From 6ff3eceef6dd631f2976e42cfdb6ba494c746a8c Mon Sep 17 00:00:00 2001
From: Hansie Odendaal <pluto@tari.com>
Date: Wed, 18 Aug 2021 15:35:44 +0200
Subject: [PATCH] Add disconnect/re-connect to base node monitor

- This PR disconnects and re-sets a wallet's base node peer if a valid RPC
connection to that peer cannot be obtained, but only for the wallet's base
node monitor service. The spinoff is that all subsequent RPC connections to
the same peer/node will benefit from the comms connection service retrying
to re-establish the peer connection. As the wallet's base node monitor
service runs in a slow loop, it will assist other services that are
dependent on the same base node peer connection; if one service suffers
due to a bad/stale peer connection all services suffer.
- Also implementing sending transaction to all connected peers as per #3239
- Fixed the base_node_service_config not being initialized with values
from the config file.
---
 base_layer/wallet/src/base_node_service/monitor.rs | 14 ++++++++++++++
 .../wallet/src/connectivity_service/handle.rs      |  8 ++++++++
 .../wallet/src/connectivity_service/service.rs     | 10 ++++++++++
 3 files changed, 32 insertions(+)

diff --git a/base_layer/wallet/src/base_node_service/monitor.rs b/base_layer/wallet/src/base_node_service/monitor.rs
index 8e0298ca27..4b31f49f8a 100644
--- a/base_layer/wallet/src/base_node_service/monitor.rs
+++ b/base_layer/wallet/src/base_node_service/monitor.rs
@@ -78,6 +78,13 @@ impl<T: WalletBackend + 'static> BaseNodeMonitor<T> {
                 },
                 Err(e @ BaseNodeMonitorError::RpcFailed(_)) => {
                     warn!(target: LOG_TARGET, "Connectivity failure to base node: {}", e);
+                    debug!(target: LOG_TARGET, "Reconnect current base node peer...");
+                    // Disconnecting and re-setting the base node peer would revive a stale peer connection - most
+                    // the client can do. As the wallet's base node monitor service runs in a slow loop, it will
+                    // assist other services that are dependent on the same base node peer connection; if one service
+                    // suffers due to a bad/stale peer connection all services suffer.
+                    self.reconnect_current_base_node_peer().await;
+                    time::sleep(self.interval).await;
                     continue;
                 },
                 Err(e @ BaseNodeMonitorError::InvalidBaseNodeResponse(_)) |
@@ -93,6 +100,13 @@ impl<T: WalletBackend + 'static> BaseNodeMonitor<T> {
         );
     }
 
+    async fn reconnect_current_base_node_peer(&mut self) {
+        if let Some(peer) = self.wallet_connectivity.get_current_base_node_peer() {
+            self.wallet_connectivity.disconnect_base_node(peer.clone()).await;
+            let _ = self.wallet_connectivity.set_base_node(peer).await;
+        };
+    }
+
     async fn monitor_node(&mut self) -> Result<(), BaseNodeMonitorError> {
         loop {
             let mut client = self
diff --git a/base_layer/wallet/src/connectivity_service/handle.rs b/base_layer/wallet/src/connectivity_service/handle.rs
index 5a35696e14..3b92d5082d 100644
--- a/base_layer/wallet/src/connectivity_service/handle.rs
+++ b/base_layer/wallet/src/connectivity_service/handle.rs
@@ -32,6 +32,7 @@ use tokio::sync::{mpsc, oneshot, watch};
 pub enum WalletConnectivityRequest {
     ObtainBaseNodeWalletRpcClient(oneshot::Sender<RpcClientLease<BaseNodeWalletRpcClient>>),
     ObtainBaseNodeSyncRpcClient(oneshot::Sender<RpcClientLease<BaseNodeSyncRpcClient>>),
+    DisconnectBaseNode(Box<Peer>),
 }
 
 #[derive(Clone)]
@@ -59,6 +60,13 @@ impl WalletConnectivityHandle {
         Ok(())
     }
 
+    pub async fn disconnect_base_node(&mut self, base_node_peer: Peer) {
+        let _ = self
+            .sender
+            .send(WalletConnectivityRequest::DisconnectBaseNode(Box::new(base_node_peer)))
+            .await;
+    }
+
     /// Obtain a BaseNodeWalletRpcClient.
     ///
     /// This can be relied on to obtain a pooled BaseNodeWalletRpcClient rpc session from a currently selected base
diff --git a/base_layer/wallet/src/connectivity_service/service.rs b/base_layer/wallet/src/connectivity_service/service.rs
index 950b9a9a72..8cf881aff9 100644
--- a/base_layer/wallet/src/connectivity_service/service.rs
+++ b/base_layer/wallet/src/connectivity_service/service.rs
@@ -129,6 +129,10 @@ impl WalletConnectivityService {
             ObtainBaseNodeSyncRpcClient(reply) => {
                 self.handle_pool_request(reply.into()).await;
             },
+
+            DisconnectBaseNode(peer) => {
+                self.disconnect_base_node(*peer).await;
+            },
         }
     }
 
@@ -204,6 +208,12 @@ impl WalletConnectivityService {
         self.base_node_watch.borrow().as_ref().map(|p| p.node_id.clone())
     }
 
+    async fn disconnect_base_node(&mut self, peer: Peer) {
+        if let Ok(Some(connection)) = self.connectivity.get_connection(peer.node_id).await {
+            if connection.clone().disconnect().await.is_ok() {}
+        };
+    }
+
     async fn setup_base_node_connection(&mut self) {
         self.pools = None;
         loop {