From a7467999e6bf875619ad34b492e71142193e57a1 Mon Sep 17 00:00:00 2001 From: bagowix Date: Tue, 1 Sep 2026 12:32:24 +0400 Subject: [PATCH 1/4] feat: let a breaker recover when its probes never reach the dependency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A HALF_OPEN probe asks one question — has the dependency recovered? — and every failure was taken as its answer, including failures that never left the process: no free connection in the local pool, no bulkhead permit. Whenever the local cause outlives the outage that opened the breaker, the question becomes unanswerable — each probe round fails on the local cause, the breaker returns to OPEN, and the cycle repeats for the life of the process, however healthy the dependency has become. Only a restart clears it. Such a call is now recorded through the new `unreachable` flag on StateMachine.record, which hands the probe's slot back without a verdict instead of counting a failure the probe never observed. Recording it as a success would be the mirror-image lie, so the outcome is dropped. The round still has to end: once as many probes have come back inconclusive as the round permits, the breaker re-opens, because nothing was learned and waiting is the only honest move left. CLOSED is deliberately untouched — there an exhausted pool is usually the dependency holding connections open, and shedding load is exactly what should happen. The httpx and httpx2 transports pass PoolTimeout that way out of the box; other guards take the set through `unreachable_exceptions` on CircuitBreaker, Registry and Engine. Alongside it, the open wait can now grow. wait_duration_in_open was a constant, so a breaker that could not recover retried at full rate indefinitely, each round hammering a dependency already in trouble. Config.wait_duration_backoff_multiplier lengthens the wait after each consecutive failed round and Config.wait_duration_in_open_max caps it; a passing round resets both. The default multiplier of 1.0 keeps the historical constant wait, so nothing changes until it is raised. --- CHANGELOG.md | 27 ++++ docs/guides/configuration.md | 2 + docs/guides/states.md | 58 +++++++ docs/llms-full.txt | 60 ++++++++ interlock/_engine.py | 54 ++++++- interlock/_state_machine.py | 66 +++++++- interlock/breaker.py | 9 ++ interlock/config.py | 24 ++- interlock/integrations/_registry.py | 2 + interlock/integrations/httpx.py | 12 ++ interlock/integrations/httpx2.py | 7 + interlock/registry.py | 11 ++ tests/test_auto_transition.py | 57 +++++++ tests/test_config.py | 17 +++ tests/test_engine.py | 126 ++++++++++++++++ tests/test_httpx.py | 33 ++++ tests/test_httpx2.py | 33 ++++ tests/test_state_machine.py | 224 ++++++++++++++++++++++++++++ 18 files changed, 814 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ace6df..6e1c4fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,33 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Fixed +- **A probe that never reached the dependency no longer decides the round.** A + `HALF_OPEN` probe asks one question — has the dependency recovered? — and + every failure was taken as its answer, including failures that never left the + process: no free connection in the local pool, no bulkhead permit. A single + such probe could re-open the breaker on evidence it did not have, and while + the local cause persisted every round failed the same way, however healthy + the dependency had become. Such a call + is now recorded through the new `unreachable` flag on `StateMachine.record`, + which hands the probe's slot back without a verdict instead of counting a + failure the probe never observed. The httpx and httpx2 transports pass + `PoolTimeout` that way out of the box; other integrations take the set + through `unreachable_exceptions` on `CircuitBreaker`, `Registry` and + `Engine`. `CLOSED` is deliberately untouched — an exhausted pool is a real + signal there, and shedding load is the point. + +### Added + +- **The open wait can now grow while probe rounds keep failing.** + `wait_duration_in_open` was a constant, so a breaker that could not recover + retried at full rate indefinitely, each round hammering a dependency already + in trouble. `Config.wait_duration_backoff_multiplier` lengthens the wait + after each consecutive failed round and `Config.wait_duration_in_open_max` + caps it; a passing round resets both. The default multiplier of `1.0` keeps + the historical constant wait, so nothing changes until it is raised. The + growing interval doubles as a signal: a breaker that is merely waiting out a + blip looks nothing like one that has failed ten rounds in a row. + - **A Dependabot pull request no longer fails CI on the Codecov upload.** A run triggered by Dependabot resolves `secrets.*` against the separate Dependabot secret store, so `CODECOV_TOKEN` has to be maintained in two places — and a diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index 432c464..0458bae 100644 --- a/docs/guides/configuration.md +++ b/docs/guides/configuration.md @@ -32,6 +32,8 @@ config = Config( | `permitted_calls_in_half_open` | `10` | Probe calls allowed while `HALF_OPEN`. | | `max_concurrent_probes` | `1` | Cap on **simultaneous** probes in `HALF_OPEN`. Must be in `[1, permitted_calls_in_half_open]`. | | `wait_duration_in_open` | `60.0` | Seconds to stay `OPEN` before the first probe is allowed. | +| `wait_duration_backoff_multiplier` | `1.0` | Multiplies the open wait after each consecutive failed probe round. `1.0` keeps it constant. Must be `>= 1`. | +| `wait_duration_in_open_max` | `None` | Ceiling for the backed-off wait, in seconds. `None` leaves it uncapped; when set, must be `>= wait_duration_in_open`. | | `auto_transition` | `False` | When `True`, a timer moves the breaker `OPEN → HALF_OPEN` once the wait elapses, instead of waiting for the next call. See [States](states.md#proactive-transition-auto_transition). | | `window_type` | `COUNT_BASED` | `COUNT_BASED` or `TIME_BASED`. | | `window_size` | `100` | Last N calls (count-based) or last N seconds (time-based). | diff --git a/docs/guides/states.md b/docs/guides/states.md index 61b779c..84e143c 100644 --- a/docs/guides/states.md +++ b/docs/guides/states.md @@ -26,6 +26,64 @@ stateDiagram-v2 thresholds as `CLOSED`: rates below the thresholds close it, at or above re-open it. Calls beyond the probe caps are rejected while the round runs. +## When a probe cannot reach the dependency + +A probe asks one question: has the dependency recovered? Some failures cannot +answer it, because the call never left the process — no free connection in the +local pool, no bulkhead permit. Counting one as a probe failure re-opens the +breaker on evidence it does not have, and if the local cause outlives the outage +that opened the breaker, every round fails the same way and the breaker never +closes again. + +Integrations mark those failures for you. The httpx and httpx2 transports treat +`PoolTimeout` this way: in `HALF_OPEN` the probe hands its slot back without a +verdict, and the round continues. `CLOSED` is untouched — there an exhausted +pool usually *is* the dependency holding connections open, and shedding load is +exactly what should happen. + +A round still has to end. Once as many probes have come back inconclusive as the +round permits, the breaker re-opens: nothing was learned, so waiting is the only +honest move left. + +Other guards pass their own set: + +```python +from interlock import CircuitBreaker, Registry + + +class NoLocalSlot(Exception): + """Raised by the guard's own pool when it has no permit to give out.""" + + +breaker = CircuitBreaker(name='payments', unreachable_exceptions=(NoLocalSlot,)) +registry = Registry(unreachable_exceptions=(NoLocalSlot,)) +``` + +## Backing off between probe rounds + +`wait_duration_in_open` is constant by default: a breaker that cannot recover +retries at exactly the same rate forever, hammering a dependency that is already +in trouble. Set `wait_duration_backoff_multiplier` above `1.0` to lengthen the +wait after each consecutive failed round, and `wait_duration_in_open_max` to cap +it. A round that passes resets both. + +```python +from interlock import CircuitBreaker, Config + +breaker = CircuitBreaker( + name='payments', + config=Config( + wait_duration_in_open=5.0, + wait_duration_backoff_multiplier=2.0, + wait_duration_in_open_max=120.0, + ), +) +# Failed rounds wait 5s, then 10s, 20s, 40s… up to 120s. +``` + +The growing interval is also a signal in its own right: a breaker waiting out a +blip looks nothing like one that has failed ten rounds in a row. + ## Proactive transition (`auto_transition`) By default the `OPEN → HALF_OPEN` move is **lazy**: it happens on the first call diff --git a/docs/llms-full.txt b/docs/llms-full.txt index ae07ccb..7b3dd29 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -1516,6 +1516,8 @@ config = Config( | `permitted_calls_in_half_open` | `10` | Probe calls allowed while `HALF_OPEN`. | | `max_concurrent_probes` | `1` | Cap on **simultaneous** probes in `HALF_OPEN`. Must be in `[1, permitted_calls_in_half_open]`. | | `wait_duration_in_open` | `60.0` | Seconds to stay `OPEN` before the first probe is allowed. | +| `wait_duration_backoff_multiplier` | `1.0` | Multiplies the open wait after each consecutive failed probe round. `1.0` keeps it constant. Must be `>= 1`. | +| `wait_duration_in_open_max` | `None` | Ceiling for the backed-off wait, in seconds. `None` leaves it uncapped; when set, must be `>= wait_duration_in_open`. | | `auto_transition` | `False` | When `True`, a timer moves the breaker `OPEN → HALF_OPEN` once the wait elapses, instead of waiting for the next call. See [States](states.md#proactive-transition-auto_transition). | | `window_type` | `COUNT_BASED` | `COUNT_BASED` or `TIME_BASED`. | | `window_size` | `100` | Last N calls (count-based) or last N seconds (time-based). | @@ -1598,6 +1600,64 @@ stateDiagram-v2 thresholds as `CLOSED`: rates below the thresholds close it, at or above re-open it. Calls beyond the probe caps are rejected while the round runs. +## When a probe cannot reach the dependency + +A probe asks one question: has the dependency recovered? Some failures cannot +answer it, because the call never left the process — no free connection in the +local pool, no bulkhead permit. Counting one as a probe failure re-opens the +breaker on evidence it does not have, and if the local cause outlives the outage +that opened the breaker, every round fails the same way and the breaker never +closes again. + +Integrations mark those failures for you. The httpx and httpx2 transports treat +`PoolTimeout` this way: in `HALF_OPEN` the probe hands its slot back without a +verdict, and the round continues. `CLOSED` is untouched — there an exhausted +pool usually *is* the dependency holding connections open, and shedding load is +exactly what should happen. + +A round still has to end. Once as many probes have come back inconclusive as the +round permits, the breaker re-opens: nothing was learned, so waiting is the only +honest move left. + +Other guards pass their own set: + +```python +from interlock import CircuitBreaker, Registry + + +class NoLocalSlot(Exception): + """Raised by the guard's own pool when it has no permit to give out.""" + + +breaker = CircuitBreaker(name='payments', unreachable_exceptions=(NoLocalSlot,)) +registry = Registry(unreachable_exceptions=(NoLocalSlot,)) +``` + +## Backing off between probe rounds + +`wait_duration_in_open` is constant by default: a breaker that cannot recover +retries at exactly the same rate forever, hammering a dependency that is already +in trouble. Set `wait_duration_backoff_multiplier` above `1.0` to lengthen the +wait after each consecutive failed round, and `wait_duration_in_open_max` to cap +it. A round that passes resets both. + +```python +from interlock import CircuitBreaker, Config + +breaker = CircuitBreaker( + name='payments', + config=Config( + wait_duration_in_open=5.0, + wait_duration_backoff_multiplier=2.0, + wait_duration_in_open_max=120.0, + ), +) +# Failed rounds wait 5s, then 10s, 20s, 40s… up to 120s. +``` + +The growing interval is also a signal in its own right: a breaker waiting out a +blip looks nothing like one that has failed ten rounds in a row. + ## Proactive transition (`auto_transition`) By default the `OPEN → HALF_OPEN` move is **lazy**: it happens on the first call diff --git a/interlock/_engine.py b/interlock/_engine.py index ec27dc0..d48a2b1 100644 --- a/interlock/_engine.py +++ b/interlock/_engine.py @@ -54,6 +54,28 @@ _MANUAL_STATES = frozenset({State.FORCED_OPEN, State.DISABLED, State.METRICS_ONLY}) +def validate_unreachable_exceptions( + types: tuple[type[Exception], ...], +) -> tuple[type[Exception], ...]: + """Reject entries that could never be classified, before any call runs. + + The annotation is a promise, not a guarantee: an untyped caller can still + pass a string. Left to ``_settle``, the ``isinstance`` would raise on the + first real failure — masking the protected exception and stranding the probe + slot it should have returned. + + Raises: + TypeError: If an entry is not an ``Exception`` subclass. + """ + for entry in cast('tuple[object, ...]', types): + if not (isinstance(entry, type) and issubclass(entry, Exception)): + raise TypeError( + f'unreachable_exceptions must hold Exception subclasses, got: {entry!r}' + ) + + return types + + @dataclass(frozen=True, slots=True) class Admission: """What ``_admit`` granted: the era it happened in, and probe provenance.""" @@ -91,11 +113,13 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, storage: Storage | AsyncStorage | None = None, + unreachable_exceptions: tuple[type[Exception], ...] = (), ) -> None: self._name = name self._config = config self._clock = clock self._classifier = classifier if classifier is not None else DefaultFailureClassifier() + self._unreachable_exceptions = validate_unreachable_exceptions(unreachable_exceptions) self._listener = listener self._machine = StateMachine( config=config, @@ -464,11 +488,26 @@ def _settle( failure = self._classifier.is_failure(result=result, exception=exception) slow = duration >= self._config.slow_call_duration_threshold outcome = _OUTCOME_BY_FLAGS[failure, slow] + coordinator = self._sync_coordinator or self._async_coordinator + # Only HALF_OPEN cares; the machine owns that check because only it knows + # its own state. Outside a probe the same exception stays an ordinary + # failure — shedding load is the point. The ``is not None`` guard keeps + # the common path — a call that returned — off ``isinstance``. + # + # A coordinated probe holds a lease that only an outcome returns, and + # ``Storage`` has no way to hand one back unspent, so dropping the verdict + # there would strand the shared budget until its TTL. Until the protocol + # grows that operation, a shared probe keeps the old behaviour. + unreachable = ( + coordinator is None + and exception is not None + and isinstance(exception, self._unreachable_exceptions) + ) with self._lock: effective_before = self._effective_state_locked() before = self._machine.state - self._machine.record(outcome, generation=admission.generation) + self._machine.record(outcome, generation=admission.generation, unreachable=unreachable) after = self._machine.state effective_after = self._effective_state_locked() if failure and exception is not None: @@ -481,7 +520,6 @@ def _settle( notify(self._listener, 'on_call', name=self._name, outcome=outcome, duration=duration) self._emit_transitions(before, after, effective_before, effective_after) - coordinator = self._sync_coordinator or self._async_coordinator if coordinator is not None: if admission.probe: coordinator.notify_probe_outcome(outcome) @@ -590,11 +628,21 @@ def _schedule_auto_transition(self) -> None: pending one never blocks interpreter shutdown, and it is cancelled the moment the breaker leaves ``OPEN`` by any path (a real probe, ``reset``, ``force_open`` or the timer itself). + + The delay comes from ``retry_after()`` rather than the configured wait: + with a backoff in play they differ, and a timer armed for the base wait + would fire early, be refused, and leave nothing scheduled for the rest of + the interval. """ if not self._config.auto_transition: return - timer = threading.Timer(self._config.wait_duration_in_open, self._fire_auto_transition) + with self._lock: + delay = self._machine.retry_after() + if delay is None: + return # left OPEN before the timer could be armed + + timer = threading.Timer(delay, self._fire_auto_transition) timer.daemon = True with self._timer_lock: if self._closed: diff --git a/interlock/_state_machine.py b/interlock/_state_machine.py index 6dc0006..0db426e 100644 --- a/interlock/_state_machine.py +++ b/interlock/_state_machine.py @@ -57,6 +57,7 @@ def __init__( self._state = initial_state self._opened_at = 0.0 self._generation = 0 + self._failed_rounds = 0 self._reset_probes() @property @@ -91,7 +92,7 @@ def retry_after(self) -> float | None: return None elapsed = self._clock.monotonic() - self._opened_at - return max(0.0, self._config.wait_duration_in_open - elapsed) + return max(0.0, self._wait_duration() - elapsed) def acquire(self) -> bool: """Decide whether one call may proceed, mutating state lazily. @@ -111,11 +112,20 @@ def acquire(self) -> bool: return False # FORCED_OPEN - def record(self, outcome: Outcome, *, generation: int | None = None) -> None: + def record( + self, outcome: Outcome, *, generation: int | None = None, unreachable: bool = False + ) -> None: """Record one completed call's outcome and evaluate any transition. ``generation`` is the era captured at admission; an outcome from an earlier era is dropped (see ``generation``). ``None`` skips the check. + + ``unreachable`` marks a call that failed without reaching the dependency + — no local connection slot, no bulkhead permit. In ``CLOSED`` it changes + nothing: the caller cannot serve traffic either way, and shedding load is + the point. It only matters for a probe, which asks the narrower question + "has the dependency recovered?" and cannot answer it from a call that + never left the process. """ if generation is not None and generation != self._generation: return @@ -124,7 +134,10 @@ def record(self, outcome: Outcome, *, generation: int | None = None) -> None: self._window.record(outcome) self._evaluate_closed() elif self._state is State.HALF_OPEN: - self._record_probe(outcome) + if unreachable: + self._record_inconclusive_probe() + else: + self._record_probe(outcome) elif self._state is State.METRICS_ONLY: self._window.record(outcome) @@ -195,7 +208,23 @@ def _begin_probing_if_elapsed(self) -> bool: return self._admit_probe() def _wait_elapsed(self) -> bool: - return self._clock.monotonic() - self._opened_at >= self._config.wait_duration_in_open + return self._clock.monotonic() - self._opened_at >= self._wait_duration() + + def _wait_duration(self) -> float: + """How long this OPEN spell lasts, lengthened by consecutive failed rounds. + + The first spell after a trip — every spell, with the default multiplier + of ``1.0`` — is exactly ``wait_duration_in_open``, so the exponentiation + is skipped rather than computed as a no-op on the rejection path. + """ + if not self._failed_rounds: + return self._config.wait_duration_in_open + + duration = self._config.wait_duration_in_open * ( + self._config.wait_duration_backoff_multiplier**self._failed_rounds + ) + ceiling = self._config.wait_duration_in_open_max + return duration if ceiling is None else min(duration, ceiling) def _admit_probe(self) -> bool: # Cap total probes (don't hammer a barely-recovered dependency) and how @@ -230,6 +259,28 @@ def _record_probe(self, outcome: Outcome) -> None: if self._probes_completed >= self._config.permitted_calls_in_half_open: self._evaluate_probes() + def _record_inconclusive_probe(self) -> None: + """Hand a probe's slot back without a verdict, and count it against the round. + + Counting an unreachable probe as a failure re-opens the breaker on + evidence it does not have; counting it as a success would be the + mirror-image lie. So the outcome is dropped and the slot returned — but + the round still has to end. Once as many probes have come back + inconclusive as the round permits, waiting is the only honest move left, + so the breaker re-opens, and the backoff applies because nothing was + learned. A probe still running is left to answer first: re-opening now + would bump the generation and throw its verdict away. + """ + self._probes_in_flight -= 1 + self._probes_admitted -= 1 + self._probes_inconclusive += 1 + + if ( + self._probes_inconclusive >= self._config.permitted_calls_in_half_open + and not self._probes_in_flight + ): + self._open() + def _evaluate_probes(self) -> None: completed = self._probes_completed if self._exceeds_threshold( @@ -247,6 +298,11 @@ def _exceeds_threshold(self, *, failure_rate: float, slow_call_rate: float) -> b ) def _open(self) -> None: + # Counted here rather than in ``_evaluate_probes`` so that every route + # back to OPEN from a probe round feeds the backoff, including a round + # that ended with nothing learned. + if self._state is State.HALF_OPEN: + self._failed_rounds += 1 self._state = State.OPEN self._opened_at = self._clock.monotonic() self._generation += 1 @@ -258,6 +314,7 @@ def _to_half_open(self) -> None: def _close(self) -> None: self._state = State.CLOSED + self._failed_rounds = 0 self._window = build_window(config=self._config, clock=self._clock) self._generation += 1 self._reset_probes() @@ -268,3 +325,4 @@ def _reset_probes(self) -> None: self._probes_completed = 0 self._probe_failures = 0 self._probe_slows = 0 + self._probes_inconclusive = 0 diff --git a/interlock/breaker.py b/interlock/breaker.py index a5c0839..7aed8a7 100644 --- a/interlock/breaker.py +++ b/interlock/breaker.py @@ -61,6 +61,13 @@ class CircuitBreaker: state. A coordinated breaker matches its storage's runtime: a sync ``Storage`` serves only the sync API, an ``AsyncStorage`` only the async one; without a storage the breaker stays fully dual. + unreachable_exceptions: Exception types that mean the call never reached + the dependency — no local connection slot, no bulkhead permit. In + ``HALF_OPEN`` such a probe returns its slot without a verdict rather + than counting as a failure it never observed; ``CLOSED`` is + unaffected. Defaults to none. A coordinated (storage-backed) probe + keeps the plain behaviour: its lease can only be returned by an + outcome. Raises: ValueError: If ``initial_state`` is not a supported stable state. @@ -76,6 +83,7 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, storage: Storage | AsyncStorage | None = None, + unreachable_exceptions: tuple[type[Exception], ...] = (), ) -> None: self._name = name self._engine = Engine( @@ -86,6 +94,7 @@ def __init__( classifier=classifier, listener=listener, storage=storage, + unreachable_exceptions=unreachable_exceptions, ) # Guarded-block bookkeeping lives in a ContextVar, not on the instance: # each thread and each asyncio task sees its own stack, so overlapping diff --git a/interlock/config.py b/interlock/config.py index 5e6a9dc..5f8d988 100644 --- a/interlock/config.py +++ b/interlock/config.py @@ -19,6 +19,13 @@ class Config: treat calls slower than 60s as slow (but never trip on slowness alone until tuned), stay open 60s before a single probe is allowed. + ``wait_duration_backoff_multiplier`` lengthens the open wait after each probe + round that fails back-to-back, capped by ``wait_duration_in_open_max``. It + defaults to ``1.0`` — a constant wait, the historical behaviour. Raise it when + a breaker can fail its probes for a reason waiting alone will not fix: a + constant wait then retries forever at full rate, and the growing interval is + itself the signal that the dependency is not merely slow to recover. + ``auto_transition`` opts into a timer that proactively moves a breaker from ``OPEN`` to ``HALF_OPEN`` once ``wait_duration_in_open`` elapses, emitting the state change without waiting for the next call. It defaults to ``False``, @@ -35,11 +42,13 @@ class Config: permitted_calls_in_half_open: int = 10 max_concurrent_probes: int = 1 wait_duration_in_open: float = 60.0 + wait_duration_backoff_multiplier: float = 1.0 + wait_duration_in_open_max: float | None = None auto_transition: bool = False window_type: WindowType = WindowType.COUNT_BASED window_size: int = 100 - def __post_init__(self) -> None: + def __post_init__(self) -> None: # noqa: C901 - a flat list of guards, not branching logic if not 0.0 < self.failure_rate_threshold <= 1.0: raise ValueError( f'failure_rate_threshold must be in (0, 1], got {self.failure_rate_threshold!r}' @@ -61,6 +70,19 @@ def __post_init__(self) -> None: raise ValueError( f'wait_duration_in_open must be > 0, got {self.wait_duration_in_open!r}' ) + if self.wait_duration_backoff_multiplier < 1.0: + raise ValueError( + f'wait_duration_backoff_multiplier must be >= 1, ' + f'got {self.wait_duration_backoff_multiplier!r}' + ) + if ( + self.wait_duration_in_open_max is not None + and self.wait_duration_in_open_max < self.wait_duration_in_open + ): + raise ValueError( + f'wait_duration_in_open_max must be >= wait_duration_in_open ' + f'({self.wait_duration_in_open!r}), got {self.wait_duration_in_open_max!r}' + ) if self.permitted_calls_in_half_open < 1: raise ValueError( f'permitted_calls_in_half_open must be >= 1, ' diff --git a/interlock/integrations/_registry.py b/interlock/integrations/_registry.py index a9d3133..9eca5ef 100644 --- a/interlock/integrations/_registry.py +++ b/interlock/integrations/_registry.py @@ -45,6 +45,7 @@ def resolve_registry( # noqa: PLR0913 - mirrors the integration constructors classifier: FailureClassifier | None, listener: CoreEventListener | StorageEventListener | None, default_classifier: FailureClassifier, + unreachable_exceptions: tuple[type[Exception], ...] = (), ) -> tuple[Registry, bool]: """Return the effective registry and whether the integration owns it.""" if registry is not None: @@ -57,6 +58,7 @@ def resolve_registry( # noqa: PLR0913 - mirrors the integration constructors initial_state=initial_state, classifier=classifier if classifier is not None else default_classifier, listener=listener, + unreachable_exceptions=unreachable_exceptions, ), True, ) diff --git a/interlock/integrations/httpx.py b/interlock/integrations/httpx.py index ff09e5c..51bda78 100644 --- a/interlock/integrations/httpx.py +++ b/interlock/integrations/httpx.py @@ -88,6 +88,16 @@ UnsupportedProtocol, ) +# A ``PoolTimeout`` in CLOSED is a real signal — see ``_EXCLUDED_EXCEPTIONS``, +# an exhausted pool is usually the dependency holding connections open, and +# shedding load then is the point. A HALF_OPEN probe asks a narrower question: +# has the dependency recovered? A request that never got a connection out of +# the local pool never asked it, so it cannot answer. Counting it re-opens the +# breaker on evidence it does not have, and a pool that stays saturated (a +# noisy neighbour on the same client, a leak) then wedges the breaker open for +# good. Listed here, such a probe returns its slot instead. +_UNREACHABLE_EXCEPTIONS: tuple[type[Exception], ...] = (PoolTimeout,) + def _exception_types( excluded: Iterable[type[Exception]] | None, @@ -350,6 +360,7 @@ def __init__( classifier=classifier, listener=listener, default_classifier=HttpStatusClassifier(), + unreachable_exceptions=_UNREACHABLE_EXCEPTIONS, ) @property @@ -454,6 +465,7 @@ def __init__( classifier=classifier, listener=listener, default_classifier=HttpStatusClassifier(), + unreachable_exceptions=_UNREACHABLE_EXCEPTIONS, ) @property diff --git a/interlock/integrations/httpx2.py b/interlock/integrations/httpx2.py index b3db353..751bef4 100644 --- a/interlock/integrations/httpx2.py +++ b/interlock/integrations/httpx2.py @@ -87,6 +87,11 @@ UnsupportedProtocol, ) +# See the httpx integration for the reasoning: an exhausted pool is a real +# signal in CLOSED, but a HALF_OPEN probe that never got a connection never +# reached the dependency and cannot report on it. +_UNREACHABLE_EXCEPTIONS: tuple[type[Exception], ...] = (PoolTimeout,) + def _exception_types( excluded: Iterable[type[Exception]] | None, @@ -350,6 +355,7 @@ def __init__( classifier=classifier, listener=listener, default_classifier=HttpStatusClassifier(), + unreachable_exceptions=_UNREACHABLE_EXCEPTIONS, ) @property @@ -455,6 +461,7 @@ def __init__( classifier=classifier, listener=listener, default_classifier=HttpStatusClassifier(), + unreachable_exceptions=_UNREACHABLE_EXCEPTIONS, ) @property diff --git a/interlock/registry.py b/interlock/registry.py index 81413d1..cd9e9e8 100644 --- a/interlock/registry.py +++ b/interlock/registry.py @@ -9,6 +9,7 @@ from contextlib import AsyncExitStack, ExitStack from interlock._clock import SystemClock +from interlock._engine import validate_unreachable_exceptions from interlock._initial_state import validate_initial_state from interlock.breaker import CircuitBreaker from interlock.config import Config @@ -41,6 +42,13 @@ class Registry: no observation. storage: Shared backend for coordinated state, handed to every breaker (each coordinates under its own name). Defaults to local state. + unreachable_exceptions: Exception types that mean the call never reached + the dependency — no local connection slot, no bulkhead permit. In + ``HALF_OPEN`` such a probe returns its slot without a verdict rather + than counting as a failure it never observed; ``CLOSED`` is + unaffected. Defaults to none. A coordinated (storage-backed) probe + keeps the plain behaviour: its lease can only be returned by an + outcome. Raises: ValueError: If ``initial_state`` is not a supported stable state. @@ -55,6 +63,7 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, storage: Storage | AsyncStorage | None = None, + unreachable_exceptions: tuple[type[Exception], ...] = (), ) -> None: validate_initial_state(initial_state) self._config = config if config is not None else Config() @@ -63,6 +72,7 @@ def __init__( self._classifier = classifier self._listener = listener self._storage = storage + self._unreachable_exceptions = validate_unreachable_exceptions(unreachable_exceptions) self._breakers: dict[str, CircuitBreaker] = {} self._lock = threading.Lock() @@ -96,6 +106,7 @@ def get(self, name: str, *, config: Config | None = None) -> CircuitBreaker: classifier=self._classifier, listener=self._listener, storage=self._storage, + unreachable_exceptions=self._unreachable_exceptions, ) self._breakers[name] = breaker diff --git a/tests/test_auto_transition.py b/tests/test_auto_transition.py index 2db1224..84fe5a7 100644 --- a/tests/test_auto_transition.py +++ b/tests/test_auto_transition.py @@ -262,3 +262,60 @@ def test__timer__survives_storage_degradation_and_recovery(timed_engine: Engine) timed_engine._on_storage_recovered() assert timed_engine._timer is armed + + +def test__schedule__backoff_in_play__timer_waits_the_grown_interval( + fake_clock: FakeClock, +) -> None: + """The timer must follow ``retry_after()``, not the configured base wait. + + Armed for the base wait, it would fire while the backoff still had time to + run, be refused by ``attempt_auto_transition``, and leave nothing scheduled + for the remainder — the breaker would then sit in ``OPEN`` until a caller + happened to arrive. + """ + engine = Engine( + name='t', + config=Config( + minimum_number_of_calls=2, + window_size=10, + permitted_calls_in_half_open=1, + max_concurrent_probes=1, + wait_duration_in_open=100.0, + wait_duration_backoff_multiplier=3.0, + auto_transition=True, + ), + clock=fake_clock, + ) + + def boom() -> None: + raise ValueError('boom') + + for _ in range(2): + with pytest.raises(ValueError, match='boom'): + engine.call_sync(boom) + assert engine._timer is not None + assert engine._timer.interval == pytest.approx(100.0) + + fake_clock.advance(100.0) + with pytest.raises(ValueError, match='boom'): + engine.call_sync(boom) # the probe fails, the round with it + + assert engine.state is State.OPEN + assert engine._timer is not None + assert engine._timer.interval == pytest.approx(300.0) + engine.close() + + +def test__schedule__no_longer_open__arms_nothing(fake_clock: FakeClock) -> None: + """``retry_after()`` is estimable only in ``OPEN``; anywhere else, stand down.""" + engine = Engine( + name='t', + config=Config(wait_duration_in_open=100.0, auto_transition=True), + clock=fake_clock, + ) + + engine._schedule_auto_transition() # the machine is CLOSED + + assert engine._timer is None + engine.close() diff --git a/tests/test_config.py b/tests/test_config.py index 9aee411..6c9a2aa 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -63,3 +63,20 @@ def test__config__auto_transition__defaults_to_false() -> None: def test__config__auto_transition__can_be_enabled() -> None: assert Config(auto_transition=True).auto_transition is True + + +def test__config__backoff_multiplier_below_one__raises() -> None: + with pytest.raises(ValueError, match='wait_duration_backoff_multiplier'): + Config(wait_duration_backoff_multiplier=0.9) + + +def test__config__wait_duration_max_below_base__raises() -> None: + with pytest.raises(ValueError, match='wait_duration_in_open_max'): + Config(wait_duration_in_open=60.0, wait_duration_in_open_max=30.0) + + +def test__config__backoff_defaults__preserve_constant_wait() -> None: + config = Config() + + assert config.wait_duration_backoff_multiplier == 1.0 + assert config.wait_duration_in_open_max is None diff --git a/tests/test_engine.py b/tests/test_engine.py index e8be501..cde5aaf 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -1,4 +1,5 @@ import asyncio +from typing import cast import pytest @@ -407,3 +408,128 @@ async def echo(first: int, *, second: int) -> tuple[int, int]: return first, second assert await engine.call_async(echo, 1, second=2) == (1, 2) + + +class _NoLocalSlot(Exception): + """Stands in for a pool/bulkhead exhaustion that never reaches the dependency.""" + + +def _unreachable_engine(config: Config, clock: FakeClock) -> Engine: + return Engine( + name='test', + config=config, + clock=clock, + classifier=DefaultFailureClassifier(), + unreachable_exceptions=(_NoLocalSlot,), + ) + + +def _raise_no_slot() -> None: + raise _NoLocalSlot + + +def test__closed__unreachable_exception__still_trips_the_breaker( + config: Config, fake_clock: FakeClock +) -> None: + engine = _unreachable_engine(config, fake_clock) + + for _ in range(2): + with pytest.raises(_NoLocalSlot): + engine.call_sync(_raise_no_slot) + + assert engine.state is State.OPEN + + +def test__half_open__unreachable_probe__does_not_reopen( + config: Config, fake_clock: FakeClock +) -> None: + engine = _unreachable_engine(config, fake_clock) + _trip_to_open(engine) + fake_clock.advance(5.0) + + with pytest.raises(_NoLocalSlot): + engine.call_sync(_raise_no_slot) + + assert engine.state is State.HALF_OPEN + + +def test__half_open__unreachable_probe_then_success__closes( + config: Config, fake_clock: FakeClock +) -> None: + engine = _unreachable_engine(config, fake_clock) + _trip_to_open(engine) + fake_clock.advance(5.0) + with pytest.raises(_NoLocalSlot): + engine.call_sync(_raise_no_slot) + + for _ in range(2): + engine.call_sync(lambda: 'ok') + + assert engine.state is State.CLOSED + + +def test__half_open__every_probe_unreachable__reopens( + config: Config, fake_clock: FakeClock +) -> None: + engine = _unreachable_engine(config, fake_clock) + _trip_to_open(engine) + fake_clock.advance(5.0) + + for _ in range(config.permitted_calls_in_half_open): + with pytest.raises(_NoLocalSlot): + engine.call_sync(_raise_no_slot) + + assert engine.state is State.OPEN + + +def test__unreachable_not_configured__probe_reopens_as_before( + config: Config, fake_clock: FakeClock +) -> None: + engine = Engine( + name='test', config=config, clock=fake_clock, classifier=DefaultFailureClassifier() + ) + _trip_to_open(engine) + fake_clock.advance(5.0) + + for _ in range(config.permitted_calls_in_half_open): + with pytest.raises(_NoLocalSlot): + engine.call_sync(_raise_no_slot) + + assert engine.state is State.OPEN + + +@pytest.mark.asyncio +async def test__half_open__unreachable_async_probe__does_not_reopen( + config: Config, fake_clock: FakeClock +) -> None: + engine = _unreachable_engine(config, fake_clock) + + async def boom() -> None: + raise ValueError('boom') + + for _ in range(2): + with pytest.raises(ValueError, match='boom'): + await engine.call_async(boom) + fake_clock.advance(5.0) + + async def no_slot() -> None: + raise _NoLocalSlot + + with pytest.raises(_NoLocalSlot): + await engine.call_async(no_slot) + + assert engine.state is State.HALF_OPEN + + +def test__unreachable_exceptions__not_an_exception_type__rejected_at_construction( + config: Config, fake_clock: FakeClock +) -> None: + """A bad entry must fail here, not from inside ``isinstance`` on a real failure.""" + for bad in ('nope', int): # not a type at all, and a type that is not an Exception + with pytest.raises(TypeError, match='unreachable_exceptions'): + Engine( + name='test', + config=config, + clock=fake_clock, + unreachable_exceptions=cast('tuple[type[Exception], ...]', (bad,)), + ) diff --git a/tests/test_httpx.py b/tests/test_httpx.py index fdbc8bb..255bfeb 100644 --- a/tests/test_httpx.py +++ b/tests/test_httpx.py @@ -996,3 +996,36 @@ def boom(_request: httpx.Request) -> httpx.Response: def test__dialect_errors__request_not_set__keep_httpx_contract(error: httpx.HTTPError) -> None: with pytest.raises(RuntimeError, match='has not been set'): _ = error.request + + +def test__half_open__pool_timeout_probe__does_not_reopen_the_breaker() -> None: + """A probe that never got a connection carries no verdict about the host.""" + clock = FakeClock() + config = Config( + minimum_number_of_calls=2, + window_size=10, + permitted_calls_in_half_open=2, + max_concurrent_probes=1, + wait_duration_in_open=5.0, + ) + responses: list[object] = [ + httpx.Response(500), + httpx.Response(500), + httpx.PoolTimeout('no connection available'), + ] + + def handler(request: httpx.Request) -> httpx.Response: # noqa: ARG001 + outcome = responses.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + transport = CircuitBreakerTransport(httpx.MockTransport(handler), config=config, clock=clock) + with httpx.Client(transport=transport) as client: + for _ in range(2): + client.get('https://service.test/path') + clock.advance(5.0) + with pytest.raises(httpx.PoolTimeout): + client.get('https://service.test/path') + + assert transport.registry.get('service.test').state is State.HALF_OPEN diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py index 6828744..2ffb455 100644 --- a/tests/test_httpx2.py +++ b/tests/test_httpx2.py @@ -757,3 +757,36 @@ def boom(_request: Request) -> Response: def test__dialect_errors__request_not_set__keep_httpx2_contract(error: httpx2.HTTPError) -> None: with pytest.raises(RuntimeError, match='has not been set'): _ = error.request + + +def test__half_open__pool_timeout_probe__does_not_reopen_the_breaker() -> None: + """A probe that never got a connection carries no verdict about the host.""" + clock = FakeClock() + config = Config( + minimum_number_of_calls=2, + window_size=10, + permitted_calls_in_half_open=2, + max_concurrent_probes=1, + wait_duration_in_open=5.0, + ) + responses: list[object] = [ + httpx2.Response(500), + httpx2.Response(500), + httpx2.PoolTimeout('no connection available'), + ] + + def handler(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001 + outcome = responses.pop(0) + if isinstance(outcome, Exception): + raise outcome + return cast('httpx2.Response', outcome) + + transport = CircuitBreakerTransport(httpx2.MockTransport(handler), config=config, clock=clock) + with httpx2.Client(transport=transport) as client: + for _ in range(2): + client.get('https://service.test/path') + clock.advance(5.0) + with pytest.raises(httpx2.PoolTimeout): + client.get('https://service.test/path') + + assert transport.registry.get('service.test').state is State.HALF_OPEN diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index f2f604f..65c0b3d 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -646,3 +646,227 @@ def test__close__time_based_window__is_rebuilt_with_the_injected_clock( machine.record(Outcome.SUCCESS) assert machine.snapshot().total_calls == 1 + + +def _fail_probe_round(machine: StateMachine, clock: FakeClock, *, wait: float) -> None: + """Advance past the open wait, then fail every probe of the resulting round.""" + clock.advance(wait) + machine.acquire() + for _ in range(machine._config.permitted_calls_in_half_open): + machine.record(Outcome.FAILURE) + + +def test__open__default_multiplier__wait_duration_stays_constant( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + _trip_to_open(machine, 2) + + for _ in range(3): + _fail_probe_round(machine, fake_clock, wait=5.0) + + assert machine.state is State.OPEN + assert machine.retry_after() == pytest.approx(5.0) + + +def test__open__consecutive_failed_rounds__wait_duration_grows( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine( + config=replace(config, wait_duration_backoff_multiplier=2.0), clock=fake_clock + ) + _trip_to_open(machine, 2) + + assert machine.retry_after() == pytest.approx(5.0) + + _fail_probe_round(machine, fake_clock, wait=5.0) + assert machine.retry_after() == pytest.approx(10.0) + + _fail_probe_round(machine, fake_clock, wait=10.0) + assert machine.retry_after() == pytest.approx(20.0) + + # The third round separates growth from mere multiplication: 5 * 2**3 is 40, + # while 5 * (2 * 3) would be 30. + _fail_probe_round(machine, fake_clock, wait=20.0) + assert machine.retry_after() == pytest.approx(40.0) + + +def test__open__backoff__capped_at_configured_max(config: Config, fake_clock: FakeClock) -> None: + machine = StateMachine( + config=replace( + config, wait_duration_backoff_multiplier=10.0, wait_duration_in_open_max=12.0 + ), + clock=fake_clock, + ) + _trip_to_open(machine, 2) + _fail_probe_round(machine, fake_clock, wait=5.0) + + assert machine.retry_after() == pytest.approx(12.0) + + +def test__half_open__probe_round_passes__backoff_resets( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine( + config=replace(config, wait_duration_backoff_multiplier=2.0), clock=fake_clock + ) + _trip_to_open(machine, 2) + _fail_probe_round(machine, fake_clock, wait=5.0) + assert machine.retry_after() == pytest.approx(10.0) + + fake_clock.advance(10.0) + machine.acquire() + for _ in range(config.permitted_calls_in_half_open): + machine.record(Outcome.SUCCESS) + assert machine.state is State.CLOSED + + _trip_to_open(machine, 2) + assert machine.retry_after() == pytest.approx(5.0) + + +def test__open__grown_wait__probe_not_admitted_before_it_elapses( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine( + config=replace(config, wait_duration_backoff_multiplier=2.0), clock=fake_clock + ) + _trip_to_open(machine, 2) + _fail_probe_round(machine, fake_clock, wait=5.0) + + fake_clock.advance(5.0) # enough for the base wait, not for the doubled one + + assert machine.acquire() is False + assert machine.state is State.OPEN + + +def test__reset__clears_backoff(config: Config, fake_clock: FakeClock) -> None: + machine = StateMachine( + config=replace(config, wait_duration_backoff_multiplier=2.0), clock=fake_clock + ) + _trip_to_open(machine, 2) + _fail_probe_round(machine, fake_clock, wait=5.0) + + machine.reset() + _trip_to_open(machine, 2) + + assert machine.retry_after() == pytest.approx(5.0) + + +def test__half_open__inconclusive_probe__returns_slot_without_verdict( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + machine.acquire() + + machine.record(Outcome.FAILURE, generation=machine.generation, unreachable=True) + + assert machine.state is State.HALF_OPEN + assert machine.acquire() is True + + +def test__half_open__inconclusive_then_successful_probes__closes( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + assert machine.acquire() is True + machine.record(Outcome.FAILURE, generation=machine.generation, unreachable=True) + + for _ in range(config.permitted_calls_in_half_open): + assert machine.acquire() is True + machine.record(Outcome.SUCCESS) + + assert machine.state is State.CLOSED + + +def test__half_open__every_probe_inconclusive__reopens( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + + for _ in range(config.permitted_calls_in_half_open): + assert machine.acquire() is True + machine.record(Outcome.FAILURE, generation=machine.generation, unreachable=True) + + assert machine.state is State.OPEN + + +def test__half_open__inconclusive_probe__stale_generation_ignored( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + machine.acquire() + stale = machine.generation - 1 + + machine.record(Outcome.FAILURE, generation=stale, unreachable=True) + + assert machine.state is State.HALF_OPEN + + +def test__closed__unreachable_outcome__still_counted_in_window( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock) + + for _ in range(2): + machine.record(Outcome.FAILURE, unreachable=True) + + assert machine.state is State.OPEN + + +def test__metrics_only__unreachable_outcome__recorded_without_tripping( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine(config=config, clock=fake_clock, initial_state=State.METRICS_ONLY) + + for _ in range(2): + machine.record(Outcome.FAILURE, unreachable=True) + + assert machine.state is State.METRICS_ONLY + assert machine.snapshot().failure_rate == 1.0 + + +def test__half_open__inconclusive_probe__frees_exactly_one_concurrency_slot( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine( + config=replace(config, permitted_calls_in_half_open=10, max_concurrent_probes=1), + clock=fake_clock, + ) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + assert machine.acquire() is True + assert machine.acquire() is False # concurrency cap reached + + machine.record(Outcome.FAILURE, generation=machine.generation, unreachable=True) + + assert machine.acquire() is True # the in-flight slot came back + assert machine.acquire() is False # and only the one + + +def test__half_open__inconclusive_probe__frees_exactly_one_round_slot( + config: Config, fake_clock: FakeClock +) -> None: + machine = StateMachine( + config=replace(config, permitted_calls_in_half_open=3, max_concurrent_probes=3), + clock=fake_clock, + ) + _trip_to_open(machine, 2) + fake_clock.advance(5.0) + for _ in range(3): + assert machine.acquire() is True + for _ in range(2): + machine.record(Outcome.SUCCESS, generation=machine.generation) + assert machine.acquire() is False # the round's admission budget is spent + + machine.record(Outcome.FAILURE, generation=machine.generation, unreachable=True) + + assert machine.acquire() is True # one admission returned to the round + assert machine.acquire() is False # and only the one From 226facbf3efbbf629430dca904f4f94d8ab91f12 Mon Sep 17 00:00:00 2001 From: bagowix Date: Tue, 1 Sep 2026 13:28:55 +0400 Subject: [PATCH 2/4] fix: bound the backoff exponent so a long outage cannot wedge the breaker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The round counter grew without limit while the wait it feeds was capped, so a ceiling did not protect anything: with `wait_duration_in_open_max` set, the wait stops growing but the exponent does not, and a dependency broken for a few days walks it into float's range. `wait_duration_in_open * multiplier ** rounds` then returns infinity — later raising OverflowError outright — `_wait_elapsed()` is never true again, and the breaker stays OPEN for the life of the process. That is the exact failure this branch set out to remove, reintroduced through the mechanism meant to soften it. Growth now stops at 64 rounds. Any sane multiplier has passed every ceiling long before, so nothing is lost, and the exponent stays far from the range where the arithmetic breaks. --- docs/guides/states.md | 6 ++++++ docs/llms-full.txt | 6 ++++++ interlock/_state_machine.py | 12 +++++++++++- tests/test_state_machine.py | 28 ++++++++++++++++++++++++++++ 4 files changed, 51 insertions(+), 1 deletion(-) diff --git a/docs/guides/states.md b/docs/guides/states.md index 84e143c..97e926b 100644 --- a/docs/guides/states.md +++ b/docs/guides/states.md @@ -84,6 +84,12 @@ breaker = CircuitBreaker( The growing interval is also a signal in its own right: a breaker waiting out a blip looks nothing like one that has failed ten rounds in a row. +Growth stops after 64 consecutive failed rounds. Any sane multiplier has long +since passed `wait_duration_in_open_max` by then, and an unbounded exponent +would eventually overflow to an infinite wait that never elapses — leaving the +breaker open for good, which is precisely the failure this release set out to +remove. + ## Proactive transition (`auto_transition`) By default the `OPEN → HALF_OPEN` move is **lazy**: it happens on the first call diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 7b3dd29..ac68c0f 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -1658,6 +1658,12 @@ breaker = CircuitBreaker( The growing interval is also a signal in its own right: a breaker waiting out a blip looks nothing like one that has failed ten rounds in a row. +Growth stops after 64 consecutive failed rounds. Any sane multiplier has long +since passed `wait_duration_in_open_max` by then, and an unbounded exponent +would eventually overflow to an infinite wait that never elapses — leaving the +breaker open for good, which is precisely the failure this release set out to +remove. + ## Proactive transition (`auto_transition`) By default the `OPEN → HALF_OPEN` move is **lazy**: it happens on the first call diff --git a/interlock/_state_machine.py b/interlock/_state_machine.py index 0db426e..4bb1ef0 100644 --- a/interlock/_state_machine.py +++ b/interlock/_state_machine.py @@ -22,6 +22,8 @@ (admit all, record metrics, never trip). """ +from typing import Final + from interlock._initial_state import validate_initial_state from interlock._windows import build_window from interlock.config import Config @@ -34,6 +36,14 @@ _PERMIT_ALL = frozenset({State.CLOSED, State.DISABLED, State.METRICS_ONLY}) +# The exponent stops here. Sixty-four rounds of any sane multiplier already dwarf +# every ceiling a caller would set, while an unbounded count walks a long-broken +# dependency into float's ceiling: the wait becomes ``inf``, never elapses, and +# the breaker can never leave OPEN again. With ``wait_duration_in_open_max`` set +# the wait itself stops growing but the count does not, so days of failure are +# enough to get there. +_MAX_BACKOFF_ROUNDS: Final[int] = 64 + class StateMachine: """Owns breaker state and drives transitions from recorded outcomes. @@ -302,7 +312,7 @@ def _open(self) -> None: # back to OPEN from a probe round feeds the backoff, including a round # that ended with nothing learned. if self._state is State.HALF_OPEN: - self._failed_rounds += 1 + self._failed_rounds = min(self._failed_rounds + 1, _MAX_BACKOFF_ROUNDS) self._state = State.OPEN self._opened_at = self._clock.monotonic() self._generation += 1 diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index 65c0b3d..a4c3778 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -870,3 +870,31 @@ def test__half_open__inconclusive_probe__frees_exactly_one_round_slot( assert machine.acquire() is True # one admission returned to the round assert machine.acquire() is False # and only the one + + +def test__open__endless_failed_rounds__wait_stays_finite( + config: Config, fake_clock: FakeClock +) -> None: + """The growth has to stop somewhere: a float that reached infinity never elapses. + + With a ceiling in play the wait itself stops growing, but the round counter + does not — so a dependency broken for days walks the exponent up until the + multiplication overflows, ``retry_after()`` returns infinity, and the breaker + can never leave ``OPEN`` again. + """ + machine = StateMachine( + config=replace( + config, wait_duration_backoff_multiplier=2.0, wait_duration_in_open_max=20.0 + ), + clock=fake_clock, + ) + _trip_to_open(machine, 2) + + for _ in range(1100): + wait = machine.retry_after() + assert wait is not None + _fail_probe_round(machine, fake_clock, wait=wait) + + remaining = machine.retry_after() + assert remaining is not None + assert remaining == pytest.approx(20.0) # pinned at the ceiling, not infinite From f077b7e0eebc98e185f2ac439eaca55b05038298 Mon Sep 17 00:00:00 2001 From: bagowix Date: Tue, 1 Sep 2026 13:51:42 +0400 Subject: [PATCH 3/4] =?UTF-8?q?fix:=20address=20review=20=E2=80=94=20probe?= =?UTF-8?q?=20classification,=20validation=20and=20documented=20limits?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A *leased* probe is what must keep its plain verdict, not any probe taken while a coordinator exists. When storage degrades, admission falls back to the local machine with no lease to strand, and the old check still suppressed the inconclusive path there. `admission.probe` marks exactly the leased ones. - Reject a multiplier whose backoff stops being a finite number: an infinite wait never elapses, so the breaker could never leave OPEN. Checked at construction, where the offending value is in front of the caller. - Reject a non-tuple container for `unreachable_exceptions`: `isinstance` raises its own TypeError on a list, which would surface on the first real failure instead of here. - Document the TypeError contract on both public constructors. - Say plainly that the backoff is local: under shared storage the coordinated lane reopens on the base wait and keeps no failed-round count. --- docs/guides/states.md | 5 +++++ docs/llms-full.txt | 5 +++++ interlock/_engine.py | 24 +++++++++++++++++------- interlock/_state_machine.py | 14 ++------------ interlock/breaker.py | 2 ++ interlock/config.py | 35 +++++++++++++++++++++++++++++++++++ interlock/registry.py | 2 ++ tests/test_config.py | 13 +++++++++++++ tests/test_engine.py | 13 +++++++++++++ 9 files changed, 94 insertions(+), 19 deletions(-) diff --git a/docs/guides/states.md b/docs/guides/states.md index 97e926b..2a2ff85 100644 --- a/docs/guides/states.md +++ b/docs/guides/states.md @@ -84,6 +84,11 @@ breaker = CircuitBreaker( The growing interval is also a signal in its own right: a breaker waiting out a blip looks nothing like one that has failed ten rounds in a row. +The backoff is local. Under a shared [storage](../integrations/redis.md) the +coordinated lane reopens on `wait_duration_in_open` and keeps no failed-round +count, so a coordinated breaker retries on the base wait no matter how many +rounds have failed. + Growth stops after 64 consecutive failed rounds. Any sane multiplier has long since passed `wait_duration_in_open_max` by then, and an unbounded exponent would eventually overflow to an infinite wait that never elapses — leaving the diff --git a/docs/llms-full.txt b/docs/llms-full.txt index ac68c0f..def849e 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -1658,6 +1658,11 @@ breaker = CircuitBreaker( The growing interval is also a signal in its own right: a breaker waiting out a blip looks nothing like one that has failed ten rounds in a row. +The backoff is local. Under a shared [storage](../integrations/redis.md) the +coordinated lane reopens on `wait_duration_in_open` and keeps no failed-round +count, so a coordinated breaker retries on the base wait no matter how many +rounds have failed. + Growth stops after 64 consecutive failed rounds. Any sane multiplier has long since passed `wait_duration_in_open_max` by then, and an unbounded exponent would eventually overflow to an infinite wait that never elapses — leaving the diff --git a/interlock/_engine.py b/interlock/_engine.py index d48a2b1..e519ee9 100644 --- a/interlock/_engine.py +++ b/interlock/_engine.py @@ -64,9 +64,17 @@ def validate_unreachable_exceptions( first real failure — masking the protected exception and stranding the probe slot it should have returned. + The container is checked too: ``isinstance`` rejects a list of types with a + ``TypeError`` of its own, so a caller who passes one would hit it on the + first real failure rather than here. + Raises: - TypeError: If an entry is not an ``Exception`` subclass. + TypeError: If the container is not a tuple, or an entry is not an + ``Exception`` subclass. """ + if not isinstance(cast('object', types), tuple): + raise TypeError(f'unreachable_exceptions must be a tuple, got: {type(types).__name__}') + for entry in cast('tuple[object, ...]', types): if not (isinstance(entry, type) and issubclass(entry, Exception)): raise TypeError( @@ -488,18 +496,19 @@ def _settle( failure = self._classifier.is_failure(result=result, exception=exception) slow = duration >= self._config.slow_call_duration_threshold outcome = _OUTCOME_BY_FLAGS[failure, slow] - coordinator = self._sync_coordinator or self._async_coordinator # Only HALF_OPEN cares; the machine owns that check because only it knows # its own state. Outside a probe the same exception stays an ordinary # failure — shedding load is the point. The ``is not None`` guard keeps # the common path — a call that returned — off ``isinstance``. # - # A coordinated probe holds a lease that only an outcome returns, and - # ``Storage`` has no way to hand one back unspent, so dropping the verdict - # there would strand the shared budget until its TTL. Until the protocol - # grows that operation, a shared probe keeps the old behaviour. + # A *leased* probe is the exception: it holds a remote grant that only an + # outcome returns, and ``Storage`` has no way to hand one back unspent, so + # dropping the verdict would strand the shared budget until its TTL. + # ``admission.probe`` marks exactly those, which also keeps the local + # fallback covered — when storage degrades the admission comes from the + # local machine, with no lease to strand. unreachable = ( - coordinator is None + not admission.probe and exception is not None and isinstance(exception, self._unreachable_exceptions) ) @@ -520,6 +529,7 @@ def _settle( notify(self._listener, 'on_call', name=self._name, outcome=outcome, duration=duration) self._emit_transitions(before, after, effective_before, effective_after) + coordinator = self._sync_coordinator or self._async_coordinator if coordinator is not None: if admission.probe: coordinator.notify_probe_outcome(outcome) diff --git a/interlock/_state_machine.py b/interlock/_state_machine.py index 4bb1ef0..39135c1 100644 --- a/interlock/_state_machine.py +++ b/interlock/_state_machine.py @@ -22,11 +22,9 @@ (admit all, record metrics, never trip). """ -from typing import Final - from interlock._initial_state import validate_initial_state from interlock._windows import build_window -from interlock.config import Config +from interlock.config import MAX_BACKOFF_ROUNDS, Config from interlock.outcome import Outcome from interlock.protocols import Clock from interlock.state import State @@ -36,14 +34,6 @@ _PERMIT_ALL = frozenset({State.CLOSED, State.DISABLED, State.METRICS_ONLY}) -# The exponent stops here. Sixty-four rounds of any sane multiplier already dwarf -# every ceiling a caller would set, while an unbounded count walks a long-broken -# dependency into float's ceiling: the wait becomes ``inf``, never elapses, and -# the breaker can never leave OPEN again. With ``wait_duration_in_open_max`` set -# the wait itself stops growing but the count does not, so days of failure are -# enough to get there. -_MAX_BACKOFF_ROUNDS: Final[int] = 64 - class StateMachine: """Owns breaker state and drives transitions from recorded outcomes. @@ -312,7 +302,7 @@ def _open(self) -> None: # back to OPEN from a probe round feeds the backoff, including a round # that ended with nothing learned. if self._state is State.HALF_OPEN: - self._failed_rounds = min(self._failed_rounds + 1, _MAX_BACKOFF_ROUNDS) + self._failed_rounds = min(self._failed_rounds + 1, MAX_BACKOFF_ROUNDS) self._state = State.OPEN self._opened_at = self._clock.monotonic() self._generation += 1 diff --git a/interlock/breaker.py b/interlock/breaker.py index 7aed8a7..ac0c6c6 100644 --- a/interlock/breaker.py +++ b/interlock/breaker.py @@ -70,6 +70,8 @@ class CircuitBreaker: outcome. Raises: + TypeError: If ``unreachable_exceptions`` is not a tuple of ``Exception`` + subclasses. ValueError: If ``initial_state`` is not a supported stable state. """ diff --git a/interlock/config.py b/interlock/config.py index 5f8d988..0f09a75 100644 --- a/interlock/config.py +++ b/interlock/config.py @@ -1,11 +1,20 @@ """Immutable circuit breaker configuration with eager validation.""" +import math from dataclasses import dataclass +from typing import Final from interlock.window import WindowType __all__ = ('Config',) +# The exponent stops here. Sixty-four rounds of any sane multiplier already dwarf +# every ceiling a caller would set, while an unbounded count walks a long-broken +# dependency into float's range: the wait becomes infinite, never elapses, and the +# breaker can never leave OPEN again. With wait_duration_in_open_max set the wait +# itself stops growing but the count does not, so days of failure get there. +MAX_BACKOFF_ROUNDS: Final[int] = 64 + @dataclass(frozen=True, kw_only=True, slots=True) class Config: @@ -26,6 +35,11 @@ class Config: constant wait then retries forever at full rate, and the growing interval is itself the signal that the dependency is not merely slow to recover. + The backoff is a local decision: with a shared ``Storage`` the coordinated + lane asks the backend to reopen after ``wait_duration_in_open``, and the + failed-round count lives in no shared state, so a coordinated breaker retries + on the base wait however many rounds have failed. + ``auto_transition`` opts into a timer that proactively moves a breaker from ``OPEN`` to ``HALF_OPEN`` once ``wait_duration_in_open`` elapses, emitting the state change without waiting for the next call. It defaults to ``False``, @@ -75,6 +89,13 @@ def __post_init__(self) -> None: # noqa: C901 - a flat list of guards, not bran f'wait_duration_backoff_multiplier must be >= 1, ' f'got {self.wait_duration_backoff_multiplier!r}' ) + if not math.isfinite(self._peak_wait_duration()): + raise ValueError( + f'wait_duration_backoff_multiplier is too large for ' + f'wait_duration_in_open={self.wait_duration_in_open!r}: after ' + f'{MAX_BACKOFF_ROUNDS} rounds the wait stops being a finite number, ' + f'got {self.wait_duration_backoff_multiplier!r}' + ) if ( self.wait_duration_in_open_max is not None and self.wait_duration_in_open_max < self.wait_duration_in_open @@ -95,3 +116,17 @@ def __post_init__(self) -> None: # noqa: C901 - a flat list of guards, not bran ) if self.window_size < 1: raise ValueError(f'window_size must be >= 1, got {self.window_size!r}') + + def _peak_wait_duration(self) -> float: + """The longest wait the backoff can produce, or infinity if it overruns. + + A wait that is not a finite number never elapses, so the breaker it + governs could never leave ``OPEN``. Catching that here keeps the failure + at construction, where the offending value is in front of the caller. + """ + try: + return self.wait_duration_in_open * ( + self.wait_duration_backoff_multiplier**MAX_BACKOFF_ROUNDS + ) + except OverflowError: + return math.inf diff --git a/interlock/registry.py b/interlock/registry.py index cd9e9e8..fcef294 100644 --- a/interlock/registry.py +++ b/interlock/registry.py @@ -51,6 +51,8 @@ class Registry: outcome. Raises: + TypeError: If ``unreachable_exceptions`` is not a tuple of ``Exception`` + subclasses. ValueError: If ``initial_state`` is not a supported stable state. """ diff --git a/tests/test_config.py b/tests/test_config.py index 6c9a2aa..79618e7 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -80,3 +80,16 @@ def test__config__backoff_defaults__preserve_constant_wait() -> None: assert config.wait_duration_backoff_multiplier == 1.0 assert config.wait_duration_in_open_max is None + + +def test__config__multiplier_overflowing_the_bounded_backoff__raises() -> None: + """A wait that stops being a finite number never elapses, so refuse it here.""" + with pytest.raises(ValueError, match='too large'): + Config(wait_duration_in_open=60.0, wait_duration_backoff_multiplier=1e308) + + +def test__config__multiplier_at_the_edge__accepted() -> None: + """The bound is on the peak wait, not on the multiplier in isolation.""" + config = Config(wait_duration_in_open=60.0, wait_duration_backoff_multiplier=2.0) + + assert config.wait_duration_backoff_multiplier == 2.0 diff --git a/tests/test_engine.py b/tests/test_engine.py index cde5aaf..d176702 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -533,3 +533,16 @@ def test__unreachable_exceptions__not_an_exception_type__rejected_at_constructio clock=fake_clock, unreachable_exceptions=cast('tuple[type[Exception], ...]', (bad,)), ) + + +def test__unreachable_exceptions__not_a_tuple__rejected_at_construction( + config: Config, fake_clock: FakeClock +) -> None: + """``isinstance`` refuses a list of types, so refuse it here with a clear message.""" + with pytest.raises(TypeError, match='must be a tuple'): + Engine( + name='test', + config=config, + clock=fake_clock, + unreachable_exceptions=cast('tuple[type[Exception], ...]', [_NoLocalSlot]), + ) From 28839c198bf72313887d9b5f7440246d9bc1db49 Mon Sep 17 00:00:00 2001 From: bagowix Date: Tue, 1 Sep 2026 14:00:16 +0400 Subject: [PATCH 4/4] perf: keep the backoff validation off the default construction path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The peak-wait check exponentiates, and `Config()` is on a measured path. At the default multiplier of 1.0 there is no backoff to validate — the peak is the base wait — so both multiplier guards now sit behind a single comparison and the arithmetic only runs when a caller asked for a backoff. --- interlock/config.py | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/interlock/config.py b/interlock/config.py index 0f09a75..1440f55 100644 --- a/interlock/config.py +++ b/interlock/config.py @@ -84,18 +84,22 @@ def __post_init__(self) -> None: # noqa: C901 - a flat list of guards, not bran raise ValueError( f'wait_duration_in_open must be > 0, got {self.wait_duration_in_open!r}' ) - if self.wait_duration_backoff_multiplier < 1.0: - raise ValueError( - f'wait_duration_backoff_multiplier must be >= 1, ' - f'got {self.wait_duration_backoff_multiplier!r}' - ) - if not math.isfinite(self._peak_wait_duration()): - raise ValueError( - f'wait_duration_backoff_multiplier is too large for ' - f'wait_duration_in_open={self.wait_duration_in_open!r}: after ' - f'{MAX_BACKOFF_ROUNDS} rounds the wait stops being a finite number, ' - f'got {self.wait_duration_backoff_multiplier!r}' - ) + # Both multiplier guards sit behind one comparison: at the default of 1.0 + # there is no backoff to check, and the exponentiation would be a no-op + # computed on every construction. + if self.wait_duration_backoff_multiplier != 1.0: + if self.wait_duration_backoff_multiplier < 1.0: + raise ValueError( + f'wait_duration_backoff_multiplier must be >= 1, ' + f'got {self.wait_duration_backoff_multiplier!r}' + ) + if not math.isfinite(self._peak_wait_duration()): + raise ValueError( + f'wait_duration_backoff_multiplier is too large for ' + f'wait_duration_in_open={self.wait_duration_in_open!r}: after ' + f'{MAX_BACKOFF_ROUNDS} rounds the wait stops being a finite number, ' + f'got {self.wait_duration_backoff_multiplier!r}' + ) if ( self.wait_duration_in_open_max is not None and self.wait_duration_in_open_max < self.wait_duration_in_open