stream: fix async iterator destroyed error propagation

There was an edge case where if _destroy calls the error callback later than one tick the iterator would complete early and not propgate the error. PR-URL: #31314 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Ruben Bridgewater <[email protected]> Reviewed-By: Rich Trott <[email protected]> Reviewed-By: Minwoo Jung <[email protected]> Reviewed-By: James M Snell <[email protected]>
nodejs · Jan 14, 2020 · d15b8ea · d15b8ea
1 parent eb5e1dd
commit d15b8ea
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 11 deletions.
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
@@ -78,18 +78,20 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({
     }
 
     if (this[kStream].destroyed) {
-      // We need to defer via nextTick because if .destroy(err) is
-      // called, the error will be emitted via nextTick, and
-      // we cannot guarantee that there is no error lingering around
-      // waiting to be emitted.
       return new Promise((resolve, reject) => {
-        process.nextTick(() => {
-          if (this[kError]) {
-            reject(this[kError]);
-          } else {
-            resolve(createIterResult(undefined, true));
-          }
-        });
+        if (this[kError]) {
+          reject(this[kError]);
+        } else if (this[kEnded]) {
+          resolve(createIterResult(undefined, true));
+        } else {
+          finished(this[kStream], (err) => {
+            if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
+              reject(err);
+            } else {
+              resolve(createIterResult(undefined, true));
+            }
+          });
+        }
       });
     }
 

diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
@@ -484,6 +484,23 @@ async function tests() {
       assert.strictEqual(e, err);
     })()]);
   }
+
+  {
+    const _err = new Error('asd');
+    const r = new Readable({
+      read() {
+      },
+      destroy(err, callback) {
+        setTimeout(() => callback(_err), 1);
+      }
+    });
+
+    r.destroy();
+    const it = r[Symbol.asyncIterator]();
+    it.next().catch(common.mustCall((err) => {
+      assert.strictEqual(err, _err);
+    }));
+  }
 }
 
 {