feat(wallet): distributed transaction sending

examples/program_deployment/src/bin/run_hello_world_with_authorization.rs

@@ -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(&[account_id])
+        .get_accounts_nonces(vec![account_id])
         .await
         .expect("Node should be reachable to query account data");
     let signing_keys = [&signing_key];

lez/wallet/src/cli/mod.rs

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

lez/wallet/src/config.rs

@@ -41,14 +41,14 @@ pub struct MultiSequencerClientConfig {
     /// Maximum numbers of sequencers to send requests
     pub distribution_limit: usize,
     /// Limit number of sequencer polls during callibration, should not be zero
-    pub callibration_limit: usize,
+    pub calibration_limit: usize,
 }
 
 impl Default for MultiSequencerClientConfig {
     fn default() -> Self {
         Self {
             distribution_limit: 1,
-            callibration_limit: 100,
+            calibration_limit: 100,
         }
     }
 }

lez/wallet/src/lib.rs

@@ -174,7 +174,7 @@ impl WalletCore {
         let mut statistics = extract_statistics_from_path(&statistics_path)?;
 
         let multi_sequencer_client = MultiSequencerClient::new(
-            &config.sequencers_conn_data,
+            &config.sequencers,
             &mut statistics,
             config.multi_sequencer_client_config.clone(),
         )
@@ -204,17 +204,21 @@ impl WalletCore {
 
     #[must_use]
     pub fn optimal_poller(&self) -> TxPoller {
-        TxPoller::new(self.config(), self.leader_owned())
+        TxPoller::new(self.config(), self.helm_owned())
     }
 
     #[must_use]
-    pub fn leader_owned(&self) -> SequencerClient {
+    pub fn helm_owned(&self) -> SequencerClient {
-        self.multi_sequencer_client.leader().clone()
+        self.multi_sequencer_client.helm().0.clone()
     }
 
     #[must_use]
-    pub fn leader_url(&self) -> Url {
+    pub fn helm_url(&self) -> Url {
-        self.multi_sequencer_client.leader_url().clone()
+        self.multi_sequencer_client.helm().1.clone()
+    }
+
+    pub fn leaders(&self) -> &[(SequencerClient, Url)] {
+        self.multi_sequencer_client.leaders()
     }
 
     /// Get storage.
@@ -255,20 +259,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)
+            .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?;
 
@@ -486,16 +485,16 @@ 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?)
     }
 
     /// Get accounts nonces.
-    pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result<Vec<Nonce>> {
+    pub async fn get_accounts_nonces(&self, accs: Vec<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
+                client.get_accounts_nonces(accs).await
             })
             .await?)
     }
@@ -516,14 +515,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?)
     }
 
@@ -533,7 +532,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?)
     }
 
@@ -541,7 +540,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?)
     }
 
@@ -580,7 +579,7 @@ 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?)
     }
 
