Node.js stream.pipeline() vs .pipe(): Handling Stream Errors

A stream chain is several objects pretending to be one operation. A file read stream, a gzip transform and an HTTP response each have their own error event, their own close event and their own idea of whether they’re finished. When one of them fails, the others don’t find out unless something tells them. That something is stream.pipeline(), and the reason it exists is the gap left by the older .pipe() method.

This is a common beginner interview topic because the bug is invisible in the happy path. Every example works until a disk fills up or a client closes the browser tab.

What .pipe() leaves behind

.pipe() moves data from a readable into a writable and handles backpressure. It does not forward errors, and it does not destroy the other streams when one fails. Here’s a source that produces rows forever, piped into a destination that fails on the third write:

import { Readable, Writable } from 'node:stream';

function source(name) {
  let i = 0;
  const r = new Readable({ read() { this.push(`row ${i++}\n`); } });
  r.on('close', () => console.log(`${name}: source closed, destroyed=${r.destroyed}`));
  return r;
}

function failingDest() {
  let n = 0;
  return new Writable({
    write(chunk, enc, cb) { if (++n === 3) cb(new Error('disk full')); else cb(); },
  });
}

const s1 = source('pipe');
const d1 = failingDest();
d1.on('error', (e) => console.log('pipe: dest error:', e.message));
s1.pipe(d1);

setTimeout(() => {
  console.log(`pipe: after 100 ms source destroyed=${s1.destroyed}, readableFlowing=${s1.readableFlowing}, buffered=${s1.readableLength}`);
}, 100);

On Node 22:

pipe: dest error: disk full
pipe: after 100 ms source destroyed=false, readableFlowing=false, buffered=65544

The destination failed and the source is still open, paused, with 64 KB sitting in its buffer. If that source were a file, the file descriptor would stay open. If it were a database cursor, the cursor would hold a connection. .pipe() unpiped the destination when it errored and left you to clean up the rest. Note also that the error handler was mandatory: without d1.on('error', ...), the disk full error would be an unhandled 'error' event and crash the process.

To do this correctly with .pipe() you’d attach an error handler to every stream and destroy every other stream from each one. That’s exactly the bookkeeping pipeline() does.

What pipeline() does instead

Same source, same failing destination:

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

const s2 = source('pipeline');
try {
  await pipeline(s2, failingDest());
} catch (e) {
  console.log('pipeline rejected:', e.message, '| source destroyed =', s2.destroyed);
}

pipeline: source closed, destroyed=true
pipeline rejected: disk full | source destroyed = true

Three things happened. Every stream in the chain was destroyed, so the source closed its resource. The promise rejected with the original error, not a wrapper. And there was one place to catch it. The promise resolves only when the last stream has finished, which for a file write stream means the data has been flushed and the file closed.

There are two forms. require('node:stream').pipeline(a, b, c, callback) is the original callback version, and require('node:stream/promises').pipeline(a, b, c) returns a promise. Use the promise form in new code. A mistake interviewers sometimes probe for is treating the return value of .pipe() as completion: src.pipe(dest) returns dest immediately, so code after it runs before a single byte has moved.

pipeline also accepts async generator functions as stages, which is often simpler than writing a Transform class:

await pipeline(
  src,
  async function* (rows) { for await (const r of rows) yield r.toString().toUpperCase(); },
  sink,
  { signal: AbortSignal.timeout(300) },
);

The final options object takes an AbortSignal. With a source that emits a row every 50 ms, the 300 ms timeout aborted the chain:

AbortError ABORT_ERR | source destroyed = true

That’s how you cancel a stream chain from outside: a request deadline, a shutdown signal, or a user pressing cancel.

A gzip export over HTTP when the client leaves

Here’s the case from the original brief: a server streams a large CSV export through gzip to an HTTP response. The generator stands in for a report query that produces 32 KB rows.

import http from 'node:http';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';
import { randomBytes } from 'node:crypto';

function reportRows() {
  let n = 0;
  const r = Readable.from((async function* () {
    while (true) { n++; yield randomBytes(16 * 1024).toString('hex') + '\n'; }
  })());
  r.on('close', () => console.log(`server: report source closed after ${n} chunks`));
  return r;
}

