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().

Bắt đầu bằng cách chạy repro.mjs được cung cấp trên phiên bản Node.js bị ảnh hưởng, sau đó kiểm tra newReadableStreamFromStreamReadable trong lib/internal/webstreams/adapters.js, đặc biệt là onData và việc xử lý hủy. Xác minh rằng việc hủy trong khi đang có một lần tiếp tục được lập lịch không còn gây ra uncaughtException chưa được bắt và nguồn vẫn được dọn dẹp một cách im lặng.