observability and reliability: prometheus counters, crash recovery, TLS tests

crates/ember-core/src/engine.rs

@@ -36,6 +36,8 @@ pub struct EngineConfig {
     /// [`ReplicationEvent`] so replication clients can stream it to
     /// replicas.
     pub replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    /// Optional channel to receive expired key names. Used for keyspace notifications.
+    pub expired_tx: Option<broadcast::Sender<String>>,
     /// Optional schema registry for protobuf value validation.
     /// When set, enables PROTO.* commands.
     #[cfg(feature = "protobuf")]
@@ -53,6 +55,7 @@ pub struct EngineConfig {
 pub struct Engine {
     shards: Vec<ShardHandle>,
     replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    expired_tx: Option<broadcast::Sender<String>>,
     #[cfg(feature = "protobuf")]
     schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 }
@@ -96,6 +99,7 @@ impl Engine {
                     config.persistence.clone(),
                     Some(drop_handle.clone()),
                     config.replication_tx.clone(),
+                    config.expired_tx.clone(),
                     #[cfg(feature = "protobuf")]
                     config.schema_registry.clone(),
                 )
@@ -105,6 +109,7 @@ impl Engine {
         Self {
             shards,
             replication_tx: config.replication_tx,
+            expired_tx: config.expired_tx,
             #[cfg(feature = "protobuf")]
             schema_registry: config.schema_registry,
         }
@@ -144,6 +149,7 @@ impl Engine {
                 config.persistence.clone(),
                 Some(drop_handle.clone()),
                 config.replication_tx.clone(),
+                config.expired_tx.clone(),
                 #[cfg(feature = "protobuf")]
                 config.schema_registry.clone(),
             );
@@ -154,6 +160,7 @@ impl Engine {
         let engine = Self {
             shards: handles,
             replication_tx: config.replication_tx,
+            expired_tx: config.expired_tx,
             #[cfg(feature = "protobuf")]
             schema_registry: config.schema_registry,
         };
@@ -197,6 +204,14 @@ impl Engine {
         self.replication_tx.as_ref().map(|tx| tx.subscribe())
     }
 
+    /// Creates a new broadcast receiver for expired key names.
+    ///
+    /// Returns `None` if no expired-key channel was configured. Used by the
+    /// server to subscribe a background task that fires keyspace notifications.
+    pub fn subscribe_expired(&self) -> Option<broadcast::Receiver<String>> {
+        self.expired_tx.as_ref().map(|tx| tx.subscribe())
+    }
+
     /// Sends a request to a specific shard by index.
     ///
     /// Used by SCAN to iterate through shards sequentially.

crates/ember-core/src/expiry.rs

@@ -18,15 +18,16 @@ const MAX_ROUNDS: usize = 3;
 
 /// Runs one active expiration cycle on the keyspace.
 ///
-/// Samples up to `SAMPLE_SIZE` random keys per round, removes expired
-/// ones, and repeats if more than 25% of the sample was expired (up to
-/// `MAX_ROUNDS` total). Returns the total number of keys removed.
-pub fn run_expiration_cycle(ks: &mut Keyspace) -> usize {
-    let mut total_removed = 0;
+/// Returns the keys removed this cycle. The caller can use this to fire
+/// keyspace notifications. When nobody is listening, the caller drops
+/// the returned Vec immediately (no allocation if empty).
+pub fn run_expiration_cycle(ks: &mut Keyspace) -> Vec<String> {
+    let mut expired_keys = Vec::new();
 
     for _ in 0..MAX_ROUNDS {
-        let removed = ks.expire_sample(SAMPLE_SIZE);
-        total_removed += removed;
+        let prev_len = expired_keys.len();
+        ks.expire_sample(SAMPLE_SIZE, &mut expired_keys);
+        let removed = expired_keys.len() - prev_len;
 
         // if we removed fewer than 25% of the sample, the keyspace
         // is reasonably clean — stop early
@@ -35,7 +36,7 @@ pub fn run_expiration_cycle(ks: &mut Keyspace) -> usize {
         }
     }
 
-    total_removed
+    expired_keys
 }
 
 #[cfg(test)]
@@ -52,7 +53,7 @@ mod tests {
             ks.set(format!("key:{i}"), Bytes::from("val"), None, false, false);
         }
         let removed = run_expiration_cycle(&mut ks);
-        assert_eq!(removed, 0);
+        assert!(removed.is_empty());
         assert_eq!(ks.len(), 10);
     }
 
@@ -77,7 +78,7 @@ mod tests {
         thread::sleep(Duration::from_millis(20));
 
         let removed = run_expiration_cycle(&mut ks);
-        assert_eq!(removed, 10);
+        assert_eq!(removed.len(), 10);
         assert_eq!(ks.len(), 5);
     }
 
@@ -94,14 +95,14 @@ mod tests {
             );
         }
         let removed = run_expiration_cycle(&mut ks);
-        assert_eq!(removed, 0);
+        assert!(removed.is_empty());
         assert_eq!(ks.len(), 10);
     }
 
     #[test]
     fn empty_keyspace_is_fine() {
         let mut ks = Keyspace::new();
         let removed = run_expiration_cycle(&mut ks);
-        assert_eq!(removed, 0);
+        assert!(removed.is_empty());
     }
 }

crates/ember-core/src/keyspace/mod.rs

@@ -1262,9 +1262,11 @@ impl Keyspace {
 
     /// Randomly samples up to `count` keys and removes any that have expired.
     ///
-    /// Returns the number of keys actually removed. Used by the active
-    /// expiration cycle to clean up keys that no one is reading.
-    pub fn expire_sample(&mut self, count: usize) -> usize {
+    /// Samples up to `count` random keys and removes any that have expired.
+    ///
+    /// Expired key names are appended to `out` so the caller can emit
+    /// keyspace notifications. Returns the number of keys removed.
+    pub(crate) fn expire_sample(&mut self, count: usize, out: &mut Vec<String>) -> usize {
         if self.entries.is_empty() {
             return 0;
         }
@@ -1280,8 +1282,9 @@ impl Keyspace {
             .collect();
 
         let mut removed = 0;
-        for key in &keys_to_check {
-            if self.remove_if_expired(key) {
+        for key in keys_to_check {
+            if self.remove_if_expired(&key) {
+                out.push(key);
                 removed += 1;
             }
         }

crates/ember-core/src/shard/mod.rs

@@ -959,6 +959,8 @@ pub struct PreparedShard {
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
     replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    /// Optional channel to broadcast expired key names for keyspace notifications.
+    expired_tx: Option<broadcast::Sender<String>>,
     #[cfg(feature = "protobuf")]
     schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 }
@@ -973,6 +975,7 @@ pub fn prepare_shard(
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
     replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    expired_tx: Option<broadcast::Sender<String>>,
     #[cfg(feature = "protobuf")] schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 ) -> (ShardHandle, PreparedShard) {
     let (tx, rx) = mpsc::channel(buffer);
@@ -982,6 +985,7 @@ pub fn prepare_shard(
         persistence,
         drop_handle,
         replication_tx,
+        expired_tx,
         #[cfg(feature = "protobuf")]
         schema_registry,
     };
@@ -999,6 +1003,7 @@ pub async fn run_prepared(prepared: PreparedShard) {
         prepared.persistence,
         prepared.drop_handle,
         prepared.replication_tx,
+        prepared.expired_tx,
         #[cfg(feature = "protobuf")]
         prepared.schema_registry,
     )
@@ -1019,6 +1024,7 @@ pub fn spawn_shard(
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
     replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    expired_tx: Option<broadcast::Sender<String>>,
     #[cfg(feature = "protobuf")] schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 ) -> ShardHandle {
     let (handle, prepared) = prepare_shard(
@@ -1027,6 +1033,7 @@ pub fn spawn_shard(
         persistence,
         drop_handle,
         replication_tx,
+        expired_tx,
         #[cfg(feature = "protobuf")]
         schema_registry,
     );
@@ -1042,6 +1049,7 @@ async fn run_shard(
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
     replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
+    expired_tx: Option<broadcast::Sender<String>>,
     #[cfg(feature = "protobuf")] schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 ) {
     let shard_id = config.shard_id;
@@ -1226,7 +1234,14 @@ async fn run_shard(
                 }
             }
             _ = expiry_tick.tick() => {
-                expiry::run_expiration_cycle(&mut keyspace);
+                let expired_keys = expiry::run_expiration_cycle(&mut keyspace);
+                if let Some(ref tx) = expired_tx {
+                    if !expired_keys.is_empty() && tx.receiver_count() > 0 {
+                        for key in &expired_keys {
+                            let _ = tx.send(key.clone());
+                        }
+                    }
+                }
             }
             _ = fsync_tick.tick(), if fsync_policy == FsyncPolicy::EverySec => {
                 if let Some(ref mut writer) = aof_writer {