consensus: Misc fixes and cleanups.

src/blockchain/mod.rs

@@ -101,10 +101,8 @@ impl Blockchain {
         let blockhashes = self.order.get(slots, false)?;
 
         let mut hashes = vec![];
-        for i in blockhashes {
-            if i.is_some() {
-                hashes.push(i.unwrap());
-            }
+        for i in blockhashes.into_iter().flatten() {
+            hashes.push(i);
         }
 
         self.get_blocks_by_hash(&hashes)

src/consensus2/block.rs

@@ -241,29 +241,3 @@ impl ProposalChain {
 }
 
 impl_vec!(ProposalChain);
-
-/// Auxiliary structure used for forks syncing.
-#[derive(Debug, SerialEncodable, SerialDecodable)]
-pub struct ForkOrder {
-    /// Validator ID
-    pub id: u64,
-}
-
-impl net::Message for ForkOrder {
-    fn name() -> &'static str {
-        "forkorder"
-    }
-}
-
-/// Auxiliary structure used for forks syncing.
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct ForkResponse {
-    /// Fork chains containing block proposals
-    pub proposals: Vec<ProposalChain>,
-}
-
-impl net::Message for ForkResponse {
-    fn name() -> &'static str {
-        "forkresponse"
-    }
-}

src/consensus2/proto/mod.rs

@@ -22,6 +22,6 @@ pub use protocol_vote::ProtocolVote;
 mod protocol_sync;
 pub use protocol_sync::ProtocolSync;
 
-/// Validator forks sync protocol
-mod protocol_sync_forks;
-pub use protocol_sync_forks::ProtocolSyncForks;
+/// Validator consensus sync protocol
+mod protocol_sync_consensus;
+pub use protocol_sync_consensus::ProtocolSyncConsensus;

src/consensus2/proto/protocol_participant.rs

@@ -46,12 +46,9 @@ impl ProtocolParticipant {
 
             debug!("ProtocolParticipant::handle_receive_participant() recv: {:?}", participant);
 
-            if self.state.write().await.append_participant((*participant).clone()) {
-                let pending_participants =
-                    self.state.read().await.consensus.pending_participants.clone();
-                for participant in pending_participants {
-                    self.p2p.broadcast(participant).await?;
-                }
+            let participant_copy = (*participant).clone();
+            if self.state.write().await.append_participant(participant_copy.clone()) {
+                self.p2p.broadcast(participant_copy).await?;
             }
         }
     }

src/consensus2/proto/protocol_sync.rs

