Broadcast validators signature and collect QC (BFT-414)

node/actors/network/src/config.rs

@@ -18,6 +18,8 @@ pub struct RpcConfig {
     pub get_block_rate: limiter::Rate,
     /// Max rate of sending/receiving consensus messages.
     pub consensus_rate: limiter::Rate,
+    /// Max rate of sending/receiving l1 batch signature messages.
+    pub l1_batch_signature_rate: limiter::Rate,
 }
 
 impl Default for RpcConfig {
@@ -39,6 +41,10 @@ impl Default for RpcConfig {
                 burst: 10,
                 refresh: time::Duration::ZERO,
             },
+            l1_batch_signature_rate: limiter::Rate {
+                burst: 10,
+                refresh: time::Duration::ZERO,
+            },
         }
     }
 }

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

@@ -1,13 +1,18 @@
 //! Consensus network is a full graph of connections between all validators.
 //! BFT consensus messages are exchanged over this network.
-use crate::{config, gossip, io, noise, pool::PoolWatch, preface, rpc};
+use crate::{
+    config, gossip, io, noise,
+    pool::PoolWatch,
+    preface,
+    rpc::{self, signature::L1BatchSignatureServer},
+};
 use anyhow::Context as _;
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
 };
 use zksync_concurrency::{ctx, oneshot, scope, sync, time};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::validator::{self, L1BatchQC};
 use zksync_protobuf::kB;
 
 mod handshake;
@@ -33,6 +38,9 @@ pub(crate) struct Network {
     pub(crate) outbound: PoolWatch<validator::PublicKey>,
     /// RPC clients for all validators.
     pub(crate) clients: HashMap<validator::PublicKey, rpc::Client<rpc::consensus::Rpc>>,
+    /// RPC clients for l1 batch signature.
+    pub(crate) signature_clients: HashMap<validator::PublicKey, rpc::Client<rpc::signature::Rpc>>,
+    pub(crate) l1_batch_qc: L1BatchQC,
 }
 
 #[async_trait::async_trait]
@@ -59,13 +67,27 @@ impl rpc::Handler<rpc::consensus::Rpc> for &Network {
     }
 }
 
+#[async_trait::async_trait]
+impl rpc::Handler<rpc::signature::Rpc> for &L1BatchSignatureServer<'_> {
+    /// Here we bound the buffering of incoming consensus messages.
+    fn max_req_size(&self) -> usize {
+        self.0.gossip.cfg.max_block_size.saturating_add(kB)
+    }
+
+    async fn handle(&self, _ctx: &ctx::Ctx, req: rpc::signature::Req) -> anyhow::Result<()> {
+        self.0.l1_batch_qc.clone().add(req.0.sig);
+        Ok(())
+    }
+}
+
 impl Network {
     /// Constructs a new consensus network state.
     pub(crate) fn new(ctx: &ctx::Ctx, gossip: Arc<gossip::Network>) -> Option<Arc<Self>> {
         let key = gossip.cfg.validator_key.clone()?;
         let validators: HashSet<_> = gossip.genesis().validators.iter().cloned().collect();
         Some(Arc::new(Self {
             key,
+            l1_batch_qc: L1BatchQC::default(),
             inbound: PoolWatch::new(validators.clone(), 0),
             outbound: PoolWatch::new(validators.clone(), 0),
             clients: validators
@@ -77,12 +99,21 @@ impl Network {
                     )
                 })
                 .collect(),
+            signature_clients: validators
+                .iter()
+                .map(|peer| {
+                    (
+                        peer.clone(),
+                        rpc::Client::new(ctx, gossip.cfg.rpc.l1_batch_signature_rate),
+                    )
+                })
+                .collect(),
             gossip,
         }))
     }
 
     /// Sends a message to all validators.
