validatod: misc fixes and cleanups

script/research/validatord/src/main.rs

@@ -1,4 +1,4 @@
-use std::{net::SocketAddr, path::PathBuf, sync::Arc, thread};
+use std::{net::SocketAddr, path::PathBuf, sync::Arc, thread, time::Duration};
 
 use async_executor::Executor;
 use async_trait::async_trait;
@@ -13,9 +13,8 @@ use structopt_toml::StructOptToml;
 use darkfi::{
     consensus::{
         block::{BlockOrder, BlockResponse},
-        blockchain::{ForkOrder, ForkResponse},
         participant::Participant,
-        state::{ValidatorState, ValidatorStatePtr},
+        state::{ConsensusRequest, ConsensusResponse, ValidatorState, ValidatorStatePtr},
         tx::Tx,
     },
     net,
@@ -38,8 +37,8 @@ use darkfi::{
 
 use validatord::protocols::{
     protocol_participant::ProtocolParticipant, protocol_proposal::ProtocolProposal,
-    protocol_sync::ProtocolSync, protocol_sync_forks::ProtocolSyncForks, protocol_tx::ProtocolTx,
-    protocol_vote::ProtocolVote,
+    protocol_sync::ProtocolSync, protocol_sync_consensus::ProtocolSyncConsensus,
+    protocol_tx::ProtocolTx, protocol_vote::ProtocolVote,
 };
 
 const CONFIG_FILE: &str = r"validatord_config.toml";
@@ -152,28 +151,28 @@ async fn syncing_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Result<()>
     Ok(())
 }
 
-async fn syncing_forks_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Result<()> {
-    info!("Node starts syncing forks...");
+async fn syncing_consensus_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Result<()> {
+    info!("Node starts syncing consensus state...");
     // Using len here because is_empty() uses unstable library feature '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 message_subsytem = channel.get_message_subsystem();
-        message_subsytem.add_dispatch::<ForkResponse>().await;
+        message_subsytem.add_dispatch::<ConsensusResponse>().await;
         let response_sub = channel
-            .subscribe_msg::<ForkResponse>()
+            .subscribe_msg::<ConsensusResponse>()
             .await
-            .expect("Missing ForkResponse dispatcher!");
+            .expect("Missing ConsensusResponse dispatcher!");
 
-        // Node creates a BlockOrder and sends it
-        let order = ForkOrder { id: state.read().unwrap().id };
-        channel.send(order).await?;
+        // Node creates a ConsensusRequest and sends it
+        let request = ConsensusRequest { id: state.read().unwrap().id };
+        channel.send(request).await?;
 
         // Node stores responce data. Extra validations can be added here.
         let response = response_sub.receive().await?;
-        state.write().unwrap().consensus.proposals = response.proposals.clone();
+        state.write().unwrap().consensus = response.consensus.clone();
     } else {
         info!("Node is not connected to other nodes, resetting consensus state.");
         state.write().unwrap().reset_consensus_state()?;
@@ -184,28 +183,47 @@ async fn syncing_forks_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Resul
 }
 
 async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
-    // Node syncs its fork chains
-    let result = syncing_forks_task(p2p.clone(), state.clone()).await;
+    // 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().unwrap().next_epoch_start();
+    let one_sec = Duration::new(1, 0);
+    loop {
+        if seconds_until_next_epoch > one_sec {
+            seconds_until_next_epoch = seconds_until_next_epoch - one_sec;
+            break
+        }
+        info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
+        thread::sleep(seconds_until_next_epoch);
+        seconds_until_next_epoch = state.read().unwrap().next_epoch_start();
+    }
+    info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
+    thread::sleep(seconds_until_next_epoch);
+
+    // Node syncs its consensus state
+    let result = syncing_consensus_task(p2p.clone(), state.clone()).await;
     match result {
         Ok(()) => (),
-        Err(e) => error!("Sync forks failed. Error: {:?}", e),
+        Err(e) => error!("Sync consensus state failed. Error: {:?}", e),
     }
 
-    // Node signals the network that it starts participating
+    // Node signals the network that it will start participating
     let participant =
         Participant::new(state.read().unwrap().id, state.read().unwrap().current_epoch());
     state.write().unwrap().append_participant(participant.clone());
-    let result = p2p.broadcast(participant).await;
+    let result = p2p.broadcast(participant.clone()).await;
     match result {
         Ok(()) => info!("Participation message broadcasted successfuly."),
         Err(e) => error!("Broadcast failed. Error: {:?}", e),
     }
 
-    // After initialization node should wait for next epoch
+    // After initialization node waits for next epoch to start participating
     let seconds_until_next_epoch = state.read().unwrap().next_epoch_start();
     info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
     thread::sleep(seconds_until_next_epoch);
 
+    // Node modifies its participating flag to true
+    state.write().unwrap().participating = true;
+
     loop {
         // Node refreshes participants records
         state.write().unwrap().refresh_participants();
@@ -268,7 +286,7 @@ async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
             }
         };
 
-        // Node waits untile next epoch
+        // Node waits until next epoch
         let seconds_until_next_epoch = state.read().unwrap().next_epoch_start();
         info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
         thread::sleep(seconds_until_next_epoch);
@@ -399,7 +417,7 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
     registry
         .register(net::SESSION_ALL, move |channel, _consensus_p2p| {
             let state = state2.clone();
-            async move { ProtocolSyncForks::init(channel, state).await }
+            async move { ProtocolSyncConsensus::init(channel, state).await }
         })
         .await;
 

script/research/validatord/src/protocols/mod.rs

@@ -1,13 +1,13 @@
 pub mod protocol_participant;
 pub mod protocol_proposal;
 pub mod protocol_sync;
-pub mod protocol_sync_forks;
+pub mod protocol_sync_consensus;
 pub mod protocol_tx;
 pub mod protocol_vote;
 
 pub use protocol_participant::ProtocolParticipant;
 pub use protocol_proposal::ProtocolProposal;
 pub use protocol_sync::ProtocolSync;
-pub use protocol_sync_forks::ProtocolSyncForks;
+pub use protocol_sync_consensus::ProtocolSyncConsensus;
 pub use protocol_tx::ProtocolTx;
 pub use protocol_vote::ProtocolVote;

script/research/validatord/src/protocols/protocol_participant.rs

@@ -49,12 +49,10 @@ impl ProtocolParticipant {
                 "ProtocolParticipant::handle_receive_participant() received {:?}",
                 participant
             );
-            if self.state.write().unwrap().append_participant((*participant).clone()) {
-                let pending_participants =
-                    self.state.read().unwrap().consensus.pending_participants.clone();
-                for pending_participant in pending_participants {
-                    self.p2p.broadcast(pending_participant.clone()).await?;
-                }
+
+            let participant_copy = (*participant).clone();
+            if self.state.write().unwrap().append_participant(participant_copy.clone()) {
+                self.p2p.broadcast(participant_copy).await?;
             }
         }
     }

