darkfid: use proposals/consensus logic while syncing

Additionally, performed some cleanup on validator code and handled cleaning up fork mempool when finalizing a block

bin/darkfid/src/task/sync.rs

@@ -19,10 +19,11 @@
 use std::collections::HashMap;
 
 use darkfi::{
-    blockchain::HeaderHash, net::ChannelPtr, system::sleep, util::encoding::base64, Error, Result,
+    blockchain::HeaderHash, net::ChannelPtr, rpc::jsonrpc::JsonSubscriber, system::sleep,
+    util::encoding::base64, validator::consensus::Proposal, Error, Result,
 };
 use darkfi_serial::serialize_async;
-use log::{debug, error, info, warn};
+use log::{debug, info, warn};
 use rand::{prelude::SliceRandom, rngs::OsRng};
 use tinyjson::JsonValue;
 
@@ -40,27 +41,37 @@ use crate::{
 pub async fn sync_task(node: &Darkfid) -> Result<()> {
     info!(target: "darkfid::task::sync_task", "Starting blockchain sync...");
 
+    // Generate a new fork to be able to extend
+    info!(target: "darkfid::task::sync_task", "Generating new empty fork...");
+    node.validator.consensus.generate_empty_fork().await?;
+
+    // Grab blocks subscriber
+    let block_sub = node.subscribers.get("blocks").unwrap();
+
     // Grab synced peers
     let peers = synced_peers(node).await?;
 
     // TODO: Configure a checkpoint, filter peers that don't have that and start
     // syncing the sequence until that
 
-    // Grab last known block header
-    let mut last = last_header(node)?;
+    // Grab last known block header, including existing pending sync ones
+    let mut last = match node.validator.blockchain.headers.get_last_sync()? {
+        Some(last_sync) => (last_sync.height, last_sync.hash()),
+        None => node.validator.blockchain.last()?,
+    };
     info!(target: "darkfid::task::sync_task", "Last known block: {} - {}", last.0, last.1);
+
+    // Sync headers and blocks
     loop {
         // Grab the most common tip and the corresponding peers
         let (common_tip_height, common_tip_peers) = most_common_tip(&peers, &last.1).await?;
 
-        // Retrieve all the headers backawards until our last known one and verify them.
+        // Retrieve all the headers backwards until our last known one and verify them.
         // We use the next height, in order to also retrieve the peers tip header.
-        retrieve_headers(node, &common_tip_peers, last, common_tip_height + 1).await?;
+        retrieve_headers(node, &common_tip_peers, last.0, common_tip_height + 1).await?;
 
         // Retrieve all the blocks for those headers and apply them to canonical
-        retrieve_blocks(node, &peers).await?;
-
-        let last_received = last_header(node)?;
+        let last_received = retrieve_blocks(node, &peers, last, block_sub).await?;
         info!(target: "darkfid::task::sync_task", "Last received block: {} - {}", last_received.0, last_received.1);
 
         if last == last_received {
@@ -73,6 +84,17 @@ pub async fn sync_task(node: &Darkfid) -> Result<()> {
     // Sync best fork
     sync_best_fork(node, &peers, &last.1).await?;
 
+    // Perform finalization
+    let finalized = node.validator.finalization().await?;
+    if !finalized.is_empty() {
+        // Notify subscriber
+        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;
+    }
+
     *node.validator.synced.write().await = true;
     info!(target: "darkfid::task::sync_task", "Blockchain synced!");
     Ok(())
@@ -120,16 +142,6 @@ async fn synced_peers(node: &Darkfid) -> Result<Vec<ChannelPtr>> {
     Ok(peers)
 }
 
-/// Auxiliary function to retrieve last known block header, including existing pending sync ones.
-fn last_header(node: &Darkfid) -> Result<(u32, HeaderHash)> {
-    // First we check if we have pending sync headers
-    if let Some(last_sync) = node.validator.blockchain.headers.get_last_sync()? {
-        return Ok((last_sync.height, last_sync.hash()))
-    }
-    // Then we grab the last one from the actual canonical chain
-    node.validator.blockchain.last()
-}
-
 /// Auxiliary function to ask all peers for their current tip and find the most common one.
 async fn most_common_tip(
     peers: &[ChannelPtr],
@@ -155,34 +167,31 @@ async fn most_common_tip(
         tip_peers.push(peer.clone());
     }
 
-    // Grab the most common tip peers
-    let mut common_tips = vec![];
-    let mut common_tip_peers = vec![];
+    // Grab the most common highest tip peers
+    let mut common_tip = (0, [0u8; 32], vec![]);
     for (tip, peers) in tips {
-        if peers.len() < common_tip_peers.len() {
+        // Check if tip peers is less than the most common tip peers
+        if peers.len() < common_tip.2.len() {
             continue;
         }
-        if peers.len() == common_tip_peers.len() {
-            common_tips.push(tip);
+        // If peers are the same length, skip if tip height is less than
+        // the most common tip height.
+        if peers.len() == common_tip.2.len() || tip.0 < common_tip.0 {
             continue;
         }
-        common_tips = vec![tip];
-        common_tip_peers = peers;
-    }
-    if common_tips.len() > 1 {
-        error!(target: "darkfid::task::sync::most_common_tip", "Multiple common tips found: {:?}", common_tips);
-        return Err(Error::BlockchainSyncError)
+        // Keep the heighest tip with the most peers
+        common_tip = (tip.0, tip.1, peers);
     }
 
-    info!(target: "darkfid::task::sync::most_common_tip", "Received tip from peers: {} - {}", common_tips[0].0, HeaderHash::new(common_tips[0].1));
-    Ok((common_tips[0].0, common_tip_peers))
+    info!(target: "darkfid::task::sync::most_common_tip", "Received tip from peers: {} - {}", common_tip.0, HeaderHash::new(common_tip.1));
+    Ok((common_tip.0, common_tip.2))
 }
 
 /// Auxiliary function to retrieve headers backwards until our last known one and verify them.
 async fn retrieve_headers(
     node: &Darkfid,
     peers: &[ChannelPtr],
-    last_known: (u32, HeaderHash),
+    last_known: u32,
     tip_height: u32,
 ) -> Result<()> {
     info!(target: "darkfid::task::sync::retrieve_headers", "Retrieving missing headers from peers...");
@@ -193,7 +202,7 @@ async fn retrieve_headers(
     }
 
     // We subtract 1 since tip_height is increased by one
-    let total = tip_height - last_known.0 - 1;
+    let total = tip_height - last_known - 1;
     let mut last_tip_height = tip_height;
     'headers_loop: loop {
         for (index, peer) in peers.iter().enumerate() {
@@ -208,7 +217,7 @@ async fn retrieve_headers(
 
             // Retain only the headers after our last known
             let mut response_headers = response.headers.to_vec();
-            response_headers.retain(|h| h.height > last_known.0);
+            response_headers.retain(|h| h.height > last_known);
 
             if response_headers.is_empty() {
                 break 'headers_loop
@@ -233,9 +242,9 @@ async fn retrieve_headers(
     info!(target: "darkfid::task::sync::retrieve_headers", "Verifying headers sequence...");
     let mut verified_headers = 0;
     let total = node.validator.blockchain.headers.len_sync();
-    // First we verify the first `BATCH` sequence, using the last canonical known one as
-    // the first sync header previous.
-    let last_known = node.validator.blockchain.last()?;
+    // First we verify the first `BATCH` sequence, using the last known header
+    // as the first sync header previous.
+    let last_known = node.validator.consensus.best_fork_last_header().await?;
     let mut headers = node.validator.blockchain.headers.get_after_sync(0, BATCH)?;
     if headers[0].previous != last_known.1 || headers[0].height != last_known.0 + 1 {
         return Err(Error::BlockIsInvalid(headers[0].hash().as_string()))
@@ -276,14 +285,19 @@ async fn retrieve_headers(
 }
 
 /// Auxiliary function to retrieve blocks of provided headers and apply them to canonical.
-async fn retrieve_blocks(node: &Darkfid, peers: &[ChannelPtr]) -> Result<()> {
+async fn retrieve_blocks(
+    node: &Darkfid,
+    peers: &[ChannelPtr],
+    last_known: (u32, HeaderHash),
+    block_sub: &JsonSubscriber,
+) -> Result<(u32, HeaderHash)> {
     info!(target: "darkfid::task::sync::retrieve_blocks", "Retrieving missing blocks from peers...");
+    let mut last_received = last_known;
     // Communication setup
     let mut peer_subs = vec![];
     for peer in peers {
         peer_subs.push(peer.subscribe_msg::<SyncResponse>().await?);
     }
-    let notif_sub = node.subscribers.get("blocks").unwrap();
 
     let mut received_blocks = 0;
     let total = node.validator.blockchain.headers.len_sync();
@@ -306,27 +320,34 @@ async fn retrieve_blocks(node: &Darkfid, peers: &[ChannelPtr]) -> Result<()> {
 
             // Verify and store retrieved blocks
             debug!(target: "darkfid::task::sync::retrieve_blocks", "Processing received blocks");
-            node.validator.add_blocks(&response.blocks).await?;
+            received_blocks += response.blocks.len();
+            let mut synced_headers = Vec::with_capacity(response.blocks.len());
+            for block in &response.blocks {
+                node.validator.append_proposal(&Proposal::new(block.clone())).await?;
+                synced_headers.push(block.header.height);
+                last_received = (block.header.height, block.hash());
+            }
 
             // Remove synced headers
-            node.validator.blockchain.headers.remove_sync(
-                &response.blocks.iter().map(|b| b.header.height).collect::<Vec<u32>>(),
-            )?;
-
-            // Notify subscriber
-            for block in &response.blocks {
-                info!(target: "darkfid::task::sync::retrieve_blocks", "Appended block: {} - {}", block.header.height, block.hash());
-                let encoded_block =
-                    JsonValue::String(base64::encode(&serialize_async(block).await));
-                notif_sub.notify(vec![encoded_block].into()).await;
+            node.validator.blockchain.headers.remove_sync(&synced_headers)?;
+
+            // Perform finalization for received blocks
+            let finalized = node.validator.finalization().await?;
+            if !finalized.is_empty() {
+                // Notify subscriber
+                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;
             }
 
-            received_blocks += response.blocks.len();
             info!(target: "darkfid::task::sync::retrieve_blocks", "Blocks received: {}/{}", received_blocks, total);
         }
     }
 
-    Ok(())
+    Ok(last_received)
 }
 
 /// Auxiliary function to retrieve best fork state from a random peer.

src/contract/test-harness/src/money_pow_reward.rs

@@ -132,7 +132,7 @@ impl TestHarness {
         let mut found_owncoins = vec![];
         for holder in holders {
             let wallet = self.holders.get_mut(holder).unwrap();
-            wallet.validator.add_blocks(&[block.clone()]).await?;
+            wallet.validator.add_test_blocks(&[block.clone()]).await?;
             wallet.money_merkle_tree.append(MerkleNode::from(params.output.coin.inner()));
 
             // Attempt to decrypt the note to see if this is a coin for the holder

src/error.rs

@@ -382,9 +382,6 @@ pub enum Error {
     #[error("Block {0} contains 0 transactions")]
     BlockContainsNoTransactions(String),
 
-    #[error("Blockchain sync failed")]
-    BlockchainSyncError,
-
     #[error("Contract {0} not found in database")]
     ContractNotFound(String),
 

src/validator/consensus.rs

@@ -79,10 +79,18 @@ impl Consensus {
     /// Generate a new empty fork.
     pub async fn generate_empty_fork(&self) -> Result<()> {
         debug!(target: "validator::consensus::generate_empty_fork", "Generating new empty fork...");
-        let mut lock = self.forks.write().await;
+        let mut forks = self.forks.write().await;
+        // Check if we already have an empty fork
+        for fork in forks.iter() {
+            if fork.proposals.is_empty() {
+                debug!(target: "validator::consensus::generate_empty_fork", "An empty fork already exists.");
+                drop(forks);
+                return Ok(())
+            }
+        }
         let fork = Fork::new(self.blockchain.clone(), self.module.read().await.clone()).await?;
-        lock.push(fork);
-        drop(lock);
+        forks.push(fork);
+        drop(forks);
         debug!(target: "validator::consensus::generate_empty_fork", "Fork generated!");
         Ok(())
     }
@@ -280,8 +288,28 @@ impl Consensus {
         Ok(vec![])
     }
 
+    /// Auxiliary function to retrieve current best fork last header.
+    /// If no forks exist, grab the last header from canonical.
+    pub async fn best_fork_last_header(&self) -> Result<(u32, HeaderHash)> {
+        // Grab a lock over current forks
+        let forks = self.forks.read().await;
+
+        // Check if node has any forks
+        if forks.is_empty() {
+            drop(forks);
+            return self.blockchain.last()
+        }
+
+        // Grab best fork
+        let fork = &forks[best_fork_index(&forks)?];
+
+        // Grab its last header
+        let last = fork.last_proposal()?;
+        drop(forks);
+        Ok((last.block.header.height, last.hash))
+    }
+
     /// Auxiliary function to retrieve current best fork proposals.
-    /// If multiple best forks exist, grab the proposals of the first one
     /// If provided tip is not the canonical(finalized), or no forks exist,
     /// an empty vector is returned.
     pub async fn get_best_fork_proposals(&self, tip: HeaderHash) -> Result<Vec<Proposal>> {
@@ -314,12 +342,15 @@ impl Consensus {
 
     /// Auxiliary function to purge current forks and reset the ones starting
     /// with the provided prefix, excluding provided finalized fork.
+    /// Additionally, remove finalized transactions from the forks mempools,
+    /// along with the unporposed transactions sled trees.
     /// This function assumes that the prefix blocks have already been appended
     /// to canonical chain from the finalized fork.
     pub async fn reset_forks(
         &self,
         prefix: &[HeaderHash],
         finalized_fork_index: &usize,
+        finalized_txs: &[Transaction],
     ) -> Result<()> {
         // Grab a lock over current forks
         let mut forks = self.forks.write().await;
@@ -335,6 +366,8 @@ impl Consensus {
         let prefix_last = prefix.last().unwrap();
         let mut keep = vec![true; forks.len()];
         let mut referenced_trees = BTreeSet::new();
+        let finalized_txs_hashes: Vec<TransactionHash> =
+            finalized_txs.iter().map(|tx| tx.hash()).collect();
         for (index, fork) in forks.iter_mut().enumerate() {
             if &index == finalized_fork_index {
                 // Store its tree references
@@ -349,6 +382,8 @@ impl Consensus {
                 for tree in &overlay.state.dropped_tree_names {
                     referenced_trees.insert(tree.clone());
                 }
+                // Remove finalized proposals txs from fork's mempool
+                fork.mempool.retain(|tx| !finalized_txs_hashes.contains(tx));
                 drop(overlay);
                 drop(fork_overlay);
                 continue
@@ -362,6 +397,9 @@ impl Consensus {
                 continue
             }
 
+            // Remove finalized proposals txs from fork's mempool
+            fork.mempool.retain(|tx| !finalized_txs_hashes.contains(tx));
+
             // Remove the commited differences
             let rest_proposals = fork.proposals.split_off(excess);
             let rest_diffs = fork.diffs.split_off(excess);
@@ -423,6 +461,9 @@ impl Consensus {
         let mut iter = keep.iter();
         forks.retain(|_| *iter.next().unwrap());
 
+        // Remove finalized proposals txs from the unporposed txs sled tree
+        self.blockchain.remove_pending_txs(finalized_txs)?;
+
         // Drop forks lock
         drop(forks);
 

src/validator/mod.rs

@@ -319,7 +319,7 @@ impl Validator {
 
     /// The node checks if best fork can be finalized.
     /// If proposals can be finalized, node appends them to canonical,
-    /// and rebuilds the best fork.
+    /// and resets the current forks.
     pub async fn finalization(&self) -> Result<Vec<BlockInfo>> {
         // Grab append lock so no new proposals can be appended while
         // we execute finalization
@@ -363,18 +363,20 @@ impl Validator {
 
         // Apply finalized proposals diffs and update PoW module
         let mut module = self.consensus.module.write().await;
+        let mut finalized_txs = vec![];
         info!(target: "validator::finalization", "Finalizing proposals:");
         for (index, proposal) in finalized_proposals.iter().enumerate() {
             info!(target: "validator::finalization", "\t{} - {}", proposal, finalized_blocks[index].header.height);
             fork.overlay.lock().unwrap().overlay.lock().unwrap().apply_diff(&mut diffs[index])?;
             let next_difficulty = module.next_difficulty()?;
             module.append(finalized_blocks[index].header.timestamp, &next_difficulty);
+            finalized_txs.extend_from_slice(&finalized_blocks[index].txs);
         }
         drop(module);
         drop(forks);
 
         // Reset forks starting with the finalized blocks
-        self.consensus.reset_forks(&finalized_proposals, &finalized_fork).await?;
+        self.consensus.reset_forks(&finalized_proposals, &finalized_fork, &finalized_txs).await?;
         info!(target: "validator::finalization", "Finalization completed!");
 
         // Release append lock
@@ -383,20 +385,10 @@ impl Validator {
         Ok(finalized_blocks)
     }
 
-    // ==========================
-    // State transition functions
-    // ==========================
-    // TODO TESTNET: Write down all cases below
-    // State transition checks should be happening in the following cases for a sync node:
-    // 1) When a finalized block is received
-    // 2) When a transaction is being broadcasted to us
-    // State transition checks should be happening in the following cases for a consensus participating node:
-    // 1) When a finalized block is received
-    // 2) When a transaction is being broadcasted to us
-    // ==========================
-
     /// Validate a set of [`BlockInfo`] in sequence and apply them if all are valid.
-    pub async fn add_blocks(&self, blocks: &[BlockInfo]) -> Result<()> {
+    /// Note: this function should only be used in tests when we don't want to
+    /// perform consensus logic.
+    pub async fn add_test_blocks(&self, blocks: &[BlockInfo]) -> Result<()> {
         debug!(target: "validator::add_blocks", "Instantiating BlockchainOverlay");
         let overlay = BlockchainOverlay::new(&self.blockchain)?;