perf(commitment-generator): Run commitment generation for multiple ba…

…tches in parallel (#1984)

## What ❔

Generates commitments with configurable parallelism.

## Why ❔

- As long as produced commitments are persisted in the same order as
previously (not even atomically), there's still a cursor to be used by
the commitment generator.
- Can provide speed up during node recovery (esp. if treeless mode is
used). Also can speed up some integration tests.

## Checklist

- [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`.
- [x] Spellcheck has been run via `zk spellcheck`.

core/bin/external_node/src/config/mod.rs

@@ -745,6 +745,11 @@ pub(crate) struct ExperimentalENConfig {
     /// Maximum number of files concurrently opened by state keeper cache RocksDB. Useful to fit into OS limits; can be used
     /// as a rudimentary way to control RAM usage of the cache.
     pub state_keeper_db_max_open_files: Option<NonZeroU32>,
+
+    // Commitment generator
+    /// Maximum degree of parallelism during commitment generation, i.e., the maximum number of L1 batches being processed in parallel.
+    /// If not specified, commitment generator will use a value roughly equal to the number of CPU cores with some clamping applied.
+    pub commitment_generator_max_parallelism: Option<NonZeroU32>,
 }
 
 impl ExperimentalENConfig {
@@ -758,6 +763,7 @@ impl ExperimentalENConfig {
             state_keeper_db_block_cache_capacity_mb:
                 Self::default_state_keeper_db_block_cache_capacity_mb(),
             state_keeper_db_max_open_files: None,
+            commitment_generator_max_parallelism: None,
         }
     }
 

core/bin/external_node/src/main.rs

@@ -359,14 +359,13 @@ async fn run_core(
     );
     app_health.insert_component(batch_status_updater.health_check())?;
 
-    let commitment_generator_pool = singleton_pool_builder
-        .build()
-        .await
-        .context("failed to build a commitment_generator_pool")?;
-    let commitment_generator = CommitmentGenerator::new(
-        commitment_generator_pool,
+    let mut commitment_generator = CommitmentGenerator::new(
+        connection_pool.clone(),
         config.optional.l1_batch_commit_data_generator_mode,
     );
+    if let Some(parallelism) = config.experimental.commitment_generator_max_parallelism {
+        commitment_generator.set_max_parallelism(parallelism);
+    }
     app_health.insert_component(commitment_generator.health_check())?;
     let commitment_generator_handle = tokio::spawn(commitment_generator.run(stop_receiver.clone()));
 

core/lib/dal/src/blocks_dal.rs

@@ -164,6 +164,8 @@ impl BlocksDal<'_, '_> {
         Ok(row.number.map(|num| L1BatchNumber(num as u32)))
     }
 
+    /// Gets a number of the earliest L1 batch that is ready for commitment generation (i.e., doesn't have commitment
+    /// yet, and has tree data).
     pub async fn get_next_l1_batch_ready_for_commitment_generation(
         &mut self,
     ) -> DalResult<Option<L1BatchNumber>> {
@@ -190,6 +192,34 @@ impl BlocksDal<'_, '_> {
         Ok(row.map(|row| L1BatchNumber(row.number as u32)))
     }
 
+    /// Gets a number of the last L1 batch that is ready for commitment generation (i.e., doesn't have commitment
+    /// yet, and has tree data).
+    pub async fn get_last_l1_batch_ready_for_commitment_generation(
+        &mut self,
+    ) -> DalResult<Option<L1BatchNumber>> {
+        let row = sqlx::query!(
+            r#"
+            SELECT
+                number
+            FROM
+                l1_batches
+            WHERE
+                hash IS NOT NULL
+                AND commitment IS NULL
+            ORDER BY
+                number DESC
+            LIMIT
+                1
+            "#
+        )
+        .instrument("get_last_l1_batch_ready_for_commitment_generation")
+        .report_latency()
+        .fetch_optional(self.storage)
+        .await?;
+
+        Ok(row.map(|row| L1BatchNumber(row.number as u32)))
+    }
+
     /// Returns the number of the earliest L1 batch with metadata (= state hash) present in the DB,
     /// or `None` if there are no such L1 batches.
     pub async fn get_earliest_l1_batch_number_with_metadata(

core/lib/zksync_core_leftovers/src/lib.rs

@@ -763,8 +763,9 @@ pub async fn initialize_components(
     }
 
     if components.contains(&Component::CommitmentGenerator) {
+        let pool_size = CommitmentGenerator::default_parallelism().get();
         let commitment_generator_pool =
-            ConnectionPool::<Core>::singleton(database_secrets.master_url()?)
+            ConnectionPool::<Core>::builder(database_secrets.master_url()?, pool_size)
                 .build()
                 .await
                 .context("failed to build commitment_generator_pool")?;

core/node/commitment_generator/Cargo.toml

@@ -28,11 +28,16 @@ zk_evm_1_4_1.workspace = true
 zk_evm_1_3_3.workspace = true
 
 tokio = { workspace = true, features = ["time"] }
+futures.workspace = true
+num_cpus.workspace = true
 anyhow.workspace = true
 tracing.workspace = true
 itertools.workspace = true
 serde_json.workspace = true
 
 [dev-dependencies]
-jsonrpsee.workspace = true
 zksync_web3_decl.workspace = true
+zksync_node_genesis.workspace = true
+zksync_node_test_utils.workspace = true
+
+rand.workspace = true