feat(fortuna): Improve startup flow (#2638)

* fix(fortuna): Do not wait for all chains setup in startup

* Use blocking thread for calculating hashchain

* Put block timestamp lag next to other tracking tasks

* Retry startup

apps/fortuna/Cargo.lock

@@ -1554,7 +1554,7 @@ dependencies = [
 
 [[package]]
 name = "fortuna"
-version = "7.5.2"
+version = "7.5.3"
 dependencies = [
  "anyhow",
  "axum",

apps/fortuna/src/api.rs

@@ -43,7 +43,7 @@ pub struct ApiMetrics {
 
 #[derive(Clone)]
 pub struct ApiState {
-    pub chains: Arc<HashMap<ChainId, BlockchainState>>,
+    pub chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>>,
 
     pub metrics_registry: Arc<RwLock<Registry>>,
 
@@ -53,7 +53,7 @@ pub struct ApiState {
 
 impl ApiState {
     pub async fn new(
-        chains: HashMap<ChainId, BlockchainState>,
+        chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>>,
         metrics_registry: Arc<RwLock<Registry>>,
     ) -> ApiState {
         let metrics = ApiMetrics {
@@ -68,7 +68,7 @@ impl ApiState {
         );
 
         ApiState {
-            chains: Arc::new(chains),
+            chains,
             metrics: Arc::new(metrics),
             metrics_registry,
         }
@@ -94,6 +94,12 @@ pub struct BlockchainState {
     pub confirmed_block_status: BlockStatus,
 }
 
+#[derive(Clone)]
+pub enum ApiBlockChainState {
+    Uninitialized,
+    Initialized(BlockchainState),
+}
+
 pub enum RestError {
     /// The caller passed a sequence number that isn't within the supported range
     InvalidSequenceNumber,
@@ -108,6 +114,9 @@ pub enum RestError {
     /// The server cannot currently communicate with the blockchain, so is not able to verify
     /// which random values have been requested.
     TemporarilyUnavailable,
+    /// The server is not able to process the request because the blockchain initialization
+    /// has not been completed yet.
+    Uninitialized,
     /// A catch-all error for all other types of errors that could occur during processing.
     Unknown,
 }
@@ -137,6 +146,11 @@ impl IntoResponse for RestError {
                 "This service is temporarily unavailable",
             )
                 .into_response(),
+            RestError::Uninitialized => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "The service is not yet initialized for this chain, please try again in a few minutes",
+            )
+                .into_response(),
             RestError::Unknown => (
                 StatusCode::INTERNAL_SERVER_ERROR,
                 "An unknown error occurred processing the request",
@@ -172,6 +186,7 @@ pub fn get_register_uri(base_uri: &str, chain_id: &str) -> Result<String> {
 
 #[cfg(test)]
 mod test {
+    use crate::api::ApiBlockChainState;
     use {
         crate::{
             api::{self, ApiState, BinaryEncoding, Blob, BlockchainState, GetRandomValueResponse},
@@ -228,10 +243,16 @@ mod test {
         };
 
         let mut chains = HashMap::new();
-        chains.insert("ethereum".into(), eth_state);
-        chains.insert("avalanche".into(), avax_state);
+        chains.insert(
+            "ethereum".into(),
+            ApiBlockChainState::Initialized(eth_state),
+        );
+        chains.insert(
+            "avalanche".into(),
+            ApiBlockChainState::Initialized(avax_state),
+        );
 
-        let api_state = ApiState::new(chains, metrics_registry).await;
+        let api_state = ApiState::new(Arc::new(RwLock::new(chains)), metrics_registry).await;
 
         let app = api::routes(api_state);
         (TestServer::new(app).unwrap(), eth_read, avax_read)

apps/fortuna/src/api/chain_ids.rs

@@ -15,6 +15,12 @@ responses(
 pub async fn chain_ids(
     State(state): State<crate::api::ApiState>,
 ) -> Result<Json<Vec<ChainId>>, RestError> {
-    let chain_ids = state.chains.iter().map(|(id, _)| id.clone()).collect();
+    let chain_ids = state
+        .chains
+        .read()
+        .await
+        .iter()
+        .map(|(id, _)| id.clone())
+        .collect();
     Ok(Json(chain_ids))
 }

apps/fortuna/src/api/revelation.rs

@@ -1,3 +1,4 @@
+use crate::api::ApiBlockChainState;
 use crate::chain::reader::BlockNumber;
 use {
     crate::api::{ChainId, RequestLabel, RestError},
@@ -46,8 +47,18 @@ pub async fn revelation(
 
     let state = state
         .chains
+        .read()
+        .await
         .get(&chain_id)
-        .ok_or(RestError::InvalidChainId)?;
+        .ok_or(RestError::InvalidChainId)?
+        .clone();
+
+    let state = match state {
+        ApiBlockChainState::Initialized(state) => state,
+        ApiBlockChainState::Uninitialized => {
+            return Err(RestError::Uninitialized);
+        }
+    };
 
     let current_block_number_fut = state
         .contract

apps/fortuna/src/command/register_provider.rs

@@ -52,7 +52,7 @@ pub async fn register_provider_from_config(
 
     let commitment_length = provider_config.chain_length;
     tracing::info!("Generating hash chain");
-    let chain = PebbleHashChain::from_config(
+    let chain = PebbleHashChain::from_config_async(
         &secret,
         chain_id,
         &private_key_string.parse::<LocalWallet>()?.address(),
@@ -60,7 +60,8 @@ pub async fn register_provider_from_config(
         &random,
         commitment_length,
         provider_config.chain_sample_interval,
-    )?;
+    )
+    .await?;
     tracing::info!("Done generating hash chain");
 
     // Arguments to the contract to register our new provider.

apps/fortuna/src/command/run.rs

@@ -1,47 +1,30 @@
 use {
     crate::{
-        api::{self, BlockchainState, ChainId},
+        api::{self, ApiBlockChainState, BlockchainState, ChainId},
         chain::ethereum::InstrumentedPythContract,
         command::register_provider::CommitmentMetadata,
-        config::{Commitment, Config, EthereumConfig, RunOptions},
-        eth_utils::traced_client::{RpcMetrics, TracedClient},
+        config::{Commitment, Config, EthereumConfig, ProviderConfig, RunOptions},
+        eth_utils::traced_client::RpcMetrics,
         keeper::{self, keeper_metrics::KeeperMetrics},
         state::{HashChainState, PebbleHashChain},
     },
     anyhow::{anyhow, Error, Result},
     axum::Router,
-    ethers::{
-        middleware::Middleware,
-        types::{Address, BlockNumber},
-    },
-    futures::future::join_all,
-    prometheus_client::{
-        encoding::EncodeLabelSet,
-        metrics::{family::Family, gauge::Gauge},
-        registry::Registry,
-    },
-    std::{
-        collections::HashMap,
-        net::SocketAddr,
-        sync::Arc,
-        time::{Duration, SystemTime, UNIX_EPOCH},
-    },
+    ethers::types::Address,
+    prometheus_client::{encoding::EncodeLabelSet, registry::Registry},
+    std::{collections::HashMap, net::SocketAddr, sync::Arc},
     tokio::{
         spawn,
         sync::{watch, RwLock},
-        time,
     },
     tower_http::cors::CorsLayer,
     utoipa::OpenApi,
     utoipa_swagger_ui::SwaggerUi,
 };
 
-/// Track metrics in this interval
-const TRACK_INTERVAL: Duration = Duration::from_secs(10);
-
 pub async fn run_api(
     socket_addr: SocketAddr,
-    chains: HashMap<String, api::BlockchainState>,
+    chains: Arc<RwLock<HashMap<String, ApiBlockChainState>>>,
     metrics_registry: Arc<RwLock<Registry>>,
     mut rx_exit: watch::Receiver<bool>,
 ) -> Result<()> {
@@ -93,40 +76,6 @@ pub async fn run_api(
     Ok(())
 }
 
-pub async fn run_keeper(
-    chains: HashMap<String, api::BlockchainState>,
-    config: Config,
-    private_key: String,
-    metrics_registry: Arc<RwLock<Registry>>,
-    rpc_metrics: Arc<RpcMetrics>,
-) -> Result<()> {
-    let mut handles = Vec::new();
-    let keeper_metrics: Arc<KeeperMetrics> = Arc::new({
-        let chain_labels: Vec<(String, Address)> = chains
-            .iter()
-            .map(|(id, state)| (id.clone(), state.provider_address))
-            .collect();
-        KeeperMetrics::new(metrics_registry.clone(), chain_labels).await
-    });
-    for (chain_id, chain_config) in chains {
-        let chain_eth_config = config
-            .chains
-            .get(&chain_id)
-            .expect("All chains should be present in the config file")
-            .clone();
-        let private_key = private_key.clone();
-        handles.push(spawn(keeper::run_keeper_threads(
-            private_key,
-            chain_eth_config,
-            chain_config.clone(),
-            keeper_metrics.clone(),
-            rpc_metrics.clone(),
-        )));
-    }
-
-    Ok(())
-}
-
 pub async fn run(opts: &RunOptions) -> Result<()> {
     let config = Config::load(&opts.config.config)?;
     let secret = config.provider.secret.load()?.ok_or(anyhow!(
@@ -136,41 +85,51 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
     let metrics_registry = Arc::new(RwLock::new(Registry::default()));
     let rpc_metrics = Arc::new(RpcMetrics::new(metrics_registry.clone()).await);
 
-    let mut tasks = Vec::new();
+    let keeper_metrics: Arc<KeeperMetrics> =
+        Arc::new(KeeperMetrics::new(metrics_registry.clone()).await);
+    let keeper_private_key_option = config.keeper.private_key.load()?;
+    if keeper_private_key_option.is_none() {
+        tracing::info!("Not starting keeper service: no keeper private key specified. Please add one to the config if you would like to run the keeper service.")
+    }
+    let chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>> = Arc::new(RwLock::new(
+        config
+            .chains
+            .keys()
+            .map(|chain_id| (chain_id.clone(), ApiBlockChainState::Uninitialized))
+            .collect(),
+    ));
     for (chain_id, chain_config) in config.chains.clone() {
+        let keeper_metrics = keeper_metrics.clone();
+        let keeper_private_key_option = keeper_private_key_option.clone();
+        let chains = chains.clone();
         let secret_copy = secret.clone();
         let rpc_metrics = rpc_metrics.clone();
-        tasks.push(spawn(async move {
-            let state = setup_chain_state(
-                &config.provider.address,
-                &secret_copy,
-                config.provider.chain_sample_interval,
-                &chain_id,
-                &chain_config,
-                rpc_metrics,
-            )
-            .await;
-
-            (chain_id, state)
-        }));
-    }
-    let states = join_all(tasks).await;
-
-    let mut chains: HashMap<ChainId, BlockchainState> = HashMap::new();
-    for result in states {
-        let (chain_id, state) = result?;
-
-        match state {
-            Ok(state) => {
-                chains.insert(chain_id.clone(), state);
+        let provider_config = config.provider.clone();
+        spawn(async move {
+            loop {
+                let setup_result = setup_chain_and_run_keeper(
+                    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);
-            }
-        }
-    }
-    if chains.is_empty() {
-        return Err(anyhow!("No chains were successfully setup"));
+        });
     }
 
     // Listen for Ctrl+C so we can set the exit flag and wait for a graceful shutdown.
@@ -185,27 +144,45 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
         Ok::<(), Error>(())
     });
 
-    if let Some(keeper_private_key) = config.keeper.private_key.load()? {
-        spawn(run_keeper(
-            chains.clone(),
-            config.clone(),
+    run_api(opts.addr, chains.clone(), metrics_registry.clone(), rx_exit).await?;
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn setup_chain_and_run_keeper(
+    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,
-            metrics_registry.clone(),
+            chain_config,
+            state,
+            keeper_metrics.clone(),
             rpc_metrics.clone(),
-        ));
-    } else {
-        tracing::info!("Not starting keeper service: no keeper private key specified. Please add one to the config if you would like to run the keeper service.")
+        )
+        .await?;
     }
-
-    // Spawn a thread to track latest block lag. This helps us know if the rpc is up and updated with the latest block.
-    spawn(track_block_timestamp_lag(
-        config,
-        metrics_registry.clone(),
-        rpc_metrics.clone(),
-    ));
-
-    run_api(opts.addr, chains, metrics_registry, rx_exit).await?;
-
     Ok(())
 }
 
@@ -267,7 +244,7 @@ async fn setup_chain_state(
         let offset = commitment.original_commitment_sequence_number.try_into()?;
         offsets.push(offset);
 
-        let pebble_hash_chain = PebbleHashChain::from_config(
+        let pebble_hash_chain = PebbleHashChain::from_config_async(
             secret,
             chain_id,
             provider,
@@ -276,6 +253,7 @@ async fn setup_chain_state(
             commitment.chain_length,
             chain_sample_interval,
         )
+        .await
         .map_err(|e| anyhow!("Failed to create hash chain: {}", e))?;
         hash_chains.push(pebble_hash_chain);
     }
@@ -308,74 +286,3 @@ async fn setup_chain_state(
 pub struct ChainLabel {
     pub chain_id: String,
 }
-
-#[tracing::instrument(name = "block_timestamp_lag", skip_all, fields(chain_id = chain_id))]
-pub async fn check_block_timestamp_lag(
-    chain_id: String,
-    chain_config: EthereumConfig,
-    metrics: Family<ChainLabel, Gauge>,
-    rpc_metrics: Arc<RpcMetrics>,
-) {
-    let provider =
-        match TracedClient::new(chain_id.clone(), &chain_config.geth_rpc_addr, rpc_metrics) {
-            Ok(r) => r,
-            Err(e) => {
-                tracing::error!("Failed to create provider for chain id - {:?}", e);
-                return;
-            }
-        };
-
-    const INF_LAG: i64 = 1000000; // value that definitely triggers an alert
-    let lag = match provider.get_block(BlockNumber::Latest).await {
-        Ok(block) => match block {
-            Some(block) => {
-                let block_timestamp = block.timestamp;
-                let server_timestamp = SystemTime::now()
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs();
-                let lag: i64 = (server_timestamp as i64) - (block_timestamp.as_u64() as i64);
-                lag
-            }
-            None => {
-                tracing::error!("Block is None");
-                INF_LAG
-            }
-        },
-        Err(e) => {
-            tracing::error!("Failed to get block - {:?}", e);
-            INF_LAG
-        }
-    };
-    metrics
-        .get_or_create(&ChainLabel {
-            chain_id: chain_id.clone(),
-        })
-        .set(lag);
-}
-
-/// Tracks the difference between the server timestamp and the latest block timestamp for each chain
-pub async fn track_block_timestamp_lag(
-    config: Config,
-    metrics_registry: Arc<RwLock<Registry>>,
-    rpc_metrics: Arc<RpcMetrics>,
-) {
-    let metrics = Family::<ChainLabel, Gauge>::default();
-    metrics_registry.write().await.register(
-        "block_timestamp_lag",
-        "The difference between server timestamp and latest block timestamp",
-        metrics.clone(),
-    );
-    loop {
-        for (chain_id, chain_config) in &config.chains {
-            spawn(check_block_timestamp_lag(
-                chain_id.clone(),
-                chain_config.clone(),
-                metrics.clone(),
-                rpc_metrics.clone(),
-            ));
-        }
-
-        time::sleep(TRACK_INTERVAL).await;
-    }
-}

apps/fortuna/src/command/setup_provider.rs

@@ -112,7 +112,7 @@ async fn setup_chain_provider(
             );
             register = true;
         } else {
-            let hash_chain = PebbleHashChain::from_config(
+            let hash_chain = PebbleHashChain::from_config_async(
                 &secret,
                 chain_id,
                 &provider_address,
@@ -120,7 +120,8 @@ async fn setup_chain_provider(
                 &metadata.seed,
                 provider_config.chain_length,
                 provider_config.chain_sample_interval,
-            )?;
+            )
+            .await?;
             let chain_state = HashChainState {
                 offsets: vec![provider_info
                     .original_commitment_sequence_number

apps/fortuna/src/keeper.rs

@@ -1,3 +1,4 @@
+use crate::keeper::track::track_block_timestamp_lag;
 use {
     crate::{
         api::{BlockchainState, ChainId},
@@ -59,10 +60,10 @@ pub async fn run_keeper_threads(
     chain_state: BlockchainState,
     metrics: Arc<KeeperMetrics>,
     rpc_metrics: Arc<RpcMetrics>,
-) {
-    tracing::info!("starting keeper");
+) -> 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);
+    tracing::info!("Latest safe block: {}", &latest_safe_block);
 
     let contract = Arc::new(
         InstrumentedSignablePythContract::from_config(
@@ -71,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();
 
@@ -215,10 +215,19 @@ pub async fn run_keeper_threads(
                     )
                     .in_current_span(),
                 );
+                spawn(
+                    track_block_timestamp_lag(
+                        chain_id.clone(),
+                        contract.client(),
+                        keeper_metrics.clone(),
+                    )
+                    .in_current_span(),
+                );
 
                 time::sleep(TRACK_INTERVAL).await;
             }
         }
         .in_current_span(),
     );
+    Ok(())
 }

apps/fortuna/src/keeper/keeper_metrics.rs

@@ -43,6 +43,7 @@ pub struct KeeperMetrics {
     pub final_fee_multiplier: Family<AccountLabel, Histogram>,
     pub gas_price_estimate: Family<AccountLabel, Gauge<f64, AtomicU64>>,
     pub accrued_pyth_fees: Family<ChainIdLabel, Gauge<f64, AtomicU64>>,
+    pub block_timestamp_lag: Family<ChainIdLabel, Gauge>,
 }
 
 impl Default for KeeperMetrics {
@@ -85,15 +86,13 @@ impl Default for KeeperMetrics {
             }),
             gas_price_estimate: Family::default(),
             accrued_pyth_fees: Family::default(),
+            block_timestamp_lag: Family::default(),
         }
     }
 }
 
 impl KeeperMetrics {
-    pub async fn new(
-        registry: Arc<RwLock<Registry>>,
-        chain_labels: Vec<(String, Address)>,
-    ) -> Self {
+    pub async fn new(registry: Arc<RwLock<Registry>>) -> Self {
         let mut writable_registry = registry.write().await;
         let keeper_metrics = KeeperMetrics::default();
 
@@ -223,73 +222,56 @@ impl KeeperMetrics {
             keeper_metrics.accrued_pyth_fees.clone(),
         );
 
+        writable_registry.register(
+            "block_timestamp_lag",
+            "The difference between server timestamp and latest block timestamp",
+            keeper_metrics.block_timestamp_lag.clone(),
+        );
+
         // *Important*: When adding a new metric:
         // 1. Register it above using `writable_registry.register(...)`
-        // 2. Add a get_or_create call in the loop below to initialize it for each chain/provider pair
-
-        // Initialize accrued_pyth_fees for each chain_id
-        for (chain_id, _) in chain_labels.iter() {
-            let _ = keeper_metrics
-                .accrued_pyth_fees
-                .get_or_create(&ChainIdLabel {
-                    chain_id: chain_id.clone(),
-                });
-        }
-
-        for (chain_id, provider_address) in chain_labels {
-            let account_label = AccountLabel {
-                chain_id,
-                address: provider_address.to_string(),
-            };
-
-            let _ = keeper_metrics
-                .current_sequence_number
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .current_commitment_sequence_number
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .end_sequence_number
-                .get_or_create(&account_label);
-            let _ = keeper_metrics.balance.get_or_create(&account_label);
-            let _ = keeper_metrics.collected_fee.get_or_create(&account_label);
-            let _ = keeper_metrics.current_fee.get_or_create(&account_label);
-            let _ = keeper_metrics
-                .target_provider_fee
-                .get_or_create(&account_label);
-            let _ = keeper_metrics.total_gas_spent.get_or_create(&account_label);
-            let _ = keeper_metrics
-                .total_gas_fee_spent
-                .get_or_create(&account_label);
-            let _ = keeper_metrics.requests.get_or_create(&account_label);
-            let _ = keeper_metrics
-                .requests_processed
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .requests_processed_success
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .requests_processed_failure
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .requests_reprocessed
-                .get_or_create(&account_label);
-            let _ = keeper_metrics.reveals.get_or_create(&account_label);
-            let _ = keeper_metrics
-                .request_duration_ms
-                .get_or_create(&account_label);
-            let _ = keeper_metrics.retry_count.get_or_create(&account_label);
-            let _ = keeper_metrics
-                .final_gas_multiplier
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .final_fee_multiplier
-                .get_or_create(&account_label);
-            let _ = keeper_metrics
-                .gas_price_estimate
-                .get_or_create(&account_label);
-        }
+        // 2. Add a get_or_create call in the add_chain function below to initialize it for each chain/provider pair
 
         keeper_metrics
     }
+
+    pub fn add_chain(&self, chain_id: String, provider_address: Address) {
+        let chain_id_label = ChainIdLabel {
+            chain_id: chain_id.clone(),
+        };
+        let _ = self.accrued_pyth_fees.get_or_create(&chain_id_label);
+        let _ = self.block_timestamp_lag.get_or_create(&chain_id_label);
+
+        let account_label = AccountLabel {
+            chain_id,
+            address: provider_address.to_string(),
+        };
+
+        let _ = self.current_sequence_number.get_or_create(&account_label);
+        let _ = self
+            .current_commitment_sequence_number
+            .get_or_create(&account_label);
+        let _ = self.end_sequence_number.get_or_create(&account_label);
+        let _ = self.balance.get_or_create(&account_label);
+        let _ = self.collected_fee.get_or_create(&account_label);
+        let _ = self.current_fee.get_or_create(&account_label);
+        let _ = self.target_provider_fee.get_or_create(&account_label);
+        let _ = self.total_gas_spent.get_or_create(&account_label);
+        let _ = self.total_gas_fee_spent.get_or_create(&account_label);
+        let _ = self.requests.get_or_create(&account_label);
+        let _ = self.requests_processed.get_or_create(&account_label);
+        let _ = self
+            .requests_processed_success
+            .get_or_create(&account_label);
+        let _ = self
+            .requests_processed_failure
+            .get_or_create(&account_label);
+        let _ = self.requests_reprocessed.get_or_create(&account_label);
+        let _ = self.reveals.get_or_create(&account_label);
+        let _ = self.request_duration_ms.get_or_create(&account_label);
+        let _ = self.retry_count.get_or_create(&account_label);
+        let _ = self.final_gas_multiplier.get_or_create(&account_label);
+        let _ = self.final_fee_multiplier.get_or_create(&account_label);
+        let _ = self.gas_price_estimate.get_or_create(&account_label);
+    }
 }

apps/fortuna/src/keeper/track.rs

@@ -5,8 +5,11 @@ use {
         eth_utils::traced_client::TracedClient,
     },
     ethers::middleware::Middleware,
-    ethers::{providers::Provider, types::Address},
-    std::sync::Arc,
+    ethers::{prelude::BlockNumber, providers::Provider, types::Address},
+    std::{
+        sync::Arc,
+        time::{SystemTime, UNIX_EPOCH},
+    },
     tracing,
 };
 
@@ -41,6 +44,43 @@ pub async fn track_balance(
         .set(balance);
 }
 
+/// Tracks the difference between the server timestamp and the latest block timestamp for each chain
+#[tracing::instrument(skip_all)]
+pub async fn track_block_timestamp_lag(
+    chain_id: String,
+    provider: Arc<Provider<TracedClient>>,
+    metrics: Arc<KeeperMetrics>,
+) {
+    const INF_LAG: i64 = 1000000; // value that definitely triggers an alert
+    let lag = match provider.get_block(BlockNumber::Latest).await {
+        Ok(block) => match block {
+            Some(block) => {
+                let block_timestamp = block.timestamp;
+                let server_timestamp = SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs();
+                let lag: i64 = (server_timestamp as i64) - (block_timestamp.as_u64() as i64);
+                lag
+            }
+            None => {
+                tracing::error!("Block is None");
+                INF_LAG
+            }
+        },
+        Err(e) => {
+            tracing::error!("Failed to get block - {:?}", e);
+            INF_LAG
+        }
+    };
+    metrics
+        .block_timestamp_lag
+        .get_or_create(&ChainIdLabel {
+            chain_id: chain_id.clone(),
+        })
+        .set(lag);
+}
+
 /// tracks the collected fees and the hashchain data of the given provider address on the given chain
 /// if there is a error the function will just return
 #[tracing::instrument(skip_all)]

apps/fortuna/src/state.rs

@@ -3,6 +3,7 @@ use {
     anyhow::{ensure, Result},
     ethers::types::Address,
     sha3::{Digest, Keccak256},
+    tokio::task::spawn_blocking,
 };
 
 /// A hash chain of a specific length. The hash chain has the property that
@@ -42,23 +43,34 @@ impl PebbleHashChain {
         }
     }
 
-    pub fn from_config(
+    fn generate_secret(
         secret: &str,
         chain_id: &ChainId,
         provider_address: &Address,
         contract_address: &Address,
         random: &[u8; 32],
-        chain_length: u64,
-        sample_interval: u64,
-    ) -> Result<Self> {
+    ) -> Result<[u8; 32]> {
         let mut input: Vec<u8> = vec![];
         input.extend_from_slice(&hex::decode(secret.trim())?);
         input.extend_from_slice(chain_id.as_bytes());
         input.extend_from_slice(provider_address.as_bytes());
         input.extend_from_slice(contract_address.as_bytes());
         input.extend_from_slice(random);
-
         let secret: [u8; 32] = Keccak256::digest(input).into();
+        Ok(secret)
+    }
+
+    pub fn from_config(
+        secret: &str,
+        chain_id: &ChainId,
+        provider_address: &Address,
+        contract_address: &Address,
+        random: &[u8; 32],
+        chain_length: u64,
+        sample_interval: u64,
+    ) -> Result<Self> {
+        let secret: [u8; 32] =
+            Self::generate_secret(secret, chain_id, provider_address, contract_address, random)?;
         Ok(Self::new(
             secret,
             chain_length.try_into()?,
@@ -66,6 +78,27 @@ impl PebbleHashChain {
         ))
     }
 
+    /// Asynchronous version of `from_config` that runs the computation in a blocking thread.
+    pub async fn from_config_async(
+        secret: &str,
+        chain_id: &ChainId,
+        provider_address: &Address,
+        contract_address: &Address,
+        random: &[u8; 32],
+        chain_length: u64,
+        sample_interval: u64,
+    ) -> Result<Self> {
+        let secret: [u8; 32] =
+            Self::generate_secret(secret, chain_id, provider_address, contract_address, random)?;
+        let chain_length: usize = chain_length.try_into()?;
+        let sample_interval: usize = sample_interval.try_into()?;
+        let hash_chain = spawn_blocking(move || Self::new(secret, chain_length, sample_interval))
+            .await
+            .expect("Failed to make hash chain");
+
+        Ok(hash_chain)
+    }
+
     pub fn reveal_ith(&self, i: usize) -> Result<[u8; 32]> {
         ensure!(i < self.len(), "index not in range");