Use history in keeper

m30m · m30m · commit fcc9f9dd2bf4 · 2025-05-01T17:14:59.000-07:00
diff --git a/apps/fortuna/src/api.rs b/apps/fortuna/src/api.rs
@@ -26,7 +26,7 @@ use {
     tokio::sync::RwLock,
     url::Url,
 };
-pub use {chain_ids::*, index::*, live::*, metrics::*, ready::*, revelation::*, explorer::*};
+pub use {chain_ids::*, explorer::*, index::*, live::*, metrics::*, ready::*, revelation::*};
 
 mod chain_ids;
 mod explorer;
@@ -72,6 +72,15 @@ pub struct TimedJournalLog {
     pub log: JournalLog,
 }
 
+impl TimedJournalLog {
+    pub fn with_current_time(log: JournalLog) -> Self {
+        TimedJournalLog {
+            timestamp: chrono::Utc::now(),
+            log,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, ToSchema)]
 pub struct RequestJournal {
     pub chain_id: ChainId,
@@ -161,9 +170,9 @@ impl History {
                         min_timestamp.unwrap_or(DateTime::<chrono::Utc>::MIN_UTC),
                     )
                         ..(
-                        chain_id.clone(),
-                        max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC),
-                    ),
+                            chain_id.clone(),
+                            max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC),
+                        ),
                 );
                 range
                     .rev()
@@ -407,7 +416,12 @@ mod test {
             ApiBlockChainState::Initialized(avax_state),
         );
 
-        let api_state = ApiState::new(Arc::new(RwLock::new(chains)), metrics_registry, Default::default()).await;
+        let api_state = ApiState::new(
+            Arc::new(RwLock::new(chains)),
+            metrics_registry,
+            Default::default(),
+        )
+        .await;
 
         let app = api::routes(api_state);
         (TestServer::new(app).unwrap(), eth_read, avax_read)
diff --git a/apps/fortuna/src/api/explorer.rs b/apps/fortuna/src/api/explorer.rs
@@ -1,7 +1,4 @@
-use crate::api::{
-    ChainId, RequestJournal, RestError
-    ,
-};
+use crate::api::{ChainId, RequestJournal, RestError};
 use axum::extract::{Query, State};
 use axum::Json;
 use ethers::types::TxHash;