const server = http.createServer(async (req, res) => {
  res.setHeader('Content-Type', 'text/csv');
  res.setHeader('Content-Encoding', 'gzip');
  try {
    await pipeline(reportRows(), createGzip(), res);
    console.log('server: export finished');
  } catch (err) {
    console.log(`server: pipeline rejected: ${err.code} ${err.message} | headersSent=${res.headersSent}`);
  }
});

A test client reads about 200 KB and then destroys its request, like a user closing the tab:

client: got 204234 bytes, disconnecting
server: report source closed after 15 chunks
server: pipeline rejected: ERR_STREAM_PREMATURE_CLOSE Premature close | headersSent=true

The response closed before it finished, pipeline noticed, and it destroyed the gzip stream and the generator. The source stopped after 15 chunks. Two points are worth saying out loud in an interview. First, the source only produced 15 chunks at all because pipeline respects backpressure: the generator isn’t pulled faster than the socket drains. Second, ERR_STREAM_PREMATURE_CLOSE here is not a server bug. It’s a client that went away, so log it at a low level or count it, rather than paging someone.

The response you can’t send anymore

The harder half of error ownership is deciding what to tell the client. Suppose the report query fails after two rows have been sent:

const server = http.createServer(async (req, res) => {
  const rows = Readable.from((async function* () {
    yield 'id,amount\n'; yield '1,10\n';
    throw new Error('cursor lost connection to database');
  })());
  try {
    await pipeline(rows, res);
  } catch (err) {
    console.log(`server: ${err.message} | headersSent=${res.headersSent} destroyed=${res.destroyed}`);
    try { res.writeHead(500); } catch (e) { console.log('server: writeHead threw', e.code); }
  }
});

server: cursor lost connection to database | headersSent=true destroyed=true
server: writeHead threw ERR_HTTP_HEADERS_SENT
client: status 200
client: response error ECONNRESET aborted
client: closed, complete=false, body="id,amount\n1,10\n"

The client already received 200 OK. There’s no way to take that back, and the attempt to send a 500 throws ERR_HTTP_HEADERS_SENT. What pipeline did do is destroy the response, so the connection was reset instead of ending cleanly. That matters: with chunked encoding, a clean end would make a truncated CSV look complete. A reset lets the client detect the failure (complete=false). If you want a real error status for failures, you have to detect them before the first byte, for example by running the query and fetching the first page before calling writeHead. Express users meet the same thing as the “Cannot set headers after they are sent” error.

Partial files on disk

When the destination is a file, a failure leaves a file behind. A validating transform that rejects a bad row:

const out = 'output.csv.gz';
try {
  await pipeline(createReadStream('input.csv'), validate, createGzip(), createWriteStream(out));
} catch (err) {
  console.log('export failed:', err.message);
  console.log('partial file exists:', await stat(out).then((s) => `yes, ${s.size} bytes`, () => 'no'));
  await rm(out, { force: true });
}

export failed: bad amount: oops
partial file exists: yes, 0 bytes

createWriteStream creates the file as soon as it opens, so even a failure on the first chunk leaves an empty output.csv.gz that a downstream job might pick up. Either delete it in the catch, or write to output.csv.gz.tmp and rename it into place only after pipeline resolves. The rename approach is better when another process watches the directory, because the final name never refers to an incomplete file.

Interview practice

A colleague’s export endpoint uses fs.createReadStream(path).pipe(zlib.createGzip()).pipe(res). In production, the process slowly accumulates open file descriptors until it hits EMFILE: too many open files. Requests mostly succeed. What’s going on, and how would you fix it?

Answer

.pipe() doesn’t propagate errors or closure backward. When a client disconnects mid-download, res closes, but the gzip stream and the file read stream are never destroyed. The read stream stays paused with its file descriptor open, so every abandoned download leaks one descriptor. Successful downloads end normally, which is why most requests look fine.

The fix is await pipeline(fs.createReadStream(path), zlib.createGzip(), res) inside a try. On any failure, including ERR_STREAM_PREMATURE_CLOSE from a client disconnect, every stream is destroyed and the descriptor is closed. In the catch, I’d log client disconnects quietly and other errors loudly, and not try to send an error response if res.headersSent is true. To verify, I’d write a test that disconnects after the first chunk and assert that the read stream’s close event fires.

Further reading

• Node.js interview questions
• Node.js MCQs
• Node.js streams and backpressure
• Response compression with zlib
• Official docs: stream.pipeline() in the Node.js docs