darkfid: created task to listen for appended proposals and perform finalization check for non mining nodes

bin/darkfid/src/main.rs

@@ -58,7 +58,7 @@ mod rpc_tx;
 
 /// Validator async tasks
 mod task;
-use task::{miner_task, sync_task};
+use task::{consensus_task, miner_task, sync_task};
 
 /// P2P net protocols
 mod proto;
@@ -327,8 +327,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     darkfid.validator.purge_pending_txs().await?;
 
     // Consensus protocol
+    info!(target: "darkfid", "Starting consensus protocol task");
     let consensus_task = if blockchain_config.miner {
-        info!(target: "darkfid", "Starting consensus protocol task");
         // Grab rewards recipient public key(address)
         if blockchain_config.recipient.is_none() {
             return Err(Error::ParseFailed("Recipient address missing"))
@@ -351,10 +351,24 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
             Error::MinerTaskStopped,
             ex.clone(),
         );
-        Some(task)
+
+        task
     } else {
-        info!(target: "darkfid", "Not participating in consensus");
-        None
+        let task = StoppableTask::new();
+        task.clone().start(
+            // Weird hack to prevent lifetimes hell
+            async move { consensus_task(&darkfid).await },
+            |res| async {
+                match res {
+                    Ok(()) | Err(Error::ConsensusTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "darkfid", "Failed starting consensus task: {}", e),
+                }
+            },
+            Error::ConsensusTaskStopped,
+            ex.clone(),
+        );
+
+        task
     };
 
     // Signal handling for graceful termination.
@@ -371,11 +385,11 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     if blockchain_config.miner {
         info!(target: "darkfid", "Stopping miners P2P network...");
         miners_p2p.unwrap().stop().await;
-
-        info!(target: "darkfid", "Stopping consensus task...");
-        consensus_task.unwrap().stop().await;
     }
 
+    info!(target: "darkfid", "Stopping consensus task...");
+    consensus_task.stop().await;
+
     info!(target: "darkfid", "Flushing sled database...");
     let flushed_bytes = sled_db.flush_async().await?;
     info!(target: "darkfid", "Flushed {} bytes", flushed_bytes);

bin/darkfid/src/task/consensus.rs

@@ -0,0 +1,52 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2024 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use darkfi::{rpc::util::JsonValue, Result};
+use darkfi_serial::serialize;
+use log::info;
+
+use crate::Darkfid;
+
+// TODO: handle all ? so the task don't stop on errors
+
+/// async task used for listening for new blocks and perform consensus
+pub async fn consensus_task(node: &Darkfid) -> Result<()> {
+    info!(target: "darkfid::task::consensus_task", "Starting consensus task...");
+
+    // Grab blocks subscriber
+    let block_sub = node.subscribers.get("blocks").unwrap();
+
+    // Grab proposals subscriber and subscribe to it
+    let proposals_sub = node.subscribers.get("proposals").unwrap();
+    let subscription = proposals_sub.sub.clone().subscribe().await;
+
+    loop {
+        subscription.receive().await;
+
+        // 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(bs58::encode(&serialize(&block)).into_string()));
+            }
+            block_sub.notify(JsonValue::Array(notif_blocks)).await;
+        }
+    }
+}

bin/darkfid/src/task/mod.rs

@@ -18,8 +18,11 @@
 
 // TODO: Handle ? with matches in these files. They should be robust.
 
-pub mod sync;
-pub use sync::sync_task;
+pub mod consensus;
+pub use consensus::consensus_task;
 
 pub mod miner;
 pub use miner::miner_task;
+
+pub mod sync;
+pub use sync::sync_task;

src/error.rs

@@ -307,6 +307,9 @@ pub enum Error {
     #[error("Proposal already exists")]
     ProposalAlreadyExists,
 
+    #[error("Consensus task stopped")]
+    ConsensusTaskStopped,
+
     #[error("Miner task stopped")]
     MinerTaskStopped,