From 005a844411d994a7fd544ec688b57d8374168224 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 13:53:24 +0300 Subject: [PATCH] test: synchronize bulkhead capacity tests on events instead of sleeps The fail-fast and rejection tests assumed a fixed sleep let the holders acquire their slots and that a handler delay kept them held until the rejections ran. Under CPU load the holders released early and the extra requests went through. The handlers now hold slots until the test releases them. --- tests/test_bulkhead.py | 104 ++++++++++++++++-------------- tests/test_bulkhead_props.py | 43 ++++++++---- tests/test_bulkhead_sync.py | 38 ++++++++--- tests/test_bulkhead_sync_props.py | 36 ++++++----- 4 files changed, 136 insertions(+), 85 deletions(-) diff --git a/tests/test_bulkhead.py b/tests/test_bulkhead.py index b8e8d2c..ac4c13c 100644 --- a/tests/test_bulkhead.py +++ b/tests/test_bulkhead.py @@ -5,7 +5,6 @@ """ import asyncio -import contextlib import logging from collections.abc import Callable, Coroutine from http import HTTPStatus @@ -48,6 +47,29 @@ async def __call__(self, request: httpx2.Request) -> httpx2.Response: self.in_flight -= 1 +class _GatedHandler: + """Holds the first `capacity` requests until `release` is set; later requests return at once.""" + + def __init__(self, capacity: int) -> None: + self.capacity = capacity + self.entered = 0 + self.full = asyncio.Event() + self.release = asyncio.Event() + + async def __call__(self, request: httpx2.Request) -> httpx2.Response: + self.entered += 1 + if self.entered == self.capacity: + self.full.set() + if self.entered <= self.capacity: + await self.release.wait() + return httpx2.Response(HTTPStatus.OK, request=request) + + async def wait_full(self) -> None: + """Wait until `capacity` requests are inside the handler.""" + async with asyncio.timeout(5.0): + await self.full.wait() + + def _client( handler: Callable[[httpx2.Request], httpx2.Response] | Callable[[httpx2.Request], Coroutine[Any, Any, httpx2.Response]], @@ -150,16 +172,12 @@ async def test_max_concurrent_2_observes_at_most_2_in_flight() -> None: async def test_raises_bulkhead_full_error_when_acquire_timeout_exceeded() -> None: """Slot is held by a slow request; a second request with a tiny timeout raises BulkheadFullError.""" - handler = _SlowHandler(delay=1.0) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST) client = _client(handler, bulkhead=bulkhead) - async def _hold_slot() -> None: - await client.get("https://example.test/slow") - - task = asyncio.create_task(_hold_slot()) - # Yield to let the slow request acquire the semaphore. - await asyncio.sleep(0) + holder = asyncio.create_task(client.get("https://example.test/slow")) + await handler.wait_full() with pytest.raises(BulkheadFullError) as exc_info: await client.get("https://example.test/fast") @@ -167,34 +185,26 @@ async def _hold_slot() -> None: assert exc_info.value.max_concurrent == _MAX_CONCURRENT_1 assert exc_info.value.acquire_timeout == _ACQUIRE_TIMEOUT_FAST - # Cancel the lingering slow task to avoid polluting the event loop. - task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await task + handler.release.set() + await holder async def test_bulkhead_full_error_chains_from_timeout() -> None: """BulkheadFullError raised on the async timeout path chains from the TimeoutError.""" - handler = _SlowHandler(delay=1.0) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST) client = _client(handler, bulkhead=bulkhead) - async def _hold_slot() -> None: - await client.get("https://example.test/slow") - - task = asyncio.create_task(_hold_slot()) - # Yield to let the slow request acquire the semaphore. - await asyncio.sleep(0) + holder = asyncio.create_task(client.get("https://example.test/slow")) + await handler.wait_full() with pytest.raises(BulkheadFullError) as exc_info: await client.get("https://example.test/fast") assert isinstance(exc_info.value.__cause__, TimeoutError) - # Cancel the lingering slow task to avoid polluting the event loop. - task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await task + handler.release.set() + await holder async def test_bounded_wait_raises_bulkhead_full_error() -> None: @@ -205,34 +215,36 @@ async def test_bounded_wait_raises_bulkhead_full_error() -> None: BulkheadFullError fields (max_concurrent / acquire_timeout) carry the configured values. """ - handler = _SlowHandler(delay=_ACQUIRE_TIMEOUT_LONG) # holds slot for 100ms + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) client = _client( handler, bulkhead=AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_SHORT), ) first = asyncio.create_task(client.get("https://example.test/a")) - await asyncio.sleep(0.005) # let first acquire the slot + await handler.wait_full() with pytest.raises(BulkheadFullError) as info: await client.get("https://example.test/b") assert info.value.max_concurrent == _MAX_CONCURRENT_1 assert info.value.acquire_timeout == _ACQUIRE_TIMEOUT_SHORT - await first # cleanup + handler.release.set() + await first async def test_acquire_timeout_zero_fails_fast() -> None: """With acquire_timeout=0, the second call raises immediately without waiting.""" - handler = _SlowHandler(delay=_ACQUIRE_TIMEOUT_LONG) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) client = _client( handler, bulkhead=AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0), ) first = asyncio.create_task(client.get("https://example.test/a")) - await asyncio.sleep(0.005) + await handler.wait_full() with pytest.raises(BulkheadFullError) as info: await client.get("https://example.test/b") assert info.value.acquire_timeout == 0 + handler.release.set() await first @@ -276,18 +288,17 @@ def handler(request: httpx2.Request) -> httpx2.Response: async def test_slot_released_on_cancellation() -> None: """If the calling task is cancelled while next() runs, the slot is released.""" - handler = _SlowHandler(delay=0.5) # would block indefinitely + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0) client = _client(handler, bulkhead=bulkhead) first = asyncio.create_task(client.get("https://example.test/a")) - await asyncio.sleep(0.01) # let first acquire and start sleeping in handler + await handler.wait_full() first.cancel() with pytest.raises(asyncio.CancelledError): await first # Slot must now be released — fail-fast=0 next call proves it. - handler.delay = 0.0 # speed up the next request response = await client.get("https://example.test/b") assert response.status_code == HTTPStatus.OK @@ -300,14 +311,14 @@ async def test_cancellation_before_acquire_does_not_hold_slot() -> None: release (it must NOT take the slot the cancelled task was waiting for). And once first releases, the fresh request must complete normally. """ - handler = _SlowHandler(delay=0.05) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=None) client = _client(handler, bulkhead=bulkhead) first = asyncio.create_task(client.get("https://example.test/a")) - await asyncio.sleep(0.005) # first acquires + await handler.wait_full() second = asyncio.create_task(client.get("https://example.test/b")) # waits for slot - await asyncio.sleep(0.005) # ensure second is parked on acquire + await asyncio.sleep(0) # ensure second is parked on acquire second.cancel() with pytest.raises(asyncio.CancelledError): await second @@ -315,10 +326,11 @@ async def test_cancellation_before_acquire_does_not_hold_slot() -> None: # Third request issued while first still holds the slot — must not see a phantom # free slot left by the cancelled second. third = asyncio.create_task(client.get("https://example.test/c")) + handler.release.set() first_response, third_response = await asyncio.gather(first, third) assert first_response.status_code == HTTPStatus.OK assert third_response.status_code == HTTPStatus.OK - assert handler.calls == 2 # noqa: PLR2004 — first and third reached handler; second never did + assert handler.entered == 2 # noqa: PLR2004 — first and third reached handler; second never did # Constructed at module scope on purpose — pins the construct-outside-loop behavior. @@ -416,7 +428,7 @@ async def _sleep(_: float) -> None: # don't actually wait between retries async def test_bulkhead_full_error_is_not_retried_by_retry() -> None: """AsyncRetry does NOT retry BulkheadFullError — it's neither a StatusError nor a NetworkError/TimeoutError.""" - handler = _SlowHandler(delay=0.5) # holds the slot indefinitely + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0) transport = httpx2.MockTransport(handler) @@ -432,30 +444,23 @@ async def test_bulkhead_full_error_is_not_retried_by_retry() -> None: ], ) - # Fill the slot with a long-lived task. first = asyncio.create_task(client.get("https://example.test/holder")) - await asyncio.sleep(0.01) + await handler.wait_full() # Second call hits a full AsyncBulkhead. AsyncRetry must NOT swallow + retry it. with pytest.raises(BulkheadFullError): await client.get("https://example.test/rejected") mock_sleep.assert_not_called() # AsyncRetry never slept — it didn't try to retry - # Cleanup. - first.cancel() - with contextlib.suppress(asyncio.CancelledError): - await first + handler.release.set() + await first async def test_bulkhead_rejected_emits_observability_event(caplog: pytest.LogCaptureFixture) -> None: """When the bulkhead rejects a request via acquire_timeout, emit one WARNING on httpware.bulkhead.""" bulkhead = AsyncBulkhead(max_concurrent=1, acquire_timeout=0.0) - - async def slow_handler(request: httpx2.Request) -> httpx2.Response: - await asyncio.sleep(0.05) - return httpx2.Response(HTTPStatus.OK, request=request) - - transport = httpx2.MockTransport(slow_handler) + handler = _GatedHandler(capacity=1) + transport = httpx2.MockTransport(handler) client = AsyncClient( httpx2_client=httpx2.AsyncClient(transport=transport), middleware=[bulkhead], @@ -464,11 +469,12 @@ async def slow_handler(request: httpx2.Request) -> httpx2.Response: async with client: # First request occupies the only slot. Second should be rejected immediately. first_task = asyncio.create_task(client.get("https://example.test/x")) - await asyncio.sleep(0) # let first_task start and acquire the slot + await handler.wait_full() with caplog.at_level(logging.WARNING, logger="httpware.bulkhead"), pytest.raises(BulkheadFullError): await client.get("https://example.test/y") + handler.release.set() await first_task bulkhead_records = [r for r in caplog.records if r.name == "httpware.bulkhead"] diff --git a/tests/test_bulkhead_props.py b/tests/test_bulkhead_props.py index 28f528f..6524407 100644 --- a/tests/test_bulkhead_props.py +++ b/tests/test_bulkhead_props.py @@ -40,6 +40,29 @@ async def __call__(self, request: httpx2.Request) -> httpx2.Response: self.in_flight -= 1 +class _GatedHandler: + """Holds the first `capacity` requests until `release` is set; later requests return at once.""" + + def __init__(self, capacity: int) -> None: + self.capacity = capacity + self.entered = 0 + self.full = asyncio.Event() + self.release = asyncio.Event() + + async def __call__(self, request: httpx2.Request) -> httpx2.Response: + self.entered += 1 + if self.entered == self.capacity: + self.full.set() + if self.entered <= self.capacity: + await self.release.wait() + return httpx2.Response(HTTPStatus.OK, request=request) + + async def wait_full(self) -> None: + """Wait until `capacity` requests are inside the handler.""" + async with asyncio.timeout(5.0): + await self.full.wait() + + @given( max_concurrent=st.integers(min_value=1, max_value=8), n_requests=st.integers(min_value=1, max_value=32), @@ -71,24 +94,22 @@ async def test_fail_fast_rejects_when_at_capacity( max_concurrent: int, extra_requests: int, ) -> None: - handler = _InFlightHandler(delay=0.05) # hold slots long enough for fail-fast to fire + handler = _GatedHandler(capacity=max_concurrent) transport = httpx2.MockTransport(handler) client = AsyncClient( httpx2_client=httpx2.AsyncClient(transport=transport), middleware=[AsyncBulkhead(max_concurrent=max_concurrent, acquire_timeout=0)], ) - # Fill the bulkhead with max_concurrent long-running tasks. holders = [asyncio.create_task(client.get(f"https://example.test/hold-{i}")) for i in range(max_concurrent)] - await asyncio.sleep(0.005) # let the holders acquire their slots - - # Any extra requests should fail fast with BulkheadFullError. - for i in range(extra_requests): - with pytest.raises(BulkheadFullError): - await client.get(f"https://example.test/extra-{i}") - - # Cleanup the holders. - await asyncio.gather(*holders) + try: + await handler.wait_full() + for i in range(extra_requests): + with pytest.raises(BulkheadFullError): + await client.get(f"https://example.test/extra-{i}") + finally: + handler.release.set() + await asyncio.gather(*holders) @given( diff --git a/tests/test_bulkhead_sync.py b/tests/test_bulkhead_sync.py index 213c7a6..f728ee9 100644 --- a/tests/test_bulkhead_sync.py +++ b/tests/test_bulkhead_sync.py @@ -49,6 +49,27 @@ def __call__(self, request: httpx2.Request) -> httpx2.Response: self.in_flight -= 1 +class _GatedHandler: + """Holds the first `capacity` requests until `release` is set; later requests return at once.""" + + def __init__(self, capacity: int) -> None: + self.capacity = capacity + self.lock = threading.Lock() + self.entered = 0 + self.full = threading.Event() + self.release = threading.Event() + + def __call__(self, request: httpx2.Request) -> httpx2.Response: + with self.lock: + self.entered += 1 + entered = self.entered + if entered == self.capacity: + self.full.set() + if entered <= self.capacity: + self.release.wait() + return httpx2.Response(HTTPStatus.OK, request=request) + + def _client( handler: Callable[[httpx2.Request], httpx2.Response], *, @@ -110,7 +131,7 @@ def test_serializes_at_capacity() -> None: def test_acquire_timeout_rejects_when_no_slot_available() -> None: - handler = _SlowHandler(delay=0.1) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) client = _client( handler, bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST), @@ -118,20 +139,20 @@ def test_acquire_timeout_rejects_when_no_slot_available() -> None: holder = threading.Thread(target=client.get, args=("https://example.test/hold",)) holder.start() - # Give the holder time to acquire the only slot - time.sleep(0.01) try: + assert handler.full.wait(timeout=5.0) with pytest.raises(BulkheadFullError) as info: client.get("https://example.test/blocked") assert info.value.max_concurrent == _MAX_CONCURRENT_1 assert info.value.acquire_timeout == _ACQUIRE_TIMEOUT_FAST finally: + handler.release.set() holder.join() def test_bulkhead_full_error_no_chaining() -> None: """BulkheadFullError raised on the sync timeout path has no __cause__ (no active exception).""" - handler = _SlowHandler(delay=0.1) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) client = _client( handler, bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST), @@ -139,13 +160,13 @@ def test_bulkhead_full_error_no_chaining() -> None: holder = threading.Thread(target=client.get, args=("https://example.test/hold",)) holder.start() - # Give the holder time to acquire the only slot - time.sleep(0.01) try: + assert handler.full.wait(timeout=5.0) with pytest.raises(BulkheadFullError) as exc_info: client.get("https://example.test/blocked") assert exc_info.value.__cause__ is None finally: + handler.release.set() holder.join() @@ -171,19 +192,20 @@ def boom(request: httpx2.Request) -> httpx2.Response: # noqa: ARG001 def test_emits_rejected_event(caplog: pytest.LogCaptureFixture) -> None: - handler = _SlowHandler(delay=0.1) + handler = _GatedHandler(capacity=_MAX_CONCURRENT_1) client = _client( handler, bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST), ) holder = threading.Thread(target=client.get, args=("https://example.test/hold",)) holder.start() - time.sleep(0.01) try: + assert handler.full.wait(timeout=5.0) with caplog.at_level(logging.WARNING, logger="httpware.bulkhead"), pytest.raises(BulkheadFullError): client.get("https://example.test/blocked") assert any("bulkhead rejected" in r.getMessage() for r in caplog.records) finally: + handler.release.set() holder.join() diff --git a/tests/test_bulkhead_sync_props.py b/tests/test_bulkhead_sync_props.py index bc366eb..6c23365 100644 --- a/tests/test_bulkhead_sync_props.py +++ b/tests/test_bulkhead_sync_props.py @@ -74,16 +74,16 @@ def test_in_flight_never_exceeds_max_concurrent( class _BarrierHandler: - """Handler that signals slot acquisition via a threading.Barrier before holding the slot.""" + """Handler that signals slot acquisition via a threading.Barrier, then holds the slot until `release` is set.""" - def __init__(self, barrier: threading.Barrier) -> None: + def __init__(self, barrier: threading.Barrier, release: threading.Event) -> None: self._barrier = barrier + self._release = release def __call__(self, request: httpx2.Request) -> httpx2.Response: # Signal that this holder has acquired a bulkhead slot and is now in-flight. self._barrier.wait(timeout=5.0) - # Hold the slot long enough for the over-limit requests to be rejected. - time.sleep(0.05) + self._release.wait() return httpx2.Response(HTTPStatus.OK, request=request) @@ -100,7 +100,8 @@ def test_fail_fast_rejects_when_at_capacity( # holder has acquired its bulkhead slot (i.e. is inside the handler). # timeout=5.0 sets the default for all barrier.wait() calls. acquired_barrier = threading.Barrier(max_concurrent + 1, timeout=5.0) - handler = _BarrierHandler(acquired_barrier) + release = threading.Event() + handler = _BarrierHandler(acquired_barrier, release) transport = httpx2.MockTransport(handler) client = Client( httpx2_client=httpx2.Client(transport=transport), @@ -110,18 +111,19 @@ def test_fail_fast_rejects_when_at_capacity( # Fill the bulkhead with max_concurrent long-running threads. pool = ThreadPoolExecutor(max_workers=max_concurrent + extra_requests) holders = [pool.submit(client.get, f"https://example.test/hold-{i}") for i in range(max_concurrent)] - # Wait deterministically — barrier releases only once ALL holders are inside the handler. - acquired_barrier.wait(timeout=5.0) - - # Any extra requests should fail fast with BulkheadFullError. - for i in range(extra_requests): - with pytest.raises(BulkheadFullError): - client.get(f"https://example.test/extra-{i}") - - # Cleanup the holders. - for f in holders: - f.result() - pool.shutdown() + try: + # Wait deterministically — barrier releases only once ALL holders are inside the handler. + acquired_barrier.wait(timeout=5.0) + + # Any extra requests should fail fast with BulkheadFullError. + for i in range(extra_requests): + with pytest.raises(BulkheadFullError): + client.get(f"https://example.test/extra-{i}") + finally: + release.set() + for f in holders: + f.result() + pool.shutdown() @given(