Finished all the logic. Missing documentation, tests and metrics.

node/actors/bft/src/replica/misc.rs → node/actors/bft/src/chonky_bft/misc.rs

@@ -36,7 +36,6 @@ impl StateMachine {
     /// Tries to build a finalized block from the given CommitQC. We simply search our
     /// block proposal cache for the matching block, and if we find it we build the block.
     /// If this method succeeds, it sends the finalized block to the executor.
-    #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) async fn save_block(
         &mut self,
         ctx: &ctx::Ctx,

node/actors/bft/src/replica/mod.rs → node/actors/bft/src/chonky_bft/mod.rs

@@ -17,12 +17,14 @@ mod commit;
 mod misc;
 mod new_view;
 mod proposal;
+pub(crate) mod proposer;
+mod timeout;
+
 #[cfg(test)]
 mod tests;
-mod timeout;
 
-/// The StateMachine struct contains the state of the replica. It is responsible
-/// for validating and voting on blocks. When participating in consensus we are always a replica.
+/// The StateMachine struct contains the state of the replica and implements all the
+/// logic of ChonkyBFT.
 #[derive(Debug)]
 pub(crate) struct StateMachine {
     /// Consensus configuration.
@@ -31,6 +33,9 @@ pub(crate) struct StateMachine {
     pub(super) outbound_pipe: OutputSender,
     /// Pipe through which replica receives network requests.
     inbound_pipe: sync::prunable_mpsc::Receiver<ConsensusReq>,
+    /// The sender part of the justification watch. This is used to set the justification
+    /// and notify the proposer loop.
+    pub(crate) justification_watch: sync::watch::Sender<Option<validator::ProposalJustification>>,
 
     /// The current view number.
     pub(crate) view_number: validator::ViewNumber,
@@ -92,6 +97,8 @@ impl StateMachine {
             StateMachine::inbound_selection_function,
         );
 
+        let (justification_sender, _) = sync::watch::channel(None);
+
         let this = Self {
             config,
             outbound_pipe,
@@ -106,6 +113,7 @@ impl StateMachine {
             commit_qcs_cache: BTreeMap::new(),
             timeout_views_cache: BTreeMap::new(),
             timeout_qcs_cache: BTreeMap::new(),
+            justification_watch: justification_sender,
             timeout_deadline: time::Deadline::Finite(ctx.now() + Self::TIMEOUT_DURATION),
             phase_start: ctx.now(),
         };

node/actors/bft/src/replica/new_view.rs → node/actors/bft/src/chonky_bft/new_view.rs

@@ -115,6 +115,7 @@ impl StateMachine {
         // Update the state machine.
         self.view_number = view;
         self.phase = validator::Phase::Prepare;
+        // TODO: Update the proposer channel.
 
         // Clear the block proposal cache.
         if let Some(qc) = self.high_commit_qc.as_ref() {

node/actors/bft/src/chonky_bft/proposer.rs

@@ -0,0 +1,82 @@
+use crate::{metrics, Config, OutputSender};
+use std::sync::Arc;
+use zksync_concurrency::{ctx, error::Wrap as _, sync};
+use zksync_consensus_network::io::ConsensusInputMessage;
+use zksync_consensus_roles::validator;
+
+/// In a loop, receives a PrepareQC and sends a LeaderPrepare containing it.
+/// Every subsequent PrepareQC has to be for a higher view than the previous one (otherwise it
+/// is skipped). In case payload generation takes too long, some PrepareQC may be elided, so
+/// that the validator doesn't spend time on generating payloads for already expired views.
+pub(crate) async fn run_proposer(
+    ctx: &ctx::Ctx,
+    cfg: Arc<Config>,
+    pipe: OutputSender,
+    mut justification_watch: sync::watch::Receiver<Option<validator::ProposalJustification>>,
+) -> ctx::Result<()> {
+    loop {
+        let Some(justification) = sync::changed(ctx, &mut justification_watch).await?.clone()
+        else {
+            continue;
+        };
+
+        let genesis = cfg.genesis();
+
+        // If we are not the leader for this view, skip it.
+        if genesis.view_leader(justification.view().number) != cfg.secret_key.public() {
+            continue;
+        }
+
+        // Get the block number and check if this must be a reproposal.
+        let (block_number, opt_block_hash) = justification.get_implied_block(genesis);
+
+        let proposal_payload = match opt_block_hash {
+            // There was some proposal last view that a subquorum of replicas
+            // voted for and could have been finalized. We need to repropose it.
+            Some(_) => None,
+            // The previous proposal was finalized, so we can propose a new block.
+            None => {
+                // Defensively assume that PayloadManager cannot propose until the previous block is stored.
+                // if we don't have the previous block, this call will halt until the other replicas timeout.
+                // This is fine as we can just not propose anything and let our turn end. Eventually, some other
+                // replica will produce some block with this block number and this function will unblock.
+                if let Some(prev) = block_number.prev() {
+                    cfg.block_store.wait_until_persisted(ctx, prev).await?;
+                }
+
+                let payload = cfg
+                    .payload_manager
+                    .propose(ctx, block_number)
+                    .await
+                    .wrap("payload_manager.propose()")?;
+
+                if payload.0.len() > cfg.max_payload_size {
+                    return Err(anyhow::format_err!(
+                        "proposed payload too large: got {}B, max {}B",
+                        payload.0.len(),
+                        cfg.max_payload_size
+                    )
+                    .into());
+                }
+
+                metrics::METRICS
+                    .leader_proposal_payload_size
+                    .observe(payload.0.len());
+
+                Some(payload)
+            }
+        };
+
+        // Broadcast our proposal to all replicas (ourselves included).
+        let msg = cfg
+            .secret_key
+            .sign_msg(validator::ConsensusMsg::LeaderProposal(
+                validator::LeaderProposal {
+                    proposal_payload,
+                    justification,
+                },
+            ));
+
+        pipe.send(ConsensusInputMessage { message: msg }.into());
+    }
+}