fix(meta-service): Remove transient expire_cursor to prevent state inconsistency (#18318)

* refactor(meta-service): write test to expose a bug with expiration consistency

Tests timestamp-based expiration consistency across snapshot boundaries.

**Scenario:**
- Log-1 has timestamp T+180s, included in snapshot
- Log-2 has timestamp T+60s with expiration T+120s, not in snapshot

**Issue:** After restart, Log-2 uses current time instead of last-seen time (from Log-1)
for expiration checks, potentially causing state inconsistencies between nodes.

This bug raises in real world if:
- When meta-service restarts, there is one log(`A`) to re-apply that has
  smaller `expire_at`(`t1`) than the biggest timestamp(`LogEntry.time_ms`)
  in the snapshot,
- and all the log entries before `A` have smaller timestamp(`LogEntry.time_ms`) than `t1`.

* fix(meta-service): Remove transient `expire_cursor` to prevent state inconsistency

Fix potential state machine inconsistencies caused by the non-persisted
`expire_cursor` field. While inconsistencies were rare, the transient
state could cause meta-service nodes to have divergent views of the
expiration processing state.

Root cause:

The `expire_cursor` was maintained as transient state outside the
replicated state machine, making it possible for nodes to have different
cursor positions after restarts or network partitions.

Solution:

Remove the non-persisted `expire_cursor` entirely to ensure all state
machine operations are fully deterministic and consistently replicated
across all meta-service nodes.

This change eliminates the race condition and guarantees state machine
consistency regardless of node restart patterns or network conditions.

Author: drmingdrmer

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

@@ -688,8 +688,6 @@ where SM: StateMachineApi + 'static
             }
         }
 
-        self.sm.update_expire_cursor(log_time_ms);
-
         Ok(())
     }
 

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

