fix: clean up garbage connections on cancellation in AsyncConnectionPool

src/httpcore2/httpcore2/_async/connection.py

@@ -156,6 +156,9 @@ async def aclose(self) -> None:
             async with Trace("close", logger, None, {}):
                 await self._connection.aclose()
 
+    def is_connected(self) -> bool:
+        return self._connection is not None and self._connection.is_connected()
+
     def is_available(self) -> bool:
         if self._connection is None:
             # If HTTP/2 support is enabled, and the resulting connection could

src/httpcore2/httpcore2/_async/connection_pool.py

@@ -263,12 +263,21 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         """
         closing_connections = []
 
+        # Connections currently referenced by an active request (including
+        # connections that are in the process of being established).
+        request_connections = {r.connection for r in self._requests}
+
         # 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 not (connection.is_connected() or connection in request_connections):
+                # Garbage: a NEW-state connection whose request was cancelled
+                # before the TCP handshake completed.  Drop it without closing
+                # (there is no socket to close yet).
+                self._connections.remove(connection)
             elif connection.has_expired():
                 # log: "closing expired connection"
                 self._connections.remove(connection)

src/httpcore2/httpcore2/_async/http11.py

@@ -246,6 +246,9 @@ async def aclose(self) -> None:
     def can_handle_request(self, origin: Origin) -> bool:
         return origin == self._origin
 
+    def is_connected(self) -> bool:
+        return not self.is_closed()
+
     def is_available(self) -> bool:
         # Note that HTTP/1.1 connections in the "NEW" state are not treated as
         # being "available". The control flow which created the connection will

src/httpcore2/httpcore2/_async/http2.py

@@ -477,6 +477,9 @@ async def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int
     def can_handle_request(self, origin: Origin) -> bool:
         return origin == self._origin
 
+    def is_connected(self) -> bool:
+        return not self.is_closed()
+
     def is_available(self) -> bool:
         return (
             self._state != HTTPConnectionState.CLOSED

src/httpcore2/httpcore2/_async/http_proxy.py

@@ -204,6 +204,9 @@ async def aclose(self) -> None:
     def info(self) -> str:
         return self._connection.info()
 
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
     def is_available(self) -> bool:
         return self._connection.is_available()
 
@@ -330,6 +333,9 @@ async def aclose(self) -> None:
     def info(self) -> str:
         return self._connection.info()
 
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
     def is_available(self) -> bool:
         return self._connection.is_available()
 

src/httpcore2/httpcore2/_async/interfaces.py

@@ -95,6 +95,18 @@ def info(self) -> str:
     def can_handle_request(self, origin: Origin) -> bool:
         raise NotImplementedError()  # pragma: nocover
 
+    def is_connected(self) -> bool:
+        """
+        Return `True` if the connection is open (the underlying socket has been
+        established).  A connection in the NEW state (just created but not yet
+        connected) returns `False`.
+
+        Note: for some implementations ``is_connected() != not is_closed()``.
+        The default implementation returns ``not self.is_closed()``, which is
+        correct for connections that are never in the NEW (pre-TCP) state.
+        """
+        return not self.is_closed()  # pragma: nocover
+
     def is_available(self) -> bool:
         """
         Return `True` if the connection is currently able to accept an

src/httpcore2/httpcore2/_async/socks_proxy.py

@@ -283,6 +283,9 @@ async def aclose(self) -> None:
         if self._connection is not None:
             await self._connection.aclose()
 
+    def is_connected(self) -> bool:
+        return self._connection is not None and self._connection.is_connected()
+
     def is_available(self) -> bool:
         if self._connection is None:  # pragma: nocover
             # If HTTP/2 support is enabled, and the resulting connection could

src/httpcore2/httpcore2/_sync/connection.py

@@ -156,6 +156,9 @@ def close(self) -> None:
             with Trace("close", logger, None, {}):
                 self._connection.close()
 
+    def is_connected(self) -> bool:
+        return self._connection is not None and self._connection.is_connected()
+
     def is_available(self) -> bool:
         if self._connection is None:
             # If HTTP/2 support is enabled, and the resulting connection could

src/httpcore2/httpcore2/_sync/connection_pool.py

@@ -263,12 +263,21 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         """
         closing_connections = []
 
+        # Connections currently referenced by an active request (including
+        # connections that are in the process of being established).
+        request_connections = {r.connection for r in self._requests}
+
         # 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 not (connection.is_connected() or connection in request_connections):
+                # Garbage: a NEW-state connection whose request was cancelled
+                # before the TCP handshake completed.  Drop it without closing
+                # (there is no socket to close yet).
+                self._connections.remove(connection)
             elif connection.has_expired():
                 # log: "closing expired connection"
                 self._connections.remove(connection)

src/httpcore2/httpcore2/_sync/http11.py

@@ -246,6 +246,9 @@ def close(self) -> None:
     def can_handle_request(self, origin: Origin) -> bool:
         return origin == self._origin
 
+    def is_connected(self) -> bool:
+        return not self.is_closed()
+
     def is_available(self) -> bool:
         # Note that HTTP/1.1 connections in the "NEW" state are not treated as
         # being "available". The control flow which created the connection will

src/httpcore2/httpcore2/_sync/http2.py

@@ -477,6 +477,9 @@ def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int:
     def can_handle_request(self, origin: Origin) -> bool:
         return origin == self._origin
 
+    def is_connected(self) -> bool:
+        return not self.is_closed()
+
     def is_available(self) -> bool:
         return (
             self._state != HTTPConnectionState.CLOSED

src/httpcore2/httpcore2/_sync/http_proxy.py

@@ -204,6 +204,9 @@ def close(self) -> None:
     def info(self) -> str:
         return self._connection.info()
 
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
     def is_available(self) -> bool:
         return self._connection.is_available()
 
@@ -330,6 +333,9 @@ def close(self) -> None:
     def info(self) -> str:
         return self._connection.info()
 
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
     def is_available(self) -> bool:
         return self._connection.is_available()
 

src/httpcore2/httpcore2/_sync/interfaces.py

@@ -95,6 +95,18 @@ def info(self) -> str:
     def can_handle_request(self, origin: Origin) -> bool:
         raise NotImplementedError()  # pragma: nocover
 
+    def is_connected(self) -> bool:
+        """
+        Return `True` if the connection is open (the underlying socket has been
+        established).  A connection in the NEW state (just created but not yet
+        connected) returns `False`.
+
+        Note: for some implementations ``is_connected() != not is_closed()``.
+        The default implementation returns ``not self.is_closed()``, which is
+        correct for connections that are never in the NEW (pre-TCP) state.
+        """
+        return not self.is_closed()  # pragma: nocover
+
     def is_available(self) -> bool:
         """
         Return `True` if the connection is currently able to accept an

