Courses › Low-Level Design › Problem Walkthroughs

Implement a Rate Limiter Class

Pluggable token bucket and sliding window limiters with an injected clock and locks.

"Implement a rate limiter class" is the LLD version of the system design question. There's no Redis cluster here. The question is whether you can write a correct, thread-safe, pluggable limiter in one process. It combines the Strategy pattern, careful time handling and concurrency in one compact problem.

Requirements

• allow(client_id) -> bool: decide whether this request goes through.
• Limits are configured per client or per tier, for example 100 requests per minute.
• Several algorithms are supported, and new ones can be added without changing callers.
• It's safe to call from many threads at once.
• Memory stays bounded as clients come and go.

Design

RateLimiter (interface)        allow(key) -> bool
  TokenBucketLimiter           capacity, refill_per_sec
  SlidingWindowLogLimiter      limit, window_sec
  FixedWindowLimiter           limit, window_sec
RateLimiterRegistry            limiter per tier; for_client(client_id) -> RateLimiter
Clock                          injected; returns monotonic seconds

Two decisions are worth saying out loud:

• Inject the clock. Tests can then advance time instantly, and you use a monotonic clock (time.monotonic()) so wall-clock adjustments can't hand out or take away tokens.
• Keep per-key state inside the limiter. One limiter object serves many keys, with a small state record for each key.

Token bucket

Each key has a bucket that holds up to capacity tokens and refills continuously. A request spends one token. You don't need a background thread to refill: compute the refill lazily from the time elapsed since the last request.

import threading
import time
from abc import ABC, abstractmethod


class RateLimiter(ABC):
    @abstractmethod
    def allow(self, key: str) -> bool: ...


class TokenBucketLimiter(RateLimiter):
    def __init__(self, capacity, refill_per_sec, clock=time.monotonic):
        self.capacity = capacity
        self.rate = refill_per_sec
        self.clock = clock
        self._buckets = {}                 # key -> [tokens, last_refill]
        self._lock = threading.Lock()

    def allow(self, key):
        now = self.clock()
        with self._lock:
            tokens, last = self._buckets.get(key, (self.capacity, now))
            tokens = min(self.capacity, tokens + (now - last) * self.rate)
            allowed = tokens >= 1
            if allowed:
                tokens -= 1
            self._buckets[key] = (tokens, now)
            return allowed

Sliding window log

This is exact: allow a request if fewer than limit requests from this key happened in the last window seconds. Store the timestamps in a deque and drop the expired ones from the front.

from collections import defaultdict, deque


class SlidingWindowLogLimiter(RateLimiter):
    def __init__(self, limit, window_sec, clock=time.monotonic):
        self.limit, self.window, self.clock = limit, window_sec, clock
        self._logs = defaultdict(deque)
        self._lock = threading.Lock()

    def allow(self, key):
        now = self.clock()
        with self._lock:
            log = self._logs[key]
            while log and log[0] <= now - self.window:
                log.popleft()
            if len(log) < self.limit:
                log.append(now)
                return True
            return False

Comparing the algorithms

[!INTERVIEW] Test with a fake clock Show one test: t = [0.0]; lim = TokenBucketLimiter(2, 1, clock=lambda: t[0]). Two calls to allow succeed, the third fails, then t[0] = 1.0 and the next call succeeds. It proves the logic and shows the value of injecting the clock, both in under a minute.

Per-client configuration

class RateLimiterRegistry:
    def __init__(self, tiers, default_tier):
        self.tiers = tiers                    # {"free": TokenBucketLimiter(...), "pro": ...}
        self.default = default_tier
        self.client_tier = {}

    def allow(self, client_id):
        tier = self.client_tier.get(client_id, self.default)
        return self.tiers[tier].allow(client_id)

Follow-ups

• Contention: one global lock serializes every client. Use a lock per key, or striped locks (hash each key to one of N locks), so independent clients don't block each other.
• Memory growth: state for clients that have gone quiet piles up. Evict buckets that have been idle longer than the time needed to refill completely, either lazily or with a periodic sweep. An LRU map with a size cap also works. See Design an LRU Cache.
• Waiting instead of rejecting: add acquire(key, timeout) that sleeps until the next token is due: (1 - tokens) / rate seconds.
• Weighted requests: allow(key, cost=5) for expensive endpoints.
• Several processes: the in-memory state no longer works. Move the counters to Redis with an atomic Lua script. That's the jump to system design.

[!TIP] Practice Implement a Rate Limiter Class · Design a Circuit Breaker · Concurrency & Thread Safety