perf(shared): buffer fragmented local data-plane lines

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?.();
     }

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),
+      }
+    );
+  }
+});

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']);
+  });
 });