fix(StreamHandler): decode output across chunk boundaries

Without line buffering, each stdout/stderr chunk was decoded separately, so
a UTF-8 character split between chunks reached the task as U+FFFD: ExifTool
printed "Каналов" and tasks saw "Кана��ов". Decoding once in StreamHandler
means tasks and taskData/noTaskData listeners only ever receive strings.

Author: mceachen

CHANGELOG.md

@@ -26,6 +26,12 @@ See [Semver](http://semver.org/).
   call rescans the task's accumulated output, so a task that printed 100,000
   lines took 30 seconds instead of 55 ms. `taskData` events now carry these
   batches, as they carry chunks without these options.
+- 🐞 A multi-byte UTF-8 character split across two stdout or stderr chunks no
+  longer reaches the task as U+FFFD replacement characters. Without
+  `isRetirementRequest` (or, for stderr, `shouldIgnoreStderrLine`), each chunk
+  was decoded separately: ExifTool's `-listx` output contained `Кана��ов` where
+  it printed `Каналов`. Tasks, `taskData`, and `noTaskData` now always receive
+  decoded strings, never `Buffer`s.
 
 ## [v19.4.1](https://github.com/photostructure/batch-cluster.js/releases/tag/v19.4.1)
 

src/BatchCluster.spec.ts

@@ -879,6 +879,31 @@ describe("BatchCluster", function () {
     });
   });
 