script/research/validatord/src/protocols/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,14 +39,14 @@ impl ProtocolSync {
         message_subsytem.add_dispatch::<BlockOrder>().await;
         message_subsytem.add_dispatch::<BlockInfo>().await;
 
-        let order_sub =
+        let request_sub =
             channel.subscribe_msg::<BlockOrder>().await.expect("Missing BlockOrder dispatcher!");
         let block_sub =
             channel.subscribe_msg::<BlockInfo>().await.expect("Missing BlockInfo dispatcher!");
 
         Arc::new(Self {
             channel: channel.clone(),
-            order_sub,
+            request_sub,
             block_sub,
             jobsman: ProtocolJobsManager::new("SyncProtocol", channel),
             state,
@@ -55,14 +55,14 @@ impl ProtocolSync {
         })
     }
 
-    async fn handle_receive_order(self: Arc<Self>) -> Result<()> {
-        debug!(target: "ircd", "ProtocolSync::handle_receive_tx() [START]");
+    async fn handle_receive_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "ircd", "ProtocolSync::handle_receive_request() [START]");
         loop {
-            let order = self.order_sub.receive().await?;
+            let order = self.request_sub.receive().await?;
 
             debug!(
                 target: "ircd",
-                "ProtocolSync::handle_receive_order() received {:?}",
+                "ProtocolSync::handle_receive_request() received {:?}",
                 order
             );
 
@@ -107,7 +107,7 @@ impl ProtocolBase for ProtocolSync {
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "ircd", "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!(target: "ircd", "ProtocolSync::start() [END]");
         Ok(())

script/research/validatord/src/protocols/protocol_sync_consensus.rs

@@ -0,0 +1,72 @@
+use async_executor::Executor;
+use async_trait::async_trait;
+
+use darkfi::{
+    consensus::state::{ConsensusRequest, ConsensusResponse, ValidatorStatePtr},
+    net::{
+        ChannelPtr, MessageSubscription, ProtocolBase, ProtocolBasePtr, ProtocolJobsManager,
+        ProtocolJobsManagerPtr,
+    },
+    Result,
+};
+use log::debug;
+use std::sync::Arc;
+
+pub struct ProtocolSyncConsensus {
+    channel: ChannelPtr,
+    request_sub: MessageSubscription<ConsensusRequest>,
+    jobsman: ProtocolJobsManagerPtr,
+    state: ValidatorStatePtr,
+}
+
+impl ProtocolSyncConsensus {
+    pub async fn init(channel: ChannelPtr, state: ValidatorStatePtr) -> ProtocolBasePtr {
+        let message_subsytem = channel.get_message_subsystem();
+        message_subsytem.add_dispatch::<ConsensusRequest>().await;
+
+        let request_sub = channel
+            .subscribe_msg::<ConsensusRequest>()
+            .await
+            .expect("Missing ConsensusRequest dispatcher!");
+
+        Arc::new(Self {
+            channel: channel.clone(),
+            request_sub,
+            jobsman: ProtocolJobsManager::new("SyncConsensusProtocol", channel),
+            state,
+        })
+    }
+
+    async fn handle_receive_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "ircd", "ProtocolSyncConsensus::handle_receive_request() [START]");
+        loop {
+            let order = self.request_sub.receive().await?;
+
+            debug!(
+                target: "ircd",
+                "ProtocolSyncConsensus::handle_receive_request() received {:?}",
+                order
+            );
+
+            // Extra validations can be added here.
+            let consensus = self.state.read().unwrap().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!(target: "ircd", "ProtocolSyncConsensus::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_request(), executor.clone()).await;
+        debug!(target: "ircd", "ProtocolSyncConsensus::start() [END]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolSyncConsensus"
+    }
+}

script/research/validatord/src/protocols/protocol_sync_forks.rs

@@ -1,73 +0,0 @@
-use async_executor::Executor;
-use async_trait::async_trait;
-
-use darkfi::{
-    consensus::{
-        blockchain::{ForkOrder, ForkResponse},
-        state::ValidatorStatePtr,
-    },
-    net::{
-        ChannelPtr, MessageSubscription, ProtocolBase, ProtocolBasePtr, ProtocolJobsManager,
-        ProtocolJobsManagerPtr,
-    },
-    Result,
-};
-use log::debug;
-use std::sync::Arc;
-
-pub struct ProtocolSyncForks {
-    channel: ChannelPtr,
-    order_sub: MessageSubscription<ForkOrder>,
-    jobsman: ProtocolJobsManagerPtr,
-    state: ValidatorStatePtr,
-}
-
-impl ProtocolSyncForks {
-    pub async fn init(channel: ChannelPtr, state: ValidatorStatePtr) -> ProtocolBasePtr {
-        let message_subsytem = channel.get_message_subsystem();
-        message_subsytem.add_dispatch::<ForkOrder>().await;
-
-        let order_sub =
-            channel.subscribe_msg::<ForkOrder>().await.expect("Missing ForkOrder dispatcher!");
-
-        Arc::new(Self {
-            channel: channel.clone(),
-            order_sub,
-            jobsman: ProtocolJobsManager::new("SyncForkProtocol", channel),
-            state,
-        })
-    }
-
-    async fn handle_receive_order(self: Arc<Self>) -> Result<()> {
-        debug!(target: "ircd", "ProtocolSyncForks::handle_receive_tx() [START]");
-        loop {
-            let order = self.order_sub.receive().await?;
-
-            debug!(
-                target: "ircd",
-                "ProtocolSyncForks::handle_receive_order() received {:?}",
-                order
-            );
-
-            // Extra validations can be added here.
-            let proposals = self.state.read().unwrap().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!(target: "ircd", "ProtocolSyncForks::start() [START]");
-        self.jobsman.clone().start(executor.clone());
-        self.jobsman.clone().spawn(self.clone().handle_receive_order(), executor.clone()).await;
-        debug!(target: "ircd", "ProtocolSyncForks::start() [END]");
-        Ok(())
-    }
-
-    fn name(&self) -> &'static str {
-        "ProtocolSyncForks"
-    }
-}

src/consensus/blockchain.rs

@@ -3,7 +3,7 @@ use std::io;
 use log::debug;
 
 use crate::{
-    impl_vec, net,
+    impl_vec,
     util::serial::{Decodable, Encodable, SerialDecodable, SerialEncodable, VarInt},
     Result,
 };
@@ -201,29 +201,3 @@ impl ProposalsChain {
 }
 
 impl_vec!(ProposalsChain);
-
-/// Auxilary 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"
-    }
-}
-
-/// Auxilary structure used for forks syncing.
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct ForkResponse {
-    /// Fork chains containing block proposals
-    pub proposals: Vec<ProposalsChain>,
-}
-
-impl net::Message for ForkResponse {
-    fn name() -> &'static str {
-        "forkresponse"
-    }
-}

src/consensus/state.rs

@@ -14,6 +14,7 @@ use crate::{
         keypair::{PublicKey, SecretKey},
         schnorr::{SchnorrPublic, SchnorrSecret},
     },
+    net,
     util::serial::{deserialize, serialize, Encodable, SerialDecodable, SerialEncodable},
     Error, Result,
 };
