Node.js Streams — Missing drain Handler Causes OOM

Every Node.js server you have ever run has been silently relying on Streams and Buffers — whether you knew it or not. When you serve a 4 GB video file, parse an incoming multipart upload, or pipe data from a database cursor to an HTTP response, you are in stream territory. Get it wrong and your server leaks memory, stalls under load, or corrupts binary data in ways that are genuinely nightmarish to debug at 2 AM. Get it right and you can process files larger than your available RAM with a flat, predictable memory footprint that holds steady under load.

The core problem Streams solve is the mismatch between producer speed and consumer speed. A database might emit rows faster than your HTTP client can receive them. A file system read might outpace a gzip compressor. Without a flow-control mechanism, the fast side buffers everything into memory until something crashes. Node.js Streams solve this with backpressure — a built-in signalling protocol between producers and consumers that says 'slow down' or 'keep going' without you writing a single line of coordination logic.

I have personally traced three separate production OOM incidents back to misunderstood stream backpressure — in two cases, engineers had used pipe() and assumed it handled everything safely, not realizing that pipe() does not propagate errors and does not protect against a broken backpressure chain inside a custom Transform.

By the end of this article you will understand how the Buffer class maps onto V8 memory outside the garbage collector, what highWaterMark actually controls (it is not a hard limit, and most engineers get this wrong), how to build a production-grade Transform stream, why pipeline() is almost always safer than pipe(), and the specific failure modes that only show up under load — never in your development environment.

Buffer Internals — V8 Heap vs External Memory

Buffer is one of the most misunderstood classes in Node.js, and the misunderstanding usually surfaces in production as an OOM kill that heap profiling tools cannot explain. Engineers look at the heap snapshot, see 50 MB, and cannot reconcile it with the 2 GB RSS climbing in their dashboards. The reason is where Buffer memory actually lives.

Before Node.js 4.5, Buffer used the V8 heap, meaning every Buffer allocation competed with JavaScript objects for garbage-collected memory. Since then, Buffer.alloc() and Buffer.allocUnsafe() allocate memory from a pool managed outside V8's heap using the C++ layer through libuv. This memory is not tracked by V8's garbage collector in the same way — it is reference-counted and returned to the pool when the Buffer is dereferenced. The GC knows a Buffer exists as a JavaScript object, but the actual bytes that Buffer points to are in external memory that the GC cannot compact or move.

This has a concrete production implication: your heap snapshot will show a clean 50 MB heap while your process RSS sits at 2 GB because of accumulated Buffer allocations. Every heap profiling tool, every Chrome DevTools memory snapshot, every heapUsed metric will look completely healthy. You must monitor process.memoryUsage().external and process.memoryUsage().rss to detect Buffer-related memory growth.

Buffer.alloc(size) zero-fills the allocated memory before returning it, which is safe but slower. Buffer.allocUnsafe(size) skips zero-filling, making it roughly 10x faster for large allocations. The 'unsafe' name is precise: the returned memory may contain bytes from previous allocations — fragments of other users' data, previous tokens, partial file contents — if you read from it before writing. For security-sensitive paths, always use Buffer.alloc(). For internal processing where you guarantee write-before-read, Buffer.allocUnsafe() is the correct production choice and the performance difference is real at high allocation rates.

const { Buffer } = require('buffer');

// SAFE: zero-filled before returning — use for user-facing or security-sensitive data.
// The zero-fill is not optional safety theater — it prevents information disclosure.
const safeBuf = Buffer.alloc(1024);
console.log('alloc [0]:', safeBuf[0]); // 0 — guaranteed, always

// FAST: returned uninitialized — use for internal processing with write-before-read.
// The 'unsafe' refers to information disclosure risk, not memory safety.
const fastBuf = Buffer.allocUnsafe(1024);
console.log('allocUnsafe [0]:', fastBuf[0]); // unpredictable — could be any byte

// Small allocations (< 4 KB) share a pre-allocated 8 KB pool.
// This means two small allocUnsafe calls may share an underlying ArrayBuffer.
// Writing to one at the wrong offset can corrupt the other.
const smallA = Buffer.allocUnsafe(100);
const smallB = Buffer.allocUnsafe(200);
console.log('Same backing store:', smallA.buffer === smallB.buffer); // true for small allocs

