feat(l2): monitor handle panics (#3607)

**Motivation**
Monitor had panics in its code.
<!-- Why does this pull request exist? What are its goals? -->

**Description**
Added new variants for `MonitorError` and used them to remove the panics
in the monitor

<!-- A clear and concise general description of the changes this PR
introduces -->

**How to test**

- Add `--monitor` to the `init-l2-no-metrics` target in
`crates/l2/Makefile`.
- Run a Sequencer (I suggest `make restart` in `crates/l2`).
- Run the prover with `make init-prover` in `crates/l2`.
- Run `make test` in `crates/l2`.

<!-- Link to issues: Resolves #111, Resolves #222 -->

Closes
https://github.com/orgs/lambdaclass/projects/37/views/10?pane=issue&itemId=118823684&issue=lambdaclass%7Cethrex%7C3536

---------

Co-authored-by: Ivan Litteri <[email protected]>

Author: gianbelinche

crates/l2/monitor/app.rs

@@ -58,14 +58,14 @@ impl EthrexMonitor {
         store: Store,
         rollup_store: StoreRollup,
         cfg: &SequencerConfig,
-    ) -> Self {
+    ) -> Result<Self, MonitorError> {
         let eth_client = EthClient::new(cfg.eth.rpc_url.first().expect("No RPC URLs provided"))
             .expect("Failed to create EthClient");
         // TODO: De-hardcode the rollup client URL
         let rollup_client =
             EthClient::new("http://localhost:1729").expect("Failed to create RollupClient");
 
-        EthrexMonitor {
+        Ok(EthrexMonitor {
             title: if cfg.based.based {
                 "Based Ethrex Monitor".to_string()
             } else {
@@ -89,25 +89,25 @@ impl EthrexMonitor {
                 &eth_client,
                 &rollup_store,
             )
-            .await,
-            blocks_table: BlocksTable::new(&store).await,
+            .await?,
+            blocks_table: BlocksTable::new(&store).await?,
             l1_to_l2_messages: L1ToL2MessagesTable::new(
                 cfg.l1_watcher.bridge_address,
                 &eth_client,
                 &store,
             )
-            .await,
+            .await?,
             l2_to_l1_messages: L2ToL1MessagesTable::new(
                 cfg.l1_watcher.bridge_address,
                 &eth_client,
                 &rollup_client,
             )
-            .await,
+            .await?,
             eth_client,
             rollup_client,
             store,
             rollup_store,
-        }
+        })
     }
 
     pub async fn start(mut self) -> Result<(), MonitorError> {
@@ -147,7 +147,7 @@ impl EthrexMonitor {
 
             let timeout = Duration::from_millis(self.tick_rate).saturating_sub(last_tick.elapsed());
             if !event::poll(timeout)? {
-                self.on_tick().await;
+                self.on_tick().await?;
                 last_tick = Instant::now();
                 continue;
             }
@@ -207,22 +207,24 @@ impl EthrexMonitor {
         }
     }
 
-    pub async fn on_tick(&mut self) {
+    pub async fn on_tick(&mut self) -> Result<(), MonitorError> {
         self.node_status.on_tick(&self.store).await;
         self.global_chain_status
             .on_tick(&self.eth_client, &self.store, &self.rollup_store)
             .await;
         self.mempool.on_tick(&self.rollup_client).await;
         self.batches_table
             .on_tick(&self.eth_client, &self.rollup_store)
-            .await;
-        self.blocks_table.on_tick(&self.store).await;
+            .await?;
+        self.blocks_table.on_tick(&self.store).await?;
         self.l1_to_l2_messages
             .on_tick(&self.eth_client, &self.store)
-            .await;
+            .await?;
         self.l2_to_l1_messages
             .on_tick(&self.eth_client, &self.rollup_client)
-            .await;
+            .await?;
+
+        Ok(())
     }
 }
 

crates/l2/monitor/mod.rs

@@ -1,8 +1,7 @@
 // TODO: Handle this expects
 #![expect(clippy::expect_used)]
-#![expect(clippy::panic)]
 #![expect(clippy::indexing_slicing)]
-
+#[expect(clippy::result_large_err)]
 pub(crate) mod app;
 pub(crate) mod utils;
 pub(crate) mod widget;
@@ -21,7 +20,7 @@ pub async fn start_monitor(
     rollup_store: StoreRollup,
     cfg: SequencerConfig,
 ) -> Result<(), SequencerError> {
-    let app = EthrexMonitor::new(sequencer_state, store, rollup_store, &cfg).await;
+    let app = EthrexMonitor::new(sequencer_state, store, rollup_store, &cfg).await?;
     app.start().await?;
     Ok(())
 }

crates/l2/monitor/utils.rs

@@ -4,12 +4,14 @@ use ethrex_common::{Address, U256};
 use ethrex_rpc::{EthClient, types::receipt::RpcLog};
 use keccak_hash::keccak;
 
+use crate::sequencer::errors::MonitorError;
+
 pub async fn get_logs(
     last_block_fetched: &mut U256,
     emitter: Address,
     logs_signatures: Vec<&str>,
     client: &EthClient,
-) -> Vec<RpcLog> {
+) -> Result<Vec<RpcLog>, MonitorError> {
     let last_block_number = client
         .get_block_number()
         .await
@@ -31,13 +33,19 @@ pub async fn get_logs(
                     .collect(),
             )
             .await
-            .unwrap_or_else(|_| panic!("Failed to fetch {logs_signatures:?} logs from {emitter}"));
+            .map_err(|e| {
+                MonitorError::LogsSignatures(
+                    logs_signatures.iter().map(|s| s.to_string()).collect(),
+                    emitter,
+                    e,
+                )
+            })?;
 
         // Update the last L1 block fetched.
         *last_block_fetched = new_last_l1_fetched_block;
 
         batch_committed_logs.extend_from_slice(&logs);
     }
 
-    batch_committed_logs
+    Ok(batch_committed_logs)
 }

crates/l2/monitor/widget/batches.rs

@@ -9,7 +9,10 @@ use ratatui::{
     widgets::{Block, Row, StatefulWidget, Table, TableState},
 };
 
-use crate::monitor::widget::{HASH_LENGTH_IN_DIGITS, NUMBER_LENGTH_IN_DIGITS};
+use crate::{
+    monitor::widget::{HASH_LENGTH_IN_DIGITS, NUMBER_LENGTH_IN_DIGITS},
+    sequencer::errors::MonitorError,
+};
 
 pub struct BatchesTable {
     pub state: TableState,
@@ -25,43 +28,49 @@ impl BatchesTable {
         on_chain_proposer_address: Address,
         eth_client: &EthClient,
         rollup_store: &StoreRollup,
-    ) -> Self {
+    ) -> Result<Self, MonitorError> {
         let mut last_l1_block_fetched = 0;
         let items = Self::fetch_new_items(
             &mut last_l1_block_fetched,
             on_chain_proposer_address,
             eth_client,
             rollup_store,
         )
-        .await;
-        Self {
+        .await?;
+        Ok(Self {
             state: TableState::default(),
             items,
             last_l1_block_fetched,
             on_chain_proposer_address,
-        }
+        })
     }
 
-    pub async fn on_tick(&mut self, eth_client: &EthClient, rollup_store: &StoreRollup) {
+    pub async fn on_tick(
+        &mut self,
+        eth_client: &EthClient,
+        rollup_store: &StoreRollup,
+    ) -> Result<(), MonitorError> {
         let mut new_latest_batches = Self::fetch_new_items(
             &mut self.last_l1_block_fetched,
             self.on_chain_proposer_address,
             eth_client,
             rollup_store,
         )
-        .await;
+        .await?;
         new_latest_batches.truncate(50);
 
         let n_new_latest_batches = new_latest_batches.len();
         self.items.truncate(50 - n_new_latest_batches);
-        self.refresh_items(rollup_store).await;
+        self.refresh_items(rollup_store).await?;
         self.items.extend_from_slice(&new_latest_batches);
         self.items.rotate_right(n_new_latest_batches);
+
+        Ok(())
     }
 
-    async fn refresh_items(&mut self, rollup_store: &StoreRollup) {
+    async fn refresh_items(&mut self, rollup_store: &StoreRollup) -> Result<(), MonitorError> {
         if self.items.is_empty() {
-            return;
+            return Ok(());
         }
 
         let mut from = self.items.last().expect("Expected items in the table").0 - 1;
@@ -71,48 +80,52 @@ impl BatchesTable {
             self.items.first().expect("Expected items in the table").0,
             rollup_store,
         )
-        .await;
+        .await?;
 
         let refreshed_items = Self::process_batches(refreshed_batches).await;
 
         self.items = refreshed_items;
+
+        Ok(())
     }
 
     async fn fetch_new_items(
         last_l2_batch_fetched: &mut u64,
         on_chain_proposer_address: Address,
         eth_client: &EthClient,
         rollup_store: &StoreRollup,
-    ) -> Vec<(u64, u64, usize, Option<H256>, Option<H256>)> {
+    ) -> Result<Vec<(u64, u64, usize, Option<H256>, Option<H256>)>, MonitorError> {
         let last_l2_batch_number = eth_client
             .get_last_committed_batch(on_chain_proposer_address)
             .await
             .expect("Failed to get latest L2 batch");
 
         let new_batches =
-            Self::get_batches(last_l2_batch_fetched, last_l2_batch_number, rollup_store).await;
+            Self::get_batches(last_l2_batch_fetched, last_l2_batch_number, rollup_store).await?;
 
-        Self::process_batches(new_batches).await
+        Ok(Self::process_batches(new_batches).await)
     }
 
-    async fn get_batches(from: &mut u64, to: u64, rollup_store: &StoreRollup) -> Vec<Batch> {
+    async fn get_batches(
+        from: &mut u64,
+        to: u64,
+        rollup_store: &StoreRollup,
+    ) -> Result<Vec<Batch>, MonitorError> {
         let mut new_batches = Vec::new();
 
         for batch_number in *from + 1..=to {
             let batch = rollup_store
                 .get_batch(batch_number)
                 .await
-                .unwrap_or_else(|err| {
-                    panic!("Failed to get batch by number ({batch_number}): {err}")
-                })
-                .unwrap_or_else(|| panic!("Batch {batch_number} not found in the rollup store"));
+                .map_err(|e| MonitorError::RollupStore(batch_number, e))?
+                .ok_or(MonitorError::BatchNotFound(batch_number))?;
 
             *from = batch_number;
 
             new_batches.push(batch);
         }
 
-        new_batches
+        Ok(new_batches)
     }
 
     async fn process_batches(

crates/l2/monitor/widget/blocks.rs

@@ -11,9 +11,12 @@ use ratatui::{
     widgets::{Row, StatefulWidget, Table, TableState},
 };
 
-use crate::monitor::widget::{
-    ADDRESS_LENGTH_IN_DIGITS, BLOCK_SIZE_LENGTH_IN_DIGITS, GAS_USED_LENGTH_IN_DIGITS,
-    HASH_LENGTH_IN_DIGITS, NUMBER_LENGTH_IN_DIGITS, TX_NUMBER_LENGTH_IN_DIGITS,
+use crate::{
+    monitor::widget::{
+        ADDRESS_LENGTH_IN_DIGITS, BLOCK_SIZE_LENGTH_IN_DIGITS, GAS_USED_LENGTH_IN_DIGITS,
+        HASH_LENGTH_IN_DIGITS, NUMBER_LENGTH_IN_DIGITS, TX_NUMBER_LENGTH_IN_DIGITS,
+    },
+    sequencer::errors::MonitorError,
 };
 
 pub struct BlocksTable {
@@ -24,35 +27,37 @@ pub struct BlocksTable {
 }
 
 impl BlocksTable {
-    pub async fn new(store: &Store) -> Self {
+    pub async fn new(store: &Store) -> Result<Self, MonitorError> {
         let mut last_l2_block_known = 0;
-        let items = Self::refresh_items(&mut last_l2_block_known, store).await;
-        Self {
+        let items = Self::refresh_items(&mut last_l2_block_known, store).await?;
+        Ok(Self {
             state: TableState::default(),
             items,
             last_l2_block_known,
-        }
+        })
     }
 
-    pub async fn on_tick(&mut self, store: &Store) {
-        let mut new_blocks = Self::refresh_items(&mut self.last_l2_block_known, store).await;
+    pub async fn on_tick(&mut self, store: &Store) -> Result<(), MonitorError> {
+        let mut new_blocks = Self::refresh_items(&mut self.last_l2_block_known, store).await?;
         new_blocks.truncate(50);
 
         let n_new_blocks = new_blocks.len();
         self.items.truncate(50 - n_new_blocks);
         self.items.extend_from_slice(&new_blocks);
         self.items.rotate_right(n_new_blocks);
+
+        Ok(())
     }
 
     async fn refresh_items(
         last_l2_block_known: &mut u64,
         store: &Store,
-    ) -> Vec<(String, String, String, String, String, String, String)> {
-        let new_blocks = Self::get_blocks(last_l2_block_known, store).await;
+    ) -> Result<Vec<(String, String, String, String, String, String, String)>, MonitorError> {
+        let new_blocks = Self::get_blocks(last_l2_block_known, store).await?;
 
         let new_blocks_processed = Self::process_blocks(new_blocks).await;
 
-        new_blocks_processed
+        Ok(new_blocks_processed
             .iter()
             .map(|(number, n_txs, hash, coinbase, gas, blob_gas, size)| {
                 (
@@ -65,10 +70,13 @@ impl BlocksTable {
                     size.to_string(),
                 )
             })
-            .collect()
+            .collect())
     }
 
-    async fn get_blocks(last_l2_block_known: &mut u64, store: &Store) -> Vec<Block> {
+    async fn get_blocks(
+        last_l2_block_known: &mut u64,
+        store: &Store,
+    ) -> Result<Vec<Block>, MonitorError> {
         let last_l2_block_number = store
             .get_latest_block_number()
             .await
@@ -81,20 +89,16 @@ impl BlocksTable {
             let new_block = store
                 .get_block_by_number(new_last_l1_fetched_block)
                 .await
-                .unwrap_or_else(|_| {
-                    panic!("Failed to get block  by number ({new_last_l1_fetched_block})")
-                })
-                .unwrap_or_else(|| {
-                    panic!("Block {new_last_l1_fetched_block} not found in the store")
-                });
+                .map_err(|e| MonitorError::GetBlockByNumber(new_last_l1_fetched_block, e))?
+                .ok_or(MonitorError::BlockNotFound(new_last_l1_fetched_block))?;
 
             // Update the last L1 block fetched.
             *last_l2_block_known = new_last_l1_fetched_block;
 
             new_blocks.push(new_block);
         }
 
-        new_blocks
+        Ok(new_blocks)
     }
 
     async fn process_blocks(