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/chain.rs b/lez/wallet/src/cli/chain.rs index dfb22eba..38ec3d7a 100644 --- a/lez/wallet/src/cli/chain.rs +++ b/lez/wallet/src/cli/chain.rs @@ -33,17 +33,23 @@ impl WalletSubcommand for ChainSubcommand { ) -> Result { match self { Self::CurrentBlockId => { - let latest_block_id = wallet_core.sequencer_client.get_last_block_id().await?; + let latest_block_id = wallet_core + .optimal_sequencer_client() + .get_last_block_id() + .await?; println!("Last block id is {latest_block_id}"); } Self::Block { id } => { - let block = wallet_core.sequencer_client.get_block(id).await?; + let block = wallet_core.optimal_sequencer_client().get_block(id).await?; println!("Last block id is {block:#?}"); } Self::Transaction { hash } => { - let tx = wallet_core.sequencer_client.get_transaction(hash).await?; + let tx = wallet_core + .optimal_sequencer_client() + .get_transaction(hash) + .await?; println!("Transaction is {tx:#?}"); } diff --git a/lez/wallet/src/cli/config.rs b/lez/wallet/src/cli/config.rs index a78bd097..b6d0c523 100644 --- a/lez/wallet/src/cli/config.rs +++ b/lez/wallet/src/cli/config.rs @@ -37,7 +37,7 @@ impl ConfigSubcommand { } else if let Some(key) = key { match key.as_str() { "sequencer_addr" => { - println!("{}", config.sequencer_addr); + println!("{:?}", config.sequencers_conn_data); } "seq_poll_timeout" => { println!("{:?}", config.seq_poll_timeout); @@ -51,13 +51,6 @@ impl ConfigSubcommand { "seq_block_poll_max_amount" => { println!("{}", config.seq_block_poll_max_amount); } - "basic_auth" => { - if let Some(basic_auth) = &config.basic_auth { - println!("{basic_auth}"); - } else { - println!("Not set"); - } - } _ => { println!("Unknown field"); } @@ -76,9 +69,6 @@ impl ConfigSubcommand { ) -> Result { let mut config = wallet_core.config().clone(); match key.as_str() { - "sequencer_addr" => { - config.sequencer_addr = value.parse()?; - } "seq_poll_timeout" => { config.seq_poll_timeout = humantime::parse_duration(&value) .map_err(|e| anyhow::anyhow!("Invalid duration: {e}"))?; @@ -92,9 +82,6 @@ impl ConfigSubcommand { "seq_block_poll_max_amount" => { config.seq_block_poll_max_amount = value.parse()?; } - "basic_auth" => { - config.basic_auth = Some(value.parse()?); - } "initial_accounts" => { anyhow::bail!("Setting this field from wallet is not supported"); } diff --git a/lez/wallet/src/cli/mod.rs b/lez/wallet/src/cli/mod.rs index 43b6dc7a..4b460714 100644 --- a/lez/wallet/src/cli/mod.rs +++ b/lez/wallet/src/cli/mod.rs @@ -219,7 +219,7 @@ pub async fn execute_subcommand( } Command::CheckHealth => { let remote_program_ids = wallet_core - .sequencer_client + .optimal_sequencer_client() .get_program_ids() .await .expect("Error fetching program ids"); @@ -287,7 +287,7 @@ pub async fn execute_subcommand( let message = lee::program_deployment_transaction::Message::new(bytecode); let transaction = ProgramDeploymentTransaction::new(message); let _response = wallet_core - .sequencer_client + .optimal_sequencer_client() .send_transaction(LeeTransaction::ProgramDeployment(transaction)) .await .context("Transaction submission error")?; @@ -384,12 +384,13 @@ pub async fn execute_keys_restoration(wallet_core: &mut WalletCore, depth: u32) wallet_core.sync_to_latest_block().await?; + let optimal_sequencer_client = wallet_core.optimal_sequencer_client_owned(); + wallet_core .storage .key_chain_mut() .cleanup_trees_remove_uninit_layered(depth, |account_id| { - wallet_core - .sequencer_client + optimal_sequencer_client .get_account(account_id) .map_err(Into::into) }) 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/config.rs b/lez/wallet/src/config.rs index 77c2729f..68f65f52 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -7,6 +7,15 @@ use log::warn; use serde::{Deserialize, Serialize}; use url::Url; +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SequencerConnectionData { + /// Connection data of all known sequencers. + pub sequencer_addr: Url, + /// Basic authentication credentials + #[serde(skip_serializing_if = "Option::is_none")] + pub basic_auth: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct GasConfig { /// Gas spent per deploying one byte of data. @@ -28,8 +37,8 @@ pub struct GasConfig { #[optfield::optfield(pub WalletConfigOverrides, rewrap, attrs = (derive(Debug, Default, Clone)))] #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WalletConfig { - /// Sequencer URL. - pub sequencer_addr: Url, + /// Connection data of all known sequencers. + pub sequencers_conn_data: Vec, /// Sequencer polling duration for new blocks. #[serde(with = "humantime_serde")] pub seq_poll_timeout: Duration, @@ -39,20 +48,19 @@ 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, - /// Basic authentication credentials - #[serde(skip_serializing_if = "Option::is_none")] - pub basic_auth: Option, } impl Default for WalletConfig { fn default() -> Self { Self { - sequencer_addr: "http://127.0.0.1:3040".parse().unwrap(), + sequencers_conn_data: vec![SequencerConnectionData { + sequencer_addr: "http://127.0.0.1:3040".parse().unwrap(), + basic_auth: None, + }], seq_poll_timeout: Duration::from_secs(12), seq_tx_poll_max_blocks: 5, seq_poll_max_retries: 5, seq_block_poll_max_amount: 100, - basic_auth: None, } } } @@ -97,26 +105,24 @@ impl WalletConfig { pub fn apply_overrides(&mut self, overrides: WalletConfigOverrides) { let Self { - sequencer_addr, + sequencers_conn_data, seq_poll_timeout, seq_tx_poll_max_blocks, seq_poll_max_retries, seq_block_poll_max_amount, - basic_auth, } = self; let WalletConfigOverrides { - sequencer_addr: o_sequencer_addr, + sequencers_conn_data: o_sequencers_conn_data, seq_poll_timeout: o_seq_poll_timeout, 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, - basic_auth: o_basic_auth, } = overrides; - if let Some(v) = o_sequencer_addr { - warn!("Overriding wallet config 'sequencer_addr' to {v}"); - *sequencer_addr = v; + if let Some(v) = o_sequencers_conn_data { + warn!("Overriding wallet config 'sequencers_conn_data' to {v:?}"); + *sequencers_conn_data = v; } if let Some(v) = o_seq_poll_timeout { warn!("Overriding wallet config 'seq_poll_timeout' to {v:?}"); @@ -134,9 +140,5 @@ impl WalletConfig { warn!("Overriding wallet config 'seq_block_poll_max_amount' to {v}"); *seq_block_poll_max_amount = v; } - if let Some(v) = o_basic_auth { - warn!("Overriding wallet config 'basic_auth' to {v:#?}"); - *basic_auth = v; - } } } diff --git a/lez/wallet/src/helperfunctions.rs b/lez/wallet/src/helperfunctions.rs index 6fe78681..3898d94c 100644 --- a/lez/wallet/src/helperfunctions.rs +++ b/lez/wallet/src/helperfunctions.rs @@ -66,6 +66,15 @@ pub fn fetch_persistent_storage_path() -> Result { Ok(accs_path) } +/// Fetch path to metrics storage from default home. +/// +/// File must be created through setup beforehand. +pub fn fetch_metrics_path() -> Result { + let home = get_home()?; + let metrics_path = home.join("metrics.json"); + Ok(metrics_path) +} + #[expect(dead_code, reason = "Maybe used later")] pub(crate) fn produce_random_nonces(size: usize) -> Vec { let mut result = vec![[0; 16]; size]; diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index 923f55ae..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}; @@ -26,13 +26,18 @@ use lee_core::{ account::Nonce, compute_digest_for_path, program::InstructionData, }; use log::info; -use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; +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, MetricsUpdate, MultiSequencerClient, extract_metrics_from_path, + save_metrics_at_path_with_updates, + }, poller::TxPoller, storage::key_chain::SharedAccountEntry, }; @@ -42,6 +47,7 @@ mod account_manager; pub mod cli; pub mod config; pub mod helperfunctions; +pub mod multi_client; pub mod poller; pub mod program_facades; pub mod signing; @@ -92,45 +98,66 @@ pub struct WalletCore { storage: Storage, storage_path: PathBuf, - poller: TxPoller, - pub sequencer_client: SequencerClient, + 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, 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, config_overrides: Option, ) -> Result { let storage = Storage::from_path(&storage_path) .with_context(|| format!("Failed to load storage from {}", storage_path.display()))?; - Self::new(config_path, storage_path, config_overrides, storage) + Self::new( + config_path, + storage_path, + metrics_path, + config_overrides, + storage, + ) + .await } - pub fn new_init_storage( + pub async fn new_init_storage( config_path: PathBuf, storage_path: PathBuf, + metrics_path: PathBuf, config_overrides: Option, password: &str, ) -> Result<(Self, Mnemonic)> { let (storage, mnemonic) = Storage::new(password).context("Failed to create storage")?; - let wallet = Self::new(config_path, storage_path, config_overrides, storage)?; + let wallet = Self::new( + config_path, + storage_path, + metrics_path, + config_overrides, + storage, + ) + .await?; Ok((wallet, mnemonic)) } - fn new( + async fn new( config_path: PathBuf, storage_path: PathBuf, + metrics_path: PathBuf, config_overrides: Option, storage: Storage, ) -> Result { @@ -145,25 +172,10 @@ impl WalletCore { config.apply_overrides(config_overrides); } - let sequencer_client = { - let mut builder = SequencerClientBuilder::default(); - if let Some(basic_auth) = &config.basic_auth { - builder = builder.set_headers( - std::iter::once(( - "Authorization".parse().expect("Header name is valid"), - format!("Basic {basic_auth}") - .parse() - .context("Invalid basic auth format")?, - )) - .collect(), - ); - } - builder - .build(config.sequencer_addr.clone()) - .context("Failed to create sequencer client")? - }; + let mut metrics = extract_metrics_from_path(&metrics_path)?; - let tx_poller = TxPoller::new(&config, sequencer_client.clone()); + let multi_sequencer_client = + MultiSequencerClient::new(&config.sequencers_conn_data, &mut metrics).await?; Ok(Self { config_path, @@ -171,8 +183,10 @@ impl WalletCore { config, storage_path, storage, - poller: tx_poller, - sequencer_client, + metrics_path, + metrics, + metric_updates: vec![], + multi_sequencer_client, }) } @@ -186,6 +200,10 @@ impl WalletCore { self.config = config; } + pub fn optimal_poller(&self) -> TxPoller { + TxPoller::new(self.config(), self.multi_sequencer_client.leader_clone()) + } + /// Get storage. #[must_use] pub const fn storage(&self) -> &Storage { @@ -222,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)?; @@ -430,17 +462,33 @@ impl WalletCore { }) } + pub fn get_metrics(&self) -> Vec { + todo!(); + } + /// Get account balance. - pub async fn get_account_balance(&self, acc: AccountId) -> Result { - Ok(self.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.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) => { @@ -454,8 +502,14 @@ impl WalletCore { } /// Get public account. - pub async fn get_account_public(&self, account_id: AccountId) -> Result { - Ok(self.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] @@ -492,29 +546,36 @@ impl WalletCore { /// Poll transactions. pub async fn poll_native_token_transfer(&self, hash: HashType) -> Result { - self.poller.poll_tx(hash).await + self.optimal_poller().poll_tx(hash).await } 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.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 - .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( @@ -580,7 +641,7 @@ impl WalletCore { } pub async fn send_privacy_preserving_tx( - &self, + &mut self, accounts: Vec, instruction_data: InstructionData, program: &ProgramWithDependencies, @@ -592,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, @@ -642,16 +703,21 @@ impl WalletCore { .map(|keys| keys.ssk) .collect(); - Ok(( - self.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, @@ -661,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, @@ -706,14 +772,25 @@ impl WalletCore { let tx = lee::public_transaction::PublicTransaction::new(message, witness_set); - Ok(self - .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.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) @@ -735,7 +812,7 @@ impl WalletCore { println!("Syncing to block {block_id}. Blocks to sync: {num_of_blocks}"); - let poller = self.poller.clone(); + let poller = self.optimal_poller().clone(); let mut blocks = std::pin::pin!(poller.poll_block_range(last_synced_block.saturating_add(1)..=block_id)); diff --git a/lez/wallet/src/main.rs b/lez/wallet/src/main.rs index 838ee856..d3c27427 100644 --- a/lez/wallet/src/main.rs +++ b/lez/wallet/src/main.rs @@ -8,8 +8,7 @@ use clap::{CommandFactory as _, Parser as _}; use wallet::{ WalletCore, cli::{Args, execute_continuous_run, execute_subcommand, read_password_from_stdin}, - config::WalletConfigOverrides, - helperfunctions::{fetch_config_path, fetch_persistent_storage_path}, + helperfunctions::{fetch_config_path, fetch_metrics_path, fetch_persistent_storage_path}, }; // TODO #169: We have sample configs for sequencer, but not for wallet @@ -21,7 +20,7 @@ use wallet::{ async fn main() -> Result<()> { let Args { continuous_run, - auth, + auth: _auth, command, } = Args::parse(); @@ -30,16 +29,11 @@ async fn main() -> Result<()> { let config_path = fetch_config_path().context("Could not fetch config path")?; let storage_path = fetch_persistent_storage_path().context("Could not fetch persistent storage path")?; - - // Override basic auth if provided via CLI - let config_overrides = WalletConfigOverrides { - basic_auth: auth.map(|auth| auth.parse()).transpose()?.map(Some), - ..Default::default() - }; + let metrics_path = fetch_metrics_path().context("Could not fetch metrics path")?; if let Some(command) = command { let mut wallet = if storage_path.exists() { - WalletCore::new_update_chain(config_path, storage_path, Some(config_overrides))? + WalletCore::new_update_chain(config_path, storage_path, metrics_path, None)? } else { // TODO: Maybe move to `WalletCore::from_env()` or similar? @@ -49,7 +43,8 @@ async fn main() -> Result<()> { let (wallet, mnemonic) = WalletCore::new_init_storage( config_path, storage_path, - Some(config_overrides), + metrics_path, + None, &password, )?; @@ -68,7 +63,7 @@ async fn main() -> Result<()> { Ok(()) } else if continuous_run { let mut wallet = - WalletCore::new_update_chain(config_path, storage_path, Some(config_overrides))?; + WalletCore::new_update_chain(config_path, storage_path, metrics_path, None)?; execute_continuous_run(&mut wallet).await } else { let help = Args::command().render_long_help(); diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs new file mode 100644 index 00000000..a79be6d0 --- /dev/null +++ b/lez/wallet/src/multi_client.rs @@ -0,0 +1,248 @@ +use std::{collections::HashMap, io::Write, path::Path}; + +use anyhow::{Context, Result}; +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: 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: HashMap, + pub leader: SequencerClient, + pub leader_url: Url, +} + +impl MultiSequencerClient { + pub async fn new( + conn_data: &[SequencerConnectionData], + metrics: &mut HashMap, + ) -> Result { + let mut client_list = HashMap::new(); + + for SequencerConnectionData { + sequencer_addr, + basic_auth, + } in conn_data + { + let sequencer_client = { + let mut builder = SequencerClientBuilder::default(); + if let Some(basic_auth) = &basic_auth { + builder = builder.set_headers( + std::iter::once(( + "Authorization".parse().expect("Header name is valid"), + format!("Basic {basic_auth}") + .parse() + .context("Invalid basic auth format")?, + )) + .collect(), + ); + } + builder + .build(sequencer_addr) + .context("Failed to create 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); + } + + let (leader_url, leader) = choose_leader(&client_list, metrics); + + Ok(Self { + client_list, + leader, + leader_url, + }) + } + + pub fn leader_ref(&self) -> &SequencerClient { + &self.leader + } + + 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) +}