feat(phoenix): Add lookback_update_monitor

src/phoenix/env.rs

@@ -37,6 +37,14 @@ pub struct AppConfig {
     pub max_auction_analysis_slot_lag: u32,
     #[serde(default = "default_max_header_delay_updates_slot_lag")]
     pub max_header_delay_updates_slot_lag: u32,
+    #[serde(default = "default_max_lookback_updates_slot_lag")]
+    pub max_lookback_updates_slot_lag: u32,
+}
+
+fn default_max_lookback_updates_slot_lag() -> u32 {
+    // Note that this is more heuristic: since theoretically there could just be no updates because
+    // of the way the update algorithm works
+    600
 }
 
 fn default_max_header_delay_updates_slot_lag() -> u32 {

src/phoenix/lookback_update_monitor.rs

@@ -0,0 +1,43 @@
+use anyhow::Result;
+use sqlx::PgPool;
+use tracing::debug;
+
+use super::util::get_current_slot;
+use crate::phoenix::{Alarm, AlarmType};
+
+use super::env::APP_CONFIG;
+
+async fn get_latest_lookback_updates_slot(mev_pool: &PgPool) -> anyhow::Result<u32> {
+    sqlx::query_scalar!(
+        r#"
+        SELECT MAX(slot)
+        FROM lookback_updates
+        "#,
+    )
+    .fetch_one(mev_pool)
+    .await
+    .map(|max| {
+        max.expect("No maximum slot found from lookback_updates")
+            .try_into()
+            .expect("Maximum slot is negative")
+    })
+    .map_err(Into::into)
+}
+
+pub async fn run_lookback_updates_monitor(mev_pool: &PgPool, alarm: &mut Alarm) -> Result<()> {
+    let latest_slot = get_latest_lookback_updates_slot(mev_pool).await?;
+    let current_slot = get_current_slot()?;
+    let slot_lag = current_slot - latest_slot;
+    debug!(
+        "lookback_updates is {:} slots behind current slot",
+        slot_lag
+    );
+    if slot_lag > APP_CONFIG.max_lookback_updates_slot_lag {
+        let message = format!(
+            "lookback_updates is {:} slots behind the current slot",
+            slot_lag
+        );
+        alarm.fire(&message, &AlarmType::Telegram).await;
+    }
+    Ok(())
+}

src/phoenix/mod.rs

@@ -6,6 +6,7 @@ mod delay_update_monitor;
 mod demotion_monitor;
 mod env;
 mod inclusion_monitor;
+mod lookback_update_monitor;
 mod promotion_monitor;
 mod util;
 mod validation_node;
@@ -41,6 +42,7 @@ use self::{
     delay_update_monitor::run_header_delay_updates_monitor,
     demotion_monitor::run_demotion_monitor,
     inclusion_monitor::{run_inclusion_monitor, LokiClient},
+    lookback_update_monitor::run_lookback_updates_monitor,
     promotion_monitor::run_promotion_monitor,
 };
 
@@ -280,15 +282,20 @@ async fn run_ops_monitors() -> Result<()> {
     )
     .await?;
     let loki_client = LokiClient::new(APP_CONFIG.loki_url.clone());
+
+    // Separate alarm instances mean throtteling will be applied separately
     let mut auction_analysis_alarm = Alarm::new();
+    let mut header_delay_updates_alarm = Alarm::new();
+    let mut run_lookback_updates_monitor_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?;
-        run_header_delay_updates_monitor(&mev_pool, &mut auction_analysis_alarm).await?;
+        run_header_delay_updates_monitor(&mev_pool, &mut header_delay_updates_alarm).await?;
+        run_lookback_updates_monitor(&mev_pool, &mut run_lookback_updates_monitor_alarm).await?;
         tokio::time::sleep(Duration::minutes(1).to_std().unwrap()).await;
     }
 }