refactor: consensus vote counting

src/domains/cluster_actors/actor.rs

@@ -295,14 +295,7 @@ impl ClusterActor {
     }
 
     pub(crate) fn track_replication_progress(&mut self, res: ReplicationResponse) {
-        if let Some(mut consensus) = self.consensus_tracker.take(&res.log_idx) {
-            println!("[INFO] Received acks for log index num: {}", res.log_idx);
-            consensus.increase_vote(res.from);
-
-            if let Some(consensus) = consensus.maybe_not_finished(res.log_idx) {
-                self.consensus_tracker.insert(res.log_idx, consensus);
-            }
-        }
+        self.consensus_tracker.track_progress(res.log_idx, res.from);
     }
 
     // After send_ack: Leader updates its knowledge of follower's progress

src/domains/cluster_actors/consensus/mod.rs

@@ -1,38 +1,36 @@
 pub mod voting;
 use super::commands::ConsensusClientResponse;
-use crate::make_smart_pointer;
+use crate::{domains::peers::identifier::PeerIdentifier, make_smart_pointer};
 
 use std::collections::HashMap;
 use tokio::sync::oneshot::Sender;
 use voting::ConsensusVoting;
 
 #[derive(Default, Debug)]
-pub struct LogConsensusTracker(
-    pub(crate) HashMap<u64, ConsensusVoting<Sender<ConsensusClientResponse>>>,
-);
+pub struct LogConsensusTracker(pub(crate) HashMap<u64, ConsensusVoting>);
 impl LogConsensusTracker {
     pub(crate) fn add(
         &mut self,
         key: u64,
         value: Sender<ConsensusClientResponse>,
         replica_count: usize,
     ) {
-        self.0.insert(
+        self.insert(
             key,
             ConsensusVoting {
                 callback: value,
-                pos_vt: 0, // no need for self vote
-                neg_vt: 0,
-                replica_count,
+                cnt: 0, // no need for self vote
                 voters: Vec::with_capacity(replica_count),
             },
         );
     }
-    pub(crate) fn take(
-        &mut self,
-        offset: &u64,
-    ) -> Option<ConsensusVoting<Sender<ConsensusClientResponse>>> {
-        self.0.remove(offset)
+
+    pub(crate) fn track_progress(&mut self, log_idx: u64, from: PeerIdentifier) {
+        if let Some(consensus) = self.remove(&log_idx) {
+            if let Some(consensus) = consensus.vote_and_maybe_stay_pending(log_idx, from) {
+                self.insert(log_idx, consensus);
+            }
+        }
     }
 }
-make_smart_pointer!(LogConsensusTracker, HashMap<u64, ConsensusVoting<Sender<ConsensusClientResponse>>>);
+make_smart_pointer!(LogConsensusTracker, HashMap<u64, ConsensusVoting>);

src/domains/cluster_actors/consensus/voting.rs

@@ -5,34 +5,39 @@ use tokio::sync::oneshot::Sender;
 pub(crate) type ReplicationVote = Sender<ConsensusClientResponse>;
 
 #[derive(Debug)]
-pub struct ConsensusVoting<T> {
-    pub(crate) callback: T,
-    pub(crate) pos_vt: u8,
-    pub(crate) neg_vt: u8,
-    pub(crate) replica_count: usize,
+pub struct ConsensusVoting {
+    pub(crate) callback: ReplicationVote,
+    pub(crate) cnt: u8,
     pub(crate) voters: Vec<PeerIdentifier>,
 }
-impl<T> ConsensusVoting<T> {
-    pub(crate) fn increase_vote(&mut self, voter: PeerIdentifier) {
-        if self.voters.iter().any(|v| v == &voter) {
-            return;
+impl ConsensusVoting {
+    pub(crate) fn vote_and_maybe_stay_pending(
+        mut self,
+        log_idx: u64,
+        from: PeerIdentifier,
+    ) -> Option<Self> {
+        if self.votable(&from) {
+            println!("[INFO] Received acks for log index num: {}", log_idx);
+            self.increase_vote(from);
         }
-        self.pos_vt += 1;
+
+        if self.cnt < self.get_required_votes() {
+            return Some(self);
+        }
+
+        let _ = self.callback.send(ConsensusClientResponse::LogIndex(Some(log_idx)));
+        None
+    }
+
+    fn increase_vote(&mut self, voter: PeerIdentifier) {
+        self.cnt += 1;
         self.voters.push(voter);
     }
 
     fn get_required_votes(&self) -> u8 {
-        ((self.replica_count as f64 + 1.0) / 2.0).ceil() as u8
+        ((self.voters.capacity() as f64 + 1.0) / 2.0).ceil() as u8
     }
-}
-
-impl ConsensusVoting<ReplicationVote> {
-    pub(crate) fn maybe_not_finished(self, log_index: u64) -> Option<Self> {
-        if self.pos_vt >= self.get_required_votes() {
-            let _ = self.callback.send(ConsensusClientResponse::LogIndex(Some(log_index)));
-            None
-        } else {
-            Some(self)
-        }
+    fn votable(&self, voter: &PeerIdentifier) -> bool {
+        !self.voters.iter().any(|v| v == voter)
     }
 }