From baacda15c2da6bb968f85c8ac6e954bf89d0ebfd Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 12:51:37 +0300 Subject: [PATCH 1/2] perf: keep running totals in the rate-mode circuit breaker window --- .../middleware/resilience/circuit_breaker.py | 63 ++++++++++--------- tests/test_rolling_window.py | 53 ++++++++++++++++ 2 files changed, 88 insertions(+), 28 deletions(-) diff --git a/src/httpware/middleware/resilience/circuit_breaker.py b/src/httpware/middleware/resilience/circuit_breaker.py index 6ccfb51..f529028 100644 --- a/src/httpware/middleware/resilience/circuit_breaker.py +++ b/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]] diff --git a/tests/test_rolling_window.py b/tests/test_rolling_window.py index b407cc7..0bb348b 100644 --- a/tests/test_rolling_window.py +++ b/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) From ca486bf96180e619e27e7e739d632666abd28c91 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 12:51:37 +0300 Subject: [PATCH 2/2] 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. --- docs/resilience.md | 4 ++ src/httpware/middleware/resilience/budget.py | 73 ++++++++++++-------- tests/test_budget.py | 45 ++++++++++++ tests/test_budget_props.py | 58 ++++++++++++++++ tests/test_retry.py | 6 +- tests/test_retry_budget_threadsafety.py | 33 +++++++-- tests/test_retry_sync.py | 2 +- tests/test_threading_with_shared_budget.py | 21 +++--- 8 files changed, 194 insertions(+), 48 deletions(-) diff --git a/docs/resilience.md b/docs/resilience.md index 9d2b7ef..8a50237 100644 --- a/docs/resilience.md +++ b/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 diff --git a/src/httpware/middleware/resilience/budget.py b/src/httpware/middleware/resilience/budget.py index 278400f..0bc03a7 100644 --- a/src/httpware/middleware/resilience/budget.py +++ b/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,24 +53,41 @@ 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. @@ -78,12 +95,12 @@ def try_withdraw(self) -> bool: 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 diff --git a/tests/test_budget.py b/tests/test_budget.py index 2577589..bada74f 100644 --- a/tests/test_budget.py +++ b/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 diff --git a/tests/test_budget_props.py b/tests/test_budget_props.py index 4867b70..a947a44 100644 --- a/tests/test_budget_props.py +++ b/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) diff --git a/tests/test_retry.py b/tests/test_retry.py index 29095b8..97db121 100644 --- a/tests/test_retry.py +++ b/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: diff --git a/tests/test_retry_budget_threadsafety.py b/tests/test_retry_budget_threadsafety.py index 44fa57b..ffdf5c0 100644 --- a/tests/test_retry_budget_threadsafety.py +++ b/tests/test_retry_budget_threadsafety.py @@ -1,7 +1,7 @@ """Thread-safety test for RetryBudget. Sync Client may share a RetryBudget across a ThreadPoolExecutor. Concurrent -deposit() / try_withdraw() calls must not corrupt the internal deques. We +deposit() / try_withdraw() calls must not corrupt the internal counters. We spawn many threads doing many ops and assert no exception, sane counters. """ @@ -41,8 +41,8 @@ def worker() -> None: # Each thread did _N_OPS_PER_THREAD deposits; budget must have accepted them all # (and possibly some withdrawals — we don't assert withdrawn count; the ceiling # formula doesn't guarantee how many succeed). - assert len(budget._deposits) <= _N_THREADS * _N_OPS_PER_THREAD # noqa: SLF001 — internal state check - assert len(budget._deposits) > 0 # noqa: SLF001 + assert budget._deposits <= _N_THREADS * _N_OPS_PER_THREAD # noqa: SLF001 — internal state check + assert budget._deposits > 0 # noqa: SLF001 @pytest.mark.stress @@ -62,6 +62,29 @@ def worker() -> None: t.join() # With no withdraws and no TTL expiry (60s window, sub-second test), every - # deposit lands in the deque. Exact equality proves no deposits were lost + # deposit is counted. Exact equality proves no deposits were lost # to a race. - assert len(budget._deposits) == _N_THREADS * _N_OPS_PER_THREAD # noqa: SLF001 + assert budget._deposits == _N_THREADS * _N_OPS_PER_THREAD # noqa: SLF001 + + +@pytest.mark.stress +def test_concurrent_bucket_rotation_keeps_totals_consistent() -> None: + budget = RetryBudget(ttl=0.01, min_retries_per_sec=1000.0, percent_can_retry=0.5) + barrier = threading.Barrier(_N_THREADS) + + def worker() -> None: + barrier.wait() + for _ in range(_N_OPS_PER_THREAD): + budget.deposit() + budget.try_withdraw() + + threads = [threading.Thread(target=worker) for _ in range(_N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert budget._deposits == sum(budget._bucket_deposits) # noqa: SLF001 + assert budget._withdrawn == sum(budget._bucket_withdrawn) # noqa: SLF001 + assert min(budget._bucket_deposits) >= 0 # noqa: SLF001 + assert min(budget._bucket_withdrawn) >= 0 # noqa: SLF001 diff --git a/tests/test_retry_sync.py b/tests/test_retry_sync.py index f5d0a2d..a229a02 100644 --- a/tests/test_retry_sync.py +++ b/tests/test_retry_sync.py @@ -625,7 +625,7 @@ def test_retry_after_exceeds_max_delay_does_not_consume_budget_token_sync() -> N with pytest.raises(ServiceUnavailableError): 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 def test_retry_after_huge_digit_string_gives_up_sync() -> None: diff --git a/tests/test_threading_with_shared_budget.py b/tests/test_threading_with_shared_budget.py index c488f58..5a5f2b1 100644 --- a/tests/test_threading_with_shared_budget.py +++ b/tests/test_threading_with_shared_budget.py @@ -1,7 +1,7 @@ """Demonstrates that a single RetryBudget can be shared across a sync Client and an AsyncClient. The lock added in Task B2 makes ``RetryBudget`` thread-safe so sync threads and an asyncio -event loop can deposit/withdraw concurrently without corrupting the internal deques. +event loop can deposit/withdraw concurrently without corrupting the internal counters. """ import asyncio @@ -54,14 +54,14 @@ async def _drive_async_side(budget: RetryBudget) -> None: def _fixed_clock() -> float: - """Return a constant timestamp so RetryBudget._purge never evicts deposits during this test.""" + """Return a constant timestamp so RetryBudget never expires a deposit during this test.""" return 0.0 def test_shared_budget_across_sync_threads_and_async_loop() -> None: - # _now is pinned to a fixed timestamp: all deposits share the same timestamp, - # so the TTL window cutoff is also 0.0 and _purge evicts nothing. This makes - # the deposit-count assertion exact and independent of wall-clock elapsed time. + # _now is pinned to a fixed timestamp: all deposits land in the same bucket and none + # expires. This makes the deposit-count assertion exact and independent of wall-clock + # elapsed time. budget = RetryBudget(ttl=60.0, min_retries_per_sec=1000.0, percent_can_retry=0.5, _now=_fixed_clock) sync_transport = httpx2.MockTransport(_failing_handler) @@ -79,14 +79,13 @@ def test_shared_budget_across_sync_threads_and_async_loop() -> None: for t in threads: t.join() - # The lock kept the budget's internal deques consistent — no IndexError, no corruption. + # The lock kept the budget's internal counters consistent, with no corruption. # 0.8.3 deposit-hoist: deposits count requests, not attempts (one per __call__, - # regardless of max_attempts). The pinned clock ensures _purge never evicts any - # deposit (all timestamps are 0.0, cutoff is 0.0 - 60.0 = -60.0 < 0.0, so the - # strict `< cutoff` predicate is always False), making the deposit count exact. + # regardless of max_attempts). The pinned clock keeps every deposit in one live + # bucket, making the deposit count exact. expected_deposits = (_N_SYNC_THREADS * _N_OPS_PER_THREAD) + _N_ASYNC_TASKS - assert len(budget._deposits) == expected_deposits, ( # noqa: SLF001 - f"expected {expected_deposits} deposits, got {len(budget._deposits)}" # noqa: SLF001 + assert budget._deposits == expected_deposits, ( # noqa: SLF001 + f"expected {expected_deposits} deposits, got {budget._deposits}" # noqa: SLF001 ) sync_client.close()