Async Iterators and Stream Backpressure in Node.js

You rewrote a stream pipeline with for await because it is easier to read, and memory still climbs on large exports — the loop writes to a response faster than the client reads, or collects every row into an array “just to sort it”, or fires off an unbounded number of concurrent database calls per chunk. This guide from Node.js Stream Backpressure and Memory Growth, part of Node.js Server-Side Memory Management, explains which half of backpressure async iteration gives you for free, which half you still own, and how to write iterator-based pipelines with bounded memory.

Symptom Root Cause Immediate Action Measurable Impact
Memory grows while streaming a large export to a slow client res.write() inside for await ignores its return value Await 'drain' when write() returns false, or use pipeline() Response buffer bounded near its threshold
Heap equals the whole dataset during processing Chunks collected into an array before output Process and emit per chunk; sort/aggregate incrementally or in the database Memory ≈ one chunk, not the dataset
Hundreds of concurrent DB calls per stream Promises started per chunk without awaiting Bound concurrency (a small pool) Pending work and memory capped
Source stays open after an early exit Loop broke before the end, stream not cleaned up Rely on break destroying the stream, or finally cleanup Descriptors and buffers released
Errors leave partial pipelines running Manual loops without error propagation Use pipeline(source, generator, dest) Whole pipeline destroyed on error

Root Cause: Pull on the Reading Side, Push on the Writing Side

A Node.js Readable is an async iterable. for await (const chunk of readable) asks for the next chunk only when the loop body finishes, so the source is pulled at the speed of your processing: if the body awaits a slow database call, the stream simply reads no further (beyond its read-ahead of about highWaterMark, as described in choosing highWaterMark for Node.js streams). That is backpressure on the reading side, for free.

The writing side is different. writable.write(chunk) is not awaited; it appends to the writable’s buffer and returns false once the buffer exceeds its threshold. A loop that does for await (const row of rows) res.write(toCsv(row)) reads from the database as fast as the database delivers and pushes into the response buffer regardless of how fast the client receives — so a slow client makes the response buffer grow to the size of the entire export. You must wait for 'drain' whenever write() returns false, which events.once(res, 'drain') makes easy, or better, let stream.pipeline() connect an async generator to the destination so Node handles the writing side correctly.

Two other habits reintroduce unbounded memory. Accumulation: pushing every chunk into an array to sort, deduplicate or count before writing turns a streaming job into a load-everything job; push those operations into the database, use external sorting, or aggregate incrementally. Unbounded concurrency: starting a promise per chunk (rows.map(async …) inside the loop, or calling an async function without awaiting) makes pending promises and their payloads grow without limit — the same retention chain as unsettled promises that leak their closures. Suspended iterator frames themselves hold only the current chunk, as long as you do not keep large values alive across awaits, per how async functions and generators keep frames on the heap.

Which side of the loop has backpressure On the left, a database cursor stream is consumed with for await; it is pulled only when the loop body finishes, so reading has backpressure. On the right, res.write pushes into the response buffer; with a slow client the buffer grows unless the loop awaits the drain event when write returns false. pipeline with an async generator handles both sides. DB cursor stream pulled by for await backpressure: automatic loop body toCsv(row) res.write(line) response buffer grows with a slow client unless you await 'drain' pipeline(cursor, async function* (rows) …, res) handles both sides, errors and cleanup

Step-by-Step Fix

  1. Reproduce with a slow consumer. Stream a large export to a client that reads slowly (curl --limit-rate 100k) while logging res.writableLength and RSS. Verification: the buffer and RSS grow with the export size.
  2. Honour write backpressure. In manual loops, check write()'s return value and await once(res, 'drain') when it is false. Verification: writableLength stays near writableHighWaterMark with the slow client.
  3. Prefer pipeline() with an async generator. Express the transformation as async function* and connect pipeline(source, transform, destination). Verification: errors in any stage destroy all stages, and backpressure works in both directions.
  4. Stop accumulating. Replace in-memory arrays of all rows with per-chunk processing; move sorting and grouping to the database or an external sort. Verification: heap during the export stays roughly constant regardless of row count.
  5. Bound concurrency. When each chunk needs async work, process with a fixed concurrency (for example 4–8) instead of starting a promise per chunk. Verification: the number of in-flight operations never exceeds the limit.
  6. Clean up on early exit. Let break or thrown errors end the loop; Node destroys the stream when iteration stops early. Add finally blocks for other resources. Verification: file descriptors and connections close when a client disconnects mid-stream.
RSS during a 2 GB export to a slow client A for await loop that calls res.write without awaiting drain grows RSS to about 2.3 gigabytes as the whole export buffers in the response. The same export through pipeline with an async generator stays around 140 megabytes for the entire duration. 2.4 GB 0 res.write without drain pipeline + async generator (~140 MB) export progress (0 → 100%)

