Cancellable, backpressure-aware Transform streams (pattern and trade-offs)

When you accept async-iterator input and must both respect backpressure and stop promptly on cancel, treat the Transform as the single place that owns transformation logic and cancellation state. The producer (the async iterator) should write one item at a time and check write()'s boolean return. If write() returns false, await the stream's 'drain' event before continuing. That keeps the backlog bounded by Node's writable buffer, rather than by the source's speed.

Make cancellation visible: cancel() should be idempotent, record the first cancel reason, and destroy the Transform with that error so pipeline() surfaces the failure to callers. Distinguish an ordinary end (source finished) from an abort (user cancelled) — collapsing them loses the ability to assert why work stopped.

Expected behavior from this pattern:

• Input order is preserved and each chunk is transformed exactly once.
• A fast async-iterator producer cannot flood the stream because the helper waits for 'drain' when write() returns false.
• cancel() aborts the pipeline with the supplied error and prevents further output; repeated cancels keep the first error as the canonical reason.

Limitations and when not to use it:

• This pattern intentionally avoids hidden queues or parallel processing inside the Transform. If your application needs parallel transforms, retry policies, or smoothing over slow consumers, design those concerns separately.
• It does not prove memory behavior for arbitrary payload sizes or replace integration testing with real sinks/sources (files, sockets, databases). Use this pattern as a clear contract for ownership, then add and test additional features as needed.

The article shows a minimal, test-backed implementation and a small test suite that verifies the three core properties above. The provided example is object-mode focused so the control flow remains visible; adapt and integration-test for your real data shapes before production use.

Working example

import { Transform, Writable } from 'node:stream';
import { finished } from 'node:stream/promises';

function waitForDrain(stream) {
  return new Promise((resolve, reject) => {
    const cleanup = () => {
      stream.off('drain', drained);
      stream.off('error', failed);
      stream.off('close', closed);
    };
    const drained = () => { cleanup(); resolve(); };
    const failed = (err) => { cleanup(); reject(err); };
    const closed = () => failed(stream.errored || Object.assign(new Error('Stream closed before drain'), {
      code: 'ERR_STREAM_PREMATURE_CLOSE',
    }));
    stream.once('drain', drained);
    stream.once('error', failed);
    stream.once('close', closed);
    if (stream.destroyed) closed();
  });
}

export class CancellableTransform extends Transform {
  constructor(opts = {}) {
    super({ ...opts, objectMode: true });
    this._cancelled = false;
    this._cancelError = null;
  }

  cancel(err = new Error('Cancelled')) {
    if (this._cancelled) return;
    this._cancelled = true;
    this._cancelError = err;
    this.destroy(err);
  }

  async _transform(chunk, _enc, callback) {
    try {
      await Promise.resolve();
      if (this._cancelled) throw this._cancelError;
      this.push({ ...chunk, processed: true });
      callback();
    } catch (err) {
      callback(err);
    }
  }

  async runFromAsyncIterable(iterable) {
    // Observe failure immediately, even while the producer is awaiting a value.
    // Keep the result as a value until we can rethrow it to the caller.
    const completion = finished(this, { readable: false, cleanup: true }).then(
      () => null, (err) => err,
    );
    try {
      if (this._cancelled) throw this._cancelError;
      for await (const item of iterable) {
        if (this._cancelled) throw this._cancelError;
        if (this.destroyed) throw this.errored || new Error('Stream closed');
        if (!this.write(item)) {
          // A close without an error must also settle the wait for drain.
          await waitForDrain(this);
        }
      }
      this.end();
      const error = await completion;
      if (error) throw error;
    } catch (err) {
      if (!this.destroyed) this.destroy(err);
      await completion;
      throw err;
    }
  }
}

export function createCollector() {
  const items = [];
  const writable = new Writable({
    objectMode: true,
    write(chunk, _enc, callback) {
      items.push(chunk);
      callback();
    },
  });
  // pipeline() is the single completion/error promise; no second, easily
  // forgotten rejection is created by this collector.
  return { writable, items };
}

Read the full article: https://chriseugenerodriguez.com/blog/cancellable-backpressure-transform-streams

Error ownership and cancellation boundary: Attach pipeline() and its rejection handler before calling cancel(), including when testing an already-cancelled transform. The collector exposes only its writable stream and items; pipeline() is the single completion/error promise. The feeder observes writable completion before ending input and handles close as well as error while waiting for drain. Cancellation cannot interrupt arbitrary work inside a producer that is awaiting its next value; that source must also accept a cooperative abort signal when prompt interruption of its I/O is required.