feat: L1 batch signing (BFT-474)

Cargo.toml

@@ -206,16 +206,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(())
     }
 
@@ -480,6 +461,40 @@ impl ConsensusDal<'_, '_> {
             .number
             .map(|number| attester::BatchNumber(number as u64)))
     }
+
+    /// Get the earliest of L1 batch which does not have a corresponding quorum certificate
+    /// and need signatures to be gossiped and collected.
+    ///
+    /// On the main node this means every L1 batch, because we need QC over all of them to be
+    /// able to submit them to L1. Replicas don't necessarily have to have the QC, because once
+    /// the batch is on L1 and it's final, they can get the batch from there and don't need the
+    /// attestations. The caller will have to choose the `min_batch_number` accordingly.
+    pub async fn earliest_batch_number_to_sign(
+        &mut self,
+        min_batch_number: attester::BatchNumber,
+    ) -> DalResult<Option<attester::BatchNumber>> {
+        let row = sqlx::query!(
+            r#"
+            SELECT
+                MIN(b.number) AS "number"
+            FROM
+                l1_batches b
+                LEFT JOIN l1_batches_consensus c ON b.number = c.l1_batch_number
+            WHERE
+                b.number >= $1
+                AND c.l1_batch_number IS NULL
+            "#,
+            min_batch_number.0 as i64
+        )
+        .instrument("earliest_batch_number_to_sign")
+        .report_latency()
+        .fetch_one(self.storage)
+        .await?;
+
+        Ok(row
+            .number
+            .map(|number| attester::BatchNumber(number as u64)))
+    }
 }
 
 #[cfg(test)]
@@ -551,7 +566,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 +628,30 @@ 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();
+
+        // Check the earliest batch number to be signed at various points.
+        for (i, (min_l1_batch_number, want_earliest)) in [
+            (0, Some(batch_number - 2)), // We inserted 2 unsigned batches before the first cert
+            (batch_number, Some(batch_number + 1)), // This one has the corresponding cert
+            (batch_number + 1, Some(batch_number + 1)), // This is the one we inserted later without a cert
+            (batch_number + 2, None),                   // Querying beyond the last one
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            let min_batch_number = attester::BatchNumber(u64::from(min_l1_batch_number));
+            let earliest = conn
+                .consensus_dal()
+                .earliest_batch_number_to_sign(min_batch_number)
+                .await
+                .unwrap();
+            assert_eq!(earliest.map(|n| n.0 as u32), want_earliest, "test case {i}");
+        }
     }
 }

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,28 @@ 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: the batch interval is ~1 minute,
+                // so there will be ~60 blocks, and an Ethereum Merkle proof is ~1kB.
+                // Using 100 to be generous.
+                max_payload_size * 100 + 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 +146,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(),
             ),