Repository: nodejs/node
Pull request: stream: optimize webstreams `pipeTo` further

I've started looking more closely at the `pipeTo()` implementation, and found some low-hanging fruit to further optimize it:
* We don't have to `await writer.ready` every time, that's only needed if there is backpressure. So we check that first.
* We don't have to go through `writer.desiredSize` to check if there is backpressure, we can read that directly from the internal state of the `WritableStreamDefaultController`.
* `readableStreamDefaultControllerCallPullIfNeeded` already checks `state === 'readable' && !closeRequested` through [`readableStreamDefaultControllerCanCloseOrEnqueue`](https://github.com/nodejs/node/blob/199daab0b0822d6063a73b9362bfce8667d2a112/lib/internal/webstreams/readablestream.js#L2415-L2416), so we don't have to repeat that check.

This also fixes a few more edge cases:
* Previously, we were using `writer.desiredSize` inside `pipeTo()`, which is part of the public `WritableStreamDefaultWriter` API, [and step 15 of the spec explicitly forbids using the public API since it can be modified by user-land code](https://streams.spec.whatwg.org/#readable-stream-pipe-to). We now bypass this entirely using `dest[kState].backpressure`.
  * There doesn't seem to be much WPT test coverage for this... I might add more tests to the upstream [`patched-global.any.js`](https://github.com/web-platform-tests/wpt/blob/3a17a2e2aa7950fbc07aa08848b7b1f698c14481/streams/readable-streams/patched-global.any.js) suite.
* If the writable end becomes closed or errored in the middle of a batch of synchronous reads, we should stop immediately.

On my machine, this improves the `webstreams/pipe-to.js` benchmark by 3% to 7%:
<details>
<summary>Baseline @ 199daab0b0822d6063a73b9362bfce8667d2a112</summary>

```
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=512 n=500000: 1,639,426.502378644
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=512 n=500000: 1,672,850.078172284
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=512 n=500000: 1,684,703.098034221
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=512 n=500000: 1,658,174.6018059512
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=1024 n=500000: 1,684,320.5922340695
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=1024 n=500000: 1,663,948.3297453062
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=1024 n=500000: 1,651,671.1443471392
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=1024 n=500000: 1,630,100.2576862487
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=2048 n=500000: 1,662,200.8869503932
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=2048 n=500000: 1,638,533.5910855907
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=2048 n=500000: 1,671,184.435256644
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=2048 n=500000: 1,639,392.637815542
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=4096 n=500000: 1,644,874.8168020674
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=4096 n=500000: 1,702,132.8746198711
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=4096 n=500000: 1,608,151.915035545
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=4096 n=500000: 1,688,220.1736233155
```
</details>
<details>
<summary>This PR @ 7b31401e6d9ad9f48090334272943f3b21d29508</summary>

```
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=512 n=500000: 1,768,254.040106831
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=512 n=500000: 1,717,563.2174905646
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=512 n=500000: 1,746,963.515365767
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=512 n=500000: 1,787,073.738236587
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=1024 n=500000: 1,756,757.2790359478
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=1024 n=500000: 1,704,454.421184323
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=1024 n=500000: 1,761,150.6372019118
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=1024 n=500000: 1,766,948.0354718352
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=2048 n=500000: 1,727,179.5537866165
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=2048 n=500000: 1,768,980.3632565797
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=2048 n=500000: 1,752,746.9927243474
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=2048 n=500000: 1,793,850.0362895865
webstreams\pipe-to.js highWaterMarkW=512 highWaterMarkR=4096 n=500000: 1,769,294.5999359516
webstreams\pipe-to.js highWaterMarkW=1024 highWaterMarkR=4096 n=500000: 1,768,876.4768791925
webstreams\pipe-to.js highWaterMarkW=2048 highWaterMarkR=4096 n=500000: 1,748,478.5613798152
webstreams\pipe-to.js highWaterMarkW=4096 highWaterMarkR=4096 n=500000: 1,783,529.4621231847
```
</details>

Changed files:
filename: lib/internal/webstreams/readablestream.js
patch: @@ -1532,10 +1532,12 @@ function readableStreamPipeTo(
   }
 
   async function step() {
-    if (shuttingDown)
-      return true;
+    if (shuttingDown) return true;
 
-    await writer[kState].ready.promise;
+    if (dest[kState].backpressure) {
+      await writer[kState].ready.promise;
+      if (shuttingDown) return true;
+    }
 
     const controller = source[kState].controller;
 
@@ -1549,7 +1551,6 @@ function readableStreamPipeTo(
       while (controller[kState].queue.length > 0) {
         if (shuttingDown) return true;
 
-        source[kState].disturbed = true;
         const chunk = dequeueValue(controller);
 
         if (controller[kState].closeRequested && !controller[kState].queue.length) {
@@ -1563,20 +1564,17 @@ function readableStreamPipeTo(
         setPromiseHandled(state.currentWrite);
 
         // Check backpressure after each write
-        if (dest[kState].state === 'writable') {
-          const desiredSize = writer.desiredSize;
-          if (desiredSize !== null && desiredSize <= 0) {
-            // Backpressure - stop batch and wait for ready
-            break;
-          }
+        if (dest[kState].backpressure) {
+          // Backpressure - stop batch and wait for ready
+          break;
+        } else if (dest[kState].state !== 'writable' || writableStreamCloseQueuedOrInFlight(dest)) {
+          // Closing or erroring - stop batch and wait for shutdown
+          break;
         }
       }
 
       // Trigger pull if needed after batch
-      if (source[kState].state === 'readable' &&
-          !controller[kState].closeRequested) {
-        readableStreamDefaultControllerCallPullIfNeeded(controller);
-      }
+      readableStreamDefaultControllerCallPullIfNeeded(controller);
 
       // Check if stream closed during batch
       if (source[kState].state === 'closed') {