brian/test graceful overload

config/src/config/consensus_config.rs

@@ -13,9 +13,14 @@ use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 
 // NOTE: when changing, make sure to update QuorumStoreBackPressureConfig::backlog_txn_limit_count as well.
-const MAX_SENDING_BLOCK_TXNS: u64 = 1900;
+const MAX_SENDING_BLOCK_UNIQUE_TXNS: u64 = 1900;
 pub(crate) static MAX_RECEIVING_BLOCK_TXNS: Lazy<u64> =
-    Lazy::new(|| 10000.max(2 * MAX_SENDING_BLOCK_TXNS));
+    Lazy::new(|| 10000.max(2 * MAX_SENDING_BLOCK_UNIQUE_TXNS));
+// The receiving validator can accept upto 2k more transactions in the block than the max sending limit.
+// The extra cushion of 2k transactions is added just in case we need to increase the max sending limit in the future.
+static MAX_SENDING_BLOCK_TXNS: Lazy<u64> =
+    Lazy::new(|| MAX_SENDING_BLOCK_UNIQUE_TXNS.max(MAX_RECEIVING_BLOCK_TXNS.saturating_sub(2000)));
+
 // stop reducing size at this point, so 1MB transactions can still go through
 const MIN_BLOCK_BYTES_OVERRIDE: u64 = 1024 * 1024 + BATCH_PADDING_BYTES as u64;
 
