Feature: new RaftNetwork API with argument RCPOption

- `RaftNetwork` introduced 3 new API `append_entries`,
  `install_snapshot` and `vote` which accept an additional argument
  `RPCOption`, and deprecated the old API `send_append_entries`,
  `send_install_snapshot` and `send_vote`.

- The old API will be **removed** in `0.9`. An application can still
  implement the old API without any changes. Openraft calls only the new
  API and the default implementation will delegate to the old API.

- Implementing the new APIs will disable the old APIs.

- The new APIs accepts an additional argument `RPCOption`, to enable an
  application control the networking behaviors based on the parameters
  in `RPCOption`.

  The `hard_ttl()` and `soft_ttl()` in `RPCOption` sets the hard limit
  and the moderate limit of the duration for which an RPC should run.
  Once the `soft_ttl()` ends, the RPC implementation should start to
  gracefully cancel the RPC, and once the `hard_ttl()` ends, Openraft
  will terminate the ongoing RPC at once.

- Fix: databendlabs#819

Author: drmingdrmer

cluster_benchmark/tests/benchmark/bench_cluster.rs

@@ -112,7 +112,8 @@ async fn do_bench(bench_config: &BenchConfig) -> anyhow::Result<()> {
                 l.client_write(ClientRequest {})
                     .await
                     .map_err(|e| {
-                        tracing::error!("client_write error: {:?}", e);
+                        eprintln!("client_write error: {:?}", e);
+                        e
                     })
                     .unwrap();
             }

cluster_benchmark/tests/benchmark/network.rs

@@ -12,6 +12,7 @@ use openraft::error::InstallSnapshotError;
 use openraft::error::RPCError;
 use openraft::error::RaftError;
 use openraft::error::RemoteError;
+use openraft::network::RPCOption;
 use openraft::network::RaftNetwork;
 use openraft::network::RaftNetworkFactory;
 use openraft::raft::AppendEntriesRequest;
