darkfid: network reorg trigger added

bin/darkfid/src/proto/mod.rs

@@ -36,8 +36,10 @@ pub use protocol_proposal::{ProposalMessage, ProtocolProposalHandler, ProtocolPr
 /// Validator blockchain sync protocol
 mod protocol_sync;
 pub use protocol_sync::{
-    ForkSyncRequest, ForkSyncResponse, HeaderSyncRequest, HeaderSyncResponse, ProtocolSyncHandler,
-    ProtocolSyncHandlerPtr, SyncRequest, SyncResponse, TipRequest, TipResponse, BATCH,
+    ForkHeaderHashRequest, ForkHeaderHashResponse, ForkHeadersRequest, ForkHeadersResponse,
+    ForkProposalsRequest, ForkProposalsResponse, ForkSyncRequest, ForkSyncResponse,
+    HeaderSyncRequest, HeaderSyncResponse, ProtocolSyncHandler, ProtocolSyncHandlerPtr,
+    SyncRequest, SyncResponse, TipRequest, TipResponse, BATCH,
 };
 
 /// Transaction broadcast protocol

bin/darkfid/src/proto/protocol_sync.rs

@@ -131,6 +131,74 @@ pub struct ForkSyncResponse {
 
 impl_p2p_message!(ForkSyncResponse, "forksyncresponse");
 
+/// Structure represening a request to ask a node a fork header for the
+/// requested height. The fork is identified by the provided header hash.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkHeaderHashRequest {
+    /// Header height
+    pub height: u32,
+    /// Block header hash to identify the fork
+    pub fork_header: HeaderHash,
+}
+
+impl_p2p_message!(ForkHeaderHashRequest, "forkheaderhashrequest");
+
+/// Structure representing the response to `ForkHeaderHashRequest`,
+/// containing the requested fork header hash, if it was found.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkHeaderHashResponse {
+    /// Response fork block header hash
+    pub fork_header: Option<HeaderHash>,
+}
+
+impl_p2p_message!(ForkHeaderHashResponse, "forkheaderhashresponse");
+
+/// Structure represening a request to ask a node for up to `BATCH`
+/// fork headers for provided header hashes.  The fork is identified
+/// by the provided header hash.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkHeadersRequest {
+    /// Header hashes
+    pub headers: Vec<HeaderHash>,
+    /// Block header hash to identify the fork
+    pub fork_header: HeaderHash,
+}
+
+impl_p2p_message!(ForkHeadersRequest, "forkheadersrequest");
+
+/// Structure representing the response to `ForkHeadersRequest`,
+/// containing up to `BATCH` fork headers.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkHeadersResponse {
+    /// Response headers
+    pub headers: Vec<Header>,
+}
+
+impl_p2p_message!(ForkHeadersResponse, "forkheadersresponse");
+
+/// Structure represening a request to ask a node for up to `BATCH`
+/// fork proposals for provided header hashes.  The fork is identified
+/// by the provided header hash.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkProposalsRequest {
+    /// Header hashes
+    pub headers: Vec<HeaderHash>,
+    /// Block header hash to identify the fork
+    pub fork_header: HeaderHash,
+}
+
+impl_p2p_message!(ForkProposalsRequest, "forkproposalsrequest");
+
+/// Structure representing the response to `ForkProposalsRequest`,
+/// containing up to `BATCH` fork headers.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct ForkProposalsResponse {
+    /// Response proposals
+    pub proposals: Vec<Proposal>,
+}
+
+impl_p2p_message!(ForkProposalsResponse, "forkproposalsresponse");
+
 /// Atomic pointer to the `ProtocolSync` handler.
 pub type ProtocolSyncHandlerPtr = Arc<ProtocolSyncHandler>;
 
@@ -144,6 +212,13 @@ pub struct ProtocolSyncHandler {
     sync_handler: ProtocolGenericHandlerPtr<SyncRequest, SyncResponse>,
     /// The generic handler for `ForkSyncRequest` messages.
     fork_sync_handler: ProtocolGenericHandlerPtr<ForkSyncRequest, ForkSyncResponse>,
+    /// The generic handler for `ForkHeaderHashRequest` messages.
+    fork_header_hash_handler:
+        ProtocolGenericHandlerPtr<ForkHeaderHashRequest, ForkHeaderHashResponse>,
+    /// The generic handler for `ForkHeadersRequest` messages.
+    fork_headers_handler: ProtocolGenericHandlerPtr<ForkHeadersRequest, ForkHeadersResponse>,
+    /// The generic handler for `ForkProposalsRequest` messages.
+    fork_proposals_handler: ProtocolGenericHandlerPtr<ForkProposalsRequest, ForkProposalsResponse>,
 }
 
 impl ProtocolSyncHandler {
@@ -162,8 +237,22 @@ impl ProtocolSyncHandler {
         let sync_handler = ProtocolGenericHandler::new(p2p, "ProtocolSync", SESSION_DEFAULT).await;
         let fork_sync_handler =
             ProtocolGenericHandler::new(p2p, "ProtocolSyncFork", SESSION_DEFAULT).await;
-
-        Arc::new(Self { tip_handler, header_handler, sync_handler, fork_sync_handler })
+        let fork_header_hash_handler =
+            ProtocolGenericHandler::new(p2p, "ProtocolSyncForkHeaderHash", SESSION_DEFAULT).await;
+        let fork_headers_handler =
+            ProtocolGenericHandler::new(p2p, "ProtocolSyncForkHeaders", SESSION_DEFAULT).await;
+        let fork_proposals_handler =
+            ProtocolGenericHandler::new(p2p, "ProtocolSyncForkProposals", SESSION_DEFAULT).await;
+
+        Arc::new(Self {
+            tip_handler,
+            header_handler,
+            sync_handler,
+            fork_sync_handler,
+            fork_header_hash_handler,
+            fork_headers_handler,
+            fork_proposals_handler,
+        })
     }
 
     /// Start all `ProtocolSync` background tasks.
@@ -221,6 +310,42 @@ impl ProtocolSyncHandler {
             executor.clone(),
         );
 
+        self.fork_header_hash_handler.task.clone().start(
+            handle_receive_fork_header_hash_request(self.fork_header_hash_handler.clone(), validator.clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "darkfid::proto::protocol_sync::start", "Failed starting ProtocolSyncForkHeaderHash handler task: {e}"),
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
+        self.fork_headers_handler.task.clone().start(
+            handle_receive_fork_headers_request(self.fork_headers_handler.clone(), validator.clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "darkfid::proto::protocol_sync::start", "Failed starting ProtocolSyncForkHeaders handler task: {e}"),
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
+        self.fork_proposals_handler.task.clone().start(
+            handle_receive_fork_proposals_request(self.fork_proposals_handler.clone(), validator.clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "darkfid::proto::protocol_sync::start", "Failed starting ProtocolSyncForkProposals handler task: {e}"),
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
         debug!(
             target: "darkfid::proto::protocol_sync::start",
             "Sync protocols handlers tasks started!"
@@ -236,6 +361,9 @@ impl ProtocolSyncHandler {
         self.header_handler.task.stop().await;
         self.sync_handler.task.stop().await;
         self.fork_sync_handler.task.stop().await;
+        self.fork_header_hash_handler.task.stop().await;
+        self.fork_headers_handler.task.stop().await;
+        self.fork_proposals_handler.task.stop().await;
         debug!(target: "darkfid::proto::protocol_sync::stop", "Sync protocols handlers tasks terminated!");
     }
 }
@@ -259,50 +387,73 @@ async fn handle_receive_tip_request(
             }
         };
 
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_tip_request", "Received request: {request:?}");
+
         // Check if node has finished syncing its blockchain
-        let response = if !*validator.synced.read().await {
-            TipResponse { synced: false, height: None, hash: None }
-        } else {
-            // Check we follow the same sequence
-            match validator.blockchain.blocks.contains(&request.tip) {
-                Ok(contains) => {
-                    if !contains {
-                        debug!(
-                            target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                            "Node doesn't follow request sequence"
-                        );
-                        handler.send_action(channel, ProtocolGenericAction::Skip).await;
-                        continue
-                    }
-                }
-                Err(e) => {
-                    error!(
-                        target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                        "block_store.contains fail: {e}"
-                    );
-                    handler.send_action(channel, ProtocolGenericAction::Skip).await;
-                    continue
-                }
-            }
+        if !*validator.synced.read().await {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
+                "Node still syncing blockchain"
+            );
+            handler
+                .send_action(
+                    channel,
+                    ProtocolGenericAction::Response(TipResponse {
+                        synced: false,
+                        height: None,
+                        hash: None,
+                    }),
+                )
+                .await;
+            continue
+        }
 
-            // Grab our current tip and return it
-            let tip = match validator.blockchain.last() {
-                Ok(v) => v,
-                Err(e) => {
-                    error!(
+        // Check we follow the same sequence
+        match validator.blockchain.blocks.contains(&request.tip) {
+            Ok(contains) => {
+                if !contains {
+                    debug!(
                         target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                        "blockchain.last fail: {e}"
+                        "Node doesn't follow request sequence"
                     );
                     handler.send_action(channel, ProtocolGenericAction::Skip).await;
                     continue
                 }
-            };
+            }
+            Err(e) => {
+                error!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
+                    "block_store.contains fail: {e}"
+                );
+                handler.send_action(channel, ProtocolGenericAction::Skip).await;
+                continue
+            }
+        }
 
-            TipResponse { synced: true, height: Some(tip.0), hash: Some(tip.1) }
+        // Grab our current tip and return it
+        let tip = match validator.blockchain.last() {
+            Ok(v) => v,
+            Err(e) => {
+                error!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
+                    "blockchain.last fail: {e}"
+                );
+                handler.send_action(channel, ProtocolGenericAction::Skip).await;
+                continue
+            }
         };
 
         // Send response
-        handler.send_action(channel, ProtocolGenericAction::Response(response)).await;
+        handler
+            .send_action(
+                channel,
+                ProtocolGenericAction::Response(TipResponse {
+                    synced: true,
+                    height: Some(tip.0),
+                    hash: Some(tip.1),
+                }),
+            )
+            .await;
     }
 }
 
@@ -335,6 +486,8 @@ async fn handle_receive_header_request(
             continue
         }
 
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_header_request", "Received request: {request:?}");
+
         // Grab the corresponding headers
         let headers = match validator.blockchain.get_headers_before(request.height, BATCH) {
             Ok(v) => v,
@@ -395,6 +548,8 @@ async fn handle_receive_request(
             continue
         }
 
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_request", "Received request: {request:?}");
+
         // Grab the corresponding blocks
         let blocks = match validator.blockchain.get_blocks_by_hash(&request.headers) {
             Ok(v) => v,
@@ -447,15 +602,12 @@ async fn handle_receive_fork_request(
 
         debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_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 {
-            Some(fork_tip) => {
-                validator.consensus.get_fork_proposals(request.tip, fork_tip, BATCH as u32).await
-            }
-            None => validator.consensus.get_best_fork_proposals(request.tip, BATCH as u32).await,
-        };
-        let proposals = match proposals {
+        // Retrieve proposals sequence
+        let proposals = match validator
+            .consensus
+            .get_fork_proposals_after(request.tip, request.fork_tip, BATCH as u32)
+            .await
+        {
             Ok(p) => p,
             Err(e) => {
                 debug!(
@@ -474,3 +626,197 @@ async fn handle_receive_fork_request(
             .await;
     }
 }
+
+/// Background handler function for ProtocolSyncForkHeaderHash.
+async fn handle_receive_fork_header_hash_request(
+    handler: ProtocolGenericHandlerPtr<ForkHeaderHashRequest, ForkHeaderHashResponse>,
+    validator: ValidatorPtr,
+) -> Result<()> {
+    debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request", "START");
+    loop {
+        // Wait for a new fork header hash request message
+        let (channel, request) = match handler.receiver.recv().await {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request",
+                    "recv fail: {e}"
+                );
+                continue
+            }
+        };
+
+        // Check if node has finished syncing its blockchain
+        if !*validator.synced.read().await {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request",
+                "Node still syncing blockchain, skipping..."
+            );
+            handler.send_action(channel, ProtocolGenericAction::Skip).await;
+            continue
+        }
+
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request", "Received request: {request:?}");
+
+        // Retrieve fork header
+        let fork_header = match validator
+            .consensus
+            .get_fork_header_hash(request.height, &request.fork_header)
+            .await
+        {
+            Ok(h) => h,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_header_hash_request",
+                    "Getting fork header hash failed: {}",
+                    e
+                );
+                handler.send_action(channel, ProtocolGenericAction::Skip).await;
+                continue
+            }
+        };
+
+        // Send response
+        handler
+            .send_action(
+                channel,
+                ProtocolGenericAction::Response(ForkHeaderHashResponse { fork_header }),
+            )
+            .await;
+    }
+}
+
+/// Background handler function for ProtocolSyncForkHeaders.
+async fn handle_receive_fork_headers_request(
+    handler: ProtocolGenericHandlerPtr<ForkHeadersRequest, ForkHeadersResponse>,
+    validator: ValidatorPtr,
+) -> Result<()> {
+    debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request", "START");
+    loop {
+        // Wait for a new fork header hash request message
+        let (channel, request) = match handler.receiver.recv().await {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request",
+                    "recv fail: {e}"
+                );
+                continue
+            }
+        };
+
+        // Check if node has finished syncing its blockchain
+        if !*validator.synced.read().await {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request",
+                "Node still syncing blockchain, skipping..."
+            );
+            handler.send_action(channel, ProtocolGenericAction::Skip).await;
+            continue
+        }
+
+        // Check if request exists the configured limit
+        if request.headers.len() > BATCH {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request",
+                "Node requested more headers than allowed."
+            );
+            handler.send_action(channel, ProtocolGenericAction::Skip).await;
+            continue
+        }
+
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request", "Received request: {request:?}");
+
+        // Retrieve fork headers
+        let headers = match validator
+            .consensus
+            .get_fork_headers(&request.headers, &request.fork_header)
+            .await
+        {
+            Ok(h) => h,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_headers_request",
+                    "Getting fork headers failed: {}",
+                    e
+                );
+                handler.send_action(channel, ProtocolGenericAction::Skip).await;
+                continue
+            }
+        };
+
+        // Send response
+        handler
+            .send_action(channel, ProtocolGenericAction::Response(ForkHeadersResponse { headers }))
+            .await;
+    }
+}
+
+/// Background handler function for ProtocolSyncForkProposals.
+async fn handle_receive_fork_proposals_request(
+    handler: ProtocolGenericHandlerPtr<ForkProposalsRequest, ForkProposalsResponse>,
+    validator: ValidatorPtr,
+) -> Result<()> {
+    debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request", "START");
+    loop {
+        // Wait for a new fork header hash request message
+        let (channel, request) = match handler.receiver.recv().await {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request",
+                    "recv fail: {e}"
+                );
+                continue
+            }
+        };
+
+        // Check if node has finished syncing its blockchain
+        if !*validator.synced.read().await {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request",
+                "Node still syncing blockchain, skipping..."
+            );
+            handler.send_action(channel, ProtocolGenericAction::Skip).await;
+            continue
+        }
+
+        // Check if request exists the configured limit
+        if request.headers.len() > BATCH {
+            debug!(
+                target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request",
+                "Node requested more proposals than allowed."
+            );
+            handler.send_action(channel, ProtocolGenericAction::Skip).await;
+            continue
+        }
+
+        debug!(target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request", "Received request: {request:?}");
+
+        // Retrieve fork headers
+        let proposals = match validator
+            .consensus
+            .get_fork_proposals(&request.headers, &request.fork_header)
+            .await
+        {
+            Ok(p) => p,
+            Err(e) => {
+                debug!(
+                    target: "darkfid::proto::protocol_sync::handle_receive_fork_proposals_request",
+                    "Getting fork proposals failed: {}",
+                    e
+                );
+                handler.send_action(channel, ProtocolGenericAction::Skip).await;
+                continue
+            }
+        };
+
+        // Send response
+        handler
+            .send_action(
+                channel,
+                ProtocolGenericAction::Response(ForkProposalsResponse { proposals }),
+            )
+            .await;
+    }
+}

bin/darkfid/src/task/unknown_proposal.rs

@@ -16,19 +16,30 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use log::{debug, error, warn};
+use log::{debug, error, info, warn};
 use tinyjson::JsonValue;
 
 use darkfi::{
-    net::P2pPtr,
+    blockchain::BlockDifficulty,
+    net::{ChannelPtr, P2pPtr},
     rpc::jsonrpc::JsonSubscriber,
     util::encoding::base64,
-    validator::{consensus::Proposal, ValidatorPtr},
+    validator::{
+        consensus::{Fork, Proposal},
+        pow::PoWModule,
+        utils::{best_fork_index, header_rank},
+        verification::verify_fork_proposal,
+        ValidatorPtr,
+    },
     Error, Result,
 };
 use darkfi_serial::serialize_async;
 
-use crate::proto::{ForkSyncRequest, ForkSyncResponse, ProposalMessage};
+use crate::proto::{
+    ForkHeaderHashRequest, ForkHeaderHashResponse, ForkHeadersRequest, ForkHeadersResponse,
+    ForkProposalsRequest, ForkProposalsResponse, ForkSyncRequest, ForkSyncResponse,
+    ProposalMessage, BATCH,
+};
 
 /// Background task to handle unknown proposals.
 pub async fn handle_unknown_proposal(
@@ -84,25 +95,25 @@ pub async fn handle_unknown_proposal(
     // Response should not be empty
     if response.proposals.is_empty() {
         warn!(target: "darkfid::task::handle_unknown_proposal", "Peer responded with empty sequence, node might be out of sync!");
-        return Ok(())
+        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
     }
 
     // Sequence length must correspond to requested height
     if response.proposals.len() as u32 != proposal.block.header.height - last.0 {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence length is erroneous");
-        return Ok(())
+        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
     }
 
     // First proposal must extend canonical
     if response.proposals[0].block.header.previous != last.1 {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence doesn't extend canonical");
-        return Ok(())
+        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
     }
 
     // Last proposal must be the same as the one requested
     if response.proposals.last().unwrap().hash != proposal.hash {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence doesn't correspond to requested tip");
-        return Ok(())
+        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
     }
 
     // Process response proposals
@@ -132,3 +143,356 @@ pub async fn handle_unknown_proposal(
 
     Ok(())
 }
+
+// TODO; If a reorg trigger is erroneous, disconnect from peer.
+/// Auxiliary function to handle a potential reorg.
+/// We first find our last common block with the peer,
+/// then grab the header sequence from that block until
+/// the proposal and check if it ranks higher than our
+/// current best ranking fork, to perform a reorg.
+async fn handle_reorg(
+    validator: ValidatorPtr,
+    p2p: P2pPtr,
+    subscriber: JsonSubscriber,
+    channel: ChannelPtr,
+    proposal: Proposal,
+) -> Result<()> {
+    info!(target: "darkfid::task::handle_reorg", "Checking for potential reorg from proposal {} - {} by peer: {channel:?}", proposal.hash, proposal.block.header.height);
+
+    // Check if genesis proposal was provided
+    if proposal.block.header.height == 0 {
+        info!(target: "darkfid::task::handle_reorg", "Peer send a genesis proposal, skipping...");
+        return Ok(())
+    }
+
+    // Communication setup
+    let Ok(response_sub) = channel.subscribe_msg::<ForkHeaderHashResponse>().await else {
+        error!(target: "darkfid::task::handle_reorg", "Failure during `ForkHeaderHashResponse` communication setup with peer: {channel:?}");
+        return Ok(())
+    };
+
+    // Keep track of received header hashes sequence
+    let mut peer_header_hashes = vec![];
+
+    // Find last common header, going backwards from the proposal
+    let mut previous_height = proposal.block.header.height;
+    let mut previous_hash = proposal.hash;
+    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 Ok(())
+        };
+
+        // Node waits for response
+        let response = match response_sub
+            .receive_with_timeout(p2p.settings().read().await.outbound_connect_timeout)
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(target: "darkfid::task::handle_reorg", "Asking peer for header hash failed: {e}");
+                return Ok(())
+            }
+        };
+        debug!(target: "darkfid::task::handle_reorg", "Peer response: {response:?}");
+
+        // Check if peer returned a header
+        let Some(peer_header) = response.fork_header else {
+            info!(target: "darkfid::task::handle_reorg", "Peer responded with an empty header");
+            return Ok(())
+        };
+
+        // Check if we know this header
+        match validator.blockchain.blocks.get_order(&[height], false)?[0] {
+            Some(known_header) => {
+                if known_header == peer_header {
+                    previous_height = height;
+                    previous_hash = known_header;
+                    break
+                }
+                // Since we retrieve in right -> left order we push them in reverse order
+                peer_header_hashes.insert(0, peer_header);
+            }
+            None => peer_header_hashes.insert(0, peer_header),
+        }
+    }
+
+    // Check if we have a sequence to process
+    if peer_header_hashes.is_empty() {
+        info!(target: "darkfid::task::handle_reorg", "No headers to process, skipping...");
+        return Ok(())
+    }
+
+    // Communication setup
+    let Ok(response_sub) = channel.subscribe_msg::<ForkHeadersResponse>().await else {
+        error!(target: "darkfid::task::handle_reorg", "Failure during `ForkHeadersResponse` communication setup with peer: {channel:?}");
+        return Ok(())
+    };
+
+    // Grab last common height ranks
+    let last_common_height = previous_height;
+    let last_difficulty = match previous_height {
+        0 => BlockDifficulty::genesis(validator.blockchain.genesis_block()?.header.timestamp),
+        _ => validator.blockchain.blocks.get_difficulty(&[last_common_height], true)?[0]
+            .clone()
+            .unwrap(),
+    };
+
+    // Create a new PoW from last common height
+    let module = 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),
+    )?;
+
+    // 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());
+    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() {
+        // 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 {
+            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 Ok(())
+        };
+
+        // Node waits for response
+        let response = match response_sub
+            .receive_with_timeout(p2p.settings().read().await.outbound_connect_timeout)
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(target: "darkfid::task::handle_reorg", "Asking peer for headers sequence failed: {e}");
+                return Ok(())
+            }
+        };
+        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() {
+            error!(target: "darkfid::task::handle_reorg", "Peer responded with a different headers sequence length");
+            return Ok(())
+        }
+
+        // Process retrieved headers
+        for (peer_header_index, peer_header) in response.headers.iter().enumerate() {
+            let peer_header_hash = peer_header.hash();
+            info!(target: "darkfid::task::handle_reorg", "Processing header: {peer_header_hash} - {}", peer_header.height);
+
+            // Validate its the header we requested
+            if peer_header_hash != batch[peer_header_index] {
+                error!(target: "darkfid::task::handle_reorg", "Peer responded with a differend header: {} - {peer_header_hash}", batch[peer_header_index]);
+                return Ok(())
+            }
+
+            // Validate sequence is correct
+            if peer_header.previous != previous_hash || peer_header.height != previous_height + 1 {
+                error!(target: "darkfid::task::handle_reorg", "Invalid header sequence detected");
+                return Ok(())
+            }
+
+            // Grab next mine target and difficulty
+            let (next_target, next_difficulty) =
+                headers_module.next_mine_target_and_difficulty()?;
+
+            // Verify header hash and calculate its rank
+            let (target_distance_sq, hash_distance_sq) = match header_rank(
+                peer_header,
+                &next_target,
+            ) {
+                Ok(distances) => distances,
+                Err(e) => {
+                    error!(target: "darkfid::task::handle_reorg", "Invalid header hash detected: {e}");
+                    return Ok(())
+                }
+            };
+
+            // Update sequence ranking
+            targets_rank += target_distance_sq.clone();
+            hashes_rank += hash_distance_sq.clone();
+
+            // Update PoW headers module
+            headers_module.append(peer_header.timestamp, &next_difficulty);
+
+            // Set previous header
+            previous_height = peer_header.height;
+            previous_hash = peer_header_hash;
+        }
+
+        total_processed += response.headers.len();
+        info!(target: "darkfid::task::handle_reorg", "Headers received and verified: {total_processed}/{}", peer_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 best_fork = &forks[best_fork_index(&forks)?];
+    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...");
+        drop(forks);
+        return Ok(())
+    }
+    drop(forks);
+
+    // Communication setup
+    let Ok(response_sub) = channel.subscribe_msg::<ForkProposalsResponse>().await else {
+        error!(target: "darkfid::task::handle_reorg", "Failure during `ForkProposalsResponse` communication setup with peer: {channel:?}");
+        return Ok(())
+    };
+
+    // Create a fork from last common height
+    let mut peer_fork = Fork::new(validator.consensus.blockchain.clone(), module).await?;
+    peer_fork.targets_rank = last_difficulty.ranks.targets_rank.clone();
+    peer_fork.hashes_rank = last_difficulty.ranks.hashes_rank.clone();
+
+    // Grab all state diffs after last common height and add their inverse to the fork
+    let diffs = validator.blockchain.blocks.get_state_diffs_after(last_common_height)?;
+    for diff in diffs.iter().rev() {
+        peer_fork.overlay.lock().unwrap().overlay.lock().unwrap().add_diff(&diff.inverse())?;
+    }
+
+    // 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());
+    let mut batch = Vec::with_capacity(BATCH);
+    let mut total_processed = 0;
+    for (index, hash) in peer_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 {
+            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}");
+            return Ok(())
+        };
+
+        // Node waits for response
+        let response = match response_sub
+            .receive_with_timeout(p2p.settings().read().await.outbound_connect_timeout)
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(target: "darkfid::task::handle_reorg", "Asking peer for proposals sequence failed: {e}");
+                return Ok(())
+            }
+        };
+        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() {
+            error!(target: "darkfid::task::handle_reorg", "Peer responded with a different proposals sequence length");
+            return Ok(())
+        }
+
+        // Process retrieved proposal
+        for (peer_proposal_index, peer_proposal) in response.proposals.iter().enumerate() {
+            info!(target: "darkfid::task::handle_reorg", "Processing proposal: {} - {}", peer_proposal.hash, peer_proposal.block.header.height);
+
+            // 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);
+                return Ok(())
+            }
+
+            // Verify proposal
+            if let Err(e) =
+                verify_fork_proposal(&peer_fork, peer_proposal, validator.verify_fees).await
+            {
+                error!(target: "darkfid::task::handle_reorg", "Verify fork proposal failed: {e}");
+                return Ok(())
+            }
+
+            // Append proposal
+            if let Err(e) = peer_fork.append_proposal(peer_proposal).await {
+                error!(target: "darkfid::task::handle_reorg", "Appending proposal failed: {e}");
+                return Ok(())
+            }
+        }
+
+        total_processed += response.proposals.len();
+        info!(target: "darkfid::task::handle_reorg", "Proposals received and verified: {total_processed}/{}", peer_header_hashes.len());
+
+        // Reset batch
+        batch = Vec::with_capacity(BATCH);
+    }
+
+    // Verify trigger proposal
+    if let Err(e) = verify_fork_proposal(&peer_fork, &proposal, validator.verify_fees).await {
+        error!(target: "darkfid::task::handle_reorg", "Verify proposal failed: {e}");
+        return Ok(())
+    }
+
+    // Append trigger proposal
+    if let Err(e) = peer_fork.append_proposal(&proposal).await {
+        error!(target: "darkfid::task::handle_reorg", "Appending proposal failed: {e}");
+        return Ok(())
+    }
+
+    // Check if the peer fork ranks higher than our current best fork
+    let mut forks = validator.consensus.forks.write().await;
+    let best_fork = &forks[best_fork_index(&forks)?];
+    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...");
+        drop(forks);
+        return Ok(())
+    }
+
+    // Execute the reorg
+    info!(target: "darkfid::task::handle_reorg", "Peer fork ranks higher than our current best fork, executing reorg...");
+    *forks = vec![peer_fork];
+    drop(forks);
+
+    // Check if we can finalize anything and broadcast them
+    let finalized = match validator.finalization().await {
+        Ok(f) => f,
+        Err(e) => {
+            error!(target: "darkfid::task::handle_reorg", "Finalization failed: {e}");
+            return Ok(())
+        }
+    };
+
+    if finalized.is_empty() {
+        return Ok(())
+    }
+
+    let mut notif_blocks = Vec::with_capacity(finalized.len());
+    for block in finalized {
+        notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+    }
+    subscriber.notify(JsonValue::Array(notif_blocks)).await;
+
+    // Broadcast proposal to the network
+    let message = ProposalMessage(proposal);
+    p2p.broadcast(&message).await;
+
+    Ok(())
+}

bin/darkfid/src/tests/forks.rs

@@ -39,7 +39,7 @@ fn forks() -> Result<()> {
         let genesis_block_hash = genesis_block.hash();
 
         // Generate the PoW module
-        let module = PoWModule::new(blockchain.clone(), 90, None)?;
+        let module = PoWModule::new(blockchain.clone(), 90, None, None)?;
 
         // Create a fork
         let fork = Fork::new(blockchain.clone(), module).await?;

src/blockchain/block_store.rs

@@ -167,7 +167,7 @@ pub struct BlockOrder {
 /// Note: we only need height cummulative ranks, but we also keep its actual
 /// ranks, so we can verify the sequence and/or know specific block height
 /// ranks, if ever needed.
-#[derive(Debug, SerialEncodable, SerialDecodable)]
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 pub struct BlockRanks {
     /// Block target rank
     pub target_rank: BigUint,
@@ -195,7 +195,7 @@ impl BlockRanks {
 /// Note: we only need height cummulative difficulty, but we also keep its actual
 /// difficulty, so we can verify the sequence and/or know specific block height
 /// difficulty, if ever needed.
-#[derive(Debug, SerialEncodable, SerialDecodable)]
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 pub struct BlockDifficulty {
     /// Block height number
     pub height: u32,
@@ -548,17 +548,19 @@ impl BlockStore {
         let mut key = height;
         let mut counter = 0;
         while counter < n {
-            if let Some(found) = self.order.get_lt(key.to_be_bytes())? {
-                let (height, hash) = parse_u32_key_record(found)?;
-                key = height;
-                ret.push(hash);
-                counter += 1;
-                continue
+            let record = self.order.get_lt(key.to_be_bytes())?;
+            if record.is_none() {
+                break
             }
-            break
+            // Since the iterator grabs in right -> left order,
+            // we deserialize found records, and push them in reverse order
+            let (height, hash) = parse_u32_key_record(record.unwrap())?;
+            key = height;
+            ret.insert(0, hash);
+            counter += 1;
         }
 
-        Ok(ret.iter().rev().copied().collect())
+        Ok(ret)
     }
 
     /// Fetch all hashes after given height. In the iteration, if an order
@@ -634,6 +636,46 @@ impl BlockStore {
         Ok(last_n)
     }
 
+    /// Fetch N records before given height from the store's difficulty tree, in order.
+    /// In the iteration, if a record height is not found, the iteration stops and the
+    /// function returns what it has found so far in the store's difficulty tree.
+    pub fn get_difficulties_before(&self, height: u32, n: usize) -> Result<Vec<BlockDifficulty>> {
+        let mut ret = vec![];
+
+        let mut key = height;
+        let mut counter = 0;
+        while counter < n {
+            let record = self.difficulty.get_lt(key.to_be_bytes())?;
+            if record.is_none() {
+                break
+            }
+            // Since the iterator grabs in right -> left order,
+            // we deserialize found records, and push them in reverse order
+            let (height, difficulty) = parse_u32_key_record(record.unwrap())?;
+            key = height;
+            ret.insert(0, difficulty);
+            counter += 1;
+        }
+
+        Ok(ret)
+    }
+
+    /// Fetch all state diffs after given height. In the iteration, if a state
+    /// diff is not found, the iteration stops and the function returns what
+    /// it has found so far in the store's state diffs tree.
+    pub fn get_state_diffs_after(&self, height: u32) -> Result<Vec<SledDbOverlayStateDiff>> {
+        let mut ret = vec![];
+
+        let mut key = height;
+        while let Some(found) = self.state_diff.get_gt(key.to_be_bytes())? {
+            let (height, state_diff) = parse_u32_key_record(found)?;
+            key = height;
+            ret.push(state_diff);
+        }
+
+        Ok(ret)
+    }
+
     /// Retrieve store's order tree records count.
     pub fn len(&self) -> usize {
         self.order.len()

src/blockchain/mod.rs

@@ -514,6 +514,14 @@ impl BlockchainOverlay {
         Ok(blockhash == block.hash())
     }
 
+    /// Retrieve [`Header`]s by given hashes. Fails if any of them is not found.
+    pub fn get_headers_by_hash(&self, hashes: &[HeaderHash]) -> Result<Vec<Header>> {
+        let headers = self.headers.get(hashes, true)?;
+        let ret: Vec<Header> = headers.iter().map(|x| x.clone().unwrap()).collect();
+
+        Ok(ret)
+    }
+
     /// Retrieve [`BlockInfo`]s by given hashes. Fails if any of them is not found.
     pub fn get_blocks_by_hash(&self, hashes: &[HeaderHash]) -> Result<Vec<BlockInfo>> {
         let blocks = self.blocks.get(hashes, true)?;

src/validator/consensus.rs

@@ -28,7 +28,7 @@ use smol::lock::RwLock;
 use crate::{
     blockchain::{
         block_store::{BlockDifficulty, BlockRanks},
-        BlockInfo, Blockchain, BlockchainOverlay, BlockchainOverlayPtr, HeaderHash,
+        BlockInfo, Blockchain, BlockchainOverlay, BlockchainOverlayPtr, Header, HeaderHash,
     },
     tx::Transaction,
     validator::{
@@ -73,8 +73,12 @@ impl Consensus {
         pow_fixed_difficulty: Option<BigUint>,
     ) -> Result<Self> {
         let forks = RwLock::new(vec![]);
-        let module =
-            RwLock::new(PoWModule::new(blockchain.clone(), pow_target, pow_fixed_difficulty)?);
+        let module = RwLock::new(PoWModule::new(
+            blockchain.clone(),
+            pow_target,
+            pow_fixed_difficulty,
+            None,
+        )?);
         let append_lock = RwLock::new(());
         Ok(Self { blockchain, finalization_threshold, forks, module, append_lock })
     }
@@ -114,6 +118,16 @@ impl Consensus {
                 }
             }
         }
+        // Check if proposal is canonical
+        if let Ok(canonical_headers) =
+            self.blockchain.blocks.get_order(&[proposal.block.header.height], true)
+        {
+            if canonical_headers[0].unwrap() == proposal.hash {
+                drop(lock);
+                debug!(target: "validator::consensus::append_proposal", "Proposal {} already exists", proposal.hash);
+                return Err(Error::ProposalAlreadyExists)
+            }
+        }
         drop(lock);
 
         // Verify proposal and grab corresponding fork
@@ -252,157 +266,203 @@ impl Consensus {
         Ok(Some(index))
     }
 
-    /// 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(
+    /// Auxiliary function to retrieve the fork header hash of provided height.
+    /// The fork is identified by the provided header hash.
+    pub async fn get_fork_header_hash(
         &self,
-        tip: HeaderHash,
-        fork_tip: HeaderHash,
-        limit: u32,
-    ) -> Result<Vec<Proposal>> {
+        height: u32,
+        fork_header: &HeaderHash,
+    ) -> Result<Option<HeaderHash>> {
         // Grab a lock over current forks
         let forks = self.forks.read().await;
 
-        // 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)
+        // Find the fork containing the provided header
+        let mut found = None;
+        'outer: for (index, fork) in forks.iter().enumerate() {
+            for p in fork.proposals.iter().rev() {
+                if p == fork_header {
+                    found = Some(index);
+                    break 'outer
+                }
             }
+        }
+        if found.is_none() {
+            drop(forks);
+            return Ok(None)
+        }
+        let index = found.unwrap();
 
-            // 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)?;
+        // Grab header if it exists
+        let header = forks[index].overlay.lock().unwrap().blocks.get_order(&[height], false)?[0];
 
-            // Add everything to the return vec
-            for block in blocks {
-                canonical_blocks.push(Proposal::new(block));
-            }
-        }
+        // Drop forks lock
+        drop(forks);
 
-        // Find the fork containing the requested tip and grab its sequence
-        let mut proposals = vec![];
-        for fork in forks.iter() {
-            let mut found = false;
-            for p in fork.proposals.iter().rev() {
-                if p != &fork_tip {
-                    continue
-                }
-                found = true;
-                break
-            }
+        Ok(header)
+    }
 
-            if !found {
-                continue
-            }
+    /// Auxiliary function to retrieve the fork headers of provided hashes.
+    /// The fork is identified by the provided header hash. If fork doesn't
+    /// exists, an empty vector is returned.
+    pub async fn get_fork_headers(
+        &self,
+        headers: &[HeaderHash],
+        fork_header: &HeaderHash,
+    ) -> Result<Vec<Header>> {
+        // Grab a lock over current forks
+        let forks = self.forks.read().await;
 
-            let mut headers = vec![];
-            for p in &fork.proposals {
-                headers.push(*p);
-                if p == &fork_tip {
-                    break
+        // Find the fork containing the provided header
+        let mut found = None;
+        'outer: for (index, fork) in forks.iter().enumerate() {
+            for p in fork.proposals.iter().rev() {
+                if p == fork_header {
+                    found = Some(index);
+                    break 'outer
                 }
             }
-
-            let blocks = fork.overlay.lock().unwrap().get_blocks_by_hash(&headers)?;
-            for block in blocks {
-                proposals.push(Proposal::new(block));
-            }
         }
-
-        // 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() {
+        let Some(index) = found else {
             drop(forks);
-            return Ok(proposals)
-        }
+            return Ok(vec![])
+        };
+
+        // Grab headers
+        let headers = forks[index].overlay.lock().unwrap().get_headers_by_hash(headers)?;
 
-        // Join the two vectors and return them
-        canonical_blocks.append(&mut proposals);
+        // Drop forks lock
         drop(forks);
-        Ok(canonical_blocks)
+
+        Ok(headers)
     }
 
-    /// 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)> {
+    /// Auxiliary function to retrieve the fork proposals of provided hashes.
+    /// The fork is identified by the provided header hash. If fork doesn't
+    /// exists, an empty vector is returned.
+    pub async fn get_fork_proposals(
+        &self,
+        headers: &[HeaderHash],
+        fork_header: &HeaderHash,
+    ) -> Result<Vec<Proposal>> {
         // 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()
+        // Find the fork containing the provided header
+        let mut found = None;
+        'outer: for (index, fork) in forks.iter().enumerate() {
+            for p in fork.proposals.iter().rev() {
+                if p == fork_header {
+                    found = Some(index);
+                    break 'outer
+                }
+            }
         }
+        let Some(index) = found else {
+            drop(forks);
+            return Ok(vec![])
+        };
 
-        // Grab best fork
-        let fork = &forks[best_fork_index(&forks)?];
+        // Grab proposals
+        let blocks = forks[index].overlay.lock().unwrap().get_blocks_by_hash(headers)?;
+        let mut proposals = Vec::with_capacity(blocks.len());
+        for block in blocks {
+            proposals.push(Proposal::new(block));
+        }
 
-        // Grab its last header
-        let last = fork.last_proposal()?;
+        // Drop forks lock
         drop(forks);
-        Ok((last.block.header.height, last.hash))
+
+        Ok(proposals)
     }
 
-    /// 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(
+    /// Auxiliary function to retrieve a fork proposals, starting from provided tip.
+    /// If provided tip is too far behind, unknown, or fork doesn't exists, an empty
+    /// vector is returned. The fork is identified by the optional provided header hash.
+    /// If its `None`, we use our best fork.
+    pub async fn get_fork_proposals_after(
         &self,
         tip: HeaderHash,
+        fork_tip: Option<HeaderHash>,
         limit: u32,
     ) -> Result<Vec<Proposal>> {
         // Grab a lock over current forks
         let forks = self.forks.read().await;
 
-        // Check if node has any forks
-        if forks.is_empty() {
+        // Create return vector
+        let mut proposals = vec![];
+
+        // Grab fork index to use
+        let index = match fork_tip {
+            Some(fork_tip) => {
+                let mut found = None;
+                'outer: for (index, fork) in forks.iter().enumerate() {
+                    for p in fork.proposals.iter().rev() {
+                        if p == &fork_tip {
+                            found = Some(index);
+                            break 'outer
+                        }
+                    }
+                }
+                if found.is_none() {
+                    drop(forks);
+                    return Ok(proposals)
+                }
+                found.unwrap()
+            }
+            None => best_fork_index(&forks)?,
+        };
+
+        // Check tip exists
+        let Ok(existing_tips) = forks[index].overlay.lock().unwrap().get_blocks_by_hash(&[tip])
+        else {
             drop(forks);
-            return Ok(vec![])
+            return Ok(proposals)
+        };
+
+        // Check tip is not far behind
+        let last_block_height = forks[index].overlay.lock().unwrap().last()?.0;
+        if last_block_height - existing_tips[0].header.height >= limit {
+            drop(forks);
+            return Ok(proposals)
+        }
+
+        // Retrieve all proposals after requested one
+        let headers = self.blockchain.blocks.get_all_after(existing_tips[0].header.height)?;
+        let blocks = self.blockchain.get_blocks_by_hash(&headers)?;
+        for block in blocks {
+            proposals.push(Proposal::new(block));
+        }
+        let blocks =
+            forks[index].overlay.lock().unwrap().get_blocks_by_hash(&forks[index].proposals)?;
+        for block in blocks {
+            proposals.push(Proposal::new(block));
         }
 
-        // Retrieve our current canonical tip height
-        let last_block_height = self.blockchain.last()?.0;
+        // Drop forks lock
+        drop(forks);
 
-        // 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)
-            }
+        Ok(proposals)
+    }
 
-            // 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)?;
+    /// 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;
 
-            // Add everything to the return vec
-            for block in blocks {
-                canonical_blocks.push(Proposal::new(block));
-            }
+        // 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 proposals
-        let blocks = fork.overlay.lock().unwrap().get_blocks_by_hash(&fork.proposals)?;
-        let mut proposals = Vec::with_capacity(blocks.len());
-        for block in blocks {
-            proposals.push(Proposal::new(block));
-        }
-
-        // Join the two vectors and return them
-        canonical_blocks.append(&mut proposals);
+        // Grab its last header
+        let last = fork.last_proposal()?;
         drop(forks);
-        Ok(canonical_blocks)
+        Ok((last.block.header.height, last.hash))
     }
 
     /// Auxiliary function to purge current forks and reset the ones starting
@@ -572,6 +632,7 @@ impl Consensus {
             self.blockchain.clone(),
             module.target,
             module.fixed_difficulty.clone(),
+            None,
         )?;
         drop(module);
         debug!(target: "validator::consensus::reset_pow_module", "PoW module reset successfully!");

src/validator/mod.rs

@@ -745,7 +745,7 @@ impl Validator {
         overlay.lock().unwrap().overlay.lock().unwrap().apply()?;
 
         // Create a PoW module to validate each block
-        let mut module = PoWModule::new(blockchain, pow_target, pow_fixed_difficulty)?;
+        let mut module = PoWModule::new(blockchain, pow_target, pow_fixed_difficulty, None)?;
 
         // Validate and insert each block
         for block in &blocks[1..] {

src/validator/pow.rs

@@ -92,10 +92,13 @@ pub struct PoWModule {
 }
 
 impl PoWModule {
+    // Initialize a new `PowModule` for provided target over provided `Blockchain`.
+    // Optionally, a fixed difficulty can be set and/or initialize before some height.
     pub fn new(
         blockchain: Blockchain,
         target: u32,
         fixed_difficulty: Option<BigUint>,
+        height: Option<u32>,
     ) -> Result<Self> {
         // Retrieve genesis block timestamp
         let genesis = blockchain.genesis_block()?.header.timestamp;
@@ -104,7 +107,10 @@ impl PoWModule {
         let mut timestamps = RingBuffer::<Timestamp, BUF_SIZE>::new();
         let mut difficulties = RingBuffer::<BigUint, BUF_SIZE>::new();
         let mut cummulative_difficulty = BigUint::zero();
-        let last_n = blockchain.blocks.get_last_n_difficulties(BUF_SIZE)?;
+        let last_n = match height {
+            Some(h) => blockchain.blocks.get_difficulties_before(h, BUF_SIZE)?,
+            None => blockchain.blocks.get_last_n_difficulties(BUF_SIZE)?,
+        };
         for difficulty in last_n {
             timestamps.push(difficulty.timestamp);
             difficulties.push(difficulty.cummulative_difficulty.clone());
@@ -197,25 +203,25 @@ impl PoWModule {
         Ok((cut_begin, cut_end))
     }
 
-    /// Compute the next mine target
+    /// Compute the next mine target.
     pub fn next_mine_target(&self) -> Result<BigUint> {
         Ok(BigUint::from_bytes_be(&[0xFF; 32]) / &self.next_difficulty()?)
     }
 
-    /// Compute the next mine target and difficulty
+    /// Compute the next mine target and difficulty.
     pub fn next_mine_target_and_difficulty(&self) -> Result<(BigUint, BigUint)> {
         let difficulty = self.next_difficulty()?;
         let mine_target = BigUint::from_bytes_be(&[0xFF; 32]) / &difficulty;
         Ok((mine_target, difficulty))
     }
 
-    /// Verify provided difficulty corresponds to the next one
+    /// Verify provided difficulty corresponds to the next one.
     pub fn verify_difficulty(&self, difficulty: &BigUint) -> Result<bool> {
         Ok(difficulty == &self.next_difficulty()?)
     }
 
     /// Verify provided block timestamp is not far in the future and
-    /// check its valid acorrding to current timestamps median
+    /// check its valid acorrding to current timestamps median.
     pub fn verify_current_timestamp(&self, timestamp: Timestamp) -> Result<bool> {
         if timestamp > Timestamp::current_time().checked_add(BLOCK_FUTURE_TIME_LIMIT)? {
             return Ok(false)
@@ -224,7 +230,7 @@ impl PoWModule {
         Ok(self.verify_timestamp_by_median(timestamp))
     }
 
-    /// Verify provided block timestamp is valid and matches certain criteria
+    /// Verify provided block timestamp is valid and matches certain criteria.
     pub fn verify_timestamp_by_median(&self, timestamp: Timestamp) -> bool {
         // Check timestamp is after genesis one
         if timestamp <= self.genesis {
@@ -248,7 +254,7 @@ impl PoWModule {
         timestamp >= median(timestamps).into()
     }
 
-    /// Verify provided block timestamp and hash
+    /// Verify provided block timestamp and hash.
     pub fn verify_current_block(&self, block: &BlockInfo) -> Result<()> {
         // First we verify the block's timestamp
         if !self.verify_current_timestamp(block.header.timestamp)? {
@@ -259,9 +265,8 @@ impl PoWModule {
         self.verify_block_hash(block)
     }
 
-    /// Verify provided block corresponds to next mine target
+    /// Verify provided block corresponds to next mine target.
     pub fn verify_block_hash(&self, block: &BlockInfo) -> Result<()> {
-        // Then we verify the proof of work:
         let verifier_setup = Instant::now();
 
         // Grab the next mine target
@@ -275,7 +280,7 @@ impl PoWModule {
 
         // Compute the output hash
         let verification_time = Instant::now();
-        let out_hash = vm.hash(block.hash().inner());
+        let out_hash = vm.hash(block.header.hash().inner());
         let out_hash = BigUint::from_bytes_be(&out_hash);
 
         // Verify hash is less than the expected mine target
@@ -287,7 +292,7 @@ impl PoWModule {
         Ok(())
     }
 
-    /// Append provided timestamp and difficulty to the ring buffers
+    /// Append provided timestamp and difficulty to the ring buffers.
     pub fn append(&mut self, timestamp: Timestamp, difficulty: &BigUint) {
         self.timestamps.push(timestamp);
         self.cummulative_difficulty += difficulty;
@@ -295,7 +300,7 @@ impl PoWModule {
     }
 
     /// Append provided block difficulty to the ring buffers and insert
-    /// it to provided overlay
+    /// it to provided overlay.
     pub fn append_difficulty(
         &mut self,
         overlay: &BlockchainOverlayPtr,
@@ -305,7 +310,7 @@ impl PoWModule {
         overlay.lock().unwrap().blocks.insert_difficulty(&[difficulty])
     }
 
-    /// Mine provided block, based on next mine target
+    /// Mine provided block, based on next mine target.
     pub fn mine_block(
         &self,
         miner_block: &mut BlockInfo,
@@ -329,7 +334,7 @@ impl std::fmt::Display for PoWModule {
     }
 }
 
-/// Mine provided block, based on provided PoW module next mine target
+/// Mine provided block, based on provided PoW module next mine target.
 pub fn mine_block(
     target: &BigUint,
     miner_block: &mut BlockInfo,
@@ -441,7 +446,7 @@ mod tests {
         let blockchain = Blockchain::new(&sled_db)?;
         let genesis_block = BlockInfo::default();
         blockchain.add_block(&genesis_block)?;
-        let mut module = PoWModule::new(blockchain, DEFAULT_TEST_DIFFICULTY_TARGET, None)?;
+        let mut module = PoWModule::new(blockchain, DEFAULT_TEST_DIFFICULTY_TARGET, None, None)?;
 
         let output = Command::new("./script/research/pow/gen_wide_data.py").output().unwrap();
         let reader = Cursor::new(output.stdout);
@@ -477,7 +482,7 @@ mod tests {
         let mut genesis_block = BlockInfo::default();
         genesis_block.header.timestamp = 0.into();
         blockchain.add_block(&genesis_block)?;
-        let module = PoWModule::new(blockchain, DEFAULT_TEST_DIFFICULTY_TARGET, None)?;
+        let module = PoWModule::new(blockchain, DEFAULT_TEST_DIFFICULTY_TARGET, None, None)?;
         let (_, recvr) = smol::channel::bounded(1);
 
         // Mine next block

src/validator/utils.rs

@@ -25,7 +25,7 @@ use num_bigint::BigUint;
 use randomx::{RandomXCache, RandomXFlags, RandomXVM};
 
 use crate::{
-    blockchain::{BlockInfo, BlockchainOverlayPtr},
+    blockchain::{BlockInfo, BlockchainOverlayPtr, Header},
     runtime::vm_runtime::Runtime,
     validator::consensus::{Fork, Proposal},
     Error, Result,
@@ -108,6 +108,45 @@ pub async fn deploy_native_contracts(
     Ok(())
 }
 
+/// Verify provided header is valid for provided mining target and compute its rank.
+///
+/// Header's rank is the tuple of its squared mining target distance from max 32 bytes int,
+/// along with its squared RandomX hash number distance from max 32 bytes int.
+/// Genesis block has rank (0, 0).
+pub fn header_rank(header: &Header, target: &BigUint) -> Result<(BigUint, BigUint)> {
+    // Genesis header has rank 0
+    if header.height == 0 {
+        return Ok((0u64.into(), 0u64.into()))
+    }
+
+    // Setup RandomX verifier
+    let flags = RandomXFlags::default();
+    let cache = RandomXCache::new(flags, header.previous.inner()).unwrap();
+    let vm = RandomXVM::new(flags, &cache).unwrap();
+
+    // Compute the output hash
+    let out_hash = vm.hash(header.hash().inner());
+    let out_hash = BigUint::from_bytes_be(&out_hash);
+
+    // Verify hash is less than the expected mine target
+    if out_hash > *target {
+        return Err(Error::PoWInvalidOutHash)
+    }
+
+    // Grab the max 32 bytes int
+    let max = BigUint::from_bytes_be(&[0xFF; 32]);
+
+    // Compute the squared mining target distance
+    let target_distance = &max - target;
+    let target_distance_sq = &target_distance * &target_distance;
+
+    // Compute the output hash distance
+    let hash_distance = max - out_hash;
+    let hash_distance_sq = &hash_distance * &hash_distance;
+
+    Ok((target_distance_sq, hash_distance_sq))
+}
+
 /// Compute a block's rank, assuming that its valid, based on provided mining target.
 ///
 /// Block's rank is the tuple of its squared mining target distance from max 32 bytes int,

src/validator/verification.rs

@@ -157,7 +157,7 @@ pub fn validate_blockchain(
     pow_fixed_difficulty: Option<BigUint>,
 ) -> Result<()> {
     // Generate a PoW module
-    let mut module = PoWModule::new(blockchain.clone(), pow_target, pow_fixed_difficulty)?;
+    let mut module = PoWModule::new(blockchain.clone(), pow_target, pow_fixed_difficulty, None)?;
     // We use block order store here so we have all blocks in order
     let blocks = blockchain.blocks.get_all_order()?;
     for (index, block) in blocks[1..].iter().enumerate() {
@@ -994,8 +994,7 @@ async fn apply_transactions(
 ///
 /// A proposal is considered valid when the following rules apply:
 ///     1. Proposal hash matches the actual block one
-///     2. Block transactions don't exceed set limit
-///     3. Block is valid
+///     2. Block is valid
 /// Additional validity rules can be applied.
 pub async fn verify_proposal(
     consensus: &Consensus,
@@ -1006,7 +1005,7 @@ pub async fn verify_proposal(
     let proposal_hash = proposal.block.hash();
     if proposal.hash != proposal_hash {
         warn!(
-            target: "validator::verification::verify_pow_proposal", "Received proposal contains mismatched hashes: {} - {}",
+            target: "validator::verification::verify_proposal", "Received proposal contains mismatched hashes: {} - {}",
             proposal.hash, proposal_hash
         );
         return Err(Error::ProposalHashesMissmatchError)
@@ -1018,15 +1017,52 @@ pub async fn verify_proposal(
     // Grab overlay last block
     let previous = fork.overlay.lock().unwrap().last_block()?;
 
-    // Verify proposal block (3)
+    // Verify proposal block (2)
     if verify_block(&fork.overlay, &fork.module, &proposal.block, &previous, verify_fees)
         .await
         .is_err()
     {
-        error!(target: "validator::verification::verify_pow_proposal", "Erroneous proposal block found");
+        error!(target: "validator::verification::verify_proposal", "Erroneous proposal block found");
         fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
         return Err(Error::BlockIsInvalid(proposal.hash.as_string()))
     };
 
     Ok((fork, index))
 }
+
+/// Verify given [`Proposal`] against provided fork state.
+///
+/// A proposal is considered valid when the following rules apply:
+///     1. Proposal hash matches the actual block one
+///     2. Block is valid
+/// Additional validity rules can be applied.
+pub async fn verify_fork_proposal(
+    fork: &Fork,
+    proposal: &Proposal,
+    verify_fees: bool,
+) -> Result<()> {
+    // Check if proposal hash matches actual one (1)
+    let proposal_hash = proposal.block.hash();
+    if proposal.hash != proposal_hash {
+        warn!(
+            target: "validator::verification::verify_fork_proposal", "Received proposal contains mismatched hashes: {} - {}",
+            proposal.hash, proposal_hash
+        );
+        return Err(Error::ProposalHashesMissmatchError)
+    }
+
+    // Grab overlay last block
+    let previous = fork.overlay.lock().unwrap().last_block()?;
+
+    // Verify proposal block (2)
+    if verify_block(&fork.overlay, &fork.module, &proposal.block, &previous, verify_fees)
+        .await
+        .is_err()
+    {
+        error!(target: "validator::verification::verify_fork_proposal", "Erroneous proposal block found");
+        fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
+        return Err(Error::BlockIsInvalid(proposal.hash.as_string()))
+    };
+
+    Ok(())
+}