From b8070ed3a86755d7aec46ad17e7a8fdd4d268862 Mon Sep 17 00:00:00 2001 From: Jan Leon Date: Sat, 27 Jun 2026 03:04:33 +0200 Subject: [PATCH] Ignore disconnect races during in-band stream error handling (#5007) Integrated into release/v3.8.38 --- open-sse/handlers/audioSpeech.ts | 38 ++++++++--------- open-sse/utils/streamHandler.ts | 32 ++++++++++---- tests/unit/stream-handler.test.ts | 69 ++++++++++++++++++++++--------- 3 files changed, 93 insertions(+), 46 deletions(-) diff --git a/open-sse/handlers/audioSpeech.ts b/open-sse/handlers/audioSpeech.ts index 9bb280248c..8885d4edba 100644 --- a/open-sse/handlers/audioSpeech.ts +++ b/open-sse/handlers/audioSpeech.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "crypto"; import { CORS_HEADERS } from "../utils/cors.ts"; import { stripTrailingSlashes } from "../utils/urlSanitize.ts"; /** @@ -35,22 +34,27 @@ import { signAwsRequest } from "../utils/awsSigV4.ts"; /** * Return a CORS error response from an upstream fetch failure */ +function extractUpstreamErrorMessage(parsed) { + const detail = parsed?.detail; + const candidates = [ + parsed?.err_msg, + parsed?.error?.message, + typeof parsed?.error === "string" ? parsed.error : null, + parsed?.message, + typeof detail === "string" ? detail : detail?.message, + ]; + + const raw = candidates.find(Boolean); + return raw ? String(raw) : null; +} + function upstreamErrorResponse(res, errText) { // Always return JSON so the client can detect 401/credential errors reliably let errorMessage: string; try { const parsed = JSON.parse(errText); - // Extract a human-readable message from various error response shapes. - // Guard against `parsed.error` being an object (e.g. ElevenLabs returns - // { error: { message: "...", status_code: 401 } } or { detail: { ... } }) - const raw = - parsed?.err_msg || - parsed?.error?.message || - (typeof parsed?.error === "string" ? parsed.error : null) || - parsed?.message || - (typeof parsed?.detail === "string" ? parsed.detail : parsed?.detail?.message) || - null; - errorMessage = raw ? String(raw) : errText || `Upstream error (${res.status})`; + errorMessage = + extractUpstreamErrorMessage(parsed) || errText || `Upstream error (${res.status})`; } catch { errorMessage = errText || `Upstream error (${res.status})`; } @@ -791,8 +795,7 @@ function hexToBytes(audioHex) { } async function handleMinimaxSpeech(providerConfig, body, modelId, token) { - const voiceId = - (typeof body.voice === "string" && body.voice) || "English_expressive_narrator"; + const voiceId = (typeof body.voice === "string" && body.voice) || "English_expressive_narrator"; const res = await fetch(providerConfig.baseUrl, { method: "POST", headers: { @@ -835,12 +838,9 @@ async function handleMinimaxSpeech(providerConfig, body, modelId, token) { return upstreamErrorResponse(res, rawText); } - const baseResp = - ((data.base_resp || data.baseResp) as Record | undefined) || {}; + const baseResp = ((data.base_resp || data.baseResp) as Record | undefined) || {}; const statusCode = Number(baseResp.status_code ?? baseResp.statusCode ?? 0); - const statusMessage = String( - baseResp.status_msg || baseResp.statusMsg || data.message || "" - ); + const statusMessage = String(baseResp.status_msg || baseResp.statusMsg || data.message || ""); if (statusCode !== 0) { return errorResponse(502, `MiniMax TTS: ${statusMessage || "upstream error"}`); } diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 451cb7cffa..1229d71a8e 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -414,6 +414,13 @@ export function createDisconnectAwareStream(transformStream, streamController) { } controller.enqueue(value); } catch (error) { + if (!streamController.isConnected()) { + try { + controller.close(); + } catch {} + return; + } + streamController.handleError(error); // T35: Encapsulate mid-stream errors as SSE events instead of abruptly aborting @@ -421,15 +428,22 @@ export function createDisconnectAwareStream(transformStream, streamController) { const errorMsg = getErrorMessage(error); const statusCode = getErrorStatusCode(error); - for (const chunk of buildStreamErrorChunks( - errorMsg, - statusCode, - streamController.clientResponseFormat - )) { - controller.enqueue(chunk); + try { + for (const chunk of buildStreamErrorChunks( + errorMsg, + statusCode, + streamController.clientResponseFormat + )) { + controller.enqueue(chunk); + } + } catch { + // The downstream may have closed while we were formatting the in-band + // error event. The original stream error has already been recorded. } - controller.close(); + try { + controller.close(); + } catch {} } }, @@ -568,7 +582,9 @@ export function pipeWithDisconnect( }, }); - const transformedBody = providerResponse.body.pipeThrough(upstreamTap).pipeThrough(transformStream); + const transformedBody = providerResponse.body + .pipeThrough(upstreamTap) + .pipeThrough(transformStream); return createDisconnectAwareStream( { readable: transformedBody, writable: createNoopAbortWritable() }, wrappedController diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index 789d685d55..da2ff9ecba 100644 --- a/tests/unit/stream-handler.test.ts +++ b/tests/unit/stream-handler.test.ts @@ -504,6 +504,42 @@ test("pipeWithDisconnect does not double-clear transform errors already accounte assert.equal(pending.byAccount[connectionId][modelKey], 1); }); +test("createDisconnectAwareStream ignores reader errors after client disconnect", async () => { + let readableController!: ReadableStreamDefaultController; + let onErrorCalled = false; + const transformStream = { + readable: new ReadableStream({ + start(controller) { + readableController = controller; + }, + }), + writable: { + getWriter() { + return { + abort() {}, + }; + }, + }, + }; + const streamController = createStreamController({ + onError() { + onErrorCalled = true; + return true; + }, + }); + const stream = createDisconnectAwareStream(transformStream, streamController); + const reader = stream.getReader(); + const readPromise = reader.read(); + + streamController.handleDisconnect("ResponseAborted"); + readableController.error(new Error("Invalid state: Controller is already closed")); + + const result = await readPromise; + + assert.equal(result.done, true); + assert.equal(onErrorCalled, false, "disconnect races must not be recorded as upstream errors"); +}); + // Stall detection: tied to RAW upstream byte activity, not transform output. // Ports decolua/9router#1243 — reasoning models (Claude thinking, Kiro // EventStream binary frames) can stream raw bytes for long stretches while @@ -546,18 +582,19 @@ test("pipeWithDisconnect does NOT flag a slow but progressing upstream as stalle }, }); - const stream = pipeWithDisconnect( - new Response(source), - swallowingTransform, - streamController, - { stallTimeoutMs: 200 } - ); + const stream = pipeWithDisconnect(new Response(source), swallowingTransform, streamController, { + stallTimeoutMs: 200, + }); const text = await readStreamText(stream); // No stall error — final flush output reaches the client cleanly. assert.equal(text, "done"); - assert.equal(onErrorCalled, false, "stall watchdog must NOT fire on a slow but progressing upstream"); + assert.equal( + onErrorCalled, + false, + "stall watchdog must NOT fire on a slow but progressing upstream" + ); assert.doesNotMatch(text, /stall/i); assert.doesNotMatch(text, /"finish_reason":"error"/); }); @@ -583,12 +620,9 @@ test("pipeWithDisconnect flags a truly stalled upstream (no bytes for the full s }, }); - const stream = pipeWithDisconnect( - new Response(source), - new TransformStream(), - streamController, - { stallTimeoutMs: 80 } - ); + const stream = pipeWithDisconnect(new Response(source), new TransformStream(), streamController, { + stallTimeoutMs: 80, + }); const text = await readStreamText(stream); @@ -616,12 +650,9 @@ test("pipeWithDisconnect stall watchdog does not fire after normal stream comple }, }); - const stream = pipeWithDisconnect( - new Response(source), - new TransformStream(), - streamController, - { stallTimeoutMs: 50 } - ); + const stream = pipeWithDisconnect(new Response(source), new TransformStream(), streamController, { + stallTimeoutMs: 50, + }); const text = await readStreamText(stream);