diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 396be457..5864d335 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -326,10 +326,25 @@ struct RuntimeState { inbound: StdMutex>>, host: HostCalls, runtime: StdMutex>, - reinitialize: Option, + reinitialize: Arc>>, } type Reinitialize = Box i32 + Send + Sync>; +async fn dispose_failed_setup_callback( + callback: Arc>>, +) -> Result<()> { + // Reinitialization holds exclusion while driving setup with block_on. + // Waiting for it must leave runtime workers free to advance that future + // and its timer. Observe completion before publishing the failure. + tokio::task::spawn_blocking(move || { + callback + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + }) + .await + .map_err(|_| Error::failed("module failed-setup callback disposal panicked")) +} type ReinitializeFactory = Box< dyn FnOnce(tokio::runtime::Handle, Arc>>) -> Reinitialize + Send, >; @@ -363,6 +378,15 @@ unsafe extern "C" fn shutdown(ctx: *mut c_void, deadline_ms: u64) -> i32 { return TB_BAD_ARGUMENT; } let state = unsafe { &*(ctx.cast::()) }; + // An in-flight reinitializer finishes before disposal. Once the slot + // is empty, late calls see a tombstone. Release the lock before joining + // runtime tasks, whose failed-setup cleanup also uses this slot. + let mut callback = state + .reinitialize + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + callback.take(); + drop(callback); state.inbound.lock().expect("module inbound lock").take(); match state.runtime.lock().expect("module runtime lock").take() { Some(runtime) => { @@ -383,7 +407,11 @@ unsafe extern "C" fn reinitialize(ctx: *mut c_void, ptr: *const u8, len: usize) return TB_BAD_ARGUMENT; } let state = unsafe { &*(ctx.cast::()) }; - let Some(reinitialize) = &state.reinitialize else { + let callback = state + .reinitialize + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(reinitialize) = callback.as_ref() else { return TB_CLOSED; }; reinitialize(ptr, len) @@ -496,6 +524,24 @@ where } } +// Validate the frozen ABI prefix before configuration parsing or a factory can +// allocate state. Never read or write beyond a caller's declared allocation. +unsafe fn validate_start_arguments( + host: *const TbHostVtable, + out: *mut TbModuleVtable, + worker_threads: usize, +) -> Option<(u32, TbHostVtable)> { + if host.is_null() || out.is_null() || worker_threads == 0 { + return None; + } + let out_capacity = unsafe { out.cast::().read() }; + if out_capacity < TB_MODULE_VTABLE_BASE_SIZE { + return None; + } + let host = unsafe { TbHostVtable::read_compatible(host) }?; + Some((out_capacity, host)) +} + unsafe fn start_module_runtime( host: *const TbHostVtable, out: *mut TbModuleVtable, @@ -510,19 +556,9 @@ where Fut: Future> + Send + 'static, { match catch_unwind(AssertUnwindSafe(|| { - if host.is_null() || out.is_null() || worker_threads == 0 { - return TB_BAD_ARGUMENT; - } - // `out.size` is input as well as output: an older ABI-v1 host owns a - // shorter allocation. Read only the frozen first field until its - // capacity is known, then write no further than that boundary. - let out_capacity = unsafe { out.cast::().read() }; - if out_capacity < TB_MODULE_VTABLE_BASE_SIZE { - return TB_BAD_ARGUMENT; - } - // `size` is the frozen prefix field; do not copy the full vtable until - // the host has proved that all v1 fields are present. - let Some(host) = (unsafe { TbHostVtable::read_compatible(host) }) else { + let Some((out_capacity, host)) = + (unsafe { validate_start_arguments(host, out, worker_threads) }) + else { return TB_BAD_ARGUMENT; }; let host = HostCalls(host); @@ -587,9 +623,14 @@ where let connection_slot = Arc::new(StdMutex::new(None)); let task_connection_slot = connection_slot.clone(); - runtime.spawn(async move { + let reinitializer = + reinitialize_factory.map(|factory| factory(runtime.handle().clone(), connection_slot)); + let supports_reinitialize = reinitializer.is_some(); + let reinitializer = Arc::new(StdMutex::new(reinitializer)); + let failed_setup_callback = reinitializer.clone(); + let setup_task = runtime.spawn(async move { let connection = module_connection(transport); - let outcome = match connection.handshake().await.map(|()| connection) { + match connection.handshake().await.map(|()| connection) { Ok(connection) => { connection.__set_panic_handler(std::sync::Arc::new(move || { let location = panic_location @@ -618,17 +659,28 @@ where } } Err(error) => Err(error), - }; - if let Err(error) = outcome { - tracing::error!(error = %error, "tinybus module failed"); + } + }); + runtime.spawn(async move { + let outcome = setup_task.await; + if !matches!(outcome, Ok(Ok(()))) { + // A terminal setup failure may never receive ABI shutdown. + // Dispose its callback under the same lifecycle exclusion. + if dispose_failed_setup_callback(failed_setup_callback) + .await + .is_err() + { + tracing::error!("tinybus module failed-setup callback disposal failed"); + } + match outcome { + Ok(Err(error)) => tracing::error!(error = %error, "tinybus module failed"), + Err(_) => tracing::error!("tinybus module setup task failed"), + Ok(Ok(())) => unreachable!(), + } host.fault(); } }); - let reinitializer = - reinitialize_factory.map(|factory| factory(runtime.handle().clone(), connection_slot)); - let supports_reinitialize = reinitializer.is_some(); - let state = Box::new(RuntimeState { host, inbound: StdMutex::new(Some(inbound_tx)), @@ -775,6 +827,12 @@ where } /// Initialize one configured instance with fresh captured setup state. +/// +/// # Safety +/// Non-null pointers must address readable host ABI prefixes and writable +/// output storage matching their declared sizes. Host callbacks and context +/// must remain valid until runtime shutdown. Null pointers and invalid sizes +/// or worker counts are rejected before configuration or factory evaluation. #[doc(hidden)] pub unsafe fn start_reconfigurable_module_with_factory( host: *const TbHostVtable, @@ -785,7 +843,7 @@ pub unsafe fn start_reconfigurable_module_with_factory( ) -> i32 where C: serde::de::DeserializeOwned + Send + 'static, - F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, + F: Fn(Connection, C) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, { unsafe { @@ -794,13 +852,19 @@ where out, worker_threads, detach_on_panic, - factory, + || shared_factory_setup(factory), false, ) } } /// Initialize one linked configured instance with fresh captured setup state. +/// +/// # Safety +/// Non-null pointers must address readable host ABI prefixes and writable +/// output storage matching their declared sizes. Host callbacks and context +/// must remain valid until runtime shutdown. Null pointers and invalid sizes +/// or worker counts are rejected before configuration or factory evaluation. #[doc(hidden)] pub unsafe fn start_linked_reconfigurable_module_with_factory( host: *const TbHostVtable, @@ -811,7 +875,7 @@ pub unsafe fn start_linked_reconfigurable_module_with_factory( ) -> i32 where C: serde::de::DeserializeOwned + Send + 'static, - F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, + F: Fn(Connection, C) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, { unsafe { @@ -820,12 +884,24 @@ where out, worker_threads, detach_on_panic, - factory, + || shared_factory_setup(factory), true, ) } } +// The factory callback itself is shared, rather than cloned: a user's Clone +// may deep-copy interior state and detach initial setup from reinitialization. +fn shared_factory_setup( + factory: impl FnOnce() -> F, +) -> impl Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static +where + F: Fn(Connection, C) -> Fut + Send + Sync + 'static, +{ + let setup = Arc::new(factory()); + move |connection, config| setup(connection, config) +} + unsafe fn start_reconfigurable_module_factory_mode( host: *const TbHostVtable, out: *mut TbModuleVtable, @@ -839,6 +915,9 @@ where F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, { + if unsafe { validate_start_arguments(host, out, worker_threads) }.is_none() { + return TB_BAD_ARGUMENT; + } let parsed = catch_unwind(AssertUnwindSafe(|| unsafe { parse_config::(host) })); let config = match parsed { Ok(Ok(config)) => config, @@ -904,7 +983,9 @@ where /// Configured modules may use `setup_factory = factory` in place of `setup`. /// After valid configuration is parsed, the safe no-argument factory runs once /// per ABI initialization and returns a -/// cloneable `Fn(Connection, Config) -> Future>`. Its captured +/// `Fn(Connection, Config) -> Future>`. The callback is shared +/// by reference for initial setup and reinitialization, regardless of its Clone +/// behavior. Its captured /// state is reused for that instance's reinitializations, allowing admission to /// remain bound without process-global state. Separate instances receive fresh /// state. A factory panic returns `TB_PANICKED` without unwinding into the host. @@ -1142,5 +1223,9 @@ mod tests; mod static_link_tests; #[cfg(test)] -#[path = "setup_factory_tests.rs"] +#[path = "lib_setup_factory_tests.rs"] mod setup_factory_test; + +#[cfg(test)] +#[path = "lib_lifecycle_tests.rs"] +mod lifecycle_test; diff --git a/crates/tinybus-module/src/lib_lifecycle_tests.rs b/crates/tinybus-module/src/lib_lifecycle_tests.rs new file mode 100644 index 00000000..89756fc8 --- /dev/null +++ b/crates/tinybus-module/src/lib_lifecycle_tests.rs @@ -0,0 +1,206 @@ +use super::*; +use std::sync::atomic::AtomicUsize; + +unsafe extern "C" fn send(_: *mut c_void, _: *const u8, _: usize) -> i32 { + TB_OK +} +unsafe extern "C" fn wake(_: *mut c_void) {} +unsafe extern "C" fn log(_: *mut c_void, _: u32, _: *const u8, _: usize) {} +unsafe extern "C" fn fault(_: *mut c_void, _: *const u8, _: usize) {} +unsafe extern "C" fn ready(_: *mut c_void) {} +fn host() -> TbHostVtable { + TbHostVtable { + size: size_of::() as u32, + _reserved: 0, + host_ctx: std::ptr::null_mut(), + send, + wake, + log, + fault, + config: tinybus::module::abi::TbSlice { + ptr: b"{}".as_ptr(), + len: 2, + }, + ready, + broker_routing: None, + } +} +struct DropCounter(Arc); +impl Drop for DropCounter { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} +#[test] +fn shutdown_releases_captured_callback_while_retaining_closed_tombstone() { + let dropped = Arc::new(AtomicUsize::new(0)); + let token = DropCounter(Arc::clone(&dropped)); + let runtime = tokio::runtime::Runtime::new().unwrap(); + let mut state = RuntimeState { + inbound: StdMutex::new(None), + host: HostCalls(host()), + runtime: StdMutex::new(Some(runtime)), + reinitialize: Arc::new(StdMutex::new(Some(Box::new(move |_, _| { + let _ = &token; + TB_OK + })))), + }; + let pointer = std::ptr::from_mut(&mut state).cast(); + assert_eq!(unsafe { shutdown(pointer, 1000) }, TB_OK); + assert_eq!( + dropped.load(Ordering::SeqCst), + 1, + "ABI tombstone must not retain factory state" + ); + assert_eq!( + unsafe { reinitialize(pointer, std::ptr::null(), 0) }, + TB_CLOSED + ); + assert_eq!(unsafe { deliver(pointer, b"{}".as_ptr(), 2) }, TB_CLOSED); + assert_eq!(unsafe { shutdown(pointer, 1000) }, TB_CLOSED); +} +#[test] +fn invalid_output_or_workers_rejects_before_factory_side_effects() { + let host = host(); + let mut observations = Vec::new(); + for (output_size, workers, null_output) in [ + (size_of::() as u32, 1, true), + (0, 1, false), + (size_of::() as u32, 0, false), + ] { + let calls = Arc::new(AtomicUsize::new(0)); + let count = Arc::clone(&calls); + let mut output = std::mem::MaybeUninit::::zeroed(); + let pointer = if null_output { + std::ptr::null_mut() + } else { + unsafe { + output.as_mut_ptr().cast::().write(output_size); + } + output.as_mut_ptr() + }; + let code = unsafe { + start_linked_reconfigurable_module_with_factory::( + &host, + pointer, + workers, + true, + move || { + count.fetch_add(1, Ordering::SeqCst); + |_: Connection, _: serde_json::Value| async { Ok(()) } + }, + ) + }; + assert_eq!(code, TB_BAD_ARGUMENT); + observations.push(calls.load(Ordering::SeqCst)); + } + assert_eq!( + observations, + [0, 0, 0], + "invalid ABI must not invoke factory" + ); +} + +#[test] +fn shutdown_excludes_active_reinitializer_before_disposing_capture() { + let dropped = Arc::new(AtomicUsize::new(0)); + let token = DropCounter(Arc::clone(&dropped)); + let (entered, started) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + let released = StdMutex::new(released); + let state = Arc::new(RuntimeState { + inbound: StdMutex::new(None), + host: HostCalls(host()), + runtime: StdMutex::new(Some(tokio::runtime::Runtime::new().unwrap())), + reinitialize: Arc::new(StdMutex::new(Some(Box::new(move |_, _| { + let _ = &token; + entered.send(()).unwrap(); + released + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(5)) + .unwrap(); + TB_OK + })))), + }); + let active = Arc::clone(&state); + let reinitializer = std::thread::spawn(move || unsafe { + reinitialize(Arc::as_ptr(&active).cast_mut().cast(), std::ptr::null(), 0) + }); + started.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!( + state.reinitialize.try_lock().is_err(), + "active ABI callback must retain lifecycle exclusion" + ); + assert_eq!(dropped.load(Ordering::SeqCst), 0); + let stopping = Arc::clone(&state); + let shutdown_thread = std::thread::spawn(move || unsafe { + shutdown(Arc::as_ptr(&stopping).cast_mut().cast(), 1000) + }); + release.send(()).unwrap(); + assert_eq!(reinitializer.join().unwrap(), TB_OK); + assert_eq!(shutdown_thread.join().unwrap(), TB_OK); + assert_eq!(dropped.load(Ordering::SeqCst), 1); + assert_eq!( + unsafe { reinitialize(Arc::as_ptr(&state).cast_mut().cast(), std::ptr::null(), 0) }, + TB_CLOSED + ); +} + +#[test] +fn shutdown_disposes_callback_even_after_reinitializer_panics() { + let dropped = Arc::new(AtomicUsize::new(0)); + let token = DropCounter(Arc::clone(&dropped)); + let mut state = RuntimeState { + inbound: StdMutex::new(None), + host: HostCalls(host()), + runtime: StdMutex::new(None), + reinitialize: Arc::new(StdMutex::new(Some(Box::new(move |_, _| { + let _ = &token; + panic!("fixture callback failed") + })))), + }; + let pointer = std::ptr::from_mut(&mut state).cast(); + assert_eq!( + unsafe { reinitialize(pointer, std::ptr::null(), 0) }, + TB_PANICKED + ); + assert_eq!(unsafe { shutdown(pointer, 1000) }, TB_CLOSED); + assert_eq!(dropped.load(Ordering::SeqCst), 1); +} + +#[test] +fn failed_setup_disposal_keeps_single_runtime_worker_available() { + let callback: Arc>> = + Arc::new(StdMutex::new(Some(Box::new(|_, _| TB_OK)))); + let held = callback.lock().unwrap(); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .unwrap(); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let dispose = callback.clone(); + runtime.spawn(async move { + entered_tx.send(()).unwrap(); + dispose_failed_setup_callback(dispose).await.unwrap(); + done_tx.send(()).unwrap(); + }); + entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let (progress_tx, progress_rx) = std::sync::mpsc::channel(); + runtime.spawn(async move { + tokio::time::sleep(Duration::from_millis(1)).await; + progress_tx.send(()).unwrap(); + }); + let progressed = progress_rx.recv_timeout(Duration::from_secs(1)).is_ok(); + // Always release exclusion before asserting, including the RED path. + drop(held); + done_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(callback.lock().unwrap().is_none()); + runtime.shutdown_timeout(Duration::from_secs(1)); + assert!( + progressed, + "callback disposal must not occupy the sole timer worker" + ); +} diff --git a/crates/tinybus-module/src/lib_setup_factory_tests.rs b/crates/tinybus-module/src/lib_setup_factory_tests.rs new file mode 100644 index 00000000..1fe9cb9d --- /dev/null +++ b/crates/tinybus-module/src/lib_setup_factory_tests.rs @@ -0,0 +1,538 @@ +use super::*; +use std::{ + future::Future, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; +use tinybus::{broker::Broker, module::ModuleHost, transport::memory::MemoryBus}; + +type SetupFuture = Pin> + Send>>; +type SetupFn = fn(Connection, Config) -> SetupFuture; +static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +static FACTORIES: AtomicUsize = AtomicUsize::new(0); +#[derive(serde::Deserialize)] +struct Config { + restricted: bool, +} +struct Service { + restricted: bool, +} +#[async_trait::async_trait] +impl tinybus::service::Interface for Service { + fn name(&self) -> tinybus::InterfaceName { + "ai.tinyhumans.tinybus.Factory".try_into().unwrap() + } + fn members(&self) -> Vec { + vec!["Mode".try_into().unwrap()] + } + async fn call( + &self, + _member: &tinybus::MemberName, + _args: serde_json::Value, + ) -> tinybus::Result { + Ok(serde_json::json!(self.restricted)) + } +} +async fn legacy_setup(connection: Connection, config: Config) -> Result<()> { + connection + .serve_at( + "/factory".try_into()?, + Service { + restricted: config.restricted, + }, + ) + .await?; + connection + .request_name("ai.tinyhumans.tinybus.Factory") + .await?; + Ok(()) +} +fn setup_factory() +-> impl Fn(Connection, Config) -> Pin> + Send>> + Clone + Send + Sync +{ + FACTORIES.fetch_add(1, Ordering::SeqCst); + let mode = Arc::new(tokio::sync::Mutex::new(None)); + move |connection, config: Config| { + let mode = Arc::clone(&mode); + Box::pin(async move { + let mut admitted = mode.lock().await; + if admitted.is_some_and(|previous| previous != config.restricted) { + return Err(tinybus::Error::failed("immutable mode")); + } + *admitted = Some(config.restricted); + legacy_setup(connection, config).await + }) + } +} +module_export_static! { + setup_factory = setup_factory, + config = Config, + worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.Factory"], + methods = ["Mode"], signals = [], requires = [], optional = [], lazy = false, +} +async fn instance(restricted: bool) -> (ModuleHost, Connection, tinybus::Proxy) { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + let host = ModuleHost::new(broker); + host.attach_linked_with_config( + linked_module().unwrap(), + serde_json::json!({"restricted":restricted}), + ) + .unwrap(); + let caller = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let proxy = caller + .proxy( + "ai.tinyhumans.tinybus.Factory", + "/factory", + "ai.tinyhumans.tinybus.Factory", + ) + .unwrap(); + assert_eq!(proxy.call::("Mode", ()).await.unwrap(), restricted); + (host, caller, proxy) +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_factory_is_per_instance_and_reinitialization_keeps_its_admission() { + let _guard = TEST_LOCK.lock().await; + let before = FACTORIES.load(Ordering::SeqCst); + let ((first, _first_caller, first_proxy), (second, _second_caller, second_proxy)) = + tokio::join!(instance(true), instance(false)); + assert_eq!( + FACTORIES.load(Ordering::SeqCst) - before, + 2, + "factory must run once per ABI init" + ); + let (same, changed) = tokio::join!( + first.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), + second.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), + ); + assert!(same.is_ok()); + assert!(changed.is_err()); + assert!(first_proxy.call::("Mode", ()).await.unwrap()); + assert!(!second_proxy.call::("Mode", ()).await.unwrap()); + assert_eq!( + FACTORIES.load(Ordering::SeqCst) - before, + 2, + "reinitialization must reuse factory state" + ); +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_instance_refuses_mode_change_without_replacing_live_service() { + let _guard = TEST_LOCK.lock().await; + let (host, _caller, proxy) = instance(true).await; + let result = host + .reinitialize("tinybus-module", serde_json::json!({"restricted":false})) + .await; + assert!( + result.is_err(), + "setup factory admission must survive reinitialization" + ); + assert!(proxy.call::("Mode", ()).await.unwrap()); +} +#[tokio::test] +async fn captured_admission_serializes_concurrent_setup_attempts() { + let _guard = TEST_LOCK.lock().await; + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let connection = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let setup = setup_factory(); + let (first, second) = tokio::join!( + setup(connection.clone(), Config { restricted: true }), + setup(connection.clone(), Config { restricted: false }) + ); + assert_ne!(first.is_ok(), second.is_ok()); +} + +mod panicking { + use super::*; + fn factory() -> SetupFn { + panic!("factory failed") + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.PanickingFactory"], + methods = [], signals = [], requires = [], optional = [], lazy = false, + } +} +#[test] +fn factory_panic_is_a_native_initialization_refusal() { + let broker = Broker::new(); + let host = ModuleHost::new(broker); + let error = host + .attach_linked_with_config( + panicking::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap_err(); + assert!(matches!(error, tinybus::Error::ModuleRefused { .. })); +} +#[tokio::test] +async fn invalid_initial_configuration_does_not_construct_factory_state() { + let _guard = TEST_LOCK.lock().await; + let before = FACTORIES.load(Ordering::SeqCst); + let host = ModuleHost::new(Broker::new()); + assert!( + host.attach_linked_with_config( + linked_module().unwrap(), + serde_json::json!({"restricted":"invalid"}) + ) + .is_err() + ); + assert_eq!(FACTORIES.load(Ordering::SeqCst), before); +} + +#[test] +fn native_factory_starter_validates_host_before_constructing_state() { + let code = unsafe { + start_reconfigurable_module_with_factory::( + std::ptr::null(), + std::ptr::null_mut(), + 1, + true, + || { + panic!("invalid host must be rejected before factory"); + #[allow(unreachable_code)] + setup_factory() + }, + ) + }; + assert_eq!(code, TB_BAD_ARGUMENT); +} + +mod copying_state { + use super::*; + #[derive(Default)] + struct Mode(std::sync::Mutex>); + impl Clone for Mode { + fn clone(&self) -> Self { + Self(std::sync::Mutex::new(*self.0.lock().unwrap())) + } + } + impl Mode { + fn admit(&self, requested: bool) -> bool { + let mut mode = self.0.lock().unwrap(); + if mode.is_some_and(|previous| previous != requested) { + return false; + } + *mode = Some(requested); + true + } + } + fn factory() -> impl Fn(Connection, Config) -> SetupFuture + Clone + Send + Sync { + let mode = Mode::default(); + move |connection, config: Config| { + let accepted = mode.admit(config.restricted); + Box::pin(async move { + if !accepted { + return Err(tinybus::Error::failed("immutable mode")); + } + legacy_setup(connection, config).await + }) + } + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.Factory"], + methods = ["Mode"], signals = [], requires = [], optional = [], lazy = false, + } +} +#[tokio::test(flavor = "current_thread")] +async fn factory_callback_with_copying_clone_keeps_initial_admission_on_reinitialize() { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + let host = ModuleHost::new(broker); + host.attach_linked_with_config( + copying_state::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap(); + let caller = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let proxy = caller + .proxy( + "ai.tinyhumans.tinybus.Factory", + "/factory", + "ai.tinyhumans.tinybus.Factory", + ) + .unwrap(); + assert!(proxy.call::("Mode", ()).await.unwrap()); + assert!( + host.reinitialize("tinybus-module", serde_json::json!({"restricted":false})) + .await + .is_err(), + "a copying callback Clone must not reset the original admission" + ); + assert!(proxy.call::("Mode", ()).await.unwrap()); +} + +mod failed_setup { + use super::*; + pub(super) static DROPPED: AtomicUsize = AtomicUsize::new(0); + struct Capture; + impl Drop for Capture { + fn drop(&mut self) { + DROPPED.fetch_add(1, Ordering::SeqCst); + } + } + fn factory() -> impl Fn(Connection, Config) -> SetupFuture + Send + Sync { + let capture = Capture; + move |_, _| { + let _ = &capture; + Box::pin(async { Err(tinybus::Error::failed("setup refused")) }) + } + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.FailedFactory"], + methods = [], signals = [], requires = [], optional = [], lazy = false, + } +} +#[tokio::test(flavor = "current_thread")] +async fn failed_initial_setup_releases_factory_capture_when_host_stops_instance() { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus); + let host = ModuleHost::new(broker); + let before = failed_setup::DROPPED.load(Ordering::SeqCst); + host.attach_linked_with_config( + failed_setup::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if host.list().iter().any(|m| { + matches!( + m.state, + tinybus::module::ModuleState::Failed { .. } + | tinybus::module::ModuleState::Faulted { .. } + ) + }) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + host.shutdown(std::time::Duration::from_secs(1)).await; + assert_eq!( + failed_setup::DROPPED.load(Ordering::SeqCst) - before, + 1, + "failed instance tombstone must release callback capture" + ); +} + +mod released_setup { + use super::*; + pub(super) static DROPPED: AtomicUsize = AtomicUsize::new(0); + struct Capture; + impl Drop for Capture { + fn drop(&mut self) { + DROPPED.fetch_add(1, Ordering::SeqCst); + } + } + fn factory() -> impl Fn(Connection, Config) -> SetupFuture + Send + Sync { + let capture = Capture; + move |connection, config| { + let _ = &capture; + Box::pin(legacy_setup(connection, config)) + } + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.Factory"], + methods = ["Mode"], signals = [], requires = [], optional = [], lazy = false, + } +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_successful_shutdown_releases_factory_capture() { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + let host = ModuleHost::new(broker); + let before = released_setup::DROPPED.load(Ordering::SeqCst); + host.attach_linked_with_config( + released_setup::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap(); + let caller = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let proxy = caller + .proxy( + "ai.tinyhumans.tinybus.Factory", + "/factory", + "ai.tinyhumans.tinybus.Factory", + ) + .unwrap(); + assert!(proxy.call::("Mode", ()).await.unwrap()); + assert_eq!(released_setup::DROPPED.load(Ordering::SeqCst), before); + host.shutdown(std::time::Duration::from_secs(1)).await; + assert_eq!(released_setup::DROPPED.load(Ordering::SeqCst) - before, 1); +} +mod panicking_setup { + use super::*; + pub(super) static DROPPED: AtomicUsize = AtomicUsize::new(0); + struct Capture; + impl Drop for Capture { + fn drop(&mut self) { + DROPPED.fetch_add(1, Ordering::SeqCst); + } + } + fn factory() -> impl Fn(Connection, Config) -> SetupFuture + Send + Sync { + let capture = Capture; + move |_, _| { + let _ = &capture; + Box::pin(async { panic!("harmless initial setup fixture") }) + } + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.PanickingFactory"], + methods = [], signals = [], requires = [], optional = [], lazy = false, + } +} +#[tokio::test(flavor = "current_thread")] +async fn panicking_initial_setup_releases_factory_capture_and_reports_fault() { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus); + let host = ModuleHost::new(broker); + let before = panicking_setup::DROPPED.load(Ordering::SeqCst); + host.attach_linked_with_config( + panicking_setup::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if host + .list() + .iter() + .any(|m| matches!(m.state, tinybus::module::ModuleState::Faulted { .. })) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!(panicking_setup::DROPPED.load(Ordering::SeqCst) - before, 1); +} + +mod worker_progress_setup { + use super::*; + pub(super) type Channels = ( + std::sync::mpsc::Sender<()>, + std::sync::mpsc::Receiver<()>, + std::sync::mpsc::Sender<()>, + ); + pub(super) static CHANNELS: StdMutex> = StdMutex::new(None); + struct Capture { + started: std::sync::mpsc::Sender<()>, + release: StdMutex>, + notify: Arc, + } + impl Drop for Capture { + fn drop(&mut self) { + self.started.send(()).unwrap(); + self.notify.notify_one(); + self.release.lock().unwrap().recv().unwrap(); + } + } + fn factory() -> impl Fn(Connection, Config) -> SetupFuture + Send + Sync { + let (started, release, progress) = CHANNELS.lock().unwrap().take().unwrap(); + let notify = Arc::new(tokio::sync::Notify::new()); + let capture = Capture { + started, + release: StdMutex::new(release), + notify: notify.clone(), + }; + move |_, _| { + let _ = &capture; + let notify = notify.clone(); + let progress = progress.clone(); + Box::pin(async move { + tokio::spawn(async move { + notify.notified().await; + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + progress.send(()).unwrap(); + }); + Err(tinybus::Error::failed("fixture setup refused")) + }) + } + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.WorkerFactory"], + methods = [], signals = [], requires = [], optional = [], lazy = false, + } +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_single_worker_disposal_advances_timer_before_reporting_fault() { + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let (progress_tx, progress_rx) = std::sync::mpsc::channel(); + *worker_progress_setup::CHANNELS.lock().unwrap() = Some((started_tx, release_rx, progress_tx)); + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus); + let host = ModuleHost::new(broker); + host.attach_linked_with_config( + worker_progress_setup::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap(); + tokio::task::spawn_blocking(move || started_rx.recv_timeout(std::time::Duration::from_secs(5))) + .await + .unwrap() + .unwrap(); + let progressed = tokio::task::spawn_blocking(move || { + progress_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .is_ok() + }) + .await + .unwrap(); + let fault_before_completion = host + .list() + .iter() + .any(|m| matches!(m.state, tinybus::module::ModuleState::Faulted { .. })); + // Release disposal even on assertion failures. + release_tx.send(()).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if host + .list() + .iter() + .any(|m| matches!(m.state, tinybus::module::ModuleState::Faulted { .. })) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!( + progressed, + "compiled module's sole timer worker must remain available" + ); + assert!( + !fault_before_completion, + "Faulted must follow observed disposal completion" + ); +} diff --git a/crates/tinybus-module/src/lib_tests.rs b/crates/tinybus-module/src/lib_tests.rs index 663af561..8892d848 100644 --- a/crates/tinybus-module/src/lib_tests.rs +++ b/crates/tinybus-module/src/lib_tests.rs @@ -94,7 +94,7 @@ fn a_module_whose_queue_is_full_reports_backpressure_rather_than_blocking_the_br host: HostCalls(host(b"{}")), inbound: StdMutex::new(Some(sender)), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let bytes = b"{}"; let code = unsafe { @@ -113,7 +113,7 @@ fn a_frame_over_the_size_cap_is_rejected_rather_than_truncated() { host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let code = unsafe { deliver( @@ -205,7 +205,7 @@ fn a_panicking_shutdown_callback_reports_panicked_not_timed_out() { host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let _ = catch_unwind(AssertUnwindSafe(|| { let _guard = state.runtime.lock().unwrap(); @@ -229,7 +229,7 @@ fn deliver_and_shutdown_validate_their_arguments_and_closed_state() { host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let bytes = b"{}"; assert_eq!( @@ -603,7 +603,7 @@ fn deliver_reports_closed_when_the_modules_queue_receiver_is_gone() { host: HostCalls(host(b"{}")), inbound: StdMutex::new(Some(tx)), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let bytes = b"{}"; assert_eq!( @@ -629,7 +629,7 @@ fn reinitialize_validates_its_arguments_and_reports_closed_without_a_handler() { host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let ctx = std::ptr::from_ref(&state).cast_mut().cast(); assert_eq!( @@ -797,7 +797,7 @@ async fn admission_never_promotes_a_frame_already_queued_without_provenance() { host: HostCalls(table), inbound: StdMutex::new(Some(sender)), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let transport = ModuleTransport { host: HostCalls(table), @@ -893,7 +893,7 @@ async fn unverified_native_frames_preserve_distinct_stream_owners_and_unverified host: HostCalls(table), inbound: StdMutex::new(Some(sender)), runtime: StdMutex::new(None), - reinitialize: None, + reinitialize: Arc::new(StdMutex::new(None)), }; let transport = Arc::new(ModuleTransport { host: HostCalls(table), diff --git a/crates/tinybus-module/src/setup_factory_tests.rs b/crates/tinybus-module/src/setup_factory_tests.rs deleted file mode 100644 index f41a6adb..00000000 --- a/crates/tinybus-module/src/setup_factory_tests.rs +++ /dev/null @@ -1,208 +0,0 @@ -use super::*; -use std::{ - future::Future, - pin::Pin, - sync::{ - Arc, - atomic::{AtomicUsize, Ordering}, - }, -}; -use tinybus::{broker::Broker, module::ModuleHost, transport::memory::MemoryBus}; - -type SetupFuture = Pin> + Send>>; -type SetupFn = fn(Connection, Config) -> SetupFuture; -static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); -static FACTORIES: AtomicUsize = AtomicUsize::new(0); -#[derive(serde::Deserialize)] -struct Config { - restricted: bool, -} -struct Service { - restricted: bool, -} -#[async_trait::async_trait] -impl tinybus::service::Interface for Service { - fn name(&self) -> tinybus::InterfaceName { - "ai.tinyhumans.tinybus.Factory".try_into().unwrap() - } - fn members(&self) -> Vec { - vec!["Mode".try_into().unwrap()] - } - async fn call( - &self, - _member: &tinybus::MemberName, - _args: serde_json::Value, - ) -> tinybus::Result { - Ok(serde_json::json!(self.restricted)) - } -} -async fn legacy_setup(connection: Connection, config: Config) -> Result<()> { - connection - .serve_at( - "/factory".try_into()?, - Service { - restricted: config.restricted, - }, - ) - .await?; - connection - .request_name("ai.tinyhumans.tinybus.Factory") - .await?; - Ok(()) -} -fn setup_factory() --> impl Fn(Connection, Config) -> Pin> + Send>> + Clone + Send + Sync -{ - FACTORIES.fetch_add(1, Ordering::SeqCst); - let mode = Arc::new(tokio::sync::Mutex::new(None)); - move |connection, config: Config| { - let mode = Arc::clone(&mode); - Box::pin(async move { - let mut admitted = mode.lock().await; - if admitted.is_some_and(|previous| previous != config.restricted) { - return Err(tinybus::Error::failed("immutable mode")); - } - *admitted = Some(config.restricted); - legacy_setup(connection, config).await - }) - } -} -module_export_static! { - setup_factory = setup_factory, - config = Config, - worker_threads = 1, - provides = ["ai.tinyhumans.tinybus.Factory"], - methods = ["Mode"], signals = [], requires = [], optional = [], lazy = false, -} -async fn instance(restricted: bool) -> (ModuleHost, Connection, tinybus::Proxy) { - let bus = MemoryBus::new(); - let broker = Broker::new(); - broker.clone().spawn(bus.clone()); - let host = ModuleHost::new(broker); - host.attach_linked_with_config( - linked_module().unwrap(), - serde_json::json!({"restricted":restricted}), - ) - .unwrap(); - let caller = Connection::connect(bus.connect().await.unwrap()) - .await - .unwrap(); - let proxy = caller - .proxy( - "ai.tinyhumans.tinybus.Factory", - "/factory", - "ai.tinyhumans.tinybus.Factory", - ) - .unwrap(); - assert_eq!(proxy.call::("Mode", ()).await.unwrap(), restricted); - (host, caller, proxy) -} -#[tokio::test(flavor = "current_thread")] -async fn compiled_factory_is_per_instance_and_reinitialization_keeps_its_admission() { - let _guard = TEST_LOCK.lock().await; - let before = FACTORIES.load(Ordering::SeqCst); - let ((first, _first_caller, first_proxy), (second, _second_caller, second_proxy)) = - tokio::join!(instance(true), instance(false)); - assert_eq!( - FACTORIES.load(Ordering::SeqCst) - before, - 2, - "factory must run once per ABI init" - ); - let (same, changed) = tokio::join!( - first.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), - second.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), - ); - assert!(same.is_ok()); - assert!(changed.is_err()); - assert!(first_proxy.call::("Mode", ()).await.unwrap()); - assert!(!second_proxy.call::("Mode", ()).await.unwrap()); - assert_eq!( - FACTORIES.load(Ordering::SeqCst) - before, - 2, - "reinitialization must reuse factory state" - ); -} -#[tokio::test(flavor = "current_thread")] -async fn compiled_instance_refuses_mode_change_without_replacing_live_service() { - let _guard = TEST_LOCK.lock().await; - let (host, _caller, proxy) = instance(true).await; - let result = host - .reinitialize("tinybus-module", serde_json::json!({"restricted":false})) - .await; - assert!( - result.is_err(), - "setup factory admission must survive reinitialization" - ); - assert!(proxy.call::("Mode", ()).await.unwrap()); -} -#[tokio::test] -async fn captured_admission_serializes_concurrent_setup_attempts() { - let _guard = TEST_LOCK.lock().await; - let bus = MemoryBus::new(); - Broker::new().spawn(bus.clone()); - let connection = Connection::connect(bus.connect().await.unwrap()) - .await - .unwrap(); - let setup = setup_factory(); - let (first, second) = tokio::join!( - setup(connection.clone(), Config { restricted: true }), - setup(connection.clone(), Config { restricted: false }) - ); - assert_ne!(first.is_ok(), second.is_ok()); -} - -mod panicking { - use super::*; - fn factory() -> SetupFn { - panic!("factory failed") - } - module_export_static! { - setup_factory = factory, config = Config, worker_threads = 1, - provides = ["ai.tinyhumans.tinybus.PanickingFactory"], - methods = [], signals = [], requires = [], optional = [], lazy = false, - } -} -#[test] -fn factory_panic_is_a_native_initialization_refusal() { - let broker = Broker::new(); - let host = ModuleHost::new(broker); - let error = host - .attach_linked_with_config( - panicking::linked_module().unwrap(), - serde_json::json!({"restricted":true}), - ) - .unwrap_err(); - assert!(matches!(error, tinybus::Error::ModuleRefused { .. })); -} -#[tokio::test] -async fn invalid_initial_configuration_does_not_construct_factory_state() { - let _guard = TEST_LOCK.lock().await; - let before = FACTORIES.load(Ordering::SeqCst); - let host = ModuleHost::new(Broker::new()); - assert!( - host.attach_linked_with_config( - linked_module().unwrap(), - serde_json::json!({"restricted":"invalid"}) - ) - .is_err() - ); - assert_eq!(FACTORIES.load(Ordering::SeqCst), before); -} - -#[test] -fn native_factory_starter_validates_host_before_constructing_state() { - let code = unsafe { - start_reconfigurable_module_with_factory::( - std::ptr::null(), - std::ptr::null_mut(), - 1, - true, - || { - panic!("invalid host must be rejected before factory"); - #[allow(unreachable_code)] - setup_factory() - }, - ) - }; - assert_eq!(code, TB_BAD_ARGUMENT); -} diff --git a/docs/modules/module/README.md b/docs/modules/module/README.md index 733248cb..7747c006 100644 --- a/docs/modules/module/README.md +++ b/docs/modules/module/README.md @@ -217,10 +217,11 @@ See [abi.md](abi.md) for the binary contract and ### Per-instance setup state Configured exports can use `setup_factory = factory` instead of `setup = setup`. -The safe, no-argument factory returns a cloneable setup callback with the same +The safe, no-argument factory returns a setup callback with the same `Fn(Connection, Config) -> Future>` signature. TinyBus calls it once after parsing valid initial configuration, then retains its captured -state for that instance's initial setup and every reinitialization. Separate +state for that instance's initial setup and every reinitialization. The callback +is shared by reference; its Clone implementation cannot detach that state. Separate ABI instances get fresh state; no global registry is needed. This permits a module to capture an immutable admission or mode in an Arc and