From 94e8c5812154046d441762721df115568922c8bc Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Thu, 21 May 2026 18:55:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(core,cli):=20address=20PR=20#4366=20review?= =?UTF-8?q?=20batch=20=E2=80=94=20onAbort=20leak,=20migrate=20missed=20sit?= =?UTF-8?q?es,=20tighten=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adopted 6 of the 7 review threads (skipping the debug-logging suggestion). 1. processFunctionCalls onAbort leak (CRITICAL): the new `finally { roundAbortController.abort(); }` in _runReasoningLoopInner would fire the `onAbort` handler in `processFunctionCalls` if scheduler.schedule or batchDone threw (the explicit removeEventListener at the old happy-path exit would be skipped), emitting spurious "Tool call cancelled by user abort." TOOL_RESULT events for every un-emitted callId — corrupting the transcript and misleading the model on the next round. Fixed by wrapping schedule + batchDone in their own try/finally so removeEventListener always runs before the outer finally's abort. 2. Migrate 3 new-from-main `new AbortController()` sites that this PR's audit missed (they came in via the merge from main): - goals/goalHook.ts (2 sites: judgeController, fallback signal) — consistency - hooks/promptHookRunner.ts (1 site: internalAbortController) — real leak (manual addEventListener without {once:true} or cleanup, exactly the pattern this PR exists to fix). Switched to createChildAbortController + finally `internalAbortController.abort()` for reverse cleanup on the success path. 3. Repro script (`listener-accumulation-repro.mjs`): inlined helper diverged from production — used WeakRef on child, while production was changed to strong-ref earlier in this PR. Updated the inlined copy to match production exactly, with a comment noting the intentional WeakRef-on-parent-only pattern. 4. warningHandler.ts: documented the snapshot-and-replace trade-offs in the JSDoc (late-added listeners bypass our filter; late `removeListener` calls have no effect on our fan-out). Tried the re-snapshot-per-warning approach the reviewer suggested but it doesn't work — `removeAllListeners('warning')` permanently removes the snapshot from Node's tracking, so a `process.listeners('warning')` filter at fan-out time always returns empty for prior listeners. The current design is the right trade-off; documentation is the correct fix. 5. abortController.test.ts: added three coverage gaps the reviewer identified — - createChildAbortController forwards custom maxListeners - manual cleanup() before scheduled timeout fires cancels it - timeoutMs <= 0 is treated as "no timeout" 6. Migrated `httpHookRunner.ts:202` (the lone caller of the deprecated `createCombinedAbortSignal`) to `combineAbortSignals` directly, then deleted `combinedAbortSignal.ts` + its test. All semantics covered by `combineAbortSignals` tests in abortController.test.ts. Refreshed `migration-completeness.txt` (now empty — clean grep). Tests: 194 pass across abortController/warningHandler/agent-runtime/ followup/hooks/goal/promptHook suites. Typecheck + prettier clean. --- .../listener-accumulation-repro.mjs | 26 ++-- packages/cli/src/utils/warningHandler.ts | 20 +++- .../core/src/agents/runtime/agent-core.ts | 24 ++-- packages/core/src/goals/goalHook.ts | 5 +- .../src/hooks/combinedAbortSignal.test.ts | 111 ------------------ .../core/src/hooks/combinedAbortSignal.ts | 21 ---- packages/core/src/hooks/httpHookRunner.ts | 6 +- packages/core/src/hooks/promptHookRunner.ts | 23 ++-- .../core/src/utils/abortController.test.ts | 35 ++++++ 9 files changed, 95 insertions(+), 176 deletions(-) delete mode 100644 packages/core/src/hooks/combinedAbortSignal.test.ts delete mode 100644 packages/core/src/hooks/combinedAbortSignal.ts diff --git a/docs/verification/abort-controller-refactor/listener-accumulation-repro.mjs b/docs/verification/abort-controller-refactor/listener-accumulation-repro.mjs index 2df834e91a..ea73016178 100644 --- a/docs/verification/abort-controller-refactor/listener-accumulation-repro.mjs +++ b/docs/verification/abort-controller-refactor/listener-accumulation-repro.mjs @@ -18,22 +18,17 @@ import { getEventListeners, setMaxListeners } from 'node:events'; -// Inline copy of the production helper so this script has no build-step -// dependency on the @qwen-code/qwen-code-core package. +// Inline copy of the production helper (packages/core/src/utils/abortController.ts) +// so this script has no build-step dependency on @qwen-code/qwen-code-core. +// Kept in sync — the child is held STRONGLY by the parent's listener closure +// (no WeakRef on child) so propagation works even when a caller drops the +// controller and keeps only the signal. WeakRef is used only on the PARENT, +// to keep child cleanup from pinning a long-lived parent in memory. function createAbortController(maxListeners = 50) { const c = new AbortController(); setMaxListeners(maxListeners, c.signal); return c; } -function propagateAbort(weakChild) { - const parent = this.deref(); - weakChild.deref()?.abort(parent?.reason); -} -function removeAbortHandler(weakHandler) { - const parent = this.deref(); - const handler = weakHandler.deref(); - if (parent && handler) parent.removeEventListener('abort', handler); -} function createChildAbortController(parent) { const child = createAbortController(); if (!parent) return child; @@ -42,13 +37,16 @@ function createChildAbortController(parent) { child.abort(parentSignal.reason); return child; } - const weakChild = new WeakRef(child); const weakParent = new WeakRef(parentSignal); - const handler = propagateAbort.bind(weakParent, weakChild); + const handler = () => { + child.abort(weakParent.deref()?.reason); + }; parentSignal.addEventListener('abort', handler, { once: true }); child.signal.addEventListener( 'abort', - removeAbortHandler.bind(weakParent, new WeakRef(handler)), + () => { + weakParent.deref()?.removeEventListener('abort', handler); + }, { once: true }, ); return child; diff --git a/packages/cli/src/utils/warningHandler.ts b/packages/cli/src/utils/warningHandler.ts index fc51114a36..8108712360 100644 --- a/packages/cli/src/utils/warningHandler.ts +++ b/packages/cli/src/utils/warningHandler.ts @@ -66,8 +66,23 @@ export function initializeWarningHandler(): void { if (installedHandler) return; // Snapshot everything currently listening on 'warning' (Node's default - // onWarning printer + any external telemetry subscribers). We will fan + // onWarning printer + any third-party telemetry subscribers). We will fan // out non-suppressed warnings back to them. + // + // Trade-offs to be aware of (documented for future readers): + // - Listeners ADDED via `process.on('warning', ...)` after this init are + // independent of our snapshot. They receive `process.emit('warning')` + // directly and bypass the suppression filter. Node's default printer + // is in our snapshot (not added later), so stderr stays clean; late + // telemetry will see the full warning stream including AbortSignal + // leaks. This is intentional — telemetry should see what's happening. + // - Listeners REMOVED via `process.removeListener('warning', fn)` after + // this init have no effect: we hold our own strong reference in the + // snapshot. Re-snapshotting per warning doesn't fix this because the + // listeners are already removed from Node's list (we called + // `process.removeAllListeners` to disable Node's default printing of + // suppressed warnings). Callers who need conditional fan-out should + // install BEFORE initializeWarningHandler. const priorListeners = [...process.listeners('warning')] as Array< (warning: Error) => void >; @@ -75,8 +90,7 @@ export function initializeWarningHandler(): void { installedHandler = (warning: Error) => { // Evaluate isDebugMode() per warning so DEBUG / QWEN_DEBUG can be // toggled at runtime (e.g. via a `/debug` slash command) without - // re-running initializeWarningHandler. Warnings are rare; the cost is - // a couple of env lookups. + // re-running initializeWarningHandler. if (!isDebugMode() && isSuppressed(warning)) return; for (const fn of priorListeners) { try { diff --git a/packages/core/src/agents/runtime/agent-core.ts b/packages/core/src/agents/runtime/agent-core.ts index 3339ab00d2..be36a5c38e 100644 --- a/packages/core/src/agents/runtime/agent-core.ts +++ b/packages/core/src/agents/runtime/agent-core.ts @@ -1324,17 +1324,23 @@ export class AgentCore { } }; abortController.signal.addEventListener('abort', onAbort, { once: true }); + try { + // If already aborted before the listener was registered, resolve + // immediately to avoid blocking forever. + if (abortController.signal.aborted) { + onAbort(); + } - // If already aborted before the listener was registered, resolve - // immediately to avoid blocking forever. - if (abortController.signal.aborted) { - onAbort(); + await scheduler.schedule(requests, abortController.signal); + await batchDone; + } finally { + // Always remove `onAbort` — otherwise a throw from scheduler.schedule + // or batchDone would leak it on the round controller, and the round's + // outer try/finally `.abort()` would later fire spurious cancellation + // TOOL_RESULT events for every un-emitted callId (corrupting the + // transcript and misleading the model on the next round). + abortController.signal.removeEventListener('abort', onAbort); } - - await scheduler.schedule(requests, abortController.signal); - await batchDone; - - abortController.signal.removeEventListener('abort', onAbort); } // If all tool calls failed, inform the model so it can re-evaluate. diff --git a/packages/core/src/goals/goalHook.ts b/packages/core/src/goals/goalHook.ts index ea6fd8657b..a35564e1a7 100644 --- a/packages/core/src/goals/goalHook.ts +++ b/packages/core/src/goals/goalHook.ts @@ -21,6 +21,7 @@ import { type ActiveGoal, } from './activeGoalStore.js'; import { judgeGoal } from './goalJudge.js'; +import { createAbortController } from '../utils/abortController.js'; import { createDebugLogger } from '../utils/debugLogger.js'; const debugLogger = createDebugLogger('GOAL_HOOK'); @@ -67,7 +68,7 @@ async function judgeGoalWithTimeout( // without this `judgeGoal`'s `generateContent` keeps running in the // background — leaking one request per timeout that accumulates across // goal-loop iterations. - const judgeController = new AbortController(); + const judgeController = createAbortController(); const linkedSignal = AbortSignal.any([args.signal, judgeController.signal]); try { return await Promise.race([ @@ -166,7 +167,7 @@ export function createGoalStopHookCallback(args: { return { continue: true }; } - const signal = context?.signal ?? new AbortController().signal; + const signal = context?.signal ?? createAbortController().signal; const verdict = await judgeGoalWithTimeout(config, { condition, lastAssistantText, diff --git a/packages/core/src/hooks/combinedAbortSignal.test.ts b/packages/core/src/hooks/combinedAbortSignal.test.ts deleted file mode 100644 index 1954237bc4..0000000000 --- a/packages/core/src/hooks/combinedAbortSignal.test.ts +++ /dev/null @@ -1,111 +0,0 @@ -/** - * @license - * Copyright 2026 Qwen Team - * SPDX-License-Identifier: Apache-2.0 - */ - -import { describe, it, expect, vi } from 'vitest'; -import { createCombinedAbortSignal } from './combinedAbortSignal.js'; - -describe('createCombinedAbortSignal', () => { - it('should return a non-aborted signal by default', () => { - const { signal, cleanup } = createCombinedAbortSignal(); - expect(signal.aborted).toBe(false); - cleanup(); - }); - - it('should abort after timeout', async () => { - const { signal, cleanup } = createCombinedAbortSignal(undefined, { - timeoutMs: 50, - }); - expect(signal.aborted).toBe(false); - - await new Promise((resolve) => setTimeout(resolve, 100)); - expect(signal.aborted).toBe(true); - cleanup(); - }); - - it('should abort when external signal is aborted', () => { - const externalController = new AbortController(); - const { signal, cleanup } = createCombinedAbortSignal( - externalController.signal, - ); - expect(signal.aborted).toBe(false); - - externalController.abort(); - expect(signal.aborted).toBe(true); - cleanup(); - }); - - it('should abort immediately if external signal is already aborted', () => { - const externalController = new AbortController(); - externalController.abort(); - - const { signal, cleanup } = createCombinedAbortSignal( - externalController.signal, - ); - expect(signal.aborted).toBe(true); - cleanup(); - }); - - it('should cleanup timeout timer', async () => { - const { signal, cleanup } = createCombinedAbortSignal(undefined, { - timeoutMs: 50, - }); - - cleanup(); - - // Wait longer than timeout - should not abort because timer was cleared - await new Promise((resolve) => setTimeout(resolve, 100)); - expect(signal.aborted).toBe(false); - }); - - it('should remove external abort listener on cleanup', () => { - const externalController = new AbortController(); - const removeListenerSpy = vi.spyOn( - externalController.signal, - 'removeEventListener', - ); - const { signal, cleanup } = createCombinedAbortSignal( - externalController.signal, - ); - - cleanup(); - externalController.abort(); - - expect(removeListenerSpy).toHaveBeenCalledWith( - 'abort', - expect.any(Function), - ); - expect(signal.aborted).toBe(false); - }); - - it('should work with both external signal and timeout', async () => { - const externalController = new AbortController(); - const { signal, cleanup } = createCombinedAbortSignal( - externalController.signal, - { timeoutMs: 200 }, - ); - - // Abort external signal before timeout - externalController.abort(); - expect(signal.aborted).toBe(true); - cleanup(); - }); - - it('should timeout before external signal', async () => { - const externalController = new AbortController(); - const { signal, cleanup } = createCombinedAbortSignal( - externalController.signal, - { timeoutMs: 50 }, - ); - - // Wait for timeout - await new Promise((resolve) => setTimeout(resolve, 100)); - expect(signal.aborted).toBe(true); - - // External signal is still not aborted - expect(externalController.signal.aborted).toBe(false); - cleanup(); - }); -}); diff --git a/packages/core/src/hooks/combinedAbortSignal.ts b/packages/core/src/hooks/combinedAbortSignal.ts deleted file mode 100644 index bf3c7501dc..0000000000 --- a/packages/core/src/hooks/combinedAbortSignal.ts +++ /dev/null @@ -1,21 +0,0 @@ -/** - * @license - * Copyright 2026 Qwen Team - * SPDX-License-Identifier: Apache-2.0 - */ - -import { combineAbortSignals } from '../utils/abortController.js'; - -/** - * @deprecated Use {@link combineAbortSignals} from `utils/abortController.js`. - * Thin wrapper preserved so existing callers (httpHookRunner) keep working - * during the abort-controller helper rollout. - */ -export function createCombinedAbortSignal( - externalSignal?: AbortSignal, - options?: { timeoutMs?: number }, -): { signal: AbortSignal; cleanup: () => void } { - return combineAbortSignals([externalSignal], { - timeoutMs: options?.timeoutMs, - }); -} diff --git a/packages/core/src/hooks/httpHookRunner.ts b/packages/core/src/hooks/httpHookRunner.ts index aad909ed3f..cc72d41b7f 100644 --- a/packages/core/src/hooks/httpHookRunner.ts +++ b/packages/core/src/hooks/httpHookRunner.ts @@ -7,7 +7,7 @@ import { createDebugLogger } from '../utils/debugLogger.js'; import { interpolateHeaders, interpolateUrl } from './envInterpolator.js'; import { UrlValidator } from './urlValidator.js'; -import { createCombinedAbortSignal } from './combinedAbortSignal.js'; +import { combineAbortSignals } from '../utils/abortController.js'; import { isBlockedAddress } from './ssrfGuard.js'; import { lookup as dnsLookup } from 'dns'; import type { @@ -199,8 +199,8 @@ export class HttpHookRunner { const timeout = hookConfig.timeout ? hookConfig.timeout * 1000 : DEFAULT_HTTP_TIMEOUT; - const { signal: combinedSignal, cleanup } = createCombinedAbortSignal( - signal, + const { signal: combinedSignal, cleanup } = combineAbortSignals( + [signal], { timeoutMs: timeout }, ); diff --git a/packages/core/src/hooks/promptHookRunner.ts b/packages/core/src/hooks/promptHookRunner.ts index 1f57bbf5a5..4e9b524267 100644 --- a/packages/core/src/hooks/promptHookRunner.ts +++ b/packages/core/src/hooks/promptHookRunner.ts @@ -5,6 +5,7 @@ */ import { z } from 'zod'; +import { createChildAbortController } from '../utils/abortController.js'; import { createDebugLogger } from '../utils/debugLogger.js'; import type { PromptHookConfig, @@ -229,21 +230,14 @@ export class PromptHookRunner { }, ]; - // Create internal AbortController to abort the request on timeout - const internalAbortController = new AbortController(); + // Internal AbortController to abort the request on timeout. Use + // createChildAbortController so parent-signal propagation gets `{once:true}` + // + reverse cleanup automatically — the old manual addEventListener path + // had no `{once:true}` and never removed the listener, leaking one + // listener per prompt-hook invocation on the long-lived parent. + const internalAbortController = createChildAbortController(signal); const internalSignal = internalAbortController.signal; - // Chain external signal to internal abort controller - if (signal) { - if (signal.aborted) { - internalAbortController.abort(); - } else { - signal.addEventListener('abort', () => { - internalAbortController.abort(); - }); - } - } - // Create timeout promise that also aborts the request let timeoutId: ReturnType | undefined; const timeoutPromise = new Promise((_, reject) => { @@ -320,6 +314,9 @@ export class PromptHookRunner { if (timeoutId) { clearTimeout(timeoutId); } + // Trigger reverse-cleanup of the parent-signal listener on the + // success path; no-op if already aborted via parent/timeout. + internalAbortController.abort(); } } diff --git a/packages/core/src/utils/abortController.test.ts b/packages/core/src/utils/abortController.test.ts index e81c4778fb..e81e617c61 100644 --- a/packages/core/src/utils/abortController.test.ts +++ b/packages/core/src/utils/abortController.test.ts @@ -98,6 +98,12 @@ describe('createChildAbortController', () => { child.abort('manual'); expect(child.signal.aborted).toBe(true); }); + + it('forwards a custom maxListeners through to the child signal', () => { + const parent = createAbortController(); + const child = createChildAbortController(parent, 123); + expect(getMaxListeners(child.signal)).toBe(123); + }); }); describe('combineAbortSignals', () => { @@ -178,6 +184,35 @@ describe('combineAbortSignals', () => { expect(() => cleanup()).not.toThrow(); }); + it('manual cleanup() cancels a pending timeout so it never fires', () => { + vi.useFakeTimers(); + try { + const { signal, cleanup } = combineAbortSignals([], { timeoutMs: 50 }); + cleanup(); + vi.advanceTimersByTime(100); + // Without the clearTimeout in cleanups[], the timer would still fire + // and abort the (already-cleaned) signal with TimeoutError. + expect(signal.aborted).toBe(false); + } finally { + vi.useRealTimers(); + } + }); + + it('treats timeoutMs <= 0 as "no timeout"', () => { + vi.useFakeTimers(); + try { + const zero = combineAbortSignals([], { timeoutMs: 0 }); + const negative = combineAbortSignals([], { timeoutMs: -1 }); + vi.advanceTimersByTime(1_000_000); + expect(zero.signal.aborted).toBe(false); + expect(negative.signal.aborted).toBe(false); + zero.cleanup(); + negative.cleanup(); + } finally { + vi.useRealTimers(); + } + }); + it('aborts and stops registering listeners once an input is found aborted mid-iteration', () => { const a = createAbortController(); const b = createAbortController();