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.

Why Node.js Streams Without drain Handling Cause OOM

Node.js streams are an abstraction for handling data piece-by-piece rather than loading entire payloads into memory. The core mechanic is backpressure: when a writable stream's internal buffer exceeds highWaterMark (default 16KB), stream.write() returns false, signaling the producer to pause. If the producer ignores this signal and keeps writing, data accumulates in memory until the process runs out of heap space. The drain event fires when the buffer has drained below highWaterMark, allowing writes to resume. This push-pull contract is what prevents unbounded memory growth. In practice, the highWaterMark is not a hard limit — it's a threshold that triggers the backpressure signal. The buffer can grow beyond it if writes continue, but the stream will eventually reject writes with an error if memory pressure is extreme. The drain event is the only reliable way to know when it's safe to write again. Use streams whenever you process data that could exceed available memory — file transfers, HTTP request bodies, database result sets. Without explicit drain handling, any high-throughput pipeline risks silent OOM crashes in production, especially under load spikes where the consumer (e.g., a slow network socket or disk) cannot keep up with the producer.

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');
}

Why Streams Matter — The 10GB File Problem

Most developers learn streams when their production service OOMs on a file upload. The math is brutal. A 100MB file loaded into memory with readFileSync consumes 100MB of RAM. A 10GB file consumes 10GB. Your server has 4GB. Game over.

Streams sidestep this entirely. Instead of swallowing the whole file, they process data in 64KB chunks. That means a 10GB file uses ~64KB of memory. Not a typo. The same memory footprint whether the file is 1MB or 100GB.

This isn't just about files. HTTP requests, database cursors, compression pipelines — any data source that produces bytes over time benefits from streaming. The alternative is buffering everything into a single blob, which scales linearly with data size. Streams scale to infinity, bounded only by disk I/O and network throughput.

The catch: streams demand a different mental model. You're not writing linear code anymore. You're orchestrating a pipeline where data flows asynchronously through stages. Get it wrong, and you'll face backpressure deadlocks, memory leaks, or silent data loss. But master it, and you can process datasets that would crash any naive implementation.

// io.thecodeforge — javascript tutorial

const fs = require('fs');

// ❌ Blows up on 10GB files
function naiveCopy(source, dest) {
  const data = fs.readFileSync(source);  // 10GB in RAM
  fs.writeFileSync(dest, data);          // 10GB more in RAM
  // Peak memory: 20GB
}

// ✅ Graceful at any size
function streamCopy(source, dest) {
  const read = fs.createReadStream(source, { highWaterMark: 65536 }); // 64KB chunks
  const write = fs.createWriteStream(dest);
  read.pipe(write);
  // Peak memory: ~64KB
}

// Usage
const FILE = './giant-log-2025-01-28.bin';
try {
  naiveCopy(FILE, '/dev/null');
} catch (err) {
  console.error('OOM:', err.message);
}

streamCopy(FILE, '/dev/null');
console.log('Streamed successfully');

Error Handling in Streams — The Silent Failure Trap

You piped a stream and nothing came out. No error. No output. Just a black hole. This is the classic stream surprise. Streams don't crash by default when something goes wrong inside a pipe chain. Errors get swallowed, and data silently stops flowing.

pipe() is the culprit. It propagates data and backpressure, but it does not propagate errors. If your read stream emits an 'error' event after piping, the write stream keeps waiting forever. No data, no close, just a zombie pipeline with dangling file handles.

The fix has two layers. First, always attach error listeners to every stream in the pipe chain. Second, stop using pipe() and switch to pipeline() from the 'stream/promises' module. pipeline() forwards errors to the final stream, and handles cleanup. It's the production-grade replacement.

Missing an error handler on a writable stream means the process never exits. You'll accumulate open file descriptors until your OS kills you. Seen it happen in a log aggregator that lost error events. 4000 file handles later, the kernel said no.

// io.thecodeforge — javascript tutorial

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

// ❌ Silent failure: pipe() ignores errors
function fragilePipeline() {
  const read = fs.createReadStream('missing-file.log');
  const write = fs.createWriteStream('output.log');
  read.pipe(write);  // If read fails, write waits forever
}

// ✅ Production-ready: pipeline() propagates errors
async function robustPipeline() {
  try {
    const read = fs.createReadStream('missing-file.log');
    const write = fs.createWriteStream('output.log');
    await pipeline(read, write);
  } catch (err) {
    console.error('Pipeline failed:', err.code, err.message);
    // Cleanup is automatic — close events fire
  }
}

// Test
robustPipeline();
// Output: Pipeline failed: ENOENT no such file or directory

Scenario A: Massive Database Exports (MongoDB / PostgreSQL)

When exporting millions of rows from MongoDB or PostgreSQL, naive collection can cause OOM crashes. Streams solve this by processing rows in flight. The critical pattern is piping a database cursor through a Transform stream that formats rows (CSV/JSON) and writes to a file or HTTP response. Without backpressure handling, the database cursor outruns the file system, causing memory buildup. The fix: listen to the cursor's 'readable' event and call read() only when the downstream signals 'drain'. For PostgreSQL, use pg-query-stream's cursor; for MongoDB, use cursor.pipe(). Never load the entire dataset into an array. Always end with pipeline() for cleanup on errors.

// io.thecodeforge — javascript tutorial

import { pipeline } from 'stream/promises';
import { createCursor } from 'mongoose'; // or MongoDB driver

const cursor = Model.find().batchSize(1000).cursor();
const transform = new Transform({
  objectMode: true,
  transform(row, enc, cb) {
    this.push(JSON.stringify(row) + '\n');
    cb();
  }
});

await pipeline(cursor, transform, process.stdout);
console.log('Done — no OOM');

Scenario B: AI Text Generation (The Web Streams API)

Modern AI APIs like OpenAI and Anthropic return chat completions as streams. Using the Web Streams API directly in Node.js lets you process tokens as they arrive—critical for real-time UI updates. The pattern: fetch the endpoint with response.body (a ReadableStream), pipe through a TextDecoderStream to get strings, then process each chunk. Key nuance: AbortSignal on the fetch cancels the stream cleanly, preventing partial token leakage. Do NOT accumulate tokens in a buffer; instead, push to a Transform stream for formatting or directly to an HTTP response. This eliminates tail latency and lets you cancel mid-request without waste.

// io.thecodeforge — javascript tutorial

const response = await fetch(url, {
  headers: { 'Authorization': `Bearer ${apiKey}` },
  signal: AbortSignal.timeout(30000)
});

const reader = response.body
  .pipeThrough(new TextDecoderStream())
  .getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  process.stdout.write(value); // stream tokens immediately
}
console.log('\nStream complete');