Node.js Streams and Backpressure: Handling Large Data

The function that processes a 2GB CSV works perfectly on the 4MB sample and kills the process in production with JavaScript heap out of memory. Nothing is wrong with the logic. It read the whole file into a string first, and a string that size does not fit.

Streams are the fix, and the part that makes them work rather than merely run is backpressure: the mechanism that stops a fast producer from burying a slow consumer.

The problem, concretely

import { readFile, writeFile } from 'node:fs/promises';

// Peak memory is roughly twice the file size. Fine at 4MB.
// At 2GB the process dies before it writes anything.
const input = await readFile('data.csv', 'utf8');
const output = input.split('\n').map(transform).join('\n');
await writeFile('out.csv', output);

The default V8 heap limit is around 1.5 to 4GB depending on the Node version and platform. Raising it with --max-old-space-size is the reflex, and it converts a crash at 2GB into a crash at 8GB. The file will eventually be bigger.

Streams process a chunk at a time

A stream moves data in pieces, so peak memory depends on chunk size rather than file size. Four kinds exist and each does one thing.

import { createReadStream, createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import { Transform } from 'node:stream';

const upperCase = new Transform({
  transform(chunk, _encoding, callback) {
    // callback(error, data). Passing an error here aborts the pipeline.
    callback(null, chunk.toString().toUpperCase());
  },
});

// Peak memory is a few chunks, whatever the file size.
await pipeline(
  createReadStream('data.csv'),
  upperCase,
  createWriteStream('out.csv')
);

Backpressure, which is the actual point

Reading from a local disk is fast. Writing to a network socket or a database is slow. Without coordination the reader races ahead and the unwritten data piles up in memory, which reproduces the original problem through a longer route.

The signal is the return value of write(). It returns false when the internal buffer is over its high water mark, meaning stop, and the stream emits drain when it is ready again.

// Ignoring the return value is the bug. This buffers the entire
// source in memory when the destination is slower than the source.
readable.on('data', (chunk) => {
  writable.write(chunk); // returns false and nobody looks
});

// Respecting it.
readable.on('data', (chunk) => {
  if (!writable.write(chunk)) {
    readable.pause();
    writable.once('drain', () => readable.resume());
  }
});

Use pipeline, not pipe

.pipe() is the older API and it does not forward errors or clean up. If a stream in the middle fails, the others stay open, and you get a leaked file descriptor per failure.

// Broken. An error in gzip leaves source and destination open,
// and the error itself is unhandled, which crashes the process.
source.pipe(gzip).pipe(destination);

// Destroys every stream on failure and gives you one place to catch.
import { pipeline } from 'node:stream/promises';

try {
  await pipeline(source, gzip, destination);
} catch (error) {
  logger.error({ error }, 'pipeline failed');
}

The promise version from node:stream/promises is what I use everywhere. The callback form still exists and works identically.

Async iterators, which read better

Readable streams are async iterable, so a for await loop replaces event handlers and gets backpressure automatically, because the loop body has to finish before the next chunk is pulled.

import { createReadStream } from 'node:fs';
import { createInterface } from 'node:readline';

const lines = createInterface({
  input: createReadStream('huge.log'),
  crlfDelay: Infinity, // treat \r\n as one break
});

let errors = 0;
for await (const line of lines) {
  // Awaiting inside the loop pauses reading. Backpressure, for free.
  if (line.includes('ERROR')) {
    await recordError(line);
    errors++;
  }
}

readline is the answer to line-by-line processing, and it handles the case people get wrong when doing it manually: a line that straddles a chunk boundary. Splitting each chunk on \n independently truncates records at 64KB intervals, and on a log file that looks like occasional corrupt lines rather than an obvious bug.

Generators as transforms

import { pipeline } from 'node:stream/promises';

// An async generator is a valid pipeline stage. Often clearer than
// a Transform subclass, and it can await.
async function* parseAndEnrich(source: AsyncIterable<string>) {
  for await (const line of source) {
    if (!line.trim()) continue;
    const row = JSON.parse(line);
    yield JSON.stringify(await enrich(row)) + '\n';
  }
}

await pipeline(
  createReadStream('in.ndjson', 'utf8'),
  splitLines,
  parseAndEnrich,
  createWriteStream('out.ndjson')
);

Object mode and highWaterMark

By default streams carry buffers and the high water mark is measured in bytes, 64KB for readable file streams and 16KB for writable ones. Object mode carries arbitrary values and the mark counts objects instead, defaulting to 16.

const batcher = new Transform({
  objectMode: true,
  // Sixteen rows in flight is fine. Sixteen 50MB records is not:
  // the mark counts objects, and says nothing about their size.
  highWaterMark: 4,

  transform(row, _enc, callback) {
    this.batch = [...(this.batch ?? []), row];
    if (this.batch.length >= 500) {
      const batch = this.batch;
      this.batch = [];
      return void insertMany(batch).then(() => callback(), callback);
    }
    callback();
  },

  // Runs once at the end. Without it the last partial batch is dropped.
  flush(callback) {
    if (!this.batch?.length) return callback();
    insertMany(this.batch).then(() => callback(), callback);
  },
});

That flush is easy to forget and the resulting bug is specific: everything works except the final few hundred records, which vanish. It only shows up when the record count is not an exact multiple of the batch size, so a test with 1,000 rows and a batch of 500 passes.

Web Streams, and which to use

Node also supports the Web Streams API, the same ReadableStream the browser has. It is what fetch bodies and Next.js Route Handlers use, so any server-rendered work touches it.

import { Readable } from 'node:stream';

// Node stream -> Web stream, for a Response body.
const webStream = Readable.toWeb(createReadStream('report.csv'));
return new Response(webStream as ReadableStream, {
  headers: { 'Content-Type': 'text/csv' },
});

// Web stream -> Node stream, for a fetch body.
const nodeStream = Readable.fromWeb(response.body!);

My rule: Web Streams at the HTTP boundary because that is what the platform APIs expect, Node streams for file and database work because the ecosystem is there. Convert at the edge with those two helpers rather than trying to pick one for everything. Streaming a response is also what makes Suspense-based server rendering work, which I covered in React Suspense and lazy.

The mistakes worth naming, and the Node.js memory leaks they cause

An unhandled error event. A stream that emits error with no listener throws an uncaught exception and takes the process down. pipeline() handles this; raw .pipe() chains do not.

Mixing await into a data handler. stream.on('data', async (chunk) => ...) does not pause anything. The handler returns a promise the stream ignores, so chunks keep arriving while the previous ones are still being processed. Use for await.

Concatenating chunks into one buffer. Collecting everything into an array and joining at the end is the original problem in stream clothing.

Assuming a chunk is a record. Chunk boundaries fall wherever the buffer filled up, not on your delimiter. Use readline, a parser, or a splitter that carries the remainder forward.

When not to bother

Streams add real complexity. For a file you know is a few megabytes, or a query returning a few thousand rows, read it into memory and move on. The code is shorter and easier to follow.

One thing streams do not solve: CPU-bound work. A transform that spends 200ms hashing each chunk blocks the event loop regardless of how the data arrives, and that is what Node.js worker threads are for. Streams fix memory, not concurrency.

Reach for streams when the input size is unbounded or attacker-controlled (an upload), when it is genuinely large, or when time to first byte matters and you can start emitting before the work finishes. An upload endpoint is also exactly where a size limit and API rate limiting belong, since streaming removes the memory ceiling that was accidentally protecting you. Those three cases are where they earn the complexity. Everything else is premature.