fix: remove RefCell adapter, implement RwLock sentry_interface

Author: samparsky

adapter/src/dummy.rs

@@ -42,7 +42,7 @@ impl Adapter for DummyAdapter {
         })
     }
 
-    fn unlock(&self) -> AdapterResult<()> {
+    fn unlock(&mut self) -> AdapterResult<()> {
         Ok(())
     }
 
@@ -77,7 +77,7 @@ impl Adapter for DummyAdapter {
         }
     }
 
-    fn session_from_token(&self, token: &str) -> AdapterResult<Session> {
+    fn session_from_token(&mut self, token: &str) -> AdapterResult<Session> {
         let identity = self
             .authorization_tokens
             .iter()
@@ -95,7 +95,7 @@ impl Adapter for DummyAdapter {
         }
     }
 
-    fn get_auth(&self, _validator: &ValidatorDesc) -> AdapterResult<String> {
+    fn get_auth(&mut self, _validator: &ValidatorDesc) -> AdapterResult<String> {
         let who = self
             .session_tokens
             .iter()

adapter/src/ethereum.rs

@@ -11,7 +11,6 @@ use primitives::{
 };
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use std::cell::RefCell;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fs;
@@ -39,10 +38,10 @@ pub struct EthereumAdapter {
     keystore_pwd: Password,
     config: Config,
     // Auth tokens that we have verified (tokenId => session)
-    session_tokens: RefCell<HashMap<String, Session>>,
+    session_tokens: HashMap<String, Session>,
     // Auth tokens that we've generated to authenticate with someone (address => token)
-    authorization_tokens: RefCell<HashMap<String, String>>,
-    wallet: RefCell<Option<SafeAccount>>,
+    authorization_tokens: HashMap<String, String>,
+    wallet: Option<SafeAccount>,
 }
 
 // Enables EthereumAdapter to be able to
@@ -81,14 +80,14 @@ impl Adapter for EthereumAdapter {
             address,
             keystore_json,
             keystore_pwd: keystore_pwd.into(),
-            session_tokens: RefCell::new(HashMap::new()),
-            authorization_tokens: RefCell::new(HashMap::new()),
-            wallet: RefCell::new(None),
+            session_tokens: HashMap::new(),
+            authorization_tokens: HashMap::new(),
+            wallet: None,
             config: config.to_owned(),
         })
     }
 
-    fn unlock(&self) -> AdapterResult<()> {
+    fn unlock(&mut self) -> AdapterResult<()> {
         let account = SafeAccount::from_file(
             serde_json::from_value(self.keystore_json.clone())
                 .map_err(|_| map_error("Invalid keystore json provided"))?,
@@ -97,7 +96,7 @@ impl Adapter for EthereumAdapter {
         )
         .map_err(|_| map_error("Failed to create account"))?;
 
-        self.wallet.replace(Some(account));
+        self.wallet = Some(account);
 
         Ok(())
     }
@@ -107,18 +106,18 @@ impl Adapter for EthereumAdapter {
     }
 
     fn sign(&self, state_root: &str) -> AdapterResult<String> {
-        let message = Message::from_slice(&hash_message(state_root));
-        match self.wallet.borrow().clone() {
-            Some(wallet) => {
-                let wallet_sign = wallet
-                    .sign(&self.keystore_pwd, &message)
-                    .map_err(|_| map_error("failed to sign messages"))?;
-                let signature: Signature = wallet_sign.into_electrum().into();
-                Ok(format!("0x{}", signature))
-            }
-            None => Err(AdapterError::Configuration(
+        if let Some(wallet) = &self.wallet {
+            let message = Message::from_slice(&hash_message(state_root));
+            let wallet_sign = wallet
+                .sign(&self.keystore_pwd, &message)
+                .map_err(|_| map_error("failed to sign messages"))?;
+            let signature: Signature = wallet_sign.into_electrum().into();
+
+            Ok(format!("0x{}", signature))
+        } else {
+            Err(AdapterError::Configuration(
                 "Unlock the wallet before signing".to_string(),
-            )),
+            ))
         }
     }
 
@@ -136,7 +135,7 @@ impl Adapter for EthereumAdapter {
         let signature = Signature::from_electrum(&decoded_signature);
         let message = Message::from_slice(&hash_message(state_root));
 
-        verify_address(&address, &signature, &message).or(Ok(false))
+        verify_address(&address, &signature, &message).or_else(|_| Ok(false))
     }
 
     fn validate_channel(&self, channel: &Channel) -> AdapterResult<bool> {
@@ -179,15 +178,14 @@ impl Adapter for EthereumAdapter {
         Ok(true)
     }
 
-    fn session_from_token(&self, token: &str) -> AdapterResult<Session> {
+    fn session_from_token(&mut self, token: &str) -> AdapterResult<Session> {
         if token.len() < 16 {
             return Err(AdapterError::Failed("invaild token id".to_string()));
         }
 
         let token_id = token[token.len() - 16..].to_string();
 
-        let mut session_tokens = self.session_tokens.borrow_mut();
-        if let Some(token) = session_tokens.get(&token_id) {
+        if let Some(token) = self.session_tokens.get(&token_id) {
             return Ok(token.to_owned());
         }
 
@@ -241,16 +239,12 @@ impl Adapter for EthereumAdapter {
             },
         };
 
-        session_tokens.insert(token_id, sess.clone());
+        self.session_tokens.insert(token_id, sess.clone());
         Ok(sess)
     }
 
-    fn get_auth(&self, validator: &ValidatorDesc) -> AdapterResult<String> {
-        let mut authorization_tokens = self.authorization_tokens.borrow_mut();
-        match (
-            self.wallet.borrow().clone(),
-            authorization_tokens.get(&validator.id),
-        ) {
+    fn get_auth(&mut self, validator: &ValidatorDesc) -> AdapterResult<String> {
+        match (&self.wallet, self.authorization_tokens.get(&validator.id)) {
             (Some(_), Some(token)) => Ok(token.to_owned()),
             (Some(wallet), None) => {
                 let era = Utc::now().timestamp() as f64 / 60000.0;
@@ -260,10 +254,11 @@ impl Adapter for EthereumAdapter {
                     identity: None,
                     address: None,
                 };
-                let token = ewt_sign(&wallet, &self.keystore_pwd, &payload)
+                let token = ewt_sign(wallet, &self.keystore_pwd, &payload)
                     .map_err(|_| map_error("Failed to sign token"))?;
 
-                authorization_tokens.insert(validator.id.clone(), token.clone());
+                self.authorization_tokens
+                    .insert(validator.id.clone(), token.clone());
 
                 Ok(token)
             }
@@ -420,7 +415,7 @@ mod test {
 
     #[test]
     fn should_init_and_unlock_ethereum_adapter() {
-        let eth_adapter = setup_eth_adapter();
+        let mut eth_adapter = setup_eth_adapter();
         let unlock = eth_adapter.unlock().expect("should unlock eth adapter");
 
         assert_eq!((), unlock, "failed to unlock eth adapter");
@@ -429,7 +424,7 @@ mod test {
     #[test]
     fn should_get_whoami_sign_and_verify_messages() {
         // whoami
-        let eth_adapter = setup_eth_adapter();
+        let mut eth_adapter = setup_eth_adapter();
         let whoami = eth_adapter.whoami();
         assert_eq!(
             whoami, "0x2bDeAFAE53940669DaA6F519373f686c1f3d3393",
@@ -460,7 +455,7 @@ mod test {
 
     #[test]
     fn should_generate_correct_ewt_sign_and_verify() {
-        let eth_adapter = setup_eth_adapter();
+        let mut eth_adapter = setup_eth_adapter();
         eth_adapter.unlock().expect("should unlock eth adapter");
 
         let payload = Payload {
@@ -469,7 +464,7 @@ mod test {
             address: Some(eth_adapter.whoami()),
             identity: None,
         };
-        let wallet = eth_adapter.wallet.borrow().clone();
+        let wallet = eth_adapter.wallet.clone();
         let response = ewt_sign(&wallet.unwrap(), &eth_adapter.keystore_pwd, &payload)
             .expect("failed to generate ewt signature");
         let expected =

primitives/src/adapter.rs

@@ -56,7 +56,7 @@ pub trait Adapter: ChannelValidator + Clone + Debug {
     fn init(opts: AdapterOptions, config: &Config) -> AdapterResult<Self::Output>;
 
     /// Unlock adapter
-    fn unlock(&self) -> AdapterResult<()>;
+    fn unlock(&mut self) -> AdapterResult<()>;
 
     /// Get Adapter whoami
     fn whoami(&self) -> String;
@@ -71,8 +71,8 @@ pub trait Adapter: ChannelValidator + Clone + Debug {
     fn validate_channel(&self, channel: &Channel) -> AdapterResult<bool>;
 
     /// Get user session from token
-    fn session_from_token(&self, token: &str) -> AdapterResult<Session>;
+    fn session_from_token(&mut self, token: &str) -> AdapterResult<Session>;
 
     /// Gets authentication for specific validator
-    fn get_auth(&self, validator: &ValidatorDesc) -> AdapterResult<String>;
+    fn get_auth(&mut self, validator: &ValidatorDesc) -> AdapterResult<String>;
 }

validator_worker/src/error.rs

@@ -1,9 +1,20 @@
-#![deny(clippy::all)]
-
 use serde::{Deserialize, Serialize};
+use std::error::Error;
+use std::fmt::{Display, Formatter, Result};
+
+#[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
+pub enum ValidatorWorker {
+    Configuration(String),
+    Failed(String),
+}
+
+impl Error for ValidatorWorker {}
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
-pub enum ValidatorWorkerError {
-    ConfigurationError(String),
-    InvalidValidatorEntry(String),
+impl Display for ValidatorWorker {
+    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
+        match self {
+            ValidatorWorker::Configuration(error) => write!(f, "Configuration error: {}", error),
+            ValidatorWorker::Failed(error) => write!(f, "error: {}", error),
+        }
+    }
 }

validator_worker/src/follower.rs

@@ -72,7 +72,12 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         )));
     }
 
-    if !iface.adapter.verify(
+    let adapter = iface
+        .adapter
+        .read()
+        .expect("on_new_state: failed to acquire read lock adapter");
+
+    if !adapter.verify(
         &iface.channel.spec.validators.leader().id,
         &proposed_state_root,
         &new_state.signature,
@@ -98,7 +103,7 @@ async fn on_new_state<'a, A: Adapter + 'static>(
         )));
     }
 
-    let signature = iface.adapter.sign(&new_state.state_root)?;
+    let signature = adapter.sign(&new_state.state_root)?;
     let health_threshold = u64::from(iface.config.health_threshold_promilles).into();
 
     iface.propagate(&[&MessageTypes::ApproveState(ApproveState {

validator_worker/src/heartbeat.rs

@@ -23,7 +23,11 @@ async fn send_heartbeat<A: Adapter + 'static>(iface: &SentryApi<A>) -> Result<()
 
     let state_root_raw = get_signable_state_root(&iface.channel.id, &info_root_raw)?;
     let state_root = hex::encode(state_root_raw);
-    let signature = iface.adapter.sign(&state_root)?;
+    let signature = iface
+        .adapter
+        .read()
+        .expect("send_heartbeat: failed to acquire adapter read lock")
+        .sign(&state_root)?;
 
     let message_types = MessageTypes::Heartbeat(Heartbeat {
         signature,

validator_worker/src/leader.rs

@@ -25,7 +25,11 @@ 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.sign(&state_root)?;
+    let signature = iface
+        .adapter
+        .read()
+        .expect("on_new_accounting: failed to acquire read lock")
+        .sign(&state_root)?;
 
     iface.propagate(&[&MessageTypes::NewState(NewState {
         state_root,

validator_worker/src/lib.rs

@@ -5,13 +5,12 @@
 
 use std::error::Error;
 
+use crate::sentry_interface::SentryApi;
 use adapter::{get_balance_leaf, get_signable_state_root};
 use primitives::adapter::Adapter;
 use primitives::merkle_tree::MerkleTree;
 use primitives::BalancesMap;
 
-use crate::sentry_interface::SentryApi;
-
 pub use self::sentry_interface::all_channels;
 
 pub mod error;
@@ -54,6 +53,7 @@ mod test {
     use primitives::config::configuration;
     use primitives::util::tests::prep_db::{AUTH, DUMMY_CHANNEL, IDS};
     use primitives::{BalancesMap, Channel};
+    use std::sync::{Arc, RwLock};
 
     fn setup_iface(channel: &Channel) -> SentryApi<DummyAdapter> {
         let adapter_options = AdapterOptions {
@@ -66,7 +66,13 @@ mod test {
         let config = configuration("development", None).expect("Dev config should be available");
         let dummy_adapter = DummyAdapter::init(adapter_options, &config).expect("init adadpter");
 
-        SentryApi::new(dummy_adapter, &channel, &config, false).expect("should succeed")
+        SentryApi::new(
+            Arc::new(RwLock::new(dummy_adapter)),
+            &channel,
+            &config,
+            false,
+        )
+        .expect("should succeed")
     }
 
     #[test]