Change: Use a RaftConfig trait to configure common types

Currently only the request and response type are configured, but other types can be now easily configured as well, such as `NodeId` (to be done in the next change) or `Runtime`.

Storage and network factory types are still provided as a separate type, since otherwise it would be hard to provide storage wrappers for testing.

This addresses #204.

Author: schreter

example-raft-kv/src/bin/main.rs

@@ -2,12 +2,11 @@ use clap::Parser;
 use env_logger::Env;
 use example_raft_key_value::network::raft_network_impl::ExampleNetwork;
 use example_raft_key_value::start_example_raft_node;
-use example_raft_key_value::store::ExampleRequest;
-use example_raft_key_value::store::ExampleResponse;
 use example_raft_key_value::store::ExampleStore;
+use example_raft_key_value::ExampleConfig;
 use openraft::Raft;
 
-pub type ExampleRaft = Raft<ExampleRequest, ExampleResponse, ExampleNetwork, ExampleStore>;
+pub type ExampleRaft = Raft<ExampleConfig, ExampleNetwork, ExampleStore>;
 
 #[derive(Parser, Clone, Debug)]
 #[clap(author, version, about, long_about = None)]

example-raft-kv/src/client.rs

@@ -19,8 +19,8 @@ use serde::de::DeserializeOwned;
 use serde::Deserialize;
 use serde::Serialize;
 
+use crate::ExampleConfig;
 use crate::ExampleRequest;
