perf: cheaper hot paths in the circuit breakers, AsyncBulkhead and the event-loop guard

docs/resilience.md

@@ -287,15 +287,15 @@ Every `AsyncCircuitBreaker` parameter, with the same defaults, plus:
 
 | Parameter | Default | Effect |
 |---|---|---|
-| `key` | the request's origin | Maps a request to its circuit key. Any hashable value works. The default is the origin as a string built from scheme, host and port, such as `https://a.example` or `http://a.example:8080`. Hosts are lowercased and default ports dropped, so `https://A.example/x` and `https://a.example:443/y` share a circuit, while `http://a.example` and `https://a.example:8443` each get their own. Userinfo is never part of it. |
+| `key` | the request's origin | Maps a request to its circuit key. Any hashable value works. The default is the origin, built from scheme, host and port, which events show as `https://a.example` or `http://a.example:8080`. Hosts are lowercased and default ports dropped, so `https://A.example/x` and `https://a.example:443/y` share a circuit, while `http://a.example` and `https://a.example:8443` each get their own. Userinfo is never part of it. |
 
 ### Circuit lifetime
 
 A circuit is created on the first request for its key and kept for as long as the breaker lives. Nothing is evicted or pruned, so memory grows with the number of distinct keys. `key` must therefore map requests to a small, bounded set, such as your configured upstreams. Never derive it from user input. A key function that returns something unbounded, such as the full URL, grows the map forever.
 
 ### Observability
 
-The keyed breakers emit the same events as `AsyncCircuitBreaker` on the same `httpware.circuit_breaker` logger. Each event has one extra attribute, `circuit_key`, which is the `str()` of the request's circuit key. httpware redacts `url` but not `circuit_key`, which reaches log records and span events exactly as the key function returned it. A custom `key` must not return credentials, tokens or other sensitive values. The default origin contains none.
+The keyed breakers emit the same events as `AsyncCircuitBreaker` on the same `httpware.circuit_breaker` logger. Each event has one extra attribute, `circuit_key`: the origin string for the default key, otherwise the `str()` of the request's circuit key. httpware redacts `url` but not `circuit_key`, which reaches log records and span events exactly as the key function returned it. A custom `key` must not return credentials, tokens or other sensitive values. The default origin contains none.
 
 ### Example
 

src/httpware/middleware/resilience/_event_loop_guard.py

@@ -11,24 +11,17 @@ def check_event_loop(
     loop_lock: threading.Lock,
     message_template: str,
 ) -> None:
-    """Bind the caller to the first event loop that calls it.
+    """Bind the caller to the running loop if it has none yet, else raise RuntimeError.
 
-    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.
+    The slow path of the caller's unlocked `self._loop is asyncio.get_running_loop()` check.
+    `get_loop`/`set_loop` read and write the caller's cached-loop attribute; the read under
+    `loop_lock` sees a loop that another thread bound after the caller's unlocked read.
     """
     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)
-        elif cached is not current:
-            raise RuntimeError(message_template.format(first=cached, current=current))
+            return
+    if cached is not current:
+        raise RuntimeError(message_template.format(first=cached, current=current))

src/httpware/middleware/resilience/bulkhead.py

@@ -1,9 +1,9 @@
 """AsyncBulkhead middleware — concurrency limiter via asyncio.Semaphore.
 
 The middleware owns an asyncio.Semaphore(max_concurrent). On each request,
-it acquires a slot (bounded by acquire_timeout via asyncio.timeout) and
-releases the slot in a try/finally so success, exceptions, and cancellation
-all release deterministically.
+it acquires a slot and releases it in a try/finally so success, exceptions,
+and cancellation all release deterministically. When every slot is taken,
+asyncio.timeout bounds the wait by acquire_timeout.
 
 AsyncBulkhead is the sharable unit — pass the same instance to multiple
 AsyncClient(middleware=[shared]) calls to enforce a joint cap across clients.
@@ -110,9 +110,10 @@ def _check_loop(self) -> None:
 
     async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
         """Acquire a slot (bounded by acquire_timeout), invoke next, release."""
-        self._check_loop()
+        if self._loop is not asyncio.get_running_loop():
+            self._check_loop()
         try:
-            if self._acquire_timeout is None:
+            if self._acquire_timeout is None or not self._sem.locked():
                 await self._sem.acquire()
             else:
                 async with asyncio.timeout(self._acquire_timeout):

src/httpware/middleware/resilience/circuit_breaker.py

@@ -40,7 +40,8 @@
 AsyncCircuitBreaker relies on asyncio atomicity (no await inside a transition) plus a
 single-event-loop guard; CircuitBreaker (sync) serializes transitions with a
 threading.Lock and emits events only after releasing it, so a logging handler may send
