fix: prevent blocking operations from freezing shard threads

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<String, Entry>),
+}
+
+/// 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<Droppable>,
+}
+
+impl DropHandle {
+    /// Spawns the background drop thread and returns a handle.
+    pub fn spawn() -> Self {
+        let (tx, rx) = mpsc::sync_channel::<Droppable>(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<String, Entry>) {
+        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());
+    }
+}

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();
 

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<DropHandle>,
 }
 
 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<String, Entry> {
+        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<String> {
+        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 {

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;