diff --git a/crates/ember-server/src/config.rs b/crates/ember-server/src/config.rs index 0bc3e36b..972bcbf6 100644 --- a/crates/ember-server/src/config.rs +++ b/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 diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index 09c3e010..ae3f9b07 100644 --- a/crates/ember-server/src/main.rs +++ b/crates/ember-server/src/main.rs @@ -74,6 +74,10 @@ struct Args { #[arg(long, env = "EMBER_APPENDFSYNC")] appendfsync: Option, + /// 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, @@ -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, ) diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 20db5971..f52cdc2f 100644 --- a/crates/ember-server/src/server.rs +++ b/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, limits: ConnectionLimits, config_path: Option, + save_interval_secs: u64, #[cfg(feature = "grpc")] grpc_addr: Option, ) -> Result<(), Box> { 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, limits: ConnectionLimits, config_path: Option, + save_interval_secs: u64, #[cfg(feature = "grpc")] grpc_addr: Option, ) -> Result<(), Box> { // 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(