refactor(fortuna): streamline keeper configuration and improve event processing

- Updated config.sample.yaml by removing unnecessary blank lines.
- Changed backup_delay_seconds in README.md from 45 to 30 for consistency.
- Refactored run_keeper_threads to accept KeeperConfig directly instead of private_key.
- Enhanced run function to handle keeper configuration more effectively.
- Added comments in process_event_with_backoff to clarify primary and backup replica logic.

Author: tejasbadadare

apps/fortuna/README.md

@@ -17,6 +17,11 @@ a database to be available at build time. Create a `.env` file in the root of th
 DATABASE_URL="sqlite:fortuna.db?mode=rwc"
 ```
 
+Install sqlx for cargo with:
+```bash
+cargo install sqlx
+```
+
 Next, you need to create the database and apply the schema migrations. You can do this by running:
 
 ```bash
@@ -62,7 +67,7 @@ keeper:
     total_replicas: 2
     backup_delay_seconds: 30
 
-# Replica 1 (Green) - handles odd sequence numbers (1, 3, 5, ...)  
+# Replica 1 (Green) - handles odd sequence numbers (1, 3, 5, ...)
 keeper:
   replica_config:
     replica_id: 1
@@ -77,7 +82,7 @@ keeper:
   replica_config:
     replica_id: 0
     total_replicas: 3