+  describe("UTF-8 split across chunks", function () {
+    let bc: BatchCluster;
+    afterEach(() => shutdown(bc));
+
+    for (const stream of ["stdout", "stderr"] as const) {
+      it(`decodes a ${stream} code point split across two chunks`, async function () {
+        setFailRatePct(0);
+        bc = listen(
+          new BatchCluster({
+            ...DefaultTestOptions,
+            processFactory,
+            maxProcs: 1,
+          }),
+        );
+        const output = await bc.enqueueTask(
+          new Task("split-utf8 " + stream, (stdout, stderr) =>
+            (stream === "stdout" ? stdout : (stderr ?? "")).trim(),
+          ),
+        );
+        expect(output).to.eql("🌻");
+        postAssertions();
+      });
+    }
+  });
+
   describe("maxProcs", function () {
     const iters = 50;
     const maxProcs = 10;

src/StreamHandler.spec.ts

@@ -967,7 +967,7 @@ describe("StreamHandler", function () {
       streamHandler.processStdout(bufferData, mockContext);
 
       expect(taskDataEvents).to.have.length(1);
-      expect(taskDataEvents[0]?.data).to.eql(bufferData);
+      expect(taskDataEvents[0]?.data).to.eql("test buffer data");
     });
 
     it("should handle Buffer data in stderr", function () {
@@ -979,6 +979,50 @@ describe("StreamHandler", function () {
         streamHandler.processStderr(bufferData, mockContext);
       }).to.not.throw();
     });
+
+    for (const stream of ["stdout", "stderr"] as const) {
+      describe(`${stream} decoding`, function () {
+        const write = (data: Buffer) =>
+          stream === "stdout"
+            ? streamHandler.processStdout(data, mockContext)
+            : streamHandler.processStderr(data, mockContext);
+        const end = () =>
+          stream === "stdout"
+            ? streamHandler.endStdout(mockContext)
+            : streamHandler.endStderr(mockContext);
+        let received: string[];
+
+        beforeEach(function () {
+          received = [];
+          mockContext.getCurrentTask = () =>
+            ({
+              pending: true,
+              onStdout: (data: string | Buffer) => {
+                received.push(String(data));
+              },
+              onStderr: (data: string | Buffer) => {
+                received.push(String(data));
+              },
+            }) as unknown as Task<unknown>;
+        });
+
+        it(`decodes a code point split across ${stream} chunks`, function () {
+          const bytes = Buffer.from("🌻\n");
+          write(bytes.subarray(0, 2));
+          expect(received).to.eql([]);
+          write(bytes.subarray(2));
+          expect(received).to.eql(["🌻\n"]);
+        });
+
+        it(`delivers a ${stream} code point truncated at EOF as U+FFFD`, function () {
+          write(Buffer.from("🌻").subarray(0, 2));
+          expect(received).to.eql([]);
+          end();
+          end();
+          expect(received).to.eql(["�"]);
+        });
+      });
+    }
   });
 
   describe("integration scenarios", function () {

src/StreamHandler.ts

@@ -133,7 +133,10 @@ export class StreamHandler {
   #onStdout(data: string | Buffer, context: StreamContext): void {
     if (data == null) return;
     if (this.#isRetirementRequest == null || this.#stdoutEnded) {
-      this.#routeStdout(data, context);
+      // The decoder retains a code point split across chunks, which decoding
+      // each chunk separately would turn into U+FFFD replacement characters.
+      const output = this.#stdoutDecoder.write(data);
+      if (output.length > 0) this.#routeStdout(output, context);
       return;
     }
 
@@ -232,7 +235,12 @@ export class StreamHandler {
   }
 
   #endStdout(context: StreamContext): void {
-    if (this.#isRetirementRequest == null || this.#stdoutEnded) return;
+    if (this.#isRetirementRequest == null) {
+      const output = this.#stdoutDecoder.end();
+      if (output.length > 0) this.#routeStdout(output, context);
+      return;
+    }
+    if (this.#stdoutEnded) return;
     const hadIncompleteLine = this.hasIncompleteOutputLine;
     this.#stdoutEnded = true;
     this.#consumeStdout(this.#stdoutDecoder.end(), context);
@@ -252,11 +260,7 @@ export class StreamHandler {
     if (fragment.length > 0) this.#routeStdout(fragment, context, owner);
   }
 
-  #routeStdout(
-    data: string | Buffer,
-    context: StreamContext,
-    owner?: LineOwner,
-  ): void {
+  #routeStdout(data: string, context: StreamContext, owner?: LineOwner): void {
     const task = owner == null ? context.getCurrentTask() : owner.task;
     if (task != null && task.pending) {
       // eslint-disable-next-line @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-argument
@@ -279,12 +283,13 @@ export class StreamHandler {
       this.#shouldIgnoreStderrLine == null &&
       this.#isRetirementRequest == null
     ) {
-      this.#routeStderr(data, context);
+      // See #onStdout:
+      this.#routeStderr(this.#stderrDecoder.write(data), context);
       return;
     }
 
     if (this.#stderrEnded) {
-      this.#routeStderr(data, context);
+      this.#routeStderr(this.#stderrDecoder.write(data), context);
       return;
     }
 
@@ -434,11 +439,13 @@ export class StreamHandler {
 
   #endStderr(context: StreamContext): void {
     if (
-      (this.#shouldIgnoreStderrLine == null &&
-        this.#isRetirementRequest == null) ||
-      this.#stderrEnded
-    )
+      this.#shouldIgnoreStderrLine == null &&
+      this.#isRetirementRequest == null
+    ) {
+      this.#routeStderr(this.#stderrDecoder.end(), context);
       return;
+    }
+    if (this.#stderrEnded) return;
 
     const hadIncompleteLine = this.hasIncompleteOutputLine;
     this.#stderrEnded = true;
@@ -461,11 +468,7 @@ export class StreamHandler {
     }
   }
 
-  #routeStderr(
-    data: string | Buffer,
-    context: StreamContext,
-    owner?: LineOwner,
-  ): void {
+  #routeStderr(data: string, context: StreamContext, owner?: LineOwner): void {
     if (blank(data)) return;
 
     const task = owner == null ? context.getCurrentTask() : owner.task;

src/test.js

@@ -142,6 +142,20 @@ async function onLine(line) {
         }
         break;
       }
+      case "split-utf8": {
+        // Writes a 4-byte code point in two parts, so the parent reads them in
+        // separate chunks. Like "lines", the stderr variant ends with FAIL on
+        // stderr.
+        const stderr = tokens[0] === "stderr";
+        const bytes = Buffer.from(
+          "🌻" + newline + (stderr ? "FAIL" : "PASS") + newline,
+        );
+        const stream = stderr ? process.stderr : process.stdout;
+        stream.write(bytes.subarray(0, 2));
+        await delay(50);
+        stream.write(bytes.subarray(2));
+        break;
+      }
       case "sleep": {
         const millis = parseInt(tokens[0] ?? "100");
         if (millis > 0) await delay(millis);