Command and Code Reference

Use case: a streamed CSV export with correct backpressure.

const { pipeline } = require('node:stream/promises');

app.get('/export.csv', async (req, res) => {
  res.setHeader('content-type', 'text/csv');
  const rows = db.queryStream('SELECT id, name, total FROM orders ORDER BY id'); // cursor-backed
  try {
    await pipeline(
      rows,
      async function* toCsv(source) {
        yield 'id,name,total\n';
        for await (const r of source) {                 // pulled at the consumer's pace
          yield `${r.id},${escapeCsv(r.name)},${r.total}\n`;
        }
      },
      res,                                              // pipeline waits for drain for us
    );
  } catch (err) {
    if (!res.headersSent) res.status(500).end();       // client aborted or DB error: all destroyed
  }
});

Use case: a manual loop that awaits drain.

const { once } = require('node:events');

async function writeAll(rows, out) {
  for await (const row of rows) {
    if (!out.write(format(row))) {
      await once(out, 'drain');                          // pause reading until the buffer empties
    }
  }
  out.end();
}

Use case: bounded concurrency for per-chunk async work.

async function* enrich(source, limit = 4) {
  const inFlight = new Set();
  for await (const item of source) {
    const p = lookup(item).then((extra) => ({ ...item, extra }));
    inFlight.add(p);
    p.finally(() => inFlight.delete(p));
    if (inFlight.size >= limit) yield await Promise.race(inFlight); // wait before pulling more
  }
  while (inFlight.size) yield await Promise.race(inFlight);
}
// Note: output order may differ from input order with this simple pool.

Verification and Regression Prevention

Test streaming endpoints with slow clients and large datasets: RSS and heap should stay roughly flat for the whole transfer, writableLength should hover near the threshold, and aborting the client mid-transfer should close the database cursor and file handles. A load test with a mix of fast and throttled clients at realistic concurrency catches most regressions.

Standardise on pipeline() for connecting stream stages, and treat write() calls whose return value is ignored as review findings. Add a unit test that streams a large synthetic source into a deliberately slow writable and asserts that the writable’s buffer never exceeds a few times its threshold. For the full backpressure picture, see pipeline vs pipe for memory-safe streams.

Memory while serving a slow client Streaming a large dataset to a throttled client, code that pushes rows without honouring backpressure buffers everything the client has not read, so RSS climbs for the whole transfer. With for await and pipeline, writableLength hovers near the threshold and RSS stays roughly flat. RSS transfer to a throttled client write() without awaiting drain for await + pipeline() Abort the client mid-transfer too: the cursor and file handles must close.

Edge Cases and Gotchas

Array.fromAsync and collecting helpers

Helpers that gather an entire async iterable into an array — Array.fromAsync, readable.toArray() — load everything into memory. Use them only for bounded sources.

Order and concurrency

Processing chunks concurrently changes output order unless you reorder results. If order matters, use a bounded pool that yields results in input order, or keep concurrency at 1.

Breaking out of loops over shared streams

Breaking out of for await destroys the stream by default. If you need to continue reading the same stream later, use readable.iterator({ destroyOnReturn: false }) — and remember that the stream then stays open until you finish it.

Object-mode read-ahead

An object-mode source reads ahead up to its highWaterMark in objects. With large records, even a pulled loop holds several records at once; size object-mode thresholds for record size.

Frequently Asked Questions

Does for await provide backpressure in Node.js streams?

For reading, yes: the stream is pulled only as fast as your loop body completes. For writing, no: write() is not awaited, so you must wait for 'drain' when it returns false, or connect the stages with pipeline().

Why does my streaming export use as much memory as the whole file?

Usually because the loop writes to the response without waiting for 'drain', so the response buffers everything a slow client has not received yet — or because rows are collected into an array before writing. Honour write backpressure and process per chunk.

Is pipeline() better than a manual for await loop?

For connecting a source, transformations and a destination, yes. pipeline() handles backpressure on both sides, propagates errors and destroys all stages on failure or abort. Manual loops are fine for consumption without a writable destination.

How do I limit concurrency inside a stream loop?

Keep a small set of in-flight promises and wait for one to finish before pulling the next chunk once the set reaches its limit. Libraries and newer stream helpers offer concurrency options, but the principle is the same: never start work per chunk without a bound.

What happens if the client disconnects mid-stream?

With pipeline(), the destination errors or closes, and all stages — including your generator and the source — are destroyed, releasing buffers and connections. With manual loops, you must detect the close yourself and stop iterating.