added retry mechanism

Author: Fraccaman

Cargo.lock

@@ -26,6 +26,8 @@ lru = "0.12.5"
 bimap = { version = "0.6.3", features = ["serde"] }
 prometheus_exporter = "0.8.5"
 subtle-encoding = "0.5.1"
+thiserror = "1.0.56"
+tokio-retry2 = { version = "0.5", features = ["jitter", "tracing"] }
 
 [build-dependencies]
 vergen = { version = "8.0.0", features = ["build", "git", "gitcl"] }

Cargo.toml

@@ -6,11 +6,10 @@ use block::BlockHeigtCheck;
 use epoch::EpochCheck;
 use total_supply_native::TotalSupplyNativeCheck;
 
-
 pub enum Checks {
     BlockHeightCheck(BlockHeigtCheck),
     EpochCheck(EpochCheck),
-    TotalSupplyNative(TotalSupplyNativeCheck)
+    TotalSupplyNative(TotalSupplyNativeCheck),
 }
 
 pub fn all_checks() -> Vec<Checks> {

src/checks/mod.rs

@@ -17,6 +17,9 @@ pub struct AppConfig {
     #[clap(long, env, default_value_t = 9184)]
     pub prometheus_port: u64,
 
+    #[clap(long, env, default_value_t = 1)]
+    pub initial_block_height: u64,
+
     #[clap(long, env, default_value_t = 5)]
     pub sleep_for: u64,
 

src/config.rs

@@ -0,0 +1,19 @@
+use tokio_retry2::RetryError;
+
+pub trait AsRetryError<T> {
+    fn into_retry_error(self) -> Result<T, RetryError<std::io::Error>>;
+}
+
+impl<T> AsRetryError<T> for anyhow::Result<T> {
+    #[inline]
+    fn into_retry_error(self) -> Result<T, RetryError<std::io::Error>> {
+        self.map_err(|reason| {
+            tracing::error!(?reason, "RPC error");
+            let err = std::io::Error::new(std::io::ErrorKind::Other, reason.to_string());
+            RetryError::Transient {
+                err,
+                retry_after: None,
+            }
+        })
+    }
+}

src/error.rs

@@ -1,22 +1,30 @@
 pub mod apprise;
 pub mod checks;
 pub mod config;
+pub mod error;
 pub mod log;
 pub mod rpc;
 pub mod shared;
 pub mod state;
 
-use std::net::SocketAddr;
+use std::{net::SocketAddr, sync::Arc};
 
 use anyhow::Context;
 use apprise::AppRise;
 use checks::all_checks;
 use clap::Parser;
 use config::AppConfig;
+use error::AsRetryError;
 use prometheus_exporter::prometheus::Registry;
 use rpc::Rpc;
 use shared::checksums::Checksums;
 use state::State;
+use tokio::sync::RwLock;
+use tokio_retry2::{strategy::ExponentialBackoff, Retry};
+
+fn notify(err: &std::io::Error, duration: std::time::Duration) {
+    tracing::info!("Error {err:?} occurred at {duration:?}");
+}
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -36,37 +44,74 @@ async fn main() -> anyhow::Result<()> {
         checksums.add(code_path, code);
     }
 
-    let mut state = State::new(checksums);
-    let registry = state.prometheus_registry();
+    let state = Arc::new(RwLock::new(State::new(
+        checksums,
+        config.initial_block_height,
+    )));
+    let unlocked_state = state.read().await;
+    let registry = unlocked_state.prometheus_registry();
+    drop(unlocked_state);
 
     start_prometheus_exporter(registry, config.prometheus_port)?;
 
+    let retry_strategy = retry_strategy();
+
     loop {
-        let pre_state = state.clone();
-
-        let native_token = rpc.query_native_token().await?;
-        let block_height = state.next_block_height();
-        let epoch = rpc.query_current_epoch(block_height).await?.unwrap_or(0);
-        let block = rpc
-            .query_block(block_height, &state.checksums, epoch)
-            .await?;
-        let total_supply_native = rpc.query_total_supply(&native_token).await?;
-
-        state.update(block, total_supply_native);
-
-        for check_kind in all_checks() {
-            let check_res = match check_kind {
-                checks::Checks::BlockHeightCheck(check) => check.run(&pre_state, &state).await,
-                checks::Checks::EpochCheck(check) => check.run(&pre_state, &state).await,
-                checks::Checks::TotalSupplyNative(check) => check.run(&pre_state, &state).await,
-            };
-            if let Err(error) = check_res {
-                tracing::error!("Error: {}", error.to_string());
-                apprise.send_to_slack(error.to_string()).await?;
-            }
-        }
-
-        tracing::info!("Done block {}", block_height);
+        Retry::spawn_notify(
+            retry_strategy.clone(),
+            || async {
+                let pre_state_lock = state.read().await;
+                let block_height = pre_state_lock.next_block_height();
+                let checksums = pre_state_lock.checksums.clone();
+                let pre_state = pre_state_lock.clone();
+                drop(pre_state_lock);
+
+                let native_token = rpc.query_native_token().await.into_retry_error()?;
+                let epoch = rpc
+                    .query_current_epoch(block_height)
+                    .await
+                    .into_retry_error()?
+                    .unwrap_or(0);
+                let block = rpc
+                    .query_block(block_height, &checksums, epoch)
+                    .await
+                    .into_retry_error()?;
+                let total_supply_native = rpc
+                    .query_total_supply(&native_token)
+                    .await
+                    .into_retry_error()?;
+
+                let mut post_state_lock = state.write().await;
+                post_state_lock.update(block, total_supply_native);
+                let post_state = post_state_lock.clone();
+                drop(post_state_lock);
+
+                for check_kind in all_checks() {
+                    let check_res = match check_kind {
+                        checks::Checks::BlockHeightCheck(check) => {
+                            check.run(&pre_state, &post_state).await
+                        }
+                        checks::Checks::EpochCheck(check) => {
+                            check.run(&pre_state, &post_state).await
+                        }
+                        checks::Checks::TotalSupplyNative(check) => {
+                            check.run(&pre_state, &post_state).await
+                        }
+                    };
+                    if let Err(error) = check_res {
+                        tracing::error!("Error: {}", error.to_string());
+                        apprise
+                            .send_to_slack(error.to_string())
+                            .await
+                            .into_retry_error()?;
+                    }
+                }
+
+                Ok(())
+            },
+            notify,
+        )
+        .await?;
     }
 }
 
@@ -80,3 +125,9 @@ fn start_prometheus_exporter(registry: Registry, port: u64) -> anyhow::Result<()
 
     Ok(())
 }
