validator_worker - SentryApi & validator worker errors

Author: elpiel

validator_worker/src/error.rs

@@ -1,25 +1,47 @@
+use primitives::adapter::AdapterErrorKind;
 use primitives::ChannelId;
-use serde::{Deserialize, Serialize};
-use std::error::Error;
-use std::fmt::{Display, Formatter, Result};
+use std::fmt;
 
-#[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
-pub enum ValidatorWorker {
-    Configuration(String),
-    Failed(String),
-    Channel(ChannelId, String),
+#[derive(Debug)]
+pub enum TickError {
+    TimedOut(tokio::time::Elapsed),
+    Tick(Box<dyn std::error::Error>),
 }
 
-impl Error for ValidatorWorker {}
+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),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum Error<AE: AdapterErrorKind> {
+    SentryApi(crate::sentry_interface::Error<AE>),
+    LeaderTick(ChannelId, TickError),
+    FollowerTick(ChannelId, TickError),
+}
+
+impl<AE: AdapterErrorKind> std::error::Error for Error<AE> {}
+
+impl<AE: AdapterErrorKind> fmt::Display for Error<AE> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use Error::*;
 
-impl Display for ValidatorWorker {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
         match self {
-            ValidatorWorker::Configuration(err) => write!(f, "Configuration error: {}", err),
-            ValidatorWorker::Failed(err) => write!(f, "error: {}", err),
-            ValidatorWorker::Channel(channel_id, err) => {
-                write!(f, "Channel {}: {}", channel_id, err)
-            }
+            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
+            ),
         }
     }
 }

validator_worker/src/main.rs

@@ -16,7 +16,7 @@ use primitives::config::{configuration, Config};
 use primitives::util::tests::prep_db::{AUTH, IDS};
 use primitives::{Channel, SpecValidator, ValidatorId};
 use slog::{error, info, Logger};
-use validator_worker::error::ValidatorWorker as ValidatorWorkerError;
+use validator_worker::error::{Error as ValidatorWorkerError, TickError};
 use validator_worker::{all_channels, follower, leader, SentryApi};
 
 #[derive(Debug, Clone)]
@@ -195,39 +195,40 @@ async fn validator_tick<A: Adapter + 'static>(
     channel: Channel,
     config: &Config,
     logger: &Logger,
