Design a rate limiter class

01The design, in the six-step framework

1. Requirements and use cases

In a mock run by a Google engineer the signature was RateLimiter(N, T).isAllowed(clientId). One global mutex was rejected as fully serial; a per-client mutex passed; then came a global cap and a question about crashes. This solution follows that path. The distributed version is a separate problem in system design.

• A service asks whether a client may proceed, and at most N requests per T milliseconds are allowed per client.
• Different clients are independent.
• An operator chooses the algorithm: sliding window log, sliding window counter or token bucket.
• Many threads call concurrently; the limit is never exceeded.
• Follow-up: a global cap M across all clients.
• Follow-up: state for idle clients is reclaimed.

Out of scope: sharing state across processes, and retry advice beyond a simple hint. Non-functional constraints: is_allowed is O(1) for bucket and counter; requests for different clients do not contend on one lock; the clock is injected; and adding an algorithm does not touch the limiter, the extension named in advance.

2. Core entities and relationships

• RateLimiter: the facade; finds the client's state and asks it for a decision.
• ClientRegistry: maps a client id to its algorithm state, striped so creation does not serialise everything.
• Algorithm: the strategy interface; try_acquire(now, cost) and refund(cost).
• TokenBucket, SlidingLog, SlidingCounter: per-client implementations, each with its own lock.
• GlobalCap: a token bucket for all traffic, checked after the client check.
• Clock: now() in seconds.

The limiter composes the registry and aggregates a factory and clock. Each client's Algorithm owns its own lock, which is what stops one busy client from delaying another. See extensibility points: the factory is the single door for a new algorithm.

3. Interfaces and key methods

from typing import Callable, Optional, Protocol


class Clock(Protocol):
    def now(self) -> float: ...


class Algorithm(Protocol):
    def try_acquire(self, now: float, cost: float = 1.0) -> bool:
        """Atomic check-and-take. Returns False and changes nothing if over limit."""

    def refund(self, cost: float = 1.0) -> None:
        """Give back a grant that a later check refused."""


class RateLimiter:
    def __init__(self, factory: Callable[[], Algorithm], clock: Clock,
                 global_cap: Optional[Algorithm] = None) -> None: ...

    def is_allowed(self, client_id: str) -> bool:
        """Client limit first, then the global cap; thread-safe."""

Invariants: try_acquire is atomic, so check and update cannot interleave; a false answer consumes nothing from the client; and a client has one Algorithm at a time, created once even when two threads race on first use.

4. Design patterns and the choices behind them

Strategy over the algorithm

The force is that algorithms differ in accuracy, memory and burst behaviour, and the right one depends on the caller. Strategy keeps those details out of the limiter. The trade-offs are the discussion an interviewer wants, and they come from sliding windows and counters:

One lock per client

The force is the rejected answer from the mock: a single mutex makes the limiter serial, even though clients share nothing. With a lock inside each Algorithm, contention only happens among requests from the same client, which is the case that must serialise. The registry needs care too: two threads seeing a new client must not create two states, or both would pass. Striping the registry with setdefault under a stripe lock solves it. A concurrent map alone does not make check-and-take atomic; see thread-safe collections.

Refill on demand, with an injected clock

The force is avoiding a timer per client. A token bucket can compute what it would have gained since the last call, with one multiplication. A background refiller would be a thread looping over millions of clients. Using a monotonic clock avoids trouble when the wall clock jumps, and the injection lets a test advance time by hand, per time and expiry.

tokens = min(capacity, tokens + (now − last) × rate), where rate = N / T

Order of the two checks

The global cap is checked after the client check, so a client over its own limit does not burn global budget. If the global check fails, the client's token is refunded, so a global shortage does not punish the client twice. The machine-coding graders look for this kind of edge case, as described in the rubric.

5. Core flow and code

Flow for is_allowed("c1") on a limit of 2 per second:

1. The limiter reads the clock and gets the client's bucket from the registry, creating it if needed.
2. The bucket takes its lock, refills by the elapsed time, and sees 1.4 tokens.
3. It takes one, leaving 0.4, and releases the lock.
4. If a global cap exists, it is tried; on failure the token is returned and the answer is false.

import threading
import time
from collections import deque
from typing import Callable, Deque, Dict, List, Optional, Protocol


class Algorithm(Protocol):
    def try_acquire(self, now: float, cost: float = 1.0) -> bool: ...
    def refund(self, cost: float = 1.0) -> None: ...


class FakeClock:
    def __init__(self) -> None:
        self.t = 0.0

    def now(self) -> float:
        return self.t


class MonotonicClock:
    def now(self) -> float:
        return time.monotonic()


class TokenBucket:
    def __init__(self, n: int, period: float) -> None:
        self.capacity = float(n)
        self.rate = n / period
        self.tokens = float(n)
        self.last: Optional[float] = None
        self.lock = threading.Lock()

    def try_acquire(self, now: float, cost: float = 1.0) -> bool:
        with self.lock:
            if self.last is not None:
                elapsed = max(0.0, now - self.last)
                self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last = now
            if self.tokens + 1e-9 < cost:
                return False
            self.tokens -= cost
            return True

    def refund(self, cost: float = 1.0) -> None:
        with self.lock:
            self.tokens = min(self.capacity, self.tokens + cost)


