Requests which raise a PoolTimeout need to be removed from the pool queue. (#502)

* Requests which raise a PoolTimeout need to be removed from the pool queue.

* Update CHANGELOG.md

Author: lovelydinosaur

CHANGELOG.md

@@ -4,9 +4,10 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
-## 0.14.7
+## master
 
-- Fix AttributeError that happened when Socks5Connection were terminated
+- Requests which raise a PoolTimeout need to be removed from the pool queue. (#502)
+- Fix AttributeError that happened when Socks5Connection were terminated. (#501)
 
 ## 0.14.6 (February 1st, 2022)
 

httpcore/_async/connection_pool.py

@@ -223,7 +223,16 @@ async def handle_async_request(self, request: Request) -> Response:
         while True:
             timeouts = request.extensions.get("timeout", {})
             timeout = timeouts.get("pool", None)
-            connection = await status.wait_for_connection(timeout=timeout)
+            try:
+                connection = await status.wait_for_connection(timeout=timeout)
+            except BaseException as exc:
+                # If we timeout here, or if the task is cancelled, then make
+                # sure to remove the request from the queue before bubbling
+                # up the exception.
+                async with self._pool_lock:
+                    self._requests.remove(status)
+                    raise exc
+
             try:
                 response = await connection.handle_async_request(request)
             except ConnectionNotAvailable:

httpcore/_sync/connection_pool.py

@@ -223,7 +223,16 @@ def handle_request(self, request: Request) -> Response:
         while True:
             timeouts = request.extensions.get("timeout", {})
             timeout = timeouts.get("pool", None)
-            connection = status.wait_for_connection(timeout=timeout)
+            try:
+                connection = status.wait_for_connection(timeout=timeout)
+            except BaseException as exc:
+                # If we timeout here, or if the task is cancelled, then make
+                # sure to remove the request from the queue before bubbling
+                # up the exception.
+                with self._pool_lock:
+                    self._requests.remove(status)
+                    raise exc
+
             try:
                 response = connection.handle_request(request)
             except ConnectionNotAvailable:

tests/_async/test_connection_pool.py

@@ -3,7 +3,7 @@
 import pytest
 import trio as concurrency
 
-from httpcore import AsyncConnectionPool, ConnectError, UnsupportedProtocol
+from httpcore import AsyncConnectionPool, ConnectError, PoolTimeout, UnsupportedProtocol
 from httpcore.backends.mock import AsyncMockBackend
 
 
@@ -461,3 +461,30 @@ async def test_connection_pool_closed_while_request_in_flight():
         async with pool.stream("GET", "https://example.com/"):
             with pytest.raises(RuntimeError):
                 await pool.aclose()
+
+
+@pytest.mark.anyio
+async def test_connection_pool_timeout():
+    """
+    Ensure that exceeding max_connections can cause a request to timeout.
+    """
+    network_backend = 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 AsyncConnectionPool(
+        network_backend=network_backend, max_connections=1
+    ) as pool:
+        # Send a request to a pool that is configured to only support a single
+        # connection, and then ensure that a second concurrent request
+        # fails with a timeout.
+        async with pool.stream("GET", "https://example.com/"):
+            with pytest.raises(PoolTimeout):
+                extensions = {"timeout": {"pool": 0.0001}}
+                await pool.request("GET", "https://example.com/", extensions=extensions)

tests/_sync/test_connection_pool.py

@@ -3,7 +3,7 @@
 import pytest
 from tests import concurrency
 
-from httpcore import ConnectionPool, ConnectError, UnsupportedProtocol
+from httpcore import ConnectionPool, ConnectError, PoolTimeout, UnsupportedProtocol
 from httpcore.backends.mock import MockBackend
 
 
@@ -461,3 +461,30 @@ def test_connection_pool_closed_while_request_in_flight():
         with pool.stream("GET", "https://example.com/"):
             with pytest.raises(RuntimeError):
                 pool.close()
+
+
+
+def test_connection_pool_timeout():
+    """
+    Ensure that exceeding max_connections can cause a request to timeout.
+    """
+    network_backend = 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 ConnectionPool(
+        network_backend=network_backend, max_connections=1
+    ) as pool:
+        # Send a request to a pool that is configured to only support a single
+        # connection, and then ensure that a second concurrent request
+        # fails with a timeout.
+        with pool.stream("GET", "https://example.com/"):
+            with pytest.raises(PoolTimeout):
+                extensions = {"timeout": {"pool": 0.0001}}
+                pool.request("GET", "https://example.com/", extensions=extensions)