Stream pipelines, transforms, and the .pipe() error trap

The .pipe() error trap: leaks one fd per failure

Lesson 02 left you with the rule “use pipeline(), not manual pipe().” This is why, in mechanism. a.pipe(b).pipe(c) wires data flow and backpressure correctly — that part works — but it does two dangerous things by omission. First, it does not forward errors. If b emits an 'error', that error stays on b; it is not propagated to a or c. Second, it does not destroy the rest of the chain on failure. When b errors, a is left open and flowing, and c is never told to close. Whatever resources a and c hold — a file descriptor, a socket, an in-flight buffer — leak.

// THE TRAP: a malformed gzip body throws in gunzip; the socket leaks.
import zlib from "node:zlib";

function handler(req, res) {
  const gunzip = zlib.createGunzip();
  const parser = createNdjsonParser();

  req.pipe(gunzip).pipe(parser);   // ❌ no error wiring, no cleanup

  parser.on("data", (record) => save(record));
  parser.on("end", () => res.end("ok"));
  // gunzip throws on bad input → 'error' on gunzip only.
  // req (the socket) is never destroyed. One leaked fd per bad request.
}

On a single CLI run this is invisible: the process exits and the OS reclaims the fd. On a long-lived server it is fatal. Each leaked descriptor is permanent for the process lifetime, and they accumulate one per failure. A default ulimit -n is often 1024, raised to 4096 on busy boxes; once you’ve leaked that many, every new socket, file open, and accept() fails with EMFILE, and the service falls over — exactly the Tuesday staircase. Worse, an 'error' on a stream with no listener is itself an uncaughtException, so the other failure mode of the naked pipe is a hard crash. The trap has two exits and both are bad.

The fix: stream.pipeline()

stream.pipeline(...streams, callback) was added to close this exact hole. It wires the same data flow and backpressure as pipe(), and additionally: it forwards an error from any stage to one callback, and it destroys every stream in the chain — on failure and on normal completion. One place learns the outcome; nothing leaks. The promise form from node:stream/promises gives you the same guarantees behind await, which is the shape you want in modern async code.

// THE FIX: pipeline wires errors AND destroys the whole chain on any outcome.
import { pipeline } from "node:stream/promises";
import zlib from "node:zlib";

async function handler(req, res) {
  try {
    await pipeline(
      req,                       // Readable: the inbound socket
      zlib.createGunzip(),       // Transform: throws on a malformed body
      createNdjsonParser(),      // Transform: objectMode records out
      saveSink(),                // Writable: persists each record
    );
    res.end("ok");
  } catch (err) {
    // gunzip's error lands HERE. req, gunzip, parser, sink are ALL destroyed.
    res.statusCode = 400;
    res.end("bad body");
  }
}

When createGunzip() throws on the malformed body, pipeline rejects, every stream — including the inbound socket req — is destroyed, and the fd is released. No staircase, no uncaughtException. The callback form, pipeline(req, gunzip, parser, sink, (err) => …), is identical in guarantees; use it when you’re not in an async function. For a single stream rather than a chain, stream.finished(stream, cb) (or its promise form) tells you when one stream is done or has errored without re-implementing the listener bookkeeping. The rule sharpens: any time more than one stream is involved, reach for pipeline() first and never type .pipe( in server code.

Authoring a Transform: _transform, _flush, object mode

When you look at a data pipeline and none of the built-ins fit — you need to parse lines, strip fields, or re-encode records on the fly — writing your own Transform is the right answer. The rules are few, but violating any one of them causes failures that are hard to reproduce.

A Transform is a Duplex whose readable side is a function of its writable side — gzip, a cipher, a per-line parser. You author one by implementing _transform(chunk, encoding, callback): do the work, push zero or more outputs with this.push(...), then call callback() to signal you’re done with this chunk and ready for the next. An optional _flush(callback) runs once when the writable side ends — your chance to emit a trailer or whatever’s left in an internal accumulator. Set objectMode: true when chunks are records (parsed rows) rather than bytes; then this.push(obj) emits a JS object instead of a Buffer.

import { Transform } from "node:stream";

// Byte stream in → one object per NDJSON line out. objectMode on the read side.
class NdjsonParser extends Transform {
  constructor() {
    super({ readableObjectMode: true });  // bytes in, objects out
    this.tail = "";
  }
  _transform(chunk, _enc, cb) {
    const lines = (this.tail + chunk).split("\n");
    this.tail = lines.pop();             // last fragment may be a partial line
    try {
      for (const line of lines) {
        if (line) this.push(JSON.parse(line));   // emit one record per line
      }
      cb();                              // success: ready for next chunk
    } catch (err) {
      cb(err);                           // failure: propagate, do NOT throw + cb
    }
  }
  _flush(cb) {
    if (this.tail) this.push(JSON.parse(this.tail));  // last line, no newline
    cb();
  }
}

A Transform has two highWaterMarks — one per side — each at the default of 16384 bytes (16KB) for byte mode or 16 objects in object mode, and it participates in backpressure on both: it stops pulling from upstream when its readable side is full, and signals upstream to pause. Three failure modes bite here. Signal errors via cb(err), not by throwing synchronously inside an async continuation — a throw there escapes the stream machinery and becomes an uncaughtException, just like the EventEmitter trap. Never call cb twice: a double callback is a hard ERR_MULTIPLE_CALLBACK and corrupts the stream’s internal state. And the cardinal sin — a Transform that buffers everything in _flush (accumulate all chunks, emit at the end) silently defeats streaming: it holds the entire dataset in memory, so a 4GB input that should cost ~3×16KB across a three-stage pipeline instead costs 4GB and OOMs. The whole point is to push as you go.