class SlidingLog:
    def __init__(self, n: int, period: float) -> None:
        self.n, self.period = n, period
        self.times: Deque[float] = deque()
        self.lock = threading.Lock()

    def try_acquire(self, now: float, cost: float = 1.0) -> bool:
        with self.lock:
            while self.times and self.times[0] <= now - self.period:
                self.times.popleft()
            if len(self.times) >= self.n:
                return False
            self.times.append(now)
            return True

    def refund(self, cost: float = 1.0) -> None:
        with self.lock:
            if self.times:
                self.times.pop()


class SlidingCounter:
    """Two fixed windows, the previous one weighted by how much of it still overlaps."""

    def __init__(self, n: int, period: float) -> None:
        self.n, self.period = n, period
        self.prev, self.curr = 0, 0
        self.start = 0.0
        self.lock = threading.Lock()

    def try_acquire(self, now: float, cost: float = 1.0) -> bool:
        with self.lock:
            windows = int((now - self.start) // self.period)
            if windows >= 1:
                self.prev = self.curr if windows == 1 else 0
                self.curr = 0
                self.start += windows * self.period
            overlap = 1 - (now - self.start) / self.period
            if self.prev * overlap + self.curr + cost > self.n:
                return False
            self.curr += 1
            return True

    def refund(self, cost: float = 1.0) -> None:
        with self.lock:
            self.curr = max(0, self.curr - 1)


class ClientRegistry:
    def __init__(self, factory: Callable[[], Algorithm], stripes: int = 32) -> None:
        self.factory = factory
        self.maps: List[Dict[str, Algorithm]] = [{} for _ in range(stripes)]
        self.locks = [threading.Lock() for _ in range(stripes)]

    def get(self, client_id: str) -> Algorithm:
        i = hash(client_id) % len(self.maps)
        with self.locks[i]:
            state = self.maps[i].get(client_id)
            if state is None:
                state = self.maps[i][client_id] = self.factory()
            return state


class RateLimiter:
    def __init__(self, factory: Callable[[], Algorithm], clock=None,
                 global_cap: Optional[Algorithm] = None) -> None:
        self.registry = ClientRegistry(factory)
        self.clock = clock or MonotonicClock()
        self.global_cap = global_cap

    def is_allowed(self, client_id: str) -> bool:
        now = self.clock.now()
        state = self.registry.get(client_id)
        if not state.try_acquire(now):
            return False
        if self.global_cap and not self.global_cap.try_acquire(now):
            state.refund()
            return False
        return True


if __name__ == "__main__":
    for make in (lambda: TokenBucket(2, 1.0), lambda: SlidingLog(2, 1.0), lambda: SlidingCounter(2, 1.0)):
        clock = FakeClock()
        limiter = RateLimiter(make, clock)
        assert limiter.is_allowed("a") and limiter.is_allowed("a") and not limiter.is_allowed("a")
        assert limiter.is_allowed("b")             # clients are independent
        clock.t = 2.5
        assert limiter.is_allowed("a")

    clock = FakeClock()
    capped = RateLimiter(lambda: TokenBucket(5, 1.0), clock, global_cap=TokenBucket(2, 1.0))
    assert capped.is_allowed("a") and capped.is_allowed("b") and not capped.is_allowed("c")
    assert capped.registry.get("c").tokens == 5.0  # refunded

    shared = RateLimiter(lambda: TokenBucket(100, 1000.0), FakeClock())
    passed: List[bool] = []

    def worker() -> None:
        for _ in range(100):
            passed.append(shared.is_allowed("same"))

    threads = [threading.Thread(target=worker) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert sum(passed) == 100                      # never exceeded

6. Extensions, concurrency and what interviewers probe

• Why not one mutex. It is correct and serial. Say that first, then move to per-client state; the mock rated this progression highly.
• Idle clients. A registry that never forgets is a memory leak. A periodic sweep, added as a method on the registry, removes clients idle for longer than T, because a full bucket gives the same answer as a new one.
• Clock going backwards. max(0, ...) guards the elapsed time; a monotonic clock avoids the case entirely.
• Cost per request. The cost argument already supports weights; the billing-token limiter takes this further.
• Crash behaviour. State is in memory, so a restart gives everyone a fresh allowance. If that matters, say whether to fail open or fail closed.
• Common mistake. Reading the tokens, deciding, then writing without one lock, so two threads each take the last token.

02Where this question was reported

• Hello Interview · Hello Interview breakdown (Hard, premium)
• LLD Mastery · Google-tagged (Medium); token bucket, sliding window
• LLD Mastery · Google-tagged with Stripe/Amazon/Meta/Cloudflare; thread safety
• Prepfully · Prepfully: 'rate limiter across users or services' among recently reported Google coding questions
• PracHub · PracHub Google-tagged variant; also /design-a-distributed-rate-limiter
• systemdesign.academy · systemdesign.academy solved
• thita.ai · thita.ai LLD sheet, foundational
• interviewing.io · RateLimiter(N,T).isAllowed(clientId); one mutex rejected as fully serial -> per-client mutex; added global cap M; crash/backoff; Strong Hire
• LeetCode Discuss · Rate-limiter design question in onsite coding slate
• Medium · 'Modified rate limiting (design problem)': three hash maps, infinite request stream, async API calls, pushed to O(N); candidate rejected
• aceoffer.app · 'Rate limiter for AI output / billing tokens' (2 reports) - system-design framing
• PracHub · Google Senior+ code-review round on a Go token-bucket limiter middleware