From 8284cf02c6abafb61e40acdeb2e015cb163ff461 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Fri, 18 Sep 2026 06:38:16 +0800 Subject: [PATCH] feat(dom): implement Observable subscription lifecycle --- .../wpt-cross-current/failed-cases.txt | 1 - .../wpt-cross-current/passed-cases.txt | 3 + moli-renderer-v8/src/abort_signal_route.rs | 30 ++ .../assets/constructor_templates.rs | 8 + .../assets/prototype_bindings.rs | 1 + .../exposed_interfaces/metadata.rs | 2 + .../src/context_bootstrap/runtime_state.rs | 2 + .../src/context_bootstrap/specs/registry.rs | 8 + .../src/context_bootstrap/specs/types.rs | 1 + moli-renderer-v8/src/lib.rs | 1 + moli-renderer-v8/src/native_bridge/abort.rs | 2 +- moli-renderer-v8/src/observable.rs | 423 ++++++++++++++++++ moli-renderer-v8/src/observable/callbacks.rs | 113 +++++ moli-renderer-v8/src/observable/state.rs | 204 +++++++++ moli-renderer-v8/src/script_vm/tests/mod.rs | 1 + .../src/script_vm/tests/observable.rs | 116 +++++ moli-renderer-v8/src/web_api_interfaces.rs | 2 + moli-renderer-v8/src/worker/abort.rs | 7 + .../src/worker/thread/tests/postmessage.rs | 16 + .../tests/fixtures/observable-core.js | 146 ++++++ 20 files changed, 1085 insertions(+), 2 deletions(-) create mode 100644 moli-renderer-v8/src/observable.rs create mode 100644 moli-renderer-v8/src/observable/callbacks.rs create mode 100644 moli-renderer-v8/src/observable/state.rs create mode 100644 moli-renderer-v8/src/script_vm/tests/observable.rs create mode 100644 moli-renderer-v8/tests/fixtures/observable-core.js diff --git a/moli-benchmark/wpt-cross-current/failed-cases.txt b/moli-benchmark/wpt-cross-current/failed-cases.txt index c2014dc28..01558b0c1 100644 --- a/moli-benchmark/wpt-cross-current/failed-cases.txt +++ b/moli-benchmark/wpt-cross-current/failed-cases.txt @@ -2490,7 +2490,6 @@ dom/nodes/moveBefore/moveBefore-from-light-to-shadow.html dom/nodes/moveBefore/moveBefore-size-query.html dom/nodes/moveBefore/preserve-render-blocking-script.html dom/nodes/moveBefore/preserve-render-blocking-style.html -dom/observable/tentative/idlharness.html dom/ranges/tentative/OpaqueRange-auto-disconnect.html dom/ranges/tentative/OpaqueRange-basic.html dom/ranges/tentative/OpaqueRange-disconnect.html diff --git a/moli-benchmark/wpt-cross-current/passed-cases.txt b/moli-benchmark/wpt-cross-current/passed-cases.txt index f982ddc62..63483507d 100644 --- a/moli-benchmark/wpt-cross-current/passed-cases.txt +++ b/moli-benchmark/wpt-cross-current/passed-cases.txt @@ -4532,6 +4532,9 @@ dom/nodes/remove-unscopable.html dom/nodes/replaceWith-document-element-crash.html dom/nodes/rootNode.html dom/nodes/svg-template-querySelector.html +dom/observable/tentative/idlharness.html +dom/observable/tentative/observable-constructor.any.js?moli-wpt-any=dedicatedworker +dom/observable/tentative/observable-constructor.any.js?moli-wpt-any=window dom/ranges/Range-adopt-test.html dom/ranges/Range-attributes.html dom/ranges/Range-cloneContents.html diff --git a/moli-renderer-v8/src/abort_signal_route.rs b/moli-renderer-v8/src/abort_signal_route.rs index 64976c071..1535b50bf 100644 --- a/moli-renderer-v8/src/abort_signal_route.rs +++ b/moli-renderer-v8/src/abort_signal_route.rs @@ -28,6 +28,36 @@ pub(crate) struct ResolvedAbortSignal<'s> { } impl<'s> ResolvedAbortSignal<'s> { + /// Creates a signal in the current realm without consulting author-visible + /// constructors or maintaining another store for native algorithms. + pub(crate) fn new(scope: &mut v8::PinScope<'s, '_>) -> Option { + let signal = if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) { + crate::native_bridge::abort::create_signal( + scope, + unsafe { &mut *host_ptr }, + false, + None, + )? + } else { + crate::worker::abort::new_worker_abort_signal(scope)? + }; + Self::resolve(scope, signal) + } + + pub(crate) fn reason(self, scope: &mut v8::PinScope<'s, '_>) -> v8::Local<'s, v8::Value> { + match self.owner { + AbortSignalOwner::Window => { + context_host_ptr_from_global_bridge(scope).and_then(|host_ptr| { + unsafe { &mut *host_ptr }.abort_signal_reason(scope, self.signal) + }) + } + AbortSignalOwner::Worker => { + crate::worker::abort::worker_abort_signal_reason(scope, self.signal) + } + } + .unwrap_or_else(|| v8::undefined(scope).into()) + } + pub(crate) fn resolve( scope: &mut v8::PinScope<'s, '_>, signal: v8::Local<'s, v8::Object>, diff --git a/moli-renderer-v8/src/context_bootstrap/assets/constructor_templates.rs b/moli-renderer-v8/src/context_bootstrap/assets/constructor_templates.rs index 3e1d5aaea..7c32c8b79 100644 --- a/moli-renderer-v8/src/context_bootstrap/assets/constructor_templates.rs +++ b/moli-renderer-v8/src/context_bootstrap/assets/constructor_templates.rs @@ -296,6 +296,14 @@ pub(in crate::context_bootstrap) fn build_constructor_template_for_profile<'s>( .length(1) .build(scope) } + ConstructorKind::Observable => { + v8::FunctionTemplate::builder(moli_webapi_declare::web_api_constructor!( + web_api_interfaces::Observable, + crate::observable::constructor + )) + .length(1) + .build(scope) + } ConstructorKind::DomParser => { v8::FunctionTemplate::builder(moli_webapi_declare::web_api_constructor!( web_api_interfaces::DOMParser, diff --git a/moli-renderer-v8/src/context_bootstrap/assets/prototype_bindings.rs b/moli-renderer-v8/src/context_bootstrap/assets/prototype_bindings.rs index c8a57098a..50ff4ba99 100644 --- a/moli-renderer-v8/src/context_bootstrap/assets/prototype_bindings.rs +++ b/moli-renderer-v8/src/context_bootstrap/assets/prototype_bindings.rs @@ -412,6 +412,7 @@ pub(super) fn install_constructor_template_bindings<'s>( install_constructor_constant_template_bindings(scope, template, spec.interface.name()); install_css_style_declaration_template_accessors(scope, template, spec.interface.name()); install_abort_template_bindings(scope, template, spec.interface.name()); + crate::observable::install_template_bindings(scope, template, spec.interface.name()); install_attr_template_bindings(scope, template, spec.interface.name()); install_dom_quad_template_bindings(scope, template, spec.interface.name()); install_dom_rect_template_bindings(scope, template, spec.interface.name()); diff --git a/moli-renderer-v8/src/context_bootstrap/exposed_interfaces/metadata.rs b/moli-renderer-v8/src/context_bootstrap/exposed_interfaces/metadata.rs index 8442cff54..d3af637fb 100644 --- a/moli-renderer-v8/src/context_bootstrap/exposed_interfaces/metadata.rs +++ b/moli-renderer-v8/src/context_bootstrap/exposed_interfaces/metadata.rs @@ -31,6 +31,8 @@ pub(in crate::context_bootstrap) const WORKER_SHARED_INTERFACE_NAMES: &[&str] = "WebSocket", "AbortSignal", "AbortController", + "Observable", + "Subscriber", "Headers", "Request", "Response", diff --git a/moli-renderer-v8/src/context_bootstrap/runtime_state.rs b/moli-renderer-v8/src/context_bootstrap/runtime_state.rs index 01bee0046..adb7d4968 100644 --- a/moli-renderer-v8/src/context_bootstrap/runtime_state.rs +++ b/moli-renderer-v8/src/context_bootstrap/runtime_state.rs @@ -1895,6 +1895,8 @@ pub(crate) fn finish_context_bootstrap( ("AudioBuffer", "AudioBuffer"), ("AbortSignal", "AbortSignal"), ("AbortController", "AbortController"), + ("Observable", "Observable"), + ("Subscriber", "Subscriber"), ("TextEncoder", "TextEncoder"), ("TextDecoder", "TextDecoder"), ("ReadableStream", "ReadableStream"), diff --git a/moli-renderer-v8/src/context_bootstrap/specs/registry.rs b/moli-renderer-v8/src/context_bootstrap/specs/registry.rs index ce3dd684c..330e4d2f1 100644 --- a/moli-renderer-v8/src/context_bootstrap/specs/registry.rs +++ b/moli-renderer-v8/src/context_bootstrap/specs/registry.rs @@ -388,6 +388,14 @@ const CONSTRUCTOR_SPECS_BEFORE_STREAMS: &[ConstructorSpec] = &[ interface: web_api_interfaces::AbortController::DESCRIPTOR, kind: ConstructorKind::AbortController, }, + ConstructorSpec { + interface: web_api_interfaces::Observable::DESCRIPTOR, + kind: ConstructorKind::Observable, + }, + ConstructorSpec { + interface: web_api_interfaces::Subscriber::DESCRIPTOR, + kind: ConstructorKind::Illegal, + }, ConstructorSpec { interface: web_api_interfaces::BroadcastChannel::DESCRIPTOR, kind: ConstructorKind::BroadcastChannel, diff --git a/moli-renderer-v8/src/context_bootstrap/specs/types.rs b/moli-renderer-v8/src/context_bootstrap/specs/types.rs index 83c3c996f..bb0c7235e 100644 --- a/moli-renderer-v8/src/context_bootstrap/specs/types.rs +++ b/moli-renderer-v8/src/context_bootstrap/specs/types.rs @@ -77,6 +77,7 @@ pub(in crate::context_bootstrap) enum ConstructorKind { FileReaderSync, XmlSerializer, AbortController, + Observable, BroadcastChannel, EventSource, IdleDetector, diff --git a/moli-renderer-v8/src/lib.rs b/moli-renderer-v8/src/lib.rs index 377a010fc..be6939a81 100644 --- a/moli-renderer-v8/src/lib.rs +++ b/moli-renderer-v8/src/lib.rs @@ -74,6 +74,7 @@ mod mutation_coordinator; pub(crate) mod native_bridge; pub mod network; mod network_host; +mod observable; mod observer_runtime; mod opfs_owner_tasks; mod opfs_task_result; diff --git a/moli-renderer-v8/src/native_bridge/abort.rs b/moli-renderer-v8/src/native_bridge/abort.rs index fb42ca771..2fc32af42 100644 --- a/moli-renderer-v8/src/native_bridge/abort.rs +++ b/moli-renderer-v8/src/native_bridge/abort.rs @@ -377,7 +377,7 @@ pub(super) fn timeout_error_value<'s>( dom_exception_value(scope, "signal timed out", "TimeoutError") } -fn create_signal<'s>( +pub(crate) fn create_signal<'s>( scope: &mut v8::PinScope<'s, '_>, host: &mut super::JsContextHost, aborted: bool, diff --git a/moli-renderer-v8/src/observable.rs b/moli-renderer-v8/src/observable.rs new file mode 100644 index 000000000..0c1c909ec --- /dev/null +++ b/moli-renderer-v8/src/observable.rs @@ -0,0 +1,423 @@ +//! Observable subscriptions share one producer until their last observer leaves. +//! Callback residence is V8-traced; the Observable's link to its Subscriber is +//! weak. AbortSignal state and callback invocation remain with their existing +//! Window/worker owners. + +mod callbacks; +mod state; + +use moli_webapi_declare::WebApiFunctionTemplate; + +use crate::{ + abort_signal_route::ResolvedAbortSignal, util::set_private_value, web_api_interfaces, webidl, +}; +use callbacks::{invoke, is_current, report}; +use state::*; + +#[derive(WebApiFunctionTemplate)] +#[webapi(interface = web_api_interfaces::Observable, enumerable, receiver)] +struct ObservablePrototype { + #[webapi(method, length = 0, callback = subscribe)] + subscribe: (), +} + +#[derive(WebApiFunctionTemplate)] +#[webapi(interface = web_api_interfaces::Subscriber, enumerable, receiver)] +struct SubscriberPrototype { + #[webapi(method, length = 1, callback = next)] + next: (), + #[webapi(method, length = 1, callback = error)] + error: (), + #[webapi(method, length = 0, callback = complete)] + complete: (), + #[webapi(method, length = 1, callback = add_teardown)] + add_teardown: (), + #[webapi(accessor_property, getter = active_getter)] + active: (), + #[webapi(accessor_property, getter = signal_getter)] + signal: (), +} + +pub(crate) fn install_template_bindings<'s>( + scope: &mut v8::PinScope<'s, '_, ()>, + template: v8::Local<'s, v8::FunctionTemplate>, + name: &str, +) { + let prototype = template.prototype_template(scope); + match name { + "Observable" => ObservablePrototype::initialize_prototype_template(scope, prototype), + "Subscriber" => SubscriberPrototype::initialize_prototype_template(scope, prototype), + _ => {} + } +} + +#[derive(webidl::WebIdlArgs)] +#[webidl(prefix = "Observable")] +struct ConstructorArgs { + #[webidl(required, converter = "callback_function")] + callback: webidl::WebIdlCallbackFunction, +} + +#[derive(webidl::WebIdlArgs)] +#[webidl(prefix = "Observable.subscribe")] +struct SubscribeArgs<'scope> { + #[webidl(with = observer_arg)] + observer: v8::Local<'scope, v8::Object>, + #[webidl(with = signal_arg)] + signal: Option>, +} + +#[derive(webidl::WebIdlArgs)] +#[webidl(prefix = "Subscriber")] +struct ValueArgs<'scope> { + #[webidl(required, converter = "raw")] + value: v8::Local<'scope, v8::Value>, +} + +#[derive(webidl::WebIdlArgs)] +#[webidl(prefix = "Subscriber.addTeardown")] +struct TeardownArgs { + #[webidl(required, converter = "callback_function")] + callback: webidl::WebIdlCallbackFunction, +} + +fn dictionary<'s>( + value: v8::Local<'s, v8::Value>, + message: &'static str, +) -> Result>, webidl::WebIdlError> { + if value.is_null_or_undefined() { + return Ok(None); + } + v8::Local::::try_from(value) + .map(Some) + .map_err(|_| webidl::WebIdlError::custom_message(message)) +} + +fn observer_arg<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: &v8::FunctionCallbackArguments<'s>, + index: i32, +) -> Result, webidl::WebIdlError> { + let observer = v8::Object::new(scope); + let Some(input) = dictionary(args.get(index), "SubscriptionObserver must be an object")? else { + return Ok(observer); + }; + if input.is_callable() { + let callback = webidl::convert::( + scope, + input.into(), + webidl::Context::argument("Observable.subscribe", 1), + )?; + set_callback(scope, observer, NEXT, callback); + return Ok(observer); + } + // Web IDL dictionary member conversion is lexicographic, even though the + // subscription itself delivers next/error/complete notifications. + for (name, slot) in [("complete", COMPLETE), ("error", ERROR), ("next", NEXT)] { + let context = webidl::Context::member("SubscriptionObserver", name); + if let Some(value) = webidl::property_result(scope, input, name, context)? + && !value.is_undefined() + { + let callback = + webidl::convert::(scope, value, context)?; + set_callback(scope, observer, slot, callback); + } + } + Ok(observer) +} + +fn signal_arg<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: &v8::FunctionCallbackArguments<'s>, + index: i32, +) -> Result>, webidl::WebIdlError> { + let Some(options) = dictionary(args.get(index), "SubscribeOptions must be an object")? else { + return Ok(None); + }; + let context = webidl::Context::member("SubscribeOptions", "signal"); + let Some(value) = webidl::property_result(scope, options, "signal", context)? else { + return Ok(None); + }; + if value.is_undefined() { + return Ok(None); + } + v8::Local::::try_from(value) + .ok() + .filter(|signal| web_api_interfaces::AbortSignal::is_instance(scope, *signal)) + .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal)) + .map(Some) + .ok_or_else(|| { + webidl::WebIdlError::custom_message("SubscribeOptions.signal must be an AbortSignal") + }) +} + +pub(crate) fn constructor<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + mut rv: v8::ReturnValue<'_, v8::Value>, +) { + if !args.is_construct_call() { + webidl::throw_type_error(scope, "Observable must be constructed with new"); + return; + } + let Some(parsed) = webidl::parse_args::(scope, &args) else { + return; + }; + initialize_observable(scope, args.this(), parsed.callback); + rv.set(args.this().into()); +} + +fn subscribe<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let Some(parsed) = webidl::parse_args::>(scope, &args) else { + return; + }; + let observable = args.this(); + if !is_current(scope, observable) { + return; + } + let (subscriber, fresh) = match current_subscriber(scope, observable) + .filter(|subscriber| active(scope, *subscriber)) + { + Some(subscriber) => (subscriber, false), + None => { + let Some(subscriber) = new_subscriber(scope) else { + return; + }; + set_subscriber(scope, observable, subscriber); + (subscriber, true) + } + }; + let mut observers = list(scope, subscriber, OBSERVERS); + observers.push(parsed.observer); + set_list(scope, subscriber, OBSERVERS, &observers); + if let Some(signal) = parsed.signal { + if signal.is_aborted(scope) { + if fresh { + let reason = signal.reason(scope); + close(scope, subscriber, Some(reason)); + } else { + observers.pop(); + set_list(scope, subscriber, OBSERVERS, &observers); + } + } else { + let data = + v8::Array::new_with_elements(scope, &[subscriber.into(), parsed.observer.into()]); + let algorithm = v8::Function::builder(cancel_observer) + .data(data.into()) + .build(scope) + .expect("Observable abort algorithm should allocate"); + set_private_value(scope, parsed.observer, INPUT_SIGNAL, signal.value().into()); + set_private_value(scope, parsed.observer, ABORT_ALGORITHM, algorithm.into()); + signal.register_algorithm(scope, algorithm); + } + } + if fresh + && let Some(callback) = object_slot(scope, observable, INITIALIZER) + && let Some(exception) = invoke(scope, callback, &[subscriber.into()]) + { + subscriber_error(scope, subscriber, exception); + } +} + +fn cancel_observer<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let data = v8::Local::::try_from(args.data()).expect("Observable abort data"); + let subscriber = v8::Local::::try_from(data.get_index(scope, 0).unwrap()).unwrap(); + let observer = v8::Local::::try_from(data.get_index(scope, 1).unwrap()).unwrap(); + if !active(scope, subscriber) { + return; + } + let mut observers = list(scope, subscriber, OBSERVERS); + observers.retain(|entry| *entry != observer); + set_list(scope, subscriber, OBSERVERS, &observers); + release_abort_algorithm(scope, observer); + if observers.is_empty() { + close(scope, subscriber, Some(args.get(0))); + } +} + +fn close<'s>( + scope: &mut v8::PinScope<'s, '_>, + subscriber: v8::Local<'s, v8::Object>, + reason: Option>, +) { + if !active(scope, subscriber) { + return; + } + set_private_value( + scope, + subscriber, + ACTIVE, + v8::Boolean::new(scope, false).into(), + ); + for observer in list(scope, subscriber, OBSERVERS) { + release_abort_algorithm(scope, observer); + } + let signal = object_slot(scope, subscriber, SIGNAL) + .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal)); + let reason = reason + .filter(|reason| !reason.is_undefined()) + .unwrap_or_else(|| crate::native_bridge::abort::abort_error_value(scope)); + if let Some(signal) = signal { + signal.abort(scope, reason); + } + let teardowns = list(scope, subscriber, TEARDOWNS); + set_list(scope, subscriber, TEARDOWNS, &[]); + for callback in teardowns.into_iter().rev() { + if !is_current(scope, subscriber) { + break; + } + invoke_and_report(scope, callback, &[]); + } +} + +fn release_abort_algorithm<'s>( + scope: &mut v8::PinScope<'s, '_>, + observer: v8::Local<'s, v8::Object>, +) { + if let Some(signal) = object_slot(scope, observer, INPUT_SIGNAL) + .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal)) + && let Some(algorithm) = object_slot(scope, observer, ABORT_ALGORITHM) + .and_then(|value| v8::Local::::try_from(value).ok()) + { + signal.unregister_algorithm(scope, algorithm); + } + set_private_value(scope, observer, INPUT_SIGNAL, v8::undefined(scope).into()); + set_private_value( + scope, + observer, + ABORT_ALGORITHM, + v8::undefined(scope).into(), + ); +} + +fn invoke_and_report<'s>( + scope: &mut v8::PinScope<'s, '_>, + callback: v8::Local<'s, v8::Object>, + values: &[v8::Local<'s, v8::Value>], +) { + if let Some(error) = invoke(scope, callback, values) { + callbacks::report_callback_exception(scope, callback, error); + } +} + +fn next<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let Some(parsed) = webidl::parse_args::>(scope, &args) else { + return; + }; + let subscriber = args.this(); + if !active(scope, subscriber) || !is_current(scope, subscriber) { + return; + } + // Reentrant subscribe/cancel must not change this notification's snapshot. + for observer in list(scope, subscriber, OBSERVERS) { + if let Some(callback) = object_slot(scope, observer, NEXT) { + invoke_and_report(scope, callback, &[parsed.value]); + } + } +} + +fn subscriber_error<'s>( + scope: &mut v8::PinScope<'s, '_>, + subscriber: v8::Local<'s, v8::Object>, + error: v8::Local<'s, v8::Value>, +) { + if !active(scope, subscriber) { + report(scope, subscriber, error); + return; + } + if !is_current(scope, subscriber) { + return; + } + close(scope, subscriber, Some(error)); + let observers = list(scope, subscriber, OBSERVERS); + set_list(scope, subscriber, OBSERVERS, &[]); + for observer in observers { + if let Some(callback) = object_slot(scope, observer, ERROR) { + invoke_and_report(scope, callback, &[error]); + } else { + callbacks::report_default_error(scope, error); + } + } +} + +fn error<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let Some(parsed) = webidl::parse_args::>(scope, &args) else { + return; + }; + subscriber_error(scope, args.this(), parsed.value); +} + +fn complete<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let subscriber = args.this(); + if !active(scope, subscriber) || !is_current(scope, subscriber) { + return; + } + close(scope, subscriber, None); + let observers = list(scope, subscriber, OBSERVERS); + set_list(scope, subscriber, OBSERVERS, &[]); + for observer in observers { + if let Some(callback) = object_slot(scope, observer, COMPLETE) { + invoke_and_report(scope, callback, &[]); + } + } +} + +fn add_teardown<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + _rv: v8::ReturnValue<'_, v8::Value>, +) { + let Some(parsed) = webidl::parse_args::(scope, &args) else { + return; + }; + let subscriber = args.this(); + if !is_current(scope, subscriber) { + return; + } + let callback = callbacks::trace(scope, parsed.callback); + if active(scope, subscriber) { + let mut teardowns = list(scope, subscriber, TEARDOWNS); + teardowns.push(callback); + set_list(scope, subscriber, TEARDOWNS, &teardowns); + } else { + invoke_and_report(scope, callback, &[]); + } +} + +fn active_getter<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + mut rv: v8::ReturnValue<'_, v8::Value>, +) { + rv.set_bool(active(scope, args.this())); +} + +fn signal_getter<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, + mut rv: v8::ReturnValue<'_, v8::Value>, +) { + if let Some(signal) = object_slot(scope, args.this(), SIGNAL) { + rv.set(signal.into()); + } +} diff --git a/moli-renderer-v8/src/observable/callbacks.rs b/moli-renderer-v8/src/observable/callbacks.rs new file mode 100644 index 000000000..f7342adce --- /dev/null +++ b/moli-renderer-v8/src/observable/callbacks.rs @@ -0,0 +1,113 @@ +use crate::{ + callback_invocation::invoke_synchronous_webidl_callback_function, + exception_reporting::build_exception_report_without_stack, + util::context_host_ptr_from_global_bridge, + v8_traced_webidl_callback::V8TracedWebIdlCallbackFunction, webidl, +}; + +pub(super) fn trace<'s>( + scope: &mut v8::PinScope<'s, '_>, + callback: webidl::WebIdlCallbackFunction, +) -> v8::Local<'s, v8::Object> { + V8TracedWebIdlCallbackFunction::new(scope, callback).into_object() +} + +fn context_is_current<'s>( + scope: &mut v8::PinScope<'s, '_>, + context: v8::Local<'s, v8::Context>, +) -> bool { + if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) { + let host = unsafe { &*host_ptr }; + host.window_execution_context_identity_for_v8_context(scope, context) + .is_some_and(|identity| host.window_execution_context_identity_is_current(identity)) + } else { + crate::worker::get_worker_state(scope).is_some() + } +} + +pub(super) fn is_current<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, +) -> bool { + object + .get_creation_context(scope) + .is_some_and(|context| context_is_current(scope, context)) +} + +/// The initializer rethrows into Subscriber.error; observer and teardown +/// callbacks report errors in their own relevant realm. Neither path runs an +/// extra microtask checkpoint. +pub(super) fn invoke<'s>( + scope: &mut v8::PinScope<'s, '_>, + carrier: v8::Local<'s, v8::Object>, + arguments: &[v8::Local<'s, v8::Value>], +) -> Option> { + let callback = V8TracedWebIdlCallbackFunction::from_object(carrier).prepare(scope); + let context = callback.relevant_context(scope); + if !context_is_current(scope, context) { + return None; + } + v8::tc_scope!(let scope, scope); + let receiver = v8::undefined(scope).into(); + invoke_synchronous_webidl_callback_function(scope, &callback, receiver, arguments); + let exception = scope.exception(); + scope.reset(); + exception +} + +pub(super) fn report_callback_exception<'s>( + scope: &mut v8::PinScope<'s, '_>, + carrier: v8::Local<'s, v8::Object>, + exception: v8::Local<'s, v8::Value>, +) { + let callback = V8TracedWebIdlCallbackFunction::from_object(carrier).prepare(scope); + let context = callback.relevant_context(scope); + report_in_context(scope, context, exception); +} + +pub(super) fn report<'s>( + scope: &mut v8::PinScope<'s, '_>, + subscriber: v8::Local<'s, v8::Object>, + exception: v8::Local<'s, v8::Value>, +) { + if let Some(context) = subscriber.get_creation_context(scope) { + report_in_context(scope, context, exception); + } +} + +pub(super) fn report_default_error<'s>( + scope: &mut v8::PinScope<'s, '_>, + exception: v8::Local<'s, v8::Value>, +) { + // An internal observer's default error algorithm reports in the current + // realm. An error pushed after closure instead uses Subscriber's realm. + let context = scope.get_current_context(); + report_in_context(scope, context, exception); +} + +fn report_in_context<'s>( + scope: &mut v8::PinScope<'s, '_>, + context: v8::Local<'s, v8::Context>, + exception: v8::Local<'s, v8::Value>, +) { + if !context_is_current(scope, context) { + return; + } + let scope = &mut v8::ContextScope::new(scope, context); + let message = v8::Exception::create_message(scope, exception); + let report = build_exception_report_without_stack(scope, Some(exception), Some(message)); + if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) { + let identity = + unsafe { &*host_ptr }.window_execution_context_identity_for_v8_context(scope, context); + crate::host::report_event_callback_exception( + scope, + host_ptr, + "Observable", + identity, + None, + &report, + ); + } else { + crate::worker::dispatch_current_worker_callback_exception(scope, report); + } +} diff --git a/moli-renderer-v8/src/observable/state.rs b/moli-renderer-v8/src/observable/state.rs new file mode 100644 index 000000000..fd4524fcb --- /dev/null +++ b/moli-renderer-v8/src/observable/state.rs @@ -0,0 +1,204 @@ +use std::{cell::RefCell, collections::HashMap, rc::Rc}; + +use moli_webapi_declare::WebApiObject; + +use super::callbacks; +use crate::{ + abort_signal_route::ResolvedAbortSignal, + context_bootstrap::ensure_intrinsic_interface_prototype, + util::{get_private_value, set_private_value}, + web_api_interfaces, webidl, +}; + +const ID: &str = "__moliObservableId"; +pub(super) const INITIALIZER: &str = "__moliObservableInitializer"; +pub(super) const ACTIVE: &str = "__moliSubscriberActive"; +pub(super) const SIGNAL: &str = "__moliSubscriberSignal"; +pub(super) const OBSERVERS: &str = "__moliSubscriberObservers"; +pub(super) const TEARDOWNS: &str = "__moliSubscriberTeardowns"; +pub(super) const NEXT: &str = "__moliObserverNext"; +pub(super) const ERROR: &str = "__moliObserverError"; +pub(super) const COMPLETE: &str = "__moliObserverComplete"; +pub(super) const INPUT_SIGNAL: &str = "__moliObserverInputSignal"; +pub(super) const ABORT_ALGORITHM: &str = "__moliObserverAbortAlgorithm"; + +type Store = Rc>; + +#[derive(Default)] +struct WeakSubscribers { + next_id: u64, + entries: HashMap, +} + +struct Entry { + _observable: v8::Weak, + subscriber: Option>, +} + +pub(super) fn initialize_observable<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, + callback: webidl::WebIdlCallbackFunction, +) { + set_callback(scope, object, INITIALIZER, callback); + let store = if let Some(store) = scope.get_slot::() { + store.clone() + } else { + let store = Store::default(); + scope.set_slot(store.clone()); + store + }; + let id = { + let mut store = store.borrow_mut(); + store.next_id = store + .next_id + .checked_add(1) + .expect("Observable identity exhausted"); + store.next_id + }; + let value = v8::BigInt::new_from_u64(scope, id); + set_private_value(scope, object, ID, value.into()); + let weak_store = Rc::downgrade(&store); + let weak = v8::Weak::with_finalizer( + scope, + object, + Box::new(move |_| { + if let Some(store) = weak_store.upgrade() { + store.borrow_mut().entries.remove(&id); + } + }), + ); + store.borrow_mut().entries.insert( + id, + Entry { + _observable: weak, + subscriber: None, + }, + ); +} + +fn id<'s>(scope: &mut v8::PinScope<'s, '_>, object: v8::Local<'s, v8::Object>) -> Option { + get_private_value(scope, object, ID) + .and_then(|value| v8::Local::::try_from(value).ok()) + .map(|value| value.u64_value().0) +} + +pub(super) fn current_subscriber<'s>( + scope: &mut v8::PinScope<'s, '_>, + observable: v8::Local<'s, v8::Object>, +) -> Option> { + let id = id(scope, observable)?; + scope + .get_slot::()? + .borrow() + .entries + .get(&id)? + .subscriber + .as_ref()? + .to_local(scope) +} + +pub(super) fn set_subscriber<'s>( + scope: &mut v8::PinScope<'s, '_>, + observable: v8::Local<'s, v8::Object>, + subscriber: v8::Local<'s, v8::Object>, +) { + let id = id(scope, observable).expect("Observable identity"); + let weak = v8::Weak::new(scope, subscriber); + scope + .get_slot::() + .unwrap() + .borrow_mut() + .entries + .get_mut(&id) + .unwrap() + .subscriber = Some(weak); +} + +#[derive(WebApiObject)] +#[webapi(prototype = "Object", interface = web_api_interfaces::Subscriber)] +struct SubscriberInstance<'scope> { + #[webapi(prototype)] + prototype: v8::Local<'scope, v8::Object>, + #[webapi(slot = ACTIVE)] + active: bool, + #[webapi(slot = SIGNAL)] + signal: v8::Local<'scope, v8::Object>, + #[webapi(slot = OBSERVERS)] + observers: v8::Local<'scope, v8::Array>, + #[webapi(slot = TEARDOWNS)] + teardowns: v8::Local<'scope, v8::Array>, +} + +pub(super) fn new_subscriber<'s>( + scope: &mut v8::PinScope<'s, '_>, +) -> Option> { + let prototype = ensure_intrinsic_interface_prototype(scope, "Subscriber").ok()?; + let signal = ResolvedAbortSignal::new(scope)?.value(); + SubscriberInstance::new( + prototype, + true, + signal, + v8::Array::new(scope, 0), + v8::Array::new(scope, 0), + ) + .bind(scope) + .ok() +} + +pub(super) fn object_slot<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, + slot: &str, +) -> Option> { + get_private_value(scope, object, slot) + .and_then(|value| v8::Local::::try_from(value).ok()) +} + +pub(super) fn active<'s>( + scope: &mut v8::PinScope<'s, '_>, + subscriber: v8::Local<'s, v8::Object>, +) -> bool { + get_private_value(scope, subscriber, ACTIVE).is_some_and(|value| value.is_true()) +} + +pub(super) fn list<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, + slot: &str, +) -> Vec> { + let Some(array) = object_slot(scope, object, slot) + .and_then(|value| v8::Local::::try_from(value).ok()) + else { + return Vec::new(); + }; + (0..array.length()) + .filter_map(|i| { + array + .get_index(scope, i) + .and_then(|value| v8::Local::::try_from(value).ok()) + }) + .collect() +} + +pub(super) fn set_list<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, + slot: &str, + values: &[v8::Local<'s, v8::Object>], +) { + let values: Vec> = + values.iter().map(|value| (*value).into()).collect(); + let array = v8::Array::new_with_elements(scope, &values); + set_private_value(scope, object, slot, array.into()); +} + +pub(super) fn set_callback<'s>( + scope: &mut v8::PinScope<'s, '_>, + object: v8::Local<'s, v8::Object>, + slot: &str, + callback: webidl::WebIdlCallbackFunction, +) { + let callback = callbacks::trace(scope, callback); + set_private_value(scope, object, slot, callback.into()); +} diff --git a/moli-renderer-v8/src/script_vm/tests/mod.rs b/moli-renderer-v8/src/script_vm/tests/mod.rs index 9bea571bd..3af471e6f 100644 --- a/moli-renderer-v8/src/script_vm/tests/mod.rs +++ b/moli-renderer-v8/src/script_vm/tests/mod.rs @@ -15711,6 +15711,7 @@ mod inspector_unwrap; mod lazy_storage; mod lazy_window_surfaces; mod no_cors_header_fill; +mod observable; mod observer_callbacks; mod post_parse; mod queue_microtask; diff --git a/moli-renderer-v8/src/script_vm/tests/observable.rs b/moli-renderer-v8/src/script_vm/tests/observable.rs new file mode 100644 index 000000000..afcd1475b --- /dev/null +++ b/moli-renderer-v8/src/script_vm/tests/observable.rs @@ -0,0 +1,116 @@ +use super::*; + +#[test] +fn observable_core_lifecycle_conversion_brands_and_exceptions() { + let mut vm = new_storage_test_vm("https://observable.test/"); + let result = vm + .eval(&format!( + "JSON.stringify({})", + include_str!("../../../tests/fixtures/observable-core.js") + )) + .expect("Observable fixture should evaluate"); + let result: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(result["failures"], serde_json::json!([]), "{result}"); + assert!(result["checks"].as_u64().unwrap() >= 65, "{result}"); +} + +#[test] +fn observable_realm_guards_use_the_receiver_and_teardown_owner() { + let mut vm = new_storage_test_vm("https://observable-realms.test/"); + vm.eval( + r#" +(() => { + const frame = document.createElement('iframe'); + frame.id = 'observable-child'; + (document.body || document.documentElement || document).appendChild(frame); +})() +"#, + ) + .unwrap(); + materialize_single_child_default_realm_for_test(&mut vm, "Observable child realm"); + let result = vm + .eval( + r#" +JSON.stringify((() => { + const frame = document.getElementById('observable-child'); + const child = frame.contentWindow; + const events = []; + let subscriber, initializers = 0, conversions = 0; + const source = new child.Observable(s => { subscriber = s; initializers++; }); + child.Observable.prototype.subscribe.call(source, value => events.push(value)); + const checks = [subscriber instanceof child.Subscriber, + subscriber.signal instanceof child.AbortSignal]; + try { + child.Observable.prototype.subscribe.call({}, {get next() { conversions++; }}); + checks.push(false); + } catch (error) { + checks.push(error instanceof child.TypeError && !(error instanceof TypeError)); + } + checks.push(conversions === 0); + Subscriber.prototype.next.call(subscriber, 'live'); + const pending = new child.Observable(() => events.push('retired initializer')); + subscriber.addTeardown(() => events.push('retired teardown')); + subscriber.addTeardown(() => { events.push('remove'); frame.remove(); }); + Subscriber.prototype.complete.call(subscriber); + Observable.prototype.subscribe.call(pending); + Subscriber.prototype.next.call(subscriber, 'retired next'); + Subscriber.prototype.addTeardown.call(subscriber, () => events.push('retired added teardown')); + checks.push(initializers === 1, !subscriber.active, subscriber.signal.aborted); + return {checks, events}; +})()) +"#, + ) + .unwrap(); + assert_eq!( + result, + r#"{"checks":[true,true,true,true,true,true,true],"events":["live","remove"]}"# + ); +} + +#[test] +fn observable_weak_subscription_and_callback_cycles_are_collectible() { + let mut vm = new_storage_test_vm("https://observable-gc.test/"); + vm.eval( + r#" +(() => { + globalThis.__initializers = 0; + globalThis.__source = new Observable(s => { + __initializers++; + globalThis.__weakSubscriber = new WeakRef(s); + }); + __source.subscribe(); + // Isolate this lexical environment from the live source initializer above. + (() => { + let cycle; + cycle = new Observable(() => cycle); + globalThis.__weakCycle = new WeakRef(cycle); + })(); + (() => { + const captured = {}; + globalThis.__weakObserver = new WeakRef(captured); + globalThis.__controller = new AbortController(); + new Observable(s => s.complete()).subscribe(() => captured, {signal: __controller.signal}); + })(); +})() +"#, + ) + .unwrap(); + vm.renderer_document_isolate + .clone() + .with_entered_renderer_document_isolate(|isolate| { + isolate.clear_kept_objects(); + isolate.low_memory_notification(); + Ok(()) + }) + .unwrap(); + assert_eq!( + vm.eval( + r#" +JSON.stringify([__weakSubscriber.deref() === undefined, __weakCycle.deref() === undefined, + __weakObserver.deref() === undefined, (__source.subscribe(), __initializers)]) +"# + ) + .unwrap(), + "[true,true,true,2]" + ); +} diff --git a/moli-renderer-v8/src/web_api_interfaces.rs b/moli-renderer-v8/src/web_api_interfaces.rs index 820944fdc..9f774b6c3 100644 --- a/moli-renderer-v8/src/web_api_interfaces.rs +++ b/moli-renderer-v8/src/web_api_interfaces.rs @@ -316,6 +316,7 @@ interfaces! { NotificationEvent: ExtendableEvent; OfflineAudioCompletionEvent: Event; OfflineAudioContext: BaseAudioContext; + Observable; OffscreenCanvas; OffscreenCanvasRenderingContext2D; Option; @@ -480,6 +481,7 @@ interfaces! { StyleSheet; StyleSheetList; SubmitEvent: Event; + Subscriber; SubtleCrypto; SyncEvent: ExtendableEvent; SyncManager; diff --git a/moli-renderer-v8/src/worker/abort.rs b/moli-renderer-v8/src/worker/abort.rs index eb34e1856..5baf93c5f 100644 --- a/moli-renderer-v8/src/worker/abort.rs +++ b/moli-renderer-v8/src/worker/abort.rs @@ -332,6 +332,13 @@ fn worker_abort_store(scope: &mut v8::PinScope<'_, '_>) -> Option( + scope: &mut v8::PinScope<'s, '_>, +) -> Option> { + let store = worker_abort_store(scope)?; + create_signal(scope, &mut store.borrow_mut(), false, None) +} + pub(crate) fn worker_abort_signal_aborted<'s>( scope: &mut v8::PinScope<'s, '_>, signal: v8::Local<'s, v8::Object>, diff --git a/moli-renderer-v8/src/worker/thread/tests/postmessage.rs b/moli-renderer-v8/src/worker/thread/tests/postmessage.rs index e2ce1124f..9ff1a0d95 100644 --- a/moli-renderer-v8/src/worker/thread/tests/postmessage.rs +++ b/moli-renderer-v8/src/worker/thread/tests/postmessage.rs @@ -1,5 +1,21 @@ use super::*; +#[tokio::test] +async fn worker_observable_core_lifecycle_conversion_brands_and_exceptions() { + ensure_v8(); + let mut handle = spawn_worker( + format!( + "postMessage({}); close();", + include_str!("../../../../tests/fixtures/observable-core.js") + ), + "https://observable.test/worker.js".into(), + ); + let message = timeout(TIMEOUT, handle.recv()).await.unwrap().unwrap(); + let result: serde_json::Value = serde_json::from_str(&expect_post_json(message)).unwrap(); + assert_eq!(result["failures"], serde_json::json!([]), "{result}"); + assert!(result["checks"].as_u64().unwrap() >= 65, "{result}"); +} + #[tokio::test] async fn worker_trusted_types_webidl_surface_checks_descriptors_arguments_and_brands() { ensure_v8(); diff --git a/moli-renderer-v8/tests/fixtures/observable-core.js b/moli-renderer-v8/tests/fixtures/observable-core.js new file mode 100644 index 000000000..21099adf2 --- /dev/null +++ b/moli-renderer-v8/tests/fixtures/observable-core.js @@ -0,0 +1,146 @@ +(() => { + 'use strict'; + const failures = []; + let checks = 0; + const check = (ok, label) => { checks++; if (!ok) failures.push(label); }; + const same = (actual, expected, label) => check(JSON.stringify(actual) === JSON.stringify(expected), label); + const throws = (fn, label, expected = TypeError) => { + try { fn(); check(false, label); } + catch (error) { check(error instanceof expected, label); } + }; + const reported = []; + const onError = event => { reported.push(event.error); event.preventDefault(); }; + addEventListener('error', onError); + try { + for (const make of [() => Observable(() => {}), () => new Observable(), + () => new Observable({}), () => new Subscriber()]) throws(make, 'constructor validation'); + let calls = 0, subscriber; + const source = new Observable(function (value) { + check(this === undefined && arguments.length === 1, 'initializer callback this/arguments'); + calls++; + subscriber = value; + }); + check(calls === 0, 'construction is lazy'); + const order = []; + const controller1 = new AbortController(), controller2 = new AbortController(); + source.subscribe({ next(value) { check(this === undefined, 'observer this'); order.push('a:' + value); } }, {signal: controller1.signal}); + const signal = subscriber.signal; + check(signal !== controller1.signal && signal === subscriber.signal, 'stable independent signal'); + check(subscriber.active && !signal.aborted, 'active subscription'); + source.subscribe(value => order.push('b:' + value), {signal: controller2.signal}); + check(calls === 1, 'shared producer'); + subscriber.addTeardown(() => { order.push('first teardown'); check(!subscriber.active && signal.aborted, 'teardown sees closed state'); }); + subscriber.addTeardown(() => order.push('second teardown')); + signal.addEventListener('abort', () => order.push('abort')); + subscriber.next(1); + controller1.abort('one'); + check(subscriber.active && !signal.aborted, 'first cancellation keeps producer active'); + subscriber.next(2); + controller2.abort('two'); + check(!subscriber.active && signal.reason === 'two', 'last cancellation closes with reason'); + subscriber.next(3); + subscriber.complete(); + same(order, ['a:1', 'b:1', 'b:2', 'abort', 'second teardown', 'first teardown'], 'cancellation/teardown order'); + source.subscribe(); + check(calls === 2 && subscriber.signal !== signal, 'restart after cancellation'); + subscriber.complete(); + + const converted = []; + const observer = {}; + for (const name of ['next', 'error', 'complete']) Object.defineProperty(observer, name, {get() { converted.push(name); return undefined; }}); + source.subscribe(observer, {get signal() { converted.push('signal'); return undefined; }}); + same(converted, ['complete', 'error', 'next', 'signal'], 'dictionary conversion order'); + subscriber.complete(); + for (const observer of [1, true, 'text', {next: null}, {error: 1}, {complete: {}}]) throws(() => source.subscribe(observer), 'invalid observer'); + for (const signal of [null, {}, new Proxy(new AbortController().signal, {})]) throws(() => source.subscribe({}, {signal}), 'signal brand'); + for (const options of [1, 'text', true]) throws(() => source.subscribe({}, options), 'invalid options'); + const marker = {marker: true}; + try { source.subscribe({get next() { throw marker; }}); check(false, 'getter throws'); } + catch (error) { check(error === marker, 'getter preserves exception'); } + let conversions = 0, traps = 0; + const revoked = Proxy.revocable(source, {}); revoked.revoke(); + for (const fake of [{}, Object.create(source), new Proxy(source, {get() { traps++; }}), revoked.proxy]) { + throws(() => Observable.prototype.subscribe.call(fake, {get next() { conversions++; }}), 'Observable receiver'); + } + check(conversions === 0 && traps === 0, 'receiver precedes conversion and proxy traps'); + for (const fake of [{}, Object.create(subscriber), new Proxy(subscriber, {})]) { + for (const name of ['next', 'error', 'complete', 'addTeardown']) throws(() => Subscriber.prototype[name].call(fake, () => {}), 'Subscriber method receiver'); + for (const name of ['active', 'signal']) throws(() => Object.getOwnPropertyDescriptor(Subscriber.prototype, name).get.call(fake), 'Subscriber getter receiver'); + } + throws(() => subscriber.next(), 'next requires value'); + throws(() => subscriber.error(), 'error requires value'); + throws(() => subscriber.addTeardown(null), 'teardown requires callback'); + + const preAborted = AbortSignal.abort(marker), closed = []; + new Observable(s => { + check(!s.active && s.signal.aborted && s.signal.reason === marker, 'pre-aborted initializer runs inactive'); + s.addTeardown(() => closed.push('a')); + s.addTeardown(() => closed.push('b')); + s.next(1); s.complete(); + }).subscribe(() => closed.push('unexpected'), {signal: preAborted}); + same(closed, ['a', 'b'], 'inactive teardown runs synchronously'); + + const notifications = []; + let shared; + const multicast = new Observable(s => { shared = s; }); + const cancelled = new AbortController(); + multicast.subscribe(value => { + notifications.push('a' + value); + if (value === 1) { + multicast.subscribe(value => notifications.push('c' + value)); + cancelled.abort(); + } + }); + multicast.subscribe(value => notifications.push('b' + value), {signal: cancelled.signal}); + shared.next(1); shared.next(2); + same(notifications, ['a1', 'b1', 'a2', 'c2'], 'notification uses observer snapshot'); + shared.complete(); + + const exceptions = [new Error('initializer'), new Error('late'), new Error('observer'), new Error('teardown')]; + let handled; + new Observable(() => { throw exceptions[0]; }).subscribe({error(error) { handled = error; }}); + check(handled === exceptions[0] && reported.length === 0, 'initializer exception goes to observer'); + new Observable(s => { s.complete(); s.error(exceptions[1]); }).subscribe(); + check(reported.pop() === exceptions[1], 'late error is reported'); + const lifecycle = []; + new Observable(s => { + s.addTeardown(() => lifecycle.push('first')); + s.addTeardown(() => { lifecycle.push('throwing'); throw exceptions[3]; }); + s.addTeardown(() => { lifecycle.push('last'); s.addTeardown(() => lifecycle.push('nested')); s.complete(); }); + s.next(1); + s.complete(); + }).subscribe({next() { throw exceptions[2]; }, complete() { lifecycle.push('complete'); }}); + same(lifecycle, ['last', 'nested', 'throwing', 'first', 'complete'], 'reentrant teardown and exceptions'); + check(reported.shift() === exceptions[2] && reported.shift() === exceptions[3], 'callback exceptions reported in order'); + + // Author changes to the public constructors/methods must not redirect the + // native close algorithm or creation of its signal. + const savedAbortController = globalThis.AbortController; + const savedAbort = AbortController.prototype.abort; + const savedComplete = Subscriber.prototype.complete; + const savedError = Subscriber.prototype.error; + const savedSignal = Object.getOwnPropertyDescriptor(AbortController.prototype, 'signal'); + try { + globalThis.AbortController = () => { throw marker; }; + savedAbortController.prototype.abort = () => { throw marker; }; + Object.defineProperty(savedAbortController.prototype, 'signal', {configurable: true, get() { throw marker; }}); + Subscriber.prototype.complete = Subscriber.prototype.error = () => { throw marker; }; + let completed = false, seen; + new Observable(s => { savedComplete.call(s); }).subscribe({complete() { completed = true; }}); + new Observable(() => { throw marker; }).subscribe({error(value) { seen = value; }}); + check(completed && seen === marker, 'internal algorithms bypass public methods'); + } finally { + globalThis.AbortController = savedAbortController; + savedAbortController.prototype.abort = savedAbort; + Object.defineProperty(savedAbortController.prototype, 'signal', savedSignal); + Subscriber.prototype.complete = savedComplete; + Subscriber.prototype.error = savedError; + } + check(reported.length === 0, 'no unexpected callback errors'); + } catch (error) { + failures.push('uncaught: ' + error); + } finally { + removeEventListener('error', onError); + } + return {checks, failures}; +})()