Reclaim connections orphaned by a cancelled request

[wwmanidumaneesha]

Reclaim connections orphaned by a cancelled request

Closes #1093. Also covers the earlier reports in #658 and #830, which were closed but describe the same failure.

The problem

A request cancelled after the pool has assigned it a newly created connection, but before that connection has been established, leaves the connection behind in the pool.

That connection is then stuck:

• It cannot be reaped. With self._connection is None and _connect_failed still False, is_closed(), has_expired() and is_idle() all return False, so no branch of the cleanup loop matches it.
• It cannot be reused. is_available() returns False for HTTP/1.1 while connecting.

So it holds a pool slot that is never reclaimed. Once max_connections of them accumulate, the pool is permanently full and every subsequent request fails with PoolTimeout against a perfectly healthy server. Only recreating the pool recovers.

The pool's own repr shows the inconsistency clearly — no requests, yet a connection still occupying the only slot:

<AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 1 active, 0 idle]>
pool.connections == [<AsyncHTTPConnection [CONNECTING]>]

Scope

Worth noting for reviewers: this needs cancellation delivered while the task is suspended in wait_for_connection(), which is what asyncio.Task.cancel() and asyncio.wait_for() do.

Scope-based cancellation (anyio.move_on_after, trio) is not affected — it is delivered at the next checkpoint, which lands inside AsyncHTTPConnection.handle_async_request(), where the existing except BaseException sets _connect_failed = True and the connection is correctly reaped. I probed that path across several pool sizes and timings and saw no leak.

The fix

Rather than patching the single cancellation path, this restores an invariant the pool already relies on. As the comment on AsyncHTTP11Connection.is_available() puts it, a connection in this state "will not be acquired from the connection pool for any other request" — it belongs to exactly one request.

So: a connection that is neither available nor idle is reserved for the one request it was assigned to. If no request references it any more, nothing can ever drive it forward, and it is closed and removed. This is self-healing regardless of how the connection came to be orphaned.

Tests

test_connection_pool_does_not_leak_slot_on_cancelled_request asserts both that no connection is left behind and that the pool still works afterwards.

Verified to fail on master and pass with the fix, 8 runs each way.

[mbeijen]

Hi, unfortunately, apparently httpx does no longer get new releases. Please note how these older PRs fix the same issue and have been open for a long time.

• Fix ConnectionPool state after async cancellation #986
• Fix 2 bugs and improve performance from O(N²+NxM) to O(N+M) in _assign_requests_to_connections #1035

Pydantic decided to fork this package to httpx2, because of accruing issues like this, and I fixed this very issue there --> pydantic/httpx2#983

[avyukd]

We hit this in production and reproduced it independently. Your rule holds for a second shape of the same defect, which the current test does not cover.

If the cancellation arrives after the socket is established but before AsyncHTTP11Connection marks the connection ACTIVE, the orphan is an HTTP/1.1 NEW entry holding an open socket, rather than a CONNECTING entry with no socket. The delivery point is the _state_lock acquire at _async/http11.py:72, which is outside that method's try at line 80 and outside AsyncHTTPConnection.handle_async_request's try at _async/connection.py:75. Neither _response_closed() nor _connect_failed = True runs, so the entry reports closed=False expired=False idle=False available=False exactly like the CONNECTING one. It holds the pool slot permanently, and it also holds the file descriptor.

Reproduction against master (10a6582):

import asyncio, anyio, httpcore

RESP = [b"HTTP/1.1 200 OK\r\n", b"Content-Length: 2\r\n", b"\r\n", b"{}"]

class ParkedStream(httpcore.AsyncNetworkStream):
    async def write(self, buffer, timeout=None): await anyio.sleep(999)
    async def aclose(self): pass

class OneParkedBackend(httpcore.AsyncMockBackend):
    def __init__(self, buf): super().__init__(list(buf)); self.connects = 0
    async def connect_tcp(self, *a, **k):
        self.connects += 1
        return ParkedStream() if self.connects == 1 else await super().connect_tcp(*a, **k)

async def settle(pred, steps=200):
    for _ in range(steps):
        if pred(): return True
        await asyncio.sleep(0)
    return pred()

async def main():
    backend = OneParkedBackend(RESP)
    async with httpcore.AsyncConnectionPool(max_connections=1, network_backend=backend) as pool:
        first = asyncio.ensure_future(pool.request("GET", "http://example.com/"))
        await settle(lambda: len(pool._connections) == 1)
        second = asyncio.ensure_future(pool.request("GET", "http://example.com/"))
        await settle(lambda: any(r.is_queued() for r in pool._requests))
        first.cancel(); second.cancel()
        await asyncio.gather(first, second, return_exceptions=True)

        print(repr(pool))
        for c in pool.connections:
            print(f"  {c!r}")
            print(f"    socket={'OPEN' if c._connection else 'none'} "
                  f"connect_failed={c._connect_failed} closed={c.is_closed()} "
                  f"expired={c.has_expired()} idle={c.is_idle()} available={c.is_available()}")
        try:
            await pool.request("GET", "http://example.com/", extensions={"timeout": {"pool": 0.2}})
            print("  follow-up: OK")
        except Exception as e:
            print(f"  follow-up: {type(e).__name__}")
        print(f"  sockets opened: {backend.connects}")