-    pub(crate) async fn broadcast(
+    pub(crate) async fn broadcast_consensus_msg(
         &self,
         ctx: &ctx::Ctx,
         msg: validator::Signed<validator::ConsensusMsg>,
@@ -102,6 +133,27 @@ impl Network {
         .await
     }
 
+    /// Broadcasts a signature to all validators.
+    pub(crate) async fn broadcast_signature(
+        &self,
+        ctx: &ctx::Ctx,
+        signature: validator::Signed<validator::L1BatchSignatureMsg>,
+    ) -> anyhow::Result<()> {
+        let req = rpc::signature::Req(signature);
+        scope::run!(ctx, |ctx, s| async {
+            for (peer, client) in &self.signature_clients {
+                s.spawn(async {
+                    if let Err(err) = client.call(ctx, &req, RESP_MAX_SIZE).await {
+                        tracing::info!("send({:?},<L1BatchSignatureMsg>): {err:#}", &*peer);
+                    }
+                    Ok(())
+                });
+            }
+            Ok(())
+        })
+        .await
+    }
+
     /// Sends a message to the given validator.
     pub(crate) async fn send(
         &self,

node/actors/network/src/io.rs

@@ -10,6 +10,13 @@ pub enum InputMessage {
     Consensus(ConsensusInputMessage),
     /// Message types from the Sync Blocks actor.
     SyncBlocks(SyncBlocksInputMessage),
+    /// Message of type L1BatchSignatureMsg from the validator.
+    L1BatchSignature(L1BatchSignatureInputMessage),
+}
+
+#[derive(Debug)]
+pub struct L1BatchSignatureInputMessage {
+    pub message: validator::Signed<validator::L1BatchSignatureMsg>,
 }
 
 /// Message types from the Consensus actor.

node/actors/network/src/lib.rs

@@ -87,7 +87,11 @@ impl Network {
                     io::Target::Validator(key) => {
                         consensus.send(ctx, &key, message.message).await?
                     }
-                    io::Target::Broadcast => consensus.broadcast(ctx, message.message).await?,
+                    io::Target::Broadcast => {
+                        consensus
+                            .broadcast_consensus_msg(ctx, message.message)
+                            .await?
+                    }
                 }
             }
             io::InputMessage::SyncBlocks(io::SyncBlocksInputMessage::GetBlock {
@@ -102,6 +106,11 @@ impl Network {
                     Err(err) => Err(io::GetBlockError::Internal(err)),
                 });
             }
+            io::InputMessage::L1BatchSignature(message) => {
+                let consensus = self.consensus.as_ref().context("not a validator node")?;
+                let ctx = &ctx.with_timeout(CONSENSUS_MSG_TIMEOUT);
+                consensus.broadcast_signature(ctx, message.message).await?;
+            }
         }
         Ok(())
     }

node/actors/network/src/proto/signature.proto

@@ -0,0 +1,12 @@
+syntax = "proto3";
+
+package zksync.network.consensus;
+
+import "zksync/roles/validator.proto";
+import "zksync/std.proto";
+
+message SignatureReq {
+  optional roles.validator.Signed msg = 1;
+}
+
+message SignatureResp {}

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

@@ -27,6 +27,7 @@ mod metrics;
 pub(crate) mod ping;
 pub(crate) mod push_block_store_state;
 pub(crate) mod push_validator_addrs;
+pub(crate) mod signature;
 #[cfg(test)]
 pub(crate) mod testonly;
 #[cfg(test)]

node/actors/network/src/rpc/signature.rs

@@ -0,0 +1,49 @@
+//! Defines RPC for passing consensus messages.
+use crate::{consensus::Network, mux, proto::consensus as proto};
+use zksync_consensus_roles::validator;
+use zksync_protobuf::{read_required, ProtoFmt};
+
+/// Signature RPC.
+pub(crate) struct Rpc;
+
+impl super::Rpc for Rpc {
+    const CAPABILITY_ID: mux::CapabilityId = 5;
+    const INFLIGHT: u32 = 1;
+    const METHOD: &'static str = "signature";
+    type Req = Req;
+    type Resp = ();
+}
+pub(crate) struct L1BatchSignatureServer<'a>(pub(crate) &'a Network);
+/// Signed consensus message that the receiving peer should process.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct Req(pub(crate) validator::Signed<validator::L1BatchSignatureMsg>);
+
+/// Confirmation that the signature message has been processed.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct Resp;
+
+impl ProtoFmt for Req {
+    type Proto = proto::SignatureReq;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        read_required(&r.msg).map(Self)
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            msg: Some(self.0.build()),
+        }
+    }
+}
+
+impl ProtoFmt for Resp {
+    type Proto = proto::SignatureResp;
+
+    fn read(_r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self)
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {}
+    }
+}

node/libs/roles/src/proto/validator.proto

@@ -144,11 +144,16 @@ message NetAddress {
   optional std.Timestamp timestamp = 3; // required
 }
 
+message L1BatchSignatureMsg {
+  optional bytes signature = 1; // required
+}
+
 message Msg {
   oneof t { // required
     ConsensusMsg consensus = 1;
     bytes session_id = 2;
     NetAddress net_address = 3;
+    L1BatchSignatureMsg l1_batch_signature = 4;
   }
 }
 

node/libs/roles/src/validator/conv.rs

