Merge pull request #493 from helium/map/add-dc-transfer-reward-msg

Add GatewayReward output from verifier

mobile_verifier/src/cli/generate.rs

@@ -50,7 +50,7 @@ impl Cmd {
         let mut total_rewards = 0_u64;
         let mut owner_rewards = HashMap::<_, u64>::new();
         let transfer_rewards = TransferRewards::empty();
-        for (reward, _) in reward_shares.into_rewards(&transfer_rewards, &epoch) {
+        for (reward, _) in reward_shares.into_rewards(transfer_rewards.reward_sum(), &epoch) {
             total_rewards += reward.amount;
             *owner_rewards
                 .entry(PublicKey::try_from(reward.owner_key)?)

mobile_verifier/src/cli/reward_from_db.rs

@@ -1,12 +1,13 @@
 use crate::{
     heartbeats::Heartbeats,
-    reward_shares::{get_scheduled_tokens_for_poc_and_dc, PocShares, TransferRewards},
+    reward_shares::{get_scheduled_tokens_for_poc_and_dc, PocShares},
     speedtests::{Average, SpeedtestAverages},
     Settings,
 };
 use anyhow::Result;
 use chrono::{DateTime, NaiveDateTime, Utc};
 use helium_crypto::PublicKey;
+use rust_decimal::Decimal;
 use serde_json::json;
 use std::collections::HashMap;
 
@@ -42,8 +43,7 @@ impl Cmd {
 
         let mut total_rewards = 0_u64;
         let mut owner_rewards = HashMap::<_, u64>::new();
-        let transfer_rewards = TransferRewards::empty();
-        for (reward, _) in reward_shares.into_rewards(&transfer_rewards, &epoch) {
+        for (reward, _) in reward_shares.into_rewards(Decimal::ZERO, &epoch) {
             total_rewards += reward.amount;
             *owner_rewards
                 .entry(PublicKey::try_from(reward.owner_key)?)

mobile_verifier/src/reward_shares.rs

@@ -44,7 +44,12 @@ impl TransferRewards {
         self.reward_scale
     }
 
-    pub fn reward(&self, hotspot: &PublicKeyBinary) -> Decimal {
+    pub fn reward_sum(&self) -> Decimal {
+        self.reward_sum
+    }
+
+    #[cfg(test)]
+    fn reward(&self, hotspot: &PublicKeyBinary) -> Decimal {
         self.rewards.get(hotspot).copied().unwrap_or(Decimal::ZERO) * self.reward_scale
     }
 
@@ -107,6 +112,34 @@ impl TransferRewards {
             reward_sum: reward_sum * reward_scale,
         }
     }
+
+    pub fn into_rewards(
+        self,
+        epoch: &'_ Range<DateTime<Utc>>,
+    ) -> impl Iterator<Item = proto::MobileRewardShare> + '_ {
+        let Self {
+            reward_scale,
+            rewards,
+            ..
+        } = self;
+        let start_period = epoch.start.encode_timestamp();
+        let end_period = epoch.end.encode_timestamp();
+        rewards
+            .into_iter()
+            .map(move |(hotspot_key, reward)| proto::MobileRewardShare {
+                start_period,
+                end_period,
+                reward: Some(proto::mobile_reward_share::Reward::GatewayReward(
+                    proto::GatewayReward {
+                        hotspot_key: hotspot_key.into(),
+                        dc_transfer_reward: (reward * reward_scale)
+                            .round_dp_with_strategy(0, RoundingStrategy::ToZero)
+                            .to_u64()
+                            .unwrap_or(0),
+                    },
+                )),
+            })
+    }
 }
 
 /// Returns the equivalent amount of Mobile bones for a specified amount of Data Credits
@@ -169,30 +202,28 @@ impl PocShares {
             })
     }
 
-    pub fn into_rewards<'a>(
+    pub fn into_rewards(
         self,
-        transfer_rewards: &'a TransferRewards,
-        epoch: &'a Range<DateTime<Utc>>,
-    ) -> impl Iterator<Item = (proto::RadioRewardShare, proto::MobileRewardShare)> + 'a {
+        transfer_rewards_sum: Decimal,
+        epoch: &'_ Range<DateTime<Utc>>,
+    ) -> impl Iterator<Item = (proto::RadioRewardShare, proto::MobileRewardShare)> + '_ {
         let total_shares = self.total_shares();
-        let available_poc_rewards = get_scheduled_tokens_for_poc_and_dc(epoch.end - epoch.start)
-            - transfer_rewards.reward_sum;
+        let available_poc_rewards =
+            get_scheduled_tokens_for_poc_and_dc(epoch.end - epoch.start) - transfer_rewards_sum;
         let poc_rewards_per_share = available_poc_rewards / total_shares;
+        let start_period = epoch.start.encode_timestamp();
+        let end_period = epoch.end.encode_timestamp();
         self.hotspot_shares.into_iter().flat_map(
             move |(hotspot_key, RadioShares { radio_shares })| {
-                let mut dc_transfer_reward = Some(transfer_rewards.reward(&hotspot_key));
                 radio_shares
                     .into_iter()
                     .map(move |(cbsd_id, amount)| {
-                        let start_period = epoch.start.encode_timestamp();
-                        let end_period = epoch.end.encode_timestamp();
                         let poc_reward = poc_rewards_per_share * amount;
-                        let dc_transfer_reward = dc_transfer_reward.take().unwrap_or(Decimal::ZERO);
                         let hotspot_key: Vec<u8> = hotspot_key.clone().into();
                         let radio_reward_share = proto::RadioRewardShare {
                             owner_key: Vec::new(),
                             cbsd_id: cbsd_id.clone(),
-                            amount: (poc_reward + dc_transfer_reward)
+                            amount: poc_reward
                                 .round_dp_with_strategy(0, RoundingStrategy::ToZero)
                                 .to_u64()
                                 .unwrap_or(0),
@@ -207,14 +238,11 @@ impl PocShares {
                                 proto::RadioReward {
                                     hotspot_key,
                                     cbsd_id,
-                                    dc_transfer_reward: dc_transfer_reward
-                                        .round_dp_with_strategy(0, RoundingStrategy::ToZero)
-                                        .to_u64()
-                                        .unwrap_or(0),
                                     poc_reward: poc_reward
                                         .round_dp_with_strategy(0, RoundingStrategy::ToZero)
                                         .to_u64()
                                         .unwrap_or(0),
+                                    ..Default::default()
                                 },
                             )),
                         };
@@ -724,10 +752,9 @@ mod test {
         // calculate the rewards for the sample group
         let mut owner_rewards = HashMap::<PublicKeyBinary, u64>::new();
         let epoch = (now - Duration::hours(1))..now;
-        let transfer_rewards = TransferRewards::empty();
         for (_, mobile_reward) in PocShares::aggregate(heartbeats, speedtest_avgs)
             .await
-            .into_rewards(&transfer_rewards, &epoch)
+            .into_rewards(Decimal::ZERO, &epoch)
         {
             let radio_reward = match mobile_reward.reward {
                 Some(proto::mobile_reward_share::Reward::RadioReward(radio_reward)) => radio_reward,
@@ -809,7 +836,7 @@ mod test {
         let epoch = now - Duration::hours(1)..now;
         let transfer_rewards = TransferRewards::empty();
         let expected_hotspot = gw1;
-        for (_, mobile_reward) in owner_shares.into_rewards(&transfer_rewards, &epoch) {
+        for (_, mobile_reward) in owner_shares.into_rewards(transfer_rewards.reward_sum(), &epoch) {
             let radio_reward = match mobile_reward.reward {
                 Some(proto::mobile_reward_share::Reward::RadioReward(radio_reward)) => radio_reward,
                 _ => unreachable!(),

mobile_verifier/src/verifier.rs

@@ -154,7 +154,7 @@ impl VerifierDaemon {
         metrics::gauge!("data_transfer_rewards_scale", scale);
 
         for (radio_reward_share, mobile_reward_share) in
-            poc_rewards.into_rewards(&transfer_rewards, &scheduler.reward_period)
+            poc_rewards.into_rewards(transfer_rewards.reward_sum(), &scheduler.reward_period)
         {
             self.radio_rewards
                 .write(radio_reward_share, [])
@@ -168,6 +168,14 @@ impl VerifierDaemon {
                 .await??;
         }
 
+        for mobile_reward_share in transfer_rewards.into_rewards(&scheduler.reward_period) {
+            self.mobile_rewards
+                .write(mobile_reward_share, [])
+                .await?
+                // Await the returned one shot to ensure that we wrote the file
+                .await??;
+        }
+
         let written_files = self.mobile_rewards.commit().await?.await??;
 
         let mut transaction = self.pool.begin().await?;

reward_index/src/indexer.rs

@@ -25,6 +25,7 @@ pub struct Indexer {
 #[derive(sqlx::Type, Debug, Clone, PartialEq, Eq, Hash)]
 #[sqlx(type_name = "reward_type", rename_all = "snake_case")]
 pub enum RewardType {
+    MobileRadio,
     MobileGateway,
     IotGateway,
     IotOperational,
@@ -126,11 +127,18 @@ impl Indexer {
                 let share = MobileRewardShare::decode(msg)?;
                 match share.reward {
                     Some(MobileReward::RadioReward(r)) => Ok((
+                        RewardKey {
+                            key: PublicKeyBinary::from(r.hotspot_key).to_string(),
+                            reward_type: RewardType::MobileRadio,
+                        },
+                        r.poc_reward,
+                    )),
+                    Some(MobileReward::GatewayReward(r)) => Ok((
                         RewardKey {
                             key: PublicKeyBinary::from(r.hotspot_key).to_string(),
                             reward_type: RewardType::MobileGateway,
                         },
-                        r.dc_transfer_reward + r.poc_reward,
+                        r.dc_transfer_reward,
                     )),
                     _ => bail!("got an invalid mobile reward share"),
                 }