Independent verification: what six fresh-context rounds found in a gauntlet that was already green

demo-rate-limiter/src/ratelimiter/__init__.py

@@ -1,49 +1,99 @@
 """Sliding-window rate limiter with an injectable clock."""
 
 import math
+import threading
 from collections import deque
 from collections.abc import Callable
 
 __all__ = ["RateLimiter"]
 
 
+def _validate(limit: int, window_seconds: float) -> None:
+    """Reject any configuration that would silently never or always allow.
+
+    Extracted from __init__ so that adding the window_seconds type guard did
+    not push the constructor to the top of the complexity budget. `bool` is
+    excluded explicitly: it is a subclass of int, so True would otherwise be
+    accepted as a limit of 1 and a window of 1.0 second.
+    """
+    if isinstance(limit, bool) or not isinstance(limit, int):
+        raise ValueError(f"limit must be an integer, got {limit!r}")
+    if limit <= 0:
+        raise ValueError(f"limit must be positive, got {limit}")
+    if isinstance(window_seconds, bool) or not isinstance(window_seconds, int | float):
+        raise ValueError(f"window_seconds must be a number, got {window_seconds!r}")
+    if not math.isfinite(window_seconds) or window_seconds <= 0:
+        raise ValueError(
+            f"window_seconds must be positive and finite, got {window_seconds}"
+        )
+
+
 class RateLimiter:
     """Allow at most `limit` requests per key within any sliding window.
 
-    `clock` returns the current time in seconds; timestamps older than
-    `window_seconds` fall out of the window individually. A backward-jumping
-    clock fails closed: past hits never expire early.
+    `clock` returns the current time in seconds and MUST be monotonic
+    (`time.monotonic`); timestamps older than `window_seconds` fall out of the
+    window individually. A backward-jumping clock fails closed: past hits never
+    expire early. A forward jump expires every hit at once — that is a caller
+    obligation, not a defect (see the clock contract in spec.md).
+
+    Safe to call from multiple threads. Memory is bounded by the distinct
+    keys seen within TWO windows, not one: a key is dropped by the first sweep
+    that runs more than a window after its last hit, and sweeps are throttled
+    to at most one per window, so worst-case retention is just under 2W. The
+    bound is temporal, not cardinal — see the accepted residual risk in
+    spec.md.
     """
 
     def __init__(
         self, limit: int, window_seconds: float, clock: Callable[[], float]
     ) -> None:
-        if limit <= 0:
-            raise ValueError(f"limit must be positive, got {limit}")
-        if not math.isfinite(window_seconds) or window_seconds <= 0:
-            raise ValueError(
-                f"window_seconds must be positive and finite, got {window_seconds}"
-            )
+        _validate(limit, window_seconds)
         self._limit = limit
         self._window = window_seconds
         self._clock = clock
+        self._lock = threading.Lock()
         self._hits: dict[str, deque[float]] = {}
+        self._last_sweep = -math.inf
 
     def allow(self, key: str) -> bool:
         """Record and allow this request, or deny it. Denials store nothing."""
-        now = self._clock()
-        hits = self._prune(key, now)
-        if len(hits) >= self._limit:
-            return False
-        hits.append(now)
-        self._hits[key] = hits
-        return True
+        if not isinstance(key, str):
+            raise TypeError(f"key must be a str, got {type(key).__name__}")
+        if not key:
+            raise ValueError("key must not be empty")
+        # The clock read belongs inside the lock. Read outside it, two callers
+        # can commit in the opposite order from which they read the clock, and
+        # the deque that _prune and _sweep both assume is ascending stops being
+        # so; _sweep then reads a stale newest-hit and forgets a key that still
+        # has a live hit, resetting that caller's quota. Cost: `clock` must not
+        # call back into this limiter (see the clock contract).
+        with self._lock:
+            now = self._clock()
+            self._sweep(now)
+            hits = self._prune(key, now)
+            if len(hits) >= self._limit:
+                return False
+            hits.append(now)
+            self._hits[key] = hits
+            return True
+
+    def _sweep(self, now: float) -> None:
+        """Forget keys idle for a full window. Runs at most once per window."""
+        # The lower bound matters: after a backward clock jump `now` sits below
+        # _last_sweep, and a one-sided `<= window` test then suspends the sweep
+        # until the clock catches up — measured 20,001 keys retained. Treating a
+        # negative delta as "sweep now" re-arms the throttle at the new time.
+        if 0 <= now - self._last_sweep <= self._window:
+            return
+        self._last_sweep = now
+        idle = [k for k, hits in self._hits.items() if now - hits[-1] > self._window]
+        for key in idle:
+            del self._hits[key]
 
     def _prune(self, key: str, now: float) -> deque[float]:
