Change consensus pipeline backpressure to exclude consensus ordering …

…latency (aptos-labs#8246)

consensus/src/block_storage/block_store.rs

@@ -42,6 +42,8 @@ mod block_store_test;
 #[path = "sync_manager.rs"]
 pub mod sync_manager;
 
+const MAX_ORDERING_PIPELINE_LATENCY_REDUCTION: Duration = Duration::from_secs(1);
+
 fn update_counters_for_ordered_blocks(ordered_blocks: &[Arc<ExecutedBlock>]) {
     for block in ordered_blocks {
         observe_block(block.block().timestamp_usecs(), BlockStage::ORDERED);
@@ -540,13 +542,13 @@ impl BlockStore {
         }
 
         let latency_to_committed = latency_from_proposal(proposal_timestamp, committed_timestamp);
+        let latency_to_ordered = latency_from_proposal(proposal_timestamp, ordered_timestamp);
 
         info!(
             pending_rounds = pending_rounds,
             ordered_round = ordered_round,
             commit_round = commit_round,
-            latency_to_ordered_ms =
-                latency_from_proposal(proposal_timestamp, ordered_timestamp).as_millis() as u64,
+            latency_to_ordered_ms = latency_to_ordered.as_millis() as u64,
             latency_to_committed_ms = latency_to_committed.as_millis() as u64,
             latency_to_commit_cert_ms =
                 latency_from_proposal(proposal_timestamp, commit_cert_timestamp).as_millis() as u64,
@@ -557,6 +559,7 @@ impl BlockStore {
         counters::CONSENSUS_PROPOSAL_PENDING_DURATION.observe(latency_to_committed.as_secs_f64());
 
         latency_to_committed
+            .saturating_sub(latency_to_ordered.min(MAX_ORDERING_PIPELINE_LATENCY_REDUCTION))
     }
 }