Release reader immediately when shutting down a pipe

Author: MattiasBuelens

index.bs

@@ -2198,6 +2198,9 @@ 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. 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 +2213,9 @@ 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. 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|.
@@ -2218,10 +2224,8 @@ The following abstract operations operate on {{ReadableStream}} instances at a h
    1. [=Finalize=], passing along |error| if it was given.
   * <dfn id="rs-pipeTo-finalize"><i>Finalize</i></dfn>: both forms of shutdown will eventually ask
     to finalize, optionally with an error |error|, which means to perform the following steps:
+   1. Assert: |reader|.[=ReadableStreamGenericReader/[[stream]]=] is undefined.
    1. Perform ! [$WritableStreamDefaultWriterRelease$](|writer|).
-   1. If |reader| [=implements=] {{ReadableStreamBYOBReader}}, perform
-      ! [$ReadableStreamBYOBReaderRelease$](|reader|).
-   1. Otherwise, perform ! [$ReadableStreamDefaultReaderRelease$](|reader|).
    1. If |signal| is not undefined, [=AbortSignal/remove=] |abortAlgorithm| from |signal|.
    1. If |error| was given, [=reject=] |promise| with |error|.
    1. Otherwise, [=resolve=] |promise| with undefined.

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

@@ -200,6 +200,9 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
       }
 
       return transformPromiseWith(writer._readyPromise, () => {
+        if (shuttingDown === true) {
+          return promiseResolvedWith(true);
+        }
         return new Promise((resolveRead, rejectRead) => {
           ReadableStreamDefaultReaderRead(
             reader,
@@ -289,6 +292,7 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
         return;
       }
       shuttingDown = true;
+      ReadableStreamDefaultReaderRelease(reader);
 
       if (dest._state === 'writable' && WritableStreamCloseQueuedOrInFlight(dest) === false) {
         uponFulfillment(waitForWritesToFinish(), doTheRest);
@@ -310,6 +314,7 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
         return;
       }
       shuttingDown = true;
+      ReadableStreamDefaultReaderRelease(reader);
 
       if (dest._state === 'writable' && WritableStreamCloseQueuedOrInFlight(dest) === false) {
         uponFulfillment(waitForWritesToFinish(), () => finalize(isError, error));
@@ -319,8 +324,8 @@ function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventC
     }
 
     function finalize(isError, error) {
+      assert(reader._stream === undefined);
       WritableStreamDefaultWriterRelease(writer);
-      ReadableStreamDefaultReaderRelease(reader);
 
       if (signal !== undefined) {
         signal.removeEventListener('abort', abortAlgorithm);