Retry startup

Author: m30m

apps/fortuna/src/command/run.rs

@@ -3,7 +3,7 @@ use {
         api::{self, ApiBlockChainState, BlockchainState, ChainId},
         chain::ethereum::InstrumentedPythContract,
         command::register_provider::CommitmentMetadata,
-        config::{Commitment, Config, EthereumConfig, RunOptions},
+        config::{Commitment, Config, EthereumConfig, ProviderConfig, RunOptions},
         eth_utils::traced_client::RpcMetrics,
         keeper::{self, keeper_metrics::KeeperMetrics},
         state::{HashChainState, PebbleHashChain},
@@ -104,36 +104,29 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
         let chains = chains.clone();
         let secret_copy = secret.clone();
         let rpc_metrics = rpc_metrics.clone();
+        let provider_config = config.provider.clone();
         spawn(async move {
-            let state = setup_chain_state(
-                &config.provider.address,
-                &secret_copy,
-                config.provider.chain_sample_interval,
-                &chain_id,
-                &chain_config,
-                rpc_metrics.clone(),
-            )
-            .await;
-            match state {
-                Ok(state) => {
-                    keeper_metrics.add_chain(chain_id.clone(), state.provider_address);
-                    chains.write().await.insert(
-                        chain_id.clone(),
-                        ApiBlockChainState::Initialized(state.clone()),
-                    );
-                    if let Some(keeper_private_key) = keeper_private_key_option {
-                        spawn(keeper::run_keeper_threads(
-                            keeper_private_key,
-                            chain_config,
-                            state,
-                            keeper_metrics.clone(),
-                            rpc_metrics.clone(),
-                        ));
+            loop {
+                let setup_result = setup_chain(
+                    provider_config.clone(),
+                    &chain_id,
+                    chain_config.clone(),
+                    keeper_metrics.clone(),
+                    keeper_private_key_option.clone(),
+                    chains.clone(),
+                    &secret_copy,
+                    rpc_metrics.clone(),
+                )
+                .await;
+                match setup_result {
+                    Ok(_) => {
+                        tracing::info!("Chain {} initialized successfully", chain_id);
+                        break;
+                    }
+                    Err(e) => {
+                        tracing::error!("Failed to initialize chain {}: {}", chain_id, e);
+                        tokio::time::sleep(tokio::time::Duration::from_secs(15)).await;
                     }
-                }
-                Err(e) => {
-                    tracing::error!("Failed to setup {} {}", chain_id, e);
-                    //TODO: Retry
                 }
             }
         });
@@ -155,6 +148,44 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
     Ok(())
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn setup_chain(
+    provider_config: ProviderConfig,
+    chain_id: &ChainId,
+    chain_config: EthereumConfig,
+    keeper_metrics: Arc<KeeperMetrics>,
+    keeper_private_key_option: Option<String>,
+    chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>>,
+    secret_copy: &str,
+    rpc_metrics: Arc<RpcMetrics>,
+) -> Result<()> {
+    let state = setup_chain_state(
+        &provider_config.address,
+        secret_copy,
+        provider_config.chain_sample_interval,
+        chain_id,
+        &chain_config,
+        rpc_metrics.clone(),
+    )
+    .await?;
+    keeper_metrics.add_chain(chain_id.clone(), state.provider_address);
+    chains.write().await.insert(
+        chain_id.clone(),
+        ApiBlockChainState::Initialized(state.clone()),
+    );
+    if let Some(keeper_private_key) = keeper_private_key_option {
+        keeper::run_keeper_threads(
+            keeper_private_key,
+            chain_config,
+            state,
+            keeper_metrics.clone(),
+            rpc_metrics.clone(),
+        )
+        .await?;
+    }
+    Ok(())
+}
+
 async fn setup_chain_state(
     provider: &Address,
     secret: &str,

apps/fortuna/src/keeper.rs

@@ -60,7 +60,7 @@ pub async fn run_keeper_threads(
     chain_state: BlockchainState,
     metrics: Arc<KeeperMetrics>,
     rpc_metrics: Arc<RpcMetrics>,
-) {
+) -> anyhow::Result<()> {
     tracing::info!("Starting keeper");
     let latest_safe_block = get_latest_safe_block(&chain_state).in_current_span().await;
     tracing::info!("Latest safe block: {}", &latest_safe_block);
@@ -72,8 +72,7 @@ pub async fn run_keeper_threads(
             chain_state.id.clone(),
             rpc_metrics.clone(),
         )
-        .await
-        .expect("Chain config should be valid"),
+        .await?,
     );
     let keeper_address = contract.wallet().address();
 
@@ -230,4 +229,5 @@ pub async fn run_keeper_threads(
         }
         .in_current_span(),
     );
+    Ok(())
 }