Change: move RaftStateMachine out of RaftStorage

In Raft, the state machine is an independent storage component that
operates separately from the log store. As a result, accessing the log
store and accessing the state machine can be naturally parallelized.

This commit replaces the type parameter `RaftStorage` in
`Raft<.., S: RaftStorage>` with two type parameters: `RaftLogStorage` and
`RaftStateMachine`.

- Add: `RaftLogReaderExt` to provide additional log access methods based
  on a `RaftLogReader` implementation. Some of the methods are moved
  from `StorageHelper` to this trait.

- Add: `Adapter` to let application use the seperated log state machine
  framework without rewriting `RaftStorage` implementation.

- Refactor: shorten type names for the 2 example crates

### TODO

- [ ] Callback based log append is defined but is not used.

### Upgrade tip

Use an adapter to wrap `RaftStorage`:
```rust
// Before:
let store = MyRaftStorage::new();
Raft::new(..., store);

// After:
let store = MyRaftStorage::new();
let (log_store, sm) = Adaptoer::new(store);
Raft::new(..., log_store, sm);
```

examples/raft-kv-memstore/src/app.rs

@@ -1,17 +1,15 @@
 use std::sync::Arc;
 
-use openraft::Config;
-
-use crate::ExampleNodeId;
-use crate::ExampleRaft;
-use crate::ExampleStore;
+use crate::NodeId;
+use crate::Raft;
+use crate::Store;
 
 // Representation of an application state. This struct can be shared around to share
 // instances of raft, store and more.
