Remove executed_in_epoch table. (#21477)

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.

The only purpose of the table was to allow CheckpointBuilder to prune
transaction dependencies from prior epochs, and the above approach
suffices while removing a surprisingly expensive source of db writes.

Author: mystenmark

crates/sui-core/src/authority.rs

@@ -3317,9 +3317,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();
         }
 
         if let Err(err) = self

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

@@ -441,6 +441,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)]
@@ -751,15 +753,6 @@ impl AuthorityEpochTables {
             .safe_iter()
             .collect::<Result<_, _>>()?)
     }
-
-    fn get_all_user_signatures_for_checkpoints(
-        &self,
-    ) -> SuiResult<HashMap<TransactionDigest, Vec<GenericSignature>>> {
-        Ok(self
-            .user_signatures_for_checkpoints
-            .safe_iter()
-            .collect::<Result<_, _>>()?)
-    }
 }
 
 pub(crate) const MUTEX_TABLE_SIZE: usize = 1024;
@@ -1402,18 +1395,15 @@ impl AuthorityPerEpochStore {
         tx_digest: &TransactionDigest,
     ) -> SuiResult {
         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(())
     }
 
@@ -1430,9 +1420,10 @@ impl AuthorityPerEpochStore {
     }
 
     pub fn revert_executed_transaction(&self, tx_digest: &TransactionDigest) -> SuiResult {
+        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(())
@@ -1455,14 +1446,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],
     ) -> SuiResult<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(
@@ -1655,6 +1662,9 @@ impl AuthorityPerEpochStore {
         quarantine.update_highest_executed_checkpoint(seq, self, &mut batch)?;
         batch.write()?;
 
+        self.consensus_output_cache
+            .remove_executed_in_epoch(digests);
+
         Ok(())
     }
 
@@ -4415,31 +4425,23 @@ impl AuthorityPerEpochStore {
         self.signature_verifier.clear_signature_cache();
     }
 
-    pub(crate) fn check_all_executed_transactions_in_checkpoint(&self) -> SuiResult<()> {
-        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/sui-core/src/authority/consensus_quarantine.rs

@@ -1,8 +1,6 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::collections::{hash_map, BTreeMap, BTreeSet, HashMap, VecDeque};
-
 use crate::authority::authority_per_epoch_store::{
     AuthorityEpochTables, EncG, ExecutionIndicesWithStats, PkG,
 };
@@ -13,8 +11,12 @@ use crate::epoch::randomness::SINGLETON_KEY;
 use dashmap::DashMap;
 use fastcrypto_tbls::{dkg_v1, nodes::PartyId};
 use fastcrypto_zkp::bn254::zk_login::{JwkId, JWK};
+use moka::policy::EvictionPolicy;
+use moka::sync::SegmentedCache as MokaCache;
 use mysten_common::fatal;
 use parking_lot::Mutex;
+use rand::seq::SliceRandom;
+use std::collections::{hash_map, BTreeMap, BTreeSet, HashMap, VecDeque};
 use sui_types::authenticator_state::ActiveJwk;
 use sui_types::base_types::{AuthorityName, SequenceNumber};
 use sui_types::crypto::RandomnessRound;
@@ -383,6 +385,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>,
 }
 
@@ -396,27 +401,30 @@ impl ConsensusOutputCache {
             .get_all_deferred_transactions()
             .expect("load deferred transactions cannot fail");
 
-        if !epoch_start_configuration.is_data_quarantine_active_from_beginning_of_epoch() {
-            let shared_version_assignments =
-                Self::get_all_shared_version_assignments(epoch_start_configuration, 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 sui-node can only run after data quarantining has been enabled. Please run version 1.45.0 or later 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),
-                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),
-                user_signatures_for_checkpoints: Default::default(),
-                metrics,
-            }
+            50_000
+        };
+
+        Self {
+            shared_version_assignments: Default::default(),
+            deferred_transactions: Mutex::new(deferred_transactions),
+            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,
         }
     }
 
@@ -476,40 +484,49 @@ 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(
-        epoch_start_configuration: &EpochStartConfiguration,
-        tables: &AuthorityEpochTables,
-    ) -> Vec<(
-        TransactionKey,
-        Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
-    )> {
-        if epoch_start_configuration.use_version_assignment_tables_v3() {
-            tables
-                .assigned_shared_object_versions_v3
-                .safe_iter()
-                .collect::<Result<_, _>>()
-                .expect("db error")
-        } else {
-            tables
-                .assigned_shared_object_versions_v2
-                .safe_iter()
-                .collect::<Result<Vec<_>, _>>()
-                .expect("db error")
-                .into_iter()
-                .map(|(key, value)| {
-                    (
-                        key,
-                        value
-                            .into_iter()
-                            .map(|(id, v)| ((id, SequenceNumber::UNKNOWN), v))
-                            .collect::<Vec<_>>(),
-                    )
-                })
-                .collect()
+    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()
+    }
 }
 
 /// ConsensusOutputQuarantine holds outputs of consensus processing in memory until the checkpoints

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

@@ -68,6 +68,10 @@ pub enum EpochFlag {
     // This flag indicates whether data quarantining has been enabled from the
     // beginning of the epoch.
     DataQuarantineFromBeginningOfEpoch = 9,
+
+    // Used for `test_epoch_flag_upgrade`.
+    #[cfg(msim)]
+    DummyFlag = 10,
 }
 
 impl EpochFlag {
@@ -77,6 +81,16 @@ impl EpochFlag {
         Self::default_flags_impl()
     }
 
+    // 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::UseVersionAssignmentTablesV3,
+            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()
@@ -86,6 +100,8 @@ impl EpochFlag {
         vec![
             EpochFlag::UseVersionAssignmentTablesV3,
             EpochFlag::DataQuarantineFromBeginningOfEpoch,
+            #[cfg(msim)]
+            EpochFlag::DummyFlag,
         ]
     }
 }
@@ -124,6 +140,10 @@ impl fmt::Display for EpochFlag {
             EpochFlag::DataQuarantineFromBeginningOfEpoch => {
                 write!(f, "DataQuarantineFromBeginningOfEpoch")
             }
+            #[cfg(msim)]
+            EpochFlag::DummyFlag => {
+                write!(f, "DummyFlag")
+            }
         }
     }
 }

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

@@ -1853,7 +1853,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())

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

@@ -739,8 +739,7 @@ async fn test_epoch_flag_upgrade() {
             return None;
         }
 
-        // start with only UseVersionAssignmentTablesV3
-        let flags: Vec<EpochFlag> = vec![EpochFlag::UseVersionAssignmentTablesV3];
+        let flags: Vec<EpochFlag> = EpochFlag::mandatory_flags();
         Some(flags)
     });