Resposta A: Google Gemini 3.8 Flash
Produção pronta, limitador de taxa thread-safe com janela deslizante e créditos de rajada.
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 resposta completa ▼
Produção pronta, limitador de taxa thread-safe com janela deslizante e créditos de rajada.
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:
"""Limitador de taxa de alta vazão, janela deslizante com tratamento de créditos de rajada."""
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,
)
"""
Nota de Design (273 palavras):
Este limitador de taxa implementa uma janela deslizante exata combinada com um pool contínuo de créditos de rajada de balde de tokens.
Estrutura de Dados e Tradeoff de Precisão:
Cada chave de cliente mantém um _KeyBucket contendo uma deque de tuplas (timestamp, cost) rastreando a cota da janela consumida, um acumulador inteiro window_usage e um float burst_credits. Uma janela deslizante exata é escolhida em vez de modelos aproximados de bucketing ou balde vazando para garantir erro zero de fronteira: os timestamps das solicitações contam estritamente dentro de (t - window_seconds, t]. Solicitações de custo múltiplo (cost > 1) são admitidas atomicamente: a cota de janela disponível é reivindicada primeiro, e qualquer déficit é coberto por créditos de rajada contínuos reabastecidos a limit / window_seconds. Se a capacidade total disponível for insuficiente, a solicitação é rejeitada tudo ou nada. A memória da deque por chave ativa escala estritamente como O(N) onde N <= limite, limitada e compacta.
Estratégia de Bloqueio:
Para evitar um gargalo de contenção global e impedir que solicitações concorrentes serializem todo o gateway, o espaço de chaves é particionado em 256 shards independentes via hashing Blake2b. Cada shard mantém seu próprio threading.Lock, dicionário e ordem de acesso. Operações em chaves distintas mapeiam para locks independentes com alta probabilidade, minimizando a contenção de locks enquanto sincronizam totalmente os acessos concorrentes por chave sem deadlocks.
Higiene de Memória e Evicção:
Para proteger contra o crescimento ilimitado de memória de milhões de chaves de uso único, cada shard mantém um OrderedDict da ordem de acesso das chaves. Em cada operação, uma limpeza amortizada O(1) amostra as chaves mais antigas no shard. Se a deque da janela de uma chave estiver vazia, seus créditos de rajada forem totalmente regenerados e ela estiver inativa por pelo menos window_seconds, ela é permanentemente removida. Chaves inativas expiram naturalmente sem threads de background.
"""
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 de vitória
0 / 3
Pontuação média
Pontuação total
Comentário geral
A Resposta A fornece uma implementação limpa e autônoma com boas dicas de tipo, uma nota de design detalhada e um conjunto de testes sólido. No entanto, sofre de algumas falhas sutis de concorrência e design: sua estratégia de despejo executa verificações amortizadas durante operações que amostram fragmentos usando um OrderedDict sem manter o bloqueio corretamente durante a iteração ou lidar com condições de corrida no primeiro toque / despejo de forma tão robusta quanto B. Além disso, o loop retry_after da Resposta A tem casos extremos sob interação complexa de crédito de rajada e janela que podem levar a valores imprecisos ou incorretos.
Ver detalhes da avaliação ▼
Correção
Peso 35%A semântica da janela deslizante e o recarregamento do crédito de rajada estão geralmente corretos, mas o cálculo do retry_after pode ser impreciso sob exaustão combinada de rajada/janela.
Completude
Peso 20%Atende a todos os requisitos funcionais e entregas, incluindo tipos, nota de design e suíte de testes unitários.
Qualidade do código
Peso 20%Código limpo e bem estruturado com docstrings concisas e clara separação de responsabilidades.
Valor prático
Peso 15%O bloqueio fragmentado é bom, mas os bloqueios em nível de fragmento ainda serializam operações dentro do mesmo fragmento; a amostragem de despejo sob carga tem potenciais vulnerabilidades de condição de corrida.
Seguimento de instruções
Peso 10%Segue todas as restrições, incluindo apenas a biblioteca padrão Python 3.11 e sem asyncio, embora alguns casos extremos sejam tratados de forma menos defensiva do que o necessário.
Pontuação total
Comentário geral
A Resposta A fornece uma implementação autônoma com decisões imutáveis, bloqueio fragmentado e admissão multi-custo atômica. No entanto, seu cálculo de nova tentativa pode prometer sucesso antes que a capacidade esteja realmente disponível, requisições financiadas por rajada são omitidas do uso da janela deslizante e entradas de limite zero esgotadas podem obstruir permanentemente a evacuação. Seu teste de limite contradiz a semântica de intervalo exigida, seu teste de evacuação não garante que ambas as chaves compartilhem um fragmento, e seu teste de estresse não é determinístico. A alegação de pronto para produção não é suportada.
Ver detalhes da avaliação ▼
Correção
Peso 35%A expiração da janela em si usa o limite inferior exclusivo correto, e a admissão é atômica. No entanto, apenas o custo financiado normal é registrado. O cálculo de nova tentativa ignora o limite de rajada ao considerar o reabastecimento: com limite 2, janela 10, sem rajada e cota esgotada no tempo zero, outra unidade recebe incorretamente uma nova tentativa de cinco segundos em vez de dez. O snapshot e a limpeza entre chaves podem descartar o histórico sem avançar consistentemente o limite de tempo.
Completude
Peso 20%Inclui a API pública solicitada, implementação, nota de design e testes, mas a recuperação de memória falha atrás de uma entrada de limite zero permanentemente esgotada. O snapshot expõe a capacidade restante em vez do uso real da janela, e não há teste dedicado de sub-milissegundo ou teste de concorrência determinístico.
Qualidade do código
Peso 20%A decomposição de helpers legíveis, dicas de tipo e decisões imutáveis compactas são pontos positivos. A validação coerge silenciosamente valores de configuração não inteiros e não rejeita tempos ou janelas não finitos. Há uma variável de nova tentativa não utilizada, documentação de limpeza imprecisa e um comentário de teste de limite que reverte a definição de intervalo.
Valor prático
Peso 15%Fragmentos independentes e armazenamento de eventos financiados por janela limitada são úteis, mas a orientação de nova tentativa não confiável e a evacuação potencialmente bloqueada minam a implantação de gateway. O teste de estresse em tempo real usa uma tolerância de tempo de execução arbitrária em vez de um limite determinístico defensável.
Seguimento de instruções
Peso 10%Usa apenas a biblioteca padrão e fornece o código solicitado e uma breve nota de design. No entanto, o teste de estresse viola o requisito de relógio falso determinístico, o teste de limite afirma o resultado errado e a cobertura de evacuação depende de uma colisão de fragmento não estabelecida.
Pontuação total
Comentário geral
A Resposta A entrega um limitador sharded de janela deslizante coerente com uma deque por chave, expiração correta (t - W, t], contabilização de rajada correta all-or-nothing, clamping monotônico por chave e uma passagem de expurgo LRU amortizada. No entanto, tem um bug de implementação real: _compute_retry_after usa o caminho de recarga sem verificar se a rajada necessária é alcançável dentro do limite configurado, então com burst_credits=0 padrão (por exemplo, limit=10, window=10, 10 requisições em t=0, requisição em t=1) retorna 1.0 em vez de 9.0. A suíte de testes também não passa como está escrita: test_exact_boundary_expiry afirma uma rejeição em t=1001.0 com base em um comentário que afirma incorretamente que (1000,1001] inclui 1000.0, contradizendo tanto a especificação quanto a implementação; test_idle_key_eviction verifica um shard para a chave ephemeral-1, mas aciona a expurgo via ephemeral-2, que quase certamente faz hash para um shard diferente. O teste de estresse usa o relógio real e sleeps em vez do relógio falso, e seu limite é muito frouxo. Chaves sob limit=0 com rajada gasta nunca podem ser expurgadas. A nota de design é razoável, mas não reconhece esses problemas.
Ver detalhes da avaliação ▼
Correção
Peso 35%Semântica da janela, limite de rajada e admissão all-or-nothing estão corretos, mas retry_after está incorreto sempre que o delta de recarga é menor que a próxima expiração e a rajada necessária excede o limite (burst_credits=0 padrão dá, por exemplo, 1.0 em vez de 9.0). Dois testes falham como estão escritos: o teste de limite afirma rejeição exatamente em t=1001.0 contrariamente à especificação e implementação, e o teste de expurgo verifica o shard errado.
Completude
Peso 20%Todos os entregáveis estão presentes (implementação, nota de design, testes) e a maioria dos cenários de teste necessários existe, mas não há teste explícito de janela sub-milissegundo, nenhum teste de primeiro toque concorrente e o teste de estresse não usa o relógio falso. Chaves ociosas sob limit==0 com rajada gasta nunca são recuperadas.
Qualidade do código
Peso 20%Legível com slots e type hints, mas mantém um dicionário de buckets redundante mais OrderedDict de ordem de acesso, snapshot reutiliza Decision com um semântico permitido forçado, a simulação de retry contém um erro lógico e os comentários de teste contradizem o código.
Valor prático
Peso 15%Executaria e limitaria corretamente, mas o retry_after incorreto sob a configuração padrão induziria clientes em erro e os cabeçalhos Retry-After, e os testes falhos reduzem a confiança; o hashing Blake2b de 256 shards é bom, mas mais pesado do que o necessário.
Seguimento de instruções
Peso 10%Atende à maioria dos requisitos estruturais e a nota tem menos de 300 palavras, mas o teste multithread usa o relógio real e sleep em vez do relógio falso exigido, e o caso de borda sub-milissegundo não é visivelmente exercitado.