darkfid: fixed some sync issues

bin/darkfid/src/proto/protocol_proposal.rs

@@ -19,7 +19,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use log::{debug, error};
+use log::{debug, error, warn};
 use smol::Executor;
 use tinyjson::JsonValue;
 
@@ -174,7 +174,7 @@ impl ProtocolProposal {
 
             // Response should not be empty
             if response.proposals.is_empty() {
-                debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Peer responded with empty sequence");
+                warn!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Peer responded with empty sequence, node might be out of sync!");
                 continue
             }
 
@@ -197,12 +197,16 @@ impl ProtocolProposal {
             }
 
             for proposal in &response.proposals {
-                if let Err(e) = self.validator.append_proposal(proposal).await {
-                    error!(
-                        target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
-                        "Error while appending response proposal: {e}"
-                    );
-                    break
+                match self.validator.append_proposal(proposal).await {
+                    Ok(()) => { /* Do nothing */ }
+                    // Skip already existing proposals
+                    Err(Error::ProposalAlreadyExists) => continue,
+                    Err(e) => {
+                        error!(
+                            target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
+                            "Error while appending response proposal: {e}"
+                        );
+                    }
                 };
                 let message = ProposalMessage(proposal.clone());
                 self.p2p.broadcast_with_exclude(&message, &exclude_list).await;

bin/darkfid/src/proto/protocol_sync.rs

@@ -35,7 +35,7 @@ use darkfi::{
 use darkfi_serial::{SerialDecodable, SerialEncodable};
 
 // Constant defining how many blocks we send during syncing.
-pub const BATCH: usize = 10;
+pub const BATCH: usize = 20;
 
 /// Structure represening a request to ask a node for their current
 /// canonical(finalized) tip block hash, if they are synced. We also
@@ -370,9 +370,17 @@ impl ProtocolSync {
             // Otherwise, grab best fork proposals sequence.
             let proposals = match request.fork_tip {
                 Some(fork_tip) => {
-                    self.validator.consensus.get_fork_proposals(request.tip, fork_tip).await
+                    self.validator
+                        .consensus
+                        .get_fork_proposals(request.tip, fork_tip, BATCH as u32)
+                        .await
+                }
+                None => {
+                    self.validator
+                        .consensus
+                        .get_best_fork_proposals(request.tip, BATCH as u32)
+                        .await
                 }
-                None => self.validator.consensus.get_best_fork_proposals(request.tip).await,
             };
             let proposals = match proposals {
                 Ok(p) => p,

src/blockchain/block_store.rs

@@ -471,6 +471,22 @@ impl BlockStore {
         Ok(ret.iter().rev().copied().collect())
     }
 
+    /// Fetch all hashes after given height. In the iteration, if an order
+    /// height is not found, the iteration stops and the function returns what
+    /// it has found so far in the store's order tree.
+    pub fn get_all_after(&self, height: u32) -> Result<Vec<HeaderHash>> {
+        let mut ret = vec![];
+
+        let mut key = height;
+        while let Some(found) = self.order.get_gt(key.to_be_bytes())? {
+            let (height, hash) = parse_u32_key_record(found)?;
+            key = height;
+            ret.push(hash);
+        }
+
+        Ok(ret)
+    }
+
     /// Fetch the first block hash in the order tree, based on the `Ord`
     /// implementation for `Vec<u8>`.
     pub fn get_first(&self) -> Result<(u32, HeaderHash)> {

src/validator/consensus.rs

@@ -252,44 +252,82 @@ impl Consensus {
         Ok(Some(index))
     }
 
-    /// Auxiliary function to retrieve a fork proposals.
-    /// If provided tip is not the canonical(finalized), or fork doesn't exists,
-    /// an empty vector is returned.
+    /// Auxiliary function to retrieve a fork proposals, starting from provided tip.
+    /// If provided tip is too far behind, or fork doesn't exists, an empty vector is returned.
     pub async fn get_fork_proposals(
         &self,
         tip: HeaderHash,
         fork_tip: HeaderHash,
+        limit: u32,
     ) -> Result<Vec<Proposal>> {
-        // Tip must be canonical(finalized) blockchain last
-        if self.blockchain.last()?.1 != tip {
-            return Ok(vec![])
-        }
-
         // 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 Ok(vec![])
+        // Retrieve our current canonical tip height
+        let last_block_height = self.blockchain.last()?.0;
+
+        // Check if request tip is canonical
+        let mut canonical_blocks = vec![];
+        if let Ok(existing_tip) = self.blockchain.get_blocks_by_hash(&[tip]) {
+            // Check tip is not far behind
+            if last_block_height - existing_tip[0].header.height >= limit {
+                drop(forks);
+                return Ok(canonical_blocks)
+            }
+
+            // Retrieve all tips after requested one
+            let headers = self.blockchain.blocks.get_all_after(existing_tip[0].header.height)?;
+            let blocks = self.blockchain.get_blocks_by_hash(&headers)?;
+
+            // Add everything to the return vec
+            for block in blocks {
+                canonical_blocks.push(Proposal::new(block));
+            }
         }
 
-        // Find fork by its tip
+        // Find the fork containing the requested tip and grab its sequence
+        let mut proposals = vec![];
         for fork in forks.iter() {
-            if fork.proposals.last() == Some(&fork_tip) {
-                // Grab its proposals
-                let blocks = fork.overlay.lock().unwrap().get_blocks_by_hash(&fork.proposals)?;
-                let mut ret = Vec::with_capacity(blocks.len());
-                for block in blocks {
-                    ret.push(Proposal::new(block));
+            let mut found = false;
+            for p in fork.proposals.iter().rev() {
+                if p != &fork_tip {
+                    continue
                 }
-                drop(forks);
-                return Ok(ret)
+                found = true;
+                break
+            }
+
+            if !found {
+                continue
+            }
+
+            let mut headers = vec![];
+            for p in &fork.proposals {
+                headers.push(*p);
+                if p == &fork_tip {
+                    break
+                }
+            }
+
+            let blocks = fork.overlay.lock().unwrap().get_blocks_by_hash(&headers)?;
+            for block in blocks {
+                proposals.push(Proposal::new(block));
             }
         }
 
-        // Fork was not found
-        Ok(vec![])
+        // Check if we found anything.
+        // Even if we found canonical blocks, if the
+        // request doesn't correspond to a known fork
+        // we return an empty vector.
+        if proposals.is_empty() {
+            drop(forks);
+            return Ok(proposals)
+        }
+
+        // Join the two vectors and return them
+        canonical_blocks.append(&mut proposals);
+        drop(forks);
+        Ok(canonical_blocks)
     }
 
     /// Auxiliary function to retrieve current best fork last header.
@@ -313,15 +351,13 @@ impl Consensus {
         Ok((last.block.header.height, last.hash))
     }
 
-    /// Auxiliary function to retrieve current best fork proposals.
-    /// 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>> {
-        // Tip must be canonical(finalized) blockchain last
-        if self.blockchain.last()?.1 != tip {
-            return Ok(vec![])
-        }
-
+    /// Auxiliary function to retrieve current best fork proposals, starting from provided tip.
+    /// If provided tip is too far behind, or fork doesn't exists, an empty vector is returned.
+    pub async fn get_best_fork_proposals(
+        &self,
+        tip: HeaderHash,
+        limit: u32,
+    ) -> Result<Vec<Proposal>> {
         // Grab a lock over current forks
         let forks = self.forks.read().await;
 
@@ -331,17 +367,42 @@ impl Consensus {
             return Ok(vec![])
         }
 
+        // Retrieve our current canonical tip height
+        let last_block_height = self.blockchain.last()?.0;
+
+        // Check if request tip is canonical
+        let mut canonical_blocks = vec![];
+        if let Ok(existing_tip) = self.blockchain.get_blocks_by_hash(&[tip]) {
+            // Check tip is not far behind
+            if last_block_height - existing_tip[0].header.height >= limit {
+                drop(forks);
+                return Ok(canonical_blocks)
+            }
+
+            // Retrieve all tips after requested one
+            let headers = self.blockchain.blocks.get_all_after(existing_tip[0].header.height)?;
+            let blocks = self.blockchain.get_blocks_by_hash(&headers)?;
+
+            // Add everything to the return vec
+            for block in blocks {
+                canonical_blocks.push(Proposal::new(block));
+            }
+        }
+
         // Grab best fork
         let fork = &forks[best_fork_index(&forks)?];
 
         // Grab its proposals
         let blocks = fork.overlay.lock().unwrap().get_blocks_by_hash(&fork.proposals)?;
-        let mut ret = Vec::with_capacity(blocks.len());
+        let mut proposals = Vec::with_capacity(blocks.len());
         for block in blocks {
-            ret.push(Proposal::new(block));
+            proposals.push(Proposal::new(block));
         }
 
-        Ok(ret)
+        // Join the two vectors and return them
+        canonical_blocks.append(&mut proposals);
+        drop(forks);
+        Ok(canonical_blocks)
     }
 
     /// Auxiliary function to purge current forks and reset the ones starting

src/validator/mod.rs

@@ -36,7 +36,7 @@ use crate::{
 
 /// DarkFi consensus module
 pub mod consensus;
-use consensus::{Consensus, Proposal};
+use consensus::{Consensus, Fork, Proposal};
 
 /// DarkFi PoW module
 pub mod pow;
@@ -403,7 +403,8 @@ impl Validator {
     /// block hash matches the expected header one.
     /// Note: this function should only be used for blocks received using a
     /// checkpoint, since in that case we enforce the node to follow the sequence,
-    /// assuming they all its blocks are valid.
+    /// assuming they all its blocks are valid. Additionally, it will update
+    /// any forks to a single empty one, holding the updated module.
     pub async fn add_checkpoint_blocks(
         &self,
         blocks: &[BlockInfo],
@@ -483,7 +484,11 @@ impl Validator {
         self.blockchain.remove_pending_txs(&removed_txs)?;
 
         // Update PoW module
-        *self.consensus.module.write().await = module;
+        *self.consensus.module.write().await = module.clone();
+
+        // Update forks
+        *self.consensus.forks.write().await =
+            vec![Fork::new(self.blockchain.clone(), module).await?];
 
         Ok(())
     }