diff --git a/changelog.d/8825-async-hooks-lifecycle.md b/changelog.d/8825-async-hooks-lifecycle.md new file mode 100644 index 0000000000..1f1b1c9a50 --- /dev/null +++ b/changelog.d/8825-async-hooks-lifecycle.md @@ -0,0 +1,3 @@ +### Fixed + +- Completed Node `async_hooks` lifecycle support across async resources, event emitters, HTTP, sockets, workers, zlib, DNS, and WebCrypto. Provider scopes now restore execution and `AsyncLocalStorage` state when hooks or callbacks throw, deferred destroy hooks run at the correct lifecycle boundary, and allocation-sensitive values remain rooted across moving garbage collections. diff --git a/crates/perry-codegen/src/expr/this_super_call.rs b/crates/perry-codegen/src/expr/this_super_call.rs index ad8b47a3a8..48fdc8bdd9 100644 --- a/crates/perry-codegen/src/expr/this_super_call.rs +++ b/crates/perry-codegen/src/expr/this_super_call.rs @@ -261,7 +261,9 @@ pub(crate) fn lower(ctx: &mut FnCtx<'_>, expr: &Expr) -> Result { let async_parent = ctx .classes .get(¤t_class_name) - .and_then(|class| class.extends_name.clone()); + .filter(|class| class.extends_expr.is_none() && !class.heritage_lexically_shadowed) + .and_then(|class| class.extends_name.clone()) + .filter(|parent| !ctx.classes.contains_key(parent.as_str())); if matches!( async_parent.as_deref(), Some("EventEmitterAsyncResource" | "AsyncLocalStorage" | "AsyncResource") @@ -269,15 +271,22 @@ pub(crate) fn lower(ctx: &mut FnCtx<'_>, expr: &Expr) -> Result { let undef = double_literal(f64::from_bits(crate::nanbox::TAG_UNDEFINED)); let zero_idx = "0".to_string(); let one_idx = "1".to_string(); - let first = - ctx.block() - .call(DOUBLE, "js_array_get_f64", &[(I64, &arr), (I32, &zero_idx)]); - let second = - ctx.block() - .call(DOUBLE, "js_array_get_f64", &[(I64, &arr), (I32, &one_idx)]); - rooting::with_rooted_group(ctx, 3, |ctx, group| { + rooting::with_rooted_group(ctx, 4, |ctx, group| { let this_root = group.adopt_emitted(ctx, Repr::Boxed, &this_box, true); + let arr_root = group.adopt_emitted(ctx, Repr::Ptr, &arr, true); + let arr = group.reread_emitted(ctx, arr_root); + let first = ctx.block().call( + DOUBLE, + "js_array_get_f64", + &[(I64, &arr), (I32, &zero_idx)], + ); let first_root = group.adopt_emitted(ctx, Repr::Boxed, &first, true); + let arr = group.reread_emitted(ctx, arr_root); + let second = ctx.block().call( + DOUBLE, + "js_array_get_f64", + &[(I64, &arr), (I32, &one_idx)], + ); let second_root = group.adopt_emitted(ctx, Repr::Boxed, &second, true); let this_box = group.reread_emitted(ctx, this_root); match async_parent.as_deref() { diff --git a/crates/perry-codegen/src/lower_call/builtin.rs b/crates/perry-codegen/src/lower_call/builtin.rs index 6e3e29f1ba..41c766ed3f 100644 --- a/crates/perry-codegen/src/lower_call/builtin.rs +++ b/crates/perry-codegen/src/lower_call/builtin.rs @@ -153,6 +153,7 @@ pub(super) fn lower_builtin_new<'a>( Some(index) => group.reread(ctx, index)?, None => double_literal(f64::from_bits(crate::nanbox::TAG_UNDEFINED)), }; + let options = group.adopt_emitted(ctx, crate::rooting::Repr::Boxed, &options, true); let runtime = if import_src.is_some_and(|source| { source.strip_prefix("node:").unwrap_or(source) == "dns/promises" }) { @@ -163,12 +164,10 @@ pub(super) fn lower_builtin_new<'a>( ctx.pending_declares .push((runtime.to_string(), DOUBLE, vec![I64])); let zero = "0".to_string(); - let args_array = ctx.block().call(I64, "js_array_alloc", &[(I32, &zero)]); - let args_array = ctx.block().call( - I64, - "js_array_push_f64", - &[(I64, &args_array), (DOUBLE, &options)], - ); + let args_array = group.begin_array(ctx, &zero); + let options = group.reread_emitted(ctx, options); + group.push_array(ctx, args_array, &options); + let args_array = group.read_array(ctx, args_array); Ok(Some(ctx.block().call( DOUBLE, runtime, diff --git a/crates/perry-ext-events/src/emit_scope.rs b/crates/perry-ext-events/src/emit_scope.rs new file mode 100644 index 0000000000..ddee1fb1a3 --- /dev/null +++ b/crates/perry-ext-events/src/emit_scope.rs @@ -0,0 +1,32 @@ +use super::*; + +pub(super) struct EventEmitterEmitCall { + pub(super) handle: Handle, + pub(super) event_value: TransientRootedNanbox, + pub(super) args_ptr: TransientRootedAddr, +} + +pub(super) unsafe extern "C" fn event_emitter_emit_thunk(data: *mut c_void) -> f64 { + let call = &mut *(data as *mut EventEmitterEmitCall); + let Some(event_name) = event_name_from_bits(call.event_value.get().to_bits() as i64) else { + return f64::from_bits(0x7FFC_0000_0000_0003); + }; + js_event_emitter_emit_impl( + call.handle, + &event_name, + call.args_ptr.get() as *mut ArrayHeader, + ) +} + +pub(super) struct EventEmitterEmit0Call { + pub(super) handle: Handle, + pub(super) event_value: TransientRootedNanbox, +} + +pub(super) unsafe extern "C" fn event_emitter_emit0_thunk(data: *mut c_void) -> f64 { + let call = &mut *(data as *mut EventEmitterEmit0Call); + let Some(event_name) = event_name_from_bits(call.event_value.get().to_bits() as i64) else { + return f64::from_bits(0x7FFC_0000_0000_0003); + }; + js_event_emitter_emit0_impl(call.handle, &event_name) +} diff --git a/crates/perry-ext-events/src/lib.rs b/crates/perry-ext-events/src/lib.rs index cfeb124c79..e29413be7b 100644 --- a/crates/perry-ext-events/src/lib.rs +++ b/crates/perry-ext-events/src/lib.rs @@ -23,7 +23,8 @@ use perry_ffi::{ error_value_with_code, js_array_alloc, js_array_get, js_array_length, js_array_push, js_array_set, js_object_alloc_with_shape, js_object_set_field, nanbox_string_bits, read_string, throw_with_code, ArrayHeader, ErrorKind, Handle, JsPromise, JsString, JsValue, ObjectHeader, - Promise, RawClosureHeader, StringHeader, TransientRootScope, + Promise, RawClosureHeader, StringHeader, TransientRootScope, TransientRootedAddr, + TransientRootedNanbox, }; use std::collections::{HashMap, HashSet}; use std::ffi::c_void; @@ -31,6 +32,11 @@ use std::sync::{Mutex, MutexGuard, Once, OnceLock}; mod error_monitor; use error_monitor::dispatch_error_monitor; +mod emit_scope; +use emit_scope::{ + event_emitter_emit0_thunk, event_emitter_emit_thunk, EventEmitterEmit0Call, + EventEmitterEmitCall, +}; mod max_listeners; mod messages; mod module_helpers; @@ -686,9 +692,13 @@ unsafe fn string_from_header(ptr: *const StringHeader) -> Option { read_string(handle).map(String::from) } +fn is_raw_string_header_bits(raw: u64) -> bool { + (0x10000..MAX_HEAP_POINTER).contains(&raw) && (raw & TAG_MASK) == 0 +} + unsafe fn event_name_from_bits(event_bits: i64) -> Option { let raw = event_bits as u64; - if (0x10000..MAX_HEAP_POINTER).contains(&raw) && (raw & TAG_MASK) == 0 { + if is_raw_string_header_bits(raw) { return string_from_header(raw as *const StringHeader); } @@ -696,6 +706,15 @@ unsafe fn event_name_from_bits(event_bits: i64) -> Option { string_from_header(rendered as *const StringHeader) } +fn event_value_from_bits(event_bits: i64) -> f64 { + let raw = event_bits as u64; + if is_raw_string_header_bits(raw) { + f64::from_bits(nanbox_string_bits(raw as *mut StringHeader)) + } else { + f64::from_bits(raw) + } +} + fn event_bits_from_string_ptr(ptr: *const StringHeader) -> i64 { f64::from_bits(nanbox_string_bits(ptr as *mut StringHeader)).to_bits() as i64 } @@ -1296,16 +1315,19 @@ pub unsafe extern "C" fn js_event_emitter_emit( event_bits: i64, args_ptr: *mut ArrayHeader, ) -> f64 { - if event_name_from_bits(event_bits).is_none() { - return f64::from_bits(0x7FFC_0000_0000_0003); - } + let roots = TransientRootScope::enter(); + let event_value = roots.root_nanbox(event_value_from_bits(event_bits)); + let args_ptr = roots.root_addr(args_ptr as i64); let async_id = event_emitter_async_id(handle); if async_id == 0 { - return js_event_emitter_emit_impl(handle, event_bits, args_ptr); + let Some(event_name) = event_name_from_bits(event_value.get().to_bits() as i64) else { + return f64::from_bits(0x7FFC_0000_0000_0003); + }; + return js_event_emitter_emit_impl(handle, &event_name, args_ptr.get() as *mut ArrayHeader); } let mut call = EventEmitterEmitCall { handle, - event_bits, + event_value, args_ptr, }; js_async_hooks_provider_run_catching( @@ -1315,42 +1337,28 @@ pub unsafe extern "C" fn js_event_emitter_emit( ) } -struct EventEmitterEmitCall { - handle: Handle, - event_bits: i64, - args_ptr: *mut ArrayHeader, -} - -unsafe extern "C" fn event_emitter_emit_thunk(data: *mut std::ffi::c_void) -> f64 { - let call = &mut *(data as *mut EventEmitterEmitCall); - js_event_emitter_emit_impl(call.handle, call.event_bits, call.args_ptr) -} - unsafe fn js_event_emitter_emit_impl( handle: Handle, - event_bits: i64, + event_name: &str, args_ptr: *mut ArrayHeader, ) -> f64 { const TAG_FALSE_F64: f64 = f64::from_bits(0x7FFC_0000_0000_0003); const TAG_TRUE_F64: f64 = f64::from_bits(0x7FFC_0000_0000_0004); - let Some(event_name) = event_name_from_bits(event_bits) else { - return TAG_FALSE_F64; - }; let mut had_listeners = false; let mut domain_error: Option<(Handle, f64)> = None; let mut throw_error: Option = None; if let Some(emitter) = get_event_emitter_mut(handle) { - let snapshot: Vec = match emitter.events.get(&event_name) { + let snapshot: Vec = match emitter.events.get(event_name) { Some(v) if !v.is_empty() => v.clone(), _ => Vec::new(), }; if !snapshot.is_empty() { had_listeners = true; if snapshot.iter().any(|l| l.once) { - if let Some(v) = emitter.events.get_mut(&event_name) { + if let Some(v) = emitter.events.get_mut(event_name) { v.retain(|l| !l.once); } - emitter.prune_event_if_empty(&event_name); + emitter.prune_event_if_empty(event_name); } } @@ -1374,7 +1382,7 @@ unsafe fn js_event_emitter_emit_impl( } if domain_error.is_none() && throw_error.is_none() { - drain_pending_once_promises(emitter, &event_name, args_ptr); + drain_pending_once_promises(emitter, event_name, args_ptr); let capture_rejections = emitter.capture_rejections && event_name != "error"; for l in snapshot { @@ -1412,14 +1420,19 @@ unsafe fn js_event_emitter_emit_impl( /// `event_name_ptr` must be null or a Perry-runtime `StringHeader`. #[no_mangle] pub unsafe extern "C" fn js_event_emitter_emit0(handle: Handle, event_bits: i64) -> f64 { - if event_name_from_bits(event_bits).is_none() { - return f64::from_bits(0x7FFC_0000_0000_0003); - } + let roots = TransientRootScope::enter(); + let event_value = roots.root_nanbox(event_value_from_bits(event_bits)); let async_id = event_emitter_async_id(handle); if async_id == 0 { - return js_event_emitter_emit0_impl(handle, event_bits); + let Some(event_name) = event_name_from_bits(event_value.get().to_bits() as i64) else { + return f64::from_bits(0x7FFC_0000_0000_0003); + }; + return js_event_emitter_emit0_impl(handle, &event_name); } - let mut call = EventEmitterEmit0Call { handle, event_bits }; + let mut call = EventEmitterEmit0Call { + handle, + event_value, + }; js_async_hooks_provider_run_catching( async_id, event_emitter_emit0_thunk, @@ -1427,37 +1440,24 @@ pub unsafe extern "C" fn js_event_emitter_emit0(handle: Handle, event_bits: i64) ) } -struct EventEmitterEmit0Call { - handle: Handle, - event_bits: i64, -} - -unsafe extern "C" fn event_emitter_emit0_thunk(data: *mut std::ffi::c_void) -> f64 { - let call = &mut *(data as *mut EventEmitterEmit0Call); - js_event_emitter_emit0_impl(call.handle, call.event_bits) -} - -unsafe fn js_event_emitter_emit0_impl(handle: Handle, event_bits: i64) -> f64 { +unsafe fn js_event_emitter_emit0_impl(handle: Handle, event_name: &str) -> f64 { const TAG_FALSE_F64: f64 = f64::from_bits(0x7FFC_0000_0000_0003); const TAG_TRUE_F64: f64 = f64::from_bits(0x7FFC_0000_0000_0004); - let Some(event_name) = event_name_from_bits(event_bits) else { - return TAG_FALSE_F64; - }; let mut had_listeners = false; let mut domain_error: Option<(Handle, f64)> = None; let mut throw_error: Option = None; if let Some(emitter) = get_event_emitter_mut(handle) { - let snapshot: Vec = match emitter.events.get(&event_name) { + let snapshot: Vec = match emitter.events.get(event_name) { Some(v) if !v.is_empty() => v.clone(), _ => Vec::new(), }; if !snapshot.is_empty() { had_listeners = true; if snapshot.iter().any(|l| l.once) { - if let Some(v) = emitter.events.get_mut(&event_name) { + if let Some(v) = emitter.events.get_mut(event_name) { v.retain(|l| !l.once); } - emitter.prune_event_if_empty(&event_name); + emitter.prune_event_if_empty(event_name); } } @@ -1480,7 +1480,7 @@ unsafe fn js_event_emitter_emit0_impl(handle: Handle, event_bits: i64) -> f64 { } } if domain_error.is_none() && throw_error.is_none() { - drain_pending_once_promises(emitter, &event_name, empty_args); + drain_pending_once_promises(emitter, event_name, empty_args); let capture_rejections = emitter.capture_rejections && event_name != "error"; for l in snapshot { diff --git a/crates/perry-ext-http/src/lib.rs b/crates/perry-ext-http/src/lib.rs index 1e357215ab..5e4407fdb2 100644 --- a/crates/perry-ext-http/src/lib.rs +++ b/crates/perry-ext-http/src/lib.rs @@ -1802,8 +1802,10 @@ pub unsafe extern "C" fn js_http_once( if callback == 0 { return handle; } + let roots = perry_ffi::TransientRootScope::enter(); + let callback = roots.root_addr(callback); let wrapper = - client_request_surface::create_client_once_wrapper(handle, &event, callback, false); + client_request_surface::create_client_once_wrapper(handle, &event, callback.get(), false); let mut matched = false; with_handle_mut::(handle, |request| { request @@ -1811,7 +1813,7 @@ pub unsafe extern "C" fn js_http_once( .entry(event.clone()) .or_default() .push(ClientEventListener { - callback, + callback: callback.get(), raw_wrapper: wrapper, once: true, }); diff --git a/crates/perry-ext-http/src/server/handle_dispatch.rs b/crates/perry-ext-http/src/server/handle_dispatch.rs index 86ad3ea08b..398677b020 100644 --- a/crates/perry-ext-http/src/server/handle_dispatch.rs +++ b/crates/perry-ext-http/src/server/handle_dispatch.rs @@ -126,8 +126,6 @@ extern "C" { fn js_node_http_im_resume(handle: i64); fn js_node_http_im_destroy(handle: i64); fn js_node_http_im_on(handle: i64, event_name_ptr: *const StringHeader, callback: i64) -> f64; - fn js_node_http_im_once(handle: i64, event_name_ptr: *const StringHeader, callback: i64) - -> f64; fn js_node_http_im_set_encoding(handle: i64, encoding_ptr: *const StringHeader) -> i64; fn js_node_http_im_set_timeout(handle: i64, msecs: f64, callback: i64) -> i64; fn js_node_http_im_read(handle: i64) -> f64; @@ -374,14 +372,6 @@ pub unsafe extern "C" fn js_ext_http_server_dispatch_method( } self_ref } - "once" if args.len() >= 2 => { - let event_ptr = string_arg(args[0]); - if event_ptr.is_null() { - return self_ref; - } - js_node_http_im_once(handle, event_ptr, closure_arg(Some(args[1]))); - self_ref - } "once" if args.len() >= 2 => { let event = read_string_header(string_arg(args[0]) as *mut StringHeader).unwrap_or_default(); diff --git a/crates/perry-ext-net/src/adopt.rs b/crates/perry-ext-net/src/adopt.rs index d3d8f331ad..e972f07c91 100644 --- a/crates/perry-ext-net/src/adopt.rs +++ b/crates/perry-ext-net/src/adopt.rs @@ -53,6 +53,7 @@ pub fn adopt_upgraded_tcp_stream(stream: tokio::net::TcpStream) -> i64 { destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: None, diff --git a/crates/perry-ext-net/src/ipc.rs b/crates/perry-ext-net/src/ipc.rs index b501471f3a..afd088afdf 100644 --- a/crates/perry-ext-net/src/ipc.rs +++ b/crates/perry-ext-net/src/ipc.rs @@ -44,6 +44,7 @@ fn allocate_socket() -> (i64, mpsc::UnboundedReceiver) { destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: None, @@ -114,6 +115,7 @@ pub(crate) fn register_accepted_transport( destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: Some(server_id), diff --git a/crates/perry-ext-net/src/lib.rs b/crates/perry-ext-net/src/lib.rs index 98ee742735..068c5c1527 100644 --- a/crates/perry-ext-net/src/lib.rs +++ b/crates/perry-ext-net/src/lib.rs @@ -12,20 +12,10 @@ //! //! # Differences from the perry-stdlib version //! -//! - Uses `perry_ffi::spawn_async` to drive each socket reader / server accept -//! loop cooperatively on Perry's shared multi-thread runtime (the same -//! reactor `crate::common::async_bridge` drives), rather than spinning a -//! throwaway current-thread runtime on a blocking-pool thread per socket. -//! Keepalive comes from `js_ext_net_has_active_handles` (the socket/server is -//! registered synchronously before the spawn), not the blocking-pool -//! active-handle counter. -//! - Uses `perry_ffi::JsClosure` instead of raw `js_closure_call*` extern fns. -//! - Uses `perry_ffi::alloc_buffer` / `BufferHeader` instead of -//! `perry-runtime::buffer::*` directly. -//! - GC root scanner registered via `perry_ffi::gc_register_mutable_root_scanner`. -//! Listeners stored inside the `NET_LISTENERS` map need this — issue #35 -//! pattern — and the mutable visitor lets copied-minor GC rewrite moved -//! closure pointers in place. +//! - Uses `perry_ffi::spawn_async` on Perry's shared runtime, with keepalive +//! provided by `js_ext_net_has_active_handles`. +//! - Uses perry-ffi closures, buffers, and mutable GC root scanning; the latter +//! rewrites listener pointers after a copying minor collection. //! //! TLS is unconditionally compiled in (no `#[cfg(feature = "tls")]` gates //! like perry-stdlib has) — keeping the wrapper crate simple, the deps are @@ -276,6 +266,7 @@ pub(crate) struct SocketState { pub(crate) destroyed: bool, pub(crate) bytes_read: u64, pub(crate) bytes_written: u64, + pub(crate) bytes_queued: u64, pub(crate) timeout: Option, pub(crate) type_of_service: u8, pub(crate) server_id: Option, @@ -301,6 +292,7 @@ impl SocketState { destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: None, @@ -352,9 +344,9 @@ enum PendingNetEvent { Data(i64, Bytes), /// Peer half-closed (FIN received); public readable-side `end` event. End(i64), - /// Writable-side shutdown requested by `socket.end()`, distinct from FIN; - /// fires the public `end` event. + /// A queued `socket.write` finished with a completion token and optional error. WriteComplete(i64, u64, Option), + /// `socket.end()` writable shutdown with a completion token and optional error. ShutdownComplete(i64, u64, Option), Close(i64), Error(i64, String), @@ -554,6 +546,7 @@ pub unsafe extern "C" fn js_net_socket_alloc() -> i64 { destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: None, @@ -1081,6 +1074,7 @@ where destroyed: false, bytes_read: 0, bytes_written: 0, + bytes_queued: 0, timeout: None, type_of_service: 0, server_id: None, @@ -1238,7 +1232,9 @@ pub(crate) async fn run_socket_task( }; match command { Some(SocketCommand::Write(bytes, completion)) => { - if let Err(e) = t.write_all(&bytes).await { + if let Err(e) = + lifecycle::write_socket_bytes(t, id, &bytes).await + { let msg = format!("{}", e); if completion != 0 { push_event(PendingNetEvent::WriteComplete( @@ -1341,7 +1337,9 @@ pub(crate) async fn run_socket_task( buffer_pool::checkin(buf); match cmd { Some(SocketCommand::Write(bytes, completion)) => { - if let Err(e) = t.write_all(&bytes).await { + if let Err(e) = + lifecycle::write_socket_bytes(t, id, &bytes).await + { let msg = format!("{}", e); if completion != 0 { push_event(PendingNetEvent::WriteComplete( diff --git a/crates/perry-ext-net/src/lifecycle.rs b/crates/perry-ext-net/src/lifecycle.rs index 2dee4476c3..98f16e1590 100644 --- a/crates/perry-ext-net/src/lifecycle.rs +++ b/crates/perry-ext-net/src/lifecycle.rs @@ -22,8 +22,10 @@ use perry_ffi::{alloc_string, nanbox_string_bits, ArrayHeader, JsValue, StringHeader}; use std::collections::HashSet; +use std::io; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Mutex, OnceLock}; +use tokio::io::AsyncWriteExt; use crate::statics; use crate::string_from_header_i64; @@ -84,10 +86,17 @@ pub(crate) unsafe fn dispatch_socket_completion(completion: u64, error: Option>(); + for completion in completions { + unsafe { + dispatch_socket_completion(completion, Some("Socket is closed".to_string())); + } + } } /// NaN-box a freshly allocated runtime string as an `f64` JS value. @@ -167,14 +176,16 @@ pub unsafe extern "C" fn js_net_socket_get_bytes_read(handle: i64) -> f64 { with_socket(handle, 0u64, |s| s.bytes_read) as f64 } -/// `socket.bytesWritten` — total bytes queued for the socket. +/// `socket.bytesWritten` — bytes dispatched to the transport or still queued. /// /// # Safety /// /// See [`js_net_socket_get_pending`]. #[no_mangle] pub unsafe extern "C" fn js_net_socket_get_bytes_written(handle: i64) -> f64 { - with_socket(handle, 0u64, |s| s.bytes_written) as f64 + with_socket(handle, 0u64, |s| { + s.bytes_written.saturating_add(s.bytes_queued) + }) as f64 } /// `socket.timeout` — the value set via `setTimeout(ms)`, or `undefined`. @@ -336,20 +347,61 @@ pub unsafe extern "C" fn js_ext_net_socket_write(handle: i64, chunk_bits: i64) { fn enqueue_socket_write(handle: i64, bytes: Vec, completion: u64) { let mut sockets = statics::sockets().lock().unwrap(); - if let Some(s) = sockets.get_mut(&handle) { - s.bytes_written = s.bytes_written.saturating_add(bytes.len() as u64); + let failure = if let Some(s) = sockets.get_mut(&handle) { + let byte_len = bytes.len() as u64; if s.cmd_tx .send(crate::SocketCommand::Write(bytes, completion)) .is_err() - && completion != 0 { - socket_completions().lock().unwrap().remove(&completion); + Some("Socket write failed") + } else { + s.bytes_queued = s.bytes_queued.saturating_add(byte_len); + None } - } else if completion != 0 { - socket_completions().lock().unwrap().remove(&completion); + } else { + Some("Socket is closed") + }; + drop(sockets); + if completion != 0 { + if let Some(message) = failure { + unsafe { + dispatch_socket_completion(completion, Some(message.to_string())); + } + } + } +} + +fn record_socket_write_progress(handle: i64, written: usize) { + if written == 0 { + return; + } + if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&handle) { + let written = written as u64; + socket.bytes_queued = socket.bytes_queued.saturating_sub(written); + socket.bytes_written = socket.bytes_written.saturating_add(written); } } +pub(crate) async fn write_socket_bytes( + transport: &mut crate::Transport, + handle: i64, + bytes: &[u8], +) -> io::Result<()> { + let mut written = 0; + while written < bytes.len() { + let count = transport.write(&bytes[written..]).await?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "failed to write socket bytes", + )); + } + written += count; + record_socket_write_progress(handle, count); + } + Ok(()) +} + /// `socket.write(chunk)` under the name the static NATIVE_MODULE_TABLE path /// emits. Delegates to the collision-proof [`js_ext_net_socket_write`] via a /// crate-local call, so even when the bundled stdlib's same-named twin wins the @@ -405,7 +457,10 @@ pub unsafe extern "C" fn js_ext_net_socket_write3( let completion = register_socket_completion(handle, completion); let Some(bytes) = crate::jsvalue_to_socket_bytes(chunk) else { if completion != 0 { - socket_completions().lock().unwrap().remove(&completion); + dispatch_socket_completion( + completion, + Some("Invalid data passed to socket.write".to_string()), + ); } return; }; @@ -445,8 +500,10 @@ pub unsafe extern "C" fn js_ext_net_socket_end(handle: i64, chunk_bits: i64) { if let Some(s) = sockets.get_mut(&handle) { if let Some(bytes) = final_bytes { if !bytes.is_empty() { - s.bytes_written = s.bytes_written.saturating_add(bytes.len() as u64); - let _ = s.cmd_tx.send(crate::SocketCommand::Write(bytes, 0)); + let byte_len = bytes.len() as u64; + if s.cmd_tx.send(crate::SocketCommand::Write(bytes, 0)).is_ok() { + s.bytes_queued = s.bytes_queued.saturating_add(byte_len); + } } } let _ = s.cmd_tx.send(crate::SocketCommand::End(0)); @@ -497,8 +554,14 @@ pub unsafe extern "C" fn js_ext_net_socket_end3( let mut sockets = statics::sockets().lock().unwrap(); if let Some(socket) = sockets.get_mut(&handle) { if let Some(bytes) = final_bytes.filter(|bytes| !bytes.is_empty()) { - socket.bytes_written = socket.bytes_written.saturating_add(bytes.len() as u64); - let _ = socket.cmd_tx.send(crate::SocketCommand::Write(bytes, 0)); + let byte_len = bytes.len() as u64; + if socket + .cmd_tx + .send(crate::SocketCommand::Write(bytes, 0)) + .is_ok() + { + socket.bytes_queued = socket.bytes_queued.saturating_add(byte_len); + } } if socket .cmd_tx @@ -1179,4 +1242,39 @@ mod tests { reset_handle(handle); } + + #[test] + fn rejected_write_does_not_increase_bytes_written() { + let handle = -91_238; + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + drop(rx); + statics::sockets() + .lock() + .unwrap() + .insert(handle, crate::SocketState::for_test(tx)); + + enqueue_socket_write(handle, vec![1, 2, 3], 0); + assert_eq!(unsafe { js_net_socket_get_bytes_written(handle) }, 0.0); + + statics::sockets().lock().unwrap().remove(&handle); + } + + #[test] + fn bytes_written_includes_queue_then_keeps_only_dispatched_progress_on_close() { + let handle = -91_239; + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + statics::sockets() + .lock() + .unwrap() + .insert(handle, crate::SocketState::for_test(tx)); + + enqueue_socket_write(handle, vec![1, 2, 3, 4], 0); + assert_eq!(unsafe { js_net_socket_get_bytes_written(handle) }, 4.0); + record_socket_write_progress(handle, 2); + assert_eq!(unsafe { js_net_socket_get_bytes_written(handle) }, 4.0); + crate::server_state::mark_socket_closed(handle); + assert_eq!(unsafe { js_net_socket_get_bytes_written(handle) }, 2.0); + + statics::sockets().lock().unwrap().remove(&handle); + } } diff --git a/crates/perry-ext-net/src/server_state.rs b/crates/perry-ext-net/src/server_state.rs index e16cebf739..276b88befc 100644 --- a/crates/perry-ext-net/src/server_state.rs +++ b/crates/perry-ext-net/src/server_state.rs @@ -350,6 +350,7 @@ pub(crate) fn mark_socket_closed(socket_id: i64) { return; }; socket.is_open = false; + socket.bytes_queued = 0; let Some(server_id) = socket.server_id.take() else { return; }; diff --git a/crates/perry-ext-zlib/src/stream.rs b/crates/perry-ext-zlib/src/stream.rs index dc33937822..8cabbb3899 100644 --- a/crates/perry-ext-zlib/src/stream.rs +++ b/crates/perry-ext-zlib/src/stream.rs @@ -20,9 +20,10 @@ use perry_ffi::{ alloc_buffer, alloc_string, gc_register_mutable_root_scanner_named, notify_main_thread, BufferHeader, ErrorKind, GcRootVisitor, JsClosure, JsValue, RawClosureHeader, StringHeader, - TransientRootScope, + TransientRootScope, TransientRootedAddr, }; use std::collections::{HashMap, HashSet, VecDeque}; +use std::ffi::c_void; use std::io::{Read, Write}; use std::sync::Mutex; @@ -68,9 +69,23 @@ extern "C" { // synchronously before queuing codec work. pub(crate) fn js_zlib_validate_callback(callback: f64) -> i64; fn js_async_hooks_provider_init(type_ptr: *const u8, type_len: usize) -> u64; - fn js_async_hooks_provider_defer_destroy(async_id: u64, check_turns: u32); - fn js_async_hooks_provider_enter(async_id: u64); - fn js_async_hooks_provider_leave(async_id: u64); + fn js_async_hooks_provider_run_catching( + async_id: u64, + callback: unsafe extern "C" fn(*mut c_void) -> f64, + data: *mut c_void, + ) -> f64; + fn js_async_hooks_provider_run_catching_deferred_destroy( + async_id: u64, + check_turns: u32, + callback: unsafe extern "C" fn(*mut c_void) -> f64, + data: *mut c_void, + ) -> f64; + fn js_async_hooks_provider_run_catching_deferred_destroy_on_error( + async_id: u64, + check_turns: u32, + callback: unsafe extern "C" fn(*mut c_void) -> f64, + data: *mut c_void, + ) -> f64; fn js_native_call_method_str_key( object: f64, name_handle: i64, @@ -685,17 +700,22 @@ unsafe fn call_one_shot_callback(callback: i64, result: Result, String>) if callback == 0 { return; } + let roots = TransientRootScope::enter(); + let callback = roots.root_addr(callback); match result { Ok(bytes) => { let err = f64::from_bits(JsValue::NULL.bits()); - let out = make_buffer_f64(&bytes) - .unwrap_or_else(|| f64::from_bits(JsValue::UNDEFINED.bits())); - let _ = JsClosure::from_raw(callback as *const RawClosureHeader).call2(err, out); + let out = roots.root_nanbox( + make_buffer_f64(&bytes) + .unwrap_or_else(|| f64::from_bits(JsValue::UNDEFINED.bits())), + ); + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader) + .call2(err, out.get()); } Err(msg) => { - let err = build_error_object(&msg); - let _ = JsClosure::from_raw(callback as *const RawClosureHeader) - .call2(err, f64::from_bits(JsValue::UNDEFINED.bits())); + let err = roots.root_nanbox(build_error_object(&msg)); + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader) + .call2(err.get(), f64::from_bits(JsValue::UNDEFINED.bits())); } } } @@ -1316,6 +1336,121 @@ unsafe fn build_error_object(msg: &str) -> f64 { f64::from_bits(POINTER_TAG | (obj as u64 & POINTER_MASK)) } +struct ZlibEventDispatch { + event: Option, +} + +unsafe extern "C" fn zlib_event_dispatch_thunk(data: *mut c_void) -> f64 { + let call = &mut *(data as *mut ZlibEventDispatch); + let event = call + .event + .take() + .expect("zlib event dispatch thunk must run exactly once"); + match event { + ZlibEvent::Data(id, bytes) => { + publish_bytes_written(id); + let roots = TransientRootScope::enter(); + let callbacks = roots.root_addrs(&listeners_for(id, "data")); + let destinations = pipes_for(id) + .into_iter() + .map(|bits| roots.root_nanbox(f64::from_bits(bits))) + .collect::>(); + if callbacks.is_empty() && destinations.is_empty() { + buffer_output_for_late_consumer(&mut statics().lock().unwrap(), id, &bytes); + } else { + if !callbacks.is_empty() { + if let Some(buffer) = make_buffer_f64(&bytes) { + let buffer = roots.root_nanbox(buffer); + for callback in callbacks { + if callback.get() != 0 { + let _ = + JsClosure::from_raw(callback.get() as *const RawClosureHeader) + .call1(buffer.get()); + } + } + } + } + for destination in destinations { + forward_write(destination.get().to_bits(), &bytes); + } + } + } + ZlibEvent::Finish(id) => { + let roots = TransientRootScope::enter(); + for callback in roots.root_addrs(&listeners_for(id, "finish")) { + if callback.get() != 0 { + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader).call0(); + } + } + } + ZlibEvent::End(id) => { + publish_bytes_written(id); + let roots = TransientRootScope::enter(); + let end_callbacks = roots.root_addrs(&listeners_for(id, "end")); + let destinations = pipes_for(id) + .into_iter() + .map(|bits| roots.root_nanbox(f64::from_bits(bits))) + .collect::>(); + let close_callbacks = roots.root_addrs(&listeners_for(id, "close")); + drop_buffered_stream(&mut statics().lock().unwrap(), id); + for callback in end_callbacks { + if callback.get() != 0 { + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader).call0(); + } + } + for destination in destinations { + forward_end(destination.get().to_bits()); + } + for callback in close_callbacks { + if callback.get() != 0 { + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader).call0(); + } + } + } + ZlibEvent::Error(id, message) => { + let roots = TransientRootScope::enter(); + let callbacks = roots.root_addrs(&listeners_for(id, "error")); + drop_buffered_stream(&mut statics().lock().unwrap(), id); + let error = roots.root_nanbox(build_error_object(&message)); + for callback in callbacks { + if callback.get() != 0 { + let _ = JsClosure::from_raw(callback.get() as *const RawClosureHeader) + .call1(error.get()); + } + } + } + ZlibEvent::Callback(callback) => { + if callback != 0 { + let _ = JsClosure::from_raw(callback as *const RawClosureHeader).call0(); + } + } + ZlibEvent::OneShotCallback(_, _, _) => { + unreachable!("one-shot zlib events use the two-phase provider path") + } + } + f64::from_bits(UNDEFINED) +} + +unsafe extern "C" fn zlib_empty_phase_thunk(_data: *mut c_void) -> f64 { + f64::from_bits(UNDEFINED) +} + +struct ZlibOneShotDispatch { + callback: TransientRootedAddr, + result: Option, String>>, +} + +unsafe extern "C" fn zlib_one_shot_dispatch_thunk(data: *mut c_void) -> f64 { + let call = &mut *(data as *mut ZlibOneShotDispatch); + call_one_shot_callback( + call.callback.get(), + call.result + .take() + .expect("zlib one-shot dispatch thunk must run exactly once"), + ); + f64::from_bits(UNDEFINED) +} + /// Drain queued zlib stream events on the main thread. Wired into perry-stdlib's /// `js_stdlib_process_pending` via the external-zlib-pump feature. #[no_mangle] @@ -1345,133 +1480,78 @@ pub unsafe extern "C" fn js_ext_zlib_process_pending() -> i32 { }; count += 1; let event_async_id = event_stream_handle(&ev).map(stream_async_id).unwrap_or(0); - let mut destroy_after_dispatch = 0; - if event_async_id != 0 { - js_async_hooks_provider_enter(event_async_id); - } - match ev { - ZlibEvent::Data(id, bytes) => { - publish_bytes_written(id); - let cbs = listeners_for(id, "data"); - let dests = pipes_for(id); - if cbs.is_empty() && dests.is_empty() { - // No consumer attached yet — buffer instead of dropping, so a - // `.on('data')`/`.pipe()` that attaches later (after `await`) - // still receives the body (flushed by `flush_buffered`), - // bounded by the per-stream + global byte caps. - buffer_output_for_late_consumer(&mut statics().lock().unwrap(), id, &bytes); - } else { - if !cbs.is_empty() { - if let Some(buf_f64) = make_buffer_f64(&bytes) { - for cb in cbs { - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader) - .call1(buf_f64); - } - } - } - } - for dest in dests { - forward_write(dest, &bytes); - } - } - } - ZlibEvent::Finish(id) => { - let scope = TransientRootScope::enter(); - let callbacks = scope.root_addrs(&listeners_for(id, "finish")); - for cb in callbacks { - let cb = cb.get(); - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader).call0(); - } - } - } - ZlibEvent::End(id) => { - publish_bytes_written(id); - // Defer `'end'` (keep the stream + its buffer alive) when no - // consumer has attached yet — otherwise removing the stream here - // would strand a `.on('data')`/`.on('end')` that attaches later - // (gaxios attaches them only after `await`ing the fetch), hanging - // the body-consume. `flush_buffered` re-queues End once a - // consumer attaches and the buffer has drained. - let has_consumer = - !listeners_for(id, "data").is_empty() || !pipes_for(id).is_empty(); - if !has_consumer { - let mut g = statics().lock().unwrap(); - let deferred = match g.streams.get_mut(&id) { - Some(s) => { - s.end_buffered = true; - true - } - None => false, - }; - if deferred { - // Cap how many never-consumed ended streams we retain so - // an abandoned handle (one that never gets a `'data'` - // listener or pipe) can't pin its buffered output for the - // process lifetime; drop the oldest excess. - evict_excess_buffered_ended(&mut g); - if event_async_id != 0 { - js_async_hooks_provider_leave(event_async_id); - } - continue; + if let ZlibEvent::End(id) = &ev { + // Defer `'end'` (keep the stream + its buffer alive) when no + // consumer has attached yet. Do this before entering the provider + // so a deferred stream does not emit a lifecycle phase prematurely. + let has_consumer = !listeners_for(*id, "data").is_empty() || !pipes_for(*id).is_empty(); + if !has_consumer { + let mut g = statics().lock().unwrap(); + let deferred = match g.streams.get_mut(id) { + Some(s) => { + s.end_buffered = true; + true } - // Stream already gone — release the lock and fall through to - // the (no-op) delivery + removal below. - drop(g); - } - for cb in listeners_for(id, "end") { - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader).call0(); - } - } - for dest in pipes_for(id) { - forward_end(dest); - } - for cb in listeners_for(id, "close") { - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader).call0(); - } - } - drop_buffered_stream(&mut statics().lock().unwrap(), id); - destroy_after_dispatch = event_async_id; - } - ZlibEvent::Callback(cb) => { - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader).call0(); + None => false, + }; + if deferred { + // Cap how many never-consumed ended streams we retain so + // an abandoned handle (one that never gets a `'data'` + // listener or pipe) can't pin its buffered output for the + // process lifetime; drop the oldest excess. + evict_excess_buffered_ended(&mut g); + continue; } + // Stream already gone — release the lock and fall through to + // the (no-op) delivery + removal below. + drop(g); } - ZlibEvent::OneShotCallback(cb, result, async_id) => { + } + + let ev = match ev { + ZlibEvent::OneShotCallback(callback, result, async_id) => { let scope = TransientRootScope::enter(); - let callback = scope.root_addr(cb); + let callback = scope.root_addr(callback); // Node exposes the native codec completion and delivery of the // JavaScript callback as two phases of the same ZLIB resource. - js_async_hooks_provider_enter(async_id); - js_async_hooks_provider_leave(async_id); - js_async_hooks_provider_enter(async_id); - call_one_shot_callback(callback.get(), result); - js_async_hooks_provider_leave(async_id); - // This is queued before the callback's Promise continuation - // schedules its first user immediate, so zlib needs one more - // check turn than synchronously closed handles. - js_async_hooks_provider_defer_destroy(async_id, 4); - } - ZlibEvent::Error(id, msg) => { - let err_f64 = build_error_object(&msg); - for cb in listeners_for(id, "error") { - if cb != 0 { - let _ = JsClosure::from_raw(cb as *const RawClosureHeader).call1(err_f64); - } - } - drop_buffered_stream(&mut statics().lock().unwrap(), id); - destroy_after_dispatch = event_async_id; + js_async_hooks_provider_run_catching_deferred_destroy_on_error( + async_id, + 4, + zlib_empty_phase_thunk, + std::ptr::null_mut(), + ); + let mut call = ZlibOneShotDispatch { + callback, + result: Some(result), + }; + js_async_hooks_provider_run_catching_deferred_destroy( + async_id, + 4, + zlib_one_shot_dispatch_thunk, + &mut call as *mut ZlibOneShotDispatch as *mut c_void, + ); + continue; } - } - if event_async_id != 0 { - js_async_hooks_provider_leave(event_async_id); - } - if destroy_after_dispatch != 0 { - js_async_hooks_provider_defer_destroy(destroy_after_dispatch, 4); + event => event, + }; + + let terminal = matches!(&ev, ZlibEvent::End(_) | ZlibEvent::Error(_, _)); + let mut call = ZlibEventDispatch { event: Some(ev) }; + if event_async_id == 0 { + zlib_event_dispatch_thunk(&mut call as *mut ZlibEventDispatch as *mut c_void); + } else if terminal { + js_async_hooks_provider_run_catching_deferred_destroy( + event_async_id, + 4, + zlib_event_dispatch_thunk, + &mut call as *mut ZlibEventDispatch as *mut c_void, + ); + } else { + js_async_hooks_provider_run_catching( + event_async_id, + zlib_event_dispatch_thunk, + &mut call as *mut ZlibEventDispatch as *mut c_void, + ); } } count diff --git a/crates/perry-runtime/src/async_hooks.rs b/crates/perry-runtime/src/async_hooks.rs index 961190f9c1..e73c8985e1 100644 --- a/crates/perry-runtime/src/async_hooks.rs +++ b/crates/perry-runtime/src/async_hooks.rs @@ -24,7 +24,15 @@ pub use provider_ffi::{ defer_destroy_after_check_turns, js_async_hooks_provider_defer_destroy, js_async_hooks_provider_destroy, js_async_hooks_provider_enter, js_async_hooks_provider_init, js_async_hooks_provider_init_with_trigger, js_async_hooks_provider_leave, - js_async_hooks_provider_run_catching, js_async_hooks_provider_run_catching_with_this, + js_async_hooks_provider_run_catching, js_async_hooks_provider_run_catching_deferred_destroy, + js_async_hooks_provider_run_catching_deferred_destroy_on_error, + js_async_hooks_provider_run_catching_with_this, +}; +mod scopes; +pub use scopes::{ + enter_resource_scope, leave_resource_scope, run_provider_completion, run_resource_scope, + run_resource_scope_catching, try_enter_resource_scope, try_leave_resource_scope, + try_run_resource_scope, }; const POINTER_TAG: u64 = 0x7FFD_0000_0000_0000; @@ -58,6 +66,8 @@ per_test_global! { pub static HOOKS_ACTIVE: AtomicUsize = AtomicUsize::new(0); static PROMISE_HOOKS_ACTIVE: AtomicUsize = AtomicUsize::new(0); static TOP_LEVEL_RESOURCE: AtomicU64 = AtomicU64::new(0); + #[cfg(test)] + static TEST_FORCE_RESOLVE_GC: AtomicUsize = AtomicUsize::new(0); } #[derive(Clone, Copy)] @@ -211,11 +221,18 @@ pub(crate) fn resolve_async_resource_handle(receiver: i64) -> Option { if !crate::value::addr_class::is_plausible_heap_addr(raw) { return None; } + let scope = crate::gc::RuntimeHandleScope::new(); + let receiver = scope.root_raw_mut_ptr(raw as *mut ObjectHeader); + #[cfg(test)] + if TEST_FORCE_RESOLVE_GC.swap(0, Ordering::Relaxed) != 0 { + let _ = crate::gc::gc_collect_minor(); + } let key = js_string_from_bytes( ASYNC_RESOURCE_SUBCLASS_KEY.as_ptr(), ASYNC_RESOURCE_SUBCLASS_KEY.len() as u32, ); - let value = js_object_get_field_by_name(raw as *const ObjectHeader, key); + let value = receiver + .with_mut_ptr::(|receiver| js_object_get_field_by_name(receiver, key)); if !value.is_pointer() { return None; } @@ -223,6 +240,28 @@ pub(crate) fn resolve_async_resource_handle(receiver: i64) -> Option { is_async_resource_handle(backing).then_some(backing) } +#[cfg(test)] +pub(crate) fn test_force_next_async_resource_resolve_gc() { + TEST_FORCE_RESOLVE_GC.store(1, Ordering::Relaxed); +} + +#[cfg(test)] +pub(crate) fn test_link_async_resource_subclass(receiver: *mut ObjectHeader, backing: i64) { + let scope = crate::gc::RuntimeHandleScope::new(); + let receiver = scope.root_raw_mut_ptr(receiver); + let key = js_string_from_bytes( + ASYNC_RESOURCE_SUBCLASS_KEY.as_ptr(), + ASYNC_RESOURCE_SUBCLASS_KEY.len() as u32, + ); + receiver.with_mut_ptr::(|receiver| { + crate::object::js_object_set_field_by_name( + receiver, + key, + crate::value::js_nanbox_pointer(backing), + ); + }); +} + #[inline(always)] pub fn hooks_active() -> bool { HOOKS_ACTIVE.load(Ordering::Relaxed) != 0 @@ -873,63 +912,6 @@ pub fn destroy_promise(async_id: u64) { destroy_with_kind(async_id, true); } -/// Run a synchronous native completion as an observable async-hooks provider. -/// The operation may already have done its blocking work eagerly, but its -/// Promise settlement still needs the same provider execution/resource scope -/// Node gives a libuv completion. The returned JS value is rooted across hook -/// callbacks, which are arbitrary allocating user code. -pub fn run_provider_completion(type_name: &'static str, completion: impl FnOnce() -> f64) -> f64 { - let scope = crate::gc::RuntimeHandleScope::new(); - let resource = crate::object::js_object_alloc_null_proto(0, 0); - let resource_handle = scope.root_raw_mut_ptr(resource); - let ids = resource_handle.with_mut_ptr::(|resource| { - init_resource( - type_name, - crate::value::js_nanbox_pointer(resource as i64), - true, - ) - }); - before(ids.async_id, ids.trigger_async_id); - let result = scope.root_nanbox_f64(completion()); - after(ids.async_id); - destroy(ids.async_id); - result.get_nanbox_f64() -} - -/// Enter an existing provider's captured AsyncLocalStorage and execution-id -/// scope for one native callback phase. -pub fn enter_resource_scope(ids: AsyncResourceIds) { - let context = RESOURCES - .lock() - .unwrap() - .get(&ids.async_id) - .map(|meta| meta.context.clone()) - .unwrap_or_default(); - let previous = crate::async_context::enter_context(&context); - crate::async_context::push_context_guard( - crate::async_context::ContextGuardAction::RestoreSnapshot(previous), - ); - before(ids.async_id, ids.trigger_async_id); - crate::async_context::push_context_guard( - crate::async_context::ContextGuardAction::RestoreExecutionIds, - ); -} - -/// Leave a provider scope entered by [`enter_resource_scope`]. -pub fn leave_resource_scope(async_id: u64) { - let _ = crate::async_context::pop_context_guard(); - after(async_id); - if let Some(action) = crate::async_context::pop_context_guard() { - crate::async_context::apply_context_guard(action); - } -} - -pub fn run_resource_scope(ids: AsyncResourceIds, completion: impl FnOnce()) { - enter_resource_scope(ids); - completion(); - leave_resource_scope(ids.async_id); -} - pub fn enqueue_gc_destroy(async_id: u64) { if async_id != 0 { GC_DESTROY_QUEUE.lock().unwrap().push_back(async_id); @@ -1269,13 +1251,15 @@ pub extern "C" fn js_async_resource_subclass_init( options_handle.get_nanbox_f64(), Some(this_handle.get_nanbox_f64()), ); - let current_this = this_handle.get_nanbox_f64(); - let raw = crate::value::js_nanbox_get_pointer(current_this) as *mut ObjectHeader; + let raw = + crate::value::js_nanbox_get_pointer(this_handle.get_nanbox_f64()) as *mut ObjectHeader; if !raw.is_null() && crate::value::addr_class::is_plausible_heap_addr(raw as usize) { let key = js_string_from_bytes( ASYNC_RESOURCE_SUBCLASS_KEY.as_ptr(), ASYNC_RESOURCE_SUBCLASS_KEY.len() as u32, ); + let raw = + crate::value::js_nanbox_get_pointer(this_handle.get_nanbox_f64()) as *mut ObjectHeader; crate::object::js_object_set_field_by_name( raw, key, @@ -1341,23 +1325,31 @@ extern "C" fn async_resource_bind_method_trampoline( return TAG_UNDEFINED_F64; } let handle = js_closure_get_capture_ptr(closure, 0); - let args_array = crate::value::js_nanbox_get_pointer(rest) as *const ArrayHeader; - let args_len = if args_array.is_null() { - 0 - } else { - js_array_length(args_array) - }; - let callback = if args_len == 0 { - TAG_UNDEFINED_F64 - } else { - crate::array::js_array_get_f64(args_array, 0) - }; - let this_arg = if args_len < 2 { - TAG_UNDEFINED_F64 - } else { - crate::array::js_array_get_f64(args_array, 1) - }; - let bound = js_async_resource_bind(handle, callback, this_arg); + let scope = crate::gc::RuntimeHandleScope::new(); + let args_array = + scope.root_raw_const_ptr(crate::value::js_nanbox_get_pointer(rest) as *const ArrayHeader); + let (callback, this_arg) = args_array.with_const_ptr::(|args_array| { + let args_len = if args_array.is_null() { + 0 + } else { + js_array_length(args_array) + }; + let callback = if args_len == 0 { + TAG_UNDEFINED_F64 + } else { + crate::array::js_array_get_f64(args_array, 0) + }; + let this_arg = if args_len < 2 { + TAG_UNDEFINED_F64 + } else { + crate::array::js_array_get_f64(args_array, 1) + }; + (callback, this_arg) + }); + let callback = scope.root_nanbox_f64(callback); + let this_arg = scope.root_nanbox_f64(this_arg); + let bound = + js_async_resource_bind(handle, callback.get_nanbox_f64(), this_arg.get_nanbox_f64()); if bound == 0 { TAG_UNDEFINED_F64 } else { @@ -1443,25 +1435,33 @@ pub fn try_async_resource_method_dispatch( unsafe { std::slice::from_raw_parts(args_ptr, args_len).to_vec() } }; let arg_handles = scope.root_nanbox_f64_slice(&raw_args); - let handle = resolve_async_resource_handle(receiver)?; - let args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles); + let receiver = scope.root_raw_mut_ptr(receiver as *mut ObjectHeader); + let handle = receiver.with_mut_ptr::(|receiver| { + resolve_async_resource_handle(receiver as i64) + })?; Some(match method_name { "asyncId" => js_async_resource_async_id(handle), "triggerAsyncId" => js_async_resource_trigger_async_id(handle), "emitDestroy" => { - js_async_resource_emit_destroy(handle); - crate::value::js_nanbox_pointer(receiver) + let (_, receiver) = + receiver.across_mut::(|| js_async_resource_emit_destroy(handle)); + crate::value::js_nanbox_pointer(receiver as i64) } "runInAsyncScope" => { // runInAsyncScope(fn[, thisArg, ...args]) - let callback = args.first().copied().unwrap_or(TAG_UNDEFINED_F64); - let this_arg = args.get(1).copied().unwrap_or(TAG_UNDEFINED_F64); + let args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles); let rest = if args.len() > 2 { &args[2..] } else { &[] }; let args_array = pack_rest_args_array(rest); + // Packing the rest array may collect, so refresh callback and + // thisArg from their roots before dispatching the call. + let args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles); + let callback = args.first().copied().unwrap_or(TAG_UNDEFINED_F64); + let this_arg = args.get(1).copied().unwrap_or(TAG_UNDEFINED_F64); js_async_resource_run_in_async_scope(handle, callback, this_arg, args_array) } "bind" => { // bind(fn[, thisArg]) + let args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles); let callback = args.first().copied().unwrap_or(TAG_UNDEFINED_F64); let this_arg = args.get(1).copied().unwrap_or(TAG_UNDEFINED_F64); let bound = js_async_resource_bind(handle, callback, this_arg); @@ -1527,83 +1527,56 @@ pub extern "C" fn js_async_resource_run_in_async_scope( this_arg: f64, args_array: i64, ) -> f64 { - let Some(handle) = resolve_async_resource_handle(handle) else { - return TAG_UNDEFINED_F64; - }; - if !is_callable_value(callback_value) { - throw_apply_not_function(callback_value); - } let scope = crate::gc::RuntimeHandleScope::new(); + let receiver_handle = scope.root_raw_mut_ptr(handle as *mut ObjectHeader); let callback_handle = scope.root_nanbox_f64(callback_value); let this_arg_handle = scope.root_nanbox_f64(this_arg); + let args_array_handle = scope.root_raw_const_ptr(args_array as *const ArrayHeader); + let Some(handle) = receiver_handle + .with_mut_ptr::(|receiver| resolve_async_resource_handle(receiver as i64)) + else { + return TAG_UNDEFINED_F64; + }; + if !is_callable_value(callback_handle.get_nanbox_f64()) { + throw_apply_not_function(callback_handle.get_nanbox_f64()); + } + let ids = unsafe { (*(handle as *const AsyncResourceHandle)).ids }; let rebound_bits = crate::closure::clone_closure_rebind_this( callback_handle.get_nanbox_f64().to_bits(), this_arg_handle.get_nanbox_f64(), ); let rebound_handle = scope.root_nanbox_f64(f64::from_bits(rebound_bits)); - let callback = crate::fs::extract_closure_ptr(rebound_handle.get_nanbox_f64()); - if callback.is_null() { + if crate::fs::extract_closure_ptr(rebound_handle.get_nanbox_f64()).is_null() { throw_apply_not_function(callback_handle.get_nanbox_f64()); } - let args_array_handle = scope.root_raw_const_ptr(args_array as *const ArrayHeader); - let resource = unsafe { &*(handle as *const AsyncResourceHandle) }; - let resource_context = RESOURCES - .lock() - .unwrap() - .get(&resource.ids.async_id) - .map(|meta| meta.context.clone()) - .unwrap_or_default(); - let mut resource_context = resource_context; - let resource_context_roots = crate::async_context::root_snapshot(&scope, &resource_context); - let previous = crate::async_context::enter_context(&resource_context); - // The guard owns the previous snapshot: it is GC-scanned while held, and - // if the callback throws, `js_throw` restores it during unwind (#788). - crate::async_context::push_context_guard( - crate::async_context::ContextGuardAction::RestoreSnapshot(previous), - ); - before(resource.ids.async_id, resource.ids.trigger_async_id); - crate::async_context::push_context_guard( - crate::async_context::ContextGuardAction::RestoreExecutionIds, - ); - let prev_this = crate::object::js_implicit_this_set(this_arg_handle.get_nanbox_f64()); - // Catch locally so a throwing scope still delivers `after` and restores - // the resource/context before the exception is rethrown to user code. - // The trap is installed after our guards, so throw-time unwinding leaves - // those guards for the normal cleanup below. - let outcome = crate::exception::js_call_catching(|| { - if args_array == 0 { - unsafe { js_closure_call_array(callback as i64, ptr::null(), 0) } - } else { - let arr = args_array_handle.get_raw_const_ptr::(); - let len = js_array_length(arr) as i64; - let data = if arr.is_null() { - ptr::null() - } else { - unsafe { (arr as *const u8).add(std::mem::size_of::()) as *const f64 } - }; - unsafe { js_closure_call_array(callback as i64, data, len) } + let outcome = try_run_resource_scope(ids, || { + let callback = crate::fs::extract_closure_ptr(rebound_handle.get_nanbox_f64()); + let previous_this = scope.root_nanbox_f64(crate::object::js_implicit_this_set( + this_arg_handle.get_nanbox_f64(), + )); + let callback_outcome = crate::exception::js_call_catching(|| { + args_array_handle.with_const_ptr::(|arr| { + if arr.is_null() { + unsafe { js_closure_call_array(callback as i64, ptr::null(), 0) } + } else { + let len = js_array_length(arr) as i64; + let data = unsafe { + (arr as *const u8).add(std::mem::size_of::()) as *const f64 + }; + unsafe { js_closure_call_array(callback as i64, data, len) } + } + }) + }); + crate::object::js_implicit_this_set(previous_this.get_nanbox_f64()); + match callback_outcome { + Ok(value) => value, + Err(error) => crate::exception::js_throw(error), } }); - crate::object::js_implicit_this_set(prev_this); - let threw = outcome.is_err(); - let result_handle = scope.root_nanbox_f64(match outcome { - Ok(result) | Err(result) => result, - }); - // Normal exit: `after` fires hooks and pops the execution scope itself, - // so discard the silent-unwind guard rather than applying it. - let _ = crate::async_context::pop_context_guard(); - after(resource.ids.async_id); - crate::async_context::refresh_snapshot_from_roots( - &mut resource_context, - &resource_context_roots, - ); - if let Some(action) = crate::async_context::pop_context_guard() { - crate::async_context::apply_context_guard(action); - } - if threw { - crate::exception::js_throw(result_handle.get_nanbox_f64()); + match outcome { + Ok(value) => value, + Err(error) => crate::exception::js_throw(error), } - result_handle.get_nanbox_f64() } /// Trampoline body for `AsyncResource#bind`. Stored as the `func_ptr` of the @@ -1645,14 +1618,17 @@ fn register_bind_trampoline_once() { #[no_mangle] pub extern "C" fn js_async_resource_bind(handle: i64, callback_value: f64, this_arg: f64) -> i64 { - validate_bind_callback(callback_value); - let Some(handle) = resolve_async_resource_handle(handle) else { - return 0; - }; - register_bind_trampoline_once(); let scope = crate::gc::RuntimeHandleScope::new(); + let receiver_handle = scope.root_raw_mut_ptr(handle as *mut ObjectHeader); let callback_handle = scope.root_nanbox_f64(callback_value); let this_arg_handle = scope.root_nanbox_f64(this_arg); + validate_bind_callback(callback_handle.get_nanbox_f64()); + let Some(handle) = receiver_handle + .with_mut_ptr::(|receiver| resolve_async_resource_handle(receiver as i64)) + else { + return 0; + }; + register_bind_trampoline_once(); let closure = js_closure_alloc(async_resource_bind_trampoline as *const u8, 3); if closure.is_null() { return 0; @@ -1669,9 +1645,9 @@ pub extern "C" fn js_async_resource_bind(handle: i64, callback_value: f64, this_ 2, this_arg_handle.get_nanbox_f64(), ); - if let Some(length) = - crate::closure::closure_length(crate::fs::extract_closure_ptr(callback_value)) - { + if let Some(length) = crate::closure::closure_length(crate::fs::extract_closure_ptr( + callback_handle.get_nanbox_f64(), + )) { crate::object::set_builtin_closure_length( closure_handle.get_raw_mut_ptr::() as usize, length, diff --git a/crates/perry-runtime/src/async_hooks/provider_ffi.rs b/crates/perry-runtime/src/async_hooks/provider_ffi.rs index b50e7f3657..7570dc316a 100644 --- a/crates/perry-runtime/src/async_hooks/provider_ffi.rs +++ b/crates/perry-runtime/src/async_hooks/provider_ffi.rs @@ -1,8 +1,8 @@ //! Exception-safe callback bridges for separately linked async providers. use super::{ - destroy, enter_resource_scope, init_resource, init_resource_with_trigger, leave_resource_scope, - AsyncResourceIds, RESOURCES, + destroy, init_resource, init_resource_with_trigger, try_enter_resource_scope, + try_leave_resource_scope, AsyncResourceIds, RESOURCES, }; extern "C" fn deferred_destroy_step(closure: *const crate::closure::ClosureHeader) -> f64 { @@ -45,6 +45,19 @@ pub fn defer_destroy_after_check_turns(async_id: u64, check_turns: u32) { } } +fn provider_ids(async_id: u64) -> AsyncResourceIds { + let trigger_async_id = RESOURCES + .lock() + .unwrap() + .get(&async_id) + .map(|meta| meta.trigger_async_id) + .unwrap_or(0); + AsyncResourceIds { + async_id, + trigger_async_id, + } +} + /// C ABI used by separately-linked native providers such as perry-ext-zlib. #[no_mangle] pub unsafe extern "C" fn js_async_hooks_provider_init(type_ptr: *const u8, type_len: usize) -> u64 { @@ -83,21 +96,16 @@ pub unsafe extern "C" fn js_async_hooks_provider_init_with_trigger( #[no_mangle] pub extern "C" fn js_async_hooks_provider_enter(async_id: u64) { - let trigger_async_id = RESOURCES - .lock() - .unwrap() - .get(&async_id) - .map(|meta| meta.trigger_async_id) - .unwrap_or(0); - enter_resource_scope(AsyncResourceIds { - async_id, - trigger_async_id, - }); + if let Err(error) = try_enter_resource_scope(provider_ids(async_id)) { + crate::exception::js_throw(error); + } } #[no_mangle] pub extern "C" fn js_async_hooks_provider_leave(async_id: u64) { - leave_resource_scope(async_id); + if let Err(error) = try_leave_resource_scope(async_id) { + crate::exception::js_throw(error); + } } #[no_mangle] @@ -120,17 +128,87 @@ pub unsafe extern "C" fn js_async_hooks_provider_run_catching( callback: unsafe extern "C" fn(*mut std::ffi::c_void) -> f64, data: *mut std::ffi::c_void, ) -> f64 { - js_async_hooks_provider_enter(async_id); - let outcome = crate::exception::js_call_catching(|| callback(data)); + provider_run_catching(async_id, DestroyPolicy::Never, callback, data) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum DestroyPolicy { + Never, + Always(u32), + OnError(u32), +} + +/// Variant used by terminal external-provider events. Teardown is scheduled +/// after scope restoration and before a caught JavaScript exception is +/// rethrown, so a throwing listener cannot strand the resource. +#[no_mangle] +pub unsafe extern "C" fn js_async_hooks_provider_run_catching_deferred_destroy( + async_id: u64, + check_turns: u32, + callback: unsafe extern "C" fn(*mut std::ffi::c_void) -> f64, + data: *mut std::ffi::c_void, +) -> f64 { + provider_run_catching(async_id, DestroyPolicy::Always(check_turns), callback, data) +} + +/// Schedule terminal teardown only when scope entry, the callback, or scope +/// exit throws. This lets a multi-phase provider protect an early phase while +/// leaving its normal destroy timing to the final phase. +#[no_mangle] +pub unsafe extern "C" fn js_async_hooks_provider_run_catching_deferred_destroy_on_error( + async_id: u64, + check_turns: u32, + callback: unsafe extern "C" fn(*mut std::ffi::c_void) -> f64, + data: *mut std::ffi::c_void, +) -> f64 { + provider_run_catching( + async_id, + DestroyPolicy::OnError(check_turns), + callback, + data, + ) +} + +unsafe fn provider_run_catching( + async_id: u64, + destroy_policy: DestroyPolicy, + callback: unsafe extern "C" fn(*mut std::ffi::c_void) -> f64, + data: *mut std::ffi::c_void, +) -> f64 { let scope = crate::gc::RuntimeHandleScope::new(); + if let Err(error) = try_enter_resource_scope(provider_ids(async_id)) { + let error = scope.root_nanbox_f64(error); + if let DestroyPolicy::Always(turns) | DestroyPolicy::OnError(turns) = destroy_policy { + defer_destroy_after_check_turns(async_id, turns); + } + crate::exception::js_throw(error.get_nanbox_f64()); + } + let outcome = crate::exception::js_call_catching(|| callback(data)); let (threw, result) = match outcome { Ok(value) => (false, scope.root_nanbox_f64(value)), Err(error) => (true, scope.root_nanbox_f64(error)), }; - js_async_hooks_provider_leave(async_id); + let leave = try_leave_resource_scope(async_id); + let (leave_threw, leave_result) = match leave { + Ok(()) => ( + false, + scope.root_nanbox_f64(f64::from_bits(crate::value::TAG_UNDEFINED)), + ), + Err(error) => (true, scope.root_nanbox_f64(error)), + }; + if let DestroyPolicy::Always(turns) = destroy_policy { + defer_destroy_after_check_turns(async_id, turns); + } else if let DestroyPolicy::OnError(turns) = destroy_policy { + if threw || leave_threw { + defer_destroy_after_check_turns(async_id, turns); + } + } if threw { crate::exception::js_throw(result.get_nanbox_f64()); } + if leave_threw { + crate::exception::js_throw(leave_result.get_nanbox_f64()); + } result.get_nanbox_f64() } @@ -147,7 +225,16 @@ pub unsafe extern "C" fn js_async_hooks_provider_run_catching_with_this( ) -> f64 { let scope = crate::gc::RuntimeHandleScope::new(); let this_value = scope.root_nanbox_f64(this_value); - js_async_hooks_provider_enter(async_id); + if let Err(error) = try_enter_resource_scope(provider_ids(async_id)) { + let error = scope.root_nanbox_f64(error); + if destroy_after != 0 { + let _ = crate::exception::js_call_catching(|| { + destroy(async_id); + f64::from_bits(crate::value::TAG_UNDEFINED) + }); + } + crate::exception::js_throw(error.get_nanbox_f64()); + } let previous_this = scope.root_nanbox_f64(crate::object::js_implicit_this_set( this_value.get_nanbox_f64(), )); @@ -157,12 +244,35 @@ pub unsafe extern "C" fn js_async_hooks_provider_run_catching_with_this( Err(error) => (true, scope.root_nanbox_f64(error)), }; crate::object::js_implicit_this_set(previous_this.get_nanbox_f64()); - js_async_hooks_provider_leave(async_id); - if destroy_after != 0 { - js_async_hooks_provider_destroy(async_id); - } + let leave = try_leave_resource_scope(async_id); + let (leave_threw, leave_result) = match leave { + Ok(()) => ( + false, + scope.root_nanbox_f64(f64::from_bits(crate::value::TAG_UNDEFINED)), + ), + Err(error) => (true, scope.root_nanbox_f64(error)), + }; + let destroy_outcome = (destroy_after != 0).then(|| { + crate::exception::js_call_catching(|| { + destroy(async_id); + f64::from_bits(crate::value::TAG_UNDEFINED) + }) + }); + let (destroy_threw, destroy_result) = match destroy_outcome { + Some(Err(error)) => (true, scope.root_nanbox_f64(error)), + _ => ( + false, + scope.root_nanbox_f64(f64::from_bits(crate::value::TAG_UNDEFINED)), + ), + }; if threw { crate::exception::js_throw(result.get_nanbox_f64()); } + if leave_threw { + crate::exception::js_throw(leave_result.get_nanbox_f64()); + } + if destroy_threw { + crate::exception::js_throw(destroy_result.get_nanbox_f64()); + } result.get_nanbox_f64() } diff --git a/crates/perry-runtime/src/async_hooks/scopes.rs b/crates/perry-runtime/src/async_hooks/scopes.rs new file mode 100644 index 0000000000..f5e046dd6c --- /dev/null +++ b/crates/perry-runtime/src/async_hooks/scopes.rs @@ -0,0 +1,152 @@ +//! Exception-safe entry and cleanup for async-resource execution scopes. + +use super::{after, before, destroy, init_resource, AsyncResourceIds, RESOURCES}; + +const TAG_UNDEFINED_F64: f64 = f64::from_bits(crate::value::TAG_UNDEFINED); + +/// Run a synchronous native completion as an observable async-hooks provider. +/// The returned value stays rooted while arbitrary JavaScript hooks run. +pub fn run_provider_completion(type_name: &'static str, completion: impl FnOnce() -> f64) -> f64 { + let scope = crate::gc::RuntimeHandleScope::new(); + let resource = crate::object::js_object_alloc_null_proto(0, 0); + let resource_handle = scope.root_raw_mut_ptr(resource); + let ids = resource_handle.with_mut_ptr::(|resource| { + init_resource( + type_name, + crate::value::js_nanbox_pointer(resource as i64), + true, + ) + }); + let outcome = try_run_resource_scope(ids, completion); + let (threw, result) = match outcome { + Ok(value) => (false, scope.root_nanbox_f64(value)), + Err(error) => (true, scope.root_nanbox_f64(error)), + }; + let destroy_outcome = crate::exception::js_call_catching(|| { + destroy(ids.async_id); + TAG_UNDEFINED_F64 + }); + let destroy_error = destroy_outcome + .err() + .map(|error| scope.root_nanbox_f64(error)); + if threw { + crate::exception::js_throw(result.get_nanbox_f64()); + } + if let Some(error) = destroy_error { + crate::exception::js_throw(error.get_nanbox_f64()); + } + result.get_nanbox_f64() +} + +/// Enter an existing provider's captured AsyncLocalStorage and execution-id +/// scope for one native callback phase. +pub fn try_enter_resource_scope(ids: AsyncResourceIds) -> Result<(), f64> { + let context = RESOURCES + .lock() + .unwrap() + .get(&ids.async_id) + .map(|meta| meta.context.clone()) + .unwrap_or_default(); + let previous = crate::async_context::enter_context(&context); + crate::async_context::push_context_guard( + crate::async_context::ContextGuardAction::RestoreSnapshot(previous), + ); + crate::async_context::push_context_guard( + crate::async_context::ContextGuardAction::RestoreExecutionIds, + ); + let outcome = crate::exception::js_call_catching(|| { + before(ids.async_id, ids.trigger_async_id); + TAG_UNDEFINED_F64 + }); + if let Err(error) = outcome { + let scope = crate::gc::RuntimeHandleScope::new(); + let error = scope.root_nanbox_f64(error); + if let Some(action) = crate::async_context::pop_context_guard() { + crate::async_context::apply_context_guard(action); + } + if let Some(action) = crate::async_context::pop_context_guard() { + crate::async_context::apply_context_guard(action); + } + return Err(error.get_nanbox_f64()); + } + Ok(()) +} + +pub fn enter_resource_scope(ids: AsyncResourceIds) { + if let Err(error) = try_enter_resource_scope(ids) { + crate::exception::js_throw(error); + } +} + +/// Leave a provider scope entered by [`enter_resource_scope`]. +pub fn try_leave_resource_scope(async_id: u64) -> Result<(), f64> { + let outcome = crate::exception::js_call_catching(|| { + after(async_id); + TAG_UNDEFINED_F64 + }); + let scope = crate::gc::RuntimeHandleScope::new(); + let (threw, result) = match outcome { + Ok(value) => (false, scope.root_nanbox_f64(value)), + Err(error) => (true, scope.root_nanbox_f64(error)), + }; + if let Some(action) = crate::async_context::pop_context_guard() { + if threw { + crate::async_context::apply_context_guard(action); + } + } + if let Some(action) = crate::async_context::pop_context_guard() { + crate::async_context::apply_context_guard(action); + } + if threw { + Err(result.get_nanbox_f64()) + } else { + Ok(()) + } +} + +pub fn leave_resource_scope(async_id: u64) { + if let Err(error) = try_leave_resource_scope(async_id) { + crate::exception::js_throw(error); + } +} + +pub fn run_resource_scope(ids: AsyncResourceIds, completion: impl FnOnce()) { + let _ = run_resource_scope_catching(ids, || { + completion(); + TAG_UNDEFINED_F64 + }); +} + +/// Execute user code inside an existing provider and return its exception only +/// after the provider context and execution-id stacks have been restored. +pub fn try_run_resource_scope( + ids: AsyncResourceIds, + completion: impl FnOnce() -> f64, +) -> Result { + try_enter_resource_scope(ids)?; + let scope = crate::gc::RuntimeHandleScope::new(); + let outcome = crate::exception::js_call_catching(completion); + let (threw, result) = match outcome { + Ok(value) => (false, scope.root_nanbox_f64(value)), + Err(error) => (true, scope.root_nanbox_f64(error)), + }; + if let Err(error) = try_leave_resource_scope(ids.async_id) { + if threw { + return Err(result.get_nanbox_f64()); + } + let error = scope.root_nanbox_f64(error); + return Err(error.get_nanbox_f64()); + } + if threw { + Err(result.get_nanbox_f64()) + } else { + Ok(result.get_nanbox_f64()) + } +} + +pub fn run_resource_scope_catching(ids: AsyncResourceIds, completion: impl FnOnce() -> f64) -> f64 { + match try_run_resource_scope(ids, completion) { + Ok(value) => value, + Err(error) => crate::exception::js_throw(error), + } +} diff --git a/crates/perry-runtime/src/async_hooks/test_support.rs b/crates/perry-runtime/src/async_hooks/test_support.rs index be53907020..18efb29c0a 100644 --- a/crates/perry-runtime/src/async_hooks/test_support.rs +++ b/crates/perry-runtime/src/async_hooks/test_support.rs @@ -63,6 +63,26 @@ pub(crate) fn test_async_hooks_scanner_snapshot() -> (usize, u64) { mod tests { use super::*; + extern "C" fn throwing_lifecycle_hook(_closure: *const ClosureHeader, _async_id: f64) -> f64 { + crate::exception::js_throw(73.0) + } + + fn enable_throwing_lifecycle_hook(before_phase: bool) { + let callback = js_closure_alloc(throwing_lifecycle_hook as *const u8, 0); + let mut callbacks = HookCallbacks::empty(); + if before_phase { + callbacks.before = callback; + } else { + callbacks.after = callback; + } + HOOKS.lock().unwrap().push(HookRecord { + callbacks, + enabled: true, + track_promises: false, + }); + HOOKS_ACTIVE.store(1, Ordering::Relaxed); + } + // #7680: no lock needed here anymore. The per-test globals isolate this // thread's reset and resource-id sequence from concurrent tests. #[test] @@ -84,6 +104,47 @@ mod tests { assert_eq!(execution_async_id_u64(), 0); } + #[test] + fn resource_scope_restores_context_when_lifecycle_hooks_throw() { + const STORE: i64 = -9_401; + for before_phase in [true, false] { + reset_for_tests(); + crate::async_context::clear_store(STORE); + crate::async_context::enter_with(STORE, 11.0); + let ids = init_resource("throwing-scope", TAG_UNDEFINED_F64, true); + crate::async_context::enter_with(STORE, 22.0); + enable_throwing_lifecycle_hook(before_phase); + + let mut completion_ran = false; + let outcome = try_run_resource_scope(ids, || { + completion_ran = true; + TAG_UNDEFINED_F64 + }); + + assert_eq!(outcome.unwrap_err().to_bits(), 73.0f64.to_bits()); + assert_eq!(completion_ran, !before_phase); + assert_eq!(execution_async_id_u64(), 0); + assert!(EXECUTION_STACK.with(|stack| stack.borrow().is_empty())); + assert_eq!(crate::async_context::get_store(STORE), Some(22.0)); + crate::async_context::clear_store(STORE); + } + reset_for_tests(); + } + + #[test] + fn resource_scope_prefers_completion_error_over_after_error() { + reset_for_tests(); + let ids = init_resource("double-faulting-scope", TAG_UNDEFINED_F64, true); + enable_throwing_lifecycle_hook(false); + + let outcome = try_run_resource_scope(ids, || crate::exception::js_throw(41.0)); + + assert_eq!(outcome.unwrap_err().to_bits(), 41.0f64.to_bits()); + assert_eq!(execution_async_id_u64(), 0); + assert!(EXECUTION_STACK.with(|stack| stack.borrow().is_empty())); + reset_for_tests(); + } + #[test] fn track_promises_filters_hooks_and_activity() { reset_for_tests(); diff --git a/crates/perry-runtime/src/gc/tests/runtime_roots/hook_dispatch_handles.rs b/crates/perry-runtime/src/gc/tests/runtime_roots/hook_dispatch_handles.rs index b8a1194344..29217da550 100644 --- a/crates/perry-runtime/src/gc/tests/runtime_roots/hook_dispatch_handles.rs +++ b/crates/perry-runtime/src/gc/tests/runtime_roots/hook_dispatch_handles.rs @@ -1,5 +1,43 @@ use super::*; +extern "C" fn test_current_async_id(_closure: *const crate::closure::ClosureHeader) -> f64 { + crate::async_hooks::execution_async_id_u64() as f64 +} + +#[test] +fn test_async_resource_subclass_run_in_scope_roots_inputs_during_key_alloc_gc() { + let _async_hook_guard = AsyncHookRuntimeTestGuard::new(); + let _guard = CopyingNurseryTestGuard::new(0); + let _triggers = GcTriggerThresholdTestGuard::suppress_automatic_triggers(); + let _force_evacuation = crate::gc::knob_overrides::ForcedEvacuationTestGuard::on(); + let _verify_evacuation = crate::gc::knob_overrides::VerifyEvacuationTestGuard::on(); + register_runtime_handle_root_scanner_for_tests(); + + let resource_type = test_string_value(b"SubclassResource"); + let backing = crate::async_hooks::js_async_resource_new( + resource_type, + f64::from_bits(crate::value::TAG_UNDEFINED), + ); + let expected_async_id = crate::async_hooks::js_async_resource_async_id(backing); + let receiver = crate::object::js_object_alloc(0, 1); + crate::async_hooks::test_link_async_resource_subclass(receiver, backing); + let callback = crate::closure::js_closure_alloc(test_current_async_id as *const u8, 0); + + crate::async_hooks::test_force_next_async_resource_resolve_gc(); + let before = crate::gc::copying_minor_cycles(); + let result = crate::async_hooks::js_async_resource_run_in_async_scope( + receiver as i64, + f64::from_bits(ptr_bits(callback as usize)), + f64::from_bits(crate::value::TAG_UNDEFINED), + 0, + ); + let after = crate::gc::copying_minor_cycles(); + + assert!(after > before, "the resolver must complete a copying minor"); + assert_eq!(result, expected_async_id); + assert_eq!(crate::async_hooks::execution_async_id_u64(), 0); +} + #[test] fn test_async_hook_option_lookup_roots_callbacks_across_copied_minor_gc() { let _legacy_pacing = crate::gc::policy::force_legacy_gc_pacing(); diff --git a/crates/perry-stdlib/src/webcrypto/digest.rs b/crates/perry-stdlib/src/webcrypto/digest.rs index e56f1396de..d464638195 100644 --- a/crates/perry-stdlib/src/webcrypto/digest.rs +++ b/crates/perry-stdlib/src/webcrypto/digest.rs @@ -55,11 +55,11 @@ pub unsafe extern "C" fn js_webcrypto_digest(algo_bits: f64, data_bits: f64) -> let scope = perry_runtime::gc::RuntimeHandleScope::new(); let value = scope.root_nanbox_f64(f64::from_bits(JSValue::pointer(buf as *const u8).bits())); let promise = scope.root_raw_mut_ptr(perry_runtime::promise::js_promise_new()); + let cl = perry_runtime::closure::js_closure_alloc(webcrypto_digest_settle as *const u8, 3); + let cl = scope.root_raw_mut_ptr(cl); let promise_val = promise.with_mut_ptr(|promise: *mut Promise| { f64::from_bits(JSValue::pointer(promise as *const u8).bits()) }); - let cl = perry_runtime::closure::js_closure_alloc(webcrypto_digest_settle as *const u8, 3); - let cl = scope.root_raw_mut_ptr(cl); perry_runtime::closure::js_closure_set_capture_ptr( cl.get_raw_mut_ptr(), 0, diff --git a/crates/perry-stdlib/src/worker_threads/worker_pump.rs b/crates/perry-stdlib/src/worker_threads/worker_pump.rs index 6e05f1d72e..2fe789e91f 100644 --- a/crates/perry-stdlib/src/worker_threads/worker_pump.rs +++ b/crates/perry-stdlib/src/worker_threads/worker_pump.rs @@ -261,51 +261,53 @@ fn dispatch_worker_event(worker_id: u64, event: &str, arg: Option) { "message" | "messageerror" => async_resources[2], _ => async_resources[0], }; - perry_runtime::async_hooks::enter_resource_scope(resource); - let property_name = match event { - "message" => Some("onmessage"), - "error" => Some("onerror"), - "messageerror" => Some("onmessageerror"), - _ => None, - }; - let property_handler = property_name - .and_then(|name| object_event_handler(object_h.get_nanbox_f64().to_bits(), name)) - .map(|bits| scope.root_nanbox_f64(f64::from_bits(bits))); - let needs_event = property_handler.is_some() || callbacks.iter().any(|(_, web)| *web); - let event_handle = if needs_event { - let data = (event == "message") - .then(|| arg_handle.as_ref().map(|h| h.get_nanbox_f64())) - .flatten(); - let ev = event_object(event, object_h.get_nanbox_f64().to_bits(), data); - Some(scope.root_nanbox_f64(ev)) - } else { - None - }; - - if let (Some(callback_h), Some(event_h)) = (property_handler, event_handle.as_ref()) { - call_callback1( - callback_h.get_nanbox_f64().to_bits(), - object_h.get_nanbox_f64().to_bits(), - event_h.get_nanbox_f64(), - ); - } - - for (callback_h, web_event) in callbacks { - let closure_ptr = perry_runtime::value::js_nanbox_get_pointer(callback_h.get_nanbox_f64()); - if closure_ptr == 0 { - continue; - } - let closure = closure_ptr as *const ClosureHeader; - let call_arg = if web_event { - event_handle.as_ref().map(|h| h.get_nanbox_f64()) - } else { - arg_handle.as_ref().map(|h| h.get_nanbox_f64()) + perry_runtime::async_hooks::run_resource_scope_catching(resource, || { + let property_name = match event { + "message" => Some("onmessage"), + "error" => Some("onerror"), + "messageerror" => Some("onmessageerror"), + _ => None, }; - if let Some(arg) = call_arg { - perry_runtime::closure::js_closure_call1(closure, arg); + let property_handler = property_name + .and_then(|name| object_event_handler(object_h.get_nanbox_f64().to_bits(), name)) + .map(|bits| scope.root_nanbox_f64(f64::from_bits(bits))); + let needs_event = property_handler.is_some() || callbacks.iter().any(|(_, web)| *web); + let event_handle = if needs_event { + let data = (event == "message") + .then(|| arg_handle.as_ref().map(|h| h.get_nanbox_f64())) + .flatten(); + let ev = event_object(event, object_h.get_nanbox_f64().to_bits(), data); + Some(scope.root_nanbox_f64(ev)) } else { - perry_runtime::closure::js_closure_call0(closure); + None + }; + + if let (Some(callback_h), Some(event_h)) = (property_handler, event_handle.as_ref()) { + call_callback1( + callback_h.get_nanbox_f64().to_bits(), + object_h.get_nanbox_f64().to_bits(), + event_h.get_nanbox_f64(), + ); } - } - perry_runtime::async_hooks::leave_resource_scope(resource.async_id); + + for (callback_h, web_event) in callbacks { + let closure_ptr = + perry_runtime::value::js_nanbox_get_pointer(callback_h.get_nanbox_f64()); + if closure_ptr == 0 { + continue; + } + let closure = closure_ptr as *const ClosureHeader; + let call_arg = if web_event { + event_handle.as_ref().map(|h| h.get_nanbox_f64()) + } else { + arg_handle.as_ref().map(|h| h.get_nanbox_f64()) + }; + if let Some(arg) = call_arg { + perry_runtime::closure::js_closure_call1(closure, arg); + } else { + perry_runtime::closure::js_closure_call0(closure); + } + } + js_undefined() + }); } diff --git a/crates/perry-stdlib/src/zlib.rs b/crates/perry-stdlib/src/zlib.rs index 0b4bd1902e..14e4aac6e6 100644 --- a/crates/perry-stdlib/src/zlib.rs +++ b/crates/perry-stdlib/src/zlib.rs @@ -1351,113 +1351,182 @@ pub unsafe extern "C" fn js_zlib_process_pending() -> i32 { .and_then(|streams| streams.get(id).map(|stream| stream.async_ids)), _ => None, }; - if let Some(ids) = event_ids { - perry_runtime::async_hooks::enter_resource_scope(ids); - } - let mut destroy_after_dispatch = None; - match ev { - ZlibEvent::Data(id, bytes) => { - publish_zlib_bytes_written(id); - let cbs = listeners_for(id, "data"); - if !cbs.is_empty() { - if let Some(buf_f64) = make_buffer(&bytes) { - for cb in cbs { - if cb != 0 { - js_closure_call1(cb as *const ClosureHeader, buf_f64); + let destroy_after_dispatch = match (&ev, event_ids) { + (ZlibEvent::End(_) | ZlibEvent::Error(_, _), Some(ids)) => Some(ids.async_id), + _ => None, + }; + let dispatch = || { + match ev { + ZlibEvent::Data(id, bytes) => { + publish_zlib_bytes_written(id); + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let callbacks = listeners_for(id, "data") + .into_iter() + .map(|callback| scope.root_raw_const_ptr(callback as *const ClosureHeader)) + .collect::>(); + let destinations = pipes_for(id) + .into_iter() + .map(|destination| scope.root_nanbox_f64(f64::from_bits(destination))) + .collect::>(); + if !callbacks.is_empty() { + if let Some(buf_f64) = make_buffer(&bytes) { + let buffer = scope.root_nanbox_f64(buf_f64); + for callback in callbacks { + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call1(callback, buffer.get_nanbox_f64()); + } } } } - } - // Fresh Buffer per pipe dest (the chunk lives in the owned - // `bytes`, so this is safe even after listener callbacks GC'd). - for dest in pipes_for(id) { - forward_write(dest, &bytes); - } - } - ZlibEvent::End(id) => { - publish_zlib_bytes_written(id); - for cb in listeners_for(id, "end") { - if cb != 0 { - js_closure_call0(cb as *const ClosureHeader); + // Fresh Buffer per pipe dest (the chunk lives in the owned + // `bytes`, so this is safe even after listener callbacks GC'd). + for destination in destinations { + forward_write(destination.get_nanbox_f64().to_bits(), &bytes); } } - for cb in listeners_for(id, "finish") { - if cb != 0 { - js_closure_call0(cb as *const ClosureHeader); + ZlibEvent::End(id) => { + publish_zlib_bytes_written(id); + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let end_callbacks = listeners_for(id, "end") + .into_iter() + .map(|callback| scope.root_raw_const_ptr(callback as *const ClosureHeader)) + .collect::>(); + let finish_callbacks = listeners_for(id, "finish") + .into_iter() + .map(|callback| scope.root_raw_const_ptr(callback as *const ClosureHeader)) + .collect::>(); + let destinations = pipes_for(id) + .into_iter() + .map(|destination| scope.root_nanbox_f64(f64::from_bits(destination))) + .collect::>(); + let close_callbacks = listeners_for(id, "close") + .into_iter() + .map(|callback| scope.root_raw_const_ptr(callback as *const ClosureHeader)) + .collect::>(); + ZLIB_LISTENERS.lock().unwrap().remove(&id); + ZLIB_STREAMS.lock().unwrap().remove(&id); + for callback in end_callbacks { + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call0(callback); + } } - } - for dest in pipes_for(id) { - forward_end(dest); - } - for cb in listeners_for(id, "close") { - if cb != 0 { - js_closure_call0(cb as *const ClosureHeader); + for callback in finish_callbacks { + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call0(callback); + } + } + for destination in destinations { + forward_end(destination.get_nanbox_f64().to_bits()); + } + for callback in close_callbacks { + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call0(callback); + } } } - ZLIB_LISTENERS.lock().unwrap().remove(&id); - ZLIB_STREAMS.lock().unwrap().remove(&id); - destroy_after_dispatch = event_ids.map(|ids| ids.async_id); - } - ZlibEvent::Callback(cb) => { - if cb != 0 { - js_closure_call0(cb as *const ClosureHeader); + ZlibEvent::Callback(cb) => { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let callback = scope.root_raw_const_ptr(cb as *const ClosureHeader); + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call0(callback); + } } - } - ZlibEvent::OneShotCallback(cb, result, ids) => { - // Node exposes two ZLIB provider phases for one-shot helpers: - // native compression completion, followed by delivery of the - // JavaScript callback. They intentionally share one resource. - perry_runtime::async_hooks::run_resource_scope(ids, || {}); - perry_runtime::async_hooks::enter_resource_scope(ids); - if cb != 0 { - match result { - Ok(bytes) => { - if let Some(buf_f64) = make_buffer(&bytes) { - js_closure_call2( - cb as *const ClosureHeader, - f64::from_bits(JSValue::null().bits()), - buf_f64, - ); - } else { - let err_f64 = build_zlib_error("Buffer allocation failed"); - js_closure_call2( - cb as *const ClosureHeader, - err_f64, - f64::from_bits(JSValue::undefined().bits()), - ); + ZlibEvent::OneShotCallback(cb, result, ids) => { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let callback = scope.root_raw_const_ptr(cb as *const ClosureHeader); + // Node exposes two ZLIB provider phases for one-shot helpers: + // native compression completion, followed by delivery of the + // JavaScript callback. They intentionally share one resource. + let first_phase = + perry_runtime::async_hooks::try_run_resource_scope(ids, || { + f64::from_bits(JSValue::undefined().bits()) + }); + if let Err(error) = first_phase { + let error = scope.root_nanbox_f64(error); + perry_runtime::async_hooks::defer_destroy_after_check_turns( + ids.async_id, + 4, + ); + perry_runtime::exception::js_throw(error.get_nanbox_f64()); + } + let outcome = perry_runtime::async_hooks::try_run_resource_scope(ids, || { + if !callback.get_raw_const_ptr::().is_null() { + match result { + Ok(bytes) => { + if let Some(buf_f64) = make_buffer(&bytes) { + let buffer = scope.root_nanbox_f64(buf_f64); + js_closure_call2( + callback.get_raw_const_ptr::(), + f64::from_bits(JSValue::null().bits()), + buffer.get_nanbox_f64(), + ); + } else { + let error = scope.root_nanbox_f64(build_zlib_error( + "Buffer allocation failed", + )); + js_closure_call2( + callback.get_raw_const_ptr::(), + error.get_nanbox_f64(), + f64::from_bits(JSValue::undefined().bits()), + ); + } + } + Err(msg) => { + let error = scope.root_nanbox_f64(build_zlib_error(&msg)); + js_closure_call2( + callback.get_raw_const_ptr::(), + error.get_nanbox_f64(), + f64::from_bits(JSValue::undefined().bits()), + ); + } } } - Err(msg) => { - let err_f64 = build_zlib_error(&msg); - js_closure_call2( - cb as *const ClosureHeader, - err_f64, - f64::from_bits(JSValue::undefined().bits()), - ); - } + f64::from_bits(JSValue::undefined().bits()) + }); + let error = outcome.err().map(|error| scope.root_nanbox_f64(error)); + perry_runtime::async_hooks::defer_destroy_after_check_turns(ids.async_id, 4); + if let Some(error) = error { + perry_runtime::exception::js_throw(error.get_nanbox_f64()); } } - perry_runtime::async_hooks::leave_resource_scope(ids.async_id); - perry_runtime::async_hooks::defer_destroy_after_check_turns(ids.async_id, 4); - } - ZlibEvent::Error(id, msg) => { - let err_f64 = build_zlib_error(&msg); - for cb in listeners_for(id, "error") { - if cb != 0 { - js_closure_call1(cb as *const ClosureHeader, err_f64); + ZlibEvent::Error(id, msg) => { + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let callbacks = listeners_for(id, "error") + .into_iter() + .map(|callback| scope.root_raw_const_ptr(callback as *const ClosureHeader)) + .collect::>(); + ZLIB_LISTENERS.lock().unwrap().remove(&id); + ZLIB_STREAMS.lock().unwrap().remove(&id); + let error = scope.root_nanbox_f64(build_zlib_error(&msg)); + for callback in callbacks { + let callback = callback.get_raw_const_ptr::(); + if !callback.is_null() { + js_closure_call1(callback, error.get_nanbox_f64()); + } } } - ZLIB_LISTENERS.lock().unwrap().remove(&id); - ZLIB_STREAMS.lock().unwrap().remove(&id); - destroy_after_dispatch = event_ids.map(|ids| ids.async_id); } - } - if let Some(ids) = event_ids { - perry_runtime::async_hooks::leave_resource_scope(ids.async_id); - } + f64::from_bits(JSValue::undefined().bits()) + }; + let outcome = match event_ids { + Some(ids) => perry_runtime::async_hooks::try_run_resource_scope(ids, dispatch), + None => Ok(dispatch()), + }; + let error_scope = perry_runtime::gc::RuntimeHandleScope::new(); + let error = outcome + .err() + .map(|error| error_scope.root_nanbox_f64(error)); if let Some(async_id) = destroy_after_dispatch { perry_runtime::async_hooks::defer_destroy_after_check_turns(async_id, 4); } + if let Some(error) = error { + perry_runtime::exception::js_throw(error.get_nanbox_f64()); + } } count } diff --git a/scripts/gc_runtime_root_holders.json b/scripts/gc_runtime_root_holders.json index 8a3237de38..41757c5f6a 100644 --- a/scripts/gc_runtime_root_holders.json +++ b/scripts/gc_runtime_root_holders.json @@ -131,6 +131,12 @@ "verdict": "not_a_gc_pointer", "why": "Monotonic counter of live async-resource handles. Holds no address at all; the resource objects live in RESOURCES, which scan_async_hooks_roots_mut visits." }, + { + "file": "crates/perry-runtime/src/async_hooks.rs", + "name": "TEST_FORCE_RESOLVE_GC", + "verdict": "test_only", + "why": "#[cfg(test)] AtomicUsize one-shot flag that asks resolve_async_resource_handle to force a collection; stores only 0 or 1 and is absent from shipped binaries." + }, { "file": "crates/perry-runtime/src/child_process/reactor.rs", "name": "CP_NEXT_LIVE_ID", diff --git a/scripts/raw_handle_debt_baseline.txt b/scripts/raw_handle_debt_baseline.txt index e2cab550c5..a0394c20d6 100644 --- a/scripts/raw_handle_debt_baseline.txt +++ b/scripts/raw_handle_debt_baseline.txt @@ -1 +1 @@ -950 +949 diff --git a/scripts/raw_handle_debt_files.txt b/scripts/raw_handle_debt_files.txt index e1d0738d7b..fdb66ef785 100644 --- a/scripts/raw_handle_debt_files.txt +++ b/scripts/raw_handle_debt_files.txt @@ -53,7 +53,7 @@ 31 crates/perry-runtime/src/array/iterator.rs 4 crates/perry-runtime/src/array/push_pop.rs 14 crates/perry-runtime/src/array/sort.rs -13 crates/perry-runtime/src/async_hooks.rs +12 crates/perry-runtime/src/async_hooks.rs 5 crates/perry-runtime/src/atomics.rs 7 crates/perry-runtime/src/builtins/console.rs 12 crates/perry-runtime/src/builtins/globals.rs diff --git a/scripts/thread_local_cold_allowlist.json b/scripts/thread_local_cold_allowlist.json index db158dedeb..89f27d4433 100644 --- a/scripts/thread_local_cold_allowlist.json +++ b/scripts/thread_local_cold_allowlist.json @@ -1,6 +1,6 @@ { "_comment": "Files still declaring raw `thread_local!`. Every entry is a declaration that pays `_tlv_get_addr` on Darwin; the count is a ratchet, so adding one to an already-listed file fails too. New code should use `crate::perry_thread_local!` \u2014 see crates/perry-runtime/src/tls_hot.rs. Regenerate with scripts/check_thread_locals.py --update.", - "_hot_declarations": 263, + "_hot_declarations": 261, "files": { "crates/perry-runtime/src/agent.rs": 1, "crates/perry-runtime/src/arena/block.rs": 2, diff --git a/test-parity/node-suite/async_hooks/integrations/events-emitter.ts b/test-parity/node-suite/async_hooks/integrations/events-emitter.ts index 5f6409530d..a55bc716e5 100644 --- a/test-parity/node-suite/async_hooks/integrations/events-emitter.ts +++ b/test-parity/node-suite/async_hooks/integrations/events-emitter.ts @@ -1,5 +1,5 @@ -import { EventEmitter } from "node:events"; -import { AsyncLocalStorage } from "node:async_hooks"; +import { EventEmitter, EventEmitterAsyncResource } from "node:events"; +import { AsyncLocalStorage, executionAsyncId } from "node:async_hooks"; const storage = new AsyncLocalStorage(); @@ -30,3 +30,44 @@ await storage.run( ); console.log("events outside:", String(storage.getStore())); + +let eventNameConversions = 0; +const convertedName = { + toString() { + eventNameConversions += 1; + return "converted"; + }, +}; +const conversionEmitter = new EventEmitter(); +let convertedValue = "missing"; +conversionEmitter.on("converted", (value) => { + convertedValue = value; +}); +conversionEmitter.emit(convertedName as unknown as string, "value"); +console.log("event name conversion:", eventNameConversions, convertedValue); + +const scopedConversionEmitter = storage.run( + "conversion-resource", + () => new EventEmitterAsyncResource({ name: "ConversionEmitter" }), +); +let scopedConversions = 0; +let conversionAsyncIdsMatch = true; +const conversionStores: Array = []; +const scopedName = { + toString() { + scopedConversions += 1; + conversionAsyncIdsMatch &&= + executionAsyncId() === scopedConversionEmitter.asyncId; + conversionStores.push(storage.getStore()); + return "converted"; + }, +}; +scopedConversionEmitter.on("converted", () => {}); +storage.run("conversion-caller", () => { + scopedConversionEmitter.emit(scopedName as unknown as string); + scopedConversionEmitter.emit(scopedName as unknown as string, "value"); +}); +console.log("scoped event name conversion:", scopedConversions); +console.log("scoped event name async id:", conversionAsyncIdsMatch); +console.log("scoped event name store:", conversionStores.join(",")); +scopedConversionEmitter.emitDestroy(); diff --git a/test-parity/node-suite/async_hooks/providers/net-write-callbacks.ts b/test-parity/node-suite/async_hooks/providers/net-write-callbacks.ts index 4b1dbcfac8..04b5efdecb 100644 --- a/test-parity/node-suite/async_hooks/providers/net-write-callbacks.ts +++ b/test-parity/node-suite/async_hooks/providers/net-write-callbacks.ts @@ -20,6 +20,7 @@ try { client!.write("payload", () => { console.log("net write callback store:", storage.getStore()); }); + console.log("net write queued bytes:", client!.bytesWritten); client!.end(() => { console.log("net end callback store:", storage.getStore()); }); diff --git a/test-parity/node-suite/async_hooks/resource/shadowed-spread-parent.ts b/test-parity/node-suite/async_hooks/resource/shadowed-spread-parent.ts new file mode 100644 index 0000000000..f7d24fcc16 --- /dev/null +++ b/test-parity/node-suite/async_hooks/resource/shadowed-spread-parent.ts @@ -0,0 +1,19 @@ +// A lexical class with a builtin-looking name must still run its own spread +// constructor path instead of being lowered as a native AsyncResource parent. +class AsyncResource { + readonly marker: string; + constructor(...values: string[]) { + this.marker = `user:${values.join(",")}`; + } +} + +class ShadowedResource extends AsyncResource { + constructor(...values: string[]) { + super(...values); + } +} + +console.log( + "shadowed spread parent:", + new ShadowedResource("first", "second").marker, +);