Refactor: Enhance Election and Replication

- **Election Update Delay:**
  - The `Voting` state is now updated only after `SaveVote` is
    successfully persisted on disk. This anticipates future
    modifications where `SaveVote` will be callback-based, ensuring
    state consistency.

- **Deprecation of `VoteResponse.vote_granted`:**
  - Granting a `RequestVote` is now determined solely by whether
    `VoteResponse.vote` matches the candidate's vote. This simplifies
    the logic and reduces ambiguity in vote handling.

- **Replication Stream Reorganization:**
  - Replication stream handles have been moved from
    `RaftCore::LeaderData` to `RaftCore`. This reorganization prepares
    for supporting `RequestVote` RPCs in the replication mechanism,
    although it currently continues to support only log and snapshot
    replication.

- **Vote Comparison Enhancement:**
  - When filtering out stale messages (e.g., when a new Leader receives
    an `AppendEntries` response intended for a previous Leader), the
    responded vote is now compared with the vote in the `Leading` state.
    Previously, it was compared with `RaftCore.engine.raft_state.vote`,
    which might differ from the `Leading` state's vote in future
    scenarios.

- **Election State Management:**
  - During election, `Engine.candidate` is `Some` and `Engine.leader` is
    `None`. Upon election completion, `Engine.candidate` is cleared, and
    a new `Leader` is assigned to `Engine.leader`.
  - Renamed `Engine.internal_server_state` to `leader`, as it now only
    stores a `Leader` instance. The candidate state is moved to
    `Engine.candidate`.
  - Moved the process of establishing a Leader to `EstablishHandler`.
  - A Leader can now be established in two ways: by election or by
    re-creating the Leader upon startup.
  - Replace `InternalServerState` with `Option<Box<Leader<...>>>`

- **Logging Enhancements:**
  - Added `DisplaySliceExt` to create a displayable instance for slices
    of `impl Display`.
  - Added `DisplayInstant` to display `Instant` in a human-readable
    format, improving logging readability.

Author: drmingdrmer

openraft/Cargo.toml

@@ -18,6 +18,7 @@ repository    = { workspace = true }
 anyerror        = { workspace = true }
 anyhow          = { workspace = true, optional = true }
 byte-unit       = { workspace = true }
+chrono          = { workspace = true }
 clap            = { workspace = true }
 derive_more     = { workspace = true }
 futures         = { workspace = true }

openraft/src/core/raft_core.rs

@@ -36,7 +36,9 @@ use crate::core::sm;
 use crate::core::sm::handle;
 use crate::core::sm::CommandSeq;
 use crate::core::ServerState;
+use crate::display_ext::DisplayInstantExt;
 use crate::display_ext::DisplayOption;
+use crate::display_ext::DisplayOptionExt;
 use crate::display_ext::DisplaySlice;
 use crate::engine::Command;
 use crate::engine::Condition;
@@ -71,6 +73,7 @@ use crate::raft::AppendEntriesRequest;
 use crate::raft::AppendEntriesResponse;
 use crate::raft::ClientWriteResponse;
 use crate::raft::VoteRequest;
+use crate::raft::VoteResponse;
 use crate::raft_state::LogIOId;
 use crate::raft_state::LogStateReader;
 use crate::replication;