diff --git a/apps/fortuna/src/chain/ethereum.rs b/apps/fortuna/src/chain/ethereum.rs
@@ -1,3 +1,4 @@
+use ethers::contract::LogMeta;
 use {
     crate::{
         api::ChainId,
@@ -285,14 +286,15 @@ impl<T: JsonRpcClient + 'static> EntropyReader for PythRandom<Provider<T>> {
             .to_block(to_block)
             .topic1(provider);
 
-        let res: Vec<RequestedWithCallbackFilter> = event.query().await?;
+        let res: Vec<(RequestedWithCallbackFilter, LogMeta)> = event.query_with_meta().await?;
 
         Ok(res
             .iter()
-            .map(|r| RequestedWithCallbackEvent {
+            .map(|(r, meta)| RequestedWithCallbackEvent {
                 sequence_number: r.sequence_number,
                 user_random_number: r.user_random_number,
                 provider_address: r.request.provider,
+                tx_hash: meta.transaction_hash,
             })
             .filter(|r| r.provider_address == provider)
             .collect())
diff --git a/apps/fortuna/src/chain/reader.rs b/apps/fortuna/src/chain/reader.rs
@@ -1,3 +1,4 @@
+use ethers::types::TxHash;
 use {
     anyhow::Result,
     axum::async_trait,
@@ -34,6 +35,7 @@ pub struct RequestedWithCallbackEvent {
     pub sequence_number: u64,
     pub user_random_number: [u8; 32],
     pub provider_address: Address,
+    pub tx_hash: TxHash,
 }
 
 /// EntropyReader is the read-only interface of the Entropy contract.
diff --git a/apps/fortuna/src/keeper.rs b/apps/fortuna/src/keeper.rs
@@ -1,4 +1,5 @@
 use crate::api::History;
+use crate::keeper::block::ProcessParams;
 use crate::keeper::track::track_block_timestamp_lag;
 use {
     crate::{
@@ -60,7 +61,7 @@ pub async fn run_keeper_threads(
     chain_eth_config: EthereumConfig,
     chain_state: BlockchainState,
     metrics: Arc<KeeperMetrics>,
-    _history: Arc<RwLock<History>>,
+    history: Arc<RwLock<History>>,
     rpc_metrics: Arc<RpcMetrics>,
 ) -> anyhow::Result<()> {
     tracing::info!("Starting keeper");
@@ -82,18 +83,22 @@ pub async fn run_keeper_threads(
 
     // Spawn a thread to handle the events from last backlog_range blocks.
     let gas_limit: U256 = chain_eth_config.gas_limit.into();
+    let process_params = ProcessParams {
+        chain_state: chain_state.clone(),
+        contract: contract.clone(),
+        gas_limit,
+        escalation_policy: chain_eth_config.escalation_policy.to_policy(),
+        metrics: metrics.clone(),
+        fulfilled_requests_cache,
+        history,
+    };
     spawn(
         process_backlog(
+            process_params.clone(),
             BlockRange {
                 from: latest_safe_block.saturating_sub(chain_eth_config.backlog_range),
                 to: latest_safe_block,
             },
-            contract.clone(),
-            gas_limit,
-            chain_eth_config.escalation_policy.to_policy(),
-            chain_state.clone(),
-            metrics.clone(),
-            fulfilled_requests_cache.clone(),
             chain_eth_config.block_delays.clone(),
         )
         .in_current_span(),
@@ -114,13 +119,8 @@ pub async fn run_keeper_threads(
     // Spawn a thread for block processing with configured delays
     spawn(
         process_new_blocks(
-            chain_state.clone(),
+            process_params.clone(),
             rx,
-            Arc::clone(&contract),
-            gas_limit,
-            chain_eth_config.escalation_policy.to_policy(),
-            metrics.clone(),
-            fulfilled_requests_cache.clone(),
             chain_eth_config.block_delays.clone(),
         )
         .in_current_span(),
diff --git a/apps/fortuna/src/keeper/block.rs b/apps/fortuna/src/keeper/block.rs
@@ -1,6 +1,6 @@
 use {
     crate::{
-        api::{self, BlockchainState},
+        api::{BlockchainState, History},
         chain::{ethereum::InstrumentedSignablePythContract, reader::BlockNumber},
         eth_utils::utils::EscalationPolicy,
         keeper::keeper_metrics::KeeperMetrics,
@@ -36,6 +36,17 @@ pub struct BlockRange {
     pub to: BlockNumber,
 }
 
+#[derive(Clone)]
+pub struct ProcessParams {
+    pub contract: Arc<InstrumentedSignablePythContract>,
+    pub gas_limit: U256,
+    pub escalation_policy: EscalationPolicy,
+    pub chain_state: BlockchainState,
+    pub metrics: Arc<KeeperMetrics>,
+    pub history: Arc<RwLock<History>>,
+    pub fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
+}
+
 /// Get the latest safe block number for the chain. Retry internally if there is an error.
 pub async fn get_latest_safe_block(chain_state: &BlockchainState) -> BlockNumber {
     loop {
@@ -63,15 +74,7 @@ pub async fn get_latest_safe_block(chain_state: &BlockchainState) -> BlockNumber
 #[tracing::instrument(skip_all, fields(
     range_from_block = block_range.from, range_to_block = block_range.to
 ))]
-pub async fn process_block_range(
-    block_range: BlockRange,
-    contract: Arc<InstrumentedSignablePythContract>,
-    gas_limit: U256,
-    escalation_policy: EscalationPolicy,
-    chain_state: api::BlockchainState,
-    metrics: Arc<KeeperMetrics>,
-    fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
-) {
+pub async fn process_block_range(block_range: BlockRange, process_params: ProcessParams) {
     let BlockRange {
         from: first_block,
         to: last_block,
@@ -89,12 +92,7 @@ pub async fn process_block_range(
                 from: current_block,
                 to: to_block,
             },
-            contract.clone(),
-            gas_limit,
-            escalation_policy.clone(),
-            chain_state.clone(),
-            metrics.clone(),
-            fulfilled_requests_cache.clone(),
+            process_params.clone(),
         )
         .in_current_span()
         .await;
@@ -110,22 +108,15 @@ pub async fn process_block_range(
 #[tracing::instrument(name = "batch", skip_all, fields(
     batch_from_block = block_range.from, batch_to_block = block_range.to
 ))]
-pub async fn process_single_block_batch(
-    block_range: BlockRange,
-    contract: Arc<InstrumentedSignablePythContract>,
-    gas_limit: U256,
-    escalation_policy: EscalationPolicy,
-    chain_state: api::BlockchainState,
-    metrics: Arc<KeeperMetrics>,
-    fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
-) {
+pub async fn process_single_block_batch(block_range: BlockRange, process_params: ProcessParams) {
     loop {
-        let events_res = chain_state
+        let events_res = process_params
+            .chain_state
             .contract
             .get_request_with_callback_events(
                 block_range.from,
                 block_range.to,
-                chain_state.provider_address,
+                process_params.chain_state.provider_address,
             )
             .await;
 
@@ -134,21 +125,15 @@ pub async fn process_single_block_batch(
                 tracing::info!(num_of_events = &events.len(), "Processing",);
                 for event in &events {
                     // the write lock guarantees we spawn only one task per sequence number
-                    let newly_inserted = fulfilled_requests_cache
+                    let newly_inserted = process_params
+                        .fulfilled_requests_cache
                         .write()
                         .await
                         .insert(event.sequence_number);
                     if newly_inserted {
                         spawn(
-                            process_event_with_backoff(
-                                event.clone(),
-                                chain_state.clone(),
-                                contract.clone(),
-                                gas_limit,
-                                escalation_policy.clone(),
-                                metrics.clone(),
-                            )
-                            .in_current_span(),
+                            process_event_with_backoff(event.clone(), process_params.clone())
+                                .in_current_span(),
                         );
                     }
                 }
@@ -288,100 +273,56 @@ pub async fn watch_blocks(
 /// 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,
+    process_params: ProcessParams,
     mut rx: mpsc::Receiver<BlockRange>,
-    contract: Arc<InstrumentedSignablePythContract>,
-    gas_limit: U256,
-    escalation_policy: EscalationPolicy,
-    metrics: Arc<KeeperMetrics>,
-    fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
     block_delays: Vec<u64>,
 ) {
     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.clone(),
-                Arc::clone(&contract),
-                gas_limit,
-                escalation_policy.clone(),
-                chain_state.clone(),
-                metrics.clone(),
-                fulfilled_requests_cache.clone(),
-            )
-            .in_current_span()
-            .await;
+            process_block_range(block_range.clone(), process_params.clone())
+                .in_current_span()
+                .await;
 
             // Then process with each configured delay
             for delay in &block_delays {
                 let adjusted_range = BlockRange {
                     from: block_range.from.saturating_sub(*delay),
                     to: block_range.to.saturating_sub(*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;
+                process_block_range(adjusted_range, process_params.clone())
+                    .in_current_span()
+                    .await;
             }
         }
     }
 }
 
 /// Processes the backlog_range for a chain.
 /// It processes the backlog range for each configured block delay.
-#[allow(clippy::too_many_arguments)]
 #[tracing::instrument(skip_all)]
 pub async fn process_backlog(
+    process_params: ProcessParams,
     backlog_range: BlockRange,
-    contract: Arc<InstrumentedSignablePythContract>,
-    gas_limit: U256,
-    escalation_policy: EscalationPolicy,
-    chain_state: BlockchainState,
-    metrics: Arc<KeeperMetrics>,
-    fulfilled_requests_cache: Arc<RwLock<HashSet<u64>>>,
     block_delays: Vec<u64>,
 ) {
     tracing::info!("Processing backlog");
     // Process blocks immediately first
-    process_block_range(
-        backlog_range.clone(),
-        Arc::clone(&contract),
-        gas_limit,
-        escalation_policy.clone(),
-        chain_state.clone(),
-        metrics.clone(),
-        fulfilled_requests_cache.clone(),
-    )
-    .in_current_span()
-    .await;
+    process_block_range(backlog_range.clone(), process_params.clone())
+        .in_current_span()
+        .await;
 
     // Then process with each configured delay
     for delay in &block_delays {
         let adjusted_range = BlockRange {
             from: backlog_range.from.saturating_sub(*delay),
             to: backlog_range.to.saturating_sub(*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;
+        process_block_range(adjusted_range, process_params.clone())
+            .in_current_span()
+            .await;
     }
     tracing::info!("Backlog processed");
 }
diff --git a/apps/fortuna/src/keeper/process_event.rs b/apps/fortuna/src/keeper/process_event.rs