asyncio.run(main())

On master:

<AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 1 active, 0 idle]>
  <AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, NEW, Request Count: 0]>
    socket=OPEN connect_failed=False closed=False expired=False idle=False available=False
  follow-up: PoolTimeout
  sockets opened: 1

With this PR applied, the pool is empty and the follow-up succeeds. So the patch already handles it and needs no change. A second test case for this shape may be worth adding, since it reaches the abandoned state at a different point in the connection lifecycle than the one the current test drives.

[jpstory]

Independent confirmation from a production incident, plus a comparison with the is_connected() approach from #986.

We hit this through httpx.AsyncClient (httpx 0.28.1, httpcore 1.0.9, asyncio; the reproductions below were run on Python 3.14). Our service cancels in-flight requests with Task.cancel(). The pool filled with <AsyncHTTPConnection [CONNECTING]> entries and every request after that failed with PoolTimeout against healthy upstreams. Only a restart recovered it.

Real-socket reproduction (httpcore only, stdlib asyncio)

One "stalled" origin accepts TCP but never answers the TLS ClientHello, one "healthy" origin answers immediately. Each round sends 100 requests to each, cancels half of the tasks, and waits for the rest.

import asyncio
import random

import httpcore

MAX_CONNECTIONS = 10
STALLED_PORT, HEALTHY_PORT = 18766, 18767
TIMEOUT = {"connect": 2.0, "read": 2.0, "write": 2.0, "pool": 2.0}


async def stalled(reader, writer):
    await asyncio.sleep(60)
    writer.close()


async def healthy(reader, writer):
    while True:
        line = await reader.readline()
        if not line:
            break
        while (await reader.readline()) not in (b"\r\n", b""):
            pass
        writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
        await writer.drain()
    writer.close()


async def one(pool, url):
    try:
        await pool.request("GET", url, extensions={"timeout": TIMEOUT})
    except Exception as exc:
        return type(exc).__name__
    return "ok"


async def main():
    random.seed(0)
    await asyncio.start_server(stalled, "127.0.0.1", STALLED_PORT)
    await asyncio.start_server(healthy, "127.0.0.1", HEALTHY_PORT)
    stalled_url = f"https://127.0.0.1:{STALLED_PORT}/"
    healthy_url = f"http://127.0.0.1:{HEALTHY_PORT}/"

    async with httpcore.AsyncConnectionPool(max_connections=MAX_CONNECTIONS) as pool:
        for rnd in range(12):
            tasks = [asyncio.create_task(one(pool, stalled_url)) for _ in range(100)]
            tasks += [asyncio.create_task(one(pool, healthy_url)) for _ in range(100)]
            await asyncio.sleep(1.0)
            for t in random.sample(tasks, len(tasks) // 2):
                t.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            stuck = [c for c in pool.connections if "CONNECTING" in repr(c)]
            print(f"round {rnd:2d}: {pool!r} stuck CONNECTING={len(stuck)}")


asyncio.run(main())

On master (10a6582) the stuck count only ever grows, with no requests left in the pool:

round  0: <AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 1 active, 0 idle]> stuck CONNECTING=1
round  2: <AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 4 active, 0 idle]> stuck CONNECTING=4
round  6: <AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 8 active, 0 idle]> stuck CONNECTING=8
round 11: <AsyncConnectionPool [Requests: 0 active, 0 queued | Connections: 8 active, 0 idle]> stuck CONNECTING=8

The same script with zero cancellations (only connect timeouts and pool timeouts) leaves nothing behind, so it is cancellation specific. With this PR applied, every round ends with Connections: 0 active, 0 idle.

Comparison with #986 (is_connected() approach)

I ported the #986 source change onto current master (it applies cleanly; the original branch no longer imports on Python 3.14 because of an unrelated old typing.Union issue) and ran three reproductions:

Reproduction	master	#986 ported to master	#1099
real sockets, script above	leaks (8 of 10 slots)	no leak	no leak
queued request cancelled after being assigned a new connection (CONNECTING, no socket)	PoolTimeout	OK	OK
@avyukd's case above (HTTP/1.1 NEW with an open socket)	PoolTimeout	PoolTimeout	OK

#986 misses the second shape because AsyncHTTP11Connection.is_connected() is not is_closed(), which is True in the NEW state, so the orphan still counts as connected. The rule in this PR ("neither available nor idle, and no request references it") covers both, and it also does not depend on how the connection got orphaned. So +1 for this approach over #986.

