stream: speed up async iteration of Readable

Replace the async generator backing Symbol.asyncIterator with a
hand-rolled iterator. The generator machinery costs several extra
promise allocations and microtask hops per chunk: yield awaits the
yielded value and resolves the pending request through separate
promises. Buffered chunks are now delivered as an already-resolved
promise, one microtask sooner than before.
Thenable chunks are still awaited before delivery, requests received
while a next() is outstanding are queued, and return()/throw() before
the first next() complete the iterator without touching the stream.
The earlier delivery is observable by code racing an abort against
the first chunk. The flatMap AbortSignal test relied on such a race;
it is reworked to abort deterministically while two mappers are in
flight, asserting the concurrency limit, in-flight cancellation and
rejection, without depending on delivery timing or timers.
streams/readable-async-iterator.js sync='yes': +32.59% (***)
streams/readable-async-iterator.js sync='no': +9.84% (***)
Assisted-by: Claude Fable 5
Signed-off-by: Matteo Collina <matteo.collina@gmail.com>
PR-URL: #64447
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>

Author: mcollina

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {

lib/internal/streams/readable.js

@@ -23,13 +23,18 @@
 
 const{
  ArrayPrototypeIndexOf,
+ AsyncIteratorPrototype,
+ FunctionPrototypeCall,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperties,
  ObjectKeys,
  ObjectSetPrototypeOf,
  Promise,
+ PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  ReflectApply,
  SafeSet,
  Symbol,
@@ -100,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const{ StringDecoder }=require('string_decoder');
 constfrom=require('internal/streams/from');
+constFixedQueue=require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
@@ -1386,10 +1392,22 @@ function streamToAsyncIterator(stream, options) {
 returniter;
 }
 
-asyncfunction*createAsyncIterator(stream,options){
+// Async iterator over a Readable. Requests received while another is
+// outstanding are queued and processed in order.
+functioncreateAsyncIterator(stream,options){
 letcallback=nop;
-
-functionnext(resolve){
+leterror;// undefined: active, null: ended cleanly, else: Error
+letstarted=false;
+letcompleted=false;
+letinFlight=false;// An asynchronous request is outstanding
+letqueue=null;// Requests received while inFlight
+letdraining=false;
+letcleanup;
+
+// Used both as the 'readable' listener (where `this === stream`) and
+// as a promise executor storing the resolver that wakes up a pending
+// pump().
+functionwakeup(resolve){
 if(this===stream){
 callback();
 callback=nop;
@@ -1398,32 +1416,23 @@ async function* createAsyncIterator(stream, options) {
 }
 }
 
-stream.on('readable',next);
+functionstart(){
+started=true;
 
-leterror;
-constcleanup=eos(stream,{writable: false},(err)=>{
-error=err ? aggregateTwoErrors(error,err) : null;
-callback();
-callback=nop;
-});
+stream.on('readable',wakeup);
+
+cleanup=eos(stream,{writable: false},(err)=>{
+error=err ? aggregateTwoErrors(error,err) : null;
+callback();
+callback=nop;
+});
+}
+
+// Complete the iterator and either destroy the stream or detach
+// from it.
+functionfinalize(){
+completed=true;
 
-try{
-while(true){
-constchunk=stream.destroyed ? null : stream.read();
-if(chunk!==null){
-yieldchunk;
-}elseif(error){
-throwerror;
-}elseif(error===null){
-return;
-}else{
-awaitnewPromise(next);
-}
-}
-}catch(err){
-error=aggregateTwoErrors(error,err);
-throwerror;
-}finally{
 constpreserveHalfOpenDuplex=
 error===null&&
 stream.allowHalfOpen===true&&
@@ -1437,10 +1446,188 @@ async function* createAsyncIterator(stream, options) {
 ){
 destroyImpl.destroyer(stream,null);
 }else{
-stream.off('readable',next);
+stream.off('readable',wakeup);
 cleanup();
 }
 }
+
+functionsettleError(err,reject){
+error=aggregateTwoErrors(error,err);
+finalize();
+reject(error);
+}
+
+functiondrain(){
+// Requests settled synchronously call back into drain(); the guard
+// keeps a single loop going instead of recursing once per request.
+if(draining){
+return;
+}
+draining=true;
+try{
+while(!inFlight&&!queue.isEmpty()){
+constreq=queue.shift();
+if(req.type==='next'){
+processNext(req.resolve,req.reject);
+}elseif(req.type==='return'){
+processReturn(req.value,req.resolve);
+}else{
+processThrow(req.value,req.reject);
+}
+}
+}finally{
+draining=false;
+}
+}
+
+// Thenable chunks are unwrapped before delivery; a rejection tears
+// down the iterator and the stream.
+functiononChunkFulfilled(value){
+inFlight=false;
+if(queue!==null)drain();
+return{done: false, value };
+}
+
+functiononChunkRejected(err){
+inFlight=false;
+error=aggregateTwoErrors(error,err);
+finalize();
+if(queue!==null)drain();
+throwerror;
+}
+
+// Runs with inFlight === true; settles the request and hands over to
+// any requests that queued up behind it.
+functionpump(resolve,reject){
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or throw
+// on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+FunctionPrototypeCall(then,chunk,(value)=>{
+inFlight=false;
+resolve({done: false, value });
+if(queue!==null)drain();
+},(err)=>{
+inFlight=false;
+settleError(err,reject);
+if(queue!==null)drain();
+});
+return;
+}
+inFlight=false;
+resolve({done: false,value: chunk});
+if(queue!==null)drain();
+}elseif(error){
+inFlight=false;
+settleError(error,reject);
+if(queue!==null)drain();
+}elseif(error===null){
+inFlight=false;
+finalize();
+resolve({done: true,value: undefined});
+if(queue!==null)drain();
+}else{
+// No data buffered yet; wait for 'readable' or end-of-stream and
+// retry.
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+}
+}
+
+functionprocessNext(resolve,reject){
+if(completed){
+resolve({done: true,value: undefined});
+return;
+}
+if(!started)start();
+inFlight=true;
+pump(resolve,reject);
+}
+
+functionprocessReturn(value,resolve){
+if(!completed){
+if(started){
+finalize();
+}else{
+// Never started: complete without touching the stream.
+completed=true;
+}
+}
+resolve({done: true, value });
+}
+
+functionprocessThrow(err,reject){
+if(completed||!started){
+completed=true;
+reject(err);
+return;
+}
+settleError(err,reject);
+}
+
+return{
+__proto__: AsyncIteratorPrototype,
+next(){
+if(!inFlight&&!completed){
+if(!started)start();
+// Fast path: a chunk is already buffered.
+constchunk=stream.destroyed ? null : stream.read();
+if(chunk!==null){
+// Read `then` only once so that a getter cannot observe (or
+// throw on) a second access.
+constthen=chunk.then;
+if(typeofthen==='function'){
+inFlight=true;
+returnFunctionPrototypeCall(
+then,chunk,onChunkFulfilled,onChunkRejected);
+}
+returnPromiseResolve({done: false,value: chunk});
+}
+if(error){
+finalize();
+returnPromiseReject(error);
+}
+if(error===null){
+finalize();
+returnPromiseResolve({done: true,value: undefined});
+}
+// No data buffered yet; wait for 'readable' or end-of-stream.
+inFlight=true;
+returnnewPromise((resolve,reject)=>{
+PromisePrototypeThen(newPromise(wakeup),()=>pump(resolve,reject));
+});
+}
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'next',value: undefined, resolve, reject });
+}else{
+resolve({done: true,value: undefined});
+}
+});
+},
+return(value){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'return', value, resolve, reject });
+}else{
+processReturn(value,resolve);
+}
+});
+},
+throw(err){
+returnnewPromise((resolve,reject)=>{
+if(inFlight){
+queue??=newFixedQueue();
+queue.push({__proto__: null,type: 'throw',value: err, resolve, reject });
+}else{
+processThrow(err,reject);
+}
+});
+},
+};
 }
 
 letcomposeImpl;

