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() does everything from section 2 for you: it pauses, resumes and waits for drain. But pipe lets you down on errors. An error in one stream is not passed to the others. If nothing handles it, it crashes the process (3.8.5). Even if you handle it, the other streams in the chain are not destroyed, so file descriptors and sockets leak. The modern way fixes all of this:

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 one stage is not passed to the other stages. An 'error' that nothing handles crashes the process (3.8.5). And even when you handle it, the other streams hold on to their resources. File descriptors, sockets and half-open handles leak, and they pile up when requests are retried.

pipeline() fixes all of that:

• An error from any stage arrives in one place, either a callback or a rejected promise.
• Every stage is destroyed on failure, so resources are freed.
• Finishing is a promise (from stream/promises), so it works with async/await and try/catch.
• It accepts an AbortSignal, so cancelling tears down the whole chain (3.8.5).

It also accepts async generators as stages, which is often the clearest way to write your own transform. 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 real fix is a streaming pipeline, with backpressure from end to end:

1. A database cursor stream. Every driver offers one. Never SELECT * into an array.
2. An object-mode Transform that turns each row into a CSV line.
3. zlib.createGzip(), if the client accepts it.
4. Into res.

They are joined by pipeline(cursor, toCsv, gzip, res). Now each export holds only a few chunks in memory, however many rows there are. And, the part worth saying out loud, backpressure reaches the database. A slow client pauses the pipeline, the pipeline stops pulling from the cursor, and the database is not racing ahead of someone's mobile connection.

Also add:

• Content-Disposition and chunked transfer, so the browser saves straight to disk as data arrives.
• pipeline's error handling, so a dropped connection destroys the cursor and leaves nothing open in the database. Check how your driver cancels.
• An AbortSignal connected to req closing.
• A limit on how many exports run at once, with a queue, so month-end is an orderly 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 it with fs.readFile, or collecting the request body into one Buffer or string, holds the entire 1 GB at once. On top of that come the parsed array of row objects and V8's own overhead. So memory grows with the size of the file, and two or three uploads at the same time go over the container's limit (2.5). The files tried during development were small, so nobody saw the flaw. 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 what makes it correct, not just tidy. Here is the chain:

1. The database writer is the slowest stage, so its buffer fills up.
2. write() returns false.
3. The pause travels back, stage by stage, until the TCP socket itself stops being read.
4. The kernel's receive window then slows down the client's upload (Part 5).

The difference in speed is soaked up by the network protocol, not by your memory.

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 maximum upload size, both before and during streaming. Reject, never buffer.
• Check each row as it streams past, and fail fast with a clear 422. This data comes from outside, so use a schema per row (3.7.7).
• Write to the database in batches. Inserting one row at a time makes the database the bottleneck, and keeps the pipeline open longer.
• Cap how many uploads run at once with an explicit limiter, so many streams together cannot use up memory or the connection pool (2.4).
• Connect the request's close event to the AbortSignal.
• If transformRow is heavy on CPU, 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.