consensus: votes chopped

bin/darkfid/src/main.rs

@@ -11,7 +11,6 @@ use darkfi::{
     consensus::{
         proto::{
             ProtocolParticipant, ProtocolProposal, ProtocolSync, ProtocolSyncConsensus, ProtocolTx,
-            ProtocolVote,
         },
         state::ValidatorStatePtr,
         task::{block_sync_task, proposal_task},
@@ -410,20 +409,6 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
                 })
                 .await;
 
-            let _state = state.clone();
-            let _sync_p2p = sync_p2p.clone().unwrap();
-            registry
-                .register(net::SESSION_ALL, move |channel, p2p| {
-                    let state = _state.clone();
-                    let __sync_p2p = _sync_p2p.clone();
-                    async move {
-                        ProtocolVote::init(channel, state, __sync_p2p, p2p)
-                            .await
-                            .unwrap()
-                    }
-                })
-                .await;
-
             let _state = state.clone();
             registry
                 .register(net::SESSION_ALL, move |channel, p2p| {

src/blockchain/metadatastore.rs

@@ -25,7 +25,6 @@ impl StreamletMetadataStore {
             let genesis_block = Block::genesis_block(genesis_ts, genesis_data);
 
             let metadata = StreamletMetadata {
-                votes: vec![],
                 notarized: true,
                 finalized: true,
                 participants: vec![],

src/consensus/metadata.rs

@@ -1,4 +1,4 @@
-use super::{Participant, Vote};
+use super::Participant;
 use rand::rngs::OsRng;
 
 use crate::{
@@ -82,8 +82,6 @@ impl OuroborosMetadata {
 /// consensus protocol.
 #[derive(Debug, Clone, Default, SerialEncodable, SerialDecodable)]
 pub struct StreamletMetadata {
-    /// Slot votes
-    pub votes: Vec<Vote>,
     /// Block notarization flag
     pub notarized: bool,
     /// Block finalization flag
@@ -94,6 +92,6 @@ pub struct StreamletMetadata {
 
 impl StreamletMetadata {
     pub fn new(participants: Vec<Participant>) -> Self {
-        Self { votes: vec![], notarized: false, finalized: false, participants }
+        Self { notarized: false, finalized: false, participants }
     }
 }

src/consensus/mod.rs

@@ -12,10 +12,6 @@ pub use metadata::{
 pub mod participant;
 pub use participant::Participant;
 
-/// Consensus vote
-pub mod vote;
-pub use vote::Vote;
-
 /// Consensus state
 pub mod state;
 pub use state::{ValidatorState, ValidatorStatePtr};

src/consensus/proto/mod.rs

@@ -10,10 +10,6 @@ pub use protocol_proposal::ProtocolProposal;
 mod protocol_tx;
 pub use protocol_tx::ProtocolTx;
 
-/// Consensus vote protocol
-mod protocol_vote;
-pub use protocol_vote::ProtocolVote;
-
 /// Validator + Replicator blockchain sync protocol
 mod protocol_sync;
 pub use protocol_sync::ProtocolSync;

src/consensus/proto/protocol_proposal.rs

@@ -63,22 +63,11 @@ impl ProtocolProposal {
 
             let proposal_copy = (*proposal).clone();
 
-            let vote = match self.state.write().await.receive_proposal(&proposal_copy).await {
-                Ok(v) => {
-                    if v.is_none() {
-                        debug!("ProtocolProposal::handle_receive_proposal(): Node didn't vote for proposed block.");
-                        continue
-                    }
-                    v.unwrap()
-                }
-                Err(e) => {
-                    debug!("ProtocolProposal::handle_receive_proposal(): error processing proposal: {}", e);
-                    continue
-                }
-            };
-
-            if let Err(e) = self.state.write().await.receive_vote(&vote).await {
-                error!("ProtocolProposal::handle_receive_proposal(): receive_vote error: {}", e);
+            if let Err(e) = self.state.write().await.receive_proposal(&proposal_copy).await {
+                error!(
+                    "ProtocolProposal::handle_receive_proposal(): receive_proposal error: {}",
+                    e
+                );
                 continue
             }
 
@@ -89,11 +78,6 @@ impl ProtocolProposal {
                     e
                 );
             };
-
-            // Broadcast vote
-            if let Err(e) = self.p2p.broadcast(vote).await {
-                error!("ProtocolProposal::handle_receive_proposal(): vote broadcast fail: {}", e);
-            }
         }
     }
 }

src/consensus/proto/protocol_vote.rs

@@ -1,114 +0,0 @@
-use async_std::sync::Arc;
-
-use async_executor::Executor;
-use async_trait::async_trait;
-use log::{debug, error};
-use url::Url;
-
-use crate::{
-    consensus::{ValidatorStatePtr, Vote},
-    net::{
-        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
-        ProtocolJobsManager, ProtocolJobsManagerPtr,
-    },
-    Result,
-};
-
-pub struct ProtocolVote {
-    vote_sub: MessageSubscription<Vote>,
-    jobsman: ProtocolJobsManagerPtr,
-    state: ValidatorStatePtr,
-    sync_p2p: P2pPtr,
-    consensus_p2p: P2pPtr,
-    channel_address: Url,
-}
-
-impl ProtocolVote {
-    pub async fn init(
-        channel: ChannelPtr,
-        state: ValidatorStatePtr,
-        sync_p2p: P2pPtr,
-        consensus_p2p: P2pPtr,
-    ) -> Result<ProtocolBasePtr> {
-        debug!("Adding ProtocolVote to the protocol registry");
-        let msg_subsystem = channel.get_message_subsystem();
-        msg_subsystem.add_dispatch::<Vote>().await;
-
-        let vote_sub = channel.subscribe_msg::<Vote>().await?;
-        let channel_address = channel.address();
-
-        Ok(Arc::new(Self {
-            vote_sub,
-            jobsman: ProtocolJobsManager::new("VoteProtocol", channel),
-            state,
-            sync_p2p,
-            consensus_p2p,
-            channel_address,
-        }))
-    }
-
-    async fn handle_receive_vote(self: Arc<Self>) -> Result<()> {
-        debug!("ProtocolVote::handle_receive_vote() [START]");
-        let exclude_list = vec![self.channel_address.clone()];
-        loop {
-            let vote = match self.vote_sub.receive().await {
-                Ok(v) => v,
-                Err(e) => {
-                    error!("ProtocolVote::handle_receive_vote(): recv fail: {}", e);
-                    continue
-                }
-            };
-
-            debug!("ProtocolVote::handle_receive_vote() recv: {:?}", vote);
-
-            let vote_copy = (*vote).clone();
-
-            let (voted, to_broadcast) =
-                match self.state.write().await.receive_vote(&vote_copy).await {
-                    Ok(v) => v,
-                    Err(e) => {
-                        error!("handle_receive_vote(): receive_vote() fail: {}", e);
-                        continue
-                    }
-                };
-
-            if voted {
-                if let Err(e) =
-                    self.consensus_p2p.broadcast_with_exclude(vote_copy, &exclude_list).await
-                {
-                    error!("handle_receive_vote(): consensus p2p broadcast fail: {}", e);
-                    continue
-                };
-
-                // Broadcast finalized blocks info, if any
-                if let Some(blocks) = to_broadcast {
-                    debug!("handle_receive_vote(): Broadcasting finalized blocks");
-                    for info in blocks {
-                        if let Err(e) = self.sync_p2p.broadcast(info).await {
-                            error!("handle_receive_vote(): sync p2p broadcast fail: {}", e);
-                            // TODO: Should we quit broadcasting if one fails?
-                            continue
-                        }
-                    }
-                } else {
-                    debug!("handle_receive_vote(): No finalized blocks to broadcast");
-                };
-            }
-        }
-    }
-}
-
-#[async_trait]
-impl ProtocolBase for ProtocolVote {
-    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-        debug!("ProtocolVote::start() [START]");
-        self.jobsman.clone().start(executor.clone());
-        self.jobsman.clone().spawn(self.clone().handle_receive_vote(), executor.clone()).await;
-        debug!("ProtocolVote::start() [END]");
-        Ok(())
-    }
-
-    fn name(&self) -> &'static str {
-        "ProtocolVote"
-    }
-}

src/consensus/state.rs

@@ -14,7 +14,7 @@ use rand::rngs::OsRng;
 
 use super::{
     Block, BlockInfo, BlockProposal, Header, OuroborosMetadata, Participant, ProposalChain,
-    StreamletMetadata, Vote,
+    StreamletMetadata,
 };
 
 use crate::{
@@ -32,7 +32,7 @@ use crate::{
         state::{state_transition, ProgramState, StateUpdate},
         Client, MemoryState, State,
     },
-    serial::{serialize, Encodable, SerialDecodable, SerialEncodable},
+    serial::{serialize, SerialDecodable, SerialEncodable},
     tx::Transaction,
     util::time::Timestamp,
     Result,
@@ -54,9 +54,6 @@ pub struct ConsensusState {
     pub genesis_block: blake3::Hash,
     /// Fork chains containing block proposals
     pub proposals: Vec<ProposalChain>,
-    /// Orphan votes pool, in case a vote reaches a node before the
-    /// corresponding block
-    pub orphan_votes: Vec<Vote>,
     /// Validators currently participating in the consensus
     pub participants: BTreeMap<Address, Participant>,
     /// Validators to be added on the next slot as participants
@@ -74,7 +71,6 @@ impl ConsensusState {
             genesis_ts,
             genesis_block,
             proposals: vec![],
-            orphan_votes: vec![],
             participants: BTreeMap::new(),
             pending_participants: vec![],
             refreshed: 0,
@@ -245,11 +241,11 @@ impl ValidatorState {
         Ok(last_slot)
     }
 
-    /// Calculates seconds until next slot starting time.
-    /// Slots durationis configured using the delta value.
-    pub fn next_slot_start(&self) -> Duration {
+    /// Calculates seconds until next Nth slot starting time.
+    /// Slots duration is configured using the delta value.
+    pub fn next_n_slot_start(&self, n: u64) -> Duration {
         let start_time = NaiveDateTime::from_timestamp(self.consensus.genesis_ts.0, 0);
-        let current_slot = self.current_slot() + 1;
+        let current_slot = self.current_slot() + n;
         let next_slot_start = (current_slot * (2 * DELTA)) + (start_time.timestamp() as u64);
         let next_slot_start = NaiveDateTime::from_timestamp(next_slot_start as i64, 0);
         let current_time = NaiveDateTime::from_timestamp(Utc::now().timestamp(), 0);
@@ -287,11 +283,11 @@ impl ValidatorState {
     }
 
     /// Generate a block proposal for the current slot, containing all
-    /// unconfirmed transactions. Proposal extends the longest notarized fork
+    /// unconfirmed transactions. Proposal extends the longest fork
     /// chain the node is holding.
     pub fn propose(&self) -> Result<Option<BlockProposal>> {
         let slot = self.current_slot();
-        let (prev_hash, index) = self.longest_notarized_chain_last_hash().unwrap();
+        let (prev_hash, index) = self.longest_chain_last_hash().unwrap();
         let unproposed_txs = self.unproposed_txs(index);
 
         let mut tree = BridgeTree::<MerkleNode, MERKLE_DEPTH>::new(100);
@@ -309,6 +305,11 @@ impl ValidatorState {
         let m = StakeholderMetadata::new(signed_proposal, self.address);
         let om = OuroborosMetadata::default();
         let sm = StreamletMetadata::new(self.consensus.participants.values().cloned().collect());
+        
+        // TODO: [PLACEHOLDER] Add balance proof creation
+        // TODO: [PLACEHOLDER] Add crypsinous leader proof creation (to replace balance proof)
+        // TODO: [PLACEHOLDER] Add rewards calculation (proof?)
+        // TODO: [PLACEHOLDER] Create and add rewards transaction
         Ok(Some(BlockProposal::new(header, unproposed_txs, m, om, sm)))
     }
 
@@ -337,24 +338,24 @@ impl ValidatorState {
         unproposed_txs
     }
 
-    /// Finds the longest fully notarized blockchain the node holds and
+    /// Finds the longest blockchain the node holds and
     /// returns the last block hash and the chain index.
-    pub fn longest_notarized_chain_last_hash(&self) -> Result<(blake3::Hash, i64)> {
-        let mut longest_notarized_chain: Option<ProposalChain> = None;
+    pub fn longest_chain_last_hash(&self) -> Result<(blake3::Hash, i64)> {
+        let mut longest: Option<ProposalChain> = None;
         let mut length = 0;
         let mut index = -1;
 
         if !self.consensus.proposals.is_empty() {
             for (i, chain) in self.consensus.proposals.iter().enumerate() {
-                if chain.notarized() && chain.proposals.len() > length {
-                    longest_notarized_chain = Some(chain.clone());
+                if chain.proposals.len() > length {
+                    longest = Some(chain.clone());
                     length = chain.proposals.len();
                     index = i as i64;
                 }
             }
         }
 
-        let hash = match longest_notarized_chain {
+        let hash = match longest {
             Some(chain) => chain.proposals.last().unwrap().block.header.headerhash(),
             None => self.blockchain.last()?.1,
         };
@@ -362,13 +363,18 @@ impl ValidatorState {
         Ok((hash, index))
     }
 
-    /// Receive the proposed block, verify its sender (slot leader),
-    /// and proceed with voting on it.
-    pub async fn receive_proposal(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
+    /// Given a proposal, the node verify its sender (slot leader), finds which blockchain
+    /// it extends and check if it can be finalized. If the proposal extends
+    /// the canonical blockchain, a new fork chain is created.
+    pub async fn receive_proposal(
+        &mut self,
+        proposal: &BlockProposal,
+    ) -> Result<Option<Vec<BlockInfo>>> {
+        let current = self.current_slot();
         // Node hasn't started participating
         match self.participating {
             Some(start) => {
-                if self.current_slot() < start {
+                if current < start {
                     return Ok(None)
                 }
             }
@@ -378,7 +384,7 @@ impl ValidatorState {
         // Node refreshes participants records
         self.refresh_participants()?;
 
-        let leader = self.slot_leader();
+        let mut leader = self.slot_leader();
         if leader.address != proposal.block.m.address {
             warn!(
                 "Received proposal not from slot leader ({}), but from ({})",
@@ -387,39 +393,43 @@ impl ValidatorState {
             return Ok(None)
         }
 
-        if !leader
-            .public_key
-            .verify(proposal.block.header.headerhash().as_bytes(), &proposal.block.m.signature)
-        {
+        if !leader.public_key.verify(
+            proposal.block.header.headerhash().as_bytes(),
+            &proposal.block.m.signature,
+        ) {
             warn!("Proposer ({}) signature could not be verified", proposal.block.m.address);
             return Ok(None)
         }
 
-        self.vote(proposal).await
-    }
+        debug!("receive_proposal(): Starting state transition validation");
+        let canon_state_clone = self.state_machine.lock().await.clone();
+        let mem_state = MemoryState::new(canon_state_clone);
 
-    /// Given a proposal, the node finds which blockchain it extends.
-    /// If the proposal extends the canonical blockchain, a new fork chain
-    /// is created. The node votes on the proposal only if it extends the
-    /// longest notarized fork chain it has seen and its state transition is valid.
-    pub async fn vote(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
-        let mut proposal = proposal.clone();
-
-        // Generate proposal hash
-        let proposal_hash = proposal.block.header.headerhash();
-
-        // Add orphan votes
-        let mut orphans = Vec::new();
-        for vote in self.consensus.orphan_votes.iter() {
-            if vote.proposal == proposal_hash {
-                proposal.block.sm.votes.push(vote.clone());
-                orphans.push(vote.clone());
+        match Self::validate_state_transitions(mem_state, &proposal.block.txs) {
+            Ok(_) => {
+                debug!("receive_proposal(): State transition valid")
+            }
+            Err(e) => {
+                warn!("receive_proposal(): State transition fail: {}", e);
+                return Ok(None)
             }
         }
 
-        for vote in orphans {
-            self.consensus.orphan_votes.retain(|v| *v != vote);
+        // TODO: [PLACEHOLDER] Add balance proof validation
+        // TODO: [PLACEHOLDER] Add crypsinous proof validation (to replace balance proof)
+        // TODO: [PLACEHOLDER] Add rewards validation
+        
+        // TODO: uncomment this after adding seen attribute to participant
+        /*
+        if current > leader.seen {
+            leader.seen = current;
         }
+        */
+
+        // Invalidating quarantine
+        leader.quarantined = None;
+
+        self.consensus.participants.insert(leader.address, leader);
 
         let index = self.find_extended_chain_index(&proposal)?;
 
@@ -427,50 +437,27 @@ impl ValidatorState {
             return Ok(None)
         }
 
-        let chain = match index {
+        let mut to_broadcast = vec![];
+        match index {
             -1 => {
                 let pc = ProposalChain::new(self.consensus.genesis_block, proposal.clone());
                 self.consensus.proposals.push(pc);
-                self.consensus.proposals.last().unwrap()
             }
             _ => {
                 self.consensus.proposals[index as usize].add(&proposal);
-                &self.consensus.proposals[index as usize]
+                match self.chain_finalization(index).await {
+                    Ok(v) => {
+                        to_broadcast = v;
+                    }
+                    Err(e) => {
+                        error!("consensus: Block finalization failed: {}", e);
+                        return Err(e)
+                    }
+                }
             }
         };
 
-        if !self.extends_notarized_chain(chain) {
-            debug!("vote(): Proposal does not extend notarized chain");
-            return Ok(None)
-        }
-
-        debug!("vote(): Starting state transition validation");
-        let canon_state_clone = self.state_machine.lock().await.clone();
-        let mem_state = MemoryState::new(canon_state_clone);
-
-        match Self::validate_state_transitions(mem_state, &proposal.block.txs) {
-            Ok(_) => {
-                debug!("vote(): State transition valid")
-            }
-            Err(e) => {
-                warn!("vote(): State transition fail: {}", e);
-                return Ok(None)
-            }
-        }
-
-        let signed_hash = self.secret.sign(&serialize(&proposal_hash));
-        Ok(Some(Vote::new(signed_hash, proposal_hash, proposal.block.header.slot, self.address)))
-    }
-
-    /// Verify if the provided chain is notarized excluding the last block.
-    pub fn extends_notarized_chain(&self, chain: &ProposalChain) -> bool {
-        for proposal in &chain.proposals[..(chain.proposals.len() - 1)] {
-            if !proposal.block.sm.notarized {
-                return false
-            }
-        }
-
-        true
+        Ok(Some(to_broadcast))
     }
 
     /// Given a proposal, find the index of the chain it extends.
@@ -518,118 +505,6 @@ impl ValidatorState {
         Ok(-1)
     }
 
-    /// Receive a vote for a proposal.
-    /// First, sender is verified using their public key.
-    /// The proposal is then searched for in the node's fork chains.
-    /// If the vote wasn't received before, it is appended to the proposal
-    /// votes list.
-    /// When a node sees 2n/3 votes for a proposal, it notarizes it.
-    /// When a proposal gets notarized, the transactions it contains are
-    /// removed from the node's unconfirmed tx list.
-    /// Finally, we check if the notarization of the proposal can finalize
-    /// parent proposals in its chain.
-    pub async fn receive_vote(&mut self, vote: &Vote) -> Result<(bool, Option<Vec<BlockInfo>>)> {
-        let current_slot = self.current_slot();
-        // Node hasn't started participating
-        match self.participating {
-            Some(start) => {
-                if current_slot < start {
-                    return Ok((false, None))
-                }
-            }
-            None => return Ok((false, None)),
-        }
-
-        // Node refreshes participants records
-        self.refresh_participants()?;
-        let node_count = self.consensus.participants.len();
-
-        // Checking that the voter can actually vote.
-        match self.consensus.participants.get(&vote.address) {
-            Some(participant) => {
-                let mut participant = participant.clone();
-                let va = vote.address;
-                if current_slot <= participant.joined {
-                    warn!("consensus: Voter ({}) joined after current slot.", va);
-                    return Ok((false, None))
-                }
-
-                let mut encoded_proposal = vec![];
-
-                if let Err(e) = vote.proposal.encode(&mut encoded_proposal) {
-                    error!("consensus: Proposal encoding failed: {:?}", e);
-                    return Ok((false, None))
-                };
-
-                if !participant.public_key.verify(&encoded_proposal, &vote.vote) {
-                    warn!("consensus: Voter ({}), signature couldn't be verified", va);
-                    return Ok((false, None))
-                }
-
-                // Updating participant vote
-                match participant.voted {
-                    Some(voted) => {
-                        if vote.slot > voted {
-                            participant.voted = Some(vote.slot);
-                        }
-                    }
-                    None => participant.voted = Some(vote.slot),
-                }
-
-                // Invalidating quarantine
-                participant.quarantined = None;
-
-                self.consensus.participants.insert(participant.address, participant);
-            }
-            None => {
-                warn!("consensus: Voter ({}) is not a participant!", vote.address);
-                return Ok((false, None))
-            }
-        }
-
-        let proposal = match self.find_proposal(&vote.proposal) {
-            Ok(v) => v,
-            Err(e) => {
-                error!("consensus: find_proposal() failed: {}", e);
-                return Err(e)
-            }
-        };
-
-        if proposal.is_none() {
-            debug!(target: "consensus", "Received vote for unknown proposal.");
-            if !self.consensus.orphan_votes.contains(vote) {
-                self.consensus.orphan_votes.push(vote.clone());
-            }
-
-            return Ok((false, None))
-        }
-
-        let (proposal, chain_idx) = proposal.unwrap();
-        if proposal.block.sm.votes.contains(vote) {
-            debug!("receive_vote(): Already seen this vote");
-            return Ok((false, None))
-        }
-
-        proposal.block.sm.votes.push(vote.clone());
-
-        let mut to_broadcast = vec![];
-        if !proposal.block.sm.notarized && proposal.block.sm.votes.len() > (2 * node_count / 3) {
-            debug!("receive_vote(): Notarized a block");
-            proposal.block.sm.notarized = true;
-            match self.chain_finalization(chain_idx).await {
-                Ok(v) => {
-                    to_broadcast = v;
-                }
-                Err(e) => {
-                    error!("consensus: Block finalization failed: {}", e);
-                    return Err(e)
-                }
-            }
-        }
-
-        Ok((true, Some(to_broadcast)))
-    }
-
     /// Search the chains we're holding for the given proposal.
     pub fn find_proposal(
         &mut self,
@@ -660,9 +535,8 @@ impl ValidatorState {
 
     /// Provided an index, the node checks if the chain can be finalized.
     /// Consensus finalization logic:
-    /// - If the node has observed the notarization of 3 consecutive
-    ///   proposals in a fork chain, it finalizes (appends to canonical
-    ///   blockchain) all proposals up to the middle block.
+    /// - If the node has observed the creation of 3 proposals in a fork chain,
+    ///   it finalizes (appends to canonical blockchain) all proposals up to the last one.
     /// When fork chain proposals are finalized, the rest of fork chains not
     /// starting by those proposals are removed.
     pub async fn chain_finalization(&mut self, chain_index: i64) -> Result<Vec<BlockInfo>> {
@@ -676,33 +550,15 @@ impl ValidatorState {
             return Ok(vec![])
         }
 
-        let mut consecutive = 0;
-        for proposal in &chain.proposals {
-            if proposal.block.sm.notarized {
-                consecutive += 1;
-                continue
-            }
-
-            break
-        }
-
-        if consecutive < 3 {
-            debug!(
-                "chain_finalization(): Less than 3 notarized blocks in chain {}, nothing to finalize",
-                chain_index
-            );
-            return Ok(vec![])
-        }
-
+        let bound = chain.proposals.len() - 1;
         let mut finalized = vec![];
-        for proposal in &mut chain.proposals[..(consecutive - 1)] {
-            proposal.block.sm.finalized = true;
+        for proposal in &mut chain.proposals[..bound] {
             finalized.push(proposal.clone().into());
         }
 
-        chain.proposals.drain(0..(consecutive - 1));
+        chain.proposals.drain(0..bound);
 
-        info!("consensus: Adding {} finalized block to canonical chain", finalized.len());
+        info!("consensus: Adding {} finalized block to canonical chain.", finalized.len());
         let blockhashes = match self.blockchain.add(&finalized) {
             Ok(v) => v,
             Err(e) => {
@@ -737,18 +593,6 @@ impl ValidatorState {
             self.consensus.proposals.retain(|c| *c != chain);
         }
 
-        // Remove orphan votes
-        let mut orphans = vec![];
-        for vote in self.consensus.orphan_votes.iter() {
-            if vote.slot <= last_slot {
-                orphans.push(vote.clone());
-            }
-        }
-
-        for vote in orphans {
-            self.consensus.orphan_votes.retain(|v| *v != vote);
-        }
-
         Ok(finalized)
     }
 
@@ -896,7 +740,6 @@ impl ValidatorState {
             genesis_ts,
             genesis_block,
             proposals: vec![],
-            orphan_votes: vec![],
             participants: BTreeMap::new(),
             pending_participants: vec![],
             refreshed: 0,

src/consensus/task/proposal.rs

@@ -13,7 +13,7 @@ use crate::{
 pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: ValidatorStatePtr) {
     // Node waits just before the current or next slot end, so it can
     // start syncing latest state.
-    let mut seconds_until_next_slot = state.read().await.next_slot_start();
+    let mut seconds_until_next_slot = state.read().await.next_n_slot_start(1);
     let one_sec = Duration::new(1, 0);
 
     loop {
@@ -24,7 +24,7 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
 
         info!("consensus: Waiting for next slot ({:?} sec)", seconds_until_next_slot);
         sleep(seconds_until_next_slot.as_secs()).await;
-        seconds_until_next_slot = state.read().await.next_slot_start();
+        seconds_until_next_slot = state.read().await.next_n_slot_start(1);
     }
 
     info!("consensus: Waiting for next slot ({:?} sec)", seconds_until_next_slot);
@@ -57,7 +57,7 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
     }
 
     loop {
-        let seconds_next_slot = state.read().await.next_slot_start().as_secs();
+        let seconds_next_slot = state.read().await.next_n_slot_start(1).as_secs();
         info!("consensus: Waiting for next slot ({} sec)", seconds_next_slot);
         sleep(seconds_next_slot).await;
 
@@ -91,25 +91,14 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
 
         info!("consensus: Node is the slot leader: Proposed block: {}", proposal);
         debug!("consensus: Full proposal: {:?}", proposal);
-        let vote = state.write().await.receive_proposal(&proposal).await;
-        let vote = match vote {
-            Ok(v) => {
-                if v.is_none() {
-                    debug!("proposal_task(): Node did not vote for the proposed block");
-                    continue
+        match state.write().await.receive_proposal(&proposal).await {
+            Ok(to_broadcast) => {
+                info!("consensus: Block proposal  saved successfully");
+                // Broadcast block to other consensus nodes
+                match consensus_p2p.broadcast(proposal).await {
+                    Ok(()) => info!("consensus: Proposal broadcasted successfully"),
+                    Err(e) => error!("consensus: Failed broadcasting proposal: {}", e),
                 }
-                v.unwrap()
-            }
-            Err(e) => {
-                error!("consensus: Failed processing proposal: {}", e);
-                continue
-            }
-        };
-
-        let result = state.write().await.receive_vote(&vote).await;
-        match result {
-            Ok((_, to_broadcast)) => {
-                info!("consensus: Vote saved successfully");
                 // Broadcast finalized blocks info, if any:
                 if let Some(blocks) = to_broadcast {
                     info!("consensus: Broadcasting finalized blocks");
@@ -124,21 +113,8 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
                 }
             }
             Err(e) => {
-                error!("consensus: Vote save failed: {}", e);
-                // TODO: Is this fallthrough ok?
+                error!("consensus: Block proposal save failed: {}", e);
             }
         }
-
-        // Broadcast block to other consensus nodes
-        match consensus_p2p.broadcast(proposal).await {
-            Ok(()) => info!("consensus: Proposal broadcasted successfully"),
-            Err(e) => error!("consensus: Failed broadcasting proposal: {}", e),
-        }
-
-        // Broadcast leader vote
-        match consensus_p2p.broadcast(vote).await {
-            Ok(()) => info!("consensus: Leader vote broadcasted successfully"),
-            Err(e) => error!("consensus: Failed broadcasting leader vote: {}", e),
-        }
     }
 }

src/consensus/vote.rs

@@ -1,30 +0,0 @@
-use crate::{
-    crypto::{address::Address, schnorr::Signature},
-    net,
-    serial::{SerialDecodable, SerialEncodable},
-};
-
-/// This struct represents a `Vote` used by the Streamlet consensus
-#[derive(Debug, Clone, PartialEq, Eq, SerialDecodable, SerialEncodable)]
-pub struct Vote {
-    /// Block signature
-    pub vote: Signature,
-    /// Block proposal hash to vote on
-    pub proposal: blake3::Hash,
-    /// Slot uid, generated by the beacon
-    pub slot: u64,
-    /// Node wallet address
-    pub address: Address,
-}
-
-impl Vote {
-    pub fn new(vote: Signature, proposal: blake3::Hash, slot: u64, address: Address) -> Self {
-        Self { vote, proposal, slot, address }
-    }
-}
-
-impl net::Message for Vote {
-    fn name() -> &'static str {
-        "vote"
-    }
-}