[Quorum Store] Implementation of quorum store components

Cargo.toml

@@ -306,7 +306,7 @@ claims = "0.7"
 clap = { version = "3.2.17", features = ["derive", "env", "suggestions"] }
 clap_complete = "3.2.3"
 codespan-reporting = "0.11.1"
-console-subscriber = "0.1.6"
+console-subscriber = "0.1.8"
 const_format = "0.2.26"
 criterion = "0.3.5"
 criterion-cpu-time = "0.1.0"
@@ -439,6 +439,7 @@ tracing = "0.1.34"
 tracing-subscriber = "0.3.11"
 trybuild = "1.0.41"
 tokio = { version = "1.21.0", features = ["full"] }
+tokio-metrics = "0.1.0"
 tokio-retry = "0.3.0"
 tokio-stream = "0.1.8"
 tokio-test = "0.4.1"

config/src/config/consensus_config.rs

@@ -1,7 +1,7 @@
 // Copyright (c) Aptos
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::config::SafetyRulesConfig;
+use crate::config::{QuorumStoreConfig, SafetyRulesConfig};
 use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 
@@ -30,6 +30,7 @@ pub struct ConsensusConfig {
     // the period = (poll_count - 1) * 30ms
     pub quorum_store_poll_count: u64,
     pub intra_consensus_channel_buffer_size: usize,
+    pub quorum_store_configs: QuorumStoreConfig,
 
     // Used to decide if backoff is needed.
     // must match one of the CHAIN_HEALTH_WINDOW_SIZES values.
@@ -48,10 +49,10 @@ pub struct ChainHealthBackoffValues {
 impl Default for ConsensusConfig {
     fn default() -> ConsensusConfig {
         ConsensusConfig {
-            max_sending_block_txns: 2500,
+            max_sending_block_txns: 4000,
             // defaulting to under 0.5s to broadcast the proposal to 100 validators
             // over 1gbps link
-            max_sending_block_bytes: 600 * 1024, // 600 KB
+            max_sending_block_bytes: 1000 * 1024, // 1 MB
             max_receiving_block_txns: 10000,
             max_receiving_block_bytes: 3 * 1024 * 1024, // 3MB
             max_pruned_blocks_in_mem: 100,
@@ -64,11 +65,11 @@ impl Default for ConsensusConfig {
             round_timeout_backoff_max_exponent: 6,
             safety_rules: SafetyRulesConfig::default(),
             sync_only: false,
-            channel_size: 30, // hard-coded
-
+            channel_size: 100, // hard-coded
             quorum_store_pull_timeout_ms: 1000,
             quorum_store_poll_count: 10,
             intra_consensus_channel_buffer_size: 10,
+            quorum_store_configs: QuorumStoreConfig::default(),
 
             window_for_chain_health: 100,
             chain_health_backoff: vec![

config/src/config/mod.rs

@@ -18,6 +18,8 @@ use thiserror::Error;
 
 mod consensus_config;
 pub use consensus_config::*;
+mod quorum_store_config;
+pub use quorum_store_config::*;
 mod error;
 pub use error::*;
 mod execution_config;

config/src/config/quorum_store_config.rs

@@ -0,0 +1,61 @@
+// Copyright (c) Aptos
+// SPDX-License-Identifier: Apache-2.0
+
+use aptos_types::block_info::Round;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct QuorumStoreConfig {
+    pub channel_size: usize,
+    pub proof_timeout_ms: usize,
+    pub batch_request_num_peers: usize,
+    pub mempool_pulling_interval: usize,
+    pub end_batch_ms: u128,
+    pub max_batch_counts: usize,
+    pub max_batch_bytes: usize,
+    pub batch_request_timeout_ms: usize,
+    /// Used when setting up the expiration time for the batch initation.
+    pub batch_expiry_round_gap_when_init: Round,
+    /// Batches may have expiry set for batch_expiry_rounds_gap rounds after the
+    /// latest committed round, and it will not be cleared from storage for another
+    /// so other batch_expiry_grace_rounds rounds, so the peers on the network
+    /// can still fetch the data they fall behind (later, they would have to state-sync).
+    /// Used when checking the expiration time of the received batch against current logical time to prevent DDoS.
+    pub batch_expiry_round_gap_behind_latest_certified: Round,
+    pub batch_expiry_round_gap_beyond_latest_certified: Round,
+    pub batch_expiry_grace_rounds: Round,
+    pub memory_quota: usize,
+    pub db_quota: usize,
+    pub mempool_txn_pull_max_count: u64,
+    pub mempool_txn_pull_max_bytes: u64,
+    pub back_pressure_local_batch_num: usize,
+    pub num_workers_for_remote_fragments: usize,
+}
+
+impl Default for QuorumStoreConfig {
+    fn default() -> QuorumStoreConfig {
+        QuorumStoreConfig {
+            channel_size: 1000,
+            proof_timeout_ms: 10000,
+            batch_request_num_peers: 2,
+            mempool_pulling_interval: 100,
+            end_batch_ms: 500,
+            max_batch_counts: 300,
+            max_batch_bytes: 1000000,
+            batch_request_timeout_ms: 10000,
+            batch_expiry_round_gap_when_init: 100,
+            batch_expiry_round_gap_behind_latest_certified: 500,
+            batch_expiry_round_gap_beyond_latest_certified: 500,
+            batch_expiry_grace_rounds: 5,
+            memory_quota: 100000000,
+            db_quota: 10000000000,
+            mempool_txn_pull_max_count: 300,
+            mempool_txn_pull_max_bytes: 1000000,
+            // QS will be backpressured if the remaining local batches is more than this number
+            back_pressure_local_batch_num: 10,
+            // number of batch coordinators to handle QS Fragment messages, should be >= 1
+            num_workers_for_remote_fragments: 2,
+        }
+    }
+}

consensus/Cargo.toml

@@ -20,10 +20,12 @@ aptos-config = { workspace = true }
 aptos-consensus-notifications = { workspace = true }
 aptos-consensus-types = { workspace = true }
 aptos-crypto = { workspace = true }
+aptos-crypto-derive = { workspace = true }
 aptos-event-notifications = { workspace = true }
 aptos-executor = { workspace = true }
 aptos-executor-types = { workspace = true }
 aptos-fallible = { workspace = true }
+aptos-global-constants = { workspace = true }
 aptos-infallible = { workspace = true }
 aptos-logger = { workspace = true }
 aptos-mempool = { workspace = true }
@@ -43,6 +45,9 @@ async-trait = { workspace = true }
 bcs = { workspace = true }
 byteorder = { workspace = true }
 bytes = { workspace = true }
+chrono = { workspace = true }
+claims = { workspace = true }
+dashmap = { workspace = true }
 fail = { workspace = true }
 futures = { workspace = true }
 futures-channel = { workspace = true }
@@ -58,6 +63,7 @@ serde_bytes = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+tokio-metrics = { workspace = true }
 
 [dev-dependencies]
 aptos-config = { workspace = true, features = ["fuzzing"] }

consensus/consensus-types/Cargo.toml

@@ -26,6 +26,7 @@ futures = { workspace = true }
 itertools = { workspace = true }
 mirai-annotations = { workspace = true }
 proptest = { workspace = true, optional = true }
+rand = { workspace = true }
 rayon = { workspace = true }
 serde = { workspace = true }
 tokio = { workspace = true }

consensus/consensus-types/src/block.rs

@@ -101,6 +101,16 @@ impl Block {
         self.block_data.payload()
     }
 
+    pub fn payload_size(&self) -> usize {
+        match self.block_data.payload() {
+            None => 0,
+            Some(payload) => match payload {
+                Payload::InQuorumStore(pos) => pos.proofs.len(),
+                Payload::DirectMempool(txns) => txns.len(),
+            },
+        }
+    }
+
     pub fn quorum_cert(&self) -> &QuorumCert {
         self.block_data.quorum_cert()
     }

consensus/consensus-types/src/proof_of_store.rs

@@ -9,6 +9,7 @@ use aptos_types::{
     aggregate_signature::AggregateSignature, validator_signer::ValidatorSigner,
     validator_verifier::ValidatorVerifier, PeerId,
 };
+use rand::{seq::SliceRandom, thread_rng};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 
@@ -81,13 +82,35 @@ impl SignedDigest {
         })
     }
 
+    pub fn peer_id(&self) -> PeerId {
+        self.peer_id
+    }
+
     pub fn epoch(&self) -> u64 {
         self.epoch
     }
 
     pub fn verify(&self, validator: &ValidatorVerifier) -> anyhow::Result<()> {
         Ok(validator.verify(self.peer_id, &self.info, &self.signature)?)
     }
+
+    pub fn info(&self) -> &SignedDigestInfo {
+        &self.info
+    }
+
+    pub fn signature(self) -> bls12381::Signature {
+        self.signature
+    }
+
+    pub fn digest(&self) -> HashValue {
+        self.info.digest
+    }
+}
+
+#[derive(Debug, PartialEq)]
+pub enum SignedDigestError {
+    WrongInfo,
+    DuplicatedSignature,
 }
 
 #[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq)]
@@ -122,6 +145,14 @@ impl ProofOfStore {
             .context("Failed to verify ProofOfStore")
     }
 
+    pub fn shuffled_signers(&self, validator: &ValidatorVerifier) -> Vec<PeerId> {
+        let mut ret: Vec<PeerId> = self
+            .multi_signature
+            .get_voter_addresses(&validator.validator_addresses());
+        ret.shuffle(&mut thread_rng());
+        ret
+    }
+
     pub fn epoch(&self) -> u64 {
         self.info.expiration.epoch
     }

consensus/consensus-types/src/request_response.rs

@@ -10,8 +10,7 @@ use aptos_crypto::HashValue;
 use futures::channel::oneshot;
 use std::{fmt, fmt::Formatter};
 
-/// Message sent from Consensus to QuorumStore.
-pub enum PayloadRequest {
+pub enum BlockProposalCommand {
     /// Request to pull block to submit to consensus.
     GetBlockRequest(
         Round,
@@ -24,21 +23,30 @@ pub enum PayloadRequest {
         // callback to respond to
         oneshot::Sender<Result<ConsensusResponse>>,
     ),
-    /// Request to clean quorum store at commit logical time
-    CleanRequest(LogicalTime, Vec<HashValue>),
 }
 
-impl fmt::Display for PayloadRequest {
+impl fmt::Display for BlockProposalCommand {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match self {
-            PayloadRequest::GetBlockRequest(round, max_txns, max_bytes, excluded, _) => {
+            BlockProposalCommand::GetBlockRequest(round, max_txns, max_bytes, excluded, _) => {
                 write!(
                     f,
                     "GetBlockRequest [round: {}, max_txns: {}, max_bytes: {} excluded: {}]",
                     round, max_txns, max_bytes, excluded
                 )
             },
-            PayloadRequest::CleanRequest(logical_time, digests) => {
+        }
+    }
+}
+
+pub enum CleanCommand {
+    CleanRequest(LogicalTime, Vec<HashValue>),
+}
+
+impl fmt::Display for CleanCommand {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            CleanCommand::CleanRequest(logical_time, digests) => {
                 write!(
                     f,
                     "CleanRequest [epoch: {}, round: {}, digests: {:?}]",
@@ -51,6 +59,7 @@ impl fmt::Display for PayloadRequest {
     }
 }
 
+#[derive(Debug)]
 pub enum ConsensusResponse {
     GetBlockResponse(Payload),
 }