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