stream: reject pull() reads on abort

trivikr · web-flow · commit 822ef3afb64c · 2026-06-07T19:16:05.000Z
Make pull() race pending source reads against the provided AbortSignal so aborting can reject a pending next() even when the source is waiting before yielding data. Fixes: #63497 Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63498 Fixes: #63497 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js
@@ -14,12 +14,19 @@ const {
   ArrayPrototypeSlice,
   PromisePrototypeThen,
   PromiseResolve,
+  PromiseWithResolvers,
+  SafePromisePrototypeFinally,
+  SafePromiseRace,
   SymbolAsyncIterator,
   SymbolIterator,
   TypedArrayPrototypeGetByteLength,
   Uint8Array,
 } = primordials;
 
+const {
+  markPromiseAsHandled,
+} = internalBinding('util');
+
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
@@ -685,6 +692,81 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
   options.signal?.throwIfAborted();
 }
 
+function getOnAbort(reject, signal) {
+  return () => reject(signal.reason);
+}
+
+/**
+ * Read one item from an async iterator, rejecting early if the signal aborts.
+ * @param {AsyncIterator} iterator - The iterator to read from.
+ * @param {AbortSignal|undefined} signal - Optional abort signal.
+ * @returns {Promise<IteratorResult<Uint8Array[]>>|IteratorResult<Uint8Array[]>}
+ */
+function abortableNext(iterator, signal) {
+  if (signal === undefined) {
+    return iterator.next();
+  }
+
+  signal.throwIfAborted();
+
+  const next = iterator.next();
+  const { promise, reject } = PromiseWithResolvers();
+  const onAbort = getOnAbort(reject, signal);
+  signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
+  if (signal.aborted) {
+    onAbort();
+  }
+
+  return SafePromisePrototypeFinally(SafePromiseRace([next, promise]), () => {
+    signal.removeEventListener('abort', onAbort);
+  });
+}
+
+/**
+ * Wrap an async source so each pending read is abort-aware.
+ * @param {AsyncIterable<Uint8Array[]>} source - The source to read from.
+ * @param {AbortSignal|undefined} signal - Optional abort signal.
+ * @returns {AsyncIterable<Uint8Array[]>}
+ */
+function yieldAbortable(source, signal) {
+  if (signal === undefined) {
+    return source;
+  }
+
+  return {
+    __proto__: null,
+    async *[SymbolAsyncIterator]() {
+      const iterator = source[SymbolAsyncIterator]();
+      let completed = false;
+      let aborted = false;
+
+      try {
+        while (true) {
+          const { done, value } = await abortableNext(iterator, signal);
+          if (done) {
+            completed = true;
+            return;
+          }
+          signal.throwIfAborted();
+          yield value;
+        }
+      } catch (error) {
+        aborted = signal.aborted;
+        throw error;
+      } finally {
+        if (!completed && typeof iterator.return === 'function') {
+          const result = iterator.return();
+          if (aborted) {
+            markPromiseAsHandled(result);
+          } else {
+            await result;
+          }
+        }
+      }
+    },
+  };
+}
+
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}
@@ -693,17 +775,14 @@ async function* createAsyncPipeline(source, transforms, signal) {
   // Check for abort
   signal?.throwIfAborted();
 
-  const normalized = source;
-
   // Fast path: no transforms, just yield normalized source directly
   if (transforms.length === 0) {
-    for await (const batch of normalized) {
-      signal?.throwIfAborted();
-      yield batch;
-    }
+    yield* yieldAbortable(source, signal);
     return;
   }
 
+  const normalized = yieldAbortable(source, signal);
+
   // Create internal controller for transform cancellation.
   // Note: if signal was already aborted, we threw above - no need to check here.
   const controller = new AbortController();
diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js
@@ -153,6 +153,44 @@ async function testPullSignalAbortMidIteration() {
   await assert.rejects(() => iter.next(), { name: 'AbortError' });
 }
 
+async function testPullSignalAbortWhileSourceNextPending() {
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next() {
+          await new Promise(() => {});
+        },
+      };
+    },
+  };
+  const ac = new AbortController();
+  const iter = pull(source, { signal: ac.signal })[Symbol.asyncIterator]();
+  const next = iter.next();
+  ac.abort();
+  await assert.rejects(next, { name: 'AbortError' });
+}
+
+async function testPullSignalAbortWithTransformWhileSourceNextPending() {
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next() {
+          await new Promise(() => {});
+        },
+      };
+    },
+  };
+  const ac = new AbortController();
+  const iter = pull(
+    source,
+    (chunks) => chunks,
+    { signal: ac.signal },
+  )[Symbol.asyncIterator]();
+  const next = iter.next();
+  ac.abort();
+  await assert.rejects(next, { name: 'AbortError' });
+}
+
 // Pull consumer break (return()) cleans up transform signal
 async function testPullConsumerBreakCleanup() {
   let signalAborted = false;
@@ -361,6 +399,8 @@ async function testTransformOptionsNotShared() {
     testPullSourceError(),
     testTapCallbackError(),
     testPullSignalAbortMidIteration(),
+    testPullSignalAbortWhileSourceNextPending(),
+    testPullSignalAbortWithTransformWhileSourceNextPending(),
     testPullConsumerBreakCleanup(),
     testPullTransformReturnsPromise(),
     testPullTransformYieldsStrings(),
