Skip to content

Commit 12aa523

Browse files
committed
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.
1 parent 144521b commit 12aa523

5 files changed

Lines changed: 140 additions & 33 deletions

File tree

architecture/overview.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@ src/httpware/
3232
│ ├── retry.py # Retry + AsyncRetry
3333
│ ├── timeout.py # AsyncTimeout
3434
│ ├── circuit_breaker.py # CircuitBreaker + AsyncCircuitBreaker
35-
│ └── _backoff.py # full-jitter helper (shared)
35+
│ ├── _backoff.py # full-jitter helper (shared)
36+
│ └── _event_loop_guard.py # single-event-loop guard (shared: AsyncBulkhead + AsyncCircuitBreaker)
3637
├── decoders/ # shared (ResponseDecoder + adapters)
3738
└── _internal/
3839
├── body_cap.py # max_response_body_bytes: validate, read-capped (sync+async)
Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
---
2+
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.
3+
---
4+
5+
# Change: Share the single-event-loop guard between AsyncBulkhead and AsyncCircuitBreaker
6+
7+
**Lane:** lightweight — 4 files (above the usual ≤2 guard; one is the
8+
`architecture/overview.md` module-layout promotion this change requires, per
9+
the `2026-06-13.04`/`2026-07-13.07` precedent of the file-count guard proxying
10+
*code* risk, not file count). No public-API change, no test changes (existing
11+
suite is the parity net).
12+
13+
Spec: [`planning/audits/2026-06-14-deep-audit.md`](../../audits/2026-06-14-deep-audit.md)
14+
("Sync/async duplication, no divergence yet" — `_check_loop` in bulkhead vs
15+
circuit_breaker), the last item from that list not yet addressed by this
16+
session's refactor series (`2026-07-13.01``.06`).
17+
18+
## Goal
19+
20+
`AsyncBulkhead._check_loop` (`bulkhead.py:103-121`) and
21+
`AsyncCircuitBreaker._check_loop` (`circuit_breaker.py:340-354`) are
22+
byte-identical double-checked-locking logic — same structure, differing only
23+
in which cross-loop-message constant they format. Unlike the other items in
24+
this refactor series, this pair lives in two different classes across two
25+
different files, not a sync/async split within one file. No behavior change.
26+
27+
## Approach
28+
29+
New module `src/httpware/middleware/resilience/_event_loop_guard.py`
30+
(sibling of the existing `_backoff.py` shared helper) with one function:
31+
32+
```python
33+
def check_event_loop(
34+
get_loop: Callable[[], asyncio.AbstractEventLoop | None],
35+
set_loop: Callable[[asyncio.AbstractEventLoop], None],
36+
loop_lock: threading.Lock,
37+
message_template: str,
38+
) -> None: ...
39+
```
40+
41+
`get_loop`/`set_loop` are closures over the caller's `self._loop`, not a
42+
passed-in `self` — passing `self` and reaching into `instance._loop` from a
43+
free function would trip ruff's `SLF001` (private-member access from outside
44+
the class), which is exactly why `decoders/_caching.py`'s `_get_or_build`
45+
precedent takes the mutable container by value instead of the owning object.
46+
The inner check re-invokes `get_loop()` rather than reusing the outer
47+
snapshot, preserving the original's double-checked-locking correctness: a
48+
thread that loses the race to acquire `loop_lock` must still observe
49+
whichever loop the winner just bound, not its own stale pre-lock read
50+
(load-bearing, per the 2026-06-14 deep audit's own note that this is a
51+
correctness mechanism, not a decorative optimization).
52+
53+
Each `_check_loop` method becomes a 5-line call:
54+
55+
```python
56+
def _check_loop(self) -> None:
57+
check_event_loop(
58+
lambda: self._loop,
59+
lambda loop: setattr(self, "_loop", loop),
60+
self._loop_lock,
61+
_ASYNCBULKHEAD_CROSS_LOOP_MSG, # or _CROSS_LOOP_MSG in circuit_breaker.py
62+
)
63+
```
64+
65+
## Files
66+
67+
- `src/httpware/middleware/resilience/_event_loop_guard.py` — new, the shared
68+
`check_event_loop` function.
69+
- `src/httpware/middleware/resilience/bulkhead.py``AsyncBulkhead._check_loop`
70+
now delegates.
71+
- `src/httpware/middleware/resilience/circuit_breaker.py`
72+
`AsyncCircuitBreaker._check_loop` now delegates.
73+
- `architecture/overview.md` — module-layout tree gets the new file, matching
74+
`_backoff.py`'s existing entry.
75+
- No test file changes — `tests/test_bulkhead.py::test_cross_loop_acquire_raises_runtimeerror`
76+
and `tests/test_circuit_breaker.py::test_cross_loop_use_raises_runtimeerror`
77+
(plus the same-loop and construct-outside-loop tests) already exercise both
78+
extraction targets and stay green unmodified.
79+
80+
## Verification
81+
82+
- [x] `just test` — full suite green, 100% coverage maintained, same test
83+
count (pure refactor, no new/removed tests).
84+
- [x] `just lint-ci` — clean.
85+
- [x] Grep both files to confirm `_check_loop`'s body no longer contains
86+
`asyncio.get_running_loop()` or the inline double-checked-locking block
87+
— only the 5-line delegating call remains.
Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
"""Single-event-loop guard for async resilience middleware (private, shared)."""
2+
3+
import asyncio
4+
import threading
5+
from collections.abc import Callable
6+
7+
8+
def check_event_loop(
9+
get_loop: Callable[[], asyncio.AbstractEventLoop | None],
10+
set_loop: Callable[[asyncio.AbstractEventLoop], None],
11+
loop_lock: threading.Lock,
12+
message_template: str,
13+
) -> None:
14+
"""Bind the caller to the first event loop that calls it.
15+
16+
Raises RuntimeError on a later call from a different loop. `get_loop`/`set_loop`
17+
read and write the caller's cached-loop attribute. The inner check re-reads via
18+
`get_loop()` rather than reusing the outer snapshot, so a thread that loses the
19+
race to acquire `loop_lock` still sees whichever loop the winner just bound
20+
(double-checked locking) — the outer unlocked read handles the common
21+
already-bound case without lock overhead.
22+
"""
23+
current = asyncio.get_running_loop()
24+
cached = get_loop()
25+
if cached is current:
26+
return
27+
if cached is not None:
28+
raise RuntimeError(message_template.format(first=cached, current=current))
29+
with loop_lock:
30+
cached = get_loop()
31+
if cached is None:
32+
set_loop(current)
33+
# pragma below: inner double-check-with-lock race arm; only reachable when
34+
# two threads simultaneously pass the outer check, which single-threaded
35+
# tests can't trigger.
36+
elif cached is not current: # pragma: no cover
37+
raise RuntimeError(message_template.format(first=cached, current=current))

