feat(en): Make batch status updater work with pruned data (#863)

## What ❔

Modifies batch status updater so that it works with pruned node data
during snapshot recovery.

## Why ❔

Part of preparations of EN code to support snapshot recovery.

## 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/main.rs

@@ -218,7 +218,7 @@ async fn init_tasks(
             .await
             .context("failed to build a connection pool for BatchStatusUpdater")?,
     )
-    .await;
+    .context("failed initializing batch status updater")?;
 
     // Run the components.
     let tree_stop_receiver = stop_receiver.clone();

core/lib/dal/src/blocks_dal.rs

@@ -1746,46 +1746,6 @@ impl BlocksDal<'_, '_> {
         Ok(Some((H256::from_slice(&hash), row.timestamp as u64)))
     }
 
-    pub async fn get_newest_l1_batch_header(&mut self) -> sqlx::Result<L1BatchHeader> {
-        let last_l1_batch = sqlx::query_as!(
-            StorageL1BatchHeader,
-            r#"
-            SELECT
-                number,
-                l1_tx_count,
-                l2_tx_count,
-                timestamp,
-                is_finished,
-                fee_account_address,
-                l2_to_l1_logs,
-                l2_to_l1_messages,
-                bloom,
-                priority_ops_onchain_data,
-                used_contract_hashes,
-                base_fee_per_gas,
-                l1_gas_price,
-                l2_fair_gas_price,
-                bootloader_code_hash,
-                default_aa_code_hash,
-                protocol_version,
-                compressed_state_diffs,
-                system_logs,
-                pubdata_input
-            FROM
-                l1_batches
-            ORDER BY
-                number DESC
-            LIMIT
-                1
-            "#
-        )
-        .instrument("get_newest_l1_batch_header")
-        .fetch_one(self.storage.conn())
-        .await?;
-
-        Ok(last_l1_batch.into())
-    }
-
     pub async fn get_l1_batch_metadata(
         &mut self,
         number: L1BatchNumber,

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

@@ -183,11 +183,20 @@ impl StateKeeper {
                 .await
                 .context("ensure_genesis_state()")?;
         }
-        let last_batch = storage
+
+        let last_l1_batch_number = storage
+            .blocks_dal()
+            .get_sealed_l1_batch_number()
+            .await
+            .context("get_sealed_l1_batch_number()")?
+            .context("no L1 batches in storage")?;
+        let last_miniblock_header = storage
             .blocks_dal()
-            .get_newest_l1_batch_header()
+            .get_last_sealed_miniblock_header()
             .await
-            .context("get_newest_l1_batch_header()")?;
+            .context("get_last_sealed_miniblock_header()")?
+            .context("no miniblocks in storage")?;
+
         let pending_batch = storage
             .blocks_dal()
             .pending_batch_exists()
@@ -196,13 +205,9 @@ impl StateKeeper {
         let (actions_sender, actions_queue) = ActionQueue::new();
         Ok((
             Self {
-                last_batch: last_batch.number + if pending_batch { 1 } else { 0 },
-                last_block: storage
-                    .blocks_dal()
-                    .get_sealed_miniblock_number()
-                    .await
-                    .context("get_sealed_miniblock_number()")?,
-                last_timestamp: last_batch.timestamp,
+                last_batch: last_l1_batch_number + if pending_batch { 1 } else { 0 },
+                last_block: last_miniblock_header.number,
+                last_timestamp: last_miniblock_header.timestamp,
                 batch_sealed: !pending_batch,
                 fee_per_gas: 10,
                 gas_per_pubdata: 100,

core/lib/zksync_core/src/state_keeper/io/mempool.rs

@@ -415,11 +415,13 @@ impl MempoolIO {
         );
 
         let mut storage = pool.access_storage_tagged("state_keeper").await.unwrap();
-        let last_sealed_l1_batch_header = storage
+        // TODO (PLA-703): Support no L1 batches / miniblocks in the storage
+        let last_sealed_l1_batch_number = storage
             .blocks_dal()
-            .get_newest_l1_batch_header()
+            .get_sealed_l1_batch_number()
             .await
-            .unwrap();
+            .unwrap()
+            .expect("No L1 batches sealed");
         let last_miniblock_number = storage
             .blocks_dal()
             .get_sealed_miniblock_number()
@@ -435,7 +437,7 @@ impl MempoolIO {
             timeout_sealer: TimeoutSealer::new(config),
             filter: L2TxFilter::default(),
             // ^ Will be initialized properly on the first newly opened batch
-            current_l1_batch_number: last_sealed_l1_batch_header.number + 1,
+            current_l1_batch_number: last_sealed_l1_batch_number + 1,
             miniblock_sealer_handle,
             current_miniblock_number: last_miniblock_number + 1,
             fee_account: config.fee_account_addr,