Replace request_proposal_to_any to request_proposal_to_superiors

Author: HoOngEe

core/src/consensus/sortition/vrf_sortition.rs

@@ -35,7 +35,7 @@ pub struct VRFSortition {
     pub vrf_inst: ECVRF,
 }
 
-#[derive(Eq, PartialEq, Clone, Debug, RlpEncodable, RlpDecodable)]
+#[derive(Eq, PartialEq, Clone, Default, Debug, RlpEncodable, RlpDecodable)]
 pub struct PriorityInfo {
     signer_idx: usize,
     priority: Priority,

core/src/consensus/tendermint/message.rs

@@ -25,7 +25,7 @@ use snap;
 
 use super::super::BitSet;
 use super::{Height, Step, View};
-use crate::consensus::Priority;
+use crate::consensus::{Priority, PriorityInfo};
 
 /// Step for the sortition round.
 /// FIXME: It has a large overlap with the previous VoteStep.
@@ -123,10 +123,16 @@ const MESSAGE_ID_COMMIT: u8 = 0x07;
 #[derive(Clone, Debug, PartialEq, RlpEncodable, RlpDecodable)]
 #[cfg_attr(test, derive(Default))]
 pub struct ProposalSummary {
-    pub priority: Priority,
+    pub priority_info: PriorityInfo,
     pub block_hash: BlockHash,
 }
 
+impl ProposalSummary {
+    pub fn priority(&self) -> Priority {
+        self.priority_info.priority()
+    }
+}
+
 #[derive(Debug, PartialEq)]
 pub enum TendermintMessage {
     ConsensusMessage(Vec<Bytes>),
@@ -442,7 +448,7 @@ mod tests {
         bit_set.set(2);
         rlp_encode_and_decode_test!(TendermintMessage::StepState {
             vote_step: VoteStep::new(10, 123, Step::Prevote),
-            proposal: Some(Default::default()),
+            proposal: Box::new(Some(Default::default())),
             lock_view: Some(2),
             known_votes: bit_set
         });

core/src/consensus/tendermint/network.rs

@@ -34,7 +34,7 @@ use super::message::*;
 use super::params::TimeoutParams;
 use super::types::{PeerState, Step, View};
 use super::worker;
-use crate::consensus::{EngineError, PriorityInfo};
+use crate::consensus::{EngineError, Priority, PriorityInfo};
 
 use super::{
     ENGINE_TIMEOUT_BROADCAST_STEP_STATE, ENGINE_TIMEOUT_BROADCAT_STEP_STATE_INTERVAL, ENGINE_TIMEOUT_TOKEN_NONCE_BASE,
@@ -154,21 +154,25 @@ impl TendermintExtension {
         }
     }
 
-    fn request_proposal_to_any(&self, round: SortitionRound) {
-        for (token, peer) in &self.peers {
-            let is_future_height_and_view = round < peer.vote_step.into();
-
-            if is_future_height_and_view {
-                self.request_proposal(token, round);
-                continue
-            }
-
-            let is_same_height_and_view = round == peer.vote_step.into();
-
-            if is_same_height_and_view && peer.proposal.is_some() {
-                self.request_proposal(token, round);
-            }
-        }
+    fn request_proposal_to_superiors(&self, round: SortitionRound, my_highest: Option<Priority>) {
+        // Request to future round peers
+        self.peers
+            .iter()
+            .filter(|(_, peer)| round < peer.vote_step.into())
+            .for_each(|(token, _)| self.request_proposal(token, round));
+
+        let current_round_peers = self.peers.iter().filter(|(_, peer)| round == peer.vote_step.into());
+        // Request to current round higher peers
+        if let Some(current_round_highest_priority) = current_round_peers
+            .clone()
+            .map(|(_id, peer)| peer.priority())
+            .max()
+            .filter(|highest_priority| *highest_priority > my_highest)
+        {
+            current_round_peers
+                .filter(|(_id, peer)| peer.priority() == current_round_highest_priority)
+                .for_each(|(token, _)| self.request_proposal(token, round))
+        };
     }
 
     fn request_proposal(&self, token: &NodeId, round: SortitionRound) {
@@ -301,21 +305,44 @@ impl NetworkExtension<Event> for TendermintExtension {
                     lock_view,
                     known_votes,
                 );
-                self.update_peer_state(token, vote_step, (*proposal).clone(), known_votes);
-                let (result, receiver) = crossbeam::unbounded();
-                self.inner
-                    .send(worker::Event::StepState {
-                        token: *token,
-                        vote_step,
-                        proposal: *proposal,
-                        lock_view,
-                        known_votes: Box::from(known_votes),
-                        result,
-                    })
-                    .unwrap();
-
-                while let Ok(message) = receiver.recv() {
-                    self.api.send(token, Arc::new(message));
+                let unchanged = match self.peers.get(token) {
+                    Some(peer_state) => *proposal == peer_state.proposal,
+                    None => false,
+                };
+                let verified = unchanged || {
+                    (*proposal)
+                        .clone()
+                        .map(|summary| {
+                            let (result, receiver) = crossbeam::bounded(1);
+                            self.inner
+                                .send(worker::Event::VerifyPriorityInfo {
+                                    height: vote_step.height,
+                                    view: vote_step.view,
+                                    priority_info: summary.priority_info,
+                                    result,
+                                })
+                                .unwrap();
+                            receiver.recv().unwrap().unwrap_or(false)
+                        })
+                        .unwrap_or(true)
+                };
+                if verified {
+                    self.update_peer_state(token, vote_step, (*proposal).clone(), known_votes);
+                    let (result, receiver) = crossbeam::unbounded();
+                    self.inner
+                        .send(worker::Event::StepState {
+                            token: *token,
+                            vote_step,
+                            proposal: *proposal,
+                            lock_view,
+                            known_votes: Box::from(known_votes),
+                            result,
+                        })
+                        .unwrap();
+
+                    while let Ok(message) = receiver.recv() {
+                        self.api.send(token, Arc::new(message));
+                    }
                 }
             }
             Ok(TendermintMessage::RequestProposal {
@@ -420,10 +447,11 @@ impl NetworkExtension<Event> for TendermintExtension {
             } => {
                 self.request_messages_to_all(vote_step, requested_votes);
             }
-            Event::RequestProposalToAny {
+            Event::RequestProposalToSuperiors {
                 round,
+                current_highest,
             } => {
-                self.request_proposal_to_any(round);
+                self.request_proposal_to_superiors(round, current_highest);
             }
             Event::SetTimerStep {
                 step,
@@ -456,8 +484,9 @@ pub enum Event {
         vote_step: VoteStep,
         requested_votes: BitSet,
     },
-    RequestProposalToAny {
+    RequestProposalToSuperiors {
         round: SortitionRound,
+        current_highest: Option<Priority>,
     },
     SetTimerStep {
         step: Step,

core/src/consensus/tendermint/types.rs

@@ -191,6 +191,10 @@ impl PeerState {
             messages: BitSet::new(),
         }
     }
+
+    pub fn priority(&self) -> Option<Priority> {
+        self.proposal.as_ref().map(|summary| summary.priority())
+    }
 }
 
 pub struct TendermintSealView<'a> {

core/src/consensus/tendermint/vote_collector.rs

@@ -21,7 +21,7 @@ use ckey::SchnorrSignature;
 use ctypes::BlockHash;
 use rlp::{Encodable, RlpStream};
 
-use super::super::PriorityInfo;
+use super::super::{Priority, PriorityInfo};
 use super::stake::Action;
 use super::{ConsensusMessage, ProposalSummary, SortitionRound, Step, VoteStep};
 use crate::consensus::BitSet;
@@ -335,6 +335,10 @@ impl VoteCollector {
             .and_then(|step_collector| step_collector.priority_collector().get_highest())
     }
 
+    pub fn get_highest_priority(&self, sortition_round: SortitionRound) -> Option<Priority> {
+        self.get_highest_priority_info(sortition_round).map(|priority_info| priority_info.priority())
+    }
+
     pub fn get_highest_proposal_hash(&self, sortition_round: SortitionRound) -> Option<BlockHash> {
         self.votes.get(&sortition_round.into()).and_then(|step_collector| {
             let highest_priority_idx =

core/src/consensus/tendermint/worker.rs

@@ -48,7 +48,7 @@ use crate::consensus::signer::EngineSigner;
 use crate::consensus::validator_set::{DynamicValidator, ValidatorSet};
 use crate::consensus::{
     sortition::seed::{SeedInfo, VRFSeed},
-    EngineError, PriorityInfo, Seal, VRFSortition,
+    EngineError, Priority, PriorityInfo, Seal, VRFSortition,
 };
 use crate::encoded;
 use crate::error::{BlockError, Error};
@@ -147,6 +147,12 @@ pub enum Event {
         message: Bytes,
         result: crossbeam::Sender<Option<Arc<dyn ConsensusClient>>>,
     },
+    VerifyPriorityInfo {
+        height: Height,
+        view: View,
+        priority_info: PriorityInfo,
+        result: crossbeam::Sender<Option<bool>>,
+    },
     StepState {
         token: NodeId,
         vote_step: VoteStep,
@@ -362,6 +368,14 @@ impl Worker {
                                 let client = inner.on_commit_message(block, votes);
                                 result.send(client).unwrap();
                             }
+                            Ok(Event::VerifyPriorityInfo {
+                                height,
+                                view,
+                                priority_info,
+                                result
+                            }) => {
+                                result.send(inner.verify_priority_info(height, view, priority_info)).unwrap();
+                            }
                             Err(crossbeam::RecvError) => {
                                 cerror!(ENGINE, "The event channel for tendermint thread had been closed.");
                                 break
@@ -592,10 +606,11 @@ impl Worker {
             .unwrap();
     }
 
-    fn request_proposal_to_any(&self, round: SortitionRound) {
+    fn request_proposal_to_superiors(&self, round: SortitionRound, current_highest: Option<Priority>) {
         self.extension
-            .send(network::Event::RequestProposalToAny {
+            .send(network::Event::RequestProposalToSuperiors {
                 round,
+                current_highest,
             })
             .unwrap();
     }
@@ -707,7 +722,10 @@ impl Worker {
                         };
                     }
                 } else {
-                    self.request_proposal_to_any(vote_step.into());
+                    let sortition_round = vote_step.into();
+                    let round_highest_priority = self.votes.get_highest_priority(sortition_round);
+                    cinfo!(ENGINE, "I am not eligible to be a proposer, I'll request a proposal");
+                    self.request_proposal_to_superiors(sortition_round, round_highest_priority);
                 }
             }
             Step::Prevote => {
@@ -829,7 +847,9 @@ impl Worker {
         let received_locked_block = self.votes.has_votes_for(&vote_step, locked_proposal_hash);
 
         if !received_locked_block {
-            self.request_proposal_to_any(vote_step.into());
+            let sortition_round = vote_step.into();
+            let round_highest_priority = self.votes.get_highest_priority(sortition_round);
+            self.request_proposal_to_superiors(sortition_round, round_highest_priority);
             return Err(format!("Have a lock on {}-{}, but do not received a locked proposal", self.height, locked_view))
         }
 
@@ -1140,6 +1160,18 @@ impl Worker {
         Ok(())
     }
 
+    fn verify_priority_info(&mut self, height: Height, view: View, priority_info: PriorityInfo) -> Option<bool> {
+        let parent_seed = self.prev_vrf_seed_of_height(height)?;
+        let parent_hash = self.client().block_hash(&(height - 1).into())?;
+        let signer_idx = priority_info.signer_idx();
+        let signer_public = self.validators.get(&parent_hash, signer_idx);
+        let voting_power = self.get_voting_power(height - 1, &parent_hash, signer_idx);
+
+        priority_info
+            .verify(&parent_seed.round_msg(view), &signer_public, voting_power, &mut self.sortition_scheme)
+            .ok()
+    }
+
     fn verify_seed(&mut self, header: &Header, parent: &Header) -> Result<(), Error> {
         let current_seal_view = TendermintSealView::new(&header.seal());
         let current_seed_signer_idx = current_seal_view.vrf_seed_info().expect("Seal field verified").signer_idx();