Merge pull request #451 from Migorithm/fix/450

fix: electiont test

Author: Migorithm

duva/src/domains/cluster_actors/heartbeats/scheduler.rs

@@ -1,7 +1,10 @@
 use crate::domains::cluster_actors::commands::ClusterCommand;
-use std::time::Duration;
+use std::{ops::Range, time::Duration};
 use tokio::{select, sync::mpsc::Sender, time::interval};
 const LEADER_HEARTBEAT_INTERVAL: u64 = 300;
+pub const LEADER_HEARTBEAT_INTERVAL_MAX: u64 = LEADER_HEARTBEAT_INTERVAL * 5;
+const LEADER_HEARTBEAT_INTERVAL_RANGE: Range<u64> =
+    LEADER_HEARTBEAT_INTERVAL * 3..LEADER_HEARTBEAT_INTERVAL_MAX;
 
 #[derive(Debug)]
 pub(crate) struct HeartBeatScheduler {
@@ -80,7 +83,7 @@ impl HeartBeatScheduler {
                             ElectionTimeOutCommand::Ping => {},
                         }
                     },
-                    _ =  tokio::time::sleep(Duration::from_millis(rand::random_range(LEADER_HEARTBEAT_INTERVAL*3..LEADER_HEARTBEAT_INTERVAL*5)))=>{
+                    _ =  tokio::time::sleep(Duration::from_millis(rand::random_range(LEADER_HEARTBEAT_INTERVAL_RANGE)))=>{
                         eprintln!("\x1b[33m[WARN] Election timeout\x1b[0m");
                         let _ = cluster_handler.send(ClusterCommand::StartLeaderElection).await;
 

duva/tests/common.rs

@@ -186,6 +186,8 @@ pub fn run_server_process(env: &ServerEnv) -> TestProcessChild {
     let mut command = Command::new("cargo");
     command.args([
         "run",
+        "-p",
+        "duva",
         "--",
         "--port",
         &env.port.to_string(),

duva/tests/replication_ops/test_leader_election.rs

@@ -1,7 +1,10 @@
 use std::{thread::sleep, time::Duration};
 
 use crate::common::{ServerEnv, array, check_internodes_communication, spawn_server_process};
-use duva::clients::ClientStreamHandler;
+use duva::{
+    clients::ClientStreamHandler,
+    domains::cluster_actors::heartbeats::scheduler::LEADER_HEARTBEAT_INTERVAL_MAX,
+};
 
 #[tokio::test]
 async fn test_leader_election() {
@@ -22,7 +25,7 @@ async fn test_leader_election() {
 
     // WHEN
     leader_p.kill().unwrap();
-    sleep(Duration::from_secs(2));
+    sleep(Duration::from_millis(LEADER_HEARTBEAT_INTERVAL_MAX));
 
     // THEN
     let mut flag = false;
@@ -59,7 +62,7 @@ async fn test_set_twice_after_election() {
 
     // WHEN
     leader_p.kill().unwrap();
-    sleep(Duration::from_secs(1));
+    sleep(Duration::from_millis(LEADER_HEARTBEAT_INTERVAL_MAX));
 
     let mut flag = false;
     for f in [&follower_p1, &follower_p2] {
@@ -96,7 +99,7 @@ async fn test_leader_election_twice() {
 
     // !first leader is killed -> election happens
     leader_p.kill().unwrap();
-    sleep(Duration::from_secs(1));
+    sleep(Duration::from_millis(LEADER_HEARTBEAT_INTERVAL_MAX));
 
     let mut processes = vec![];
 
@@ -109,11 +112,12 @@ async fn test_leader_election_twice() {
         }
         let new_process =
             spawn_server_process(&ServerEnv::default().with_leader_bind_addr(f.bind_addr().into()));
-        sleep(Duration::from_secs(1));
+        sleep(Duration::from_millis(LEADER_HEARTBEAT_INTERVAL_MAX));
 
         // WHEN
         // ! second leader is killed -> election happens
         f.kill().unwrap();
+        sleep(Duration::from_millis(LEADER_HEARTBEAT_INTERVAL_MAX));
         processes.push(new_process);
     }
     assert_eq!(processes.len(), 2);