Merge branch 'improvements' into add-exhausted-property

Author: elpiel

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

@@ -99,11 +99,11 @@ pub struct Channel {
     pub targeting_rules: Rules,
     pub spec: ChannelSpec,
     #[serde(default)]
-    pub exhausted: Vec<bool>
+    pub exhausted: Vec<bool>,
 }
 
 pub fn channel_exhausted(channel: &Channel) -> bool {
-    channel.exhausted.len() == 2 && channel.exhausted.iter().all(|&x| x == true)
+    channel.exhausted.len() == 2 && channel.exhausted.iter().all(|&x| x)
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]

sentry/src/db/channel.rs

@@ -101,23 +101,22 @@ pub async fn update_exhausted_channel(
     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)),
+    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))
                 }
-            }
-        })
-        .await
-
+                Err(e) => Err((e, connection)),
+            },
+            Err(e) => Err((e, connection)),
+        }
+    })
+    .await
 }
 
 mod list_channels {

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, update_exhausted_channel};
+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;
@@ -244,7 +247,6 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
         _ => false,
     });
 
-
     match channel.spec.validators.find(&session.uid) {
         None => Err(ResponseError::Unauthorized),
         _ => {
@@ -255,11 +257,7 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
 
             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?;
+                    update_exhausted_channel(&app.pool, &channel, validator_index).await?;
                 }
             }
 

validator_worker/src/leader.rs

@@ -3,8 +3,7 @@ use std::error::Error;
 use primitives::adapter::{Adapter, AdapterErrorKind};
 use primitives::{
     validator::{Accounting, MessageTypes, NewState},
-    BalancesMap,
-    BigNum,
+    BalancesMap, BigNum,
 };
 
 use crate::heartbeat::{heartbeat, HeartbeatStatus};
@@ -49,16 +48,17 @@ async fn on_new_accounting<A: Adapter + 'static>(
 
     let signature = iface.adapter.sign(&state_root)?;
 
-    let exhausted = new_accounting.balances.values().sum::<BigNum>() == iface.channel.deposit_amount;
+    let exhausted =
+        new_accounting.balances.values().sum::<BigNum>() == iface.channel.deposit_amount;
 
     let propagation_results = iface
         .propagate(&[&MessageTypes::NewState(NewState {
             state_root,
             signature,
             balances: new_accounting.balances.clone(),
-            exhausted
+            exhausted,
         })])
         .await;
 
     Ok(propagation_results)
-}
+}