System Design in Coding Questions

import asyncio
import signal
from aiohttp import web
import os
import logging
from contextlib import asynccontextmanager

logging.basicConfig(level=logging.INFO)
GRACE_PERIOD = int(os.getenv("GRACE_PERIOD_S", "30"))
PORT = int(os.getenv("PORT", "8080"))

app = web.Application()
ready = True              # readiness flag
shutdown_event = asyncio.Event()
active_requests = 0       # guarded by lock
active_lock = asyncio.Lock()

async def set_ready(value: bool):
    global ready
    ready = value
    logging.info("Readiness set to %s", value)

async def health(request):
    # Kubernetes will use /healthz for readinessProbe
    if ready:
        return web.Response(text="ok")
    raise web.HTTPServiceUnavailable(text="draining")

@asynccontextmanager
async def track_request(request):
    global active_requests
    async with active_lock:
        active_requests += 1
    try:
        yield
    finally:
        async with active_lock:
            active_requests -= 1
            logging.info("Active requests: %d", active_requests)
            if active_requests == 0:
                shutdown_event.set()

async def long_handler(request):
    # Example handler that simulates work
    async with track_request(request):
        await asyncio.sleep(5)  # simulate in-flight work
        return web.Response(text="done")

app.router.add_get("/healthz", health)
app.router.add_get("/work", long_handler)

async def initiate_graceful_shutdown(app):
    logging.info("Initiating graceful shutdown")
    await set_ready(False)          # report not-ready immediately
    # stop accepting new connections: rely on aiohttp runner.shutdown below
    # wait for active requests to drain or timeout
    try:
        await asyncio.wait_for(shutdown_event.wait(), timeout=GRACE_PERIOD)
        logging.info("All in-flight requests finished")
    except asyncio.TimeoutError:
        logging.warning("Grace period expired; force shutting down with %d active requests", active_requests)

async def start_server():
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", PORT)
    await site.start()
    logging.info("Server started on %s", PORT)

    loop = asyncio.get_running_loop()
    stop_future = loop.create_future()

    def _signal_handler():
        logging.info("SIGTERM received")
        loop.create_task(initiate_graceful_shutdown(app))
        stop_future.set_result(None)

    loop.add_signal_handler(signal.SIGTERM, _signal_handler)
    loop.add_signal_handler(signal.SIGINT, _signal_handler)

    await stop_future  # wait until signal triggers
    # stop accepting new connections and wait for cleanup
    await runner.shutdown()   # aiohttp stops accepting new requests
    await runner.cleanup()
    logging.info("Shutdown complete, exiting")

if __name__ == "__main__":
    try:
        asyncio.run(start_server())
    except Exception as e:
        logging.exception("Server exited with error: %s", e)

# metadata: vector clock is dict node->counter
def put(key, value, client_W):
    nodes = successors(hash(key), R)
    vc = increment_vector_clock(read_existing_vc(nodes))  # read latest VC if exists or start {}
    write_req = { "key": key, "value": value, "vc": vc, "ts": current_time() }
    acks = 0
    for n in nodes:
        try:
            resp = send(n, "replicate_write", write_req, timeout=SHORT)
            if resp.ok: acks += 1
        except NodeDown:
            # hinted handoff: write to a nearby coordinator to replay later
            save_hint(n, write_req)
    if acks >= client_W:
        return {"status":"ok", "vc": vc}
    else:
        return {"status":"error", "reason":"QUORUM_NOT_REACHED"}

def get(key, client_R):
    nodes = successors(hash(key), R)
    responses = []
    for n in nodes:
        try:
            resp = send(n, "read_local", {"key":key}, timeout=SHORT)
            if resp.ok:
                responses.append(resp.data)  # {value, vc, ts}
        except NodeDown:
            continue
    if len(responses) < client_R:
        return {"status":"error", "reason":"QUORUM_NOT_REACHED"}
    # conflict resolution
    # 1) Use vector clocks to detect concurrent versions
    merged = resolve_by_vector_clocks(responses)
    if merged.is_concurrent:
        # return multiple versions to client or attempt automatic merge
        return {"status":"ok", "values": merged.versions}
    else:
        # optionally read-repair: write merged back to nodes that were stale asynchronously
        async_trigger_read_repair(key, merged.canonical)
        return {"status":"ok", "value": merged.canonical}

# helpers
def resolve_by_vector_clocks(responses):
    # compare vcs: if one dominates others, pick it; if concurrent, return all concurrent versions.
    # fallback: if no vector clocks, use latest ts (LWW)
    ...

import time
import threading
import bisect
from collections import deque, Counter

