Merge pull request #289 from samparsky/couple-fix-original-message

Couple of fixes PR #277; original message

Author: elpiel

adapter/src/ethereum.rs

@@ -536,7 +536,7 @@ mod test {
                 .expect("Valid ValidatorId"),
             payload: Payload {
                 id: "awesomeValidator".to_string(),
-                era: 100000,
+                era: 100_000,
                 address: "0x2bDeAFAE53940669DaA6F519373f686c1f3d3393".to_string(),
                 identity: None,
             },

adapter/src/ethereum/error.rs

@@ -66,8 +66,8 @@ impl fmt::Display for VerifyError {
             PublicKeyRecovery(err) => {
                 write!(f, "Recovering the public key from the signature: {}", err)
             }
-            StateRootDecoding(err) => write!(f, "Decoding the hex of the state root: {}", err),
-            SignatureDecoding(err) => write!(f, "Decoding the hex of the signature: {}", err),
+            StateRootDecoding(err) => write!(f, "Decoding state root: {}", err),
+            SignatureDecoding(err) => write!(f, "Decoding signature: {}", err),
             SignatureNotPrefixed => write!(f, "Signature is not prefixed with `0x`"),
         }
     }

primitives/src/adapter.rs

@@ -37,7 +37,7 @@ impl<AE: AdapterErrorKind> fmt::Display for Error<AE> {
             Error::Authentication(error) => write!(f, "Authentication: {}", error),
             Error::Authorization(error) => write!(f, "Authorization: {}", error),
             Error::InvalidChannel(error) => write!(f, "{}", error),
-            Error::Adapter(error) => write!(f, "Adapter specific: {}", error),
+            Error::Adapter(error) => write!(f, "Adapter: {}", error),
             Error::Domain(error) => write!(f, "Domain: {}", error),
             Error::LockedWallet => write!(f, "You must `.unlock()` the wallet first"),
         }

validator_worker/src/core/events.rs

