Problem 69 · Object & System Design

Rate Limiter: Token Bucket and Sliding-Window Log

MediumClassic DSA & systems

01Problem

Implement two per-client rate limiters with allow(client_id) -> bool: a token bucket (capacity, refill rate per second) and a sliding-window log (max N requests in any trailing W seconds). Clock must be injectable.

02Key insight

Token bucket stores just (tokens, last_refill) and refills lazily on each call, allowing bursts up to capacity. Sliding-window log is exact but O(N) memory per client; a deque lets you evict expired timestamps from the left in amortized O(1).

03Solution

solution.pypython37 lines

import time
from collections import defaultdict, deque
from typing import Callable


class TokenBucketLimiter:
    def __init__(self, capacity: float, refill_per_sec: float,
                 now: Callable[[], float] = time.monotonic) -> None:
        self.capacity, self.rate, self.now = capacity, refill_per_sec, now
        self.state: dict[str, tuple[float, float]] = {}  # client -> (tokens, last_ts)

    def allow(self, client_id: str, cost: float = 1.0) -> bool:
        t = self.now()
        tokens, last = self.state.get(client_id, (self.capacity, t))
        tokens = min(self.capacity, tokens + (t - last) * self.rate)
        allowed = tokens >= cost
        if allowed:
            tokens -= cost
        self.state[client_id] = (tokens, t)
        return allowed


class SlidingWindowLogLimiter:
    def __init__(self, limit: int, window_sec: float,
                 now: Callable[[], float] = time.monotonic) -> None:
        self.limit, self.window, self.now = limit, window_sec, now
        self.logs: defaultdict[str, deque[float]] = defaultdict(deque)

    def allow(self, client_id: str) -> bool:
        t = self.now()
        log = self.logs[client_id]
        while log and log[0] <= t - self.window:
            log.popleft()
        if len(log) < self.limit:
            log.append(t)
            return True
        return False

Try it first.

TimeO(1) bucket; amortized O(1) log

SpaceO(clients) bucket; O(clients * limit) log

tests.py · 4 passing

tests.pypython11 lines

clock = [0.0]
tb = TokenBucketLimiter(capacity=2, refill_per_sec=1, now=lambda: clock[0])
assert [tb.allow("a") for _ in range(3)] == [True, True, False]
assert tb.allow("b")                  # independent client
clock[0] = 1.0
assert tb.allow("a") and not tb.allow("a")
clock[0] = 0.0
sw = SlidingWindowLogLimiter(limit=2, window_sec=10, now=lambda: clock[0])
assert sw.allow("a") and sw.allow("a") and not sw.allow("a")
clock[0] = 9.9; assert not sw.allow("a")
clock[0] = 10.0; assert sw.allow("a")

04Staff-level follow-up

Distributed across many API servers? Keep state in Redis with an atomic Lua script per check (or GCRA), accept slight over-admission via local pre-allocation of tokens to cut round trips.