[r2r] Tendermint multiple rpcs optimization

mm2src/coins/eth.rs

@@ -82,7 +82,7 @@ pub use rlp;
 
 #[cfg(test)] mod eth_tests;
 #[cfg(target_arch = "wasm32")] mod eth_wasm_tests;
-mod web3_transport;
+pub mod web3_transport;
 
 #[path = "eth/v2_activation.rs"] pub mod v2_activation;
 use v2_activation::build_address_and_priv_key_policy;

mm2src/coins/lp_coins.rs

@@ -33,6 +33,8 @@
 #[macro_use] extern crate serde_json;
 #[macro_use] extern crate ser_error_derive;
 
+use crate::eth::web3_transport::http_transport::HttpTransportNode;
+use crate::tendermint::rpc::HttpClient;
 use async_trait::async_trait;
 use base58::FromBase58Error;
 use common::executor::{abortable_queue::{AbortableQueue, WeakSpawner},
@@ -71,6 +73,11 @@ use std::sync::Arc;
 use std::time::Duration;
 use utxo_signer::with_key_pair::UtxoSignWithKeyPairError;
 
+#[cfg(not(target_arch = "wasm32"))]
+mod z_coin_grpc {
+    tonic::include_proto!("cash.z.wallet.sdk.rpc");
+}
+
 cfg_native! {
     use crate::lightning::LightningCoin;
     use crate::lightning::ln_conf::PlatformCoinConfirmationTargets;
@@ -81,6 +88,8 @@ cfg_native! {
     use std::io;
     use zcash_primitives::transaction::Transaction as ZTransaction;
     use z_coin::ZcoinProtocolInfo;
+    use tonic::transport::Channel as TonicChannel;
+    use z_coin_grpc::compact_tx_streamer_client::CompactTxStreamerClient;
 }
 
 cfg_wasm32! {
@@ -3511,3 +3520,27 @@ where
         b.block_height.cmp(&a.block_height)
     }
 }
+
+#[derive(Display, Debug)]
+pub enum RpcCommonError {
+    FindClientError(String),
+    WrongRpcClient,
+}
+
+#[derive(Debug)]
+pub enum RpcClientEnum {
+    #[cfg(not(target_arch = "wasm32"))]
+    ZcoinRpcClient(CompactTxStreamerClient<TonicChannel>),
+    TendermintHttpClient(HttpClient),
+    Web3HttpTransportNode(HttpTransportNode),
+}
+
+/// Use trait in the case, when we have to send requests to rpc client.
+#[async_trait]
+pub trait RpcCommonOps {
+    /// get alive client from custom `RpcClient` structure, that contains one rpc client and vector of rpc urls.
+    async fn get_rpc_client(&self) -> Result<RpcClientEnum, RpcCommonError>;
+
+    /// if rpc client isn't available anymore, iterate over rpc urls, to get the first available client.
+    async fn iterate_over_urls(&self) -> Result<RpcClientEnum, RpcCommonError>;
+}

mm2src/coins/tendermint/mod.rs

@@ -3,7 +3,7 @@
 // https://docs.cosmos.network/
 
 mod iris;
-mod rpc;
+pub mod rpc;
 mod tendermint_coin;
 mod tendermint_token;
 pub mod tendermint_tx_history_v2;

mm2src/coins/tendermint/tendermint_coin.rs

@@ -9,22 +9,21 @@ use crate::{big_decimal_from_sat_unsigned, BalanceError, BalanceFut, BigDecimal,
             CoinBalance, CoinFutSpawner, FeeApproxStage, FoundSwapTxSpend, HistorySyncState, MarketCoinOps, MmCoin,
             NegotiateSwapContractAddrErr, PaymentInstructions, PaymentInstructionsErr, PrivKeyBuildPolicy,
             PrivKeyPolicyNotAllowed, RawTransactionError, RawTransactionFut, RawTransactionRequest, RawTransactionRes,
-            SearchForSwapTxSpendInput, SendMakerPaymentArgs, SendMakerRefundsPaymentArgs,
-            SendMakerSpendsTakerPaymentArgs, SendTakerPaymentArgs, SendTakerRefundsPaymentArgs,
-            SendTakerSpendsMakerPaymentArgs, SignatureError, SignatureResult, SwapOps, TradeFee, TradePreimageError,
-            TradePreimageFut, TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionEnum,
-            TransactionErr, TransactionFut, TransactionType, TxFeeDetails, TxMarshalingErr,
-            UnexpectedDerivationMethod, ValidateAddressResult, ValidateFeeArgs, ValidateInstructionsErr,
-            ValidateOtherPubKeyErr, ValidatePaymentFut, ValidatePaymentInput, VerificationError, VerificationResult,
-            WatcherOps, WatcherSearchForSwapTxSpendInput, WatcherValidatePaymentInput, WatcherValidateTakerFeeInput,
-            WithdrawError, WithdrawFut, WithdrawRequest};
-use async_std::prelude::FutureExt as AsyncStdFutureExt;
+            RpcClientEnum, RpcCommonError, RpcCommonOps, SearchForSwapTxSpendInput, SendMakerPaymentArgs,
+            SendMakerRefundsPaymentArgs, SendMakerSpendsTakerPaymentArgs, SendTakerPaymentArgs,
+            SendTakerRefundsPaymentArgs, SendTakerSpendsMakerPaymentArgs, SignatureError, SignatureResult, SwapOps,
+            TradeFee, TradePreimageError, TradePreimageFut, TradePreimageResult, TradePreimageValue,
+            TransactionDetails, TransactionEnum, TransactionErr, TransactionFut, TransactionType, TxFeeDetails,
+            TxMarshalingErr, UnexpectedDerivationMethod, ValidateAddressResult, ValidateFeeArgs,
+            ValidateInstructionsErr, ValidateOtherPubKeyErr, ValidatePaymentFut, ValidatePaymentInput,
+            VerificationError, VerificationResult, WatcherOps, WatcherSearchForSwapTxSpendInput,
+            WatcherValidatePaymentInput, WatcherValidateTakerFeeInput, WithdrawError, WithdrawFut, WithdrawRequest};
 use async_trait::async_trait;
 use bitcrypto::{dhash160, sha256};
 use common::executor::{abortable_queue::AbortableQueue, AbortableSystem};
 use common::executor::{AbortedError, Timer};
 use common::log::warn;
