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 563f0877..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 @@ -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 e1127181..f7f32f0c 100644 --- a/lez/wallet/src/account_manager.rs +++ b/lez/wallet/src/account_manager.rs @@ -600,7 +600,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/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 ede71073..f9c1b748 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 { @@ -36,6 +37,24 @@ pub struct GasConfig { pub gas_limit_runtime: u64, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MultiSequencerClientConfig { + /// 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, +} + +impl Default for MultiSequencerClientConfig { + fn default() -> Self { + Self { + distribution_limit: DEFAULT_DISTRIBUTION_LIMIT, + calibration_limit: DEFAULT_CALLIBRATION_LIMIT, + } + } +} + #[optfield::optfield(pub WalletConfigOverrides, rewrap, attrs = (derive(Debug, Default, Clone)))] #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WalletConfig { @@ -50,9 +69,8 @@ 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 calibration, should not be zero - #[serde(default = "default_calibration_limit")] - pub calibration_limit: usize, + #[serde(default = "MultiSequencerClientConfig::default")] + pub multi_sequencer_client_config: MultiSequencerClientConfig, } impl Default for WalletConfig { @@ -66,7 +84,7 @@ impl Default for WalletConfig { seq_tx_poll_max_blocks: 5, seq_poll_max_retries: 5, seq_block_poll_max_amount: 100, - calibration_limit: DEFAULT_CALLIBRATION_LIMIT, + multi_sequencer_client_config: MultiSequencerClientConfig::default(), } } } @@ -116,7 +134,7 @@ impl WalletConfig { seq_tx_poll_max_blocks, seq_poll_max_retries, seq_block_poll_max_amount, - calibration_limit, + multi_sequencer_client_config, } = self; let WalletConfigOverrides { @@ -125,7 +143,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, - calibration_limit: o_calibration_limit, + multi_sequencer_client_config: o_multi_sequencer_client_config, } = overrides; if let Some(v) = o_sequencers { @@ -148,13 +166,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_calibration_limit { - warn!("Overriding wallet config 'calibration_limit' to {v}"); - *calibration_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; } } } - -const fn default_calibration_limit() -> usize { - DEFAULT_CALLIBRATION_LIMIT -} diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index 7c8b8f69..ebe5028a 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}, }; @@ -85,6 +85,8 @@ pub enum ExecutionFailureKind { TransactionBuildError(#[from] lee::error::LeeError), #[error("Failed to sign transaction: {0}")] SignError(anyhow::Error), + #[error("Sending transaction failed for each client")] + MultiSequencerTransactionSendError, } pub struct WalletCore { @@ -173,7 +175,7 @@ impl WalletCore { let multi_sequencer_client = MultiSequencerClient::new( &config.sequencers, &mut statistics, - config.calibration_limit, + config.multi_sequencer_client_config.clone(), ) .await?; @@ -200,18 +202,32 @@ impl WalletCore { } #[must_use] - pub fn optimal_poller(&self) -> TxPoller { - TxPoller::new(self.config(), self.leader_owned()) + 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 leader_owned(&self) -> SequencerClient { - self.multi_sequencer_client.leader().clone() + pub fn poller_helm(&self) -> TxPoller { + TxPoller::new(self.config(), self.helm_owned()) } #[must_use] - pub fn leader_url(&self) -> Url { - self.multi_sequencer_client.leader_url().clone() + pub fn helm_owned(&self) -> SequencerClient { + self.multi_sequencer_client.helm().0.clone() + } + + #[must_use] + pub fn helm_url(&self) -> Url { + self.multi_sequencer_client.helm().1.clone() + } + + #[must_use] + pub fn leaders(&self) -> &[(SequencerClient, Url)] { + self.multi_sequencer_client.leaders() } /// Get storage. @@ -252,20 +268,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) - .await; + .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?; @@ -436,7 +447,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 { @@ -544,7 +555,7 @@ 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?) } @@ -552,7 +563,7 @@ impl WalletCore { pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result> { Ok(self .multi_sequencer_client - .metered_call(async |client: &SequencerClient| { + .metered_get(async |client: &SequencerClient| { client.get_accounts_nonces(accs.to_vec()).await }) .await?) @@ -574,14 +585,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?) } @@ -591,7 +602,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?) } @@ -599,7 +610,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?) } @@ -638,26 +649,23 @@ 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?) } /// 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_call(async |client: &SequencerClient| { - client.get_proofs_and_root(commitments.clone()).await + .metered_get(async |client: &SequencerClient| { + client.get_proofs_and_root(commitments.to_vec()).await }) .await?) } @@ -708,7 +716,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()?; @@ -722,7 +730,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 { @@ -800,12 +808,15 @@ 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 }) - .await; + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)?; Ok((call_res?, shared_secrets)) } @@ -868,12 +879,15 @@ 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 }) - .await?) + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??) } pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { @@ -882,12 +896,15 @@ 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 }) - .await?) + .await + .into_iter() + .find(std::result::Result::is_ok) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??) } pub async fn sync_to_latest_block(&mut self) -> Result { @@ -913,7 +930,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 499cf7a1..3c9da807 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -16,7 +16,7 @@ use serde::{Deserialize, Serialize}; use tokio::sync::RwLock; use url::Url; -use crate::config::SequencerConnectionData; +use crate::config::{MultiSequencerClientConfig, SequencerConnectionData}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Statistics { @@ -28,10 +28,12 @@ pub struct Statistics { } #[derive(Debug, Clone)] -pub struct StatisticsUpdate { - pub latency: f32, - pub new_latest_block_id: Option, - pub is_failed: bool, +pub enum StatisticsUpdate { + Success { + latency: f32, + new_latest_block_id: BlockId, + }, + Failure, } impl Statistics { @@ -76,23 +78,21 @@ impl Statistics { #[derive(Clone)] pub struct MultiSequencerClient { - // For now we store only leader, it is possible, that - // in future for important sends(for example for transactions) - // we would want to distribute call between known sequencers - leader: SequencerClient, - leader_url: Url, + /// Ordered list of leaders, from best to worst. + 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. - statistic_updates: Arc>>, + statistic_updates: Arc>>>, } impl MultiSequencerClient { async fn setup( conn_data: &[SequencerConnectionData], statistics: &mut HashMap, - calibration_limit: usize, - ) -> Result<(Url, SequencerClient)> { + multi_sequencer_client_config: &MultiSequencerClientConfig, + ) -> Result> { let mut client_list = HashMap::new(); for SequencerConnectionData { @@ -119,48 +119,62 @@ 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 + } 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); + // 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, + statistics, + multi_sequencer_client_config.distribution_limit, + ) + .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; - let res = choose_leader(&client_list, statistics) - .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?; + if leader_list.is_empty() { + anyhow::bail!( + "Leader search algorithm failure: Sorted out all clients during leader search" + ); + } - log::info!("Chosen leader is {:?}", res.0); + log::info!("Chosen leaders is {leader_list:?}"); - Ok(res) + 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, + config: multi_sequencer_client_config, + statistic_updates: Arc::new(RwLock::new(HashMap::new())), }) } @@ -169,55 +183,163 @@ 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>( + #[must_use] + pub const fn config(&self) -> &MultiSequencerClientConfig { + &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 up to a `self.config.distribution_limit`. + pub async fn metered_get Result>( &self, call: I, ) -> Result { - let (resp, statistics_update) = - tokio::join!(call(self.leader()), actualize_client(self.leader())); + // Collecting all statistics into one map to lock updates only once + let mut statistic_map: HashMap> = HashMap::new(); - log::debug!( - "Metered call for {:?}, metric updates is {:?}", - self.leader_url, - statistics_update - ); + // 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 + .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() { + 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_ok() { + break; + } + } + } { let mut statistic_updates_guard = self.statistic_updates.write().await; - statistic_updates_guard.push(statistics_update); + + #[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 } - /// 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.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 {leader_url:?}, statistic updates is {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; + + #[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 + } + + /// Update statistics of a leader, clear statistic updates log. + pub async fn update_statistics(&self, statistics: &mut HashMap) -> Result<()> { let mut statistic_updates = self.statistic_updates.write().await; - leader_statistic.apply_updates(statistic_updates.as_ref()); + + #[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) + .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(()) } } @@ -236,35 +358,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( @@ -350,18 +470,20 @@ 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] -pub fn choose_leader( +pub fn choose_leaders( client_list: &HashMap, statistics: &HashMap, -) -> Option<(Url, SequencerClient)> { + distribution_limit: usize, +) -> Option> { // Sort out all unmetered clients let mut client_vec: Vec<_> = client_list .keys() @@ -419,7 +541,7 @@ pub fn choose_leader( .expect("Ratios must be a valid numbers") }); - let client_vec = client_vec[..(client_vec + let mut client_vec = client_vec[..(client_vec .iter() .position(|item| { error_ratio( @@ -431,35 +553,46 @@ pub fn choose_leader( .to_vec(); // Choose clients with least latency and variance - let min_lat_var_addr = client_vec.iter().fold(client_vec[0], |acc, x| { - let old = statistics.get(acc).unwrap(); - let (old_lat, old_var) = (old.latency_avg, old.latency_var); - let new = statistics.get(*x).unwrap(); - let (new_lat, new_var) = (new.latency_avg, new.latency_var); + client_vec.sort_by(|a, b| { + let left = statistics.get(*a).unwrap(); + let (left_lat, left_var) = (left.latency_avg, left.latency_var); + let right = statistics.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 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] + // [-right_std < left_lat < right_lat < +left_std < +right_std] // // However one can argue that this: // - // [-old_std...................old_lat........new_lat.........+new_std.......+old_std] + // [-right_std < right_lat < left_lat < +left_std < +right_std] // // is still better, but it is up to discussion - if (new_lat <= old_lat) && ((new_lat + new_std) < (old_lat + old_std)) { - *x - } else { - acc + 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)) + } } }); - 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. @@ -535,13 +668,13 @@ 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; use crate::multi_client::{ - CumulativeUpdates, Statistics, StatisticsUpdate, choose_leader, cumulative_avg, + CumulativeUpdates, Statistics, StatisticsUpdate, choose_leaders, cumulative_avg, cumulative_var, }; @@ -564,24 +697,39 @@ 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![ - 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 { @@ -688,21 +836,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(); @@ -758,22 +900,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(); @@ -821,29 +948,16 @@ 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] 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(); @@ -891,29 +1005,16 @@ 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] 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(); @@ -961,29 +1062,16 @@ 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] 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(); @@ -1031,8 +1119,316 @@ 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] + 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); } } 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 f69c9c67..9a7df9db 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]; @@ -222,7 +222,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, - calibration_limit: 5, + multi_sequencer_client_config: MultiSequencerClientConfig { + distribution_limit: 1, + calibration_limit: 5, + }, }) }