diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 5ef74e64..48a0140a 100644 --- a/httpcore/_async/connection_pool.py +++ b/httpcore/_async/connection_pool.py @@ -297,6 +297,12 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # log: "closing idle connection" self._connections.remove(connection) closing_connections.append(connection) + elif not connection.is_idle() and not any( + request.connection is connection for request in self._requests + ): + # log: "closing orphaned connection" + 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..928f42d9 100644 --- a/httpcore/_sync/connection_pool.py +++ b/httpcore/_sync/connection_pool.py @@ -297,6 +297,12 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # log: "closing idle connection" self._connections.remove(connection) closing_connections.append(connection) + elif not connection.is_idle() and not any( + request.connection is connection for request in self._requests + ): + # log: "closing orphaned connection" + 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/_async/test_connection_pool.py b/tests/_async/test_connection_pool.py index bc4b251e..f1bb5bde 100644 --- a/tests/_async/test_connection_pool.py +++ b/tests/_async/test_connection_pool.py @@ -1,12 +1,15 @@ import logging import typing +import anyio import hpack import hyperframe.frame import pytest import trio as concurrency import httpcore +from httpcore._async import connection_pool +from httpcore._async.interfaces import AsyncConnectionInterface @pytest.mark.anyio @@ -451,6 +454,75 @@ async def trace(name, kwargs): ] +@pytest.mark.anyio +async def test_connection_pool_removes_connection_after_request_cancellation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + request_assigned = anyio.Event() + + class WaitingPoolRequest(connection_pool.AsyncPoolRequest): + async def wait_for_connection( + self, timeout: float | None = None + ) -> AsyncConnectionInterface: + await super().wait_for_connection(timeout) + request_assigned.set() + await anyio.sleep_forever() + raise AssertionError("Pool request must be cancelled") + + monkeypatch.setattr(connection_pool, "AsyncPoolRequest", WaitingPoolRequest) + + class ConnectingConnection(AsyncConnectionInterface): + def __init__(self) -> None: + self.closed = False + + async def handle_async_request( + self, request: httpcore.Request + ) -> httpcore.Response: + raise AssertionError("Cancelled request must not reach the connection") + + async def aclose(self) -> None: + self.closed = True + + def can_handle_request(self, origin: httpcore.Origin) -> bool: + return True + + def is_available(self) -> bool: + return False + + def has_expired(self) -> bool: + return False + + def is_idle(self) -> bool: + return False + + def is_closed(self) -> bool: + return False + + def info(self) -> str: + return "CONNECTING" + + class TestPool(httpcore.AsyncConnectionPool): + def create_connection( + self, origin: httpcore.Origin + ) -> AsyncConnectionInterface: + return connection + + connection = ConnectingConnection() + async with TestPool(max_connections=1) as pool: + + async def request() -> None: + with pytest.raises(TimeoutError): + with anyio.fail_after(0.01): + await pool.request("GET", "https://example.com/") + + async with anyio.create_task_group() as task_group: + task_group.start_soon(request) + await request_assigned.wait() + + assert pool.connections == [] + assert connection.closed + + @pytest.mark.anyio async def test_connection_pool_with_immediate_expiry(): """