Node.js stream backpressure: bound the whole export

Node.js stream backpressure works only while the producer follows downstream demand. Use a pull-driven source and pipeline(), put a limit on each record, and test with a consumer that deliberately stops accepting data. A stream’s highWaterMark signals pressure; it is not a ceiling on the memory used by the entire export. ^[1][2]

Scenario: an order-export job reads records, serializes each record as newline-delimited JSON, and sends the bytes to a destination. It works with a small account. A larger account, or a destination that pauses, makes the process grow until the job is killed. The investigation starts at the boundary between fetching a record and admitting its serialized bytes—not at the process’s heap setting.

This is an illustrative implementation exercise, not a customer incident or a throughput benchmark. The test should establish one boundary: when the destination stops, the source must stop advancing after its bounded in-flight buffers fill.

Where can an apparently streamed export still buffer everything?#

Consider rows.map(serialize) followed by Readable.from(chunks). The final leg is a stream, but every row and serialized chunk already exists. Changing the writable watermark cannot undo that allocation. A similar problem occurs when an SDK fetches the complete result before exposing an iterator. The interface looks incremental while the underlying storage is eager.

Start by drawing the actual owners of data: database page, fetched objects, encoded chunks, transform input, transform output and destination queue. For each owner, ask when it releases a reference and what prevents another allocation. If the answer is “the callback will eventually run,” find whether the caller actually waits for that callback before producing more work.

An asynchronous event listener deserves special scrutiny. source.on('data', async row => await send(row)) does not turn EventEmitter delivery into a sequential loop. Ordinary listeners are called synchronously; returning a promise does not establish a producer admission rule. A source that keeps emitting can create many outstanding sends. That is an application queue outside the stream’s own buffer. ^[5]

For this export, choose one owner for fetching: an async iterator that requests the next page only after consuming the current one. Choose one owner for sending: the pipeline. If the database driver prefetches pages, document its bound separately. “Uses an iterator” is a starting point for inspection, not proof of bounded storage.

What does a false return from write actually mean?#

It means the producer should pause before supplying another chunk. The current chunk has entered the writable path; treating false as “try the same chunk again” can duplicate output. The producer resumes after drain, while stream errors require their own handling. The official Node guidance makes respecting this signal part of the producer contract. ^[1][2]

A watermark also has units. In byte mode, the relevant queue contains bytes. In object mode, a small count can still retain a few enormous objects. A transform has both an input side and an output side; setting one threshold does not erase the other. Inspect the configuration of the actual stages rather than importing a default from a different Node release.

Suppose the export has a small page of ordinary orders and one order with an unbounded free-text attachment embedded in its JSON. The row count is controlled, but the record size is not. You need a record policy as well as flow control: reject, externalize or separately stream the oversized field. Raising the heap postpones the same failure and makes concurrent exports more expensive.

For the worked example, an oversized serialized record fails the job. That is a product decision: a partial export must not look complete. A service that needs partial results should instead define an explicit per-record rejection format and a completion manifest. Silently dropping the bad row is not an equivalent repair.

Build one pipeline with an explicit record policy#

The following module takes an async iterable and a destination writable. It deliberately does not know about HTTP, database drivers or object storage. That keeps the flow-control boundary testable without a provider account. Use a supported Node release with node:stream/promises; the API references checked here are pinned to Node 24.13.0. ^[1]

// bounded-export.mjs
import { pipeline } from 'node:stream/promises';

export async function writeExport(rows, destination, {
  signal,
  maxRecordBytes = 64 * 1024,
} = {}) {
  if (!Number.isSafeInteger(maxRecordBytes) || maxRecordBytes < 1) {
    throw new RangeError('maxRecordBytes must be a positive integer');
  }
  await pipeline(
    async function* ({ signal: pipelineSignal }) {
      for await (const row of rows) {
        pipelineSignal.throwIfAborted();
        const text = JSON.stringify(row);
        if (text === undefined) throw new TypeError('row is not JSON data');
        const bytes = Buffer.from(text + '\n');
        if (bytes.length > maxRecordBytes) {
          throw new RangeError('serialized record exceeds export policy');
        }
        yield bytes;
      }
    },
    destination,
    { signal },
  );
}

There is one important limit to this limit. JSON.stringify() and Buffer.from() allocate before the size check runs. This module rejects a large encoded record; it does not prevent an already enormous input object or string from being allocated. Validate field sizes earlier, avoid loading embedded binaries into the row, and use a streaming encoder when individual values can legitimately exceed the process budget.

The iterator must also cooperate with cancellation. If its next() is waiting on a database request that ignores the signal, aborting the pipeline does not magically interrupt that remote operation. Give the driver a bounded fetch and an appropriate cancellation path. A timeout signal can represent a deadline, but cancellation needs support from each participating operation. ^[4]

Use a fresh destination per attempt. Decide who owns failed artifacts. For a file export, write to an attempt-specific staging location and promote only after completion and the required storage checks. For remote uploads, define abort and reconciliation behavior with the storage client. Pipeline completion is not a universal business commit protocol.

How do you prove the source actually stops?#

A fast destination makes an eager producer look healthy. Replace it with a writable whose first write callback stays pending until the test releases it. Count how many source records were requested while that callback remained blocked. This creates an observable ordering condition instead of relying on a guessed sleep or a memory graph.

