vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-08-31 02:06:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 16

fix(core): isolate OpenAI SDK abort listener leak with per-request child controllers (#4810)

Browse source 
* fix(core): isolate OpenAI SDK abort listener leak with per-request child controllers

OpenAI SDK v5.11.0's `fetchWithTimeout` (client.mjs:324) adds an abort
listener on the caller's signal without `{once: true}` or cleanup on
request completion. PR #4366 removed the `raiseAbortListenerCap` band-aid
(which had silenced the warning by setting maxListeners to Infinity) under
the assumption that `createChildAbortController` in agent-core covered it.
However, the SDK's internal listener leak was never addressed — it
accumulates on whichever signal reaches the SDK across retries and rounds.

Fix: wrap the signal passed to `client.chat.completions.create()` in a
per-request `createChildAbortController`. The SDK's leaked listener stays
on the short-lived child signal. The `finally` block aborts the child,
triggering reverse-cleanup that removes the parent listener. For streaming,
the cleanup runs when the async generator is fully consumed or abandoned.

Closes #4423

* fix(core): skip child controller when no parent signal is provided

Tests that don't pass an abortSignal expect `signal: undefined` to reach
the SDK. Only create the per-request child when a parent signal exists.

* fix(core): capture narrowed abort controller for closure use

TS does not propagate narrowing into the nested `drainThenCleanup`
generator, so referencing `perRequestAc` inside the closure failed
with TS18048. Capture the narrowed value in a local const after the
guard so the finally block sees a non-nullable controller.

* fix(core): address review — guard streaming create() with try/catch + add cleanup tests

1. Wrap the streaming `client.chat.completions.create()` call in try/catch
   so a network/DNS/proxy error still aborts the per-request child
   controller (same pattern as the non-streaming path).

2. Add three cleanup-invariant tests:
   - child signal aborted after full stream consumption
   - child signal aborted after consumer break
   - child signal aborted when SDK create() throws

* fix(test): correct inaccurate comment and remove trailing whitespace

- Update error-handling test comment: the error does propagate to the
  consumer via the async generator, not handled only internally
- Remove trailing whitespace on blank line in early-break test

---------

Co-authored-by: tanzhenxin <tanzhenxing1987@gmail.com>
This commit is contained in:
易良 2026-06-10 10:54:45 +08:00 • committed by GitHub
parent 0def5f77ff
commit 7f08ec6aeb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 207 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

151
packages/core/src/core/openaiContentGenerator/pipeline.test.ts
View file

@ -1201,10 +1201,38 @@ describe('ContentGenerationPipeline', () => {
await pipeline.execute(request, 'test-id');
expect(mockClient.chat.completions.create).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({ signal: abortController.signal }),
// The pipeline wraps the caller's signal in a per-request child
// to isolate OpenAI SDK listener leaks, so the SDK receives a
// child AbortSignal, not the original.
const call = (mockClient.chat.completions.create as Mock).mock.calls[0];
const sdkSignal = call[1]?.signal;
expect(sdkSignal).toBeInstanceOf(AbortSignal);
expect(sdkSignal).not.toBe(abortController.signal);
});
it('should propagate parent abort to SDK child signal', async () => {
const abortController = new AbortController();
const request: GenerateContentParameters = {
model: 'test-model',
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
config: { abortSignal: abortController.signal },
};
let capturedSignal: AbortSignal | undefined;
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
(mockClient.chat.completions.create as Mock).mockImplementation(
(_req: unknown, opts: { signal: AbortSignal }) => {
capturedSignal = opts.signal;
abortController.abort();
return { choices: [{ message: { content: 'ok' } }] };
},
);
(mockConverter.convertOpenAIResponseToGemini as Mock).mockReturnValue(
new GenerateContentResponse(),
);
await pipeline.execute(request, 'test-id');
expect(capturedSignal!.aborted).toBe(true);
});
});
@ -1378,17 +1406,18 @@ describe('ContentGenerationPipeline', () => {
);
// Assert
// The stream should handle the error internally - errors during iteration don't propagate to the consumer
// Instead, they are handled internally by the pipeline
// The error propagates to the consumer via the async generator;
// errorHandler.handle() is also called internally by the pipeline.
const results = [];
let caughtError: unknown;
try {
for await (const result of resultGenerator) {
results.push(result);
}
} catch (error) {
// This is expected - the error should propagate from the stream processing
expect(error).toBe(testError);
caughtError = error;
}
expect(caughtError).toBe(testError);
expect(results).toHaveLength(0); // No results due to error
expect(mockErrorHandler.handle).toHaveBeenCalledWith(
@ -1579,10 +1608,112 @@ describe('ContentGenerationPipeline', () => {
// Consume stream
}
expect(mockClient.chat.completions.create).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({ signal: abortController.signal }),
// Per-request child signal isolates SDK listener leaks
const call = (mockClient.chat.completions.create as Mock).mock.calls[0];
const sdkSignal = call[1]?.signal;
expect(sdkSignal).toBeInstanceOf(AbortSignal);
expect(sdkSignal).not.toBe(abortController.signal);
});
it('should abort child signal after stream is fully consumed', async () => {
const abortController = new AbortController();
const request: GenerateContentParameters = {
model: 'test-model',
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
config: { abortSignal: abortController.signal },
};
const mockStream = {
async *[Symbol.asyncIterator]() {
yield {
id: 'chunk-1',
choices: [{ delta: { content: 'Hello' }, finish_reason: 'stop' }],
};
},
};
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
(mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
new GenerateContentResponse(),
);
(mockClient.chat.completions.create as Mock).mockResolvedValue(
mockStream,
);
const resultGenerator = await pipeline.executeStream(request, 'test-id');
const sdkSignal = (mockClient.chat.completions.create as Mock).mock
.calls[0][1]?.signal as AbortSignal;
expect(sdkSignal.aborted).toBe(false);
for await (const _result of resultGenerator) {
// Consume stream
}
expect(sdkSignal.aborted).toBe(true);
});
it('should abort child signal when consumer breaks early', async () => {
const abortController = new AbortController();
const request: GenerateContentParameters = {
model: 'test-model',
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
config: { abortSignal: abortController.signal },
};
const mockStream = {
async *[Symbol.asyncIterator]() {
yield {
id: 'chunk-1',
choices: [{ delta: { content: 'a' }, finish_reason: null }],
};
yield {
id: 'chunk-2',
choices: [{ delta: { content: 'b' }, finish_reason: 'stop' }],
};
},
};
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
(mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
new GenerateContentResponse(),
);
(mockClient.chat.completions.create as Mock).mockResolvedValue(
mockStream,
);
const resultGenerator = await pipeline.executeStream(request, 'test-id');
const sdkSignal = (mockClient.chat.completions.create as Mock).mock
.calls[0][1]?.signal as AbortSignal;
for await (const _result of resultGenerator) {
break;
}
expect(sdkSignal.aborted).toBe(true);
});
it('should abort child signal when SDK create() throws', async () => {
const abortController = new AbortController();
const request: GenerateContentParameters = {
model: 'test-model',
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
config: { abortSignal: abortController.signal },
};
let capturedSignal: AbortSignal | undefined;
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
(mockClient.chat.completions.create as Mock).mockImplementation(
(_req: unknown, opts: { signal: AbortSignal }) => {
capturedSignal = opts.signal;
throw new Error('network failure');
},
);
await expect(
pipeline.executeStream(request, 'test-id'),
).rejects.toThrow();
expect(capturedSignal!.aborted).toBe(true);
});
it('should merge finishReason and usageMetadata from separate chunks', async () => {

87
packages/core/src/core/openaiContentGenerator/pipeline.ts
View file

@ -19,6 +19,7 @@ import { TaggedThinkingParser } from './taggedThinkingParser.js';
import type { PipelineConfig, RequestContext } from './types.js';
import { redactProxyError } from '../../utils/runtimeFetchOptions.js';
import { runtimeDiagnostics } from '../../utils/runtimeDiagnostics.js';
import { createChildAbortController } from '../../utils/abortController.js';
/**
* Error thrown when the API returns an error embedded as stream content
@ -53,20 +54,32 @@ export class ContentGenerationPipeline {
userPromptId,
false,
async (openaiRequest, context) => {
const openaiResponse = (await this.client.chat.completions.create(
openaiRequest,
{
signal: request.config?.abortSignal,
},
)) as OpenAI.Chat.ChatCompletion;
// Wrap in a per-request child so the OpenAI SDK's leaked abort
// listener (client.mjs fetchWithTimeout — no {once:true}, no
// removeEventListener) stays on a short-lived signal instead of
// accumulating on the caller's long-lived round signal.
const parentSignal = request.config?.abortSignal;
const perRequestAc = parentSignal
? createChildAbortController(parentSignal)
: undefined;
try {
const openaiResponse = (await this.client.chat.completions.create(
openaiRequest,
{
signal: perRequestAc?.signal,
},
)) as OpenAI.Chat.ChatCompletion;
const geminiResponse =
OpenAIContentConverter.convertOpenAIResponseToGemini(
openaiResponse,
context,
);
const geminiResponse =
OpenAIContentConverter.convertOpenAIResponseToGemini(
openaiResponse,
context,
);
return geminiResponse;
return geminiResponse;
} finally {
perRequestAc?.abort();
}
},
);
}
@ -80,16 +93,48 @@ export class ContentGenerationPipeline {
userPromptId,
true,
async (openaiRequest, context) => {
// Stage 1: Create OpenAI stream
const stream = (await this.client.chat.completions.create(
openaiRequest,
{
signal: request.config?.abortSignal,
},
)) as AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
// Per-request child — same rationale as the non-streaming path.
const parentSignal = request.config?.abortSignal;
const perRequestAc = parentSignal
? createChildAbortController(parentSignal)
: undefined;
let stream: AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
try {
// Stage 1: Create OpenAI stream. Wrapped in try so a network /
// DNS / proxy error during the SDK call still cleans up the
// per-request child (same pattern as the non-streaming path).
stream = (await this.client.chat.completions.create(openaiRequest, {
signal: perRequestAc?.signal,
})) as AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
} catch (e) {
perRequestAc?.abort();
throw e;
}
// Stage 2: Process stream with conversion and logging
return this.processStreamWithLogging(stream, context, request);
// Stage 2: Process stream with conversion and logging.
// When a per-request controller exists, wrap in an async generator
// that aborts it once the stream is fully consumed or abandoned, so
// the child signal's reverse-cleanup fires and the parent listener
// is released.
if (!perRequestAc) {
return this.processStreamWithLogging(stream, context, request);
}
// Capture the narrowed controller so the closure below sees a non-
// nullable type (TS does not propagate narrowing into nested funcs).
const ac = perRequestAc;
const innerStream = this.processStreamWithLogging(
stream,
context,
request,
);
async function* drainThenCleanup(): AsyncGenerator<GenerateContentResponse> {
try {
yield* innerStream;
} finally {
ac.abort();
}
}
return drainThenCleanup();
},
);
}