feat(fortuna): implement multiple replica support with sequence number modulo filtering

Co-Authored-By: Tejas Badadare <[email protected]>

Author: devin-ai-integration[bot]

apps/fortuna/config.sample.yaml

@@ -47,6 +47,8 @@ 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

apps/fortuna/src/config.rs

@@ -94,6 +94,15 @@ impl Config {
             }
         }
 
+        if let Some(replica_config) = &config.keeper.replica_config {
+            if replica_config.total_replicas == 0 {
+                return Err(anyhow!("Keeper replica configuration is invalid. total_replicas must be greater than 0."));
+            }
+            if replica_config.replica_id >= replica_config.total_replicas {
+                return Err(anyhow!("Keeper replica configuration is invalid. replica_id must be less than total_replicas."));
+            }
+        }
+
         Ok(config)
     }
 
@@ -333,6 +342,12 @@ fn default_chain_sample_interval() -> u64 {
     1
 }
 
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct ReplicaConfig {
+    pub replica_id: u64,
+    pub total_replicas: u64,
+}
+
 /// Configuration values for the keeper service that are shared across chains.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct KeeperConfig {
@@ -342,6 +357,9 @@ pub struct KeeperConfig {
     /// This key *does not need to be a registered provider*. In particular, production deployments
     /// should ensure this is a different key in order to reduce the severity of security breaches.
     pub private_key: SecretString,
+
+    #[serde(default)]
+    pub replica_config: Option<ReplicaConfig>,
 }
 
 // A secret is a string that can be provided either as a literal in the config,

apps/fortuna/src/keeper.rs

@@ -85,6 +85,13 @@ 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,
+        },
         metrics: metrics.clone(),
         fulfilled_requests_cache,
         history,

apps/fortuna/src/keeper/block.rs

@@ -45,6 +45,7 @@ pub struct ProcessParams {
     pub gas_limit: U256,
     pub escalation_policy: EscalationPolicy,
     pub chain_state: BlockchainState,
+    pub keeper_config: crate::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,54 @@ pub async fn process_event_with_backoff(
         return Ok(());
     }
 
+    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;
+            if assigned_replica != replica_config.replica_id {
+                tracing::debug!(
+                    sequence_number = event.sequence_number,
+                    assigned_replica = assigned_replica,
+                    our_replica_id = replica_config.replica_id,
+                    "Processing request as backup replica"
+                );
+                false
+            } else {
+                true
+            }
+        } else {
+            true // No replica config, process all requests
+        };
+
+    if !is_primary_replica {
+        match chain_state
+            .contract
+            .get_request(event.provider_address, event.sequence_number)
+            .await
+        {
+            Ok(Some(_)) => {
+                tracing::info!(
+                    sequence_number = event.sequence_number,
+                    "Request still open, processing as backup replica"
+                );
+            }
+            Ok(None) => {
+                tracing::debug!(
+                    sequence_number = event.sequence_number,
+                    "Request already fulfilled by primary replica, skipping"
+                );
+                return Ok(());
+            }
+            Err(e) => {
+                tracing::warn!(
+                    sequence_number = event.sequence_number,
+                    error = ?e,
+                    "Error checking request status, skipping"
+                );
+                return Ok(());
+            }
+        }
+    }
+
     let account_label = AccountLabel {
         chain_id: chain_state.id.clone(),
         address: chain_state.provider_address.to_string(),