Skip to content

Reclaim connections orphaned by a cancelled request - #1099

Open
wwmanidumaneesha wants to merge 1 commit into
encode:masterfrom
wwmanidumaneesha:fix/reclaim-orphaned-pool-connections
Open

wwmanidumaneesha wants to merge 1 commit into
encode:masterfrom
wwmanidumaneesha:fix/reclaim-orphaned-pool-connections

Conversation

@wwmanidumaneesha

Copy link
Copy Markdown

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.

A request that is 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.

Such a connection cannot be reaped, since `is_closed()`, `has_expired()`
and `is_idle()` all report False while it has yet to connect, and it cannot
be reused, since `is_available()` reports False for HTTP/1.1. It therefore
holds a pool slot that is never reclaimed. Once enough have accumulated the
pool is permanently full, and every subsequent request fails with
`PoolTimeout` against a perfectly healthy server, until the process is
restarted.

This restores the invariant the pool already relies on elsewhere: a
connection that is neither available nor idle is reserved for the single
request it was assigned to. Connections no longer referenced by any request,
and which can no longer service one, are now closed and removed.

Refs encode#1093, and the earlier reports in encode#658 and encode#830.
@mbeijen

mbeijen commented Jul 27, 2026

Copy link
Copy Markdown

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.

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

avyukd commented Aug 12, 2026

Copy link
Copy Markdown

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

jpstory commented Oct 15, 2026

Copy link
Copy Markdown

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

jpstory commented Oct 15, 2026

Copy link
Copy Markdown

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.

garrett361 added a commit to PrimeIntellect-ai/prime-rl that referenced this pull request Nov 6, 2026
…#4408)

This PR gives the orchestrator's router-session releases their own HTTP
client and caps how many run at once. Before, a burst of releases could
leave the shared client permanently unable to send any request, so
releases silently stopped for the rest of the run. From then on the
router treated every finished rollout as still running, and placed new
rollouts on inference replicas based on that wrong picture of their
load.

**Background.** With the `sticky_least_loaded` router policy, each
rollout is a session pinned to one inference replica so later turns
reuse its prefix cache. The router places each new session on the
replica holding the fewest sessions. When an episode ends, the
orchestrator calls `/finish_session` so the router stops counting it. If
a release never arrives, the router keeps counting the session until its
time to live (TTL) expires, 2 h by default.

**Bug 1: a burst loses most of its releases.** All releases shared one
httpx client with 4 connections, a 10 s wait per attempt for a free
connection, and errors logged at DEBUG. When many episodes end at once
(for example, a sandbox outage failing thousands of episodes), thousands
of releases queue for those 4 connections:

```
 episodes ending at once        shared httpx pool (4 connections)        router
 release ─┐
 release ─┼─► wait up to 10 s ─► [ conn ][ conn ][ conn ][ conn ] ─► /finish_session
 release ─┘     for a connection
   ...  (thousands)
```

Most time out waiting, which loses those releases. On its own this is
temporary: once the burst drains, releases work again.

**Bug 2: those timeouts permanently break the client.** httpcore 1.0.9
(the latest release) has a race in exactly that timeout. It is reported
upstream in encode/httpcore#1099 (open, unmerged), where a comment
reproduces this pool-timeout case on the same versions; the pydantic
fork fixed it in pydantic/httpx2#983.

```
 time        queued request R                       pool (4 slots)
 ─────────────────────────────────────────────────────────────────────────────────
 0 s         joins the queue, 10 s timer starts     slots 1-4 busy
 ...
 10 s        (still asleep)                         request in slot 1 finishes;
                                                    pool creates connection C
                                                    (CONNECTING) and hands it to R
 10 s        timer fires before R wakes up:
             R raises PoolTimeout and leaves
             (C is never used)                      C stays in slot 1: never opened,
                                                    so never "closed", "idle" or
                                                    "expired", so never cleaned up
```

The pool opens a new connection instead of reusing the one that just
finished because the old client kept at most 1 idle connection, and
httpcore 1.0.9 counts every connection, busy or idle, against that
limit. So with 4 connections open, each one is closed the moment it
frees up.

Thousands of requests time out at nearly the same moment, so this fires
reliably and each occurrence costs one slot for good:

