Accumulate candidates and votes from previous iterations, same round

consensus/example/consensus_service.rs

@@ -174,7 +174,7 @@ pub fn fetch_blskeys_from_file(
 
     // attempt to load and decode wallet
     let ciphertext =
-        fs::read(&path).expect("path should be valid consensus keys file");
+        fs::read(path).expect("path should be valid consensus keys file");
 
     // Decrypt
     let iv = &ciphertext[..16];

consensus/src/aggregator.rs

@@ -18,7 +18,9 @@ use tracing::{debug, error, warn};
 /// voters.StepVotes Mapping of a block hash to both an aggregated signatures
 /// and a cluster of bls voters.
 #[derive(Default)]
-pub struct Aggregator(BTreeMap<Hash, (AggrSignature, Cluster<PublicKey>)>);
+pub struct Aggregator(
+    BTreeMap<(u8, Hash), (AggrSignature, Cluster<PublicKey>)>,
+);
 
 impl Aggregator {
     pub fn collect_vote(
@@ -27,6 +29,7 @@ impl Aggregator {
         header: &Header,
         signature: &[u8; 48],
     ) -> Option<(Hash, StepVotes)> {
+        let msg_step = header.step;
         // Get weight for this pubkey bls. If votes_for returns None, it means
         // the key is not a committee member, respectively we should not
         // process a vote from it.
@@ -35,7 +38,7 @@ impl Aggregator {
 
             let (aggr_sign, cluster) = self
                 .0
-                .entry(hash)
+                .entry((msg_step, hash))
                 .or_insert((AggrSignature::default(), Cluster::new()));
 
             // Each committee has 64 slots. If a Provisioner is extracted into
@@ -173,8 +176,8 @@ mod tests {
     use rand::rngs::StdRng;
     use rand::SeedableRng;
     impl Aggregator {
-        pub fn get_total(&self, hash: Hash) -> Option<usize> {
-            if let Some(value) = self.0.get(&hash) {
+        pub fn get_total(&self, step: u8, hash: Hash) -> Option<usize> {
+            if let Some(value) = self.0.get(&(step, hash)) {
                 return Some(value.1.total_occurrences());
             }
             None
@@ -264,12 +267,15 @@ mod tests {
             // Check collected votes
             assert!(a.collect_vote(&c, h, signature).is_none());
             collected_votes += expected_votes[i];
-            assert_eq!(a.get_total(block_hash), Some(collected_votes));
+            assert_eq!(a.get_total(h.step, block_hash), Some(collected_votes));
 
             // Ensure a duplicated vote is discarded
             if i == 0 {
                 assert!(a.collect_vote(&c, h, signature).is_none());
-                assert_eq!(a.get_total(block_hash), Some(collected_votes));
+                assert_eq!(
+                    a.get_total(h.step, block_hash),
+                    Some(collected_votes)
+                );
             }
         }
     }

consensus/src/agreement/accumulator.rs

@@ -4,6 +4,8 @@
 //
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
+#![allow(dead_code)]
+
 use crate::agreement::verifiers;
 use crate::user::committee::CommitteeSet;
 use crate::user::sortition;
@@ -136,6 +138,7 @@ impl Accumulator {
     /// # Panics
     ///
     /// If workers pool is not spawned, this will panic.
+
     pub async fn process(&mut self, msg: Message) {
         assert!(!self.workers.is_empty());
 

consensus/src/agreement/step.rs

@@ -194,15 +194,19 @@ impl<D: Database> Executor<D> {
             }
             Payload::Agreement(_) => {
                 // Accumulate the agreement
-                self.collect_agreement(acc, msg).await;
+                return self.collect_agreement(acc, msg).await;
             }
             _ => {}
         };
 
         None
     }
 
-    async fn collect_agreement(&mut self, acc: &mut Accumulator, msg: Message) {
+    async fn collect_agreement(
+        &mut self,
+        _acc: &mut Accumulator,
+        msg: Message,
+    ) -> Option<Block> {
         // Publish the agreement
         self.outbound_queue
             .send(msg.clone())
@@ -211,8 +215,18 @@ impl<D: Database> Executor<D> {
                 error!("unable to publish a collected agreement msg {:?}", err)
             });
 
-        // Accumulate the agreement
-        acc.process(msg.clone()).await;
+        debug!("compile block with certificate");
+
+        match msg.payload {
+            Payload::Agreement(payload) => {
+                let (cert, hash) =
+                    (payload.generate_certificate(), msg.header.block_hash);
+
+                // Create winning block
+                self.create_winning_block(&hash, &cert).await
+            }
+            _ => None,
+        }
     }
 
     /// Collects accumulator output (a list of agreements) and publishes

consensus/src/commons.rs

@@ -11,6 +11,7 @@ use crate::contract_state::Operations;
 
 use node_data::ledger::*;
 use node_data::message;
+use node_data::message::Topics;
 use tracing::Instrument;
 
 use crate::contract_state::CallParams;
@@ -80,6 +81,7 @@ pub fn spawn_send_reduction<T: Operations + 'static>(
     outbound: AsyncQueue<Message>,
     inbound: AsyncQueue<Message>,
     executor: Arc<Mutex<T>>,
+    topic: Topics,
 ) {
     let hash = to_str(&candidate.header().hash);
 
@@ -168,7 +170,7 @@ pub fn spawn_send_reduction<T: Operations + 'static>(
                 round: ru.round,
                 step,
                 block_hash: hash,
-                topic: message::Topics::Reduction as u8,
+                topic: topic.into(),
             };
 
             let signature = hdr.sign(&ru.secret_key, ru.pubkey_bls.inner());

consensus/src/config.rs

@@ -6,6 +6,9 @@
 
 /// Maximum number of steps Consensus runs per a single round.
 pub const CONSENSUS_MAX_STEP: u8 = 213;
+/// Maximum number of iterations Consensus runs per a single round.
+pub const CONSENSUS_MAX_ITER: u8 = CONSENSUS_MAX_STEP / 3;
+
 /// Percentage number that determines a quorum.
 pub const CONSENSUS_QUORUM_THRESHOLD: f64 = 0.67;
 /// Initial step timeout in milliseconds.

consensus/src/consensus.rs

@@ -7,8 +7,10 @@
 use crate::commons::{ConsensusError, Database, RoundUpdate};
 use crate::contract_state::Operations;
 use crate::phase::Phase;
+
 use node_data::ledger::{to_str, Block};
-use node_data::message::{AsyncQueue, Message, Payload};
+
+use node_data::message::{AsyncQueue, Message, Payload, Topics};
 
 use crate::agreement::step;
 use crate::execution_ctx::{ExecutionCtx, IterationCtx};
@@ -18,6 +20,7 @@ use crate::{config, selection};
 use crate::{firststep, secondstep};
 use tracing::{error, Instrument};
 
+use crate::round_ctx::RoundCtx;
 use std::sync::Arc;
 use tokio::sync::{oneshot, Mutex};
 use tokio::task::JoinHandle;
@@ -164,34 +167,68 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
                 future_msgs.lock().await.clear_round(ru.round - 1);
             }
 
+            let round_ctx = Arc::new(Mutex::new(RoundCtx::new(ru.clone())));
+
+            let sel_handler =
+                Arc::new(Mutex::new(selection::handler::Selection::new(
+                    db.clone(),
+                    round_ctx.clone(),
+                )));
+
+            let first_handler =
+                Arc::new(Mutex::new(firststep::handler::Reduction::new(
+                    db.clone(),
+                    round_ctx.clone(),
+                )));
+
+            let sec_handler = Arc::new(Mutex::new(
+                secondstep::handler::Reduction::new(round_ctx.clone()),
+            ));
+
             let mut phases = [
                 Phase::Selection(selection::step::Selection::new(
                     executor.clone(),
                     db.clone(),
+                    sel_handler.clone(),
                 )),
                 Phase::Reduction1(firststep::step::Reduction::new(
                     executor.clone(),
                     db.clone(),
+                    first_handler.clone(),
+                )),
+                Phase::Reduction2(secondstep::step::Reduction::new(
+                    executor.clone(),
+                    sec_handler.clone(),
                 )),
-                Phase::Reduction2(secondstep::step::Reduction::new(executor)),
             ];
 
             // Consensus loop
             // Initialize and run consensus loop
-            let mut step: u8 = 0;
+
+            let mut iter: u8 = 0;
 
             loop {
+                iter += 1;
+
                 let mut msg = Message::empty();
-                let mut iter_ctx = IterationCtx::new(ru.round, step + 1);
+                let mut iter_ctx = IterationCtx::new(
+                    ru.round,
+                    iter,
+                    sel_handler.clone(),
+                    first_handler.clone(),
+                    sec_handler.clone(),
+                );
 
                 // Execute a single iteration
-                for phase in phases.iter_mut() {
-                    step += 1;
+                for pos in 0..phases.len() {
+                    let phase = phases.get_mut(pos).unwrap();
+
+                    let step = (iter - 1) * 3 + (pos as u8 + 1);
                     let name = phase.name();
 
                     // Initialize new phase with message returned by previous
                     // phase.
-                    phase.reinitialize(&msg, ru.round, step);
+                    phase.reinitialize(&msg, ru.round, step).await;
 
                     // Construct phase execution context
                     let ctx = ExecutionCtx::new(
@@ -202,6 +239,7 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
                         &mut provisioners,
                         ru.clone(),
                         step,
+                        executor.clone(),
                     );
 
                     // Execute a phase.
@@ -220,6 +258,12 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
                         ))
                         .await?;
 
+                    // During execution of any step we may encounter that an
+                    // agreement is generated for previous iteration.
+                    if msg.topic() == Topics::Agreement {
+                        break;
+                    }
+
                     if step >= config::CONSENSUS_MAX_STEP {
                         return Err(ConsensusError::MaxStepReached);
                     }