vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-09 17:19:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

fix(core,cli): address PR #4366 review batch — onAbort leak, migrate missed sites, tighten tests

Browse source 
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.
This commit is contained in:
doudouOUC 2026-05-21 18:55:43 +08:00
parent d95a18e3df
commit 94e8c58121
9 changed files with 95 additions and 176 deletions
Download patch file Download diff file Expand all files Collapse all files

26
docs/verification/abort-controller-refactor/listener-accumulation-repro.mjs
View file

@ -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;

20
packages/cli/src/utils/warningHandler.ts
View file

@ -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 {

24
packages/core/src/agents/runtime/agent-core.ts
View file

@ -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.

5
packages/core/src/goals/goalHook.ts
View file

@ -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,

111
packages/core/src/hooks/combinedAbortSignal.test.ts
View file

@ -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();
});
});

21
packages/core/src/hooks/combinedAbortSignal.ts
View file

@ -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,
});
}

6
packages/core/src/hooks/httpHookRunner.ts
View file

@ -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 },
);

23
packages/core/src/hooks/promptHookRunner.ts
View file

@ -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<typeof setTimeout> | undefined;
const timeoutPromise = new Promise<never>((_, 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();
}
}

35
packages/core/src/utils/abortController.test.ts
View file

@ -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();