stream: implement TransformStream cleanup using "transformer.cancel"

Fixes: #49971 PR-URL: #50126 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
nodejs · Dec 15, 2023 · 71acd36 · 71acd36
1 parent 3ccd5fa
commit 71acd36
Show file tree

Hide file tree

Showing 72 changed files with 418 additions and 94 deletions.
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
@@ -14,6 +14,7 @@ const {
   ObjectCreate,
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
+  Promise,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseReject,
@@ -2444,7 +2445,7 @@ function setupReadableStreamDefaultController(
   const startResult = startAlgorithm();
 
   PromisePrototypeThen(
-    PromiseResolve(startResult),
+    new Promise((r) => r(startResult)),
     () => {
       controller[kState].started = true;
       assert(!controller[kState].pulling);
@@ -3243,7 +3244,7 @@ function setupReadableByteStreamController(
   const startResult = startAlgorithm();
 
   PromisePrototypeThen(
-    PromiseResolve(startResult),
+    new Promise((r) => r(startResult)),
     () => {
       controller[kState].started = true;
       assert(!controller[kState].pulling);

diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js
@@ -6,7 +6,6 @@ const {
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
   PromisePrototypeThen,
-  PromiseResolve,
   SymbolToStringTag,
   Symbol,
 } = primordials;
@@ -47,6 +46,7 @@ const {
   nonOpFlush,
   kType,
   kState,
+  nonOpCancel,
 } = require('internal/webstreams/util');
 
 const {
@@ -384,8 +384,7 @@ function initializeTransformStream(
       return transformStreamDefaultSourcePullAlgorithm(stream);
     },
     cancel(reason) {
-      transformStreamErrorWritableAndUnblockWrite(stream, reason);
-      return PromiseResolve();
+      return transformStreamDefaultSourceCancelAlgorithm(stream, reason);
     },
   }, {
     highWaterMark: readableHighWaterMark,
@@ -427,6 +426,10 @@ function transformStreamErrorWritableAndUnblockWrite(stream, error) {
   writableStreamDefaultControllerErrorIfNeeded(
     writable[kState].controller,
     error);
+  transformStreamUnblockWrite(stream);
+}
+
+function transformStreamUnblockWrite(stream) {
   if (stream[kState].backpressure)
     transformStreamSetBackpressure(stream, false);
 }
@@ -443,13 +446,15 @@ function setupTransformStreamDefaultController(
   stream,
   controller,
   transformAlgorithm,
-  flushAlgorithm) {
+  flushAlgorithm,
+  cancelAlgorithm) {
   assert(isTransformStream(stream));
   assert(stream[kState].controller === undefined);
   controller[kState] = {
     stream,
     transformAlgorithm,
     flushAlgorithm,
+    cancelAlgorithm,
   };
   stream[kState].controller = controller;
 }
@@ -460,21 +465,26 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const controller = new TransformStreamDefaultController(kSkipThrow);
   const transform = transformer?.transform || defaultTransformAlgorithm;
   const flush = transformer?.flush || nonOpFlush;
+  const cancel = transformer?.cancel || nonOpCancel;
   const transformAlgorithm =
     FunctionPrototypeBind(transform, transformer);
   const flushAlgorithm =
     FunctionPrototypeBind(flush, transformer);
+  const cancelAlgorithm =
+    FunctionPrototypeBind(cancel, transformer);
 
   setupTransformStreamDefaultController(
     stream,
     controller,
     transformAlgorithm,
-    flushAlgorithm);
+    flushAlgorithm,
+    cancelAlgorithm);
 }
 
 function transformStreamDefaultControllerClearAlgorithms(controller) {
   controller[kState].transformAlgorithm = undefined;
   controller[kState].flushAlgorithm = undefined;
+  controller[kState].cancelAlgorithm = undefined;
 }
 
 function transformStreamDefaultControllerEnqueue(controller, chunk) {
@@ -563,7 +573,40 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
 }
 
 async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
-  transformStreamError(stream, reason);
+  const {
+    controller,
+    readable,
+  } = stream[kState];
+
+  if (controller[kState].finishPromise !== undefined) {
+    return controller[kState].finishPromise;
+  }
+
+  const { promise, resolve, reject } = createDeferredPromise();
+  controller[kState].finishPromise = promise;
+  const cancelPromise = ensureIsPromise(
+    controller[kState].cancelAlgorithm,
+    controller,
+    reason);
+  transformStreamDefaultControllerClearAlgorithms(controller);
+
+  PromisePrototypeThen(
+    cancelPromise,
+    () => {
+      if (readable[kState].state === 'errored')
+        reject(readable[kState].storedError);
+      else {
+        readableStreamDefaultControllerError(readable[kState].controller, reason);
+        resolve();
+      }
+    },
+    (error) => {
+      readableStreamDefaultControllerError(readable[kState].controller, error);
+      reject(error);
+    },
+  );
+
+  return controller[kState].finishPromise;
 }
 
 function transformStreamDefaultSinkCloseAlgorithm(stream) {
@@ -572,23 +615,32 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
     controller,
   } = stream[kState];
 
+  if (controller[kState].finishPromise !== undefined) {
+    return controller[kState].finishPromise;
+  }
+  const { promise, resolve, reject } = createDeferredPromise();
+  controller[kState].finishPromise = promise;
   const flushPromise =
     ensureIsPromise(
       controller[kState].flushAlgorithm,
       controller,
       controller);
   transformStreamDefaultControllerClearAlgorithms(controller);
-  return PromisePrototypeThen(
+  PromisePrototypeThen(
     flushPromise,
     () => {
       if (readable[kState].state === 'errored')
-        throw readable[kState].storedError;
-      readableStreamDefaultControllerClose(readable[kState].controller);
+        reject(readable[kState].storedError);
+      else {
+        readableStreamDefaultControllerClose(readable[kState].controller);
+        resolve();
+      }
     },
     (error) => {
-      transformStreamError(stream, error);
-      throw readable[kState].storedError;
+      readableStreamDefaultControllerError(readable[kState].controller, error);
+      reject(error);
     });
+  return controller[kState].finishPromise;
 }
 
 function transformStreamDefaultSourcePullAlgorithm(stream) {
@@ -598,6 +650,48 @@ function transformStreamDefaultSourcePullAlgorithm(stream) {
   return stream[kState].backpressureChange.promise;
 }
 
+function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
+  const {
+    controller,
+    writable,
+  } = stream[kState];
+
+  if (controller[kState].finishPromise !== undefined) {
+    return controller[kState].finishPromise;
+  }
+
+  const { promise, resolve, reject } = createDeferredPromise();
+  controller[kState].finishPromise = promise;
+  const cancelPromise = ensureIsPromise(
+    controller[kState].cancelAlgorithm,
+    controller,
+    reason);
+  transformStreamDefaultControllerClearAlgorithms(controller);
+
+  PromisePrototypeThen(cancelPromise,
+                       () => {
+                         if (writable[kState].state === 'errored')
+                           reject(writable[kState].storedError);
+                         else {
+                           writableStreamDefaultControllerErrorIfNeeded(
+                             writable[kState].controller,
+                             reason);
+                           transformStreamUnblockWrite(stream);
+                           resolve();
+                         }
+                       },
+                       (error) => {
+                         writableStreamDefaultControllerErrorIfNeeded(
+                           writable[kState].controller,
+                           error);
+                         transformStreamUnblockWrite(stream);
+                         reject(error);
+                       },
+  );
+
+  return controller[kState].finishPromise;
+}
+
 module.exports = {
   TransformStream,
   TransformStreamDefaultController,

diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js
@@ -7,6 +7,7 @@ const {
   FunctionPrototypeCall,
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
+  Promise,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseReject,
@@ -1295,7 +1296,7 @@ function setupWritableStreamDefaultController(
   const startResult = startAlgorithm();
 
   PromisePrototypeThen(
-    PromiseResolve(startResult),
+    new Promise((r) => r(startResult)),
     () => {
       assert(stream[kState].state === 'writable' ||
              stream[kState].state === 'erroring');

diff --git a/test/fixtures/wpt/README.md b/test/fixtures/wpt/README.md
@@ -27,7 +27,7 @@ Last update:
 - performance-timeline: https://github.com/web-platform-tests/wpt/tree/17ebc3aea0/performance-timeline
 - resource-timing: https://github.com/web-platform-tests/wpt/tree/22d38586d0/resource-timing
 - resources: https://github.com/web-platform-tests/wpt/tree/1e140d63ec/resources
-- streams: https://github.com/web-platform-tests/wpt/tree/517e945bbf/streams
+- streams: https://github.com/web-platform-tests/wpt/tree/a8872d92b1/streams
 - url: https://github.com/web-platform-tests/wpt/tree/c2d7e70b52/url
 - user-timing: https://github.com/web-platform-tests/wpt/tree/5ae85bf826/user-timing
 - wasm/jsapi: https://github.com/web-platform-tests/wpt/tree/cde25e7e3c/wasm/jsapi

diff --git a/test/fixtures/wpt/streams/piping/abort.any.js b/test/fixtures/wpt/streams/piping/abort.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/recording-streams.js
 // META: script=../resources/test-utils.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/close-propagation-backward.any.js b/test/fixtures/wpt/streams/piping/close-propagation-backward.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/recording-streams.js
 'use strict';
 

diff --git a/test/fixtures/wpt/streams/piping/close-propagation-forward.any.js b/test/fixtures/wpt/streams/piping/close-propagation-forward.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/crashtests/cross-piping.html b/test/fixtures/wpt/streams/piping/crashtests/cross-piping.html
@@ -0,0 +1,12 @@
+<!DOCTYPE html>
+<script type="module">
+  let a = new ReadableStream();
+  let b = self.open()
+  let f = new b.WritableStream();
+  a.pipeThrough(
+    { "readable": a, "writable": f },
+    { "signal": AbortSignal.abort() }
+  )
+  await new Promise(setTimeout);
+  structuredClone(undefined, { "transfer": [f] })
+</script>
diff --git a/test/fixtures/wpt/streams/piping/error-propagation-backward.any.js b/test/fixtures/wpt/streams/piping/error-propagation-backward.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/error-propagation-forward.any.js b/test/fixtures/wpt/streams/piping/error-propagation-forward.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/flow-control.any.js b/test/fixtures/wpt/streams/piping/flow-control.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/rs-utils.js
 // META: script=../resources/recording-streams.js

diff --git a/test/fixtures/wpt/streams/piping/general-addition.any.js b/test/fixtures/wpt/streams/piping/general-addition.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 promise_test(async t => {

diff --git a/test/fixtures/wpt/streams/piping/general.any.js b/test/fixtures/wpt/streams/piping/general.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/multiple-propagation.any.js b/test/fixtures/wpt/streams/piping/multiple-propagation.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/pipe-through.any.js b/test/fixtures/wpt/streams/piping/pipe-through.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/rs-utils.js
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js

diff --git a/test/fixtures/wpt/streams/piping/then-interception.any.js b/test/fixtures/wpt/streams/piping/then-interception.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/test-utils.js
 // META: script=../resources/recording-streams.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/piping/throwing-options.any.js b/test/fixtures/wpt/streams/piping/throwing-options.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 class ThrowingOptions {

diff --git a/test/fixtures/wpt/streams/piping/transform-streams.any.js b/test/fixtures/wpt/streams/piping/transform-streams.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 promise_test(() => {

diff --git a/test/fixtures/wpt/streams/queuing-strategies.any.js b/test/fixtures/wpt/streams/queuing-strategies.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 const highWaterMarkConversions = new Map([

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/bad-buffers-and-views.any.js b/test/fixtures/wpt/streams/readable-byte-streams/bad-buffers-and-views.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 promise_test(() => {

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/construct-byob-request.any.js b/test/fixtures/wpt/streams/readable-byte-streams/construct-byob-request.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/rs-utils.js
 'use strict';
 

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/enqueue-with-detached-buffer.any.js b/test/fixtures/wpt/streams/readable-byte-streams/enqueue-with-detached-buffer.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 
 promise_test(async t => {
   const error = new Error('cannot proceed');

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/general.any.js b/test/fixtures/wpt/streams/readable-byte-streams/general.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 // META: script=../resources/rs-utils.js
 // META: script=../resources/test-utils.js
 'use strict';

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/non-transferable-buffers.any.js b/test/fixtures/wpt/streams/readable-byte-streams/non-transferable-buffers.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 'use strict';
 
 promise_test(async t => {

diff --git a/test/fixtures/wpt/streams/readable-byte-streams/respond-after-enqueue.any.js b/test/fixtures/wpt/streams/readable-byte-streams/respond-after-enqueue.any.js
@@ -1,4 +1,4 @@
-// META: global=window,worker
+// META: global=window,worker,shadowrealm
 
 'use strict';