@@ -46,14 +46,6 @@ pub struct SMV003 {
 impl StateMachineApi for SMV003 {
     type Map = LeveledMap;
 
-    fn get_expire_cursor(&self) -> ExpireKey {
-        self.expire_cursor
-    }
-
-    fn set_expire_cursor(&mut self, cursor: ExpireKey) {
-        self.expire_cursor = cursor;
-    }
-
     fn map_ref(&self) -> &Self::Map {
         &self.levels
     }

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

@@ -14,6 +14,7 @@
 
 use databend_common_meta_types::seq_value::SeqV;
 use databend_common_meta_types::seq_value::SeqValue;
+use databend_common_meta_types::CmdContext;
 use databend_common_meta_types::UpsertKV;
 use futures_util::TryStreamExt;
 use map_api::map_api_ro::MapApiRO;
@@ -137,26 +138,6 @@ async fn test_two_level_upsert_get_range() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn test_update_expire_index() -> anyhow::Result<()> {
-    let mut sm = SMV003::default();
-
-    sm.update_expire_cursor(1);
-    assert_eq!(sm.get_expire_cursor(), ExpireKey::new(1, 0));
-
-    sm.update_expire_cursor(2);
-    assert_eq!(sm.get_expire_cursor(), ExpireKey::new(2, 0));
-
-    sm.update_expire_cursor(1);
-    assert_eq!(
-        sm.get_expire_cursor(),
-        ExpireKey::new(2, 0),
-        "expire cursor can not go back"
-    );
-
-    Ok(())
-}
-
 /// The subscript is internal_seq:
 ///
 ///    | kv             | expire
@@ -221,7 +202,7 @@ async fn test_internal_expire_index() -> anyhow::Result<()> {
 
 #[tokio::test]
 async fn test_list_expire_index() -> anyhow::Result<()> {
-    let mut sm = build_sm_with_expire().await?;
+    let sm = build_sm_with_expire().await?;
 
     let curr_time_ms = 5000;
     let got = sm
@@ -251,15 +232,14 @@ async fn test_list_expire_index() -> anyhow::Result<()> {
         (ExpireKey::new(20000, 3), s("c")),
     ]);
 
-    sm.update_expire_cursor(15000);
-
     let curr_time_ms = 20_001;
     let got = sm
         .list_expire_index(curr_time_ms)
         .await?
         .try_collect::<Vec<_>>()
         .await?;
     assert_eq!(got, vec![
+        (ExpireKey::new(5000, 2), s("b")),
         (ExpireKey::new(15000, 4), s("a")),
         (ExpireKey::new(20000, 3), s("c")),
     ]);
@@ -270,9 +250,8 @@ async fn test_list_expire_index() -> anyhow::Result<()> {
 async fn test_inserting_expired_becomes_deleting() -> anyhow::Result<()> {
     let mut sm = build_sm_with_expire().await?;
 
-    sm.update_expire_cursor(15_000);
-
     let mut a = sm.new_applier();
+    a.cmd_ctx = CmdContext::from_millis(15_000);
 
     // Inserting an expired entry will delete it.
     a.upsert_kv(&UpsertKV::update("a", b"a1").with_expire_sec(10))
@@ -290,7 +269,7 @@ async fn test_inserting_expired_becomes_deleting() -> anyhow::Result<()> {
         .await?
         .try_collect::<Vec<_>>()
         .await?;
-    assert!(got.is_empty());
+    assert_eq!(got, vec![(ExpireKey::new(5_000, 2), s("b")),]);
 
     let got = sm
         .list_expire_index(20_001)
@@ -299,6 +278,7 @@ async fn test_inserting_expired_becomes_deleting() -> anyhow::Result<()> {
         .await?;
     assert_eq!(got, vec![
         //
+        (ExpireKey::new(5_000, 2), s("b")),
         (ExpireKey::new(20_000, 3), s("c")),
     ]);
 

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

@@ -50,20 +50,6 @@ where T: SMEventSender
 pub trait StateMachineApi: Send + Sync {
     type Map: MapApi<String, KVMeta> + MapApi<ExpireKey, KVMeta> + 'static;
 
-    /// Returns the current expire key cursor position.
-    ///
-    /// The expiry key cursor marks a boundary in the key space:
-    /// - All keys before this cursor (exclusive) have already been processed and deleted
-    /// - This cursor position is used to track progress when incrementally cleaning up expired keys
-    fn get_expire_cursor(&self) -> ExpireKey;
-
-    /// Updates the expiry key cursor position.
-    ///
-    /// This method is called after a batch of expired keys have been processed and deleted.
-    /// The new cursor position indicates that all keys before it (exclusive) have been
-    /// successfully cleaned up.
-    fn set_expire_cursor(&mut self, cursor: ExpireKey);
-
     /// Returns a reference to the map that stores application data.
     ///
     /// This method provides read-only access to the underlying key-value store

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

@@ -73,11 +73,12 @@ pub trait StateMachineApiExt: StateMachineApi {
         };
 
         let expire_ms = kv_meta.expires_at_ms();
-        if expire_ms < self.get_expire_cursor().time_ms {
+        let curr_time_ms = cmd_ctx.time().millis();
+        if expire_ms < curr_time_ms {
             warn!(
-                "upsert_kv_primary_index: expired key inserted: {} < expire-cursor: {}; key: {}",
+                "upsert_kv_primary_index: expired key inserted: {} < timestamp in log entry: {}; key: {}",
                 Duration::from_millis(expire_ms).display_unix_timestamp_short(),
-                Duration::from_millis(self.get_expire_cursor().time_ms)
+                Duration::from_millis(curr_time_ms)
                     .display_unix_timestamp_short(),
                 upsert_kv.key
             );
@@ -177,42 +178,21 @@ pub trait StateMachineApiExt: StateMachineApi {
         &self,
         curr_time_ms: u64,
     ) -> Result<IOResultStream<(ExpireKey, String)>, io::Error> {
-        let start = self.get_expire_cursor();
-
         // curr_time > expire_at => expired
         let end = ExpireKey::new(curr_time_ms, 0);
 
-        // There is chance the raft leader produce smaller timestamp for later logs.
-        if start >= end {
-            return Ok(futures::stream::empty().boxed());
-        }
-
-        let strm = self.map_ref().expire_map().range(start..end).await?;
+        let strm = self.map_ref().expire_map().range(..end).await?;
 
-        let strm =
-            add_cooperative_yielding(strm, format!("list_expire_index since {start} to {end}"))
-                // Return only non-deleted records
-                .try_filter_map(|(k, marked)| {
-                    let expire_entry = marked.unpack().map(|(v, _v_meta)| (k, v));
-                    future::ready(Ok(expire_entry))
-                });
+        let strm = add_cooperative_yielding(strm, format!("list_expire_index up to {end}"))
+            // Return only non-deleted records
+            .try_filter_map(|(k, marked)| {
+                let expire_entry = marked.unpack().map(|(v, _v_meta)| (k, v));
+                future::ready(Ok(expire_entry))
+            });
 
         Ok(strm.boxed())
     }
 
-    fn update_expire_cursor(&mut self, log_time_ms: u64) {
-        if log_time_ms < self.get_expire_cursor().time_ms {
-            warn!(
-                "update_last_cleaned: log_time_ms {} < last_cleaned_expire.time_ms {}",
-                log_time_ms,
-                self.get_expire_cursor().time_ms
-            );
-            return;
-        }
-
-        self.set_expire_cursor(ExpireKey::new(log_time_ms, 0));
-    }
-
     /// Get a cloned value by key.
     ///
     /// It does not check expiration of the returned entry.

src/meta/service/tests/it/meta_node/mod.rs

@@ -17,3 +17,4 @@ pub(crate) mod meta_node_kv_api_expire;
 pub(crate) mod meta_node_lifecycle;
 pub(crate) mod meta_node_replication;
 pub(crate) mod meta_node_request_forwarding;
+pub(crate) mod t90_time_revert_cross_snapshot_boundary;

src/meta/service/tests/it/meta_node/t90_time_revert_cross_snapshot_boundary.rs

@@ -0,0 +1,140 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::time::Duration;
+
+use databend_common_meta_kvapi::kvapi::KVApi;
+use databend_common_meta_types::raft_types::new_log_id;
+use databend_common_meta_types::Cmd;
+use databend_common_meta_types::LogEntry;
+use databend_common_meta_types::SeqV;
+use databend_common_meta_types::TxnOp;
+use databend_common_meta_types::TxnRequest;
+use databend_meta::meta_service::MetaNode;
+use log::info;
+use maplit::btreeset;
+use test_harness::test;
+use tokio::time::sleep;
+
+use crate::testing::meta_service_test_harness;
+use crate::tests::meta_node::start_meta_node_cluster;
+use crate::tests::meta_node::timeout;
+
+/// Tests timestamp-based expiration consistency across snapshot boundaries.
+///
+/// **Scenario:**
+/// - Log-1 has timestamp T+180s, included in snapshot
+/// - Log-2 has timestamp T+60s with expiration T+120s, not in snapshot
+///
+/// **Issue:** After restart, Log-2 uses current time instead of last-seen time (from Log-1)
+/// for expiration checks, potentially causing state inconsistencies between nodes.
+///
+/// This bug raises in real world if:
+/// - When meta-service restarts, there is one log(`A`) to re-apply that has smaller `expire_at`(`t1`)
+///   than the biggest timestamp(`LogEntry.time_ms`) in the snapshot,
+/// - and all the log entries before `A` have smaller timestamp(`LogEntry.time_ms`) than `t1`.
+#[test(harness = meta_service_test_harness)]
+#[fastrace::trace]
+async fn test_meta_node_log_time_revert_cross_snapshot_boundary() -> anyhow::Result<()> {
+    let now_ms = SeqV::<()>::now_ms();
+
+    // Log with later timestamp (T+180s) - will be included in snapshot
+    let log_later = LogEntry {
+        txid: None,
+        time_ms: Some(now_ms + 180_000),
+        cmd: Cmd::Transaction(TxnRequest::new(vec![], vec![TxnOp::put(
+            "k1",
+            b"v1".to_vec(),
+        )])),
+    };
+
+    // Log with earlier timestamp (T+60s) but expires at T+120s - not in snapshot
+    let log_earlier = LogEntry {
+        txid: None,
+        time_ms: Some(now_ms + 60_000),
+        cmd: Cmd::Transaction(TxnRequest::new(vec![], vec![TxnOp::put(
+            "k1",
+            b"v2".to_vec(),
+        )
+        .with_expires_at_ms(Some(now_ms + 120_000))])),
+    };
+
+    let result_with_restart = write_two_logs(log_later.clone(), log_earlier.clone(), true).await?;
+    let result_without_restart = write_two_logs(log_later, log_earlier, false).await?;
+
+    assert_eq!(
+        result_with_restart.is_some(),
+        result_without_restart.is_some()
+    );
+
+    Ok(())
+}
+
+/// Writes two logs with optional snapshot and restart between them.
+///
+/// Tests how state machine behavior differs when last-seen time context is lost
+/// during restart. Returns the final value of key "k1".
+async fn write_two_logs(
+    first_log: LogEntry,
+    second_log: LogEntry,
+    restart: bool,
+) -> anyhow::Result<Option<SeqV>> {
+    info!("Testing log sequence with restart={}", restart);
+
+    let (mut log_index, mut test_contexts) =
+        start_meta_node_cluster(btreeset![0], btreeset![]).await?;
+
+    let mut tc0 = test_contexts.remove(0);
+    let mut meta_node = tc0.meta_node.take().unwrap();
+
+    // Apply first log
+    meta_node.raft.client_write(first_log).await?;
+    log_index += 1;
+
+    // Restart node with snapshot if requested
+    if restart {
+        info!("Taking snapshot at log_index={}", log_index);
+        meta_node.raft.trigger().snapshot().await?;
+        meta_node
+            .raft
+            .wait(timeout())
+            .snapshot(new_log_id(1, 0, log_index), "purged")
+            .await?;
+
+        info!("Restarting meta-node");
+        meta_node.stop().await?;
+        drop(meta_node);
+
+        sleep(Duration::from_secs(2)).await;
+
+        meta_node = MetaNode::open(&tc0.config.raft_config).await?;
+    }
+
+    // Apply second log
+    meta_node.raft.client_write(second_log).await?;
+
+    // Check final state
+    let result = meta_node
+        .raft_store
+        .state_machine
+        .read()
+        .await
+        .kv_api()
+        .get_kv("k1")
+        .await?;
+
+    info!("Final state for k1 (restart={}): {:?}", restart, result);
+
+    Ok(result)
+}