change: use LogId to track last applied instead of using just an index.

It provides more info by Using LogId to track last applied log.
E.g. when creating a snapshot, it need to walk through logs to find the
term of the last applied log, just like it did in memstore impl.

Using LogId{term, index} is a more natural way in every aspect.

changes: RaftCore: change type of `last_applied` from u64 to LogId.

Author: drmingdrmer

async-raft/src/core/append_entries.rs

@@ -9,6 +9,7 @@ use crate::raft::Entry;
 use crate::raft::EntryPayload;
 use crate::AppData;
 use crate::AppDataResponse;
+use crate::LogId;
 use crate::RaftNetwork;
 use crate::RaftStorage;
 use crate::Update;
@@ -209,7 +210,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             return;
         }
         // If we don't have any new entries to replicate, then do nothing.
-        if self.commit_index <= self.last_applied {
+        if self.commit_index <= self.last_applied.index {
             return;
         }
         // If we have no cached entries, then do nothing.
@@ -219,13 +220,13 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         };
 
         // Drain entries from the beginning of the cache up to commit index.
-        let mut last_entry_seen: Option<u64> = None;
+        let mut last_entry_seen: Option<LogId> = None;
         let entries: Vec<_> = (first_idx..=self.commit_index)
             .filter_map(|idx| {
                 if let Some(entry) = self.entries_cache.remove(&idx) {
-                    last_entry_seen = Some(entry.log_id.index);
+                    last_entry_seen = Some(entry.log_id);
                     match entry.payload {
-                        EntryPayload::Normal(inner) => Some((entry.log_id.index, inner.data)),
+                        EntryPayload::Normal(inner) => Some((entry.log_id, inner.data)),
                         _ => None,
                     }
                 } else {
@@ -236,8 +237,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
         // If we have no data entries to apply, then do nothing.
         if entries.is_empty() {
-            if let Some(index) = last_entry_seen {
-                self.last_applied = index;
+            if let Some(log_id) = last_entry_seen {
+                self.last_applied = log_id;
                 self.report_metrics(Update::Ignore);
             }
             return;
@@ -264,7 +265,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     #[tracing::instrument(level = "trace", skip(self))]
     async fn initial_replicate_to_state_machine(&mut self) {
         let stop = std::cmp::min(self.commit_index, self.last_log_id.index) + 1;
-        let start = self.last_applied + 1;
+        let start = self.last_applied.index + 1;
         let storage = self.storage.clone();
 
         // If we already have an active replication task, then do nothing.
@@ -274,15 +275,15 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
         // Fetch the series of entries which must be applied to the state machine, then apply them.
         let handle = tokio::spawn(async move {
-            let mut new_last_applied: Option<u64> = None;
+            let mut new_last_applied: Option<LogId> = None;
             let entries = storage.get_log_entries(start, stop).await?;
             if let Some(entry) = entries.last() {
-                new_last_applied = Some(entry.log_id.index);
+                new_last_applied = Some(entry.log_id);
             }
             let data_entries: Vec<_> = entries
                 .iter()
                 .filter_map(|entry| match &entry.payload {
-                    EntryPayload::Normal(inner) => Some((&entry.log_id.index, &inner.data)),
+                    EntryPayload::Normal(inner) => Some((&entry.log_id, &inner.data)),
                     _ => None,
                 })
                 .collect();

async-raft/src/core/client.rs

@@ -334,7 +334,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 }
             }
             ClientOrInternalResponseTx::Internal(tx) => {
-                self.core.last_applied = req.entry.log_id.index;
+                self.core.last_applied = req.entry.log_id;
                 self.leader_report_metrics();
                 let _ = tx.send(Ok(req.entry.log_id.index));
             }
@@ -355,7 +355,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
 
         let index = log_id.index;
 
-        let expected_next_index = self.core.last_applied + 1;
+        let expected_next_index = self.core.last_applied.index + 1;
         if index != expected_next_index {
             let entries = self
                 .core
@@ -365,13 +365,13 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 .map_err(|err| self.core.map_fatal_storage_error(err))?;
 
             if let Some(entry) = entries.last() {
-                self.core.last_applied = entry.log_id.index;
+                self.core.last_applied = entry.log_id;
             }
 
             let data_entries: Vec<_> = entries
                 .iter()
                 .filter_map(|entry| match &entry.payload {
-                    EntryPayload::Normal(inner) => Some((&entry.log_id.index, &inner.data)),
+                    EntryPayload::Normal(inner) => Some((&entry.log_id, &inner.data)),
                     _ => None,
                 })
                 .collect();
@@ -393,7 +393,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             }
         }
         // Apply this entry to the state machine and return its data response.
-        let res = self.core.storage.apply_entry_to_state_machine(&index, entry).await.map_err(|err| {
+        let res = self.core.storage.apply_entry_to_state_machine(&log_id, entry).await.map_err(|err| {
             if err.downcast_ref::<S::ShutdownError>().is_some() {
                 // If this is an instance of the storage impl's shutdown error, then trigger shutdown.
                 self.core.map_fatal_storage_error(err)
@@ -402,7 +402,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 RaftError::RaftStorage(err)
             }
         });
-        self.core.last_applied = index;
+        self.core.last_applied = *log_id;
         self.leader_report_metrics();
         res
     }

async-raft/src/core/install_snapshot.rs

@@ -174,7 +174,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         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.meta.last_log_id;
-        self.last_applied = req.meta.last_log_id.index;
+        self.last_applied = req.meta.last_log_id;
         self.snapshot_last_log_id = req.meta.last_log_id;
         self.report_metrics(Update::Ignore);
         Ok(())

async-raft/src/core/mod.rs

@@ -84,13 +84,15 @@ pub struct RaftCore<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftSt
     /// Is initialized to 0, and increases monotonically. This is always based on the leader's
     /// commit index which is communicated to other members via the AppendEntries protocol.
     commit_index: u64,
-    /// The index of the highest log entry which has been applied to the local state machine.
+
+    /// The log id of the highest log entry which has been applied to the local state machine.
     ///
-    /// Is initialized to 0 for a pristine node; else, for nodes with existing state it is
+    /// Is initialized to 0,0 for a pristine node; else, for nodes with existing state it is
     /// is initialized to the value returned from the `RaftStorage::get_initial_state` on startup.
     /// This value increases following the `commit_index` as logs are applied to the state
     /// machine (via the storage interface).
-    last_applied: u64,
+    last_applied: LogId,
+
     /// The current term.
     ///
     /// Is initialized to 0 on first boot, and increases monotonically. This is normally based on
@@ -130,7 +132,7 @@ pub struct RaftCore<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftSt
     /// This abstraction is needed to ensure that replicating to the state machine does not block
     /// the AppendEntries RPC flow, and to ensure that we have a smooth transition to becoming
     /// leader without concern over duplicate application of entries to the state machine.
-    replicate_to_sm_handle: FuturesOrdered<JoinHandle<anyhow::Result<Option<u64>>>>,
+    replicate_to_sm_handle: FuturesOrdered<JoinHandle<anyhow::Result<Option<LogId>>>>,
     /// A bool indicating if this system has performed its initial replication of
     /// outstanding entries to the state machine.
     has_completed_initial_replication_to_sm: bool,
@@ -168,7 +170,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             storage,
             target_state: State::Follower,
             commit_index: 0,
-            last_applied: 0,
+            last_applied: LogId { term: 0, index: 0 },
             current_term: 0,
             current_leader: None,
             voted_for: None,
@@ -281,7 +283,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             state: self.target_state,
             current_term: self.current_term,
             last_log_index: self.last_log_id.index,
-            last_applied: self.last_applied,
+            last_applied: self.last_applied.index,
             current_leader: self.current_leader,
             membership_config: self.membership.clone(),
             snapshot: self.snapshot_last_log_id,
@@ -425,13 +427,13 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         }
         let SnapshotPolicy::LogsSinceLast(threshold) = &self.config.snapshot_policy;
         // Check to ensure we have actual entries for compaction.
-        if self.last_applied == 0 || self.last_applied < self.snapshot_last_log_id.index {
+        if self.last_applied.index == 0 || self.last_applied.index < self.snapshot_last_log_id.index {
             return;
         }
 
         if !force {
             // If we are below the threshold, then there is nothing to do.
-            if self.last_applied < self.snapshot_last_log_id.index + *threshold {
+            if self.last_applied.index < self.snapshot_last_log_id.index + *threshold {
                 return;
             }
         }
@@ -471,7 +473,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
     /// Handle the output of an async task replicating entries to the state machine.
     #[tracing::instrument(level = "trace", skip(self, res))]
-    pub(self) fn handle_replicate_to_sm_result(&mut self, res: anyhow::Result<Option<u64>>) -> RaftResult<()> {
+    pub(self) fn handle_replicate_to_sm_result(&mut self, res: anyhow::Result<Option<LogId>>) -> RaftResult<()> {
         let last_applied_opt = res.map_err(|err| self.map_fatal_storage_error(err))?;
         if let Some(last_applied) = last_applied_opt {
             self.last_applied = last_applied;

async-raft/src/storage.rs

@@ -58,8 +58,8 @@ pub struct InitialState {
     /// The last entry.
     pub last_log_id: LogId,
 
-    /// The index of the last log applied to the state machine.
-    pub last_applied_log: u64,
+    /// The LogId of the last log applied to the state machine.
+    pub last_applied_log: LogId,
     /// The saved hard state of the node.
     pub hard_state: HardState,
     /// The latest cluster membership configuration found in the log, else a new initial
@@ -75,7 +75,7 @@ impl InitialState {
     pub fn new_initial(id: NodeId) -> Self {
         Self {
             last_log_id: LogId { term: 0, index: 0 },
-            last_applied_log: 0,
+            last_applied_log: LogId { term: 0, index: 0 },
             hard_state: HardState {
                 current_term: 0,
                 voted_for: None,
@@ -186,15 +186,15 @@ where
     ///
     /// It is important to note that even in cases where an application specific error is returned,
     /// implementations should still record that the entry has been applied to the state machine.
-    async fn apply_entry_to_state_machine(&self, index: &u64, data: &D) -> Result<R>;
+    async fn apply_entry_to_state_machine(&self, index: &LogId, data: &D) -> Result<R>;
 
     /// Apply the given payload of entries to the state machine, as part of replication.
     ///
     /// The Raft protocol guarantees that only logs which have been _committed_, that is, logs which
     /// have been replicated to a majority of the cluster, will be applied to the state machine.
     ///
     /// Errors returned from this method will cause Raft to go into shutdown.
-    async fn replicate_to_state_machine(&self, entries: &[(&u64, &D)]) -> Result<()>;
+    async fn replicate_to_state_machine(&self, entries: &[(&LogId, &D)]) -> Result<()>;
 
     /// Perform log compaction, returning a handle to the generated snapshot.
     ///

async-raft/tests/client_writes.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use anyhow::Result;
 use async_raft::raft::MembershipConfig;
 use async_raft::Config;
+use async_raft::LogId;
 use async_raft::State;
 use fixtures::RaftRouter;
 use futures::prelude::*;
@@ -67,7 +68,7 @@ async fn client_writes() -> Result<()> {
             1,
             want,
             Some(0),
-            want,
+            LogId { term: 1, index: want },
             Some(((5000..5100).into(), 1, MembershipConfig {
                 members: hashset![0, 1, 2],
                 members_after_consensus: None,

async-raft/tests/compaction.rs

@@ -65,7 +65,7 @@ async fn compaction() -> Result<()> {
             1,
             want,
             Some(0),
-            want,
+            LogId { term: 1, index: want },
             Some((want.into(), 1, MembershipConfig {
                 members: hashset![0],
                 members_after_consensus: None,
@@ -87,7 +87,15 @@ async fn compaction() -> Result<()> {
         members: hashset![0u64],
         members_after_consensus: None,
     }));
-    router.assert_storage_state(1, want, None /* non-voter does not vote */, want, expected_snap).await;
+    router
+        .assert_storage_state(
+            1,
+            want,
+            None, /* non-voter does not vote */
+            LogId { term: 1, index: want },
+            expected_snap,
+        )
+        .await;
 
     Ok(())
 }

async-raft/tests/fixtures/mod.rs

@@ -460,7 +460,7 @@ impl RaftRouter {
         expect_term: u64,
         expect_last_log: u64,
         expect_voted_for: Option<u64>,
-        expect_sm_last_applied_log: u64,
+        expect_sm_last_applied_log: LogId,
         expect_snapshot: Option<(ValueTest<u64>, u64, MembershipConfig)>,
     ) {
         let rt = self.routing_table.read().await;

async-raft/tests/singlenode.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use async_raft::Config;
+use async_raft::LogId;
 use async_raft::State;
 use fixtures::RaftRouter;
 use maplit::hashset;
@@ -46,7 +47,7 @@ async fn singlenode() -> Result<()> {
     // Write some data to the single node cluster.
     router.client_request_many(0, "0", 1000).await;
     router.assert_stable_cluster(Some(1), Some(1001)).await;
-    router.assert_storage_state(1, 1001, Some(0), 1001, None).await;
+    router.assert_storage_state(1, 1001, Some(0), LogId { term: 1, index: 1001 }, None).await;
 
     // Read some data from the single node cluster.
     router.client_read(0).await?;

async-raft/tests/snapshot_chunk_size.rs

@@ -63,7 +63,7 @@ async fn snapshot_chunk_size() -> Result<()> {
 
         router.wait_for_log(&hashset![0], want, None, "send log to trigger snapshot").await?;
         router.wait_for_snapshot(&hashset![0], LogId { term: 1, index: want }, None, "snapshot").await?;
-        router.assert_storage_state(1, want, Some(0), want, want_snap).await;
+        router.assert_storage_state(1, want, Some(0), LogId { term: 1, index: want }, want_snap).await;
     }
 
     tracing::info!("--- add non-voter to receive snapshot and logs");
@@ -78,7 +78,15 @@ async fn snapshot_chunk_size() -> Result<()> {
 
         router.wait_for_log(&hashset![0, 1], want, None, "add non-voter").await?;
         router.wait_for_snapshot(&hashset![1], LogId { term: 1, index: want }, None, "").await?;
-        router.assert_storage_state(1, want, None /* non-voter does not vote */, want, want_snap).await;
+        router
+            .assert_storage_state(
+                1,
+                want,
+                None, /* non-voter does not vote */
+                LogId { term: 1, index: want },
+                want_snap,
+            )
+            .await;
     }
 
     Ok(())

async-raft/tests/snapshot_ge_half_threshold.rs

@@ -68,7 +68,7 @@ async fn snapshot_ge_half_threshold() -> Result<()> {
                 1,
                 want,
                 Some(0),
-                want,
+                LogId { term: 1, index: want },
                 Some((want.into(), 1, MembershipConfig {
                     members: hashset![0],
                     members_after_consensus: None,
@@ -94,7 +94,15 @@ async fn snapshot_ge_half_threshold() -> Result<()> {
             members_after_consensus: None,
         }));
         router.wait_for_snapshot(&hashset![1], LogId { term: 1, index: want }, None, "").await?;
-        router.assert_storage_state(1, want, None /* non-voter does not vote */, want, expected_snap).await;
+        router
+            .assert_storage_state(
+                1,
+                want,
+                None, /* non-voter does not vote */
+                LogId { term: 1, index: want },
+                expected_snap,
+            )
+            .await;
     }
 
     Ok(())

async-raft/tests/snapshot_overrides_membership.rs

@@ -70,7 +70,7 @@ async fn snapshot_overrides_membership() -> Result<()> {
                 1,
                 want,
                 Some(0),
-                want,
+                LogId { term: 1, index: want },
                 Some((want.into(), 1, MembershipConfig {
                     members: hashset![0],
                     members_after_consensus: None,
@@ -127,7 +127,15 @@ async fn snapshot_overrides_membership() -> Result<()> {
                 members_after_consensus: None,
             }));
             router.wait_for_snapshot(&hashset![1], LogId { term: 1, index: want }, None, "").await?;
-            router.assert_storage_state(1, want, None /* non-voter does not vote */, want, expected_snap).await;
+            router
+                .assert_storage_state(
+                    1,
+                    want,
+                    None, /* non-voter does not vote */
+                    LogId { term: 1, index: want },
+                    expected_snap,
+                )
+                .await;
 
             let m = sto.get_membership_config().await?;
             assert_eq!(