Design a rate limiter (LLD)

TL;DR

• You build a gateway middleware: a rule decides which limit applies, a ClientKey decides whose budget it is, and one of five algorithms decides whether this request passes.
• Three decisions carry the interview: the storage owns atomicity, not the algorithm (that is what lets the same class run on a dict today and a Redis script tomorrow), the state map is striped, and every answer is a Decision object carrying Retry-After, not a bare boolean.
• Patterns that earn their place: Strategy (the algorithms), Template Method (the shared request path), Factory (rule to limiter), Pipeline/Middleware (the gateway stage), Dependency Injection (Clock, Storage).

Problem statement

"Design the rate limiter that sits in front of an API. Different endpoints have different limits — five order submissions per minute per user, a thousand reads per minute per API key — and a request that exceeds its limit gets a 429 with a Retry-After header. Support several algorithms and let operations switch between them per endpoint without a deploy. It runs in a multi-threaded gateway process, and the state must eventually live in Redis. Show me the classes, the algorithms and what happens when two threads hit the same key."

Requirements

Functional

• Per-caller keys: user id, API key, client IP, or one global bucket.
• Five algorithms: token bucket, leaky bucket, fixed window, sliding window log, sliding window counter.
• Rules per endpoint (method plus path prefix), most specific match wins, with an optional default.
• Every call answers allow or deny with the limit, what is left, and how long to wait.
• Thread-safe under concurrent requests for the same key.
• Pluggable storage: an in-process implementation now, a Redis-shaped one later, behind one Protocol.
• Middleware integration: one pipeline stage, not an annotation on every handler.
• Metrics: allowed and denied counts per rule.
• Cost-weighted requests: an expensive endpoint may spend more than one unit.

Non-functional and constraints

• The decision is on the request path, so it must be O(1) in time and bounded in memory per key.
• Configuration is hot-reloadable: changing a limit must not need a restart.
• Deterministic and testable: the clock is injected, so no test ever sleeps to watch a window roll.
• Fail-open or fail-closed is a policy question you should raise; here the limiter is in-process, so it cannot fail independently.

Out of scope: the distributed counter itself (Lua scripts, cell-based sync), client-side backoff, and quota billing.

Clarifying questions and assumptions

Question to ask	Assumption taken here
Is the limit per user, per IP, or per API key?	Per rule — KeyScope is a field of the rule, so the same endpoint can be limited differently for different tenants.
What happens to an anonymous caller under a per-user rule?	Fall back to the IP. Keying every anonymous request the same way would give them one shared budget, which one script could exhaust.
Do the limits need to be exact?	No. The sliding window counter is approximate and that is the right default; sliding log is offered for small, expensive limits.
One process or many?	One here. The Storage protocol is the seam to a shared store, and the page says exactly what changes.
Do requests all cost the same?	No — Request.cost defaults to 1, and a fan-out search can declare 10.
What does a denied request get?	429 with X-RateLimit-* and Retry-After, never below 1 second.
What happens to keys that go quiet?	evict_idle sweeps them. In Redis you would set a key TTL instead and let the server forget.

Core entities and relationships

• RateLimitRule — a frozen value object: method, path prefix, scope, algorithm, limit, window, optional burst. It validates itself and knows how specific it is.
• RuleRegistry — the ordered ruleset, sorted most specific first, swappable in one assignment for a hot reload.
• ClientKey — scope plus identity. It does not know which rule is asking; the limiter prefixes its own name so two rules can never share a counter.
• KeyExtractor (Protocol) with DefaultKeyExtractor — request to ClientKey, including the anonymous fallback.
• RateLimiter (Protocol) — allow(key, cost) -> Decision. StorageBackedLimiter is the abstract base the five algorithms extend.
• Storage (Protocol) with InMemoryStorage — one atomic apply per key, plus idle eviction. LimiterState is the record it keeps.
• LimiterFactory — rule to limiter, cached by rule value.
• RateLimitMiddleware — the pipeline stage. RateLimitMetrics and RuleCounters — the counters beside it.
• Decision — allowed, limit, remaining, retry-after, and the headers() an HTTP layer needs.

Multiplicities: registry 1 → * rules, rule 1 → 1 limiter, limiter 1 → 1 storage, storage 1 → * states, middleware 1 → 1 registry and 1 → 1 factory.

Class diagram

The request path: rule, key, decision, response.

classDiagram
    class Request {
        +method: str
        +path: str
        +client_ip: str
        +user_id: str
        +api_key: str
        +cost: int
    }
    class Response {
        +status: int
        +body: str
        +with_headers(extra) Response
    }
    class RateLimitMiddleware {
        +metrics: RateLimitMetrics
        +__call__(request, next_handler) Response
    }
    class RuleRegistry {
        -_lock: Lock
        +rule_for(method, path) RateLimitRule
        +replace(rules)
        +rules() Tuple
    }
    class RateLimitRule {
        +name: str
        +scope: KeyScope
        +algorithm: Algorithm
        +limit: int
        +window_seconds: float
        +matches(method, path) bool
        +specificity() Tuple
    }
    class ClientKey {
        +scope: KeyScope
        +value: str
        +storage_key() str
    }
    class KeyExtractor {
        <<interface>>
        +extract(request, scope) ClientKey
    }
    class DefaultKeyExtractor
    class LimiterFactory {
        +for_rule(rule) RateLimiter
        +prune(active) int
    }
    class RateLimitMetrics {
        -_lock: Lock
        +record(rule_name, decision)
        +snapshot() Dict~str,RuleCounters~
    }
    class RuleCounters {
        +allowed: int
        +denied: int
        +denied_ratio: float
    }
    class Decision {
        +allowed: bool
        +limit: int
        +remaining: int
        +retry_after: float
        +headers() Dict~str,str~
    }
    RateLimitMiddleware --> RuleRegistry
    RateLimitMiddleware --> KeyExtractor
    RateLimitMiddleware --> LimiterFactory
    RateLimitMiddleware --> RateLimitMetrics
    RateLimitMiddleware ..> Request : reads
    RateLimitMiddleware ..> Response : returns
    RuleRegistry "1" o-- "many" RateLimitRule
    KeyExtractor <|.. DefaultKeyExtractor
    DefaultKeyExtractor ..> ClientKey : creates
    RateLimitMetrics "1" o-- "many" RuleCounters
    RateLimitMiddleware ..> Decision : records

