Replace runtime in orchestrator to enable multithreading

orchestrator/cosmos_gravity/src/build.rs

@@ -16,7 +16,7 @@ use std::collections::BTreeMap;
 
 use crate::crypto::PrivateKey as CosmosPrivateKey;
 
-lazy_static!{
+lazy_static! {
     static ref DENOM_REGEX: Regex = Regex::new("^[a-zA-Z][a-zA-Z0-9/-]{2,127}$").unwrap();
 }
 
@@ -247,7 +247,7 @@ pub fn ethereum_event_messages(
 // be rejected by the chain when processed.
 pub fn denom_string(input_denom: String) -> String {
     if !DENOM_REGEX.is_match(input_denom.as_str()) {
-        return "invalid".to_string()
+        return "invalid".to_string();
     }
 
     input_denom

orchestrator/cosmos_gravity/src/send.rs

@@ -226,4 +226,4 @@ fn log_send_error(messages: &Vec<Msg>, err: GravityError) {
         msg_types,
         err
     );
-}
+}

orchestrator/ethereum_gravity/src/logic_call.rs

@@ -9,7 +9,7 @@ use gravity_abi::gravity::*;
 use gravity_utils::ethereum::{bytes_to_hex_str, vec_u8_to_fixed_32};
 use gravity_utils::types::*;
 use gravity_utils::{error::GravityError, message_signatures::encode_logic_call_confirm_hashed};
-use std::{result::Result, time::Duration, collections::HashMap};
+use std::{collections::HashMap, result::Result, time::Duration};
 
 /// this function generates an appropriate Ethereum transaction
 /// to submit the provided logic call
@@ -262,7 +262,8 @@ impl LogicCallSkips {
         if id_skips.is_none() {
             // first time we've seen this invalidation id, start at 2 skips
             let new_id_skips = HashMap::from([(call.invalidation_nonce, new_skip_state)]);
-            self.skip_map.insert(call.invalidation_id.clone(), new_id_skips);
+            self.skip_map
+                .insert(call.invalidation_id.clone(), new_id_skips);
         } else {
             let id_skips = id_skips.unwrap();
             let skip_state = id_skips.get_mut(&call.invalidation_nonce);

orchestrator/ethereum_gravity/src/utils.rs

@@ -5,7 +5,7 @@ use ethers::prelude::*;
 use ethers::types::Address as EthAddress;
 use gravity_abi::gravity::*;
 use gravity_utils::error::GravityError;
-use gravity_utils::ethereum::{downcast_to_u64, vec_u8_to_fixed_32, hex_str_to_bytes};
+use gravity_utils::ethereum::{downcast_to_u64, hex_str_to_bytes, vec_u8_to_fixed_32};
 use gravity_utils::types::{decode_gravity_error, GravityContractError};
 use std::result::Result;
 
@@ -199,15 +199,23 @@ pub fn handle_contract_error(gravity_error: GravityError) -> bool {
     if gravity_contract_error.is_some() {
         match gravity_contract_error.unwrap() {
             GravityContractError::InvalidLogicCallNonce(nonce_error) => {
-                info!("LogicCall already processed, skipping until observed on chain: {}", nonce_error.message());
+                info!(
+                    "LogicCall already processed, skipping until observed on chain: {}",
+                    nonce_error.message()
+                );
                 return true;
             }
             GravityContractError::LogicCallTimedOut(timeout_error) => {
-                info!("LogicCall is timed out, will be skipped until timeout on chain: {}", timeout_error.message());
+                info!(
+                    "LogicCall is timed out, will be skipped until timeout on chain: {}",
+                    timeout_error.message()
+                );
                 return true;
             }
             // TODO(bolten): implement other cases if necessary
-            _ => { error!("Unspecified gravity contract error: {}", error_string) }
+            _ => {
+                error!("Unspecified gravity contract error: {}", error_string)
+            }
         }
     } else {
         error!("Non-gravity contract error: {}", error_string);
@@ -220,48 +228,47 @@ pub fn handle_contract_error(gravity_error: GravityError) -> bool {
 // results in this nightmare
 pub fn extract_gravity_contract_error(gravity_error: GravityError) -> Option<GravityContractError> {
     match gravity_error {
-        GravityError::EthersContractError(ce) => {
-            match ce {
-                ethers::contract::ContractError::MiddlewareError(me) => {
-                    match me {
-                        ethers::middleware::signer::SignerMiddlewareError::MiddlewareError(sme) => {
-                            match sme {
-                                ethers::providers::ProviderError::JsonRpcClientError(jrpce) => {
-                                    if jrpce.is::<ethers::providers::HttpClientError>() {
-                                        let httpe = *jrpce.downcast::<ethers::providers::HttpClientError>().unwrap();
-                                        match httpe {
-                                            ethers::providers::HttpClientError::JsonRpcError(jre) => {
-                                                if jre.code == 3 && jre.data.is_some() {
-                                                    let data = jre.data.unwrap();
-                                                    if data.is_string() {
-                                                        let data_bytes = hex_str_to_bytes(data.as_str().unwrap());
-                                                        if data_bytes.is_ok() {
-                                                            decode_gravity_error(data_bytes.unwrap())
-                                                        } else {
-                                                            None
-                                                        }
-                                                    } else {
-                                                        None
-                                                    }
+        GravityError::EthersContractError(ce) => match ce {
+            ethers::contract::ContractError::MiddlewareError(me) => match me {
+                ethers::middleware::signer::SignerMiddlewareError::MiddlewareError(sme) => {
+                    match sme {
+                        ethers::providers::ProviderError::JsonRpcClientError(jrpce) => {
+                            if jrpce.is::<ethers::providers::HttpClientError>() {
+                                let httpe = *jrpce
+                                    .downcast::<ethers::providers::HttpClientError>()
+                                    .unwrap();
+                                match httpe {
+                                    ethers::providers::HttpClientError::JsonRpcError(jre) => {
+                                        if jre.code == 3 && jre.data.is_some() {
+                                            let data = jre.data.unwrap();
+                                            if data.is_string() {
+                                                let data_bytes =
+                                                    hex_str_to_bytes(data.as_str().unwrap());
+                                                if data_bytes.is_ok() {
+                                                    decode_gravity_error(data_bytes.unwrap())
                                                 } else {
                                                     None
                                                 }
+                                            } else {
+                                                None
                                             }
-                                            _ => None
+                                        } else {
+                                            None
                                         }
-                                    } else {
-                                        None
                                     }
+                                    _ => None,
                                 }
-                                _ => None
+                            } else {
+                                None
                             }
                         }
-                        _ => None
+                        _ => None,
                     }
                 }
-                _ => None
-            }
-        }
-        _ => None
+                _ => None,
+            },
+            _ => None,
+        },
+        _ => None,
     }
 }

orchestrator/gorc/Cargo.toml

@@ -21,7 +21,6 @@ relayer = { path = "../relayer" }
 deep_space = { git = "https://github.com/iqlusioninc/deep_space/", branch = "master" }
 ethers = { git = "https://github.com/iqlusioninc/ethers-rs.git", branch = "zaki/error_abi_support", features = ["abigen"] }
 clarity = "0.4.12"
-actix-rt = "2.5"
 rpassword = "5"
 bip32 = "0.2"
 k256 = { version = "0.9", features = ["pem"] }

orchestrator/gravity_abi/src/lib.rs

@@ -1,2 +1,2 @@
 pub mod erc20;
-pub mod gravity;
+pub mod gravity;

orchestrator/gravity_utils/src/types/gravity_contract_errors.rs

@@ -1,7 +1,7 @@
-use std::convert::TryInto;
 use ethers::prelude::*;
 use ethers::utils::keccak256;
 use lazy_static::lazy_static;
+use std::convert::TryInto;
 
 fn err_to_selector(error: &str) -> [u8; 4] {
     keccak256(error)[0..4].try_into().unwrap()
@@ -11,15 +11,19 @@ fn err_to_selector(error: &str) -> [u8; 4] {
 // here by calculating the first hash bytes rather than computing them by hand in advance
 lazy_static! {
     static ref INVALID_SIGNATURE: [u8; 4] = err_to_selector("InvalidSignature()");
-    static ref INVALID_VALSET_NONCE: [u8; 4] = err_to_selector("InvalidValsetNonce(uint256,uint256)");
+    static ref INVALID_VALSET_NONCE: [u8; 4] =
+        err_to_selector("InvalidValsetNonce(uint256,uint256)");
     static ref INVALID_BATCH_NONCE: [u8; 4] = err_to_selector("InvalidBatchNonce(uint256,uint256)");
-    static ref INVALID_LOGIC_CALL_NONCE: [u8; 4] = err_to_selector("InvalidLogicCallNonce(uint256,uint256)");
-    static ref INVALID_LOGIC_CALL_TRANSFERS: [u8; 4] = err_to_selector("InvalidLogicCallTransfers()");
+    static ref INVALID_LOGIC_CALL_NONCE: [u8; 4] =
+        err_to_selector("InvalidLogicCallNonce(uint256,uint256)");
+    static ref INVALID_LOGIC_CALL_TRANSFERS: [u8; 4] =
+        err_to_selector("InvalidLogicCallTransfers()");
     static ref INVALID_LOGIC_CALL_FEES: [u8; 4] = err_to_selector("InvalidLogicCallFees()");
     static ref INVALID_SEND_TO_COSMOS: [u8; 4] = err_to_selector("InvalidSendToCosmos()");
     static ref INCORRECT_CHECKPOINT: [u8; 4] = err_to_selector("IncorrectCheckpoint()");
     static ref MALFORMED_NEW_VALIDATOR_SET: [u8; 4] = err_to_selector("MalformedNewValidatorSet()");
-    static ref MALFORMED_CURRENT_VALIDATOR_SET: [u8; 4] = err_to_selector("MalformedCurrentValidatorSet()");
+    static ref MALFORMED_CURRENT_VALIDATOR_SET: [u8; 4] =
+        err_to_selector("MalformedCurrentValidatorSet()");
     static ref MALFORMED_BATCH: [u8; 4] = err_to_selector("MalformedBatch()");
     static ref INSUFFICIENT_POWER: [u8; 4] = err_to_selector("InsufficientPower(uint256,uint256)");
     static ref BATCH_TIMED_OUT: [u8; 4] = err_to_selector("BatchTimedOut()");
@@ -53,87 +57,105 @@ pub fn decode_gravity_error(data: Vec<u8>) -> Option<GravityContractError> {
     let selector: [u8; 4] = data[0..4].try_into().unwrap();
 
     if selector == INVALID_SIGNATURE.as_slice() {
-        return Some(GravityContractError::InvalidSignature(InvalidSignature{}))
+        return Some(GravityContractError::InvalidSignature(InvalidSignature {}));
     }
 
     if selector == INVALID_VALSET_NONCE.as_slice() {
         if data.len() != 68 {
-            return None
+            return None;
         }
 
-        return Some(GravityContractError::InvalidValsetNonce(InvalidValsetNonce{
-            new_nonce: data[4..36].into(),
-            current_nonce: data[36..].into(),
-        }))
+        return Some(GravityContractError::InvalidValsetNonce(
+            InvalidValsetNonce {
+                new_nonce: data[4..36].into(),
+                current_nonce: data[36..].into(),
+            },
+        ));
     }
 
     if selector == INVALID_BATCH_NONCE.as_slice() {
         if data.len() != 68 {
-            return None
+            return None;
         }
 
-        return Some(GravityContractError::InvalidBatchNonce(InvalidBatchNonce{
+        return Some(GravityContractError::InvalidBatchNonce(InvalidBatchNonce {
             new_nonce: data[4..36].into(),
             current_nonce: data[36..].into(),
-        }))
+        }));
     }
 
     if selector == INVALID_LOGIC_CALL_NONCE.as_slice() {
         if data.len() != 68 {
-            return None
+            return None;
         }
 
-        return Some(GravityContractError::InvalidLogicCallNonce(InvalidLogicCallNonce{
-            new_nonce: data[4..36].into(),
-            current_nonce: data[36..].into(),
-        }))
+        return Some(GravityContractError::InvalidLogicCallNonce(
+            InvalidLogicCallNonce {
+                new_nonce: data[4..36].into(),
+                current_nonce: data[36..].into(),
+            },
+        ));
     }
 
     if selector == INVALID_LOGIC_CALL_TRANSFERS.as_slice() {
-        return Some(GravityContractError::InvalidLogicCallTransfers(InvalidLogicCallTransfers{}))
+        return Some(GravityContractError::InvalidLogicCallTransfers(
+            InvalidLogicCallTransfers {},
+        ));
     }
 
     if selector == INVALID_LOGIC_CALL_FEES.as_slice() {
-        return Some(GravityContractError::InvalidLogicCallFees(InvalidLogicCallFees{}))
+        return Some(GravityContractError::InvalidLogicCallFees(
+            InvalidLogicCallFees {},
+        ));
     }
 
     if selector == INVALID_SEND_TO_COSMOS.as_slice() {
-        return Some(GravityContractError::InvalidSendToCosmos(InvalidSendToCosmos{}))
+        return Some(GravityContractError::InvalidSendToCosmos(
+            InvalidSendToCosmos {},
+        ));
     }
 
     if selector == INCORRECT_CHECKPOINT.as_slice() {
-        return Some(GravityContractError::InvalidLogicCallTransfers(InvalidLogicCallTransfers{}))
+        return Some(GravityContractError::InvalidLogicCallTransfers(
+            InvalidLogicCallTransfers {},
+        ));
     }
 
     if selector == MALFORMED_NEW_VALIDATOR_SET.as_slice() {
-        return Some(GravityContractError::MalformedNewValidatorSet(MalformedNewValidatorSet{}))
+        return Some(GravityContractError::MalformedNewValidatorSet(
+            MalformedNewValidatorSet {},
+        ));
     }
 
     if selector == MALFORMED_CURRENT_VALIDATOR_SET.as_slice() {
-        return Some(GravityContractError::MalformedCurrentValidatorSet(MalformedCurrentValidatorSet{}))
+        return Some(GravityContractError::MalformedCurrentValidatorSet(
+            MalformedCurrentValidatorSet {},
+        ));
     }
 
     if selector == MALFORMED_BATCH.as_slice() {
-        return Some(GravityContractError::MalformedBatch(MalformedBatch{}))
+        return Some(GravityContractError::MalformedBatch(MalformedBatch {}));
     }
 
     if selector == INSUFFICIENT_POWER.as_slice() {
         if data.len() != 68 {
-            return None
+            return None;
         }
 
-        return Some(GravityContractError::InsufficientPower(InsufficientPower{
+        return Some(GravityContractError::InsufficientPower(InsufficientPower {
             cumulative_power: data[4..36].into(),
             power_threshold: data[36..].into(),
-        }))
+        }));
     }
 
     if selector == BATCH_TIMED_OUT.as_slice() {
-        return Some(GravityContractError::BatchTimedOut(BatchTimedOut{}))
+        return Some(GravityContractError::BatchTimedOut(BatchTimedOut {}));
     }
 
     if selector == LOGIC_CALL_TIMED_OUT.as_slice() {
-        return Some(GravityContractError::LogicCallTimedOut(LogicCallTimedOut{}))
+        return Some(GravityContractError::LogicCallTimedOut(
+            LogicCallTimedOut {},
+        ));
     }
 
     info!("Did not find gravity error");
@@ -156,8 +178,10 @@ pub struct InvalidValsetNonce {
 
 impl InvalidValsetNonce {
     pub fn message(&self) -> String {
-        format!("Invalid valset nonce, new nonce {}, current nonce {}",
-            self.new_nonce, self.current_nonce)
+        format!(
+            "Invalid valset nonce, new nonce {}, current nonce {}",
+            self.new_nonce, self.current_nonce
+        )
     }
 }
 
@@ -168,8 +192,10 @@ pub struct InvalidBatchNonce {
 
 impl InvalidBatchNonce {
     pub fn message(&self) -> String {
-        format!("Invalid batch nonce, new nonce {}, current nonce {}",
-            self.new_nonce, self.current_nonce)
+        format!(
+            "Invalid batch nonce, new nonce {}, current nonce {}",
+            self.new_nonce, self.current_nonce
+        )
     }
 }
 
@@ -180,8 +206,10 @@ pub struct InvalidLogicCallNonce {
 
 impl InvalidLogicCallNonce {
     pub fn message(&self) -> String {
-        format!("Invalid logic call nonce, new nonce {}, current nonce {}",
-            self.new_nonce, self.current_nonce)
+        format!(
+            "Invalid logic call nonce, new nonce {}, current nonce {}",
+            self.new_nonce, self.current_nonce
+        )
     }
 }
 
@@ -248,8 +276,10 @@ pub struct InsufficientPower {
 
 impl InsufficientPower {
     pub fn message(&self) -> String {
-        format!("Insufficient power, cumulative power {}, power threshold {}",
-            self.cumulative_power, self.power_threshold)
+        format!(
+            "Insufficient power, cumulative power {}, power threshold {}",
+            self.cumulative_power, self.power_threshold
+        )
     }
 }
 
@@ -267,4 +297,4 @@ impl LogicCallTimedOut {
     pub fn message(&self) -> String {
         "Logic call timed out".to_string()
     }
-}
+}