upstream(core): Remove executed_in_epoch table (#8550)

# Description of change

- Upstream range: [v1.44.3, v1.45.3)
- Port commit: 
-
MystenLabs/sui@efdbe99
- Description:
The table is replaced with:
- An in-memory "dirty set" which holds executed but un-checkpointed
transaction digests. Transactions are removed from the dirty set by
CheckpointExecutor.
- An additional bounded cache intended to lessen the number of db reads
by CheckpointBuilder
- Last-resort reads go to the `executed_transactions_to_checkpoint`
table.


## Links to any relevant issues

Part of #6153 

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [ ] Patch-specific tests (correctness, functionality coverage)
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have checked that new and existing unit tests pass locally with
my changes

Author: jkrvivian

crates/iota-core/src/authority.rs

@@ -3225,9 +3225,7 @@ impl AuthorityState {
         );
 
         if cfg!(debug_assertions) {
-            cur_epoch_store
-                .check_all_executed_transactions_in_checkpoint()
-                .expect("failed to check all executed transactions in checkpoint");
+            cur_epoch_store.check_all_executed_transactions_in_checkpoint();
         }
 
         self.get_reconfig_api()

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

@@ -21,7 +21,10 @@ use futures::{
     FutureExt,
     future::{Either, join_all, select},
 };
-use iota_common::sync::{notify_once::NotifyOnce, notify_read::NotifyRead};
+use iota_common::{
+    fatal,
+    sync::{notify_once::NotifyOnce, notify_read::NotifyRead},
+};
 use iota_config::node::ExpensiveSafetyCheckConfig;
 use iota_execution::{self, Executor};
 use iota_macros::{fail_point, fail_point_arg};
@@ -537,6 +540,8 @@ pub struct AuthorityEpochTables {
     transaction_cert_signatures: DBMap<TransactionDigest, AuthorityStrongQuorumSignInfo>,
 
     /// Transactions that were executed in the current epoch.
+    #[allow(dead_code)]
+    #[deprecated]
     executed_in_epoch: DBMap<TransactionDigest, ()>,
 
     #[allow(dead_code)]
@@ -824,15 +829,6 @@ impl AuthorityEpochTables {
             .safe_iter()
             .collect::<Result<_, _>>()?)
     }
-
-    fn get_all_user_signatures_for_checkpoints(
-        &self,
-    ) -> IotaResult<HashMap<TransactionDigest, Vec<GenericSignature>>> {
-        Ok(self
-            .user_signatures_for_checkpoints
-            .safe_iter()
-            .collect::<Result<_, _>>()?)
-    }
 }
 
 pub(crate) const MUTEX_TABLE_SIZE: usize = 1024;
@@ -1303,18 +1299,15 @@ impl AuthorityPerEpochStore {
         tx_digest: &TransactionDigest,
     ) -> IotaResult {
         let tables = self.tables()?;
-        let mut batch = self.tables()?.executed_in_epoch.batch();
 
-        batch.insert_batch(&tables.executed_in_epoch, [(tx_digest, ())])?;
-
-        if !matches!(tx_key, TransactionKey::Digest(_)) {
-            batch.insert_batch(&tables.transaction_key_to_digest, [(tx_key, tx_digest)])?;
-        }
-        batch.write()?;
+        self.consensus_output_cache
+            .insert_executed_in_epoch(*tx_digest);
 
         if !matches!(tx_key, TransactionKey::Digest(_)) {
+            tables.transaction_key_to_digest.insert(tx_key, tx_digest)?;
             self.executed_digests_notify_read.notify(tx_key, tx_digest);
         }
+
         Ok(())
     }
 
