Low-level design
Part 2 of 3 · LLM Gateway Rate LimiterRate Limiter - Token Bucket + Sliding Window Solution & Tests
Runnable TokenBucket, SlidingWindowLog, and GatewayLimiter, with a fake clock and an 80-thread burst flood.
- 1Gist
- 2Maps
- 3Q&A
- 4Sandbox
Voice readout needs Web Speech Synthesis in this browser.
Question ladder
L1
How do you run the tests?
Answer
From the rate_limiter directory, python3 test_limiter.py -v. Five tests pass.
L2
What does FakeClock do?
Answer
It returns self.t and advance adds seconds. Nothing sleeps.
L3
What does test_allows_then_blocks prove?
Answer
Capacity 2 blocks the third call, and one second of refill allows one more.
L4
What does test_keys_isolated prove?
Answer
Spending tenant x does not spend tenant y.
L5
What does test_window prove?
Answer
Two events fill a limit of 2. Advancing 10.1 seconds drops them and allows another.
L6
What does test_burst_then_rpm prove?
Answer
With refill 0 and burst 2, the third call is BURST_LIMIT while RPM is still 5.
L7
What does the flood prove?
Answer
Eighty threads, burst 50, refill 0. Exactly 50 allows and 30 denies.
Failure modes
Sleep in the test
A slow host changes retry_after_ms. The fake clock does not.
Shared default clock
time.monotonic is the default. The tests that care about edges pass a FakeClock.
Oversell
Without the locks, the flood can exceed the burst. The test expects 50.
Misconceptions
The window test needs a 60 second window.
SlidingWindowLog takes the window length. The test uses 10 seconds.
retry_after_ms is always positive.
A refill rate of 0 returns 0 milliseconds from the bucket, because there is no time at which the deficit clears.
The sliding window stores a weighted previous count.
SlidingWindowLog stores timestamps in a deque. The weighted counter is the concept page.
Interviewer traps
Paste a shorter stub than the tests import.
The module and the unittest file on this page are the full source.
Assert a particular interleaving of the 80 threads.
The assertion is the count of allows, which is the burst.
Design scenario
Same prompt for every reader.
Requirements
Stdlib only. Injected clock. Isolated keys. Window expiry. BURST_LIMIT. Concurrent cap.
Traffic / scale
Eighty threads, one tenant, burst 50, refill 0.
Latency
No sleep in the tests.
Consistency
The number of successful allows equals the burst.
Availability
A deny is a Decision with a reason. It is not an exception.
Failure assumptions
- The third call on a capacity-2 bucket happens before the clock advances.
- Two tenants share a limiter with capacity 1.
Constraints
- Do not read time.monotonic in the edge tests.
- Do not let the flood exceed the burst.
Prompt
Implement the three classes and the five tests.
API
What does Decision contain when the bucket blocks?
Data
Where are tokens and timestamps stored?
Architecture
Which locks does GatewayLimiter.allow acquire?
Where the clock comes from
Prefer
Pass a fake clock
advance moves the tests. Refill and the window edge stay exact.
- No sleep.
- retry_after_ms is computed, not measured.
- Keys still have separate state.
Alternative
Call the clock inside the test
The same asserts become timing-sensitive.
- A slow host changes the edge.
- The window cutoff is hard to hit.
- The flood is no longer only about the lock.
Put limiter.py and test_limiter.py on the path and run python3 test_limiter.py -v. You want the bucket to block then refill, isolated keys, a window that opens after 10.1 seconds, BURST_LIMIT on the third gateway call, and exactly 50 allows from 80 threads.
Overview
This page is the runnable limiter. TokenBucket, SlidingWindowLog, and GatewayLimiter live in limiter.py. The tests pass an explicit clock wherever an edge matters. Five tests cover burst, isolation, the window, composition, and the concurrent cap.
Run
python3 test_limiter.py -vSolution
"""LLM Gateway Rate Limiter - token bucket + sliding window.
Sandbox: python3 -c "from limiter import TokenBucket, SlidingWindowLog; ..."
Tests: python3 test_limiter.py
stdlib only.
"""
from __future__ import annotations
import threading
import time
from collections import defaultdict, deque
from dataclasses import dataclass
from typing import Deque, Optional
@dataclass(frozen=True)
class Decision:
allowed: bool
remaining: float
retry_after_ms: int = 0
reason: str = "OK"
class TokenBucket:
"""Per-key token bucket with fractional tokens and thread safety."""
def __init__(self, capacity: float, refill_per_sec: float, clock=time.monotonic) -> None:
if capacity <= 0 or refill_per_sec < 0:
raise ValueError("INVALID_CAPACITY")
self.capacity = float(capacity)
self.refill_per_sec = float(refill_per_sec)
self._clock = clock
self._lock = threading.RLock()
self._tokens: dict[str, float] = {}
self._updated: dict[str, float] = {}
def _refill(self, key: str, now: float) -> None:
last = self._updated.get(key, now)
tokens = self._tokens.get(key, self.capacity)
elapsed = max(0.0, now - last)
tokens = min(self.capacity, tokens + elapsed * self.refill_per_sec)
self._tokens[key] = tokens
self._updated[key] = now
def allow(self, key: str, cost: float = 1.0) -> Decision:
if cost <= 0:
raise ValueError("INVALID_COST")
with self._lock:
now = self._clock()
self._refill(key, now)
tokens = self._tokens[key]
if tokens >= cost:
self._tokens[key] = tokens - cost
return Decision(True, self._tokens[key], 0, "OK")
need = cost - tokens
retry_ms = int((need / self.refill_per_sec) * 1000) if self.refill_per_sec > 0 else 0
return Decision(False, tokens, retry_ms, "RATE_LIMITED")
class SlidingWindowLog:
"""Per-key sliding window using event timestamps (exact count)."""
def __init__(self, limit: int, window_sec: float, clock=time.monotonic) -> None:
if limit < 1 or window_sec <= 0:
raise ValueError("INVALID_WINDOW")
self.limit = limit
self.window_sec = float(window_sec)
self._clock = clock
self._lock = threading.RLock()
self._events: dict[str, Deque[float]] = defaultdict(deque)
def allow(self, key: str) -> Decision:
with self._lock:
now = self._clock()
q = self._events[key]
cutoff = now - self.window_sec
while q and q[0] <= cutoff:
q.popleft()
if len(q) < self.limit:
q.append(now)
return Decision(True, float(self.limit - len(q)), 0, "OK")
retry_ms = int((q[0] + self.window_sec - now) * 1000)
return Decision(False, 0.0, max(0, retry_ms), "RATE_LIMITED")
class GatewayLimiter:
"""Compose RPM sliding window + burst token bucket for an LLM gateway."""
def __init__(
self,
rpm: int = 60,
burst: float = 10.0,
burst_refill_per_sec: float = 5.0,
clock=time.monotonic,
) -> None:
self.rpm = SlidingWindowLog(rpm, 60.0, clock=clock)
self.burst = TokenBucket(burst, burst_refill_per_sec, clock=clock)
self._lock = threading.RLock()
def allow(self, tenant: str, tokens: float = 1.0) -> Decision:
with self._lock:
d1 = self.rpm.allow(tenant)
if not d1.allowed:
return Decision(False, d1.remaining, d1.retry_after_ms, "RPM_LIMIT")
d2 = self.burst.allow(tenant, tokens)
if not d2.allowed:
return Decision(False, d2.remaining, d2.retry_after_ms, "BURST_LIMIT")
return Decision(True, min(d1.remaining, d2.remaining), 0, "OK")Tests
"""Tests for LLM gateway rate limiter."""
from __future__ import annotations
import threading
import unittest
from limiter import GatewayLimiter, SlidingWindowLog, TokenBucket
class FakeClock:
def __init__(self) -> None:
self.t = 0.0
def __call__(self) -> float:
return self.t
def advance(self, s: float) -> None:
self.t += s
class TokenBucketTests(unittest.TestCase):
def test_allows_then_blocks(self):
clock = FakeClock()
b = TokenBucket(2, 1, clock=clock)
self.assertTrue(b.allow("a").allowed)
self.assertTrue(b.allow("a").allowed)
d = b.allow("a")
self.assertFalse(d.allowed)
self.assertEqual(d.reason, "RATE_LIMITED")
clock.advance(1.0)
self.assertTrue(b.allow("a").allowed)
def test_keys_isolated(self):
b = TokenBucket(1, 0)
self.assertTrue(b.allow("x").allowed)
self.assertTrue(b.allow("y").allowed)
self.assertFalse(b.allow("x").allowed)
class SlidingWindowTests(unittest.TestCase):
def test_window(self):
clock = FakeClock()
w = SlidingWindowLog(2, 10, clock=clock)
self.assertTrue(w.allow("k").allowed)
self.assertTrue(w.allow("k").allowed)
self.assertFalse(w.allow("k").allowed)
clock.advance(10.1)
self.assertTrue(w.allow("k").allowed)
class GatewayTests(unittest.TestCase):
def test_burst_then_rpm(self):
clock = FakeClock()
g = GatewayLimiter(rpm=5, burst=2, burst_refill_per_sec=0, clock=clock)
self.assertTrue(g.allow("t").allowed)
self.assertTrue(g.allow("t").allowed)
d = g.allow("t")
self.assertFalse(d.allowed)
self.assertEqual(d.reason, "BURST_LIMIT")
def test_concurrent_unique_success_bound(self):
g = GatewayLimiter(rpm=100, burst=50, burst_refill_per_sec=0)
ok = []
lock = threading.Lock()
def worker():
d = g.allow("tenant")
with lock:
ok.append(d.allowed)
threads = [threading.Thread(target=worker) for _ in range(80)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(sum(1 for x in ok if x), 50)
self.assertEqual(sum(1 for x in ok if not x), 30)
if __name__ == "__main__":
unittest.main()What the five tests pin
Diagram 1. The same order as the hub. The tests pin refill, the window, and the burst cap.
- 1
Refill
Capacity 2 blocks, then one second of the fake clock allows one more token. - 2
Isolation
Tenant y still has its own token after tenant x is empty. - 3
Window
Two events fill the log. Advancing past the window drops them. - 4
Compose
Burst 2 and RPM 5. The third call is BURST_LIMIT. - 5
Flood
Eighty threads and burst 50. Exactly 50 allows.
Decisions
- 1
1. Request tenant and cost
- next2. Sliding window RPM check
- 2
2. Sliding window RPM check
- next3. Under RPM?
- ?
3. Under RPM?
- No4. Failure: RPM_LIMIT
- Yes5. Token bucket burst check
- 4
4. Failure: RPM_LIMIT
- 5
5. Token bucket burst check
- next6. Tokens enough?
- ?
6. Tokens enough?
- No7. Failure: BURST_LIMIT
- Yes8. Success: deduct and allow
- 7
7. Failure: BURST_LIMIT
- 8
8. Success: deduct and allow
Lesson map
Rate Limiter - Token Bucket + Sliding Window Solution & Tests
Runnable TokenBucket, SlidingWindowLog, and GatewayLimiter, with a fake clock and an 80-thread burst flood.
Architecture. Architecture
Select a node to see why it exists, or an edge to see the protocol, direction, effect, and consequence.
Mermaid export
flowchart TB a["1. Request tenant and cost"] b["2. Sliding window RPM check"] c["3. Under RPM?"] d["4. Failure: RPM_LIMIT"] e["5. Token bucket burst check"] f["6. Tokens enough?"] g["7. Failure: BURST_LIMIT"] h["8. Success: deduct and allow"] a -->|continues| b b -->|continues| c c -->|No| d c -->|Yes| e e -->|continues| f f -->|No| g f -->|Yes| h
Concepts used, learn more
Read the underlying idea on its own study page. This lesson applies it. It does not replace those pages.
- Rate Limiting: Token Bucket, Leaky Bucket & Sliding Window
- Token Bucket vs Leaky Bucket vs Sliding Window
- Redis + Lua Atomic Rate Limiters
- Distributed Rate Limits Across Gateways
- Mutexes, Condition Variables, Deadlocks & Happens-Before
- Mutex vs RWLock
- When Locks Win — Contention, Fairness & Hybrid Designs
- Atomics vs Locks
- Low-Level Design Under Time — Interfaces, State & Tradeoffs
Interview Q&A
Why is the default clock time.monotonic?
Answer
Production calls need a monotonic clock. Tests that care about edges replace it. The isolation test can keep the default because it never advances time.
What does a refill rate of 0 do to retry_after_ms?
Answer
The bucket cannot name a future time when the deficit clears, so the retry is 0. The Decision is still denied.
Why is the window cutoff a less-or-equal compare?
Answer
An event at exactly now minus window_sec is outside the open window and is popped. The test advances 10.1 seconds so the two events are clearly gone.
Why 50 allows, not 80 or 100?
Answer
The gateway burst is 50 and refill is 0. RPM is 100, so the bucket is the tighter cap. The lock makes the successful count equal that cap.
Which import does the test need?
Answer
from limiter import GatewayLimiter, SlidingWindowLog, TokenBucket.
Does this page replace the rate-limiting series?
Answer
No. Those pages compare the algorithms and the distributed design. This page is the coding-round module.
Related
The series pager also walks these pages.