feat: implement actor model for Argus keeper service

Co-Authored-By: Tejas Badadare <tejas@dourolabs.xyz>

apps/argus/Cargo.toml

@@ -19,6 +19,7 @@ hex = "0.4.3"
 prometheus-client = { version = "0.21.2" }
 pythnet-sdk = { path = "../../pythnet/pythnet_sdk", features = ["strum"] }
 rand = "0.8.5"
+ractor = "0.15.3"
 reqwest = { version = "0.11.22", features = ["json", "blocking"] }
 serde = { version = "1.0.188", features = ["derive"] }
 serde_qs = { version = "0.12.0", features = ["axum"] }

apps/argus/src/actors/chain_price_listener.rs

@@ -0,0 +1,111 @@
+use {
+    crate::actors::types::*,
+    anyhow::Result,
+    async_trait::async_trait,
+    ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort},
+    std::{collections::{HashMap, HashSet}, sync::Arc, time::Duration},
+    tokio::{sync::RwLock, time},
+    tracing,
+};
+
+pub struct ChainPriceListener {
+    chain_id: String,
+    contract: Arc<dyn PulseContractInterface + Send + Sync>,
+    feed_ids: HashSet<PriceId>,
+    latest_prices: Arc<RwLock<HashMap<PriceId, Price>>>,
+    poll_interval: Duration,
+}
+
+#[async_trait]
+pub trait PulseContractInterface {
+    async fn get_price_unsafe(&self, subscription_id: SubscriptionId, feed_id: &PriceId) -> Result<Option<Price>>;
+    
+    async fn subscribe_to_price_events(&self) -> Result<()>;
+}
+
+impl Actor for ChainPriceListener {
+    type Msg = ChainPriceListenerMessage;
+    type State = Self;
+    type Arguments = (String, Arc<dyn PulseContractInterface + Send + Sync>, Duration);
+
+    async fn pre_start(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        (chain_id, contract, poll_interval): Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        let listener = ChainPriceListener {
+            chain_id,
+            contract,
+            feed_ids: HashSet::new(),
+            latest_prices: Arc::new(RwLock::new(HashMap::new())),
+            poll_interval,
+        };
+
+        if let Err(e) = listener.contract.subscribe_to_price_events().await {
+            tracing::error!(
+                chain_id = listener.chain_id,
+                error = %e,
+                "Failed to subscribe to price events"
+            );
+        }
+
+        let poll_interval = listener.poll_interval;
+        let myself_clone = myself.clone();
+        tokio::spawn(async move {
+            let mut interval = time::interval(poll_interval);
+            loop {
+                interval.tick().await;
+                tracing::debug!(chain_id = chain_id, "Polling for on-chain price updates");
+            }
+        });
+
+        Ok(listener)
+    }
+
+    async fn handle(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        _state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            ChainPriceListenerMessage::GetLatestPrice(_) => {
+            }
+            ChainPriceListenerMessage::UpdateFeedIdSet(_) => {
+            }
+        }
+        Ok(())
+    }
+
+    async fn handle_rpc(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+        reply_port: RpcReplyPort<ChainPriceListenerResponse>,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            ChainPriceListenerMessage::GetLatestPrice(feed_id) => {
+                let price = None;
+                
+                let _ = reply_port.send(ChainPriceListenerResponse::LatestPrice(price));
+            }
+            ChainPriceListenerMessage::UpdateFeedIdSet(feed_ids) => {
+                let old_count = state.feed_ids.len();
+                let new_count = feed_ids.len();
+                
+                state.feed_ids = feed_ids;
+                
+                tracing::info!(
+                    chain_id = state.chain_id,
+                    old_count = old_count,
+                    new_count = new_count,
+                    "Updated feed ID set"
+                );
+                
+                let _ = reply_port.send(ChainPriceListenerResponse::FeedIdSetUpdated);
+            }
+        }
+        Ok(())
+    }
+}

apps/argus/src/actors/controller.rs