class RollingBuckets:
    # Fixed-window histogram buckets (ms). Simple but effective for SRE use.
    BUCKETS = [1,2,5,10,20,50,100,200,500,1000,2000,5000]
    def __init__(self, window_seconds=60):
        self.window = deque()  # stores (timestamp, value)
        self.lock = threading.Lock()

    def add(self, ms):
        with self.lock:
            self.window.append((time.time(), ms))
            self._evict()

    def _evict(self):
        cutoff = time.time() - self.window_size
        # not used; eviction in quantile compute

    def percentiles(self, p_list=(50,95,99)):
        now = time.time()
        cutoff = now - 60
        with self.lock:
            vals = [v for (t,v) in self.window if t >= cutoff]
            # compact evict
            while self.window and self.window[0][0] < cutoff:
                self.window.popleft()
        if not vals:
            return {p:0 for p in p_list}
        vals.sort()
        res = {}
        n = len(vals)
        for p in p_list:
            idx = min(n-1, int(p/100.0*n))
            res[p] = vals[idx]
        return res

class HysteresisPolicy:
    def __init__(self, high_threshold_ms, low_threshold_ms, cooldown_seconds=30):
        self.high = high_threshold_ms
        self.low = low_threshold_ms
        self.cooldown = cooldown_seconds
        self.state = 'PRIMARY'  # PRIMARY, FALLBACK
        self.last_switch = 0

    def evaluate(self, p95_ms):
        now = time.time()
        if self.state == 'PRIMARY':
            if p95_ms >= self.high and now - self.last_switch > 0:
                self.state = 'FALLBACK'
                self.last_switch = now
        else: # FALLBACK
            if p95_ms <= self.low and now - self.last_switch >= self.cooldown:
                self.state = 'PRIMARY'
                self.last_switch = now
        return self.state

# Router uses policy to decide where to send reads
class Router:
    def __init__(self, buckets, policy, stale_counter):
        self.buckets = buckets
        self.policy = policy
        self.stale_counter = stale_counter

    def route(self, read_fn_primary, read_fn_replica, read_fn_cache, *args, **kwargs):
        percentiles = self.buckets.percentiles((50,95,99))
        decision = self.policy.evaluate(percentiles[95])
        if decision == 'PRIMARY':
            return read_fn_primary(*args, **kwargs)
        # FALLBACK: try cache, then replica, and mark if served stale
        res, from_cache, is_stale = read_fn_cache(*args, **kwargs)
        if res is not None:
            if is_stale:
                self.stale_counter.increment()
            return res
        return read_fn_replica(*args, **kwargs)

class Counter:
    def __init__(self):
        self.v = 0
        self.lock = threading.Lock()
    def increment(self, n=1):
        with self.lock:
            self.v += n
    def value(self):
        with self.lock:
            return self.v

import json
import time
import redis
from typing import Callable, Any

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

IDEMP_TTL = 60 * 5  # 5 minutes for in-progress; adjust per operation latency
RESULT_TTL = 60 * 60 * 24  # 1 day to keep successful responses if desired
LOCK_TOKEN_PREFIX = "lock:"  # to store token for safe overwrite

def handle_request(idempotency_key: str, request_fn: Callable[[], Any]) -> Any:
    """
    Atomically claim idempotency_key and run request_fn if not seen.
    Returns cached result for retries.
    """
    lock_key = f"idemp:{idempotency_key}"
    token = str(time.time()) + "-" + str(idempotency_key)
    # Try to set lock with NX and TTL atomically
    claimed = r.set(lock_key, json.dumps({"status": "in-progress", "token": token}), nx=True, ex=IDEMP_TTL)
    if claimed:
        try:
            result = request_fn()
            payload = {"status": "done", "result": result}
            # Overwrite with final result and longer TTL
            r.set(lock_key, json.dumps(payload), ex=RESULT_TTL)
            return result
        except Exception as exc:
            # Option: delete key so callers can retry immediately, or leave for TTL to expire
            r.delete(lock_key)
            raise
    else:
        # Another worker claimed it—poll for completion or return cached result if available
        for _ in range(50):  # polling for up to ~5s (tunable)
            raw = r.get(lock_key)
            if not raw:
                # key disappeared, try again (best-effort retry)
                time.sleep(0.1)
                continue
            obj = json.loads(raw)
            if obj.get("status") == "done":
                return obj.get("result")
            time.sleep(0.1)
        # Fallback: either raise or attempt to claim again
        raise RuntimeError("Idempotency timeout; no completed result found.")

import time
from math import floor
from typing import Dict, Tuple

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        # user_id -> (window_start, count)
        self.users: Dict[str, Tuple[float, int]] = {}

    def allow_request(self, user_id: str, timestamp: float) -> bool:
        window_start = floor(timestamp / self.window_seconds) * self.window_seconds
        entry = self.users.get(user_id)
        if entry is None or entry[0] != window_start:
            # new window for this user
            self.users[user_id] = (window_start, 1)
            return True
        # same window: increment and check
        ws, cnt = entry
        if cnt < self.max_requests:
            self.users[user_id] = (ws, cnt + 1)
            return True
        return False