fix(supervisor): bound provider install by dispatch deadline

crates/nexum-runtime/src/host/extension.rs

@@ -171,6 +171,8 @@ pub trait ProviderKind<T: RuntimeTypes>: Send + Sync + 'static {
 
     /// Instantiate and install one provider; [`Installed::Dead`] is a failed
     /// guest `init`, `Err` a boot error.
+    /// The host bounds this call by the dispatch deadline and drops the future
+    /// at its next await, so publish into `service` only after the last await.
     async fn install(
         &self,
         instance: ProviderInstance<'_, T>,

crates/nexum-runtime/src/supervisor/load.rs

@@ -183,6 +183,8 @@ pub(super) async fn instantiate_module<T: RuntimeTypes>(
 
 /// Builds the kind's linker and installs on the given store; a `Dead`
 /// verdict carries no error, its meaning stays with the caller.
+/// `event_deadline` bounds the whole install (instantiation, guest `init`,
+/// extension wiring), not only the guest call a module's bound covers.
 pub(super) async fn install_provider<T: RuntimeTypes>(
     shared: &Shared<T>,
     row: &ProviderRow<T>,
@@ -193,8 +195,13 @@ pub(super) async fn install_provider<T: RuntimeTypes>(
 ) -> Result<Installed> {
     let (kind, service) = row;
     let linker = build_provider_linker::<T>(&shared.engine, kind.as_ref())?;
-    kind.install(seed.instance(&linker, sections, store, liveness), service)
-        .await
+    with_dispatch_deadline(
+        seed.event_deadline,
+        kind.install(seed.instance(&linker, sections, store, liveness), service),
+    )
+    .await
+    .map_err(Error::from)
+    .with_context(|| format!("provider kind {} did not install in time", kind.kind()))?
 }
 
 /// A failed `init` loads the module dead; the dispatcher skips it.

crates/nexum-runtime/src/supervisor/tests/lifecycle.rs

@@ -539,12 +539,13 @@ impl Extension<CoreRuntime> for ScriptedExtension {
     }
 }
 
-/// The shared wiring the sweep reinstalls through, with the scripted kind
-/// registered. The `TempDir` keeps the local store alive.
-fn scripted_shared(live: &Arc<AtomicBool>) -> (tempfile::TempDir, Shared<CoreRuntime>) {
+/// The shared wiring the sweep reinstalls through, with the given
+/// extension's kind registered. The `TempDir` keeps the local store alive.
+fn kind_shared(
+    extension: Arc<dyn Extension<CoreRuntime>>,
+) -> (tempfile::TempDir, Shared<CoreRuntime>) {
     let (dir, store) = temp_local_store();
-    let extensions: Vec<Arc<dyn Extension<CoreRuntime>>> =
-        vec![Arc::new(ScriptedExtension(live.clone()))];
+    let extensions = vec![extension];
     let services = HostServices::from_extensions(&extensions).expect("services");
     let kinds =
         crate::supervisor::admission::provider_kinds(&extensions, &services).expect("kinds");
@@ -560,13 +561,16 @@ fn scripted_shared(live: &Arc<AtomicBool>) -> (tempfile::TempDir, Shared<CoreRun
 }
 
 /// A booted provider at run 0. The component is an empty valid one: the
-/// scripted kind never instantiates it.
-fn scripted_provider(engine: &wasmtime::Engine) -> crate::supervisor::load::LoadedProvider {
+/// test kinds never instantiate it.
+fn provider_at_run_zero(
+    engine: &wasmtime::Engine,
+    kind: &'static str,
+) -> crate::supervisor::load::LoadedProvider {
     const EMPTY_COMPONENT: &[u8] = b"(component)";
     let limits = ModuleLimits::default();
     crate::supervisor::load::LoadedProvider {
         name: "scripted".into(),
-        kind: "scripted-adapter",
+        kind,
         sections: manifest::ExtensionSections::default(),
         seed: crate::supervisor::load::Seed {
             artifact: crate::supervisor::load::CachedArtifact {
@@ -597,8 +601,8 @@ fn scripted_provider(engine: &wasmtime::Engine) -> crate::supervisor::load::Load
 #[tokio::test]
 async fn a_dead_provider_reinstall_defers_without_committing_a_run() {
     let live = Arc::new(AtomicBool::new(false));
-    let (_dir, shared) = scripted_shared(&live);
-    let mut provider = scripted_provider(&shared.engine);
+    let (_dir, shared) = kind_shared(Arc::new(ScriptedExtension(live.clone())));
+    let mut provider = provider_at_run_zero(&shared.engine, "scripted-adapter");
     let policy = crate::runtime::poison_policy::PoisonPolicy::new(9, Duration::from_secs(600));
 
     // The actor trapped: the sweep discovers the death through the shared
@@ -635,3 +639,135 @@ async fn a_dead_provider_reinstall_defers_without_committing_a_run() {
         "a reinstall is a fresh instance, so the curve resets",
     );
 }
+
+/// Distinct from every default, so an assertion on elapsed paused time pins
+/// which duration bounded the install.
+const INSTALL_DEADLINE: Duration = Duration::from_secs(7);
+
+/// Long enough that a wrapped install always wins the race; a hung unwrapped
+/// one rides this instead and fails the test.
+const OUTER_TIMEOUT: Duration = Duration::from_secs(3_600);
+
+/// A provider kind whose `install` never returns, for the deadline gate.
+struct HangingKind;
+
+#[async_trait::async_trait]
+impl ProviderKind<CoreRuntime> for HangingKind {
+    fn kind(&self) -> &'static str {
+        "hanging-adapter"
+    }
+
+    fn link(&self, _linker: &mut Linker<HostState<CoreRuntime>>) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    async fn install(
+        &self,
+        _instance: ProviderInstance<'_, CoreRuntime>,
+        _service: &Arc<dyn HostService>,
+    ) -> anyhow::Result<Installed> {
+        std::future::pending().await
+    }
+}
+
+struct HangingExtension;
+
+impl Extension<CoreRuntime> for HangingExtension {
+    fn namespace(&self) -> &'static str {
+        "hanging"
+    }
+
+    fn capabilities(&self) -> manifest::NamespaceCaps {
+        manifest::NamespaceCaps {
+            prefix: "test:hanging/",
+            ifaces: &[],
+        }
+    }
+
+    fn link(&self, _linker: &mut Linker<HostState<CoreRuntime>>) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    fn service(&self) -> Option<Arc<dyn HostService>> {
+        Some(Arc::new(ScriptedService))
+    }
+
+    fn provider(&self) -> Option<Box<dyn ProviderKind<CoreRuntime>>> {
+        Some(Box::new(HangingKind))
+    }
+}
+
+/// A hung `install` fails by the dispatch deadline and defers, instead of
+/// parking the sweep (and with it all dispatch) forever.
+#[tokio::test(start_paused = true)]
+async fn a_hanging_provider_install_fails_by_deadline() {
+    let (_dir, shared) = kind_shared(Arc::new(HangingExtension));
+    let mut provider = provider_at_run_zero(&shared.engine, "hanging-adapter");
+    provider.seed.event_deadline = INSTALL_DEADLINE;
+    let policy = crate::runtime::poison_policy::PoisonPolicy::new(9, Duration::from_secs(600));
+
+    provider.liveness.mark_dead();
+    let died_at = provider.liveness.dead_since().expect("marked dead");
+    let now = died_at + Duration::from_secs(5);
+    let started = tokio::time::Instant::now();
+    // Paused time auto-advances only through timers; an unwrapped hung
+    // install would ride this outer timeout instead of its own deadline.
+    tokio::time::timeout(
+        OUTER_TIMEOUT,
+        sweep(&shared, std::slice::from_mut(&mut provider), policy, now),
+    )
+    .await
+    .expect("sweep completed: the install deadline bounded the hung install");
+
+    assert_eq!(
+        started.elapsed(),
+        INSTALL_DEADLINE,
+        "the install rode the seed's deadline, not another timer",
+    );
+    assert_eq!(provider.run.seq, 0, "a timed-out install commits no run");
+    assert!(
+        !provider.liveness.is_alive(),
+        "a timed-out install leaves the liveness dead",
+    );
+    assert_eq!(
+        provider.health.failure_count(),
+        2,
+        "one recorded trap plus one deferred restart",
+    );
+    assert!(
+        provider.health.due_restart(now + Duration::from_secs(60)),
+        "a deadline hit defers rather than killing the provider permanently",
+    );
+}
+
+/// The boot call site carries the same bound: a hung `install` refuses the
+/// boot instead of parking the launch forever.
+#[tokio::test(start_paused = true)]
+async fn a_hanging_provider_install_refuses_the_boot_by_deadline() {
+    let extension: Arc<dyn Extension<CoreRuntime>> = Arc::new(HangingExtension);
+    let scenario = BootScenario::new()
+        .extensions([extension])
+        .limits(ModuleLimits {
+            event_deadline_secs: Some(INSTALL_DEADLINE.as_secs()),
+            ..Default::default()
+        });
+    let wasm = scenario.dir().join("hanging.wasm");
+    std::fs::write(&wasm, b"(component)").expect("write component");
+
+    let started = tokio::time::Instant::now();
+    let refusal = tokio::time::timeout(
+        OUTER_TIMEOUT,
+        scenario
+            .adapter(Entry::new(TestManifest::new("hanging").kind("hanging-adapter")).wasm(wasm))
+            .expect_refusal(),
+    )
+    .await
+    .expect("boot returned: the install deadline bounded the hung install");
+
+    assert_eq!(
+        started.elapsed(),
+        INSTALL_DEADLINE,
+        "the install rode the configured deadline, not another timer",
+    );
+    refusal.names("did not install in time");
+}