test/parallel/test-stream-flatMap.js

@@ -72,10 +72,23 @@ function oneTo5() {
 
 {
 // Concurrency + AbortSignal
+// Two mappers are started concurrently and block until their signal
+// is aborted. Aborting while both are in flight must cancel them and
+// reject the iteration, without ever starting a third mapper.
 constac=newAbortController();
-conststream=oneTo5().flatMap(common.mustNotCall(async(_,{ signal })=>{
-awaitsetTimeout(100,{ signal });
-}),{signal: ac.signal,concurrency: 2});
+conststream=oneTo5().flatMap(common.mustCall(async(x,{ signal })=>{
+if(x===2){
+// Both mappers allowed by `concurrency` are now in flight.
+ac.abort();
+}
+const{ promise, reject }=Promise.withResolvers();
+if(signal.aborted){
+reject(signal.reason);
+}
+signal.addEventListener('abort',()=>reject(signal.reason),{once: true});
+// Promise is expected to reject.
+awaitpromise;
+},2),{signal: ac.signal,concurrency: 2});
 // pump
 assert.rejects(async()=>{
 forawait(constitemofstream){
@@ -85,10 +98,6 @@ function oneTo5() {
 },{
 name: 'AbortError',
 }).then(common.mustCall());
-
-queueMicrotask(()=>{
-ac.abort();
-});
 }
 
 {