3.8.4 — Streams: Data in Motion, Memory at Rest

Here is the abstraction that separates naive from competent server code. fs.readFile on a 2 GB file loads all 2 GB into memory before your callback sees byte one; a few concurrent requests and the process is OOM-killed (2.5). A stream processes data in chunks as it arrives, so memory stays small and constant regardless of total size — and its companion concept, backpressure, is the feedback signal that keeps a fast producer from drowning a slow consumer. Streams are Node's Unix-pipe inheritance (2.8) made programmable; this page covers the four classes, the two reading modes, the exact mechanics of backpressure, pipe vs pipeline, building your own Transform, object mode, and the modern async-iteration face.

1. The four classes

Every stream is one of four shapes — all EventEmitters (3.8.5) with a shared core:

• Readable — a source you pull from: fs.createReadStream, an HTTP request body, process.stdin, a DB cursor.
• Writable — a sink you push to: fs.createWriteStream, an HTTP response, process.stdout, a socket's write side.
• Duplex — independently both, like a phone line: a TCP socket (net.Socket) — what you read isn't what you wrote.
• Transform — a Duplex where output is computed from input: zlib.createGzip(), crypto.createCipheriv, a CSV parser. The pipeline workhorse.

// stream a file to an HTTP client, compressing en route — constant memory
// whether the file is 1 MB or 100 GB:
fs.createReadStream("big.log")     // Readable   (source)
  .pipe(zlib.createGzip())         // Transform  (compress in transit)
  .pipe(response);                 // Writable   (sink — the HTTP response)

2. Backpressure: the mechanism, precisely

The scenario that makes streams necessary rather than nice: disk reads at ~500 MB/s; a mobile client drains at ~1 MB/s. Without coordination, 499 MB/s of difference accumulates in your process's memory until it dies. Backpressure is the feedback loop that prevents it, and its whole mechanism is three parts: ^{⚑What is backpressure? [EQ-66]}

1. Every stream has an internal buffer with a target ceiling — the highWaterMark (default 64 KB for byte streams; 16 objects in object mode). It's a threshold, not a hard cap.
2. writable.write(chunk) returns false when the internal buffer has reached the ceiling — the polite signal "stop sending."
3. When the buffer drains below the ceiling, the writable emits 'drain' — "resume."

Figure 1 — Backpressure's feedback loop. The writable's buffer absorbs the speed mismatch up to its highWaterMark; beyond that, write() returning false pauses the producer until 'drain'. Memory stays bounded no matter how mismatched the speeds.

Manual honoring — worth writing once in your life so pipe stops being magic:

function copy(src, dst) {
  src.on("data", (chunk) => {
    if (!dst.write(chunk)) {        // (1) returns false: ceiling reached
      src.pause();                  // (2) stop the flow at the SOURCE
      dst.once("drain", () => src.resume());   // (3) resume when drained
    }
  });
  src.on("end", () => dst.end());
}

Omit steps 2–3 — i.e. wire data straight to write() and ignore the return value — and the code works, in dev, with small files: write never refuses, it just buffers unboundedly, and you've rebuilt the OOM with extra steps. This is the single most common stream bug. highWaterMark tuning is the throughput/memory dial (bigger = fewer pauses, more RAM per stream × concurrent streams — the multiplication that matters on a busy server).

3. pipe vs pipeline — and why production code uses the latter

