From bf237daf711fcab3a05024ac5853cac1e998dab2 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Tue, 14 Jul 2026 13:10:32 +0300 Subject: [PATCH] fix(tintegration_tests): tests work --- .../tests/auth_transfer/private.rs | 6 +- integration_tests/tests/private_pda.rs | 8 +- integration_tests/tests/wallet_ffi.rs | 18 +- lee/state_machine/src/state/mod.rs | 2 +- lee/state_machine/src/state/tests/claiming.rs | 6 +- .../src/state/tests/privacy_preserving.rs | 2 +- lez/wallet-ffi/src/vault.rs | 8 +- lez/wallet-ffi/src/wallet.rs | 7 +- lez/wallet/src/cli/account.rs | 7 +- lez/wallet/src/cli/mod.rs | 7 + lez/wallet/src/cli/programs/amm.rs | 10 +- lez/wallet/src/cli/programs/ata.rs | 2 +- .../src/cli/programs/native_token_transfer.rs | 4 +- lez/wallet/src/cli/programs/pinata.rs | 6 +- lez/wallet/src/cli/programs/token.rs | 8 +- lez/wallet/src/config.rs | 2 +- lez/wallet/src/lib.rs | 55 ++- lez/wallet/src/multi_client.rs | 371 ++++++++++-------- test_fixtures/src/lib.rs | 2 +- 19 files changed, 300 insertions(+), 231 deletions(-) diff --git a/integration_tests/tests/auth_transfer/private.rs b/integration_tests/tests/auth_transfer/private.rs index af016dc6..fb2fde5b 100644 --- a/integration_tests/tests/auth_transfer/private.rs +++ b/integration_tests/tests/auth_transfer/private.rs @@ -249,11 +249,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/private_pda.rs b/integration_tests/tests/private_pda.rs index 0d74fdfe..0b187283 100644 --- a/integration_tests/tests/private_pda.rs +++ b/integration_tests/tests/private_pda.rs @@ -37,7 +37,7 @@ use wallet::{AccountIdentity, WalletCore}; reason = "test helper — grouping args would obscure intent" )] async fn fund_private_pda( - wallet: &mut WalletCore, + wallet: &WalletCore, sender: AccountId, npk: NullifierPublicKey, vpk: ViewingPublicKey, @@ -107,7 +107,7 @@ async fn fund_private_pda( reason = "test helper — grouping args would obscure intent" )] async fn spend_private_pda( - wallet: &mut WalletCore, + wallet: &WalletCore, pda_account_id: AccountId, recipient_npk: NullifierPublicKey, recipient_vpk: ViewingPublicKey, @@ -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!( diff --git a/integration_tests/tests/wallet_ffi.rs b/integration_tests/tests/wallet_ffi.rs index 321c874d..3b8ea05e 100644 --- a/integration_tests/tests/wallet_ffi.rs +++ b/integration_tests/tests/wallet_ffi.rs @@ -43,12 +43,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 +293,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 +310,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(), ) }; @@ -370,14 +375,17 @@ fn new_wallet_ffi_with_default_config(password: &str) -> Result Result 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] diff --git a/lee/state_machine/src/state/mod.rs b/lee/state_machine/src/state/mod.rs index fcf71a21..df1c6343 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/src/vault.rs b/lez/wallet-ffi/src/vault.rs index e4d9dd6a..6db1fa66 100644 --- a/lez/wallet-ffi/src/vault.rs +++ b/lez/wallet-ffi/src/vault.rs @@ -113,7 +113,7 @@ pub unsafe extern "C" fn wallet_ffi_vault_claim( return WalletFfiError::NullPointer; } - let mut wallet = match wrapper.core.lock() { + let wallet = match wrapper.core.lock() { Ok(w) => w, Err(e) => { print_error(format!("Failed to lock wallet: {e}")); @@ -124,7 +124,7 @@ pub unsafe extern "C" fn wallet_ffi_vault_claim( let owner_id = AccountId::new(unsafe { (*owner).data }); let amount = u128::from_le_bytes(unsafe { *amount }); - match block_on(Vault(&mut wallet).send_claim(owner_id, amount)) { + match block_on(Vault(&wallet).send_claim(owner_id, amount)) { Ok(tx_hash) => { let tx_hash = CString::new(tx_hash.to_string()) .map_or(ptr::null_mut(), std::ffi::CString::into_raw); @@ -185,7 +185,7 @@ pub unsafe extern "C" fn wallet_ffi_vault_claim_private( return WalletFfiError::NullPointer; } - let mut wallet = match wrapper.core.lock() { + let wallet = match wrapper.core.lock() { Ok(w) => w, Err(e) => { print_error(format!("Failed to lock wallet: {e}")); @@ -196,7 +196,7 @@ pub unsafe extern "C" fn wallet_ffi_vault_claim_private( let owner_id = AccountId::new(unsafe { (*owner).data }); let amount = u128::from_le_bytes(unsafe { *amount }); - match block_on(Vault(&mut wallet).send_claim_private_owner(owner_id, amount)) { + match block_on(Vault(&wallet).send_claim_private_owner(owner_id, amount)) { Ok((tx_hash, _shared_key)) => { 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/wallet.rs b/lez/wallet-ffi/src/wallet.rs index c6c58987..703da3a5 100644 --- a/lez/wallet-ffi/src/wallet.rs +++ b/lez/wallet-ffi/src/wallet.rs @@ -240,7 +240,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}")); @@ -248,7 +248,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.store_metrics())) + { Ok(()) => WalletFfiError::Success, Err(e) => { print_error(format!("Failed to save wallet: {e}")); diff --git a/lez/wallet/src/cli/account.rs b/lez/wallet/src/cli/account.rs index 21e35c17..c700d2e9 100644 --- a/lez/wallet/src/cli/account.rs +++ b/lez/wallet/src/cli/account.rs @@ -314,7 +314,7 @@ impl AccountSubcommand { raw: bool, keys: bool, account_id: CliAccountMention, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let resolved = account_id.resolve(wallet_core.storage())?; wallet_core @@ -403,10 +403,7 @@ impl AccountSubcommand { } } - async fn handle_list( - long: bool, - wallet_core: &mut WalletCore, - ) -> Result { + 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(); diff --git a/lez/wallet/src/cli/mod.rs b/lez/wallet/src/cli/mod.rs index b3eedb5a..d01cd860 100644 --- a/lez/wallet/src/cli/mod.rs +++ b/lez/wallet/src/cli/mod.rs @@ -295,6 +295,13 @@ pub async fn execute_subcommand( } }; + // Kind of a sledgehammer solution, but it is not clear if there is the case to not store + // metrics + wallet_core + .store_metrics() + .await + .context("Failed to store metrics")?; + Ok(subcommand_ret) } diff --git a/lez/wallet/src/cli/programs/amm.rs b/lez/wallet/src/cli/programs/amm.rs index b95746ce..0f544c42 100644 --- a/lez/wallet/src/cli/programs/amm.rs +++ b/lez/wallet/src/cli/programs/amm.rs @@ -125,7 +125,7 @@ impl AmmProgramAgnosticSubcommand { user_holding_lp: CliAccountMention, balance_a: u128, balance_b: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let a_id = user_holding_a.resolve(wallet_core.storage())?; let b_id = user_holding_b.resolve(wallet_core.storage())?; @@ -162,7 +162,7 @@ impl AmmProgramAgnosticSubcommand { amount_in: u128, min_amount_out: u128, token_definition: AccountId, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let a_id = user_holding_a.resolve(wallet_core.storage())?; let b_id = user_holding_b.resolve(wallet_core.storage())?; @@ -194,7 +194,7 @@ impl AmmProgramAgnosticSubcommand { exact_amount_out: u128, max_amount_in: u128, token_definition: AccountId, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let a_id = user_holding_a.resolve(wallet_core.storage())?; let b_id = user_holding_b.resolve(wallet_core.storage())?; @@ -227,7 +227,7 @@ impl AmmProgramAgnosticSubcommand { min_amount_lp: u128, max_amount_a: u128, max_amount_b: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let a_id = user_holding_a.resolve(wallet_core.storage())?; let b_id = user_holding_b.resolve(wallet_core.storage())?; @@ -266,7 +266,7 @@ impl AmmProgramAgnosticSubcommand { balance_lp: u128, min_amount_a: u128, min_amount_b: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let a_id = user_holding_a.resolve(wallet_core.storage())?; let b_id = user_holding_b.resolve(wallet_core.storage())?; diff --git a/lez/wallet/src/cli/programs/ata.rs b/lez/wallet/src/cli/programs/ata.rs index 044ddb75..1d207b35 100644 --- a/lez/wallet/src/cli/programs/ata.rs +++ b/lez/wallet/src/cli/programs/ata.rs @@ -186,7 +186,7 @@ impl AtaSubcommand { async fn handle_list( owner: AccountId, token_definition: Vec, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let ata_program_id = programs::ata().id(); diff --git a/lez/wallet/src/cli/programs/native_token_transfer.rs b/lez/wallet/src/cli/programs/native_token_transfer.rs index 514fd0e6..aaa3efe1 100644 --- a/lez/wallet/src/cli/programs/native_token_transfer.rs +++ b/lez/wallet/src/cli/programs/native_token_transfer.rs @@ -425,7 +425,7 @@ impl NativeTokenTransferProgramSubcommandShielded { to_vpk: String, to_identifier: Option, amount: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let (to_npk, to_vpk) = crate::cli::decode_npk_vpk(&to_npk, &to_vpk)?; @@ -497,7 +497,7 @@ impl NativeTokenTransferProgramSubcommand { from: Option, to: Option, amount: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let tx_hash = NativeTokenTransfer(wallet_core) .send_public_transfer( diff --git a/lez/wallet/src/cli/programs/pinata.rs b/lez/wallet/src/cli/programs/pinata.rs index dd26f7e6..f0f49c5e 100644 --- a/lez/wallet/src/cli/programs/pinata.rs +++ b/lez/wallet/src/cli/programs/pinata.rs @@ -167,7 +167,7 @@ impl WalletSubcommand for PinataProgramSubcommand { } async fn ensure_public_recipient_initialized( - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, winner_account_id: AccountId, ) -> Result<()> { let account = wallet_core @@ -187,7 +187,7 @@ async fn ensure_public_recipient_initialized( } fn ensure_private_owned_recipient_initialized( - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, winner_account_id: AccountId, ) -> Result<()> { let Some(account) = wallet_core.get_account_private(winner_account_id) else { @@ -210,7 +210,7 @@ fn ensure_private_owned_recipient_initialized( Ok(()) } -async fn find_solution(wallet: &mut WalletCore, pinata_account_id: AccountId) -> Result { +async fn find_solution(wallet: &WalletCore, pinata_account_id: AccountId) -> Result { let account = wallet.get_account_public(pinata_account_id).await?; let data: [u8; 33] = account .data diff --git a/lez/wallet/src/cli/programs/token.rs b/lez/wallet/src/cli/programs/token.rs index 4686909d..6486f58c 100644 --- a/lez/wallet/src/cli/programs/token.rs +++ b/lez/wallet/src/cli/programs/token.rs @@ -817,7 +817,7 @@ impl TokenProgramSubcommandPublic { sender_account_id: CliAccountMention, recipient_account_id: CliAccountMention, balance_to_move: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let sender = sender_account_id.resolve(wallet_core.storage())?; let recipient = recipient_account_id.resolve(wallet_core.storage())?; @@ -844,7 +844,7 @@ impl TokenProgramSubcommandPublic { definition_account_id: AccountId, holder_account_id: CliAccountMention, amount: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let holder = holder_account_id.resolve(wallet_core.storage())?; let AccountIdWithPrivacy::Public(holder_id) = holder else { @@ -868,7 +868,7 @@ impl TokenProgramSubcommandPublic { definition_account_id: CliAccountMention, holder_account_id: CliAccountMention, amount: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let definition = definition_account_id.resolve(wallet_core.storage())?; let holder = holder_account_id.resolve(wallet_core.storage())?; @@ -1535,7 +1535,7 @@ impl CreateNewTokenProgramSubcommand { supply_account_id: CliAccountMention, name: String, total_supply: u128, - wallet_core: &mut WalletCore, + wallet_core: &WalletCore, ) -> Result { let definition = definition_account_id.resolve(wallet_core.storage())?; let supply = supply_account_id.resolve(wallet_core.storage())?; diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 319bde4f..0063c06d 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -11,7 +11,7 @@ use url::Url; pub struct SequencerConnectionData { /// Connection data of all known sequencers. pub sequencer_addr: Url, - /// Basic authentication credentials + /// Basic authentication credentials. #[serde(skip_serializing_if = "Option::is_none")] pub basic_auth: Option, } diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index c9c3a04e..bc65d5c6 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -38,10 +38,7 @@ use url::Url; use crate::{ account::{AccountIdWithPrivacy, Label}, config::WalletConfigOverrides, - multi_client::{ - Metrics, MetricsUpdate, MultiSequencerClient, extract_metrics_from_path, - save_metrics_at_path_with_updates, - }, + multi_client::{Metrics, MetricsUpdate, MultiSequencerClient, extract_metrics_from_path}, poller::TxPoller, storage::key_chain::SharedAccountEntry, }; @@ -93,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, @@ -104,9 +100,9 @@ pub struct WalletCore { metrics_path: PathBuf, metrics: HashMap, - /// Wrapping metric updates in Arc to not break intefaces too much. + /// Wrapping metric updates in Arc to not break interfaces too much. /// - /// It is assumed, that wallet methods can be accesed via immutable reference + /// It is assumed, that wallet methods can be accesed via immutable reference. metric_updates: Arc>>, multi_sequencer_client: MultiSequencerClient, @@ -211,14 +207,17 @@ 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() } @@ -259,21 +258,38 @@ impl WalletCore { Ok(()) } - pub async fn store_metrics(&self) -> Result<()> { + async fn update_metrics(&mut self) -> Result<()> { + let leader_metric = self + .metrics + .get_mut(&self.multi_sequencer_client.leader_url) + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in metrics"))?; + { let metric_updates = self.metric_updates.read().await; - save_metrics_at_path_with_updates( - self.metrics.clone(), - &self.multi_sequencer_client.leader_url, - &metric_updates, - &self.storage_path, - ) - .await - .with_context(|| { - format!("Failed to store metrics at {}", self.storage_path.display()) - })?; + leader_metric.apply_updates(metric_updates.as_ref()); } + // Clear updates + { + let mut metric_updates = self.metric_updates.write().await; + metric_updates.clear(); + } + + Ok(()) + } + + pub async fn store_metrics(&mut self) -> Result<()> { + self.update_metrics().await?; + + let metrics_serialized = serde_json::to_vec_pretty(&self.metrics)?; + let mut file = tokio::fs::File::create(&self.metrics_path) + .await + .context("Failed to create file")?; + file.write_all(&metrics_serialized) + .await + .context("Failed to write to file")?; + file.sync_all().await.context("Failed to sync file")?; + println!("Stored metrics at {}", self.metrics_path.display()); Ok(()) @@ -487,6 +503,7 @@ impl WalletCore { }) } + #[must_use] pub fn get_metrics(&self, sequencer_url: &Url) -> Option<&Metrics> { self.metrics.get(sequencer_url) } @@ -686,7 +703,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, diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 461898a3..f4b01e21 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -1,27 +1,21 @@ +#![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}; -use anyhow::{Context, Result}; -use sequencer_service_rpc::{RpcClient, SequencerClient, SequencerClientBuilder}; +use anyhow::{Context as _, Result}; +use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; use serde::{Deserialize, Serialize}; -use tokio::io::AsyncWriteExt; use url::Url; use crate::config::SequencerConnectionData; -pub fn extract_metrics_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!("Metrics not found, choosing empty"); - Ok(HashMap::new()) - } - Err(err) => Err(err).context("IO error"), - } -} - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Metrics { pub latency_avg: f32, @@ -39,21 +33,23 @@ pub struct MetricsUpdate { } impl Metrics { - fn apply_updates(&mut self, updates: &[MetricsUpdate]) { + pub fn apply_updates(&mut self, updates: &[MetricsUpdate]) { let CumulativeUpdates { failure_count, latest_block_id, cumulative_latency, cumulative_latency_squares, additional_sample_size, - } = cumulative_updates(updates); + } = CumulativeUpdates::from_metric_updates(updates); - self.errors += failure_count; + 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; @@ -72,7 +68,7 @@ impl Metrics { self.latency_avg = latency_avg_new; self.latency_var = latency_var_new; - self.sample_size += additional_sample_size; + self.sample_size = self.sample_size.saturating_add(additional_sample_size); } } @@ -117,32 +113,41 @@ impl MultiSequencerClient { }; // If there is no metrics for client, callibrate it - if !metrics.contains_key(sequencer_addr) { + if metrics.contains_key(sequencer_addr) { + let metric_updates = actualize_client(&sequencer_client).await; + + log::debug!( + "Metered call for {sequencer_addr:?}, metric updates is {metric_updates:?}" + ); + + let metric_mut = metrics.get_mut(sequencer_addr).unwrap(); + metric_mut.apply_updates(&[metric_updates]); + // Otherwise actualize client data + } else { metrics.insert( sequencer_addr.clone(), callibrate_client(&sequencer_client, callibration_limit).await, ); - // Otherwise actualize client data - } else { - let metric_updates = actualize_client(&sequencer_client).await; - let metric_mut = metrics.get_mut(sequencer_addr).unwrap(); - metric_mut.apply_updates(&[metric_updates]); } client_list.insert(sequencer_addr.clone(), sequencer_client); } - let (leader_url, leader) = - choose_leader(&client_list, metrics).ok_or(anyhow::anyhow!("Failed to find leader"))?; + let (leader_url, leader) = choose_leader(&client_list, metrics) + .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; + + log::info!("Chosen leader is {leader_url:?}"); // Dropping client list, for reasons why, see comment in structure definition. Ok(Self { leader, leader_url }) } - pub fn leader_ref(&self) -> &SequencerClient { + #[must_use] + pub const fn leader_ref(&self) -> &SequencerClient { &self.leader } + #[must_use] pub fn leader_clone(&self) -> SequencerClient { self.leader.clone() } @@ -152,14 +157,89 @@ impl MultiSequencerClient { &self, call: I, ) -> (Result, MetricsUpdate) { - tokio::join!(call(self.leader_ref()), actualize_client(self.leader_ref())) + let resp = tokio::join!(call(self.leader_ref()), actualize_client(self.leader_ref())); + + log::debug!( + "Metered call for {:?}, metric updates is {:?}", + self.leader_url, + resp.1 + ); + + resp + } +} + +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: &[MetricsUpdate]) -> 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 MetricsUpdate { + 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_metrics_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!("Metrics not found, choosing empty"); + Ok(HashMap::new()) + } + Err(err) => Err(err).context("IO error"), } } pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usize) -> Metrics { let mut latencies = vec![]; let mut latest_block_id = 0; - let mut errors = 0; + let mut errors: u64 = 0; // ToDo: Add some DDoS adaptation for _ in 0..callibration_limit { @@ -170,7 +250,7 @@ pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usi let latency = tokio::time::Instant::now().duration_since(now).as_millis(); let Ok(block_id) = block_id else { - errors += 1; + errors = errors.saturating_add(1); continue; }; @@ -180,9 +260,11 @@ pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usi // Precision loss if fine there let sample_size = latencies.len(); - let latency_avg = (latencies.iter().fold(0, |acc, x| acc + x) as f32) / (sample_size as f32); - let latency_var = latencies.iter().fold(0f32, |acc, x| { - acc + ((*x as f32) - latency_avg) * ((*x as f32) - latency_avg) + #[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); Metrics { @@ -199,6 +281,7 @@ pub async fn actualize_client(client: &SequencerClient) -> MetricsUpdate { let block_id = client.get_last_block_id().await.ok(); + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] let latency = tokio::time::Instant::now().duration_since(now).as_millis() as f32; MetricsUpdate { @@ -208,6 +291,7 @@ pub async fn actualize_client(client: &SequencerClient) -> MetricsUpdate { } } +#[must_use] pub fn choose_leader( client_list: &HashMap, metrics: &HashMap, @@ -215,11 +299,10 @@ pub fn choose_leader( let mut client_vec = vec![]; // Sort out all unmetered clients - for addr in client_list.keys() { - if metrics.contains_key(addr) { - client_vec.push(addr); - } - } + client_vec = client_list + .keys() + .filter(|item| metrics.contains_key(*item)) + .collect(); if client_vec.is_empty() { return None; @@ -244,19 +327,16 @@ pub fn choose_leader( .filter_map(|x| { let latest_block_id = metrics.get(*x).unwrap().latest_block_id; - if latest_block_id == max_block_id { - Some(*x) - } else { - None - } + (latest_block_id == max_block_id).then_some(*x) }) .collect(); // Get the clients with lesser or equal to average error count - let avg_err_count = client_vec - .iter() - .fold(0, |acc, x| acc + metrics.get(*x).unwrap().errors) - / (client_vec.len() as u64); + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let avg_err_count = (client_vec.iter().fold(0_u64, |acc, x| { + acc.saturating_add(metrics.get(*x).unwrap().errors) + }) as f32) + / (client_vec.len() as f32); client_vec.sort_by(|a, b| { metrics @@ -266,9 +346,10 @@ pub fn choose_leader( .cmp(&metrics.get(*b).unwrap().errors) }); - client_vec = client_vec[..(client_vec + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let client_vec = client_vec[..(client_vec .iter() - .position(|item| metrics.get(*item).unwrap().errors > avg_err_count) + .position(|item| (metrics.get(*item).unwrap().errors as f32) > avg_err_count) .unwrap_or(client_vec.len()))] .to_vec(); @@ -304,103 +385,42 @@ pub fn choose_leader( )) } -pub async fn save_metrics_at_path( - metrics: &HashMap, - path: &Path, -) -> Result<(), anyhow::Error> { - let metrics_serialized = serde_json::to_vec_pretty(metrics)?; - let mut file = tokio::fs::File::create(path) - .await - .context("Failed to create file")?; - file.write_all(&metrics_serialized) - .await - .context("Failed to write to file")?; - file.sync_all().await.context("Failed to sync file")?; - - Ok(()) -} - -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, -} - -fn cumulative_updates(metric_updates: &[MetricsUpdate]) -> CumulativeUpdates { - let (failure_count, latest_block_id, cumulative_latency, cumulative_latency_squares) = - metric_updates - .iter() - .fold((0u64, None, 0f32, 0_f32), |acc, x| { - let MetricsUpdate { - latency, - new_latest_block_id, - is_failed, - } = x; - ( - if *is_failed { acc.0 + 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 + latency } else { acc.2 }, - if !*is_failed { - acc.3 + latency * latency - } else { - acc.3 - }, - ) - }); - - CumulativeUpdates { - failure_count, - latest_block_id: latest_block_id.copied(), - cumulative_latency, - cumulative_latency_squares, - additional_sample_size: metric_updates.len() - (failure_count as usize), - } -} - -/// Helperfunction to calculate cumulative average +/// 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 +/// where average for `N` is known. /// /// To do so we need: /// - old average value -/// - sum_{i=1}^{N_1}{n_i} -/// - N -/// - N_1 +/// - 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 * orig_size_f + cumulative_latency) / (orig_size_f + mod_size_f) + latency_avg_old.mul_add(orig_size_f, cumulative_latency) / (orig_size_f + mod_size_f) } -/// Helperfunction to calculate cumulative variance +/// 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 +/// 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 +/// - 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, @@ -410,39 +430,23 @@ fn cumulative_var( orig_size_f: f32, mod_size_f: f32, ) -> f32 { - (latency_var * orig_size_f - + (latency_avg_new - latency_avg_old) * (latency_avg_new - latency_avg_old) * orig_size_f - + cumulative_latency_squares - + mod_size_f * (latency_avg_new * latency_avg_new) - - 2_f32 * cumulative_latency * latency_avg_new) + // 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) } -pub fn update_metrics( - metrics: &mut HashMap, - leader_url: &Url, - metric_updates: &[MetricsUpdate], -) -> Result<(), anyhow::Error> { - let leader_metric = metrics - .get_mut(leader_url) - .ok_or(anyhow::anyhow!("Leader URL is not present in metrics"))?; - - leader_metric.apply_updates(metric_updates); - - Ok(()) -} - -pub async fn save_metrics_at_path_with_updates( - mut metrics: HashMap, - leader_url: &Url, - metric_updates: &[MetricsUpdate], - path: &Path, -) -> Result<(), anyhow::Error> { - update_metrics(&mut metrics, leader_url, metric_updates)?; - - save_metrics_at_path(&metrics, path).await -} - #[cfg(test)] mod tests { use std::collections::HashMap; @@ -451,10 +455,23 @@ mod tests { use url::Url; use crate::multi_client::{ - CumulativeUpdates, Metrics, MetricsUpdate, choose_leader, cumulative_avg, - cumulative_updates, cumulative_var, update_metrics, + CumulativeUpdates, Metrics, MetricsUpdate, choose_leader, cumulative_avg, cumulative_var, }; + fn update_metrics( + metrics: &mut HashMap, + leader_url: &Url, + metric_updates: &[MetricsUpdate], + ) -> Result<(), anyhow::Error> { + let leader_metric = metrics + .get_mut(leader_url) + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in metrics"))?; + + leader_metric.apply_updates(metric_updates); + + Ok(()) + } + fn client_from_url_unchecked(url: &Url) -> SequencerClient { let builder = SequencerClientBuilder::default(); builder.build(url).unwrap() @@ -486,11 +503,11 @@ mod tests { cumulative_latency, cumulative_latency_squares, additional_sample_size, - } = cumulative_updates(&metrics_updates_vec); + } = CumulativeUpdates::from_metric_updates(&metrics_updates_vec); let epsilon = 0.01_f32; - let sum_squared_manual = 100_f32 * 100_f32 + 115_f32 * 115_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); @@ -503,17 +520,25 @@ mod tests { 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); @@ -527,27 +552,35 @@ mod tests { 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| acc + (x - old_avg) * (x - old_avg)) + .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| acc + (*x) * (*x)); + 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| acc + (x - new_avg) * (x - new_avg)) + .fold(0_f32, |acc, x| (x - new_avg).mul_add(x - new_avg, acc)) / (sample.len() as f32); let new_var_2 = cumulative_var( @@ -596,7 +629,7 @@ mod tests { }; let cumulative_latency = 100_f32 + 115_f32; - let cumulative_latency_squares = 100_f32 * 100_f32 + 115_f32 * 115_f32; + let cumulative_latency_squares = 100_f32.mul_add(100_f32, 115_f32 * 115_f32); let avg_manual = cumulative_avg( leader_metrics.latency_avg, @@ -625,15 +658,15 @@ mod tests { sample_size, latest_block_id, errors, - } = metric_map.get(&addr_leader).unwrap(); + } = 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); + 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] diff --git a/test_fixtures/src/lib.rs b/test_fixtures/src/lib.rs index e885e357..28fae660 100644 --- a/test_fixtures/src/lib.rs +++ b/test_fixtures/src/lib.rs @@ -421,7 +421,7 @@ impl BlockingTestContext { self.ctx.as_ref().expect("TestContext is set") } - pub fn ctx_mut(&mut self) -> &mut TestContext { + pub const fn ctx_mut(&mut self) -> &mut TestContext { self.ctx.as_mut().expect("TestContext is set") }