From c2d821bf392f11d7c2f52c212370af38b8921d3b Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 16 Jul 2026 23:45:57 +0000 Subject: [PATCH] runtime: carry the venue registry through the services map and delete the privileged field The client face resolves the registry from the store's service map under the videre namespace; no service means every call resolves to unknown-venue. The boot path seeds the registry into the map until the videre extension takes it over, provider stores carry an empty map so the registry never cycles back into a store it owns, and the supervisor field is gone. --- crates/nexum-runtime/src/builder.rs | 16 ++-- crates/nexum-runtime/src/host/extension.rs | 14 ++++ .../src/host/impls/venue_client.rs | 37 +++++---- crates/nexum-runtime/src/host/state.rs | 8 +- .../nexum-runtime/src/host/venue_registry.rs | 9 +-- crates/nexum-runtime/src/supervisor.rs | 77 +++++++------------ crates/nexum-runtime/src/supervisor/tests.rs | 2 +- 7 files changed, 79 insertions(+), 84 deletions(-) diff --git a/crates/nexum-runtime/src/builder.rs b/crates/nexum-runtime/src/builder.rs index f5559506..beaa2eb6 100644 --- a/crates/nexum-runtime/src/builder.rs +++ b/crates/nexum-runtime/src/builder.rs @@ -267,10 +267,14 @@ impl LaunchRuntime for AssembledRuntime<'_, T> { let logs = components.logs.clone(); let chain_log_subs = supervisor.chain_log_subscriptions(); // Status polling runs only when it can produce something a module - // will see: at least one intent-status subscriber and at least one - // installed adapter to poll. - let poll_statuses = supervisor.has_intent_status_subscribers() - && supervisor.venue_registry().venue_count() > 0; + // will see: at least one intent-status subscriber, a registered + // venue-registry service, and at least one installed adapter to poll. + let status_registry = supervisor + .has_intent_status_subscribers() + .then(|| supervisor.venue_registry()) + .flatten() + .filter(|registry| registry.venue_count() > 0); + let poll_statuses = status_registry.is_some(); // No subscriptions: nothing to drive. Return a handle whose event loop // is already complete so `wait` resolves immediately. @@ -308,9 +312,9 @@ impl LaunchRuntime for AssembledRuntime<'_, T> { &executor, &mut reconnect_tasks, ); - let intent_status_stream = poll_statuses.then(|| { + let intent_status_stream = status_registry.map(|registry| { event_loop::open_intent_status_stream( - supervisor.venue_registry(), + registry, engine_cfg.limits.status_poll_interval(), &executor, &mut reconnect_tasks, diff --git a/crates/nexum-runtime/src/host/extension.rs b/crates/nexum-runtime/src/host/extension.rs index ed14de95..a282d4ba 100644 --- a/crates/nexum-runtime/src/host/extension.rs +++ b/crates/nexum-runtime/src/host/extension.rs @@ -143,6 +143,20 @@ impl HostServices { pub fn raw(&self, namespace: &str) -> Option<&Arc> { self.0.get(namespace) } + + /// Publish `service` under `namespace`, refusing a duplicate. The boot + /// path seeds a service no extension registers yet. + pub fn with_service( + self, + namespace: &'static str, + service: Arc, + ) -> anyhow::Result { + let mut map = Arc::unwrap_or_clone(self.0); + if map.insert(namespace, service).is_some() { + anyhow::bail!("duplicate extension service namespace {namespace}"); + } + Ok(Self(Arc::new(map))) + } } #[cfg(test)] diff --git a/crates/nexum-runtime/src/host/impls/venue_client.rs b/crates/nexum-runtime/src/host/impls/venue_client.rs index 0fac7d94..8b86e8a4 100644 --- a/crates/nexum-runtime/src/host/impls/venue_client.rs +++ b/crates/nexum-runtime/src/host/impls/venue_client.rs @@ -1,25 +1,36 @@ -//! `videre:venue/client`: the keeper-facing venue import. Every method is a -//! thin delegation to the shared -//! [`VenueRegistry`](crate::host::venue_registry) carried in the store; the -//! registry owns the venue resolution, per-adapter serialisation, guard -//! seam (advisory-only for now), and quota. The caller identity the registry -//! meters against is this store's module namespace. +//! `videre:venue/client`: the keeper-facing venue import. Every method +//! resolves the shared [`VenueRegistry`] from the store's service map under +//! the videre namespace and delegates; the registry owns the venue +//! resolution, per-adapter serialisation, guard seam (advisory-only for +//! now), and quota. The caller identity the registry meters against is this +//! store's module namespace. No registry service means no venues, so every +//! call resolves to `unknown-venue`. + +use std::sync::Arc; use crate::bindings::client::Host; use crate::bindings::{IntentStatus, Quotation, SubmitOutcome, VenueError}; use crate::host::component::RuntimeTypes; use crate::host::state::HostState; -use crate::host::venue_registry::VenueId; +use crate::host::venue_registry::{VenueId, VenueRegistry}; + +/// The registry published under the videre service namespace. +fn registry(state: &HostState) -> Result, VenueError> { + state + .services + .get::(VenueRegistry::NAMESPACE) + .ok_or(VenueError::UnknownVenue) +} impl Host for HostState { async fn quote(&mut self, venue: String, body: Vec) -> Result { - self.venue_registry + registry(self)? .quote(&self.run.module, &VenueId::from(venue), body) .await } async fn submit(&mut self, venue: String, body: Vec) -> Result { - self.venue_registry + registry(self)? .submit(&self.run.module, &VenueId::from(venue), body) .await } @@ -29,14 +40,10 @@ impl Host for HostState { venue: String, receipt: Vec, ) -> Result { - self.venue_registry - .status(&VenueId::from(venue), receipt) - .await + registry(self)?.status(&VenueId::from(venue), receipt).await } async fn cancel(&mut self, venue: String, receipt: Vec) -> Result<(), VenueError> { - self.venue_registry - .cancel(&VenueId::from(venue), receipt) - .await + registry(self)?.cancel(&VenueId::from(venue), receipt).await } } diff --git a/crates/nexum-runtime/src/host/state.rs b/crates/nexum-runtime/src/host/state.rs index 3d85117d..b1b103cb 100644 --- a/crates/nexum-runtime/src/host/state.rs +++ b/crates/nexum-runtime/src/host/state.rs @@ -14,7 +14,6 @@ use super::component::{Handle, RuntimeTypes}; use super::extension::HostServices; use super::http::HttpGate; use super::logs::{LogRouter, RunId}; -use super::venue_registry::VenueRegistry; /// Per-module host state, generic over the [`RuntimeTypes`] lattice /// binding the backend seams. The composition root supplies the @@ -52,12 +51,9 @@ pub struct HostState { /// `local-store` backend - per-module handle with pre-computed /// keccak256 namespace prefix. pub store: Handle, - /// The venue registry the `videre:venue/client` import dispatches to. - /// Every module store carries the same shared handle; an adapter store, - /// which cannot call the client face, carries an empty one. - pub venue_registry: VenueRegistry, /// Extension-owned host services, keyed by extension namespace and - /// downcast at the call site. One shared map across every store. + /// downcast at the call site. One shared map across every module store; + /// a provider store carries an empty map. pub services: HostServices, } diff --git a/crates/nexum-runtime/src/host/venue_registry.rs b/crates/nexum-runtime/src/host/venue_registry.rs index 8e93942d..46c5485c 100644 --- a/crates/nexum-runtime/src/host/venue_registry.rs +++ b/crates/nexum-runtime/src/host/venue_registry.rs @@ -378,12 +378,9 @@ pub struct VenueRegistry { impl HostService for VenueRegistry {} impl VenueRegistry { - /// An empty registry: no adapters, the unit guard, the default quota. - /// This is what an adapter store (which cannot call the client face) and - /// the single-module `just run` path carry. - pub fn empty() -> Self { - VenueRegistryBuilder::new(SubmitQuota::default()).build() - } + /// Service namespace the registry publishes under: the videre + /// extension's. + pub const NAMESPACE: &'static str = "videre"; /// Install an adapter under its venue id. Rejects a duplicate id: two /// adapters answering the same venue would silently shadow one another, diff --git a/crates/nexum-runtime/src/supervisor.rs b/crates/nexum-runtime/src/supervisor.rs index 154e8193..92894f6b 100644 --- a/crates/nexum-runtime/src/supervisor.rs +++ b/crates/nexum-runtime/src/supervisor.rs @@ -64,13 +64,6 @@ use crate::manifest::{ /// the component seam backends. pub struct Supervisor { modules: Vec>, - /// The venue registry: every installed venue adapter's serialising - /// store, keyed by venue id. Cached so a module restart rebuilds a store - /// carrying the same shared handle. Adapters boot through the same store, - /// fuel, and memory machinery as modules but carry no subscriptions: - /// modules reach them through this registry, not through dispatch. Folding - /// adapters into the restart and poison sweeps is still a later change. - venue_registry: VenueRegistry, /// Venue adapters loaded at boot, whether or not `init` succeeded. adapters_total: usize, /// Adapters whose `init` succeeded and that are installed for routing. @@ -320,25 +313,22 @@ impl Supervisor { clocks: Option, ) -> Result { let registry = capability_registry(extensions); - let services = HostServices::from_extensions(extensions)?; - let venue_registry = VenueRegistryBuilder::new(engine_cfg.limits.quota()) - .with_watch_limit(engine_cfg.limits.watch()) - .build(); - // Provider kinds the boot loop resolves manifest kinds against: - // every extension-registered kind plus the venue-adapter row, seeded - // here while the registry lives in-core; the videre extension takes - // it over. + // The venue registry rides the generic service map under the videre + // namespace, seeded here while it lives in-core; the videre + // extension takes it over. Same for the venue-adapter kind row. + let venue_service: Arc = Arc::new( + VenueRegistryBuilder::new(engine_cfg.limits.quota()) + .with_watch_limit(engine_cfg.limits.watch()) + .build(), + ); + let services = HostServices::from_extensions(extensions)? + .with_service(VenueRegistry::NAMESPACE, Arc::clone(&venue_service))?; + // Provider kinds the boot loop resolves manifest kinds against. let mut kinds = provider_kinds(extensions, &services)?; - register_kind( - &mut kinds, - Box::new(VenueAdapterKind), - Arc::new(venue_registry.clone()), - )?; + register_kind(&mut kinds, Box::new(VenueAdapterKind), venue_service)?; // Providers boot first into the shared registry handle, so every // module store built below already routes to the installed venues. - // Providers link only their kind's scoped imports, and their own - // stores carry an empty registry since a provider cannot call the - // client face. + // Providers link only their kind's scoped imports. let provider_registry = CapabilityRegistry::provider(); let adapters_total = engine_cfg.adapters.len(); let mut adapters_alive = 0; @@ -350,7 +340,6 @@ impl Supervisor { &engine_cfg.limits, &provider_registry, clocks.as_ref(), - services.clone(), &kinds, ) .await @@ -370,7 +359,6 @@ impl Supervisor { &engine_cfg.limits, ®istry, clocks.as_ref(), - venue_registry.clone(), services.clone(), ) .await @@ -387,7 +375,6 @@ impl Supervisor { ); Ok(Self { modules, - venue_registry, adapters_total, adapters_alive, engine: engine.clone(), @@ -423,9 +410,8 @@ impl Supervisor { manifest: manifest.map(Path::to_path_buf), }; // The single-module override path serves `just run`; adapters are - // configured through `engine.toml`, so the registry is empty here and - // every client call resolves to `unknown-venue`. - let venue_registry = VenueRegistry::empty(); + // configured through `engine.toml`, so no registry service is + // published and every client call resolves to `unknown-venue`. let loaded = Self::load_one( engine, linker, @@ -434,13 +420,11 @@ impl Supervisor { limits, ®istry, clocks.as_ref(), - venue_registry.clone(), services.clone(), ) .await?; Ok(Self { modules: vec![loaded], - venue_registry, adapters_total: 0, adapters_alive: 0, engine: engine.clone(), @@ -471,7 +455,6 @@ impl Supervisor { chain_response_max_bytes: usize, state_quota: u64, clocks: Option<&WasiClockOverride>, - venue_registry: VenueRegistry, services: HostServices, ) -> Result> { let namespace: &str = &run.module; @@ -530,7 +513,6 @@ impl Supervisor { chain: components.chain.clone(), chain_response_max_bytes, store: module_store, - venue_registry, services, }, ); @@ -539,8 +521,7 @@ impl Supervisor { Ok(store) } - // One flat argument per shared input threaded onto the store, plus the - // venue registry the module's `videre:venue/client` import dispatches to. + // One flat argument per shared input threaded onto the store. #[allow(clippy::too_many_arguments)] async fn load_one( engine: &Engine, @@ -550,7 +531,6 @@ impl Supervisor { limits_cfg: &ModuleLimits, registry: &CapabilityRegistry, clocks: Option<&WasiClockOverride>, - venue_registry: VenueRegistry, services: HostServices, ) -> Result> { let manifest_path = resolve_manifest_path(&entry.path, entry.manifest.as_deref()); @@ -616,7 +596,6 @@ impl Supervisor { limits_cfg.chain_response_max_bytes(), state_bytes, clocks, - venue_registry, services, )?; let bindings = EventModule::instantiate_async(&mut store, &component, linker) @@ -724,7 +703,6 @@ impl Supervisor { limits_cfg: &ModuleLimits, registry: &CapabilityRegistry, clocks: Option<&WasiClockOverride>, - services: HostServices, kinds: &ProviderKinds, ) -> Result { let manifest_path = resolve_manifest_path(&entry.path, entry.manifest.as_deref()); @@ -804,10 +782,9 @@ impl Supervisor { let linker = build_provider_linker::(engine, kind.as_ref())?; let run = RunId::new(namespace.clone(), 0); - // A provider store cannot call the client face, so it carries an - // empty registry; this also keeps the real registry out of the - // provider's `HostState`, so there is no reference cycle back into - // the registry that owns it. + // A provider links no service-consuming import, so its store carries + // an empty service map; the shared map holds the registry that owns + // the provider's store, and carrying it here would cycle. let store = Self::build_store( engine, components, @@ -820,8 +797,7 @@ impl Supervisor { limits_cfg.chain_response_max_bytes(), limits_cfg.state_bytes(), clocks, - VenueRegistry::empty(), - services, + HostServices::default(), )?; let config: Config = if loaded_manifest.config.is_empty() { @@ -969,10 +945,9 @@ impl Supervisor { let linker = build_linker::(&self.engine, &self.extensions)?; // Borrowed before the `&mut self.modules[idx]` reborrow so the restart - // path applies the same clock override and the same shared registry + // path applies the same clock override and the same shared services // as the initial boot. let clocks = self.clocks.clone(); - let venue_registry = self.venue_registry.clone(); let services = self.services.clone(); let module = &mut self.modules[idx]; // A restart is a new run: bump the sequence so its logs key @@ -990,7 +965,6 @@ impl Supervisor { module.chain_response_max_bytes, module.local_store_bytes, clocks.as_ref(), - venue_registry, services, )?; let bindings = EventModule::instantiate_async(&mut store, &module.component, &linker) @@ -1244,9 +1218,12 @@ impl Supervisor { }) } - /// The shared venue registry carried by every module store. - pub fn venue_registry(&self) -> VenueRegistry { - self.venue_registry.clone() + /// The venue registry published under the videre service namespace, + /// when one is. Shared by every module store through the service map. + pub fn venue_registry(&self) -> Option { + self.services + .get::(VenueRegistry::NAMESPACE) + .map(|registry| (*registry).clone()) } /// Shared per-module dispatch path: refuel, call `on_event`, and diff --git a/crates/nexum-runtime/src/supervisor/tests.rs b/crates/nexum-runtime/src/supervisor/tests.rs index 68093413..3830e6f3 100644 --- a/crates/nexum-runtime/src/supervisor/tests.rs +++ b/crates/nexum-runtime/src/supervisor/tests.rs @@ -898,7 +898,7 @@ async fn e2e_echo_module_registry_adapter_round_trip() { // Poll the registry the module submitted through and fan its transitions // back to the module. echo-venue settles instantly, so the first poll // reports a terminal status and the watch is pruned. - let registry = supervisor.venue_registry(); + let registry = supervisor.venue_registry().expect("registry service"); let mut delivered = 0; for _ in 0..2 { for update in registry.poll_status_transitions().await {