iota_core/authority/

authority_store.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{iter, mem, sync::Arc, thread};

use either::Either;
use fastcrypto::hash::{HashFunction, Sha3_256};
use futures::stream::FuturesUnordered;
use iota_common::sync::notify_read::NotifyRead;
use iota_config::{migration_tx_data::MigrationTxData, node::AuthorityStorePruningConfig};
use iota_genesis_common::MigrationTxDataExt;
use iota_macros::fail_point_arg;
use iota_sdk_types::{TransactionEffects, TransactionEvents, Version};
use iota_storage::mutex_table::{MutexGuard, MutexTable};
use iota_types::{
    base_types::VerifiedExecutionData,
    effects::TransactionEffectsExt,
    error::UserInputError,
    execution::TypeLayoutStore,
    fp_bail, fp_ensure,
    global_state_hash::GlobalStateHash,
    iota_system_state::{
        get_iota_system_state, iota_system_state_summary::IotaSystemStateSummaryV2,
    },
    messages_checkpoint::CheckpointContentsExt,
    storage::{
        BackingPackageStore, MarkerValue, ObjectKey, ObjectOrTombstone, ObjectStore, get_module,
    },
};
use itertools::izip;
use move_core_types::resolver::ModuleResolver;
use tokio::time::Instant;
use tracing::{debug, info, trace};
use typed_store::{
    TypedStoreError,
    rocks::{DBBatch, DBMap},
    traits::Map,
};

use super::{
    authority_store_tables::{AuthorityPerpetualTables, LiveObject},
    *,
};
use crate::{
    authority::{
        authority_per_epoch_store::{AuthorityPerEpochStore, LockDetails},
        authority_store_pruner::{
            AuthorityStorePruner, AuthorityStorePruningMetrics, EPOCH_DURATION_MS_FOR_TESTING,
        },
        authority_store_tables::TotalIotaSupplyCheck,
        authority_store_types::{StoreObject, StoreObjectWrapper, get_store_object},
        epoch_start_configuration::{EpochFlag, EpochStartConfiguration},
    },
    global_state_hasher::GlobalStateHashStore,
    grpc_indexes::GrpcIndexesStore,
    transaction_outputs::TransactionOutputs,
};

const NUM_SHARDS: usize = 4096;

struct AuthorityStoreMetrics {
    iota_conservation_check_latency: IntGauge,
    iota_conservation_live_object_count: IntGauge,
    iota_conservation_live_object_size: IntGauge,
    iota_conservation_imbalance: IntGauge,
    iota_conservation_storage_fund: IntGauge,
    iota_conservation_storage_fund_imbalance: IntGauge,
    epoch_flags: IntGaugeVec,
}

