change: track committed log id instead of just a commit index

Author: drmingdrmer

async-raft/src/core/admin.rs

@@ -121,7 +121,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
 
         // The last membership config is not committed yet.
         // Can not process the next one.
-        if self.core.commit_index < self.core.effective_membership.log_id.index {
+        if self.core.committed < self.core.effective_membership.log_id {
             let _ = tx.send(Err(ClientWriteError::ChangeMembershipError(
                 ChangeMembershipError::InProgress {
                     membership_log_id: self.core.effective_membership.log_id,

async-raft/src/core/append_entries.rs

@@ -58,8 +58,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         //   Then it will update commit_index to 3 and apply {2,3}
 
         // TODO(xp): cleanup commit index at sender side.
-        let valid_commit_index = msg_entries.last().map(|x| x.log_id).unwrap_or_else(|| msg.prev_log_id).index;
-        let valid_commit_index = std::cmp::min(msg.leader_commit, valid_commit_index);
+        let valid_commit_index = msg_entries.last().map(|x| x.log_id).unwrap_or_else(|| msg.prev_log_id);
+        let valid_committed = std::cmp::min(msg.leader_commit, valid_commit_index);
 
         tracing::debug!("start to check and update to latest term/leader");
         {
@@ -94,7 +94,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         //              +----------------+------------------------+
         //              ` 0              ` last_applied           ` last_log_id
 
-        return self.append_apply_log_entries(&msg.prev_log_id, msg_entries, valid_commit_index).await;
+        return self.append_apply_log_entries(&msg.prev_log_id, msg_entries, valid_committed).await;
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -227,13 +227,13 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         &mut self,
         prev_log_id: &LogId,
         entries: &[Entry<D>],
-        commit_index: u64,
+        committed: LogId,
     ) -> RaftResult<AppendEntriesResponse> {
         let matching = self.does_log_id_match(prev_log_id).await?;
         tracing::debug!(
-            "check prev_log_id {} match: commit_index: {}, matches: {}",
+            "check prev_log_id {} match: committed: {}, matches: {}",
             prev_log_id,
-            self.commit_index,
+            self.committed,
             matching,
         );
 
@@ -263,7 +263,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         let (n_matching, entries) = self.skip_matching_entries(entries).await?;
 
         tracing::debug!(
-            commit_index = self.commit_index,
+            %self.committed,
             n_matching,
             entries = %entries.summary(),
             "skip matching entries",
@@ -278,7 +278,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
         // commit index must not > last_log_id.index
         // This is guaranteed by caller.
-        self.commit_index = commit_index;
+        self.committed = committed;
 
         self.replicate_to_state_machine_if_needed().await?;
 
@@ -306,7 +306,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             let log_id = entries[i].log_id;
             let index = log_id.index;
 
-            if index <= self.commit_index {
+            if index <= self.committed.index {
                 continue;
             }
 
@@ -333,7 +333,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         let index = remote_log_id.index;
 
         // Committed entries are always safe and are consistent to a valid leader.
-        if index <= self.commit_index {
+        if index <= self.committed.index {
             return Ok(true);
         }
 
@@ -410,10 +410,10 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         }
 
         // If we don't have any new entries to replicate, then do nothing.
-        if self.commit_index <= self.last_applied.index {
+        if self.committed <= self.last_applied {
             tracing::debug!(
-                "commit_index({}) <= last_applied({}), return",
-                self.commit_index,
+                "committed({}) <= last_applied({}), return",
+                self.committed,
                 self.last_applied
             );
             return Ok(());
@@ -423,7 +423,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
         let entries = self
             .storage
-            .get_log_entries(self.last_applied.index + 1..=self.commit_index)
+            .get_log_entries(self.last_applied.index + 1..=self.committed.index)
             .await
             .map_err(|e| self.map_storage_error(e))?;
 
@@ -452,11 +452,11 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     /// from the AppendEntries RPC handler.
     #[tracing::instrument(level = "debug", skip(self))]
     async fn initial_replicate_to_state_machine(&mut self) -> Result<(), RaftError> {
-        let stop = std::cmp::min(self.commit_index, self.last_log_id.index) + 1;
+        let stop = std::cmp::min(self.committed.index, self.last_log_id.index) + 1;
         let start = self.last_applied.index + 1;
         let storage = self.storage.clone();
 
-        tracing::debug!(start, stop, self.commit_index, %self.last_log_id, "start stop");
+        tracing::debug!(start, stop, %self.committed, %self.last_log_id, "start stop");
 
         // when self.commit_index is not initialized, e.g. the first heartbeat from leader always has a commit_index to
         // be 0, because the leader needs one round of heartbeat to find out the commit index.

async-raft/src/core/client.rs

@@ -139,7 +139,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 leader_id: self.core.id,
                 prev_log_id: node.matched,
                 entries: vec![],
-                leader_commit: self.core.commit_index,
+                leader_commit: self.core.committed,
             };
             let target = *id;
             let network = self.core.network.clone();
@@ -273,8 +273,9 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             self.awaiting_committed.push(req);
         } else {
             // Else, there are no voting nodes for replication, so the payload is now committed.
-            self.core.commit_index = entry_arc.log_id.index;
-            tracing::debug!(self.core.commit_index, "update commit index, no need to replicate");
+            self.core.committed = entry_arc.log_id;
+            tracing::debug!(%self.core.committed, "update committed, no need to replicate");
+
             self.leader_report_metrics();
             self.client_request_post_commit(req).await;
         }
@@ -283,7 +284,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             let _ = node.repl_stream.repl_tx.send((
                 RaftEvent::Replicate {
                     entry: entry_arc.clone(),
-                    commit_index: self.core.commit_index,
+                    committed: self.core.committed,
                 },
                 tracing::debug_span!("CH"),
             ));

async-raft/src/core/install_snapshot.rs

@@ -204,7 +204,6 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             .map_err(|e| self.map_storage_error(e))?;
 
         tracing::debug!("update after apply or install-snapshot: {:?}", changes);
-        println!("update after apply or install-snapshot: {:?}", changes);
 
         // After installing snapshot, no inconsistent log is removed.
         // This does not affect raft consistency.
@@ -219,16 +218,16 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             // snapshot is installed
             self.last_applied = last_applied;
 
-            if self.commit_index < self.last_applied.index {
-                self.commit_index = self.last_applied.index;
+            if self.committed < self.last_applied {
+                self.committed = self.last_applied;
             }
             if self.last_log_id < self.last_applied {
                 self.last_log_id = self.last_applied;
             }
 
             // There could be unknown membership in the snapshot.
             let membership = self.storage.get_membership_config().await.map_err(|err| self.map_storage_error(err))?;
-            println!("storage membership: {:?}", membership);
+            tracing::debug!("storage membership: {:?}", membership);
 
             self.update_membership(membership)?;
 

async-raft/src/core/mod.rs

@@ -103,12 +103,12 @@ pub struct RaftCore<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftSt
     /// The target state of the system.
     target_state: State,
 
-    /// The index of the last known committed entry.
+    /// The log id of the last known committed entry.
     ///
-    /// I.e.:
+    /// Committed means:
     /// - a log that is replicated to a quorum of the cluster and it is of the term of the leader.
     /// - A quorum could be a joint quorum.
-    commit_index: u64,
+    committed: LogId,
 
     /// The log id of the highest log entry which has been applied to the local state machine.
     last_applied: LogId,
@@ -182,14 +182,14 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             network,
             storage,
             target_state: State::Follower,
-            commit_index: 0,
-            last_applied: LogId { term: 0, index: 0 },
+            committed: LogId::new(0, 0),
+            last_applied: LogId::new(0, 0),
             current_term: 0,
             current_leader: None,
             voted_for: None,
-            last_log_id: LogId { term: 0, index: 0 },
+            last_log_id: LogId::new(0, 0),
             snapshot_state: None,
-            snapshot_last_log_id: LogId { term: 0, index: 0 },
+            snapshot_last_log_id: LogId::new(0, 0),
             has_completed_initial_replication_to_sm: false,
             last_heartbeat: None,
             next_election_timeout: None,
@@ -213,10 +213,11 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         self.voted_for = state.hard_state.voted_for;
         self.effective_membership = state.last_membership.clone();
         self.last_applied = state.last_applied;
+
         // NOTE: this is repeated here for clarity. It is unsafe to initialize the node's commit
         // index to any other value. The commit index must be determined by a leader after
         // successfully committing a new log to the cluster.
-        self.commit_index = 0;
+        self.committed = LogId::new(0, 0);
 
         // Fetch the most recent snapshot in the system.
         if let Some(snapshot) = self.storage.get_current_snapshot().await.map_err(|err| self.map_storage_error(err))? {

async-raft/src/core/replication.rs

@@ -40,7 +40,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             self.core.current_term,
             self.core.config.clone(),
             self.core.last_log_id,
-            self.core.commit_index,
+            self.core.committed,
             self.core.network.clone(),
             self.core.storage.clone(),
             self.replication_tx.clone(),
@@ -116,24 +116,24 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             self.update_leader_metrics(target, matched);
         }
 
-        if matched.index <= self.core.commit_index {
+        if matched <= self.core.committed {
             self.leader_report_metrics();
             return Ok(());
         }
 
-        let commit_index = self.calc_commit_index();
+        let commit_index = self.calc_commit_log_id();
 
         // Determine if we have a new commit index, accounting for joint consensus.
         // If a new commit index has been established, then update a few needed elements.
 
-        if commit_index > self.core.commit_index {
-            self.core.commit_index = commit_index;
+        if commit_index > self.core.committed {
+            self.core.committed = commit_index;
 
             // Update all replication streams based on new commit index.
             for node in self.nodes.values() {
                 let _ = node.repl_stream.repl_tx.send((
-                    RaftEvent::UpdateCommitIndex {
-                        commit_index: self.core.commit_index,
+                    RaftEvent::UpdateCommittedLogId {
+                        committed: self.core.committed,
                     },
                     tracing::debug_span!("CH"),
                 ));
@@ -144,7 +144,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
                 .awaiting_committed
                 .iter()
                 .enumerate()
-                .take_while(|(_idx, elem)| elem.entry.log_id.index <= self.core.commit_index)
+                .take_while(|(_idx, elem)| elem.entry.log_id <= self.core.committed)
                 .last()
                 .map(|(idx, _)| idx);
 
@@ -168,16 +168,16 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     }
 
     #[tracing::instrument(level = "trace", skip(self))]
-    fn calc_commit_index(&self) -> u64 {
+    fn calc_commit_log_id(&self) -> LogId {
         let repl_indexes = self.get_match_log_indexes();
 
         let committed = self.core.effective_membership.membership.greatest_majority_value(&repl_indexes);
 
-        *committed.unwrap_or(&self.core.commit_index)
+        *committed.unwrap_or(&self.core.committed)
     }
 
     /// Collect indexes of the greatest matching log on every replica(include the leader itself)
-    fn get_match_log_indexes(&self) -> BTreeMap<NodeId, u64> {
+    fn get_match_log_indexes(&self) -> BTreeMap<NodeId, LogId> {
         let node_ids = self.core.effective_membership.membership.all_nodes();
 
         let mut res = BTreeMap::new();
@@ -194,7 +194,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             // and that new leader may overrides any lower term logs.
             // Thus it is not considered as committed.
             if matched.term == self.core.current_term {
-                res.insert(*id, matched.index);
+                res.insert(*id, matched);
             }
         }
 

async-raft/src/raft.rs

@@ -501,8 +501,8 @@ pub struct AppendEntriesRequest<D: AppData> {
     #[serde(bound = "D: AppData")]
     pub entries: Vec<Entry<D>>,
 
-    /// The leader's commit index.
-    pub leader_commit: u64,
+    /// The leader's committed log id.
+    pub leader_commit: LogId,
 }
 
 impl<D: AppData> MessageSummary for AppendEntriesRequest<D> {

async-raft/src/raft_types.rs

@@ -14,7 +14,7 @@ pub struct LogId {
 
 impl From<(u64, u64)> for LogId {
     fn from(v: (u64, u64)) -> Self {
-        LogId { term: v.0, index: v.1 }
+        LogId::new(v.0, v.1)
     }
 }
 
@@ -26,6 +26,10 @@ impl Display for LogId {
 
 impl LogId {
     pub fn new(term: u64, index: u64) -> Self {
+        if term == 0 || index == 0 {
+            assert_eq!(index, 0, "zero-th log entry must be (0,0), but {}, {}", term, index);
+            assert_eq!(term, 0, "zero-th log entry must be (0,0), but {}, {}", term, index);
+        }
         LogId { term, index }
     }
 }