LibWeb: Implement WritableStream transfer

Author: https://github.com/trflynn89
Commit: cca08ad833
Pull-request: https://github.com/LadybirdBrowser/ladybird/pull/4828
Reviewed-by: https://github.com/shannonbooth ✅

Libraries/LibWeb/HTML/StructuredSerialize.cpp

@@ -50,6 +50,7 @@
 #include <LibWeb/HTML/MessagePort.h>
 #include <LibWeb/HTML/StructuredSerialize.h>
 #include <LibWeb/Streams/ReadableStream.h>
+#include <LibWeb/Streams/WritableStream.h>
 #include <LibWeb/WebIDL/DOMException.h>
 #include <LibWeb/WebIDL/ExceptionOr.h>
 
@@ -1296,6 +1297,8 @@ static bool is_interface_exposed_on_target_realm(TransferType name, JS::Realm& r
         return intrinsics.is_exposed("MessagePort"sv);
     case TransferType::ReadableStream:
         return intrinsics.is_exposed("ReadableStream"sv);
+    case TransferType::WritableStream:
+        return intrinsics.is_exposed("WritableStream"sv);
     case TransferType::Unknown:
         dbgln("Unknown interface type for transfer: {}", to_underlying(name));
         break;
@@ -1318,6 +1321,11 @@ static WebIDL::ExceptionOr<GC::Ref<Bindings::PlatformObject>> create_transferred
         TRY(readable_stream->transfer_receiving_steps(transfer_data_holder));
         return readable_stream;
     }
+    case TransferType::WritableStream: {
+        auto writable_stream = target_realm.create<Streams::WritableStream>(target_realm);
+        TRY(writable_stream->transfer_receiving_steps(transfer_data_holder));
+        return writable_stream;
+    }
     case TransferType::ArrayBuffer:
     case TransferType::ResizableArrayBuffer:
         dbgln("ArrayBuffer ({}) is not a platform object.", to_underlying(name));

Libraries/LibWeb/HTML/StructuredSerialize.h

@@ -50,6 +50,7 @@ enum class TransferType : u8 {
     ArrayBuffer = 2,
     ResizableArrayBuffer = 3,
     ReadableStream = 4,
+    WritableStream = 5,
 };
 
 WebIDL::ExceptionOr<SerializationRecord> structured_serialize(JS::VM& vm, JS::Value);

Libraries/LibWeb/Streams/WritableStream.cpp

@@ -7,7 +7,11 @@
 #include <LibJS/Runtime/PromiseCapability.h>
 #include <LibWeb/Bindings/Intrinsics.h>
 #include <LibWeb/Bindings/WritableStreamPrototype.h>
+#include <LibWeb/HTML/MessagePort.h>
+#include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
 #include <LibWeb/Streams/AbstractOperations.h>
+#include <LibWeb/Streams/ReadableStream.h>
+#include <LibWeb/Streams/ReadableStreamOperations.h>
 #include <LibWeb/Streams/UnderlyingSink.h>
 #include <LibWeb/Streams/WritableStream.h>
 #include <LibWeb/Streams/WritableStreamDefaultController.h>
@@ -51,6 +55,34 @@ WebIDL::ExceptionOr<GC::Ref<WritableStream>> WritableStream::construct_impl(JS::
     return writable_stream;
 }
 
+WritableStream::WritableStream(JS::Realm& realm)
+    : Bindings::PlatformObject(realm)
+{
+}
+
+void WritableStream::initialize(JS::Realm& realm)
+{
+    WEB_SET_PROTOTYPE_FOR_INTERFACE(WritableStream);
+    Base::initialize(realm);
+}
+
+void WritableStream::visit_edges(Cell::Visitor& visitor)
+{
+    Base::visit_edges(visitor);
+    visitor.visit(m_close_request);
+    visitor.visit(m_controller);
+    visitor.visit(m_in_flight_write_request);
+    visitor.visit(m_in_flight_close_request);
+    if (m_pending_abort_request.has_value()) {
+        visitor.visit(m_pending_abort_request->promise);
+        visitor.visit(m_pending_abort_request->reason);
+    }
+    visitor.visit(m_stored_error);
+    visitor.visit(m_writer);
+    for (auto& write_request : m_write_requests)
+        visitor.visit(write_request);
+}
+
 // https://streams.spec.whatwg.org/#ws-locked
 bool WritableStream::locked() const
 {
@@ -101,32 +133,62 @@ WebIDL::ExceptionOr<GC::Ref<WritableStreamDefaultWriter>> WritableStream::get_wr
     return acquire_writable_stream_default_writer(*this);
 }
 
