[consensus] reduce retry timeout for submission and subscription (#24412)

## Description 

Besides the PR title, also refactor `ExponentialBackoff` to accept an
`initial_delay`, and improve logging a bit.

## Test plan 

CI
`SIM_STRESS_TEST_DURATION_SECS=300 python
./scripts/simtest/seed-search.py test_mainnet_config --test simtest
--num-seeds 1000`

Author: mwtian

consensus/core/src/subscriber.rs

@@ -118,13 +118,15 @@ impl<C: NetworkClient, S: NetworkService> Subscriber<C, S> {
         last_received: Round,
     ) {
         const IMMEDIATE_RETRIES: i64 = 3;
+        const MIN_TIMEOUT: Duration = Duration::from_millis(500);
         // When not immediately retrying, limit retry delay between 100ms and 10s.
-        const INITIAL_RETRY_INTERVAL: Duration = Duration::from_millis(100);
-        const MAX_RETRY_INTERVAL: Duration = Duration::from_secs(10);
-        const RETRY_INTERVAL_MULTIPLIER: f32 = 1.2;
+        let mut backoff = mysten_common::backoff::ExponentialBackoff::new(
+            Duration::from_millis(100),
+            Duration::from_secs(10),
+        );
+
         let peer_hostname = &context.committee.authority(peer).hostname;
         let mut retries: i64 = 0;
-        let mut delay = INITIAL_RETRY_INTERVAL;
         'subscription: loop {
             context
                 .metrics
@@ -133,29 +135,26 @@ impl<C: NetworkClient, S: NetworkService> Subscriber<C, S> {
                 .with_label_values(&[peer_hostname])
                 .set(0);
 
+            let mut delay = Duration::ZERO;
             if retries > IMMEDIATE_RETRIES {
+                delay = backoff.next().unwrap();
                 debug!(
                     "Delaying retry {} of peer {} subscription, in {} seconds",
                     retries,
                     peer_hostname,
                     delay.as_secs_f32(),
                 );
                 sleep(delay).await;
-                // Update delay for the next retry.
-                delay = delay
-                    .mul_f32(RETRY_INTERVAL_MULTIPLIER)
-                    .min(MAX_RETRY_INTERVAL);
             } else if retries > 0 {
                 // Retry immediately, but still yield to avoid monopolizing the thread.
                 tokio::task::yield_now().await;
-            } else {
-                // First attempt, reset delay for next retries but no waiting.
-                delay = INITIAL_RETRY_INTERVAL;
             }
             retries += 1;
 
+            // Use longer timeout when retry delay is long, to adapt to slow network.
+            let request_timeout = MIN_TIMEOUT.max(delay);
             let mut blocks = match network_client
-                .subscribe_blocks(peer, last_received, MAX_RETRY_INTERVAL)
+                .subscribe_blocks(peer, last_received, request_timeout)
                 .await
             {
                 Ok(blocks) => {

crates/mysten-common/src/backoff.rs

@@ -16,15 +16,14 @@ use rand::Rng as _;
 /// use mysten_common::backoff::ExponentialBackoff;
 ///
 /// // Basic example:
-/// let mut backoff = ExponentialBackoff::new(Duration::from_secs(10));
+/// let mut backoff = ExponentialBackoff::new(Duration::from_millis(100), Duration::from_secs(10));
 /// for (attempt, delay) in backoff.enumerate() {
 ///     println!("Attempt {attempt}: Delay: {:?}", delay);
 /// }
 ///
-/// // Specifying initial, maximum delay and jitter:
-/// let mut backoff = ExponentialBackoff::new(Duration::from_secs(60))
-///     .base(Duration::from_secs(5))
-///     .factor(1.2)
+/// // Specifying backoff factor and maximum jitter:
+/// let mut backoff = ExponentialBackoff::new(Duration::from_secs(5), Duration::from_secs(60))
+///     .factor(2.0)
 ///     .max_jitter(Duration::from_secs(1));
 /// loop {
 ///     // next() should always return a Some(Duration).
@@ -34,42 +33,35 @@ use rand::Rng as _;
 /// ```
 #[derive(Debug, Clone)]
 pub struct ExponentialBackoff {
-    current: Duration,
+    next: Duration,
     factor: f64,
     max_delay: Duration,
     max_jitter: Duration,
 }
 
 impl ExponentialBackoff {
     /// Constructs a new exponential backoff generator, specifying the maximum delay.
-    pub fn new(max_delay: Duration) -> ExponentialBackoff {
+    pub fn new(initial_delay: Duration, max_delay: Duration) -> ExponentialBackoff {
         ExponentialBackoff {
-            current: Duration::from_millis(50),
-            factor: 1.5,
+            next: initial_delay,
+            factor: 1.2,
             max_delay,
-            max_jitter: Duration::from_millis(50),
+            max_jitter: initial_delay,
         }
     }
 
-    /// Sets the base delay for computing the next delay, before increasing by the exponential factor and adding jitter.
-    ///
-    /// Default base delay is 50ms.
-    pub fn base(mut self, base: Duration) -> ExponentialBackoff {
-        self.current = base;
-        self
-    }
-
     /// Sets the approximate ratio of consecutive backoff delays, before jitters are applied.
     /// Setting this to Duration::ZERO disables jittering.
     ///
-    /// Default factor is 1.5.
+    /// Default is 1.2.
     pub fn factor(mut self, factor: f64) -> ExponentialBackoff {
         self.factor = factor;
         self
     }
 
     /// Sets the maximum jitter per delay.
-    /// Default maximum jitter is 50ms.
+    ///
+    /// Default is the initial delay.
     pub fn max_jitter(mut self, max_jitter: Duration) -> ExponentialBackoff {
         self.max_jitter = max_jitter;
         self
@@ -81,32 +73,31 @@ impl Iterator for ExponentialBackoff {
 
     /// Yields backoff delays. Never terminates.
     fn next(&mut self) -> Option<Duration> {
+        let current = self.next;
+
         let jitter = if self.max_jitter.is_zero() {
             Duration::ZERO
         } else {
             Duration::from_secs_f64(
                 rand::thread_rng().gen_range(0.0..self.max_jitter.as_secs_f64()),
             )
         };
-        let next = self
-            .current
+        self.next = current
             .mul_f64(self.factor)
             .min(self.max_delay)
             .saturating_add(jitter);
 
-        self.current = next;
-
-        Some(next)
+        Some(current)
     }
 }
 
 #[test]
 fn test_exponential_backoff_default() {
-    let mut backoff = ExponentialBackoff::new(Duration::from_secs(10));
+    let mut backoff = ExponentialBackoff::new(Duration::from_millis(50), Duration::from_secs(10));
 
     let bounds = vec![
-        (Duration::from_millis(75), Duration::from_millis(125)),
-        (Duration::from_millis(110), Duration::from_millis(250)),
+        (Duration::from_millis(50), Duration::from_millis(100)),
+        (Duration::from_millis(60), Duration::from_millis(170)),
     ];
     for ((lower, upper), delay) in bounds.into_iter().zip(backoff.next()) {
         assert!(delay >= lower && delay <= upper);
@@ -115,23 +106,23 @@ fn test_exponential_backoff_default() {
 
 #[test]
 fn test_exponential_backoff_base_100_factor_2_no_jitter() {
-    let mut backoff = ExponentialBackoff::new(Duration::from_secs(10))
-        .base(Duration::from_millis(100))
+    let mut backoff = ExponentialBackoff::new(Duration::from_millis(100), Duration::from_secs(10))
         .factor(2.0)
         .max_jitter(Duration::ZERO);
 
+    assert_eq!(backoff.next(), Some(Duration::from_millis(100)));
     assert_eq!(backoff.next(), Some(Duration::from_millis(200)));
     assert_eq!(backoff.next(), Some(Duration::from_millis(400)));
     assert_eq!(backoff.next(), Some(Duration::from_millis(800)));
 }
 
 #[test]
 fn test_exponential_backoff_max_delay() {
-    let mut backoff = ExponentialBackoff::new(Duration::from_secs(1))
-        .base(Duration::from_millis(200))
+    let mut backoff = ExponentialBackoff::new(Duration::from_millis(200), Duration::from_secs(1))
         .factor(3.0)
         .max_jitter(Duration::ZERO);
 
+    assert_eq!(backoff.next(), Some(Duration::from_millis(200)));
     assert_eq!(backoff.next(), Some(Duration::from_millis(600)));
 
     for _ in 0..10 {

crates/sui-core/src/consensus_adapter.rs

@@ -739,9 +739,11 @@ impl ConsensusAdapter {
         // In addition to that, within_alive_epoch ensures that all pending consensus
         // adapter tasks are stopped before reconfiguration can proceed.
         //
-        // This is essential because narwhal workers reuse same ports when narwhal restarts,
-        // this means we might be sending transactions from previous epochs to narwhal of
-        // new epoch if we have not had this barrier.
+        // This is essential because after epoch change, this validator may exit the committee and become a full node.
+        // So it is no longer able to submit to consensus.
+        //
+        // Also, submission to consensus is not gated on epoch. Although it is ok to submit user transactions
+        // to the new epoch, we want to cancel system transaction submissions from the current epoch to the new epoch.
         epoch_store
             .within_alive_epoch(self.submit_and_wait_inner(
                 transactions,
@@ -769,7 +771,7 @@ impl ConsensusAdapter {
                 "Performing a ping check, pinging consensus to get a consensus position in next block"
             );
             let (consensus_positions, _status_waiter) = self
-                .submit_inner(&transactions, epoch_store, &[], "ping", false)
+                .submit_inner(&transactions, epoch_store, &[], "ping")
                 .await;
 
             if let Some(tx_consensus_positions) = tx_consensus_positions.take() {
@@ -917,13 +919,7 @@ impl ConsensusAdapter {
                 loop {
                     // Submit the transaction to consensus and return the submit result with a status waiter
                     let (consensus_positions, status_waiter) = self
-                        .submit_inner(
-                            &transactions,
-                            epoch_store,
-                            &transaction_keys,
-                            tx_type,
-                            is_soft_bundle,
-                        )
+                        .submit_inner(&transactions, epoch_store, &transaction_keys, tx_type)
                         .await;
 
                     if let Some(tx_consensus_positions) = tx_consensus_positions.take() {
@@ -1050,11 +1046,13 @@ impl ConsensusAdapter {
         epoch_store: &Arc<AuthorityPerEpochStore>,
         transaction_keys: &[SequencedConsensusTransactionKey],
         tx_type: &str,
-        is_soft_bundle: bool,
     ) -> (Vec<ConsensusPosition>, BlockStatusReceiver) {
         let ack_start = Instant::now();
         let mut retries: u32 = 0;
-        let is_dkg = !transactions.is_empty() && transactions[0].is_dkg();
+        let mut backoff = mysten_common::backoff::ExponentialBackoff::new(
+            Duration::from_millis(100),
+            Duration::from_secs(10),
+        );
 
         let (consensus_positions, status_waiter) = loop {
             let span = debug_span!("client_submit");
@@ -1065,11 +1063,10 @@ impl ConsensusAdapter {
                 .await
             {
                 Err(err) => {
-                    // This can happen during reconfig, or when consensus has full internal buffers
-                    // and needs to back pressure, so retry a few times before logging warnings.
-                    if retries > 30 || (retries > 3 && (is_soft_bundle || !is_dkg)) {
+                    // This can happen during reconfig, so keep retrying until succeed.
+                    if cfg!(msim) || retries > 3 {
                         warn!(
-                            "Failed to submit transactions {transaction_keys:?} to consensus: {err:?}. Retry #{retries}"
+                            "Failed to submit transactions {transaction_keys:?} to consensus: {err}. Retry #{retries}"
                         );
                     }
                     self.metrics
@@ -1078,13 +1075,7 @@ impl ConsensusAdapter {
                         .inc();
                     retries += 1;
 
-                    if is_dkg {
-                        // Shorter delay for DKG messages, which are time-sensitive and happen at
-                        // start-of-epoch when submit errors due to active reconfig are likely.
-                        time::sleep(Duration::from_millis(100)).await;
-                    } else {
-                        time::sleep(Duration::from_secs(10)).await;
-                    };
+                    time::sleep(backoff.next().unwrap()).await;
                 }
                 Ok((consensus_positions, status_waiter)) => {
                     break (consensus_positions, status_waiter);
@@ -1462,7 +1453,7 @@ impl SubmitToConsensus for Arc<ConsensusAdapter> {
 
                 let result = tokio::time::timeout(
                     timeout,
-                    this.submit_inner(&[transaction], &epoch_store, &[key], tx_type, false),
+                    this.submit_inner(&[transaction], &epoch_store, &[key], tx_type),
                 )
                 .await;
 

crates/sui-core/src/transaction_driver/effects_certifier.rs

@@ -586,7 +586,8 @@ impl EffectsCertifier {
         A: AuthorityAPI + Send + Sync + 'static + Clone,
     {
         let effects_start = Instant::now();
-        let backoff = ExponentialBackoff::new(MAX_WAIT_FOR_EFFECTS_RETRY_DELAY);
+        let backoff =
+            ExponentialBackoff::new(Duration::from_millis(100), MAX_WAIT_FOR_EFFECTS_RETRY_DELAY);
         let ping_type = raw_request.get_ping_type();
         // This loop should only retry errors that are retriable without new submission.
         for (attempt, delay) in backoff.enumerate() {

crates/sui-core/src/transaction_driver/mod.rs

@@ -178,7 +178,10 @@ where
             .with_label_values(&[tx_type.as_str(), ping_label])
             .inc();
 
-        let mut backoff = ExponentialBackoff::new(MAX_DRIVE_TRANSACTION_RETRY_DELAY);
+        let mut backoff = ExponentialBackoff::new(
+            Duration::from_millis(100),
+            MAX_DRIVE_TRANSACTION_RETRY_DELAY,
+        );
         let mut attempts = 0;
         let mut latest_retriable_error = None;
 

crates/sui-surfer/src/surfer_state.rs

@@ -344,6 +344,8 @@ impl SurferState {
             rgp,
         );
         let tx = self.cluster.wallet.sign_transaction(&tx_data).await;
+        let tx_digest = *tx.digest();
+        info!(?tx_digest, "Publishing package");
         let start = Instant::now();
         let response = loop {
             match self
@@ -358,12 +360,12 @@ impl SurferState {
                 Err(err) => {
                     if start.elapsed() > Duration::from_secs(120) {
                         fatal!(
-                            "Failed to publish package after 120 seconds: {:?} {}",
+                            "Failed to publish package after 120 seconds: {} {}",
                             err,
                             tx.digest()
                         );
                     }
-                    error!("Failed to publish package: {:?} {}", err, tx.digest());
+                    error!(?tx_digest, "Failed to publish package: {}", err);
                     tokio::time::sleep(Duration::from_secs(1)).await;
                 }
             }