feat(phoenix): Add auction_analysis_monitor

src/phoenix/auction_analysis_monitor.rs

@@ -0,0 +1,64 @@
+use anyhow::Result;
+use chrono::{DateTime, Utc};
+use lazy_static::lazy_static;
+use sqlx::PgPool;
+use tracing::debug;
+
+use crate::{
+    env::{Network, ToNetwork},
+    phoenix::{Alarm, AlarmType},
+};
+
+use super::env::APP_CONFIG;
+
+lazy_static! {
+    static ref GENESIS_TIMESTAMP: DateTime<Utc> = {
+        match &APP_CONFIG.env.to_network() {
+            Network::Mainnet => "2020-12-01T12:00:23Z".parse().unwrap(),
+            Network::Holesky => "2023-09-28T12:00:00Z".parse().unwrap(),
+        }
+    };
+}
+
+const SECONDS_PER_SLOT: u8 = 12;
+
+async fn get_latest_auction_analysis_slot(mev_pool: &PgPool) -> anyhow::Result<u32> {
+    sqlx::query_scalar!(
+        r#"
+        SELECT MAX(slot)
+        FROM auction_analysis
+        "#,
+    )
+    .fetch_one(mev_pool)
+    .await
+    .map(|max| {
+        max.expect("No maximum slot found from auction_analysis")
+            .try_into()
+            .expect("Maximum slot is negative")
+    })
+    .map_err(Into::into)
+}
+
+fn get_current_slot() -> Result<u32> {
+    let now = Utc::now();
+    let seconds_since_genesis: u32 = (now - *GENESIS_TIMESTAMP).num_seconds().try_into()?;
+    Ok(seconds_since_genesis / SECONDS_PER_SLOT as u32)
+}
+
+pub async fn run_auction_analysis_monitor(mev_pool: &PgPool, alarm: &mut Alarm) -> Result<()> {
+    let latest_slot = get_latest_auction_analysis_slot(mev_pool).await?;
+    let current_slot = get_current_slot()?;
+    let slot_lag = current_slot - latest_slot;
+    debug!(
+        "Auction analysis is {:} slots behind current slot",
+        slot_lag
+    );
+    if slot_lag > APP_CONFIG.max_auction_analysis_slot_lag {
+        let message = format!(
+            "Auction analysis is {:} slots behind the current slot",
+            slot_lag
+        );
+        alarm.fire(&message, &AlarmType::Telegram).await;
+    }
+    Ok(())
+}

src/phoenix/env.rs

@@ -33,6 +33,12 @@ pub struct AppConfig {
     pub unsynced_nodes_threshold_tg_warning: usize,
     #[serde(default = "default_unsynced_nodes_threshold_og_alert")]
     pub unsynced_nodes_threshold_og_alert: usize,
+    #[serde(default = "default_max_auction_analysis_slot_lag")]
+    pub max_auction_analysis_slot_lag: u32,
+}
+
+fn default_max_auction_analysis_slot_lag() -> u32 {
+    50
 }
 
 fn default_unsynced_nodes_threshold_og_alert() -> usize {

src/phoenix/mod.rs

@@ -1,4 +1,5 @@
 mod alerts;
+mod auction_analysis_monitor;
 mod checkpoint;
 mod consensus_node;
 mod demotion_monitor;
@@ -8,7 +9,9 @@ mod promotion_monitor;
 mod validation_node;
 
 use std::{
-    collections::HashMap, net::SocketAddr, sync::{Arc, Mutex}
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{Arc, Mutex},
 };
 
 use anyhow::{anyhow, Result};
@@ -32,6 +35,7 @@ use self::{
         telegram::{self, TelegramAlerts, TelegramSafeAlert},
         SendAlert,
     },
+    auction_analysis_monitor::run_auction_analysis_monitor,
     demotion_monitor::run_demotion_monitor,
     inclusion_monitor::{run_inclusion_monitor, LokiClient},
     promotion_monitor::run_promotion_monitor,
@@ -99,17 +103,16 @@ impl Alarm {
 }
 
 struct NodeAlarm {
-    alarm: Alarm
+    alarm: Alarm,
 }
 
 impl NodeAlarm {
     fn new() -> Self {
         Self {
-            alarm: Alarm::new()
+            alarm: Alarm::new(),
         }
     }
 
-
     async fn fire_age_over_limit(&mut self, name: &str) {
         let message = format!(
             "{} hasn't updated for more than {} seconds",
@@ -129,10 +132,8 @@ impl NodeAlarm {
             self.alarm.fire(&message, &AlarmType::Telegram).await;
         }
     }
-
 }
 
-
 struct Phoenix {
     name: &'static str,
     last_seen: DateTime<Utc>,
@@ -276,12 +277,14 @@ async fn run_ops_monitors() -> Result<()> {
     )
     .await?;
     let loki_client = LokiClient::new(APP_CONFIG.loki_url.clone());
+    let mut auction_analysis_alarm = Alarm::new();
 
     loop {
         let canonical_horizon = Utc::now() - Duration::minutes(APP_CONFIG.canonical_wait_minutes);
         run_demotion_monitor(&relay_pool, &mev_pool).await?;
         run_inclusion_monitor(&relay_pool, &mev_pool, &canonical_horizon, &loki_client).await?;
         run_promotion_monitor(&relay_pool, &mev_pool, &canonical_horizon).await?;
+        run_auction_analysis_monitor(&mev_pool, &mut auction_analysis_alarm).await?;
         tokio::time::sleep(Duration::minutes(1).to_std().unwrap()).await;
     }
 }