feat: add log-level validation

Author: Migorithm

.config/nextest.toml

@@ -1,6 +1,6 @@
 [profile.default]
-retries = 0          # this is the default, so it doesn't need to be specified
-slow-timeout = "15s"
+retries = 0                                            # this is the default, so it doesn't need to be specified
+slow-timeout = { period = "15s", terminate-after = 4 }
 
 
 [profile.ci]

duva/src/domains/cluster_actors/actor.rs

@@ -385,20 +385,42 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
         if self.find_replica_mut(&request_vote.candidate_id).is_none() {
             return;
         };
-        REQUESTS_BLOCKED_BY_ELECTION.store(true, Ordering::Relaxed);
 
-        let grant_vote = self.logger().last_log_index <= request_vote.last_log_index
-            && self.replication.become_follower_if_term_higher_and_votable(
-                &request_vote.candidate_id,
-                request_vote.term,
+        let mut grant_vote = false;
+
+        // 1. If candidate's term is less than current term, reject
+        if request_vote.term < self.replication.term {
+            info!(
+                "Rejecting vote for {} (term {}). My term is higher ({}).",
+                request_vote.candidate_id, request_vote.term, self.replication.term
             );
+        }
+        // 2. If candidate's term is greater than or equal to current term
+        else {
+            // If candidate's term is higher, update own term and step down
+            if request_vote.term > self.replication.term {
+                self.replication.term = request_vote.term;
+                self.replication.vote_for(None); // Clear votedFor
+                self.step_down().await; // Ensure follower mode
+            }
+
+            // 3. Check if log is up-to-date and if not already voted in this term or voted for this candidate
+            if self
+                .replication
+                .is_log_up_to_date(request_vote.last_log_index, request_vote.last_log_term)
+                && self.replication.election_state.is_votable(&request_vote.candidate_id)
+            {
+                self.replication.vote_for(Some(request_vote.candidate_id.clone()));
+                grant_vote = true;
+            }
+            REQUESTS_BLOCKED_BY_ELECTION.store(true, Ordering::Relaxed);
+        }
 
         info!(
             "Voting for {} with term {} and granted: {grant_vote}",
             request_vote.candidate_id, request_vote.term
         );
 
-        // ! vote may have been made for requester, maybe not. Therefore, we have to set the term for self
         let vote = ElectionVote { term: self.replication.term, vote_granted: grant_vote };
         let Some(peer) = self.find_replica_mut(&request_vote.candidate_id) else {
             return;
@@ -1041,7 +1063,7 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
     /// 2) step down operation is given from user
     async fn step_down(&mut self) {
         self.replication.vote_for(None);
-        self.heartbeat_scheduler.turn_follower_mode().await;
+        self.heartbeat_scheduler.turn_follower_mode();
     }
 
     async fn become_leader(&mut self) {

duva/src/domains/cluster_actors/actor/heartbeat_scheduler.rs

@@ -124,7 +124,7 @@ impl HeartBeatScheduler {
         self.controller = controller;
     }
 
-    pub(crate) async fn turn_follower_mode(&mut self) {
+    pub(crate) fn turn_follower_mode(&mut self) {
         let controller = match self.controller.take() {
             | Some(SchedulerMode::Leader(sender)) => {
                 sender.send(());

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

@@ -131,7 +131,7 @@ async fn test_vote_election_deny_vote_older_log() {
 
     assert_expected_queryio(
         &candidate_fake_buf,
-        ElectionVote { term: initial_term, vote_granted: false },
+        ElectionVote { term: initial_term + 1, vote_granted: false },
     )
     .await;
 }

duva/src/domains/cluster_actors/replication.rs

@@ -1,5 +1,6 @@
 use super::consensus::election::ElectionState;
 
+use crate::domains::operation_logs::interfaces::TWriteAheadLog;
 use crate::domains::operation_logs::logger::ReplicatedLogs;
 use crate::domains::peers::command::BannedPeer;
 use crate::domains::peers::command::HeartBeat;
@@ -25,7 +26,7 @@ pub(crate) struct ReplicationState<T> {
     pub(crate) last_applied: u64,
 }
 
-impl<T> ReplicationState<T> {
+impl<T: TWriteAheadLog> ReplicationState<T> {
     pub(crate) fn new(
         replid: ReplicationId,
         role: ReplicationRole,
@@ -91,31 +92,6 @@ impl<T> ReplicationState<T> {
         }
     }
 
-    pub(super) fn become_follower_if_term_higher_and_votable(
-        &mut self,
-        candidate_id: &PeerIdentifier,
-        election_term: u64,
-    ) -> bool {
-        // If the candidate's term is less than mine -> reject
-        if election_term < self.term {
-            return false;
-        }
-
-        // When a node sees a higher term, it must forget any vote it cast in a prior term, because:
-        if election_term > self.term {
-            self.term = election_term;
-            self.vote_for(None);
-        }
-
-        if !self.election_state.is_votable(candidate_id) {
-            return false;
-        }
-
-        self.vote_for(Some(candidate_id.clone()));
-
-        true
-    }
-
     pub(super) fn vote_for(&mut self, leader_id: Option<PeerIdentifier>) {
         self.election_state = ElectionState::Follower { voted_for: leader_id };
         self.role = ReplicationRole::Follower;
@@ -124,6 +100,20 @@ impl<T> ReplicationState<T> {
     pub(crate) fn is_leader(&self) -> bool {
         self.role == ReplicationRole::Leader
     }
+
+    pub(super) fn is_log_up_to_date(
+        &self,
+        candidate_last_log_index: u64,
+        candidate_last_log_term: u64,
+    ) -> bool {
+        if candidate_last_log_term > self.logger.last_log_term {
+            true
+        } else if candidate_last_log_term == self.logger.last_log_term {
+            candidate_last_log_index >= self.logger.last_log_index
+        } else {
+            false
+        }
+    }
 }
 
 pub(crate) fn time_in_secs() -> anyhow::Result<u64> {

duva/tests/replication_ops/mod.rs

@@ -12,10 +12,14 @@ fn panic_if_election_not_done(order: &str, port1: u16, port2: u16, num_possible_
     let mut first_election_cnt = 0;
     let mut flag = false;
     let mut h1 = Client::new(port1);
+    let mut h2 = Client::new(port2);
 
     let start = std::time::Instant::now();
     while first_election_cnt < 50 {
-        let res = h1.send_and_get_vec("role", num_possible_nodes);
+        let mut res = h1.send_and_get_vec("role", num_possible_nodes);
+        if res.is_empty() {
+            res = h2.send_and_get_vec("role", num_possible_nodes);
+        }
         println!(
             "[{}ms] Poll {}: port1={} port2={} res={:?}",
             start.elapsed().as_millis(),