apollo_consensus: SM tracks last self votes not the SHC

Author: asmaastarkware

crates/apollo_consensus/src/single_height_consensus.rs

@@ -137,8 +137,6 @@ pub(crate) struct SingleHeightConsensus {
     state_machine: StateMachine,
     // Tracks rounds for which we started validating a proposal to avoid duplicate validations.
     pending_validation_rounds: HashSet<Round>,
-    last_prevote: Option<Vote>,
-    last_precommit: Option<Vote>,
 }
 
 impl SingleHeightConsensus {
@@ -154,14 +152,7 @@ impl SingleHeightConsensus {
         let n_validators =
             u64::try_from(validators.len()).expect("Should have way less than u64::MAX validators");
         let state_machine = StateMachine::new(height, id, n_validators, is_observer, quorum_type);
-        Self {
-            validators,
-            timeouts,
-            state_machine,
-            pending_validation_rounds: HashSet::new(),
-            last_prevote: None,
-            last_precommit: None,
-        }
+        Self { validators, timeouts, state_machine, pending_validation_rounds: HashSet::new() }
     }
 
     pub(crate) fn current_round(&self) -> Round {
@@ -242,7 +233,7 @@ impl SingleHeightConsensus {
                 self.handle_timeout(context, event).await
             }
             StateMachineEvent::Prevote(vote) => {
-                let Some(last_vote) = &self.last_prevote else {
+                let Some(last_vote) = self.state_machine.last_self_prevote() else {
                     return Err(ConsensusError::InternalInconsistency(
                         "No prevote to send".to_string(),
                     ));
@@ -252,14 +243,14 @@ impl SingleHeightConsensus {
                     return Ok(ShcReturn::Tasks(Vec::new()));
                 }
                 trace_every_n_sec!(REBROADCAST_LOG_PERIOD_SECS, "Rebroadcasting {last_vote:?}");
-                context.broadcast(last_vote.clone()).await?;
+                context.broadcast(last_vote).await?;
                 Ok(ShcReturn::Tasks(vec![ShcTask::Prevote(
                     self.timeouts.prevote_timeout,
                     StateMachineEvent::Prevote(vote),
                 )]))
             }
             StateMachineEvent::Precommit(vote) => {
-                let Some(last_vote) = &self.last_precommit else {
+                let Some(last_vote) = self.state_machine.last_self_precommit() else {
                     return Err(ConsensusError::InternalInconsistency(
                         "No precommit to send".to_string(),
                     ));
@@ -269,7 +260,7 @@ impl SingleHeightConsensus {
                     return Ok(ShcReturn::Tasks(Vec::new()));
                 }
                 trace_every_n_sec!(REBROADCAST_LOG_PERIOD_SECS, "Rebroadcasting {last_vote:?}");
-                context.broadcast(last_vote.clone()).await?;
+                context.broadcast(last_vote).await?;
                 Ok(ShcReturn::Tasks(vec![ShcTask::Precommit(
                     self.timeouts.precommit_timeout,
                     StateMachineEvent::Precommit(vote),
@@ -488,37 +479,18 @@ impl SingleHeightConsensus {
         context: &mut ContextT,
         vote: Vote,
     ) -> Result<Vec<ShcTask>, ConsensusError> {
-        let (last_vote, task) = match vote.vote_type {
-            VoteType::Prevote => (
-                &mut self.last_prevote,
-                ShcTask::Prevote(
-                    self.timeouts.prevote_timeout,
-                    StateMachineEvent::Prevote(vote.clone()),
-                ),
+        let task = match vote.vote_type {
+            VoteType::Prevote => ShcTask::Prevote(
+                self.timeouts.prevote_timeout,
+                StateMachineEvent::Prevote(vote.clone()),
             ),
-            VoteType::Precommit => (
-                &mut self.last_precommit,
-                ShcTask::Precommit(
-                    self.timeouts.precommit_timeout,
-                    StateMachineEvent::Precommit(vote.clone()),
-                ),
+            VoteType::Precommit => ShcTask::Precommit(
+                self.timeouts.precommit_timeout,
+                StateMachineEvent::Precommit(vote.clone()),
             ),
         };
         // Ensure the voter matches this node.
         assert_eq!(vote.voter, self.state_machine.id());
-        *last_vote = match last_vote {
-            None => Some(vote.clone()),
-            Some(last_vote) if vote.round > last_vote.round => Some(vote.clone()),
-            Some(_) => {
-                // According to the Tendermint paper, the state machine should only vote for its
-                // current round. It should monotonically increase its round. It should only vote
-                // once per step.
-                return Err(ConsensusError::InternalInconsistency(format!(
-                    "State machine must progress in time: last_vote: {last_vote:?} new_vote: \
-                     {vote:?}",
-                )));
-            }
-        };
 
         info!("Broadcasting {vote:?}");
         context.broadcast(vote).await?;

crates/apollo_consensus/src/state_machine.rs

@@ -90,6 +90,9 @@ pub(crate) struct StateMachine {
     prevote_quorum: HashSet<Round>,
     mixed_prevote_quorum: HashSet<Round>,
     mixed_precommit_quorum: HashSet<Round>,
+    // Tracks the latest self votes for efficient rebroadcasts.
+    last_self_prevote: Option<Vote>,
+    last_self_precommit: Option<Vote>,
 }
 
 impl StateMachine {
@@ -122,6 +125,8 @@ impl StateMachine {
             prevote_quorum: HashSet::new(),
             mixed_prevote_quorum: HashSet::new(),
             mixed_precommit_quorum: HashSet::new(),
+            last_self_prevote: None,
+            last_self_precommit: None,
         }
     }
 
@@ -159,13 +164,42 @@ impl StateMachine {
         &self.proposals
     }
 
+    pub(crate) fn last_self_prevote(&self) -> Option<Vote> {
+        self.last_self_prevote.clone()
+    }
+
+    pub(crate) fn last_self_precommit(&self) -> Option<Vote> {
+        self.last_self_precommit.clone()
+    }
+
     fn make_self_vote(
-        &self,
+        &mut self,
         vote_type: VoteType,
         round: Round,
         proposal_commitment: Option<ProposalCommitment>,
     ) -> Vote {
-        Vote { vote_type, height: self.height.0, round, proposal_commitment, voter: self.id }
+        let vt = vote_type.clone();
+        let vote = Vote {
+            vote_type: vt,
+            height: self.height.0,
+            round,
+            proposal_commitment,
+            voter: self.id,
+        };
+        // update the latest self vote.
+        match vote_type {
+            VoteType::Prevote => match &self.last_self_prevote {
+                None => self.last_self_prevote = Some(vote.clone()),
+                Some(last) if round > last.round => self.last_self_prevote = Some(vote.clone()),
+                _ => {}
+            },
+            VoteType::Precommit => match &self.last_self_precommit {
+                None => self.last_self_precommit = Some(vote.clone()),
+                Some(last) if round > last.round => self.last_self_precommit = Some(vote.clone()),
+                _ => {}
+            },
+        }
+        vote
     }
 
     /// Starts the state machine, effectively calling `StartRound(0)` from the paper. This is