Feature: Add Raft::begin_receiving_snapshot()

`Raft::begin_receiving_snapshot()` request the state machine to return a
`SnapshotData` for receiving snapshot from the leader.
Internally it calls `RaftStateMachine::begin_receiving_snapshot()`

Handling snapshot receiving is moved out of state-machine worker task.
Now it is in implemented outside the `RaftCore`.
Receiving snapshot could be totally application specific and should not
be part of Openraft.

The in sm-worker snapshot receiving is removed.

- Part of databendlabs#606

openraft/src/core/raft_core.rs

@@ -30,7 +30,6 @@ use crate::core::raft_msg::external_command::ExternalCommand;
 use crate::core::raft_msg::AppendEntriesTx;
 use crate::core::raft_msg::ClientReadTx;
 use crate::core::raft_msg::ClientWriteTx;
-use crate::core::raft_msg::InstallSnapshotTx;
 use crate::core::raft_msg::RaftMsg;
 use crate::core::raft_msg::ResultSender;
 use crate::core::raft_msg::VoteTx;
@@ -69,7 +68,6 @@ use crate::quorum::QuorumSet;
 use crate::raft::AppendEntriesRequest;
 use crate::raft::AppendEntriesResponse;
 use crate::raft::ClientWriteResponse;
-use crate::raft::InstallSnapshotRequest;
 use crate::raft::VoteRequest;
 use crate::raft_state::LogStateReader;
 use crate::replication;
@@ -607,21 +605,6 @@ where
         });
     }
 
-    /// Invoked by leader to send chunks of a snapshot to a follower.
-    ///
-    /// Leaders always send chunks in order. It is important to note that, according to the Raft
-    /// spec, a node may only have one snapshot at any time. As snapshot contents are application
-    /// specific.
-    #[tracing::instrument(level = "debug", skip_all)]
-    pub(crate) fn handle_install_snapshot_request(
-        &mut self,
-        req: InstallSnapshotRequest<C>,
-        tx: InstallSnapshotTx<C::NodeId>,
-    ) {
-        tracing::info!(req = display(req.summary()), "{}", func_name!());
-        self.engine.handle_install_snapshot(req, tx);
-    }
-
     /// Trigger a snapshot building(log compaction) job if there is no pending building job.
     #[tracing::instrument(level = "debug", skip(self))]
     pub(crate) fn trigger_snapshot(&mut self) {
@@ -1125,14 +1108,8 @@ where
 
                 self.handle_vote_request(rpc, tx);
             }