@@ -598,8 +597,8 @@ impl WalletCore {
     ) -> Result<(Vec<Option<MembershipProof>>, CommitmentSetDigest)> {
         Ok(self
             .multi_sequencer_client
-            .metered_call(async |client: &SequencerClient| {
+            .metered_get(async |client: &SequencerClient| {
-                client.get_proofs_and_root(commitments.clone()).await
+                client.get_proofs_and_root(commitments).await
             })
             .await?)
     }
@@ -742,7 +741,7 @@ 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
@@ -810,7 +809,7 @@ 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
@@ -824,7 +823,7 @@ 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

lez/wallet/src/multi_client.rs

@@ -76,21 +76,21 @@ impl Statistics {
 
 #[derive(Clone)]
 pub struct MultiSequencerClient {
-    /// Ordered list of leaders, from best to worst
+    /// Ordered list of leaders, from best to worst.
     pub leader_list: Vec<(SequencerClient, Url)>,
     pub multi_sequencer_client_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.
-    statistic_updates: Arc<RwLock<Vec<StatisticsUpdate>>>,
+    statistic_updates: Arc<RwLock<HashMap<Url, Vec<StatisticsUpdate>>>>,
 }
 
 impl MultiSequencerClient {
     async fn setup(
         conn_data: &[SequencerConnectionData],
         statistics: &mut HashMap<Url, Statistics>,
-        multi_sequencer_client_config: MultiSequencerClientConfig,
+        multi_sequencer_client_config: &MultiSequencerClientConfig,
-    ) -> Result<Self> {
+    ) -> Result<Vec<(SequencerClient, Url)>> {
         let mut client_list = HashMap::new();
 
         for SequencerConnectionData {
@@ -117,55 +117,52 @@ impl MultiSequencerClient {
             };
 
             // If there is statistics for client, actualize it
-            if let Some(metric_mut) = statistics.get_mut(sequencer_addr) {
+            if let Some(statistic_mut) = statistics.get_mut(sequencer_addr) {
-                let metric_updates = actualize_client(&sequencer_client).await;
+                let statistic_updates = actualize_client(&sequencer_client).await;
 
                 log::debug!(
-                    "Metered call for {sequencer_addr:?}, metric updates is {metric_updates:?}"
+                    "Metered call for {sequencer_addr:?}, statistic updates is {statistic_updates:?}"
                 );
 
-                metric_mut.apply_updates(&[metric_updates]);
+                statistic_mut.apply_updates(&[statistic_updates]);
+
+                client_list.insert(sequencer_addr.clone(), sequencer_client);
             // Otherwise calibrate client data
             } else if let Some(client_statistics) =
-                calibrate_client(&sequencer_client, calibration_limit).await
+                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);
+            // There is no point in adding uncalibrated client
             } else {
-                log::warn!("Client {sequencer_addr:?} failed all {calibration_limit} calibration attempts, it may be unhealthy.
+                log::warn!("Client {sequencer_addr:?} failed all {} calibration attempts, it may be unhealthy.
-                    \n Consider bumping calibration_limit or remove this client altogether");
+                    \n Consider bumping calibration_limit or remove this client altogether", multi_sequencer_client_config.calibration_limit);
             }
-
-            client_list.insert(sequencer_addr.clone(), sequencer_client);
         }
 
-        // Dropping client list, for reasons why, see comment in structure definition.
-
         let leader_list = choose_leaders(
             &client_list,
-            metrics,
+            statistics,
             multi_sequencer_client_config.distribution_limit,
         )
         .ok_or_else(|| anyhow::anyhow!("Failed to find leader"))?;
 
         log::info!("Chosen leaders is {leader_list:?}");
 
-        Ok(Self {
+        Ok(leader_list)
-            leader_list,
-            multi_sequencer_client_config,
-        })
     }
 
     pub async fn new(
         conn_data: &[SequencerConnectionData],
         statistics: &mut HashMap<Url, Statistics>,
-        calibration_limit: usize,
+        multi_sequencer_client_config: MultiSequencerClientConfig,
     ) -> Result<Self> {
-        let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?;
+        let leader_list = Self::setup(conn_data, statistics, &multi_sequencer_client_config).await?;
 
         Ok(Self {
-            leader,
+            leader_list,
-            leader_url,
+            multi_sequencer_client_config,
-            statistic_updates: Arc::new(RwLock::new(vec![])),
+            statistic_updates: Arc::new(RwLock::new(HashMap::new())),
         })
     }
 
@@ -174,54 +171,95 @@ impl MultiSequencerClient {
         &mut self,
         conn_data: &[SequencerConnectionData],
         statistics: &mut HashMap<Url, Statistics>,
-        calibration_limit: usize,
+        multi_sequencer_client_config: &MultiSequencerClientConfig,
     ) -> Result<()> {
-        let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?;
+        let leader_list = Self::setup(conn_data, statistics, multi_sequencer_client_config).await?;
 
-        log::info!("Chosen leader is {leader_url:?}");
+        log::info!("Chosen leaders is {leader_list:#?}");
 
-        self.leader = leader;
+        self.leader_list = leader_list;
-        self.leader_url = leader_url;
 
         Ok(())
     }
 
     #[must_use]
-    pub const fn leader(&self) -> &SequencerClient {
+    pub fn leaders(&self) -> &[(SequencerClient, Url)] {
-        &self.leader
+        &self.leader_list.as_ref()
     }
 
     #[must_use]
-    pub const fn leader_url(&self) -> &Url {
+    pub fn helm(&self) -> &(SequencerClient, Url) {
-        &self.leader_url
+        &self.leader_list.first().expect("At least one leader must be set")
     }
 
-    // Keeping this call abstract, in case if we need to do more than one request
+    /// Metered call for main leader(helm), to get data, necessary for send call.
-    pub async fn metered_call<R, E, I: AsyncFn(&SequencerClient) -> Result<R, E>>(
+    pub async fn metered_get<R, E, I: AsyncFnOnce(&SequencerClient) -> Result<R, E>>(
         &self,
         call: I,
     ) -> Result<R, E> {
+        let (helm, helm_url) = self.helm();
+
         let (resp, statistics_update) =
-            tokio::join!(call(self.leader()), actualize_client(self.leader()));
+            tokio::join!(call(helm), actualize_client(helm));
 
         log::debug!(
-            "Metered call for {:?}, metric updates is {:?}",
+            "Metered call for {:?}, statistic updates is {:?}",
-            self.leader_url,
+            helm_url,
             statistics_update
         );
 
         {
             let mut statistic_updates_guard = self.statistic_updates.write().await;
-            statistic_updates_guard.push(statistics_update);
+            statistic_updates_guard.entry(helm_url.clone()).or_default().push(statistics_update);
         }
 
         resp
     }
 
-    /// Update statistics of a leader, clear statist updates log.
+    /// Metered call for `distribution_limit` amount of leaders for sending data, usually transaction.
-    pub async fn update_statistics(&self, leader_statistic: &mut Statistics) {
+    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);
+
+        // 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));
+
+            log::debug!(
+            "Metered call for {:?}, statistic updates is {:?}",
+            leader_url,
+            statistics_update
+            );
+
+            statistic_map.entry(leader_url.clone()).or_default().push(statistics_update);
+            results.push(resp);
+        }
+
+        {
+            let mut statistic_updates_guard = self.statistic_updates.write().await;
+
+            statistic_updates_guard.extend(statistic_map.into_iter());
+        }
+
+        results
+    }
+
+    /// Update statistics of a leader, clear statistic updates log.
+    pub async fn update_statistics(&self, statistics: &mut HashMap<Url, Statistics>,) {
         let mut statistic_updates = self.statistic_updates.write().await;
-        leader_statistic.apply_updates(statistic_updates.as_ref());
+
+        for (addr, statistic_updates_vec) in statistic_updates.iter() {
+            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();
     }
 }
@@ -838,9 +876,11 @@ mod tests {
             },
         );
 
-        let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap();
+        let leaders = choose_leaders(&client_list, &statistics, 1).unwrap();
 
-        assert_eq!(leader_url, addr_leader);
+        let (_, leader_url) = leaders.first().unwrap(); 
+
+        assert_eq!(leader_url, &addr_leader);
     }
 
     #[test]
@@ -908,9 +948,11 @@ mod tests {
             },
         );
 
-        let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap();
+        let leaders = choose_leaders(&client_list, &statistics, 1).unwrap();
 
-        assert_eq!(leader_url, addr_leader);
+        let (_, leader_url) = leaders.first().unwrap(); 
+
+        assert_eq!(leader_url, &addr_leader);
     }
 
     #[test]
@@ -978,9 +1020,11 @@ mod tests {
             },
         );
 
-        let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap();
+        let leaders = choose_leaders(&client_list, &statistics, 1).unwrap();
 
-        assert_eq!(leader_url, addr_leader);
+        let (_, leader_url) = leaders.first().unwrap(); 
+
+        assert_eq!(leader_url, &addr_leader);
     }
 
     #[test]
@@ -1048,8 +1092,10 @@ mod tests {
             },
         );
 
-        let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap();
+        let leaders = choose_leaders(&client_list, &statistics, 1).unwrap();
 
-        assert_eq!(leader_url, addr_leader);
+        let (_, leader_url) = leaders.first().unwrap(); 
+
+        assert_eq!(leader_url, &addr_leader);
     }
 }