@@ -138,20 +141,14 @@ impl<C: RaftTypeConfig> Debug for ApplyResult<C> {
 ///
 /// It is created when RaftCore enters leader state, and will be dropped when it quits leader state.
 pub(crate) struct LeaderData<C: RaftTypeConfig> {
-    /// A mapping of node IDs the replication state of the target node.
-    // TODO(xp): make it a field of RaftCore. it does not have to belong to leader.
-    //           It requires the Engine to emit correct add/remove replication commands
-    pub(super) replications: BTreeMap<C::NodeId, ReplicationHandle<C>>,
-
     /// The time to send next heartbeat.
     pub(crate) next_heartbeat: <C::AsyncRuntime as AsyncRuntime>::Instant,
 }
 
 impl<C: RaftTypeConfig> LeaderData<C> {
     pub(crate) fn new() -> Self {
         Self {
-            replications: BTreeMap::new(),
-            next_heartbeat: <C::AsyncRuntime as AsyncRuntime>::Instant::now(),
+            next_heartbeat: InstantOf::<C>::now(),
         }
     }
 }
@@ -187,6 +184,9 @@ where
     /// Channels to send result back to client when logs are applied.
     pub(crate) client_resp_channels: BTreeMap<u64, ResponderOf<C>>,
 
+    /// A mapping of node IDs the replication state of the target node.
+    pub(crate) replications: BTreeMap<C::NodeId, ReplicationHandle<C>>,
+
     pub(crate) leader_data: Option<LeaderData<C>>,
 
     #[allow(dead_code)]
@@ -320,7 +320,7 @@ where
         let mut pending = FuturesUnordered::new();
 
         let voter_progresses = {
-            let l = &self.engine.internal_server_state.leading().unwrap();
+            let l = &self.engine.leader.as_ref().unwrap();
             l.progress.iter().filter(|(id, _v)| l.progress.is_voter(id) == Some(true))
         };
 
@@ -533,7 +533,7 @@ where
 
     #[tracing::instrument(level = "debug", skip_all)]
     pub fn flush_metrics(&mut self) {
-        let leader_metrics = if let Some(leader) = self.engine.internal_server_state.leading() {
+        let leader_metrics = if let Some(leader) = self.engine.leader.as_ref() {
             let prog = &leader.progress;
             Some(prog.iter().map(|(id, p)| (*id, *p.borrow())).collect())
         } else {
@@ -702,7 +702,7 @@ where
     /// from a quorum of followers, indicating its leadership is current and recognized.
     /// If the node is not a leader or no acknowledgment has been received, `None` is returned.
     fn last_quorum_acked_time(&mut self) -> Option<InstantOf<C>> {
-        let leading = self.engine.internal_server_state.leading_mut();
+        let leading = self.engine.leader.as_mut();
         leading.and_then(|l| l.last_quorum_acked_time())
     }
 
@@ -831,7 +831,9 @@ where
         let network = self.network.new_client(target, target_node).await;
         let snapshot_network = self.network.new_client(target, target_node).await;
 
-        let session_id = ReplicationSessionId::new(*self.engine.state.vote_ref(), *membership_log_id);
+        let leader = self.engine.leader.as_ref().unwrap();
+
+        let session_id = ReplicationSessionId::new(leader.vote, *membership_log_id);
 
         ReplicationCore::<C, N, LS>::spawn(
             target,
@@ -853,27 +855,23 @@ where
     pub async fn remove_all_replication(&mut self) {
         tracing::info!("remove all replication");
 
-        if let Some(l) = &mut self.leader_data {
-            let nodes = std::mem::take(&mut l.replications);
+        let nodes = std::mem::take(&mut self.replications);
 
-            tracing::debug!(
-                targets = debug(nodes.iter().map(|x| *x.0).collect::<Vec<_>>()),
-                "remove all targets from replication_metrics"
-            );
+        tracing::debug!(
+            targets = debug(nodes.iter().map(|x| *x.0).collect::<Vec<_>>()),
+            "remove all targets from replication_metrics"
+        );
 
-            for (target, s) in nodes {
-                let handle = s.join_handle;
+        for (target, s) in nodes {
+            let handle = s.join_handle;
 
-                // Drop sender to notify the task to shutdown
-                drop(s.tx_repl);
+            // Drop sender to notify the task to shutdown
+            drop(s.tx_repl);
 
-                tracing::debug!("joining removed replication: {}", target);
-                let _x = handle.await;
-                tracing::info!("Done joining removed replication : {}", target);
-            }
-        } else {
-            unreachable!("it has to be a leader!!!");
-        };
+            tracing::debug!("joining removed replication: {}", target);
+            let _x = handle.await;
+            tracing::info!("Done joining removed replication : {}", target);
+        }
     }
 
     /// Run as many commands as possible.
@@ -1154,8 +1152,8 @@ where
             RaftMsg::RequestVote { rpc, tx } => {
                 let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
                 tracing::info!(
-                    now = debug(now),
-                    vote_request = display(rpc.summary()),
+                    now = display(now.display()),
+                    vote_request = display(&rpc),
                     "received RaftMsg::RequestVote: {}",
                     func_name!()
                 );
@@ -1237,36 +1235,36 @@ where
             Notify::VoteResponse {
                 target,
                 resp,
-                sender_vote: vote,
+                sender_vote,
             } => {
                 let now = <C::AsyncRuntime as AsyncRuntime>::Instant::now();
 
                 tracing::info!(
-                    now = debug(now),
-                    resp = display(resp.summary()),
+                    now = display(now.display()),
+                    resp = display(&resp),
                     "received Notify::VoteResponse: {}",
                     func_name!()
                 );
 
-                if self.does_vote_match(&vote, "VoteResponse") {
+                if self.does_vote_match(&sender_vote, "VoteResponse") {
                     self.engine.handle_vote_resp(target, resp);
                 }
             }
 
             Notify::HigherVote {
                 target,
                 higher,
-                sender_vote: vote,
+                sender_vote,
             } => {
                 tracing::info!(
                     target = display(target),
                     higher_vote = display(&higher),
-                    sending_vote = display(&vote),
+                    sending_vote = display(&sender_vote),
                     "received Notify::HigherVote: {}",
                     func_name!()
                 );
 
-                if self.does_vote_match(&vote, "HigherVote") {
+                if self.does_vote_match(&sender_vote, "HigherVote") {
                     // Rejected vote change is ok.
                     let _ = self.engine.vote_handler().update_vote(&higher);
                 }
@@ -1282,6 +1280,7 @@ where
 
                 // TODO: test: fixture: make isolated_nodes a single-way isolating.
 
+                // TODO: check if it is Leader with Engine
                 // Leader send heartbeat
                 let heartbeat_at = self.leader_data.as_ref().map(|x| x.next_heartbeat);
                 if let Some(t) = heartbeat_at {
@@ -1347,16 +1346,20 @@ where
                         return Err(Fatal::from(error));
                     }
 
-                    replication::Response::HigherVote { target, higher, vote } => {
+                    replication::Response::HigherVote {
+                        target,
+                        higher,
+                        sender_vote,
+                    } => {
                         tracing::info!(
                             target = display(target),
                             higher_vote = display(&higher),
-                            sending_vote = display(&vote),
+                            sender_vote = display(&sender_vote),
                             "received Notify::HigherVote: {}",
                             func_name!()
                         );
 
-                        if self.does_vote_match(&vote, "HigherVote") {
+                        if self.does_vote_match(&sender_vote, "HigherVote") {
                             // Rejected vote change is ok.
                             let _ = self.engine.vote_handler().update_vote(&higher);
                         }
@@ -1506,30 +1509,40 @@ where
             "handle_replication_progress"
         );
 
+        #[allow(clippy::collapsible_if)]
         if tracing::enabled!(Level::DEBUG) {
-            if let Some(l) = &self.leader_data {
-                if !l.replications.contains_key(&target) {
-                    tracing::warn!("leader has removed target: {}", target);
-                };
-            } else {
-                // TODO: A leader may have stepped down.
-            }
+            if !self.replications.contains_key(&target) {
+                tracing::warn!("leader has removed target: {}", target);
+            };
         }
 
         // A leader may have stepped down.
-        if self.engine.internal_server_state.is_leading() {
+        if self.engine.leader.is_some() {
             self.engine.replication_handler().update_progress(target, request_id, result);
         }
     }
 
     /// If a message is sent by a previous server state but is received by current server state,
     /// it is a stale message and should be just ignored.
-    fn does_vote_match(&self, vote: &Vote<C::NodeId>, msg: impl Display) -> bool {
-        if vote != self.engine.state.vote_ref() {
+    fn does_vote_match(&self, sender_vote: &Vote<C::NodeId>, msg: impl 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.vote)
+        } else {
+            // If it finished voting, Candidate's vote is None.
+            let candidate = self.engine.candidate_ref();
+            candidate.map(|x| *x.vote_ref())
+        };
+
+        if Some(*sender_vote) != my_vote {
             tracing::warn!(
-                "vote changed: msg sent by: {:?}; curr: {}; ignore when ({})",
-                vote,
-                self.engine.state.vote_ref(),
+                "A message will be ignored because vote changed: msg sent by vote: {}; current my vote: {}; when ({})",
+                sender_vote,
+                my_vote.display(),
                 msg
             );
             false
@@ -1544,7 +1557,7 @@ where
         session_id: &ReplicationSessionId<C::NodeId>,
         msg: impl Display + Copy,
     ) -> bool {
-        if !self.does_vote_match(&session_id.vote, msg) {
+        if !self.does_vote_match(session_id.vote_ref(), msg) {
             return false;
         }
 
@@ -1624,6 +1637,13 @@ where
             Command::SaveVote { vote } => {
                 self.log_store.save_vote(&vote).await?;
                 self.engine.state.io_state_mut().update_vote(vote);
+
+                let _ = self.tx_notify.send(Notify::VoteResponse {
+                    target: self.id,
+                    // last_log_id is not used when sending VoteRequest to local node
+                    resp: VoteResponse::new(vote, None),
+                    sender_vote: vote,
+                });
             }
             Command::PurgeLog { upto } => {
                 self.log_store.purge(upto).await?;
@@ -1656,12 +1676,8 @@ where
                 self.spawn_parallel_vote_requests(&vote_req).await;
             }
             Command::ReplicateCommitted { committed } => {
-                if let Some(l) = &self.leader_data {
-                    for node in l.replications.values() {
-                        let _ = node.tx_repl.send(Replicate::Committed(committed));
-                    }
-                } else {
-                    unreachable!("it has to be a leader!!!");
+                for node in self.replications.values() {
+                    let _ = node.tx_repl.send(Replicate::Committed(committed));
                 }
             }
             Command::Commit {
@@ -1673,38 +1689,29 @@ where
                 self.apply_to_state_machine(seq, already_committed.next_index(), upto.index).await?;
             }
             Command::Replicate { req, target } => {
-                if let Some(l) = &self.leader_data {
-                    let node = l.replications.get(&target).expect("replication to target node exists");
+                let node = self.replications.get(&target).expect("replication to target node exists");
 
-                    match req {
-                        Inflight::None => {
-                            let _ = node.tx_repl.send(Replicate::Heartbeat);
-                        }
-                        Inflight::Logs { id, log_id_range } => {
-                            let _ = node.tx_repl.send(Replicate::logs(RequestId::new_append_entries(id), log_id_range));
-                        }
-                        Inflight::Snapshot { id, last_log_id } => {
-                            // unwrap: The replication channel must not be dropped or it is a bug.
-                            node.tx_repl.send(Replicate::snapshot(RequestId::new_snapshot(id), last_log_id)).map_err(
-                                |_e| StorageIOError::read_snapshot(None, AnyError::error("replication channel closed")),
-                            )?;
-                        }
+                match req {
+                    Inflight::None => {
+                        let _ = node.tx_repl.send(Replicate::Heartbeat);
+                    }
+                    Inflight::Logs { id, log_id_range } => {
+                        let _ = node.tx_repl.send(Replicate::logs(RequestId::new_append_entries(id), log_id_range));
+                    }
+                    Inflight::Snapshot { id, last_log_id } => {
+                        // unwrap: The replication channel must not be dropped or it is a bug.
+                        node.tx_repl.send(Replicate::snapshot(RequestId::new_snapshot(id), last_log_id)).map_err(
+                            |_e| StorageIOError::read_snapshot(None, AnyError::error("replication channel closed")),
+                        )?;
                     }
-                } else {
-                    unreachable!("it has to be a leader!!!");
                 }
             }
             Command::RebuildReplicationStreams { targets } => {
                 self.remove_all_replication().await;
 
                 for (target, matching) in targets.iter() {
                     let handle = self.spawn_replication_stream(*target, *matching).await;
-
-                    if let Some(l) = &mut self.leader_data {
-                        l.replications.insert(*target, handle);
-                    } else {
-                        unreachable!("it has to be a leader!!!");
-                    }
+                    self.replications.insert(*target, handle);
                 }
             }
             Command::StateMachine { command } => {