run restore on all stf periodically

bin/mosaic/config/config.a.toml

@@ -62,6 +62,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 60
 
 [rpc]
 bind_addr = "127.0.0.1:8000"

bin/mosaic/config/config.b.toml

@@ -62,6 +62,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 60
 
 [rpc]
 bind_addr = "127.0.0.1:8001"

bin/mosaic/config/config.c.toml

@@ -62,6 +62,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 60
 
 [rpc]
 bind_addr = "127.0.0.1:8002"

bin/mosaic/config/config.d.toml

@@ -62,6 +62,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 60
 
 [rpc]
 bind_addr = "127.0.0.1:8003"

bin/mosaic/config/config.e.toml

@@ -62,6 +62,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 60
 
 [rpc]
 bind_addr = "127.0.0.1:8004"

bin/mosaic/config/config.example.toml

@@ -48,6 +48,7 @@ chunk_timeout_secs = 30
 
 [sm_executor]
 command_queue_size = 256
+restore_interval_secs = 600
 
 [rpc]
 bind_addr = "127.0.0.1:8000"

bin/mosaic/src/config.rs

@@ -70,6 +70,7 @@ impl MosaicConfig {
         SmExecutorConfig {
             command_queue_size: self.sm_executor.command_queue_size,
             known_peers,
+            restore_interval_secs: self.sm_executor.restore_interval_secs,
         }
     }
 
@@ -352,6 +353,8 @@ impl Default for GarblingSection {
 pub(crate) struct SmExecutorSection {
     #[serde(default = "default_command_queue_size")]
     pub(crate) command_queue_size: usize,
+    /// Seconds between periodic `restore_known_peers` calls. Omitted or `0` disables.
+    pub(crate) restore_interval_secs: Option<u64>,
 }
 
 #[derive(Debug, Clone, Deserialize)]

crates/sm-executor-api/src/config.rs

@@ -7,13 +7,17 @@ pub struct SmExecutorConfig {
     pub command_queue_size: usize,
     /// Peers to restore at startup.
     pub known_peers: Vec<PeerId>,
+    /// Interval in seconds between periodic `restore_known_peers` runs.
+    /// `None` or `Some(0)` disables periodic restore.
+    pub restore_interval_secs: Option<u64>,
 }
 
 impl Default for SmExecutorConfig {
     fn default() -> Self {
         Self {
             command_queue_size: 256,
             known_peers: Vec::new(),
+            restore_interval_secs: None,
         }
     }
 }

crates/sm-executor/src/lib.rs

@@ -138,6 +138,13 @@ async fn recv_shutdown(
     }
 }
 
+async fn restore_timer_tick(interval: Option<std::time::Duration>) {
+    match interval {
+        Some(dur) => monoio::time::sleep(dur).await,
+        None => std::future::pending().await,
+    }
+}
+
 /// SM executor.
 #[derive(Debug)]
 pub struct SmExecutor<S>
@@ -196,6 +203,7 @@ where
             .name("sm-executor".to_string())
             .spawn(move || {
                 let mut runtime = monoio::RuntimeBuilder::<monoio::FusionDriver>::new()
+                    .enable_timer()
                     .build()
                     .expect("failed to build sm-executor monoio runtime");
                 let result = runtime.block_on(self.run_inner(Some(shutdown_rx)));
@@ -225,6 +233,12 @@ where
             self.restore_known_peers().await?;
             tracing::info!("sm executor restore completed; entering main loop");
 
+            let restore_interval = self
+                .config
+                .restore_interval_secs
+                .filter(|&s| s > 0)
+                .map(std::time::Duration::from_secs);
+
             loop {
                 monoio::select! {
                     shutdown = recv_shutdown(shutdown_rx.as_ref()) => {
@@ -301,6 +315,16 @@ where
                             }
                         }
                     }
+                    _ = restore_timer_tick(restore_interval) => {
+                        tracing::info!(source = "restore_tick", "periodic restore_known_peers triggered");
+                        if let Err(err) = self.restore_known_peers().await {
+                            if Self::is_fatal_processing_error(&err) {
+                                tracing::error!(source = "restore_tick", error = ?err, "fatal error during periodic restore; stopping sm executor");
+                                return Err(err);
+                            }
+                            tracing::warn!(source = "restore_tick", error = ?err, "periodic restore_known_peers failed; will retry at next interval");
+                        }
+                    }
                 }
             }
         }
@@ -1082,6 +1106,7 @@ mod tests {
         F: Future<Output = ()> + 'static,
     {
         monoio::RuntimeBuilder::<monoio::FusionDriver>::new()
+            .enable_timer()
             .build()
             .expect("build monoio runtime")
             .block_on(future);
@@ -1309,6 +1334,7 @@ mod tests {
             let config = SmExecutorConfig {
                 command_queue_size: 8,
                 known_peers: vec![peer_id],
+                restore_interval_secs: None,
             };
             let (executor, _handle) = SmExecutor::new(config, provider, job_handle, net_client);
 
@@ -1359,6 +1385,7 @@ mod tests {
             let config = SmExecutorConfig {
                 command_queue_size: 8,
                 known_peers: vec![peer_id],
+                restore_interval_secs: None,
             };
             let (executor, _handle) = SmExecutor::new(config, provider, job_handle, net_client);