perf(sse): avoid flattening fragmented block carry

src/server/sse-payload-rewrite.ts

@@ -96,6 +96,8 @@ export function createSseBlockBuffer(
   let buffer = "";
   let offset = 0;
   let scanOffset = 0;
+  let needsScan = false;
+  let lastCodeUnit = Number.NaN;
   let bufferBytes = 0;
   const delimiterPattern = /(?:\r\n|\r(?!\n)|\n){2}/g;
 
@@ -111,7 +113,7 @@ export function createSseBlockBuffer(
       if (!fragment) return;
       let fragmentBytes = Buffer.byteLength(fragment, "utf8");
       // Decoder output never splits a surrogate pair, but keep the helper exact for string callers.
-      const last = buffer.charCodeAt(buffer.length - 1);
+      const last = lastCodeUnit;
       const first = fragment.charCodeAt(0);
       if (offset < buffer.length && last >= 0xd800 && last <= 0xdbff && first >= 0xdc00 && first <= 0xdfff) {
         fragmentBytes -= 2;
@@ -125,17 +127,24 @@ export function createSseBlockBuffer(
         reservation.commitRetained();
         budget.releaseRetained(bufferBytes, scope);
         bufferBytes = nextBytes;
+        lastCodeUnit = fragment.charCodeAt(fragment.length - 1);
+        needsScan ||= fragment.includes("\r") || fragment.includes("\n");
       } catch (error) {
         reservation.release();
         throw error;
       }
     },
     next() {
+      if (!needsScan) {
+        scanOffset = Math.max(offset, buffer.length - 3);
+        return null;
+      }
       delimiterPattern.lastIndex = scanOffset;
       const match = delimiterPattern.exec(buffer);
       if (!match) {
         // A delimiter is at most four code units. Revisit only its possible split prefix.
         scanOffset = Math.max(offset, buffer.length - 3);
+        needsScan = false;
         return null;
       }
       const start = match.index;
@@ -153,6 +162,8 @@ export function createSseBlockBuffer(
         buffer = "";
         offset = 0;
         scanOffset = 0;
+        needsScan = false;
+        lastCodeUnit = Number.NaN;
       }
       return { block, delimiter };
     },
@@ -164,6 +175,8 @@ export function createSseBlockBuffer(
       buffer = "";
       offset = 0;
       scanOffset = 0;
+      needsScan = false;
+      lastCodeUnit = Number.NaN;
       bufferBytes = 0;
     },
   };

structure/transports/byte-accounting.md

@@ -138,6 +138,11 @@ cancellation or consumer closure; a buffer-limit failure terminates without anot
 `src/server/sse-payload-rewrite.ts` shares an incremental block buffer with native Chat. It scans
 only new input plus a three-character delimiter prefix and counts consumed blocks rather than
 remaining suffixes. CR, LF and CRLF share partial-event, injection/drop and EOF behavior.
+After an unsuccessful delimiter search, fragments without CR or LF cannot create a delimiter;
+the buffer advances its scan cursor without passing the growing carry to the regex again.
+Queued appends retain scan eligibility until drained. This adds no payload buffer or budget operation.
+Surrogate-boundary accounting caches the final code unit from each fragment rather than indexing
+the growing carry. A failed scan retains that boundary; emptying or clearing the buffer resets it.
 Output admission precedes its single UTF-8 encoding;
 failed enqueue and cancellation release the reservation without re-entering a disposed rewrite.
 Old/new buffer overlap remains charged against the same translator cap.

tests/responses/sse-payload-rewrite.test.ts

@@ -176,6 +176,85 @@ describe("SSE payload rewrite composition", () => {
     }
   });
 
+  test("does not send a growing unfinished frame to the delimiter regex on each fragment", () => {
+    const text = `data: ${"x".repeat(256 * 1024)}\r\n\r\n`;
+    const budget = createTestTranslatorBudget();
+    const buffer = createSseBlockBuffer(budget);
+    const original = RegExp.prototype.exec;
+    const pattern = /(?:\r\n|\r(?!\n)|\n){2}/g.source;
+    let regexInputUnits = 0;
+    const exec = spyOn(RegExp.prototype, "exec").mockImplementation(function (this: RegExp, input: string) {
+      if (this.source === pattern && this.global) regexInputUnits += input.length;
+      return original.call(this, input);
+    });
+    try {
+      let delivered = "";
+      for (let index = 0; index < text.length; index += 512) {
+        buffer.append(text.slice(index, index + 512));
+        let block;
+        while ((block = buffer.next())) delivered += block.block + block.delimiter;
+      }
+      expect(delivered).toBe(text);
+      // Deterministic work bound: even a near-end regex search can flatten a
+      // growing rope, so count the full subject passed to that native operation.
+      expect(regexInputUnits).toBeLessThan(text.length * 4);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { exec.mockRestore(); buffer.clear(); }
+  });
+
+  test("append boundary checks do not index into the growing carry on each fragment", () => {
+    const text = `data: ${"x".repeat(256 * 1024)}\n\n`;
+    const budget = createTestTranslatorBudget(), buffer = createSseBlockBuffer(budget);
+    const original = String.prototype.charCodeAt;
+    let indexedSubjectUnits = 0;
+    const charCodeAt = spyOn(String.prototype, "charCodeAt").mockImplementation(function (this: string, position: number) {
+      indexedSubjectUnits += this.length;
+      return original.call(this, position);
+    });
+    try {
+      let delivered = "";
+      for (let index = 0; index < text.length; index += 512) {
+        buffer.append(text.slice(index, index + 512));
+        let block;
+        while ((block = buffer.next())) delivered += block.block + block.delimiter;
+      }
+      expect(delivered).toBe(text);
+      expect(indexedSubjectUnits).toBeLessThan(text.length * 4);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { charCodeAt.mockRestore(); buffer.clear(); }
+  });
+
+  test("a failed delimiter scan retains the surrogate boundary of the carry", () => {
+    const budget = createTestTranslatorBudget(), buffer = createSseBlockBuffer(budget);
+    buffer.append("data: \r\ud800");
+    expect(buffer.next()).toBeNull();
+    buffer.append("\udc00");
+    expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength("data: \r\ud800\udc00", "utf8"));
+    buffer.clear();
+    expect(budget.snapshot().currentBytes).toBe(0);
+  });
+
+  test("scan eligibility survives queued appends, compaction and clear", () => {
+    const budget = createTestTranslatorBudget();
+    const buffer = createSseBlockBuffer(budget);
+    let reference = "";
+    for (const fragments of [["data: one", "x", "\r", "\n", "\r", "\n", "tail"], ["plain", "x", "\n", "\n", "data: two\r\r", "last"], ["\ud800", "\udc00", "\n", "\n"]]) {
+      for (const fragment of fragments) { buffer.append(fragment); reference += fragment; }
+      for (;;) {
+        const expected = nextSseBlock(reference), actual = buffer.next();
+        expect(actual).toEqual(expected && {block: expected.block, delimiter: expected.delimiter});
+        if (!expected) break;
+        reference = expected.rest;
+        buffer.compact();
+      }
+      expect(buffer.tail()).toBe(reference);
+      expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(reference, "utf8"));
+      buffer.clear(); reference = "";
+      buffer.append("plain"); expect(buffer.next()).toBeNull(); buffer.clear();
+    }
+    expect(budget.snapshot().currentBytes).toBe(0);
+  });
+
   test("rejects an oversized replacement before encoding or delivering it", async () => {
     const upstream = streamFromText("data: small\n\n");
     const budget = createTestTranslatorBudget({ maxTurnBytes: 64 });