From fc76a42bd123a4a64d9d830d95c2cf09aa84c6dd Mon Sep 17 00:00:00 2001 From: wwmanidumaneesha Date: Sat, 25 Jul 2026 17:58:05 +0530 Subject: [PATCH] Reclaim connections orphaned by a cancelled request A request that is cancelled after the pool has assigned it a newly created connection, but before that connection has been established, leaves the connection behind in the pool. Such a connection cannot be reaped, since `is_closed()`, `has_expired()` and `is_idle()` all report False while it has yet to connect, and it cannot be reused, since `is_available()` reports False for HTTP/1.1. It therefore holds a pool slot that is never reclaimed. Once enough have accumulated the pool is permanently full, and every subsequent request fails with `PoolTimeout` against a perfectly healthy server, until the process is restarted. This restores the invariant the pool already relies on elsewhere: a connection that is neither available nor idle is reserved for the single request it was assigned to. Connections no longer referenced by any request, and which can no longer service one, are now closed and removed. Refs #1093, and the earlier reports in #658 and #830. --- CHANGELOG.md | 1 + httpcore/_async/connection_pool.py | 28 +++++++++- httpcore/_sync/connection_pool.py | 28 +++++++++- tests/test_cancellations.py | 84 ++++++++++++++++++++++++++++++ 4 files changed, 137 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd0..25689d1f0 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 5ef74e649..55b50be39 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 4b26f9c63..0e29405de 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 033acef60..1b4e84166 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())