httpcore: Request errors on overloaded systems may poison the connection pool

import asyncio
import traceback
from collections import Counter
from socket import socket

import httpcore

extensions = {
    "timeout": {
        "connect": .2,
        "read": .2,
        "pool": .2,
        "write": .2,
    },
    # "trace": log,
}


def report(pool):
    print("-- Report:", f"--- {pool.connections=}", f"--- {pool._requests=}", sep="\n")
    if pool.connections:
        conn = pool.connections[0]
        print(f"--- {conn=}")
        print(f"--- Connection has expired? - {conn.has_expired()}")
        print(f"--- Inner connection is {conn._connection}")


async def main():
    _socket = socket()
    _socket.bind(("localhost", 0))
    _socket.listen()

    async with httpcore.AsyncConnectionPool(
            max_connections=1,
            max_keepalive_connections=0,
    ) as pool:
        # Control
        await pool.request("GET", "https://en.wikipedia.org/wiki/2023")
        print("- Control is OK")
        report(pool)

        # Saturate pool
        num_tasks = 600
        url = "http://{}:{}".format(*_socket.getsockname())
        results = await asyncio.gather(
            *[
                asyncio.create_task(pool.request(
                    "GET",
                    url,
                    extensions=extensions,
                ))
                for _ in range(num_tasks)
            ],
            return_exceptions=True,
        )
        print(f"- Finished saturation\n-- ({num_tasks=}): {dict(Counter(type(res) for res in results))}")
        report(pool)

        # Control
        try:
            await pool.request("GET", "https://en.wikipedia.org/wiki/2023", extensions=extensions)  # PoolTimeout
            # await pool.request("GET", url, extensions=extensions)  Expected ReadTimeout, got PoolTimeout
        except httpcore.PoolTimeout:
            print("- ERROR: Got pool timeout")
            traceback.print_exc(chain=True)
        else:
            print("- No pool timeout!")

First one (lets name it t1) succeeded and passes through status.wait_for_connection in AsyncConnectionPool.handle_async_request while other tasks hang here.

Then t1 can create a real connection in connection.handle_async_request(request) but fails with ReadTimeout.

This leads to call self.response_closed(status) and real connection is closed and connection wrapper is removed from the pool, request status is removed from separate register either. After this new request status creates another connection wrapper with empty connection. Looks like under normal conditions acquiring of new connection wrapper triggers status.wait_for_connection in some task and then new real connection can be established so on and so forth. But under some load asyncio event loop is not fast enough to change coroutines and happens PoolTimeout. So no one can establish real connection (because all tasks has been finished with PoolTimeout and removed) and there is connection wrapper inside pool without real connection.

Now two things can happen.

First one is pool receives new request to some different url. It tries to handle it but hangs until PoolTimeout on status.wait_for_connection because pool can’t create new connection wrapper and old one neither can process request nor can be removed (connection wrapper has no real connection so it is not idle or expired, it is “CONNECTING”).

Second is pool receives new request to same url as it is in connection wrapper inside pool (“CONNECTING” one). In this case request can’t be processed because connection wrapper can handle request but wrapper is not available at the same time (failed is False and connection is None and httpx2 property is False either)

import trio
import traceback
from collections import Counter
from socket import socket

import httpcore

extensions = {
    "timeout": {
        "connect": .2,
        "read": .2,
        "pool": .2,
        "write": .2,
    },
    # "trace": log,
}


def report(pool):
    print("-- Report:", f"--- {pool.connections=}", f"--- {pool._requests=}", sep="\n")
    if pool.connections:
        conn = pool.connections[0]
        print(f"--- {conn=}")
        print(f"--- Connection has expired? - {conn.has_expired()}")
        print(f"--- Inner connection is {conn._connection}")


async def make_request(pool, url, extensions):
    try:
        r = await pool.request(
            "GET",
            url,
            extensions=extensions,
        )
    except httpcore.TimeoutException as exc:
        print(type(exc))
    else:
        print(r)


async def main():
    _socket = socket()
    _socket.bind(("localhost", 0))
    _socket.listen()

    async with httpcore.AsyncConnectionPool(
            max_connections=1,
            max_keepalive_connections=0,
    ) as pool:
        # Control
        await pool.request("GET", "https://en.wikipedia.org/wiki/2023")
        print("- Control is OK")
        report(pool)

        # Saturate pool
        num_tasks = 300
        url = "http://{}:{}".format(*_socket.getsockname())
        async with trio.open_nursery() as nursery:
            for _ in range(num_tasks):
                nursery.start_soon(make_request, pool, url, extensions)

        print(f"- Finished saturation\n")
        report(pool)

        # Control
        try:
            await pool.request("GET", "https://en.wikipedia.org/wiki/2023", extensions=extensions)  # PoolTimeout
        except httpcore.PoolTimeout:
            print("- ERROR: Got pool timeout")
            traceback.print_exc(chain=True)
        else:
            print("- No pool timeout!")


