Ignore disconnect races during in-band stream error handling (#5007)

Integrated into release/v3.8.38

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<string, unknown> | undefined) || {};
+  const baseResp = ((data.base_resp || data.baseResp) as Record<string, unknown> | 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"}`);
   }

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

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