How I Traced a Node.js Streams Bug in Duplex.from()

A few days ago, I made my first contribution to Node.js.

The change itself was small. The part that took time was figuring out why a stream could stay alive even after the async function behind it had already returned.

The issue was reported in Node.js as #55077, and I eventually fixed it in PR #65963.

This is how I understood the bug and ended up fixing it.

The problem

The issue involved code like this:

const readable = Readable.from(['foo', 'bar', 'baz']);

pipeline(
  readable,
  Duplex.from(async function () {
    // Don't consume the input
  }),
  callback
);

At first glance, there doesn’t seem to be much going on here.

Duplex.from() receives an async function. The function returns. You would expect the stream pipeline to clean itself up.

That wasn’t happening.

The upstream Readable wasn’t being destroyed, and in the problematic cases the pipeline() callback could never run.

So the problem wasn’t that the async function failed.

It actually succeeded.

That was what made the bug interesting.

---

Following Duplex.from()

I started tracing the implementation instead of trying to reason about Duplex.from() as a black box.

Duplex.from() itself is basically a wrapper around duplexify():

Duplex.from = function(body) {
  duplexify ??= require('internal/streams/duplexify');
  return duplexify(body, 'body');
};

From there, things get more interesting.

When the argument is a function, duplexify() uses fromAsyncGen().

That helper creates an internal async generator and uses promises as a sort of handshake between the stream and the generator.

The important part looks roughly like this:

function fromAsyncGen(fn) {
  let { promise, resolve } = PromiseWithResolvers();

  const value = fn(async function*() {
    while (true) {
      const _promise = promise;
      promise = null;

      const { chunk, done, cb } = await _promise;

      process.nextTick(cb);

      if (done)
        return;

      yield chunk;
    }
  }());

  // ...
}

The exact implementation has more involved handling for aborts and cleanup, but this was the part I needed to understand.

The function passed to Duplex.from() is given an async iterable.

If the function consumes that iterable, the handshake progresses normally.

But what happens if it doesn’t?

That’s where the problem starts.

---

There are actually two different paths

One thing that initially made this harder to reason about was that Duplex.from(function) doesn’t necessarily mean one single execution path.

After fromAsyncGen() calls the function, Node looks at what the function returned.

An async generator returns an async iterable:

async function* () {
  // ...
}

An ordinary async function returns a Promise:

async function () {
  // ...
}

Those two cases go through different paths.

The async-generator case becomes a transform-like stream.

The AsyncFunction case becomes a writable-only Duplexify.

That distinction turned out to be important.

The bug in #55077 was in the AsyncFunction / Promise path.

---

What happens when the async function returns early?

Consider this:

Duplex.from(async function () {
  return;
});

The async function returns a fulfilled Promise.

But the stream may still have input waiting to be consumed.

The internal machinery is expecting the async iterable to be consumed. The write() side resolves part of that handshake, and the internal async generator is supposed to continue the process.

If the user function has already returned without consuming the iterable, that consumer is no longer doing the work the stream expects.

The result is a stalled handshake.

In the problematic case, the write callback doesn’t get to complete, so the writable side doesn’t progress normally.

That means the stream doesn’t reach the normal finalization path.

And because the destination is still waiting, pipeline() has no reason to consider the upstream stream finished.

So you end up with something roughly like this:

Readable
   |
   v
Duplex.from(async function)
   |
   v
AsyncFunction returns
   |
   v
Internal handshake is still waiting
   |
   v
Writable doesn't finish
   |
   v
Pipeline keeps waiting
   |
   v
Readable isn't destroyed

That was the behavior I needed to fix.

---

The part that confused me at first: final()

The AsyncFunction path has a final() implementation.

Before my change, the important structure was essentially:

let d;

const promise = FunctionPrototypeCall(
  then,
  value,
  (val) => {
    // successful completion
  },
  (err) => {
    destroyer(d, err);
  }
);

return d = new Duplexify({
  objectMode: true,
  readable: false,

  write,

  final(cb) {
    final(async () => {
      try {
        await promise;
        process.nextTick(cb, null);
      } catch (err) {
        process.nextTick(cb, err);
      }
    });
  },

  destroy,
});

There was an interesting asymmetry here.

If the AsyncFunction rejected, the rejection handler already did this:

destroyer(d, err);

But if the AsyncFunction resolved successfully, there was no equivalent cleanup.

That meant the successful early-return case could leave the Duplex around even though the function responsible for consuming its input was already finished.

That was the missing piece.

---

The fix

The fix ended up being quite small.

I added a flag:

let finalized = false;

Then, when final() starts:

final(cb) {
  finalized = true;

  final(async () => {
    try {
      await promise;
      process.nextTick(cb, null);
    } catch (err) {
      process.nextTick(cb, err);
    }
  });
}

And when the AsyncFunction successfully resolves:

(val) => {
  if (val != null) {
    throw new ERR_INVALID_RETURN_VALUE('nully', 'body', val);
  }

  if (!finalized) {
    destroyer(d);
  }
}

The important question is now:

Has final() already started?

