feat: add delayed block processing and bump minor version

Co-Authored-By: Jayant Krishnamurthy <[email protected]>

Author: devin-ai-integration[bot]

apps/fortuna/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "fortuna"
-version = "7.3.0"
+version = "7.4.0"
 edition = "2021"
 
 [dependencies]

apps/fortuna/Cargo.toml

@@ -78,3 +78,6 @@ keeper:
     value: 0xabcd
     # For production, you can store the private key in a file.
     # file: keeper-key.txt
+  
+  # Number of blocks to wait before processing new blocks (in addition to reveal_delay_blocks)
+  delay_blocks: 5

apps/fortuna/config.sample.yaml

@@ -341,6 +341,14 @@ 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,
+
+    /// Number of blocks to wait before processing new blocks (in addition to reveal_delay_blocks)
+    #[serde(default = "default_delay_blocks")]
+    pub delay_blocks: u64,
+}
+
+fn default_delay_blocks() -> u64 {
+    5
 }
 
 // A secret is a string that can be provided either as a literal in the config,

apps/fortuna/src/config.rs

@@ -335,7 +335,7 @@ pub async fn run_keeper_threads(
         .in_current_span(),
     );
 
-    let (tx, rx) = mpsc::channel::<BlockRange>(1000);
+    let (tx, _rx) = mpsc::channel::<BlockRange>(1000);
     // Spawn a thread to watch for new blocks and send the range of blocks for which events has not been handled to the `tx` channel.
     spawn(
         watch_blocks_wrapper(
@@ -346,16 +346,57 @@ pub async fn run_keeper_threads(
         )
         .in_current_span(),
     );
-    // Spawn a thread that listens for block ranges on the `rx` channel and processes the events for those blocks.
+    // Create channels for both immediate and delayed block processing
+    let (tx_immediate, rx_immediate) = mpsc::channel::<BlockRange>(1000);
+    let (tx_delayed, rx_delayed) = mpsc::channel::<BlockRange>(1000);
+
+    // Spawn a thread to watch for new blocks and send the range of blocks to both channels
+    spawn(
+        watch_blocks_wrapper(
+            chain_state.clone(),
+            latest_safe_block,
+            tx_immediate,
+            chain_eth_config.geth_rpc_wss.clone(),
+        )
+        .in_current_span(),
+    );
+
+    // Clone the tx_delayed channel for the second watch_blocks_wrapper
+    spawn(
+        watch_blocks_wrapper(
+            chain_state.clone(),
+            latest_safe_block,
+            tx_delayed,
+            chain_eth_config.geth_rpc_wss.clone(),
+        )
+        .in_current_span(),
+    );
+
+    // Spawn a thread for immediate block processing
     spawn(
         process_new_blocks(
             chain_state.clone(),
-            rx,
+            rx_immediate,
+            Arc::clone(&contract),
+            gas_limit,
+            chain_eth_config.escalation_policy.clone(),
+            metrics.clone(),
+            fulfilled_requests_cache.clone(),
+        )
+        .in_current_span(),
+    );
+
+    // Spawn a thread for delayed block processing
+    spawn(
+        process_new_blocks_delayed(
+            chain_state.clone(),
+            rx_delayed,
             Arc::clone(&contract),
             gas_limit,
             chain_eth_config.escalation_policy.clone(),
             metrics.clone(),
             fulfilled_requests_cache.clone(),
+            5,
         )
         .in_current_span(),
     );
@@ -994,6 +1035,43 @@ pub async fn process_new_blocks(
     }
 }
 
+#[tracing::instrument(skip_all)]
+pub async fn process_new_blocks_delayed(
+    chain_state: BlockchainState,
+    mut rx: mpsc::Receiver<BlockRange>,
+    contract: Arc<InstrumentedSignablePythContract>,
+    gas_limit: U256,
+    escalation_policy: EscalationPolicyConfig,
+    metrics: Arc<KeeperMetrics>,
+    fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
+    delay_blocks: u64,
+) {
+    tracing::info!(
+        "Waiting for new block ranges to process with {} block delay",
+        delay_blocks
+    );
+    loop {
+        if let Some(block_range) = rx.recv().await {
+            let from_block_after_delay = block_range.from + delay_blocks;
+            let adjusted_range = BlockRange {
+                from: from_block_after_delay,
+                to: block_range.to + delay_blocks,
+            };
+            process_block_range(
+                adjusted_range,
+                Arc::clone(&contract),
+                gas_limit,
+                escalation_policy.clone(),
+                chain_state.clone(),
+                metrics.clone(),
+                fulfilled_requests_cache.clone(),
+            )
+            .in_current_span()
+            .await;
+        }
+    }
+}
+
 /// Processes the backlog_range for a chain.
 #[tracing::instrument(skip_all)]
 pub async fn process_backlog(