databendlabs · drmingdrmer · Aug 12, 2024 · Aug 10, 2024
diff --git a/Cargo.toml b/Cargo.toml
@@ -3,7 +3,7 @@ version = "0.10.0"
 edition = "2021"
 authors = [
  "Databend Authors <[email protected]>",
- "Anthony Dodd <[email protected]>"
+ "Anthony Dodd <[email protected]>",
 ]
 categories = ["algorithms", "asynchronous", "data-structures"]
 description = "Advanced Raft consensus"
@@ -18,16 +18,16 @@ repository = "https://github.com/datafuselabs/openraft"
 anyerror = { version = "0.1.10" }
 anyhow = "1.0.63"
 async-entry = "0.3.1"
-byte-unit = "4.0.12"
+byte-unit = "5.1.4"
 bytes = "1.0"
 chrono = { version = "0.4" }
 clap = { version = "4.1.11", features = ["derive", "env"] }
-derive_more = { version = "0.99.9" }
+derive_more = { version = "1.0", features = ["std", "from", "try_into", "display"] }
 futures = "0.3"
 lazy_static = "1.4.0"
 maplit = "1.0.2"
 pretty_assertions = "1.0.0"
-proc-macro2 = { version = ">=1.0.0,<1.0.80", features = [] }
+proc-macro2 = "1.0"
 quote = "1.0"
 rand = "0.8"
 semver = "1.0.14"
@@ -37,7 +37,14 @@ syn = "2.0"
 tempfile = { version = "3.4.0" }
 test-harness = "0.3.0"
 thiserror = "1.0.49"
-tokio = { version = "1.22", default-features = false, features = ["io-util", "macros", "rt", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1.22", default-features = false, features = [
+ "io-util",
+ "macros",
+ "rt",
+ "rt-multi-thread",
+ "sync",
+ "time",
+] }
 tracing = { version = "0.1.40" }
 tracing-appender = "0.2.0"
 tracing-futures = "0.2.4"
@@ -64,5 +71,5 @@ exclude = [
  "examples/raft-kv-memstore-network-v2",
  "examples/raft-kv-memstore-opendal-snapshot-data",
  "examples/raft-kv-rocksdb",
- "rt-monoio"
+ "rt-monoio",
 ]
diff --git a/examples/raft-kv-memstore-network-v2/Cargo.toml b/examples/raft-kv-memstore-network-v2/Cargo.toml
@@ -19,8 +19,6 @@ repository = "https://github.com/datafuselabs/openraft"
 memstore = { path = "../memstore", features = [] }
 openraft = { path = "../../openraft", features = ["serde", "type-alias"] }
 
-clap = { version = "4.1.11", features = ["derive", "env"] }
-reqwest = { version = "0.11.9", features = ["json"] }
 serde = { version = "1.0.114", features = ["derive"] }
 serde_json = "1.0.57"
 tokio = { version = "1.0", default-features = false, features = ["sync"] }

diff --git a/examples/raft-kv-memstore-opendal-snapshot-data/Cargo.toml b/examples/raft-kv-memstore-opendal-snapshot-data/Cargo.toml
@@ -7,7 +7,7 @@ edition = "2021"
 authors = [
  "drdr xp <[email protected]>",
  "Pedro Paulo de Amorim <[email protected]>",
- "Xuanwo <[email protected]>"
+ "Xuanwo <[email protected]>",
 ]
 categories = ["algorithms", "asynchronous", "data-structures"]
 description = "An example distributed key-value store built upon `openraft`."
@@ -20,12 +20,13 @@ repository = "https://github.com/datafuselabs/openraft"
 memstore = { path = "../memstore", features = [] }
 openraft = { path = "../../openraft", features = ["serde", "type-alias"] }
 
+bytes = "1.0"
 serde = { version = "1.0.114", features = ["derive"] }
 serde_json = "1.0.57"
 tokio = { version = "1.0", default-features = false, features = ["sync"] }
 tracing = "0.1.29"
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
-opendal = "0.45.0"
+opendal = "0.48.0"
 
 [dev-dependencies]
 maplit = "1.0.2"

diff --git a/examples/raft-kv-memstore-opendal-snapshot-data/src/lib.rs b/examples/raft-kv-memstore-opendal-snapshot-data/src/lib.rs
@@ -3,6 +3,8 @@
 
 use std::sync::Arc;
 
