refactor(resilience): share single-event-loop guard between bulkhead/circuit_breaker (#106)

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.

Author: lesnik512

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)

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.

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))

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."""

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: