runtime: fold venue adapters into the restart and poison sweeps

.github/workflows/ci.yml

@@ -78,7 +78,7 @@ jobs:
       - uses: taiki-e/install-action@43aecc8d72668fbcfe75c31400bc4f890f1c5853 # v2.83.2
         with:
           tool: nextest
-      # Build all 15 guest module wasms ONCE (release/wasm32-wasip2): the single
+      # Build all 16 guest module wasms ONCE (release/wasm32-wasip2): the single
       # source of truth for guest buildability and the artifacts the integration
       # tests load. Replaces the deleted 9-way build-module matrix, which recompiled
       # the shared wasm dependency graph ~9x cold. Per-module size report folded in;
@@ -88,8 +88,8 @@ jobs:
           cargo build --release --target wasm32-wasip2 --locked \
             -p example -p twap-monitor -p ethflow-watcher -p price-alert \
             -p balance-tracker -p stop-loss -p http-probe -p echo-venue \
-            -p echo-client -p clock-reader -p flaky-bomb -p fuel-bomb \
-            -p memory-bomb -p panic-bomb -p slow-host
+            -p echo-client -p clock-reader -p flaky-bomb -p flaky-venue \
+            -p fuel-bomb -p memory-bomb -p panic-bomb -p slow-host
           {
             echo "### module .wasm sizes"
             echo "| module | bytes |"

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "modules/examples/stop-loss",
     "modules/fixtures/clock-reader",
     "modules/fixtures/flaky-bomb",
+    "modules/fixtures/flaky-venue",
     "modules/fixtures/fuel-bomb",
     "modules/fixtures/memory-bomb",
     "modules/fixtures/panic-bomb",

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

@@ -4,7 +4,8 @@
 //! caller, and each instance sits behind an [`ActorSlot`] async mutex held
 //! across the guest await, so one store never runs two guest calls at once.
 
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Instant;
 
 use tokio::sync::Mutex as AsyncMutex;
 use wasmtime::Store;
@@ -16,6 +17,47 @@ use super::state::HostState;
 /// is not `Sync`; concurrent callers queue here.
 pub type ActorSlot<A> = Arc<AsyncMutex<A>>;
 
+/// Shared liveness of one supervised component. The store marks it dead on
+/// a trap, recording when, so the supervisor's restart sweep can count the
+/// backoff from the death rather than from the sweep that observed it.
+/// Cloning shares the flag. Starts alive.
+#[derive(Clone, Debug, Default)]
+pub struct Liveness(Arc<Mutex<Option<Instant>>>);
+
+impl Liveness {
+    /// Whether the component is currently callable.
+    pub fn is_alive(&self) -> bool {
+        self.lock().is_none()
+    }
+
+    /// When the component died, while it is dead.
+    pub fn dead_since(&self) -> Option<Instant> {
+        *self.lock()
+    }
+
+    /// Mark the component dead: its store trapped and is unusable. Keeps
+    /// the first death instant when already dead.
+    pub fn mark_dead(&self) {
+        let mut died_at = self.lock();
+        if died_at.is_none() {
+            *died_at = Some(Instant::now());
+        }
+    }
+
+    /// Mark the component alive again after a restart.
+    pub fn mark_alive(&self) {
+        *self.lock() = None;
+    }
+
+    /// The flag, recovered from a poisoned lock: the state is a bare
+    /// `Option<Instant>`, valid under any interleaving.
+    fn lock(&self) -> MutexGuard<'_, Option<Instant>> {
+        self.0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+}
+
 /// A guest call failed outside the component's typed error space.
 #[derive(Debug, thiserror::Error)]
 pub enum ActorFault {
@@ -30,29 +72,36 @@ pub enum ActorFault {
 
 /// A supervised component store: refuelled before each guest call so every
 /// invocation starts from a full budget, with traps projected onto
-/// [`ActorFault`].
+/// [`ActorFault`] and recorded on the shared [`Liveness`].
 pub struct SupervisedStore<T: RuntimeTypes> {
     store: Store<HostState<T>>,
     fuel_per_call: u64,
+    liveness: Liveness,
 }
 
 impl<T: RuntimeTypes> SupervisedStore<T> {
-    /// Supervise an instantiated store with a per-call fuel budget.
-    pub fn new(store: Store<HostState<T>>, fuel_per_call: u64) -> Self {
+    /// Supervise an instantiated store with a per-call fuel budget,
+    /// reporting traps on `liveness`.
+    pub fn new(store: Store<HostState<T>>, fuel_per_call: u64, liveness: Liveness) -> Self {
         Self {
             store,
             fuel_per_call,
+            liveness,
         }
     }
 
-    /// Refuel, then run one guest call against the store.
+    /// Refuel, then run one guest call against the store. A trap marks the
+    /// shared liveness dead: the store is poisoned until reinstantiated.
     pub async fn call<R>(
         &mut self,
         call: impl AsyncFnOnce(&mut Store<HostState<T>>) -> wasmtime::Result<R>,
     ) -> Result<R, ActorFault> {
         self.store
             .set_fuel(self.fuel_per_call)
             .map_err(ActorFault::Refuel)?;
-        call(&mut self.store).await.map_err(ActorFault::Trap)
+        call(&mut self.store).await.map_err(|trap| {
+            self.liveness.mark_dead();
+            ActorFault::Trap(trap)
+        })
     }
 }

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

@@ -11,6 +11,7 @@ use async_trait::async_trait;
 use wasmtime::Store;
 use wasmtime::component::{Component, Linker};
 
+use crate::host::actor::Liveness;
 use crate::host::component::RuntimeTypes;
 use crate::host::state::HostState;
 use crate::manifest::NamespaceCaps;
@@ -84,6 +85,9 @@ pub struct ProviderInstance<'a, T: RuntimeTypes> {
     pub config: Vec<(String, String)>,
     /// Fuel budget applied before each routed guest call.
     pub fuel_per_call: u64,
+    /// Shared liveness the installed instance reports traps on and the
+    /// supervisor's restart sweep reads.
+    pub liveness: Liveness,
 }
 
 /// Outcome of one provider install.

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

@@ -41,7 +41,7 @@ use crate::bindings::{
     IntentHeader, IntentStatus, IntentStatusUpdate, Quotation, RateLimit, SubmitOutcome,
     VenueAdapter, VenueError, nexum,
 };
-use crate::host::actor::{ActorFault, ActorSlot, SupervisedStore};
+use crate::host::actor::{ActorFault, ActorSlot, Liveness, SupervisedStore};
 use crate::host::component::RuntimeTypes;
 use crate::host::extension::{
     HostService, Installed, ProviderInstance, ProviderKind, downcast_service,
@@ -225,10 +225,16 @@ pub struct VenueActor<T: RuntimeTypes> {
 }
 
 impl<T: RuntimeTypes> VenueActor<T> {
-    /// Wrap an instantiated adapter store for routing.
-    pub fn new(store: Store<HostState<T>>, bindings: VenueAdapter, fuel_per_call: u64) -> Self {
+    /// Wrap an instantiated adapter store for routing, reporting traps on
+    /// the shared `liveness`.
+    pub fn new(
+        store: Store<HostState<T>>,
+        bindings: VenueAdapter,
+        fuel_per_call: u64,
+        liveness: Liveness,
+    ) -> Self {
         Self {
-            actor: SupervisedStore::new(store, fuel_per_call),
+            actor: SupervisedStore::new(store, fuel_per_call, liveness),
             bindings,
         }
     }
@@ -302,6 +308,15 @@ impl<T: RuntimeTypes> VenueInvoker for VenueActor<T> {
 /// One installed adapter behind its serialising slot.
 type AdapterSlot = ActorSlot<dyn VenueInvoker>;
 
+/// One installed venue: the adapter slot plus the liveness the supervisor's
+/// sweep shares with the actor. A dead entry stays installed, so the venue
+/// resolves to `unavailable` (temporarily dead) rather than `unknown-venue`
+/// (never installed) until the sweep restarts it.
+struct InstalledVenue {
+    slot: AdapterSlot,
+    liveness: Liveness,
+}
+
 /// Per-caller charge history, pruned to the quota window on each touch.
 #[derive(Default)]
 struct QuotaLedger {
@@ -356,7 +371,7 @@ fn status_body(status: IntentStatus) -> StatusBody {
 /// install through the shared handle at provider boot, before any client
 /// call routes.
 struct VenueRegistryInner {
-    adapters: Mutex<HashMap<VenueId, AdapterSlot>>,
+    adapters: Mutex<HashMap<VenueId, InstalledVenue>>,
     guard: Arc<dyn EgressGuard>,
     quota: SubmitQuota,
     ledger: Mutex<QuotaLedger>,
@@ -382,31 +397,44 @@ impl VenueRegistry {
     /// extension's.
     pub const NAMESPACE: &'static str = "videre";
 
-    /// Install an adapter under its venue id. Rejects a duplicate id: two
+    /// Install an adapter under its venue id, sharing `liveness` with its
+    /// invoker. Rejects a duplicate id while the incumbent is alive: two
     /// adapters answering the same venue would silently shadow one another,
-    /// which is a config error worth failing boot over.
+    /// which is a config error worth failing boot over. A dead incumbent is
+    /// replaced: that is the sweep restarting a trapped adapter.
     pub fn install(
         &self,
         venue: VenueId,
+        liveness: Liveness,
         invoker: impl VenueInvoker + 'static,
     ) -> Result<(), DuplicateVenue> {
         let mut adapters = self.inner.adapters.lock().expect("adapter map poisoned");
-        if adapters.contains_key(&venue) {
+        if adapters.get(&venue).is_some_and(|v| v.liveness.is_alive()) {
             return Err(DuplicateVenue { venue });
         }
-        adapters.insert(venue, Arc::new(AsyncMutex::new(invoker)));
+        adapters.insert(
+            venue,
+            InstalledVenue {
+                slot: Arc::new(AsyncMutex::new(invoker)),
+                liveness,
+            },
+        );
         Ok(())
     }
 
-    /// Resolve a venue id to its installed adapter slot.
+    /// Resolve a venue id to its installed adapter slot. An uninstalled
+    /// venue is `unknown-venue`; an installed but dead one is `unavailable`
+    /// pending the supervisor's restart sweep, without touching its
+    /// poisoned store.
     fn resolve(&self, venue: &VenueId) -> Result<AdapterSlot, VenueError> {
-        self.inner
-            .adapters
-            .lock()
-            .expect("adapter map poisoned")
-            .get(venue)
-            .cloned()
-            .ok_or(VenueError::UnknownVenue)
+        let adapters = self.inner.adapters.lock().expect("adapter map poisoned");
+        let installed = adapters.get(venue).ok_or(VenueError::UnknownVenue)?;
+        if !installed.liveness.is_alive() {
+            return Err(VenueError::Unavailable(format!(
+                "venue {venue} is dead pending restart"
+            )));
+        }
+        Ok(Arc::clone(&installed.slot))
     }
 
     /// Whether `caller` has budget left in the current window. Read-only: it
@@ -580,8 +608,8 @@ impl VenueRegistry {
         }
         let mut updates = Vec::new();
         for (venue, receipt) in snapshot {
-            // Installed adapters never leave the registry, so a resolve
-            // failure here is unreachable; skip defensively regardless.
+            // A dead venue fails to resolve; its watch stays for the
+            // cadence after the sweep restarts the adapter.
             let Ok(slot) = self.resolve(&venue) else {
                 continue;
             };
@@ -724,6 +752,7 @@ impl<T: RuntimeTypes> ProviderKind<T> for VenueAdapterKind {
             mut store,
             config,
             fuel_per_call,
+            liveness,
         } = instance;
         let bindings = VenueAdapter::instantiate_async(&mut store, component, linker)
             .await
@@ -750,7 +779,8 @@ impl<T: RuntimeTypes> ProviderKind<T> for VenueAdapterKind {
         registry
             .install(
                 venue_id.clone(),
-                VenueActor::new(store, bindings, fuel_per_call),
+                liveness.clone(),
+                VenueActor::new(store, bindings, fuel_per_call, liveness),
             )
             .with_context(|| format!("install adapter {venue_id}"))?;
         Ok(Installed::Live)
@@ -1062,7 +1092,9 @@ mod tests {
             builder = builder.with_guard(guard);
         }
         let registry = builder.build();
-        registry.install(cow(), adapter).expect("install adapter");
+        registry
+            .install(cow(), Liveness::default(), adapter)
+            .expect("install adapter");
         registry
     }
 
@@ -1367,14 +1399,61 @@ mod tests {
         let a = Arc::new(StubCalls::default());
         let b = Arc::new(StubCalls::default());
         registry
-            .install(cow(), StubAdapter::new(a))
+            .install(cow(), Liveness::default(), StubAdapter::new(a))
             .expect("first install");
         let err = registry
-            .install(cow(), StubAdapter::new(b))
+            .install(cow(), Liveness::default(), StubAdapter::new(b))
             .expect_err("second install collides");
         assert_eq!(err.venue, cow());
     }
 
+    #[tokio::test]
+    async fn dead_venue_is_unavailable_not_unknown() {
+        let calls = Arc::new(StubCalls::default());
+        let liveness = Liveness::default();
+        let registry = VenueRegistryBuilder::new(SubmitQuota::default()).build();
+        registry
+            .install(cow(), liveness.clone(), StubAdapter::new(calls.clone()))
+            .expect("install adapter");
+        liveness.mark_dead();
+
+        // Temporarily dead resolves distinctly from never installed, and
+        // the dead adapter's slot is never entered.
+        assert!(matches!(
+            registry.submit("mod-a", &cow(), b"b".to_vec()).await,
+            Err(VenueError::Unavailable(_))
+        ));
+        assert!(matches!(
+            registry
+                .submit("mod-a", &VenueId::from("unlisted"), b"b".to_vec())
+                .await,
+            Err(VenueError::UnknownVenue)
+        ));
+        assert_eq!(calls.derive.load(Ordering::SeqCst), 0);
+    }
+
+    #[test]
+    fn a_dead_incumbent_is_replaced_on_reinstall() {
+        let registry = VenueRegistryBuilder::new(SubmitQuota::default()).build();
+        let liveness = Liveness::default();
+        registry
+            .install(
+                cow(),
+                liveness.clone(),
+                StubAdapter::new(Arc::new(StubCalls::default())),
+            )
+            .expect("first install");
+        liveness.mark_dead();
+        registry
+            .install(
+                cow(),
+                Liveness::default(),
+                StubAdapter::new(Arc::new(StubCalls::default())),
+            )
+            .expect("a restart replaces the dead incumbent");
+        assert_eq!(registry.venue_count(), 1);
+    }
+
     #[test]
     fn zero_quota_saturates_to_one() {
         let registry =
@@ -1523,7 +1602,9 @@ mod tests {
         let registry = VenueRegistryBuilder::new(SubmitQuota::default())
             .with_watch_limit(watch_limit)
             .build();
-        registry.install(cow(), adapter).expect("install adapter");
+        registry
+            .install(cow(), Liveness::default(), adapter)
+            .expect("install adapter");
         registry
     }