darkfid: minor cleanup

bin/darkfid/src/rpc_tx.rs

@@ -134,7 +134,7 @@ impl Darkfid {
         };
 
         self.p2p.broadcast(&tx).await;
-        if self.p2p.hosts().channels().is_empty() {
+        if !self.p2p.is_connected() {
             warn!(target: "darkfid::rpc::tx_broadcast", "No connected channels to broadcast tx");
         }
 

bin/darkfid/src/task/consensus.rs

@@ -220,27 +220,31 @@ async fn consensus_task(
                 continue
             }
         };
-        if !finalized.is_empty() {
-            let mut notif_blocks = Vec::with_capacity(finalized.len());
-            for block in finalized {
-                notif_blocks
-                    .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
-            }
-            block_sub.notify(JsonValue::Array(notif_blocks)).await;
-
-            // Invoke the detached garbage collection task
-            gc_task.clone().stop().await;
-            gc_task.clone().start(
-                garbage_collect_task(node.clone()),
-                |res| async {
-                    match res {
-                        Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
-                        Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
-                    }
-                },
-                Error::GarbageCollectionTaskStopped,
-                ex.clone(),
-            );
+
+        if finalized.is_empty() {
+            continue
         }
+
+        let mut notif_blocks = Vec::with_capacity(finalized.len());
+        for block in finalized {
+            notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+        }
+        block_sub.notify(JsonValue::Array(notif_blocks)).await;
+
+        // Invoke the detached garbage collection task
+        gc_task.clone().stop().await;
+        gc_task.clone().start(
+            garbage_collect_task(node.clone()),
+            |res| async {
+                match res {
+                    Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!(target: "darkfid", "Failed starting garbage collection task: {}", e)
+                    }
+                }
+            },
+            Error::GarbageCollectionTaskStopped,
+            ex.clone(),
+        );
     }
 }

bin/darkfid/src/task/miner.rs

@@ -106,15 +106,18 @@ pub async fn miner_task(
 
             // Check if we can finalize anything and broadcast them
             let finalized = node.validator.finalization().await?;
-            if !finalized.is_empty() {
-                let mut notif_blocks = Vec::with_capacity(finalized.len());
-                for block in finalized {
-                    notif_blocks
-                        .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
-                }
-                block_sub.notify(JsonValue::Array(notif_blocks)).await;
-                break;
+
+            if finalized.is_empty() {
+                continue
             }
+
+            let mut notif_blocks = Vec::with_capacity(finalized.len());
+            for block in finalized {
+                notif_blocks
+                    .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+            }
+            block_sub.notify(JsonValue::Array(notif_blocks)).await;
+            break;
         }
     }
 
@@ -200,28 +203,32 @@ pub async fn miner_task(
                 continue
             }
         };
-        if !finalized.is_empty() {
-            let mut notif_blocks = Vec::with_capacity(finalized.len());
-            for block in finalized {
-                notif_blocks
-                    .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
-            }
-            block_sub.notify(JsonValue::Array(notif_blocks)).await;
 
-            // Invoke the detached garbage collection task
-            gc_task.clone().stop().await;
-            gc_task.clone().start(
-                garbage_collect_task(node.clone()),
-                |res| async {
-                    match res {
-                        Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
-                        Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
-                    }
-                },
-                Error::GarbageCollectionTaskStopped,
-                ex.clone(),
-            );
+        if finalized.is_empty() {
+            continue
+        }
+
+        let mut notif_blocks = Vec::with_capacity(finalized.len());
+        for block in finalized {
+            notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
         }
+        block_sub.notify(JsonValue::Array(notif_blocks)).await;
+
+        // Invoke the detached garbage collection task
+        gc_task.clone().stop().await;
+        gc_task.clone().start(
+            garbage_collect_task(node.clone()),
+            |res| async {
+                match res {
+                    Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!(target: "darkfid", "Failed starting garbage collection task: {}", e)
+                    }
+                }
+            },
+            Error::GarbageCollectionTaskStopped,
+            ex.clone(),
+        );
     }
 }
 
