From 036557f47e019405819f29cf94c3de9d184a730c Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 13:51:14 -0500 Subject: [PATCH 1/5] feat: add background value dropper for lazy free adds a dedicated OS thread that receives large values and drops them off the hot path. this prevents expensive destructors (large lists, hashes, sorted sets) from blocking shard threads. - dropper.rs: bounded channel (4096 capacity) with try_send fallback - memory.rs: is_large_value() helper with 64-element threshold - strings always drop inline (Bytes is O(1) ref-counted) --- crates/ember-core/src/dropper.rs | 144 +++++++++++++++++++++++++++++++ crates/ember-core/src/lib.rs | 1 + crates/ember-core/src/memory.rs | 64 ++++++++++++++ 3 files changed, 209 insertions(+) create mode 100644 crates/ember-core/src/dropper.rs diff --git a/crates/ember-core/src/dropper.rs b/crates/ember-core/src/dropper.rs new file mode 100644 index 00000000..50112d8f --- /dev/null +++ b/crates/ember-core/src/dropper.rs @@ -0,0 +1,144 @@ +//! Background value dropper for lazy free. +//! +//! Expensive destructor work (dropping large lists, hashes, sorted sets) +//! is offloaded to a dedicated OS thread so shard loops stay responsive. +//! This is the same strategy Redis uses with its `lazyfree` threads. +//! +//! The dropper runs as a plain `std::thread` rather than a tokio task +//! because dropping data structures is CPU-bound work that would starve +//! the async executor. + +use std::collections::HashMap; +use std::sync::mpsc::{self, SyncSender, TrySendError}; + +use crate::keyspace::Entry; +use crate::memory::is_large_value; +use crate::types::Value; + +/// Bounded channel capacity. Large enough to absorb bursts without +/// meaningful memory overhead (~4096 pointers). +const DROP_CHANNEL_CAPACITY: usize = 4096; + +/// Items that can be sent to the background drop thread. +/// +/// The fields are never explicitly read — the whole point is that the +/// drop thread receives them and lets their destructors run. +#[allow(dead_code)] +enum Droppable { + /// A single value removed from the keyspace (e.g. DEL, UNLINK, eviction). + Value(Value), + /// All entries from a FLUSHDB ASYNC — dropped in bulk. + Entries(HashMap), +} + +/// A cloneable handle for deferring expensive drops to the background thread. +/// +/// When all handles are dropped, the background thread's channel closes +/// and it exits cleanly. +#[derive(Debug, Clone)] +pub struct DropHandle { + tx: SyncSender, +} + +impl DropHandle { + /// Spawns the background drop thread and returns a handle. + pub fn spawn() -> Self { + let (tx, rx) = mpsc::sync_channel::(DROP_CHANNEL_CAPACITY); + + std::thread::Builder::new() + .name("ember-drop".into()) + .spawn(move || { + // just drain the channel — dropping each item frees the memory + while rx.recv().is_ok() {} + }) + .expect("failed to spawn drop thread"); + + Self { tx } + } + + /// Defers dropping a value to the background thread if it's large enough + /// to be worth the channel overhead. Small values are dropped inline. + /// + /// If the channel is full, falls back to inline drop — never blocks. + pub fn defer_value(&self, value: Value) { + if !is_large_value(&value) { + return; // small value — inline drop is fine + } + // try_send: never block the shard even if the drop thread is behind + match self.tx.try_send(Droppable::Value(value)) { + Ok(()) => {} + Err(TrySendError::Full(item)) => { + // channel full — drop inline as fallback + drop(item); + } + Err(TrySendError::Disconnected(_)) => { + // drop thread gone — nothing we can do, value drops here + } + } + } + + /// Defers dropping all entries from a flush operation. Always deferred + /// since a full keyspace is always worth offloading. + pub(crate) fn defer_entries(&self, entries: HashMap) { + if entries.is_empty() { + return; + } + match self.tx.try_send(Droppable::Entries(entries)) { + Ok(()) => {} + Err(TrySendError::Full(item)) => { + drop(item); + } + Err(TrySendError::Disconnected(_)) => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use std::collections::VecDeque; + + #[test] + fn defer_small_value_drops_inline() { + let handle = DropHandle::spawn(); + // small string — should not be sent to channel + handle.defer_value(Value::String(Bytes::from("hello"))); + } + + #[test] + fn defer_large_list() { + let handle = DropHandle::spawn(); + let mut list = VecDeque::new(); + for i in 0..100 { + list.push_back(Bytes::from(format!("item-{i}"))); + } + handle.defer_value(Value::List(list)); + // give the drop thread a moment to process + std::thread::sleep(std::time::Duration::from_millis(10)); + } + + #[test] + fn defer_entries_from_flush() { + let handle = DropHandle::spawn(); + let mut entries = HashMap::new(); + for i in 0..10 { + entries.insert( + format!("key-{i}"), + Entry { + value: Value::String(Bytes::from(format!("val-{i}"))), + expires_at_ms: 0, + last_access_ms: 0, + }, + ); + } + handle.defer_entries(entries); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + + #[test] + fn empty_entries_skipped() { + let handle = DropHandle::spawn(); + handle.defer_entries(HashMap::new()); + } +} diff --git a/crates/ember-core/src/lib.rs b/crates/ember-core/src/lib.rs index 3a1b18d2..a39fe166 100644 --- a/crates/ember-core/src/lib.rs +++ b/crates/ember-core/src/lib.rs @@ -5,6 +5,7 @@ //! where each shard independently manages a partition of keys. pub mod concurrent; +pub mod dropper; pub mod engine; pub mod error; pub mod expiry; diff --git a/crates/ember-core/src/memory.rs b/crates/ember-core/src/memory.rs index 60410585..a6082cd3 100644 --- a/crates/ember-core/src/memory.rs +++ b/crates/ember-core/src/memory.rs @@ -141,6 +141,28 @@ impl Default for MemoryTracker { } } +/// Element count threshold below which values are dropped inline rather +/// than sent to the background drop thread. Strings are always inline +/// (Bytes::drop is O(1)), but collections with more than this many +/// elements get deferred. +pub const LAZY_FREE_THRESHOLD: usize = 64; + +/// Returns `true` if dropping this value is expensive enough to justify +/// sending it to the background drop thread. +/// +/// Strings are always cheap to drop (reference-counted `Bytes`). +/// Collections are considered large when they exceed [`LAZY_FREE_THRESHOLD`] +/// elements. +pub fn is_large_value(value: &Value) -> bool { + match value { + Value::String(_) => false, + Value::List(d) => d.len() > LAZY_FREE_THRESHOLD, + Value::SortedSet(ss) => ss.len() > LAZY_FREE_THRESHOLD, + Value::Hash(m) => m.len() > LAZY_FREE_THRESHOLD, + Value::Set(s) => s.len() > LAZY_FREE_THRESHOLD, + } +} + /// Estimates the total memory footprint of a single entry. /// /// key heap allocation + value bytes + fixed per-entry overhead. @@ -325,4 +347,46 @@ mod tests { fn effective_limit_zero() { assert_eq!(effective_limit(0), 0); } + + #[test] + fn string_is_never_large() { + let val = Value::String(Bytes::from(vec![0u8; 10_000])); + assert!(!is_large_value(&val)); + } + + #[test] + fn small_list_is_not_large() { + let mut d = std::collections::VecDeque::new(); + for _ in 0..LAZY_FREE_THRESHOLD { + d.push_back(Bytes::from("x")); + } + assert!(!is_large_value(&Value::List(d))); + } + + #[test] + fn big_list_is_large() { + let mut d = std::collections::VecDeque::new(); + for _ in 0..=LAZY_FREE_THRESHOLD { + d.push_back(Bytes::from("x")); + } + assert!(is_large_value(&Value::List(d))); + } + + #[test] + fn big_hash_is_large() { + let mut m = std::collections::HashMap::new(); + for i in 0..=LAZY_FREE_THRESHOLD { + m.insert(format!("f{i}"), Bytes::from("v")); + } + assert!(is_large_value(&Value::Hash(m))); + } + + #[test] + fn big_set_is_large() { + let mut s = std::collections::HashSet::new(); + for i in 0..=LAZY_FREE_THRESHOLD { + s.insert(format!("m{i}")); + } + assert!(is_large_value(&Value::Set(s))); + } } From 6becf256f648e11d7ffb14a9f9c7adbe75ce6078 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 13:52:03 -0500 Subject: [PATCH 2/5] feat: add unlink and flush_async to keyspace with lazy free integrates the background dropper into keyspace operations: - unlink(): like del but always defers the destructor - flush_async(): swaps entries map and returns old entries for deferred drop - del/try_evict/remove_if_expired: defer large value drops when handle is set - keys(): warn when scanning >10k keys (suggest SCAN instead) --- crates/ember-core/src/keyspace.rs | 69 +++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index cda853e1..60e03541 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -11,6 +11,9 @@ use std::time::Duration; use bytes::Bytes; use rand::seq::IteratorRandom; +use tracing::warn; + +use crate::dropper::DropHandle; use crate::memory::{self, MemoryTracker}; use crate::time; use crate::types::sorted_set::{SortedSet, ZAddFlags}; @@ -269,6 +272,9 @@ pub struct Keyspace { expired_total: u64, /// Cumulative count of keys removed by eviction. evicted_total: u64, + /// When set, large values are dropped on a background thread instead + /// of inline on the shard thread. See [`crate::dropper`]. + drop_handle: Option, } impl Keyspace { @@ -286,9 +292,17 @@ impl Keyspace { expiry_count: 0, expired_total: 0, evicted_total: 0, + drop_handle: None, } } + /// Attaches a background drop handle for lazy free. When set, large + /// values removed by del/eviction/expiration are dropped on a + /// background thread instead of blocking the shard. + pub fn set_drop_handle(&mut self, handle: DropHandle) { + self.drop_handle = Some(handle); + } + /// Retrieves the string value for `key`, or `None` if missing/expired. /// /// Returns `Err(WrongType)` if the key holds a non-string value. @@ -394,6 +408,7 @@ impl Keyspace { self.expiry_count = self.expiry_count.saturating_sub(1); } self.evicted_total += 1; + self.defer_drop(entry.value); return true; } } @@ -426,6 +441,9 @@ impl Keyspace { } /// Removes a key. Returns `true` if the key existed (and wasn't expired). + /// + /// When a drop handle is set, large values are dropped on the + /// background thread instead of inline. pub fn del(&mut self, key: &str) -> bool { if self.remove_if_expired(key) { return false; @@ -435,12 +453,47 @@ impl Keyspace { if entry.expires_at_ms != 0 { self.expiry_count = self.expiry_count.saturating_sub(1); } + self.defer_drop(entry.value); + true + } else { + false + } + } + + /// Removes a key like `del`, but always defers the value's destructor + /// to the background drop thread (when available). Semantically + /// identical to DEL — the key is gone immediately, memory is + /// accounted for immediately, but the actual deallocation happens + /// off the hot path. + pub fn unlink(&mut self, key: &str) -> bool { + if self.remove_if_expired(key) { + return false; + } + if let Some(entry) = self.entries.remove(key) { + self.memory.remove(key, &entry.value); + if entry.expires_at_ms != 0 { + self.expiry_count = self.expiry_count.saturating_sub(1); + } + // always defer for UNLINK, regardless of value size + if let Some(ref handle) = self.drop_handle { + handle.defer_value(entry.value); + } true } else { false } } + /// Replaces the entries map with an empty one and resets memory + /// tracking. Returns the old entries so the caller can send them + /// to the background drop thread. + pub(crate) fn flush_async(&mut self) -> HashMap { + let old = std::mem::take(&mut self.entries); + self.memory.reset(); + self.expiry_count = 0; + old + } + /// Returns `true` if the key exists and hasn't expired. pub fn exists(&mut self, key: &str) -> bool { if self.remove_if_expired(key) { @@ -679,6 +732,13 @@ impl Keyspace { /// Warning: O(n) scan of the entire keyspace. Use SCAN for production /// workloads with large key counts. pub fn keys(&self, pattern: &str) -> Vec { + let len = self.entries.len(); + if len > 10_000 { + warn!( + key_count = len, + "KEYS on large keyspace, consider SCAN instead" + ); + } self.entries .iter() .filter(|(_, entry)| !entry.is_expired()) @@ -1752,10 +1812,19 @@ impl Keyspace { self.expiry_count = self.expiry_count.saturating_sub(1); } self.expired_total += 1; + self.defer_drop(entry.value); } } expired } + + /// Sends a value to the background drop thread if one is configured + /// and the value is large enough to justify the overhead. + fn defer_drop(&self, value: Value) { + if let Some(ref handle) = self.drop_handle { + handle.defer_value(value); + } + } } impl Default for Keyspace { From 749915931337d31c1b38953dd0fed98a7216fb6e Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 13:53:17 -0500 Subject: [PATCH 3/5] feat: wire up UNLINK and FLUSHDB ASYNC through shard and engine - adds Unlink and FlushDbAsync shard request variants - spawn_shard now accepts an optional DropHandle for lazy free - engine creates a shared DropHandle and passes it to all shards - FlushDbAsync handled in the main loop: swaps entries, defers drop - Unlink maps to AofRecord::Del for replay (same semantics) --- crates/ember-core/src/engine.rs | 13 +++++++- crates/ember-core/src/shard.rs | 53 +++++++++++++++++++++++++-------- 2 files changed, 53 insertions(+), 13 deletions(-) diff --git a/crates/ember-core/src/engine.rs b/crates/ember-core/src/engine.rs index a00aedcf..b8499514 100644 --- a/crates/ember-core/src/engine.rs +++ b/crates/ember-core/src/engine.rs @@ -7,6 +7,7 @@ use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; +use crate::dropper::DropHandle; use crate::error::ShardError; use crate::keyspace::ShardConfig; use crate::shard::{self, ShardHandle, ShardPersistenceConfig, ShardRequest, ShardResponse}; @@ -46,15 +47,25 @@ impl Engine { /// Creates an engine with `shard_count` shards and the given config. /// + /// Spawns a single background drop thread shared by all shards for + /// lazy-freeing large values. + /// /// Panics if `shard_count` is zero. pub fn with_config(shard_count: usize, config: EngineConfig) -> Self { assert!(shard_count > 0, "shard count must be at least 1"); + let drop_handle = DropHandle::spawn(); + let shards = (0..shard_count) .map(|i| { let mut shard_config = config.shard.clone(); shard_config.shard_id = i as u16; - shard::spawn_shard(SHARD_BUFFER, shard_config, config.persistence.clone()) + shard::spawn_shard( + SHARD_BUFFER, + shard_config, + config.persistence.clone(), + Some(drop_handle.clone()), + ) }) .collect(); diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index 97d851ef..05d05531 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -15,6 +15,7 @@ use ember_persistence::snapshot::{self, SnapEntry, SnapValue, SnapshotWriter}; use tokio::sync::{mpsc, oneshot}; use tracing::{info, warn}; +use crate::dropper::DropHandle; use crate::error::ShardError; use crate::expiry; use crate::keyspace::{ @@ -94,6 +95,10 @@ pub enum ShardRequest { Del { key: String, }, + /// Like DEL but defers value deallocation to the background drop thread. + Unlink { + key: String, + }, Exists { key: String, }, @@ -234,6 +239,8 @@ pub enum ShardRequest { RewriteAof, /// Clears all keys from the keyspace. FlushDb, + /// Clears all keys, deferring deallocation to the background drop thread. + FlushDbAsync, /// Scans keys in the keyspace. Scan { cursor: u64, @@ -346,14 +353,16 @@ impl ShardHandle { /// Spawns a shard task and returns the handle for communicating with it. /// /// `buffer` controls the mpsc channel capacity — higher values absorb -/// burst traffic at the cost of memory. +/// burst traffic at the cost of memory. When `drop_handle` is provided, +/// large value deallocations are deferred to the background drop thread. pub fn spawn_shard( buffer: usize, config: ShardConfig, persistence: Option, + drop_handle: Option, ) -> ShardHandle { let (tx, rx) = mpsc::channel(buffer); - tokio::spawn(run_shard(rx, config, persistence)); + tokio::spawn(run_shard(rx, config, persistence, drop_handle)); ShardHandle { tx } } @@ -363,10 +372,15 @@ async fn run_shard( mut rx: mpsc::Receiver, config: ShardConfig, persistence: Option, + drop_handle: Option, ) { let shard_id = config.shard_id; let mut keyspace = Keyspace::with_config(config); + if let Some(handle) = drop_handle.clone() { + keyspace.set_drop_handle(handle); + } + // -- recovery -- if let Some(ref pcfg) = persistence { let result = recovery::recover_shard(&pcfg.data_dir, shard_id); @@ -467,6 +481,15 @@ async fn run_shard( let _ = msg.reply.send(resp); continue; } + RequestKind::FlushDbAsync => { + let old_entries = keyspace.flush_async(); + if let Some(ref handle) = drop_handle { + handle.defer_entries(old_entries); + } + // else: old_entries drops inline here + let _ = msg.reply.send(ShardResponse::Ok); + continue; + } RequestKind::Other => {} } @@ -494,11 +517,12 @@ async fn run_shard( } } -/// Lightweight tag so we can identify snapshot/rewrite requests after -/// dispatch without borrowing the request again. +/// Lightweight tag so we can identify requests that need special +/// handling after dispatch without borrowing the request again. enum RequestKind { Snapshot, RewriteAof, + FlushDbAsync, Other, } @@ -506,6 +530,7 @@ fn describe_request(req: &ShardRequest) -> RequestKind { match req { ShardRequest::Snapshot => RequestKind::Snapshot, ShardRequest::RewriteAof => RequestKind::RewriteAof, + ShardRequest::FlushDbAsync => RequestKind::FlushDbAsync, _ => RequestKind::Other, } } @@ -588,6 +613,7 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { } } ShardRequest::Del { key } => ShardResponse::Bool(ks.del(key)), + ShardRequest::Unlink { key } => ShardResponse::Bool(ks.unlink(key)), ShardRequest::Exists { key } => ShardResponse::Bool(ks.exists(key)), ShardRequest::Expire { key, seconds } => ShardResponse::Bool(ks.expire(key, *seconds)), ShardRequest::Ttl { key } => ShardResponse::Ttl(ks.ttl(key)), @@ -757,8 +783,10 @@ fn dispatch(ks: &mut Keyspace, req: &ShardRequest) -> ShardResponse { Ok(count) => ShardResponse::Len(count), Err(_) => ShardResponse::WrongType, }, - // snapshot/rewrite are handled in the main loop, not here - ShardRequest::Snapshot | ShardRequest::RewriteAof => ShardResponse::Ok, + // snapshot/rewrite/flush_async are handled in the main loop, not here + ShardRequest::Snapshot | ShardRequest::RewriteAof | ShardRequest::FlushDbAsync => { + ShardResponse::Ok + } } } @@ -779,7 +807,8 @@ fn to_aof_record(req: &ShardRequest, resp: &ShardResponse) -> Option expire_ms, }) } - (ShardRequest::Del { key }, ShardResponse::Bool(true)) => { + (ShardRequest::Del { key }, ShardResponse::Bool(true)) + | (ShardRequest::Unlink { key }, ShardResponse::Bool(true)) => { Some(AofRecord::Del { key: key.clone() }) } (ShardRequest::Expire { key, seconds }, ShardResponse::Bool(true)) => { @@ -1080,7 +1109,7 @@ mod tests { #[tokio::test] async fn shard_round_trip() { - let handle = spawn_shard(16, ShardConfig::default(), None); + let handle = spawn_shard(16, ShardConfig::default(), None, None); let resp = handle .send(ShardRequest::Set { @@ -1110,7 +1139,7 @@ mod tests { #[tokio::test] async fn expired_key_through_shard() { - let handle = spawn_shard(16, ShardConfig::default(), None); + let handle = spawn_shard(16, ShardConfig::default(), None, None); handle .send(ShardRequest::Set { @@ -1134,7 +1163,7 @@ mod tests { #[tokio::test] async fn active_expiration_cleans_up_without_access() { - let handle = spawn_shard(16, ShardConfig::default(), None); + let handle = spawn_shard(16, ShardConfig::default(), None, None); // set a key with a short TTL handle @@ -1198,7 +1227,7 @@ mod tests { // write some keys then trigger a snapshot { - let handle = spawn_shard(16, config.clone(), Some(pcfg.clone())); + let handle = spawn_shard(16, config.clone(), Some(pcfg.clone()), None); handle .send(ShardRequest::Set { key: "a".into(), @@ -1239,7 +1268,7 @@ mod tests { // start a new shard with the same config — should recover { - let handle = spawn_shard(16, config, Some(pcfg)); + let handle = spawn_shard(16, config, Some(pcfg), None); // give it a moment to recover tokio::time::sleep(Duration::from_millis(50)).await; From 161cedd8e46d0c886df30ba23b936dd2cfed79d0 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 13:53:54 -0500 Subject: [PATCH 4/5] feat: add UNLINK command parsing and FLUSHDB ASYNC support - Command::Unlink: parses like DEL, accepts one or more keys - Command::FlushDb: now has async_mode field, accepts optional ASYNC arg - parse_unlink mirrors parse_del exactly - parse_flushdb updated to accept optional ASYNC argument --- crates/ember-protocol/src/command.rs | 82 +++++++++++++++++++++++++--- 1 file changed, 74 insertions(+), 8 deletions(-) diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index dbf3a887..87f4100e 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -71,6 +71,9 @@ pub enum Command { /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, + /// UNLINK `key` \[key ...\]. Like DEL but frees memory in the background. + Unlink { keys: Vec }, + /// EXISTS `key` \[key ...\]. Returns the number of keys that exist. Exists { keys: Vec }, @@ -107,8 +110,8 @@ pub enum Command { /// BGREWRITEAOF. Triggers an AOF rewrite (snapshot + truncate). BgRewriteAof, - /// FLUSHDB. Removes all keys from the database. - FlushDb, + /// FLUSHDB \[ASYNC\]. Removes all keys from the database. + FlushDb { async_mode: bool }, /// SCAN `cursor` \[MATCH pattern\] \[COUNT count\]. Iterates keys. Scan { @@ -371,6 +374,7 @@ impl Command { Command::Keys { .. } => "keys", Command::Rename { .. } => "rename", Command::Del { .. } => "del", + Command::Unlink { .. } => "unlink", Command::Exists { .. } => "exists", Command::MGet { .. } => "mget", Command::MSet { .. } => "mset", @@ -383,7 +387,7 @@ impl Command { Command::Info { .. } => "info", Command::BgSave => "bgsave", Command::BgRewriteAof => "bgrewriteaof", - Command::FlushDb => "flushdb", + Command::FlushDb { .. } => "flushdb", Command::Scan { .. } => "scan", Command::LPush { .. } => "lpush", Command::RPush { .. } => "rpush", @@ -487,6 +491,7 @@ impl Command { "KEYS" => parse_keys(&frames[1..]), "RENAME" => parse_rename(&frames[1..]), "DEL" => parse_del(&frames[1..]), + "UNLINK" => parse_unlink(&frames[1..]), "EXISTS" => parse_exists(&frames[1..]), "MGET" => parse_mget(&frames[1..]), "MSET" => parse_mset(&frames[1..]), @@ -902,10 +907,27 @@ fn parse_bgrewriteaof(args: &[Frame]) -> Result { } fn parse_flushdb(args: &[Frame]) -> Result { - if !args.is_empty() { - return Err(ProtocolError::WrongArity("FLUSHDB".into())); + if args.is_empty() { + return Ok(Command::FlushDb { async_mode: false }); } - Ok(Command::FlushDb) + if args.len() == 1 { + let arg = extract_string(&args[0])?; + if arg.eq_ignore_ascii_case("ASYNC") { + return Ok(Command::FlushDb { async_mode: true }); + } + } + Err(ProtocolError::WrongArity("FLUSHDB".into())) +} + +fn parse_unlink(args: &[Frame]) -> Result { + if args.is_empty() { + return Err(ProtocolError::WrongArity("UNLINK".into())); + } + let keys = args + .iter() + .map(extract_string) + .collect::, _>>()?; + Ok(Command::Unlink { keys }) } fn parse_scan(args: &[Frame]) -> Result { @@ -2236,7 +2258,7 @@ mod tests { fn flushdb_basic() { assert_eq!( Command::from_frame(cmd(&["FLUSHDB"])).unwrap(), - Command::FlushDb, + Command::FlushDb { async_mode: false }, ); } @@ -2244,7 +2266,23 @@ mod tests { fn flushdb_case_insensitive() { assert_eq!( Command::from_frame(cmd(&["flushdb"])).unwrap(), - Command::FlushDb, + Command::FlushDb { async_mode: false }, + ); + } + + #[test] + fn flushdb_async() { + assert_eq!( + Command::from_frame(cmd(&["FLUSHDB", "ASYNC"])).unwrap(), + Command::FlushDb { async_mode: true }, + ); + } + + #[test] + fn flushdb_async_case_insensitive() { + assert_eq!( + Command::from_frame(cmd(&["flushdb", "async"])).unwrap(), + Command::FlushDb { async_mode: true }, ); } @@ -2254,6 +2292,34 @@ mod tests { assert!(matches!(err, ProtocolError::WrongArity(_))); } + // --- unlink --- + + #[test] + fn unlink_single() { + assert_eq!( + Command::from_frame(cmd(&["UNLINK", "mykey"])).unwrap(), + Command::Unlink { + keys: vec!["mykey".into()] + }, + ); + } + + #[test] + fn unlink_multiple() { + assert_eq!( + Command::from_frame(cmd(&["UNLINK", "a", "b", "c"])).unwrap(), + Command::Unlink { + keys: vec!["a".into(), "b".into(), "c".into()] + }, + ); + } + + #[test] + fn unlink_no_args() { + let err = Command::from_frame(cmd(&["UNLINK"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + // --- lpush --- #[test] From e8f888de1d889a67c8e5ac4c383b7e24cab41272 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 8 Feb 2026 13:54:32 -0500 Subject: [PATCH 5/5] feat: handle UNLINK and FLUSHDB ASYNC in connection handlers - sharded mode: UNLINK routes through multi_key_bool like DEL - sharded mode: FLUSHDB broadcasts FlushDb or FlushDbAsync based on flag - concurrent mode: UNLINK handled same as DEL (DashMap has no blocking issue) - concurrent mode: FlushDb pattern updated for async_mode field --- crates/ember-server/src/concurrent_handler.rs | 4 ++-- crates/ember-server/src/connection.rs | 19 +++++++++++++++---- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/crates/ember-server/src/concurrent_handler.rs b/crates/ember-server/src/concurrent_handler.rs index fb3d68a7..f3815178 100644 --- a/crates/ember-server/src/concurrent_handler.rs +++ b/crates/ember-server/src/concurrent_handler.rs @@ -186,7 +186,7 @@ async fn execute_concurrent( } } - Command::Del { keys } => { + Command::Del { keys } | Command::Unlink { keys } => { let mut count = 0i64; for key in keys { if keyspace.del(&key) { @@ -224,7 +224,7 @@ async fn execute_concurrent( Command::DbSize => Frame::Integer(keyspace.len() as i64), - Command::FlushDb => { + Command::FlushDb { .. } => { keyspace.clear(); Frame::Simple("OK".into()) } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 87751970..ed0a7415 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -712,6 +712,10 @@ async fn execute( multi_key_bool(engine, &keys, |k| ShardRequest::Del { key: k }).await } + Command::Unlink { keys } => { + multi_key_bool(engine, &keys, |k| ShardRequest::Unlink { key: k }).await + } + Command::Exists { keys } => { multi_key_bool(engine, &keys, |k| ShardRequest::Exists { key: k }).await } @@ -800,10 +804,17 @@ async fn execute( Err(e) => Frame::Error(format!("ERR {e}")), }, - Command::FlushDb => match engine.broadcast(|| ShardRequest::FlushDb).await { - Ok(_) => Frame::Simple("OK".into()), - Err(e) => Frame::Error(format!("ERR {e}")), - }, + Command::FlushDb { async_mode } => { + let req = if async_mode { + || ShardRequest::FlushDbAsync + } else { + || ShardRequest::FlushDb + }; + match engine.broadcast(req).await { + Ok(_) => Frame::Simple("OK".into()), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } Command::Keys { pattern } => { match engine