If it has, let the normal finalization path continue.

If it hasn’t, and the AsyncFunction has already successfully returned, the Duplex needs to be destroyed.

So the lifecycle becomes:

AsyncFunction resolves
        |
        v
Has final() started?
      /   \
    yes    no
     |      |
     v      v
 normal   destroy
  path      |
            v
      pipeline cleanup

One subtle detail here is that finalized doesn’t mean “the stream has finished.”

It means final() has started.

That distinction matters because the flag is set synchronously at the beginning of final().

---

Why not just change fromAsyncGen()?

This was probably the most important part of looking through the history of the code.

There had already been an attempt to solve related cleanup behavior in PR #55096.

That change was broader. It included the AsyncFunction behavior I needed, but it also changed the shared fromAsyncGen() machinery, including handling around .return() and .throw().

That broader change was later reverted in #56278 after it caused problems with async-generator transforms.

So I didn’t want to bring that whole change back.

The problem I was fixing was specifically in the AsyncFunction path.

The final patch therefore leaves the shared fromAsyncGen() implementation alone and only changes the Promise/AsyncFunction branch.

That gives us:

AsyncFunction path
       |
       +---- fix early successful completion
       
AsyncGenerator path
       |
       +---- leave existing behavior alone

This ended up being a good example of why understanding the history of a piece of code can be just as important as understanding the code itself.

A more general-looking fix isn’t necessarily a better fix.

---

The regression test

Once the behavior was understood, the regression test was fairly simple.

The test creates a Readable, pipes it into an AsyncFunction that intentionally doesn’t consume its input, and checks that the original Readable is eventually destroyed.

const r = Readable.from(['foo', 'bar', 'baz']);

pipeline(
  r,
  Duplex.from(async function() {
    // Intentionally do not consume the async iterable input.
  }),
  common.mustCall(() => {
    assert.strictEqual(r.destroyed, true);
  }),
);

Before the fix, the pipeline could remain stuck and the callback wouldn’t be called.

After the fix, destroying the Duplex allows the normal pipeline cleanup to propagate to the upstream Readable.

The test specifically captures the zero-consumption case because it’s the simplest way to reproduce the problem.

---

One thing I had to correct in my own mental model

While investigating this, I initially thought about the problem mostly as:

“The async function returned, but final() never ran.”

That’s close, but not quite precise enough.

The earlier stall can happen at the write() handshake itself.

The internal async generator is waiting for something to consume the input, and if that consumer has already returned, the callback associated with the write can’t make normal progress.

So final() is better understood as part of the lifecycle that never gets a chance to take over normally.

Another useful distinction is that destroying the stream doesn’t necessarily mean the pipeline completed successfully.

destroyer(d) can destroy an unfinished stream with an AbortError. The important behavior for this bug is that the stream gets torn down and the upstream cleanup can happen instead of leaving the pipeline hanging forever.

---

The final flow

Before the fix:

Readable
   |
   v
pipeline()
   |
   v
Duplex.from(async function)
   |
   v
fromAsyncGen()
   |
   v
AsyncFunction returns successfully
   |
   v
Success handler does nothing
   |
   v
Async iterable isn't being consumed
   |
   v
Internal handshake stalls
   |
   v
Duplex doesn't finish
   |
   v
pipeline() keeps waiting
   |
   v
Readable isn't destroyed

After the fix:

Readable
   |
   v
pipeline()
   |
   v
Duplex.from(async function)
   |
   v
AsyncFunction returns successfully
   |
   v
final() hasn't started
   |
   v
destroyer(d)
   |
   v
Duplex destroyed
   |
   v
Pipeline cleanup propagates
   |
   v
Readable destroyed

The actual code change was only a few lines.

Getting to those lines was the interesting part.

---

Contributing to Node.js

This was also my first contribution to Node.js, so there was a second learning curve outside the code itself.

The PR went through the normal review and CI process. I initially had Jenkins failures that I couldn’t inspect because I didn’t have permission to read the Jenkins logs.

After the CI was rerun, the checks passed and the review process continued.

Eventually, PR #65963 was merged.

That was a nice moment, but honestly, the more useful part for me was the investigation itself.

Working on an application, it’s easy to stay within the code you wrote yourself.

Working on a project like Node.js is different. A relatively small change can take you through several layers of internal machinery, historical fixes, regression tests, and assumptions that aren’t obvious from the public API.

---

What I took away from it

The biggest lesson for me was that asynchronous bugs are often lifecycle bugs.

There wasn’t some complicated algorithm hiding here.

There were just several things happening at the same time:

• an async function had its own lifecycle
  

• the async iterable had another lifecycle
  

• the writable side had its own finalization lifecycle
  

• pipeline() was waiting for the stream lifecycle to settle
  

• destroying one part of the pipeline affected cleanup elsewhere
  

The hard part was understanding how those lifecycles were supposed to interact.

The final fix was small:

if (!finalized) {
  destroyer(d);
}

But that line only makes sense once you understand why the AsyncFunction is allowed to finish before final() and what needs to happen when it does.

That’s what made this contribution interesting for me.