darkfid/task/unknown_proposal: full refactor for cleaner code and usage of append lock while generating the reorg fork

bin/darkfid/src/task/consensus.rs

@@ -110,7 +110,6 @@ pub async fn consensus_init_task(
             Err(Error::NetworkNotConnected) => {
                 // Sync node again
                 *node.validator.synced.write().await = false;
-                node.validator.consensus.purge_forks().await?;
                 if !config.skip_sync {
                     loop {
                         match sync_task(&node, checkpoint).await {
@@ -202,13 +201,12 @@ async fn consensus_task(
             continue
         }
 
-        // Purge all unreferenced contract trees from the database iff
-        // the node is not executing a reorg.
-        let reorg_lock = node.validator.synced.write().await;
-        if !*reorg_lock {
-            purge_unreferenced_trees(node).await;
-        }
-        drop(reorg_lock);
+        // Grab the append lock so no other proposal gets processed
+        // while the node is purging all unreferenced contract trees
+        // from the database.
+        let append_lock = node.validator.consensus.append_lock.write().await;
+        purge_unreferenced_trees(node).await;
+        drop(append_lock);
 
         let mut notif_blocks = Vec::with_capacity(confirmed.len());
         for block in confirmed {

bin/darkfid/src/task/unknown_proposal.rs

@@ -21,12 +21,13 @@ use std::{
     sync::Arc,
 };
 
+use num_bigint::BigUint;
 use smol::{channel::Receiver, lock::RwLock};
 use tinyjson::JsonValue;
 use tracing::{debug, error, info};
 
 use darkfi::{
-    blockchain::BlockDifficulty,
+    blockchain::{BlockDifficulty, HeaderHash},
     net::{ChannelPtr, P2pPtr},
     rpc::jsonrpc::JsonSubscriber,
     util::{encoding::base64, time::Timestamp},
@@ -37,7 +38,8 @@ use darkfi::{
         verification::verify_fork_proposal,
         ValidatorPtr,
     },
-    Error, Result,
+    Error::{Custom, DatabaseError, PoWInvalidOutHash, ProposalAlreadyExists},
+    Result,
 };
 use darkfi_serial::serialize_async;
 
@@ -206,7 +208,7 @@ async fn handle_unknown_proposal(
         match validator.append_proposal(proposal).await {
             Ok(()) => { /* Do nothing */ }
             // Skip already existing proposals
-            Err(Error::ProposalAlreadyExists) => continue,
+            Err(ProposalAlreadyExists) => continue,
             Err(e) => {
                 debug!(
                     target: "darkfid::task::handle_unknown_proposal",
@@ -237,6 +239,8 @@ async fn handle_unknown_proposal(
 ///
 /// Note: Always remember to purge new trees from the database if not
 /// needed.
+//  TODO: We keep everything in memory which can result in OOM for a
+//        valid long fork. We could use some disk space to store stuff.
 async fn handle_reorg(
     validator: &ValidatorPtr,
     p2p: &P2pPtr,
@@ -253,11 +257,188 @@ async fn handle_reorg(
         return true
     }
 
-    // Communication setup
-    let Ok(response_sub) = channel.subscribe_msg::<ForkHeaderHashResponse>().await else {
-        debug!(target: "darkfid::task::handle_reorg", "Failure during `ForkHeaderHashResponse` communication setup with peer: {channel:?}");
+    // Retrieve communications timeout
+    let comms_timeout =
+        p2p.settings().read_arc().await.outbound_connect_timeout(channel.address().scheme());
+
+    // Find last common header and its sequence, going backwards from
+    // the proposal.
+    let (last_common_height, last_common_hash, peer_header_hashes) =
+        match retrieve_peer_header_hashes(validator, (&channel, &comms_timeout), proposal).await {
+            Ok(t) => t,
+            Err(DatabaseError(e)) => {
+                error!(target: "darkfid::task::handle_reorg", "Internal error while retrieving peer headers hashes: {e}");
+                return false
+            }
+            Err(e) => {
+                error!(target: "darkfid::task::handle_reorg", "Retrieving peer headers hashes failed: {e}");
+                return true
+            }
+        };
+
+    // Check if we have a sequence to process
+    if peer_header_hashes.is_empty() {
+        debug!(target: "darkfid::task::handle_reorg", "No headers to process, skipping...");
+        return true
+    }
+
+    // Create a new PoW module from last common height
+    let module = match PoWModule::new(
+        validator.consensus.blockchain.clone(),
+        validator.consensus.module.read().await.target,
+        validator.consensus.module.read().await.fixed_difficulty.clone(),
+        Some(last_common_height + 1),
+    ) {
+        Ok(m) => m,
+        Err(e) => {
+            error!(target: "darkfid::task::handle_reorg", "PoWModule generation failed: {e}");
+            return false
+        }
+    };
+
+    // Grab last common height ranks
+    let last_difficulty = match last_common_height {
+        0 => {
+            let genesis_timestamp = match validator.blockchain.genesis_block() {
+                Ok(b) => b.header.timestamp,
+                Err(e) => {
+                    error!(target: "darkfid::task::handle_reorg", "Retrieving genesis block failed: {e}");
+                    return false
+                }
+            };
+            BlockDifficulty::genesis(genesis_timestamp)
+        }
+        _ => match validator.blockchain.blocks.get_difficulty(&[last_common_height], true) {
+            Ok(d) => d[0].clone().unwrap(),
+            Err(e) => {
+                error!(target: "darkfid::task::handle_reorg", "Retrieving block difficulty failed: {e}");
+                return false
+            }
+        },
+    };
+
+    // Retrieve the headers of the hashes sequence and its ranking
+    let (targets_rank, hashes_rank) = match retrieve_peer_headers_sequence_ranking(
+        (&last_common_height, &last_common_hash, &module, &last_difficulty),
+        (&channel, &comms_timeout),
+        &proposal.hash,
+        &peer_header_hashes,
+    )
+    .await
+    {
+        Ok(p) => p,
+        Err(DatabaseError(e)) => {
+            error!(target: "darkfid::task::handle_reorg", "Internal error while retrieving peer headers: {e}");
+            return false
+        }
+        Err(e) => {
+            error!(target: "darkfid::task::handle_reorg", "Retrieving peer headers failed: {e}");
+            return true
+        }
+    };
+
+    // Grab the append lock so no other proposal gets processed while
+    // we are verifying the sequence.
+    let append_lock = validator.consensus.append_lock.write().await;
+
+    // Check if the sequence ranks higher than our current best fork
+    let mut forks = validator.consensus.forks.write().await;
+    let index = match best_fork_index(&forks) {
+        Ok(i) => i,
+        Err(e) => {
+            debug!(target: "darkfid::task::handle_reorg", "Retrieving best fork index failed: {e}");
+            return false
+        }
+    };
+    let best_fork = &forks[index];
+    if targets_rank < best_fork.targets_rank ||
+        (targets_rank == best_fork.targets_rank && hashes_rank <= best_fork.hashes_rank)
+    {
+        info!(target: "darkfid::task::handle_reorg", "Peer sequence ranks lower than our current best fork, skipping...");
+        return true
+    }
+
+    // Generate the peer fork and retrieve its ranking
+    let peer_fork = match retrieve_peer_fork(
+        validator,
+        (&last_common_height, &module, &last_difficulty),
+        (&channel, &comms_timeout),
+        proposal,
+        &peer_header_hashes,
+    )
+    .await
+    {
+        Ok(p) => p,
+        Err(DatabaseError(e)) => {
+            error!(target: "darkfid::task::handle_reorg", "Internal error while retrieving peer fork: {e}");
+            return false
+        }
+        Err(e) => {
+            error!(target: "darkfid::task::handle_reorg", "Retrieving peer fork failed: {e}");
+            return true
+        }
+    };
+
+    // Check if the peer fork ranks higher than our current best fork
+    if peer_fork.targets_rank < best_fork.targets_rank ||
+        (peer_fork.targets_rank == best_fork.targets_rank &&
+            peer_fork.hashes_rank <= best_fork.hashes_rank)
+    {
+        info!(target: "darkfid::task::handle_reorg", "Peer fork ranks lower than our current best fork, skipping...");
         return true
+    }
+
+    // Execute the reorg
+    info!(target: "darkfid::task::handle_reorg", "Peer fork ranks higher than our current best fork, executing reorg...");
+    if let Err(e) = validator.blockchain.reset_to_height(last_common_height) {
+        error!(target: "darkfid::task::handle_reorg", "Applying full inverse diff failed: {e}");
+        return false
     };
+    *validator.consensus.module.write().await = module;
+    *forks = vec![peer_fork];
+    drop(forks);
+    drop(append_lock);
+
+    // Check if we can confirm anything and broadcast them
+    let confirmed = match validator.confirmation().await {
+        Ok(f) => f,
+        Err(e) => {
+            error!(target: "darkfid::task::handle_reorg", "Confirmation failed: {e}");
+            return false
+        }
+    };
+
+    if !confirmed.is_empty() {
+        let mut notif_blocks = Vec::with_capacity(confirmed.len());
+        for block in confirmed {
+            notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+        }
+        blocks_sub.notify(JsonValue::Array(notif_blocks)).await;
+    }
+
+    // Broadcast proposal to the network
+    let message = ProposalMessage(proposal.clone());
+    p2p.broadcast(&message).await;
+
+    // Notify proposals subscriber
+    let enc_prop = JsonValue::String(base64::encode(&serialize_async(proposal).await));
+    proposals_sub.notify(vec![enc_prop].into()).await;
+
+    false
+}
+
+/// Auxiliary function to retrieve the last common header and height,
+/// along with the headers sequence up to provided peer proposal.
+async fn retrieve_peer_header_hashes(
+    // Validator pointer
+    validator: &ValidatorPtr,
+    // Peer channel and its communications timeout
+    channel: (&ChannelPtr, &u64),
+    // Peer fork proposal
+    proposal: &Proposal,
+) -> Result<(u32, HeaderHash, Vec<HeaderHash>)> {
+    // Communication setup
+    let response_sub = channel.0.subscribe_msg::<ForkHeaderHashResponse>().await?;
 
     // Keep track of received header hashes sequence
     let mut peer_header_hashes = vec![];
@@ -268,36 +449,21 @@ async fn handle_reorg(
     for height in (0..proposal.block.header.height).rev() {
         // Request peer header hash for this height
         let request = ForkHeaderHashRequest { height, fork_header: proposal.hash };
-        if let Err(e) = channel.send(&request).await {
-            debug!(target: "darkfid::task::handle_reorg", "Channel send failed: {e}");
-            return true
-        };
+        channel.0.send(&request).await?;
 
-        let comms_timeout =
-            p2p.settings().read_arc().await.outbound_connect_timeout(channel.address().scheme());
         // Node waits for response
-        let response = match response_sub.receive_with_timeout(comms_timeout).await {
-            Ok(r) => r,
-            Err(e) => {
-                debug!(target: "darkfid::task::handle_reorg", "Asking peer for header hash failed: {e}");
-                return true
-            }
-        };
+        let response = response_sub.receive_with_timeout(*channel.1).await?;
         debug!(target: "darkfid::task::handle_reorg", "Peer response: {response:?}");
 
         // Check if peer returned a header
         let Some(peer_header) = response.fork_header else {
-            debug!(target: "darkfid::task::handle_reorg", "Peer responded with an empty header");
-            return true
+            return Err(Custom(String::from("Peer responded with an empty header")))
         };
 
         // Check if we know this header
         let headers = match validator.blockchain.blocks.get_order(&[height], false) {
-            Ok(r) => r,
-            Err(e) => {
-                error!(target: "darkfid::task::handle_reorg", "Retrieving headers failed: {e}");
-                return false
-            }
+            Ok(h) => h,
+            Err(e) => return Err(DatabaseError(format!("Retrieving headers failed: {e}"))),
         };
         match headers[0] {
             Some(known_header) => {
@@ -313,93 +479,56 @@ async fn handle_reorg(
         }
     }
 
-    // Check if we have a sequence to process
-    if peer_header_hashes.is_empty() {
-        debug!(target: "darkfid::task::handle_reorg", "No headers to process, skipping...");
-        return true
-    }
+    Ok((previous_height, previous_hash, peer_header_hashes))
+}
 
+/// Auxiliary function to retrieve provided peer headers hashes
+/// sequence and its ranking, based on provided last common
+/// information.
+async fn retrieve_peer_headers_sequence_ranking(
+    // Last common header, PoW module and difficulty
+    last_common_info: (&u32, &HeaderHash, &PoWModule, &BlockDifficulty),
+    // Peer channel and its communications timeout
+    channel: (&ChannelPtr, &u64),
+    // Peer fork proposal header for our requests
+    fork_tip: &HeaderHash,
+    // Peer header hashes sequence
+    header_hashes: &[HeaderHash],
+) -> Result<(BigUint, BigUint)> {
     // Communication setup
-    let Ok(response_sub) = channel.subscribe_msg::<ForkHeadersResponse>().await else {
-        debug!(target: "darkfid::task::handle_reorg", "Failure during `ForkHeadersResponse` communication setup with peer: {channel:?}");
-        return true
-    };
-
-    // Grab last common height ranks
-    let last_common_height = previous_height;
-    let last_difficulty = match previous_height {
-        0 => {
-            let genesis_timestamp = match validator.blockchain.genesis_block() {
-                Ok(b) => b.header.timestamp,
-                Err(e) => {
-                    error!(target: "darkfid::task::handle_reorg", "Retrieving genesis block failed: {e}");
-                    return false
-                }
-            };
-            BlockDifficulty::genesis(genesis_timestamp)
-        }
-        _ => match validator.blockchain.blocks.get_difficulty(&[last_common_height], true) {
-            Ok(d) => d[0].clone().unwrap(),
-            Err(e) => {
-                error!(target: "darkfid::task::handle_reorg", "Retrieving block difficulty failed: {e}");
-                return false
-            }
-        },
-    };
-
-    // Create a new PoW from last common height
-    let module = match PoWModule::new(
-        validator.consensus.blockchain.clone(),
-        validator.consensus.module.read().await.target,
-        validator.consensus.module.read().await.fixed_difficulty.clone(),
-        Some(last_common_height + 1),
-    ) {
-        Ok(m) => m,
-        Err(e) => {
-            error!(target: "darkfid::task::handle_reorg", "PoWModule generation failed: {e}");
-            return false
-        }
-    };
+    let response_sub = channel.0.subscribe_msg::<ForkHeadersResponse>().await?;
 
     // Retrieve the headers of the hashes sequence, in batches, keeping track of the sequence ranking
-    info!(target: "darkfid::task::handle_reorg", "Retrieving {} headers from peer...", peer_header_hashes.len());
+    info!(target: "darkfid::task::handle_reorg", "Retrieving {} headers from peer...", header_hashes.len());
+    let mut previous_height = *last_common_info.0;
+    let mut previous_hash = *last_common_info.1;
+    let mut module = last_common_info.2.clone();
+    let mut targets_rank = last_common_info.3.ranks.targets_rank.clone();
+    let mut hashes_rank = last_common_info.3.ranks.hashes_rank.clone();
     let mut batch = Vec::with_capacity(BATCH);
     let mut total_processed = 0;
-    let mut targets_rank = last_difficulty.ranks.targets_rank.clone();
-    let mut hashes_rank = last_difficulty.ranks.hashes_rank.clone();
-    let mut headers_module = module.clone();
-    for (index, hash) in peer_header_hashes.iter().enumerate() {
+    for (index, hash) in header_hashes.iter().enumerate() {
         // Add hash in batch sequence
         batch.push(*hash);
 
         // Check if batch is full so we can send it
-        if batch.len() < BATCH && index != peer_header_hashes.len() - 1 {
+        if batch.len() < BATCH && index != header_hashes.len() - 1 {
             continue
         }
 
         // Request peer headers
-        let request = ForkHeadersRequest { headers: batch.clone(), fork_header: proposal.hash };
-        if let Err(e) = channel.send(&request).await {
-            debug!(target: "darkfid::task::handle_reorg", "Channel send failed: {e}");
-            return true
-        };
+        let request = ForkHeadersRequest { headers: batch.clone(), fork_header: *fork_tip };
+        channel.0.send(&request).await?;
 
-        let comms_timeout =
-            p2p.settings().read_arc().await.outbound_connect_timeout(channel.address().scheme());
         // Node waits for response
-        let response = match response_sub.receive_with_timeout(comms_timeout).await {
-            Ok(r) => r,
-            Err(e) => {
-                debug!(target: "darkfid::task::handle_reorg", "Asking peer for headers sequence failed: {e}");
-                return true
-            }
-        };
+        let response = response_sub.receive_with_timeout(*channel.1).await?;
         debug!(target: "darkfid::task::handle_reorg", "Peer response: {response:?}");
 
         // Response sequence must be the same length as the one requested
         if response.headers.len() != batch.len() {
-            debug!(target: "darkfid::task::handle_reorg", "Peer responded with a different headers sequence length");
-            return true
+            return Err(Custom(String::from(
+                "Peer responded with a different headers sequence length",
+            )))
         }
 
         // Process retrieved headers
@@ -409,41 +538,33 @@ async fn handle_reorg(
 
             // Validate its the header we requested
             if peer_header_hash != batch[peer_header_index] {
-                debug!(target: "darkfid::task::handle_reorg", "Peer responded with a differend header: {} - {peer_header_hash}", batch[peer_header_index]);
-                return true
+                return Err(Custom(format!(
+                    "Peer responded with a differend header: {} - {peer_header_hash}",
+                    batch[peer_header_index]
+                )))
             }
 
             // Validate sequence is correct
             if peer_header.previous != previous_hash || peer_header.height != previous_height + 1 {
-                debug!(target: "darkfid::task::handle_reorg", "Invalid header sequence detected");
-                return true
+                return Err(Custom(String::from("Invalid header sequence detected")))
             }
 
             // Verify header hash and calculate its rank
-            let (next_difficulty, target_distance_sq, hash_distance_sq) = match header_rank(
-                &headers_module,
-                peer_header,
-            ) {
-                Ok(tuple) => tuple,
-                Err(Error::PoWInvalidOutHash) => {
-                    debug!(target: "darkfid::task::handle_reorg", "Invalid header hash detected");
-                    return true
-                }
-                Err(e) => {
-                    debug!(target: "darkfid::task::handle_reorg", "Computing header rank failed: {e}");
-                    return false
-                }
-            };
+            let (next_difficulty, target_distance_sq, hash_distance_sq) =
+                match header_rank(&module, peer_header) {
+                    Ok(tuple) => tuple,
+                    Err(PoWInvalidOutHash) => return Err(PoWInvalidOutHash),
+                    Err(e) => {
+                        return Err(DatabaseError(format!("Computing header rank failed: {e}")))
+                    }
+                };
 
             // Update sequence ranking
             targets_rank += target_distance_sq.clone();
             hashes_rank += hash_distance_sq.clone();
 
             // Update PoW headers module
-            if let Err(e) = headers_module.append(peer_header, &next_difficulty) {
-                debug!(target: "darkfid::task::handle_reorg", "Error while appending header to module: {e}");
-                return true
-            };
+            module.append(peer_header, &next_difficulty)?;
 
             // Set previous header
             previous_height = peer_header.height;
@@ -451,72 +572,54 @@ async fn handle_reorg(
         }
 
         total_processed += response.headers.len();
-        info!(target: "darkfid::task::handle_reorg", "Headers received and verified: {total_processed}/{}", peer_header_hashes.len());
+        info!(target: "darkfid::task::handle_reorg", "Headers received and verified: {total_processed}/{}", header_hashes.len());
 
         // Reset batch
         batch = Vec::with_capacity(BATCH);
     }
 
-    // Check if the sequence ranks higher than our current best fork
-    let forks = validator.consensus.forks.read().await;
-    let index = match best_fork_index(&forks) {
-        Ok(i) => i,
-        Err(e) => {
-            debug!(target: "darkfid::task::handle_reorg", "Retrieving best fork index failed: {e}");
-            return false
-        }
-    };
-    let best_fork = &forks[index];
-    if targets_rank < best_fork.targets_rank ||
-        (targets_rank == best_fork.targets_rank && hashes_rank <= best_fork.hashes_rank)
-    {
-        info!(target: "darkfid::task::handle_reorg", "Peer sequence ranks lower than our current best fork, skipping...");
-        return true
-    }
-    drop(forks);
+    Ok((targets_rank, hashes_rank))
+}
 
+/// Auxiliary function to generate provided peer headers hashes fork
+/// and its ranking, based on provided last common information.
+async fn retrieve_peer_fork(
+    // Validator pointer
+    validator: &ValidatorPtr,
+    // Last common header height, PoW module and difficulty
+    last_common_info: (&u32, &PoWModule, &BlockDifficulty),
+    // Peer channel and its communications timeout
+    channel: (&ChannelPtr, &u64),
+    // Peer fork trigger proposal
+    proposal: &Proposal,
+    // Peer header hashes sequence
+    header_hashes: &[HeaderHash],
+) -> Result<Fork> {
     // Communication setup
-    let Ok(response_sub) = channel.subscribe_msg::<ForkProposalsResponse>().await else {
-        debug!(target: "darkfid::task::handle_reorg", "Failure during `ForkProposalsResponse` communication setup with peer: {channel:?}");
-        return true
-    };
-
-    // Update the node reorg flag
-    *validator.reorg.write().await = true;
+    let response_sub = channel.0.subscribe_msg::<ForkProposalsResponse>().await?;
 
     // Create a fork from last common height
     let mut peer_fork =
-        match Fork::new(validator.consensus.blockchain.clone(), module.clone()).await {
+        match Fork::new(validator.consensus.blockchain.clone(), last_common_info.1.clone()).await {
             Ok(f) => f,
-            Err(e) => {
-                error!(target: "darkfid::task::handle_reorg", "Generating peer fork failed: {e}");
-                *validator.reorg.write().await = false;
-                return false
-            }
+            Err(e) => return Err(DatabaseError(format!("Generating peer fork failed: {e}"))),
         };
-    peer_fork.targets_rank = last_difficulty.ranks.targets_rank.clone();
-    peer_fork.hashes_rank = last_difficulty.ranks.hashes_rank.clone();
+    peer_fork.targets_rank = last_common_info.2.ranks.targets_rank.clone();
+    peer_fork.hashes_rank = last_common_info.2.ranks.hashes_rank.clone();
 
     // Grab all state inverse diffs after last common height, and add them to the fork
-    let inverse_diffs = match validator
-        .blockchain
-        .blocks
-        .get_state_inverse_diffs_after(last_common_height)
-    {
-        Ok(i) => i,
-        Err(e) => {
-            error!(target: "darkfid::task::handle_reorg", "Retrieving state inverse diffs failed: {e}");
-            *validator.reorg.write().await = false;
-            return false
-        }
-    };
+    let inverse_diffs =
+        match validator.blockchain.blocks.get_state_inverse_diffs_after(*last_common_info.0) {
+            Ok(i) => i,
+            Err(e) => {
+                return Err(DatabaseError(format!("Retrieving state inverse diffs failed: {e}")))
+            }
+        };
     for inverse_diff in inverse_diffs.iter().rev() {
         let result =
             peer_fork.overlay.lock().unwrap().overlay.lock().unwrap().add_diff(inverse_diff);
         if let Err(e) = result {
-            error!(target: "darkfid::task::handle_reorg", "Applying inverse diff failed: {e}");
-            *validator.reorg.write().await = false;
-            return false
+            return Err(DatabaseError(format!("Applying state inverse diff failed: {e}")))
         }
     }
 
@@ -527,53 +630,37 @@ async fn handle_reorg(
     let diff = match diff {
         Ok(d) => d,
         Err(e) => {
-            error!(target: "darkfid::task::handle_reorg", "Generate full inverse diff failed: {e}");
-            *validator.reorg.write().await = false;
-            return false
+            return Err(DatabaseError(format!("Generate full state inverse diff failed: {e}")))
         }
     };
     peer_fork.diffs = vec![diff];
 
     // Retrieve the proposals of the hashes sequence, in batches
-    info!(target: "darkfid::task::handle_reorg", "Peer sequence ranks higher than our current best fork, retrieving {} proposals from peer...", peer_header_hashes.len());
+    info!(target: "darkfid::task::handle_reorg", "Peer sequence ranks higher than our current best fork, retrieving {} proposals from peer...", header_hashes.len());
     let mut batch = Vec::with_capacity(BATCH);
     let mut total_processed = 0;
-    for (index, hash) in peer_header_hashes.iter().enumerate() {
+    for (index, hash) in header_hashes.iter().enumerate() {
         // Add hash in batch sequence
         batch.push(*hash);
 
         // Check if batch is full so we can send it
-        if batch.len() < BATCH && index != peer_header_hashes.len() - 1 {
+        if batch.len() < BATCH && index != header_hashes.len() - 1 {
             continue
         }
 
         // Request peer proposals
         let request = ForkProposalsRequest { headers: batch.clone(), fork_header: proposal.hash };
-        if let Err(e) = channel.send(&request).await {
-            debug!(target: "darkfid::task::handle_reorg", "Channel send failed: {e}");
-            *validator.reorg.write().await = false;
-            return true
-        };
-
-        let comms_timeout =
-            p2p.settings().read_arc().await.outbound_connect_timeout(channel.address().scheme());
+        channel.0.send(&request).await?;
 
         // Node waits for response
-        let response = match response_sub.receive_with_timeout(comms_timeout).await {
-            Ok(r) => r,
-            Err(e) => {
-                debug!(target: "darkfid::task::handle_reorg", "Asking peer for proposals sequence failed: {e}");
-                *validator.reorg.write().await = false;
-                return true
-            }
-        };
+        let response = response_sub.receive_with_timeout(*channel.1).await?;
         debug!(target: "darkfid::task::handle_reorg", "Peer response: {response:?}");
 
         // Response sequence must be the same length as the one requested
         if response.proposals.len() != batch.len() {
-            debug!(target: "darkfid::task::handle_reorg", "Peer responded with a different proposals sequence length");
-            *validator.reorg.write().await = false;
-            return true
+            return Err(Custom(String::from(
+                "Peer responded with a different proposals sequence length",
+            )))
         }
 
         // Process retrieved proposal
@@ -582,113 +669,34 @@ async fn handle_reorg(
 
             // Validate its the proposal we requested
             if peer_proposal.hash != batch[peer_proposal_index] {
-                error!(target: "darkfid::task::handle_reorg", "Peer responded with a differend proposal: {} - {}", batch[peer_proposal_index], peer_proposal.hash);
-                *validator.reorg.write().await = false;
-                return true
+                return Err(Custom(format!(
+                    "Peer responded with a differend proposal: {} - {}",
+                    batch[peer_proposal_index], peer_proposal.hash
+                )))
             }
 
             // Verify proposal
-            if let Err(e) =
-                verify_fork_proposal(&mut peer_fork, peer_proposal, validator.verify_fees).await
-            {
-                error!(target: "darkfid::task::handle_reorg", "Verify fork proposal failed: {e}");
-                *validator.reorg.write().await = false;
-                return true
-            }
+            verify_fork_proposal(&mut peer_fork, peer_proposal, validator.verify_fees).await?;
 
             // Append proposal
-            if let Err(e) = peer_fork.append_proposal(peer_proposal).await {
-                error!(target: "darkfid::task::handle_reorg", "Appending proposal failed: {e}");
-                *validator.reorg.write().await = false;
-                return true
-            }
+            peer_fork.append_proposal(peer_proposal).await?;
         }
 
         total_processed += response.proposals.len();
-        info!(target: "darkfid::task::handle_reorg", "Proposals received and verified: {total_processed}/{}", peer_header_hashes.len());
+        info!(target: "darkfid::task::handle_reorg", "Proposals received and verified: {total_processed}/{}", header_hashes.len());
 
         // Reset batch
         batch = Vec::with_capacity(BATCH);
     }
 
     // Verify trigger proposal
-    if let Err(e) = verify_fork_proposal(&mut peer_fork, proposal, validator.verify_fees).await {
-        error!(target: "darkfid::task::handle_reorg", "Verify proposal failed: {e}");
-        *validator.reorg.write().await = false;
-        return true
-    }
+    verify_fork_proposal(&mut peer_fork, proposal, validator.verify_fees).await?;
 
     // Append trigger proposal
-    if let Err(e) = peer_fork.append_proposal(proposal).await {
-        error!(target: "darkfid::task::handle_reorg", "Appending proposal failed: {e}");
-        *validator.reorg.write().await = false;
-        return true
-    }
+    peer_fork.append_proposal(proposal).await?;
 
-    // Check if the peer fork ranks higher than our current best fork
-    let mut forks = validator.consensus.forks.write().await;
-    let index = match best_fork_index(&forks) {
-        Ok(i) => i,
-        Err(e) => {
-            debug!(target: "darkfid::task::handle_reorg", "Retrieving best fork index failed: {e}");
-            *validator.reorg.write().await = false;
-            return false
-        }
-    };
-    let best_fork = &forks[index];
-    if peer_fork.targets_rank < best_fork.targets_rank ||
-        (peer_fork.targets_rank == best_fork.targets_rank &&
-            peer_fork.hashes_rank <= best_fork.hashes_rank)
-    {
-        info!(target: "darkfid::task::handle_reorg", "Peer fork ranks lower than our current best fork, skipping...");
-        *validator.reorg.write().await = false;
-        return true
-    }
+    // Remove the reorg diff from the fork
+    peer_fork.diffs.remove(0);
 
-    // Execute the reorg
-    info!(target: "darkfid::task::handle_reorg", "Peer fork ranks higher than our current best fork, executing reorg...");
-    let result = peer_fork
-        .overlay
-        .lock()
-        .unwrap()
-        .overlay
-        .lock()
-        .unwrap()
-        .apply_diff(&peer_fork.diffs.remove(0));
-    if let Err(e) = result {
-        error!(target: "darkfid::task::handle_reorg", "Applying full inverse diff failed: {e}");
-        *validator.reorg.write().await = false;
-        return false
-    };
-    *validator.consensus.module.write().await = module;
-    *forks = vec![peer_fork];
-    *validator.reorg.write().await = false;
-    drop(forks);
-
-    // Check if we can confirm anything and broadcast them
-    let confirmed = match validator.confirmation().await {
-        Ok(f) => f,
-        Err(e) => {
-            error!(target: "darkfid::task::handle_reorg", "Confirmation failed: {e}");
-            return false
-        }
-    };
-
-    if !confirmed.is_empty() {
-        let mut notif_blocks = Vec::with_capacity(confirmed.len());
-        for block in confirmed {
-            notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
-        }
-        blocks_sub.notify(JsonValue::Array(notif_blocks)).await;
-    }
-
-    // Broadcast proposal to the network
-    let message = ProposalMessage(proposal.clone());
-    p2p.broadcast(&message).await;
-
-    // Notify proposals subscriber
-    let enc_prop = JsonValue::String(base64::encode(&serialize_async(proposal).await));
-    proposals_sub.notify(vec![enc_prop].into()).await;
-
-    false
+    Ok(peer_fork)
 }

src/validator/mod.rs

@@ -88,8 +88,6 @@ pub struct Validator {
     pub consensus: Consensus,
     /// Flag signalling if the node is synced
     pub synced: RwLock<bool>,
-    /// Flag signalling if the node is trying to execute a reorg
-    pub reorg: RwLock<bool>,
     /// Flag to enable tx fee verification
     pub verify_fees: bool,
 }
@@ -135,7 +133,6 @@ impl Validator {
             blockchain,
             consensus,
             synced: RwLock::new(false),
-            reorg: RwLock::new(false),
             verify_fees: config.verify_fees,
         });