+use bytes::Buf;
+use opendal::Buffer;
 use opendal::Operator;
 use openraft::Config;
 
@@ -68,6 +70,10 @@ pub fn decode<T: serde::de::DeserializeOwned>(s: &str) -> T {
  serde_json::from_str(s).unwrap()
 }
 
+pub fn decode_buffer<T: serde::de::DeserializeOwned>(b: Buffer) -> T {
+ serde_json::from_reader(b.reader()).unwrap()
+}
+
 pub async fn new_raft(node_id: NodeId, router: Router, op: Operator) -> (typ::Raft, App) {
  // Create a configuration for the raft instance.
  let config = Config {

diff --git a/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs b/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs
@@ -17,7 +17,7 @@ use openraft::StoredMembership;
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::decode;
+use crate::decode_buffer;
 use crate::encode;
 use crate::typ;
 use crate::NodeId;
@@ -209,7 +209,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
  // Update the state machine.
  {
  let bs = self.storage.read(&new_snapshot.data).await.unwrap();
- let updated_state_machine: StateMachineData = decode(&String::from_utf8_lossy(&bs));
+ let updated_state_machine: StateMachineData = decode_buffer(bs);
  let mut state_machine = self.state_machine.lock().unwrap();
  *state_machine = updated_state_machine;
  }

diff --git a/examples/raft-kv-memstore-opendal-snapshot-data/tests/cluster/test_cluster.rs b/examples/raft-kv-memstore-opendal-snapshot-data/tests/cluster/test_cluster.rs
@@ -59,7 +59,7 @@ async fn test_cluster() {
 
  // This test only use memory service for simplicity.
  // Feel free to test against fs or s3.
- let op = opendal::Operator::via_map(opendal::Scheme::Memory, HashMap::default()).unwrap();
+ let op = opendal::Operator::via_iter(opendal::Scheme::Memory, HashMap::<String, String>::default()).unwrap();
 
  let router = Router::default();
 

diff --git a/examples/raft-kv-memstore-singlethreaded/Cargo.toml b/examples/raft-kv-memstore-singlethreaded/Cargo.toml
@@ -16,10 +16,12 @@ license = "MIT OR Apache-2.0"
 repository = "https://github.com/datafuselabs/openraft"
 
 [dependencies]
-openraft = { path = "../../openraft", features = ["serde", "singlethreaded", "type-alias"] }
+openraft = { path = "../../openraft", features = [
+ "serde",
+ "singlethreaded",
+ "type-alias",
+] }
 
-clap = { version = "4.1.11", features = ["derive", "env"] }
-reqwest = { version = "0.11.9", features = ["json"] }
 serde = { version = "1.0.114", features = ["derive"] }
 serde_json = "1.0.57"
 tokio = { version = "1.0", default-features = false, features = ["sync"] }

diff --git a/examples/raft-kv-memstore/Cargo.toml b/examples/raft-kv-memstore/Cargo.toml
@@ -25,7 +25,7 @@ openraft = { path = "../../openraft", features = ["serde", "type-alias"] }
 
 actix-web = "4.0.0-rc.2"
 clap = { version = "4.1.11", features = ["derive", "env"] }
-reqwest = { version = "0.11.9", features = ["json"] }
+reqwest = { version = "0.12.5", features = ["json"] }
 serde = { version = "1.0.114", features = ["derive"] }
 serde_json = "1.0.57"
 tokio = { version = "1.0", default-features = false, features = ["sync"] }

diff --git a/examples/raft-kv-rocksdb/Cargo.toml b/examples/raft-kv-rocksdb/Cargo.toml
@@ -5,9 +5,9 @@ readme = "README.md"
 
 edition = "2021"
 authors = [
-  "drdr xp <[email protected]>",
-  "Pedro Paulo de Amorim <[email protected]>",
-  "The Tremor Team",
+ "drdr xp <[email protected]>",
+ "Pedro Paulo de Amorim <[email protected]>",
+ "The Tremor Team",
 ]
 categories = ["algorithms", "asynchronous", "data-structures"]
 description = "An example distributed key-value store built upon `openraft`."
@@ -26,14 +26,14 @@ openraft = { path = "../../openraft", features = ["serde", "type-alias"] }
 tokio = { version = "1.35.1", features = ["full"] }
 byteorder = "1.4.3"
 clap = { version = "4.1.11", features = ["derive", "env"] }
-reqwest = { version = "0.11.9", features = ["json"] }
+reqwest = { version = "0.12.5", features = ["json"] }
 rocksdb = "0.22.0"
 serde = { version = "1.0.114", features = ["derive"] }
 serde_json = "1.0.57"
 tide = { version = "0.16" }
 # for toy-rpc, use `serde_json` instead of the default `serde_bincode`:
 # bincode which enabled by default by toy-rpc, does not support `#[serde(flatten)]`: https://docs.rs/bincode/2.0.0-alpha.1/bincode/serde/index.html#known-issues
-toy-rpc = { version = "0.8.6", features = [
+toy-rpc = { version = "0.10.0", features = [
  "ws_tokio",
  "server",
  "client",

diff --git a/openraft/Cargo.toml b/openraft/Cargo.toml
@@ -27,15 +27,12 @@ maplit = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true, optional = true }
 serde_json = { workspace = true, optional = true }
-tempfile = { workspace = true, optional = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, optional = true }
 tracing = { workspace = true }
 tracing-futures = { workspace = true }
 validit = { workspace = true }
 
-or07 = { package = "openraft", version = "0.7.4", optional = true }
-
 
 [dev-dependencies]
 anyhow = { workspace = true }
@@ -128,4 +125,4 @@ no-default-features = false
 #
 # Sort modules by appearance order for crate `docs`.
 # https://doc.rust-lang.org/rustdoc/unstable-features.html#--sort-modules-by-appearance-control-how-items-on-module-pages-are-sorted
-# rustdoc-args = ["-Z", "unstable-options", "--sort-modules-by-appearance"]
+# rustdoc-args = ["-Z", "unstable-options", "--sort-modules-by-appearance"]
diff --git a/openraft/src/config/config.rs b/openraft/src/config/config.rs
@@ -1,6 +1,7 @@
 //! Raft runtime configuration.
 
 use std::ops::Deref;
+use std::str::FromStr;
 use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 
@@ -54,7 +55,7 @@ fn parse_bytes_with_unit(src: &str) -> Result<u64, ConfigError> {
  reason: e.to_string(),
  })?;
 
- Ok(res.get_bytes() as u64)
+ Ok(res.as_u64())
 }
 
 fn parse_snapshot_policy(src: &str) -> Result<SnapshotPolicy, ConfigError> {

diff --git a/openraft/src/core/sm/worker.rs b/openraft/src/core/sm/worker.rs
@@ -43,7 +43,7 @@ where
  /// [`RaftLogStorage`]: `crate::storage::RaftLogStorage`
  log_reader: LR,
 
- /// Raed command from RaftCore to execute.
+ /// Read command from RaftCore to execute.
  cmd_rx: MpscUnboundedReceiverOf<C, Command<C>>,
 
  /// Send back the result of the command to RaftCore.

diff --git a/openraft/src/docs/docs.md b/openraft/src/docs/docs.md
@@ -42,4 +42,4 @@ Contributors who want to understand the internals of Openraft can find relevant
 Finally, the archived and discarded documents:
 - [`obsolete`](crate::docs::obsolete) describes obsolete design documents and why they are discarded;
  - [`blank-log-heartbeeat`](`crate::docs::obsolete::heartbeat`);
- - [`fast-commit`](`crate::docs::obsolete::fast_commit`);
+ - [`fast-commit`](`crate::docs::obsolete::fast_commit`);
diff --git a/openraft/src/docs/mod.rs b/openraft/src/docs/mod.rs
@@ -1,3 +1,4 @@
+#![allow(rustdoc::redundant_explicit_links)]
 #![doc = include_str!("docs.md")]
 
 #[rustfmt::skip]

diff --git a/openraft/src/docs/protocol/snapshot_replication.md b/openraft/src/docs/protocol/snapshot_replication.md
@@ -112,8 +112,7 @@ Installing snapshot includes two steps:
 The final step is to purge logs up to [`snapshot_meta.last_log_id`].
 This step is necessary because:
 
-- 1) A local log that is <= [`snapshot_meta.last_log_id`] may conflict with the leader, and can not be used
- anymore.
+- 1) A local log that is <= [`snapshot_meta.last_log_id`] may conflict with the leader, and can not be used anymore.
 
 - 2) There may be a hole in the logs, if `snapshot_last_log_id > local_last_log_id`:
 

diff --git a/openraft/src/raft/message/append_entries.rs b/openraft/src/raft/message/append_entries.rs
@@ -30,9 +30,7 @@ pub struct AppendEntriesRequest<C: RaftTypeConfig> {
  pub leader_commit: Option<LogId<C::NodeId>>,
 }
 
-impl<C: RaftTypeConfig> fmt::Debug for AppendEntriesRequest<C>
-where C::D: fmt::Debug
-{
+impl<C: RaftTypeConfig> fmt::Debug for AppendEntriesRequest<C> {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
  f.debug_struct("AppendEntriesRequest")
  .field("vote", &self.vote)

diff --git a/openraft/src/raft/message/install_snapshot.rs b/openraft/src/raft/message/install_snapshot.rs
@@ -41,7 +41,7 @@ impl<C: RaftTypeConfig> fmt::Display for InstallSnapshotRequest<C> {
 #[derive(Debug)]
 #[derive(PartialEq, Eq)]
 #[derive(derive_more::Display)]
-#[display(fmt = "{{vote:{}}}", vote)]
+#[display("{{vote:{}}}", vote)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize), serde(bound = ""))]
 pub struct InstallSnapshotResponse<C: RaftTypeConfig> {
  pub vote: Vote<C::NodeId>,
@@ -51,7 +51,7 @@ pub struct InstallSnapshotResponse<C: RaftTypeConfig> {
 #[derive(Debug)]
 #[derive(PartialEq, Eq)]
 #[derive(derive_more::Display)]
-#[display(fmt = "SnapshotResponse{{vote:{}}}", vote)]
+#[display("SnapshotResponse{{vote:{}}}", vote)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize), serde(bound = ""))]
 pub struct SnapshotResponse<C: RaftTypeConfig> {
  pub vote: Vote<C::NodeId>,

diff --git a/openraft/src/testing/log/suite.rs b/openraft/src/testing/log/suite.rs
@@ -48,7 +48,7 @@ macro_rules! btreeset {
  }};
 }
 
-/// Allows [`RaftLogStorage`] to access methods provided by [`RaftLogReader`] in ths test.
+/// Allows [`RaftLogStorage`] to access methods provided by [`RaftLogReader`] in this test.
 trait ReaderExt<C>: RaftLogStorage<C>
 where C: RaftTypeConfig
 {

diff --git a/rust-toolchain b/rust-toolchain
@@ -1 +1 @@
-nightly-2024-03-20
+nightly-2024-07-02
diff --git a/tests/tests/fixtures/mod.rs b/tests/tests/fixtures/mod.rs
@@ -10,7 +10,7 @@
 use std::env;
 use std::fmt;
 use std::future::Future;
 use std::panic::PanicInfo;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -142,7 +142,7 @@
 }));
 }

 pub fn log_panic(panic: &PanicInfo) {
 let backtrace = {
 #[cfg(feature = "bt")]
 {
@@ -226,16 +226,21 @@
 /// Pre-hook result, which does not return remote Error.
 pub type PreHookResult = Result<(), RPCError<MemConfig, Infallible>>;
 
+#[derive(Debug)]
 #[derive(derive_more::From, derive_more::TryInto)]
-pub enum RPCRequest<C: RaftTypeConfig> {
+pub enum RPCRequest<C: RaftTypeConfig>
+where C::SnapshotData: fmt::Debug
+{
  AppendEntries(AppendEntriesRequest<C>),
  InstallSnapshot(InstallSnapshotRequest<C>),
  InstallFullSnapshot(Snapshot<C>),
  Vote(VoteRequest<C>),
  TransferLeader(TransferLeaderRequest<C>),
 }
 
-impl<C: RaftTypeConfig> RPCRequest<C> {
+impl<C: RaftTypeConfig> RPCRequest<C>
+where C::SnapshotData: fmt::Debug
+{
  pub fn get_type(&self) -> RPCTypes {
  match self {
  RPCRequest::AppendEntries(_) => RPCTypes::AppendEntries,