# interlock — full documentation
> A modern circuit breaker for Python: sync and async in a single class,
> sliding-window failure-rate and slow-call detection, a type-safe decorator
> API, and transparent per-host integrations for httpx2, aiohttp and requests.
> Zero-dependency core (standard library only); integrations ship as optional
> extras.
This file inlines every documentation page in reading order. It is generated
from the Markdown sources by ``scripts/build_llms_full.py`` — edit the pages in
``docs/``, not this file.
---
# Getting started
## Install
=== "uv"
```bash
uv add interlock-cb
```
=== "pip"
```bash
pip install interlock-cb
```
=== "poetry"
```bash
poetry add interlock-cb
```
The core is pure standard library. External integrations are optional extras —
add the ones you need (same names with `pip install` / `poetry add`):
```bash
uv add 'interlock-cb[fastapi]' # CircuitOpenError -> 503 + Retry-After
uv add 'interlock-cb[litestar]' # same for Litestar
uv add 'interlock-cb[httpx2]' # per-host httpx2 transport
uv add 'interlock-cb[httpx]' # per-host httpx transport
uv add 'interlock-cb[aiohttp]' # per-host aiohttp client middleware
uv add 'interlock-cb[requests]' # per-host requests session adapter
uv add 'interlock-cb[tenacity]' # retry x breaker composition helpers
uv add 'interlock-cb[redis]' # shared distributed state
uv add 'interlock-cb[otel]' # OpenTelemetry metrics listener
```
## Create a breaker
A breaker is named and configured once, then reused:
```python
from interlock import CircuitBreaker, Config
breaker = CircuitBreaker(
name='payments',
config=Config(failure_rate_threshold=0.5, minimum_number_of_calls=20),
)
```
The defaults follow resilience4j: trip at a 50% failure rate over at least 10
calls, stay open for 60s, then admit up to 10 probe calls (one at a time) and
decide from their outcomes. See [Configuration](guides/configuration.md) for
every option.
For an existing production dependency, start with
`initial_state=State.METRICS_ONLY`: calls are always admitted while their
outcomes populate the window. Tune thresholds from real traffic, then deploy a
new breaker in the default `CLOSED` state. See
[Safe production rollout](guides/states.md#safe-rollout).
## Three ways to protect work
All three run over the same `call()` primitive.
### Decorator
```python
@breaker
def charge(amount: int) -> str:
return gateway.charge(amount)
```
The decorator preserves the wrapped signature and its sync/async nature — type
checkers still see `charge` as `(int) -> str`.
### `breaker.call`
```python
result = breaker.call(gateway.charge, 100)
```
`call` inspects the callable to pick the sync or the async path. Where that is
already known — a synchronous client, an awaited handler — `call_sync` and
`call_async` skip the inspection:
```python
result = breaker.call_sync(gateway.charge, 100)
result = await breaker.call_async(client.get, url)
```
`call_sync` never awaits, so passing a coroutine function to it records the
coroutine's creation rather than its outcome. Reach for these on a hot path that
calls the same shape every time; `call` is the right default everywhere else.
### Context manager
```python
with breaker:
gateway.charge(100)
```
!!! note "Contract difference"
The decorator and `call` see a callable, so result-based classification and
slow-call detection both apply. The context manager sees only the block —
its exception and duration — so classification by **return value** is not
available there. Need result-based classification? Use the decorator or
`call`.
## Async
The same instance handles async. The decorator and `call` detect a coroutine
function; the instance is also an async context manager:
```python
@breaker
async def fetch(url: str) -> bytes:
return await client.get(url)
result = await breaker.call(client.get, url)
async with breaker:
await client.get(url)
```
## Handle rejections
When the circuit is not closed, the call is rejected with `CircuitOpenError`:
```python
from interlock import CircuitOpenError
try:
breaker.call(gateway.charge, 100)
except CircuitOpenError as exc:
# exc.breaker_name, exc.retry_after (seconds, may be None), exc.last_failure
raise
```
## Inspect state
```python
breaker.state # State.CLOSED / OPEN / HALF_OPEN / ...
breaker.snapshot() # WindowSnapshot: total_calls, failed_calls, slow_calls,
# .failure_rate, .slow_call_rate
```
## Next steps
- [Runnable demo](demo.md) — watch a breaker trip and recover, then debug it
- [Configuration](guides/configuration.md)
- [States & manual control](guides/states.md)
- [Failure classification](guides/failure-classification.md)
- [Observability](guides/observability.md)
- [Integrations](integrations/index.md) — FastAPI, Litestar, httpx2, httpx, aiohttp, requests, tenacity, Redis
- [Resilience pipeline](guides/pipeline.md) — compose timeout, bulkhead, retry and fallback around the breaker
---
# Runnable demo
Three self-contained scripts in [`examples/`](https://github.com/bagowix/interlock/tree/main/examples)
show a breaker doing its job — standard library plus `interlock-cb` only, no
network, no services to stand up. The output is deterministic: what you see
below is exactly what you get, so set a breakpoint anywhere and step through.
```bash
pip install interlock-cb # or: uv add interlock-cb
python examples/lifecycle.py # one breaker through its full state cycle
python examples/two_clients.py # two clients, one outage, no collateral damage
python examples/pipeline.py # timeout + breaker + fallback around a hanging service
```
Both scripts print through an `EventListener` — every line tagged
`[listener]` comes from the breaker itself, not from the demo code. See
[Observability](guides/observability.md) for the hook reference.
## `lifecycle.py` — one breaker, full cycle
A fake payment gateway is healthy, goes down, and recovers. The breaker is
configured tightly so the whole story fits in seven calls: trip at 50%
failures over a 4-call window, stay open 1 second, close after 2 good probes.
??? example "lifecycle.py — full source"
```python
"""Walk one breaker through its full lifecycle: CLOSED -> OPEN -> HALF_OPEN -> CLOSED.
Zero dependencies, no network — run it directly and watch every transition:
python examples/lifecycle.py
The gateway is deterministic (healthy, then down, then recovered), so the
output is always the same. Tweak the ``Config`` values and re-run to see how
the thresholds change the story. Explained line by line in
https://bagowix.github.io/interlock/demo/.
"""
import time
from interlock import CircuitBreaker, CircuitOpenError, Config, Outcome, State
class GatewayError(Exception):
"""The fake dependency's failure mode."""
class FlakyGateway:
"""A payment gateway you can switch between healthy and down."""
def __init__(self) -> None:
self.healthy = True
def charge(self, amount: int) -> str:
if not self.healthy:
raise GatewayError('503 from gateway')
return f'charged ${amount}'
class PrintListener:
"""An EventListener that narrates everything the breaker does."""
def on_state_change(self, *, name: str, old: State, new: State) -> None:
print(f' [listener] {name}: state {old.name} -> {new.name}')
def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
print(f' [listener] {name}: recorded {outcome.name} ({duration:.3f}s)')
def on_rejected(self, *, name: str) -> None:
print(f' [listener] {name}: call rejected — circuit is open')
def on_reset(self, *, name: str) -> None:
print(f' [listener] {name}: manual reset')
gateway = FlakyGateway()
breaker = CircuitBreaker(
name='payments-api',
config=Config(
failure_rate_threshold=0.5, # trip at >= 50% failures ...
minimum_number_of_calls=4, # ... once the window holds 4 calls
window_size=4,
wait_duration_in_open=1.0, # stay OPEN for 1s, then allow probes
permitted_calls_in_half_open=2, # 2 good probes close the circuit
),
listener=PrintListener(),
)
@breaker
def charge(amount: int) -> str:
"""The protected call: the decorator preserves this signature."""
return gateway.charge(amount)
def attempt(number: int, amount: int) -> None:
"""Make one charge attempt and print what came back."""
try:
print(f'call {number}: {charge(amount)}')
except GatewayError as exc:
print(f'call {number}: dependency failed — {exc}')
except CircuitOpenError as exc:
print(
f'call {number}: REJECTED in ~0ms, the gateway was never called '
f'(retry_after={exc.retry_after:.1f}s)'
)
def main() -> None:
"""Run the four phases of the lifecycle."""
print('phase 1 — healthy dependency, breaker CLOSED')
attempt(1, 10)
attempt(2, 20)
print()
print('phase 2 — the gateway goes down; failures fill the window')
gateway.healthy = False
attempt(3, 30)
attempt(4, 40) # 2 failures / 4 calls = 50% -> the breaker trips
print()
print('phase 3 — circuit OPEN: calls fail fast, the gateway gets a break')
attempt(5, 50)
print()
print('phase 4 — after wait_duration_in_open the breaker probes')
gateway.healthy = True # ops fixed the gateway meanwhile
time.sleep(1.1)
attempt(6, 60) # probe 1 (OPEN -> HALF_OPEN on admission)
attempt(7, 70) # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED
print()
print(f'final state: {breaker.state.name}, window reset: {breaker.snapshot()}')
if __name__ == '__main__':
main()
```
Running it prints:
```text
phase 1 — healthy dependency, breaker CLOSED
[listener] payments-api: recorded SUCCESS (0.000s)
call 1: charged $10
[listener] payments-api: recorded SUCCESS (0.000s)
call 2: charged $20
```
Every completed call is recorded into the sliding window — the listener's
`on_call` hook fires with the classified `Outcome` and the duration.
```text
phase 2 — the gateway goes down; failures fill the window
[listener] payments-api: recorded FAILURE (0.000s)
call 3: dependency failed — 503 from gateway
[listener] payments-api: recorded FAILURE (0.000s)
[listener] payments-api: state CLOSED -> OPEN
call 4: dependency failed — 503 from gateway
```
Call 4 is the moment the window reaches `minimum_number_of_calls=4` with 2
failures out of 4 — exactly the 50% threshold — so recording it trips the
circuit: `on_state_change` fires *inside* call 4's bookkeeping, before the
demo's own `call 4:` line prints.
```text
phase 3 — circuit OPEN: calls fail fast, the gateway gets a break
[listener] payments-api: call rejected — circuit is open
call 5: REJECTED in ~0ms, the gateway was never called (retry_after=1.0s)
```
Call 5 never reaches the gateway. `CircuitOpenError` is raised immediately
and carries `retry_after` — the estimate until the next probe is allowed.
This is the entire point of a breaker: while the dependency is down, callers
spend no timeouts on it and it gets quiet time to recover.
```text
phase 4 — after wait_duration_in_open the breaker probes
[listener] payments-api: state OPEN -> HALF_OPEN
[listener] payments-api: recorded SUCCESS (0.000s)
call 6: charged $60
[listener] payments-api: recorded SUCCESS (0.000s)
[listener] payments-api: state HALF_OPEN -> CLOSED
call 7: charged $70
final state: CLOSED, window reset: WindowSnapshot(total_calls=0, failed_calls=0, slow_calls=0)
```
The transition to `HALF_OPEN` is lazy: it happens when call 6 asks for
admission after the 1-second wait, not on a timer (set
`Config.auto_transition=True` for the eager variant). Calls 6 and 7 are the
two permitted probes; both succeed, so recording the second one closes the
circuit and starts a fresh window.
**Things to try:** set `gateway.healthy = False` before phase 4 and watch the
failed probe reopen the circuit; raise `permitted_calls_in_half_open`; put a
breakpoint in `PrintListener.on_state_change` and inspect
`breaker.snapshot()` at each transition.
## `two_clients.py` — isolation under a partial outage
One asyncio app talks to two dependencies. A [`Registry`](reference.md) hands
each its own breaker, so when `recommendations` goes down during rounds 3–6,
only its circuit opens — the app serves a cached fallback — while `payments`
keeps charging as if nothing happened.
??? example "two_clients.py — full source"
```python
"""Two guarded clients in one event loop: one dependency dies, the other keeps serving.
Zero dependencies, no network — run it directly:
python examples/two_clients.py
A ``Registry`` hands out one independent breaker per dependency. The
``recommendations`` service goes down during rounds 3-6: its breaker opens
and the app falls back to a cached list, while ``payments`` — its own
breaker untouched — keeps charging without a hiccup. Explained line by line
in https://bagowix.github.io/interlock/demo/.
"""
import asyncio
from interlock import CircuitOpenError, Config, Outcome, Registry, State
class RecsDownError(Exception):
"""The recommendations service's failure mode."""
class PrintListener:
"""An EventListener that narrates state changes and rejections."""
def on_state_change(self, *, name: str, old: State, new: State) -> None:
print(f' [listener] {name}: state {old.name} -> {new.name}')
def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
pass # per-call noise is off; see examples/lifecycle.py for it
def on_rejected(self, *, name: str) -> None:
print(f' [listener] {name}: call rejected — circuit is open')
def on_reset(self, *, name: str) -> None:
print(f' [listener] {name}: manual reset')
registry = Registry(
config=Config(
failure_rate_threshold=0.5,
minimum_number_of_calls=4,
window_size=4,
wait_duration_in_open=1.0,
permitted_calls_in_half_open=2,
),
listener=PrintListener(),
)
outage = False
@registry.get('recommendations')
async def fetch_recommendations(user: str) -> list[str]:
"""Call the flaky recommendations service."""
if outage:
raise RecsDownError('recommendations service timed out')
return [f'{user}-pick-1', f'{user}-pick-2']
@registry.get('payments')
async def charge(user: str, amount: int) -> str:
"""Call the payments service, which stays healthy throughout."""
return f'charged {user} ${amount}'
async def round_trip(number: int, user: str) -> None:
"""One request round: hit both dependencies concurrently."""
print(f'round {number}:')
payment, recs = await asyncio.gather(
charge(user, 25),
fetch_recommendations(user),
return_exceptions=True,
)
print(f' payments -> {payment}')
if isinstance(recs, CircuitOpenError):
print(' recommendations -> rejected instantly -> fallback: cached picks')
elif isinstance(recs, RecsDownError):
print(f' recommendations -> failed ({recs}) -> fallback: cached picks')
else:
print(f' recommendations -> {recs}')
async def main() -> None:
"""Run eight rounds across the outage and the recovery."""
global outage # noqa: PLW0603 - a module-level switch keeps the demo flat
print('rounds 1-2 — both dependencies healthy')
await round_trip(1, 'alice')
await round_trip(2, 'bob')
print()
print('rounds 3-6 — recommendations goes down; payments must not care')
outage = True
await round_trip(3, 'carol')
await round_trip(4, 'dave') # 50% failures -> recommendations breaker opens
await round_trip(5, 'erin') # rejected in ~0ms: no timeout is burned
await round_trip(6, 'frank')
print()
print('rounds 7-8 — the outage is over; the breaker probes and closes')
outage = False
await asyncio.sleep(1.1) # let wait_duration_in_open elapse
await round_trip(7, 'grace')
await round_trip(8, 'heidi')
print()
for name in ('payments', 'recommendations'):
print(f'final state of {name}: {registry.get(name).state.name}')
if __name__ == '__main__':
asyncio.run(main())
```
The interesting part of the output:
```text
round 4:
[listener] recommendations: state CLOSED -> OPEN
payments -> charged dave $25
recommendations -> failed (recommendations service timed out) -> fallback: cached picks
round 5:
[listener] recommendations: call rejected — circuit is open
payments -> charged erin $25
recommendations -> rejected instantly -> fallback: cached picks
```
Round 4's failure is the second in a 4-call window — the `recommendations`
breaker trips. From round 5 on the difference matters: the round-4 request
*waited* for the dependency to fail, the round-5 request is rejected in
microseconds, so the user still gets their page (with cached picks) at full
speed. Notice what is absent: the `payments` breaker never logs a single
state change for the whole run.
```text
rounds 7-8 — the outage is over; the breaker probes and closes
round 7:
[listener] recommendations: state OPEN -> HALF_OPEN
payments -> charged grace $25
recommendations -> ['grace-pick-1', 'grace-pick-2']
round 8:
[listener] recommendations: state HALF_OPEN -> CLOSED
payments -> charged heidi $25
recommendations -> ['heidi-pick-1', 'heidi-pick-2']
```
Recovery is gradual by design: two successful probes (rounds 7 and 8) must
complete before the circuit closes again.
**Things to try:** extend the outage past round 7 and watch a failed probe
send the circuit straight back to `OPEN`; give the two breakers different
configs via `registry.get(name, config=...)`; replace the hand-rolled
fallback with a [tenacity retry](integrations/tenacity.md) that waits
exactly `retry_after`.
## `pipeline.py` — composition against a quiet death
The nastiest failure mode gets the [v2 pipeline](guides/pipeline.md)
treatment: the inventory service never raises — it *hangs*. On its own that
trips nothing and starves every caller. Three composed strategies turn it
into a non-event: a timeout makes hangs classifiable, the breaker counts
them, a fallback keeps serving.
??? example "pipeline.py — full source"
```python
"""Compose timeout + breaker + fallback around one dependency that dies quietly.
Zero dependencies, no network — run it directly:
python examples/pipeline.py
The inventory service never raises: it *hangs*. Alone, that starves callers
and trips nothing. The pipeline turns the hang into a failure chain: the
timeout cancels the attempt, the breaker counts it and opens, the fallback
serves a cached snapshot — and while the circuit is open, requests cost
~0.0s instead of a timeout each. Add `.retry(...)` between the fallback and
the breaker for the full stack (needs `interlock-cb[tenacity]`). Explained
line by line in https://bagowix.github.io/interlock/demo/.
"""
import asyncio
import time
from interlock import (
CallTimeoutError,
CircuitBreaker,
CircuitOpenError,
Config,
Outcome,
Pipeline,
State,
)
CACHED_SNAPSHOT = ['widget (cached)', 'gadget (cached)']
class PrintListener:
"""Narrates the breaker's transitions and the pipeline's decisions."""
def on_state_change(self, *, name: str, old: State, new: State) -> None:
print(f' [listener] {name}: state {old.name} -> {new.name}')
def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
pass # per-call noise is off; see examples/lifecycle.py for it
def on_rejected(self, *, name: str) -> None:
print(f' [listener] {name}: call rejected — circuit is open')
def on_reset(self, *, name: str) -> None:
print(f' [listener] {name}: manual reset')
def on_fallback(self, *, name: str, error: BaseException) -> None:
print(f' [listener] {name}: fallback served instead of {type(error).__name__}')
events = PrintListener()
hanging = False
breaker = CircuitBreaker(
name='inventory',
config=Config(
failure_rate_threshold=0.5, # trip at >= 50% failures ...
minimum_number_of_calls=4, # ... once the window holds 4 calls
window_size=4,
wait_duration_in_open=1.0, # stay OPEN for 1s, then allow probes
permitted_calls_in_half_open=2, # 2 good probes close the circuit
),
listener=events,
)
pipeline = (
Pipeline.builder()
.fallback(
lambda _exc: CACHED_SNAPSHOT,
on=(CircuitOpenError, CallTimeoutError),
name='inventory',
listener=events,
)
.circuit_breaker(breaker)
.timeout(0.2) # a hanging attempt becomes CallTimeoutError after 0.2s
.build()
)
@pipeline
async def fetch_inventory() -> list[str]:
"""The protected call: the dependency hangs instead of erroring."""
if hanging:
await asyncio.sleep(5) # never finishes within the timeout
return ['widget', 'gadget']
async def request(number: int) -> None:
"""Serve one request and print what it cost."""
start = time.perf_counter()
items = await fetch_inventory()
elapsed = time.perf_counter() - start
print(f'request {number}: {items} in ~{elapsed:.1f}s')
async def main() -> None:
"""Run the outage story end to end."""
global hanging # noqa: PLW0603 - a module-level switch keeps the demo flat
print('phase 1 — healthy and fast, breaker CLOSED')
await request(1)
await request(2)
print()
print('phase 2 — the service starts hanging; timeouts become failures')
hanging = True
await request(3) # waits the full 0.2s timeout, then serves the cache
await request(4) # 2 timeouts / 4 calls = 50% -> the breaker trips
print()
print('phase 3 — circuit OPEN: no timeout is burned, the cache is instant')
await request(5)
print()
print('phase 4 — the service recovers; probes close the circuit')
hanging = False
await asyncio.sleep(1.1) # let wait_duration_in_open elapse
await request(6) # probe 1 (OPEN -> HALF_OPEN on admission)
await request(7) # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED
print()
print(f'final state: {breaker.state.name}')
if __name__ == '__main__':
asyncio.run(main())
```
The interesting part of the output:
```text
phase 2 — the service starts hanging; timeouts become failures
[listener] inventory: fallback served instead of CallTimeoutError
request 3: ['widget (cached)', 'gadget (cached)'] in ~0.2s
[listener] inventory: state CLOSED -> OPEN
[listener] inventory: fallback served instead of CallTimeoutError
request 4: ['widget (cached)', 'gadget (cached)'] in ~0.2s
phase 3 — circuit OPEN: no timeout is burned, the cache is instant
[listener] inventory: call rejected — circuit is open
[listener] inventory: fallback served instead of CircuitOpenError
request 5: ['widget (cached)', 'gadget (cached)'] in ~0.0s
```
Watch the latency column. Requests 3–4 each pay the full 0.2 s timeout —
that is the *detection* cost while the window fills. Request 4 tips the
failure rate to 50% and the circuit opens; from request 5 on the rejection
is immediate and the cached snapshot costs ~0.0 s. The user never saw an
error: every response during the outage came from the fallback, and the
listener logged each substitution.
```text
phase 4 — the service recovers; probes close the circuit
[listener] inventory: state OPEN -> HALF_OPEN
request 6: ['widget', 'gadget'] in ~0.0s
[listener] inventory: state HALF_OPEN -> CLOSED
request 7: ['widget', 'gadget'] in ~0.0s
```
**Things to try:** raise `wait_duration_in_open` and watch how long the
cache serves; put a `.retry(...)` step between the fallback and the breaker
(needs `interlock-cb[tenacity]`) and see attempts in the listener via
`on_retry`; drop the `.timeout(...)` step and watch the outage become
invisible again — no strategy ever fires.
## Where to go next
These demos hand-roll their guarded clients so you can debug the mechanics.
In real code you usually don't have to: the [integrations](integrations/index.md)
apply the same per-dependency pattern to httpx2, httpx, aiohttp, requests and
FastAPI transparently, and the [resilience pipeline](guides/pipeline.md)
composes the breaker with timeout, bulkhead, retry and fallback declaratively.
---
# Comparison
Picking a circuit breaker is mostly about which failure model, runtime and
coordination story you need. This page compares interlock-cb with the
established Python circuit breakers — honestly. interlock-cb is the youngest
of the five (first released in 2026); [pybreaker](https://github.com/danielfm/pybreaker),
[circuitbreaker](https://github.com/fabfuel/circuitbreaker),
[aiobreaker](https://github.com/arlyon/aiobreaker) and
[purgatory](https://github.com/mardiros/purgatory) have carried production
traffic for years, and for many projects one of them is exactly the right
choice.
## Feature table
| Feature | interlock-cb | pybreaker | circuitbreaker | aiobreaker | purgatory |
|---|:---:|:---:|:---:|:---:|:---:|
| Core states (closed / open / half-open) | ✅ | ✅ | ✅ | ✅ | ✅ |
| Choose which exceptions count as failures | ✅ | ✅ | ✅ | ✅ | ✅ |
| Zero-dependency core | ✅ | ✅ | ✅ | ✅ | ✅ |
| `async` / `await` (asyncio) | ✅ | Tornado | ✅ | ✅ | ✅ |
| Sync and async in **one** breaker class | ✅ | — | ✅ | ✅ | separate factories |
| Trip condition | failure **rate** over a window | consecutive count | consecutive count | consecutive count | consecutive count |
| Time-based sliding window | ✅ | — | — | — | — |
| Slow-call detection | ✅ | — | — | — | — |
| Result-based failure classification | ✅ | — | — | — | — |
| Event / state-change listeners | ✅ | ✅ | — | ✅ | ✅ |
| Shared state across processes (Redis) | ✅ | ✅ | — | ✅ | ✅ |
| Globally budgeted recovery probes | ✅ | — | — | — | — |
| Fallback function | ✅ | — | ✅ | — | — |
| Composable resilience pipeline (retry × breaker × bulkhead × timeout × fallback) | ✅ | — | — | — | — |
| Fully typed API (`py.typed`) | ✅ | — | — | — | ✅ |
| Signature-preserving decorator (`ParamSpec`) | ✅ | — | — | — | — |
| HTTP client integrations (httpx2 / httpx / aiohttp / requests) | ✅ | — | — | — | — |
| Retry composition helpers (tenacity) | ✅ | — | — | — | — |
| OpenTelemetry metrics | ✅ | — | — | — | — |
| Operator overrides (force-open / disable / shadow mode) | ✅ | — | — | — | — |
| Years of production use | new | ✅ | ✅ | ✅ | ✅ |
| Latest release (as of September 2026) | 2.8.0 · 2026 | 1.4.1 · 2025 | 2.1.3 | 1.2.0 · 2021 | 3.0.1 · 2024 |
| Python | ≥ 3.11 | ≥ 3.9 | ≥ 3.8 | ≥ 3.6 | ≥ 3.9 |
Compared against pybreaker 1.4.1, circuitbreaker 2.1.3, aiobreaker 1.2.0
and purgatory 3.0.1, as documented in July 2026. Something out of date or
unfair? Please
[open a PR](https://github.com/bagowix/interlock/pulls).
## The four established libraries, honestly
**pybreaker** is the original Python circuit breaker and the most proven one:
a small, stable sync breaker with listeners and optional Redis-backed state,
maintained for well over a decade. Its async support targets Tornado, not
asyncio, and it trips on a consecutive-failure count rather than a failure
rate. If you run a synchronous stack and want the most battle-tested option,
start here.
**circuitbreaker** has the smallest API of the five: one `@circuit` decorator
that also handles async functions, plus the only built-in **fallback
function** in this table. There are no listeners, no shared state and no rate
window — which is precisely its appeal when you need a guard on a handful of
call sites and nothing else.
**aiobreaker** is pybreaker ported to native asyncio, with the same
listener and Redis-storage features. Its last release was in 2021; for new
projects, prefer an actively maintained alternative.
**purgatory** brings a fully typed sync + async breaker with Redis storage
and event hooks. Sync and async live in separate factory classes
(`SyncCircuitBreakerFactory` / `AsyncCircuitBreakerFactory`), and tripping is
a consecutive-failure threshold with a TTL on the open state.
## Where interlock-cb differs
- **Rate over a window, not a streak.** A consecutive-failure counter resets
on any single success, so a dependency failing 90% of requests under load
can keep a breaker closed indefinitely. interlock trips on the failure
*rate* across a count- or time-based sliding window
([configuration](guides/configuration.md)).
- **Slow calls are failures too.** A dependency that answers in 30 s can be
worse than one that errors fast. `slow_call_duration_threshold` +
`slow_call_rate_threshold` trip the breaker on latency degradation alone.
- **One class, both runtimes.** The same `CircuitBreaker` instance guards sync
and async callables — decorator, context manager, or `call()` — and the
decorator preserves the wrapped signature for type checkers.
- **Coordination built for fleets.** With the Redis storage, tripping is
atomic across racing instances and half-open probes are budgeted globally,
with graceful degradation to local state when Redis is down
([Redis integration](integrations/redis.md)).
- **Meets your stack where it is.** Per-host breakers ship for httpx2, httpx,
aiohttp and requests; tenacity glue composes retries correctly; FastAPI and
Litestar map rejections to `503 + Retry-After`
([integrations overview](integrations/index.md)).
The honest trade-off: interlock-cb requires Python ≥ 3.11 and has not had
years in production. Reach for an established
library if that maturity matters more than the feature gap; choose
interlock-cb when you want rate-based windows, slow-call detection,
coordinated state and a fully typed API.
---
# Migrating from pybreaker / circuitbreaker
Already using [pybreaker](https://github.com/danielfm/pybreaker) or
[circuitbreaker](https://github.com/fabfuel/circuitbreaker)? Moving to
interlock-cb is mostly mechanical — swap the import, translate the constructor
arguments, and keep the rest of your call sites. This page maps every concept
across, one library at a time.
If you have not decided *whether* to move yet, read the
[Comparison](comparison.md) first — this page assumes you have.
## The one conceptual change
Every established Python breaker trips on a **consecutive-failure count**:
`fail_max` / `failure_threshold` failures *in a row* open the circuit, and a
single success resets the counter. interlock trips on a **failure rate over a
sliding window** instead ([configuration](guides/configuration.md)).
That difference is the whole reason to migrate — a dependency failing 40% of
requests under load never trips a streak counter, because successes keep
resetting it — but it means the threshold numbers do **not** carry over
one-to-one. Everything else (timing, decorator, `call`, listeners) has a direct
equivalent.
Translate the trip condition like this:
| Old (streak) | interlock (rate over a window) |
|---|---|
| `fail_max=5` / `failure_threshold=5` | `minimum_number_of_calls` = how many calls to observe before trusting a rate; `failure_rate_threshold` = the fraction that trips |
| "open on 5 failures in a row" | e.g. `Config(minimum_number_of_calls=5, failure_rate_threshold=0.8)` — trip when ≥ 80% of the last window is failing |
| `reset_timeout` / `recovery_timeout` | `wait_duration_in_open` (seconds) — direct equivalent |
There is no exact arithmetic conversion, because the two models answer
different questions. A safe way to pick numbers is to run in
[shadow mode](guides/states.md#safe-rollout) first (see
[Roll out incrementally](#roll-out-incrementally) below) and read the real
rates off `breaker.snapshot()` before enforcing.
---
## From pybreaker
### Constructor
```python
# before — pybreaker
import pybreaker
breaker = pybreaker.CircuitBreaker(
fail_max=5,
reset_timeout=60,
exclude=[ValueError],
name='payments',
)
```
```python
# after — interlock
from interlock import CircuitBreaker, Config
breaker = CircuitBreaker(
name='payments',
config=Config(
minimum_number_of_calls=5,
failure_rate_threshold=0.8,
wait_duration_in_open=60.0,
),
)
```
| pybreaker | interlock | Notes |
|---|---|---|
| `fail_max` | `minimum_number_of_calls` + `failure_rate_threshold` | Streak → rate; see [above](#the-one-conceptual-change). |
| `reset_timeout` | `Config.wait_duration_in_open` | Direct, in seconds. |
| `success_threshold` | `Config.permitted_calls_in_half_open` | Probes admitted before the breaker re-decides. interlock decides on the probe *rate*, not a fixed success count. |
| `exclude=[...]` | a `FailureClassifier` | See [below](#exclude-failureclassifier). |
| `name=` | `name=` | Same. |
| `state_storage=CircuitRedisStorage(...)` | `storage=RedisStorage(...)` | See [below](#redis-shared-state). |
| `listeners=[...]` | `listener=` | See [below](#listeners). |
### Decorator and `call`
Both keep working with the same shape — only the object changes:
```python
# before # after
@breaker @breaker
def charge(amount): ... def charge(amount): ...
breaker.call(charge, 100) breaker.call(charge, 100)
```
interlock's decorator additionally **preserves the wrapped signature** for type
checkers, and the same instance also works as a `with` block and on `async`
functions — no separate class ([getting started](getting-started.md#async)).
### `exclude` → `FailureClassifier`
pybreaker's `exclude` lists exceptions that should *not* count as failures.
interlock expresses the same policy as a
[classifier](guides/failure-classification.md):
```python
# before — pybreaker
breaker = pybreaker.CircuitBreaker(exclude=[ValueError])
```
```python
# after — interlock
class IgnoreValueError:
def is_failure(self, *, result: object, exception: Exception | None) -> bool:
if isinstance(exception, ValueError):
return False # business error, not a dependency problem
return exception is not None
breaker = CircuitBreaker(name='payments', classifier=IgnoreValueError())
```
The classifier is strictly more capable: it also sees the **return value**, so
you can count an HTTP `503` response object as a failure — something no
exclude-list can do.
### Listeners
pybreaker's `CircuitBreakerListener` (`before_call` / `state_change` /
`failure` / `success`) maps onto interlock's
[`EventListener`](guides/observability.md):
| pybreaker | interlock |
|---|---|
| `state_change(cb, old, new)` | `on_state_change(*, name, old, new)` |
| `failure(cb, exc)` / `success(cb)` | `on_call(*, name, outcome, duration)` — `outcome.is_failure` distinguishes them |
| `before_call(cb, func, ...)` | — (no per-call pre-hook; use `on_call` after the fact) |
| — | `on_rejected(*, name)` fires when an open circuit rejects a call |
```python
# after — interlock
from interlock import State, Outcome
class Payments:
def on_state_change(self, *, name: str, old: State, new: State) -> None:
print(f'{name}: {old} -> {new}')
def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None: ...
def on_rejected(self, *, name: str) -> None: ...
def on_reset(self, *, name: str) -> None: ...
breaker = CircuitBreaker(name='payments', listener=Payments())
```
A built-in `LoggingEventListener` covers the common case with zero code.
### Redis (shared state)
```python
# before — pybreaker
import redis
import pybreaker
breaker = pybreaker.CircuitBreaker(
state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, redis.StrictRedis()),
)
```
```python
# after — interlock
import redis
from interlock import CircuitBreaker
from interlock.integrations.redis import RedisStorage
breaker = CircuitBreaker(
name='payments',
storage=RedisStorage(redis.Redis(host='redis.internal')),
)
```
interlock's coordination is stronger than shared counters: tripping is atomic
across racing instances, half-open probes are **budgeted globally** (N
instances send at most `permitted_calls_in_half_open` probes *in total*), and a
Redis outage degrades to local state instead of failing your calls
([Redis integration](integrations/redis.md)).
### State inspection and the open error
```python
breaker.current_state # pybreaker → 'open' / 'half-open' / 'closed'
breaker.state # interlock → State.OPEN / HALF_OPEN / CLOSED
```
```python
# before # after
from pybreaker import CircuitBreakerError from interlock import CircuitOpenError
try: try:
breaker.call(charge, 100) breaker.call(charge, 100)
except CircuitBreakerError: except CircuitOpenError as exc:
... # exc.retry_after, exc.breaker_name,
# exc.last_failure
...
```
`CircuitOpenError` carries a `retry_after` estimate (seconds until the next
probe), which the [FastAPI](integrations/fastapi.md) /
[Litestar](integrations/litestar.md) extras turn into `503 + Retry-After`
automatically.
---
## From circuitbreaker
### Decorator
```python
# before — circuitbreaker
from circuitbreaker import circuit
@circuit(failure_threshold=5, recovery_timeout=30, expected_exception=ConnectionError)
def external_call(): ...
```
```python
# after — interlock
from interlock import CircuitBreaker, Config
breaker = CircuitBreaker(
name='external_call',
config=Config(
minimum_number_of_calls=5,
failure_rate_threshold=0.8,
wait_duration_in_open=30.0,
),
classifier=OnlyConnectionError(), # see below
)
@breaker
def external_call(): ...
```
| circuitbreaker | interlock | Notes |
|---|---|---|
| `failure_threshold` | `minimum_number_of_calls` + `failure_rate_threshold` | Streak → rate. |
| `recovery_timeout` | `Config.wait_duration_in_open` | Direct, in seconds. |
| `expected_exception` | a `FailureClassifier` | Only these count as failures; see [below](#expected_exception-failureclassifier). |
| `fallback_function` | `FallbackStrategy` in a pipeline | See [below](#fallback_function-fallbackstrategy). |
| `name=` | `name=` | interlock requires a name explicitly. |
interlock separates the breaker object from the decorator: build one
`CircuitBreaker` and apply it with `@breaker`, rather than configuring a fresh
circuit at each decoration site. To reuse config across many call sites, share
a [`Registry`](guides/configuration.md#sharing-config-with-a-registry).
### `expected_exception` → `FailureClassifier`
circuitbreaker's `expected_exception` is the *inverse* of pybreaker's
`exclude` — it names the exceptions that **do** count. Same classifier tool:
```python
class OnlyConnectionError:
def is_failure(self, *, result: object, exception: Exception | None) -> bool:
return isinstance(exception, ConnectionError)
```
### Class-based breakers
```python
# before — circuitbreaker
from circuitbreaker import CircuitBreaker
class ApiBreaker(CircuitBreaker):
FAILURE_THRESHOLD = 10
RECOVERY_TIMEOUT = 60
EXPECTED_EXCEPTION = ConnectionError
```
There is no subclassing in interlock — the same intent is a reusable `Config`
value (and a classifier), which you pass wherever you need it:
```python
# after — interlock
from interlock import Config
API_CONFIG = Config(
minimum_number_of_calls=10,
failure_rate_threshold=0.8,
wait_duration_in_open=60.0,
)
```
### `fallback_function` → `FallbackStrategy`
circuitbreaker calls `fallback_function` when the circuit is open. interlock
keeps the breaker a pure gate and layers the fallback with the v2.0
[pipeline](guides/pipeline.md), so the substitution is explicit about *which*
failures it stands in for:
```python
# before — circuitbreaker
@circuit(fallback_function=lambda: [])
def recommendations(): ...
```
```python
# after — interlock
from interlock import CircuitBreaker, CircuitOpenError, Pipeline
breaker = CircuitBreaker(name='recommendations')
pipeline = (
Pipeline.builder()
.fallback(lambda exc: [], on=(CircuitOpenError,))
.circuit_breaker(breaker)
.build()
)
@pipeline
def recommendations(): ...
```
The pipeline is also where you compose retries, bulkheads and timeouts around
the same breaker — none of which the old decorator offers.
### Async and monitoring
`@circuit` handles async functions; so does interlock's `@breaker` — with the
**same instance**, no separate class or import. circuitbreaker's
`CircuitBreakerMonitor` (enumerate all circuits, check `all_closed()`) has no
direct analogue; hold your breakers in a [`Registry`](guides/configuration.md)
and iterate that, or observe transitions through the
[listener](guides/observability.md).
`CircuitBreakerError` → `CircuitOpenError`, exactly as in the
[pybreaker section](#state-inspection-and-the-open-error).
---
## aiobreaker and purgatory
[aiobreaker](https://github.com/arlyon/aiobreaker) is pybreaker ported to
asyncio — follow the [pybreaker section](#from-pybreaker); its `fail_max` /
`timeout_duration` / `CircuitBreakerListener` map the same way, and interlock's
single class removes the need for an asyncio-specific breaker at all.
[purgatory](https://github.com/mardiros/purgatory) splits sync and async into
`SyncCircuitBreakerFactory` / `AsyncCircuitBreakerFactory` with a
`default_threshold` (consecutive) and a `default_ttl` on the open state. Map the
factory + `get_breaker(name)` pattern onto a single interlock
[`Registry`](guides/configuration.md#sharing-config-with-a-registry) whose
`registry.get(name)` returns one dual sync/async breaker; `default_ttl` becomes
`wait_duration_in_open`, and `default_threshold` follows the same streak → rate
translation.
---
## What actually changes at runtime
After migrating, expect these behavioural differences — all intended:
- **Trips reflect the rate, not a streak.** A dependency that fails
intermittently under load will now trip where a streak counter never did.
Conversely, a single burst of failures below `minimum_number_of_calls` will
*not* trip — the window has to fill first.
- **Slow calls can trip too, but only when you opt in.**
`slow_call_rate_threshold` defaults to `1.0`, which trips only when every call
in the window is slow, so latency is effectively off until you tune it down
([configuration](guides/configuration.md#why-slow-calls-matter)).
- **Half-open is a budgeted probe round, not a single trial call.** Up to
`permitted_calls_in_half_open` probes run (with a concurrency cap), and the
breaker re-decides from their rate ([states](guides/states.md)).
## Roll out incrementally
You do not have to trust new threshold numbers on day one. Ship the breaker in
[shadow mode](guides/states.md#safe-rollout) — it records real
failure and slow-call rates without rejecting anything — tune against live data,
then switch to enforcing:
```python
breaker.metrics_only() # observe production, reject nothing
# ... read breaker.snapshot().failure_rate / .slow_call_rate over real traffic ...
breaker.reset() # start enforcing with a clean window
```
## Next steps
- [Configuration](guides/configuration.md) — pick your thresholds and window
- [Failure classification](guides/failure-classification.md) — port `exclude` / `expected_exception`
- [Observability](guides/observability.md) — port your listeners
- [Resilience pipeline](guides/pipeline.md) — port `fallback_function`, add retries and timeouts
- [Redis integration](integrations/redis.md) — port shared state
---
# Correctness and testing
interlock-cb is young — [Comparison](comparison.md) says so plainly. What it
can offer instead of years in production is a bar most libraries in this space
do not hold themselves to, and a way to check that the bar is real rather than
promised. Every claim below links to the config, workflow or test file that
enforces it, so it can be verified against the repository at any commit, not
just taken on trust.
## Types
- **Three independent checkers, strict mode**: [mypy](https://mypy-lang.org/)
(`strict = true`), [pyright](https://microsoft.github.io/pyright/)
(`typeCheckingMode = "strict"`) and
[pyrefly](https://pyrefly.org/) (`preset = "strict"`) all run over
`interlock/` and `tests/typing_surface.py` on every pull request
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml),
[`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)).
Three implementations rather than one because they disagree often enough for
it to matter — each catches inference gaps the others miss.
- **The public surface is asserted, not just checked.**
[`tests/typing_surface.py`](https://github.com/bagowix/interlock/blob/main/tests/typing_surface.py)
uses `assert_type` on `call()`'s overloads and the decorator's signature
preservation. It is never executed; it is included in the mypy/pyright/pyrefly
scope, so a regression in what a type checker *infers* at a call site fails
CI even though nothing here runs at test time.
- **Public API breakage is diffed, not just typed.**
[`api-compatibility.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/api-compatibility.yml)
runs [`griffe check`](https://mkdocstrings.github.io/griffe/) against the
latest release tag on every pull request — covering `interlock/__init__.py`'s
re-exports and `interlock/integrations/*`, which is public without being
re-exported. A detected breakage fails the build unless the PR carries the
`breaking-change` label.
- Ships `py.typed`; `ParamSpec` + `TypeVar` decorators preserve both the
wrapped signature and its sync/async nature.
## Coverage
- **100% branch coverage, enforced, not aspirational**:
`fail_under = 100` under `[tool.coverage.report]`
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)),
checked with `branch = true` so a branch taken only one way still fails the
gate. Reported through [Codecov](https://codecov.io/gh/bagowix/interlock)
(project **and** patch targets at 100% —
[`codecov.yml`](https://github.com/bagowix/interlock/blob/main/.github/codecov.yml)).
- **What coverage doesn't prove** is exactly why the next four sections exist —
see [Mutation testing](#mutation-testing) below.
## Property-based and model-based tests
Coverage proves every line ran; it says nothing about whether a test would
notice if the logic were wrong. Two Hypothesis suites target the state machine
specifically, both driven by an injected `FakeClock`
([`tests/conftest.py`](https://github.com/bagowix/interlock/blob/main/tests/conftest.py)),
never real time:
- [`test_state_machine_properties.py`](https://github.com/bagowix/interlock/blob/main/tests/test_state_machine_properties.py) —
`@given` properties over hand-written sequences, checking four invariants:
the minimum-calls gate, a saturated window, the `OPEN` wait, and the probe
caps.
- [`test_state_machine_model.py`](https://github.com/bagowix/interlock/blob/main/tests/test_state_machine_model.py) —
a Hypothesis `RuleBasedStateMachine` that generates the *sequence itself*:
interleaved outcomes, clock advances, admissions and operator overrides,
checked against an independently predicted state, generation, window and
probe budget after every step. This is the suite that catches
order-dependent bugs — an override landing mid-probe-round, a probe settling
a generation late — that a fixed sequence cannot reach. Reaching those
interleavings is not left to chance: the rule mix and a `target()` objective
steer the search into `HALF_OPEN`, and the share of examples that get there
is measured whenever a rule changes, because a generated sequence that never
reaches the interesting state proves nothing.
When the model finds a counterexample, the shrunk sequence gets pinned as a
named regression test next to the model, so the reproducer survives the model
being changed later.
## Mutation testing
100% branch coverage proves every line executes; it does not prove an
assertion would catch a wrong value. [`mutmut`](https://mutmut.readthedocs.io/)
closes that gap for the two modules where a surviving mutant is a real bug:
[`interlock/_state_machine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_state_machine.py)
(threshold arithmetic, transition ordering, probe admission) and
[`interlock/_engine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_engine.py)
(lock scope, dispatch, recording order) —
[`[tool.mutmut]`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)
in `pyproject.toml`.
Baseline: **526 of 549 mutants killed (95.8%)**. Every survivor is an
equivalent mutant in one of five enumerated classes — an unread sentinel
default, a comparison that only ever uses `!=`, a value used solely as a
boolean — documented with the reasoning in
[`CONTRIBUTING.md`](https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md#mutation-testing).
A survivor outside those classes is treated as a missing test, not as grounds
to raise the budget.
Run weekly and on demand
([`mutation.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/mutation.yml)),
never as a pull-request gate — the signal is a slowly-moving score, not
something that should block an unrelated change. The hypothesis suites are
excluded from the run: `mutmut` maps tests to functions from a single stats
pass, and a randomised test that reaches a branch only sometimes would make
both that mapping and the score irreproducible.
## An I/O-free state machine, driven by an injected clock
[`interlock/_state_machine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_state_machine.py)
never touches a socket, a file, or `time.time()` — all time comes from a
`Clock` passed in at construction. Every test in the suite that exercises
timed transitions does so with `FakeClock.advance(seconds)`
([`tests/conftest.py`](https://github.com/bagowix/interlock/blob/main/tests/conftest.py)),
never `sleep`; `test_examples.py` is the one deliberate exception, since it
runs the `examples/` scripts themselves as a subprocess smoke test. This is
what makes the property and model-based suites above practical at all — a
`RuleBasedStateMachine` generating thousands of transition sequences would be
too slow to run against a real clock.
## Concurrency, including free-threaded CPython
- The `threading.Lock` in
[`interlock/_engine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_engine.py)
covers only the two await-free critical sections (admission and recording);
the protected callable itself runs outside it.
- [`tests/test_concurrency.py`](https://github.com/bagowix/interlock/blob/main/tests/test_concurrency.py)
drives one breaker from many real threads at once and checks window counts
add up, `snapshot()` is never torn, the `HALF_OPEN` caps hold, and a
concurrent trip emits exactly one `CLOSED → OPEN` event.
- **CI runs the full matrix on `3.14t`, the free-threaded build, as a required
job** — Python 3.11, 3.12, 3.13, 3.14 and 3.14t, `fail-fast: false`
([`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)).
A GIL-enabled interpreter cannot falsify a thread-safety claim resting on one
lock: with `Engine._lock` removed, `test_concurrency.py` fails on 3.14t and
passes everywhere else. This is the reason the job exists rather than being
aspirational — it is the only Python build in the matrix that can actually
disprove the claim.
## Storage contract and coordinated (distributed) breakers
- [`tests/test_storage_contract.py`](https://github.com/bagowix/interlock/blob/main/tests/test_storage_contract.py)
runs a single behavioural contract suite against the in-memory reference
storage
([`tests/inmemory_storage.py`](https://github.com/bagowix/interlock/blob/main/tests/inmemory_storage.py)) —
the same contract every `Storage` / `AsyncStorage` implementation, including
Redis, is expected to satisfy.
- [`tests/test_coordination.py`](https://github.com/bagowix/interlock/blob/main/tests/test_coordination.py)
covers trip propagation, the global probe budget, coordinated close, and
degradation-and-recovery across a shared storage — fully deterministic, on a
shared `FakeClock` with a manually driven `poll_once()`.
- [`tests/test_redis_storage.py`](https://github.com/bagowix/interlock/blob/main/tests/test_redis_storage.py)
runs against in-process `fakeredis` by default (so `uv run pytest` needs no
server) and against a real Redis service container in CI, which is the
authoritative check for Lua-script atomicity under concurrency. `3.14t`
specifically requires the real server: `fakeredis`'s Lua engine (`lupa`)
re-enables the GIL on import, which `filterwarnings = "error"` turns into a
collection failure.
## Lower-bound dependency versions are tested, not guessed
Every optional extra declares a minimum version in `pyproject.toml`
(`httpx>=0.27.0`, `httpx2>=2.4.0`, `redis>=5.0.0`, and so on). The
`extras-min` job re-pins each
one to exactly that floor and runs the integration suite against it
([`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)) —
so a lower bound is a tested claim, not an untested guess about how far back
compatibility actually reaches. The latest versions of each extra are covered
separately by the main matrix.
## Warnings are errors
`filterwarnings = ["error"]`
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml))
turns every `DeprecationWarning` and `RuntimeWarning` raised during a test run
into a failure, with one narrow, commented exception for a third-party import
warning on CPython 3.11. This is what makes the `3.14t` / `fakeredis` /
`lupa` interaction above a hard collection failure instead of a silent GIL
re-enable, and it means an internal deprecation (`InterlockDeprecationWarning`)
must be explicitly asserted in the test that triggers it, never silenced.
## Linting
`ruff` runs with `select = ["ALL"]`
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)),
including the `S` (flake8-bandit) security ruleset and `flake8-tidy-imports`
bans on legacy `typing` aliases. Ignored rules are listed individually in
`pyproject.toml`, each implicitly scoped to what it actually silences, not
disabled wholesale.
## Hot-path performance
[CodSpeed](https://codspeed.io) measures CPU instructions (not wall-clock
time, so results stay stable on shared CI runners) over the call paths, the
state machine, the sliding windows and the pipeline on every pull request
([`codspeed.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/codspeed.yml)).
A regression is a signal, not silence — this is a performance floor, not a
correctness one, and is listed here for completeness rather than as a
correctness claim.
## Supply chain
Correctness of the code is only half the trust question; the other half is
whether what gets published is what was reviewed. See
[`SECURITY.md`](https://github.com/bagowix/interlock/blob/main/SECURITY.md#supply-chain)
for the full breakdown — trusted publishing with no long-lived PyPI token,
Sigstore-backed [PEP 740](https://peps.python.org/pep-0740/) provenance on
every release artifact, every GitHub Actions `uses:` pinned to a full commit
SHA and audited by [zizmor](https://docs.zizmor.sh) on every pull request, and
CodeQL default setup over both Python and the workflows themselves. The
[OpenSSF Scorecard](https://scorecard.dev/viewer/?uri=github.com/bagowix/interlock)
badge in the README is the standing summary of this section, refreshed weekly
and on every push to `main`. The project also holds the
[OpenSSF Best Practices](https://www.bestpractices.dev/projects/13932) badge —
the self-certified checklist covering release process, vulnerability
reporting, static/dynamic analysis and secure-development practices, most of
it satisfied by the mechanisms this page already documents.
## Known limits
An honest boundary is more persuasive than a claimed clean sweep:
- **Mutation testing covers two modules.** `_state_machine.py` and `_engine.py`
are where a surviving mutant is unambiguously a bug; the rest of the package
(integrations, the pipeline strategies, the registry) is covered by branch
coverage and example-based tests only, not mutation testing.
- **Mutation testing excludes the Hypothesis suites**, for reproducibility —
see [Mutation testing](#mutation-testing) above. A branch reached only by a
property or model-based test can show up as a mutation survivor even when a
property test does in fact cover it.
- **Free-threaded coverage is one test file's job.** `test_concurrency.py` is
the suite specifically designed to fail without the engine's lock; the rest
of the suite runs on 3.14t too, but is not designed to detect races the way
that file is.
- **No production track record.** interlock-cb was first released in 2026 —
see [Comparison](comparison.md). None of the above substitutes for years of
real-world traffic; it is what can be verified today in place of that
history.
- **Fuzzing is not part of this suite.** The OpenSSF Scorecard `Fuzzing` check
is expected to stay red — the input surface here is typed function calls and
config values, not a parser or protocol decoder, so structure-aware fuzzing
does not apply the way it would to a format library.
- **Branch protection is not machine-verifiable from the public Scorecard
run.** The `Branch-Protection` check needs a PAT with read access to
repository settings that the workflow does not have; the settings themselves
are configured in GitHub, not in a file this page can link to.
---
# Configuration
`Config` is an immutable (frozen) dataclass validated on construction. Pass it
to a `CircuitBreaker` or share it across a `Registry`. All fields are
keyword-only.
```python
from interlock import Config
from interlock import WindowType
config = Config(
failure_rate_threshold=0.5,
minimum_number_of_calls=20,
slow_call_duration_threshold=2.0,
slow_call_rate_threshold=1.0,
permitted_calls_in_half_open=10,
max_concurrent_probes=1,
wait_duration_in_open=30.0,
window_type=WindowType.COUNT_BASED,
window_size=100,
)
```
## Fields
| Field | Default | Meaning |
|-------|---------|---------|
| `failure_rate_threshold` | `0.5` | Trip when the failure rate reaches this fraction. Range `(0, 1]`. |
| `minimum_number_of_calls` | `10` | Minimum calls in the window before a rate is trusted. Guards against `1/1 = 100%`. |
| `slow_call_duration_threshold` | `60.0` | Calls at or above this many seconds are **slow**. |
| `slow_call_rate_threshold` | `1.0` | Trip when the slow-call rate reaches this fraction. Range `(0, 1]`. |
| `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`, and must stay `1.0` on a breaker with a shared storage. |
| `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). |
Validation raises `ValueError` eagerly for out-of-range or inconsistent values,
so a misconfigured breaker fails at construction rather than in production.
## Windows
- **Count-based** keeps the last `window_size` calls. Predictable memory,
independent of traffic rate. The default.
- **Time-based** keeps calls from the last `window_size` seconds. The right
choice for high-throughput services where "last N calls" is a moving target.
```python
from interlock import Config, WindowType
# Trip on a 50% failure rate observed over the last 30 seconds.
Config(window_type=WindowType.TIME_BASED, window_size=30)
```
## Why slow calls matter
A dependency that answers slowly but never errors will never trip a
failure-rate breaker, yet it still exhausts your timeouts and threads.
Slow-call detection treats latency as a first-class failure signal. The default
`slow_call_rate_threshold=1.0` trips only when every call in the window is slow,
so latency is effectively off until you tune it down — safe to leave on while
you observe.
## Sharing config with a Registry
```python
from interlock import Config, Registry, State
registry = Registry(
config=Config(minimum_number_of_calls=20),
initial_state=State.METRICS_ONLY,
)
payments = registry.get('payments') # shared default
search = registry.get('search', config=Config(window_size=500)) # per-name override
```
The override applies only when the breaker is first created; later `get` calls
with the same name return the existing instance and ignore the `config`
argument. `initial_state` is also assigned once, inside the registry lock,
before a newly created breaker can serve traffic. See
[Safe rollout](states.md#safe-rollout) for using
`METRICS_ONLY` in production.
---
# States & manual control
A breaker has three core states plus three operator overrides.
## Core lifecycle
```mermaid
stateDiagram-v2
CLOSED --> OPEN: failure/slow rate crosses threshold
OPEN --> HALF_OPEN: first call after wait_duration_in_open (or timer, if auto_transition)
HALF_OPEN --> CLOSED: probe round passes
HALF_OPEN --> OPEN: probe round fails
```
- **`CLOSED`** — traffic flows; outcomes are recorded. When the failure rate (or
slow-call rate) crosses its threshold over at least `minimum_number_of_calls`,
the breaker trips to `OPEN`.
- **`OPEN`** — calls are rejected immediately with `CircuitOpenError`. After
`wait_duration_in_open` seconds, the **next** call lazily moves the breaker to
`HALF_OPEN`. Enable [`auto_transition`](#proactive-transition-auto_transition)
to have a timer make that move on its own.
- **`HALF_OPEN`** — up to `permitted_calls_in_half_open` probe calls are
admitted, with a cap on how many run concurrently, so a barely-recovered
dependency is not hit by the full parallel load at once. Once the round
completes, the breaker decides from the probes' outcomes using the same
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.
The backoff is local, and deliberately refuses to pretend otherwise. Reopening
a coordinated breaker is decided by the backend from `wait_duration_in_open` and
its own clock, and no failed-round count crosses the wire, so a multiplier above
`1.0` alongside a shared [storage](../integrations/redis.md) raises `ValueError`
at construction rather than being accepted and ignored. Coordinated backoff is
[under discussion](https://github.com/bagowix/interlock/issues); until then a
coordinated breaker waits a constant interval.
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
after `wait_duration_in_open` elapses. A low-traffic service can therefore sit in
`OPEN` longer than necessary, and — since nothing changes until that call — the
state-change event is not emitted, leaving a blind spot on dashboards.
Set `auto_transition=True` to arm a timer that performs the move on its own when
the wait elapses, emitting `on_state_change` without waiting for a call:
```python
from interlock import CircuitBreaker, Config
breaker = CircuitBreaker(
name='payments',
config=Config(wait_duration_in_open=30.0, auto_transition=True),
)
# 30s after opening, the breaker moves to HALF_OPEN and emits the event,
# even if no call arrives.
```
The lazy path stays authoritative: the timer only flips the state (it admits no
probe), so the first real call still becomes the first probe. If a call arrives
exactly as the timer fires, a lock ensures the transition and its event happen
exactly once. The timer is cancelled automatically on `reset()`, `force_open()`,
or when a call makes the move first.
The timer is a daemon thread, used uniformly for sync and async breakers (the
breaker's critical sections are guarded by a `threading.Lock`, never an event
loop), so a pending timer never blocks interpreter shutdown.
## Operator overrides
Three special states are set manually and stay until you `reset()`:
| Method | State | Behaviour |
|--------|-------|-----------|
| `breaker.force_open()` | `FORCED_OPEN` | Reject all traffic regardless of metrics. |
| `breaker.disable()` | `DISABLED` | Admit all traffic but record no outcome — thresholds are never evaluated and `snapshot()` gets nothing new. Listener `on_call` events still fire. |
| `breaker.metrics_only()` | `METRICS_ONLY` | Admit all traffic, record metrics, but never trip. |
| `breaker.reset()` | `CLOSED` | Return to closed with a fresh, empty window. In coordinated mode, resume the cached shared state instead. |
```python
breaker.metrics_only() # observe in production without enforcing
# ... inspect breaker.snapshot() until thresholds look right ...
breaker.reset() # start enforcing with a clean window
```
### What an override does to your metrics
Two observability surfaces are in play, and an override does not move them
together:
- the sliding **window** — what `snapshot()` reports and what the thresholds
read. `METRICS_ONLY` keeps filling it (that is the whole point of shadow
mode); `DISABLED` records nothing and `FORCED_OPEN` admits nothing to record,
so neither feeds it. A count-based window then keeps its last contents
unchanged; a time-based one drains as its buckets expire;
- the **`EventListener`**, which observes calls rather than the window.
`on_call` fires whenever an admitted call settles, whatever the state —
`DISABLED` included; `on_rejected` fires for every rejected call,
`FORCED_OPEN` included.
So `disable()` is not a way to silence a listener: `LoggingEventListener`, the
`OTelEventListener` or a Prometheus exporter keeps reporting outcomes and
durations for a disabled breaker, with the classifier still deciding success
from failure. That is deliberate — dashboards going dark the moment an operator
disables a breaker looks exactly like an outage. To stop the events, drop the
listener instead (construct the breaker without one).
Switching a rollout from `metrics_only()` to `disable()` therefore keeps
listener-exported dashboards alive, and only stops threshold evaluation and
`snapshot()`.
### Safe rollout
Shadow mode is the key to introducing a breaker without risk: it records the
exact failure and slow-call rates real traffic produces, so you can tune
thresholds against live data before letting the breaker reject anything. It
costs almost nothing to leave on.
Set the mode at construction when no call may be admitted first:
```python
from interlock import CircuitBreaker, Registry, State
breaker = CircuitBreaker(name='payments', initial_state=State.METRICS_ONLY)
registry = Registry(initial_state=State.METRICS_ONLY)
```
`Registry` applies the state while holding its creation lock and publishes the
breaker only afterwards. Every name created later therefore starts in shadow
mode too. Construction is not a state transition, so it does not emit a
synthetic `CLOSED → METRICS_ONLY` listener event.
Only stable states are valid at construction: `CLOSED`, `FORCED_OPEN`,
`DISABLED` and `METRICS_ONLY`. `OPEN` and `HALF_OPEN` require timing, probe and
failure history, so passing either as `initial_state` raises `ValueError`.
For a production rollout:
1. Deploy with `initial_state=State.METRICS_ONLY` and an `EventListener` that
exports call outcomes.
2. Observe failure and slow-call rates, then tune `Config` against real traffic.
3. Deploy a new breaker, registry or transport with `initial_state=State.CLOSED`
(the default). The enforcing instance starts with a fresh window.
Prefer a new deployment for step 3. Calling `reset()` enforces immediately for
an existing breaker, but a registry configured with `METRICS_ONLY` would still
apply that original initial state to hosts first seen later.
For local diagnosis, `registry.get_existing(name)` returns a cached breaker or
`None` without creating one. Inspect its `state` and `snapshot()`; use listeners
rather than polling snapshots for production metrics.
When the names are not known in advance — the HTTP transports create one
breaker per host, lazily — `registry.items()` lists every breaker created so
far, and `registry.names()` just their names. Both return a point-in-time copy,
so they also drive bulk operator actions:
```python
for _, breaker in registry.items():
breaker.metrics_only()
```
## Coordinated state (optional)
With a shared [storage](../integrations/redis.md), `OPEN` and `HALF_OPEN` can
also be *adopted* from other instances: a trip anywhere in the fleet gates
admission everywhere, and the HALF_OPEN probe budget is shared globally.
`breaker.state` then reports the effective state — the shared one when it
governs admission, the local one otherwise (including while the storage is
unreachable).
Local operator overrides always take precedence over a healthy shared view:
`force_open()` rejects locally, while `disable()` and `metrics_only()` admit
locally without claiming a shared HALF_OPEN probe. `reset()` clears that local
override and freshens local metrics; it does not change the cluster. The
instance immediately resumes the cached shared `OPEN` or `HALF_OPEN` state.
## Observing transitions
Every transition (and reset) is delivered to the breaker's
[`EventListener`](observability.md), so you can log or export state changes
without polling `breaker.state`.
---
# Failure classification
What counts as a failure is a separate concern from *when to trip* (thresholds,
in [Config](configuration.md)). It is decided by a `FailureClassifier`.
## Default policy
By default, a call is a failure exactly when it **raises**, and any returned
value is a success:
```python
from interlock import CircuitBreaker
breaker = CircuitBreaker(name='svc') # DefaultFailureClassifier
```
This is right for code that signals errors by raising. It is *not* enough when
failure is encoded in a **return value** — for example an HTTP response object
whose `503` status means the dependency is unhealthy.
## Classify by result
A classifier implements one method. The `result`/`exception` pair is mutually
exclusive: when `exception` is not `None` the call raised; otherwise `result`
holds the return value. `exception` is always an `Exception` — a
`BaseException` such as `CancelledError` says nothing about the dependency, so
the breaker releases the call without classifying it.
```python
from interlock import CircuitBreaker
class StatusClassifier:
def is_failure(self, *, result: object, exception: Exception | None) -> bool:
if exception is not None:
return True
return getattr(result, 'status_code', 200) >= 500
breaker = CircuitBreaker(name='api', classifier=StatusClassifier())
result = breaker.call(client.get, url) # a 503 response now counts as a failure
```
Result-based classification needs the return value, so it works with the
**decorator** and **`call`**, but not the context manager (which only sees
exceptions and duration).
## Ignore expected errors
Business errors — a `404`, a validation failure — should not open the circuit.
Encode that by treating only the exceptions you care about as failures:
```python
class IgnoreNotFound:
def is_failure(self, *, result: object, exception: Exception | None) -> bool:
if isinstance(exception, NotFoundError):
return False # expected, not a dependency problem
return exception is not None
```
An ignored exception is recorded as a **success** — the sliding window has only
two outcomes — and still propagates to the caller. It therefore dilutes the
failure rate rather than being invisible to it.
## HTTP out of the box
For HTTP clients, you do not need to write this yourself — the
[httpx2](../integrations/httpx2.md), [httpx](../integrations/httpx.md),
[aiohttp](../integrations/aiohttp.md), and
[requests](../integrations/requests.md) integrations ship
`HttpStatusClassifier`, which treats the canonical retryable statuses
(`429, 500, 502, 503, 504`) and every non-excluded transport exception as
failures.
Errors the *caller* caused are excluded for the same reason as a `404`: a
scheme-less URL or a local protocol violation is a bug in your code, not
evidence that the dependency is unhealthy, and counting it would open the
circuit of a host that is answering fine. The default set differs per
integration, because each client library raises different things inside the
guarded call:
| Integration | Excluded by default |
|---|---|
| [httpx2](../integrations/httpx2.md), [httpx](../integrations/httpx.md) | `UnsupportedProtocol`, `LocalProtocolError` |
| [requests](../integrations/requests.md) | `InvalidURL` (and its `InvalidProxyURL` subclass) |
| [aiohttp](../integrations/aiohttp.md) | nothing — it rejects malformed URLs before the middleware chain runs, so every handler exception counts unless you exclude it |
All four take `excluded_exceptions=(...)` to replace that set — `()` counts
every exception as a failure. Entries must be `Exception` subclasses;
anything else raises `TypeError` at construction.
---
# Observability
A breaker reports everything it does through an `EventListener`. The same
hooks back logging, metrics, and any custom sink.
## The hooks
```python
from typing import Protocol
from interlock import Outcome, State
class CoreEventListener(Protocol):
def on_state_change(self, *, name: str, old: State, new: State) -> None: ...
def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None: ...
def on_rejected(self, *, name: str) -> None: ...
def on_reset(self, *, name: str) -> None: ...
class StorageEventListener(Protocol):
def on_storage_degraded(self, *, name: str, error: BaseException) -> None: ...
def on_storage_recovered(self, *, name: str) -> None: ...
def on_storage_write_dropped(self, *, name: str) -> None: ...
class PipelineEventListener(Protocol):
def on_retry(self, *, name: str, attempt: int, delay: float) -> None: ...
def on_bulkhead_rejected(self, *, name: str) -> None: ...
def on_fallback(self, *, name: str, error: BaseException) -> None: ...
class EventListener(
CoreEventListener,
StorageEventListener,
PipelineEventListener,
Protocol,
): ...
```
Core listeners are called **outside** the breaker's lock, so a slow listener
never serialises throughput. `on_call` runs after the protected call completes;
transition, rejection, and reset hooks run immediately after their corresponding
breaker operation. Storage hooks run from the coordinated breaker's background
lane when its backend degrades, recovers, or drops a queued write.
Pipeline hooks run inside their strategy's execution path: before retry backoff,
when bulkhead admission fails, or before a fallback returns its substitute.
The protocols follow the owner of each event. `CoreEventListener` observes the
breaker itself, `StorageEventListener` observes breakers coordinated through a
shared [storage](../integrations/redis.md), and `PipelineEventListener` observes
[pipeline strategies](pipeline.md) given a `listener=`. `EventListener` combines
all three for sinks that observe the complete library.
The `name` namespace follows the same boundary: core and storage hooks receive
the **breaker name**, while pipeline hooks receive the **strategy name**. A
listener can therefore use `name` directly as its breaker or strategy label
without rediscovering the distinction from the call site.
Every hook is dispatched only if present, and each protocol supplies no-op
implementations for subclasses. A listener can override just the hooks it cares
about and keeps working when a later interlock version adds a new hook.
## Listener failures are isolated
Observability is optional; the protected call is not. A listener runs on the
breaker's own paths, so a bug in one — a metrics exporter with a stale label,
a logging handler with a full queue — would otherwise become a new failure
source. interlock guarantees it cannot:
- an `Exception` raised by a hook is logged to the `interlock` logger at
`ERROR` (with the traceback) and then ignored;
- the protected call keeps its result, and a failing call keeps *its own*
exception — a listener never masks the dependency's error;
- state transitions, probe accounting and the coordinated-mode background lane
continue unaffected;
- a raising hook is never reported as storage degradation.
`BaseException` is **not** caught: `KeyboardInterrupt` and
`asyncio.CancelledError` propagate from a hook exactly as they do everywhere
else in interlock.
This covers `EventListener` hooks only. User-supplied *policy* callbacks are
part of the call's behaviour rather than observations of it, and keep raising
as before: a [`FailureClassifier`](failure-classification.md), a
[fallback](pipeline.md) function, a tenacity `before_sleep` hook.
To be notified when your own listener misbehaves, watch the `interlock`
logger:
```python
import logging
logging.getLogger('interlock').setLevel(logging.ERROR)
```
Attach one per breaker, or share one across a `Registry`:
```python
breaker = CircuitBreaker(name='payments', listener=my_listener)
registry = Registry(listener=my_listener) # every breaker reports here
```
## Logging (zero dependencies)
`LoggingEventListener` is built in. State changes and rejections log at
`WARNING`, resets at `INFO`, and individual calls at `DEBUG`:
```python
from interlock import CircuitBreaker, LoggingEventListener
breaker = CircuitBreaker(name='payments', listener=LoggingEventListener())
```
Pass your own logger to control routing:
```python
import logging
LoggingEventListener(logging.getLogger('myapp.breakers'))
```
## OpenTelemetry metrics
The OTel listener lives in the `interlock-cb[otel]` extra and is imported
explicitly, so the core stays dependency-free:
```bash
uv add 'interlock-cb[otel]'
```
Supports `opentelemetry-api>=1.20.0` — the listener only calls
`get_meter`/`create_histogram`/`create_counter`, stable across any
`opentelemetry-distro`/SDK release from that version on.
```python
from interlock import CircuitBreaker
from interlock.integrations.otel import OTelEventListener
breaker = CircuitBreaker(name='payments', listener=OTelEventListener())
```
It records five instruments on the `interlock` meter (or a meter you pass in):
| Instrument | Type | Labels |
|------------|------|--------|
| `interlock.call.duration` | histogram (s) | `breaker`, `outcome` |
| `interlock.call.rejected` | counter | `breaker` |
| `interlock.state.changes` | counter | `breaker`, `from`, `to` |
| `interlock.reset` | counter | `breaker` |
| `interlock.storage.events` | counter | `breaker`, `event` (`degraded`/`recovered`/`write_dropped`), `error` |
## Custom listeners
For a partial listener that strict type checkers can verify, inherit the narrowest
protocol for its owner and override only the hooks you need. Every inherited hook
is a no-op:
```python
from interlock import CoreEventListener
class RejectionCounter(CoreEventListener):
def __init__(self) -> None:
self.rejected = 0
def on_rejected(self, *, name: str) -> None:
self.rejected += 1
```
Use `StorageEventListener` for storage-only sinks, `PipelineEventListener` for
strategy-only sinks, or `EventListener` when one object handles every group.
Inheritance is optional for a listener that structurally implements the relevant
protocol. At runtime, dispatch is still by name and skips any missing hook,
including on older listener objects that inherit none of them.
---
# Timeout
A circuit breaker without a timeout is unsafe. A call that hangs forever is
never counted as slow or failed — it just holds a resource indefinitely.
`timeout` bounds an awaited block and turns a hang into a `CallTimeoutError`,
which a surrounding breaker records as a (slow) failure.
```python
from interlock import timeout
async with timeout(2.0):
await client.get(url) # raises CallTimeoutError after 2 seconds
```
## Composing with a breaker
Compose `timeout` with a breaker manually — pipeline composition is a v2
feature. Put the timeout *inside* the protected callable so the breaker observes
the `CallTimeoutError`:
```python
from interlock import CircuitBreaker, timeout
breaker = CircuitBreaker(name='search')
@breaker
async def search(q: str) -> bytes:
async with timeout(2.0):
return await client.get('/search', params={'q': q})
```
Now a request that exceeds 2 seconds raises `CallTimeoutError`; the breaker
counts it as a failure and, once the failure rate crosses the threshold, opens
the circuit — converting slow hangs into fast rejections.
## Synchronous code
`timeout` relies on asyncio cancelling the coroutine in place, which has no
synchronous equivalent: a blocking call cannot be interrupted from outside its
own thread, and `signal.SIGALRM` only works in the main thread, so it breaks in
threaded servers. `sync_timeout` instead runs the callable in a daemon worker
thread and joins it with a deadline. It is a decorator, so it wraps a *callable*
rather than a block:
```python
from interlock import CircuitBreaker, sync_timeout
breaker = CircuitBreaker(name='search')
@breaker
@sync_timeout(2.0)
def search(q: str) -> bytes:
return client.get('/search', params={'q': q}).content
```
A call that exceeds 2 seconds raises `CallTimeoutError`, which the breaker
records exactly as with the async path. The decorator preserves the wrapped
function's signature, arguments and return value.
!!! warning "The worker keeps running after a timeout"
Python cannot forcibly kill a thread. After `sync_timeout` raises, the
worker thread keeps running in the background until the call returns on its
own — it cannot be cancelled, so it may still hold the resource it was
waiting on. The caller is unblocked immediately, but the underlying work is
not stopped. Prefer the async `timeout` wherever you control an event loop;
reach for `sync_timeout` only in genuinely synchronous code.
## Why not bake it in?
interlock keeps retry, fallback and timeout as explicit, observable features
rather than hidden magic inside the breaker. You decide the deadline at the call
site, and the failure it produces flows through the same classification and
metrics as any other.
---
# Retries and circuit breakers
Retries and circuit breakers pull in opposite directions: a retry *adds*
load to a struggling dependency, a breaker *sheds* it. Combined carelessly
they cancel each other out — retries hammer a dependency the breaker is
trying to protect, or the breaker's window never sees the real failure rate.
This guide fixes the composition; the ready-made tenacity helpers live in the
[tenacity integration](../integrations/tenacity.md).
## Which goes on the outside?
Both orders are valid — they answer different questions. What changes is what
the breaker's sliding window *sees*:
| Order | What the window sees | When to choose |
|---|---|---|
| **Retry outside → breaker inside** (recommended) | Every attempt individually — honest failure rate, the breaker trips as early as the dependency deserves | Default. Also the order used by Polly and resilience4j |
| Breaker outside → retry inside | One aggregated outcome per *operation* (all attempts folded into it) | When thresholds are tuned per business operation, not per request |
With retry outside, a rejected attempt is also visible to the retry loop —
which is exactly where the two failure modes below come from.
## Failure mode 1: retrying an open circuit
`CircuitOpenError` is not a transient error. The breaker rejects instantly,
so an exponential backoff loop around it burns its attempt budget in
milliseconds, never reaches the dependency, and buries the real signal in log
noise. Stop retrying the moment the circuit opens:
```python
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter
from interlock.integrations.tenacity import retry_unless_open
retrying = Retrying(
retry=retry_unless_open(TimeoutError, ConnectionError),
wait=wait_exponential_jitter(),
stop=stop_after_attempt(5),
reraise=True,
)
```
## Failure mode 2: blind waiting
Sometimes waiting *is* the right call — a nightly job would rather sleep
than fail. But `2^n` seconds is the wrong amount: the breaker already knows
when it will allow the next probe (`CircuitOpenError.retry_after`). Wait
exactly that long:
```python
from tenacity import AsyncRetrying, retry_if_exception_type, stop_after_attempt
from tenacity import wait_exponential_jitter
from interlock import CircuitOpenError
from interlock.integrations.tenacity import wait_probe
retrying = AsyncRetrying(
retry=retry_if_exception_type((TimeoutError, CircuitOpenError)),
wait=wait_probe(wait_exponential_jitter()),
stop=stop_after_attempt(10),
reraise=True,
)
```
Pick one mode per call site. Fail fast at request/latency-sensitive
boundaries; be patient in background work.
## Retrying on HTTP statuses
The HTTP integrations classify statuses for the *breaker* without raising —
a `503` response is returned to you, recorded as a failure. tenacity,
however, is exception-driven. Do **not** reach for `retry_if_result`: with
aiohttp a retried-away response is never released and leaks its connection.
Turn bad statuses into exceptions instead, then retry exceptions:
```python
import requests
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter
from interlock.integrations.requests import CircuitBreakerAdapter
from interlock.integrations.tenacity import retry_unless_open
session = requests.Session()
session.mount('https://', CircuitBreakerAdapter())
def fetch_orders() -> dict:
response = session.get('https://api.example.com/orders')
response.raise_for_status()
return response.json()
retrying = Retrying(
retry=retry_unless_open(requests.HTTPError, requests.ConnectionError),
wait=wait_exponential_jitter(),
stop=stop_after_attempt(5),
reraise=True,
)
orders = retrying(fetch_orders)
```
The breaker still classifies by status (no exception needed), the retry loop
reacts to `raise_for_status()` — each tool sees the signal in its native
form. To align which statuses trip the breaker, pass
`HttpStatusClassifier(failure_statuses={...})` to the integration.
## Anti-patterns
- **Unbounded retries.** Always set a `stop` condition. A breaker caps
concurrent damage, not the lifetime of a stubborn loop.
- **Retries without a breaker.** N clients × M retries is an N·M-fold
amplification aimed at a dependency that is already failing — the classic
retry storm. The breaker inside the loop is what breaks it.
- **Retrying non-transient errors.** A `404` or a validation error will not
succeed on attempt five. List transient exception types explicitly in
`retry_unless_open(...)` rather than retrying everything.
- **Nested retry layers.** urllib3's `max_retries`, your service mesh and
tenacity each multiply attempts. Budget them together — one deliberate
retry layer beats three accidental ones.
## Declarative composition
Everything on this page stays valid with the v2 [resilience
pipeline](pipeline.md) — `RetryStrategy` packages the same predicates and
wait strategies as a layer, so the manual recipe becomes::
```python
pipeline = (
Pipeline.builder()
.retry(attempts=4) # retry outside — the recommended order
.circuit_breaker(breaker)
.timeout(2.0)
.build()
)
```
---
# Resilience pipeline
v2.0 turns interlock's primitives into composable **strategies**: timeout,
bulkhead, circuit breaker, retry and fallback applied around one call in an
explicit order, mirroring [Polly's](https://www.pollydocs.org/pipelines)
`ResiliencePipeline` semantics.
The pipeline is an additive layer — the standalone
[`CircuitBreaker`](../getting-started.md) remains a first-class primitive, and
existing v1 code keeps working unchanged. Reach for a pipeline when one
concern is not enough.
## At a glance
```python
from interlock import CircuitBreaker, CircuitOpenError, Pipeline
breaker = CircuitBreaker(name='recommendations')
pipeline = (
Pipeline.builder()
.fallback(lambda exc: [], on=(CircuitOpenError,)) # outermost
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0) # innermost
.build()
)
@pipeline
async def fetch_picks(user: str) -> list[str]:
return await client.get_picks(user)
```
One pipeline serves sync and async callables alike: the decorator and
`pipeline.call(fn, ...)` detect the callable's nature and dispatch, exactly
like `CircuitBreaker.call`. The decorator preserves the wrapped signature for
type checkers.
## Order is explicit — first is outermost
Strategies apply in declaration order: the first strategy sees everything the
inner layers produce. `Pipeline(A, B)` means `A(B(call))`. There is no hidden
"correct" order in the code; the recommended one is a documented default:
| Layer (outer → inner) | Why here |
|---|---|
| `fallback(...)` | Substitutes a value for whatever the inner stack gave up on — including rejections raised by the strategies themselves |
| `retry(...)` | Each attempt below is a complete guarded call: the breaker sees honest per-attempt statistics and stops the retry loop the moment the circuit opens |
| `circuit_breaker(...)` | Counts timeouts and failures of every attempt; open circuit rejects before the bulkhead slot or a connection is touched |
| `bulkhead(...)` | Inside retry — otherwise every backoff-and-retry cycle would multiply slot occupancy |
| `timeout(...)` | Innermost: bounds a single attempt, so one hung attempt cannot eat the whole retry budget |
Deviating is legitimate — e.g. a breaker *outside* retry counts one
aggregated outcome per operation instead of one per attempt (see
[Retries and circuit breakers](retries.md) for that trade-off) — but do it
deliberately.
## The strategies
### `CircuitBreakerStrategy`
Wraps a standalone [`CircuitBreaker`](../reference.md) without touching it:
the window, events, manual controls and the breaker's own listener behave
exactly as in direct use, and the same instance can still be called directly.
An open circuit raises `CircuitOpenError` before any inner layer runs.
```python
from interlock import CircuitBreakerStrategy, Pipeline
pipeline = Pipeline(CircuitBreakerStrategy(breaker))
```
### `TimeoutStrategy`
Bounds every attempt using the v1 primitives: `asyncio.timeout` on the async
path (the attempt is cancelled), `sync_timeout` on the sync path — which
inherits its [worker-thread limitation](timeout.md): the caller gets
`CallTimeoutError` on time, but Python cannot kill the overrunning thread.
### `BulkheadStrategy`
Caps how many calls run through the layer concurrently. With no free slot the
call fails immediately (`max_wait=0`, the default) or waits up to `max_wait`
seconds, then raises `BulkheadFullError`:
```python
from interlock import BulkheadStrategy, Pipeline
pipeline = Pipeline(BulkheadStrategy(8, max_wait=0.5))
```
`BulkheadFullError` is deliberately not `CircuitOpenError`: a full bulkhead
means *this process* is saturated, not that the dependency is unhealthy — the
right reaction is shedding load, not backing off. Sync calls share a
`threading.Semaphore`, async calls an `asyncio.Semaphore`; one configuration,
two independent pools.
### `FallbackStrategy`
Substitutes an explicit value for selected failures — nothing silent:
```python
from interlock import CircuitOpenError, FallbackStrategy, Pipeline
cached: list[str] = []
strategy = FallbackStrategy(lambda exc: cached, on=(CircuitOpenError,))
```
- The substitution happens **only** for exception types named in `on`;
anything else propagates.
- The `fallback` callable receives the exception it stands in for.
- `on` accepts `Exception` subclasses exclusively — cancellation and
`KeyboardInterrupt` always propagate.
- The strategy's own result type is the honest union `T | F`, not `Any`. At
the pipeline level the substitute is expected to be shaped like the call's
result (the same contract as Polly and resilience4j).
- A fallback never masks shadow-mode statistics: a `metrics_only` breaker
below it keeps recording every failure.
### `RetryStrategy` (the `tenacity` extra)
interlock ships no retry engine; the strategy delegates all policy to
[tenacity](https://tenacity.readthedocs.io/) and packages the
[retry × breaker glue](../integrations/tenacity.md) for the pipeline:
```python
from interlock.integrations.tenacity import RetryStrategy
strategy = RetryStrategy(attempts=4) # fail-fast: never retries CircuitOpenError
```
Attempts are always capped, the original exception is re-raised when the
budget runs out, and the default predicate stops retrying the moment the
circuit opens. For the *patient* mode (wait exactly until the breaker's next
probe) pass `wait=wait_probe(...)` — see the
[tenacity integration](../integrations/tenacity.md). The builder step
`.retry(...)` imports the extra lazily, so the pipeline core stays
zero-dependency.
## Two usage forms, not three
A pipeline works as a decorator and as `pipeline.call(fn, *args, **kwargs)` —
the same signature-preserving contracts as the breaker's. There is
deliberately **no context manager**: a `with` block cannot be re-run, so a
retry layer inside it is semantically impossible. This is the same honesty as
the v1 breaker's context manager not supporting result-based classification —
rather than a form that silently ignores half the strategies, the form does
not exist.
## Migrating from v1 (nothing breaks)
The v1 API is untouched — migration is wrapping, not rewriting:
```python
# v1: the breaker guards the call directly
result = breaker.call(fetch_orders, user_id)
# v2: the same breaker, now composed with a timeout
pipeline = Pipeline(CircuitBreakerStrategy(breaker), TimeoutStrategy(2.0))
result = pipeline.call(fetch_orders, user_id)
```
The manual composition recipe from the
[retries guide](retries.md) — `Retrying` wrapped around `breaker.call` —
keeps working and remains the most flexible form; the pipeline is that recipe
made declarative.
## Observability
`RetryStrategy`, `BulkheadStrategy` and `FallbackStrategy` (and their builder
steps) accept `name=` and `listener=`. Three optional
[`EventListener`](observability.md) hooks make the pipeline's decisions
visible — `on_retry(name, attempt, delay)`, `on_bulkhead_rejected(name)` and
`on_fallback(name, error)`:
```python
from interlock import LoggingEventListener, Pipeline
events = LoggingEventListener()
pipeline = (
Pipeline.builder()
.fallback(lambda exc: [], on=(CircuitOpenError,), name='recs', listener=events)
.circuit_breaker(breaker) # the breaker keeps its own listener
.bulkhead(8, name='recs', listener=events)
.timeout(2.0)
.build()
)
```
Strategy hooks go through the same dispatcher as the breaker's own: dispatched
only if defined (listeners written before v2.0 keep working unchanged), and a
hook that raises is logged and ignored rather than failing the call — see
[listener failures are isolated](observability.md#listener-failures-are-isolated).
A `fallback` function is *not* a hook: it shapes the result, so its errors
propagate. `LoggingEventListener` logs retries at INFO and bulkhead rejections
/ fallbacks at WARNING; `OTelEventListener` counts all three in the
`interlock.pipeline.events` counter.
## Custom strategies
Any object with `execute` / `execute_async` is a strategy — the `Strategy`
protocol is structural:
```python
import time
from collections.abc import Awaitable, Callable
from typing import TypeVar
T = TypeVar('T')
class Measuring:
"""Times every layer below it."""
def execute(self, call: Callable[[], T]) -> T:
start = time.perf_counter()
try:
return call()
finally:
print(f'took {time.perf_counter() - start:.3f}s')
async def execute_async(self, call: Callable[[], Awaitable[T]]) -> T:
start = time.perf_counter()
try:
return await call()
finally:
print(f'took {time.perf_counter() - start:.3f}s')
pipeline = Pipeline.builder().add(Measuring()).timeout(2.0).build()
```
The contract, in full:
- Run the zero-argument next layer, return its result, let exceptions
propagate. Never swallow `BaseException` — cancellation must cross every
layer untouched.
- `execute_async` always receives a real coroutine function, so
detect-dispatching primitives (like `breaker.call`) treat it as async.
---
# Integrations
interlock plugs into the HTTP client, framework or retry library you already
use — you configure thresholds once and the breaker applies **per host** (or
per named dependency) with no decorators in call sites.
## Supported integrations
| Integration | Extra | What you get |
|---|---|---|
| [FastAPI](fastapi.md) | `interlock-cb[fastapi]` | `Depends`-injected breakers and a `CircuitOpenError → 503 + Retry-After` handler |
| [Litestar](litestar.md) | `interlock-cb[litestar]` | `Provide`-injected breakers and a `CircuitOpenError → 503 + Retry-After` handler (Litestar ≥ 2.23) |
| [httpx2](httpx2.md) | `interlock-cb[httpx2]` | `CircuitBreakerTransport` / `AsyncCircuitBreakerTransport` — per-host breaker at the transport level |
| [httpx](httpx.md) | `interlock-cb[httpx]` | `CircuitBreakerTransport` / `AsyncCircuitBreakerTransport` — per-host transport for httpx clients |
| [aiohttp](aiohttp.md) | `interlock-cb[aiohttp]` | `CircuitBreakerMiddleware` — per-host breaker as a client middleware (aiohttp ≥ 3.12) |
| [requests](requests.md) | `interlock-cb[requests]` | `CircuitBreakerAdapter` — per-host breaker mounted on a `Session` |
| [LLM SDKs](llm.md) | — (recipe) | Guard OpenAI / Anthropic SDK calls with a breaker + bounded retries |
| [tenacity](tenacity.md) | `interlock-cb[tenacity]` | Retry × breaker glue: stop retrying when the circuit opens, or wait exactly until the next probe |
| [Redis](redis.md) | `interlock-cb[redis]` | Shared breaker state across processes with graceful degradation |
| [Flask / Django](frameworks.md) | — (recipe) | Map `CircuitOpenError` to `503 + Retry-After` in other web frameworks |
## How integrations are built
Every integration follows the same rules, so learning one means knowing all:
- **Native extension points only.** A transport (httpx2/httpx), a client middleware
(aiohttp), an adapter (requests), an exception handler (FastAPI). No
monkey-patching, no private APIs — an integration survives minor releases
of its host library.
- **One breaker per host.** HTTP integrations key breakers by request host:
a failing `api.a` never trips `api.b`. Breakers are created lazily in a
shared [`Registry`](../reference.md).
- **Safe rollout before enforcement.** Pass
`initial_state=State.METRICS_ONLY` to record real traffic without rejecting
it. The state is applied before each lazy breaker serves its first request;
deploy a new integration with `CLOSED` after tuning thresholds.
- **One classification model.** Responses are classified by an
`HttpStatusClassifier` — by default the canonical retryable set
(`429, 500, 502, 503, 504`) plus every non-excluded transport exception
counts as a failure, while `4xx` client mistakes do not. Exceptions the
*caller* caused say nothing about the dependency's health, so each
integration excludes the ones its own library raises inside the guarded
call: `UnsupportedProtocol` / `LocalProtocolError` for httpx2 and httpx,
`InvalidURL` for requests, nothing for aiohttp (it rejects malformed URLs
before the middleware chain runs). Pass
`HttpStatusClassifier(failure_statuses={...}, excluded_exceptions=(...))` or
your own `FailureClassifier` to change the policy.
- **One rejection signal, in two dialects.** An open circuit always raises
[`CircuitOpenError`](../reference.md) — carrying the breaker name, a
`retry_after` estimate and the last recorded failure — *before* a
connection is attempted. Each HTTP client integration raises a subclass that
is *also* a native error of that library, so an application's existing
degradation path catches it: `CircuitOpenTransportError` (an
`httpx.TransportError` for the httpx integration, an `httpx2.TransportError`
for the httpx2 one), `CircuitOpenClientError`
(an `aiohttp.ClientConnectionError`), `CircuitOpenRequestError`
(a `requests.exceptions.ConnectionError`). The host base is always the
broadest "the request never completed" type, never a leaf such as
`ConnectError` — leaves are what retry predicates key on, and retrying a
rejection only burns an attempt against a circuit that is still open.
- **Zero-dependency core.** Integrations live in `interlock.integrations.*`
as optional extras; `import interlock` itself never pulls anything beyond
the standard library.
- **Explicit ownership and teardown.** Transports and adapters close their
native connection resources together with their breaker registry. The
aiohttp middleware exposes `aclose()` because `ClientSession` does not own
middleware resources.
## Support tiers
- **Tier 1 — shipped code.** Modules under `interlock.integrations.*`,
covered by the test suite and CI against both the minimum supported and the
latest version of the host library. Semver applies.
- **Tier 2 — recipes.** Documented, runnable patterns (LLM SDKs,
Flask/Django) that need no dedicated glue code. They can graduate to Tier 1
when demand shows up.
Missing an integration — gRPC, SQLAlchemy, Kafka, Celery?
[Open an issue](https://github.com/bagowix/interlock/issues): the next wave
is prioritised by demand.
---
# FastAPI
The `interlock-cb[fastapi]` extra protects a route's outgoing dependency with a
shared `Registry` and turns a tripped breaker into a clean
`503 Service Unavailable` response with a `Retry-After` header.
=== "uv"
```bash
uv add 'interlock-cb[fastapi]'
```
=== "pip"
```bash
pip install 'interlock-cb[fastapi]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[fastapi]'
```
## Usage
Install the exception handler once, then inject a per-name breaker into any route
with `Depends`:
```python
from typing import Annotated
from fastapi import Depends, FastAPI
from interlock import CircuitBreaker, Registry
from interlock.integrations.fastapi import breaker_dependency, install_exception_handler
app = FastAPI()
registry = Registry()
install_exception_handler(app)
orders_db = breaker_dependency('orders-db', registry=registry)
@app.get('/orders')
async def orders(breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> list[dict]:
return await breaker.call(fetch_orders)
```
When `fetch_orders` fails often enough, the breaker opens. The next request is
rejected with `CircuitOpenError` *before* `fetch_orders` runs, and the installed
handler converts it into:
```http
HTTP/1.1 503 Service Unavailable
Retry-After: 30
Content-Type: application/json
{"detail": "Circuit 'orders-db' is open"}
```
## How it works
- **`breaker_dependency(name, *, registry)`** returns a FastAPI dependency that
yields the named breaker from the shared `Registry`. The breaker is created
lazily on first use and reused on every later request, so all requests to that
route share one breaker (and one view of the dependency's health).
- **`install_exception_handler(app)`** registers a handler for
`CircuitOpenError`. It responds `503` and sets `Retry-After` to the breaker's
`retry_after` estimate, rounded up to whole seconds (per RFC 7231). The header
is omitted when there is no estimate (for example after `force_open()`).
You protect the *outgoing* call (`breaker.call(...)`) rather than the route
itself: only the dependency you wrap counts toward the breaker, and the breaker's
own admission logic (probes, half-open) keeps working.
## Sharing breakers across routes
Reuse the same `name` (and the same `registry`) to share one breaker across
several routes that all depend on the same downstream:
```python
orders_db = breaker_dependency('orders-db', registry=registry)
@app.get('/orders')
async def list_orders(breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> list[dict]:
return await breaker.call(fetch_orders)
@app.get('/orders/{order_id}')
async def get_order(order_id: int, breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> dict:
return await breaker.call(fetch_order, order_id)
```
Pass `config`, `clock`, `classifier` or `listener` to the `Registry` to tune
every breaker it creates, or override per name via `registry.get(name, config=...)`.
## Custom responses
For a different response shape, register your own handler instead of
`install_exception_handler`:
```python
from fastapi import Request, Response
from interlock import CircuitOpenError
@app.exception_handler(CircuitOpenError)
async def on_open(request: Request, exc: CircuitOpenError) -> Response: ...
```
---
# Litestar
The `interlock-cb[litestar]` extra (Litestar ≥ 2.23) protects a route's
outgoing dependency with a shared `Registry` and turns a tripped breaker into
a clean `503 Service Unavailable` response with a `Retry-After` header.
=== "uv"
```bash
uv add 'interlock-cb[litestar]'
```
=== "pip"
```bash
pip install 'interlock-cb[litestar]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[litestar]'
```
## Usage
Litestar wires exception handlers and dependencies at construction time —
declare both on the app (or a router / controller) and annotate the handler
parameter with `NamedDependency`:
```python
from litestar import Litestar, get
from litestar.di import NamedDependency
from interlock import CircuitBreaker, CircuitOpenError, Registry
from interlock.integrations.litestar import breaker_dependency, circuit_open_handler
registry = Registry()
@get('/orders')
async def orders(breaker: NamedDependency[CircuitBreaker]) -> list[dict]:
return await breaker.call(fetch_orders)
app = Litestar(
route_handlers=[orders],
dependencies={'breaker': breaker_dependency('orders-db', registry=registry)},
exception_handlers={CircuitOpenError: circuit_open_handler},
)
```
When `fetch_orders` fails often enough, the breaker opens. The next request is
rejected with `CircuitOpenError` *before* `fetch_orders` runs, and the handler
converts it into:
```http
HTTP/1.1 503 Service Unavailable
Retry-After: 30
Content-Type: application/json
{"detail": "Circuit 'orders-db' is open"}
```
## How it works
- **`breaker_dependency(name, *, registry)`** returns a Litestar
[`Provide`](https://docs.litestar.dev/latest/usage/dependency-injection.html)
that yields the named breaker from the shared `Registry`. The breaker is
created lazily on first use and reused on every later request, so all
requests sharing the dependency share one breaker (and one view of the
downstream's health). Declare it at any layer — app, router, controller or
handler.
- **`circuit_open_handler`** maps `CircuitOpenError` to `503` and sets
`Retry-After` to the breaker's `retry_after` estimate, rounded up to whole
seconds (per RFC 7231). The header is omitted when there is no estimate
(for example after `force_open()`).
You protect the *outgoing* call (`breaker.call(...)`) rather than the route
itself: only the dependency you wrap counts toward the breaker, and the
breaker's own admission logic (probes, half-open) keeps working.
## Sharing breakers across routes
Reuse the same `name` (and the same `registry`) wherever routes depend on the
same downstream — one `dependencies={...}` declaration on the app covers them
all, and every route sees the same circuit state.
Pass `config`, `clock`, `classifier` or `listener` to the `Registry` to tune
every breaker it creates, or override per name via
`registry.get(name, config=...)`.
## Custom responses
For a different response shape, register your own handler instead:
```python
from litestar import Request, Response
from interlock import CircuitOpenError
def on_open(request: Request, exc: CircuitOpenError) -> Response[dict[str, str]]: ...
app = Litestar(..., exception_handlers={CircuitOpenError: on_open})
```
---
# httpx2
The `interlock-cb[httpx2]` extra wraps an [httpx2](https://pypi.org/project/httpx2/)
transport so a circuit breaker is applied **per host** transparently — no
decorators or `call` wrappers in your request code.
`interlock-cb` is listed in httpx2's official
[third-party packages directory](https://github.com/pydantic/httpx2/blob/main/docs/third_party_packages.md#interlock-cb).
=== "uv"
```bash
uv add 'interlock-cb[httpx2]'
```
=== "pip"
```bash
pip install 'interlock-cb[httpx2]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[httpx2]'
```
## Synchronous client
```python
import httpx2
from interlock.integrations.httpx2 import CircuitBreakerTransport
transport = CircuitBreakerTransport(httpx2.HTTPTransport())
client = httpx2.Client(transport=transport)
response = client.get('https://api.example.com/v1/users')
```
## Asynchronous client
```python
import httpx2
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport())
client = httpx2.AsyncClient(transport=transport)
response = await client.get('https://api.example.com/v1/users')
```
Use the client as a context manager. Context entry and exit are delegated to
the wrapped transport, including for custom transports that acquire resources
in `__enter__` or `__aenter__`. Closing the client closes both the wrapped
connection pool and every breaker created by the transport.
## Safe production rollout
Start in shadow mode when introducing the integration to existing traffic:
```python
import httpx2
from interlock import LoggingEventListener, State
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(
httpx2.AsyncHTTPTransport(),
initial_state=State.METRICS_ONLY,
listener=LoggingEventListener(),
)
```
Every host created later starts in `METRICS_ONLY` before its first request: it
records outcomes but never raises `CircuitOpenError`. `LoggingEventListener`
writes every event through stdlib logging; swap it for an `EventListener` that
exports to your metrics backend. For local diagnosis,
`transport.registry.get_existing(host)` returns an existing breaker without
creating one; inspect its `state` and `snapshot()`. Hosts are only known at
runtime, so `transport.registry.items()` lists the breakers created so far and
`names()` just the names they were created under. Both are point-in-time copies:
a breaker created afterwards is not in them.
After tuning thresholds, deploy a new transport with the default
`initial_state=State.CLOSED`. Do not reset only the currently known hosts: a
transport configured for shadow mode would still create future hosts in
`METRICS_ONLY`. See the complete [safe-rollout guide](../guides/states.md#safe-rollout).
## Per-host isolation
Each host gets its own breaker, created lazily and cached. A failing
`api.a.example.com` trips only its own breaker; requests to
`api.b.example.com` are unaffected. Per-instance, per-host state is usually more
correct than global state — each host's health is observed independently.
When a host's breaker is open, its requests raise `CircuitOpenTransportError`
before reaching the network.
## What a rejection looks like
An open circuit rejects the request with `CircuitOpenTransportError`, which is
both an `httpx2.TransportError` and interlock's `CircuitOpenError`:
```python
import httpx2
from interlock.integrations.httpx2 import CircuitOpenTransportError
try:
response = client.get('https://api.example.com/v1/users')
except httpx2.TransportError as exc:
# The dependency being unreachable and the breaker rejecting both land here.
if isinstance(exc, CircuitOpenTransportError):
... # rejected before any I/O; the next probe is exc.retry_after away
raise
```
That is the point of the type: the degradation paths an application already
writes in httpx2's own idiom keep working the day a breaker leaves shadow mode.
The rejection is still a `CircuitOpenError` too, so `except CircuitOpenError`,
a `FallbackStrategy(on=(CircuitOpenError,))` or a framework exception handler
registered for it catch exactly what they caught before. It carries
`breaker_name`, `retry_after` and `last_failure`, plus httpx2's own `.request`.
The base type is deliberately `TransportError` and never a leaf such as
`ConnectError` or `TimeoutException`: nothing was connected and nothing timed
out, and those leaves are exactly what retry predicates key on — a retried
rejection burns an attempt against a circuit that is still open.
Two further types cover the other interlock errors that can reach the caller
through the transport, raised when a layer of your own inside the wrapped
transport (a pipeline timeout, a bulkhead) fails the request:
| interlock error | dialect type | httpx2 base |
|---|---|---|
| `CircuitOpenError` | `CircuitOpenTransportError` | `httpx2.TransportError` |
| `CallTimeoutError` | `CallTimeoutTransportError` | `httpx2.TimeoutException` |
| `BulkheadFullError` | `BulkheadFullTransportError` | `httpx2.PoolTimeout` |
Those two describe transient *local* conditions — a deadline, a busy slot pool
— so unlike a rejection they sit under httpx2's timeout types on purpose, where
retry predicates do fire on them. An error raised by the wrapped transport
itself is never retyped, and neither is one that already carries an httpx2
hierarchy.
## Custom breaker keys
Pass `name_resolver` when the request host is not the logical dependency
identity. The callback receives the native `httpx2.Request` and returns the
breaker name:
```python
import httpx2
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(
httpx2.AsyncHTTPTransport(),
name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```
Returning the same name for several discovery hosts gives them one breaker;
deriving a name from the path can split independent upstreams behind one
gateway host. The result must be a non-empty string containing something other
than whitespace. Invalid results raise `ValueError` with the request URL before
the wrapped transport performs I/O.
The resolved name is used consistently as the registry key, in
`CircuitOpenError`, and in every listener event. Resolve the name here instead
of rewriting listener labels so metrics always identify the breaker whose
state they report. Both synchronous and asynchronous transports accept the
option.
## Share one registry across clients
Pass a caller-owned `Registry` when several clients reach the same dependency.
Requests resolving to the same name then use one breaker instance and one
sliding window, even when they travel through different transports:
```python
import httpx2
from interlock import Config, Registry
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport, HttpStatusClassifier
registry = Registry(
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
classifier=HttpStatusClassifier(),
)
client_a = httpx2.AsyncClient(
transport=AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport(), registry=registry)
)
client_b = httpx2.AsyncClient(
transport=AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport(), registry=registry)
)
```
`Registry` uses exception-only classification by default. Always configure
`HttpStatusClassifier` as above when you want the integration's normal HTTP
status policy; otherwise a returned `503` counts as a success. The supplied
registry owns `config`, `clock`, `initial_state`, `classifier`, and `listener`,
so none of those options may also be passed to the transport.
Share such a registry with HTTP clients only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.
Closing a client automatically closes its breakers only when the transport
owns the registry. An injected registry remains open while the wrapped
connection pool closes; the application must explicitly call
`await registry.aclose_all()` during async shutdown (or `registry.close_all()`
when all guarded clients are synchronous).
## Reach the wrapped transport
`transport.wrapped` returns the transport being guarded, so a composed object
can be unwrapped without touching private attributes — verifying the pool
limits, TLS context or proxy the inner transport was built with, inspecting it
in a REPL, or walking a chain of wrappers:
```python
import httpx2
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport
inner = httpx2.AsyncHTTPTransport(limits=httpx2.Limits(max_connections=20))
transport = AsyncCircuitBreakerTransport(inner)
assert transport.wrapped is inner
```
The property is read-only: the wrapped transport is fixed at construction. Both
the synchronous and asynchronous classes expose it.
## What counts as a failure
By default the transport uses `HttpStatusClassifier`:
- a transport exception (connect/read errors) → failure;
- a response with status `429, 500, 502, 503, 504` → failure;
- everything else, including `4xx` client errors like `404`, → success;
- `UnsupportedProtocol` and `LocalProtocolError` → success.
This mirrors the retryable set used by urllib3, AWS and Google clients.
Permanent `5xx` (`501`, `505`) are deliberately excluded — retrying or tripping
the breaker cannot fix a contract or protocol error.
The two excluded exceptions are the caller's own bug — a scheme-less or
unsupported URL, and the local side violating HTTP. They are deterministic and
say nothing about the dependency, so a burst of them must not open the circuit
of a healthy host; they still propagate to the caller unchanged.
`PoolTimeout` is *not* excluded: an exhausted pool is usually the dependency
holding connections open, and shedding load then is the point. Exclude it
explicitly when your pool is sized below your own burst:
```python
import httpx2
from interlock.integrations.httpx2 import HttpStatusClassifier
classifier = HttpStatusClassifier(
excluded_exceptions=(
httpx2.LocalProtocolError,
httpx2.UnsupportedProtocol,
httpx2.PoolTimeout,
),
)
```
`excluded_exceptions` replaces the default set — pass `()` to count every
exception as a failure. An excluded exception is recorded as a *success*: the
sliding window has no third outcome.
## Tuning
Pass any of `config`, `clock`, `classifier`, `listener` to the transport; they
flow to every breaker:
```python
from interlock import Config, LoggingEventListener
from interlock.integrations.httpx2 import CircuitBreakerTransport
transport = CircuitBreakerTransport(
httpx2.HTTPTransport(),
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
listener=LoggingEventListener(),
)
```
Supply your own `classifier` to change the failure policy — for example to also
fail on `408 Request Timeout`.
---
# httpx
The `interlock-cb[httpx]` extra wraps an
[httpx](https://www.python-httpx.org/) transport so a circuit breaker is
applied **per host** transparently. It supports httpx 0.27.0 and newer.
=== "uv"
```bash
uv add 'interlock-cb[httpx]'
```
=== "pip"
```bash
pip install 'interlock-cb[httpx]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[httpx]'
```
## Synchronous client
```python
import httpx
from interlock.integrations.httpx import CircuitBreakerTransport
transport = CircuitBreakerTransport(httpx.HTTPTransport())
client = httpx.Client(transport=transport)
response = client.get('https://api.example.com/v1/users')
```
## Asynchronous client
```python
import httpx
from interlock.integrations.httpx import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport())
client = httpx.AsyncClient(transport=transport)
response = await client.get('https://api.example.com/v1/users')
```
Use the client as a context manager. Closing it delegates `close()` or
`aclose()` to the wrapped transport and releases both the connection pool and
every breaker created by the transport.
## Safe production rollout
Start in shadow mode when introducing the integration to existing traffic:
```python
import httpx
from interlock import LoggingEventListener, State
from interlock.integrations.httpx import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(
httpx.AsyncHTTPTransport(),
initial_state=State.METRICS_ONLY,
listener=LoggingEventListener(),
)
```
Every host created later starts in `METRICS_ONLY` before its first request: it
records outcomes but never raises `CircuitOpenError`. `LoggingEventListener`
writes every event through stdlib logging; swap it for an `EventListener` that
exports to your metrics backend. For local diagnosis,
`transport.registry.get_existing(host)` returns an existing breaker without
creating one; inspect its `state` and `snapshot()`. Hosts are only known at
runtime, so `transport.registry.items()` lists the breakers created so far and
`names()` just the names they were created under. Both are point-in-time copies:
a breaker created afterwards is not in them.
After tuning thresholds, deploy a new transport with the default
`initial_state=State.CLOSED`. Do not reset only the currently known hosts: a
transport configured for shadow mode would still create future hosts in
`METRICS_ONLY`. See the complete [safe-rollout guide](../guides/states.md#safe-rollout).
## Per-host isolation
Each host gets its own lazily created breaker. A failing
`api.a.example.com` trips only that host; traffic to `api.b.example.com`
continues normally. An open breaker raises `CircuitOpenTransportError`
before the wrapped transport performs I/O. A request URL without a host raises
`ValueError` for the same reason: there is no dependency identity to key on.
## What a rejection looks like
An open circuit rejects the request with `CircuitOpenTransportError`, which is
both an `httpx.TransportError` and interlock's `CircuitOpenError`:
```python
import httpx
from interlock.integrations.httpx import CircuitOpenTransportError
try:
response = client.get('https://api.example.com/v1/users')
except httpx.TransportError as exc:
# The dependency being unreachable and the breaker rejecting both land here.
if isinstance(exc, CircuitOpenTransportError):
... # rejected before any I/O; the next probe is exc.retry_after away
raise
```
That is the point of the type: the degradation paths an application already
writes in httpx's own idiom keep working the day a breaker leaves shadow mode.
The rejection is still a `CircuitOpenError` too, so `except CircuitOpenError`,
a `FallbackStrategy(on=(CircuitOpenError,))` or a framework exception handler
registered for it catch exactly what they caught before. It carries
`breaker_name`, `retry_after` and `last_failure`, plus httpx's own `.request`.
The base type is deliberately `TransportError` and never a leaf such as
`ConnectError` or `TimeoutException`: nothing was connected and nothing timed
out, and those leaves are exactly what retry predicates key on — a retried
rejection burns an attempt against a circuit that is still open.
Two further types cover the other interlock errors that can reach the caller
through the transport, raised when a layer of your own inside the wrapped
transport (a pipeline timeout, a bulkhead) fails the request:
| interlock error | dialect type | httpx base |
|---|---|---|
| `CircuitOpenError` | `CircuitOpenTransportError` | `httpx.TransportError` |
| `CallTimeoutError` | `CallTimeoutTransportError` | `httpx.TimeoutException` |
| `BulkheadFullError` | `BulkheadFullTransportError` | `httpx.PoolTimeout` |
Those two describe transient *local* conditions — a deadline, a busy slot pool
— so unlike a rejection they sit under httpx's timeout types on purpose, where
retry predicates do fire on them. An error raised by the wrapped transport
itself is never retyped, and neither is one that already carries an httpx
hierarchy.
## Custom breaker keys
Pass `name_resolver` when the request host is transport plumbing rather than
the logical dependency identity. The callback receives the native
`httpx.Request` and returns the breaker name:
```python
import httpx
from interlock.integrations.httpx import AsyncCircuitBreakerTransport
transport = AsyncCircuitBreakerTransport(
httpx.AsyncHTTPTransport(),
name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```
The same callback can split one gateway host into independent breakers, for
example by returning a name derived from the first path segment. It must return
a non-empty string containing something other than whitespace; invalid results
raise `ValueError` with the request URL before the wrapped transport performs
I/O.
The resolved name is the registry key and the name carried by
`CircuitOpenError` and every listener event. Use the resolver, rather than
rewriting labels in a listener, so breaker state and observability labels stay
aligned. Both synchronous and asynchronous transports accept the option.
## Share one registry across clients
Inject one caller-owned `Registry` when several clients should observe the
same dependency health. The transports then resolve the same name to the same
breaker and contribute to one sliding window:
```python
import httpx
from interlock import Config, Registry
from interlock.integrations.httpx import AsyncCircuitBreakerTransport, HttpStatusClassifier
registry = Registry(
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
classifier=HttpStatusClassifier(),
)
client_a = httpx.AsyncClient(
transport=AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport(), registry=registry)
)
client_b = httpx.AsyncClient(
transport=AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport(), registry=registry)
)
```
The classifier is intentional: a bare `Registry` classifies raised exceptions
but treats returned responses, including `503`, as successes. Configure
`HttpStatusClassifier` to retain the transport's default status policy. A
supplied registry also owns `config`, `clock`, `initial_state`, `classifier`,
and `listener`; combining `registry` with any of those transport options raises
`ValueError` instead of silently ignoring one source of configuration.
Share such a registry with HTTP clients only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.
Closing a client automatically closes its breakers only when the transport
owns the registry. An injected registry remains open while the wrapped
connection pool closes; the application must explicitly call
`await registry.aclose_all()` during async shutdown, or `registry.close_all()`
when every guarded client is synchronous.
## Reach the wrapped transport
`transport.wrapped` returns the transport being guarded, so a composed object
can be unwrapped without touching private attributes — verifying the pool
limits, TLS context or proxy the inner transport was built with, inspecting it
in a REPL, or walking a chain of wrappers:
```python
import httpx
from interlock.integrations.httpx import AsyncCircuitBreakerTransport
inner = httpx.AsyncHTTPTransport(limits=httpx.Limits(max_connections=20))
transport = AsyncCircuitBreakerTransport(inner)
assert transport.wrapped is inner
```
The property is read-only: the wrapped transport is fixed at construction. Both
the synchronous and asynchronous classes expose it.
## What counts as a failure
The default `HttpStatusClassifier` counts these as failures:
- transport exceptions raised before a response is returned;
- response statuses `429, 500, 502, 503, 504`.
Other responses, including caller errors such as `404`, count as successes.
So are the transport exceptions httpx raises for the *caller's* own bug —
`UnsupportedProtocol` (a scheme-less or unsupported URL) and
`LocalProtocolError` (the local side violating HTTP). They are deterministic
and say nothing about the dependency, so a burst of them must not open the
circuit of a healthy host. They still propagate to the caller unchanged.
`PoolTimeout` is *not* excluded: an exhausted pool is usually the dependency
holding connections open, and shedding load then is the point. Exclude it
explicitly when your pool is sized below your own burst:
```python
import httpx
from interlock.integrations.httpx import HttpStatusClassifier
classifier = HttpStatusClassifier(
excluded_exceptions=(httpx.LocalProtocolError, httpx.UnsupportedProtocol, httpx.PoolTimeout),
)
```
`excluded_exceptions` replaces the default set — pass `()` to count every
exception as a failure. An excluded exception is recorded as a *success*: the
sliding window has no third outcome.
Pass `HttpStatusClassifier(failure_statuses={...})` or another
`FailureClassifier` to change the status side of the policy.
## Streaming responses
The wrapper returns the original `httpx.Response` unchanged, so sync and async
streaming remain lazy and connection cleanup keeps httpx's normal semantics.
Context entry and exit are delegated to the wrapped transport, including for
custom transports that acquire resources in `__enter__` or `__aenter__`.
Because the circuit-breaker call completes when response headers arrive, an
exception raised later while consuming a streaming body is outside that call
and is not recorded by the breaker.
## Tuning
`config`, `clock`, `initial_state`, `classifier`, and `listener` are shared by
every breaker created by the transport:
```python
import httpx
from interlock import Config, LoggingEventListener
from interlock.integrations.httpx import CircuitBreakerTransport
transport = CircuitBreakerTransport(
httpx.HTTPTransport(),
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
listener=LoggingEventListener(),
)
```
The transport never retries. If another layer owns retries, keep them bounded
and stop retrying when the breaker rejects: `CircuitOpenTransportError` sits
outside the httpx leaf types retry predicates key on, so a predicate written
against `ConnectError` or `TimeoutException` already leaves it alone.
---
# aiohttp
The `interlock-cb[aiohttp]` extra guards every request a `ClientSession`
sends with a circuit breaker **per host**, wired in as a client middleware —
no decorators in call sites. Requires aiohttp ≥ 3.12 (client middlewares).
=== "uv"
```bash
uv add 'interlock-cb[aiohttp]'
```
=== "pip"
```bash
pip install 'interlock-cb[aiohttp]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[aiohttp]'
```
## Usage
```python
import aiohttp
from interlock.integrations.aiohttp import CircuitBreakerMiddleware
middleware = CircuitBreakerMiddleware()
async with aiohttp.ClientSession(middlewares=(middleware,)) as session:
async with session.get('https://api.example.com/orders') as response:
orders = await response.json()
await middleware.aclose()
```
`ClientSession` does not own middleware resources. Call `middleware.aclose()`
during application shutdown; it releases every per-host breaker and is
idempotent.
Each host gets its own breaker (a failing `api.a` never trips `api.b`),
created lazily and shared across requests. When a host's circuit is open the
request raises `CircuitOpenClientError` *before* a connection is made.
The breaker observes the time to *response headers*; reading the body happens
outside the guarded call — the same semantics as the
[httpx2 transport](httpx2.md).
## What a rejection looks like
An open circuit rejects the request with `CircuitOpenClientError`, which is
both an `aiohttp.ClientConnectionError` and interlock's `CircuitOpenError`:
```python
import aiohttp
from interlock.integrations.aiohttp import CircuitOpenClientError
try:
response = await session.get('https://api.example.com/v1/users')
except aiohttp.ClientError as exc:
# The dependency being unreachable and the breaker rejecting both land here.
if isinstance(exc, CircuitOpenClientError):
... # rejected before any I/O; the next probe is exc.retry_after away
raise
```
That is the point of the type: the degradation paths an application already
writes in aiohttp's own idiom keep working the day a breaker leaves shadow
mode. The rejection is still a `CircuitOpenError` too, so `except
CircuitOpenError`, a `FallbackStrategy(on=(CircuitOpenError,))` or a framework
exception handler registered for it catch exactly what they caught before. It
carries `breaker_name`, `retry_after` and `last_failure`. aiohttp re-raises a
`ClientError` from the middleware chain untouched, so it reaches the caller
exactly as raised.
The base type is deliberately the broad `ClientConnectionError` and never a
leaf such as `ClientOSError` or `ServerDisconnectedError`: nothing was
connected and no server dropped anything, and those leaves are exactly what
retry predicates key on — a retried rejection burns an attempt against a
circuit that is still open. It also stays outside `ClientResponseError`, which
would claim a response that never arrived.
Only the rejection is retyped. The httpx transports also pair `CallTimeoutError`
with `httpx.TimeoutException` and `BulkheadFullError` with `httpx.PoolTimeout`;
aiohttp has no honest counterpart for "no local slot was free" — the nearest type
is `aiohttp.ClientConnectionError`, which the rejection already uses, so the pairing would claim a
distinction it cannot express. A `CallTimeoutError` raised by a pipeline of your
own inside the guarded call therefore stays an interlock error on its way out.
## Custom breaker keys
Pass `name_resolver` when host-based isolation does not match the logical
dependencies. The callback receives the native `aiohttp.ClientRequest` and
returns the breaker name:
```python
from interlock.integrations.aiohttp import CircuitBreakerMiddleware
middleware = CircuitBreakerMiddleware(
name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```
A resolver can collapse several discovery hosts onto one breaker or derive a
name from the request path to separate upstreams behind a shared gateway. It
must return a non-empty string containing something other than whitespace;
invalid results raise `ValueError` with the request URL before the handler
performs I/O.
The resolved name is used by the registry, `CircuitOpenError`, and every
listener event. Resolve it in the middleware rather than rewriting listener
labels so observed names always match the breaker whose state they describe.
## Share one registry across sessions
Several middleware instances can share one caller-owned registry, so traffic
resolving to the same name contributes to one breaker and one sliding window:
```python
from interlock import Config, Registry
from interlock.integrations.aiohttp import CircuitBreakerMiddleware, HttpStatusClassifier
registry = Registry(
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
classifier=HttpStatusClassifier(),
)
middleware_a = CircuitBreakerMiddleware(registry=registry)
middleware_b = CircuitBreakerMiddleware(registry=registry)
```
Do not omit the classifier when HTTP statuses should affect the breaker. A
bare `Registry` uses exception-only classification, so a returned `503` counts
as a success. The registry owns `config`, `clock`, `initial_state`,
`classifier`, and `listener`; passing any of them to the middleware together
with `registry` raises `ValueError`.
Share such a registry with HTTP sessions only. `HttpStatusClassifier` reads
`.status` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.
Calling `aclose()` automatically closes the breakers only when the middleware
owns the registry. An injected registry remains open until the application
explicitly calls `await registry.aclose_all()` during shutdown.
## Safe production rollout
Pass `initial_state=State.METRICS_ONLY` to record real outcomes without
rejecting requests. The state is applied to every lazily created host before
its first request:
```python
from interlock import LoggingEventListener, State
from interlock.integrations.aiohttp import CircuitBreakerMiddleware
middleware = CircuitBreakerMiddleware(
initial_state=State.METRICS_ONLY,
listener=LoggingEventListener(),
)
```
The public `middleware.registry` supports local diagnosis with
`get_existing(host)`, `state` and `snapshot()`, plus `names()` and `items()` for
the breakers created so far — point-in-time copies, without the ones created
afterwards. `LoggingEventListener` writes every event through stdlib logging;
swap it for an `EventListener` that exports to your metrics backend, then
deploy a new middleware with the default `CLOSED` state to begin enforcement.
See [Safe rollout](../guides/states.md#safe-rollout).
## Failure policy
By default a response counts as a failure when its status is in the canonical
retryable set (`429, 500, 502, 503, 504`) and any exception raised while
sending (connect/read errors) is a failure; `4xx` client mistakes like `404`
are successes.
Nothing is excluded by default: aiohttp rejects a malformed or non-HTTP URL
before the middleware chain runs, so no caller-side error of its own reaches
the classifier. Middlewares of your own that sit inside this one are the
exception — an auth middleware refusing to sign a request is your bug, not the
dependency's, so exclude what it raises with
`excluded_exceptions=(MissingCredentials,)`. An excluded exception is recorded
as a *success*, since the sliding window has no third outcome, and still
propagates to the caller.
Change the statuses, or the whole policy:
```python
from interlock import Config
from interlock.integrations.aiohttp import CircuitBreakerMiddleware, HttpStatusClassifier
middleware = CircuitBreakerMiddleware(
config=Config(failure_rate_threshold=0.3),
classifier=HttpStatusClassifier(failure_statuses={408, 429, 500, 502, 503, 504}),
)
```
Any custom `FailureClassifier` works too — see
[Failure classification](../guides/failure-classification.md).
## Tuning and observability
The middleware accepts the same collaborators as `CircuitBreaker` — `config`,
`clock`, `initial_state`, `classifier`, `listener`. One middleware instance
holds one registry of resolved breakers; reuse the instance across sessions to
share breaker state, or create separate instances to isolate them. For
application-level retries combine with the [tenacity integration](tenacity.md)
and read [Retries and circuit breakers](../guides/retries.md) first.
---
# requests
The `interlock-cb[requests]` extra guards every request a `Session` sends
with a circuit breaker **per host** — mounted once, no decorators in call
sites.
=== "uv"
```bash
uv add 'interlock-cb[requests]'
```
=== "pip"
```bash
pip install 'interlock-cb[requests]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[requests]'
```
## Usage
`CircuitBreakerAdapter` subclasses `requests.adapters.HTTPAdapter` — the
library's native transport extension point — so it mounts like any adapter:
```python
import requests
from interlock.integrations.requests import CircuitBreakerAdapter
session = requests.Session()
adapter = CircuitBreakerAdapter()
session.mount('https://', adapter)
session.mount('http://', adapter)
response = session.get('https://api.example.com/orders')
```
Closing the session closes the adapter's connection pools and every breaker it
created.
Each host gets its own breaker (a failing `api.a` never trips `api.b`),
created lazily and shared across requests. When a host's circuit is open the
request raises `CircuitOpenRequestError` *before* a connection is made.
## What a rejection looks like
An open circuit rejects the request with `CircuitOpenRequestError`, which is
both a `requests.exceptions.ConnectionError` and interlock's
`CircuitOpenError`:
```python
import requests
from interlock.integrations.requests import CircuitOpenRequestError
try:
response = session.get('https://api.example.com/v1/users')
except requests.exceptions.RequestException as exc:
# The dependency being unreachable and the breaker rejecting both land here.
if isinstance(exc, CircuitOpenRequestError):
... # rejected before any I/O; the next probe is exc.retry_after away
raise
```
That is the point of the type: the degradation paths an application already
writes in requests' own idiom keep working the day a breaker leaves shadow
mode. The rejection is still a `CircuitOpenError` too, so `except
CircuitOpenError`, a `FallbackStrategy(on=(CircuitOpenError,))` or a framework
exception handler registered for it catch exactly what they caught before. It
carries `breaker_name`, `retry_after` and `last_failure`, plus requests' own
`.request` and `.response`.
The base type is deliberately `ConnectionError` and never a leaf such as
`SSLError` or `ConnectTimeout`: no connection was attempted and nothing timed
out, and those leaves are exactly what retry predicates key on — a retried
rejection burns an attempt against a circuit that is still open. urllib3's own
`Retry` never sees it either: that runs inside `HTTPAdapter.send`, which the
rejection replaces rather than enters.
Only the rejection is retyped. The httpx transports also pair `CallTimeoutError`
with `httpx.TimeoutException` and `BulkheadFullError` with `httpx.PoolTimeout`;
requests has no honest counterpart for "no local slot was free" — the nearest type
is `requests.exceptions.ConnectionError`, which the rejection already uses, so the pairing would claim a
distinction it cannot express. A `CallTimeoutError` raised by a pipeline of your
own inside the guarded call therefore stays an interlock error on its way out.
## Custom breaker keys
Pass `name_resolver` when the request host is not the logical dependency
identity. The callback receives the native `requests.PreparedRequest` and
returns the breaker name. For example, the first path segment can separate
independent upstreams behind one gateway host:
```python
from interlock.integrations.requests import CircuitBreakerAdapter
adapter = CircuitBreakerAdapter(
name_resolver=lambda request: request.path_url.split('/')[1],
)
```
Returning one name for several discovery hosts instead makes them share a
breaker. The result must be a non-empty string containing something other than
whitespace; invalid results raise `ValueError` with the request URL before the
adapter performs I/O.
The resolved name is used consistently as the registry key, in
`CircuitOpenError`, and in every listener event. Resolve the identity here
instead of rewriting listener labels so metrics remain aligned with breaker
state.
## Share one registry across sessions
Inject a caller-owned `Registry` when independent sessions should use one
breaker and one sliding window for the same resolved name:
```python
import requests
from interlock import Config, Registry
from interlock.integrations.requests import CircuitBreakerAdapter, HttpStatusClassifier
registry = Registry(
config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
classifier=HttpStatusClassifier(),
)
session_a = requests.Session()
session_a.mount('https://', CircuitBreakerAdapter(registry=registry))
session_b = requests.Session()
session_b.mount('https://', CircuitBreakerAdapter(registry=registry))
```
The explicit classifier preserves the adapter's normal status policy. Without
it, a bare `Registry` classifies exceptions only and a returned `503` counts
as a success. The registry owns `config`, `clock`, `initial_state`,
`classifier`, and `listener`; combining `registry` with any of those adapter
options raises `ValueError`.
Share such a registry with HTTP sessions only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.
Closing a session automatically closes its breakers only when the adapter owns
the registry. An injected registry remains open while its connection pools
close; the application must explicitly call `registry.close_all()` during
shutdown.
## Safe production rollout
Pass `initial_state=State.METRICS_ONLY` to record real outcomes without
rejecting requests. The state is applied to every lazily created host before
its first request:
```python
from interlock import LoggingEventListener, State
from interlock.integrations.requests import CircuitBreakerAdapter
adapter = CircuitBreakerAdapter(
initial_state=State.METRICS_ONLY,
listener=LoggingEventListener(),
)
```
The public `adapter.registry` supports local diagnosis with
`get_existing(host)`, `state` and `snapshot()`, plus `names()` and `items()` for
the breakers created so far — point-in-time copies, without the ones created
afterwards. `LoggingEventListener` writes every event through stdlib logging;
swap it for an `EventListener` that exports to your metrics backend, then
deploy a new adapter with the default `CLOSED` state to begin enforcement.
See [Safe rollout](../guides/states.md#safe-rollout).
## Failure policy
By default a response counts as a failure when its status is in the canonical
retryable set (`429, 500, 502, 503, 504`) and a transport exception
(connect/read errors) is a failure; `4xx` client mistakes like `404` are
successes.
`InvalidURL` — raised by the adapter when the request or proxy URL carries no
host, and the parent of `InvalidProxyURL` — is a success too. It is the
caller's own bug: deterministic, and no evidence about the dependency, so a
burst of them must not open the circuit of a healthy host. The exception still
propagates to the caller. Replace that set with `excluded_exceptions=(...)`,
or pass `()` to count every exception as a failure; an excluded exception is
recorded as a *success*, since the sliding window has no third outcome.
Change the statuses, or the whole policy:
```python
from interlock import Config
from interlock.integrations.requests import CircuitBreakerAdapter, HttpStatusClassifier
adapter = CircuitBreakerAdapter(
config=Config(failure_rate_threshold=0.3),
classifier=HttpStatusClassifier(failure_statuses={408, 429, 500, 502, 503, 504}),
)
```
Any custom `FailureClassifier` works too — see
[Failure classification](../guides/failure-classification.md).
## Tuning and observability
The adapter accepts the same collaborators as `CircuitBreaker` — `config`,
`clock`, `initial_state`, `classifier`, `listener` — and forwards everything
else
(`pool_connections`, `max_retries`, ...) to `HTTPAdapter`. Note that
`max_retries` is urllib3's connection-level retry; for application-level
retries combine with the [tenacity integration](tenacity.md) and read
[Retries and circuit breakers](../guides/retries.md) first.
---
# LLM SDKs (OpenAI, Anthropic) — recipe
LLM APIs fail in exactly the ways circuit breakers exist for: rate limits
(`429`), overloaded backends (`529`/`503`), long hangs. A breaker around your
LLM calls stops a degraded provider from stalling every request thread, and
bounded retries recover from blips without amplifying an outage.
You can protect an SDK at either its call boundary, which enables SDK-specific
classification and slow-call detection, or at the httpx transport,
which applies transparently to every request made by that client.
## Classify SDK errors
Both SDKs raise `APIStatusError` subclasses carrying `status_code`, plus
connection/timeout errors. Not every error should trip the circuit: an
invalid request (`400`) or a missing model (`404`) is your bug, not the
provider's outage.
```python
import anthropic
class LLMFailureClassifier:
"""Trip on provider-side trouble, not on caller mistakes."""
_FAILURE_STATUSES = frozenset({429, 500, 502, 503, 504, 529})
def is_failure(self, *, result: object, exception: Exception | None) -> bool:
if exception is None:
return False
if isinstance(exception, anthropic.APIStatusError):
return exception.status_code in self._FAILURE_STATUSES
return isinstance(exception, (anthropic.APIConnectionError, anthropic.APITimeoutError))
```
For OpenAI, swap the exception types (`openai.APIStatusError`,
`openai.APIConnectionError`, `openai.APITimeoutError`) — the shape is
identical.
## Guard the calls
```python
import anthropic
from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter
from interlock import CircuitBreaker, Config
from interlock.integrations.tenacity import retry_unless_open
client = anthropic.AsyncAnthropic()
breaker = CircuitBreaker(
name='anthropic',
config=Config(slow_call_duration_threshold=30.0),
classifier=LLMFailureClassifier(),
)
@breaker
async def complete(prompt: str) -> str:
message = await client.messages.create(
model='claude-sonnet-5',
max_tokens=1024,
messages=[{'role': 'user', 'content': prompt}],
)
return message.content[0].text
retrying = AsyncRetrying(
retry=retry_unless_open(
anthropic.APIStatusError,
anthropic.APIConnectionError,
anthropic.APITimeoutError,
),
wait=wait_exponential_jitter(initial=1.0, max=30.0),
stop=stop_after_attempt(4),
reraise=True,
)
answer = await retrying(complete, 'Summarise this document...')
```
What each layer contributes:
- **Slow-call detection** (`slow_call_duration_threshold`) counts calls
slower than 30s as failures — a provider that still answers but takes a
minute per completion trips the breaker too. No other signal catches this.
- **The breaker** stops sending after the failure rate crosses the threshold;
while open, callers get `CircuitOpenError` in microseconds instead of
hanging — fail over to a second provider or degrade gracefully.
- **`retry_unless_open`** retries provider blips with jittered backoff but
stops the moment the circuit opens. The SDK's own retries overlap here —
either set `max_retries=0` on the client and let tenacity own retries, or
keep the SDK's and drop the tenacity layer; running both multiplies
attempts.
## Multiple providers, one pattern
Give each provider its own breaker name (`anthropic`, `openai`, ...) via a
shared `Registry` and check `breaker.state` to route around an open provider.
The [states guide](../guides/states.md) covers manual failover controls.
## Transport-level protection
OpenAI and Anthropic clients accept an httpx client. Install
`interlock-cb[httpx]`, wrap the SDK's underlying transport, and every endpoint
on the provider host shares the same breaker:
```python
import httpx
from openai import DefaultHttpxClient, OpenAI
from interlock.integrations.httpx import CircuitBreakerTransport
transport = CircuitBreakerTransport(httpx.HTTPTransport())
with OpenAI(
http_client=DefaultHttpxClient(transport=transport),
max_retries=0,
) as client:
response = client.responses.create(model='gpt-5.5', input='Summarise this document...')
```
Use `AsyncCircuitBreakerTransport`, `httpx.AsyncHTTPTransport`, and the SDK's
async client for async applications. `max_retries=0` gives the breaker one
observable attempt per SDK call; if retries are required, keep one explicit,
bounded retry owner instead of stacking SDK and application retries.
The transport classifier sees HTTP statuses directly, so no SDK exception
classifier is needed. Choose the call-boundary recipe above when you also need
to classify SDK-specific exceptions, measure the complete SDK operation as a
slow call, or use a breaker name that is not derived from the request host.
---
# tenacity (retries)
interlock deliberately ships no retry engine of its own:
[tenacity](https://tenacity.readthedocs.io/) already does backoff, jitter,
stop conditions and predicates well. The `interlock-cb[tenacity]` extra adds
the glue where retry × breaker composition goes wrong in practice.
=== "uv"
```bash
uv add 'interlock-cb[tenacity]'
```
=== "pip"
```bash
pip install 'interlock-cb[tenacity]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[tenacity]'
```
Read [Retries and circuit breakers](../guides/retries.md) first if you are
deciding *how* to combine the two patterns; this page documents the helpers.
## Fail fast (recommended default)
`retry_unless_open(*transient)` retries the listed transient exceptions but
stops as soon as the breaker opens. `CircuitOpenError` is not transient: the
breaker rejects instantly, so backing off and retrying it only burns the
attempt budget without ever reaching the dependency.
```python
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter
from interlock import CircuitBreaker
from interlock.integrations.tenacity import retry_unless_open
breaker = CircuitBreaker(name='payments')
@breaker
def charge(amount: int) -> str:
return gateway.charge(amount)
retrying = Retrying(
retry=retry_unless_open(TimeoutError, ConnectionError),
wait=wait_exponential_jitter(),
stop=stop_after_attempt(5),
reraise=True,
)
result = retrying(charge, 100)
```
Called without arguments, `retry_unless_open()` retries any ordinary
`Exception` — still never `CircuitOpenError`.
## Patient mode (wait for the probe)
Background jobs often prefer waiting over failing. `wait_probe(fallback)` is
a wait strategy: when the last attempt was rejected with a `retry_after`
estimate, it sleeps *exactly* until the breaker allows the next probe (plus a
small jitter so concurrent waiters do not storm the single probe slot). Any
other outcome delegates to the `fallback` strategy.
```python
from tenacity import (
AsyncRetrying,
retry_if_exception_type,
stop_after_attempt,
wait_exponential_jitter,
)
from interlock import CircuitOpenError
from interlock.integrations.tenacity import wait_probe
retrying = AsyncRetrying(
retry=retry_if_exception_type((TimeoutError, CircuitOpenError)),
wait=wait_probe(wait_exponential_jitter()),
stop=stop_after_attempt(10),
reraise=True,
)
report = await retrying(nightly_export)
```
Note the retry predicate: patient mode deliberately *does* retry
`CircuitOpenError` — that is what makes `wait_probe` see the rejection and
wait the right amount. Keep a `stop` condition anyway; a dependency can stay
down longer than any job should wait.
`wait_probe(..., jitter=0.5)` widens the random extra wait (seconds) added on
top of `retry_after`; when the rejection carries no estimate (for example
after `force_open()`), the fallback strategy decides.
## `RetryStrategy` — the pipeline layer
The same policies package into a [pipeline](../guides/pipeline.md) strategy:
```python
from interlock import Pipeline
from interlock.integrations.tenacity import RetryStrategy
pipeline = (
Pipeline.builder()
.retry(attempts=4) # this step builds a RetryStrategy
.circuit_breaker(breaker)
.timeout(2.0)
.build()
)
```
Attempts are always capped and the original exception is re-raised when the
budget runs out — no `RetryError` wrapping. The default predicate is
`retry_unless_open()` (fail fast on an open circuit); pass
`wait=wait_probe(...)` and a predicate that includes `CircuitOpenError` for
the patient mode. `name=` and `listener=` make every retry visible through
the `on_retry` hook ([observability](../guides/observability.md)).
## Everything else is plain tenacity
`Retrying`, `AsyncRetrying`, the `@retry` decorator, stop and wait strategies
compose as usual — the helpers are ordinary tenacity predicates and wait
objects, so you can combine them with `|`, `retry_any`, `wait_chain` and
friends.
---
# Redis (shared state)
The `interlock-cb[redis]` extra coordinates breaker state across processes and
machines through Redis: when one instance trips, every instance backs off, and
recovery probes are budgeted globally instead of per process.
=== "uv"
```bash
uv add 'interlock-cb[redis]'
```
=== "pip"
```bash
pip install 'interlock-cb[redis]'
```
=== "poetry"
```bash
poetry add 'interlock-cb[redis]'
```
## When to share state — and when not to
Per-instance state is the default for a reason. A local breaker reacts only to
what *this* process observes, cannot be affected by another instance's problem,
and keeps working when Redis does not.
Reach for shared state when all of these hold:
- Many instances call the **same downstream**, and its failure affects all of
them equally (a shared database, a rate-limited third-party API).
- You want **coordinated back-off**: once the downstream is declared unhealthy,
no instance should keep hammering it just because its own window has not
filled yet.
- You want **bounded recovery probing**: N instances should send at most
`permitted_calls_in_half_open` probes *in total*, not each.
Stay local when instances see genuinely different views of the dependency
(per-AZ endpoints, canary deployments), or when one instance's network problems
must not silence the whole fleet. A shared OPEN gates traffic *everywhere* —
that is the point, and the risk. It is a trade-off you opt into, not a default.
## Usage
Pass a storage to the breaker (or to a `Registry`, which hands it to every
breaker it creates — each coordinates under its own name):
```python
import redis
from interlock import CircuitBreaker, Registry
from interlock.integrations.redis import RedisStorage
storage = RedisStorage(redis.Redis(host='redis.internal'))
breaker = CircuitBreaker(name='payments', storage=storage)
registry = Registry(storage=storage) # or share one storage across many breakers
```
Async services use the async client and storage:
```python
import redis.asyncio
from interlock import CircuitBreaker
from interlock.integrations.redis import AsyncRedisStorage
storage = AsyncRedisStorage(redis.asyncio.Redis(host='redis.internal'))
breaker = CircuitBreaker(name='payments', storage=storage)
```
A coordinated breaker matches its storage's runtime: a `RedisStorage` serves
only the sync API (`with`, sync `call`), an `AsyncRedisStorage` only the async
one (`async with`, async `call`); mixing the styles raises `InterlockError`
with a clear message. A breaker *without* a storage stays fully dual.
## How coordination works
The local state machine keeps owning the sliding window and trip detection;
Redis owns the shared OPEN/HALF_OPEN state and the global probe budget. All
state for one breaker lives in a single hash (`interlock:cb:` by
default), and every transition runs as a Lua script, so racing instances stay
consistent.
The protected path stays fast:
- **CLOSED / OPEN admission** reads a locally cached view of the shared state —
zero inline Redis calls. A background poller refreshes the cache every
`poll_interval` seconds, so a trip on one instance reaches the others within
roughly one interval.
- **HALF_OPEN admission** is the single inline Redis operation: an atomic probe
lease that decrements the shared budget, bounding probes across the fleet.
- **Writes** (propagating a local trip, tallying probe outcomes, the final
close-or-reopen decision) are fire-and-forget on a background worker; they
never block a protected call.
Time comparisons ("has `wait_duration_in_open` elapsed?") use the *Redis
server's* clock, since instance clocks are not comparable. After the last probe
of a round, the deciding instance applies the same thresholds as the local
state machine and writes the transition guarded by a version check, so a
delayed decision can never overwrite a newer state.
## Coordinated mode contract
`RedisStorage` implements `Storage` / `AsyncStorage`; a third-party backend can
too. Four behaviours are part of that contract, not implementation detail —
skipping any of them breaks the guarantees the coordinator relies on:
1. **Fencing is mandatory.** `trip_open` and `close` accept `expected_version`
and must apply the write only when the backend's current version matches
it, otherwise no-op and return the current state. The coordinator relies on
this for the probe-round decision: a delayed instance computing "probes
passed" off a stale view must lose to a state another instance already
wrote, never overwrite it.
2. **A leaked probe slot is bounded only by `state_ttl`.** `lease_probe`
decrements a shared budget and has no corresponding un-lease operation. A
`BaseException` — cancellation, process kill — between `lease_probe` and
`record_probe` never returns that slot; only the key's TTL expiry does.
With several instances interrupted mid-probe, `HALF_OPEN` can stall for up
to `state_ttl`. Size it with this in mind — it is not just an
abandoned-key cleanup knob.
3. **`record_probe` tallies only while `HALF_OPEN`.** Every coordinated write
is best-effort *and* bounded: dropped while degraded, dropped when the
write queue is full, superseded by a later decision, reconciled by the next
poll rather than retried.
4. **Teardown is explicit, not automatic.** `close()` / `aclose()` stop the
lane deterministically — see [Shutdown](#shutdown) below. A coordinated
breaker left to be garbage-collected can outlive its owning scope instead.
Also: a breaker's `name` becomes a storage key (`interlock:cb:` for
`RedisStorage`). Never build one from untrusted input — an attacker-controlled
name can collide with, and corrupt, unrelated breaker state.
### Checklist for a custom `Storage` implementation
- [ ] `trip_open` / `close` honor `expected_version`: apply only on a match,
otherwise no-op and return the current state.
- [ ] Every operation is atomic against concurrent callers (a Lua script, a
locked transaction, ...) — no read-modify-write races.
- [ ] `ttl` is refreshed on every write so an abandoned key self-expires.
- [ ] `lease_probe` grants only while `HALF_OPEN` and budget remains.
- [ ] `record_probe` tallies only while `HALF_OPEN`; a late or out-of-round
outcome is dropped, not applied.
- [ ] No method raises into the protected path — the engine treats any
exception as a signal to degrade to local state, so a backend that
raises for a routine outcome (e.g. a fencing miss) breaks the contract;
that case must return the current state instead.
- [ ] Time comparisons (has `wait_duration_in_open` elapsed?) use the
backend's own clock, not the caller's — instance clocks are not
comparable across a fleet.
## Manual controls
Manual controls are local to one process and take precedence over Redis while
they are active. `force_open()` rejects every local call; `disable()` and
`metrics_only()` admit local calls without consuming a Redis HALF_OPEN probe.
`reset()` clears the local override and local metrics, but does not reset Redis
for the fleet: the breaker immediately resumes its cached shared `OPEN` or
`HALF_OPEN` state.
## Degradation: Redis down ≠ breaker down
A storage error never reaches your calls. On the first failure the breaker
switches to its local state and keeps protecting the process on its own window;
pending shared writes are dropped, and Redis is left alone before the poller
tries again. That delay starts at `retry_backoff` seconds and, by default,
stays there for as long as the outage lasts — the same fixed cadence every
release before this one used. Set `retry_backoff_multiplier` above `1.0` to
grow it geometrically with each further consecutive failure (capped at
`retry_backoff_max`), plus `retry_jitter` to spread out the retries of a fleet
of instances recovering from the same outage instead of having them all probe
Redis in the same instant. On the first successful operation the shared view
becomes authoritative again — including adopting a shared OPEN that happened
while this instance was cut off — and the failure count resets, so the next
outage starts back at `retry_backoff`.
Both edges are observable through the listener:
```python
class StorageWatch:
def on_storage_degraded(
self, *, name: str, error: BaseException
) -> None: ... # alert: running on local state
def on_storage_recovered(self, *, name: str) -> None: ... # back to coordinated state
```
`LoggingEventListener` logs degradation at `WARNING` and recovery at `INFO`;
`OTelEventListener` counts both on `interlock.storage.events`. Listeners
written before these hooks existed keep working — the engine calls them only if
present.
## Backpressure: the write queue is bounded
Coordinated writes are fire-and-forget: a local trip and each probe outcome are
queued for the background lane instead of being written on the protected path.
That queue holds at most `write_queue_size` writes (128 by default).
It is not a per-call queue — a healthy lane keeps it near empty, because writes
happen per *transition*, not per call, and probes are capped by
`permitted_calls_in_half_open`. The bound only matters when the lane stops
draining at all: a Redis client blocking without a timeout, or an async lane
whose event loop is gone. Without it, that lane would grow the queue for as
long as the process lives.
When the queue is full the *arriving* write is dropped — never blocked, never
raised into the call that produced it — and reported:
```python
class StorageWatch:
def on_storage_write_dropped(self, *, name: str) -> None: ... # shared state falling behind
```
Nothing is retried: the shared state is reconciled by the next successful poll
and, failing that, by `state_ttl` expiring the key. Locally the breaker keeps
protecting the process on its own window exactly as it does while degraded.
The hook is the signal that a lane is wedged — treat a non-zero rate as an
alert, not a tuning hint. `LoggingEventListener` logs it at `WARNING` and
`OTelEventListener` counts it on `interlock.storage.events`.
## Shutdown
A coordinated breaker owns a background lane — a daemon thread for a sync
storage, an asyncio task for an async one. It polls the shared view and drains
fire-and-forget writes. Left alone, it ends only when the breaker is garbage
collected, which is why an async lane can outlive `asyncio.run()` and log
"task was destroyed but it is pending".
`close()` (or `aclose()` for an async storage) ends it deterministically:
```python
breaker = CircuitBreaker(name='payments', storage=storage)
try:
... # serve traffic
finally:
breaker.close() # drains queued writes, stops the lane, joins it
```
`Registry` does the whole set at once:
```python
registry = Registry(storage=storage)
...
registry.close_all() # or: await registry.aclose_all()
```
What it guarantees:
- **Queued writes are drained first.** Ops already on the queue run in order
before the lane exits, so a trip recorded just before shutdown still reaches
Redis. Writes that the degraded gate would drop are still dropped.
- **No waiting out `poll_interval`.** A parked lane is woken immediately.
- **The `auto_transition` timer is cancelled**, and no new one is armed.
- **It is idempotent** and safe to call from any thread.
Two consequences worth knowing:
- **Shutdown is terminal.** The lane never restarts. Afterwards the breaker
keeps protecting calls on its local state, and shared writes are dropped —
the same behaviour as a degraded storage. `Registry.get()` keeps returning the
closed instance rather than silently starting a fresh lane.
- **The cached shared view is dropped.** Nothing refreshes it once the lane is
gone, so keeping it would pin the breaker in whatever a peer last published —
a shared `OPEN` would never expire. The fallback to local state is reported
through `on_state_change` like any other.
`close()` is teardown, not a state change: it does **not** close the circuit.
That is `reset()`.
## Tuning
All knobs live on the storage constructor; the core `Config` stays
storage-agnostic:
```python
RedisStorage(
client,
key_prefix='interlock:cb:', # hash key namespace
state_ttl=300.0, # key lifetime (s); refreshed on every write
poll_interval=1.0, # cache refresh cadence (s)
retry_backoff=5.0, # local-only time after a storage failure (s)
retry_backoff_multiplier=1.0, # growth per consecutive failure; 1.0 = fixed delay
retry_backoff_max=None, # cap on the delay (s), or None for no cap
retry_jitter=0.0, # proportional random spread added to the (capped) delay
write_queue_size=128, # max pending coordinated writes; further ones are dropped
)
```
- **`state_ttl`** keeps abandoned state from lingering: if every instance
disappears, the key expires and the breaker starts CLOSED. Keep it well above
`wait_duration_in_open`.
- **`poll_interval`** is the propagation latency of a coordinated trip. Each
breaker costs about one Redis read per interval.
- **`retry_backoff`** is the delay before the first retry after a storage
failure, and the fixed delay for every retry after that while
`retry_backoff_multiplier` stays at its default of `1.0`.
- **`retry_backoff_multiplier`** grows the delay geometrically with each
further *consecutive* failure (`retry_backoff * retry_backoff_multiplier **
attempts`); a recovery resets the attempt count. Must be `>= 1.0`.
- **`retry_backoff_max`** caps the grown delay in seconds. `None` (the
default) leaves it uncapped.
- **`retry_jitter`** adds up to this fraction of the (capped) delay as random
spread, so instances that degraded at the same moment do not all retry
Redis at the same moment too. Deterministic given the same clock reading,
attempt number and breaker name — it does not depend on process-global
random state.
- **`write_queue_size`** bounds the pending coordinated writes; see
[Backpressure](#backpressure-the-write-queue-is-bounded). The default of 128
is far above what a draining lane ever holds, so raising it does not buy
throughput — it only lets a wedged lane hold more memory before dropping.
## Compatibility
`RedisStorage` speaks plain commands and `EVAL` — no server-specific features —
so it works against Redis, [Valkey](https://valkey.io), or any RESP-compatible
server. The scripts call `TIME` before writing, which requires effect-based
script replication: **Redis 5.0 or newer**, or any Valkey release. (The
`redis>=5.0.0` dependency pin is the *client* library's version, not the
server's.)
---
# Flask / Django — recipe
When a route's outgoing dependency trips its breaker, the raised
[`CircuitOpenError`](../reference.md) should become a clean
`503 Service Unavailable` with a `Retry-After` header — the same behaviour
the [FastAPI](fastapi.md) and [Litestar](litestar.md) extras ship as code.
For other frameworks the handler is a few lines; no extra needed.
## Flask
```python
import math
from flask import Flask, jsonify
from interlock import CircuitOpenError
app = Flask(__name__)
@app.errorhandler(CircuitOpenError)
def on_circuit_open(exc: CircuitOpenError):
response = jsonify({'detail': str(exc)})
response.status_code = 503
if exc.retry_after is not None:
response.headers['Retry-After'] = str(math.ceil(exc.retry_after))
return response
```
## Django
```python
# middleware.py
import json
import math
from django.http import HttpResponse
from interlock import CircuitOpenError
class CircuitOpenMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
return self.get_response(request)
def process_exception(self, request, exception):
if not isinstance(exception, CircuitOpenError):
return None
response = HttpResponse(
json.dumps({'detail': str(exception)}),
status=503,
content_type='application/json',
)
if exception.retry_after is not None:
response['Retry-After'] = str(math.ceil(exception.retry_after))
return response
```
Add it to `MIDDLEWARE` in `settings.py`.
## Where the breakers live
The handler only translates the rejection. The breakers themselves guard your
*outgoing* calls — share one `Registry` across the app and wrap the
dependencies:
```python
from interlock import Registry
registry = Registry()
payments = registry.get('payments-api')
def charge(amount: int) -> str:
return payments.call(gateway.charge, amount)
```
`Retry-After` is rounded up to whole seconds (per RFC 7231) and omitted when
the breaker cannot estimate the next probe (for example after
`force_open()`).
---
# API reference
Everything below is importable from the top-level `interlock` package, except
the integration adapters, which live in their own modules to keep the core
dependency-free.
## `CircuitBreaker`
```python
CircuitBreaker(*, name, config=None, clock=None, initial_state=State.CLOSED,
classifier=None, listener=None, storage=None)
```
A named breaker for sync and async callables.
- **Use as** a decorator (`@breaker`), a sync/async context manager
(`with` / `async with`), or `breaker.call(fn, *args, **kwargs)`.
- **`call_sync(fn, ...)` / `call_async(fn, ...)`** — the same protection with the
sync/async dispatch skipped, for callers that already know their own nature
(the transport integrations use them per request). `call_sync` never awaits: a
coroutine function passed to it is recorded as an immediate success.
`call_async` awaits whatever `fn` returns, so it accepts any
awaitable-returning callable, not only a coroutine function.
- **Properties:** `name: str`, `state: State`.
- **`initial_state`** — one of `CLOSED`, `FORCED_OPEN`, `DISABLED` or
`METRICS_ONLY`; transitional `OPEN` / `HALF_OPEN` raise `ValueError`.
- **`snapshot() -> WindowSnapshot`** — current, self-consistent window aggregates;
concurrent call settlement cannot expose a partially updated window.
- **Manual control:** `reset()`, `force_open()`, `disable()`, `metrics_only()`.
- **`close()` / `aclose()`** — release background resources: the coordinator
lane and the `auto_transition` timer. Teardown, not a state change: neither
closes the circuit (`reset()` does). Idempotent, and terminal — the lane never
restarts. See [Shutdown](integrations/redis.md#shutdown).
- **`storage`** — optional shared backend (`Storage` or `AsyncStorage`) for
coordinated state across instances; see the
[Redis integration](integrations/redis.md). A coordinated breaker matches its
storage's runtime (sync storage → sync API, async storage → async API);
without a storage the breaker stays fully dual.
## `Config`
Frozen dataclass of thresholds, window and timing; validated on construction.
See [Configuration](guides/configuration.md) for every field. Raises
`ValueError` on invalid input.
## `Registry`
```python
Registry(*, config=None, clock=None, initial_state=State.CLOSED,
classifier=None, listener=None, storage=None)
registry.get(name, *, config=None) -> CircuitBreaker
registry.get_existing(name) -> CircuitBreaker | None
registry.names() -> tuple[str, ...]
registry.items() -> tuple[tuple[str, CircuitBreaker], ...]
registry.close_all() / await registry.aclose_all()
```
Creates and caches named breakers. The same name always returns the same
instance; the per-call `config` override applies only at creation. A cache hit
is served without taking the registry lock, so a registry shared by every
request of a service does not serialise on lookups. A `storage`
is handed to every breaker the registry creates; each coordinates under its own
name. `initial_state` is assigned before a lazy breaker is published;
`get_existing()` inspects the cache without creating a missing name.
`names()` and `items()` enumerate the breakers created so far — the only way to
see the ones the HTTP transports create lazily, one per host. Both take a
point-in-time copy under the registry lock: a breaker created afterwards is not
in it, and the returned tuple never changes.
```python
for name, breaker in registry.items():
print(name, breaker.state, breaker.snapshot())
```
`close_all()` / `aclose_all()` close every breaker created so far. The cache is
kept, so `get()` keeps returning the same, torn-down instances instead of
silently starting a fresh lane after shutdown.
## Enums
- **`State`** — `CLOSED`, `OPEN`, `HALF_OPEN`, `FORCED_OPEN`, `DISABLED`,
`METRICS_ONLY`. A `StrEnum`; values are stable lowercase identifiers.
- **`Outcome`** — `SUCCESS`, `FAILURE`, `SLOW_SUCCESS`, `SLOW_FAILURE`, with
`.is_failure` and `.is_slow` properties.
- **`WindowType`** — `COUNT_BASED`, `TIME_BASED`.
## `WindowSnapshot`
Frozen dataclass: `total_calls`, `failed_calls`, `slow_calls`, plus
`.failure_rate` and `.slow_call_rate` properties (both `0.0` when empty).
## Errors & warnings
- **`InterlockError`** — base of all interlock errors.
- **`CircuitOpenError(breaker_name, *, retry_after=None, last_failure=None)`** —
raised on rejection; attributes `breaker_name`, `retry_after`, `last_failure`.
- **`CallTimeoutError(timeout)`** — raised by `timeout` and `sync_timeout`;
attribute `timeout`.
- **`BulkheadFullError(max_concurrent, *, max_wait=0.0)`** — raised by a
pipeline bulkhead when no concurrency slot frees up in time; attributes
`max_concurrent`, `max_wait`.
- **`InterlockDeprecationWarning`** — subclasses `UserWarning`, visible by
default.
Every HTTP client integration retypes the **rejection** — and only the
rejection — as a subclass that is *also* a native error of the host library, so
the idiom that library teaches and `except CircuitOpenError` catch the same
error. The native base differs per integration; one `except` clause covers one
client, not all four:
| Integration | Rejection type | Native base |
|---|---|---|
| httpx | `CircuitOpenTransportError` | `httpx.TransportError` |
| httpx2 | `CircuitOpenTransportError` | `httpx2.TransportError` |
| aiohttp | `CircuitOpenClientError` | `aiohttp.ClientConnectionError` |
| requests | `CircuitOpenRequestError` | `requests.exceptions.ConnectionError` |
`CallTimeoutError` and `BulkheadFullError` are paired only by the httpx and
httpx2 transports, as `CallTimeoutTransportError` (a `TimeoutException`) and
`BulkheadFullTransportError` (a `PoolTimeout`); they surface a timeout or
bulkhead raised by a layer inside the wrapped transport. The aiohttp middleware
and the requests adapter leave both untouched — neither hierarchy has an honest
type for "no local slot was free" — so they reach the caller as plain interlock
errors. See the integration sections below.
## `timeout` / `sync_timeout`
```python
async with timeout(seconds):
... # async block
@sync_timeout(seconds) # synchronous callable
def work(): ...
```
`timeout` is an async context manager that raises `CallTimeoutError` if the
block exceeds `seconds`. `sync_timeout` is a decorator that runs a synchronous
callable in a daemon worker thread and raises `CallTimeoutError` if it overruns
`seconds`; the worker keeps running after a timeout (Python cannot kill a
thread). See [Timeout](guides/timeout.md).
## Pipeline
Compose strategies around one call, outermost first — see the
[pipeline guide](guides/pipeline.md):
- **`Pipeline(*strategies)`** — the executor; works as a signature-preserving
decorator and as `pipeline.call(fn, *args, **kwargs)` (detect-dispatching,
like the breaker's). No context manager by design.
- **`Pipeline.builder()` / `PipelineBuilder`** — step-by-step assembly:
`.fallback(...)`, `.retry(...)` (lazy `tenacity` extra),
`.circuit_breaker(breaker)`, `.bulkhead(...)`, `.timeout(seconds)`,
`.add(custom)`, `.build()`.
- **`Strategy`** — the structural protocol: `execute(call)` /
`execute_async(call)`; `execute_async` always receives a real coroutine
function.
- **`CircuitBreakerStrategy(breaker)`** — wraps a standalone breaker unchanged.
- **`TimeoutStrategy(seconds)`** — bounds every attempt via the v1 primitives.
- **`BulkheadStrategy(max_concurrent, *, max_wait=0.0, name='bulkhead',
listener=None)`** — concurrency cap; raises `BulkheadFullError`.
- **`FallbackStrategy(fallback, *, on=(Exception,), name='fallback',
listener=None)`** — explicit substitution for selected failures; result
typed `T | F`.
- **`RetryStrategy(...)`** — lives in `interlock.integrations.tenacity` (see
below).
## Protocols (extension points)
Implement any of these to swap a core behaviour:
- **`Clock`** — `monotonic() -> float`. Inject a fake for deterministic tests.
- **`SlidingWindow`** — `record(outcome)`, `snapshot() -> WindowSnapshot`.
- **`Storage`** / **`AsyncStorage`** — shared-state backend as atomic *intent*
operations: `read`, `trip_open`, `begin_half_open_if_elapsed`, `lease_probe`,
`record_probe`, `close`. `trip_open`/`close` take an optional
`expected_version` (version-fenced CAS); every write carries a `ttl`.
Mechanism only — threshold policy stays in the core. `AsyncStorage` is the
awaitable mirror. See the [Redis integration](integrations/redis.md).
- **`FailureClassifier`** — `is_failure(*, result, exception) -> bool`, where
`exception` is an `Exception` or `None` — cancellation and shutdown are
released without being classified. See
[Failure classification](guides/failure-classification.md).
- **`EventListener`** — `on_state_change`, `on_call`, `on_rejected`, `on_reset`,
plus `on_storage_degraded` / `on_storage_recovered` /
`on_storage_write_dropped` for coordinated breakers
and `on_retry` / `on_bulkhead_rejected` / `on_fallback` for pipeline
strategies (all optional hooks are dispatched only if present, so older
listeners keep working). See [Observability](guides/observability.md).
## Shared-state types
- **`SharedState`** — frozen snapshot of one breaker's coordinated state:
`state`, `opened_at` (backend time), `version` (for fencing), and the
HALF_OPEN probe accounting (`probes_permitted`, `probes_remaining`,
`probes_completed`, `probe_failures`, `probe_slows`).
`SharedState.closed()` is the baseline an absent key implies.
- **`ProbeLease`** — result of `lease_probe`: `granted: bool` plus the
post-attempt `state: SharedState`.
## Listeners
- **`LoggingEventListener(logger=None)`** — top-level; zero dependencies.
- **`interlock.integrations.otel.OTelEventListener(meter=None)`** — extra `interlock-cb[otel]`.
## httpx2 adapters
Extra `interlock-cb[httpx2]`, module `interlock.integrations.httpx2`:
- **`CircuitBreakerTransport(transport, *, config=None, clock=None,
initial_state=State.CLOSED, classifier=None, listener=None, registry=None,
name_resolver=)`**
- **`AsyncCircuitBreakerTransport(transport, *, ...)`**
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
fails on transport exceptions and statuses `429, 500, 502, 503, 504`
(override the set via `failure_statuses`). `UnsupportedProtocol` and
`LocalProtocolError` are caller-side and count as successes; replace that set
via `excluded_exceptions`.
- **`CircuitOpenTransportError(breaker_name, *, retry_after=None,
last_failure=None, request=None)`** — the rejection: an
`httpx2.TransportError` *and* a `CircuitOpenError`.
- **`CallTimeoutTransportError(timeout, *, request=None)`** — an
`httpx2.TimeoutException` *and* a `CallTimeoutError`.
- **`BulkheadFullTransportError(max_concurrent, *, max_wait=0.0,
request=None)`** — an `httpx2.PoolTimeout` *and* a `BulkheadFullError`.
Both transports expose their per-host `registry` and the guarded transport as a
read-only `wrapped`. `close()` / `aclose()` release the wrapped transport and,
when the transport owns the registry, every breaker in it; a caller-owned
registry stays open and is closed by its owner. Every adapter below takes the
same `registry` option — a caller-owned `Registry` shared between clients — and
combining it with any breaker-construction option (`config`, `clock`,
`initial_state`, `classifier`, `listener`) raises `ValueError`, since the
registry already owns them. `name_resolver` maps each request to its breaker
name and defaults to the request host. See the
[httpx2 integration](integrations/httpx2.md).
## httpx adapters
Extra `interlock-cb[httpx]` (httpx ≥ 0.27.0), module
`interlock.integrations.httpx`:
- **`CircuitBreakerTransport(transport, *, config=None, clock=None,
initial_state=State.CLOSED, classifier=None, listener=None, registry=None,
name_resolver=)`**
- **`AsyncCircuitBreakerTransport(transport, *, ...)`**
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
same policy, including the caller-side `UnsupportedProtocol` /
`LocalProtocolError` exclusions.
- **`CircuitOpenTransportError`**, **`CallTimeoutTransportError`**,
**`BulkheadFullTransportError`** — the same three dialect errors as the
httpx2 adapters, built on `httpx.TransportError`, `httpx.TimeoutException`
and `httpx.PoolTimeout`.
Both transports expose their per-host `registry` and the guarded transport as a
read-only `wrapped`, and preserve streaming responses. `close()` / `aclose()`
release the wrapped transport and, when the transport owns the registry, every
breaker in it; a caller-owned registry stays open and is closed by its owner.
See the [httpx integration](integrations/httpx.md).
## aiohttp adapters
Extra `interlock-cb[aiohttp]` (aiohttp ≥ 3.12), module `interlock.integrations.aiohttp`:
- **`CircuitBreakerMiddleware(*, config=None, clock=None,
initial_state=State.CLOSED, classifier=None, listener=None, registry=None,
name_resolver=)`** —
client middleware for `ClientSession(middlewares=(...,))`; one breaker per
request host.
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
same canonical HTTP policy, reading `ClientResponse.status`; nothing is
excluded by default (aiohttp rejects bad URLs before the middleware runs).
- **`CircuitOpenClientError(breaker_name, *, retry_after=None,
last_failure=None)`** — the rejection: an `aiohttp.ClientConnectionError`
*and* a `CircuitOpenError`.
It exposes its `registry`; call `await middleware.aclose()` during application
shutdown. See the [aiohttp integration](integrations/aiohttp.md).
## requests adapters
Extra `interlock-cb[requests]`, module `interlock.integrations.requests`:
- **`CircuitBreakerAdapter(*, config=None, clock=None,
initial_state=State.CLOSED, classifier=None, listener=None, registry=None,
name_resolver=, **adapter_kwargs)`** —
`HTTPAdapter` subclass for `session.mount(...)`; one breaker per request
host. Extra kwargs go to `HTTPAdapter`.
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
same policy, reading `Response.status_code`; the caller-side `InvalidURL`
counts as a success.
- **`CircuitOpenRequestError(breaker_name, *, retry_after=None,
last_failure=None, request=None)`** — the rejection: a
`requests.exceptions.ConnectionError` *and* a `CircuitOpenError`; carries
requests' own `.request` and `.response`.
It exposes its `registry`; closing the adapter or owning session releases both
the connection pools and breaker resources. See the
[requests integration](integrations/requests.md).
## tenacity helpers
Extra `interlock-cb[tenacity]`, module `interlock.integrations.tenacity`:
- **`retry_unless_open(*transient)`** — tenacity retry predicate: retries the
listed transient exception types (default: any `Exception`), never
`CircuitOpenError`.
- **`wait_probe(fallback, *, jitter=0.1)`** — tenacity wait strategy: sleeps
`CircuitOpenError.retry_after` (+ up to `jitter` seconds) after a
rejection, delegates to `fallback` otherwise.
- **`RetryStrategy(*, attempts=3, retry=None, wait=None, sleep=None,
async_sleep=None, before_sleep=None, name='retry', listener=None)`** — a
bounded retry layer for the pipeline: policy delegated to tenacity,
attempts always capped, the original exception re-raised when the budget
runs out, `CircuitOpenError` not retried by default.
See the [tenacity integration](integrations/tenacity.md) and the
[retries guide](guides/retries.md).
## FastAPI adapters
Extra `interlock-cb[fastapi]`, module `interlock.integrations.fastapi`:
- **`breaker_dependency(name, *, registry)`** — returns a `Depends`-compatible
callable yielding the named breaker from a shared `Registry`.
- **`install_exception_handler(app)`** — registers a handler mapping
`CircuitOpenError` to `503` with a `Retry-After` header.
- **`circuit_open_handler(request, exc)`** — the handler itself, for custom
registration.
See the [FastAPI integration](integrations/fastapi.md).
## Litestar adapters
Extra `interlock-cb[litestar]` (Litestar ≥ 2.23), module
`interlock.integrations.litestar`:
- **`breaker_dependency(name, *, registry)`** — returns a `Provide` yielding
the named breaker from a shared `Registry`; annotate handler parameters
with `NamedDependency[CircuitBreaker]`.
- **`circuit_open_handler(request, exc)`** — maps `CircuitOpenError` to `503`
with a `Retry-After` header; pass it in the app's `exception_handlers`.
See the [Litestar integration](integrations/litestar.md).
## Redis adapters
Extra `interlock-cb[redis]`, module `interlock.integrations.redis`:
- **`RedisStorage(client, *, key_prefix='interlock:cb:', state_ttl=300.0, poll_interval=1.0, retry_backoff=5.0, retry_backoff_multiplier=1.0, retry_backoff_max=None, retry_jitter=0.0, write_queue_size=128)`** —
sync `Storage` over a `redis.Redis` client.
- **`AsyncRedisStorage(client, *, ...)`** — async mirror over
`redis.asyncio.Redis`.
One Redis hash per breaker; every transition is a Lua script (atomic across
racing instances), elapse checks use the server's `TIME`. Works against Redis
(5.0+), Valkey, or any RESP-compatible server.
See the [Redis integration](integrations/redis.md).