fix(module): wait for setup before reporting a module Ready

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<ModuleInfo> {
+        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

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<usize>,
@@ -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<HostCallbackBinding>, 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<HostCallbackBinding>, serde_json::Value)>,
     ) -> Result<(), String> {
         let runtime = self.runtime().await.map_err(|_| {
             format!(

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<Option<SetupHandle>> = 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<LinkedModule> {
+                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!(