@@ -0,0 +1,227 @@
+use {
+    crate::actors::types::*,
+    anyhow::Result,
+    ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort, call_t},
+    std::{collections::{HashMap, HashSet}, sync::Arc, time::Duration},
+    tokio::time,
+    tracing,
+};
+
+pub struct Controller {
+    chain_id: String,
+    subscription_listener: ActorRef<SubscriptionListenerMessage>,
+    pyth_price_listener: ActorRef<PythPriceListenerMessage>,
+    chain_price_listener: ActorRef<ChainPriceListenerMessage>,
+    price_pusher: ActorRef<PricePusherMessage>,
+    update_interval: Duration,
+    update_loop_running: bool,
+    active_subscriptions: HashMap<SubscriptionId, Subscription>,
+    feed_ids: HashSet<PriceId>,
+}
+
+impl Actor for Controller {
+    type Msg = ControllerMessage;
+    type State = Self;
+    type Arguments = (
+        String,
+        ActorRef<SubscriptionListenerMessage>,
+        ActorRef<PythPriceListenerMessage>,
+        ActorRef<ChainPriceListenerMessage>,
+        ActorRef<PricePusherMessage>,
+        Duration,
+    );
+
+    async fn pre_start(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        (
+            chain_id,
+            subscription_listener,
+            pyth_price_listener,
+            chain_price_listener,
+            price_pusher,
+            update_interval,
+        ): Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        let controller = Controller {
+            chain_id,
+            subscription_listener,
+            pyth_price_listener,
+            chain_price_listener,
+            price_pusher,
+            update_interval,
+            update_loop_running: false,
+            active_subscriptions: HashMap::new(),
+            feed_ids: HashSet::new(),
+        };
+
+        let _ = myself.cast(ControllerMessage::StartUpdateLoop);
+
+        Ok(controller)
+    }
+
+    async fn handle(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            ControllerMessage::StartUpdateLoop => {
+                if !state.update_loop_running {
+                    state.update_loop_running = true;
+                    
+                    let update_interval = state.update_interval;
+                    let myself_clone = myself.clone();
+                    tokio::spawn(async move {
+                        let mut interval = time::interval(update_interval);
+                        while let Ok(ControllerResponse::UpdateLoopStarted) = call_t!(myself_clone, ControllerMessage::StartUpdateLoop).await {
+                            interval.tick().await;
+                            let _ = myself_clone.cast(ControllerMessage::CheckForUpdates);
+                        }
+                    });
+                    
+                    tracing::info!(chain_id = state.chain_id, "Update loop started");
+                }
+            }
+            ControllerMessage::StopUpdateLoop => {
+                state.update_loop_running = false;
+                tracing::info!(chain_id = state.chain_id, "Update loop stopped");
+            }
+            ControllerMessage::CheckForUpdates => {
+                match call_t!(state.subscription_listener, SubscriptionListenerMessage::GetActiveSubscriptions).await {
+                    Ok(SubscriptionListenerResponse::ActiveSubscriptions(subscriptions)) => {
+                        state.active_subscriptions = subscriptions;
+                        
+                        let mut feed_ids = HashSet::new();
+                        for subscription in state.active_subscriptions.values() {
+                            for price_id in &subscription.price_ids {
+                                feed_ids.insert(*price_id);
+                            }
+                        }
+                        
+                        if feed_ids != state.feed_ids {
+                            state.feed_ids = feed_ids.clone();
+                            
+                            let _ = call_t!(state.pyth_price_listener, PythPriceListenerMessage::UpdateFeedIdSet(feed_ids.clone())).await;
+                            
+                            let _ = call_t!(state.chain_price_listener, ChainPriceListenerMessage::UpdateFeedIdSet(feed_ids)).await;
+                        }
+                        
+                        for (subscription_id, subscription) in &state.active_subscriptions {
+                            if state.should_update_subscription(subscription).await {
+                                let push_request = PushRequest {
+                                    subscription_id: *subscription_id,
+                                    price_ids: subscription.price_ids.clone(),
+                                };
+                                
+                                let _ = state.price_pusher.cast(PricePusherMessage::PushPriceUpdates(push_request));
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        tracing::error!(
+                            chain_id = state.chain_id,
+                            error = %e,
+                            "Failed to get active subscriptions"
+                        );
+                    }
+                    _ => {}
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn handle_rpc(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+        reply_port: RpcReplyPort<ControllerResponse>,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            ControllerMessage::StartUpdateLoop => {
+                state.update_loop_running = true;
+                let _ = reply_port.send(ControllerResponse::UpdateLoopStarted);
+            }
+            ControllerMessage::StopUpdateLoop => {
+                state.update_loop_running = false;
+                let _ = reply_port.send(ControllerResponse::UpdateLoopStopped);
+            }
+            ControllerMessage::CheckForUpdates => {
+                let _ = reply_port.send(ControllerResponse::UpdateCheckResult(vec![]));
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Controller {
+    async fn should_update_subscription(&self, subscription: &Subscription) -> bool {
+        if subscription.update_criteria.update_on_heartbeat {
+            let current_time = chrono::Utc::now().timestamp() as u64;
+            let time_since_last_update = current_time.saturating_sub(subscription.last_updated_at);
+            
+            if time_since_last_update >= subscription.update_criteria.heartbeat_seconds as u64 {
+                tracing::info!(
+                    chain_id = self.chain_id,
+                    subscription_id = subscription.id,
+                    time_since_last_update = time_since_last_update,
+                    heartbeat_seconds = subscription.update_criteria.heartbeat_seconds,
+                    "Subscription needs update due to heartbeat"
+                );
+                return true;
+            }
+        }
+        
+        if subscription.update_criteria.update_on_deviation {
+            for price_id in &subscription.price_ids {
+                let pyth_price = match call_t!(self.pyth_price_listener, PythPriceListenerMessage::GetLatestPrice(*price_id)).await {
+                    Ok(PythPriceListenerResponse::LatestPrice(Some(price))) => price,
+                    _ => continue,
+                };
+                
+                let chain_price = match call_t!(self.chain_price_listener, ChainPriceListenerMessage::GetLatestPrice(*price_id)).await {
+                    Ok(ChainPriceListenerResponse::LatestPrice(Some(price))) => price,
+                    _ => continue,
+                };
+                
+                let deviation = calculate_deviation(pyth_price.price, chain_price.price);
+                
+                if deviation >= subscription.update_criteria.deviation_threshold_bps as u64 {
+                    tracing::info!(
+                        chain_id = self.chain_id,
+                        subscription_id = subscription.id,
+                        price_id = hex::encode(price_id),
+                        deviation = deviation,
+                        threshold = subscription.update_criteria.deviation_threshold_bps,
+                        "Subscription needs update due to price deviation"
+                    );
+                    return true;
+                }
+            }
+        }
+        
+        false
+    }
+}
+
+fn calculate_deviation(price1: i64, price2: i64) -> u64 {
+    if price1 == 0 || price2 == 0 {
+        return 0;
+    }
+    
+    let price1_abs = price1.abs() as u64;
+    let price2_abs = price2.abs() as u64;
+    
+    let diff = if price1_abs > price2_abs {
+        price1_abs - price2_abs
+    } else {
+        price2_abs - price1_abs
+    };
+    
+    let base = std::cmp::max(price1_abs, price2_abs);
+    
+    (diff * 10000) / base
+}

apps/argus/src/actors/mod.rs

@@ -0,0 +1,13 @@
+pub mod subscription_listener;
+pub mod pyth_price_listener;
+pub mod chain_price_listener;
+pub mod controller;
+pub mod price_pusher;
+pub mod types;
+
+pub use subscription_listener::SubscriptionListener;
+pub use pyth_price_listener::PythPriceListener;
+pub use chain_price_listener::ChainPriceListener;
+pub use controller::Controller;
+pub use price_pusher::PricePusher;
+pub use types::*;

apps/argus/src/actors/price_pusher.rs

@@ -0,0 +1,199 @@
+use {
+    crate::actors::types::*,
+    anyhow::Result,
+    async_trait::async_trait,
+    backoff::{backoff::Backoff, ExponentialBackoff},
+    ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort},
+    std::{sync::Arc, time::Duration},
+    tokio::time,
+    tracing,
+};
+
+pub struct PricePusher {
+    chain_id: String,
+    contract: Arc<dyn PulseContractInterface + Send + Sync>,
+    hermes_client: Arc<dyn HermesClient + Send + Sync>,
+    backoff_policy: ExponentialBackoff,
+}
+
+#[async_trait]
+pub trait PulseContractInterface {
+    async fn update_price_feeds(
+        &self,
+        subscription_id: SubscriptionId,
+        price_ids: &[PriceId],
+        update_data: &[Vec<u8>],
+    ) -> Result<String>;
+}
+
+#[async_trait]
+pub trait HermesClient {
+    async fn get_price_update_data(&self, feed_ids: &[PriceId]) -> Result<Vec<Vec<u8>>>;
+}
+
+impl Actor for PricePusher {
+    type Msg = PricePusherMessage;
+    type State = Self;
+    type Arguments = (
+        String,
+        Arc<dyn PulseContractInterface + Send + Sync>,
+        Arc<dyn HermesClient + Send + Sync>,
+        ExponentialBackoff,
+    );
+
+    async fn pre_start(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        (chain_id, contract, hermes_client, backoff_policy): Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        let pusher = PricePusher {
+            chain_id,
+            contract,
+            hermes_client,
+            backoff_policy,
+        };
+
+        Ok(pusher)
+    }
+
+    async fn handle(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            PricePusherMessage::PushPriceUpdates(push_request) => {
+                let price_ids = push_request.price_ids.clone();
+                match state.hermes_client.get_price_update_data(&price_ids).await {
+                    Ok(update_data) => {
+                        let mut backoff = state.backoff_policy.clone();
+                        let mut attempt = 0;
+                        
+                        loop {
+                            attempt += 1;
+                            
+                            match state.contract.update_price_feeds(
+                                push_request.subscription_id,
+                                &price_ids,
+                                &update_data,
+                            ).await {
+                                Ok(tx_hash) => {
+                                    tracing::info!(
+                                        chain_id = state.chain_id,
+                                        subscription_id = push_request.subscription_id,
+                                        tx_hash = tx_hash,
+                                        attempt = attempt,
+                                        "Successfully pushed price updates"
+                                    );
+                                    break;
+                                }
+                                Err(e) => {
+                                    if let Some(duration) = backoff.next_backoff() {
+                                        tracing::warn!(
+                                            chain_id = state.chain_id,
+                                            subscription_id = push_request.subscription_id,
+                                            error = %e,
+                                            attempt = attempt,
+                                            retry_after_ms = duration.as_millis(),
+                                            "Failed to push price updates, retrying"
+                                        );
+                                        time::sleep(duration).await;
+                                    } else {
+                                        tracing::error!(
+                                            chain_id = state.chain_id,
+                                            subscription_id = push_request.subscription_id,
+                                            error = %e,
+                                            attempt = attempt,
+                                            "Failed to push price updates, giving up"
+                                        );
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        tracing::error!(
+                            chain_id = state.chain_id,
+                            subscription_id = push_request.subscription_id,
+                            error = %e,
+                            "Failed to get price update data from Hermes"
+                        );
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn handle_rpc(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+        reply_port: RpcReplyPort<PricePusherResponse>,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            PricePusherMessage::PushPriceUpdates(push_request) => {
+                let price_ids = push_request.price_ids.clone();
+                match state.hermes_client.get_price_update_data(&price_ids).await {
+                    Ok(update_data) => {
+                        match state.contract.update_price_feeds(
+                            push_request.subscription_id,
+                            &price_ids,
+                            &update_data,
+                        ).await {
+                            Ok(tx_hash) => {
+                                tracing::info!(
+                                    chain_id = state.chain_id,
+                                    subscription_id = push_request.subscription_id,
+                                    tx_hash = tx_hash,
+                                    "Successfully pushed price updates"
+                                );
+                                
+                                let _ = reply_port.send(PricePusherResponse::PushResult {
+                                    success: true,
+                                    subscription_id: push_request.subscription_id,
+                                    tx_hash: Some(tx_hash),
+                                    error: None,
+                                });
+                            }
+                            Err(e) => {
+                                tracing::error!(
+                                    chain_id = state.chain_id,
+                                    subscription_id = push_request.subscription_id,
+                                    error = %e,
+                                    "Failed to push price updates"
+                                );
+                                
+                                let _ = reply_port.send(PricePusherResponse::PushResult {
+                                    success: false,
+                                    subscription_id: push_request.subscription_id,
+                                    tx_hash: None,
+                                    error: Some(e.to_string()),
+                                });
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        tracing::error!(
+                            chain_id = state.chain_id,
+                            subscription_id = push_request.subscription_id,
+                            error = %e,
+                            "Failed to get price update data from Hermes"
+                        );
+                        
+                        let _ = reply_port.send(PricePusherResponse::PushResult {
+                            success: false,
+                            subscription_id: push_request.subscription_id,
+                            tx_hash: None,
+                            error: Some(format!("Failed to get price update data: {}", e)),
+                        });
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+}

apps/argus/src/actors/pyth_price_listener.rs

@@ -0,0 +1,120 @@
+use {
+    crate::actors::types::*,
+    anyhow::Result,
+    async_trait::async_trait,
+    ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort},
+    std::{collections::{HashMap, HashSet}, sync::Arc},
+    tokio::sync::RwLock,
+    tracing,
+};
+
+pub struct PythPriceListener {
+    chain_id: String,
+    hermes_client: Arc<dyn HermesClient + Send + Sync>,
+    feed_ids: HashSet<PriceId>,
+    latest_prices: Arc<RwLock<HashMap<PriceId, Price>>>,
+}
+
+#[async_trait]
+pub trait HermesClient {
+    async fn connect(&self) -> Result<()>;
+    
+    async fn subscribe_to_price_updates(&self, feed_ids: &HashSet<PriceId>) -> Result<()>;
+    
+    async fn get_latest_price(&self, feed_id: &PriceId) -> Result<Option<Price>>;
+}
+
+impl Actor for PythPriceListener {
+    type Msg = PythPriceListenerMessage;
+    type State = Self;
+    type Arguments = (String, Arc<dyn HermesClient + Send + Sync>);
+
+    async fn pre_start(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        (chain_id, hermes_client): Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        let listener = PythPriceListener {
+            chain_id,
+            hermes_client,
+            feed_ids: HashSet::new(),
+            latest_prices: Arc::new(RwLock::new(HashMap::new())),
+        };
+
+        if let Err(e) = listener.hermes_client.connect().await {
+            tracing::error!(
+                chain_id = listener.chain_id,
+                error = %e,
+                "Failed to connect to Hermes"
+            );
+        }
+
+        Ok(listener)
+    }
+
+    async fn handle(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        _state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            PythPriceListenerMessage::GetLatestPrice(_) => {
+            }
+            PythPriceListenerMessage::UpdateFeedIdSet(_) => {
+            }
+        }
+        Ok(())
+    }
+
+    async fn handle_rpc(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+        reply_port: RpcReplyPort<PythPriceListenerResponse>,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            PythPriceListenerMessage::GetLatestPrice(feed_id) => {
+                let price = match state.hermes_client.get_latest_price(&feed_id).await {
+                    Ok(price) => price,
+                    Err(e) => {
+                        tracing::error!(
+                            chain_id = state.chain_id,
+                            feed_id = hex::encode(feed_id),
+                            error = %e,
+                            "Failed to get latest price from Hermes"
+                        );
+                        None
+                    }
+                };
+                
+                let _ = reply_port.send(PythPriceListenerResponse::LatestPrice(price));
+            }
+            PythPriceListenerMessage::UpdateFeedIdSet(feed_ids) => {
+                let old_count = state.feed_ids.len();
+                let new_count = feed_ids.len();
+                
+                state.feed_ids = feed_ids.clone();
+                
+                if let Err(e) = state.hermes_client.subscribe_to_price_updates(&feed_ids).await {
+                    tracing::error!(
+                        chain_id = state.chain_id,
+                        error = %e,
+                        "Failed to subscribe to price updates"
+                    );
+                }
+                
+                tracing::info!(
+                    chain_id = state.chain_id,
+                    old_count = old_count,
+                    new_count = new_count,
+                    "Updated feed ID set"
+                );
+                
+                let _ = reply_port.send(PythPriceListenerResponse::FeedIdSetUpdated);
+            }
+        }
+        Ok(())
+    }
+}

apps/argus/src/actors/subscription_listener.rs

@@ -0,0 +1,152 @@
+use {
+    crate::actors::types::*,
+    anyhow::Result,
+    async_trait::async_trait,
+    ethers::providers::Middleware,
+    ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort},
+    std::{collections::{HashMap, HashSet}, sync::Arc, time::Duration},
+    tokio::time,
+    tracing,
+};
+
+pub struct SubscriptionListener {
+    chain_id: String,
+    contract: Arc<dyn PulseContractInterface + Send + Sync>,
+    active_subscriptions: HashMap<SubscriptionId, Subscription>,
+    poll_interval: Duration,
+}
+
+#[async_trait]
+pub trait PulseContractInterface {
+    async fn get_active_subscriptions(&self) -> Result<HashMap<SubscriptionId, Subscription>>;
+    
+    async fn subscribe_to_events(&self) -> Result<()>;
+}
+
+impl Actor for SubscriptionListener {
+    type Msg = SubscriptionListenerMessage;
+    type State = Self;
+    type Arguments = (String, Arc<dyn PulseContractInterface + Send + Sync>, Duration);
+
+    async fn pre_start(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        (chain_id, contract, poll_interval): Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        let mut listener = SubscriptionListener {
+            chain_id,
+            contract,
+            active_subscriptions: HashMap::new(),
+            poll_interval,
+        };
+
+        match listener.refresh_subscriptions().await {
+            Ok(_) => {
+                tracing::info!(
+                    chain_id = listener.chain_id,
+                    "Loaded {} active subscriptions",
+                    listener.active_subscriptions.len()
+                );
+            }
+            Err(e) => {
+                tracing::error!(
+                    chain_id = listener.chain_id,
+                    error = %e,
+                    "Failed to load active subscriptions"
+                );
+            }
+        }
+
+        if let Err(e) = listener.contract.subscribe_to_events().await {
+            tracing::error!(
+                chain_id = listener.chain_id,
+                error = %e,
+                "Failed to subscribe to contract events"
+            );
+        }
+
+        let poll_interval = listener.poll_interval;
+        tokio::spawn(async move {
+            let mut interval = time::interval(poll_interval);
+            loop {
+                interval.tick().await;
+                let _ = myself.cast(SubscriptionListenerMessage::RefreshSubscriptions);
+            }
+        });
+
+        Ok(listener)
+    }
+
+    async fn handle(
+        &self,
+        myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            SubscriptionListenerMessage::GetActiveSubscriptions => {
+            }
+            SubscriptionListenerMessage::RefreshSubscriptions => {
+                if let Err(e) = state.refresh_subscriptions().await {
+                    tracing::error!(
+                        chain_id = state.chain_id,
+                        error = %e,
+                        "Failed to refresh subscriptions"
+                    );
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn handle_rpc(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        message: Self::Msg,
+        state: &mut Self::State,
+        reply_port: RpcReplyPort<SubscriptionListenerResponse>,
+    ) -> Result<(), ActorProcessingErr> {
+        match message {
+            SubscriptionListenerMessage::GetActiveSubscriptions => {
+                let _ = reply_port.send(SubscriptionListenerResponse::ActiveSubscriptions(
+                    state.active_subscriptions.clone(),
+                ));
+            }
+            SubscriptionListenerMessage::RefreshSubscriptions => {
+                if let Err(e) = state.refresh_subscriptions().await {
+                    tracing::error!(
+                        chain_id = state.chain_id,
+                        error = %e,
+                        "Failed to refresh subscriptions"
+                    );
+                }
+                let _ = reply_port.send(SubscriptionListenerResponse::RefreshAcknowledged);
+            }
+        }
+        Ok(())
+    }
+}
+
+impl SubscriptionListener {
+    async fn refresh_subscriptions(&mut self) -> Result<()> {
+        let subscriptions = self.contract.get_active_subscriptions().await?;
+        
+        let old_ids: HashSet<_> = self.active_subscriptions.keys().cloned().collect();
+        let new_ids: HashSet<_> = subscriptions.keys().cloned().collect();
+        
+        let added = new_ids.difference(&old_ids).count();
+        let removed = old_ids.difference(&new_ids).count();
+        
+        if added > 0 || removed > 0 {
+            tracing::info!(
+                chain_id = self.chain_id,
+                added = added,
+                removed = removed,
+                "Subscription changes detected"
+            );
+        }
+        
+        self.active_subscriptions = subscriptions;
+        Ok(())
+    }
+}

apps/argus/src/actors/types.rs

@@ -0,0 +1,107 @@
+use {
+    ethers::types::{Address, U256},
+    serde::{Deserialize, Serialize},
+    std::collections::{HashMap, HashSet},
+};
+
+pub type PriceId = [u8; 32];
+
+pub type SubscriptionId = u64;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Price {
+    pub price: i64,
+    pub conf: u64,
+    pub expo: i32,
+    pub publish_time: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Subscription {
+    pub id: SubscriptionId,
+    pub price_ids: Vec<PriceId>,
+    pub manager: Address,
+    pub is_active: bool,
+    pub update_criteria: UpdateCriteria,
+    pub last_updated_at: u64,
+    pub balance: U256,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct UpdateCriteria {
+    pub update_on_heartbeat: bool,
+    pub heartbeat_seconds: u32,
+    pub update_on_deviation: bool,
+    pub deviation_threshold_bps: u32,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum SubscriptionListenerMessage {
+    GetActiveSubscriptions,
+    RefreshSubscriptions,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum PythPriceListenerMessage {
+    GetLatestPrice(PriceId),
+    UpdateFeedIdSet(HashSet<PriceId>),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum ChainPriceListenerMessage {
+    GetLatestPrice(PriceId),
+    UpdateFeedIdSet(HashSet<PriceId>),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum ControllerMessage {
+    StartUpdateLoop,
+    StopUpdateLoop,
+    CheckForUpdates,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PushRequest {
+    pub subscription_id: SubscriptionId,
+    pub price_ids: Vec<PriceId>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum PricePusherMessage {
+    PushPriceUpdates(PushRequest),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum SubscriptionListenerResponse {
+    ActiveSubscriptions(HashMap<SubscriptionId, Subscription>),
+    RefreshAcknowledged,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum PythPriceListenerResponse {
+    LatestPrice(Option<Price>),
+    FeedIdSetUpdated,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum ChainPriceListenerResponse {
+    LatestPrice(Option<Price>),
+    FeedIdSetUpdated,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum ControllerResponse {
+    UpdateLoopStarted,
+    UpdateLoopStopped,
+    UpdateCheckResult(Vec<PushRequest>),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum PricePusherResponse {
+    PushResult {
+        success: bool,
+        subscription_id: SubscriptionId,
+        tx_hash: Option<String>,
+        error: Option<String>,
+    },
+}

apps/argus/src/keeper.rs

@@ -1,12 +1,25 @@
 use {
     crate::{
-        api::BlockchainState, chain::ethereum::InstrumentedSignablePythContract,
+        actors::{
+            chain_price_listener::{ChainPriceListener, PulseContractInterface as ChainPriceContractInterface},
+            controller::Controller,
+            price_pusher::{PricePusher, PulseContractInterface as PricePusherContractInterface, HermesClient},
+            pyth_price_listener::{PythPriceListener, HermesClient as PythPriceHermesClient},
+            subscription_listener::{SubscriptionListener, PulseContractInterface as SubscriptionContractInterface},
+            types::*,
+        },
+        api::BlockchainState, 
+        chain::ethereum::InstrumentedSignablePythContract,
         config::EthereumConfig,
     },
+    anyhow::Result,
+    async_trait::async_trait,
+    backoff::ExponentialBackoff,
     ethers::signers::Signer,
     fortuna::eth_utils::traced_client::RpcMetrics,
     keeper_metrics::KeeperMetrics,
-    std::sync::Arc,
+    ractor::{Actor, ActorRef, SupervisionStrategy},
+    std::{collections::{HashMap, HashSet}, sync::Arc, time::Duration},
     tracing,
 };
 
@@ -17,7 +30,7 @@ pub async fn run_keeper_threads(
     private_key: String,
     chain_eth_config: EthereumConfig,
     chain_state: BlockchainState,
-    _metrics: Arc<KeeperMetrics>,
+    metrics: Arc<KeeperMetrics>,
     rpc_metrics: Arc<RpcMetrics>,
 ) {
     tracing::info!("starting keeper");
@@ -32,7 +45,209 @@ pub async fn run_keeper_threads(
         .await
         .expect("Chain config should be valid"),
     );
-    let _keeper_address = contract.wallet().address();
+    let keeper_address = contract.wallet().address();
+    
+    tracing::info!(
+        chain_id = chain_state.id,
+        keeper_address = %keeper_address,
+        "Keeper address"
+    );
+
+    let subscription_contract = Arc::new(PulseContractAdapter {
+        contract: contract.clone(),
+        chain_id: chain_state.id.clone(),
+    });
+    
+    let chain_price_contract = Arc::new(PulseContractAdapter {
+        contract: contract.clone(),
+        chain_id: chain_state.id.clone(),
+    });
+    
+    let price_pusher_contract = Arc::new(PulseContractAdapter {
+        contract: contract.clone(),
+        chain_id: chain_state.id.clone(),
+    });
+    
+    let hermes_client = Arc::new(HermesClientAdapter {
+        chain_id: chain_state.id.clone(),
+    });
+    
+    let subscription_poll_interval = Duration::from_secs(60); // Poll for subscriptions every 60 seconds
+    let chain_price_poll_interval = Duration::from_secs(10); // Poll for on-chain prices every 10 seconds
+    let controller_update_interval = Duration::from_secs(5); // Run the update loop every 5 seconds
+    
+    let backoff_policy = ExponentialBackoff {
+        initial_interval: Duration::from_secs(1),
+        max_interval: Duration::from_secs(60),
+        multiplier: 2.0,
+        max_elapsed_time: Some(Duration::from_secs(300)), // Give up after 5 minutes
+        ..ExponentialBackoff::default()
+    };
+    
+    let (subscription_listener, _) = Actor::spawn(
+        None,
+        SubscriptionListener,
+        (
+            chain_state.id.clone(),
+            subscription_contract as Arc<dyn SubscriptionContractInterface + Send + Sync>,
+            subscription_poll_interval,
+        ),
+        SupervisionStrategy::Restart(5),
+    )
+    .await
+    .expect("Failed to spawn SubscriptionListener actor");
+    
+    let (pyth_price_listener, _) = Actor::spawn(
+        None,
+        PythPriceListener,
+        (
+            chain_state.id.clone(),
+            hermes_client.clone() as Arc<dyn PythPriceHermesClient + Send + Sync>,
+        ),
+        SupervisionStrategy::Restart(5),
+    )
+    .await
+    .expect("Failed to spawn PythPriceListener actor");
+    
+    let (chain_price_listener, _) = Actor::spawn(
+        None,
+        ChainPriceListener,
+        (
+            chain_state.id.clone(),
+            chain_price_contract as Arc<dyn ChainPriceContractInterface + Send + Sync>,
+            chain_price_poll_interval,
+        ),
+        SupervisionStrategy::Restart(5),
+    )
+    .await
+    .expect("Failed to spawn ChainPriceListener actor");
+    
+    let (price_pusher, _) = Actor::spawn(
+        None,
+        PricePusher,
+        (
+            chain_state.id.clone(),
+            price_pusher_contract as Arc<dyn PricePusherContractInterface + Send + Sync>,
+            hermes_client as Arc<dyn HermesClient + Send + Sync>,
+            backoff_policy,
+        ),
+        SupervisionStrategy::Restart(5),
+    )
+    .await
+    .expect("Failed to spawn PricePusher actor");
+    
+    let (_controller, _) = Actor::spawn(
+        None,
+        Controller,
+        (
+            chain_state.id.clone(),
+            subscription_listener,
+            pyth_price_listener,
+            chain_price_listener,
+            price_pusher,
+            controller_update_interval,
+        ),
+        SupervisionStrategy::Restart(5),
+    )
+    .await
+    .expect("Failed to spawn Controller actor");
+    
+    tracing::info!(chain_id = chain_state.id, "Keeper actors started");
+}
+
+struct PulseContractAdapter {
+    contract: Arc<InstrumentedSignablePythContract>,
+    chain_id: String,
+}
+
+#[async_trait]
+impl SubscriptionContractInterface for PulseContractAdapter {
+    async fn get_active_subscriptions(&self) -> Result<HashMap<SubscriptionId, Subscription>> {
+        tracing::debug!(chain_id = self.chain_id, "Getting active subscriptions");
+        Ok(HashMap::new())
+    }
+    
+    async fn subscribe_to_events(&self) -> Result<()> {
+        tracing::debug!(chain_id = self.chain_id, "Subscribing to contract events");
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl ChainPriceContractInterface for PulseContractAdapter {
+    async fn get_price_unsafe(&self, subscription_id: SubscriptionId, feed_id: &PriceId) -> Result<Option<Price>> {
+        tracing::debug!(
+            chain_id = self.chain_id,
+            subscription_id = subscription_id,
+            feed_id = hex::encode(feed_id),
+            "Getting on-chain price"
+        );
+        Ok(None)
+    }
+    
+    async fn subscribe_to_price_events(&self) -> Result<()> {
+        tracing::debug!(chain_id = self.chain_id, "Subscribing to price events");
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl PricePusherContractInterface for PulseContractAdapter {
+    async fn update_price_feeds(
+        &self,
+        subscription_id: SubscriptionId,
+        price_ids: &[PriceId],
+        update_data: &[Vec<u8>],
+    ) -> Result<String> {
+        tracing::debug!(
+            chain_id = self.chain_id,
+            subscription_id = subscription_id,
+            price_ids_count = price_ids.len(),
+            update_data_count = update_data.len(),
+            "Updating price feeds"
+        );
+        Ok("0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef".to_string())
+    }
+}
+
+struct HermesClientAdapter {
+    chain_id: String,
+}
+
+#[async_trait]
+impl PythPriceHermesClient for HermesClientAdapter {
+    async fn connect(&self) -> Result<()> {
+        tracing::debug!(chain_id = self.chain_id, "Connecting to Hermes");
+        Ok(())
+    }
+    
+    async fn subscribe_to_price_updates(&self, feed_ids: &HashSet<PriceId>) -> Result<()> {
+        tracing::debug!(
+            chain_id = self.chain_id,
+            feed_ids_count = feed_ids.len(),
+            "Subscribing to price updates"
+        );
+        Ok(())
+    }
+    
+    async fn get_latest_price(&self, feed_id: &PriceId) -> Result<Option<Price>> {
+        tracing::debug!(
+            chain_id = self.chain_id,
+            feed_id = hex::encode(feed_id),
+            "Getting latest price from Hermes"
+        );
+        Ok(None)
+    }
+}
 
-    // TODO: Spawn actors here
+#[async_trait]
+impl HermesClient for HermesClientAdapter {
+    async fn get_price_update_data(&self, feed_ids: &[PriceId]) -> Result<Vec<Vec<u8>>> {
+        tracing::debug!(
+            chain_id = self.chain_id,
+            feed_ids_count = feed_ids.len(),
+            "Getting price update data from Hermes"
+        );
+        Ok(vec![vec![0u8; 32]; feed_ids.len()])
+    }
 }

apps/argus/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod actors;
 pub mod api;
 pub mod chain;
 pub mod command;