[r2r] Multi lightwalletd servers integration

mm2src/coins/z_coin.rs

@@ -796,7 +796,6 @@ impl<'a> UtxoCoinWithIguanaPrivKeyBuilder for ZCoinBuilder<'a> {
             ZcoinRpcMode::Light {
                 light_wallet_d_servers, ..
             } => {
-                // TODO multi lightwalletd servers support will be added on the next iteration
                 init_light_client(
                     light_wallet_d_servers.clone(),
                     blocks_db,

mm2src/coins/z_coin/z_coin_errors.rs

@@ -10,21 +10,42 @@ use derive_more::Display;
 use http::uri::InvalidUri;
 use mm2_number::BigDecimal;
 use rpc::v1::types::{Bytes as BytesJson, H256 as H256Json};
+use std::fmt;
 use zcash_client_sqlite::error::SqliteClientError;
 use zcash_client_sqlite::error::SqliteClientError as ZcashClientError;
 use zcash_primitives::transaction::builder::Error as ZTxBuilderError;
 
-#[derive(Debug, Display)]
+#[derive(Debug)]
 #[non_exhaustive]
 pub enum UpdateBlocksCacheErr {
     GrpcError(tonic::Status),
+    GrpcVecError(Vec<tonic::Status>),
     BlocksDbError(SqliteError),
     ZcashSqliteError(ZcashClientError),
     UtxoRpcError(UtxoRpcError),
     InternalError(String),
     JsonRpcError(JsonRpcError),
 }
 
+impl fmt::Display for UpdateBlocksCacheErr {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match *self {
+            UpdateBlocksCacheErr::GrpcError(ref err) => write!(f, "{}", err),
+            UpdateBlocksCacheErr::GrpcVecError(ref err) => {
+                for e in err {
+                    write!(f, "\t{}", e)?;
+                }
+                Ok(())
+            },
+            UpdateBlocksCacheErr::BlocksDbError(ref err) => write!(f, "{}", err),
+            UpdateBlocksCacheErr::ZcashSqliteError(ref err) => write!(f, "{}", err),
+            UpdateBlocksCacheErr::UtxoRpcError(ref err) => write!(f, "{}", err),
+            UpdateBlocksCacheErr::InternalError(ref err) => write!(f, "{}", err),
+            UpdateBlocksCacheErr::JsonRpcError(ref err) => write!(f, "{}", err),
+        }
+    }
+}
+
 impl From<tonic::Status> for UpdateBlocksCacheErr {
     fn from(err: tonic::Status) -> Self { UpdateBlocksCacheErr::GrpcError(err) }
 }
@@ -55,6 +76,7 @@ pub enum ZcoinClientInitError {
     ZcashSqliteError(ZcashClientError),
     EmptyLightwalletdUris,
     InvalidUri(InvalidUri),
+    UrlIterFailure(String),
 }
 
 impl From<ZcashClientError> for ZcoinClientInitError {

mm2src/coins/z_coin/z_rpc.rs

@@ -16,8 +16,10 @@ use mm2_err_handle::prelude::*;
 use parking_lot::Mutex;
 use prost::Message;
 use protobuf::Message as ProtobufMessage;
+use std::future::Future;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
+use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::task::block_in_place;
@@ -69,12 +71,15 @@ pub trait ZRpcOps {
 }
 
 #[async_trait]
-impl ZRpcOps for CompactTxStreamerClient<Channel> {
+impl ZRpcOps for Vec<CompactTxStreamerClient<Channel>> {
     async fn get_block_height(&mut self) -> Result<u64, MmError<UpdateBlocksCacheErr>> {
-        let request = tonic::Request::new(ChainSpec {});
-        let block = self.get_latest_block(request).await?;
-        let res = block.into_inner().height;
-        Ok(res)
+        let block = send_multi_light_wallet_request_for_tonic(self, |client| {
+            let request = tonic::Request::new(ChainSpec {});
+            client.get_latest_block(request)
+        })
+        .await
+        .map_to_mm(UpdateBlocksCacheErr::GrpcVecError)?;
+        Ok(block.height)
     }
 
     async fn scan_blocks(
@@ -83,48 +88,55 @@ impl ZRpcOps for CompactTxStreamerClient<Channel> {
         last_block: u64,
         on_block: &mut OnCompactBlockFn,
     ) -> Result<(), MmError<UpdateBlocksCacheErr>> {
-        let request = tonic::Request::new(BlockRange {
-            start: Some(BlockId {
-                height: start_block,
-                hash: Vec::new(),
-            }),
-            end: Some(BlockId {
-                height: last_block,
-                hash: Vec::new(),
-            }),
-        });
-        let mut response = self.get_block_range(request).await?;
-        while let Some(block) = response.get_mut().message().await? {
+        let mut response = send_multi_light_wallet_request_for_tonic(self, |client| {
+            let request = tonic::Request::new(BlockRange {
+                start: Some(BlockId {
+                    height: start_block,
+                    hash: Vec::new(),
+                }),
+                end: Some(BlockId {
+                    height: last_block,
+                    hash: Vec::new(),
+                }),
+            });
+            client.get_block_range(request)
+        })
+        .await
+        .map_to_mm(UpdateBlocksCacheErr::GrpcVecError)?;
+        // without Pin method get_mut is not found in current scope
+        while let Some(block) = Pin::new(&mut response).get_mut().message().await? {
             debug!("Got block {:?}", block);
             on_block(block)?;
         }
         Ok(())
     }
 
     async fn check_tx_existence(&mut self, tx_id: TxId) -> bool {
-        let mut attempts = 0;
-        loop {
-            let filter = TxFilter {
-                block: None,
-                index: 0,
-                hash: tx_id.0.into(),
-            };
-            let request = tonic::Request::new(filter);
-            match self.get_transaction(request).await {
-                Ok(_) => break,
-                Err(e) => {
-                    error!("Error on getting tx {}", tx_id);
-                    if e.message().contains(NO_TX_ERROR_CODE) {
-                        if attempts >= 3 {
-                            return false;
+        for client in self {
+            let mut attempts = 0;
+            loop {
+                let filter = TxFilter {
+                    block: None,
+                    index: 0,
+                    hash: tx_id.0.into(),
+                };
+                let request = tonic::Request::new(filter);
+                match client.get_transaction(request).await {
+                    Ok(_) => return true,
+                    Err(e) => {
+                        error!("Error on getting tx {}", tx_id);
+                        if e.message().contains(NO_TX_ERROR_CODE) {
+                            if attempts >= 3 {
+                                break;
+                            }
+                            attempts += 1;
                         }
-                        attempts += 1;
-                    }
-                    Timer::sleep(30.).await;
-                },
+                        Timer::sleep(30.).await;
+                    },
+                }
             }
         }
-        true
+        false
     }
 }
 
@@ -370,12 +382,49 @@ pub(super) async fn init_light_client(
 ) -> Result<(AsyncMutex<SaplingSyncConnector>, WalletDbShared), MmError<ZcoinClientInitError>> {
     let (sync_status_notifier, sync_watcher) = channel(1);
     let (on_tx_gen_notifier, on_tx_gen_watcher) = channel(1);
-
-    let lightwalletd_url = Uri::from_str(
-        lightwalletd_urls
-            .first()
-            .or_mm_err(|| ZcoinClientInitError::EmptyLightwalletdUris)?,
-    )?;
+    let mut rpc_clients = Vec::new();
+    let mut errors = Vec::new();
+    if !lightwalletd_urls.is_empty() {
+        for url in lightwalletd_urls {
+            let uri = match Uri::from_str(&*url).map_to_mm(ZcoinClientInitError::InvalidUri) {
+                Ok(uri) => uri,
+                Err(err) => {
+                    errors.push(format!("{:?}", err));
+                    continue;
+                },
+            };
+            let endpoint = match Channel::builder(uri)
+                .tls_config(ClientTlsConfig::new())
+                .map_to_mm(ZcoinClientInitError::TlsConfigFailure)
+            {
+                Ok(endpoint) => endpoint,
+                Err(err) => {
+                    errors.push(format!("{:?}", err));
+                    continue;
+                },
+            };
+            let tonic_channel = match endpoint
+                .connect()
+                .await
+                .map_to_mm(ZcoinClientInitError::ConnectionFailure)
+            {
+                Ok(tonic_channel) => tonic_channel,
+                Err(err) => {
+                    errors.push(format!("{:?}", err));
+                    continue;
+                },
+            };
+            let rpc_copy = CompactTxStreamerClient::new(tonic_channel);
+            rpc_clients.push(rpc_copy);
+        }
+        drop_mutability!(errors);
+        // check if rpc_clients is empty, then for loop wasnt successful
+        if rpc_clients.is_empty() {
+            return Err(format_init_client_e(errors));
+        }
+    } else {
+        return Err(MmError::from(ZcoinClientInitError::EmptyLightwalletdUris));
+    }
 
     let sync_handle = SaplingSyncLoopHandle {
         current_block: BlockHeight::from_u32(0),
@@ -387,14 +436,8 @@ pub(super) async fn init_light_client(
         watch_for_tx: None,
     };
 
-    let tonic_channel = Channel::builder(lightwalletd_url)
-        .tls_config(ClientTlsConfig::new())
-        .map_to_mm(ZcoinClientInitError::TlsConfigFailure)?
-        .connect()
-        .await
-        .map_to_mm(ZcoinClientInitError::ConnectionFailure)?;
-    let rpc_copy = CompactTxStreamerClient::new(tonic_channel);
-    let abort_handle = spawn_abortable(light_wallet_db_sync_loop(sync_handle, Box::new(rpc_copy)));
+    drop_mutability!(rpc_clients);
+    let abort_handle = spawn_abortable(light_wallet_db_sync_loop(sync_handle, Box::new(rpc_clients)));
 
     Ok((
         SaplingSyncConnector::new_mutex_wrapped(sync_watcher, on_tx_gen_notifier, abort_handle),
@@ -726,3 +769,27 @@ pub(super) struct SaplingSyncGuard<'a> {
     pub(super) _connector_guard: AsyncMutexGuard<'a, SaplingSyncConnector>,
     pub(super) respawn_guard: SaplingSyncRespawnGuard,
 }
+
+fn format_init_client_e(errors: Vec<String>) -> MmError<ZcoinClientInitError> {
+    let errors: String = errors.iter().map(|e| format!("{:?}", e)).collect();
+    let error = format!("Init client failed during urls iteration: {}", errors);
+    MmError::from(ZcoinClientInitError::UrlIterFailure(error))
+}
+
+async fn send_multi_light_wallet_request_for_tonic<'a, Res, Fut, Fn>(
+    clients: &'a mut [CompactTxStreamerClient<Channel>],
+    mut req_fn: Fn,
+) -> Result<Res, Vec<tonic::Status>>
+where
+    Fut: Future<Output = Result<tonic::Response<Res>, tonic::Status>>,
+    Fn: FnMut(&'a mut CompactTxStreamerClient<Channel>) -> Fut,
+{
+    let mut errors = Vec::new();
+    for client in clients.iter_mut() {
+        match req_fn(client).await {
+            Ok(res) => return Ok(res.into_inner()),
+            Err(e) => errors.push(e),
+        }
+    }
+    Err(errors)
+}