feat(meta): optimize expired key cleanup with time-range filtering (#18441)

Update state-machine-api from v0.1.0 to v0.1.1 and improve expired
key cleanup efficiency:

- Add `cleanup_start_timestamp` tracking to avoid re-scanning processed
  tombstone entries
- Simplify expired key removal logic by eliminating redundant checks

Author: drmingdrmer

Cargo.lock

@@ -503,7 +503,7 @@ socket2 = "0.5.3"
 span-map = { version = "0.2.0" }
 sqlx = { version = "0.8", features = ["mysql", "runtime-tokio"] }
 state = "0.6.0"
-state-machine-api = { version = "0.1.0" }
+state-machine-api = { version = "0.1.1" }
 stream-more = "0.1.3"
 strength_reduce = "0.2.4"
 stringslice = "0.2.0"
@@ -666,7 +666,7 @@ openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-al
 orc-rust = { git = "https://github.com/datafuse-extras/orc-rust", rev = "d82aa6d" }
 recursive = { git = "https://github.com/datafuse-extras/recursive.git", rev = "6af35a1" }
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1" }
-state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.1.0" }
+state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.1.1" }
 sub-cache = { git = "https://github.com/databendlabs/sub-cache", tag = "v0.2.1" }
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }
 tantivy-common = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370", package = "tantivy-common" }

Cargo.toml

@@ -660,6 +660,11 @@ where SM: StateMachineApi<SysData> + 'static
         let mut to_clean = vec![];
         let mut strm = self.sm.list_expire_index(log_time_ms).await?;
 
+        // Save the log time for next cleaning.
+        // Avoid listing tombstone records.
+        self.sm
+            .set_cleanup_start_timestamp(Duration::from_millis(log_time_ms));
+
         {
             let mut strm = std::pin::pin!(strm);
             while let Some((expire_key, key)) = strm.try_next().await? {
@@ -672,19 +677,9 @@ where SM: StateMachineApi<SysData> + 'static
         }
 
         for (expire_key, key) in to_clean {
-            let curr = self.sm.get_maybe_expired_kv(&key).await?;
-
-            if let Some(seq_v) = &curr {
-                assert_eq!(expire_key.seq, seq_v.seq);
-                info!("clean expired: {}, {}", key, expire_key);
-
-                self.upsert_kv(&UpsertKV::delete(key.clone())).await?;
-            } else {
-                unreachable!(
-                    "trying to remove un-cleanable: {}, {}, kv-entry: {:?}",
-                    key, expire_key, curr
-                );
-            }
+            let upsert = UpsertKV::delete(key);
+            self.upsert_kv(&upsert).await?;
+            info!("clean expired: {} {expire_key}", upsert.key);
         }
 
         Ok(())

src/meta/raft-store/src/applier/mod.rs

@@ -15,6 +15,7 @@
 use std::fmt;
 use std::fmt::Formatter;
 use std::io;
+use std::time::Duration;
 
 use databend_common_meta_types::raft_types::Entry;
 use databend_common_meta_types::raft_types::StorageError;
@@ -40,6 +41,9 @@ type OnChange = Box<dyn Fn((String, Option<SeqV>, Option<SeqV>)) + Send + Sync>;
 pub struct SMV003 {
     levels: LeveledMap,
 
+    /// Since when to start cleaning expired keys.
+    cleanup_start_time_ms: Duration,
+
     /// Callback when a change is applied to state machine
     pub(crate) on_change_applied: Option<OnChange>,
 }
@@ -77,6 +81,14 @@ impl StateMachineApi<SysData> for SMV003 {
         &mut self.levels
     }
 
+    fn cleanup_start_timestamp(&self) -> Duration {
+        self.cleanup_start_time_ms
+    }
+
+    fn set_cleanup_start_timestamp(&mut self, timestamp: Duration) {
+        self.cleanup_start_time_ms = timestamp;
+    }
+
     fn sys_data_mut(&mut self) -> &mut SysData {
         self.levels.sys_data_mut()
     }

src/meta/raft-store/src/sm_v003/sm_v003.rs

@@ -27,6 +27,7 @@ use display_more::DisplayUnixTimeStampExt;
 use futures_util::StreamExt;
 use futures_util::TryStreamExt;
 use log::debug;
+use log::info;
 use log::warn;
 use map_api::map_api::MapApi;
 use map_api::map_api_ro::MapApiRO;
@@ -192,12 +193,36 @@ pub trait StateMachineApiExt: StateMachineApi<SysData> {
         &self,
         curr_time_ms: u64,
     ) -> Result<IOResultStream<(ExpireKey, String)>, io::Error> {
+        // Since the last saved cleanup timestamp
+        let start_ms = self.cleanup_start_timestamp().as_millis() as u64;
+        let start = ExpireKey::new(start_ms, 0);
+
         // curr_time > expire_at => expired
         let end = ExpireKey::new(curr_time_ms, 0);
 
-        let strm = self.expire_map().range(..end).await?;
+        let msg = {
+            let start = Duration::from_millis(start_ms);
+            let end = Duration::from_millis(curr_time_ms);
+            let msg = format!(
+                "list_expire_index: [{}, {}); interval: {:?}",
+                start.display_unix_timestamp_short(),
+                end.display_unix_timestamp_short(),
+                end.saturating_sub(start)
+            );
+
+            info!("{}", msg);
+
+            msg
+        };
+
+        if start >= end {
+            // No expired entries
+            return Ok(futures_util::stream::empty().boxed());
+        }
+
+        let strm = self.expire_map().range(start..end).await?;
 
-        let strm = add_cooperative_yielding(strm, format!("list_expire_index up to {end}"))
+        let strm = add_cooperative_yielding(strm, msg)
             // Return only non-deleted records
             .try_filter_map(|(k, seq_marked)| {
                 let expire_entry = seq_marked.into_data().map(|v| (k, v));

src/meta/raft-store/src/state_machine_api_ext.rs

@@ -31,11 +31,9 @@ where
     T: Send + 'static,
 {
     stream.enumerate().then(move |(index, item)| {
-        // Yield control every 100 items to prevent blocking other tasks
-        let to_yield = if index % 100 == 0 {
-            if index % 5000 == 0 {
-                info!("{stream_name} yield control to allow other tasks to run: index={index}");
-            }
+        // Yield control every n items to prevent blocking other tasks
+        let to_yield = if index > 0 && index % 5000 == 0 {
+            info!("{stream_name} yield control to allow other tasks to run: index={index}");
             true
         } else {
             false