Persist service registry

packages/wavs/benches/dev_triggers/setup.rs

@@ -137,7 +137,7 @@ impl DevTriggersRuntime {
         assert_eq!(digest, component_digest);
 
         // Register service
-        futures::executor::block_on(dispatcher.add_service_direct(service.clone()))
+        futures::executor::block_on(dispatcher.add_service_direct(service.clone(), None))
             .expect("add service to dispatcher");
 
         // Start dispatcher (store handle + context for graceful shutdown)

packages/wavs/src/dispatcher.rs

@@ -45,6 +45,7 @@ use wavs_types::{
 use wavs_types::{Service, ServiceError, ServiceId, SignerResponse, TriggerAction};
 
 use crate::config::Config;
+use crate::service_registry::{RegistryError, ServiceRegistry};
 use crate::services::{Services, ServicesError};
 use crate::subsystems::aggregator::error::AggregatorError;
 use crate::subsystems::aggregator::{Aggregator, AggregatorCommand};
@@ -76,6 +77,7 @@ pub struct Dispatcher<S: CAStorage> {
     pub dispatcher_to_submission_tx: crossbeam::channel::Sender<SubmissionCommand>,
     pub dispatcher_to_aggregator_tx: crossbeam::channel::Sender<AggregatorCommand>,
     pub db_storage: WavsDb,
+    pub service_registry: ServiceRegistry,
     /// Cached EVM HTTP providers per chain to avoid creating new connections for each query
     evm_http_providers: Arc<RwLock<HashMap<ChainKey, DynProvider>>>,
     /// Cached Cosmos query clients per chain to avoid creating new connections for each query
@@ -117,6 +119,7 @@ impl Dispatcher<FileStorage> {
 
         let file_storage = FileStorage::new(config.data.join("ca"))?;
         let db_storage = WavsDb::new()?;
+        let service_registry = ServiceRegistry::load(&config.data)?;
 
         let services = Services::new(db_storage.clone());
 
@@ -170,6 +173,7 @@ impl Dispatcher<FileStorage> {
             aggregator,
             services,
             db_storage,
+            service_registry,
             chain_configs: config.chains.clone(),
             metrics: metrics.dispatcher.clone(),
             ipfs_gateway: config.ipfs_gateway.clone(),
@@ -369,8 +373,113 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
             }
         });
 
-        // populate the initial triggers
-        let initial_services = self.services.list(Bound::Unbounded, Bound::Unbounded)?;
+        // Restore services from the persisted registry
+        let registry_entries = self.service_registry.entries();
+        let chain_configs_for_restore = self.chain_configs.read().unwrap().clone();
+        let ipfs_gateway_for_restore = self.ipfs_gateway.clone();
+        let evm_providers_for_restore = self.evm_http_providers.clone();
+        let cosmos_clients_for_restore = self.cosmos_query_clients.clone();
+
+        let initial_services: Vec<Service> = ctx.rt.block_on(async {
+            // Fetch all services from chain in parallel (bounded concurrency)
+            const MAX_CONCURRENT_RESTORES: usize = 10;
+            let fetched: Vec<_> = stream::iter(&registry_entries)
+                .map(|entry| {
+                    let chain_configs = &chain_configs_for_restore;
+                    let ipfs_gateway = &ipfs_gateway_for_restore;
+                    let evm_providers = &evm_providers_for_restore;
+                    let cosmos_clients = &cosmos_clients_for_restore;
+                    async move {
+                        let (chain, address) = match &entry.service_manager {
+                            ServiceManager::Evm { chain, address } => {
+                                (chain.clone(), layer_climb::prelude::Address::from(*address))
+                            }
+                            ServiceManager::Cosmos { chain, address } => (
+                                chain.clone(),
+                                layer_climb::prelude::Address::from(address.clone()),
+                            ),
+                        };
+                        let result = query_service_from_address(
+                            chain,
+                            address,
+                            chain_configs,
+                            ipfs_gateway,
+                            evm_providers,
+                            cosmos_clients,
+                        )
+                        .await;
+                        (entry, result)
+                    }
+                })
+                .buffer_unordered(MAX_CONCURRENT_RESTORES)
+                .collect::<Vec<_>>()
+                .await;
+
+            // Process results sequentially (DB writes and manager registration are not async-safe to parallelize)
+            let mut restored = Vec::new();
+            for (entry, result) in fetched {
+                match result {
+                    Ok(service) => {
+                        // Store the service in DB
+                        if let Err(err) = self.services.save(&service) {
+                            tracing::error!(
+                                "Failed to save restored service {} to DB: {:?}",
+                                service.name,
+                                err
+                            );
+                            continue;
+                        }
+
+                        // Store components
+                        if let Err(err) = self
+                            .engine_manager
+                            .store_components_for_service(&service)
+                            .await
+                        {
+                            tracing::error!(
+                                "Failed to store components for restored service {}: {:?}",
+                                service.name,
+                                err
+                            );
+                            continue;
+                        }
+
+                        // Add to managers with explicit HD index from registry
+                        if let Err(err) = add_service_to_managers(
+                            &service,
+                            &self.trigger_manager,
+                            &self.submission_manager,
+                            &self.dispatcher_to_aggregator_tx,
+                            Some(entry.hd_index),
+                        ) {
+                            tracing::error!(
+                                "Failed to add restored service {} to managers: {:?}",
+                                service.name,
+                                err
+                            );
+                            continue;
+                        }
+
+                        tracing::info!(
+                            "Restored service {} [{:?}] with HD index {}",
+                            service.name,
+                            service.manager,
+                            entry.hd_index
+                        );
+                        restored.push(service);
+                    }
+                    Err(err) => {
+                        tracing::error!(
+                            "Failed to restore service from chain for {:?}: {:?}",
+                            entry.service_manager,
+                            err
+                        );
+                    }
+                }
+            }
+            restored
+        });
+
         let total_workflows: usize = initial_services.iter().map(|s| s.workflows.len()).sum();
         tracing::info!(
             "Initializing dispatcher: services={}, workflows={}, components={}",
@@ -379,16 +488,6 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
             self.list_component_digests()?.len()
         );
 
-        for service in initial_services.iter() {
-            add_service_to_managers(
-                service,
-                &self.trigger_manager,
-                &self.submission_manager,
-                &self.dispatcher_to_aggregator_tx,
-                None,
-            )?;
-        }
-
         // Check ServiceURI for each service at startup and update if needed (bounded concurrency)
         let chain_configs = self.chain_configs.read().unwrap().clone();
         let ipfs_gateway = self.ipfs_gateway.clone();
@@ -491,13 +590,14 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
         &self,
         service_manager: ServiceManager,
     ) -> Result<Service, DispatcherError> {
-        let (chain, address) = match service_manager {
+        let (chain, address) = match &service_manager {
             ServiceManager::Evm { chain, address } => {
-                (chain, layer_climb::prelude::Address::from(address))
-            }
-            ServiceManager::Cosmos { chain, address } => {
-                (chain, layer_climb::prelude::Address::from(address))
+                (chain.clone(), layer_climb::prelude::Address::from(*address))
             }
+            ServiceManager::Cosmos { chain, address } => (
+                chain.clone(),
+                layer_climb::prelude::Address::from(address.clone()),
+            ),
         };
         let chain_configs = self.chain_configs.read().unwrap().clone();
         let service = query_service_from_address(
@@ -510,7 +610,19 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
         )
         .await?;
 
-        self.add_service_direct(service.clone()).await?;
+        // Allocate HD index from the registry (single source of truth for key derivation indices)
+        let hd_index = self.service_registry.append(service_manager.clone())?;
+
+        if let Err(e) = self
+            .add_service_direct(service.clone(), Some(hd_index))
+            .await
+        {
+            // Roll back the registry entry so we don't leave a stale entry on disk
+            if let Err(remove_err) = self.service_registry.remove(&service_manager) {
+                tracing::error!("Failed to roll back registry entry after add_service_direct failure: {remove_err}");
+            }
+            return Err(e);
+        }
 
         // Get current service count for logging
         let current_services = self.services.list(Bound::Unbounded, Bound::Unbounded)?;
@@ -524,7 +636,11 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
 
     // this is public just so we can call it from tests
     #[instrument(skip(self), fields(subsys = "Dispatcher", service.name = %service.name, service.manager = ?service.manager))]
-    pub async fn add_service_direct(&self, service: Service) -> Result<(), DispatcherError> {
+    pub async fn add_service_direct(
+        &self,
+        service: Service,
+        hd_index: Option<u32>,
+    ) -> Result<(), DispatcherError> {
         let service_id = service.id();
         tracing::info!("Adding service: {} [{:?}]", service.name, service.manager);
         // Check if service is already registered
@@ -546,14 +662,30 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
             &self.trigger_manager,
             &self.submission_manager,
             &self.dispatcher_to_aggregator_tx,
-            None,
+            hd_index,
         )?;
 
         Ok(())
     }
 
     #[instrument(skip(self), fields(subsys = "Dispatcher"))]
     pub fn remove_service(&self, id: ServiceId) -> Result<(), DispatcherError> {
+        // Look up the ServiceManager before removing so we can delete it from the registry
+        let service_manager = self.services.get(&id).ok().map(|s| s.manager.clone());
+
+        self.remove_service_inner(id.clone())?;
+
+        // Remove from persistent registry
+        if let Some(sm) = service_manager {
+            self.service_registry.remove(&sm)?;
+        }
+
+        Ok(())
+    }
+
+    /// Remove a service from in-memory state without mutating the persistent registry.
+    /// Used by `change_service_inner` where the ServiceManager hasn't changed.
+    fn remove_service_inner(&self, id: ServiceId) -> Result<(), DispatcherError> {
         self.services.remove(&id)?;
         self.engine_manager.engine.remove_storage(&id);
         self.trigger_manager.remove_service(id.clone())?;
@@ -642,8 +774,9 @@ impl<S: CAStorage + 'static> Dispatcher<S> {
             .store_components_for_service(&service)
             .await?;
 
-        // Remove the old service - after this, no await points until the new service is added
-        self.remove_service(service_id.clone())?;
+        // Remove the old service from in-memory state only (ServiceManager hasn't changed,
+        // so no registry mutation needed)
+        self.remove_service_inner(service_id.clone())?;
 
         // Store the service BEFORE setting up triggers/P2P subscription
         // This ensures the service is in the database before any triggers can fire
@@ -923,4 +1056,7 @@ pub enum DispatcherError {
 
     #[error("Cosmos query error: {0}")]
     CosmosQuery(anyhow::Error),
+
+    #[error("Service registry: {0}")]
+    Registry(#[from] RegistryError),
 }

packages/wavs/src/http/handlers/service/add.rs

@@ -64,7 +64,7 @@ pub async fn handle_add_service_direct(
 
 async fn add_service_direct_inner(state: HttpState, service_hash: String) -> HttpResult<()> {
     let service = get_service_inner_hash(&state, service_hash).await?;
-    state.dispatcher.add_service_direct(service).await?;
+    state.dispatcher.add_service_direct(service, None).await?;
 
     state.metrics.increment_registered_services();
 

packages/wavs/src/lib.rs

@@ -7,6 +7,7 @@ pub mod config;
 pub mod dispatcher; // where we have the high-level dispatcher
 pub mod health;
 pub mod http;
+pub mod service_registry;
 pub mod services;
 pub mod subsystems; // subsystems: engine, submission, and trigger // services lookup