-use common::{get_utc_timestamp, log, now_ms, Future01CompatExt, DEX_FEE_ADDR_PUBKEY};
+use common::{get_utc_timestamp, now_ms, Future01CompatExt, DEX_FEE_ADDR_PUBKEY};
 use cosmrs::bank::MsgSend;
 use cosmrs::crypto::secp256k1::SigningKey;
 use cosmrs::proto::cosmos::auth::v1beta1::{BaseAccount, QueryAccountRequest, QueryAccountResponse};
@@ -60,7 +59,6 @@ use std::convert::TryFrom;
 use std::ops::Deref;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
 use uuid::Uuid;
 
 // ABCI Request Paths
@@ -164,9 +162,52 @@ impl TendermintConf {
     }
 }
 
+struct TendermintRpcClient {
+    rpc_urls: Vec<String>,
+    rpc_client: AsyncMutex<HttpClient>,
+}
+
+#[async_trait]
+impl RpcCommonOps for TendermintRpcClient {
+    async fn get_rpc_client(&self) -> Result<RpcClientEnum, RpcCommonError> {
+        let mut rpc_client = self.rpc_client.lock().await;
+        match rpc_client.perform(HealthRequest).await {
+            Ok(_) => Ok(RpcClientEnum::TendermintHttpClient(rpc_client.clone())),
+            // try HealthRequest one more time
+            Err(_) => match rpc_client.perform(HealthRequest).await {
+                Ok(_) => Ok(RpcClientEnum::TendermintHttpClient(rpc_client.clone())),
+                Err(_) => {
+                    let new_client = self.iterate_over_urls().await?;
+                    match new_client {
+                        RpcClientEnum::TendermintHttpClient(client) => {
+                            *rpc_client = client.clone();
+                            Ok(RpcClientEnum::TendermintHttpClient(client))
+                        },
+                        _ => Err(RpcCommonError::WrongRpcClient),
+                    }
+                },
+            },
+        }
+    }
+
+    async fn iterate_over_urls(&self) -> Result<RpcClientEnum, RpcCommonError> {
+        let urls = &self.rpc_urls;
+        for url in urls {
+            let client = HttpClient::new(url.as_str());
+            if let Ok(client) = client {
+                if client.perform(HealthRequest).await.is_ok() {
+                    return Ok(RpcClientEnum::TendermintHttpClient(client.clone()));
+                }
+            }
+        }
+        Err(RpcCommonError::FindClientError(
+            "All the current rpc nodes are unavailable.".to_string(),
+        ))
+    }
+}
+
 pub struct TendermintCoinImpl {
     ticker: String,
-    rpc_clients: Vec<HttpClient>,
     /// As seconds
     avg_blocktime: u8,
     /// My address
@@ -183,6 +224,7 @@ pub struct TendermintCoinImpl {
     /// or on [`MmArc::stop`].
     pub(super) abortable_system: AbortableQueue,
     pub(crate) history_sync_state: Mutex<HistorySyncState>,
+    client_impl: TendermintRpcClient,
 }
 
 #[derive(Clone)]
@@ -206,6 +248,7 @@ pub enum TendermintInitErrorKind {
     InvalidPrivKey(String),
     CouldNotGenerateAccountId(String),
     EmptyRpcUrls,
+    #[display(fmt = "Fail to init HttpClient during rpc urls iteration {}", _0)]
     RpcClientInitError(String),
     InvalidChainId(String),
     InvalidDenom(String),
@@ -226,12 +269,18 @@ pub enum TendermintCoinRpcError {
     Prost(DecodeError),
     InvalidResponse(String),
     PerformError(String),
+    WrongRpcClient,
+    RpcClientError(RpcCommonError),
 }
 
 impl From<DecodeError> for TendermintCoinRpcError {
     fn from(err: DecodeError) -> Self { TendermintCoinRpcError::Prost(err) }
 }
 
+impl From<RpcCommonError> for TendermintCoinRpcError {
+    fn from(err: RpcCommonError) -> Self { TendermintCoinRpcError::RpcClientError(err) }
+}
+
 impl From<TendermintCoinRpcError> for WithdrawError {
     fn from(err: TendermintCoinRpcError) -> Self { WithdrawError::Transport(err.to_string()) }
 }
@@ -242,6 +291,8 @@ impl From<TendermintCoinRpcError> for BalanceError {
             TendermintCoinRpcError::InvalidResponse(e) => BalanceError::InvalidResponse(e),
             TendermintCoinRpcError::Prost(e) => BalanceError::InvalidResponse(e.to_string()),
             TendermintCoinRpcError::PerformError(e) => BalanceError::Transport(e),
+            TendermintCoinRpcError::WrongRpcClient => BalanceError::Internal("Wrong rpc client type".to_string()),
+            TendermintCoinRpcError::RpcClientError(e) => BalanceError::Transport(format!("{}", e)),
         }
     }
 }