src/httpware/middleware/resilience/bulkhead.py

Lines changed: 7 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
from httpware._internal.observability import _emit_event
2828
from httpware.errors import BulkheadFullError
2929
from httpware.middleware import AsyncNext, Next
30+
from httpware.middleware.resilience._event_loop_guard import check_event_loop
3031

3132

3233
_MAX_CONCURRENT_INVALID = "max_concurrent must be >= 1"
@@ -101,24 +102,12 @@ def __init__(
101102
self._loop_lock = threading.Lock()
102103

103104
def _check_loop(self) -> None:
104-
current = asyncio.get_running_loop()
105-
cached = self._loop
106-
if cached is current:
107-
return
108-
if cached is not None:
109-
raise RuntimeError(
110-
_ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=cached, current=current),
111-
)
112-
with self._loop_lock:
113-
if self._loop is None:
114-
self._loop = current
115-
# pragma below: inner double-check-with-lock race arm; only
116-
# reachable when two threads simultaneously pass the outer
117-
# cached-loop check, which single-threaded tests can't trigger.
118-
elif self._loop is not current: # pragma: no cover
119-
raise RuntimeError(
120-
_ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=self._loop, current=current),
121-
)
105+
check_event_loop(
106+
lambda: self._loop,
107+
lambda loop: setattr(self, "_loop", loop),
108+
self._loop_lock,
109+
_ASYNCBULKHEAD_CROSS_LOOP_MSG,
110+
)
122111

123112
async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response: # noqa: A002
124113
"""Acquire a slot (bounded by acquire_timeout), invoke next, release."""

src/httpware/middleware/resilience/circuit_breaker.py

Lines changed: 7 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
from httpware._internal.observability import _emit_event
4747
from httpware.errors import CircuitOpenError, NetworkError, StatusError, TimeoutError # noqa: A004
4848
from httpware.middleware import AsyncNext, Next
49+
from httpware.middleware.resilience._event_loop_guard import check_event_loop
4950

5051

5152
_FAILURE_THRESHOLD_INVALID = "failure_threshold must be >= 1"
@@ -338,20 +339,12 @@ def __init__( # noqa: PLR0913 — breaker has many orthogonal knobs; a dataclas
338339
self._loop_lock = threading.Lock()
339340

340341
def _check_loop(self) -> None:
341-
current = asyncio.get_running_loop()
342-
cached = self._loop
343-
if cached is current:
344-
return
345-
if cached is not None:
346-
raise RuntimeError(_CROSS_LOOP_MSG.format(first=cached, current=current))
347-
with self._loop_lock:
348-
if self._loop is None:
349-
self._loop = current
350-
# pragma below: inner double-check-with-lock race arm; only reachable when
351-
# two threads simultaneously pass the outer check, which single-threaded
352-
# tests can't trigger.
353-
elif self._loop is not current: # pragma: no cover
354-
raise RuntimeError(_CROSS_LOOP_MSG.format(first=self._loop, current=current))
342+
check_event_loop(
343+
lambda: self._loop,
344+
lambda loop: setattr(self, "_loop", loop),
345+
self._loop_lock,
346+
_CROSS_LOOP_MSG,
347+
)
355348

356349
@property
357350
def state(self) -> CircuitState:

0 commit comments

Comments
 (0)