diff --git a/lez/wallet/src/account_manager.rs b/lez/wallet/src/account_manager.rs index dae8fd31..65930e69 100644 --- a/lez/wallet/src/account_manager.rs +++ b/lez/wallet/src/account_manager.rs @@ -192,7 +192,7 @@ pub struct AccountManager { impl AccountManager { pub async fn new( - wallet: &WalletCore, + wallet: &mut WalletCore, accounts: Vec, ) -> Result { let mut states = Vec::with_capacity(accounts.len()); @@ -475,7 +475,7 @@ struct AccountPreparedData { } async fn prepare_public_state( - wallet: &WalletCore, + wallet: &mut WalletCore, account_id: AccountId, lookup_signing_key: bool, ) -> Result { @@ -493,7 +493,7 @@ async fn prepare_public_state( } async fn prepare_public_keycard_state( - wallet: &WalletCore, + wallet: &mut WalletCore, account_id: AccountId, key_path: String, ) -> Result { @@ -506,7 +506,7 @@ async fn prepare_public_keycard_state( } async fn private_key_tree_acc_preparation( - wallet: &WalletCore, + wallet: &mut WalletCore, account_id: AccountId, is_pda: bool, ) -> Result { @@ -520,16 +520,16 @@ async fn private_key_tree_acc_preparation( let from_npk = from_keys.nullifier_public_key; let from_vpk = from_keys.viewing_public_key.clone(); + // TODO: Technically we could allow unauthorized owned accounts, but currently we don't have + // support from that in the wallet. + let sender_pre = AccountWithMetadata::new(from_acc.account.clone(), true, account_id); + // TODO: Remove this unwrap, error types must be compatible let proof = wallet .check_private_account_initialized(account_id) .await .unwrap(); - // TODO: Technically we could allow unauthorized owned accounts, but currently we don't have - // support from that in the wallet. - let sender_pre = AccountWithMetadata::new(from_acc.account.clone(), true, account_id); - let eph_holder = EphemeralKeyHolder::new(&from_vpk); let ssk = eph_holder.calculate_shared_secret_sender(); let epk = eph_holder.ephemeral_public_key().clone(); @@ -548,7 +548,7 @@ async fn private_key_tree_acc_preparation( } async fn private_shared_acc_preparation( - wallet: &WalletCore, + wallet: &mut WalletCore, account_id: AccountId, nsk: NullifierSecretKey, npk: NullifierPublicKey, diff --git a/lez/wallet/src/cli/programs/ata.rs b/lez/wallet/src/cli/programs/ata.rs index 1d207b35..044ddb75 100644 --- a/lez/wallet/src/cli/programs/ata.rs +++ b/lez/wallet/src/cli/programs/ata.rs @@ -186,7 +186,7 @@ impl AtaSubcommand { async fn handle_list( owner: AccountId, token_definition: Vec, - wallet_core: &WalletCore, + wallet_core: &mut WalletCore, ) -> Result { let ata_program_id = programs::ata().id(); diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index caaf34c2..2741827f 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -7,7 +7,7 @@ reason = "Most of the shadows come from args parsing which is ok" )] -use std::path::PathBuf; +use std::{collections::HashMap, path::PathBuf}; pub use account_manager::AccountIdentity; use anyhow::{Context as _, Result}; @@ -29,11 +29,15 @@ use log::info; use sequencer_service_rpc::{RpcClient as _, SequencerClient}; use storage::Storage; use tokio::io::AsyncWriteExt as _; +use url::Url; use crate::{ account::{AccountIdWithPrivacy, Label}, config::WalletConfigOverrides, - multi_client::{Metrics, MultiSequencerClient}, + multi_client::{ + Metrics, MetricsUpdate, MultiSequencerClient, extract_metrics_from_path, + save_metrics_at_path_with_updates, + }, poller::TxPoller, storage::key_chain::SharedAccountEntry, }; @@ -94,22 +98,24 @@ pub struct WalletCore { storage: Storage, storage_path: PathBuf, - _metrics_path: PathBuf, + metrics_path: PathBuf, + metrics: HashMap, + metric_updates: Vec, pub multi_sequencer_client: MultiSequencerClient, } impl WalletCore { /// Construct wallet using [`HOME_DIR_ENV_VAR`] env var for paths or user home dir if not set. - pub fn from_env() -> Result { + pub async fn from_env() -> Result { let config_path = helperfunctions::fetch_config_path()?; let storage_path = helperfunctions::fetch_persistent_storage_path()?; let metrics_path = helperfunctions::fetch_metrics_path()?; - Self::new_update_chain(config_path, storage_path, metrics_path, None) + Self::new_update_chain(config_path, storage_path, metrics_path, None).await } - pub fn new_update_chain( + pub async fn new_update_chain( config_path: PathBuf, storage_path: PathBuf, metrics_path: PathBuf, @@ -125,9 +131,10 @@ impl WalletCore { config_overrides, storage, ) + .await } - pub fn new_init_storage( + pub async fn new_init_storage( config_path: PathBuf, storage_path: PathBuf, metrics_path: PathBuf, @@ -141,12 +148,13 @@ impl WalletCore { metrics_path, config_overrides, storage, - )?; + ) + .await?; Ok((wallet, mnemonic)) } - fn new( + async fn new( config_path: PathBuf, storage_path: PathBuf, metrics_path: PathBuf, @@ -164,7 +172,10 @@ impl WalletCore { config.apply_overrides(config_overrides); } - let multi_sequencer_client = MultiSequencerClient::new(&config.sequencers_conn_data)?; + let mut metrics = extract_metrics_from_path(&metrics_path)?; + + let multi_sequencer_client = + MultiSequencerClient::new(&config.sequencers_conn_data, &mut metrics).await?; Ok(Self { config_path, @@ -172,7 +183,9 @@ impl WalletCore { config, storage_path, storage, - _metrics_path: metrics_path, + metrics_path, + metrics, + metric_updates: vec![], multi_sequencer_client, }) } @@ -188,24 +201,7 @@ impl WalletCore { } pub fn optimal_poller(&self) -> TxPoller { - let metrics = self.get_metrics(); - - TxPoller::new( - self.config(), - self.multi_sequencer_client.optimal_client_clone(&metrics), - ) - } - - pub fn optimal_sequencer_client(&self) -> &SequencerClient { - let metrics = self.get_metrics(); - - self.multi_sequencer_client.optimal_client_ref(&metrics) - } - - pub fn optimal_sequencer_client_owned(&self) -> SequencerClient { - let metrics = self.get_metrics(); - - self.multi_sequencer_client.optimal_client_clone(&metrics) + TxPoller::new(self.config(), self.multi_sequencer_client.leader_clone()) } /// Get storage. @@ -244,6 +240,20 @@ impl WalletCore { Ok(()) } + pub fn store_metrics(&self) -> Result<()> { + save_metrics_at_path_with_updates( + self.metrics.clone(), + &self.multi_sequencer_client.leader_url, + &self.metric_updates, + &self.storage_path, + ) + .with_context(|| format!("Failed to store metrics at {}", self.storage_path.display()))?; + + println!("Stored metrics at {}", self.metrics_path.display()); + + Ok(()) + } + /// Store persistent data at home. pub async fn store_config_changes(&self) -> Result<()> { let config = serde_json::to_vec_pretty(&self.config)?; @@ -457,22 +467,28 @@ impl WalletCore { } /// Get account balance. - pub async fn get_account_balance(&self, acc: AccountId) -> Result { - Ok(self - .optimal_sequencer_client() - .get_account_balance(acc) - .await?) + pub async fn get_account_balance(&mut self, acc: AccountId) -> Result { + let call_f = async |client: &SequencerClient| client.get_account_balance(acc).await; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + Ok(call_res?) } /// Get accounts nonces. - pub async fn get_accounts_nonces(&self, accs: Vec) -> Result> { - Ok(self - .optimal_sequencer_client() - .get_accounts_nonces(accs) - .await?) + pub async fn get_accounts_nonces(&mut self, accs: Vec) -> Result> { + let call_f = async |client: &SequencerClient| client.get_accounts_nonces(accs).await; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + Ok(call_res?) } - pub async fn get_account(&self, account_id: AccountIdWithPrivacy) -> Result { + pub async fn get_account(&mut self, account_id: AccountIdWithPrivacy) -> Result { match account_id { AccountIdWithPrivacy::Public(acc_id) => self.get_account_public(acc_id).await, AccountIdWithPrivacy::Private(acc_id) => { @@ -486,11 +502,14 @@ impl WalletCore { } /// Get public account. - pub async fn get_account_public(&self, account_id: AccountId) -> Result { - Ok(self - .optimal_sequencer_client() - .get_account(account_id) - .await?) + pub async fn get_account_public(&mut self, account_id: AccountId) -> Result { + let call_f = async |client: &SequencerClient| client.get_account(account_id).await; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + Ok(call_res?) } #[must_use] @@ -531,25 +550,32 @@ impl WalletCore { } pub async fn check_private_account_initialized( - &self, + &mut self, account_id: AccountId, ) -> Result> { if let Some(acc_comm) = self.get_private_account_commitment(account_id) { - self.optimal_sequencer_client() - .get_proof_for_commitment(acc_comm) - .await - .map_err(Into::into) + let call_f = + async |client: &SequencerClient| client.get_proof_for_commitment(acc_comm).await; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + call_res.map_err(Into::into) } else { Ok(None) } } - pub async fn get_commitment_root(&self) -> Result> { - let proof = self - .optimal_sequencer_client() - .get_proof_for_commitment(DUMMY_COMMITMENT) - .await?; - Ok(proof.map(|p| compute_digest_for_path(&DUMMY_COMMITMENT, &p))) + pub async fn get_commitment_root(&mut self) -> Result> { + let call_f = async |client: &SequencerClient| { + client.get_proof_for_commitment(DUMMY_COMMITMENT).await + }; + + let (proof, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + Ok(proof?.map(|p| compute_digest_for_path(&DUMMY_COMMITMENT, &p))) } pub fn decode_insert_privacy_preserving_transaction_results( @@ -615,7 +641,7 @@ impl WalletCore { } pub async fn send_privacy_preserving_tx( - &self, + &mut self, accounts: Vec, instruction_data: InstructionData, program: &ProgramWithDependencies, @@ -627,7 +653,7 @@ impl WalletCore { } pub async fn send_privacy_preserving_tx_with_pre_check( - &self, + &mut self, accounts: Vec, instruction_data: InstructionData, program: &ProgramWithDependencies, @@ -677,16 +703,21 @@ impl WalletCore { .map(|keys| keys.ssk) .collect(); - Ok(( - self.optimal_sequencer_client() + let call_f = async |client: &SequencerClient| { + client .send_transaction(LeeTransaction::PrivacyPreserving(tx)) - .await?, - shared_secrets, - )) + .await + }; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + Ok((call_res?, shared_secrets)) } pub async fn send_pub_tx( - &self, + &mut self, accounts: Vec, instruction_data: InstructionData, program_id: ProgramId, @@ -696,7 +727,7 @@ impl WalletCore { } pub async fn send_pub_tx_with_pre_check( - &self, + &mut self, accounts: Vec, instruction_data: InstructionData, program_id: ProgramId, @@ -741,14 +772,25 @@ impl WalletCore { let tx = lee::public_transaction::PublicTransaction::new(message, witness_set); - Ok(self - .optimal_sequencer_client() - .send_transaction(LeeTransaction::Public(tx)) - .await?) + let call_f = async |client: &SequencerClient| { + client.send_transaction(LeeTransaction::Public(tx)).await + }; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + Ok(call_res?) } pub async fn sync_to_latest_block(&mut self) -> Result { - let latest_block_id = self.optimal_sequencer_client().get_last_block_id().await?; + let call_f = async |client: &SequencerClient| client.get_last_block_id().await; + + let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; + + self.metric_updates.push(metrics_update); + + let latest_block_id = call_res?; println!("Latest block is {latest_block_id}"); self.sync_to_block(latest_block_id).await?; Ok(latest_block_id) diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 01c3ab18..a79be6d0 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -1,24 +1,57 @@ +use std::{collections::HashMap, io::Write, path::Path}; + use anyhow::{Context, Result}; -use sequencer_service_rpc::{SequencerClient, SequencerClientBuilder}; +use sequencer_service_rpc::{RpcClient, SequencerClient, SequencerClientBuilder}; use serde::{Deserialize, Serialize}; +use url::Url; use crate::config::SequencerConnectionData; +pub const CALLIBRATION_LIMIT: usize = 100; + +pub fn extract_metrics_from_path(path: &Path) -> Result, anyhow::Error> { + match std::fs::File::open(path) { + Ok(file) => { + let reader = std::io::BufReader::new(file); + Ok(serde_json::from_reader(reader)?) + } + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + println!("Metrics not found, choosing empty"); + Ok(HashMap::new()) + } + Err(err) => Err(err).context("IO error"), + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Metrics { - pub latency_avg: u64, - pub latency_var: u64, - pub last_block_id: u64, + pub latency_avg: f32, + pub latency_var: f32, + pub sample_size: usize, + pub latest_block_id: u64, + pub errors: u64, +} + +#[derive(Debug, Clone)] +pub struct MetricsUpdate { + pub latency: f32, + pub new_latest_block_id: Option, + pub is_failed: bool, } #[derive(Clone)] pub struct MultiSequencerClient { - pub client_list: Vec, + pub client_list: HashMap, + pub leader: SequencerClient, + pub leader_url: Url, } impl MultiSequencerClient { - pub fn new(conn_data: &[SequencerConnectionData]) -> Result { - let mut client_list = vec![]; + pub async fn new( + conn_data: &[SequencerConnectionData], + metrics: &mut HashMap, + ) -> Result { + let mut client_list = HashMap::new(); for SequencerConnectionData { sequencer_addr, @@ -43,17 +76,173 @@ impl MultiSequencerClient { .context("Failed to create sequencer client")? }; - client_list.push(sequencer_client); + // If there is no metrics for client, callibrate it + if !metrics.contains_key(sequencer_addr) { + metrics.insert( + sequencer_addr.clone(), + callibration(sequencer_client.clone()).await, + ); + } + + client_list.insert(sequencer_addr.clone(), sequencer_client); } - Ok(Self { client_list }) + let (leader_url, leader) = choose_leader(&client_list, metrics); + + Ok(Self { + client_list, + leader, + leader_url, + }) } - pub fn optimal_client_ref(&self, _metrics: &[Metrics]) -> &SequencerClient { - todo!(); + pub fn leader_ref(&self) -> &SequencerClient { + &self.leader } - pub fn optimal_client_clone(&self, _metrics: &[Metrics]) -> SequencerClient { - todo!(); + pub fn leader_clone(&self) -> SequencerClient { + self.leader.clone() + } + + pub async fn metered_call Result>( + &self, + call: I, + ) -> (Result, MetricsUpdate) { + let call_last_block = self.leader_ref().get_last_block_id(); + + let now = tokio::time::Instant::now(); + + let (call_future_res, call_last_block_res) = + tokio::join!(call(self.leader_ref()), call_last_block); + + let latency = tokio::time::Instant::now().duration_since(now).as_millis() as f32; + let is_failed = call_future_res.is_err() || call_last_block_res.is_err(); + let mut new_last_block = None; + + if let Ok(last_block) = call_last_block_res { + new_last_block = Some(last_block); + } + + let metrics_update = MetricsUpdate { + latency, + new_latest_block_id: new_last_block, + is_failed, + }; + + (call_future_res, metrics_update) } } + +pub async fn callibration(client: SequencerClient) -> Metrics { + let mut latencies = vec![]; + let mut latest_block_id = 0; + let mut errors = 0; + + for _ in 0..CALLIBRATION_LIMIT { + let now = tokio::time::Instant::now(); + + let block_id = client.get_last_block_id().await; + + let latency = tokio::time::Instant::now().duration_since(now).as_millis(); + + let Ok(block_id) = block_id else { + errors += 1; + continue; + }; + + latest_block_id = block_id; + latencies.push(latency); + } + + // Precision loss if fine there + let sample_size = latencies.len(); + let latency_avg = (latencies.iter().fold(0, |acc, x| acc + x) as f32) / (sample_size as f32); + let latency_var = (latencies.iter().fold(0f32, |acc, x| { + acc + ((*x as f32) - latency_avg) * ((*x as f32) - latency_avg) + }) / (sample_size as f32)) + .sqrt(); + + Metrics { + latency_avg, + latency_var, + sample_size, + latest_block_id, + errors, + } +} + +pub fn choose_leader( + _client_list: &HashMap, + _metrics: &HashMap, +) -> (Url, SequencerClient) { + todo!() +} + +pub fn save_metrics_at_path( + metrics: &HashMap, + path: &Path, +) -> Result<(), anyhow::Error> { + let metrics_serialized = serde_json::to_vec_pretty(metrics)?; + let mut file = std::fs::File::create(path).context("Failed to create file")?; + file.write_all(&metrics_serialized) + .context("Failed to write to file")?; + file.sync_all().context("Failed to sync file")?; + + Ok(()) +} + +pub fn save_metrics_at_path_with_updates( + mut metrics: HashMap, + leader_url: &Url, + metric_updates: &[MetricsUpdate], + path: &Path, +) -> Result<(), anyhow::Error> { + let leader_metric = metrics + .get_mut(leader_url) + .ok_or(anyhow::anyhow!("Leader URL is in present in metrics"))?; + + let (failure_count, latest_block_id, cumulative_latency) = + metric_updates.iter().fold((0u64, None, 0f32), |acc, x| { + let MetricsUpdate { + latency, + new_latest_block_id, + is_failed, + } = x; + ( + if *is_failed { acc.0 + 1 } else { acc.0 }, + acc.1.map(|inner| { + if let Some(new_latest_block_id) = new_latest_block_id { + if inner < *new_latest_block_id { + *new_latest_block_id + } else { + inner + } + } else { + inner + } + }), + acc.2 + latency, + ) + }); + + leader_metric.errors += failure_count; + if let Some(latest_block_id) = latest_block_id { + leader_metric.latest_block_id = latest_block_id; + } + + let orig_size_f = leader_metric.sample_size as f32; + let mod_size_f = (leader_metric.sample_size + metric_updates.len()) as f32; + + let latency_avg_old = leader_metric.latency_avg; + let latency_avg_new = + (leader_metric.latency_avg * orig_size_f + cumulative_latency) / mod_size_f; + + let latency_disp_new = (latency_avg_old - latency_avg_new) + * ((3f32 * latency_avg_old + latency_avg_new) * orig_size_f / mod_size_f) + + (leader_metric.latency_var * leader_metric.latency_var) * orig_size_f / mod_size_f; + + leader_metric.latency_avg = latency_avg_new; + leader_metric.latency_var = latency_disp_new.sqrt(); + + save_metrics_at_path(&metrics, path) +}