impl AuthorityStoreMetrics {
    pub fn new(registry: &Registry) -> Self {
        Self {
            iota_conservation_check_latency: register_int_gauge_with_registry!(
                "iota_conservation_check_latency",
                "Number of seconds took to scan all live objects in the store for IOTA conservation check",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            iota_conservation_live_object_count: register_int_gauge_with_registry!(
                "iota_conservation_live_object_count",
                "Number of live objects in the store",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            iota_conservation_live_object_size: register_int_gauge_with_registry!(
                "iota_conservation_live_object_size",
                "Size in bytes of live objects in the store",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            iota_conservation_imbalance: register_int_gauge_with_registry!(
                "iota_conservation_imbalance",
                "Total amount of IOTA in the network - 10B * 10^9. This delta shows the amount of imbalance",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            iota_conservation_storage_fund: register_int_gauge_with_registry!(
                "iota_conservation_storage_fund",
                "Storage Fund pool balance (only includes the storage fund proper that represents object storage)",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            iota_conservation_storage_fund_imbalance: register_int_gauge_with_registry!(
                "iota_conservation_storage_fund_imbalance",
                "Imbalance of storage fund, computed with storage_fund_balance - total_object_storage_rebates",
                registry;
                MetricLevel::Warn,
            ).unwrap(),
            epoch_flags: register_int_gauge_vec_with_registry!(
                "epoch_flags",
                "Local flags of the currently running epoch",
                &["flag"],
                registry,
            ).unwrap(),
        }
    }
}

/// The `AuthorityStore` manages the state and operations of an authority's
/// store. It includes a `mutex_table` to handle concurrent writes to the
/// database and references to various tables stored in
/// `AuthorityPerpetualTables`. The struct provides mechanisms for initializing
/// and accessing locks, managing objects and transactions, and performing
/// epoch-specific operations. It also includes methods for recovering from
/// crashes, checking IOTA conservation, and handling object markers and states
/// during epoch transitions.
pub struct AuthorityStore {
    /// Internal vector of locks to manage concurrent writes to the database
    mutex_table: MutexTable<ObjectDigest>,

    pub(crate) perpetual_tables: Arc<AuthorityPerpetualTables>,

    pub(crate) root_state_notify_read:
        NotifyRead<EpochId, (CheckpointSequenceNumber, GlobalStateHash)>,

    /// Whether to enable expensive IOTA conservation check at epoch boundaries.
    enable_epoch_iota_conservation_check: bool,

    metrics: AuthorityStoreMetrics,
}

pub type ExecutionLockReadGuard<'a> = tokio::sync::RwLockReadGuard<'a, EpochId>;
pub type ExecutionLockWriteGuard<'a> = tokio::sync::RwLockWriteGuard<'a, EpochId>;

impl AuthorityStore {
    /// Open an authority store by directory path.
    /// If the store is empty, initialize it using genesis.
    pub async fn open(
        perpetual_tables: Arc<AuthorityPerpetualTables>,
        genesis: &Genesis,
        config: &NodeConfig,
        registry: &Registry,
        migration_tx_data: Option<&MigrationTxData>,
    ) -> IotaResult<Arc<Self>> {
        let enable_epoch_iota_conservation_check = config
            .expensive_safety_check_config
            .enable_epoch_iota_conservation_check();

        let epoch_start_configuration = if perpetual_tables.database_is_empty()? {
            info!("Creating new epoch start config from genesis");

            #[cfg_attr(not(any(msim, fail_points)), expect(unused_mut))]
            let mut initial_epoch_flags = EpochFlag::default_flags_for_new_epoch(config);
            fail_point_arg!("initial_epoch_flags", |flags: Vec<EpochFlag>| {
                info!("Setting initial epoch flags to {:?}", flags);
                initial_epoch_flags = flags;
            });

            let epoch_start_configuration = EpochStartConfiguration::new(
                genesis.iota_system_object().into_epoch_start_state(),
                *genesis.checkpoint().digest(),
                &genesis.objects(),
                initial_epoch_flags,
            )?;
            perpetual_tables.set_epoch_start_configuration(&epoch_start_configuration)?;
            epoch_start_configuration
        } else {
            info!("Loading epoch start config from DB");
            perpetual_tables
                .epoch_start_configuration
                .get(&())?
                .expect("Epoch start configuration must be set in non-empty DB")
        };
        let cur_epoch = perpetual_tables.get_recovery_epoch_at_restart()?;
        info!("Epoch start config: {:?}", epoch_start_configuration);
        info!("Cur epoch: {:?}", cur_epoch);
        let this = Self::open_inner(
            genesis,
            perpetual_tables,
            enable_epoch_iota_conservation_check,
            registry,
            migration_tx_data,
        )
        .await?;
        this.update_epoch_flags_metrics(&[], epoch_start_configuration.flags());
        Ok(this)
    }

    pub fn update_epoch_flags_metrics(&self, old: &[EpochFlag], new: &[EpochFlag]) {
        for flag in old {
            self.metrics
                .epoch_flags
                .with_label_values(&[&flag.to_string()])
                .set(0);
        }
        for flag in new {
            self.metrics
                .epoch_flags
                .with_label_values(&[&flag.to_string()])
                .set(1);
        }
    }

    // NB: This must only be called at time of reconfiguration. We take the
    // execution lock write guard as an argument to ensure that this is the
    // case.
    pub fn clear_object_per_epoch_marker_table(
        &self,
        _execution_guard: &ExecutionLockWriteGuard<'_>,
    ) -> IotaResult<()> {
        // We can safely delete all entries in the per epoch marker table since this is
        // only called at epoch boundaries (during reconfiguration). Therefore
        // any entries that currently exist can be removed. Because of this we
        // can use the `schedule_delete_all` method.
        Ok(self
            .perpetual_tables
            .object_per_epoch_marker_table
            .schedule_delete_all()?)
    }

    pub async fn open_with_committee_for_testing(
        perpetual_tables: Arc<AuthorityPerpetualTables>,
        committee: &Committee,
        genesis: &Genesis,
    ) -> IotaResult<Arc<Self>> {
        // TODO: Since we always start at genesis, the committee should be technically
        // the same as the genesis committee.
        assert_eq!(committee.epoch, 0);
        Self::open_inner(genesis, perpetual_tables, true, &Registry::new(), None).await
    }

    async fn open_inner(
        genesis: &Genesis,
        perpetual_tables: Arc<AuthorityPerpetualTables>,
        enable_epoch_iota_conservation_check: bool,
        registry: &Registry,
        migration_tx_data: Option<&MigrationTxData>,
    ) -> IotaResult<Arc<Self>> {
        let store = Arc::new(Self {
            mutex_table: MutexTable::new(NUM_SHARDS),
            perpetual_tables,
            root_state_notify_read: NotifyRead::<
                EpochId,
                (CheckpointSequenceNumber, GlobalStateHash),
            >::new(),
            enable_epoch_iota_conservation_check,
            metrics: AuthorityStoreMetrics::new(registry),
        });
        // Only initialize an empty database.
        if store
            .database_is_empty()
            .expect("database read should not fail at init.")
        {
            // Initialize with genesis data
            // First insert genesis objects
            store
                .bulk_insert_genesis_objects(genesis.objects())
                .expect("cannot bulk insert genesis objects");

            // Then insert txn and effects of genesis
            let transaction = VerifiedTransaction::new_unchecked(genesis.transaction().clone());
            store
                .perpetual_tables
                .transactions
                .insert(transaction.digest(), transaction.serializable_ref())
                .expect("cannot insert genesis transaction");
            store
                .perpetual_tables
                .effects
                .insert(&genesis.effects().digest(), genesis.effects())
                .expect("cannot insert genesis effects");

            // In the previous step we don't insert the effects to executed_effects yet
            // because the genesis tx hasn't but will be executed. This is
            // important for fullnodes to be able to generate indexing data
            // right now.
            if genesis.effects().events_digest().is_some() {
                store
                    .perpetual_tables
                    .events_2
                    .insert(transaction.digest(), genesis.events())
                    .unwrap();
            }

            // Initialize with migration data if genesis contained migration transactions
            if let Some(migration_transactions) = migration_tx_data {
                // This migration data was validated during the loading into the node (invoked
                // by the caller of this function)
                let txs_data = migration_transactions.txs_data();

                // We iterate over the contents of the genesis checkpoint, that includes all
                // migration transactions execution digest. Thus we cover all transactions that
                // were considered during the creation of the genesis blob.
                for (_, execution_digest) in genesis
                    .checkpoint_contents()
                    .enumerate_transactions(&genesis.checkpoint())
                {
                    let tx_digest = &execution_digest.transaction;
                    // We can skip the genesis transaction and its data because above it was already
                    // stored in the perpetual_tables.
                    if tx_digest == genesis.transaction().digest() {
                        continue;
                    }
                    // Now we can store in the perpetual_tables this migration transaction, together
                    // with its effects, events and created objects.
                    let Some((tx, effects, events)) = txs_data.get(tx_digest) else {
                        panic!("tx digest not found in migrated objects blob");
                    };
                    let transaction = VerifiedTransaction::new_unchecked(tx.clone());
                    let objects = migration_transactions
                        .objects_by_tx_digest(*tx_digest)
                        .expect("the migration data is corrupted");
                    store
                        .bulk_insert_genesis_objects(&objects)
                        .expect("cannot bulk insert migrated objects");
                    store
                        .perpetual_tables
                        .transactions
                        .insert(transaction.digest(), transaction.serializable_ref())
                        .expect("cannot insert migration transaction");
                    store
                        .perpetual_tables
                        .effects
                        .insert(&effects.digest(), effects)
                        .expect("cannot insert migration effects");
                    if effects.events_digest().is_some() {
                        store
                            .perpetual_tables
                            .events_2
                            .insert(transaction.digest(), events)
                            .unwrap();
                    }
                }
            }
        }

        Ok(store)
    }

    /// Open authority store without any operations that require
    /// genesis, such as constructing EpochStartConfiguration
    /// or inserting genesis objects.
    pub fn open_no_genesis(
        perpetual_tables: Arc<AuthorityPerpetualTables>,
        enable_epoch_iota_conservation_check: bool,
        registry: &Registry,
    ) -> IotaResult<Arc<Self>> {
        let store = Arc::new(Self {
            mutex_table: MutexTable::new(NUM_SHARDS),
            perpetual_tables,
            root_state_notify_read: NotifyRead::<
                EpochId,
                (CheckpointSequenceNumber, GlobalStateHash),
            >::new(),
            enable_epoch_iota_conservation_check,
            metrics: AuthorityStoreMetrics::new(registry),
        });
        Ok(store)
    }

    pub fn get_recovery_epoch_at_restart(&self) -> IotaResult<EpochId> {
        self.perpetual_tables.get_recovery_epoch_at_restart()
    }

    pub fn get_effects(
        &self,
        effects_digest: &TransactionEffectsDigest,
    ) -> IotaResult<Option<TransactionEffects>> {
        Ok(self.perpetual_tables.effects.get(effects_digest)?)
    }

    /// Returns true if we have an effects structure for this transaction digest
    pub fn effects_exists(&self, effects_digest: &TransactionEffectsDigest) -> IotaResult<bool> {
        self.perpetual_tables
            .effects
            .contains_key(effects_digest)
            .map_err(|e| e.into())
    }

    pub fn get_events(
        &self,
        digest: &TransactionDigest,
    ) -> Result<Option<TransactionEvents>, TypedStoreError> {
        self.perpetual_tables.events_2.get(digest)
    }

    pub fn multi_get_events(
        &self,
        event_digests: &[TransactionDigest],
    ) -> IotaResult<Vec<Option<TransactionEvents>>> {
        Ok(event_digests
            .iter()
            .map(|digest| self.get_events(digest))
            .collect::<Result<Vec<_>, _>>()?)
    }

    pub fn multi_get_effects<'a>(
        &self,
        effects_digests: impl Iterator<Item = &'a TransactionEffectsDigest>,
    ) -> Result<Vec<Option<TransactionEffects>>, TypedStoreError> {
        self.perpetual_tables.effects.multi_get(effects_digests)
    }

    pub fn get_executed_effects(
        &self,
        tx_digest: &TransactionDigest,
    ) -> Result<Option<TransactionEffects>, TypedStoreError> {
        let effects_digest = self.perpetual_tables.executed_effects.get(tx_digest)?;
        match effects_digest {
            Some(digest) => Ok(self.perpetual_tables.effects.get(&digest)?),
            None => Ok(None),
        }
    }

    /// Given a list of transaction digests, returns a list of the corresponding
    /// effects only if they have been executed. For transactions that have
    /// not been executed, None is returned.
    pub fn multi_get_executed_effects_digests(
        &self,
        digests: &[TransactionDigest],
    ) -> Result<Vec<Option<TransactionEffectsDigest>>, TypedStoreError> {
        self.perpetual_tables.executed_effects.multi_get(digests)
    }

    /// Given a list of transaction digests, returns a list of the corresponding
    /// effects only if they have been executed. For transactions that have
    /// not been executed, None is returned.
    pub fn multi_get_executed_effects(
        &self,
        digests: &[TransactionDigest],
    ) -> Result<Vec<Option<TransactionEffects>>, TypedStoreError> {
        let executed_effects_digests = self.perpetual_tables.executed_effects.multi_get(digests)?;
        let effects = self.multi_get_effects(executed_effects_digests.iter().flatten())?;
        let mut tx_to_effects_map = effects
            .into_iter()
            .flatten()
            .map(|effects| (*effects.transaction_digest(), effects))
            .collect::<HashMap<_, _>>();
        Ok(digests
            .iter()
            .map(|digest| tx_to_effects_map.remove(digest))
            .collect())
    }

    pub fn is_tx_already_executed(&self, digest: &TransactionDigest) -> IotaResult<bool> {
        Ok(self
            .perpetual_tables
            .executed_effects
            .contains_key(digest)?)
    }

    pub fn get_marker_value(
        &self,
        object_id: &ObjectId,
        version: &Version,
        epoch_id: EpochId,
    ) -> IotaResult<Option<MarkerValue>> {
        let object_key = (epoch_id, ObjectKey(*object_id, *version));
        Ok(self
            .perpetual_tables
            .object_per_epoch_marker_table
            .get(&object_key)?)
    }

    pub fn get_latest_marker(
        &self,
        object_id: &ObjectId,
        epoch_id: EpochId,
    ) -> IotaResult<Option<(Version, MarkerValue)>> {
        let marker_entry = self
            .perpetual_tables
            .object_per_epoch_marker_table
            .safe_iter_with_prefix_reversed(&(epoch_id, *object_id))
            .next();
        match marker_entry {
            Some(Ok(((epoch, key), marker))) => {
                // because of the iterator bounds these cannot fail
                assert_eq!(epoch, epoch_id);
                assert_eq!(key.0, *object_id);
                Ok(Some((key.1, marker)))
            }
            Some(Err(e)) => Err(e.into()),
            None => Ok(None),
        }
    }

    /// Returns future containing the state hash for the given epoch
    /// once available
    pub async fn notify_read_root_state_hash(
        &self,
        epoch: EpochId,
    ) -> IotaResult<(CheckpointSequenceNumber, GlobalStateHash)> {
        // We need to register waiters _before_ reading from the database to avoid race
        // conditions
        let registration = self.root_state_notify_read.register_one(&epoch);
        let hash = self.perpetual_tables.root_state_hash_by_epoch.get(&epoch)?;

        let result = match hash {
            // Note that Some() clause also drops registration that is already fulfilled
            Some(ready) => Either::Left(futures::future::ready(ready)),
            None => Either::Right(registration),
        }
        .await;

        Ok(result)
    }

    // Implementation of the corresponding method of `CheckpointCache` trait.
    pub(crate) fn insert_finalized_transactions_perpetual_checkpoints(
        &self,
        digests: &[TransactionDigest],
        epoch: EpochId,
        sequence: CheckpointSequenceNumber,
    ) -> IotaResult {
        let mut batch = self
            .perpetual_tables
            .executed_transactions_to_checkpoint
            .batch();
        batch.insert_batch(
            &self.perpetual_tables.executed_transactions_to_checkpoint,
            digests.iter().map(|d| (*d, (epoch, sequence))),
        )?;
        batch.write()?;
        trace!("Transactions {digests:?} finalized at checkpoint {sequence} epoch {epoch}");
        Ok(())
    }

    // Implementation of the corresponding method of `CheckpointCache` trait.
    pub(crate) fn get_transaction_perpetual_checkpoint(
        &self,
        digest: &TransactionDigest,
    ) -> IotaResult<Option<(EpochId, CheckpointSequenceNumber)>> {
        Ok(self
            .perpetual_tables
            .executed_transactions_to_checkpoint
            .get(digest)?)
    }

    // Implementation of the corresponding method of `CheckpointCache` trait.
    pub(crate) fn multi_get_transactions_perpetual_checkpoints(
        &self,
        digests: &[TransactionDigest],
    ) -> IotaResult<Vec<Option<(EpochId, CheckpointSequenceNumber)>>> {
        Ok(self
            .perpetual_tables
            .executed_transactions_to_checkpoint
            .multi_get(digests)?)
    }

    /// Returns true if there are no objects in the database
    pub fn database_is_empty(&self) -> IotaResult<bool> {
        self.perpetual_tables.database_is_empty()
    }

    /// A function that acquires all locks associated with the objects (in order
    /// to avoid deadlocks).
    fn acquire_locks(&self, input_objects: &[ObjectReference]) -> Vec<MutexGuard> {
        self.mutex_table
            .acquire_locks(input_objects.iter().map(|object_ref| object_ref.digest))
    }

    pub fn object_exists_by_key(
        &self,
        object_id: &ObjectId,
        version: VersionNumber,
    ) -> IotaResult<bool> {
        Ok(self
            .perpetual_tables
            .objects
            .contains_key(&ObjectKey(*object_id, version))?)
    }

    pub fn multi_object_exists_by_key(&self, object_keys: &[ObjectKey]) -> IotaResult<Vec<bool>> {
        Ok(self
            .perpetual_tables
            .objects
            .multi_contains_keys(object_keys.to_vec())?
            .into_iter()
            .collect())
    }

    pub fn multi_get_objects_by_key(
        &self,
        object_keys: &[ObjectKey],
    ) -> Result<Vec<Option<Object>>, IotaError> {
        let wrappers = self
            .perpetual_tables
            .objects
            .multi_get(object_keys.to_vec())?;
        let mut ret = vec![];

        for (idx, w) in wrappers.into_iter().enumerate() {
            ret.push(
                w.map(|object| self.perpetual_tables.object(&object_keys[idx], object))
                    .transpose()?
                    .flatten(),
            );
        }
        Ok(ret)
    }

    /// Get many objects
    pub fn get_objects(&self, objects: &[ObjectId]) -> Result<Vec<Option<Object>>, IotaError> {
        let mut result = Vec::new();
        for id in objects {
            result.push(self.try_get_object(id)?);
        }
        Ok(result)
    }

    pub fn have_deleted_owned_object_at_version_or_after(
        &self,
        object_id: &ObjectId,
        version: VersionNumber,
        epoch_id: EpochId,
    ) -> Result<bool, IotaError> {
        // Find the most recent version of the object that was deleted or wrapped.
        // Return true if the version is >= `version`. Otherwise return false.
        let marker_entry = self
            .perpetual_tables
            .object_per_epoch_marker_table
            .safe_iter_with_prefix_reversed(&(epoch_id, *object_id))
            .next();
        match marker_entry.transpose()? {
            Some(((epoch, key), marker)) => {
                // Make sure object id matches and version is >= `version`
                let object_data_ok = key.0 == *object_id && key.1 >= version;
                // Make sure we don't have a stale epoch for some reason (e.g., a revert)
                let epoch_data_ok = epoch == epoch_id;
                // Make sure the object was deleted or wrapped.
                let mark_data_ok = marker == MarkerValue::OwnedDeleted;
                Ok(object_data_ok && epoch_data_ok && mark_data_ok)
            }
            None => Ok(false),
        }
    }

    // Methods to mutate the store

    /// Insert a genesis object.
    /// TODO: delete this method entirely (still used by authority_tests.rs)
    pub(crate) fn insert_genesis_object(&self, object: Object) -> IotaResult {
        // We only side load objects with a genesis parent transaction.
        debug_assert!(object.previous_transaction == TransactionDigest::GENESIS_MARKER);
        let object_ref = object.object_ref();
        self.insert_genesis_object_direct(object_ref, &object)
    }

    /// Insert an object directly into the store, and also update relevant
    /// tables NOTE: does not handle transaction lock.
    /// This is used to insert genesis objects
    fn insert_genesis_object_direct(
        &self,
        object_ref: ObjectReference,
        object: &Object,
    ) -> IotaResult {
        let mut write_batch = self.perpetual_tables.objects.batch();

        // Genesis objects are produced by the genesis checkpoint (sequence 0).
        let store_object = get_store_object(object.clone(), Some(0));
        write_batch.insert_batch(
            &self.perpetual_tables.objects,
            std::iter::once((ObjectKey::from(object_ref), store_object)),
        )?;

        // Update the index
        if object.single_owner().is_some() {
            // Only initialize live object markers for address owned objects.
            if !object.is_child_object() {
                self.initialize_live_object_markers_impl(&mut write_batch, &[object_ref])?;
            }
        }

        write_batch.write()?;

        Ok(())
    }

    /// This function should only be used for initializing genesis and should
    /// remain private.
    #[instrument(level = "debug", skip_all)]
    pub(crate) fn bulk_insert_genesis_objects(&self, objects: &[Object]) -> IotaResult<()> {
        let mut batch = self.perpetual_tables.objects.batch();
        let ref_and_objects: Vec<_> = objects.iter().map(|o| (o.object_ref(), o)).collect();

        // Genesis objects are produced by the genesis checkpoint (sequence 0).
        batch.insert_batch(
            &self.perpetual_tables.objects,
            ref_and_objects.iter().map(|(oref, o)| {
                (
                    ObjectKey::from(oref),
                    get_store_object((*o).clone(), Some(0)),
                )
            }),
        )?;

        let non_child_object_refs: Vec<_> = ref_and_objects
            .iter()
            .filter(|(_, object)| !object.is_child_object())
            .map(|(oref, _)| *oref)
            .collect();

        self.initialize_live_object_markers_impl(&mut batch, &non_child_object_refs)?;

        batch.write()?;

        Ok(())
    }

    pub fn bulk_insert_live_objects(
        perpetual_db: &AuthorityPerpetualTables,
        live_objects: impl Iterator<Item = LiveObject>,
        expected_sha3_digest: &[u8; 32],
    ) -> IotaResult<()> {
        let mut hasher = Sha3_256::default();
        let mut batch = perpetual_db.objects.batch();
        for live_object in live_objects {
            hasher.update(live_object.object_reference().digest.inner());
            let LiveObject {
                object,
                previous_transaction_checkpoint,
            } = live_object;
            let store_object_wrapper =
                get_store_object(object.clone(), previous_transaction_checkpoint);
            batch.insert_batch(
                &perpetual_db.objects,
                std::iter::once((ObjectKey::from(object.object_ref()), store_object_wrapper)),
            )?;
            if !object.is_child_object() {
                Self::initialize_live_object_markers(
                    &perpetual_db.live_owned_object_markers,
                    &mut batch,
                    &[object.object_ref()],
                )?;
            }
        }
        let sha3_digest = hasher.finalize().digest;
        if *expected_sha3_digest != sha3_digest {
            error!(
                "Sha does not match! expected: {:?}, actual: {:?}",
                expected_sha3_digest, sha3_digest
            );
            return Err(IotaError::from("Sha does not match"));
        }
        batch.write()?;
        Ok(())
    }

    pub fn set_epoch_start_configuration(
        &self,
        epoch_start_configuration: &EpochStartConfiguration,
    ) -> IotaResult {
        self.perpetual_tables
            .set_epoch_start_configuration(epoch_start_configuration)?;
        Ok(())
    }

    pub fn get_epoch_start_configuration(&self) -> IotaResult<Option<EpochStartConfiguration>> {
        Ok(self.perpetual_tables.epoch_start_configuration.get(&())?)
    }

    /// Updates the state resulting from the execution of a certificate.
    ///
    /// Internally it checks that all locks for active inputs are at the correct
    /// version, and then writes objects, certificates, parents and clean up
    /// locks atomically.
    ///
    /// `checkpoint_sequence_number` is stamped onto each newly written object's
    /// `StoreObjectValueV2.previous_transaction_checkpoint` field.
    ///
    /// **Invariant** Every `TransactionOutputs` in `tx_outputs` must belong to
    /// the checkpoint identified by `checkpoint_sequence_number`.
    #[instrument(level = "debug", skip_all)]
    pub fn build_db_batch(
        &self,
        epoch_id: EpochId,
        checkpoint_sequence_number: CheckpointSequenceNumber,
        tx_outputs: &[Arc<TransactionOutputs>],
    ) -> IotaResult<DBBatch> {
        let mut written = Vec::with_capacity(tx_outputs.len());
        for outputs in tx_outputs {
            written.extend(outputs.written.values().cloned());
        }

        let mut write_batch = self.perpetual_tables.transactions.batch();
        for outputs in tx_outputs {
            self.write_one_transaction_outputs(
                &mut write_batch,
                epoch_id,
                checkpoint_sequence_number,
                outputs,
            )?;
        }
        // test crashing before writing the batch
        fail_point!("crash");

        trace!(
            "built batch for committed transactions: {:?}",
            tx_outputs
                .iter()
                .map(|tx| tx.transaction.digest())
                .collect::<Vec<_>>()
        );

        // test crashing before notifying
        fail_point!("crash");

        Ok(write_batch)
    }

    fn write_one_transaction_outputs(
        &self,
        write_batch: &mut DBBatch,
        epoch_id: EpochId,
        checkpoint_sequence_number: CheckpointSequenceNumber,
        tx_outputs: &TransactionOutputs,
    ) -> IotaResult {
        let TransactionOutputs {
            transaction,
            effects,
            markers,
            wrapped,
            deleted,
            written,
            events,
            live_object_markers_to_delete,
            new_live_object_markers_to_init,
            ..
        } = tx_outputs;

        // Store the certificate indexed by transaction digest
        let transaction_digest = transaction.digest();
        write_batch.insert_batch(
            &self.perpetual_tables.transactions,
            iter::once((transaction_digest, transaction.serializable_ref())),
        )?;

        // Add batched writes for objects and locks.
        let effects_digest = effects.digest();

        write_batch.insert_batch(
            &self.perpetual_tables.object_per_epoch_marker_table,
            markers
                .iter()
                .map(|(key, marker_value)| ((epoch_id, *key), *marker_value)),
        )?;

        write_batch.insert_batch(
            &self.perpetual_tables.objects,
            deleted
                .iter()
                .map(|key| (key, StoreObject::Deleted))
                .chain(wrapped.iter().map(|key| (key, StoreObject::Wrapped)))
                .map(|(key, store_object)| (key, StoreObjectWrapper::from(store_object))),
        )?;

        // Insert each output object into the stores
        let new_objects = written.iter().map(|(id, new_object)| {
            let version = new_object.version();
            trace!(?id, ?version, "writing object");
            let store_object =
                get_store_object(new_object.clone(), Some(checkpoint_sequence_number));
            (ObjectKey(*id, version), store_object)
        });

        write_batch.insert_batch(&self.perpetual_tables.objects, new_objects)?;

        // Write events into the new table keyed off of transaction_digest
        if effects.events_digest().is_some() {
            write_batch.insert_batch(
                &self.perpetual_tables.events_2,
                [(transaction_digest, events)],
            )?;
        }

        self.initialize_live_object_markers_impl(write_batch, new_live_object_markers_to_init)?;

        // Note: deletes live object markers for received objects as well (but not for
        // objects that were in `Receiving` arguments which were not received)
        self.delete_live_object_markers(write_batch, live_object_markers_to_delete)?;

        write_batch
            .insert_batch(
                &self.perpetual_tables.effects,
                [(effects_digest, effects.clone())],
            )?
            .insert_batch(
                &self.perpetual_tables.executed_effects,
                [(transaction_digest, effects_digest)],
            )?;

        debug!(effects_digest = ?effects.digest(), "commit_transaction finished");

        Ok(())
    }

    /// Commits transactions only (not effects or other transaction outputs) to
    /// the db. See ExecutionCache::persist_transaction for more info
    pub(crate) fn persist_transaction(&self, tx: &VerifiedExecutableTransaction) -> IotaResult {
        let mut batch = self.perpetual_tables.transactions.batch();
        batch.insert_batch(
            &self.perpetual_tables.transactions,
            [(tx.digest(), tx.clone().into_unsigned().serializable_ref())],
        )?;
        batch.write()?;
        Ok(())
    }

    pub fn acquire_transaction_locks(
        &self,
        epoch_store: &AuthorityPerEpochStore,
        owned_input_objects: &[ObjectReference],
        transaction: VerifiedSignedTransaction,
    ) -> IotaResult {
        let tx_digest = *transaction.digest();
        // Other writers may be attempting to acquire locks on the same objects, so a
        // mutex is required.
        // TODO: replace with optimistic db_transactions (i.e. set lock to tx if none)
        let _mutexes = self.acquire_locks(owned_input_objects);

        trace!(?owned_input_objects, "acquire_transaction_locks");
        let mut locks_to_write = Vec::new();

        let live_object_markers = self
            .perpetual_tables
            .live_owned_object_markers
            .multi_get(owned_input_objects)?;

        let epoch_tables = epoch_store.tables()?;

        let locks = epoch_tables.multi_get_locked_transactions(owned_input_objects)?;

        assert_eq!(locks.len(), live_object_markers.len());

        for (live_marker, lock, obj_ref) in izip!(
            live_object_markers.into_iter(),
            locks.into_iter(),
            owned_input_objects
        ) {
            if live_marker.is_none() {
                // object at that version does not exist
                let latest_live_version =
                    self.get_latest_live_version_for_object_id(obj_ref.object_id)?;
                fp_bail!(
                    UserInputError::ObjectVersionUnavailableForConsumption {
                        provided_obj_ref: *obj_ref,
                        current_version: latest_live_version.version
                    }
                    .into()
                );
            };

            if let Some(previous_tx_digest) = &lock {
                if previous_tx_digest == &tx_digest {
                    // no need to re-write lock
                    continue;
                } else {
                    // TODO: add metrics here
                    info!(prev_tx_digest = ?previous_tx_digest,
                          cur_tx_digest = ?tx_digest,
                          "Cannot acquire lock: conflicting transaction!");
                    return Err(IotaError::ObjectLockConflict {
                        obj_ref: *obj_ref,
                        pending_transaction: *previous_tx_digest,
                    });
                }
            }

            locks_to_write.push((*obj_ref, tx_digest));
        }

        if !locks_to_write.is_empty() {
            trace!(?locks_to_write, "Writing locks");
            epoch_tables.write_transaction_locks(transaction, locks_to_write.into_iter())?;
        }

        Ok(())
    }

    /// Gets ObjectLockInfo that represents state of lock on an object.
    /// Returns UserInputError::ObjectNotFound if cannot find lock record for
    /// this object
    pub(crate) fn get_lock(
        &self,
        obj_ref: ObjectReference,
        epoch_store: &AuthorityPerEpochStore,
    ) -> IotaLockResult {
        if self
            .perpetual_tables
            .live_owned_object_markers
            .get(&obj_ref)?
            .is_none()
        {
            // object at that version does not exist
            return Ok(ObjectLockStatus::LockedAtDifferentVersion {
                locked_ref: self.get_latest_live_version_for_object_id(obj_ref.object_id)?,
            });
        }

        let tables = epoch_store.tables()?;
        if let Some(tx_digest) = tables.get_locked_transaction(&obj_ref)? {
            Ok(ObjectLockStatus::LockedToTx {
                locked_by_tx: tx_digest,
            })
        } else {
            Ok(ObjectLockStatus::Initialized)
        }
    }

    /// Returns UserInputError::ObjectNotFound if no lock records found for this
    /// object.
    pub(crate) fn get_latest_live_version_for_object_id(
        &self,
        object_id: ObjectId,
    ) -> IotaResult<ObjectReference> {
        let mut iterator = self
            .perpetual_tables
            .live_owned_object_markers
            .safe_iter_with_prefix_reversed(&object_id);
        Ok(iterator
            .next()
            .transpose()?
            .filter(|&value| value.0.object_id == object_id)
            .ok_or_else(|| {
                IotaError::from(UserInputError::ObjectNotFound {
                    object_id,
                    version: None,
                })
            })?
            .0)
    }

    /// Checks multiple object locks exist.
    /// Returns UserInputError::ObjectNotFound if cannot find lock record for at
    /// least one of the objects.
    /// Returns UserInputError::ObjectVersionUnavailableForConsumption if at
    /// least one object lock is not initialized     at the given version.
    pub fn check_owned_objects_are_live(&self, objects: &[ObjectReference]) -> IotaResult {
        let live_markers = self
            .perpetual_tables
            .live_owned_object_markers
            .multi_get(objects)?;
        for (live_marker, obj_ref) in live_markers.into_iter().zip(objects) {
            if live_marker.is_none() {
                // object at that version does not exist
                let latest_live_version =
                    self.get_latest_live_version_for_object_id(obj_ref.object_id)?;
                fp_bail!(
                    UserInputError::ObjectVersionUnavailableForConsumption {
                        provided_obj_ref: *obj_ref,
                        current_version: latest_live_version.version
                    }
                    .into()
                );
            }
        }
        Ok(())
    }

    /// Initialize live object markers for a given list of ObjectRefs.
    fn initialize_live_object_markers_impl(
        &self,
        write_batch: &mut DBBatch,
        objects: &[ObjectReference],
    ) -> IotaResult {
        AuthorityStore::initialize_live_object_markers(
            &self.perpetual_tables.live_owned_object_markers,
            write_batch,
            objects,
        )
    }

    pub fn initialize_live_object_markers(
        live_object_marker_table: &DBMap<ObjectReference, ()>,
        write_batch: &mut DBBatch,
        objects: &[ObjectReference],
    ) -> IotaResult {
        trace!(?objects, "initialize_live_object_markers");

        write_batch.insert_batch(
            live_object_marker_table,
            objects.iter().map(|obj_ref| (obj_ref, ())),
        )?;
        Ok(())
    }

    /// Removes locks for a given list of ObjectRefs.
    fn delete_live_object_markers(
        &self,
        write_batch: &mut DBBatch,
        objects: &[ObjectReference],
    ) -> IotaResult {
        trace!(?objects, "delete_live_object_markers");
        write_batch.delete_batch(
            &self.perpetual_tables.live_owned_object_markers,
            objects.iter(),
        )?;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn reset_locks_and_live_markers_for_test(
        &self,
        transactions: &[TransactionDigest],
        objects: &[ObjectReference],
        epoch_store: &AuthorityPerEpochStore,
    ) {
        for tx in transactions {
            epoch_store.delete_signed_transaction_for_test(tx);
            epoch_store.delete_object_locks_for_test(objects);
        }

        let mut batch = self.perpetual_tables.live_owned_object_markers.batch();
        batch
            .delete_batch(
                &self.perpetual_tables.live_owned_object_markers,
                objects.iter(),
            )
            .unwrap();
        batch.write().unwrap();

        let mut batch = self.perpetual_tables.live_owned_object_markers.batch();
        self.initialize_live_object_markers_impl(&mut batch, objects)
            .unwrap();
        batch.write().unwrap();
    }

    /// This function is called at the end of epoch for each transaction that's
    /// executed locally on the validator but didn't make to the last
    /// checkpoint. The effects of the execution is reverted here.
    /// The following things are reverted:
    /// 1. All new object states are deleted.
    /// 2. owner_index table change is reverted.
    ///
    /// NOTE: transaction and effects are intentionally not deleted. It's
    /// possible that if this node is behind, the network will execute the
    /// transaction in a later epoch. In that case, we need to keep it saved
    /// so that when we receive the checkpoint that includes it from state
    /// sync, we are able to execute the checkpoint.
    /// TODO: implement GC for transactions that are no longer needed.
    pub fn revert_state_update(&self, tx_digest: &TransactionDigest) -> IotaResult {
        let Some(effects) = self.get_executed_effects(tx_digest)? else {
            info!("Not reverting {:?} as it was not executed", tx_digest);
            return Ok(());
        };

        info!(?tx_digest, ?effects, "reverting transaction");

        // We should never be reverting shared object transactions.
        assert!(effects.input_shared_objects().is_empty());

        let mut write_batch = self.perpetual_tables.transactions.batch();
        write_batch.delete_batch(
            &self.perpetual_tables.executed_effects,
            iter::once(tx_digest),
        )?;
        if effects.events_digest().is_some() {
            write_batch.delete_batch(&self.perpetual_tables.events_2, [tx_digest])?;
        }

        let tombstones = effects
            .all_tombstones()
            .into_iter()
            .map(|(id, version)| ObjectKey(id, version));
        write_batch.delete_batch(&self.perpetual_tables.objects, tombstones)?;

        let all_new_object_keys = effects
            .all_changed_objects()
            .into_iter()
            .map(|(object_ref, _, _)| ObjectKey(object_ref.object_id, object_ref.version));
        write_batch.delete_batch(&self.perpetual_tables.objects, all_new_object_keys.clone())?;

        let modified_object_keys = effects
            .modified_at_versions()
            .into_iter()
            .map(|(id, version)| ObjectKey(id, version));

        macro_rules! get_objects_and_locks {
            ($object_keys: expr) => {
                self.perpetual_tables
                    .objects
                    .multi_get($object_keys.clone())?
                    .into_iter()
                    .zip($object_keys)
                    .filter_map(|(obj_opt, key)| {
                        let obj = self
                            .perpetual_tables
                            .object(
                                &key,
                                obj_opt.unwrap_or_else(|| {
                                    panic!("Older object version not found: {:?}", key)
                                }),
                            )
                            .expect("Matching indirect object not found")?;

                        if obj.is_immutable() {
                            return None;
                        }

                        let obj_ref = obj.object_ref();
                        Some(obj.is_address_owned().then_some(obj_ref))
                    })
            };
        }

        let old_locks = get_objects_and_locks!(modified_object_keys);
        let new_locks = get_objects_and_locks!(all_new_object_keys);

        let old_locks: Vec<_> = old_locks.flatten().collect();

        // Re-create old live markers.
        self.initialize_live_object_markers_impl(&mut write_batch, &old_locks)?;

        // Delete new live markers
        write_batch.delete_batch(
            &self.perpetual_tables.live_owned_object_markers,
            new_locks.flatten(),
        )?;

        write_batch.write()?;

        Ok(())
    }

    /// Return the object with version less then or eq to the provided seq
    /// number. This is used by indexer to find the correct version of
    /// dynamic field child object. We do not store the version of the child
    /// object, but because of lamport timestamp, we know the child must
    /// have version number less then or eq to the parent.
    pub fn find_object_lt_or_eq_version(
        &self,
        object_id: ObjectId,
        version: Version,
    ) -> IotaResult<Option<Object>> {
        self.perpetual_tables
            .find_object_lt_or_eq_version(object_id, version)
    }

    /// Returns the latest object reference we have for this object_id in the
    /// objects table.
    ///
    /// The method may also return the reference to a deleted object with a
    /// digest of ObjectDigest::deleted() or ObjectDigest::wrapped() and
    /// lamport version of a transaction that deleted the object.
    /// Note that a deleted object may re-appear if the deletion was the result
    /// of the object being wrapped in another object.
    ///
    /// If no entry for the object_id is found, return None.
    pub fn get_latest_object_ref_or_tombstone(
        &self,
        object_id: ObjectId,
    ) -> Result<Option<ObjectReference>, IotaError> {
        self.perpetual_tables
            .get_latest_object_ref_or_tombstone(object_id)
    }

    /// Returns the latest object reference if and only if the object is still
    /// live (i.e. it does not return tombstones)
    pub fn get_latest_object_ref_if_alive(
        &self,
        object_id: ObjectId,
    ) -> Result<Option<ObjectReference>, IotaError> {
        match self.get_latest_object_ref_or_tombstone(object_id)? {
            Some(objref) if objref.digest.is_alive() => Ok(Some(objref)),
            _ => Ok(None),
        }
    }

    /// Returns the latest object we have for this object_id in the objects
    /// table.
    ///
    /// If no entry for the object_id is found, return None.
    pub fn get_latest_object_or_tombstone(
        &self,
        object_id: ObjectId,
    ) -> Result<Option<(ObjectKey, ObjectOrTombstone)>, IotaError> {
        let Some((object_key, store_object)) = self
            .perpetual_tables
            .get_latest_object_or_tombstone(object_id)?
        else {
            return Ok(None);
        };

        if let Some(object_ref) = self
            .perpetual_tables
            .tombstone_reference(&object_key, &store_object)?
        {
            return Ok(Some((object_key, ObjectOrTombstone::Tombstone(object_ref))));
        }

        let object = self
            .perpetual_tables
            .object(&object_key, store_object)?
            .expect("Non tombstone store object could not be converted to object");

        Ok(Some((object_key, ObjectOrTombstone::Object(object))))
    }

    pub fn insert_transaction_and_effects(
        &self,
        transaction: &VerifiedTransaction,
        transaction_effects: &TransactionEffects,
    ) -> Result<(), TypedStoreError> {
        let mut write_batch = self.perpetual_tables.transactions.batch();
        write_batch
            .insert_batch(
                &self.perpetual_tables.transactions,
                [(transaction.digest(), transaction.serializable_ref())],
            )?
            .insert_batch(
                &self.perpetual_tables.effects,
                [(transaction_effects.digest(), transaction_effects)],
            )?;

        write_batch.write()?;
        Ok(())
    }

    pub fn multi_insert_transaction_and_effects<'a>(
        &self,
        transactions: impl Iterator<Item = &'a VerifiedExecutionData>,
    ) -> Result<(), TypedStoreError> {
        let mut write_batch = self.perpetual_tables.transactions.batch();
        for tx in transactions {
            write_batch
                .insert_batch(
                    &self.perpetual_tables.transactions,
                    [(tx.transaction.digest(), tx.transaction.serializable_ref())],
                )?
                .insert_batch(
                    &self.perpetual_tables.effects,
                    [(tx.effects.digest(), &tx.effects)],
                )?;
        }

        write_batch.write()?;
        Ok(())
    }

    pub fn multi_get_transaction_blocks(
        &self,
        tx_digests: &[TransactionDigest],
    ) -> Result<Vec<Option<VerifiedTransaction>>, TypedStoreError> {
        self.perpetual_tables
            .transactions
            .multi_get(tx_digests)
            .map(|v| v.into_iter().map(|v| v.map(|v| v.into())).collect())
    }

    pub fn get_transaction_block(
        &self,
        tx_digest: &TransactionDigest,
    ) -> Result<Option<VerifiedTransaction>, TypedStoreError> {
        self.perpetual_tables
            .transactions
            .get(tx_digest)
            .map(|v| v.map(|v| v.into()))
    }

    /// This function reads the DB directly to get the system state object.
    /// If reconfiguration is happening at the same time, there is no guarantee
    /// whether we would be getting the old or the new system state object.
    /// Hence this function should only be called during RPC reads where data
    /// race is not a major concern. In general we should avoid this as much
    /// as possible. If the intent is for testing, you can use
    /// AuthorityState:: get_iota_system_state_object_for_testing.
    pub fn get_iota_system_state_object_unsafe(&self) -> IotaResult<IotaSystemState> {
        get_iota_system_state(self.perpetual_tables.as_ref())
    }

    pub fn expensive_check_iota_conservation<T>(
        self: &Arc<Self>,
        type_layout_store: T,
        old_epoch_store: &AuthorityPerEpochStore,
        epoch_supply_change: Option<i64>,
    ) -> IotaResult
    where
        T: TypeLayoutStore + Send + Copy,
    {
        if !self.enable_epoch_iota_conservation_check {
            return Ok(());
        }

        let executor = old_epoch_store.executor();
        info!("Starting IOTA conservation check. This may take a while..");
        let cur_time = Instant::now();
        let mut pending_objects = vec![];
        let mut count = 0;
        let mut size = 0;
        let (mut total_iota, mut total_storage_rebate) = thread::scope(|s| {
            let pending_tasks = FuturesUnordered::new();
            for o in self.iter_live_object_set() {
                let object = o.object;
                size += object.object_size_for_gas_metering();
                count += 1;
                pending_objects.push(object);
                if count % 1_000_000 == 0 {
                    let mut task_objects = vec![];
                    mem::swap(&mut pending_objects, &mut task_objects);
                    pending_tasks.push(s.spawn(move || {
                        let mut layout_resolver =
                            executor.type_layout_resolver(Box::new(type_layout_store));
                        let mut total_storage_rebate = 0;
                        let mut total_iota = 0;
                        for object in task_objects {
                            total_storage_rebate += object.storage_rebate;
                            // get_total_iota includes storage rebate, however all storage rebate is
                            // also stored in the storage fund, so we need to subtract it here.
                            total_iota += object.get_total_iota(layout_resolver.as_mut()).unwrap()
                                - object.storage_rebate;
                        }
                        if count % 50_000_000 == 0 {
                            info!("Processed {} objects", count);
                        }
                        (total_iota, total_storage_rebate)
                    }));
                }
            }
            pending_tasks.into_iter().fold((0, 0), |init, result| {
                let result = result.join().unwrap();
                (init.0 + result.0, init.1 + result.1)
            })
        });
        let mut layout_resolver = executor.type_layout_resolver(Box::new(type_layout_store));
        for object in pending_objects {
            total_storage_rebate += object.storage_rebate;
            total_iota +=
                object.get_total_iota(layout_resolver.as_mut()).unwrap() - object.storage_rebate;
        }
        info!(
            "Scanned {} live objects, took {:?}",
            count,
            cur_time.elapsed()
        );
        self.metrics
            .iota_conservation_live_object_count
            .set(count as i64);
        self.metrics
            .iota_conservation_live_object_size
            .set(size as i64);
        self.metrics
            .iota_conservation_check_latency
            .set(cur_time.elapsed().as_secs() as i64);

        // It is safe to call this function because we are in the middle of
        // reconfiguration.
        let system_state: IotaSystemStateSummaryV2 = self
            .get_iota_system_state_object_unsafe()
            .expect("Reading iota system state object cannot fail")
            .into_iota_system_state_summary()
            .try_into()?;
        let storage_fund_balance = system_state.storage_fund_total_object_storage_rebates;
        info!(
            "Total IOTA amount in the network: {}, storage fund balance: {}, total storage rebate: {} at beginning of epoch {}",
            total_iota, storage_fund_balance, total_storage_rebate, system_state.epoch
        );

        let imbalance = (storage_fund_balance as i64) - (total_storage_rebate as i64);
        self.metrics
            .iota_conservation_storage_fund
            .set(storage_fund_balance as i64);
        self.metrics
            .iota_conservation_storage_fund_imbalance
            .set(imbalance);
        self.metrics
            .iota_conservation_imbalance
            .set((total_iota as i128 - system_state.iota_total_supply as i128) as i64);

        if let Some(expected_imbalance) = self
            .perpetual_tables
            .expected_storage_fund_imbalance
            .get(&())
            .map_err(|err| {
                IotaError::from(
                    format!("failed to read expected storage fund imbalance: {err}").as_str(),
                )
            })?
        {
            fp_ensure!(
                imbalance == expected_imbalance,
                IotaError::from(
                    format!(
                        "Inconsistent state detected at epoch {}: total storage rebate: {}, storage fund balance: {}, expected imbalance: {}",
                        system_state.epoch, total_storage_rebate, storage_fund_balance, expected_imbalance
                    ).as_str()
                )
            );
        } else {
            self.perpetual_tables
                .expected_storage_fund_imbalance
                .insert(&(), &imbalance)
                .map_err(|err| {
                    IotaError::from(
                        format!("failed to write expected storage fund imbalance: {err}").as_str(),
                    )
                })?;
        }

        let total_supply = self
            .perpetual_tables
            .total_iota_supply
            .get(&())
            .map_err(|err| {
                IotaError::from(format!("failed to read total iota supply: {err}").as_str())
            })?;

        match total_supply.zip(epoch_supply_change) {
            // Only execute the check if both are set and the supply value was set in the last
            // epoch. We have to assume the supply changes every epoch and therefore we
            // cannot run the check with a supply value from any epoch earlier than the
            // last one. This can happen if the check was disabled for some time.
            Some((old_supply, epoch_supply_change))
                if old_supply.last_check_epoch + 1 == old_epoch_store.epoch() =>
            {
                let expected_new_supply = if epoch_supply_change >= 0 {
                    old_supply
                        .total_supply
                        .checked_add(epoch_supply_change.unsigned_abs())
                        .ok_or_else(|| {
                            IotaError::from(
                                format!(
                                    "Inconsistent state detected at epoch {}: old supply {} + supply change {} overflowed",
                                    system_state.epoch, old_supply.total_supply, epoch_supply_change
                                ).as_str())
                        })?
                } else {
                    old_supply.total_supply.checked_sub(epoch_supply_change.unsigned_abs()).ok_or_else(|| {
                        IotaError::from(
                            format!(
                                "Inconsistent state detected at epoch {}: old supply {} - supply change {} underflowed",
                                system_state.epoch, old_supply.total_supply, epoch_supply_change
                            ).as_str())
                    })?
                };

                fp_ensure!(
                    total_iota == expected_new_supply,
                    IotaError::from(
                        format!(
                            "Inconsistent state detected at epoch {}: total iota: {}, expecting {}",
                            system_state.epoch, total_iota, expected_new_supply
                        )
                        .as_str()
                    )
                );

                let new_supply = TotalIotaSupplyCheck {
                    total_supply: expected_new_supply,
                    last_check_epoch: old_epoch_store.epoch(),
                };

                self.perpetual_tables
                    .total_iota_supply
                    .insert(&(), &new_supply)
                    .map_err(|err| {
                        IotaError::from(
                            format!("failed to write total iota supply: {err}").as_str(),
                        )
                    })?;
            }
            // If either one is None or if the last value is from an older epoch,
            // we update the value in the table since we're at genesis and cannot execute the check.
            _ => {
                info!("Skipping total supply check");

                let supply = TotalIotaSupplyCheck {
                    total_supply: total_iota,
                    last_check_epoch: old_epoch_store.epoch(),
                };

                self.perpetual_tables
                    .total_iota_supply
                    .insert(&(), &supply)
                    .map_err(|err| {
                        IotaError::from(
                            format!("failed to write total iota supply: {err}").as_str(),
                        )
                    })?;

                return Ok(());
            }
        };

        Ok(())
    }

    pub async fn prune_objects_and_compact_for_testing(
        &self,
        checkpoint_store: &Arc<CheckpointStore>,
        grpc_indexes_store: Option<&GrpcIndexesStore>,
    ) {
        let pruning_config = AuthorityStorePruningConfig {
            num_epochs_to_retain: 0,
            ..Default::default()
        };
        let _ = AuthorityStorePruner::prune_objects_for_eligible_epochs(
            &self.perpetual_tables,
            checkpoint_store,
            grpc_indexes_store,
            None,
            pruning_config,
            AuthorityStorePruningMetrics::new_for_test(),
            EPOCH_DURATION_MS_FOR_TESTING,
            None,
        )
        .await;
        let _ = AuthorityStorePruner::compact(&self.perpetual_tables);
    }

    #[cfg(test)]
    pub async fn prune_objects_immediately_for_testing(
        &self,
        transaction_effects: Vec<TransactionEffects>,
    ) -> anyhow::Result<()> {
        let mut wb = self.perpetual_tables.objects.batch();

        let mut object_keys_to_prune = vec![];
        for effects in &transaction_effects {
            for (object_id, seq_number) in effects.modified_at_versions() {
                info!("Pruning object {} version {:?}", object_id, seq_number);
                object_keys_to_prune.push(ObjectKey(object_id, seq_number));
            }
        }

        wb.delete_batch(&self.perpetual_tables.objects, object_keys_to_prune)?;
        wb.write()?;
        Ok(())
    }

    // Counts the number of versions exist in object store for `object_id`. This
    // includes tombstone.
    #[cfg(msim)]
    pub fn count_object_versions(&self, object_id: ObjectId) -> usize {
        self.perpetual_tables
            .objects
            .safe_iter_with_prefix(&object_id)
            .collect::<Result<Vec<_>, _>>()
            .unwrap()
            .len()
    }
}

impl GlobalStateHashStore for AuthorityStore {
    fn get_root_state_hash_for_epoch(
        &self,
        epoch: EpochId,
    ) -> IotaResult<Option<(CheckpointSequenceNumber, GlobalStateHash)>> {
        self.perpetual_tables
            .root_state_hash_by_epoch
            .get(&epoch)
            .map_err(Into::into)
    }

    fn get_root_state_hash_for_highest_epoch(
        &self,
    ) -> IotaResult<Option<(EpochId, (CheckpointSequenceNumber, GlobalStateHash))>> {
        Ok(self
            .perpetual_tables
            .root_state_hash_by_epoch
            .safe_range_iter_reversed(..)
            .next()
            .transpose()?)
    }

    fn insert_state_hash_for_epoch(
        &self,
        epoch: EpochId,
        last_checkpoint_of_epoch: &CheckpointSequenceNumber,
        acc: &GlobalStateHash,
    ) -> IotaResult {
        self.perpetual_tables
            .root_state_hash_by_epoch
            .insert(&epoch, &(*last_checkpoint_of_epoch, acc.clone()))?;
        self.root_state_notify_read
            .notify(&epoch, &(*last_checkpoint_of_epoch, acc.clone()));

        Ok(())
    }

    fn iter_live_object_set(&self) -> Box<dyn Iterator<Item = LiveObject> + '_> {
        Box::new(self.perpetual_tables.iter_live_object_set())
    }
}

impl ObjectStore for AuthorityStore {
    /// Read an object and return it, or Ok(None) if the object was not found.
    fn try_get_object(
        &self,
        object_id: &ObjectId,
    ) -> Result<Option<Object>, iota_types::storage::error::Error> {
        self.perpetual_tables.as_ref().try_get_object(object_id)
    }

    fn try_get_object_by_key(
        &self,
        object_id: &ObjectId,
        version: VersionNumber,
    ) -> Result<Option<Object>, iota_types::storage::error::Error> {
        self.perpetual_tables
            .try_get_object_by_key(object_id, version)
    }
}

/// A wrapper to make Orphan Rule happy
pub struct ResolverWrapper {
    pub resolver: Arc<dyn BackingPackageStore + Send + Sync>,
    pub metrics: Arc<ResolverMetrics>,
}

impl ResolverWrapper {
    pub fn new(
        resolver: Arc<dyn BackingPackageStore + Send + Sync>,
        metrics: Arc<ResolverMetrics>,
    ) -> Self {
        metrics.module_cache_size.set(0);
        ResolverWrapper { resolver, metrics }
    }

    fn inc_cache_size_gauge(&self) {
        // reset the gauge after a restart of the cache
        let current = self.metrics.module_cache_size.get();
        self.metrics.module_cache_size.set(current + 1);
    }
}

impl ModuleResolver for ResolverWrapper {
    type Error = IotaError;
    fn get_module(&self, module_id: &ModuleId) -> Result<Option<Vec<u8>>, Self::Error> {
        self.inc_cache_size_gauge();
        get_module(&*self.resolver, module_id)
    }
}

pub enum UpdateType {
    Transaction(TransactionEffectsDigest),
    Genesis,
}

pub type IotaLockResult = IotaResult<ObjectLockStatus>;

#[derive(Debug, PartialEq, Eq)]
pub enum ObjectLockStatus {
    Initialized,
    LockedToTx { locked_by_tx: LockDetails }, // no need to use wrapper, not stored or serialized
    LockedAtDifferentVersion { locked_ref: ObjectReference },
}