diff --git a/docs/resilience.md b/docs/resilience.md index 8a50237..71c3cea 100644 --- a/docs/resilience.md +++ b/docs/resilience.md @@ -287,7 +287,7 @@ 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 @@ -295,7 +295,7 @@ A circuit is created on the first request for its key and kept for as long as th ### 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 diff --git a/src/httpware/middleware/resilience/_event_loop_guard.py b/src/httpware/middleware/resilience/_event_loop_guard.py index d7b60d3..03983ea 100644 --- a/src/httpware/middleware/resilience/_event_loop_guard.py +++ b/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)) diff --git a/src/httpware/middleware/resilience/bulkhead.py b/src/httpware/middleware/resilience/bulkhead.py index 09111fd..bcba5ff 100644 --- a/src/httpware/middleware/resilience/bulkhead.py +++ b/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): diff --git a/src/httpware/middleware/resilience/circuit_breaker.py b/src/httpware/middleware/resilience/circuit_breaker.py index f529028..4e589e3 100644 --- a/src/httpware/middleware/resilience/circuit_breaker.py +++ b/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,15 +480,19 @@ 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 @@ -479,14 +500,23 @@ def __init__(self, key: Callable[[httpx2.Request], Hashable], **config: typing.A 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) diff --git a/tests/test_bulkhead.py b/tests/test_bulkhead.py index cc054eb..b8e8d2c 100644 --- a/tests/test_bulkhead.py +++ b/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)) diff --git a/tests/test_circuit_breaker_freethreaded_stress.py b/tests/test_circuit_breaker_freethreaded_stress.py index 4a7de88..d9bcdfb 100644 --- a/tests/test_circuit_breaker_freethreaded_stress.py +++ b/tests/test_circuit_breaker_freethreaded_stress.py @@ -11,13 +11,16 @@ """ import threading +import typing from http import HTTPStatus import httpx2 import pytest -from httpware import CircuitBreaker, CircuitState, Client +from httpware import CircuitBreaker, CircuitState, Client, InternalServerError from httpware.errors import CircuitOpenError, StatusError +from httpware.middleware import Next +from httpware.middleware.resilience.circuit_breaker import _FAILURE, _SUCCESS, _HalfOpened _N_THREADS = 16 @@ -58,3 +61,132 @@ def worker() -> None: assert len(errors) == _N_THREADS * _N_OPS assert all(isinstance(exc, (StatusError, CircuitOpenError)) for exc in errors) assert breaker.state is CircuitState.OPEN + + +_FAILURE_THRESHOLD = 3 +_SUCCESS_THRESHOLD = 2 +_FAILURES_PER_TEN = 3 +_N_STRESS_OPS = 2000 + + +class _Recorder: + """Wraps a breaker's state to check its invariants from inside the breaker's lock. + + The sync breaker calls `admit` (when the circuit is not closed) and `record` under its lock, + so the snapshots taken there are consistent. + """ + + def __init__(self, breaker: CircuitBreaker, monkeypatch: pytest.MonkeyPatch) -> None: + self.state = breaker._state # noqa: SLF001 + self.admit = self.state.admit + self.record = self.state.record + self.probes_in_flight = 0 + self.max_probes_in_flight = 0 + self.records: list[tuple[bool, tuple[typing.Any, ...], tuple[typing.Any, ...], bool, bool]] = [] + self.probe_outcomes: list[tuple[int, int]] = [] + monkeypatch.setattr(self.state, "admit", self._admit) + monkeypatch.setattr(self.state, "record", self._record) + + def _snapshot(self) -> tuple[typing.Any, ...]: + state = self.state + return ( + state.state, + state._epoch, # noqa: SLF001 + state.closed_ticket, + state._consecutive_failures, # noqa: SLF001 + state._consecutive_successes, # noqa: SLF001 + state._probe_in_flight, # noqa: SLF001 + ) + + def _admitted(self, ticket: int) -> None: + if ticket % 2: + self.probes_in_flight += 1 + self.max_probes_in_flight = max(self.max_probes_in_flight, self.probes_in_flight) + + def _admit(self, request: httpx2.Request) -> int: + try: + ticket = self.admit(request) + except _HalfOpened as half_opened: + self._admitted(half_opened.ticket) + raise + self._admitted(ticket) + return ticket + + def _record(self, ticket: int, outcome: int, request: httpx2.Request) -> list[typing.Any] | None: + if ticket % 2: + self.probes_in_flight -= 1 + self.probe_outcomes.append((ticket, outcome)) + before = self._snapshot() + stale = ticket not in (before[1], before[1] + 1) + events = self.record(ticket, outcome, request) + after = self._snapshot() + circuit, epoch, closed_ticket, failures, successes, probe_in_flight = after + consistent = ( + closed_ticket == (epoch if circuit is CircuitState.CLOSED else None) + and failures < _FAILURE_THRESHOLD + and successes < _SUCCESS_THRESHOLD + and probe_in_flight == (self.probes_in_flight == 1) + ) + self.records.append((stale, before, after, events is None, consistent)) + return events + + +def _next(fail: bool) -> Next: + def _send(request: httpx2.Request) -> httpx2.Response: + if fail: + raise InternalServerError(httpx2.Response(HTTPStatus.INTERNAL_SERVER_ERROR, request=request)) + return httpx2.Response(HTTPStatus.OK, request=request) + + return _send + + +def _probe_eras(probe_outcomes: list[tuple[int, int]]) -> list[list[int]]: + eras: dict[int, list[int]] = {} + for ticket, outcome in probe_outcomes: + eras.setdefault(ticket, []).append(outcome) + return list(eras.values()) + + +@pytest.mark.stress +def test_lock_free_admission_keeps_the_circuit_invariants(monkeypatch: pytest.MonkeyPatch) -> None: + """Probes never overlap, success_threshold probes close the circuit, stale outcomes change nothing.""" + breaker = CircuitBreaker( + failure_threshold=_FAILURE_THRESHOLD, + reset_timeout=0.0, + success_threshold=_SUCCESS_THRESHOLD, + ) + recorder = _Recorder(breaker, monkeypatch) + succeeded: list[int] = [] + errors: list[Exception] = [] + guard = threading.Lock() + + def worker(seed: int) -> None: + request = httpx2.Request("GET", "https://example.test/x") + for op in range(_N_STRESS_OPS): + try: + breaker(request, _next((seed + op * op) % 10 < _FAILURES_PER_TEN)) + with guard: + succeeded.append(op) + except Exception as exc: # noqa: BLE001 + with guard: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(_N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(succeeded) + len(errors) == _N_THREADS * _N_STRESS_OPS + assert all(isinstance(exc, (CircuitOpenError, InternalServerError)) for exc in errors) + assert recorder.max_probes_in_flight == 1 + assert recorder.probes_in_flight == 0 + assert all(consistent for *_, consistent in recorder.records) + assert all(after == before and no_events for stale, before, after, no_events, _ in recorder.records if stale) + eras = _probe_eras(recorder.probe_outcomes) + assert eras + for era in eras: + successes = era.count(_SUCCESS) + assert successes <= _SUCCESS_THRESHOLD + assert era in ([_SUCCESS] * successes, [*[_SUCCESS] * successes, _FAILURE]) + assert all(_FAILURE in era or len(era) == _SUCCESS_THRESHOLD for era in eras[:-1]) diff --git a/tests/test_circuit_breaker_sync.py b/tests/test_circuit_breaker_sync.py index 7c7eb36..524099e 100644 --- a/tests/test_circuit_breaker_sync.py +++ b/tests/test_circuit_breaker_sync.py @@ -978,3 +978,15 @@ def test_suppressed_never_counts_a_previous_outage(caplog: pytest.LogCaptureFixt _open_now(breaker) _reject(breaker) assert [r.suppressed for r in _events(caplog, "circuit.rejected")] == [0, 0] # ty: ignore[unresolved-attribute] + + +def test_locked_admission_into_a_circuit_closed_since_the_unlocked_read_gets_the_closed_ticket() -> None: + """The sync slow path admits under the lock after an unlocked read saw a circuit that was not closed.""" + breaker = CircuitBreaker(failure_threshold=1, reset_timeout=0.0) + with _client(_StatusSequence([HTTPStatus.INTERNAL_SERVER_ERROR]), breaker=breaker) as client: + with pytest.raises(InternalServerError): + client.get("https://example.test/x") + client.get("https://example.test/x") + state = breaker._state # noqa: SLF001 + assert state.state is CircuitState.CLOSED + assert state.admit(httpx2.Request("GET", "https://example.test/x")) == state.closed_ticket == state._epoch # noqa: SLF001 diff --git a/tests/test_circuit_breaker_sync_props.py b/tests/test_circuit_breaker_sync_props.py new file mode 100644 index 0000000..4f0ed0f --- /dev/null +++ b/tests/test_circuit_breaker_sync_props.py @@ -0,0 +1,165 @@ +"""Property tests for the sync CircuitBreaker state machine. + +Mirrors the interleaving test in tests/test_circuit_breaker_props.py. Each call runs on its own +thread and blocks inside `next` until the test finishes it, so calls overlap in any order the +strategy picks while the steps themselves stay sequential and deterministic. +""" + +import dataclasses +import threading +import typing +from http import HTTPStatus + +import httpx2 +from hypothesis import given, settings +from hypothesis import strategies as st + +from httpware import CircuitOpenError, InternalServerError +from httpware.middleware.resilience.circuit_breaker import CircuitBreaker, CircuitState + + +_FOREIGN = 0 + + +def _request() -> httpx2.Request: + return httpx2.Request("GET", "https://example.test/x") + + +@dataclasses.dataclass +class _Call: + generation: int + settled: threading.Event = dataclasses.field(default_factory=threading.Event) + release: threading.Event = dataclasses.field(default_factory=threading.Event) + status: int = HTTPStatus.OK + probe: bool = False + error: Exception | None = None + thread: threading.Thread = dataclasses.field(init=False) + + +class _Harness: + """Starts and finishes sync breaker calls, checking the circuit's invariants after every step. + + The generation counts observed state changes; each step causes at most one. + """ + + def __init__(self, breaker: CircuitBreaker, *, failure_threshold: int, success_threshold: int) -> None: + self.breaker = breaker + self.failure_threshold = failure_threshold + self.success_threshold = success_threshold + self.generation = 0 + self.observed = breaker.state + self.in_flight: list[_Call] = [] + self.probes_in_flight = 0 + self.probe_successes = 0 + + def _observe(self) -> None: + state = self.breaker.state + if state is not self.observed: + if self.observed is CircuitState.HALF_OPEN and state is CircuitState.CLOSED: + assert self.probe_successes == self.success_threshold + if state is CircuitState.HALF_OPEN: + self.probe_successes = 0 + self.generation += 1 + self.observed = state + self._assert_consistent() + + def _assert_consistent(self) -> None: + state = self.breaker._state # noqa: SLF001 + closed = state.state is CircuitState.CLOSED + assert state.closed_ticket == (state._epoch if closed else None) # noqa: SLF001 + assert state._probe_in_flight == (self.probes_in_flight == 1) # noqa: SLF001 + assert state._consecutive_failures < self.failure_threshold # noqa: SLF001 + assert state._consecutive_successes < self.success_threshold # noqa: SLF001 + + def _next_for(self, call: _Call) -> typing.Callable[[httpx2.Request], httpx2.Response]: + def _next(request: httpx2.Request) -> httpx2.Response: + call.probe = self.breaker.state is CircuitState.HALF_OPEN + self.probes_in_flight += call.probe + call.settled.set() + call.release.wait() + self.probes_in_flight -= call.probe + if call.status == _FOREIGN: + msg = "foreign" + raise ValueError(msg) + if call.status >= HTTPStatus.INTERNAL_SERVER_ERROR: + raise InternalServerError(httpx2.Response(call.status, request=request)) + return httpx2.Response(call.status, request=request) + + return _next + + def _run(self, call: _Call) -> None: + try: + self.breaker(_request(), self._next_for(call)) + except Exception as exc: # noqa: BLE001 + call.error = exc + finally: + call.settled.set() + + def start(self) -> None: + call = _Call(self.generation) + call.thread = threading.Thread(target=self._run, args=(call,), daemon=True) + call.thread.start() + call.settled.wait() + assert self.probes_in_flight <= 1 + self._observe() + call.generation = self.generation + if isinstance(call.error, CircuitOpenError): + call.thread.join() + else: + self.in_flight.append(call) + + def finish(self, index: int, status: int) -> None: + call = self.in_flight.pop(index % len(self.in_flight)) + before = self.breaker.state + call.status = status + call.release.set() + call.thread.join() + assert call.error is None or isinstance(call.error, (InternalServerError, ValueError)) + if call.probe and status != _FOREIGN and status < HTTPStatus.INTERNAL_SERVER_ERROR: + self.probe_successes += 1 + if call.generation != self.generation: + assert self.breaker.state is before + self._observe() + + def drain(self) -> None: + while self.in_flight: + self.finish(0, HTTPStatus.OK) + + +_operations = st.lists( + st.one_of( + st.just(("start",)), + st.tuples( + st.just("finish"), st.integers(min_value=0, max_value=31), st.sampled_from([200, 404, 500, _FOREIGN]) + ), + ), + min_size=8, + max_size=40, +) + + +@given( + failure_threshold=st.integers(min_value=1, max_value=2), + success_threshold=st.integers(min_value=1, max_value=3), + operations=_operations, +) +@settings(max_examples=100, deadline=None) +def test_outcomes_only_move_the_state_they_were_admitted_under( + failure_threshold: int, + success_threshold: int, + operations: list[tuple[typing.Any, ...]], +) -> None: + """Stale outcomes leave the circuit alone, probes never overlap, and success_threshold probes close it.""" + breaker = CircuitBreaker( + failure_threshold=failure_threshold, + reset_timeout=0.0, + success_threshold=success_threshold, + _now=lambda: 0.0, + ) + harness = _Harness(breaker, failure_threshold=failure_threshold, success_threshold=success_threshold) + for operation in operations: + if operation[0] == "start": + harness.start() + elif harness.in_flight: + harness.finish(operation[1], operation[2]) + harness.drain() diff --git a/tests/test_event_loop_guard.py b/tests/test_event_loop_guard.py index b50ac94..59b34bf 100644 --- a/tests/test_event_loop_guard.py +++ b/tests/test_event_loop_guard.py @@ -8,12 +8,11 @@ async def test_raises_when_another_loop_binds_between_unlocked_and_locked_read() -> None: other_loop = asyncio.new_event_loop() - reads = iter([None, other_loop]) bound: list[asyncio.AbstractEventLoop] = [] try: with pytest.raises(RuntimeError, match="bound to"): check_event_loop( - lambda: next(reads), + lambda: other_loop, bound.append, threading.Lock(), "bound to {first}, called from {current}", diff --git a/tests/test_keyed_circuit_breaker.py b/tests/test_keyed_circuit_breaker.py index 4857ce6..0ac2dfd 100644 --- a/tests/test_keyed_circuit_breaker.py +++ b/tests/test_keyed_circuit_breaker.py @@ -205,6 +205,16 @@ async def test_circuit_key_is_the_normalized_origin_without_userinfo(caplog: pyt assert keys == ["https://a.test", "http://a.test:8080"] +async def test_circuit_key_of_a_custom_key_is_its_str(caplog: pytest.LogCaptureFixture) -> None: + handler = _PerHost(failing={"a.test"}) + breaker = AsyncKeyedCircuitBreaker(failure_threshold=1, key=lambda request: (request.url.host, 1)) + async with _client(handler, breaker=breaker) as client: + with caplog.at_level(logging.WARNING, logger="httpware.circuit_breaker"): + await _fail_n(client, "https://a.test/x", 1) + keys = [r.circuit_key for r in caplog.records if r.name == "httpware.circuit_breaker"] # ty: ignore[unresolved-attribute] + assert keys == ["('a.test', 1)"] + + def test_cross_loop_use_raises_runtimeerror() -> None: breaker = AsyncKeyedCircuitBreaker() handler = _PerHost(failing=set()) diff --git a/tests/test_keyed_circuit_breaker_freethreaded_stress.py b/tests/test_keyed_circuit_breaker_freethreaded_stress.py index 8c10ac9..693003c 100644 --- a/tests/test_keyed_circuit_breaker_freethreaded_stress.py +++ b/tests/test_keyed_circuit_breaker_freethreaded_stress.py @@ -6,13 +6,14 @@ test_circuit_breaker_freethreaded_stress.py for why). """ +import contextlib import threading from http import HTTPStatus import httpx2 import pytest -from httpware import Client, KeyedCircuitBreaker +from httpware import Client, InternalServerError, KeyedCircuitBreaker from httpware.errors import CircuitOpenError, StatusError @@ -55,3 +56,32 @@ def worker(index: int) -> None: with pytest.raises(CircuitOpenError): client.get(f"https://{host}/x") client.close() + + +@pytest.mark.stress +def test_racing_first_requests_share_one_circuit_per_origin() -> None: + """Every thread finds the same circuit for an origin, so no recorded failure is lost to a duplicate.""" + breaker = KeyedCircuitBreaker(failure_threshold=_N_THREADS * _N_OPS + 1) + start = threading.Barrier(_N_THREADS) + + def worker(index: int) -> None: + start.wait() + for op in range(_N_OPS): + request = httpx2.Request("GET", f"https://{_HOSTS[(index + op) % len(_HOSTS)]}/x") + with contextlib.suppress(InternalServerError): + breaker(request, _raise_500) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(_N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + states = breaker._states._states # noqa: SLF001 + assert len(states) == len(_HOSTS) + assert sum(state._consecutive_failures for state in states.values()) == _N_THREADS * _N_OPS # noqa: SLF001 + assert {state._consecutive_failures for state in states.values()} == {_N_THREADS * _N_OPS // len(_HOSTS)} # noqa: SLF001 + + +def _raise_500(request: httpx2.Request) -> httpx2.Response: + raise InternalServerError(httpx2.Response(HTTPStatus.INTERNAL_SERVER_ERROR, request=request)) diff --git a/tests/test_keyed_circuit_breaker_sync.py b/tests/test_keyed_circuit_breaker_sync.py index 305c758..150f747 100644 --- a/tests/test_keyed_circuit_breaker_sync.py +++ b/tests/test_keyed_circuit_breaker_sync.py @@ -206,6 +206,18 @@ def test_circuit_key_is_the_normalized_origin_without_userinfo(caplog: pytest.Lo assert keys == ["https://a.test", "http://a.test:8080"] +def test_circuit_key_of_a_custom_key_is_its_str(caplog: pytest.LogCaptureFixture) -> None: + handler = _PerHost(failing={"a.test"}) + breaker = KeyedCircuitBreaker(failure_threshold=1, key=lambda request: (request.url.host, 1)) + with ( + _client(handler, breaker=breaker) as client, + caplog.at_level(logging.WARNING, logger="httpware.circuit_breaker"), + ): + _fail_n(client, "https://a.test/x", 1) + keys = [r.circuit_key for r in caplog.records if r.name == "httpware.circuit_breaker"] # ty: ignore[unresolved-attribute] + assert keys == ["('a.test', 1)"] + + class _Held: """A `next` that blocks until released, then answers `status`."""