From 89c5e0aaacd97dc5fe9cc3716031f2b6df40af30 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:19:37 +0300 Subject: [PATCH 1/8] fix(module): share setup factory callback across reinitialization The reinitialize callback is now held behind a mutex so shutdown can take it and in-flight reinitializations finish before the runtime stops, and late calls see a tombstone. Setup factories are shared by reference instead of cloned, so a Clone implementation that deep-copies interior state can no longer detach initial admission from reinitialization. Start argument validation was also factored out and applied before configuration parsing or Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 94 +++++++--- crates/tinybus-module/src/lib_tests.rs | 16 +- crates/tinybus-module/src/lifecycle_tests.rs | 170 ++++++++++++++++++ .../tinybus-module/src/setup_factory_tests.rs | 126 +++++++++++++ docs/modules/module/README.md | 5 +- 5 files changed, 380 insertions(+), 31 deletions(-) create mode 100644 crates/tinybus-module/src/lifecycle_tests.rs diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 396be457..359b86dd 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -326,7 +326,7 @@ struct RuntimeState { inbound: StdMutex>>, host: HostCalls, runtime: StdMutex>, - reinitialize: Option, + reinitialize: StdMutex>, } type Reinitialize = Box i32 + Send + Sync>; @@ -363,6 +363,13 @@ unsafe extern "C" fn shutdown(ctx: *mut c_void, deadline_ms: u64) -> i32 { return TB_BAD_ARGUMENT; } let state = unsafe { &*(ctx.cast::()) }; + // Hold the callback lifecycle lock while stopping the runtime. An + // in-flight reinitializer finishes first; late calls see a tombstone. + let mut callback = state + .reinitialize + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + callback.take(); state.inbound.lock().expect("module inbound lock").take(); match state.runtime.lock().expect("module runtime lock").take() { Some(runtime) => { @@ -383,7 +390,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 +507,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 +539,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); @@ -633,7 +652,7 @@ where host, inbound: StdMutex::new(Some(inbound_tx)), runtime: StdMutex::new(Some(runtime)), - reinitialize: reinitializer, + reinitialize: StdMutex::new(reinitializer), }); let state = Box::into_raw(state).cast::(); unsafe { write_module_vtable(out, out_capacity, state, supports_reinitialize) }; @@ -775,6 +794,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 +810,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 +819,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 +842,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 +851,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 +882,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 +950,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. @@ -1144,3 +1192,7 @@ mod static_link_tests; #[cfg(test)] #[path = "setup_factory_tests.rs"] mod setup_factory_test; + +#[cfg(test)] +#[path = "lifecycle_tests.rs"] +mod lifecycle_test; diff --git a/crates/tinybus-module/src/lib_tests.rs b/crates/tinybus-module/src/lib_tests.rs index 663af561..92da9175 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: 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: 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: 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: 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: 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: 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: 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: StdMutex::new(None), }; let transport = Arc::new(ModuleTransport { host: HostCalls(table), diff --git a/crates/tinybus-module/src/lifecycle_tests.rs b/crates/tinybus-module/src/lifecycle_tests.rs new file mode 100644 index 00000000..beb8bca0 --- /dev/null +++ b/crates/tinybus-module/src/lifecycle_tests.rs @@ -0,0 +1,170 @@ +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: 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: 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: 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); +} diff --git a/crates/tinybus-module/src/setup_factory_tests.rs b/crates/tinybus-module/src/setup_factory_tests.rs index f41a6adb..e5828e7e 100644 --- a/crates/tinybus-module/src/setup_factory_tests.rs +++ b/crates/tinybus-module/src/setup_factory_tests.rs @@ -206,3 +206,129 @@ fn native_factory_starter_validates_host_before_constructing_state() { }; 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" + ); +} 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 From 4f286c40cfd80344f15194c81f48f8df10ac5061 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:21:41 +0300 Subject: [PATCH 2/8] fix(tinybus-module): dispose reinitialize callback on terminal setup failure The reinitialize callback is now shared behind an Arc so the setup task can dispose it when setup fails terminally, since such a failure may never receive an ABI shutdown call. Disposal happens under the same lifecycle exclusion used by shutdown, preventing the callback from being leaked or invoked after fault. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 33 ++++++++++++++------ crates/tinybus-module/src/lib_tests.rs | 16 +++++----- crates/tinybus-module/src/lifecycle_tests.rs | 12 +++---- 3 files changed, 38 insertions(+), 23 deletions(-) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 359b86dd..b59289f4 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -326,7 +326,7 @@ struct RuntimeState { inbound: StdMutex>>, host: HostCalls, runtime: StdMutex>, - reinitialize: StdMutex>, + reinitialize: Arc>>, } type Reinitialize = Box i32 + Send + Sync>; @@ -606,7 +606,12 @@ 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) { Ok(connection) => { @@ -638,21 +643,31 @@ where } Err(error) => Err(error), }; - if let Err(error) = outcome { - tracing::error!(error = %error, "tinybus module failed"); + outcome + }); + 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. + failed_setup_callback + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + 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)), runtime: StdMutex::new(Some(runtime)), - reinitialize: StdMutex::new(reinitializer), + reinitialize: reinitializer, }); let state = Box::into_raw(state).cast::(); unsafe { write_module_vtable(out, out_capacity, state, supports_reinitialize) }; diff --git a/crates/tinybus-module/src/lib_tests.rs b/crates/tinybus-module/src/lib_tests.rs index 92da9175..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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(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: StdMutex::new(None), + reinitialize: Arc::new(StdMutex::new(None)), }; let transport = Arc::new(ModuleTransport { host: HostCalls(table), diff --git a/crates/tinybus-module/src/lifecycle_tests.rs b/crates/tinybus-module/src/lifecycle_tests.rs index beb8bca0..47943095 100644 --- a/crates/tinybus-module/src/lifecycle_tests.rs +++ b/crates/tinybus-module/src/lifecycle_tests.rs @@ -40,10 +40,10 @@ fn shutdown_releases_captured_callback_while_retaining_closed_tombstone() { inbound: StdMutex::new(None), host: HostCalls(host()), runtime: StdMutex::new(Some(runtime)), - reinitialize: StdMutex::new(Some(Box::new(move |_, _| { + 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); @@ -112,7 +112,7 @@ fn shutdown_excludes_active_reinitializer_before_disposing_capture() { inbound: StdMutex::new(None), host: HostCalls(host()), runtime: StdMutex::new(Some(tokio::runtime::Runtime::new().unwrap())), - reinitialize: StdMutex::new(Some(Box::new(move |_, _| { + reinitialize: Arc::new(StdMutex::new(Some(Box::new(move |_, _| { let _ = &token; entered.send(()).unwrap(); released @@ -121,7 +121,7 @@ fn shutdown_excludes_active_reinitializer_before_disposing_capture() { .recv_timeout(Duration::from_secs(5)) .unwrap(); TB_OK - }))), + })))), }); let active = Arc::clone(&state); let reinitializer = std::thread::spawn(move || unsafe { @@ -155,10 +155,10 @@ fn shutdown_disposes_callback_even_after_reinitializer_panics() { inbound: StdMutex::new(None), host: HostCalls(host()), runtime: StdMutex::new(None), - reinitialize: StdMutex::new(Some(Box::new(move |_, _| { + 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!( From c3976618ef3fa5887d75acecc68d429a959e0784 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:22:32 +0300 Subject: [PATCH 3/8] test(module): cover factory capture release on shutdown and setup panic Add tests asserting that the setup factory's captured state is dropped exactly once when a module shuts down after a successful initial setup, and when initial setup panics and the module transitions to Faulted. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../tinybus-module/src/setup_factory_tests.rs | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/crates/tinybus-module/src/setup_factory_tests.rs b/crates/tinybus-module/src/setup_factory_tests.rs index e5828e7e..ae3a5266 100644 --- a/crates/tinybus-module/src/setup_factory_tests.rs +++ b/crates/tinybus-module/src/setup_factory_tests.rs @@ -332,3 +332,103 @@ async fn failed_initial_setup_releases_factory_capture_when_host_stops_instance( "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); +} From 16a98cff106bd8f957b10bbf2dac7add8f5c2c03 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:30:52 +0300 Subject: [PATCH 4/8] fix(tinybus-module): release callback lock before joining runtime tasks The shutdown path held the callback lifecycle lock while stopping the runtime, which could deadlock against failed-setup cleanup that also touches the slot. The lock is now dropped once the callback is taken, so late calls still see a tombstone but task joining can proceed. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index b59289f4..57575625 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -363,13 +363,15 @@ unsafe extern "C" fn shutdown(ctx: *mut c_void, deadline_ms: u64) -> i32 { return TB_BAD_ARGUMENT; } let state = unsafe { &*(ctx.cast::()) }; - // Hold the callback lifecycle lock while stopping the runtime. An - // in-flight reinitializer finishes first; late calls see a tombstone. + // 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) => { From 43de92568b4cc59e81cbce03df1d705408d31d7f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:34:03 +0300 Subject: [PATCH 5/8] refactor(tinybus-module): return handshake match directly The setup task bound the handshake match to a local and returned it on the next line, so the binding is dropped and the match is returned directly. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 57575625..732e9c06 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -615,7 +615,7 @@ where 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 @@ -644,8 +644,7 @@ where } } Err(error) => Err(error), - }; - outcome + } }); runtime.spawn(async move { let outcome = setup_task.await; From bc36c191418493c69c9ab04d6342c848ade7e961 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:45:57 +0300 Subject: [PATCH 6/8] fix(tinybus-module): avoid blocking runtime worker during failed setup disposal Move the failed-setup callback disposal into an async helper so the synchronous mutex lock is not held across an await point on the runtime worker. Add a regression test asserting that disposal does not occupy the sole timer worker. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 11 +++--- crates/tinybus-module/src/lifecycle_tests.rs | 36 ++++++++++++++++++++ 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 732e9c06..d9f6218b 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -330,6 +330,12 @@ struct RuntimeState { } type Reinitialize = Box i32 + Send + Sync>; +async fn dispose_failed_setup_callback(callback: Arc>>) { + callback + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); +} type ReinitializeFactory = Box< dyn FnOnce(tokio::runtime::Handle, Arc>>) -> Reinitialize + Send, >; @@ -651,10 +657,7 @@ where if !matches!(outcome, Ok(Ok(()))) { // A terminal setup failure may never receive ABI shutdown. // Dispose its callback under the same lifecycle exclusion. - failed_setup_callback - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take(); + dispose_failed_setup_callback(failed_setup_callback).await; match outcome { Ok(Err(error)) => tracing::error!(error = %error, "tinybus module failed"), Err(_) => tracing::error!("tinybus module setup task failed"), diff --git a/crates/tinybus-module/src/lifecycle_tests.rs b/crates/tinybus-module/src/lifecycle_tests.rs index 47943095..829c426a 100644 --- a/crates/tinybus-module/src/lifecycle_tests.rs +++ b/crates/tinybus-module/src/lifecycle_tests.rs @@ -168,3 +168,39 @@ fn shutdown_disposes_callback_even_after_reinitializer_panics() { 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; + 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" + ); +} From 56514a1dd3737ce8e7078ed2b8df193693b513c0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:48:49 +0300 Subject: [PATCH 7/8] fix(tinybus-module): dispose failed setup callback off the runtime worker The failed-setup callback disposal now runs on a blocking thread so that waiting for the reinitialization lock cannot occupy the sole runtime worker needed to drive the setup future and its timer. Test modules were renamed to match the lib_ prefix convention. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 18 ++++++++++++------ ...fecycle_tests.rs => lib_lifecycle_tests.rs} | 0 ...ory_tests.rs => lib_setup_factory_tests.rs} | 0 3 files changed, 12 insertions(+), 6 deletions(-) rename crates/tinybus-module/src/{lifecycle_tests.rs => lib_lifecycle_tests.rs} (100%) rename crates/tinybus-module/src/{setup_factory_tests.rs => lib_setup_factory_tests.rs} (100%) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index d9f6218b..d6dc705b 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -331,10 +331,16 @@ struct RuntimeState { type Reinitialize = Box i32 + Send + Sync>; async fn dispose_failed_setup_callback(callback: Arc>>) { - callback - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take(); + // 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. + let _ = tokio::task::spawn_blocking(move || { + callback + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + }) + .await; } type ReinitializeFactory = Box< dyn FnOnce(tokio::runtime::Handle, Arc>>) -> Reinitialize + Send, @@ -1209,9 +1215,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 = "lifecycle_tests.rs"] +#[path = "lib_lifecycle_tests.rs"] mod lifecycle_test; diff --git a/crates/tinybus-module/src/lifecycle_tests.rs b/crates/tinybus-module/src/lib_lifecycle_tests.rs similarity index 100% rename from crates/tinybus-module/src/lifecycle_tests.rs rename to crates/tinybus-module/src/lib_lifecycle_tests.rs diff --git a/crates/tinybus-module/src/setup_factory_tests.rs b/crates/tinybus-module/src/lib_setup_factory_tests.rs similarity index 100% rename from crates/tinybus-module/src/setup_factory_tests.rs rename to crates/tinybus-module/src/lib_setup_factory_tests.rs From eb1c7448b9cdea72dd4e9d4281f8f0739f0308df Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 20:49:12 +0300 Subject: [PATCH 8/8] fix(tinybus-module): propagate failed-setup callback disposal errors The failed-setup callback disposal now returns an error when its blocking task panics instead of silently discarding the join result, and the caller logs that failure. A regression test covers a compiled single-worker module whose timer must keep advancing before the fault is reported. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus-module/src/lib.rs | 16 ++- .../tinybus-module/src/lib_lifecycle_tests.rs | 2 +- .../src/lib_setup_factory_tests.rs | 104 ++++++++++++++++++ 3 files changed, 117 insertions(+), 5 deletions(-) diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index d6dc705b..5864d335 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -330,17 +330,20 @@ struct RuntimeState { } type Reinitialize = Box i32 + Send + Sync>; -async fn dispose_failed_setup_callback(callback: Arc>>) { +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. - let _ = tokio::task::spawn_blocking(move || { + tokio::task::spawn_blocking(move || { callback .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .take(); }) - .await; + .await + .map_err(|_| Error::failed("module failed-setup callback disposal panicked")) } type ReinitializeFactory = Box< dyn FnOnce(tokio::runtime::Handle, Arc>>) -> Reinitialize + Send, @@ -663,7 +666,12 @@ where if !matches!(outcome, Ok(Ok(()))) { // A terminal setup failure may never receive ABI shutdown. // Dispose its callback under the same lifecycle exclusion. - dispose_failed_setup_callback(failed_setup_callback).await; + 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"), diff --git a/crates/tinybus-module/src/lib_lifecycle_tests.rs b/crates/tinybus-module/src/lib_lifecycle_tests.rs index 829c426a..89756fc8 100644 --- a/crates/tinybus-module/src/lib_lifecycle_tests.rs +++ b/crates/tinybus-module/src/lib_lifecycle_tests.rs @@ -184,7 +184,7 @@ fn failed_setup_disposal_keeps_single_runtime_worker_available() { let dispose = callback.clone(); runtime.spawn(async move { entered_tx.send(()).unwrap(); - dispose_failed_setup_callback(dispose).await; + dispose_failed_setup_callback(dispose).await.unwrap(); done_tx.send(()).unwrap(); }); entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); diff --git a/crates/tinybus-module/src/lib_setup_factory_tests.rs b/crates/tinybus-module/src/lib_setup_factory_tests.rs index ae3a5266..1fe9cb9d 100644 --- a/crates/tinybus-module/src/lib_setup_factory_tests.rs +++ b/crates/tinybus-module/src/lib_setup_factory_tests.rs @@ -432,3 +432,107 @@ async fn panicking_initial_setup_releases_factory_capture_and_reports_fault() { .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" + ); +}