.pipe() implements section 2 automatically — pause, resume, drain, unpipe on error… except error handling is where pipe fails you: an error in any stream does not propagate to the others; un-handled, it crashes the process (3.8.5's error event rule), and even handled, the other streams in the chain aren't destroyed — leaking file descriptors and sockets. The modern spelling fixes all of it:

import { pipeline } from "stream/promises";

await pipeline(                       // [!code highlight] // the production idiom:
  fs.createReadStream("big.log"),
  zlib.createGzip(),
  uploadStream,
  { signal }                          // AbortSignal support built in (3.8.5)
);
// throws on ANY stage's failure — and destroys EVERY stage, leaking nothing

pipeline() propagates errors across the chain, destroys all streams on any failure, resolves/rejects as a promise, and accepts an AbortSignal. Rule: pipe for quick scripts; pipeline wherever failure matters — which is production, always. ^{⚑How does Node.js handle large files efficiently? [EQ-67b]}

4. Building your own: Transform, object mode, async iteration

A Transform is "a function over a stream" — implement _transform(chunk, encoding, callback):

import { Transform } from "stream";

const lineSplitter = new Transform({
  readableObjectMode: true,           // output side emits OBJECTS (lines)
  transform(chunk, _enc, cb) {
    this.tail = (this.tail ?? "") + chunk.toString("utf8");   // (StringDecoder in
    const lines = this.tail.split("\n");                       //  real code — 3.8.3)
    this.tail = lines.pop();          // keep the incomplete last line for next chunk
    for (const line of lines) this.push(line);   // push 0..n outputs per input
    cb();                             // "ready for the next chunk" — THE backpressure hook
  },
  flush(cb) {                         // end-of-input: emit the held tail
    if (this.tail) this.push(this.tail);
    cb();
  }
});

Note what the API shape gives you: calling cb() is how you ask for more — delay it and upstream pauses (backpressure participation is built into the contract); push() returning false tells you downstream is full. The held-tail logic is the same boundary discipline as 3.8.3's StringDecoder — streaming code is boundary code.

Object mode lifts streams from bytes to arbitrary values — the line-splitter above emits strings; a CSV parser emits row objects; a DB writer consumes them. highWaterMark counts objects (default 16). This is what makes streams a general dataflow tool, not just an I/O one — with the caveat that per-object overhead makes it a poor fit for millions of tiny values in hot loops (batch them).

Async iteration — the modern consumption face (3.6.6): every Readable is an async iterable, and for await honors backpressure by construction (you don't pull the next chunk until you're done with this one):

for await (const line of readable.pipe(lineSplitter)) {
  await handle(line);                 // slow handler? upstream simply waits. ✅
}

Async generators slot straight into pipeline(src, async function* (source) { … }, dst) — the generator-as-Transform pattern, often the clearest spelling of custom stages.

5. The expert lens

Backpressure is a universal law wearing a Node API. Fast producer, slow consumer, bounded buffer, feedback signal: you've now seen it as kernel pipe buffers (2.8), TCP flow-control windows (Part 5's receive window is exactly this), Node streams here, and you'll meet it as consumer lag and bounded queues in Kafka (Part 10). Any system lacking the feedback leg — "just buffer it" — has chosen the unbounded-memory failure mode and will meet it at scale. When reviewing any producer/consumer design, your first question is now permanent: where does the slow-consumer signal travel backward?

Constant memory is the difference between a service and a demo. readFile-then-process works flawlessly until real data: memory proportional to input × concurrency is a cliff, not a slope (2.5's OOM killer doesn't warn). The professional reflex: any path whose input size you don't control — uploads, exports, logs, third-party feeds — is stream-shaped by default, and "how big can this get × how many at once" is asked at design time, not incident time.

Streams are Node's composition story. Single-purpose stages — split, parse, validate, batch — glued by pipeline mirror Unix's pipe philosophy (2.8) with types (object mode) and with backpressure, which shell pipes give you free and most in-memory abstractions forget. The design payoff is the same as Unix's: stages test in isolation, recombine freely, and the memory bound holds end-to-end because every joint speaks the same protocol.

Next: the machinery under every stream — 3.8.5: EventEmitter internals, the special error event, and cancellation with AbortController.

Recall

• Four classes: Readable (source), Writable (sink), Duplex (both, independent — sockets), Transform (output computed from input — gzip, parsers). All EventEmitters; all speak one backpressure protocol.
• Backpressure mechanics: internal buffer targets highWaterMark (64 KB bytes / 16 objects); write() → false at the ceiling = pause; 'drain' = resume. Ignoring the return value = unbounded buffering = the OOM you were avoiding. for await honors it by construction.
• pipeline() over .pipe() in production: propagates errors across stages, destroys all streams on failure (no fd/socket leaks), promise-based, takes AbortSignal. pipe = scripts only.
• Custom stages: Transform with _transform(chunk, enc, cb) — cb() requests the next chunk (backpressure hook), push() emits 0..n outputs, flush drains held state (boundary discipline — tails, split characters). Object mode streams values (rows, events) with object-counted highWaterMark; async generators are the modern Transform spelling.
• Design laws: memory must not scale with input size; every producer/consumer joint needs the backward slow-consumer signal.

