Node.js Streams skill for Codex

Done well means: memory stays flat no matter the input size, every stream chain is built with pipeline so one error destroys all stages and rejects one promise, producers respect backpressure, and cancellation (client disconnect, timeout) tears the chain down.

Before changing anything

1. Read AGENTS.md and package.json (engines.node, ESM or CommonJS). The patterns here need Node 18+; Node 20+ is assumed for Readable.fromWeb stability and stream.compose.
2. Find the current data path: rg -n "readFileSync|readFile\(|\.pipe\(|arrayBuffer\(\)|Buffer\.concat" src.
3. Measure first. Run the job with a production-sized input and record peak memory: /usr/bin/time -l node job.js on macOS, /usr/bin/time -v on Linux (max RSS), or log process.memoryUsage().rss periodically.

Decision table

Situation	Use
File or stream through transforms to a sink	await pipeline(src, ...transforms, sink)
Per-record logic (parse, filter, map)	Async generator function as a pipeline stage
Need a reusable transform with state and flush	Transform subclass or new Transform({ transform, flush })
Consume a stream in application code	for await (const chunk of readable)
fetch response body into Node APIs	Readable.fromWeb(res.body)
Return a Node stream from a Web-style handler	Readable.toWeb(nodeStream)
Small file, known bounded size	fs.readFile is fine; do not stream for its own sake

The default shape

import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";

await pipeline(
  createReadStream("events.ndjson", { encoding: "utf8" }),
  splitLines,                       // async generator stage
  async function* (lines) {
    for await (const line of lines) {
      if (!line) continue;
      const event = JSON.parse(line);
      if (event.type === "purchase") yield JSON.stringify(event) + "\n";
    }
  },
  createGzip(),
  createWriteStream("purchases.ndjson.gz"),
);

async function* splitLines(source) {
  let rest = "";
  for await (const chunk of source) {
    const parts = (rest + chunk).split("\n");
    rest = parts.pop();
    yield* parts;
  }
  if (rest) yield rest;
}

For text, pass encoding: "utf8" (or call setEncoding) so Node decodes across chunk boundaries, or decode with a TextDecoder and { stream: true }; concatenating raw Buffer chunks as strings breaks multi-byte characters split across chunk boundaries. For line reading alone, readline.createInterface({ input, crlfDelay: Infinity }) is also fine.

Why pipeline and not .pipe():

• .pipe() does not forward errors. If the source errors, the destination stays open and file descriptors leak; if the destination errors, the source keeps reading.
• pipeline destroys every stage on failure, waits for the last stage to finish, and rejects with the first error.
• Pass { signal } as the last argument to cancel: await pipeline(a, b, c, { signal }).

Backpressure when writing by hand

import { once } from "node:events";

async function writeAll(rows, out) {
  for await (const row of rows) {
    if (!out.write(serialize(row))) {
      await once(out, "drain");        // buffer is above highWaterMark: wait
    }
  }
  out.end();
  await once(out, "finish");
}

Ignoring the false from write() buffers everything in memory: the classic cause of an OOM in an export job. Better still, express the producer as an async generator and let pipeline handle it.

highWaterMark and objectMode

Stream type	Default highWaterMark	Unit
Byte streams (Node 22+)	64 KiB	bytes
Byte streams (Node 20 and earlier)	16 KiB (fs read streams 64 KiB)	bytes
objectMode: true	16	objects

• Raise it (e.g. 1 MiB) for high-throughput file copies; lower it for objects that are large.
• In objectMode the limit counts objects, not bytes: 16 rows of 10 MB each is 160 MB.
• Generators used as pipeline stages are object-mode-like; yield bounded chunks.

HTTP

Stream a large download to the client without buffering, and stop work when the client goes away:

import { pipeline } from "node:stream/promises";

server.on("request", async (req, res) => {
  res.setHeader("Content-Type", "text/csv");
  try {
    await pipeline(queryRowsAsCsv(db, req), res); // generator yielding CSV lines
  } catch (err) {
    if (err.code !== "ERR_STREAM_PREMATURE_CLOSE") logger.error(err);
  }
});

When the client disconnects, res closes, pipeline destroys the generator, and its finally block runs: release database cursors there.

Proxy an upstream fetch without buffering:

const upstream = await fetch(url, { signal: AbortSignal.timeout(30_000) });
if (!upstream.ok || !upstream.body) throw new Error(`upstream ${upstream.status}`);
await pipeline(Readable.fromWeb(upstream.body), res);

Uploading a Node stream with fetch requires duplex: "half" in the request init: fetch(url, { method: "PUT", body: Readable.toWeb(fileStream), duplex: "half" }).

Writing a Transform

import { Transform } from "node:stream";

export function countBytes(onDone) {
  let total = 0;
  return new Transform({
    transform(chunk, _enc, cb) {
      total += chunk.length;
      cb(null, chunk);
    },
    flush(cb) {
      onDone(total);
      cb();
    },
  });
}

Always call the callback exactly once per chunk, and pass errors to it (cb(err)) rather than throwing; a thrown error inside transform escapes as an uncaught exception.

Testing

• Build inputs with Readable.from(["a\nb", "\nc"]) to hit chunk-boundary cases.
• Collect output with await Array.fromAsync(readable) (Node 22+) or a for await loop.
• Test errors: a source that throws mid-stream must make pipeline reject and must close the sink (assert the file handle or mock socket was destroyed).
• Test memory: run with --max-old-space-size=64 against a file larger than that; a streaming implementation still passes.

Checklist

• Every chain uses pipeline from node:stream/promises
• No readFile, Buffer.concat or await res.text() on unbounded input
• Hand-written producers await drain
• Cancellation wired (signal, client disconnect) and cleanup in finally
• Peak RSS measured before and after on a production-sized input

Gotchas

• Attaching a data listener switches a stream to flowing mode and ignores backpressure unless you pause(); prefer async iteration.
• Mixing for await on a stream with .on("data") on the same stream loses chunks.
• Readable.from("big string") or Readable.from(buffer) emits one single chunk; pass an array or generator when a test needs real chunk boundaries.
• zlib streams are CPU-bound; compressing many large files in parallel saturates the libuv thread pool (default 4, UV_THREADPOOL_SIZE to change).