httpcore.AsyncConnectionPool poisoning due to asyncio.CancelledError

[nikitagashkov]

Also ported to pydantic/httpx2#982

Hello,

I’d like to report an issue we are encountering with the httpx/httpcore stack.

In a real-world scenario, our application (O) concurrently calls two downstream services (T1 and T2). We observed the following sequence:

1. T1 began returning 5xx errors while T2 remained healthy.
2. After some time, T1 recovered, but all requests to T2 started failing with httpx.PoolTimeout.
3. Restarting the O pods immediately resolved the httpx.PoolTimeout errors.

A simplified version of the real code looks like this:

@router.get("/")
async def view(
    t1_client: Annotated[httpx.AsyncClient, Depends(get_t1_client)],
    t2_client: Annotated[httpx.AsyncClient, Depends(get_t2_client)],
) -> dict[str, str]:
    async with asyncio.TaskGroup() as tg:
        t1_task = tg.create_task(http_get(t1_client, "/data"))
        t2_task = tg.create_task(http_get(t2_client, "/more-data"))
    return t1_task.result() | t2_task.result()


async def http_get(client: httpx.AsyncClient, path: str) -> dict[str, str]:
    resp = await client.get(path)
    resp.raise_for_status()
    return resp.json()

Our hypothesis is that t2_task starts execution but gets cancelled due to an exception in t1_task. As a result, some connections may not be cleaned up properly, eventually exhausting the connection pool.

I was able to reproduce what appears to be the same underlying issue with the following minimal example:

import asyncio
import httpcore


async def main() -> None:
    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
        tasks = [asyncio.create_task(pool.request("GET", f"http://localhost:8888/{i}")) for i in range(2)]
        await asyncio.sleep(0)
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

        print(pool)  # <AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 1 active, 0 idle]>
        await pool.request("GET", "http://localhost:8888/", extensions={"timeout": {"pool": 1}})  # httpcore.PoolTimeout


if __name__ == "__main__":
    asyncio.run(main())

At this point, the pool reports one active connection even though there are no active or queued requests, and subsequent requests fail with httpcore.PoolTimeout. This suggests that task cancellation during pool.request() can leave the pool in an inconsistent state where connections are not properly released.

Thank you for taking a look! Let me know if additional details or diagnostics would be helpful.

[nikoswsn]

I had a similar issue where the pool would stay congested even though it shouldn't and the only thing that could resolve the issue was restarting the service.
Based on my investigation connections get stuck in CONNECTING state indefinitely, occupying pool slots and eventually preventing all new requests from completing.

Root Cause (Hypothesis)

1. Pool is saturated: All max_connections slots are occupied by in-flight requests
2. New request B enters: No slot is free, so B is queued in self._requests
3. B's pool timeout fires: The timeout expires, raising PoolTimeout in B's context
4. Simultaneously, request A completes: A's cleanup handler:
   • Acquires the pool lock
   • Removes A's connection
   • Creates a new connection for queued request B
   • Calls B.pool_request.assign_to_connection(new_connection) [ref]
   • Appends the new connection to self._connections
5. B's PoolTimeout handler runs: B removes itself from self._requests and exits

Result: The connection created in step 4 is now orphaned since it exists in self._connections but no request in self._requests holds a reference to it.

Why Orphans Are Permanent

For an AsyncHTTPConnection with _connection is None and _connect_failed is False (never used, never failed):

Method	Returns	Cleanup Path
is_available()	False	Not assigned to new requests
is_idle()	False	Not cleaned as surplus idle
is_closed()	False	Not removed as closed
has_expired()	False	Not removed as expired
info()	"CONNECTING"	Visible in debug output only

The orphan is invisible to every cleanup path in _assign_requests_to_connections. It permanently occupies a pool slot.

Minimal Reproduction

Requirements: httpcore[asyncio]
Tested withPython@3.13.2 and httpcore@1.0.9

code

import asyncio
import sys
import time
import httpcore

MAX_CONNS = 5
POOL_TIMEOUT = 0.05
SERVER_DELAY = 0.5


async def start_server():
    async def handle(reader, writer):
        await reader.readuntil(b"\r\n\r\n")
        await asyncio.sleep(SERVER_DELAY)
        writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK")
        await writer.drain()
        writer.close()
    return await asyncio.start_server(handle, "127.0.0.1", 9876)


async def main():
    server = await start_server()
    pool = httpcore.AsyncConnectionPool(max_connections=MAX_CONNS)
    stop, timeouts, successes = asyncio.Event(), 0, 0

    async def worker():
        nonlocal timeouts, successes
        while not stop.is_set():
            try:
                await pool.request("GET", "http://127.0.0.1:9876/",
                    extensions={"timeout": {"pool": POOL_TIMEOUT, "read": SERVER_DELAY + 1}})
                successes += 1
            except httpcore.PoolTimeout:
                timeouts += 1
            except Exception:
                pass
            await asyncio.sleep(0)

    workers = [asyncio.create_task(worker()) for _ in range(20)]
    print(f"httpcore {httpcore.__version__}, python {sys.version.split()[0]}\n")

    t0, prev_ok = time.monotonic(), 0
    for _ in range(30):
        await asyncio.sleep(1)
        zombies = sum(1 for c in pool._connections if c.info() == "CONNECTING")
        rps, prev_ok = successes - prev_ok, successes
        print(f"  t={time.monotonic()-t0:4.1f}s  zombies={zombies}/{MAX_CONNS}  "
              f"throughput={rps}/s  pool_timeouts={timeouts}")
        if zombies >= MAX_CONNS:
            break

    stop.set()
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

    zombies = sum(1 for c in pool._connections if c.info() == "CONNECTING")
    if zombies == 0:
        print("\nNo zombies — re-run (timing-dependent race).")
    else:
        print(f"\n{zombies} zombie(s). Waiting 5s with zero traffic...")
        for i in range(5):
            await asyncio.sleep(1)
            print(f"  +{i+1}s  zombies={sum(1 for c in pool._connections if c.info() == 'CONNECTING')}/{MAX_CONNS}")
        print("\nPool connections:")
        for c in pool._connections:
            print(f"  {c.info()}")

    await pool.aclose()
    server.close()
    await server.wait_closed()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nInterrupted")

Monkey Patch Workaround

code

from httpcore._async.connection_pool import AsyncConnectionPool

def apply_httpcore_pool_fix() -> None:
  logger.info('Applying httpcore monkeypatch for zombie connection cleanup')
  
  _original = AsyncConnectionPool._assign_requests_to_connections
  
  def _fixed_assign(self: AsyncConnectionPool) -> Any:
      for connection in list(self._connections):
          if connection.info() == 'CONNECTING' and not any(
              r.connection is connection for r in self._requests
          ):
              self._connections.remove(connection)
  
      return _original(self)
  
  AsyncConnectionPool._assign_requests_to_connections = _fixed_assign  # type: ignore[method-assign]