feat(merkle tree): Finalize metadata calculator snapshot recovery log…

…ic (#798)

## What ❔

- Integrates reading snapshot metadata in the metadata calculator.
- Tests full recovery lifecycle for the metadata calculator.

## Why ❔

Right now, metadata calculator never enters recovery mode (since there's
no snapshot metadata stored in Postgres yet).

## 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.
- [ ] Documentation comments have been added / updated.
- [x] Code has been formatted via `zk fmt` and `zk lint`.
- [x] Spellcheck has been run via `cargo spellcheck
--cfg=./spellcheck/era.cfg --code 1`.

core/bin/external_node/src/main.rs

@@ -399,12 +399,15 @@ async fn main() -> anyhow::Result<()> {
             L1ExecutedBatchesRevert::Allowed,
         );
 
-        let mut connection = connection_pool.access_storage().await.unwrap();
+        let mut connection = connection_pool.access_storage().await?;
         let sealed_l1_batch_number = connection
             .blocks_dal()
             .get_sealed_l1_batch_number()
             .await
-            .unwrap();
+            .context("Failed getting sealed L1 batch number")?
+            .context(
+                "Cannot roll back pending L1 batch since there are no L1 batches in Postgres",
+            )?;
         drop(connection);
 
         tracing::info!("Rolling back to l1 batch number {sealed_l1_batch_number}");
@@ -418,9 +421,7 @@ async fn main() -> anyhow::Result<()> {
     }
 
     let sigint_receiver = setup_sigint_handler();
-
     tracing::warn!("The external node is in the alpha phase, and should be used with caution.");
-
     tracing::info!("Started the external node");
     tracing::info!("Main node URL is: {}", main_node_url);
 

core/bin/snapshots_creator/src/main.rs

@@ -148,9 +148,9 @@ async fn run(
 
     // We subtract 1 so that after restore, EN node has at least one L1 batch to fetch
     let sealed_l1_batch_number = conn.blocks_dal().get_sealed_l1_batch_number().await?;
-    assert_ne!(
-        sealed_l1_batch_number,
-        L1BatchNumber(0),
+    let sealed_l1_batch_number = sealed_l1_batch_number.context("No L1 batches in Postgres")?;
+    anyhow::ensure!(
+        sealed_l1_batch_number != L1BatchNumber(0),
         "Cannot create snapshot when only the genesis L1 batch is present in Postgres"
     );
     let l1_batch_number = sealed_l1_batch_number - 1;

core/lib/dal/src/blocks_dal.rs

@@ -43,8 +43,8 @@ impl BlocksDal<'_, '_> {
         Ok(count == 0)
     }
 
-    pub async fn get_sealed_l1_batch_number(&mut self) -> anyhow::Result<L1BatchNumber> {
-        let number = sqlx::query!(
+    pub async fn get_sealed_l1_batch_number(&mut self) -> sqlx::Result<Option<L1BatchNumber>> {
+        let row = sqlx::query!(
             r#"
             SELECT
                 MAX(number) AS "number"
@@ -57,11 +57,9 @@ impl BlocksDal<'_, '_> {
         .instrument("get_sealed_block_number")
         .report_latency()
         .fetch_one(self.storage.conn())
-        .await?
-        .number
-        .context("DAL invocation before genesis")?;
+        .await?;
 
-        Ok(L1BatchNumber(number as u32))
+        Ok(row.number.map(|num| L1BatchNumber(num as u32)))
     }
 
     pub async fn get_sealed_miniblock_number(&mut self) -> sqlx::Result<MiniblockNumber> {

core/lib/state/src/rocksdb/mod.rs

@@ -135,15 +135,16 @@ impl RocksdbStorage {
     /// in Postgres.
     pub async fn update_from_postgres(&mut self, conn: &mut StorageProcessor<'_>) {
         let latency = METRICS.update.start();
-        let latest_l1_batch_number = conn
+        let Some(latest_l1_batch_number) = conn
             .blocks_dal()
             .get_sealed_l1_batch_number()
             .await
-            .unwrap();
-        tracing::debug!(
-            "loading storage for l1 batch number {}",
-            latest_l1_batch_number.0
-        );
+            .unwrap()
+        else {
+            // No L1 batches are persisted in Postgres; update is not necessary.
+            return;
+        };
+        tracing::debug!("Loading storage for l1 batch number {latest_l1_batch_number}");
 
         let mut current_l1_batch_number = self.l1_batch_number().0;
         assert!(

core/lib/zksync_core/src/consensus/testonly.rs

@@ -306,7 +306,8 @@ async fn run_mock_metadata_calculator(ctx: &ctx::Ctx, pool: ConnectionPool) -> a
             .blocks_dal()
             .get_sealed_l1_batch_number()
             .await
-            .context("get_sealed_l1_batch_number()")?;
+            .context("get_sealed_l1_batch_number()")?
+            .context("no L1 batches in Postgres")?;
 
         while n <= last {
             let metadata = create_l1_batch_metadata(n.0);

core/lib/zksync_core/src/eth_sender/aggregator.rs

@@ -95,11 +95,15 @@ impl Aggregator {
         protocol_version_id: ProtocolVersionId,
         l1_verifier_config: L1VerifierConfig,
     ) -> Option<AggregatedOperation> {
-        let last_sealed_l1_batch_number = storage
+        let Some(last_sealed_l1_batch_number) = storage
             .blocks_dal()
             .get_sealed_l1_batch_number()
             .await
-            .unwrap();
+            .unwrap()
+        else {
+            return None; // No L1 batches in Postgres; no operations are ready yet
+        };
+
         if let Some(op) = self
             .get_execute_operations(
                 storage,

core/lib/zksync_core/src/house_keeper/blocks_state_reporter.rs

@@ -20,14 +20,16 @@ impl L1BatchMetricsReporter {
     }
 
     async fn report_metrics(&self) {
+        let mut block_metrics = vec![];
         let mut conn = self.connection_pool.access_storage().await.unwrap();
-        let mut block_metrics = vec![(
-            conn.blocks_dal()
-                .get_sealed_l1_batch_number()
-                .await
-                .unwrap(),
-            BlockStage::Sealed,
-        )];
+        let last_l1_batch = conn
+            .blocks_dal()
+            .get_sealed_l1_batch_number()
+            .await
+            .unwrap();
+        if let Some(number) = last_l1_batch {
+            block_metrics.push((number, BlockStage::Sealed));
+        }
 
         let last_l1_batch_with_metadata = conn
             .blocks_dal()