Node.js streams in practice: pipeline(), backpressure and async generators

Most Node.js memory problems in small services start the same way: a route or a script reads a whole file, response or database export into memory, transforms it, and writes it out. It works with test data and then fails in production when the file is 2 GB instead of 2 MB. Streams solve this by moving data through your code in small chunks, so memory use stays roughly flat whatever the input size.

The streams API has a reputation for being difficult, mostly because of older patterns built on events and .pipe(). Modern Node.js gives you a much simpler path: pipeline() from node:stream/promises, combined with async generator functions. This tutorial shows that pattern on a realistic task: filtering a large newline-delimited JSON log, writing a compressed CSV, and serving it over HTTP.

You will learn what backpressure is and why it matters, how to write transformation steps as plain async function* functions, how to cancel a pipeline with an AbortSignal, and which mistakes to avoid. The code was run on Node.js 24 on 7 October 2026 and uses only built-in modules.

Why streams: the problem with reading everything

Here is the version most of us write first:

// Anti-pattern: reads the whole file into memory first
import { readFile, writeFile } from 'node:fs/promises';

const text = await readFile('requests.ndjson', 'utf8');
const rows = text.split('\n').filter(Boolean).map((line) => JSON.parse(line));
await writeFile('out.json', JSON.stringify(rows));

This holds the raw text, the array of lines and the array of parsed objects in memory at the same time. Memory grows with the file, and a large enough input crashes the process. The streaming version below reads, transforms and writes one chunk at a time.

Backpressure in one paragraph

A stream has an internal buffer. The highWaterMark option sets how much it buffers before it signals "stop": for byte streams that is a number of bytes, for object-mode streams a number of objects. When a writable stream's buffer reaches the threshold, write() returns false. A well-behaved producer then waits for the 'drain' event before writing more. This flow control is called backpressure, and it is what keeps a fast reader (a local disk) from overwhelming a slow writer (a client on a mobile network).

The Node.js documentation is explicit that highWaterMark is a threshold, not a hard memory limit. You can check the defaults on your platform:

import { getDefaultHighWaterMark } from 'node:stream';

console.log(getDefaultHighWaterMark(false)); // byte streams: 65536 on Linux and macOS
console.log(getDefaultHighWaterMark(true));  // object mode: 16 objects

According to the v24 documentation, the default for byte streams is 65536 bytes (64 KiB) on non-Windows platforms and 16384 bytes (16 KiB) on Windows; Node.js 22 raised it from the older value.

Step 1: generate a large test file while respecting backpressure

To have something worth streaming, this script writes one million log records. It is also the clearest example of manual backpressure: check the return value of write() and await 'drain' when it is false.

// make-sample.js
import { createWriteStream } from 'node:fs';
import { once } from 'node:events';

const out = createWriteStream('requests.ndjson');
const levels = ['info', 'info', 'info', 'warn', 'error'];

for (let i = 0; i < 1_000_000; i++) {
  const line = JSON.stringify({
    time: new Date(Date.UTC(2026, 9, 7, 0, 0, i % 86400)).toISOString(),
    level: levels[i % levels.length],
    path: `/api/items/${i % 500}`,
    ms: i % 900,
  }) + '\n';

  // write() returns false when the internal buffer is above highWaterMark.
  if (!out.write(line)) {
    await once(out, 'drain'); // wait instead of buffering a million lines in memory
  }
}
out.end();
await once(out, 'finish');

Without the if (!out.write(line)) check, the loop would finish almost instantly and leave every line queued in memory, waiting for the disk. With it, the loop pauses whenever the buffer is full.

Most of the time you will not write this by hand, because pipeline() does it for you.

Step 2: a pipeline of async generators

pipeline() from node:stream/promises connects a source, any number of transforms and a destination. It returns a promise that resolves when everything has finished, and rejects if any step fails. When a step fails, the other streams are destroyed, so file handles and sockets are not left open.

A transform step can be an async generator function. It receives the previous step's output as an async iterable and yields values for the next step. That means you can write stream transformations as ordinary loops:

// slow-errors.js
import { createReadStream, createWriteStream } from 'node:fs';
import { createInterface } from 'node:readline';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';

const [input = 'requests.ndjson', output = 'slow-errors.csv.gz'] = process.argv.slice(2);

// Stage 1: turn a byte stream into lines.
async function* toLines(source) {
  const rl = createInterface({ input: source, crlfDelay: Infinity });
  for await (const line of rl) {
    if (line) yield line;
  }
}

// Stage 2: keep only slow error records, emit CSV rows.
async function* slowErrorsAsCsv(lines) {
  yield 'time,path,ms\n';
  for await (const line of lines) {
    const record = JSON.parse(line);
    if (record.level === 'error' && record.ms >= 500) {
      yield `${record.time},${record.path},${record.ms}\n`;
    }
  }
}

