Merge remote-tracking branch 'origin/main' into feat/send-with-response

Author: lesnik512

docs/middleware.md

@@ -142,7 +142,6 @@ The same protocol shape, sync flavor. Use these when wiring middleware into a sy
 
 ```python
 from httpware import Middleware, Next, before_request, after_response, on_error
-from httpware.middleware.chain import compose
 ```
 
 A sync `Middleware` is a structural protocol — any callable with the right signature satisfies it:

docs/testing.md

@@ -56,6 +56,9 @@ def test_get_returns_typed_response() -> None:
 For tests that need to vary the response by call count or assert on the requests that came in, use a handler with instance state:
 
 ```python
+from httpware import AsyncRetry
+
+
 class _ResponseSequence:
     """Returns each status in order; records every request received."""
 

src/httpware/middleware/resilience/bulkhead.py

@@ -9,6 +9,13 @@
 
 AsyncBulkhead is the sharable unit — pass the same instance to multiple
 AsyncClient(middleware=[shared]) calls to enforce a joint cap across clients.
+
+AsyncBulkhead is single-event-loop: the underlying asyncio.Semaphore binds
+to whichever loop first awaits it, and cross-loop wake-ups are not thread
+safe. A single instance acquired from a second event loop (e.g. another
+thread running asyncio.run) raises RuntimeError on entry rather than
+deadlocking silently. To cap a sync+async or cross-thread workload, use
+a Bulkhead and an AsyncBulkhead with matching max_concurrent.
 """
 
 import asyncio
@@ -24,6 +31,11 @@
 
 _MAX_CONCURRENT_INVALID = "max_concurrent must be >= 1"
 _ACQUIRE_TIMEOUT_INVALID = "acquire_timeout must be >= 0"
+_ASYNCBULKHEAD_CROSS_LOOP_MSG = (
+    "AsyncBulkhead is bound to a single event loop. First seen on {first!r}; "
+    "current request is on {current!r}. Use one AsyncBulkhead per loop; "
+    "cross-thread sharing requires the sync Bulkhead primitive."
+)
 
 _LOGGER = logging.getLogger("httpware.bulkhead")
 
@@ -42,7 +54,8 @@ class AsyncBulkhead:
         Defaults to ``1.0``. ``None`` waits forever; ``0`` fails fast. Must be
         ``>= 0`` (or ``None``).
 
-    See the module docstring for the algorithm and middleware-ordering guidance.
+    See the module docstring for the algorithm, middleware-ordering guidance,
+    and the single-event-loop constraint.
 
     """
 
@@ -59,9 +72,29 @@ def __init__(
         self._max_concurrent = max_concurrent
         self._acquire_timeout = acquire_timeout
         self._sem = asyncio.Semaphore(max_concurrent)
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._loop_lock = threading.Lock()
+
+    def _check_loop(self) -> None:
+        current = asyncio.get_running_loop()
+        cached = self._loop
+        if cached is current:
+            return
+        if cached is not None:
+            raise RuntimeError(
+                _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=cached, current=current),
+            )
+        with self._loop_lock:
+            if self._loop is None:
+                self._loop = current
+            elif self._loop is not current:
+                raise RuntimeError(
+                    _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=self._loop, current=current),
+                )
 
     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()
         try:
             if self._acquire_timeout is None:
                 await self._sem.acquire()

tests/test_bulkhead.py

@@ -431,3 +431,45 @@ async def slow_handler(request: httpx2.Request) -> httpx2.Response:
     assert record.acquire_timeout == 0.0  # ty: ignore[unresolved-attribute]
     assert record.method == "GET"  # ty: ignore[unresolved-attribute]
     assert "example.test/y" in record.url  # ty: ignore[unresolved-attribute]
+
+
+# ───── Single-event-loop guard ──────────────────────────────────────────────
+
+
+async def test_first_acquire_captures_running_loop() -> None:
+    """AsyncBulkhead binds to whichever loop first acquires a slot."""
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1)
+    assert bulkhead._loop is None  # noqa: SLF001
+    handler = _SlowHandler(delay=0.0)
+    async with _client(handler, bulkhead=bulkhead) as client:
+        await client.get("https://example.test/x")
+    assert bulkhead._loop is asyncio.get_running_loop()  # noqa: SLF001
+
+
+async def test_same_loop_succeeds_across_multiple_acquires() -> None:
+    """Repeated acquires on the same loop never trigger the cross-loop guard."""
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_2)
+    handler = _SlowHandler(delay=0.0)
+    async with _client(handler, bulkhead=bulkhead) as client:
+        for _ in range(5):
+            response = await client.get("https://example.test/x")
+            assert response.status_code == HTTPStatus.OK
+
+
+def test_cross_loop_acquire_raises_runtimeerror() -> None:
+    """A bulkhead first used on one loop, then reused on another, raises RuntimeError.
+
+    Each asyncio.run() call creates a fresh event loop and tears it down on
+    exit. Sharing one AsyncBulkhead instance across two asyncio.run() calls
+    is the cross-loop case the guard prevents.
+    """
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1)
+    handler = _SlowHandler(delay=0.0)
+
+    async def _run_once() -> None:
+        async with _client(handler, bulkhead=bulkhead) as client:
+            await client.get("https://example.test/x")
+
+    asyncio.run(_run_once())  # captures loop L1, then L1 closes
+    with pytest.raises(RuntimeError, match="AsyncBulkhead is bound to a single event loop"):
+        asyncio.run(_run_once())