DEN-4233: preserve UTF-8 across JSON stream chunk boundaries

.github/workflows/ci.yml

@@ -0,0 +1,33 @@
+name: ci
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ci-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  parser-regressions:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - name: Check out source
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+
+      - name: Set up Node.js
+        uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7
+        with:
+          node-version: '22.23.1'
+          package-manager-cache: false
+
+      - name: Install test dependencies without native addon build
+        run: npm install --ignore-scripts --no-audit --no-fund --package-lock=false
+
+      - name: Build and test parser
+        run: npm test

package.json

@@ -30,7 +30,7 @@
     "bench:latency": "npm run build && bash bench/latency.sh",
     "bench:load": "npm run build && bash bench/load.sh",
     "bench:compare-load": "npm run build && node bench/compare-load.mjs",
-    "test": "npm run build && node test/run.js"
+    "test": "npm run build && node test/run.js && node test/chunk-boundary.test.js"
   },
   "repository": {
     "type": "git",

src/main.ts

@@ -2,6 +2,7 @@
 
 import * as stream from 'stream';
 import * as assert from 'node:assert/strict';
+import {StringDecoder} from 'node:string_decoder';
 
 import {RawStringSymbol, RawJSONBytesSymbol, JSONBytesSymbol} from './symbols.js';
 export {RawStringSymbol, RawJSONBytesSymbol, JSONBytesSymbol};
@@ -53,6 +54,7 @@ export class JSONParser<T = any> extends stream.Transform {
   delay = false;
   count = 1;
   wrapMetadata = false;
+  decoder = new StringDecoder('utf8');
 
   constructor(opts ?: JSONParserOpts) {
     super({objectMode: true, highWaterMark: 1});
@@ -189,19 +191,24 @@ export class JSONParser<T = any> extends stream.Transform {
   }
 
   _transform(chunk: any, encoding: string, cb: EVCb<void>) {
+    const bytes = Buffer.isBuffer(chunk)
+      ? chunk
+      : ArrayBuffer.isView(chunk)
+        ? Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength)
+        : Buffer.from(String(chunk ?? ''), encoding && encoding !== 'buffer' ? encoding as BufferEncoding : 'utf8');
 
     if (this.isTrackBytesRead) {
-      this.jpBytesRead += chunk.length;
+      this.jpBytesRead += bytes.length;
     }
 
-    let data = String(chunk || '');
+    let data = this.decoder.write(bytes);
 
     if (this.lastLineData) {
       data = this.lastLineData + data;
     }
 
     const lines = data.split(this.delimiter);
-    this.lastLineData = lines.pop();
+    this.lastLineData = lines.pop() || '';
 
     for (let l of lines) {
 
@@ -226,6 +233,10 @@ export class JSONParser<T = any> extends stream.Transform {
   }
 
   _flush(cb: Function) {
+    const decoderTail = this.decoder.end();
+    if (decoderTail) {
+      this.lastLineData += decoderTail;
+    }
 
     if (this.lastLineData) {
       this.handleJSON(this.lastLineData);

test/chunk-boundary.test.js

@@ -0,0 +1,49 @@
+#!/usr/bin/env node
+'use strict';
+
+import assert from 'node:assert/strict';
+import { PassThrough } from 'node:stream';
+
+import { JSONParser } from '../dist/main.js';
+
+async function collectStream(readable) {
+  return await new Promise((resolve, reject) => {
+    const out = [];
+    readable.on('data', value => out.push(value));
+    readable.on('error', reject);
+    readable.on('end', () => resolve(out));
+  });
+}
+
+async function parseAtSplit(payload, splitAt, opts = {}) {
+  const parser = new JSONParser(opts);
+  const input = new PassThrough();
+  const output = collectStream(input.pipe(parser));
+  const bytes = Buffer.from(payload, 'utf8');
+
+  input.write(bytes.subarray(0, splitAt));
+  input.end(bytes.subarray(splitAt));
+
+  return await output;
+}
+
+async function assertEveryByteSplit(payload, expected, opts = {}) {
+  const bytes = Buffer.from(payload, 'utf8');
+  for (let splitAt = 1; splitAt < bytes.length; splitAt += 1) {
+    const actual = await parseAtSplit(payload, splitAt, opts);
+    assert.deepEqual(actual, expected, `failed at byte split ${splitAt}/${bytes.length}`);
+  }
+}
+
+await assertEveryByteSplit(
+  '{"city":"Lima","note":"café ☕ 🚲"}\n{"ok":true}\n',
+  [{ city: 'Lima', note: 'café ☕ 🚲' }, { ok: true }]
+);
+
+await assertEveryByteSplit(
+  '{"first":"mañana"}∆∆∆{"second":"東京"}∆∆∆',
+  [{ first: 'mañana' }, { second: '東京' }],
+  { delimiter: '∆∆∆' }
+);
+
+process.stdout.write('ok - UTF-8 JSON and multibyte delimiters survive every byte split\n');