TopolearnIgnoring backpressure is how a fast producer and slow consumer turn into an out-of-memory crash.
Every stream has an internal buffer whose target size is the highWaterMark (64 KiB for byte streams in Node 22, 16 objects in object mode). It is a threshold, not a hard limit, and it is how backpressure works: the mechanism by which a slow consumer tells a fast producer to wait.
On the writable side, writable.write(chunk) returns a boolean. false means "my buffer is above the highWaterMark — stop until I emit 'drain'". The data is still accepted; nothing enforces the pause. Code that ignores the return value keeps buffering, and memory grows until the process dies.
if (!out.write(chunk)) await once(out, "drain");On the readable side, a stream reads ahead only until its buffer reaches the highWaterMark, then waits for the consumer. In a custom Readable, push() returning false is your signal to stop producing until _read() is called again.
The good news: `pipeline()` and `for await` handle all of this for you. pipeline pauses sources when destinations fill up; async iteration only pulls the next chunk when your loop body finishes. Manual on("data") handlers are where backpressure gets lost, because 'data' keeps firing however slowly you process it.
When writing a Transform, call the callback only when you have finished with the chunk, push output with this.push(), and use flush(callback) to emit anything buffered at the end. Errors go to callback(err), which destroys the stream and propagates through pipeline.
Tune highWaterMark deliberately: bigger buffers improve throughput for large sequential copies, smaller ones reduce memory per connection. Measure before changing it.
Example
import { Transform, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createWriteStream } from "node:fs";
// Transform: object rows -> CSV lines, with a header emitted lazily
class ToCsv extends Transform {
constructor() { super({ writableObjectMode: true }); this.header = false; }
_transform(row, _enc, callback) {
if (!this.header) { this.push(Object.keys(row).join(",") + "\n"); this.header = true; }
callback(null, Object.values(row).join(",") + "\n");
}
_flush(callback) { this.push("# end\n"); callback(); }
}
// A large lazy source: rows are produced only as fast as the file accepts them
function* rows(n) { for (let i = 0; i < n; i++) yield { id: i, square: i * i }; }
await pipeline(Readable.from(rows(1_000_000)), new ToCsv(), createWriteStream("squares.csv"));
console.log("rss MB:", Math.round(process.memoryUsage().rss / 1e6)); // bounded, whatever n isRemember
write()returning false means wait for'drain'pipelineandfor awaitapply backpressure automatically- In Transforms, call the callback when done and use
flushfor trailing output
Watch out for
- Writing in a loop and ignoring
write()'s return value, buffering unbounded data. - Consuming with
on('data')and slow async work, which defeats backpressure.