From 20f8ed0e374181ad64379c84451e4ece847666ab Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 02:54:13 +0000 Subject: [PATCH 1/5] feat(module): add settle to await module setup before snapshotting Adds a helper that waits until the module owning a bus name has finished setup, faulted, failed or been closed, or until a deadline passes, then returns its snapshot. Loaded is not the same as serving, so callers that reported a module usable at load time could report one about to lose its name. Lazy modules that have not been asked to initialise are returned at once, and None is returned when no loaded module owns the name. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinybus/src/module/host.rs | 33 +++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs index 81cfe0c0..0b71c7f8 100644 --- a/crates/tinybus/src/module/host.rs +++ b/crates/tinybus/src/module/host.rs @@ -407,6 +407,39 @@ impl ModuleHost { modules } + /// Wait until the module that owns `bus_name` has finished setup, then + /// snapshot it. + /// + /// "Loaded" is not "serving": an eager module's `init` can return while + /// its setup (opening a socket, starting a server) is still running, and + /// that setup can fault afterwards. A caller that reports the module usable + /// at load time reports a module that is about to lose its name. This + /// returns once the module signalled ready, faulted, failed or was closed, + /// or after `deadline`; a module still `Initializing` in the snapshot timed + /// out. A lazy module that has not been asked to initialise yet is + /// returned at once, since nothing is running whose outcome could be + /// awaited. `None` when no loaded module owns the name. + pub(crate) async fn settle(&self, bus_name: &str, deadline: Duration) -> Option { + let transport = { + let loaded = self.inner.loaded.lock().expect("module list lock"); + let module = loaded + .iter() + .find(|module| module.info.manifest.bus_name.as_str() == bus_name)?; + if matches!(module.snapshot().state, ModuleState::Resolved) { + return Some(module.snapshot()); + } + module.transport.clone() + }; + let _ = tokio::time::timeout(deadline, transport.wait_ready()).await; + self.inner + .loaded + .lock() + .expect("module list lock") + .iter() + .find(|module| module.info.manifest.bus_name.as_str() == bus_name) + .map(LoadedModule::snapshot) + } + /// Return the broker-assigned unique peer name for a live, ready module. /// /// `module_name` is the module's sanitized descriptor name as returned by From da58a10f25f77a7e700d9ac1ac6e78594fbc244c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 02:55:18 +0000 Subject: [PATCH 2/5] refactor(module): remove unused settle helper from module host The settle method on ModuleHost was no longer called anywhere, so it has been dropped along with its doc comment. Tests covering the setup race it guarded are added in resolve_tests. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinybus/src/module/host.rs | 33 ----- crates/tinybus/src/module/resolve_tests.rs | 165 +++++++++++++++++++++ 2 files changed, 165 insertions(+), 33 deletions(-) diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs index 0b71c7f8..81cfe0c0 100644 --- a/crates/tinybus/src/module/host.rs +++ b/crates/tinybus/src/module/host.rs @@ -407,39 +407,6 @@ impl ModuleHost { modules } - /// Wait until the module that owns `bus_name` has finished setup, then - /// snapshot it. - /// - /// "Loaded" is not "serving": an eager module's `init` can return while - /// its setup (opening a socket, starting a server) is still running, and - /// that setup can fault afterwards. A caller that reports the module usable - /// at load time reports a module that is about to lose its name. This - /// returns once the module signalled ready, faulted, failed or was closed, - /// or after `deadline`; a module still `Initializing` in the snapshot timed - /// out. A lazy module that has not been asked to initialise yet is - /// returned at once, since nothing is running whose outcome could be - /// awaited. `None` when no loaded module owns the name. - pub(crate) async fn settle(&self, bus_name: &str, deadline: Duration) -> Option { - let transport = { - let loaded = self.inner.loaded.lock().expect("module list lock"); - let module = loaded - .iter() - .find(|module| module.info.manifest.bus_name.as_str() == bus_name)?; - if matches!(module.snapshot().state, ModuleState::Resolved) { - return Some(module.snapshot()); - } - module.transport.clone() - }; - let _ = tokio::time::timeout(deadline, transport.wait_ready()).await; - self.inner - .loaded - .lock() - .expect("module list lock") - .iter() - .find(|module| module.info.manifest.bus_name.as_str() == bus_name) - .map(LoadedModule::snapshot) - } - /// Return the broker-assigned unique peer name for a live, ready module. /// /// `module_name` is the module's sanitized descriptor name as returned by diff --git a/crates/tinybus/src/module/resolve_tests.rs b/crates/tinybus/src/module/resolve_tests.rs index 19a076d2..ad410fbd 100644 --- a/crates/tinybus/src/module/resolve_tests.rs +++ b/crates/tinybus/src/module/resolve_tests.rs @@ -1354,6 +1354,171 @@ mod resolver { assert_eq!(f.resolver.mark_terminal(once.clone()), once); } + /// What a module's `init` leaves behind for the test to drive its setup: + /// the host's `fault`/`ready` entry points and context, as plain integers. + struct SetupHandle { + host_ctx: usize, + fault: unsafe extern "C" fn(*mut std::ffi::c_void, *const u8, usize), + ready: unsafe extern "C" fn(*mut std::ffi::c_void), + } + + impl SetupHandle { + fn fault(&self) { + unsafe { (self.fault)(self.host_ctx as *mut _, std::ptr::null(), 0) }; + } + + fn ready(&self) { + unsafe { (self.ready)(self.host_ctx as *mut _) }; + } + } + + /// A module whose `init` returns at once while its setup is still running: + /// it neither calls `ready` nor `fault` until the test says so. + macro_rules! slow_setup_module { + ($init:ident, $factory:ident, $handle:ident, $entered:ident, $id:literal) => { + static $handle: Mutex> = Mutex::new(None); + static $entered: tokio::sync::Notify = tokio::sync::Notify::const_new(); + + unsafe extern "C" fn $init( + host: *const crate::module::abi::TbHostVtable, + module: *mut crate::module::abi::TbModuleVtable, + ) -> i32 { + let host = unsafe { *host }; + unsafe { + *module = crate::module::abi::TbModuleVtable { + size: crate::module::abi::TB_MODULE_VTABLE_BASE_SIZE, + _reserved: 0, + module_ctx: Box::into_raw(Box::new(())).cast(), + deliver: accept_raced_message, + shutdown: stop_raced_module, + reinitialize: None, + } + }; + *$handle.lock().unwrap() = Some(SetupHandle { + host_ctx: host.host_ctx as usize, + fault: host.fault, + ready: host.ready, + }); + $entered.notify_one(); + crate::module::abi::TB_OK + } + + fn $factory() -> crate::Result { + let mut manifest = super::module($id, &[], &[]); + manifest.bus_name = crate::BusName::new(concat!("ai.tinyhumans.test.", $id))?; + manifest.object_path = crate::ObjectPath::new("/ai/tinyhumans/test/Module")?; + Ok(LinkedModule { + descriptor: crate::module::abi::TbAbiDescriptor::current($id, "1.0.0"), + manifest, + init: $init, + }) + } + }; + } + + fn linked_resolver(record: &'static ReleaseRecord, factory: LinkedFactory) -> Resolver { + Resolver::new(ResolverInputs { + enabled: true, + install_dir: None, + allow_download: false, + bundled_dir: None, + sources: HashMap::from([(record.id.to_string(), ModuleSource::Linked)]), + linked: HashMap::from([(record.id.to_string(), factory)]), + search_skip: Vec::new(), + gate: None, + on_resolved: None, + records: vec![record], + terminal_marker: "This is terminal for the running process".into(), + runtime: Arc::new(|| Box::pin(async { Ok(shared_runtime()) })), + module_config: Arc::new(|_| serde_json::json!({})), + reporter: Arc::new(FailureReporter::new(Arc::new(Recording::default()))), + }) + } + + slow_setup_module!( + init_then_fault, + faulting_factory, + FAULTING_HANDLE, + FAULTING_ENTERED, + "resolver_setup_faults" + ); + + #[tokio::test] + async fn a_module_whose_setup_faults_after_load_fails_the_load_with_its_reason() { + let record = record("resolver_setup_faults"); + let resolver = linked_resolver(record, faulting_factory); + let load = tokio::spawn({ + let resolver = resolver.clone(); + async move { resolver.ensure_loaded(record.id).await } + }); + FAULTING_ENTERED.notified().await; + FAULTING_HANDLE.lock().unwrap().as_ref().unwrap().fault(); + + let error = load.await.unwrap().unwrap_err(); + assert!( + error.contains("faulted during setup") + && error.contains("module reported an unrecoverable fault"), + "load must carry the module's reason, got: {error}" + ); + assert_eq!(resolver.state_of(record.id), Availability::Failed); + let again = resolver.ensure_loaded(record.id).await.unwrap_err(); + assert!(again.contains("faulted during setup"), "{again}"); + } + + slow_setup_module!( + init_then_ready, + pending_factory, + PENDING_HANDLE, + PENDING_ENTERED, + "resolver_setup_pending" + ); + + #[tokio::test] + async fn a_module_is_not_ready_until_its_setup_signals_ready() { + let record = record("resolver_setup_pending"); + let resolver = linked_resolver(record, pending_factory); + let load = tokio::spawn({ + let resolver = resolver.clone(); + async move { resolver.ensure_loaded(record.id).await } + }); + PENDING_ENTERED.notified().await; + assert!( + !load.is_finished() && resolver.state_of(record.id) == Availability::Loading, + "the module was reported ready while its setup was still running" + ); + PENDING_HANDLE.lock().unwrap().as_ref().unwrap().ready(); + + load.await.unwrap().unwrap(); + assert_eq!(resolver.state_of(record.id), Availability::Ready); + } + + slow_setup_module!( + init_ready_later_faults, + late_fault_factory, + LATE_HANDLE, + LATE_ENTERED, + "resolver_late_fault" + ); + + #[tokio::test] + async fn a_cached_ready_is_not_served_once_the_module_has_faulted() { + let record = record("resolver_late_fault"); + let resolver = linked_resolver(record, late_fault_factory); + let load = tokio::spawn({ + let resolver = resolver.clone(); + async move { resolver.ensure_loaded(record.id).await } + }); + LATE_ENTERED.notified().await; + LATE_HANDLE.lock().unwrap().as_ref().unwrap().ready(); + load.await.unwrap().unwrap(); + assert_eq!(resolver.state_of(record.id), Availability::Ready); + + LATE_HANDLE.lock().unwrap().as_ref().unwrap().fault(); + let error = resolver.ensure_loaded(record.id).await.unwrap_err(); + assert!(error.contains("faulted during setup"), "{error}"); + assert_eq!(resolver.state_of(record.id), Availability::Failed); + } + #[test] fn wire_forms_of_source_and_availability_are_stable() { assert_eq!( From 0ca3de2b0e4be57d1ff077f7c4f6fb769fb06dbd Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 02:55:36 +0000 Subject: [PATCH 3/5] test(resolve): bound the pending-setup wait with a deadline The test that checks a module is not ready until its setup signals ready now uses ensure_loaded_within with a short deadline instead of spawning the load and polling is_finished, so the wait is bounded rather than relying on scheduling. The module is still asserted to be Loading while setup is pending and Ready once the handle signals. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinybus/src/module/resolve_tests.rs | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/crates/tinybus/src/module/resolve_tests.rs b/crates/tinybus/src/module/resolve_tests.rs index ad410fbd..348d1caf 100644 --- a/crates/tinybus/src/module/resolve_tests.rs +++ b/crates/tinybus/src/module/resolve_tests.rs @@ -1477,18 +1477,20 @@ mod resolver { async fn a_module_is_not_ready_until_its_setup_signals_ready() { let record = record("resolver_setup_pending"); let resolver = linked_resolver(record, pending_factory); - let load = tokio::spawn({ - let resolver = resolver.clone(); - async move { resolver.ensure_loaded(record.id).await } - }); - PENDING_ENTERED.notified().await; - assert!( - !load.is_finished() && resolver.state_of(record.id) == Availability::Loading, + // A deadline, not a sleep: nothing signals ready, so a bounded wait + // must run out while the module is still loading. + assert_eq!( + resolver + .ensure_loaded_within(record.id, Some(Duration::from_millis(200))) + .await, + Err(LoadError::StillLoading), "the module was reported ready while its setup was still running" ); + assert_eq!(resolver.state_of(record.id), Availability::Loading); + PENDING_ENTERED.notified().await; PENDING_HANDLE.lock().unwrap().as_ref().unwrap().ready(); - load.await.unwrap().unwrap(); + resolver.ensure_loaded(record.id).await.unwrap(); assert_eq!(resolver.state_of(record.id), Availability::Ready); } From a14d88fce7f7cb5b08b22ec85aecad91d7987adf Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 02:55:57 +0000 Subject: [PATCH 4/5] fix(module): wait for module setup before reporting it usable Resolution now waits for a loaded module to finish its setup before treating it as serving, since an eager module's init can return while its server or socket is still starting and may fault afterwards. A cached Ready is re-checked against the host so a module that faulted after being reported ready surfaces its own reason instead of a bare Unavailable. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinybus/src/module/host.rs | 33 +++++++++ crates/tinybus/src/module/resolve.rs | 83 +++++++++++++++++++++- crates/tinybus/src/module/resolve_tests.rs | 8 ++- 3 files changed, 120 insertions(+), 4 deletions(-) diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs index 81cfe0c0..0b71c7f8 100644 --- a/crates/tinybus/src/module/host.rs +++ b/crates/tinybus/src/module/host.rs @@ -407,6 +407,39 @@ impl ModuleHost { modules } + /// Wait until the module that owns `bus_name` has finished setup, then + /// snapshot it. + /// + /// "Loaded" is not "serving": an eager module's `init` can return while + /// its setup (opening a socket, starting a server) is still running, and + /// that setup can fault afterwards. A caller that reports the module usable + /// at load time reports a module that is about to lose its name. This + /// returns once the module signalled ready, faulted, failed or was closed, + /// or after `deadline`; a module still `Initializing` in the snapshot timed + /// out. A lazy module that has not been asked to initialise yet is + /// returned at once, since nothing is running whose outcome could be + /// awaited. `None` when no loaded module owns the name. + pub(crate) async fn settle(&self, bus_name: &str, deadline: Duration) -> Option { + let transport = { + let loaded = self.inner.loaded.lock().expect("module list lock"); + let module = loaded + .iter() + .find(|module| module.info.manifest.bus_name.as_str() == bus_name)?; + if matches!(module.snapshot().state, ModuleState::Resolved) { + return Some(module.snapshot()); + } + module.transport.clone() + }; + let _ = tokio::time::timeout(deadline, transport.wait_ready()).await; + self.inner + .loaded + .lock() + .expect("module list lock") + .iter() + .find(|module| module.info.manifest.bus_name.as_str() == bus_name) + .map(LoadedModule::snapshot) + } + /// Return the broker-assigned unique peer name for a live, ready module. /// /// `module_name` is the module's sanitized descriptor name as returned by diff --git a/crates/tinybus/src/module/resolve.rs b/crates/tinybus/src/module/resolve.rs index b6f00de4..c1a47dc0 100644 --- a/crates/tinybus/src/module/resolve.rs +++ b/crates/tinybus/src/module/resolve.rs @@ -67,6 +67,35 @@ fn is_live_match(info: &ModuleInfo, record: &ReleaseRecord) -> bool { ) } +/// How long a resolution waits for a loaded module to finish setup before it +/// gives up on it. Setup (starting a server, opening a socket) runs after the +/// module's `init` returns, so loading alone says nothing about it. +const SETUP_DEADLINE: Duration = Duration::from_secs(30); + +/// Why `info` is not serving, or `Ok` when it is serving or has nothing running +/// whose outcome could be awaited (a lazy module that has not been called yet). +fn setup_outcome(info: &ModuleInfo, id: &str) -> Result<(), String> { + match &info.state { + ModuleState::Faulted { reason } | ModuleState::Failed { reason } => Err(format!( + "module '{id}' faulted during setup: {reason}" + )), + ModuleState::Rejected { reason } | ModuleState::Unresolved { reason } => { + Err(format!("module '{id}' was refused: {reason}")) + } + ModuleState::Stopped | ModuleState::Disabled => { + Err(format!("module '{id}' is not running")) + } + ModuleState::Initializing => Err(format!( + "module '{id}' did not finish setup within {} seconds", + SETUP_DEADLINE.as_secs() + )), + ModuleState::Discovered + | ModuleState::Resolved + | ModuleState::Ready + | ModuleState::Serving => Ok(()), + } +} + /// Result indexes refer to the input manifest slice. pub(crate) struct LoadOrder { pub(crate) order: Vec, @@ -516,7 +545,7 @@ impl Resolver { "module '{id}' is already loaded without this callback binding" ))) } else { - Ok(()) + self.still_serving(record).await }; } Claim::Done(Resolution::Failed(reason)) => { @@ -542,6 +571,7 @@ impl Resolver { }; match resolution::global().wait(id, receiver, within).await { Waited::Ready => { + self.still_serving(record).await?; if let Some((binding, _)) = callbacks.as_ref() { let runtime = self.runtime().await.map_err(|_| { LoadError::Failed(format!( @@ -564,6 +594,32 @@ impl Resolver { } } + /// Re-check a cached `Ready` against the host. A module can fault after it + /// was reported ready; the host then drops its name, so serving the cached + /// outcome would turn every later call into a bare `Unavailable`. The slot + /// settles as failed instead and the caller gets the module's reason. + async fn still_serving(&self, record: &'static ReleaseRecord) -> Result<(), LoadError> { + let Ok(runtime) = self.runtime().await else { + return Ok(()); + }; + let faulted = runtime.host().list().into_iter().find_map(|info| { + (info.manifest.bus_name.as_str() == record.bus_name + && matches!( + info.state, + ModuleState::Faulted { .. } | ModuleState::Failed { .. } + )) + .then(|| setup_outcome(&info, record.id).err()) + .flatten() + }); + let Some(reason) = faulted else { + return Ok(()); + }; + let reason = self.mark_terminal(reason); + resolution::global().mark_faulted(record.id, reason.clone()); + self.report(record, Reason::ResolutionFailed); + Err(LoadError::Failed(reason)) + } + /// The state of `id` as the resolution table reports it, without touching it. #[must_use] pub fn state_of(&self, id: &str) -> Availability { @@ -669,11 +725,34 @@ impl Resolver { } } - /// Get `record` serving: loaded, override, search path, release cache, prune. + /// Get `record` serving: load it, then wait for its setup to finish. + /// + /// Every rung (file, linked, search path, release cache, already loaded) + /// ends here, so none of them can report `Ready` for a module whose setup is + /// still running or has faulted. async fn ladder( &self, record: &'static ReleaseRecord, callbacks: Option<(Arc, serde_json::Value)>, + ) -> Result<(), String> { + self.load_rungs(record, callbacks).await?; + let runtime = self.runtime().await.map_err(|_| { + format!( + "module '{}' is unavailable: the module bus could not start", + record.id + ) + })?; + match runtime.host().settle(record.bus_name, SETUP_DEADLINE).await { + Some(info) => setup_outcome(&info, record.id), + None => Ok(()), + } + } + + /// Load: loaded, override, search path, release cache, prune. + async fn load_rungs( + &self, + record: &'static ReleaseRecord, + callbacks: Option<(Arc, serde_json::Value)>, ) -> Result<(), String> { let runtime = self.runtime().await.map_err(|_| { format!( diff --git a/crates/tinybus/src/module/resolve_tests.rs b/crates/tinybus/src/module/resolve_tests.rs index 348d1caf..62118752 100644 --- a/crates/tinybus/src/module/resolve_tests.rs +++ b/crates/tinybus/src/module/resolve_tests.rs @@ -1477,6 +1477,11 @@ mod resolver { async fn a_module_is_not_ready_until_its_setup_signals_ready() { let record = record("resolver_setup_pending"); let resolver = linked_resolver(record, pending_factory); + let first = tokio::spawn({ + let resolver = resolver.clone(); + async move { resolver.ensure_loaded(record.id).await } + }); + PENDING_ENTERED.notified().await; // A deadline, not a sleep: nothing signals ready, so a bounded wait // must run out while the module is still loading. assert_eq!( @@ -1487,10 +1492,9 @@ mod resolver { "the module was reported ready while its setup was still running" ); assert_eq!(resolver.state_of(record.id), Availability::Loading); - PENDING_ENTERED.notified().await; PENDING_HANDLE.lock().unwrap().as_ref().unwrap().ready(); - resolver.ensure_loaded(record.id).await.unwrap(); + first.await.unwrap().unwrap(); assert_eq!(resolver.state_of(record.id), Availability::Ready); } From b12f71ab4b21d078b5fe7776d1e199545175e50a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 02:56:49 +0000 Subject: [PATCH 5/5] docs(module): document setup readiness and fault reporting The module resolution docs now explain that a module only counts as serving once its setup has finished, and that a cached Ready is re-checked against the host so later faults surface as failures instead of bare Unavailable. The setup_outcome match arm was also reformatted without changing behaviour. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinybus/src/module/resolve.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/tinybus/src/module/resolve.rs b/crates/tinybus/src/module/resolve.rs index c1a47dc0..dfa577e3 100644 --- a/crates/tinybus/src/module/resolve.rs +++ b/crates/tinybus/src/module/resolve.rs @@ -14,6 +14,12 @@ //! search path, then the release cache (verified against the host's pinned //! digest, downloaded on a miss when allowed), then pruning of stale versions. //! +//! A module counts as serving only once its setup has finished: every rung ends +//! by waiting for the module to signal ready, and a fault or failure during +//! setup fails the load with the module's reason. A cached `Ready` is re-checked +//! against the host on each call, so a module that faults later is reported as +//! failed rather than answering every call with a bare `Unavailable`. +//! //! Each module has one slot in [`super::resolution`]. The first caller runs the //! resolution as a process-lifetime task and every other caller waits on its //! outcome; two modules never queue behind each other. @@ -76,9 +82,9 @@ const SETUP_DEADLINE: Duration = Duration::from_secs(30); /// whose outcome could be awaited (a lazy module that has not been called yet). fn setup_outcome(info: &ModuleInfo, id: &str) -> Result<(), String> { match &info.state { - ModuleState::Faulted { reason } | ModuleState::Failed { reason } => Err(format!( - "module '{id}' faulted during setup: {reason}" - )), + ModuleState::Faulted { reason } | ModuleState::Failed { reason } => { + Err(format!("module '{id}' faulted during setup: {reason}")) + } ModuleState::Rejected { reason } | ModuleState::Unresolved { reason } => { Err(format!("module '{id}' was refused: {reason}")) }