refactor: use wallet connectivity in wallet services (#3391)

* refactor: reintroduce generic substream in client rpc

Generic substream was removed in order to more closely test RPC/yamux
interaction but that was incorrect. The tests should have just been
updated without removing the generic. This PR reintroduces them so that
an RPC client mock can more easily be created.

* refactor: use wallet connectivity in wallet services

- use wallet connectivity in broadcast protocol
- remove base node update code in multiple services
- use wallet connectivity in transaction validation protocol
- use wallet connectivity in TXO validation task
- update/fix tests

applications/tari_console_wallet/Cargo.toml

@@ -56,4 +56,4 @@ default-features = false
 features = ["crossterm"]
 
 [features]
-avx2 = []
+avx2 = []

applications/tari_console_wallet/src/init/mod.rs

@@ -386,7 +386,6 @@ pub async fn init_wallet(
             base_node_query_timeout: config.base_node_query_timeout,
             prevent_fee_gt_amount: config.prevent_fee_gt_amount,
             event_channel_size: config.output_manager_event_channel_size,
-            base_node_update_publisher_channel_size: config.base_node_update_publisher_channel_size,
             num_confirmations_required: config.transaction_num_confirmations_required,
             ..Default::default()
         }),

applications/tari_console_wallet/src/ui/components/base_node.rs

