From 9cd7f51e9710c7ed896008cdc2d665b0955ab75b Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Wed, 15 Jul 2026 16:36:45 +0300 Subject: [PATCH 1/9] fix(wallet): multiple leaders inclusion --- lez/wallet/src/config.rs | 32 ++++++++--- lez/wallet/src/lib.rs | 2 +- lez/wallet/src/multi_client.rs | 100 ++++++++++++++++++++------------- test_fixtures/src/config.rs | 7 ++- 4 files changed, 90 insertions(+), 51 deletions(-) diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 0063c06d..80a68756 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -34,6 +34,23 @@ pub struct GasConfig { pub gas_limit_runtime: u64, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MultiSequencerClientConfig { + /// Maximum numbers of sequencers to send requests + pub distribution_limit: usize, + /// Limit number of sequencer polls during callibration, should not be zero + pub callibration_limit: usize, +} + +impl Default for MultiSequencerClientConfig { + fn default() -> Self { + Self { + distribution_limit: 1, + callibration_limit: 100, + } + } +} + #[optfield::optfield(pub WalletConfigOverrides, rewrap, attrs = (derive(Debug, Default, Clone)))] #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WalletConfig { @@ -48,8 +65,7 @@ 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, - /// Limit number of sequencer polls during callibration, should not be zero - pub callibration_limit: usize, + pub multi_sequencer_client_config: MultiSequencerClientConfig, } impl Default for WalletConfig { @@ -63,7 +79,7 @@ impl Default for WalletConfig { seq_tx_poll_max_blocks: 5, seq_poll_max_retries: 5, seq_block_poll_max_amount: 100, - callibration_limit: 100, + multi_sequencer_client_config: MultiSequencerClientConfig::default(), } } } @@ -113,7 +129,7 @@ impl WalletConfig { seq_tx_poll_max_blocks, seq_poll_max_retries, seq_block_poll_max_amount, - callibration_limit, + multi_sequencer_client_config, } = self; let WalletConfigOverrides { @@ -122,7 +138,7 @@ impl WalletConfig { 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, - callibration_limit: o_callibration_limit, + multi_sequencer_client_config: o_multi_sequencer_client_config, } = overrides; if let Some(v) = o_sequencers_conn_data { @@ -145,9 +161,9 @@ impl WalletConfig { warn!("Overriding wallet config 'seq_block_poll_max_amount' to {v}"); *seq_block_poll_max_amount = v; } - if let Some(v) = o_callibration_limit { - warn!("Overriding wallet config 'callibration_limit' to {v}"); - *callibration_limit = v; + if let Some(v) = o_multi_sequencer_client_config { + warn!("Overriding wallet config 'multi_sequencer_client_config' to {v:?}"); + *multi_sequencer_client_config = v; } } } diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index d9f1b891..055a0bfa 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -181,7 +181,7 @@ impl WalletCore { let multi_sequencer_client = MultiSequencerClient::new( &config.sequencers_conn_data, &mut metrics, - config.callibration_limit, + config.multi_sequencer_client_config.clone(), ) .await?; diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index f4b01e21..421a4194 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -14,7 +14,7 @@ use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuil use serde::{Deserialize, Serialize}; use url::Url; -use crate::config::SequencerConnectionData; +use crate::config::{MultiSequencerClientConfig, SequencerConnectionData}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Metrics { @@ -74,18 +74,16 @@ impl Metrics { #[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 - pub leader: SequencerClient, - pub leader_url: Url, + /// Ordered list of leaders, from best to worst + pub leader_list: Vec<(SequencerClient, Url)>, + pub multi_sequencer_client_config: MultiSequencerClientConfig, } impl MultiSequencerClient { pub async fn new( conn_data: &[SequencerConnectionData], metrics: &mut HashMap, - callibration_limit: usize, + multi_sequencer_client_config: MultiSequencerClientConfig, ) -> Result { let mut client_list = HashMap::new(); @@ -126,30 +124,40 @@ impl MultiSequencerClient { } else { metrics.insert( sequencer_addr.clone(), - callibrate_client(&sequencer_client, callibration_limit).await, + callibrate_client( + &sequencer_client, + multi_sequencer_client_config.callibration_limit, + ) + .await, ); } client_list.insert(sequencer_addr.clone(), sequencer_client); } - let (leader_url, leader) = choose_leader(&client_list, metrics) - .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; + let leader_list = choose_leaders( + &client_list, + metrics, + multi_sequencer_client_config.distribution_limit, + ) + .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; - log::info!("Chosen leader is {leader_url:?}"); + log::info!("Chosen leaders is {leader_list:?}"); - // Dropping client list, for reasons why, see comment in structure definition. - Ok(Self { leader, leader_url }) + Ok(Self { + leader_list, + multi_sequencer_client_config, + }) } #[must_use] - pub const fn leader_ref(&self) -> &SequencerClient { - &self.leader + pub fn leader_ref(&self) -> &(SequencerClient, Url) { + &self.leader_list.first().expect("Must be at least one leader") } #[must_use] - pub fn leader_clone(&self) -> SequencerClient { - self.leader.clone() + pub fn leader_clone(&self) -> (SequencerClient, Url) { + self.leader_list.first().expect("Must be at least one leader").clone() } // Keeping this call abstract, in case if we need to do more than one request @@ -157,11 +165,11 @@ impl MultiSequencerClient { &self, call: I, ) -> (Result, MetricsUpdate) { - let resp = tokio::join!(call(self.leader_ref()), actualize_client(self.leader_ref())); + let resp = tokio::join!(call(&(*self.leader_ref()).0), actualize_client(&(*self.leader_ref()).0)); log::debug!( "Metered call for {:?}, metric updates is {:?}", - self.leader_url, + self.leader_ref().1, resp.1 ); @@ -292,10 +300,11 @@ pub async fn actualize_client(client: &SequencerClient) -> MetricsUpdate { } #[must_use] -pub fn choose_leader( +pub fn choose_leaders( client_list: &HashMap, metrics: &HashMap, -) -> Option<(Url, SequencerClient)> { + distribution_limit: usize, +) -> Option> { let mut client_vec = vec![]; // Sort out all unmetered clients @@ -347,42 +356,53 @@ pub fn choose_leader( }); #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] - let client_vec = client_vec[..(client_vec + let mut client_vec = client_vec[..(client_vec .iter() .position(|item| (metrics.get(*item).unwrap().errors as f32) > avg_err_count) .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 = metrics.get(acc).unwrap(); - let (old_lat, old_var) = (old.latency_avg, old.latency_var); - let new = metrics.get(*x).unwrap(); - let (new_lat, new_var) = (new.latency_avg, new.latency_var); + client_vec.sort_by(|a, b| { + let left = metrics.get(*a).unwrap(); + let (left_lat, left_var) = (left.latency_avg, left.latency_var); + let right = metrics.get(*b).unwrap(); + let (right_lat, right_var) = (right.latency_avg, right.latency_var); - let new_std = new_var.sqrt(); - let old_std = old_var.sqrt(); + let right_std = right_var.sqrt(); + let left_std = left_var.sqrt(); - // Client is better if its averabe is better and variance does not make it worse + // 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] + // [-left_std............right_lat.......left_lat...............+right_std......... + // +left_std] // // However one can argue that this: // - // [-old_std...................old_lat........new_lat.........+new_std.......+old_std] + // [-left_std...................left_lat........right_lat.........+right_std....... + // +left_std] // // is still better, but it is up to discussion - if (new_lat <= old_lat) && ((new_lat + new_std) < (old_lat + old_std)) { - *x + if (right_lat <= right_lat) && ((right_lat + right_std) < (left_lat + left_std)) { + std::cmp::Ordering::Less } else { - acc + std::cmp::Ordering::Greater } }); - Some(( - min_lat_var_addr.clone(), - client_list.get(min_lat_var_addr).unwrap().clone(), - )) + Some( + client_vec + .iter() + .take(distribution_limit) + .map(|addr| { + let client = client_list + .get(*addr) + .expect("Missing clients already sorted out"); + + (client.clone(), (*addr).clone()) + }) + .collect(), + ) } /// Helperfunction to calculate cumulative average. @@ -455,7 +475,7 @@ mod tests { use url::Url; use crate::multi_client::{ - CumulativeUpdates, Metrics, MetricsUpdate, choose_leader, cumulative_avg, cumulative_var, + CumulativeUpdates, Metrics, MetricsUpdate, choose_leaders, cumulative_avg, cumulative_var, }; fn update_metrics( diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index bd7cbc32..cacca77d 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::{SequencerConnectionData, WalletConfig}; +use wallet::config::{MultiSequencerClientConfig, 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]; @@ -203,7 +203,10 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { seq_tx_poll_max_blocks: 15, seq_poll_max_retries: 10, seq_block_poll_max_amount: 100, - callibration_limit: 1, + multi_sequencer_client_config: MultiSequencerClientConfig { + distribution_limit: 1, + callibration_limit: 1, + }, }) } From 580c4661245f67c822c2a876d00a5b10960f8c8c Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Mon, 20 Jul 2026 17:10:01 +0300 Subject: [PATCH 2/9] feat(wallet): distributed transaction sending --- .../bin/run_hello_world_with_authorization.rs | 2 +- lez/wallet/src/cli/mod.rs | 2 +- lez/wallet/src/config.rs | 4 +- lez/wallet/src/lib.rs | 53 +++---- lez/wallet/src/multi_client.rs | 148 ++++++++++++------ 5 files changed, 127 insertions(+), 82 deletions(-) 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 563f0877..32aa0d4c 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 @@ -61,7 +61,7 @@ async fn main() { // Construct the public transaction // Query the current nonce from the node let nonces = wallet_core - .get_accounts_nonces(&[account_id]) + .get_accounts_nonces(vec![account_id]) .await .expect("Node should be reachable to query account data"); let signing_keys = [&signing_key]; diff --git a/lez/wallet/src/cli/mod.rs b/lez/wallet/src/cli/mod.rs index d436de52..16b6073a 100644 --- a/lez/wallet/src/cli/mod.rs +++ b/lez/wallet/src/cli/mod.rs @@ -387,7 +387,7 @@ 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(); + let leader_client = wallet_core.helm_owned(); wallet_core .storage diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 354b322e..06a8ea6b 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -41,14 +41,14 @@ pub struct MultiSequencerClientConfig { /// Maximum numbers of sequencers to send requests pub distribution_limit: usize, /// Limit number of sequencer polls during callibration, should not be zero - pub callibration_limit: usize, + pub calibration_limit: usize, } impl Default for MultiSequencerClientConfig { fn default() -> Self { Self { distribution_limit: 1, - callibration_limit: 100, + calibration_limit: 100, } } } diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index 588f9e0f..f46ceba6 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -174,7 +174,7 @@ impl WalletCore { let mut statistics = extract_statistics_from_path(&statistics_path)?; let multi_sequencer_client = MultiSequencerClient::new( - &config.sequencers_conn_data, + &config.sequencers, &mut statistics, config.multi_sequencer_client_config.clone(), ) @@ -204,17 +204,21 @@ impl WalletCore { #[must_use] pub fn optimal_poller(&self) -> TxPoller { - TxPoller::new(self.config(), self.leader_owned()) + TxPoller::new(self.config(), self.helm_owned()) } #[must_use] - pub fn leader_owned(&self) -> SequencerClient { - self.multi_sequencer_client.leader().clone() + pub fn helm_owned(&self) -> SequencerClient { + self.multi_sequencer_client.helm().0.clone() } #[must_use] - pub fn leader_url(&self) -> Url { - self.multi_sequencer_client.leader_url().clone() + pub fn helm_url(&self) -> Url { + self.multi_sequencer_client.helm().1.clone() + } + + pub fn leaders(&self) -> &[(SequencerClient, Url)] { + self.multi_sequencer_client.leaders() } /// Get storage. @@ -255,20 +259,15 @@ impl WalletCore { /// 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) + .update_statistics(&mut self.statistics) .await; self.multi_sequencer_client .rotate( &self.config.sequencers, &mut self.statistics, - self.config.calibration_limit, + &self.config.multi_sequencer_client_config, ) .await?; @@ -486,16 +485,16 @@ impl WalletCore { pub async fn get_account_balance(&self, acc: AccountId) -> Result { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| client.get_account_balance(acc).await) + .metered_get(async |client: &SequencerClient| client.get_account_balance(acc).await) .await?) } /// Get accounts nonces. - pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result> { + pub async fn get_accounts_nonces(&self, accs: Vec) -> Result> { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { - client.get_accounts_nonces(accs.to_vec()).await + .metered_get(async |client: &SequencerClient| { + client.get_accounts_nonces(accs).await }) .await?) } @@ -516,14 +515,14 @@ 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) + .metered_get(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) + .metered_get(async |client: &SequencerClient| client.get_block(block_id).await) .await?) } @@ -533,7 +532,7 @@ impl WalletCore { ) -> Result> { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| client.get_transaction(hash).await) + .metered_get(async |client: &SequencerClient| client.get_transaction(hash).await) .await?) } @@ -541,7 +540,7 @@ impl WalletCore { pub async fn get_account_public(&self, account_id: AccountId) -> Result { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| client.get_account(account_id).await) + .metered_get(async |client: &SequencerClient| client.get_account(account_id).await) .await?) } @@ -580,7 +579,7 @@ impl WalletCore { pub async fn get_program_ids(&self) -> Result> { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| client.get_program_ids().await) + .metered_get(async |client: &SequencerClient| client.get_program_ids().await) .await?) } @@ -598,8 +597,8 @@ impl WalletCore { ) -> Result<(Vec>, CommitmentSetDigest)> { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { - client.get_proofs_and_root(commitments.clone()).await + .metered_get(async |client: &SequencerClient| { + client.get_proofs_and_root(commitments).await }) .await?) } @@ -742,7 +741,7 @@ impl WalletCore { let call_res = self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { + .metered_send(async |client: &SequencerClient| { client .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) .await @@ -810,7 +809,7 @@ impl WalletCore { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { + .metered_send(async |client: &SequencerClient| { client .send_transaction(LeeTransaction::Public(tx.clone())) .await @@ -824,7 +823,7 @@ impl WalletCore { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { + .metered_send(async |client: &SequencerClient| { client .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) .await diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 71a6ce8c..0f3dee04 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -76,21 +76,21 @@ impl Statistics { #[derive(Clone)] pub struct MultiSequencerClient { - /// Ordered list of leaders, from best to worst + /// Ordered list of leaders, from best to worst. pub leader_list: Vec<(SequencerClient, Url)>, pub multi_sequencer_client_config: MultiSequencerClientConfig, /// 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>>, + statistic_updates: Arc>>>, } impl MultiSequencerClient { async fn setup( conn_data: &[SequencerConnectionData], statistics: &mut HashMap, - multi_sequencer_client_config: MultiSequencerClientConfig, - ) -> Result { + multi_sequencer_client_config: &MultiSequencerClientConfig, + ) -> Result> { let mut client_list = HashMap::new(); for SequencerConnectionData { @@ -117,55 +117,52 @@ impl MultiSequencerClient { }; // 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; + if let Some(statistic_mut) = statistics.get_mut(sequencer_addr) { + let statistic_updates = actualize_client(&sequencer_client).await; log::debug!( - "Metered call for {sequencer_addr:?}, metric updates is {metric_updates:?}" + "Metered call for {sequencer_addr:?}, statistic updates is {statistic_updates:?}" ); - metric_mut.apply_updates(&[metric_updates]); + statistic_mut.apply_updates(&[statistic_updates]); + + client_list.insert(sequencer_addr.clone(), sequencer_client); // Otherwise calibrate client data } else if let Some(client_statistics) = - calibrate_client(&sequencer_client, calibration_limit).await + calibrate_client(&sequencer_client, multi_sequencer_client_config.calibration_limit).await { statistics.insert(sequencer_addr.clone(), client_statistics); + client_list.insert(sequencer_addr.clone(), sequencer_client); + // There is no point in adding uncalibrated client } 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"); + log::warn!("Client {sequencer_addr:?} failed all {} calibration attempts, it may be unhealthy. + \n Consider bumping calibration_limit or remove this client altogether", multi_sequencer_client_config.calibration_limit); } - - client_list.insert(sequencer_addr.clone(), sequencer_client); } - // Dropping client list, for reasons why, see comment in structure definition. - let leader_list = choose_leaders( &client_list, - metrics, + statistics, multi_sequencer_client_config.distribution_limit, ) .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; log::info!("Chosen leaders is {leader_list:?}"); - Ok(Self { - leader_list, - multi_sequencer_client_config, - }) + Ok(leader_list) } pub async fn new( conn_data: &[SequencerConnectionData], statistics: &mut HashMap, - calibration_limit: usize, + multi_sequencer_client_config: MultiSequencerClientConfig, ) -> Result { - let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + let leader_list = Self::setup(conn_data, statistics, &multi_sequencer_client_config).await?; Ok(Self { - leader, - leader_url, - statistic_updates: Arc::new(RwLock::new(vec![])), + leader_list, + multi_sequencer_client_config, + statistic_updates: Arc::new(RwLock::new(HashMap::new())), }) } @@ -174,54 +171,95 @@ impl MultiSequencerClient { &mut self, conn_data: &[SequencerConnectionData], statistics: &mut HashMap, - calibration_limit: usize, + multi_sequencer_client_config: &MultiSequencerClientConfig, ) -> Result<()> { - let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + let leader_list = Self::setup(conn_data, statistics, multi_sequencer_client_config).await?; - log::info!("Chosen leader is {leader_url:?}"); + log::info!("Chosen leaders is {leader_list:#?}"); - self.leader = leader; - self.leader_url = leader_url; + self.leader_list = leader_list; Ok(()) } #[must_use] - pub const fn leader(&self) -> &SequencerClient { - &self.leader + pub fn leaders(&self) -> &[(SequencerClient, Url)] { + &self.leader_list.as_ref() } #[must_use] - pub const fn leader_url(&self) -> &Url { - &self.leader_url + pub fn helm(&self) -> &(SequencerClient, Url) { + &self.leader_list.first().expect("At least one leader must be set") } - // Keeping this call abstract, in case if we need to do more than one request - pub async fn metered_call Result>( + /// Metered call for main leader(helm), to get data, necessary for send call. + pub async fn metered_get Result>( &self, call: I, ) -> Result { + let (helm, helm_url) = self.helm(); + let (resp, statistics_update) = - tokio::join!(call(self.leader()), actualize_client(self.leader())); + tokio::join!(call(helm), actualize_client(helm)); log::debug!( - "Metered call for {:?}, metric updates is {:?}", - self.leader_url, + "Metered call for {:?}, statistic updates is {:?}", + helm_url, statistics_update ); { let mut statistic_updates_guard = self.statistic_updates.write().await; - statistic_updates_guard.push(statistics_update); + statistic_updates_guard.entry(helm_url.clone()).or_default().push(statistics_update); } resp } - /// Update statistics of a leader, clear statist updates log. - pub async fn update_statistics(&self, leader_statistic: &mut Statistics) { + /// Metered call for `distribution_limit` amount of leaders for sending data, usually transaction. + pub async fn metered_send Result>( + &self, + call: I, + ) -> Vec> { + let leaders = self.leaders().iter().take(self.multi_sequencer_client_config.distribution_limit); + + // Collecting all statistics into one map to lock updates only once + let mut statistic_map: HashMap> = HashMap::new(); + + let mut results = vec![]; + + for (leader, leader_url) in leaders { + let (resp, statistics_update) = + tokio::join!(call(leader), actualize_client(leader)); + + log::debug!( + "Metered call for {:?}, statistic updates is {:?}", + leader_url, + statistics_update + ); + + statistic_map.entry(leader_url.clone()).or_default().push(statistics_update); + results.push(resp); + } + + { + let mut statistic_updates_guard = self.statistic_updates.write().await; + + statistic_updates_guard.extend(statistic_map.into_iter()); + } + + results + } + + /// Update statistics of a leader, clear statistic updates log. + pub async fn update_statistics(&self, statistics: &mut HashMap,) { let mut statistic_updates = self.statistic_updates.write().await; - leader_statistic.apply_updates(statistic_updates.as_ref()); + + for (addr, statistic_updates_vec) in statistic_updates.iter() { + let leader_statistic = statistics.get_mut(addr).expect("Leader statistic must be present after setup"); + leader_statistic.apply_updates(statistic_updates_vec.as_slice()); + } + statistic_updates.clear(); } } @@ -838,9 +876,11 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - assert_eq!(leader_url, addr_leader); + let (_, leader_url) = leaders.first().unwrap(); + + assert_eq!(leader_url, &addr_leader); } #[test] @@ -908,9 +948,11 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - assert_eq!(leader_url, addr_leader); + let (_, leader_url) = leaders.first().unwrap(); + + assert_eq!(leader_url, &addr_leader); } #[test] @@ -978,9 +1020,11 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - assert_eq!(leader_url, addr_leader); + let (_, leader_url) = leaders.first().unwrap(); + + assert_eq!(leader_url, &addr_leader); } #[test] @@ -1048,8 +1092,10 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); + let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - assert_eq!(leader_url, addr_leader); + let (_, leader_url) = leaders.first().unwrap(); + + assert_eq!(leader_url, &addr_leader); } } From 0c32025c4b3d292191f667b11e9507987a049122 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Tue, 21 Jul 2026 18:01:04 +0300 Subject: [PATCH 3/9] feat(wallet_ffi): multi-poller --- .../src/bin/run_hello_world.rs | 2 +- .../bin/run_hello_world_through_tail_call.rs | 2 +- .../bin/run_hello_world_with_authorization.rs | 4 +- ...uthorization_through_tail_call_with_pda.rs | 2 +- .../bin/run_hello_world_with_move_function.rs | 4 +- integration_tests/tests/private_pda.rs | 2 +- lez/wallet-ffi/src/wallet.rs | 2 +- lez/wallet/src/account_manager.rs | 2 +- lez/wallet/src/config.rs | 15 +- lez/wallet/src/lib.rs | 56 +++++--- lez/wallet/src/multi_client.rs | 134 +++++++++++++----- lez/wallet/src/poller.rs | 21 +++ test_fixtures/src/config.rs | 2 +- 13 files changed, 177 insertions(+), 71 deletions(-) diff --git a/examples/program_deployment/src/bin/run_hello_world.rs b/examples/program_deployment/src/bin/run_hello_world.rs index a43bd7f1..f58f2ec0 100644 --- a/examples/program_deployment/src/bin/run_hello_world.rs +++ b/examples/program_deployment/src/bin/run_hello_world.rs @@ -59,7 +59,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); 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 18620220..00254bfd 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 @@ -55,7 +55,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); 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 32aa0d4c..f7ad36b1 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 @@ -61,7 +61,7 @@ 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]; @@ -73,7 +73,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_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 9139189e..a29bd263 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 @@ -57,7 +57,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_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 b99f41f4..c03563ad 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 @@ -88,7 +88,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); @@ -127,7 +127,7 @@ async fn main() { // Submit the transaction let _response = wallet_core - .leader_owned() + .helm_owned() .send_transaction(LeeTransaction::Public(tx)) .await .unwrap(); diff --git a/integration_tests/tests/private_pda.rs b/integration_tests/tests/private_pda.rs index 0b187283..af78aa74 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 - .leader_owned() + .helm_owned() .send_transaction(LeeTransaction::PrivacyPreserving(tx)) .await .map_err(|e| anyhow::anyhow!("send transaction failed: {e}"))?; diff --git a/lez/wallet-ffi/src/wallet.rs b/lez/wallet-ffi/src/wallet.rs index b2846912..cbee32a7 100644 --- a/lez/wallet-ffi/src/wallet.rs +++ b/lez/wallet-ffi/src/wallet.rs @@ -360,7 +360,7 @@ pub unsafe extern "C" fn wallet_ffi_get_sequencer_addr(handle: *mut WalletHandle } }; - let addr = wallet.leader_url().to_string(); + let addr = wallet.helm_url().to_string(); match std::ffi::CString::new(addr) { Ok(s) => s.into_raw(), diff --git a/lez/wallet/src/account_manager.rs b/lez/wallet/src/account_manager.rs index 824a994d..eb423650 100644 --- a/lez/wallet/src/account_manager.rs +++ b/lez/wallet/src/account_manager.rs @@ -611,7 +611,7 @@ async fn fetch_private_proofs_and_root( .unzip(); let (proofs, root) = wallet - .get_proofs_and_root(commitments.clone()) + .get_proofs_and_root(&commitments) .await .map_err(ExecutionFailureKind::SequencerError)?; diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 06a8ea6b..7fa1ba31 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use url::Url; const DEFAULT_CALLIBRATION_LIMIT: usize = 100; +const DEFAULT_DISTRIBUTION_LIMIT: usize = 1; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SequencerConnectionData { @@ -38,17 +39,17 @@ pub struct GasConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MultiSequencerClientConfig { - /// Maximum numbers of sequencers to send requests + /// Maximum numbers of sequencers to send requests. pub distribution_limit: usize, - /// Limit number of sequencer polls during callibration, should not be zero + /// Limit number of sequencer polls during callibration, should not be zero. pub calibration_limit: usize, } impl Default for MultiSequencerClientConfig { fn default() -> Self { Self { - distribution_limit: 1, - calibration_limit: 100, + distribution_limit: DEFAULT_DISTRIBUTION_LIMIT, + calibration_limit: DEFAULT_CALLIBRATION_LIMIT, } } } @@ -67,7 +68,7 @@ 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, - /// CURR_TODO: Add default serialization + #[serde(default = "MultiSequencerClientConfig::default")] pub multi_sequencer_client_config: MultiSequencerClientConfig, } @@ -170,7 +171,3 @@ impl WalletConfig { } } } - -const fn default_calibration_limit() -> usize { - DEFAULT_CALLIBRATION_LIMIT -} diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index f46ceba6..ed2b2b84 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -39,7 +39,7 @@ use crate::{ account::{AccountIdWithPrivacy, Label}, config::WalletConfigOverrides, multi_client::{MultiSequencerClient, Statistics, extract_statistics_from_path}, - poller::TxPoller, + poller::{TxPoller, multi_poll}, storage::key_chain::{NullifierIndex, SharedAccountEntry}, }; @@ -203,7 +203,16 @@ impl WalletCore { } #[must_use] - pub fn optimal_poller(&self) -> TxPoller { + pub fn poller_vec(&self) -> Vec { + self.leaders() + .iter() + .take(self.multi_sequencer_client.config().distribution_limit) + .map(|(leader, _)| TxPoller::new(self.config(), leader.clone())) + .collect() + } + + #[must_use] + pub fn poller_helm(&self) -> TxPoller { TxPoller::new(self.config(), self.helm_owned()) } @@ -217,6 +226,7 @@ impl WalletCore { self.multi_sequencer_client.helm().1.clone() } + #[must_use] pub fn leaders(&self) -> &[(SequencerClient, Url)] { self.multi_sequencer_client.leaders() } @@ -490,11 +500,11 @@ impl WalletCore { } /// Get accounts nonces. - pub async fn get_accounts_nonces(&self, accs: Vec) -> Result> { + pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result> { Ok(self .multi_sequencer_client .metered_get(async |client: &SequencerClient| { - client.get_accounts_nonces(accs).await + client.get_accounts_nonces(accs.to_vec()).await }) .await?) } @@ -584,21 +594,18 @@ impl WalletCore { } /// 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 poll_transaction(&self, tx_hash: HashType) -> Result<(LeeTransaction, BlockId)> { + multi_poll(self.poller_vec(), tx_hash).await } pub async fn get_proofs_and_root( &self, - commitments: Vec, + commitments: &[Commitment], ) -> Result<(Vec>, CommitmentSetDigest)> { Ok(self .multi_sequencer_client .metered_get(async |client: &SequencerClient| { - client.get_proofs_and_root(commitments).await + client.get_proofs_and_root(commitments.to_vec()).await }) .await?) } @@ -649,7 +656,7 @@ impl WalletCore { tx_hash: HashType, ) -> Result { println!("Transaction hash is {tx_hash}"); - let (tx, block_id) = self.optimal_poller().poll_tx(tx_hash).await?; + let (tx, block_id) = self.poll_transaction(tx_hash).await?; println!("Transaction is included in block {block_id}"); println!("Transaction data is {tx:?}"); self.store_persistent_data()?; @@ -663,7 +670,7 @@ impl WalletCore { acc_decode_data: &[AccDecodeData], ) -> Result { println!("Transaction hash is {tx_hash}"); - let (tx, block_id) = self.optimal_poller().poll_tx(tx_hash).await?; + let (tx, block_id) = self.poll_transaction(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 { @@ -746,7 +753,12 @@ impl WalletCore { .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) .await }) - .await; + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( + "All sequencers rejected transaction" + )))?; Ok((call_res?, shared_secrets)) } @@ -814,7 +826,12 @@ impl WalletCore { .send_transaction(LeeTransaction::Public(tx.clone())) .await }) - .await?) + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( + "All sequencers rejected transaction" + )))??) } pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { @@ -828,7 +845,12 @@ impl WalletCore { .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) .await }) - .await?) + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( + "All sequencers rejected transaction" + )))??) } pub async fn sync_to_latest_block(&mut self) -> Result { @@ -854,7 +876,7 @@ impl WalletCore { println!("Syncing to block {block_id}. Blocks to sync: {num_of_blocks}"); - let poller = self.optimal_poller(); + let poller = self.poller_helm(); let mut blocks = std::pin::pin!(poller.poll_block_range(last_synced_block.saturating_add(1)..=block_id)); diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 0f3dee04..510ed73b 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -34,6 +34,17 @@ pub struct StatisticsUpdate { pub is_failed: bool, } +impl StatisticsUpdate { + #[must_use] + pub const fn failure() -> Self { + Self { + latency: 0_f32, + new_latest_block_id: None, + is_failed: true, + } + } +} + impl Statistics { pub fn apply_updates(&mut self, updates: &[StatisticsUpdate]) { let CumulativeUpdates { @@ -77,8 +88,8 @@ impl Statistics { #[derive(Clone)] pub struct MultiSequencerClient { /// Ordered list of leaders, from best to worst. - pub leader_list: Vec<(SequencerClient, Url)>, - pub multi_sequencer_client_config: MultiSequencerClientConfig, + leader_list: Vec<(SequencerClient, Url)>, + config: MultiSequencerClientConfig, /// Wrapping statistic updates in Arc to not break interfaces too much. /// /// It is assumed, that wallet methods can be accesed via immutable reference. @@ -128,8 +139,11 @@ impl MultiSequencerClient { client_list.insert(sequencer_addr.clone(), sequencer_client); // Otherwise calibrate client data - } else if let Some(client_statistics) = - calibrate_client(&sequencer_client, multi_sequencer_client_config.calibration_limit).await + } else if let Some(client_statistics) = calibrate_client( + &sequencer_client, + multi_sequencer_client_config.calibration_limit, + ) + .await { statistics.insert(sequencer_addr.clone(), client_statistics); client_list.insert(sequencer_addr.clone(), sequencer_client); @@ -147,6 +161,8 @@ impl MultiSequencerClient { ) .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; + assert_ne!(leader_list.len(), 0); + log::info!("Chosen leaders is {leader_list:?}"); Ok(leader_list) @@ -157,11 +173,12 @@ impl MultiSequencerClient { statistics: &mut HashMap, multi_sequencer_client_config: MultiSequencerClientConfig, ) -> Result { - let leader_list = Self::setup(conn_data, statistics, &multi_sequencer_client_config).await?; + let leader_list = + Self::setup(conn_data, statistics, &multi_sequencer_client_config).await?; Ok(Self { leader_list, - multi_sequencer_client_config, + config: multi_sequencer_client_config, statistic_updates: Arc::new(RwLock::new(HashMap::new())), }) } @@ -184,61 +201,107 @@ impl MultiSequencerClient { #[must_use] pub fn leaders(&self) -> &[(SequencerClient, Url)] { - &self.leader_list.as_ref() + self.leader_list.as_ref() } #[must_use] pub fn helm(&self) -> &(SequencerClient, Url) { - &self.leader_list.first().expect("At least one leader must be set") + self.leader_list + .first() + .expect("At least one leader must be set") + } + + #[must_use] + pub const fn config(&self) -> &MultiSequencerClientConfig { + &self.config } /// Metered call for main leader(helm), to get data, necessary for send call. - pub async fn metered_get Result>( + /// + /// If current leader errors, we ask next one in list. + pub async fn metered_get Result>( &self, call: I, ) -> Result { + // Collecting all statistics into one map to lock updates only once + let mut statistic_map: HashMap> = HashMap::new(); + let (helm, helm_url) = self.helm(); - let (resp, statistics_update) = - tokio::join!(call(helm), actualize_client(helm)); + let (mut resp, statistics_update) = tokio::join!(call(helm), actualize_client(helm)); - log::debug!( - "Metered call for {:?}, statistic updates is {:?}", - helm_url, - statistics_update - ); + log::debug!("Metered call for {helm_url:?}, statistic updates is {statistics_update:?}",); + + statistic_map + .entry(helm_url.clone()) + .or_default() + .push(statistics_update); + + if resp.is_err() { + statistic_map + .entry(helm_url.clone()) + .or_default() + .push(StatisticsUpdate::failure()); + + for (leader, leader_url) in self.leaders().iter().skip(1) { + let (curr_resp, statistics_update) = + tokio::join!(call(leader), actualize_client(leader)); + + log::debug!( + "Metered call for {leader_url:?}, statistic updates is {statistics_update:?}", + ); + + statistic_map + .entry(leader_url.clone()) + .or_default() + .push(statistics_update); + + resp = curr_resp; + + if resp.is_err() { + statistic_map + .entry(leader_url.clone()) + .or_default() + .push(StatisticsUpdate::failure()); + } else { + break; + } + } + } { let mut statistic_updates_guard = self.statistic_updates.write().await; - statistic_updates_guard.entry(helm_url.clone()).or_default().push(statistics_update); + + statistic_updates_guard.extend(statistic_map.into_iter()); } resp } - /// Metered call for `distribution_limit` amount of leaders for sending data, usually transaction. + /// Metered call for `distribution_limit` amount of leaders for sending data, usually + /// transaction. pub async fn metered_send Result>( &self, call: I, ) -> Vec> { - let leaders = self.leaders().iter().take(self.multi_sequencer_client_config.distribution_limit); + let leaders = self.leaders().iter().take(self.config().distribution_limit); // Collecting all statistics into one map to lock updates only once let mut statistic_map: HashMap> = HashMap::new(); let mut results = vec![]; - + for (leader, leader_url) in leaders { - let (resp, statistics_update) = - tokio::join!(call(leader), actualize_client(leader)); + let (resp, statistics_update) = tokio::join!(call(leader), actualize_client(leader)); log::debug!( - "Metered call for {:?}, statistic updates is {:?}", - leader_url, - statistics_update + "Metered call for {leader_url:?}, statistic updates is {statistics_update:?}", ); - statistic_map.entry(leader_url.clone()).or_default().push(statistics_update); + statistic_map + .entry(leader_url.clone()) + .or_default() + .push(statistics_update); results.push(resp); } @@ -252,14 +315,17 @@ impl MultiSequencerClient { } /// Update statistics of a leader, clear statistic updates log. - pub async fn update_statistics(&self, statistics: &mut HashMap,) { + pub async fn update_statistics(&self, statistics: &mut HashMap) { let mut statistic_updates = self.statistic_updates.write().await; + #[expect(clippy::iter_over_hash_type, reason = "Ordering is unnecesary here")] for (addr, statistic_updates_vec) in statistic_updates.iter() { - let leader_statistic = statistics.get_mut(addr).expect("Leader statistic must be present after setup"); + let leader_statistic = statistics + .get_mut(addr) + .expect("Leader statistic must be present after setup"); leader_statistic.apply_updates(statistic_updates_vec.as_slice()); } - + statistic_updates.clear(); } } @@ -495,7 +561,7 @@ pub fn choose_leaders( // +left_std] // // is still better, but it is up to discussion - if (right_lat <= right_lat) && ((right_lat + right_std) < (left_lat + left_std)) { + if (right_lat <= left_lat) && ((right_lat + right_std) < (left_lat + left_std)) { std::cmp::Ordering::Less } else { std::cmp::Ordering::Greater @@ -878,7 +944,7 @@ mod tests { let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - let (_, leader_url) = leaders.first().unwrap(); + let (_, leader_url) = leaders.first().unwrap(); assert_eq!(leader_url, &addr_leader); } @@ -950,7 +1016,7 @@ mod tests { let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - let (_, leader_url) = leaders.first().unwrap(); + let (_, leader_url) = leaders.first().unwrap(); assert_eq!(leader_url, &addr_leader); } @@ -1022,7 +1088,7 @@ mod tests { let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - let (_, leader_url) = leaders.first().unwrap(); + let (_, leader_url) = leaders.first().unwrap(); assert_eq!(leader_url, &addr_leader); } @@ -1094,7 +1160,7 @@ mod tests { let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); - let (_, leader_url) = leaders.first().unwrap(); + let (_, leader_url) = leaders.first().unwrap(); assert_eq!(leader_url, &addr_leader); } diff --git a/lez/wallet/src/poller.rs b/lez/wallet/src/poller.rs index 63ba2275..80f2a0c5 100644 --- a/lez/wallet/src/poller.rs +++ b/lez/wallet/src/poller.rs @@ -5,6 +5,7 @@ use common::{HashType, block::Block, transaction::LeeTransaction}; use lee_core::BlockId; use log::{info, warn}; use sequencer_service_rpc::{RpcClient as _, SequencerClient}; +use tokio::task::JoinSet; use crate::config::WalletConfig; @@ -87,3 +88,23 @@ impl TxPoller { } } } + +pub async fn multi_poll( + pollers: Vec, + tx_hash: HashType, +) -> Result<(LeeTransaction, BlockId)> { + let mut set = JoinSet::new(); + + for poller in pollers { + set.spawn(async move { poller.poll_tx(tx_hash).await }); + } + + while let Some(res) = set.join_next().await { + if let Ok(Ok(tx_res)) = res { + return Ok(tx_res); + } + // There is no point handling failed poll here + } + + anyhow::bail!("All pollers failed") +} diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index 75310000..e6f5fc92 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -205,7 +205,7 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { seq_block_poll_max_amount: 100, multi_sequencer_client_config: MultiSequencerClientConfig { distribution_limit: 1, - callibration_limit: 5, + calibration_limit: 5, }, }) } From 3f3f5d77d6169a15e9bc4702ed3a2d589dc6b545 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Wed, 22 Jul 2026 10:13:01 +0300 Subject: [PATCH 4/9] fix(wallet): unit test fix --- lez/wallet/src/config.rs | 3 +- lez/wallet/src/multi_client.rs | 454 ++++++++++++++++++++++++++------- 2 files changed, 365 insertions(+), 92 deletions(-) diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 7fa1ba31..f9c1b748 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -39,7 +39,8 @@ pub struct GasConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MultiSequencerClientConfig { - /// Maximum numbers of sequencers to send requests. + /// Maximum numbers of sequencers to send requests. Client can have AT MOST + /// `distribution_limit` active clients. pub distribution_limit: usize, /// Limit number of sequencer polls during callibration, should not be zero. pub calibration_limit: usize, diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 510ed73b..1addf7db 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -216,6 +216,26 @@ impl MultiSequencerClient { &self.config } + /// Helperfunction for the `metered_get`. + async fn metered_get_helper Result>( + &self, + call: &I, + leader: &SequencerClient, + leader_url: &Url, + statistic_map: &mut HashMap>, + ) -> Result { + let (resp, statistics_update) = tokio::join!(call(leader), actualize_client(leader)); + + log::debug!("Metered call for {leader_url:?}, statistic updates is {statistics_update:?}",); + + statistic_map + .entry(leader_url.clone()) + .or_default() + .push(statistics_update); + + resp + } + /// Metered call for main leader(helm), to get data, necessary for send call. /// /// If current leader errors, we ask next one in list. @@ -226,17 +246,14 @@ impl MultiSequencerClient { // Collecting all statistics into one map to lock updates only once let mut statistic_map: HashMap> = HashMap::new(); + // We need helm response to avoid constructing guard Result, which we can not do with generics. let (helm, helm_url) = self.helm(); - let (mut resp, statistics_update) = tokio::join!(call(helm), actualize_client(helm)); - - log::debug!("Metered call for {helm_url:?}, statistic updates is {statistics_update:?}",); - - statistic_map - .entry(helm_url.clone()) - .or_default() - .push(statistics_update); + let mut resp = self + .metered_get_helper(&call, helm, helm_url, &mut statistic_map) + .await; + // Not the cleanest approach, but I am not sure how to have it both clean and async. if resp.is_err() { statistic_map .entry(helm_url.clone()) @@ -244,19 +261,9 @@ impl MultiSequencerClient { .push(StatisticsUpdate::failure()); for (leader, leader_url) in self.leaders().iter().skip(1) { - let (curr_resp, statistics_update) = - tokio::join!(call(leader), actualize_client(leader)); - - log::debug!( - "Metered call for {leader_url:?}, statistic updates is {statistics_update:?}", - ); - - statistic_map - .entry(leader_url.clone()) - .or_default() - .push(statistics_update); - - resp = curr_resp; + resp = self + .metered_get_helper(&call, leader, leader_url, &mut statistic_map) + .await; if resp.is_err() { statistic_map @@ -552,16 +559,14 @@ pub fn choose_leaders( // Client is better if its average is better and variance does not make it worse // So basically we want this: - // [-left_std............right_lat.......left_lat...............+right_std......... - // +left_std] + // [-right_std < left_lat < right_lat < +left_std < +right_std] // // However one can argue that this: // - // [-left_std...................left_lat........right_lat.........+right_std....... - // +left_std] + // [-right_std < right_lat < left_lat < +left_std < +right_std] // // is still better, but it is up to discussion - if (right_lat <= left_lat) && ((right_lat + right_std) < (left_lat + left_std)) { + if (left_lat <= right_lat) && ((left_lat + left_std) < (right_lat + right_std)) { std::cmp::Ordering::Less } else { std::cmp::Ordering::Greater @@ -656,7 +661,7 @@ fn error_ratio(errors: u64, size: usize) -> f32 { #[cfg(test)] mod tests { - use std::collections::HashMap; + use std::collections::{HashMap, HashSet}; use sequencer_service_rpc::{SequencerClient, SequencerClientBuilder}; use url::Url; @@ -685,6 +690,27 @@ mod tests { builder.build(url).unwrap() } + fn four_client_list() -> (HashMap, [Url; 4]) { + 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); + + (client_list, [addr_leader, addr_1, addr_2, addr_3]) + } + #[test] fn cumulative_updates_test() { let statistics_updates_vec = vec![ @@ -879,22 +905,7 @@ mod tests { #[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 (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); let mut statistics = HashMap::new(); @@ -951,22 +962,7 @@ mod tests { #[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 (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); let mut statistics = HashMap::new(); @@ -1023,22 +1019,7 @@ mod tests { #[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 (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); let mut statistics = HashMap::new(); @@ -1095,22 +1076,7 @@ mod tests { #[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 (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); let mut statistics = HashMap::new(); @@ -1164,4 +1130,310 @@ mod tests { assert_eq!(leader_url, &addr_leader); } + + #[test] + fn choose_multiple_leaders_latest_block() { + let (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); + + 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.clone(), + Statistics { + latency_avg: 100_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 leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + + let mut url_set_origin = HashSet::new(); + let mut url_set_res = HashSet::new(); + + let (_, leader_url_first) = leaders[0].clone(); + let (_, leader_url_second) = leaders[1].clone(); + + url_set_origin.insert(addr_leader); + url_set_origin.insert(addr_1); + + url_set_res.insert(leader_url_first); + url_set_res.insert(leader_url_second); + + assert_eq!(url_set_origin, url_set_res); + assert_eq!(leaders.len(), 2); + } + + #[test] + fn choose_multiple_leaders_latest_block_still_chooses_one_best() { + let (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); + + 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 leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + + let (_, helm) = leaders.first().unwrap(); + + assert_eq!(&addr_leader, helm); + assert_eq!(leaders.len(), 1); + } + + #[test] + fn choose_multiple_leaders_least_errors() { + let (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); + + 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: 6, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 6, + }, + ); + + statistics.insert( + addr_1.clone(), + 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 leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + + let (_, leader_url_first) = leaders[0].clone(); + let (_, leader_url_second) = leaders[1].clone(); + + assert_eq!(addr_leader, leader_url_first); + assert_eq!(addr_1, leader_url_second); + assert_eq!(leaders.len(), 2); + } + + #[test] + fn choose_multiple_leaders_simple_latency_check() { + let (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); + + 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: 6, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 102_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + statistics.insert( + addr_1.clone(), + Statistics { + latency_avg: 101_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + + let (_, leader_url_first) = leaders[0].clone(); + let (_, leader_url_second) = leaders[1].clone(); + + assert_eq!(addr_leader, leader_url_first); + assert_eq!(addr_1, leader_url_second); + assert_eq!(leaders.len(), 2); + } + + #[test] + fn choose_multiple_leaders_var_check() { + let (client_list, [addr_leader, addr_1, addr_2, addr_3]) = four_client_list(); + + 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: 6, + }, + ); + + statistics.insert( + addr_2, + Statistics { + latency_avg: 100_f32, + latency_var: 12_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + statistics.insert( + addr_1.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 11_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + statistics.insert( + addr_leader.clone(), + Statistics { + latency_avg: 100_f32, + latency_var: 10_f32, + sample_size: 10, + latest_block_id: 100, + errors: 4, + }, + ); + + let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + + let (_, leader_url_first) = leaders[0].clone(); + let (_, leader_url_second) = leaders[1].clone(); + + assert_eq!(addr_leader, leader_url_first); + assert_eq!(addr_1, leader_url_second); + assert_eq!(leaders.len(), 2); + } } From ff0efddccccc606a9d905a2609c52f6a6d9c6ae6 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Wed, 22 Jul 2026 10:47:37 +0300 Subject: [PATCH 5/9] fix(fmt): fmt --- lez/wallet/src/multi_client.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 1addf7db..45df1256 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -246,7 +246,8 @@ impl MultiSequencerClient { // Collecting all statistics into one map to lock updates only once let mut statistic_map: HashMap> = HashMap::new(); - // We need helm response to avoid constructing guard Result, which we can not do with generics. + // We need helm response to avoid constructing guard Result, which we can not do with + // generics. let (helm, helm_url) = self.helm(); let mut resp = self From b6b5b4397a8cf531363a3ebaae66e1d49f6044de Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Wed, 22 Jul 2026 13:21:27 +0300 Subject: [PATCH 6/9] fix(wallet): suggestions fix --- lez/wallet/src/lib.rs | 14 ++-- lez/wallet/src/multi_client.rs | 116 +++++++++++++-------------------- 2 files changed, 52 insertions(+), 78 deletions(-) diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index ed2b2b84..3604c406 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -88,6 +88,8 @@ pub enum ExecutionFailureKind { SignError(anyhow::Error), #[error(transparent)] KeycardError(#[from] pyo3::PyErr), + #[error("Sending transaction failed for each client")] + MultiSequencerTransactionSendError, } pub struct WalletCore { @@ -756,9 +758,7 @@ impl WalletCore { .await .into_iter() .find(std::result::Result::is_ok) - .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( - "All sequencers rejected transaction" - )))?; + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)?; Ok((call_res?, shared_secrets)) } @@ -829,9 +829,7 @@ impl WalletCore { .await .into_iter() .find(std::result::Result::is_ok) - .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( - "All sequencers rejected transaction" - )))??) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??) } pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { @@ -848,9 +846,7 @@ impl WalletCore { .await .into_iter() .find(std::result::Result::is_ok) - .ok_or(ExecutionFailureKind::SequencerError(anyhow::anyhow!( - "All sequencers rejected transaction" - )))??) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??) } pub async fn sync_to_latest_block(&mut self) -> Result { diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 45df1256..d987c55c 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -28,21 +28,12 @@ pub struct Statistics { } #[derive(Debug, Clone)] -pub struct StatisticsUpdate { - pub latency: f32, - pub new_latest_block_id: Option, - pub is_failed: bool, -} - -impl StatisticsUpdate { - #[must_use] - pub const fn failure() -> Self { - Self { - latency: 0_f32, - new_latest_block_id: None, - is_failed: true, - } - } +pub enum StatisticsUpdate { + Success { + latency: f32, + new_latest_block_id: BlockId, + }, + Failure, } impl Statistics { @@ -259,7 +250,7 @@ impl MultiSequencerClient { statistic_map .entry(helm_url.clone()) .or_default() - .push(StatisticsUpdate::failure()); + .push(StatisticsUpdate::Failure); for (leader, leader_url) in self.leaders().iter().skip(1) { resp = self @@ -270,7 +261,7 @@ impl MultiSequencerClient { statistic_map .entry(leader_url.clone()) .or_default() - .push(StatisticsUpdate::failure()); + .push(StatisticsUpdate::Failure); } else { break; } @@ -353,35 +344,33 @@ impl CumulativeUpdates { 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) - }, - ) + .fold((0_u64, None, 0_f32, 0_f32), |mut acc, x| { + match x { + StatisticsUpdate::Success { + latency, + new_latest_block_id, + } => { + match acc.1 { + Some(val_old) => { + acc.1 = Some(std::cmp::max(val_old, *new_latest_block_id)); + } + None => { + acc.1 = Some(*new_latest_block_id); + } + } + acc.2 += latency; + acc.3 = latency.mul_add(*latency, acc.3); + } + StatisticsUpdate::Failure => { + acc.0 = acc.0.saturating_add(1); + } + } + acc }); Self { failure_count, - latest_block_id: latest_block_id.copied(), + latest_block_id, cumulative_latency, cumulative_latency_squares, additional_sample_size: metric_updates.len().saturating_sub( @@ -467,11 +456,12 @@ pub async fn actualize_client(client: &SequencerClient) -> StatisticsUpdate { #[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(), - } + block_id.map_or(StatisticsUpdate::Failure, |new_latest_block_id| { + StatisticsUpdate::Success { + latency, + new_latest_block_id, + } + }) } #[must_use] @@ -715,21 +705,15 @@ mod tests { #[test] fn cumulative_updates_test() { let statistics_updates_vec = vec![ - StatisticsUpdate { + StatisticsUpdate::Success { latency: 100_f32, - new_latest_block_id: Some(15), - is_failed: false, + new_latest_block_id: 15, }, - StatisticsUpdate { + StatisticsUpdate::Success { latency: 115_f32, - new_latest_block_id: Some(16), - is_failed: false, - }, - StatisticsUpdate { - latency: 50_f32, - new_latest_block_id: None, - is_failed: true, + new_latest_block_id: 16, }, + StatisticsUpdate::Failure, ]; let CumulativeUpdates { @@ -836,21 +820,15 @@ mod tests { #[test] fn metric_updates_correctness() { let statistics_updates_vec = vec![ - StatisticsUpdate { + StatisticsUpdate::Success { latency: 100_f32, - new_latest_block_id: Some(105), - is_failed: false, + new_latest_block_id: 105, }, - StatisticsUpdate { + StatisticsUpdate::Success { latency: 115_f32, - new_latest_block_id: Some(106), - is_failed: false, - }, - StatisticsUpdate { - latency: 50_f32, - new_latest_block_id: None, - is_failed: true, + new_latest_block_id: 106, }, + StatisticsUpdate::Failure, ]; let addr_leader = Url::parse("https://127.0.0.1:3040").unwrap(); From e1e55e72e5c02a4d1b627252ebb1c22923e5f5a4 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Thu, 23 Jul 2026 08:30:05 +0300 Subject: [PATCH 7/9] fix(wallet): suggestions fix 2 --- lez/wallet/src/multi_client.rs | 46 +++++++++++++++++++++------------- 1 file changed, 28 insertions(+), 18 deletions(-) diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index d987c55c..a8a63861 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -229,7 +229,7 @@ impl MultiSequencerClient { /// Metered call for main leader(helm), to get data, necessary for send call. /// - /// If current leader errors, we ask next one in list. + /// If current leader errors, we ask next one in list up to a `self.config.distribution_limit`. pub async fn metered_get Result>( &self, call: I, @@ -247,22 +247,12 @@ impl MultiSequencerClient { // Not the cleanest approach, but I am not sure how to have it both clean and async. if resp.is_err() { - statistic_map - .entry(helm_url.clone()) - .or_default() - .push(StatisticsUpdate::Failure); - for (leader, leader_url) in self.leaders().iter().skip(1) { resp = self .metered_get_helper(&call, leader, leader_url, &mut statistic_map) .await; - if resp.is_err() { - statistic_map - .entry(leader_url.clone()) - .or_default() - .push(StatisticsUpdate::Failure); - } else { + if resp.is_ok() { break; } } @@ -271,7 +261,16 @@ impl MultiSequencerClient { { let mut statistic_updates_guard = self.statistic_updates.write().await; - statistic_updates_guard.extend(statistic_map.into_iter()); + #[expect( + clippy::iter_over_hash_type, + reason = "Ordering of map updates is not important" + )] + for (url, updates) in statistic_map { + statistic_updates_guard + .entry(url) + .or_insert_with(Vec::new) + .extend(updates); + } } resp @@ -307,7 +306,16 @@ impl MultiSequencerClient { { let mut statistic_updates_guard = self.statistic_updates.write().await; - statistic_updates_guard.extend(statistic_map.into_iter()); + #[expect( + clippy::iter_over_hash_type, + reason = "Ordering of map updates is not important" + )] + for (url, updates) in statistic_map { + statistic_updates_guard + .entry(url) + .or_insert_with(Vec::new) + .extend(updates); + } } results @@ -557,10 +565,12 @@ pub fn choose_leaders( // [-right_std < right_lat < left_lat < +left_std < +right_std] // // is still better, but it is up to discussion - if (left_lat <= right_lat) && ((left_lat + left_std) < (right_lat + right_std)) { - std::cmp::Ordering::Less - } else { - std::cmp::Ordering::Greater + let first_ordering = left_lat.total_cmp(&right_lat); + match first_ordering { + std::cmp::Ordering::Greater => first_ordering, + std::cmp::Ordering::Less | std::cmp::Ordering::Equal => { + (left_lat + left_std).total_cmp(&(right_lat + right_std)) + } } }); From 70ac6b3294dfe6f5997a3386dc01d66194aace41 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Thu, 23 Jul 2026 14:59:54 +0300 Subject: [PATCH 8/9] fix(wallet): merge postfix --- lez/wallet/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index 19e57ccd..e38b33b9 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -450,7 +450,7 @@ impl WalletCore { let mut index = NullifierIndex::default(); index.track_initialization(account_id); - let poller = self.optimal_poller(); + let poller = self.poller_helm(); let mut blocks = std::pin::pin!(poller.poll_block_range(1..=cursor)); while let Some(block) = blocks.try_next().await? { for tx in block.body.transactions { From 93eed48112b7c59864831f72b56fd1a1ff8186d8 Mon Sep 17 00:00:00 2001 From: Pravdyvy Date: Fri, 24 Jul 2026 08:02:03 +0300 Subject: [PATCH 9/9] fix(wallet): suggestion fix 3 --- lez/wallet/src/lib.rs | 2 +- lez/wallet/src/multi_client.rs | 12 +++++++++--- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index e38b33b9..3388755d 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -273,7 +273,7 @@ impl WalletCore { pub async fn client_rotation(&mut self) -> Result<()> { self.multi_sequencer_client .update_statistics(&mut self.statistics) - .await; + .await?; self.multi_sequencer_client .rotate( diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index a8a63861..3c9da807 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -152,7 +152,11 @@ impl MultiSequencerClient { ) .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; - assert_ne!(leader_list.len(), 0); + if leader_list.is_empty() { + anyhow::bail!( + "Leader search algorithm failure: Sorted out all clients during leader search" + ); + } log::info!("Chosen leaders is {leader_list:?}"); @@ -322,18 +326,20 @@ impl MultiSequencerClient { } /// Update statistics of a leader, clear statistic updates log. - pub async fn update_statistics(&self, statistics: &mut HashMap) { + pub async fn update_statistics(&self, statistics: &mut HashMap) -> Result<()> { let mut statistic_updates = self.statistic_updates.write().await; #[expect(clippy::iter_over_hash_type, reason = "Ordering is unnecesary here")] for (addr, statistic_updates_vec) in statistic_updates.iter() { let leader_statistic = statistics .get_mut(addr) - .expect("Leader statistic must be present after setup"); + .ok_or_else(|| anyhow::anyhow!("Leader statistic must be present after setup"))?; leader_statistic.apply_updates(statistic_updates_vec.as_slice()); } statistic_updates.clear(); + + Ok(()) } }