From 4a53a1659fdcd26913417f65a9bac0d46da1e8be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?= =?UTF-8?q?=29?= Date: Thu, 4 Jun 2026 22:26:46 +0800 Subject: [PATCH] fix(proxy): timeout stalled OpenAI-compatible streams (#548) Guard OpenAI-compatible proxy streaming bodies with the configured AI request timeout so a provider that emits partial SSE and then idles cannot leave proxy consumers waiting forever. This is a proxy-level fix found while investigating #548; it does not claim to close the broader desktop interruption issue. Tested: bun test src/server/__tests__/proxy-network-settings.test.ts Tested: bun run check:server Confidence: medium Scope-risk: narrow --- .../__tests__/proxy-network-settings.test.ts | 61 ++++++++++++++++++- src/server/proxy/handler.ts | 58 +++++++++++++++++- 2 files changed, 114 insertions(+), 5 deletions(-) diff --git a/src/server/__tests__/proxy-network-settings.test.ts b/src/server/__tests__/proxy-network-settings.test.ts index d4bdb812..eae73730 100644 --- a/src/server/__tests__/proxy-network-settings.test.ts +++ b/src/server/__tests__/proxy-network-settings.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' import * as fs from 'fs/promises' import * as os from 'os' import * as path from 'path' -import { handleProxyRequest } from '../proxy/handler.js' +import { handleProxyRequest, withStreamIdleTimeout } from '../proxy/handler.js' import { ProviderService } from '../services/providerService.js' import { resetSettingsCache } from '../../utils/settings/settingsCache.js' @@ -180,7 +180,7 @@ describe('proxy network settings', () => { } }) - test('uses configured AI request timeout only while opening streaming upstream requests', async () => { + test('uses configured AI request timeout while opening and reading streaming upstream requests', async () => { await fs.writeFile( path.join(tmpDir, 'settings.json'), JSON.stringify({ @@ -262,7 +262,11 @@ describe('proxy network settings', () => { expect(res.status).toBe(200) expect(timeoutCalls).toEqual([]) - expect(timers).toEqual([{ ms: 180_000, cleared: true }]) + expect(timers).toEqual([ + { ms: 180_000, cleared: true }, + { ms: 180_000, cleared: true }, + { ms: 180_000, cleared: true }, + ]) } finally { globalThis.clearTimeout = originalClearTimeout globalThis.setTimeout = originalSetTimeout @@ -270,4 +274,55 @@ describe('proxy network settings', () => { globalThis.fetch = originalFetch } }) + + test('fails a streaming upstream body that stops producing chunks', async () => { + const stalled = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n')) + }, + }) + + await expect(new Response(withStreamIdleTimeout(stalled, 20)).text()) + .rejects + .toThrow('Upstream stream idle timeout after 20ms') + }) + + test('propagates streaming upstream body errors before the idle timeout fires', async () => { + let pulls = 0 + const upstream = new ReadableStream({ + pull(controller) { + pulls += 1 + if (pulls === 1) { + controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n')) + return + } + controller.error(new Error('upstream body failed')) + }, + }) + const reader = withStreamIdleTimeout(upstream, 1_000).getReader() + + expect(await reader.read()).toEqual({ + done: false, + value: new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n'), + }) + await expect(reader.read()).rejects.toThrow('upstream body failed') + }) + + test('cancels the upstream body when the downstream stream is canceled', async () => { + let cancelReason: unknown = null + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n')) + }, + cancel(reason) { + cancelReason = reason + }, + }) + const reader = withStreamIdleTimeout(upstream, 1_000).getReader() + + expect((await reader.read()).done).toBe(false) + await reader.cancel('downstream closed') + + expect(cancelReason).toBe('downstream closed') + }) }) diff --git a/src/server/proxy/handler.ts b/src/server/proxy/handler.ts index 2a1583ac..4cf64a10 100644 --- a/src/server/proxy/handler.ts +++ b/src/server/proxy/handler.ts @@ -69,6 +69,58 @@ async function fetchUpstreamWithTimeout( } } +export function withStreamIdleTimeout( + upstream: ReadableStream, + timeoutMs: number, +): ReadableStream { + let reader: ReadableStreamDefaultReader | null = null + let timer: ReturnType | null = null + + const clearIdleTimer = () => { + if (timer) { + clearTimeout(timer) + timer = null + } + } + + return new ReadableStream({ + async start(controller) { + reader = upstream.getReader() + let timedOut = false + + const armIdleTimer = () => { + clearIdleTimer() + timer = setTimeout(() => { + timedOut = true + void reader?.cancel('stream idle timeout').catch(() => undefined) + controller.error(new Error(`Upstream stream idle timeout after ${timeoutMs}ms`)) + }, timeoutMs) + } + + try { + armIdleTimer() + while (true) { + const { done, value } = await reader.read() + if (done) break + if (timedOut) break + + controller.enqueue(value) + armIdleTimer() + } + clearIdleTimer() + if (!timedOut) controller.close() + } catch (err) { + clearIdleTimer() + if (!timedOut) controller.error(err) + } + }, + cancel(reason) { + clearIdleTimer() + return reader?.cancel(reason) + }, + }) +} + export async function handleProxyRequest(req: Request, url: URL): Promise { const providerMatch = url.pathname.match(/^\/proxy\/providers\/([^/]+)\/v1\/messages$/) const providerId = providerMatch ? decodeURIComponent(providerMatch[1]!) : undefined @@ -207,7 +259,8 @@ async function handleOpenaiChat( { status: 502 }, ) } - const anthropicStream = openaiChatStreamToAnthropic(upstream.body, body.model) + const upstreamBody = withStreamIdleTimeout(upstream.body, aiRequestTimeoutMs) + const anthropicStream = openaiChatStreamToAnthropic(upstreamBody, body.model) return new Response(anthropicStream, { status: 200, headers: { @@ -278,7 +331,8 @@ async function handleOpenaiResponses( { status: 502 }, ) } - const anthropicStream = openaiResponsesStreamToAnthropic(upstream.body, body.model) + const upstreamBody = withStreamIdleTimeout(upstream.body, aiRequestTimeoutMs) + const anthropicStream = openaiResponsesStreamToAnthropic(upstreamBody, body.model) return new Response(anthropicStream, { status: 200, headers: {