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