Merge pull request #2474 from dusk-network/open-consensus

consensus: fix OPEN_CONSENSUS mode

consensus/src/config.rs

@@ -66,6 +66,11 @@ pub fn is_emergency_iter(iter: u8) -> bool {
     iter >= EMERGENCY_MODE_ITERATION_THRESHOLD
 }
 
+/// Returns if the next iteration generator needs to be excluded
+pub fn exclude_next_generator(iter: u8) -> bool {
+    iter < CONSENSUS_MAX_ITER - 1
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

consensus/src/execution_ctx.rs

@@ -16,7 +16,7 @@ use crate::user::committee::Committee;
 use crate::user::provisioners::Provisioners;
 
 use node_data::bls::PublicKeyBytes;
-use node_data::ledger::Block;
+use node_data::ledger::{to_str, Block};
 use node_data::message::{AsyncQueue, Message, Payload};
 
 use node_data::StepName;
@@ -136,7 +136,7 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
         // emergency block is accepted
         let timeout = if open_consensus_mode {
             let dur = Duration::new(u32::MAX as u64, 0);
-            info!(event = "run event_loop in open_mode", ?dur);
+            info!(event = "run event_loop", ?dur, mode = "open_consensus",);
             dur
         } else {
             let dur = self.iter_ctx.get_timeout(self.step_name());
@@ -156,15 +156,39 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
                         self.process_inbound_msg(phase.clone(), msg).await
                     {
                         if open_consensus_mode {
+                            info!(
+                                mode = "open_consensus",
+                                event = "message received",
+                                topic = ?step_result.topic()
+                            );
+                            if let Payload::Quorum(q) = &step_result.payload {
+                                let vote = q.att.result.vote();
+                                if let Vote::Valid(hash) = vote {
+                                    info!(
+                                        mode = "open_consensus",
+                                        event = "send quorum",
+                                        hash = to_str(hash)
+                                    );
+                                    self.quorum_sender
+                                        .send_quorum(step_result)
+                                        .await;
+                                } else {
+                                    info!(
+                                        mode = "open_consensus",
+                                        event = "ignoring failed quorum",
+                                        ?vote
+                                    );
+                                }
+                            }
                             // In open consensus mode, consensus step is never
                             // terminated.
                             // The acceptor will cancel the consensus if a
                             // block is accepted
                             continue;
+                        } else {
+                            self.report_elapsed_time().await;
+                            return Ok(step_result);
                         }
-
-                        self.report_elapsed_time().await;
-                        return Ok(step_result);
                     }
                 }
                 Ok(Err(e)) => {

consensus/src/iteration_ctx.rs

@@ -7,7 +7,9 @@
 use crate::commons::{Database, RoundUpdate, TimeoutSet};
 use std::cmp;
 
-use crate::config::{CONSENSUS_MAX_ITER, MAX_STEP_TIMEOUT, TIMEOUT_INCREASE};
+use crate::config::{
+    exclude_next_generator, MAX_STEP_TIMEOUT, TIMEOUT_INCREASE,
+};
 use crate::msg_handler::HandleMsgOutput;
 use crate::msg_handler::MsgHandler;
 
@@ -168,7 +170,7 @@ impl<DB: Database> IterationCtx<DB> {
         // generator for the exclusion list. So we extract, it if necessary.
         //
         // This is not necessary in the last iteration, so we skip it
-        if step_name != StepName::Proposal && iteration < CONSENSUS_MAX_ITER - 1
+        if step_name != StepName::Proposal && exclude_next_generator(iteration)
         {
             let prop = StepName::Proposal;
             let next_prop_step = prop.to_step(iteration + 1);
@@ -202,7 +204,7 @@ impl<DB: Database> IterationCtx<DB> {
                 exclusion_list.push(cur_generator);
 
                 // Exclude generator for next iteration
-                if iteration < CONSENSUS_MAX_ITER - 1 {
+                if exclude_next_generator(iteration) {
                     let next_generator =
                         self.get_generator(iteration + 1).expect(
                             "Next Proposal committee to be already generated",
@@ -228,6 +230,8 @@ impl<DB: Database> IterationCtx<DB> {
 
         debug!(
             event = "committee_generated",
+            step = config_step.step,
+            config = ?config_step,
             members = format!("{}", &step_committee)
         );
 

consensus/src/quorum/verifiers.rs

@@ -17,7 +17,7 @@ use crate::user::cluster::Cluster;
 use crate::user::committee::{Committee, CommitteeSet};
 use crate::user::sortition;
 
-use crate::config::CONSENSUS_MAX_ITER;
+use crate::config::exclude_next_generator;
 use dusk_bytes::Serializable as BytesSerializable;
 use execution_core::signatures::bls::{
     MultisigPublicKey as BlsMultisigPublicKey,
@@ -45,7 +45,7 @@ pub async fn verify_step_votes(
 
     exclusion_list.push(generator);
 
-    if iteration < CONSENSUS_MAX_ITER {
+    if exclude_next_generator(iteration) {
         let next_generator = committees_set
             .read()
             .await
@@ -123,7 +123,8 @@ pub fn verify_votes(
     if !skip_quorum && !quorum_result.quorum_reached() {
         tracing::error!(
             desc = "vote_set_too_small",
-            committee = format!("{:#?}", sub_committee),
+            committee = format!("{committee}"),
+            sub_committee = format!("{:#?}", sub_committee),
             bitset,
             target_quorum,
             total,
@@ -220,7 +221,7 @@ async fn get_step_committee(
     exclusion_list.push(generator);
 
     // exclude next-iteration generator
-    if iteration < CONSENSUS_MAX_ITER {
+    if exclude_next_generator(iteration) {
         let next_generator = committees_set
             .read()
             .await

consensus/src/step_votes_reg.rs

@@ -176,6 +176,9 @@ impl AttInfoRegistry {
         quorum_reached: bool,
         generator: &PublicKeyBytes,
     ) -> Option<Message> {
+        if sv == StepVotes::default() {
+            return None;
+        }
         let att = self
             .att_list
             .entry(iteration)