-use crate::ExampleResponse;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Empty {}
@@ -54,7 +54,7 @@ impl ExampleClient {
     pub async fn write(
         &self,
         req: &ExampleRequest,
-    ) -> Result<ClientWriteResponse<ExampleResponse>, RPCError<ClientWriteError>> {
+    ) -> Result<ClientWriteResponse<ExampleConfig>, RPCError<ClientWriteError>> {
         self.send_rpc_to_leader("write", Some(req)).await
     }
 
@@ -91,7 +91,7 @@ impl ExampleClient {
     pub async fn change_membership(
         &self,
         req: &BTreeSet<NodeId>,
-    ) -> Result<ClientWriteResponse<ExampleResponse>, RPCError<ClientWriteError>> {
+    ) -> Result<ClientWriteResponse<ExampleConfig>, RPCError<ClientWriteError>> {
         self.send_rpc_to_leader("change-membership", Some(req)).await
     }
 

example-raft-kv/src/lib.rs

@@ -8,6 +8,7 @@ use actix_web::HttpServer;
 use openraft::Config;
 use openraft::NodeId;
 use openraft::Raft;
+use openraft::RaftConfig;
 
 use crate::app::ExampleApp;
 use crate::network::api;
@@ -23,7 +24,14 @@ pub mod client;
 pub mod network;
 pub mod store;
 
-pub type ExampleRaft = Raft<ExampleRequest, ExampleResponse, ExampleNetwork, Arc<ExampleStore>>;
+pub struct ExampleConfig {}
+
+impl RaftConfig for ExampleConfig {
+    type D = ExampleRequest;
+    type R = ExampleResponse;
+}
+
+pub type ExampleRaft = Raft<ExampleConfig, ExampleNetwork, Arc<ExampleStore>>;
 
 pub async fn start_example_raft_node(node_id: NodeId, http_addr: String) -> std::io::Result<()> {
     // Create a configuration for the raft instance.

example-raft-kv/src/network/raft.rs

@@ -8,7 +8,7 @@ use openraft::raft::VoteRequest;
 use web::Json;
 
 use crate::app::ExampleApp;
-use crate::store::ExampleRequest;
+use crate::ExampleConfig;
 
 // --- Raft communication
 
@@ -21,7 +21,7 @@ pub async fn vote(app: Data<ExampleApp>, req: Json<VoteRequest>) -> actix_web::R
 #[post("/raft-append")]
 pub async fn append(
     app: Data<ExampleApp>,
-    req: Json<AppendEntriesRequest<ExampleRequest>>,
+    req: Json<AppendEntriesRequest<ExampleConfig>>,
 ) -> actix_web::Result<impl Responder> {
     let res = app.raft.append_entries(req.0).await;
     Ok(Json(res))

example-raft-kv/src/network/raft_network_impl.rs

@@ -18,7 +18,7 @@ use openraft::RaftNetworkFactory;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
 
-use crate::store::ExampleRequest;
+use crate::ExampleConfig;
 
 pub struct ExampleNetwork {}
 
@@ -50,7 +50,7 @@ impl ExampleNetwork {
 
 // NOTE: This could be implemented also on `Arc<ExampleNetwork>`, but since it's empty, implemented directly.
 #[async_trait]
-impl RaftNetworkFactory<ExampleRequest> for ExampleNetwork {
+impl RaftNetworkFactory<ExampleConfig> for ExampleNetwork {
     type Network = ExampleNetworkConnection;
 
     async fn connect(&mut self, target: NodeId, node: Option<&Node>) -> Self::Network {
@@ -69,10 +69,10 @@ pub struct ExampleNetworkConnection {
 }
 
 #[async_trait]
-impl RaftNetwork<ExampleRequest> for ExampleNetworkConnection {
+impl RaftNetwork<ExampleConfig> for ExampleNetworkConnection {
     async fn send_append_entries(
         &mut self,
-        req: AppendEntriesRequest<ExampleRequest>,
+        req: AppendEntriesRequest<ExampleConfig>,
     ) -> Result<AppendEntriesResponse, RPCError<AppendEntriesError>> {
         self.owner.send_rpc(self.target, self.target_node.as_ref(), "raft-append", req).await
     }

example-raft-kv/src/store/mod.rs

@@ -29,6 +29,8 @@ use serde::Deserialize;
 use serde::Serialize;
 use tokio::sync::RwLock;
 
+use crate::ExampleConfig;
+
 /**
  * Here you will set the types of request that will interact with the raft nodes.
  * For example the `Set` will be used to write data (key and value) to the raft database.
@@ -87,7 +89,7 @@ pub struct ExampleStore {
     last_purged_log_id: RwLock<Option<LogId>>,
 
     /// The Raft log.
-    log: RwLock<BTreeMap<u64, Entry<ExampleRequest>>>,
+    log: RwLock<BTreeMap<u64, Entry<ExampleConfig>>>,
 
     /// The Raft state machine.
     pub state_machine: RwLock<ExampleStateMachine>,
@@ -101,7 +103,7 @@ pub struct ExampleStore {
 }
 
 #[async_trait]
-impl RaftLogReader<ExampleRequest, ExampleResponse> for Arc<ExampleStore> {
+impl RaftLogReader<ExampleConfig> for Arc<ExampleStore> {
     async fn get_log_state(&mut self) -> Result<LogState, StorageError> {
         let log = self.log.read().await;
         let last = log.iter().rev().next().map(|(_, ent)| ent.log_id);
@@ -122,15 +124,15 @@ impl RaftLogReader<ExampleRequest, ExampleResponse> for Arc<ExampleStore> {
     async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug + Send + Sync>(
         &mut self,
         range: RB,
-    ) -> Result<Vec<Entry<ExampleRequest>>, StorageError> {
+    ) -> Result<Vec<Entry<ExampleConfig>>, StorageError> {
         let log = self.log.read().await;
         let response = log.range(range.clone()).map(|(_, val)| val.clone()).collect::<Vec<_>>();
         Ok(response)
     }
 }
 
 #[async_trait]
-impl RaftSnapshotBuilder<ExampleRequest, ExampleResponse, Cursor<Vec<u8>>> for Arc<ExampleStore> {
+impl RaftSnapshotBuilder<ExampleConfig, Cursor<Vec<u8>>> for Arc<ExampleStore> {
     #[tracing::instrument(level = "trace", skip(self))]
     async fn build_snapshot(&mut self) -> Result<Snapshot<Cursor<Vec<u8>>>, StorageError> {
         let (data, last_applied_log);
@@ -185,7 +187,7 @@ impl RaftSnapshotBuilder<ExampleRequest, ExampleResponse, Cursor<Vec<u8>>> for A
 }
 
 #[async_trait]
-impl RaftStorage<ExampleRequest, ExampleResponse> for Arc<ExampleStore> {
+impl RaftStorage<ExampleConfig> for Arc<ExampleStore> {
     type SnapshotData = Cursor<Vec<u8>>;
     type LogReader = Self;
     type SnapshotBuilder = Self;
@@ -203,7 +205,7 @@ impl RaftStorage<ExampleRequest, ExampleResponse> for Arc<ExampleStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self, entries))]
-    async fn append_to_log(&mut self, entries: &[&Entry<ExampleRequest>]) -> Result<(), StorageError> {
+    async fn append_to_log(&mut self, entries: &[&Entry<ExampleConfig>]) -> Result<(), StorageError> {
         let mut log = self.log.write().await;
         for entry in entries {
             log.insert(entry.log_id.index, (*entry).clone());
@@ -254,7 +256,7 @@ impl RaftStorage<ExampleRequest, ExampleResponse> for Arc<ExampleStore> {
     #[tracing::instrument(level = "trace", skip(self, entries))]
     async fn apply_to_state_machine(
         &mut self,
-        entries: &[&Entry<ExampleRequest>],
+        entries: &[&Entry<ExampleConfig>],
     ) -> Result<Vec<ExampleResponse>, StorageError> {
         let mut res = Vec::with_capacity(entries.len());
 

memstore/src/lib.rs

@@ -25,6 +25,7 @@ use openraft::EffectiveMembership;
 use openraft::ErrorSubject;
 use openraft::ErrorVerb;
 use openraft::LogId;
+use openraft::RaftConfig;
 use openraft::RaftStorage;
 use openraft::RaftStorageDebug;
 use openraft::SnapshotMeta;
@@ -62,6 +63,13 @@ pub struct ClientResponse(Option<String>);
 
 impl AppDataResponse for ClientResponse {}
 
+pub struct Config {}
+
+impl RaftConfig for Config {
+    type D = ClientRequest;
+    type R = ClientResponse;
+}
+
 /// The application snapshot type which the `MemStore` works with.
 #[derive(Debug)]
 pub struct MemStoreSnapshot {
@@ -89,7 +97,7 @@ pub struct MemStore {
     last_purged_log_id: RwLock<Option<LogId>>,
 
     /// The Raft log.
-    log: RwLock<BTreeMap<u64, Entry<ClientRequest>>>,
+    log: RwLock<BTreeMap<u64, Entry<Config>>>,
 
     /// The Raft state machine.
     sm: RwLock<MemStoreStateMachine>,
@@ -134,11 +142,11 @@ impl RaftStorageDebug<MemStoreStateMachine> for Arc<MemStore> {
 }
 
 #[async_trait]
-impl RaftLogReader<ClientRequest, ClientResponse> for Arc<MemStore> {
+impl RaftLogReader<Config> for Arc<MemStore> {
     async fn try_get_log_entries<RB: RangeBounds<u64> + Clone + Debug + Send + Sync>(
         &mut self,
         range: RB,
-    ) -> Result<Vec<Entry<ClientRequest>>, StorageError> {
+    ) -> Result<Vec<Entry<Config>>, StorageError> {
         let res = {
             let log = self.log.read().await;
             log.range(range.clone()).map(|(_, val)| val.clone()).collect::<Vec<_>>()
@@ -166,7 +174,7 @@ impl RaftLogReader<ClientRequest, ClientResponse> for Arc<MemStore> {
 }
 
 #[async_trait]
-impl RaftSnapshotBuilder<ClientRequest, ClientResponse, Cursor<Vec<u8>>> for Arc<MemStore> {
+impl RaftSnapshotBuilder<Config, Cursor<Vec<u8>>> for Arc<MemStore> {
     #[tracing::instrument(level = "trace", skip(self))]
     async fn build_snapshot(&mut self) -> Result<Snapshot<Cursor<Vec<u8>>>, StorageError> {
         let (data, last_applied_log);
@@ -225,7 +233,7 @@ impl RaftSnapshotBuilder<ClientRequest, ClientResponse, Cursor<Vec<u8>>> for Arc
 }
 
 #[async_trait]
-impl RaftStorage<ClientRequest, ClientResponse> for Arc<MemStore> {
+impl RaftStorage<Config> for Arc<MemStore> {
     type SnapshotData = Cursor<Vec<u8>>;
 
     #[tracing::instrument(level = "trace", skip(self))]
@@ -285,7 +293,7 @@ impl RaftStorage<ClientRequest, ClientResponse> for Arc<MemStore> {
     }
 
     #[tracing::instrument(level = "trace", skip(self, entries))]
-    async fn append_to_log(&mut self, entries: &[&Entry<ClientRequest>]) -> Result<(), StorageError> {
+    async fn append_to_log(&mut self, entries: &[&Entry<Config>]) -> Result<(), StorageError> {
         let mut log = self.log.write().await;
         for entry in entries {
             log.insert(entry.log_id.index, (*entry).clone());
@@ -296,7 +304,7 @@ impl RaftStorage<ClientRequest, ClientResponse> for Arc<MemStore> {
     #[tracing::instrument(level = "trace", skip(self, entries))]
     async fn apply_to_state_machine(
         &mut self,
-        entries: &[&Entry<ClientRequest>],
+        entries: &[&Entry<Config>],
     ) -> Result<Vec<ClientResponse>, StorageError> {
         let mut res = Vec::with_capacity(entries.len());
 

openraft/src/core/admin.rs

@@ -23,17 +23,16 @@ use crate::raft::ClientWriteResponse;
 use crate::raft::EntryPayload;
 use crate::raft::RaftRespTx;
 use crate::raft_types::LogIdOptionExt;
-use crate::AppData;
-use crate::AppDataResponse;
 use crate::LogId;
 use crate::Membership;
 use crate::Node;
 use crate::NodeId;
+use crate::RaftConfig;
 use crate::RaftNetworkFactory;
 use crate::RaftStorage;
 use crate::StorageError;
 
-impl<'a, D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D, R>> LearnerState<'a, D, R, N, S> {
+impl<'a, C: RaftConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LearnerState<'a, C, N, S> {
     /// Handle the admin `init_with_config` command.
     #[tracing::instrument(level = "debug", skip(self))]
     pub(super) async fn handle_init_with_config(&mut self, members: EitherNodesOrIds) -> Result<(), InitializeError> {
@@ -67,7 +66,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorag
     }
 }
 
-impl<'a, D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D, R>> LeaderState<'a, D, R, N, S> {
+impl<'a, C: RaftConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> LeaderState<'a, C, N, S> {
     // add node into learner,return true if the node is already a member or learner
     #[tracing::instrument(level = "debug", skip(self))]
     async fn add_learner_into_membership(
@@ -162,7 +161,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorag
         members: BTreeSet<NodeId>,
         blocking: bool,
         turn_to_learner: bool,
-        tx: RaftRespTx<ClientWriteResponse<R>, ClientWriteError>,
+        tx: RaftRespTx<ClientWriteResponse<C>, ClientWriteError>,
     ) -> Result<(), StorageError> {
         // Ensure cluster will have at least one node.
         if members.is_empty() {
@@ -251,7 +250,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorag
     pub async fn append_membership_log(
         &mut self,
         mem: Membership,
-        resp_tx: Option<RaftRespTx<ClientWriteResponse<R>, ClientWriteError>>,
+        resp_tx: Option<RaftRespTx<ClientWriteResponse<C>, ClientWriteError>>,
     ) -> Result<(), StorageError> {
         let payload = EntryPayload::Membership(mem.clone());
         let entry = self.core.append_payload_to_log(payload).await?;

openraft/src/core/append_entries.rs

@@ -7,24 +7,23 @@ use crate::raft::AppendEntriesResponse;
 use crate::raft::Entry;
 use crate::raft::EntryPayload;
 use crate::raft_types::LogIdOptionExt;
-use crate::AppData;
-use crate::AppDataResponse;
 use crate::EffectiveMembership;
 use crate::LogId;
 use crate::MessageSummary;
+use crate::RaftConfig;
 use crate::RaftNetworkFactory;
 use crate::RaftStorage;
 use crate::StorageError;
 use crate::Update;
 
-impl<D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D, R>> RaftCore<D, R, N, S> {
+impl<C: RaftConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C, N, S> {
     /// An RPC invoked by the leader to replicate log entries (§5.3); also used as heartbeat (§5.2).
     ///
     /// See `receiver implementation: AppendEntries RPC` in raft-essentials.md in this repo.
     #[tracing::instrument(level = "debug", skip(self, req))]
     pub(super) async fn handle_append_entries_request(
         &mut self,
-        req: AppendEntriesRequest<D>,
+        req: AppendEntriesRequest<C>,
     ) -> Result<AppendEntriesResponse, AppendEntriesError> {
         tracing::debug!(last_log_id=?self.last_log_id, ?self.last_applied, msg=%req.summary(), "handle_append_entries_request");
 
@@ -134,10 +133,10 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D,
     /// R5 2 4 4
     /// ```
     ///
-    /// If log 5 is committed by R1, and log 3 is not removed, R5 in future could become a new leader and overrides log
+    /// If log 5 is committed by R1, and log 3 is not removeC5 in future could become a new leader and overrides log
     /// 5 on R3.
     #[tracing::instrument(level="trace", skip(self, msg_entries), fields(msg_entries=%msg_entries.summary()))]
-    async fn find_and_delete_conflict_logs(&mut self, msg_entries: &[Entry<D>]) -> Result<(), StorageError> {
+    async fn find_and_delete_conflict_logs(&mut self, msg_entries: &[Entry<C>]) -> Result<(), StorageError> {
         // all msg_entries are inconsistent logs
 
         tracing::debug!(msg_entries=%msg_entries.summary(), "try to delete_inconsistent_log");
@@ -172,7 +171,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D,
     async fn append_apply_log_entries(
         &mut self,
         prev_log_id: Option<LogId>,
-        entries: &[Entry<D>],
+        entries: &[Entry<C>],
         committed: Option<LogId>,
     ) -> Result<AppendEntriesResponse, StorageError> {
         let mismatched = self.does_log_id_match(prev_log_id).await?;
@@ -239,8 +238,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D,
     /// Filter them out.
     pub async fn skip_matching_entries<'s, 'e>(
         &'s mut self,
-        entries: &'e [Entry<D>],
-    ) -> Result<(usize, &'e [Entry<D>]), StorageError> {
+        entries: &'e [Entry<C>],
+    ) -> Result<(usize, &'e [Entry<C>]), StorageError> {
         let l = entries.len();
 
         for i in 0..l {
@@ -307,7 +306,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetworkFactory<D>, S: RaftStorage<D,
     /// Configuration changes are also detected and applied here. See `configuration changes`
     /// in the raft-essentials.md in this repo.
     #[tracing::instrument(level = "trace", skip(self, entries), fields(entries=%entries.summary()))]
-    async fn append_log_entries(&mut self, entries: &[Entry<D>]) -> Result<(), StorageError> {
+    async fn append_log_entries(&mut self, entries: &[Entry<C>]) -> Result<(), StorageError> {
         if entries.is_empty() {
             return Ok(());
         }