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

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

@@ -1825,46 +1825,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/state_keeper/io/mempool.rs

@@ -417,11 +417,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()
@@ -437,7 +439,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,