chore: stop run when an RPC fails to get latest block number (#263)

Author: Maikol

crates/oracle/src/metrics.rs

@@ -18,6 +18,8 @@ pub struct Metrics {
     registry: Registry,
     jrpc_request_duration_seconds: HistogramVec,
     jrpc_failure: IntCounterVec,
+    jrpc_protocol_chain_failure: IntCounterVec,
+    jrpc_indexed_chain_failure: IntCounterVec,
     current_epoch: IntGaugeVec,
     last_sent_message: Gauge,
     latest_block_number: IntGaugeVec,
@@ -45,6 +47,20 @@ impl Metrics {
             registry
         )?;
 
+        let jrpc_protocol_chain_failure = register_int_counter_vec_with_registry!(
+            "epoch_block_oracle_jrpc_protocol_chain_failure_total",
+            "JSON RPC Protocol Chain Request Failure",
+            &["network"],
+            registry
+        )?;
+
+        let jrpc_indexed_chain_failure = register_int_counter_vec_with_registry!(
+            "epoch_block_oracle_jrpc_indexed_chain_failure_total",
+            "JSON RPC Indexed Chain Request Failure",
+            &["network"],
+            registry
+        )?;
+
         let current_epoch = register_int_gauge_vec_with_registry!(
             "epoch_block_oracle_current_epoch",
             "Current Epoch",
@@ -93,6 +109,8 @@ impl Metrics {
             registry,
             jrpc_request_duration_seconds,
             jrpc_failure,
+            jrpc_protocol_chain_failure,
+            jrpc_indexed_chain_failure,
             current_epoch,
             last_sent_message,
             latest_block_number,
@@ -162,6 +180,20 @@ impl Metrics {
             .unwrap()
             .inc();
     }
+
+    pub fn track_jrpc_protocol_chain_failure(&self, network: &str) {
+        self.jrpc_protocol_chain_failure
+            .get_metric_with_label_values(&[network])
+            .unwrap()
+            .inc();
+    }
+
+    pub fn track_jrpc_indexed_chain_failure(&self, network: &str) {
+        self.jrpc_indexed_chain_failure
+            .get_metric_with_label_values(&[network])
+            .unwrap()
+            .inc();
+    }
 }
 
 pub async fn metrics_server(metrics: &'static Metrics, port: u16) {

crates/oracle/src/runner/oracle.rs

@@ -87,9 +87,15 @@ impl Oracle {
             Err(other) => return Err(other),
         };
 
-        let protocol_chain_current_block = get_latest_block(self.protocol_chain.web3.clone())
+        let protocol_chain_current_block = match get_latest_block(self.protocol_chain.web3.clone())
             .await
-            .map_err(Error::BadJrpcProtocolChain)?;
+        {
+            Ok(block) => block,
+            Err(e) => {
+                METRICS.track_jrpc_protocol_chain_failure(self.config.protocol_chain.id.as_str());
+                return Err(Error::BadJrpcProtocolChain(e));
+            }
+        };
         debug!(
             number = protocol_chain_current_block.number,
             hash = hex::encode(protocol_chain_current_block.hash).as_str(),
@@ -155,72 +161,63 @@ impl Oracle {
         info!("Collecting latest block information from all indexed chains.");
 
         let latest_jrpc_blocks_res = get_latest_blocks(&self.indexed_chains).await;
-        let latest_jrpc_blocks: BTreeMap<Caip2ChainId, BlockPtr> = latest_jrpc_blocks_res
-            .iter()
-            .filter_map(|(chain_id, res)| -> Option<(Caip2ChainId, BlockPtr)> {
-                match res {
-                    Ok(block) => {
-                        METRICS.set_latest_block_number(
-                            chain_id.as_str(),
-                            "jrpc",
-                            block.number as i64,
-                        );
-                        Some((chain_id.clone(), *block))
-                    }
-                    Err(e) => {
-                        warn!(
-                            chain_id = chain_id.as_str(),
-                            error = e.to_string().as_str(),
-                            "Failed to get latest block from chain. Skipping."
-                        );
-                        None
-                    }
+        let mut latest_jrpc_blocks = BTreeMap::new();
+        latest_jrpc_blocks_res
+            .into_iter()
+            .try_for_each(|(chain_id, res)| match res {
+                Ok(block) => {
+                    METRICS.set_latest_block_number(chain_id.as_str(), "jrpc", block.number as i64);
+                    latest_jrpc_blocks.insert(chain_id.clone(), block);
+                    Ok(())
                 }
-            })
-            .collect();
+                Err(e) => {
+                    METRICS.track_jrpc_indexed_chain_failure(chain_id.as_str());
+                    Err(Error::BadJrpcIndexedChain {
+                        chain_id: chain_id.clone(),
+                        error: web3::Error::InvalidResponse(e.to_string()),
+                    })
+                }
+            })?;
 
         let latest_blockmeta_blocks_res =
             get_latest_blockmeta_blocks(&self.blockmeta_indexed_chains).await;
-        let latest_blockmeta_blocks: BTreeMap<Caip2ChainId, BlockPtr> = latest_blockmeta_blocks_res
-            .iter()
-            .filter_map(|(chain_id, res)| -> Option<(Caip2ChainId, BlockPtr)> {
-                match res {
-                    Ok(block) => {
-                        METRICS.set_latest_block_number(
-                            chain_id.as_str(),
-                            "blockmeta",
-                            block.num as i64,
-                        );
-
-                        match block.id.clone().parse::<BlockHash>() {
-                            Ok(hash) => {
-                                let block_ptr = BlockPtr {
-                                    number: block.num,
-                                    hash: hash.0,
-                                };
-                                Some((chain_id.clone(), block_ptr))
-                            }
-                            Err(e) => {
-                                warn!(
-                                    chain_id = chain_id.as_str(),
-                                    error = e.to_string().as_str(),
-                                    "Failed to parse block hash. Skipping."
-                                );
-                                None
-                            }
+        let mut latest_blockmeta_blocks = BTreeMap::new();
+        latest_blockmeta_blocks_res
+            .into_iter()
+            .try_for_each(|(chain_id, res)| match res {
+                Ok(block) => {
+                    METRICS.set_latest_block_number(
+                        chain_id.as_str(),
+                        "blockmeta",
+                        block.num as i64,
+                    );
+
+                    match block.id.clone().parse::<BlockHash>() {
+                        Ok(hash) => {
+                            let block_ptr = BlockPtr {
+                                number: block.num,
+                                hash: hash.0,
+                            };
+                            latest_blockmeta_blocks.insert(chain_id.clone(), block_ptr);
+                            Ok(())
+                        }
+                        Err(e) => {
+                            METRICS.track_jrpc_indexed_chain_failure(chain_id.as_str());
+                            Err(Error::BadJrpcIndexedChain {
+                                chain_id: chain_id.clone(),
+                                error: web3::Error::InvalidResponse(e.to_string()),
+                            })
                         }
-                    }
-                    Err(e) => {
-                        warn!(
-                            chain_id = chain_id.as_str(),
-                            error = e.to_string().as_str(),
-                            "Failed to get latest block from chain. Skipping."
-                        );
-                        None
                     }
                 }
-            })
-            .collect();
+                Err(e) => {
+                    METRICS.track_jrpc_indexed_chain_failure(chain_id.as_str());
+                    Err(Error::BadJrpcIndexedChain {
+                        chain_id: chain_id.clone(),
+                        error: web3::Error::InvalidResponse(e.to_string()),
+                    })
+                }
+            })?;
 
         let latest_blocks: BTreeMap<Caip2ChainId, BlockPtr> = latest_jrpc_blocks
             .into_iter()
@@ -395,13 +392,13 @@ mod freshness {
     /// targeting the DataEdge contract.
     ///
     /// To assert that, the Block Oracle will need to get the latest block from a JSON RPC provider
-    /// and compare its number with the subgraph’s current block.
+    /// and compare its number with the subgraph's current block.
     ///
     /// If they are way too different, then the subgraph is not fresh, and we should gracefully
     /// handle that error.
     ///
     /// Otherwise, if block numbers are under a certain threshold apart, we could scan the blocks
-    /// in between and ensure they’re not relevant to the DataEdge contract.
+    /// in between and ensure they're not relevant to the DataEdge contract.
     pub async fn subgraph_is_fresh<T>(
         subgraph_latest_block: U64,
         current_block: U64,