feat(consensus): update consensus so that it runs with voting (#2189)

Currently the context is hard coded with 4 validators, so at least 3 must vote in favor.

Author: matan-starkware

crates/papyrus_protobuf/src/consensus.rs

@@ -29,3 +29,12 @@ pub enum ConsensusMessage {
     Proposal(Proposal),
     Vote(Vote),
 }
+
+impl ConsensusMessage {
+    pub fn height(&self) -> u64 {
+        match self {
+            ConsensusMessage::Proposal(proposal) => proposal.height,
+            ConsensusMessage::Vote(vote) => vote.height,
+        }
+    }
+}

crates/sequencing/papyrus_consensus/src/lib.rs

@@ -27,6 +27,69 @@ pub mod types;
 
 use futures::StreamExt;
 
+#[instrument(skip(context, validator_id, network_receiver, cached_messages), level = "info")]
+#[allow(missing_docs)]
+async fn run_height<BlockT: ConsensusBlock>(
+    context: Arc<dyn ConsensusContext<Block = BlockT>>,
+    height: BlockNumber,
+    validator_id: ValidatorId,
+    network_receiver: &mut SubscriberReceiver<ConsensusMessage>,
+    cached_messages: &mut Vec<ConsensusMessage>,
+) -> Result<BlockT, ConsensusError>
+where
+    ProposalWrapper:
+        Into<(ProposalInit, mpsc::Receiver<BlockT::ProposalChunk>, oneshot::Receiver<BlockHash>)>,
+{
+    let mut shc = SingleHeightConsensus::new(height, context, validator_id).await;
+
+    if let Some(decision) = shc.start().await? {
+        return Ok(decision);
+    }
+
+    let mut current_height_messages = Vec::new();
+    for msg in std::mem::take(cached_messages) {
+        match height.0.cmp(&msg.height()) {
+            std::cmp::Ordering::Less => cached_messages.push(msg),
+            std::cmp::Ordering::Equal => current_height_messages.push(msg),
+            std::cmp::Ordering::Greater => {}
+        }
+    }
+
+    loop {
+        let message = current_height_messages.pop().unwrap_or(
+            // TODO(matan): Handle parsing failures and utilize ReportCallback.
+            network_receiver
+                .next()
+                .await
+                .expect("Network receiver closed unexpectedly")
+                .0
+                .expect("Failed to parse consensus message"),
+        );
+
+        if message.height() != height.0 {
+            debug!("Received a message for a different height. {:?}", message);
+            if message.height() > height.0 {
+                cached_messages.push(message);
+            }
+            continue;
+        }
+
+        let maybe_block = match message {
+            ConsensusMessage::Proposal(proposal) => {
+                // Special case due to fake streaming.
+                let (proposal_init, content_receiver, fin_receiver) =
+                    ProposalWrapper(proposal).into();
+                shc.handle_proposal(proposal_init, content_receiver, fin_receiver).await?
+            }
+            _ => shc.handle_message(message).await?,
+        };
+
+        if let Some(block) = maybe_block {
+            return Ok(block);
+        }
+    }
+}
+
 // TODO(dvir): add test for this.
 #[instrument(skip(context, start_height, network_receiver), level = "info")]
 #[allow(missing_docs)]
@@ -41,30 +104,16 @@ where
         Into<(ProposalInit, mpsc::Receiver<BlockT::ProposalChunk>, oneshot::Receiver<BlockHash>)>,
 {
     let mut current_height = start_height;
+    let mut future_messages = Vec::new();
     loop {
-        debug!("Starting consensus for height {current_height}");
-        let mut shc =
-            SingleHeightConsensus::new(current_height, context.clone(), validator_id).await;
-
-        let block = if let Some(block) = shc.start().await? {
-            block
-        } else {
-            info!("Validator flow height {current_height}");
-            let ConsensusMessage::Proposal(proposal) = network_receiver
-                .next()
-                .await
-                .expect("Failed to receive a message from network")
-                .0
-                .expect("Network receiver closed unexpectedly")
-            else {
-                todo!("Handle votes");
-            };
-            let (proposal_init, content_receiver, fin_receiver) = ProposalWrapper(proposal).into();
-
-            shc.handle_proposal(proposal_init, content_receiver, fin_receiver)
-                .await?
-                .expect("Failed to handle proposal")
-        };
+        let block = run_height(
+            Arc::clone(&context),
+            current_height,
+            validator_id,
+            &mut network_receiver,
+            &mut future_messages,
+        )
+        .await?;
 
         info!(
             "Finished consensus for height: {current_height}. Agreed on block with id: {:x}",

crates/sequencing/papyrus_consensus/src/papyrus_consensus_context.rs

@@ -158,7 +158,7 @@ impl ConsensusContext for PapyrusConsensusContext {
     }
 
     async fn validators(&self, _height: BlockNumber) -> Vec<ValidatorId> {
-        vec![0u8.into(), 1u8.into()]
+        vec![0u8.into(), 1u8.into(), 2u8.into(), 3u8.into()]
     }
 
     fn proposer(&self, _validators: &[ValidatorId], height: BlockNumber) -> ValidatorId {
@@ -193,7 +193,13 @@ impl ConsensusContext for PapyrusConsensusContext {
                 transactions,
                 block_hash,
             };
-            debug!("Sending proposal: {proposal:?}");
+            debug!(
+                "Sending proposal: height={:?} id={:?} num_txs={} block_hash={:?}",
+                proposal.height,
+                proposal.proposer,
+                proposal.transactions.len(),
+                proposal.block_hash
+            );
 
             broadcast_sender
                 .lock()