stream: ensure streamable protocols precede fast paths

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: https://github.com/nodejs/node/pull/66030
Reviewed-By: Filip Skokan <panva.ip@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

lib/internal/streams/iter/from.js

@@ -661,6 +661,13 @@ function fromSync(input) {
     };
   }
 
+  // Check toStreamable protocol (takes precedence over iteration protocols).
+  // toAsyncStreamable is ignored entirely in fromSync.
+  const streamableMethod = getProtocolMethod(input, toStreamable);
+  if (streamableMethod !== undefined) {
+    return fromSync(FunctionPrototypeCall(streamableMethod, input));
+  }
+
   // Fast path: Uint8Array[] - yield in bounded sub-batches.
   // Yielding the entire array as one batch forces downstream transforms
   // to process all data at once, causing peak memory proportional to total
@@ -696,13 +703,6 @@ function fromSync(input) {
     }
   }
 
-  // Check toStreamable protocol (takes precedence over iteration protocols).
-  // toAsyncStreamable is ignored entirely in fromSync.
-  const streamableMethod = getProtocolMethod(input, toStreamable);
-  if (streamableMethod !== undefined) {
-    return fromSync(FunctionPrototypeCall(streamableMethod, input));
-  }
-
   const isIterable = isSyncIterable(input);
 
   // Reject explicit async-only inputs
@@ -751,11 +751,6 @@ function from(input) {
     throw new ERR_INVALID_ARG_TYPE('input', 'a non-null value', input);
   }
 
-  // Fast path: validated source already yields valid Uint8Array[] batches
-  if (input[kValidatedSource]) {
-    return input;
-  }
-
   // Check for primitives first (ByteInput)
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
@@ -767,6 +762,34 @@ function from(input) {
     };
   }
 
+  // Check toAsyncStreamable protocol (takes precedence over toStreamable and
+  // iteration protocols)
+  const asyncStreamableMethod = getProtocolMethod(input, toAsyncStreamable);
+  if (asyncStreamableMethod !== undefined) {
+    let result = FunctionPrototypeCall(asyncStreamableMethod, input);
+    if (isPromise(result)) {
+      result = PromisePrototypeThen(result, undefined, undefined);
+      markPromiseAsHandled(result);
+    }
+    // Synchronous validated source (e.g. Readable batched iterator)
+    if (result?.[kValidatedSource]) {
+      return result;
+    }
+    return createNormalizationSource(
+      (context) => normalizeAsyncStreamableResult(result, context));
+  }
+
+  // Check toStreamable protocol (takes precedence over iteration protocols)
+  const streamableMethod = getProtocolMethod(input, toStreamable);
+  if (streamableMethod !== undefined) {
+    return from(FunctionPrototypeCall(streamableMethod, input));
+  }
+
+  // Fast path: validated source already yields valid Uint8Array[] batches
+  if (input[kValidatedSource]) {
+    return input;
+  }
+
   // Fast path: Uint8Array[] - yield in bounded sub-batches.
   // Yielding the entire array as one batch forces downstream transforms
   // to process all data at once, causing peak memory proportional to total
@@ -801,29 +824,6 @@ function from(input) {
     }
   }
 
-  // Check toAsyncStreamable protocol (takes precedence over toStreamable and
-  // iteration protocols)
-  const asyncStreamableMethod = getProtocolMethod(input, toAsyncStreamable);
-  if (asyncStreamableMethod !== undefined) {
-    let result = FunctionPrototypeCall(asyncStreamableMethod, input);
-    if (isPromise(result)) {
-      result = PromisePrototypeThen(result, undefined, undefined);
-      markPromiseAsHandled(result);
-    }
-    // Synchronous validated source (e.g. Readable batched iterator)
-    if (result?.[kValidatedSource]) {
-      return result;
-    }
-    return createNormalizationSource(
-      (context) => normalizeAsyncStreamableResult(result, context));
-  }
-
-  // Check toStreamable protocol (takes precedence over iteration protocols)
-  const streamableMethod = getProtocolMethod(input, toStreamable);
-  if (streamableMethod !== undefined) {
-    return from(FunctionPrototypeCall(streamableMethod, input));
-  }
-
   // Must be a Streamable (sync or async iterable)
   if (!isSyncIterable(input) && !isAsyncIterable(input)) {
     throw new ERR_INVALID_ARG_TYPE(

test/parallel/test-stream-iter-property-access.js

@@ -3,6 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
+const { Readable } = require('stream');
 const {
   Broadcast,
   Share,
@@ -21,13 +22,14 @@ const {
   shareProtocol,
   shareSync,
   shareSyncProtocol,
+  text,
+  textSync,
   toAsyncStreamable,
   toStreamable,
 } = require('stream/iter');
 
-function protocolFixture(symbol, result) {
+function protocolFixture(symbol, result, input = {}) {
   let accesses = 0;
-  const input = {};
   const method = common.mustCall(function() {
     assert.strictEqual(this, input);
     return result;
@@ -83,6 +85,38 @@ async function testFromSyncSnapshotsProtocolMethods() {
   assert.strictEqual(fixture.accesses, 1);
 }
 
+async function testArrayFastPathsHonorProtocols() {
+  const asyncInputs = [
+    protocolFixture(toAsyncStreamable, 'empty-async', []),
+    protocolFixture(toStreamable, 'batch-async', [new Uint8Array([0])]),
+  ];
+  for (const fixture of asyncInputs) {
+    assert.match(await text(from(fixture.input)), /-async$/);
+    assert.strictEqual(fixture.accesses, 1);
+  }
+
+  const syncInputs = [
+    protocolFixture(toStreamable, 'empty-sync', []),
+    protocolFixture(toStreamable, 'batch-sync', [new Uint8Array([0])]),
+  ];
+  for (const fixture of syncInputs) {
+    assert.match(textSync(fromSync(fixture.input)), /-sync$/);
+    assert.strictEqual(fixture.accesses, 1);
+  }
+}
+
+async function testValidatedSourceHonorsProtocol() {
+  const readable = Readable.from(['ignored']);
+  const validated = readable[toAsyncStreamable]();
+  assert.strictEqual(from(validated), validated);
+
+  const fixture = protocolFixture(
+    toAsyncStreamable, 'validated-protocol', validated);
+  assert.strictEqual(await text(from(fixture.input)), 'validated-protocol');
+  assert.strictEqual(fixture.accesses, 1);
+  readable.destroy();
+}
+
 async function testPullSnapshotsStatefulTransform() {
   const fixture = statefulTransformFixture();
   const controller = new AbortController();
@@ -136,6 +170,8 @@ async function testDrainableProtocolSnapshotMethod() {
 Promise.all([
   testFromSnapshotsProtocolMethods(),
   testFromSyncSnapshotsProtocolMethods(),
+  testArrayFastPathsHonorProtocols(),
+  testValidatedSourceHonorsProtocol(),
   testPullSnapshotsStatefulTransform(),
   testPullSyncSnapshotsStatefulTransform(),
   testMultiConsumerProtocolsSnapshotMethods(),