Backpressure in Node.js streams: how a misconfigured pipe takes down an ingestion microservice

Every time we investigate an ingestion microservice whose memory climbs without limit until the process dies with an OutOfMemory or the orchestrator restarts it for exceeding its RAM budget, the pattern repeats: someone wrote data into an output stream without checking whether that stream was ready to receive it. The code works in development, where volumes are small and the source never produces faster than the destination can consume. It fails in production, where a large file, a slow connection to the destination, or a traffic spike breaks that implicit assumption. The mechanism that exists specifically to prevent this is called backpressure, and Node has exposed it in its streams API from the start. The tool isn't missing — pipe() just leaves too much room to ignore it.

What backpressure actually is in a stream

A Node stream connects a data source to a consumer, and the two sides rarely process at the same speed. A Readable reading from a TCP socket or a file on disk can produce chunks far faster than a Writable writing to another service over HTTP, to a slow disk, or to a database can accept them. Without any control mechanism, the reading side would keep pushing data into the writer, and that data would pile up in an internal buffer while it waits to be processed. That buffer has no natural ceiling: it grows with every chunk that arrives before the previous one finishes writing, and that unchecked growth is exactly what exhausts the process's memory.

Backpressure is the signal a Writable stream sends back when its internal buffer reaches the configured highWaterMark — 16 KB by default for streams in binary mode. A Writable's write() method returns false when this happens, and that return value is an explicit instruction: stop writing until the stream emits the drain event, which signals that the buffer has dropped enough to accept more data.

const { createWriteStream } = require('fs');

const destination = createWriteStream('/var/data/output.log');

function writeBatch(data, index) {
  if (index >= data.length) return;

  const canContinue = destination.write(data[index]);

  if (canContinue) {
    writeBatch(data, index + 1);
  } else {
    destination.once('drain', () => writeBatch(data, index + 1));
  }
}

This manual pattern works, but nobody writes it this way in a real microservice with several chained transformations. That's where pipe() comes in, and that's where the problem this article is named after begins.

Why pipe() isn't enough on its own

pipe() does respect backpressure on the data-flow side: if the destination Writable returns false, pipe() automatically pauses the source Readable until the next drain. That much it does exactly as advertised. The problem shows up in error handling, which is where most ingestion microservices actually fail.

const { createReadStream, createWriteStream } = require('fs');
const { createGunzip } = require('zlib');

const source = createReadStream('/tmp/input.csv.gz');
const decompressor = createGunzip();
const destination = createWriteStream('/var/data/output.csv');

source.pipe(decompressor).pipe(destination);

If destination emits an error — say, because the disk filled up or the network connection backing that stream dropped — pipe() does not propagate that error to source or to decompressor. Each stream in the chain keeps its own error listener, and if one wasn't installed on every single one, Node throws an uncaught exception for whichever stream actually has the unhandled error, while the earlier streams in the chain are left open, never closing their file descriptors or releasing their buffers. In a microservice processing thousands of files per hour, that dangling file descriptor and that unreleased buffer accumulate until the process runs out of memory or available file descriptors, and the symptom that gets reported is "intermittent memory leak," not "a middle stream in a pipe chain was missing an error handler."

The other problem with pipe() is that it doesn't automatically close the streams in a chain when one finishes before the others. If source finishes emitting data but destination still has a full buffer processing the last batch, resource cleanup is left to whoever wrote the code, and it's common for nobody to write it, because in local tests with small files it never surfaces.

pipeline(): the same backpressure, with lifecycle handled

stream.pipeline(), available since Node 10 and with promise support since Node 15 via stream/promises, fixes exactly these two points without changing the underlying backpressure mechanism, which is still the same write() returning false and the same drain event.

const { pipeline } = require('stream/promises');
const { createReadStream, createWriteStream } = require('fs');
const { createGunzip } = require('zlib');

async function processFile(inputPath, outputPath) {
  try {
    await pipeline(
      createReadStream(inputPath),
      createGunzip(),
      createWriteStream(outputPath)
    );
    console.log('processing complete');
  } catch (error) {
    console.error('stream chain failed', error);
  }
}

pipeline() propagates the error from any stream in the chain to a single handling point, and it also destroys every stream involved — calling its internal destroy() — as soon as any of them fails or finishes, regardless of where in the chain that happened. That's what eliminates the dangling file descriptors and orphaned listeners that pipe() leaves behind when something breaks mid-chain.

The typical pattern that saturates memory in an ingestion microservice

The case we see most often isn't a simple pipe() like the example above, but a microservice that receives an input stream over HTTP, transforms it with a custom Transform, and forwards it to another service, also over HTTP:

app.post('/ingest', (req, res) => {
  const transformer = new MyTransform();

  req.pipe(transformer).pipe(remoteDestination);

  remoteDestination.on('finish', () => res.status(200).end());
});

If remoteDestination is a stream to an outgoing HTTP connection that turns slow — say, because the downstream service starts degrading under load — backpressure propagates correctly back to req thanks to pipe(), and up to that point there's no leak. The problem shows up when that custom Transform keeps its own internal buffer inside its _transform method that doesn't respect the callback correctly: if the Transform calls push() with more data than it received, or if it accumulates chunks in an internal array "to process in batches" without checking push()'s return value, that array grows unchecked, because stream backpressure only controls the flow between native streams — not the internal data structures a poorly written Transform decides to keep on its own.

class BadTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.buffer = this.buffer || [];
    this.buffer.push(process(chunk));

    if (this.buffer.length >= 1000) {
      for (const item of this.buffer) this.push(item);
      this.buffer = [];
    }
    callback();
  }
}

This Transform accumulates up to a thousand items before emitting anything, and while it accumulates, it keeps accepting chunks from req because callback() gets called on every invocation regardless of how large this.buffer has grown. If remoteDestination is slow and the inbound flow is steady, this internal buffer grows indefinitely because nothing in the streams API knows it exists: respecting backpressure on the way in, not just passing it along on the way out, is entirely the Transform's own responsibility.

The fix isn't complicated, but it requires understanding that push() inside a Transform also returns false when the read side is saturated, and that accumulating data "to batch process" inside a stream voids the guarantee the rest of the chain takes for granted:

class FixedTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(process(chunk));
    callback();
  }
}

Emitting each processed chunk immediately, with no intermediate accumulation, lets backpressure flow end to end through the chain with no blind spots.

What to check before signing off on a stream chain

Replacing pipe() with pipeline() fixes resource cleanup and error propagation, but it doesn't replace reviewing every custom Transform in the chain: any internal accumulation of data that doesn't go through push() as soon as it's ready is a point where backpressure stops propagating, no matter how well the rest of the pipeline is written. The checklist we run before approving a new ingestion microservice is short: use pipeline() instead of hand-chained pipe(), check that no Transform accumulates chunks without checking backpressure, and confirm that each stream's highWaterMark is sized for the actual chunk size the source produces, not for the default value someone assumed without measuring the service's real traffic.