logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-15 10:20:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tintegration_tests): tests work

Browse Source
This commit is contained in:
Pravdyvy 2026-07-14 13:10:32 +03:00
parent 1615a06604
commit bf237daf71
19 changed files with 300 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
integration_tests/tests/auth_transfer/private.rs
View File

@ -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);

8
integration_tests/tests/private_pda.rs
View File

@ -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!(

18
integration_tests/tests/wallet_ffi.rs
View File

@ -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<FfiCreateWalletOutput> {
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<FfiCreateWalletO
let tempdir = tempdir()?;
let config_path = tempdir.path().join("wallet_config.json");
let storage_path = tempdir.path().join("storage.json");
let metrics_path = tempdir.path().join("metrics.json");
let config_path_c = CString::new(config_path.to_str().unwrap())?;
let storage_path_c = CString::new(storage_path.to_str().unwrap())?;
let metrics_path_c = CString::new(metrics_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(),
metrics_path_c.as_ptr(),
password.as_ptr(),
)
};
@ -388,10 +396,18 @@ fn new_wallet_ffi_with_default_config(password: &str) -> Result<FfiCreateWalletO
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]

2
lee/state_machine/src/state/mod.rs
View File

@ -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);
}

6
lee/state_machine/src/state/tests/claiming.rs
View File

@ -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;

2
lee/state_machine/src/state/tests/privacy_preserving.rs
View File

@ -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;

8
lez/wallet-ffi/src/vault.rs
View File

@ -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);

7
lez/wallet-ffi/src/wallet.rs
View File

@ -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}"));

7
lez/wallet/src/cli/account.rs
View File

@ -314,7 +314,7 @@ impl AccountSubcommand {
raw: bool,
keys: bool,
account_id: CliAccountMention,
wallet_core: &mut WalletCore,
wallet_core: &WalletCore,
) -> Result<SubcommandReturnValue> {
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<SubcommandReturnValue> {
async fn handle_list(long: bool, wallet_core: &WalletCore) -> Result<SubcommandReturnValue> {
let (public_account_ids, private_account_ids) = {
let key_chain = &wallet_core.storage.key_chain();

7
lez/wallet/src/cli/mod.rs
View File

@ -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)
}

10
lez/wallet/src/cli/programs/amm.rs
View File

@ -125,7 +125,7 @@ impl AmmProgramAgnosticSubcommand {
user_holding_lp: CliAccountMention,
balance_a: u128,
balance_b: u128,
wallet_core: &mut WalletCore,
wallet_core: &WalletCore,
) -> Result<SubcommandReturnValue> {
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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
let a_id = user_holding_a.resolve(wallet_core.storage())?;
let b_id = user_holding_b.resolve(wallet_core.storage())?;

2
lez/wallet/src/cli/programs/ata.rs
View File

@ -186,7 +186,7 @@ impl AtaSubcommand {
async fn handle_list(
owner: AccountId,
token_definition: Vec<AccountId>,
wallet_core: &mut WalletCore,
wallet_core: &WalletCore,
) -> Result<SubcommandReturnValue> {
let ata_program_id = programs::ata().id();

4
lez/wallet/src/cli/programs/native_token_transfer.rs
View File

@ -425,7 +425,7 @@ impl NativeTokenTransferProgramSubcommandShielded {
to_vpk: String,
to_identifier: Option<u128>,
amount: u128,
wallet_core: &mut WalletCore,
wallet_core: &WalletCore,
) -> Result<SubcommandReturnValue> {
let (to_npk, to_vpk) = crate::cli::decode_npk_vpk(&to_npk, &to_vpk)?;
@ -497,7 +497,7 @@ impl NativeTokenTransferProgramSubcommand {
from: Option<AccountIdentity>,
to: Option<AccountIdentity>,
amount: u128,
wallet_core: &mut WalletCore,
wallet_core: &WalletCore,
) -> Result<SubcommandReturnValue> {
let tx_hash = NativeTokenTransfer(wallet_core)
.send_public_transfer(

6
lez/wallet/src/cli/programs/pinata.rs
View File

@ -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<u128> {
async fn find_solution(wallet: &WalletCore, pinata_account_id: AccountId) -> Result<u128> {
let account = wallet.get_account_public(pinata_account_id).await?;
let data: [u8; 33] = account
.data

8
lez/wallet/src/cli/programs/token.rs
View File

@ -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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
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<SubcommandReturnValue> {
let definition = definition_account_id.resolve(wallet_core.storage())?;
let supply = supply_account_id.resolve(wallet_core.storage())?;

2
lez/wallet/src/config.rs
View File

@ -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<BasicAuth>,
}

55
lez/wallet/src/lib.rs
View File

@ -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<WalletConfigOverrides>,
@ -104,9 +100,9 @@ pub struct WalletCore {
metrics_path: PathBuf,
metrics: HashMap<Url, Metrics>,
/// Wrapping metric updates in Arc<RwLock> to not break intefaces too much.
/// Wrapping metric updates in Arc<RwLock> 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<RwLock<Vec<MetricsUpdate>>>,
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,

371
lez/wallet/src/multi_client.rs
View File

@ -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<HashMap<Url, Metrics>, 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<R, E>, 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<u64>,
/// 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<HashMap<Url, Metrics>, 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::<u128>() 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<Url, SequencerClient>,
metrics: &HashMap<Url, Metrics>,
@ -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<Url, Metrics>,
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<u64>,
/// 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<Url, Metrics>,
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<Url, Metrics>,
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<Url, Metrics>,
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::<f32>() / 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::<f32>() / 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::<f32>() / 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]

2
test_fixtures/src/lib.rs
View File

@ -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")
}