stream: reject pull() reads on abort

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

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(),

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

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(),

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

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(),

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

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(),

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

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(),

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

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(),

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

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(),

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

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(),