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 diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index 94e8b832..56194ff5 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -77,6 +77,59 @@ 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 {} + +/// 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)] @@ -500,11 +553,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 @@ -532,6 +585,136 @@ 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), + } + } + + /// 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 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). @@ -1581,6 +1764,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 +3638,229 @@ 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", 2.72).unwrap(); + let f: f64 = result.parse().unwrap(); + assert!((f - 2.72).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 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"); + 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(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 10c2513d..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, 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 4434c0f6..97d851ef 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; @@ -62,6 +63,34 @@ pub enum ShardRequest { Decr { key: String, }, + IncrBy { + key: String, + delta: i64, + }, + DecrBy { + key: String, + delta: i64, + }, + IncrByFloat { + key: String, + delta: f64, + }, + Append { + key: String, + value: Bytes, + }, + 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, }, @@ -251,6 +280,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. @@ -518,6 +549,44 @@ 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::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::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::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)), @@ -755,6 +824,36 @@ 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, + }) + } + // 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, + }) + } + // APPEND: record the appended value for replay + (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() }) } @@ -1245,6 +1344,132 @@ 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 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_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(); + let resp = dispatch( + &mut ks, + &ShardRequest::IncrByFloat { + key: "new".into(), + delta: 2.72, + }, + ); + match resp { + ShardResponse::BulkString(val) => { + let f: f64 = val.parse().unwrap(); + assert!((f - 2.72).abs() < 0.001); + } + other => panic!("expected BulkString, got {other:?}"), + } + } + #[test] fn to_aof_record_for_incr() { let req = ShardRequest::Incr { key: "c".into() }; @@ -1261,6 +1486,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(); @@ -1657,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 4fd0d4b2..96380037 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -54,6 +54,10 @@ 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; +const TAG_APPEND: u8 = 21; +const TAG_RENAME: u8 = 22; /// A single mutation record stored in the AOF. #[derive(Debug, Clone, PartialEq)] @@ -108,6 +112,14 @@ 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 }, + /// APPEND key value. + Append { key: String, value: Bytes }, + /// RENAME key newkey. + Rename { key: String, newkey: String }, } impl AofRecord { @@ -231,6 +243,26 @@ 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"); + } + 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"); + } + 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 } @@ -361,6 +393,26 @@ 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 }) + } + TAG_APPEND => { + let key = read_string(&mut cursor, "key")?; + 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 eaa86656..dba0e0a2 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/crates/ember-persistence/src/recovery.rs @@ -285,6 +285,28 @@ 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::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::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 7247e4ec..007f2be8 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -47,6 +47,27 @@ 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 }, + + /// 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 }, + + /// 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 }, @@ -331,6 +352,13 @@ impl Command { Command::Set { .. } => "set", Command::Incr { .. } => "incr", Command::Decr { .. } => "decr", + Command::IncrBy { .. } => "incrby", + Command::DecrBy { .. } => "decrby", + Command::IncrByFloat { .. } => "incrbyfloat", + Command::Append { .. } => "append", + Command::Strlen { .. } => "strlen", + Command::Keys { .. } => "keys", + Command::Rename { .. } => "rename", Command::Del { .. } => "del", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", @@ -438,6 +466,13 @@ 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..]), + "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..]), @@ -646,6 +681,75 @@ 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_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_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_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())); @@ -3499,4 +3603,165 @@ 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(_))); + } + + // --- 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(_))); + } + + // --- 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(_))); + } + + // --- 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 fd1e1e17..dd25a75c 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,76 @@ 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}")), + } + } + + 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(), + 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}")), @@ -708,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,