[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

@@ -18,6 +18,8 @@ use zcash_primitives::transaction::builder::Error as ZTxBuilderError;
 #[non_exhaustive]
 pub enum UpdateBlocksCacheErr {
     GrpcError(tonic::Status),
+    #[display(fmt = "Fail to send requests during clients iteration {:?}", _0)]
+    GrpcMultiError(Vec<tonic::Status>),
     BlocksDbError(SqliteError),
     ZcashSqliteError(ZcashClientError),
     UtxoRpcError(UtxoRpcError),
@@ -48,21 +50,23 @@ impl From<JsonRpcError> for UpdateBlocksCacheErr {
 #[derive(Debug, Display)]
 #[non_exhaustive]
 pub enum ZcoinClientInitError {
-    TlsConfigFailure(tonic::transport::Error),
-    ConnectionFailure(tonic::transport::Error),
     BlocksDbInitFailure(SqliteError),
     WalletDbInitFailure(SqliteError),
     ZcashSqliteError(ZcashClientError),
     EmptyLightwalletdUris,
-    InvalidUri(InvalidUri),
+    #[display(fmt = "Fail to init clients while iterating lightwalletd urls {:?}", _0)]
+    UrlIterFailure(Vec<UrlIterError>),
 }
 
 impl From<ZcashClientError> for ZcoinClientInitError {
     fn from(err: ZcashClientError) -> Self { ZcoinClientInitError::ZcashSqliteError(err) }
 }
 
-impl From<InvalidUri> for ZcoinClientInitError {
-    fn from(err: InvalidUri) -> Self { ZcoinClientInitError::InvalidUri(err) }
+#[derive(Debug, Display)]
+pub enum UrlIterError {
+    InvalidUri(InvalidUri),
+    TlsConfigFailure(tonic::transport::Error),
+    ConnectionFailure(tonic::transport::Error),
 }
 
 #[derive(Debug, Display)]

mm2src/coins/z_coin/z_rpc.rs

@@ -16,8 +16,9 @@ use mm2_err_handle::prelude::*;
 use parking_lot::Mutex;
 use prost::Message;
 use protobuf::Message as ProtobufMessage;
-use std::ops::Deref;
+use std::future::Future;
 use std::path::{Path, PathBuf};
+use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::task::block_in_place;
@@ -51,8 +52,7 @@ struct CompactBlockRow {
     data: Vec<u8>,
 }
 
-pub type OnCompactBlockFn<'a> =
-    dyn FnMut(TonicCompactBlock) -> Result<(), MmError<UpdateBlocksCacheErr>> + Send + Sync + 'a;
+pub type OnCompactBlockFn<'a> = dyn FnMut(TonicCompactBlock) -> Result<(), MmError<UpdateBlocksCacheErr>> + Send + 'a;
 
 #[async_trait]
 pub trait ZRpcOps {
@@ -69,12 +69,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(self, |client| {
+            let request = tonic::Request::new(ChainSpec {});
+            client.get_latest_block(request)
+        })
+        .await
+        .map_to_mm(UpdateBlocksCacheErr::GrpcMultiError)?;
+        Ok(block.height)
     }
 
     async fn scan_blocks(
@@ -83,18 +86,23 @@ 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(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::GrpcMultiError)?;
+        // 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)?;
         }
@@ -104,17 +112,22 @@ impl ZRpcOps for CompactTxStreamerClient<Channel> {
     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 {
+            match send_multi_light_wallet_request(self, |client| {
+                let filter = TxFilter {
+                    block: None,
+                    index: 0,
+                    hash: tx_id.0.into(),
+                };
+                let request = tonic::Request::new(filter);
+                client.get_transaction(request)
+            })
+            .await
+            {
                 Ok(_) => break,
                 Err(e) => {
                     error!("Error on getting tx {}", tx_id);
-                    if e.message().contains(NO_TX_ERROR_CODE) {
+                    let mut e = e;
+                    if e.remove(0).message().contains(NO_TX_ERROR_CODE) {
                         if attempts >= 3 {
                             return false;
                         }
@@ -370,31 +383,53 @@ 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() {
+        return MmError::err(ZcoinClientInitError::EmptyLightwalletdUris);
+    }
+    for url in lightwalletd_urls {
+        let uri = match Uri::from_str(&*url) {
+            Ok(uri) => uri,
+            Err(err) => {
+                errors.push(UrlIterError::InvalidUri(err));
+                continue;
+            },
+        };
+        let endpoint = match Channel::builder(uri).tls_config(ClientTlsConfig::new()) {
+            Ok(endpoint) => endpoint,
+            Err(err) => {
+                errors.push(UrlIterError::TlsConfigFailure(err));
+                continue;
+            },
+        };
+        let tonic_channel = match endpoint.connect().await {
+            Ok(tonic_channel) => tonic_channel,
+            Err(err) => {
+                errors.push(UrlIterError::ConnectionFailure(err));
+                continue;
+            },
+        };
+        rpc_clients.push(CompactTxStreamerClient::new(tonic_channel));
+    }
+    drop_mutability!(errors);
+    // check if rpc_clients is empty, then for loop wasn't successful
+    if rpc_clients.is_empty() {
+        return MmError::err(ZcoinClientInitError::UrlIterFailure(errors));
+    }
 
     let sync_handle = SaplingSyncLoopHandle {
         current_block: BlockHeight::from_u32(0),
-        blocks_db: Mutex::new(blocks_db),
+        blocks_db,
         wallet_db: wallet_db.clone(),
         consensus_params,
         sync_status_notifier,
         on_tx_gen_watcher,
         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),
@@ -413,7 +448,7 @@ pub(super) async fn init_native_client(
 
     let sync_handle = SaplingSyncLoopHandle {
         current_block: BlockHeight::from_u32(0),
-        blocks_db: Mutex::new(blocks_db),
+        blocks_db,
         wallet_db: wallet_db.clone(),
         consensus_params,
         sync_status_notifier,
@@ -479,7 +514,7 @@ pub enum SyncStatus {
 
 pub struct SaplingSyncLoopHandle {
     current_block: BlockHeight,
-    blocks_db: Mutex<BlockDb>,
+    blocks_db: BlockDb,
     wallet_db: WalletDbShared,
     consensus_params: ZcoinConsensusParams,
     /// Notifies about sync status without stopping the loop, e.g. on coin activation
@@ -528,7 +563,7 @@ impl SaplingSyncLoopHandle {
         rpc: &mut (dyn ZRpcOps + Send),
     ) -> Result<(), MmError<UpdateBlocksCacheErr>> {
         let current_block = rpc.get_block_height().await?;
-        let current_block_in_db = block_in_place(|| self.blocks_db.lock().get_latest_block())?;
+        let current_block_in_db = block_in_place(|| self.blocks_db.get_latest_block())?;
         let extrema = block_in_place(|| self.wallet_db.lock().block_height_extrema())?;
         let mut from_block = self
             .consensus_params
@@ -540,11 +575,7 @@ impl SaplingSyncLoopHandle {
         }
         if current_block >= from_block {
             rpc.scan_blocks(from_block, current_block, &mut |block: TonicCompactBlock| {
-                block_in_place(|| {
-                    self.blocks_db
-                        .lock()
-                        .insert_block(block.height as u32, block.encode_to_vec())
-                })?;
+                block_in_place(|| self.blocks_db.insert_block(block.height as u32, block.encode_to_vec()))?;
                 self.notify_blocks_cache_status(block.height, current_block);
                 Ok(())
             })
@@ -564,7 +595,7 @@ impl SaplingSyncLoopHandle {
 
         if let Err(e) = validate_chain(
             &self.consensus_params,
-            self.blocks_db.lock().deref(),
+            &self.blocks_db,
             wallet_ops.get_max_height_hash()?,
         ) {
             match e {
@@ -575,13 +606,13 @@ impl SaplingSyncLoopHandle {
                         BlockHeight::from_u32(0)
                     };
                     wallet_ops.rewind_to_height(rewind_height)?;
-                    self.blocks_db.lock().rewind_to_height(rewind_height.into())?;
+                    self.blocks_db.rewind_to_height(rewind_height.into())?;
                 },
                 e => return MmError::err(e),
             }
         }
 
-        let current_block = BlockHeight::from_u32(self.blocks_db.lock().get_latest_block()?);
+        let current_block = BlockHeight::from_u32(self.blocks_db.get_latest_block()?);
         loop {
             match wallet_ops.block_height_extrema()? {
                 Some((_, max_in_wallet)) => {
@@ -593,12 +624,7 @@ impl SaplingSyncLoopHandle {
                 },
                 None => self.notify_building_wallet_db(0, current_block.into()),
             }
-            scan_cached_blocks(
-                &self.consensus_params,
-                self.blocks_db.lock().deref(),
-                &mut wallet_ops,
-                Some(1000),
-            )?;
+            scan_cached_blocks(&self.consensus_params, &self.blocks_db, &mut wallet_ops, Some(1000))?;
         }
         Ok(())
     }
@@ -726,3 +752,21 @@ pub(super) struct SaplingSyncGuard<'a> {
     pub(super) _connector_guard: AsyncMutexGuard<'a, SaplingSyncConnector>,
     pub(super) respawn_guard: SaplingSyncRespawnGuard,
 }
+
+async fn send_multi_light_wallet_request<'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)
+}