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.

Author: ronag

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));
+            }
+          });
+        }
       });
     }
 

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

@@ -484,6 +484,22 @@ async function tests() {
       assert.strictEqual(e, err);
     })()]);
   }
+
+  {
+    const r = new Readable({
+      read () {
+      },
+      destroy(err, callback) {
+        setTimeout(() => callback(new Error('asd')), 1);
+      }
+    });
+
+    r.destroy();
+    const it = r[Symbol.asyncIterator]();
+    it.next().catch(common.mustCall((err) => {
+      assert.strictEqual(err.message, 'asd');
+    }));
+  }
 }
 
 {