[TransactionOrchestrator] enhance retries on transient failures (#24649)

- Process transactions in a separate task in TO, to avoid cancelled by
user dropping the request.
- Retry transient errors in the background.
- Avoid casting reject votes because of consensus overload, against
transactions already in consensus. This should reduce the chance of
rejecting valid transactions.

CI

Author: mwtian

crates/sui-core/src/authority.rs

@@ -1281,6 +1281,7 @@ impl AuthorityState {
         &self,
         epoch_store: &Arc<AuthorityPerEpochStore>,
         transaction: &VerifiedTransaction,
+        enforce_live_input_objects: bool,
     ) -> SuiResult<()> {
         if !epoch_store
             .get_reconfig_state_read_lock_guard()
@@ -1313,6 +1314,17 @@ impl AuthorityState {
                     .into());
                 }
 
+                if enforce_live_input_objects && live_object.version() < obj_ref.1 {
+                    // Returns a non-retriable error when the input object version is required to be live.
+                    return Err(SuiErrorKind::UserInputError {
+                        error: UserInputError::ObjectVersionUnavailableForConsumption {
+                            provided_obj_ref: *obj_ref,
+                            current_version: live_object.version(),
+                        },
+                    }
+                    .into());
+                }
+
                 // If version matches, verify digest also matches
                 if obj_ref.1 == live_object.version() && obj_ref.2 != live_object.digest() {
                     return Err(SuiErrorKind::UserInputError {

crates/sui-core/src/consensus_validator.rs

@@ -22,10 +22,11 @@ use tracing::{debug, info, instrument, warn};
 use crate::{
     authority::{AuthorityState, authority_per_epoch_store::AuthorityPerEpochStore},
     checkpoints::CheckpointServiceNotify,
-    consensus_adapter::ConsensusOverloadChecker,
+    consensus_adapter::{ConsensusOverloadChecker, NoopConsensusOverloadChecker},
 };
 
-/// Allows verifying the validity of transactions
+/// Validates transactions from consensus and votes on whether to execute the transactions
+/// based on their validity and the current state of the authority.
 #[derive(Clone)]
 pub struct SuiTxValidator {
     authority_state: Arc<AuthorityState>,
@@ -37,7 +38,6 @@ pub struct SuiTxValidator {
 impl SuiTxValidator {
     pub fn new(
         authority_state: Arc<AuthorityState>,
-        consensus_overload_checker: Arc<dyn ConsensusOverloadChecker>,
         checkpoint_service: Arc<dyn CheckpointServiceNotify + Send + Sync>,
         metrics: Arc<SuiTxValidatorMetrics>,
     ) -> Self {
@@ -46,6 +46,8 @@ impl SuiTxValidator {
             "SuiTxValidator constructed for epoch {}",
             epoch_store.epoch()
         );
+        // Intentionally do not check consensus overload, because this is validating transactions already in consensus.
+        let consensus_overload_checker = Arc::new(NoopConsensusOverloadChecker {});
         Self {
             authority_state,
             consensus_overload_checker,
@@ -330,9 +332,8 @@ mod tests {
     use crate::{
         authority::test_authority_builder::TestAuthorityBuilder,
         checkpoints::CheckpointServiceNoop,
-        consensus_adapter::{
-            NoopConsensusOverloadChecker,
-            consensus_tests::{test_certificates, test_gas_objects, test_user_transaction},
+        consensus_adapter::consensus_tests::{
+            test_certificates, test_gas_objects, test_user_transaction,
         },
         consensus_validator::{SuiTxValidator, SuiTxValidatorMetrics},
     };
@@ -364,12 +365,8 @@ mod tests {
         .unwrap();
 
         let metrics = SuiTxValidatorMetrics::new(&Default::default());
-        let validator = SuiTxValidator::new(
-            state.clone(),
-            Arc::new(NoopConsensusOverloadChecker {}),
-            Arc::new(CheckpointServiceNoop {}),
-            metrics,
-        );
+        let validator =
+            SuiTxValidator::new(state.clone(), Arc::new(CheckpointServiceNoop {}), metrics);
         let res = validator.verify_batch(&[&first_transaction_bytes]);
         assert!(res.is_ok(), "{res:?}");
 
@@ -481,7 +478,6 @@ mod tests {
 
         let validator = SuiTxValidator::new(
             state.clone(),
-            Arc::new(NoopConsensusOverloadChecker {}),
             Arc::new(CheckpointServiceNoop {}),
             SuiTxValidatorMetrics::new(&Default::default()),
         );
@@ -565,7 +561,6 @@ mod tests {
 
         let validator = SuiTxValidator::new(
             state.clone(),
-            Arc::new(NoopConsensusOverloadChecker {}),
             Arc::new(CheckpointServiceNoop {}),
             SuiTxValidatorMetrics::new(&Default::default()),
         );
@@ -619,7 +614,6 @@ mod tests {
 
         let validator = SuiTxValidator::new(
             state.clone(),
-            Arc::new(NoopConsensusOverloadChecker {}),
             Arc::new(CheckpointServiceNoop {}),
             SuiTxValidatorMetrics::new(&Default::default()),
         );
@@ -680,7 +674,6 @@ mod tests {
         .unwrap();
         let validator = SuiTxValidator::new(
             state.clone(),
-            Arc::new(NoopConsensusOverloadChecker {}),
             Arc::new(CheckpointServiceNoop {}),
             SuiTxValidatorMetrics::new(&Default::default()),
         );