Refactor: introducing CommittedVote, NonCommittedVote, and RefVote types

- **Introduce Specialized Vote Types:**
  - Added `CommittedVote` and `NonCommittedVote` to distinguish between votes that have been
    accepted by a quorum and those that have not.
  - Introduced `RefVote` to provide a reference-based representation of votes, facilitating safer
    and more efficient comparisons.

- **Update Vote Management Logic:**
  - Replaced direct usage of `Vote` with `CommittedVote` and `NonCommittedVote` across various
    modules (`heartbeat/worker.rs`, `notification.rs`, `raft_core.rs`, etc.) to enhance type
    safety and clarity.
  - Modified comparison logic to leverage the new vote types, ensuring accurate and meaningful
    ordering based on vote status.

- **Enhance Notification Mechanisms:**
  - Updated notification structures to use the new vote types, improving the accuracy of messages
    related to vote responses and higher votes observed.

openraft/src/core/heartbeat/worker.rs

@@ -1,5 +1,4 @@
 use std::fmt;
-use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -83,7 +82,7 @@ where
             let option = RPCOption::new(timeout);
 
             let payload = AppendEntriesRequest {
-                vote: *heartbeat.session_id.leader_vote.deref(),
+                vote: heartbeat.session_id.leader_vote.into_vote(),
                 prev_log_id: None,
                 leader_commit: heartbeat.committed,
                 entries: vec![],

openraft/src/core/notification.rs

@@ -7,6 +7,8 @@ use crate::raft_state::IOId;
 use crate::replication;
 use crate::replication::ReplicationSessionId;
 use crate::type_config::alias::InstantOf;
+use crate::vote::CommittedVote;
+use crate::vote::NonCommittedVote;
 use crate::RaftTypeConfig;
 use crate::StorageError;
 use crate::Vote;
@@ -22,21 +24,19 @@ where C: RaftTypeConfig
         /// The candidate that sent the vote request.
         ///
         /// A vote identifies a unique server state.
-        sender_vote: Vote<C::NodeId>,
+        candidate_vote: NonCommittedVote<C>,
     },
 
