feat(server): automatic snapshot scheduling

crates/ember-server/src/config.rs

@@ -156,6 +156,9 @@ pub struct EmberConfig {
     pub active_expiry_interval_ms: u64,
     #[serde(rename = "aof-fsync-interval-secs")]
     pub aof_fsync_interval_secs: u64,
+    /// automatically trigger a background save every N seconds. 0 = disabled.
+    #[serde(rename = "save-interval-secs")]
+    pub save_interval_secs: u64,
 
     // -- monitoring --
     #[serde(rename = "metrics-port")]
@@ -220,6 +223,7 @@ impl Default for EmberConfig {
             appendfsync: "everysec".into(),
             active_expiry_interval_ms: 100,
             aof_fsync_interval_secs: 1,
+            save_interval_secs: 0,
 
             metrics_port: 0,
             slowlog_log_slower_than: 10_000,
@@ -435,6 +439,10 @@ impl EmberConfig {
             "aof-fsync-interval-secs".into(),
             self.aof_fsync_interval_secs.to_string(),
         );
+        params.insert(
+            "save-interval-secs".into(),
+            self.save_interval_secs.to_string(),
+        );
         params.insert("max-key-len".into(), self.max_key_len.clone());
         params.insert("max-value-len".into(), self.max_value_len.clone());
         params.insert(
@@ -724,6 +732,11 @@ impl ConfigRegistry {
             .parse()
             .map_err(|_| bad_value("aof-fsync-interval-secs", &raw))?;
 
+        let raw = get("save-interval-secs", "0");
+        cfg.save_interval_secs = raw
+            .parse()
+            .map_err(|_| bad_value("save-interval-secs", &raw))?;
+
         // monitoring
         let raw = get("slowlog-log-slower-than", "10000");
         cfg.slowlog_log_slower_than = raw

crates/ember-server/src/main.rs

@@ -74,6 +74,10 @@ struct Args {
     #[arg(long, env = "EMBER_APPENDFSYNC")]
     appendfsync: Option<String>,
 
+    /// automatically trigger a background snapshot every N seconds. 0 = disabled
+    #[arg(long, default_value_t = 0, env = "EMBER_SAVE_INTERVAL")]
+    save_interval: u64,
+
     /// port for prometheus metrics HTTP endpoint (0 = disabled)
     #[arg(long, env = "EMBER_METRICS_PORT")]
     metrics_port: Option<u16>,
@@ -227,6 +231,9 @@ fn apply_args(cfg: &mut config::EmberConfig, args: &Args) {
     if let Some(ref fsync) = args.appendfsync {
         cfg.appendfsync = fsync.clone();
     }
+    if args.save_interval > 0 {
+        cfg.save_interval_secs = args.save_interval;
+    }
     if let Some(port) = args.metrics_port {
         cfg.metrics_port = port;
     }
@@ -805,6 +812,7 @@ async fn main() {
             config_registry,
             limits,
             config_path,
+            cfg.save_interval_secs,
             #[cfg(feature = "grpc")]
             grpc_addr,
         )
@@ -828,6 +836,7 @@ async fn main() {
             config_registry,
             limits,
             config_path,
+            cfg.save_interval_secs,
             #[cfg(feature = "grpc")]
             grpc_addr,
         )

crates/ember-server/src/server.rs

@@ -10,7 +10,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
-use ember_core::{ConcurrentKeyspace, Engine, EngineConfig, EvictionPolicy};
+use ember_core::{ConcurrentKeyspace, Engine, EngineConfig, EvictionPolicy, ShardRequest};
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
@@ -266,6 +266,7 @@ pub async fn run_concurrent(
     config_registry: Arc<ConfigRegistry>,
     limits: ConnectionLimits,
     config_path: Option<PathBuf>,
+    save_interval_secs: u64,
     #[cfg(feature = "grpc")] grpc_addr: Option<SocketAddr>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let aof_enabled = config
@@ -322,6 +323,30 @@ pub async fn run_concurrent(
     let slow_log = Arc::new(SlowLog::new(slowlog_config));
     let pubsub = Arc::new(PubSubManager::new());
 
+    if save_interval_secs > 0 {
+        let engine_snap = engine.clone();
+        let ctx_snap = Arc::clone(&ctx);
+        tokio::spawn(async move {
+            let interval = Duration::from_secs(save_interval_secs);
+            loop {
+                tokio::time::sleep(interval).await;
+                match engine_snap.broadcast(|| ShardRequest::Snapshot).await {
+                    Ok(_) => {
+                        let ts = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .unwrap_or_default()
+                            .as_secs();
+                        ctx_snap
+                            .last_save_timestamp
+                            .store(ts, std::sync::atomic::Ordering::Relaxed);
+                        tracing::info!(interval = save_interval_secs, "automatic snapshot completed");
+                    }
+                    Err(e) => tracing::warn!("automatic snapshot failed: {e}"),
+                }
+            }
+        });
+    }
+
     // spawn gRPC listener if configured
     #[cfg(feature = "grpc")]
     let _grpc_handle = if let Some(grpc_addr) = grpc_addr {
@@ -501,6 +526,7 @@ pub async fn run_threaded(
     config_registry: Arc<ConfigRegistry>,
     limits: ConnectionLimits,
     config_path: Option<PathBuf>,
+    save_interval_secs: u64,
     #[cfg(feature = "grpc")] grpc_addr: Option<SocketAddr>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // ensure data directory exists if persistence is configured
@@ -567,6 +593,30 @@ pub async fn run_threaded(
     let slow_log = Arc::new(SlowLog::new(slowlog_config));
     let pubsub = Arc::new(PubSubManager::new());
 
+    if save_interval_secs > 0 {
+        let engine_snap = engine.clone();
+        let ctx_snap = Arc::clone(&ctx);
+        tokio::spawn(async move {
+            let interval = Duration::from_secs(save_interval_secs);
+            loop {
+                tokio::time::sleep(interval).await;
+                match engine_snap.broadcast(|| ShardRequest::Snapshot).await {
+                    Ok(_) => {
+                        let ts = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .unwrap_or_default()
+                            .as_secs();
+                        ctx_snap
+                            .last_save_timestamp
+                            .store(ts, std::sync::atomic::Ordering::Relaxed);
+                        tracing::info!(interval = save_interval_secs, "automatic snapshot completed");
+                    }
+                    Err(e) => tracing::warn!("automatic snapshot failed: {e}"),
+                }
+            }
+        });
+    }
+
     #[cfg(feature = "grpc")]
     let _grpc_handle = if let Some(grpc_addr) = grpc_addr {
         let svc = crate::grpc::EmberService::new(