logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-09 15:29:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 40fe9ff209d663710562b2996c854f6f642e732c into a0ba6008c38665882e160fb83a9f7c52bc1789aa

Browse Source
This commit is contained in:
Pravdyvy 2026-07-08 11:19:01 +00:00 committed by GitHub
commit c6fd8f99bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 451 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
lez/wallet/src/account_manager.rs
View File

@ -192,7 +192,7 @@ pub struct AccountManager {
impl AccountManager {
pub async fn new(
wallet: &WalletCore,
wallet: &mut WalletCore,
accounts: Vec<AccountIdentity>,
) -> Result<Self, ExecutionFailureKind> {
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<State, ExecutionFailureKind> {
@ -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<State, ExecutionFailureKind> {
@ -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<AccountPreparedData, ExecutionFailureKind> {
@ -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,

12
lez/wallet/src/cli/chain.rs
View File

@ -33,17 +33,23 @@ impl WalletSubcommand for ChainSubcommand {
) -> Result<SubcommandReturnValue> {
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:#?}");
}

15
lez/wallet/src/cli/config.rs
View File

@ -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<SubcommandReturnValue> {
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");
}

9
lez/wallet/src/cli/mod.rs
View File

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

2
lez/wallet/src/cli/programs/ata.rs
View File

@ -186,7 +186,7 @@ impl AtaSubcommand {
async fn handle_list(
owner: AccountId,
token_definition: Vec<AccountId>,
wallet_core: &WalletCore,
wallet_core: &mut WalletCore,
) -> Result<SubcommandReturnValue> {
let ata_program_id = programs::ata().id();

38
lez/wallet/src/config.rs
View File

@ -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<BasicAuth>,
}
#[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<SequencerConnectionData>,
/// 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<BasicAuth>,
}
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;
}
}
}

9
lez/wallet/src/helperfunctions.rs
View File

@ -66,6 +66,15 @@ pub fn fetch_persistent_storage_path() -> Result<PathBuf> {
Ok(accs_path)
}
/// Fetch path to metrics storage from default home.
///
/// File must be created through setup beforehand.
pub fn fetch_metrics_path() -> Result<PathBuf> {
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<Nonce> {
let mut result = vec![[0; 16]; size];

207
lez/wallet/src/lib.rs
View File

@ -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<Url, Metrics>,
metric_updates: Vec<MetricsUpdate>,
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<Self> {
pub async fn from_env() -> Result<Self> {
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<WalletConfigOverrides>,
) -> Result<Self> {
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<WalletConfigOverrides>,
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<WalletConfigOverrides>,
storage: Storage,
) -> Result<Self> {
@ -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<Metrics> {
todo!();
}
/// Get account balance.
pub async fn get_account_balance(&self, acc: AccountId) -> Result<u128> {
Ok(self.sequencer_client.get_account_balance(acc).await?)
pub async fn get_account_balance(&mut self, acc: AccountId) -> Result<u128> {
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<AccountId>) -> Result<Vec<Nonce>> {
Ok(self.sequencer_client.get_accounts_nonces(accs).await?)
pub async fn get_accounts_nonces(&mut self, accs: Vec<AccountId>) -> Result<Vec<Nonce>> {
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<Account> {
pub async fn get_account(&mut self, account_id: AccountIdWithPrivacy) -> Result<Account> {
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<Account> {
Ok(self.sequencer_client.get_account(account_id).await?)
pub async fn get_account_public(&mut self, account_id: AccountId) -> Result<Account> {
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<LeeTransaction> {
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<Option<MembershipProof>> {
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<Option<CommitmentSetDigest>> {
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<Option<CommitmentSetDigest>> {
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<AccountIdentity>,
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<AccountIdentity>,
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<AccountIdentity>,
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<AccountIdentity>,
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<u64> {
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));

19
lez/wallet/src/main.rs
View File

@ -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();

248
lez/wallet/src/multi_client.rs Normal file
View File

@ -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<HashMap<Url, Metrics>, 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<u64>,
pub is_failed: bool,
}
#[derive(Clone)]
pub struct MultiSequencerClient {
pub client_list: HashMap<Url, SequencerClient>,
pub leader: SequencerClient,
pub leader_url: Url,
}
impl MultiSequencerClient {
pub async fn new(
conn_data: &[SequencerConnectionData],
metrics: &mut HashMap<Url, Metrics>,
) -> Result<Self> {
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<R, E, I: AsyncFnOnce(&SequencerClient) -> Result<R, E>>(
&self,
call: I,
) -> (Result<R, E>, 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<Url, SequencerClient>,
_metrics: &HashMap<Url, Metrics>,
) -> (Url, SequencerClient) {
todo!()
}
pub fn save_metrics_at_path(
metrics: &HashMap<Url, Metrics>,
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<Url, Metrics>,
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)
}