diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd..393eb518 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## Unreleased +- Fix the connection pool closing a connection it had already assigned to a queued request. (#1110) - Fix `max_keepalive_connections` not being properly handled. (#1000) ## Version 1.0.9 (April 24th, 2025) diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 5ef74e64..6d642d8a 100644 --- a/httpcore/_async/connection_pool.py +++ b/httpcore/_async/connection_pool.py @@ -279,19 +279,35 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: """ closing_connections = [] + # A connection assigned to a request still reports as idle until that + # request starts on it. It is spoken for: closing it here would either + # send the request back to the queue, or close the socket underneath + # the request if it started in the meantime. + assigned_connections = { + request.connection + for request in self._requests + if request.connection is not None + } + + def is_idle(connection: AsyncConnectionInterface) -> bool: + return connection.is_idle() and connection not in assigned_connections + # First we handle cleaning up any connections that are closed, # have expired their keep-alive, or surplus idle connections. for connection in list(self._connections): if connection.is_closed(): # log: "removing closed connection" self._connections.remove(connection) + elif connection in assigned_connections: + # log: "keeping assigned connection" + pass elif connection.has_expired(): # log: "closing expired connection" self._connections.remove(connection) closing_connections.append(connection) elif ( connection.is_idle() - and sum(connection.is_idle() for connection in self._connections) + and sum(is_idle(connection) for connection in self._connections) > self._max_keepalive_connections ): # log: "closing idle connection" @@ -308,7 +324,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: if connection.can_handle_request(origin) and connection.is_available() ] idle_connections = [ - connection for connection in self._connections if connection.is_idle() + connection for connection in self._connections if is_idle(connection) ] # There are three cases for how we may be able to handle the request: @@ -321,6 +337,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # log: "reusing existing connection" connection = available_connections[0] pool_request.assign_to_connection(connection) + assigned_connections.add(connection) elif len(self._connections) < self._max_connections: # log: "creating new connection" connection = self.create_connection(origin) diff --git a/httpcore/_sync/connection_pool.py b/httpcore/_sync/connection_pool.py index 4b26f9c6..a6a400fc 100644 --- a/httpcore/_sync/connection_pool.py +++ b/httpcore/_sync/connection_pool.py @@ -279,19 +279,35 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: """ closing_connections = [] + # A connection assigned to a request still reports as idle until that + # request starts on it. It is spoken for: closing it here would either + # send the request back to the queue, or close the socket underneath + # the request if it started in the meantime. + assigned_connections = { + request.connection + for request in self._requests + if request.connection is not None + } + + def is_idle(connection: ConnectionInterface) -> bool: + return connection.is_idle() and connection not in assigned_connections + # First we handle cleaning up any connections that are closed, # have expired their keep-alive, or surplus idle connections. for connection in list(self._connections): if connection.is_closed(): # log: "removing closed connection" self._connections.remove(connection) + elif connection in assigned_connections: + # log: "keeping assigned connection" + pass elif connection.has_expired(): # log: "closing expired connection" self._connections.remove(connection) closing_connections.append(connection) elif ( connection.is_idle() - and sum(connection.is_idle() for connection in self._connections) + and sum(is_idle(connection) for connection in self._connections) > self._max_keepalive_connections ): # log: "closing idle connection" @@ -308,7 +324,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: if connection.can_handle_request(origin) and connection.is_available() ] idle_connections = [ - connection for connection in self._connections if connection.is_idle() + connection for connection in self._connections if is_idle(connection) ] # There are three cases for how we may be able to handle the request: @@ -321,6 +337,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # log: "reusing existing connection" connection = available_connections[0] pool_request.assign_to_connection(connection) + assigned_connections.add(connection) elif len(self._connections) < self._max_connections: # log: "creating new connection" connection = self.create_connection(origin) diff --git a/scripts/unasync.py b/scripts/unasync.py index 5a5627d7..f8ca7590 100644 --- a/scripts/unasync.py +++ b/scripts/unasync.py @@ -24,6 +24,7 @@ ('@pytest.mark.anyio', ''), ('@pytest.mark.trio', ''), ('AutoBackend', 'SyncBackend'), + ('httpcore._async', 'httpcore._sync'), ] COMPILED_SUBS = [ (re.compile(r'(^|\b)' + regex + r'($|\b)'), repl) diff --git a/tests/_async/test_connection_pool.py b/tests/_async/test_connection_pool.py index bc4b251e..424d3cc3 100644 --- a/tests/_async/test_connection_pool.py +++ b/tests/_async/test_connection_pool.py @@ -7,6 +7,7 @@ import trio as concurrency import httpcore +from httpcore._async.connection_pool import AsyncPoolRequest @pytest.mark.anyio @@ -518,6 +519,44 @@ async def test_connection_pool_with_no_keepalive_connections_allowed(): assert info == [] +@pytest.mark.anyio +async def test_connection_pool_keeps_connection_assigned_to_queued_request(): + """ + A connection assigned to a queued request stays IDLE until the request + starts on it. A concurrent pass over the pool must not close it as + surplus in the meantime. + """ + 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(network_backend=network_backend) as pool: + # An initial request leaves an IDLE connection in the pool. + await pool.request("GET", "https://example.com/") + [connection] = pool.connections + + # Queue a request and assign the IDLE connection to it, as the pass + # triggered by another request completing would. + pool_request = AsyncPoolRequest(httpcore.Request("GET", "https://example.com/")) + pool._requests.append(pool_request) + assert pool._assign_requests_to_connections() == [] + assert pool_request.connection is connection + + # Before the request starts on its connection, another pass finds + # more IDLE connections than keep-alive allows. + pool._max_keepalive_connections = 0 + assert pool._assign_requests_to_connections() == [] + assert pool.connections == [connection] + + pool._requests.remove(pool_request) + + @pytest.mark.trio async def test_connection_pool_concurrency(): """ diff --git a/tests/_sync/test_connection_pool.py b/tests/_sync/test_connection_pool.py index 7adc3f5c..32434d1a 100644 --- a/tests/_sync/test_connection_pool.py +++ b/tests/_sync/test_connection_pool.py @@ -7,6 +7,7 @@ from tests import concurrency import httpcore +from httpcore._sync.connection_pool import PoolRequest @@ -519,6 +520,44 @@ def test_connection_pool_with_no_keepalive_connections_allowed(): +def test_connection_pool_keeps_connection_assigned_to_queued_request(): + """ + A connection assigned to a queued request stays IDLE until the request + starts on it. A concurrent pass over the pool must not close it as + surplus in the meantime. + """ + network_backend = httpcore.MockBackend( + [ + 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!", + ] + ) + + with httpcore.ConnectionPool(network_backend=network_backend) as pool: + # An initial request leaves an IDLE connection in the pool. + pool.request("GET", "https://example.com/") + [connection] = pool.connections + + # Queue a request and assign the IDLE connection to it, as the pass + # triggered by another request completing would. + pool_request = PoolRequest(httpcore.Request("GET", "https://example.com/")) + pool._requests.append(pool_request) + assert pool._assign_requests_to_connections() == [] + assert pool_request.connection is connection + + # Before the request starts on its connection, another pass finds + # more IDLE connections than keep-alive allows. + pool._max_keepalive_connections = 0 + assert pool._assign_requests_to_connections() == [] + assert pool.connections == [connection] + + pool._requests.remove(pool_request) + + + def test_connection_pool_concurrency(): """ HTTP/1.1 requests made in concurrency must not ever exceed the maximum number