From d2c9c356845c68e66c44483988aaa7fb1ede1beb Mon Sep 17 00:00:00 2001 From: luvs01 Date: Fri, 7 Aug 2026 16:54:01 +0900 Subject: [PATCH] fix(responses): bound terminal SSE relay --- src/server/relay.ts | 10 +++++++++- src/server/responses/core.ts | 7 ++++++- tests/passthrough-abort.test.ts | 1 + tests/relay-eager.test.ts | 16 ++++++++++++++++ tests/sse-failed-tail.test.ts | 13 ++++++++++++- 5 files changed, 44 insertions(+), 3 deletions(-) diff --git a/src/server/relay.ts b/src/server/relay.ts index 3b5aae2366..b24c2cc590 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -17,6 +17,7 @@ const nativePassthroughSseResponses = new WeakSet(); const eagerRelaySseResponses = new WeakSet(); export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024; +export const MAX_CLIENT_SSE_FRAME_BYTES = 4 * 1024 * 1024; export const MAX_COMPLETED_OUTPUT_ITEMS = 256; export const MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES = 8 * 1024 * 1024; export const MAX_TAIL_ERROR_MESSAGE_CHARS = 512; @@ -150,7 +151,14 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { feed(chunk) { if (disposed || terminal) return new Uint8Array(0); buffer += decoder!.decode(chunk, { stream: true }); - return process(false); + const output = process(false); + if (encoder.encode(buffer).byteLength > MAX_CLIENT_SSE_FRAME_BYTES) { + // Drop the oversized partial before the relay catch path calls finish(); + // reflecting it into a synthetic failure would defeat the memory cap. + buffer = ""; + throw new Error("upstream SSE frame exceeded the safe limit"); + } + return output; }, finish() { if (disposed || terminal) return new Uint8Array(0); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 2d00ab0c43..cb2cefc5ea 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2221,7 +2221,12 @@ async function handleResponsesInner( const rewrittenBody = clientBlockRewrite !== undefined || payloadRewrites.length > 0 ? relaySseWithBlockRewrite(nativeBody, clientBlockRewrite ?? payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)), translatorBudget) : nativeBody; - const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); + // Known-bad Windows Bun runtimes must keep the no-rewrite client branch + // as a native tee relay. A JS pull wrapper reintroduces Bun#32111 when the + // client disconnects; fixed runtimes already took the eager path above. + const clientBody = process.platform === "win32" && !needsClientRewrite + ? nativeBody + : relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); return markNativePassthroughSseResponse(new Response(clientBody, { status: upstreamResponse.status, headers, diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index 5c782b6072..134dfc83e5 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -62,6 +62,7 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(sseBranch).toContain("rewritePayload: composeSsePayloadRewrites(...payloadRewrites)"); // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed. expect(sseBranch).toContain("relaySseWithFailedTail(rewrittenBody, upstream"); + expect(sseBranch).toContain('process.platform === "win32" && !needsClientRewrite'); expect(sseBranch).toContain("new Response(clientBody"); expect(sseBranch).toContain("markNativePassthroughSseResponse"); // #314/phase 100 two-platform contract: the real core gate delegates to the diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index 9840ead4b0..66edc065e7 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -284,6 +284,22 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { expect(budget.snapshot().currentBytes).toBe(0); }); + test("bounds delimiter-free frames before they can bypass the client queue cap", async () => { + const up = controlledUpstream(); + const upstream = new AbortController(); + const { hooks } = makeHooks(); + const relayed = relaySseEagerBounded(up.stream, upstream, hooks, { maxQueueBytes: 1 }); + const reading = readAll(relayed); + + up.push(new Uint8Array(4 * 1024 * 1024 + 1).fill(120)); + up.close(); + + const text = await reading; + expect(text).not.toContain("x".repeat(64)); + expect(text).toContain("upstream SSE frame exceeded the safe limit"); + expect(upstream.signal.aborted).toBe(true); + }); + test("blocks without a data field pass through untouched before the terminal", async () => { const up = controlledUpstream(); const { hooks } = makeHooks(); diff --git a/tests/sse-failed-tail.test.ts b/tests/sse-failed-tail.test.ts index 70876dafdb..72a59d9ef0 100644 --- a/tests/sse-failed-tail.test.ts +++ b/tests/sse-failed-tail.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; import { relaySseWithFailedTail, relayWithAbort } from "../src/server"; import { relaySseEagerBounded, type EagerRelayHooks } from "../src/server/relay-eager"; -import { MAX_TAIL_ERROR_MESSAGE_CHARS } from "../src/server/relay"; +import { MAX_CLIENT_SSE_FRAME_BYTES, MAX_TAIL_ERROR_MESSAGE_CHARS } from "../src/server/relay"; import { TranslatorBudgetExceededError } from "../src/lib/translator-budget"; const encoder = new TextEncoder(); @@ -140,6 +140,17 @@ describe("relaySseWithFailedTail", () => { expect(out.endsWith("data: [DONE]\n\n")).toBe(true); }); + test("fails closed when an unterminated SSE frame exceeds the client buffer cap", async () => { + const upstream = new AbortController(); + const chunk = "x".repeat(MAX_CLIENT_SSE_FRAME_BYTES / 2); + const out = await drain(relaySseWithFailedTail(sourceStream([chunk, chunk, "x"]), upstream)); + + expect(out).not.toContain(chunk); + expect(out).toContain("upstream SSE frame exceeded the safe limit"); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + expect(upstream.signal.aborted).toBe(true); + }); + test("translator overflow failed tail preserves translation_buffer_limit", async () => { const upstream = new AbortController(); const error = new TranslatorBudgetExceededError("live_transient", 32 * 1024 * 1024);