darkfid/task: use network disconnect subscription

bin/darkfid/src/task/consensus.rs

@@ -144,7 +144,7 @@ pub async fn consensus_init_task(
 
         match result {
             Ok(_) => return Ok(()),
-            Err(Error::NetworkOperationFailed) => {
+            Err(Error::NetworkNotConnected) => {
                 // Sync node again
                 *node.validator.synced.write().await = false;
                 if !config.skip_sync {
@@ -162,30 +162,27 @@ pub async fn consensus_init_task(
 async fn replicator_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     // Grab proposals subscriber and subscribe to it
     let proposals_sub = node.subscribers.get("proposals").unwrap();
-    let subscription = proposals_sub.publisher.clone().subscribe().await;
+    let prop_subscription = proposals_sub.publisher.clone().subscribe().await;
 
-    smol::future::or(
-        monitor_network(node.clone(), &subscription),
-        consensus_task(node, &subscription, ex),
+    // Subscribe to the network disconnect subscriber
+    let net_subscription = node.p2p.hosts().subscribe_disconnect().await;
+
+    let result = smol::future::or(
+        monitor_network(&net_subscription),
+        consensus_task(node, &prop_subscription, ex),
     )
-    .await
+    .await;
+
+    // Terminate the subscriptions
+    prop_subscription.unsubscribe().await;
+    net_subscription.unsubscribe().await;
+
+    result
 }
 
 /// Async task to monitor network disconnections.
-async fn monitor_network(
-    node: Arc<Darkfid>,
-    subscription: &Subscription<JsonNotification>,
-) -> Result<()> {
-    loop {
-        // Check if we are connected to the network
-        if node.p2p.hosts().channels().await.is_empty() {
-            error!(target: "darkfid::task::consensus::monitor_network", "Node disconnected from the network");
-            subscription.unsubscribe().await;
-            return Err(Error::NetworkOperationFailed)
-        }
-
-        sleep(node.p2p.settings().outbound_connect_timeout).await;
-    }
+async fn monitor_network(subscription: &Subscription<Error>) -> Result<()> {
+    Err(subscription.receive().await)
 }
 
 /// Async task used for listening for new blocks and perform consensus.

bin/darkfid/src/task/miner.rs

@@ -175,10 +175,10 @@ pub async fn miner_task(
         .await
         {
             Ok(_) => { /* Do nothing */ }
-            Err(Error::NetworkOperationFailed) => {
+            Err(Error::NetworkNotConnected) => {
                 error!(target: "darkfid::task::miner_task", "Node disconnected from the network");
                 subscription.unsubscribe().await;
-                return Err(Error::NetworkOperationFailed)
+                return Err(Error::NetworkNotConnected)
             }
             Err(e) => {
                 error!(
@@ -332,7 +332,7 @@ async fn mine_next_block(
 
     // Check if we are connected to the network
     if !skip_sync && node.p2p.hosts().channels().await.is_empty() {
-        return Err(Error::NetworkOperationFailed)
+        return Err(Error::NetworkNotConnected)
     }
 
     // Append the mined block as a proposal

bin/darkfid/src/task/sync.rs

@@ -200,6 +200,11 @@ async fn synced_peers(
         }
 
         warn!(target: "darkfid::task::sync::synced_peers", "Node is not connected to other nodes, waiting to retry...");
+        let subscription = node.p2p.hosts().subscribe_channel().await;
+        let _ = subscription.receive().await;
+        subscription.unsubscribe().await;
+
+        info!(target: "darkfid::task::sync::synced_peers", "Sleeping a bit to allow for more nodes to connect...");
         sleep(node.p2p.settings().outbound_connect_timeout).await;
     }