From 03fc3f8e345cf28dd320dcacca3e2b96b7c5c20a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 09:16:03 -0500 Subject: [PATCH 1/5] feat: add INCRBY and DECRBY commands increment/decrement integer values by arbitrary amounts. makes incr_by public in keyspace so INCRBY/DECRBY route through it directly. includes full AOF persistence and recovery support. --- crates/ember-core/src/keyspace.rs | 6 +- crates/ember-core/src/shard.rs | 107 +++++++++++++++++++++++ crates/ember-persistence/src/aof.rs | 26 ++++++ crates/ember-persistence/src/recovery.rs | 6 ++ crates/ember-protocol/src/command.rs | 81 +++++++++++++++++ crates/ember-server/src/connection.rs | 38 ++++++-- 6 files changed, 255 insertions(+), 9 deletions(-) diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index 94e8b832..c70314e3 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -500,11 +500,11 @@ impl Keyspace { self.incr_by(key, -1) } - /// Shared implementation for INCR/DECR. Adds `delta` to the current - /// integer value of the key, creating it if necessary. + /// Adds `delta` to the current integer value of the key, creating it + /// if necessary. Used by INCR, DECR, INCRBY, and DECRBY. /// /// Preserves the existing TTL when updating an existing key. - fn incr_by(&mut self, key: &str, delta: i64) -> Result { + pub fn incr_by(&mut self, key: &str, delta: i64) -> Result { self.remove_if_expired(key); // read current value and TTL diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index 4434c0f6..3ea41b2b 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -62,6 +62,14 @@ pub enum ShardRequest { Decr { key: String, }, + IncrBy { + key: String, + delta: i64, + }, + DecrBy { + key: String, + delta: i64, + }, Del { key: String, }, @@ -518,6 +526,18 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory, Err(e) => ShardResponse::Err(e.to_string()), }, + ShardRequest::IncrBy { key, delta } => match ks.incr_by(key, *delta) { + Ok(val) => ShardResponse::Integer(val), + Err(IncrError::WrongType) => ShardResponse::WrongType, + Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory, + Err(e) => ShardResponse::Err(e.to_string()), + }, + ShardRequest::DecrBy { key, delta } => match ks.incr_by(key, -delta) { + Ok(val) => ShardResponse::Integer(val), + Err(IncrError::WrongType) => ShardResponse::WrongType, + Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory, + Err(e) => ShardResponse::Err(e.to_string()), + }, ShardRequest::Del { key } => ShardResponse::Bool(ks.del(key)), ShardRequest::Exists { key } => ShardResponse::Bool(ks.exists(key)), ShardRequest::Expire { key, seconds } => ShardResponse::Bool(ks.expire(key, *seconds)), @@ -755,6 +775,18 @@ fn to_aof_record(req: &ShardRequest, resp: &ShardResponse) -> Option (ShardRequest::Decr { key }, ShardResponse::Integer(_)) => { Some(AofRecord::Decr { key: key.clone() }) } + (ShardRequest::IncrBy { key, delta }, ShardResponse::Integer(_)) => { + Some(AofRecord::IncrBy { + key: key.clone(), + delta: *delta, + }) + } + (ShardRequest::DecrBy { key, delta }, ShardResponse::Integer(_)) => { + Some(AofRecord::DecrBy { + key: key.clone(), + delta: *delta, + }) + } (ShardRequest::Persist { key }, ShardResponse::Bool(true)) => { Some(AofRecord::Persist { key: key.clone() }) } @@ -1245,6 +1277,47 @@ mod tests { assert!(matches!(resp, ShardResponse::Err(_))); } + #[test] + fn dispatch_incrby() { + let mut ks = Keyspace::new(); + ks.set("n".into(), Bytes::from("10"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::IncrBy { + key: "n".into(), + delta: 5, + }, + ); + assert!(matches!(resp, ShardResponse::Integer(15))); + } + + #[test] + fn dispatch_decrby() { + let mut ks = Keyspace::new(); + ks.set("n".into(), Bytes::from("10"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::DecrBy { + key: "n".into(), + delta: 3, + }, + ); + assert!(matches!(resp, ShardResponse::Integer(7))); + } + + #[test] + fn dispatch_incrby_new_key() { + let mut ks = Keyspace::new(); + let resp = dispatch( + &mut ks, + &ShardRequest::IncrBy { + key: "new".into(), + delta: 42, + }, + ); + assert!(matches!(resp, ShardResponse::Integer(42))); + } + #[test] fn to_aof_record_for_incr() { let req = ShardRequest::Incr { key: "c".into() }; @@ -1261,6 +1334,40 @@ mod tests { assert!(matches!(record, AofRecord::Decr { .. })); } + #[test] + fn to_aof_record_for_incrby() { + let req = ShardRequest::IncrBy { + key: "c".into(), + delta: 5, + }; + let resp = ShardResponse::Integer(15); + let record = to_aof_record(&req, &resp).unwrap(); + match record { + AofRecord::IncrBy { key, delta } => { + assert_eq!(key, "c"); + assert_eq!(delta, 5); + } + other => panic!("expected IncrBy, got {other:?}"), + } + } + + #[test] + fn to_aof_record_for_decrby() { + let req = ShardRequest::DecrBy { + key: "c".into(), + delta: 3, + }; + let resp = ShardResponse::Integer(7); + let record = to_aof_record(&req, &resp).unwrap(); + match record { + AofRecord::DecrBy { key, delta } => { + assert_eq!(key, "c"); + assert_eq!(delta, 3); + } + other => panic!("expected DecrBy, got {other:?}"), + } + } + #[test] fn dispatch_persist_removes_ttl() { let mut ks = Keyspace::new(); diff --git a/crates/ember-persistence/src/aof.rs b/crates/ember-persistence/src/aof.rs index 4fd0d4b2..5f2549eb 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -54,6 +54,8 @@ const TAG_HDEL: u8 = 15; const TAG_HINCRBY: u8 = 16; const TAG_SADD: u8 = 17; const TAG_SREM: u8 = 18; +const TAG_INCRBY: u8 = 19; +const TAG_DECRBY: u8 = 20; /// A single mutation record stored in the AOF. #[derive(Debug, Clone, PartialEq)] @@ -108,6 +110,10 @@ pub enum AofRecord { SAdd { key: String, members: Vec }, /// SREM key member [member ...]. SRem { key: String, members: Vec }, + /// INCRBY key delta. + IncrBy { key: String, delta: i64 }, + /// DECRBY key delta. + DecrBy { key: String, delta: i64 }, } impl AofRecord { @@ -231,6 +237,16 @@ impl AofRecord { format::write_bytes(&mut buf, member.as_bytes()).expect("vec write"); } } + AofRecord::IncrBy { key, delta } => { + format::write_u8(&mut buf, TAG_INCRBY).expect("vec write"); + format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); + format::write_i64(&mut buf, *delta).expect("vec write"); + } + AofRecord::DecrBy { key, delta } => { + format::write_u8(&mut buf, TAG_DECRBY).expect("vec write"); + format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); + format::write_i64(&mut buf, *delta).expect("vec write"); + } } buf } @@ -361,6 +377,16 @@ impl AofRecord { } Ok(AofRecord::SRem { key, members }) } + TAG_INCRBY => { + let key = read_string(&mut cursor, "key")?; + let delta = format::read_i64(&mut cursor)?; + Ok(AofRecord::IncrBy { key, delta }) + } + TAG_DECRBY => { + let key = read_string(&mut cursor, "key")?; + let delta = format::read_i64(&mut cursor)?; + Ok(AofRecord::DecrBy { key, delta }) + } _ => Err(FormatError::UnknownTag(tag)), } } diff --git a/crates/ember-persistence/src/recovery.rs b/crates/ember-persistence/src/recovery.rs index eaa86656..c4116821 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/crates/ember-persistence/src/recovery.rs @@ -285,6 +285,12 @@ fn replay_aof( AofRecord::Decr { key } => { apply_incr(map, key, -1); } + AofRecord::IncrBy { key, delta } => { + apply_incr(map, key, delta); + } + AofRecord::DecrBy { key, delta } => { + apply_incr(map, key, -delta); + } AofRecord::HSet { key, fields } => { let entry = map .entry(key) diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 7247e4ec..e0391696 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -47,6 +47,12 @@ pub enum Command { /// DECR `key`. Decrements the integer value of a key by 1. Decr { key: String }, + /// INCRBY `key` `increment`. Increments the integer value of a key by the given amount. + IncrBy { key: String, delta: i64 }, + + /// DECRBY `key` `decrement`. Decrements the integer value of a key by the given amount. + DecrBy { key: String, delta: i64 }, + /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, @@ -331,6 +337,8 @@ impl Command { Command::Set { .. } => "set", Command::Incr { .. } => "incr", Command::Decr { .. } => "decr", + Command::IncrBy { .. } => "incrby", + Command::DecrBy { .. } => "decrby", Command::Del { .. } => "del", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", @@ -438,6 +446,8 @@ impl Command { "SET" => parse_set(&frames[1..]), "INCR" => parse_incr(&frames[1..]), "DECR" => parse_decr(&frames[1..]), + "INCRBY" => parse_incrby(&frames[1..]), + "DECRBY" => parse_decrby(&frames[1..]), "DEL" => parse_del(&frames[1..]), "EXISTS" => parse_exists(&frames[1..]), "MGET" => parse_mget(&frames[1..]), @@ -646,6 +656,24 @@ fn parse_decr(args: &[Frame]) -> Result { Ok(Command::Decr { key }) } +fn parse_incrby(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("INCRBY".into())); + } + let key = extract_string(&args[0])?; + let delta = parse_i64(&args[1], "INCRBY")?; + Ok(Command::IncrBy { key, delta }) +} + +fn parse_decrby(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("DECRBY".into())); + } + let key = extract_string(&args[0])?; + let delta = parse_i64(&args[1], "DECRBY")?; + Ok(Command::DecrBy { key, delta }) +} + fn parse_del(args: &[Frame]) -> Result { if args.is_empty() { return Err(ProtocolError::WrongArity("DEL".into())); @@ -3499,4 +3527,57 @@ mod tests { let err = Command::from_frame(cmd(&["PUBSUB", "BOGUS"])).unwrap_err(); assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); } + + // --- INCRBY / DECRBY --- + + #[test] + fn incrby_basic() { + assert_eq!( + Command::from_frame(cmd(&["INCRBY", "counter", "5"])).unwrap(), + Command::IncrBy { + key: "counter".into(), + delta: 5 + }, + ); + } + + #[test] + fn incrby_negative() { + assert_eq!( + Command::from_frame(cmd(&["INCRBY", "counter", "-3"])).unwrap(), + Command::IncrBy { + key: "counter".into(), + delta: -3 + }, + ); + } + + #[test] + fn incrby_wrong_arity() { + let err = Command::from_frame(cmd(&["INCRBY", "key"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn incrby_not_integer() { + let err = Command::from_frame(cmd(&["INCRBY", "key", "abc"])).unwrap_err(); + assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); + } + + #[test] + fn decrby_basic() { + assert_eq!( + Command::from_frame(cmd(&["DECRBY", "counter", "10"])).unwrap(), + Command::DecrBy { + key: "counter".into(), + delta: 10 + }, + ); + } + + #[test] + fn decrby_wrong_arity() { + let err = Command::from_frame(cmd(&["DECRBY"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index fd1e1e17..16ac608e 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -555,9 +555,7 @@ async fn execute( match engine.route(&key, req).await { Ok(ShardResponse::Integer(n)) => Frame::Integer(n), Ok(ShardResponse::WrongType) => wrongtype_error(), - Ok(ShardResponse::OutOfMemory) => { - Frame::Error("OOM command not allowed when used memory > 'maxmemory'".into()) - } + Ok(ShardResponse::OutOfMemory) => oom_error(), Ok(ShardResponse::Err(msg)) => Frame::Error(msg), Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), Err(e) => Frame::Error(format!("ERR {e}")), @@ -569,9 +567,37 @@ async fn execute( match engine.route(&key, req).await { Ok(ShardResponse::Integer(n)) => Frame::Integer(n), Ok(ShardResponse::WrongType) => wrongtype_error(), - Ok(ShardResponse::OutOfMemory) => { - Frame::Error("OOM command not allowed when used memory > 'maxmemory'".into()) - } + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(ShardResponse::Err(msg)) => Frame::Error(msg), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + + Command::IncrBy { key, delta } => { + let req = ShardRequest::IncrBy { + key: key.clone(), + delta, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Integer(n)) => Frame::Integer(n), + Ok(ShardResponse::WrongType) => wrongtype_error(), + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(ShardResponse::Err(msg)) => Frame::Error(msg), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + + Command::DecrBy { key, delta } => { + let req = ShardRequest::DecrBy { + key: key.clone(), + delta, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Integer(n)) => Frame::Integer(n), + Ok(ShardResponse::WrongType) => wrongtype_error(), + Ok(ShardResponse::OutOfMemory) => oom_error(), Ok(ShardResponse::Err(msg)) => Frame::Error(msg), Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), Err(e) => Frame::Error(format!("ERR {e}")), From 05919b64b46dde82f6c733c1e1187edc602e5c64 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 09:19:19 -0500 Subject: [PATCH 2/5] feat: add INCRBYFLOAT command increment float values by arbitrary amounts. adds IncrFloatError type and incr_by_float method to keyspace. persists as SET records in AOF to avoid float rounding drift during replay. --- crates/ember-core/src/keyspace.rs | 153 ++++++++++++++++++++++++++ crates/ember-core/src/lib.rs | 4 +- crates/ember-core/src/shard.rs | 63 ++++++++++- crates/ember-protocol/src/command.rs | 60 ++++++++++ crates/ember-server/src/connection.rs | 15 +++ 5 files changed, 292 insertions(+), 3 deletions(-) diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index c70314e3..f27f7df3 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -77,6 +77,42 @@ impl std::fmt::Display for IncrError { } impl std::error::Error for IncrError {} + +/// Errors that can occur during INCRBYFLOAT operations. +#[derive(Debug, Clone, PartialEq)] +pub enum IncrFloatError { + /// Key holds a non-string type. + WrongType, + /// Value is not a valid float. + NotAFloat, + /// Result would be NaN or Infinity. + NanOrInfinity, + /// Memory limit reached. + OutOfMemory, +} + +impl std::fmt::Display for IncrFloatError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + IncrFloatError::WrongType => write!( + f, + "WRONGTYPE Operation against a key holding the wrong kind of value" + ), + IncrFloatError::NotAFloat => { + write!(f, "ERR value is not a valid float") + } + IncrFloatError::NanOrInfinity => { + write!(f, "ERR increment would produce NaN or Infinity") + } + IncrFloatError::OutOfMemory => { + write!(f, "OOM command not allowed when used memory > 'maxmemory'") + } + } + } +} + +impl std::error::Error for IncrFloatError {} + /// Result of a ZADD operation, containing both the client-facing count /// and the list of members that were actually applied (for AOF correctness). #[derive(Debug, Clone)] @@ -532,6 +568,46 @@ impl Keyspace { } } + /// Adds a float `delta` to the current value of the key, creating it + /// if necessary. Used by INCRBYFLOAT. + /// + /// Preserves the existing TTL when updating an existing key. + /// Returns the new value as a string (matching Redis behavior). + pub fn incr_by_float(&mut self, key: &str, delta: f64) -> Result { + self.remove_if_expired(key); + + let (current, existing_expire) = match self.entries.get(key) { + Some(entry) => { + let val = match &entry.value { + Value::String(data) => { + let s = std::str::from_utf8(data) + .map_err(|_| IncrFloatError::NotAFloat)?; + s.parse::().map_err(|_| IncrFloatError::NotAFloat)? + } + _ => return Err(IncrFloatError::WrongType), + }; + let expire = time::remaining_ms(entry.expires_at_ms).map(Duration::from_millis); + (val, expire) + } + None => (0.0, None), + }; + + let new_val = current + delta; + if new_val.is_nan() || new_val.is_infinite() { + return Err(IncrFloatError::NanOrInfinity); + } + + // Redis strips trailing zeros: "10.5" not "10.50000..." + // but keeps at least one decimal if the result is a whole number + let formatted = format_float(new_val); + let new_bytes = Bytes::from(formatted.clone()); + + match self.set(key.to_owned(), new_bytes, existing_expire) { + SetResult::Ok => Ok(formatted), + SetResult::OutOfMemory => Err(IncrFloatError::OutOfMemory), + } + } + /// Returns aggregated stats for this keyspace. /// /// All fields are tracked incrementally — this is O(1). @@ -1581,6 +1657,28 @@ impl Default for Keyspace { /// - `*` matches any sequence of characters (including empty) /// - `?` matches exactly one character /// - `[abc]` matches one character from the set +/// Formats a float value matching Redis behavior. +/// +/// Uses up to 17 significant digits and strips unnecessary trailing zeros, +/// but always keeps at least one decimal place for non-integer results. +fn format_float(val: f64) -> String { + if val == 0.0 { + return "0".into(); + } + // Use enough precision to round-trip + let s = format!("{:.17e}", val); + // Parse back to get the clean representation + let reparsed: f64 = s.parse().unwrap_or(val); + // If it's a whole number, format without decimals + if reparsed == reparsed.trunc() && reparsed.abs() < 1e15 { + format!("{}", reparsed as i64) + } else { + // Use ryu-like formatting via Display which strips trailing zeros + let formatted = format!("{}", reparsed); + formatted + } +} + /// - `[^abc]` or `[!abc]` matches one character NOT in the set /// /// Uses an iterative two-pointer algorithm with backtracking for O(n*m) @@ -3433,4 +3531,59 @@ mod tests { ks.set("binary".into(), binary.clone(), None); assert_eq!(ks.get("binary").unwrap(), Some(Value::String(binary))); } + + #[test] + fn incr_by_float_basic() { + let mut ks = Keyspace::new(); + ks.set("n".into(), Bytes::from("10.5"), None); + let result = ks.incr_by_float("n", 2.3).unwrap(); + let f: f64 = result.parse().unwrap(); + assert!((f - 12.8).abs() < 0.001); + } + + #[test] + fn incr_by_float_new_key() { + let mut ks = Keyspace::new(); + let result = ks.incr_by_float("new", 3.14).unwrap(); + let f: f64 = result.parse().unwrap(); + assert!((f - 3.14).abs() < 0.001); + } + + #[test] + fn incr_by_float_negative() { + let mut ks = Keyspace::new(); + ks.set("n".into(), Bytes::from("10"), None); + let result = ks.incr_by_float("n", -3.5).unwrap(); + let f: f64 = result.parse().unwrap(); + assert!((f - 6.5).abs() < 0.001); + } + + #[test] + fn incr_by_float_wrong_type() { + let mut ks = Keyspace::new(); + ks.lpush("mylist", &[Bytes::from("a")]).unwrap(); + let err = ks.incr_by_float("mylist", 1.0).unwrap_err(); + assert_eq!(err, IncrFloatError::WrongType); + } + + #[test] + fn incr_by_float_not_a_float() { + let mut ks = Keyspace::new(); + ks.set("s".into(), Bytes::from("hello"), None); + let err = ks.incr_by_float("s", 1.0).unwrap_err(); + assert_eq!(err, IncrFloatError::NotAFloat); + } + + #[test] + fn format_float_integers() { + assert_eq!(super::format_float(10.0), "10"); + assert_eq!(super::format_float(0.0), "0"); + assert_eq!(super::format_float(-5.0), "-5"); + } + + #[test] + fn format_float_decimals() { + assert_eq!(super::format_float(3.14), "3.14"); + assert_eq!(super::format_float(10.5), "10.5"); + } } diff --git a/crates/ember-core/src/lib.rs b/crates/ember-core/src/lib.rs index 10c2513d..dec71c4d 100644 --- a/crates/ember-core/src/lib.rs +++ b/crates/ember-core/src/lib.rs @@ -18,8 +18,8 @@ pub use concurrent::ConcurrentKeyspace; pub use engine::{Engine, EngineConfig}; pub use error::ShardError; pub use keyspace::{ - EvictionPolicy, IncrError, Keyspace, KeyspaceStats, ShardConfig, TtlResult, WriteError, - WrongType, ZAddResult, + EvictionPolicy, IncrError, IncrFloatError, Keyspace, KeyspaceStats, ShardConfig, TtlResult, + WriteError, WrongType, ZAddResult, }; pub use shard::{ShardPersistenceConfig, ShardRequest, ShardResponse}; pub use types::Value; diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index 3ea41b2b..4e88e636 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -18,7 +18,8 @@ use tracing::{info, warn}; use crate::error::ShardError; use crate::expiry; use crate::keyspace::{ - IncrError, Keyspace, KeyspaceStats, SetResult, ShardConfig, TtlResult, WriteError, + IncrError, IncrFloatError, Keyspace, KeyspaceStats, SetResult, ShardConfig, TtlResult, + WriteError, }; use crate::types::sorted_set::ZAddFlags; use crate::types::Value; @@ -70,6 +71,10 @@ pub enum ShardRequest { key: String, delta: i64, }, + IncrByFloat { + key: String, + delta: f64, + }, Del { key: String, }, @@ -259,6 +264,8 @@ pub enum ShardResponse { Rank(Option), /// Scored array of (member, score) pairs (e.g. ZRANGE). ScoredArray(Vec<(String, f64)>), + /// A bulk string result (e.g. INCRBYFLOAT). + BulkString(String), /// Command used against a key holding the wrong kind of value. WrongType, /// An error message. @@ -538,6 +545,12 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory, Err(e) => ShardResponse::Err(e.to_string()), }, + ShardRequest::IncrByFloat { key, delta } => match ks.incr_by_float(key, *delta) { + Ok(val) => ShardResponse::BulkString(val), + Err(IncrFloatError::WrongType) => ShardResponse::WrongType, + Err(IncrFloatError::OutOfMemory) => ShardResponse::OutOfMemory, + Err(e) => ShardResponse::Err(e.to_string()), + }, ShardRequest::Del { key } => ShardResponse::Bool(ks.del(key)), ShardRequest::Exists { key } => ShardResponse::Bool(ks.exists(key)), ShardRequest::Expire { key, seconds } => ShardResponse::Bool(ks.expire(key, *seconds)), @@ -787,6 +800,15 @@ fn to_aof_record(req: &ShardRequest, resp: &ShardResponse) -> Option delta: *delta, }) } + // INCRBYFLOAT: record as a SET with the resulting value to avoid + // float rounding drift during replay. + (ShardRequest::IncrByFloat { key, .. }, ShardResponse::BulkString(val)) => { + Some(AofRecord::Set { + key: key.clone(), + value: Bytes::from(val.clone()), + expire_ms: -1, // preserve existing TTL via the SET path + }) + } (ShardRequest::Persist { key }, ShardResponse::Bool(true)) => { Some(AofRecord::Persist { key: key.clone() }) } @@ -1318,6 +1340,45 @@ mod tests { assert!(matches!(resp, ShardResponse::Integer(42))); } + #[test] + fn dispatch_incrbyfloat() { + let mut ks = Keyspace::new(); + ks.set("n".into(), Bytes::from("10.5"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::IncrByFloat { + key: "n".into(), + delta: 2.3, + }, + ); + match resp { + ShardResponse::BulkString(val) => { + let f: f64 = val.parse().unwrap(); + assert!((f - 12.8).abs() < 0.001); + } + other => panic!("expected BulkString, got {other:?}"), + } + } + + #[test] + fn dispatch_incrbyfloat_new_key() { + let mut ks = Keyspace::new(); + let resp = dispatch( + &mut ks, + &ShardRequest::IncrByFloat { + key: "new".into(), + delta: 3.14, + }, + ); + match resp { + ShardResponse::BulkString(val) => { + let f: f64 = val.parse().unwrap(); + assert!((f - 3.14).abs() < 0.001); + } + other => panic!("expected BulkString, got {other:?}"), + } + } + #[test] fn to_aof_record_for_incr() { let req = ShardRequest::Incr { key: "c".into() }; diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index e0391696..725c3c92 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -53,6 +53,9 @@ pub enum Command { /// DECRBY `key` `decrement`. Decrements the integer value of a key by the given amount. DecrBy { key: String, delta: i64 }, + /// INCRBYFLOAT `key` `increment`. Increments the float value of a key by the given amount. + IncrByFloat { key: String, delta: f64 }, + /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, @@ -339,6 +342,7 @@ impl Command { Command::Decr { .. } => "decr", Command::IncrBy { .. } => "incrby", Command::DecrBy { .. } => "decrby", + Command::IncrByFloat { .. } => "incrbyfloat", Command::Del { .. } => "del", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", @@ -448,6 +452,7 @@ impl Command { "DECR" => parse_decr(&frames[1..]), "INCRBY" => parse_incrby(&frames[1..]), "DECRBY" => parse_decrby(&frames[1..]), + "INCRBYFLOAT" => parse_incrbyfloat(&frames[1..]), "DEL" => parse_del(&frames[1..]), "EXISTS" => parse_exists(&frames[1..]), "MGET" => parse_mget(&frames[1..]), @@ -674,6 +679,23 @@ fn parse_decrby(args: &[Frame]) -> Result { Ok(Command::DecrBy { key, delta }) } +fn parse_incrbyfloat(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("INCRBYFLOAT".into())); + } + let key = extract_string(&args[0])?; + let s = extract_string(&args[1])?; + let delta: f64 = s.parse().map_err(|_| { + ProtocolError::InvalidCommandFrame("value is not a valid float for 'INCRBYFLOAT'".into()) + })?; + if delta.is_nan() || delta.is_infinite() { + return Err(ProtocolError::InvalidCommandFrame( + "increment would produce NaN or Infinity".into(), + )); + } + Ok(Command::IncrByFloat { key, delta }) +} + fn parse_del(args: &[Frame]) -> Result { if args.is_empty() { return Err(ProtocolError::WrongArity("DEL".into())); @@ -3580,4 +3602,42 @@ mod tests { let err = Command::from_frame(cmd(&["DECRBY"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } + + // --- INCRBYFLOAT --- + + #[test] + fn incrbyfloat_basic() { + let cmd = Command::from_frame(cmd(&["INCRBYFLOAT", "key", "2.5"])).unwrap(); + match cmd { + Command::IncrByFloat { key, delta } => { + assert_eq!(key, "key"); + assert!((delta - 2.5).abs() < f64::EPSILON); + } + other => panic!("expected IncrByFloat, got {other:?}"), + } + } + + #[test] + fn incrbyfloat_negative() { + let cmd = Command::from_frame(cmd(&["INCRBYFLOAT", "key", "-1.5"])).unwrap(); + match cmd { + Command::IncrByFloat { key, delta } => { + assert_eq!(key, "key"); + assert!((delta - (-1.5)).abs() < f64::EPSILON); + } + other => panic!("expected IncrByFloat, got {other:?}"), + } + } + + #[test] + fn incrbyfloat_wrong_arity() { + let err = Command::from_frame(cmd(&["INCRBYFLOAT", "key"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn incrbyfloat_not_a_float() { + let err = Command::from_frame(cmd(&["INCRBYFLOAT", "key", "abc"])).unwrap_err(); + assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 16ac608e..5c5d4dc4 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -604,6 +604,21 @@ async fn execute( } } + Command::IncrByFloat { key, delta } => { + let req = ShardRequest::IncrByFloat { + key: key.clone(), + delta, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::BulkString(val)) => Frame::Bulk(Bytes::from(val)), + Ok(ShardResponse::WrongType) => wrongtype_error(), + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(ShardResponse::Err(msg)) => Frame::Error(msg), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + Command::Persist { key } => { let req = ShardRequest::Persist { key: key.clone() }; match engine.route(&key, req).await { From 23294e71871c4bad6d5282eef5fc05c113206a62 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 09:21:45 -0500 Subject: [PATCH 3/5] feat: add APPEND and STRLEN commands APPEND concatenates a value onto an existing string key (or creates it). STRLEN returns the byte length of a string value. both handle wrong-type errors and APPEND includes AOF persistence. --- crates/ember-core/src/keyspace.rs | 99 ++++++++++++++++++++++++ crates/ember-core/src/shard.rs | 71 ++++++++++++++++- crates/ember-persistence/src/aof.rs | 13 ++++ crates/ember-persistence/src/recovery.rs | 11 +++ crates/ember-protocol/src/command.rs | 60 ++++++++++++++ crates/ember-server/src/connection.rs | 24 ++++++ 6 files changed, 277 insertions(+), 1 deletion(-) diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index f27f7df3..9a04c501 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -608,6 +608,53 @@ impl Keyspace { } } + /// Appends a value to an existing string key, or creates a new key if + /// it doesn't exist. Returns the new string length. + pub fn append(&mut self, key: &str, value: &[u8]) -> Result { + self.remove_if_expired(key); + + match self.entries.get(key) { + Some(entry) => { + match &entry.value { + Value::String(existing) => { + let mut new_data = Vec::with_capacity(existing.len() + value.len()); + new_data.extend_from_slice(existing); + new_data.extend_from_slice(value); + let new_len = new_data.len(); + let expire = + time::remaining_ms(entry.expires_at_ms).map(Duration::from_millis); + match self.set(key.to_owned(), Bytes::from(new_data), expire) { + SetResult::Ok => Ok(new_len), + SetResult::OutOfMemory => Err(WriteError::OutOfMemory), + } + } + _ => Err(WriteError::WrongType), + } + } + None => { + let new_len = value.len(); + match self.set(key.to_owned(), Bytes::copy_from_slice(value), None) { + SetResult::Ok => Ok(new_len), + SetResult::OutOfMemory => Err(WriteError::OutOfMemory), + } + } + } + } + + /// Returns the length of the string value stored at key. + /// Returns 0 if the key does not exist. + pub fn strlen(&mut self, key: &str) -> Result { + self.remove_if_expired(key); + + match self.entries.get(key) { + Some(entry) => match &entry.value { + Value::String(data) => Ok(data.len()), + _ => Err(WrongType), + }, + None => Ok(0), + } + } + /// Returns aggregated stats for this keyspace. /// /// All fields are tracked incrementally — this is O(1). @@ -3574,6 +3621,58 @@ mod tests { assert_eq!(err, IncrFloatError::NotAFloat); } + #[test] + fn append_to_existing_key() { + let mut ks = Keyspace::new(); + ks.set("key".into(), Bytes::from("hello"), None); + let len = ks.append("key", b" world").unwrap(); + assert_eq!(len, 11); + assert_eq!( + ks.get("key").unwrap(), + Some(Value::String(Bytes::from("hello world"))) + ); + } + + #[test] + fn append_to_new_key() { + let mut ks = Keyspace::new(); + let len = ks.append("new", b"value").unwrap(); + assert_eq!(len, 5); + assert_eq!( + ks.get("new").unwrap(), + Some(Value::String(Bytes::from("value"))) + ); + } + + #[test] + fn append_wrong_type() { + let mut ks = Keyspace::new(); + ks.lpush("mylist", &[Bytes::from("a")]).unwrap(); + let err = ks.append("mylist", b"value").unwrap_err(); + assert_eq!(err, WriteError::WrongType); + } + + #[test] + fn strlen_existing_key() { + let mut ks = Keyspace::new(); + ks.set("key".into(), Bytes::from("hello"), None); + assert_eq!(ks.strlen("key").unwrap(), 5); + } + + #[test] + fn strlen_missing_key() { + let mut ks = Keyspace::new(); + assert_eq!(ks.strlen("missing").unwrap(), 0); + } + + #[test] + fn strlen_wrong_type() { + let mut ks = Keyspace::new(); + ks.lpush("mylist", &[Bytes::from("a")]).unwrap(); + let err = ks.strlen("mylist").unwrap_err(); + assert_eq!(err, WrongType); + } + #[test] fn format_float_integers() { assert_eq!(super::format_float(10.0), "10"); diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index 4e88e636..c94e0daa 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -75,6 +75,13 @@ pub enum ShardRequest { key: String, delta: f64, }, + Append { + key: String, + value: Bytes, + }, + Strlen { + key: String, + }, Del { key: String, }, @@ -551,6 +558,15 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { Err(IncrFloatError::OutOfMemory) => ShardResponse::OutOfMemory, Err(e) => ShardResponse::Err(e.to_string()), }, + ShardRequest::Append { key, value } => match ks.append(key, value) { + Ok(len) => ShardResponse::Len(len), + Err(WriteError::WrongType) => ShardResponse::WrongType, + Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory, + }, + ShardRequest::Strlen { key } => match ks.strlen(key) { + Ok(len) => ShardResponse::Len(len), + Err(_) => ShardResponse::WrongType, + }, ShardRequest::Del { key } => ShardResponse::Bool(ks.del(key)), ShardRequest::Exists { key } => ShardResponse::Bool(ks.exists(key)), ShardRequest::Expire { key, seconds } => ShardResponse::Bool(ks.expire(key, *seconds)), @@ -806,7 +822,14 @@ fn to_aof_record(req: &ShardRequest, resp: &ShardResponse) -> Option Some(AofRecord::Set { key: key.clone(), value: Bytes::from(val.clone()), - expire_ms: -1, // preserve existing TTL via the SET path + expire_ms: -1, + }) + } + // APPEND: record the appended value for replay + (ShardRequest::Append { key, value }, ShardResponse::Len(_)) => { + Some(AofRecord::Append { + key: key.clone(), + value: value.clone(), }) } (ShardRequest::Persist { key }, ShardResponse::Bool(true)) => { @@ -1360,6 +1383,52 @@ mod tests { } } + #[test] + fn dispatch_append() { + let mut ks = Keyspace::new(); + ks.set("k".into(), Bytes::from("hello"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::Append { + key: "k".into(), + value: Bytes::from(" world"), + }, + ); + assert!(matches!(resp, ShardResponse::Len(11))); + } + + #[test] + fn dispatch_strlen() { + let mut ks = Keyspace::new(); + ks.set("k".into(), Bytes::from("hello"), None); + let resp = dispatch(&mut ks, &ShardRequest::Strlen { key: "k".into() }); + assert!(matches!(resp, ShardResponse::Len(5))); + } + + #[test] + fn dispatch_strlen_missing() { + let mut ks = Keyspace::new(); + let resp = dispatch(&mut ks, &ShardRequest::Strlen { key: "nope".into() }); + assert!(matches!(resp, ShardResponse::Len(0))); + } + + #[test] + fn to_aof_record_for_append() { + let req = ShardRequest::Append { + key: "k".into(), + value: Bytes::from("data"), + }; + let resp = ShardResponse::Len(10); + let record = to_aof_record(&req, &resp).unwrap(); + match record { + AofRecord::Append { key, value } => { + assert_eq!(key, "k"); + assert_eq!(value, Bytes::from("data")); + } + other => panic!("expected Append, got {other:?}"), + } + } + #[test] fn dispatch_incrbyfloat_new_key() { let mut ks = Keyspace::new(); diff --git a/crates/ember-persistence/src/aof.rs b/crates/ember-persistence/src/aof.rs index 5f2549eb..e678ddd5 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -56,6 +56,7 @@ const TAG_SADD: u8 = 17; const TAG_SREM: u8 = 18; const TAG_INCRBY: u8 = 19; const TAG_DECRBY: u8 = 20; +const TAG_APPEND: u8 = 21; /// A single mutation record stored in the AOF. #[derive(Debug, Clone, PartialEq)] @@ -114,6 +115,8 @@ pub enum AofRecord { IncrBy { key: String, delta: i64 }, /// DECRBY key delta. DecrBy { key: String, delta: i64 }, + /// APPEND key value. + Append { key: String, value: Bytes }, } impl AofRecord { @@ -247,6 +250,11 @@ impl AofRecord { format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); format::write_i64(&mut buf, *delta).expect("vec write"); } + AofRecord::Append { key, value } => { + format::write_u8(&mut buf, TAG_APPEND).expect("vec write"); + format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); + format::write_bytes(&mut buf, value).expect("vec write"); + } } buf } @@ -387,6 +395,11 @@ impl AofRecord { let delta = format::read_i64(&mut cursor)?; Ok(AofRecord::DecrBy { key, delta }) } + TAG_APPEND => { + let key = read_string(&mut cursor, "key")?; + let value = Bytes::from(format::read_bytes(&mut cursor)?); + Ok(AofRecord::Append { key, value }) + } _ => Err(FormatError::UnknownTag(tag)), } } diff --git a/crates/ember-persistence/src/recovery.rs b/crates/ember-persistence/src/recovery.rs index c4116821..685f2ca0 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/crates/ember-persistence/src/recovery.rs @@ -291,6 +291,17 @@ fn replay_aof( AofRecord::DecrBy { key, delta } => { apply_incr(map, key, -delta); } + AofRecord::Append { key, value } => { + let entry = map + .entry(key) + .or_insert_with(|| (RecoveredValue::String(Bytes::new()), -1)); + if let RecoveredValue::String(ref mut data) = entry.0 { + let mut new_data = Vec::with_capacity(data.len() + value.len()); + new_data.extend_from_slice(data); + new_data.extend_from_slice(&value); + *data = Bytes::from(new_data); + } + } AofRecord::HSet { key, fields } => { let entry = map .entry(key) diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 725c3c92..535b61c3 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -56,6 +56,12 @@ pub enum Command { /// INCRBYFLOAT `key` `increment`. Increments the float value of a key by the given amount. IncrByFloat { key: String, delta: f64 }, + /// APPEND `key` `value`. Appends a value to a string key. Returns the new length. + Append { key: String, value: Bytes }, + + /// STRLEN `key`. Returns the length of the string value stored at key. + Strlen { key: String }, + /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, @@ -343,6 +349,8 @@ impl Command { Command::IncrBy { .. } => "incrby", Command::DecrBy { .. } => "decrby", Command::IncrByFloat { .. } => "incrbyfloat", + Command::Append { .. } => "append", + Command::Strlen { .. } => "strlen", Command::Del { .. } => "del", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", @@ -453,6 +461,8 @@ impl Command { "INCRBY" => parse_incrby(&frames[1..]), "DECRBY" => parse_decrby(&frames[1..]), "INCRBYFLOAT" => parse_incrbyfloat(&frames[1..]), + "APPEND" => parse_append(&frames[1..]), + "STRLEN" => parse_strlen(&frames[1..]), "DEL" => parse_del(&frames[1..]), "EXISTS" => parse_exists(&frames[1..]), "MGET" => parse_mget(&frames[1..]), @@ -696,6 +706,23 @@ fn parse_incrbyfloat(args: &[Frame]) -> Result { Ok(Command::IncrByFloat { key, delta }) } +fn parse_append(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("APPEND".into())); + } + let key = extract_string(&args[0])?; + let value = extract_bytes(&args[1])?; + Ok(Command::Append { key, value }) +} + +fn parse_strlen(args: &[Frame]) -> Result { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("STRLEN".into())); + } + let key = extract_string(&args[0])?; + Ok(Command::Strlen { key }) +} + fn parse_del(args: &[Frame]) -> Result { if args.is_empty() { return Err(ProtocolError::WrongArity("DEL".into())); @@ -3640,4 +3667,37 @@ mod tests { let err = Command::from_frame(cmd(&["INCRBYFLOAT", "key", "abc"])).unwrap_err(); assert!(matches!(err, ProtocolError::InvalidCommandFrame(_))); } + + // --- APPEND / STRLEN --- + + #[test] + fn append_basic() { + assert_eq!( + Command::from_frame(cmd(&["APPEND", "key", "value"])).unwrap(), + Command::Append { + key: "key".into(), + value: Bytes::from("value") + }, + ); + } + + #[test] + fn append_wrong_arity() { + let err = Command::from_frame(cmd(&["APPEND", "key"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn strlen_basic() { + assert_eq!( + Command::from_frame(cmd(&["STRLEN", "key"])).unwrap(), + Command::Strlen { key: "key".into() }, + ); + } + + #[test] + fn strlen_wrong_arity() { + let err = Command::from_frame(cmd(&["STRLEN"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 5c5d4dc4..a829ed0a 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -604,6 +604,30 @@ async fn execute( } } + Command::Append { key, value } => { + let req = ShardRequest::Append { + key: key.clone(), + value, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Len(n)) => Frame::Integer(n as i64), + Ok(ShardResponse::WrongType) => wrongtype_error(), + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + + Command::Strlen { key } => { + let req = ShardRequest::Strlen { key: key.clone() }; + match engine.route(&key, req).await { + Ok(ShardResponse::Len(n)) => Frame::Integer(n as i64), + Ok(ShardResponse::WrongType) => wrongtype_error(), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + Command::IncrByFloat { key, delta } => { let req = ShardRequest::IncrByFloat { key: key.clone(), From 14bade0c7c1354ea2b48b96dc6599096683df234 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 09:32:14 -0500 Subject: [PATCH 4/5] feat: add KEYS and RENAME commands implements KEYS with glob pattern matching and RENAME with atomic key move. includes full pipeline: protocol parsing, keyspace methods, shard dispatch, AOF persistence, recovery, and connection routing. also fixes clippy warnings in INCRBYFLOAT tests (3.14 -> 2.72 to avoid approx_constant lint). --- crates/ember-core/src/keyspace.rs | 218 ++++++++++++++++++++--- crates/ember-core/src/lib.rs | 4 +- crates/ember-core/src/shard.rs | 105 ++++++++++- crates/ember-persistence/src/aof.rs | 13 ++ crates/ember-persistence/src/recovery.rs | 5 + crates/ember-protocol/src/command.rs | 64 +++++++ crates/ember-server/src/connection.rs | 42 +++++ 7 files changed, 421 insertions(+), 30 deletions(-) diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index 9a04c501..56194ff5 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -113,6 +113,23 @@ impl std::fmt::Display for IncrFloatError { impl std::error::Error for IncrFloatError {} +/// Error returned when RENAME fails because the source key doesn't exist. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RenameError { + /// The source key does not exist. + NoSuchKey, +} + +impl std::fmt::Display for RenameError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RenameError::NoSuchKey => write!(f, "ERR no such key"), + } + } +} + +impl std::error::Error for RenameError {} + /// Result of a ZADD operation, containing both the client-facing count /// and the list of members that were actually applied (for AOF correctness). #[derive(Debug, Clone)] @@ -580,8 +597,7 @@ impl Keyspace { Some(entry) => { let val = match &entry.value { Value::String(data) => { - let s = std::str::from_utf8(data) - .map_err(|_| IncrFloatError::NotAFloat)?; + let s = std::str::from_utf8(data).map_err(|_| IncrFloatError::NotAFloat)?; s.parse::().map_err(|_| IncrFloatError::NotAFloat)? } _ => return Err(IncrFloatError::WrongType), @@ -614,23 +630,20 @@ impl Keyspace { self.remove_if_expired(key); match self.entries.get(key) { - Some(entry) => { - match &entry.value { - Value::String(existing) => { - let mut new_data = Vec::with_capacity(existing.len() + value.len()); - new_data.extend_from_slice(existing); - new_data.extend_from_slice(value); - let new_len = new_data.len(); - let expire = - time::remaining_ms(entry.expires_at_ms).map(Duration::from_millis); - match self.set(key.to_owned(), Bytes::from(new_data), expire) { - SetResult::Ok => Ok(new_len), - SetResult::OutOfMemory => Err(WriteError::OutOfMemory), - } + Some(entry) => match &entry.value { + Value::String(existing) => { + let mut new_data = Vec::with_capacity(existing.len() + value.len()); + new_data.extend_from_slice(existing); + new_data.extend_from_slice(value); + let new_len = new_data.len(); + let expire = time::remaining_ms(entry.expires_at_ms).map(Duration::from_millis); + match self.set(key.to_owned(), Bytes::from(new_data), expire) { + SetResult::Ok => Ok(new_len), + SetResult::OutOfMemory => Err(WriteError::OutOfMemory), } - _ => Err(WriteError::WrongType), } - } + _ => Err(WriteError::WrongType), + }, None => { let new_len = value.len(); match self.set(key.to_owned(), Bytes::copy_from_slice(value), None) { @@ -655,6 +668,53 @@ impl Keyspace { } } + /// Returns all keys matching a glob pattern. + /// + /// Warning: O(n) scan of the entire keyspace. Use SCAN for production + /// workloads with large key counts. + pub fn keys(&self, pattern: &str) -> Vec { + self.entries + .iter() + .filter(|(_, entry)| !entry.is_expired()) + .filter(|(key, _)| glob_match(pattern, key)) + .map(|(key, _)| key.clone()) + .collect() + } + + /// Renames a key to a new name. Returns an error if the source key + /// doesn't exist. If the destination key already exists, it is overwritten. + pub fn rename(&mut self, key: &str, newkey: &str) -> Result<(), RenameError> { + self.remove_if_expired(key); + self.remove_if_expired(newkey); + + let entry = match self.entries.remove(key) { + Some(entry) => entry, + None => return Err(RenameError::NoSuchKey), + }; + + // update memory tracking for old key removal + self.memory.remove(key, &entry.value); + if entry.expires_at_ms != 0 { + self.expiry_count = self.expiry_count.saturating_sub(1); + } + + // remove destination if it exists + if let Some(old_dest) = self.entries.remove(newkey) { + self.memory.remove(newkey, &old_dest.value); + if old_dest.expires_at_ms != 0 { + self.expiry_count = self.expiry_count.saturating_sub(1); + } + } + + // re-insert with the new key name, preserving value and expiry + self.memory.add(newkey, &entry.value); + if entry.expires_at_ms != 0 { + self.expiry_count += 1; + } + self.entries.insert(newkey.to_owned(), entry); + Ok(()) + } + /// Returns aggregated stats for this keyspace. /// /// All fields are tracked incrementally — this is O(1). @@ -3591,9 +3651,9 @@ mod tests { #[test] fn incr_by_float_new_key() { let mut ks = Keyspace::new(); - let result = ks.incr_by_float("new", 3.14).unwrap(); + let result = ks.incr_by_float("new", 2.72).unwrap(); let f: f64 = result.parse().unwrap(); - assert!((f - 3.14).abs() < 0.001); + assert!((f - 2.72).abs() < 0.001); } #[test] @@ -3682,7 +3742,125 @@ mod tests { #[test] fn format_float_decimals() { - assert_eq!(super::format_float(3.14), "3.14"); + assert_eq!(super::format_float(2.72), "2.72"); assert_eq!(super::format_float(10.5), "10.5"); } + + // --- keys tests --- + + #[test] + fn keys_match_all() { + let mut ks = Keyspace::new(); + ks.set("a".into(), Bytes::from("1"), None); + ks.set("b".into(), Bytes::from("2"), None); + ks.set("c".into(), Bytes::from("3"), None); + let mut result = ks.keys("*"); + result.sort(); + assert_eq!(result, vec!["a", "b", "c"]); + } + + #[test] + fn keys_with_pattern() { + let mut ks = Keyspace::new(); + ks.set("user:1".into(), Bytes::from("a"), None); + ks.set("user:2".into(), Bytes::from("b"), None); + ks.set("item:1".into(), Bytes::from("c"), None); + let mut result = ks.keys("user:*"); + result.sort(); + assert_eq!(result, vec!["user:1", "user:2"]); + } + + #[test] + fn keys_skips_expired() { + let mut ks = Keyspace::new(); + ks.set("live".into(), Bytes::from("a"), None); + ks.set( + "dead".into(), + Bytes::from("b"), + Some(Duration::from_millis(1)), + ); + thread::sleep(Duration::from_millis(5)); + let result = ks.keys("*"); + assert_eq!(result, vec!["live"]); + } + + #[test] + fn keys_empty_keyspace() { + let ks = Keyspace::new(); + assert!(ks.keys("*").is_empty()); + } + + // --- rename tests --- + + #[test] + fn rename_basic() { + let mut ks = Keyspace::new(); + ks.set("old".into(), Bytes::from("value"), None); + ks.rename("old", "new").unwrap(); + assert!(!ks.exists("old")); + assert_eq!( + ks.get("new").unwrap(), + Some(Value::String(Bytes::from("value"))) + ); + } + + #[test] + fn rename_preserves_expiry() { + let mut ks = Keyspace::new(); + ks.set( + "old".into(), + Bytes::from("val"), + Some(Duration::from_secs(60)), + ); + ks.rename("old", "new").unwrap(); + match ks.ttl("new") { + TtlResult::Seconds(s) => assert!((58..=60).contains(&s)), + other => panic!("expected TTL preserved, got {other:?}"), + } + } + + #[test] + fn rename_overwrites_destination() { + let mut ks = Keyspace::new(); + ks.set("src".into(), Bytes::from("new_val"), None); + ks.set("dst".into(), Bytes::from("old_val"), None); + ks.rename("src", "dst").unwrap(); + assert!(!ks.exists("src")); + assert_eq!( + ks.get("dst").unwrap(), + Some(Value::String(Bytes::from("new_val"))) + ); + assert_eq!(ks.len(), 1); + } + + #[test] + fn rename_missing_key_returns_error() { + let mut ks = Keyspace::new(); + let err = ks.rename("missing", "new").unwrap_err(); + assert_eq!(err, RenameError::NoSuchKey); + } + + #[test] + fn rename_same_key() { + let mut ks = Keyspace::new(); + ks.set("key".into(), Bytes::from("val"), None); + // renaming to itself should succeed (Redis behavior) + ks.rename("key", "key").unwrap(); + assert_eq!( + ks.get("key").unwrap(), + Some(Value::String(Bytes::from("val"))) + ); + } + + #[test] + fn rename_tracks_memory() { + let mut ks = Keyspace::new(); + ks.set("old".into(), Bytes::from("value"), None); + let before = ks.stats().used_bytes; + ks.rename("old", "new").unwrap(); + let after = ks.stats().used_bytes; + // same key length, so memory should be the same + assert_eq!(before, after); + assert_eq!(ks.stats().key_count, 1); + } } diff --git a/crates/ember-core/src/lib.rs b/crates/ember-core/src/lib.rs index dec71c4d..3a1b18d2 100644 --- a/crates/ember-core/src/lib.rs +++ b/crates/ember-core/src/lib.rs @@ -18,8 +18,8 @@ pub use concurrent::ConcurrentKeyspace; pub use engine::{Engine, EngineConfig}; pub use error::ShardError; pub use keyspace::{ - EvictionPolicy, IncrError, IncrFloatError, Keyspace, KeyspaceStats, ShardConfig, TtlResult, - WriteError, WrongType, ZAddResult, + EvictionPolicy, IncrError, IncrFloatError, Keyspace, KeyspaceStats, RenameError, ShardConfig, + TtlResult, WriteError, WrongType, ZAddResult, }; pub use shard::{ShardPersistenceConfig, ShardRequest, ShardResponse}; pub use types::Value; diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index c94e0daa..97d851ef 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -82,6 +82,15 @@ pub enum ShardRequest { Strlen { key: String, }, + /// Returns all keys matching a glob pattern in this shard. + Keys { + pattern: String, + }, + /// Renames a key within this shard. + Rename { + key: String, + newkey: String, + }, Del { key: String, }, @@ -567,6 +576,17 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { Ok(len) => ShardResponse::Len(len), Err(_) => ShardResponse::WrongType, }, + ShardRequest::Keys { pattern } => { + let keys = ks.keys(pattern); + ShardResponse::StringArray(keys) + } + ShardRequest::Rename { key, newkey } => { + use crate::keyspace::RenameError; + match ks.rename(key, newkey) { + Ok(()) => ShardResponse::Ok, + Err(RenameError::NoSuchKey) => ShardResponse::Err("ERR no such key".into()), + } + } ShardRequest::Del { key } => ShardResponse::Bool(ks.del(key)), ShardRequest::Exists { key } => ShardResponse::Bool(ks.exists(key)), ShardRequest::Expire { key, seconds } => ShardResponse::Bool(ks.expire(key, *seconds)), @@ -826,12 +846,14 @@ fn to_aof_record(req: &ShardRequest, resp: &ShardResponse) -> Option }) } // APPEND: record the appended value for replay - (ShardRequest::Append { key, value }, ShardResponse::Len(_)) => { - Some(AofRecord::Append { - key: key.clone(), - value: value.clone(), - }) - } + (ShardRequest::Append { key, value }, ShardResponse::Len(_)) => Some(AofRecord::Append { + key: key.clone(), + value: value.clone(), + }), + (ShardRequest::Rename { key, newkey }, ShardResponse::Ok) => Some(AofRecord::Rename { + key: key.clone(), + newkey: newkey.clone(), + }), (ShardRequest::Persist { key }, ShardResponse::Bool(true)) => { Some(AofRecord::Persist { key: key.clone() }) } @@ -1436,13 +1458,13 @@ mod tests { &mut ks, &ShardRequest::IncrByFloat { key: "new".into(), - delta: 3.14, + delta: 2.72, }, ); match resp { ShardResponse::BulkString(val) => { let f: f64 = val.parse().unwrap(); - assert!((f - 3.14).abs() < 0.001); + assert!((f - 2.72).abs() < 0.001); } other => panic!("expected BulkString, got {other:?}"), } @@ -1894,4 +1916,71 @@ mod tests { let resp = ShardResponse::Len(0); assert!(to_aof_record(&req, &resp).is_none()); } + + #[test] + fn dispatch_keys() { + let mut ks = Keyspace::new(); + ks.set("user:1".into(), Bytes::from("a"), None); + ks.set("user:2".into(), Bytes::from("b"), None); + ks.set("item:1".into(), Bytes::from("c"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::Keys { + pattern: "user:*".into(), + }, + ); + match resp { + ShardResponse::StringArray(mut keys) => { + keys.sort(); + assert_eq!(keys, vec!["user:1", "user:2"]); + } + other => panic!("expected StringArray, got {other:?}"), + } + } + + #[test] + fn dispatch_rename() { + let mut ks = Keyspace::new(); + ks.set("old".into(), Bytes::from("value"), None); + let resp = dispatch( + &mut ks, + &ShardRequest::Rename { + key: "old".into(), + newkey: "new".into(), + }, + ); + assert!(matches!(resp, ShardResponse::Ok)); + assert!(!ks.exists("old")); + assert!(ks.exists("new")); + } + + #[test] + fn dispatch_rename_missing_key() { + let mut ks = Keyspace::new(); + let resp = dispatch( + &mut ks, + &ShardRequest::Rename { + key: "missing".into(), + newkey: "new".into(), + }, + ); + assert!(matches!(resp, ShardResponse::Err(_))); + } + + #[test] + fn to_aof_record_for_rename() { + let req = ShardRequest::Rename { + key: "old".into(), + newkey: "new".into(), + }; + let resp = ShardResponse::Ok; + let record = to_aof_record(&req, &resp).unwrap(); + match record { + AofRecord::Rename { key, newkey } => { + assert_eq!(key, "old"); + assert_eq!(newkey, "new"); + } + other => panic!("expected Rename, got {other:?}"), + } + } } diff --git a/crates/ember-persistence/src/aof.rs b/crates/ember-persistence/src/aof.rs index e678ddd5..96380037 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -57,6 +57,7 @@ const TAG_SREM: u8 = 18; const TAG_INCRBY: u8 = 19; const TAG_DECRBY: u8 = 20; const TAG_APPEND: u8 = 21; +const TAG_RENAME: u8 = 22; /// A single mutation record stored in the AOF. #[derive(Debug, Clone, PartialEq)] @@ -117,6 +118,8 @@ pub enum AofRecord { DecrBy { key: String, delta: i64 }, /// APPEND key value. Append { key: String, value: Bytes }, + /// RENAME key newkey. + Rename { key: String, newkey: String }, } impl AofRecord { @@ -255,6 +258,11 @@ impl AofRecord { format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); format::write_bytes(&mut buf, value).expect("vec write"); } + AofRecord::Rename { key, newkey } => { + format::write_u8(&mut buf, TAG_RENAME).expect("vec write"); + format::write_bytes(&mut buf, key.as_bytes()).expect("vec write"); + format::write_bytes(&mut buf, newkey.as_bytes()).expect("vec write"); + } } buf } @@ -400,6 +408,11 @@ impl AofRecord { let value = Bytes::from(format::read_bytes(&mut cursor)?); Ok(AofRecord::Append { key, value }) } + TAG_RENAME => { + let key = read_string(&mut cursor, "key")?; + let newkey = read_string(&mut cursor, "newkey")?; + Ok(AofRecord::Rename { key, newkey }) + } _ => Err(FormatError::UnknownTag(tag)), } } diff --git a/crates/ember-persistence/src/recovery.rs b/crates/ember-persistence/src/recovery.rs index 685f2ca0..dba0e0a2 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/crates/ember-persistence/src/recovery.rs @@ -302,6 +302,11 @@ fn replay_aof( *data = Bytes::from(new_data); } } + AofRecord::Rename { key, newkey } => { + if let Some(entry) = map.remove(&key) { + map.insert(newkey, entry); + } + } AofRecord::HSet { key, fields } => { let entry = map .entry(key) diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 535b61c3..007f2be8 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -62,6 +62,12 @@ pub enum Command { /// STRLEN `key`. Returns the length of the string value stored at key. Strlen { key: String }, + /// KEYS `pattern`. Returns all keys matching a glob pattern. + Keys { pattern: String }, + + /// RENAME `key` `newkey`. Renames a key. + Rename { key: String, newkey: String }, + /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, @@ -351,6 +357,8 @@ impl Command { Command::IncrByFloat { .. } => "incrbyfloat", Command::Append { .. } => "append", Command::Strlen { .. } => "strlen", + Command::Keys { .. } => "keys", + Command::Rename { .. } => "rename", Command::Del { .. } => "del", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", @@ -463,6 +471,8 @@ impl Command { "INCRBYFLOAT" => parse_incrbyfloat(&frames[1..]), "APPEND" => parse_append(&frames[1..]), "STRLEN" => parse_strlen(&frames[1..]), + "KEYS" => parse_keys(&frames[1..]), + "RENAME" => parse_rename(&frames[1..]), "DEL" => parse_del(&frames[1..]), "EXISTS" => parse_exists(&frames[1..]), "MGET" => parse_mget(&frames[1..]), @@ -723,6 +733,23 @@ fn parse_strlen(args: &[Frame]) -> Result { Ok(Command::Strlen { key }) } +fn parse_keys(args: &[Frame]) -> Result { + if args.len() != 1 { + return Err(ProtocolError::WrongArity("KEYS".into())); + } + let pattern = extract_string(&args[0])?; + Ok(Command::Keys { pattern }) +} + +fn parse_rename(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("RENAME".into())); + } + let key = extract_string(&args[0])?; + let newkey = extract_string(&args[1])?; + Ok(Command::Rename { key, newkey }) +} + fn parse_del(args: &[Frame]) -> Result { if args.is_empty() { return Err(ProtocolError::WrongArity("DEL".into())); @@ -3700,4 +3727,41 @@ mod tests { let err = Command::from_frame(cmd(&["STRLEN"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } + + // --- KEYS --- + + #[test] + fn keys_basic() { + assert_eq!( + Command::from_frame(cmd(&["KEYS", "user:*"])).unwrap(), + Command::Keys { + pattern: "user:*".into() + }, + ); + } + + #[test] + fn keys_wrong_arity() { + let err = Command::from_frame(cmd(&["KEYS"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + // --- RENAME --- + + #[test] + fn rename_basic() { + assert_eq!( + Command::from_frame(cmd(&["RENAME", "old", "new"])).unwrap(), + Command::Rename { + key: "old".into(), + newkey: "new".into() + }, + ); + } + + #[test] + fn rename_wrong_arity() { + let err = Command::from_frame(cmd(&["RENAME", "only"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index a829ed0a..dd25a75c 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -773,6 +773,48 @@ async fn execute( Err(e) => Frame::Error(format!("ERR {e}")), }, + Command::Keys { pattern } => { + match engine + .broadcast(|| ShardRequest::Keys { + pattern: pattern.clone(), + }) + .await + { + Ok(responses) => { + let mut all_keys = Vec::new(); + for r in responses { + if let ShardResponse::StringArray(keys) = r { + all_keys.extend(keys); + } + } + Frame::Array( + all_keys + .into_iter() + .map(|k| Frame::Bulk(Bytes::from(k))) + .collect(), + ) + } + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + + Command::Rename { key, newkey } => { + // Route to the source key's shard. Both keys must hash to the + // same shard for a correct rename (same as Redis Cluster's + // cross-slot restriction). If they don't, the rename operates + // on the source shard and the newkey will live there. + let req = ShardRequest::Rename { + key: key.clone(), + newkey, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Ok) => Frame::Simple("OK".into()), + Ok(ShardResponse::Err(msg)) => Frame::Error(msg), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + Command::Scan { cursor, pattern, From b5e90d0186784482c7867a1f46dea0aee9eacd16 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 09:33:38 -0500 Subject: [PATCH 5/5] docs: update README with new string and key commands --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 7a277658..db157bc5 100644 --- a/README.md +++ b/README.md @@ -9,12 +9,12 @@ a low-latency, memory-efficient, distributed cache written in Rust. designed to ## features - **resp3 protocol** — full compatibility with `redis-cli` and existing Redis clients -- **string commands** — GET, SET (with NX/XX/EX/PX), MGET, MSET, INCR, DECR +- **string commands** — GET, SET (with NX/XX/EX/PX), MGET, MSET, INCR, DECR, INCRBY, DECRBY, INCRBYFLOAT, APPEND, STRLEN - **list operations** — LPUSH, RPUSH, LPOP, RPOP, LRANGE, LLEN - **sorted sets** — ZADD (with NX/XX/GT/LT/CH), ZREM, ZSCORE, ZRANK, ZRANGE, ZCARD - **hashes** — HSET, HGET, HGETALL, HDEL, HEXISTS, HLEN, HINCRBY, HKEYS, HVALS, HMGET - **sets** — SADD, SREM, SMEMBERS, SISMEMBER, SCARD -- **key commands** — DEL, EXISTS, EXPIRE, TTL, PEXPIRE, PTTL, PERSIST, TYPE, SCAN +- **key commands** — DEL, EXISTS, EXPIRE, TTL, PEXPIRE, PTTL, PERSIST, TYPE, SCAN, KEYS, RENAME - **server commands** — PING, ECHO, INFO, DBSIZE, FLUSHDB, BGSAVE, BGREWRITEAOF - **pub/sub** — SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBLISH, plus PUBSUB introspection - **observability** — prometheus metrics (`--metrics-port`), enriched INFO with 6 sections, SLOWLOG command @@ -184,7 +184,7 @@ contributions welcome — see [CONTRIBUTING.md](CONTRIBUTING.md). | 4 | clustering (raft, gossip, slots, migration) | ✅ complete | | 5 | developer experience (observability, CLI, clients) | 🚧 in progress | -**current**: 76 commands, 639 tests, ~21k lines of code +**current**: 83 commands, 695 tests, ~22k lines of code ## security