logos-blockchain/lssa
1
0
Fork 0
mirror of https://github.com/logos-blockchain/lssa.git synced 2026-07-23 16:13:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(wallet_ffi): multi-poller

Browse Source
This commit is contained in:
Pravdyvy 2026-07-21 18:01:04 +03:00
parent 4968db3dd7
commit 0c32025c4b
13 changed files with 177 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/program_deployment/src/bin/run_hello_world.rs
View File

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

2
examples/program_deployment/src/bin/run_hello_world_through_tail_call.rs
View File

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

4
examples/program_deployment/src/bin/run_hello_world_with_authorization.rs
View File

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

2
examples/program_deployment/src/bin/run_hello_world_with_authorization_through_tail_call_with_pda.rs
View File

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

4
examples/program_deployment/src/bin/run_hello_world_with_move_function.rs
View File

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

2
integration_tests/tests/private_pda.rs
View File

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

2
lez/wallet-ffi/src/wallet.rs
View File

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

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

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

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

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

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

@ -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<TxPoller> {
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<AccountId>) -> Result<Vec<Nonce>> {
pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result<Vec<Nonce>> {
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<Commitment>,
commitments: &[Commitment],
) -> Result<(Vec<Option<MembershipProof>>, 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<cli::SubcommandReturnValue> {
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<cli::SubcommandReturnValue> {
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<u8>) -> Result<HashType> {
@ -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<u64> {
@ -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));

134
lez/wallet/src/multi_client.rs
View File

@ -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<RwLock> 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<Url, Statistics>,
multi_sequencer_client_config: MultiSequencerClientConfig,
) -> Result<Self> {
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<R, E, I: AsyncFnOnce(&SequencerClient) -> Result<R, E>>(
///
/// If current leader errors, we ask next one in list.
pub async fn metered_get<R, E, I: AsyncFn(&SequencerClient) -> Result<R, E>>(
&self,
call: I,
) -> Result<R, E> {
// Collecting all statistics into one map to lock updates only once
let mut statistic_map: HashMap<Url, Vec<StatisticsUpdate>> = 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<R, E, I: AsyncFn(&SequencerClient) -> Result<R, E>>(
&self,
call: I,
) -> Vec<Result<R, E>> {
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<Url, Vec<StatisticsUpdate>> = 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<Url, Statistics>,) {
pub async fn update_statistics(&self, statistics: &mut HashMap<Url, Statistics>) {
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);
}

21
lez/wallet/src/poller.rs
View File

@ -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<TxPoller>,
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")
}

2
test_fixtures/src/config.rs
View File

@ -205,7 +205,7 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result<WalletConfig> {
seq_block_poll_max_amount: 100,
multi_sequencer_client_config: MultiSequencerClientConfig {
distribution_limit: 1,
callibration_limit: 5,
calibration_limit: 5,
},
})
}