Merge pull request #635 from drmingdrmer/30-internal-state

Refactor: move leading state out of RaftState

Author: drmingdrmer

openraft/src/core/raft_core.rs

@@ -251,7 +251,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
         // Spawn parallel requests, all with the standard timeout for heartbeats.
         let mut pending = FuturesUnordered::new();
 
-        let voter_progresses = if let Some(l) = &self.engine.state.internal_server_state.leading() {
+        let voter_progresses = if let Some(l) = &self.engine.internal_server_state.leading() {
             l.progress
                 .iter()
                 .filter(|(id, _v)| l.progress.is_voter(id) == Some(true))
@@ -399,7 +399,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
 
         let curr = &self.engine.state.membership_state.effective;
         if curr.contains(&target) {
-            let matched = if let Some(l) = &self.engine.state.internal_server_state.leading() {
+            let matched = if let Some(l) = &self.engine.internal_server_state.leading() {
                 *l.progress.get(&target)
             } else {
                 unreachable!("it has to be a leader!!!");
@@ -538,7 +538,7 @@ impl<C: RaftTypeConfig, N: RaftNetworkFactory<C>, S: RaftStorage<C>> RaftCore<C,
                 Expectation::AtLineRate => {
                     // Expect to be at line rate but not.
 
-                    let matched = if let Some(l) = &self.engine.state.internal_server_state.leading() {
+                    let matched = if let Some(l) = &self.engine.internal_server_state.leading() {
                         *l.progress.get(node_id)
                     } else {
                         unreachable!("it has to be a leader!!!");

openraft/src/engine/elect_test.rs

@@ -47,7 +47,7 @@ fn test_elect() -> anyhow::Result<()> {
         assert_eq!(Vote::new_committed(1, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Leader, eng.state.server_state);
@@ -106,15 +106,15 @@ fn test_elect() -> anyhow::Result<()> {
 
         // Build in-progress election state
         eng.state.vote = Vote::new_committed(1, 2);
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
 
         eng.elect();
 
         assert_eq!(Vote::new_committed(2, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Leader, eng.state.server_state);
@@ -177,7 +177,7 @@ fn test_elect() -> anyhow::Result<()> {
         assert_eq!(Vote::new(1, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Candidate, eng.state.server_state);

openraft/src/engine/engine_impl.rs

@@ -9,6 +9,7 @@ use crate::error::NotAllowed;
 use crate::error::NotInMembers;
 use crate::error::RejectVoteRequest;
 use crate::internal_server_state::InternalServerState;
+use crate::leader::Leader;
 use crate::membership::EffectiveMembership;
 use crate::membership::NodeRole;
 use crate::node::Node;
@@ -79,6 +80,9 @@ where
     /// The state of this raft node.
     pub(crate) state: RaftState<NID, N>,
 
+    /// The internal server state used by Engine.
+    pub(crate) internal_server_state: InternalServerState<NID>,
+
     /// Tracks what kind of metrics changed
     pub(crate) metrics_flags: MetricsChangeFlags,
 
@@ -96,6 +100,7 @@ where
             id,
             config,
             state: init_state.clone(),
+            internal_server_state: InternalServerState::default(),
             metrics_flags: MetricsChangeFlags::default(),
             commands: vec![],
         }
@@ -162,7 +167,7 @@ where
         self.handle_vote_change(&Vote::new(self.state.vote.term + 1, self.id)).unwrap();
 
         // Safe unwrap()
-        let leader = self.state.internal_server_state.leading_mut().unwrap();
+        let leader = self.internal_server_state.leading_mut().unwrap();
         leader.grant_vote_by(self.id);
         let quorum_granted = leader.is_vote_granted();
 
@@ -233,7 +238,7 @@ where
         );
 
         // If this node is no longer a leader(i.e., electing), just ignore the delayed vote_resp.
-        let leader = match &mut self.state.internal_server_state {
+        let leader = match &mut self.internal_server_state {
             InternalServerState::Leading(l) => l,
             InternalServerState::Following => return,
         };
@@ -674,7 +679,7 @@ where
         let end = self.state.last_log_id().next_index();
 
         // If membership changes, the progress should be upgraded.
-        if let Some(leader) = &mut self.state.internal_server_state.leading_mut() {
+        if let Some(leader) = &mut self.internal_server_state.leading_mut() {
             let old_progress = leader.progress.clone();
             let learner_ids = em.learner_ids().collect::<Vec<_>>();
 
@@ -683,7 +688,7 @@ where
         }
 
         // A leader that is removed will be shut down when this membership log is committed.
-        if self.state.internal_server_state.is_leading() {
+        if self.internal_server_state.is_leading() {
             self.update_replications()
         }
 
@@ -699,7 +704,7 @@ where
         tracing::debug!("update_progress: node_id:{} log_id:{:?}", node_id, log_id);
 
         let committed = {
-            let leader = match self.state.internal_server_state.leading_mut() {
+            let leader = match self.internal_server_state.leading_mut() {
                 None => {
                     // TODO: is it a bug if trying to update progress when it is not in leading state?
                     return;
@@ -943,11 +948,11 @@ where
     pub(crate) fn enter_leading(&mut self) {
         debug_assert_eq!(self.state.vote.node_id, self.id);
         // debug_assert!(
-        //     self.state.internal_server_state.is_following(),
+        //     self.internal_server_state.is_following(),
         //     "can not enter leading twice"
         // );
 
-        self.state.new_leader();
+        self.new_leader();
     }
 
     /// Leave leading state and enter following state(vote.node_id != self.id).
@@ -962,7 +967,7 @@ where
         // debug_assert_ne!(self.state.vote.node_id, self.id);
 
         // debug_assert!(
-        //     self.state.internal_server_state.is_leading(),
+        //     self.internal_server_state.is_leading(),
         //     "can not enter following twice"
         // );
 
@@ -980,11 +985,11 @@ where
             self.push_command(Command::InstallElectionTimer { can_be_leader: true });
         }
 
-        if self.state.internal_server_state.is_following() {
+        if self.internal_server_state.is_following() {
             return;
         }
 
-        self.state.internal_server_state = InternalServerState::Following;
+        self.internal_server_state = InternalServerState::Following;
 
         self.update_server_state_if_changed();
     }
@@ -1005,6 +1010,17 @@ where
         self.append_blank_log();
     }
 
+    /// Create a new Leader, when raft enters candidate state.
+    /// In openraft, Leader and Candidate shares the same state.
+    pub(crate) fn new_leader(&mut self) {
+        let em = &self.state.membership_state.effective;
+        self.internal_server_state = InternalServerState::Leading(Leader::new(
+            em.membership.to_quorum_set(),
+            em.learner_ids(),
+            self.state.last_log_id().index(),
+        ));
+    }
+
     fn append_blank_log(&mut self) {
         let log_id = LogId {
             leader_id: self.state.vote.leader_id(),
@@ -1018,7 +1034,7 @@ where
 
     /// update replication streams to reflect replication progress change.
     fn update_replications(&mut self) {
-        if let Some(leader) = self.state.internal_server_state.leading() {
+        if let Some(leader) = self.internal_server_state.leading() {
             let mut targets = vec![];
             for (node_id, matched) in leader.progress.iter() {
                 if node_id != &self.id {

openraft/src/engine/handle_vote_req_test.rs

@@ -31,7 +31,7 @@ fn eng() -> Engine<u64, ()> {
     eng.state.vote = Vote::new(2, 1);
     eng.state.server_state = ServerState::Candidate;
     eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m01()));
-    eng.state.new_leader();
+    eng.new_leader();
     eng
 }
 
@@ -54,7 +54,7 @@ fn test_handle_vote_req_reject_smaller_vote() -> anyhow::Result<()> {
     );
 
     assert_eq!(Vote::new(2, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_leading());
+    assert!(eng.internal_server_state.is_leading());
 
     assert_eq!(ServerState::Candidate, eng.state.server_state);
     assert_eq!(
@@ -91,7 +91,7 @@ fn test_handle_vote_req_reject_smaller_last_log_id() -> anyhow::Result<()> {
     );
 
     assert_eq!(Vote::new(2, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_leading());
+    assert!(eng.internal_server_state.is_leading());
 
     assert_eq!(ServerState::Candidate, eng.state.server_state);
     assert_eq!(
@@ -129,7 +129,7 @@ fn test_handle_vote_req_granted_equal_vote_and_last_log_id() -> anyhow::Result<(
     );
 
     assert_eq!(Vote::new(2, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_following());
+    assert!(eng.internal_server_state.is_following());
 
     assert_eq!(ServerState::Follower, eng.state.server_state);
     assert_eq!(
@@ -174,7 +174,7 @@ fn test_handle_vote_req_granted_greater_vote() -> anyhow::Result<()> {
     );
 
     assert_eq!(Vote::new(3, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_following());
+    assert!(eng.internal_server_state.is_following());
 
     assert_eq!(ServerState::Follower, eng.state.server_state);
     assert_eq!(

openraft/src/engine/handle_vote_resp_test.rs

@@ -50,7 +50,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         });
 
         assert_eq!(Vote::new(2, 1), eng.state.vote);
-        assert!(eng.state.internal_server_state.is_following());
+        assert!(eng.internal_server_state.is_following());
 
         assert_eq!(ServerState::Follower, eng.state.server_state);
         assert_eq!(
@@ -71,8 +71,8 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         eng.id = 1;
         eng.state.vote = Vote::new(2, 1);
         eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m12()));
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
         eng.state.server_state = ServerState::Candidate;
 
         eng.handle_vote_resp(2, VoteResponse {
@@ -84,7 +84,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         assert_eq!(Vote::new(2, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Candidate, eng.state.server_state);
@@ -110,8 +110,8 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         eng.state.vote = Vote::new(2, 1);
         eng.state.log_ids = LogIdList::new(vec![log_id(3, 3)]);
         eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m12()));
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
         eng.state.server_state = ServerState::Candidate;
 
         eng.handle_vote_resp(2, VoteResponse {
@@ -121,7 +121,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         });
 
         assert_eq!(Vote::new(2, 2), eng.state.vote);
-        assert!(eng.state.internal_server_state.is_leading());
+        assert!(eng.internal_server_state.is_leading());
 
         assert_eq!(ServerState::Candidate, eng.state.server_state);
         assert_eq!(
@@ -148,8 +148,8 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         eng.id = 1;
         eng.state.vote = Vote::new(2, 1);
         eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m12()));
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
         eng.state.server_state = ServerState::Candidate;
 
         eng.handle_vote_resp(2, VoteResponse {
@@ -161,7 +161,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         assert_eq!(Vote::new(2, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Candidate, eng.state.server_state);
@@ -186,8 +186,8 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         eng.id = 1;
         eng.state.vote = Vote::new(2, 1);
         eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m1234()));
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
         eng.state.server_state = ServerState::Candidate;
 
         eng.handle_vote_resp(2, VoteResponse {
@@ -199,7 +199,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         assert_eq!(Vote::new(2, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1,2},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Candidate, eng.state.server_state);
@@ -221,8 +221,8 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         eng.id = 1;
         eng.state.vote = Vote::new(2, 1);
         eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m12()));
-        eng.state.new_leader();
-        eng.state.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
+        eng.new_leader();
+        eng.internal_server_state.leading_mut().map(|l| l.vote_granted_by.insert(1));
         eng.state.server_state = ServerState::Candidate;
 
         eng.handle_vote_resp(2, VoteResponse {
@@ -234,7 +234,7 @@ fn test_handle_vote_resp() -> anyhow::Result<()> {
         assert_eq!(Vote::new_committed(2, 1), eng.state.vote);
         assert_eq!(
             Some(btreeset! {1,2},),
-            eng.state.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
+            eng.internal_server_state.leading().map(|x| x.vote_granted_by.clone())
         );
 
         assert_eq!(ServerState::Leader, eng.state.server_state);

openraft/src/engine/internal_handle_vote_req_test.rs

@@ -30,7 +30,7 @@ fn eng() -> Engine<u64, ()> {
     eng.state.vote = Vote::new(2, 1);
     eng.state.server_state = ServerState::Candidate;
     eng.state.membership_state.effective = Arc::new(EffectiveMembership::new(Some(log_id(1, 1)), m01()));
-    eng.state.new_leader();
+    eng.new_leader();
     eng
 }
 
@@ -43,7 +43,7 @@ fn test_handle_vote_change_reject_smaller_vote() -> anyhow::Result<()> {
     assert_eq!(Err(RejectVoteRequest::ByVote(Vote::new(2, 1))), resp);
 
     assert_eq!(Vote::new(2, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_leading());
+    assert!(eng.internal_server_state.is_leading());
 
     assert_eq!(ServerState::Candidate, eng.state.server_state);
     assert_eq!(
@@ -70,7 +70,7 @@ fn test_handle_vote_change_committed_vote() -> anyhow::Result<()> {
     assert_eq!(Ok(()), resp);
 
     assert_eq!(Vote::new_committed(3, 2), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_following());
+    assert!(eng.internal_server_state.is_following());
 
     assert_eq!(ServerState::Follower, eng.state.server_state);
     assert_eq!(
@@ -108,7 +108,7 @@ fn test_handle_vote_change_granted_equal_vote() -> anyhow::Result<()> {
     assert_eq!(Ok(()), resp);
 
     assert_eq!(Vote::new(2, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_following());
+    assert!(eng.internal_server_state.is_following());
 
     assert_eq!(ServerState::Follower, eng.state.server_state);
     assert_eq!(
@@ -142,7 +142,7 @@ fn test_handle_vote_change_granted_greater_vote() -> anyhow::Result<()> {
     assert_eq!(Ok(()), resp);
 
     assert_eq!(Vote::new(3, 1), eng.state.vote);
-    assert!(eng.state.internal_server_state.is_following());
+    assert!(eng.internal_server_state.is_following());
 
     assert_eq!(ServerState::Follower, eng.state.server_state);
     assert_eq!(