fix(core): add SSE stream idle watchdog to abort hung streams (Fixes #4177) - #5330
fix(core): add SSE stream idle watchdog to abort hung streams (Fixes #4177)#5330aspnmy wants to merge 3 commits into
Conversation
qqqys
left a comment
qqqys
left a comment
There was a problem hiding this comment.
Critical: the watchdog never observes the stream iterator. processStreamResponse now accepts _streamWatchdog/_cleanupAbortLink, but the generator still uses for-await over streamResponse directly and never calls streamWatchdog.next(...); the only cleanup path added is around synchronous generator creation. If the provider hangs while awaiting the next SSE chunk, that for-await remains stuck forever, the timeout never rejects, and the abort controller is never triggered. This leaves the core session hang from #4177 unfixed after merge. Please route each streamResponse.next() through the watchdog and clean it up when iteration finishes or throws.
|
@qwen-code /triage |
|
Re-run (2026-06-18): PR unchanged since last triage — no new commits pushed. All findings below still stand. Thanks for the PR! Template looks good ✓ On direction: this solves a real user problem — hung SSE streams on weak networks are painful, and the only recovery today is Ctrl+C. Claude Code's CHANGELOG confirms this is production-relevant: they shipped a default 5-minute idle timeout ( On approach: the watchdog is created but never actually wired into the stream. This is a correctness blocker that makes the entire feature non-functional as shipped:
The missing integration looks something like replacing the Additional concerns:
Flagging these for discussion before diving deeper. The direction is right, but the implementation needs the actual wiring before code review is meaningful. 中文说明重新运行（2026-06-18）：PR 自上次 triage 以来无变化——没有新提交。以下所有发现仍然有效。 感谢贡献！ 模板完整 ✓ 方向：解决的是真实用户问题——弱网环境下 SSE 流挂起非常痛苦，目前只能 Ctrl+C 恢复。Claude Code 的 CHANGELOG 证实这是生产环境需要的功能。 方案：看门狗被创建了但从未接入流处理。 这是一个正确性缺陷，导致整个功能无效。 其他顾虑：无测试、循环导入、90 秒默认超时较激进。 先提出来讨论，再深入看代码。方向是对的，但实现需要补上实际接入才有意义。 — Qwen Code · qwen3.7-max |
|
Re-run (2026-06-18): PR unchanged. Re-confirmed all findings below. Added tmux code-analysis evidence. Code ReviewTypecheck: clean (zero errors). The critical integration bug is confirmed after reading the actual code:
The watchdog module itself ( const iter = streamResponse[Symbol.asyncIterator]();
try {
while (true) {
const result = streamWatchdog
? await streamWatchdog.next(iter.next())
: await iter.next();
if (result.done) break;
const chunk = result.value;
// ... existing chunk processing
}
} finally {
streamWatchdog?.cleanup();
cleanupAbortLink?.();
}Other code-level findings:
Real-Scenario TestingCode analysis evidence (tmux)Network-drop testResult: SKIP (justified) The feature is non-functional by code inspection — the watchdog timer never fires because Once the wiring is fixed, a meaningful test would:
中文说明重新运行（2026-06-18）：PR 无变化。重新确认了所有发现。 代码审查类型检查：通过（零错误）。 关键集成缺陷已确认： 看门狗模块本身单独看是合理的——但从未被调用。修复方式是把 其他发现：循环导入、underscore 前缀掩盖未使用参数、try/catch 范围问题。 真实场景测试通过代码分析确认功能无效——看门狗从未接入迭代循环。tmux 输出已在上方内联。修复接入后需要进行网络切断测试。 — Qwen Code · qwen3.7-max |
|
Re-run (2026-06-18): PR unchanged since last triage — single commit ( Stepping back: the direction is right and the problem is real. Weak-network stream hangs are a genuine pain point, and Claude Code's production deployment (with their 5-minute But the PR is missing its own core integration. The watchdog object is created, passed as a parameter, and then ignored. The The try/catch placement is also wrong for the timeout case — cleanup lives in the caller's catch block, but the timeout error would throw during iteration in the caller's caller. The My independent proposal: I'd solve this with ~15 lines directly in The fix is straightforward — maybe 10 lines to replace Verdict: Request changes. The architecture is sound, the integration is incomplete. One focused fix away from being reviewable. 中文说明重新运行（2026-06-18）：PR 自上次 triage 以来无变化。结论不变。 退一步看：方向正确，问题真实存在。看门狗模块本身设计清晰，但 PR 缺少了自己的核心集成。 我的独立方案：直接在 修复方式很直接：约 10 行代码 + 把清理移到 结论：请求修改。 架构合理，集成不完整。一个专注的修复就能进入可审查状态。 — Qwen Code · qwen3.7-max |
qwen-code-ci-bot
left a comment
qwen-code-ci-bot
left a comment
There was a problem hiding this comment.
Needs the core wiring fix — the watchdog module is created but never invoked in processStreamResponse. See my notes above. The fix is ~10 lines: replace for await with an explicit loop calling streamWatchdog.next(iter.next()), and move cleanup into a finally inside processStreamResponse. 🙏
|
@qwen-code /triage |
| }); | ||
|
|
||
| return this.processStreamResponse(model, streamResponse); | ||
| try { |
There was a problem hiding this comment.
[Critical] Dead try/catch around an async generator — cleanup is unreachable
processStreamResponse is declared async *, so calling it returns an AsyncGenerator object synchronously without executing any function body. Errors during stream iteration happen in the caller's for await loop, not here. The catch block (which contains the only calls to streamWatchdog.cleanup() and cleanupAbortLink()) can never fire.
This means cleanup of both the watchdog timers and the abort-signal listener never runs on any exit path — not on normal completion, not on error, not on caller abort.
Fix: Remove this try/catch and move cleanup into a try/finally inside processStreamResponse's generator body (wrapping the for await loop). Generator finally blocks run on .return(), .throw(), and normal completion.
— qwen3.7-max via Qwen Code /review
| prompt_id: string, | ||
| ): Promise<AsyncGenerator<GenerateContentResponse>> { | ||
| const streamAbortController = new AbortController(); | ||
| const cleanupAbortLink = linkAbortSignal(params.config?.abortSignal, streamAbortController); |
There was a problem hiding this comment.
[Critical] Abort event listener leak — cleanupAbortLink is never called
linkAbortSignal adds an 'abort' listener to the caller's AbortSignal. The returned cleanup function is stored as cleanupAbortLink but is only called in the dead catch block at line 2671 (see other comment). On the normal completion path, the listener is never removed.
Each stream call in a long-lived session leaks one listener on the parent signal. With enough turns, this triggers MaxListenersExceededWarning and unbounded listener growth.
Fix: Call _cleanupAbortLink?.() from a finally block inside processStreamResponse's generator body, ensuring it runs on all exit paths.
— qwen3.7-max via Qwen Code /review
| model, | ||
| contents: requestContents, | ||
| config: { ...this.generationConfig, ...params.config }, | ||
| config: { ...this.generationConfig, ...params.config, abortSignal: streamAbortController.signal }, |
There was a problem hiding this comment.
[Critical] Existing test failure — config: {} assertion no longer matches
This line injects abortSignal: streamAbortController.signal into the config passed to generateContentStream. The existing test at geminiChat.test.ts:1476 asserts config: {}, which now fails because the actual config is { abortSignal: AbortSignal{...} }.
CI confirms: Test fails on all 3 platforms (macOS, Windows, Linux).
| config: { ...this.generationConfig, ...params.config, abortSignal: streamAbortController.signal }, | |
| config: { ...this.generationConfig, ...params.config, abortSignal: streamAbortController.signal }, |
Update the test to use an asymmetric matcher:
config: { abortSignal: expect.any(AbortSignal) },— qwen3.7-max via Qwen Code /review
| )); | ||
| }, timeoutMs); | ||
| }); | ||
| return Promise.race([nextPromise, timeoutPromise]).finally(clearTimers); |
There was a problem hiding this comment.
[Critical] Unhandled promise rejection when timeout wins the race
When the watchdog timeout fires before nextPromise settles, Promise.race rejects with the timeout error. However, nextPromise (the underlying stream's .next() call) is still pending. The timeout handler calls abortController.abort(), which causes the stream to eventually reject. Since Promise.race already settled, the nextPromise rejection has no handler attached — triggering an unhandled promise rejection warning (or crash in strict mode).
| return Promise.race([nextPromise, timeoutPromise]).finally(clearTimers); | |
| return Promise.race([nextPromise, timeoutPromise]).finally(clearTimers); | |
| }, |
Should be:
next<T>(nextPromise: Promise<IteratorResult<T>>): Promise<IteratorResult<T>> {
nextPromise.catch(() => {}); // prevent unhandled rejection if timeout wins
const timeoutPromise = new Promise<never>((_, reject) => {
// ...
});
return Promise.race([nextPromise, timeoutPromise]).finally(clearTimers);
},The .catch(() => {}) doesn't prevent Promise.race from seeing the original rejection — it just silences the unhandled rejection when the losing promise later rejects.
— qwen3.7-max via Qwen Code /review
| warningId = setTimeout(() => { | ||
| debugLogger.warn('Stream idle for ' + warningMs + 'ms from ' + model); | ||
| }, warningMs); | ||
| timeoutId = setTimeout(() => { |
There was a problem hiding this comment.
[Suggestion] abortController.abort() called without a reason argument
When the watchdog timeout fires, abort() is called with no reason. Downstream consumers that inspect signal.reason (retry predicates, error classifiers, telemetry) see undefined instead of the InvalidStreamError that is simultaneously thrown via reject().
| timeoutId = setTimeout(() => { | |
| const err = new InvalidStreamError( | |
| 'Stream idle timeout after ' + timeoutMs + 'ms', | |
| 'STREAM_IDLE_TIMEOUT', | |
| ); | |
| abortController.abort(err); | |
| reject(err); |
This ensures signal.reason matches the thrown error and propagates correctly through linkAbortSignal's controller.abort(signal.reason) chain.
— qwen3.7-max via Qwen Code /review
| @@ -0,0 +1,79 @@ | |||
| import { createDebugLogger } from '../utils/debugLogger.js'; | |||
There was a problem hiding this comment.
[Suggestion] No test file for this new module
This module exports 4 functions with non-trivial logic: env var parsing, timer management via Promise.race, abort signal linking (with pre-aborted signal handling), and cleanup. No streamIdleWatchdog.test.ts exists.
Key untested scenarios:
isStreamWatchdogDisabled: env var'1','true', absent, other valuesgetStreamIdleTimeoutMs: disabled path, invalid values ('abc','-1','0')linkAbortSignal: undefined signal, already-aborted signal, cleanup removing listenercreateStreamIdleWatchdog.next(): timeout fires, promise resolves first, cleanup clears timers
— qwen3.7-max via Qwen Code /review
Maintainer verification — local merge build (🔴 not mergeable as-is)Verified the 3-way merge result (current 1. 🔴 The watchdog is plumbed but never invoked — the fix doesn't function
I confirmed the watchdog unit itself is correct (fake-timer test): (This compiles cleanly precisely because the unused params are 2. 🔴 CI Test jobs fail —
|
geminiChat.test.ts |
|
|---|---|
origin/main (no PR) |
✅ 192 / 192 |
| merge result (with PR) | ❌ 191 / 192 |
This matches the PR's CI (Test macOS + ubuntu: fail).
3. ⚠️ Cleanup sits in a catch that can never fire
try {
return this.processStreamResponse(model, streamResponse, streamWatchdog, cleanupAbortLink);
} catch (error) {
streamWatchdog?.cleanup(); cleanupAbortLink?.(); throw error;
}processStreamResponse is an async generator — calling it returns the generator synchronously without running the body, so this catch only catches synchronous construction errors (there are none). Stream/iteration errors happen later, during consumption, and never reach here. Net effect: cleanupAbortLink() (which removes the abort listener added by linkAbortSignal) is never called on the normal path or on iteration errors → a per-stream abort-listener leak. Cleanup needs to live inside the generator (e.g. a finally) or wrap the iteration.
4. ⚠️ Format + tests
prettier --check packages/core/src/core/streamIdleWatchdog.tsreports style issues (the new file isn't formatted).eslintpasses.- No tests are added for the watchdog. A hung-stream test (fake timers, a stalled
next()) would have caught that the watchdog never fires.
What it needs before merge
- Wire
streamWatchdog.next(...)into the stream loop inprocessStreamResponse(replace the barefor awaitwith a manual loop that races eachstreamResponse.next()through the watchdog, or wrapstreamResponsein a watchdog generator). Drop the_once it's actually used. - Move cleanup into a
finallyinside the generator socleanupAbortLink()/cleanup()always run (the currenttry/catcharound the generator construction is dead). - Update
geminiChat.test.tsfor the newabortSignalconfig field (turn the failing assertion green). prettier --writethe new file; add a watchdog unit test (the timeout-fires / timers-cleared cases).
Verdict
The watchdog primitive is well-designed and verified to work in isolation, but as integrated it is inert (never invoked) and the PR is red on CI (a test wasn't updated for the new abortSignal), plus a dead-catch cleanup leak and an unformatted file. 🔴 Changes requested — wire next() into the loop, fix the cleanup placement, and update the test; then re-verify that a stalled stream actually aborts.
Verified by maintainer @wenshao: 3-way merge build (main e234cf6 + PR 0ae2a54) + whole-file usage grep (watchdog next() never called) + a fake-timer unit test of the watchdog (passes in isolation) + geminiChat.test.ts on main (192/192) vs merge (191/192). Env: Darwin arm64, Node v22.22.2.
中文版（点击展开）
维护者验证 —— 本地合并构建（🔴 当前不可合并）
在 macOS（Darwin arm64，Node v22.22.2）上验证了三方合并结果（当前 origin/main e234cf6 + PR head 0ae2a54）。PR 与 main 同步（diff：新增 streamIdleWatchdog.ts + geminiChat.ts 接入，+95/-4）。两处阻塞：CI 的 Test job 红了,而且这个 watchdog 根本没被调用 —— 所以 #4177 的修复并没有生效。
1. 🔴 watchdog 接线了却从未被调用 —— 修复不生效
createStreamIdleWatchdog() 返回 { next, cleanup },其中 next(streamStep) 才是机制本身:它把每个流步骤与空闲超时做竞速,卡住就 abort。但在 geminiChat.ts 里,这个 watchdog 只是被创建（2611）、传给 processStreamResponse（2668）、尝试 cleanup（2670） —— 任何地方都没有 streamWatchdog.next(...) 调用。processStreamResponse 以 _streamWatchdog（下划线=有意未用）接收它,并直接迭代流（for await (const chunk of streamResponse)),完全绕过了 watchdog。
我确认了 watchdog 单元本身是对的（fake-timer 测试）:next() 对卡住的 promise 会在超时后以 STREAM_IDLE_TIMEOUT reject 并 abort;对已 resolve 的步骤会清掉两个定时器。所以这块积木是好的 —— 只是从没接进流循环。按现状,卡死的 SSE 流不会被这个 watchdog 检测或中止。
（它能干净编译,恰恰是因为未用参数加了 _ 前缀,no-unused-vars 不报 —— 这也是它在"什么都不做"的同时还能过 lint/typecheck 的原因。）
2. 🔴 CI Test 失败 —— geminiChat.test.ts 没为新增的 abortSignal 更新
接入后会把 abortSignal: streamAbortController.signal 传进 generateContentStream 的 config。既有断言 should call generateContentStream with the correct parameters 校验的是精确 config,没有更新,于是因为多出的字段而失败。
geminiChat.test.ts |
|
|---|---|
origin/main（无 PR） |
✅ 192 / 192 |
| 合并结果（含 PR） | ❌ 191 / 192 |
这与 PR 的 CI 一致（Test macOS + ubuntu: fail）。
3. ⚠️ cleanup 放在了永远不会触发的 catch 里
try {
return this.processStreamResponse(model, streamResponse, streamWatchdog, cleanupAbortLink);
} catch (error) {
streamWatchdog?.cleanup(); cleanupAbortLink?.(); throw error;
}processStreamResponse 是异步生成器 —— 调用它会同步返回生成器、并不执行函数体,所以这个 catch 只能捕获同步构造期的错误（没有）。流/迭代错误发生在之后的消费阶段,根本到不了这里。净效果:cleanupAbortLink()(用于移除 linkAbortSignal 添加的 abort 监听器)在正常路径和迭代错误路径上都不会被调用 → 每条流泄漏一个 abort 监听器。cleanup 需要放进生成器内部（如 finally)或包住迭代。
4. ⚠️ 格式 + 测试
prettier --check packages/core/src/core/streamIdleWatchdog.ts报告格式问题(新文件未格式化)。eslint通过。- 没有为 watchdog 加测试。一个卡死流的测试(fake timers + 卡住的
next())本可以发现 watchdog 从未触发。
合并前需要
- 把
streamWatchdog.next(...)接进processStreamResponse的流循环(把裸for await换成手动循环、让每次streamResponse.next()经过 watchdog 竞速,或把streamResponse包成 watchdog 生成器)。真正用上后去掉_。 - 把 cleanup 移到生成器内部的
finally,让cleanupAbortLink()/cleanup()总会执行(当前包在生成器构造外的try/catch是死的)。 - 更新
geminiChat.test.ts以适配新增的abortSignalconfig 字段(把失败的断言改绿)。 - 对新文件
prettier --write;补一个 watchdog 单测(超时触发 / 定时器清理两种情形)。
结论
watchdog 这个原语设计不错、单独验证也能工作,但接入后是惰性的(从未被调用),且 PR 在 CI 上是红的(有个测试没为新的 abortSignal 更新),外加一处 dead-catch 的 cleanup 泄漏和一个未格式化的文件。🔴 请求修改 —— 把 next() 接进循环、修正 cleanup 位置、更新测试;然后再验证一次"卡住的流确实会被中止"。
维护者 @wenshao 验证:三方合并构建（main e234cf6 + PR 0ae2a54)+ 全文件用法 grep（watchdog next() 从未被调用）+ watchdog 的 fake-timer 单测（单独能工作）+ geminiChat.test.ts 在 main（192/192）对比合并（191/192）。环境:Darwin arm64,Node v22.22.2。
5 participants
What this PR does
Adds a configurable SSE stream idle watchdog that aborts hung streams in weak-network scenarios. When an SSE connection silently drops between events, the stream would previously hang indefinitely — the user saw a frozen spinner and the only recovery was Ctrl+C → restart.
This PR adds a
setTimeout-based idle watchdog insideprocessStreamResponsethat:QWEN_CODE_STREAM_IDLE_TIMEOUT_MS(default 90s) of silenceAbortController, which triggers the existingInvalidStreamErrorretry budgetWhy it's needed
Fixes #4177. The PR #4176 (merged) fixed the partial-history side of weak-network failures. This PR addresses the remaining case where the SSE stream goes dead-silent, producing no chunk to trigger our existing timeout logic.
Design based on Claude Code's
CLAUDE_ENABLE_STREAM_WATCHDOGsubsystem (Layer 1 of the Streaming Resilience proposal), adapted for qwen-code's codebase conventions.Reviewer Test Plan
How to verify
QWEN_CODE_STREAM_IDLE_TIMEOUT_MS=5000for a short test timeoutInvalidStreamError('STREAM_IDLE_TIMEOUT')Evidence (Before & After)
N/A (non-UI, stream transport layer change).
Tested on
Risk & Scope
AbortController.abort()may surface differently across content generators. Tested with the default Gemini provider path.QWEN_CODE_DISABLE_STREAM_WATCHDOG=1.upstream/mainwith correct UTF-8 encoding.Linked Issues
Fixes #4177
Credit
Design and original implementation: @Alexxigang in PR #4256
Rebased and encoding-fixed: this PR
中文说明
本 PR 做了什么
添加了一个可配置的 SSE 流空闲看门狗，在弱网环境下中止挂起的流。当 SSE 连接在事件之间静默断开时，流会无限挂起——用户看到冻结的 spinner，只能 Ctrl+C 重启。
实现方式：在
processStreamResponse内添加基于setTimeout的空闲看门狗，每收到 chunk 重置计时器，QWEN_CODE_STREAM_IDLE_TIMEOUT_MS（默认 90 秒）无数据后触发中止。为什么需要
修复 #4177。PR #4176（已合并）修复了弱网故障中"部分历史丢失"的问题。本 PR 处理剩余情况：SSE 流完全静默断开。
审查者测试计划
QWEN_CODE_STREAM_IDLE_TIMEOUT_MS=5000缩短测试超时风险与范围
AbortController.abort()在不同内容生成器上表现可能不同QWEN_CODE_DISABLE_STREAM_WATCHDOG=1禁用关联 Issue
Fixes #4177