Remove with_retry for port collisions.

Author: akonring

Cargo.lock

@@ -11,7 +11,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 
 use alloy::eips::{BlockNumberOrTag, Encodable2718};
 use bytes::Bytes;
-use cliquenet::{Address, AddressableCommittee, Network, NetworkMetrics};
+use cliquenet::{Address, AddressableCommittee};
 use metrics::NoMetrics;
 use multisig::Keypair;
 use multisig::{Committee, x25519};
@@ -23,7 +23,6 @@ use timeboost::crypto::prelude::DkgDecKey;
 use timeboost::sequencer::{Sequencer, SequencerConfig};
 use timeboost::types::{BlockNumber, BundleVariant, DecryptionKeyCell, KeyStore, Transaction};
 use timeboost_utils::load_generation::make_bundle;
-use timeboost_utils::with_retry;
 use tokio::sync::broadcast;
 use tokio::time::{Duration, sleep};
 use tracing::warn;
@@ -194,41 +193,3 @@ impl Round2Block {
             .or_insert_with(|| self.counter.fetch_add(1, Ordering::Relaxed).into())
     }
 }
-
-async fn start_sequencer_with_retry(seq_conf: SequencerConfig) -> Sequencer {
-    with_retry(
-        || async { Sequencer::new(seq_conf.clone(), &NoMetrics).await },
-        |e| format!("failed to create sequencer: {e}"),
-    )
-    .await
-}
-
-async fn start_certifier_with_retry(cert_conf: CertifierConfig) -> Certifier {
-    with_retry(
-        || async { Certifier::new(cert_conf.clone(), &NoMetrics).await },
-        |e| format!("failed to create certifer: {e}"),
-    )
-    .await
-}
-
-async fn start_network_with_retry(cfg: &SequencerConfig) -> Network {
-    with_retry(
-        || async {
-            Network::create(
-                "sailfish",
-                cfg.sailfish_address().clone(),
-                cfg.sign_keypair().public_key(),
-                cfg.dh_keypair().clone(),
-                cfg.sailfish_committee().entries(),
-                NetworkMetrics::new(
-                    "sailfish",
-                    &NoMetrics,
-                    cfg.sailfish_committee().parties().copied(),
-                ),
-            )
-            .await
-        },
-        |e| format!("Network::create failed: {e}"),
-    )
-    .await
-}

tests/src/tests/timeboost.rs

@@ -4,8 +4,10 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::Duration;
 
+use metrics::NoMetrics;
 use multisig::Certificate;
-use timeboost::sequencer::Output;
+use timeboost::builder::Certifier;
+use timeboost::sequencer::{Output, Sequencer};
 use timeboost::types::{Block, BlockInfo};
 use timeboost_utils::types::logging::init_logging;
 use tokio::select;
@@ -16,9 +18,7 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tracing::{debug, error, info};
 
-use crate::tests::timeboost::{
-    Round2Block, hash, start_certifier_with_retry, start_sequencer_with_retry,
-};
+use crate::tests::timeboost::{Round2Block, hash};
 
 use super::{gen_bundles, make_configs};
 
@@ -50,9 +50,14 @@ async fn block_order() {
                 // delay start of a recovering node:
                 sleep(Duration::from_secs(5)).await
             }
-            let mut s = start_sequencer_with_retry(c).await;
-            let mut p = start_certifier_with_retry(b).await;
+            let mut s = Sequencer::new(c.clone(), &NoMetrics)
+                .await
+                .unwrap_or_else(|e| panic!("failed to create sequencer: {e}"));
+            let mut p = Certifier::new(b.clone(), &NoMetrics)
+                .await
+                .unwrap_or_else(|e| panic!("failed to create certifer: {e}"));
             let mut r = None;
