diff --git a/planning/changes/2026-07-13.04-bulkhead-shared-validation-rejection.md b/planning/changes/2026-07-13.04-bulkhead-shared-validation-rejection.md new file mode 100644 index 0000000..6a4093b --- /dev/null +++ b/planning/changes/2026-07-13.04-bulkhead-shared-validation-rejection.md @@ -0,0 +1,166 @@ +--- +summary: Share Bulkhead/AsyncBulkhead's hand-duplicated constructor validation and rejection-event-plus-BulkheadFullError construction via two module-level functions. +--- + +# Design: Share Bulkhead's validation and rejection logic + +## Summary + +`AsyncBulkhead` and `Bulkhead` (`src/httpware/middleware/resilience/bulkhead.py`, +185 lines) hand-duplicate two blocks: constructor validation +(`max_concurrent`/`acquire_timeout` range checks) and the on-timeout +rejection path (emit a `bulkhead.rejected` event, construct +`BulkheadFullError`). Only the semaphore type and acquire-timeout +mechanics genuinely differ. This change extracts both blocks into two +module-level functions in the same file — `_validate_bulkhead_config` and +`_emit_bulkhead_rejected` — closing a deferral this repo's own planning +history named for itself. No behavior change. + +## Motivation + +- `planning/changes/2026-06-23.01-retry-policy-extraction.md`'s + Out-of-scope section states: "Not extending the same treatment to + Bulkhead in this change." That change gave `circuit_breaker.py` and + `retry.py` shared decision objects (`_CircuitBreakerState`, + `_RetryPolicy`); `bulkhead.py` never got the equivalent treatment. +- **Verified duplication:** `AsyncBulkhead.__init__` (`:62-76`) and + `Bulkhead.__init__` (`:143-155`) run the identical two `if`/`raise + ValueError` checks. `AsyncBulkhead.__call__`'s `except TimeoutError` + branch (`:107-123`) and `Bulkhead.__call__`'s `if not acquired:` branch + (`:163-179`) build the identical `_emit_event(...)` call (same event + name, level, message, attributes shape) and the identical + `BulkheadFullError(max_concurrent=..., acquire_timeout=...)` + construction. +- **Deletion test:** delete the two shared functions and both blocks + reappear verbatim at all four call sites (2 `__init__`s, 2 `__call__`s) + — real, load-bearing logic, not a pass-through. + +## Non-goals + +- No behavior change. Validation error messages, the emitted event's + name/level/message/attributes, and `BulkheadFullError`'s fields stay + byte-identical. +- Not a stateful class (unlike `_CircuitBreakerState`/`_RetryPolicy`). + Both extracted operations are one-shot/per-call pure functions with no + config or evolving state reused across calls — a class would be a + namespace with no benefit, the same reasoning already applied to + `client.py`'s request-assembly extraction earlier this session. +- Not moving the functions to `_internal/`. `bulkhead.py`'s logic is used + only by its own two classes — same situation as `_CircuitBreakerState` + in `circuit_breaker.py` and `_RetryPolicy` in `retry.py`, both of which + stayed local to their file. +- Not touching `AsyncBulkhead._check_loop`, the semaphore acquire + mechanics, or anything else in either `__call__`/`__init__` beyond the + two blocks named above. + +## Design + +### 1. `_validate_bulkhead_config` + +```python +def _validate_bulkhead_config(*, max_concurrent: int, acquire_timeout: float | None) -> None: + if max_concurrent < 1: + raise ValueError(_MAX_CONCURRENT_INVALID) + if acquire_timeout is not None and acquire_timeout < 0: + raise ValueError(_ACQUIRE_TIMEOUT_INVALID) +``` + +Replaces the two `if`/`raise` lines in both `__init__`s with: + +```python +_validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout) +``` + +Uses `-> None` (not `typing.NoReturn`), matching the existing house idiom +for validation helpers that raise conditionally +(`_validate_max_response_body_bytes`, `_validate_httpx2_client_conflict`) +— no precedent for `NoReturn` exists in this codebase, and this function +doesn't always raise (only on the invalid branches), so `None` is also the +technically correct annotation here regardless of style. + +### 2. `_emit_bulkhead_rejected` — builds and returns, does not raise + +```python +def _emit_bulkhead_rejected( + request: httpx2.Request, + *, + max_concurrent: int, + acquire_timeout: float | None, +) -> BulkheadFullError: + _emit_event( + _LOGGER, + "bulkhead.rejected", + level=logging.WARNING, + message="bulkhead rejected request — acquire_timeout exceeded", + attributes={ + "max_concurrent": max_concurrent, + "acquire_timeout": acquire_timeout, + "method": request.method, + "url": str(request.url), + }, + ) + return BulkheadFullError(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout) +``` + +**Deliberately returns the exception rather than raising it.** The two +call sites need different exception-chaining behavior: `AsyncBulkhead`'s +rejection fires inside `except TimeoutError as exc:` and today does +`raise BulkheadFullError(...) from exc` (explicit `__cause__`); +`Bulkhead`'s rejection fires inside `if not acquired:` — no active +exception, bare `raise BulkheadFullError(...)`. If the function raised +internally, the async call site's explicit `from exc` would be lost, +silently downgrading to implicit `__context__` chaining (a different +traceback presentation — "the above exception was the direct cause of" +vs "during handling of the above exception"). Returning the constructed +exception lets each call site keep its own `raise` statement exactly as +today: + +```python +# AsyncBulkhead.__call__, inside except TimeoutError as exc: +raise _emit_bulkhead_rejected( + request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout, +) from exc + +# Bulkhead.__call__, inside if not acquired: +raise _emit_bulkhead_rejected( + request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout, +) +``` + +### 3. Placement + +Both functions go in `bulkhead.py`, module level, alongside the existing +module constants (`_MAX_CONCURRENT_INVALID`, `_ACQUIRE_TIMEOUT_INVALID`, +`_LOGGER`) — before `class AsyncBulkhead:`. + +## Testing + +- **Parity net:** existing `tests/test_bulkhead.py` and + `tests/test_bulkhead_sync.py` stay green with the assertions they + already have (message content, `BulkheadFullError` field values, + `bulkhead.rejected` event via `caplog`) — byte-identical behavior is the + bar. +- **New chaining tests** (the one subtlety this refactor is careful to + preserve, and no existing test asserts it): in `test_bulkhead.py`, add + an assertion that the raised `BulkheadFullError`'s `__cause__` is the + `TimeoutError` that triggered it; in `test_bulkhead_sync.py`, add an + assertion that `__cause__` is `None`. Mirrors how `DecodeError.__cause__` + is explicitly tested in `test_errors.py`. +- **No new direct-unit-test file** for the two shared functions — the + existing construction-error tests already exercise + `_validate_bulkhead_config` fully (both branches, both classes), and + the existing `BulkheadFullError`-field/event tests plus the new + chaining tests already exercise `_emit_bulkhead_rejected` fully. +- `just lint && just test` both clean; 100% coverage maintained. + +## Risk + +- **Chaining regression** (the main risk this design addresses directly): + *Mitigation:* returning rather than raising keeps each `raise` statement + exactly where it is today; the new chaining tests make a future + regression (e.g. someone "simplifying" the sync call site to also use + `from exc` where no exception exists, or vice versa) fail loudly. +- **Behavioral drift in the shared validation/event logic** (unlikely × + low): *Mitigation:* extract verbatim under the existing green suites, + which assert exact message text and field values; do not edit those + assertions in this change (only add the two new chaining assertions). diff --git a/src/httpware/middleware/resilience/bulkhead.py b/src/httpware/middleware/resilience/bulkhead.py index 7d41db4..927a3db 100644 --- a/src/httpware/middleware/resilience/bulkhead.py +++ b/src/httpware/middleware/resilience/bulkhead.py @@ -40,6 +40,34 @@ _LOGGER = logging.getLogger("httpware.bulkhead") +def _validate_bulkhead_config(*, max_concurrent: int, acquire_timeout: float | None) -> None: + if max_concurrent < 1: + raise ValueError(_MAX_CONCURRENT_INVALID) + if acquire_timeout is not None and acquire_timeout < 0: + raise ValueError(_ACQUIRE_TIMEOUT_INVALID) + + +def _emit_bulkhead_rejected( + request: httpx2.Request, + *, + max_concurrent: int, + acquire_timeout: float | None, +) -> BulkheadFullError: + _emit_event( + _LOGGER, + "bulkhead.rejected", + level=logging.WARNING, + message="bulkhead rejected request — acquire_timeout exceeded", + attributes={ + "max_concurrent": max_concurrent, + "acquire_timeout": acquire_timeout, + "method": request.method, + "url": str(request.url), + }, + ) + return BulkheadFullError(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout) + + class AsyncBulkhead: """Async concurrency limiter middleware backed by ``asyncio.Semaphore``. @@ -65,10 +93,7 @@ def __init__( max_concurrent: int, acquire_timeout: float | None = 1.0, ) -> None: - if max_concurrent < 1: - raise ValueError(_MAX_CONCURRENT_INVALID) - if acquire_timeout is not None and acquire_timeout < 0: - raise ValueError(_ACQUIRE_TIMEOUT_INVALID) + _validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout) self._max_concurrent = max_concurrent self._acquire_timeout = acquire_timeout self._sem = asyncio.Semaphore(max_concurrent) @@ -105,19 +130,8 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res async with asyncio.timeout(self._acquire_timeout): await self._sem.acquire() except TimeoutError as exc: - _emit_event( - _LOGGER, - "bulkhead.rejected", - level=logging.WARNING, - message="bulkhead rejected request — acquire_timeout exceeded", - attributes={ - "max_concurrent": self._max_concurrent, - "acquire_timeout": self._acquire_timeout, - "method": request.method, - "url": str(request.url), - }, - ) - raise BulkheadFullError( + raise _emit_bulkhead_rejected( + request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout, ) from exc @@ -146,10 +160,7 @@ def __init__( max_concurrent: int, acquire_timeout: float | None = 1.0, ) -> None: - if max_concurrent < 1: - raise ValueError(_MAX_CONCURRENT_INVALID) - if acquire_timeout is not None and acquire_timeout < 0: - raise ValueError(_ACQUIRE_TIMEOUT_INVALID) + _validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout) self._max_concurrent = max_concurrent self._acquire_timeout = acquire_timeout self._sem = threading.Semaphore(max_concurrent) @@ -161,19 +172,8 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response: # n # False otherwise). Both match AsyncBulkhead's contract. acquired = self._sem.acquire(timeout=self._acquire_timeout) if not acquired: - _emit_event( - _LOGGER, - "bulkhead.rejected", - level=logging.WARNING, - message="bulkhead rejected request — acquire_timeout exceeded", - attributes={ - "max_concurrent": self._max_concurrent, - "acquire_timeout": self._acquire_timeout, - "method": request.method, - "url": str(request.url), - }, - ) - raise BulkheadFullError( + raise _emit_bulkhead_rejected( + request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout, ) diff --git a/tests/test_bulkhead.py b/tests/test_bulkhead.py index b5863c5..383cda6 100644 --- a/tests/test_bulkhead.py +++ b/tests/test_bulkhead.py @@ -148,6 +148,30 @@ async def _hold_slot() -> None: await task +async def test_bulkhead_full_error_chains_from_timeout() -> None: + """BulkheadFullError raised on the async timeout path chains from the TimeoutError.""" + handler = _SlowHandler(delay=1.0) + bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST) + client = _client(handler, bulkhead=bulkhead) + + async def _hold_slot() -> None: + await client.get("https://example.test/slow") + + task = asyncio.create_task(_hold_slot()) + # Yield to let the slow request acquire the semaphore. + await asyncio.sleep(0) + + with pytest.raises(BulkheadFullError) as exc_info: + await client.get("https://example.test/fast") + + assert isinstance(exc_info.value.__cause__, TimeoutError) + + # Cancel the lingering slow task to avoid polluting the event loop. + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def test_bounded_wait_raises_bulkhead_full_error() -> None: """With max_concurrent=1 and acquire_timeout=0.02, the second call raises after ~20ms. diff --git a/tests/test_bulkhead_sync.py b/tests/test_bulkhead_sync.py index c427e67..213c7a6 100644 --- a/tests/test_bulkhead_sync.py +++ b/tests/test_bulkhead_sync.py @@ -129,6 +129,26 @@ def test_acquire_timeout_rejects_when_no_slot_available() -> None: holder.join() +def test_bulkhead_full_error_no_chaining() -> None: + """BulkheadFullError raised on the sync timeout path has no __cause__ (no active exception).""" + handler = _SlowHandler(delay=0.1) + client = _client( + handler, + bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST), + ) + + holder = threading.Thread(target=client.get, args=("https://example.test/hold",)) + holder.start() + # Give the holder time to acquire the only slot + time.sleep(0.01) + try: + with pytest.raises(BulkheadFullError) as exc_info: + client.get("https://example.test/blocked") + assert exc_info.value.__cause__ is None + finally: + holder.join() + + def test_releases_slot_on_exception() -> None: """A handler that raises must still cause the slot to be released.""" calls = []