@@ -1331,9 +1324,10 @@ impl AuthorityPerEpochStore {
     }
 
     pub fn revert_executed_transaction(&self, tx_digest: &TransactionDigest) -> IotaResult {
+        self.consensus_output_cache
+            .remove_reverted_transaction(tx_digest);
         let tables = self.tables()?;
         let mut batch = tables.effects_signatures.batch();
-        batch.delete_batch(&tables.executed_in_epoch, [*tx_digest])?;
         batch.delete_batch(&tables.effects_signatures, [*tx_digest])?;
         batch.write()?;
         Ok(())
@@ -1356,14 +1350,30 @@ impl AuthorityPerEpochStore {
         Ok(())
     }
 
-    pub fn transactions_executed_in_cur_epoch<'a>(
+    pub fn transactions_executed_in_cur_epoch(
         &self,
-        digests: impl IntoIterator<Item = &'a TransactionDigest>,
+        digests: &[TransactionDigest],
     ) -> IotaResult<Vec<bool>> {
-        Ok(self
-            .tables()?
-            .executed_in_epoch
-            .multi_contains_keys(digests)?)
+        let tables = self.tables()?;
+        Ok(do_fallback_lookup(
+            digests,
+            |digest| {
+                if self
+                    .consensus_output_cache
+                    .executed_in_current_epoch(digest)
+                {
+                    CacheResult::Hit(true)
+                } else {
+                    CacheResult::Miss
+                }
+            },
+            |digests| {
+                tables
+                    .executed_transactions_to_checkpoint
+                    .multi_contains_keys(digests)
+                    .expect("db error")
+            },
+        ))
     }
 
     pub fn get_effects_signature(
@@ -1522,6 +1532,9 @@ impl AuthorityPerEpochStore {
         quarantine.update_highest_executed_checkpoint(seq, self, &mut batch)?;
         batch.write()?;
 
+        self.consensus_output_cache
+            .remove_executed_in_epoch(digests);
+
         Ok(())
     }
 
@@ -4348,32 +4361,23 @@ impl AuthorityPerEpochStore {
         self.signature_verifier.clear_signature_cache();
     }
 
-    pub(crate) fn check_all_executed_transactions_in_checkpoint(&self) -> IotaResult<()> {
-        let tables = self.tables().unwrap();
-
-        info!("Verifying that all executed transactions are in a checkpoint");
-
-        let mut executed_iter = tables.executed_in_epoch.safe_iter();
-        let mut checkpointed_iter = tables.executed_transactions_to_checkpoint.safe_iter();
+    pub(crate) fn check_all_executed_transactions_in_checkpoint(&self) {
+        let uncheckpointed_transactions = self
+            .consensus_output_cache
+            .get_uncheckpointed_transactions();
 
-        // verify that the two iterators (which are both sorted) are identical
-        loop {
-            let executed = executed_iter.next().transpose()?;
-            let checkpointed = checkpointed_iter.next().transpose()?;
-            match (executed, checkpointed) {
-                (Some((left, ())), Some((right, _))) => {
-                    if left != right {
-                        panic!(
-                            "Executed transactions and checkpointed transactions do not match: {left:?} {right:?}"
-                        );
-                    }
-                }
-                (None, None) => break Ok(()),
-                (left, right) => panic!(
-                    "Executed transactions and checkpointed transactions do not match: {left:?} {right:?}"
-                ),
-            }
+        if uncheckpointed_transactions.is_empty() {
+            info!("Verified that all executed transactions are in a checkpoint");
+            return;
         }
+
+        // TODO: should this be debug_fatal? Its potentially very serious in that it
+        // could indicate that we broke the checkpoint inclusion guarantee, but
+        // we won't be able to do anything about it if it happens.
+        fatal!(
+            "The following transactions were neither reverted nor checkpointed: {:?}",
+            uncheckpointed_transactions
+        );
     }
 }
 

crates/iota-core/src/authority/consensus_quarantine.rs

@@ -17,7 +17,9 @@ use iota_types::{
     messages_consensus::{TimestampMs, VersionedDkgConfirmation},
     signature::GenericSignature,
 };
+use moka::{policy::EvictionPolicy, sync::SegmentedCache as MokaCache};
 use parking_lot::Mutex;
+use rand::seq::SliceRandom;
 use tracing::{debug, info};
 use typed_store::{Map, rocks::DBBatch};
 
@@ -316,6 +318,9 @@ pub(crate) struct ConsensusOutputCache {
     pub(super) user_signatures_for_checkpoints:
         Mutex<HashMap<TransactionDigest, Vec<GenericSignature>>>,
 
+    executed_in_epoch: RwLock<DashMap<TransactionDigest, ()>>,
+    executed_in_epoch_cache: MokaCache<TransactionDigest, ()>,
+
     metrics: Arc<EpochMetrics>,
 }
 
@@ -333,28 +338,31 @@ impl ConsensusOutputCache {
             .get_all_deferred_transactions_v2()
             .expect("load deferred transactions v2 cannot fail");
 
-        if !epoch_start_configuration.is_data_quarantine_active_from_beginning_of_epoch() {
-            let shared_version_assignments = Self::get_all_shared_version_assignments(tables);
-
-            let user_signatures_for_checkpoints = tables
-                .get_all_user_signatures_for_checkpoints()
-                .expect("load user signatures for checkpoints cannot fail");
+        assert!(
+            epoch_start_configuration.is_data_quarantine_active_from_beginning_of_epoch(),
+            "This version of iota-node can only run after data quarantining has been enabled. Please run the latest version to the end of the current epoch and retry"
+        );
 
-            Self {
-                shared_version_assignments: shared_version_assignments.into_iter().collect(),
-                deferred_transactions: Mutex::new(deferred_transactions),
-                deferred_transactions_v2: Mutex::new(deferred_transactions_v2),
-                user_signatures_for_checkpoints: Mutex::new(user_signatures_for_checkpoints),
-                metrics,
-            }
+        let executed_in_epoch_cache_capacity = if cfg!(msim) {
+            // Ensure that we test under conditions of constant, frequent,
+            // and rare cache evictions.
+            *[2, 100, 50000].choose(&mut rand::thread_rng()).unwrap()
         } else {
-            Self {
-                shared_version_assignments: Default::default(),
-                deferred_transactions: Mutex::new(deferred_transactions),
-                deferred_transactions_v2: Mutex::new(deferred_transactions_v2),
-                user_signatures_for_checkpoints: Default::default(),
-                metrics,
-            }
+            50_000
+        };
+
+        Self {
+            shared_version_assignments: Default::default(),
+            deferred_transactions: Mutex::new(deferred_transactions),
+            deferred_transactions_v2: Mutex::new(deferred_transactions_v2),
+            user_signatures_for_checkpoints: Default::default(),
+            executed_in_epoch: RwLock::new(DashMap::with_shard_amount(2048)),
+            executed_in_epoch_cache: MokaCache::builder(8)
+                // most queries should be for recent transactions
+                .max_capacity(executed_in_epoch_cache_capacity)
+                .eviction_policy(EvictionPolicy::lru())
+                .build(),
+            metrics,
         }
     }
 
@@ -414,17 +422,50 @@ impl ConsensusOutputCache {
             .sub(removed_count as i64);
     }
 
-    // Used to read pre-existing shared object versions from the database after a
-    // crash. TODO: remove this once all nodes have upgraded to
-    // data-quarantining.
-    fn get_all_shared_version_assignments(
-        tables: &AuthorityEpochTables,
-    ) -> Vec<(TransactionKey, Vec<(ObjectID, SequenceNumber)>)> {
-        tables
-            .assigned_shared_object_versions
-            .safe_iter()
-            .collect::<Result<Vec<_>, _>>()
-            .expect("db error")
+    pub fn executed_in_current_epoch(&self, digest: &TransactionDigest) -> bool {
+        self.executed_in_epoch
+            .read()
+            .contains_key(digest) ||
+            // we use get instead of contains key to mark the entry as read
+            self.executed_in_epoch_cache.get(digest).is_some()
+    }
+
+    // Called by execution
+    pub fn insert_executed_in_epoch(&self, tx_digest: TransactionDigest) {
+        assert!(
+            self.executed_in_epoch
+                .read()
+                .insert(tx_digest, ())
+                .is_none(),
+            "transaction already executed"
+        );
+        self.executed_in_epoch_cache.insert(tx_digest, ());
+    }
+
+    // CheckpointExecutor calls this (indirectly) in order to prune the in-memory
+    // cache of executed transactions. By the time this is called, the
+    // transaction digests will have been committed to
+    // the `executed_transactions_to_checkpoint` table.
+    pub fn remove_executed_in_epoch(&self, tx_digests: &[TransactionDigest]) {
+        let executed_in_epoch = self.executed_in_epoch.read();
+        for tx_digest in tx_digests {
+            executed_in_epoch.remove(tx_digest);
+        }
+    }
+
+    pub fn remove_reverted_transaction(&self, tx_digest: &TransactionDigest) {
+        // reverted transactions are not guaranteed to have been executed
+        self.executed_in_epoch.read().remove(tx_digest);
+    }
+
+    /// At reconfig time, all checkpointed transactions must have been removed
+    /// from self.executed_in_epoch
+    pub fn get_uncheckpointed_transactions(&self) -> Vec<TransactionDigest> {
+        self.executed_in_epoch
+            .write() // exclusive lock to ensure consistent view
+            .iter()
+            .map(|e| *e.key())
+            .collect()
     }
 }
 

crates/iota-core/src/authority/epoch_start_configuration.rs

@@ -52,7 +52,7 @@ pub trait EpochStartConfigTrait {
 // is a collision in the value of some variant, the branch which has been
 // released should take precedence. In this case, the picked-from branch is
 // inconsistent with the released branch, and must be fixed.
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd)]
 pub enum EpochFlag {
     // When switching between different cache types mid-epoch, partial checkpoint transactions
     // might already be on disk. During lock initialization, we check if there is any existing
@@ -63,21 +63,36 @@ pub enum EpochFlag {
     // This flag indicates whether data quarantining has been enabled from the
     // beginning of the epoch.
     DataQuarantineFromBeginningOfEpoch = 1,
+
+    // Used for `test_epoch_flag_upgrade`.
+    #[cfg(msim)]
+    DummyFlag = 2,
 }
 
 impl EpochFlag {
     pub fn default_flags_for_new_epoch(config: &NodeConfig) -> Vec<Self> {
         Self::default_flags_impl(config.execution_cache)
     }
 
+    // Return flags that are mandatory for the current version of the code. This is
+    // used so that `test_epoch_flag_upgrade` can still work correctly even when
+    // there are no optional flags.
+    pub fn mandatory_flags() -> Vec<Self> {
+        vec![EpochFlag::DataQuarantineFromBeginningOfEpoch]
+    }
+
     /// For situations in which there is no config available (e.g. setting up a
     /// downloaded snapshot).
     pub fn default_for_no_config() -> Vec<Self> {
         Self::default_flags_impl(Default::default())
     }
 
     fn default_flags_impl(cache_type: ExecutionCacheType) -> Vec<Self> {
-        let mut new_flags = vec![EpochFlag::DataQuarantineFromBeginningOfEpoch];
+        let mut new_flags = vec![
+            EpochFlag::DataQuarantineFromBeginningOfEpoch,
+            #[cfg(msim)]
+            EpochFlag::DummyFlag,
+        ];
 
         // Load cache type from env
         if matches!(cache_type.cache_type(), ExecutionCacheType::WritebackCache) {
@@ -97,6 +112,10 @@ impl fmt::Display for EpochFlag {
             EpochFlag::DataQuarantineFromBeginningOfEpoch => {
                 write!(f, "DataQuarantineFromBeginningOfEpoch")
             }
+            #[cfg(msim)]
+            EpochFlag::DummyFlag => {
+                write!(f, "DummyFlag")
+            }
         }
     }
 }

crates/iota-core/src/checkpoints/mod.rs

@@ -1823,7 +1823,7 @@ impl CheckpointBuilder {
 
                 let existing_effects = self
                     .epoch_store
-                    .transactions_executed_in_cur_epoch(effect.dependencies().iter())?;
+                    .transactions_executed_in_cur_epoch(effect.dependencies())?;
 
                 for (dependency, effects_signature_exists) in
                     effect.dependencies().iter().zip(existing_effects.iter())