[to] recover transactions through TransactionDriver (#24385)

## Description 

Recover transactions through Transaction Driver instead of Quorum
Driver, and clear recovered transactions explicitly.

Also remove more usages of Quorum Driver via Transaction Orchestrator.

## Test plan 

CI

Author: mwtian

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

@@ -87,6 +87,7 @@ impl<A> TransactionDriver<A>
 where
     A: AuthorityAPI + Send + Sync + 'static + Clone,
 {
+    // TODO: accept a TransactionDriverConfig to set default allowed & blocked validators.
     pub fn new(
         authority_aggregator: Arc<AuthorityAggregator<A>>,
         reconfig_observer: Arc<dyn ReconfigObserver<A> + Sync + Send>,
@@ -120,109 +121,17 @@ where
         driver
     }
 
-    // Runs a background task to send ping transactions to all validators to perform latency checks to test both the fast path and the consensus path.
-    async fn run_latency_checks(self: Arc<Self>) {
-        const INTERVAL_BETWEEN_RUNS: Duration = Duration::from_secs(15);
-        const MAX_JITTER: Duration = Duration::from_secs(10);
-        const PING_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
-
-        let mut interval = interval(INTERVAL_BETWEEN_RUNS);
-        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-
-        loop {
-            interval.tick().await;
-
-            let mut tasks = JoinSet::new();
-
-            for tx_type in [TxType::SingleWriter, TxType::SharedObject] {
-                Self::ping_for_tx_type(
-                    self.clone(),
-                    &mut tasks,
-                    tx_type,
-                    MAX_JITTER,
-                    PING_REQUEST_TIMEOUT,
-                );
-            }
-
-            while let Some(result) = tasks.join_next().await {
-                if let Err(e) = result {
-                    tracing::debug!("Error while driving ping transaction: {}", e);
-                }
-            }
-        }
+    /// Returns the authority aggregator wrapper which upgrades on epoch changes.
+    pub fn authority_aggregator(&self) -> &Arc<ArcSwap<AuthorityAggregator<A>>> {
+        &self.authority_aggregator
     }
 
-    /// Pings all validators for e2e latency with the provided transaction type.
-    fn ping_for_tx_type(
-        self: Arc<Self>,
-        tasks: &mut JoinSet<()>,
-        tx_type: TxType,
-        max_jitter: Duration,
-        ping_timeout: Duration,
-    ) {
-        // We are iterating over the single writer and shared object transaction types to test both the fast path and the consensus path.
-        let auth_agg = self.authority_aggregator.load().clone();
-        let validators = auth_agg.committee.names().cloned().collect::<Vec<_>>();
-
-        self.metrics
-            .latency_check_runs
-            .with_label_values(&[tx_type.as_str()])
-            .inc();
-
-        for name in validators {
-            let display_name = auth_agg.get_display_name(&name);
-            let delay_ms = rand::thread_rng().gen_range(0..max_jitter.as_millis()) as u64;
-            let self_clone = self.clone();
-
-            let task = async move {
-                // Add some random delay to the task to avoid all tasks running at the same time
-                if delay_ms > 0 {
-                    tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
-                }
-                let start_time = Instant::now();
-
-                let ping_type = if tx_type == TxType::SingleWriter {
-                    PingType::FastPath
-                } else {
-                    PingType::Consensus
-                };
-
-                // Now send a ping transaction to the chosen validator for the provided tx type
-                match self_clone
-                    .drive_transaction(
-                        SubmitTxRequest::new_ping(ping_type),
-                        SubmitTransactionOptions {
-                            allowed_validators: vec![display_name.clone()],
-                            ..Default::default()
-                        },
-                        Some(ping_timeout),
-                    )
-                    .await
-                {
-                    Ok(_) => {
-                        tracing::debug!(
-                            "Ping transaction to validator {} for tx type {} completed end to end in {} seconds",
-                            display_name,
-                            tx_type.as_str(),
-                            start_time.elapsed().as_secs_f64()
-                        );
-                    }
-                    Err(err) => {
-                        tracing::debug!(
-                            "Failed to get certified finalized effects for tx type {}, for ping transaction to validator {}: {}",
-                            tx_type.as_str(),
-                            display_name,
-                            err
-                        );
-                    }
-                }
-            };
-
-            tasks.spawn(task);
-        }
-    }
-
-    /// Drives transaction to submission and effects certification. Ping requests are derived from the submitted payload.
+    /// Drives transaction to finalization.
+    ///
+    /// Internally, retries the attempt to finalize a transaction until:
+    /// - The transaction is finalized.
+    /// - The transaction observes a non-retriable error.
+    /// - Timeout is reached.
     #[instrument(level = "error", skip_all, fields(tx_digest = ?request.transaction.as_ref().map(|t| t.digest()), ping = %request.ping_type.is_some()))]
     pub async fn drive_transaction(
         &self,
@@ -438,6 +347,108 @@ where
         result
     }
 
+    // Runs a background task to send ping transactions to all validators to perform latency checks to test both the fast path and the consensus path.
+    async fn run_latency_checks(self: Arc<Self>) {
+        const INTERVAL_BETWEEN_RUNS: Duration = Duration::from_secs(15);
+        const MAX_JITTER: Duration = Duration::from_secs(10);
+        const PING_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+        let mut interval = interval(INTERVAL_BETWEEN_RUNS);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+        loop {
+            interval.tick().await;
+
+            let mut tasks = JoinSet::new();
+
+            for tx_type in [TxType::SingleWriter, TxType::SharedObject] {
+                Self::ping_for_tx_type(
+                    self.clone(),
+                    &mut tasks,
+                    tx_type,
+                    MAX_JITTER,
+                    PING_REQUEST_TIMEOUT,
+                );
+            }
+
+            while let Some(result) = tasks.join_next().await {
+                if let Err(e) = result {
+                    tracing::debug!("Error while driving ping transaction: {}", e);
+                }
+            }
+        }
+    }
+
+    /// Pings all validators for e2e latency with the provided transaction type.
+    fn ping_for_tx_type(
+        self: Arc<Self>,
+        tasks: &mut JoinSet<()>,
+        tx_type: TxType,
+        max_jitter: Duration,
+        ping_timeout: Duration,
+    ) {
+        // We are iterating over the single writer and shared object transaction types to test both the fast path and the consensus path.
+        let auth_agg = self.authority_aggregator.load().clone();
+        let validators = auth_agg.committee.names().cloned().collect::<Vec<_>>();
+
+        self.metrics
+            .latency_check_runs
+            .with_label_values(&[tx_type.as_str()])
+            .inc();
+
+        for name in validators {
+            let display_name = auth_agg.get_display_name(&name);
+            let delay_ms = rand::thread_rng().gen_range(0..max_jitter.as_millis()) as u64;
+            let self_clone = self.clone();
+
+            let task = async move {
+                // Add some random delay to the task to avoid all tasks running at the same time
+                if delay_ms > 0 {
+                    tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
+                }
+                let start_time = Instant::now();
+
+                let ping_type = if tx_type == TxType::SingleWriter {
+                    PingType::FastPath
+                } else {
+                    PingType::Consensus
+                };
+
+                // Now send a ping transaction to the chosen validator for the provided tx type
+                match self_clone
+                    .drive_transaction(
+                        SubmitTxRequest::new_ping(ping_type),
+                        SubmitTransactionOptions {
+                            allowed_validators: vec![display_name.clone()],
+                            ..Default::default()
+                        },
+                        Some(ping_timeout),
+                    )
+                    .await
+                {
+                    Ok(_) => {
+                        tracing::debug!(
+                            "Ping transaction to validator {} for tx type {} completed end to end in {} seconds",
+                            display_name,
+                            tx_type.as_str(),
+                            start_time.elapsed().as_secs_f64()
+                        );
+                    }
+                    Err(err) => {
+                        tracing::debug!(
+                            "Failed to get certified finalized effects for tx type {}, for ping transaction to validator {}: {}",
+                            tx_type.as_str(),
+                            display_name,
+                            err
+                        );
+                    }
+                }
+            };
+
+            tasks.spawn(task);
+        }
+    }
+
     fn enable_reconfig(
         self: &Arc<Self>,
         reconfig_observer: Arc<dyn ReconfigObserver<A> + Sync + Send>,