[mfp] accept recently finalized transactions

consensus/core/src/commit_finalizer.rs

@@ -732,7 +732,7 @@ impl CommitFinalizer {
         }
         // current_commit_index is the commit index which includes the current / voting block.
         // When proposing the current block, the latest possible GC round is the GC round computed
-        // from the leader of of the previous commit (current_commit_index - 1).
+        // from the leader of the previous commit (current_commit_index - 1).
         let (commit_index, gc_round) = *gc_rounds
             .get((current_commit_index - 1 - pending_commit_index) as usize)
             .unwrap();

crates/sui-core/src/authority.rs

@@ -1249,13 +1249,13 @@ impl AuthorityState {
             return Err(SuiErrorKind::ValidatorHaltedAtEpochEnd.into());
         }
 
-        // Accept executed transactions, instead of voting to reject them.
-        // Execution is limited to the current epoch. Otherwise there can be a race where
-        // the transaction is accepted but the executed effects are pruned.
-        if let Some(effects) = self
-            .get_transaction_cache_reader()
-            .get_executed_effects(transaction.digest())
-            && effects.executed_epoch() == epoch_store.epoch()
+        // Accept finalized transactions, instead of voting to reject them.
+        // Checking executed transactions is limited to the current epoch.
+        // Otherwise there can be a race where the transaction is accepted because it has executed,
+        // but the executed effects are pruned post consensus, leading to failures.
+        let tx_digest = *transaction.digest();
+        if epoch_store.is_recently_finalized(&tx_digest)
+            || epoch_store.transactions_executed_in_cur_epoch(&[tx_digest])?[0]
         {
             return Ok(());
         }

crates/sui-core/src/authority/authority_per_epoch_store.rs

@@ -18,8 +18,10 @@ use fastcrypto_zkp::bn254::zk_login_api::ZkLoginEnv;
 use futures::FutureExt;
 use futures::future::{Either, join_all, select};
 use itertools::{Itertools, izip};
+use moka::sync::SegmentedCache as MokaCache;
 use move_bytecode_utils::module_cache::SyncModuleCache;
 use mysten_common::assert_reachable;
+use mysten_common::random_util::randomize_cache_capacity_in_tests;
 use mysten_common::sync::notify_once::NotifyOnce;
 use mysten_common::sync::notify_read::NotifyRead;
 use mysten_common::{debug_fatal, fatal};
