Introduction

This issue appears when a readable stream emits end, but the asynchronous work started inside readable callbacks is still running. That is expected behavior: Node streams know when no more chunks are available, but they do not automatically wait for your custom promises. To make completion reliable, you need to connect stream completion and async task completion explicitly.

Why end Does Not Wait for Your Async Work

The readable event only tells you that data can be pulled from the internal buffer. If your handler starts an asynchronous function, the stream does not track it. The end event fires once the source has no more data, not when your database writes, HTTP calls, or timers finish.

This simplified example shows the race:

1const { Readable } = require("stream");
2
3const input = Readable.from(["a", "b", "c"]);
4
5input.on("readable", async () => {
6  let chunk;
7  while ((chunk = input.read()) !== null) {
8    await new Promise((resolve) => setTimeout(resolve, 50));
9    console.log("processed", chunk);
10  }
11});
12
13input.on("end", () => {
14  console.log("stream ended");
15});

stream ended can appear before the last processed line because the stream only knows that reading is complete. It has no knowledge of the await inside your handler.

Prefer Async Iteration for Sequential Processing

If you want to read each chunk and wait for async work before moving on, for await...of is the cleanest pattern. It ties chunk consumption to your async function naturally.

1const { Readable } = require("stream");
2
3async function main() {
4  const input = Readable.from(["a", "b", "c"]);
5
6  for await (const chunk of input) {
7    await new Promise((resolve) => setTimeout(resolve, 50));
8    console.log("processed", chunk);
9  }
10
11  console.log("all async work finished");
12}
13
14main().catch(console.error);

This approach is easier to reason about than mixing event listeners and await. It also makes error handling straightforward because thrown exceptions become rejected promises that you can catch in one place.

Track In-Flight Work If You Must Use Events

Sometimes you need to keep an event-based design. In that case, maintain your own list of pending promises and wait for them after end.

1const { Readable } = require("stream");
2
3const input = Readable.from(["a", "b", "c"]);
4const pending = [];
5
6input.on("readable", () => {
7  let chunk;
8  while ((chunk = input.read()) !== null) {
9    const task = new Promise((resolve) => {
10      setTimeout(() => {
11        console.log("processed", chunk);
12        resolve();
13      }, 50);
14    });
15
16    pending.push(task);
17  }
18});
19
20input.on("end", async () => {
21  await Promise.all(pending);
22  console.log("stream ended and tasks finished");
23});
24
25input.on("error", (err) => {
26  console.error(err);
27});

This works, but you are now responsible for managing memory, failures, and backpressure. If readable can produce thousands of chunks quickly, pending may grow too large.

Use pipeline and Transforms for Stream-Shaped Work

When the job is truly stream processing rather than chunk collection, a Transform stream or pipeline is often the better model. It keeps flow control inside the stream system.

1const { Readable, Transform } = require("stream");
2const { pipeline } = require("stream/promises");
3
4async function main() {
5  const input = Readable.from(["a", "b", "c"]);
6
7  const upper = new Transform({
8    objectMode: true,
9    async transform(chunk, encoding, callback) {
10      await new Promise((resolve) => setTimeout(resolve, 50));
11      callback(null, chunk.toUpperCase());
12    }
13  });
14
15  const output = new Transform({
16    objectMode: true,
17    transform(chunk, encoding, callback) {
18      console.log("processed", chunk);
19      callback(null, chunk);
20    }
21  });
22
23  await pipeline(input, upper, output);
24  console.log("pipeline finished");
25}
26
27main().catch(console.error);

pipeline gives you one promise that settles when the whole chain finishes or fails. That is usually the completion signal people expect from end.

Choose the Pattern Based on Ownership

If you own the whole consumer, prefer async iteration. If you are extending an event-based system, track pending work explicitly. If the problem is stream processing end to end, redesign around pipeline. The bug usually comes from mixing models: event handlers are being used as if they were promise-aware completion hooks.

Common Pitfalls

• Assuming end means all asynchronous work launched from readable has completed.
• Using an async event listener and expecting Node to wait for the returned promise automatically.
• Accumulating unbounded pending promises when readable emits faster than downstream work completes.
• Ignoring error handling while focusing only on readable and end.
• Mixing event-driven reading with stream transforms when a pipeline design would be simpler and safer.

Summary

• 'end only signals that no more stream data will arrive.'
• Async work started in readable handlers must be tracked separately.
• 'for await...of is the clearest pattern for sequential async chunk processing.'
• 'Promise.all over tracked tasks can work in event-based code, but requires care.'
• 'pipeline is often the best fit when the whole workflow is naturally stream-shaped.'