feat(consensus): optionally configure max block time

applications/tari_dan_app_utilities/src/consensus_constants.rs

@@ -26,6 +26,7 @@ pub struct ConsensusConstants {
     pub committee_size: u32,
     pub max_base_layer_blocks_ahead: u64,
     pub max_base_layer_blocks_behind: u64,
+    pub pacemaker_max_base_time: std::time::Duration,
 }
 
 impl ConsensusConstants {
@@ -35,6 +36,7 @@ impl ConsensusConstants {
             committee_size: 7,
             max_base_layer_blocks_ahead: 5,
             max_base_layer_blocks_behind: 5,
+            pacemaker_max_base_time: std::time::Duration::from_secs(10),
         }
     }
 }

applications/tari_validator_node/src/consensus/mod.rs

@@ -99,6 +99,7 @@ pub async fn spawn(
         HotstuffConfig {
             max_base_layer_blocks_behind: consensus_constants.max_base_layer_blocks_behind,
             max_base_layer_blocks_ahead: consensus_constants.max_base_layer_blocks_ahead,
+            pacemaker_max_base_time: consensus_constants.pacemaker_max_base_time,
         },
     );
     let current_view = hotstuff_worker.pacemaker().current_view().clone();

applications/tari_validator_node/src/p2p/rpc/service_impl.rs