@@ -331,7 +338,7 @@ async fn mine_next_block(
     extended_fork.module.verify_current_block(&next_block)?;
 
     // Check if we are connected to the network
-    if !skip_sync && node.p2p.hosts().channels().is_empty() {
+    if !skip_sync && !node.p2p.is_connected() {
         return Err(Error::NetworkNotConnected)
     }
 

bin/darkfid/src/task/sync.rs

@@ -144,37 +144,15 @@ async fn synced_peers(
         // Grab channels
         let peers = node.p2p.hosts().channels();
 
-        // Check anyone is connected
-        if !peers.is_empty() {
-            // Ask each peer if they are synced
-            for peer in peers {
-                // If a checkpoint was provider, we check that the peer follows that sequence
-                if let Some(c) = checkpoint {
-                    // Communication setup
-                    let response_sub = peer.subscribe_msg::<HeaderSyncResponse>().await?;
-
-                    // Node creates a `HeaderSyncRequest` and sends it
-                    let request = HeaderSyncRequest { height: c.0 + 1 };
-                    peer.send(&request).await?;
-
-                    // Node waits for response
-                    let Ok(response) = response_sub.receive_with_timeout(comms_timeout).await
-                    else {
-                        continue
-                    };
-
-                    // Handle response
-                    if response.headers.is_empty() || response.headers.last().unwrap().hash() != c.1
-                    {
-                        continue
-                    }
-                }
-
+        // Ask each peer(if we got any) if they are synced
+        for peer in peers {
+            // If a checkpoint was provider, we check that the peer follows that sequence
+            if let Some(c) = checkpoint {
                 // Communication setup
-                let response_sub = peer.subscribe_msg::<TipResponse>().await?;
+                let response_sub = peer.subscribe_msg::<HeaderSyncResponse>().await?;
 
-                // Node creates a `TipRequest` and sends it
-                let request = TipRequest { tip: *last_tip };
+                // Node creates a `HeaderSyncRequest` and sends it
+                let request = HeaderSyncRequest { height: c.0 + 1 };
                 peer.send(&request).await?;
 
                 // Node waits for response
@@ -183,15 +161,32 @@ async fn synced_peers(
                 };
 
                 // Handle response
-                if response.synced && response.height.is_some() && response.hash.is_some() {
-                    let tip = (response.height.unwrap(), *response.hash.unwrap().inner());
-                    let Some(tip_peers) = tips.get_mut(&tip) else {
-                        tips.insert(tip, vec![peer.clone()]);
-                        continue
-                    };
-                    tip_peers.push(peer.clone());
+                if response.headers.is_empty() || response.headers.last().unwrap().hash() != c.1 {
+                    continue
                 }
             }
+
+            // Communication setup
+            let response_sub = peer.subscribe_msg::<TipResponse>().await?;
+
+            // Node creates a `TipRequest` and sends it
+            let request = TipRequest { tip: *last_tip };
+            peer.send(&request).await?;
+
+            // Node waits for response
+            let Ok(response) = response_sub.receive_with_timeout(comms_timeout).await else {
+                continue
+            };
+
+            // Handle response
+            if response.synced && response.height.is_some() && response.hash.is_some() {
+                let tip = (response.height.unwrap(), *response.hash.unwrap().inner());
+                let Some(tip_peers) = tips.get_mut(&tip) else {
+                    tips.insert(tip, vec![peer.clone()]);
+                    continue
+                };
+                tip_peers.push(peer.clone());
+            }
         }
 
         // Check if we got any tips
@@ -199,7 +194,7 @@ async fn synced_peers(
             break
         }
 
-        warn!(target: "darkfid::task::sync::synced_peers", "Node is not connected to other nodes, waiting to retry...");
+        warn!(target: "darkfid::task::sync::synced_peers", "Node is not connected to other synced nodes, waiting to retry...");
         let subscription = node.p2p.hosts().subscribe_channel().await;
         let _ = subscription.receive().await;
         subscription.unsubscribe().await;