stream: narrow pipeTo guard to writer release check

The state.shuttingDown guard introduced in the previous commit is too
broad: it skips writes even when the destination is still writable.
The WHATWG Streams spec requires that already-read chunks are written
to a still-writable destination during shutdown (step 15, shutdown
substeps 3.1-3.2).

Replace the guard with a check for writer[kState].stream === undefined
which is only true after finalize() has released the writer. This
precisely targets the crash condition without violating the spec
requirement.

Restore shuttingDown as a closure-local variable since it no longer
needs to be shared with PipeToReadableStreamReadRequest.

Add a test case verifying that an already-read chunk is written when
an AbortSignal fires after enqueue while the destination is still
writable.

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

Author: colinaaa

lib/internal/webstreams/readablestream.js

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

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

@@ -33,3 +33,34 @@ const { ReadableStream, WritableStream } = require('stream/web');
     sourceController.enqueue('chunk');
   }));
 }
+
+{
+  const ac = new AbortController();
+  let sourceController;
+  const chunks = [];
+
+  const source = new ReadableStream({
+    start(controller) {
+      sourceController = controller;
+    },
+  }, { highWaterMark: 0 });
+
+  const dest = new WritableStream({
+    write: common.mustCall((chunk) => {
+      chunks.push(chunk);
+    }),
+  }, { highWaterMark: 1 });
+
+  source.pipeTo(dest, { signal: ac.signal }).then(
+    common.mustNotCall('pipeTo should not resolve'),
+    common.mustCall((err) => {
+      assert.strictEqual(err.name, 'AbortError');
+      assert.deepStrictEqual(chunks, ['chunk']);
+    })
+  );
+
+  setImmediate(common.mustCall(() => {
+    sourceController.enqueue('chunk');
+    ac.abort();
+  }));
+}