feat: L1 batch signing (BFT-474) (#2414)

## What ❔

Implements the methods on `PersistentBatchStore` required by consensus
to sign and publish attestations over L1 batches.
Calculates the hash value which needs to be signed by attestors using
`SignedBatchInfo::hash()`.
## Why ❔

This allows consensus to figure out which is the first batch that needs
to be signed after the node starts, and subsequently to retrieve the
payload to be signed as well.

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [x] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [x] Tests for the changes have been added / updated.
- [x] Documentation comments have been added / updated.
- [x] Code has been formatted via `zk fmt` and `zk lint`.

---------

Co-authored-by: Bruno França <[email protected]>

Cargo.toml

@@ -207,16 +207,16 @@ zk_evm_1_4_1 = { package = "zk_evm", version = "0.141.0" }
 zk_evm_1_5_0 = { package = "zk_evm", version = "0.150.0" }
 
 # Consensus dependencies.
-zksync_concurrency = "=0.1.0-rc.1"
-zksync_consensus_bft = "=0.1.0-rc.1"
-zksync_consensus_crypto = "=0.1.0-rc.1"
-zksync_consensus_executor = "=0.1.0-rc.1"
-zksync_consensus_network = "=0.1.0-rc.1"
-zksync_consensus_roles = "=0.1.0-rc.1"
-zksync_consensus_storage = "=0.1.0-rc.1"
-zksync_consensus_utils = "=0.1.0-rc.1"
-zksync_protobuf = "=0.1.0-rc.1"
-zksync_protobuf_build = "=0.1.0-rc.1"
+zksync_concurrency = "=0.1.0-rc.2"
+zksync_consensus_bft = "=0.1.0-rc.2"
+zksync_consensus_crypto = "=0.1.0-rc.2"
+zksync_consensus_executor = "=0.1.0-rc.2"
+zksync_consensus_network = "=0.1.0-rc.2"
+zksync_consensus_roles = "=0.1.0-rc.2"
+zksync_consensus_storage = "=0.1.0-rc.2"
+zksync_consensus_utils = "=0.1.0-rc.2"
+zksync_protobuf = "=0.1.0-rc.2"
+zksync_protobuf_build = "=0.1.0-rc.2"
 
 # "Local" dependencies
 zksync_multivm = { path = "core/lib/multivm" }

core/lib/config/src/configs/consensus.rs

@@ -120,6 +120,13 @@ pub struct ConsensusConfig {
     /// Maximal allowed size of the payload in bytes.
     pub max_payload_size: usize,
 
+    /// Maximal allowed size of the sync-batch payloads in bytes.
+    ///
+    /// The batch consists of block payloads and a Merkle proof of inclusion on L1 (~1kB),
+    /// so the maximum batch size should be the maximum payload size times the maximum number
+    /// of blocks in a batch.
+    pub max_batch_size: usize,
+
     /// Limit on the number of inbound connections outside
     /// of the `static_inbound` set.
     pub gossip_dynamic_inbound_limit: usize,

core/lib/config/src/testonly.rs

@@ -755,6 +755,7 @@ impl Distribution<configs::consensus::ConsensusConfig> for EncodeDist {
             server_addr: self.sample(rng),
             public_addr: Host(self.sample(rng)),
             max_payload_size: self.sample(rng),
+            max_batch_size: self.sample(rng),
             gossip_dynamic_inbound_limit: self.sample(rng),
             gossip_static_inbound: self
                 .sample_range(rng)

core/lib/dal/src/consensus_dal.rs

@@ -7,7 +7,7 @@ use zksync_db_connection::{
     error::{DalError, DalResult, SqlxContext},
     instrument::{InstrumentExt, Instrumented},
 };
-use zksync_types::{L1BatchNumber, L2BlockNumber};
+use zksync_types::L2BlockNumber;
 
 pub use crate::consensus::Payload;
 use crate::{Core, CoreDal};
@@ -409,29 +409,12 @@ impl ConsensusDal<'_, '_> {
     ///
     /// Insertion is allowed even if it creates gaps in the L1 batch history.
     ///
-    /// It fails if the batch payload is missing or it's not consistent with the QC.
+    /// This method assumes that all payload validation has been carried out by the caller.
     pub async fn insert_batch_certificate(
         &mut self,
         cert: &attester::BatchQC,
     ) -> Result<(), InsertCertificateError> {
-        use InsertCertificateError as E;
-        let mut txn = self.storage.start_transaction().await?;
-
-        let l1_batch_number = L1BatchNumber(cert.message.number.0 as u32);
-        let _l1_batch_header = txn
-            .blocks_dal()
-            .get_l1_batch_header(l1_batch_number)
-            .await?
-            .ok_or(E::MissingPayload)?;
-
-        // TODO: Verify that the certificate matches the stored batch:
-        // * add the hash of the batch to the `BatchQC`
-        // * find out which field in the `l1_batches` table contains the hash we need to match
-        // * ideally move the responsibility of validation outside this method
-
-        // if header.payload != want_payload.encode().hash() {
-        //     return Err(E::PayloadMismatch);
-        // }
+        let l1_batch_number = cert.message.number.0 as i64;
 
         let res = sqlx::query!(
             r#"
@@ -441,20 +424,18 @@ impl ConsensusDal<'_, '_> {
                 ($1, $2, NOW(), NOW())
             ON CONFLICT (l1_batch_number) DO NOTHING
             "#,
-            i64::from(l1_batch_number.0),
+            l1_batch_number,
             zksync_protobuf::serde::serialize(cert, serde_json::value::Serializer).unwrap(),
         )
         .instrument("insert_batch_certificate")
         .report_latency()
-        .execute(&mut txn)
+        .execute(self.storage)
         .await?;
 
         if res.rows_affected().is_zero() {
-            tracing::debug!(%l1_batch_number, "duplicate batch certificate");
+            tracing::debug!(l1_batch_number, "duplicate batch certificate");
         }
 
-        txn.commit().await.context("commit")?;
-
         Ok(())
     }
 
@@ -551,7 +532,8 @@ mod tests {
         // Insert some mock L2 blocks and L1 batches
         let mut block_number = 0;
         let mut batch_number = 0;
-        for _ in 0..3 {
+        let num_batches = 3;
+        for _ in 0..num_batches {
             for _ in 0..3 {
                 block_number += 1;
                 let l2_block = create_l2_block_header(block_number);
@@ -612,5 +594,11 @@ mod tests {
             .insert_batch_certificate(&cert3)
             .await
             .expect_err("missing payload");
+
+        // Insert one more L1 batch without a certificate.
+        conn.blocks_dal()
+            .insert_mock_l1_batch(&create_l1_batch_header(batch_number + 1))
+            .await
+            .unwrap();
     }
 }

core/lib/protobuf_config/src/consensus.rs

@@ -4,7 +4,7 @@ use zksync_config::configs::consensus::{
     AttesterPublicKey, ConsensusConfig, GenesisSpec, Host, NodePublicKey, ProtocolVersion,
     RpcConfig, ValidatorPublicKey, WeightedAttester, WeightedValidator,
 };
-use zksync_protobuf::{read_optional, repr::ProtoRepr, required, ProtoFmt};
+use zksync_protobuf::{kB, read_optional, repr::ProtoRepr, required, ProtoFmt};
 
 use crate::{proto::consensus as proto, read_optional_repr};
 
@@ -100,14 +100,31 @@ impl ProtoRepr for proto::Config {
             let addr = Host(required(&e.addr).context("addr")?.clone());
             anyhow::Ok((key, addr))
         };
+
+        let max_payload_size = required(&self.max_payload_size)
+            .and_then(|x| Ok((*x).try_into()?))
+            .context("max_payload_size")?;
+
+        let max_batch_size = match self.max_batch_size {
+            Some(x) => x.try_into().context("max_batch_size")?,
+            None => {
+                // Compute a default batch size, so operators are not caught out by the missing setting
+                // while we're still working on batch syncing. The batch interval is ~1 minute,
+                // so there will be ~60 blocks, and an Ethereum Merkle proof is ~1kB, but under high
+                // traffic there can be thousands of huge transactions that quickly fill up blocks
+                // and there could be more blocks in a batch then expected. We chose a generous
+                // limit so as not to prevent any legitimate batch from being transmitted.
+                max_payload_size * 5000 + kB
+            }
+        };
+
         Ok(Self::Type {
             server_addr: required(&self.server_addr)
                 .and_then(|x| Ok(x.parse()?))
                 .context("server_addr")?,
             public_addr: Host(required(&self.public_addr).context("public_addr")?.clone()),
-            max_payload_size: required(&self.max_payload_size)
-                .and_then(|x| Ok((*x).try_into()?))
-                .context("max_payload_size")?,
+            max_payload_size,
+            max_batch_size,
             gossip_dynamic_inbound_limit: required(&self.gossip_dynamic_inbound_limit)
                 .and_then(|x| Ok((*x).try_into()?))
                 .context("gossip_dynamic_inbound_limit")?,
@@ -132,6 +149,7 @@ impl ProtoRepr for proto::Config {
             server_addr: Some(this.server_addr.to_string()),
             public_addr: Some(this.public_addr.0.clone()),
             max_payload_size: Some(this.max_payload_size.try_into().unwrap()),
+            max_batch_size: Some(this.max_batch_size.try_into().unwrap()),
             gossip_dynamic_inbound_limit: Some(
                 this.gossip_dynamic_inbound_limit.try_into().unwrap(),
             ),

core/lib/protobuf_config/src/proto/core/consensus.proto

@@ -78,6 +78,9 @@ message Config {
   // Maximal allowed size of the payload.
   optional uint64 max_payload_size = 4; // required; bytes
 
+  // Maximal allowed size of the sync batches.
+  optional uint64 max_batch_size = 10; // required; bytes
+
   // Inbound connections that should be unconditionally accepted on the gossip network.
   repeated string gossip_static_inbound = 5; // required; NodePublicKey
 

core/node/consensus/src/config.rs

@@ -126,6 +126,7 @@ pub(super) fn executor(
         server_addr: cfg.server_addr,
         public_addr: net::Host(cfg.public_addr.0.clone()),
         max_payload_size: cfg.max_payload_size,
+        max_batch_size: cfg.max_batch_size,
         node_key: node_key(secrets)
             .context("node_key")?
             .context("missing node_key")?,

core/node/consensus/src/storage/connection.rs

@@ -3,6 +3,7 @@ use zksync_concurrency::{ctx, error::Wrap as _, time};
 use zksync_consensus_roles::{attester, validator};
 use zksync_consensus_storage::{self as storage, BatchStoreState};
 use zksync_dal::{consensus_dal::Payload, Core, CoreDal, DalError};
+use zksync_l1_contract_interface::i_executor::structures::StoredBatchInfo;
 use zksync_node_sync::{fetcher::IoCursorExt as _, ActionQueueSender, SyncState};
 use zksync_state_keeper::io::common::IoCursor;
 use zksync_types::{commitment::L1BatchWithMetadata, L1BatchNumber};
@@ -120,6 +121,26 @@ impl<'a> Connection<'a> {
         ctx: &ctx::Ctx,
         cert: &attester::BatchQC,
     ) -> Result<(), InsertCertificateError> {
+        use crate::storage::consensus_dal::InsertCertificateError as E;
+
+        let l1_batch_number = L1BatchNumber(cert.message.number.0 as u32);
+
+        let Some(l1_batch) = self
+            .0
+            .blocks_dal()
+            .get_l1_batch_metadata(l1_batch_number)
+            .await
+            .map_err(E::Dal)?
+        else {
+            return Err(E::MissingPayload.into());
+        };
+
+        let l1_batch_info = StoredBatchInfo::from(&l1_batch);
+
+        if l1_batch_info.hash().0 != *cert.message.hash.0.as_bytes() {
+            return Err(E::PayloadMismatch.into());
+        }
+
         Ok(ctx
             .wait(self.0.consensus_dal().insert_batch_certificate(cert))
             .await??)
@@ -344,8 +365,8 @@ impl<'a> Connection<'a> {
 
         // TODO: Fill out the proof when we have the stateless L1 batch validation story finished.
         // It is supposed to be a Merkle proof that the rolling hash of the batch has been included
-        // in the L1 state tree. The state root hash of L1 won't be available in the DB, it requires
-        // an API client.
+        // in the L1 system contract state tree. It is *not* the Ethereum state root hash, so producing
+        // it can be done without an L1 client, which is only required for validation.
         let batch = attester::SyncBatch {
             number,
             payloads,