feat(argus): Implement Keeper state and main request processing loop (#2542)

* chore: remove fortuna-specific keeper code

* feat: first pass at keeper fulfillment state and logic

* refactor: replace EscalationPolicy with a stub for now

* fix: address pr comments

Author: tejasbadadare

apps/argus/Cargo.lock

@@ -40,7 +40,9 @@ chrono = { version = "0.4.38", features = [
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 thiserror = "1.0.61"
 futures-locks = "0.7.1"
+async-trait = "0.1.88"
 
 
 [dev-dependencies]
+mockall = "0.13.1"
 axum-test = "13.1.1"

apps/argus/Cargo.toml

@@ -21,7 +21,9 @@ use {
 };
 
 pub(crate) mod fee;
+pub(crate) mod fulfillment_task;
 pub(crate) mod keeper_metrics;
+pub(crate) mod state;
 pub(crate) mod track;
 
 /// Track metrics in this interval

apps/argus/src/keeper.rs

@@ -3,13 +3,13 @@ use {
         api::BlockchainState, chain::ethereum::InstrumentedSignablePythContract,
         keeper::AccountLabel, keeper::ChainId, keeper::KeeperMetrics,
     },
-    fortuna::eth_utils::utils::{estimate_tx_cost, send_and_confirm},
     anyhow::{anyhow, Result},
     ethers::{
         middleware::Middleware,
         signers::Signer,
         types::{Address, U256},
     },
+    fortuna::eth_utils::utils::{estimate_tx_cost, send_and_confirm},
     std::sync::Arc,
     tokio::time::{self, Duration},
     tracing::{self, Instrument},

apps/argus/src/keeper/fee.rs

@@ -0,0 +1,51 @@
+use anyhow::Result;
+use tokio::task::JoinHandle;
+
+use super::state::{EscalationPolicy, PulseRequest};
+use async_trait::async_trait;
+
+#[allow(dead_code)]
+#[derive(Debug)]
+pub struct RequestFulfillmentTask {
+    /// If None, the task hasn't been spawned. If Some(fut), task is in flight or completed.
+    pub task: Option<JoinHandle<Result<()>>>,
+    pub retries: u32,
+    pub success: bool,
+
+    // The error received during fulfillment if `success` is false.
+    // We don't consider the consumer callback reverting as a failure since we catch those
+    // in the Pulse contract. Thus, this should only happen if there's a transient RPC error
+    // (tx failed to land, out of gas, etc)
+    pub error: Option<String>,
+}
+
+#[async_trait]
+pub trait RequestFulfiller: Send + Sync + 'static {
+    #[allow(dead_code)]
+    async fn fulfill_request(
+        &self,
+        request: PulseRequest,
+        hermes_url: &str,
+        escalation_policy: EscalationPolicy,
+    ) -> Result<()>;
+}
+
+#[allow(dead_code)]
+pub struct DefaultRequestFulfiller;
+
+#[async_trait]
+impl RequestFulfiller for DefaultRequestFulfiller {
+    /// Core logic of fulfilling a Pulse request
+    async fn fulfill_request(
+        &self,
+        _request: PulseRequest,
+        _hermes_url: &str,
+        _escalation_policy: EscalationPolicy,
+    ) -> Result<()> {
+        // TODO:
+        // 1. get price update by calling hermes
+        // 2. create contract call and submit it with escalation policy
+        // 3. validate receipt from tx
+        Ok(())
+    }
+}

apps/argus/src/keeper/fulfillment_task.rs

@@ -22,6 +22,7 @@ pub struct ChainIdLabel {
 }
 
 pub struct KeeperMetrics {
+    // TODO: reevaluate what metrics are useful for argus
     pub current_sequence_number: Family<AccountLabel, Gauge>,
     pub end_sequence_number: Family<AccountLabel, Gauge>,
     pub balance: Family<AccountLabel, Gauge<f64, AtomicU64>>,