stream: normalize fused stateless transform results

Normalize each stateless transform result before passing it to the next transform in a fused run. This ensures that subsequent transforms always receive Uint8Array[] batches in both synchronous and asynchronous pipelines. Signed-off-by: Kamat, Trivikram <[email protected]> Assisted-by: codex:gpt-5.6-sol PR-URL: #65367 Fixes: #65366 Reviewed-By: James M Snell <[email protected]>
aduh95 · Sep 8, 2026 · Aug 2, 2026 · Aug 4, 2026 · Aug 5, 2026 · Aug 5, 2026
commit 9e2c477e26df8b084f7c9bc3a4ec87ff64975931
diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js
@@ -434,6 +434,16 @@ async function appendTransformResultAsyncSlow(target, result) {
   }
 }
 
+function normalizeTransformResultFast(result) {
+  if (isUint8ArrayBatch(result)) {
+    return result.length === 0 ? null : result;
+  }
+  if (isUint8Array(result)) return [result];
+  if (typeof result === 'string') return [toUint8Array(result)];
+  if (isAnyArrayBuffer(result)) return [new Uint8Array(result)];
+  if (ArrayBufferIsView(result)) return [arrayBufferViewToUint8Array(result)];
+}
+
 // =============================================================================
 // Sync Pipeline Implementation
 // =============================================================================
@@ -457,7 +467,17 @@ function* applyFusedStatelessSyncTransforms(source, run) {
         current = null;
         break;
       }
-      current = result;
+      if (i === run.length - 1) {
+        current = result;
+        continue;
+      }
+      current = normalizeTransformResultFast(result);
+      if (current === undefined) {
+        const normalized = [];
+        appendTransformResultSync(normalized, result);
+        current = normalized.length === 0 ? null : normalized[0];
+      }
+      if (current === null) break;
     }
     if (current === null) continue;
     // Inline normalization with Uint8Array[] batch as the fast path,
@@ -570,21 +590,24 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
   for await (const chunks of source) {
     let current = chunks;
     for (let i = 0; i < run.length; i++) {
-      const result = run[i](current, { __proto__: null, signal });
+      let result = run[i](current, { __proto__: null, signal });
+      if (isPromise(result)) result = await result;
       if (result === null) {
         current = null;
         break;
       }
-      if (isPromise(result)) {
-        const resolved = await result;
-        if (resolved === null) {
-          current = null;
-          break;
-        }
-        current = resolved;
-      } else {
+      if (i === run.length - 1) {
         current = result;
+        continue;
+      }
+      current = normalizeTransformResultFast(result);
+      if (current === undefined) {
+        const normalized = [];
+        const pendingResult = appendTransformResultAsync(normalized, result);
+        if (pendingResult !== undefined) await pendingResult;
+        current = normalized.length === 0 ? null : normalized[0];
       }
+      if (current === null) break;
     }
     if (current === null) continue;
     // Normalize the final output

diff --git a/test/parallel/test-stream-iter-transform-output.js b/test/parallel/test-stream-iter-transform-output.js
@@ -59,6 +59,36 @@ async function testSyncTransformReturnsFloat32Array() {
   assert.strictEqual(data.byteLength, 4);
 }
 
+// Consecutive stateless transforms normalize intermediate output (async)
+async function testConsecutiveTransformsNormalizeIntermediateOutput() {
+  const first = (chunks) => {
+    return chunks === null ? null : new Uint8Array([65]);
+  };
+  let receivedBatch = false;
+  const second = (chunks) => {
+    if (chunks !== null) receivedBatch = Array.isArray(chunks);
+    return chunks;
+  };
+  const data = await bytes(pull(from('x'), first, second));
+  assert.ok(receivedBatch);
+  assert.deepStrictEqual(data, new Uint8Array([65]));
+}
+
+// Consecutive stateless transforms normalize intermediate output (sync)
+async function testConsecutiveSyncTransformsNormalizeIntermediateOutput() {
+  const first = (chunks) => {
+    return chunks === null ? null : new Uint8Array([65]);
+  };
+  let receivedBatch = false;
+  const second = (chunks) => {
+    if (chunks !== null) receivedBatch = Array.isArray(chunks);
+    return chunks;
+  };
+  const data = bytesSync(pullSync(fromSync('x'), first, second));
+  assert.ok(receivedBatch);
+  assert.deepStrictEqual(data, new Uint8Array([65]));
+}
+
 // Stateless transform returns a sync generator (iterable)
 async function testTransformReturnsGenerator() {
   const tx = (chunks) => {
@@ -233,6 +263,8 @@ Promise.all([
   testSyncTransformReturnsArrayBuffer(),
   testTransformReturnsFloat32Array(),
   testSyncTransformReturnsFloat32Array(),
+  testConsecutiveTransformsNormalizeIntermediateOutput(),
+  testConsecutiveSyncTransformsNormalizeIntermediateOutput(),
   testTransformReturnsGenerator(),
   testSyncTransformReturnsGenerator(),
   testTransformReturnsAsyncGenerator(),