stream: apply source normalization once at call time

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

doc/api/stream_iter.md

@@ -685,8 +685,10 @@ added: v24.20.0
   * `signal` {AbortSignal} Abort the pipeline.
 * Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array\[]}
 
-Create a lazy async pipeline. Data is not read from `source` until the
-returned iterable is consumed. Transforms are applied in order.
+Create a lazy async pipeline. Source conversion and streamable protocol
+dispatch occur when `pull()` is called, but data is not read from `source`
+until the returned iterable is consumed. A signal that is already aborted is
+thrown synchronously after source conversion. Transforms are applied in order.
 
 ```mjs
 import { from, pull, text } from 'node:stream/iter';
@@ -754,7 +756,8 @@ added: v24.20.0
 * `...transforms` {Function|Object} Zero or more sync transforms.
 * Returns: {Iterable} whose chunks return {Uint8Array\[]}
 
-Synchronous version of [`pull()`][]. All transforms must be synchronous.
+Synchronous version of [`pull()`][]. Source conversion and streamable protocol
+dispatch occur when `pullSync()` is called. All transforms must be synchronous.
 
 ## Push streams
 

lib/internal/streams/iter/pull.js

@@ -802,6 +802,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
  * @returns {Iterable<Uint8Array[]>}
  */
 function pullSync(source, ...transforms) {
+  const normalized = fromSync(source);
   for (let i = 0; i < transforms.length; i++) {
     if (!isTransform(transforms[i])) {
       throw new ERR_INVALID_ARG_TYPE(
@@ -812,7 +813,7 @@ function pullSync(source, ...transforms) {
   return {
     __proto__: null,
     *[SymbolIterator]() {
-      yield* createSyncPipeline(fromSync(source), transforms);
+      yield* createSyncPipeline(normalized, transforms);
     },
   };
 }
@@ -829,17 +830,9 @@ function pull(source, ...args) {
   const signal = options?.signal;
   if (signal !== undefined) {
     validateAbortSignal(signal, 'options.signal');
-    // Eagerly check abort at call time per spec
-    if (signal.aborted) {
-      return {
-        __proto__: null,
-        // eslint-disable-next-line require-yield
-        async *[SymbolAsyncIterator]() {
-          throw signal.reason;
-        },
-      };
-    }
   }
+  const normalized = from(source);
+  signal?.throwIfAborted();
 
   return {
     __proto__: null,
@@ -849,7 +842,7 @@ function pull(source, ...args) {
         controller.signal : AbortSignal.any([signal, controller.signal]);
 
       async function* pipeline() {
-        yield* createAsyncPipeline(from(source), transforms, iteratorSignal);
+        yield* createAsyncPipeline(normalized, transforms, iteratorSignal);
       }
       const iterator = pipeline();
 
@@ -884,9 +877,6 @@ function pullWithConsumerCleanup(source, transforms, signal) {
       return sourceIterator;
     },
   };
-  const pipeline = signal === undefined ?
-    pull(pipelineSource, ...transforms) :
-    pull(pipelineSource, ...transforms, { __proto__: null, signal });
   let sourceClosed = false;
   let abortHandler;
 
@@ -903,6 +893,21 @@ function pullWithConsumerCleanup(source, transforms, signal) {
     }
   }
 
+  if (signal?.aborted) {
+    closeSource('throw', signal.reason);
+    return {
+      __proto__: null,
+      // eslint-disable-next-line require-yield
+      async *[SymbolAsyncIterator]() {
+        throw signal.reason;
+      },
+    };
+  }
+
+  const pipeline = signal === undefined ?
+    pull(pipelineSource, ...transforms) :
+    pull(pipelineSource, ...transforms, { __proto__: null, signal });
+
   if (signal !== undefined) {
     abortHandler = () => closeSource('throw', signal.reason);
     signal.addEventListener('abort', abortHandler,

test/parallel/test-stream-iter-pull-async.js

@@ -11,6 +11,7 @@ const {
   share,
   tap,
   text,
+  toAsyncStreamable,
 } = require('stream/iter');
 
 async function testPullIdentity() {
@@ -53,18 +54,54 @@ async function testPullWithAbortSignal() {
     yield [new Uint8Array([1])];
   }
 
-  const result = pull(gen(), { signal: AbortSignal.abort() });
-  await assert.rejects(
-    async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const _ of result) {
-        assert.fail('Should not reach here');
-      }
-    },
+  assert.throws(
+    () => pull(gen(), { signal: AbortSignal.abort() }),
     { name: 'AbortError' },
   );
 }
 
+async function testPullNormalizesSourceAtCallTime() {
+  let protocolCalls = 0;
+  let iteratorCalls = 0;
+  const source = {
+    [toAsyncStreamable]() {
+      protocolCalls++;
+      return {
+        async *[Symbol.asyncIterator]() {
+          iteratorCalls++;
+          yield 'data';
+        },
+      };
+    },
+  };
+
+  const result = pull(source);
+  assert.strictEqual(protocolCalls, 1);
+  assert.strictEqual(iteratorCalls, 0);
+  assert.strictEqual(await text(result), 'data');
+  assert.strictEqual(protocolCalls, 1);
+  assert.strictEqual(iteratorCalls, 1);
+}
+
+function testPullPreAbortOrdering() {
+  const reason = new Error('already aborted');
+  let protocolCalls = 0;
+  const source = {
+    [toAsyncStreamable]() {
+      protocolCalls++;
+      return from('data');
+    },
+  };
+  const signal = AbortSignal.abort(reason);
+
+  assert.throws(() => pull(source, { signal }), (error) => error === reason);
+  assert.strictEqual(protocolCalls, 1);
+  assert.throws(
+    () => pull(null, { signal }),
+    { code: 'ERR_INVALID_ARG_TYPE' },
+  );
+}
+
 async function testPullChainedTransforms() {
   const enc = new TextEncoder();
   const transforms = [
@@ -475,6 +512,8 @@ async function testTransformOptionsNotShared() {
     testPullStatelessTransform(),
     testPullStatefulTransform(),
     testPullWithAbortSignal(),
+    testPullNormalizesSourceAtCallTime(),
+    testPullPreAbortOrdering(),
     testPullChainedTransforms(),
     testPullSourceError(),
     testTapCallbackError(),

test/parallel/test-stream-iter-pull-sync.js

@@ -3,14 +3,44 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { pullSync, fromSync, bytesSync, tapSync } = require('stream/iter');
+const {
+  pullSync,
+  fromSync,
+  bytesSync,
+  tapSync,
+  toStreamable,
+} = require('stream/iter');
 
 function testPullSyncIdentity() {
   // No transforms - just pass through
   const data = bytesSync(pullSync(fromSync('hello')));
   assert.deepStrictEqual(data, new TextEncoder().encode('hello'));
 }
 
+function testPullSyncNormalizesSourceAtCallTime() {
+  let protocolCalls = 0;
+  let iteratorCalls = 0;
+  const source = {
+    [toStreamable]() {
+      protocolCalls++;
+      return {
+        *[Symbol.iterator]() {
+          iteratorCalls++;
+          yield 'data';
+        },
+      };
+    },
+  };
+
+  const result = pullSync(source);
+  assert.strictEqual(protocolCalls, 1);
+  assert.strictEqual(iteratorCalls, 0);
+  assert.strictEqual(new TextDecoder().decode(bytesSync(result)), 'data');
+  assert.strictEqual(protocolCalls, 1);
+  assert.strictEqual(iteratorCalls, 1);
+  assert.throws(() => pullSync(null), { code: 'ERR_INVALID_ARG_TYPE' });
+}
+
 function testPullSyncStatelessTransform() {
   const upper = (chunks) => {
     if (chunks === null) return null;
@@ -177,6 +207,7 @@ function testPullSyncInvalidTransform() {
 
 Promise.all([
   testPullSyncIdentity(),
+  testPullSyncNormalizesSourceAtCallTime(),
   testPullSyncStatelessTransform(),
   testPullSyncStatefulTransform(),
   testPullSyncChainedTransforms(),