change: CurrentSnapshotData: merge term and index into included.

databendlabs · Jul 9, 2021 · 85859d0 · 85859d0
1 parent 933e0b3
commit 85859d0
Show file tree

Hide file tree

Showing 6 changed files with 19 additions and 21 deletions.
diff --git a/async-raft/src/core/mod.rs b/async-raft/src/core/mod.rs
@@ -211,7 +211,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         if let Some(snapshot) =
             self.storage.get_current_snapshot().await.map_err(|err| self.map_fatal_storage_error(err))?
         {
-            self.snapshot_index = snapshot.index;
+            self.snapshot_index = snapshot.included.index;
         }
 
         let has_log = self.last_log_index != u64::min_value();
@@ -452,8 +452,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
                 match res {
                     Ok(res) => match res {
                         Ok(snapshot) => {
-                            let _ = tx_compaction.try_send(SnapshotUpdate::SnapshotComplete(snapshot.index));
-                            let _ = chan_tx.send(snapshot.index); // This will always succeed.
+                            let _ = tx_compaction.try_send(SnapshotUpdate::SnapshotComplete(snapshot.included.index));
+                            let _ = chan_tx.send(snapshot.included.index); // This will always succeed.
                         }
                         Err(err) => {
                             tracing::error!({error=%err}, "error while generating snapshot");

diff --git a/async-raft/src/core/replication.rs b/async-raft/src/core/replication.rs
@@ -288,7 +288,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
         if let Some(snapshot) = current_snapshot_opt {
             // If snapshot exists, ensure its distance from the leader's last log index is <= half
             // of the configured snapshot threshold, else create a new snapshot.
-            if snapshot_is_within_half_of_threshold(&snapshot.index, &self.core.last_log_index, &threshold) {
+            if snapshot_is_within_half_of_threshold(&snapshot.included.index, &self.core.last_log_index, &threshold) {
                 let _ = tx.send(snapshot);
                 return Ok(());
             }

diff --git a/async-raft/src/replication/mod.rs b/async-raft/src/replication/mod.rs
@@ -832,8 +832,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     async fn stream_snapshot(&mut self, mut snapshot: CurrentSnapshotData<S::Snapshot>) -> RaftResult<()> {
         let snapshot_id = snapshot.snapshot_id.clone();
         let mut offset = 0;
-        self.core.next_index = snapshot.index + 1;
-        self.core.matched = (snapshot.term, snapshot.index).into();
+        self.core.next_index = snapshot.included.index + 1;
+        self.core.matched = snapshot.included;
         let mut buf = Vec::with_capacity(self.core.config.snapshot_max_chunk_size as usize);
         loop {
             // Build the RPC.
@@ -844,7 +844,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 term: self.core.term,
                 leader_id: self.core.id,
                 snapshot_id: snapshot_id.clone(),
-                last_included: (snapshot.term, snapshot.index).into(),
+                last_included: snapshot.included,
                 offset,
                 data: Vec::from(&buf[..nread]),
                 done,

diff --git a/async-raft/src/storage.rs b/async-raft/src/storage.rs
@@ -15,16 +15,16 @@ use crate::raft::MembershipConfig;
 use crate::raft_types::SnapshotId;
 use crate::AppData;
 use crate::AppDataResponse;
+use crate::LogId;
 use crate::NodeId;
 
 /// The data associated with the current snapshot.
 pub struct CurrentSnapshotData<S>
 where S: AsyncRead + AsyncSeek + Send + Unpin + 'static
 {
-    /// The snapshot entry's term.
-    pub term: u64,
-    /// The snapshot entry's index.
-    pub index: u64,
+    // Log entries upto which this snapshot includes, inclusive.
+    pub included: LogId,
+
     /// The latest membership configuration covered by the snapshot.
     pub membership: MembershipConfig,
 

diff --git a/async-raft/tests/fixtures/mod.rs b/async-raft/tests/fixtures/mod.rs
@@ -485,22 +485,22 @@ impl RaftRouter {
                     .unwrap_or_else(|| panic!("no snapshot present for node {}", id));
                 match index_test {
                     ValueTest::Exact(index) => assert_eq!(
-                        &snap.index, index,
+                        &snap.included.index, index,
                         "expected node {} to have snapshot with index {}, got {}",
-                        id, index, snap.index
+                        id, index, snap.included.index
                     ),
                     ValueTest::Range(range) => assert!(
-                        range.contains(&snap.index),
+                        range.contains(&snap.included.index),
                         "expected node {} to have snapshot within range {:?}, got {}",
                         id,
                         range,
-                        snap.index
+                        snap.included.index
                     ),
                 }
                 assert_eq!(
-                    &snap.term, term,
+                    &snap.included.term, term,
                     "expected node {} to have snapshot with term {}, got {}",
-                    id, term, snap.term
+                    id, term, snap.included.term
                 );
                 assert_eq!(
                     &snap.membership, cfg,

diff --git a/memstore/src/lib.rs b/memstore/src/lib.rs
@@ -352,8 +352,7 @@ impl RaftStorage<ClientRequest, ClientResponse> for MemStore {
 
         tracing::trace!({ snapshot_size = snapshot_bytes.len() }, "log compaction complete");
         Ok(CurrentSnapshotData {
-            term,
-            index: last_applied_log,
+            included: (term, last_applied_log).into(),
             membership: membership_config.clone(),
             snapshot_id,
             snapshot: Box::new(Cursor::new(snapshot_bytes)),
@@ -423,8 +422,7 @@ impl RaftStorage<ClientRequest, ClientResponse> for MemStore {
             Some(snapshot) => {
                 let reader = serde_json::to_vec(&snapshot)?;
                 Ok(Some(CurrentSnapshotData {
-                    index: snapshot.index,
-                    term: snapshot.term,
+                    included: (snapshot.term, snapshot.index).into(),
                     membership: snapshot.membership.clone(),
                     snapshot_id: snapshot.snapshot_id.clone(),
                     snapshot: Box::new(Cursor::new(reader)),