apollo_consensus: prevent dup/conflict vote queuing during awaiting_finished_building (#10616)

Author: asmaastarkware

crates/apollo_consensus/src/single_height_consensus.rs

@@ -10,7 +10,9 @@ mod single_height_consensus_test;
 
 use std::collections::{HashSet, VecDeque};
 
+use crate::state_machine::VoteStatus;
 const REBROADCAST_LOG_PERIOD_SECS: u64 = 10;
+const DUPLICATE_VOTE_LOG_PERIOD_SECS: u64 = 10;
 
 use apollo_consensus_config::config::TimeoutsConfig;
 use apollo_infra_utils::trace_every_n_sec;
@@ -274,29 +276,32 @@ impl SingleHeightConsensus {
             return Ok(ShcReturn::Requests(VecDeque::new()));
         }
 
-        // Check duplicates/conflicts from SM stored votes.
-        let (votes_map, sm_vote) = match vote.vote_type {
-            VoteType::Prevote => {
-                (self.state_machine.prevotes_ref(), StateMachineEvent::Prevote(vote.clone()))
-            }
-            VoteType::Precommit => {
-                (self.state_machine.precommits_ref(), StateMachineEvent::Precommit(vote.clone()))
-            }
-        };
-        if let Some((old_vote, _)) = votes_map.get(&(vote.round, vote.voter)) {
-            if old_vote.proposal_commitment == vote.proposal_commitment {
+        // Check if vote has already been received.
+        match self.state_machine.received_vote(&vote) {
+            VoteStatus::Duplicate => {
                 // Duplicate - ignore.
+                trace_every_n_sec!(
+                    DUPLICATE_VOTE_LOG_PERIOD_SECS,
+                    "Ignoring duplicate vote: {vote:?}"
+                );
                 return Ok(ShcReturn::Requests(VecDeque::new()));
-            } else {
+            }
+            VoteStatus::Conflict(old_vote, new_vote) => {
                 // Conflict - ignore and record.
-                warn!("Conflicting votes: old={old_vote:?}, new={vote:?}");
+                warn!("Conflicting votes: old={old_vote:?}, new={new_vote:?}");
                 CONSENSUS_CONFLICTING_VOTES.increment(1);
                 return Ok(ShcReturn::Requests(VecDeque::new()));
             }
+            VoteStatus::New => {
+                // Vote is new, proceed to process it.
+            }
         }
 
         info!("Accepting {:?}", vote);
-        // TODO(Asmaa): consider calling handle_prevote/precommit instead of sending the vote event.
+        let sm_vote = match vote.vote_type {
+            VoteType::Prevote => StateMachineEvent::Prevote(vote),
+            VoteType::Precommit => StateMachineEvent::Precommit(vote),
+        };
         let requests = self.state_machine.handle_event(sm_vote, leader_fn);
         self.handle_state_machine_requests(requests)
     }

crates/apollo_consensus/src/single_height_consensus_test.rs

@@ -361,3 +361,52 @@ fn repropose() {
         assert!(reqs.is_empty());
     });
 }
+
+#[tokio::test]
+async fn duplicate_votes_during_awaiting_finished_building_are_ignored() {
+    // This test verifies that receiving 3 identical prevotes during awaiting_finished_building
+    // results in only one vote being processed, so no TimeoutPrevote is triggered.
+    let mut shc = SingleHeightConsensus::new(
+        BlockNumber(0),
+        false,
+        *PROPOSER_ID,
+        VALIDATORS.to_vec(),
+        QuorumType::Byzantine,
+        TIMEOUTS.clone(),
+    );
+    let leader_fn = |_round| -> ValidatorId { *PROPOSER_ID };
+    let ret = shc.start(&leader_fn).unwrap();
+    assert_matches!(ret, ShcReturn::Requests(mut reqs) => {
+        assert_matches!(reqs.pop_front(), Some(SMRequest::StartBuildProposal(ROUND_0)));
+        assert!(reqs.is_empty());
+    });
+
+    // Receive enough identical prevotes during awaiting_finished_building to trigger Timeout
+    // (if they weren't duplicates)
+    let duplicate_vote = prevote(Some(BLOCK.id.0), HEIGHT_0, ROUND_0, *VALIDATOR_ID_1);
+
+    // First vote gets queued
+    assert_matches!(
+        shc.handle_vote(&leader_fn, duplicate_vote.clone()).unwrap(),
+        ShcReturn::Requests(reqs) if reqs.is_empty()
+    );
+
+    // Remaining votes are duplicates - should be ignored
+    for _ in 1..VALIDATORS.len() {
+        assert_matches!(
+            shc.handle_vote(&leader_fn, duplicate_vote.clone()).unwrap(),
+            ShcReturn::Requests(reqs) if reqs.is_empty()
+        );
+    }
+
+    // Finish building - processes the queue
+    // Only one vote was queued (duplicates were ignored), so no TimeoutPrevote should be triggered,
+    // only a broadcast vote
+    let ret = shc
+        .handle_event(&leader_fn, StateMachineEvent::FinishedBuilding(Some(BLOCK.id), ROUND_0))
+        .unwrap();
+    assert_matches!(ret, ShcReturn::Requests(mut reqs) => {
+        assert_matches!(reqs.pop_front(), Some(SMRequest::BroadcastVote(v)) if v.vote_type == VoteType::Prevote && v.round == ROUND_0);
+        assert!(reqs.is_empty());
+    });
+}

crates/apollo_consensus/src/state_machine.rs

