made max_payload_size configurable

node/actors/bft/src/config.rs

@@ -12,6 +12,9 @@ pub struct Config {
     pub secret_key: validator::SecretKey,
     /// A vector of public keys for all the validators in the network.
     pub validator_set: validator::ValidatorSet,
+    /// The maximum size of the payload of a block, in bytes. We will
+    /// reject blocks with payloads larger than this.
+    pub max_payload_size: usize,
     /// Block store.
     pub block_store: Arc<storage::BlockStore>,
     /// Replica store.
@@ -21,10 +24,6 @@ pub struct Config {
 }
 
 impl Config {
-    /// The maximum size of the payload of a block, in bytes. We will
-    /// reject blocks with payloads larger than this.
-    pub(crate) const PAYLOAD_MAX_SIZE: usize = 500 * zksync_protobuf::kB;
-
     /// Computes the validator for the given view.
     #[instrument(level = "trace", ret)]
     pub fn view_leader(&self, view_number: validator::ViewNumber) -> validator::PublicKey {

node/actors/bft/src/leader/state_machine.rs

@@ -173,6 +173,14 @@ impl StateMachine {
                     .payload_manager
                     .propose(ctx, highest_qc.header().number.next())
                     .await?;
+                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());

node/actors/bft/src/replica/block.rs

@@ -1,5 +1,5 @@
 use super::StateMachine;
-use tracing::{info, instrument};
+use tracing::instrument;
 use zksync_concurrency::ctx;
 use zksync_consensus_roles::validator;
 
@@ -30,9 +30,10 @@ impl StateMachine {
             justification: commit_qc.clone(),
         };
 
-        info!(
-            "Finalized a block!\nFinal block: {:#?}",
-            block.header().hash()
+        tracing::info!(
+            "Finalized block {}: {:#?}",
+            block.header().number,
+            block.header().hash(),
         );
         self.config
             .block_store

node/actors/bft/src/replica/leader_prepare.rs

@@ -1,5 +1,4 @@
 use super::StateMachine;
-use crate::Config;
 use std::collections::HashMap;
 use tracing::instrument;
 use zksync_concurrency::{ctx, error::Wrap};
@@ -229,7 +228,7 @@ impl StateMachine {
             // The leader proposed a new block.
             Some(payload) => {
                 // Check that the payload doesn't exceed the maximum size.
-                if payload.0.len() > Config::PAYLOAD_MAX_SIZE {
+                if payload.0.len() > self.config.max_payload_size {
                     return Err(Error::ProposalOversizedPayload {
                         payload_size: payload.0.len(),
                         header: message.proposal,

node/actors/bft/src/replica/tests.rs

@@ -1,5 +1,8 @@
 use super::{leader_commit, leader_prepare};
-use crate::{testonly, testonly::ut_harness::UTHarness, Config};
+use crate::{
+    testonly,
+    testonly::ut_harness::{UTHarness, MAX_PAYLOAD_SIZE},
+};
 use assert_matches::assert_matches;
 use rand::Rng;
 use zksync_concurrency::{ctx, scope};
@@ -271,7 +274,7 @@ async fn leader_prepare_proposal_oversized_payload() {
         let (mut util, runner) = UTHarness::new(ctx, 1).await;
         s.spawn_bg(runner.run(ctx));
 
-        let payload_oversize = Config::PAYLOAD_MAX_SIZE + 1;
+        let payload_oversize = MAX_PAYLOAD_SIZE + 1;
         let payload_vec = vec![0; payload_oversize];
         let mut leader_prepare = util.new_leader_prepare(ctx).await.msg;
         leader_prepare.proposal_payload = Some(Payload(payload_vec));

node/actors/bft/src/testonly/fuzz.rs

@@ -1,3 +1,4 @@
+use crate::testonly::node::MAX_PAYLOAD_SIZE;
 use rand::{seq::SliceRandom, Rng};
 use zksync_consensus_roles::validator;
 
@@ -155,10 +156,10 @@ impl Fuzz for validator::Signers {
 impl Fuzz for validator::Payload {
     fn mutate(&mut self, rng: &mut impl Rng) {
         // Push bytes into the payload until it exceeds the limit.
-        let num_bytes = crate::Config::PAYLOAD_MAX_SIZE + 1 - self.0.len();
+        let num_bytes = MAX_PAYLOAD_SIZE - self.0.len() + 1;
         let bytes: Vec<u8> = (0..num_bytes).map(|_| rng.gen()).collect();
         self.0.extend_from_slice(&bytes);
-        assert!(self.0.len() > crate::Config::PAYLOAD_MAX_SIZE);
+        assert!(self.0.len() > MAX_PAYLOAD_SIZE);
     }
 }
 

node/actors/bft/src/testonly/make.rs

@@ -1,12 +1,12 @@
 //! This module contains utilities that are only meant for testing purposes.
-use crate::{Config, PayloadManager};
+use crate::PayloadManager;
 use rand::Rng as _;
 use zksync_concurrency::ctx;
 use zksync_consensus_roles::validator;
 
-/// Produces random payload.
+/// Produces random payload of a given size.
 #[derive(Debug)]
-pub struct RandomPayload;
+pub struct RandomPayload(pub usize);
 
 #[async_trait::async_trait]
 impl PayloadManager for RandomPayload {
@@ -15,7 +15,7 @@ impl PayloadManager for RandomPayload {
         ctx: &ctx::Ctx,
         _number: validator::BlockNumber,
     ) -> ctx::Result<validator::Payload> {
-        let mut payload = validator::Payload(vec![0; Config::PAYLOAD_MAX_SIZE]);
+        let mut payload = validator::Payload(vec![0; self.0]);
         ctx.rng().fill(&mut payload.0[..]);
         Ok(payload)
     }
@@ -77,29 +77,3 @@ impl PayloadManager for RejectPayload {
         Err(anyhow::anyhow!("invalid payload").into())
     }
 }
-
-/// Creates a genesis block with the given payload
-/// and a validator set for the chain.
-pub fn make_genesis(
-    keys: &[validator::SecretKey],
-    payload: validator::Payload,
-    block_number: validator::BlockNumber,
-) -> (validator::FinalBlock, validator::ValidatorSet) {
-    let header = validator::BlockHeader::genesis(payload.hash(), block_number);
-    let validator_set = validator::ValidatorSet::new(keys.iter().map(|k| k.public())).unwrap();
-    let signed_messages: Vec<_> = keys
-        .iter()
-        .map(|sk| {
-            sk.sign_msg(validator::ReplicaCommit {
-                protocol_version: validator::ProtocolVersion::EARLIEST,
-                view: validator::ViewNumber(0),
-                proposal: header,
-            })
-        })
-        .collect();
-    let final_block = validator::FinalBlock {
-        payload,
-        justification: validator::CommitQC::from(&signed_messages, &validator_set).unwrap(),
-    };
-    (final_block, validator_set)
-}

node/actors/bft/src/testonly/node.rs

@@ -1,12 +1,16 @@
 use super::Fuzz;
 use crate::{io, testonly, PayloadManager};
+use anyhow::Context as _;
 use rand::Rng;
 use std::sync::Arc;
 use zksync_concurrency::{ctx, scope};
 use zksync_consensus_network as network;
 use zksync_consensus_network::io::ConsensusInputMessage;
 use zksync_consensus_storage as storage;
-use zksync_consensus_utils::pipe::DispatcherPipe;
+use zksync_consensus_storage::testonly::in_memory;
+use zksync_consensus_utils::pipe;
+
+pub(crate) const MAX_PAYLOAD_SIZE: usize = 1000;
 
 /// Enum representing the behavior of the node.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -29,32 +33,50 @@ impl Behavior {
     pub(crate) fn payload_manager(&self) -> Box<dyn PayloadManager> {
         match self {
             Self::HonestNotProposing => Box::new(testonly::PendingPayload),
-            _ => Box::new(testonly::RandomPayload),
+            _ => Box::new(testonly::RandomPayload(MAX_PAYLOAD_SIZE)),
         }
     }
 }
 
 /// Struct representing a node.
 pub(super) struct Node {
-    pub(crate) net: network::testonly::Instance,
+    pub(crate) net: network::Config,
     pub(crate) behavior: Behavior,
     pub(crate) block_store: Arc<storage::BlockStore>,
 }
 
 impl Node {
     /// Runs a mock executor.
-    pub(crate) async fn run_executor(
+    pub(crate) async fn run(
         &self,
         ctx: &ctx::Ctx,
-        consensus_pipe: DispatcherPipe<io::InputMessage, io::OutputMessage>,
-        network_pipe: DispatcherPipe<network::io::InputMessage, network::io::OutputMessage>,
+        network_pipe: &mut pipe::DispatcherPipe<
+            network::io::InputMessage,
+            network::io::OutputMessage,
+        >,
     ) -> anyhow::Result<()> {
         let rng = &mut ctx.rng();
-        let mut net_recv = network_pipe.recv;
-        let net_send = network_pipe.send;
+        let net_recv = &mut network_pipe.recv;
+        let net_send = &mut network_pipe.send;
+        let (consensus_actor_pipe, consensus_pipe) = pipe::new();
         let mut con_recv = consensus_pipe.recv;
         let con_send = consensus_pipe.send;
         scope::run!(ctx, |ctx, s| async {
+            s.spawn(async {
+                let validator_key = self.net.consensus.as_ref().unwrap().key.clone();
+                let validator_set = self.net.validators.clone();
+                crate::Config {
+                    secret_key: validator_key.clone(),
+                    validator_set,
+                    block_store: self.block_store.clone(),
+                    replica_store: Box::new(in_memory::ReplicaStore::default()),
+                    payload_manager: self.behavior.payload_manager(),
+                    max_payload_size: MAX_PAYLOAD_SIZE,
+                }
+                .run(ctx, consensus_actor_pipe)
+                .await
+                .context("consensus.run()")
+            });
             s.spawn(async {
                 while let Ok(network_message) = net_recv.recv(ctx).await {
                     match network_message {