// The key metric to watch in production — NOT heapUsed.
// Buffer memory shows up in external and rss, not heapUsed.
const mem = process.memoryUsage();
console.log({
  rss:          `${(mem.rss          / 1024 / 1024).toFixed(1)} MB`, // total process memory
  heapUsed:     `${(mem.heapUsed     / 1024 / 1024).toFixed(1)} MB`, // V8 JS objects
  external:     `${(mem.external     / 1024 / 1024).toFixed(1)} MB`, // Buffer bytes
  arrayBuffers: `${(mem.arrayBuffers / 1024 / 1024).toFixed(1)} MB`  // ArrayBuffer bytes
});

// If external climbs while heapUsed stays flat — Buffer leak.

Stream Types and Their Internal State Machines

Node.js provides five stream types, each with a distinct role in a data pipeline. Understanding their internal state machines is not academic — it is what lets you diagnose production issues where streams silently stop flowing, emit data after destruction, or hold memory that the GC cannot reclaim.

Every Readable stream has two operating modes: paused and flowing. In paused mode — the default — data is buffered internally and you must explicitly call read() to pull chunks out. In flowing mode, data is pushed to you automatically via data events as fast as the underlying source can produce it. Calling .resume(), piping to a Writable, or attaching a data listener switches to flowing mode. The most common cause of 'stream hang' bugs I have debugged is a Readable created and then left in paused mode with no consumer attached — data accumulates in the internal buffer, the highWaterMark is crossed, and the underlying source pauses, and nothing ever flows. The process looks healthy. No error is emitted. Everything is just silently stuck.

Writable streams have a simpler state machine driven by the callback in _write(). When _write() invokes its callback, the stream is ready to receive the next chunk. When the internal buffer crosses highWaterMark, write() returns false — this is the backpressure signal. The drain event fires when the buffer drops back below highWaterMark.

Duplex streams like TCP sockets combine both — independent Readable and Writable sides with independent state machines sharing one underlying resource. Transform streams like zlib.createGzip() are Duplex streams where the write side feeds into the read side through your _transform() implementation. PassThrough streams are identity Transforms useful for injecting inspection points into a pipeline without modifying data.

const { Readable, Writable, Transform, PassThrough } = require('stream');

// --- Readable state inspection ---
const readable = new Readable({
  highWaterMark: 16 * 1024, // 16 KB internal buffer
  read(size) {
    // This is called when the consumer wants data.
    const shouldContinue = this.push(Buffer.from('data chunk'));
    if (!shouldContinue) {
      // Consumer not reading fast enough — stop producing.
      // The stream will call read() again when the consumer is ready.
    }
    this.push(null); // null signals end of stream
  }
});

console.log('Initial state:', {
  readableFlowing: readable.readableFlowing, // null = paused, no listeners
  readableLength:  readable.readableLength,   // 0 = nothing buffered yet
  readableEnded:   readable.readableEnded     // false = not done
});

readable.resume(); // switch to flowing mode — data events start firing

console.log('After resume:', {
  readableFlowing: readable.readableFlowing, // true
});

// --- Transform with destroy guard ---
const safeTransform = new Transform({
  highWaterMark: 16 * 1024,
  transform(chunk, encoding, callback) {
    if (this.destroyed) return callback();
    const processed = chunk.toString().toUpperCase();
    callback(null, Buffer.from(processed));
  },
  flush(callback) {
    // emit buffered remainder here, if any
    callback();
  }
});

// --- PassThrough for pipeline inspection ---
const inspector = new PassThrough();
let bytesThrough = 0;
inspector.on('data', chunk => {
  bytesThrough += chunk.length;
});
// Insert inspector between any two pipeline stages without affecting data flow.

Backpressure — The Flow Control Protocol

Backpressure is the single most important concept in Node.js Streams, and the one most frequently misunderstood in practice. It is not a rate limiter, not a throttle, and not a buffer size configuration. It is a cooperative protocol between a Readable and a Writable where the Writable signals 'I need you to slow down' and the Readable obliges — if the producer is paying attention.

