refactor(meta): upgrade OpenRaft to v0.10.0-alpha.9

Cargo.toml

@@ -397,7 +397,6 @@ opendal = { version = "0.51.2", features = [
 openraft = { version = "0.10.0", features = [
     "serde",
     "tracing-log",
-    "loosen-follower-log-revert", # allows removing all data from a follower and restoring from the leader.
 ] }
 opensrv-mysql = { git = "https://github.com/databendlabs/opensrv.git", rev = "a1fb4da", features = ["tls"] }
 orc-rust = "0.5.0"
@@ -636,7 +635,7 @@ deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "c149502" }
 ethnum = { git = "https://github.com/datafuse-extras/ethnum-rs", rev = "4cb05f1" }
 map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.2.3" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
-openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.7" }
+openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.9" }
 orc-rust = { git = "https://github.com/youngsofun/orc-rust", rev = "94ab8e9" }
 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" }

src/meta/control/src/import.rs

@@ -38,7 +38,7 @@ use databend_common_meta_sled_store::init_get_sled_db;
 use databend_common_meta_sled_store::openraft::storage::RaftLogStorageExt;
 use databend_common_meta_sled_store::openraft::RaftSnapshotBuilder;
 use databend_common_meta_types::node::Node;
-use databend_common_meta_types::raft_types::CommittedLeaderId;
+use databend_common_meta_types::raft_types::new_log_id;
 use databend_common_meta_types::raft_types::Entry;
 use databend_common_meta_types::raft_types::EntryPayload;
 use databend_common_meta_types::raft_types::LogId;
@@ -314,10 +314,10 @@ async fn init_new_cluster(
     };
 
     let last_log_id = std::cmp::max(last_applied, max_log_id);
-    let mut log_id = last_log_id.unwrap_or(LogId::new(CommittedLeaderId::new(0, 0), 0));
+    let mut log_id = last_log_id.unwrap_or(new_log_id(0, 0, 0));
 
     let node_ids = nodes.keys().copied().collect::<BTreeSet<_>>();
-    let membership = Membership::new(vec![node_ids], ());
+    let membership = Membership::new_with_defaults(vec![node_ids], []);
 
     // Update snapshot: Replace nodes set and membership config.
     {

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

@@ -51,7 +51,10 @@ async fn test_compact_copied_value_and_kv() -> anyhow::Result<()> {
     assert_eq!(compacted.iter_immutable_levels().count(), 1);
     assert_eq!(
         d.last_membership_ref(),
-        &StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()))
+        &StoredMembership::new(
+            Some(log_id(3, 3, 3)),
+            Membership::new_with_defaults(vec![], [])
+        )
     );
     assert_eq!(d.last_applied_ref(), &Some(log_id(3, 3, 3)));
     assert_eq!(
@@ -203,8 +206,10 @@ async fn build_3_levels() -> anyhow::Result<LeveledMap> {
     let mut lm = LeveledMap::default();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(1, 1, 1)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(1, 1, 1)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(1, 1, 1));
     *sd.nodes_mut() = btreemap! {1=>Node::new("1", Endpoint::new("1", 1))};
 
@@ -217,8 +222,10 @@ async fn build_3_levels() -> anyhow::Result<LeveledMap> {
     lm.freeze_writable();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(2, 2, 2)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(2, 2, 2)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(2, 2, 2));
     *sd.nodes_mut() = btreemap! {2=>Node::new("2", Endpoint::new("2", 2))};
 
@@ -230,8 +237,10 @@ async fn build_3_levels() -> anyhow::Result<LeveledMap> {
     lm.freeze_writable();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(3, 3, 3)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(3, 3, 3));
     *sd.nodes_mut() = btreemap! {3=>Node::new("3", Endpoint::new("3", 3))};
 

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

@@ -44,7 +44,10 @@ async fn test_leveled_query_with_db() -> anyhow::Result<()> {
     assert_eq!(lm.curr_seq(), 7);
     assert_eq!(
         lm.last_membership_ref(),
-        &StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()))
+        &StoredMembership::new(
+            Some(log_id(3, 3, 3)),
+            Membership::new_with_defaults(vec![], [])
+        )
     );
     assert_eq!(lm.last_applied_ref(), &Some(log_id(3, 3, 3)));
     assert_eq!(
@@ -93,7 +96,7 @@ async fn test_leveled_query_with_expire_index() -> anyhow::Result<()> {
     assert_eq!(lm.curr_seq(), 4);
     assert_eq!(
         lm.last_membership_ref(),
-        &StoredMembership::new(None, Membership::new(vec![], ()))
+        &StoredMembership::new(None, Membership::new_with_defaults(vec![], []))
     );
     assert_eq!(lm.last_applied_ref(), &None);
     assert_eq!(lm.nodes_ref(), &btreemap! {});
@@ -161,7 +164,10 @@ async fn test_compact() -> anyhow::Result<()> {
     assert_eq!(db.curr_seq(), 7);
     assert_eq!(
         db.last_membership_ref(),
-        &StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()))
+        &StoredMembership::new(
+            Some(log_id(3, 3, 3)),
+            Membership::new_with_defaults(vec![], [])
+        )
     );
     assert_eq!(db.last_applied_ref(), &Some(log_id(3, 3, 3)));
     assert_eq!(
@@ -210,7 +216,7 @@ async fn test_compact_expire_index() -> anyhow::Result<()> {
     assert_eq!(db.curr_seq(), 4);
     assert_eq!(
         db.last_membership_ref(),
-        &StoredMembership::new(None, Membership::new(vec![], ()))
+        &StoredMembership::new(None, Membership::new_with_defaults(vec![], []))
     );
     assert_eq!(db.last_applied_ref(), &None);
     assert_eq!(db.nodes_ref(), &btreemap! {});
@@ -274,7 +280,10 @@ async fn test_compact_output_3_level() -> anyhow::Result<()> {
     assert_eq!(sys_data.curr_seq(), 7);
     assert_eq!(
         sys_data.last_membership_ref(),
-        &StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()))
+        &StoredMembership::new(
+            Some(log_id(3, 3, 3)),
+            Membership::new_with_defaults(vec![], [])
+        )
     );
     assert_eq!(sys_data.last_applied_ref(), &Some(log_id(3, 3, 3)));
     assert_eq!(
@@ -305,8 +314,10 @@ async fn build_3_levels() -> anyhow::Result<(LeveledMap, impl Drop)> {
     let mut lm = LeveledMap::default();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(1, 1, 1)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(1, 1, 1)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(1, 1, 1));
     *sd.nodes_mut() = btreemap! {1=>Node::new("1", Endpoint::new("1", 1))};
 
@@ -319,8 +330,10 @@ async fn build_3_levels() -> anyhow::Result<(LeveledMap, impl Drop)> {
     lm.freeze_writable();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(2, 2, 2)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(2, 2, 2)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(2, 2, 2));
     *sd.nodes_mut() = btreemap! {2=>Node::new("2", Endpoint::new("2", 2))};
 
@@ -332,8 +345,10 @@ async fn build_3_levels() -> anyhow::Result<(LeveledMap, impl Drop)> {
     lm.freeze_writable();
     let sd = lm.writable_mut().sys_data_mut();
 
-    *sd.last_membership_mut() =
-        StoredMembership::new(Some(log_id(3, 3, 3)), Membership::new(vec![], ()));
+    *sd.last_membership_mut() = StoredMembership::new(
+        Some(log_id(3, 3, 3)),
+        Membership::new_with_defaults(vec![], []),
+    );
     *sd.last_applied_mut() = Some(log_id(3, 3, 3));
     *sd.nodes_mut() = btreemap! {3=>Node::new("3", Endpoint::new("3", 3))};
 

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

@@ -21,7 +21,7 @@ use databend_common_meta_types::snapshot_db::DB;
 use databend_common_meta_types::sys_data::SysData;
 use databend_common_meta_types::AppliedState;
 use log::info;
-use openraft::RaftLogId;
+use openraft::entry::RaftEntry;
 
 use crate::applier::Applier;
 use crate::leveled_store::leveled_map::compactor::Compactor;
@@ -137,7 +137,7 @@ impl SMV003 {
         let mut res = vec![];
 
         for ent in entries.into_iter() {
-            let log_id = *ent.get_log_id();
+            let log_id = ent.log_id();
             let r = applier
                 .apply(&ent)
                 .await

src/meta/raft-store/src/state_machine/testing.rs

@@ -29,7 +29,10 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
     let logs = vec![
         Entry {
             log_id: new_log_id(1, 0, 1),
-            payload: EntryPayload::Membership(Membership::new(vec![btreeset![1, 2, 3]], ())),
+            payload: EntryPayload::Membership(Membership::new_with_defaults(
+                vec![btreeset![1, 2, 3]],
+                [],
+            )),
         },
         Entry::new_blank(new_log_id(1, 0, 2)),
         Entry::new_blank(new_log_id(1, 0, 3)),
@@ -43,7 +46,10 @@ pub fn snapshot_logs() -> (Vec<Entry>, Vec<String>) {
         },
         Entry {
             log_id: new_log_id(1, 0, 5),
-            payload: EntryPayload::Membership(Membership::new(vec![btreeset![4, 5, 6]], ())),
+            payload: EntryPayload::Membership(Membership::new_with_defaults(
+                vec![btreeset![4, 5, 6]],
+                [],
+            )),
         },
         Entry {
             log_id: new_log_id(1, 0, 6),

src/meta/service/src/meta_service/meta_node.rs

@@ -43,10 +43,9 @@ use databend_common_meta_types::protobuf::raft_service_server::RaftServiceServer
 use databend_common_meta_types::protobuf::watch_request::FilterType;
 use databend_common_meta_types::protobuf::WatchRequest;
 use databend_common_meta_types::protobuf::WatchResponse;
-use databend_common_meta_types::raft_types::CommittedLeaderId;
+use databend_common_meta_types::raft_types::new_log_id;
 use databend_common_meta_types::raft_types::ForwardToLeader;
 use databend_common_meta_types::raft_types::InitializeError;
-use databend_common_meta_types::raft_types::LogId;
 use databend_common_meta_types::raft_types::MembershipNode;
 use databend_common_meta_types::raft_types::NodeId;
 use databend_common_meta_types::raft_types::RaftMetrics;
@@ -243,6 +242,9 @@ impl MetaNode {
             snapshot_policy: SnapshotPolicy::LogsSinceLast(config.snapshot_logs_since_last),
             max_in_snapshot_log_to_keep: config.max_applied_log_to_keep,
             snapshot_max_chunk_size: config.snapshot_chunk_size,
+            // Allow Leader to reset replication if a follower clears its log.
+            // Usefull in a testing environment.
+            allow_log_reversion: Some(true),
             ..Default::default()
         }
         .validate()
@@ -907,9 +909,7 @@ impl MetaNode {
             is_leader: metrics.state == openraft::ServerState::Leader,
             current_term: metrics.current_term,
             last_log_index: metrics.last_log_index.unwrap_or(0),
-            last_applied: metrics
-                .last_applied
-                .unwrap_or(LogId::new(CommittedLeaderId::new(0, 0), 0)),
+            last_applied: metrics.last_applied.unwrap_or(new_log_id(0, 0, 0)),
             snapshot_last_log_id: metrics.snapshot,
             purged: metrics.purged,
             leader,

src/meta/service/src/meta_service/raft_service_impl.rs

@@ -169,7 +169,7 @@ impl RaftServiceImpl {
 
             let snapshot = Snapshot {
                 meta: snapshot_meta.clone(),
-                snapshot: Box::new(db),
+                snapshot: db,
             };
 
             let resp = raft.install_full_snapshot(req_vote, snapshot).await?;
@@ -206,7 +206,7 @@ impl RaftServiceImpl {
 
         let snapshot = Snapshot {
             meta: snapshot_meta,
-            snapshot: Box::new(db),
+            snapshot: db,
         };
 
         let res = self

src/meta/service/src/store/raft_log_storage_impl.rs

@@ -21,12 +21,12 @@ use databend_common_base::display::display_option::DisplayOptionExt;
 use databend_common_meta_raft_store::raft_log_v004;
 use databend_common_meta_raft_store::raft_log_v004::codec_wrapper::Cw;
 use databend_common_meta_raft_store::raft_log_v004::io_desc::IODesc;
+use databend_common_meta_sled_store::openraft::entry::RaftEntry;
 use databend_common_meta_sled_store::openraft::storage::RaftLogStorage;
 use databend_common_meta_sled_store::openraft::EntryPayload;
 use databend_common_meta_sled_store::openraft::LogIdOptionExt;
 use databend_common_meta_sled_store::openraft::LogState;
 use databend_common_meta_sled_store::openraft::OptionalSend;
-use databend_common_meta_sled_store::openraft::RaftLogId;
 use databend_common_meta_sled_store::openraft::RaftLogReader;
 use databend_common_meta_types::raft_types::Entry;
 use databend_common_meta_types::raft_types::IOFlushed;
@@ -70,7 +70,7 @@ impl RaftLogReader<TypeConfig> for RaftStore {
 
                 debug!(
                     "RaftStore::limited_get_log_entries: got log: log_id: {}, size: {}",
-                    ent.get_log_id(),
+                    ent.log_id(),
                     size
                 );
 
@@ -82,8 +82,8 @@ impl RaftLogReader<TypeConfig> for RaftStore {
                         "RaftStore::limited_get_log_entries: too many logs, early return: entries cnt: {}, total size: {}, res: [{}, {}]",
                         res.len(),
                         total_size,
-                        res.first().map(|x| x.get_log_id()).unwrap(),
-                        res.last().map(|x| x.get_log_id()).unwrap(),
+                        res.first().map(|x| x.log_id()).unwrap(),
+                        res.last().map(|x| x.log_id()).unwrap(),
                     );
 
                     return Ok(res);

src/meta/service/src/store/raft_state_machine_impl.rs

@@ -74,19 +74,19 @@ impl RaftStateMachine<TypeConfig> for RaftStore {
 
     // This method is not used
     #[fastrace::trace]
-    async fn begin_receiving_snapshot(&mut self) -> Result<Box<DB>, StorageError> {
+    async fn begin_receiving_snapshot(&mut self) -> Result<DB, StorageError> {
         let ss_store = SnapshotStoreV004::new(self.inner.config.clone());
         let db = ss_store
             .new_temp()
             .map_err(|e| StorageError::write_snapshot(None, &e))?;
-        Ok(Box::new(db))
+        Ok(db)
     }
 
     #[fastrace::trace]
     async fn install_snapshot(
         &mut self,
         meta: &SnapshotMeta,
-        snapshot: Box<DB>,
+        snapshot: DB,
     ) -> Result<(), StorageError> {
         let data_size = snapshot.file_size();
 
@@ -131,7 +131,7 @@ impl RaftStateMachine<TypeConfig> for RaftStore {
 
         let snapshot = db.map(|x| Snapshot {
             meta: x.snapshot_meta().clone(),
-            snapshot: Box::new(x),
+            snapshot: x,
         });
 
         info!(