The algorithms and the storage they all share.

classDiagram
    class RateLimiter {
        <<interface>>
        +allow(key, cost) Decision
    }
    class StorageBackedLimiter {
        <<abstract>>
        -_capacity: int
        -_window: float
        +allow(key, cost) Decision
        +_initial(now) LimiterState
        +_decide(state, now, cost) Decision
    }
    class TokenBucket
    class LeakyBucket
    class FixedWindowCounter
    class SlidingWindowLog
    class SlidingWindowCounter
    class Storage {
        <<interface>>
        +apply(key, now, mutate) Decision
        +evict_idle(cutoff) int
    }
    class InMemoryStorage {
        -_locks: List~Lock~
        -_states: List~Dict~
        +apply(key, now, mutate) Decision
        +evict_idle(cutoff) int
        +keys() List~str~
    }
    class LimiterState {
        +updated_at: float
        +level: float
        +window_start: float
        +count: int
        +previous: int
        +log: Deque
    }
    class LimiterFactory {
        +BUILDERS: Dict
        +for_rule(rule) RateLimiter
    }
    RateLimiter <|.. StorageBackedLimiter
    StorageBackedLimiter <|-- TokenBucket
    StorageBackedLimiter <|-- LeakyBucket
    StorageBackedLimiter <|-- FixedWindowCounter
    StorageBackedLimiter <|-- SlidingWindowLog
    StorageBackedLimiter <|-- SlidingWindowCounter
    StorageBackedLimiter --> Storage
    Storage <|.. InMemoryStorage
    InMemoryStorage "1" *-- "many" LimiterState
    LimiterFactory ..> StorageBackedLimiter : builds

Design patterns applied

Pattern	Where	Why it earns its place
Strategy	RateLimiter with five implementations	The algorithm is the thing the interviewer swaps. Because it is a Protocol, a rule's algorithm field is the only thing that changes when operations move an endpoint from fixed window to sliding counter.
Template Method	StorageBackedLimiter.allow	Validation, the single clock read and the storage round trip are identical across five algorithms. Each subclass writes only _initial and _decide, so the five algorithms read as five paragraphs of arithmetic.
Factory	LimiterFactory.BUILDERS	Configuration names an algorithm as a string; the registry maps it to a class. Keying the cache by the rule value means a changed limit yields a new limiter with no invalidation logic.
Pipeline / Middleware	RateLimitMiddleware.__call__(request, next_handler)	The limit belongs to the route, not to the handler function. One stage covers every endpoint, and it composes with auth and logging stages in a known order.
Dependency Injection	Clock, Storage, KeyExtractor	FakeClock rolls a window instantly; a Redis-shaped storage swaps in without touching an algorithm.
Value Object	Decision, RateLimitRule, ClientKey	Frozen, hashable, and safe to share across threads. Decision is what turns "denied" into a response an HTTP client can act on.

What was deliberately not used: Decorator per handler (@rate_limit(5, "1/min")). It reads nicely and it is the wrong seam — the limit becomes source code, so operations cannot change it during an incident, and an endpoint added without the decorator is silently unlimited. Singleton for the storage is also avoided: one instance built in main and injected lets a test build a fresh limiter per case, which is why the test file has no shared state.

Key flows

One request through the gateway: rule, key, decision, then pass or 429.

sequenceDiagram
    autonumber
    actor C as Client
    participant M as RateLimitMiddleware
    participant R as RuleRegistry
    participant F as LimiterFactory
    participant L as TokenBucket
    participant S as InMemoryStorage
    participant H as Handler
    C->>M: POST /api/orders
    M->>R: rule_for(POST, /api/orders)
    R-->>M: orders-write, 5 per minute, per user
    M->>M: extract ClientKey(user, u-1)
    M->>F: for_rule(rule)
    F-->>M: cached TokenBucket
    M->>L: allow(user|u-1, cost)
    L->>S: apply(orders-write|user|u-1, now, mutate)
    S->>S: lock the stripe, refill, spend, unlock
    S-->>L: Decision
    L-->>M: Decision
    alt allowed
        M->>H: next_handler(request)
        H-->>M: 200
        M-->>C: 200 with X-RateLimit headers
    else denied
        M-->>C: 429 with Retry-After
    end

Why the storage owns atomicity: the same two threads, with and without the stripe lock.

sequenceDiagram
    autonumber
    participant T1 as Thread1
    participant T2 as Thread2
    participant S as InMemoryStorage
    Note over T1,T2: one token left, two requests
    T1->>S: apply(k), takes stripe lock
    T2->>S: apply(k), blocks on the same stripe
    S->>S: read level 1, spend, write level 0
    S-->>T1: allowed
    S->>S: thread 2 proceeds, reads level 0
    S-->>T2: denied, retry_after 12 s
    Note over T1,T2: without the lock both read 1 and both spend it

Implementation

Write it in the order the request travels: what a rule is, what an answer is, then the algorithms, then the storage and the middleware.

The enums are the configuration vocabulary — an operator writes "sliding_counter" in a config file and it lands here:

code/lld/rate_limiter/models.py — algorithms, scopes and errors

class Algorithm(StrEnum):
    TOKEN_BUCKET = "token_bucket"  # bursty, smooth refill, two numbers per key
    LEAKY_BUCKET = "leaky_bucket"  # shapes traffic: admits at a constant rate
    FIXED_WINDOW = "fixed_window"  # cheapest, leaks 2x the limit at a boundary
    SLIDING_LOG = "sliding_log"  # exact, O(limit) memory per key
    SLIDING_COUNTER = "sliding_counter"  # approximate, O(1) memory, no boundary burst


class KeyScope(StrEnum):
    """What "one caller" means for a rule. Two rules may scope the same request differently."""

    USER = "user"
    API_KEY = "api_key"
    IP = "ip"
    GLOBAL = "global"


class RuleNotFoundError(NotFoundError):
    """No rule matches the request and the registry has no default."""


class UnknownAlgorithmError(ValidationError):
    """A rule names an algorithm the factory cannot build."""

RateLimitRule validates itself in __post_init__, so an impossible rule fails at load time rather than on the first request that matches it. specificity is the sort key that makes "most specific wins" one line in the registry.

code/lld/rate_limiter/models.py — rule and client key

@dataclass(frozen=True, slots=True)
class RateLimitRule:
    """One configured limit: which requests it covers, per whom, how much.

    ``burst`` only means something to the token bucket: it is the capacity, so
    ``limit=60, window_seconds=60, burst=10`` reads as "one request per second on
    average, but ten may arrive at once".
    """

    name: str
    method: str  # "GET", "POST", or "*"
    path_prefix: str
    scope: KeyScope
    algorithm: Algorithm
    limit: int
    window_seconds: float
    burst: int | None = None

    def __post_init__(self) -> None:
        if self.limit <= 0 or self.window_seconds <= 0:
            raise ValidationError(f"rule {self.name}: limit and window must be positive")
        if self.burst is not None and self.burst <= 0:
            raise ValidationError(f"rule {self.name}: burst must be positive")
        if not self.path_prefix.startswith("/"):
            raise ValidationError(f"rule {self.name}: path_prefix must start with '/'")

    @property
    def capacity(self) -> int:
        """What a single request is measured against: the burst for buckets, the limit otherwise."""
        return self.burst or self.limit

    @property
    def refill_rate(self) -> float:
        """Tokens per second. 60 per minute is 1.0 per second."""
        return self.limit / self.window_seconds

    def matches(self, method: str, path: str) -> bool:
        method_ok = self.method in (WILDCARD, method)
        return method_ok and path.startswith(self.path_prefix)

    def specificity(self) -> tuple[int, int]:
        """Sort key: the longest path wins, an exact method beats a wildcard."""
        return len(self.path_prefix), 0 if self.method == WILDCARD else 1


@dataclass(frozen=True, slots=True)
class ClientKey:
    """Who is being limited: a scope and the identity inside it.

    It deliberately does *not* know which rule is asking. The limiter prefixes its
    own rule name, so a caller cannot forget to namespace and two rules covering
    the same user - a strict one on ``POST /orders``, a loose one on everything -
    can never end up sharing a counter.
    """

    scope: KeyScope
    value: str

    def storage_key(self) -> str:
        return f"{self.scope}|{self.value}"

Decision is the object that separates a good answer from a bare boolean: the caller learns the ceiling, what is left and when to come back, and headers() renders exactly what an HTTP client reads.

code/lld/rate_limiter/models.py — decision, counters and messages

@dataclass(frozen=True, slots=True)
class Decision:
    """The answer to one request, carrying everything an HTTP response needs."""

    allowed: bool
    limit: int
    remaining: int
    retry_after: float = 0.0  # seconds until a request of this cost would pass
    rule: str = ""
    delay: float = 0.0  # shaping limiters only: how long the admitted request waits

    def headers(self) -> dict[str, str]:
        out = {
            "X-RateLimit-Limit": str(self.limit),
            "X-RateLimit-Remaining": str(max(0, self.remaining)),
        }
        if not self.allowed:
            # Retry-After is an integer number of seconds, and never 0: a client
            # that reads 0 retries immediately and is denied again.
            out["Retry-After"] = str(max(1, math.ceil(self.retry_after)))
        return out


@dataclass(frozen=True, slots=True)
class RuleCounters:
    allowed: int = 0
    denied: int = 0

    @property
    def total(self) -> int:
        return self.allowed + self.denied

    @property
    def denied_ratio(self) -> float:
        return self.denied / self.total if self.total else 0.0


@dataclass(frozen=True, slots=True)
class Request:
    """The slice of an HTTP request this middleware needs. Frozen: stages copy, never mutate."""

    method: str
    path: str
    client_ip: str = "0.0.0.0"
    user_id: str | None = None
    api_key: str | None = None
    cost: int = 1  # a search that fans out to ten shards can declare cost=10

    def __post_init__(self) -> None:
        if self.cost <= 0:
            raise ValidationError("cost must be positive")


@dataclass(frozen=True, slots=True)
class Response:
    status: int
    body: str
    headers: Mapping[str, str] = field(default_factory=dict)

    def with_headers(self, extra: Mapping[str, str]) -> Response:
        return replace(self, headers={**self.headers, **extra})

One record type serves all five algorithms. Say why out loud: it keeps the Storage protocol free of algorithm-specific methods, and a Redis-backed store would persist only the fields its algorithm touches.

code/lld/rate_limiter/models.py — the stored record

