Issue #70 validation tick timeout (#71)

* domain - re-export `SpecValidator`

* domain - channel - ChannelId - impl `Display` instead of `ToString`

* validator - domain - re-export some structs & traits

* validator - infra - TickWorker - Validation tick timeout

Author: elpiel

domain/src/channel.rs

@@ -1,4 +1,5 @@
 use std::convert::TryFrom;
+use std::fmt;
 
 use chrono::serde::{ts_milliseconds, ts_seconds};
 use chrono::{DateTime, Utc};
@@ -17,7 +18,7 @@ pub struct ChannelId {
     pub id: [u8; 32],
 }
 
-impl ToString for ChannelId {
+impl fmt::Display for ChannelId {
     /// Converts a ChannelId to hex string with prefix
     ///
     /// Example:
@@ -26,11 +27,12 @@ impl ToString for ChannelId {
     ///
     /// let hex_string = "0x061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088";
     /// let channel_id = ChannelId::try_from_hex(&hex_string).expect("Should be a valid hex string already");
-    ///
-    /// assert_eq!("0x061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088", channel_id.to_string());
+    /// let channel_hex_string = format!("{}", channel_id);
+    /// assert_eq!("0x061d5e2a67d0a9a10f1c732bca12a676d83f79663a396f7d87b3e30b9b411088", &channel_hex_string);
     /// ```
-    fn to_string(&self) -> String {
-        SerHex::<StrictPfx>::into_hex(&self.id).unwrap()
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
+        let hex_string = SerHex::<StrictPfx>::into_hex(&self.id).unwrap();
+        write!(f, "{}", hex_string)
     }
 }
 

domain/src/lib.rs

@@ -9,7 +9,7 @@ pub use util::tests as test_util;
 pub use self::ad_unit::AdUnit;
 pub use self::asset::Asset;
 pub use self::bignum::BigNum;
-pub use self::channel::{Channel, ChannelId, ChannelSpec, SpecValidators};
+pub use self::channel::{Channel, ChannelId, ChannelSpec, SpecValidator, SpecValidators};
 pub use self::event_submission::EventSubmission;
 #[cfg(feature = "repositories")]
 pub use self::repository::*;

validator/src/domain.rs

@@ -1,3 +1,5 @@
+pub use self::channel::ChannelRepository;
+pub use self::validator::{Validator, ValidatorError, ValidatorFuture};
 pub use self::worker::{Worker, WorkerFuture};
 
 pub mod channel;

validator/src/infrastructure/worker.rs

@@ -3,24 +3,26 @@ pub use self::single::TickWorker;
 
 pub mod single {
     use std::sync::Arc;
+    use std::time::Duration;
 
-    use futures::future::FutureExt;
+    use futures::compat::Future01CompatExt;
+    use futures::future::{FutureExt, TryFutureExt};
+    use tokio::util::FutureExt as TokioFutureExt;
 
-    use domain::channel::SpecValidator;
-    use domain::Channel;
+    use domain::{Channel, SpecValidator};
 
-    use crate::domain::channel::ChannelRepository;
-    use crate::domain::validator::Validator;
-    use crate::domain::{Worker, WorkerFuture};
-    use crate::infrastructure::validator::follower::Follower;
-    use crate::infrastructure::validator::leader::Leader;
+    use crate::domain::{ChannelRepository, Validator, Worker, WorkerFuture};
+    use crate::infrastructure::validator::{Follower, Leader};
 
     #[derive(Clone)]
     pub struct TickWorker {
         pub leader: Leader,
         pub follower: Follower,
         pub channel_repository: Arc<dyn ChannelRepository>,
+        // @TODO: use the adapter(maybe?) instead of repeating the identity
         pub identity: String,
+        // @TODO: Pass configuration by which this can be set
+        pub validation_tick_timeout: Duration,
     }
 
     /// Single tick worker
@@ -45,15 +47,33 @@ pub mod single {
 
             match &channel.spec.validators.find(&self.identity) {
                 SpecValidator::Leader(_) => {
-                    self.leader.tick(channel);
-                    eprintln!("Channel {} handled as __Leader__", channel_id.to_string());
+                    let tick_future = self.leader.tick(channel);
+
+                    let tick_result = await!(tick_future
+                        .compat()
+                        .timeout(self.validation_tick_timeout)
+                        .compat());
+
+                    match tick_result {
+                        Ok(_) => println!("Channel {} handled as __Leader__", channel_id),
+                        Err(_) => eprintln!("Channel {} Timed out", channel_id),
+                    }
                 }
                 SpecValidator::Follower(_) => {
-                    self.follower.tick(channel);
-                    eprintln!("Channel {} handled as __Follower__", channel_id.to_string());
+                    let tick_future = self.follower.tick(channel);
+
+                    let tick_result = await!(tick_future
+                        .compat()
+                        .timeout(self.validation_tick_timeout)
+                        .compat());
+
+                    match tick_result {
+                        Ok(_) => println!("Channel {} handled as __Follower__", channel_id),
+                        Err(_) => eprintln!("Channel {} Timed out", channel_id),
+                    }
                 }
                 SpecValidator::None => {
-                    eprintln!("Channel {} is not validated by us", channel_id.to_string());
+                    eprintln!("Channel {} is not validated by us", channel_id);
                 }
             };
 

validator/src/main.rs

@@ -14,7 +14,14 @@ lazy_static! {
             .unwrap()
             .parse()
             .unwrap();
+
+        let validation_tick_timeout = std::env::var("VALIDATOR_VALIDATION_TICK_TIMEOUT")
+            .unwrap()
+            .parse()
+            .unwrap();
+
         Config {
+            validation_tick_timeout: Duration::from_millis(validation_tick_timeout),
             ticks_wait_time: Duration::from_millis(ticks_wait_time),
             sentry_url: std::env::var("VALIDATOR_SENTRY_URL")
                 .unwrap()
@@ -89,6 +96,7 @@ fn run(is_single_tick: bool, adapter: impl Adapter) {
         follower: Follower {},
         channel_repository,
         identity: adapter.config().identity.to_string(),
+        validation_tick_timeout: CONFIG.validation_tick_timeout,
     };
 
     if !is_single_tick {
@@ -118,6 +126,7 @@ fn run(is_single_tick: bool, adapter: impl Adapter) {
 }
 
 struct Config {
+    pub validation_tick_timeout: Duration,
     pub ticks_wait_time: Duration,
     pub sentry_url: String,
 }