[Quorum Store] Implementation of quorum store components

Cargo.toml

@@ -308,7 +308,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"
@@ -445,6 +445,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"

api/src/tests/resource_groups.rs

@@ -6,7 +6,7 @@ use aptos_api_test_context::{current_function_name, TestContext};
 use aptos_cached_packages::aptos_stdlib;
 use aptos_framework::BuiltPackage;
 use aptos_sdk::types::LocalAccount;
-use aptos_types::account_address::AccountAddress;
+use aptos_types::{account_address::AccountAddress, transaction::TransactionPayload};
 use serde_json::{json, Value};
 use std::path::PathBuf;
 
@@ -37,13 +37,21 @@ async fn test_read_resoure_group() {
         ("resource_groups_secondary".to_string(), admin1.address()),
     ];
 
-    let path = PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
-        .join("../aptos-move/move-examples/resource_groups/primary");
-    publish_package(&mut context, &mut admin0, path, named_addresses.clone()).await;
-
-    let path = PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
-        .join("../aptos-move/move-examples/resource_groups/secondary");
-    publish_package(&mut context, &mut admin1, path, named_addresses.clone()).await;
+    let named_addresses_clone = named_addresses.clone();
+    let txn = futures::executor::block_on(async move {
+        let path = PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
+            .join("../aptos-move/move-examples/resource_groups/primary");
+        build_package(path, named_addresses_clone)
+    });
+    publish_package(&mut context, &mut admin0, txn).await;
+
+    let named_addresses_clone = named_addresses.clone();
+    let txn = futures::executor::block_on(async move {
+        let path = PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
+            .join("../aptos-move/move-examples/resource_groups/secondary");
+        build_package(path, named_addresses_clone)
+    });
+    publish_package(&mut context, &mut admin1, txn).await;
 
     // Read default data
     let primary = format!("0x{}::{}::{}", admin0.address(), "primary", "Primary");
@@ -150,23 +158,28 @@ async fn read_resource(
     context.get(&request).await
 }
 
-async fn publish_package(
-    context: &mut TestContext,
-    publisher: &mut LocalAccount,
+fn build_package(
     path: PathBuf,
     named_addresses: Vec<(String, AccountAddress)>,
-) {
+) -> TransactionPayload {
     let mut build_options = aptos_framework::BuildOptions::default();
     let _ = named_addresses
         .into_iter()
         .map(|(name, address)| build_options.named_addresses.insert(name, address))
         .collect::<Vec<_>>();
 
-    let package = BuiltPackage::build(path.to_owned(), build_options).unwrap();
+    let package = BuiltPackage::build(path, build_options).unwrap();
     let code = package.extract_code();
     let metadata = package.extract_metadata().unwrap();
 
-    let payload = aptos_stdlib::code_publish_package_txn(bcs::to_bytes(&metadata).unwrap(), code);
+    aptos_stdlib::code_publish_package_txn(bcs::to_bytes(&metadata).unwrap(), code)
+}
+
+async fn publish_package(
+    context: &mut TestContext,
+    publisher: &mut LocalAccount,
+    payload: TransactionPayload,
+) {
     let txn =
         publisher.sign_with_transaction_builder(context.transaction_factory().payload(payload));
     let bcs_txn = bcs::to_bytes(&txn).unwrap();

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.
@@ -65,10 +66,10 @@ impl Default for ConsensusConfig {
             safety_rules: SafetyRulesConfig::default(),
             sync_only: false,
             channel_size: 30, // 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.get_ordered_account_addresses());
+        ret.shuffle(&mut thread_rng());
+        ret
+    }
+
     pub fn epoch(&self) -> u64 {
         self.info.expiration.epoch
     }