From 31883d53203d4191d38e91d84711907acbe63df5 Mon Sep 17 00:00:00 2001 From: Matt Van Horn Date: Thu, 2 Jul 2026 22:31:56 -0700 Subject: [PATCH] fix(core): raise stream idle timeout default to 5m and hint the env knob (#6107) The streaming inactivity watchdog aborted any stream silent for 120000ms, which breaks large-prompt ingest, /compress reprocessing, and long thinking phases on local models and reasoning proxies that legitimately emit no SSE deltas for over two minutes. Raise DEFAULT_STREAM_IDLE_TIMEOUT_MS to 300000 (5 minutes) and append an actionable hint to the timeout error pointing at QWEN_STREAM_IDLE_TIMEOUT_MS (or 0 to disable). Refs #5975 Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Co-authored-by: Shaojin Wen --- .../core/openaiContentGenerator/constants.ts | 6 ++-- .../openaiContentGenerator/pipeline.test.ts | 35 ++++++++++++++++--- .../core/openaiContentGenerator/pipeline.ts | 5 ++- 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/constants.ts b/packages/core/src/core/openaiContentGenerator/constants.ts index cc444d70a1..04f94122bb 100644 --- a/packages/core/src/core/openaiContentGenerator/constants.ts +++ b/packages/core/src/core/openaiContentGenerator/constants.ts @@ -1,8 +1,10 @@ export const DEFAULT_TIMEOUT = 120000; // Inactivity (no-chunk) timeout for streaming responses. The SDK `timeout` // only bounds connect + first response, so a stream that returns 200 then -// goes silent is otherwise unbounded; this watchdog aborts it. -export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 120000; +// goes silent is otherwise unbounded. The 4-minute default gives large-prompt +// ingest and long thinking phases room while staying below +// LoggingContentGenerator's stream-span idle timer, so this watchdog fires first. +export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 240000; // Env override (deployment knob) for the streaming inactivity timeout, so a // daemon deployment can tune it without code — the same way the QWEN_SERVE_* // params are set. An explicit ContentGeneratorConfig.streamIdleTimeoutMs still diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts index d2c2f3a1c5..9aca5a51be 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts @@ -3148,7 +3148,9 @@ describe('ContentGenerationPipeline', () => { const err = await captured; expect(err).toBeInstanceOf(StreamInactivityTimeoutError); expect((err as Error).message).toBe( - 'No stream activity for 1000ms after 0 chunks (stream lifetime: 1000ms)', + 'No stream activity for 1000ms after 0 chunks ' + + '(stream lifetime: 1000ms). Set QWEN_STREAM_IDLE_TIMEOUT_MS ' + + 'to increase this window (or 0 to disable it).', ); expect(err).toMatchObject({ code: 'ETIMEDOUT' }); expect((err as StreamInactivityTimeoutError).chunksReceived).toBe(0); @@ -3157,6 +3159,29 @@ describe('ContentGenerationPipeline', () => { expect(mockErrorHandler.handle).not.toHaveBeenCalled(); }); + it('includes the idle detail and env override hint in timeout errors', async () => { + const gated = gatedStream(); // never push/end → silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(1000); + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + const captured = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch((e: unknown) => e); + await vi.advanceTimersByTimeAsync(1000); + const err = await captured; + expect(err).toBeInstanceOf(StreamInactivityTimeoutError); + const message = (err as Error).message; + expect(message).toContain('No stream activity for 1000ms after 0 chunks'); + expect(message).toContain('QWEN_STREAM_IDLE_TIMEOUT_MS'); + }); + it('uses the default stream idle timeout when no override is configured', async () => { const gated = gatedStream(); // never push/end → silent (mockClient.chat.completions.create as Mock).mockResolvedValue( @@ -3524,7 +3549,7 @@ describe('ContentGenerationPipeline', () => { (mockClient.chat.completions.create as Mock).mockResolvedValue( gated.stream, ); - const p = buildPipeline(); // no config; invalid env → default (120000ms) + const p = buildPipeline(); // no config; invalid env → default const gen = await p.executeStream( streamingRequest(new AbortController().signal), 'id', @@ -3557,7 +3582,7 @@ describe('ContentGenerationPipeline', () => { (mockClient.chat.completions.create as Mock).mockResolvedValue( gated.stream, ); - const p = buildPipeline(); // no config; oversized env → default (120000ms) + const p = buildPipeline(); // no config; oversized env → default const gen = await p.executeStream( streamingRequest(new AbortController().signal), 'id', @@ -3657,7 +3682,7 @@ describe('ContentGenerationPipeline', () => { (mockClient.chat.completions.create as Mock).mockResolvedValue( gated.stream, ); - // Config is oversized → rejected; env = 4000 → used (not default 120000). + // Config is oversized → rejected; env = 4000 → used (not default). const p = buildPipeline(MAX_STREAM_IDLE_TIMEOUT_MS + 1); const gen = await p.executeStream( streamingRequest(new AbortController().signal), @@ -3673,7 +3698,7 @@ describe('ContentGenerationPipeline', () => { expect(settled).toBe(false); // not yet at the env value await vi.advanceTimersByTimeAsync(1); await consume; - expect(settled).toBe(true); // trips at 4000ms from the env (not 120000 default) + expect(settled).toBe(true); // trips at 4000ms from the env (not default) }); it('disables the watchdog when QWEN_STREAM_IDLE_TIMEOUT_MS=0', async () => { diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index 18b9ec6cd0..154c366ab9 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts @@ -56,7 +56,10 @@ export class StreamInactivityTimeoutError extends Error { readonly streamLifetimeMs: number, ) { super( - `No stream activity for ${idleMs}ms after ${chunksReceived} chunks (stream lifetime: ${streamLifetimeMs}ms)`, + `No stream activity for ${idleMs}ms after ${chunksReceived} chunks ` + + `(stream lifetime: ${streamLifetimeMs}ms). Set ` + + `${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV} to increase this window ` + + `(or 0 to disable it).`, ); this.name = 'StreamInactivityTimeoutError'; }