stream: fix unhandled rejection for errored pipeTo writes

lib/internal/webstreams/readablestream.js

@@ -1495,6 +1495,19 @@ function readableStreamFromIterable(iterable) {
   return stream;
 }
 
+// markPromiseAsHandled only silences unhandled-rejection if the promise is
+// still pending. writableStreamDefaultWriterWrite can return an already-
+// rejected promise when the destination has errored or closed; in that case
+// attach a real handler via setPromiseHandled.
+function markPipeToWritePromiseHandled(promise, dest) {
+  if (dest[kState].state === 'writable' &&
+      !writableStreamCloseQueuedOrInFlight(dest)) {
+    markPromiseAsHandled(promise);
+  } else {
+    setPromiseHandled(promise);
+  }
+}
+
 function readableStreamPipeTo(
   source,
   dest,
@@ -1668,7 +1681,7 @@ function readableStreamPipeTo(
         // Write the chunk - we're already in a separate microtask from enqueue
         // because we awaited the writer ready promise above.
         state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
-        markPromiseAsHandled(state.currentWrite);
+        markPipeToWritePromiseHandled(state.currentWrite, dest);
 
         // Check backpressure after each write
         if (dest[kState].backpressure) {
@@ -1772,7 +1785,9 @@ class PipeToReadableStreamReadRequest {
     // "ReadableStreamPipeTo" step 15's "chunk steps".
     queueMicrotask(() => {
       this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
-      markPromiseAsHandled(this.state.currentWrite);
+      markPipeToWritePromiseHandled(
+        this.state.currentWrite,
+        this.writer[kState].stream);
       this.promise.resolve(false);
     });
   }

test/parallel/test-whatwg-pipeto-transform-error.js

@@ -0,0 +1,49 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const {
+  ReadableStream,
+  TransformStream,
+  WritableStream,
+} = require('stream/web');
+
+// A late write after the destination has errored returns an already-rejected
+// promise. That must not surface as an unhandled rejection when the pipeTo()
+// rejection is already handled.
+// Refs: https://github.com/nodejs/node/issues/64561
+
+process.on('unhandledRejection', common.mustNotCall());
+
+(async () => {
+  const source = new ReadableStream({
+    start(controller) {
+      controller.enqueue('a');
+      controller.enqueue('b');
+      controller.close();
+    },
+  });
+
+  const asyncPassThrough = new TransformStream({
+    async transform(chunk, controller) {
+      controller.enqueue(chunk);
+    },
+  });
+
+  const identity = new TransformStream();
+
+  const failingTransform = new TransformStream({
+    transform() {
+      throw new Error('boom');
+    },
+  });
+
+  await assert.rejects(
+    source
+      .pipeThrough(asyncPassThrough)
+      .pipeThrough(identity)
+      .pipeThrough(failingTransform)
+      .pipeTo(new WritableStream()),
+    { message: 'boom' },
+  );
+})().then(common.mustCall());