channel: add connectivity online offline events

Author: jjyr

crates/fiber-lib/src/fiber/channel.rs

@@ -3902,6 +3902,28 @@ impl ChannelActorState {
         }
     }
 
+    fn notify_channel_connectivity(&self, connectivity_state: ChannelConnectivityState) {
+        let Some(channel_outpoint) = self.get_funding_transaction_outpoint() else {
+            return;
+        };
+        let event = match connectivity_state {
+            ChannelConnectivityState::Online => NetworkServiceEvent::ChannelOnline(
+                self.get_remote_pubkey(),
+                self.get_id(),
+                channel_outpoint,
+            ),
+            ChannelConnectivityState::Offline => NetworkServiceEvent::ChannelOffline(
+                self.get_remote_pubkey(),
+                self.get_id(),
+                channel_outpoint,
+            ),
+            ChannelConnectivityState::Syncing => return,
+        };
+        self.network()
+            .send_message(NetworkActorMessage::new_notification(event))
+            .expect(ASSUME_NETWORK_ACTOR_ALIVE);
+    }
+
     pub(crate) fn mark_reestablishing_offline(&mut self) {
         self.clear_waiting_peer_response();
         self.reestablishing = true;
@@ -3913,6 +3935,7 @@ impl ChannelActorState {
 
     fn on_peer_disconnected(&mut self) {
         self.mark_reestablishing_offline();
+        self.notify_channel_connectivity(ChannelConnectivityState::Offline);
         if let Some(outpoint) = self.get_funding_transaction_outpoint() {
             self.network()
                 .send_message(NetworkActorMessage::new_event(
@@ -6048,6 +6071,7 @@ impl ChannelActorState {
         self.increment_local_commitment_number();
         self.increment_remote_commitment_number();
         self.connectivity_state = ChannelConnectivityState::Online;
+        self.notify_channel_connectivity(ChannelConnectivityState::Online);
         let pubkey = self.get_remote_pubkey();
         self.on_owned_channel_updated(myself, false);
         self.network()
@@ -6068,6 +6092,7 @@ impl ChannelActorState {
 
         self.reestablishing = false;
         self.connectivity_state = ChannelConnectivityState::Online;
+        self.notify_channel_connectivity(ChannelConnectivityState::Online);
 
         // If the channel is already ready, we should notify the network actor.
         // so that we update the network.outpoint_channel_map

crates/fiber-lib/src/fiber/network.rs

@@ -526,6 +526,10 @@ pub enum NetworkServiceEvent {
     // The channel is ready to use (with funding transaction confirmed
     // and both parties sent ChannelReady messages).
     ChannelReady(Pubkey, Hash256, OutPoint),
+    // The channel connectivity is online and normal operations may resume.
+    ChannelOnline(Pubkey, Hash256, OutPoint),
+    // The channel connectivity is offline and normal operations are paused.
+    ChannelOffline(Pubkey, Hash256, OutPoint),
     ChannelClosed(Pubkey, Hash256, Byte32),
     ChannelAbandon(Hash256),
     ChannelFundingAborted(Hash256),

crates/fiber-lib/src/fiber/tests/channel.rs

@@ -499,31 +499,69 @@ async fn do_test_owned_channel_saved_to_graph_on_reconnected(public: bool) {
         })
         .await;
 
-    tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
+    node1
+        .expect_event(|event| match event {
+            NetworkServiceEvent::ChannelOffline(pubkey, channel_id, _) => {
+                assert_eq!(pubkey, &node2.pubkey);
+                assert_eq!(channel_id, &_new_channel_id);
+                true
+            }
+            _ => false,
+        })
+        .await;
+    node2
+        .expect_event(|event| match event {
+            NetworkServiceEvent::ChannelOffline(pubkey, channel_id, _) => {
+                assert_eq!(pubkey, &node1.pubkey);
+                assert_eq!(channel_id, &_new_channel_id);
+                true
+            }
+            _ => false,
+        })
+        .await;
+
+    wait_until_async_timeout(|| async {
+        node1.get_network_graph_channels().await.is_empty()
+            && node2.get_network_graph_channels().await.is_empty()
+    })
+    .await;
+
     let node1_channels = node1.get_network_graph_channels().await;
     assert_eq!(node1_channels, vec![]);
     let node2_channels = node2.get_network_graph_channels().await;
     assert_eq!(node2_channels, vec![]);
 
-    node1.connect_to(&mut node2).await;
+    node1.connect_to_nonblocking(&node2).await;
 
     node1
-        .expect_debug_event("Reestablished channel in ChannelReady")
+        .expect_event(|event| match event {
+            NetworkServiceEvent::ChannelOnline(pubkey, channel_id, _) => {
+                assert_eq!(pubkey, &node2.pubkey);
+                assert_eq!(channel_id, &_new_channel_id);
+                true
+            }
+            _ => false,
+        })
         .await;
     node2
-        .expect_debug_event("Reestablished channel in ChannelReady")
+        .expect_event(|event| match event {
+            NetworkServiceEvent::ChannelOnline(pubkey, channel_id, _) => {
+                assert_eq!(pubkey, &node1.pubkey);
+                assert_eq!(channel_id, &_new_channel_id);
+                true
+            }
+            _ => false,
+        })
         .await;
 
-    let mut node1_channels = vec![];
-    let mut node2_channels = vec![];
-    for _ in 0..100 {
-        node1_channels = node1.get_network_graph_channels().await;
-        node2_channels = node2.get_network_graph_channels().await;
-        if !node1_channels.is_empty() && !node2_channels.is_empty() {
-            break;
-        }
-        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-    }
+    wait_until_async_timeout(|| async {
+        !node1.get_network_graph_channels().await.is_empty()
+            && !node2.get_network_graph_channels().await.is_empty()
+    })
+    .await;
+
+    let node1_channels = node1.get_network_graph_channels().await;
+    let node2_channels = node2.get_network_graph_channels().await;
     assert_ne!(node1_channels, vec![]);
     assert_ne!(node2_channels, vec![]);
 }