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

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]