stream: fix pipeTo assert when writer is released during microtask

The deferred write introduced by commit 65aa8f6 ("stream: fix
pipeTo to defer writes per WHATWG spec") wraps the write operation in
queueMicrotask(). A race condition exists where the pipe can shutdown
and release the writer between when [kChunk] schedules the microtask
and when it executes, causing an ERR_INTERNAL_ASSERTION because
writer[[stream]] is undefined.

Fix by moving the shuttingDown flag into the shared pipe state object
so that PipeToReadableStreamReadRequest can check it. The microtask
now skips the write if the pipe has already begun shutting down.

This aligns with the WHATWG Streams spec step 15 which states:
"if shuttingDown becomes true, the user agent must not initiate
further reads from reader, and must only perform writes of
already-read chunks".

Fixes: #63732
Signed-off-by: Qingyu Wang <wangqingyu.c0l1n@bytedance.com>

Author: colinaaa

lib/internal/webstreams/readablestream.js

@@ -1424,8 +1424,6 @@ function readableStreamPipeTo(
 
   source[kState].disturbed = true;
 
-  let shuttingDown = false;
-
   if (signal !== undefined) {
     try {
       validateAbortSignal(signal, 'options.signal');
@@ -1438,6 +1436,7 @@ function readableStreamPipeTo(
 
   const state = {
     currentWrite: PromiseResolve(),
+    shuttingDown: false,
   };
 
   // The error here can be undefined. The rejected arg
@@ -1462,8 +1461,8 @@ function readableStreamPipeTo(
   }
 
   function shutdownWithAnAction(action, rejected, originalError) {
-    if (shuttingDown) return;
-    shuttingDown = true;
+    if (state.shuttingDown) return;
+    state.shuttingDown = true;
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
@@ -1483,8 +1482,8 @@ function readableStreamPipeTo(
   }
 
   function shutdown(rejected, error) {
-    if (shuttingDown) return;
-    shuttingDown = true;
+    if (state.shuttingDown) return;
+    state.shuttingDown = true;
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
@@ -1546,11 +1545,11 @@ function readableStreamPipeTo(
   }
 
   async function step() {
-    if (shuttingDown) return true;
+    if (state.shuttingDown) return true;
 
     if (dest[kState].backpressure) {
       await writer[kState].ready.promise;
-      if (shuttingDown) return true;
+      if (state.shuttingDown) return true;
     }
 
     const controller = source[kState].controller;
@@ -1563,7 +1562,7 @@ function readableStreamPipeTo(
         controller[kState].queue.length > 0) {
 
       while (controller[kState].queue.length > 0) {
-        if (shuttingDown) return true;
+        if (state.shuttingDown) return true;
 
         const chunk = dequeueValue(controller);
 
@@ -1678,6 +1677,10 @@ class PipeToReadableStreamReadRequest {
     // synchronous write during enqueue(). See WHATWG Streams spec
     // "ReadableStreamPipeTo" step 15's "chunk steps".
     queueMicrotask(() => {
+      if (this.state.shuttingDown) {
+        this.promise.resolve(false);
+        return;
+      }
       this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
       markPromiseAsHandled(this.state.currentWrite);
       this.promise.resolve(false);

test/parallel/test-webstreams-pipeto-writer-released-race.js

@@ -0,0 +1,35 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { ReadableStream, WritableStream } = require('stream/web');
+
+{
+  let sourceController;
+  let destController;
+
+  const source = new ReadableStream({
+    start(controller) {
+      sourceController = controller;
+    },
+  });
+
+  const dest = new WritableStream({
+    start(controller) {
+      destController = controller;
+    },
+    write() {},
+  });
+
+  source.pipeTo(dest, { preventCancel: true }).then(
+    common.mustNotCall('pipeTo should not resolve'),
+    common.mustCall((err) => {
+      assert.strictEqual(err.message, 'destination errored');
+    })
+  );
+
+  setImmediate(common.mustCall(() => {
+    destController.error(new Error('destination errored'));
+    sourceController.enqueue('chunk');
+  }));
+}