fix; validator_worker: remove RwLock from Adapter

Author: samparsky

adapter/src/ethereum.rs

@@ -11,7 +11,6 @@ use primitives::{
 };
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::error::Error;
 use std::fs;
@@ -38,8 +37,6 @@ pub struct EthereumAdapter {
     keystore_json: Value,
     keystore_pwd: Password,
     config: Config,
-    // Auth tokens that we've generated to authenticate with someone (address => token)
-    authorization_tokens: HashMap<String, String>,
     wallet: Option<SafeAccount>,
 }
 
@@ -70,7 +67,6 @@ impl EthereumAdapter {
             address,
             keystore_json,
             keystore_pwd: opts.keystore_pwd.into(),
-            authorization_tokens: HashMap::new(),
             wallet: None,
             config: config.to_owned(),
         })
@@ -228,9 +224,10 @@ impl Adapter for EthereumAdapter {
     }
 
     fn get_auth(&self, validator: &ValidatorId) -> AdapterResult<String> {
-        let wallet = self.wallet.clone().ok_or_else(|| AdapterError::Configuration(
-                "failed to unlock wallet".to_string()
-        ))?;
+        let wallet = self
+            .wallet
+            .clone()
+            .ok_or_else(|| AdapterError::Configuration("failed to unlock wallet".to_string()))?;
 
         let era = Utc::now().timestamp_millis() as f64 / 60000.0;
         let payload = Payload {
@@ -241,7 +238,7 @@ impl Adapter for EthereumAdapter {
         };
         let token = ewt_sign(&wallet, &self.keystore_pwd, &payload)
             .map_err(|_| map_error("Failed to sign token"))?;
-        
+
         Ok(token)
     }
 }

primitives/src/adapter.rs

@@ -59,7 +59,7 @@ pub struct Session {
     pub uid: ValidatorId,
 }
 
