perf: count RetryBudget and the rate-mode window in time buckets (#181)

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

* perf: count RetryBudget deposits and withdrawals in time buckets

The window now moves in steps of ttl / 10, so a deposit or withdrawal stops counting between 0.9 * ttl and ttl after it was made. Memory stays constant and no call has to drop a backlog of timestamps under the lock.

Author: lesnik512

docs/resilience.md

@@ -84,6 +84,8 @@ ceiling = ceil(len(deposits_in_window) * percent_can_retry) + int(min_retries_pe
 
 The percent term rounds up (`math.ceil`); the floor term truncates (`int`). A withdrawal fails when `len(withdrawn_in_window) >= ceiling`.
 
+The window moves in steps of `ttl / 10`, so a deposit or withdrawal stops counting between `0.9 * ttl` and `ttl` after it was made.
+
 ### Why a floor matters
 
 With no traffic yet, the percent term is zero, and without the floor the first retry would be refused. The floor lets low-traffic clients retry occasional failures. For high-traffic clients the percent term is much larger and the floor stops mattering.
@@ -226,6 +228,8 @@ breaker = AsyncCircuitBreaker(
 
 Once the window holds at least `minimum_calls` outcomes, the rate is checked after every outcome, successes included. A success can therefore be the call that opens the circuit. With `minimum_calls=4` and a threshold of `0.5`, three failures leave the circuit closed because the window holds only three outcomes; a fourth call that succeeds brings it to four, the rate is 3/4, and the circuit opens. That call still returns its response to the caller, and the next request is rejected with `CircuitOpenError`.
 
+The window moves in steps of `window_seconds / 10`, so an outcome stops counting between `0.9 * window_seconds` and `window_seconds` after it was recorded.
+
 In rate mode `failure_threshold` is ignored. Recovery through HALF_OPEN works the same in both modes, and the sync `CircuitBreaker` takes the same parameters.
 
 ### State introspection

src/httpware/middleware/resilience/budget.py

@@ -1,22 +1,20 @@
 """Finagle-style token-bucket retry budget.
 
-Thread-safe and asyncio-safe: all mutations go through a threading.Lock,
-which ensures no torn state across concurrent accesses. When a RetryBudget
-is shared between a sync Client (pool thread) and an AsyncClient (event-loop
-thread), a sync thread holding the lock can briefly block the loop thread's
-acquisition; the critical section (purge + append/compare) is intentionally
-tiny to bound this latency. Safe to share across threads, across coroutines
-on one event loop, and across (sync Client, AsyncClient) pairs in the same
+Thread-safe and asyncio-safe: all mutations go through a threading.Lock held for a
+constant-time update of per-bucket counters, so a sync thread holding it delays an
+event loop sharing the budget only briefly. Safe to share across threads, across
+coroutines on one event loop, and across (sync Client, AsyncClient) pairs in the same
 process.
 """
 
 import math
 import threading
 import time
-from collections import deque
 from collections.abc import Callable
 
 
+_BUCKET_COUNT = 10
+
 _TTL_INVALID = "ttl must be a finite number > 0"
 _MIN_RETRIES_PER_SEC_INVALID = "min_retries_per_sec must be a finite number >= 0"
 _PERCENT_CAN_RETRY_INVALID = "percent_can_retry must be a finite number >= 0"
@@ -36,7 +34,9 @@ class RetryBudget:
 
     Each request deposits a token; each retry attempts to withdraw one.
     Available retries are bounded by `percent_can_retry` of recent deposits,
-    plus a `min_retries_per_sec * ttl` floor.
+    plus a `min_retries_per_sec * ttl` floor. Deposits and withdrawals are counted in
+    ten buckets of `ttl / 10` each, so one stops counting between 0.9 * ttl and ttl
+    after it was recorded.
     """
 
     def __init__(
@@ -53,37 +53,54 @@ def __init__(
         self._percent_can_retry = percent_can_retry
         self._now = _now
         self._lock = threading.Lock()
-        self._deposits: deque[float] = deque()
-        self._withdrawn: deque[float] = deque()
+        self._bucket_width = ttl / _BUCKET_COUNT
+        self._floor = int(min_retries_per_sec * ttl)
+        self._slot: float = -math.inf
+        self._index = 0
+        self._bucket_deposits = [0] * _BUCKET_COUNT
+        self._bucket_withdrawn = [0] * _BUCKET_COUNT
+        self._deposits = 0
+        self._withdrawn = 0
 
-    def _purge(self, now: float) -> None:
-        # Caller must hold self._lock.
-        # Strict `< cutoff` keeps entries at exactly `now - ttl`: window is [now - ttl, now].
-        cutoff = now - self._ttl
-        while self._deposits and self._deposits[0] < cutoff:
-            self._deposits.popleft()
-        while self._withdrawn and self._withdrawn[0] < cutoff:
-            self._withdrawn.popleft()
+    def _rotate(self, slot: float) -> None:
+        """Make `slot` the current bucket, dropping buckets that fell out of the window. The caller holds self._lock."""
+        current = int(slot)
+        if current - self._slot >= _BUCKET_COUNT:
+            self._bucket_deposits = [0] * _BUCKET_COUNT
+            self._bucket_withdrawn = [0] * _BUCKET_COUNT
+            self._deposits = 0
+            self._withdrawn = 0
+        else:
+            for stale in range(int(self._slot) + 1, current + 1):
+                index = stale % _BUCKET_COUNT
+                self._deposits -= self._bucket_deposits[index]
+                self._withdrawn -= self._bucket_withdrawn[index]
+                self._bucket_deposits[index] = 0
+                self._bucket_withdrawn[index] = 0
+        self._slot = current
+        self._index = current % _BUCKET_COUNT
 
     def deposit(self) -> None:
         """Record a request (success or failure attempt). Adds one token."""
-        now = self._now()
+        slot = self._now() // self._bucket_width
         with self._lock:
-            self._purge(now)
-            self._deposits.append(now)
+            if slot > self._slot:
+                self._rotate(slot)
+            self._bucket_deposits[self._index] += 1
+            self._deposits += 1
 
     def try_withdraw(self) -> bool:
         """Atomically attempt to spend one retry token.
 
         Returns True if a retry is permitted, False if the budget is exhausted.
         Never blocks.
         """
-        now = self._now()
+        slot = self._now() // self._bucket_width
         with self._lock:
-            self._purge(now)
-            floor = int(self._min_retries_per_sec * self._ttl)
-            ceiling = math.ceil(len(self._deposits) * self._percent_can_retry) + floor
-            if len(self._withdrawn) >= ceiling:
+            if slot > self._slot:
+                self._rotate(slot)
+            if self._withdrawn >= math.ceil(self._deposits * self._percent_can_retry) + self._floor:
                 return False
-            self._withdrawn.append(now)
+            self._bucket_withdrawn[self._index] += 1
+            self._withdrawn += 1
             return True

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_budget.py

@@ -136,3 +136,48 @@ def test_boundary_budget_values_accepted() -> None:
     assert budget.try_withdraw() is True
     assert budget.try_withdraw() is True
     assert budget.try_withdraw() is False
+
+
+def test_deposit_counts_until_its_bucket_leaves_the_window() -> None:
+    # ttl=10 gives 1 s buckets; a deposit at 0.5 lives in bucket [0, 1) and stops counting at 10.0
+    clock = _Clock(0.5)
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=0.0, percent_can_retry=1.0, _now=clock.now)
+    budget.deposit()
+    clock.advance(9.49)
+    assert budget.try_withdraw() is True
+    clock.advance(0.01)
+    assert budget.try_withdraw() is False
+
+
+def test_expired_buckets_release_only_their_own_counts() -> None:
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=0.0, percent_can_retry=1.0, _now=clock.now)
+    budget.deposit()
+    clock.advance(5.0)
+    budget.deposit()
+    budget.deposit()
+    clock.advance(5.0)
+    # the first deposit has expired, the two later ones still count
+    assert budget.try_withdraw() is True
+    assert budget.try_withdraw() is True
+    assert budget.try_withdraw() is False
+
+
+def test_withdrawal_expires_with_its_bucket() -> None:
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=0.1, percent_can_retry=0.0, _now=clock.now)
+    assert budget.try_withdraw() is True
+    clock.advance(9.0)
+    assert budget.try_withdraw() is False
+    clock.advance(1.0)
+    assert budget.try_withdraw() is True
+
+
+def test_earlier_timestamp_counts_in_the_current_bucket() -> None:
+    times = iter([5.0, 4.0, 14.5])
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=0.0, percent_can_retry=1.0, _now=lambda: next(times))
+    budget.deposit()
+    budget.deposit()
+    # both deposits sit in the bucket for [5, 6), which is still live at 14.5
+    assert budget._deposits == 2  # noqa: SLF001, PLR2004
+    assert budget.try_withdraw() is True

tests/test_budget_props.py

@@ -6,6 +6,9 @@
    beyond the floor).
 3. `deposit()` is monotonically non-decreasing in permitted retries (more deposits cannot
    reduce the budget).
+4. Against a reference that keeps every timestamp, each `try_withdraw()` answer is one the
+   reference allows when events stop counting anywhere between 0.9 * ttl and ttl after
+   they were recorded.
 """
 
 import math
@@ -114,3 +117,58 @@ def test_more_deposits_never_decreases_budget(
         budget2.deposit()
     new_permitted = sum(1 for _ in range(base_deposits + extra_deposits + 200) if budget2.try_withdraw())
     assert new_permitted >= initial_permitted
+
+
+_EDGE = 1e-9
+
+
+def _live(timestamps: list[float], now: float, max_age: float) -> int:
+    return sum(1 for t in timestamps if now - t <= max_age)
+
+
+def _ceiling(deposits: int, percent: float, floor: int) -> int:
+    return math.ceil(deposits * percent) + floor
+
+
+@given(
+    ttl=st.floats(min_value=0.1, max_value=60.0, allow_nan=False, allow_infinity=False),
+    min_rps=st.floats(min_value=0.0, max_value=5.0, allow_nan=False, allow_infinity=False),
+    percent=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
+    start=st.floats(min_value=0.0, max_value=1e6, allow_nan=False, allow_infinity=False),
+    ops=st.lists(
+        st.tuples(st.booleans(), st.floats(min_value=0.0, max_value=0.5, allow_nan=False, allow_infinity=False)),
+        max_size=300,
+    ),
+)
+@settings(max_examples=200, deadline=None)
+def test_matches_timestamp_reference_within_one_bucket(
+    ttl: float,
+    min_rps: float,
+    percent: float,
+    start: float,
+    ops: list[tuple[bool, float]],
+) -> None:
+    clock = _Clock()
+    clock.advance(start)
+    budget = _budget(ttl=ttl, min_retries_per_sec=min_rps, percent_can_retry=percent, now=clock.now)
+    floor = int(min_rps * ttl)
+    shortest = 0.9 * ttl * (1 - _EDGE)
+    longest = ttl * (1 + _EDGE)
+    deposits: list[float] = []
+    withdrawn: list[float] = []
+    for is_deposit, step in ops:
+        clock.advance(step * ttl)
+        now = clock.now()
+        if is_deposit:
+            budget.deposit()
+            deposits.append(now)
+            continue
+        permitted = budget.try_withdraw()
+        must_permit = _live(withdrawn, now, longest) < _ceiling(_live(deposits, now, shortest), percent, floor)
+        must_refuse = _live(withdrawn, now, shortest) >= _ceiling(_live(deposits, now, longest), percent, floor)
+        if must_permit:
+            assert permitted
+        if must_refuse:
+            assert not permitted
+        if permitted:
+            withdrawn.append(now)

tests/test_retry.py

@@ -569,8 +569,8 @@ async def streamed_body() -> typing.AsyncIterator[bytes]:
         await client.put("https://example.test/upload", content=streamed_body())
 
     # Budget should be untouched: deposits OK (every attempt deposits), but no withdrawals.
-    # Check via _withdrawn deque emptiness.
-    assert len(budget._withdrawn) == 0  # noqa: SLF001 — implementation-detail access for invariant
+    # Check via the withdrawn counter.
+    assert budget._withdrawn == 0  # noqa: SLF001 — implementation-detail access for invariant
 
 
 async def test_retry_refuses_streamed_body_network_error_idempotent() -> None:
@@ -776,7 +776,7 @@ async def test_retry_after_exceeds_max_delay_does_not_consume_budget_token() ->
     with pytest.raises(ServiceUnavailableError):
         await client.get("https://example.test/x")
     # Retry-After > max_delay: give up without spending a budget token.
-    assert len(budget._withdrawn) == 0  # noqa: SLF001 — implementation-detail access for invariant
+    assert budget._withdrawn == 0  # noqa: SLF001 — implementation-detail access for invariant
 
 
 async def test_retry_after_huge_digit_string_gives_up() -> None: