From 12aa5237d040cba7e6b76633735012487649019b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Mon, 13 Jul 2026 23:11:15 +0300 Subject: [PATCH] refactor(resilience): share single-event-loop guard between bulkhead/circuit_breaker AsyncBulkhead._check_loop and AsyncCircuitBreaker._check_loop were byte-identical double-checked-locking logic, differing only in the cross-loop error message. Extracted into check_event_loop in a new _event_loop_guard.py, closing the last item from the 2026-06-14 deep audit's "sync/async duplication, no divergence yet" list not yet addressed by this session's refactor series. --- architecture/overview.md | 3 +- .../2026-07-13.10-event-loop-guard-shared.md | 87 +++++++++++++++++++ .../resilience/_event_loop_guard.py | 37 ++++++++ .../middleware/resilience/bulkhead.py | 25 ++---- .../middleware/resilience/circuit_breaker.py | 21 ++--- 5 files changed, 140 insertions(+), 33 deletions(-) create mode 100644 planning/changes/2026-07-13.10-event-loop-guard-shared.md create mode 100644 src/httpware/middleware/resilience/_event_loop_guard.py diff --git a/architecture/overview.md b/architecture/overview.md index cad77b5..91d8386 100644 --- a/architecture/overview.md +++ b/architecture/overview.md @@ -32,7 +32,8 @@ src/httpware/ │ ├── retry.py # Retry + AsyncRetry │ ├── timeout.py # AsyncTimeout │ ├── circuit_breaker.py # CircuitBreaker + AsyncCircuitBreaker -│ └── _backoff.py # full-jitter helper (shared) +│ ├── _backoff.py # full-jitter helper (shared) +│ └── _event_loop_guard.py # single-event-loop guard (shared: AsyncBulkhead + AsyncCircuitBreaker) ├── decoders/ # shared (ResponseDecoder + adapters) └── _internal/ ├── body_cap.py # max_response_body_bytes: validate, read-capped (sync+async) diff --git a/planning/changes/2026-07-13.10-event-loop-guard-shared.md b/planning/changes/2026-07-13.10-event-loop-guard-shared.md new file mode 100644 index 0000000..ba24ab1 --- /dev/null +++ b/planning/changes/2026-07-13.10-event-loop-guard-shared.md @@ -0,0 +1,87 @@ +--- +summary: Share AsyncBulkhead/AsyncCircuitBreaker's identical single-event-loop guard via one module-level check_event_loop function in a new _event_loop_guard.py. +--- + +# Change: Share the single-event-loop guard between AsyncBulkhead and AsyncCircuitBreaker + +**Lane:** lightweight — 4 files (above the usual ≤2 guard; one is the +`architecture/overview.md` module-layout promotion this change requires, per +the `2026-06-13.04`/`2026-07-13.07` precedent of the file-count guard proxying +*code* risk, not file count). No public-API change, no test changes (existing +suite is the parity net). + +Spec: [`planning/audits/2026-06-14-deep-audit.md`](../../audits/2026-06-14-deep-audit.md) +("Sync/async duplication, no divergence yet" — `_check_loop` in bulkhead vs +circuit_breaker), the last item from that list not yet addressed by this +session's refactor series (`2026-07-13.01`–`.06`). + +## Goal + +`AsyncBulkhead._check_loop` (`bulkhead.py:103-121`) and +`AsyncCircuitBreaker._check_loop` (`circuit_breaker.py:340-354`) are +byte-identical double-checked-locking logic — same structure, differing only +in which cross-loop-message constant they format. Unlike the other items in +this refactor series, this pair lives in two different classes across two +different files, not a sync/async split within one file. No behavior change. + +## Approach + +New module `src/httpware/middleware/resilience/_event_loop_guard.py` +(sibling of the existing `_backoff.py` shared helper) with one function: + +```python +def check_event_loop( + get_loop: Callable[[], asyncio.AbstractEventLoop | None], + set_loop: Callable[[asyncio.AbstractEventLoop], None], + loop_lock: threading.Lock, + message_template: str, +) -> None: ... +``` + +`get_loop`/`set_loop` are closures over the caller's `self._loop`, not a +passed-in `self` — passing `self` and reaching into `instance._loop` from a +free function would trip ruff's `SLF001` (private-member access from outside +the class), which is exactly why `decoders/_caching.py`'s `_get_or_build` +precedent takes the mutable container by value instead of the owning object. +The inner check re-invokes `get_loop()` rather than reusing the outer +snapshot, preserving the original's double-checked-locking correctness: a +thread that loses the race to acquire `loop_lock` must still observe +whichever loop the winner just bound, not its own stale pre-lock read +(load-bearing, per the 2026-06-14 deep audit's own note that this is a +correctness mechanism, not a decorative optimization). + +Each `_check_loop` method becomes a 5-line call: + +```python +def _check_loop(self) -> None: + check_event_loop( + lambda: self._loop, + lambda loop: setattr(self, "_loop", loop), + self._loop_lock, + _ASYNCBULKHEAD_CROSS_LOOP_MSG, # or _CROSS_LOOP_MSG in circuit_breaker.py + ) +``` + +## Files + +- `src/httpware/middleware/resilience/_event_loop_guard.py` — new, the shared + `check_event_loop` function. +- `src/httpware/middleware/resilience/bulkhead.py` — `AsyncBulkhead._check_loop` + now delegates. +- `src/httpware/middleware/resilience/circuit_breaker.py` — + `AsyncCircuitBreaker._check_loop` now delegates. +- `architecture/overview.md` — module-layout tree gets the new file, matching + `_backoff.py`'s existing entry. +- No test file changes — `tests/test_bulkhead.py::test_cross_loop_acquire_raises_runtimeerror` + and `tests/test_circuit_breaker.py::test_cross_loop_use_raises_runtimeerror` + (plus the same-loop and construct-outside-loop tests) already exercise both + extraction targets and stay green unmodified. + +## Verification + +- [x] `just test` — full suite green, 100% coverage maintained, same test + count (pure refactor, no new/removed tests). +- [x] `just lint-ci` — clean. +- [x] Grep both files to confirm `_check_loop`'s body no longer contains + `asyncio.get_running_loop()` or the inline double-checked-locking block + — only the 5-line delegating call remains. diff --git a/src/httpware/middleware/resilience/_event_loop_guard.py b/src/httpware/middleware/resilience/_event_loop_guard.py new file mode 100644 index 0000000..043ea4d --- /dev/null +++ b/src/httpware/middleware/resilience/_event_loop_guard.py @@ -0,0 +1,37 @@ +"""Single-event-loop guard for async resilience middleware (private, shared).""" + +import asyncio +import threading +from collections.abc import Callable + + +def check_event_loop( + get_loop: Callable[[], asyncio.AbstractEventLoop | None], + set_loop: Callable[[asyncio.AbstractEventLoop], None], + loop_lock: threading.Lock, + message_template: str, +) -> None: + """Bind the caller to the first event loop that calls it. + + Raises RuntimeError on a later call from a different loop. `get_loop`/`set_loop` + read and write the caller's cached-loop attribute. The inner check re-reads via + `get_loop()` rather than reusing the outer snapshot, so a thread that loses the + race to acquire `loop_lock` still sees whichever loop the winner just bound + (double-checked locking) — the outer unlocked read handles the common + already-bound case without lock overhead. + """ + current = asyncio.get_running_loop() + cached = get_loop() + if cached is current: + return + if cached is not None: + raise RuntimeError(message_template.format(first=cached, current=current)) + with loop_lock: + cached = get_loop() + if cached is None: + set_loop(current) + # pragma below: inner double-check-with-lock race arm; only reachable when + # two threads simultaneously pass the outer check, which single-threaded + # tests can't trigger. + elif cached is not current: # pragma: no cover + raise RuntimeError(message_template.format(first=cached, current=current)) diff --git a/src/httpware/middleware/resilience/bulkhead.py b/src/httpware/middleware/resilience/bulkhead.py index 927a3db..8b56949 100644 --- a/src/httpware/middleware/resilience/bulkhead.py +++ b/src/httpware/middleware/resilience/bulkhead.py @@ -27,6 +27,7 @@ from httpware._internal.observability import _emit_event from httpware.errors import BulkheadFullError from httpware.middleware import AsyncNext, Next +from httpware.middleware.resilience._event_loop_guard import check_event_loop _MAX_CONCURRENT_INVALID = "max_concurrent must be >= 1" @@ -101,24 +102,12 @@ def __init__( self._loop_lock = threading.Lock() def _check_loop(self) -> None: - current = asyncio.get_running_loop() - cached = self._loop - if cached is current: - return - if cached is not None: - raise RuntimeError( - _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=cached, current=current), - ) - with self._loop_lock: - if self._loop is None: - self._loop = current - # pragma below: inner double-check-with-lock race arm; only - # reachable when two threads simultaneously pass the outer - # cached-loop check, which single-threaded tests can't trigger. - elif self._loop is not current: # pragma: no cover - raise RuntimeError( - _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=self._loop, current=current), - ) + check_event_loop( + lambda: self._loop, + lambda loop: setattr(self, "_loop", loop), + self._loop_lock, + _ASYNCBULKHEAD_CROSS_LOOP_MSG, + ) async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response: # noqa: A002 """Acquire a slot (bounded by acquire_timeout), invoke next, release.""" diff --git a/src/httpware/middleware/resilience/circuit_breaker.py b/src/httpware/middleware/resilience/circuit_breaker.py index 3017dbb..a254b73 100644 --- a/src/httpware/middleware/resilience/circuit_breaker.py +++ b/src/httpware/middleware/resilience/circuit_breaker.py @@ -46,6 +46,7 @@ from httpware._internal.observability import _emit_event from httpware.errors import CircuitOpenError, NetworkError, StatusError, TimeoutError # noqa: A004 from httpware.middleware import AsyncNext, Next +from httpware.middleware.resilience._event_loop_guard import check_event_loop _FAILURE_THRESHOLD_INVALID = "failure_threshold must be >= 1" @@ -338,20 +339,12 @@ def __init__( # noqa: PLR0913 — breaker has many orthogonal knobs; a dataclas self._loop_lock = threading.Lock() def _check_loop(self) -> None: - current = asyncio.get_running_loop() - cached = self._loop - if cached is current: - return - if cached is not None: - raise RuntimeError(_CROSS_LOOP_MSG.format(first=cached, current=current)) - with self._loop_lock: - if self._loop is None: - self._loop = current - # pragma below: inner double-check-with-lock race arm; only reachable when - # two threads simultaneously pass the outer check, which single-threaded - # tests can't trigger. - elif self._loop is not current: # pragma: no cover - raise RuntimeError(_CROSS_LOOP_MSG.format(first=self._loop, current=current)) + check_event_loop( + lambda: self._loop, + lambda loop: setattr(self, "_loop", loop), + self._loop_lock, + _CROSS_LOOP_MSG, + ) @property def state(self) -> CircuitState: