node: Reward Voters/Validators of previous block

consensus/src/aggregator.rs

@@ -249,6 +249,7 @@ mod tests {
                 secret_key,
                 &tip_header,
                 HashMap::new(),
+                vec![],
             );
 
             let msg = crate::build_validation_payload(

consensus/src/commons.rs

@@ -14,6 +14,7 @@ use std::collections::HashMap;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use thiserror::Error;
 
+use crate::operations::VoterWithCredits;
 use execution_core::{BlsSigError, StakeSecretKey};
 use node_data::bls::PublicKey;
 use node_data::message::{AsyncQueue, Message, Payload};
@@ -34,6 +35,7 @@ pub struct RoundUpdate {
     seed: Seed,
     hash: [u8; 32],
     att: Attestation,
+    att_voters: Vec<VoterWithCredits>,
     timestamp: u64,
 
     pub base_timeouts: TimeoutSet,
@@ -45,6 +47,7 @@ impl RoundUpdate {
         secret_key: StakeSecretKey,
         tip_header: &Header,
         base_timeouts: TimeoutSet,
+        att_voters: Vec<VoterWithCredits>,
     ) -> Self {
         let round = tip_header.height + 1;
         RoundUpdate {
@@ -56,6 +59,7 @@ impl RoundUpdate {
             seed: tip_header.seed,
             timestamp: tip_header.timestamp,
             base_timeouts,
+            att_voters,
         }
     }
 
@@ -74,6 +78,10 @@ impl RoundUpdate {
     pub fn timestamp(&self) -> u64 {
         self.timestamp
     }
+
+    pub fn att_voters(&self) -> &Vec<VoterWithCredits> {
+        &self.att_voters
+    }
 }
 
 #[derive(Debug, Clone, Copy, Error)]

consensus/src/consensus.rs

@@ -38,7 +38,7 @@ pub struct Consensus<T: Operations, D: Database> {
     future_msgs: Arc<Mutex<MsgRegistry<Message>>>,
 
     /// Reference to the executor of any EST-related call
-    executor: Arc<Mutex<T>>,
+    executor: Arc<T>,
 
     // Database
     db: Arc<Mutex<D>>,
@@ -58,7 +58,7 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
         inbound: AsyncQueue<Message>,
         outbound: AsyncQueue<Message>,
         future_msgs: Arc<Mutex<MsgRegistry<Message>>>,
-        executor: Arc<Mutex<T>>,
+        executor: Arc<T>,
         db: Arc<Mutex<D>>,
     ) -> Self {
         Self {
@@ -169,7 +169,6 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
                     validation_handler.clone(),
                 )),
                 Phase::Ratification(ratification::step::RatificationStep::new(
-                    executor.clone(),
                     ratification_handler.clone(),
                 )),
             ];

consensus/src/execution_ctx.rs

@@ -50,7 +50,7 @@ pub struct ExecutionCtx<'a, DB: Database, T> {
     step: StepName,
     step_start_time: Option<Instant>,
 
-    pub client: Arc<Mutex<T>>,
+    pub client: Arc<T>,
 
     pub sv_registry: SafeAttestationInfoRegistry,
     quorum_sender: QuorumMsgSender,
@@ -68,7 +68,7 @@ impl<'a, DB: Database, T: Operations + 'static> ExecutionCtx<'a, DB, T> {
         round_update: RoundUpdate,
         iteration: u8,
         step: StepName,
-        client: Arc<Mutex<T>>,
+        client: Arc<T>,
         sv_registry: SafeAttestationInfoRegistry,
         quorum_sender: QuorumMsgSender,
     ) -> Self {
@@ -191,7 +191,7 @@ impl<'a, DB: Database, T: Operations + 'static> ExecutionCtx<'a, DB, T> {
 
         if let Some(committee) = self.iter_ctx.committees.get_committee(step) {
             if self.am_member(committee) {
-                RatificationStep::<T, DB>::try_vote(
+                RatificationStep::<DB>::try_vote(
                     &self.round_update,
                     msg_iteration,
                     validation,
@@ -456,8 +456,6 @@ impl<'a, DB: Database, T: Operations + 'static> ExecutionCtx<'a, DB, T> {
 
         let _ = self
             .client
-            .lock()
-            .await
             .add_step_elapsed_time(
                 self.round_update.round,
                 self.step_name(),

consensus/src/operations.rs

@@ -13,6 +13,7 @@ use node_data::StepName;
 
 pub type StateRoot = [u8; 32];
 pub type EventHash = [u8; 32];
+pub type VoterWithCredits = (StakePublicKey, usize);
 
 #[derive(Debug)]
 pub enum Error {
@@ -26,6 +27,7 @@ pub struct CallParams {
     pub block_gas_limit: u64,
     pub generator_pubkey: node_data::bls::PublicKey,
     pub missed_generators: Vec<StakePublicKey>,
+    pub voters_pubkey: Option<Vec<VoterWithCredits>>,
 }
 
 #[derive(Default)]
@@ -58,11 +60,12 @@ pub trait Operations: Send + Sync {
         &self,
         candidate_header: &Header,
         disable_winning_att_check: bool,
-    ) -> Result<(), Error>;
+    ) -> Result<(u8, Vec<VoterWithCredits>, Vec<VoterWithCredits>), Error>;
 
     async fn verify_state_transition(
         &self,
         blk: &Block,
+        voters: &[VoterWithCredits],
     ) -> Result<VerificationOutput, Error>;
 
     async fn execute_state_transition(

consensus/src/phase.rs

@@ -28,7 +28,7 @@ macro_rules! await_phase {
 pub enum Phase<T: Operations, D: Database> {
     Proposal(proposal::step::ProposalStep<T, D>),
     Validation(validation::step::ValidationStep<T>),
-    Ratification(ratification::step::RatificationStep<T, D>),
+    Ratification(ratification::step::RatificationStep<D>),
 }
 
 impl<T: Operations + 'static, D: Database + 'static> Phase<T, D> {

consensus/src/proposal/block_generator.rs

@@ -5,7 +5,7 @@
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
 use crate::commons::{get_current_timestamp, RoundUpdate};
-use crate::operations::{CallParams, Operations};
+use crate::operations::{CallParams, Operations, VoterWithCredits};
 use node_data::ledger::{to_str, Attestation, Block, IterationsInfo, Seed};
 use std::cmp::max;
 
@@ -19,15 +19,14 @@ use node_data::message::payload::Candidate;
 use node_data::message::{ConsensusHeader, Message, SignInfo, StepMessage};
 use std::sync::Arc;
 use std::time::Instant;
-use tokio::sync::Mutex;
 use tracing::{debug, info};
 
 pub struct Generator<T: Operations> {
-    executor: Arc<Mutex<T>>,
+    executor: Arc<T>,
 }
 
 impl<T: Operations> Generator<T> {
-    pub fn new(executor: Arc<Mutex<T>>) -> Self {
+    pub fn new(executor: Arc<T>) -> Self {
         Self { executor }
     }
 
@@ -46,7 +45,13 @@ impl<T: Operations> Generator<T> {
         let start = Instant::now();
 
         let candidate = self
-            .generate_block(ru, Seed::from(seed), iteration, failed_iterations)
+            .generate_block(
+                ru,
+                Seed::from(seed),
+                iteration,
+                failed_iterations,
+                ru.att_voters(),
+            )
             .await?;
 
         info!(
@@ -81,6 +86,7 @@ impl<T: Operations> Generator<T> {
         seed: Seed,
         iteration: u8,
         failed_iterations: IterationsInfo,
+        voters: &[VoterWithCredits],
     ) -> Result<Block, crate::operations::Error> {
         let missed_generators = failed_iterations
             .to_missed_generators()
@@ -91,14 +97,11 @@ impl<T: Operations> Generator<T> {
             block_gas_limit: config::DEFAULT_BLOCK_GAS_LIMIT,
             generator_pubkey: ru.pubkey_bls.clone(),
             missed_generators,
+            voters_pubkey: Some(voters.to_owned()),
         };
 
-        let result = self
-            .executor
-            .lock()
-            .await
-            .execute_state_transition(call_params)
-            .await?;
+        let result =
+            self.executor.execute_state_transition(call_params).await?;
 
         let tx_hashes: Vec<_> =
             result.txs.iter().map(|t| t.inner.hash()).collect();

consensus/src/proposal/step.rs

@@ -31,7 +31,7 @@ where
 
 impl<T: Operations + 'static, D: Database> ProposalStep<T, D> {
     pub fn new(
-        executor: Arc<Mutex<T>>,
+        executor: Arc<T>,
         _db: Arc<Mutex<D>>,
         handler: Arc<Mutex<handler::ProposalHandler<D>>>,
     ) -> Self {

consensus/src/quorum/verifiers.rs

@@ -74,7 +74,7 @@ pub async fn verify_step_votes(
     committees_set: &RwLock<CommitteeSet<'_>>,
     seed: Seed,
     step: StepName,
-) -> Result<QuorumResult, StepSigError> {
+) -> Result<(QuorumResult, Committee), StepSigError> {
     let round = header.round;
     let iteration = header.iteration;
 
@@ -94,7 +94,8 @@ pub async fn verify_step_votes(
     let set = committees_set.read().await;
     let committee = set.get(&cfg).expect("committee to be created");
 
-    verify_votes(header, step, vote, sv, committee)
+    let quorum_result = verify_votes(header, step, vote, sv, committee)?;
+    Ok((quorum_result, committee.clone()))
 }
 
 #[derive(Default)]

consensus/src/ratification/step.rs

@@ -21,15 +21,12 @@ use tokio::sync::Mutex;
 
 use tracing::{error, info, Instrument};
 
-pub struct RatificationStep<T, DB> {
+pub struct RatificationStep<DB> {
     handler: Arc<Mutex<handler::RatificationHandler>>,
-
-    _executor: Arc<Mutex<T>>,
-
     marker: PhantomData<DB>,
 }
 
-impl<T: Operations + 'static, DB: Database> RatificationStep<T, DB> {
+impl<DB: Database> RatificationStep<DB> {
     pub async fn try_vote(
         ru: &RoundUpdate,
         iteration: u8,
@@ -77,14 +74,12 @@ pub fn build_ratification_payload(
     ratification
 }
 
-impl<T: Operations + 'static, DB: Database> RatificationStep<T, DB> {
+impl<DB: Database> RatificationStep<DB> {
     pub(crate) fn new(
-        executor: Arc<Mutex<T>>,
         handler: Arc<Mutex<handler::RatificationHandler>>,
     ) -> Self {
         Self {
             handler,
-            _executor: executor,
             marker: PhantomData,
         }
     }
@@ -119,7 +114,7 @@ impl<T: Operations + 'static, DB: Database> RatificationStep<T, DB> {
         )
     }
 
-    pub async fn run(
+    pub async fn run<T: Operations + 'static>(
         &mut self,
         mut ctx: ExecutionCtx<'_, DB, T>,
     ) -> Result<Message, ConsensusError> {

consensus/src/user/committee.rs

@@ -73,6 +73,10 @@ impl Committee {
         self.members.get(pubkey_bls).copied()
     }
 
+    pub fn members(&self) -> &BTreeMap<PublicKey, usize> {
+        &self.members
+    }
+
     // get_occurrences returns values in a vec
     pub fn get_occurrences(&self) -> Vec<usize> {
         self.members.values().copied().collect()
@@ -145,6 +149,7 @@ impl fmt::Display for &Committee {
 }
 
 /// Implements a cache of generated committees so that they can be reused.
+#[derive(Clone)]
 pub struct CommitteeSet<'p> {
     committees: HashMap<sortition::Config, Committee>,
     provisioners: &'p Provisioners,