From b37da6e3b73044b67d1158275db4d8588cf53e27 Mon Sep 17 00:00:00 2001 From: StreamClash Dev Date: Wed, 21 Oct 2026 17:15:42 -0400 Subject: [PATCH] perf(sse): avoid flattening fragmented block carry --- src/server/sse-payload-rewrite.ts | 15 +++- structure/transports/byte-accounting.md | 5 ++ tests/responses/sse-payload-rewrite.test.ts | 79 +++++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 99c67056eaa..d476b7e16cc 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/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; }, }; diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 048fc8ec2cf..1a15e792a58 100644 --- a/structure/transports/byte-accounting.md +++ b/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. diff --git a/tests/responses/sse-payload-rewrite.test.ts b/tests/responses/sse-payload-rewrite.test.ts index 92eef790e78..0b2bd55a17f 100644 --- a/tests/responses/sse-payload-rewrite.test.ts +++ b/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 });