@@ -350,8 +350,8 @@ impl ValidatorNodeRpcService for ValidatorNodeRpcServiceImpl {
             .get_local_committee_info(prev_epoch)
             .await
             .optional()
-            .map_err(RpcStatus::log_internal_error(LOG_TARGET))? else
-        {
+            .map_err(RpcStatus::log_internal_error(LOG_TARGET))?
+        else {
             return Err(RpcStatus::bad_request(format!(
                 "This validator node is not registered for the previous epoch {prev_epoch}"
             )));

dan_layer/consensus/src/hotstuff/config.rs

@@ -5,4 +5,5 @@
 pub struct HotstuffConfig {
     pub max_base_layer_blocks_ahead: u64,
     pub max_base_layer_blocks_behind: u64,
+    pub pacemaker_max_base_time: std::time::Duration,
 }

dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs

@@ -564,12 +564,11 @@ impl<TConsensusSpec: ConsensusSpec> OnReceiveLocalProposalHandler<TConsensusSpec
 
             let Some(last_dummy) = dummy_blocks.last() else {
                 warn!(target: LOG_TARGET, "❌ Bad proposal, does not justify parent for candidate block {}", candidate_block);
-                return Err(
-                    ProposalValidationError::CandidateBlockDoesNotExtendJustify {
-                        justify_block_height: justify_block.height(),
-                        candidate_block_height: candidate_block.height(),
-                    }.into()
-                );
+                return Err(ProposalValidationError::CandidateBlockDoesNotExtendJustify {
+                    justify_block_height: justify_block.height(),
+                    candidate_block_height: candidate_block.height(),
+                }
+                .into());
             };
 
             if candidate_block.parent() != last_dummy.id() {

dan_layer/consensus/src/hotstuff/pacemaker.rs

@@ -1,5 +1,6 @@
 //  Copyright 2022 The Tari Project
 //  SPDX-License-Identifier: BSD-3-Clause
+
 use std::{
     cmp,
     time::{Duration, Instant},
@@ -20,17 +21,17 @@ use crate::hotstuff::{
 
 const LOG_TARGET: &str = "tari::dan::consensus::hotstuff::pacemaker";
 const MAX_DELTA: Duration = Duration::from_secs(300);
-const BLOCK_TIME: Duration = Duration::from_secs(10);
 
 pub struct PaceMaker {
     pace_maker_handle: PaceMakerHandle,
     handle_receiver: mpsc::Receiver<PacemakerRequest>,
     current_view: CurrentView,
     current_high_qc_height: NodeHeight,
+    block_time: Duration,
 }
 
 impl PaceMaker {
-    pub fn new() -> Self {
+    pub fn new(max_base_time: Duration) -> Self {
         let (sender, receiver) = mpsc::channel(100);
 
         let on_beat = OnBeat::new();
@@ -49,6 +50,7 @@ impl PaceMaker {
             ),
             current_view: current_height,
             current_high_qc_height: NodeHeight(0),
+            block_time: max_base_time,
         }
     }
 
@@ -101,7 +103,7 @@ impl PaceMaker {
                                 info!(target: LOG_TARGET, "Reset! Current height: {}, Delta: {:.2?}", self.current_view, delta);
                                 leader_timeout.as_mut().reset(tokio::time::Instant::now() + delta);
                                 // set a timer for when we must send a block...
-                                block_timer.as_mut().reset(tokio::time::Instant::now() + BLOCK_TIME);
+                                block_timer.as_mut().reset(tokio::time::Instant::now() + self.block_time);
                            },
                             PacemakerRequest::Start { high_qc_height } => {
                                 info!(target: LOG_TARGET, "🚀 Starting pacemaker at leaf height {} and high QC: {}", self.current_view, high_qc_height);
@@ -112,7 +114,7 @@ impl PaceMaker {
                                 let delta = self.delta_time();
                                 info!(target: LOG_TARGET, "Reset! Current height: {}, Delta: {:.2?}", self.current_view, delta);
                                 leader_timeout.as_mut().reset(tokio::time::Instant::now() + delta);
-                                block_timer.as_mut().reset(tokio::time::Instant::now() + BLOCK_TIME);
+                                block_timer.as_mut().reset(tokio::time::Instant::now() + self.block_time);
                                 on_beat.beat();
                                 started = true;
                             }
@@ -130,11 +132,11 @@ impl PaceMaker {
                     }
                 },
                 () = &mut block_timer => {
-                    block_timer.as_mut().reset(tokio::time::Instant::now() + BLOCK_TIME);
+                    block_timer.as_mut().reset(tokio::time::Instant::now() + self.block_time);
                     on_force_beat.beat(None);
                 }
                 () = &mut leader_timeout => {
-                    block_timer.as_mut().reset(tokio::time::Instant::now() + BLOCK_TIME);
+                    block_timer.as_mut().reset(tokio::time::Instant::now() + self.block_time);
 
                     let delta = self.delta_time();
                     leader_timeout.as_mut().reset(tokio::time::Instant::now() + delta);
@@ -156,7 +158,7 @@ impl PaceMaker {
         let current_height = self.current_view.get_height();
         if current_height.is_zero() || self.current_high_qc_height.is_zero() {
             // Allow extra time for the first block
-            return BLOCK_TIME * 2;
+            return self.block_time * 2;
         }
         let exp = u32::try_from(cmp::min(
             u64::from(u32::MAX),
@@ -169,7 +171,7 @@ impl PaceMaker {
         );
         // TODO: get real avg latency
         let avg_latency = Duration::from_secs(2);
-        BLOCK_TIME + delta + avg_latency
+        self.block_time + delta + avg_latency
     }
 }
 

dan_layer/consensus/src/hotstuff/substate_store/pending_store.rs

@@ -79,7 +79,9 @@ impl<'a, 'tx, TStore: StateStore + 'a + 'tx> ReadableSubstateStore for PendingSu
         }
 
         let Some(substate) = SubstateRecord::get(self.read_transaction(), &id.to_substate_address()).optional()? else {
-            return Err(SubstateStoreError::SubstateNotFound { address: id.to_substate_address() });
+            return Err(SubstateStoreError::SubstateNotFound {
+                address: id.to_substate_address(),
+            });
         };
         Ok(substate.into_substate())
     }

dan_layer/consensus/src/hotstuff/worker.rs

@@ -97,7 +97,7 @@ impl<TConsensusSpec: ConsensusSpec> HotstuffWorker<TConsensusSpec> {
         config: HotstuffConfig,
     ) -> Self {
         let (tx_missing_transactions, rx_missing_transactions) = mpsc::unbounded_channel();
-        let pacemaker = PaceMaker::new();
+        let pacemaker = PaceMaker::new(config.pacemaker_max_base_time);
         let vote_receiver = VoteReceiver::new(
             network,
             state_store.clone(),

dan_layer/consensus_tests/src/support/validator/builder.rs

@@ -133,6 +133,7 @@ impl ValidatorBuilder {
             HotstuffConfig {
                 max_base_layer_blocks_ahead: 5,
                 max_base_layer_blocks_behind: 5,
+                pacemaker_max_base_time: std::time::Duration::from_secs(10),
             },
         );
 

dan_layer/engine/src/runtime/working_state.rs

@@ -197,7 +197,7 @@ impl WorkingState {
                 Ok(Some((before, after)))
             })?;
 
-        let Some((before, after))= maybe_before_and_after else {
+        let Some((before, after)) = maybe_before_and_after else {
             return Ok(());
         };