Streams: make writable streams transferrable (#36588)

Making writable streams transferrable, part of
https://github.com/servo/servo/issues/34676

Signed-off-by: gterzian <2792687+gterzian@users.noreply.github.com>

components/script/dom/bindings/structuredclone.rs

@@ -43,6 +43,7 @@ use crate::dom::globalscope::GlobalScope;
 use crate::dom::messageport::MessagePort;
 use crate::dom::readablestream::ReadableStream;
 use crate::dom::types::DOMException;
+use crate::dom::writablestream::WritableStream;
 use crate::realms::{AlreadyInRealm, InRealm, enter_realm};
 use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
 
@@ -61,6 +62,7 @@ pub(super) enum StructuredCloneTags {
     DomPoint = 0xFFFF8005,
     ReadableStream = 0xFFFF8006,
     DomException = 0xFFFF8007,
+    WritableStream = 0xFFFF8008,
     Max = 0xFFFFFFFF,
 }
 
@@ -80,6 +82,7 @@ impl From<TransferrableInterface> for StructuredCloneTags {
         match v {
             TransferrableInterface::MessagePort => StructuredCloneTags::MessagePort,
             TransferrableInterface::ReadableStream => StructuredCloneTags::ReadableStream,
+            TransferrableInterface::WritableStream => StructuredCloneTags::WritableStream,
         }
     }
 }
@@ -259,6 +262,7 @@ fn receiver_for_type(
     match val {
         TransferrableInterface::MessagePort => receive_object::<MessagePort>,
         TransferrableInterface::ReadableStream => receive_object::<ReadableStream>,
+        TransferrableInterface::WritableStream => receive_object::<WritableStream>,
     }
 }
 
@@ -385,6 +389,7 @@ fn transfer_for_type(val: TransferrableInterface) -> TransferOperation {
     match val {
         TransferrableInterface::MessagePort => try_transfer::<MessagePort>,
         TransferrableInterface::ReadableStream => try_transfer::<ReadableStream>,
+        TransferrableInterface::WritableStream => try_transfer::<WritableStream>,
     }
 }
 
@@ -432,6 +437,7 @@ unsafe fn can_transfer_for_type(
     match transferable {
         TransferrableInterface::MessagePort => can_transfer::<MessagePort>(obj, cx),
         TransferrableInterface::ReadableStream => can_transfer::<ReadableStream>(obj, cx),
+        TransferrableInterface::WritableStream => can_transfer::<WritableStream>(obj, cx),
     }
 }
 
@@ -522,7 +528,12 @@ pub(crate) struct StructuredDataReader {
     pub(crate) points: Option<HashMap<DomPointId, DomPoint>>,
     /// A map of serialized exceptions.
     pub(crate) exceptions: Option<HashMap<DomExceptionId, DomException>>,
+
+    /// <https://streams.spec.whatwg.org/#rs-transfer>
     pub(crate) readable_streams: Option<Vec<DomRoot<ReadableStream>>>,
+
+    /// <https://streams.spec.whatwg.org/#ws-transfer>
+    pub(crate) writable_streams: Option<Vec<DomRoot<WritableStream>>>,
 }
 
 /// A data holder for transferred and serialized objects.
