Circuit breaker state machine.
CLOSED, OPEN and HALF-OPEN with a sliding failure window, a recovery timer on a fake clock, and business errors that don't trip it.
When the bank API is down, every call that waits 3 seconds for a timeout holds a thread, and the payment service dies with it. A circuit breaker notices the failures and fails fast instead (Day 3, chapter 1). Build the lesson's breaker from section 2.2:
| From | To | When |
|---|---|---|
| CLOSED | OPEN | failures inside the window reach failure_threshold |
| OPEN | HALF_OPEN | recovery_timeout seconds have passed since the last failure |
| HALF_OPEN | CLOSED | success_threshold trial calls succeed |
| HALF_OPEN | OPEN | a trial call fails |
Implement CircuitBreaker(name, clock, failure_threshold=5, failure_window=60, recovery_timeout=30, success_threshold=1, failure_exceptions=(Exception,)):
| Member | Does |
|---|---|
call(func, *args, **kwargs) |
runs func and returns its result, re-raising its exceptions unchanged. While OPEN it raises CircuitOpenError without calling func |
state (property) |
the current CircuitState, after any time-based transition |
The state machine, exactly as the lesson implements it:
- CLOSED: record the time of each failure. Drop failures older than
failure_windowseconds. When the failures still in the window reachfailure_threshold, open. A success in CLOSED does not clear the window: it is a time window, not a consecutive-failure counter. - OPEN: once
recovery_timeoutseconds have passed since the last failure, move to HALF_OPEN (checked lazily, oncallorstate). Rejected calls don't restart the timer. - HALF_OPEN: let calls through as trials.
success_thresholdsuccesses close the circuit and clear the failures. One failure re-opens it and restarts the timer. - Only exceptions that are instances of
failure_exceptionscount as failures. Anything else (say, a declined card) propagates to the caller and is neither a failure nor a success: the service answered correctly (Day 3, "What counts as a failure?"). - Read time only through
clock.now().
Write your code, then run 3 example tests and 10 hidden tests. Each run is limited to 8 seconds: an infinite loop gets stopped and Python restarts, your code stays.