The mechanism works through the return value of write(). When you call writable.write(chunk), the method returns true if the internal buffer is below highWaterMark and false if it is at or above it. When write() returns false, the protocol says the producer should stop writing and wait for the drain event before sending more data. If the producer ignores this signal and keeps calling write(), the data is still buffered — but in memory, without any bound, until the process runs out of memory and the OOM killer fires. There is no automatic enforcement. Backpressure is cooperative, not mandatory.

The highWaterMark is not a hard limit. This is the specific detail that most engineers who have read about streams still get wrong in interviews. It is a heuristic threshold — 16,384 bytes for binary streams, 16 objects for objectMode streams by default — where write() starts returning false to signal the producer to pause. But the stream will still accept data beyond this point. The buffer can grow arbitrarily beyond highWaterMark if the producer ignores the signal. Think of highWaterMark as the 'please slow down' sign on a highway, not the physical guardrail at the edge of a cliff.

pipe() handles backpressure automatically within its direct neighbours: when the destination's write() returns false, pipe() calls readable.pause(). When drain fires, pipe() calls readable.resume(). This is why pipe() seems to work in simple cases. The failure mode — which only appears in production under sustained load — is when a custom Transform breaks the backpressure chain by calling its callback immediately regardless of whether the downstream stream has drained.

const fs = require('fs');

// Manual backpressure implementation — shown to illustrate the protocol.
// In production, use pipeline() which implements this correctly for you.
function copyWithBackpressure(sourcePath, destPath) {
  const readable = fs.createReadStream(sourcePath);
  const writable = fs.createWriteStream(destPath);

  readable.on('data', (chunk) => {
    const canContinue = writable.write(chunk);
    if (!canContinue) {
      // Backpressure engaged — pause the producer.
      readable.pause();
      // Resume only when writable has drained.
      writable.once('drain', () => readable.resume());
    }
  });

  readable.on('end', () => writable.end());

  // Error handling on both sides — without this, errors crash the process.
  readable.on('error', (err) => {
    console.error('Read error:', err.message);
    writable.destroy(err); // destroy the other stream too
  });
  writable.on('error', (err) => {
    console.error('Write error:', err.message);
    readable.destroy(err);
  });

  return new Promise((resolve, reject) => {
    writable.on('finish', resolve);
    writable.on('error', reject);
  });
}

pipe() vs pipeline() — Error Propagation and Resource Cleanup

pipe() is the most commonly used stream API, and in production it is also the most dangerous one when used without fully understanding its limitations. I have seen three separate post-mortem write-ups at different companies trace back to the same root cause: pipe() does not propagate errors, and it does not destroy streams on error.

Here is the concrete failure mode. If you have readable.pipe(transform).pipe(writable) and the transform stream emits an error, the error is emitted only on the transform. The readable stream is not notified, not paused, not destroyed — it keeps emitting data into a transform that is in an error state. The writable stream is not notified and not destroyed — it keeps waiting for data that may never arrive or may arrive in a corrupted state. Both streams hold their underlying resources: the readable holds an open file descriptor, the writable holds an open socket or file handle. Under sustained error conditions — a flaky upstream service that errors on 5% of requests — this accumulates EMFILE errors as file descriptors exhaust the OS limit.

pipeline() from stream/promises solves both problems with one API call. When any stream in the chain errors or closes prematurely, pipeline() automatically destroys all other streams in the chain, propagates the error as a rejected Promise, and ensures all resources are cleaned up. In Node.js 18+, pipeline() also supports async generators as pipeline stages, which allows you to inject stateful processing logic — like computing a hash or accumulating metrics — inline without writing a full Transform class.

stream.finished() is the complementary utility for monitoring a single stream's completion. It returns a Promise that resolves when a stream emits 'finish' or 'end', and rejects on error or premature close. Use it when you need to wait for a stream to complete without piping it anywhere — for example, waiting for a write stream to flush before reading the file it wrote.

const { pipeline } = require('stream/promises');
const { finished }  = require('stream/promises');
const fs     = require('fs');
const zlib   = require('zlib');
const crypto = require('crypto');

// Production pattern: process an upload, compute its hash, and write compressed.
// The async generator stage is a pipeline-compatible way to do inline processing
// without writing a full Transform class.
async function processUpload(inputStream, outputPath) {
  const gzip       = zlib.createGzip({ level: 6 });
  const fileStream = fs.createWriteStream(outputPath);
  const hasher     = crypto.createHash('sha256');

  await pipeline(
    inputStream,
    gzip,
    // Async generator as a pipeline stage
    async function* (source) {
      for await (const chunk of source) {
        hasher.update(chunk);
        yield chunk;
      }
    },
    fileStream
  );

  return hasher.digest('hex');
}