@@ -619,6 +630,7 @@ pub(crate) fn read(
         exceptions: data.exceptions.take(),
         errors: DOMErrorRecord { message: None },
         readable_streams: None,
+        writable_streams: None,
     };
     let sc_reader_ptr = &mut sc_reader as *mut _;
     unsafe {

components/script/dom/readablestream.rs

@@ -845,7 +845,7 @@ impl ReadableStream {
     }
 
     #[cfg_attr(crown, allow(crown::unrooted_must_root))]
-    fn new_with_proto(
+    pub(crate) fn new_with_proto(
         global: &GlobalScope,
         proto: Option<SafeHandleObject>,
         can_gc: CanGc,
@@ -1619,7 +1619,7 @@ impl ReadableStream {
 
     /// <https://streams.spec.whatwg.org/#readable-stream-pipe-to>
     #[allow(clippy::too_many_arguments)]
-    fn pipe_to(
+    pub(crate) fn pipe_to(
         &self,
         cx: SafeJSContext,
         global: &GlobalScope,
@@ -1782,7 +1782,7 @@ impl ReadableStream {
     }
 
     /// <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformreadable>
-    fn setup_cross_realm_transform_readable(
+    pub(crate) fn setup_cross_realm_transform_readable(
         &self,
         cx: SafeJSContext,
         port: &MessagePort,

components/script/dom/writablestream.rs

@@ -3,11 +3,13 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use std::cell::{Cell, RefCell};
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::mem;
 use std::ptr::{self};
 use std::rc::Rc;
 
+use base::id::MessagePortId;
+use constellation_traits::MessagePortImpl;
 use dom_struct::dom_struct;
 use js::jsapi::{Heap, JSObject};
 use js::jsval::{JSVal, ObjectValue, UndefinedValue};
@@ -25,13 +27,15 @@ use crate::dom::bindings::conversions::ConversionResult;
 use crate::dom::bindings::error::{Error, Fallible};
 use crate::dom::bindings::reflector::{DomGlobal, Reflector, reflect_dom_object_with_proto};
 use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
+use crate::dom::bindings::structuredclone::{StructuredData, StructuredDataReader};
+use crate::dom::bindings::transferable::Transferable;
 use crate::dom::countqueuingstrategy::{extract_high_water_mark, extract_size_algorithm};
 use crate::dom::domexception::{DOMErrorName, DOMException};
 use crate::dom::globalscope::GlobalScope;
 use crate::dom::messageport::MessagePort;
 use crate::dom::promise::Promise;
 use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
-use crate::dom::readablestream::get_type_and_value_from_message;
+use crate::dom::readablestream::{ReadableStream, get_type_and_value_from_message};
 use crate::dom::writablestreamdefaultcontroller::{
     UnderlyingSinkType, WritableStreamDefaultController,
 };
@@ -1107,3 +1111,85 @@ impl CrossRealmTransformWritable {
         global.disentangle_port(port);
     }
 }
+
+/// <https://streams.spec.whatwg.org/#ws-transfer>
+impl Transferable for WritableStream {
+    type Id = MessagePortId;
+    type Data = MessagePortImpl;
+
+    /// <https://streams.spec.whatwg.org/#ref-for-writablestream%E2%91%A0%E2%91%A4>
+    fn transfer(&self) -> Result<(MessagePortId, MessagePortImpl), ()> {
+        // If ! IsWritableStreamLocked(value) is true, throw a "DataCloneError" DOMException.
+        if self.is_locked() {
+            return Err(());
+        }
+
+        let global = self.global();
+        let realm = enter_realm(&*global);
+        let comp = InRealm::Entered(&realm);
+        let cx = GlobalScope::get_cx();
+        let can_gc = CanGc::note();
+
+        // Let port1 be a new MessagePort in the current Realm.
+        let port_1 = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port_1, None);
+
+        // Let port2 be a new MessagePort in the current Realm.
+        let port_2 = MessagePort::new(&global, can_gc);
+        global.track_message_port(&port_2, None);
+
+        // Entangle port1 and port2.
+        global.entangle_ports(*port_1.message_port_id(), *port_2.message_port_id());
+
+        // Let readable be a new ReadableStream in the current Realm.
+        let readable = ReadableStream::new_with_proto(&global, None, can_gc);
+
+        // Perform ! SetUpCrossRealmTransformReadable(readable, port1).
+        readable.setup_cross_realm_transform_readable(cx, &port_1, can_gc);
+
+        // Let promise be ! ReadableStreamPipeTo(readable, value, false, false, false).
+        let promise = readable.pipe_to(cx, &global, self, false, false, false, comp, can_gc);
+
+        // Set promise.[[PromiseIsHandled]] to true.
+        promise.set_promise_is_handled();
+
+        // Set dataHolder.[[port]] to ! StructuredSerializeWithTransfer(port2, « port2 »).
+        port_2.transfer()
+    }
+
+    /// <https://streams.spec.whatwg.org/#ref-for-writablestream%E2%91%A0%E2%91%A4>
+    fn transfer_receive(
+        owner: &GlobalScope,
+        id: MessagePortId,
+        port_impl: MessagePortImpl,
+    ) -> Result<DomRoot<Self>, ()> {
+        let cx = GlobalScope::get_cx();
+        let can_gc = CanGc::note();
+
+        // Their transfer-receiving steps, given dataHolder and value, are:
+        // Note: dataHolder is used in `structuredclone.rs`, and value is created here.
+        let value = WritableStream::new_with_proto(owner, None, can_gc);
+
+        // Let deserializedRecord be ! StructuredDeserializeWithTransfer(dataHolder.[[port]], the current Realm).
+        // Done with the `Deserialize` derive of `MessagePortImpl`.
+
+        // Let port be deserializedRecord.[[Deserialized]].
+        let transferred_port = MessagePort::transfer_receive(owner, id, port_impl)?;
+
+        // Perform ! SetUpCrossRealmTransformWritable(value, port).
+        value.setup_cross_realm_transform_writable(cx, &transferred_port, can_gc);
+        Ok(value)
+    }
+
+    /// Note: we are relying on the port transfer, so the data returned here are related to the port.
+    fn serialized_storage(data: StructuredData<'_>) -> &mut Option<HashMap<Self::Id, Self::Data>> {
+        match data {
+            StructuredData::Reader(r) => &mut r.port_impls,
+            StructuredData::Writer(w) => &mut w.ports,
+        }
+    }
+
+    fn deserialized_storage(reader: &mut StructuredDataReader) -> &mut Option<Vec<DomRoot<Self>>> {
+        &mut reader.writable_streams
+    }
+}

components/shared/constellation/message_port.rs

@@ -233,6 +233,8 @@ pub enum Transferrable {
     MessagePort,
     /// The `ReadableStream` interface.
     ReadableStream,
+    /// The `WritableStream` interface.
+    WritableStream,
 }
 
 impl StructuredSerializedData {
@@ -243,6 +245,7 @@ impl StructuredSerializedData {
         match val {
             Transferrable::MessagePort => is_field_empty(&self.ports),
             Transferrable::ReadableStream => is_field_empty(&self.ports),
+            Transferrable::WritableStream => is_field_empty(&self.ports),
         }
     }
 

tests/wpt/meta/streams/transferable/deserialize-error.window.js.ini

@@ -1,7 +1,2 @@
 [deserialize-error.window.html]
   expected: ERROR
-  [a WritableStream deserialization failure should result in a DataCloneError]
-    expected: TIMEOUT
-
-  [a ReadableStream deserialization failure should result in a DataCloneError]
-    expected: TIMEOUT

tests/wpt/meta/streams/transferable/transfer-with-messageport.window.js.ini

@@ -1,13 +1,7 @@
 [transfer-with-messageport.window.html]
-  [Transferring a MessagePort with a WritableStream should set `.ports`]
-    expected: FAIL
-
   [Transferring a MessagePort with a TransformStream should set `.ports`]
     expected: FAIL
 
-  [Transferring a MessagePort with a WritableStream should set `.ports`, advanced]
-    expected: FAIL
-
   [Transferring a MessagePort with a TransformStream should set `.ports`, advanced]
     expected: FAIL
 

tests/wpt/meta/streams/transferable/writable-stream.html.ini

@@ -1,21 +1,4 @@
 [writable-stream.html]
-  [window.postMessage should be able to transfer a WritableStream]
-    expected: FAIL
-
+  expected: ERROR
   [window.postMessage should be able to transfer a {readable, writable} pair]
     expected: FAIL
-
-  [desiredSize for a newly-transferred stream should be 1]
-    expected: FAIL
-
-  [effective queue size of a transferred writable should be 2]
-    expected: FAIL
-
-  [second write should wait for first underlying write to complete]
-    expected: FAIL
-
-  [abort() should work]
-    expected: FAIL
-
-  [writing a unclonable object should error the stream]
-    expected: FAIL