Node.js Streams Backpressure: Handling Large Payloads Safely

What backpressure is and how stream buffers work

Backpressure is the mechanism that lets a slow consumer tell a fast producer to stop sending data. Without it, the gap between the two speeds becomes memory. If a client uploads faster than your storage backend accepts writes, the difference has to live somewhere, and by default that place is your process heap.

Buffers and highWaterMark

Every Node.js stream holds an internal buffer. Readables queue data that has been produced but not yet consumed. Writables queue data that has been accepted but not yet flushed. Each buffer has a threshold called highWaterMark. For byte streams it is measured in bytes. In objectMode it is a count of objects, regardless of how large each object is.

highWaterMark is a signal to stop, not a memory cap. When a writable's buffer reaches the threshold, write() returns false, but the chunk you passed is still accepted and queued. A producer that ignores that return value can push the buffer far past the limit. The same applies in object mode: sixteen 50 MB objects are still sixteen objects. The limit only works if the producer respects the signal.

How the signal travels up a chain

Consider an HTTP request piped through a gzip transform into a file or object-storage writable:

1. The storage writable falls behind, and its buffer fills.
2. The transform stops being drained, so its output side fills and it stops reading input.
3. The request stream's buffer fills, and Node stops reading from the socket.
4. The kernel's receive buffer fills, and TCP flow control slows the client.

Each stage holds at most about one highWaterMark of data. Memory stays bounded across the whole path, from the network to the disk.

This matters to anyone building uploads, downloads, reverse proxies, or ETL jobs. In all of these, the input size is unknown and the producer's speed is controlled by someone else.

CopyDATA FLOW ------------------------------------------------------------->

+----------+   +----------+   +----------+   +----------+   +----------+
|  Client  |-->|  Kernel  |-->| HTTP req |-->|   gzip   |-->| Storage  |
| (upload) |   | TCP recv |   | Readable |   |Transform |   | Writable |
|          |   |  buffer  |   |[buf<=HWM]|   |[in] [out]|   |[buf<=HWM]|
+----------+   +----------+   +----------+   +----------+   +----------+
     ^              ^              ^              ^              |
     +<-------------+<-------------+<-------------+<-------------+
     BACKPRESSURE <------------------------------------------ slow sink

client slowed  (4) rcv buffer (3) req buffer (2) gzip out   (1) buf >= HWM
by TCP flow    full, TCP win  full, socket   full, stops    write() returns
control        shrinks        not read       reading input  false

Each stage holds about one highWaterMark, so memory stays bounded end to end.
HWM is a stop signal: write() past it still queues the chunk, so the
producer must wait for 'drain' or the buffer grows without limit.

Data flows left to right from client to storage, while backpressure moves right to left: a full storage buffer stalls the gzip transform, then the request stream and socket reads, and finally TCP flow control slows the client.

The write and read contracts: write(), drain, flowing and paused mode

Unknown. The writable side gives you a single signal to stay in step with it: the return value of write().

The write contract

write(chunk) always accepts the chunk. Its boolean return value tells you whether the internal buffer is still below highWaterMark. When it returns false, the stream is asking you to stop until it emits 'drain'. A correct manual loop writes chunks until it sees false, waits for a single 'drain' event, and then continues.

The broken version calls write() in a tight loop and ignores the return value. Nothing throws and nothing warns, because every chunk is accepted and queued. If the source reads from local disk faster than a slow socket can send, the whole file ends up in the writable's buffer in memory. The stream does not drop data, so it keeps it all.

The read contract

A readable starts in paused mode, and data moves only when you call read(). Adding a 'data' listener, calling resume(), or calling pipe() switches it to flowing mode, where chunks are pushed to you as fast as the source can produce them.

Flowing mode is where async handlers go wrong. The stream does not await a 'data' callback. If the callback starts a database insert and returns a promise, the next chunk arrives immediately and the in-flight work piles up without limit. To apply backpressure manually, call pause() before the async work and resume() after it finishes. pipe() does this for you: it pauses the source when dest.write() returns false and resumes it on 'drain'.

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