-WritableStream::WritableStream(JS::Realm& realm)
+// https://streams.spec.whatwg.org/#ref-for-transfer-steps①
-    : Bindings::PlatformObject(realm)
+WebIDL::ExceptionOr<void> WritableStream::transfer_steps(HTML::TransferDataHolder& data_holder)
 {
+    auto& realm = this->realm();
+    auto& vm = realm.vm();
+
+    HTML::TemporaryExecutionContext execution_context { realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
+
+    // 1. If ! IsWritableStreamLocked(value) is true, throw a "DataCloneError" DOMException.
+    if (is_writable_stream_locked(*this))
+        return WebIDL::DataCloneError::create(realm, "Cannot transfer locked WritableStream"_string);
+
+    // 2. Let port1 be a new MessagePort in the current Realm.
+    auto port1 = HTML::MessagePort::create(realm);
+
+    // 3. Let port2 be a new MessagePort in the current Realm.
+    auto port2 = HTML::MessagePort::create(realm, HTML::TransferType::WritableStream);
+
+    // 4. Entangle port1 and port2.
+    port1->entangle_with(port2);
+
+    // 5. Let readable be a new ReadableStream in the current Realm.
+    auto readable = realm.create<ReadableStream>(realm);
+
+    // 6. Perform ! SetUpCrossRealmTransformReadable(readable, port1).
+    set_up_cross_realm_transform_readable(realm, readable, port1);
+
+    // 7. Let promise be ! ReadableStreamPipeTo(readable, value, false, false, false).
+    auto promise = readable_stream_pipe_to(readable, *this, false, false, false);
+
+    // 8. Set promise.[[PromiseIsHandled]] to true.
+    WebIDL::mark_promise_as_handled(promise);
+
+    // 9. Set dataHolder.[[port]] to ! StructuredSerializeWithTransfer(port2, « port2 »).
+    auto result = MUST(HTML::structured_serialize_with_transfer(vm, port2, { { GC::Root { port2 } } }));
+    data_holder = move(result.transfer_data_holders.first());
+
+    return {};
 }
 
-void WritableStream::initialize(JS::Realm& realm)
+// https://streams.spec.whatwg.org/#ref-for-transfer-receiving-steps①
+WebIDL::ExceptionOr<void> WritableStream::transfer_receiving_steps(HTML::TransferDataHolder& data_holder)
 {
-    WEB_SET_PROTOTYPE_FOR_INTERFACE(WritableStream);
+    auto& realm = this->realm();
-    Base::initialize(realm);
-}
 
-void WritableStream::visit_edges(Cell::Visitor& visitor)
+    HTML::TemporaryExecutionContext execution_context { realm, HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
-{
+
-    Base::visit_edges(visitor);
+    // 1. Let deserializedRecord be ! StructuredDeserializeWithTransfer(dataHolder.[[port]], the current Realm).
-    visitor.visit(m_close_request);
+    // 2. Let port be deserializedRecord.[[Deserialized]].
-    visitor.visit(m_controller);
+    auto port = HTML::MessagePort::create(realm);
-    visitor.visit(m_in_flight_write_request);
+    TRY(port->transfer_receiving_steps(data_holder));
-    visitor.visit(m_in_flight_close_request);
+
-    if (m_pending_abort_request.has_value()) {
+    // 3. Perform ! SetUpCrossRealmTransformWritable(value, port).
-        visitor.visit(m_pending_abort_request->promise);
+    set_up_cross_realm_transform_writable(realm, *this, port);
-        visitor.visit(m_pending_abort_request->reason);
+
-    }
+    return {};
-    visitor.visit(m_stored_error);
-    visitor.visit(m_writer);
-    for (auto& write_request : m_write_requests)
-        visitor.visit(write_request);
 }
 
 }

Libraries/LibWeb/Streams/WritableStream.h

@@ -10,6 +10,7 @@
 #include <AK/SinglyLinkedList.h>
 #include <LibJS/Forward.h>
 #include <LibWeb/Bindings/PlatformObject.h>
+#include <LibWeb/Bindings/Transferable.h>
 #include <LibWeb/Forward.h>
 #include <LibWeb/Streams/QueuingStrategy.h>
 #include <LibWeb/WebIDL/Promise.h>
@@ -32,7 +33,9 @@ struct PendingAbortRequest {
 };
 
 // https://streams.spec.whatwg.org/#writablestream
-class WritableStream final : public Bindings::PlatformObject {
+class WritableStream final
+    : public Bindings::PlatformObject
+    , public Bindings::Transferable {
     WEB_PLATFORM_OBJECT(WritableStream, Bindings::PlatformObject);
     GC_DECLARE_ALLOCATOR(WritableStream);
 
@@ -85,6 +88,11 @@ public:
 
     SinglyLinkedList<GC::Ref<WebIDL::Promise>>& write_requests() { return m_write_requests; }
 
+    // ^Transferable
+    virtual WebIDL::ExceptionOr<void> transfer_steps(HTML::TransferDataHolder&) override;
+    virtual WebIDL::ExceptionOr<void> transfer_receiving_steps(HTML::TransferDataHolder&) override;
+    virtual HTML::TransferType primary_interface() const override { return HTML::TransferType::WritableStream; }
+
 private:
     explicit WritableStream(JS::Realm&);
 
@@ -104,10 +112,6 @@ private:
     // A WritableStreamDefaultController created with the ability to control the state and queue of this stream
     GC::Ptr<WritableStreamDefaultController> m_controller;
 
-    // https://streams.spec.whatwg.org/#writablestream-detached
-    // A boolean flag set to true when the stream is transferred
-    bool m_detached { false };
-
     // https://streams.spec.whatwg.org/#writablestream-inflightwriterequest
     // A slot set to the promise for the current in-flight write operation while the underlying sink's write algorithm is executing and has not yet fulfilled, used to prevent reentrant calls
     GC::Ptr<WebIDL::Promise> m_in_flight_write_request;

Tests/LibWeb/Text/expected/wpt-import/streams/transferable/writable-stream.txt

@@ -0,0 +1,13 @@
+Harness status: OK
+
+Found 8 tests
+
+8 Pass
+Pass	window.postMessage should be able to transfer a WritableStream
+Pass	a locked WritableStream should not be transferable
+Pass	window.postMessage should be able to transfer a {readable, writable} pair
+Pass	desiredSize for a newly-transferred stream should be 1
+Pass	effective queue size of a transferred writable should be 2
+Pass	second write should wait for first underlying write to complete
+Pass	abort() should work
+Pass	writing a unclonable object should error the stream

Tests/LibWeb/Text/input/wpt-import/streams/transferable/writable-stream.html

@@ -0,0 +1,146 @@
+<!DOCTYPE html>
+<meta charset="utf-8">
+<script src="../../resources/testharness.js"></script>
+<script src="../../resources/testharnessreport.js"></script>
+<script src="resources/helpers.js"></script>
+<script src="../resources/test-utils.js"></script>
+<script src="../resources/recording-streams.js"></script>
+<script>
+'use strict';
+
+promise_test(t => {
+  const orig = new WritableStream();
+  const promise = new Promise(resolve => {
+    addEventListener('message', t.step_func(evt => {
+      const transferred = evt.data;
+      assert_equals(transferred.constructor, WritableStream,
+                    'transferred should be a WritableStream in this realm');
+      assert_true(transferred instanceof WritableStream,
+                  'instanceof check should pass');
+
+      // Perform a brand-check on |transferred|.
+      const writer = WritableStream.prototype.getWriter.call(transferred);
+      resolve();
+    }), {once: true});
+  });
+  postMessage(orig, '*', [orig]);
+  assert_true(orig.locked, 'the original stream should be locked');
+  return promise;
+}, 'window.postMessage should be able to transfer a WritableStream');
+
+test(() => {
+  const ws = new WritableStream();
+  const writer = ws.getWriter();
+  assert_throws_dom('DataCloneError', () => postMessage(ws, '*', [ws]),
+                    'postMessage should throw');
+}, 'a locked WritableStream should not be transferable');
+
+promise_test(t => {
+  const {writable, readable} = new TransformStream();
+  const promise = new Promise(resolve => {
+    addEventListener('message', t.step_func(async evt => {
+      const {writable, readable} = evt.data;
+      const reader = readable.getReader();
+      const writer = writable.getWriter();
+      const writerPromises = Promise.all([
+        writer.write('hi'),
+        writer.close(),
+      ]);
+      const {value, done} = await reader.read();
+      assert_false(done, 'we should not be done');
+      assert_equals(value, 'hi', 'chunk should have been delivered');
+      const readResult = await reader.read();
+      assert_true(readResult.done, 'readable should be closed');
+      await writerPromises;
+      resolve();
+    }), {once: true});
+  });
+  postMessage({writable, readable}, '*', [writable, readable]);
+  return promise;
+}, 'window.postMessage should be able to transfer a {readable, writable} pair');
+
+function transfer(stream) {
+  return new Promise(resolve => {
+    addEventListener('message', evt => resolve(evt.data), { once: true });
+    postMessage(stream, '*', [stream]);
+  });
+}
+
+promise_test(async () => {
+  const orig = new WritableStream(
+    {}, new ByteLengthQueuingStrategy({ highWaterMark: 65536 }));
+  const transferred = await transfer(orig);
+  const writer = transferred.getWriter();
+  assert_equals(writer.desiredSize, 1, 'desiredSize should be 1');
+}, 'desiredSize for a newly-transferred stream should be 1');
+
+promise_test(async () => {
+  const orig = new WritableStream({
+    write() {
+      return new Promise(() => {});
+    }
+  });
+  const transferred = await transfer(orig);
+  const writer = transferred.getWriter();
+  await writer.write('a');
+  assert_equals(writer.desiredSize, 1, 'desiredSize should be 1');
+}, 'effective queue size of a transferred writable should be 2');
+
+promise_test(async () => {
+  const [writeCalled, resolveWriteCalled] = makePromiseAndResolveFunc();
+  let resolveWrite;
+  const orig = new WritableStream({
+    write() {
+      resolveWriteCalled();
+      return new Promise(resolve => {
+        resolveWrite = resolve;
+      });
+    }
+  });
+  const transferred = await transfer(orig);
+  const writer = transferred.getWriter();
+  await writer.write('a');
+  let writeDone = false;
+  const writePromise = writer.write('b').then(() => {
+    writeDone = true;
+  });
+  await writeCalled;
+  assert_false(writeDone, 'second write should not have resolved yet');
+  resolveWrite();
+  await writePromise; // (makes sure this resolves)
+}, 'second write should wait for first underlying write to complete');
+
+async function transferredWritableStreamWithAbortPromise() {
+  const [abortCalled, resolveAbortCalled] = makePromiseAndResolveFunc();
+  const orig = recordingWritableStream({
+    abort() {
+      resolveAbortCalled();
+    }
+  });
+  const transferred = await transfer(orig);
+  return { orig, transferred, abortCalled };
+}
+
+promise_test(async t => {
+  const { orig, transferred, abortCalled } = await transferredWritableStreamWithAbortPromise();
+  transferred.abort('p');
+  await abortCalled;
+  assert_array_equals(orig.events, ['abort', 'p'],
+                      'abort() should have been called');
+}, 'abort() should work');
+
+promise_test(async t => {
+  const { orig, transferred, abortCalled } = await transferredWritableStreamWithAbortPromise();
+  const writer = transferred.getWriter();
+  // A WritableStream object cannot be cloned.
+  await promise_rejects_dom(t, 'DataCloneError', writer.write(new WritableStream()),
+                            'the write should reject');
+  await promise_rejects_dom(t, 'DataCloneError', writer.closed,
+                            'the stream should be errored');
+  await abortCalled;
+  assert_equals(orig.events.length, 2, 'abort should have been called');
+  assert_equals(orig.events[0], 'abort', 'first event should be abort');
+  assert_equals(orig.events[1].name, 'DataCloneError',
+                'reason should be a DataCloneError');
+}, 'writing a unclonable object should error the stream');
+</script>