@@ -21,7 +21,7 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use crate::ui::{components::Component, state::AppState};
-use tari_wallet::connectivity_service::OnlineStatus;
+use tari_wallet::connectivity_service::{OnlineStatus, WalletConnectivityInterface};
 use tui::{
     backend::Backend,
     layout::Rect,

applications/tari_console_wallet/src/ui/state/wallet_event_monitor.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 use tari_comms::{connectivity::ConnectivityEvent, peer_manager::Peer};
 use tari_wallet::{
     base_node_service::{handle::BaseNodeEvent, service::BaseNodeState},
+    connectivity_service::WalletConnectivityInterface,
     output_manager_service::{handle::OutputManagerEvent, TxId},
     transaction_service::handle::TransactionEvent,
 };
@@ -62,6 +63,7 @@ impl WalletEventMonitor {
         let mut connectivity_events = self.app_state_inner.read().await.get_connectivity_event_stream();
         let wallet_connectivity = self.app_state_inner.read().await.get_wallet_connectivity();
         let mut connectivity_status = wallet_connectivity.get_connectivity_status_watch();
+        let mut base_node_changed = wallet_connectivity.get_current_base_node_watcher();
 
         let mut base_node_events = self.app_state_inner.read().await.get_base_node_event_stream();
         let mut software_update_notif = self
@@ -166,6 +168,13 @@ impl WalletEventMonitor {
                             Err(broadcast::error::RecvError::Closed) => {}
                         }
                     },
+                    _ = base_node_changed.changed() => {
+                        let peer = base_node_changed.borrow().as_ref().cloned();
+                        if let Some(peer) = peer {
+                            self.trigger_base_node_peer_refresh(peer).await;
+                            self.trigger_balance_refresh();
+                        }
+                    }
                     result = base_node_events.recv() => {
                         match result {
                             Ok(msg) => {
@@ -174,10 +183,6 @@ impl WalletEventMonitor {
                                     BaseNodeEvent::BaseNodeStateChanged(state) => {
                                         self.trigger_base_node_state_refresh(state).await;
                                     }
-                                    BaseNodeEvent::BaseNodePeerSet(peer) => {
-                                        self.trigger_base_node_peer_refresh(*peer).await;
-                                        self.trigger_balance_refresh();
-                                    }
                                 }
                             },
                             Err(broadcast::error::RecvError::Lagged(n)) => {

applications/tari_console_wallet/src/wallet_modes.rs

@@ -223,7 +223,7 @@ pub fn tui_mode(config: WalletModeConfig, mut wallet: WalletSqlite) -> Result<()
     base_node_config.base_node_custom = base_node_custom.clone();
     if let Some(peer) = base_node_custom {
         base_node_selected = peer;
-    } else if let Some(peer) = handle.block_on(wallet.get_base_node_peer())? {
+    } else if let Some(peer) = handle.block_on(wallet.get_base_node_peer()) {
         base_node_selected = peer;
     }
 

base_layer/wallet/Cargo.toml

@@ -19,27 +19,28 @@ tari_shutdown = { version = "^0.10", path = "../../infrastructure/shutdown" }
 tari_storage = { version = "^0.10", path = "../../infrastructure/storage" }
 
 aes-gcm = "^0.8"
+async-trait = "0.1.50"
+bincode = "1.3.1"
 blake2 = "0.9.0"
 chrono = { version = "0.4.6", features = ["serde"] }
 crossbeam-channel = "0.3.8"
-digest = "0.9.0"
 diesel = { version = "1.4.7", features = ["sqlite", "serde_json", "chrono"] }
 diesel_migrations = "1.4.0"
-libsqlite3-sys = { version = ">=0.8.0, <0.13.0", features = ["bundled"], optional = true }
+digest = "0.9.0"
 fs2 = "0.3.0"
 futures = { version = "^0.3.1", features = ["compat", "std"] }
+libsqlite3-sys = { version = ">=0.8.0, <0.13.0", features = ["bundled"], optional = true }
+lmdb-zero = "0.4.4"
 log = "0.4.6"
 log4rs = { version = "1.0.0", features = ["console_appender", "file_appender", "yaml_format"] }
-lmdb-zero = "0.4.4"
 rand = "0.8"
 serde = { version = "1.0.89", features = ["derive"] }
 serde_json = "1.0.39"
-tokio = { version = "1.11", features = ["sync", "macros"] }
-tower = "0.3.0-alpha.2"
 tempfile = "3.1.0"
-time = { version = "0.1.39" }
 thiserror = "1.0.26"
-bincode = "1.3.1"
+time = { version = "0.1.39" }
+tokio = { version = "1.11", features = ["sync", "macros"] }
+tower = "0.3.0-alpha.2"
 
 [dependencies.tari_core]
 path = "../../base_layer/core"

base_layer/wallet/src/base_node_service/handle.rs

@@ -23,7 +23,6 @@
 use super::{error::BaseNodeServiceError, service::BaseNodeState};
 use std::{sync::Arc, time::Duration};
 use tari_common_types::chain_metadata::ChainMetadata;
-use tari_comms::peer_manager::Peer;
 use tari_service_framework::reply_channel::SenderService;
 use tokio::sync::broadcast;
 use tower::Service;
@@ -34,22 +33,17 @@ pub type BaseNodeEventReceiver = broadcast::Receiver<Arc<BaseNodeEvent>>;
 #[derive(Debug)]
 pub enum BaseNodeServiceRequest {
     GetChainMetadata,
-    SetBaseNodePeer(Box<Peer>),
-    GetBaseNodePeer,
     GetBaseNodeLatency,
 }
 /// API Response enum
 #[derive(Debug)]
 pub enum BaseNodeServiceResponse {
     ChainMetadata(Option<ChainMetadata>),
-    BaseNodePeerSet,
-    BaseNodePeer(Option<Box<Peer>>),
     Latency(Option<Duration>),
 }
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
 pub enum BaseNodeEvent {
     BaseNodeStateChanged(BaseNodeState),
-    BaseNodePeerSet(Box<Peer>),
 }
 
 /// The Base Node Service Handle is a struct that contains the interfaces used to communicate with a running
@@ -82,24 +76,6 @@ impl BaseNodeServiceHandle {
         }
     }
 
-    pub async fn set_base_node_peer(&mut self, peer: Peer) -> Result<(), BaseNodeServiceError> {
-        match self
-            .handle
-            .call(BaseNodeServiceRequest::SetBaseNodePeer(Box::new(peer)))
-            .await??
-        {
-            BaseNodeServiceResponse::BaseNodePeerSet => Ok(()),
-            _ => Err(BaseNodeServiceError::UnexpectedApiResponse),
-        }
-    }
-
-    pub async fn get_base_node_peer(&mut self) -> Result<Option<Peer>, BaseNodeServiceError> {
-        match self.handle.call(BaseNodeServiceRequest::GetBaseNodePeer).await?? {
-            BaseNodeServiceResponse::BaseNodePeer(peer) => Ok(peer.map(|p| *p)),
-            _ => Err(BaseNodeServiceError::UnexpectedApiResponse),
-        }
-    }
-
     pub async fn get_base_node_latency(&mut self) -> Result<Option<Duration>, BaseNodeServiceError> {
         match self.handle.call(BaseNodeServiceRequest::GetBaseNodeLatency).await?? {
             BaseNodeServiceResponse::Latency(latency) => Ok(latency),

base_layer/wallet/src/base_node_service/mock_base_node_service.rs

@@ -103,24 +103,12 @@ impl MockBaseNodeService {
         }
     }
 
-    fn set_base_node_peer(&mut self, peer: Peer) {
-        self.base_node_peer = Some(peer);
-    }
-
     /// This handler is called when requests arrive from the various streams
     fn handle_request(
         &mut self,
         request: BaseNodeServiceRequest,
     ) -> Result<BaseNodeServiceResponse, BaseNodeServiceError> {
         match request {
-            BaseNodeServiceRequest::SetBaseNodePeer(peer) => {
-                self.set_base_node_peer(*peer);
-                Ok(BaseNodeServiceResponse::BaseNodePeerSet)
-            },
-            BaseNodeServiceRequest::GetBaseNodePeer => {
-                let peer = self.base_node_peer.clone();
-                Ok(BaseNodeServiceResponse::BaseNodePeer(peer.map(Box::new)))
-            },
             BaseNodeServiceRequest::GetChainMetadata => Ok(BaseNodeServiceResponse::ChainMetadata(
                 self.state.chain_metadata.clone(),
             )),

base_layer/wallet/src/base_node_service/monitor.rs

@@ -25,7 +25,7 @@ use crate::{
         handle::{BaseNodeEvent, BaseNodeEventSender},
         service::BaseNodeState,
     },
-    connectivity_service::WalletConnectivityHandle,
+    connectivity_service::WalletConnectivityInterface,
     error::WalletStorageError,
     storage::database::{WalletBackend, WalletDatabase},
 };
@@ -42,20 +42,24 @@ use tokio::{sync::RwLock, time};
 
 const LOG_TARGET: &str = "wallet::base_node_service::chain_metadata_monitor";
 
-pub struct BaseNodeMonitor<T> {
+pub struct BaseNodeMonitor<TBackend, TWalletConnectivity> {
     interval: Duration,
     state: Arc<RwLock<BaseNodeState>>,
-    db: WalletDatabase<T>,
-    wallet_connectivity: WalletConnectivityHandle,
+    db: WalletDatabase<TBackend>,
+    wallet_connectivity: TWalletConnectivity,
     event_publisher: BaseNodeEventSender,
 }
 
-impl<T: WalletBackend + 'static> BaseNodeMonitor<T> {
+impl<TBackend, TWalletConnectivity> BaseNodeMonitor<TBackend, TWalletConnectivity>
+where
+    TBackend: WalletBackend + 'static,
+    TWalletConnectivity: WalletConnectivityInterface,
+{
     pub fn new(
         interval: Duration,
         state: Arc<RwLock<BaseNodeState>>,
-        db: WalletDatabase<T>,
-        wallet_connectivity: WalletConnectivityHandle,
+        db: WalletDatabase<TBackend>,
+        wallet_connectivity: TWalletConnectivity,
         event_publisher: BaseNodeEventSender,
     ) -> Self {
         Self {

base_layer/wallet/src/base_node_service/service.rs

@@ -23,7 +23,7 @@
 use super::{
     config::BaseNodeServiceConfig,
     error::BaseNodeServiceError,
-    handle::{BaseNodeEvent, BaseNodeEventSender, BaseNodeServiceRequest, BaseNodeServiceResponse},
+    handle::{BaseNodeEventSender, BaseNodeServiceRequest, BaseNodeServiceResponse},
 };
 use crate::{
     base_node_service::monitor::BaseNodeMonitor,
@@ -35,7 +35,6 @@ use futures::{future, StreamExt};
 use log::*;
 use std::{sync::Arc, time::Duration};
 use tari_common_types::chain_metadata::ChainMetadata;
-use tari_comms::peer_manager::Peer;
 use tari_service_framework::reply_channel::Receiver;
 use tari_shutdown::ShutdownSignal;
 use tokio::sync::RwLock;
@@ -153,12 +152,6 @@ where T: WalletBackend + 'static
         Ok(())
     }
 
-    async fn set_base_node_peer(&mut self, peer: Peer) -> Result<(), BaseNodeServiceError> {
-        self.wallet_connectivity.set_base_node(peer.clone()).await?;
-        self.publish_event(BaseNodeEvent::BaseNodePeerSet(Box::new(peer)));
-        Ok(())
-    }
-
     /// This handler is called when requests arrive from the various streams
     async fn handle_request(
         &mut self,
@@ -169,14 +162,6 @@ where T: WalletBackend + 'static
             "Handling Wallet Base Node Service Request: {:?}", request
         );
         match request {
-            BaseNodeServiceRequest::SetBaseNodePeer(peer) => {
-                self.set_base_node_peer(*peer).await?;
-                Ok(BaseNodeServiceResponse::BaseNodePeerSet)
-            },
-            BaseNodeServiceRequest::GetBaseNodePeer => {
-                let peer = self.wallet_connectivity.get_current_base_node_peer().map(Box::new);
-                Ok(BaseNodeServiceResponse::BaseNodePeer(peer))
-            },
             BaseNodeServiceRequest::GetChainMetadata => match self.get_state().await.chain_metadata.clone() {
                 Some(metadata) => Ok(BaseNodeServiceResponse::ChainMetadata(Some(metadata))),
                 None => {
@@ -190,14 +175,4 @@ where T: WalletBackend + 'static
             },
         }
     }
-
-    fn publish_event(&self, event: BaseNodeEvent) {
-        trace!(target: LOG_TARGET, "Publishing event: {:?}", event);
-        let _ = self.event_publisher.send(Arc::new(event)).map_err(|_| {
-            trace!(
-                target: LOG_TARGET,
-                "Could not publish BaseNodeEvent as there are no subscribers"
-            )
-        });
-    }
 }

base_layer/wallet/src/connectivity_service/handle.rs

@@ -21,10 +21,11 @@
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use super::service::OnlineStatus;
-use crate::connectivity_service::{error::WalletConnectivityError, watch::Watch};
+use crate::{connectivity_service::WalletConnectivityInterface, util::watch::Watch};
 use tari_comms::{
     peer_manager::{NodeId, Peer},
     protocol::rpc::RpcClientLease,
+    types::CommsPublicKey,
 };
 use tari_core::base_node::{rpc::BaseNodeWalletRpcClient, sync::rpc::BaseNodeSyncRpcClient};
 use tokio::sync::{mpsc, oneshot, watch};
@@ -53,10 +54,16 @@ impl WalletConnectivityHandle {
             online_status_rx,
         }
     }
+}
+
+#[async_trait::async_trait]
+impl WalletConnectivityInterface for WalletConnectivityHandle {
+    fn set_base_node(&mut self, base_node_peer: Peer) {
+        self.base_node_watch.send(Some(base_node_peer));
+    }
 
-    pub async fn set_base_node(&mut self, base_node_peer: Peer) -> Result<(), WalletConnectivityError> {
-        self.base_node_watch.broadcast(Some(base_node_peer));
-        Ok(())
+    fn get_current_base_node_watcher(&self) -> watch::Receiver<Option<Peer>> {
+        self.base_node_watch.get_receiver()
     }
 
     /// Obtain a BaseNodeWalletRpcClient.
@@ -65,7 +72,7 @@ impl WalletConnectivityHandle {
     /// node/nodes. It will block until this happens. The ONLY other time it will return is if the node is
     /// shutting down, where it will return None. Use this function whenever no work can be done without a
     /// BaseNodeWalletRpcClient RPC session.
-    pub async fn obtain_base_node_wallet_rpc_client(&mut self) -> Option<RpcClientLease<BaseNodeWalletRpcClient>> {
+    async fn obtain_base_node_wallet_rpc_client(&mut self) -> Option<RpcClientLease<BaseNodeWalletRpcClient>> {
         let (reply_tx, reply_rx) = oneshot::channel();
         // Under what conditions do the (1) mpsc channel and (2) oneshot channel error?
         // (1) when the receiver has been dropped
@@ -88,7 +95,7 @@ impl WalletConnectivityHandle {
     /// node/nodes. It will block until this happens. The ONLY other time it will return is if the node is
     /// shutting down, where it will return None. Use this function whenever no work can be done without a
     /// BaseNodeSyncRpcClient RPC session.
-    pub async fn obtain_base_node_sync_rpc_client(&mut self) -> Option<RpcClientLease<BaseNodeSyncRpcClient>> {
+    async fn obtain_base_node_sync_rpc_client(&mut self) -> Option<RpcClientLease<BaseNodeSyncRpcClient>> {
         let (reply_tx, reply_rx) = oneshot::channel();
         self.sender
             .send(WalletConnectivityRequest::ObtainBaseNodeSyncRpcClient(reply_tx))
@@ -98,19 +105,27 @@ impl WalletConnectivityHandle {
         reply_rx.await.ok()
     }
 
-    pub fn get_connectivity_status(&mut self) -> OnlineStatus {
+    fn get_connectivity_status(&mut self) -> OnlineStatus {
         *self.online_status_rx.borrow()
     }
 
-    pub fn get_connectivity_status_watch(&self) -> watch::Receiver<OnlineStatus> {
+    fn get_connectivity_status_watch(&self) -> watch::Receiver<OnlineStatus> {
         self.online_status_rx.clone()
     }
 
-    pub fn get_current_base_node_peer(&self) -> Option<Peer> {
+    fn get_current_base_node_peer(&self) -> Option<Peer> {
         self.base_node_watch.borrow().clone()
     }
 
-    pub fn get_current_base_node_id(&self) -> Option<NodeId> {
+    fn get_current_base_node_peer_public_key(&self) -> Option<CommsPublicKey> {
+        self.base_node_watch.borrow().as_ref().map(|p| p.public_key.clone())
+    }
+
+    fn get_current_base_node_id(&self) -> Option<NodeId> {
         self.base_node_watch.borrow().as_ref().map(|p| p.node_id.clone())
     }
+
+    fn is_base_node_set(&self) -> bool {
+        self.base_node_watch.borrow().is_some()
+    }
 }

base_layer/wallet/src/connectivity_service/initializer.rs

@@ -28,8 +28,12 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use super::{handle::WalletConnectivityHandle, service::WalletConnectivityService, watch::Watch};
-use crate::{base_node_service::config::BaseNodeServiceConfig, connectivity_service::service::OnlineStatus};
+use super::{handle::WalletConnectivityHandle, service::WalletConnectivityService};
+use crate::{
+    base_node_service::config::BaseNodeServiceConfig,
+    connectivity_service::service::OnlineStatus,
+    util::watch::Watch,
+};
 use tari_service_framework::{async_trait, ServiceInitializationError, ServiceInitializer, ServiceInitializerContext};
 use tokio::sync::mpsc;