@@ -10,7 +10,7 @@ pub(crate) fn merge_aggrs(
     accounting: &Accounting,
     aggregates: &[EventAggregate],
     channel: &Channel,
-) -> Result<(BalancesMap, Accounting), DomainError> {
+) -> Result<Accounting, DomainError> {
     let deposit = channel.deposit_amount.clone();
 
     let last_event_aggregate = [accounting.last_event_aggregate]
@@ -35,10 +35,10 @@ pub(crate) fn merge_aggrs(
     let new_accounting = Accounting {
         last_event_aggregate,
         balances_before_fees,
-        balances: balances.clone(),
+        balances,
     };
 
-    Ok((balances, new_accounting))
+    Ok(new_accounting)
 }
 
 fn merge_payouts_into_balances<'a, T: Iterator<Item = &'a AggregateEvents>>(
@@ -115,11 +115,9 @@ mod test {
             balances: BalancesMap::default(),
         };
 
-        let (balances, new_accounting) =
-            merge_aggrs(&acc, &[gen_ev_aggr(5, &IDS["publisher"])], &channel)
-                .expect("Something went wrong");
+        let new_accounting = merge_aggrs(&acc, &[gen_ev_aggr(5, &IDS["publisher"])], &channel)
+            .expect("Something went wrong");
 
-        assert_eq!(balances, new_accounting.balances, "balances is the same");
         assert_eq!(
             new_accounting.balances_before_fees[&IDS["publisher"]],
             150.into(),
@@ -166,11 +164,9 @@ mod test {
             balances: BalancesMap::default(),
         };
 
-        let (balances, new_accounting) =
-            merge_aggrs(&acc, &[gen_ev_aggr(1_001, &IDS["publisher"])], &channel)
-                .expect("Something went wrong");
+        let new_accounting = merge_aggrs(&acc, &[gen_ev_aggr(1_001, &IDS["publisher"])], &channel)
+            .expect("Something went wrong");
 
-        assert_eq!(balances, new_accounting.balances, "balances is the same");
         assert_eq!(
             new_accounting.balances_before_fees[&IDS["publisher"]],
             9_800.into(),

validator_worker/src/error.rs

@@ -11,7 +11,7 @@ pub enum TickError {
 impl fmt::Display for TickError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            TickError::TimedOut(err) => write!(f, "Tick timed out ({})", err),
+            TickError::TimedOut(err) => write!(f, "Tick TimedOut: ({})", err),
             TickError::Tick(err) => write!(f, "Tick: {}", err),
         }
     }
@@ -31,13 +31,9 @@ impl<AE: AdapterErrorKind> fmt::Display for Error<AE> {
         use Error::*;
 
         match self {
-            SentryApi(err) => write!(f, "Sentry Api: {}", err),
-            LeaderTick(channel_id, err) => {
-                write!(f, "Error for Leader tick of {:#?}: {}", channel_id, err)
-            }
-            FollowerTick(channel_id, err) => {
-                write!(f, "Error for Follower tick of {:#?}: {}", channel_id, err)
-            }
+            SentryApi(err) => write!(f, "SentryApi: {}", err),
+            LeaderTick(channel_id, err) => write!(f, "LeaderTick {:#?}: {}", channel_id, err),
+            FollowerTick(channel_id, err) => write!(f, "FollowerTick {:#?}: {}", channel_id, err),
         }
     }
 }

validator_worker/src/follower.rs

@@ -36,12 +36,12 @@ impl fmt::Display for InvalidNewState {
 
 #[derive(Debug)]
 pub enum ApproveStateResult<AE: AdapterErrorKind> {
-    Sent(Vec<PropagationResult<AE>>),
-    /// Conditions for handling the new state haven't been met
-    NotSent,
+    /// If None, Conditions for handling the new state haven't been met
+    Sent(Option<Vec<PropagationResult<AE>>>),
     RejectedState {
         reason: InvalidNewState,
-        reject_state_propagation: Vec<PropagationResult<AE>>,
+        state_root: String,
+        propagation: Vec<PropagationResult<AE>>,
     },
 }
 
@@ -79,13 +79,13 @@ pub async fn tick<A: Adapter + 'static>(
 
     let producer_tick = producer::tick(&iface).await?;
     let balances = match &producer_tick {
-        producer::TickStatus::Sent { balances, .. } => balances,
+        producer::TickStatus::Sent { new_accounting, .. } => &new_accounting.balances,
         producer::TickStatus::NoNewEventAggr(balances) => balances,
     };
     let approve_state_result = if let (Some(new_state), false) = (new_msg, latest_is_responded_to) {
         on_new_state(&iface, &balances, &new_state).await?
     } else {
-        ApproveStateResult::NotSent
+        ApproveStateResult::Sent(None)
     };
 
     Ok(TickStatus {
@@ -147,15 +147,15 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         })])
         .await;
 
-    Ok(ApproveStateResult::Sent(propagation_result))
+    Ok(ApproveStateResult::Sent(Some(propagation_result)))
 }
 
 async fn on_error<'a, A: Adapter + 'static>(
     iface: &'a SentryApi<A>,
     new_state: &'a NewState,
     status: InvalidNewState,
 ) -> ApproveStateResult<A::AdapterError> {
-    let reject_state_propagation = iface
+    let propagation = iface
         .propagate(&[&MessageTypes::RejectState(RejectState {
             reason: status.to_string(),
             state_root: new_state.state_root.clone(),
@@ -168,6 +168,7 @@ async fn on_error<'a, A: Adapter + 'static>(
 
     ApproveStateResult::RejectedState {
         reason: status,
-        reject_state_propagation,
+        state_root: new_state.state_root.clone(),
+        propagation,
     }
 }

validator_worker/src/heartbeat.rs

@@ -5,18 +5,14 @@ use chrono::{Duration, Utc};
 
 use adapter::get_signable_state_root;
 use byteorder::{BigEndian, ByteOrder};
-use primitives::adapter::{Adapter, AdapterErrorKind};
+use primitives::adapter::Adapter;
 use primitives::merkle_tree::MerkleTree;
 use primitives::validator::{Heartbeat, MessageTypes};
 use primitives::{BalancesMap, BigNum, Channel};
 
 use crate::sentry_interface::{PropagationResult, SentryApi};
 
-#[derive(Debug)]
-pub enum HeartbeatStatus<AE: AdapterErrorKind> {
-    Sent(Vec<PropagationResult<AE>>),
-    NotSent,
-}
+pub type HeartbeatStatus<A> = Option<Vec<PropagationResult<A>>>;
 
 async fn send_heartbeat<A: Adapter + 'static>(
     iface: &SentryApi<A>,
@@ -59,9 +55,9 @@ pub async fn heartbeat<A: Adapter + 'static>(
     });
 
     if should_send {
-        Ok(HeartbeatStatus::Sent(send_heartbeat(&iface).await?))
+        Ok(Some(send_heartbeat(&iface).await?))
     } else {
-        Ok(HeartbeatStatus::NotSent)
+        Ok(None)
     }
 }
 

validator_worker/src/leader.rs

@@ -2,24 +2,16 @@ use std::error::Error;
 
 use primitives::adapter::{Adapter, AdapterErrorKind};
 use primitives::validator::{Accounting, MessageTypes, NewState};
-use primitives::BalancesMap;
 
 use crate::heartbeat::{heartbeat, HeartbeatStatus};
 use crate::sentry_interface::{PropagationResult, SentryApi};
 use crate::{get_state_root_hash, producer};
 
-#[derive(Debug)]
-pub enum NewStateResult<AE: AdapterErrorKind> {
-    Sent(Vec<PropagationResult<AE>>),
-    /// Conditions for sending the new state haven't been met
-    NotSent,
-}
-
 #[derive(Debug)]
 pub struct TickStatus<AE: AdapterErrorKind> {
     pub heartbeat: HeartbeatStatus<AE>,
     /// If None, then the conditions for handling a new state haven't been met
-    pub new_state: NewStateResult<AE>,
+    pub new_state: Option<Vec<PropagationResult<AE>>>,
     pub producer_tick: producer::TickStatus<AE>,
 }
 
@@ -28,15 +20,11 @@ pub async fn tick<A: Adapter + 'static>(
 ) -> Result<TickStatus<A::AdapterError>, Box<dyn Error>> {
     let producer_tick = producer::tick(&iface).await?;
     let (balances, new_state) = match &producer_tick {
-        producer::TickStatus::Sent {
-            balances,
-            new_accounting,
-            ..
-        } => {
-            let new_state = on_new_accounting(&iface, (balances, new_accounting)).await?;
-            (balances, new_state)
+        producer::TickStatus::Sent { new_accounting, .. } => {
+            let new_state = on_new_accounting(&iface, new_accounting).await?;
+            (&new_accounting.balances, Some(new_state))
         }
-        producer::TickStatus::NoNewEventAggr(balances) => (balances, NewStateResult::NotSent),
+        producer::TickStatus::NoNewEventAggr(balances) => (balances, None),
     };
 
     Ok(TickStatus {
@@ -48,9 +36,9 @@ pub async fn tick<A: Adapter + 'static>(
 
 async fn on_new_accounting<A: Adapter + 'static>(
     iface: &SentryApi<A>,
-    (balances, new_accounting): (&BalancesMap, &Accounting),
-) -> Result<NewStateResult<A::AdapterError>, Box<dyn Error>> {
-    let state_root_raw = get_state_root_hash(&iface, &balances)?;
+    new_accounting: &Accounting,
+) -> Result<Vec<PropagationResult<A::AdapterError>>, Box<dyn Error>> {
+    let state_root_raw = get_state_root_hash(&iface, &new_accounting.balances)?;
     let state_root = hex::encode(state_root_raw);
 
     let signature = iface.adapter.sign(&state_root)?;
@@ -63,5 +51,5 @@ async fn on_new_accounting<A: Adapter + 'static>(
         })])
         .await;
 
-    Ok(NewStateResult::Sent(propagation_results))
+    Ok(propagation_results)
 }

validator_worker/src/producer.rs

@@ -13,7 +13,6 @@ use crate::sentry_interface::{PropagationResult, SentryApi};
 pub enum TickStatus<AE: AdapterErrorKind> {
     Sent {
         channel: ChannelId,
-        balances: BalancesMap,
         new_accounting: Accounting,
         accounting_propagation: Vec<PropagationResult<AE>>,
         event_counts: usize,
@@ -40,14 +39,13 @@ pub async fn tick<A: Adapter + 'static>(
         .await?;
 
     if !aggrs.events.is_empty() {
-        let (balances, new_accounting) = merge_aggrs(&accounting, &aggrs.events, &iface.channel)?;
+        let new_accounting = merge_aggrs(&accounting, &aggrs.events, &iface.channel)?;
 
         let message_types = MessageTypes::Accounting(new_accounting.clone());
 
         Ok(TickStatus::Sent {
             channel: iface.channel.id,
             accounting_propagation: iface.propagate(&[&message_types]).await,
-            balances,
             new_accounting,
             event_counts: aggrs.events.len(),
         })