// Broken: ignores write()'s return value, so every chunk is queued in memory
function writeAllBroken(dest, chunks) {
  for (const chunk of chunks) dest.write(chunk); // returns false; nobody checks
  dest.end();
}

// Correct: stop when write() returns false, wait for one 'drain', then continue
async function writeAll(dest, chunks) {
  for (const chunk of chunks) {
    if (!dest.write(chunk)) {
      await once(dest, 'drain'); // buffer has emptied; safe to write again
    }
  }
  dest.end();
  await once(dest, 'finish');
}

// Flowing mode with async work: the stream does not await this callback,
// so pause the source until the work for this chunk is done
function consume(src, saveChunk) {
  src.on('data', async (chunk) => {
    src.pause(); // no more 'data' events while the insert runs
    try {
      await saveChunk(chunk);
    } catch (err) {
      src.destroy(err);
      return;
    }
    src.resume(); // ask for the next chunk
  });
}

// Usage: 10,000 chunks of 64 KiB written to a file without unbounded buffering
function* makeChunks(n) {
  for (let i = 0; i < n; i++) yield Buffer.alloc(64 * 1024, i % 256);
}
writeAll(fs.createWriteStream('/tmp/out.bin'), makeChunks(10_000))
  .then(() => console.log('done'));

The example contrasts a write loop that ignores write()'s return value with one that waits for 'drain', then uses pause() and resume() in a flowing-mode 'data' handler so async work applies backpressure the way pipe() does.

Implementing it with stream.pipeline() and async iteration

That covers the happy path. The weakness of .pipe() is what happens on failure. Errors do not travel along the chain. If the destination fails, the source is unpiped but not destroyed, and it keeps its file descriptor or socket open until something else closes it. Each stream needs its own error listener, and there is no single place to learn that the whole transfer finished. Under steady error traffic, those orphaned handles pile up.

pipeline() from stream/promises fixes all three problems. An error from any stage rejects one promise, and every stream in the chain is destroyed, so their underlying resources are released. You await it once, and that is your completion point.

pipeline() also accepts async generator functions as stages. Each one receives the previous stage as an async iterable and yields output chunks. Backpressure comes from the pull model: the generator body only runs when the next stage asks for data. If you await a database write or an HTTP call inside the for await...of loop, no further chunks are read until that call settles. You get flow control without pause(), resume() or drain.

Async generators add promise and iterator work on every chunk. A hand-written Transform avoids that and gives finer control over buffering. Whether the difference matters depends on chunk size and throughput, so profile your own workload before choosing one over the other.