+
             let handle = p.handle();
             loop {
                 select! {

tests/src/tests/timeboost/block_order.rs

@@ -5,15 +5,16 @@ use std::num::NonZeroUsize;
 use std::time::Duration;
 
 use alloy::eips::BlockNumberOrTag;
-use cliquenet::{Address, AddressableCommittee, Overlay};
+use cliquenet::{Address, AddressableCommittee, Network, NetworkMetrics, Overlay};
 use futures::FutureExt;
 use futures::stream::{self, StreamExt};
+use metrics::NoMetrics;
 use multisig::{Committee, CommitteeId, Keypair, x25519};
 use sailfish::consensus::Consensus;
 use sailfish::rbc::Rbc;
 use sailfish::types::{ConsensusTime, RoundNumber, Timestamp};
 use sailfish::{Coordinator, Event};
-use timeboost::config::{ChainConfig, ParentChain};
+use timeboost::config::{ChainConfig, DECRYPTER_PORT_OFFSET, ParentChain};
 use timeboost::crypto::prelude::DkgDecKey;
 use timeboost::sequencer::SequencerConfig;
 use timeboost::types::{DecryptionKeyCell, KeyStore};
@@ -26,8 +27,6 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::info;
 use url::Url;
 
-use super::start_network_with_retry;
-
 #[derive(Debug, Clone)]
 enum Cmd {
     NextCommittee(ConsensusTime, AddressableCommittee),
@@ -80,7 +79,7 @@ where
 
     let de_addrs = sf_addrs
         .iter()
-        .map(|addr| Address::from((Ipv4Addr::LOCALHOST, addr.port() + 1000)))
+        .map(|addr| Address::from((Ipv4Addr::LOCALHOST, addr.port() + DECRYPTER_PORT_OFFSET)))
         .collect::<Vec<_>>();
 
     let committee = Committee::new(
@@ -173,7 +172,20 @@ where
 ///
 /// NB that the decryption parts of the config are not used yet.
 async fn mk_node(cfg: &SequencerConfig) -> Coordinator<Timestamp, Rbc<Timestamp>> {
-    let mut net = start_network_with_retry(cfg).await;
+    let mut net = Network::create(
+        "sailfish",
+        cfg.sailfish_address().clone(),
+        cfg.sign_keypair().public_key(),
+        cfg.dh_keypair().clone(),
+        cfg.sailfish_committee().entries(),
+        NetworkMetrics::new(
+            "sailfish",
+            &NoMetrics,
+            cfg.sailfish_committee().parties().copied(),
+        ),
+    )
+    .await
+    .unwrap();
 
     if let Some(prev) = &cfg.previous_sailfish_committee() {
         let old = prev.diff(cfg.sailfish_committee());

tests/src/tests/timeboost/handover.rs

@@ -71,8 +71,8 @@ async fn run_handover(
         let r2b = round2block.clone();
 
         tasks.spawn(async move {
-            let mut s = start_sequencer_with_retry(seq_conf).await;
-            let mut c = start_certifier_with_retry(cert_conf).await;
+            let mut s = Sequencer::new(seq_conf.clone(), &NoMetrics).await.unwrap();
+            let mut c = Certifier::new(cert_conf.clone(), &NoMetrics).await.unwrap();
             let mut r: Option<RoundNumber> = None;
             let c_handle = c.handle();
 
@@ -179,8 +179,12 @@ async fn run_handover(
         let r2b = round2block.clone();
 
         tasks.spawn(async move {
-            let mut s = start_sequencer_with_retry(seq_conf).await;
-            let mut c = start_certifier_with_retry(cert_conf).await;
+          let mut s = Sequencer::new(seq_conf, &NoMetrics)
+                .await
+                .unwrap_or_else(|e| panic!("failed to create sequencer: {e}"));
+            let mut c = Certifier::new(cert_conf, &NoMetrics)
+                .await
+                .unwrap_or_else(|e| panic!("failed to create certifer: {e}"));
             let mut r: Option<sailfish_types::RoundNumber> = None;
             let c_handle = c.handle();
 

tests/src/tests/timeboost/timeboost_handover.rs

@@ -4,8 +4,9 @@ use std::num::NonZeroUsize;
 use std::time::Duration;
 
 use alloy::primitives::B256;
+use metrics::NoMetrics;
 use sailfish_types::RoundNumber;
-use timeboost::sequencer::Output;
+use timeboost::sequencer::{Output, Sequencer};
 use timeboost_utils::types::logging::init_logging;
 use tokio::select;
 use tokio::sync::broadcast::error::RecvError;
@@ -15,8 +16,6 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tracing::{debug, info};
 
-use crate::tests::timeboost::start_sequencer_with_retry;
-
 use super::{gen_bundles, make_configs};
 
 const NUM_OF_TRANSACTIONS: usize = 500;
@@ -53,7 +52,9 @@ async fn transaction_order() {
                 // delay start of a recovering node:
                 sleep(Duration::from_secs(5)).await
             }
-            let mut s = start_sequencer_with_retry(c).await;
+            let mut s = Sequencer::new(c.clone(), &NoMetrics)
+                .await
+                .unwrap_or_else(|e| panic!("failed to create sequencer: {e}"));
             loop {
                 select! {
                     trx = brx.recv() => match trx {

tests/src/tests/timeboost/transaction_order.rs

@@ -23,7 +23,6 @@ smallvec = { workspace = true }
 thiserror = { workspace = true }
 timeboost-config = { path = "../timeboost-config" }
 timeboost-types = { path = "../timeboost-types" }
-timeboost-utils = { path = "../timeboost-utils" }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 tracing = { workspace = true }

timeboost-builder/Cargo.toml

@@ -1479,8 +1479,7 @@ mod tests {
         sync::Arc,
         time::Instant,
     };
-
-    use timeboost_utils::{types::logging, with_retry};
+    use timeboost_utils::types::logging;
 
     use cliquenet::AddressableCommittee;
     use multisig::{Committee, KeyId, Keypair, SecretKey, Signed, VoteAccumulator, x25519};
@@ -2445,7 +2444,13 @@ mod tests {
                 .threshold_dec_key(encryption_key_cell.clone())
                 .build();
 
-            let decrypter = start_decrypter_with_retry(decrypter_config).await;
+            let decrypter = Decrypter::new(
+                decrypter_config.clone(),
+                &NoMetrics,
+                Arc::new(SequencerMetrics::default()),
+            )
+            .await
+            .unwrap_or_else(|e| panic!("failed to create decrypter: {e}"));
             decrypters.push(decrypter);
             encryption_key_cells.push(encryption_key_cell);
         }
@@ -2460,19 +2465,4 @@ mod tests {
             ),
         )
     }
-
-    async fn start_decrypter_with_retry(dec_config: DecrypterConfig) -> Decrypter {
-        with_retry(
-            || async {
-                Decrypter::new(
-                    dec_config.clone(),
-                    &NoMetrics,
-                    Arc::new(SequencerMetrics::default()),
-                )
-                .await
-            },
-            |e| format!("failed to create decrypter: {e}"),
-        )
-        .await
-    }
 }

timeboost-sequencer/src/decrypt.rs

@@ -2,12 +2,12 @@ pub mod load_generation;
 pub mod types;
 pub mod until;
 
-use std::{fmt::Display, time::Duration};
+use std::time::Duration;
 
 use cliquenet::Address;
 use multisig::x25519;
 use tokio::time::sleep;
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
 pub fn unsafe_zero_keypair<N: Into<u64>>(i: N) -> multisig::Keypair {
     sig_keypair_from_seed_indexed([0u8; 32], i.into())
@@ -68,25 +68,3 @@ pub async fn wait_for_live_peer(host: &Address) -> anyhow::Result<()> {
         sleep(Duration::from_secs(3)).await;
     }
 }
-
-/// Retries an async operation with exponential backoff until it succeeds.
-pub async fn with_retry<T, E, F, Fut>(op: F, err: impl Fn(&E) -> String) -> T
-where
-    E: Display,
-    F: Fn() -> Fut,
-    Fut: Future<Output = Result<T, E>>,
-{
-    let mut delay = Duration::from_millis(100);
-    const MAX_DELAY: Duration = Duration::from_secs(5);
-
-    loop {
-        match op().await {
-            Ok(result) => return result,
-            Err(e) => {
-                warn!("{}: {e}, retrying in {:?}...", err(&e), delay);
-                sleep(delay).await;
-                delay = (delay * 2).min(MAX_DELAY);
-            }
-        }
-    }
-}