trio.run(main)

class HangingStream(MockStream):
    def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
        if self._closed:
            raise ReadError("Connection closed")
        time.sleep(timeout or 0.1)
        raise ReadTimeout


class MockBackend(NetworkBackend):
    def __init__(
            self,
            buffer: typing.List[bytes],
            http2: bool = False,
            resp_stream_cls: Optional[Type[NetworkStream]] = None,
    ) -> None:
        self._buffer = buffer
        self._http2 = http2
        self._resp_stream_cls: Type[MockStream] = resp_stream_cls or MockStream

    def connect_tcp(
        self,
        host: str,
        port: int,
        timeout: Optional[float] = None,
        local_address: Optional[str] = None,
    ) -> NetworkStream:
        return self._resp_stream_cls(list(self._buffer), http2=self._http2)

    def connect_unix_socket(
        self, path: str, timeout: Optional[float] = None
    ) -> NetworkStream:
        return self._resp_stream_cls(list(self._buffer), http2=self._http2)

    def sleep(self, seconds: float) -> None:
        pass


class AsyncHangingStream(AsyncMockStream):
    async def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
        if self._closed:
            raise ReadError("Connection closed")
        await anyio.sleep(timeout or 0.1)
        raise ReadTimeout


class AsyncMockBackend(AsyncNetworkBackend):
    def __init__(
            self,
            buffer: typing.List[bytes],
            http2: bool = False,
            resp_stream_cls: Optional[Type[AsyncNetworkStream]] = None,
    ) -> None:
        self._buffer = buffer
        self._http2 = http2
        self._resp_stream_cls: Type[AsyncMockStream] = resp_stream_cls or AsyncMockStream

    async def connect_tcp(
        self,
        host: str,
        port: int,
        timeout: Optional[float] = None,
        local_address: Optional[str] = None,
    ) -> AsyncNetworkStream:
        return self._resp_stream_cls(list(self._buffer), http2=self._http2)

    async def connect_unix_socket(
        self, path: str, timeout: Optional[float] = None
    ) -> AsyncNetworkStream:
        return self._resp_stream_cls(list(self._buffer), http2=self._http2)

    async def sleep(self, seconds: float) -> None:
        pass

@pytest.mark.trio
async def test_pool_under_load():
    """
    Pool must remain operational after some peak load.
    """
    network_backend = AsyncMockBackend([], resp_stream_cls=AsyncHangingStream)

    async def fetch(_pool: AsyncConnectionPool, *exceptions: Type[BaseException]):
        with contextlib.suppress(*exceptions):
            async with pool.stream(
                    "GET",
                    "http://a.com/",
                    extensions={
                        "timeout": {
                            "connect": 0.1,
                            "read": 0.1,
                            "pool": 0.1,
                            "write": 0.1,
                        },
                    },
            ) as response:
                await response.aread()

    async with AsyncConnectionPool(
        max_connections=1, network_backend=network_backend
    ) as pool:
        async with concurrency.open_nursery() as nursery:
            for _ in range(300):
                # Sending many requests to the same url. All of them but one will have PoolTimeout. One will
                # be finished with ReadTimeout
                nursery.start_soon(fetch, pool, PoolTimeout, ReadTimeout)
        if pool.connections:  # There is one connection in pool in "CONNECTING" state
            assert pool.connections[0].is_connecting()
        with pytest.raises(ReadTimeout):  # ReadTimeout indicates that connection could be retrieved
            await fetch(pool)

def test_pool_under_load():
    """
    Pool must remain operational after some peak load.
    """
    network_backend = MockBackend([], resp_stream_cls=HangingStream)

    def fetch(_pool: ConnectionPool, *exceptions: Type[BaseException]):
        with contextlib.suppress(*exceptions):
            with pool.stream(
                    "GET",
                    "http://a.com/",
                    extensions={
                        "timeout": {
                            "connect": 0.1,
                            "read": 0.1,
                            "pool": 0.1,
                            "write": 0.1,
                        },
                    },
            ) as response:
                response.read()

    with ConnectionPool(
        max_connections=1, network_backend=network_backend
    ) as pool:
        with concurrency.open_nursery() as nursery:
            for _ in range(300):
                # Sending many requests to the same url. All of them but one will have PoolTimeout. One will
                # be finished with ReadTimeout
                nursery.start_soon(fetch, pool, PoolTimeout, ReadTimeout)
        if pool.connections:  # There is one connection in pool in "CONNECTING" state
            assert pool.connections[0].is_connecting()
        with pytest.raises(ReadTimeout):  # ReadTimeout indicates that connection could be retrieved
            fetch(pool)