@@ -83,6 +83,17 @@ pub(crate) enum Step {
     Precommit,
 }
 
+/// Status of a vote when checking if it has been received.
+#[derive(Debug, PartialEq)]
+pub(crate) enum VoteStatus {
+    /// Vote is new and has not been received yet.
+    New,
+    /// Vote is a duplicate (same proposal_commitment) - already received or queued.
+    Duplicate,
+    /// Vote conflicts with an existing vote (different proposal_commitment).
+    Conflict(Vote, Vote),
+}
+
 /// State Machine. Major assumptions:
 /// 1. SHC handles: authentication, replays, and conflicts.
 /// 2. SM must handle "out of order" messages (E.g. vote arrives before proposal).
@@ -167,10 +178,6 @@ impl StateMachine {
         self.height
     }
 
-    pub(crate) fn prevotes_ref(&self) -> &VotesMap {
-        &self.prevotes
-    }
-
     pub(crate) fn precommits_ref(&self) -> &VotesMap {
         &self.precommits
     }
@@ -191,6 +198,44 @@ impl StateMachine {
         self.last_self_precommit.clone()
     }
 
+    /// Check if a vote has already been received (either in the vote maps or queued).
+    /// Returns the status of the vote: NotReceived, Duplicate, or Conflict.
+    pub(crate) fn received_vote(&self, vote: &Vote) -> VoteStatus {
+        let determine_status = |old: &Vote, new: &Vote| {
+            if old.proposal_commitment == new.proposal_commitment {
+                VoteStatus::Duplicate
+            } else {
+                VoteStatus::Conflict(old.clone(), new.clone())
+            }
+        };
+
+        // Check Map
+        let key = (vote.round, vote.voter);
+        let map_entry = match vote.vote_type {
+            VoteType::Prevote => self.prevotes.get(&key),
+            VoteType::Precommit => self.precommits.get(&key),
+        };
+
+        if let Some((old_vote, _)) = map_entry {
+            return determine_status(old_vote, vote);
+        }
+
+        // Check Queue
+        for event in &self.events_queue {
+            let queued_vote = match (event, vote.vote_type) {
+                (StateMachineEvent::Prevote(v), VoteType::Prevote) => v,
+                (StateMachineEvent::Precommit(v), VoteType::Precommit) => v,
+                _ => continue,
+            };
+
+            if queued_vote.round == vote.round && queued_vote.voter == vote.voter {
+                return determine_status(queued_vote, vote);
+            }
+        }
+
+        VoteStatus::New
+    }
+
     fn make_self_vote(
         &mut self,
         vote_type: VoteType,
@@ -274,10 +319,7 @@ impl StateMachine {
         self.handle_enqueued_events(leader_fn)
     }
 
-    pub(crate) fn handle_enqueued_events<LeaderFn>(
-        &mut self,
-        leader_fn: &LeaderFn,
-    ) -> VecDeque<SMRequest>
+    fn handle_enqueued_events<LeaderFn>(&mut self, leader_fn: &LeaderFn) -> VecDeque<SMRequest>
     where
         LeaderFn: Fn(Round) -> ValidatorId,
     {
@@ -311,7 +353,7 @@ impl StateMachine {
         output_requests
     }
 
-    pub(crate) fn handle_event_internal<LeaderFn>(
+    fn handle_event_internal<LeaderFn>(
         &mut self,
         event: StateMachineEvent,
         leader_fn: &LeaderFn,
@@ -344,7 +386,7 @@ impl StateMachine {
         }
     }
 
-    pub(crate) fn handle_finished_building<LeaderFn>(
+    fn handle_finished_building<LeaderFn>(
         &mut self,
         proposal_id: Option<ProposalCommitment>,
         round: u32,
@@ -362,7 +404,7 @@ impl StateMachine {
         self.map_round_to_upons(round, leader_fn)
     }
 
-    pub(crate) fn handle_finished_validation<LeaderFn>(
+    fn handle_finished_validation<LeaderFn>(
         &mut self,
         proposal_id: Option<ProposalCommitment>,
         round: u32,
@@ -377,7 +419,7 @@ impl StateMachine {
         self.map_round_to_upons(round, leader_fn)
     }
 
-    pub(crate) fn handle_timeout_propose(&mut self, round: u32) -> VecDeque<SMRequest> {
+    fn handle_timeout_propose(&mut self, round: u32) -> VecDeque<SMRequest> {
         if self.step != Step::Propose || round != self.round {
             return VecDeque::new();
         };
@@ -392,11 +434,7 @@ impl StateMachine {
     }
 
     // A prevote from a peer node.
-    pub(crate) fn handle_prevote<LeaderFn>(
-        &mut self,
-        vote: Vote,
-        leader_fn: &LeaderFn,
-    ) -> VecDeque<SMRequest>
+    fn handle_prevote<LeaderFn>(&mut self, vote: Vote, leader_fn: &LeaderFn) -> VecDeque<SMRequest>
     where
         LeaderFn: Fn(Round) -> ValidatorId,
     {
@@ -411,7 +449,7 @@ impl StateMachine {
         self.map_round_to_upons(round, leader_fn)
     }
 
-    pub(crate) fn handle_timeout_prevote(&mut self, round: u32) -> VecDeque<SMRequest> {
+    fn handle_timeout_prevote(&mut self, round: u32) -> VecDeque<SMRequest> {
         if self.step != Step::Prevote || round != self.round {
             return VecDeque::new();
         };