Clock inconsistency: IdempotencyStore uses time.time() (system clock) while TokenBucketLimiter uses time.monotonic() — an NTP jump could wrongly expire/replay events. Should be monotonic everywhere.
No max-body cap: handle_client does reader.readexactly(content_length) with no limit — a hostile Content-Length could force a huge allocation (DoS). The rate limiter doesn't protect pre-parse.
forward_timestamps list grows unbounded (append-only, only cleared in tests) — memory leak for a long-running service.
HTTP reason phrase is the raw message string (HTTP/1.1 200 Forwarded) — works for the bundled test client but is not valid HTTP for real clients/proxies.
No Content-Type validation on incoming requests (accepts any).
Idempotency eviction is lazy (only on is_seen) — a quiet store retains stale entries until next access; not a leak in steady state but imperfect.
Tests are mildly timing-flaky: 5% random 429 in discord_send + a tight '>1.0s' threshold; no tests for the 400 (bad JSON) or missing-signature 401 paths even though the code handles them.
▮ RECOMMENDED USE
Best non-LFU result for this model (75 vs TTS 49, Rust 50). Runs clean, passes all 4 tests, implements HMAC + idempotency + token-bucket rate-limit + 429 backoff correctly. Safe to offload single-handler HTTP/bridge logic (webhooks, signature verification, rate-limited forwarding). AVOID for multi-task orchestration (TTS) and typed/compiled languages (Rust).
▮ REFACTORED PATCH
# FIX 1 (clock): use monotonic for TTL too.
class IdempotencyStore:
def is_seen(self, event_id):
now = time.monotonic()
...
def mark(self, event_id):
self.store[event_id] = time.monotonic()
# FIX 2 (body cap): reject oversized bodies before reading.
MAX_BODY = 64 * 1024
content_length = int(headers.get('content-length', 0))
if content_length > MAX_BODY:
writer.write(b'HTTP/1.1 413 Payload Too Large\r\nContent-Length: 0\r\n\r\n'); await writer.drain(); return
body = await reader.readexactly(content_length) if 0 < content_length <= MAX_BODY else b''
# FIX 3 (leak): bound forward_timestamps (deque maxlen=N) or drop it if unused.
from collections import deque
self.forward_timestamps: deque = deque(maxlen=1000)
# FIX 4 (reason phrase): use a fixed map.
REASON = {200:'OK',400:'Bad Request',401:'Unauthorized',404:'Not Found',502:'Bad Gateway',500:'Internal Server Error'}
response = f'HTTP/1.1 {status} {REASON.get(status,"OK")}\r\n...'
# FIX 5: add tests for the 400 (malformed JSON) and missing-signature 401 paths.