consensus: sync fixes and more checks to trigger resync

src/consensus/proto/protocol_proposal.rs

@@ -87,21 +87,28 @@ impl ProtocolProposal {
                 continue
             }
 
-            if let Err(e) = lock.receive_proposal(&proposal_copy, None).await {
-                error!(
-                    "ProtocolProposal::handle_receive_proposal(): receive_proposal error: {}",
-                    e
-                );
-                continue
+            match lock.receive_proposal(&proposal_copy, None).await {
+                Ok(broadcast) => {
+                    if broadcast {
+                        // Broadcast proposal to rest of nodes
+                        if let Err(e) =
+                            self.p2p.broadcast_with_exclude(proposal_copy, &exclude_list).await
+                        {
+                            error!(
+                                "ProtocolProposal::handle_receive_proposal(): proposal broadcast fail: {}",
+                                e
+                            );
+                        };
+                    }
+                }
+                Err(e) => {
+                    error!(
+                        "ProtocolProposal::handle_receive_proposal(): receive_proposal error: {}",
+                        e
+                    );
+                    continue
+                }
             }
-
-            // Broadcast block to rest of nodes
-            if let Err(e) = self.p2p.broadcast_with_exclude(proposal_copy, &exclude_list).await {
-                error!(
-                    "ProtocolProposal::handle_receive_proposal(): proposal broadcast fail: {}",
-                    e
-                );
-            };
         }
     }
 }

src/consensus/proto/protocol_sync.rs

