fix(eth-watch): return internal errors from BatchRootProcessor (#4285)

## What ❔

Return internal errors from BatchRootProcessor instead of client errors

## Why ❔

BatchRootProcessor has its state -- merkle tree. It must be consistent
with postgres: after each `process_events` call, set of batches that
have `batch_chain_merkle_path` in postgres and set of batches present in
the merkle tree should be the same. Otherwise on restart eth watch will
be in crashloop.
Eth watch treats client error as retryable so if merkle tree was updated
and then client returns error -> process_events returns before
batch_chain_merkle_path is stored. Returning internal error fixes the
issue because eth watch exits and restarts on them

## Is this a breaking change?
- [ ] Yes
- [x] 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: perekopskiy

core/node/eth_watch/src/event_processors/appended_chain_batch_root.rs

@@ -65,7 +65,8 @@ impl EventProcessor for BatchRootProcessor {
         let mut transaction = storage
             .start_transaction()
             .await
-            .map_err(DalError::generalize)?;
+            .map_err(DalError::generalize)
+            .map_err(EventProcessorError::internal)?;
 
         let grouped_events: Vec<_> = events
             .into_iter()
@@ -113,13 +114,21 @@ impl EventProcessor for BatchRootProcessor {
                 }
             });
 
-        let sl_chain_id = self.sl_l2_client.chain_id().await?;
+        let sl_chain_id = self
+            .sl_l2_client
+            .chain_id()
+            .await
+            .context("sl_l2_client.chain_id()")
+            .map_err(EventProcessorError::internal)?;
         for (sl_l1_batch_number, chain_batches) in new_events {
             let chain_agg_proof = self
                 .sl_l2_client
                 .get_chain_log_proof(sl_l1_batch_number, self.l2_chain_id)
-                .await?
-                .context("Missing chain log proof for finalized batch")?;
+                .await
+                .context("sl_l2_client.get_chain_log_proof()")
+                .map_err(EventProcessorError::internal)?
+                .context("Missing chain log proof for finalized batch")
+                .map_err(EventProcessorError::internal)?;
             let chain_proof_vector =
                 Self::chain_proof_vector(sl_l1_batch_number, chain_agg_proof, sl_chain_id);
 
@@ -128,8 +137,10 @@ impl EventProcessor for BatchRootProcessor {
                     .blocks_dal()
                     .get_l1_batch_l2_l1_merkle_root(*batch_number)
                     .await
-                    .map_err(DalError::generalize)?
-                    .context("Missing l2_l1_merkle_root for finalized batch")?;
+                    .map_err(DalError::generalize)
+                    .map_err(EventProcessorError::internal)?
+                    .context("Missing l2_l1_merkle_root for finalized batch")
+                    .map_err(EventProcessorError::internal)?;
                 assert_eq!(root_from_db, *batch_root);
 
                 self.merkle_tree
@@ -141,7 +152,9 @@ impl EventProcessor for BatchRootProcessor {
             let chain_root_remote = self
                 .sl_l2_client
                 .get_chain_root_l2(sl_l1_batch_number, self.l2_chain_id)
-                .await?;
+                .await
+                .context("sl_l2_client.get_chain_root_l2()")
+                .map_err(EventProcessorError::internal)?;
             assert_eq!(
                 chain_root_local,
                 chain_root_remote.unwrap(),
@@ -172,11 +185,16 @@ impl EventProcessor for BatchRootProcessor {
                     .blocks_dal()
                     .set_batch_chain_merkle_path(*batch_number, proof)
                     .await
-                    .map_err(DalError::generalize)?;
+                    .map_err(DalError::generalize)
+                    .map_err(EventProcessorError::internal)?;
             }
         }
 
-        transaction.commit().await.map_err(DalError::generalize)?;
+        transaction
+            .commit()
+            .await
+            .map_err(DalError::generalize)
+            .map_err(EventProcessorError::internal)?;
 
         Ok(events_count)
     }

core/node/eth_watch/src/event_processors/decentralized_upgrades.rs

@@ -75,38 +75,58 @@ impl EventProcessor for DecentralizedUpgradesEventProcessor {
     ) -> Result<usize, EventProcessorError> {
         let mut upgrades = Vec::new();
         for event in &events {
-            let version = event.topics.get(1).copied().context("missing topic 1")?;
+            let version = event
+                .topics
+                .get(1)
+                .copied()
+                .context("missing topic 1")
+                .map_err(EventProcessorError::internal)?;
             let timestamp: u64 = U256::from_big_endian(&event.data.0)
                 .try_into()
                 .ok()
-                .context("upgrade timestamp is too big")?;
+                .context("upgrade timestamp is too big")
+                .map_err(EventProcessorError::internal)?;
 
             let diamond_cut = self
                 .sl_client
                 .diamond_cut_by_version(version)
-                .await?
-                .context("missing upgrade data on STM")?;
+                .await
+                .map_err(EventProcessorError::client)?
+                .context("missing upgrade data on STM")
+                .map_err(EventProcessorError::internal)?;
 
             let upgrade = ProtocolUpgrade {
                 timestamp,
                 ..ProtocolUpgrade::try_from_diamond_cut(
                     &diamond_cut,
                     self.l1_client.as_ref(),
-                    self.l1_client.get_chain_gateway_upgrade_info().await?,
+                    self.l1_client
+                        .get_chain_gateway_upgrade_info()
+                        .await
+                        .map_err(EventProcessorError::contract_call)?,
                 )
-                .await?
+                .await
+                .map_err(EventProcessorError::internal)?
             };
 
             // Scheduler VK is not present in proposal event. It is hard coded in verifier contract.
             let scheduler_vk_hash = if let Some(address) = upgrade.verifier_address {
-                Some(self.sl_client.scheduler_vk_hash(address).await?)
+                Some(
+                    self.sl_client
+                        .scheduler_vk_hash(address)
+                        .await
+                        .map_err(EventProcessorError::contract_call)?,
+                )
             } else {
                 None
             };
 
             // Scheduler VK is not present in proposal event. It is hard coded in verifier contract.
             let fflonk_scheduler_vk_hash = if let Some(address) = upgrade.verifier_address {
-                self.sl_client.fflonk_scheduler_vk_hash(address).await?
+                self.sl_client
+                    .fflonk_scheduler_vk_hash(address)
+                    .await
+                    .map_err(EventProcessorError::contract_call)?
             } else {
                 None
             };
@@ -135,21 +155,25 @@ impl EventProcessor for DecentralizedUpgradesEventProcessor {
                 .protocol_versions_dal()
                 .latest_semantic_version()
                 .await
-                .map_err(DalError::generalize)?
-                .context("expected some version to be present in DB")?;
+                .map_err(DalError::generalize)
+                .map_err(EventProcessorError::internal)?
+                .context("expected some version to be present in DB")
+                .map_err(EventProcessorError::internal)?;
 
             if upgrade.version > latest_semantic_version {
                 let latest_version = storage
                     .protocol_versions_dal()
                     .get_protocol_version_with_latest_patch(latest_semantic_version.minor)
                     .await
-                    .map_err(DalError::generalize)?
+                    .map_err(DalError::generalize)
+                    .map_err(EventProcessorError::internal)?
                     .with_context(|| {
                         format!(
                             "expected minor version {} to be present in DB",
                             latest_semantic_version.minor as u16
                         )
-                    })?;
+                    })
+                    .map_err(EventProcessorError::internal)?;
 
                 let new_version = latest_version.apply_upgrade(
                     upgrade,
@@ -168,7 +192,8 @@ impl EventProcessor for DecentralizedUpgradesEventProcessor {
                     .protocol_versions_dal()
                     .save_protocol_version_with_tx(&new_version)
                     .await
-                    .map_err(DalError::generalize)?;
+                    .map_err(DalError::generalize)
+                    .map_err(EventProcessorError::internal)?;
             }
         }
         stage_latency.observe();

core/node/eth_watch/src/event_processors/gateway_migration.rs

@@ -34,7 +34,12 @@ impl EventProcessor for GatewayMigrationProcessor {
         events: Vec<Log>,
     ) -> Result<usize, EventProcessorError> {
         for event in &events {
-            let main_topic = event.topics.first().copied().context("missing topic 0")?;
+            let main_topic = event
+                .topics
+                .first()
+                .copied()
+                .context("missing topic 0")
+                .map_err(EventProcessorError::internal)?;
             if !self.possible_main_topics.contains(&main_topic) {
                 continue;
             }
@@ -44,7 +49,8 @@ impl EventProcessor for GatewayMigrationProcessor {
                     .topics
                     .get(1)
                     .copied()
-                    .context("missing topic 1")?
+                    .context("missing topic 1")
+                    .map_err(EventProcessorError::internal)?
                     .as_bytes(),
             );
 

core/node/eth_watch/src/event_processors/mod.rs

@@ -18,19 +18,31 @@ mod priority_ops;
 /// Errors issued by an [`EventProcessor`].
 #[derive(Debug, thiserror::Error)]
 pub(super) enum EventProcessorError {
+    #[error("Fatal: {0:?}")]
+    Fatal(#[from] FatalError),
+    #[error("Transient: {0:?}")]
+    Transient(#[from] TransientError),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(super) enum FatalError {
     #[error("failed parsing a log into {log_kind}: {source:?}")]
     LogParse {
         log_kind: &'static str,
         #[source]
         source: anyhow::Error,
     },
+    /// Internal errors are considered fatal (i.e., they bubble up and lead to the watcher termination).
+    #[error("internal processing error: {0:?}")]
+    Internal(#[from] anyhow::Error),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(super) enum TransientError {
     #[error("Eth client error: {0}")]
     Client(#[from] EnrichedClientError),
     #[error("Contract call error: {0}")]
     ContractCall(#[from] ContractCallError),
-    /// Internal errors are considered fatal (i.e., they bubble up and lead to the watcher termination).
-    #[error("internal processing error: {0:?}")]
-    Internal(#[from] anyhow::Error),
 }
 
 #[derive(Debug)]
@@ -41,10 +53,22 @@ pub(super) enum EventsSource {
 
 impl EventProcessorError {
     pub fn log_parse(source: impl Into<anyhow::Error>, log_kind: &'static str) -> Self {
-        Self::LogParse {
+        Self::Fatal(FatalError::LogParse {
             log_kind,
             source: source.into(),
-        }
+        })
+    }
+
+    pub fn internal(source: impl Into<anyhow::Error>) -> Self {
+        Self::Fatal(FatalError::Internal(source.into()))
+    }
+
+    pub fn client(source: impl Into<EnrichedClientError>) -> Self {
+        Self::Transient(TransientError::Client(source.into()))
+    }
+
+    pub fn contract_call(source: impl Into<ContractCallError>) -> Self {
+        Self::Transient(TransientError::ContractCall(source.into()))
     }
 }
 

core/node/eth_watch/src/event_processors/priority_ops.rs

@@ -88,7 +88,11 @@ impl EventProcessor for PriorityOpsEventProcessor {
         let stage_latency = METRICS.poll_eth_node[&PollStage::PersistL1Txs].start();
         APP_METRICS.processed_txs[&TxStage::added_to_mempool()].inc();
         APP_METRICS.processed_l1_txs[&TxStage::added_to_mempool()].inc();
-        let processed_priority_transactions = self.sl_client.get_total_priority_txs().await?;
+        let processed_priority_transactions = self
+            .sl_client
+            .get_total_priority_txs()
+            .await
+            .map_err(EventProcessorError::contract_call)?;
         let ops_to_insert: Vec<&L1Tx> = new_ops
             .iter()
             .take_while(|op| processed_priority_transactions > op.serial_id().0)
@@ -99,7 +103,8 @@ impl EventProcessor for PriorityOpsEventProcessor {
                 .transactions_dal()
                 .insert_transaction_l1(new_op, new_op.eth_block())
                 .await
-                .map_err(DalError::generalize)?;
+                .map_err(DalError::generalize)
+                .map_err(EventProcessorError::internal)?;
         }
         stage_latency.observe();
         if let Some(last_op) = ops_to_insert.last() {

core/node/eth_watch/src/lib.rs

@@ -162,13 +162,11 @@ impl EthWatch {
                     /* everything went fine */
                     METRICS.eth_poll.inc();
                 }
-                Err(EventProcessorError::Internal(err)) => {
-                    tracing::error!("Internal error processing new blocks: {err:?}");
-                    return Err(err);
+                Err(EventProcessorError::Fatal(err)) => {
+                    tracing::error!("Fatal error processing new blocks: {err:?}");
+                    return Err(err.into());
                 }
-                Err(err) => {
-                    // This is an error because otherwise we could potentially miss a priority operation
-                    // thus entering priority mode, which is not desired.
+                Err(EventProcessorError::Transient(err)) => {
                     tracing::error!("Failed to process new blocks: {err}");
                 }
             }
@@ -188,13 +186,17 @@ impl EthWatch {
                 EventsSource::L1 => self.l1_client.as_ref(),
                 EventsSource::SL => self.sl_client.as_ref(),
             };
-            let chain_id = client.chain_id().await?;
+            let chain_id = client
+                .chain_id()
+                .await
+                .map_err(EventProcessorError::client)?;
 
             let to_block = if processor.only_finalized_block() {
-                client.finalized_block_number().await?
+                client.finalized_block_number().await
             } else {
-                client.confirmed_block_number().await?
-            };
+                client.confirmed_block_number().await
+            }
+            .map_err(EventProcessorError::client)?;
 
             let from_block = storage
                 .eth_watcher_dal()
@@ -204,7 +206,8 @@ impl EthWatch {
                     to_block.saturating_sub(self.event_expiration_blocks),
                 )
                 .await
-                .map_err(DalError::generalize)?;
+                .map_err(DalError::generalize)
+                .map_err(EventProcessorError::internal)?;
 
             // There are no new blocks so there is nothing to be done
             if from_block > to_block {
@@ -219,7 +222,8 @@ impl EthWatch {
                     processor.topic2(),
                     RETRY_LIMIT,
                 )
-                .await?;
+                .await
+                .map_err(EventProcessorError::client)?;
             let processed_events_count = processor
                 .process_events(storage, processor_events.clone())
                 .await?;
@@ -245,7 +249,8 @@ impl EthWatch {
                     next_block_to_process,
                 )
                 .await
-                .map_err(DalError::generalize)?;
+                .map_err(DalError::generalize)
+                .map_err(EventProcessorError::internal)?;
         }
 
         Ok(())