diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd..25689d1f 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 5ef74e64..55b50be3 100644 --- a/httpcore/_async/connection_pool.py +++ b/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()] diff --git a/httpcore/_sync/connection_pool.py b/httpcore/_sync/connection_pool.py index 4b26f9c6..0e29405d 100644 --- a/httpcore/_sync/connection_pool.py +++ b/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()] diff --git a/tests/test_cancellations.py b/tests/test_cancellations.py index 033acef6..1b4e8416 100644 --- a/tests/test_cancellations.py +++ b/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())