Change: use term, node_id, index to identify a log entry

Simplify election: in a `term`, it allows to elect more than one leader
and the one with greatest `Vote`, i.e., `term, node_id` wins.

Only the last established leader is valid.

A node is able to grant more than one vote within one term.
A greater `Vote` will always be granted.

This modification turns the `Vote` from partial order to a total order
value. Thus the vote granting is simple as comparing two value.

- Change: Vote now is a tuple of `term, node_id` and is compared in
  dictionary order.

  Greater Vote wins.

  There is another field `committed` in `Vote`, which is not necessary
  to be persisted by storage layer.
  Saving it won't affect correctness.

- Add: LeaderId into LogId, LogId is identified by `term, node_id, index`.

Author: drmingdrmer

memstore/src/lib.rs

@@ -292,7 +292,10 @@ impl RaftStorage<ClientRequest, ClientResponse> for MemStore {
             *l
         };
 
-        let snapshot_id = format!("{}-{}-{}", last_applied_log.term, last_applied_log.index, snapshot_idx);
+        let snapshot_id = format!(
+            "{}-{}-{}",
+            last_applied_log.leader_id, last_applied_log.index, snapshot_idx
+        );
 
         let meta = SnapshotMeta {
             last_log_id: last_applied_log,

openraft/src/core/admin.rs

@@ -38,7 +38,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
 
         if self.core.last_log_id.is_some() || self.core.vote.term != 0 {
             tracing::error!(
-                last_log_id=?self.core.last_log_id, %self.core.vote,
+                last_log_id=?self.core.last_log_id, ?self.core.vote,
                 "rejecting init_with_config request as last_log_index is not None or current_term is not 0");
             return Err(InitializeError::NotAllowed);
         }

openraft/src/core/append_entries.rs

@@ -31,9 +31,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         let msg_entries = req.entries.as_slice();
 
         // Partial order compare: smaller than or incomparable
-        #[allow(clippy::neg_cmp_op_on_partial_ord)]
-        if !(req.vote >= self.vote) {
-            tracing::debug!(%self.vote, %req.vote, "AppendEntries RPC term is less than current term");
+        if req.vote < self.vote {
+            tracing::debug!(?self.vote, %req.vote, "AppendEntries RPC term is less than current term");
 
             return Ok(AppendEntriesResponse {
                 vote: self.vote,

openraft/src/core/client.rs

@@ -23,7 +23,6 @@ use crate::raft::Entry;
 use crate::raft::EntryPayload;
 use crate::raft::RaftRespTx;
 use crate::replication::RaftEvent;
-use crate::vote::Vote;
 use crate::AppData;
 use crate::AppDataResponse;
 use crate::MessageSummary;
@@ -157,8 +156,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             };
 
             // If we receive a response with a greater term, then revert to follower and abort this request.
-            if data.vote.term != self.core.vote.term {
-                self.core.vote = Vote::new_uncommitted(data.vote.term, None);
+            if data.vote > self.core.vote {
+                self.core.vote = data.vote;
                 // TODO(xp): deal with storage error
                 self.core.save_vote().await.unwrap();
                 // TODO(xp): if receives error about a higher term, it should stop at once?

openraft/src/core/install_snapshot.rs

@@ -35,9 +35,8 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         &mut self,
         req: InstallSnapshotRequest,
     ) -> Result<InstallSnapshotResponse, InstallSnapshotError> {
-        #[allow(clippy::neg_cmp_op_on_partial_ord)]
-        if !(req.vote >= self.vote) {
-            tracing::debug!(%self.vote, %req.vote, "InstallSnapshot RPC term is less than current term");
+        if req.vote < self.vote {
+            tracing::debug!(?self.vote, %req.vote, "InstallSnapshot RPC term is less than current term");
 
             return Ok(InstallSnapshotResponse { vote: self.vote });
         }

openraft/src/core/mod.rs

@@ -55,6 +55,7 @@ use crate::replication::ReplicationStream;
 use crate::vote::Vote;
 use crate::AppData;
 use crate::AppDataResponse;
+use crate::LeaderId;
 use crate::LogId;
 use crate::Membership;
 use crate::MessageSummary;
@@ -82,7 +83,8 @@ pub struct EffectiveMembership {
 impl EffectiveMembership {
     pub fn new_initial(node_id: u64) -> Self {
         EffectiveMembership {
-            log_id: LogId::new(0, 0),
+            // TODO(xp): avoid using Vote::default()
+            log_id: LogId::new(LeaderId::default(), 0),
             membership: Membership::new_initial(node_id),
         }
     }
@@ -182,7 +184,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             target_state: State::Follower,
             committed: None,
             last_applied: None,
-            vote: Vote::new_uncommitted(0, None),
+            vote: Vote::default(),
             last_log_id: None,
             snapshot_state: None,
             snapshot_last_log_id: None,
@@ -519,7 +521,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
     #[tracing::instrument(level = "debug", skip(self, payload))]
     pub(super) async fn append_payload_to_log(&mut self, payload: EntryPayload<D>) -> Result<Entry<D>, StorageError> {
-        let log_id = LogId::new(self.vote.term, self.last_log_id.next_index());
+        let log_id = LogId::new(self.vote.leader_id(), self.last_log_id.next_index());
 
         let entry = Entry { log_id, payload };
         self.storage.append_to_log(&[&entry]).await?;
@@ -539,20 +541,20 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
 
     #[tracing::instrument(level = "debug", skip(self))]
     pub fn current_leader(&self) -> Option<NodeId> {
-        let l = self.vote.leader();
-
-        if let Some(id) = l {
-            if id == self.id {
-                if self.target_state == State::Leader {
-                    Some(id)
-                } else {
-                    None
-                }
-            } else {
+        if !self.vote.committed {
+            return None;
+        }
+
+        let id = self.vote.node_id;
+
+        if id == self.id {
+            if self.target_state == State::Leader {
                 Some(id)
+            } else {
+                None
             }
         } else {
-            None
+            Some(id)
         }
     }
 }
@@ -724,6 +726,11 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     /// Transition to the Raft leader state.
     #[tracing::instrument(level="debug", skip(self), fields(id=self.core.id, raft_state="leader"))]
     pub(self) async fn run(mut self) -> Result<(), Fatal> {
+        // Setup state as leader.
+        self.core.last_heartbeat = None;
+        self.core.next_election_timeout = None;
+        self.core.vote.commit();
+
         // Spawn replication streams.
         let targets = self
             .core
@@ -746,11 +753,6 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             self.nodes.insert(*node_id, state);
         }
 
-        // Setup state as leader.
-        self.core.last_heartbeat = None;
-        self.core.next_election_timeout = None;
-        self.core.vote = Vote::new_committed(self.core.vote.term, self.core.id);
-
         self.leader_report_metrics();
 
         self.commit_initial_leader_entry().await?;
@@ -904,7 +906,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             // Setup new term.
             self.core.update_next_election_timeout(false); // Generates a new rand value within range.
 
-            self.core.vote = Vote::new_uncommitted(self.core.vote.term + 1, Some(self.core.id));
+            self.core.vote = Vote::new(self.core.vote.term + 1, self.core.id);
 
             self.core.save_vote().await?;
             self.core.report_metrics(Update::Update(None));

openraft/src/core/replication.rs

@@ -35,9 +35,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
         caller_tx: Option<RaftRespTx<AddLearnerResponse, AddLearnerError>>,
     ) -> ReplicationState {
         let repl_stream = ReplicationStream::new(
-            self.core.id,
             target,
-            self.core.vote.term,
+            self.core.vote,
             self.core.config.clone(),
             self.core.last_log_id,
             self.core.committed,
@@ -60,8 +59,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
         event: ReplicaEvent<S::SnapshotData>,
     ) -> Result<(), StorageError> {
         match event {
-            ReplicaEvent::RevertToFollower { target, term } => {
-                self.handle_revert_to_follower(target, term).await?;
+            ReplicaEvent::RevertToFollower { target, vote } => {
+                self.handle_revert_to_follower(target, vote).await?;
             }
             ReplicaEvent::UpdateMatched { target, matched } => {
                 self.handle_update_matched(target, matched).await?;
@@ -82,10 +81,10 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     }
 
     /// Handle events from replication streams for when this node needs to revert to follower state.
-    #[tracing::instrument(level = "trace", skip(self, term))]
-    async fn handle_revert_to_follower(&mut self, _: NodeId, term: u64) -> Result<(), StorageError> {
-        if term > self.core.vote.term {
-            self.core.vote = Vote::new_uncommitted(term, None);
+    #[tracing::instrument(level = "trace", skip(self))]
+    async fn handle_revert_to_follower(&mut self, _: NodeId, vote: Vote) -> Result<(), StorageError> {
+        if vote > self.core.vote {
+            self.core.vote = vote;
             self.core.save_vote().await?;
             self.core.set_target_state(State::Follower);
         }
@@ -208,7 +207,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 let Some(log_id) = matched {
-                if log_id.term == self.core.vote.term {
+                if log_id.leader_id == self.core.vote.leader_id() {
                     res.insert(*id, log_id);
                 }
             }

openraft/src/core/replication_state_test.rs

@@ -1,10 +1,11 @@
 use crate::core::is_matched_upto_date;
 use crate::Config;
+use crate::LeaderId;
 use crate::LogId;
 
 #[test]
 fn test_is_line_rate() -> anyhow::Result<()> {
-    let m = Some(LogId { term: 1, index: 10 });
+    let m = Some(LogId::new(LeaderId::new(1, 0), 10));
 
     let cfg = |n| Config {
         replication_lag_threshold: n,
@@ -13,33 +14,47 @@ fn test_is_line_rate() -> anyhow::Result<()> {
 
     assert!(is_matched_upto_date(&None, &None, &cfg(0)), "matched, threshold=0");
     assert!(
-        is_matched_upto_date(&None, &Some(LogId { term: 2, index: 0 }), &cfg(1)),
+        is_matched_upto_date(
+            &None,
+            &Some(LogId {
+                leader_id: LeaderId::new(2, 0),
+                index: 0
+            }),
+            &cfg(1)
+        ),
         "matched, threshold=1"
     );
     assert!(
-        !is_matched_upto_date(&None, &Some(LogId { term: 2, index: 0 }), &cfg(0)),
+        !is_matched_upto_date(
+            &None,
+            &Some(LogId {
+                leader_id: LeaderId::new(2, 0),
+                index: 0
+            }),
+            &cfg(0)
+        ),
         "not matched, threshold=1"
     );
 
     assert!(
-        is_matched_upto_date(&Some(LogId::new(0, 0)), &None, &cfg(0)),
+        is_matched_upto_date(&Some(LogId::new(LeaderId::new(0, 0), 0)), &None, &cfg(0)),
         "matched, threshold=0"
     );
 
     assert!(
-        is_matched_upto_date(&m, &Some(LogId { term: 2, index: 10 }), &cfg(0)),
+        is_matched_upto_date(&m, &Some(LogId::new(LeaderId::new(2, 0), 10)), &cfg(0)),
         "matched, threshold=0"
     );
     assert!(
-        is_matched_upto_date(&m, &Some(LogId { term: 2, index: 9 }), &cfg(0)),
+        is_matched_upto_date(&m, &Some(LogId::new(LeaderId::new(2, 0), 9)), &cfg(0)),
         "overflow, threshold=0"
     );
     assert!(
-        !is_matched_upto_date(&m, &Some(LogId { term: 2, index: 11 }), &cfg(0)),
+        !is_matched_upto_date(&m, &Some(LogId::new(LeaderId::new(2, 0), 11)), &cfg(0)),
         "not caught up, threshold=0"
     );
     assert!(
-        is_matched_upto_date(&m, &Some(LogId { term: 2, index: 11 }), &cfg(1)),
+        is_matched_upto_date(&m, &Some(LogId::new(LeaderId::new(2, 0), 11)), &cfg(1)),
         "caught up, threshold=1"
     );
     Ok(())

openraft/src/core/vote.rs

@@ -9,7 +9,6 @@ use crate::error::VoteError;
 use crate::raft::VoteRequest;
 use crate::raft::VoteResponse;
 use crate::summary::MessageSummary;
-use crate::vote::Vote;
 use crate::AppData;
 use crate::AppDataResponse;
 use crate::NodeId;
@@ -25,16 +24,15 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     pub(super) async fn handle_vote_request(&mut self, req: VoteRequest) -> Result<VoteResponse, VoteError> {
         tracing::debug!(
             %req.vote,
-            %self.vote,
+            ?self.vote,
             "start handle_vote_request"
         );
         let last_log_id = self.last_log_id;
 
-        #[allow(clippy::neg_cmp_op_on_partial_ord)]
-        if !(req.vote >= self.vote) {
+        if req.vote < self.vote {
             tracing::debug!(
                 %req.vote,
-                %self.vote,
+                ?self.vote,
                 "RequestVote RPC term is less than current term"
             );
             return Ok(VoteResponse {
@@ -62,15 +60,6 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             }
         }
 
-        // Always save a higher term
-        if req.vote.term > self.vote.term {
-            self.update_next_election_timeout(false);
-            self.vote = Vote::new_uncommitted(req.vote.term, None);
-            self.save_vote().await?;
-
-            self.set_target_state(State::Follower);
-        }
-
         // Check if candidate's log is at least as up-to-date as this node's.
         // If candidate's log is not at least as up-to-date as this node, then reject.
         if req.last_log_id < last_log_id {
@@ -107,8 +96,8 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
     pub(super) async fn handle_vote_response(&mut self, res: VoteResponse, target: NodeId) -> Result<(), StorageError> {
         // If peer's term is greater than current term, revert to follower state.
 
-        if res.vote.term > self.core.vote.term {
-            self.core.vote = Vote::new_uncommitted(res.vote.term, None);
+        if res.vote > self.core.vote {
+            self.core.vote = res.vote;
             self.core.save_vote().await?;
 
             // If a quorum of nodes have higher `last_log_id`, I have no chance to become a leader.
@@ -121,7 +110,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             } else {
                 tracing::debug!(
                     id = %self.core.id,
-                    %self.core.vote,
+                    ?self.core.vote,
                     %res.vote,
                     self_last_log_id=?self.core.last_log_id,
                     res_last_log_id=?res.last_log_id,

openraft/src/error.rs

@@ -14,6 +14,7 @@ use crate::LogId;
 use crate::NodeId;
 use crate::RPCTypes;
 use crate::StorageError;
+use crate::Vote;
 
 /// Fatal is unrecoverable and shuts down raft at once.
 #[derive(Debug, Clone, thiserror::Error, PartialEq, Eq, Serialize, Deserialize)]
@@ -179,7 +180,7 @@ impl From<StorageError> for AddLearnerError {
 #[allow(clippy::large_enum_variant)]
 pub enum ReplicationError {
     #[error(transparent)]
-    HigherTerm(#[from] HigherTerm),
+    HigherVote(#[from] HigherVote),
 
     #[error("Replication is closed")]
     Closed,
@@ -231,10 +232,10 @@ impl<T: std::error::Error> RemoteError<T> {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
-#[error("seen a higher term: {higher} GT mine: {mine}")]
-pub struct HigherTerm {
-    pub higher: u64,
-    pub mine: u64,
+#[error("seen a higher vote: {higher} GT mine: {mine}")]
+pub struct HigherVote {
+    pub higher: Vote,
+    pub mine: Vote,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]

openraft/src/lib.rs

@@ -58,7 +58,7 @@ pub use crate::storage_error::Violation;
 pub use crate::store_ext::StoreExt;
 pub use crate::store_wrapper::Wrapper;
 pub use crate::summary::MessageSummary;
-pub use crate::vote::CommittedState;
+pub use crate::vote::LeaderId;
 pub use crate::vote::Vote;
 
 /// A Raft node's ID.