-requests through the same breaker. Both are sharable across clients (one shared circuit);
+requests through the same breaker. A request to a closed circuit is admitted without the
+lock; its outcome is still recorded under it. Both are sharable across clients (one shared circuit);
 a sync instance cannot be shared with an async one.
 """
 
@@ -196,6 +197,7 @@ class _CircuitBreakerState:
     the caller takes them with the outcome and emits them outside any lock.
     `admit` hands out a ticket: the epoch, which every transition advances by two, plus one
     for the probe. `record` ignores a ticket from an earlier epoch.
+    `closed_ticket` is the epoch while CLOSED and None otherwise, so one read admits without the lock.
     """
 
     def __init__(  # noqa: PLR0913 — breaker state has many orthogonal knobs; a dataclass would be worse
@@ -244,6 +246,7 @@ def __init__(  # noqa: PLR0913 — breaker state has many orthogonal knobs; a da
         self._opened_at = 0.0
         self._probe_in_flight = False
         self._epoch = 0
+        self.closed_ticket: int | None = 0
         self._rejected_emitted_at = -math.inf
         self._rejections_suppressed = 0
         self._episode_rejections = 0
@@ -256,8 +259,9 @@ def state(self) -> CircuitState:
 
     def admit(self, request: httpx2.Request) -> int:
         """Return the request's ticket, or raise _RejectionError or _HalfOpened. No await inside."""
-        if self._state is CircuitState.CLOSED:
-            return self._epoch
+        ticket = self.closed_ticket
+        if ticket is not None:
+            return ticket
         now = self._now()
         if self._state is CircuitState.OPEN:
             elapsed = now - self._opened_at
@@ -341,6 +345,7 @@ def _record_probe_success(self, request: httpx2.Request) -> None:
     def _move_to(self, state: CircuitState) -> None:
         self._state = state
         self._epoch += _EPOCH_STEP
+        self.closed_ticket = self._epoch if state is CircuitState.CLOSED else None
 
     def _enter_open(self, request: httpx2.Request, message: str, attributes: dict[str, typing.Any]) -> None:
         if self._state is CircuitState.CLOSED:
@@ -414,13 +419,29 @@ def _rejected(rejection: _RejectionError) -> CircuitOpenError:
     return CircuitOpenError(retry_after=rejection.retry_after)
 
 
-async def _call_async(state: _CircuitBreakerState, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
+def _admit(state: _CircuitBreakerState, request: httpx2.Request) -> int:
+    try:
+        return state.admit(request)
+    except _RejectionError as rejection:
+        raise _rejected(rejection) from None
+    except _HalfOpened as half_opened:
+        return _half_opened(half_opened)
+
+
+def _admit_locked(state: _CircuitBreakerState, lock: threading.Lock, request: httpx2.Request) -> int:
     try:
-        ticket = state.admit(request)
+        with lock:
+            return state.admit(request)
     except _RejectionError as rejection:
         raise _rejected(rejection) from None
     except _HalfOpened as half_opened:
-        ticket = _half_opened(half_opened)
+        return _half_opened(half_opened)
+
+
+async def _call_async(state: _CircuitBreakerState, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
+    ticket = state.closed_ticket
+    if ticket is None:
+        ticket = _admit(state, request)
     try:
         response = await next(request)
     except BaseException as exc:
@@ -440,13 +461,9 @@ def _call_sync(
     request: httpx2.Request,
     next: Next,  # noqa: A002
 ) -> httpx2.Response:
-    try:
-        with lock:
-            ticket = state.admit(request)
-    except _RejectionError as rejection:
-        raise _rejected(rejection) from None
-    except _HalfOpened as half_opened:
-        ticket = _half_opened(half_opened)
+    ticket = state.closed_ticket
+    if ticket is None:
+        ticket = _admit_locked(state, lock, request)
     try:
         response = next(request)
     except BaseException as exc:
@@ -463,30 +480,43 @@ def _call_sync(
     return response
 
 
-def _origin(request: httpx2.Request) -> str:
+def _origin(request: httpx2.Request) -> tuple[str, bytes, int | None]:
+    url = request.url
+    return url.scheme, url.raw_host, url.port
+
+
+def _origin_label(request: httpx2.Request) -> str:
     return f"{request.url.scheme}://{request.url.netloc.decode('ascii')}"
 
 
 class _KeyedStates:
     """One lazily created _CircuitBreakerState per circuit key; the map is never pruned.
 
-    Not synchronized: the async wrapper relies on its single event loop, the sync wrapper
-    calls `get` under its lock.
+    Lookups read the map without a lock; creating a circuit takes `_create_lock`.
     """
 
     def __init__(self, key: Callable[[httpx2.Request], Hashable], **config: typing.Any) -> None:  # noqa: ANN401
         self._key = key
         self._new_state = functools.partial(_CircuitBreakerState, **config)
         self._new_state()  # validates the config at construction instead of on the first request
         self._states: dict[Hashable, _CircuitBreakerState] = {}
+        self._create_lock = threading.Lock()
 
     def get(self, request: httpx2.Request) -> _CircuitBreakerState:
         circuit_key = self._key(request)
         state = self._states.get(circuit_key)
         if state is None:
-            state = self._states[circuit_key] = self._new_state(key=str(circuit_key))
+            state = self._create(circuit_key, request)
         return state
 
+    def _create(self, circuit_key: Hashable, request: httpx2.Request) -> _CircuitBreakerState:
+        label = _origin_label(request) if self._key is _origin else str(circuit_key)
+        with self._create_lock:
+            state = self._states.get(circuit_key)
+            if state is None:
+                state = self._states[circuit_key] = self._new_state(key=label)
+            return state
+
 
 class AsyncCircuitBreaker:
     """Async classic circuit breaker middleware. See the module docstring for the contract."""
@@ -534,7 +564,8 @@ def state(self) -> CircuitState:
 
     async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
         """Admit, forward, then record the outcome. Fast-fail when the circuit is not closed."""
-        self._check_loop()
+        if self._loop is not asyncio.get_running_loop():
+            self._check_loop()
         return await _call_async(self._state, request, next)
 
 
@@ -585,8 +616,8 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
 class AsyncKeyedCircuitBreaker:
     """Async circuit breaker with one independent circuit per circuit key.
 
-    `key` maps a request to its circuit key; the default is the request URL's origin as a string
-    such as "https://api.example" (scheme, host and non-default port; never userinfo). Every
+    `key` maps a request to its circuit key; the default is the request URL's origin (scheme, host
+    and non-default port; never userinfo), shown in events as "https://api.example". Every
     circuit behaves exactly like an AsyncCircuitBreaker built with the same arguments. Circuits
     are created on first use and kept for the breaker's lifetime, so the key must take a small,
     bounded set of values.
@@ -629,14 +660,16 @@ def _check_loop(self) -> None:
 
     async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
         """Pick the request's circuit, then admit, forward and record the outcome on it."""
-        self._check_loop()
+        if self._loop is not asyncio.get_running_loop():
+            self._check_loop()
         return await _call_async(self._states.get(request), request, next)
 
 
 class KeyedCircuitBreaker:
     """Sync circuit breaker with one independent circuit per circuit key. Mirror of AsyncKeyedCircuitBreaker.
 
-    One threading.Lock serializes circuit lookup and every transition across all keys.
+    One threading.Lock serializes every transition across all keys. Looking up an existing
+    circuit and admitting into a closed one take no lock.
     """
 
     def __init__(  # noqa: PLR0913 — breaker has many orthogonal knobs; a dataclass would be worse
@@ -667,6 +700,4 @@ def __init__(  # noqa: PLR0913 — breaker has many orthogonal knobs; a dataclas
 
     def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # noqa: A002
         """Pick the request's circuit, then admit, forward and record the outcome on it."""
-        with self._lock:
-            state = self._states.get(request)
-        return _call_sync(state, self._lock, request, next)
+        return _call_sync(self._states.get(request), self._lock, request, next)

tests/test_bulkhead.py

@@ -110,6 +110,31 @@ async def test_serializes_at_capacity() -> None:
     assert handler.max_in_flight == 1  # cap honored
 
 
+async def test_a_freed_slot_goes_to_the_queued_waiter_before_a_new_call() -> None:
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_LONG)
+    request = httpx2.Request("GET", "https://example.test/x")
+    release = asyncio.Event()
+    order: list[str] = []
+
+    def _next(name: str) -> Callable[[httpx2.Request], Coroutine[Any, Any, httpx2.Response]]:
+        async def _send(request: httpx2.Request) -> httpx2.Response:
+            order.append(name)
+            if name == "holder":
+                await release.wait()
+            return httpx2.Response(HTTPStatus.OK, request=request)
+
+        return _send
+
+    holder = asyncio.create_task(bulkhead(request, _next("holder")))
+    await asyncio.sleep(0)
+    queued = asyncio.create_task(bulkhead(request, _next("queued")))
+    await asyncio.sleep(0)
+    release.set()
+    late = asyncio.create_task(bulkhead(request, _next("late")))
+    await asyncio.gather(holder, queued, late)
+    assert order == ["holder", "queued", "late"]
+
+
 async def test_max_concurrent_2_observes_at_most_2_in_flight() -> None:
     handler = _SlowHandler(delay=0.02)
     client = _client(handler, bulkhead=AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_2, acquire_timeout=None))