Reward granularity (#698)

* Initial work

* Add higher level of granularity to reward output

* Clippy

* ...clippy

* Move location_trust_score_multiplier into database

* Fix tests

* Update helium-proto

* Fmt

* Update proto branch to master

---------

Co-authored-by: Matthew Plant <[email protected]>

file_store/src/cli/dump.rs

@@ -205,7 +205,6 @@ impl Cmd {
                     print_json(&json!({
                         "cbsd_id": heartbeat.cbsd_id,
                         "pub_key": PublicKey::try_from(heartbeat.pub_key)?,
-                        "reward_multiplier": heartbeat.reward_multiplier,
                         "timestamp": heartbeat.timestamp,
                         "cell_type": heartbeat.cell_type,
                         "validity": heartbeat.validity,

file_store/src/heartbeat.rs

@@ -95,7 +95,6 @@ pub mod cli {
     pub struct ValidatedHeartbeat {
         pub cbsd_id: String,
         pub pub_key: PublicKeyBinary,
-        pub reward_multiplier: f32,
         pub timestamp: DateTime<Utc>,
         pub cell_type: CellType,
         pub validity: HeartbeatValidity,
@@ -113,7 +112,6 @@ pub mod cli {
             Ok(Self {
                 cbsd_id: v.cbsd_id.clone(),
                 pub_key: v.pub_key.clone().into(),
-                reward_multiplier: v.reward_multiplier,
                 timestamp: Utc
                     .timestamp_opt(v.timestamp as i64, 0)
                     .single()

mobile_verifier/migrations/24_location_trust_multiplier.sql

@@ -0,0 +1,18 @@
+ALTER TABLE wifi_heartbeats ADD COLUMN location_trust_score_multiplier DECIMAL;
+
+UPDATE wifi_heartbeats SET location_trust_score_multiplier =
+       CASE WHEN location_validation_timestamp IS NULL THEN
+       	    0.25
+       WHEN distance_to_asserted > 100 THEN
+       	    0.25
+       ELSE
+            1.0
+       END;
+
+ALTER TABLE wifi_heartbeats ALTER COLUMN location_trust_score_multiplier SET NOT NULL;
+
+ALTER TABLE cbrs_heartbeats ADD COLUMN location_trust_score_multiplier DECIMAL;
+
+UPDATE cbrs_heartbeats SET location_trust_score_multiplier = 1.0;
+
+ALTER TABLE cbrs_heartbeats ALTER COLUMN location_trust_score_multiplier SET NOT NULL;

mobile_verifier/src/cli/reward_from_db.rs

@@ -35,8 +35,7 @@ impl Cmd {
         let (shutdown_trigger, _shutdown_listener) = triggered::trigger();
         let pool = settings.database.connect(env!("CARGO_PKG_NAME")).await?;
 
-        let heartbeats =
-            HeartbeatReward::validated(&pool, &epoch, settings.max_asserted_distance_deviation);
+        let heartbeats = HeartbeatReward::validated(&pool, &epoch);
         let speedtest_averages =
             SpeedtestAverages::aggregate_epoch_averages(epoch.end, &pool).await?;
         let reward_shares =

mobile_verifier/src/cli/server.rs

@@ -114,6 +114,7 @@ impl Cmd {
             gateway_client.clone(),
             cbrs_heartbeats,
             settings.modeled_coverage_start(),
+            settings.max_asserted_distance_deviation,
             valid_heartbeats.clone(),
             seniority_updates.clone(),
         );
@@ -123,6 +124,7 @@ impl Cmd {
             gateway_client.clone(),
             wifi_heartbeats,
             settings.modeled_coverage_start(),
+            settings.max_asserted_distance_deviation,
             valid_heartbeats,
             seniority_updates,
         );
@@ -227,7 +229,6 @@ impl Cmd {
             mobile_rewards,
             reward_manifests,
             price_tracker,
-            settings.max_asserted_distance_deviation,
         );
 
         // subscriber location

mobile_verifier/src/data_session.rs

@@ -15,13 +15,19 @@ pub struct DataSessionIngestor {
     pub pool: PgPool,
 }
 
+#[derive(Default)]
+pub struct HotspotReward {
+    pub rewardable_bytes: u64,
+    pub rewardable_dc: u64,
+}
+
 #[derive(Clone, Debug)]
 pub struct ServiceProviderDataSession {
     pub service_provider: ServiceProvider,
     pub total_dcs: Decimal,
 }
 
-pub type HotspotMap = HashMap<PublicKeyBinary, u64>;
+pub type HotspotMap = HashMap<PublicKeyBinary, HotspotReward>;
 
 impl DataSessionIngestor {
     pub fn new(pool: sqlx::Pool<sqlx::Postgres>) -> Self {
@@ -180,7 +186,9 @@ pub async fn data_sessions_to_dc<'a>(
     tokio::pin!(stream);
     let mut map = HotspotMap::new();
     while let Some(session) = stream.try_next().await? {
-        *map.entry(session.pub_key).or_default() += session.num_dcs as u64
+        let rewards = map.entry(session.pub_key).or_default();
+        rewards.rewardable_dc += session.num_dcs as u64;
+        rewards.rewardable_bytes += session.upload_bytes as u64 + session.download_bytes as u64;
     }
     Ok(map)
 }

mobile_verifier/src/heartbeats/cbrs.rs

@@ -22,6 +22,7 @@ pub struct HeartbeatDaemon<GIR> {
     gateway_info_resolver: GIR,
     heartbeats: Receiver<FileInfoStream<CbrsHeartbeatIngestReport>>,
     modeled_coverage_start: DateTime<Utc>,
+    max_distance_to_asserted: u32,
     heartbeat_sink: FileSinkClient,
     seniority_sink: FileSinkClient,
 }
@@ -35,6 +36,7 @@ where
         gateway_info_resolver: GIR,
         heartbeats: Receiver<FileInfoStream<CbrsHeartbeatIngestReport>>,
         modeled_coverage_start: DateTime<Utc>,
+        max_distance_to_asserted: u32,
         heartbeat_sink: FileSinkClient,
         seniority_sink: FileSinkClient,
     ) -> Self {
@@ -43,6 +45,7 @@ where
             gateway_info_resolver,
             heartbeats,
             modeled_coverage_start,
+            max_distance_to_asserted,
             heartbeat_sink,
             seniority_sink,
         }
@@ -111,6 +114,7 @@ where
                 &self.gateway_info_resolver,
                 heartbeats,
                 coverage_objects,
+                self.max_distance_to_asserted,
                 &epoch,
             ),
             heartbeat_cache,