-            RaftMsg::InstallSnapshot { rpc, tx } => {
-                tracing::info!(
-                    req = display(rpc.summary()),
-                    "received RaftMst::InstallSnapshot: {}",
-                    func_name!()
-                );
-
-                self.handle_install_snapshot_request(rpc, tx);
+            RaftMsg::BeginReceivingSnapshot { vote, tx } => {
+                self.engine.handle_begin_receiving_snapshot(vote, tx);
             }
             RaftMsg::InstallCompleteSnapshot { vote, snapshot, tx } => {
                 self.engine.handle_install_complete_snapshot(vote, snapshot, tx);
@@ -1375,9 +1352,6 @@ where
                         let st = self.engine.state.io_state_mut();
                         st.update_snapshot(last_log_id);
                     }
-                    sm::Response::ReceiveSnapshotChunk(_) => {
-                        tracing::info!("sm::StateMachine command done: ReceiveSnapshotChunk: {}", func_name!());
-                    }
                     sm::Response::InstallSnapshot(meta) => {
                         tracing::info!(
                             "sm::StateMachine command done: InstallSnapshot: {}: {}",

openraft/src/core/raft_msg/mod.rs

@@ -5,20 +5,20 @@ use tokio::sync::oneshot;
 use crate::core::raft_msg::external_command::ExternalCommand;
 use crate::error::CheckIsLeaderError;
 use crate::error::ClientWriteError;
+use crate::error::HigherVote;
 use crate::error::Infallible;
 use crate::error::InitializeError;
-use crate::error::InstallSnapshotError;
 use crate::raft::AppendEntriesRequest;
 use crate::raft::AppendEntriesResponse;
 use crate::raft::BoxCoreFn;
 use crate::raft::ClientWriteResponse;
-use crate::raft::InstallSnapshotRequest;
 use crate::raft::InstallSnapshotResponse;
 use crate::raft::VoteRequest;
 use crate::raft::VoteResponse;
 use crate::type_config::alias::LogIdOf;
 use crate::type_config::alias::NodeIdOf;
 use crate::type_config::alias::NodeOf;
+use crate::type_config::alias::SnapshotDataOf;
 use crate::ChangeMembers;
 use crate::MessageSummary;
 use crate::RaftTypeConfig;
@@ -32,9 +32,6 @@ pub(crate) type ResultSender<T, E = Infallible> = oneshot::Sender<Result<T, E>>;
 
 pub(crate) type ResultReceiver<T, E = Infallible> = oneshot::Receiver<Result<T, E>>;
 
-/// TX for Install Snapshot Response
-pub(crate) type InstallSnapshotTx<NID> = ResultSender<InstallSnapshotResponse<NID>, InstallSnapshotError>;
-
 /// TX for Vote Response
 pub(crate) type VoteTx<NID> = ResultSender<VoteResponse<NID>>;
 
@@ -64,17 +61,22 @@ where C: RaftTypeConfig
         tx: VoteTx<C::NodeId>,
     },
 
-    InstallSnapshot {
-        rpc: InstallSnapshotRequest<C>,
-        tx: InstallSnapshotTx<C::NodeId>,
-    },
-
     InstallCompleteSnapshot {
         vote: Vote<C::NodeId>,
         snapshot: Snapshot<C>,
         tx: ResultSender<InstallSnapshotResponse<C::NodeId>>,
     },
 
+    /// Begin receiving a snapshot from the leader.
+    ///
+    /// Returns a handle to a snapshot data ready for receiving if successful.
+    /// Otherwise, it is an error because of the `vote` is not GE the local `vote`, the local `vote`
+    /// will be returned in a Err
+    BeginReceivingSnapshot {
+        vote: Vote<C::NodeId>,
+        tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
+    },
+
     ClientWriteRequest {
         app_data: C::D,
         tx: ClientWriteTx<C>,
@@ -119,8 +121,8 @@ where C: RaftTypeConfig
             RaftMsg::RequestVote { rpc, .. } => {
                 format!("RequestVote: {}", rpc.summary())
             }
-            RaftMsg::InstallSnapshot { rpc, .. } => {
-                format!("InstallSnapshot: {}", rpc.summary())
+            RaftMsg::BeginReceivingSnapshot { vote, .. } => {
+                format!("BeginReceivingSnapshot: vote: {}", vote)
             }
             RaftMsg::InstallCompleteSnapshot { vote, snapshot, .. } => {
                 format!("InstallCompleteSnapshot: vote: {}, snapshot: {}", vote, snapshot)

openraft/src/core/sm/command.rs

@@ -3,12 +3,11 @@ use std::fmt::Formatter;
 
 use crate::core::raft_msg::ResultSender;
 use crate::display_ext::DisplaySlice;
+use crate::error::HigherVote;
 use crate::log_id::RaftLogId;
-use crate::raft::InstallSnapshotRequest;
-use crate::MessageSummary;
+use crate::type_config::alias::SnapshotDataOf;
 use crate::RaftTypeConfig;
 use crate::Snapshot;
-use crate::SnapshotMeta;
 
 #[derive(PartialEq)]
 pub(crate) struct Command<C>
@@ -60,17 +59,8 @@ where C: RaftTypeConfig
         Command::new(payload)
     }
 
-    pub(crate) fn receive(req: InstallSnapshotRequest<C>) -> Self {
-        let payload = CommandPayload::ReceiveSnapshotChunk { req };
-        Command::new(payload)
-    }
-
-    // TODO: all sm command should have a command seq.
-    pub(crate) fn install_snapshot(snapshot_meta: SnapshotMeta<C::NodeId, C::Node>) -> Self {
-        let payload = CommandPayload::FinalizeSnapshot {
-            install: true,
-            snapshot_meta,
-        };
+    pub(crate) fn begin_receiving_snapshot(tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>) -> Self {
+        let payload = CommandPayload::BeginReceivingSnapshot { tx };
         Command::new(payload)
     }
 
@@ -79,14 +69,6 @@ where C: RaftTypeConfig
         Command::new(payload)
     }
 
-    pub(crate) fn cancel_snapshot(snapshot_meta: SnapshotMeta<C::NodeId, C::Node>) -> Self {
-        let payload = CommandPayload::FinalizeSnapshot {
-            install: false,
-            snapshot_meta,
-        };
-        Command::new(payload)
-    }
-
     pub(crate) fn apply(entries: Vec<C::Entry>) -> Self {
         let payload = CommandPayload::Apply { entries };
         Command::new(payload)
@@ -112,21 +94,8 @@ where C: RaftTypeConfig
         tx: ResultSender<Option<Snapshot<C>>>,
     },
 
-    /// Receive a chunk of snapshot.
-    ///
-    /// If it is the final chunk, the snapshot stream will be closed and saved.
-    ///
-    /// Installing a snapshot includes two steps: ReceiveSnapshotChunk and FinalizeSnapshot.
-    ReceiveSnapshotChunk {
-        req: InstallSnapshotRequest<C>,
-    },
-
-    /// After receiving all chunks, finalize the snapshot by installing it or discarding it,
-    /// if the snapshot is stale(the snapshot last log id is smaller than the local committed).
-    FinalizeSnapshot {
-        /// To install it, or just discard it.
-        install: bool,
-        snapshot_meta: SnapshotMeta<C::NodeId, C::Node>,
+    BeginReceivingSnapshot {
+        tx: ResultSender<Box<SnapshotDataOf<C>>, HigherVote<C::NodeId>>,
     },
 
     InstallCompleteSnapshot {
@@ -146,15 +115,12 @@ where C: RaftTypeConfig
         match self {
             CommandPayload::BuildSnapshot => write!(f, "BuildSnapshot"),
             CommandPayload::GetSnapshot { .. } => write!(f, "GetSnapshot"),
-            CommandPayload::ReceiveSnapshotChunk { req, .. } => {
-                write!(f, "ReceiveSnapshotChunk: {}", req.summary())
-            }
-            CommandPayload::FinalizeSnapshot { install, snapshot_meta } => {
-                write!(f, "FinalizeSnapshot: install:{} {:?}", install, snapshot_meta)
-            }
             CommandPayload::InstallCompleteSnapshot { snapshot } => {
                 write!(f, "InstallCompleteSnapshot: meta: {:?}", snapshot.meta)
             }
+            CommandPayload::BeginReceivingSnapshot { .. } => {
+                write!(f, "BeginReceivingSnapshot")
+            }
             CommandPayload::Apply { entries } => write!(f, "Apply: {}", DisplaySlice::<_>(entries)),
         }
     }
@@ -168,20 +134,11 @@ where C: RaftTypeConfig
         match (self, other) {
             (CommandPayload::BuildSnapshot, CommandPayload::BuildSnapshot) => true,
             (CommandPayload::GetSnapshot { .. }, CommandPayload::GetSnapshot { .. }) => true,
+            (CommandPayload::BeginReceivingSnapshot { .. }, CommandPayload::BeginReceivingSnapshot { .. }) => true,
             (
-                CommandPayload::ReceiveSnapshotChunk { req: req1, .. },
-                CommandPayload::ReceiveSnapshotChunk { req: req2, .. },
-            ) => req1 == req2,
-            (
-                CommandPayload::FinalizeSnapshot {
-                    install: install1,
-                    snapshot_meta: meta1,
-                },
-                CommandPayload::FinalizeSnapshot {
-                    install: install2,
-                    snapshot_meta: meta2,
-                },
-            ) => install1 == install2 && meta1 == meta2,
+                CommandPayload::InstallCompleteSnapshot { snapshot: s1 },
+                CommandPayload::InstallCompleteSnapshot { snapshot: s2 },
+            ) => s1.meta == s2.meta,
             (CommandPayload::Apply { entries: entries1 }, CommandPayload::Apply { entries: entries2 }) => {
                 // Entry may not be `Eq`, we just compare log id.
                 // This would be enough for testing.