diff --git a/crates/ember-core/src/keyspace/mod.rs b/crates/ember-core/src/keyspace/mod.rs index 149569c6..fa40ff3b 100644 --- a/crates/ember-core/src/keyspace/mod.rs +++ b/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, + 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). diff --git a/crates/ember-core/src/shard/mod.rs b/crates/ember-core/src/shard/mod.rs index d5877f7b..fc4a139c 100644 --- a/crates/ember-core/src/shard/mod.rs +++ b/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, + 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, + 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, } diff --git a/crates/ember-protocol/src/command/attributes.rs b/crates/ember-protocol/src/command/attributes.rs index 0658e3eb..b0a5459f 100644 --- a/crates/ember-protocol/src/command/attributes.rs +++ b/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, diff --git a/crates/ember-protocol/src/command/mod.rs b/crates/ember-protocol/src/command/mod.rs index 5be23a8e..92ffef20 100644 --- a/crates/ember-protocol/src/command/mod.rs +++ b/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 }, diff --git a/crates/ember-protocol/src/command/parse.rs b/crates/ember-protocol/src/command/parse.rs index f4d2d7ac..97143572 100644 --- a/crates/ember-protocol/src/command/parse.rs +++ b/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 Err(wrong_arity(cmd)), } } + +fn parse_wait(args: &[Frame]) -> Result { + 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::().map_err(|_| { + ProtocolError::InvalidCommandFrame("WAIT numreplicas must be an integer".into()) + })?; + let timeout_ms = timeout_ms_str.parse::().map_err(|_| { + ProtocolError::InvalidCommandFrame("WAIT timeout must be an integer".into()) + })?; + Ok(Command::Wait { + numreplicas, + timeout_ms, + }) +} diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs index 42afda37..63d0e86e 100644 --- a/crates/ember-server/src/cluster.rs +++ b/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) { + pub async fn start_replication_server( + self: &Arc, + tracker: Arc, + ) { 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}"); } diff --git a/crates/ember-server/src/config.rs b/crates/ember-server/src/config.rs index 972bcbf6..50fa7721 100644 --- a/crates/ember-server/src/config.rs +++ b/crates/ember-server/src/config.rs @@ -59,6 +59,14 @@ pub fn parse_byte_size(input: &str) -> Result { .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 { + parse_byte_size(input) +} + /// Parses an eviction policy name from a CLI string. pub fn parse_eviction_policy(input: &str) -> Result { 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 { + 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 -- diff --git a/crates/ember-server/src/connection/execute.rs b/crates/ember-server/src/connection/execute.rs index f7bb6893..6792f48e 100644 --- a/crates/ember-server/src/connection/execute.rs +++ b/crates/ember-server/src/connection/execute.rs @@ -494,6 +494,21 @@ pub(super) async fn execute( if let Ok(len) = value.parse::() { 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, 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, 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); + } + } +} diff --git a/crates/ember-server/src/replication.rs b/crates/ember-server/src/replication.rs index 900fb193..20bfeb45 100644 --- a/crates/ember-server/src/replication.rs +++ b/crates/ember-server/src/replication.rs @@ -28,9 +28,11 @@ //! [MSG_RESYNC: 1B] primary closes the connection; replica reconnects //! ``` +use std::collections::HashMap; use std::io; use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use bytes::Bytes; @@ -52,6 +54,75 @@ const MSG_SHARD_SYNC: u8 = 2; const MSG_SHARD_OFFSET: u8 = 3; const MSG_RECORD: u8 = 4; const MSG_RESYNC: u8 = 5; +const MSG_ACK: u8 = 6; + +/// Tracks per-replica acknowledged write offsets for the WAIT command. +/// +/// The primary increments `write_offset` for each record forwarded to +/// replicas. Each replica sends back MSG_ACK frames reporting its +/// current offset. WAIT polls `count_at_or_above(target)` with a +/// deadline to determine when enough replicas are in sync. +#[derive(Debug)] +pub struct ReplicaTracker { + /// Monotonically increasing counter of records forwarded by this primary. + pub write_offset: AtomicU64, + /// Per-replica last acknowledged offset. Keyed by a u64 replica ID + /// assigned sequentially at connection time. + offsets: Mutex>, + /// Next replica ID to assign. + next_id: AtomicU64, +} + +impl ReplicaTracker { + pub fn new() -> Self { + Self { + write_offset: AtomicU64::new(0), + offsets: Mutex::new(HashMap::new()), + next_id: AtomicU64::new(0), + } + } + + /// Registers a new replica connection. Returns its unique ID. + pub fn register(&self) -> u64 { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + if let Ok(mut map) = self.offsets.lock() { + map.insert(id, 0); + } + id + } + + /// Removes a replica connection from tracking. + pub fn remove(&self, replica_id: u64) { + if let Ok(mut map) = self.offsets.lock() { + map.remove(&replica_id); + } + } + + /// Updates the acknowledged offset for a replica. + /// + /// Only advances forward — never decrements. + pub fn update(&self, replica_id: u64, offset: u64) { + if let Ok(mut map) = self.offsets.lock() { + let entry = map.entry(replica_id).or_insert(0); + if offset > *entry { + *entry = offset; + } + } + } + + /// Returns the number of replicas whose acked offset is >= `target`. + pub fn count_at_or_above(&self, target: u64) -> usize { + self.offsets + .lock() + .map(|map| map.values().filter(|&&v| v >= target).count()) + .unwrap_or(0) + } + + /// Returns the total number of currently connected replicas. + pub fn connected_count(&self) -> usize { + self.offsets.lock().map(|map| map.len()).unwrap_or(0) + } +} // -- framed I/O primitives -- // @@ -129,6 +200,7 @@ async fn expect_tag( pub struct ReplicationServer { engine: Arc, primary_id: String, + tracker: Arc, } impl ReplicationServer { @@ -136,12 +208,21 @@ impl ReplicationServer { /// /// Runs indefinitely in the background; returns immediately after /// spawning the accept loop task. - pub async fn start(engine: Arc, primary_id: String, port: u16) -> std::io::Result<()> { + pub async fn start( + engine: Arc, + primary_id: String, + port: u16, + tracker: Arc, + ) -> std::io::Result<()> { let bind_addr = format!("0.0.0.0:{port}"); let listener = TcpListener::bind(&bind_addr).await?; info!(port, "replication server listening"); - let server = Arc::new(Self { engine, primary_id }); + let server = Arc::new(Self { + engine, + primary_id, + tracker, + }); tokio::spawn(async move { loop { @@ -173,7 +254,7 @@ impl ReplicationServer { // a single syscall. reads are delegated to the inner TcpStream. let mut stream = BufWriter::with_capacity(65536, stream); - // read replica handshake + // --- handshake --- let replica_version = read_u8(&mut stream).await?; if replica_version != REPL_VERSION { return Err(std::io::Error::new( @@ -184,7 +265,6 @@ impl ReplicationServer { let replica_shards = read_u16_le(&mut stream).await?; let our_shards = self.engine.shard_count() as u16; - // send primary handshake response write_u8(&mut stream, REPL_VERSION).await?; write_u16_le(&mut stream, our_shards).await?; let id_bytes = self.primary_id.as_bytes(); @@ -200,8 +280,8 @@ impl ReplicationServer { } write_u8(&mut stream, STATUS_OK).await?; - // subscribe to the broadcast channel before snapshotting so we don't - // miss events that happen between snapshot and stream start + // subscribe before snapshotting so we don't miss events that + // happen between snapshot and incremental stream start let mut rx = match self.engine.subscribe_replication() { Some(rx) => rx, None => { @@ -211,7 +291,7 @@ impl ReplicationServer { } }; - // full sync: snapshot each shard and send + // --- full sync --- for shard_idx in 0..self.engine.shard_count() { let resp = self .engine @@ -239,8 +319,6 @@ impl ReplicationServer { write_u32_le(&mut stream, data_len).await?; stream.write_all(&data).await?; - // send the current replication offset for this shard (0 until we - // track per-shard offsets; good enough for gap detection) write_u8(&mut stream, MSG_SHARD_OFFSET).await?; write_u16_le(&mut stream, shard_id).await?; write_u64_le(&mut stream, 0u64).await?; @@ -249,7 +327,47 @@ impl ReplicationServer { stream.flush().await?; info!("full sync complete, starting incremental stream"); - // incremental stream: relay events from the broadcast channel + // --- split stream for concurrent read (ACKs) and write (records) --- + // Take the inner TcpStream back from BufWriter, then split. + let inner = stream.into_inner(); + let (read_half, write_inner) = tokio::io::split(inner); + let mut writer = BufWriter::with_capacity(65536, write_inner); + + // Register this replica and spawn an ACK reader task. + let replica_id = self.tracker.register(); + let tracker = Arc::clone(&self.tracker); + let mut ack_reader = read_half; + + let ack_task = tokio::spawn(async move { + loop { + match read_u8(&mut ack_reader).await { + Ok(MSG_ACK) => match read_u64_le(&mut ack_reader).await { + Ok(offset) => tracker.update(replica_id, offset), + Err(_) => break, + }, + Ok(_) => {} // unknown or future message types — ignore + Err(_) => break, + } + } + }); + + // --- incremental stream --- + let result = self.stream_records(&mut writer, &mut rx).await; + + // clean up regardless of result + ack_task.abort(); + self.tracker.remove(replica_id); + + result + } + + /// Streams replication records to the replica until the connection closes + /// or the broadcast channel is exhausted. + async fn stream_records( + &self, + writer: &mut BufWriter>, + rx: &mut broadcast::Receiver, + ) -> std::io::Result<()> { loop { match rx.recv().await { Ok(event) => { @@ -260,19 +378,21 @@ impl ReplicationServer { std::io::Error::new(std::io::ErrorKind::InvalidData, "record too large") })?; - write_u8(&mut stream, MSG_RECORD).await?; - write_u16_le(&mut stream, event.shard_id).await?; - write_u64_le(&mut stream, event.offset).await?; - write_u32_le(&mut stream, record_len).await?; - stream.write_all(&record_bytes).await?; - // flush so the replica receives the record without waiting - // for the 64 KiB buffer to fill (reduces replication lag) - stream.flush().await?; + write_u8(writer, MSG_RECORD).await?; + write_u16_le(writer, event.shard_id).await?; + write_u64_le(writer, event.offset).await?; + write_u32_le(writer, record_len).await?; + writer.write_all(&record_bytes).await?; + writer.flush().await?; + + // advance the primary's write offset AFTER successfully + // flushing to the replica's TCP buffer + self.tracker.write_offset.fetch_add(1, Ordering::Relaxed); } Err(broadcast::error::RecvError::Lagged(count)) => { warn!("replication stream lagged by {count} events; triggering resync"); - let _ = write_u8(&mut stream, MSG_RESYNC).await; - let _ = stream.flush().await; + let _ = write_u8(writer, MSG_RESYNC).await; + let _ = writer.flush().await; return Ok(()); } Err(broadcast::error::RecvError::Closed) => { @@ -394,6 +514,7 @@ impl ReplicationClient { info!("full sync applied, starting incremental replay"); // incremental stream + let mut local_offset: u64 = 0; loop { let msg = read_u8(&mut stream).await?; match msg { @@ -418,6 +539,11 @@ impl ReplicationClient { } } } + + // acknowledge this record to the primary so WAIT can count us + local_offset += 1; + write_u8(&mut stream, MSG_ACK).await?; + write_u64_le(&mut stream, local_offset).await?; } MSG_RESYNC => { info!("primary requested resync; reconnecting"); diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 0811aefb..ed525961 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -80,6 +80,16 @@ pub struct ServerContext { pub monitor_tx: tokio::sync::broadcast::Sender, /// ACL state. `None` = legacy mode (no ACL, zero overhead). pub acl: crate::acl::SharedAclState, + /// Live memory limit in bytes. Updated atomically by CONFIG SET maxmemory. + /// 0 means unlimited. Used by INFO to display the current effective limit. + pub max_memory_limit: AtomicU64, + /// Tracks WAIT-command state: how many records this primary has forwarded + /// to replicas, and each replica's current acknowledged offset. + /// + /// Always present (even without cluster mode) but only meaningful when + /// there are active replica connections. The write_offset advances inside + /// the replication background task — zero overhead on the GET/SET path. + pub replica_tracker: Arc, /// Unix timestamp of the last successful save (BGSAVE/snapshot). pub last_save_timestamp: AtomicU64, /// Monotonically increasing client ID generator. @@ -290,6 +300,7 @@ pub async fn run_concurrent( let metrics_enabled = metrics.is_some(); let stats_poll_interval = limits.stats_poll_interval; let clients: ClientRegistry = Arc::new(Mutex::new(HashMap::new())); + let replica_tracker = Arc::new(crate::replication::ReplicaTracker::new()); let ctx = Arc::new(ServerContext { start_time: Instant::now(), version: env!("CARGO_PKG_VERSION"), @@ -310,6 +321,8 @@ pub async fn run_concurrent( memory_used_bytes: MemoryUsedBytes::new(), acl: None, monitor_tx: tokio::sync::broadcast::channel(256).0, + max_memory_limit: AtomicU64::new(max_memory.unwrap_or(0) as u64), + replica_tracker: Arc::clone(&replica_tracker), last_save_timestamp: AtomicU64::new(0), next_client_id: AtomicU64::new(1), clients, @@ -552,11 +565,15 @@ pub async fn run_threaded( let (engine, prepared_shards) = Engine::prepare(shard_count, config); + let replica_tracker = Arc::new(crate::replication::ReplicaTracker::new()); + // wire replication: give the cluster coordinator access to the engine // and start the replication server so replicas can connect if let Some(ref coordinator) = cluster { coordinator.set_engine(Arc::new(engine.clone())); - coordinator.start_replication_server().await; + coordinator + .start_replication_server(Arc::clone(&replica_tracker)) + .await; } let max_conn = max_connections.unwrap_or(DEFAULT_MAX_CONNECTIONS); @@ -585,6 +602,8 @@ pub async fn run_threaded( memory_used_bytes: MemoryUsedBytes::new(), acl: None, monitor_tx: tokio::sync::broadcast::channel(256).0, + max_memory_limit: AtomicU64::new(max_memory.unwrap_or(0) as u64), + replica_tracker: Arc::clone(&replica_tracker), last_save_timestamp: AtomicU64::new(0), next_client_id: AtomicU64::new(1), clients,