consensus: broadcasting finalized slot checkpoints impl added, minor fixes

src/blockchain/mod.rs

@@ -243,4 +243,12 @@ impl Blockchain {
         debug!("get_slot_checkpoints_by_slot(): {:?}", slots);
         self.slot_checkpoints.get(slots, true)
     }
+
+    /// Check if the given [`SlotCheckpoint`] is in the database and all trees.
+    pub fn has_slot_checkpoint(&self, slot_checkpoint: &SlotCheckpoint) -> Result<bool> {
+        if let Err(_) = self.slot_checkpoints.get(&[slot_checkpoint.slot], true) {
+            return Ok(false)
+        }
+        Ok(true)
+    }
 }

src/consensus/proto/protocol_sync.rs

@@ -24,7 +24,7 @@ use smol::Executor;
 use crate::{
     consensus::{
         block::{BlockInfo, BlockOrder, BlockResponse},
-        state::{SlotCheckpointRequest, SlotCheckpointResponse},
+        state::{SlotCheckpoint, SlotCheckpointRequest, SlotCheckpointResponse},
         ValidatorStatePtr,
     },
     net::{
@@ -42,6 +42,7 @@ pub struct ProtocolSync {
     request_sub: MessageSubscription<BlockOrder>,
     slot_checkpoin_request_sub: MessageSubscription<SlotCheckpointRequest>,
     block_sub: MessageSubscription<BlockInfo>,
+    slot_checkpoints_sub: MessageSubscription<SlotCheckpoint>,
     jobsman: ProtocolJobsManagerPtr,
     state: ValidatorStatePtr,
     p2p: P2pPtr,
@@ -59,16 +60,19 @@ impl ProtocolSync {
         msg_subsystem.add_dispatch::<BlockOrder>().await;
         msg_subsystem.add_dispatch::<SlotCheckpointRequest>().await;
         msg_subsystem.add_dispatch::<BlockInfo>().await;
+        msg_subsystem.add_dispatch::<SlotCheckpoint>().await;
 
         let request_sub = channel.subscribe_msg::<BlockOrder>().await?;
         let slot_checkpoin_request_sub = channel.subscribe_msg::<SlotCheckpointRequest>().await?;
         let block_sub = channel.subscribe_msg::<BlockInfo>().await?;
+        let slot_checkpoints_sub = channel.subscribe_msg::<SlotCheckpoint>().await?;
 
         Ok(Arc::new(Self {
             channel: channel.clone(),
             request_sub,
             slot_checkpoin_request_sub,
             block_sub,
+            slot_checkpoints_sub,
             jobsman: ProtocolJobsManager::new("SyncProtocol", channel),
             state,
             p2p,
@@ -107,6 +111,52 @@ impl ProtocolSync {
         }
     }
 
+    async fn handle_receive_block(self: Arc<Self>) -> Result<()> {
+        // Consensus-mode enabled nodes have already performed these steps,
+        // during proposal finalization.
+        if self.consensus_mode && self.state.read().await.consensus.participating.is_some() {
+            debug!(
+                "ProtocolSync::handle_receive_block(): node runs in consensus mode, skipping..."
+            );
+            return Ok(())
+        }
+
+        debug!("ProtocolSync::handle_receive_block() [START]");
+        let exclude_list = vec![self.channel.address()];
+        loop {
+            let info = match self.block_sub.receive().await {
+                Ok(v) => v,
+                Err(e) => {
+                    error!("ProtocolSync::handle_receive_block(): recv fail: {}", e);
+                    continue
+                }
+            };
+
+            info!("ProtocolSync::handle_receive_block(): Received block: {}", info.blockhash());
+
+            debug!("ProtocolSync::handle_receive_block(): Processing received block");
+            let info_copy = (*info).clone();
+            match self.state.write().await.receive_finalized_block(info_copy.clone()).await {
+                Ok(v) => {
+                    if v {
+                        debug!("ProtocolProposal::handle_receive_block(): block processed successfully, broadcasting...");
+                        if let Err(e) =
+                            self.p2p.broadcast_with_exclude(info_copy, &exclude_list).await
+                        {
+                            error!(
+                                "ProtocolSync::handle_receive_block(): p2p broadcast fail: {}",
+                                e
+                            );
+                        };
+                    }
+                }
+                Err(e) => {
+                    debug!("ProtocolSync::handle_receive_block(): error processing finalized block: {}", e);
+                }
+            };
+        }
+    }
+
     async fn handle_receive_slot_checkpoint_request(self: Arc<Self>) -> Result<()> {
         debug!("ProtocolSync::handle_receive_slot_checkpoint_request() [START]");
         loop {
@@ -153,47 +203,58 @@ impl ProtocolSync {
         }
     }
 
-    async fn handle_receive_block(self: Arc<Self>) -> Result<()> {
+    async fn handle_receive_slot_checkpoint(self: Arc<Self>) -> Result<()> {
         // Consensus-mode enabled nodes have already performed these steps,
         // during proposal finalization.
-        if self.consensus_mode {
+        if self.consensus_mode && self.state.read().await.consensus.participating.is_some() {
             debug!(
-                "ProtocolSync::handle_receive_block(): node runs in consensus mode, skipping..."
+                "ProtocolSync::handle_receive_slot_checkpoint(): node runs in consensus mode, skipping..."
             );
             return Ok(())
         }
 
-        debug!("ProtocolSync::handle_receive_block() [START]");
+        debug!("ProtocolSync::handle_receive_slot_checkpoint() [START]");
         let exclude_list = vec![self.channel.address()];
         loop {
-            let info = match self.block_sub.receive().await {
+            let slot_checkpoint = match self.slot_checkpoints_sub.receive().await {
                 Ok(v) => v,
                 Err(e) => {
-                    error!("ProtocolSync::handle_receive_block(): recv fail: {}", e);
+                    error!("ProtocolSync::handle_receive_slot_checkpoint(): recv fail: {}", e);
                     continue
                 }
             };
 
-            info!("ProtocolSync::handle_receive_block(): Received block: {}", info.blockhash());
+            info!(
+                "ProtocolSync::handle_receive_slot_checkpoint(): Received slot checkpoint: {}",
+                slot_checkpoint.slot
+            );
 
-            debug!("ProtocolSync::handle_receive_block(): Processing received block");
-            let info_copy = (*info).clone();
-            match self.state.write().await.receive_finalized_block(info_copy.clone()).await {
+            debug!("ProtocolSync::handle_receive_slot_checkpoint(): Processing received slot checkpoint");
+            let slot_checkpoint_copy = (*slot_checkpoint).clone();
+            match self
+                .state
+                .write()
+                .await
+                .receive_finalized_slot_checkpoints(slot_checkpoint_copy.clone())
+                .await
+            {
                 Ok(v) => {
                     if v {
-                        debug!("ProtocolProposal::handle_receive_block(): block processed successfully, broadcasting...");
-                        if let Err(e) =
-                            self.p2p.broadcast_with_exclude(info_copy, &exclude_list).await
+                        debug!("ProtocolProposal::handle_receive_slot_checkpoint(): slot checkpoint processed successfully, broadcasting...");
+                        if let Err(e) = self
+                            .p2p
+                            .broadcast_with_exclude(slot_checkpoint_copy, &exclude_list)
+                            .await
                         {
                             error!(
-                                "ProtocolSync::handle_receive_block(): p2p broadcast fail: {}",
+                                "ProtocolSync::handle_receive_slot_checkpoint(): p2p broadcast fail: {}",
                                 e
                             );
                         };
                     }
                 }
                 Err(e) => {
-                    debug!("ProtocolSync::handle_receive_block(): error processing finalized block: {}", e);
+                    debug!("ProtocolSync::handle_receive_slot_checkpoint(): error processing finalized slot checkpoint: {}", e);
                 }
             };
         }
@@ -211,6 +272,10 @@ impl ProtocolBase for ProtocolSync {
             .spawn(self.clone().handle_receive_slot_checkpoint_request(), executor.clone())
             .await;
         self.jobsman.clone().spawn(self.clone().handle_receive_block(), executor.clone()).await;
+        self.jobsman
+            .clone()
+            .spawn(self.clone().handle_receive_slot_checkpoint(), executor.clone())
+            .await;
         debug!("ProtocolSync::start() [END]");
         Ok(())
     }

src/consensus/state.rs

@@ -585,14 +585,13 @@ impl ConsensusState {
         Err(Error::SlotCheckpointNotFound(slot))
     }
 
-    /// Auxillary function to update all fork state checkpoints to nodes current canonical states.
+    /// Auxillary function to update all fork state checkpoints to nodes coins current canonical states.
     /// Note: This function should only be invoked once on nodes' coins creation.
     pub fn update_forks_checkpoints(&mut self) {
         for fork in &mut self.forks {
             for state_checkpoint in &mut fork.sequence {
                 state_checkpoint.coins = self.coins.clone();
                 state_checkpoint.coins_tree = self.coins_tree.clone();
-                state_checkpoint.nullifiers = self.nullifiers.clone();
             }
         }
     }
@@ -657,6 +656,12 @@ impl SlotCheckpoint {
     }
 }
 
+impl net::Message for SlotCheckpoint {
+    fn name() -> &'static str {
+        "slotcheckpoint"
+    }
+}
+
 /// Auxiliary structure used for slot checkpoints syncing
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct SlotCheckpointRequest {

src/consensus/task/proposal.rs

@@ -72,11 +72,11 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
 
         // Check if any forks can be finalized
         match state.write().await.chain_finalization().await {
-            Ok(to_broadcast) => {
+            Ok((to_broadcast_block, to_broadcast_slot_checkpoints)) => {
                 // Broadcast finalized blocks info, if any:
-                if to_broadcast.len() > 0 {
+                if to_broadcast_block.len() > 0 {
                     info!("consensus: Broadcasting finalized blocks");
-                    for info in to_broadcast {
+                    for info in to_broadcast_block {
                         match sync_p2p.broadcast(info).await {
                             Ok(()) => info!("consensus: Broadcasted block"),
                             Err(e) => error!("consensus: Failed broadcasting block: {}", e),
@@ -85,6 +85,21 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
                 } else {
                     info!("consensus: No finalized blocks to broadcast");
                 }
+
+                // Broadcast finalized slot checkpoints, if any:
+                if to_broadcast_slot_checkpoints.len() > 0 {
+                    info!("consensus: Broadcasting finalized slot checkpoints");
+                    for slot_checkpoint in to_broadcast_slot_checkpoints {
+                        match sync_p2p.broadcast(slot_checkpoint).await {
+                            Ok(()) => info!("consensus: Broadcasted slot_checkpoint"),
+                            Err(e) => {
+                                error!("consensus: Failed broadcasting slot_checkpoint: {}", e)
+                            }
+                        }
+                    }
+                } else {
+                    info!("consensus: No finalized slot checkpoints to broadcast");
+                }
             }
             Err(e) => {
                 error!("consensus: Finalization check failed: {}", e);

src/consensus/validator.rs

@@ -549,7 +549,7 @@ impl ValidatorState {
     ///   all proposals up to the last one.
     /// When fork chain proposals are finalized, the rest of fork chains are removed and all
     /// slot checkpoints until current slot are apppended to canonical state.
-    pub async fn chain_finalization(&mut self) -> Result<Vec<BlockInfo>> {
+    pub async fn chain_finalization(&mut self) -> Result<(Vec<BlockInfo>, Vec<SlotCheckpoint>)> {
         let slot = self.consensus.current_slot();
         debug!("chain_finalization(): Started finalization check for slot: {}", slot);
         // Set last slot finalization check occured to current slot
@@ -591,12 +591,12 @@ impl ValidatorState {
             -2 => {
                 debug!("chain_finalization(): Eligible forks with same height exist, nothing to finalize.");
                 self.consensus.set_leader_history(index_for_history);
-                return Ok(vec![])
+                return Ok((vec![], vec![]))
             }
             -1 => {
                 debug!("chain_finalization(): All chains have less than 3 proposals, nothing to finalize.");
                 self.consensus.set_leader_history(index_for_history);
-                return Ok(vec![])
+                return Ok((vec![], vec![]))
             }
             _ => debug!("chain_finalization(): Chain {} can be finalized!", fork_index),
         }
@@ -695,7 +695,7 @@ impl ValidatorState {
             }
         };
 
-        Ok(finalized)
+        Ok((finalized, finalized_slot_checkpoints))
     }
 
     // ==========================
@@ -999,4 +999,28 @@ impl ValidatorState {
 
         Ok(())
     }
+
+    /// Validate and append to canonical state received finalized slot checkpoint.
+    /// Returns boolean flag indicating already existing slot checkpoint.
+    pub async fn receive_finalized_slot_checkpoints(
+        &mut self,
+        slot_checkpoint: SlotCheckpoint,
+    ) -> Result<bool> {
+        match self.blockchain.has_slot_checkpoint(&slot_checkpoint) {
+            Ok(v) => {
+                if v {
+                    debug!(
+                        "receive_finalized_slot_checkpoints(): Existing slot checkpoint received"
+                    );
+                    return Ok(false)
+                }
+            }
+            Err(e) => {
+                error!("receive_finalized_slot_checkpoints(): failed checking for has_slot_checkpoint(): {}", e);
+                return Ok(false)
+            }
+        };
+        self.receive_slot_checkpoints(&[slot_checkpoint]).await?;
+        Ok(true)
+    }
 }