Node.js Stream piping 에러 해결

Introduction

Node.js streams are powerful for handling large amounts of data, but they come with their own set of error handling challenges. Improperly handled stream pipes can lead to memory leaks, data corruption, and unhandled exceptions. This post covers the most common stream piping errors and their solutions.

Environment

node --version
# v20.11.0

# Testing with large files
ls -lh large-dataset.csv
# -rw-r--r-- 1 user user 2.1G Jul 10 10:30 large-dataset.csv

Problem

Here is a typical file copy operation that fails with stream errors:

const fs = require('fs');

const readStream = fs.createReadStream('large-dataset.csv');
const writeStream = fs.createWriteStream('copy-dataset.csv');

readStream.pipe(writeStream);

After processing about 500MB, the process crashes with:

Error: read EPIPE
    at afterReadDispatch [as original] (node:internal/stream_base_commons:98:11)
    at onStreamRead (node:internal/stream_base_commons:241:5)

Error [ERR_STREAM_WRITE_AFTER_END]: write after end
    at WriteStream.write (node:internal/streams/writable:493:15)
    at ReadStream.ondata (node:internal/streams/readable:758:22)

Analysis

Stream piping errors occur for several reasons:

1. The destination stream closes before the source finishes (EPIPE)
2. Writing to a stream that has already ended (ERR_STREAM_WRITE_AFTER_END)
3. Backpressure not handled properly - writing faster than the destination can handle
4. Error events not caught on either stream

The pipe lifecycle looks like this:

Readable → pipe() → Writable
         ↓
    backpressure signal
         ↓
    pauses reading
         ↓
    drains write buffer
         ↓
    resumes reading

If any step in this lifecycle fails without proper error handling, the process crashes.

Solution

Solution 1: Proper error handling with pipe

const fs = require('fs');

const readStream = fs.createReadStream('large-dataset.csv');
const writeStream = fs.createWriteStream('copy-dataset.csv');

readStream.on('error', (err) => {
  console.error('Read error:', err.message);
  writeStream.destroy(err);
});

writeStream.on('error', (err) => {
  console.error('Write error:', err.message);
  readStream.destroy(err);
});

writeStream.on('finish', () => {
  console.log('Copy complete');
});

readStream.pipe(writeStream);

Solution 2: Using pipeline() (recommended for Node.js 10+)

const { pipeline } = require('stream');
const fs = require('fs');

const readStream = fs.createReadStream('large-dataset.csv');
const writeStream = fs.createWriteStream('copy-dataset.csv');

pipeline(
  readStream,
  writeStream,
  (err) => {
    if (err) {
      console.error('Pipeline failed:', err.message);
    } else {
      console.log('Pipeline succeeded');
    }
  }
);

Solution 3: Promisified pipeline

const { pipeline } = require('stream');
const { promisify } = require('util');
const fs = require('fs');

const pipelineAsync = promisify(pipeline);

async function copyFile(src, dest) {
  const readStream = fs.createReadStream(src);
  const writeStream = fs.createWriteStream(dest);
  
  try {
    await pipelineAsync(readStream, writeStream);
    console.log('File copied successfully');
  } catch (err) {
    console.error('Copy failed:', err.message);
  }
}

copyFile('large-dataset.csv', 'copy-dataset.csv');

Solution 4: Transform streams with error handling

const { Transform, pipeline } = require('stream');

const csvParser = new Transform({
  transform(chunk, encoding, callback) {
    try {
      const lines = chunk.toString().split('\n');
      const parsed = lines
        .filter(line => line.trim())
        .map(line => {
          const [name, email] = line.split(',');
          return JSON.stringify({ name: name?.trim(), email: email?.trim() });
        })
        .join('\n');
      
      callback(null, parsed + '\n');
    } catch (err) {
      callback(err);
    }
  }
});

pipeline(
  fs.createReadStream('contacts.csv'),
  csvParser,
  fs.createWriteStream('contacts.jsonl'),
  (err) => {
    if (err) console.error('Transform failed:', err);
    else console.log('Transform complete');
  }
);

Solution 5: Handling backpressure explicitly

const readable = getLargeDataStream();
const writable = fs.createWriteStream('output.bin');

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);
  
  if (!canContinue) {
    readable.pause();
    writable.once('drain', () => {
      readable.resume();
    });
  }
});

readable.on('end', () => {
  writable.end();
});

Lessons Learned

1. Always use pipeline() instead of .pipe() - It handles cleanup and error propagation automatically
2. Never ignore stream errors - Use .on('error', handler) on every stream
3. Destroy streams on error - Call .destroy(err) to clean up resources
4. Test with large files - Stream issues often only appear with data larger than the internal buffer (typically 16KB)
5. Monitor memory usage - Improper backpressure handling causes memory to grow unbounded

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.