fix(EN): properly set chain_id for old batch transactions (#4488)

## What ❔

If received batch details contain None/Null chain_id (commit_chain_id,
prove_chain_id, execute_chain_id) set it to config provided chain id for
L1. This happens for old batches.

## Why ❔

If you try syning EN from scratch with a "old" chain you will encounter
following error:
```
Error: One or more tasks failed: ["Task transaction_finality_updater failed: No batch transactions to process for chain id 1 while there are some for None chain_id.Error is thrown so node can restart and reload SL data. If node doesn't make any progress after restart, then it's bug, please contact developers."]
```

Logic for fetching batch transactions to process their finality status
assumes chain_ids are set properly and no transaction should have
chain_id = None. However this is the case for old chains. When fetching
they return *_chain_id: null in zks_getL1BatchDetails for some old
batches.

## Is this a breaking change?
- [ ] Yes
- [ ] No

## Operational changes
<!-- Any config changes? Any new flags? Any changes to any scripts? -->
<!-- Please add anything that non-Matter Labs entities running their own
ZK Chain may need to know -->

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zkstack dev fmt` and `zkstack dev
lint`.

Author: cytadela8

core/bin/external_node/src/node_builder.rs

@@ -281,7 +281,9 @@ impl ExternalNodeBuilder {
     }
 
     fn add_batch_transaction_fetcher_layer(mut self) -> anyhow::Result<Self> {
-        self.node.add_layer(BatchStatusUpdaterLayer);
+        self.node.add_layer(BatchStatusUpdaterLayer::new(
+            self.config.local.networks.l1_chain_id,
+        ));
         Ok(self)
     }
 

core/node/node_sync/src/batch_transaction_fetcher/mod.rs

@@ -13,7 +13,8 @@ use zksync_dal::{Connection, ConnectionPool, Core, CoreDal};
 use zksync_health_check::{Health, HealthStatus, HealthUpdater, ReactiveHealthCheck};
 use zksync_shared_metrics::EN_METRICS;
 use zksync_types::{
-    aggregated_operations::L1BatchAggregatedActionType, api, L1BatchNumber, SLChainId, H256,
+    aggregated_operations::L1BatchAggregatedActionType, api, L1BatchNumber, L1ChainId, SLChainId,
+    H256,
 };
 use zksync_web3_decl::{
     client::{DynClient, L2},
@@ -41,7 +42,7 @@ struct BatchStatusChange {
     number: L1BatchNumber,
     l1_tx_hash: H256,
     happened_at: DateTime<Utc>,
-    sl_chain_id: Option<SLChainId>,
+    sl_chain_id: SLChainId,
 }
 
 #[derive(Debug, Default)]
@@ -103,10 +104,14 @@ struct UpdaterCursor {
     last_executed_l1_batch: L1BatchNumber,
     last_proven_l1_batch: L1BatchNumber,
     last_committed_l1_batch: L1BatchNumber,
+    default_sl_chain_id: SLChainId,
 }
 
 impl UpdaterCursor {
-    async fn new(storage: &mut Connection<'_, Core>) -> anyhow::Result<Self> {
+    async fn new(
+        storage: &mut Connection<'_, Core>,
+        default_sl_chain_id: SLChainId,
+    ) -> anyhow::Result<Self> {
         let first_l1_batch_number = projected_first_l1_batch(storage).await?;
         // Use the snapshot L1 batch, or the genesis batch if we are not using a snapshot. Technically, the snapshot L1 batch
         // is not necessarily proven / executed yet, but since it and earlier batches are not stored, it serves
@@ -132,6 +137,7 @@ impl UpdaterCursor {
             last_executed_l1_batch,
             last_proven_l1_batch,
             last_committed_l1_batch,
+            default_sl_chain_id,
         })
     }
 
@@ -211,7 +217,7 @@ impl UpdaterCursor {
             number: batch_info.number,
             l1_tx_hash,
             happened_at,
-            sl_chain_id,
+            sl_chain_id: sl_chain_id.unwrap_or(self.default_sl_chain_id),
         });
         tracing::info!("Batch {}: {action_str}", batch_info.number);
         FETCHER_METRICS.l1_batch[&stage.into()].set(batch_info.number.0.into());
@@ -230,6 +236,7 @@ impl UpdaterCursor {
 /// L1 transaction information in `zks_getBlockDetails` and `zks_getL1BatchDetails` RPC methods.
 #[derive(Debug)]
 pub struct BatchStatusUpdater {
+    l1_chain_id: SLChainId,
     client: Box<dyn MainNodeClient>,
     pool: ConnectionPool<Core>,
     health_updater: HealthUpdater,
@@ -242,20 +249,27 @@ pub struct BatchStatusUpdater {
 impl BatchStatusUpdater {
     const DEFAULT_SLEEP_INTERVAL: Duration = Duration::from_secs(5);
 
-    pub fn new(client: Box<DynClient<L2>>, pool: ConnectionPool<Core>) -> Self {
+    pub fn new(
+        l1_chain_id: L1ChainId,
+        client: Box<DynClient<L2>>,
+        pool: ConnectionPool<Core>,
+    ) -> Self {
         Self::from_parts(
+            l1_chain_id.into(), // needed because this is used when l1 is sl
             Box::new(client.for_component("batch_transaction_fetcher")),
             pool,
             Self::DEFAULT_SLEEP_INTERVAL,
         )
     }
 
     fn from_parts(
+        l1_chain_id: SLChainId,
         client: Box<dyn MainNodeClient>,
         pool: ConnectionPool<Core>,
         sleep_interval: Duration,
     ) -> Self {
         Self {
+            l1_chain_id,
             client,
             pool,
             health_updater: ReactiveHealthCheck::new("batch_transaction_fetcher").1,
@@ -271,7 +285,7 @@ impl BatchStatusUpdater {
 
     pub async fn run(self, mut stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
         let mut storage = self.pool.connection_tagged("sync_layer").await?;
-        let mut cursor = UpdaterCursor::new(&mut storage).await?;
+        let mut cursor = UpdaterCursor::new(&mut storage, self.l1_chain_id).await?;
         drop(storage);
         tracing::info!("Initialized batch status updater cursor: {cursor:?}");
         self.health_updater
@@ -400,7 +414,7 @@ impl BatchStatusUpdater {
                     change.number,
                     L1BatchAggregatedActionType::Commit,
                     change.l1_tx_hash,
-                    change.sl_chain_id,
+                    Some(change.sl_chain_id),
                 )
                 .await?;
             cursor.last_committed_l1_batch = change.number;
@@ -424,7 +438,7 @@ impl BatchStatusUpdater {
                     change.number,
                     L1BatchAggregatedActionType::PublishProofOnchain,
                     change.l1_tx_hash,
-                    change.sl_chain_id,
+                    Some(change.sl_chain_id),
                 )
                 .await?;
             cursor.last_proven_l1_batch = change.number;
@@ -448,7 +462,7 @@ impl BatchStatusUpdater {
                     change.number,
                     L1BatchAggregatedActionType::Execute,
                     change.l1_tx_hash,
-                    change.sl_chain_id,
+                    Some(change.sl_chain_id),
                 )
                 .await?;
             cursor.last_executed_l1_batch = change.number;

core/node/node_sync/src/batch_transaction_fetcher/tests.rs

@@ -237,7 +237,7 @@ fn mock_change(number: L1BatchNumber) -> BatchStatusChange {
         number,
         l1_tx_hash: H256::zero(),
         happened_at: DateTime::default(),
-        sl_chain_id: Some(SLChainId(0)),
+        sl_chain_id: SLChainId(0),
     }
 }
 
@@ -246,8 +246,12 @@ fn mock_updater(
     pool: ConnectionPool<Core>,
 ) -> (BatchStatusUpdater, mpsc::UnboundedReceiver<StatusChanges>) {
     let (changes_sender, changes_receiver) = mpsc::unbounded_channel();
-    let mut updater =
-        BatchStatusUpdater::from_parts(Box::new(client), pool, Duration::from_millis(10));
+    let mut updater = BatchStatusUpdater::from_parts(
+        SLChainId(1),
+        Box::new(client),
+        pool,
+        Duration::from_millis(10),
+    );
     updater.changes_sender = changes_sender;
     (updater, changes_receiver)
 }
@@ -263,7 +267,9 @@ async fn updater_cursor_for_storage_with_genesis_block() {
         seal_l1_batch(&mut storage, L1BatchNumber(number)).await;
     }
 
-    let mut cursor = UpdaterCursor::new(&mut storage).await.unwrap();
+    let mut cursor = UpdaterCursor::new(&mut storage, SLChainId(1))
+        .await
+        .unwrap();
     assert_eq!(cursor.last_committed_l1_batch, L1BatchNumber(0));
     assert_eq!(cursor.last_proven_l1_batch, L1BatchNumber(0));
     assert_eq!(cursor.last_executed_l1_batch, L1BatchNumber(0));
@@ -283,7 +289,9 @@ async fn updater_cursor_for_storage_with_genesis_block() {
     assert_eq!(cursor.last_proven_l1_batch, L1BatchNumber(1));
     assert_eq!(cursor.last_executed_l1_batch, L1BatchNumber(0));
 
-    let restored_cursor = UpdaterCursor::new(&mut storage).await.unwrap();
+    let restored_cursor = UpdaterCursor::new(&mut storage, SLChainId(1))
+        .await
+        .unwrap();
     assert_eq!(restored_cursor, cursor);
 }
 
@@ -293,7 +301,9 @@ async fn updater_cursor_after_snapshot_recovery() {
     let mut storage = pool.connection().await.unwrap();
     prepare_recovery_snapshot(&mut storage, L1BatchNumber(23), L2BlockNumber(42), &[]).await;
 
-    let cursor = UpdaterCursor::new(&mut storage).await.unwrap();
+    let cursor = UpdaterCursor::new(&mut storage, SLChainId(1))
+        .await
+        .unwrap();
     assert_eq!(cursor.last_committed_l1_batch, L1BatchNumber(23));
     assert_eq!(cursor.last_proven_l1_batch, L1BatchNumber(23));
     assert_eq!(cursor.last_executed_l1_batch, L1BatchNumber(23));

core/node/node_sync/src/node/batch_transaction_fetcher.rs

@@ -8,6 +8,7 @@ use zksync_node_framework::{
     wiring_layer::{WiringError, WiringLayer},
     FromContext, IntoContext,
 };
+use zksync_types::L1ChainId;
 use zksync_web3_decl::client::{DynClient, L2};
 
 use crate::batch_transaction_fetcher::BatchStatusUpdater;
@@ -28,7 +29,15 @@ pub struct Output {
 
 /// Wiring layer for `BatchStatusUpdater`, part of the external node.
 #[derive(Debug)]
-pub struct BatchStatusUpdaterLayer;
+pub struct BatchStatusUpdaterLayer {
+    l1_chain_id: L1ChainId,
+}
+
+impl BatchStatusUpdaterLayer {
+    pub fn new(l1_chain_id: L1ChainId) -> Self {
+        Self { l1_chain_id }
+    }
+}
 
 #[async_trait::async_trait]
 impl WiringLayer for BatchStatusUpdaterLayer {
@@ -46,7 +55,8 @@ impl WiringLayer for BatchStatusUpdaterLayer {
             app_health,
         } = input;
 
-        let updater = BatchStatusUpdater::new(main_node_client, pool.get().await?);
+        let updater =
+            BatchStatusUpdater::new(self.l1_chain_id, main_node_client, pool.get().await?);
 
         // Insert healthcheck
         app_health