const started = performance.now();
await pipeline(
  createReadStream(input),
  toLines,
  slowErrorsAsCsv,
  createGzip(),
  createWriteStream(output),
  { signal: AbortSignal.timeout(60_000) }, // give up after 60 seconds
);
const mb = Math.round(process.memoryUsage().rss / 1024 / 1024);
console.log(`Done in ${Math.round(performance.now() - started)} ms, RSS ${mb} MB`);

Walking through it:

1. createReadStream(input) produces chunks of bytes.
2. toLines uses readline.createInterface() to turn those chunks into complete lines. crlfDelay: Infinity treats Windows line endings (\r\n) as a single line break.
3. slowErrorsAsCsv parses each line, keeps slow error records and yields CSV rows. Because it is a generator, it only processes the next line when the next step asks for more data, so backpressure flows all the way back to the file read.
4. createGzip() compresses the CSV text.
5. createWriteStream(output) writes the compressed bytes.

Run both scripts:

node make-sample.js          # writes requests.ndjson (about 80 MB)
node slow-errors.js          # writes slow-errors.csv.gz
gunzip -c slow-errors.csv.gz | head -3

Only a few chunks are in flight at any moment, so memory use depends on the buffer sizes and the Node.js runtime itself rather than on the size of the input file. To confirm this on your own hardware, double or triple the number of records in make-sample.js and compare the RSS figure that the script prints with process.memoryUsage().

Step 3: timeouts and cancellation with AbortSignal

The last argument to pipeline() can be an options object with a signal. When the signal aborts, the pipeline destroys its streams and the promise rejects with an AbortError. AbortSignal.timeout(60_000) creates a signal that aborts after 60 seconds, which is a simple guard against a job that hangs on a stuck network share.

The same mechanism lets you cancel work when a user goes away. For example, an HTTP handler can create an AbortController and call abort() when the request's 'close' event fires before the response has finished.

Async generator steps also receive the signal: the documentation shows a transform written as async function* (source, { signal }), which you can pass on to fetch() or database calls inside the loop.

Step 4: stream a file to an HTTP client

An HTTP response is a writable stream, so the same function serves downloads without loading the file into memory:

// serve.js
import { createReadStream } from 'node:fs';
import { createServer } from 'node:http';
import { pipeline } from 'node:stream/promises';

createServer(async (req, res) => {
  res.writeHead(200, {
    'content-type': 'text/csv; charset=utf-8',
    'content-encoding': 'gzip',
  });
  try {
    // pipeline respects backpressure from slow clients and destroys
    // the file stream if the client disconnects part-way.
    await pipeline(createReadStream('slow-errors.csv.gz'), res);
  } catch (err) {
    console.error('download aborted:', err.code);
  }
}).listen(3000);

Two details matter here. First, pipeline() respects backpressure from slow clients: if a client reads slowly, the file is read slowly too. Second, if the client disconnects part-way, the pipeline rejects (we saw ERR_STREAM_PREMATURE_CLOSE in testing) and the file stream is closed. Catch that rejection; otherwise it becomes an unhandled promise rejection.

Because the file on disk is already gzip-compressed, the handler sends it as-is with content-encoding: gzip instead of compressing it again for every request.

Common mistakes and how to avoid them

One more pattern to recognise:

// Anti-pattern: .pipe() does not forward errors or clean up
createReadStream('in.log').pipe(createGzip()).pipe(createWriteStream('in.log.gz'));

If the write stream fails (a full disk, a permission error), the read stream keeps its file handle open and nothing tells your code. The pipeline() equivalent is one line longer and handles both problems.

Where streams fit in real projects

• Exports and reports: stream database rows into CSV and straight into an HTTP response or object storage upload.
• Imports: parse uploaded CSV or NDJSON files line by line and insert in batches.
• Log processing: filter and aggregate large log files, as in this tutorial.
• Proxies and uploads: pass request bodies to another service without buffering them.

The pipeline() function also accepts web streams (ReadableStream, WritableStream, TransformStream) since Node.js 18.16, which helps when you combine Node streams with the body of a fetch() response.

Summary

• Use streams whenever input size is not under your control.
• Prefer pipeline() from node:stream/promises over .pipe(): it propagates errors, cleans up and returns a promise.
• Write transforms as async function* generators; yield results one at a time.
• Respect backpressure: check write() or let pipeline() manage it.
• Add an AbortSignal for timeouts and cancellation, and always catch the pipeline promise.

Next step: find one script or route in your code base that calls readFile() on user-sized data and rewrite it as a pipeline.