Repository navigation
Reclaim connections orphaned by a cancelled request - #1099
wwmanidumaneesha wants to merge 1 commit into
Conversation
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.
|
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 |
|
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 Reproduction against master ( 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: 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. |
|
Independent confirmation from a production incident, plus a comparison with the We hit this through 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 ( 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 Comparison with #986 (
|
| 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.
|
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
So switching to |
…#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>
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:
self._connection is Noneand_connect_failedstillFalse,is_closed(),has_expired()andis_idle()all returnFalse, so no branch of the cleanup loop matches it.is_available()returnsFalsefor HTTP/1.1 while connecting.So it holds a pool slot that is never reclaimed. Once
max_connectionsof them accumulate, the pool is permanently full and every subsequent request fails withPoolTimeoutagainst 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:
Scope
Worth noting for reviewers: this needs cancellation delivered while the task is suspended in
wait_for_connection(), which is whatasyncio.Task.cancel()andasyncio.wait_for()do.Scope-based cancellation (
anyio.move_on_after, trio) is not affected — it is delivered at the next checkpoint, which lands insideAsyncHTTPConnection.handle_async_request(), where the existingexcept BaseExceptionsets_connect_failed = Trueand 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_requestasserts both that no connection is left behind and that the pool still works afterwards.Verified to fail on
masterand pass with the fix, 8 runs each way.