Readable.toWeb(): uncaughtException (ERR_INVALID_STATE "Controller is already closed") when the stream is canceled during backpressure resume

Version

v22.23.1 (also affects main — the unguarded code is still present)

Platform

Linux (Docker, x86_64) — also reproduced on Darwin 25.5.0 arm64

Subsystem

stream, webstreams

What steps will reproduce the bug?

Convert a Node Readable (e.g. a PassThrough) to a web ReadableStream with Readable.toWeb(), pipe it to a slow sink so backpressure pause/resume cycles occur, and cancel/abort mid-transfer (as happens whenever an HTTP client disconnects while a server streams a response built from new Response(Readable.toWeb(nodeStream))):

// repro.mjs — node repro.mjs
import { PassThrough, Readable, pipeline as pump } from "node:stream";

let uncaught = 0;
process.on("uncaughtExceptionMonitor", (e) => { if (++uncaught === 1) console.log(e.stack); });
process.on("uncaughtException", () => {});

async function one() {
  const src = new Readable({
    read() {
      this.push(Buffer.alloc(16 * 1024, 1));
      if ((this.bytes = (this.bytes || 0) + 16384) > 512 * 1024) this.push(null);
    },
  });
  const pt = new PassThrough({ highWaterMark: 16384 });
  pump(src, pt, () => {});
  const web = Readable.toWeb(pt);

  const ac = new AbortController();
  const writer = new WritableStream(
    { async write() { await new Promise((r) => setTimeout(r, 1)); } },  // slow sink → backpressure
    { highWaterMark: 1 },
  );
  setTimeout(() => ac.abort(), Math.floor(Math.random() * 12));          // client disconnect
  try { await web.pipeTo(writer, { signal: ac.signal }); } catch {}
  await new Promise((r) => setImmediate(r));
}

for (let i = 0; i < 400; i++) await one();
console.log(`uncaught=${uncaught}/400 node=${process.version}`);

How often does it reproduce? Is there a required condition?

400/400 iterations on v22.23.1 with the harness above. Required conditions: the source is flowing under backpressure (pause → pull() → resume() cycles) and the web stream is canceled (directly or via pipeTo abort) while a resume tick is already scheduled.

What is the expected behavior? Why is that the expected behavior?

Canceling a ReadableStream produced by Readable.toWeb() should tear the pipeline down quietly. A consumer abort is a normal event; it should never escalate to an uncaughtException the user cannot catch.

What do you see instead?

TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed
    at ReadableStreamDefaultController.enqueue (node:internal/webstreams/readablestream:1077:13)
    at PassThrough.onData (node:internal/webstreams/adapters:468:16)
    at PassThrough.emit (node:events:519:28)
    at Readable.read (node:internal/streams/readable:782:10)
    at flow (node:internal/streams/readable:1283:53)
    at resume_ (node:internal/streams/readable:1262:3)
    at process.processTicksAndRejections (node:internal/process/task_queues:89:21)

The throw happens synchronously inside an EventEmitter 'data' callback, so it surfaces as an uncaughtException — unreachable by any user-level try/catch.

Additional information

Root cause in lib/internal/webstreams/adapters.js (newReadableStreamFromStreamReadable):

function onData(chunk) {
  // Copy the Buffer to detach it from the pool.
  if (Buffer.isBuffer(chunk) && !objectMode)
    chunk = new Uint8Array(chunk);
  controller.enqueue(chunk);        // ← no closed/canceled guard
  if (controller.desiredSize <= 0)
    streamReadable.pause();
}

Per spec, ReadableStreamCancel sets the stream state to "closed" before invoking the underlying source's cancel() (which destroys the Node stream), and the 'data' listener is never removed on cancel. A resume_-scheduled flow tick can therefore deliver one more 'data' event after the controller is closed but before destroy() takes effect, and enqueue throws.

Suggested fix: guard onData with wasCanceled (and/or wrap the enqueue in try/catch → destroy the source), or remove the 'data' listener in cancel().

Real-world impact: any server doing new Response(Readable.toWeb(nodeStream)) (e.g. a Next.js route handler proxying object-storage bytes) hits this whenever a browser aborts mid-download — <video> seek/probe aborts made this fire on every deploy cutover for us. Note the same unguarded pattern is reachable from Next.js's own action-handler.ts (Readable.toWeb(sizeLimitedBody)).

Workaround: use pull-based ReadableStream.from(nodeReadable) instead of Readable.toWeb() — 0/400 with the same harness, and cancel still propagates destroy to the source via the async iterator's return().

Führe zunächst die bereitgestellte repro.mjs auf der betroffenen Node.js-Version aus und untersuche anschließend newReadableStreamFromStreamReadable in lib/internal/webstreams/adapters.js, insbesondere onData und die Behandlung von Abbrüchen. Überprüfe, dass ein Abbruch während einer geplanten Fortsetzung keine ungefangene uncaughtException mehr auslöst und dass die Quelle weiterhin unbemerkt heruntergefahren wird.