Node.js Streams Backpressure: The Memory Leak That Kills Servers at 3AM

// ❌ Classic backpressure bug — write() return ignored
const readable = fs.createReadStream('huge-file.csv');
const writable = someSlowDestination.createWriteStream();

readable.on('data', (chunk) => {
 writable.write(chunk); // Returns false when buffer full — IGNORED!
 // readable keeps firing 'data' events, chunks pile up in memory
});
// Result: memory grows until process crashes

// ✅ Fix 1: Respect the write() return value manually
readable.on('data', (chunk) => {
 const ok = writable.write(chunk);
 if (!ok) {
 readable.pause(); // Stop reading when buffer full
 writable.once('drain', () => readable.resume()); // Resume when drained
 }
});
readable.on('end', () => writable.end());

// ✅ pipe() handles backpressure automatically
// It pauses/resumes the readable based on writable drain events
readable.pipe(writable);

// With transform streams (e.g., gzip compression)
const zlib = require('zlib');
fs.createReadStream('input.csv')
 .pipe(zlib.createGzip()) // Transform: compress
 .pipe(fs.createWriteStream('output.csv.gz')); // Write compressed

// Limitation of pipe(): errors don't propagate cleanly
// If transform throws, writable is not automatically closed

const { pipeline } = require('stream/promises'); // Node.js 15+
const zlib = require('zlib');

async function processFile() {
 try {
 await pipeline(
 fs.createReadStream('input.csv'),
 zlib.createGzip(),
 fs.createWriteStream('output.csv.gz')
 );
 console.log('Done — all streams closed cleanly');
 } catch (err) {
 // All streams automatically destroyed on error
 console.error('Pipeline failed:', err);
 }
}
// pipeline() advantages over pipe():
// ✅ Propagates errors to all streams automatically
// ✅ Destroys all streams on error (prevents leaks)
// ✅ Returns a Promise (works with async/await)
// ✅ Handles backpressure identically to pipe()

// Monitor stream buffer sizes to detect backpressure
const { Writable } = require('stream');

class MonitoredWritable extends Writable {
 _write(chunk, encoding, callback) {
 const bufferLength = this.writableLength;
 const hwm = this.writableHighWaterMark;
 
 if (bufferLength > hwm * 0.8) {
 console.warn(`Buffer at ${Math.round(bufferLength/hwm*100)}% capacity — backpressure building`);
 }
 
 // Simulate slow write
 setTimeout(() => callback(), 10);
 }
}

// Check stream buffer programmatically
setInterval(() => {
 const memUsage = process.memoryUsage();
 console.log({
 heapUsed: Math.round(memUsage.heapUsed / 1024 / 1024) + 'MB',
 external: Math.round(memUsage.external / 1024 / 1024) + 'MB',
 });
}, 1000);
// Growing 'external' memory = stream buffer accumulation = backpressure bug

const express = require('express');
const app = express();

// ❌ BAD: No backpressure handling — client slow = server OOM
app.get('/download', (req, res) => {
 const stream = fs.createReadStream('huge-dataset.json');
 stream.on('data', chunk => res.write(chunk)); // Ignores backpressure!
 stream.on('end', () => res.end());
});

// ✅ GOOD: pipeline() handles everything
app.get('/download', async (req, res) => {
 res.setHeader('Content-Type', 'application/json');
 res.setHeader('Transfer-Encoding', 'chunked');
 
 try {
 await pipeline(
 fs.createReadStream('huge-dataset.json'),
 res // res is a Writable stream — pipeline handles backpressure
 );
 } catch (err) {
 if (err.code !== 'ERR_STREAM_DESTROYED') {
 // Client disconnected — this is normal, not an error
 res.status(500).end('Stream error');
 }
 }
});

// Default highWaterMark is 16KB — often too small for large file transfers
// Increase for better throughput, decrease to limit memory usage

// Large file transfer — bigger buffer = fewer pause/resume cycles
const readable = fs.createReadStream('video.mp4', {
 highWaterMark: 64 * 1024 // 64KB chunks instead of default 16KB
});

// Object mode streams (e.g., database rows)
const dbStream = db.query('SELECT * FROM logs').stream({
 highWaterMark: 100 // Buffer 100 objects (rows) instead of default 16
});

// Rule of thumb:
// - High throughput, large files: highWaterMark = 64KB–1MB
// - Memory-constrained: highWaterMark = 4KB–16KB (default)
// - Object streams: highWaterMark = number of objects to buffer