fix(comms): spawn liveness check after address is final (#4919)

Description
---
- Spawns the liveness check after the public address has been finalised

Motivation and Context
---
Fixes #4915 and fixes an edge case where the old address is used for liveness checks if the user deletes their tor identity before starting the node.

How Has This Been Tested?
---
Manually

applications/tari_base_node/src/commands/command/status.rs

@@ -131,7 +131,7 @@ impl CommandContext {
             ),
         );
 
-        match self.comms.listening_info().liveness_status() {
+        match self.comms.liveness_status() {
             LivenessStatus::Disabled => {},
             LivenessStatus::Checking => {
                 status_line.add("⏳️️");

comms/core/src/builder/comms_node.rs

@@ -26,7 +26,7 @@ use log::*;
 use tari_shutdown::ShutdownSignal;
 use tokio::{
     io::{AsyncRead, AsyncWrite},
-    sync::{broadcast, mpsc},
+    sync::{broadcast, mpsc, watch},
 };
 
 use super::{CommsBuilderError, CommsShutdown};
@@ -37,6 +37,8 @@ use crate::{
         ConnectionManagerRequest,
         ConnectionManagerRequester,
         ListenerInfo,
+        LivenessCheck,
+        LivenessStatus,
     },
     connectivity::{ConnectivityEventRx, ConnectivityManager, ConnectivityRequest, ConnectivityRequester},
     multiaddr::Multiaddr,
@@ -132,6 +134,7 @@ impl UnspawnedCommsNode {
     }
 
     /// Spawn a new node using the specified [Transport](crate::transports::Transport).
+    #[allow(clippy::too_many_lines)]
     pub async fn spawn_with_transport<TTransport>(self, transport: TTransport) -> Result<CommsNode, CommsBuilderError>
     where
         TTransport: Transport + Unpin + Send + Sync + Clone + 'static,
@@ -187,8 +190,8 @@ impl UnspawnedCommsNode {
         let noise_config = NoiseConfig::new(node_identity.clone());
 
         let mut connection_manager = ConnectionManager::new(
-            connection_manager_config,
-            transport,
+            connection_manager_config.clone(),
+            transport.clone(),
             noise_config,
             dial_backoff,
             connection_manager_request_rx,
@@ -235,13 +238,27 @@ impl UnspawnedCommsNode {
             node_identity.public_address()
         );
 
+        // Spawn liveness check now that we have the final address
+        let liveness_watch = connection_manager_config
+            .liveness_self_check_interval
+            .map(|interval| {
+                LivenessCheck::spawn(
+                    transport,
+                    node_identity.public_address(),
+                    interval,
+                    shutdown_signal.clone(),
+                )
+            })
+            .unwrap_or_else(|| watch::channel(LivenessStatus::Disabled).1);
+
         Ok(CommsNode {
             shutdown_signal,
             connection_manager_requester,
             connectivity_requester,
             listening_info,
             node_identity,
             peer_manager,
+            liveness_watch,
             hidden_service,
             complete_signals: ext_context.drain_complete_signals(),
         })
@@ -286,6 +303,8 @@ pub struct CommsNode {
     peer_manager: Arc<PeerManager>,
     /// The bind addresses of the listener(s)
     listening_info: ListenerInfo,
+    /// Current liveness status
+    liveness_watch: watch::Receiver<LivenessStatus>,
     /// `Some` if the comms node is configured to run via a hidden service, otherwise `None`
     hidden_service: Option<tor::HiddenService>,
     /// The 'reciprocal' shutdown signals for each comms service
@@ -328,6 +347,11 @@ impl CommsNode {
         &self.listening_info
     }
 
+    /// Returns the current liveness status
+    pub fn liveness_status(&self) -> LivenessStatus {
+        *self.liveness_watch.borrow()
+    }
+
     /// Return the Ip/Tcp address that this node is listening on
     pub fn hidden_service(&self) -> Option<&tor::HiddenService> {
         self.hidden_service.as_ref()

comms/core/src/connection_manager/listener.rs

@@ -36,7 +36,7 @@ use log::*;
 use tari_shutdown::{oneshot_trigger, oneshot_trigger::OneshotTrigger, ShutdownSignal};
 use tokio::{
     io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
-    sync::{mpsc, watch},
+    sync::mpsc,
     time,
 };
 use tokio_stream::StreamExt;
@@ -53,7 +53,7 @@ use super::{
 use crate::{
     bounded_executor::BoundedExecutor,
     connection_manager::{
-        liveness::{LivenessCheck, LivenessSession, LivenessStatus},
+        liveness::LivenessSession,
         metrics,
         wire_mode::{WireMode, LIVENESS_WIRE_MODE},
     },
@@ -83,7 +83,7 @@ pub struct PeerListener<TTransport> {
     node_identity: Arc<NodeIdentity>,
     our_supported_protocols: Vec<ProtocolId>,
     liveness_session_count: Arc<AtomicUsize>,
-    on_listening: OneshotTrigger<Result<(Multiaddr, watch::Receiver<LivenessStatus>), ConnectionManagerError>>,
+    on_listening: OneshotTrigger<Result<Multiaddr, ConnectionManagerError>>,
 }
 
 impl<TTransport> PeerListener<TTransport>
@@ -121,10 +121,7 @@ where
     /// in binding the listener socket
     // This returns an impl Future and is not async because we want to exclude &self from the future so that it has a
     // 'static lifetime as well as to flatten the oneshot result for ergonomics
-    pub fn on_listening(
-        &self,
-    ) -> impl Future<Output = Result<(Multiaddr, watch::Receiver<LivenessStatus>), ConnectionManagerError>> + 'static
-    {
+    pub fn on_listening(&self) -> impl Future<Output = Result<Multiaddr, ConnectionManagerError>> + 'static {
         let signal = self.on_listening.to_signal();
         signal.map(|r| r.ok_or(ConnectionManagerError::ListenerOneshotCancelled)?)
     }
@@ -135,7 +132,7 @@ where
         self
     }
 
-    pub async fn listen(self) -> Result<(Multiaddr, watch::Receiver<LivenessStatus>), ConnectionManagerError> {
+    pub async fn listen(self) -> Result<Multiaddr, ConnectionManagerError> {
         let on_listening = self.on_listening();
         runtime::current().spawn(self.run());
         on_listening.await
@@ -148,9 +145,7 @@ where
             Ok((mut inbound, address)) => {
                 info!(target: LOG_TARGET, "Listening for peer connections on '{}'", address);
 
-                let liveness_watch = self.spawn_liveness_check();
-
-                self.on_listening.broadcast(Ok((address, liveness_watch)));
+                self.on_listening.broadcast(Ok(address));
 
                 loop {
                     tokio::select! {
@@ -234,21 +229,6 @@ where
         });
     }
 
-    fn spawn_liveness_check(&self) -> watch::Receiver<LivenessStatus> {
-        match self.config.liveness_self_check_interval {
-            Some(interval) => LivenessCheck::spawn(
-                self.transport.clone(),
-                self.node_identity.public_address(),
-                interval,
-                self.shutdown_signal.clone(),
-            ),
-            None => {
-                let (_, rx) = watch::channel(LivenessStatus::Disabled);
-                rx
-            },
-        }
-    }
-
     async fn spawn_listen_task(&self, mut socket: TTransport::Output, peer_addr: Multiaddr) {
         let node_identity = self.node_identity.clone();
         let peer_manager = self.peer_manager.clone();

comms/core/src/connection_manager/manager.rs

@@ -28,7 +28,7 @@ use tari_shutdown::{Shutdown, ShutdownSignal};
 use time::Duration;
 use tokio::{
     io::{AsyncRead, AsyncWrite},
-    sync::{broadcast, mpsc, oneshot, watch},
+    sync::{broadcast, mpsc, oneshot},
     task,
     time,
 };
@@ -43,7 +43,7 @@ use super::{
 };
 use crate::{
     backoff::Backoff,
-    connection_manager::{liveness::LivenessStatus, metrics, ConnectionDirection, ConnectionId},
+    connection_manager::{metrics, ConnectionDirection, ConnectionId},
     multiplexing::Substream,
     noise::NoiseConfig,
     peer_manager::{NodeId, NodeIdentity, PeerManagerError},
@@ -149,7 +149,6 @@ impl Default for ConnectionManagerConfig {
 pub struct ListenerInfo {
     bind_address: Multiaddr,
     aux_bind_address: Option<Multiaddr>,
-    liveness_watch: watch::Receiver<LivenessStatus>,
 }
 
 impl ListenerInfo {
@@ -163,17 +162,6 @@ impl ListenerInfo {
     pub fn auxiliary_bind_address(&self) -> Option<&Multiaddr> {
         self.aux_bind_address.as_ref()
     }
-
-    /// Returns the current liveness status
-    pub fn liveness_status(&self) -> LivenessStatus {
-        *self.liveness_watch.borrow()
-    }
-
-    /// Waits for liveness status to change from the last time the value was checked.
-    pub async fn liveness_status_changed(&mut self) -> Option<LivenessStatus> {
-        self.liveness_watch.changed().await.ok()?;
-        Some(*self.liveness_watch.borrow())
-    }
 }
 
 /// The actor responsible for connection management.
@@ -346,17 +334,16 @@ where
         listener.set_supported_protocols(self.protocols.get_supported_protocols());
 
         let mut listener_info = match listener.listen().await {
-            Ok((bind_address, liveness_watch)) => ListenerInfo {
+            Ok(bind_address) => ListenerInfo {
                 bind_address,
                 aux_bind_address: None,
-                liveness_watch,
             },
             Err(err) => return Err(err),
         };
 
         if let Some(mut listener) = self.aux_listener.take() {
             listener.set_supported_protocols(self.protocols.get_supported_protocols());
-            let (addr, _) = listener.listen().await?;
+            let addr = listener.listen().await?;
             debug!(target: LOG_TARGET, "Aux TCP listener bound to address {}", addr);
             listener_info.aux_bind_address = Some(addr);
         }

comms/core/src/connection_manager/mod.rs

@@ -52,6 +52,7 @@ mod peer_connection;
 pub use peer_connection::{ConnectionId, NegotiatedSubstream, PeerConnection, PeerConnectionRequest};
 
 mod liveness;
+pub(crate) use liveness::LivenessCheck;
 pub use liveness::LivenessStatus;
 
 mod wire_mode;

comms/core/src/connection_manager/tests/listener_dialer.rs

@@ -66,7 +66,7 @@ async fn listen() -> Result<(), Box<dyn Error>> {
         shutdown.to_signal(),
     );
 
-    let (mut bind_addr, _) = listener.listen().await?;
+    let mut bind_addr = listener.listen().await?;
 
     unpack_enum!(Protocol::Memory(port) = bind_addr.pop().unwrap());
     assert!(port > 0);
@@ -103,7 +103,7 @@ async fn smoke() {
     listener.set_supported_protocols(supported_protocols.clone());
 
     // Get the listening address of the peer
-    let (address, _) = listener.listen().await.unwrap();
+    let address = listener.listen().await.unwrap();
 
     let node_identity2 = build_node_identity(PeerFeatures::COMMUNICATION_NODE);
     let noise_config2 = NoiseConfig::new(node_identity2.clone());
@@ -207,7 +207,7 @@ async fn banned() {
     listener.set_supported_protocols(supported_protocols.clone());
 
     // Get the listener address of the peer
-    let (address, _) = listener.listen().await.unwrap();
+    let address = listener.listen().await.unwrap();
 
     let node_identity2 = build_node_identity(PeerFeatures::COMMUNICATION_NODE);
     // The listener has banned the dialer peer