refactor: move block delays to EthereumConfig and support multiple de…

…lays

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

apps/fortuna/config.sample.yaml

@@ -39,6 +39,11 @@ chains:
     target_profit_pct: 20
     max_profit_pct: 100
 
+    # A list of block delays for processing blocks multiple times. Each number represents
+    # how many blocks to wait before processing. For example, [5, 10, 20] means process
+    # 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
@@ -78,5 +83,3 @@ 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/src/config.rs

@@ -177,6 +177,16 @@ pub struct EthereumConfig {
     /// Maximum number of hashes to record in a request.
     /// This should be set according to the maximum gas limit the provider supports for callbacks.
     pub max_num_hashes: Option<u32>,
+
+    /// A list of delays (in blocks) that indicates how many blocks should be delayed
+    /// before we process a block. For retry logic, we can process blocks multiple times
+    /// at each specified delay. For example: [5, 10, 20].
+    #[serde(default = "default_block_delays")]
+    pub block_delays: Vec<u64>,
+}
+
+fn default_block_delays() -> Vec<u64> {
+    vec![5]
 }
 
 fn default_priority_fee_multiplier_pct() -> u64 {
@@ -341,14 +351,6 @@ 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/keeper.rs

@@ -250,7 +250,7 @@ impl KeeperMetrics {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BlockRange {
     pub from: BlockNumber,
     pub to: BlockNumber,
@@ -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,57 +346,18 @@ pub async fn run_keeper_threads(
         )
         .in_current_span(),
     );
-    // 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 a thread for block processing with configured delays
     spawn(
         process_new_blocks(
             chain_state.clone(),
-            rx_immediate,
+            rx,
             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,
+            chain_eth_config.clone(),
         )
         .in_current_span(),
     );
@@ -1006,8 +967,10 @@ pub async fn watch_blocks(
     }
 }
 
-/// It waits on rx channel to receive block ranges and then calls process_block_range to process them.
+/// It waits on rx channel to receive block ranges and then calls process_block_range to process them
+/// for each configured block delay.
 #[tracing::instrument(skip_all)]
+#[allow(clippy::too_many_arguments)]
 pub async fn process_new_blocks(
     chain_state: BlockchainState,
     mut rx: mpsc::Receiver<BlockRange>,
@@ -1016,12 +979,14 @@ pub async fn process_new_blocks(
     escalation_policy: EscalationPolicyConfig,
     metrics: Arc<KeeperMetrics>,
     fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
+    chain_eth_config: EthereumConfig,
 ) {
     tracing::info!("Waiting for new block ranges to process");
     loop {
         if let Some(block_range) = rx.recv().await {
+            // Process blocks immediately first
             process_block_range(
-                block_range,
+                block_range.clone(),
                 Arc::clone(&contract),
                 gas_limit,
                 escalation_policy.clone(),
@@ -1031,44 +996,25 @@ pub async fn process_new_blocks(
             )
             .in_current_span()
             .await;
-        }
-    }
-}
 
-#[tracing::instrument(skip_all)]
-#[allow(clippy::too_many_arguments)]
-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;
+            // Then process with each configured delay
+            for delay in &chain_eth_config.block_delays {
+                let adjusted_range = BlockRange {
+                    from: block_range.from + delay,
+                    to: block_range.to + delay,
+                };
+                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;
+            }
         }
     }
 }