@dataclass(slots=True)
class LimiterState:
    """Every field the five algorithms may keep for one key, in one record.

    A Redis-backed store keeps only what its algorithm touches - a token bucket
    is two numbers in a hash, a sliding log is a sorted set - but in process one
    record keeps the ``Storage`` protocol free of algorithm-specific methods.
    """

    updated_at: float
    level: float = 0.0  # token bucket: tokens left. leaky bucket: work queued
    window_start: float = 0.0
    count: int = 0  # requests inside the current window
    previous: int = 0  # sliding counter only: the previous window's final count
    log: deque[float] = field(default_factory=deque)  # sliding log only

    def idle_since(self, now: float) -> float:
        return now - self.updated_at

Now the two protocols. The contract on Storage.apply is the sentence to say in the room: read, decide and write are one indivisible step. A storage offering plain get and set would turn every algorithm into a read-modify-write race.

code/lld/rate_limiter/strategies.py — the two protocols

class Storage(Protocol):
    """Where limiter state lives, with one hard contract: ``apply`` is atomic per key.

    Read, decide and write must be indivisible. In process that is a lock; in
    Redis it is a Lua script or a ``WATCH``/``MULTI`` loop - which is why the
    distributed version ships the algorithm to the data instead of pulling the
    data to the algorithm. A storage that offered plain ``get`` and ``set`` would
    make every limiter a read-modify-write race.
    """

    def apply(self, key: str, now: float, mutate: Mutation) -> Decision: ...

    def evict_idle(self, cutoff: float) -> int:
        """Forget keys untouched since ``cutoff``. Returns how many were dropped."""

    def __len__(self) -> int: ...


class RateLimiter(Protocol):
    """One decision per request. ``key`` already identifies rule plus caller."""

    def allow(self, key: str, cost: int = 1) -> Decision: ...

The abstract base holds everything the five algorithms share, including the single clock read and the rule namespacing:

code/lld/rate_limiter/strategies.py — the shared request path

class StorageBackedLimiter(ABC):
    """Template Method: read the clock once, then mutate exactly one key atomically.

    Every algorithm answers the same two questions and nothing else: what does a
    key look like the first time it is seen, and what does one request do to it.
    Validation, the clock read and the storage round trip are written once here.
    """

    def __init__(
        self,
        capacity: int,
        window_seconds: float,
        storage: Storage,
        clock: Clock | None = None,
        rule: str = "",
    ) -> None:
        if capacity <= 0 or window_seconds <= 0:
            raise ValidationError("capacity and window must be positive")
        self._capacity = capacity
        self._window = window_seconds
        self._storage = storage
        self._clock = clock or SystemClock()
        self._rule = rule

    @property
    def capacity(self) -> int:
        return self._capacity

    def allow(self, key: str, cost: int = 1) -> Decision:
        if cost <= 0:
            raise ValidationError("cost must be positive")
        if cost > self._capacity:
            raise ValidationError(f"cost {cost} can never fit under a limit of {self._capacity}")
        now = self._clock.now()

        def mutate(state: LimiterState | None) -> tuple[LimiterState, Decision]:
            record = self._initial(now) if state is None else state
            return record, self._decide(record, now, cost)

        return self._storage.apply(self._namespaced(key), now, mutate)

    def _namespaced(self, key: str) -> str:
        """Every rule owns its own key space; two rules can never share one counter."""
        return f"{self._rule}|{key}"

    @abstractmethod
    def _initial(self, now: float) -> LimiterState:
        """The record for a key nobody has asked about yet."""

    @abstractmethod
    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        """Mutate the record for one request and answer."""

    def _verdict(self, allowed: bool, remaining: float, retry_after: float = 0.0, delay: float = 0.0) -> Decision:
        return Decision(
            allowed=allowed,
            limit=self._capacity,
            remaining=max(0, math.floor(remaining)),
            retry_after=max(0.0, retry_after),
            rule=self._rule,
            delay=delay,
        )

The bucket algorithms are continuous: they compute the refill or the drain from elapsed time, so an idle key costs nothing and there is no timer thread anywhere in the design.

code/lld/rate_limiter/strategies.py — token bucket and leaky bucket

class TokenBucket(StorageBackedLimiter):
    """``capacity`` tokens refilled continuously at ``capacity / window`` per second.

    The default choice for a public API: it absorbs a burst up to the capacity and
    then settles to the sustained rate, and an idle key costs nothing because the
    refill is computed from the elapsed time rather than by a timer.
    """

    def _initial(self, now: float) -> LimiterState:
        return LimiterState(updated_at=now, level=float(self._capacity))

    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        rate = self._capacity / self._window
        state.level = min(float(self._capacity), state.level + (now - state.updated_at) * rate)
        if state.level >= cost:
            state.level -= cost
            return self._verdict(True, state.level)
        return self._verdict(False, state.level, retry_after=(cost - state.level) / rate)


class LeakyBucket(StorageBackedLimiter):
    """A queue of depth ``capacity`` draining at a constant rate: a shaper, not a gate.

    It admits the same requests a token bucket would, but each admitted request
    carries the delay until the work queued ahead of it has drained, so a burst
    leaves at an even rate instead of all at once. Use it in front of something
    that cannot absorb a spike at all, such as a payment provider.
    """

    def _initial(self, now: float) -> LimiterState:
        return LimiterState(updated_at=now, level=0.0)

    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        rate = self._capacity / self._window
        state.level = max(0.0, state.level - (now - state.updated_at) * rate)
        room = self._capacity - state.level
        if cost <= room:
            delay = state.level / rate
            state.level += cost
            return self._verdict(True, self._capacity - state.level, delay=delay)
        return self._verdict(False, room, retry_after=(cost - room) / rate)

The window algorithms trade memory against accuracy in three steps: a counter that leaks at the boundary, an exact log that costs O(limit) per key, and the weighted estimate that most gateways actually run.

