add support for unallocated service provider rewards

file_store/src/cli/dump.rs

@@ -230,6 +230,10 @@ impl Cmd {
                             "service_provider": reward.service_provider_id,
                             "amount": reward.amount,
                         }))?,
+                        Some(Reward::UnallocatedReward(reward)) => print_json(&json!({
+                            "unallocated_reward_type": reward.reward_type,
+                            "amount": reward.amount,
+                        }))?,
                         _ => (),
                     }
                 }

mobile_verifier/src/reward_shares.rs

@@ -10,6 +10,7 @@ use futures::{Stream, StreamExt};
 use helium_crypto::PublicKeyBinary;
 use helium_proto::services::{
     poc_mobile as proto, poc_mobile::mobile_reward_share::Reward as ProtoReward,
+    poc_mobile::UnallocatedReward, poc_mobile::UnallocatedRewardType,
 };
 use mobile_config::client::carrier_service_client::CarrierServiceVerifier;
 use mobile_config::client::ClientError;
@@ -262,7 +263,7 @@ impl ServiceProviderShares {
         self,
         reward_period: &'_ Range<DateTime<Utc>>,
         reward_per_share: Decimal,
-    ) -> impl Iterator<Item = proto::MobileRewardShare> + '_ {
+    ) -> impl Iterator<Item = (u64, proto::MobileRewardShare)> + '_ {
         self.shares
             .into_iter()
             .map(move |share| proto::ServiceProviderReward {
@@ -273,13 +274,36 @@ impl ServiceProviderShares {
                     .unwrap_or(0),
             })
             .filter(|service_provider_reward| service_provider_reward.amount > 0)
-            .map(|service_provider_reward| proto::MobileRewardShare {
-                start_period: reward_period.start.encode_timestamp(),
-                end_period: reward_period.end.encode_timestamp(),
-                reward: Some(ProtoReward::ServiceProviderReward(service_provider_reward)),
+            .map(|service_provider_reward| {
+                (
+                    service_provider_reward.amount,
+                    proto::MobileRewardShare {
+                        start_period: reward_period.start.encode_timestamp(),
+                        end_period: reward_period.end.encode_timestamp(),
+                        reward: Some(ProtoReward::ServiceProviderReward(service_provider_reward)),
+                    },
+                )
             })
     }
 
+    pub fn unallocated_reward(
+        unallocated_amount: Decimal,
+        reward_period: &'_ Range<DateTime<Utc>>,
+    ) -> anyhow::Result<proto::MobileRewardShare> {
+        let reward = UnallocatedReward {
+            reward_type: UnallocatedRewardType::ServiceProvider as i32,
+            amount: unallocated_amount
+                .round_dp_with_strategy(0, RoundingStrategy::ToZero)
+                .to_u64()
+                .unwrap_or(0),
+        };
+        Ok(proto::MobileRewardShare {
+            start_period: reward_period.start.encode_timestamp(),
+            end_period: reward_period.end.encode_timestamp(),
+            reward: Some(ProtoReward::UnallocatedReward(reward)),
+        })
+    }
+
     fn normalize_service_provider_rewards(
         total_sp_rewards_used: Decimal,
         total_sp_rewards: Decimal,
@@ -550,7 +574,9 @@ mod test {
             .unwrap();
 
         let mut sp_rewards = HashMap::new();
-        for sp_reward in sp_shares.into_service_provider_rewards(&epoch, rewards_per_share) {
+        for (_amount, sp_reward) in
+            sp_shares.into_service_provider_rewards(&epoch, rewards_per_share)
+        {
             if let Some(MobileReward::ServiceProviderReward(r)) = sp_reward.reward {
                 sp_rewards.insert(r.service_provider_id, r.amount);
             }
@@ -597,7 +623,9 @@ mod test {
             .unwrap();
 
         let mut sp_rewards = HashMap::new();
-        for sp_reward in sp_shares.into_service_provider_rewards(&epoch, rewards_per_share) {
+        for (_amount, sp_reward) in
+            sp_shares.into_service_provider_rewards(&epoch, rewards_per_share)
+        {
             if let Some(MobileReward::ServiceProviderReward(r)) = sp_reward.reward {
                 sp_rewards.insert(r.service_provider_id, r.amount);
             }
@@ -637,7 +665,9 @@ mod test {
             .unwrap();
 
         let mut sp_rewards = HashMap::new();
-        for sp_reward in sp_shares.into_service_provider_rewards(&epoch, rewards_per_share) {
+        for (_amount, sp_reward) in
+            sp_shares.into_service_provider_rewards(&epoch, rewards_per_share)
+        {
             if let Some(MobileReward::ServiceProviderReward(r)) = sp_reward.reward {
                 sp_rewards.insert(r.service_provider_id, r.amount);
             }
@@ -669,7 +699,9 @@ mod test {
             .unwrap();
 
         let mut sp_rewards = HashMap::new();
-        for sp_reward in sp_shares.into_service_provider_rewards(&epoch, rewards_per_share) {
+        for (_amount, sp_reward) in
+            sp_shares.into_service_provider_rewards(&epoch, rewards_per_share)
+        {
             if let Some(MobileReward::ServiceProviderReward(r)) = sp_reward.reward {
                 sp_rewards.insert(r.service_provider_id, r.amount);
             }

mobile_verifier/src/rewarder.rs

@@ -10,6 +10,7 @@ use anyhow::bail;
 use chrono::{DateTime, Duration, TimeZone, Utc};
 use db_store::meta;
 use file_store::{file_sink::FileSinkClient, traits::TimestampEncode};
+
 use helium_proto::RewardManifest;
 use mobile_config::client::{carrier_service_client::CarrierServiceVerifier, ClientError};
 use price::PriceTracker;
@@ -229,13 +230,28 @@ where
             .get_scheduled_tokens_for_service_providers(reward_period.end - reward_period.start);
         let rewards_per_share = sp_shares.rewards_per_share(total_sp_rewards, mobile_bone_price)?;
         // translate service provider shares into service provider rewards
-        for sp_share in sp_shares.into_service_provider_rewards(reward_period, rewards_per_share) {
+        // track the amount of allocated reward value as we go
+        let mut allocated_sp_rewards = 0_u64;
+        for (amount, sp_share) in
+            sp_shares.into_service_provider_rewards(reward_period, rewards_per_share)
+        {
+            allocated_sp_rewards += amount;
             self.mobile_rewards
                 .write(sp_share.clone(), [])
                 .await?
                 // Await the returned one shot to ensure that we wrote the file
                 .await??;
         }
+
+        // write out any unallocated service provider reward
+        let unallocated_sp_reward_amount = total_sp_rewards - Decimal::from(allocated_sp_rewards);
+        let unallocated_sp_reward =
+            ServiceProviderShares::unallocated_reward(unallocated_sp_reward_amount, reward_period)?;
+        self.mobile_rewards
+            .write(unallocated_sp_reward, [])
+            .await?
+            .await??;
+
         let written_files = self.mobile_rewards.commit().await?.await??;
 
         let mut transaction = self.pool.begin().await?;