Update mobile verifier to record last_reward_end_time metric on boot

mobile_verifier/src/cli/server.rs

@@ -1,6 +1,7 @@
 use crate::{
     data_session::DataSessionIngestor, heartbeats::HeartbeatDaemon, rewarder::Rewarder,
-    speedtests::SpeedtestDaemon, subscriber_location::SubscriberLocationIngestor, Settings,
+    speedtests::SpeedtestDaemon, subscriber_location::SubscriberLocationIngestor, telemetry,
+    Settings,
 };
 use anyhow::{Error, Result};
 use chrono::Duration;
@@ -38,6 +39,8 @@ impl Cmd {
             .await?;
         sqlx::migrate!().run(&pool).await?;
 
+        telemetry::initialize(&pool).await?;
+
         let (file_upload_tx, file_upload_rx) = file_upload::message_channel();
         let file_upload =
             file_upload::FileUpload::from_settings(&settings.output, file_upload_rx).await?;

mobile_verifier/src/heartbeats.rs

@@ -214,7 +214,7 @@ impl Heartbeat {
                     timestamp: self.timestamp.timestamp() as u64,
                     coverage_object: Vec::with_capacity(0), // Placeholder so the project compiles
                 },
-                [],
+                &[("validity", self.validity.as_str_name())],
             )
             .await?;
         Ok(())

mobile_verifier/src/lib.rs

@@ -5,6 +5,7 @@ mod reward_shares;
 mod settings;
 mod speedtests;
 mod subscriber_location;
+mod telemetry;
 
 pub mod cli;
 pub mod rewarder;

mobile_verifier/src/rewarder.rs

@@ -3,7 +3,7 @@ use crate::{
     heartbeats::HeartbeatReward,
     reward_shares::{MapperShares, PocShares, TransferRewards},
     speedtests::SpeedtestAverages,
-    subscriber_location,
+    subscriber_location, telemetry,
 };
 use anyhow::bail;
 use chrono::{DateTime, Duration, TimeZone, Utc};
@@ -54,8 +54,8 @@ impl Rewarder {
 
     pub async fn run(self, shutdown: triggered::Listener) -> anyhow::Result<()> {
         loop {
-            let last_rewarded_end_time = self.last_rewarded_end_time().await?;
-            let next_rewarded_end_time = self.next_rewarded_end_time().await?;
+            let last_rewarded_end_time = last_rewarded_end_time(&self.pool).await?;
+            let next_rewarded_end_time = next_rewarded_end_time(&self.pool).await?;
             let scheduler = Scheduler::new(
                 self.reward_period_duration,
                 last_rewarded_end_time,
@@ -88,18 +88,6 @@ impl Rewarder {
         Ok(())
     }
 
-    async fn last_rewarded_end_time(&self) -> db_store::Result<DateTime<Utc>> {
-        Utc.timestamp_opt(meta::fetch(&self.pool, "last_rewarded_end_time").await?, 0)
-            .single()
-            .ok_or(db_store::Error::DecodeError)
-    }
-
-    async fn next_rewarded_end_time(&self) -> db_store::Result<DateTime<Utc>> {
-        Utc.timestamp_opt(meta::fetch(&self.pool, "next_rewarded_end_time").await?, 0)
-            .single()
-            .ok_or(db_store::Error::DecodeError)
-    }
-
     async fn disable_complete_data_checks_until(&self) -> db_store::Result<DateTime<Utc>> {
         Utc.timestamp_opt(
             meta::fetch(&self.pool, "disable_complete_data_checks_until").await?,
@@ -180,7 +168,7 @@ impl Rewarder {
         let Some(scale) = transfer_rewards.reward_scale().to_f64() else {
             bail!("The data transfer rewards scale cannot be converted to a float");
         };
-        metrics::gauge!("data_transfer_rewards_scale", scale);
+        telemetry::data_transfer_rewards_scale(scale);
 
         for mobile_reward_share in
             poc_rewards.into_rewards(transfer_rewards.reward_sum(), reward_period)
@@ -270,14 +258,23 @@ impl Rewarder {
             .await??;
 
         self.reward_manifests.commit().await?;
-        metrics::gauge!(
-            "last_rewarded_end_time",
-            next_reward_period.start.timestamp() as f64
-        );
+        telemetry::last_rewarded_end_time(next_reward_period.start);
         Ok(())
     }
 }
 
+pub async fn last_rewarded_end_time(db: &Pool<Postgres>) -> db_store::Result<DateTime<Utc>> {
+    Utc.timestamp_opt(meta::fetch(db, "last_rewarded_end_time").await?, 0)
+        .single()
+        .ok_or(db_store::Error::DecodeError)
+}
+
+async fn next_rewarded_end_time(db: &Pool<Postgres>) -> db_store::Result<DateTime<Utc>> {
+    Utc.timestamp_opt(meta::fetch(db, "next_rewarded_end_time").await?, 0)
+        .single()
+        .ok_or(db_store::Error::DecodeError)
+}
+
 async fn save_last_rewarded_end_time(
     exec: impl PgExecutor<'_>,
     value: &DateTime<Utc>,

mobile_verifier/src/speedtests.rs

@@ -243,7 +243,7 @@ impl SpeedtestRollingAverage {
     pub async fn write(&self, averages: &file_sink::FileSinkClient) -> file_store::Result {
         // Write out the speedtests to S3
         let average = Average::from(&self.speedtests);
-        let validity = average.validity() as i32;
+        let validity = average.validity();
         // this is guaratneed to safely convert and not panic as it can only be one of
         // four possible decimal values based on the speedtest average tier
         let reward_multiplier = average.reward_multiplier().try_into().unwrap();
@@ -272,10 +272,10 @@ impl SpeedtestRollingAverage {
                         latency_ms: st.latency as u32,
                     })
                     .collect(),
-                    validity,
+                    validity: validity as i32,
                     reward_multiplier,
                 },
-                [],
+                &[("validity", validity.as_str_name())],
             )
             .await?;
 

mobile_verifier/src/telemetry.rs

@@ -0,0 +1,21 @@
+use chrono::{DateTime, Utc};
+use sqlx::{Pool, Postgres};
+
+use crate::rewarder;
+
+const LAST_REWARDED_END_TIME: &str = "last_rewarded_end_time";
+const DATA_TRANSFER_REWARDS_SCALE: &str = "data_transfer_rewards_scale";
+
+pub async fn initialize(db: &Pool<Postgres>) -> anyhow::Result<()> {
+    last_rewarded_end_time(rewarder::last_rewarded_end_time(db).await?);
+
+    Ok(())
+}
+
+pub fn last_rewarded_end_time(timestamp: DateTime<Utc>) {
+    metrics::gauge!(LAST_REWARDED_END_TIME, timestamp.timestamp() as f64);
+}
+
+pub fn data_transfer_rewards_scale(scale: f64) {
+    metrics::gauge!(DATA_TRANSFER_REWARDS_SCALE, scale);
+}