rejig clear tables after rewarding

mobile_verifier/src/heartbeats.rs

@@ -287,3 +287,14 @@ async fn validate_heartbeat(
 
     Ok((cell_type, proto::HeartbeatValidity::Valid))
 }
+
+pub async fn clear_heartbeats(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    reward_period: &Range<DateTime<Utc>>,
+) -> Result<(), sqlx::Error> {
+    sqlx::query("DELETE FROM heartbeats WHERE truncated_timestamp < $1")
+        .bind(reward_period.start)
+        .execute(&mut *tx)
+        .await?;
+    Ok(())
+}

mobile_verifier/src/rewarder.rs

@@ -1,8 +1,9 @@
 use crate::{
     data_session,
-    heartbeats::HeartbeatReward,
+    heartbeats::{self, HeartbeatReward},
     reward_shares::{MapperShares, PocShares, TransferRewards},
-    speedtests_average::{SpeedtestAverages, SPEEDTEST_LAPSE},
+    speedtests,
+    speedtests_average::SpeedtestAverages,
     subscriber_location, telemetry,
 };
 use anyhow::bail;
@@ -229,21 +230,9 @@ impl Rewarder {
 
         let mut transaction = self.pool.begin().await?;
 
-        // Clear the heartbeats table of old heartbeats:
-        sqlx::query("DELETE FROM heartbeats WHERE truncated_timestamp < $1")
-            .bind(reward_period.start)
-            .execute(&mut transaction)
-            .await?;
-
-        // Clear the speedtests table of tests older than hours defined by SPEEDTEST_LAPSE
-        // We end up with tests older than we need here but erroring on side of caution
-        // and the volume of tests is low so no big impact there
-        sqlx::query("DELETE FROM speedtests where timestamp < $1")
-            .bind(Utc::now() - Duration::hours(SPEEDTEST_LAPSE))
-            .execute(&mut transaction)
-            .await?;
-
-        // clear the db of data sessions data & subscriber location data for the epoch
+        // clear out the various db tables
+        heartbeats::clear_heartbeats(&mut transaction, reward_period).await?;
+        speedtests::clear_speedtests(&mut transaction).await?;
         data_session::clear_hotspot_data_sessions(&mut transaction, reward_period).await?;
         subscriber_location::clear_location_shares(&mut transaction, reward_period).await?;
 

mobile_verifier/src/speedtests.rs

@@ -1,5 +1,5 @@
-use crate::speedtests_average::SpeedtestAverage;
-use chrono::{DateTime, Utc};
+use crate::speedtests_average::{SpeedtestAverage, SPEEDTEST_LAPSE};
+use chrono::{DateTime, Duration, Utc};
 use file_store::{
     file_info_poller::FileInfoStream, file_sink::FileSinkClient,
     speedtest::CellSpeedtestIngestReport,
@@ -190,3 +190,14 @@ pub async fn save_speedtest_to_db(
     .await?;
     Ok(())
 }
+
+// Clear the speedtests table of tests older than hours defined by SPEEDTEST_LAPSE
+pub async fn clear_speedtests(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+) -> Result<(), sqlx::Error> {
+    sqlx::query("DELETE FROM speedtests where timestamp < $1")
+        .bind(Utc::now() - Duration::hours(SPEEDTEST_LAPSE))
+        .execute(&mut *tx)
+        .await?;
+    Ok(())
+}