refactor: Implement earliest_batch_number_to_sign (BFT-474)

Cargo.toml

@@ -287,13 +287,13 @@ zksync_tee_verifier_input_producer = { path = "core/node/tee_verifier_input_prod
 zksync_base_token_adjuster = { path = "core/node/base_token_adjuster" }
 
 [patch.crates-io]
-zksync_concurrency = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_bft = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_crypto = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_executor = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_network = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_roles = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_storage = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_consensus_utils = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_protobuf = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
-zksync_protobuf_build = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "78ade978b38480cab2b4335d9f41155e24d5c9a1" }
+zksync_concurrency = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_bft = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_crypto = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_executor = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_network = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_roles = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_storage = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_consensus_utils = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_protobuf = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }
+zksync_protobuf_build = { version = "0.1.0-rc.1", git = "https://github.com/matter-labs/era-consensus.git", rev = "ed5832d90c3c202dcb28ebcd87926eac51f961d2" }

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

@@ -462,45 +462,38 @@ impl ConsensusDal<'_, '_> {
             .map(|number| attester::BatchNumber(number as u64)))
     }
 
-    /// Get the numbers of L1 batches which do not have a corresponding L1 quorum certificate
+    /// 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.
-    pub async fn unsigned_batch_numbers(
+    /// 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<Vec<attester::BatchNumber>> {
-        Ok(sqlx::query!(
+    ) -> DalResult<Option<attester::BatchNumber>> {
+        let row = sqlx::query!(
             r#"
             SELECT
-                b.number
+                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 NOT EXISTS (
-                    SELECT
-                        1
-                    FROM
-                        l1_batches_consensus c
-                    WHERE
-                        c.l1_batch_number = b.number
-                )
-            ORDER BY
-                b.number
+                AND c.l1_batch_number IS NULL
             "#,
             min_batch_number.0 as i64
         )
-        .instrument("unsigned_batch_numbers")
+        .instrument("earliest_batch_number_to_sign")
         .report_latency()
-        .fetch_all(self.storage)
-        .await?
-        .into_iter()
-        .map(|row| attester::BatchNumber(row.number as u64))
-        .collect())
+        .fetch_one(self.storage)
+        .await?;
+
+        Ok(row
+            .number
+            .map(|number| attester::BatchNumber(number as u64)))
     }
 }
 
@@ -642,25 +635,23 @@ mod tests {
             .await
             .unwrap();
 
-        let num_batches = num_batches + 1;
-        let num_unsigned = num_batches - 1;
-
-        // Check how many unsigned L1 batches we can find.
-        for (i, (min_l1_batch_number, exp_num_unsigned)) in [
-            (0, num_unsigned),
-            (batch_number, 1),     // This one has the corresponding cert
-            (batch_number + 1, 1), // This is the one we inserted later
-            (batch_number + 2, 0), // Querying beyond the last one
+        // 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 unsigned = conn
+            let min_batch_number = attester::BatchNumber(u64::from(min_l1_batch_number));
+            let earliest = conn
                 .consensus_dal()
-                .unsigned_batch_numbers(attester::BatchNumber(u64::from(min_l1_batch_number)))
+                .earliest_batch_number_to_sign(min_batch_number)
                 .await
                 .unwrap();
-            assert_eq!(unsigned.len(), exp_num_unsigned, "test case {i}");
+            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(),
             ),

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

@@ -431,19 +431,19 @@ impl<'a> Connection<'a> {
         })
     }
 
-    /// Wrapper for `consensus_dal().unsigned_batch_numbers()`.
-    pub async fn unsigned_batch_numbers(
+    /// Wrapper for `consensus_dal().earliest_batch_number_to_sign()`.
+    pub async fn earliest_batch_number_to_sign(
         &mut self,
         ctx: &ctx::Ctx,
         min_batch_number: attester::BatchNumber,
-    ) -> ctx::Result<Vec<attester::BatchNumber>> {
+    ) -> ctx::Result<Option<attester::BatchNumber>> {
         Ok(ctx
             .wait(
                 self.0
                     .consensus_dal()
-                    .unsigned_batch_numbers(min_batch_number),
+                    .earliest_batch_number_to_sign(min_batch_number),
             )
             .await?
-            .context("unsigned_batch_numbers()")?)
+            .context("earliest_batch_number_to_sign()")?)
     }
 }