Merge pull request #426 from AdExNetwork/v5-validator-workflow

V5 validator workflow

Author: elpiel

Cargo.lock

@@ -42,14 +42,22 @@ pub fn get_signable_state_root(
     Ok(res)
 }
 
-pub fn get_balance_leaf(acc: &Address, amnt: &BigNum) -> Result<[u8; 32], Box<dyn Error>> {
-    let tokens = [
-        Token::Address(EthAddress::from_slice(acc.as_bytes())),
-        Token::Uint(
-            U256::from_dec_str(&amnt.to_str_radix(10))
-                .map_err(|_| ChannelError::InvalidArgument("failed to parse amt".into()))?,
-        ),
-    ];
+pub fn get_balance_leaf(
+    is_spender: bool,
+    acc: &Address,
+    amnt: &BigNum,
+) -> Result<[u8; 32], Box<dyn Error>> {
+    let address = Token::Address(EthAddress::from_slice(acc.as_bytes()));
+    let amount = Token::Uint(
+        U256::from_dec_str(&amnt.to_str_radix(10))
+            .map_err(|_| ChannelError::InvalidArgument("failed to parse amt".into()))?,
+    );
+
+    let tokens = if is_spender {
+        vec![Token::String("spender".into()), address, amount]
+    } else {
+        vec![address, amount]
+    };
     let encoded = encode(&tokens).to_vec();
 
     let mut result = Keccak::new_keccak256();

adapter/src/lib.rs

@@ -38,6 +38,14 @@ impl Balances<UncheckedState> {
 }
 
 impl<S: BalancesState> Balances<S> {
+    pub fn new() -> Balances<CheckedState> {
+        Balances {
+            earners: Default::default(),
+            spenders: Default::default(),
+            state: Default::default(),
+        }
+    }
+
     pub fn spend(
         &mut self,
         spender: Address,
@@ -78,6 +86,19 @@ impl<S: BalancesState> Balances<S> {
             state: PhantomData::default(),
         }
     }
+
+    /// Returns a tuple of the sum of `(earners, spenders)`
+    pub fn sum(&self) -> Option<(UnifiedNum, UnifiedNum)> {
+        self.spenders
+            .values()
+            .sum::<Option<UnifiedNum>>()
+            .map(|spenders| {
+                let earners = self.earners.values().sum::<Option<UnifiedNum>>()?;
+
+                Some((earners, spenders))
+            })
+            .flatten()
+    }
 }
 
 #[derive(Debug, Error)]

primitives/src/balances.rs

@@ -4,7 +4,7 @@ use std::fmt;
 
 use crate::{Address, ChannelId, Validator, ValidatorId};
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]
 #[serde(rename_all = "camelCase")]
 pub struct Channel {
     pub leader: ValidatorId,
@@ -45,7 +45,7 @@ impl Channel {
 }
 
 /// The nonce is an Unsigned 256 number
-#[derive(Clone, Copy, PartialEq, Eq)]
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
 pub struct Nonce(pub U256);
 
 impl Nonce {

primitives/src/channel_v5.rs

@@ -1,6 +1,5 @@
 use crate::{
     balances::BalancesState,
-    channel_v5::Channel as ChannelV5,
     spender::Spender,
     validator::{ApproveState, Heartbeat, MessageTypes, NewState, Type as MessageType},
     Address, Balances, BigNum, Channel, ChannelId, ValidatorId, IPFS,
@@ -11,12 +10,11 @@ use std::{collections::HashMap, fmt, hash::Hash};
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-pub struct Accounting<S: BalancesState> {
-    pub channel: ChannelV5,
+/// Channel Accounting response
+/// A collection of all `Accounting`s for a specific `Channel`
+pub struct AccountingResponse<S: BalancesState> {
     #[serde(flatten, bound = "S: BalancesState")]
     pub balances: Balances<S>,
-    pub updated: Option<DateTime<Utc>>,
-    pub created: DateTime<Utc>,
 }
 
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
@@ -187,6 +185,15 @@ pub struct AggregateEvents {
 #[serde(rename_all = "camelCase")]
 pub struct ChannelListResponse {
     pub channels: Vec<Channel>,
+    // TODO: Replace with `Pagination`
+    pub total_pages: u64,
+    pub total: u64,
+    pub page: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Pagination {
     pub total_pages: u64,
     pub total: u64,
     pub page: u64,
@@ -317,6 +324,36 @@ pub mod channel_list {
     }
 }
 
+pub mod campaign {
+    use crate::{Address, Campaign, ValidatorId};
+    use chrono::{serde::ts_seconds, DateTime, Utc};
+    use serde::{Deserialize, Serialize};
+
+    use super::Pagination;
+
+    #[derive(Debug, Serialize, Deserialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct CampaignListResponse {
+        pub campaigns: Vec<Campaign>,
+        #[serde(flatten)]
+        pub pagination: Pagination,
+    }
+
+    #[derive(Debug, Serialize, Deserialize)]
+    pub struct CampaignListQuery {
+        #[serde(default)]
+        // default is `u64::default()` = `0`
+        pub page: u64,
+        /// filters the list on `active.to >= active_to_ge`
+        /// It should be the same timestamp format as the `Campaign.active.to`: **seconds**
+        #[serde(with = "ts_seconds", default = "Utc::now", rename = "activeTo")]
+        pub active_to_ge: DateTime<Utc>,
+        pub creator: Option<Address>,
+        /// filters the campaigns containing a specific validator if provided
+        pub validator: Option<ValidatorId>,
+    }
+}
+
 pub mod campaign_create {
     use chrono::{serde::ts_milliseconds, DateTime, Utc};
     use serde::{Deserialize, Serialize};

primitives/src/sentry.rs

@@ -3,6 +3,7 @@ use std::{convert::TryFrom, fmt, str::FromStr};
 use parse_display::Display;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
+pub use url::ParseError;
 use url::Url;
 
 // `url::Url::scheme()` returns lower-cased ASCII string without `:`
@@ -46,12 +47,15 @@ impl ApiUrl {
 
     /// The Endpoint of which we want to get an url to (strips prefixed `/` from the endpoint),
     /// which can can include:
+    ///
     /// - path
     /// - query
     /// - fragments - usually should not be used for requesting API resources from server
+    ///
     /// This method does **not** check if a file is present
+    ///
     /// This method strips the starting `/` of the endpoint, if there is one
-    pub fn join(&self, endpoint: &str) -> Result<Url, url::ParseError> {
+    pub fn join(&self, endpoint: &str) -> Result<Url, ParseError> {
         let stripped = endpoint.strip_prefix('/').unwrap_or(endpoint);
         // this join is safe, since we always prefix the Url with `/`
         self.0.join(stripped)

primitives/src/util/api.rs

@@ -246,7 +246,7 @@ async fn create_or_update_spendable_document(
     };
 
     let spendable = Spendable {
-        channel: channel.clone(),
+        channel: *channel,
         deposit: Deposit {
             total,
             still_on_create2,

sentry/src/routes/channel.rs

@@ -29,11 +29,13 @@ futures = "0.3"
 tokio = { version = "1", features = ["time", "rt-multi-thread"] }
 # API client
 reqwest = { version = "0.11", features = ["json"] }
-# Configuration
+# Other
 lazy_static = "^1.4"
+thiserror = "^1.0"
 # (De)Serialization
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = "1.0"
+serde_urlencoded = "0.7"
 toml = "0.5"
 # CLI
 clap = "^2.33"

validator_worker/Cargo.toml

@@ -0,0 +1,120 @@
+use crate::{
+    error::{Error, TickError},
+    follower,
+    sentry_interface::{campaigns::all_campaigns, Validator, Validators},
+    SentryApi,
+};
+use primitives::{adapter::Adapter, channel_v5::Channel, config::Config, util::ApiUrl, ChannelId};
+// use slog::{error, info, Logger};
+use slog::Logger;
+use std::collections::{hash_map::Entry, HashSet};
+
+pub async fn channel_tick<A: Adapter + 'static>(
+    adapter: A,
+    config: &Config,
+    logger: &Logger,
+    channel: Channel,
+    validators: Validators,
+) -> Result<ChannelId, Error<A::AdapterError>> {
+    let tick = channel
+        .find_validator(adapter.whoami())
+        .ok_or(Error::ChannelNotIntendedForUs)?;
+
+    let sentry = SentryApi::init(
+        adapter,
+        logger.clone(),
+        config.clone(),
+        (channel, validators),
+    )?;
+    // `GET /channel/:id/spender/all`
+    let all_spenders = sentry.get_all_spenders().await?;
+
+    // `GET /channel/:id/accounting`
+    // Validation #1:
+    // sum(Accounting.spenders) == sum(Accounting.earners)
+    let accounting = sentry.get_accounting(channel.id()).await?;
+
+    // Validation #2:
+    // spender.spender_leaf.total_deposit >= accounting.balances.spenders[spender.address]
+    if !all_spenders.iter().all(|(address, spender)| {
+        spender.total_deposited
+            >= accounting
+                .balances
+                .spenders
+                .get(address)
+                .cloned()
+                .unwrap_or_default()
+    }) {
+        return Err(Error::Validation);
+    }
+
+    // TODO: Add timeout
+    let _tick_result = match tick {
+        primitives::Validator::Leader(_v) => todo!(),
+        primitives::Validator::Follower(_v) => {
+            follower::tick(&sentry, channel, accounting.balances)
+                .await
+                .map_err(|err| Error::FollowerTick(channel.id(), TickError::Tick(err)))?
+        }
+    };
+
+    // Validation #3
+    // Accounting.balances != NewState.balances
+
+    // Validation #4
+    // OUTPACE Rules:
+
+    Ok(channel.id())
+}
+
+/// Fetches all `Campaign`s from Sentry and builds the `Channel`s to be processed
+/// along side all the `Validator`s' url & auth token
+pub async fn collect_channels<A: Adapter + 'static>(
+    adapter: A,
+    sentry_url: &ApiUrl,
+    _config: &Config,
+    _logger: &Logger,
+) -> Result<(HashSet<Channel>, Validators), reqwest::Error> {
+    let whoami = adapter.whoami();
+
+    let campaigns = all_campaigns(sentry_url, whoami).await?;
+    let channels = campaigns
+        .iter()
+        .map(|campaign| campaign.channel)
+        .collect::<HashSet<_>>();
+
+    let validators = campaigns
+        .into_iter()
+        .fold(Validators::new(), |mut acc, campaign| {
+            for validator_desc in campaign.validators.iter() {
+                // if Validator is already there, we can just skip it
+                // remember, the campaigns are ordered by `created DESC`
+                // so we will always get the latest Validator url first
+                match acc.entry(validator_desc.id) {
+                    Entry::Occupied(_) => continue,
+                    Entry::Vacant(entry) => {
+                        // try to parse the url of the Validator Desc
+                        let validator_url = validator_desc.url.parse::<ApiUrl>();
+                        // and also try to find the Auth token in the config
+
+                        // if there was an error with any of the operations, skip this `ValidatorDesc`
+                        let auth_token = adapter.get_auth(&validator_desc.id);
+
+                        // only if `ApiUrl` parsing is `Ok` & Auth Token is found in the `Adapter`
+                        if let (Ok(url), Ok(auth_token)) = (validator_url, auth_token) {
+                            // add an entry for propagation
+                            entry.insert(Validator {
+                                url,
+                                token: auth_token,
+                            });
+                        }
+                        // otherwise it will try to do the same things on the next encounter of this `ValidatorId`
+                    }
+                }
+            }
+
+            acc
+        });
+
+    Ok((channels, validators))
+}