stream: normalize fused stateless transform results

Project: nodejs/node | Language: JavaScript | Concepts: data-structures, refactoring, error-handling, testing-strategy | Difficulty: mid
Link: https://github.com/nodejs/node/pull/65367

The problem

When chaining ("fusing") multiple stateless transforms in Node.js streams, the intermediate results passed between transforms could be inconsistently typed—sometimes a single Uint8Array, sometimes a batch (Uint8Array[]), or even other types. This inconsistency could break downstream transforms that expect a normalized batch format, especially in mixed sync/async pipelines.

Why it's non-trivial

A naive fix might just pass the output of one transform directly to the next. However, transforms can return many different types (e.g., strings, buffers, array buffers, or arrays), and downstream code may expect a normalized batch (Uint8Array[]). Without careful normalization, subtle bugs and type errors can occur, especially when transforms are chained or when mixing sync and async pipelines.

The approach

The PR introduces a normalization step for transform results between fused stateless transforms, ensuring that each transform receives a consistent batch format.

1. Add a normalization helper:

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

This function quickly checks if the result is already a batch, a single chunk, or a string/ArrayBuffer, and normalizes it to a non-empty Uint8Array[] or null for empty batches.

2. Apply normalization in the sync pipeline:

-      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;

For all but the last transform, normalize the result. If the fast path doesn't handle it, fall back to a slower, more general normalization (appendTransformResultSync). If the result is empty, set to null.

3. Apply normalization in the async pipeline:

-      if (isPromise(result)) {
-        const resolved = await result;
-        if (resolved === null) {
-          current = null;
-          break;
-        }
-        current = resolved;
-      } else {
-        current = result;
-      }
+      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;

Same logic as sync, but handles promises and async normalization.

4. Add tests to verify normalization:

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

These tests check that the intermediate output between transforms is always a batch (Array.isArray(chunks)), both for sync and async pipelines.

From the review thread

Question:

shouldn't we return result as default if none of the if cases match ?

Response:

The implicit undefined is intentional here. It indicates that the result was not handled by the fast path, so the caller falls back to appendTransformResultSync/Async to normalize iterables/generators and validate unsupported values.

Returning result by default would pass those values directly to the next fused transform without normalization.

Takeaways

• Always normalize intermediate data structures when chaining abstractions, so downstream code can rely on consistent input types.
• Use fast-path checks for common cases, but provide a general fallback for less common or complex cases.
• Test not just end-to-end output, but also the shape and type of intermediate values in pipelines, especially when refactoring or fusing logic.