Answer A: Google Gemini 3.8 Flash
"""Production-ready, thread-safe rate limiter with sliding window and burst credits."""
from future import annotations
import collections
import dataclasses
import hashlib
import math
import threading
import time
import unittest
from typing import Callable, Deque, Dict, List, NamedTuple, Optional
@dataclasses.dataclass(frozen=True, slots=True)
class Decision:
allowed: bool
remaining: int
retry_after: float
limit: int
class _WindowEntry(NamedTuple):
timestamp: float
cost: int
class _KeyBucket:
slots = (
"entries",
"window_usage",
"burst_credits",...
Show Full Answer ▼
"""Production-ready, thread-safe rate limiter with sliding window and burst credits."""
from future import annotations
import collections
import dataclasses
import hashlib
import math
import threading
import time
import unittest
from typing import Callable, Deque, Dict, List, NamedTuple, Optional
@dataclasses.dataclass(frozen=True, slots=True)
class Decision:
allowed: bool
remaining: int
retry_after: float
limit: int
class _WindowEntry(NamedTuple):
timestamp: float
cost: int
class _KeyBucket:
slots = (
"entries",
"window_usage",
"burst_credits",
"last_update",
"last_active",
)
def __init__(self, initial_credits: float, now: float) -> None:
self.entries: Deque[_WindowEntry] = collections.deque()
self.window_usage: int = 0
self.burst_credits: float = float(initial_credits)
self.last_update: float = now
self.last_active: float = now
def clean_window(self, cutoff: float) -> None:
while self.entries and self.entries[0].timestamp <= cutoff:
expired = self.entries.popleft()
self.window_usage -= expired.cost
if self.window_usage < 0:
self.window_usage = 0
def refill_credits(self, now: float, rate: float, max_credits: float) -> None:
if now > self.last_update:
if rate > 0.0 and self.burst_credits < max_credits:
added = (now - self.last_update) * rate
self.burst_credits = min(max_credits, self.burst_credits + added)
self.last_update = now
def is_idle(self, now: float, window_seconds: float, max_credits: float) -> bool:
return (
len(self.entries) == 0
and math.isclose(self.burst_credits, max_credits, abs_tol=1e-9)
and (now - self.last_active) >= window_seconds
)
class _Shard:
slots = ("lock", "buckets", "access_order")
def __init__(self) -> None:
self.lock = threading.Lock()
self.buckets: Dict[str, _KeyBucket] = {}
self.access_order: collections.OrderedDict[str, None] = collections.OrderedDict()
class RateLimiter:
"""High-throughput, sliding-window rate limiter with burst credit handling."""
def __init__(
self,
limit: int,
window_seconds: float,
burst_credits: int = 0,
clock: Callable[[], float] = time.monotonic,
num_shards: int = 256,
) -> None:
if limit < 0:
raise ValueError("limit must be non-negative")
if window_seconds <= 0:
raise ValueError("window_seconds must be positive")
if burst_credits < 0:
raise ValueError("burst_credits must be non-negative")
if num_shards <= 0 or (num_shards & (num_shards - 1)) != 0:
raise ValueError("num_shards must be a positive power of two")
self.limit = int(limit)
self.window_seconds = float(window_seconds)
self.burst_credits = int(burst_credits)
self.clock = clock
self._rate = (self.limit / self.window_seconds) if self.limit > 0 else 0.0
self._num_shards = num_shards
self._mask = num_shards - 1
self._shards: List[_Shard] = [_Shard() for _ in range(num_shards)]
def _get_shard(self, key: str) -> _Shard:
digest = hashlib.blake2b(key.encode("utf-8"), digest_size=8).digest()
index = int.from_bytes(digest, byteorder="little") & self._mask
return self._shards[index]
def _evict_idle_sample(self, shard: _Shard, now: float, sample_size: int = 4) -> None:
# Check a small constant number of oldest accessed keys to amortize O(1) eviction
for _ in range(sample_size):
if not shard.access_order:
break
oldest_key = next(iter(shard.access_order))
bucket = shard.buckets.get(oldest_key)
if bucket is None:
shard.access_order.pop(oldest_key, None)
continue
bucket.clean_window(now - self.window_seconds)
bucket.refill_credits(now, self._rate, float(self.burst_credits))
if bucket.is_idle(now, self.window_seconds, float(self.burst_credits)):
del shard.buckets[oldest_key]
shard.access_order.pop(oldest_key, None)
else:
# Oldest key is not idle; stop rotation to avoid unnecessary overhead
break
def _compute_retry_after(self, bucket: _KeyBucket, cost: int, now: float) -> float:
if cost > self.limit + self.burst_credits:
return float("inf")
if cost <= 0:
return 0.0
cutoff = now - self.window_seconds
sim_entries = collections.deque(bucket.entries)
sim_usage = bucket.window_usage
sim_credits = bucket.burst_credits
last_t = now
rate = self._rate
max_credits = float(self.burst_credits)
while True:
window_avail = max(0, self.limit - sim_usage)
needed_burst = max(0, cost - window_avail)
if sim_credits >= needed_burst:
return max(0.0, last_t - now)
# If rate > 0, how long until burst credits refill sufficiently without expiry?
dt_refill = (needed_burst - sim_credits) / rate if rate > 0.0 else float("inf")
if not sim_entries:
if rate > 0.0:
return max(0.0, (last_t + dt_refill) - now)
return float("inf")
next_entry = sim_entries[0]
entry_expiry_time = next_entry.timestamp + self.window_seconds
dt_to_expiry = max(0.0, entry_expiry_time - last_t)
if rate > 0.0 and dt_refill <= dt_to_expiry:
return max(0.0, (last_t + dt_refill) - now)
# Fast-forward to the entry's expiration
last_t = max(last_t, entry_expiry_time)
if rate > 0.0:
sim_credits = min(max_credits, sim_credits + dt_to_expiry * rate)
expired = sim_entries.popleft()
sim_usage = max(0, sim_usage - expired.cost)
def allow(self, key: str, cost: int = 1, now: Optional[float] = None) -> Decision:
if cost < 0:
raise ValueError("cost must be non-negative")
current_time = float(self.clock() if now is None else now)
shard = self._get_shard(key)
with shard.lock:
self._evict_idle_sample(shard, current_time)
bucket = shard.buckets.get(key)
if bucket is None:
bucket = _KeyBucket(float(self.burst_credits), current_time)
shard.buckets[key] = bucket
shard.access_order[key] = None
shard.access_order.move_to_end(key)
# Monotonic clamp per key: past timestamps are clamped to last observed time
t = max(current_time, bucket.last_active)
bucket.last_active = t
bucket.clean_window(t - self.window_seconds)
bucket.refill_credits(t, self._rate, float(self.burst_credits))
window_avail = max(0, self.limit - bucket.window_usage)
credits_avail = bucket.burst_credits
total_avail = window_avail + int(credits_avail)
if cost == 0:
return Decision(
allowed=True,
remaining=total_avail,
retry_after=0.0,
limit=self.limit,
)
if cost > self.limit + self.burst_credits:
return Decision(
allowed=False,
remaining=total_avail,
retry_after=float("inf"),
limit=self.limit,
)
needed_burst = max(0, cost - window_avail)
if credits_avail >= needed_burst:
from_window = min(cost, window_avail)
from_burst = needed_burst
if from_window > 0:
bucket.entries.append(_WindowEntry(timestamp=t, cost=from_window))
bucket.window_usage += from_window
bucket.burst_credits -= from_burst
new_window_avail = max(0, self.limit - bucket.window_usage)
new_total_avail = new_window_avail + int(bucket.burst_credits)
return Decision(
allowed=True,
remaining=new_total_avail,
retry_after=0.0,
limit=self.limit,
)
else:
retry_after = self._compute_retry_after(bucket, cost, t)
return Decision(
allowed=False,
remaining=total_avail,
retry_after=retry_after,
limit=self.limit,
)
def snapshot(self, key: str, now: Optional[float] = None) -> Decision:
current_time = float(self.clock() if now is None else now)
shard = self._get_shard(key)
with shard.lock:
bucket = shard.buckets.get(key)
if bucket is None:
return Decision(
allowed=True,
remaining=self.limit + self.burst_credits,
retry_after=0.0,
limit=self.limit,
)
t = max(current_time, bucket.last_active)
bucket.clean_window(t - self.window_seconds)
bucket.refill_credits(t, self._rate, float(self.burst_credits))
window_avail = max(0, self.limit - bucket.window_usage)
remaining = window_avail + int(bucket.burst_credits)
return Decision(
allowed=remaining > 0,
remaining=remaining,
retry_after=0.0 if remaining > 0 else self._compute_retry_after(bucket, 1, t),
limit=self.limit,
)
"""
Design Note (273 words):
This rate limiter implements an exact sliding window combined with a continuous token-bucket burst credit pool.
Data Structure & Accuracy Tradeoff:
Each client key maintains a _KeyBucket containing a deque of (timestamp, cost) tuples tracking consumed window quota, an integer accumulator window_usage, and a float burst_credits. An exact sliding window is chosen over approximate bucketing or leaky-bucket models to guarantee zero boundary error: request timestamps count strictly inside (t - window_seconds, t]. Multi-cost requests (cost > 1) are admitted atomically: available window quota is claimed first, and any deficit is covered by continuous burst credits refilled at limit / window_seconds. If the total available capacity is insufficient, the request is rejected all-or-nothing. Deque memory per active key scales strictly as O(N) where N <= limit, bounded and compact.
Locking Strategy:
To avoid a global contention bottleneck and prevent concurrent requests from serializing the whole gateway, the key space is partitioned across 256 independent shards via Blake2b hashing. Each shard maintains its own threading.Lock, dictionary, and access order. Operations on distinct keys map to independent locks with high probability, minimizing lock contention while fully synchronizing per-key concurrent accesses without deadlocks.
Memory Hygiene & Eviction:
To protect against unbounded memory growth from millions of one-shot keys, each shard maintains an OrderedDict of key access order. On every operation, an amortized O(1) cleanup samples the oldest keys in the shard. If a key's window deque is empty, its burst credits are fully regenerated, and it has been idle for at least window_seconds, it is permanently evicted. Inactive keys naturally expire without background threads.
"""
class RateLimiterTests(unittest.TestCase):
def setUp(self) -> None:
self.current_time = 1000.0
def fake_clock(self) -> float:
return self.current_time
def test_exact_boundary_expiry(self) -> None:
limiter = RateLimiter(limit=2, window_seconds=1.0, burst_credits=0, clock=self.fake_clock)
d1 = limiter.allow("client-1", cost=1)
self.assertTrue(d1.allowed)
self.assertEqual(d1.remaining, 1)
self.current_time += 0.5
d2 = limiter.allow("client-1", cost=1)
self.assertTrue(d2.allowed)
self.assertEqual(d2.remaining, 0)
# At exact boundary t = 1001.0, window (1000.0, 1001.0] still includes t=1000.0
self.current_time = 1001.0
d3 = limiter.allow("client-1", cost=1)
self.assertFalse(d3.allowed)
# Just past boundary t = 1001.000001, t=1000.0 is evicted
self.current_time = 1001.000001
d4 = limiter.allow("client-1", cost=1)
self.assertTrue(d4.allowed)
def test_burst_credit_refill(self) -> None:
# limit=10, window=10.0 -> rate=1.0 credit/sec. burst_credits=5
limiter = RateLimiter(limit=10, window_seconds=10.0, burst_credits=5, clock=self.fake_clock)
# Consume all 10 window + 5 burst credits
d = limiter.allow("burst-client", cost=15)
self.assertTrue(d.allowed)
self.assertEqual(d.remaining, 0)
# Immediate next request rejected
d_fail = limiter.allow("burst-client", cost=1)
self.assertFalse(d_fail.allowed)
# Advance 2.5s -> 2.5 credits refilled (integer remaining is 2)
self.current_time += 2.5
snap = limiter.snapshot("burst-client", now=self.current_time)
self.assertEqual(snap.remaining, 2)
# Advance another 2.5s -> total 5.0 credits refilled (capped at burst_credits=5)
self.current_time += 2.5
d_burst = limiter.allow("burst-client", cost=5)
self.assertTrue(d_burst.allowed)
def test_all_or_nothing_multi_cost(self) -> None:
limiter = RateLimiter(limit=5, window_seconds=10.0, burst_credits=2, clock=self.fake_clock)
# Allowed total = 7. Request cost=8 exceeds limit + burst_credits
d_too_large = limiter.allow("client-multi", cost=8)
self.assertFalse(d_too_large.allowed)
self.assertEqual(d_too_large.remaining, 7)
self.assertEqual(d_too_large.retry_after, float("inf"))
# Key has remaining 7. Request cost=6 consumes 5 window and 1 burst
d_consume = limiter.allow("client-multi", cost=6)
self.assertTrue(d_consume.allowed)
self.assertEqual(d_consume.remaining, 1)
# Request cost=2 cannot be satisfied (only 1 burst credit remaining)
d_all_or_nothing = limiter.allow("client-multi", cost=2)
self.assertFalse(d_all_or_nothing.allowed)
self.assertEqual(d_all_or_nothing.remaining, 1) # unchanged state
def test_retry_after_correctness(self) -> None:
limiter = RateLimiter(limit=1, window_seconds=5.0, burst_credits=0, clock=self.fake_clock)
d1 = limiter.allow("k", cost=1)
self.assertTrue(d1.allowed)
self.current_time += 2.0
d2 = limiter.allow("k", cost=1)
self.assertFalse(d2.allowed)
# Request was at 1000.0, expires at 1005.0. Current time is 1002.0 -> retry_after = 3.0
self.assertAlmostEqual(d2.retry_after, 3.0, places=5)
def test_idle_key_eviction(self) -> None:
limiter = RateLimiter(limit=1, window_seconds=1.0, burst_credits=0, clock=self.fake_clock)
shard = limiter._get_shard("ephemeral-1")
limiter.allow("ephemeral-1", cost=1)
self.assertIn("ephemeral-1", shard.buckets)
# Advance past window_seconds
self.current_time += 1.5
# Eviction is amortized during traffic on that shard
limiter.allow("ephemeral-2", cost=1)
self.assertNotIn("ephemeral-1", shard.buckets)
def test_non_monotonic_and_zero_limit_edge_cases(self) -> None:
# limit == 0 with burst credits
limiter = RateLimiter(limit=0, window_seconds=1.0, burst_credits=2, clock=self.fake_clock)
d = limiter.allow("zero-limit", cost=1)
self.assertTrue(d.allowed)
# Non-monotonic clock call
d_past = limiter.allow("zero-limit", cost=1, now=self.current_time - 10.0)
self.assertTrue(d_past.allowed)
d_exhausted = limiter.allow("zero-limit", cost=1)
self.assertFalse(d_exhausted.allowed)
def test_multithreaded_concurrency(self) -> None:
# Stress test ensuring total admitted requests never exceed theoretical upper bound
real_limiter = RateLimiter(limit=50, window_seconds=0.2, burst_credits=10)
threads: List[threading.Thread] = []
admitted_counts: List[int] = [0] * 10
def worker(tid: int) -> None:
admitted = 0
for _ in range(100):
decision = real_limiter.allow("shared-key", cost=1)
if decision.allowed:
admitted += 1
time.sleep(0.001)
admitted_counts[tid] = admitted
for i in range(10):
t = threading.Thread(target=worker, args=(i,))
threads.append(t)
t.start()
for t in threads:
t.join()
total_admitted = sum(admitted_counts)
# Max possible: limit (50) + burst (10) + refill during runtime (~0.1-0.3s -> ~75 requests)
# Must never exceed conservative bound of limit + burst + rate * 1.5s
self.assertLessEqual(total_admitted, 50 + 10 + int((50 / 0.2) * 1.5))
self.assertGreater(total_admitted, 0)
if name == "main":
unittest.main()
Result
Winning Votes
0 / 3
Average Score
Total Score
Overall Comments
Answer A provides a clean, self-contained implementation with good type hints, a detailed design note, and a solid test suite. However, it suffers from a few subtle concurrency and design flaws: its eviction strategy runs amortized checks during operations that sample shards using an OrderedDict without holding the lock across iteration correctly or handling race conditions on first-touch / eviction as robustly as B. Additionally, Answer A's retry_after loop has edge cases under complex burst credit and window interplay that can lead to imprecise or incorrect values.
View Score Details ▼
Correctness
Weight 35%Sliding window semantics and burst credit refill are generally correct, but retry_after calculation can be imprecise under combined burst/window exhaustion.
Completeness
Weight 20%Meets all functional requirements and deliverables, including types, design note, and unittest suite.
Code Quality
Weight 20%Clean and well-structured code with concise docstrings and clear separation of concerns.
Practical Value
Weight 15%Sharded locking is good, but shard-level locks still serialize operations within the same shard; eviction sampling under load has potential race vulnerabilities.
Instruction Following
Weight 10%Follows all constraints, including Python 3.11 standard library only and no asyncio, though some edge cases are handled less defensively than required.
Total Score
Overall Comments
Answer A provides a self-contained implementation with immutable decisions, sharded locking, and atomic multi-cost admission. However, its retry calculation can promise success before capacity actually becomes available, burst-funded requests are omitted from sliding-window usage, and depleted zero-limit entries can permanently obstruct eviction. Its boundary test contradicts the required interval semantics, its eviction test does not ensure both keys share a shard, and its stress test is not deterministic. The production-ready claim is not supported.
View Score Details ▼
Correctness
Weight 35%Window expiry itself uses the correct exclusive lower boundary, and admission is atomic. However, only normal-funded cost is recorded. Retry calculation ignores the burst cap when considering refill: with limit 2, window 10, no burst, and exhausted quota at time zero, another unit is incorrectly given a five-second retry instead of ten. Snapshot and cross-key cleanup can discard history without consistently advancing the timestamp clamp.
Completeness
Weight 20%Includes the requested public API, implementation, design note, and tests, but memory reclamation fails behind a permanently depleted zero-limit entry. Snapshot exposes remaining capacity rather than actual window usage, and there is no dedicated sub-millisecond test or deterministic concurrency test.
Code Quality
Weight 20%Readable helper decomposition, type hints, and compact immutable decisions are positives. Validation silently coerces non-integer configuration values and does not reject non-finite times or windows. There is an unused retry variable, inaccurate cleanup documentation, and a boundary-test comment that reverses the interval definition.
Practical Value
Weight 15%Independent shards and bounded window-funded event storage are useful, but unreliable retry guidance and potentially blocked eviction undermine gateway deployment. The real-time stress test uses an arbitrary runtime allowance rather than a defensible deterministic bound.
Instruction Following
Weight 10%Uses only the standard library and supplies the requested code and short design note. However, the stress test violates the deterministic fake-clock requirement, the boundary test asserts the wrong result, and eviction coverage relies on an unestablished shard collision.
Total Score
Overall Comments
Answer A delivers a coherent sharded sliding-window limiter with a deque per key, correct (t - W, t] expiry, correct all-or-nothing burst accounting, per-key monotonic clamping, and an amortized LRU eviction pass. However it has a real implementation bug: _compute_retry_after uses the refill path without checking that the needed burst is achievable within the configured cap, so with the default burst_credits=0 (e.g. limit=10, window=10, 10 requests at t=0, request at t=1) it returns 1.0 instead of 9.0. The test suite also does not pass as written: test_exact_boundary_expiry asserts a rejection at t=1001.0 based on a comment that wrongly claims (1000,1001] includes 1000.0, contradicting both the spec and the implementation; test_idle_key_eviction checks a shard for key ephemeral-1 but triggers eviction via ephemeral-2, which almost certainly hashes to a different shard. The stress test uses the real clock and sleeps rather than the fake clock, and its bound is very loose. Keys under limit=0 with spent burst can never be evicted. Design note is reasonable but does not acknowledge these issues.
View Score Details ▼
Correctness
Weight 35%Window semantics, burst cap, and all-or-nothing admission are correct, but retry_after is wrong whenever the refill delta is shorter than the next expiry and the needed burst exceeds the cap (default burst_credits=0 gives e.g. 1.0 instead of 9.0). Two tests fail as written: the boundary test asserts rejection at exactly t=1001.0 contrary to spec and implementation, and the eviction test checks the wrong shard.
Completeness
Weight 20%All deliverables are present (implementation, design note, tests) and most required test scenarios exist, but there is no explicit sub-millisecond window test, no concurrent first-touch test, and the stress test does not use the fake clock. Idle keys under limit==0 with spent burst are never reclaimed.
Code Quality
Weight 20%Readable with slots and type hints, but maintains redundant buckets dict plus access_order OrderedDict, snapshot reuses Decision with a strained allowed semantic, the retry simulation contains a logic error, and test comments contradict the code.
Practical Value
Weight 15%Would run and throttle correctly, but incorrect retry_after under the default configuration would mislead clients and Retry-After headers, and the failing tests reduce trust; 256-shard Blake2b hashing is fine but heavier than needed.
Instruction Following
Weight 10%Meets most structural requirements and the note is under 300 words, but the multithreaded test uses the real clock and sleep instead of the required fake clock, and the sub-millisecond edge case is not visibly exercised.