Broadcast validators signature and collect QC (BFT-414) (#76)

## What ❔

Implement a mechanism for validators to broadcast their signatures to
other nodes in the network and collect them in a new certificate.

## Why ❔

This is essential for signing L1 batches and sending them to L1 for
verification. Validators need to broadcast their signatures and gather
them in a certificate in case the majority signs the new batch.

---------

Co-authored-by: Grzegorz Prusak <[email protected]>
Co-authored-by: Ignacio Avecilla <[email protected]>
Co-authored-by: Bruno França <[email protected]>

.github/workflows/protobuf.yaml

@@ -2,7 +2,7 @@ name: protobuf_compatibility
 
 on:
   pull_request:
-    branches: [ "*" ]
+    branches: ["*"]
   push:
     # protobuf compatibility is a transitive property,
     # but it requires all the transitions to be checked.
@@ -11,7 +11,7 @@ on:
     # (unless we improve our github setup).
     # Therefore on post-merge we will execute the
     # compatibility check as well (TODO: alerting).
-    branches: [ "main" ]
+    branches: ["main"]
 
 permissions:
   id-token: write
@@ -33,17 +33,16 @@ jobs:
   compatibility:
     runs-on: [ubuntu-22.04-github-hosted-16core]
     steps:
-      - uses: mozilla-actions/[email protected]     
-      
+      - uses: mozilla-actions/[email protected]
+
       # before
       - uses: actions/checkout@v4
         with:
           ref: ${{ env.BASE }}
           path: before
           fetch-depth: 0 # fetches all branches and tags, which is needed to compute the LCA.
       - name: checkout LCA
-        run:
-          git checkout $(git merge-base $BASE $HEAD)
+        run: git checkout $(git merge-base $BASE $HEAD)
         working-directory: ./before
       - name: compile before
         run: cargo build --all-targets
@@ -53,7 +52,7 @@ jobs:
           perl -ne 'print "$1\n" if /PROTOBUF_DESCRIPTOR="(.*)"/'
           `find ./before/node/target/debug/build/*/output`
           | xargs cat > ./before.binpb
- 
+
       # after
       - uses: actions/checkout@v4
         with:
@@ -67,7 +66,7 @@ jobs:
           perl -ne 'print "$1\n" if /PROTOBUF_DESCRIPTOR="(.*)"/'
           `find ./after/node/target/debug/build/*/output`
           | xargs cat > ./after.binpb
-      
+
       # compare
       - uses: bufbuild/buf-setup-action@v1
         with:

node/Cargo.toml

@@ -9,7 +9,7 @@ members = [
     "libs/protobuf_build",
     "libs/roles",
     "libs/storage",
-    "libs/utils", 
+    "libs/utils",
     "tests",
     "tools",
 ]
@@ -55,7 +55,7 @@ ff_ce = "0.14.3"
 heck = "0.5.0"
 hex = "0.4.3"
 im = "15.1.0"
-jsonrpsee = { version = "0.21.0", features = ["server", "http-client"]  }
+jsonrpsee = { version = "0.21.0", features = ["server", "http-client"] }
 k8s-openapi = { version = "0.21.0", features = ["latest"] }
 kube = { version = "0.88.1", features = ["runtime", "derive"] }
 num-bigint = "0.4.4"

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

@@ -51,7 +51,7 @@ impl StateMachine {
         let author = &signed_message.key;
 
         // Check that the message signer is in the validator committee.
-        if !self.config.genesis().committee.contains(author) {
+        if !self.config.genesis().validators.contains(author) {
             return Err(Error::NonValidatorSigner {
                 signer: author.clone(),
             });
@@ -103,7 +103,7 @@ impl StateMachine {
             .expect("Could not add message to CommitQC");
 
         // Calculate the CommitQC signers weight.
-        let weight = self.config.genesis().committee.weight(&commit_qc.signers);
+        let weight = self.config.genesis().validators.weight(&commit_qc.signers);
 
         // Update commit message current view number for author
         self.replica_commit_views
@@ -118,7 +118,7 @@ impl StateMachine {
             .retain(|view_number, _| active_views.contains(view_number));
 
         // Now we check if we have enough weight to continue.
-        if weight < self.config.genesis().committee.threshold() {
+        if weight < self.config.genesis().validators.threshold() {
             return Ok(());
         };
 

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

@@ -63,7 +63,7 @@ impl StateMachine {
         let author = &signed_message.key;
 
         // Check that the message signer is in the validator set.
-        if !self.config.genesis().committee.contains(author) {
+        if !self.config.genesis().validators.contains(author) {
             return Err(Error::NonValidatorSigner {
                 signer: author.clone(),
             });
@@ -114,7 +114,7 @@ impl StateMachine {
             .expect("Could not add message to PrepareQC");
 
         // Calculate the PrepareQC signers weight.
-        let weight = prepare_qc.weight(&self.config.genesis().committee);
+        let weight = prepare_qc.weight(&self.config.genesis().validators);
 
         // Update prepare message current view number for author
         self.replica_prepare_views
@@ -129,7 +129,7 @@ impl StateMachine {
             .retain(|view_number, _| active_views.contains(view_number));
 
         // Now we check if we have enough weight to continue.
-        if weight < self.config.genesis().committee.threshold() {
+        if weight < self.config.genesis().validators.threshold() {
             return Ok(());
         }
 

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

@@ -374,7 +374,7 @@ async fn replica_prepare_different_messages() {
 
         let mut replica_commit_result = None;
         // The rest of the validators until threshold sign other_replica_prepare
-        for i in validators / 2..util.genesis().committee.threshold() as usize {
+        for i in validators / 2..util.genesis().validators.threshold() as usize {
             replica_commit_result = util
                 .process_replica_prepare(ctx, util.keys[i].sign_msg(other_replica_prepare.clone()))
                 .await

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

@@ -61,7 +61,7 @@ impl UTHarness {
         let (send, recv) = ctx::channel::unbounded();
 
         let cfg = Arc::new(Config {
-            secret_key: setup.keys[0].clone(),
+            secret_key: setup.validator_keys[0].clone(),
             block_store: block_store.clone(),
             replica_store: Box::new(in_memory::ReplicaStore::default()),
             payload_manager,
@@ -75,7 +75,7 @@ impl UTHarness {
             leader,
             replica,
             pipe: recv,
-            keys: setup.keys.clone(),
+            keys: setup.validator_keys.clone(),
             leader_send,
         };
         let _: Signed<ReplicaPrepare> = this.try_recv().unwrap();
@@ -86,7 +86,7 @@ impl UTHarness {
     pub(crate) async fn new_many(ctx: &ctx::Ctx) -> (UTHarness, BlockStoreRunner) {
         let num_validators = 6;
         let (util, runner) = UTHarness::new(ctx, num_validators).await;
-        assert!(util.genesis().committee.max_faulty_weight() > 0);
+        assert!(util.genesis().validators.max_faulty_weight() > 0);
         (util, runner)
     }
 
@@ -223,8 +223,8 @@ impl UTHarness {
         for (i, msg) in msgs.into_iter().enumerate() {
             let res = self.process_replica_prepare(ctx, msg).await;
             match (
-                (i + 1) as u64 * self.genesis().committee.iter().next().unwrap().weight
-                    < self.genesis().committee.threshold(),
+                (i + 1) as u64 * self.genesis().validators.iter().next().unwrap().weight
+                    < self.genesis().validators.threshold(),
                 first_match,
             ) {
                 (true, _) => assert!(res.unwrap().is_none()),
@@ -258,8 +258,8 @@ impl UTHarness {
                 .leader
                 .process_replica_commit(ctx, key.sign_msg(msg.clone()));
             match (
-                (i + 1) as u64 * self.genesis().committee.iter().next().unwrap().weight
-                    < self.genesis().committee.threshold(),
+                (i + 1) as u64 * self.genesis().validators.iter().next().unwrap().weight
+                    < self.genesis().validators.threshold(),
                 first_match,
             ) {
                 (true, _) => res.unwrap(),

node/actors/executor/src/lib.rs

@@ -3,7 +3,6 @@ use crate::io::Dispatcher;
 use anyhow::Context as _;
 use std::{
     collections::{HashMap, HashSet},
-    fmt,
     sync::Arc,
 };
 use zksync_concurrency::{ctx, limiter, net, scope, time};
@@ -19,6 +18,7 @@ mod io;
 mod tests;
 
 /// Validator-related part of [`Executor`].
+#[derive(Debug)]
 pub struct Validator {
     /// Consensus network configuration.
     pub key: validator::SecretKey,
@@ -28,14 +28,6 @@ pub struct Validator {
     pub payload_manager: Box<dyn bft::PayloadManager>,
 }
 
-impl fmt::Debug for Validator {
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt.debug_struct("ValidatorExecutor")
-            .field("key", &self.key)
-            .finish()
-    }
-}
-
 /// Config of the node executor.
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -47,7 +39,6 @@ pub struct Config {
     pub public_addr: net::Host,
     /// Maximal size of the block payload.
     pub max_payload_size: usize,
-
     /// Key of this node. It uniquely identifies the node.
     /// It should match the secret key provided in the `node_key` file.
     pub node_key: node::SecretKey,
@@ -129,7 +120,7 @@ impl Executor {
             if !self
                 .block_store
                 .genesis()
-                .committee
+                .validators
                 .contains(&validator.key.public())
             {
                 tracing::warn!(

node/actors/network/src/config.rs

@@ -20,6 +20,8 @@ pub struct RpcConfig {
     pub get_block_timeout: Option<time::Duration>,
     /// Max rate of sending/receiving consensus messages.
     pub consensus_rate: limiter::Rate,
+    /// Max rate of sending/receiving l1 batch votes messages.
+    pub push_batch_votes_rate: limiter::Rate,
 }
 
 impl Default for RpcConfig {
@@ -42,6 +44,10 @@ impl Default for RpcConfig {
                 burst: 10,
                 refresh: time::Duration::ZERO,
             },
+            push_batch_votes_rate: limiter::Rate {
+                burst: 2,
+                refresh: time::Duration::milliseconds(500),
+            },
         }
     }
 }

node/actors/network/src/consensus/mod.rs

@@ -151,7 +151,7 @@ impl Network {
     /// Constructs a new consensus network state.
     pub(crate) fn new(gossip: Arc<gossip::Network>) -> Option<Arc<Self>> {
         let key = gossip.cfg.validator_key.clone()?;
-        let validators: HashSet<_> = gossip.genesis().committee.keys().cloned().collect();
+        let validators: HashSet<_> = gossip.genesis().validators.keys().cloned().collect();
         Some(Arc::new(Self {
             key,
             inbound: PoolWatch::new(validators.clone(), 0),

node/actors/network/src/consensus/tests.rs

@@ -166,12 +166,14 @@ async fn test_genesis_mismatch() {
             .gossip
             .validator_addrs
             .update(
-                &setup.genesis.committee,
-                &[Arc::new(setup.keys[1].sign_msg(validator::NetAddress {
-                    addr: *cfgs[1].server_addr,
-                    version: 0,
-                    timestamp: ctx.now_utc(),
-                }))],
+                &setup.genesis.validators,
+                &[Arc::new(setup.validator_keys[1].sign_msg(
+                    validator::NetAddress {
+                        addr: *cfgs[1].server_addr,
+                        version: 0,
+                        timestamp: ctx.now_utc(),
+                    },
+                ))],
             )
             .await
             .unwrap();
@@ -185,7 +187,7 @@ async fn test_genesis_mismatch() {
             .context("preface::accept()")?;
         assert_eq!(endpoint, preface::Endpoint::ConsensusNet);
         tracing::info!("Expect the handshake to fail");
-        let res = handshake::inbound(ctx, &setup.keys[1], rng.gen(), &mut stream).await;
+        let res = handshake::inbound(ctx, &setup.validator_keys[1], rng.gen(), &mut stream).await;
         assert_matches!(res, Err(handshake::Error::GenesisMismatch));
 
         tracing::info!("Try to connect to a node with a mismatching genesis.");
@@ -195,10 +197,10 @@ async fn test_genesis_mismatch() {
                 .context("preface::connect")?;
         let res = handshake::outbound(
             ctx,
-            &setup.keys[1],
+            &setup.validator_keys[1],
             rng.gen(),
             &mut stream,
-            &setup.keys[0].public(),
+            &setup.validator_keys[0].public(),
         )
         .await;
         tracing::info!(