Merge pull request #354 from samparsky/add-exhausted-property

Add exhausted property

Author: elpiel

README.md

@@ -22,6 +22,7 @@ Requirements:
 - Docker
 
 #### Linux
+
 - `build-essentials` is required to build the project (error: `linker ``cc`` not found`)
 - The crate `openssl-sys` requires `libssl-dev` and `pkg-config` for Ubuntu.
 

adapter/src/ethereum.rs

@@ -679,6 +679,7 @@ mod test {
                 ad_units: vec![],
                 pricing_bounds: None,
             },
+            exhausted: Default::default(),
         };
 
         // convert to eth channel

hooks/post_checkout

@@ -0,0 +1,4 @@
+#!/bin/bash
+# Docker hub does a recursive clone, then checks the branch out,
+# so when a PR adds a submodule (or updates it), it fails.
+git submodule update --init

primitives/src/channel.rs

@@ -98,6 +98,12 @@ pub struct Channel {
     #[serde(default)]
     pub targeting_rules: Rules,
     pub spec: ChannelSpec,
+    #[serde(default)]
+    pub exhausted: Vec<bool>,
+}
+
+pub fn channel_exhausted(channel: &Channel) -> bool {
+    channel.exhausted.len() == 2 && channel.exhausted.iter().all(|&x| x)
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
@@ -226,6 +232,16 @@ impl SpecValidators {
         }
     }
 
+    pub fn find_index(&self, validator_id: &ValidatorId) -> Option<i32> {
+        if &self.leader().id == validator_id {
+            Some(0)
+        } else if &self.follower().id == validator_id {
+            Some(1)
+        } else {
+            None
+        }
+    }
+
     pub fn iter(&self) -> Iter<'_> {
         Iter::new(&self)
     }
@@ -387,6 +403,7 @@ pub mod postgres {
                 valid_until: row.get("valid_until"),
                 targeting_rules: row.get::<_, Json<Rules>>("targeting_rules").0,
                 spec: row.get::<_, Json<ChannelSpec>>("spec").0,
+                exhausted: row.get("exhausted"),
             }
         }
     }

primitives/src/util/tests/prep_db.rs

@@ -83,6 +83,7 @@ lazy_static! {
                 ad_units: vec![],
                 pricing_bounds: Some(PricingBounds {impression: None, click: Some(Pricing { max: 0.into(), min: 0.into()})}),
             },
+            exhausted: Default::default(),
         }
     };
 

primitives/src/validator.rs

@@ -165,6 +165,8 @@ pub struct ApproveState {
     pub state_root: String,
     pub signature: String,
     pub is_healthy: bool,
+    #[serde(default)]
+    pub exhausted: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -173,6 +175,8 @@ pub struct NewState {
     pub state_root: String,
     pub signature: String,
     pub balances: BalancesMap,
+    #[serde(default)]
+    pub exhausted: bool,
 }
 
 #[derive(Default, Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]

sentry/migrations/20190806011140_initial-tables/up.sql

@@ -6,6 +6,7 @@ CREATE TABLE channels
     deposit_amount VARCHAR(255)             NOT NULL,
     valid_until    TIMESTAMP(2) WITH TIME ZONE NOT NULL,
     spec           JSONB                    NOT NULL,
+    exhausted      BOOLEAN[2]
 
     PRIMARY KEY (id)
 );

sentry/src/db/channel.rs

@@ -96,6 +96,29 @@ pub async fn insert_validator_messages(
         .await
 }
 
+pub async fn update_exhausted_channel(
+    pool: &DbPool,
+    channel: &Channel,
+    index: i32,
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool.run(move |connection| async move {
+        match connection
+            .prepare("UPDATE channels SET exhausted[$1] = true WHERE id = $2")
+            .await
+        {
+            Ok(stmt) => match connection.execute(&stmt, &[&index, &channel.id]).await {
+                Ok(row) => {
+                    let updated = row == 1;
+                    Ok((updated, connection))
+                }
+                Err(e) => Err((e, connection)),
+            },
+            Err(e) => Err((e, connection)),
+        }
+    })
+    .await
+}
+
 mod list_channels {
     use crate::db::DbPool;
     use bb8::RunError;

sentry/src/routes/channel.rs

@@ -1,5 +1,8 @@
 use crate::db::event_aggregate::{latest_approve_state, latest_heartbeats, latest_new_state};
-use crate::db::{get_channel_by_id, insert_channel, insert_validator_messages, list_channels};
+use crate::db::{
+    get_channel_by_id, insert_channel, insert_validator_messages, list_channels,
+    update_exhausted_channel,
+};
 use crate::{success_response, Application, Auth, ResponseError, RouteParams, Session};
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::error;
@@ -238,6 +241,12 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
         .get("messages")
         .ok_or_else(|| ResponseError::BadRequest("missing messages body".to_string()))?;
 
+    let channel_is_exhausted = messages.iter().any(|message| match message {
+        MessageTypes::ApproveState(approve) => approve.exhausted,
+        MessageTypes::NewState(new_state) => new_state.exhausted,
+        _ => false,
+    });
+
     match channel.spec.validators.find(&session.uid) {
         None => Err(ResponseError::Unauthorized),
         _ => {
@@ -246,6 +255,12 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
             }))
             .await?;
 
+            if channel_is_exhausted {
+                if let Some(validator_index) = channel.spec.validators.find_index(&session.uid) {
+                    update_exhausted_channel(&app.pool, &channel, validator_index).await?;
+                }
+            }
+
             Ok(success_response(serde_json::to_string(&SuccessResponse {
                 success: true,
             })?))

validator_worker/src/follower.rs

@@ -3,7 +3,7 @@ use std::fmt;
 
 use primitives::adapter::{Adapter, AdapterErrorKind};
 use primitives::validator::{ApproveState, MessageTypes, NewState, RejectState};
-use primitives::BalancesMap;
+use primitives::{BalancesMap, BigNum};
 
 use crate::core::follower_rules::{get_health, is_valid_transition};
 use crate::heartbeat::{heartbeat, HeartbeatStatus};
@@ -140,12 +140,14 @@ async fn on_new_state<'a, A: Adapter + 'static>(
     let signature = iface.adapter.sign(&new_state.state_root)?;
     let health_threshold = u64::from(iface.config.health_threshold_promilles);
     let is_healthy = health >= health_threshold;
+    let exhausted = proposed_balances.values().sum::<BigNum>() == iface.channel.deposit_amount;
 
     let propagation_result = iface
         .propagate(&[&MessageTypes::ApproveState(ApproveState {
             state_root: proposed_state_root,
             signature,
             is_healthy,
+            exhausted,
         })])
         .await;