stream: implement TransformStream cleanup using "transformer.cancel"

Fixes: #49971 PR-URL: #50126 Reviewed-By: Matteo Collina <[email protected]>

Author: debadree25

lib/internal/webstreams/readablestream.js

@@ -14,6 +14,7 @@ const{
  ObjectCreate,
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
+ Promise,
  PromisePrototypeThen,
  PromiseResolve,
  PromiseReject,
@@ -2444,7 +2445,7 @@ function setupReadableStreamDefaultController(
 conststartResult=startAlgorithm();
 
 PromisePrototypeThen(
-PromiseResolve(startResult),
+newPromise((r)=>r(startResult)),
 ()=>{
 controller[kState].started=true;
 assert(!controller[kState].pulling);
@@ -3243,7 +3244,7 @@ function setupReadableByteStreamController(
 conststartResult=startAlgorithm();
 
 PromisePrototypeThen(
-PromiseResolve(startResult),
+newPromise((r)=>r(startResult)),
 ()=>{
 controller[kState].started=true;
 assert(!controller[kState].pulling);

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(
 returntransformStreamDefaultSourcePullAlgorithm(stream);
 },
 cancel(reason){
-transformStreamErrorWritableAndUnblockWrite(stream,reason);
-returnPromiseResolve();
+returntransformStreamDefaultSourceCancelAlgorithm(stream,reason);
 },
 },{
 highWaterMark: readableHighWaterMark,
@@ -427,6 +426,10 @@ function transformStreamErrorWritableAndUnblockWrite(stream, error){
 writableStreamDefaultControllerErrorIfNeeded(
 writable[kState].controller,
 error);
+transformStreamUnblockWrite(stream);
+}
+
+functiontransformStreamUnblockWrite(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(
 constcontroller=newTransformStreamDefaultController(kSkipThrow);
 consttransform=transformer?.transform||defaultTransformAlgorithm;
 constflush=transformer?.flush||nonOpFlush;
+constcancel=transformer?.cancel||nonOpCancel;
 consttransformAlgorithm=
 FunctionPrototypeBind(transform,transformer);
 constflushAlgorithm=
 FunctionPrototypeBind(flush,transformer);
+constcancelAlgorithm=
+FunctionPrototypeBind(cancel,transformer);
 
 setupTransformStreamDefaultController(
 stream,
 controller,
 transformAlgorithm,
-flushAlgorithm);
+flushAlgorithm,
+cancelAlgorithm);
 }
 
 functiontransformStreamDefaultControllerClearAlgorithms(controller){
 controller[kState].transformAlgorithm=undefined;
 controller[kState].flushAlgorithm=undefined;
+controller[kState].cancelAlgorithm=undefined;
 }
 
 functiontransformStreamDefaultControllerEnqueue(controller,chunk){
@@ -563,7 +573,40 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk){
 }
 
 asyncfunctiontransformStreamDefaultSinkAbortAlgorithm(stream,reason){
-transformStreamError(stream,reason);
+const{
+ controller,
+ readable,
+}=stream[kState];
+
+if(controller[kState].finishPromise!==undefined){
+returncontroller[kState].finishPromise;
+}
+
+const{ promise, resolve, reject }=createDeferredPromise();
+controller[kState].finishPromise=promise;
+constcancelPromise=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);
+},
+);
+
+returncontroller[kState].finishPromise;
 }
 
 functiontransformStreamDefaultSinkCloseAlgorithm(stream){
@@ -572,23 +615,32 @@ function transformStreamDefaultSinkCloseAlgorithm(stream){
  controller,
 }=stream[kState];
 
+if(controller[kState].finishPromise!==undefined){
+returncontroller[kState].finishPromise;
+}
+const{ promise, resolve, reject }=createDeferredPromise();
+controller[kState].finishPromise=promise;
 constflushPromise=
 ensureIsPromise(
 controller[kState].flushAlgorithm,
 controller,
 controller);
 transformStreamDefaultControllerClearAlgorithms(controller);
-returnPromisePrototypeThen(
+PromisePrototypeThen(
 flushPromise,
 ()=>{
 if(readable[kState].state==='errored')
-throwreadable[kState].storedError;
-readableStreamDefaultControllerClose(readable[kState].controller);
+reject(readable[kState].storedError);
+else{
+readableStreamDefaultControllerClose(readable[kState].controller);
+resolve();
+}
 },
 (error)=>{
-transformStreamError(stream,error);
-throwreadable[kState].storedError;
+readableStreamDefaultControllerError(readable[kState].controller,error);
+reject(error);
 });
+returncontroller[kState].finishPromise;
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
@@ -598,6 +650,48 @@ function transformStreamDefaultSourcePullAlgorithm(stream){
 returnstream[kState].backpressureChange.promise;
 }
 
+functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){
+const{
+ controller,
+ writable,
+}=stream[kState];
+
+if(controller[kState].finishPromise!==undefined){
+returncontroller[kState].finishPromise;
+}
+
+const{ promise, resolve, reject }=createDeferredPromise();
+controller[kState].finishPromise=promise;
+constcancelPromise=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);
+},
+);
+
+returncontroller[kState].finishPromise;
+}
+
 module.exports={
  TransformStream,
  TransformStreamDefaultController,

lib/internal/webstreams/writablestream.js

@@ -7,6 +7,7 @@ const{
  FunctionPrototypeCall,
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
+ Promise,
  PromisePrototypeThen,
  PromiseResolve,
  PromiseReject,
@@ -1295,7 +1296,7 @@ function setupWritableStreamDefaultController(
 conststartResult=startAlgorithm();
 
 PromisePrototypeThen(
-PromiseResolve(startResult),
+newPromise((r)=>r(startResult)),
 ()=>{
 assert(stream[kState].state==='writable'||
 stream[kState].state==='erroring');

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

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';

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';
 

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';

test/fixtures/wpt/streams/piping/crashtests/cross-piping.html

@@ -0,0 +1,12 @@
+<!DOCTYPE html>
+<scripttype="module">
+leta=newReadableStream();
+letb=self.open()
+letf=newb.WritableStream();
+a.pipeThrough(
+{"readable": a,"writable": f},
+{"signal": AbortSignal.abort()}
+)
+awaitnewPromise(setTimeout);
+structuredClone(undefined,{"transfer": [f]})
+</script>

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';

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';