LibWeb: Update spec steps for closing TransformStreams

This was actually an older change to the Streams spec that we missed
when we implemented TransformStreams. This fixes a crash in the imported
WPT tests.

See: https://github.com/whatwg/streams/commit/007d729

Libraries/LibWeb/Streams/AbstractOperations.cpp

@@ -5421,7 +5421,7 @@ GC::Ref<WebIDL::Promise> transform_stream_default_sink_abort_algorithm(Transform
 
     // 2. If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
     if (controller->finish_promise())
-        return GC::Ref { *controller->finish_promise() };
+        return *controller->finish_promise();
 
     // 3. Let readable be stream.[[readable]].
     auto readable = stream.readable();
@@ -5436,8 +5436,7 @@ GC::Ref<WebIDL::Promise> transform_stream_default_sink_abort_algorithm(Transform
     transform_stream_default_controller_clear_algorithms(*controller);
 
     // 7. React to cancelPromise:
-    WebIDL::react_to_promise(
-        *cancel_promise,
+    WebIDL::react_to_promise(cancel_promise,
         // 1. If cancelPromise was fulfilled, then:
         GC::create_function(realm.heap(), [&realm, readable, controller](JS::Value reason) -> WebIDL::ExceptionOr<JS::Value> {
             // 1. If readable.[[state]] is "errored", reject controller.[[finishPromise]] with readable.[[storedError]].
@@ -5453,8 +5452,10 @@ GC::Ref<WebIDL::Promise> transform_stream_default_sink_abort_algorithm(Transform
                 // 2. Resolve controller.[[finishPromise]] with undefined.
                 WebIDL::resolve_promise(realm, *controller->finish_promise(), JS::js_undefined());
             }
+
             return JS::js_undefined();
         }),
+
         // 2. If cancelPromise was rejected with reason r, then:
         GC::create_function(realm.heap(), [&realm, readable, controller](JS::Value reason) -> WebIDL::ExceptionOr<JS::Value> {
             VERIFY(readable->controller().has_value() && readable->controller()->has<GC::Ref<ReadableStreamDefaultController>>());
@@ -5476,43 +5477,58 @@ GC::Ref<WebIDL::Promise> transform_stream_default_sink_close_algorithm(Transform
 {
     auto& realm = stream.realm();
 
-    // 1. Let readable be stream.[[readable]].
-    auto readable = stream.readable();
-
-    // 2. Let controller be stream.[[controller]].
+    // 1. Let controller be stream.[[controller]].
     auto controller = stream.controller();
 
-    // 3. Let flushPromise be the result of performing controller.[[flushAlgorithm]].
+    // 2. If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
+    if (controller->finish_promise())
+        return *controller->finish_promise();
+
+    // 3. Let readable be stream.[[readable]].
+    auto readable = stream.readable();
+
+    // 4. Let controller.[[finishPromise]] be a new promise.
+    controller->set_finish_promise(WebIDL::create_promise(realm));
+
+    // 5. Let flushPromise be the result of performing controller.[[flushAlgorithm]].
     auto flush_promise = controller->flush_algorithm()->function()();
 
-    // 4. Perform ! TransformStreamDefaultControllerClearAlgorithms(controller).
+    // 6. Perform ! TransformStreamDefaultControllerClearAlgorithms(controller).
     transform_stream_default_controller_clear_algorithms(*controller);
 
-    // 5. Return the result of reacting to flushPromise:
-    auto react_result = WebIDL::react_to_promise(
-        *flush_promise,
+    // 7. React to flushPromise:
+    WebIDL::react_to_promise(flush_promise,
         // 1. If flushPromise was fulfilled, then:
-        GC::create_function(realm.heap(), [readable](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
-            // 1. If readable.[[state]] is "errored", throw readable.[[storedError]].
-            if (readable->state() == ReadableStream::State::Errored)
-                return JS::throw_completion(readable->stored_error());
+        GC::create_function(realm.heap(), [&realm, controller, readable](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
+            // 1. If readable.[[state]] is "errored", reject controller.[[finishPromise]] with readable.[[storedError]].
+            if (readable->state() == ReadableStream::State::Errored) {
+                WebIDL::reject_promise(realm, *controller->finish_promise(), readable->stored_error());
+            }
+            // 2. Otherwise:
+            else {
+                // 1. Perform ! ReadableStreamDefaultControllerClose(readable.[[controller]]).
+                readable_stream_default_controller_close(readable->controller().value().get<GC::Ref<ReadableStreamDefaultController>>());
 
-            VERIFY(readable->controller().has_value() && readable->controller()->has<GC::Ref<ReadableStreamDefaultController>>());
-            // 2. Perform ! ReadableStreamDefaultControllerClose(readable.[[controller]]).
-            readable_stream_default_controller_close(readable->controller().value().get<GC::Ref<ReadableStreamDefaultController>>());
+                // 2. Resolve controller.[[finishPromise]] with undefined.
+                WebIDL::resolve_promise(realm, *controller->finish_promise(), JS::js_undefined());
+            }
 
             return JS::js_undefined();
         }),
-        // 2. If flushPromise was rejected with reason r, then:
-        GC::create_function(realm.heap(), [&stream, readable](JS::Value reason) -> WebIDL::ExceptionOr<JS::Value> {
-            // 1. Perform ! TransformStreamError(stream, r).
-            transform_stream_error(stream, reason);
 
-            // 2. Throw readable.[[storedError]].
-            return JS::throw_completion(readable->stored_error());
+        // 2. If flushPromise was rejected with reason r, then:
+        GC::create_function(realm.heap(), [&realm, controller, readable](JS::Value reason) -> WebIDL::ExceptionOr<JS::Value> {
+            // 1. Perform ! ReadableStreamDefaultControllerError(readable.[[controller]], r).
+            readable_stream_default_controller_error(readable->controller().value().get<GC::Ref<ReadableStreamDefaultController>>(), reason);
+
+            // 2. Reject controller.[[finishPromise]] with r.
+            WebIDL::reject_promise(realm, *controller->finish_promise(), reason);
+
+            return JS::js_undefined();
         }));
 
-    return react_result;
+    // 8. Return controller.[[finishPromise]].
+    return *controller->finish_promise();
 }
 
 // https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm
@@ -5587,7 +5603,7 @@ GC::Ref<WebIDL::Promise> transform_stream_default_source_cancel_algorithm(Transf
 
     // 2. If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
     if (controller->finish_promise())
-        return GC::Ref { *controller->finish_promise() };
+        return *controller->finish_promise();
 
     // 3. Let writable be stream.[[writable]].
     auto writable = stream.writable();
@@ -5602,8 +5618,7 @@ GC::Ref<WebIDL::Promise> transform_stream_default_source_cancel_algorithm(Transf
     transform_stream_default_controller_clear_algorithms(*controller);
 
     // 7. React to cancelPromise:
-    WebIDL::react_to_promise(
-        *cancel_promise,
+    WebIDL::react_to_promise(cancel_promise,
         // 1. If cancelPromise was fulfilled, then:
         GC::create_function(realm.heap(), [&realm, writable, controller, &stream, reason](JS::Value) -> WebIDL::ExceptionOr<JS::Value> {
             // 1. If writable.[[state]] is "errored", reject controller.[[finishPromise]] with writable.[[storedError]].
@@ -5614,21 +5629,28 @@ GC::Ref<WebIDL::Promise> transform_stream_default_source_cancel_algorithm(Transf
             else {
                 // 1. Perform ! WritableStreamDefaultControllerErrorIfNeeded(writable.[[controller]], reason).
                 writable_stream_default_controller_error_if_needed(*writable->controller(), reason);
+
                 // 2. Perform ! TransformStreamUnblockWrite(stream).
                 transform_stream_unblock_write(stream);
+
                 // 3. Resolve controller.[[finishPromise]] with undefined.
                 WebIDL::resolve_promise(realm, *controller->finish_promise(), JS::js_undefined());
             }
+
             return JS::js_undefined();
         }),
+
         // 2. If cancelPromise was rejected with reason r, then:
         GC::create_function(realm.heap(), [&realm, writable, &stream, controller](JS::Value reason) -> WebIDL::ExceptionOr<JS::Value> {
             // 1. Perform ! WritableStreamDefaultControllerErrorIfNeeded(writable.[[controller]], r).
             writable_stream_default_controller_error_if_needed(*writable->controller(), reason);
+
             // 2. Perform ! TransformStreamUnblockWrite(stream).
             transform_stream_unblock_write(stream);
+
             // 3. Reject controller.[[finishPromise]] with r.
             WebIDL::reject_promise(realm, *controller->finish_promise(), reason);
+
             return JS::js_undefined();
         }));
 

Tests/LibWeb/Text/expected/wpt-import/streams/transform-streams/cancel.any.txt

@@ -0,0 +1,16 @@
+Harness status: OK
+
+Found 11 tests
+
+11 Pass
+Pass	cancelling the readable side should call transformer.cancel()
+Pass	cancelling the readable side should reject if transformer.cancel() throws
+Pass	aborting the writable side should call transformer.abort()
+Pass	aborting the writable side should reject if transformer.cancel() throws
+Pass	closing the writable side should reject if a parallel transformer.cancel() throws
+Pass	readable.cancel() and a parallel writable.close() should reject if a transformer.cancel() calls controller.error()
+Pass	writable.abort() and readable.cancel() should reject if a transformer.cancel() calls controller.error()
+Pass	readable.cancel() should not call cancel() when flush() is already called from writable.close()
+Pass	readable.cancel() should not call cancel() again when already called from writable.abort()
+Pass	writable.close() should not call flush() when cancel() is already called from readable.cancel()
+Pass	writable.abort() should not call cancel() again when already called from readable.cancel()

Tests/LibWeb/Text/input/wpt-import/streams/transform-streams/cancel.any.html

@@ -0,0 +1,15 @@
+<!doctype html>
+<meta charset=utf-8>
+
+<script>
+self.GLOBAL = {
+  isWindow: function() { return true; },
+  isWorker: function() { return false; },
+  isShadowRealm: function() { return false; },
+};
+</script>
+<script src="../../resources/testharness.js"></script>
+<script src="../../resources/testharnessreport.js"></script>
+<script src="../resources/test-utils.js"></script>
+<div id=log></div>
+<script src="../../streams/transform-streams/cancel.any.js"></script>

Tests/LibWeb/Text/input/wpt-import/streams/transform-streams/cancel.any.js

@@ -0,0 +1,205 @@
+// META: global=window,worker,shadowrealm
+// META: script=../resources/test-utils.js
+'use strict';
+
+const thrownError = new Error('bad things are happening!');
+thrownError.name = 'error1';
+
+const originalReason = new Error('original reason');
+originalReason.name = 'error2';
+
+promise_test(async t => {
+  let cancelled = undefined;
+  const ts = new TransformStream({
+    cancel(reason) {
+      cancelled = reason;
+    }
+  });
+  const res = await ts.readable.cancel(thrownError);
+  assert_equals(res, undefined, 'readable.cancel() should return undefined');
+  assert_equals(cancelled, thrownError, 'transformer.cancel() should be called with the passed reason');
+}, 'cancelling the readable side should call transformer.cancel()');
+
+promise_test(async t => {
+  const ts = new TransformStream({
+    cancel(reason) {
+      assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason');
+      throw thrownError;
+    }
+  });
+  const writer = ts.writable.getWriter();
+  const cancelPromise = ts.readable.cancel(originalReason);
+  await promise_rejects_exactly(t, thrownError, cancelPromise, 'readable.cancel() should reject with thrownError');
+  await promise_rejects_exactly(t, thrownError, writer.closed, 'writer.closed should reject with thrownError');
+}, 'cancelling the readable side should reject if transformer.cancel() throws');
+
+promise_test(async t => {
+  let aborted = undefined;
+  const ts = new TransformStream({
+    cancel(reason) {
+      aborted = reason;
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  const res = await ts.writable.abort(thrownError);
+  assert_equals(res, undefined, 'writable.abort() should return undefined');
+  assert_equals(aborted, thrownError, 'transformer.abort() should be called with the passed reason');
+}, 'aborting the writable side should call transformer.abort()');
+
+promise_test(async t => {
+  const ts = new TransformStream({
+    cancel(reason) {
+      assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason');
+      throw thrownError;
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  const reader = ts.readable.getReader();
+  const abortPromise = ts.writable.abort(originalReason);
+  await promise_rejects_exactly(t, thrownError, abortPromise, 'writable.abort() should reject with thrownError');
+  await promise_rejects_exactly(t, thrownError, reader.closed, 'reader.closed should reject with thrownError');
+}, 'aborting the writable side should reject if transformer.cancel() throws');
+
+promise_test(async t => {
+  const ts = new TransformStream({
+    async cancel(reason) {
+      assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason');
+      throw thrownError;
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  const cancelPromise = ts.readable.cancel(originalReason);
+  const closePromise = ts.writable.close();
+  await Promise.all([
+    promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError'),
+    promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError'),
+  ]);
+}, 'closing the writable side should reject if a parallel transformer.cancel() throws');
+
+promise_test(async t => {
+  let controller;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    async cancel(reason) {
+      assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason');
+      controller.error(thrownError);
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  const cancelPromise = ts.readable.cancel(originalReason);
+  const closePromise = ts.writable.close();
+  await Promise.all([
+    promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError'),
+    promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError'),
+  ]);
+}, 'readable.cancel() and a parallel writable.close() should reject if a transformer.cancel() calls controller.error()');
+
+promise_test(async t => {
+  let controller;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    async cancel(reason) {
+      assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason');
+      controller.error(thrownError);
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  const cancelPromise = ts.writable.abort(originalReason);
+  await promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError');
+  const closePromise = ts.readable.cancel(1);
+  await promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError');
+}, 'writable.abort() and readable.cancel() should reject if a transformer.cancel() calls controller.error()');
+
+promise_test(async t => {
+  const cancelReason = new Error('cancel reason');
+  let controller;
+  let cancelPromise;
+  let flushCalled = false;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    flush() {
+      flushCalled = true;
+      cancelPromise = ts.readable.cancel(cancelReason);
+    },
+    cancel: t.unreached_func('cancel should not be called')
+  });
+  await flushAsyncEvents(); // ensure stream is started
+  await ts.writable.close();
+  assert_true(flushCalled, 'flush() was called');
+  await cancelPromise;
+}, 'readable.cancel() should not call cancel() when flush() is already called from writable.close()');
+
+promise_test(async t => {
+  const cancelReason = new Error('cancel reason');
+  const abortReason = new Error('abort reason');
+  let cancelCalls = 0;
+  let controller;
+  let cancelPromise;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    cancel() {
+      if (++cancelCalls === 1) {
+        cancelPromise = ts.readable.cancel(cancelReason);
+      }
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  await flushAsyncEvents(); // ensure stream is started
+  await ts.writable.abort(abortReason);
+  assert_equals(cancelCalls, 1);
+  await cancelPromise;
+  assert_equals(cancelCalls, 1);
+}, 'readable.cancel() should not call cancel() again when already called from writable.abort()');
+
+promise_test(async t => {
+  const cancelReason = new Error('cancel reason');
+  let controller;
+  let closePromise;
+  let cancelCalled = false;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    cancel() {
+      cancelCalled = true;
+      closePromise = ts.writable.close();
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  await flushAsyncEvents(); // ensure stream is started
+  await ts.readable.cancel(cancelReason);
+  assert_true(cancelCalled, 'cancel() was called');
+  await closePromise;
+}, 'writable.close() should not call flush() when cancel() is already called from readable.cancel()');
+
+promise_test(async t => {
+  const cancelReason = new Error('cancel reason');
+  const abortReason = new Error('abort reason');
+  let cancelCalls = 0;
+  let controller;
+  let abortPromise;
+  const ts = new TransformStream({
+    start(c) {
+      controller = c;
+    },
+    cancel() {
+      if (++cancelCalls === 1) {
+        abortPromise = ts.writable.abort(abortReason);
+      }
+    },
+    flush: t.unreached_func('flush should not be called')
+  });
+  await flushAsyncEvents(); // ensure stream is started
+  await promise_rejects_exactly(t, abortReason, ts.readable.cancel(cancelReason));
+  assert_equals(cancelCalls, 1);
+  await promise_rejects_exactly(t, abortReason, abortPromise);
+  assert_equals(cancelCalls, 1);
+}, 'writable.abort() should not call cancel() again when already called from readable.cancel()');