Handle Node.js Stream Backpressure

In Node.js applications, writing data to a slow writable stream faster than it can process leads to high memory usage and potential crashes, a phenomenon known as backpressure. This article explains how backpressure occurs in Node.js streams and provides practical, efficient solutions to handle it, such as using the .pipe() method, listening to the 'drain' event, and utilizing the robust stream.pipeline() utility.

Understanding Backpressure in Node.js

Backpressure occurs when the data producer (readable stream) sends data faster than the data consumer (writable stream) can process it. When this happens, Node.js buffers the incoming data in system memory. If the buffer continues to grow, it can lead to high memory consumption and eventually crash the application with an “Out of Memory” error.

To manage this, the writable.write(chunk) method returns a boolean value: * true: The internal buffer is below the highWaterMark threshold, and you can safely continue writing. * false: The buffer limit has been reached. You should stop writing data until the stream signals that it is ready.

Solution 1: Use stream.pipe()

The easiest and most common way to handle backpressure is by using the .pipe() method. This method automatically manages the data flow between the readable and writable streams. It pauses the readable stream when the writable stream’s buffer is full and resumes it once the buffer has cleared.

const fs = require('fs');

const readable = fs.createReadStream('large-file.txt');
const writable = fs.createWriteStream('destination.txt');

// pipe automatically handles backpressure
readable.pipe(writable);

Solution 2: Use stream.pipeline() for Better Error Handling

While .pipe() manages backpressure automatically, it does not handle stream errors or close streams properly if one fails. The stream.pipeline() utility solves this by handling backpressure automatically while also offering clean error handling and resource cleanup.

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

const readable = fs.createReadStream('large-file.txt');
const writable = fs.createWriteStream('destination.txt');

pipeline(readable, writable, (err) => {
  if (err) {
    console.error('Pipeline failed:', err);
  } else {
    console.log('Pipeline completed successfully.');
  }
});

Solution 3: Manual Backpressure Handling with the ‘drain’ Event

If you are generating data dynamically (for example, in a loop) and writing directly to a writable stream without a readable stream source, you must handle backpressure manually.

To do this, you must pause writing when writable.write() returns false, and wait for the 'drain' event to fire before writing more data.

const fs = require('fs');
const writer = fs.createWriteStream('output.txt');

function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // Last write, trigger the callback when finished
        writer.write(data, encoding, callback);
      } else {
        // Write data and check if we should continue (true) or wait (false)
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    
    if (i > 0) {
      // Stopped writing because of backpressure.
      // Wait for the 'drain' event to resume.
      writer.once('drain', write);
    }
  }
  
  write();
}

writeOneMillionTimes(writer, 'some data\n', 'utf-8', () => {
  console.log('Finished writing all data safely.');
});

Solution 4: Using Async Iterators (Modern Approach)

With modern Node.js, you can consume readable streams and write to writable streams using for await...of loops. This approach natively respects backpressure because the loop waits for the write operation to resolve before fetching the next chunk.

const fs = require('fs');
const { finished } = require('stream/promises');

async function writeWithAsyncIterator(readable, writable) {
  for await (const chunk of readable) {
    if (!writable.write(chunk)) {
      // Handle backpressure by waiting for the drain event
      await new Promise((resolve) => writable.once('drain', resolve));
    }
  }
  writable.end();
  await finished(writable);
}

const readable = fs.createReadStream('large-file.txt');
const writable = fs.createWriteStream('destination.txt');

writeWithAsyncIterator(readable, writable)
  .then(() => console.log('Writing finished.'))
  .catch((err) => console.error('Writing failed:', err));