[consensus] return quorum rounds from round prober

consensus/core/src/authority_service.rs

@@ -600,20 +600,21 @@ async fn make_recv_future<T: Clone>(
 
 #[cfg(test)]
 mod tests {
-    use crate::commit::CommitRange;
-    use crate::test_dag_builder::DagBuilder;
     use crate::{
         authority_service::AuthorityService,
         block::BlockAPI,
         block::{BlockRef, SignedBlock, TestBlock, VerifiedBlock},
+        commit::CommitRange,
         commit_vote_monitor::CommitVoteMonitor,
         context::Context,
         core_thread::{CoreError, CoreThreadDispatcher},
         dag_state::DagState,
         error::ConsensusResult,
         network::{BlockStream, NetworkClient, NetworkService},
+        round_prober::QuorumRound,
         storage::mem_store::MemStore,
         synchronizer::Synchronizer,
+        test_dag_builder::DagBuilder,
         Round,
     };
     use async_trait::async_trait;
@@ -665,7 +666,11 @@ mod tests {
             todo!()
         }
 
-        fn set_propagation_delay(&self, _delay: Round) -> Result<(), CoreError> {
+        fn set_propagation_delay_and_quorum_rounds(
+            &self,
+            _delay: Round,
+            _quorum_rounds: Vec<QuorumRound>,
+        ) -> Result<(), CoreError> {
             todo!()
         }
 

consensus/core/src/core.rs

@@ -30,6 +30,7 @@ use crate::{
     dag_state::DagState,
     error::{ConsensusError, ConsensusResult},
     leader_schedule::LeaderSchedule,
+    round_prober::QuorumRound,
     stake_aggregator::{QuorumThreshold, StakeAggregator},
     threshold_clock::ThresholdClock,
     transaction::TransactionConsumer,
@@ -645,7 +646,12 @@ impl Core {
     }
 
     /// Sets the delay by round for propagating blocks to a quorum.
-    pub(crate) fn set_propagation_delay(&mut self, delay: Round) {
+    // TODO: Will set the quorum round per authority in ancestor state manager.
+    pub(crate) fn set_propagation_delay_and_quorum_rounds(
+        &mut self,
+        delay: Round,
+        _quorum_rounds: Vec<QuorumRound>,
+    ) {
         info!("Propagation round delay set to: {delay}");
         self.propagation_delay = delay;
     }
@@ -1726,7 +1732,7 @@ mod test {
         );
 
         // Use a large propagation delay to disable proposing.
-        core.set_propagation_delay(1000);
+        core.set_propagation_delay_and_quorum_rounds(1000, vec![]);
 
         // Make propagation delay the only reason for not proposing.
         core.set_subscriber_exists(true);
@@ -1735,7 +1741,7 @@ mod test {
         assert!(core.try_propose(true).unwrap().is_none());
 
         // Let Core know there is no propagation delay.
-        core.set_propagation_delay(0);
+        core.set_propagation_delay_and_quorum_rounds(0, vec![]);
 
         // Proposing now would succeed.
         assert!(core.try_propose(true).unwrap().is_some());

consensus/core/src/core_thread.rs

@@ -27,6 +27,7 @@ use crate::{
     core_thread::CoreError::Shutdown,
     dag_state::DagState,
     error::{ConsensusError, ConsensusResult},
+    round_prober::QuorumRound,
     BlockAPI as _,
 };
 
@@ -65,8 +66,13 @@ pub trait CoreThreadDispatcher: Sync + Send + 'static {
     /// It is not a guarantee that produced blocks will be accepted by peers.
     fn set_subscriber_exists(&self, exists: bool) -> Result<(), CoreError>;
 
-    /// Sets the estimated delay to propagate a block to a quorum of peers, in number of rounds.
-    fn set_propagation_delay(&self, delay: Round) -> Result<(), CoreError>;
+    /// Sets the estimated delay to propagate a block to a quorum of peers, in
+    /// number of rounds, and the quorum rounds for all authorities.
+    fn set_propagation_delay_and_quorum_rounds(
+        &self,
+        delay: Round,
+        quorum_rounds: Vec<QuorumRound>,
+    ) -> Result<(), CoreError>;
 
     fn set_last_known_proposed_round(&self, round: Round) -> Result<(), CoreError>;
 
@@ -91,7 +97,7 @@ struct CoreThread {
     core: Core,
     receiver: Receiver<CoreThreadCommand>,
     rx_subscriber_exists: watch::Receiver<bool>,
-    rx_propagation_delay: watch::Receiver<Round>,
+    rx_propagation_delay_and_quorum_rounds: watch::Receiver<(Round, Vec<QuorumRound>)>,
     rx_last_known_proposed_round: watch::Receiver<Round>,
     context: Arc<Context>,
 }
@@ -141,11 +147,11 @@ impl CoreThread {
                         self.core.new_block(Round::MAX, true)?;
                     }
                 }
-                _ = self.rx_propagation_delay.changed() => {
-                    let _scope = monitored_scope("CoreThread::loop::set_propagation_delay");
+                _ = self.rx_propagation_delay_and_quorum_rounds.changed() => {
+                    let _scope = monitored_scope("CoreThread::loop::set_propagation_delay_and_quorum_rounds");
                     let should_propose_before = self.core.should_propose();
-                    let delay = *self.rx_propagation_delay.borrow();
-                    self.core.set_propagation_delay(delay);
+                    let (delay, quorum_rounds) = self.rx_propagation_delay_and_quorum_rounds.borrow().clone();
+                    self.core.set_propagation_delay_and_quorum_rounds(delay, quorum_rounds);
                     if !should_propose_before && self.core.should_propose() {
                         // If core cannnot propose before but can propose now, try to produce a new block to ensure liveness,
                         // because block proposal could have been skipped.
@@ -164,7 +170,7 @@ pub(crate) struct ChannelCoreThreadDispatcher {
     context: Arc<Context>,
     sender: WeakSender<CoreThreadCommand>,
     tx_subscriber_exists: Arc<watch::Sender<bool>>,
-    tx_propagation_delay: Arc<watch::Sender<Round>>,
+    tx_propagation_delay_and_quorum_rounds: Arc<watch::Sender<(Round, Vec<QuorumRound>)>>,
     tx_last_known_proposed_round: Arc<watch::Sender<Round>>,
     highest_received_rounds: Arc<Vec<AtomicU32>>,
 }
@@ -190,16 +196,17 @@ impl ChannelCoreThreadDispatcher {
         let (sender, receiver) =
             channel("consensus_core_commands", CORE_THREAD_COMMANDS_CHANNEL_SIZE);
         let (tx_subscriber_exists, mut rx_subscriber_exists) = watch::channel(false);
-        let (tx_propagation_delay, mut rx_propagation_delay) = watch::channel(0);
+        let (tx_propagation_delay_and_quorum_rounds, mut rx_propagation_delay_and_quorum_rounds) =
+            watch::channel((0, vec![(0, 0); context.committee.size()]));
         let (tx_last_known_proposed_round, mut rx_last_known_proposed_round) = watch::channel(0);
         rx_subscriber_exists.mark_unchanged();
-        rx_propagation_delay.mark_unchanged();
+        rx_propagation_delay_and_quorum_rounds.mark_unchanged();
         rx_last_known_proposed_round.mark_unchanged();
         let core_thread = CoreThread {
             core,
             receiver,
             rx_subscriber_exists,
-            rx_propagation_delay,
+            rx_propagation_delay_and_quorum_rounds,
             rx_last_known_proposed_round,
             context: context.clone(),
         };
@@ -221,7 +228,9 @@ impl ChannelCoreThreadDispatcher {
             context,
             sender: sender.downgrade(),
             tx_subscriber_exists: Arc::new(tx_subscriber_exists),
-            tx_propagation_delay: Arc::new(tx_propagation_delay),
+            tx_propagation_delay_and_quorum_rounds: Arc::new(
+                tx_propagation_delay_and_quorum_rounds,
+            ),
             tx_last_known_proposed_round: Arc::new(tx_last_known_proposed_round),
             highest_received_rounds: Arc::new(highest_received_rounds),
         };
@@ -279,9 +288,13 @@ impl CoreThreadDispatcher for ChannelCoreThreadDispatcher {
             .map_err(|e| Shutdown(e.to_string()))
     }
 
-    fn set_propagation_delay(&self, delay: Round) -> Result<(), CoreError> {
-        self.tx_propagation_delay
-            .send(delay)
+    fn set_propagation_delay_and_quorum_rounds(
+        &self,
+        delay: Round,
+        quorum_rounds: Vec<QuorumRound>,
+    ) -> Result<(), CoreError> {
+        self.tx_propagation_delay_and_quorum_rounds
+            .send((delay, quorum_rounds))
             .map_err(|e| Shutdown(e.to_string()))
     }
 
@@ -356,7 +369,11 @@ impl CoreThreadDispatcher for MockCoreThreadDispatcher {
         todo!()
     }
 
-    fn set_propagation_delay(&self, _delay: Round) -> Result<(), CoreError> {
+    fn set_propagation_delay_and_quorum_rounds(
+        &self,
+        _delay: Round,
+        _quorum_rounds: Vec<QuorumRound>,
+    ) -> Result<(), CoreError> {
         todo!()
     }
 

consensus/core/src/leader_timeout.rs

@@ -133,6 +133,7 @@ mod tests {
     use crate::core::CoreSignals;
     use crate::core_thread::{CoreError, CoreThreadDispatcher};
     use crate::leader_timeout::LeaderTimeoutTask;
+    use crate::round_prober::QuorumRound;
 
     #[derive(Clone, Default)]
     struct MockCoreThreadDispatcher {
@@ -171,7 +172,11 @@ mod tests {
             todo!()
         }
 
-        fn set_propagation_delay(&self, _delay: Round) -> Result<(), CoreError> {
+        fn set_propagation_delay_and_quorum_rounds(
+            &self,
+            _delay: Round,
+            _quorum_rounds: Vec<QuorumRound>,
+        ) -> Result<(), CoreError> {
             todo!()
         }
 

consensus/core/src/metrics.rs

@@ -173,6 +173,7 @@ pub(crate) struct NodeMetrics {
     pub(crate) commit_sync_fetch_once_latency: Histogram,
     pub(crate) commit_sync_fetch_once_errors: IntCounterVec,
     pub(crate) round_prober_quorum_round_gaps: IntGaugeVec,
+    pub(crate) round_prober_low_quorum_round: IntGaugeVec,
     pub(crate) round_prober_current_round_gaps: IntGaugeVec,
     pub(crate) round_prober_propagation_delays: Histogram,
     pub(crate) round_prober_last_propagation_delay: IntGauge,
@@ -611,6 +612,12 @@ impl NodeMetrics {
                 &["authority"],
                 registry
             ).unwrap(),
+            round_prober_low_quorum_round: register_int_gauge_vec_with_registry!(
+                "round_prober_low_quorum_round",
+                "Low quorum round among peers for blocks proposed from each authority",
+                &["authority"],
+                registry
+            ).unwrap(),
             round_prober_current_round_gaps: register_int_gauge_vec_with_registry!(
                 "round_prober_current_round_gaps",
                 "Round gaps from local last proposed round to the low quorum round of each peer. Can be negative.",

consensus/core/src/round_prober.rs

@@ -207,6 +207,10 @@ impl<C: NetworkClient> RoundProber<C> {
                 .round_prober_quorum_round_gaps
                 .with_label_values(&[&authority.hostname])
                 .set((high - low) as i64);
+            node_metrics
+                .round_prober_low_quorum_round
+                .with_label_values(&[&authority.hostname])
+                .set(*low as i64);
             // The gap can be negative if this validator is lagging behind the network.
             node_metrics
                 .round_prober_current_round_gaps
@@ -231,7 +235,7 @@ impl<C: NetworkClient> RoundProber<C> {
             .set(propagation_delay as i64);
         if let Err(e) = self
             .core_thread_dispatcher
-            .set_propagation_delay(propagation_delay)
+            .set_propagation_delay_and_quorum_rounds(propagation_delay, quorum_rounds.clone())
         {
             tracing::warn!(
                 "Failed to set propagation delay {propagation_delay} on Core: {:?}",
@@ -293,6 +297,7 @@ mod test {
     use consensus_config::AuthorityIndex;
     use parking_lot::{Mutex, RwLock};
 
+    use super::QuorumRound;
     use crate::{
         block::BlockRef,
         commit::CommitRange,
@@ -309,19 +314,25 @@ mod test {
     struct FakeThreadDispatcher {
         highest_received_rounds: Vec<Round>,
         propagation_delay: Mutex<Round>,
+        quorum_rounds: Mutex<Vec<QuorumRound>>,
     }
 
     impl FakeThreadDispatcher {
         fn new(highest_received_rounds: Vec<Round>) -> Self {
             Self {
                 highest_received_rounds,
                 propagation_delay: Mutex::new(0),
+                quorum_rounds: Mutex::new(Vec::new()),
             }
         }
 
         fn propagation_delay(&self) -> Round {
             *self.propagation_delay.lock()
         }
+
+        fn quorum_rounds(&self) -> Vec<QuorumRound> {
+            self.quorum_rounds.lock().clone()
+        }
     }
 
     #[async_trait]
@@ -345,7 +356,13 @@ mod test {
             unimplemented!()
         }
 
-        fn set_propagation_delay(&self, delay: Round) -> Result<(), CoreError> {
+        fn set_propagation_delay_and_quorum_rounds(
+            &self,
+            delay: Round,
+            quorum_rounds: Vec<QuorumRound>,
+        ) -> Result<(), CoreError> {
+            let mut quorum_round_per_authority = self.quorum_rounds.lock();
+            *quorum_round_per_authority = quorum_rounds;
             let mut propagation_delay = self.propagation_delay.lock();
             *propagation_delay = delay;
             Ok(())
@@ -492,6 +509,18 @@ mod test {
             ]
         );
 
+        assert_eq!(
+            core_thread_dispatcher.quorum_rounds(),
+            vec![
+                (100, 105),
+                (0, 115),
+                (103, 130),
+                (0, 0),
+                (105, 150),
+                (106, 160),
+                (107, 170)
+            ]
+        );
         // 110 - 100 = 10
         assert_eq!(propagation_delay, 10);
         assert_eq!(core_thread_dispatcher.propagation_delay(), 10);