code/lld/rate_limiter/strategies.py — the three window algorithms

class FixedWindowCounter(StorageBackedLimiter):
    """One counter per aligned window. The cheapest algorithm and the leakiest.

    ``limit`` requests just before a boundary plus ``limit`` just after pass back
    to back, so a limit of 100 per minute can serve 200 in two seconds. Say that
    trade out loud before the interviewer does.
    """

    def _initial(self, now: float) -> LimiterState:
        return LimiterState(updated_at=now, window_start=self._align(now))

    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        start = self._align(now)
        if state.window_start != start:
            state.window_start, state.count = start, 0
        if state.count + cost <= self._capacity:
            state.count += cost
            return self._verdict(True, self._capacity - state.count)
        return self._verdict(False, self._capacity - state.count, retry_after=start + self._window - now)

    def _align(self, now: float) -> float:
        return math.floor(now / self._window) * self._window


class SlidingWindowLog(StorageBackedLimiter):
    """Exact: the timestamps of the accepted requests in the trailing window.

    Memory is O(limit) per key, which makes it right for small limits on
    expensive actions - login attempts, password resets, one-time-code sends -
    and wrong for a public API at thousands of requests per key per minute.
    """

    def _initial(self, now: float) -> LimiterState:
        return LimiterState(updated_at=now)

    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        horizon = now - self._window
        while state.log and state.log[0] <= horizon:
            state.log.popleft()
        if len(state.log) + cost <= self._capacity:
            state.log.extend([now] * cost)
            return self._verdict(True, self._capacity - len(state.log))
        # the request fits once this many of the oldest entries have aged out
        index = len(state.log) + cost - self._capacity - 1
        return self._verdict(False, 0, retry_after=state.log[index] + self._window - now)


class SlidingWindowCounter(StorageBackedLimiter):
    """Approximate: this window's count plus the previous one's, weighted by overlap.

    Two integers per key, no boundary burst, and one assumption - that the
    previous window's requests were spread evenly across it. That assumption is
    wrong for a single spike and close enough for aggregate traffic, which is why
    this is what most gateways actually run.
    """

    def _initial(self, now: float) -> LimiterState:
        return LimiterState(updated_at=now, window_start=self._align(now))

    def _decide(self, state: LimiterState, now: float, cost: int) -> Decision:
        start = self._align(now)
        if state.window_start != start:
            carried = state.count if state.window_start == start - self._window else 0
            state.window_start, state.previous, state.count = start, carried, 0
        overlap = 1.0 - (now - start) / self._window
        estimate = state.previous * overlap + state.count
        if estimate + cost <= self._capacity:
            state.count += cost
            return self._verdict(True, self._capacity - estimate - cost)
        return self._verdict(False, self._capacity - estimate, retry_after=self._retry_after(state, now, start, cost))

    def _retry_after(self, state: LimiterState, now: float, start: float, cost: int) -> float:
        room = self._capacity - state.count - cost  # what the previous window may still occupy
        if state.previous <= 0 or room < 0:
            return start + self._window - now  # only a fresh window can help
        return start + self._window * (1 - room / state.previous) - now

    def _align(self, now: float) -> float:
        return math.floor(now / self._window) * self._window

The storage is where the locks are. Note that each stripe owns its own dict rather than sharing one:

code/lld/rate_limiter/services.py — striped storage

class InMemoryStorage:
    """Process-local limiter state behind striped locks: one lock *and one dict* per stripe.

    Striping is the answer to "one lock for the whole gateway is a bottleneck".
    Two keys that land on different stripes never wait for each other, so
    contention falls roughly with the stripe count, while requests for the *same*
    key still serialise - which is what you want, because the hot key is exactly
    the one that has to be counted correctly.

    Each stripe owns its own dict on purpose. One shared dict mutated under
    several different locks would be relying on CPython's internals for safety
    rather than on a lock, and that is not a claim you want to defend.
    """

    def __init__(self, stripes: int = DEFAULT_STRIPES) -> None:
        if stripes < 1:
            raise ValidationError("stripes must be at least 1")
        self._locks = [threading.Lock() for _ in range(stripes)]
        self._states: list[dict[str, LimiterState]] = [{} for _ in range(stripes)]

    def apply(self, key: str, now: float, mutate: Mutation) -> Decision:
        index = hash(key) % len(self._locks)
        with self._locks[index]:
            states = self._states[index]
            record, decision = mutate(states.get(key))
            record.updated_at = now  # the storage stamps it, so no algorithm can forget to
            states[key] = record
            return decision

    def evict_idle(self, cutoff: float) -> int:
        """Drop keys untouched since ``cutoff``. In Redis this is a key TTL instead."""
        removed = 0
        for index, lock in enumerate(self._locks):
            with lock:
                states = self._states[index]
                stale = [key for key, state in states.items() if state.updated_at <= cutoff]
                for key in stale:
                    del states[key]
                removed += len(stale)
        return removed

    def __len__(self) -> int:
        return sum(len(states) for states in self._states)

    def keys(self) -> list[str]:
        out: list[str] = []
        for index, lock in enumerate(self._locks):
            with lock:
                out.extend(self._states[index])
        return out

The registry and the key extractor are the configuration half. Copy-on-write is the trick worth naming: readers copy a reference to an immutable tuple and scan outside the lock.

code/lld/rate_limiter/services.py — rules and keys

