AbortController: integrate with stream piping. (#37244)

Start using abort signal in Use in
https://streams.spec.whatwg.org/#readablestream-pipe-to-signal

Part of https://github.com/servo/servo/issues/34866

Will also cover https://github.com/servo/servo/issues/37230 and
https://github.com/servo/servo/issues/37232

---------

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

components/script/dom/abortcontroller.rs

@@ -10,6 +10,7 @@ use crate::dom::bindings::codegen::Bindings::AbortControllerBinding::AbortContro
 use crate::dom::bindings::reflector::{Reflector, reflect_dom_object_with_proto};
 use crate::dom::bindings::root::{Dom, DomRoot};
 use crate::dom::globalscope::GlobalScope;
+use crate::realms::InRealm;
 use crate::script_runtime::{CanGc, JSContext};
 
 /// <https://dom.spec.whatwg.org/#abortcontroller>
@@ -23,23 +24,27 @@ pub(crate) struct AbortController {
 
 impl AbortController {
     /// <https://dom.spec.whatwg.org/#dom-abortcontroller-abortcontroller>
-    fn new_inherited() -> AbortController {
-        // The new AbortController() constructor steps are:
-        // Let signal be a new AbortSignal object.
+    fn new_inherited(signal: &AbortSignal) -> AbortController {
+        // Note: continuation of the constructor steps.
+
         // Set this’s signal to signal.
         AbortController {
             reflector_: Reflector::new(),
-            signal: Dom::from_ref(&AbortSignal::new_inherited()),
+            signal: Dom::from_ref(signal),
         }
     }
 
+    /// <https://dom.spec.whatwg.org/#dom-abortcontroller-abortcontroller>
     fn new_with_proto(
         global: &GlobalScope,
         proto: Option<HandleObject>,
         can_gc: CanGc,
     ) -> DomRoot<AbortController> {
+        // The new AbortController() constructor steps are:
+        // Let signal be a new AbortSignal object.
+        let signal = AbortSignal::new_with_proto(global, None, can_gc);
         reflect_dom_object_with_proto(
-            Box::new(AbortController::new_inherited()),
+            Box::new(AbortController::new_inherited(&signal)),
             global,
             proto,
             can_gc,
@@ -47,9 +52,9 @@ impl AbortController {
     }
 
     /// <https://dom.spec.whatwg.org/#abortcontroller-signal-abort>
-    fn signal_abort(&self, cx: JSContext, reason: HandleValue, can_gc: CanGc) {
+    fn signal_abort(&self, cx: JSContext, reason: HandleValue, realm: InRealm, can_gc: CanGc) {
         // signal abort on controller’s signal with reason if it is given.
-        self.signal.signal_abort(cx, reason, can_gc);
+        self.signal.signal_abort(cx, reason, realm, can_gc);
     }
 }
 
@@ -64,10 +69,10 @@ impl AbortControllerMethods<crate::DomTypeHolder> for AbortController {
     }
 
     /// <https://dom.spec.whatwg.org/#dom-abortcontroller-abort>
-    fn Abort(&self, cx: JSContext, reason: HandleValue, can_gc: CanGc) {
+    fn Abort(&self, cx: JSContext, reason: HandleValue, realm: InRealm, can_gc: CanGc) {
         // The abort(reason) method steps are
         // to signal abort on this with reason if it is given.
-        self.signal_abort(cx, reason, can_gc);
+        self.signal_abort(cx, reason, realm, can_gc);
     }
 
     /// <https://dom.spec.whatwg.org/#dom-abortcontroller-signal>

components/script/dom/abortsignal.rs

@@ -3,7 +3,6 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use std::cell::RefCell;
-use std::mem;
 
 use dom_struct::dom_struct;
 use js::jsapi::{ExceptionStackBehavior, Heap, JS_SetPendingException};
@@ -17,18 +16,23 @@ use crate::dom::bindings::reflector::{DomGlobal, reflect_dom_object_with_proto};
 use crate::dom::bindings::root::DomRoot;
 use crate::dom::eventtarget::EventTarget;
 use crate::dom::globalscope::GlobalScope;
-use crate::script_runtime::{CanGc, JSContext};
+use crate::dom::readablestream::PipeTo;
+use crate::realms::InRealm;
+use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
+
+impl js::gc::Rootable for AbortAlgorithm {}
 
 /// <https://dom.spec.whatwg.org/#abortcontroller-api-integration>
 /// TODO: implement algorithms at call point,
 /// in order to integrate the abort signal with its various use cases.
-#[derive(JSTraceable, MallocSizeOf)]
+#[derive(Clone, JSTraceable, MallocSizeOf)]
+#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
 #[allow(dead_code)]
-enum AbortAlgorithm {
+pub(crate) enum AbortAlgorithm {
     /// <https://dom.spec.whatwg.org/#add-an-event-listener>
     DomEventLister,
     /// <https://streams.spec.whatwg.org/#readable-stream-pipe-to>
-    StreamPiping,
+    StreamPiping(PipeTo),
     /// <https://fetch.spec.whatwg.org/#dom-global-fetch>
     Fetch,
 }
@@ -47,7 +51,7 @@ pub(crate) struct AbortSignal {
 }
 
 impl AbortSignal {
-    pub(crate) fn new_inherited() -> AbortSignal {
+    fn new_inherited() -> AbortSignal {
         AbortSignal {
             eventtarget: EventTarget::new_inherited(),
             abort_reason: Default::default(),
@@ -56,7 +60,7 @@ impl AbortSignal {
     }
 
     #[allow(dead_code)]
-    fn new_with_proto(
+    pub(crate) fn new_with_proto(
         global: &GlobalScope,
         proto: Option<HandleObject>,
         can_gc: CanGc,
@@ -70,7 +74,15 @@ impl AbortSignal {
     }
 
     /// <https://dom.spec.whatwg.org/#abortsignal-signal-abort>
-    pub(crate) fn signal_abort(&self, cx: JSContext, reason: HandleValue, can_gc: CanGc) {
+    pub(crate) fn signal_abort(
+        &self,
+        cx: SafeJSContext,
+        reason: HandleValue,
+        realm: InRealm,
+        can_gc: CanGc,
+    ) {
+        let global = self.global();
+
         // If signal is aborted, then return.
         if self.Aborted() {
             return;
@@ -84,7 +96,7 @@ impl AbortSignal {
         } else {
             // otherwise to a new "AbortError" DOMException.
             rooted!(in(*cx) let mut rooted_error = UndefinedValue());
-            Error::Abort.to_jsval(cx, &self.global(), rooted_error.handle_mut(), can_gc);
+            Error::Abort.to_jsval(cx, &global, rooted_error.handle_mut(), can_gc);
             self.abort_reason.set(rooted_error.get())
         }
 
@@ -93,23 +105,60 @@ impl AbortSignal {
         // TODO: #36936
 
         // Run the abort steps for signal.
-        self.run_the_abort_steps(can_gc);
+        self.run_the_abort_steps(cx, &global, realm, can_gc);
 
         // For each dependentSignal of dependentSignalsToAbort, run the abort steps for dependentSignal.
         // TODO: #36936
     }
 
+    /// <https://dom.spec.whatwg.org/#abortsignal-add>
+    pub(crate) fn add(&self, algorithm: &AbortAlgorithm) {
+        // If signal is aborted, then return.
+        if self.aborted() {
+            return;
+        }
+
+        // Append algorithm to signal’s abort algorithms.
+        self.abort_algorithms.borrow_mut().push(algorithm.clone());
+    }
+
+    /// Run a specific abort algorithm.
+    pub(crate) fn run_abort_algorithm(
+        &self,
+        cx: SafeJSContext,
+        global: &GlobalScope,
+        algorithm: &AbortAlgorithm,
+        realm: InRealm,
+        can_gc: CanGc,
+    ) {
+        match algorithm {
+            AbortAlgorithm::StreamPiping(pipe) => {
+                rooted!(in(*cx) let mut reason = UndefinedValue());
+                reason.set(self.abort_reason.get());
+                pipe.abort_with_reason(cx, global, reason.handle(), realm, can_gc);
+            },
+            _ => {
+                // TODO: match on variant and implement algo steps.
+                // See the various items of #34866
+            },
+        }
+    }
+
     /// <https://dom.spec.whatwg.org/#run-the-abort-steps>
-    fn run_the_abort_steps(&self, can_gc: CanGc) {
+    fn run_the_abort_steps(
+        &self,
+        cx: SafeJSContext,
+        global: &GlobalScope,
+        realm: InRealm,
+        can_gc: CanGc,
+    ) {
         // For each algorithm of signal’s abort algorithms: run algorithm.
-        let algos = mem::take(&mut *self.abort_algorithms.borrow_mut());
-        for _algo in algos {
-            // TODO: match on variant and implement algo steps.
-            // See the various items of #34866
+        for algo in self.abort_algorithms.borrow().iter() {
+            self.run_abort_algorithm(cx, global, algo, realm, can_gc);
         }
 
         // Empty signal’s abort algorithms.
-        // Done above with `take`.
+        self.abort_algorithms.borrow_mut().clear();
 
         // Fire an event named abort at signal.
         self.upcast::<EventTarget>()
@@ -117,7 +166,7 @@ impl AbortSignal {
     }
 
     /// <https://dom.spec.whatwg.org/#abortsignal-aborted>
-    fn aborted(&self) -> bool {
+    pub(crate) fn aborted(&self) -> bool {
         // An AbortSignal object is aborted when its abort reason is not undefined.
         !self.abort_reason.get().is_undefined()
     }
@@ -131,7 +180,7 @@ impl AbortSignalMethods<crate::DomTypeHolder> for AbortSignal {
     }
 
     /// <https://dom.spec.whatwg.org/#dom-abortsignal-reason>
-    fn Reason(&self, _cx: JSContext, mut rval: MutableHandleValue) {
+    fn Reason(&self, _cx: SafeJSContext, mut rval: MutableHandleValue) {
         // The reason getter steps are to return this’s abort reason.
         rval.set(self.abort_reason.get());
     }

components/script/dom/promise.rs

@@ -11,6 +11,7 @@
 //! native Promise values that refer to the same JS value yet are distinct native objects
 //! (ie. address equality for the native objects is meaningless).
 
+use std::cell::{Cell, RefCell};
 use std::ptr;
 use std::rc::Rc;
 
@@ -22,7 +23,7 @@ use js::jsapi::{
     NewFunctionWithReserved, PromiseState, PromiseUserInputEventHandlingState, RemoveRawValueRoot,
     SetFunctionNativeReserved,
 };
-use js::jsval::{Int32Value, JSVal, ObjectValue, UndefinedValue};
+use js::jsval::{Int32Value, JSVal, NullValue, ObjectValue, UndefinedValue};
 use js::rust::wrappers::{
     AddPromiseReactions, CallOriginalPromiseReject, CallOriginalPromiseResolve,
     GetPromiseIsHandled, GetPromiseState, IsPromiseObject, NewPromiseObject, RejectPromise,
@@ -35,7 +36,7 @@ use crate::dom::bindings::error::{Error, ErrorToJsval};
 use crate::dom::bindings::reflector::{DomGlobal, DomObject, MutDomObject, Reflector};
 use crate::dom::bindings::settings_stack::AutoEntryScript;
 use crate::dom::globalscope::GlobalScope;
-use crate::dom::promisenativehandler::PromiseNativeHandler;
+use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
 use crate::realms::{AlreadyInRealm, InRealm, enter_realm};
 use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
 use crate::script_thread::ScriptThread;
@@ -405,3 +406,221 @@ impl FromJSValConvertibleRc for Promise {
         Ok(ConversionResult::Success(promise))
     }
 }
+
+/// The success steps of <https://webidl.spec.whatwg.org/#wait-for-all>
+type WaitForAllSuccessSteps = Rc<dyn Fn(Vec<HandleValue>)>;
+
+/// The failure steps of <https://webidl.spec.whatwg.org/#wait-for-all>
+type WaitForAllFailureSteps = Rc<dyn Fn(HandleValue)>;
+
+/// The fulfillment handler for the list of promises in
+/// <https://webidl.spec.whatwg.org/#wait-for-all>.
+#[derive(JSTraceable, MallocSizeOf)]
+#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
+struct WaitForAllFulfillmentHandler {
+    /// The steps to call when all promises are resolved.
+    #[ignore_malloc_size_of = "Rc is hard"]
+    #[no_trace]
+    success_steps: WaitForAllSuccessSteps,
+
+    /// The results of the promises.
+    #[ignore_malloc_size_of = "Rc is hard"]
+    #[allow(clippy::vec_box)]
+    result: Rc<RefCell<Vec<Box<Heap<JSVal>>>>>,
+
+    /// The index identifying which promise this handler is attached to.
+    promise_index: usize,
+
+    /// A count of fulfilled promises.
+    #[ignore_malloc_size_of = "Rc is hard"]
+    fulfilled_count: Rc<RefCell<usize>>,
+}
+
+impl Callback for WaitForAllFulfillmentHandler {
+    #[allow(unsafe_code)]
+    fn callback(&self, _cx: SafeJSContext, v: HandleValue, _realm: InRealm, _can_gc: CanGc) {
+        // Let fulfillmentHandler be the following steps given arg:
+
+        let equals_total = {
+            // Set result[promiseIndex] to arg.
+            let result = self.result.borrow_mut();
+            result[self.promise_index].set(v.get());
+
+            // Set fulfilledCount to fulfilledCount + 1.
+            let mut fulfilled_count = self.fulfilled_count.borrow_mut();
+            *fulfilled_count += 1;
+
+            *fulfilled_count == result.len()
+        };
+
+        // If fulfilledCount equals total, then perform successSteps given result.
+        if equals_total {
+            // Safety: the values are kept alive by the Heap
+            // while their handles are passed to the the success steps.
+            let result_handles: Vec<HandleValue> = unsafe {
+                self.result
+                    .borrow()
+                    .iter()
+                    .map(|val| HandleValue::from_raw(val.handle()))
+                    .collect()
+            };
+            (self.success_steps)(result_handles);
+        }
+    }
+}
+
+/// The rejection handler for the list of promises in
+/// <https://webidl.spec.whatwg.org/#wait-for-all>.
+#[derive(Clone, JSTraceable, MallocSizeOf)]
+struct WaitForAllRejectionHandler {
+    /// The steps to call if any promise rejects.
+    #[ignore_malloc_size_of = "Rc is hard"]
+    #[no_trace]
+    failure_steps: WaitForAllFailureSteps,
+
+    /// Whether any promises have been rejected already.
+    rejected: Cell<bool>,
+}
+
+impl Callback for WaitForAllRejectionHandler {
+    fn callback(&self, _cx: SafeJSContext, v: HandleValue, _realm: InRealm, _can_gc: CanGc) {
+        // Let rejectionHandlerSteps be the following steps given arg:
+
+        if self.rejected.replace(true) {
+            // If rejected is true, abort these steps.
+            return;
+        }
+
+        // Set rejected to true.
+        // Done above with `replace`.
+        (self.failure_steps)(v);
+    }
+}
+
+/// <https://webidl.spec.whatwg.org/#wait-for-all>
+pub(crate) fn wait_for_all(
+    cx: SafeJSContext,
+    global: &GlobalScope,
+    promises: Vec<Rc<Promise>>,
+    success_steps: WaitForAllSuccessSteps,
+    failure_steps: WaitForAllFailureSteps,
+    realm: InRealm,
+    can_gc: CanGc,
+) {
+    // Let fulfilledCount be 0.
+    let fulfilled_count: Rc<RefCell<usize>> = Default::default();
+
+    // Let rejected be false.
+    // Note: done below when constructing a rejection handler.
+
+    // Let rejectionHandlerSteps be the following steps given arg:
+    // Note: implemented with the `WaitForAllRejectionHandler`.
+
+    // Let rejectionHandler be CreateBuiltinFunction(rejectionHandlerSteps, « »):
+    // Note: done as part of attaching the `WaitForAllRejectionHandler` as native rejection handler.
+    let rejection_handler = WaitForAllRejectionHandler {
+        failure_steps,
+        rejected: Default::default(),
+    };
+
+    // Let total be promises’s size.
+    // Note: done using the len of result.
+
+    // If total is 0, then:
+    // Queue a microtask to perform successSteps given « ».
+    // TODO: #37259
+
+    // Let index be 0.
+    // Note: done with `enumerate` below.
+
+    // Let result be a list containing total null values.
+    let result: Rc<RefCell<Vec<Box<Heap<JSVal>>>>> = Default::default();
+
+    // For each promise of promises:
+    for (promise_index, promise) in promises.into_iter().enumerate() {
+        let result = result.clone();
+
+        {
+            // Note: adding a null value for this promise result.
+            let mut result_list = result.borrow_mut();
+            rooted!(in(*cx) let null_value = NullValue());
+            result_list.push(Heap::boxed(null_value.get()));
+        }
+
+        // Let promiseIndex be index.
+        // Note: done with `enumerate` above.
+
+        // Let fulfillmentHandler be the following steps given arg:
+        // Note: implemented with the `WaitForAllFulFillmentHandler`.
+
+        // Let fulfillmentHandler be CreateBuiltinFunction(fulfillmentHandler, « »):
+        // Note: passed below to avoid the need to root it.
+
+        // Perform PerformPromiseThen(promise, fulfillmentHandler, rejectionHandler).
+        let handler = PromiseNativeHandler::new(
+            global,
+            Some(Box::new(WaitForAllFulfillmentHandler {
+                success_steps: success_steps.clone(),
+                result,
+                promise_index,
+                fulfilled_count: fulfilled_count.clone(),
+            })),
+            Some(Box::new(rejection_handler.clone())),
+            can_gc,
+        );
+        promise.append_native_handler(&handler, realm, can_gc);
+
+        // Set index to index + 1.
+        // Note: done above with `enumerate`.
+    }
+}
+
+/// <https://webidl.spec.whatwg.org/#waiting-for-all-promise>
+pub(crate) fn wait_for_all_promise(
+    cx: SafeJSContext,
+    global: &GlobalScope,
+    promises: Vec<Rc<Promise>>,
+    realm: InRealm,
+    can_gc: CanGc,
+) -> Rc<Promise> {
+    // Let promise be a new promise of type Promise<sequence<T>> in realm.
+    let promise = Promise::new(global, can_gc);
+    let success_promise = promise.clone();
+    let failure_promise = promise.clone();
+
+    // Let successSteps be the following steps, given results:
+    let success_steps = Rc::new(move |results: Vec<HandleValue>| {
+        // Resolve promise with results.
+        success_promise.resolve_native(&results, can_gc);
+    });
+
+    // Let failureSteps be the following steps, given reason:
+    let failure_steps = Rc::new(move |reason: HandleValue| {
+        // Reject promise with reason.
+        failure_promise.reject_native(&reason, can_gc);
+    });
+
+    if promises.is_empty() {
+        // Note: part of `wait_for_all`.
+        // Done here by using `resolve_native`.
+        // TODO: #37259
+        // If total is 0, then:
+        // Queue a microtask to perform successSteps given « ».
+        let empty_list: Vec<HandleValue> = vec![];
+        promise.resolve_native(&empty_list, can_gc);
+    } else {
+        // Wait for all with promises, given successSteps and failureSteps.
+        wait_for_all(
+            cx,
+            global,
+            promises,
+            success_steps,
+            failure_steps,
+            realm,
+            can_gc,
+        );
+    }
+
+    // Return promise.
+    promise
+}

components/script/dom/readablestream.rs

@@ -14,7 +14,7 @@ use dom_struct::dom_struct;
 use ipc_channel::ipc::IpcSharedMemory;
 use js::conversions::ToJSValConvertible;
 use js::jsapi::{Heap, JSObject};
-use js::jsval::{JSVal, NullValue, ObjectValue, UndefinedValue};
+use js::jsval::{JSVal, ObjectValue, UndefinedValue};
 use js::rust::{
     HandleObject as SafeHandleObject, HandleValue as SafeHandleValue,
     MutableHandleValue as SafeMutableHandleValue,
@@ -29,8 +29,9 @@ use crate::dom::bindings::codegen::Bindings::ReadableStreamBinding::{
 use script_bindings::str::DOMString;
 
 use crate::dom::domexception::{DOMErrorName, DOMException};
-use script_bindings::conversions::StringificationBehavior;
+use script_bindings::conversions::{is_array_like, StringificationBehavior};
 use super::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategySize;
+use crate::dom::abortsignal::{AbortAlgorithm, AbortSignal};
 use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::ReadableStreamDefaultReaderMethods;
 use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultControllerBinding::ReadableStreamDefaultController_Binding::ReadableStreamDefaultControllerMethods;
 use crate::dom::bindings::codegen::Bindings::UnderlyingSourceBinding::UnderlyingSource as JsUnderlyingSource;
@@ -46,7 +47,7 @@ use crate::dom::bindings::utils::get_dictionary_property;
 use crate::dom::countqueuingstrategy::{extract_high_water_mark, extract_size_algorithm};
 use crate::dom::readablestreamgenericreader::ReadableStreamGenericReader;
 use crate::dom::globalscope::GlobalScope;
-use crate::dom::promise::Promise;
+use crate::dom::promise::{wait_for_all_promise, Promise};
 use crate::dom::readablebytestreamcontroller::ReadableByteStreamController;
 use crate::dom::readablestreambyobreader::ReadableStreamBYOBReader;
 use crate::dom::readablestreamdefaultcontroller::ReadableStreamDefaultController;
@@ -99,6 +100,8 @@ enum ShutdownAction {
     ReadableStreamCancel,
     /// <https://streams.spec.whatwg.org/#writable-stream-default-writer-close-with-error-propagation>
     WritableStreamDefaultWriterCloseWithErrorPropagation,
+    /// <https://streams.spec.whatwg.org/#ref-for-rs-pipeTo-shutdown-with-action>
+    Abort,
 }
 
 impl js::gc::Rootable for PipeTo {}
@@ -113,7 +116,7 @@ impl js::gc::Rootable for PipeTo {}
 /// - Error and close states must be propagated: we'll do this by checking these states at every step.
 #[derive(Clone, JSTraceable, MallocSizeOf)]
 #[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
-struct PipeTo {
+pub(crate) struct PipeTo {
     /// <https://streams.spec.whatwg.org/#ref-for-readablestream%E2%91%A7%E2%91%A0>
     reader: Dom<ReadableStreamDefaultReader>,
 
@@ -144,10 +147,15 @@ struct PipeTo {
     #[ignore_malloc_size_of = "Rc are hard"]
     shutting_down: Rc<Cell<bool>>,
 
+    /// The abort reason of the abort signal,
+    /// stored here because we must keep it across a microtask.
+    #[ignore_malloc_size_of = "mozjs"]
+    abort_reason: Rc<Heap<JSVal>>,
+
     /// The error potentially passed to shutdown,
     /// stored here because we must keep it across a microtask.
     #[ignore_malloc_size_of = "mozjs"]
-    shutdown_error: Rc<Heap<JSVal>>,
+    shutdown_error: Rc<RefCell<Option<Heap<JSVal>>>>,
 
     /// The promise returned by a shutdown action.
     /// We keep it to only continue when it is not pending anymore.
@@ -160,6 +168,49 @@ struct PipeTo {
     result_promise: Rc<Promise>,
 }
 
+impl PipeTo {
+    /// Run the `abortAlgorithm` defined at
+    /// <https://streams.spec.whatwg.org/#readable-stream-pipe-to>
+    pub(crate) fn abort_with_reason(
+        &self,
+        cx: SafeJSContext,
+        global: &GlobalScope,
+        reason: SafeHandleValue,
+        realm: InRealm,
+        can_gc: CanGc,
+    ) {
+        // Abort should do nothing if we are already shutting down.
+        if self.shutting_down.get() {
+            return;
+        }
+
+        // Let error be signal’s abort reason.
+        // Note: storing it because it may need to be kept across a microtask,
+        // and see the note below as to why it is kept separately from `shutdown_error`.
+        self.abort_reason.set(reason.get());
+
+        // Note: setting the error now,
+        // will result in a rejection of the pipe promise, with this error.
+        // Unless any shutdown action raise their own error,
+        // in which case this error will be overwritten by the shutdown action error.
+        {
+            let mut error = Some(Heap::default());
+            // Setting the value on the heap after it has been moved.
+            if let Some(heap) = error.as_mut() {
+                heap.set(reason.get())
+            }
+            *self.shutdown_error.borrow_mut() = error;
+        }
+
+        // Let actions be an empty ordered set.
+        // Note: the actions are defined, and performed, inside `shutdown_with_an_action`.
+
+        // Shutdown with an action consisting of getting a promise to wait for all of the actions in actions,
+        // and with error.
+        self.shutdown(cx, global, Some(ShutdownAction::Abort), realm, can_gc);
+    }
+}
+
 impl Callback for PipeTo {
     /// The pipe makes progress one microtask at a time.
     /// Note: we use one struct as the callback for all promises,
@@ -265,6 +316,11 @@ impl Callback for PipeTo {
                 // Write the chunk.
                 self.write_chunk(cx, &global, result, can_gc);
 
+                // An early return is necessary if the write algorithm aborted the pipe.
+                if self.shutting_down.get() {
+                    return;
+                }
+
                 // Wait for the writer to be ready again.
                 self.wait_for_writer_ready(&global, realm, can_gc);
             },
@@ -296,12 +352,33 @@ impl Callback for PipeTo {
                     return;
                 }
 
+                let is_array_like = {
+                    if !result.is_object() {
+                        false
+                    } else {
+                        unsafe { is_array_like::<crate::DomTypeHolder>(*cx, result) }
+                    }
+                };
+
                 // Finalize, passing along error if it was given.
-                if !result.is_undefined() {
-                    // All actions either resolve with undefined,
+                if !result.is_undefined() && !is_array_like {
+                    // Most actions either resolve with undefined,
                     // or reject with an error,
                     // and the error should be used when finalizing.
-                    self.shutdown_error.set(result.get());
+                    // One exception is the `Abort` action,
+                    // which resolves with a list of undefined values.
+
+                    // If `result` isn't undefined or array-like,
+                    // then it is an error
+                    // and should overwrite the current shutdown error.
+                    {
+                        let mut error = Some(Heap::default());
+                        // Setting the value on the heap after it has been moved.
+                        if let Some(heap) = error.as_mut() {
+                            heap.set(result.get())
+                        }
+                        *self.shutdown_error.borrow_mut() = error;
+                    }
                 }
                 self.finalize(cx, &global, can_gc);
             },
@@ -426,7 +503,14 @@ impl PipeTo {
         if source.is_errored() {
             rooted!(in(*cx) let mut source_error = UndefinedValue());
             source.get_stored_error(source_error.handle_mut());
-            self.shutdown_error.set(source_error.get());
+            {
+                let mut error = Some(Heap::default());
+                // Setting the value on the heap after it has been moved.
+                if let Some(heap) = error.as_mut() {
+                    heap.set(source_error.get())
+                }
+                *self.shutdown_error.borrow_mut() = error;
+            }
 
             // If preventAbort is false,
             if !self.prevent_abort {
@@ -469,7 +553,14 @@ impl PipeTo {
         if dest.is_errored() {
             rooted!(in(*cx) let mut dest_error = UndefinedValue());
             dest.get_stored_error(dest_error.handle_mut());
-            self.shutdown_error.set(dest_error.get());
+            {
+                let mut error = Some(Heap::default());
+                // Setting the value on the heap after it has been moved.
+                if let Some(heap) = error.as_mut() {
+                    heap.set(dest_error.get())
+                }
+                *self.shutdown_error.borrow_mut() = error;
+            }
 
             // If preventCancel is false,
             if !self.prevent_cancel {
@@ -558,7 +649,14 @@ impl PipeTo {
             let error =
                 Error::Type("Destination is closed or has closed queued or in flight".to_string());
             error.to_jsval(cx, global, dest_closed.handle_mut(), can_gc);
-            self.shutdown_error.set(dest_closed.get());
+            {
+                let mut error = Some(Heap::default());
+                // Setting the value on the heap after it has been moved.
+                if let Some(heap) = error.as_mut() {
+                    heap.set(dest_closed.get())
+                }
+                *self.shutdown_error.borrow_mut() = error;
+            }
 
             // If preventCancel is false,
             if !self.prevent_cancel {
@@ -629,7 +727,11 @@ impl PipeTo {
         realm: InRealm,
         can_gc: CanGc,
     ) {
-        rooted!(in(*cx) let mut error = self.shutdown_error.get());
+        rooted!(in(*cx) let mut error = UndefinedValue());
+        if let Some(shutdown_error) = self.shutdown_error.borrow().as_ref() {
+            error.set(shutdown_error.get());
+        }
+
         *self.state.borrow_mut() = PipeToState::ShuttingDownPendingAction;
 
         // Let p be the result of performing action.
@@ -648,6 +750,55 @@ impl PipeTo {
             ShutdownAction::WritableStreamDefaultWriterCloseWithErrorPropagation => {
                 self.writer.close_with_error_propagation(cx, global, can_gc)
             },
+            ShutdownAction::Abort => {
+                // Note: implementation of the the `abortAlgorithm`
+                // of the signal associated with this piping operation.
+
+                // Let error be signal’s abort reason.
+                rooted!(in(*cx) let mut error = UndefinedValue());
+                error.set(self.abort_reason.get());
+
+                // Let actions be an empty ordered set.
+                let mut actions = vec![];
+
+                // If preventAbort is false, append the following action to actions:
+                if !self.prevent_abort {
+                    let dest = self
+                        .writer
+                        .get_stream()
+                        .expect("Destination stream must be set");
+
+                    // If dest.[[state]] is "writable",
+                    let promise = if dest.is_writable() {
+                        // return ! WritableStreamAbort(dest, error)
+                        dest.abort(cx, global, error.handle(), can_gc)
+                    } else {
+                        // Otherwise, return a promise resolved with undefined.
+                        Promise::new_resolved(global, cx, (), can_gc)
+                    };
+                    actions.push(promise);
+                }
+
+                // If preventCancel is false, append the following action action to actions:
+                if !self.prevent_cancel {
+                    let source = self.reader.get_stream().expect("Source stream must be set");
+
+                    // If source.[[state]] is "readable",
+                    let promise = if source.is_readable() {
+                        // return ! ReadableStreamCancel(source, error).
+                        source.cancel(cx, global, error.handle(), can_gc)
+                    } else {
+                        // Otherwise, return a promise resolved with undefined.
+                        Promise::new_resolved(global, cx, (), can_gc)
+                    };
+                    actions.push(promise);
+                }
+
+                // Shutdown with an action consisting
+                // of getting a promise to wait for all of the actions in actions,
+                // and with error.
+                wait_for_all_promise(cx, global, actions, realm, can_gc)
+            },
         };
 
         // Upon fulfillment of p, finalize, passing along originalError if it was given.
@@ -679,10 +830,13 @@ impl PipeTo {
             .expect("Releasing the reader should not fail");
 
         // If signal is not undefined, remove abortAlgorithm from signal.
-        // TODO: implement AbortSignal.
+        // Note: since `self.shutdown` is true at this point,
+        // the abort algorithm is a no-op,
+        // so for now not implementing this step.
 
-        rooted!(in(*cx) let mut error = self.shutdown_error.get());
-        if !error.is_null() {
+        if let Some(shutdown_error) = self.shutdown_error.borrow().as_ref() {
+            rooted!(in(*cx) let mut error = UndefinedValue());
+            error.set(shutdown_error.get());
             // If error was given, reject promise with error.
             self.result_promise.reject_native(&error.handle(), can_gc);
         } else {
@@ -1636,9 +1790,10 @@ impl ReadableStream {
         cx: SafeJSContext,
         global: &GlobalScope,
         dest: &WritableStream,
+        prevent_close: bool,
         prevent_abort: bool,
         prevent_cancel: bool,
-        prevent_close: bool,
+        signal: Option<&AbortSignal>,
         realm: InRealm,
         can_gc: CanGc,
     ) -> Rc<Promise> {
@@ -1649,7 +1804,7 @@ impl ReadableStream {
 
         // If signal was not given, let signal be undefined.
         // Assert: either signal is undefined, or signal implements AbortSignal.
-        // TODO: implement AbortSignal.
+        // Note: done with the `signal` argument.
 
         // Assert: ! IsReadableStreamLocked(source) is false.
         assert!(!self.is_locked());
@@ -1682,9 +1837,6 @@ impl ReadableStream {
         // Let promise be a new promise.
         let promise = Promise::new(global, can_gc);
 
-        // If signal is not undefined,
-        // TODO: implement AbortSignal.
-
         // In parallel, but not really, using reader and writer, read all chunks from source and write them to dest.
         rooted!(in(*cx) let pipe_to = PipeTo {
             reader: Dom::from_ref(&reader),
@@ -1695,15 +1847,28 @@ impl ReadableStream {
             prevent_cancel,
             prevent_close,
             shutting_down: Default::default(),
+            abort_reason: Default::default(),
             shutdown_error: Default::default(),
             shutdown_action_promise:  Default::default(),
             result_promise: promise.clone(),
         });
 
-        // Note: set the shutdown error to null,
-        // to distinguish it from cases
-        // where the error is set to undefined.
-        pipe_to.shutdown_error.set(NullValue());
+        // If signal is not undefined,
+        // Note: moving the steps to here, so that the `PipeTo` is available.
+        if let Some(signal) = signal {
+            // Let abortAlgorithm be the following steps:
+            // Note: steps are implemented at call site.
+            rooted!(in(*cx) let abort_algorithm = AbortAlgorithm::StreamPiping(pipe_to.clone()));
+
+            // If signal is aborted, perform abortAlgorithm and return promise.
+            if signal.aborted() {
+                signal.run_abort_algorithm(cx, global, &abort_algorithm, realm, can_gc);
+                return promise;
+            }
+
+            // Add abortAlgorithm to signal.
+            signal.add(&abort_algorithm);
+        }
 
         // Note: perfom checks now, since streams can start as closed or errored.
         pipe_to.check_and_propagate_errors_forward(cx, global, realm, can_gc);
@@ -1992,16 +2157,17 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
         }
 
         // Let signal be options["signal"] if it exists, or undefined otherwise.
-        // TODO: implement AbortSignal.
+        let signal = options.signal.as_deref();
 
         // Return ! ReadableStreamPipeTo.
         self.pipe_to(
             cx,
             &global,
             destination,
+            options.preventClose,
             options.preventAbort,
             options.preventCancel,
-            options.preventClose,
+            signal,
             realm,
             can_gc,
         )
@@ -2029,7 +2195,7 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
         }
 
         // Let signal be options["signal"] if it exists, or undefined otherwise.
-        // TODO: implement AbortSignal.
+        let signal = options.signal.as_deref();
 
         // Let promise be ! ReadableStreamPipeTo(this, transform["writable"],
         // options["preventClose"], options["preventAbort"], options["preventCancel"], signal).
@@ -2037,9 +2203,10 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
             cx,
             &global,
             &transform.writable,
+            options.preventClose,
             options.preventAbort,
             options.preventCancel,
-            options.preventClose,
+            signal,
             realm,
             can_gc,
         );
@@ -2270,7 +2437,9 @@ impl Transferable for ReadableStream {
         writable.setup_cross_realm_transform_writable(cx, &port_1, can_gc);
 
         // Let promise be ! ReadableStreamPipeTo(value, writable, false, false, false).
-        let promise = self.pipe_to(cx, &global, &writable, false, false, false, comp, can_gc);
+        let promise = self.pipe_to(
+            cx, &global, &writable, false, false, false, None, comp, can_gc,
+        );
 
         // Set promise.[[PromiseIsHandled]] to true.
         promise.set_promise_is_handled();

components/script/dom/transformstream.rs

@@ -1053,7 +1053,9 @@ impl Transferable for TransformStream {
         let proxy_readable = ReadableStream::new_with_proto(&global, None, can_gc);
         proxy_readable.setup_cross_realm_transform_readable(cx, &port1, can_gc);
         proxy_readable
-            .pipe_to(cx, &global, &writable, false, false, false, comp, can_gc)
+            .pipe_to(
+                cx, &global, &writable, false, false, false, None, comp, can_gc,
+            )
             .set_promise_is_handled();
 
         // Second port pair (proxy readable → writable)
@@ -1075,6 +1077,7 @@ impl Transferable for TransformStream {
                 false,
                 false,
                 false,
+                None,
                 comp,
                 can_gc,
             )

components/script/dom/writablestream.rs

@@ -1241,7 +1241,7 @@ impl Transferable for WritableStream {
         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);
+        let promise = readable.pipe_to(cx, &global, self, false, false, false, None, comp, can_gc);
 
         // Set promise.[[PromiseIsHandled]] to true.
         promise.set_promise_is_handled();