stream: reject iter consumers on abort

Make stream/iter async consumers observe abort signals while waiting
for a pending async iterator read. This lets bytes(), text(),
arrayBuffer(), and array() reject promptly with the abort reason
instead of waiting for another batch.
Move the shared abort-aware iterator wrapper to stream/iter utils so
pull and consumers use the same helper.
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #64066
Backport-PR-URL: #64675Fixes: #64065
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),

lib/internal/streams/iter/consumers.js

@@ -52,6 +52,7 @@ const {
 
 const{
  concatBytes,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -125,7 +126,9 @@ async function collectAsync(source, signal, limit) {
 signal?.throwIfAborted();
 
 // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
-constnormalized=from(source);
+constabortableSource=signal&&isAsyncIterable(source) ?
+yieldAbortable(source,signal) : source;
+constnormalized=from(abortableSource);
 constchunks=[];
 
 // Fast path: no signal and no limit
@@ -140,8 +143,9 @@ async function collectAsync(source, signal, limit) {
 
 // Slow path: with signal or limit checks
 lettotalBytes=0;
+constiterable=signal ? yieldAbortable(normalized,signal) : normalized;
 
-forawait(constbatchofnormalized){
+forawait(constbatchofiterable){
 signal?.throwIfAborted();
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];

lib/internal/streams/iter/pull.js

@@ -14,19 +14,12 @@ const {
  ArrayPrototypeSlice,
  PromisePrototypeThen,
  PromiseResolve,
- PromiseWithResolvers,
- SafePromisePrototypeFinally,
- SafePromiseRace,
  SymbolAsyncIterator,
  SymbolIterator,
  TypedArrayPrototypeGetByteLength,
  Uint8Array,
 }=primordials;
 
-const{
- markPromiseAsHandled,
-}=internalBinding('util');
-
 const{
 codes: {
 ERR_INVALID_ARG_TYPE,
@@ -60,6 +53,7 @@ const {
  parsePullArgs,
  toUint8Array,
  wrapError,
+ yieldAbortable,
 }=require('internal/streams/iter/utils');
 
 const{
@@ -690,81 +684,6 @@ async function* applyValidatedStatefulAsyncTransform(source, transform, options)
 options.signal?.throwIfAborted();
 }
 
-functiongetOnAbort(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[]>}
- */
-functionabortableNext(iterator,signal){
-if(signal===undefined){
-returniterator.next();
-}
-
-signal.throwIfAborted();
-
-constnext=iterator.next();
-const{ promise, reject }=PromiseWithResolvers();
-constonAbort=getOnAbort(reject,signal);
-signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
-if(signal.aborted){
-onAbort();
-}
-
-returnSafePromisePrototypeFinally(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[]>}
- */
-functionyieldAbortable(source,signal){
-if(signal===undefined){
-returnsource;
-}
-
-return{
-__proto__: null,
-async*[SymbolAsyncIterator](){
-constiterator=source[SymbolAsyncIterator]();
-letcompleted=false;
-letaborted=false;
-
-try{
-while(true){
-const{ done, value }=awaitabortableNext(iterator,signal);
-if(done){
-completed=true;
-return;
-}
-signal.throwIfAborted();
-yieldvalue;
-}
-}catch(error){
-aborted=signal.aborted;
-throwerror;
-}finally{
-if(!completed&&typeofiterator.return==='function'){
-constresult=iterator.return();
-if(aborted){
-markPromiseAsHandled(result);
-}else{
-awaitresult;
-}
-}
-}
-},
-};
-}
-
 /**
  * Create an async pipeline from source through transforms.
  * @yields {Uint8Array[]}

lib/internal/streams/iter/utils.js

@@ -8,14 +8,22 @@ const {
  MathMin,
  NumberMAX_SAFE_INTEGER,
  PromiseResolve,
+ PromiseWithResolvers,
+ SafePromisePrototypeFinally,
+ SafePromiseRace,
  String,
+ SymbolAsyncIterator,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
+const{
+ markPromiseAsHandled,
+}=internalBinding('util');
+
 const{ TextEncoder }=require('internal/encoding');
 const{
 codes: {
@@ -69,6 +77,81 @@ function onSignalAbort(signal, handler) {
 }
 }
 
+functiongetOnAbort(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[]>}
+ */
+functionabortableNext(iterator,signal){
+if(signal===undefined){
+returniterator.next();
+}
+
+signal.throwIfAborted();
+
+constnext=iterator.next();
+const{ promise, reject }=PromiseWithResolvers();
+constonAbort=getOnAbort(reject,signal);
+signal.addEventListener('abort',onAbort,{__proto__: null,once: true});
+if(signal.aborted){
+onAbort();
+}
+
+returnSafePromisePrototypeFinally(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[]>}
+ */
+functionyieldAbortable(source,signal){
+if(signal===undefined){
+returnsource;
+}
+
+return{
+__proto__: null,
+async*[SymbolAsyncIterator](){
+constiterator=source[SymbolAsyncIterator]();
+letcompleted=false;
+letaborted=false;
+
+try{
+while(true){
+const{ done, value }=awaitabortableNext(iterator,signal);
+if(done){
+completed=true;
+return;
+}
+signal.throwIfAborted();
+yieldvalue;
+}
+}catch(error){
+aborted=signal.aborted;
+throwerror;
+}finally{
+if(!completed&&typeofiterator.return==='function'){
+constresult=iterator.return();
+if(aborted){
+markPromiseAsHandled(result);
+}else{
+awaitresult;
+}
+}
+}
+},
+};
+}
+
 /**
  * Compute the minimum cursor across a set of consumers and count how many
  * consumers are at that cursor.
@@ -301,4 +384,5 @@ module.exports = {
  toUint8Array,
  validateBackpressure,
  wrapError,
+ yieldAbortable,
 };

test/parallel/test-stream-iter-consumers-bytes.js

@@ -14,6 +14,7 @@ const {
  arrayBufferSync,
  array,
  arraySync,
+ toAsyncStreamable,
 }=require('stream/iter');
 
 // =============================================================================
@@ -53,6 +54,55 @@ async function testBytesAsyncAbort() {
 );
 }
 
+asyncfunctiontestAsyncConsumersAbortPendingNext(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} boom`);
+
+asyncfunction*never(){
+awaitnewPromise(()=>{});
+yield[];
+}
+
+constpromise=consumer(never(),{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
+asyncfunctiontestAsyncConsumersAbortPendingNormalization(){
+constconsumers=[
+['bytes',bytes],
+['text',text],
+['arrayBuffer',arrayBuffer],
+['array',array],
+];
+
+for(const[name,consumer]ofconsumers){
+constac=newAbortController();
+constreason=newError(`${name} normalization boom`);
+constsource={
+__proto__: null,
+[toAsyncStreamable](){
+returnnewPromise(()=>{});
+},
+};
+
+constpromise=consumer(source,{__proto__: null,signal: ac.signal});
+ac.abort(reason);
+
+awaitassert.rejects(promise,reason);
+}
+}
+
 asyncfunctiontestBytesEmpty(){
 constdata=awaitbytes(from([]));
 assert.ok(datainstanceofUint8Array);
@@ -205,6 +255,8 @@ Promise.all([
 testBytesAsync(),
 testBytesAsyncLimit(),
 testBytesAsyncAbort(),
+testAsyncConsumersAbortPendingNext(),
+testAsyncConsumersAbortPendingNormalization(),
 testBytesEmpty(),
 testArrayBufferSyncBasic(),
 testArrayBufferAsync(),