Stop serving requests whose client has already hung up - #22
Conversation
Reviewer's GuideThe PR closes the non-streaming disconnect gap by racing upstream aggregation against ASGI client-disconnect events, cancelling and awaiting abandoned work, returning a quiet 204, and preserving audit and concurrency behavior; it adds focused ASGI regression tests and documentation. Sequence diagram for cancelling non-streaming work after client hangupsequenceDiagram
participant Client
participant Handler
participant await_or_hangup
participant Upstream
participant AuditMiddleware
Client->>Handler: POST non-streaming request
Handler->>await_or_hangup: await_or_hangup(_fetch_checked_chat(...), request)
await_or_hangup->>Upstream: _fetch_checked_chat(...)
par Await upstream aggregation
Upstream-->>await_or_hangup: completion
and Listen for disconnect
Client--xawait_or_hangup: http.disconnect
end
alt Upstream completes first
await_or_hangup-->>Handler: collected result
Handler-->>Client: JSONResponse 200
else Client disconnects first
await_or_hangup->>Upstream: cancel and await cleanup
await_or_hangup-->>Handler: ClientHungUp
Handler->>Handler: _hungup_response(...)
Handler-->>Client: Response 204
AuditMiddleware-->>AuditMiddleware: record outcome=cancelled
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="tests/test_nonstream_disconnect.py" line_range="72" />
<code_context>
+ def handle_upstream(self, request):
+ if HANG_MARKER in request.content.decode("utf-8", "replace"):
+ self.upstream_seen.set()
+ return httpx.Response(200, content=_HangingBody(self.upstream_closed),
+ headers={"content-type": "text/event-stream"})
+ return super().handle_upstream(request)
</code_context>
<issue_to_address>
**issue (testing):** `httpx.Response(..., content=_HangingBody(...))` raises `TypeError` because an `AsyncByteStream` must be supplied through the `stream=` parameter, not `content=`. The new disconnect tests therefore fail while constructing the mocked upstream response instead of exercising cancellation.
**Triggers:** When any test sends a payload containing `synthetic-hang-up`.
**Suggested fix:** Pass the stream as `stream=_HangingBody(self.upstream_closed)` instead of `content=...`.
```suggestion
return httpx.Response(200, stream=_HangingBody(self.upstream_closed),
```
</issue_to_address>Sourcery assessment
Needs a human reviewer. 1 finding to address first, and if the disconnect race is wrong, an in-flight upstream generation could be cancelled prematurely and the audit record could persist an incorrect cancelled outcome; reverting would stop future cancellations but would not repair those records or any upstream work already affected. The impact is bounded and normally repairable, rather than an authorization, deletion, or payment decision.
Blocking findings: tests/test_nonstream_disconnect.py:72
`StreamingResponse` watches for the client leaving: for `spec_version < 2.4` -- both of uvicorn's HTTP protocols declare 2.3 -- Starlette runs `stream_response` and `listen_for_disconnect` in one task group, so a disconnect cancels the pending upstream read. A non-streaming endpoint has no equivalent. It awaits the whole aggregation inside the handler and only reaches `send` once the answer is finished, so nobody consumes `http.disconnect` for that entire window. What that costs, per abandoned request: - the upstream call runs to completion, up to the 300s read timeout; - the credits for it are spent, for an answer nobody will read; - `ConcurrencyLimitMiddleware` holds the slot for the whole duration -- slots are taken from the moment the request enters until the app returns, so enough hung-up clients turn the gateway into a 503 for everyone else; - the audit row says `outcome=success` with HTTP 200, because the aggregation really did complete; only the socket was gone. The endpoints now race the aggregation against a disconnect listener (`app/client_hangup.await_or_hangup`), wrapped outside `_routed_fetch` so the whole credential-failover sequence is one cancellable unit rather than only its last attempt. If the client leaves first, the in-flight call is cancelled and the cancellation is awaited, so httpx closes the response instead of leaving a half-open connection behind; the handler returns an empty 204 (nothing is delivered either way, and Starlette sends no content-length for 204, so the framing stays legal), and `AuditMiddleware` sees the disconnect through the same `receive` and records `cancelled`. If the upstream answers first, the result is returned exactly as before -- including a completion that races the client's last millisecond -- and the listener is cancelled. Callers without a `Request` (`request=None`) keep the plain await. This is not a policy change: it is the missing half of guarantees the gateway already advertises for streaming. There is no new switch, because "stop working for a client that is gone" has no cost side worth opting out of. Tests (`tests/test_nonstream_disconnect.py`, driving the ASGI app directly, since TestClient cannot deliver a mid-request disconnect): - a hangup during aggregation cancels the in-flight upstream read; - with `max_concurrent=1`, a request whose client has hung up stops refusing new ones with 503 `concurrency_limit`, and the next request gets its answer; - the hangup is audited as `cancelled`, never as a completed inference; - control: an aggregation for a client that stayed is unaffected (the listener must not eat a live request), including the trailing disconnect servers synthesise after the response; - control: the same hangup on a streaming request already behaved this way, which is how the gap was narrowed to `stream=false`. Verified against a real uvicorn with a real socket close, not just the harness: the instrumented cancellation lands 0.10s after the client hangs up, the unwatched path is still asleep 1.5s later, a follow-up request returns 200, and uvicorn logs no "Exception in ASGI application".
adf058c to
3740302
Compare
There was a problem hiding this comment.
已核验当前 head 3740302：未发现阻塞性的运行逻辑缺陷，取消范围、三协议行为、资源释放和审计上下文均符合预期。Python 3.12 下 PR 定向测试 43 项通过，另做的三协议/连接关闭及相关回归检查 15 项通过；未重跑全量套件或真实 TCP 实验。
请按两条行内意见补齐仓库内的回归覆盖，并收束本次新增的长注释。这两项是测试与可维护性要求，不是已复现的实现故障。Sourcery 关于 content=AsyncByteStream 的意见已对照 httpx 0.28.1 源码及测试确认是误报，无需据此改动。
| await asyncio.wait_for(task, STEP) # 修复前：没人监听断连，这一句必然超时 | ||
|
|
||
| asyncio.run(scenario()) | ||
| self.assertTrue(self.upstream_closed.is_set(), "挂断之后，挂着的上游读要被取消") |
There was a problem hiding this comment.
请把新增断连用例参数化覆盖 /v1/chat/completions、/v1/responses、/v1/messages，并补上换凭证后的聚合断连与外层任务取消场景。
这里的 upstream_closed 在迭代器 finally 中置位，只证明读取被取消，不能单独证明响应的异步 aclose() 已完成。请再用独立关闭标记断言清理完成，并验证取消审计及后续请求取得并发名额。上述边界在额外验证中已通过，希望固化为本 PR 的回归测试；不需要改变当前生产逻辑。
There was a problem hiding this comment.
Done in 9771473 — tests and comments only, no production change.
Parameterized over all three protocols. Every case in the file now runs as a subTest over fixtures.GENERATIONS (/v1/chat/completions, /v1/responses, /v1/messages), reusing the existing endpoint/profile fixtures instead of a private list. Two things that turned out to matter for that layout: reset_upstream() has to be called per sub-case (all sub-cases share one setUp, so attempt counters otherwise bleed across protocols), and each case builds its own ConcurrencyLimitMiddleware(max_concurrent=1) because the gate attached to converter.app is reused across cases and leaks state.
The two scenarios requested.
test_hangup_after_credential_failover_cancels_the_whole_sequence: the first shot is poisoned with 429 so the replay lands on a second credential and that one hangs. Asserts exactly two upstream attempts on two distinctx-user-idcredentials, nofailover_recoveredtag (nothing was recovered),cancelledin the audit, the slot released, and a normal 200 for the client that is still there.test_outer_task_cancellation_cancels_and_closes_the_upstream_call:task.cancel()on the ASGI task itself, same assertions. To be fair to the record, this one already passed onaf5b306— it pins a contract that holds rather than reproducing a gap.- Both controls kept and parameterized: a client that stayed connected must not be disturbed, and the streaming path already behaved this way.
Independent close marker — and a correction to my reply on the Sourcery thread. You are right that the finally in __aiter__ only proves the read was cancelled. Splitting the markers (read_cancelled from the iterator's finally, stream_closed from aclose()) also showed the fixture was the weak link: built with content=, httpx takes the AsyncIterable branch in encode_content and re-wraps the body in AsyncIteratorByteStream, whose aclose() does not forward to the wrapped object. Measured — Response.aclose() completed normally while our hook stayed unset, so a stream_closed assertion written against that fixture can never fire regardless of what production does. The fixture response is therefore built with stream= plus an explicit content-type, which is what makes the two markers genuinely independent. Sourcery's suggested parameter was right even though its stated reason (TypeError at construction) was not — which refines what I said there.
For completeness: a real-TCP probe (httpcore/h11, not MockTransport) showed the server observing the connection close ~2ms after the cancel, so closure semantics were never missing on the production path; the mock just could not see them.
Red/green. On af5b306 the three hang-up cases fail as 9 subtests (3 cases x 3 protocols); the outer-cancel case and both controls pass on base. On this head: tests/test_nonstream_disconnect.py 44 passed / 129 subtests, full suite 883 passed / 2901 subtests (-p no:randomly).
| """非流式（聚合）端点的下游断连监听。 | ||
|
|
||
| `StreamingResponse` 自带这层保护：Starlette 在 `spec_version < 2.4`（uvicorn 的两个 HTTP 协议 | ||
| 都是 2.3）把 `stream_response` 和 `listen_for_disconnect` 放进同一个任务组，客户端中途挂断， | ||
| 挂着的上游读就一起被取消。非流式端点没有对应机制 —— 端点直接 `await` 完整聚合，拿到响应对象 | ||
| 之后才第一次 `send`，于是整个聚合窗口里没有任何人消费 `http.disconnect`。 | ||
|
|
||
| 后果不是「回不去」而已（响应本来就发不出去了）：这枪上游照旧跑到完，最长是读超时那样长；额度 | ||
| 照旧烧；`ConcurrencyLimitMiddleware` 的名额照旧被这个已经没有客户端的请求占着，名额用满之后 | ||
| 网关对所有人回 503；审计照旧写 `outcome=success`，因为它确实完整收到了一次上游响应。 | ||
|
|
||
| 本模块把「等一次上游调用」和「等一次断连」放在同一场竞速里：谁先到算谁。断连先到就取消那次 | ||
| 调用，并把取消等干净 —— httpx 的 `async with` 收尾要在 `finally` 里跑完，连接才不会半开着。 | ||
| """ |
There was a problem hiding this comment.
请收束本 PR 新增的长注释：模块和函数 docstring 原则上保留一句职责说明，仅对非显然的竞态、取消收尾语义保留一两行就近注释。
这里的故障背景、计费影响、Starlette 行为和修复过程已在 PR 描述中说明，不必在源码重复展开。同样压缩 converter.py::_hungup_response 和 tests/test_nonstream_disconnect.py 的长说明，去掉叙事化措辞；保留必要契约，不改变行为。
There was a problem hiding this comment.
Done in 9771473 — collapsed the new prose, no behavioural change.
app/client_hangup.py: module docstring,ClientHungUpand_listen_for_hangupare one line each.await_or_hangupkeeps its one-line contract plus the two lines that are not readable off the code: why the cancellation is awaited before it propagates (httpx closes the response in theasync withfinally, so not waiting leaves the connection half-closed), and whatrequest=Nonedegrades to.converter.py::_hungup_response: down to the 204 being the formality of "an ASGI call must deliver a response" (Starlette writes no content-length for it) and the audit verdict coming fromAuditMiddleware, in two lines.tests/test_nonstream_disconnect.py: module docstring now says only what the file pins (Starlette covers streaming, the aggregation window has no equivalent) plus the two extra shapes it exercises; per-case intent moved into the test names; the failure narrative is gone from the source.- Kept deliberately: the one-line reason beside the fixture's
stream=choice, since that trap is invisible to a reader who hits it.
… comments Follow-up on review. No production behaviour changes. Tests (tests/test_nonstream_disconnect.py), each parameterized over fixtures.GENERATIONS (/v1/chat/completions, /v1/responses, /v1/messages): - hangup during aggregation cancels the in-flight read and closes the response; - the hangup is audited as `cancelled` (never as a completed inference) and the `max_concurrent=1` slot is released to the next request; - new: a hangup after a credential failover cancels the whole replay sequence -- first attempt poisoned with 429, replay on a second credential hangs; asserts exactly two attempts on two distinct credentials, no `failover_recovered` tag (nothing was recovered), `cancelled` in the audit, slot released; - new: the caller cancelling the ASGI task itself reaches the upstream read and the response close the same way (this one already held before the fix, so it pins the contract rather than reproducing a gap); - both controls kept and parameterized: a client that stayed connected must not be disturbed, and the same hangup on a streaming request already behaved so. The close marker is now independent of the read marker, which also required building the fixture response with `stream=` rather than `content=`: `content=` makes httpx re-wrap the stream in `AsyncIteratorByteStream`, whose `aclose()` never reaches ours, so a `stream_closed` assertion there would have been vacuous (measured: `Response.aclose()` completed while the hook stayed unset). `content-type` is passed explicitly, so nothing else changes. Comments: module and function docstrings are down to the contract they carry -- why the cancellation is awaited, what `request=None` degrades to, and why the hangup response is an empty 204. The failure narrative stays in the PR body. Red on `af5b306` (9 subtests: 3 hangup cases x 3 protocols), green here. Full suite: 883 passed / 2901 subtests.
|
Both inline points are addressed in 9771473 (
One correction worth stating out loud, since it revises a claim I made earlier on this PR: making the close marker observable required building the fixture response with Verification on the new head: targeted file 44 passed / 129 subtests, full suite 883 passed / 2901 subtests; the three hang-up cases are red on |
2 participants
Problem
A non-streaming request whose client has already hung up keeps being served. Nothing consumes
http.disconnectwhile the response is aggregated, so the gateway spends the upstream call, the credits behind it and a concurrency slot on an answer nobody will read — and then records it as a success.Measured on af5b306,
/v1/chat/completionswithstream=false: the client closes the socket mid-request and the upstream answers 0.5 s later.outcome=success,status_code=200outcome=cancelled,status_code=204max_concurrent=1503 concurrency_limit200The worst case is not 0.5 s but the 300 s read timeout, and enough abandoned requests turn the gateway into a 503 for everyone.
Streaming is not affected: for
spec_version < 2.4— both of uvicorn's HTTP protocols declare 2.3 —StreamingResponse.__call__runsstream_responseandlisten_for_disconnectin one task group, so a disconnect cancels the pending upstream read. The gap is specific tostream=false, on all three endpoints that aggregate (/v1/chat/completions,/v1/responses,/v1/messages).Root cause
The non-streaming endpoints
awaitthe entire aggregation inside the handler and only reachsendonce the answer is finished. The disconnect listener lives inStreamingResponse, and a plainJSONResponsehas none, so for the whole aggregation window there is no consumer ofhttp.disconnect. Nothing else closes the gap either: uvicorn'sconnection_lostsetscycle.disconnected, wakesmessage_eventand makes latersendcalls silent — it never cancels the ASGI task.What changes
app/client_hangup.py.await_or_hangup(awaitable, request)races one upstream operation against areceiveloop. It is wrapped outside_routed_fetch, so the whole credential-failover sequence is one cancellable unit rather than only its last attempt. If the client leaves first, the in-flight call is cancelled and the cancellation is awaited, so httpx closes the response instead of leaving a half-open connection behind; the losing listener is always cancelled on the way out.Requestdown (the shared_nonstream_adaptedgrows arequest=keyword).request=Nonekeeps a plain await for any caller that has no request.204: the socket is gone so nothing is delivered either way, Starlette omitscontent-lengthfor 204 so the framing stays legal, and no exception escapes to uvicorn, so the log stays clean.AuditMiddlewaresees the same disconnect throughreceiveand writescancelled.docs/advanced.md,docs/advanced.zh-CN.md).No new switch. "Stop working for a client that is gone" has no cost side worth opting out of, and streaming already does it unconditionally.
Tests
tests/test_nonstream_disconnect.pydrives the ASGI app directly, becauseTestClientonly synthesiseshttp.disconnectafter the response is complete and cannot express a mid-request hangup.max_concurrent=1: a second request is refused with503 concurrency_limitwhile the dead one holds the slot, and the request after the hangup gets its200;cancelled, never as a completed inference;stream=truealready released the upstream and the slot, which is how the gap was narrowed tostream=false. That case asserts an invariant rather than a status code, because Report streaming failures with the real status, and fail over locally #18 also made a pre-first-byte hangup send no headers at all.The first three are red on af5b306 — awaiting that request task simply never returns — and green here.
Verification
python -B -m unittest tests.test_nonstream_disconnect: 43 passed (5 new, plus the inherited routing fixtures).200, and uvicorn logs noException in ASGI application.Notes
Rebased onto af5b306 (#18), which is why the guard wraps
_routed_fetchrather than_fetch_checked_chat. Independent of #18's behaviour otherwise: that PR is about an upstream that fails before the first byte, this is about a client that leaves mid-request.Summary by Sourcery
Cancel non-streaming inference as soon as a client disconnects so resources and concurrency capacity are released promptly and abandoned requests are audited accurately.
New Features:
Bug Fixes:
Documentation:
Tests: