feat(consensus): single height consensus returns Decision with Precommits (#2206)

Author: matan-starkware

crates/sequencing/papyrus_consensus/src/lib.rs

@@ -12,7 +12,14 @@ use papyrus_protobuf::consensus::{ConsensusMessage, Proposal};
 use single_height_consensus::SingleHeightConsensus;
 use starknet_api::block::{BlockHash, BlockNumber};
 use tracing::{debug, info, instrument};
-use types::{ConsensusBlock, ConsensusContext, ConsensusError, ProposalInit, ValidatorId};
+use types::{
+    ConsensusBlock,
+    ConsensusContext,
+    ConsensusError,
+    Decision,
+    ProposalInit,
+    ValidatorId,
+};
 
 pub mod config;
 #[allow(missing_docs)]
@@ -36,7 +43,7 @@ async fn run_height<BlockT: ConsensusBlock>(
     validator_id: ValidatorId,
     network_receiver: &mut SubscriberReceiver<ConsensusMessage>,
     cached_messages: &mut Vec<ConsensusMessage>,
-) -> Result<BlockT, ConsensusError>
+) -> Result<Decision<BlockT>, ConsensusError>
 where
     ProposalWrapper:
         Into<(ProposalInit, mpsc::Receiver<BlockT::ProposalChunk>, oneshot::Receiver<BlockHash>)>,
@@ -75,7 +82,7 @@ where
             continue;
         }
 
