[dag] stop voting based on pipeline latency

config/src/config/dag_consensus_config.rs

@@ -135,15 +135,32 @@ impl Default for DagRoundStateConfig {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct DagHealthConfig {
+    pub chain_backoff_config: Vec<ChainHealthBackoffValues>,
+    pub voter_pipeline_latency_limit_ms: u64,
+    pub pipeline_backpressure_config: Vec<PipelineBackpressureValues>,
+}
+
+impl Default for DagHealthConfig {
+    fn default() -> Self {
+        Self {
+            chain_backoff_config: Vec::new(),
+            voter_pipeline_latency_limit_ms: 30_000,
+            pipeline_backpressure_config: Vec::new(),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)]
 #[serde(default, deny_unknown_fields)]
 pub struct DagConsensusConfig {
     pub node_payload_config: DagPayloadConfig,
     pub rb_config: ReliableBroadcastConfig,
     pub fetcher_config: DagFetcherConfig,
     pub round_state_config: DagRoundStateConfig,
-    pub chain_backoff_config: Vec<ChainHealthBackoffValues>,
-    pub pipeline_backpressure_config: Vec<PipelineBackpressureValues>,
+    pub health_config: DagHealthConfig,
     #[serde(default = "QuorumStoreConfig::default_for_dag")]
     pub quorum_store: QuorumStoreConfig,
 }

consensus/src/dag/adapter.rs

@@ -206,7 +206,7 @@ impl OrderedNotifier for OrderedNotifierAdapter {
                       commit_decision: LedgerInfoWithSignatures| {
                     block_created_ts
                         .write()
-                        .split_off(&(commit_decision.commit_info().round() + 1));
+                        .retain(|&round, _| round > commit_decision.commit_info().round());
                     dag.write()
                         .commit_callback(commit_decision.commit_info().round());
                     ledger_info_provider

consensus/src/dag/bootstrap.rs

@@ -593,11 +593,17 @@ impl DagBootstrapper {
         );
 
         let chain_health: Arc<dyn TChainHealth> = ChainHealthBackoff::new(
-            ChainHealthBackoffConfig::new(self.config.chain_backoff_config.clone()),
+            ChainHealthBackoffConfig::new(self.config.health_config.chain_backoff_config.clone()),
             commit_history.clone(),
         );
         let pipeline_health = PipelineLatencyBasedBackpressure::new(
-            PipelineBackpressureConfig::new(self.config.pipeline_backpressure_config.clone()),
+            Duration::from_millis(self.config.health_config.voter_pipeline_latency_limit_ms),
+            PipelineBackpressureConfig::new(
+                self.config
+                    .health_config
+                    .pipeline_backpressure_config
+                    .clone(),
+            ),
             ordered_notifier.clone(),
         );
         let health_backoff =
@@ -616,7 +622,7 @@ impl DagBootstrapper {
             round_state,
             self.onchain_config.dag_ordering_causal_history_window as Round,
             self.config.node_payload_config.clone(),
-            health_backoff,
+            health_backoff.clone(),
             self.quorum_store_enabled,
         );
         let rb_handler = NodeBroadcastHandler::new(
@@ -628,6 +634,7 @@ impl DagBootstrapper {
             self.config.node_payload_config.clone(),
             self.vtxn_config.clone(),
             self.features.clone(),
+            health_backoff,
         );
         let fetch_handler = FetchRequestHandler::new(dag_store.clone(), self.epoch_state.clone());
 

consensus/src/dag/errors.rs

@@ -14,6 +14,8 @@ pub enum NodeBroadcastHandleError {
     MissingParents,
     #[error("stale round number")]
     StaleRound(Round),
+    #[error("refused to vote")]
+    VoteRefused,
 }
 
 #[derive(Clone, Debug, ThisError, Serialize, Deserialize)]

consensus/src/dag/health/backoff.rs

@@ -6,6 +6,7 @@ use aptos_consensus_types::common::Round;
 use aptos_types::epoch_state::EpochState;
 use std::{sync::Arc, time::Duration};
 
+#[derive(Clone)]
 pub struct HealthBackoff {
     epoch_state: Arc<EpochState>,
     chain_health: Arc<dyn TChainHealth>,
@@ -77,4 +78,8 @@ impl HealthBackoff {
             .unwrap_or_default()
             .max(pipeline_backoff.unwrap_or_default())
     }
+
+    pub fn stop_voting(&self) -> bool {
+        self.pipeline_health.stop_voting()
+    }
 }

consensus/src/dag/health/pipeline_health.rs

@@ -9,6 +9,8 @@ pub trait TPipelineHealth: Send + Sync {
     fn get_backoff(&self) -> Option<Duration>;
 
     fn get_payload_limits(&self) -> Option<(u64, u64)>;
+
+    fn stop_voting(&self) -> bool;
 }
 
 pub struct NoPipelineBackpressure {}
@@ -27,19 +29,26 @@ impl TPipelineHealth for NoPipelineBackpressure {
     fn get_payload_limits(&self) -> Option<(u64, u64)> {
         None
     }
+
+    fn stop_voting(&self) -> bool {
+        false
+    }
 }
 
 pub struct PipelineLatencyBasedBackpressure {
+    voter_pipeline_latency_limit: Duration,
     pipeline_config: PipelineBackpressureConfig,
     adapter: Arc<OrderedNotifierAdapter>,
 }
 
 impl PipelineLatencyBasedBackpressure {
     pub(in crate::dag) fn new(
+        voter_pipeline_latency_limit: Duration,
         pipeline_config: PipelineBackpressureConfig,
         adapter: Arc<OrderedNotifierAdapter>,
     ) -> Arc<Self> {
         Arc::new(Self {
+            voter_pipeline_latency_limit,
             pipeline_config,
             adapter,
         })
@@ -63,4 +72,9 @@ impl TPipelineHealth for PipelineLatencyBasedBackpressure {
             )
         })
     }
+
+    fn stop_voting(&self) -> bool {
+        let latency = self.adapter.pipeline_pending_latency();
+        latency > self.voter_pipeline_latency_limit
+    }
 }

consensus/src/dag/rb_handler.rs

@@ -31,6 +31,8 @@ use aptos_types::{
 use async_trait::async_trait;
 use std::{collections::BTreeMap, mem, sync::Arc};
 
+use super::health::HealthBackoff;
+
 pub(crate) struct NodeBroadcastHandler {
     dag: Arc<RwLock<Dag>>,
     votes_by_round_peer: BTreeMap<Round, BTreeMap<Author, Vote>>,
@@ -41,6 +43,7 @@ pub(crate) struct NodeBroadcastHandler {
     payload_config: DagPayloadConfig,
     vtxn_config: ValidatorTxnConfig,
     features: Features,
+    health_backoff: HealthBackoff,
 }
 
 impl NodeBroadcastHandler {
@@ -53,6 +56,7 @@ impl NodeBroadcastHandler {
         payload_config: DagPayloadConfig,
         vtxn_config: ValidatorTxnConfig,
         features: Features,
+        health_backoff: HealthBackoff,
     ) -> Self {
         let epoch = epoch_state.epoch;
         let votes_by_round_peer = read_votes_from_storage(&storage, epoch);
@@ -67,6 +71,7 @@ impl NodeBroadcastHandler {
             payload_config,
             vtxn_config,
             features,
+            health_backoff,
         }
     }
 
@@ -189,6 +194,11 @@ impl RpcHandler for NodeBroadcastHandler {
     type Response = Vote;
 
     async fn process(&mut self, node: Self::Request) -> anyhow::Result<Self::Response> {
+        ensure!(
+            !self.health_backoff.stop_voting(),
+            NodeBroadcastHandleError::VoteRefused
+        );
+
         let node = self.validate(node)?;
         observe_node(node.timestamp(), NodeStage::NodeReceived);
         debug!(LogSchema::new(LogEvent::ReceiveNode)

consensus/src/dag/tests/rb_handler_tests.rs

@@ -5,6 +5,7 @@ use crate::dag::{
     dag_fetcher::TFetchRequester,
     dag_store::Dag,
     errors::NodeBroadcastHandleError,
+    health::{HealthBackoff, NoChainHealth, NoPipelineBackpressure},
     rb_handler::NodeBroadcastHandler,
     storage::DAGStorage,
     tests::{
@@ -57,6 +58,12 @@ async fn test_node_broadcast_receiver_succeed() {
         TEST_DAG_WINDOW,
     )));
 
+    let health_backoff = HealthBackoff::new(
+        epoch_state.clone(),
+        NoChainHealth::new(),
+        NoPipelineBackpressure::new(),
+    );
+
     let wellformed_node = new_node(1, 10, signers[0].author(), vec![]);
     let equivocating_node = new_node(1, 20, signers[0].author(), vec![]);
 
@@ -70,7 +77,11 @@ async fn test_node_broadcast_receiver_succeed() {
         Arc::new(MockFetchRequester {}),
         DagPayloadConfig::default(),
         ValidatorTxnConfig::default_disabled(),
+<<<<<<< HEAD
         Features::default(),
+=======
+        health_backoff,
+>>>>>>> 85978d5bc8 ([dag] stop voting based on pipeline latency)
     );
 
     let expected_result = Vote::new(
@@ -117,7 +128,15 @@ async fn test_node_broadcast_receiver_failure() {
                 Arc::new(MockFetchRequester {}),
                 DagPayloadConfig::default(),
                 ValidatorTxnConfig::default_disabled(),
+<<<<<<< HEAD
                 Features::default(),
+=======
+                HealthBackoff::new(
+                    epoch_state.clone(),
+                    NoChainHealth::new(),
+                    NoPipelineBackpressure::new(),
+                ),
+>>>>>>> 85978d5bc8 ([dag] stop voting based on pipeline latency)
             )
         })
         .collect();
@@ -201,7 +220,15 @@ async fn test_node_broadcast_receiver_storage() {
         Arc::new(MockFetchRequester {}),
         DagPayloadConfig::default(),
         ValidatorTxnConfig::default_disabled(),
+<<<<<<< HEAD
         Features::default(),
+=======
+        HealthBackoff::new(
+            epoch_state.clone(),
+            NoChainHealth::new(),
+            NoPipelineBackpressure::new(),
+        ),
+>>>>>>> 85978d5bc8 ([dag] stop voting based on pipeline latency)
     );
     let sig = rb_receiver.process(node).await.expect("must succeed");
 
@@ -213,12 +240,20 @@ async fn test_node_broadcast_receiver_storage() {
     let mut rb_receiver = NodeBroadcastHandler::new(
         dag,
         signers[3].clone(),
-        epoch_state,
+        epoch_state.clone(),
         storage.clone(),
         Arc::new(MockFetchRequester {}),
         DagPayloadConfig::default(),
         ValidatorTxnConfig::default_disabled(),
+<<<<<<< HEAD
         Features::default(),
+=======
+        HealthBackoff::new(
+            epoch_state,
+            NoChainHealth::new(),
+            NoPipelineBackpressure::new(),
+        ),
+>>>>>>> 85978d5bc8 ([dag] stop voting based on pipeline latency)
     );
     assert_ok!(rb_receiver.gc_before_round(2));
     assert_eq!(storage.get_votes().unwrap().len(), 0);