The test below uses large-enough fixture rows and many records, then asserts that the source has not been exhausted. It does not prescribe an exact queue length, since several internal buffers may read ahead. It establishes the narrower property we need: the entire fixture is not consumed while the destination is stopped.

// gated-example.mjs
import assert from 'node:assert/strict';
import { Writable } from 'node:stream';
import { writeExport } from './bounded-export.mjs';

let produced = 0;
let release;
let entered;
const firstWrite = new Promise(resolve => { entered = resolve; });
const rows = {
  async *[Symbol.asyncIterator]() {
    for (let id = 0; id < 200; id++) {
      produced++;
      yield { id, note: 'x'.repeat(4096) };
    }
  },
};
const destination = new Writable({
  highWaterMark: 8192,
  write(chunk, encoding, callback) {
    if (!release) {
      release = callback;
      entered();
    } else {
      setImmediate(callback);
    }
  },
});
const pending = writeExport(rows, destination);
await firstWrite;
await new Promise(resolve => setImmediate(resolve));
assert(produced < 200, 'source was exhausted behind a stopped sink');
release();
await pending;
assert.equal(produced, 200);

Add a negative control: materialize every row before connecting the destination. The same source-count assertion should fail for that deliberately eager variant. Otherwise, a passing assertion may simply be too weak to distinguish the failure you are investigating.

Then verify output separately. Count decoded records, compare representative fields, and include a multibyte string near the byte limit. Character count is not byte count. If an export can contain embedded newlines, confirm the JSON encoding preserves one physical line per record. The destination fixture should collect small test output only; a production verification helper that accumulates the whole export would reintroduce the original problem.

Parallel enrichment introduces another place to hold work. If each order triggers a lookup, launching a promise for every order and later streaming the results has already admitted the whole lookup set. Keep enrichment sequential for the first implementation, or define a finite worker count and a finite result queue. An ordered export also needs a policy for a slow early lookup, because completed later rows may otherwise accumulate while waiting for it.

Test that boundary with a second gate around the lookup itself. Count started lookups while the first group is held, then release one and confirm that only the permitted next work begins. This separates destination backpressure from enrichment admission. A passing stopped-sink test cannot prove that a parallel helper buried upstream has a bound.

Which failure changes the recovery plan?#

An export that pauses correctly can still leave a misleading artifact or keep a remote query alive. Treat failure behavior as part of the implementation, not a follow-up operational chore.

Inject a destination error after several writes and assert that the export promise rejects. Inject an abort while the sink is gated and assert rejection without releasing more source work. Confirm the source iterator’s cleanup runs. Finally, test an oversized record and ensure the destination never receives that record. These are behavior checks, not estimates of a real storage system’s failure rate.

Do not automatically retry an export into the same visible object. A destination may have accepted some bytes before a connection failed. Give each attempt an identity and expose only verified complete results. If the business operation also sends a notification or creates a report record, its retry policy needs a separate action identity.

How much memory should you measure?#

Measure several boundaries together: produced records, current page size, active exports, readable and writable queue lengths, and process memory. Node’s heapUsed describes V8 allocations; arrayBuffers includes Buffer storage, while RSS covers a wider process footprint. A stable heap graph alone cannot establish that a byte-heavy export is stable. ^[3]

Compare the same fixture with a normal sink, a slow sink and a stopped sink. Keep the input and stage configuration fixed. Record the Node version and explicit watermarks. If you report a maximum queue observation, name the queue and its units; do not call it the process memory maximum.

The job queue needs its own admission limit. A bounded export multiplied by unbounded simultaneous jobs still creates an unbounded commitment. Put an admission limit in front of the job and a bounded waiting queue or rejection policy behind that limit. Choose the limit from observed resource use in the deployed environment rather than multiplying an internal watermark and assuming the result accounts for everything.

For this scenario, acceptance means that the gate stops source progress, successful output preserves the fixture, errors and aborts reject, and oversized rows have an explicit result. Production acceptance adds driver prefetch, storage cleanup, job admission and real workload memory observations. Save the driver version, destination settings and stopped-consumer result beside the release.

For the related distinction between a caller deadline and physical work completion, see the asyncio worker-lifetime example. Coordinate any export retries with one bounded retry owner.

Is highWaterMark a hard memory limit?

No. It is a threshold used by stream flow control. Input objects, encoded records, other stages and application queues can all hold additional memory. An oversized single record can exceed the threshold even when the producer respects pressure. ^[1]

Should I retry a chunk when write returns false?

No. Pause before producing the next chunk. The return value is a pressure signal, not an instruction to resend the current chunk. Wait for drain when using the manual write interface and handle errors separately. ^[1][2]

Does an async data listener process records sequentially?

Ordinary EventEmitter delivery does not await the returned promise. Use a pull-driven iterator or an explicitly bounded concurrency mechanism when each record triggers asynchronous work. Confirm the source’s actual behavior rather than relying on the async keyword. ^[5]

Does pipeline completion prove the export is published safely?

It proves completion of the pipeline’s participating stages under their contracts. A storage commit, final filename, completion manifest or notification is a separate business step. Define those steps so a failed attempt cannot be mistaken for a complete export.