change: reduce one unnecessary snapshot serialization

- Change: `get_current_snapshot()`: remove double-serialization:
  convert MemStoreSnapshot to CurrentSnapshotData instead of serializing
  MemStoreSnapshot:

  Before:
  ```
  MemStoreSnapshot.data = serialize(state-machine)
  CurrentSnapshotData.data = serialize(MemStoreSnapshot)
  ```

  After:
  ```
  MemStoreSnapshot.data = serialize(state-machine)
  CurrentSnapshotData.data = MemStoreSnapshot.data
  ```

  when `finalize_snapshot_installation`, extract snapshot meta info from
  `InstallSnapshotRequest`. Reduce one unnecessary deserialization.

- Change: InstallSnapshotRequest: merge `snapshot_id`, `last_log_id`,
  `membership` into one field `meta`.

- Refactor: use SnapshotMeta(`snapshot_id`, `last_log_id`, `membership`) as
  a container of metadata of a snapshot.
  Reduce parameters.

- Refactor: remove redundent param `delete_through` from
  `finalize_snapshot_installation`.

Author: drmingdrmer

async-raft/src/core/install_snapshot.rs

@@ -74,8 +74,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
                 return self.begin_installing_snapshot(req).await;
             }
             Some(SnapshotState::Streaming { snapshot, id, offset }) => {
-                if req.snapshot_id == id {
-                    return self.continue_installing_snapshot(req, offset, id, snapshot).await;
+                if req.meta.snapshot_id == id {
+                    return self.continue_installing_snapshot(req, offset, snapshot).await;
                 }
 
                 if req.offset == 0 {
@@ -85,7 +85,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
                 Err(RaftError::SnapshotMismatch {
                     expect: SnapshotSegmentId { id: id.clone(), offset },
                     got: SnapshotSegmentId {
-                        id: req.snapshot_id.clone(),
+                        id: req.meta.snapshot_id.clone(),
                         offset: req.offset,
                     },
                 })
@@ -96,19 +96,19 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     #[tracing::instrument(level = "trace", skip(self, req))]
     async fn begin_installing_snapshot(&mut self, req: InstallSnapshotRequest) -> RaftResult<InstallSnapshotResponse> {
         // Create a new snapshot and begin writing its contents.
-        let id = req.snapshot_id.clone();
+        let id = req.meta.snapshot_id.clone();
         let mut snapshot = self.storage.create_snapshot().await.map_err(|err| self.map_fatal_storage_error(err))?;
         snapshot.as_mut().write_all(&req.data).await?;
 
         // If this was a small snapshot, and it is already done, then finish up.
         if req.done {
-            self.finalize_snapshot_installation(req, id, snapshot).await?;
+            self.finalize_snapshot_installation(req, snapshot).await?;
             return Ok(InstallSnapshotResponse {
                 term: self.current_term,
             });
         }
 
-        // Else, retain snapshot components for later segments & respod.
+        // Else, retain snapshot components for later segments & respond.
         self.snapshot_state = Some(SnapshotState::Streaming {
             offset: req.data.len() as u64,
             id,
@@ -124,9 +124,10 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         &mut self,
         req: InstallSnapshotRequest,
         mut offset: u64,
-        id: String,
         mut snapshot: Box<S::Snapshot>,
     ) -> RaftResult<InstallSnapshotResponse> {
+        let id = req.meta.snapshot_id.clone();
+
         // Always seek to the target offset if not an exact match.
         if req.offset != offset {
             if let Err(err) = snapshot.as_mut().seek(SeekFrom::Start(req.offset)).await {
@@ -145,7 +146,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
         // If the snapshot stream is done, then finalize.
         if req.done {
-            self.finalize_snapshot_installation(req, id, snapshot).await?;
+            self.finalize_snapshot_installation(req, snapshot).await?;
         } else {
             self.snapshot_state = Some(SnapshotState::Streaming { offset, id, snapshot });
         }
@@ -161,30 +162,20 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     async fn finalize_snapshot_installation(
         &mut self,
         req: InstallSnapshotRequest,
-        id: String,
         mut snapshot: Box<S::Snapshot>,
     ) -> RaftResult<()> {
         snapshot.as_mut().shutdown().await.map_err(|err| self.map_fatal_storage_error(err.into()))?;
-        let delete_through = if self.last_log_id.index > req.last_log_id.index {
-            Some(req.last_log_id.index)
-        } else {
-            None
-        };
+
         self.storage
-            .finalize_snapshot_installation(
-                req.last_log_id.index,
-                req.last_log_id.term,
-                delete_through,
-                id,
-                snapshot,
-            )
+            .finalize_snapshot_installation(&req.meta, snapshot)
             .await
             .map_err(|err| self.map_fatal_storage_error(err))?;
+
         let membership = self.storage.get_membership_config().await.map_err(|err| self.map_fatal_storage_error(err))?;
         self.update_membership(membership)?;
-        self.last_log_id = req.last_log_id;
-        self.last_applied = req.last_log_id.index;
-        self.snapshot_last_log_id = req.last_log_id;
+        self.last_log_id = req.meta.last_log_id;
+        self.last_applied = req.meta.last_log_id.index;
+        self.snapshot_last_log_id = req.meta.last_log_id;
         self.report_metrics(Update::Ignore);
         Ok(())
     }

async-raft/src/raft.rs

@@ -22,13 +22,13 @@ use crate::error::RaftError;
 use crate::error::RaftResult;
 use crate::metrics::RaftMetrics;
 use crate::metrics::Wait;
-use crate::raft_types::SnapshotId;
 use crate::AppData;
 use crate::AppDataResponse;
 use crate::LogId;
 use crate::NodeId;
 use crate::RaftNetwork;
 use crate::RaftStorage;
+use crate::SnapshotMeta;
 
 struct RaftInner<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> {
     tx_api: mpsc::UnboundedSender<RaftMsg<D, R>>,
@@ -422,21 +422,14 @@ pub struct Entry<D: AppData> {
 }
 
 impl<D: AppData> Entry<D> {
-    /// Create a new snapshot pointer from the given data.
-    ///
-    /// ### index & term
-    /// The index and term of the entry being replaced by this snapshot pointer entry.
-    ///
-    /// ### id
-    /// The ID of the associated snapshot.
-    ///
-    /// ### membership
-    /// The cluster membership config which is contained in the snapshot, which will always be the
-    /// latest membership covered by the snapshot.
-    pub fn new_snapshot_pointer(index: u64, term: u64, id: String, membership: MembershipConfig) -> Self {
+    /// Create a new snapshot pointer from the given snapshot meta.
+    pub fn new_snapshot_pointer(meta: &SnapshotMeta) -> Self {
         Entry {
-            log_id: LogId { term, index },
-            payload: EntryPayload::SnapshotPointer(EntrySnapshotPointer { id, membership }),
+            log_id: meta.last_log_id,
+            payload: EntryPayload::SnapshotPointer(EntrySnapshotPointer {
+                id: meta.snapshot_id.clone(),
+                membership: meta.membership.clone(),
+            }),
         }
     }
 }
@@ -581,15 +574,15 @@ pub struct InstallSnapshotRequest {
     pub term: u64,
     /// The leader's ID. Useful in redirecting clients.
     pub leader_id: u64,
-    /// The Id of a snapshot.
-    /// Every two snapshots should have different snapshot id.
-    pub snapshot_id: SnapshotId,
-    /// The snapshot replaces all log entries up through and including this log.
-    pub last_log_id: LogId,
+
+    /// Metadata of a snapshot: snapshot_id, last_log_ed membership etc.
+    pub meta: SnapshotMeta,
+
     /// The byte offset where this chunk of data is positioned in the snapshot file.
     pub offset: u64,
     /// The raw bytes of the snapshot chunk, starting at `offset`.
     pub data: Vec<u8>,
+
     /// Will be `true` if this is the last chunk in the snapshot.
     pub done: bool,
 }

async-raft/src/replication/mod.rs

@@ -829,7 +829,6 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
 
     #[tracing::instrument(level = "trace", skip(self, snapshot))]
     async fn stream_snapshot(&mut self, mut snapshot: CurrentSnapshotData<S::Snapshot>) -> RaftResult<()> {
-        let snapshot_id = snapshot.meta.snapshot_id.clone();
         let mut offset = 0;
         self.core.next_index = snapshot.meta.last_log_id.index + 1;
         self.core.matched = snapshot.meta.last_log_id;
@@ -842,8 +841,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             let req = InstallSnapshotRequest {
                 term: self.core.term,
                 leader_id: self.core.id,
-                snapshot_id: snapshot_id.clone(),
-                last_log_id: snapshot.meta.last_log_id,
+                meta: snapshot.meta.clone(),
                 offset,
                 data: Vec::from(&buf[..nread]),
                 done,

async-raft/src/storage.rs

@@ -219,10 +219,9 @@ where
 
     /// Finalize the installation of a snapshot which has finished streaming from the cluster leader.
     ///
-    /// Delete all entries in the log through `delete_through`, unless `None`, in which case
-    /// all entries of the log are to be deleted.
+    /// Delete all entries in the log through `meta.last_log_id.index`.
     ///
-    /// Write a new snapshot pointer to the log at the given `index`. The snapshot pointer should be
+    /// Write a new snapshot pointer to the log at the given `meta.last_log_id.index`. The snapshot pointer should be
     /// constructed via the `Entry::new_snapshot_pointer` constructor and the other parameters
     /// provided to this method.
     ///
@@ -235,14 +234,7 @@ where
     /// made to the snapshot.
     ///
     /// Errors returned from this method will cause Raft to go into shutdown.
-    async fn finalize_snapshot_installation(
-        &self,
-        index: u64,
-        term: u64,
-        delete_through: Option<u64>,
-        id: String,
-        snapshot: Box<Self::Snapshot>,
-    ) -> Result<()>;
+    async fn finalize_snapshot_installation(&self, meta: &SnapshotMeta, snapshot: Box<Self::Snapshot>) -> Result<()>;
 
     /// Get a readable handle to the current snapshot, along with its metadata.
     ///

async-raft/tests/api_install_snapshot.rs

@@ -6,6 +6,7 @@ use anyhow::Result;
 use async_raft::raft::InstallSnapshotRequest;
 use async_raft::Config;
 use async_raft::LogId;
+use async_raft::SnapshotMeta;
 use async_raft::State;
 use fixtures::RaftRouter;
 use maplit::hashset;
@@ -46,8 +47,11 @@ async fn snapshot_ge_half_threshold() -> Result<()> {
     let req0 = InstallSnapshotRequest {
         term: 1,
         leader_id: 0,
-        snapshot_id: "ss1".into(),
-        last_log_id: LogId { term: 1, index: 0 },
+        meta: SnapshotMeta {
+            snapshot_id: "ss1".into(),
+            last_log_id: LogId { term: 1, index: 0 },
+            membership: Default::default(),
+        },
         offset: 0,
         data: vec![1, 2, 3],
         done: false,
@@ -62,7 +66,7 @@ async fn snapshot_ge_half_threshold() -> Result<()> {
     {
         let mut req = req0.clone();
         req.offset = 3;
-        req.snapshot_id = "ss2".into();
+        req.meta.snapshot_id = "ss2".into();
         let res = n.0.install_snapshot(req).await;
         assert_eq!("expect: ss1+3, got: ss2+3", res.unwrap_err().to_string());
     }
@@ -71,20 +75,20 @@ async fn snapshot_ge_half_threshold() -> Result<()> {
     {
         let mut req = req0.clone();
         req.offset = 0;
-        req.snapshot_id = "ss2".into();
+        req.meta.snapshot_id = "ss2".into();
         n.0.install_snapshot(req).await?;
 
         let mut req = req0.clone();
         req.offset = 3;
-        req.snapshot_id = "ss2".into();
+        req.meta.snapshot_id = "ss2".into();
         n.0.install_snapshot(req).await?;
     }
 
     tracing::info!("-- continue write with mismatched offset is allowed");
     {
         let mut req = req0.clone();
         req.offset = 8;
-        req.snapshot_id = "ss2".into();
+        req.meta.snapshot_id = "ss2".into();
         n.0.install_snapshot(req).await?;
     }
     Ok(())