feat(utxo): prioritize electrum connections

README.md

@@ -186,4 +186,4 @@ The current state can be considered as a alpha version.
 
 ## Help and troubleshooting
 
-If you have any question/want to report a bug/suggest an improvement feel free to [open an issue](https://github.com/artemii235/SuperNET/issues/new) or join the  [Komodo Platform Discord](https://discord.gg/PGxVm2y) `dev-marketmaker` channel.  
+If you have any question/want to report a bug/suggest an improvement feel free to [open an issue](https://github.com/artemii235/SuperNET/issues/new) or join the  [Komodo Platform Discord](https://discord.gg/PGxVm2y) `dev-marketmaker` channel.  buildbusi

mm2src/adex_cli/src/adex_proc/adex_proc_impl.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, bail, Result};
-use log::{error, info, warn};
+use log::{debug, error, info, warn};
 use mm2_rpc::data::legacy::{BalanceResponse, CoinInitResponse, GetEnabledResponse, Mm2RpcResult, MmVersionResponse,
                             OrderbookRequest, OrderbookResponse, SellBuyRequest, SellBuyResponse, Status};
 use serde_json::{json, Value as Json};
@@ -38,7 +38,7 @@ impl<T: Transport, P: ResponseHandler, C: AdexConfig + 'static> AdexProc<'_, '_,
 
         let activation_scheme = get_activation_scheme()?;
         let activation_method = activation_scheme.get_activation_method(asset)?;
-
+        debug!("Got activation scheme for the coin: {}, {:?}", asset, activation_method);
         let enable = Command::builder()
             .flatten_data(activation_method)
             .userpass(self.get_rpc_password()?)

mm2src/adex_cli/src/rpc_data.rs

@@ -85,4 +85,19 @@ pub(super) struct Server {
     protocol: ElectrumProtocol,
     #[serde(default)]
     disable_cert_verification: bool,
+    #[serde(default)]
+    pub priority: Priority,
+    pub timeout_sec: Option<u64>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Priority {
+    Primary,
+    Secondary,
+    Tertiary,
+}
+
+impl Default for Priority {
+    fn default() -> Self { Priority::Secondary }
 }

mm2src/adex_cli/src/tests/mod.rs

@@ -1,3 +1,4 @@
+use log::trace;
 use std::io::Write;
 use std::time::Duration;
 use tokio::io::AsyncWriteExt;

mm2src/coins/lp_coins.rs

@@ -81,6 +81,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering as AtomicOrdering;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::Notify;
 use utxo_signer::with_key_pair::UtxoSignWithKeyPairError;
 use zcash_primitives::transaction::Transaction as ZTransaction;
 
@@ -2843,35 +2844,43 @@ pub enum CoinProtocol {
 
 pub type RpcTransportEventHandlerShared = Arc<dyn RpcTransportEventHandler + Send + Sync + 'static>;
 
-/// Common methods to measure the outgoing requests and incoming responses statistics.
+#[async_trait]
 pub trait RpcTransportEventHandler {
     fn debug_info(&self) -> String;
 
     fn on_outgoing_request(&self, data: &[u8]);
 
     fn on_incoming_response(&self, data: &[u8]);
 
-    fn on_connected(&self, address: String) -> Result<(), String>;
+    fn on_connected(&self, address: String, conn_spawner: WeakSpawner, verified: Arc<Notify>) -> Result<(), String>;
+
+    fn on_connected_async(&self, _address: String) -> Result<(), String> { Ok(()) }
 
-    fn on_disconnected(&self, address: String) -> Result<(), String>;
+    fn on_disconnected(&self, address: String, conn_spawner: WeakSpawner) -> Result<(), String>;
 }
 
 impl fmt::Debug for dyn RpcTransportEventHandler + Send + Sync {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}", self.debug_info()) }
 }
 
+#[async_trait]
 impl RpcTransportEventHandler for RpcTransportEventHandlerShared {
     fn debug_info(&self) -> String { self.deref().debug_info() }
 
     fn on_outgoing_request(&self, data: &[u8]) { self.as_ref().on_outgoing_request(data) }
 
     fn on_incoming_response(&self, data: &[u8]) { self.as_ref().on_incoming_response(data) }
 
-    fn on_connected(&self, address: String) -> Result<(), String> { self.as_ref().on_connected(address) }
+    fn on_connected(&self, address: String, conn_spawner: WeakSpawner, verified: Arc<Notify>) -> Result<(), String> {
+        self.as_ref().on_connected(address, conn_spawner, verified)
+    }
 
-    fn on_disconnected(&self, address: String) -> Result<(), String> { self.as_ref().on_disconnected(address) }
+    fn on_disconnected(&self, address: String, conn_spawner: WeakSpawner) -> Result<(), String> {
+        self.as_ref().on_disconnected(address, conn_spawner)
+    }
 }
 
+#[async_trait]
 impl<T: RpcTransportEventHandler> RpcTransportEventHandler for Vec<T> {
     fn debug_info(&self) -> String {
         let selfi: Vec<String> = self.iter().map(|x| x.debug_info()).collect();
@@ -2890,16 +2899,16 @@ impl<T: RpcTransportEventHandler> RpcTransportEventHandler for Vec<T> {
         }
     }
 
-    fn on_connected(&self, address: String) -> Result<(), String> {
+    fn on_connected(&self, address: String, conn_spawner: WeakSpawner, verified: Arc<Notify>) -> Result<(), String> {
         for handler in self {
-            try_s!(handler.on_connected(address.clone()))
+            try_s!(handler.on_connected(address.clone(), conn_spawner.clone(), verified.clone()))
         }
         Ok(())
     }
 
-    fn on_disconnected(&self, address: String) -> Result<(), String> {
+    fn on_disconnected(&self, address: String, conn_spawner: WeakSpawner) -> Result<(), String> {
         for handler in self {
-            try_s!(handler.on_disconnected(address.clone()))
+            try_s!(handler.on_disconnected(address.clone(), conn_spawner.clone()))
         }
         Ok(())
     }
@@ -2943,6 +2952,7 @@ impl CoinTransportMetrics {
     fn into_shared(self) -> RpcTransportEventHandlerShared { Arc::new(self) }
 }
 
+#[async_trait]
 impl RpcTransportEventHandler for CoinTransportMetrics {
     fn debug_info(&self) -> String { "CoinTransportMetrics".into() }
 
@@ -2960,13 +2970,13 @@ impl RpcTransportEventHandler for CoinTransportMetrics {
             "coin" => self.ticker.to_owned(), "client" => self.client.to_owned());
     }
 
-    fn on_connected(&self, _address: String) -> Result<(), String> {
+    fn on_connected(&self, _address: String, _conn_spawner: WeakSpawner, _verified: Arc<Notify>) -> Result<(), String> {
         // Handle a new connected endpoint if necessary.
         // Now just return the Ok
         Ok(())
     }
 
-    fn on_disconnected(&self, _address: String) -> Result<(), String> {
+    fn on_disconnected(&self, _address: String, _conn_spawner: WeakSpawner) -> Result<(), String> {
         // Handle disconnected endpoint if necessary.
         // Now just return the Ok
         Ok(())

mm2src/coins/utxo.rs

@@ -44,11 +44,11 @@ use bitcoin::network::constants::Network as BitcoinNetwork;
 pub use bitcrypto::{dhash160, sha256, ChecksumType};
 pub use chain::Transaction as UtxoTx;
 use chain::{OutPoint, TransactionOutput, TxHashAlgo};
-use common::executor::abortable_queue::AbortableQueue;
+use common::executor::abortable_queue::{AbortableQueue, WeakSpawner};
 #[cfg(not(target_arch = "wasm32"))]
 use common::first_char_to_upper;
 use common::jsonrpc_client::JsonRpcError;
-use common::log::LogOnError;
+use common::log::{debug, LogOnError};
 use common::{now_sec, now_sec_u32};
 use crypto::{Bip32DerPathOps, Bip32Error, Bip44Chain, ChildNumber, DerivationPath, Secp256k1ExtendedPublicKey,
              StandardHDPathError, StandardHDPathToAccount, StandardHDPathToCoin};
@@ -89,21 +89,22 @@ use std::path::{Path, PathBuf};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU64};
 use std::sync::{Arc, Mutex, Weak};
+use tokio::sync::Notify;
 use utxo_builder::UtxoConfBuilder;
 use utxo_common::{big_decimal_from_sat, UtxoTxBuilder};
 use utxo_signer::with_key_pair::sign_tx;
 use utxo_signer::{TxProvider, TxProviderError, UtxoSignTxError, UtxoSignTxResult};
 
-use self::rpc_clients::{electrum_script_hash, ElectrumClient, ElectrumRpcRequest, EstimateFeeMethod, EstimateFeeMode,
-                        NativeClient, UnspentInfo, UnspentMap, UtxoRpcClientEnum, UtxoRpcError, UtxoRpcFut,
-                        UtxoRpcResult};
+use self::rpc_clients::{electrum_script_hash, ElectrumClient, ElectrumConnSettings, EstimateFeeMethod,
+                        EstimateFeeMode, NativeClient, UnspentInfo, UnspentMap, UtxoRpcClientEnum, UtxoRpcError,
+                        UtxoRpcFut, UtxoRpcResult};
 use super::{big_decimal_from_sat_unsigned, BalanceError, BalanceFut, BalanceResult, CoinBalance, CoinFutSpawner,
             CoinsContext, DerivationMethod, FeeApproxStage, FoundSwapTxSpend, HistorySyncState, KmdRewardsDetails,
             MarketCoinOps, MmCoin, NumConversError, NumConversResult, PrivKeyActivationPolicy, PrivKeyPolicy,
             PrivKeyPolicyNotAllowed, RawTransactionFut, RawTransactionRequest, RawTransactionResult,
-            RpcTransportEventHandler, RpcTransportEventHandlerShared, TradeFee, TradePreimageError, TradePreimageFut,
-            TradePreimageResult, Transaction, TransactionDetails, TransactionEnum, TransactionErr,
-            UnexpectedDerivationMethod, VerificationError, WithdrawError, WithdrawRequest};
+            RpcTransportEventHandler, TradeFee, TradePreimageError, TradePreimageFut, TradePreimageResult,
+            Transaction, TransactionDetails, TransactionEnum, TransactionErr, UnexpectedDerivationMethod,
+            VerificationError, WithdrawError, WithdrawRequest};
 use crate::coin_balance::{EnableCoinScanPolicy, EnabledCoinBalanceParams, HDAddressBalanceScanner};
 use crate::hd_wallet::{HDAccountOps, HDAccountsMutex, HDAddress, HDAddressId, HDWalletCoinOps, HDWalletOps,
                        InvalidBip44ChainError};
@@ -1320,25 +1321,24 @@ struct ElectrumProtoVerifier {
     on_event_tx: UnboundedSender<ElectrumProtoVerifierEvent>,
 }
 
-impl ElectrumProtoVerifier {
-    fn into_shared(self) -> RpcTransportEventHandlerShared { Arc::new(self) }
-}
-
+#[async_trait]
 impl RpcTransportEventHandler for ElectrumProtoVerifier {
     fn debug_info(&self) -> String { "ElectrumProtoVerifier".into() }
 
     fn on_outgoing_request(&self, _data: &[u8]) {}
 
     fn on_incoming_response(&self, _data: &[u8]) {}
 
-    fn on_connected(&self, address: String) -> Result<(), String> {
+    fn on_connected(&self, address: String, _conn_spawner: WeakSpawner, _verified: Arc<Notify>) -> Result<(), String> {
+        debug!("Connected to the electrum server: {}", address);
         try_s!(self
             .on_event_tx
             .unbounded_send(ElectrumProtoVerifierEvent::Connected(address)));
         Ok(())
     }
 
-    fn on_disconnected(&self, address: String) -> Result<(), String> {
+    fn on_disconnected(&self, address: String, _conn_spawner: WeakSpawner) -> Result<(), String> {
+        debug!("Disconnected from the electrum server: {}", address);
         try_s!(self
             .on_event_tx
             .unbounded_send(ElectrumProtoVerifierEvent::Disconnected(address)));
@@ -1437,7 +1437,7 @@ impl UtxoActivationParams {
 #[serde(tag = "rpc", content = "rpc_data")]
 pub enum UtxoRpcMode {
     Native,
-    Electrum { servers: Vec<ElectrumRpcRequest> },
+    Electrum { servers: Vec<ElectrumConnSettings> },
 }
 
 impl UtxoRpcMode {