[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(crate) 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

@@ -3511,3 +3511,13 @@ where
         b.block_height.cmp(&a.block_height)
     }
 }
+
+/// Use trait in the case, when we have to send requests to rpc client.
+#[async_trait]
+pub trait RpcCommonOps {
+    type RpcClient;
+    type Error;
+
+    /// Returns an alive RPC client or returns an error if no RPC endpoint is currently available.
+    async fn get_live_client(&self) -> Result<Self::RpcClient, Self::Error>;
+}

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,7 +9,7 @@ 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,
+            RpcCommonOps, SearchForSwapTxSpendInput, SendMakerPaymentArgs, SendMakerRefundsPaymentArgs,
             SendMakerSpendsTakerPaymentArgs, SendTakerPaymentArgs, SendTakerRefundsPaymentArgs,
             SendTakerSpendsMakerPaymentArgs, SignatureError, SignatureResult, SwapOps, TradeFee, TradePreimageError,
             TradePreimageFut, TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionEnum,
@@ -18,13 +18,12 @@ use crate::{big_decimal_from_sat_unsigned, BalanceError, BalanceFut, BigDecimal,
             ValidateOtherPubKeyErr, ValidatePaymentFut, ValidatePaymentInput, VerificationError, VerificationResult,
             WatcherOps, WatcherSearchForSwapTxSpendInput, WatcherValidatePaymentInput, WatcherValidateTakerFeeInput,
             WithdrawError, WithdrawFut, WithdrawRequest};
-use async_std::prelude::FutureExt as AsyncStdFutureExt;
 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,56 @@ impl TendermintConf {
     }
 }
 
+struct TendermintRpcClient(AsyncMutex<TendermintRpcClientImpl>);
+
+#[allow(dead_code)]
+struct TendermintRpcClientImpl {
+    rpc_urls: Vec<String>,
+    rpc_client: HttpClient,
+    current: usize,
+}
+
+#[async_trait]
+impl RpcCommonOps for TendermintCoin {
+    type RpcClient = HttpClient;
+    type Error = TendermintCoinRpcError;
+
+    async fn get_live_client(&self) -> Result<Self::RpcClient, Self::Error> {
+        let mut client_impl = self.client.0.lock().await;
+        let current_client = client_impl.rpc_client.clone();
+        match current_client.perform(HealthRequest).await {
+            Ok(_) => return Ok(current_client),
+            // try HealthRequest one more time
+            Err(_) => {
+                if current_client.perform(HealthRequest).await.is_ok() {
+                    return Ok(current_client);
+                }
+            },
+        }
+        let mut rpc_urls = client_impl.rpc_urls.clone();
+        let current_url = rpc_urls.remove(client_impl.current);
+        // push current url to the end
+        rpc_urls.push(current_url);
+        // try to find first live client
+        for (i, url) in rpc_urls.iter().enumerate() {
+            let client = HttpClient::new(url.as_str());
+            if let Ok(new_client) = client {
+                if new_client.perform(HealthRequest).await.is_ok() {
+                    // replace an unavailable client with a new one
+                    client_impl.rpc_client = new_client.clone();
+                    client_impl.current = i;
+                    return Ok(new_client);
+                }
+            }
+        }
+        return Err(TendermintCoinRpcError::RpcClientError(
+            "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 +228,7 @@ pub struct TendermintCoinImpl {
     /// or on [`MmArc::stop`].
     pub(super) abortable_system: AbortableQueue,
     pub(crate) history_sync_state: Mutex<HistorySyncState>,
+    client: TendermintRpcClient,
 }
 
 #[derive(Clone)]
@@ -226,6 +272,7 @@ pub enum TendermintCoinRpcError {
     Prost(DecodeError),
     InvalidResponse(String),
     PerformError(String),
+    RpcClientError(String),
 }
 
 impl From<DecodeError> for TendermintCoinRpcError {
@@ -242,6 +289,7 @@ 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::RpcClientError(e) => BalanceError::Transport(e),
         }
     }
 }
@@ -252,6 +300,7 @@ 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::RpcClientError(e) => ValidatePaymentError::Transport(e),
         }
     }
 }
@@ -381,25 +430,9 @@ 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.
+    #[inline(always)]
     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),
-            };
-        }
-
-        MmError::err(TendermintCoinRpcError::PerformError(
-            "All the current rpc nodes are unavailable.".to_string(),
-        ))
+        self.get_live_client().await.map_to_mm(|e| e)
     }
 }
 
@@ -430,17 +463,16 @@ 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, current) = find_client(rpc_urls.as_ref()).map_to_mm(|e| TendermintInitError {
+            ticker: ticker.clone(),
+            kind: TendermintInitErrorKind::RpcClientInitError(e),
+        })?;
 
-        let rpc_clients = rpc_clients?;
+        let client_impl = TendermintRpcClientImpl {
+            rpc_urls,
+            rpc_client,
+            current,
+        };
 
         let chain_id = ChainId::try_from(protocol_info.chain_id).map_to_mm(|e| TendermintInitError {
             ticker: ticker.clone(),
@@ -470,7 +502,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 +514,7 @@ impl TendermintCoin {
             tokens_info: PaMutex::new(HashMap::new()),
             abortable_system,
             history_sync_state: Mutex::new(history_sync_state),
+            client: TendermintRpcClient(AsyncMutex::new(client_impl)),
         })))
     }
 
@@ -1357,6 +1389,29 @@ impl TendermintCoin {
     }
 }
 
+fn find_client(rpc_urls: &[String]) -> Result<(HttpClient, usize), String> {
+    let mut clients = Vec::new();
+    let mut errors = Vec::new();
+    // check that all urls are valid
+    // keep all invalid urls in one vector to show all of them in error
+    for url in rpc_urls.iter() {
+        match HttpClient::new(url.as_str()) {
+            Ok(client) => clients.push(client),
+            Err(e) => errors.push(format!("Url {} is invalid, got error {}", url, e)),
+        }
+    }
+    if !errors.is_empty() {
+        let errors: String = errors.iter().map(|e| format!("{:?}", e)).collect();
+        return Err(errors);
+    }
+    let current = clients.len() - 1;
+    if let Some(client) = clients.pop() {
+        Ok((client, current))
+    } else {
+        Err("None client was found".to_string())
+    }
+}
+
 #[async_trait]
 #[allow(unused_variables)]
 impl MmCoin for TendermintCoin {