@@ -20,7 +20,7 @@ const BATCH: u64 = 10;
 
 pub struct ProtocolSync {
     channel: ChannelPtr,
-    order_sub: MessageSubscription<BlockOrder>,
+    request_sub: MessageSubscription<BlockOrder>,
     block_sub: MessageSubscription<BlockInfo>,
     jobsman: ProtocolJobsManagerPtr,
     state: ValidatorStatePtr,
@@ -39,12 +39,12 @@ impl ProtocolSync {
         msg_subsystem.add_dispatch::<BlockOrder>().await;
         msg_subsystem.add_dispatch::<BlockInfo>().await;
 
-        let order_sub = channel.subscribe_msg::<BlockOrder>().await?;
+        let request_sub = channel.subscribe_msg::<BlockOrder>().await?;
         let block_sub = channel.subscribe_msg::<BlockInfo>().await?;
 
         Ok(Arc::new(Self {
             channel: channel.clone(),
-            order_sub,
+            request_sub,
             block_sub,
             jobsman: ProtocolJobsManager::new("SyncProtocol", channel),
             state,
@@ -53,19 +53,21 @@ impl ProtocolSync {
         }))
     }
 
-    async fn handle_receive_order(self: Arc<Self>) -> Result<()> {
-        debug!("ProtocolSync::handle_receive_order() [START]");
+    async fn handle_receive_request(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolSync::handle_receive_request() [START]");
         loop {
-            let order = self.order_sub.receive().await?;
+            let order = self.request_sub.receive().await?;
 
-            debug!("ProtocolSync::handle_receive_order() received {:?}", order);
+            debug!("ProtocolSync::handle_receive_request() received {:?}", order);
 
             // Extra validations can be added here
             let key = order.sl;
-            let slot_range: Vec<u64> = (key..=(key + BATCH)).collect();
-            debug!("ProtocolSync::handle_receive_order(): Querying block range: {:?}", slot_range);
-            let blocks = self.state.read().await.blockchain.get_blocks_by_slot(&slot_range)?;
-            debug!("ProtocolSync::handle_receive_order(): Found {} blocks", blocks.len());
+            let range: Vec<u64> = (key..=(key + BATCH)).collect();
+
+            debug!("ProtocolSync::handle_receive_request(): Querying block range: {:?}", range);
+            let blocks = self.state.read().await.blockchain.get_blocks_by_slot(&range)?;
+            debug!("ProtocolSync::handle_receive_request(): Found {} blocks", blocks.len());
+
             let response = BlockResponse { blocks };
             self.channel.send(response).await?;
         }
@@ -100,7 +102,7 @@ impl ProtocolBase for ProtocolSync {
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!("ProtocolSync::start() [START]");
         self.jobsman.clone().start(executor.clone());
-        self.jobsman.clone().spawn(self.clone().handle_receive_order(), executor.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_receive_request(), executor.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_receive_block(), executor.clone()).await;
         debug!("ProtocolSync::start() [END]");
         Ok(())

src/consensus2/proto/protocol_sync_consensus.rs

@@ -0,0 +1,72 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use log::debug;
+
+use crate::{
+    consensus2::{
+        state::{ConsensusRequest, ConsensusResponse},
+        ValidatorStatePtr,
+    },
+    net::{
+        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    Result,
+};
+
+pub struct ProtocolSyncConsensus {
+    channel: ChannelPtr,
+    request_sub: MessageSubscription<ConsensusRequest>,
+    jobsman: ProtocolJobsManagerPtr,
+    state: ValidatorStatePtr,
+}
+
+impl ProtocolSyncConsensus {
+    pub async fn init(
+        channel: ChannelPtr,
+        state: ValidatorStatePtr,
+        _p2p: P2pPtr,
+    ) -> Result<ProtocolBasePtr> {
+        let msg_subsystem = channel.get_message_subsystem();
+        msg_subsystem.add_dispatch::<ConsensusRequest>().await;
+
+        let request_sub = channel.subscribe_msg::<ConsensusRequest>().await?;
+
+        Ok(Arc::new(Self {
+            channel: channel.clone(),
+            request_sub,
+            jobsman: ProtocolJobsManager::new("SyncConsensusProtocol", channel),
+            state,
+        }))
+    }
+
+    async fn handle_receive_request(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolSyncConsensus::handle_receive_request() [START]");
+        loop {
+            let order = self.request_sub.receive().await?;
+
+            debug!("ProtocolSyncConsensuss::handle_receive_request() received {:?}", order);
+
+            // Extra validations can be added here.
+            let consensus = self.state.read().await.consensus.clone();
+            let response = ConsensusResponse { consensus };
+            self.channel.send(response).await?;
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolSyncConsensus {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!("ProtocolSyncConsensus::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_request(), executor.clone()).await;
+        debug!("ProtocolSyncConsensus::start() [END]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolSyncConsensus"
+    }
+}

src/consensus2/proto/protocol_sync_forks.rs

@@ -1,72 +0,0 @@
-use async_executor::Executor;
-use async_std::sync::Arc;
-use async_trait::async_trait;
-use log::debug;
-
-use crate::{
-    consensus2::{
-        block::{ForkOrder, ForkResponse},
-        state::ValidatorStatePtr,
-    },
-    net::{
-        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
-        ProtocolJobsManager, ProtocolJobsManagerPtr,
-    },
-    Result,
-};
-
-pub struct ProtocolSyncForks {
-    channel: ChannelPtr,
-    order_sub: MessageSubscription<ForkOrder>,
-    jobsman: ProtocolJobsManagerPtr,
-    state: ValidatorStatePtr,
-}
-
-impl ProtocolSyncForks {
-    pub async fn init(
-        channel: ChannelPtr,
-        state: ValidatorStatePtr,
-        _p2p: P2pPtr,
-    ) -> Result<ProtocolBasePtr> {
-        let msg_subsystem = channel.get_message_subsystem();
-        msg_subsystem.add_dispatch::<ForkOrder>().await;
-
-        let order_sub = channel.subscribe_msg::<ForkOrder>().await?;
-
-        Ok(Arc::new(Self {
-            channel: channel.clone(),
-            order_sub,
-            jobsman: ProtocolJobsManager::new("SyncForkProtocol", channel),
-            state,
-        }))
-    }
-
-    async fn handle_receive_order(self: Arc<Self>) -> Result<()> {
-        debug!("ProtocolSyncForks::handle_receive_order() [START]");
-        loop {
-            let order = self.order_sub.receive().await?;
-
-            debug!("ProtocolSyncForks::handle_receive_order() received {:?}", order);
-
-            // Extra validations can be added here.
-            let proposals = self.state.read().await.consensus.proposals.clone();
-            let response = ForkResponse { proposals };
-            self.channel.send(response).await?;
-        }
-    }
-}
-
-#[async_trait]
-impl ProtocolBase for ProtocolSyncForks {
-    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-        debug!("ProtocolSyncForks::start() [START]");
-        self.jobsman.clone().start(executor.clone());
-        self.jobsman.clone().spawn(self.clone().handle_receive_order(), executor.clone()).await;
-        debug!("ProtocolSyncForks::start() [END]");
-        Ok(())
-    }
-
-    fn name(&self) -> &'static str {
-        "ProtocolSyncForks"
-    }
-}

src/consensus2/state.rs

@@ -1,10 +1,10 @@
 // TODO: Use sets instead of vectors where possible.
 
-use std::time::Duration;
+use std::{io, time::Duration};
 
 use async_std::sync::{Arc, RwLock};
 use chrono::{NaiveDateTime, Utc};
-use fxhash::FxBuildHasher;
+use fxhash::{FxBuildHasher, FxHasher};
 use indexmap::IndexMap;
 use log::{debug, error, info, warn};
 use rand::{rngs::OsRng, Rng};
@@ -19,16 +19,18 @@ use crate::{
         keypair::{PublicKey, SecretKey},
         schnorr::{SchnorrPublic, SchnorrSecret},
     },
-    util::serial::{serialize, Encodable},
+    net,
+    util::serial::{serialize, Decodable, Encodable, SerialDecodable, SerialEncodable, VarInt},
     Result,
 };
 
 type FxIndexMap<K, V> = IndexMap<K, V, FxBuildHasher>;
 
-const DELTA: u64 = 10;
+/// `2 * DELTA` represents epoch time
+pub const DELTA: u64 = 10;
 
 /// This struct represents the information required by the consensus algorithm
-#[derive(Debug)]
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct ConsensusState {
     /// Genesis block creation timestamp
     pub genesis_ts: Timestamp,
@@ -61,6 +63,32 @@ impl ConsensusState {
     }
 }
 
+/// Auxiliary structure used for consensus syncing.
+#[derive(Debug, SerialEncodable, SerialDecodable)]
+pub struct ConsensusRequest {
+    /// Validator ID
+    pub id: u64,
+}
+
+impl net::Message for ConsensusRequest {
+    fn name() -> &'static str {
+        "consensusrequest"
+    }
+}
+
+/// Auxiliary structure used for consensus syncing.
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct ConsensusResponse {
+    /// Hot/live data used by the consensus algorithm
+    pub consensus: ConsensusState,
+}
+
+impl net::Message for ConsensusResponse {
+    fn name() -> &'static str {
+        "consensusresponse"
+    }
+}
+
 /// Atomic pointer to validator state.
 pub type ValidatorStatePtr = Arc<RwLock<ValidatorState>>;
 
@@ -78,6 +106,8 @@ pub struct ValidatorState {
     pub blockchain: Blockchain,
     /// Pending transactions
     pub unconfirmed_txs: Vec<Tx>,
+    /// Participation flag
+    pub participating: bool,
 }
 
 impl ValidatorState {
@@ -94,6 +124,7 @@ impl ValidatorState {
         let consensus = ConsensusState::new(genesis_ts, genesis_data)?;
         let blockchain = Blockchain::new(db, genesis_ts, genesis_data)?;
         let unconfirmed_txs = vec![];
+        let participating = false;
 
         let state = Arc::new(RwLock::new(ValidatorState {
             id,
@@ -102,6 +133,7 @@ impl ValidatorState {
             consensus,
             blockchain,
             unconfirmed_txs,
+            participating,
         }));
 
         Ok(state)
@@ -239,6 +271,11 @@ impl ValidatorState {
     /// Receive the proposed block, verify its sender (epoch leader),
     /// and proceed with voting on it.
     pub fn receive_proposal(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
+        // Node hasn't started participating
+        if !self.participating {
+            return Ok(None)
+        }
+
         let leader = self.epoch_leader();
         if leader != proposal.id {
             warn!(
@@ -270,7 +307,6 @@ impl ValidatorState {
     /// is created. The node votes on the proposal only if it extends the
     /// longest notarized fork chain it has seen.
     pub fn vote(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
-        self.zero_participants_check();
         let mut proposal = proposal.clone();
 
         // Generate proposal hash
@@ -328,7 +364,7 @@ impl ValidatorState {
     }
 
     /// Given a proposal, find the index of the chain it extends.
-    pub fn find_extended_chain_index(&self, proposal: &BlockProposal) -> Result<i64> {
+    pub fn find_extended_chain_index(&mut self, proposal: &BlockProposal) -> Result<i64> {
         for (index, chain) in self.consensus.proposals.iter().enumerate() {
             let last = chain.proposals.last().unwrap();
             let hash = last.hash();
@@ -362,6 +398,11 @@ impl ValidatorState {
     /// Finally, we check if the notarization of the proposal can finalize
     /// parent proposals in its chain.
     pub fn receive_vote(&mut self, vote: &Vote) -> Result<(bool, Option<Vec<BlockInfo>>)> {
+        // Node hasn't started participating
+        if !self.participating {
+            return Ok((false, None))
+        }
+
         let mut encoded_proposal = vec![];
 
         match vote.proposal.encode(&mut encoded_proposal) {
@@ -378,7 +419,6 @@ impl ValidatorState {
         }
 
         let node_count = self.consensus.participants.len();
-        self.zero_participants_check();
 
         // Checking that the voter can actually vote.
         match self.consensus.participants.get(&vote.id) {
@@ -578,23 +618,6 @@ impl ValidatorState {
         true
     }
 
-    /// Prevent the extreme case scenario where network is initialized, but
-    /// some nodes have not pushed the initial participants in the map.
-    pub fn zero_participants_check(&mut self) {
-        if self.consensus.participants.is_empty() {
-            debug!("zero_participants_check(): Participants are empty, trying to add pending ones");
-            for participant in &self.consensus.pending_participants {
-                self.consensus.participants.insert(participant.id, participant.clone());
-            }
-
-            if self.consensus.participants.is_empty() {
-                debug!("zero_participants_check(): Didn't manage to add any participant, pending were empty");
-            }
-
-            self.consensus.pending_participants = Vec::new();
-        }
-    }
-
     /// Refresh the participants map, to retain only the active ones.
     /// Active nodes are considered those who joined or voted on a previous epoch.
     pub fn refresh_participants(&mut self) {
@@ -632,12 +655,19 @@ impl ValidatorState {
         for index in inactive {
             self.consensus.participants.remove(&index);
         }
+
+        if self.consensus.participants.is_empty() {
+            // If no nodes are active, node becomes a single node network.
+            let participant = Participant::new(self.id, self.current_epoch());
+            self.consensus.pending_participants.push(participant);
+        }
     }
 
     /// Utility function to reset the current consensus state.
     pub fn reset_consensus_state(&mut self) -> Result<()> {
-        let genesis_ts = self.consensus.genesis_ts.clone();
-        let genesis_block = self.consensus.genesis_block.clone();
+        let genesis_ts = self.consensus.genesis_ts;
+        let genesis_block = self.consensus.genesis_block;
+
         let consensus = ConsensusState {
             genesis_ts,
             genesis_block,
@@ -651,3 +681,26 @@ impl ValidatorState {
         Ok(())
     }
 }
+
+impl Encodable for indexmap::IndexMap<u64, Participant, std::hash::BuildHasherDefault<FxHasher>> {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += VarInt(self.len() as u64).encode(&mut s)?;
+        for c in self.iter() {
+            len += c.1.encode(&mut s)?;
+        }
+        Ok(len)
+    }
+}
+
+impl Decodable for indexmap::IndexMap<u64, Participant, std::hash::BuildHasherDefault<FxHasher>> {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let len = VarInt::decode(&mut d)?.0;
+        let mut ret = FxIndexMap::with_hasher(FxBuildHasher::default());
+        for _ in 0..len {
+            let participant: Participant = Decodable::decode(&mut d)?;
+            ret.insert(participant.id, participant);
+        }
+        Ok(ret)
+    }
+}

src/consensus2/task/fork_sync.rs → src/consensus2/task/consensus_sync.rs

@@ -2,39 +2,40 @@ use log::{info, warn};
 
 use crate::{
     consensus2::{
-        block::{ForkOrder, ForkResponse},
+        state::{ConsensusRequest, ConsensusResponse},
         ValidatorStatePtr,
     },
-    net, Result,
+    net::P2pPtr,
+    Result,
 };
 
-/// async task used for consensus fork syncing.
-pub async fn fork_sync_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Result<()> {
-    info!("Starting forks sync...");
+/// async task used for consensus state syncing.
+pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Result<()> {
+    info!("Starting consensus state sync...");
 
     // Using len here beacuse is_empty() uses unstable library feature
     // called 'exact_size_is_empty'.
     if p2p.channels().lock().await.values().len() != 0 {
-        // Nodes ask for the fork chains of the last channel peer
+        // Nodes ask for the consensus state of the last channel peer
         let channel = p2p.channels().lock().await.values().last().unwrap().clone();
 
         // Communication setup
         let msg_subsystem = channel.get_message_subsystem();
-        msg_subsystem.add_dispatch::<ForkResponse>().await;
-        let response_sub = channel.subscribe_msg::<ForkResponse>().await?;
+        msg_subsystem.add_dispatch::<ConsensusResponse>().await;
+        let response_sub = channel.subscribe_msg::<ConsensusResponse>().await?;
 
-        // Node creates a `ForkOrder` and sends it
-        let order = ForkOrder { id: state.read().await.id };
-        channel.send(order).await?;
+        // Node creates a `ConsensusRequest` and sends it
+        let request = ConsensusRequest { id: state.read().await.id };
+        channel.send(request).await?;
 
         // Node stores response data. Extra validations can be added here.
         let response = response_sub.receive().await?;
-        state.write().await.consensus.proposals = response.proposals.clone();
+        state.write().await.consensus = response.consensus.clone();
     } else {
         warn!("Node is not connected to other nodes, resetting consensus state.");
         state.write().await.reset_consensus_state()?;
     }
 
-    info!("Forks synced!");
+    info!("Consensus state synced!");
     Ok(())
 }

src/consensus2/task/mod.rs

@@ -1,8 +1,10 @@
+// TODO: Handle ? with matches in these files. They should be robust.
+
 mod block_sync;
 pub use block_sync::block_sync_task;
 
-mod fork_sync;
-pub use fork_sync::fork_sync_task;
+mod consensus_sync;
+pub use consensus_sync::consensus_sync_task;
 
 mod proposal;
 pub use proposal::proposal_task;

src/consensus2/task/proposal.rs

@@ -1,5 +1,8 @@
+use std::time::Duration;
+
 use log::{debug, error, info};
 
+use super::consensus_sync_task;
 use crate::{
     consensus2::{state::ValidatorStatePtr, Participant},
     net,
@@ -8,7 +11,32 @@ use crate::{
 
 /// async task used for participating in the consensus protocol
 pub async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
-    // Node signals the network that it starts participating
+    // Node waits just before the current or next epoch end,
+    // so it can start syncing latest state.
+    let mut seconds_until_next_epoch = state.read().await.next_epoch_start();
+    let one_sec = Duration::new(1, 0);
+    loop {
+        if seconds_until_next_epoch > one_sec {
+            seconds_until_next_epoch -= one_sec;
+            break
+        }
+        info!("Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
+        sleep(seconds_until_next_epoch.as_secs()).await;
+        seconds_until_next_epoch = state.read().await.next_epoch_start();
+    }
+    info!("Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
+    sleep(seconds_until_next_epoch.as_secs()).await;
+
+    // Node syncs its consensus state
+    match consensus_sync_task(p2p.clone(), state.clone()).await {
+        Ok(()) => {}
+        Err(e) => {
+            error!("Failed syncing consensus state: {}. Quitting consensus.", e);
+            return
+        }
+    }
+
+    // Node signals the network that it will start participating
     let participant = Participant::new(state.read().await.id, state.read().await.current_epoch());
     state.write().await.append_participant(participant.clone());
 
@@ -17,6 +45,9 @@ pub async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
         Err(e) => error!("Failed broadcasting consensus participation: {}", e),
     }
 
+    // Note modifies its participating flag to true.
+    state.write().await.participating = true;
+
     loop {
         let seconds_until_next_epoch = state.read().await.next_epoch_start().as_secs();
         info!(target: "consensus", "Waiting for next epoch ({}) sec)...", seconds_until_next_epoch);