Respuesta 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",...
Mostrar respuesta completa ▼
"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()
Resultado
Votos ganadores
0 / 3
Puntuación media
Puntuación total
Comentario general
La respuesta A proporciona una implementación limpia y autónoma con buenas sugerencias de tipos, una nota de diseño detallada y un sólido conjunto de pruebas. Sin embargo, sufre de algunos fallos sutiles de concurrencia y diseño: su estrategia de desalojo ejecuta comprobaciones amortizadas durante operaciones que muestrean fragmentos usando un OrderedDict sin mantener el bloqueo correctamente durante la iteración o manejar las condiciones de carrera en el primer acceso / desalojo de forma tan robusta como la B. Además, el bucle retry_after de la respuesta A tiene casos extremos bajo la interacción compleja de créditos de ráfaga y ventanas que pueden llevar a valores imprecisos o incorrectos.
Ver detalle de evaluación ▼
Corrección
Peso 35%La semántica de la ventana deslizante y el recarga de créditos de ráfaga son generalmente correctas, pero el cálculo de retry_after puede ser impreciso bajo agotamiento combinado de ráfaga/ventana.
Integridad
Peso 20%Cumple con todos los requisitos funcionales y entregables, incluyendo tipos, nota de diseño y suite de unittest.
Calidad del código
Peso 20%Código limpio y bien estructurado con docstrings concisas y clara separación de responsabilidades.
Valor práctico
Peso 15%El bloqueo fragmentado es bueno, pero los bloqueos a nivel de fragmento aún serializan las operaciones dentro del mismo fragmento; el muestreo de desalojo bajo carga tiene posibles vulnerabilidades de carrera.
Seguimiento de instrucciones
Peso 10%Sigue todas las restricciones, incluyendo solo la biblioteca estándar de Python 3.11 y sin asyncio, aunque algunos casos extremos se manejan de forma menos defensiva de lo requerido.
Puntuación total
Comentario general
La respuesta A proporciona una implementación autocontenida con decisiones inmutables, bloqueo fragmentado y admisión multi-costo atómica. Sin embargo, su cálculo de reintentos puede prometer éxito antes de que la capacidad esté realmente disponible, las solicitudes financiadas por ráfagas se omiten del uso de la ventana deslizante y las entradas agotadas de límite cero pueden obstruir permanentemente la desalojación. Su prueba de límites contradice la semántica del intervalo requerida, su prueba de desalojación no garantiza que ambas claves compartan un fragmento, y su prueba de estrés no es determinista. La afirmación de estar lista para producción no está respaldada.
Ver detalle de evaluación ▼
Corrección
Peso 35%La expiración de la ventana en sí utiliza el límite inferior exclusivo correcto, y la admisión es atómica. Sin embargo, solo se registra el costo financiado normal. El cálculo de reintentos ignora el límite de ráfaga al considerar la recarga: con un límite de 2, ventana de 10, sin ráfaga y cuota agotada en el tiempo cero, a otra unidad se le da incorrectamente un reintento de cinco segundos en lugar de diez. La instantánea y la limpieza entre claves pueden descartar el historial sin avanzar consistentemente la marca de tiempo límite.
Integridad
Peso 20%Incluye la API pública solicitada, la implementación, la nota de diseño y las pruebas, pero la recuperación de memoria falla detrás de una entrada de límite cero permanentemente agotada. La instantánea expone la capacidad restante en lugar del uso real de la ventana, y no hay una prueba dedicada de sub-milisegundos ni una prueba de concurrencia determinista.
Calidad del código
Peso 20%La descomposición de ayudantes legibles, las sugerencias de tipo y las decisiones inmutables compactas son puntos positivos. La validación coacciona silenciosamente los valores de configuración no enteros y no rechaza tiempos o ventanas no finitos. Hay una variable de reintento no utilizada, documentación de limpieza inexacta y un comentario de prueba de límites que invierte la definición del intervalo.
Valor práctico
Peso 15%Los fragmentos independientes y el almacenamiento de eventos financiados por ventanas acotadas son útiles, pero la guía de reintentos poco fiable y la desalojación potencialmente bloqueada socavan el despliegue de la pasarela. La prueba de estrés en tiempo real utiliza una asignación de tiempo de ejecución arbitraria en lugar de un límite determinista defendible.
Seguimiento de instrucciones
Peso 10%Utiliza solo la biblioteca estándar y proporciona el código solicitado y una breve nota de diseño. Sin embargo, la prueba de estrés viola el requisito del reloj falso determinista, la prueba de límites afirma el resultado incorrecto y la cobertura de desalojación depende de una colisión de fragmentos no establecida.
Puntuación total
Comentario general
La respuesta A ofrece un limitador coherente de ventana deslizante fragmentada con una cola por clave, caducidad correcta (t - W, t], contabilidad de ráfagas correcta de todo o nada, limitación monotónica por clave y un pase de expulsión LRU amortizado. Sin embargo, tiene un error de implementación real: _compute_retry_after utiliza la ruta de recarga sin comprobar que la ráfaga necesaria sea alcanzable dentro del límite configurado, por lo que con burst_credits=0 por defecto (por ejemplo, limit=10, window=10, 10 solicitudes en t=0, solicitud en t=1) devuelve 1.0 en lugar de 9.0. El conjunto de pruebas tampoco pasa tal como está escrito: test_exact_boundary_expiry afirma un rechazo en t=1001.0 basándose en un comentario que afirma erróneamente que (1000,1001] incluye 1000.0, lo que contradice tanto la especificación como la implementación; test_idle_key_eviction comprueba una partición para la clave ephemeral-1 pero activa la expulsión a través de ephemeral-2, que casi con toda seguridad se hashifica en una partición diferente. La prueba de estrés utiliza el reloj real y las pausas en lugar del reloj falso, y su límite es muy laxo. Las claves con limit=0 y ráfaga consumida nunca pueden ser expulsadas. La nota de diseño es razonable pero no reconoce estos problemas.
Ver detalle de evaluación ▼
Corrección
Peso 35%La semántica de la ventana, el límite de ráfaga y la admisión de todo o nada son correctos, pero retry_after es incorrecto siempre que el delta de recarga sea más corto que la siguiente caducidad y la ráfaga necesaria exceda el límite (burst_credits=0 por defecto da, por ejemplo, 1.0 en lugar de 9.0). Dos pruebas fallan tal como están escritas: la prueba de límite afirma un rechazo exactamente en t=1001.0 en contra de la especificación y la implementación, y la prueba de expulsión comprueba la partición incorrecta.
Integridad
Peso 20%Todos los entregables están presentes (implementación, nota de diseño, pruebas) y la mayoría de los escenarios de prueba requeridos existen, pero no hay una prueba explícita de ventana sub-milisegundo, no hay una prueba concurrente de primer acceso y la prueba de estrés no utiliza el reloj falso. Las claves inactivas con limit==0 y ráfaga consumida nunca se recuperan.
Calidad del código
Peso 20%Legible con slots y type hints, pero mantiene un diccionario de buckets redundante además de OrderedDict con orden de acceso, snapshot reutiliza Decision con una semántica permitida forzada, la simulación de reintento contiene un error lógico y los comentarios de prueba contradicen el código.
Valor práctico
Peso 15%Se ejecutaría y limitaría correctamente, pero un retry_after incorrecto bajo la configuración por defecto induciría a error a los clientes y a las cabeceras Retry-After, y las pruebas fallidas reducen la confianza; el hashing Blake2b de 256 particiones está bien pero es más pesado de lo necesario.
Seguimiento de instrucciones
Peso 10%Cumple la mayoría de los requisitos estructurales y la nota tiene menos de 300 palabras, pero la prueba multihilo utiliza el reloj real y sleep en lugar del reloj falso requerido, y el caso límite sub-milisegundo no se ejerce visiblemente.