-pub struct ExampleApp {
-    pub id: ExampleNodeId,
+pub struct App {
+    pub id: NodeId,
     pub addr: String,
-    pub raft: ExampleRaft,
-    pub store: Arc<ExampleStore>,
-    pub config: Arc<Config>,
+    pub raft: Raft,
+    pub store: Arc<Store>,
+    pub config: Arc<openraft::Config>,
 }

examples/raft-kv-memstore/src/bin/main.rs

@@ -1,13 +1,7 @@
 use clap::Parser;
-use openraft::Raft;
-use raft_kv_memstore::network::raft_network_impl::ExampleNetwork;
 use raft_kv_memstore::start_example_raft_node;
-use raft_kv_memstore::store::ExampleStore;
-use raft_kv_memstore::ExampleTypeConfig;
 use tracing_subscriber::EnvFilter;
 
-pub type ExampleRaft = Raft<ExampleTypeConfig, ExampleNetwork, ExampleStore>;
-
 #[derive(Parser, Clone, Debug)]
 #[clap(author, version, about, long_about = None)]
 pub struct Opt {

examples/raft-kv-memstore/src/client.rs

@@ -5,20 +5,18 @@ use std::time::Duration;
 
 use openraft::error::ForwardToLeader;
 use openraft::error::NetworkError;
-use openraft::error::RPCError;
 use openraft::error::RemoteError;
 use openraft::BasicNode;
 use openraft::RaftMetrics;
 use openraft::TryAsRef;
-use reqwest::Client;
 use serde::de::DeserializeOwned;
 use serde::Deserialize;
 use serde::Serialize;
 use tokio::time::timeout;
 
 use crate::typ;
-use crate::ExampleNodeId;
-use crate::ExampleRequest;
+use crate::NodeId;
+use crate::Request;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Empty {}
@@ -27,14 +25,14 @@ pub struct ExampleClient {
     /// The leader node to send request to.
     ///
     /// All traffic should be sent to the leader in a cluster.
-    pub leader: Arc<Mutex<(ExampleNodeId, String)>>,
+    pub leader: Arc<Mutex<(NodeId, String)>>,
 
-    pub inner: Client,
+    pub inner: reqwest::Client,
 }
 
 impl ExampleClient {
     /// Create a client with a leader node id and a node manager to get node address by node id.
-    pub fn new(leader_id: ExampleNodeId, leader_addr: String) -> Self {
+    pub fn new(leader_id: NodeId, leader_addr: String) -> Self {
         Self {
             leader: Arc::new(Mutex::new((leader_id, leader_addr))),
             inner: reqwest::Client::new(),
@@ -49,10 +47,7 @@ impl ExampleClient {
     /// will be applied to state machine.
     ///
     /// The result of applying the request will be returned.
-    pub async fn write(
-        &self,
-        req: &ExampleRequest,
-    ) -> Result<typ::ClientWriteResponse, typ::RPCError<typ::ClientWriteError>> {
+    pub async fn write(&self, req: &Request) -> Result<typ::ClientWriteResponse, typ::RPCError<typ::ClientWriteError>> {
         self.send_rpc_to_leader("write", Some(req)).await
     }
 
@@ -87,7 +82,7 @@ impl ExampleClient {
     /// The node to add has to exist, i.e., being added with `write(ExampleRequest::AddNode{})`
     pub async fn add_learner(
         &self,
-        req: (ExampleNodeId, String),
+        req: (NodeId, String),
     ) -> Result<typ::ClientWriteResponse, typ::RPCError<typ::ClientWriteError>> {
         self.send_rpc_to_leader("add-learner", Some(&req)).await
     }
@@ -98,7 +93,7 @@ impl ExampleClient {
     /// or an error [`LearnerNotFound`] will be returned.
     pub async fn change_membership(
         &self,
-        req: &BTreeSet<ExampleNodeId>,
+        req: &BTreeSet<NodeId>,
     ) -> Result<typ::ClientWriteResponse, typ::RPCError<typ::ClientWriteError>> {
         self.send_rpc_to_leader("change-membership", Some(req)).await
     }
@@ -108,7 +103,7 @@ impl ExampleClient {
     /// Metrics contains various information about the cluster, such as current leader,
     /// membership config, replication status etc.
     /// See [`RaftMetrics`].
-    pub async fn metrics(&self) -> Result<RaftMetrics<ExampleNodeId, BasicNode>, typ::RPCError> {
+    pub async fn metrics(&self) -> Result<RaftMetrics<NodeId, BasicNode>, typ::RPCError> {
         self.do_send_rpc_to_leader("metrics", None::<&()>).await
     }
 
@@ -118,7 +113,8 @@ impl ExampleClient {
     ///
     /// It sends out a POST request if `req` is Some. Otherwise a GET request.
     /// The remote endpoint must respond a reply in form of `Result<T, E>`.
-    /// An `Err` happened on remote will be wrapped in an [`RPCError::RemoteError`].
+    /// An `Err` happened on remote will be wrapped in an
+    /// [`openraft::error::RPCError::RemoteError`].
     async fn do_send_rpc_to_leader<Req, Resp, Err>(
         &self,
         uri: &str,
@@ -150,22 +146,22 @@ impl ExampleClient {
 
         let res = timeout(Duration::from_millis(3_000), fu).await;
         let resp = match res {
-            Ok(x) => x.map_err(|e| RPCError::Network(NetworkError::new(&e)))?,
+            Ok(x) => x.map_err(|e| typ::RPCError::Network(NetworkError::new(&e)))?,
             Err(timeout_err) => {
                 tracing::error!("timeout {} to url: {}", timeout_err, url);
-                return Err(RPCError::Network(NetworkError::new(&timeout_err)));
+                return Err(typ::RPCError::Network(NetworkError::new(&timeout_err)));
             }
         };
 
         let res: Result<Resp, typ::RaftError<Err>> =
-            resp.json().await.map_err(|e| RPCError::Network(NetworkError::new(&e)))?;
+            resp.json().await.map_err(|e| typ::RPCError::Network(NetworkError::new(&e)))?;
         tracing::debug!(
             "<<< client recv reply from {}: {}",
             url,
             serde_json::to_string_pretty(&res).unwrap()
         );
 
-        res.map_err(|e| RPCError::RemoteError(RemoteError::new(leader_id, e)))
+        res.map_err(|e| typ::RPCError::RemoteError(RemoteError::new(leader_id, e)))
     }
 
     /// Try the best to send a request to the leader.

examples/raft-kv-memstore/src/lib.rs

@@ -6,54 +6,54 @@ use std::sync::Arc;
 use actix_web::middleware;
 use actix_web::middleware::Logger;
 use actix_web::web::Data;
-use actix_web::App;
 use actix_web::HttpServer;
+use openraft::storage::Adaptor;
 use openraft::BasicNode;
 use openraft::Config;
-use openraft::Raft;
 
-use crate::app::ExampleApp;
+use crate::app::App;
 use crate::network::api;
 use crate::network::management;
 use crate::network::raft;
-use crate::network::raft_network_impl::ExampleNetwork;
-use crate::store::ExampleRequest;
-use crate::store::ExampleResponse;
-use crate::store::ExampleStore;
+use crate::network::Network;
+use crate::store::Request;
+use crate::store::Response;
+use crate::store::Store;
 
 pub mod app;
 pub mod client;
 pub mod network;
 pub mod store;
 
-pub type ExampleNodeId = u64;
+pub type NodeId = u64;
 
 openraft::declare_raft_types!(
     /// Declare the type configuration for example K/V store.
-    pub ExampleTypeConfig: D = ExampleRequest, R = ExampleResponse, NodeId = ExampleNodeId, Node = BasicNode, Entry = openraft::Entry<ExampleTypeConfig>
+    pub TypeConfig: D = Request, R = Response, NodeId = NodeId, Node = BasicNode, Entry = openraft::Entry<TypeConfig>
 );
 
-pub type ExampleRaft = Raft<ExampleTypeConfig, ExampleNetwork, Arc<ExampleStore>>;
+pub type LogStore = Adaptor<TypeConfig, Arc<Store>>;
+pub type StateMachineStore = Adaptor<TypeConfig, Arc<Store>>;
+pub type Raft = openraft::Raft<TypeConfig, Network, LogStore, StateMachineStore>;
 
 pub mod typ {
     use openraft::BasicNode;
 
-    use crate::ExampleNodeId;
-    use crate::ExampleTypeConfig;
+    use crate::NodeId;
+    use crate::TypeConfig;
 
-    pub type RaftError<E = openraft::error::Infallible> = openraft::error::RaftError<ExampleNodeId, E>;
-    pub type RPCError<E = openraft::error::Infallible> =
-        openraft::error::RPCError<ExampleNodeId, BasicNode, RaftError<E>>;
+    pub type RaftError<E = openraft::error::Infallible> = openraft::error::RaftError<NodeId, E>;
+    pub type RPCError<E = openraft::error::Infallible> = openraft::error::RPCError<NodeId, BasicNode, RaftError<E>>;
 
-    pub type ClientWriteError = openraft::error::ClientWriteError<ExampleNodeId, BasicNode>;
-    pub type CheckIsLeaderError = openraft::error::CheckIsLeaderError<ExampleNodeId, BasicNode>;
-    pub type ForwardToLeader = openraft::error::ForwardToLeader<ExampleNodeId, BasicNode>;
-    pub type InitializeError = openraft::error::InitializeError<ExampleNodeId, BasicNode>;
+    pub type ClientWriteError = openraft::error::ClientWriteError<NodeId, BasicNode>;
+    pub type CheckIsLeaderError = openraft::error::CheckIsLeaderError<NodeId, BasicNode>;
+    pub type ForwardToLeader = openraft::error::ForwardToLeader<NodeId, BasicNode>;
+    pub type InitializeError = openraft::error::InitializeError<NodeId, BasicNode>;
 
-    pub type ClientWriteResponse = openraft::raft::ClientWriteResponse<ExampleTypeConfig>;
+    pub type ClientWriteResponse = openraft::raft::ClientWriteResponse<TypeConfig>;
 }
 
-pub async fn start_example_raft_node(node_id: ExampleNodeId, http_addr: String) -> std::io::Result<()> {
+pub async fn start_example_raft_node(node_id: NodeId, http_addr: String) -> std::io::Result<()> {
     // Create a configuration for the raft instance.
     let config = Config {
         heartbeat_interval: 500,
@@ -65,18 +65,20 @@ pub async fn start_example_raft_node(node_id: ExampleNodeId, http_addr: String)
     let config = Arc::new(config.validate().unwrap());
 
     // Create a instance of where the Raft data will be stored.
-    let store = Arc::new(ExampleStore::default());
+    let store = Arc::new(Store::default());
+
+    let (log_store, state_machine) = Adaptor::new(store.clone());
 
     // Create the network layer that will connect and communicate the raft instances and
     // will be used in conjunction with the store created above.
-    let network = ExampleNetwork {};
+    let network = Network {};
 
     // Create a local raft instance.
-    let raft = Raft::new(node_id, config.clone(), network, store.clone()).await.unwrap();
+    let raft = openraft::Raft::new(node_id, config.clone(), network, log_store, state_machine).await.unwrap();
 
     // Create an application that will store all the instances created above, this will
     // be later used on the actix-web services.
-    let app = Data::new(ExampleApp {
+    let app_data = Data::new(App {
         id: node_id,
         addr: http_addr.clone(),
         raft,
@@ -86,11 +88,11 @@ pub async fn start_example_raft_node(node_id: ExampleNodeId, http_addr: String)
 
     // Start the actix-web server.
     let server = HttpServer::new(move || {
-        App::new()
+        actix_web::App::new()
             .wrap(Logger::default())
             .wrap(Logger::new("%a %{User-Agent}i"))
             .wrap(middleware::Compress::default())
-            .app_data(app.clone())
+            .app_data(app_data.clone())
             // raft internal RPC
             .service(raft::append)
             .service(raft::snapshot)

examples/raft-kv-memstore/src/network/api.rs

@@ -8,9 +8,9 @@ use openraft::error::RaftError;
 use openraft::BasicNode;
 use web::Json;
 
-use crate::app::ExampleApp;
-use crate::store::ExampleRequest;
-use crate::ExampleNodeId;
+use crate::app::App;
+use crate::store::Request;
+use crate::NodeId;
 
 /**
  * Application API
@@ -22,13 +22,13 @@ use crate::ExampleNodeId;
  *  - `POST - /read` attempt to find a value from a given key.
  */
 #[post("/write")]
-pub async fn write(app: Data<ExampleApp>, req: Json<ExampleRequest>) -> actix_web::Result<impl Responder> {
+pub async fn write(app: Data<App>, req: Json<Request>) -> actix_web::Result<impl Responder> {
     let response = app.raft.client_write(req.0).await;
     Ok(Json(response))
 }
 
 #[post("/read")]
-pub async fn read(app: Data<ExampleApp>, req: Json<String>) -> actix_web::Result<impl Responder> {
+pub async fn read(app: Data<App>, req: Json<String>) -> actix_web::Result<impl Responder> {
     let state_machine = app.store.state_machine.read().await;
     let key = req.0;
     let value = state_machine.data.get(&key).cloned();
@@ -38,7 +38,7 @@ pub async fn read(app: Data<ExampleApp>, req: Json<String>) -> actix_web::Result
 }
 
 #[post("/consistent_read")]
-pub async fn consistent_read(app: Data<ExampleApp>, req: Json<String>) -> actix_web::Result<impl Responder> {
+pub async fn consistent_read(app: Data<App>, req: Json<String>) -> actix_web::Result<impl Responder> {
     let ret = app.raft.is_leader().await;
 
     match ret {
@@ -47,7 +47,7 @@ pub async fn consistent_read(app: Data<ExampleApp>, req: Json<String>) -> actix_
             let key = req.0;
             let value = state_machine.data.get(&key).cloned();
 
-            let res: Result<String, RaftError<ExampleNodeId, CheckIsLeaderError<ExampleNodeId, BasicNode>>> =
+            let res: Result<String, RaftError<NodeId, CheckIsLeaderError<NodeId, BasicNode>>> =
                 Ok(value.unwrap_or_default());
             Ok(Json(res))
         }

examples/raft-kv-memstore/src/network/management.rs

@@ -3,16 +3,15 @@ use std::collections::BTreeSet;
 
 use actix_web::get;
 use actix_web::post;
-use actix_web::web;
 use actix_web::web::Data;
+use actix_web::web::Json;
 use actix_web::Responder;
 use openraft::error::Infallible;
 use openraft::BasicNode;
 use openraft::RaftMetrics;
-use web::Json;
 
-use crate::app::ExampleApp;
-use crate::ExampleNodeId;
+use crate::app::App;
+use crate::NodeId;
 
 // --- Cluster management
 
@@ -22,10 +21,7 @@ use crate::ExampleNodeId;
 /// This should be done before adding a node as a member into the cluster
 /// (by calling `change-membership`)
 #[post("/add-learner")]
-pub async fn add_learner(
-    app: Data<ExampleApp>,
-    req: Json<(ExampleNodeId, String)>,
-) -> actix_web::Result<impl Responder> {
+pub async fn add_learner(app: Data<App>, req: Json<(NodeId, String)>) -> actix_web::Result<impl Responder> {
     let node_id = req.0 .0;
     let node = BasicNode { addr: req.0 .1.clone() };
     let res = app.raft.add_learner(node_id, node, true).await;
@@ -34,17 +30,14 @@ pub async fn add_learner(
 
 /// Changes specified learners to members, or remove members.
 #[post("/change-membership")]
-pub async fn change_membership(
-    app: Data<ExampleApp>,
-    req: Json<BTreeSet<ExampleNodeId>>,
-) -> actix_web::Result<impl Responder> {
+pub async fn change_membership(app: Data<App>, req: Json<BTreeSet<NodeId>>) -> actix_web::Result<impl Responder> {
     let res = app.raft.change_membership(req.0, false).await;
     Ok(Json(res))
 }
 
 /// Initialize a single-node cluster.
 #[post("/init")]
-pub async fn init(app: Data<ExampleApp>) -> actix_web::Result<impl Responder> {
+pub async fn init(app: Data<App>) -> actix_web::Result<impl Responder> {
     let mut nodes = BTreeMap::new();
     nodes.insert(app.id, BasicNode { addr: app.addr.clone() });
     let res = app.raft.initialize(nodes).await;
@@ -53,9 +46,9 @@ pub async fn init(app: Data<ExampleApp>) -> actix_web::Result<impl Responder> {
 
 /// Get the latest metrics of the cluster
 #[get("/metrics")]
-pub async fn metrics(app: Data<ExampleApp>) -> actix_web::Result<impl Responder> {
+pub async fn metrics(app: Data<App>) -> actix_web::Result<impl Responder> {
     let metrics = app.raft.metrics().borrow().clone();
 
-    let res: Result<RaftMetrics<ExampleNodeId, BasicNode>, Infallible> = Ok(metrics);
+    let res: Result<RaftMetrics<NodeId, BasicNode>, Infallible> = Ok(metrics);
     Ok(Json(res))
 }

examples/raft-kv-memstore/src/network/mod.rs

@@ -1,4 +1,7 @@
 pub mod api;
 pub mod management;
 pub mod raft;
-pub mod raft_network_impl;
+mod raft_network_impl;
+
+pub use raft_network_impl::Network;
+pub use raft_network_impl::NetworkConnection;