Merge pull request #511 from EspressoSystems/tw/submit-randomly

Submit to random client.

Author: twittner

timeboost-builder/Cargo.toml

@@ -17,6 +17,7 @@ metrics = { path = "../metrics" }
 minicbor = { workspace = true }
 multisig = { path = "../multisig" }
 parking_lot = { workspace = true }
+rand = { workspace = true }
 robusta = { path = "../robusta" }
 serde = { workspace = true }
 smallvec = { workspace = true }
@@ -28,5 +29,4 @@ tokio-util = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]
-rand = { workspace = true }
 tracing-subscriber = { workspace = true }

timeboost-builder/src/submit.rs

@@ -7,7 +7,8 @@ use std::{
 
 use bon::Builder;
 use multisig::{Committee, PublicKey, Validated};
-use robusta::{Client, espresso_types::NamespaceId};
+use rand::seq::IndexedRandom;
+use robusta::{Client, Config, espresso_types::NamespaceId};
 use timeboost_types::{
     CertifiedBlock,
     sailfish::{CommitteeVec, Empty},
@@ -49,27 +50,26 @@ impl Submitter {
     where
         M: ::metrics::Metrics,
     {
-        let client = Client::new(cfg.robusta.0.clone());
         let verified = Verified::default();
         let committees = Arc::new(Mutex::new(CommitteeVec::new(cfg.committee.clone())));
         let metrics = Arc::new(BuilderMetrics::new(metrics));
         let verifier = Verifier::builder()
             .label(cfg.pubkey)
             .nsid(cfg.namespace)
             .committees(committees.clone())
-            .client(client.clone())
+            .client(Client::new(cfg.robusta.0.clone()))
             .verified(verified.clone())
             .metrics(metrics.clone())
             .build();
         let (tx, rx) = mpsc::channel(10_000);
         let sender = Sender::builder()
             .label(cfg.pubkey)
             .nsid(cfg.namespace)
-            .client(client)
             .verified(verified.clone())
             .receiver(rx)
             .clock(Instant::now())
             .size_limit(cfg.max_transaction_size)
+            .config(cfg.robusta.0.clone())
             .build();
         let mut configs = vec![cfg.robusta.0.clone()];
         configs.extend(cfg.robusta.1.iter().cloned());
@@ -79,8 +79,8 @@ impl Submitter {
             committees,
             metrics,
             sender: tx,
-            verify_task: spawn(verifier.verify(configs)),
-            sender_task: spawn(sender.go()),
+            verify_task: spawn(verifier.verify(configs.clone())),
+            sender_task: spawn(sender.send(configs)),
         }
     }
 
@@ -109,17 +109,20 @@ pub struct SenderTaskDown(());
 struct Sender {
     label: PublicKey,
     nsid: NamespaceId,
-    client: Client,
     verified: Verified<15_000>,
     receiver: mpsc::Receiver<CertifiedBlock<Validated>>,
     clock: Instant,
     #[builder(default)]
     pending: BTreeMap<Instant, Vec<CertifiedBlock<Validated>>>,
     size_limit: usize,
+    config: Config,
 }
 
 impl Sender {
-    async fn go(mut self) {
+    async fn send(mut self, configs: Vec<Config>) {
+        let clients: Vec<Client> = configs.into_iter().map(Client::new).collect();
+        assert!(!clients.is_empty());
+
         // Blocks we receive from the application:
         let mut inbox = Vec::new();
         // Blocks scheduled for submission:
@@ -131,6 +134,12 @@ impl Sender {
             v.retain(|b| !self.verified.contains(b.cert().data().num()));
         };
 
+        let random_client = || {
+            clients
+                .choose(&mut rand::rng())
+                .expect("Vec<Client> is non-empty")
+        };
+
         let mut checkpoints = interval(Duration::from_secs(1));
         checkpoints.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
@@ -186,9 +195,9 @@ impl Sender {
 
                 debug!(node = %self.label, blocks = %transaction.len(), %size, "submitting blocks");
 
-                let mut delays = self.client.config().delay_iter();
+                let mut delays = self.config.delay_iter();
 
-                while let Err(err) = self.client.submit(self.nsid, &transaction).await {
+                while let Err(err) = random_client().submit(self.nsid, &transaction).await {
                     warn!(node= %self.label, %err, "error submitting blocks");
                     let d = delays.next().expect("delay iterator repeats");
                     sleep(d).await;