Keep the pool from closing a connection assigned to a queued request

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)

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)

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)

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)

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():
     """

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