```
 slot 1   busy ──► dead C ──────────────────────────────────────────────►
 slot 2   busy ─────────► dead C ───────────────────────────────────────►
 slot 3   busy ───────────────► dead C ─────────────────────────────────►
 slot 4   busy ─────────────────────► dead C ───────────────────────────►
                                      └─ from here on, every release waits 10 s
                                         for a slot and fails with PoolTimeout
```

A fresh client works immediately, which is how the reproducer confirms
the pool, not the router, is at fault.

**Consequence.** The router keeps treating finished sessions as active
until its TTL removes them. It sends each new session to the replica
with the fewest sessions, so a replica holding many finished ones looks
busier than it is and receives less new work, while a replica holding
few receives more. The session counts look even across replicas, but the
actual work is not. If the table reaches `VLLM_ROUTER_SLL_MAX_SESSIONS`
(100,000 by default), new sessions are no longer pinned to one replica
and lose their prefix cache on every turn (we have not hit this; the
largest table we have seen peaked at about a third of the limit).
Training data and rewards are unaffected; this is purely a
routing-efficiency problem, and it was invisible because nothing was
logged above DEBUG.

**The fix.** A release now needs one of 32 tickets (a semaphore) before
it sends anything, and the release client has exactly 32 connections.
Releases that find no free ticket wait in `clients.py`, not inside
httpx, so httpx's pool wait and its race never come into play:

```
 thousands of releases arrive at once (one per finished session)
        │
        ▼
 ┌─ waiting for a ticket ───────────┐
 │ (semaphore in clients.py)        │──► waited 30 s, still no ticket?
 │ R  R  R  R  R  R  ...            │    give up: never sent, counted as failed,
 └───────────────┬──────────────────┘    router removes the session after its TTL
                 │ takes 1 of 32 tickets
                 ▼
 ┌─ httpx client, 32 connections ───┐
 │ a ticket holder always finds an  │──► POST /finish_session ──► router
 │ idle connection, so it never     │
 │ waits inside httpx               │
 └───────────────┬──────────────────┘
                 │ request done (success or error)
                 ▼
          ticket returned; the next release in line goes
```

- Releases use a dedicated client with one connection per ticket, so
requests never wait inside httpx's pool.
- A `PoolTimeout` can then only mean a leaked connection, so the release
path replaces the client and the next releases go out on a fresh one.
- Failures log a WARNING each time their total doubles, so a burst logs
a handful of lines, and `sessions/released` / `sessions/release_failed`
report cumulative counts.

| Reproducer (httpx 0.28.1, httpcore 1.0.9) | Before | After |
|---|---|---|
| Burst of 6,000 releases, 10 ms server | 21 delivered (0.35%) | all
delivered |
| Single releases after each burst of 4,000 against a 2 s server | 0/3,
every burst | 3/3, every burst |
| Pool state after those bursts | 4 connections stuck `CONNECTING` | 32
`IDLE` |

Trade-offs a reviewer should know:

- If the router answers releases slowly (for example, under heavy load),
a release that waits more than 30 s for a ticket is dropped, so an
episode never waits longer than that to finish.
- Replacing a broken client fails up to 31 releases already in flight on
it; those sessions wait for the router TTL.
- Two other call sites keep the original 4-connection client unchanged:
weight-update admin calls (`AdminPlane`) and the Dynamo frontend's
startup checks. Each sends only a few requests at a time, so none ever
waits for a connection.

## Verification

Commit bbd2d3b adds `tools/repro_session_release_pool_leak.py` and the
next commit removes it, so it stays in this PR's history without landing
on `main`. It copies the old and new release paths, runs them against a
local `/finish_session` server through the dispatcher's cancellation
pattern, and prints delivery, follow-up releases and pool state per
burst. It runs anywhere, without the repo's dependencies:

```
git show bbd2d3b:tools/repro_session_release_pool_leak.py > /tmp/repro.py
uv run --no-project --with httpx==0.28.1 --with httpcore==1.0.9 --with anyio==4.13.0 \
    --with tenacity==9.1.4 --with starlette --with uvicorn python /tmp/repro.py old   # or: new
```

`--server-delay 0.01 --episodes 3000 --bursts 1` gives the first table
row; `new --undersized-pool` gives the pool 4 connections for 32
tickets, so its connections leak and the client replacement path
recovers it.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Development

Successfully merging this pull request may close these issues.

Cancelling requests under pool contention permanently leaks connection slots (AsyncConnectionPool → PoolTimeout)

4 participants