[r2r] Multi lightwalletd servers integration

mm2src/coins/z_coin.rs

@@ -793,7 +793,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

@@ -23,6 +23,7 @@ pub enum UpdateBlocksCacheErr {
     UtxoRpcError(UtxoRpcError),
     InternalError(String),
     JsonRpcError(JsonRpcError),
+    ClientIterError(String),
 }
 
 impl From<tonic::Status> for UpdateBlocksCacheErr {
@@ -55,6 +56,7 @@ pub enum ZcoinClientInitError {
     ZcashSqliteError(ZcashClientError),
     EmptyLightwalletdUris,
     InvalidUri(InvalidUri),
+    UrlIterFailure(String),
 }
 
 impl From<ZcashClientError> for ZcoinClientInitError {

mm2src/coins/z_coin/z_rpc.rs

@@ -69,12 +69,21 @@ 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 mut errors = Vec::new();
+        for client in self {
+            let request = tonic::Request::new(ChainSpec {});
+            match client.get_latest_block(request).await {
+                Ok(block) => return Ok(block.into_inner().height),
+                Err(err) => {
+                    errors.push(format!("{:?}", err));
+                    continue;
+                },
+            };
+        }
+        drop_mutability!(errors);
+        Err(format_update_blocks_e(errors))
     }
 
     async fn scan_blocks(
@@ -83,48 +92,74 @@ 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? {
-            debug!("Got block {:?}", block);
-            on_block(block)?;
+        let mut errors = Vec::new();
+        for client in self {
+            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 = match client.get_block_range(request).await {
+                Ok(response) => response,
+                Err(err) => {
+                    errors.push(format!("{:?}", err));
+                    continue;
+                },
+            };
+            loop {
+                match response.get_mut().message().await {
+                    Ok(block) => match block {
+                        Some(block) => {
+                            debug!("Got block {:?}", block);
+                            if let Err(err) = on_block(block) {
+                                errors.push(format!("{:?}", err));
+                                break;
+                            }
+                        },
+                        _ => return Ok(()),
+                    },
+                    Err(err) => {
+                        errors.push(format!("{:?}", err));
+                        break;
+                    },
+                }
+            }
         }
-        Ok(())
+        drop_mutability!(errors);
+        Err(format_update_blocks_e(errors))
     }
 
     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 +405,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 +459,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 +792,15 @@ pub(super) struct SaplingSyncGuard<'a> {
     pub(super) _connector_guard: AsyncMutexGuard<'a, SaplingSyncConnector>,
     pub(super) respawn_guard: SaplingSyncRespawnGuard,
 }
+
+fn format_update_blocks_e(errors: Vec<String>) -> MmError<UpdateBlocksCacheErr> {
+    let errors: String = errors.iter().map(|e| format!("{:?}", e)).collect();
+    let error = format!("Update blocks cache error during client iteration: {}", errors);
+    MmError::from(UpdateBlocksCacheErr::ClientIterError(error))
+}
+
+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))
+}