Merge pull request #854 from Migorithm/fix/election-failure

fix: election failure due to scheduler synchronization

Author: Migorithm

duva/src/domains/cluster_actors/actor.rs

@@ -363,6 +363,10 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
 
     #[instrument(level = tracing::Level::DEBUG, skip(self))]
     pub(crate) async fn send_rpc(&mut self) {
+        if !self.replication.is_leader() {
+            return;
+        }
+
         if self.replicas().count() == 0 {
             return;
         }
@@ -409,6 +413,9 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         from: &PeerIdentifier,
         repl_res: ReplicationAck,
     ) {
+        if !self.replication.is_leader() {
+            return;
+        }
         let Some(peer) = self.members.get_mut(from) else {
             return;
         };
@@ -426,6 +433,9 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
 
     #[instrument(level = tracing::Level::DEBUG, skip(self, heartbeat), fields(peer_id = %heartbeat.from))]
     pub(crate) async fn append_entries_rpc(&mut self, heartbeat: HeartBeat) {
+        if self.replication.is_leader() {
+            return;
+        }
         if self.check_term_outdated(&heartbeat).await {
             err!("Term Outdated received:{} self:{}", heartbeat.term, self.replication.term);
             return;
@@ -875,7 +885,8 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
     }
 
     async fn replicate(&mut self, mut heartbeat: HeartBeat) {
-        if self.replication.is_leader() {
+        if heartbeat.leader_commit_idx.is_none() {
+            err!("It must have leader commit index!");
             return;
         }
 
@@ -951,15 +962,17 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
             // ! Term mismatch -> triggers log truncation
             err!("Term mismatch: {} != {}", prev_entry.term, prev_log_term);
             self.replication.logger.truncate_after(prev_log_index);
-
-            return Err(RejectionReason::LogInconsistency);
         }
 
         Ok(())
     }
 
     #[instrument(level = tracing::Level::INFO, skip(self))]
     pub(crate) async fn run_for_election(&mut self) {
+        if self.replication.is_leader() {
+            return;
+        }
+
         self.become_candidate();
         let request_vote = RequestVote::new(self.replication.info(), &self.replication.logger);
 

duva/src/domains/cluster_actors/actor/tests/replications.rs

@@ -374,8 +374,8 @@ async fn follower_truncates_log_on_term_mismatch() {
     let result = cluster_actor.replicate_log_entries(&mut heartbeat).await;
 
     // THEN: Expect truncation and rejection
-    assert_eq!(cluster_actor.replication.logger.target.writer.len(), 1);
-    assert!(result.is_err(), "Should reject due to term mismatch");
+    assert_eq!(cluster_actor.replication.logger.target.writer.len(), 2);
+    assert!(result.is_ok(), "Should pass as follower will just truncate and write");
 }
 
 #[tokio::test]

duva/src/domains/cluster_actors/service.rs

@@ -34,6 +34,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         Ok(self)
     }
 
+    // ! Beware scheduler is a separate task that asychrnously sends message, therefore leadership change in between may not be observed.
     async fn process_scheduler_message(&mut self, msg: SchedulerMessage) {
         use SchedulerMessage::*;
         match msg {

duva/tests/replication_ops/mod.rs

@@ -8,15 +8,14 @@ mod test_raft_happy_case;
 mod test_set_twice_after_election;
 mod test_sync;
 
-fn panic_if_election_not_done(port1: u16, port2: u16) {
+fn panic_if_election_not_done(order: &str, port1: u16, port2: u16, num_possible_nodes: u32) {
     let mut first_election_cnt = 0;
     let mut flag = false;
     let mut h1 = Client::new(port1);
-    h1.read_timeout = Duration::from_secs(4);
 
     let start = std::time::Instant::now();
     while first_election_cnt < 50 {
-        let res = h1.send_and_get_vec("role", 2);
+        let res = h1.send_and_get_vec("role", num_possible_nodes);
         println!(
             "[{}ms] Poll {}: port1={} port2={} res={:?}",
             start.elapsed().as_millis(),
@@ -43,5 +42,5 @@ fn panic_if_election_not_done(port1: u16, port2: u16) {
             port2
         );
     }
-    assert!(flag, "first election fail");
+    assert!(flag, "{order} election fail");
 }

duva/tests/replication_ops/test_leader_election.rs

@@ -16,7 +16,7 @@ fn run_leader_election(with_append_only: bool) -> anyhow::Result<()> {
     drop(leader_p);
 
     // THEN
-    panic_if_election_not_done(follower_p1.port, follower_p2.port);
+    panic_if_election_not_done("first", follower_p1.port, follower_p2.port, 3);
 
     Ok(())
 }

duva/tests/replication_ops/test_leader_election_twice.rs

@@ -6,25 +6,22 @@ use crate::{
 /// following test is to see if election works even after the first election.
 fn run_leader_election_twice(with_append_only: bool) -> anyhow::Result<()> {
     // GIVEN
-    let server_env = ServerEnv::default().with_append_only(with_append_only);
-    let server_env = server_env;
-    let mut leader_env = server_env;
+    let mut leader_env = ServerEnv::default().with_append_only(with_append_only);
     let mut follower_env1 = ServerEnv::default().with_append_only(with_append_only);
     let mut follower_env2 = ServerEnv::default().with_append_only(with_append_only);
 
     let [leader_p, follower_p1, follower_p2] =
         form_cluster([&mut leader_env, &mut follower_env1, &mut follower_env2]);
 
-    // !first leader is killed -> election happens
-
+    // WHEN: The first leader is killed, triggering an election between F1 and F2
     drop(leader_p);
-    panic_if_election_not_done(follower_p1.port, follower_p2.port);
+    panic_if_election_not_done("first", follower_p1.port, follower_p2.port, 3);
 
     let mut processes = vec![];
     for f in [follower_p1, follower_p2] {
         let mut handler = Client::new(f.port);
-        let res = handler.send_and_get_vec("info replication", 4);
-        if !res.contains(&"role:leader".to_string()) {
+        let res = handler.send_and_get_vec("role", 3);
+        if !res.contains(&format!("127.0.0.1:{}:{}", f.port, "leader")) {
             processes.push(f);
             continue;
         }
@@ -41,7 +38,7 @@ fn run_leader_election_twice(with_append_only: bool) -> anyhow::Result<()> {
     }
     assert_eq!(processes.len(), 2);
 
-    panic_if_election_not_done(processes[0].port, processes[1].port);
+    panic_if_election_not_done("second", processes[0].port, processes[1].port, 3);
 
     Ok(())
 }

duva/tests/replication_ops/test_set_twice_after_election.rs

@@ -28,7 +28,7 @@ fn run_set_twice_after_election(with_append_only: bool) -> anyhow::Result<()> {
     let mut h1 = Client::new(follower_p1.port);
     let mut h2 = Client::new(follower_p2.port);
 
-    panic_if_election_not_done(follower_p1.port, follower_p2.port);
+    panic_if_election_not_done("first", follower_p1.port, follower_p2.port, 3);
 
     let res = h1.send_and_get_vec("role", 3);