fix(external-node): Check txs at insert time instead of read time (#555)

ENs are broken due to tx mismatch. This PR will allow ENs to restart and
hopefully update the state. More information
[here](https://www.notion.so/matterlabs/EN-is-broken-on-testnet-due-to-mismatched-txs-11c68bca2a4047cb8730e21e21f4ac41).

This was discussed with @danil.

## What ❔

<!-- What are the changes this PR brings about? -->
<!-- Example: This PR adds a PR template to the repo. -->
<!-- (For bigger PRs adding more context is appreciated) -->

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## 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 `zk fmt` and `zk lint`.

core/lib/zksync_core/src/sync_layer/batch_status_updater.rs

@@ -139,24 +139,10 @@ impl BatchStatusUpdater {
             .unwrap()
             .number;
 
-        // We don't want to change the internal state until we actually persist the changes.
         let mut last_committed_l1_batch = self.last_committed_l1_batch;
         let mut last_proven_l1_batch = self.last_proven_l1_batch;
         let mut last_executed_l1_batch = self.last_executed_l1_batch;
 
-        assert!(
-            last_executed_l1_batch <= last_proven_l1_batch,
-            "Incorrect local state: executed batch must be proven"
-        );
-        assert!(
-            last_proven_l1_batch <= last_committed_l1_batch,
-            "Incorrect local state: proven batch must be committed"
-        );
-        assert!(
-            last_committed_l1_batch <= last_sealed_batch,
-            "Incorrect local state: unkonwn batch marked as committed"
-        );
-
         let mut batch = last_executed_l1_batch.next();
         // In this loop we try to progress on the batch statuses, utilizing the same request to the node to potentially
         // update all three statuses (e.g. if the node is still syncing), but also skipping the gaps in the statuses
@@ -289,7 +275,16 @@ impl BatchStatusUpdater {
     /// tables to be ever accessed by the `eth_sender` module.
     async fn apply_status_changes(&mut self, changes: StatusChanges) {
         let total_latency = EN_METRICS.batch_status_updater_loop_iteration.start();
-        let mut storage = self.pool.access_storage_tagged("sync_layer").await.unwrap();
+        let mut connection = self.pool.access_storage_tagged("sync_layer").await.unwrap();
+
+        let mut transaction = connection.start_transaction().await.unwrap();
+
+        let last_sealed_batch = transaction
+            .blocks_dal()
+            .get_newest_l1_batch_header()
+            .await
+            .unwrap()
+            .number;
 
         for change in changes.commit.into_iter() {
             tracing::info!(
@@ -298,7 +293,13 @@ impl BatchStatusUpdater {
                 change.l1_tx_hash,
                 change.happened_at
             );
-            storage
+
+            assert!(
+                change.number <= last_sealed_batch,
+                "Incorrect update state: unknown batch marked as committed"
+            );
+
+            transaction
                 .eth_sender_dal()
                 .insert_bogus_confirmed_eth_tx(
                     change.number,
@@ -317,7 +318,13 @@ impl BatchStatusUpdater {
                 change.l1_tx_hash,
                 change.happened_at
             );
-            storage
+
+            assert!(
+                change.number <= self.last_committed_l1_batch,
+                "Incorrect update state: proven batch must be committed"
+            );
+
+            transaction
                 .eth_sender_dal()
                 .insert_bogus_confirmed_eth_tx(
                     change.number,
@@ -337,7 +344,12 @@ impl BatchStatusUpdater {
                 change.happened_at
             );
 
-            storage
+            assert!(
+                change.number <= self.last_proven_l1_batch,
+                "Incorrect update state: executed batch must be proven"
+            );
+
+            transaction
                 .eth_sender_dal()
                 .insert_bogus_confirmed_eth_tx(
                     change.number,
@@ -350,6 +362,8 @@ impl BatchStatusUpdater {
             self.last_executed_l1_batch = change.number;
         }
 
+        transaction.commit().await.unwrap();
+
         total_latency.observe();
     }
 }