feat: WAIT command and CONFIG SET live-apply (maxmemory, maxmemory-policy)

crates/ember-core/src/keyspace/mod.rs

@@ -1105,6 +1105,21 @@ impl Keyspace {
         Ok(true)
     }
 
+    /// Updates the memory limit and eviction policy in-place.
+    ///
+    /// Takes effect immediately for all subsequent write commands.
+    /// `track_access` is synchronized with the new policy so LRU
+    /// sampling stays consistent.
+    pub fn update_memory_config(
+        &mut self,
+        max_memory: Option<usize>,
+        eviction_policy: EvictionPolicy,
+    ) {
+        self.config.max_memory = max_memory;
+        self.config.eviction_policy = eviction_policy;
+        self.track_access = matches!(eviction_policy, EvictionPolicy::AllKeysLru);
+    }
+
     /// Returns aggregated stats for this keyspace.
     ///
     /// All fields are tracked incrementally — this is O(1).

crates/ember-core/src/shard/mod.rs

@@ -63,8 +63,8 @@ use crate::dropper::DropHandle;
 use crate::error::ShardError;
 use crate::expiry;
 use crate::keyspace::{
-    IncrError, IncrFloatError, Keyspace, KeyspaceStats, LsetError, SetResult, ShardConfig,
-    TtlResult, WriteError,
+    EvictionPolicy, IncrError, IncrFloatError, Keyspace, KeyspaceStats, LsetError, SetResult,
+    ShardConfig, TtlResult, WriteError,
 };
 use crate::types::sorted_set::{ScoreBound, ZAddFlags};
 use crate::types::Value;
@@ -484,6 +484,14 @@ pub enum ShardRequest {
     KeyVersion {
         key: String,
     },
+    /// Applies a live memory configuration update to this shard.
+    ///
+    /// Sent by the server when CONFIG SET maxmemory or maxmemory-policy
+    /// is changed at runtime. Takes effect on the next write check.
+    UpdateMemoryConfig {
+        max_memory: Option<usize>,
+        eviction_policy: EvictionPolicy,
+    },
     /// Triggers a snapshot write.
     Snapshot,
     /// Serializes the current shard state to bytes (in-memory snapshot).
@@ -1444,6 +1452,15 @@ fn process_single(mut request: ShardRequest, reply: ReplySender, ctx: &mut Proce
             reply.send(ShardResponse::Ok);
             return;
         }
+        RequestKind::UpdateMemoryConfig {
+            max_memory,
+            eviction_policy,
+        } => {
+            ctx.keyspace
+                .update_memory_config(max_memory, eviction_policy);
+            reply.send(ShardResponse::Ok);
+            return;
+        }
         RequestKind::Other => {}
     }
 
@@ -1457,6 +1474,10 @@ enum RequestKind {
     SerializeSnapshot,
     RewriteAof,
     FlushDbAsync,
+    UpdateMemoryConfig {
+        max_memory: Option<usize>,
+        eviction_policy: EvictionPolicy,
+    },
     Other,
 }
 
@@ -1466,6 +1487,13 @@ fn describe_request(req: &ShardRequest) -> RequestKind {
         ShardRequest::SerializeSnapshot => RequestKind::SerializeSnapshot,
         ShardRequest::RewriteAof => RequestKind::RewriteAof,
         ShardRequest::FlushDbAsync => RequestKind::FlushDbAsync,
+        ShardRequest::UpdateMemoryConfig {
+            max_memory,
+            eviction_policy,
+        } => RequestKind::UpdateMemoryConfig {
+            max_memory: *max_memory,
+            eviction_policy: *eviction_policy,
+        },
         _ => RequestKind::Other,
     }
 }
@@ -2175,6 +2203,7 @@ fn dispatch(
         | ShardRequest::SerializeSnapshot
         | ShardRequest::RewriteAof
         | ShardRequest::FlushDbAsync
+        | ShardRequest::UpdateMemoryConfig { .. }
         | ShardRequest::BLPop { .. }
         | ShardRequest::BRPop { .. } => ShardResponse::Ok,
     }

crates/ember-protocol/src/command/attributes.rs

