darkfid: fixed fork sync issues

bin/darkfid/src/proto/protocol_proposal.rs

@@ -133,6 +133,11 @@ impl ProtocolProposal {
 
             // If proposal fork chain was not found, we ask our peer for its sequence
             debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Asking peer for fork sequence");
+
+            // Cleanup subscriber
+            self.proposals_response_sub.clean().await?;
+
+            // Grab last known block to create the request and execute it
             let last = self.validator.blockchain.last()?;
             let request = ForkSyncRequest { tip: last.1, fork_tip: Some(proposal_copy.0.hash) };
             self.channel.send(&request).await?;
@@ -149,6 +154,7 @@ impl ProtocolProposal {
                     continue
                 }
             };
+            debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Peer response: {response:?}");
 
             // Verify and store retrieved proposals
             debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Processing received proposals");

bin/darkfid/src/proto/protocol_sync.rs

@@ -223,6 +223,8 @@ impl ProtocolSync {
                 continue
             }
 
+            debug!(target: "darkfid::proto::protocol_sync::handle_receive_request", "Received request: {request:?}");
+
             // If a fork tip is provided, grab its fork proposals sequence.
             // Otherwise, grab best fork proposals sequence.
             let proposals = match request.fork_tip {
@@ -244,6 +246,7 @@ impl ProtocolSync {
             };
 
             let response = ForkSyncResponse { proposals };
+            debug!(target: "darkfid::proto::protocol_sync::handle_receive_request", "Response: {response:?}");
             if let Err(e) = self.channel.send(&response).await {
                 debug!(
                     target: "darkfid::proto::protocol_sync::handle_receive_fork_request",

bin/darkfid/src/tests/mod.rs

@@ -41,17 +41,17 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let config = HarnessConfig {
         pow_target,
         pow_fixed_difficulty: pow_fixed_difficulty.clone(),
-        finalization_threshold: 6,
+        finalization_threshold: 3,
         alice_initial: 1000,
         bob_initial: 500,
     };
     let th = Harness::new(config, true, &ex).await?;
 
     // Retrieve genesis block
-    let previous = th.alice.validator.blockchain.last_block()?;
+    let genesis = th.alice.validator.blockchain.last_block()?;
 
     // Generate next blocks
-    let block1 = th.generate_next_block(&previous).await?;
+    let block1 = th.generate_next_block(&genesis).await?;
     let block2 = th.generate_next_block(&block1).await?;
     let block3 = th.generate_next_block(&block2).await?;
     let block4 = th.generate_next_block(&block3).await?;
@@ -99,10 +99,11 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
 
     // Nodes must have two forks with 2 blocks each
     th.validate_fork_chains(2, vec![2, 2]).await;
-    // If Charlie already had the small fork as its best,
-    // it will have a single fork with 2 blocks.
+    // Check charlie has the correct forks
     let charlie_forks = charlie.consensus.forks.read().await;
     if small_best {
+        // If Charlie already had the small fork as its best,
+        // it will have a single fork with 2 blocks.
         assert_eq!(charlie_forks.len(), 1);
         assert_eq!(charlie_forks[0].proposals.len(), 2);
     } else {

bin/darkfid/src/tests/sync_forks.rs

@@ -18,7 +18,7 @@
 
 use std::sync::Arc;
 
-use darkfi::{net::Settings, Result};
+use darkfi::{net::Settings, validator::utils::best_fork_index, Result};
 use darkfi_contract_test_harness::init_logger;
 use darkfi_sdk::num_traits::One;
 use num_bigint::BigUint;
@@ -72,10 +72,14 @@ async fn sync_forks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let charlie =
         generate_node(&th.vks, &th.validator_config, &settings, &ex, false, false).await?;
 
-    // Verify node synced the big(best) fork
+    // Verify node synced the best fork
+    let forks = th.alice.validator.consensus.forks.read().await;
+    let best_fork = &forks[best_fork_index(&forks)?];
     let charlie_forks = charlie.validator.consensus.forks.read().await;
     assert_eq!(charlie_forks.len(), 1);
-    assert_eq!(charlie_forks[0].proposals.len(), 3);
+    assert_eq!(charlie_forks[0].proposals.len(), best_fork.proposals.len());
+    let small_best = best_fork.proposals.len() == 1;
+    drop(forks);
     drop(charlie_forks);
 
     // Extend the small fork sequences and add it to nodes
@@ -85,12 +89,22 @@ async fn sync_forks_real(ex: Arc<Executor<'static>>) -> Result<()> {
     let block7 = th.generate_next_block(&block5).await?;
     th.add_blocks(&vec![block7]).await?;
 
-    // Check charlie has all the forks
+    // Check charlie has the correct forks
     let charlie_forks = charlie.validator.consensus.forks.read().await;
-    assert_eq!(charlie_forks.len(), 3);
-    assert_eq!(charlie_forks[0].proposals.len(), 3);
-    assert_eq!(charlie_forks[1].proposals.len(), 2);
-    assert_eq!(charlie_forks[2].proposals.len(), 2);
+    if small_best {
+        // If Charlie already had a small fork as its best,
+        // it will have two forks with 2 blocks each.
+        assert_eq!(charlie_forks.len(), 2);
+        assert_eq!(charlie_forks[0].proposals.len(), 2);
+        assert_eq!(charlie_forks[1].proposals.len(), 2);
+    } else {
+        // Charlie didn't originaly have the forks, but they
+        // should be synced when their proposals were received
+        assert_eq!(charlie_forks.len(), 3);
+        assert_eq!(charlie_forks[0].proposals.len(), 3);
+        assert_eq!(charlie_forks[1].proposals.len(), 2);
+        assert_eq!(charlie_forks[2].proposals.len(), 2);
+    }
     drop(charlie_forks);
 
     // Thanks for reading