Release reader immediately when shutting down a pipe

index.bs

@@ -2198,6 +2198,17 @@ The following abstract operations operate on {{ReadableStream}} instances at a h
     |originalError|, then:
    1. If |shuttingDown| is true, abort these substeps.
    1. Set |shuttingDown| to true.
+   1. If |reader| [=implements=] {{ReadableStreamBYOBReader}}, perform
+      ! [$ReadableStreamBYOBReaderRelease$](|reader|).
+   1. Otherwise, perform ! [$ReadableStreamDefaultReaderRelease$](|reader|).
+   1. Set |reader| to ! [$AcquireReadableStreamDefaultReader$](|source|).
+      <p class="note">The initial reader is released to ensure that any pending read requests
+      are immediately aborted, and no more chunks are pulled from |source|. A new reader is
+      acquired in order to keep |source| locked until the shutdown is [=finalized=], for example
+      to [=cancel a readable stream|cancel=] |source| if necessary.
+      This exchange of readers is not observable to author code and the user agent is free to
+      implement this differently, for example by keeping the same reader and internally aborting
+      its pending read requests.
    1. If |dest|.[=WritableStream/[[state]]=] is "`writable`" and !
       [$WritableStreamCloseQueuedOrInFlight$](|dest|) is false,
      1. If any [=chunks=] have been read but not yet written, write them to |dest|.
@@ -2210,6 +2221,10 @@ The following abstract operations operate on {{ReadableStream}} instances at a h
     ask to shutdown, optionally with an error |error|, then:
    1. If |shuttingDown| is true, abort these substeps.
    1. Set |shuttingDown| to true.
+   1. If |reader| [=implements=] {{ReadableStreamBYOBReader}}, perform
+      ! [$ReadableStreamBYOBReaderRelease$](|reader|).
+   1. Otherwise, perform ! [$ReadableStreamDefaultReaderRelease$](|reader|).
+   1. Set |reader| to ! [$AcquireReadableStreamDefaultReader$](|source|).
    1. If |dest|.[=WritableStream/[[state]]=] is "`writable`" and !
       [$WritableStreamCloseQueuedOrInFlight$](|dest|) is false,
     1. If any [=chunks=] have been read but not yet written, write them to |dest|.

reference-implementation/lib/abstract-ops/readable-streams.js

@@ -134,7 +134,7 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
   assert(IsReadableStreamLocked(source) === false);
   assert(IsWritableStreamLocked(dest) === false);
 
-  const reader = AcquireReadableStreamDefaultReader(source);
+  let reader = AcquireReadableStreamDefaultReader(source);
   const writer = AcquireWritableStreamDefaultWriter(dest);
 
   source._disturbed = true;
@@ -200,6 +200,12 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
       }
 
       return transformPromiseWith(writer._readyPromise, () => {
+        if (shuttingDown === true) {
+          return promiseResolvedWith(true);
+        }
+        if (dest._state !== 'writable' || WritableStreamCloseQueuedOrInFlight(dest) === true) {
+          return promiseResolvedWith(true);
+        }
         return new Promise((resolveRead, rejectRead) => {
           ReadableStreamDefaultReaderRead(
             reader,
@@ -289,6 +295,8 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
         return;
       }
       shuttingDown = true;
+      ReadableStreamDefaultReaderRelease(reader);
+      reader = AcquireReadableStreamDefaultReader(source);
 
       if (dest._state === 'writable' && WritableStreamCloseQueuedOrInFlight(dest) === false) {
         uponFulfillment(waitForWritesToFinish(), doTheRest);
@@ -310,6 +318,8 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
         return;
       }
       shuttingDown = true;
+      ReadableStreamDefaultReaderRelease(reader);
+      reader = AcquireReadableStreamDefaultReader(source);
 
       if (dest._state === 'writable' && WritableStreamCloseQueuedOrInFlight(dest) === false) {
         uponFulfillment(waitForWritesToFinish(), () => finalize(isError, error));