consensus: Deterministic timeout

consensus/src/aggregator.rs

@@ -242,7 +242,7 @@ mod tests {
     };
     use hex::FromHex;
     use node_data::ledger::{Header, Seed};
-    use std::collections::HashMap;
+    use std::time::Duration;
 
     impl<V> Aggregator<V> {
         pub fn get_total(&self, step: u8, vote: Vote) -> Option<usize> {
@@ -301,7 +301,7 @@ mod tests {
                 pubkey_bls,
                 secret_key,
                 &tip_header,
-                HashMap::new(),
+                Duration::default(),
                 vec![],
             );
 

consensus/src/commons.rs

@@ -8,19 +8,15 @@
 // Provisioners, the BidList, the Seed and the Hash.
 
 use node_data::ledger::*;
-use std::collections::HashMap;
 
 use std::time::Duration;
 
 use execution_core::signatures::bls::SecretKey as BlsSecretKey;
 use node_data::bls::PublicKey;
 use node_data::message::{AsyncQueue, Message, Payload};
-use node_data::StepName;
 
 use crate::operations::Voter;
 
-pub type TimeoutSet = HashMap<StepName, Duration>;
-
 #[derive(Clone, Default, Debug)]
 pub struct RoundUpdate {
     // Current round number of the ongoing consensus
@@ -36,15 +32,15 @@ pub struct RoundUpdate {
     att_voters: Vec<Voter>,
     timestamp: u64,
 
-    pub base_timeouts: TimeoutSet,
+    pub base_timeout: Duration,
 }
 
 impl RoundUpdate {
     pub fn new(
         pubkey_bls: PublicKey,
         secret_key: BlsSecretKey,
         tip_header: &Header,
-        base_timeouts: TimeoutSet,
+        base_timeout: Duration,
         att_voters: Vec<Voter>,
     ) -> Self {
         let round = tip_header.height + 1;
@@ -56,7 +52,7 @@ impl RoundUpdate {
             hash: tip_header.hash,
             seed: tip_header.seed,
             timestamp: tip_header.timestamp,
-            base_timeouts,
+            base_timeout,
             att_voters,
         }
     }

consensus/src/consensus.rs

@@ -182,7 +182,7 @@ impl<T: Operations + 'static, D: Database + 'static> Consensus<T, D> {
                 validation_handler,
                 ratification_handler,
                 proposal_handler,
-                ru.base_timeouts.clone(),
+                ru.base_timeout,
             );
 
             let (prev_block_hash, saved_iter) =

consensus/src/errors.rs

@@ -76,8 +76,6 @@ pub enum OperationError {
     InvalidEST(anyhow::Error),
     #[error("failed to verify header {0}")]
     InvalidHeader(HeaderError),
-    #[error("Unable to update metrics {0}")]
-    MetricsUpdate(anyhow::Error),
     #[error("Invalid Iteration Info {0}")]
     InvalidIterationInfo(io::Error),
     #[error("Invalid Faults {0}")]

consensus/src/execution_ctx.rs

@@ -139,7 +139,7 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
             info!(event = "run event_loop", ?dur, mode = "open_consensus",);
             dur
         } else {
-            let dur = self.iter_ctx.get_timeout(self.step_name());
+            let dur = self.iter_ctx.get_timeout();
             debug!(event = "run event_loop", ?dur);
             dur
         };
@@ -186,7 +186,6 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
                             // block is accepted
                             continue;
                         } else {
-                            self.report_elapsed_time().await;
                             return Ok(step_result);
                         }
                     }
@@ -470,8 +469,6 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
         &mut self,
         phase: Arc<Mutex<C>>,
     ) {
-        self.iter_ctx.on_timeout_event(self.step_name());
-
         if let Some(msg) = phase
             .lock()
             .await
@@ -536,24 +533,6 @@ impl<'a, T: Operations + 'static, DB: Database> ExecutionCtx<'a, T, DB> {
         None
     }
 
-    /// Reports step elapsed time to the client
-    async fn report_elapsed_time(&mut self) {
-        let elapsed = self
-            .step_start_time
-            .take()
-            .expect("valid start time")
-            .elapsed();
-
-        let _ = self
-            .client
-            .add_step_elapsed_time(
-                self.round_update.round,
-                self.step_name(),
-                elapsed,
-            )
-            .await;
-    }
-
     pub(crate) fn get_curr_generator(&self) -> Option<PublicKeyBytes> {
         self.iter_ctx.get_generator(self.iteration)
     }

consensus/src/iteration_ctx.rs

@@ -4,11 +4,11 @@
 //
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
-use crate::commons::{Database, RoundUpdate, TimeoutSet};
-use std::cmp;
+use crate::commons::{Database, RoundUpdate};
 
 use crate::config::{
-    exclude_next_generator, MAX_STEP_TIMEOUT, TIMEOUT_INCREASE,
+    exclude_next_generator, MAX_STEP_TIMEOUT, MIN_STEP_TIMEOUT,
+    TIMEOUT_INCREASE,
 };
 use crate::msg_handler::HandleMsgOutput;
 use crate::msg_handler::MsgHandler;
@@ -23,7 +23,6 @@ use node_data::bls::PublicKeyBytes;
 use node_data::ledger::Seed;
 use node_data::message::Message;
 use std::collections::HashMap;
-use std::ops::Add;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Mutex;
@@ -79,8 +78,8 @@ pub struct IterationCtx<DB: Database> {
     /// iteration of current round
     pub(crate) committees: RoundCommittees,
 
-    /// Implements the adaptive timeout algorithm
-    timeouts: TimeoutSet,
+    /// Timeout for each step of the current iteration
+    step_timeout: Duration,
 }
 
 impl<DB: Database> IterationCtx<DB> {
@@ -92,7 +91,7 @@ impl<DB: Database> IterationCtx<DB> {
             Mutex<ratification::handler::RatificationHandler>,
         >,
         proposal_handler: Arc<Mutex<proposal::handler::ProposalHandler<DB>>>,
-        timeouts: TimeoutSet,
+        base_step_timeout: Duration,
     ) -> Self {
         Self {
             round,
@@ -101,14 +100,23 @@ impl<DB: Database> IterationCtx<DB> {
             validation_handler,
             ratification_handler,
             committees: Default::default(),
-            timeouts,
+            step_timeout: base_step_timeout,
             proposal_handler,
         }
     }
 
     /// Executed on starting a new iteration, before Proposal step execution
     pub(crate) fn on_begin(&mut self, iter: u8) {
         self.iter = iter;
+
+        if iter > 0 {
+            self.step_timeout += TIMEOUT_INCREASE;
+        }
+
+        self.step_timeout = self
+            .step_timeout
+            .max(MIN_STEP_TIMEOUT)
+            .min(MAX_STEP_TIMEOUT);
     }
 
     /// Executed on closing an iteration, after Ratification step execution
@@ -122,21 +130,9 @@ impl<DB: Database> IterationCtx<DB> {
         self.join_set.abort_all();
     }
 
-    /// Handles an event of a Phase timeout
-    pub(crate) fn on_timeout_event(&mut self, step_name: StepName) {
-        let curr_step_timeout =
-            self.timeouts.get_mut(&step_name).expect("valid timeout");
-
-        *curr_step_timeout =
-            cmp::min(MAX_STEP_TIMEOUT, curr_step_timeout.add(TIMEOUT_INCREASE));
-    }
-
-    /// Calculates and returns the adjusted timeout for the specified step
-    pub(crate) fn get_timeout(&self, step_name: StepName) -> Duration {
-        *self
-            .timeouts
-            .get(&step_name)
-            .expect("valid timeout per step")
+    /// Return the step timeout for the current iteration
+    pub(crate) fn get_timeout(&self) -> Duration {
+        self.step_timeout
     }
 
     fn get_sortition_config(

consensus/src/operations.rs

@@ -5,14 +5,12 @@
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
 use std::fmt;
-use std::time::Duration;
 
 use node_data::bls::PublicKey;
 use node_data::bls::PublicKeyBytes;
 use node_data::ledger::{
     Block, Fault, Header, Slash, SpentTransaction, Transaction,
 };
-use node_data::StepName;
 
 use crate::errors::*;
 
@@ -87,12 +85,5 @@ pub trait Operations: Send + Sync {
         params: CallParams,
     ) -> Result<Output, OperationError>;
 
-    async fn add_step_elapsed_time(
-        &self,
-        round: u64,
-        step_name: StepName,
-        elapsed: Duration,
-    ) -> Result<(), OperationError>;
-
     async fn get_block_gas_limit(&self) -> u64;
 }

consensus/src/phase.rs

@@ -57,7 +57,7 @@ impl<T: Operations + 'static, D: Database + 'static> Phase<T, D> {
     ) -> Result<Message, ConsensusError> {
         ctx.set_start_time();
 
-        let timeout = ctx.iter_ctx.get_timeout(ctx.step_name());
+        let timeout = ctx.iter_ctx.get_timeout();
         debug!(event = "execute_step", ?timeout);
 
         // Execute step

node/benches/accept.rs

@@ -68,7 +68,7 @@ fn create_step_votes(
                 pk.clone(),
                 sk.clone(),
                 tip_header,
-                HashMap::default(),
+                Duration::default(),
                 vec![],
             );
             let sig = match step {