@@ -28,11 +29,11 @@ use super::{
     vote::Vote,
 };
 
-const DELTA: u64 = 60;
+pub const DELTA: u64 = 10;
 const SLED_CONSESUS_STATE_TREE: &[u8] = b"_consensus_state";
 
 /// This struct represents the information required by the consensus algorithm.
-#[derive(Debug, SerialEncodable, SerialDecodable)]
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct ConsensusState {
     /// Genesis block creation timestamp
     pub genesis: Timestamp,
@@ -67,6 +68,32 @@ impl ConsensusState {
     }
 }
 
+/// Auxilary 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"
+    }
+}
+
+/// Auxilary 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>>;
 
@@ -88,6 +115,8 @@ pub struct ValidatorState {
     pub unconfirmed_txs: Vec<Tx>,
     /// Genesis block hash, used for validations
     pub genesis_block: blake3::Hash,
+    /// Participation flag
+    pub participating: bool,
 }
 
 impl ValidatorState {
@@ -100,6 +129,7 @@ impl ValidatorState {
         let blockchain = Blockchain::new(&db, genesis)?;
         let unconfirmed_txs = Vec::new();
         let genesis_block = blake3::hash(&serialize(&Block::genesis_block(genesis)));
+        let participating = false;
         Ok(Arc::new(RwLock::new(ValidatorState {
             id,
             secret,
@@ -109,6 +139,7 @@ impl ValidatorState {
             blockchain,
             unconfirmed_txs,
             genesis_block,
+            participating,
         })))
     }
 
@@ -148,7 +179,6 @@ impl ValidatorState {
         let epoch = self.current_epoch();
         let mut hasher = DefaultHasher::new();
         epoch.hash(&mut hasher);
-        self.zero_participants_check();
         let pos = hasher.finish() % (self.consensus.participants.len() as u64);
         self.consensus.participants.iter().nth(pos as usize).unwrap().1.id
     }
@@ -241,6 +271,11 @@ impl ValidatorState {
     /// Node receives the proposed block, verifies its sender(epoch leader),
     /// and proceeds 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 {
             debug!(
@@ -269,7 +304,6 @@ impl ValidatorState {
     /// If proposal extends the canonical blockchain, a new fork chain is created.
     /// 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
@@ -331,7 +365,7 @@ impl ValidatorState {
     }
 
     /// Given a proposal, node finds 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();
@@ -363,6 +397,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![];
         let result = vote.proposal.encode(&mut encoded_proposal);
         match result {
@@ -379,8 +418,6 @@ impl ValidatorState {
         }
 
         let nodes_count = self.consensus.participants.len();
-        self.zero_participants_check();
-
         // Checking that the voter can actually vote.
         match self.consensus.participants.get(&vote.id) {
             Some(participant) => {
@@ -538,17 +575,6 @@ impl ValidatorState {
         true
     }
 
-    /// This 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.len() == 0 {
-            for participant in &self.consensus.pending_participants {
-                self.consensus.participants.insert(participant.id, participant.clone());
-            }
-            self.consensus.pending_participants = Vec::new();
-        }
-    }
-
     /// Node refreshes participants map, to retain only the active ones.
     /// Active nodes are considered those who joined or voted on previous epoch.
     pub fn refresh_participants(&mut self) {
@@ -577,6 +603,12 @@ 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);
+        }
     }
 
     /// Util function to save the current consensus state to provided file path.