class RuleRegistry:
    """Which rule covers a request, and the seam that makes configuration hot-reloadable.

    ``_lock`` guards one thing: the reference to an immutable tuple of rules. A
    reader copies that reference under the lock and then scans outside it, so a
    reload publishes a whole ruleset at once and no request can ever observe a
    half-applied configuration. Copy-on-write beats locking the read path.
    """

    def __init__(self, rules: Iterable[RateLimitRule] = (), default: RateLimitRule | None = None) -> None:
        self._lock = threading.Lock()
        self._rules = self._ordered(rules)
        self._default = default

    def rules(self) -> tuple[RateLimitRule, ...]:
        with self._lock:
            return self._rules

    def replace(self, rules: Iterable[RateLimitRule]) -> None:
        """Hot reload: swap the whole ruleset in one assignment."""
        ordered = self._ordered(rules)
        with self._lock:
            self._rules = ordered

    def rule_for(self, method: str, path: str) -> RateLimitRule:
        for rule in self.rules():  # already sorted most specific first
            if rule.matches(method, path):
                return rule
        if self._default is not None:
            return self._default
        raise RuleNotFoundError(f"no rate limit rule matches {method} {path}")

    @staticmethod
    def _ordered(rules: Iterable[RateLimitRule]) -> tuple[RateLimitRule, ...]:
        return tuple(sorted(rules, key=lambda r: r.specificity(), reverse=True))


class KeyExtractor(Protocol):
    """Turns a request into the identity a rule counts against."""

    def extract(self, request: Request, scope: KeyScope) -> ClientKey: ...


class DefaultKeyExtractor:
    """User id, API key or client IP, with the IP as the fallback for anonymous callers.

    The fallback is a real decision: keying every anonymous request as ``user:none``
    would give all of them one shared budget, so a single script could exhaust the
    quota of every logged-out visitor at once.
    """

    def extract(self, request: Request, scope: KeyScope) -> ClientKey:
        if scope is KeyScope.USER and request.user_id:
            return ClientKey(KeyScope.USER, request.user_id)
        if scope is KeyScope.API_KEY and request.api_key:
            return ClientKey(KeyScope.API_KEY, request.api_key)
        if scope is KeyScope.GLOBAL:
            return ClientKey(KeyScope.GLOBAL, "all")
        return ClientKey(KeyScope.IP, request.client_ip)

code/lld/rate_limiter/services.py — the factory

class LimiterFactory:
    """Factory: a rule names an algorithm, this turns it into an object - once.

    The cache is keyed by the rule *value*, not by its name. A frozen dataclass
    hashes by its fields, so editing a limit produces a different key and a fresh
    limiter automatically: no invalidation logic, and no stale ceiling surviving
    a hot reload.
    """

    BUILDERS: dict[Algorithm, type[StorageBackedLimiter]] = {
        Algorithm.TOKEN_BUCKET: TokenBucket,
        Algorithm.LEAKY_BUCKET: LeakyBucket,
        Algorithm.FIXED_WINDOW: FixedWindowCounter,
        Algorithm.SLIDING_LOG: SlidingWindowLog,
        Algorithm.SLIDING_COUNTER: SlidingWindowCounter,
    }

    def __init__(self, storage: Storage, clock: Clock | None = None) -> None:
        self._storage = storage
        self._clock = clock or SystemClock()
        self._lock = threading.Lock()
        self._limiters: dict[RateLimitRule, RateLimiter] = {}

    def for_rule(self, rule: RateLimitRule) -> RateLimiter:
        with self._lock:
            limiter = self._limiters.get(rule)
            if limiter is None:
                limiter = self._limiters[rule] = self._build(rule)
            return limiter

    def prune(self, active: Iterable[RateLimitRule]) -> int:
        """Drop limiters for rules that are no longer configured."""
        keep = set(active)
        with self._lock:
            dead = [rule for rule in self._limiters if rule not in keep]
            for rule in dead:
                del self._limiters[rule]
            return len(dead)

    def _build(self, rule: RateLimitRule) -> RateLimiter:
        try:
            builder = self.BUILDERS[rule.algorithm]
        except KeyError as exc:
            raise UnknownAlgorithmError(f"no limiter for algorithm {rule.algorithm!r}") from exc
        return builder(
            capacity=rule.capacity,
            window_seconds=rule.window_seconds,
            storage=self._storage,
            clock=self._clock,
            rule=rule.name,
        )

The middleware is then eight lines of glue, which is the sign the seams are in the right places:

code/lld/rate_limiter/services.py — metrics and middleware

