refactor(meta-service): remove unused txid in raft log (#18392)

drmingdrmer · web-flow · commit a8b79f12c888 · 2025-07-19T17:10:29.000+08:00
diff --git a/src/meta/binaries/meta/entry.rs b/src/meta/binaries/meta/entry.rs
@@ -209,7 +209,6 @@ async fn do_register(meta_node: &Arc<MetaNode>, conf: &Config) -> Result<(), Met
     println!();
 
     let ent = LogEntry {
-        txid: None,
         time_ms: None,
         cmd: Cmd::AddNode {
             node_id,
diff --git a/src/meta/control/src/import.rs b/src/meta/control/src/import.rs
@@ -251,11 +251,7 @@ async fn init_new_cluster(
 
             let entry: Entry = Entry {
                 log_id,
-                payload: EntryPayload::Normal(LogEntry {
-                    txid: None,
-                    time_ms: None,
-                    cmd,
-                }),
+                payload: EntryPayload::Normal(LogEntry { time_ms: None, cmd }),
             };
 
             sto.blocking_append([entry]).await?;
diff --git a/src/meta/process/src/kv_processor.rs b/src/meta/process/src/kv_processor.rs
@@ -139,7 +139,6 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
             Cmd::SetFeature { .. } => Ok(None),
             Cmd::UpsertKV(ups) => {
                 let x = LogEntry {
-                    txid: log_entry.txid,
                     time_ms: log_entry.time_ms,
                     cmd: Cmd::UpsertKV(unwrap_or_return!(self.proc_upsert_kv(ups)?)),
                 };
@@ -162,7 +161,6 @@ where F: Fn(&str, Vec<u8>) -> Result<Vec<u8>, anyhow::Error>
                 }
 
                 Ok(Some(LogEntry {
-                    txid: log_entry.txid,
                     time_ms: log_entry.time_ms,
                     cmd: Cmd::Transaction(TxnRequest::new(condition, if_then).with_else(else_then)),
                 }))
diff --git a/src/meta/raft-store/src/applier/mod.rs b/src/meta/raft-store/src/applier/mod.rs
@@ -117,7 +117,6 @@ where SM: StateMachineApi<SysData> + 'static
             }
             EntryPayload::Normal(ref data) => {
                 info!("apply: normal: {} {}", log_id, data);
-                assert!(data.txid.is_none(), "txid is disabled");
 
                 self.apply_cmd(&data.cmd).await?
             }
diff --git a/src/meta/raft-store/src/state_machine/testing.rs b/src/meta/raft-store/src/state_machine/testing.rs
@@ -18,7 +18,6 @@ use databend_common_meta_types::raft_types::Entry;
 use databend_common_meta_types::raft_types::EntryPayload;
 use databend_common_meta_types::Cmd;
 use databend_common_meta_types::LogEntry;
-use databend_common_meta_types::RaftTxId;
 use databend_common_meta_types::UpsertKV;
 use maplit::btreeset;
 use openraft::entry::RaftEntry;
@@ -39,7 +38,6 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
         Entry {
             log_id: new_log_id(1, 0, 4),
             payload: EntryPayload::Normal(LogEntry {
-                txid: None,
                 time_ms: None,
                 cmd: Cmd::UpsertKV(UpsertKV::update("a", b"A")),
             }),
@@ -63,7 +61,6 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
         Entry {
             log_id: new_log_id(1, 0, 9),
             payload: EntryPayload::Normal(LogEntry {
-                txid: None,
                 time_ms: None,
                 cmd: Cmd::AddNode {
                     node_id: 5,
@@ -85,25 +82,3 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
 
     (logs, want)
 }
-
-// test cases fro Cmd::IncrSeq:
-// case_name, txid, key, want
-pub fn cases_incr_seq() -> Vec<(&'static str, Option<RaftTxId>, &'static str, u64)> {
-    vec![
-        ("incr on none", Some(RaftTxId::new("foo", 1)), "k1", 1),
-        ("incr on existent", Some(RaftTxId::new("foo", 2)), "k1", 2),
-        (
-            "dup: same serial, even with diff key, got the previous result",
-            Some(RaftTxId::new("foo", 2)),
-            "k2",
-            2,
-        ),
-        (
-            "diff client, same serial, not a dup request",
-            Some(RaftTxId::new("bar", 2)),
-            "k2",
-            1,
-        ),
-        ("no txid, no de-dup", None, "k2", 2),
-    ]
-}
diff --git a/src/meta/service/src/meta_service/meta_leader.rs b/src/meta/service/src/meta_service/meta_leader.rs
@@ -193,7 +193,6 @@ impl<'a> MetaLeader<'a> {
         }
 
         let ent = LogEntry {
-            txid: None,
             time_ms: None,
             cmd: Cmd::AddNode {
                 node_id,
@@ -251,7 +250,6 @@ impl<'a> MetaLeader<'a> {
 
         // 2. Remove node info
         let ent = LogEntry {
-            txid: None,
             time_ms: None,
             cmd: Cmd::RemoveNode { node_id },
         };
diff --git a/src/meta/service/src/meta_service/meta_node.rs b/src/meta/service/src/meta_service/meta_node.rs
@@ -1115,7 +1115,6 @@ impl MetaNode {
         node_id: NodeId,
         node: Node,
     ) -> Result<AppliedState, MetaAPIError> {
-        // TODO: use txid?
         let cmd = Cmd::AddNode {
             node_id,
             node,
diff --git a/src/meta/service/tests/it/grpc/metasrv_grpc_export.rs b/src/meta/service/tests/it/grpc/metasrv_grpc_export.rs
@@ -83,11 +83,11 @@ async fn test_export() -> anyhow::Result<()> {
         r#"["raft_log",{"Purged":null}]"#,
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":0,"node_id":0},"index":0},"payload":{"Membership":{"configs":[[0]],"nodes":{"0":{}}}}}}]"#,
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":1},"payload":"Blank"}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":2},"payload":{"Normal":{"txid":null,"time_ms":1111111111111,"cmd":{"AddNode":{"node_id":0,"node":{"name":"0","endpoint":{"addr":"localhost","port":29000},"grpc_api_advertise_address":"127.0.0.1:29000"},"overriding":false}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":2},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"AddNode":{"node_id":0,"node":{"name":"0","endpoint":{"addr":"localhost","port":29000},"grpc_api_advertise_address":"127.0.0.1:29000"},"overriding":false}}}}}}]"#,
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":3},"payload":{"Membership":{"configs":[[0]],"nodes":{"0":{}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":4},"payload":{"Normal":{"txid":null,"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"foo","seq":{"GE":0},"value":{"Update":[102,111,111]},"value_meta":null}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":5},"payload":{"Normal":{"txid":null,"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"bar","seq":{"GE":0},"value":{"Update":[98,97,114]},"value_meta":null}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":6},"payload":{"Normal":{"txid":null,"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"wow","seq":{"GE":0},"value":{"Update":[119,111,119]},"value_meta":null}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":4},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"foo","seq":{"GE":0},"value":{"Update":[102,111,111]},"value_meta":null}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":5},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"bar","seq":{"GE":0},"value":{"Update":[98,97,114]},"value_meta":null}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":6},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"UpsertKV":{"key":"wow","seq":{"GE":0},"value":{"Update":[119,111,119]},"value_meta":null}}}}}}]"#,
         r#"["state_machine/0",{"Sequences":{"key":"generic-kv","value":3}}]"#,
         r#"["state_machine/0",{"StateMachineMeta":{"key":"LastApplied","value":{"LogId":{"leader_id":{"term":1,"node_id":0},"index":6}}}}]"#,
         r#"["state_machine/0",{"StateMachineMeta":{"key":"LastMembership","value":{"Membership":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":3},"membership":{"configs":[[0]],"nodes":{"0":{}}}}}}}]"#,
diff --git a/src/meta/service/tests/it/meta_node/meta_node_lifecycle.rs b/src/meta/service/tests/it/meta_node/meta_node_lifecycle.rs
@@ -724,7 +724,6 @@ async fn test_meta_node_restart_single_node() -> anyhow::Result<()> {
             .assume_leader()
             .await?
             .write(LogEntry {
-                txid: None,
                 time_ms: None,
                 cmd: Cmd::UpsertKV(UpsertKV::update("foo", b"1")),
             })
@@ -817,7 +816,6 @@ async fn assert_upsert_kv_synced(meta_nodes: Vec<Arc<MetaNode>>, key: &str) -> a
             .assume_leader()
             .await?
             .write(LogEntry {
-                txid: None,
                 time_ms: None,
                 cmd: Cmd::UpsertKV(UpsertKV::update(key, key.as_bytes())),
             })
diff --git a/src/meta/service/tests/it/meta_node/meta_node_request_forwarding.rs b/src/meta/service/tests/it/meta_node/meta_node_request_forwarding.rs
@@ -48,7 +48,6 @@ async fn test_meta_node_forward_to_leader() -> anyhow::Result<()> {
         let maybe_leader = MetaLeader::new(mn);
         let rst = maybe_leader
             .write(LogEntry {
-                txid: None,
                 time_ms: None,
                 cmd: Cmd::UpsertKV(UpsertKV::update(key, key.as_bytes())),
             })
diff --git a/src/meta/service/tests/it/meta_node/t90_time_revert_cross_snapshot_boundary.rs b/src/meta/service/tests/it/meta_node/t90_time_revert_cross_snapshot_boundary.rs
@@ -52,7 +52,6 @@ async fn test_meta_node_log_time_revert_cross_snapshot_boundary() -> anyhow::Res
 
     // 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",
@@ -62,7 +61,6 @@ async fn test_meta_node_log_time_revert_cross_snapshot_boundary() -> anyhow::Res
 
     // 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",
diff --git a/src/meta/types/src/lib.rs b/src/meta/types/src/lib.rs
@@ -27,7 +27,6 @@ mod grpc_helper;
 mod log_entry;
 mod message;
 mod operation;
-mod raft_txid;
 mod seq_num;
 mod time;
 mod with;
@@ -105,7 +104,6 @@ pub use protobuf::TxnPutRequest;
 pub use protobuf::TxnPutResponse;
 pub use protobuf::TxnReply;
 pub use protobuf::TxnRequest;
-pub use raft_txid::RaftTxId;
 pub use seq_num::SeqNum;
 pub use time::Interval;
 pub use time::Time;
diff --git a/src/meta/types/src/log_entry.rs b/src/meta/types/src/log_entry.rs
@@ -21,7 +21,6 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::Cmd;
-use crate::RaftTxId;
 
 /// The application data request type which the `metasrv` works with.
 ///
@@ -30,9 +29,6 @@ use crate::RaftTxId;
 /// "client" and "serial", thus the raft engine is able to distinguish if a request is duplicated.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, deepsize::DeepSizeOf)]
 pub struct LogEntry {
-    /// When not None, it is used to filter out duplicated logs, which are caused by retries by client.
-    pub txid: Option<RaftTxId>,
-
     /// The time in millisecond when this log is proposed by the leader.
     ///
     /// State machine depends on clock time to expire values.
@@ -47,9 +43,6 @@ pub struct LogEntry {
 
 impl Display for LogEntry {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        if let Some(txid) = &self.txid {
-            write!(f, "txid: {:?}", txid)?;
-        }
         if let Some(time_ms) = &self.time_ms {
             write!(
                 f,
@@ -64,14 +57,36 @@ impl Display for LogEntry {
 
 impl LogEntry {
     pub fn new(cmd: Cmd) -> Self {
-        Self {
-            txid: None,
-            time_ms: None,
-            cmd,
-        }
+        Self { time_ms: None, cmd }
     }
-    pub fn with_txid(mut self, txid: Option<RaftTxId>) -> Self {
-        self.txid = txid;
-        self
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// With txid before 2025-07-19
+    #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, deepsize::DeepSizeOf)]
+    pub struct LogEntryWithTxid {
+        pub txid: Option<()>,
+
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub time_ms: Option<u64>,
+
+        /// The action a client want to take.
+        pub cmd: Cmd,
+    }
+
+    #[test]
+    fn test_compat_with_and_without_txid() {
+        let json_with_txid = r#"{"txid":null,"time_ms":1667290824603,"cmd":{"UpsertKV":{"key":"__fd_clusters","seq":{"Exact":0},"value":{"Update":[123]},"value_meta":{"expire_at":1667290884}}}}"#;
+
+        let log_entry: LogEntry = serde_json::from_str(json_with_txid).unwrap();
+        println!("{:?}", log_entry);
+
+        let json_without_txid = r#"{"time_ms":1667290824603,"cmd":{"UpsertKV":{"key":"__fd_clusters","seq":{"Exact":0},"value":{"Update":[123]},"value_meta":{"expire_at":1667290884}}}}"#;
+
+        let log_entry: LogEntryWithTxid = serde_json::from_str(json_without_txid).unwrap();
+        println!("{:?}", log_entry);
     }
 }
diff --git a/src/meta/types/src/raft_txid.rs b/src/meta/types/src/raft_txid.rs
diff --git a/tests/metactl/want_exported_v004 b/tests/metactl/want_exported_v004

Original file line number	Diff line number	Diff line change
`@@ -117,7 +117,6 @@ where SM: StateMachineApi<SysData> + 'static`
`117`	`117`	`}`
`118`	`118`	`EntryPayload::Normal(ref data) => {`
`119`	`119`	`info!("apply: normal: {} {}", log_id, data);`
`120`		`- assert!(data.txid.is_none(), "txid is disabled");`
`121`	`120`
`122`	`121`	`self.apply_cmd(&data.cmd).await?`
`123`	`122`	`}`