Copyimport { createReadStream } from 'node:fs';
import { createGunzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

// Placeholder for a real DB client call; it simulates a slow write.
const insertRecord = (record) => new Promise((r) => setTimeout(r, 5, record));

// Async generator stage: turns byte chunks into complete text lines.
async function* splitLines(source) {
  const decoder = new TextDecoder(); // handles multi-byte chars split across chunks
  let remainder = '';
  for await (const chunk of source) {
    remainder += decoder.decode(chunk, { stream: true });
    const lines = remainder.split('\n');
    remainder = lines.pop(); // keep the partial last line for the next chunk
    for (const line of lines) if (line) yield line;
  }
  remainder += decoder.decode();
  if (remainder) yield remainder;
}

// Final stage: consumes lines. The await is the backpressure point.
// No more data is pulled from upstream until the insert settles.
async function writeToDb(lines) {
  for await (const line of lines) {
    await insertRecord(JSON.parse(line));
  }
}

try {
  // One promise for the whole chain: on any error, every stream is destroyed.
  await pipeline(
    createReadStream('./exports/events.ndjson.gz'),
    createGunzip(),
    splitLines,
    writeToDb,
  );
  console.log('import complete');
} catch (err) {
  console.error('import failed, all streams closed:', err);
  process.exitCode = 1;
}

A promise-based pipeline() streams a gzipped NDJSON file through an async generator. Awaiting each database write inside for await...of throttles the read side without pause(), resume() or drain, and a single try/catch covers errors and cleanup for every stage.

Streaming a large file upload with size limits and cancellation

An incoming http.IncomingMessage is already a Readable, so the upload path is a single pipeline(): the request, a byte-counting Transform, a hashing Transform, and a destination writable. The counter keeps a running total and calls its callback with an error as soon as the total exceeds the limit. Don't rely on Content-Length for this, because clients can omit it or send a false value. The error makes pipeline() destroy every stage, so nothing beyond the limit reaches the disk.

Pass an AbortSignal in the pipeline options and abort it when the response closes before you have finished. A client disconnect then tears down the whole chain instead of leaving the writable open and waiting. Write to a temporary path and rename it only after the pipeline resolves and the hash checks out. On rejection, unlink the temporary file. If you skip this, partial files build up on every aborted upload.

Slow clients cost you a socket and a file descriptor for as long as they keep trickling bytes. Set the server's headersTimeout and requestTimeout, and add an idle timer inside the pipeline that aborts when no chunk arrives within a set interval.

When the destination is object storage, backpressure only works if the client library respects write() return values. Some SDKs read the entire body into memory first, or need a known length before they start. Retries add another constraint: a request stream that has been consumed cannot be replayed. Multipart uploads with bounded, individually retryable parts are the usual way around both problems.

Copyimport http from 'node:http';
import { createHash, randomUUID } from 'node:crypto';
import { createWriteStream } from 'node:fs';
import { rename, unlink } from 'node:fs/promises';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

const MAX_BYTES = 100 * 1024 * 1024; // hard cap enforced on actual bytes, not Content-Length
const IDLE_MS = 15_000;              // abort if no chunk arrives for this long

const server = http.createServer(async (req, res) => {
  const ac = new AbortController();
  const tmp = `/var/uploads/.tmp-${randomUUID()}`;
  const hash = createHash('sha256');
  let bytes = 0;
  const idle = setTimeout(() => ac.abort(new Error('idle timeout')), IDLE_MS);
  // Client went away before we answered: destroy every stage, including the file stream.
  res.on('close', () => { if (!res.writableFinished) ac.abort(new Error('client disconnected')); });

  const guard = new Transform({
    transform(chunk, _enc, cb) {
      idle.refresh(); // data is flowing, so push the idle deadline forward
      bytes += chunk.length;
      if (bytes > MAX_BYTES) return cb(Object.assign(new Error('payload too large'), { status: 413 }));
      hash.update(chunk); // count and hash in one pass, no buffering
      cb(null, chunk);
    },
  });

  try {
    await pipeline(req, guard, createWriteStream(tmp), { signal: ac.signal });
    const expected = req.headers['x-content-sha256'];
    if (expected && hash.digest('hex') !== expected) throw Object.assign(new Error('hash mismatch'), { status: 422 });
    await rename(tmp, `/var/uploads/${randomUUID()}`); // publish only a complete, verified file
    res.writeHead(201).end();
  } catch (err) {
    await unlink(tmp).catch(() => {}); // never leave half-written files behind
    if (!res.headersSent && !res.destroyed) res.writeHead(err.status ?? 500, { connection: 'close' }).end();
  } finally { clearTimeout(idle); }
});

server.headersTimeout = 10_000;  // limit slowloris-style header trickling
server.requestTimeout = 120_000; // upper bound on total time to receive a request
server.listen(8080);

A single pipeline() streams the request through a combined byte-limit and SHA-256 transform into a temp file. An AbortSignal tied to client disconnects and an idle timer tears the chain down, and the file is renamed only after its hash verifies, or unlinked on any failure.

Diagnosing memory leaks in Node streams and tuning highWaterMark

Leaked or stalled streams show a recognizable pattern: resident memory (RSS) grows in proportion to payload size instead of staying flat, and writableLength on the destination or readableLength on the source keeps rising during a transfer. Both values are byte counts of data queued in the stream's internal buffer. If they settle near highWaterMark, backpressure is working. If they keep climbing, a producer is ignoring the false return value from write(), or the consumer has stopped reading. writableNeedDrain set to true for a long time points to a destination that is not draining. Usually this is a slow client socket, or a 'drain' listener that was never attached. Buffer contents are allocated outside the V8 heap, so check external and arrayBuffers from process.memoryUsage() alongside RSS. To confirm the cause, take two heap snapshots a few seconds apart. Then follow the retainer path from the growing Buffer objects back through the stream's internal buffer list to the request or socket that holds them.

Illustrative estimate (assumed: 50 concurrent 500 MB uploads, 64 KiB highWaterMark on both the readable and writable side; per-request overhead ignored):

Raising highWaterMark improves throughput when the cost per chunk dominates, for example syscalls or small writes to a high-latency destination. Once the slowest stage is saturated, a larger buffer adds no speed and multiplies memory by the number of connections. highWaterMark is also a threshold, not a hard cap: a single oversized chunk is still accepted whole. In objectMode, the threshold counts objects rather than bytes, so a handful of multi-megabyte objects can exhaust memory while the stream appears to be within its limit.

When to avoid streaming, and the alternatives

Streaming adds code paths for partial failure, cleanup, and ordering. For small, bounded payloads, such as a JSON body capped at a limit you enforce before reading, full buffering is simpler and equally safe. The limit is what makes it safe. Without it, buffering lets a client decide how much memory you allocate.

Some checks need the whole payload: a checksum the client sends up front, a full JSON Schema validation, or a virus scan. You cannot act on chunk one until you have seen the last chunk, so choose where the bytes wait.

Also note that pipeline() destroys every stream when one fails. If the destination is an HTTP response that should carry a 4xx error body, that response is torn down before you can write to it. Pipe into an intermediate stage instead, and handle the response yourself.

Anti-patterns checklist

• Readable.from() on a giant array: the data is already fully in memory, so streaming it saves nothing.
• Collecting chunks, then calling Buffer.concat: this is buffering without a size limit.
• JSON.parse on an unbounded body: it needs the whole string and blocks the event loop while it runs.
• Unhandled 'error' events: they crash the process or leak file descriptors.
• Transforms that ignore push()'s return value: they defeat backpressure when one input produces many outputs.

Impact on development cost and engineering time

Streaming code costs more to write than await readFile(). Every stage needs explicit error handling. Partial writes need cleanup, such as deleting a half-written temp file or aborting a multipart upload. Backpressure also requires end-to-end cooperation: if one stage buffers internally, for example a parser that accumulates a whole record set or a transform that ignores push()'s return value, the whole pipeline loses its memory bound. The tests still pass. The failure shows up only when load rises, which is the most expensive time to find it.

Most of that cost is paid once if pipeline() and async iteration become the team default rather than a per-author choice. Several leak classes then stop appearing: streams left open after an error, unclosed file descriptors, and listeners that keep buffers alive. Incident debugging also gets shorter, because a stalled or failed transfer surfaces as one rejected promise with a cause instead of a silent hang.

Predictable memory per connection makes capacity planning a calculation instead of a guess:

Illustrative estimate (assumed numbers, not a benchmark) peak buffered bytes ≈ concurrent connections × buffering stages × per-stage buffer budget With 500 uploads, 3 stages and a 64 KiB budget per stage: 500 × 3 × 64 KiB ≈ 94 MiB. This holds only if every stage honors backpressure.

Review checklists catch the cheap mistakes: hand-wired .pipe() chains, ignored return values and missing cleanup. Load tests with realistic concurrency and slow clients catch the stage that buffers internally, which code review usually misses.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to node.js streams backpressure, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Adding or removing a column goes through one review with the inheriting services listed, instead of ad-hoc messages to each team.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• Services that depend on a column get the chance to object before it disappears, avoiding emergency fixes.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.