@@ -252,6 +303,10 @@ impl From<TendermintCoinRpcError> for ValidatePaymentError {
             TendermintCoinRpcError::InvalidResponse(e) => ValidatePaymentError::InvalidRpcResponse(e),
             TendermintCoinRpcError::Prost(e) => ValidatePaymentError::InvalidRpcResponse(e.to_string()),
             TendermintCoinRpcError::PerformError(e) => ValidatePaymentError::Transport(e),
+            TendermintCoinRpcError::WrongRpcClient => {
+                ValidatePaymentError::InternalError("Wrong rpc client type".to_string())
+            },
+            TendermintCoinRpcError::RpcClientError(e) => ValidatePaymentError::Transport(format!("{}", e)),
         }
     }
 }
@@ -381,25 +436,12 @@ impl TendermintCommons for TendermintCoin {
         Ok(result)
     }
 
-    // TODO
-    // Save one working client to the coin context, only try others once it doesn't
-    // work anymore.
-    // Also, try couple times more on health check errors.
     async fn rpc_client(&self) -> MmResult<HttpClient, TendermintCoinRpcError> {
-        for rpc_client in self.rpc_clients.iter() {
-            match rpc_client.perform(HealthRequest).timeout(Duration::from_secs(3)).await {
-                Ok(Ok(_)) => return Ok(rpc_client.clone()),
-                Ok(Err(e)) => log::warn!(
-                    "Recieved error from Tendermint rpc node during health check. Error: {:?}",
-                    e
-                ),
-                Err(_) => log::warn!("Tendermint rpc node: {:?} got timeout during health check", rpc_client),
-            };
+        let client_enum = self.client_impl.get_rpc_client().await?;
+        match client_enum {
+            RpcClientEnum::TendermintHttpClient(client) => Ok(client),
+            _ => MmError::err(TendermintCoinRpcError::WrongRpcClient),
         }
-
-        MmError::err(TendermintCoinRpcError::PerformError(
-            "All the current rpc nodes are unavailable.".to_string(),
-        ))
     }
 }
 