// Usage with specific error handling
async function handleUpload(req, outputPath) {
  try {
    const sha256 = await processUpload(req, outputPath);
    console.log('Upload complete. SHA256:', sha256);
    return { success: true, hash: sha256 };
  } catch (err) {
    if (err.code === 'ERR_STREAM_PREMATURE_CLOSE') {
      console.info('Client disconnected before upload completed');
      return { success: false, reason: "client_disconnect" };
    }
    console.error('Upload failed:', err.message);
    return { success: false, reason: "processing_error", error: err.message };
  }
}

// stream.finished() — wait for a single stream to complete
async function waitForFlush(writeStream) {
  await finished(writeStream);
  console.log('Write stream fully flushed — safe to read the file now');
}

Building a Custom Transform Stream for Production

Custom Transform streams are where most backpressure bugs are born. You write a _transform() method, call the callback, and assume everything works. Then under load, memory grows, or data gets corrupted, or the stream hangs. The problem is almost always the same: the Transform breaks the backpressure chain by not waiting for the downstream consumer to drain before signalling readiness to the upstream producer.

The contract of _transform() is simple but unforgiving: you receive a chunk, you process it, and you call the callback with the result (or null if you want to pass it through). The stream uses the timing of that callback to decide whether to ask for more data from upstream. If you call callback() synchronously on every chunk — even when the downstream is struggling — the upstream never pauses, and your Transform becomes an unbounded buffer.

A production-grade Transform must respect the backpressure signal from its own writable side. Concretely: if this.push() returns false because the readable buffer is full, you should not call the callback until the drain event fires on the readable side. The built-in Transform class handles this in most cases, but if you are using a custom push mechanism or if you are writing to multiple destinations, you must implement the pause/drain cycle yourself.

Another critical pattern: always guard _transform() with a this.destroyed check. The destroy() method sets the destroyed flag but does not abort in-flight _transform() calls. Without the guard, chunks queued before destroy will still be processed, pushed to an already-closed stream, causing ERR_STREAM_DESTROYED or silent data loss.

And don't forget _flush(). It's called when the writable side ends. If your Transform buffers data across chunks (like a CSV row parser waiting for a newline), _flush() is where you emit the remainder. Forgetting _flush() means data loss at the end of every stream.

const { Transform } = require('stream');

class LineParser extends Transform {
  constructor(options = {}) {
    options.objectMode = true; // emit full lines as strings
    super(options);
    this._buffer = '';
    this._paused = false;
  }

  _transform(chunk, encoding, callback) {
    // 1. Guard against post-destroy processing
    if (this.destroyed) {
      return callback();
    }

    this._buffer += chunk.toString();
    const lines = this._buffer.split('\n');
    // Keep the last (potentially incomplete) piece in buffer
    this._buffer = lines.pop();

    for (const line of lines) {
      const processed = this._processLine(line);
      const shouldContinue = this.push(processed);
      if (!shouldContinue) {
        // Backpressure: stop processing and wait for drain
        this._paused = true;
        this.once('drain', () => {
          this._paused = false;
          this._flushBuffer(callback);
        });
        return; // don't call callback yet
      }
    }

    callback();
  }

  _flush(callback) {
    // Emit the final partial line (if any) when writable side ends
    if (this._buffer.length > 0) {
      this.push(this._processLine(this._buffer));
      this._buffer = '';
    }
    callback();
  }

  _processLine(line) {
    // Example processing: trim and uppercase
    return line.trim().toUpperCase();
  }

  _flushBuffer(callback) {
    // If we were paused, resume emitting from buffer when drain fires
    // This is a simplified version; production code would re-emit queued items
    callback();
  }
}

// Usage
const { pipeline } = require('stream/promises');
const fs = require('fs');

async function processFile(inputPath, outputPath) {
  const readable = fs.createReadStream(inputPath, { encoding: 'utf8' });
  const writable = fs.createWriteStream(outputPath);
  const parser = new LineParser();

  await pipeline(readable, parser, writable);
  console.log('File processed');
}