@@ -98,25 +99,28 @@ pub struct Network {
 
 #[async_trait]
 impl RaftNetwork<MemConfig> for Network {
-    async fn send_append_entries(
+    async fn append_entries(
         &mut self,
         rpc: AppendEntriesRequest<MemConfig>,
+        _option: RPCOption,
     ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, (), RaftError<NodeId>>> {
         let resp = self.target_raft.append_entries(rpc).await.map_err(|e| RemoteError::new(self.target, e))?;
         Ok(resp)
     }
 
-    async fn send_install_snapshot(
+    async fn install_snapshot(
         &mut self,
         rpc: InstallSnapshotRequest<MemConfig>,
+        _option: RPCOption,
     ) -> Result<InstallSnapshotResponse<NodeId>, RPCError<NodeId, (), RaftError<NodeId, InstallSnapshotError>>> {
         let resp = self.target_raft.install_snapshot(rpc).await.map_err(|e| RemoteError::new(self.target, e))?;
         Ok(resp)
     }
 
-    async fn send_vote(
+    async fn vote(
         &mut self,
         rpc: VoteRequest<NodeId>,
+        _option: RPCOption,
     ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, (), RaftError<NodeId>>> {
         let resp = self.target_raft.vote(rpc).await.map_err(|e| RemoteError::new(self.target, e))?;
         Ok(resp)

openraft/src/core/raft_core.rs

@@ -51,6 +51,7 @@ use crate::log_id::LogIdOptionExt;
 use crate::log_id::RaftLogId;
 use crate::metrics::RaftMetrics;
 use crate::metrics::ReplicationMetrics;
+use crate::network::RPCOption;
 use crate::network::RPCTypes;
 use crate::network::RaftNetwork;
 use crate::network::RaftNetworkFactory;
@@ -307,8 +308,10 @@ where
             let target_node = eff_mem.get_node(&target).unwrap().clone();
             let mut client = self.network.new_client(target, &target_node).await;
 
+            let option = RPCOption::new(ttl);
+
             let fu = async move {
-                let outer_res = timeout(ttl, client.send_append_entries(rpc)).await;
+                let outer_res = timeout(ttl, client.append_entries(rpc, option)).await;
                 match outer_res {
                     Ok(append_res) => match append_res {
                         Ok(x) => Ok((target, x)),
@@ -991,10 +994,11 @@ where
 
             let ttl = Duration::from_millis(self.config.election_timeout_min);
             let id = self.id;
+            let option = RPCOption::new(ttl);
 
             let _ = tokio::spawn(
                 async move {
-                    let tm_res = timeout(ttl, client.send_vote(req)).await;
+                    let tm_res = timeout(ttl, client.vote(req, option)).await;
                     let res = match tm_res {
                         Ok(res) => res,
 

openraft/src/network/mod.rs

@@ -3,9 +3,11 @@
 mod backoff;
 mod factory;
 #[allow(clippy::module_inception)] mod network;
+mod rpc_option;
 mod rpc_type;
 
 pub use backoff::Backoff;
 pub use factory::RaftNetworkFactory;
 pub use network::RaftNetwork;
+pub use rpc_option::RPCOption;
 pub use rpc_type::RPCTypes;

openraft/src/network/network.rs

@@ -5,6 +5,7 @@ use async_trait::async_trait;
 use crate::error::InstallSnapshotError;
 use crate::error::RPCError;
 use crate::error::RaftError;
+use crate::network::rpc_option::RPCOption;
 use crate::network::Backoff;
 use crate::raft::AppendEntriesRequest;
 use crate::raft::AppendEntriesResponse;
@@ -19,35 +20,92 @@ use crate::RaftTypeConfig;
 /// See the [network chapter of the guide](https://datafuselabs.github.io/openraft/getting-started.html#3-impl-raftnetwork)
 /// for details and discussion on this trait and how to implement it.
 ///
-/// Typically, the network implementation as such will be hidden behind a `Box<T>` or `Arc<T>` and
-/// this interface implemented on the `Box<T>` or `Arc<T>`.
-///
 /// A single network instance is used to connect to a single target node. The network instance is
 /// constructed by the [`RaftNetworkFactory`](`crate::network::RaftNetworkFactory`).
+///
+/// ### 2023-05-03: New API with options
+///
+/// - This trait introduced 3 new API `append_entries`, `install_snapshot` and `vote` which accept
+///   an additional argument [`RPCOption`], and deprecated the old API `send_append_entries`,
+///   `send_install_snapshot` and `send_vote`.
+///
+/// - The old API will be **removed** in `0.9`. An application can still implement the old API
+///   without any changes. Openraft calls only the new API and the default implementation will
+///   delegate to the old API.
+///
+/// - Implementing the new APIs will disable the old APIs.
 #[async_trait]
 pub trait RaftNetwork<C>: Send + Sync + 'static
 where C: RaftTypeConfig
 {
+    /// Send an AppendEntries RPC to the target.
+    async fn append_entries(
+        &mut self,
+        rpc: AppendEntriesRequest<C>,
+        option: RPCOption,
+    ) -> Result<AppendEntriesResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>> {
+        let _ = option;
+        #[allow(deprecated)]
+        self.send_append_entries(rpc).await
+    }
+
+    /// Send an InstallSnapshot RPC to the target.
+    async fn install_snapshot(
+        &mut self,
+        rpc: InstallSnapshotRequest<C>,
+        option: RPCOption,
+    ) -> Result<
+        InstallSnapshotResponse<C::NodeId>,
+        RPCError<C::NodeId, C::Node, RaftError<C::NodeId, InstallSnapshotError>>,
+    > {
+        let _ = option;
+        #[allow(deprecated)]
+        self.send_install_snapshot(rpc).await
+    }
+
+    /// Send a RequestVote RPC to the target.
+    async fn vote(
+        &mut self,
+        rpc: VoteRequest<C::NodeId>,
+        option: RPCOption,
+    ) -> Result<VoteResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>> {
+        let _ = option;
+        #[allow(deprecated)]
+        self.send_vote(rpc).await
+    }
+
     /// Send an AppendEntries RPC to the target Raft node (§5).
+    #[deprecated(note = "use `append_entries` instead. This method will be removed in 0.9")]
     async fn send_append_entries(
         &mut self,
         rpc: AppendEntriesRequest<C>,
-    ) -> Result<AppendEntriesResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>>;
+    ) -> Result<AppendEntriesResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>> {
+        let _ = rpc;
+        unimplemented!("send_append_entries is deprecated")
+    }
 
     /// Send an InstallSnapshot RPC to the target Raft node (§7).
+    #[deprecated(note = "use `install_snapshot` instead. This method will be removed in 0.9")]
     async fn send_install_snapshot(
         &mut self,
         rpc: InstallSnapshotRequest<C>,
     ) -> Result<
         InstallSnapshotResponse<C::NodeId>,
         RPCError<C::NodeId, C::Node, RaftError<C::NodeId, InstallSnapshotError>>,
-    >;
+    > {
+        let _ = rpc;
+        unimplemented!("send_install_snapshot is deprecated")
+    }
 
     /// Send a RequestVote RPC to the target Raft node (§5).
+    #[deprecated(note = "use `vote` instead. This method will be removed in 0.9")]
     async fn send_vote(
         &mut self,
         rpc: VoteRequest<C::NodeId>,
-    ) -> Result<VoteResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>>;
+    ) -> Result<VoteResponse<C::NodeId>, RPCError<C::NodeId, C::Node, RaftError<C::NodeId>>> {
+        let _ = rpc;
+        unimplemented!("send_vote is deprecated")
+    }
 
     /// Build a backoff instance if the target node is temporarily(or permanently) unreachable.
     ///

openraft/src/network/rpc_option.rs

@@ -0,0 +1,42 @@
+use std::time::Duration;
+
+/// An additional argument to the [`RaftNetwork`] methods to allow applications to customize
+/// networking behaviors.
+///
+/// [`RaftNetwork`]: `crate::network::RaftNetwork`
+pub struct RPCOption {
+    /// The expected time-to-last for an RPC.
+    ///
+    /// The caller will cancel an RPC if it takes longer than this duration.
+    hard_ttl: Duration,
+}
+
+impl RPCOption {
+    pub fn new(hard_ttl: Duration) -> Self {
+        Self { hard_ttl }
+    }
+
+    /// The moderate max interval an RPC should last for.
+    ///
+    /// The [`hard_ttl()`] and `soft_ttl()` of `RPCOption` sets the hard limit and the moderate
+    /// limit of the duration for which an RPC should run. Once the `soft_ttl()` ends, the RPC
+    /// implementation should start to gracefully cancel the RPC, and once the `hard_ttl()` ends,
+    /// Openraft will terminate the ongoing RPC at once.
+    ///
+    /// `soft_ttl` is smaller than [`hard_ttl()`] so that the RPC implementation can cancel the RPC
+    /// gracefully after `soft_ttl` and before `hard_ttl`.
+    ///
+    /// `soft_ttl` is 3/4 of `hard_ttl` but it may change in future, do not rely on this ratio.
+    ///
+    /// [`hard_ttl()`]: `Self::hard_ttl`
+    pub fn soft_ttl(&self) -> Duration {
+        self.hard_ttl * 3 / 4
+    }
+
+    /// The hard limit of the interval an RPC should last for.
+    ///
+    /// When exceeding this limit, the RPC will be dropped by Openraft at once.
+    pub fn hard_ttl(&self) -> Duration {
+        self.hard_ttl
+    }
+}

openraft/src/replication/mod.rs

@@ -33,6 +33,7 @@ use crate::error::Timeout;
 use crate::log_id::LogIdOptionExt;
 use crate::log_id_range::LogIdRange;
 use crate::network::Backoff;
+use crate::network::RPCOption;
 use crate::network::RPCTypes;
 use crate::network::RaftNetwork;
 use crate::network::RaftNetworkFactory;
@@ -292,7 +293,8 @@ where
         );
 
         let the_timeout = Duration::from_millis(self.config.heartbeat_interval);
-        let res = timeout(the_timeout, self.network.send_append_entries(payload)).await;
+        let option = RPCOption::new(the_timeout);
+        let res = timeout(the_timeout, self.network.append_entries(payload, option)).await;
 
         tracing::debug!("append_entries res: {:?}", res);
 
@@ -576,7 +578,9 @@ where
                 self.config.send_snapshot_timeout()
             };
 
-            let res = timeout(snap_timeout, self.network.send_install_snapshot(req)).await;
+            let option = RPCOption::new(snap_timeout);
+
+            let res = timeout(snap_timeout, self.network.install_snapshot(req, option)).await;
 
             let res = match res {
                 Ok(outer_res) => match outer_res {

tests/tests/append_entries/t10_conflict_with_empty_entries.rs

@@ -1,6 +1,8 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::Result;
+use openraft::network::RPCOption;
 use openraft::network::RaftNetwork;
 use openraft::network::RaftNetworkFactory;
 use openraft::raft::AppendEntriesRequest;
@@ -57,7 +59,8 @@ async fn conflict_with_empty_entries() -> Result<()> {
         leader_commit: Some(LogId::new(CommittedLeaderId::new(1, 0), 5)),
     };
 
-    let resp = router.new_client(0, &()).await.send_append_entries(rpc).await?;
+    let option = RPCOption::new(Duration::from_millis(1_000));
+    let resp = router.new_client(0, &()).await.append_entries(rpc, option).await?;
     assert!(!resp.is_success());
     assert!(resp.is_conflict());
 
@@ -77,7 +80,9 @@ async fn conflict_with_empty_entries() -> Result<()> {
         leader_commit: Some(LogId::new(CommittedLeaderId::new(1, 0), 5)),
     };
 
-    let resp = router.new_client(0, &()).await.send_append_entries(rpc).await?;
+    let option = RPCOption::new(Duration::from_millis(1_000));
+
+    let resp = router.new_client(0, &()).await.append_entries(rpc, option).await?;
     assert!(resp.is_success());
     assert!(!resp.is_conflict());
 
@@ -90,7 +95,9 @@ async fn conflict_with_empty_entries() -> Result<()> {
         leader_commit: Some(LogId::new(CommittedLeaderId::new(1, 0), 5)),
     };
 
-    let resp = router.new_client(0, &()).await.send_append_entries(rpc).await?;
+    let option = RPCOption::new(Duration::from_millis(1_000));
+
+    let resp = router.new_client(0, &()).await.append_entries(rpc, option).await?;
     assert!(!resp.is_success());
     assert!(resp.is_conflict());
 

tests/tests/append_entries/t10_see_higher_vote.rs

@@ -3,6 +3,7 @@ use std::time::Duration;
 
 use anyhow::Result;
 use maplit::btreeset;
+use openraft::network::RPCOption;
 use openraft::network::RaftNetwork;
 use openraft::network::RaftNetworkFactory;
 use openraft::raft::VoteRequest;
@@ -40,13 +41,18 @@ async fn append_sees_higher_vote() -> Result<()> {
         // Let leader lease expire
         sleep(Duration::from_millis(800)).await;
 
+        let option = RPCOption::new(Duration::from_millis(1_000));
+
         let resp = router
             .new_client(1, &())
             .await
-            .send_vote(VoteRequest {
-                vote: Vote::new(10, 1),
-                last_log_id: Some(LogId::new(CommittedLeaderId::new(10, 1), 5)),
-            })
+            .vote(
+                VoteRequest {
+                    vote: Vote::new(10, 1),
+                    last_log_id: Some(LogId::new(CommittedLeaderId::new(10, 1), 5)),
+                },
+                option,
+            )
             .await?;
 
         assert!(resp.vote_granted);

tests/tests/append_entries/t11_append_entries_with_bigger_term.rs

@@ -1,7 +1,9 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::Result;
 use maplit::btreeset;
+use openraft::network::RPCOption;
 use openraft::network::RaftNetwork;
 use openraft::network::RaftNetworkFactory;
 use openraft::raft::AppendEntriesRequest;
@@ -51,7 +53,9 @@ async fn append_entries_with_bigger_term() -> Result<()> {
         leader_commit: Some(LogId::new(CommittedLeaderId::new(1, 0), log_index)),
     };
 
-    let resp = router.new_client(0, &()).await.send_append_entries(req).await?;
+    let option = RPCOption::new(Duration::from_millis(1_000));
+
+    let resp = router.new_client(0, &()).await.append_entries(req, option).await?;
     assert!(resp.is_success());
 
     // after append entries, check hard state in term 2 and vote for node 1