@@ -1,8 +1,9 @@
 use super::{
     AggregateSignature, BlockHeader, BlockHeaderHash, BlockNumber, CommitQC, ConsensusMsg,
-    FinalBlock, Fork, ForkNumber, Genesis, GenesisHash, LeaderCommit, LeaderPrepare, Msg, MsgHash,
-    NetAddress, Payload, PayloadHash, Phase, PrepareQC, ProtocolVersion, PublicKey, ReplicaCommit,
-    ReplicaPrepare, Signature, Signed, Signers, ValidatorSet, View, ViewNumber,
+    FinalBlock, Fork, ForkNumber, Genesis, GenesisHash, L1BatchSignatureMsg, LeaderCommit,
+    LeaderPrepare, Msg, MsgHash, NetAddress, Payload, PayloadHash, Phase, PrepareQC,
+    ProtocolVersion, PublicKey, ReplicaCommit, ReplicaPrepare, Signature, Signed, Signers,
+    ValidatorSet, View, ViewNumber,
 };
 use crate::{node::SessionId, proto::validator as proto};
 use anyhow::Context as _;
@@ -355,6 +356,20 @@ impl ProtoFmt for NetAddress {
     }
 }
 
+impl ProtoFmt for L1BatchSignatureMsg {
+    type Proto = proto::L1BatchSignatureMsg;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self(ByteFmt::decode(required(&r.signature)?)?))
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            signature: Some(self.0.encode()),
+        }
+    }
+}
+
 impl ProtoFmt for Msg {
     type Proto = proto::Msg;
 
@@ -364,6 +379,9 @@ impl ProtoFmt for Msg {
             T::Consensus(r) => Self::Consensus(ProtoFmt::read(r).context("Consensus")?),
             T::SessionId(r) => Self::SessionId(SessionId(r.clone())),
             T::NetAddress(r) => Self::NetAddress(ProtoFmt::read(r).context("NetAddress")?),
+            T::L1BatchSignature(r) => {
+                Self::L1BatchSignature(ProtoFmt::read(r).context("L1BatchSignature")?)
+            }
         })
     }
 
@@ -374,6 +392,7 @@ impl ProtoFmt for Msg {
             Self::Consensus(x) => T::Consensus(x.build()),
             Self::SessionId(x) => T::SessionId(x.0.clone()),
             Self::NetAddress(x) => T::NetAddress(x.build()),
+            Self::L1BatchSignature(x) => T::L1BatchSignature(x.build()),
         };
 
         Self::Proto { t: Some(t) }

node/libs/roles/src/validator/messages/l1_batch.rs

@@ -0,0 +1,20 @@
+use crate::validator::{self, Signature};
+
+/// A message to send by validators to the gossip network.
+/// It contains the validators signature to sign the block batches to be sent to L1.
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct L1BatchSignatureMsg(pub Signature);
+
+/// A certificate for a batch of L2 blocks to be sent to L1.
+/// It contains the signatures of the validators that signed the batch.
+#[derive(Debug, Clone, Default, Eq, PartialEq)]
+pub struct L1BatchQC {
+    /// The aggregate signature of the signed L1 batches.
+    pub signatures: validator::AggregateSignature,
+}
+
+impl L1BatchQC {
+    pub fn add(&mut self, signature: Signature) {
+        self.signatures.add(&signature);
+    }
+}

node/libs/roles/src/validator/messages/mod.rs

@@ -3,6 +3,7 @@
 mod block;
 mod consensus;
 mod discovery;
+mod l1_batch;
 mod leader_commit;
 mod leader_prepare;
 mod msg;
@@ -12,6 +13,7 @@ mod replica_prepare;
 pub use block::*;
 pub use consensus::*;
 pub use discovery::*;
+pub use l1_batch::*;
 pub use leader_commit::*;
 pub use leader_prepare::*;
 pub use msg::*;

node/libs/roles/src/validator/messages/msg.rs

@@ -1,6 +1,9 @@
 //! Generic message types.
-use super::{ConsensusMsg, NetAddress};
-use crate::{node::SessionId, validator, validator::Error};
+use super::{ConsensusMsg, L1BatchSignatureMsg, NetAddress};
+use crate::{
+    node::SessionId,
+    validator::{self, Error},
+};
 use std::fmt;
 use zksync_consensus_crypto::{keccak256, ByteFmt, Text, TextFmt};
 use zksync_consensus_utils::enum_util::{BadVariantError, Variant};
@@ -14,6 +17,8 @@ pub enum Msg {
     SessionId(SessionId),
     /// validator discovery
     NetAddress(NetAddress),
+    /// l1 batch signature
+    L1BatchSignature(L1BatchSignatureMsg),
 }
 
 impl Msg {
@@ -59,6 +64,18 @@ impl Variant<Msg> for NetAddress {
     }
 }
 
+impl Variant<Msg> for L1BatchSignatureMsg {
+    fn insert(self) -> Msg {
+        Msg::L1BatchSignature(self)
+    }
+    fn extract(msg: Msg) -> Result<Self, BadVariantError> {
+        let Msg::L1BatchSignature(this) = msg else {
+            return Err(BadVariantError);
+        };
+        Ok(this)
+    }
+}
+
 /// Hash of a message.
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub struct MsgHash(pub(crate) keccak256::Keccak256);