@@ -427,6 +429,9 @@ pub struct AuthorityPerEpochStore {
     /// A cache that tracks submitted transactions to prevent DoS through excessive resubmissions.
     pub(crate) submitted_transaction_cache: SubmittedTransactionCache,
 
+    /// A cache which tracks recently finalized transactions.
+    pub(crate) finalized_transactions_cache: MokaCache<TransactionDigest, ()>,
+
     /// Waiters for settlement transactions. Used by execution scheduler to wait for
     /// settlement transaction keys to resolve to transactions.
     /// Stored in AuthorityPerEpochStore so that it is automatically cleaned up at the end of the epoch.
@@ -1141,6 +1146,11 @@ impl AuthorityPerEpochStore {
         let submitted_transaction_cache =
             SubmittedTransactionCache::new(None, submitted_transaction_cache_metrics);
 
+        let finalized_transactions_cache = MokaCache::builder(8)
+            .max_capacity(randomize_cache_capacity_in_tests(100_000))
+            .eviction_policy(moka::policy::EvictionPolicy::lru())
+            .build();
+
         let s = Arc::new(Self {
             name,
             committee: committee.clone(),
@@ -1183,6 +1193,7 @@ impl AuthorityPerEpochStore {
             consensus_tx_status_cache,
             tx_reject_reason_cache,
             submitted_transaction_cache,
+            finalized_transactions_cache,
             settlement_registrations: Default::default(),
         });
 
@@ -3642,6 +3653,18 @@ impl AuthorityPerEpochStore {
         }
     }
 
+    /// Caches recent finalized transactions, to avoid revoting them.
+    pub(crate) fn cache_recently_finalized_transaction(&self, tx_digest: TransactionDigest) {
+        self.finalized_transactions_cache.insert(tx_digest, ());
+    }
+
+    /// If true, transaction is recently finalized and should not be voted on.
+    /// If false, the transaction may never be finalized, or has been finalized
+    /// but the info has been evicted from the cache.
+    pub(crate) fn is_recently_finalized(&self, tx_digest: &TransactionDigest) -> bool {
+        self.finalized_transactions_cache.contains_key(tx_digest)
+    }
+
     /// Only used by admin API
     pub async fn get_estimated_tx_cost(&self, tx: &TransactionData) -> Option<u64> {
         self.execution_time_estimator

crates/sui-core/src/consensus_handler.rs

@@ -1915,7 +1915,7 @@ impl<C: CheckpointServiceNotify + Send + Sync> ConsensusHandler<C> {
             // TODO: consider only messages within 1~3 rounds of the leader?
             self.last_consensus_stats.stats.inc_num_messages(author);
 
-            // Set the "ping" transaction status for this block. This is ncecessary as there might be some ping requests waiting for the ping transaction to be certified.
+            // Set the "ping" transaction status for this block. This is necessary as there might be some ping requests waiting for the ping transaction to be certified.
             self.epoch_store.set_consensus_tx_status(
                 ConsensusPosition::ping(epoch, block),
                 ConsensusTxStatus::Finalized,
@@ -2187,9 +2187,14 @@ impl<C: CheckpointServiceNotify + Send + Sync> ConsensusHandler<C> {
             };
 
             let key = verified_transaction.0.key();
+
+            if let Some(tx_digest) = key.user_transaction_digest() {
+                self.epoch_store
+                    .cache_recently_finalized_transaction(tx_digest);
+            }
+
             let in_set = !processed_set.insert(key.clone());
             let in_cache = self.processed_cache.put(key.clone(), ()).is_some();
-
             if in_set || in_cache {
                 self.metrics.skipped_consensus_txns_cache_hit.inc();
                 continue;
@@ -2573,6 +2578,18 @@ pub enum SequencedConsensusTransactionKey {
     System(TransactionDigest),
 }
 
+impl SequencedConsensusTransactionKey {
+    pub fn user_transaction_digest(&self) -> Option<TransactionDigest> {
+        match self {
+            SequencedConsensusTransactionKey::External(key) => match key {
+                ConsensusTransactionKey::Certificate(digest) => Some(*digest),
+                _ => None,
+            },
+            SequencedConsensusTransactionKey::System(_) => None,
+        }
+    }
+}
+
 impl SequencedConsensusTransactionKind {
     pub fn key(&self) -> SequencedConsensusTransactionKey {
         match self {

crates/sui-core/src/consensus_validator.rs

@@ -312,18 +312,21 @@ mod tests {
     use sui_protocol_config::{Chain, ProtocolConfig, ProtocolVersion};
     use sui_types::crypto::deterministic_random_account_key;
     use sui_types::error::{SuiErrorKind, UserInputError};
+    use sui_types::executable_transaction::VerifiedExecutableTransaction;
     use sui_types::messages_checkpoint::{
         CheckpointContents, CheckpointSignatureMessage, CheckpointSummary, SignedCheckpointSummary,
     };
     use sui_types::messages_consensus::ConsensusPosition;
     use sui_types::{
         base_types::{ExecutionDigests, ObjectID},
         crypto::Ed25519SuiSignature,
+        effects::TransactionEffectsAPI as _,
         messages_consensus::ConsensusTransaction,
         object::Object,
         signature::GenericSignature,
     };
 
+    use crate::authority::ExecutionEnv;
     use crate::{
         authority::test_authority_builder::TestAuthorityBuilder,
         checkpoints::CheckpointServiceNoop,
@@ -624,4 +627,68 @@ mod tests {
         let res = validator.verify_batch(&[&bytes]);
         assert!(res.is_ok(), "{res:?}");
     }
+
+    #[sim_test]
+    async fn accept_already_executed_transaction() {
+        let (sender, keypair) = deterministic_random_account_key();
+
+        let gas_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
+        let owned_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
+
+        let network_config =
+            sui_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir()
+                .committee_size(NonZeroUsize::new(1).unwrap())
+                .with_objects(vec![gas_object.clone(), owned_object.clone()])
+                .build();
+
+        let state = TestAuthorityBuilder::new()
+            .with_network_config(&network_config, 0)
+            .build()
+            .await;
+
+        let epoch_store = state.load_epoch_store_one_call_per_task();
+
+        // Create a transaction and execute it.
+        let transaction = test_user_transaction(
+            &state,
+            sender,
+            &keypair,
+            gas_object.clone(),
+            vec![owned_object.clone()],
+        )
+        .await;
+        let tx_digest = *transaction.digest();
+        let cert = VerifiedExecutableTransaction::new_from_quorum_execution(transaction.clone(), 0);
+        let (executed_effects, _) = state
+            .try_execute_immediately(&cert, ExecutionEnv::new(), &state.epoch_store_for_testing())
+            .await
+            .unwrap();
+
+        // Verify the transaction is executed.
+        let read_effects = state
+            .get_transaction_cache_reader()
+            .get_executed_effects(&tx_digest)
+            .expect("Transaction should be executed");
+        assert_eq!(read_effects, executed_effects);
+        assert_eq!(read_effects.executed_epoch(), epoch_store.epoch());
+
+        // Now try to vote on the already executed transaction
+        let serialized_tx = bcs::to_bytes(&ConsensusTransaction::new_user_transaction_message(
+            &state.name,
+            transaction.into_inner().clone(),
+        ))
+        .unwrap();
+        let validator = SuiTxValidator::new(
+            state.clone(),
+            Arc::new(NoopConsensusOverloadChecker {}),
+            Arc::new(CheckpointServiceNoop {}),
+            SuiTxValidatorMetrics::new(&Default::default()),
+        );
+        let rejected_transactions = validator
+            .verify_and_vote_batch(&BlockRef::MAX, &[&serialized_tx])
+            .expect("Verify and vote should succeed");
+
+        // The executed transaction should NOT be rejected.
+        assert!(rejected_transactions.is_empty());
+    }
 }

docs/content/references/ide/debugger.mdx

@@ -92,7 +92,7 @@ There is no stepping "through" native commands. Once the state is inspected, you
 
 ## Usage
 
-Use the debugger to debug unit tests and existing on-chain transactions (including [PTB](/concepts/transactions/prog-txn-blocks) debugging support) with the help of of the [replay tool](/references/cli/replay).
+Use the debugger to debug unit tests and existing on-chain transactions (including [PTB](/concepts/transactions/prog-txn-blocks) debugging support) with the help of the [replay tool](/references/cli/replay).
 
 ### Debugging unit tests