@@ -68,6 +68,7 @@ impl Command {
             Command::Time => "time",
             Command::LastSave => "lastsave",
             Command::Role => "role",
+            Command::Wait { .. } => "wait",
             Command::BgSave => "bgsave",
             Command::BgRewriteAof => "bgrewriteaof",
             Command::FlushDb { .. } => "flushdb",
@@ -461,7 +462,9 @@ impl Command {
             // server
             Command::DbSize => SERVER | KEYSPACE | READ | FAST,
             Command::Info { .. } => SERVER | SLOW,
-            Command::Time | Command::LastSave | Command::Role => SERVER | FAST,
+            Command::Time | Command::LastSave | Command::Role | Command::Wait { .. } => {
+                SERVER | FAST
+            }
             Command::BgSave | Command::BgRewriteAof => SERVER | ADMIN | SLOW,
             Command::FlushDb { .. } => KEYSPACE | WRITE | ADMIN | DANGEROUS | SLOW,
             Command::ConfigGet { .. } => SERVER | ADMIN | SLOW,

crates/ember-protocol/src/command/mod.rs

@@ -723,6 +723,14 @@ pub enum Command {
     /// ROLE. Returns the replication role of the server.
     Role,
 
+    /// WAIT numreplicas timeout-ms
+    ///
+    /// Blocks until `numreplicas` replicas have acknowledged all write
+    /// commands processed before this WAIT, or until `timeout_ms`
+    /// milliseconds elapse. Returns the count of replicas that
+    /// acknowledged in time.
+    Wait { numreplicas: u64, timeout_ms: u64 },
+
     /// OBJECT ENCODING `key`. Returns the internal encoding of the value.
     ObjectEncoding { key: String },
 

crates/ember-protocol/src/command/parse.rs

@@ -212,6 +212,7 @@ impl Command {
             "TIME" => parse_no_args("TIME", &frames[1..], Command::Time),
             "LASTSAVE" => parse_no_args("LASTSAVE", &frames[1..], Command::LastSave),
             "ROLE" => parse_no_args("ROLE", &frames[1..], Command::Role),
+            "WAIT" => parse_wait(&frames[1..]),
             "OBJECT" => parse_object(&frames[1..]),
             "COPY" => parse_copy(&frames[1..]),
             "CLIENT" => parse_client(&frames[1..]),
@@ -3193,3 +3194,21 @@ fn parse_zset_multi(cmd: &'static str, args: &[Frame]) -> Result<Command, Protoc
         _ => Err(wrong_arity(cmd)),
     }
 }
+
+fn parse_wait(args: &[Frame]) -> Result<Command, ProtocolError> {
+    if args.len() != 2 {
+        return Err(wrong_arity("WAIT"));
+    }
+    let numreplicas_str = extract_string(&args[0])?;
+    let timeout_ms_str = extract_string(&args[1])?;
+    let numreplicas = numreplicas_str.parse::<u64>().map_err(|_| {
+        ProtocolError::InvalidCommandFrame("WAIT numreplicas must be an integer".into())
+    })?;
+    let timeout_ms = timeout_ms_str.parse::<u64>().map_err(|_| {
+        ProtocolError::InvalidCommandFrame("WAIT timeout must be an integer".into())
+    })?;
+    Ok(Command::Wait {
+        numreplicas,
+        timeout_ms,
+    })
+}

crates/ember-server/src/cluster.rs

@@ -1211,7 +1211,10 @@ impl ClusterCoordinator {
     /// Binds a TCP listener on `bind_addr.port() + gossip_port_offset + 2`
     /// and accepts replica connections indefinitely. This is a no-op if
     /// no engine has been attached via `set_engine`.
-    pub async fn start_replication_server(self: &Arc<Self>) {
+    pub async fn start_replication_server(
+        self: &Arc<Self>,
+        tracker: Arc<crate::replication::ReplicaTracker>,
+    ) {
         let Some(engine) = self.engine.get() else {
             warn!("start_replication_server called before set_engine; skipping");
             return;
@@ -1226,9 +1229,13 @@ impl ClusterCoordinator {
         };
 
         let local_id = self.local_id.to_string();
-        if let Err(e) =
-            crate::replication::ReplicationServer::start(Arc::clone(engine), local_id, repl_port)
-                .await
+        if let Err(e) = crate::replication::ReplicationServer::start(
+            Arc::clone(engine),
+            local_id,
+            repl_port,
+            tracker,
+        )
+        .await
         {
             error!("failed to start replication server on port {repl_port}: {e}");
         }

crates/ember-server/src/config.rs

@@ -59,6 +59,14 @@ pub fn parse_byte_size(input: &str) -> Result<usize, String> {
         .ok_or_else(|| format!("byte size overflow: '{input}'"))
 }
 
+/// Parses a memory size string. Accepts all formats from `parse_byte_size`
+/// plus "0" as a special value meaning "unlimited".
+///
+/// This is the right function for maxmemory values.
+pub fn parse_memory_size(input: &str) -> Result<usize, String> {
+    parse_byte_size(input)
+}
+
 /// Parses an eviction policy name from a CLI string.
 pub fn parse_eviction_policy(input: &str) -> Result<EvictionPolicy, String> {
     match input.to_ascii_lowercase().as_str() {
@@ -575,13 +583,16 @@ impl std::fmt::Debug for ConfigRegistry {
 ///
 /// Slowlog params take effect immediately. Connection-related params
 /// (maxclients, idle-timeout-secs, max-pipeline-depth) are stored in the
-/// registry and take effect for new connections.
+/// registry and take effect for new connections. Memory params are validated
+/// here and broadcast to shards by the server layer.
 const MUTABLE_PARAMS: &[&str] = &[
     "slowlog-log-slower-than",
     "slowlog-max-len",
     "maxclients",
     "idle-timeout-secs",
     "max-pipeline-depth",
+    "maxmemory",
+    "maxmemory-policy",
 ];
 
 impl ConfigRegistry {
@@ -661,6 +672,21 @@ impl ConfigRegistry {
                     format!("ERR Invalid argument '{value}' for CONFIG SET '{key}'")
                 })?;
             }
+            "maxmemory" => {
+                // Allow "0" (unlimited) or a memory string like "100M", "1G".
+                parse_memory_size(value).map_err(|_| {
+                    format!("ERR Invalid argument '{value}' for CONFIG SET '{key}'")
+                })?;
+            }
+            "maxmemory-policy" => match value.to_ascii_lowercase().as_str() {
+                "noeviction" | "allkeys-lru" => {}
+                _ => {
+                    return Err(format!(
+                        "ERR Invalid argument '{value}' for CONFIG SET '{key}': \
+                             supported policies are noeviction and allkeys-lru"
+                    ));
+                }
+            },
             _ => {}
         }
 
@@ -669,6 +695,27 @@ impl ConfigRegistry {
         Ok(())
     }
 
+    /// Returns the current effective memory limit in bytes, or `None` for unlimited.
+    pub fn memory_limit(&self) -> Option<usize> {
+        let params = self.params.read().unwrap_or_else(|e| e.into_inner());
+        params
+            .get("maxmemory")
+            .and_then(|v| parse_memory_size(v).ok())
+            .filter(|&n| n > 0)
+    }
+
+    /// Returns the current eviction policy.
+    pub fn eviction_policy(&self) -> EvictionPolicy {
+        let params = self.params.read().unwrap_or_else(|e| e.into_inner());
+        params
+            .get("maxmemory-policy")
+            .map(|v| match v.to_ascii_lowercase().as_str() {
+                "allkeys-lru" => EvictionPolicy::AllKeysLru,
+                _ => EvictionPolicy::NoEviction,
+            })
+            .unwrap_or(EvictionPolicy::NoEviction)
+    }
+
     /// Writes the current configuration to a TOML file.
     ///
     /// Reads all current parameter values, builds an `EmberConfig`,
@@ -953,12 +1000,25 @@ mod tests {
     }
 
     #[test]
-    fn config_registry_set_immutable_rejected() {
+    fn config_registry_set_maxmemory_accepted() {
         let mut params = HashMap::new();
-        params.insert("maxmemory".into(), "100".into());
+        params.insert("maxmemory".into(), "100M".into());
+        let registry = ConfigRegistry::new(params);
+
+        // maxmemory is mutable at runtime
+        assert!(registry.set("maxmemory", "200M").is_ok());
+        assert!(registry.set("maxmemory", "0").is_ok()); // 0 = unlimited
+        assert!(registry.set("maxmemory", "not-a-size").is_err());
+    }
+
+    #[test]
+    fn config_registry_set_immutable_rejected() {
+        let params = HashMap::new();
         let registry = ConfigRegistry::new(params);
 
-        assert!(registry.set("maxmemory", "200").is_err());
+        // truly immutable params are still rejected
+        assert!(registry.set("bind", "0.0.0.0").is_err());
+        assert!(registry.set("port", "6380").is_err());
     }
 
     // -- EmberConfig tests --

crates/ember-server/src/connection/execute.rs

@@ -494,6 +494,21 @@ pub(super) async fn execute(
                     if let Ok(len) = value.parse::<usize>() {
                         slow_log.update_max_len(len);
                     }
+                } else if key == "maxmemory" || key == "maxmemory-policy" {
+                    let limit = ctx.config.memory_limit();
+                    let policy = ctx.config.eviction_policy();
+                    // broadcast is fallible but config is already stored — log and continue
+                    let _ = engine
+                        .broadcast(move || ShardRequest::UpdateMemoryConfig {
+                            max_memory: limit,
+                            eviction_policy: policy,
+                        })
+                        .await;
+                    // keep the INFO-visible limit in sync
+                    ctx.max_memory_limit.store(
+                        limit.unwrap_or(0) as u64,
+                        std::sync::atomic::Ordering::Relaxed,
+                    );
                 }
                 Frame::Simple("OK".into())
             }
@@ -577,6 +592,11 @@ pub(super) async fn execute(
             }
         }
 
+        Command::Wait {
+            numreplicas,
+            timeout_ms,
+        } => handle_wait(ctx, numreplicas, timeout_ms).await,
+
         Command::FlushDb { async_mode } => {
             let req = if async_mode {
                 || ShardRequest::FlushDbAsync
@@ -2781,10 +2801,13 @@ async fn render_info(engine: &Engine, ctx: &Arc<ServerContext>, section: Option<
                 out.push_str(&format!("used_memory_rss:{rss}\r\n"));
                 out.push_str(&format!("used_memory_rss_human:{}\r\n", human_bytes(rss)));
             }
-            if let Some(max) = ctx.max_memory {
-                let effective = ember_core::memory::effective_limit(max);
-                out.push_str(&format!("max_memory:{max}\r\n"));
-                out.push_str(&format!("max_memory_human:{}\r\n", human_bytes(max)));
+            let max_bytes = ctx
+                .max_memory_limit
+                .load(std::sync::atomic::Ordering::Relaxed) as usize;
+            if max_bytes > 0 {
+                let effective = ember_core::memory::effective_limit(max_bytes);
+                out.push_str(&format!("max_memory:{max_bytes}\r\n"));
+                out.push_str(&format!("max_memory_human:{}\r\n", human_bytes(max_bytes)));
                 out.push_str(&format!("max_memory_effective:{effective}\r\n"));
                 out.push_str(&format!(
                     "max_memory_effective_human:{}\r\n",
@@ -2902,3 +2925,41 @@ pub(super) fn resolve_collection_scan(
 pub(super) fn oom_error() -> Frame {
     Frame::Error("OOM command not allowed when used memory > 'maxmemory'".into())
 }
+
+/// Implements the WAIT command: blocks until `needed` replicas have
+/// acknowledged all writes at or before the current primary offset,
+/// or until `timeout_ms` milliseconds elapse.
+///
+/// Returns the count of replicas that acknowledged in time as a
+/// RESP integer. When there are no replicas or no writes, returns
+/// immediately without sleeping.
+async fn handle_wait(ctx: &Arc<ServerContext>, numreplicas: u64, timeout_ms: u64) -> Frame {
+    use std::sync::atomic::Ordering;
+    use std::time::Duration;
+
+    let needed = numreplicas as usize;
+    let tracker = &ctx.replica_tracker;
+
+    // fast path: no replicas connected
+    if tracker.connected_count() == 0 {
+        return Frame::Integer(0);
+    }
+
+    let target = tracker.write_offset.load(Ordering::Relaxed);
+
+    // fast path: already satisfied or no timeout needed
+    let count = tracker.count_at_or_above(target);
+    if count >= needed || timeout_ms == 0 {
+        return Frame::Integer(count as i64);
+    }
+
+    // poll until enough replicas have caught up or the deadline passes
+    let deadline = tokio::time::Instant::now() + Duration::from_millis(timeout_ms);
+    loop {
+        tokio::time::sleep(Duration::from_millis(25)).await;
+        let c = tracker.count_at_or_above(target);
+        if c >= needed || tokio::time::Instant::now() >= deadline {
+            return Frame::Integer(c as i64);
+        }
+    }
+}