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

Merge pull request #623 from logos-blockchain/Pravdyvy/request-distribution-for-multiple-sequencers

Browse Source 
feat(wallet)!: Request distribution for multiple sequencers
This commit is contained in:
Pravdyvy 2026-07-24 12:50:22 +03:00 committed by GitHub
commit 7e5728b535
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 718 additions and 267 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();

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

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

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

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

@ -387,7 +387,7 @@ pub async fn execute_keys_restoration(wallet_core: &mut WalletCore, depth: u32)
wallet_core.sync_to_latest_block().await?;
let leader_client = wallet_core.leader_owned();
let leader_client = wallet_core.helm_owned();
wallet_core
.storage

40
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 {
@ -36,6 +37,24 @@ pub struct GasConfig {
pub gas_limit_runtime: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MultiSequencerClientConfig {
/// Maximum numbers of sequencers to send requests. Client can have AT MOST
/// `distribution_limit` active clients.
pub distribution_limit: usize,
/// 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: DEFAULT_DISTRIBUTION_LIMIT,
calibration_limit: DEFAULT_CALLIBRATION_LIMIT,
}
}
}
#[optfield::optfield(pub WalletConfigOverrides, rewrap, attrs = (derive(Debug, Default, Clone)))]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WalletConfig {
@ -50,9 +69,8 @@ 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,
/// Limit number of sequencer polls during calibration, should not be zero
#[serde(default = "default_calibration_limit")]
pub calibration_limit: usize,
#[serde(default = "MultiSequencerClientConfig::default")]
pub multi_sequencer_client_config: MultiSequencerClientConfig,
}
impl Default for WalletConfig {
@ -66,7 +84,7 @@ impl Default for WalletConfig {
seq_tx_poll_max_blocks: 5,
seq_poll_max_retries: 5,
seq_block_poll_max_amount: 100,
calibration_limit: DEFAULT_CALLIBRATION_LIMIT,
multi_sequencer_client_config: MultiSequencerClientConfig::default(),
}
}
}
@ -116,7 +134,7 @@ impl WalletConfig {
seq_tx_poll_max_blocks,
seq_poll_max_retries,
seq_block_poll_max_amount,
calibration_limit,
multi_sequencer_client_config,
} = self;
let WalletConfigOverrides {
@ -125,7 +143,7 @@ impl WalletConfig {
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,
calibration_limit: o_calibration_limit,
multi_sequencer_client_config: o_multi_sequencer_client_config,
} = overrides;
if let Some(v) = o_sequencers {
@ -148,13 +166,9 @@ impl WalletConfig {
warn!("Overriding wallet config 'seq_block_poll_max_amount' to {v}");
*seq_block_poll_max_amount = v;
}
if let Some(v) = o_calibration_limit {
warn!("Overriding wallet config 'calibration_limit' to {v}");
*calibration_limit = v;
if let Some(v) = o_multi_sequencer_client_config {
warn!("Overriding wallet config 'multi_sequencer_client_config' to {v:?}");
*multi_sequencer_client_config = v;
}
}
}
const fn default_calibration_limit() -> usize {
DEFAULT_CALLIBRATION_LIMIT
}

99
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},
};
@ -85,6 +85,8 @@ pub enum ExecutionFailureKind {
TransactionBuildError(#[from] lee::error::LeeError),
#[error("Failed to sign transaction: {0}")]
SignError(anyhow::Error),
#[error("Sending transaction failed for each client")]
MultiSequencerTransactionSendError,
}
pub struct WalletCore {
@ -173,7 +175,7 @@ impl WalletCore {
let multi_sequencer_client = MultiSequencerClient::new(
&config.sequencers,
&mut statistics,
config.calibration_limit,
config.multi_sequencer_client_config.clone(),
)
.await?;
@ -200,18 +202,32 @@ impl WalletCore {
}
#[must_use]
pub fn optimal_poller(&self) -> TxPoller {
TxPoller::new(self.config(), self.leader_owned())
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 leader_owned(&self) -> SequencerClient {
self.multi_sequencer_client.leader().clone()
pub fn poller_helm(&self) -> TxPoller {
TxPoller::new(self.config(), self.helm_owned())
}
#[must_use]
pub fn leader_url(&self) -> Url {
self.multi_sequencer_client.leader_url().clone()
pub fn helm_owned(&self) -> SequencerClient {
self.multi_sequencer_client.helm().0.clone()
}
#[must_use]
pub fn helm_url(&self) -> Url {
self.multi_sequencer_client.helm().1.clone()
}
#[must_use]
pub fn leaders(&self) -> &[(SequencerClient, Url)] {
self.multi_sequencer_client.leaders()
}
/// Get storage.
@ -252,20 +268,15 @@ impl WalletCore {
/// Rotates multi-client and stores metrics.
pub async fn client_rotation(&mut self) -> Result<()> {
let leader_statistic = self
.statistics
.get_mut(self.multi_sequencer_client.leader_url())
.ok_or_else(|| anyhow::anyhow!("Leader URL is not present in statistics"))?;
self.multi_sequencer_client
.update_statistics(leader_statistic)
.await;
.update_statistics(&mut self.statistics)
.await?;
self.multi_sequencer_client
.rotate(
&self.config.sequencers,
&mut self.statistics,
self.config.calibration_limit,
&self.config.multi_sequencer_client_config,
)
.await?;
@ -436,7 +447,7 @@ impl WalletCore {
let mut index = NullifierIndex::default();
index.track_initialization(account_id);
let poller = self.optimal_poller();
let poller = self.poller_helm();
let mut blocks = std::pin::pin!(poller.poll_block_range(1..=cursor));
while let Some(block) = blocks.try_next().await? {
for tx in block.body.transactions {
@ -544,7 +555,7 @@ impl WalletCore {
pub async fn get_account_balance(&self, acc: AccountId) -> Result<u128> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_account_balance(acc).await)
.metered_get(async |client: &SequencerClient| client.get_account_balance(acc).await)
.await?)
}
@ -552,7 +563,7 @@ impl WalletCore {
pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result<Vec<Nonce>> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| {
.metered_get(async |client: &SequencerClient| {
client.get_accounts_nonces(accs.to_vec()).await
})
.await?)
@ -574,14 +585,14 @@ impl WalletCore {
pub async fn get_last_block_id(&self) -> Result<u64> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_last_block_id().await)
.metered_get(async |client: &SequencerClient| client.get_last_block_id().await)
.await?)
}
pub async fn get_block(&self, block_id: u64) -> Result<Option<Block>> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_block(block_id).await)
.metered_get(async |client: &SequencerClient| client.get_block(block_id).await)
.await?)
}
@ -591,7 +602,7 @@ impl WalletCore {
) -> Result<Option<(LeeTransaction, BlockId)>> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_transaction(hash).await)
.metered_get(async |client: &SequencerClient| client.get_transaction(hash).await)
.await?)
}
@ -599,7 +610,7 @@ impl WalletCore {
pub async fn get_account_public(&self, account_id: AccountId) -> Result<Account> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_account(account_id).await)
.metered_get(async |client: &SequencerClient| client.get_account(account_id).await)
.await?)
}
@ -638,26 +649,23 @@ impl WalletCore {
pub async fn get_program_ids(&self) -> Result<BTreeMap<String, ProgramId>> {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| client.get_program_ids().await)
.metered_get(async |client: &SequencerClient| client.get_program_ids().await)
.await?)
}
/// 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_call(async |client: &SequencerClient| {
client.get_proofs_and_root(commitments.clone()).await
.metered_get(async |client: &SequencerClient| {
client.get_proofs_and_root(commitments.to_vec()).await
})
.await?)
}
@ -708,7 +716,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()?;
@ -722,7 +730,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 {
@ -800,12 +808,15 @@ impl WalletCore {
let call_res = self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| {
.metered_send(async |client: &SequencerClient| {
client
.send_transaction(LeeTransaction::PrivacyPreserving(tx.clone()))
.await
})
.await;
.await
.into_iter()
.find(std::result::Result::is_ok)
.ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)?;
Ok((call_res?, shared_secrets))
}
@ -868,12 +879,15 @@ impl WalletCore {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| {
.metered_send(async |client: &SequencerClient| {
client
.send_transaction(LeeTransaction::Public(tx.clone()))
.await
})
.await?)
.await
.into_iter()
.find(std::result::Result::is_ok)
.ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??)
}
pub async fn send_program_deployment_transaction(&self, bytecode: Vec<u8>) -> Result<HashType> {
@ -882,12 +896,15 @@ impl WalletCore {
Ok(self
.multi_sequencer_client
.metered_call(async |client: &SequencerClient| {
.metered_send(async |client: &SequencerClient| {
client
.send_transaction(LeeTransaction::ProgramDeployment(transaction.clone()))
.await
})
.await?)
.await
.into_iter()
.find(std::result::Result::is_ok)
.ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??)
}
pub async fn sync_to_latest_block(&mut self) -> Result<u64> {
@ -913,7 +930,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));

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

File diff suppressed because it is too large Load Diff

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

7
test_fixtures/src/config.rs
View File

@ -8,7 +8,7 @@ use lee::{AccountId, PrivateKey, PublicKey};
use lee_core::Identifier;
use sequencer_core::config::{BedrockConfig, CrossZoneConfig, GenesisAction, SequencerConfig};
use url::Url;
use wallet::config::{SequencerConnectionData, WalletConfig};
use wallet::config::{MultiSequencerClientConfig, SequencerConnectionData, WalletConfig};
pub const INITIAL_PUBLIC_BALANCES_FOR_WALLET: [u128; 2] = [10_000, 20_000];
pub const INITIAL_PRIVATE_BALANCES_FOR_WALLET: [u128; 2] = [10_000, 20_000];
@ -222,7 +222,10 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result<WalletConfig> {
seq_tx_poll_max_blocks: 15,
seq_poll_max_retries: 10,
seq_block_poll_max_amount: 100,
calibration_limit: 5,
multi_sequencer_client_config: MultiSequencerClientConfig {
distribution_limit: 1,
calibration_limit: 5,
},
})
}