stream: fix fromAsyncGen

Fixes: #40497

Author: ronag

lib/internal/streams/duplexify.js

@@ -210,21 +210,26 @@ function fromAsyncGen(fn) {
   const value = fn(async function*() {
     while (true) {
       const { chunk, done, cb } = await promise;
+      promise = null;
       process.nextTick(cb);
       if (done) return;
       if (signal.aborted) throw new AbortError();
-      yield chunk;
       ({ promise, resolve } = createDeferredPromise());
+      yield chunk;
     }
   }(), { signal });
 
   return {
     value,
     write(chunk, encoding, cb) {
-      resolve({ chunk, done: false, cb });
+      const _resolve = resolve;
+      resolve = null;
+      _resolve({ chunk, done: false, cb });
     },
     final(cb) {
-      resolve({ done: true, cb });
+      const _resolve = resolve;
+      resolve = null;
+      _resolve({ done: true, cb });
     },
     destroy(err, cb) {
       ac.abort();

test/parallel/test-stream-duplex-from.js

@@ -2,7 +2,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { Duplex, Readable, Writable } = require('stream');
+const { Duplex, Readable, Writable, pipeline } = require('stream');
 
 {
   const d = Duplex.from({
@@ -118,3 +118,29 @@ const { Duplex, Readable, Writable } = require('stream');
     assert.strictEqual(d.readable, false);
   }));
 }
+
+{
+  // https://github.com/nodejs/node/issues/40497
+  pipeline(
+    ['abc\ndef\nghi'],
+    Duplex.from(async function * (source) {
+      let rest = '';
+      for await (const chunk of source) {
+        const lines = (rest + chunk.toString()).split('\n');
+        rest = lines.pop();
+        for (const line of lines) {
+          yield line;
+        }
+      }
+      yield rest;
+    }),
+    async function * (source) {
+      let ret = '';
+      for await (const x of source) {
+        ret += x;
+      }
+      assert.strictEqual(ret, 'abcdefghi');
+    },
+    common.mustCall(() => {}),
+  );
+}