diff --git a/src/server/relay.ts b/src/server/relay.ts index 3b5aae236..c1d8c6b6a 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -6,7 +6,6 @@ import { addFinalRequestLog, httpStatusForRequestLogTerminal, inspectResponseLogJson, - inspectResponseLogSsePayload, inspectResponseLogSsePayloadParsed, recordFirstOutput, type RequestLogContext, @@ -359,53 +358,42 @@ export function trackSseForRequestLog( onFirstOutput?: () => void, ): ReadableStream { const reader = body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; let terminalReported = false; - const reportFirstOutput = createFirstOutputReporter(onFirstOutput); const reportTerminal = (status: ResponsesTerminalStatus) => { if (terminalReported) return; terminalReported = true; onTerminal(status); }; - - const inspectPayload = (payload: string | null) => { - if (!payload) return; - if (logCtx) inspectResponseLogSsePayload(logCtx, payload); - reportFirstOutput.payload(payload); - const status = terminalStatusFromSsePayload(payload); - if (status) reportTerminal(status); - }; - - const inspectChunk = (value: Uint8Array) => { - buffer += decoder.decode(value, { stream: true }); - let next: { block: string; rest: string } | null; - while ((next = nextSseBlock(buffer))) { - buffer = next.rest; - inspectPayload(sseDataPayload(next.block)); - } - }; + // Reuse the byte-bounded inspector so translated responses cannot retain an + // unterminated upstream frame or parse the same event once per observer. + const inspector = createSseInspector({ + onTerminal: reportTerminal, + logCtx, + onFirstOutput, + }); return new ReadableStream({ async pull(controller) { try { const { done, value } = await reader.read(); if (done) { - buffer += decoder.decode(); - if (buffer.trim()) inspectPayload(sseDataPayload(buffer)); + inspector.finish(); if (!terminalReported) reportTerminal("incomplete"); + inspector.dispose(); controller.close(); return; } - inspectChunk(value); + inspector.feed(value); controller.enqueue(value); } catch (err) { if (!terminalReported) reportTerminal("incomplete"); + inspector.dispose(); try { controller.error(err); } catch { /* already torn down */ } } }, cancel(reason) { + inspector.dispose(); onCancel(); reader.cancel(reason).catch(() => {}); }, @@ -516,38 +504,23 @@ export function relaySseWithHeartbeat( ): ReadableStream | null { if (!body) return null; const reader = body.getReader(); - const decoder = new TextDecoder(); const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n"); let timer: ReturnType | undefined; let closed = false; let clientCancelled = false; let terminalReported = false; - let buffer = ""; const reportTerminal = (status: ResponsesTerminalStatus) => { if (terminalReported || clientCancelled || closed) return; terminalReported = true; onTerminal?.(status); }; - - const inspectPayload = (payload: string | null) => { - if (!payload) return; - const status = terminalStatusFromSsePayload(payload); - if (status) reportTerminal(status); - }; - - const inspectChunk = (value: Uint8Array) => { - buffer += decoder.decode(value, { stream: true }); - let next: { block: string; rest: string } | null; - while ((next = nextSseBlock(buffer))) { - buffer = next.rest; - inspectPayload(sseDataPayload(next.block)); - } - }; + const inspector = createSseInspector({ onTerminal: reportTerminal }); const cleanup = () => { if (closed) return; closed = true; + inspector.dispose(); if (timer) clearInterval(timer); timer = undefined; options?.onDone?.(); @@ -569,14 +542,13 @@ export function relaySseWithHeartbeat( try { const { done, value } = await reader.read(); if (done) { - buffer += decoder.decode(); - if (buffer.trim()) inspectPayload(sseDataPayload(buffer)); + inspector.finish(); if (!terminalReported && !clientCancelled) reportTerminal("incomplete"); cleanup(); controller.close(); return; } - inspectChunk(value); + inspector.feed(value); controller.enqueue(value); } catch (err) { if (!clientCancelled) reportTerminal("incomplete"); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index ed401de19..24535af60 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -61,6 +61,11 @@ The two-shape contract is mirror-commented in `src/server/index.ts`; the real and the platform matrix lives in `tests/bun-stream-caps.test.ts`. Keep all three in lockstep with any passthrough-policy change. +Translated response request-log tracking and the heartbeat relay also reuse +`createSseInspector`. This keeps every client-facing SSE observation path on +the same byte-bounded, discard-and-resynchronize frame policy and ensures the +request-log, first-output, and terminal observers share one payload parse. + ## Standalone Search and exact account selectors `POST /v1/alpha/search` retains the selected model in its request body. When that value is an diff --git a/tests/sse-inspector-bounds.test.ts b/tests/sse-inspector-bounds.test.ts index 82c25ad5e..8ecc9621c 100644 --- a/tests/sse-inspector-bounds.test.ts +++ b/tests/sse-inspector-bounds.test.ts @@ -4,7 +4,9 @@ import { getInspectionCounters, MAX_COMPLETED_OUTPUT_ITEMS, MAX_INSPECTION_SSE_FRAME_BYTES, + relaySseWithHeartbeat, resetInspectionCountersForTest, + trackSseForRequestLog, } from "../src/server/relay"; import type { RequestLogContext } from "../src/server/request-log"; @@ -379,3 +381,66 @@ describe("createSseInspector parse-once", () => { } }); }); + +describe("client-facing SSE wrapper bounds", () => { + test("translated request-log tracking discards an oversized frame, resynchronizes, and preserves bytes", async () => { + const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`); + const delimiter = encoder.encode("\n\n"); + const terminal = frame(completedEvent("translated-after-cap")); + const chunks = [oversized, delimiter, terminal]; + const terminals: string[] = []; + const tracked = trackSseForRequestLog( + streamFromChunks(chunks), + status => terminals.push(status), + () => {}, + {} as RequestLogContext, + () => {}, + ); + + expect(await readAllBytes(tracked)).toEqual(joinBytes(chunks)); + expect(terminals).toEqual(["completed"]); + expect(getInspectionCounters().frameCapOverflows).toBe(1); + expect(getInspectionCounters().frameBufferHighWaterBytes) + .toBeLessThanOrEqual(MAX_INSPECTION_SSE_FRAME_BYTES); + }); + + test("translated request-log tracking parses a complete payload once for all observers", async () => { + const originalParse = JSON.parse; + let parses = 0; + JSON.parse = ((text: string) => { + parses += 1; + return originalParse(text); + }) as typeof JSON.parse; + try { + const tracked = trackSseForRequestLog( + streamFromChunks([frame(completedEvent("translated-parse-once"))]), + () => {}, + () => {}, + {} as RequestLogContext, + () => {}, + ); + await readAllBytes(tracked); + expect(parses).toBe(1); + } finally { + JSON.parse = originalParse; + } + }); + + test("heartbeat relay applies the same frame bound without changing upstream bytes", async () => { + const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`); + const delimiter = encoder.encode("\r\n\r\n"); + const terminal = frame(completedEvent("heartbeat-after-cap")); + const chunks = [oversized, delimiter, terminal]; + const terminals: string[] = []; + const relayed = relaySseWithHeartbeat( + streamFromChunks(chunks), + new AbortController(), + 60_000, + status => terminals.push(status), + )!; + + expect(await readAllBytes(relayed)).toEqual(joinBytes(chunks)); + expect(terminals).toEqual(["completed"]); + expect(getInspectionCounters().frameCapOverflows).toBe(1); + }); +});