Reclaim connections orphaned by a cancelled request

CHANGELOG.md

@@ -7,6 +7,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ## Unreleased
 
 - Fix `max_keepalive_connections` not being properly handled. (#1000)
+- Reclaim connections orphaned by a cancelled request, which previously held a pool slot permanently and eventually caused every request to fail with `PoolTimeout`. (#1093)
 
 ## Version 1.0.9 (April 24th, 2025)
 

httpcore/_async/connection_pool.py

@@ -279,8 +279,18 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         """
         closing_connections = []
 
-        # First we handle cleaning up any connections that are closed,
-        # have expired their keep-alive, or surplus idle connections.
+        # A connection that is neither "available" nor "idle" is reserved for
+        # the one request it was assigned to, and only that request is able to
+        # drive it forward. We record which connections are still spoken for,
+        # so that any which have been abandoned can be cleaned up below.
+        #
+        # Identity is used here rather than equality, since connection classes
+        # are not required to be hashable or comparable.
+        reserved_connections = {id(request.connection) for request in self._requests}
+
+        # First we handle cleaning up any connections that are closed, have
+        # expired their keep-alive, are surplus idle connections, or have been
+        # orphaned by the request they were created for.
         for connection in list(self._connections):
             if connection.is_closed():
                 # log: "removing closed connection"
@@ -297,6 +307,20 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
                 # log: "closing idle connection"
                 self._connections.remove(connection)
                 closing_connections.append(connection)
+            elif (
+                id(connection) not in reserved_connections
+                and not connection.is_available()
+                and not connection.is_idle()
+            ):
+                # log: "closing orphaned connection"
+                #
+                # The request this connection was reserved for was cancelled or
+                # failed before the connection was established. Nothing will
+                # ever complete the connection, and it cannot be handed out to
+                # another request, so without this it would occupy a pool slot
+                # indefinitely and eventually starve the pool.
+                self._connections.remove(connection)
+                closing_connections.append(connection)
 
         # Assign queued requests to connections.
         queued_requests = [request for request in self._requests if request.is_queued()]

httpcore/_sync/connection_pool.py

@@ -279,8 +279,18 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         """
         closing_connections = []
 
-        # First we handle cleaning up any connections that are closed,
-        # have expired their keep-alive, or surplus idle connections.
+        # A connection that is neither "available" nor "idle" is reserved for
+        # the one request it was assigned to, and only that request is able to
+        # drive it forward. We record which connections are still spoken for,
+        # so that any which have been abandoned can be cleaned up below.
+        #
+        # Identity is used here rather than equality, since connection classes
+        # are not required to be hashable or comparable.
+        reserved_connections = {id(request.connection) for request in self._requests}
+
+        # First we handle cleaning up any connections that are closed, have
+        # expired their keep-alive, are surplus idle connections, or have been
+        # orphaned by the request they were created for.
         for connection in list(self._connections):
             if connection.is_closed():
                 # log: "removing closed connection"
@@ -297,6 +307,20 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
                 # log: "closing idle connection"
                 self._connections.remove(connection)
                 closing_connections.append(connection)
+            elif (
+                id(connection) not in reserved_connections
+                and not connection.is_available()
+                and not connection.is_idle()
+            ):
+                # log: "closing orphaned connection"
+                #
+                # The request this connection was reserved for was cancelled or
+                # failed before the connection was established. Nothing will
+                # ever complete the connection, and it cannot be handed out to
+                # another request, so without this it would occupy a pool slot
+                # indefinitely and eventually starve the pool.
+                self._connections.remove(connection)
+                closing_connections.append(connection)
 
         # Assign queued requests to connections.
         queued_requests = [request for request in self._requests if request.is_queued()]

tests/test_cancellations.py

@@ -1,3 +1,4 @@
+import asyncio
 import typing
 
 import anyio
@@ -243,3 +244,86 @@ async def test_h2_timeout_during_response():
 
         assert not conn.is_closed()
         assert conn.is_idle()
+
+
+def test_connection_pool_does_not_leak_slot_on_cancelled_request():
+    """
+    A request which is cancelled after the pool has assigned it a newly
+    created connection, but before that connection has been established,
+    must not leave the connection behind holding a pool slot.
+
+    Such a connection can never be reaped, because it is neither closed,
+    expired, nor idle, and can never be reused, because it is not available.
+    Every occurrence therefore consumes a slot permanently, until the pool
+    fails every subsequent request with `PoolTimeout`, even though the server
+    is perfectly healthy.
+
+    This drives `asyncio` directly rather than going through `anyio`, because
+    it relies on a task being cancelled while it is suspended waiting for a
+    connection to be assigned to it. Scope based cancellation, as used by
+    `anyio.move_on_after` and by trio, is only delivered at the next
+    checkpoint, which falls inside the connection's own error handling where
+    it is already dealt with correctly.
+
+    See https://github.com/encode/httpcore/issues/1093
+    """
+
+    async def scenario() -> None:
+        network_backend = httpcore.AsyncMockBackend(
+            [
+                b"HTTP/1.1 200 OK\r\n",
+                b"Content-Type: plain/text\r\n",
+                b"Content-Length: 13\r\n",
+                b"\r\n",
+                b"Hello, world!",
+            ]
+        )
+
+        async with httpcore.AsyncConnectionPool(
+            max_connections=1, network_backend=network_backend
+        ) as pool:
+            # Occupy the pool's only slot, leaving the response body unread so
+            # that the connection remains checked out.
+            response = await pool.handle_async_request(
+                httpcore.Request(
+                    "GET", "http://a.example/", headers={"Host": "a.example"}
+                )
+            )
+
+            # A request to a different origin queues up behind it.
+            task = asyncio.ensure_future(
+                pool.handle_async_request(
+                    httpcore.Request(
+                        "GET", "http://b.example/", headers={"Host": "b.example"}
+                    )
+                )
+            )
+            for _ in range(10):
+                await asyncio.sleep(0)
+            assert not task.done()
+
+            # Closing the first response frees the slot. The pool evicts the
+            # now idle connection and assigns a freshly created one to the
+            # queued request. Cancelling that request before it is scheduled
+            # to resume leaves nothing to establish the new connection.
+            await response.aclose()
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+            # The abandoned connection must not be left occupying the pool.
+            assert pool.connections == []
+
+            # ...and so the pool must still be usable.
+            response = await pool.handle_async_request(
+                httpcore.Request(
+                    "GET",
+                    "http://c.example/",
+                    headers={"Host": "c.example"},
+                    extensions={"timeout": {"pool": 1.0}},
+                )
+            )
+            await response.aread()
+            assert response.status == 200
+
+    asyncio.run(scenario())