src/httpcore2/httpcore2/_sync/socks_proxy.py

@@ -283,6 +283,9 @@ def close(self) -> None:
         if self._connection is not None:
             self._connection.close()
 
+    def is_connected(self) -> bool:
+        return self._connection is not None and self._connection.is_connected()
+
     def is_available(self) -> bool:
         if self._connection is None:  # pragma: nocover
             # If HTTP/2 support is enabled, and the resulting connection could

tests/httpcore2/_async/test_connection.py

@@ -37,6 +37,7 @@ async def test_http_connection() -> None:
         assert not conn.is_closed()
         assert not conn.is_available()
         assert not conn.has_expired()
+        assert not conn.is_connected()
         assert repr(conn) == "<AsyncHTTPConnection [CONNECTING]>"
 
         async with conn.stream("GET", "https://example.com/") as response:
@@ -50,6 +51,7 @@ async def test_http_connection() -> None:
         assert not conn.is_closed()
         assert conn.is_available()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert repr(conn) == "<AsyncHTTPConnection ['https://example.com:443', HTTP/1.1, IDLE, Request Count: 1]>"
 
 

tests/httpcore2/_async/test_http11.py

@@ -24,6 +24,7 @@ async def test_http11_connection() -> None:
         assert not conn.is_closed()
         assert conn.is_available()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert repr(conn) == "<AsyncHTTP11Connection ['https://example.com:443', IDLE, Request Count: 1]>"
 
 

tests/httpcore2/_async/test_http2.py

@@ -33,6 +33,7 @@ async def test_http2_connection() -> None:
         assert conn.is_available()
         assert not conn.is_closed()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert conn.info() == "'https://example.com:443', HTTP/2, IDLE, Request Count: 1"
         assert repr(conn) == "<AsyncHTTP2Connection ['https://example.com:443', IDLE, Request Count: 1]>"
 

tests/httpcore2/_sync/test_connection.py

@@ -37,6 +37,7 @@ def test_http_connection() -> None:
         assert not conn.is_closed()
         assert not conn.is_available()
         assert not conn.has_expired()
+        assert not conn.is_connected()
         assert repr(conn) == "<HTTPConnection [CONNECTING]>"
 
         with conn.stream("GET", "https://example.com/") as response:
@@ -50,6 +51,7 @@ def test_http_connection() -> None:
         assert not conn.is_closed()
         assert conn.is_available()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert repr(conn) == "<HTTPConnection ['https://example.com:443', HTTP/1.1, IDLE, Request Count: 1]>"
 
 

tests/httpcore2/_sync/test_http11.py

@@ -24,6 +24,7 @@ def test_http11_connection() -> None:
         assert not conn.is_closed()
         assert conn.is_available()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert repr(conn) == "<HTTP11Connection ['https://example.com:443', IDLE, Request Count: 1]>"
 
 

tests/httpcore2/_sync/test_http2.py

@@ -33,6 +33,7 @@ def test_http2_connection() -> None:
         assert conn.is_available()
         assert not conn.is_closed()
         assert not conn.has_expired()
+        assert conn.is_connected()
         assert conn.info() == "'https://example.com:443', HTTP/2, IDLE, Request Count: 1"
         assert repr(conn) == "<HTTP2Connection ['https://example.com:443', IDLE, Request Count: 1]>"
 

tests/httpcore2/test_cancellations.py

@@ -1,4 +1,5 @@
 import typing
+from unittest.mock import patch
 
 import anyio
 import hpack
@@ -129,6 +130,29 @@ async def test_connection_pool_timeout_during_response() -> None:
         assert not pool.connections
 
 
+@pytest.mark.anyio
+async def test_connection_pool_cancellation_during_waiting_for_connection() -> None:
+    """
+    A cancellation while a request is waiting for a connection should leave
+    the pool in a consistent state.
+
+    In this case, that means the new (not-yet-connected) connection is
+    discarded and no longer remains in the pool.
+    """
+
+    async def wait_for_connection(self: typing.Any, *args: typing.Any, **kwargs: typing.Any) -> None:
+        await anyio.sleep(999)
+
+    with patch(
+        "httpcore2._async.connection_pool.AsyncPoolRequest.wait_for_connection",
+        new=wait_for_connection,
+    ):
+        async with httpcore2.AsyncConnectionPool() as pool:
+            with anyio.move_on_after(0.01):
+                await pool.request("GET", "http://example.com")
+            assert not pool.connections
+
+
 @pytest.mark.anyio
 async def test_h11_timeout_during_request() -> None:
     """