-) -> Result<(), ValidatorWorkerError> {
+) -> Result<(), ValidatorWorkerError<A::AdapterError>> {
     let whoami = adapter.whoami().clone();
     // Cloning the `Logger` is cheap, see documentation for more info
-    let sentry = SentryApi::init(adapter, &channel, &config, logger.clone())?;
+    let sentry = SentryApi::init(adapter, &channel, &config, logger.clone())
+        .map_err(ValidatorWorkerError::SentryApi)?;
     let duration = Duration::from_millis(config.validator_tick_timeout as u64);
 
     match channel.spec.validators.find(&whoami) {
         SpecValidator::Leader(_) => match timeout(duration, leader::tick(&sentry)).await {
-            Err(timeout_e) => Err(ValidatorWorkerError::Channel(
+            Err(timeout_e) => Err(ValidatorWorkerError::LeaderTick(
                 channel.id,
-                timeout_e.to_string(),
+                TickError::TimedOut(timeout_e),
             )),
-            Ok(Err(tick_e)) => Err(ValidatorWorkerError::Channel(
+            Ok(Err(tick_e)) => Err(ValidatorWorkerError::LeaderTick(
                 channel.id,
-                tick_e.to_string(),
+                TickError::Tick(tick_e),
             )),
             _ => Ok(()),
         },
         SpecValidator::Follower(_) => match timeout(duration, follower::tick(&sentry)).await {
-            Err(timeout_e) => Err(ValidatorWorkerError::Channel(
+            Err(timeout_e) => Err(ValidatorWorkerError::FollowerTick(
                 channel.id,
-                timeout_e.to_string(),
+                TickError::TimedOut(timeout_e),
             )),
-            Ok(Err(tick_e)) => Err(ValidatorWorkerError::Channel(
+            Ok(Err(tick_e)) => Err(ValidatorWorkerError::FollowerTick(
                 channel.id,
-                tick_e.to_string(),
+                TickError::Tick(tick_e),
             )),
             _ => Ok(()),
         },
-        SpecValidator::None => Err(ValidatorWorkerError::Channel(
-            channel.id,
-            "validatorTick: processing a channel which we are not validating".to_string(),
-        )),
+        // @TODO: Can we make this so that we don't have this check at all? maybe something with the SentryApi struct?
+        SpecValidator::None => {
+            unreachable!("SentryApi makes a check if validator is in Channel spec on `init()`")
+        }
     }
 }
 

validator_worker/src/sentry_interface.rs

@@ -1,19 +1,20 @@
-use crate::error::ValidatorWorker;
+use std::collections::HashMap;
+use std::fmt;
+use std::time::Duration;
+
 use chrono::{DateTime, Utc};
-use futures::future::try_join_all;
-use futures::future::TryFutureExt;
-use primitives::adapter::Adapter;
+use futures::future::{try_join_all, TryFutureExt};
+use reqwest::{Client, Response};
+use slog::{error, Logger};
+
+use primitives::adapter::{Adapter, AdapterErrorKind, Error as AdapterError};
 use primitives::channel::SpecValidator;
 use primitives::sentry::{
     ChannelListResponse, EventAggregateResponse, LastApprovedResponse, SuccessResponse,
     ValidatorMessageResponse,
 };
 use primitives::validator::MessageTypes;
-use primitives::{Channel, Config, ToETHChecksum, ValidatorDesc, ValidatorId};
-use reqwest::{Client, Response};
-use slog::{error, Logger};
-use std::collections::HashMap;
-use std::time::Duration;
+use primitives::{Channel, ChannelId, Config, ToETHChecksum, ValidatorDesc, ValidatorId};
 
 #[derive(Debug, Clone)]
 pub struct SentryApi<T: Adapter> {
@@ -26,17 +27,63 @@ pub struct SentryApi<T: Adapter> {
     pub propagate_to: Vec<(ValidatorDesc, String)>,
 }
 
-impl<T: Adapter + 'static> SentryApi<T> {
+#[derive(Debug)]
+pub enum Error<AE: AdapterErrorKind> {
+    BuildingClient(reqwest::Error),
+    Request(reqwest::Error),
+    ValidatorAuthentication(AdapterError<AE>),
+    MissingWhoamiInChannelValidators {
+        channel: ChannelId,
+        validators: Vec<ValidatorId>,
+        whoami: ValidatorId,
+    },
+}
+
+impl<AE: AdapterErrorKind> std::error::Error for Error<AE> {}
+
+impl<AE: AdapterErrorKind> fmt::Display for Error<AE> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use Error::*;
+
+        match self {
+            BuildingClient(err) => write!(f, "Building client - {}", err),
+            Request(err) => write!(f, "Making a request - {}", err),
+            ValidatorAuthentication(err) => {
+                write!(f, "Getting authentication for validator - {}", err)
+            }
+            MissingWhoamiInChannelValidators {
+                channel,
+                validators,
+                whoami,
+            } => {
+                let validator_ids = validators
+                    .iter()
+                    .map(ToString::to_string)
+                    .collect::<Vec<_>>()
+                    .join(", ");
+                write!(
+                    f,
+                    "We cannot find validator entry for whoami ({}) in channel {} with validators: {}",
+                    whoami,
+                    channel,
+                    validator_ids
+                )
+            }
+        }
+    }
+}
+
+impl<A: Adapter + 'static> SentryApi<A> {
     pub fn init(
-        adapter: T,
+        adapter: A,
         channel: &Channel,
         config: &Config,
         logger: Logger,
-    ) -> Result<Self, ValidatorWorker> {
+    ) -> Result<Self, Error<A::AdapterError>> {
         let client = Client::builder()
             .timeout(Duration::from_millis(config.fetch_timeout.into()))
             .build()
-            .map_err(|e| ValidatorWorker::Failed(format!("building Client error: {}", e)))?;
+            .map_err(Error::BuildingClient)?;
 
         // validate that we are to validate the channel
         match channel.spec.validators.find(adapter.whoami()) {
@@ -51,12 +98,7 @@ impl<T: Adapter + 'static> SentryApi<T> {
                         adapter
                             .get_auth(&validator.id)
                             .map(|auth| (validator.to_owned(), auth))
-                            .map_err(|e| {
-                                ValidatorWorker::Failed(format!(
-                                    "Propagation error: get auth failed {}",
-                                    e
-                                ))
-                            })
+                            .map_err(Error::ValidatorAuthentication)
                     })
                     .collect::<Result<Vec<_>, _>>()?;
 
@@ -70,28 +112,36 @@ impl<T: Adapter + 'static> SentryApi<T> {
                     config: config.to_owned(),
                 })
             }
-            SpecValidator::None => Err(ValidatorWorker::Failed(
-                "We can not find validator entry for whoami".to_string(),
-            )),
+            SpecValidator::None => Err(Error::MissingWhoamiInChannelValidators {
+                channel: channel.id,
+                validators: channel
+                    .spec
+                    .validators
+                    .iter()
+                    .map(|v| v.id.clone())
+                    .collect(),
+                whoami: adapter.whoami().clone(),
+            }),
         }
     }
 
+    // @TODO: Remove logging & fix `try_join_all` @see: https://github.com/AdExNetwork/adex-validator-stack-rust/issues/278
     pub async fn propagate(&self, messages: &[&MessageTypes]) {
         let channel_id = format!("0x{}", hex::encode(&self.channel.id));
         if let Err(e) = try_join_all(self.propagate_to.iter().map(|(validator, auth_token)| {
             propagate_to(&channel_id, &auth_token, &self.client, &validator, messages)
         }))
         .await
         {
-            error!(&self.logger, "Propagation error: {}", e; "module" => "sentry_interface", "in" => "SentryApi");
+            error!(&self.logger, "Propagation error - {}", e; "module" => "sentry_interface", "in" => "SentryApi");
         }
     }
 
     pub async fn get_latest_msg(
         &self,
         from: &ValidatorId,
         message_types: &[&str],
-    ) -> Result<Option<MessageTypes>, reqwest::Error> {
+    ) -> Result<Option<MessageTypes>, Error<A::AdapterError>> {
         let message_type = message_types.join("+");
         let url = format!(
             "{}/validator-messages/{}/{}?limit=1",
@@ -104,6 +154,7 @@ impl<T: Adapter + 'static> SentryApi<T> {
             .get(&url)
             .send()
             .and_then(|res: Response| res.json::<ValidatorMessageResponse>())
+            .map_err(Error::Request)
             .await?;
 
         Ok(result.validator_messages.first().map(|m| m.msg.clone()))
@@ -112,38 +163,40 @@ impl<T: Adapter + 'static> SentryApi<T> {
     pub async fn get_our_latest_msg(
         &self,
         message_types: &[&str],
-    ) -> Result<Option<MessageTypes>, reqwest::Error> {
+    ) -> Result<Option<MessageTypes>, Error<A::AdapterError>> {
         self.get_latest_msg(self.adapter.whoami(), message_types)
             .await
     }
 
-    pub async fn get_last_approved(&self) -> Result<LastApprovedResponse, reqwest::Error> {
+    pub async fn get_last_approved(&self) -> Result<LastApprovedResponse, Error<A::AdapterError>> {
         self.client
             .get(&format!("{}/last-approved", self.validator_url))
             .send()
             .and_then(|res: Response| res.json::<LastApprovedResponse>())
+            .map_err(Error::Request)
             .await
     }
 
-    pub async fn get_last_msgs(&self) -> Result<LastApprovedResponse, reqwest::Error> {
+    pub async fn get_last_msgs(&self) -> Result<LastApprovedResponse, Error<A::AdapterError>> {
         self.client
             .get(&format!(
                 "{}/last-approved?withHeartbeat=true",
                 self.validator_url
             ))
             .send()
             .and_then(|res: Response| res.json::<LastApprovedResponse>())
+            .map_err(Error::Request)
             .await
     }
 
     pub async fn get_event_aggregates(
         &self,
         after: DateTime<Utc>,
-    ) -> Result<EventAggregateResponse, Box<ValidatorWorker>> {
+    ) -> Result<EventAggregateResponse, Error<A::AdapterError>> {
         let auth_token = self
             .adapter
             .get_auth(self.adapter.whoami())
-            .map_err(|e| Box::new(ValidatorWorker::Failed(e.to_string())))?;
+            .map_err(Error::ValidatorAuthentication)?;
 
         let url = format!(
             "{}/events-aggregates?after={}",
@@ -155,10 +208,10 @@ impl<T: Adapter + 'static> SentryApi<T> {
             .get(&url)
             .bearer_auth(&auth_token)
             .send()
-            .map_err(|e| Box::new(ValidatorWorker::Failed(e.to_string())))
+            .map_err(Error::Request)
             .await?
             .json()
-            .map_err(|e| Box::new(ValidatorWorker::Failed(e.to_string())))
+            .map_err(Error::Request)
             .await
     }
 }