From 6fc5f45b25e4161c5f69d986a94ebb1273f2c4c2 Mon Sep 17 00:00:00 2001 From: moe Date: Fri, 4 Sep 2026 19:16:47 -0400 Subject: [PATCH] perf(shared): buffer fragmented local data-plane lines Model: gpt-6 --- packages/shared/src/local-loro-data-plane.ts | 51 ++++++++++----- .../local-loro-data-plane-splitter.bench.ts | 49 +++++++++++++++ ...ocal-loro-data-plane-utf8-chunking.test.ts | 62 +++++++++++++++++++ 3 files changed, 148 insertions(+), 14 deletions(-) create mode 100644 packages/shared/tests/local-loro-data-plane-splitter.bench.ts diff --git a/packages/shared/src/local-loro-data-plane.ts b/packages/shared/src/local-loro-data-plane.ts index ca354d9c8..d24ad642e 100644 --- a/packages/shared/src/local-loro-data-plane.ts +++ b/packages/shared/src/local-loro-data-plane.ts @@ -578,10 +578,16 @@ export function createJsonLineSplitter(options: { onOverflow?: () => void; }): (chunk: string | Uint8Array) => void { const decodeChunk = createUtf8StreamDecoder(); - let buffer = ''; + let parts: string[] = []; + let bufferedLength = 0; + let retryChunk = ''; let discardingOversizedLine = false; return (input: string | Uint8Array) => { let chunk = typeof input === 'string' ? input : decodeChunk(input); + if (retryChunk) { + chunk = retryChunk + chunk; + retryChunk = ''; + } if (discardingOversizedLine) { const newlineIndex = chunk.indexOf('\n'); if (newlineIndex < 0) { @@ -590,22 +596,39 @@ export function createJsonLineSplitter(options: { discardingOversizedLine = false; chunk = chunk.slice(newlineIndex + 1); } - buffer += chunk; - let newlineIndex = buffer.indexOf('\n'); - while (newlineIndex >= 0) { - const line = buffer.slice(0, newlineIndex).trim(); - buffer = buffer.slice(newlineIndex + 1); - if (line) { - if (options.maxBufferBytes !== undefined && line.length > options.maxBufferBytes) { - options.onOverflow?.(); - } else { - options.onLine(line); + let start = 0; + let newlineIndex = chunk.indexOf('\n'); + try { + while (newlineIndex >= 0) { + let line = chunk.slice(start, newlineIndex); + if (parts.length > 0) { + parts.push(line); + line = parts.join(''); + parts = []; + bufferedLength = 0; + } + line = line.trim(); + start = newlineIndex + 1; + if (line) { + if (options.maxBufferBytes !== undefined && line.length > options.maxBufferBytes) { + options.onOverflow?.(); + } else { + options.onLine(line); + } } + newlineIndex = chunk.indexOf('\n', start); } - newlineIndex = buffer.indexOf('\n'); + } catch (error) { + retryChunk = chunk.slice(start); + throw error; + } + if (start < chunk.length) { + parts.push(chunk.slice(start)); + bufferedLength += chunk.length - start; } - if (options.maxBufferBytes !== undefined && buffer.length > options.maxBufferBytes) { - buffer = ''; + if (options.maxBufferBytes !== undefined && bufferedLength > options.maxBufferBytes) { + parts = []; + bufferedLength = 0; discardingOversizedLine = true; options.onOverflow?.(); } diff --git a/packages/shared/tests/local-loro-data-plane-splitter.bench.ts b/packages/shared/tests/local-loro-data-plane-splitter.bench.ts new file mode 100644 index 000000000..103518d7a --- /dev/null +++ b/packages/shared/tests/local-loro-data-plane-splitter.bench.ts @@ -0,0 +1,49 @@ +import { strict as assert } from 'node:assert'; +import { bench, describe } from 'vitest'; +import { createJsonLineSplitter } from '../src/local-loro-data-plane'; + +// Run: pnpm --filter @lody/shared exec vitest bench --run local-loro-data-plane-splitter +const largeLine = JSON.stringify({ payload: 'x'.repeat(1_100_000) }); +const shortLine = JSON.stringify({ type: 'ping', sequence: 1 }); +const encoder = new TextEncoder(); +const cases = [ + ...[64, 4096, 65536].map((chunkSize) => ({ + name: `1.1 MB line / ${chunkSize} byte chunks`, + line: largeLine, + count: 1, + chunkSize, + })), + { name: '10000 short frames / 64 KiB batches', line: shortLine, count: 10_000, chunkSize: 65536 }, +]; + +describe('JSON line splitter', () => { + for (const { name, line, count, chunkSize } of cases) { + const bytes = encoder.encode(`${line}\n`.repeat(count)); + const chunks: Uint8Array[] = []; + for (let offset = 0; offset < bytes.length; offset += chunkSize) { + chunks.push(bytes.subarray(offset, offset + chunkSize)); + } + + const lines: string[] = []; + const verify = createJsonLineSplitter({ onLine: (value) => lines.push(value) }); + for (const chunk of chunks) verify(chunk); + assert.equal(lines.length, count); + for (const value of lines) assert.equal(value, line); + + let consumed = 0; + bench( + name, + () => { + const split = createJsonLineSplitter({ onLine: (value) => (consumed += value.length) }); + for (const chunk of chunks) split(chunk); + }, + { + time: 200, + iterations: 3, + warmupTime: 100, + warmupIterations: 1, + teardown: () => assert.ok(consumed > 0), + } + ); + } +}); diff --git a/packages/shared/tests/local-loro-data-plane-utf8-chunking.test.ts b/packages/shared/tests/local-loro-data-plane-utf8-chunking.test.ts index 2c8c5dea5..925111389 100644 --- a/packages/shared/tests/local-loro-data-plane-utf8-chunking.test.ts +++ b/packages/shared/tests/local-loro-data-plane-utf8-chunking.test.ts @@ -52,4 +52,66 @@ describe('local data-plane UTF-8 chunk decoding', () => { splitLines('2}\n'); expect(lines).toEqual(['{"a":1}', '{"b":2}']); }); + + it('reassembles a large frame from small byte chunks without emitting a partial line', () => { + const frame = JSON.stringify({ data: 'x'.repeat(1_100_000) }); + const bytes = new TextEncoder().encode(frame); + const lines: string[] = []; + const splitLines = createJsonLineSplitter({ onLine: (line) => lines.push(line) }); + for (let offset = 0; offset < bytes.length; offset += 64) { + splitLines(bytes.subarray(offset, offset + 64)); + } + expect(lines).toEqual([]); + splitLines('\n{"next":true}\n'); + expect(lines).toEqual([frame, '{"next":true}']); + }); + + it('trims complete lines before checking the cap but caps untrimmed partial lines', () => { + const events: string[] = []; + const splitLines = createJsonLineSplitter({ + maxBufferBytes: 4, + onLine: (line) => events.push(line), + onOverflow: () => events.push('overflow'), + }); + splitLines(' 12'); + splitLines('34 \r\n\t \n'); + splitLines('1234'); + splitLines(''); + splitLines('\n'); + splitLines(' '); + splitLines('discarded'); + splitLines('\n12345\n12\n3'); + splitLines('4\n'); + expect(events).toEqual(['1234', '1234', 'overflow', 'overflow', '12', '34']); + }); + + it('counts decoded characters across multibyte chunk boundaries for the cap', () => { + const events: string[] = []; + const splitLines = createJsonLineSplitter({ + maxBufferBytes: 4, + onLine: (line) => events.push(line), + onOverflow: () => events.push('overflow'), + }); + const bytes = new TextEncoder().encode('软😀件\n软😀件多\n好\n'); + for (const byte of bytes) { + splitLines(Uint8Array.of(byte)); + } + expect(events).toEqual(['软😀件', 'overflow', '好']); + }); + + it('retains unprocessed lines and a partial tail when a callback throws', () => { + const lines: string[] = []; + const splitLines = createJsonLineSplitter({ + onLine: (line) => { + lines.push(line); + if (line === 'first') { + throw new Error('callback failed'); + } + }, + }); + splitLines('fir'); + expect(() => splitLines('st\nsecond\nthi')).toThrow('callback failed'); + splitLines('rd\n'); + expect(lines).toEqual(['first', 'second', 'third']); + }); });