Merge pull request #142 from vacp2p/schouhy/add-sequencer-tps-test

Add tps integration test

integration_tests/Cargo.toml

@@ -39,3 +39,6 @@ path = "../common"
 [dependencies.nssa]
 path = "../nssa"
 features = ["no_docker"]
+
+[dependencies.key_protocol]
+path = "../key_protocol"

integration_tests/configs/debug/sequencer/sequencer_config.json

@@ -4,6 +4,7 @@
     "genesis_id": 1,
     "is_genesis_random": true,
     "max_num_tx_in_block": 20,
+    "mempool_max_size": 10000,
     "block_create_timeout_millis": 10000,
     "port": 3040,
     "initial_accounts": [

integration_tests/src/lib.rs

@@ -16,13 +16,15 @@ use sequencer_runner::startup_sequencer;
 use tempfile::TempDir;
 use tokio::task::JoinHandle;
 
-use crate::test_suite_map::prepare_function_map;
+use crate::test_suite_map::{prepare_function_map, tps_test};
 
 #[macro_use]
 extern crate proc_macro_test_attribute;
 
 pub mod test_suite_map;
 
+mod tps_test_utils;
+
 #[derive(Parser, Debug)]
 #[clap(version)]
 struct Args {
@@ -116,9 +118,12 @@ pub async fn main_tests_runner() -> Result<()> {
 
     match test_name.as_str() {
         "all" => {
+            // Tests that use default config
             for (_, fn_pointer) in function_map {
                 fn_pointer(home_dir.clone()).await;
             }
+            // Run TPS test with its own specific config
+            tps_test().await;
         }
         _ => {
             let fn_pointer = function_map.get(&test_name).expect("Unknown test name");

integration_tests/src/test_suite_map.rs

@@ -1,9 +1,19 @@
-use std::{collections::HashMap, path::PathBuf, pin::Pin, time::Duration};
+use anyhow::Result;
+use std::{
+    collections::HashMap,
+    path::PathBuf,
+    pin::Pin,
+    time::{Duration, Instant},
+};
 
+use actix_web::dev::ServerHandle;
 use common::{PINATA_BASE58, sequencer_client::SequencerClient};
 use log::info;
 use nssa::{Address, ProgramDeploymentTransaction, program::Program};
 use nssa_core::{NullifierPublicKey, encryption::shared_key_derivation::Secp256k1Point};
+use sequencer_runner::startup_sequencer;
+use tempfile::TempDir;
+use tokio::task::JoinHandle;
 use wallet::{
     Command, SubcommandReturnValue, WalletCore,
     cli::{
@@ -20,7 +30,8 @@ use crate::{
     ACC_RECEIVER, ACC_RECEIVER_PRIVATE, ACC_SENDER, ACC_SENDER_PRIVATE,
     NSSA_PROGRAM_FOR_TEST_DATA_CHANGER, TIME_TO_WAIT_FOR_BLOCK_SECONDS,
     fetch_privacy_preserving_tx, make_private_account_input_from_str,
-    make_public_account_input_from_str,
+    make_public_account_input_from_str, replace_home_dir_with_temp_dir_in_configs,
+    tps_test_utils::TpsTestManager,
 };
 use crate::{post_test, pre_test, verify_commitment_is_in_state};
 
@@ -1582,3 +1593,83 @@ pub fn prepare_function_map() -> HashMap<String, TestFunction> {
 
     function_map
 }
+
+#[allow(clippy::type_complexity)]
+async fn pre_tps_test(
+    test: &TpsTestManager,
+) -> Result<(ServerHandle, JoinHandle<Result<()>>, TempDir)> {
+    info!("Generating tps test config");
+    let mut sequencer_config = test.generate_tps_test_config();
+    info!("Done");
+
+    let temp_dir_sequencer = replace_home_dir_with_temp_dir_in_configs(&mut sequencer_config);
+
+    let (seq_http_server_handle, sequencer_loop_handle) =
+        startup_sequencer(sequencer_config).await?;
+
+    Ok((
+        seq_http_server_handle,
+        sequencer_loop_handle,
+        temp_dir_sequencer,
+    ))
+}
+
+pub async fn tps_test() {
+    let num_transactions = 300 * 5;
+    let target_tps = 12;
+    let tps_test = TpsTestManager::new(target_tps, num_transactions);
+
+    let target_time = tps_test.target_time();
+    info!("Target time: {:?} seconds", target_time.as_secs());
+    let res = pre_tps_test(&tps_test).await.unwrap();
+
+    let wallet_config = fetch_config().await.unwrap();
+    let seq_client = SequencerClient::new(wallet_config.sequencer_addr.clone()).unwrap();
+
+    info!("TPS test begin");
+    let txs = tps_test.build_public_txs();
+    let now = Instant::now();
+
+    let mut tx_hashes = vec![];
+    for (i, tx) in txs.into_iter().enumerate() {
+        let tx_hash = seq_client.send_tx_public(tx).await.unwrap().tx_hash;
+        info!("Sent tx {i}");
+        tx_hashes.push(tx_hash);
+    }
+
+    for (i, tx_hash) in tx_hashes.iter().enumerate() {
+        loop {
+            if now.elapsed().as_millis() > target_time.as_millis() {
+                panic!("TPS test failed by timout");
+            }
+
+            let tx_obj = seq_client
+                .get_transaction_by_hash(tx_hash.clone())
+                .await
+                .inspect_err(|err| {
+                    log::warn!(
+                        "Failed to get transaction by hash {tx_hash:#?} with error: {err:#?}"
+                    )
+                });
+
+            if let Ok(tx_obj) = tx_obj
+                && tx_obj.transaction.is_some()
+            {
+                info!("Found tx {i} with hash {tx_hash}");
+                break;
+            }
+        }
+    }
+    let time_elapsed = now.elapsed().as_secs();
+
+    info!("TPS test finished successfully");
+    info!("Target TPS: {}", target_tps);
+    info!(
+        "Processed {} transactions in {}s",
+        tx_hashes.len(),
+        time_elapsed
+    );
+    info!("Target time: {:?}s", target_time.as_secs());
+
+    post_test(res).await;
+}

integration_tests/src/tps_test_utils.rs

@@ -0,0 +1,187 @@
+use std::time::Duration;
+
+use key_protocol::key_management::ephemeral_key_holder::EphemeralKeyHolder;
+use nssa::{
+    Account, AccountId, Address, PrivacyPreservingTransaction, PrivateKey, PublicKey,
+    PublicTransaction,
+    privacy_preserving_transaction::{self as pptx, circuit},
+    program::Program,
+    public_transaction as putx,
+};
+use nssa_core::{
+    MembershipProof, NullifierPublicKey, account::AccountWithMetadata,
+    encryption::IncomingViewingPublicKey,
+};
+use sequencer_core::config::{AccountInitialData, CommitmentsInitialData, SequencerConfig};
+
+pub(crate) struct TpsTestManager {
+    public_keypairs: Vec<(PrivateKey, Address)>,
+    target_tps: u64,
+}
+
+impl TpsTestManager {
+    /// Generates public account keypairs. These are used to populate the config and to generate valid
+    /// public transactions for the tps test.
+    pub(crate) fn new(target_tps: u64, number_transactions: usize) -> Self {
+        let public_keypairs = (1..(number_transactions + 2))
+            .map(|i| {
+                let mut private_key_bytes = [0u8; 32];
+                private_key_bytes[..8].copy_from_slice(&i.to_le_bytes());
+                let private_key = PrivateKey::try_new(private_key_bytes).unwrap();
+                let public_key = PublicKey::new_from_private_key(&private_key);
+                let address = Address::from(&public_key);
+                (private_key, address)
+            })
+            .collect::<Vec<_>>();
+        Self {
+            public_keypairs,
+            target_tps,
+        }
+    }
+
+    pub(crate) fn target_time(&self) -> Duration {
+        let number_transactions = (self.public_keypairs.len() - 1) as u64;
+        Duration::from_secs_f64(number_transactions as f64 / self.target_tps as f64)
+    }
+
+    ///
+    /// Build a batch of public transactions to submit to the node.
+    pub fn build_public_txs(&self) -> Vec<PublicTransaction> {
+        // Create valid public transactions
+        let program = Program::authenticated_transfer_program();
+        let public_txs: Vec<PublicTransaction> = self
+            .public_keypairs
+            .windows(2)
+            .map(|pair| {
+                let amount: u128 = 1;
+                let message = putx::Message::try_new(
+                    program.id(),
+                    [pair[0].1, pair[1].1].to_vec(),
+                    [0u128].to_vec(),
+                    amount,
+                )
+                .unwrap();
+                let witness_set =
+                    nssa::public_transaction::WitnessSet::for_message(&message, &[&pair[0].0]);
+                PublicTransaction::new(message, witness_set)
+            })
+            .collect();
+
+        public_txs
+    }
+
+    /// Generates a sequencer configuration with initial balance in a number of public accounts.
+    /// The transactions generated with the function `build_public_txs` will be valid in a node started
+    /// with the config from this method.
+    pub(crate) fn generate_tps_test_config(&self) -> SequencerConfig {
+        // Create public public keypairs
+        let initial_public_accounts = self
+            .public_keypairs
+            .iter()
+            .map(|(_, addr)| AccountInitialData {
+                addr: addr.to_string(),
+                balance: 10,
+            })
+            .collect();
+
+        // Generate an initial commitment to be used with the privacy preserving transaction
+        // created with the `build_privacy_transaction` function.
+        let sender_nsk = [1; 32];
+        let sender_npk = NullifierPublicKey::from(&sender_nsk);
+        let account = Account {
+            balance: 100,
+            nonce: 0xdeadbeef,
+            program_owner: Program::authenticated_transfer_program().id(),
+            data: vec![],
+        };
+        let initial_commitment = CommitmentsInitialData {
+            npk: sender_npk,
+            account,
+        };
+
+        SequencerConfig {
+            home: ".".into(),
+            override_rust_log: None,
+            genesis_id: 1,
+            is_genesis_random: true,
+            max_num_tx_in_block: 300,
+            mempool_max_size: 10000,
+            block_create_timeout_millis: 12000,
+            port: 3040,
+            initial_accounts: initial_public_accounts,
+            initial_commitments: vec![initial_commitment],
+            signing_key: [37; 32],
+        }
+    }
+}
+
+/// Builds a single privacy transaction to use in stress tests. This involves generating a proof so
+/// it may take a while to run. In normal execution of the node this transaction will be accepted
+/// only once. Disabling the node's nullifier uniqueness check allows to submit this transaction
+/// multiple times with the purpose of testing the node's processing performance.
+#[allow(unused)]
+fn build_privacy_transaction() -> PrivacyPreservingTransaction {
+    let program = Program::authenticated_transfer_program();
+    let sender_nsk = [1; 32];
+    let sender_isk = [99; 32];
+    let sender_ipk = IncomingViewingPublicKey::from_scalar(sender_isk);
+    let sender_npk = NullifierPublicKey::from(&sender_nsk);
+    let sender_pre = AccountWithMetadata::new(
+        Account {
+            balance: 100,
+            nonce: 0xdeadbeef,
+            program_owner: program.id(),
+            data: vec![],
+        },
+        true,
+        AccountId::from(&sender_npk),
+    );
+    let recipient_nsk = [2; 32];
+    let recipient_isk = [99; 32];
+    let recipient_ipk = IncomingViewingPublicKey::from_scalar(recipient_isk);
+    let recipient_npk = NullifierPublicKey::from(&recipient_nsk);
+    let recipient_pre =
+        AccountWithMetadata::new(Account::default(), false, AccountId::from(&recipient_npk));
+
+    let eph_holder_from = EphemeralKeyHolder::new(&sender_npk);
+    let sender_ss = eph_holder_from.calculate_shared_secret_sender(&sender_ipk);
+    let sender_epk = eph_holder_from.generate_ephemeral_public_key();
+
+    let eph_holder_to = EphemeralKeyHolder::new(&recipient_npk);
+    let recipient_ss = eph_holder_to.calculate_shared_secret_sender(&recipient_ipk);
+    let recipient_epk = eph_holder_from.generate_ephemeral_public_key();
+
+    let balance_to_move: u128 = 1;
+    let proof: MembershipProof = (
+        1,
+        vec![[
+            170, 10, 217, 228, 20, 35, 189, 177, 238, 235, 97, 129, 132, 89, 96, 247, 86, 91, 222,
+            214, 38, 194, 216, 67, 56, 251, 208, 226, 0, 117, 149, 39,
+        ]],
+    );
+    let (output, proof) = circuit::execute_and_prove(
+        &[sender_pre, recipient_pre],
+        &Program::serialize_instruction(balance_to_move).unwrap(),
+        &[1, 2],
+        &[0xdeadbeef1, 0xdeadbeef2],
+        &[
+            (sender_npk.clone(), sender_ss),
+            (recipient_npk.clone(), recipient_ss),
+        ],
+        &[(sender_nsk, proof)],
+        &program,
+    )
+    .unwrap();
+    let message = pptx::message::Message::try_from_circuit_output(
+        vec![],
+        vec![],
+        vec![
+            (sender_npk, sender_ipk, sender_epk),
+            (recipient_npk, recipient_ipk, recipient_epk),
+        ],
+        output,
+    )
+    .unwrap();
+    let witness_set = pptx::witness_set::WitnessSet::for_message(&message, proof, &[]);
+    pptx::PrivacyPreservingTransaction::new(message, witness_set)
+}

sequencer_core/src/config.rs

@@ -28,6 +28,8 @@ pub struct SequencerConfig {
     pub is_genesis_random: bool,
     ///Maximum number of transactions in block
     pub max_num_tx_in_block: usize,
+    ///Mempool maximum size
+    pub mempool_max_size: usize,
     ///Interval in which blocks produced
     pub block_create_timeout_millis: u64,
     ///Port to listen

sequencer_core/src/lib.rs

@@ -1,4 +1,4 @@
-use std::fmt::Display;
+use std::{fmt::Display, time::Instant};
 
 use anyhow::Result;
 use common::{
@@ -103,7 +103,7 @@ impl SequencerCore {
         })?;
 
         let mempool_size = self.mempool.len();
-        if mempool_size >= self.sequencer_config.max_num_tx_in_block {
+        if mempool_size >= self.sequencer_config.mempool_max_size {
             return Err(TransactionMalformationError::MempoolFullForRound);
         }
 
@@ -146,6 +146,7 @@ impl SequencerCore {
 
     ///Produces new block from transactions in mempool
     pub fn produce_new_block_with_mempool_transactions(&mut self) -> Result<u64> {
+        let now = Instant::now();
         let new_block_height = self.chain_height + 1;
 
         let mut num_valid_transactions_in_block = 0;
@@ -175,6 +176,8 @@ impl SequencerCore {
 
         let curr_time = chrono::Utc::now().timestamp_millis() as u64;
 
+        let num_txs_in_block = valid_transactions.len();
+
         let hashable_data = HashableBlockData {
             block_id: new_block_height,
             transactions: valid_transactions,
@@ -188,6 +191,12 @@ impl SequencerCore {
 
         self.chain_height = new_block_height;
 
+        log::info!(
+            "Created block with {} transactions in {} seconds",
+            num_txs_in_block,
+            now.elapsed().as_secs()
+        );
+
         Ok(self.chain_height)
     }
 }
@@ -219,6 +228,7 @@ mod tests {
             genesis_id: 1,
             is_genesis_random: false,
             max_num_tx_in_block: 10,
+            mempool_max_size: 10000,
             block_create_timeout_millis: 1000,
             port: 8080,
             initial_accounts,
@@ -548,7 +558,7 @@ mod tests {
     #[test]
     fn test_push_tx_into_mempool_fails_mempool_full() {
         let config = SequencerConfig {
-            max_num_tx_in_block: 1,
+            mempool_max_size: 1,
             ..setup_sequencer_config()
         };
         let mut sequencer = SequencerCore::start_from_config(config);

sequencer_rpc/src/process.rs

@@ -361,6 +361,7 @@ mod tests {
             genesis_id: 1,
             is_genesis_random: false,
             max_num_tx_in_block: 10,
+            mempool_max_size: 1000,
             block_create_timeout_millis: 1000,
             port: 8080,
             initial_accounts,

sequencer_runner/configs/debug/sequencer_config.json

@@ -4,6 +4,7 @@
     "genesis_id": 1,
     "is_genesis_random": true,
     "max_num_tx_in_block": 20,
+    "mempool_max_size": 1000,
     "block_create_timeout_millis": 10000,
     "port": 3040,
     "initial_accounts": [