Merge pull request #135 from AdExNetwork/issue-120

Issue #120 merge_aggrs & merge_payouts_into_balances

Author: samparsky

Cargo.lock

@@ -1,5 +1,5 @@
-use std::collections::HashMap;
+use std::collections::BTreeMap;
 
 use crate::BigNum;
 
-pub type BalancesMap = HashMap<String, BigNum>;
+pub type BalancesMap = BTreeMap<String, BigNum>;

primitives/src/balances_map.rs

@@ -5,8 +5,7 @@ use std::ops::{Add, AddAssign, Div, Mul, Sub};
 use std::str::FromStr;
 
 use num::rational::Ratio;
-use num::Integer;
-use num_bigint::BigUint;
+use num::{BigUint, CheckedSub, Integer};
 use num_derive::{Num, NumOps, One, Zero};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
@@ -151,6 +150,12 @@ impl<'a> Sum<&'a BigNum> for BigNum {
     }
 }
 
+impl CheckedSub for BigNum {
+    fn checked_sub(&self, v: &Self) -> Option<Self> {
+        self.0.checked_sub(&v.0).map(Self)
+    }
+}
+
 impl Mul<&Ratio<BigNum>> for &BigNum {
     type Output = BigNum;
 

primitives/src/big_num.rs

@@ -16,6 +16,7 @@ primitives = { path = "../primitives" }
 adapter = { version = "0.1", path = "../adapter" }
 chrono = { version = "0.4", features = ["serde"] }
 num = "0.2.0"
+num-traits = "0.2.0"
 # Futures
 futures-preview = { version = "=0.3.0-alpha.16", features = ["compat", "io-compat"] }
 futures_legacy = { version = "0.1", package = "futures" }

validator_worker/Cargo.toml

@@ -4,10 +4,10 @@ use chrono::{DateTime, Utc};
 use num_traits::CheckedSub;
 use serde::{Deserialize, Serialize};
 
-use domain::balances_map::get_balances_after_fees_tree;
-use domain::validator::message::Accounting;
-use domain::{BalancesMap, BigNum, Channel, ChannelId, DomainError};
-
+use crate::core::fees::get_balances_after_fees_tree;
+use primitives::sentry::{AggregateEvents, EventAggregate};
+use primitives::validator::Accounting;
+use primitives::{BalancesMap, BigNum, Channel, DomainError};
 
 #[allow(dead_code)]
 fn merge_aggrs(
@@ -25,7 +25,7 @@ fn merge_aggrs(
         .to_owned();
 
     // Build an intermediary balances representation
-    let mut balances_before_fees = accounting.pre_fees.clone();
+    let mut balances_before_fees = accounting.balances_before_fees.clone();
 
     // Merge in all the aggrs
     for aggr in aggregates {
@@ -38,7 +38,7 @@ fn merge_aggrs(
 
     let new_accounting = Accounting {
         last_event_aggregate,
-        pre_fees: balances_before_fees,
+        balances_before_fees,
         balances: balances.clone(),
     };
 
@@ -83,20 +83,29 @@ fn merge_payouts_into_balances<'a, T: Iterator<Item = &'a AggregateEvents>>(
 #[cfg(test)]
 mod test {
     use super::*;
-    use domain::channel::fixtures::{get_channel, get_channel_spec, ValidatorsOption};
-    use domain::fixtures::get_channel_id;
-    use domain::validator::fixtures::get_validator;
+    use primitives::util::tests::prep_db::{
+        DUMMY_CHANNEL, DUMMY_VALIDATOR_FOLLOWER, DUMMY_VALIDATOR_LEADER,
+    };
+    use primitives::{Channel, ChannelSpec, ValidatorDesc};
 
     #[test]
     fn should_merge_event_aggrs_and_apply_fees() {
         // fees: 100
-        // deposit: 10000
-        let leader = get_validator("one", Some(50.into()));
-        let follower = get_validator("two", Some(50.into()));
+        // deposit: 10 000
+        let leader = ValidatorDesc {
+            fee: 50.into(),
+            ..DUMMY_VALIDATOR_LEADER.clone()
+        };
+        let follower = ValidatorDesc {
+            fee: 50.into(),
+            ..DUMMY_VALIDATOR_FOLLOWER.clone()
+        };
 
-        let spec = get_channel_spec(ValidatorsOption::Pair { leader, follower });
-        let mut channel = get_channel("channel", &None, Some(spec));
-        channel.deposit_amount = 10_000.into();
+        let mut channel = Channel {
+            deposit_amount: 10_000.into(),
+            ..DUMMY_CHANNEL.clone()
+        };
+        channel.spec.validators = [leader, follower].into();
 
         let balances_before_fees: BalancesMap =
             vec![("a".to_string(), 100.into()), ("b".to_string(), 200.into())]
@@ -105,7 +114,7 @@ mod test {
 
         let acc = Accounting {
             last_event_aggregate: Utc::now(),
-            pre_fees: balances_before_fees,
+            balances_before_fees,
             balances: BalancesMap::default(),
         };
 
@@ -114,7 +123,7 @@ mod test {
 
         assert_eq!(balances, new_accounting.balances, "balances is the same");
         assert_eq!(
-            new_accounting.pre_fees["a"],
+            new_accounting.balances_before_fees["a"],
             150.into(),
             "balance of recipient incremented accordingly"
         );
@@ -127,12 +136,24 @@ mod test {
 
     #[test]
     fn should_never_allow_exceeding_the_deposit() {
-        let leader = get_validator("one", Some(50.into()));
-        let follower = get_validator("two", Some(50.into()));
+        let leader = ValidatorDesc {
+            fee: 50.into(),
+            ..DUMMY_VALIDATOR_LEADER.clone()
+        };
+        let follower = ValidatorDesc {
+            fee: 50.into(),
+            ..DUMMY_VALIDATOR_FOLLOWER.clone()
+        };
 
-        let spec = get_channel_spec(ValidatorsOption::Pair { leader, follower });
-        let mut channel = get_channel("channel", &None, Some(spec));
-        channel.deposit_amount = 10_000.into();
+        let spec = ChannelSpec {
+            validators: [leader, follower].into(),
+            ..DUMMY_CHANNEL.spec.clone()
+        };
+        let channel = Channel {
+            deposit_amount: 10_000.into(),
+            spec,
+            ..DUMMY_CHANNEL.clone()
+        };
 
         let balances_before_fees: BalancesMap =
             vec![("a".to_string(), 100.into()), ("b".to_string(), 200.into())]
@@ -141,7 +162,7 @@ mod test {
 
         let acc = Accounting {
             last_event_aggregate: Utc::now(),
-            pre_fees: balances_before_fees,
+            balances_before_fees,
             balances: BalancesMap::default(),
         };
 
@@ -150,12 +171,12 @@ mod test {
 
         assert_eq!(balances, new_accounting.balances, "balances is the same");
         assert_eq!(
-            new_accounting.pre_fees["a"],
+            new_accounting.balances_before_fees["a"],
             9_800.into(),
             "balance of recipient incremented accordingly"
         );
         assert_eq!(
-            new_accounting.pre_fees["b"],
+            new_accounting.balances_before_fees["b"],
             200.into(),
             "balances of non-recipient remains the same"
         );
@@ -165,7 +186,7 @@ mod test {
             "balanceAfterFees is ok"
         );
         assert_eq!(
-            &new_accounting.pre_fees.values().sum::<BigNum>(),
+            &new_accounting.balances_before_fees.values().sum::<BigNum>(),
             &channel.deposit_amount,
             "sum(balancesBeforeFees) == depositAmount"
         );
@@ -187,11 +208,11 @@ mod test {
         };
 
         EventAggregate {
-            channel_id: get_channel_id("one"),
+            channel_id: DUMMY_CHANNEL.id.to_owned(),
             created: Utc::now(),
             events: vec![("IMPRESSION".to_string(), aggregate_events)]
                 .into_iter()
                 .collect(),
         }
     }
-}
+}

validator_worker/src/core/events.rs

@@ -1,17 +1,13 @@
-use std::collections::BTreeMap;
-
 use num::rational::Ratio;
-use primitives::{BigNum, Channel, DomainError, ValidatorDesc};
-
-pub type BalancesMap = BTreeMap<String, BigNum>;
+use primitives::{BalancesMap, BigNum, Channel, DomainError, ValidatorDesc};
 
 pub fn get_balances_after_fees_tree(
     balances: &BalancesMap,
     channel: &Channel,
 ) -> Result<BalancesMap, DomainError> {
     let distribution = Distribution::new(balances, &channel)?;
 
-    let mut balances_after_fees = BTreeMap::default();
+    let mut balances_after_fees = BalancesMap::default();
     let mut total = BigNum::from(0);
 
     for (key, value) in balances.iter() {

validator_worker/src/core/fees.rs

@@ -13,6 +13,7 @@ pub use self::leader::Leader;
 pub use self::sentry_interface::all_channels;
 
 pub mod core {
+    pub mod events;
     pub mod fees;
     pub mod follower_rules;
 }