-        let maybe_block = match message {
+        let maybe_decision = match message {
             ConsensusMessage::Proposal(proposal) => {
                 // Special case due to fake streaming.
                 let (proposal_init, content_receiver, fin_receiver) =
@@ -85,8 +92,8 @@ where
             _ => shc.handle_message(message).await?,
         };
 
-        if let Some(block) = maybe_block {
-            return Ok(block);
+        if let Some(decision) = maybe_decision {
+            return Ok(decision);
         }
     }
 }
@@ -107,7 +114,7 @@ where
     let mut current_height = start_height;
     let mut future_messages = Vec::new();
     loop {
-        let block = run_height(
+        let decision = run_height(
             Arc::clone(&context),
             current_height,
             validator_id,
@@ -118,8 +125,9 @@ where
 
         info!(
             "Finished consensus for height: {current_height}. Agreed on block with id: {:x}",
-            block.id().0
+            decision.block.id().0
         );
+        debug!("Decision: {:?}", decision);
         metrics::gauge!(papyrus_metrics::PAPYRUS_CONSENSUS_HEIGHT, current_height.0 as f64);
         current_height = current_height.unchecked_next();
     }

crates/sequencing/papyrus_consensus/src/single_height_consensus.rs

@@ -15,6 +15,7 @@ use crate::types::{
     ConsensusBlock,
     ConsensusContext,
     ConsensusError,
+    Decision,
     ProposalInit,
     Round,
     ValidatorId,
@@ -26,10 +27,7 @@ const ROUND_ZERO: Round = 0;
 /// call to `start`, which is relevant if we are the proposer for this height's first round.
 /// SingleHeightConsensus receives messages directly as parameters to function calls. It can send
 /// out messages "directly" to the network, and returning a decision to the caller.
-pub(crate) struct SingleHeightConsensus<BlockT>
-where
-    BlockT: ConsensusBlock,
-{
+pub(crate) struct SingleHeightConsensus<BlockT: ConsensusBlock> {
     height: BlockNumber,
     context: Arc<dyn ConsensusContext<Block = BlockT>>,
     validators: Vec<ValidatorId>,
@@ -40,10 +38,7 @@ where
     precommits: HashMap<(Round, ValidatorId), Vote>,
 }
 
-impl<BlockT> SingleHeightConsensus<BlockT>
-where
-    BlockT: ConsensusBlock,
-{
+impl<BlockT: ConsensusBlock> SingleHeightConsensus<BlockT> {
     pub(crate) async fn new(
         height: BlockNumber,
         context: Arc<dyn ConsensusContext<Block = BlockT>>,
@@ -65,7 +60,7 @@ where
     }
 
     #[instrument(skip(self), fields(height=self.height.0), level = "debug")]
-    pub(crate) async fn start(&mut self) -> Result<Option<BlockT>, ConsensusError> {
+    pub(crate) async fn start(&mut self) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         info!("Starting consensus with validators {:?}", self.validators);
         let events = self.state_machine.start();
         self.handle_state_machine_events(events).await
@@ -83,7 +78,7 @@ where
         init: ProposalInit,
         p2p_messages_receiver: mpsc::Receiver<<BlockT as ConsensusBlock>::ProposalChunk>,
         fin_receiver: oneshot::Receiver<BlockHash>,
-    ) -> Result<Option<BlockT>, ConsensusError> {
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         debug!(
             "Received proposal: proposal_height={}, proposer={:?}",
             init.height.0, init.proposer
@@ -137,7 +132,7 @@ where
     pub(crate) async fn handle_message(
         &mut self,
         message: ConsensusMessage,
-    ) -> Result<Option<BlockT>, ConsensusError> {
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         debug!("Received message: {:?}", message);
         match message {
             ConsensusMessage::Proposal(_) => {
@@ -148,7 +143,10 @@ where
     }
 
     #[instrument(skip_all)]
-    async fn handle_vote(&mut self, vote: Vote) -> Result<Option<BlockT>, ConsensusError> {
+    async fn handle_vote(
+        &mut self,
+        vote: Vote,
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         let (votes, sm_vote) = match vote.vote_type {
             VoteType::Prevote => {
                 (&mut self.prevotes, StateMachineEvent::Prevote(vote.block_hash, ROUND_ZERO))
@@ -180,7 +178,7 @@ where
     async fn handle_state_machine_events(
         &mut self,
         mut events: VecDeque<StateMachineEvent>,
-    ) -> Result<Option<BlockT>, ConsensusError> {
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         while let Some(event) = events.pop_front() {
             trace!("Handling event: {:?}", event);
             match event {
@@ -194,16 +192,7 @@ where
                     // sent this out when responding to a StartRound.
                 }
                 StateMachineEvent::Decision(block_hash, round) => {
-                    let block = self
-                        .proposals
-                        .remove(&round)
-                        .expect("StateMachine arrived at an unknown decision");
-                    assert_eq!(
-                        block.id(),
-                        block_hash,
-                        "StateMachine block hash should match the stored block"
-                    );
-                    return Ok(Some(block));
+                    return self.handle_state_machine_decision(block_hash, round).await;
                 }
                 StateMachineEvent::Prevote(block_hash, round) => {
                     self.handle_state_machine_vote(block_hash, round, VoteType::Prevote).await?;
@@ -255,12 +244,13 @@ where
         self.state_machine.handle_event(StateMachineEvent::StartRound(Some(id), round))
     }
 
+    #[instrument(skip_all)]
     async fn handle_state_machine_vote(
         &mut self,
         block_hash: BlockHash,
         round: Round,
         vote_type: VoteType,
-    ) -> Result<Option<BlockT>, ConsensusError> {
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
         let votes = match vote_type {
             VoteType::Prevote => &mut self.prevotes,
             VoteType::Precommit => &mut self.precommits,
@@ -273,4 +263,29 @@ where
         self.context.broadcast(ConsensusMessage::Vote(vote)).await?;
         Ok(None)
     }
+
+    #[instrument(skip_all)]
+    async fn handle_state_machine_decision(
+        &mut self,
+        block_hash: BlockHash,
+        round: Round,
+    ) -> Result<Option<Decision<BlockT>>, ConsensusError> {
+        let block =
+            self.proposals.remove(&round).expect("StateMachine arrived at an unknown decision");
+        assert_eq!(block.id(), block_hash, "StateMachine block hash should match the stored block");
+        let supporting_precommits: Vec<Vote> = self
+            .validators
+            .iter()
+            .filter_map(|v| {
+                let vote = self.precommits.get(&(round, *v))?;
+                if vote.block_hash != block_hash {
+                    return None;
+                }
+                Some(vote.clone())
+            })
+            .collect();
+        // TODO(matan): Check actual weights.
+        assert!(supporting_precommits.len() >= self.state_machine.quorum_size() as usize);
+        Ok(Some(Decision { precommits: supporting_precommits, block }))
+    }
 }

crates/sequencing/papyrus_consensus/src/single_height_consensus_test.rs

@@ -63,10 +63,23 @@ async fn proposer() {
 
     assert_eq!(shc.handle_message(prevote(block.id(), 0, 2_u32.into())).await, Ok(None));
     assert_eq!(shc.handle_message(prevote(block.id(), 0, 3_u32.into())).await, Ok(None));
-    assert_eq!(shc.handle_message(precommit(block.id(), 0, 2_u32.into())).await, Ok(None));
-    let decision =
-        shc.handle_message(precommit(block.id(), 0, 3_u32.into())).await.unwrap().unwrap();
-    assert_eq!(decision, block);
+
+    let precommits = vec![
+        precommit(block.id(), 0, 1_u32.into()),
+        precommit(BlockHash(Felt::TWO), 0, 4_u32.into()), // Ignores since disagrees.
+        precommit(block.id(), 0, 2_u32.into()),
+        precommit(block.id(), 0, 3_u32.into()),
+    ];
+    assert_eq!(shc.handle_message(precommits[1].clone()).await, Ok(None));
+    assert_eq!(shc.handle_message(precommits[2].clone()).await, Ok(None));
+    let decision = shc.handle_message(precommits[3].clone()).await.unwrap().unwrap();
+    assert_eq!(decision.block, block);
+    assert!(
+        decision
+            .precommits
+            .into_iter()
+            .all(|item| precommits.contains(&ConsensusMessage::Vote(item)))
+    );
 
     // Check the fin sent to the network.
     let fin = Arc::into_inner(fin_receiver).unwrap().take().unwrap().await.unwrap();
@@ -120,9 +133,19 @@ async fn validator() {
 
     assert_eq!(shc.handle_message(prevote(block.id(), 0, 2_u32.into())).await, Ok(None));
     assert_eq!(shc.handle_message(prevote(block.id(), 0, 3_u32.into())).await, Ok(None));
-    assert_eq!(shc.handle_message(precommit(block.id(), 0, 2_u32.into())).await, Ok(None));
 
-    let decision =
-        shc.handle_message(precommit(block.id(), 0, 3_u32.into())).await.unwrap().unwrap();
-    assert_eq!(decision, block);
+    let precommits = vec![
+        precommit(block.id(), 0, 2_u32.into()),
+        precommit(block.id(), 0, 3_u32.into()),
+        precommit(block.id(), 0, node_id),
+    ];
+    assert_eq!(shc.handle_message(precommits[0].clone()).await, Ok(None));
+    let decision = shc.handle_message(precommits[1].clone()).await.unwrap().unwrap();
+    assert_eq!(decision.block, block);
+    assert!(
+        decision
+            .precommits
+            .into_iter()
+            .all(|item| precommits.contains(&ConsensusMessage::Vote(item)))
+    );
 }

crates/sequencing/papyrus_consensus/src/state_machine.rs

@@ -84,6 +84,10 @@ impl StateMachine {
         }
     }
 
+    pub fn quorum_size(&self) -> u32 {
+        self.quorum
+    }
+
     /// Starts the state machine, effectively calling `StartRound(0)` from the paper. This is needed
     /// to trigger the first leader to propose. See [`StartRound`](StateMachineEvent::StartRound)
     pub fn start(&mut self) -> VecDeque<StateMachineEvent> {

crates/sequencing/papyrus_consensus/src/types.rs

@@ -2,9 +2,11 @@
 #[path = "types_test.rs"]
 mod types_test;
 
+use std::fmt::Debug;
+
 use async_trait::async_trait;
 use futures::channel::{mpsc, oneshot};
-use papyrus_protobuf::consensus::ConsensusMessage;
+use papyrus_protobuf::consensus::{ConsensusMessage, Vote};
 use starknet_api::block::{BlockHash, BlockNumber};
 use starknet_api::core::ContractAddress;
 
@@ -144,6 +146,21 @@ pub trait ConsensusContext: Send + Sync {
     ) -> Result<(), ConsensusError>;
 }
 
+#[derive(PartialEq)]
+pub struct Decision<BlockT: ConsensusBlock> {
+    pub precommits: Vec<Vote>,
+    pub block: BlockT,
+}
+
+impl<BlockT: ConsensusBlock> Debug for Decision<BlockT> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Decision")
+            .field("block_id", &self.block.id())
+            .field("precommits", &self.precommits)
+            .finish()
+    }
+}
+
 #[derive(PartialEq, Debug, Clone)]
 pub struct ProposalInit {
     pub height: BlockNumber,