Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion architecture/overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
87 changes: 87 additions & 0 deletions planning/changes/2026-07-13.10-event-loop-guard-shared.md
Original file line number Diff line number Diff line change
@@ -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.
37 changes: 37 additions & 0 deletions src/httpware/middleware/resilience/_event_loop_guard.py
Original file line number Diff line number Diff line change
@@ -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))
25 changes: 7 additions & 18 deletions src/httpware/middleware/resilience/bulkhead.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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."""
Expand Down
21 changes: 7 additions & 14 deletions src/httpware/middleware/resilience/circuit_breaker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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:
Expand Down