@@ -127,11 +127,19 @@ impl ProtocolSync {
             // Consensus-mode enabled nodes have already performed these steps,
             // during proposal finalization. They still listen to this sub,
             // in case they go out of sync and become a none-consensus node.
-            if self.consensus_mode && self.state.read().await.consensus.participating.is_some() {
-                debug!(
-                    "ProtocolSync::handle_receive_block(): node runs in consensus mode, skipping..."
-                );
-                continue
+            if self.consensus_mode {
+                let lock = self.state.read().await;
+                let current = lock.consensus.current_slot();
+                let participating = lock.consensus.participating;
+                if participating.is_some() {
+                    let slot = participating.unwrap();
+                    if current >= slot {
+                        debug!(
+                            "ProtocolSync::handle_receive_block(): node runs in consensus mode, skipping..."
+                        );
+                        continue
+                    }
+                }
             }
 
             info!("ProtocolSync::handle_receive_block(): Received block: {}", info.blockhash());
@@ -221,11 +229,19 @@ impl ProtocolSync {
             // Consensus-mode enabled nodes have already performed these steps,
             // during proposal finalization. They still listen to this sub,
             // in case they go out of sync and become a none-consensus node.
-            if self.consensus_mode && self.state.read().await.consensus.participating.is_some() {
-                debug!(
-                    "ProtocolSync::handle_receive_slot_checkpoint(): node runs in consensus mode, skipping..."
-                );
-                continue
+            if self.consensus_mode {
+                let lock = self.state.read().await;
+                let current = lock.consensus.current_slot();
+                let participating = lock.consensus.participating;
+                if participating.is_some() {
+                    let slot = participating.unwrap();
+                    if current >= slot {
+                        debug!(
+                            "ProtocolSync::handle_receive_block(): node runs in consensus mode, skipping..."
+                        );
+                        continue
+                    }
+                }
             }
 
             info!(

src/consensus/state.rs

@@ -335,7 +335,6 @@ impl ConsensusState {
         }
         // Retrieve longest fork length, to also those proposals in the calculation
         let max_fork_length = self.longest_chain_length() as u64;
-
         current_slot - blocks - self.get_current_offset(current_slot) - max_fork_length
     }
 
@@ -678,6 +677,7 @@ impl ConsensusState {
     pub fn reset(&mut self) {
         self.participating = None;
         self.proposing = false;
+        self.offset = None;
         self.forks = vec![];
         self.slot_checkpoints = vec![];
         self.leaders_history = vec![0];
@@ -686,7 +686,7 @@ impl ConsensusState {
 }
 
 /// Auxiliary structure used for consensus syncing.
-#[derive(Debug, SerialEncodable, SerialDecodable)]
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct ConsensusRequest {}
 
 impl net::Message for ConsensusRequest {

src/consensus/task/consensus_sync.rs

@@ -35,6 +35,7 @@ use crate::{
 /// so it can immediately start proposing proposals.
 pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Result<bool> {
     info!("Starting consensus state sync...");
+    let current_slot = state.read().await.consensus.current_slot();
     // Loop through connected channels
     let channels_map = p2p.channels().lock().await;
     let values = channels_map.values();
@@ -43,7 +44,7 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
     if values.len() == 0 {
         warn!("Node is not connected to other nodes");
         let mut lock = state.write().await;
-        lock.consensus.bootstrap_slot = lock.consensus.current_slot();
+        lock.consensus.bootstrap_slot = current_slot;
         lock.consensus.init_coins().await?;
         info!("Consensus state synced!");
         return Ok(true)
@@ -56,14 +57,12 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
         let msg_subsystem = channel.get_message_subsystem();
         msg_subsystem.add_dispatch::<ConsensusSlotCheckpointsResponse>().await;
         let response_sub = channel.subscribe_msg::<ConsensusSlotCheckpointsResponse>().await?;
-
         // Node creates a `ConsensusSlotCheckpointsRequest` and sends it
         let request = ConsensusSlotCheckpointsRequest {};
         channel.send(request).await?;
-
         // Node checks response
         let response = response_sub.receive().await?;
-        if response.bootstrap_slot == state.read().await.consensus.current_slot() {
+        if response.bootstrap_slot == current_slot {
             warn!("Network was just bootstraped, checking rest nodes");
             continue
         }
@@ -71,7 +70,6 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
             warn!("Node has not seen any slot checkpoints, retrying...");
             continue
         }
-
         // Keep peer to ask for consensus state
         peer = Some(channel.clone());
         break
@@ -85,7 +83,7 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
     if peer.is_none() {
         warn!("No node that has seen any slot checkpoints was found, or network was just boostrapped.");
         let mut lock = state.write().await;
-        lock.consensus.bootstrap_slot = lock.consensus.current_slot();
+        lock.consensus.bootstrap_slot = current_slot;
         lock.consensus.init_coins().await?;
         info!("Consensus state synced!");
         return Ok(true)
@@ -106,14 +104,33 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
     let msg_subsystem = peer.get_message_subsystem();
     msg_subsystem.add_dispatch::<ConsensusResponse>().await;
     let response_sub = peer.subscribe_msg::<ConsensusResponse>().await?;
-
     // Node creates a `ConsensusRequest` and sends it
-    let request = ConsensusRequest {};
-    peer.send(request).await?;
+    peer.send(ConsensusRequest {}).await?;
 
     // Node verifies response came from a participating node.
     // Extra validations can be added here.
-    let response = response_sub.receive().await?;
+    let mut response = response_sub.receive().await?;
+    // Verify that peer has finished finalizing forks
+    loop {
+        if response.forks.len() != 1 || response.forks[0].sequence.len() != 1 {
+            warn!("Peer has not finished finalization, retrying...");
+            peer.send(ConsensusRequest {}).await?;
+            response = response_sub.receive().await?;
+            continue
+        }
+        break
+    }
+
+    // Verify that the node has received all finalized blocks
+    loop {
+        let lock = state.read().await;
+        let last_finalized = lock.consensus.current_slot() - 1;
+        if lock.blockchain.last().unwrap().0 != last_finalized {
+            warn!("Node has not finished finalization, retrying...");
+            continue
+        }
+        break
+    }
 
     // Node stores response data.
     let mut lock = state.write().await;

src/consensus/task/proposal.rs

@@ -35,14 +35,30 @@ pub async fn proposal_task(
     state: ValidatorStatePtr,
     ex: Arc<smol::Executor<'_>>,
 ) {
-    // Check if network is configured to start in the future
-    // NOTE: This should always be true when bootstrapping or restarting a network.
+    // Check if network is configured to start in the future,
+    // otherwise wait for current or next slot finalization period for optimal sync conditions.
+    // NOTE: Network beign configured to start in the future should always be the case
+    // when bootstrapping or restarting a network.
     let current_ts = Timestamp::current_time();
     let genesis_ts = state.read().await.consensus.genesis_ts;
     if current_ts < genesis_ts {
         let diff = genesis_ts.0 - current_ts.0;
         info!("consensus: Waiting for network bootstrap: {} seconds", diff);
         sleep(diff as u64).await;
+    } else {
+        let mut sleep_time = state.read().await.consensus.next_n_slot_start(1);
+        let sync_offset = Duration::new(constants::FINAL_SYNC_DUR, 0);
+        loop {
+            if sleep_time > sync_offset {
+                sleep_time -= sync_offset;
+                break
+            }
+            info!("consensus: Waiting for next slot ({:?})", sleep_time);
+            sleep(sleep_time.as_secs()).await;
+            sleep_time = state.read().await.consensus.next_n_slot_start(1);
+        }
+        info!("consensus: Waiting for finalization sync period ({:?})", sleep_time);
+        sleep(sleep_time.as_secs()).await;
     }
 
     let mut retries = 0;
@@ -200,11 +216,23 @@ async fn propose_period(consensus_p2p: P2pPtr, state: ValidatorStatePtr) -> bool
         }
     };
 
+    // Node checks if it missed finalization period due to proposal creation
+    let next_slot_start = state.read().await.consensus.next_n_slot_start(1);
+    if next_slot_start.as_secs() <= constants::FINAL_SYNC_DUR {
+        warn!(
+            "consensus: Node missed slot {} finalization period due to proposal creation, resyncing...",
+            state.read().await.consensus.current_slot()
+        );
+        return true
+    }
+
     // Node stores the proposal and broadcast to rest nodes
     info!("consensus: Node is the slot leader: Proposed block: {}", proposal);
     debug!("consensus: Full proposal: {:?}", proposal);
     match state.write().await.receive_proposal(&proposal, Some((idx, coin))).await {
-        Ok(()) => {
+        Ok(_) => {
+            // Here we don't have to check to broadcast, because the flag
+            // will always be true, since the node is able to produce proposals
             info!("consensus: Block proposal saved successfully");
             // Broadcast proposal to other consensus nodes
             match consensus_p2p.broadcast(proposal).await {
@@ -230,13 +258,18 @@ async fn finalization_period(
 ) -> bool {
     // Node sleeps until finalization sync period starts
     let next_slot_start = state.read().await.consensus.next_n_slot_start(1);
-    let seconds_sync_period = if next_slot_start.as_secs() > constants::FINAL_SYNC_DUR {
-        (next_slot_start - Duration::new(constants::FINAL_SYNC_DUR, 0)).as_secs()
+    if next_slot_start.as_secs() > constants::FINAL_SYNC_DUR {
+        let seconds_sync_period =
+            (next_slot_start - Duration::new(constants::FINAL_SYNC_DUR, 0)).as_secs();
+        info!("consensus: Waiting for finalization sync period ({} sec)", seconds_sync_period);
+        sleep(seconds_sync_period).await;
     } else {
-        next_slot_start.as_secs()
-    };
-    info!("consensus: Waiting for finalization sync period ({} sec)", seconds_sync_period);
-    sleep(seconds_sync_period).await;
+        warn!(
+            "consensus: Node missed slot {} finalization period due to proposals processing, resyncing...",
+            state.read().await.consensus.current_slot()
+        );
+        return true
+    }
 
     // Keep a record of slot to verify if next slot got skipped during processing
     let completed_slot = state.read().await.consensus.current_slot();

src/consensus/validator.rs

@@ -348,20 +348,22 @@ impl ValidatorState {
 
     /// Given a proposal, the node verify its sender (slot leader) and finds which blockchain
     /// it extends. If the proposal extends the canonical blockchain, a new fork chain is created.
+    /// Returns flag to signal if proposal should be broadcasted. Only active consensus participants
+    /// should broadcast proposals.
     pub async fn receive_proposal(
         &mut self,
         proposal: &BlockProposal,
         coin: Option<(usize, LeadCoin)>,
-    ) -> Result<()> {
+    ) -> Result<bool> {
         let current = self.consensus.current_slot();
         // Node hasn't started participating
         match self.consensus.participating {
             Some(start) => {
                 if current < start {
-                    return Ok(())
+                    return Ok(false)
                 }
             }
-            None => return Ok(()),
+            None => return Ok(false),
         }
 
         // Node have already checked for finalization in this slot
@@ -569,7 +571,7 @@ impl ValidatorState {
             }
         };
 
-        Ok(())
+        Ok(true)
     }
 
     /// Remove provided transactions vector from unconfirmed_txs if they exist.