Merge pull request #281 from AdExNetwork/issue-278-propagation-logging

Issue #278 Propagation logging

Author: elpiel

adapter/src/ethereum.rs

@@ -341,7 +341,7 @@ fn hash_message(message: &[u8]) -> [u8; 32] {
 }
 
 // Ethereum Web Tokens
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct Payload {
     pub id: String,
     pub era: i64,
@@ -350,7 +350,7 @@ pub struct Payload {
     pub identity: Option<ValidatorId>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct VerifyPayload {
     pub from: ValidatorId,
     pub payload: Payload,
@@ -704,15 +704,23 @@ mod test {
         let expected = "eyJ0eXBlIjoiSldUIiwiYWxnIjoiRVRIIn0.eyJpZCI6ImF3ZXNvbWVWYWxpZGF0b3IiLCJlcmEiOjEwMDAwMCwiYWRkcmVzcyI6IjB4MmJEZUFGQUU1Mzk0MDY2OURhQTZGNTE5MzczZjY4NmMxZjNkMzM5MyJ9.gGw_sfnxirENdcX5KJQWaEt4FVRvfEjSLD4f3OiPrJIltRadeYP2zWy9T2GYcK5xxD96vnqAw4GebAW7rMlz4xw";
         assert_eq!(response, expected, "generated wrong ewt signature");
 
-        let expected_verification_response = r#"VerifyPayload { from: ValidatorId([43, 222, 175, 174, 83, 148, 6, 105, 218, 166, 245, 25, 55, 63, 104, 108, 31, 61, 51, 147]), payload: Payload { id: "awesomeValidator", era: 100000, address: "0x2bDeAFAE53940669DaA6F519373f686c1f3d3393", identity: None } }"#;
+        let expected_verification_response = VerifyPayload {
+            from: ValidatorId::try_from("0x2bdeafae53940669daa6f519373f686c1f3d3393")
+                .expect("Valid ValidatorId"),
+            payload: Payload {
+                id: "awesomeValidator".to_string(),
+                era: 100000,
+                address: "0x2bDeAFAE53940669DaA6F519373f686c1f3d3393".to_string(),
+                identity: None,
+            },
+        };
 
         let parts: Vec<&str> = expected.split('.').collect();
         let verification =
             ewt_verify(parts[0], parts[1], parts[2]).expect("Failed to verify ewt token");
 
         assert_eq!(
-            expected_verification_response,
-            format!("{:?}", verification),
+            expected_verification_response, verification,
             "generated wrong verification payload"
         );
     }

primitives/src/channel.rs

@@ -11,7 +11,7 @@ use crate::{AdUnit, EventSubmission, TargetingTag, ValidatorDesc, ValidatorId};
 use hex::{FromHex, FromHexError};
 use std::ops::Deref;
 
-#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Copy, Clone, Hash)]
+#[derive(Serialize, Deserialize, PartialEq, Eq, Copy, Clone, Hash)]
 #[serde(transparent)]
 pub struct ChannelId(
     #[serde(
@@ -21,6 +21,12 @@ pub struct ChannelId(
     [u8; 32],
 );
 
+impl fmt::Debug for ChannelId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "ChannelId({})", self)
+    }
+}
+
 fn channel_id_from_str<'de, D>(deserializer: D) -> Result<[u8; 32], D::Error>
 where
     D: Deserializer<'de>,
@@ -151,23 +157,19 @@ pub struct ChannelSpec {
 /// A (leader, follower) tuple
 pub struct SpecValidators(ValidatorDesc, ValidatorDesc);
 
+#[derive(Debug)]
 pub enum SpecValidator<'a> {
     Leader(&'a ValidatorDesc),
     Follower(&'a ValidatorDesc),
-    None,
 }
 
 impl<'a> SpecValidator<'a> {
-    pub fn is_some(&self) -> bool {
-        match &self {
-            SpecValidator::None => false,
-            _ => true,
+    pub fn validator(&self) -> &'a ValidatorDesc {
+        match self {
+            SpecValidator::Leader(validator) => validator,
+            SpecValidator::Follower(validator) => validator,
         }
     }
-
-    pub fn is_none(&self) -> bool {
-        !self.is_some()
-    }
 }
 
 impl SpecValidators {
@@ -183,13 +185,13 @@ impl SpecValidators {
         &self.1
     }
 
-    pub fn find(&self, validator_id: &ValidatorId) -> SpecValidator<'_> {
+    pub fn find(&self, validator_id: &ValidatorId) -> Option<SpecValidator<'_>> {
         if &self.leader().id == validator_id {
-            SpecValidator::Leader(&self.leader())
+            Some(SpecValidator::Leader(&self.leader()))
         } else if &self.follower().id == validator_id {
-            SpecValidator::Follower(&self.follower())
+            Some(SpecValidator::Follower(&self.follower()))
         } else {
-            SpecValidator::None
+            None
         }
     }
 

primitives/src/channel_validator.rs

@@ -14,8 +14,10 @@ pub trait ChannelValidator {
     ) -> Result<(), ChannelError> {
         let adapter_channel_validator = match channel.spec.validators.find(validator_identity) {
             // check if the channel validators include our adapter identity
-            SpecValidator::None => return Err(ChannelError::AdapterNotIncluded),
-            SpecValidator::Leader(validator) | SpecValidator::Follower(validator) => validator,
+            None => return Err(ChannelError::AdapterNotIncluded),
+            Some(SpecValidator::Leader(validator)) | Some(SpecValidator::Follower(validator)) => {
+                validator
+            }
         };
 
         if channel.valid_until < Utc::now() {

primitives/src/validator.rs

@@ -14,7 +14,7 @@ pub enum ValidatorError {
     InvalidTransition,
 }
 
-#[derive(Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 #[serde(transparent)]
 pub struct ValidatorId(
     #[serde(
@@ -24,6 +24,12 @@ pub struct ValidatorId(
     [u8; 20],
 );
 
+impl fmt::Debug for ValidatorId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "ValidatorId({})", self.to_hex_prefix_string())
+    }
+}
+
 fn validator_id_from_str<'de, D>(deserializer: D) -> Result<[u8; 20], D::Error>
 where
     D: Deserializer<'de>,

sentry/src/lib.rs

@@ -252,8 +252,6 @@ async fn channels_router<A: Adapter + 'static>(
 ) -> Result<Response<Body>, ResponseError> {
     let (path, method) = (req.uri().path().to_owned(), req.method());
 
-    // example with
-    // @TODO remove later
     // regex matching for routes with params
     if let (Some(caps), &Method::GET) = (LAST_APPROVED_BY_CHANNEL_ID.captures(&path), method) {
         let param = RouteParams(vec![caps

sentry/src/routes/channel.rs

@@ -12,7 +12,6 @@ use futures::future::try_join_all;
 use hex::FromHex;
 use hyper::{Body, Request, Response};
 use primitives::adapter::Adapter;
-use primitives::channel::SpecValidator;
 use primitives::sentry::{Event, LastApproved, LastApprovedResponse, SuccessResponse};
 use primitives::validator::MessageTypes;
 use primitives::{Channel, ChannelId};
@@ -235,7 +234,7 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
         .ok_or_else(|| ResponseError::BadRequest("missing messages body".to_string()))?;
 
     match channel.spec.validators.find(&session.uid) {
-        SpecValidator::None => Err(ResponseError::Unauthorized),
+        None => Err(ResponseError::Unauthorized),
         _ => {
             try_join_all(messages.iter().map(|message| {
                 insert_validator_messages(&app.pool, &channel, &session.uid, &message)

validator_worker/src/error.rs

@@ -11,8 +11,8 @@ 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::Tick(err) => write!(f, "Tick error - {}", err),
+            TickError::TimedOut(err) => write!(f, "Tick timed out ({})", err),
+            TickError::Tick(err) => write!(f, "Tick: {}", err),
         }
     }
 }
@@ -31,17 +31,13 @@ impl<AE: AdapterErrorKind> fmt::Display for Error<AE> {
         use Error::*;
 
         match self {
-            SentryApi(err) => write!(f, "Sentry Api error - {}", err),
-            LeaderTick(channel_id, err) => write!(
-                f,
-                "Error for Leader tick of Channel ({}) - {}",
-                channel_id, err
-            ),
-            FollowerTick(channel_id, err) => write!(
-                f,
-                "Error for Follower tick of Channel ({}) - {}",
-                channel_id, err
-            ),
+            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)
+            }
         }
     }
 }

validator_worker/src/follower.rs

@@ -1,28 +1,44 @@
 use std::error::Error;
 
-use primitives::adapter::Adapter;
+use primitives::adapter::{Adapter, AdapterErrorKind};
 use primitives::validator::{ApproveState, MessageTypes, NewState, RejectState};
 use primitives::BalancesMap;
 
 use crate::core::follower_rules::{get_health, is_valid_transition};
-use crate::heartbeat::heartbeat;
-use crate::sentry_interface::SentryApi;
+use crate::heartbeat::{heartbeat, HeartbeatStatus};
+use crate::sentry_interface::{PropagationResult, SentryApi};
 use crate::{get_state_root_hash, producer};
 use chrono::Utc;
 
-enum InvalidNewState {
+#[derive(Debug)]
+pub enum InvalidNewState {
     RootHash,
     Signature,
     Transition,
     Health,
 }
 
-enum NewStateResult {
-    Ok,
-    Err(InvalidNewState),
+#[derive(Debug)]
+pub enum ApproveStateResult<AE: AdapterErrorKind> {
+    Sent(Vec<PropagationResult<AE>>),
+    /// Conditions for handling the new state haven't been met
+    NotSent,
+    Err {
+        new_state: InvalidNewState,
+        reject_state_propagation: Vec<PropagationResult<AE>>,
+    },
 }
 
-pub async fn tick<A: Adapter + 'static>(iface: &SentryApi<A>) -> Result<(), Box<dyn Error>> {
+#[derive(Debug)]
+pub struct TickStatus<AE: AdapterErrorKind> {
+    pub heartbeat: HeartbeatStatus<AE>,
+    pub approve_state: ApproveStateResult<AE>,
+    pub producer_tick: producer::TickStatus<AE>,
+}
+
+pub async fn tick<A: Adapter + 'static>(
+    iface: &SentryApi<A>,
+) -> Result<TickStatus<A::AdapterError>, Box<dyn Error>> {
     let from = &iface.channel.spec.validators.leader().id;
     let new_msg_response = iface.get_latest_msg(from, &["NewState"]).await?;
     let new_msg = match new_msg_response {
@@ -42,21 +58,32 @@ pub async fn tick<A: Adapter + 'static>(iface: &SentryApi<A>) -> Result<(), Box<
 
     let latest_is_responded_to = match (&new_msg, &our_latest_msg_state_root) {
         (Some(new_msg), Some(state_root)) => &new_msg.state_root == state_root,
-        (_, _) => false,
+        _ => false,
     };
 
-    let (balances, _) = producer::tick(&iface).await?;
-    if let (Some(new_state), false) = (new_msg, latest_is_responded_to) {
-        on_new_state(&iface, &balances, &new_state).await?;
-    }
-    heartbeat(&iface, balances).await.map(|_| ())
+    let producer_tick = producer::tick(&iface).await?;
+    let balances = match &producer_tick {
+        producer::TickStatus::AccountingSent { balances, .. } => balances,
+        producer::TickStatus::AccountingNotSent(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
+    };
+
+    Ok(TickStatus {
+        heartbeat: heartbeat(&iface, balances).await?,
+        approve_state: approve_state_result,
+        producer_tick,
+    })
 }
 
 async fn on_new_state<'a, A: Adapter + 'static>(
     iface: &'a SentryApi<A>,
     balances: &'a BalancesMap,
     new_state: &'a NewState,
-) -> Result<NewStateResult, Box<dyn Error>> {
+) -> Result<ApproveStateResult<A::AdapterError>, Box<dyn Error>> {
     let proposed_balances = new_state.balances.clone();
     let proposed_state_root = new_state.state_root.clone();
     if proposed_state_root != hex::encode(get_state_root_hash(&iface, &proposed_balances)?) {
@@ -96,22 +123,22 @@ async fn on_new_state<'a, A: Adapter + 'static>(
     let health_threshold = u64::from(iface.config.health_threshold_promilles);
     let is_healthy = health >= health_threshold;
 
-    iface
+    let propagation_result = iface
         .propagate(&[&MessageTypes::ApproveState(ApproveState {
             state_root: proposed_state_root,
             signature,
             is_healthy,
         })])
         .await;
 
-    Ok(NewStateResult::Ok)
+    Ok(ApproveStateResult::Sent(propagation_result))
 }
 
 async fn on_error<'a, A: Adapter + 'static>(
     iface: &'a SentryApi<A>,
     new_state: &'a NewState,
     status: InvalidNewState,
-) -> NewStateResult {
+) -> ApproveStateResult<A::AdapterError> {
     use InvalidNewState::*;
     let reason = match &status {
         RootHash => "InvalidRootHash",
@@ -121,7 +148,7 @@ async fn on_error<'a, A: Adapter + 'static>(
     }
     .to_string();
 
-    iface
+    let reject_state_propagation = iface
         .propagate(&[&MessageTypes::RejectState(RejectState {
             reason,
             state_root: new_state.state_root.clone(),
@@ -132,5 +159,8 @@ async fn on_error<'a, A: Adapter + 'static>(
         })])
         .await;
 
-    NewStateResult::Err(status)
+    ApproveStateResult::Err {
+        reject_state_propagation,
+        new_state: status,
+    }
 }