fix(server): bound client-facing SSE frame retention

src/server/relay.ts

@@ -11,11 +11,16 @@ import {
   type RequestLogContext,
   type RequestLogEntry,
 } from "./request-log";
+import {
+  BoundedSseFrameBuffer,
+  joinSseFrameBytes,
+  MAX_CLIENT_SSE_FRAME_BYTES,
+} from "./sse-frame-buffer";
 
 const nativePassthroughSseResponses = new WeakSet<Response>();
 const eagerRelaySseResponses = new WeakSet<Response>();
 
-export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024;
+export const MAX_INSPECTION_SSE_FRAME_BYTES = MAX_CLIENT_SSE_FRAME_BYTES;
 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;
@@ -104,30 +109,29 @@ export type SseTerminalOutputBoundary = {
 
 /**
  * Frame-aware client output boundary shared by both native Responses relays.
- * It buffers only the current incomplete SSE block, forwards complete blocks
- * through the first Responses terminal, and drops every later block/byte.
+ * It buffers only the current incomplete SSE block under the same hard byte
+ * cap as inspection, forwards complete blocks through the first Responses
+ * terminal, and drops every later block/byte.
  */
 export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary {
-  let decoder: TextDecoder | null = new TextDecoder();
-  const encoder = new TextEncoder();
-  let buffer = "";
+  const decoder = new TextDecoder();
+  const framer = new BoundedSseFrameBuffer(MAX_INSPECTION_SSE_FRAME_BYTES);
   let terminal = false;
   let done = false;
   let disposed = false;
 
-  const process = (flush: boolean): Uint8Array => {
-    if (disposed || terminal) return new Uint8Array(0);
-    let output = "";
+  const processFrames = (
+    frames: ReturnType<BoundedSseFrameBuffer["feed"]>,
+  ): Uint8Array => {
+    if (disposed || terminal || frames.length === 0) return new Uint8Array(0);
+    const output: Uint8Array[] = [];
     let responsesTerminal = false;
-    for (;;) {
-      const next = nextSseBlock(buffer);
-      if (!next) break;
-      buffer = next.rest;
-      const payload = sseDataPayload(next.block);
-      if (!responsesTerminal) output += next.block + next.delimiter;
+    for (const frame of frames) {
+      const payload = sseDataPayload(decoder.decode(frame.block));
+      if (!responsesTerminal) output.push(frame.block, frame.delimiter);
       if (payload === "[DONE]") {
         done = true;
-        if (responsesTerminal) output += next.block + next.delimiter;
+        if (responsesTerminal) output.push(frame.block, frame.delimiter);
         continue;
       }
       if (!responsesTerminal && payload && terminalStatusFromSsePayload(payload)) {
@@ -136,33 +140,26 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary {
     }
     if (responsesTerminal) {
       terminal = true;
-      buffer = "";
-    }
-    if (flush && !terminal && buffer.length > 0) {
-      output += buffer;
-      buffer = "";
+      framer.dispose();
     }
-    return encoder.encode(output);
+    return joinSseFrameBytes(output);
   };
 
   return {
     feed(chunk) {
       if (disposed || terminal) return new Uint8Array(0);
-      buffer += decoder!.decode(chunk, { stream: true });
-      return process(false);
+      return processFrames(framer.feed(chunk));
     },
     finish() {
       if (disposed || terminal) return new Uint8Array(0);
-      buffer += decoder!.decode();
-      return process(true);
+      return framer.finish();
     },
     terminalSeen: () => terminal,
     doneSeen: () => done,
     dispose() {
       if (disposed) return;
       disposed = true;
-      decoder = null;
-      buffer = "";
+      framer.dispose();
     },
   };
 }
@@ -622,17 +619,6 @@ function delimiterLengthAt(
   return byteAt(index + 3) === 10 ? 4 : 0;
 }
 
-function joinedBytes(slices: readonly Uint8Array[], byteLength: number): Uint8Array {
-  if (slices.length === 1 && slices[0]!.byteLength === byteLength) return slices[0]!;
-  const joined = new Uint8Array(byteLength);
-  let offset = 0;
-  for (const slice of slices) {
-    joined.set(slice, offset);
-    offset += slice.byteLength;
-  }
-  return joined;
-}
-
 /**
  * Per-chunk SSE inspection state machine shared by consumeForInspection,
  * consumeForResponseLogMetadata, and the eager bounded relay (relay-eager.ts).
@@ -653,8 +639,8 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
   let reported = false;
   let sawTerminal = false;
   let disposed = false;
-  let delimiterTail = new Uint8Array(0);
-  let candidateSlices: Uint8Array[] = [];
+  let delimiterTail: Uint8Array = new Uint8Array(0);
+  let candidate: Uint8Array = new Uint8Array(0);
   let candidateBytes = 0;
   let discardingOversizedFrame = false;
   const reportFirstOutput = createFirstOutputReporter(handlers.onFirstOutput);
@@ -668,7 +654,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
 
   const clearFrameState = (): void => {
     delimiterTail = new Uint8Array(0);
-    candidateSlices = [];
+    candidate = new Uint8Array(0);
     candidateBytes = 0;
     discardingOversizedFrame = false;
   };
@@ -688,6 +674,30 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
     firstResponseId = undefined;
   };
 
+  const ensureCandidateCapacity = (requiredBytes: number): void => {
+    if (candidate.byteLength >= requiredBytes) return;
+    let capacity = candidate.byteLength === 0
+      ? Math.min(MAX_INSPECTION_SSE_FRAME_BYTES, Math.max(requiredBytes, 4096))
+      : candidate.byteLength;
+    while (capacity < requiredBytes) {
+      capacity = Math.min(
+        MAX_INSPECTION_SSE_FRAME_BYTES,
+        Math.max(requiredBytes, capacity * 2),
+      );
+    }
+    const grown = new Uint8Array(capacity);
+    if (candidateBytes > 0) grown.set(candidate.subarray(0, candidateBytes));
+    candidate = grown;
+  };
+
+  const takeCandidate = (): Uint8Array => {
+    if (candidateBytes === 0) return new Uint8Array(0);
+    const frame = candidate.slice(0, candidateBytes);
+    candidate = new Uint8Array(0);
+    candidateBytes = 0;
+    return frame;
+  };
+
   const retainCandidateSlice = (slice: Uint8Array): void => {
     if (slice.byteLength === 0 || discardingOversizedFrame) return;
     const nextBytes = candidateBytes + slice.byteLength;
@@ -696,7 +706,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
       Math.min(nextBytes, MAX_INSPECTION_SSE_FRAME_BYTES),
     );
     if (nextBytes > MAX_INSPECTION_SSE_FRAME_BYTES) {
-      candidateSlices = [];
+      candidate = new Uint8Array(0);
       candidateBytes = 0;
       discardingOversizedFrame = true;
       inspectionCounters.frameCapOverflows += 1;
@@ -706,10 +716,8 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
       reconstructionTainted = true;
       return;
     }
-    // `subarray()` aliases the upstream chunk's backing buffer. Copy only the
-    // live candidate bytes so a tiny trailing frame cannot pin a multi-MiB
-    // chunk whose preceding frames have already been consumed.
-    candidateSlices.push(slice.slice());
+    ensureCandidateCapacity(nextBytes);
+    candidate.set(slice, candidateBytes);
     candidateBytes = nextBytes;
   };
 
@@ -845,9 +853,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
       return;
     }
     const sourceBytes = candidateBytes;
-    const frame = joinedBytes(candidateSlices, sourceBytes);
-    candidateSlices = [];
-    candidateBytes = 0;
+    const frame = takeCandidate();
     if (reported && !handlers.onCompletedResponse) return;
     const decoded = decoder!.decode(frame);
     scanPayload(sseDataPayload(decoded), sourceBytes);
@@ -904,7 +910,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
         delimiterTail = new Uint8Array(0);
         if (!discardingOversizedFrame && candidateBytes > 0 && !reported) {
           const sourceBytes = candidateBytes;
-          const decoded = decoder!.decode(joinedBytes(candidateSlices, sourceBytes));
+          const decoded = decoder!.decode(takeCandidate());
           scanPayload(decoded.trim() ? sseDataPayload(decoded) : null, sourceBytes);
         }
       } finally {