fix(comms): fixes edge case where online status event does not get pu…

…blished (#4756)

Description
---
Ensures that connectivity status events are always published if a status change has occurred.

Motivation and Context
---
An online/degraded event should be emitted if any connections are present, including if only connected to walllets.
Some edge cases where connectivity status event was not emitted. These events are typically only used for logging and tests, but could be used in an application to wait for conenctivity to become available.

How Has This Been Tested?
---
Added to existing unit test, manually with DAN node

comms/core/src/connectivity/manager.rs

@@ -480,7 +480,7 @@ impl ConnectivityManagerActor {
         entry.failed_attempts()
     }
 
-    async fn handle_peer_connection_failure(&mut self, node_id: &NodeId) -> Result<(), ConnectivityError> {
+    async fn on_peer_connection_failure(&mut self, node_id: &NodeId) -> Result<(), ConnectivityError> {
         if self.status.is_offline() {
             debug!(
                 target: LOG_TARGET,
@@ -532,12 +532,22 @@ impl ConnectivityManagerActor {
     async fn handle_connection_manager_event(
         &mut self,
         event: &ConnectionManagerEvent,
+    ) -> Result<(), ConnectivityError> {
+        self.update_state_on_connectivity_event(event).await?;
+        self.update_connectivity_status();
+        self.update_connectivity_metrics();
+        Ok(())
+    }
+
+    async fn update_state_on_connectivity_event(
+        &mut self,
+        event: &ConnectionManagerEvent,
     ) -> Result<(), ConnectivityError> {
         use ConnectionManagerEvent::{PeerConnectFailed, PeerConnected, PeerDisconnected};
         debug!(target: LOG_TARGET, "Received event: {}", event);
         match event {
             PeerConnected(new_conn) => {
-                match self.handle_new_connection_tie_break(new_conn).await {
+                match self.on_new_connection(new_conn).await {
                     TieBreak::KeepExisting => {
                         // Ignore event, we discarded the new connection and keeping the current one
                         return Ok(());
@@ -572,6 +582,7 @@ impl ConnectivityManagerActor {
                             target: LOG_TARGET,
                             "Ignoring DialCancelled({}) event because an inbound connection already exists", node_id
                         );
+
                         return Ok(());
                     }
                 }
@@ -586,7 +597,7 @@ impl ConnectivityManagerActor {
                     target: LOG_TARGET,
                     "Connection to peer '{}' failed because '{:?}'", node_id, err
                 );
-                self.handle_peer_connection_failure(node_id).await?;
+                self.on_peer_connection_failure(node_id).await?;
                 (&*node_id, ConnectionStatus::Failed, None)
             },
             _ => return Ok(()),
@@ -635,12 +646,10 @@ impl ConnectivityManagerActor {
             },
         }
 
-        self.update_connectivity_status();
-        self.update_connectivity_metrics();
         Ok(())
     }
 
-    async fn handle_new_connection_tie_break(&mut self, new_conn: &PeerConnection) -> TieBreak {
+    async fn on_new_connection(&mut self, new_conn: &PeerConnection) -> TieBreak {
         match self.pool.get_connection(new_conn.peer_node_id()).cloned() {
             Some(existing_conn) if !existing_conn.is_connected() => {
                 debug!(
@@ -749,6 +758,8 @@ impl ConnectivityManagerActor {
             n if n == 0 => {
                 if num_connected_clients == 0 {
                     self.transition(ConnectivityStatus::Offline, min_peers);
+                } else {
+                    self.transition(ConnectivityStatus::Degraded(n), min_peers);
                 }
             },
             _ => unreachable!("num_connected is unsigned and only negative pattern covered on this branch"),

comms/core/src/connectivity/test.rs

@@ -1,24 +1,25 @@
 //  Copyright 2020, The Tari Project
 //
-//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
-//  following conditions are met:
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that
+// the  following conditions are met:
 //
-//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
-//  disclaimer.
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the
+// following  disclaimer.
 //
 //  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
 //  following disclaimer in the documentation and/or other materials provided with the distribution.
 //
 //  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
 //  products derived from this software without specific prior written permission.
 //
-//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
-//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
-//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
-//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
-//  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.
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED
+// WARRANTIES,  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+// PARTICULAR PURPOSE ARE  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
+// DIRECT, INDIRECT, INCIDENTAL,  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+// CAUSED AND ON ANY THEORY OF LIABILITY,  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 std::{sync::Arc, time::Duration};
 
 use futures::{future, StreamExt};
@@ -137,8 +138,9 @@ async fn connecting_peers() {
     }
 }
 
+#[allow(clippy::too_many_lines)]
 #[runtime::test]
-async fn online_then_offline() {
+async fn online_then_offline_then_online() {
     let (mut connectivity, mut event_stream, node_identity, peer_manager, cm_mock_state, _shutdown) =
         setup_connectivity_manager(ConnectivityConfig {
             min_connectivity: 2,
@@ -239,6 +241,32 @@ async fn online_then_offline() {
 
     let is_offline = connectivity.get_connectivity_status().await.unwrap().is_offline();
     assert!(is_offline);
+
+    // Create a fresh set of connections since the previous connections are now in a disconnected state
+    let connections = future::join_all(
+        (0..5)
+            .map(|i| peers[i].clone())
+            .map(|peer| create_peer_connection_mock_pair(node_identity.to_peer(), peer)),
+    )
+    .await
+    .into_iter()
+    .map(|(conn, _, _, _)| conn)
+    .collect::<Vec<_>>();
+    for conn in connections.iter().skip(1) {
+        cm_mock_state.publish_event(ConnectionManagerEvent::PeerConnected(conn.clone()));
+    }
+
+    streams::assert_in_broadcast(
+        &mut event_stream,
+        |item| match item {
+            ConnectivityEvent::ConnectivityStateOnline(2) => Some(()),
+            _ => None,
+        },
+        Duration::from_secs(10),
+    )
+    .await;
+
+    assert!(connectivity.get_connectivity_status().await.unwrap().is_online());
 }
 
 #[runtime::test]