-        """Drop hits older than the window; forget keys with none left."""
+        """Drop hits older than the window. Forgetting keys is _sweep's job."""
         hits = self._hits.get(key, deque())
         while hits and now - hits[0] > self._window:
             hits.popleft()
-        if not hits:
-            self._hits.pop(key, None)
         return hits

demo-rate-limiter/tests/test_properties.py

@@ -6,15 +6,35 @@
 
 from ratelimiter import RateLimiter
 
-requests = st.lists(
-    st.tuples(
-        st.floats(min_value=0, max_value=1000, allow_nan=False, allow_infinity=False),
-        st.sampled_from("abc"),
-    ),
-    max_size=60,
+timestamps = st.floats(
+    min_value=0, max_value=1000, allow_nan=False, allow_infinity=False
 )
+# [REVISION 4] Keys were `st.sampled_from("abc")`, so the whole suite ever saw
+# three distinct keys and an implementation hardcoded to them scored 100%
+# coverage and 8/8 mutants. A small alphabet keeps collisions frequent (which
+# is what makes P1 bite) while ranging far outside any hardcoded set.
+# Widening this too far blunts the layer: with 258 possible keys and limits up
+# to 20, hypothesis almost never drives one key to its limit, so the deny
+# branch goes unexercised and the fail-open mutant M5 survives the property
+# suite. 12 keys keeps collisions frequent while still ranging outside any
+# hardcoded key set. Measured by the layer-attribution run, not guessed.
+keys = st.text(alphabet="abc", min_size=1, max_size=2)
+requests = st.lists(st.tuples(timestamps, keys), max_size=60)
+# P2 needs the target key to recur, so it keeps a small pool — but the pool is
+# no longer three single characters. The widening above was applied to P1 only
+# in the first pass, directly under the comment explaining it; P2 was left
+# behind. What the widening fixed is the strip()-merge blindness (the pool held
+# "c " but not "c", so no two members could merge). It did NOT change P2's
+# attribution: measured, P2 alone still kills none of M1/M5/M12.
+isolation_keys = st.sampled_from(["a", "ab", "Ab", "b", "bc", "c", "c "])
+isolation_requests = st.lists(st.tuples(timestamps, isolation_keys), max_size=60)
 limits = st.integers(min_value=1, max_value=5)
-windows = st.floats(min_value=0.1, max_value=100, allow_nan=False)
+# Mostly ordinary windows, sometimes far outside the tested range, so an
+# implementation that special-cases large windows cannot hide.
+windows = st.one_of(
+    st.floats(min_value=0.1, max_value=100, allow_nan=False),
+    st.floats(min_value=1001, max_value=5000, allow_nan=False),
+)
 
 
 def run(
@@ -43,14 +63,20 @@ def test_p1_allowed_count_within_any_window_never_exceeds_limit(
         assert in_window <= limit
 
 
-@given(steps=requests, limit=limits, window=windows)
+@given(steps=isolation_requests, limit=limits, window=windows)
 def test_p2_other_keys_traffic_never_changes_one_keys_outcomes(
     steps: list[tuple[float, str]], limit: int, window: float
 ) -> None:
+    if not steps:
+        return
     steps.sort(key=lambda s: s[0])
+    # Take the target from the data rather than hardcoding "a": with a wider
+    # key pool a fixed target is often absent, and the property then holds
+    # vacuously over two empty lists.
+    target = steps[0][1]
     clock_full = FakeClock()
     full = run(RateLimiter(limit, window, clock_full), clock_full, steps)
-    only_a = [s for s in steps if s[1] == "a"]
+    only_target = [s for s in steps if s[1] == target]
     clock_solo = FakeClock()
-    solo = run(RateLimiter(limit, window, clock_solo), clock_solo, only_a)
-    assert [o for o in full if o[1] == "a"] == solo
+    solo = run(RateLimiter(limit, window, clock_solo), clock_solo, only_target)
+    assert [o for o in full if o[1] == target] == solo