Implement priority message handling

core/src/consensus/tendermint/network.rs

@@ -277,6 +277,24 @@ impl NetworkExtension<Event> for TendermintExtension {
                     }
                 }
             }
+            Ok(TendermintMessage::PriorityMessage {
+                height,
+                view,
+                signer_idx,
+                message,
+            }) => {
+                let (result, receiver) = crossbeam::bounded(1);
+                self.inner
+                    .send(worker::Event::Priority {
+                        height,
+                        view,
+                        signer_idx,
+                        message,
+                        result,
+                    })
+                    .unwrap();
+                if let Some(c) = receiver.recv().unwrap() {}
+            }
             Ok(TendermintMessage::ProposalBlock {
                 signature,
                 view,

core/src/consensus/tendermint/worker.rs

@@ -21,20 +21,24 @@ use std::sync::{Arc, Weak};
 use std::thread::{Builder, JoinHandle};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use ccrypto::sha256;
 use ckey::{public_to_address, verify_schnorr, Address, SchnorrSignature};
 use cnetwork::{EventSender, NodeId};
 use crossbeam_channel as crossbeam;
 use ctypes::transaction::{Action, Transaction};
 use ctypes::util::unexpected::Mismatch;
 use ctypes::{BlockHash, BlockNumber, Header};
+use parking_lot::RwLock;
 use primitives::{u256_from_u128, Bytes, U256};
 use rlp::{Encodable, Rlp};
+use vrf::openssl::{CipherSuite, ECVRF};
 
 use super::super::BitSet;
 use super::backup::{backup, restore, BackupView};
 use super::message::*;
 use super::network;
 use super::params::TimeGapParams;
+use super::sortition_info_collector::SortitionInfoCollector;
 use super::stake::CUSTOM_ACTION_HANDLER_ID;
 use super::types::{Height, Proposal, Step, TendermintSealView, TendermintState, TwoThirdsMajority, View};
 use super::vote_collector::{DoubleVote, VoteCollector};
@@ -47,7 +51,7 @@ use crate::block::*;
 use crate::client::ConsensusClient;
 use crate::consensus::signer::EngineSigner;
 use crate::consensus::validator_set::{DynamicValidator, ValidatorSet};
-use crate::consensus::{EngineError, Seal};
+use crate::consensus::{EngineError, Seal, VRFSortition};
 use crate::encoded;
 use crate::error::{BlockError, Error};
 use crate::transaction::{SignedTransaction, UnverifiedTransaction};
@@ -94,6 +98,9 @@ struct Worker {
     validators: Arc<DynamicValidator>,
     /// Channel to the network extension, must be set later.
     extension: EventSender<network::Event>,
+    /// VRF instance to calculate VRF hash.
+    vrf_inst: Arc<RwLock<ECVRF>>,
+    sortition_infos: SortitionInfoCollector,
     time_gap_params: TimeGapParams,
     timeout_token_nonce: usize,
     vote_regression_checker: VoteRegressionChecker,
@@ -137,6 +144,13 @@ pub enum Event {
         address: Address,
     },
     Restore(crossbeam::Sender<()>),
+    Priority {
+        height: Height,
+        view: View,
+        signer_idx: usize,
+        message: PriorityMessage,
+        result: crossbeam::Sender<Option<Arc<dyn ConsensusClient>>>,
+    },
     ProposalBlock {
         signature: SchnorrSignature,
         view: View,
@@ -194,10 +208,12 @@ impl Worker {
             finalized_view_of_current_block: None,
             validators,
             extension,
+            sortition_infos: Default::default(),
             votes_received: MutTrigger::new(BitSet::new()),
             time_gap_params,
             timeout_token_nonce: ENGINE_TIMEOUT_TOKEN_NONCE_BASE,
             vote_regression_checker: VoteRegressionChecker::new(),
+            vrf_inst: Arc::new(RwLock::new(ECVRF::from_suite(CipherSuite::SECP256K1_SHA256_SVDW).unwrap())),
         }
     }
 
@@ -312,6 +328,16 @@ impl Worker {
                                 inner.restore();
                                 result.send(()).unwrap();
                             }
+                            Ok(Event::Priority {
+                                height,
+                                view,
+                                signer_idx,
+                                message,
+                                result,
+                            }) => {
+                                let client = inner.on_priority_message(height, view, signer_idx, message);
+                                result.send(client).unwrap();
+                            }
                             Ok(Event::ProposalBlock {
                                 signature,
                                 view,
@@ -1728,6 +1754,35 @@ impl Worker {
         result.send(message.rlp_bytes().into_vec()).unwrap();
     }
 
+    fn on_priority_message(
+        &mut self,
+        height: Height,
+        view: View,
+        signer_idx: usize,
+        message: PriorityMessage,
+    ) -> Option<Arc<dyn ConsensusClient>> {
+        let c = self.client.upgrade()?;
+        c.block_hash(&BlockId::Number(height - 1)).and_then(|parent_hash| {
+            let signer_public = self.validators.get(&parent_hash, signer_idx);
+            //FIXME: Do not recalculate seed values with heavy VRF operations.
+            //FIXME: store block_seed into each block.
+            let concatenated = [height.to_be_bytes(), view.to_be_bytes()].concat();
+            let seed = self.signer.vrf_hash(sha256(concatenated), &mut self.vrf_inst.write());
+            let voting_power = 50;
+            //FIXME: store sortition scheme.
+            let sortition_scheme = VRFSortition {
+                total_power: 100,
+                expectation: 7.0,
+                vrf_inst: Arc::clone(&self.vrf_inst),
+            };
+            if let Ok(true) = message.verify(&signer_public, voting_power, &sortition_scheme) {
+                Some(c)
+            } else {
+                None
+            }
+        })
+    }
+
     fn on_proposal_message(
         &mut self,
         signature: SchnorrSignature,