A second test for the NEW-with-socket shape, as suggested above, would lock that in. Here is one in the style of tests/test_cancellations.py that fails on master and passes with this PR:

class ParkedWriteStream(httpcore.AsyncNetworkStream):
    async def write(self, buffer, timeout=None):
        await anyio.sleep(999)  # pragma: nocover

    async def aclose(self):
        pass


class FirstConnectionParkedBackend(httpcore.AsyncMockBackend):
    def __init__(self, buffer):
        super().__init__(buffer)
        self.connects = 0

    async def connect_tcp(self, *args, **kwargs):
        self.connects += 1
        if self.connects == 1:
            return ParkedWriteStream()
        return await super().connect_tcp(*args, **kwargs)


def test_cancelled_request_does_not_leak_new_http11_connection():
    async def scenario():
        network_backend = FirstConnectionParkedBackend(
            [b"HTTP/1.1 200 OK\r\n", b"Content-Length: 2\r\n", b"\r\n", b"ok"]
        )
        async with httpcore.AsyncConnectionPool(
            max_connections=1, network_backend=network_backend
        ) as pool:
            first = asyncio.ensure_future(pool.request("GET", "http://a.example/"))
            for _ in range(200):
                if pool.connections:
                    break
                await asyncio.sleep(0)
            second = asyncio.ensure_future(pool.request("GET", "http://a.example/"))
            for _ in range(200):
                if any(request.is_queued() for request in pool._requests):
                    break
                await asyncio.sleep(0)

            first.cancel()
            second.cancel()
            await asyncio.gather(first, second, return_exceptions=True)

            assert pool.connections == []
            response = await pool.request(
                "GET", "http://a.example/", extensions={"timeout": {"pool": 1.0}}
            )
            assert response.status == 200

    asyncio.run(scenario())

If a smaller blast radius is preferred, the same two cases are also fixed by doing the check only on the failing request's exit path in handle_async_request() (release pool_request.connection when it is not available, not idle, not closed, and no other queued/active request references it), leaving _assign_requests_to_connections() untouched. That version passes the full test suite (216 passed, coverage 100%) and the three reproductions above. I am happy to share the diff, but I would rather see this PR merged than open a competing one.

It also happens without any explicit cancellation

After working around the Task.cancel() path on our side (shielding requests from caller cancellation), the pool wedged again under a burst. No task was cancelled this time: the pool timeout alone is enough. When a queued request is assigned a new connection at the moment its own pool timeout fires, the except path in handle_async_request() drops the request and the freshly created CONNECTING connection stays in the pool. It is easy to hit when the origin closes connections (Connection: close), because every release then creates a new connection for a waiter.

import asyncio
import httpx

async def close_each(reader, writer):
    await reader.readline()
    while (await reader.readline()) not in (b"\r\n", b""):
        pass
    await asyncio.sleep(0.02)
    writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
    await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(close_each, "127.0.0.1", 0)
    url = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/"
    client = httpx.AsyncClient(timeout=httpx.Timeout(5.0, pool=1.0), limits=httpx.Limits(max_connections=20))
    pool = client._transport._pool

    async def one():
        try:
            await client.get(url)
        except httpx.PoolTimeout:
            pass

    for rnd in range(10):
        await asyncio.gather(*(one() for _ in range(600)))
        await asyncio.sleep(0.5)
        orphans = [c for c in pool.connections if not (c.is_idle() or c.is_closed())]
        print(rnd, "orphans:", len(orphans))
        if len(orphans) >= 20:
            break
    try:
        await client.get(url, timeout=httpx.Timeout(5.0, pool=2.0))
        print("pool ok")
    except httpx.PoolTimeout:
        print("pool wedged")

asyncio.run(main())

httpcore	orphans after the run	single request afterwards
master (10a6582)	20 (all CONNECTING), usually within the first round	PoolTimeout
this PR	0 over 10 rounds	ok

So this PR also fixes the cancellation-free trigger, which I think makes it more pressing: any service that sees PoolTimeout under load can end up with a permanently wedged pool.

[jpstory]

Follow-up for anyone landing here with the same symptom: the fork already has this fixed.

pydantic/httpx2#983 (merged 2026-07-14) cleans up these orphaned connections. I ran the reproductions from my comment above against httpcore2 2.13.1, the current release:

Reproduction	httpcore master (10a6582)	httpcore2 2.13.1
Task.cancel() while connecting (real sockets)	leaks	no leak
pool timeout only, no cancellation (Connection: close origin)	20 of 20 slots orphaned within 2 rounds, then PoolTimeout	0 orphans over 10 rounds, follow-up request ok

So switching to httpx2 / httpcore2 resolves both triggers. If staying on httpx for now, the workaround we use is: run each request in its own task behind asyncio.shield so caller cancellation never reaches httpcore, and cap concurrent requests with a semaphore at max_connections so nothing waits inside the pool. Both are needed; either one alone still leaked in our tests.