use log::{info, warn}; use crate::{ consensus::{ state::{ConsensusRequest, ConsensusResponse}, ValidatorStatePtr, }, net::P2pPtr, Result, }; /// async task used for consensus state syncing. pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Result<()> { info!("Starting consensus state sync..."); let channels_map = p2p.channels().lock().await; let values = channels_map.values(); // Using len here because is_empty() uses unstable library feature // called 'exact_size_is_empty'. if values.len() != 0 { // Node iterates the channel peers to ask for their consensus state for channel in values { // Communication setup let msg_subsystem = channel.get_message_subsystem(); msg_subsystem.add_dispatch::().await; let response_sub = channel.subscribe_msg::().await?; // Node creates a `ConsensusRequest` and sends it let request = ConsensusRequest { address: state.read().await.address }; channel.send(request).await?; // Node verifies response came from a participating node. // Extra validations can be added here. let response = response_sub.receive().await?; if response.participants.is_empty() { warn!("Retrieved consensus state from a new node, retrying..."); continue } // Node stores response data. let mut lock = state.write().await; lock.consensus.proposals = response.proposals.clone(); lock.consensus.participants = response.participants.clone(); break } } else { warn!("Node is not connected to other nodes, resetting consensus state."); state.write().await.reset_consensus_state()?; } info!("Consensus state synced!"); Ok(()) }