stream: drop noop .then in pipe orchestration

lib/internal/webstreams/readablestream.js

@@ -111,6 +111,7 @@
   kState,
   kType,
   lazyTransfer,
+  markPromiseAsHandled,
   nonOpCancel,
   nonOpPull,
   nonOpStart,
@@ -387,7 +388,11 @@
       !!preventAbort,
       !!preventCancel,
       signal);
-    setPromiseHandled(promise);
+    // The pipeTo promise's rejection is also propagated through the
+    // destination stream's state, which the returned readable side
+    // observes via its own error path. markAsHandled silences the
+    // unhandled-rejection event without allocating a no-op .then chain.
+    markPromiseAsHandled(promise);
 
     return readable;
   }
@@ -1617,7 +1622,14 @@
     disposable = addAbortListener(signal, abortAlgorithm);
   }
 
-  setPromiseHandled(run());
+  // run() rejects when step() throws — i.e. when a write or a
+  // writer.ready/closed await fails. Those same errors reach pipeTo via
+  // watchErrored on writer.closed / reader.closed (registered just below),
+  // which calls shutdownWithAnAction and ultimately resolves pipeTo's outer
+  // promise. The run() promise itself is otherwise unobserved, so a
+  // markAsHandled is sufficient to silence the unhandled-rejection event
+  // without allocating a no-op .then chain.
+  markPromiseAsHandled(run());
 
   watchErrored(source, reader[kState].close.promise, (error) => {
     if (!preventAbort) {

lib/internal/webstreams/transfer.js

@@ -9,7 +9,7 @@ const {
 
 const {
   kState,
-  setPromiseHandled,
+  markPromiseAsHandled,
 } = require('internal/webstreams/util');
 
 const {
@@ -279,7 +279,10 @@ function newCrossRealmReadableStream(writable, port) {
 
   const promise = readableStreamPipeTo(readable, writable, false, false, false);
 
-  setPromiseHandled(promise);
+  // pipeTo's rejection here flows back via the MessagePort and the wrapped
+  // source/sink; nothing else observes this internal promise. markAsHandled
+  // silences the unhandled-rejection event without allocating .then chain.
+  markPromiseAsHandled(promise);
 
   return {
     readable,
@@ -295,7 +298,10 @@ function newCrossRealmWritableSink(readable, port) {
 
   const promise = readableStreamPipeTo(readable, writable, false, false, false);
 
-  setPromiseHandled(promise);
+  // pipeTo's rejection here flows back via the MessagePort and the wrapped
+  // source/sink; nothing else observes this internal promise. markAsHandled
+  // silences the unhandled-rejection event without allocating .then chain.
+  markPromiseAsHandled(promise);
 
   return {
     writable,

lib/internal/webstreams/util.js

@@ -35,6 +35,7 @@ const {
     kPending,
   },
   getPromiseDetails,
+  markPromiseAsHandled,
 } = internalBinding('util');
 
 const assert = require('internal/assert');
@@ -225,6 +226,7 @@ module.exports = {
   kState,
   kType,
   lazyTransfer,
+  markPromiseAsHandled,
   nonOpCancel,
   nonOpFlush,
   nonOpPull,