class RateLimitMetrics:
    """Allowed and denied counts per rule, behind their own lock.

    Deliberately *not* the storage lock: a metrics update must never run inside
    the critical section that every request for a hot key already serialises on.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._counters: dict[str, RuleCounters] = {}

    def record(self, rule_name: str, decision: Decision) -> None:
        with self._lock:
            current = self._counters.get(rule_name, RuleCounters())
            self._counters[rule_name] = RuleCounters(
                allowed=current.allowed + (1 if decision.allowed else 0),
                denied=current.denied + (0 if decision.allowed else 1),
            )

    def snapshot(self) -> dict[str, RuleCounters]:
        with self._lock:
            return dict(self._counters)


class RateLimitMiddleware:
    """One stage of the request pipeline: find the rule, build the key, ask, then pass or 429.

    It is a middleware rather than a decorator on each handler because the limit
    belongs to the route, not to the function: adding an endpoint should not mean
    remembering to annotate it.
    """

    def __init__(
        self,
        registry: RuleRegistry,
        factory: LimiterFactory,
        extractor: KeyExtractor | None = None,
        metrics: RateLimitMetrics | None = None,
    ) -> None:
        self._registry = registry
        self._factory = factory
        self._extractor = extractor or DefaultKeyExtractor()
        self.metrics = metrics or RateLimitMetrics()

    def __call__(self, request: Request, next_handler: Handler) -> Response:
        try:
            rule = self._registry.rule_for(request.method, request.path)
        except RuleNotFoundError:
            return next_handler(request)  # unlimited routes are a configuration choice
        key = self._extractor.extract(request, rule.scope)
        decision = self._factory.for_rule(rule).allow(key.storage_key(), cost=request.cost)
        self.metrics.record(rule.name, decision)
        if not decision.allowed:
            return Response(TOO_MANY_REQUESTS, f"rate limit exceeded for rule {rule.name}", decision.headers())
        return next_handler(request).with_headers(decision.headers())

Running python -m lld.rate_limiter.demo puts three rules behind the middleware and then runs the same burst through all five algorithms:

--- 7 POST /api/orders from user u-1, limit 5 per minute ---
  allowed 5, denied 2, last body: rate limit exceeded for rule orders-write
  429 headers: {'X-RateLimit-Limit': '5', 'X-RateLimit-Remaining': '0', 'Retry-After': '12'}
--- the same user on a different route uses a different budget ---
  cost=5 against a limit of 10: 200, 200, then 429
--- an anonymous request falls through to the IP rule ---
  GET /health -> 200, {'X-RateLimit-Limit': '100', 'X-RateLimit-Remaining': '99'}
--- limit 5 per second, 7 requests arrive together at t=0 ---
  token bucket    : 5 allowed, 2 denied, Retry-After 0.20 s
  leaky bucket    : 5 allowed, 2 denied, Retry-After 0.20 s
  fixed window    : 5 allowed, 2 denied, Retry-After 1.00 s
  sliding log     : 5 allowed, 2 denied, Retry-After 1.00 s
  sliding counter : 5 allowed, 2 denied, Retry-After 1.00 s
--- boundary burst: 5 requests at t+0.9 s, 5 more at t+1.0 s ---
  fixed window    : 5 allowed, then 5 more within 0.1 s
  sliding counter : 5 allowed, then 0 more within 0.1 s
--- housekeeping ---
  keys held: 10, evicted after 10 min idle: 10
  catch-all: 1 allowed / 0 denied, orders-write: 5 allowed / 2 denied, search: 2 allowed / 1 denied

Concurrency and edge cases

Which lock protects what. Four locks, each with one job.

1. InMemoryStorage._locks[i] guards _states[i], one dict per stripe. The race it prevents is the read-modify-write: "read the level, subtract the cost, write it back" is three steps, and two threads that interleave them both spend the same token. With 64 stripes, two keys collide only when their hashes agree modulo 64, so unrelated tenants never wait for each other — while requests for the same key still serialise, which is right: the hot key is the one that must be counted correctly.
2. RuleRegistry._lock guards a single reference to an immutable tuple. A reload assigns a new tuple; readers copy the reference under the lock and scan outside it. No request can observe a half-applied ruleset, and the read path costs one uncontended acquire, about 17 ns.
3. LimiterFactory._lock guards the rule-to-limiter cache: a check-then-insert, so two threads missing the same rule build one limiter, not two.
4. RateLimitMetrics._lock guards the counters, and deliberately not the storage stripe — a metrics update must never run inside the critical section every request for a hot key already queues on.

Why each stripe owns its dict. One shared dict mutated under several different locks leans on CPython's internals rather than on a lock — not a claim you want to defend in a review.

Where atomicity really lives. The algorithms never lock. They hand a closure to Storage.apply and the storage makes read-decide-write indivisible. That is the seam that survives the move to Redis: in process a stripe lock; in Redis a Lua script, which ships the algorithm to the data instead of pulling the data to the algorithm. The cost difference is the whole argument for keeping a local tier: an in-process decision is one uncontended mutex, about 17 ns, against a 500 µs same-datacenter round trip for the Redis call — roughly 30,000 times cheaper.

Idle keys. Every distinct caller creates a record, so the map grows with your IP space unless something removes it. evict_idle(cutoff) sweeps per stripe; in Redis you would set a key expiry and let the server forget. Sizing then tells you which algorithm you can afford: a token bucket record is a level and a timestamp, 8 B each, while a sliding log at a limit of 1,000 per key is 1,000 timestamps, 8 KB — so a million active keys is 8 GB of nothing but timestamps. That is why sliding log belongs on login attempts and not on a public read API.

Edge cases handled: a cost larger than the limit is rejected at once instead of being denied forever; Retry-After is never rendered as 0, because a client reading 0 retries immediately and is denied again; an anonymous caller under a per-user rule falls back to its IP; a request that matches no rule passes, a configuration choice this page states rather than a silent default; raising a limit through a hot reload changes the ceiling and the refill rate at once but does not hand back tokens already spent; and the sliding window counter's retry_after solves for when enough of the previous window will have aged out, rather than always pointing at the next boundary.

Common mistake

Returning True/False from allow. The caller then has to invent the Retry-After value, and every caller invents a different one — usually the whole window, which turns a one-second wait into a sixty-second one and makes clients bunch up at the next boundary. Return a Decision, compute retry_after from the algorithm's own state, and let headers() render it.

Extensibility and follow-ups

• Distributed counters: replace InMemoryStorage with one that runs the same decision as a Redis Lua script. The algorithm classes do not change, because the atomicity contract already lives in Storage. That conversation — local-then-global, cell-based sync, eventual accuracy — is the HLD case study.
• Hierarchical limits: a request may have to satisfy per-user and per-tenant and global rules. Change rule_for to return every match and require all of them to allow, and make sure you spend the tokens only once every rule has agreed, or a denial by the last rule will have silently charged the first.
• Cost-weighted requests: already here as Request.cost. The next step is deriving the cost from the response (a search that touched ten shards) and settling the difference afterwards, which turns the limiter into an accounting problem.
• Fail-open versus fail-closed: once the storage is remote it can be unavailable. Decide out loud — fail-open protects availability and lets an attack through, fail-closed protects the backend and turns a cache outage into an outage.
• Hot reload from a config service: RuleRegistry.replace is already the seam; add a watcher that validates and swaps. LimiterFactory.prune drops the limiters built from retired rules.
• A radix tree for routes: the registry scans an ordered tuple, which is fine for tens of rules. At hundreds you would index by method and path segment — and only then, because scanning 20 rules costs less than the lock beside it.

Sizing: a gateway at 10k QPS calls the limiter once per request, so a Redis instance rated at roughly 100k ops/s absorbs ten such nodes — until a second rule per request halves that to five.

Interview tip

Lead with the comparison table, not with code. "Fixed window is one integer and leaks twice the limit at a boundary; sliding log is exact and costs O(limit) memory per key; sliding counter is two integers and approximates by overlap; token bucket allows a burst and then the sustained rate." Four sentences buys you the whole design discussion, and then you write the one the interviewer picks.

Tests

tests/test_rate_limiter.py has 23 cases. The two worth walking through are the boundary comparison, because it is the answer to the question every interviewer asks, and the concurrency test.

code/lld/rate_limiter/tests/test_rate_limiter.py — the boundary burst

def test_fixed_window_leaks_at_the_boundary_and_the_sliding_counter_does_not(
    clock: FakeClock, storage: InMemoryStorage
) -> None:
    """The one comparison every interviewer asks for, in six lines."""
    fixed = build(FixedWindowCounter, clock, storage)
    sliding = build(SlidingWindowCounter, clock, storage)
    clock.set(1_000_000.9)  # late in the window
    assert sum(fixed.allow("u-1").allowed for _ in range(5)) == 5
    assert sum(sliding.allow("u-1").allowed for _ in range(5)) == 5
    clock.set(1_000_001.0)  # the next window starts
    assert sum(fixed.allow("u-1").allowed for _ in range(5)) == 5  # 10 in 100 ms
    assert sum(sliding.allow("u-1").allowed for _ in range(5)) == 0  # the overlap still counts

With an injected clock, the flaw everyone describes in words becomes four assertions: ten requests pass in 100 milliseconds under a fixed window of five per second, and none pass under the sliding counter.

code/lld/rate_limiter/tests/test_rate_limiter.py — concurrency

def test_concurrent_requests_never_exceed_the_limit(clock: FakeClock, storage: InMemoryStorage) -> None:
    """8 threads, 400 requests, a frozen clock, capacity 100: exactly 100 pass.

    Without the stripe lock, "read the level, subtract the cost, write it back" is
    three steps, and two threads that interleave them both spend the same token.
    The frozen clock removes refill from the picture, so any number above 100
    is a lost update rather than time passing.
    """
    limiter = build(TokenBucket, clock, storage, capacity=100)
    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(lambda _: limiter.allow("hot-tenant").allowed, range(400)))
    assert sum(results) == 100
    assert len(storage) == 1  # one key, one stripe, one contended lock


def test_independent_keys_do_not_interfere(clock: FakeClock, storage: InMemoryStorage) -> None:
    limiter = build(TokenBucket, clock, storage, capacity=10)

    def hammer(tenant: int) -> int:
        return sum(limiter.allow(f"tenant-{tenant}").allowed for _ in range(20))

    with ThreadPoolExecutor(max_workers=8) as pool:
        allowed = list(pool.map(hammer, range(8)))
    assert allowed == [10] * 8 and len(storage) == 8

Freezing the clock is what makes the concurrency assertion sharp: with no refill possible, any number above the capacity is a lost update rather than time passing. The sibling test proves the other half — eight tenants on eight keys each get exactly their own limit, so striping does not leak budget between callers.

The rest cover: the token bucket's burst-then-rate behaviour and its exact retry_after; all five algorithms admitting exactly the limit in one burst, via parametrize; the sliding log pointing retry_after at the moment its oldest entry ages out; a cost larger than the limit and a cost of zero; four invalid rule configurations; most-specific rule matching and a missing rule; two rules never sharing a caller's budget; the anonymous IP fallback; the middleware's 429, headers and metrics; cost-weighted spending; a hot reload taking effect without a restart; an unknown algorithm; and idle-key eviction. Run them with uv run pytest code/lld/rate_limiter -q.

45-minute pacing

Minutes	What to do	What to say or write
0–5	Clarify	Per user, IP or API key? Exact or approximate? One process or a fleet? What does a denied caller get back?
5–10	Algorithms	The four-sentence comparison, then commit to token bucket as the default and say why.
10–16	Entities	RateLimitRule, ClientKey, Decision, RateLimiter, Storage on the board. Name the split: rules configure, limiters decide, storage keeps.
16–26	Code the limiter	Decision first, then TokenBucket._decide. Say "refill from elapsed time, no timer thread" while writing it.
26–34	Storage and concurrency	The read-modify-write race, why apply takes a closure, why stripes, and why each stripe owns its dict.
34–40	Middleware and rules	The pipeline stage, most-specific matching, hot reload by swapping an immutable tuple.
40–45	Distribution	Redis with a Lua script, local-then-global, fail-open versus fail-closed, and hand off to the HLD version.

Related

• Rate limiting — the algorithms in system-design terms, placement and the accuracy trade
• Design a distributed rate limiter — the same problem once the counter has to be shared by a fleet
• Strategy — the pattern behind the five interchangeable algorithms
• Pipeline and Middleware — the stage-and-next-handler shape this middleware follows
• Factory Method — the registry idiom behind LimiterFactory
• Concurrency for LLD in Python — striped locks, copy-on-write configuration and read-modify-write races
• Primary sources: RFC 6585 (HTTP 429) and RFC 9110 for Retry-After; the Redis INCR/Lua rate-limiting patterns in the official Redis documentation