perf: keep running totals in the rate-mode circuit breaker window

Author: lesnik512

src/httpware/middleware/resilience/circuit_breaker.py

@@ -114,51 +114,58 @@ class CircuitState(enum.Enum):
 class _RollingWindow:
     """Time-bucketed success/failure counters over a rolling window.
 
-    `window_seconds` is split into `_BUCKET_COUNT` buckets. Each bucket holds
-    [successes, failures] tagged with the integer time-slot it represents; a
-    bucket whose slot is stale is reset on write, and `totals` filters to the
-    live slot range so data older than the window never counts. Every method is
-    synchronous and reads `now` from its caller (so the breaker's critical
-    section owns the clock read).
+    `window_seconds` is split into `_BUCKET_COUNT` buckets, one per integer time-slot; the
+    window is the current slot and the `_BUCKET_COUNT - 1` before it. Running totals are
+    kept beside the buckets, so recording and reading are constant-time. Every method is
+    synchronous and reads `now` from its caller (so the breaker's critical section owns
+    the clock read).
     """
 
     def __init__(self, window_seconds: float) -> None:
         self._bucket_width = window_seconds / _BUCKET_COUNT
-        self._slot = [-1] * _BUCKET_COUNT
+        self._slot: float = -math.inf
+        self._index = 0
         self._success = [0] * _BUCKET_COUNT
         self._failure = [0] * _BUCKET_COUNT
-
-    def _current_slot(self, now: float) -> int:
-        return int(now // self._bucket_width)
+        self._total = 0
+        self._failures = 0
+
+    def _advance(self, now: float) -> None:
+        slot = now // self._bucket_width
+        if slot <= self._slot:
+            return
+        current = int(slot)
+        if current - self._slot >= _BUCKET_COUNT:
+            self.clear()
+        else:
+            for stale in range(int(self._slot) + 1, current + 1):
+                index = stale % _BUCKET_COUNT
+                self._total -= self._success[index] + self._failure[index]
+                self._failures -= self._failure[index]
+                self._success[index] = 0
+                self._failure[index] = 0
+        self._slot = current
+        self._index = current % _BUCKET_COUNT
 
     def record(self, now: float, *, failed: bool) -> None:
-        slot = self._current_slot(now)
-        index = slot % _BUCKET_COUNT
-        if self._slot[index] != slot:  # bucket reused for a new slot — evict
-            self._slot[index] = slot
-            self._success[index] = 0
-            self._failure[index] = 0
+        self._advance(now)
         if failed:
-            self._failure[index] += 1
+            self._failure[self._index] += 1
+            self._failures += 1
         else:
-            self._success[index] += 1
+            self._success[self._index] += 1
+        self._total += 1
 
     def totals(self, now: float) -> tuple[int, int]:
         """Return (total, failures) across buckets still inside the window at `now`."""
-        slot = self._current_slot(now)
-        oldest = slot - _BUCKET_COUNT + 1
-        total = 0
-        failures = 0
-        for i in range(_BUCKET_COUNT):
-            if oldest <= self._slot[i] <= slot:
-                total += self._success[i] + self._failure[i]
-                failures += self._failure[i]
-        return total, failures
+        self._advance(now)
+        return self._total, self._failures
 
     def clear(self) -> None:
-        self._slot = [-1] * _BUCKET_COUNT
         self._success = [0] * _BUCKET_COUNT
         self._failure = [0] * _BUCKET_COUNT
+        self._total = 0
+        self._failures = 0
 
 
 _Event = tuple[str, int, str, dict[str, typing.Any]]

tests/test_rolling_window.py

@@ -68,3 +68,56 @@ def test_totals_match_live_events(events: list[tuple[float, bool]]) -> None:
     assert total == len(expected_live)
     assert failures == sum(1 for _, f in expected_live if f)
     assert 0 <= failures <= total
+
+
+def test_partial_advance_evicts_only_expired_buckets() -> None:
+    w = _RollingWindow(window_seconds=10.0)
+    w.record(0.0, failed=True)
+    w.record(4.0, failed=False)
+    w.record(5.0, failed=True)
+    assert w.totals(10.0) == (2, 1)
+    assert w.totals(14.0) == (1, 1)
+    assert w.totals(15.0) == (0, 0)
+
+
+def test_earlier_timestamp_counts_in_the_current_bucket() -> None:
+    w = _RollingWindow(window_seconds=10.0)
+    w.record(5.0, failed=False)
+    w.record(4.0, failed=True)
+    assert w.totals(14.5) == (2, 1)
+    assert w.totals(15.0) == (0, 0)
+
+
+_EDGE = 1e-9
+
+
+@given(
+    window_seconds=st.floats(min_value=0.1, max_value=120.0, allow_nan=False, allow_infinity=False),
+    start=st.floats(min_value=0.0, max_value=1e6, allow_nan=False, allow_infinity=False),
+    steps=st.lists(
+        st.tuples(st.floats(min_value=0.0, max_value=0.5, allow_nan=False, allow_infinity=False), st.booleans()),
+        min_size=1,
+        max_size=300,
+    ),
+)
+@settings(max_examples=200, deadline=None)
+def test_totals_within_one_bucket_of_timestamp_reference(
+    window_seconds: float,
+    start: float,
+    steps: list[tuple[float, bool]],
+) -> None:
+    """An outcome counts until somewhere between 0.9 and 1.0 window lengths after it was recorded."""
+    w = _RollingWindow(window_seconds=window_seconds)
+    shortest = 0.9 * window_seconds * (1 - _EDGE)
+    longest = window_seconds * (1 + _EDGE)
+    now = start
+    events: list[tuple[float, bool]] = []
+    for step, failed in steps:
+        now += step * window_seconds
+        w.record(now, failed=failed)
+        events.append((now, failed))
+        total, failures = w.totals(now)
+        assert sum(1 for t, _ in events if now - t <= shortest) <= total
+        assert total <= sum(1 for t, _ in events if now - t <= longest)
+        assert sum(1 for t, f in events if f and now - t <= shortest) <= failures
+        assert failures <= sum(1 for t, f in events if f and now - t <= longest)