-    /// Seen a higher `vote`.
+    /// A Leader sees a higher `vote` when replicating.
     HigherVote {
         /// The ID of the target node from which the new term was observed.
         target: C::NodeId,
 
         /// The higher vote observed.
         higher: Vote<C::NodeId>,
 
-        /// The candidate or leader that sent the vote request.
-        ///
-        /// A vote identifies a unique server state.
-        sender_vote: Vote<C::NodeId>,
+        /// The Leader that sent replication request.
+        leader_vote: CommittedVote<C>,
         // TODO: need this?
         // /// The cluster this replication works for.
         // membership_log_id: Option<LogId<C::NodeId>>,
@@ -84,18 +84,18 @@ where C: RaftTypeConfig
             Self::VoteResponse {
                 target,
                 resp,
-                sender_vote,
+                candidate_vote,
             } => {
                 write!(
                     f,
-                    "VoteResponse: from target={}, to sender_vote: {}, {}",
-                    target, sender_vote, resp
+                    "VoteResponse: from target={}, to candidate_vote: {}, {}",
+                    target, candidate_vote, resp
                 )
             }
             Self::HigherVote {
                 ref target,
                 higher: ref new_vote,
-                sender_vote: ref vote,
+                leader_vote: ref vote,
             } => {
                 write!(
                     f,

openraft/src/core/raft_core.rs

@@ -2,7 +2,6 @@ use std::borrow::Borrow;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::fmt::Debug;
-use std::ops::Deref;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
@@ -93,14 +92,16 @@ use crate::type_config::alias::ResponderOf;
 use crate::type_config::alias::WatchSenderOf;
 use crate::type_config::async_runtime::MpscUnboundedReceiver;
 use crate::type_config::TypeConfigExt;
+use crate::vote::vote_status::VoteStatus;
+use crate::vote::CommittedVote;
+use crate::vote::NonCommittedVote;
 use crate::ChangeMembers;
 use crate::Instant;
 use crate::LogId;
 use crate::Membership;
 use crate::OptionalSend;
 use crate::RaftTypeConfig;
 use crate::StorageError;
-use crate::Vote;
 
 /// A temp struct to hold the data for a node that is being applied.
 #[derive(Debug)]
@@ -393,7 +394,7 @@ where
                     let send_res = core_tx.send(Notification::HigherVote {
                         target,
                         higher: vote,
-                        sender_vote: my_vote,
+                        leader_vote: my_vote.into_committed(),
                     });
 
                     if let Err(_e) = send_res {
@@ -566,7 +567,7 @@ where
 
             // --- data ---
             current_term: st.vote_ref().leader_id().get_term(),
-            vote: st.io_state().io_progress.flushed().map(|x| *x.vote_ref()).unwrap_or_default(),
+            vote: st.io_state().io_progress.flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
             last_log_index: st.last_log_id().index(),
             last_applied: st.io_applied().copied(),
             snapshot: st.io_snapshot_last_log_id().copied(),
@@ -598,7 +599,7 @@ where
 
         let server_metrics = RaftServerMetrics {
             id: self.id,
-            vote: st.io_state().io_progress.flushed().map(|x| *x.vote_ref()).unwrap_or_default(),
+            vote: st.io_state().io_progress.flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
             state: st.server_state,
             current_leader,
             membership_config,
@@ -1102,7 +1103,7 @@ where
                             let _ = tx.send(Notification::VoteResponse {
                                 target,
                                 resp,
-                                sender_vote: vote,
+                                candidate_vote: vote.into_non_committed(),
                             });
                         }
                         Err(err) => tracing::error!({error=%err, target=display(target)}, "while requesting vote"),
@@ -1306,7 +1307,7 @@ where
             Notification::VoteResponse {
                 target,
                 resp,
-                sender_vote,
+                candidate_vote,
             } => {
                 let now = C::now();
 
@@ -1319,7 +1320,7 @@ where
 
                 #[allow(clippy::collapsible_if)]
                 if self.engine.candidate.is_some() {
-                    if self.does_vote_match(&sender_vote, "VoteResponse") {
+                    if self.does_candidate_vote_match(&candidate_vote, "VoteResponse") {
                         self.engine.handle_vote_resp(target, resp);
                     }
                 }
@@ -1328,17 +1329,17 @@ where
             Notification::HigherVote {
                 target,
                 higher,
-                sender_vote,
+                leader_vote,
             } => {
                 tracing::info!(
                     target = display(target),
                     higher_vote = display(&higher),
-                    sending_vote = display(&sender_vote),
+                    sending_vote = display(&leader_vote),
                     "received Notification::HigherVote: {}",
                     func_name!()
                 );
 
-                if self.does_vote_match(&sender_vote, "HigherVote") {
+                if self.does_leader_vote_match(&leader_vote, "HigherVote") {
                     // Rejected vote change is ok.
                     let _ = self.engine.vote_handler().update_vote(&higher);
                 }
@@ -1407,7 +1408,7 @@ where
                         // local log wont revert when membership changes.
                         #[allow(clippy::collapsible_if)]
                         if self.engine.leader.is_some() {
-                            if self.does_vote_match(io_id.vote_ref(), "LocalIO Notification") {
+                            if self.does_leader_vote_match(&log_io_id.committed_vote, "LocalIO Notification") {
                                 self.engine.replication_handler().update_local_progress(log_io_id.log_id);
                             }
                         }
@@ -1551,27 +1552,53 @@ where
         self.engine.elect();
     }
 
-    /// If a message is sent by a previous server state but is received by current server state,
+    /// If a message is sent by a previous Candidate but is received by current Candidate,
     /// it is a stale message and should be just ignored.
-    fn does_vote_match(&self, sender_vote: &Vote<C::NodeId>, msg: impl fmt::Display) -> bool {
-        // Get the current leading vote:
-        // - If input `sender_vote` is committed, it is sent by a Leader. Therefore we check against current
-        //   Leader's vote.
-        // - Otherwise, it is sent by a Candidate, we check against the current in progress voting state.
-        let my_vote = if sender_vote.is_committed() {
-            let l = self.engine.leader.as_ref();
-            l.map(|x| *x.committed_vote.deref())
+    fn does_candidate_vote_match(&self, candidate_vote: &NonCommittedVote<C>, msg: impl fmt::Display) -> bool {
+        // If it finished voting, Candidate's vote is None.
+        let Some(my_vote) = self.engine.candidate_ref().map(|x| *x.vote_ref()) else {
+            tracing::warn!(
+                "A message will be ignored because this node is no longer Candidate: \
+                 msg sent by vote: {}; when ({})",
+                candidate_vote,
+                msg
+            );
+            return false;
+        };
+
+        if candidate_vote.leader_id() != my_vote.leader_id() {
+            tracing::warn!(
+                "A message will be ignored because vote changed: \
+                msg sent by vote: {}; current my vote: {}; when ({})",
+                candidate_vote,
+                my_vote,
+                msg
+            );
+            false
         } else {
-            // If it finished voting, Candidate's vote is None.
-            let candidate = self.engine.candidate_ref();
-            candidate.map(|x| *x.vote_ref())
+            true
+        }
+    }
+
+    /// If a message is sent by a previous Leader but is received by current Leader,
+    /// it is a stale message and should be just ignored.
+    fn does_leader_vote_match(&self, leader_vote: &CommittedVote<C>, msg: impl fmt::Display) -> bool {
+        let Some(my_vote) = self.engine.leader.as_ref().map(|x| x.committed_vote) else {
+            tracing::warn!(
+                "A message will be ignored because this node is no longer Leader: \
+                msg sent by vote: {}; when ({})",
+                leader_vote,
+                msg
+            );
+            return false;
         };
 
-        if Some(*sender_vote) != my_vote {
+        if leader_vote != &my_vote {
             tracing::warn!(
-                "A message will be ignored because vote changed: msg sent by vote: {}; current my vote: {}; when ({})",
-                sender_vote,
-                my_vote.display(),
+                "A message will be ignored because vote changed: \
+                msg sent by vote: {}; current my vote: {}; when ({})",
+                leader_vote,
+                my_vote,
                 msg
             );
             false
@@ -1587,7 +1614,7 @@ where
         session_id: &ReplicationSessionId<C>,
         msg: impl fmt::Display + Copy,
     ) -> bool {
-        if !self.does_vote_match(session_id.vote_ref(), msg) {
+        if !self.does_leader_vote_match(&session_id.committed_vote(), msg) {
             return false;
         }
 
@@ -1683,7 +1710,7 @@ where
                 let last_log_id = *entries.last().unwrap().get_log_id();
                 tracing::debug!("AppendInputEntries: {}", DisplaySlice::<_>(&entries),);
 
-                let io_id = IOId::new_log_io(vote.into_committed(), Some(last_log_id));
+                let io_id = IOId::new_log_io(vote, Some(last_log_id));
                 let notify = Notification::LocalIO { io_id };
                 let callback = IOFlushed::new(notify, self.tx_notification.downgrade());
 
@@ -1706,12 +1733,16 @@ where
 
                 let _ = self.tx_notification.send(Notification::LocalIO { io_id: IOId::new(vote) });
 
-                let _ = self.tx_notification.send(Notification::VoteResponse {
-                    target: self.id,
-                    // last_log_id is not used when sending VoteRequest to local node
-                    resp: VoteResponse::new(vote, None, true),
-                    sender_vote: vote,
-                });
+                // If a non-committed vote is saved,
+                // there may be a candidate waiting for the response.
+                if let VoteStatus::Pending(non_committed) = vote.into_vote_status() {
+                    let _ = self.tx_notification.send(Notification::VoteResponse {
+                        target: self.id,
+                        // last_log_id is not used when sending VoteRequest to local node
+                        resp: VoteResponse::new(vote, None, true),
+                        candidate_vote: non_committed,
+                    });
+                }
             }
             Command::PurgeLog { upto } => {
                 self.log_store.purge(upto).await?;

openraft/src/engine/engine_impl.rs

@@ -1,4 +1,3 @@
-use std::ops::Deref;
 use std::time::Duration;
 
 use validit::Valid;
@@ -645,10 +644,10 @@ where C: RaftTypeConfig
         // Before sending any log, update the vote.
         // This could not fail because `internal_server_state` will be cleared
         // once `state.vote` is changed to a value of other node.
-        let _res = self.vote_handler().update_vote(&vote);
+        let _res = self.vote_handler().update_vote(&vote.into_vote());
         debug_assert!(_res.is_ok(), "commit vote can not fail but: {:?}", _res);
 
-        self.state.accept_io(IOId::new_log_io(vote.into_committed(), last_log_id));
+        self.state.accept_io(IOId::new_log_io(vote, last_log_id));
 
         // No need to submit UpdateIOProgress command,
         // IO progress is updated by the new blank log
@@ -752,7 +751,7 @@ where C: RaftTypeConfig
         };
 
         debug_assert!(
-            leader.committed_vote_ref().deref() >= self.state.vote_ref(),
+            leader.committed_vote_ref().into_vote() >= *self.state.vote_ref(),
             "leader.vote({}) >= state.vote({})",
             leader.committed_vote_ref(),
             self.state.vote_ref()

openraft/src/engine/handler/establish_handler/mod.rs

@@ -31,7 +31,7 @@ where C: RaftTypeConfig
 
         if let Some(l) = self.leader.as_ref() {
             #[allow(clippy::neg_cmp_op_on_partial_ord)]
-            if !(&vote > l.committed_vote_ref()) {
+            if !(vote > l.committed_vote_ref().into_vote()) {
                 tracing::warn!(
                     "vote is not greater than current existing leader vote. Do not establish new leader and quit"
                 );

openraft/src/engine/handler/leader_handler/mod.rs

@@ -121,7 +121,11 @@ where C: RaftTypeConfig
         self.state.vote.disable_lease();
 
         self.output.push_command(Command::BroadcastTransferLeader {
-            req: TransferLeaderRequest::new(*self.leader.committed_vote, to, self.leader.last_log_id().copied()),
+            req: TransferLeaderRequest::new(
+                self.leader.committed_vote.into_vote(),
+                to,
+                self.leader.last_log_id().copied(),
+            ),
         });
     }
 

openraft/src/engine/handler/vote_handler/mod.rs

@@ -167,7 +167,7 @@ where C: RaftTypeConfig
         if let Some(l) = self.leader.as_mut() {
             tracing::debug!("leading vote: {}", l.committed_vote,);
 
-            if l.committed_vote.leader_id() == self.state.vote_ref().leader_id() {
+            if l.committed_vote.into_vote().leader_id() == self.state.vote_ref().leader_id() {
                 tracing::debug!(
                     "vote still belongs to the same leader. Just updating vote is enough: node-{}, {}",
                     self.config.id,

openraft/src/proposer/leader.rs

@@ -200,7 +200,7 @@ where
         // Thus vote.voted_for() is this node.
 
         // Safe unwrap: voted_for() is always non-None in Openraft
-        let node_id = self.committed_vote.leader_id().voted_for().unwrap();
+        let node_id = self.committed_vote.into_vote().leader_id().voted_for().unwrap();
         let now = C::now();
 
         tracing::debug!(