Self-test: Walk the three-part backpressure mechanism and the exact bug when the write() return is ignored. Why is pipeline strictly better than pipe in production — name all three wins. In a Transform, what do cb() and push()'s return value each signal? Why does for await get backpressure for free? What does highWaterMark trade against what?

Quiz Bank

FoundationalWhat are the four stream types, with a real example of each?

Readable — a chunk source: fs.createReadStream, an incoming HTTP request body, process.stdin, a database cursor. Writable — a chunk sink: fs.createWriteStream, the HTTP response, process.stdout. Duplex — independently readable and writable, two unrelated channels in one object: a TCP socket (what you read from the peer isn't what you wrote). Transform — a Duplex whose readable side is computed from its writable side: zlib.createGzip(), crypto cipher streams, CSV/line parsers — the class you implement most, because it's "a function lifted over a stream." All four are EventEmitters sharing one backpressure protocol, which is what lets arbitrary chains compose safely.

FoundationalWhat is backpressure and what exactly happens without it?

Backpressure is the feedback signal that matches a fast producer to a slow consumer through a bounded buffer: a Writable buffers incoming chunks up to its highWaterMark; at the ceiling, write() returns false ("pause"), and when the buffer drains, the 'drain' event says "resume." pipe/pipeline/for await implement the pause/resume automatically. Without it — the classic bug is wiring data events directly to write() and ignoring its return — nothing fails visibly: write still accepts every chunk, buffering them in memory without bound, so a 500 MB/s disk feeding a 1 MB/s client accumulates the difference in RAM until the OOM killer arrives (2.5). It works in dev (small files, fast local consumers) and dies in production — memory proportional to speed-mismatch × time × concurrency.

AppliedWhy should production code use pipeline() instead of .pipe()?

.pipe() handles backpressure but not failure: an error in any stage does not propagate to the other stages; an unhandled 'error' on any stream crashes the process (3.8.5), and even when you handle it, the other streams keep their resources — leaked file descriptors, sockets, and half-open handles that accumulate under retry load. pipeline() fixes the full list: errors from any stage propagate to one place (callback or rejected promise); every stage is destroyed on failure (resources freed); completion is a promise (stream/promises) that composes with async/await and try/catch; and it accepts an AbortSignal so cancellation tears the whole chain down (3.8.5). It also admits async generators as stages — often the clearest custom-Transform spelling. Rule: pipe in throwaway scripts; pipeline anywhere failure or cancellation matters — i.e., production, always.

InterviewYou implement a Transform. Explain the roles of cb() in _transform, the return value of this.push(), and flush — and connect each to backpressure or correctness.

_transform(chunk, enc, cb) is called with one upstream chunk; calling cb() is the demand signal — the stream machinery won't deliver the next chunk until you do, so a slow stage automatically pauses everything upstream (your stage participates in backpressure just by construction; delaying cb while awaiting I/O is legitimate throttling). this.push(out) emits 0..n output chunks per input; its return value false means the readable side's buffer hit its highWaterMark — downstream is full — and a well-behaved stage stops pushing until it can continue (with pipeline the machinery largely manages this, but bulk-pushing loops should check it). flush(cb) runs at end-of-input, before 'end' is emitted downstream — the place to emit held state: the incomplete last line a splitter kept, the final block a compressor buffers, StringDecoder tails (3.8.3). Forgetting flush is the classic "last record missing" bug. Together: cb = upstream demand, push's return = downstream capacity, flush = boundary correctness.

StaffDesign review: a report service builds a 400 MB CSV export by accumulating rows into an array, then res.send(csv). It OOMs at month-end. A teammate proposes doubling the container's memory. Give the senior response.

Doubling memory moves the cliff, keeps the architecture: memory still scales with report size × concurrent exports, so the next big tenant or two simultaneous month-ends re-OOMs — and meanwhile every export holds hundreds of MB hostage, degrading the whole service (2.5). The structural fix is a streaming pipeline with backpressure end to end: a DB cursor stream (every driver offers one — never SELECT * into an array), an object-mode Transform serializing rows to CSV lines, zlib.createGzip() if the client accepts it, into res — glued by pipeline(cursor, toCsv, gzip, res). Now memory is a few chunks per export regardless of row count, and — the part worth saying aloud — backpressure reaches the database: a slow client pauses the pipeline, the pipeline stops pulling the cursor, and the DB isn't racing ahead of a mobile connection. Add: Content-Disposition/chunked transfer so the browser streams to disk; pipeline's error propagation so a dropped connection destroys the cursor (no orphaned DB resources — check the driver's cancel semantics); an AbortSignal tied to req closing; and bounded export concurrency with a queue so month-end is a line, not a stampede (2.4). Frame the principle for the review: memory that scales with input size is a design bug, not a sizing problem — capacity changes buy time, streaming removes the failure class. (Then note the same shape upgrades their imports too — [the CSV-upload drill below] is this pipeline reversed.)

Flashcards

FlashFour stream classes

Readable (source) · Writable (sink) · Duplex (both, independent — socket) · Transform (output = f(input) — gzip, parsers).

FlashBackpressure mechanics

Buffer targets highWaterMark (64 KB / 16 objects) → write() returns false = pause → 'drain' = resume. Ignoring the return = unbounded buffering.

Flashpipeline() wins

Cross-stage error propagation, all stages destroyed on failure (no leaks), promise API, AbortSignal, async-generator stages. pipe = scripts only.

FlashTransform contract

cb() = give me the next chunk (upstream demand); push() false = downstream full; flush = emit held tails or lose the last record.

FlashObject mode

Streams of values (rows/events), highWaterMark counts objects (16). General dataflow tool; batch tiny values.

Flashfor await over a Readable

Every Readable is async-iterable; the loop pulls only when ready — backpressure by construction (3.6.6).

Scenario Drill

DrillAn endpoint accepts a CSV upload, transforms each row, and writes results to a database. It works in testing but crashes with out-of-memory on real 1 GB files. Redesign it with this page's machinery, end to end.

The crash comes from loading the whole file into memory: reading with fs.readFile, or accumulating the request body into one Buffer/string, materializes the entire 1 GB — plus the parsed array of row objects, plus V8 overhead — so memory is proportional to file size, and a couple of concurrent uploads exceed the container limit (2.5). Test files were small; the flaw was invisible. Redesign as a backpressured pipeline — the request is already a Readable arriving in chunks:

await pipeline(
  request,                        // Readable — the upload, chunk by chunk
  csvParser(),                    // Transform — bytes → row objects (object mode;
                                  //   holds split lines/characters across chunks — 3.8.3)
  async function* (rows) {        // Transform as async generator — the row logic,
    for await (const row of rows) //   with backpressure by construction (3.6.6)
      yield transformRow(row);
  },
  batchedDbWriter({ size: 500 }), // Writable — batches inserts; its slowness
  { signal: controller.signal }   //   PAUSES the whole chain back to the socket
);

Only a few chunks and one batch exist in memory at any moment — 1 MB or 100 GB, same footprint. Backpressure is the correctness, not the tidiness: the DB writer is the slowest stage, so its buffer fills, write() returns false, and the pause propagates stage by stage until the TCP socket itself stops being read — the kernel's receive window then throttles the client's upload (Part 5): the speed mismatch is absorbed by the network protocol, not your RAM.

pipeline (not pipe) so a mid-upload disconnect or DB failure destroys every stage — no leaked descriptors, no orphaned partial state un-noticed. Hardening: enforce a max upload size before and during streaming (reject, don't buffer); validate rows in-stream and fail fast with a clear 422 (this is a trust boundary — 3.7.7 schemas per row); batch DB writes (per-row inserts make the DB the bottleneck and hold the pipeline open longer); cap concurrent uploads with an explicit limiter so N streams can't jointly exhaust memory or the connection pool (2.4); wire the request's close to the AbortSignal; and if transformRow is CPU-heavy, move it to workers (3.8.6) so the loop keeps serving. State the principle in the PR: never let memory scale with input size — stream it, and let backpressure run the speed negotiation all the way back to the sender.