diff --git a/Cargo.lock b/Cargo.lock index 16861330..ca8c2a0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11019,13 +11019,11 @@ version = "0.1.0" dependencies = [ "bip39", "cbindgen", - "common", "key_protocol", "lee", "lee_core", "programs", "risc0-zkvm", - "sequencer_service_rpc", "serde_json", "tokio", "vault_core", diff --git a/Justfile b/Justfile index cf7c833e..277403c7 100644 --- a/Justfile +++ b/Justfile @@ -131,5 +131,6 @@ clean: rm -rf lez/sequencer/service/rocksdb rm -rf lez/indexer/service/rocksdb* rm -rf lez/wallet/configs/debug/storage.json + rm -rf lez/wallet/configs/debug/statistics.json rm -rf rocksdb* cd bedrock && docker compose down -v diff --git a/artifacts/lee/privacy_preserving_circuit/privacy_preserving_circuit.bin b/artifacts/lee/privacy_preserving_circuit/privacy_preserving_circuit.bin index f54520a8..4eafd160 100644 Binary files a/artifacts/lee/privacy_preserving_circuit/privacy_preserving_circuit.bin and b/artifacts/lee/privacy_preserving_circuit/privacy_preserving_circuit.bin differ diff --git a/examples/program_deployment/src/bin/run_hello_world.rs b/examples/program_deployment/src/bin/run_hello_world.rs index 3f2223a1..a43bd7f1 100644 --- a/examples/program_deployment/src/bin/run_hello_world.rs +++ b/examples/program_deployment/src/bin/run_hello_world.rs @@ -27,7 +27,7 @@ use wallet::WalletCore; #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the program binary @@ -59,7 +59,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/examples/program_deployment/src/bin/run_hello_world_private.rs b/examples/program_deployment/src/bin/run_hello_world_private.rs index f6202433..64fc39c2 100644 --- a/examples/program_deployment/src/bin/run_hello_world_private.rs +++ b/examples/program_deployment/src/bin/run_hello_world_private.rs @@ -23,7 +23,7 @@ use wallet::{AccountIdentity, WalletCore}; #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the program binary diff --git a/examples/program_deployment/src/bin/run_hello_world_through_tail_call.rs b/examples/program_deployment/src/bin/run_hello_world_through_tail_call.rs index 6ebba70f..18620220 100644 --- a/examples/program_deployment/src/bin/run_hello_world_through_tail_call.rs +++ b/examples/program_deployment/src/bin/run_hello_world_through_tail_call.rs @@ -27,7 +27,7 @@ use wallet::WalletCore; #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the program binary @@ -55,7 +55,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/examples/program_deployment/src/bin/run_hello_world_through_tail_call_private.rs b/examples/program_deployment/src/bin/run_hello_world_through_tail_call_private.rs index d35e6521..c1282dae 100644 --- a/examples/program_deployment/src/bin/run_hello_world_through_tail_call_private.rs +++ b/examples/program_deployment/src/bin/run_hello_world_through_tail_call_private.rs @@ -26,7 +26,7 @@ use wallet::{AccountIdentity, WalletCore}; #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the simple_tail_call program binary diff --git a/examples/program_deployment/src/bin/run_hello_world_with_authorization.rs b/examples/program_deployment/src/bin/run_hello_world_with_authorization.rs index 0d257db2..563f0877 100644 --- a/examples/program_deployment/src/bin/run_hello_world_with_authorization.rs +++ b/examples/program_deployment/src/bin/run_hello_world_with_authorization.rs @@ -29,7 +29,7 @@ use wallet::WalletCore; #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the program binary @@ -52,7 +52,8 @@ async fn main() { .storage() .key_chain() .pub_account_signing_key(account_id) - .expect("Input account should be a self owned public account"); + .expect("Input account should be a self owned public account") + .clone(); // Define the desired greeting in ASCII let greeting: Vec = vec![72, 111, 108, 97, 32, 109, 117, 110, 100, 111, 33]; @@ -60,10 +61,10 @@ async fn main() { // Construct the public transaction // Query the current nonce from the node let nonces = wallet_core - .get_accounts_nonces(vec![account_id]) + .get_accounts_nonces(&[account_id]) .await .expect("Node should be reachable to query account data"); - let signing_keys = [signing_key]; + let signing_keys = [&signing_key]; let message = Message::try_new(program.id(), vec![account_id], nonces, greeting).unwrap(); // Pass the signing key to sign the message. This will be used by the node // to flag the pre_state as `is_authorized` when executing the program @@ -72,7 +73,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/examples/program_deployment/src/bin/run_hello_world_with_authorization_through_tail_call_with_pda.rs b/examples/program_deployment/src/bin/run_hello_world_with_authorization_through_tail_call_with_pda.rs index 70688cfd..9139189e 100644 --- a/examples/program_deployment/src/bin/run_hello_world_with_authorization_through_tail_call_with_pda.rs +++ b/examples/program_deployment/src/bin/run_hello_world_with_authorization_through_tail_call_with_pda.rs @@ -35,7 +35,7 @@ const PDA_SEED: PdaSeed = PdaSeed::new([37; 32]); #[tokio::main] async fn main() { // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); // Parse arguments // First argument is the path to the program binary @@ -57,7 +57,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/examples/program_deployment/src/bin/run_hello_world_with_move_function.rs b/examples/program_deployment/src/bin/run_hello_world_with_move_function.rs index a77fe2e6..b99f41f4 100644 --- a/examples/program_deployment/src/bin/run_hello_world_with_move_function.rs +++ b/examples/program_deployment/src/bin/run_hello_world_with_move_function.rs @@ -66,7 +66,7 @@ async fn main() { let program = Program::new(bytecode.into()).unwrap(); // Initialize wallet - let wallet_core = WalletCore::from_env().unwrap(); + let wallet_core = WalletCore::from_env().await.unwrap(); match cli.command { Command::WritePublic { @@ -88,7 +88,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); @@ -127,7 +127,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/integration_tests/tests/auth_transfer/private.rs b/integration_tests/tests/auth_transfer/private.rs index ddea3ab8..655ca1a9 100644 --- a/integration_tests/tests/auth_transfer/private.rs +++ b/integration_tests/tests/auth_transfer/private.rs @@ -248,11 +248,7 @@ async fn shielded_transfer_to_foreign_account() -> Result<()> { let acc_1_balance = account_balance(&ctx, from).await?; assert!( - verify_commitment_is_in_state( - tx.message.new_commitments[0].clone(), - ctx.sequencer_client() - ) - .await + verify_commitment_is_in_state(tx.message.new_commitments[0], ctx.sequencer_client()).await ); assert_eq!(acc_1_balance, 9900); diff --git a/integration_tests/tests/bridge.rs b/integration_tests/tests/bridge.rs index 9c241ebb..8da35f4d 100644 --- a/integration_tests/tests/bridge.rs +++ b/integration_tests/tests/bridge.rs @@ -441,7 +441,7 @@ async fn bedrock_deposit_claim_and_withdraw_round_trip_succeeds() -> anyhow::Res // Now claim funds from vault back to recipient let nonces = ctx .wallet() - .get_accounts_nonces(vec![recipient_id]) + .get_accounts_nonces(&[recipient_id]) .await .context("Failed to get nonce for vault claim")?; diff --git a/integration_tests/tests/private_pda.rs b/integration_tests/tests/private_pda.rs index 0c67049f..0b187283 100644 --- a/integration_tests/tests/private_pda.rs +++ b/integration_tests/tests/private_pda.rs @@ -91,7 +91,7 @@ async fn fund_private_pda( let tx = PrivacyPreservingTransaction::new(message, witness_set); wallet - .sequencer_client + .leader_owned() .send_transaction(LeeTransaction::PrivacyPreserving(tx)) .await .map_err(|e| anyhow::anyhow!("send transaction failed: {e}"))?; @@ -180,7 +180,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { info!("Sending to alice_pda_0 (identifier=0)"); fund_private_pda( - ctx.wallet(), + ctx.wallet_mut(), sender_0, alice_npk, alice_vpk.clone(), @@ -194,7 +194,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { info!("Sending to alice_pda_1 (identifier=1)"); fund_private_pda( - ctx.wallet(), + ctx.wallet_mut(), sender_1, alice_npk, alice_vpk.clone(), @@ -231,7 +231,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { .get_private_account_commitment(alice_pda_0_id) .context("commitment for alice_pda_0 missing")?; assert!( - verify_commitment_is_in_state(commitment_0.clone(), ctx.sequencer_client()).await, + verify_commitment_is_in_state(commitment_0, ctx.sequencer_client()).await, "alice_pda_0 commitment not in state after receive" ); @@ -240,7 +240,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { .get_private_account_commitment(alice_pda_1_id) .context("commitment for alice_pda_1 missing")?; assert!( - verify_commitment_is_in_state(commitment_1.clone(), ctx.sequencer_client()).await, + verify_commitment_is_in_state(commitment_1, ctx.sequencer_client()).await, "alice_pda_1 commitment not in state after receive" ); assert_ne!( @@ -262,7 +262,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { info!("Alice spending from alice_pda_0"); spend_private_pda( - ctx.wallet(), + ctx.wallet_mut(), alice_pda_0_id, recipient_npk_0, recipient_vpk_0, @@ -275,7 +275,7 @@ async fn private_pda_family_members_receive_and_spend() -> Result<()> { info!("Alice spending from alice_pda_1"); spend_private_pda( - ctx.wallet(), + ctx.wallet_mut(), alice_pda_1_id, recipient_npk_1, recipient_vpk_1, diff --git a/integration_tests/tests/program_deployment.rs b/integration_tests/tests/program_deployment.rs index c92daeb3..3c620168 100644 --- a/integration_tests/tests/program_deployment.rs +++ b/integration_tests/tests/program_deployment.rs @@ -31,7 +31,7 @@ async fn deploy_and_execute_program() -> Result<()> { let account_id = new_account(&mut ctx, false, None).await?; - let nonces = ctx.wallet().get_accounts_nonces(vec![account_id]).await?; + let nonces = ctx.wallet_mut().get_accounts_nonces(&[account_id]).await?; let private_key = ctx .wallet() .get_account_public_signing_key(account_id) diff --git a/integration_tests/tests/wallet_ffi.rs b/integration_tests/tests/wallet_ffi.rs index 321c874d..a36eb069 100644 --- a/integration_tests/tests/wallet_ffi.rs +++ b/integration_tests/tests/wallet_ffi.rs @@ -28,7 +28,6 @@ use lee::{ }; use lee_core::program::DEFAULT_PROGRAM_ID; use log::info; -use tempfile::tempdir; use wallet::{account::HumanReadableAccount, program_facades::vault::Vault}; use wallet_ffi::{ FfiAccount, FfiAccountIdWithPrivacy, FfiAccountIdentity, FfiAccountList, FfiBytes32, @@ -43,12 +42,14 @@ unsafe extern "C" { fn wallet_ffi_create_new( config_path: *const c_char, storage_path: *const c_char, + metrics_path: *const c_char, password: *const c_char, ) -> FfiCreateWalletOutput; fn wallet_ffi_open( config_path: *const c_char, storage_path: *const c_char, + metrics_path: *const c_char, ) -> *mut WalletHandle; fn wallet_ffi_destroy(handle: *mut WalletHandle); @@ -291,6 +292,7 @@ fn new_wallet_ffi_with_test_context_config( ) -> Result { let config_path = home.join("wallet_config.json"); let storage_path = home.join("storage.json"); + let metrics_path = home.join("metrics.json"); let mut config = ctx.ctx().wallet().config().to_owned(); if let Some(config_overrides) = ctx.ctx().wallet().config_overrides().clone() { config.apply_overrides(config_overrides); @@ -307,12 +309,14 @@ fn new_wallet_ffi_with_test_context_config( let config_path = CString::new(config_path.to_str().unwrap())?; let storage_path = CString::new(storage_path.to_str().unwrap())?; + let metrics_path = CString::new(metrics_path.to_str().unwrap())?; let password = CString::new(ctx.ctx().wallet_password())?; let create_wallet_result = unsafe { wallet_ffi_create_new( config_path.as_ptr(), storage_path.as_ptr(), + metrics_path.as_ptr(), password.as_ptr(), ) }; @@ -366,47 +370,37 @@ fn new_wallet_ffi_with_test_context_config( Ok(create_wallet_result) } -fn new_wallet_ffi_with_default_config(password: &str) -> Result { - let tempdir = tempdir()?; - let config_path = tempdir.path().join("wallet_config.json"); - let storage_path = tempdir.path().join("storage.json"); - let config_path_c = CString::new(config_path.to_str().unwrap())?; - let storage_path_c = CString::new(storage_path.to_str().unwrap())?; - let password = CString::new(password)?; - - let create_wallet_result = unsafe { - wallet_ffi_create_new( - config_path_c.as_ptr(), - storage_path_c.as_ptr(), - password.as_ptr(), - ) - }; - - Ok(create_wallet_result) -} - fn load_existing_ffi_wallet(home: &Path) -> Result<*mut WalletHandle> { let config_path = home.join("wallet_config.json"); let storage_path = home.join("storage.json"); + let metrics_path = home.join("metrics.json"); let config_path = CString::new(config_path.to_str().unwrap())?; let storage_path = CString::new(storage_path.to_str().unwrap())?; + let metrics_path = CString::new(metrics_path.to_str().unwrap())?; - Ok(unsafe { wallet_ffi_open(config_path.as_ptr(), storage_path.as_ptr()) }) + Ok(unsafe { + wallet_ffi_open( + config_path.as_ptr(), + storage_path.as_ptr(), + metrics_path.as_ptr(), + ) + }) } #[test] fn wallet_ffi_create_public_accounts() -> Result<()> { - let password = "password_for_tests"; + let ctx = BlockingTestContext::new()?; let n_accounts = 10; // Create `n_accounts` public accounts with wallet FFI let new_public_account_ids_ffi = unsafe { let mut account_ids = Vec::new(); + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: wallet_ffi_handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; for _ in 0..n_accounts { let mut out_account_id = FfiBytes32::from_bytes([0; 32]); wallet_ffi_create_account_public(wallet_ffi_handle, &raw mut out_account_id).unwrap(); @@ -436,16 +430,17 @@ fn wallet_ffi_create_public_accounts() -> Result<()> { #[test] fn wallet_ffi_create_private_accounts() -> Result<()> { - let password = "password_for_tests"; + let ctx = BlockingTestContext::new()?; let n_accounts = 10; // Create `n_accounts` receiving keys with wallet FFI let new_npks_ffi = unsafe { let mut npks = Vec::new(); + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: wallet_ffi_handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; for _ in 0..n_accounts { let mut out_keys = FfiPrivateAccountKeys::default(); wallet_ffi_create_private_accounts_key(wallet_ffi_handle, &raw mut out_keys).unwrap(); @@ -510,14 +505,14 @@ fn wallet_ffi_save_and_load_persistent_storage() -> Result<()> { #[test] fn test_wallet_ffi_list_accounts() -> Result<()> { - let password = "password_for_tests"; - + let ctx = BlockingTestContext::new()?; // Create the wallet FFI and track which account IDs were created as public/private let (wallet_ffi_handle, created_public_ids) = unsafe { + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; let mut public_ids: Vec<[u8; 32]> = Vec::new(); // Create 5 public accounts and 5 receiving keys diff --git a/lee/state_machine/core/src/commitment.rs b/lee/state_machine/core/src/commitment.rs index 92085d7d..537a9643 100644 --- a/lee/state_machine/core/src/commitment.rs +++ b/lee/state_machine/core/src/commitment.rs @@ -32,7 +32,7 @@ pub const DUMMY_COMMITMENT_HASH: [u8; 32] = [ #[derive(Serialize, Deserialize, BorshSerialize, BorshDeserialize)] #[cfg_attr( any(feature = "host", test), - derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord) + derive(Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord) )] pub struct Commitment(pub(super) [u8; 32]); diff --git a/lee/state_machine/src/state/mod.rs b/lee/state_machine/src/state/mod.rs index 97a77978..94d919ca 100644 --- a/lee/state_machine/src/state/mod.rs +++ b/lee/state_machine/src/state/mod.rs @@ -44,7 +44,7 @@ impl CommitmentSet { /// Inserts a list of commitments to the `CommitmentSet`. pub(crate) fn extend(&mut self, commitments: &[Commitment]) { - for commitment in commitments.iter().cloned() { + for commitment in commitments.iter().copied() { let index = self.merkle_tree.insert(commitment.to_byte_array()); self.commitments.insert(commitment, index); } diff --git a/lee/state_machine/src/state/tests/claiming.rs b/lee/state_machine/src/state/tests/claiming.rs index acfe9834..d45f39bd 100644 --- a/lee/state_machine/src/state/tests/claiming.rs +++ b/lee/state_machine/src/state/tests/claiming.rs @@ -308,7 +308,7 @@ fn authorized_public_account_claiming_succeeds_when_executed_privately() { let sender_commitment = Commitment::new(&sender_account_id, &sender_private_account); let sender_init_nullifier = Nullifier::for_account_initialization(&sender_account_id); let mut state = - V03State::new().with_private_accounts([(sender_commitment.clone(), sender_init_nullifier)]); + V03State::new().with_private_accounts([(sender_commitment, sender_init_nullifier)]); let sender_pre = AccountWithMetadata::new( sender_private_account, true, @@ -401,8 +401,8 @@ fn private_chained_call(number_of_calls: u32) { let to_init_nullifier = Nullifier::for_account_initialization(&to_account_id); let mut state = V03State::new() .with_private_accounts([ - (from_commitment.clone(), from_init_nullifier), - (to_commitment.clone(), to_init_nullifier), + (from_commitment, from_init_nullifier), + (to_commitment, to_init_nullifier), ]) .with_test_programs(); let amount: u128 = 37; diff --git a/lee/state_machine/src/state/tests/privacy_preserving.rs b/lee/state_machine/src/state/tests/privacy_preserving.rs index 8c33e202..174aa1b3 100644 --- a/lee/state_machine/src/state/tests/privacy_preserving.rs +++ b/lee/state_machine/src/state/tests/privacy_preserving.rs @@ -505,7 +505,7 @@ fn malicious_authorization_changer_should_fail_in_privacy_preserving_circuit() { sender_account.account_id, sender_account.account.balance, )])) - .with_private_accounts([(recipient_commitment.clone(), recipient_init_nullifier)]) + .with_private_accounts([(recipient_commitment, recipient_init_nullifier)]) .with_test_programs(); let balance_to_transfer = 10_u128; diff --git a/lez/wallet-ffi/Cargo.toml b/lez/wallet-ffi/Cargo.toml index b5fe7635..5440bee2 100644 --- a/lez/wallet-ffi/Cargo.toml +++ b/lez/wallet-ffi/Cargo.toml @@ -14,8 +14,6 @@ crate-type = ["rlib", "cdylib", "staticlib"] wallet.workspace = true lee.workspace = true lee_core.workspace = true -sequencer_service_rpc = { workspace = true, features = ["client"] } -common.workspace = true programs.workspace = true tokio.workspace = true diff --git a/lez/wallet-ffi/src/program_deployment.rs b/lez/wallet-ffi/src/program_deployment.rs index 2086fa5c..2e5550ae 100644 --- a/lez/wallet-ffi/src/program_deployment.rs +++ b/lez/wallet-ffi/src/program_deployment.rs @@ -1,9 +1,5 @@ use std::{ffi::CString, ptr, slice}; -use common::transaction::LeeTransaction; -use lee::ProgramDeploymentTransaction; -use sequencer_service_rpc::RpcClient as _; - use crate::{ block_on, error::{print_error, WalletFfiError}, @@ -60,14 +56,7 @@ pub unsafe extern "C" fn wallet_ffi_program_deployment( let elf = unsafe { slice::from_raw_parts(elf_data, elf_size) }.to_vec(); - let message = lee::program_deployment_transaction::Message::new(elf); - let transaction = ProgramDeploymentTransaction::new(message); - - match block_on( - wallet - .sequencer_client - .send_transaction(LeeTransaction::ProgramDeployment(transaction)), - ) { + match block_on(wallet.send_program_deployment_transaction(elf)) { Ok(tx_hash) => { let tx_hash = CString::new(tx_hash.to_string()) .map_or(ptr::null_mut(), std::ffi::CString::into_raw); diff --git a/lez/wallet-ffi/src/sync.rs b/lez/wallet-ffi/src/sync.rs index 5f7a4413..b65a0944 100644 --- a/lez/wallet-ffi/src/sync.rs +++ b/lez/wallet-ffi/src/sync.rs @@ -1,7 +1,5 @@ //! Block synchronization functions. -use sequencer_service_rpc::RpcClient as _; - use crate::{ block_on, error::{print_error, WalletFfiError}, @@ -136,7 +134,7 @@ pub unsafe extern "C" fn wallet_ffi_get_current_block_height( } }; - match block_on(wallet.sequencer_client.get_last_block_id()) { + match block_on(wallet.get_last_block_id()) { Ok(last_block_id) => { unsafe { *out_block_height = last_block_id; diff --git a/lez/wallet-ffi/src/wallet.rs b/lez/wallet-ffi/src/wallet.rs index b19aaae5..b2846912 100644 --- a/lez/wallet-ffi/src/wallet.rs +++ b/lez/wallet-ffi/src/wallet.rs @@ -86,6 +86,7 @@ fn c_str_to_path(ptr: *const c_char, name: &str) -> Result Result FfiCreateWalletOutput { let Ok(config_path) = c_str_to_path(config_path, "config_path") else { @@ -112,7 +114,17 @@ pub unsafe extern "C" fn wallet_ffi_create_new( return FfiCreateWalletOutput::default(); }; - match WalletCore::new_init_storage(config_path, storage_path, None, &password) { + let Ok(statistics_path) = c_str_to_path(statistics_path, "statistics_path") else { + return FfiCreateWalletOutput::default(); + }; + + match block_on(WalletCore::new_init_storage( + config_path, + storage_path, + statistics_path, + None, + &password, + )) { Ok((core, mnemonic)) => { let wrapper = Box::new(WalletWrapper { core: Mutex::new(core), @@ -143,7 +155,8 @@ pub unsafe extern "C" fn wallet_ffi_create_new( /// /// # Parameters /// - `config_path`: Path to the wallet configuration file (JSON) -/// - `storage_path`: Path where wallet data is stored +/// - `storage_path`: Path to the wallet storage (JSON) +/// - `statistics_path`: Path to the wallet statistics file (JSON) /// /// # Returns /// - Opaque wallet handle on success @@ -155,6 +168,7 @@ pub unsafe extern "C" fn wallet_ffi_create_new( pub unsafe extern "C" fn wallet_ffi_open( config_path: *const c_char, storage_path: *const c_char, + statistics_path: *const c_char, ) -> *mut WalletHandle { let Ok(config_path) = c_str_to_path(config_path, "config_path") else { return ptr::null_mut(); @@ -164,7 +178,16 @@ pub unsafe extern "C" fn wallet_ffi_open( return ptr::null_mut(); }; - match WalletCore::new_update_chain(config_path, storage_path, None) { + let Ok(statistics_path) = c_str_to_path(statistics_path, "statistics_path") else { + return ptr::null_mut(); + }; + + match block_on(WalletCore::new_update_chain( + config_path, + storage_path, + statistics_path, + None, + )) { Ok(core) => { let wrapper = Box::new(WalletWrapper { core: Mutex::new(core), @@ -216,7 +239,7 @@ pub unsafe extern "C" fn wallet_ffi_save(handle: *mut WalletHandle) -> WalletFfi Err(e) => return e, }; - let wallet = match wrapper.core.lock() { + let mut wallet = match wrapper.core.lock() { Ok(w) => w, Err(e) => { print_error(format!("Failed to lock wallet: {e}")); @@ -224,7 +247,10 @@ pub unsafe extern "C" fn wallet_ffi_save(handle: *mut WalletHandle) -> WalletFfi } }; - match wallet.store_persistent_data() { + match wallet + .store_persistent_data() + .and_then(|()| block_on(wallet.client_rotation())) + { Ok(()) => WalletFfiError::Success, Err(e) => { print_error(format!("Failed to save wallet: {e}")); @@ -334,7 +360,7 @@ pub unsafe extern "C" fn wallet_ffi_get_sequencer_addr(handle: *mut WalletHandle } }; - let addr = wallet.config().sequencer_addr.clone().to_string(); + let addr = wallet.leader_url().to_string(); match std::ffi::CString::new(addr) { Ok(s) => s.into_raw(), diff --git a/lez/wallet-ffi/wallet_ffi.h b/lez/wallet-ffi/wallet_ffi.h index 4104fb61..bbd7da1f 100644 --- a/lez/wallet-ffi/wallet_ffi.h +++ b/lez/wallet-ffi/wallet_ffi.h @@ -1612,6 +1612,7 @@ enum WalletFfiError wallet_ffi_vault_claim_private(struct WalletHandle *handle, * # Parameters * - `config_path`: Path to the wallet configuration file (JSON) * - `storage_path`: Path where wallet data will be stored + * - `statistics_path`: Path to the wallet statistics file (JSON) * - `password`: Password for encrypting the wallet seed * * # Returns @@ -1623,6 +1624,7 @@ enum WalletFfiError wallet_ffi_vault_claim_private(struct WalletHandle *handle, */ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, const char *storage_path, + const char *statistics_path, const char *password); /** @@ -1632,7 +1634,8 @@ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, * * # Parameters * - `config_path`: Path to the wallet configuration file (JSON) - * - `storage_path`: Path where wallet data is stored + * - `storage_path`: Path to the wallet storage (JSON) + * - `statistics_path`: Path to the wallet statistics file (JSON) * * # Returns * - Opaque wallet handle on success @@ -1641,7 +1644,9 @@ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, * # Safety * All string parameters must be valid null-terminated UTF-8 strings. */ -struct WalletHandle *wallet_ffi_open(const char *config_path, const char *storage_path); +struct WalletHandle *wallet_ffi_open(const char *config_path, + const char *storage_path, + const char *statistics_path); /** * Destroy a wallet handle and free its resources. diff --git a/lez/wallet/configs/debug/wallet_config.json b/lez/wallet/configs/debug/wallet_config.json index 926ee298..c8bd872b 100644 --- a/lez/wallet/configs/debug/wallet_config.json +++ b/lez/wallet/configs/debug/wallet_config.json @@ -1,7 +1,10 @@ { - "sequencer_addr": "http://127.0.0.1:3040", + "sequencers": [{ + "sequencer_addr": "http://127.0.0.1:3040" + }], "seq_poll_timeout": "30s", "seq_tx_poll_max_blocks": 15, "seq_poll_max_retries": 10, - "seq_block_poll_max_amount": 100 + "seq_block_poll_max_amount": 100, + "calibration_limit": 100 } \ No newline at end of file diff --git a/lez/wallet/src/cli/account.rs b/lez/wallet/src/cli/account.rs index 275e13fc..c700d2e9 100644 --- a/lez/wallet/src/cli/account.rs +++ b/lez/wallet/src/cli/account.rs @@ -384,38 +384,61 @@ impl AccountSubcommand { Ok(SubcommandReturnValue::Empty) } - async fn handle_list(long: bool, wallet_core: &WalletCore) -> Result { - let key_chain = &wallet_core.storage.key_chain(); - let storage = wallet_core.storage(); + fn format_with_label( + wallet_core: &WalletCore, + id: AccountIdWithPrivacy, + chain_index: Option<&ChainIndex>, + ) -> String { + let id_str = chain_index.map_or_else(|| id.to_string(), |cci| format!("{cci} {id}")); - let format_with_label = |id: AccountIdWithPrivacy, chain_index: Option<&ChainIndex>| { - let id_str = chain_index.map_or_else(|| id.to_string(), |cci| format!("{cci} {id}")); - - let labels = storage.labels_for_account(id).format(", ").to_string(); - if labels.is_empty() { - id_str - } else { - format!("{id_str} [{labels}]") - } - }; - - if !long { - let accounts = key_chain - .account_ids() - .map(|(id, idx)| format_with_label(id, idx)) - .format("\n"); - println!("{accounts}"); - - return Ok(SubcommandReturnValue::Empty); + let labels = wallet_core + .storage() + .labels_for_account(id) + .format(", ") + .to_string(); + if labels.is_empty() { + id_str + } else { + format!("{id_str} [{labels}]") } + } + + async fn handle_list(long: bool, wallet_core: &WalletCore) -> Result { + let (public_account_ids, private_account_ids) = { + let key_chain = &wallet_core.storage.key_chain(); + + if !long { + let accounts = key_chain + .account_ids() + .map(|(id, idx)| Self::format_with_label(wallet_core, id, idx)) + .format("\n"); + println!("{accounts}"); + + return Ok(SubcommandReturnValue::Empty); + } + + let public_account_ids: Vec<_> = key_chain + .public_account_ids() + .map(|(id, chain_index)| (id, chain_index.cloned())) + .collect(); + let private_account_ids: Vec<_> = key_chain + .private_account_ids() + .map(|(id, chain_index)| (id, chain_index.cloned())) + .collect(); + + (public_account_ids, private_account_ids) + }; // Detailed listing with --long flag - // Public key tree accounts - for (id, chain_index) in key_chain.public_account_ids() { + for (id, chain_index) in public_account_ids { println!( "{}", - format_with_label(AccountIdWithPrivacy::Public(id), chain_index) + Self::format_with_label( + wallet_core, + AccountIdWithPrivacy::Public(id), + chain_index.as_ref() + ) ); match wallet_core.get_account_public(id).await { Ok(account) if account != Account::default() => { @@ -429,10 +452,14 @@ impl AccountSubcommand { } // Private key tree accounts - for (id, chain_index) in key_chain.private_account_ids() { + for (id, chain_index) in private_account_ids { println!( "{}", - format_with_label(AccountIdWithPrivacy::Private(id), chain_index) + Self::format_with_label( + wallet_core, + AccountIdWithPrivacy::Private(id), + chain_index.as_ref() + ) ); match wallet_core.get_account_private(id) { Some(account) if account != Account::default() => { diff --git a/lez/wallet/src/cli/chain.rs b/lez/wallet/src/cli/chain.rs index dfb22eba..b14a5aa3 100644 --- a/lez/wallet/src/cli/chain.rs +++ b/lez/wallet/src/cli/chain.rs @@ -1,7 +1,6 @@ use anyhow::Result; use clap::Subcommand; use common::HashType; -use sequencer_service_rpc::RpcClient as _; use crate::{ WalletCore, @@ -33,17 +32,17 @@ impl WalletSubcommand for ChainSubcommand { ) -> Result { match self { Self::CurrentBlockId => { - let latest_block_id = wallet_core.sequencer_client.get_last_block_id().await?; + let latest_block_id = wallet_core.get_last_block_id().await?; println!("Last block id is {latest_block_id}"); } Self::Block { id } => { - let block = wallet_core.sequencer_client.get_block(id).await?; + let block = wallet_core.get_block(id).await?; println!("Last block id is {block:#?}"); } Self::Transaction { hash } => { - let tx = wallet_core.sequencer_client.get_transaction(hash).await?; + let tx = wallet_core.get_transaction(hash).await?; println!("Transaction is {tx:#?}"); } diff --git a/lez/wallet/src/cli/config.rs b/lez/wallet/src/cli/config.rs index a78bd097..8c8c4b5c 100644 --- a/lez/wallet/src/cli/config.rs +++ b/lez/wallet/src/cli/config.rs @@ -36,8 +36,8 @@ impl ConfigSubcommand { println!("{config_str}"); } else if let Some(key) = key { match key.as_str() { - "sequencer_addr" => { - println!("{}", config.sequencer_addr); + "sequencers" => { + println!("{:?}", config.sequencers); } "seq_poll_timeout" => { println!("{:?}", config.seq_poll_timeout); @@ -51,13 +51,6 @@ impl ConfigSubcommand { "seq_block_poll_max_amount" => { println!("{}", config.seq_block_poll_max_amount); } - "basic_auth" => { - if let Some(basic_auth) = &config.basic_auth { - println!("{basic_auth}"); - } else { - println!("Not set"); - } - } _ => { println!("Unknown field"); } @@ -76,9 +69,6 @@ impl ConfigSubcommand { ) -> Result { let mut config = wallet_core.config().clone(); match key.as_str() { - "sequencer_addr" => { - config.sequencer_addr = value.parse()?; - } "seq_poll_timeout" => { config.seq_poll_timeout = humantime::parse_duration(&value) .map_err(|e| anyhow::anyhow!("Invalid duration: {e}"))?; @@ -92,9 +82,6 @@ impl ConfigSubcommand { "seq_block_poll_max_amount" => { config.seq_block_poll_max_amount = value.parse()?; } - "basic_auth" => { - config.basic_auth = Some(value.parse()?); - } "initial_accounts" => { anyhow::bail!("Setting this field from wallet is not supported"); } diff --git a/lez/wallet/src/cli/mod.rs b/lez/wallet/src/cli/mod.rs index 44b578a1..d436de52 100644 --- a/lez/wallet/src/cli/mod.rs +++ b/lez/wallet/src/cli/mod.rs @@ -3,10 +3,9 @@ use std::{io::Write as _, path::PathBuf, str::FromStr}; use anyhow::{Context as _, Result}; use bip39::Mnemonic; use clap::{Parser, Subcommand}; -use common::{HashType, transaction::LeeTransaction}; +use common::HashType; use derive_more::Display; use futures::TryFutureExt as _; -use lee::ProgramDeploymentTransaction; use lee_core::BlockId; use sequencer_service_rpc::RpcClient as _; @@ -108,9 +107,6 @@ pub struct Args { /// Continious run flag. #[arg(short, long)] pub continuous_run: bool, - /// Basic authentication in the format `user` or `user:password`. - #[arg(long)] - pub auth: Option, /// Wallet command. #[command(subcommand)] pub command: Option, @@ -220,7 +216,6 @@ pub async fn execute_subcommand( } Command::CheckHealth => { let remote_program_ids = wallet_core - .sequencer_client .get_program_ids() .await .expect("Error fetching program ids"); @@ -285,21 +280,25 @@ pub async fn execute_subcommand( "Failed to read program binary at {}", binary_filepath.display() ))?; - let message = lee::program_deployment_transaction::Message::new(bytecode); - let transaction = ProgramDeploymentTransaction::new(message); - let tx_hash = wallet_core - .sequencer_client - .send_transaction(LeeTransaction::ProgramDeployment(transaction)) + let response = wallet_core + .send_program_deployment_transaction(bytecode) .await .context("Transaction submission error")?; wallet_core - .poll_and_finalize_public_transaction(tx_hash) + .poll_and_finalize_public_transaction(response) .await .context("Transaction finalization error")? } }; + // Kind of a sledgehammer solution, but it is not clear if there is the case to not store + // statistics + wallet_core + .client_rotation() + .await + .context("Failed to rotate wallet")?; + Ok(subcommand_ret) } @@ -388,14 +387,13 @@ pub async fn execute_keys_restoration(wallet_core: &mut WalletCore, depth: u32) wallet_core.sync_to_latest_block().await?; + let leader_client = wallet_core.leader_owned(); + wallet_core .storage .key_chain_mut() .cleanup_trees_remove_uninit_layered(depth, |account_id| { - wallet_core - .sequencer_client - .get_account(account_id) - .map_err(Into::into) + leader_client.get_account(account_id).map_err(Into::into) }) .await?; diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 77c2729f..ede71073 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -7,6 +7,17 @@ use log::warn; use serde::{Deserialize, Serialize}; use url::Url; +const DEFAULT_CALLIBRATION_LIMIT: usize = 100; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SequencerConnectionData { + /// Connection data of all known sequencers. + pub sequencer_addr: Url, + /// Basic authentication credentials. + #[serde(skip_serializing_if = "Option::is_none")] + pub basic_auth: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct GasConfig { /// Gas spent per deploying one byte of data. @@ -28,8 +39,8 @@ pub struct GasConfig { #[optfield::optfield(pub WalletConfigOverrides, rewrap, attrs = (derive(Debug, Default, Clone)))] #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WalletConfig { - /// Sequencer URL. - pub sequencer_addr: Url, + /// Connection data of all known sequencers. + pub sequencers: Vec, /// Sequencer polling duration for new blocks. #[serde(with = "humantime_serde")] pub seq_poll_timeout: Duration, @@ -39,20 +50,23 @@ pub struct WalletConfig { pub seq_poll_max_retries: u64, /// Max amount of blocks to poll in one request. pub seq_block_poll_max_amount: u64, - /// Basic authentication credentials - #[serde(skip_serializing_if = "Option::is_none")] - pub basic_auth: Option, + /// Limit number of sequencer polls during calibration, should not be zero + #[serde(default = "default_calibration_limit")] + pub calibration_limit: usize, } impl Default for WalletConfig { fn default() -> Self { Self { - sequencer_addr: "http://127.0.0.1:3040".parse().unwrap(), + sequencers: vec![SequencerConnectionData { + sequencer_addr: "http://127.0.0.1:3040".parse().unwrap(), + basic_auth: None, + }], seq_poll_timeout: Duration::from_secs(12), seq_tx_poll_max_blocks: 5, seq_poll_max_retries: 5, seq_block_poll_max_amount: 100, - basic_auth: None, + calibration_limit: DEFAULT_CALLIBRATION_LIMIT, } } } @@ -97,26 +111,26 @@ impl WalletConfig { pub fn apply_overrides(&mut self, overrides: WalletConfigOverrides) { let Self { - sequencer_addr, + sequencers, seq_poll_timeout, seq_tx_poll_max_blocks, seq_poll_max_retries, seq_block_poll_max_amount, - basic_auth, + calibration_limit, } = self; let WalletConfigOverrides { - sequencer_addr: o_sequencer_addr, + sequencers: o_sequencers, seq_poll_timeout: o_seq_poll_timeout, seq_tx_poll_max_blocks: o_seq_tx_poll_max_blocks, seq_poll_max_retries: o_seq_poll_max_retries, seq_block_poll_max_amount: o_seq_block_poll_max_amount, - basic_auth: o_basic_auth, + calibration_limit: o_calibration_limit, } = overrides; - if let Some(v) = o_sequencer_addr { - warn!("Overriding wallet config 'sequencer_addr' to {v}"); - *sequencer_addr = v; + if let Some(v) = o_sequencers { + warn!("Overriding wallet config 'sequencers' to {v:?}"); + *sequencers = v; } if let Some(v) = o_seq_poll_timeout { warn!("Overriding wallet config 'seq_poll_timeout' to {v:?}"); @@ -134,9 +148,13 @@ impl WalletConfig { warn!("Overriding wallet config 'seq_block_poll_max_amount' to {v}"); *seq_block_poll_max_amount = v; } - if let Some(v) = o_basic_auth { - warn!("Overriding wallet config 'basic_auth' to {v:#?}"); - *basic_auth = v; + if let Some(v) = o_calibration_limit { + warn!("Overriding wallet config 'calibration_limit' to {v}"); + *calibration_limit = v; } } } + +const fn default_calibration_limit() -> usize { + DEFAULT_CALLIBRATION_LIMIT +} diff --git a/lez/wallet/src/helperfunctions.rs b/lez/wallet/src/helperfunctions.rs index 6fe78681..0af99c1f 100644 --- a/lez/wallet/src/helperfunctions.rs +++ b/lez/wallet/src/helperfunctions.rs @@ -66,6 +66,15 @@ pub fn fetch_persistent_storage_path() -> Result { Ok(accs_path) } +/// Fetch path to statistics storage from default home. +/// +/// File must be created through setup beforehand. +pub fn fetch_statistics_path() -> Result { + let home = get_home()?; + let statistics_path = home.join("statistics.json"); + Ok(statistics_path) +} + #[expect(dead_code, reason = "Maybe used later")] pub(crate) fn produce_random_nonces(size: usize) -> Vec { let mut result = vec![[0; 16]; size]; diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index fdd208af..4984f784 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -7,33 +7,38 @@ reason = "Most of the shadows come from args parsing which is ok" )] -use std::{collections::HashSet, path::PathBuf}; +use std::{ + collections::{BTreeMap, HashMap, HashSet}, + path::PathBuf, +}; pub use account_manager::AccountIdentity; use anyhow::{Context as _, Result}; use bip39::Mnemonic; -use common::{HashType, transaction::LeeTransaction}; +use common::{HashType, block::Block, transaction::LeeTransaction}; use config::WalletConfig; use key_protocol::key_management::key_tree::chain_index::ChainIndex; use lee::{ - Account, AccountId, PrivacyPreservingTransaction, ProgramId, + Account, AccountId, PrivacyPreservingTransaction, ProgramDeploymentTransaction, ProgramId, privacy_preserving_transaction::{ circuit::ProgramWithDependencies, message::{EncryptedAccountData, Message}, }, }; use lee_core::{ - Commitment, CommitmentSetDigest, MembershipProof, SharedSecretKey, account::Nonce, + BlockId, Commitment, CommitmentSetDigest, MembershipProof, SharedSecretKey, account::Nonce, program::InstructionData, }; use log::info; -use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; +use sequencer_service_rpc::{RpcClient as _, SequencerClient}; use storage::Storage; use tokio::io::AsyncWriteExt as _; +use url::Url; use crate::{ account::{AccountIdWithPrivacy, Label}, config::WalletConfigOverrides, + multi_client::{MultiSequencerClient, Statistics, extract_statistics_from_path}, poller::TxPoller, storage::key_chain::{NullifierIndex, SharedAccountEntry}, }; @@ -43,6 +48,7 @@ mod account_manager; pub mod cli; pub mod config; pub mod helperfunctions; +pub mod multi_client; pub mod poller; pub mod program_facades; pub mod signing; @@ -84,7 +90,6 @@ pub enum ExecutionFailureKind { KeycardError(#[from] pyo3::PyErr), } -#[expect(clippy::partial_pub_fields, reason = "TODO: make all fields private")] pub struct WalletCore { config_path: PathBuf, config_overrides: Option, @@ -93,45 +98,65 @@ pub struct WalletCore { storage: Storage, storage_path: PathBuf, - poller: TxPoller, - pub sequencer_client: SequencerClient, + statistics_path: PathBuf, + statistics: HashMap, + + multi_sequencer_client: MultiSequencerClient, } impl WalletCore { /// Construct wallet using [`HOME_DIR_ENV_VAR`] env var for paths or user home dir if not set. - pub fn from_env() -> Result { + pub async fn from_env() -> Result { let config_path = helperfunctions::fetch_config_path()?; let storage_path = helperfunctions::fetch_persistent_storage_path()?; + let statistics_path = helperfunctions::fetch_statistics_path()?; - Self::new_update_chain(config_path, storage_path, None) + Self::new_update_chain(config_path, storage_path, statistics_path, None).await } - pub fn new_update_chain( + pub async fn new_update_chain( config_path: PathBuf, storage_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, ) -> Result { let storage = Storage::from_path(&storage_path) .with_context(|| format!("Failed to load storage from {}", storage_path.display()))?; - Self::new(config_path, storage_path, config_overrides, storage) + Self::new( + config_path, + storage_path, + statistics_path, + config_overrides, + storage, + ) + .await } - pub fn new_init_storage( + pub async fn new_init_storage( config_path: PathBuf, storage_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, password: &str, ) -> Result<(Self, Mnemonic)> { let (storage, mnemonic) = Storage::new(password).context("Failed to create storage")?; - let wallet = Self::new(config_path, storage_path, config_overrides, storage)?; + let wallet = Self::new( + config_path, + storage_path, + statistics_path, + config_overrides, + storage, + ) + .await?; Ok((wallet, mnemonic)) } - fn new( + async fn new( config_path: PathBuf, storage_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, storage: Storage, ) -> Result { @@ -146,34 +171,24 @@ impl WalletCore { config.apply_overrides(config_overrides); } - let sequencer_client = { - let mut builder = SequencerClientBuilder::default(); - if let Some(basic_auth) = &config.basic_auth { - builder = builder.set_headers( - std::iter::once(( - "Authorization".parse().expect("Header name is valid"), - format!("Basic {basic_auth}") - .parse() - .context("Invalid basic auth format")?, - )) - .collect(), - ); - } - builder - .build(config.sequencer_addr.clone()) - .context("Failed to create sequencer client")? - }; + let mut statistics = extract_statistics_from_path(&statistics_path)?; - let tx_poller = TxPoller::new(&config, sequencer_client.clone()); + let multi_sequencer_client = MultiSequencerClient::new( + &config.sequencers, + &mut statistics, + config.calibration_limit, + ) + .await?; Ok(Self { config_path, config_overrides, config, - storage_path, storage, - poller: tx_poller, - sequencer_client, + storage_path, + statistics_path, + statistics, + multi_sequencer_client, }) } @@ -187,6 +202,21 @@ impl WalletCore { self.config = config; } + #[must_use] + pub fn optimal_poller(&self) -> TxPoller { + TxPoller::new(self.config(), self.leader_owned()) + } + + #[must_use] + pub fn leader_owned(&self) -> SequencerClient { + self.multi_sequencer_client.leader().clone() + } + + #[must_use] + pub fn leader_url(&self) -> Url { + self.multi_sequencer_client.leader_url().clone() + } + /// Get storage. #[must_use] pub const fn storage(&self) -> &Storage { @@ -223,6 +253,39 @@ impl WalletCore { Ok(()) } + /// Rotates multi-client and stores metrics. + pub async fn client_rotation(&mut self) -> Result<()> { + let leader_statistic = self + .statistics + .get_mut(self.multi_sequencer_client.leader_url()) + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in statistics"))?; + + self.multi_sequencer_client + .update_statistics(leader_statistic) + .await; + + self.multi_sequencer_client + .rotate( + &self.config.sequencers, + &mut self.statistics, + self.config.calibration_limit, + ) + .await?; + + let statistics_serialized = serde_json::to_vec_pretty(&self.statistics)?; + let mut file = tokio::fs::File::create(&self.statistics_path) + .await + .context("Failed to create file")?; + file.write_all(&statistics_serialized) + .await + .context("Failed to write to file")?; + file.sync_all().await.context("Failed to sync file")?; + + println!("Stored statistics at {}", self.statistics_path.display()); + + Ok(()) + } + /// Store persistent data at home. pub async fn store_config_changes(&self) -> Result<()> { let config = serde_json::to_vec_pretty(&self.config)?; @@ -414,14 +477,27 @@ impl WalletCore { }) } + #[must_use] + pub fn get_statistics(&self, sequencer_url: &Url) -> Option<&Statistics> { + self.statistics.get(sequencer_url) + } + /// Get account balance. pub async fn get_account_balance(&self, acc: AccountId) -> Result { - Ok(self.sequencer_client.get_account_balance(acc).await?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_account_balance(acc).await) + .await?) } /// Get accounts nonces. - pub async fn get_accounts_nonces(&self, accs: Vec) -> Result> { - Ok(self.sequencer_client.get_accounts_nonces(accs).await?) + pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result> { + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client.get_accounts_nonces(accs.to_vec()).await + }) + .await?) } pub async fn get_account(&self, account_id: AccountIdWithPrivacy) -> Result { @@ -437,9 +513,36 @@ impl WalletCore { } } + pub async fn get_last_block_id(&self) -> Result { + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_last_block_id().await) + .await?) + } + + pub async fn get_block(&self, block_id: u64) -> Result> { + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_block(block_id).await) + .await?) + } + + pub async fn get_transaction( + &self, + hash: HashType, + ) -> Result> { + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_transaction(hash).await) + .await?) + } + /// Get public account. pub async fn get_account_public(&self, account_id: AccountId) -> Result { - Ok(self.sequencer_client.get_account(account_id).await?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_account(account_id).await) + .await?) } #[must_use] @@ -474,14 +577,31 @@ impl WalletCore { Some(Commitment::new(&account_id, account)) } + pub async fn get_program_ids(&self) -> Result> { + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_program_ids().await) + .await?) + } + + /// Poll transactions. + pub async fn poll_native_token_transfer( + &self, + hash: HashType, + ) -> Result<(LeeTransaction, BlockId)> { + self.optimal_poller().poll_tx(hash).await + } + pub async fn get_proofs_and_root( &self, commitments: Vec, ) -> Result<(Vec>, CommitmentSetDigest)> { - self.sequencer_client - .get_proofs_and_root(commitments) - .await - .map_err(Into::into) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client.get_proofs_and_root(commitments.clone()).await + }) + .await?) } pub fn decode_insert_privacy_preserving_transaction_results( @@ -499,7 +619,7 @@ impl WalletCore { match acc_decode_data { AccDecodeData::Decode(secret, acc_account_id) => { let acc_ead = tx.message.encrypted_private_post_states[output_index].clone(); - let acc_comm = tx.message.new_commitments[output_index].clone(); + let acc_comm = tx.message.new_commitments[output_index]; let (kind, res_acc) = lee_core::EncryptionScheme::decrypt( &acc_ead.ciphertext, @@ -530,7 +650,7 @@ impl WalletCore { tx_hash: HashType, ) -> Result { println!("Transaction hash is {tx_hash}"); - let (tx, block_id) = self.poller.poll_tx(tx_hash).await?; + let (tx, block_id) = self.optimal_poller().poll_tx(tx_hash).await?; println!("Transaction is included in block {block_id}"); println!("Transaction data is {tx:?}"); self.store_persistent_data()?; @@ -544,7 +664,7 @@ impl WalletCore { acc_decode_data: &[AccDecodeData], ) -> Result { println!("Transaction hash is {tx_hash}"); - let (tx, block_id) = self.poller.poll_tx(tx_hash).await?; + let (tx, block_id) = self.optimal_poller().poll_tx(tx_hash).await?; println!("Transaction is included in block {block_id}"); println!("Transaction data is {tx:?}"); if let common::transaction::LeeTransaction::PrivacyPreserving(private_tx) = tx { @@ -620,12 +740,16 @@ impl WalletCore { .map(|keys| keys.ssk) .collect(); - Ok(( - self.sequencer_client - .send_transaction(LeeTransaction::PrivacyPreserving(tx)) - .await?, - shared_secrets, - )) + let call_res = self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) + .await + }) + .await; + + Ok((call_res?, shared_secrets)) } pub async fn send_pub_tx( @@ -685,13 +809,31 @@ impl WalletCore { let tx = lee::public_transaction::PublicTransaction::new(message, witness_set); Ok(self - .sequencer_client - .send_transaction(LeeTransaction::Public(tx)) + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::Public(tx.clone())) + .await + }) + .await?) + } + + pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { + let message = lee::program_deployment_transaction::Message::new(bytecode); + let transaction = ProgramDeploymentTransaction::new(message); + + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) + .await + }) .await?) } pub async fn sync_to_latest_block(&mut self) -> Result { - let latest_block_id = self.sequencer_client.get_last_block_id().await?; + let latest_block_id = self.get_last_block_id().await?; println!("Latest block is {latest_block_id}"); self.sync_to_block(latest_block_id).await?; Ok(latest_block_id) @@ -713,7 +855,7 @@ impl WalletCore { println!("Syncing to block {block_id}. Blocks to sync: {num_of_blocks}"); - let poller = self.poller.clone(); + let poller = self.optimal_poller(); let mut blocks = std::pin::pin!(poller.poll_block_range(last_synced_block.saturating_add(1)..=block_id)); diff --git a/lez/wallet/src/main.rs b/lez/wallet/src/main.rs index 838ee856..c9cf0820 100644 --- a/lez/wallet/src/main.rs +++ b/lez/wallet/src/main.rs @@ -8,8 +8,7 @@ use clap::{CommandFactory as _, Parser as _}; use wallet::{ WalletCore, cli::{Args, execute_continuous_run, execute_subcommand, read_password_from_stdin}, - config::WalletConfigOverrides, - helperfunctions::{fetch_config_path, fetch_persistent_storage_path}, + helperfunctions::{fetch_config_path, fetch_persistent_storage_path, fetch_statistics_path}, }; // TODO #169: We have sample configs for sequencer, but not for wallet @@ -21,7 +20,6 @@ use wallet::{ async fn main() -> Result<()> { let Args { continuous_run, - auth, command, } = Args::parse(); @@ -30,16 +28,11 @@ async fn main() -> Result<()> { let config_path = fetch_config_path().context("Could not fetch config path")?; let storage_path = fetch_persistent_storage_path().context("Could not fetch persistent storage path")?; - - // Override basic auth if provided via CLI - let config_overrides = WalletConfigOverrides { - basic_auth: auth.map(|auth| auth.parse()).transpose()?.map(Some), - ..Default::default() - }; + let statistics_path = fetch_statistics_path().context("Could not fetch statistics path")?; if let Some(command) = command { let mut wallet = if storage_path.exists() { - WalletCore::new_update_chain(config_path, storage_path, Some(config_overrides))? + WalletCore::new_update_chain(config_path, storage_path, statistics_path, None).await? } else { // TODO: Maybe move to `WalletCore::from_env()` or similar? @@ -49,9 +42,11 @@ async fn main() -> Result<()> { let (wallet, mnemonic) = WalletCore::new_init_storage( config_path, storage_path, - Some(config_overrides), + statistics_path, + None, &password, - )?; + ) + .await?; println!(); println!("IMPORTANT: Write down your recovery phrase and store it securely."); @@ -68,7 +63,7 @@ async fn main() -> Result<()> { Ok(()) } else if continuous_run { let mut wallet = - WalletCore::new_update_chain(config_path, storage_path, Some(config_overrides))?; + WalletCore::new_update_chain(config_path, storage_path, statistics_path, None).await?; execute_continuous_run(&mut wallet).await } else { let help = Args::command().render_long_help(); diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs new file mode 100644 index 00000000..499cf7a1 --- /dev/null +++ b/lez/wallet/src/multi_client.rs @@ -0,0 +1,1038 @@ +#![expect( + clippy::float_arithmetic, + reason = "One should expect floating point arithmetic in statistic calculations" +)] +#![expect( + clippy::cast_precision_loss, + reason = "Operated numbers is not big enough to have precision loss" +)] + +use std::{collections::HashMap, path::Path, sync::Arc}; + +use anyhow::{Context as _, Result}; +use lee_core::BlockId; +use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; +use url::Url; + +use crate::config::SequencerConnectionData; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Statistics { + pub latency_avg: f32, + pub latency_var: f32, + pub sample_size: usize, + pub latest_block_id: BlockId, + pub errors: u64, +} + +#[derive(Debug, Clone)] +pub struct StatisticsUpdate { + pub latency: f32, + pub new_latest_block_id: Option, + pub is_failed: bool, +} + +impl Statistics { + pub fn apply_updates(&mut self, updates: &[StatisticsUpdate]) { + let CumulativeUpdates { + failure_count, + latest_block_id, + cumulative_latency, + cumulative_latency_squares, + additional_sample_size, + } = CumulativeUpdates::from_metric_updates(updates); + + self.errors = self.errors.saturating_add(failure_count); + if let Some(latest_block_id) = latest_block_id { + self.latest_block_id = latest_block_id; + } + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let orig_size_f = self.sample_size as f32; + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let mod_size_f = additional_sample_size as f32; + + let latency_avg_old = self.latency_avg; + let latency_avg_new = + cumulative_avg(latency_avg_old, cumulative_latency, orig_size_f, mod_size_f); + + let latency_var_new = cumulative_var( + latency_avg_old, + latency_avg_new, + self.latency_var, + cumulative_latency, + cumulative_latency_squares, + orig_size_f, + mod_size_f, + ); + + self.latency_avg = latency_avg_new; + self.latency_var = latency_var_new; + self.sample_size = self.sample_size.saturating_add(additional_sample_size); + } +} + +#[derive(Clone)] +pub struct MultiSequencerClient { + // For now we store only leader, it is possible, that + // in future for important sends(for example for transactions) + // we would want to distribute call between known sequencers + leader: SequencerClient, + leader_url: Url, + /// Wrapping statistic updates in Arc to not break interfaces too much. + /// + /// It is assumed, that wallet methods can be accesed via immutable reference. + statistic_updates: Arc>>, +} + +impl MultiSequencerClient { + async fn setup( + conn_data: &[SequencerConnectionData], + statistics: &mut HashMap, + calibration_limit: usize, + ) -> Result<(Url, SequencerClient)> { + let mut client_list = HashMap::new(); + + for SequencerConnectionData { + sequencer_addr, + basic_auth, + } in conn_data + { + let sequencer_client = { + let mut builder = SequencerClientBuilder::default(); + if let Some(basic_auth) = &basic_auth { + builder = builder.set_headers( + std::iter::once(( + "Authorization".parse().expect("Header name is valid"), + format!("Basic {basic_auth}") + .parse() + .context("Invalid basic auth format")?, + )) + .collect(), + ); + } + builder + .build(sequencer_addr) + .context("Failed to create sequencer client")? + }; + + // If there is statistics for client, actualize it + if let Some(metric_mut) = statistics.get_mut(sequencer_addr) { + let metric_updates = actualize_client(&sequencer_client).await; + + log::debug!( + "Metered call for {sequencer_addr:?}, metric updates is {metric_updates:?}" + ); + + metric_mut.apply_updates(&[metric_updates]); + // Otherwise calibrate client data + } else if let Some(client_statistics) = + calibrate_client(&sequencer_client, calibration_limit).await + { + statistics.insert(sequencer_addr.clone(), client_statistics); + } else { + log::warn!("Client {sequencer_addr:?} failed all {calibration_limit} calibration attempts, it may be unhealthy. + \n Consider bumping calibration_limit or remove this client altogether"); + } + + client_list.insert(sequencer_addr.clone(), sequencer_client); + } + + // Dropping client list, for reasons why, see comment in structure definition. + + let res = choose_leader(&client_list, statistics) + .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; + + log::info!("Chosen leader is {:?}", res.0); + + Ok(res) + } + + pub async fn new( + conn_data: &[SequencerConnectionData], + statistics: &mut HashMap, + calibration_limit: usize, + ) -> Result { + let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + + Ok(Self { + leader, + leader_url, + statistic_updates: Arc::new(RwLock::new(vec![])), + }) + } + + /// Re-choose leader, `statistic_updates` must be empty. + pub async fn rotate( + &mut self, + conn_data: &[SequencerConnectionData], + statistics: &mut HashMap, + calibration_limit: usize, + ) -> Result<()> { + let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + + log::info!("Chosen leader is {leader_url:?}"); + + self.leader = leader; + self.leader_url = leader_url; + + Ok(()) + } + + #[must_use] + pub const fn leader(&self) -> &SequencerClient { + &self.leader + } + + #[must_use] + pub const fn leader_url(&self) -> &Url { + &self.leader_url + } + + // Keeping this call abstract, in case if we need to do more than one request + pub async fn metered_call Result>( + &self, + call: I, + ) -> Result { + let (resp, statistics_update) = + tokio::join!(call(self.leader()), actualize_client(self.leader())); + + log::debug!( + "Metered call for {:?}, metric updates is {:?}", + self.leader_url, + statistics_update + ); + + { + let mut statistic_updates_guard = self.statistic_updates.write().await; + statistic_updates_guard.push(statistics_update); + } + + resp + } + + /// Update statistics of a leader, clear statist updates log. + pub async fn update_statistics(&self, leader_statistic: &mut Statistics) { + let mut statistic_updates = self.statistic_updates.write().await; + leader_statistic.apply_updates(statistic_updates.as_ref()); + statistic_updates.clear(); + } +} + +struct CumulativeUpdates { + pub failure_count: u64, + pub latest_block_id: Option, + /// Necessary for cumulative average calculation. + pub cumulative_latency: f32, + /// Necessary for cumulative variance calculation. + pub cumulative_latency_squares: f32, + pub additional_sample_size: usize, +} + +impl CumulativeUpdates { + fn from_metric_updates(metric_updates: &[StatisticsUpdate]) -> Self { + let (failure_count, latest_block_id, cumulative_latency, cumulative_latency_squares) = + metric_updates + .iter() + .fold((0_u64, None, 0_f32, 0_f32), |acc, x| { + let StatisticsUpdate { + latency, + new_latest_block_id, + is_failed, + } = x; + ( + if *is_failed { + acc.0.saturating_add(1) + } else { + acc.0 + }, + match (acc.1, new_latest_block_id) { + (None, None) => None, + (None, Some(val)) | (Some(val), None) => Some(val), + (Some(val_old), Some(val_new)) => Some(std::cmp::max(val_old, val_new)), + }, + if *is_failed { acc.2 } else { acc.2 + latency }, + if *is_failed { + acc.3 + } else { + latency.mul_add(*latency, acc.3) + }, + ) + }); + + Self { + failure_count, + latest_block_id: latest_block_id.copied(), + cumulative_latency, + cumulative_latency_squares, + additional_sample_size: metric_updates.len().saturating_sub( + usize::try_from(failure_count).expect("Sample size should fit usize"), + ), + } + } +} + +pub fn extract_statistics_from_path( + path: &Path, +) -> Result, anyhow::Error> { + match std::fs::File::open(path) { + Ok(file) => { + let reader = std::io::BufReader::new(file); + Ok(serde_json::from_reader(reader)?) + } + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + println!("Statistics not found, choosing empty"); + Ok(HashMap::new()) + } + Err(err) => Err(err).context("IO error"), + } +} + +/// Measuring `get_last_block_id` as it should be the fastest request on sequencer. +async fn measure_request_duration(client: &SequencerClient) -> (u128, Option) { + let now = tokio::time::Instant::now(); + let block_id = client.get_last_block_id().await.ok(); + ( + tokio::time::Instant::now().duration_since(now).as_millis(), + block_id, + ) +} + +pub async fn calibrate_client( + client: &SequencerClient, + calibration_limit: usize, +) -> Option { + let mut latencies = vec![]; + let mut latest_block_id = 0; + let mut errors: u64 = 0; + + // ToDo: Add some DDoS adaptation + for _ in 0..calibration_limit { + let (latency, block_id) = measure_request_duration(client).await; + + let Some(block_id) = block_id else { + errors = errors.saturating_add(1); + continue; + }; + + latest_block_id = block_id; + latencies.push(latency); + } + + // There is no point in guard numbers, exclude client if it fails all requests. + if latencies.is_empty() { + return None; + } + + // Precision loss is fine there + let sample_size = latencies.len(); + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let latency_avg = (latencies.iter().sum::() as f32) / (sample_size as f32); + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let latency_var = latencies.iter().fold(0_f32, |acc, x| { + ((*x as f32) - latency_avg).mul_add((*x as f32) - latency_avg, acc) + }) / (sample_size as f32); + + Some(Statistics { + latency_avg, + latency_var, + sample_size, + latest_block_id, + errors, + }) +} + +pub async fn actualize_client(client: &SequencerClient) -> StatisticsUpdate { + let (latency, block_id) = measure_request_duration(client).await; + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let latency = latency as f32; + + StatisticsUpdate { + latency, + new_latest_block_id: block_id, + is_failed: block_id.is_none(), + } +} + +#[must_use] +pub fn choose_leader( + client_list: &HashMap, + statistics: &HashMap, +) -> Option<(Url, SequencerClient)> { + // Sort out all unmetered clients + let mut client_vec: Vec<_> = client_list + .keys() + .filter(|item| statistics.contains_key(*item)) + .collect(); + + if client_vec.is_empty() { + return None; + } + + // Considering the nature of our requests, the latest_block_id is the dominant characteristic + let max_block_id_addr = client_vec.iter().fold(client_vec[0], |acc, x| { + let old_latest_block_id = statistics.get(acc).unwrap().latest_block_id; + let new_latest_block_id = statistics.get(*x).unwrap().latest_block_id; + if new_latest_block_id > old_latest_block_id { + *x + } else { + acc + } + }); + + let max_block_id = statistics.get(max_block_id_addr).unwrap().latest_block_id; + + // Sort out all clients running late + client_vec = client_vec + .iter() + .filter_map(|x| { + let latest_block_id = statistics.get(*x).unwrap().latest_block_id; + + (latest_block_id == max_block_id).then_some(*x) + }) + .collect(); + + // Get the clients with lesser or equal to average error ratio + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let avg_err_ratio = client_vec.iter().fold(0_f32, |acc, x| { + acc + error_ratio( + statistics.get(*x).unwrap().errors, + statistics.get(*x).unwrap().sample_size, + ) + }) / (client_vec.len() as f32); + + client_vec.sort_by(|a, b| { + let err_ratio_a = error_ratio( + statistics.get(*a).unwrap().errors, + statistics.get(*a).unwrap().sample_size, + ); + let err_ratio_b = error_ratio( + statistics.get(*b).unwrap().errors, + statistics.get(*b).unwrap().sample_size, + ); + + err_ratio_a + .partial_cmp(&err_ratio_b) + .expect("Ratios must be a valid numbers") + }); + + let client_vec = client_vec[..(client_vec + .iter() + .position(|item| { + error_ratio( + statistics.get(*item).unwrap().errors, + statistics.get(*item).unwrap().sample_size, + ) > avg_err_ratio + }) + .unwrap_or(client_vec.len()))] + .to_vec(); + + // Choose clients with least latency and variance + let min_lat_var_addr = client_vec.iter().fold(client_vec[0], |acc, x| { + let old = statistics.get(acc).unwrap(); + let (old_lat, old_var) = (old.latency_avg, old.latency_var); + let new = statistics.get(*x).unwrap(); + let (new_lat, new_var) = (new.latency_avg, new.latency_var); + + let new_std = new_var.sqrt(); + let old_std = old_var.sqrt(); + + // Client is better if its average is better and variance does not make it worse + // So basically we want this: + // [-old_std............new_lat.......old_lat...............+new_std.........+old_std] + // + // However one can argue that this: + // + // [-old_std...................old_lat........new_lat.........+new_std.......+old_std] + // + // is still better, but it is up to discussion + if (new_lat <= old_lat) && ((new_lat + new_std) < (old_lat + old_std)) { + *x + } else { + acc + } + }); + + Some(( + min_lat_var_addr.clone(), + client_list.get(min_lat_var_addr).unwrap().clone(), + )) +} + +/// Helperfunction to calculate cumulative average. +/// +/// Cumulative average calculation is the following problem: +/// +/// We want to calculate avarage of a sample of size `N + N_1` +/// where average for `N` is known. +/// +/// To do so we need: +/// - old average value +/// - sum_{`i=1}^{N_1}{n_i`} +/// - `N` +/// - `N_1` +fn cumulative_avg( + latency_avg_old: f32, + cumulative_latency: f32, + orig_size_f: f32, + mod_size_f: f32, +) -> f32 { + latency_avg_old.mul_add(orig_size_f, cumulative_latency) / (orig_size_f + mod_size_f) +} + +/// Helperfunction to calculate cumulative variance. +/// +/// Cumulative variance calculation is the following problem: +/// +/// We want to calculate variance of a sample of size `N + N_1` +/// where average for `N` is known. +/// +/// To do so we need: +/// - old average value +/// - new average value +/// - old variance +/// - sum_{`i=1}^{N_1}{n_i`} +/// - sum_{`i=1}^{N_1}{n_i^2`} +/// - `N` +/// - `N_1` +fn cumulative_var( + latency_avg_old: f32, + latency_avg_new: f32, + latency_var: f32, + cumulative_latency: f32, + cumulative_latency_squares: f32, + orig_size_f: f32, + mod_size_f: f32, +) -> f32 { + // The formula was atrocious before. + // `mul_add` function have less precision loss with drawback of being absolutely unreadable + ((2_f32 * cumulative_latency).mul_add( + -latency_avg_new, + mod_size_f.mul_add( + latency_avg_new * latency_avg_new, + latency_var.mul_add( + orig_size_f, + (latency_avg_new - latency_avg_old) + * (latency_avg_new - latency_avg_old) + * orig_size_f, + ), + ), + ) + cumulative_latency_squares) + / (orig_size_f + mod_size_f) +} + +fn error_ratio(errors: u64, size: usize) -> f32 { + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let errors_f = errors as f32; + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let size_f = size as f32; + + errors_f / (errors_f + size_f) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use sequencer_service_rpc::{SequencerClient, SequencerClientBuilder}; + use url::Url; + + use crate::multi_client::{ + CumulativeUpdates, Statistics, StatisticsUpdate, choose_leader, cumulative_avg, + cumulative_var, + }; + + fn update_statistics( + statistics: &mut HashMap, + leader_url: &Url, + metric_updates: &[StatisticsUpdate], + ) -> Result<(), anyhow::Error> { + let leader_metric = statistics + .get_mut(leader_url) + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in statistics"))?; + + leader_metric.apply_updates(metric_updates); + + Ok(()) + } + + fn client_from_url_unchecked(url: &Url) -> SequencerClient { + let builder = SequencerClientBuilder::default(); + builder.build(url).unwrap() + } + + #[test] + fn cumulative_updates_test() { + let statistics_updates_vec = vec![ + StatisticsUpdate { + latency: 100_f32, + new_latest_block_id: Some(15), + is_failed: false, + }, + StatisticsUpdate { + latency: 115_f32, + new_latest_block_id: Some(16), + is_failed: false, + }, + StatisticsUpdate { + latency: 50_f32, + new_latest_block_id: None, + is_failed: true, + }, + ]; + + let CumulativeUpdates { + failure_count, + latest_block_id, + cumulative_latency, + cumulative_latency_squares, + additional_sample_size, + } = CumulativeUpdates::from_metric_updates(&statistics_updates_vec); + + let epsilon = 0.01_f32; + + let sum_squared_manual = 100_f32.mul_add(100_f32, 115_f32 * 115_f32); + + assert_eq!(additional_sample_size, 2); + assert_eq!(failure_count, 1); + assert_eq!(latest_block_id, Some(16)); + assert!((cumulative_latency - 215_f32).abs() < epsilon); + assert!((cumulative_latency_squares - sum_squared_manual).abs() < epsilon); + } + + #[test] + fn cumulative_avg_test() { + let mut sample = vec![100_f32; 40]; + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let old_sample_size_f = sample.len() as f32; + + let old_avg = sample.iter().sum::() / old_sample_size_f; + + let new_samples = vec![ + 101_f32, 110_f32, 112_f32, 97_f32, 78_f32, 25_f32, 75_f32, 189_f32, 120_f32, 50_f32, + ]; + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let mod_sample_size_f = new_samples.len() as f32; + + let cumulative = new_samples.iter().sum(); + + sample.extend_from_slice(&new_samples); + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let new_sample_size_f = sample.len() as f32; + + let new_avg_1 = sample.iter().sum::() / new_sample_size_f; + + let new_avg_2 = cumulative_avg(old_avg, cumulative, old_sample_size_f, mod_sample_size_f); + + let epsilon = 0.01_f32; + + assert!((new_avg_1 - new_avg_2).abs() < epsilon); + } + + #[test] + fn cumulative_var_test() { + let mut sample = vec![100_f32; 40]; + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let old_sample_size_f = sample.len() as f32; + let old_avg = sample.iter().sum::() / old_sample_size_f; + + let old_var = sample + .iter() + .fold(0_f32, |acc, x| (x - old_avg).mul_add(x - old_avg, acc)) + / old_sample_size_f; + + let new_samples = vec![ + 101_f32, 110_f32, 112_f32, 97_f32, 78_f32, 25_f32, 75_f32, 189_f32, 120_f32, 50_f32, + ]; + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let mod_sample_size_f = new_samples.len() as f32; + + let cumulative = new_samples.iter().sum(); + let cumulative_squares = new_samples + .iter() + .fold(0_f32, |acc, x| (*x).mul_add(*x, acc)); + + let new_avg = cumulative_avg(old_avg, cumulative, old_sample_size_f, mod_sample_size_f); + + sample.extend_from_slice(&new_samples); + + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let new_var_1 = sample + .iter() + .fold(0_f32, |acc, x| (x - new_avg).mul_add(x - new_avg, acc)) + / (sample.len() as f32); + + let new_var_2 = cumulative_var( + old_avg, + new_avg, + old_var, + cumulative, + cumulative_squares, + old_sample_size_f, + mod_sample_size_f, + ); + + let epsilon = 0.01_f32; + + assert!((new_var_1 - new_var_2).abs() < epsilon); + } + + #[test] + fn metric_updates_correctness() { + let statistics_updates_vec = vec![ + StatisticsUpdate { + latency: 100_f32, + new_latest_block_id: Some(105), + is_failed: false, + }, + StatisticsUpdate { + latency: 115_f32, + new_latest_block_id: Some(106), + is_failed: false, + }, + StatisticsUpdate { + latency: 50_f32, + new_latest_block_id: None, + is_failed: true, + }, + ]; + + let addr_leader = Url::parse("https://127.0.0.1:3040").unwrap(); + + let leader_statistics = Statistics { + latency_avg: 100_f32, + latency_var: 25_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }; + + let cumulative_latency = 100_f32 + 115_f32; + let cumulative_latency_squares = 100_f32.mul_add(100_f32, 115_f32 * 115_f32); + + let avg_manual = cumulative_avg( + leader_statistics.latency_avg, + cumulative_latency, + 10_f32, + 2_f32, + ); + let var_manual = cumulative_var( + leader_statistics.latency_avg, + avg_manual, + leader_statistics.latency_var, + cumulative_latency, + cumulative_latency_squares, + 10_f32, + 2_f32, + ); + + let mut metric_map = HashMap::new(); + metric_map.insert(addr_leader.clone(), leader_statistics); + + update_statistics(&mut metric_map, &addr_leader, &statistics_updates_vec).unwrap(); + + let Statistics { + latency_avg, + latency_var, + sample_size, + latest_block_id, + errors, + } = metric_map[&addr_leader]; + + let epsilon = 0.01_f32; + + assert_eq!(errors, 6); + assert_eq!(latest_block_id, 106); + assert_eq!(sample_size, 12); + assert!((latency_avg - avg_manual).abs() < epsilon); + assert!((latency_var - var_manual).abs() < epsilon); + } + + #[test] + fn choose_leader_latest_block() { + let addr_leader = Url::parse("http://127.0.0.1:3040").unwrap(); + let addr_1 = Url::parse("http://127.0.0.1:3041").unwrap(); + let addr_2 = Url::parse("http://127.0.0.1:3042").unwrap(); + let addr_3 = Url::parse("http://127.0.0.1:3043").unwrap(); + + let leader = client_from_url_unchecked(&addr_leader); + let client_1 = client_from_url_unchecked(&addr_1); + let client_2 = client_from_url_unchecked(&addr_2); + let client_3 = client_from_url_unchecked(&addr_3); + + let mut client_list = HashMap::new(); + + client_list.insert(addr_leader.clone(), leader); + client_list.insert(addr_1.clone(), client_1); + client_list.insert(addr_2.clone(), client_2); + client_list.insert(addr_3.clone(), client_3); + + let mut statistics = HashMap::new(); + + statistics.insert( + addr_3, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 97, + errors: 5, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 98, + errors: 5, + }, + ); + + statistics.insert( + addr_1, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 99, + errors: 5, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + + assert_eq!(leader_url, addr_leader); + } + + #[test] + fn choose_leader_least_errors() { + let addr_leader = Url::parse("http://127.0.0.1:3040").unwrap(); + let addr_1 = Url::parse("http://127.0.0.1:3041").unwrap(); + let addr_2 = Url::parse("http://127.0.0.1:3042").unwrap(); + let addr_3 = Url::parse("http://127.0.0.1:3043").unwrap(); + + let leader = client_from_url_unchecked(&addr_leader); + let client_1 = client_from_url_unchecked(&addr_1); + let client_2 = client_from_url_unchecked(&addr_2); + let client_3 = client_from_url_unchecked(&addr_3); + + let mut client_list = HashMap::new(); + + client_list.insert(addr_leader.clone(), leader); + client_list.insert(addr_1.clone(), client_1); + client_list.insert(addr_2.clone(), client_2); + client_list.insert(addr_3.clone(), client_3); + + let mut statistics = HashMap::new(); + + statistics.insert( + addr_3, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + statistics.insert( + addr_1, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 3, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 2, + }, + ); + + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + + assert_eq!(leader_url, addr_leader); + } + + #[test] + fn choose_leader_simple_latency_check() { + let addr_leader = Url::parse("http://127.0.0.1:3040").unwrap(); + let addr_1 = Url::parse("http://127.0.0.1:3041").unwrap(); + let addr_2 = Url::parse("http://127.0.0.1:3042").unwrap(); + let addr_3 = Url::parse("http://127.0.0.1:3043").unwrap(); + + let leader = client_from_url_unchecked(&addr_leader); + let client_1 = client_from_url_unchecked(&addr_1); + let client_2 = client_from_url_unchecked(&addr_2); + let client_3 = client_from_url_unchecked(&addr_3); + + let mut client_list = HashMap::new(); + + client_list.insert(addr_leader.clone(), leader); + client_list.insert(addr_1.clone(), client_1); + client_list.insert(addr_2.clone(), client_2); + client_list.insert(addr_3.clone(), client_3); + + let mut statistics = HashMap::new(); + + statistics.insert( + addr_3, + Statistics { + latency_avg: 103_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 102_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_1, + Statistics { + latency_avg: 101_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + + assert_eq!(leader_url, addr_leader); + } + + #[test] + fn choose_leader_latency_var_check() { + let addr_leader = Url::parse("http://127.0.0.1:3040").unwrap(); + let addr_1 = Url::parse("http://127.0.0.1:3041").unwrap(); + let addr_2 = Url::parse("http://127.0.0.1:3042").unwrap(); + let addr_3 = Url::parse("http://127.0.0.1:3043").unwrap(); + + let leader = client_from_url_unchecked(&addr_leader); + let client_1 = client_from_url_unchecked(&addr_1); + let client_2 = client_from_url_unchecked(&addr_2); + let client_3 = client_from_url_unchecked(&addr_3); + + let mut client_list = HashMap::new(); + + client_list.insert(addr_leader.clone(), leader); + client_list.insert(addr_1.clone(), client_1); + client_list.insert(addr_2.clone(), client_2); + client_list.insert(addr_3.clone(), client_3); + + let mut statistics = HashMap::new(); + + statistics.insert( + addr_3, + Statistics { + latency_avg: 100_f32, + latency_var: 13_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 100_f32, + latency_var: 12_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_1, + Statistics { + latency_avg: 100_f32, + latency_var: 11_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 5, + }, + ); + + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + + assert_eq!(leader_url, addr_leader); + } +} diff --git a/test_fixtures/fixtures/prebuilt_sequencer_db.dump b/test_fixtures/fixtures/prebuilt_sequencer_db.dump index 2aa0c51f..a926c65e 100644 Binary files a/test_fixtures/fixtures/prebuilt_sequencer_db.dump and b/test_fixtures/fixtures/prebuilt_sequencer_db.dump differ diff --git a/test_fixtures/src/bin/regenerate_test_fixture.rs b/test_fixtures/src/bin/regenerate_test_fixture.rs index 80da8fdc..2aa24c8b 100644 --- a/test_fixtures/src/bin/regenerate_test_fixture.rs +++ b/test_fixtures/src/bin/regenerate_test_fixture.rs @@ -65,6 +65,7 @@ async fn generate_prebuilt_fixture(dest: &Path) -> Result<()> { &initial_private_accounts, WalletConfigOverrides::default(), ) + .await .context("Failed to setup wallet for fixture generation")?; setup_public_accounts_with_initial_supply(&mut wallet, &initial_public_accounts) diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index 960f349b..449308eb 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -8,7 +8,7 @@ use lee::{AccountId, PrivateKey, PublicKey}; use lee_core::Identifier; use sequencer_core::config::{BedrockConfig, CrossZoneConfig, GenesisAction, SequencerConfig}; use url::Url; -use wallet::config::WalletConfig; +use wallet::config::{SequencerConnectionData, WalletConfig}; pub const INITIAL_PUBLIC_BALANCES_FOR_WALLET: [u128; 2] = [10_000, 20_000]; pub const INITIAL_PRIVATE_BALANCES_FOR_WALLET: [u128; 2] = [10_000, 20_000]; @@ -194,13 +194,16 @@ pub fn genesis_from_accounts( pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { Ok(WalletConfig { - sequencer_addr: addr_to_url(UrlProtocol::Http, sequencer_addr) - .context("Failed to convert sequencer addr to URL")?, + sequencers: vec![SequencerConnectionData { + sequencer_addr: addr_to_url(UrlProtocol::Http, sequencer_addr) + .context("Failed to convert sequencer addr to URL")?, + basic_auth: None, + }], seq_poll_timeout: Duration::from_secs(30), seq_tx_poll_max_blocks: 15, seq_poll_max_retries: 10, seq_block_poll_max_amount: 100, - basic_auth: None, + calibration_limit: 5, }) } diff --git a/test_fixtures/src/lib.rs b/test_fixtures/src/lib.rs index 687da077..8aa201a7 100644 --- a/test_fixtures/src/lib.rs +++ b/test_fixtures/src/lib.rs @@ -393,6 +393,7 @@ impl TestContextBuilder { &initial_private_accounts, wallet_config_overrides, ) + .await .context("Failed to setup wallet")?; if use_prebuilt { @@ -456,6 +457,10 @@ impl BlockingTestContext { self.ctx.as_ref().expect("TestContext is set") } + pub const fn ctx_mut(&mut self) -> &mut TestContext { + self.ctx.as_mut().expect("TestContext is set") + } + pub const fn runtime(&self) -> &tokio::runtime::Runtime { &self.runtime } diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index 325e1628..08ce6e88 100644 --- a/test_fixtures/src/setup.rs +++ b/test_fixtures/src/setup.rs @@ -214,7 +214,7 @@ async fn setup_sequencer_inner( Ok((sequencer_handle, temp_sequencer_dir)) } -pub fn setup_wallet( +pub async fn setup_wallet( sequencer_addr: SocketAddr, initial_public_accounts: &[(PrivateKey, u128)], initial_private_accounts: &[InitialPrivateAccountForWallet], @@ -232,14 +232,17 @@ pub fn setup_wallet( .context("Failed to write wallet config in temp dir")?; let storage_path = temp_wallet_dir.path().join("storage.json"); + let metrics_path = temp_wallet_dir.path().join("metrics.json"); let wallet_password = "test_pass".to_owned(); let (mut wallet, _mnemonic) = WalletCore::new_init_storage( config_path, storage_path, + metrics_path, Some(config_overrides), &wallet_password, ) + .await .context("Failed to init wallet")?; for (private_key, _balance) in initial_public_accounts {