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..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. @@ -67,6 +73,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 +551,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 +577,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 +600,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 +731,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 19a076d2..62118752 100644 --- a/crates/tinybus/src/module/resolve_tests.rs +++ b/crates/tinybus/src/module/resolve_tests.rs @@ -1354,6 +1354,177 @@ 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 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!( + 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_HANDLE.lock().unwrap().as_ref().unwrap().ready(); + + first.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!(