@@ -430,17 +472,15 @@ impl TendermintCoin {
                 }
             })?;
 
-        let rpc_clients: Result<Vec<HttpClient>, _> = rpc_urls
-            .iter()
-            .map(|url| {
-                HttpClient::new(url.as_str()).map_to_mm(|e| TendermintInitError {
-                    ticker: ticker.clone(),
-                    kind: TendermintInitErrorKind::RpcClientInitError(e.to_string()),
-                })
-            })
-            .collect();
+        let (rpc_client, rpc_urls) = find_client(rpc_urls).map_to_mm(|e| TendermintInitError {
+            ticker: ticker.clone(),
+            kind: TendermintInitErrorKind::RpcClientInitError(e),
+        })?;
 
-        let rpc_clients = rpc_clients?;
+        let client_impl = TendermintRpcClient {
+            rpc_urls,
+            rpc_client: AsyncMutex::new(rpc_client),
+        };
 
         let chain_id = ChainId::try_from(protocol_info.chain_id).map_to_mm(|e| TendermintInitError {
             ticker: ticker.clone(),
@@ -470,7 +510,6 @@ impl TendermintCoin {
 
         Ok(TendermintCoin(Arc::new(TendermintCoinImpl {
             ticker,
-            rpc_clients,
             account_id,
             account_prefix: protocol_info.account_prefix,
             priv_key: priv_key.to_vec(),
@@ -483,6 +522,7 @@ impl TendermintCoin {
             tokens_info: PaMutex::new(HashMap::new()),
             abortable_system,
             history_sync_state: Mutex::new(history_sync_state),
+            client_impl,
         })))
     }
 
@@ -1357,6 +1397,32 @@ impl TendermintCoin {
     }
 }
 
+fn find_client(rpc_urls: Vec<String>) -> Result<(HttpClient, Vec<String>), String> {
+    let mut res_urls = rpc_urls.clone();
+    let mut errors = Vec::new();
+    let mut clients = Vec::new();
+    for (i, url) in rpc_urls.iter().enumerate() {
+        match HttpClient::new(url.as_str()) {
+            Ok(client) => clients.push(client),
+            Err(e) => {
+                errors.push(e);
+                res_urls.remove(i);
+            },
+        }
+    }
+    if let Some(client) = clients.pop() {
+        if !rpc_urls.is_empty() {
+            Ok((client, rpc_urls))
+        } else {
+            let errors: String = errors.iter().map(|e| format!("{:?}", e)).collect();
+            Err(errors)
+        }
+    } else {
+        let errors: String = errors.iter().map(|e| format!("{:?}", e)).collect();
+        Err(errors)
+    }
+}
+
 #[async_trait]
 #[allow(unused_variables)]
 impl MmCoin for TendermintCoin {