-pub trait Adapter: ChannelValidator + Send + Clone + Debug {
+pub trait Adapter: ChannelValidator + Send + Sync + Clone + Debug {
     /// Unlock adapter
     fn unlock(&mut self) -> AdapterResult<()>;
 

validator_worker/src/follower.rs

@@ -64,9 +64,7 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         return Ok(on_error(&iface, &new_state, InvalidNewState::RootHash).await);
     }
 
-    let adapter = iface.adapter.read().await.clone();
-
-    if !adapter.verify(
+    if !iface.adapter.verify(
         &iface.channel.spec.validators.leader().id,
         &proposed_state_root,
         &new_state.signature,
@@ -84,7 +82,7 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         return Ok(on_error(&iface, &new_state, InvalidNewState::Transition).await);
     }
 
-    let signature = adapter.sign(&new_state.state_root)?;
+    let signature = iface.adapter.sign(&new_state.state_root)?;
     let health_threshold = u64::from(iface.config.health_threshold_promilles).into();
     let health = is_healthy(
         &iface.channel,

validator_worker/src/heartbeat.rs

@@ -23,7 +23,7 @@ async fn send_heartbeat<A: Adapter + 'static>(iface: &SentryApi<A>) -> Result<()
     let state_root_raw = get_signable_state_root(&iface.channel.id, &merkle_tree.root())?;
     let state_root = hex::encode(state_root_raw);
 
-    let signature = iface.adapter.read().await.sign(&state_root)?;
+    let signature = iface.adapter.sign(&state_root)?;
 
     let message_types = MessageTypes::Heartbeat(Heartbeat {
         signature,

validator_worker/src/leader.rs

@@ -25,7 +25,7 @@ async fn on_new_accounting<A: Adapter + 'static>(
     let state_root_raw = get_state_root_hash(&iface, &balances)?;
     let state_root = hex::encode(state_root_raw);
 
-    let signature = iface.adapter.read().await.sign(&state_root)?;
+    let signature = iface.adapter.sign(&state_root)?;
 
     iface
         .propagate(&[&MessageTypes::NewState(NewState {

validator_worker/src/lib.rs

@@ -44,12 +44,10 @@ mod test {
     use super::*;
 
     use adapter::DummyAdapter;
-    use async_std::sync::RwLock;
     use primitives::adapter::DummyAdapterOptions;
     use primitives::config::configuration;
     use primitives::util::tests::prep_db::{AUTH, DUMMY_CHANNEL, IDS};
     use primitives::{BalancesMap, Channel};
-    use std::sync::Arc;
 
     fn setup_iface(channel: &Channel) -> SentryApi<DummyAdapter> {
         let adapter_options = DummyAdapterOptions {
@@ -61,14 +59,7 @@ mod test {
         let dummy_adapter = DummyAdapter::init(adapter_options, &config);
         let whoami = dummy_adapter.whoami().clone();
 
-        SentryApi::init(
-            Arc::new(RwLock::new(dummy_adapter)),
-            &channel,
-            &config,
-            false,
-            &whoami,
-        )
-        .expect("should succeed")
+        SentryApi::init(dummy_adapter, &channel, &config, false, &whoami).expect("should succeed")
     }
 
     #[test]

validator_worker/src/main.rs

@@ -4,7 +4,6 @@
 use clap::{App, Arg};
 
 use adapter::{AdapterTypes, DummyAdapter, EthereumAdapter};
-use async_std::sync::RwLock;
 use futures::compat::Future01CompatExt;
 use futures::future::try_join_all;
 use futures::future::{join, FutureExt, TryFutureExt};
@@ -15,7 +14,6 @@ use primitives::{Channel, SpecValidator, ValidatorId};
 use std::convert::TryFrom;
 use std::error::Error;
 use std::ops::Add;
-use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::timer::Delay;
 use tokio::util::FutureExt as TokioFutureExt;
@@ -26,7 +24,7 @@ use validator_worker::{all_channels, follower, leader, SentryApi};
 struct Args<A: Adapter> {
     sentry_url: String,
     config: Config,
-    adapter: Arc<RwLock<A>>,
+    adapter: A,
     whoami: ValidatorId,
 }
 
@@ -126,7 +124,9 @@ fn run<A: Adapter + 'static>(
     config: &Config,
     adapter: A,
 ) -> Result<(), Box<dyn Error>> {
-    let sentry_adapter = Arc::new(RwLock::new(adapter.clone()));
+    let mut sentry_adapter = adapter.clone();
+    // unlock adapter
+    sentry_adapter.unlock()?;
     let whoami = adapter.whoami().to_owned();
 
     let args = Args {
@@ -188,7 +188,7 @@ async fn iterate_channels<A: Adapter + 'static>(args: Args<A>) -> Result<(), ()>
 }
 
 async fn validator_tick<A: Adapter + 'static>(
-    adapter: Arc<RwLock<A>>,
+    adapter: A,
     channel: Channel,
     config: &Config,
     whoami: &ValidatorId,

validator_worker/src/sentry_interface.rs

@@ -1,5 +1,4 @@
 use crate::error::ValidatorWorker;
-use async_std::sync::RwLock;
 use chrono::{DateTime, Utc};
 use futures::compat::Future01CompatExt;
 use futures::future::try_join_all;
@@ -14,23 +13,23 @@ use primitives::validator::MessageTypes;
 use primitives::{Channel, Config, ValidatorDesc, ValidatorId};
 use reqwest::r#async::{Client, Response};
 use std::collections::HashMap;
-use std::sync::Arc;
 use std::time::Duration;
 
 #[derive(Debug, Clone)]
 pub struct SentryApi<T: Adapter> {
-    pub adapter: Arc<RwLock<T>>,
+    pub adapter: T,
     pub validator_url: String,
     pub client: Client,
     pub logging: bool,
     pub channel: Channel,
     pub config: Config,
     pub whoami: ValidatorId,
+    pub propagate_to: Vec<(ValidatorDesc, String)>,
 }
 
 impl<T: Adapter + 'static> SentryApi<T> {
     pub fn init(
-        adapter: Arc<RwLock<T>>,
+        adapter: T,
         channel: &Channel,
         config: &Config,
         logging: bool,
@@ -46,12 +45,29 @@ impl<T: Adapter + 'static> SentryApi<T> {
             SpecValidator::Leader(v) | SpecValidator::Follower(v) => {
                 let channel_id = format!("0x{}", hex::encode(&channel.id));
                 let validator_url = format!("{}/channel/{}", v.url, channel_id);
+                let propagate_to = channel
+                    .spec
+                    .validators
+                    .into_iter()
+                    .map(|validator| {
+                        adapter
+                            .get_auth(&validator.id)
+                            .map(|auth| (validator.to_owned(), auth))
+                            .map_err(|e| {
+                                ValidatorWorker::Failed(format!(
+                                    "propagate error: get auth failed {}",
+                                    e
+                                ))
+                            })
+                    })
+                    .collect::<Result<Vec<_>, _>>()?;
 
                 Ok(Self {
                     adapter,
                     validator_url,
                     client,
                     logging,
+                    propagate_to,
                     channel: channel.to_owned(),
                     config: config.to_owned(),
                     whoami: whoami.to_owned(),
@@ -64,29 +80,13 @@ impl<T: Adapter + 'static> SentryApi<T> {
     }
 
     pub async fn propagate(&self, messages: &[&MessageTypes]) {
-        let mut adapter = self.adapter.write().await;
-
         let channel_id = format!("0x{}", hex::encode(&self.channel.id));
-
-        for validator in self.channel.spec.validators.into_iter() {
-            let auth_token = adapter.get_auth(&validator.id);
-
-            if let Err(e) = auth_token {
-                println!("propagate error: get auth failed {}", e);
-                continue;
-            }
-
-            if let Err(e) = propagate_to(
-                &channel_id,
-                &auth_token.unwrap(),
-                &self.client,
-                &validator,
-                messages,
-            )
-            .await
-            {
-                handle_http_error(e, &validator.url);
-            }
+        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
+        {
+            handle_http_error(e);
         }
     }
 
@@ -148,8 +148,6 @@ impl<T: Adapter + 'static> SentryApi<T> {
     ) -> Result<EventAggregateResponse, Box<ValidatorWorker>> {
         let auth_token = self
             .adapter
-            .write()
-            .await
             .get_auth(&self.whoami)
             .map_err(|e| Box::new(ValidatorWorker::Failed(e.to_string())))?;
 
@@ -201,11 +199,11 @@ async fn propagate_to(
     Ok(())
 }
 
-fn handle_http_error(e: reqwest::Error, url: &str) {
+fn handle_http_error(e: reqwest::Error) {
     if e.is_http() {
         match e.url() {
             None => println!("No Url given"),
-            Some(url) => println!("Problem making request to: {}", url),
+            Some(url) => println!("erorr sending http request for validator {}", url),
         }
     }
     // Inspect the internal error and output it
@@ -217,10 +215,6 @@ fn handle_http_error(e: reqwest::Error, url: &str) {
         println!("problem parsing information {}", serde_error);
     }
 
-    if e.is_client_error() {
-        println!("erorr sending http request for validator {}", url)
-    }
-
     if e.is_redirect() {
         println!("server redirecting too many times or making loop");
     }