Speed up SDK read polling (#24434)

## Description 

Waiting for 2s after the 1st failure to poll transaction can be too
long.

## Test plan 

CI
`python ./scripts/simtest/seed-search.py per_epoch_config_stress_test
--test per_epoch_config_stress_tests --num-seeds 10`

Author: mwtian

Cargo.lock

@@ -16,6 +16,7 @@ use sui_types::programmable_transaction_builder::ProgrammableTransactionBuilder;
 use sui_types::transaction::{CallArg, ObjectArg, SharedObjectMutability, TransactionData};
 use sui_types::{SUI_DENY_LIST_OBJECT_ID, SUI_FRAMEWORK_PACKAGE_ID};
 use test_cluster::{TestCluster, TestClusterBuilder};
+use tracing::info;
 
 const DENY_ADDRESS: SuiAddress = SuiAddress::ZERO;
 
@@ -34,13 +35,13 @@ async fn per_epoch_config_stress_test() {
     let handle1 = {
         let test_env = test_env.clone();
         tokio::spawn(async move {
-            run_thread(test_env, target_epoch, gas1.0, create_transfer_tx, true).await
+            run_thread(1, test_env, target_epoch, gas1.0, create_transfer_tx, true).await
         })
     };
     let handle2 = {
         let test_env = test_env.clone();
         tokio::spawn(async move {
-            run_thread(test_env, target_epoch, gas2.0, create_deny_tx, false).await
+            run_thread(2, test_env, target_epoch, gas2.0, create_deny_tx, false).await
         })
     };
     tokio::time::timeout(Duration::from_secs(600), async {
@@ -52,6 +53,7 @@ async fn per_epoch_config_stress_test() {
 }
 
 async fn run_thread<F, Fut>(
+    thread_id: u64,
     test_env: Arc<TestEnv>,
     target_epoch: EpochId,
     gas_id: ObjectID,
@@ -61,12 +63,15 @@ async fn run_thread<F, Fut>(
     F: Fn(Arc<TestEnv>, ObjectRef) -> Fut,
     Fut: Future<Output = TransactionData>,
 {
+    info!(?thread_id, "Thread started");
     let mut num_tx_succeeded = 0;
     let mut num_tx_failed = 0;
     loop {
         let gas = test_env.get_latest_object_ref(&gas_id).await;
         let tx_data = tx_creation_func(test_env.clone(), gas).await;
         let tx = test_env.test_cluster.sign_transaction(&tx_data).await;
+        let tx_digest = *tx.digest();
+        info!(?thread_id, ?tx_digest, "Sending transaction");
         let Ok(effects) = test_env
             .test_cluster
             .wallet
@@ -79,19 +84,32 @@ async fn run_thread<F, Fut>(
             continue;
         };
         if effects.status().is_ok() {
+            info!(?thread_id, ?tx_digest, "Transaction succeeded");
             num_tx_succeeded += 1;
         } else {
+            info!(?thread_id, ?tx_digest, "Transaction failed");
             num_tx_failed += 1;
         }
         let executed_epoch = effects.executed_epoch();
         if executed_epoch >= target_epoch {
+            info!(
+                ?thread_id,
+                "Reached target epoch {target_epoch}. Current {executed_epoch}."
+            );
             break;
         }
     }
     if !tx_may_fail {
         assert_eq!(num_tx_failed, 0);
     }
-    assert!(num_tx_succeeded + num_tx_failed > 5);
+    assert!(
+        num_tx_succeeded + num_tx_failed > 5,
+        "Thread {thread_id} succeeded {num_tx_succeeded} transactions and failed {num_tx_failed} transactions"
+    );
+    info!(
+        ?thread_id,
+        "Thread {thread_id} finished. Succeeded {num_tx_succeeded} transactions and failed {num_tx_failed} transactions."
+    );
 }
 
 async fn create_deny_tx(test_env: Arc<TestEnv>, gas: ObjectRef) -> TransactionData {

crates/sui-e2e-tests/tests/per_epoch_config_stress_tests.rs

@@ -31,6 +31,7 @@ bcs.workspace = true
 thiserror.workspace = true
 reqwest.workspace = true
 
+mysten-common.workspace = true
 sui-json-rpc-api.workspace = true
 sui-transaction-builder.workspace = true
 sui-json-rpc-types.workspace = true

crates/sui-sdk/Cargo.toml

@@ -42,9 +42,8 @@ use sui_types::sui_serde::BigInt;
 use sui_types::sui_system_state::sui_system_state_summary::SuiSystemStateSummary;
 use sui_types::transaction::{Transaction, TransactionData, TransactionKind};
 
-const WAIT_FOR_LOCAL_EXECUTION_DELAY: Duration = Duration::from_millis(200);
-
-const WAIT_FOR_LOCAL_EXECUTION_INTERVAL: Duration = Duration::from_secs(2);
+const WAIT_FOR_LOCAL_EXECUTION_MIN_INTERVAL: Duration = Duration::from_millis(100);
+const WAIT_FOR_LOCAL_EXECUTION_MAX_INTERVAL: Duration = Duration::from_secs(2);
 
 /// The main read API structure with functions for retrieving data about different objects and transactions
 #[derive(Debug)]
@@ -1180,12 +1179,15 @@ impl QuorumDriverApi {
             Duration::from_secs(60)
         };
         let mut poll_response = tokio::time::timeout(wait_for_local_execution_timeout, async {
-            // Apply a short delay to give the full node a chance to catch up.
-            tokio::time::sleep(WAIT_FOR_LOCAL_EXECUTION_DELAY).await;
-
-            let mut interval = tokio::time::interval(WAIT_FOR_LOCAL_EXECUTION_INTERVAL);
+            let mut backoff = mysten_common::backoff::ExponentialBackoff::new(
+                WAIT_FOR_LOCAL_EXECUTION_MIN_INTERVAL,
+                WAIT_FOR_LOCAL_EXECUTION_MAX_INTERVAL,
+            );
             loop {
-                interval.tick().await;
+                // Intentionally waiting for a short delay (MIN_INTERVAL) before the 1st iteration,
+                // to leave time for the checkpoint containing the transaction to be certified, propagate
+                // to the full node, and get executed.
+                tokio::time::sleep(backoff.next().unwrap()).await;
 
                 if let Ok(poll_response) = self
                     .api