fixup: address review comments

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/streams/readable.js

@@ -23,6 +23,8 @@
 
 const {
   ArrayPrototypeIndexOf,
+  AsyncIteratorPrototype,
+  FunctionPrototypeCall,
   NumberIsInteger,
   NumberIsNaN,
   NumberParseInt,
@@ -103,6 +105,7 @@ const FastBuffer = Buffer[SymbolSpecies];
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const FixedQueue = require('internal/fixed_queue');
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -1394,6 +1397,7 @@ function createAsyncIterator(stream, options) {
   let completed = false;
   let inFlight = false; // An asynchronous request is outstanding
   let queue = null;     // Requests received while inFlight
+  let draining = false;
   let cleanup;
 
   // Used both as the 'readable' listener (where `this === stream`) and
@@ -1450,15 +1454,25 @@ function createAsyncIterator(stream, options) {
   }
 
   function drain() {
-    while (!inFlight && queue.length > 0) {
-      const req = queue.shift();
-      if (req.type === 'next') {
-        processNext(req.resolve, req.reject);
-      } else if (req.type === 'return') {
-        processReturn(req.value, req.resolve);
-      } else {
-        processThrow(req.value, req.reject);
+    // 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()) {
+        const req = queue.shift();
+        if (req.type === 'next') {
+          processNext(req.resolve, req.reject);
+        } else if (req.type === 'return') {
+          processReturn(req.value, req.resolve);
+        } else {
+          processThrow(req.value, req.reject);
+        }
       }
+    } finally {
+      draining = false;
     }
   }
 
@@ -1483,8 +1497,11 @@ function createAsyncIterator(stream, options) {
   function pump(resolve, reject) {
     const chunk = stream.destroyed ? null : stream.read();
     if (chunk !== null) {
-      if (typeof chunk.then === 'function') {
-        PromisePrototypeThen(PromiseResolve(chunk), (value) => {
+      // Read `then` only once so that a getter cannot observe (or throw
+      // on) a second access.
+      const then = chunk.then;
+      if (typeof then === 'function') {
+        FunctionPrototypeCall(then, chunk, (value) => {
           inFlight = false;
           resolve({ done: false, value });
           if (queue !== null) drain();
@@ -1546,16 +1563,20 @@ function createAsyncIterator(stream, options) {
   }
 
   return {
+    __proto__: AsyncIteratorPrototype,
     next() {
       if (!inFlight && !completed) {
         if (!started) start();
         // Fast path: a chunk is already buffered.
         const chunk = stream.destroyed ? null : stream.read();
         if (chunk !== null) {
-          if (typeof chunk.then === 'function') {
+          // Read `then` only once so that a getter cannot observe (or
+          // throw on) a second access.
+          const then = chunk.then;
+          if (typeof then === 'function') {
             inFlight = true;
-            return PromisePrototypeThen(
-              PromiseResolve(chunk), onChunkFulfilled, onChunkRejected);
+            return FunctionPrototypeCall(
+              then, chunk, onChunkFulfilled, onChunkRejected);
           }
           return PromiseResolve({ done: false, value: chunk });
         }
@@ -1575,8 +1596,8 @@ function createAsyncIterator(stream, options) {
       }
       return new Promise((resolve, reject) => {
         if (inFlight) {
-          queue ??= [];
-          queue.push({ type: 'next', value: undefined, resolve, reject });
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'next', value: undefined, resolve, reject });
         } else {
           resolve({ done: true, value: undefined });
         }
@@ -1585,8 +1606,8 @@ function createAsyncIterator(stream, options) {
     return(value) {
       return new Promise((resolve, reject) => {
         if (inFlight) {
-          queue ??= [];
-          queue.push({ type: 'return', value, resolve, reject });
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'return', value, resolve, reject });
         } else {
           processReturn(value, resolve);
         }
@@ -1595,8 +1616,8 @@ function createAsyncIterator(stream, options) {
     throw(err) {
       return new Promise((resolve, reject) => {
         if (inFlight) {
-          queue ??= [];
-          queue.push({ type: 'throw', value: err, resolve, reject });
+          queue ??= new FixedQueue();
+          queue.push({ __proto__: null, type: 'throw', value: err, resolve, reject });
         } else {
           processThrow(err, reject);
         }

test/parallel/test-stream-flatMap.js

@@ -81,13 +81,13 @@ function oneTo5() {
       // Both mappers allowed by `concurrency` are now in flight.
       ac.abort();
     }
-    await new Promise((resolve, reject) => {
-      if (signal.aborted) {
-        reject(signal.reason);
-        return;
-      }
-      signal.addEventListener('abort', () => reject(signal.reason), { once: true });
-    });
+    const { promise, reject } = Promise.withResolvers();
+    if (signal.aborted) {
+      reject(signal.reason);
+    }
+    signal.addEventListener('abort', () => reject(signal.reason), { once: true });
+    // Promise is expected to reject.
+    await promise;
   }, 2), { signal: ac.signal, concurrency: 2 });
   // pump
   assert.rejects(async () => {

test/parallel/test-stream-readable-async-iterators.js

@@ -963,5 +963,31 @@ async function tests() {
   })().then(common.mustCall());
 }
 
+{
+  // Draining a large number of queued next() calls must not overflow
+  // the call stack.
+  // Refs: https://github.com/nodejs/node/pull/64447#discussion_r3566240419
+  (async () => {
+    const count = 20_000;
+    const r = new Readable({ objectMode: true, read() {} });
+    const it = r[Symbol.asyncIterator]();
+
+    const requests = [];
+    for (let i = 0; i < count; i++) {
+      requests.push(it.next());
+    }
+    for (let i = 0; i < count; i++) {
+      r.push(i);
+    }
+    r.push(null);
+
+    const results = await Promise.all(requests);
+    for (let i = 0; i < count; i++) {
+      assert.deepStrictEqual(results[i], { done: false, value: i });
+    }
+    assert.strictEqual((await it.next()).done, true);
+  })().then(common.mustCall());
+}
+
 // To avoid missing some tests if a promise does not resolve
 tests().then(common.mustCall());