+
+fn retry_strategy() -> ExponentialBackoff {
+    ExponentialBackoff::from_millis(1000)
+        .factor(1)
+        .max_delay_millis(10000)
+}

src/main.rs

@@ -3,7 +3,7 @@ use std::str::FromStr;
 use anyhow::Context;
 use futures::FutureExt;
 use namada_sdk::{
-    address::Address as NamadaAddress, hash::Hash, ibc::context::client, io::Client, rpc,
+    address::Address as NamadaAddress, hash::Hash, io::Client, rpc,
     state::Key,
 };
 use tendermint_rpc::{HttpClient, Url};

src/rpc.rs

@@ -56,7 +56,9 @@ impl PrometheusMetrics {
             .register(Box::new(block_height_counter.clone()))
             .unwrap();
         registry.register(Box::new(epoch_counter.clone())).unwrap();
-        registry.register(Box::new(total_supply_native_token.clone())).unwrap();
+        registry
+            .register(Box::new(total_supply_native_token.clone()))
+            .unwrap();
 
         Self {
             block_height_counter,
@@ -76,9 +78,9 @@ impl PrometheusMetrics {
 }
 
 impl State {
-    pub fn new(checksums: Checksums) -> Self {
+    pub fn new(checksums: Checksums, block_height: u64) -> Self {
         Self {
-            latest_block_height: None,
+            latest_block_height: Some(block_height),
             latest_epoch: None,
             latest_total_supply_native: None,
             checksums,
@@ -115,9 +117,11 @@ impl State {
         if let Some(total_supply) = self.latest_total_supply_native {
             self.metrics
                 .total_supply_native_token
-                .inc_by(total_supply - total_supply);
+                .inc_by(total_supply_native - total_supply);
         } else {
-            self.metrics.total_supply_native_token.inc_by(total_supply_native);
+            self.metrics
+                .total_supply_native_token
+                .inc_by(total_supply_native);
         }
         self.latest_total_supply_native = Some(total_supply_native);