-    backup_delay_seconds: 45
+    backup_delay_seconds: 30
 ```
 
 ### Deployment Considerations

apps/fortuna/config.sample.yaml

@@ -47,8 +47,6 @@ chains:
     # blocks after 5 blocks, then again after 10 blocks, and finally after 20 blocks.
     block_delays: [5, 10, 20]
 
-
-
     # Historical commitments -- delete this block for local development purposes
     commitments:
       # prettier-ignore
@@ -88,8 +86,8 @@ keeper:
     value: 0xabcd
     # For production, you can store the private key in a file.
     # file: keeper-key.txt
-# Multi-replica configuration documentation
 
+  # Multi-replica configuration
   # Optional: Multi-replica configuration for high availability and load distribution
   # Uncomment and configure for production deployments with multiple Fortuna instances
   # replica_config:
@@ -98,7 +96,7 @@ keeper:
   #   backup_delay_seconds: 30   # Seconds to wait before processing other replicas' requests
   #
   # Example configurations:
-  # 
+  #
   # Two-replica setup (Blue/Green):
   # - Replica 0: handles even sequence numbers (0, 2, 4, ...)
   # - Replica 1: handles odd sequence numbers (1, 3, 5, ...)

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, ProviderConfig, RunOptions},
+        config::{Commitment, Config, EthereumConfig, KeeperConfig, ProviderConfig, RunOptions},
         eth_utils::traced_client::RpcMetrics,
         history::History,
         keeper::{self, keeper_metrics::KeeperMetrics},
@@ -98,6 +98,11 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
     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 keeper_config_option = if keeper_private_key_option.is_some() {
+        Some(config.keeper.clone())
+    } else {
+        None
+    };
     let chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>> = Arc::new(RwLock::new(
         config
             .chains
@@ -109,7 +114,7 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
     for (chain_id, chain_config) in config.chains.clone() {
         keeper_metrics.add_chain(chain_id.clone(), config.provider.address);
         let keeper_metrics = keeper_metrics.clone();
-        let keeper_private_key_option = keeper_private_key_option.clone();
+        let keeper_config_option = keeper_config_option.clone();
         let chains = chains.clone();
         let secret_copy = secret.clone();
         let rpc_metrics = rpc_metrics.clone();
@@ -122,7 +127,7 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
                     &chain_id,
                     chain_config.clone(),
                     keeper_metrics.clone(),
-                    keeper_private_key_option.clone(),
+                    keeper_config_option.clone(),
                     chains.clone(),
                     &secret_copy,
                     history.clone(),
@@ -172,7 +177,7 @@ async fn setup_chain_and_run_keeper(
     chain_id: &ChainId,
     chain_config: EthereumConfig,
     keeper_metrics: Arc<KeeperMetrics>,
-    keeper_private_key_option: Option<String>,
+    keeper_config: Option<KeeperConfig>,
     chains: Arc<RwLock<HashMap<ChainId, ApiBlockChainState>>>,
     secret_copy: &str,
     history: Arc<History>,
@@ -192,9 +197,9 @@ async fn setup_chain_and_run_keeper(
         chain_id.clone(),
         ApiBlockChainState::Initialized(state.clone()),
     );
-    if let Some(keeper_private_key) = keeper_private_key_option {
+    if let Some(keeper_config) = keeper_config {
         keeper::run_keeper_threads(
-            keeper_private_key,
+            keeper_config,
             chain_config,
             state,
             keeper_metrics.clone(),

apps/fortuna/src/keeper.rs

@@ -56,7 +56,7 @@ pub enum RequestState {
 /// handle any events for the new blocks.
 #[tracing::instrument(name = "keeper", skip_all, fields(chain_id = chain_state.id))]
 pub async fn run_keeper_threads(
-    private_key: String,
+    keeper_config: crate::config::KeeperConfig,
     chain_eth_config: EthereumConfig,
     chain_state: BlockchainState,
     metrics: Arc<KeeperMetrics>,
@@ -67,6 +67,11 @@ pub async fn run_keeper_threads(
     let latest_safe_block = get_latest_safe_block(&chain_state).in_current_span().await;
     tracing::info!("Latest safe block: {}", &latest_safe_block);
 
+    let private_key = keeper_config
+        .private_key
+        .load()?
+        .ok_or(anyhow::anyhow!("Keeper private key must be provided"))?;
+
     let contract = Arc::new(InstrumentedSignablePythContract::from_config(
         &chain_eth_config,
         &private_key,
@@ -85,13 +90,7 @@ pub async fn run_keeper_threads(
         contract: contract.clone(),
         gas_limit,
         escalation_policy: chain_eth_config.escalation_policy.to_policy(),
-        keeper_config: crate::config::KeeperConfig {
-            private_key: crate::config::SecretString {
-                value: None,
-                file: None,
-            },
-            replica_config: None,
-        },
+        keeper_config,
         metrics: metrics.clone(),
         fulfilled_requests_cache,
         history,

apps/fortuna/src/keeper/block.rs

@@ -2,6 +2,7 @@ use {
     crate::{
         api::BlockchainState,
         chain::{ethereum::InstrumentedSignablePythContract, reader::BlockNumber},
+        config::KeeperConfig,
         eth_utils::utils::EscalationPolicy,
         history::History,
         keeper::{
@@ -45,7 +46,7 @@ pub struct ProcessParams {
     pub gas_limit: U256,
     pub escalation_policy: EscalationPolicy,
     pub chain_state: BlockchainState,
-    pub keeper_config: crate::config::KeeperConfig,
+    pub keeper_config: KeeperConfig,
     pub metrics: Arc<KeeperMetrics>,
     pub history: Arc<History>,
     pub fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,

apps/fortuna/src/keeper/process_event.rs

@@ -35,6 +35,8 @@ pub async fn process_event_with_backoff(
         return Ok(());
     }
 
+    // Check if we are the primary replica for this request.
+    // If not, we will wait for a delay before processing the request.
     let is_primary_replica =
         if let Some(replica_config) = &process_param.keeper_config.replica_config {
             let assigned_replica = event.sequence_number % replica_config.total_replicas;
@@ -66,6 +68,8 @@ pub async fn process_event_with_backoff(
             .await;
         }
 
+        // Check if the request is still open after the delay.
+        // If it is, we will process it as a backup replica.
         match chain_state
             .contract
             .get_request(event.provider_address, event.sequence_number)
@@ -88,9 +92,8 @@ pub async fn process_event_with_backoff(
                 tracing::warn!(
                     sequence_number = event.sequence_number,
                     error = ?e,
-                    "Error checking request status after delay, skipping"
+                    "Error checking request status after delay, processing as backup replica"
                 );
-                return Ok(());
             }
         }
     }