From 2a94ca6c777e7ca829c24d91a38e2e731ca5a2e9 Mon Sep 17 00:00:00 2001 From: bagowix Date: Sat, 8 Aug 2026 00:09:40 +0300 Subject: [PATCH 1/3] feat: make transport breaker keys configurable --- CHANGELOG.md | 9 ++ docs/integrations/aiohttp.md | 34 ++++++-- docs/integrations/httpx.md | 31 ++++++- docs/integrations/httpx2.md | 35 +++++++- docs/integrations/requests.md | 27 +++++- docs/llms-full.txt | 127 ++++++++++++++++++++++++++--- interlock/integrations/aiohttp.py | 55 +++++++++---- interlock/integrations/httpx.py | 67 +++++++++------ interlock/integrations/httpx2.py | 67 +++++++++------ interlock/integrations/requests.py | 55 +++++++++---- tests/test_aiohttp.py | 45 ++++++++++ tests/test_httpx.py | 110 +++++++++++++++++++++++++ tests/test_httpx2.py | 47 +++++++++++ tests/test_requests.py | 44 ++++++++++ 14 files changed, 653 insertions(+), 100 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 109379d..ff8a4af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,15 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Added +- **Transport integrations can key breakers by logical dependency instead of + raw host.** Service-discovery suffixes and shared gateway hosts previously + made the httpx2/httpx transports, aiohttp middleware, and requests adapter + merge unrelated dependencies or split one dependency across several + breakers. Their new `name_resolver=` callback receives the native request + and supplies the single name used by the registry, open-circuit errors, and + listener events. Host-based naming remains the default, while empty custom + names fail before any network I/O. + - **Listener contracts now match the events a component actually emits.** `CoreEventListener`, `StorageEventListener` and `PipelineEventListener` let a breaker-only, coordination-only or strategy-only sink pass strict type diff --git a/docs/integrations/aiohttp.md b/docs/integrations/aiohttp.md index 153ff64..f7788e6 100644 --- a/docs/integrations/aiohttp.md +++ b/docs/integrations/aiohttp.md @@ -51,10 +51,33 @@ The breaker observes the time to *response headers*; reading the body happens outside the guarded call — the same semantics as the [httpx2 transport](httpx2.md). +## Custom breaker keys + +Pass `name_resolver` when host-based isolation does not match the logical +dependencies. The callback receives the native `aiohttp.ClientRequest` and +returns the breaker name: + +```python +from interlock.integrations.aiohttp import CircuitBreakerMiddleware + +middleware = CircuitBreakerMiddleware( + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +A resolver can collapse several discovery hosts onto one breaker or derive a +name from the request path to separate upstreams behind a shared gateway. It +must return a non-empty, non-whitespace name; invalid names raise `ValueError` +with the request URL before the handler performs I/O. + +The resolved name is used by the registry, `CircuitOpenError`, and every +listener event. Resolve it in the middleware rather than rewriting listener +labels so observed names always match the breaker whose state they describe. + ## Share one registry across sessions Several middleware instances can share one caller-owned registry, so traffic -to the same host contributes to one breaker and one sliding window: +resolving to the same name contributes to one breaker and one sliding window: ```python from interlock import Config, Registry @@ -123,8 +146,7 @@ Any custom `FailureClassifier` works too — see The middleware accepts the same collaborators as `CircuitBreaker` — `config`, `clock`, `initial_state`, `classifier`, `listener`. One middleware instance -holds one registry -of per-host breakers; reuse the instance across sessions to share breaker -state, or create separate instances to isolate them. For application-level -retries combine with the [tenacity integration](tenacity.md) and read -[Retries and circuit breakers](../guides/retries.md) first. +holds one registry of resolved breakers; reuse the instance across sessions to +share breaker state, or create separate instances to isolate them. For +application-level retries combine with the [tenacity integration](tenacity.md) +and read [Retries and circuit breakers](../guides/retries.md) first. diff --git a/docs/integrations/httpx.md b/docs/integrations/httpx.md index 1e6eb5d..5ae8456 100644 --- a/docs/integrations/httpx.md +++ b/docs/integrations/httpx.md @@ -86,10 +86,37 @@ continues normally. An open breaker raises `CircuitOpenError` before the wrapped transport performs I/O. A request URL without a host raises `ValueError` for the same reason: there is no dependency identity to key on. +## Custom breaker keys + +Pass `name_resolver` when the request host is transport plumbing rather than +the logical dependency identity. The callback receives the native +`httpx.Request` and returns the breaker name: + +```python +import httpx + +from interlock.integrations.httpx import AsyncCircuitBreakerTransport + +transport = AsyncCircuitBreakerTransport( + httpx.AsyncHTTPTransport(), + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +The same callback can split one gateway host into independent breakers, for +example by returning a name derived from the first path segment. It must return +a non-empty, non-whitespace name; invalid names raise `ValueError` with the +request URL before the wrapped transport performs I/O. + +The resolved name is the registry key and the name carried by +`CircuitOpenError` and every listener event. Use the resolver, rather than +rewriting labels in a listener, so breaker state and observability labels stay +aligned. Both synchronous and asynchronous transports accept the option. + ## Share one registry across clients Inject one caller-owned `Registry` when several clients should observe the -same dependency health. The transports then resolve the same host to the same +same dependency health. The transports then resolve the same name to the same breaker and contribute to one sliding window: ```python @@ -148,7 +175,7 @@ and is not recorded by the breaker. ## Tuning `config`, `clock`, `initial_state`, `classifier`, and `listener` are shared by -every per-host breaker created by the transport: +every breaker created by the transport: ```python import httpx diff --git a/docs/integrations/httpx2.md b/docs/integrations/httpx2.md index e89d4c2..83fd70c 100644 --- a/docs/integrations/httpx2.md +++ b/docs/integrations/httpx2.md @@ -92,11 +92,40 @@ correct than global state — each host's health is observed independently. When a host's breaker is open, its requests raise `CircuitOpenError` before reaching the network. +## Custom breaker keys + +Pass `name_resolver` when the request host is not the logical dependency +identity. The callback receives the native `httpx2.Request` and returns the +breaker name: + +```python +import httpx2 + +from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport + +transport = AsyncCircuitBreakerTransport( + httpx2.AsyncHTTPTransport(), + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +Returning the same name for several discovery hosts gives them one breaker; +deriving a name from the path can split independent upstreams behind one +gateway host. The result must be non-empty and contain something other than +whitespace. Invalid names raise `ValueError` with the request URL before the +wrapped transport performs I/O. + +The resolved name is used consistently as the registry key, in +`CircuitOpenError`, and in every listener event. Resolve the name here instead +of rewriting listener labels so metrics always identify the breaker whose +state they report. Both synchronous and asynchronous transports accept the +option. + ## Share one registry across clients Pass a caller-owned `Registry` when several clients reach the same dependency. -Requests for the same host then use one breaker instance and one sliding -window, even when they travel through different transports: +Requests resolving to the same name then use one breaker instance and one +sliding window, even when they travel through different transports: ```python import httpx2 @@ -144,7 +173,7 @@ the breaker cannot fix a contract or protocol error. ## Tuning Pass any of `config`, `clock`, `classifier`, `listener` to the transport; they -flow to every per-host breaker: +flow to every breaker: ```python from interlock import Config, LoggingEventListener diff --git a/docs/integrations/requests.md b/docs/integrations/requests.md index f1d5f18..dbcb41a 100644 --- a/docs/integrations/requests.md +++ b/docs/integrations/requests.md @@ -48,10 +48,35 @@ created lazily and shared across requests. When a host's circuit is open the request raises [`CircuitOpenError`](../reference.md) *before* a connection is made. +## Custom breaker keys + +Pass `name_resolver` when the request host is not the logical dependency +identity. The callback receives the native `requests.PreparedRequest` and +returns the breaker name. For example, the first path segment can separate +independent upstreams behind one gateway host: + +```python +from interlock.integrations.requests import CircuitBreakerAdapter + +adapter = CircuitBreakerAdapter( + name_resolver=lambda request: request.path_url.split('/')[1], +) +``` + +Returning one name for several discovery hosts instead makes them share a +breaker. The result must be non-empty and contain something other than +whitespace; invalid names raise `ValueError` with the request URL before the +adapter performs I/O. + +The resolved name is used consistently as the registry key, in +`CircuitOpenError`, and in every listener event. Resolve the identity here +instead of rewriting listener labels so metrics remain aligned with breaker +state. + ## Share one registry across sessions Inject a caller-owned `Registry` when independent sessions should use one -breaker and one sliding window for the same host: +breaker and one sliding window for the same resolved name: ```python import requests diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 75d93a4..f4ddf24 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -2811,11 +2811,40 @@ correct than global state — each host's health is observed independently. When a host's breaker is open, its requests raise `CircuitOpenError` before reaching the network. +## Custom breaker keys + +Pass `name_resolver` when the request host is not the logical dependency +identity. The callback receives the native `httpx2.Request` and returns the +breaker name: + +```python +import httpx2 + +from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport + +transport = AsyncCircuitBreakerTransport( + httpx2.AsyncHTTPTransport(), + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +Returning the same name for several discovery hosts gives them one breaker; +deriving a name from the path can split independent upstreams behind one +gateway host. The result must be non-empty and contain something other than +whitespace. Invalid names raise `ValueError` with the request URL before the +wrapped transport performs I/O. + +The resolved name is used consistently as the registry key, in +`CircuitOpenError`, and in every listener event. Resolve the name here instead +of rewriting listener labels so metrics always identify the breaker whose +state they report. Both synchronous and asynchronous transports accept the +option. + ## Share one registry across clients Pass a caller-owned `Registry` when several clients reach the same dependency. -Requests for the same host then use one breaker instance and one sliding -window, even when they travel through different transports: +Requests resolving to the same name then use one breaker instance and one +sliding window, even when they travel through different transports: ```python import httpx2 @@ -2863,7 +2892,7 @@ the breaker cannot fix a contract or protocol error. ## Tuning Pass any of `config`, `clock`, `classifier`, `listener` to the transport; they -flow to every per-host breaker: +flow to every breaker: ```python from interlock import Config, LoggingEventListener @@ -2971,10 +3000,37 @@ continues normally. An open breaker raises `CircuitOpenError` before the wrapped transport performs I/O. A request URL without a host raises `ValueError` for the same reason: there is no dependency identity to key on. +## Custom breaker keys + +Pass `name_resolver` when the request host is transport plumbing rather than +the logical dependency identity. The callback receives the native +`httpx.Request` and returns the breaker name: + +```python +import httpx + +from interlock.integrations.httpx import AsyncCircuitBreakerTransport + +transport = AsyncCircuitBreakerTransport( + httpx.AsyncHTTPTransport(), + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +The same callback can split one gateway host into independent breakers, for +example by returning a name derived from the first path segment. It must return +a non-empty, non-whitespace name; invalid names raise `ValueError` with the +request URL before the wrapped transport performs I/O. + +The resolved name is the registry key and the name carried by +`CircuitOpenError` and every listener event. Use the resolver, rather than +rewriting labels in a listener, so breaker state and observability labels stay +aligned. Both synchronous and asynchronous transports accept the option. + ## Share one registry across clients Inject one caller-owned `Registry` when several clients should observe the -same dependency health. The transports then resolve the same host to the same +same dependency health. The transports then resolve the same name to the same breaker and contribute to one sliding window: ```python @@ -3033,7 +3089,7 @@ and is not recorded by the breaker. ## Tuning `config`, `clock`, `initial_state`, `classifier`, and `listener` are shared by -every per-host breaker created by the transport: +every breaker created by the transport: ```python import httpx @@ -3108,10 +3164,33 @@ The breaker observes the time to *response headers*; reading the body happens outside the guarded call — the same semantics as the [httpx2 transport](httpx2.md). +## Custom breaker keys + +Pass `name_resolver` when host-based isolation does not match the logical +dependencies. The callback receives the native `aiohttp.ClientRequest` and +returns the breaker name: + +```python +from interlock.integrations.aiohttp import CircuitBreakerMiddleware + +middleware = CircuitBreakerMiddleware( + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), +) +``` + +A resolver can collapse several discovery hosts onto one breaker or derive a +name from the request path to separate upstreams behind a shared gateway. It +must return a non-empty, non-whitespace name; invalid names raise `ValueError` +with the request URL before the handler performs I/O. + +The resolved name is used by the registry, `CircuitOpenError`, and every +listener event. Resolve it in the middleware rather than rewriting listener +labels so observed names always match the breaker whose state they describe. + ## Share one registry across sessions Several middleware instances can share one caller-owned registry, so traffic -to the same host contributes to one breaker and one sliding window: +resolving to the same name contributes to one breaker and one sliding window: ```python from interlock import Config, Registry @@ -3180,11 +3259,10 @@ Any custom `FailureClassifier` works too — see The middleware accepts the same collaborators as `CircuitBreaker` — `config`, `clock`, `initial_state`, `classifier`, `listener`. One middleware instance -holds one registry -of per-host breakers; reuse the instance across sessions to share breaker -state, or create separate instances to isolate them. For application-level -retries combine with the [tenacity integration](tenacity.md) and read -[Retries and circuit breakers](../guides/retries.md) first. +holds one registry of resolved breakers; reuse the instance across sessions to +share breaker state, or create separate instances to isolate them. For +application-level retries combine with the [tenacity integration](tenacity.md) +and read [Retries and circuit breakers](../guides/retries.md) first. --- @@ -3240,10 +3318,35 @@ created lazily and shared across requests. When a host's circuit is open the request raises [`CircuitOpenError`](../reference.md) *before* a connection is made. +## Custom breaker keys + +Pass `name_resolver` when the request host is not the logical dependency +identity. The callback receives the native `requests.PreparedRequest` and +returns the breaker name. For example, the first path segment can separate +independent upstreams behind one gateway host: + +```python +from interlock.integrations.requests import CircuitBreakerAdapter + +adapter = CircuitBreakerAdapter( + name_resolver=lambda request: request.path_url.split('/')[1], +) +``` + +Returning one name for several discovery hosts instead makes them share a +breaker. The result must be non-empty and contain something other than +whitespace; invalid names raise `ValueError` with the request URL before the +adapter performs I/O. + +The resolved name is used consistently as the registry key, in +`CircuitOpenError`, and in every listener event. Resolve the identity here +instead of rewriting listener labels so metrics remain aligned with breaker +state. + ## Share one registry across sessions Inject a caller-owned `Registry` when independent sessions should use one -breaker and one sliding window for the same host: +breaker and one sliding window for the same resolved name: ```python import requests diff --git a/interlock/integrations/aiohttp.py b/interlock/integrations/aiohttp.py index dc0e185..d5d2973 100644 --- a/interlock/integrations/aiohttp.py +++ b/interlock/integrations/aiohttp.py @@ -1,7 +1,8 @@ """aiohttp client integration — requires the ``aiohttp`` extra. Pass ``CircuitBreakerMiddleware`` to a session and every request it sends is -guarded by a circuit breaker **per host** — no decorators in call sites:: +guarded by a circuit breaker **per host by default** — no decorators in call +sites:: import aiohttp from interlock.integrations.aiohttp import CircuitBreakerMiddleware @@ -26,7 +27,7 @@ """ import http -from collections.abc import Iterable +from collections.abc import Callable, Iterable from typing import cast from aiohttp import ClientHandlerType, ClientRequest, ClientResponse @@ -81,19 +82,44 @@ def is_failure(self, *, result: object, exception: BaseException | None) -> bool return cast('ClientResponse', result).status in self._failure_statuses +def _host(request: ClientRequest) -> str: + """Return the host key, rejecting URLs that cannot identify a dependency.""" + host = request.url.host + if not host: + raise ValueError(f'Request URL has no host to key a breaker on: {request.url!s}') + + return host + + +def _breaker_name( + request: ClientRequest, + name_resolver: Callable[[ClientRequest], str], +) -> str: + """Resolve and validate the dependency identity before registry access.""" + name = name_resolver(request) + if not name.strip(): + raise ValueError( + f'Name resolver returned an empty breaker name for request URL: {request.url!s}' + ) + + return name + + class CircuitBreakerMiddleware: - """A client middleware that guards each host with a circuit breaker. + """A client middleware that guards each dependency with a circuit breaker. Args: - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers; inject a fake for deterministic tests. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. Raises: ValueError: If ``initial_state`` is unsupported or ``registry`` is @@ -110,7 +136,9 @@ def __init__( # noqa: PLR0913 - mirrors Registry's breaker collaborators classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[ClientRequest], str] = _host, ) -> None: + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -123,7 +151,7 @@ def __init__( # noqa: PLR0913 - mirrors Registry's breaker collaborators @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry async def aclose(self) -> None: @@ -132,17 +160,14 @@ async def aclose(self) -> None: await self._registry.aclose_all() async def __call__(self, request: ClientRequest, handler: ClientHandlerType) -> ClientResponse: - """Run the request under its host's breaker. + """Run the request under its resolved dependency's breaker. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL carries no host to key a breaker on. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - host = request.url.host - if not host: - raise ValueError(f'Request URL has no host to key a breaker on: {request.url!s}') - - breaker = self._registry.get(host) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) # The composed handler is not guaranteed to be a coroutine *function* # (middleware chains may hand over plain callables returning diff --git a/interlock/integrations/httpx.py b/interlock/integrations/httpx.py index 3b6f0c2..2efb8d6 100644 --- a/interlock/integrations/httpx.py +++ b/interlock/integrations/httpx.py @@ -10,8 +10,8 @@ transport = CircuitBreakerTransport(httpx.HTTPTransport()) client = httpx.Client(transport=transport) -The wrapper applies one circuit breaker **per host** transparently: no -decorators in user code. Each host gets its own breaker (a slow or failing +The wrapper applies one circuit breaker **per host by default** transparently: +no decorators in user code. Each host gets its own breaker (a slow or failing ``api.a`` must not trip ``api.b``), created lazily and shared across requests. Responses are returned unchanged, preserving httpx's streaming semantics. @@ -22,7 +22,7 @@ """ import http -from collections.abc import Iterable +from collections.abc import Callable, Iterable from contextlib import AsyncExitStack, ExitStack from types import TracebackType from typing import Self, cast @@ -92,19 +92,32 @@ def _host(request: Request) -> str: return host +def _breaker_name(request: Request, name_resolver: Callable[[Request], str]) -> str: + """Resolve and validate the dependency identity before registry access.""" + name = name_resolver(request) + if not name.strip(): + raise ValueError( + f'Name resolver returned an empty breaker name for request URL: {request.url!s}' + ) + + return name + + class CircuitBreakerTransport(BaseTransport): - """Guard every host reached by a synchronous httpx transport. + """Guard every dependency reached by a synchronous httpx transport. Args: transport: Wrapped transport that performs requests. - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. Raises: ValueError: If ``initial_state`` is unsupported or ``registry`` is @@ -122,8 +135,10 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[Request], str] = _host, ) -> None: self._transport = transport + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -136,17 +151,18 @@ def __init__( @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry def handle_request(self, request: Request) -> Response: - """Run a request under the breaker for its host. + """Run a request under the breaker for its resolved dependency. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL has no host. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - breaker = self._registry.get(_host(request)) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_request) return guarded(request) @@ -167,7 +183,7 @@ def __exit__( self._transport.__exit__(exc_type, exc_value, traceback) def close(self) -> None: - """Release the wrapped transport and every owned per-host breaker.""" + """Release the wrapped transport and every owned breaker.""" with ExitStack() as stack: stack.callback(self._close_registry) self._transport.close() @@ -178,18 +194,20 @@ def _close_registry(self) -> None: class AsyncCircuitBreakerTransport(AsyncBaseTransport): - """Guard every host reached by an asynchronous httpx transport. + """Guard every dependency reached by an asynchronous httpx transport. Args: transport: Wrapped async transport that performs requests. - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. Raises: ValueError: If ``initial_state`` is unsupported or ``registry`` is @@ -207,8 +225,10 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[Request], str] = _host, ) -> None: self._transport = transport + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -221,17 +241,18 @@ def __init__( @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry async def handle_async_request(self, request: Request) -> Response: - """Run a request under the breaker for its host. + """Run a request under the breaker for its resolved dependency. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL has no host. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - breaker = self._registry.get(_host(request)) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_async_request) return await guarded(request) @@ -252,7 +273,7 @@ async def __aexit__( await self._transport.__aexit__(exc_type, exc_value, traceback) async def aclose(self) -> None: - """Release the wrapped transport and every owned per-host breaker.""" + """Release the wrapped transport and every owned breaker.""" async with AsyncExitStack() as stack: stack.push_async_callback(self._aclose_registry) await self._transport.aclose() diff --git a/interlock/integrations/httpx2.py b/interlock/integrations/httpx2.py index 6608d57..ec31071 100644 --- a/interlock/integrations/httpx2.py +++ b/interlock/integrations/httpx2.py @@ -10,8 +10,8 @@ transport = CircuitBreakerTransport(httpx2.HTTPTransport()) client = httpx2.Client(transport=transport) -The wrapper applies one circuit breaker **per host** transparently: no -decorators in user code. Each host gets its own breaker (a slow or failing +The wrapper applies one circuit breaker **per host by default** transparently: +no decorators in user code. Each host gets its own breaker (a slow or failing ``api.a`` must not trip ``api.b``), created lazily and shared across requests. By default a response counts as a failure when its status is one of @@ -21,7 +21,7 @@ """ import http -from collections.abc import Iterable +from collections.abc import Callable, Iterable from contextlib import AsyncExitStack, ExitStack from types import TracebackType from typing import Self, cast @@ -91,20 +91,33 @@ def _host(request: Request) -> str: return host +def _breaker_name(request: Request, name_resolver: Callable[[Request], str]) -> str: + """Resolve and validate the dependency identity before registry access.""" + name = name_resolver(request) + if not name.strip(): + raise ValueError( + f'Name resolver returned an empty breaker name for request URL: {request.url!s}' + ) + + return name + + class CircuitBreakerTransport(BaseTransport): - """A synchronous transport that guards each host with a circuit breaker. + """A synchronous transport that guards each dependency with a circuit breaker. Args: transport: The wrapped transport that performs the actual request. - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers; inject a fake for deterministic tests. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. Raises: ValueError: If ``initial_state`` is unsupported or ``registry`` is @@ -122,8 +135,10 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[Request], str] = _host, ) -> None: self._transport = transport + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -136,17 +151,18 @@ def __init__( @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry def handle_request(self, request: Request) -> Response: - """Run the request under its host's breaker. + """Run the request under its resolved dependency's breaker. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL carries no host to key a breaker on. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - breaker = self._registry.get(_host(request)) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_request) return guarded(request) @@ -167,7 +183,7 @@ def __exit__( self._transport.__exit__(exc_type, exc_value, traceback) def close(self) -> None: - """Release the wrapped transport and every owned per-host breaker.""" + """Release the wrapped transport and every owned breaker.""" with ExitStack() as stack: stack.callback(self._close_registry) self._transport.close() @@ -178,19 +194,21 @@ def _close_registry(self) -> None: class AsyncCircuitBreakerTransport(AsyncBaseTransport): - """An asynchronous transport that guards each host with a circuit breaker. + """An asynchronous transport that guards each dependency with a circuit breaker. Args: transport: The wrapped async transport that performs the request. - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers; inject a fake for deterministic tests. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. Raises: ValueError: If ``initial_state`` is unsupported or ``registry`` is @@ -208,8 +226,10 @@ def __init__( classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[Request], str] = _host, ) -> None: self._transport = transport + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -222,17 +242,18 @@ def __init__( @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry async def handle_async_request(self, request: Request) -> Response: - """Run the request under its host's breaker. + """Run the request under its resolved dependency's breaker. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL carries no host to key a breaker on. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - breaker = self._registry.get(_host(request)) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_async_request) return await guarded(request) @@ -253,7 +274,7 @@ async def __aexit__( await self._transport.__aexit__(exc_type, exc_value, traceback) async def aclose(self) -> None: - """Release the wrapped transport and every owned per-host breaker.""" + """Release the wrapped transport and every owned breaker.""" async with AsyncExitStack() as stack: stack.push_async_callback(self._aclose_registry) await self._transport.aclose() diff --git a/interlock/integrations/requests.py b/interlock/integrations/requests.py index db09d16..3aac8b9 100644 --- a/interlock/integrations/requests.py +++ b/interlock/integrations/requests.py @@ -1,7 +1,8 @@ """requests integration — requires the ``requests`` extra. Mount ``CircuitBreakerAdapter`` on a session and every request it sends is -guarded by a circuit breaker **per host** — no decorators in call sites:: +guarded by a circuit breaker **per host by default** — no decorators in call +sites:: import requests from interlock.integrations.requests import CircuitBreakerAdapter @@ -22,7 +23,7 @@ """ import http -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from contextlib import ExitStack from typing import cast from urllib.parse import urlsplit @@ -83,19 +84,44 @@ def is_failure(self, *, result: object, exception: BaseException | None) -> bool return cast('Response', result).status_code in self._failure_statuses +def _host(request: PreparedRequest) -> str: + """Return the host key, rejecting URLs that cannot identify a dependency.""" + host = urlsplit(request.url or '').hostname + if not host: + raise ValueError(f'Request URL has no host to key a breaker on: {request.url!r}') + + return host + + +def _breaker_name( + request: PreparedRequest, + name_resolver: Callable[[PreparedRequest], str], +) -> str: + """Resolve and validate the dependency identity before registry access.""" + name = name_resolver(request) + if not name.strip(): + raise ValueError( + f'Name resolver returned an empty breaker name for request URL: {request.url!s}' + ) + + return name + + class CircuitBreakerAdapter(HTTPAdapter): - """An ``HTTPAdapter`` that guards each host with a circuit breaker. + """An ``HTTPAdapter`` that guards each dependency with a circuit breaker. Args: - config: Thresholds, window and timing for every host's breaker. + config: Thresholds, window and timing for every breaker. clock: Time source for the breakers; inject a fake for deterministic tests. - initial_state: Stable state assigned before a host's first request. + initial_state: Stable state assigned before a breaker's first request. Use ``State.METRICS_ONLY`` for shadow mode. classifier: Failure policy. Defaults to ``HttpStatusClassifier``. - listener: Observability hooks shared by every host's breaker. + listener: Observability hooks shared by every breaker. registry: Caller-owned registry shared with other clients. Mutually exclusive with all breaker-construction options above. + name_resolver: Maps each request to its breaker name. Defaults to the + request host. adapter_kwargs: Passed through to ``HTTPAdapter`` (pool sizes, ``max_retries``, ...). @@ -114,9 +140,11 @@ def __init__( # noqa: PLR0913 - mirrors Registry's breaker collaborators classifier: FailureClassifier | None = None, listener: CoreEventListener | StorageEventListener | None = None, registry: Registry | None = None, + name_resolver: Callable[[PreparedRequest], str] = _host, **adapter_kwargs: object, ) -> None: super().__init__(**adapter_kwargs) # type: ignore[arg-type] + self._name_resolver = name_resolver self._registry, self._owns_registry = resolve_registry( registry=registry, config=config, @@ -129,7 +157,7 @@ def __init__( # noqa: PLR0913 - mirrors Registry's breaker collaborators @property def registry(self) -> Registry: - """The per-host registry, exposed for diagnostics and operator control.""" + """The breaker registry, exposed for diagnostics and operator control.""" return self._registry def close(self) -> None: @@ -151,17 +179,14 @@ def send( # noqa: PLR0913, PLR0917 - mirrors HTTPAdapter.send, the native exten cert: _Cert = None, proxies: Mapping[str, str] | None = None, ) -> Response: - """Run the request under its host's breaker. + """Run the request under its resolved dependency's breaker. Raises: - CircuitOpenError: If the host's breaker is open. - ValueError: If the request URL carries no host to key a breaker on. + CircuitOpenError: If the resolved dependency's breaker is open. + ValueError: If the request URL has no host under the default + resolver, or the configured resolver returns an empty name. """ - host = urlsplit(request.url or '').hostname - if not host: - raise ValueError(f'Request URL has no host to key a breaker on: {request.url!r}') - - breaker = self._registry.get(host) + breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(super().send) return guarded( request, diff --git a/tests/test_aiohttp.py b/tests/test_aiohttp.py index d55afb6..6370d33 100644 --- a/tests/test_aiohttp.py +++ b/tests/test_aiohttp.py @@ -105,6 +105,51 @@ async def test__middleware__open_host__other_host_unaffected(fake_clock: FakeClo assert handler.calls == 3 +@pytest.mark.asyncio +async def test__middleware__name_resolver__collapses_hosts_onto_one_breaker( + fake_clock: FakeClock, +) -> None: + middleware = CircuitBreakerMiddleware( + config=_TRIP_FAST, + clock=fake_clock, + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), + ) + handler = _handler([503, 503]) + + await middleware( + _request('https://orders.query.consul/v1'), + cast('ClientHandlerType', handler), + ) + await middleware(_request('https://orders/v2'), cast('ClientHandlerType', handler)) + + with pytest.raises(CircuitOpenError): + await middleware( + _request('https://orders.query.consul/v3'), + cast('ClientHandlerType', handler), + ) + breaker = middleware.registry.get_existing('orders') + assert breaker is not None + assert breaker.snapshot().failed_calls == 2 + + +@pytest.mark.asyncio +async def test__middleware__empty_resolved_name__raises_before_handler( + fake_clock: FakeClock, +) -> None: + middleware = CircuitBreakerMiddleware( + clock=fake_clock, + name_resolver=lambda _request: ' ', + ) + handler = _handler([200]) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='empty breaker name') as raised: + await middleware(request, cast('ClientHandlerType', handler)) + + assert str(request.url) in str(raised.value) + assert handler.calls == 0 + + @pytest.mark.asyncio async def test__middleware__client_errors__do_not_trip(fake_clock: FakeClock) -> None: middleware = CircuitBreakerMiddleware(config=_TRIP_FAST, clock=fake_clock) diff --git a/tests/test_httpx.py b/tests/test_httpx.py index ad703f8..307aa8b 100644 --- a/tests/test_httpx.py +++ b/tests/test_httpx.py @@ -239,6 +239,64 @@ def handler(request: httpx.Request) -> httpx.Response: assert transport.handle_request(_request('https://good.example.com/')).status_code == 200 +def test__sync_transport__name_resolver__collapses_hosts_onto_one_breaker( + fake_clock: FakeClock, +) -> None: + inner = _SyncStub(lambda _request: httpx.Response(503)) + transport = CircuitBreakerTransport( + inner, + config=_TRIP_FAST, + clock=fake_clock, + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), + ) + + transport.handle_request(_request('https://orders.query.consul/v1')) + transport.handle_request(_request('https://orders/v2')) + + with pytest.raises(CircuitOpenError): + transport.handle_request(_request('https://orders.query.consul/v3')) + breaker = transport.registry.get_existing('orders') + assert breaker is not None + assert breaker.snapshot().failed_calls == 2 + assert transport.registry.get_existing('orders.query.consul') is None + + +def test__sync_transport__name_resolver__aligns_listener_name( + fake_clock: FakeClock, + listener: RecordingListener, +) -> None: + transport = CircuitBreakerTransport( + _SyncStub(lambda _request: httpx.Response(200)), + clock=fake_clock, + listener=listener, + name_resolver=lambda _request: 'orders', + ) + + transport.handle_request(_request()) + + assert listener.names == ['orders'] + + +@pytest.mark.parametrize('resolved_name', ['', ' ']) +def test__sync_transport__empty_resolved_name__raises_before_io( + fake_clock: FakeClock, + resolved_name: str, +) -> None: + inner = _SyncStub(lambda _request: httpx.Response(200)) + transport = CircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda _request: resolved_name, + ) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='empty breaker name') as raised: + transport.handle_request(request) + + assert str(request.url) in str(raised.value) + assert inner.calls == 0 + + def test__sync_transports__shared_registry__merge_window( fake_clock: FakeClock, ) -> None: @@ -471,6 +529,58 @@ def handler(request: httpx.Request) -> httpx.Response: assert response.status_code == 200 +@pytest.mark.asyncio +async def test__async_transport__name_resolver__splits_host_by_path( + fake_clock: FakeClock, +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return ( + httpx.Response(503) if request.url.path.startswith('/orders') else httpx.Response(200) + ) + + inner = _AsyncStub(handler) + transport = AsyncCircuitBreakerTransport( + inner, + config=_TRIP_FAST, + clock=fake_clock, + name_resolver=lambda request: request.url.path.split('/')[1], + ) + + await transport.handle_async_request(_request('https://gateway.example.com/orders/1')) + await transport.handle_async_request(_request('https://gateway.example.com/orders/2')) + + with pytest.raises(CircuitOpenError): + await transport.handle_async_request(_request('https://gateway.example.com/orders/3')) + response = await transport.handle_async_request( + _request('https://gateway.example.com/payments/1') + ) + orders = transport.registry.get_existing('orders') + payments = transport.registry.get_existing('payments') + assert response.status_code == 200 + assert orders is not None + assert payments is not None + assert orders is not payments + + +@pytest.mark.asyncio +async def test__async_transport__empty_resolved_name__raises_before_io( + fake_clock: FakeClock, +) -> None: + inner = _AsyncStub(lambda _request: httpx.Response(200)) + transport = AsyncCircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda _request: '', + ) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='empty breaker name') as raised: + await transport.handle_async_request(request) + + assert str(request.url) in str(raised.value) + assert inner.calls == 0 + + @pytest.mark.asyncio async def test__async_transports__shared_registry__survives_one_transport_closing( fake_clock: FakeClock, diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py index fd841af..82cabc3 100644 --- a/tests/test_httpx2.py +++ b/tests/test_httpx2.py @@ -192,6 +192,36 @@ def handler(request: Request) -> Response: assert transport.handle_request(_request('https://good.example.com/')).status_code == 200 +def test__sync_transport__name_resolver__uses_custom_breaker_key(fake_clock: FakeClock) -> None: + inner = _SyncStub(lambda _request: Response(200)) + transport = CircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda request: request.url.host.removesuffix('.query.consul'), + ) + + transport.handle_request(_request('https://orders.query.consul/v1')) + + assert transport.registry.get_existing('orders') is not None + assert transport.registry.get_existing('orders.query.consul') is None + + +def test__sync_transport__empty_resolved_name__raises_before_io(fake_clock: FakeClock) -> None: + inner = _SyncStub(lambda _request: Response(200)) + transport = CircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda _request: ' ', + ) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='empty breaker name') as raised: + transport.handle_request(request) + + assert str(request.url) in str(raised.value) + assert inner.calls == 0 + + def test__sync_transport__close__releases_wrapped_transport_and_registry( mocker: MockerFixture, ) -> None: @@ -244,6 +274,23 @@ async def test__async_transport__success_response__passes_through(fake_clock: Fa assert response.status_code == 200 +@pytest.mark.asyncio +async def test__async_transport__name_resolver__uses_custom_breaker_key( + fake_clock: FakeClock, +) -> None: + inner = _AsyncStub(lambda _request: Response(200)) + transport = AsyncCircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda request: request.url.path.split('/')[1], + ) + + await transport.handle_async_request(_request('https://gateway.example.com/orders/1')) + + assert transport.registry.get_existing('orders') is not None + assert transport.registry.get_existing('gateway.example.com') is None + + @pytest.mark.asyncio async def test__async_transport__context_manager__delegates_wrapped_lifecycle() -> None: inner = _AsyncLifecycleStub() diff --git a/tests/test_requests.py b/tests/test_requests.py index 4c64bc3..b402c57 100644 --- a/tests/test_requests.py +++ b/tests/test_requests.py @@ -90,6 +90,50 @@ def test__adapter__open_host__other_host_unaffected( assert transport.call_count == 3 +def test__adapter__name_resolver__splits_host_by_path( + mocker: MockerFixture, + fake_clock: FakeClock, +) -> None: + transport = _patch_transport(mocker, [_response(503), _response(503), _response(200)]) + adapter = CircuitBreakerAdapter( + config=_TRIP_FAST, + clock=fake_clock, + name_resolver=lambda request: request.path_url.split('/')[1], + ) + + adapter.send(_prepared('https://gateway.example.com/orders/1')) + adapter.send(_prepared('https://gateway.example.com/orders/2')) + + with pytest.raises(CircuitOpenError): + adapter.send(_prepared('https://gateway.example.com/orders/3')) + response = adapter.send(_prepared('https://gateway.example.com/payments/1')) + orders = adapter.registry.get_existing('orders') + payments = adapter.registry.get_existing('payments') + assert response.status_code == 200 + assert orders is not None + assert payments is not None + assert orders is not payments + assert transport.call_count == 3 + + +def test__adapter__empty_resolved_name__raises_before_transport( + mocker: MockerFixture, + fake_clock: FakeClock, +) -> None: + transport = _patch_transport(mocker, [_response(200)]) + adapter = CircuitBreakerAdapter( + clock=fake_clock, + name_resolver=lambda _request: '', + ) + request = _prepared('https://api.example.com/v1') + + with pytest.raises(ValueError, match='empty breaker name') as raised: + adapter.send(request) + + assert request.url in str(raised.value) + assert transport.call_count == 0 + + def test__adapter__client_errors__do_not_trip(mocker: MockerFixture, fake_clock: FakeClock) -> None: transport = _patch_transport(mocker, [_response(404)] * 5) adapter = CircuitBreakerAdapter(config=_TRIP_FAST, clock=fake_clock) From ec695be6008ff6d0b5b469687a815ab40e281d2c Mon Sep 17 00:00:00 2001 From: bagowix Date: Sat, 8 Aug 2026 00:34:43 +0300 Subject: [PATCH 2/3] fix: validate resolved transport breaker names --- CHANGELOG.md | 2 +- docs/integrations/aiohttp.md | 5 ++-- docs/integrations/httpx.md | 5 ++-- docs/integrations/httpx2.md | 6 ++--- docs/integrations/requests.md | 4 +-- docs/llms-full.txt | 20 ++++++++------- interlock/integrations/aiohttp.py | 9 +++++-- interlock/integrations/httpx.py | 12 ++++++--- interlock/integrations/httpx2.py | 12 ++++++--- interlock/integrations/requests.py | 17 ++++++++++--- tests/test_aiohttp.py | 20 +++++++++++++++ tests/test_httpx.py | 22 ++++++++++++++++- tests/test_httpx2.py | 22 ++++++++++++++++- tests/test_requests.py | 39 ++++++++++++++++++++++++++++++ 14 files changed, 162 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff8a4af..005d2ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,7 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). breakers. Their new `name_resolver=` callback receives the native request and supplies the single name used by the registry, open-circuit errors, and listener events. Host-based naming remains the default, while empty custom - names fail before any network I/O. + names and non-string results fail before any network I/O. - **Listener contracts now match the events a component actually emits.** `CoreEventListener`, `StorageEventListener` and `PipelineEventListener` let a diff --git a/docs/integrations/aiohttp.md b/docs/integrations/aiohttp.md index f7788e6..5a16bd4 100644 --- a/docs/integrations/aiohttp.md +++ b/docs/integrations/aiohttp.md @@ -67,8 +67,9 @@ middleware = CircuitBreakerMiddleware( A resolver can collapse several discovery hosts onto one breaker or derive a name from the request path to separate upstreams behind a shared gateway. It -must return a non-empty, non-whitespace name; invalid names raise `ValueError` -with the request URL before the handler performs I/O. +must return a non-empty string containing something other than whitespace; +invalid results raise `ValueError` with the request URL before the handler +performs I/O. The resolved name is used by the registry, `CircuitOpenError`, and every listener event. Resolve it in the middleware rather than rewriting listener diff --git a/docs/integrations/httpx.md b/docs/integrations/httpx.md index 5ae8456..36439f6 100644 --- a/docs/integrations/httpx.md +++ b/docs/integrations/httpx.md @@ -105,8 +105,9 @@ transport = AsyncCircuitBreakerTransport( The same callback can split one gateway host into independent breakers, for example by returning a name derived from the first path segment. It must return -a non-empty, non-whitespace name; invalid names raise `ValueError` with the -request URL before the wrapped transport performs I/O. +a non-empty string containing something other than whitespace; invalid results +raise `ValueError` with the request URL before the wrapped transport performs +I/O. The resolved name is the registry key and the name carried by `CircuitOpenError` and every listener event. Use the resolver, rather than diff --git a/docs/integrations/httpx2.md b/docs/integrations/httpx2.md index 83fd70c..acbf0df 100644 --- a/docs/integrations/httpx2.md +++ b/docs/integrations/httpx2.md @@ -111,9 +111,9 @@ transport = AsyncCircuitBreakerTransport( Returning the same name for several discovery hosts gives them one breaker; deriving a name from the path can split independent upstreams behind one -gateway host. The result must be non-empty and contain something other than -whitespace. Invalid names raise `ValueError` with the request URL before the -wrapped transport performs I/O. +gateway host. The result must be a non-empty string containing something other +than whitespace. Invalid results raise `ValueError` with the request URL before +the wrapped transport performs I/O. The resolved name is used consistently as the registry key, in `CircuitOpenError`, and in every listener event. Resolve the name here instead diff --git a/docs/integrations/requests.md b/docs/integrations/requests.md index dbcb41a..6fb9e63 100644 --- a/docs/integrations/requests.md +++ b/docs/integrations/requests.md @@ -64,8 +64,8 @@ adapter = CircuitBreakerAdapter( ``` Returning one name for several discovery hosts instead makes them share a -breaker. The result must be non-empty and contain something other than -whitespace; invalid names raise `ValueError` with the request URL before the +breaker. The result must be a non-empty string containing something other than +whitespace; invalid results raise `ValueError` with the request URL before the adapter performs I/O. The resolved name is used consistently as the registry key, in diff --git a/docs/llms-full.txt b/docs/llms-full.txt index f4ddf24..3c66aa9 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -2830,9 +2830,9 @@ transport = AsyncCircuitBreakerTransport( Returning the same name for several discovery hosts gives them one breaker; deriving a name from the path can split independent upstreams behind one -gateway host. The result must be non-empty and contain something other than -whitespace. Invalid names raise `ValueError` with the request URL before the -wrapped transport performs I/O. +gateway host. The result must be a non-empty string containing something other +than whitespace. Invalid results raise `ValueError` with the request URL before +the wrapped transport performs I/O. The resolved name is used consistently as the registry key, in `CircuitOpenError`, and in every listener event. Resolve the name here instead @@ -3019,8 +3019,9 @@ transport = AsyncCircuitBreakerTransport( The same callback can split one gateway host into independent breakers, for example by returning a name derived from the first path segment. It must return -a non-empty, non-whitespace name; invalid names raise `ValueError` with the -request URL before the wrapped transport performs I/O. +a non-empty string containing something other than whitespace; invalid results +raise `ValueError` with the request URL before the wrapped transport performs +I/O. The resolved name is the registry key and the name carried by `CircuitOpenError` and every listener event. Use the resolver, rather than @@ -3180,8 +3181,9 @@ middleware = CircuitBreakerMiddleware( A resolver can collapse several discovery hosts onto one breaker or derive a name from the request path to separate upstreams behind a shared gateway. It -must return a non-empty, non-whitespace name; invalid names raise `ValueError` -with the request URL before the handler performs I/O. +must return a non-empty string containing something other than whitespace; +invalid results raise `ValueError` with the request URL before the handler +performs I/O. The resolved name is used by the registry, `CircuitOpenError`, and every listener event. Resolve it in the middleware rather than rewriting listener @@ -3334,8 +3336,8 @@ adapter = CircuitBreakerAdapter( ``` Returning one name for several discovery hosts instead makes them share a -breaker. The result must be non-empty and contain something other than -whitespace; invalid names raise `ValueError` with the request URL before the +breaker. The result must be a non-empty string containing something other than +whitespace; invalid results raise `ValueError` with the request URL before the adapter performs I/O. The resolved name is used consistently as the registry key, in diff --git a/interlock/integrations/aiohttp.py b/interlock/integrations/aiohttp.py index d5d2973..27a52d3 100644 --- a/interlock/integrations/aiohttp.py +++ b/interlock/integrations/aiohttp.py @@ -96,7 +96,11 @@ def _breaker_name( name_resolver: Callable[[ClientRequest], str], ) -> str: """Resolve and validate the dependency identity before registry access.""" - name = name_resolver(request) + name = cast('object', name_resolver(request)) + if not isinstance(name, str): + raise ValueError( # noqa: TRY004 - resolver contract uses ValueError + f'Name resolver returned a non-string breaker name for request URL: {request.url!s}' + ) if not name.strip(): raise ValueError( f'Name resolver returned an empty breaker name for request URL: {request.url!s}' @@ -165,7 +169,8 @@ async def __call__(self, request: ClientRequest, handler: ClientHandlerType) -> Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) diff --git a/interlock/integrations/httpx.py b/interlock/integrations/httpx.py index 2efb8d6..8ecccc5 100644 --- a/interlock/integrations/httpx.py +++ b/interlock/integrations/httpx.py @@ -94,7 +94,11 @@ def _host(request: Request) -> str: def _breaker_name(request: Request, name_resolver: Callable[[Request], str]) -> str: """Resolve and validate the dependency identity before registry access.""" - name = name_resolver(request) + name = cast('object', name_resolver(request)) + if not isinstance(name, str): + raise ValueError( # noqa: TRY004 - resolver contract uses ValueError + f'Name resolver returned a non-string breaker name for request URL: {request.url!s}' + ) if not name.strip(): raise ValueError( f'Name resolver returned an empty breaker name for request URL: {request.url!s}' @@ -160,7 +164,8 @@ def handle_request(self, request: Request) -> Response: Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_request) @@ -250,7 +255,8 @@ async def handle_async_request(self, request: Request) -> Response: Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_async_request) diff --git a/interlock/integrations/httpx2.py b/interlock/integrations/httpx2.py index ec31071..b523a7d 100644 --- a/interlock/integrations/httpx2.py +++ b/interlock/integrations/httpx2.py @@ -93,7 +93,11 @@ def _host(request: Request) -> str: def _breaker_name(request: Request, name_resolver: Callable[[Request], str]) -> str: """Resolve and validate the dependency identity before registry access.""" - name = name_resolver(request) + name = cast('object', name_resolver(request)) + if not isinstance(name, str): + raise ValueError( # noqa: TRY004 - resolver contract uses ValueError + f'Name resolver returned a non-string breaker name for request URL: {request.url!s}' + ) if not name.strip(): raise ValueError( f'Name resolver returned an empty breaker name for request URL: {request.url!s}' @@ -160,7 +164,8 @@ def handle_request(self, request: Request) -> Response: Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_request) @@ -251,7 +256,8 @@ async def handle_async_request(self, request: Request) -> Response: Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(self._transport.handle_async_request) diff --git a/interlock/integrations/requests.py b/interlock/integrations/requests.py index 3aac8b9..6b1437f 100644 --- a/interlock/integrations/requests.py +++ b/interlock/integrations/requests.py @@ -86,9 +86,13 @@ def is_failure(self, *, result: object, exception: BaseException | None) -> bool def _host(request: PreparedRequest) -> str: """Return the host key, rejecting URLs that cannot identify a dependency.""" - host = urlsplit(request.url or '').hostname + url = request.url + if url is None: + raise ValueError(f'Request URL has no host to key a breaker on: {url!r}') + + host = urlsplit(url).hostname if not host: - raise ValueError(f'Request URL has no host to key a breaker on: {request.url!r}') + raise ValueError(f'Request URL has no host to key a breaker on: {url!r}') return host @@ -98,7 +102,11 @@ def _breaker_name( name_resolver: Callable[[PreparedRequest], str], ) -> str: """Resolve and validate the dependency identity before registry access.""" - name = name_resolver(request) + name = cast('object', name_resolver(request)) + if not isinstance(name, str): + raise ValueError( # noqa: TRY004 - resolver contract uses ValueError + f'Name resolver returned a non-string breaker name for request URL: {request.url!s}' + ) if not name.strip(): raise ValueError( f'Name resolver returned an empty breaker name for request URL: {request.url!s}' @@ -184,7 +192,8 @@ def send( # noqa: PLR0913, PLR0917 - mirrors HTTPAdapter.send, the native exten Raises: CircuitOpenError: If the resolved dependency's breaker is open. ValueError: If the request URL has no host under the default - resolver, or the configured resolver returns an empty name. + resolver, or the configured resolver returns a non-string or + empty name. """ breaker = self._registry.get(_breaker_name(request, self._name_resolver)) guarded = breaker(super().send) diff --git a/tests/test_aiohttp.py b/tests/test_aiohttp.py index 6370d33..f191662 100644 --- a/tests/test_aiohttp.py +++ b/tests/test_aiohttp.py @@ -150,6 +150,26 @@ async def test__middleware__empty_resolved_name__raises_before_handler( assert handler.calls == 0 +@pytest.mark.parametrize('resolved_name', [None, b'orders']) +@pytest.mark.asyncio +async def test__middleware__non_string_resolved_name__raises_before_handler( + fake_clock: FakeClock, + resolved_name: object, +) -> None: + middleware = CircuitBreakerMiddleware( + clock=fake_clock, + name_resolver=lambda _request: cast('str', resolved_name), + ) + handler = _handler([200]) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='non-string breaker name') as raised: + await middleware(request, cast('ClientHandlerType', handler)) + + assert str(request.url) in str(raised.value) + assert handler.calls == 0 + + @pytest.mark.asyncio async def test__middleware__client_errors__do_not_trip(fake_clock: FakeClock) -> None: middleware = CircuitBreakerMiddleware(config=_TRIP_FAST, clock=fake_clock) diff --git a/tests/test_httpx.py b/tests/test_httpx.py index 307aa8b..871ba9e 100644 --- a/tests/test_httpx.py +++ b/tests/test_httpx.py @@ -1,6 +1,6 @@ from collections.abc import AsyncIterator, Callable, Iterator from types import TracebackType -from typing import Self +from typing import Self, cast import httpx import pytest @@ -297,6 +297,26 @@ def test__sync_transport__empty_resolved_name__raises_before_io( assert inner.calls == 0 +@pytest.mark.parametrize('resolved_name', [None, b'orders']) +def test__sync_transport__non_string_resolved_name__raises_before_io( + fake_clock: FakeClock, + resolved_name: object, +) -> None: + inner = _SyncStub(lambda _request: httpx.Response(200)) + transport = CircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda _request: cast('str', resolved_name), + ) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='non-string breaker name') as raised: + transport.handle_request(request) + + assert str(request.url) in str(raised.value) + assert inner.calls == 0 + + def test__sync_transports__shared_registry__merge_window( fake_clock: FakeClock, ) -> None: diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py index 82cabc3..57e5366 100644 --- a/tests/test_httpx2.py +++ b/tests/test_httpx2.py @@ -1,6 +1,6 @@ from collections.abc import Callable from types import TracebackType -from typing import Self +from typing import Self, cast import httpx2 import pytest @@ -222,6 +222,26 @@ def test__sync_transport__empty_resolved_name__raises_before_io(fake_clock: Fake assert inner.calls == 0 +@pytest.mark.parametrize('resolved_name', [None, b'orders']) +def test__sync_transport__non_string_resolved_name__raises_before_io( + fake_clock: FakeClock, + resolved_name: object, +) -> None: + inner = _SyncStub(lambda _request: Response(200)) + transport = CircuitBreakerTransport( + inner, + clock=fake_clock, + name_resolver=lambda _request: cast('str', resolved_name), + ) + request = _request('https://api.example.com/v1') + + with pytest.raises(ValueError, match='non-string breaker name') as raised: + transport.handle_request(request) + + assert str(request.url) in str(raised.value) + assert inner.calls == 0 + + def test__sync_transport__close__releases_wrapped_transport_and_registry( mocker: MockerFixture, ) -> None: diff --git a/tests/test_requests.py b/tests/test_requests.py index b402c57..6f326c0 100644 --- a/tests/test_requests.py +++ b/tests/test_requests.py @@ -1,6 +1,7 @@ """Tests for the requests integration (``interlock.integrations.requests``).""" from collections.abc import Callable +from typing import cast import pytest import requests @@ -134,6 +135,26 @@ def test__adapter__empty_resolved_name__raises_before_transport( assert transport.call_count == 0 +@pytest.mark.parametrize('resolved_name', [None, b'orders']) +def test__adapter__non_string_resolved_name__raises_before_transport( + mocker: MockerFixture, + fake_clock: FakeClock, + resolved_name: object, +) -> None: + transport = _patch_transport(mocker, [_response(200)]) + adapter = CircuitBreakerAdapter( + clock=fake_clock, + name_resolver=lambda _request: cast('str', resolved_name), + ) + request = _prepared('https://api.example.com/v1') + + with pytest.raises(ValueError, match='non-string breaker name') as raised: + adapter.send(request) + + assert request.url in str(raised.value) + assert transport.call_count == 0 + + def test__adapter__client_errors__do_not_trip(mocker: MockerFixture, fake_clock: FakeClock) -> None: transport = _patch_transport(mocker, [_response(404)] * 5) adapter = CircuitBreakerAdapter(config=_TRIP_FAST, clock=fake_clock) @@ -260,6 +281,24 @@ def test__adapter__url_without_host__raises_value_error(fake_clock: FakeClock) - adapter.send(request) +def test__adapter__missing_url__raises_before_url_parsing( + mocker: MockerFixture, + fake_clock: FakeClock, +) -> None: + adapter = CircuitBreakerAdapter(config=_TRIP_FAST, clock=fake_clock) + request = PreparedRequest() + request.url = None + urlsplit = mocker.patch( + 'interlock.integrations.requests.urlsplit', + side_effect=AssertionError('missing URL must not reach urlsplit'), + ) + + with pytest.raises(ValueError, match='no host'): + adapter.send(request) + + urlsplit.assert_not_called() + + def test__adapter__send_kwargs__forwarded_to_transport( mocker: MockerFixture, fake_clock: FakeClock ) -> None: From 64a9a5bdd0620920f36ce7d4582b23015fa1ed81 Mon Sep 17 00:00:00 2001 From: bagowix Date: Sat, 8 Aug 2026 00:41:59 +0300 Subject: [PATCH 3/3] test: verify invalid names bypass registry --- tests/test_httpx.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/test_httpx.py b/tests/test_httpx.py index 871ba9e..4751c6f 100644 --- a/tests/test_httpx.py +++ b/tests/test_httpx.py @@ -280,6 +280,7 @@ def test__sync_transport__name_resolver__aligns_listener_name( @pytest.mark.parametrize('resolved_name', ['', ' ']) def test__sync_transport__empty_resolved_name__raises_before_io( fake_clock: FakeClock, + mocker: MockerFixture, resolved_name: str, ) -> None: inner = _SyncStub(lambda _request: httpx.Response(200)) @@ -288,18 +289,21 @@ def test__sync_transport__empty_resolved_name__raises_before_io( clock=fake_clock, name_resolver=lambda _request: resolved_name, ) + registry_get = mocker.spy(transport.registry, 'get') request = _request('https://api.example.com/v1') with pytest.raises(ValueError, match='empty breaker name') as raised: transport.handle_request(request) assert str(request.url) in str(raised.value) + registry_get.assert_not_called() assert inner.calls == 0 @pytest.mark.parametrize('resolved_name', [None, b'orders']) def test__sync_transport__non_string_resolved_name__raises_before_io( fake_clock: FakeClock, + mocker: MockerFixture, resolved_name: object, ) -> None: inner = _SyncStub(lambda _request: httpx.Response(200)) @@ -308,12 +312,14 @@ def test__sync_transport__non_string_resolved_name__raises_before_io( clock=fake_clock, name_resolver=lambda _request: cast('str', resolved_name), ) + registry_get = mocker.spy(transport.registry, 'get') request = _request('https://api.example.com/v1') with pytest.raises(ValueError, match='non-string breaker name') as raised: transport.handle_request(request) assert str(request.url) in str(raised.value) + registry_get.assert_not_called() assert inner.calls == 0