From bb43abbe7b755af023976b4e23e00e3fb1a77e19 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 11:43:16 +0300 Subject: [PATCH 1/3] fix: circuit breaker ignores straggler outcomes, unwraps budget exhaustion, rate-limits circuit.rejected Outcomes now carry an epoch ticket from admission, so a request that finishes after its circuit changed state no longer reopens, closes or frees the probe slot. A RetryBudgetExhaustedError is classified by its last_exception. circuit.rejected is emitted outside the sync lock, at most once per second per circuit, with a suppressed count. --- CONTEXT.md | 6 +- docs/observability.md | 2 +- docs/resilience.md | 6 +- .../middleware/resilience/circuit_breaker.py | 225 +++++++++-------- tests/test_circuit_breaker.py | 191 ++++++++++++++ tests/test_circuit_breaker_props.py | 123 ++++++++- tests/test_circuit_breaker_sync.py | 234 ++++++++++++++++++ tests/test_keyed_circuit_breaker.py | 42 ++++ tests/test_keyed_circuit_breaker_sync.py | 80 ++++++ 9 files changed, 803 insertions(+), 106 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 71455d7..ce6e716 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -46,9 +46,9 @@ _Avoid_: adapter, which names the third-party object a decoder builds and memoiz **Counted failure**: The circuit breaker's unit of account: a `NetworkError`, an httpware `TimeoutError`, or a `StatusError` whose status is in the effective failure set (default: all 5xx). Every other 4xx — -429 included — counts as a *success*, and any other exception type propagates without touching -circuit state. "Failure" alone is ambiguous here: a failed request is very often not a counted -failure. +429 included — counts as a *success*. A `RetryBudgetExhaustedError` is classified by its +`last_exception`, and any other exception type propagates without touching circuit state. "Failure" +alone is ambiguous here: a failed request is very often not a counted failure. **Circuit key**: The value a keyed circuit breaker computes from a request to pick its circuit; requests with equal diff --git a/docs/observability.md b/docs/observability.md index 7d532f6..ed0b3af 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -11,7 +11,7 @@ The logger and event names below are a stable public contract: | `httpware.circuit_breaker` | `circuit.opened` (WARNING), `circuit.rejected` (WARNING), `circuit.half_open` (INFO), `circuit.closed` (INFO) | | `httpware.timeout` | `timeout.exceeded` (WARNING) | -Each log record has an `event` field holding the event name, such as `event="circuit.opened"`, which you can filter on in your log aggregator. Events from `AsyncKeyedCircuitBreaker` and `KeyedCircuitBreaker` also carry `circuit_key`, the circuit they belong to. [Resilience](resilience.md) describes when each event fires. +Each log record has an `event` field holding the event name, such as `event="circuit.opened"`, which you can filter on in your log aggregator. `circuit.rejected` is emitted at most once per second per circuit, so an open circuit under heavy traffic does not flood your logs; its `suppressed` attribute counts the rejections that were left out since the previous one. Events from `AsyncKeyedCircuitBreaker` and `KeyedCircuitBreaker` also carry `circuit_key`, the circuit they belong to. [Resilience](resilience.md) describes when each event fires. ```python import logging diff --git a/docs/resilience.md b/docs/resilience.md index 5adec2d..b03ae63 100644 --- a/docs/resilience.md +++ b/docs/resilience.md @@ -160,6 +160,8 @@ A circuit breaker counts consecutive failures and, past a threshold, stops sendi - OPEN: until `reset_timeout` has passed, every request fails at once with `CircuitOpenError`, whose `retry_after` says how many seconds remain. The first request after that moves the circuit to HALF_OPEN and becomes the probe. - HALF_OPEN: one probe at a time is let through. After `success_threshold` successful probes in a row the circuit closes; one failed probe opens it again. +An outcome only counts against the state its request was admitted under. A request that is still in flight when the circuit changes state, such as a slow call sent before the circuit opened, finishes without touching the circuit. It cannot reopen a HALF_OPEN circuit, count toward `success_threshold`, or free the probe slot. + ### Constructor | Parameter | Default | Effect | @@ -176,6 +178,8 @@ A circuit breaker counts consecutive failures and, past a threshold, stops sendi A counted failure is a `NetworkError`, an httpware `TimeoutError`, or a `StatusError` whose status is in `failure_status_codes`. Other exceptions pass through without changing the circuit. +A `RetryBudgetExhaustedError` from a retry placed after the breaker is classified by its `last_exception`. If the last attempt failed with a counted failure, the breaker counts one; if it failed with a 4xx, that is a success. + 4xx responses, 429 included, count as successes. A 429 means the service is up and throttling you; opening the circuit on it would add circuit-open rejections on top of the throttling. ### `CircuitOpenError` @@ -189,7 +193,7 @@ Emitted on logger `httpware.circuit_breaker`: | Event | When | |---|---| | `circuit.opened` | The failure threshold was reached and the circuit went from CLOSED to OPEN. | -| `circuit.rejected` | A request failed fast because the circuit was OPEN or a HALF_OPEN probe was already running. | +| `circuit.rejected` | A request failed fast because the circuit was OPEN or a HALF_OPEN probe was already running. Emitted at most once per second per circuit; its `suppressed` attribute counts the rejections since the previous `circuit.rejected` that were not emitted. | | `circuit.half_open` | The reset timeout passed and the circuit went from OPEN to HALF_OPEN. | | `circuit.closed` | The success threshold was reached and the circuit went from HALF_OPEN to CLOSED. | diff --git a/src/httpware/middleware/resilience/circuit_breaker.py b/src/httpware/middleware/resilience/circuit_breaker.py index 5521d9a..59fcdc9 100644 --- a/src/httpware/middleware/resilience/circuit_breaker.py +++ b/src/httpware/middleware/resilience/circuit_breaker.py @@ -6,7 +6,8 @@ A counted failure is a NetworkError, an httpware TimeoutError, or a StatusError whose status_code is in the effective failure set (default: all 5xx). 4xx — including 429 — count as successes: 429 means healthy-but-throttling, and tripping on it amplifies -incidents. Any other exception propagates without affecting circuit state. In +incidents. A RetryBudgetExhaustedError is classified by its last_exception. Any other +exception propagates without affecting circuit state. In particular, non-NetworkError transport problems — e.g. httpx2.InvalidURL from a malformed URL — are foreign: they propagate unchanged and do not increment the failure counter, so programming errors cannot trip the breaker. @@ -18,6 +19,12 @@ HALF_OPEN — admit exactly one probe at a time; success_threshold consecutive probe successes close the circuit; one probe failure re-opens it. +An outcome applies only to the state its request was admitted under: a request still in +flight when the circuit changes state finishes without touching the circuit. + +circuit.rejected is emitted at most once per second per circuit; its `suppressed` +attribute counts the rejections not emitted since the previous one. + Trip modes: Classic (default) — opens when consecutive counted-failures reach failure_threshold. Set failure_threshold to use this mode; leave failure_rate_threshold unset. @@ -38,6 +45,7 @@ import enum import functools import logging +import math import threading import time import typing @@ -46,7 +54,13 @@ import httpx2 from httpware._internal.observability import _emit_event -from httpware.errors import CircuitOpenError, NetworkError, StatusError, TimeoutError # noqa: A004 +from httpware.errors import ( + CircuitOpenError, + NetworkError, + RetryBudgetExhaustedError, + StatusError, + TimeoutError, # noqa: A004 +) from httpware.middleware import AsyncNext, Next from httpware.middleware.resilience._event_loop_guard import check_event_loop @@ -72,8 +86,14 @@ _BUCKET_COUNT = 10 -_ROLE_CLOSED = "closed" -_ROLE_PROBE = "probe" +_SUCCESS = 0 +_FAILURE = 1 +_FOREIGN = 2 + +_PROBE = 1 +_EPOCH_STEP = 2 + +_REJECTED_EVENT_INTERVAL = 1.0 _LOGGER = logging.getLogger("httpware.circuit_breaker") @@ -136,12 +156,23 @@ def clear(self) -> None: self._failure = [0] * _BUCKET_COUNT +class _RejectionError(Exception): + """Raised by `admit` to refuse a request; `message` is None when the event is suppressed.""" + + def __init__(self, retry_after: float | None, message: str | None, suppressed: int) -> None: + self.retry_after = retry_after + self.message = message + self.suppressed = suppressed + + class _CircuitBreakerState: """Lock-free circuit-breaker state machine shared by the sync + async wrappers. Every method is synchronous and performs no I/O beyond logging. The async wrapper calls these directly (atomic under a single event loop because no await occurs - inside a transition); the sync wrapper wraps each call in a threading.Lock. + inside a transition); the sync wrapper wraps `admit` and `record` in a threading.Lock. + `admit` hands out a ticket: the epoch, which every transition advances by two, plus one + for the probe. `record` ignores a ticket from an earlier epoch. """ def __init__( # noqa: PLR0913 — breaker state has many orthogonal knobs; a dataclass would be worse @@ -189,86 +220,99 @@ def __init__( # noqa: PLR0913 — breaker state has many orthogonal knobs; a da self._consecutive_successes = 0 self._opened_at = 0.0 self._probe_in_flight = False - - def is_failure_status(self, status_code: int) -> bool: - return status_code in self._failure_status_codes + self._epoch = 0 + self._rejected_emitted_at = -math.inf + self._rejections_suppressed = 0 @property def state(self) -> CircuitState: """The circuit's current stored state (raw read; no lazy OPEN→HALF_OPEN transition).""" return self._state - def admit(self, request: httpx2.Request) -> str: - """Decide the request's role, or raise CircuitOpenError. No await inside.""" + def admit(self, request: httpx2.Request) -> int: + """Return the request's ticket, or raise _RejectionError. No await inside.""" if self._state is CircuitState.CLOSED: - return _ROLE_CLOSED + return self._epoch + now = self._now() if self._state is CircuitState.OPEN: - elapsed = self._now() - self._opened_at - if elapsed >= self._reset_timeout: - self._state = CircuitState.HALF_OPEN - self._probe_in_flight = True - self._emit(request, "circuit.half_open", logging.INFO, "circuit half-open — admitting probe", {}) - return _ROLE_PROBE - retry_after = max(0.0, self._reset_timeout - elapsed) - self._emit( - request, - "circuit.rejected", - logging.WARNING, - "circuit open — rejecting request", - {"retry_after": retry_after}, - ) - raise CircuitOpenError(retry_after=retry_after) - # HALF_OPEN - if self._probe_in_flight: + elapsed = now - self._opened_at + if elapsed < self._reset_timeout: + self._reject(now, max(0.0, self._reset_timeout - elapsed), "circuit open — rejecting request") + self._move_to(CircuitState.HALF_OPEN) + self._emit(request, "circuit.half_open", logging.INFO, "circuit half-open — admitting probe", {}) + elif self._probe_in_flight: + self._reject(now, None, "circuit half-open — rejecting request (probe in flight)") + self._probe_in_flight = True + return self._epoch + _PROBE + + def classify(self, exc: BaseException | None) -> int: + """Map an exception raised past the breaker to _SUCCESS, _FAILURE or _FOREIGN.""" + if isinstance(exc, RetryBudgetExhaustedError): + exc = exc.last_exception + if isinstance(exc, StatusError): + return _FAILURE if exc.response.status_code in self._failure_status_codes else _SUCCESS + if isinstance(exc, (NetworkError, TimeoutError)): + return _FAILURE + return _FOREIGN + + def record(self, ticket: int, outcome: int, request: httpx2.Request) -> None: + """Apply an admitted request's outcome; a no-op if the circuit has transitioned since.""" + if ticket == self._epoch: + if outcome == _SUCCESS and not self._rate_mode: + self._consecutive_failures = 0 + elif outcome != _FOREIGN: + self._record_closed(request, failed=outcome == _FAILURE) + elif ticket == self._epoch + _PROBE: + self._probe_in_flight = False + if outcome == _SUCCESS: + self._record_probe_success(request) + elif outcome == _FAILURE: + self._open(request, failures=1) # 1 = the single probe failure that re-opened the circuit + + def rejected(self, request: httpx2.Request, rejection: _RejectionError) -> CircuitOpenError: + """Emit the rejection's event unless `admit` suppressed it; return the error to raise.""" + if rejection.message is not None: self._emit( request, "circuit.rejected", logging.WARNING, - "circuit half-open — rejecting request (probe in flight)", - {"retry_after": None}, + rejection.message, + {"retry_after": rejection.retry_after, "suppressed": rejection.suppressed}, ) - raise CircuitOpenError(retry_after=None) - self._probe_in_flight = True - return _ROLE_PROBE - - def on_success(self, role: str, request: httpx2.Request) -> None: - if role == _ROLE_PROBE: - self._probe_in_flight = False - if self._state is CircuitState.CLOSED: + return CircuitOpenError(retry_after=rejection.retry_after) + + def _reject(self, now: float, retry_after: float | None, message: str) -> typing.NoReturn: + if now - self._rejected_emitted_at < _REJECTED_EVENT_INTERVAL: + self._rejections_suppressed += 1 + raise _RejectionError(retry_after, None, 0) + self._rejected_emitted_at = now + suppressed, self._rejections_suppressed = self._rejections_suppressed, 0 + raise _RejectionError(retry_after, message, suppressed) + + def _record_closed(self, request: httpx2.Request, *, failed: bool) -> None: + if self._rate_mode: + self._record_outcome(request, failed=failed) + elif failed: + self._consecutive_failures += 1 + if self._consecutive_failures >= self._failure_threshold: + self._open(request, failures=self._consecutive_failures) + + def _record_probe_success(self, request: httpx2.Request) -> None: + self._consecutive_successes += 1 + if self._consecutive_successes >= self._success_threshold: + self._move_to(CircuitState.CLOSED) + self._consecutive_failures = 0 + self._consecutive_successes = 0 if self._rate_mode: - self._record_outcome(request, failed=False) - else: - self._consecutive_failures = 0 - elif self._state is CircuitState.HALF_OPEN: - self._consecutive_successes += 1 - if self._consecutive_successes >= self._success_threshold: - self._state = CircuitState.CLOSED - self._consecutive_failures = 0 - self._consecutive_successes = 0 - if self._rate_mode: - self._window.clear() # ty: ignore[unresolved-attribute] - self._emit(request, "circuit.closed", logging.INFO, "circuit closed — service recovered", {}) + self._window.clear() # ty: ignore[unresolved-attribute] + self._emit(request, "circuit.closed", logging.INFO, "circuit closed — service recovered", {}) - def on_failure(self, role: str, request: httpx2.Request) -> None: - if role == _ROLE_PROBE: - self._probe_in_flight = False - if self._state is CircuitState.CLOSED: - if self._rate_mode: - self._record_outcome(request, failed=True) - else: - self._consecutive_failures += 1 - if self._consecutive_failures >= self._failure_threshold: - self._open(request, failures=self._consecutive_failures) - elif self._state is CircuitState.HALF_OPEN: - self._open(request, failures=1) # 1 = the single probe failure that re-opened the circuit - - def release_probe(self, role: str) -> None: - """Release the probe slot without recording success or failure (non-counted exc).""" - if role == _ROLE_PROBE: - self._probe_in_flight = False + def _move_to(self, state: CircuitState) -> None: + self._state = state + self._epoch += _EPOCH_STEP def _enter_open(self, request: httpx2.Request, message: str, attributes: dict[str, typing.Any]) -> None: - self._state = CircuitState.OPEN + self._move_to(CircuitState.OPEN) self._opened_at = self._now() self._consecutive_failures = 0 self._consecutive_successes = 0 @@ -317,22 +361,16 @@ def _emit( async def _call_async(state: _CircuitBreakerState, request: httpx2.Request, next: AsyncNext) -> httpx2.Response: # noqa: A002 - role = state.admit(request) + try: + ticket = state.admit(request) + except _RejectionError as rejection: + raise state.rejected(request, rejection) from None try: response = await next(request) - except StatusError as exc: - if state.is_failure_status(exc.response.status_code): - state.on_failure(role, request) - else: - state.on_success(role, request) + except BaseException as exc: + state.record(ticket, state.classify(exc), request) raise - except (NetworkError, TimeoutError): - state.on_failure(role, request) - raise - except BaseException: - state.release_probe(role) - raise - state.on_success(role, request) + state.record(ticket, _SUCCESS, request) return response @@ -342,27 +380,20 @@ def _call_sync( request: httpx2.Request, next: Next, # noqa: A002 ) -> httpx2.Response: - with lock: - role = state.admit(request) try: - response = next(request) - except StatusError as exc: with lock: - if state.is_failure_status(exc.response.status_code): - state.on_failure(role, request) - else: - state.on_success(role, request) - raise - except (NetworkError, TimeoutError): - with lock: - state.on_failure(role, request) - raise - except BaseException: + ticket = state.admit(request) + except _RejectionError as rejection: + raise state.rejected(request, rejection) from None + try: + response = next(request) + except BaseException as exc: + outcome = state.classify(exc) with lock: - state.release_probe(role) + state.record(ticket, outcome, request) raise with lock: - state.on_success(role, request) + state.record(ticket, _SUCCESS, request) return response diff --git a/tests/test_circuit_breaker.py b/tests/test_circuit_breaker.py index 528d522..f5d5eb4 100644 --- a/tests/test_circuit_breaker.py +++ b/tests/test_circuit_breaker.py @@ -17,12 +17,15 @@ from httpware import ( AsyncClient, + AsyncRetry, CircuitOpenError, CircuitState, InternalServerError, NetworkError, NotFoundError, RateLimitedError, + RetryBudget, + RetryBudgetExhaustedError, ServiceUnavailableError, TimeoutError, # noqa: A004 — intentional: httpware.TimeoutError shadows the builtin ) @@ -700,3 +703,191 @@ async def test_rate_mode_probe_failure_reopens_with_classic_attributes(caplog: p assert rec.failures == 1 # ty: ignore[unresolved-attribute] # classic probe-failure shape assert hasattr(rec, "failure_threshold") assert not hasattr(rec, "failure_rate") # NOT the rate shape + + +# ── stragglers: outcomes apply to the state the request was admitted under ── + + +class _HeldCalls: + """Starts breaker calls whose `next` blocks until `finish` gives it a status. + + Tracks how many calls that entered `next` while the circuit was HALF_OPEN are in flight. + """ + + def __init__(self, breaker: AsyncCircuitBreaker) -> None: + self._breaker = breaker + self._futures: list[asyncio.Future[int]] = [] + self._tasks: list[asyncio.Task[httpx2.Response]] = [] + self.probes_in_flight = 0 + self.max_probes_in_flight = 0 + + async def _next(self, request: httpx2.Request) -> httpx2.Response: + probe = self._breaker.state is CircuitState.HALF_OPEN + if probe: + self.probes_in_flight += 1 + self.max_probes_in_flight = max(self.max_probes_in_flight, self.probes_in_flight) + future: asyncio.Future[int] = asyncio.get_running_loop().create_future() + self._futures.append(future) + try: + status = await future + finally: + if probe: + self.probes_in_flight -= 1 + if status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(status, request=request)) + return httpx2.Response(status, request=request) + + async def start(self) -> int: + """Start a call; return its index once it is blocked in `next` (or already rejected).""" + task = asyncio.create_task(self._breaker(httpx2.Request("GET", "https://example.test/x"), self._next)) + self._tasks.append(task) + await asyncio.sleep(0) + return len(self._tasks) - 1 + + async def finish(self, index: int, status: int) -> None: + self._futures[index].set_result(status) + with contextlib.suppress(InternalServerError): + await self._tasks[index] + + async def drain(self) -> None: + for future in self._futures: + if not future.done(): + future.set_result(HTTPStatus.OK) + await asyncio.gather(*self._tasks, return_exceptions=True) + + +async def _open_now(breaker: AsyncCircuitBreaker) -> None: + async def _five_hundred(request: httpx2.Request) -> httpx2.Response: + raise InternalServerError(httpx2.Response(HTTPStatus.INTERNAL_SERVER_ERROR, request=request)) + + with pytest.raises(InternalServerError): + await breaker(httpx2.Request("GET", "https://example.test/x"), _five_hundred) + + +async def test_stale_closed_era_failure_does_not_reopen_half_open_circuit() -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + calls = _HeldCalls(breaker) + straggler = await calls.start() + await _open_now(breaker) + clock.advance(1.0) + probe = await calls.start() + await calls.finish(straggler, HTTPStatus.INTERNAL_SERVER_ERROR) + assert breaker.state is CircuitState.HALF_OPEN + await calls.finish(probe, HTTPStatus.OK) + assert breaker.state is CircuitState.CLOSED + + +async def test_stale_closed_era_successes_do_not_close_half_open_circuit() -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=1.0, success_threshold=3, _now=clock) + calls = _HeldCalls(breaker) + stragglers = [await calls.start(), await calls.start()] + await _open_now(breaker) + clock.advance(1.0) + probe = await calls.start() + for straggler in stragglers: + await calls.finish(straggler, HTTPStatus.OK) + await calls.finish(probe, HTTPStatus.OK) + assert breaker.state is CircuitState.HALF_OPEN + + +async def test_stragglers_never_let_a_second_probe_in() -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=1.0, success_threshold=2, _now=clock) + calls = _HeldCalls(breaker) + straggler = await calls.start() + await _open_now(breaker) + clock.advance(1.0) + await calls.start() + await calls.finish(straggler, HTTPStatus.INTERNAL_SERVER_ERROR) + clock.advance(1.0) + await calls.start() + await calls.drain() + assert calls.max_probes_in_flight == 1 + + +# ── a RetryBudgetExhaustedError is classified by its last_exception ── + + +def _no_retry_budget() -> RetryBudget: + return RetryBudget(min_retries_per_sec=0.0, percent_can_retry=0.0) + + +def _behind_starved_retry( + handler: Callable[[httpx2.Request], httpx2.Response], breaker: AsyncCircuitBreaker +) -> AsyncClient: + return AsyncClient( + httpx2_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)), + middleware=[ + breaker, + AsyncRetry(budget=_no_retry_budget(), base_delay=0.0, retry_status_codes=frozenset({429, 503})), + ], + ) + + +async def test_budget_exhausted_on_5xx_is_a_counted_failure() -> None: + breaker = AsyncCircuitBreaker(failure_threshold=1, _now=_Clock()) + async with _behind_starved_retry(_StatusSequence([503]), breaker) as client: + with pytest.raises(RetryBudgetExhaustedError): + await client.get("https://example.test/x") + with pytest.raises(CircuitOpenError): + await client.get("https://example.test/x") + + +async def test_budget_exhausted_on_network_error_is_a_counted_failure() -> None: + def _refuse(request: httpx2.Request) -> httpx2.Response: + msg = "refused" + raise httpx2.ConnectError(msg, request=request) + + breaker = AsyncCircuitBreaker(failure_threshold=1, _now=_Clock()) + async with _behind_starved_retry(_refuse, breaker) as client: + with pytest.raises(RetryBudgetExhaustedError) as info: + await client.get("https://example.test/x") + assert isinstance(info.value.last_exception, NetworkError) + with pytest.raises(CircuitOpenError): + await client.get("https://example.test/x") + + +async def test_budget_exhausted_on_4xx_is_a_success() -> None: + breaker = AsyncCircuitBreaker(failure_threshold=2, _now=_Clock()) + async with _behind_starved_retry(_StatusSequence([500, 429, 500, 200]), breaker) as client: + with pytest.raises(InternalServerError): + await client.get("https://example.test/x") + with pytest.raises(RetryBudgetExhaustedError): + await client.get("https://example.test/x") + with pytest.raises(InternalServerError): + await client.get("https://example.test/x") + response = await client.get("https://example.test/x") + assert response.status_code == HTTPStatus.OK + + +async def test_budget_exhausted_with_a_foreign_last_exception_leaves_the_circuit_alone() -> None: + async def _exhausted(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001 + raise RetryBudgetExhaustedError(last_response=None, last_exception=ValueError("boom"), attempts=1) + + breaker = AsyncCircuitBreaker(failure_threshold=1, _now=_Clock()) + for _ in range(3): + with pytest.raises(RetryBudgetExhaustedError): + await breaker(httpx2.Request("GET", "https://example.test/x"), _exhausted) + assert breaker.state is CircuitState.CLOSED + + +# ── circuit.rejected is rate limited per circuit ── + + +async def test_rejected_event_is_emitted_at_most_once_per_second(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=30.0, _now=clock) + await _open_now(breaker) + + async def _unreachable(request: httpx2.Request) -> httpx2.Response: # pragma: no cover — OPEN never forwards + return httpx2.Response(HTTPStatus.OK, request=request) + + with caplog.at_level(logging.WARNING, logger="httpware.circuit_breaker"): + for step in (0.0, 0.0, 0.0, 0.5, 0.49, 0.01, 0.2): + clock.advance(step) + with pytest.raises(CircuitOpenError): + await breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) + rejected = [r for r in caplog.records if r.event == "circuit.rejected"] # ty: ignore[unresolved-attribute] + assert [r.suppressed for r in rejected] == [0, 4] # ty: ignore[unresolved-attribute] diff --git a/tests/test_circuit_breaker_props.py b/tests/test_circuit_breaker_props.py index 8255971..a941b84 100644 --- a/tests/test_circuit_breaker_props.py +++ b/tests/test_circuit_breaker_props.py @@ -1,16 +1,21 @@ -"""Property test: while OPEN and before reset_timeout, the breaker never forwards. +"""Property tests for the AsyncCircuitBreaker state machine. -Drives the AsyncCircuitBreaker directly with a stub `next` that records calls. -Hypothesis generates random advance/outcome sequences. Time is injected via a clock. +Drives the breaker directly with stub `next` callables; time is injected via a clock. """ +import asyncio +import contextlib +import dataclasses +import typing +from http import HTTPStatus + import httpx2 import pytest from hypothesis import given, settings from hypothesis import strategies as st from httpware import CircuitOpenError, InternalServerError -from httpware.middleware.resilience.circuit_breaker import AsyncCircuitBreaker +from httpware.middleware.resilience.circuit_breaker import AsyncCircuitBreaker, CircuitState class _Clock: @@ -66,3 +71,113 @@ async def _five_hundred(request: httpx2.Request) -> httpx2.Response: await breaker(_request(), _ok) assert forwarded == 0 # `next` (_ok) was never called while OPEN pre-timeout + + +_FOREIGN = 0 + + +@dataclasses.dataclass +class _InFlight: + future: asyncio.Future[int] + task: asyncio.Task[httpx2.Response] + generation: int + + +class _Harness: + """Starts and finishes breaker calls, tracking the circuit's generation and probes in flight. + + The generation counts observed state changes; each operation causes at most one. + """ + + def __init__(self, breaker: AsyncCircuitBreaker) -> None: + self.breaker = breaker + self.generation = 0 + self.observed = breaker.state + self.in_flight: list[_InFlight] = [] + self.probes_in_flight = 0 + + def _observe(self) -> None: + if self.breaker.state is not self.observed: + self.generation += 1 + self.observed = self.breaker.state + + async def _next(self, request: httpx2.Request) -> httpx2.Response: + call = self.in_flight[-1] + probe = self.breaker.state is CircuitState.HALF_OPEN + self.probes_in_flight += probe + assert self.probes_in_flight <= 1 + try: + status = await call.future + finally: + self.probes_in_flight -= probe + if status == _FOREIGN: + msg = "foreign" + raise ValueError(msg) + if status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(status, request=request)) + return httpx2.Response(status, request=request) + + async def start(self) -> None: + future: asyncio.Future[int] = asyncio.get_running_loop().create_future() + call = _InFlight(future, asyncio.create_task(self.breaker(_request(), self._next)), self.generation) + self.in_flight.append(call) + await asyncio.sleep(0) + self._observe() + call.generation = self.generation + if call.task.done(): + self.in_flight.remove(call) + assert isinstance(call.task.exception(), CircuitOpenError) + + async def finish(self, index: int, status: int) -> None: + call = self.in_flight.pop(index % len(self.in_flight)) + before = self.breaker.state + call.future.set_result(status) + with contextlib.suppress(InternalServerError, ValueError): + await call.task + if call.generation != self.generation: + assert self.breaker.state is before + self._observe() + + async def drain(self) -> None: + for call in self.in_flight: + call.future.set_result(HTTPStatus.OK) + await asyncio.gather(*(call.task for call in self.in_flight)) + + +_operations = st.lists( + st.one_of( + st.just(("start",)), + st.tuples( + st.just("finish"), st.integers(min_value=0, max_value=31), st.sampled_from([200, 404, 500, _FOREIGN]) + ), + ), + min_size=8, + max_size=40, +) + + +@given( + failure_threshold=st.integers(min_value=1, max_value=2), + success_threshold=st.integers(min_value=1, max_value=3), + operations=_operations, +) +@settings(max_examples=200, deadline=None) +async def test_outcomes_only_move_the_state_they_were_admitted_under( + failure_threshold: int, + success_threshold: int, + operations: list[tuple[typing.Any, ...]], +) -> None: + """A call finishing after the circuit has changed state leaves it alone, and probes never overlap.""" + breaker = AsyncCircuitBreaker( + failure_threshold=failure_threshold, + reset_timeout=0.0, + success_threshold=success_threshold, + _now=_Clock(), + ) + harness = _Harness(breaker) + for operation in operations: + if operation[0] == "start": + await harness.start() + elif harness.in_flight: + await harness.finish(operation[1], operation[2]) + await harness.drain() diff --git a/tests/test_circuit_breaker_sync.py b/tests/test_circuit_breaker_sync.py index 7516d71..1b5f4c0 100644 --- a/tests/test_circuit_breaker_sync.py +++ b/tests/test_circuit_breaker_sync.py @@ -17,6 +17,9 @@ NetworkError, NotFoundError, RateLimitedError, + Retry, + RetryBudget, + RetryBudgetExhaustedError, ServiceUnavailableError, TimeoutError, # noqa: A004 — intentional: httpware.TimeoutError shadows the builtin ) @@ -589,3 +592,234 @@ def test_rate_mode_probe_failure_reopens_with_classic_attributes(caplog: pytest. assert rec.failures == 1 # ty: ignore[unresolved-attribute] # classic probe-failure shape assert hasattr(rec, "failure_threshold") assert not hasattr(rec, "failure_rate") # NOT the rate shape + + +# ── stragglers: outcomes apply to the state the request was admitted under ── + + +class _HeldCall: + """One breaker call on its own thread whose `next` blocks until `finish` gives it a status.""" + + def __init__(self, calls: "_HeldCalls") -> None: + self._calls = calls + self._settled = threading.Event() + self._release = threading.Event() + self._status = HTTPStatus.OK + self._thread = threading.Thread(target=self._run) + self._thread.start() + assert self._settled.wait(timeout=5.0) + + def _next(self, request: httpx2.Request) -> httpx2.Response: + probe = self._calls.enter() + self._settled.set() + try: + self._release.wait(timeout=5.0) + finally: + self._calls.leave(probe) + if self._status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(self._status, request=request)) + return httpx2.Response(self._status, request=request) + + def _run(self) -> None: + with contextlib.suppress(InternalServerError, CircuitOpenError): + self._calls.breaker(httpx2.Request("GET", "https://example.test/x"), self._next) + self._settled.set() + + def finish(self, status: int) -> None: + self._status = status + self._release.set() + self._thread.join(timeout=5.0) + assert not self._thread.is_alive() + + +class _HeldCalls: + """Counts how many calls that entered `next` while the circuit was HALF_OPEN are in flight.""" + + def __init__(self, breaker: CircuitBreaker) -> None: + self.breaker = breaker + self._lock = threading.Lock() + self._started: list[_HeldCall] = [] + self.probes_in_flight = 0 + self.max_probes_in_flight = 0 + + def enter(self) -> bool: + probe = self.breaker.state is CircuitState.HALF_OPEN + if probe: + with self._lock: + self.probes_in_flight += 1 + self.max_probes_in_flight = max(self.max_probes_in_flight, self.probes_in_flight) + return probe + + def leave(self, probe: bool) -> None: + if probe: + with self._lock: + self.probes_in_flight -= 1 + + def start(self) -> _HeldCall: + call = _HeldCall(self) + self._started.append(call) + return call + + def drain(self) -> None: + for call in self._started: + call.finish(HTTPStatus.OK) + + +def _open_now(breaker: CircuitBreaker) -> None: + def _five_hundred(request: httpx2.Request) -> httpx2.Response: + raise InternalServerError(httpx2.Response(HTTPStatus.INTERNAL_SERVER_ERROR, request=request)) + + with pytest.raises(InternalServerError): + breaker(httpx2.Request("GET", "https://example.test/x"), _five_hundred) + + +def test_stale_closed_era_failure_does_not_reopen_half_open_circuit() -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + calls = _HeldCalls(breaker) + straggler = calls.start() + _open_now(breaker) + clock.advance(1.0) + probe = calls.start() + straggler.finish(HTTPStatus.INTERNAL_SERVER_ERROR) + assert breaker.state is CircuitState.HALF_OPEN + probe.finish(HTTPStatus.OK) + assert breaker.state is CircuitState.CLOSED + + +def test_stale_closed_era_successes_do_not_close_half_open_circuit() -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=1.0, success_threshold=3, _now=clock) + calls = _HeldCalls(breaker) + stragglers = [calls.start(), calls.start()] + _open_now(breaker) + clock.advance(1.0) + probe = calls.start() + for straggler in stragglers: + straggler.finish(HTTPStatus.OK) + probe.finish(HTTPStatus.OK) + assert breaker.state is CircuitState.HALF_OPEN + + +def test_stragglers_never_let_a_second_probe_in() -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=1.0, success_threshold=2, _now=clock) + calls = _HeldCalls(breaker) + straggler = calls.start() + _open_now(breaker) + clock.advance(1.0) + calls.start() + straggler.finish(HTTPStatus.INTERNAL_SERVER_ERROR) + clock.advance(1.0) + calls.start() + calls.drain() + assert calls.max_probes_in_flight == 1 + + +# ── a RetryBudgetExhaustedError is classified by its last_exception ── + + +def _behind_starved_retry(handler: Callable[[httpx2.Request], httpx2.Response], breaker: CircuitBreaker) -> Client: + return Client( + httpx2_client=httpx2.Client(transport=httpx2.MockTransport(handler)), + middleware=[ + breaker, + Retry( + budget=RetryBudget(min_retries_per_sec=0.0, percent_can_retry=0.0), + base_delay=0.0, + retry_status_codes=frozenset({429, 503}), + ), + ], + ) + + +def test_budget_exhausted_on_5xx_is_a_counted_failure() -> None: + breaker = CircuitBreaker(failure_threshold=1, _now=_Clock()) + with _behind_starved_retry(_StatusSequence([503]), breaker) as client: + with pytest.raises(RetryBudgetExhaustedError): + client.get("https://example.test/x") + with pytest.raises(CircuitOpenError): + client.get("https://example.test/x") + + +def test_budget_exhausted_on_network_error_is_a_counted_failure() -> None: + def _refuse(request: httpx2.Request) -> httpx2.Response: + msg = "refused" + raise httpx2.ConnectError(msg, request=request) + + breaker = CircuitBreaker(failure_threshold=1, _now=_Clock()) + with _behind_starved_retry(_refuse, breaker) as client: + with pytest.raises(RetryBudgetExhaustedError) as info: + client.get("https://example.test/x") + assert isinstance(info.value.last_exception, NetworkError) + with pytest.raises(CircuitOpenError): + client.get("https://example.test/x") + + +def test_budget_exhausted_on_4xx_is_a_success() -> None: + breaker = CircuitBreaker(failure_threshold=2, _now=_Clock()) + with _behind_starved_retry(_StatusSequence([500, 429, 500, 200]), breaker) as client: + with pytest.raises(InternalServerError): + client.get("https://example.test/x") + with pytest.raises(RetryBudgetExhaustedError): + client.get("https://example.test/x") + with pytest.raises(InternalServerError): + client.get("https://example.test/x") + response = client.get("https://example.test/x") + assert response.status_code == HTTPStatus.OK + + +def test_budget_exhausted_with_a_foreign_last_exception_leaves_the_circuit_alone() -> None: + def _exhausted(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001 + raise RetryBudgetExhaustedError(last_response=None, last_exception=ValueError("boom"), attempts=1) + + breaker = CircuitBreaker(failure_threshold=1, _now=_Clock()) + for _ in range(3): + with pytest.raises(RetryBudgetExhaustedError): + breaker(httpx2.Request("GET", "https://example.test/x"), _exhausted) + assert breaker.state is CircuitState.CLOSED + + +# ── circuit.rejected is rate limited per circuit and emitted outside the lock ── + + +def _unreachable(request: httpx2.Request) -> httpx2.Response: # pragma: no cover — OPEN never forwards + return httpx2.Response(HTTPStatus.OK, request=request) + + +def test_rejected_event_is_emitted_at_most_once_per_second(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=30.0, _now=clock) + _open_now(breaker) + with caplog.at_level(logging.WARNING, logger="httpware.circuit_breaker"): + for step in (0.0, 0.0, 0.0, 0.5, 0.49, 0.01, 0.2): + clock.advance(step) + with pytest.raises(CircuitOpenError): + breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) + rejected = [r for r in caplog.records if r.event == "circuit.rejected"] # ty: ignore[unresolved-attribute] + assert [r.suppressed for r in rejected] == [0, 4] # ty: ignore[unresolved-attribute] + + +class _LockProbe(logging.Handler): + def __init__(self, lock: threading.Lock) -> None: + super().__init__() + self._lock = lock + self.held_during: list[bool] = [] + + def emit(self, record: logging.LogRecord) -> None: + if getattr(record, "event", None) == "circuit.rejected": + self.held_during.append(self._lock.locked()) + + +def test_rejected_event_is_emitted_outside_the_lock() -> None: + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=30.0, _now=_Clock()) + _open_now(breaker) + probe = _LockProbe(breaker._lock) # noqa: SLF001 — the lock is the thing under test + logger = logging.getLogger("httpware.circuit_breaker") + logger.addHandler(probe) + try: + with pytest.raises(CircuitOpenError): + breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) + finally: + logger.removeHandler(probe) + assert probe.held_during == [False] diff --git a/tests/test_keyed_circuit_breaker.py b/tests/test_keyed_circuit_breaker.py index aa85de2..a08dfc2 100644 --- a/tests/test_keyed_circuit_breaker.py +++ b/tests/test_keyed_circuit_breaker.py @@ -216,3 +216,45 @@ async def _run_once(url: str) -> None: asyncio.run(_run_once("https://a.test/x")) with pytest.raises(RuntimeError, match="AsyncKeyedCircuitBreaker is bound to a single event loop"): asyncio.run(_run_once("https://b.test/x")) + + +class _Held: + """A `next` that blocks until released, then answers `status`.""" + + def __init__(self, status: int) -> None: + self.status = status + self.entered = asyncio.Event() + self.release = asyncio.Event() + + async def __call__(self, request: httpx2.Request) -> httpx2.Response: + self.entered.set() + await self.release.wait() + if self.status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(self.status, request=request)) + return httpx2.Response(self.status, request=request) + + +async def test_stale_closed_era_failure_does_not_reopen_a_half_open_circuit() -> None: + clock = _Clock() + breaker = AsyncKeyedCircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + request = httpx2.Request("GET", "https://a.test/x") + straggler_next = _Held(HTTPStatus.INTERNAL_SERVER_ERROR) + straggler = asyncio.create_task(breaker(request, straggler_next)) + await straggler_next.entered.wait() + opener = _Held(HTTPStatus.INTERNAL_SERVER_ERROR) + opener.release.set() + with pytest.raises(InternalServerError): + await breaker(request, opener) + clock.t = 1.0 + probe_next = _Held(HTTPStatus.OK) + probe = asyncio.create_task(breaker(request, probe_next)) + await probe_next.entered.wait() + straggler_next.release.set() + with pytest.raises(InternalServerError): + await straggler + probe_next.release.set() + await probe + after = _Held(HTTPStatus.OK) + after.release.set() + response = await breaker(request, after) + assert response.status_code == HTTPStatus.OK diff --git a/tests/test_keyed_circuit_breaker_sync.py b/tests/test_keyed_circuit_breaker_sync.py index edb9810..cbf49e7 100644 --- a/tests/test_keyed_circuit_breaker_sync.py +++ b/tests/test_keyed_circuit_breaker_sync.py @@ -1,5 +1,6 @@ """Tests for the sync KeyedCircuitBreaker middleware. Mirror of test_keyed_circuit_breaker.py.""" +import contextlib import logging import threading from collections.abc import Callable @@ -202,3 +203,82 @@ def test_circuit_key_is_the_normalized_origin_without_userinfo(caplog: pytest.Lo _fail_n(client, "http://a.test:8080/x", 1) keys = [r.circuit_key for r in caplog.records if r.name == "httpware.circuit_breaker"] # ty: ignore[unresolved-attribute] assert keys == ["https://a.test", "http://a.test:8080"] + + +class _Held: + """A `next` that blocks until released, then answers `status`.""" + + def __init__(self, status: int) -> None: + self.status = status + self.entered = threading.Event() + self.release = threading.Event() + + def __call__(self, request: httpx2.Request) -> httpx2.Response: + self.entered.set() + self.release.wait(timeout=5.0) + if self.status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(self.status, request=request)) + return httpx2.Response(self.status, request=request) + + +def _in_thread(breaker: KeyedCircuitBreaker, request: httpx2.Request, held: _Held) -> threading.Thread: + def _run() -> None: + with contextlib.suppress(InternalServerError): + breaker(request, held) + + thread = threading.Thread(target=_run) + thread.start() + assert held.entered.wait(timeout=5.0) + return thread + + +def test_stale_closed_era_failure_does_not_reopen_a_half_open_circuit() -> None: + clock = _Clock() + breaker = KeyedCircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + request = httpx2.Request("GET", "https://a.test/x") + straggler_next = _Held(HTTPStatus.INTERNAL_SERVER_ERROR) + straggler = _in_thread(breaker, request, straggler_next) + opener = _Held(HTTPStatus.INTERNAL_SERVER_ERROR) + opener.release.set() + with pytest.raises(InternalServerError): + breaker(request, opener) + clock.t = 1.0 + probe_next = _Held(HTTPStatus.OK) + probe = _in_thread(breaker, request, probe_next) + straggler_next.release.set() + straggler.join(timeout=5.0) + probe_next.release.set() + probe.join(timeout=5.0) + after = _Held(HTTPStatus.OK) + after.release.set() + response = breaker(request, after) + assert response.status_code == HTTPStatus.OK + + +class _LockProbe(logging.Handler): + def __init__(self, lock: threading.Lock) -> None: + super().__init__() + self._lock = lock + self.held_during: list[bool] = [] + + def emit(self, record: logging.LogRecord) -> None: + if getattr(record, "event", None) == "circuit.rejected": + self.held_during.append(self._lock.locked()) + + +def test_rejected_event_is_emitted_outside_the_lock() -> None: + breaker = KeyedCircuitBreaker(failure_threshold=1, reset_timeout=30.0, _now=_Clock()) + request = httpx2.Request("GET", "https://a.test/x") + opener = _Held(HTTPStatus.INTERNAL_SERVER_ERROR) + opener.release.set() + with pytest.raises(InternalServerError): + breaker(request, opener) + probe = _LockProbe(breaker._lock) # noqa: SLF001 — the lock is the thing under test + logger = logging.getLogger("httpware.circuit_breaker") + logger.addHandler(probe) + try: + with pytest.raises(CircuitOpenError): + breaker(request, _Held(HTTPStatus.OK)) + finally: + logger.removeHandler(probe) + assert probe.held_during == [False] From ed8e87d49c68e835854eea18c9e9dadc5b231674 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 11:50:58 +0300 Subject: [PATCH 2/3] test: keep the async straggler harness covered on Python 3.11 On 3.11, awaiting a task that ended with an exception drops coverage line events in the awaiting frame until its next suspension. The harness tasks now swallow the breaker's error themselves. --- tests/test_circuit_breaker.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/tests/test_circuit_breaker.py b/tests/test_circuit_breaker.py index f5d5eb4..d065c19 100644 --- a/tests/test_circuit_breaker.py +++ b/tests/test_circuit_breaker.py @@ -717,7 +717,7 @@ class _HeldCalls: def __init__(self, breaker: AsyncCircuitBreaker) -> None: self._breaker = breaker self._futures: list[asyncio.Future[int]] = [] - self._tasks: list[asyncio.Task[httpx2.Response]] = [] + self._tasks: list[asyncio.Task[None]] = [] self.probes_in_flight = 0 self.max_probes_in_flight = 0 @@ -737,23 +737,29 @@ async def _next(self, request: httpx2.Request) -> httpx2.Response: raise InternalServerError(httpx2.Response(status, request=request)) return httpx2.Response(status, request=request) + async def _call(self) -> None: + with contextlib.suppress(InternalServerError, CircuitOpenError): + await self._breaker(httpx2.Request("GET", "https://example.test/x"), self._next) + async def start(self) -> int: - """Start a call; return its index once it is blocked in `next` (or already rejected).""" - task = asyncio.create_task(self._breaker(httpx2.Request("GET", "https://example.test/x"), self._next)) - self._tasks.append(task) + """Start a call; return its index once it is blocked in `next` (or already rejected). + + The task swallows the breaker's error itself: on 3.11, awaiting a task that failed loses + coverage line events in the awaiting test. + """ + self._tasks.append(asyncio.create_task(self._call())) await asyncio.sleep(0) return len(self._tasks) - 1 async def finish(self, index: int, status: int) -> None: self._futures[index].set_result(status) - with contextlib.suppress(InternalServerError): - await self._tasks[index] + await self._tasks[index] async def drain(self) -> None: for future in self._futures: if not future.done(): future.set_result(HTTPStatus.OK) - await asyncio.gather(*self._tasks, return_exceptions=True) + await asyncio.gather(*self._tasks) async def _open_now(breaker: AsyncCircuitBreaker) -> None: From 7ba2f2ebbf552e4360fd4bb3e34e64680a6e9296 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 12:07:05 +0300 Subject: [PATCH 3/3] fix: emit circuit breaker events outside the lock and count rejections per outage Breaker transitions queue their events and the drivers emit them after releasing the sync lock, so a logging handler that sends a request through the same breaker no longer deadlocks. Rejection accounting restarts with each outage: the first rejection is always emitted and circuit.closed reports the outage's total in a new rejected attribute. --- docs/observability.md | 2 +- docs/resilience.md | 8 +- .../middleware/resilience/circuit_breaker.py | 143 +++++++++++----- tests/test_circuit_breaker.py | 82 ++++++++- tests/test_circuit_breaker_sync.py | 159 +++++++++++++++++- tests/test_keyed_circuit_breaker_sync.py | 41 +++++ 6 files changed, 385 insertions(+), 50 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index ed0b3af..36c030c 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -11,7 +11,7 @@ The logger and event names below are a stable public contract: | `httpware.circuit_breaker` | `circuit.opened` (WARNING), `circuit.rejected` (WARNING), `circuit.half_open` (INFO), `circuit.closed` (INFO) | | `httpware.timeout` | `timeout.exceeded` (WARNING) | -Each log record has an `event` field holding the event name, such as `event="circuit.opened"`, which you can filter on in your log aggregator. `circuit.rejected` is emitted at most once per second per circuit, so an open circuit under heavy traffic does not flood your logs; its `suppressed` attribute counts the rejections that were left out since the previous one. Events from `AsyncKeyedCircuitBreaker` and `KeyedCircuitBreaker` also carry `circuit_key`, the circuit they belong to. [Resilience](resilience.md) describes when each event fires. +Each log record has an `event` field holding the event name, such as `event="circuit.opened"`, which you can filter on in your log aggregator. `circuit.rejected` is emitted at most once per second per circuit, so an open circuit under heavy traffic does not flood your logs. The first rejection of each outage is always emitted, `suppressed` counts the rejections left out since the previous one, and `circuit.closed` reports the outage's total in `rejected`. Events from `AsyncKeyedCircuitBreaker` and `KeyedCircuitBreaker` also carry `circuit_key`, the circuit they belong to. [Resilience](resilience.md) describes when each event fires. ```python import logging diff --git a/docs/resilience.md b/docs/resilience.md index b03ae63..a9d1f72 100644 --- a/docs/resilience.md +++ b/docs/resilience.md @@ -193,9 +193,13 @@ Emitted on logger `httpware.circuit_breaker`: | Event | When | |---|---| | `circuit.opened` | The failure threshold was reached and the circuit went from CLOSED to OPEN. | -| `circuit.rejected` | A request failed fast because the circuit was OPEN or a HALF_OPEN probe was already running. Emitted at most once per second per circuit; its `suppressed` attribute counts the rejections since the previous `circuit.rejected` that were not emitted. | +| `circuit.rejected` | A request failed fast because the circuit was OPEN or a HALF_OPEN probe was already running. Emitted at most once per second per circuit, and always for the first rejection of an outage. Its `suppressed` attribute counts the rejections in the same outage that were not emitted since the previous `circuit.rejected`. | | `circuit.half_open` | The reset timeout passed and the circuit went from OPEN to HALF_OPEN. | -| `circuit.closed` | The success threshold was reached and the circuit went from HALF_OPEN to CLOSED. | +| `circuit.closed` | The success threshold was reached and the circuit went from HALF_OPEN to CLOSED. Its `rejected` attribute is the number of requests rejected during the outage, emitted or not. | + +An outage runs from the moment the circuit opens until it closes again; a probe failure that reopens it stays in the same outage. + +The breaker emits its events after releasing its internal lock, so a logging handler may send requests through a client that uses the same breaker. ### Time-based failure-rate mode diff --git a/src/httpware/middleware/resilience/circuit_breaker.py b/src/httpware/middleware/resilience/circuit_breaker.py index 59fcdc9..46e4331 100644 --- a/src/httpware/middleware/resilience/circuit_breaker.py +++ b/src/httpware/middleware/resilience/circuit_breaker.py @@ -22,8 +22,10 @@ An outcome applies only to the state its request was admitted under: a request still in flight when the circuit changes state finishes without touching the circuit. -circuit.rejected is emitted at most once per second per circuit; its `suppressed` -attribute counts the rejections not emitted since the previous one. +An outage episode runs from CLOSED to OPEN until the circuit is CLOSED again. Within one, +circuit.rejected is emitted at most once per second; its `suppressed` attribute counts the +rejections not emitted since the previous one, and the first rejection of an episode is +always emitted. circuit.closed carries `rejected`, the episode's total rejection count. Trip modes: Classic (default) — opens when consecutive counted-failures reach failure_threshold. @@ -37,8 +39,9 @@ The lock-free _CircuitBreakerState holds the transition logic, shared by both wrappers. AsyncCircuitBreaker relies on asyncio atomicity (no await inside a transition) plus a single-event-loop guard; CircuitBreaker (sync) serializes transitions with a -threading.Lock. Both are sharable across clients (one shared circuit); a sync instance -cannot be shared with an async one. +threading.Lock and emits events only after releasing it, so a logging handler may send +requests through the same breaker. Both are sharable across clients (one shared circuit); +a sync instance cannot be shared with an async one. """ import asyncio @@ -156,21 +159,32 @@ def clear(self) -> None: self._failure = [0] * _BUCKET_COUNT +_Event = tuple[str, int, str, dict[str, typing.Any]] + + +class _HalfOpened(Exception): # noqa: N818 + """Raised by `admit` when it moved the circuit to HALF_OPEN; carries the probe's ticket and events.""" + + def __init__(self, ticket: int, events: list[_Event] | None) -> None: + self.ticket = ticket + self.events = events + + class _RejectionError(Exception): - """Raised by `admit` to refuse a request; `message` is None when the event is suppressed.""" + """Raised by `admit` to refuse a request, carrying the events to emit once the lock is released.""" - def __init__(self, retry_after: float | None, message: str | None, suppressed: int) -> None: + def __init__(self, retry_after: float | None, events: list[_Event] | None) -> None: self.retry_after = retry_after - self.message = message - self.suppressed = suppressed + self.events = events class _CircuitBreakerState: """Lock-free circuit-breaker state machine shared by the sync + async wrappers. - Every method is synchronous and performs no I/O beyond logging. The async wrapper - calls these directly (atomic under a single event loop because no await occurs - inside a transition); the sync wrapper wraps `admit` and `record` in a threading.Lock. + Every method is synchronous and performs no I/O. The async wrapper calls these directly + (atomic under a single event loop because no await occurs inside a transition); the sync + wrapper wraps `admit` and `record` in a threading.Lock. Transitions queue their events; + the caller takes them with the outcome and emits them outside any lock. `admit` hands out a ticket: the epoch, which every transition advances by two, plus one for the probe. `record` ignores a ticket from an earlier epoch. """ @@ -223,6 +237,8 @@ def __init__( # noqa: PLR0913 — breaker state has many orthogonal knobs; a da self._epoch = 0 self._rejected_emitted_at = -math.inf self._rejections_suppressed = 0 + self._episode_rejections = 0 + self._events: list[_Event] | None = None @property def state(self) -> CircuitState: @@ -230,18 +246,20 @@ def state(self) -> CircuitState: return self._state def admit(self, request: httpx2.Request) -> int: - """Return the request's ticket, or raise _RejectionError. No await inside.""" + """Return the request's ticket, or raise _RejectionError or _HalfOpened. No await inside.""" if self._state is CircuitState.CLOSED: return self._epoch now = self._now() if self._state is CircuitState.OPEN: elapsed = now - self._opened_at if elapsed < self._reset_timeout: - self._reject(now, max(0.0, self._reset_timeout - elapsed), "circuit open — rejecting request") + self._reject(request, now, max(0.0, self._reset_timeout - elapsed), "circuit open — rejecting request") self._move_to(CircuitState.HALF_OPEN) - self._emit(request, "circuit.half_open", logging.INFO, "circuit half-open — admitting probe", {}) - elif self._probe_in_flight: - self._reject(now, None, "circuit half-open — rejecting request (probe in flight)") + self._probe_in_flight = True + self._queue(request, "circuit.half_open", logging.INFO, "circuit half-open — admitting probe", {}) + raise _HalfOpened(self._epoch + _PROBE, self.take_events()) + if self._probe_in_flight: + self._reject(request, now, None, "circuit half-open — rejecting request (probe in flight)") self._probe_in_flight = True return self._epoch + _PROBE @@ -255,8 +273,11 @@ def classify(self, exc: BaseException | None) -> int: return _FAILURE return _FOREIGN - def record(self, ticket: int, outcome: int, request: httpx2.Request) -> None: - """Apply an admitted request's outcome; a no-op if the circuit has transitioned since.""" + def record(self, ticket: int, outcome: int, request: httpx2.Request) -> list[_Event] | None: + """Apply an admitted request's outcome and return the events it queued. + + A no-op if the circuit has transitioned since the ticket was issued. + """ if ticket == self._epoch: if outcome == _SUCCESS and not self._rate_mode: self._consecutive_failures = 0 @@ -268,26 +289,26 @@ def record(self, ticket: int, outcome: int, request: httpx2.Request) -> None: self._record_probe_success(request) elif outcome == _FAILURE: self._open(request, failures=1) # 1 = the single probe failure that re-opened the circuit - - def rejected(self, request: httpx2.Request, rejection: _RejectionError) -> CircuitOpenError: - """Emit the rejection's event unless `admit` suppressed it; return the error to raise.""" - if rejection.message is not None: - self._emit( - request, - "circuit.rejected", - logging.WARNING, - rejection.message, - {"retry_after": rejection.retry_after, "suppressed": rejection.suppressed}, - ) - return CircuitOpenError(retry_after=rejection.retry_after) - - def _reject(self, now: float, retry_after: float | None, message: str) -> typing.NoReturn: + events = self._events + if events is not None: + self._events = None + return events + + def take_events(self) -> list[_Event] | None: + """Return and clear the queued events.""" + events, self._events = self._events, None + return events + + def _reject(self, request: httpx2.Request, now: float, retry_after: float | None, message: str) -> typing.NoReturn: + self._episode_rejections += 1 if now - self._rejected_emitted_at < _REJECTED_EVENT_INTERVAL: self._rejections_suppressed += 1 - raise _RejectionError(retry_after, None, 0) + raise _RejectionError(retry_after, None) self._rejected_emitted_at = now suppressed, self._rejections_suppressed = self._rejections_suppressed, 0 - raise _RejectionError(retry_after, message, suppressed) + attributes = {"retry_after": retry_after, "suppressed": suppressed} + self._queue(request, "circuit.rejected", logging.WARNING, message, attributes) + raise _RejectionError(retry_after, self.take_events()) def _record_closed(self, request: httpx2.Request, *, failed: bool) -> None: if self._rate_mode: @@ -305,18 +326,23 @@ def _record_probe_success(self, request: httpx2.Request) -> None: self._consecutive_successes = 0 if self._rate_mode: self._window.clear() # ty: ignore[unresolved-attribute] - self._emit(request, "circuit.closed", logging.INFO, "circuit closed — service recovered", {}) + attributes = {"rejected": self._episode_rejections} + self._queue(request, "circuit.closed", logging.INFO, "circuit closed — service recovered", attributes) def _move_to(self, state: CircuitState) -> None: self._state = state self._epoch += _EPOCH_STEP def _enter_open(self, request: httpx2.Request, message: str, attributes: dict[str, typing.Any]) -> None: + if self._state is CircuitState.CLOSED: + self._rejected_emitted_at = -math.inf + self._rejections_suppressed = 0 + self._episode_rejections = 0 self._move_to(CircuitState.OPEN) self._opened_at = self._now() self._consecutive_failures = 0 self._consecutive_successes = 0 - self._emit(request, "circuit.opened", logging.WARNING, message, attributes) + self._queue(request, "circuit.opened", logging.WARNING, message, attributes) def _open(self, request: httpx2.Request, *, failures: int) -> None: self._enter_open( @@ -346,7 +372,7 @@ def _record_outcome(self, request: httpx2.Request, *, failed: bool) -> None: if threshold is not None and total >= self._minimum_calls and failures / total >= threshold: self._open_rate(request, total=total, failures=failures) - def _emit( + def _queue( self, request: httpx2.Request, event_name: str, @@ -357,20 +383,45 @@ def _emit( attributes = {**attributes, "method": request.method, "url": str(request.url)} if self._key is not None: attributes["circuit_key"] = self._key + if self._events is None: + self._events = [] + self._events.append((event_name, level, message, attributes)) + + +def _emit_events(events: list[_Event]) -> None: + for event_name, level, message, attributes in events: _emit_event(_LOGGER, event_name, level=level, message=message, attributes=attributes) +def _half_opened(half_opened: _HalfOpened) -> int: + if half_opened.events is not None: + _emit_events(half_opened.events) + return half_opened.ticket + + +def _rejected(rejection: _RejectionError) -> CircuitOpenError: + if rejection.events is not None: + _emit_events(rejection.events) + return CircuitOpenError(retry_after=rejection.retry_after) + + async def _call_async(state: _CircuitBreakerState, request: httpx2.Request, next: AsyncNext) -> httpx2.Response: # noqa: A002 try: ticket = state.admit(request) except _RejectionError as rejection: - raise state.rejected(request, rejection) from None + raise _rejected(rejection) from None + except _HalfOpened as half_opened: + ticket = _half_opened(half_opened) try: response = await next(request) except BaseException as exc: - state.record(ticket, state.classify(exc), request) + events = state.record(ticket, state.classify(exc), request) + if events is not None: + _emit_events(events) raise - state.record(ticket, _SUCCESS, request) + events = state.record(ticket, _SUCCESS, request) + if events is not None: + _emit_events(events) return response @@ -384,16 +435,22 @@ def _call_sync( with lock: ticket = state.admit(request) except _RejectionError as rejection: - raise state.rejected(request, rejection) from None + raise _rejected(rejection) from None + except _HalfOpened as half_opened: + ticket = _half_opened(half_opened) try: response = next(request) except BaseException as exc: outcome = state.classify(exc) with lock: - state.record(ticket, outcome, request) + events = state.record(ticket, outcome, request) + if events is not None: + _emit_events(events) raise with lock: - state.record(ticket, _SUCCESS, request) + events = state.record(ticket, _SUCCESS, request) + if events is not None: + _emit_events(events) return response diff --git a/tests/test_circuit_breaker.py b/tests/test_circuit_breaker.py index d065c19..476d6b7 100644 --- a/tests/test_circuit_breaker.py +++ b/tests/test_circuit_breaker.py @@ -744,8 +744,7 @@ async def _call(self) -> None: async def start(self) -> int: """Start a call; return its index once it is blocked in `next` (or already rejected). - The task swallows the breaker's error itself: on 3.11, awaiting a task that failed loses - coverage line events in the awaiting test. + The task swallows the breaker's error itself; on 3.11 awaiting a failed task loses coverage. """ self._tasks.append(asyncio.create_task(self._call())) await asyncio.sleep(0) @@ -897,3 +896,82 @@ async def _unreachable(request: httpx2.Request) -> httpx2.Response: # pragma: n await breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) rejected = [r for r in caplog.records if r.event == "circuit.rejected"] # ty: ignore[unresolved-attribute] assert [r.suppressed for r in rejected] == [0, 4] # ty: ignore[unresolved-attribute] + + +# ── rejection accounting runs per outage episode ── + + +async def _ok(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(HTTPStatus.OK, request=request) + + +async def _reject(breaker: AsyncCircuitBreaker) -> None: + async def _unreachable(request: httpx2.Request) -> httpx2.Response: # pragma: no cover — OPEN never forwards + return httpx2.Response(HTTPStatus.OK, request=request) + + with pytest.raises(CircuitOpenError): + await breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) + + +def _events(caplog: pytest.LogCaptureFixture, name: str) -> list[logging.LogRecord]: + return [r for r in caplog.records if getattr(r, "event", None) == name] + + +async def test_first_rejection_of_a_new_outage_is_emitted(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=0.1, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + await _open_now(breaker) + await _reject(breaker) + clock.advance(0.1) + await breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + await _open_now(breaker) + await _reject(breaker) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 0] # ty: ignore[unresolved-attribute] + + +async def test_closed_event_reports_every_rejection_of_the_outage(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=5.0, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + await _open_now(breaker) + await _reject(breaker) + await _reject(breaker) + clock.advance(5.0) + await _open_now(breaker) + await _reject(breaker) + await _reject(breaker) + clock.advance(5.0) + await breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 1] # ty: ignore[unresolved-attribute] + assert [r.rejected for r in _events(caplog, "circuit.closed")] == [4] # ty: ignore[unresolved-attribute] + + +async def test_suppressed_never_counts_a_previous_outage(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=5.0, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + await _open_now(breaker) + for _ in range(3): + await _reject(breaker) + clock.advance(5.0) + await breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + await _open_now(breaker) + await _reject(breaker) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 0] # ty: ignore[unresolved-attribute] + + +async def test_events_of_one_call_arrive_in_transition_order(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = AsyncCircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + await _open_now(breaker) + await _reject(breaker) + clock.advance(1.0) + await breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + assert [r.event for r in caplog.records] == [ # ty: ignore[unresolved-attribute] + "circuit.opened", + "circuit.rejected", + "circuit.half_open", + "circuit.closed", + ] diff --git a/tests/test_circuit_breaker_sync.py b/tests/test_circuit_breaker_sync.py index 1b5f4c0..7c7eb36 100644 --- a/tests/test_circuit_breaker_sync.py +++ b/tests/test_circuit_breaker_sync.py @@ -13,6 +13,7 @@ CircuitOpenError, CircuitState, Client, + ClientError, InternalServerError, NetworkError, NotFoundError, @@ -801,13 +802,17 @@ def test_rejected_event_is_emitted_at_most_once_per_second(caplog: pytest.LogCap class _LockProbe(logging.Handler): - def __init__(self, lock: threading.Lock) -> None: + """Records whether `lock` was held when each breaker event in `events` (None: all) arrived.""" + + def __init__(self, lock: threading.Lock, events: frozenset[str] | None = frozenset({"circuit.rejected"})) -> None: super().__init__() self._lock = lock + self._events = events self.held_during: list[bool] = [] def emit(self, record: logging.LogRecord) -> None: - if getattr(record, "event", None) == "circuit.rejected": + event = getattr(record, "event", None) + if event is not None and (self._events is None or event in self._events): self.held_during.append(self._lock.locked()) @@ -823,3 +828,153 @@ def test_rejected_event_is_emitted_outside_the_lock() -> None: finally: logger.removeHandler(probe) assert probe.held_during == [False] + + +class _CallOnEvent(logging.Handler): + """Sends one request through `call` from inside every breaker event, like a log shipper.""" + + def __init__(self, call: Callable[[], object]) -> None: + super().__init__() + self._call = call + self._active = threading.local() + self.events: list[str] = [] + + def handle(self, record: logging.LogRecord) -> bool: + """Emit without taking the handler lock, so a deadlocked call cannot hang logging.shutdown.""" + self.emit(record) + return True + + def emit(self, record: logging.LogRecord) -> None: + self.events.append(getattr(record, "event", "")) + if getattr(self._active, "on", False): + return + self._active.on = True + try: + with contextlib.suppress(ClientError): + self._call() + finally: + self._active.on = False + + +def _in_thread_with_handler(handler: logging.Handler, target: Callable[[], object]) -> bool: + """Run `target` on a daemon thread with `handler` attached; return whether it finished in time.""" + logger = logging.getLogger("httpware.circuit_breaker") + logger.addHandler(handler) + previous = logger.level + logger.setLevel(logging.INFO) + + def _run() -> None: + with contextlib.suppress(ClientError): + target() + + thread = threading.Thread(target=_run, daemon=True) + try: + thread.start() + thread.join(timeout=5.0) + finally: + logger.removeHandler(handler) + logger.setLevel(previous) + return not thread.is_alive() + + +def test_event_handler_may_send_through_the_same_breaker() -> None: + clock = _Clock() + handler = _StatusSequence([500]) + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + client = _client(handler, breaker=breaker) + shipper = _CallOnEvent(lambda: client.get("https://logs.example.test/ingest")) + + def _outage_and_recovery() -> None: + with contextlib.suppress(InternalServerError): + client.get("https://example.test/x") + clock.advance(1.0) + client.get("https://example.test/x") + + with client: + assert _in_thread_with_handler(shipper, _outage_and_recovery) + assert shipper.events == [ + "circuit.opened", + "circuit.rejected", + "circuit.half_open", + "circuit.rejected", + "circuit.closed", + ] + + +def test_every_event_is_emitted_outside_the_lock() -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=1.0, _now=clock) + probe = _LockProbe(breaker._lock, events=None) # noqa: SLF001 — the lock is the thing under test + logger = logging.getLogger("httpware.circuit_breaker") + logger.addHandler(probe) + previous = logger.level + logger.setLevel(logging.INFO) + try: + _open_now(breaker) + _reject(breaker) + clock.advance(1.0) + breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + finally: + logger.removeHandler(probe) + logger.setLevel(previous) + assert probe.held_during == [False, False, False, False] + + +# ── rejection accounting runs per outage episode ── + + +def _ok(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(HTTPStatus.OK, request=request) + + +def _reject(breaker: CircuitBreaker) -> None: + with pytest.raises(CircuitOpenError): + breaker(httpx2.Request("GET", "https://example.test/x"), _unreachable) + + +def _events(caplog: pytest.LogCaptureFixture, name: str) -> list[logging.LogRecord]: + return [r for r in caplog.records if getattr(r, "event", None) == name] + + +def test_first_rejection_of_a_new_outage_is_emitted(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=0.1, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + _open_now(breaker) + _reject(breaker) + clock.advance(0.1) + breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + _open_now(breaker) + _reject(breaker) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 0] # ty: ignore[unresolved-attribute] + + +def test_closed_event_reports_every_rejection_of_the_outage(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=5.0, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + _open_now(breaker) + _reject(breaker) + _reject(breaker) + clock.advance(5.0) + _open_now(breaker) + _reject(breaker) + _reject(breaker) + clock.advance(5.0) + breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 1] # ty: ignore[unresolved-attribute] + assert [r.rejected for r in _events(caplog, "circuit.closed")] == [4] # ty: ignore[unresolved-attribute] + + +def test_suppressed_never_counts_a_previous_outage(caplog: pytest.LogCaptureFixture) -> None: + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=5.0, _now=clock) + with caplog.at_level(logging.INFO, logger="httpware.circuit_breaker"): + _open_now(breaker) + for _ in range(3): + _reject(breaker) + clock.advance(5.0) + breaker(httpx2.Request("GET", "https://example.test/x"), _ok) + _open_now(breaker) + _reject(breaker) + assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 0] # ty: ignore[unresolved-attribute] diff --git a/tests/test_keyed_circuit_breaker_sync.py b/tests/test_keyed_circuit_breaker_sync.py index cbf49e7..305c758 100644 --- a/tests/test_keyed_circuit_breaker_sync.py +++ b/tests/test_keyed_circuit_breaker_sync.py @@ -12,6 +12,7 @@ from httpware import ( CircuitOpenError, Client, + ClientError, InternalServerError, KeyedCircuitBreaker, NetworkError, @@ -282,3 +283,43 @@ def test_rejected_event_is_emitted_outside_the_lock() -> None: finally: logger.removeHandler(probe) assert probe.held_during == [False] + + +class _CallOnEvent(logging.Handler): + """Sends a request through `call` from inside every breaker event, like a log shipper.""" + + def __init__(self, call: Callable[[], object]) -> None: + super().__init__() + self._call = call + + def handle(self, record: logging.LogRecord) -> bool: + """Emit without taking the handler lock, so a deadlocked call cannot hang logging.shutdown.""" + self.emit(record) + return True + + def emit(self, record: logging.LogRecord) -> None: # noqa: ARG002 + with contextlib.suppress(ClientError): + self._call() + + +def test_event_handler_may_send_to_another_host_through_the_same_breaker() -> None: + handler = _PerHost(failing={"a.test"}) + breaker = KeyedCircuitBreaker(failure_threshold=1, _now=_Clock()) + client = _client(handler, breaker=breaker) + shipper = _CallOnEvent(lambda: client.get("https://logs.test/ingest")) + logger = logging.getLogger("httpware.circuit_breaker") + logger.addHandler(shipper) + + def _fail_once() -> None: + with contextlib.suppress(InternalServerError): + client.get("https://a.test/x") + + thread = threading.Thread(target=_fail_once, daemon=True) + try: + with client: + thread.start() + thread.join(timeout=5.0) + finally: + logger.removeHandler(shipper) + assert not thread.is_alive() + assert handler.calls == ["a.test", "logs.test"]