@@ -25,6 +30,7 @@ pub struct ConsensusConfig {
     // length of inbound queue of messages
     pub max_network_channel_size: usize,
     pub max_sending_block_txns: u64,
+    pub max_sending_block_unique_txns: u64,
     pub max_sending_block_bytes: u64,
     pub max_sending_inline_txns: u64,
     pub max_sending_inline_bytes: u64,
@@ -152,7 +158,8 @@ impl Default for ConsensusConfig {
     fn default() -> ConsensusConfig {
         ConsensusConfig {
             max_network_channel_size: 1024,
-            max_sending_block_txns: MAX_SENDING_BLOCK_TXNS,
+            max_sending_block_txns: *MAX_SENDING_BLOCK_TXNS,
+            max_sending_block_unique_txns: MAX_SENDING_BLOCK_UNIQUE_TXNS,
             max_sending_block_bytes: 3 * 1024 * 1024, // 3MB
             max_receiving_block_txns: *MAX_RECEIVING_BLOCK_TXNS,
             max_sending_inline_txns: 100,
@@ -192,14 +199,14 @@ impl Default for ConsensusConfig {
                     // pipeline once quorum on execution result among validators has been reached
                     // (so-(badly)-called "commit certificate"), meaning 2f+1 validators have finished execution.
                     back_pressure_pipeline_latency_limit_ms: 800,
-                    max_sending_block_txns_override: MAX_SENDING_BLOCK_TXNS,
+                    max_sending_block_txns_override: *MAX_SENDING_BLOCK_TXNS,
                     max_sending_block_bytes_override: 5 * 1024 * 1024,
                     backpressure_proposal_delay_ms: 100,
                     max_txns_from_block_to_execute: None,
                 },
                 PipelineBackpressureValues {
                     back_pressure_pipeline_latency_limit_ms: 1100,
-                    max_sending_block_txns_override: MAX_SENDING_BLOCK_TXNS,
+                    max_sending_block_txns_override: *MAX_SENDING_BLOCK_TXNS,
                     max_sending_block_bytes_override: 5 * 1024 * 1024,
                     backpressure_proposal_delay_ms: 200,
                     max_txns_from_block_to_execute: None,

config/src/config/quorum_store_config.rs

@@ -27,16 +27,19 @@ pub struct QuorumStoreBackPressureConfig {
 impl Default for QuorumStoreBackPressureConfig {
     fn default() -> QuorumStoreBackPressureConfig {
         QuorumStoreBackPressureConfig {
-            // QS will be backpressured if the remaining total txns is more than this number
-            // Roughly, target TPS * commit latency seconds
-            backlog_txn_limit_count: 12_000,
+            // QS will be backpressured if the remaining total txns is more than this number.
+            // The remaining total txns is the sum of batch sizes of all unexpired batches for which the
+            // validator has recieved Proof of Store but has not yet committed.
+            // The batches generated by different validators could have common transactions.
+            // Roughly, target TPS * commit latency seconds * duplication factor of a transaction in batches.
+            backlog_txn_limit_count: 36_000,
             // QS will create batches at the max rate until this number is reached
-            backlog_per_validator_batch_limit_count: 4,
+            backlog_per_validator_batch_limit_count: 20,
             decrease_duration_ms: 1000,
             increase_duration_ms: 1000,
             decrease_fraction: 0.5,
             dynamic_min_txn_per_s: 160,
-            dynamic_max_txn_per_s: 4000,
+            dynamic_max_txn_per_s: 12000,
         }
     }
 }

consensus/consensus-types/src/common.rs

@@ -51,6 +51,36 @@ impl fmt::Display for TransactionSummary {
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize, Hash, Ord, PartialOrd)]
+pub struct TransactionSynopsis {
+    pub sender: AccountAddress,
+    pub sequence_number: u64,
+    pub expiration_timestamp_secs: u64,
+    pub hash: HashValue,
+}
+
+impl TransactionSynopsis {
+    pub fn new(
+        sender: AccountAddress,
+        sequence_number: u64,
+        expiration_timestamp_secs: u64,
+        hash: HashValue,
+    ) -> Self {
+        Self {
+            sender,
+            sequence_number,
+            expiration_timestamp_secs,
+            hash,
+        }
+    }
+}
+
+impl fmt::Display for TransactionSynopsis {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}:{}", self.sender, self.sequence_number,)
+    }
+}
+
 #[derive(Clone)]
 pub struct TransactionInProgress {
     pub gas_unit_price: u64,

consensus/consensus-types/src/request_response.rs

@@ -9,7 +9,9 @@ use std::{fmt, fmt::Formatter};
 pub enum GetPayloadCommand {
     /// Request to pull block to submit to consensus.
     GetPayloadRequest(
-        // max block size
+        // max number of transactions in the block
+        u64,
+        // max number of unique transactions in the block
         u64,
         // max byte size
         u64,
@@ -31,6 +33,7 @@ impl fmt::Display for GetPayloadCommand {
         match self {
             GetPayloadCommand::GetPayloadRequest(
                 max_txns,
+                max_unique_txns,
                 max_bytes,
                 max_inline_txns,
                 max_inline_bytes,
@@ -40,8 +43,8 @@ impl fmt::Display for GetPayloadCommand {
             ) => {
                 write!(
                     f,
-                    "GetPayloadRequest [max_txns: {}, max_bytes: {}, max_inline_txns: {}, max_inline_bytes:{}, return_non_full: {},  excluded: {}]",
-                    max_txns, max_bytes, max_inline_txns, max_inline_bytes, return_non_full, excluded
+                    "GetPayloadRequest [max_txns: {}, max_unique_txns: {}, max_bytes: {}, max_inline_txns: {}, max_inline_bytes:{}, return_non_full: {},  excluded: {}]",
+                    max_txns, max_unique_txns, max_bytes, max_inline_txns, max_inline_bytes, return_non_full, excluded
                 )
             },
         }

consensus/src/dag/dag_driver.rs

@@ -257,6 +257,7 @@ impl DagDriver {
             .pull_payload(
                 Duration::from_millis(self.payload_config.payload_pull_max_poll_time_ms),
                 max_txns,
+                max_txns,
                 max_size_bytes,
                 // TODO: Set max_inline_items and max_inline_bytes correctly
                 100,

consensus/src/epoch_manager.rs

@@ -839,6 +839,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             self.time_service.clone(),
             Duration::from_millis(self.config.quorum_store_poll_time_ms),
             self.config.max_sending_block_txns,
+            self.config.max_sending_block_unique_txns,
             self.config.max_sending_block_bytes,
             self.config.max_sending_inline_txns,
             self.config.max_sending_inline_bytes,

consensus/src/liveness/proposal_generator.rs

@@ -164,6 +164,8 @@ pub struct ProposalGenerator {
     quorum_store_poll_time: Duration,
     // Max number of transactions to be added to a proposed block.
     max_block_txns: u64,
+    // Max number of unique transactions to be added to a proposed block.
+    max_block_unique_txns: u64,
     // Max number of bytes to be added to a proposed block.
     max_block_bytes: u64,
     // Max number of inline transactions to be added to a proposed block.
@@ -193,6 +195,7 @@ impl ProposalGenerator {
         time_service: Arc<dyn TimeService>,
         quorum_store_poll_time: Duration,
         max_block_txns: u64,
+        max_block_unique_txns: u64,
         max_block_bytes: u64,
         max_inline_txns: u64,
         max_inline_bytes: u64,
@@ -210,6 +213,7 @@ impl ProposalGenerator {
             time_service,
             quorum_store_poll_time,
             max_block_txns,
+            max_block_unique_txns,
             max_block_bytes,
             max_inline_txns,
             max_inline_bytes,
@@ -312,9 +316,14 @@ impl ProposalGenerator {
 
             let voting_power_ratio = proposer_election.get_voting_power_participation_ratio(round);
 
-            let (max_block_txns, max_block_bytes, max_txns_from_block_to_execute, proposal_delay) =
-                self.calculate_max_block_sizes(voting_power_ratio, timestamp, round)
-                    .await;
+            let (
+                max_block_unique_txns,
+                max_block_bytes,
+                max_txns_from_block_to_execute,
+                proposal_delay,
+            ) = self
+                .calculate_max_block_sizes(voting_power_ratio, timestamp, round)
+                .await;
 
             PROPOSER_DELAY_PROPOSAL.set(proposal_delay.as_secs_f64());
             if !proposal_delay.is_zero() {
@@ -349,7 +358,8 @@ impl ProposalGenerator {
                 .payload_client
                 .pull_payload(
                     self.quorum_store_poll_time.saturating_sub(proposal_delay),
-                    max_block_txns,
+                    self.max_block_txns,
+                    max_block_unique_txns,
                     max_block_bytes,
                     // TODO: Set max_inline_txns and max_inline_bytes correctly
                     self.max_inline_txns,

consensus/src/liveness/proposal_generator_test.rs

@@ -37,6 +37,7 @@ async fn test_proposal_generation_empty_tree() {
         Arc::new(SimulatedTimeService::new()),
         Duration::ZERO,
         1,
+        1,
         10,
         1,
         10,
@@ -81,6 +82,7 @@ async fn test_proposal_generation_parent() {
         Arc::new(SimulatedTimeService::new()),
         Duration::ZERO,
         1,
+        1,
         1000,
         1,
         500,
@@ -157,6 +159,7 @@ async fn test_old_proposal_generation() {
         Arc::new(SimulatedTimeService::new()),
         Duration::ZERO,
         1,
+        1,
         1000,
         1,
         500,
@@ -198,6 +201,7 @@ async fn test_correct_failed_authors() {
         Arc::new(SimulatedTimeService::new()),
         Duration::ZERO,
         1,
+        1,
         1000,
         1,
         500,

consensus/src/payload_client/mixed.rs

@@ -68,6 +68,7 @@ impl PayloadClient for MixedPayloadClient {
         &self,
         mut max_poll_time: Duration,
         mut max_items: u64,
+        mut max_unique_items: u64,
         mut max_bytes: u64,
         max_inline_items: u64,
         max_inline_bytes: u64,
@@ -101,6 +102,7 @@ impl PayloadClient for MixedPayloadClient {
         debug!("num_validator_txns={}", validator_txns.len());
         // Update constraints with validator txn pull results.
         max_items -= validator_txns.len() as u64;
+        max_unique_items -= validator_txns.len() as u64;
         max_bytes -= validator_txns
             .iter()
             .map(|txn| txn.size_in_bytes())
@@ -113,6 +115,7 @@ impl PayloadClient for MixedPayloadClient {
             .pull(
                 max_poll_time,
                 max_items,
+                max_unique_items,
                 max_bytes,
                 max_inline_items,
                 max_inline_bytes,
@@ -151,7 +154,8 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
     let (pulled_validator_txns, Payload::DirectMempool(pulled_user_txns)) = client
         .pull_payload(
             Duration::from_secs(1), // max_poll_time
-            99,                     // max_items
+            120,                    // max_items
+            99,                     // max_unique_items
             1048576,                // size limit: 1MB
             50,
             500000, // inline limit: 500KB
@@ -174,7 +178,8 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
     let (pulled_validator_txns, Payload::DirectMempool(pulled_user_txns)) = client
         .pull_payload(
             Duration::from_micros(500), // max_poll_time
-            99,                         // max_items
+            120,                        // max_items
+            99,                         // max_unique_items
             1048576,                    // size limit: 1MB
             50,
             500000, // inline limit: 500KB
@@ -197,7 +202,8 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
     let (pulled_validator_txns, Payload::DirectMempool(pulled_user_txns)) = client
         .pull_payload(
             Duration::from_secs(1), // max_poll_time
-            1,                      // max_items
+            2,                      // max_items
+            1,                      // max_unique_items
             1048576,                // size limit: 1MB
             0,
             0, // inline limit: 0
@@ -220,7 +226,8 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
     let (pulled_validator_txns, Payload::DirectMempool(pulled_user_txns)) = client
         .pull_payload(
             Duration::from_secs(1), // max_poll_time
-            99,                     // max_items
+            120,                    // max_items
+            99,                     // max_unique_items
             all_validator_txns[0].size_in_bytes() as u64,
             50,
             all_validator_txns[0].size_in_bytes() as u64,
@@ -261,7 +268,8 @@ async fn mixed_payload_client_should_respect_validator_txn_feature_flag() {
     let (pulled_validator_txns, Payload::DirectMempool(pulled_user_txns)) = client
         .pull_payload(
             Duration::from_millis(50), // max_poll_time
-            99,                        // max_items
+            120,                       // max_items
+            99,                        // max_unique_items
             1048576,                   // size limit: 1MB
             50,
             500000, // inline limit: 500KB

consensus/src/payload_client/mod.rs

@@ -18,6 +18,7 @@ pub trait PayloadClient: Send + Sync {
         &self,
         max_poll_time: Duration,
         max_items: u64,
+        max_unique_items: u64,
         max_bytes: u64,
         max_inline_items: u64,
         max_inline_bytes: u64,

consensus/src/payload_client/user/mod.rs

@@ -18,6 +18,7 @@ pub trait UserPayloadClient: Send + Sync {
         &self,
         max_poll_time: Duration,
         max_items: u64,
+        max_unique_items: u64,
         max_bytes: u64,
         max_inline_items: u64,
         max_inline_bytes: u64,
@@ -49,6 +50,7 @@ impl UserPayloadClient for DummyClient {
         &self,
         max_poll_time: Duration,
         mut max_items: u64,
+        mut max_unique_items: u64,
         mut max_bytes: u64,
         _max_inline_items: u64,
         _max_inline_bytes: u64,
@@ -63,6 +65,7 @@ impl UserPayloadClient for DummyClient {
         let mut txns = vec![];
         while timer.elapsed() < max_poll_time
             && max_items >= 1
+            && max_unique_items >= 1
             && max_bytes >= 1
             && nxt_txn_idx < self.txns.len()
         {
@@ -73,6 +76,7 @@ impl UserPayloadClient for DummyClient {
                 break;
             }
             max_items -= 1;
+            max_unique_items -= 1;
             max_bytes -= txn_size;
             nxt_txn_idx += 1;
             txns.push(txn);

consensus/src/payload_client/user/quorum_store_client.rs

@@ -46,6 +46,7 @@ impl QuorumStoreClient {
     async fn pull_internal(
         &self,
         max_items: u64,
+        max_unique_items: u64,
         max_bytes: u64,
         max_inline_items: u64,
         max_inline_bytes: u64,
@@ -55,6 +56,7 @@ impl QuorumStoreClient {
         let (callback, callback_rcv) = oneshot::channel();
         let req = GetPayloadCommand::GetPayloadRequest(
             max_items,
+            max_unique_items,
             max_bytes,
             max_inline_items,
             max_inline_bytes,
@@ -88,6 +90,7 @@ impl UserPayloadClient for QuorumStoreClient {
         &self,
         max_poll_time: Duration,
         max_items: u64,
+        max_unique_items: u64,
         max_bytes: u64,
         max_inline_items: u64,
         max_inline_bytes: u64,
@@ -117,6 +120,7 @@ impl UserPayloadClient for QuorumStoreClient {
             let payload = self
                 .pull_internal(
                     max_items,
+                    max_unique_items,
                     max_bytes,
                     max_inline_items,
                     max_inline_bytes,
@@ -138,6 +142,7 @@ impl UserPayloadClient for QuorumStoreClient {
             max_poll_time_ms = max_poll_time.as_millis() as u64,
             payload_len = payload.len(),
             max_items = max_items,
+            max_unique_items = max_unique_items,
             max_bytes = max_bytes,
             max_inline_items = max_inline_items,
             max_inline_bytes = max_inline_bytes,