sui_adapter_latest/

temporary_store.rs

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

use crate::gas_charger::{GasCharger, PaymentLocation};
use mysten_common::{ZipDebugEqIteratorExt, debug_fatal};
use mysten_metrics::monitored_scope;
use parking_lot::RwLock;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use sui_protocol_config::ProtocolConfig;
use sui_types::accumulator_event::AccumulatorEvent;
use sui_types::accumulator_root::AccumulatorObjId;
use sui_types::base_types::VersionDigest;
use sui_types::committee::EpochId;
use sui_types::deny_list_v2::check_coin_deny_list_v2_during_execution;
use sui_types::effects::{
    AccumulatorOperation, AccumulatorValue, AccumulatorWriteV1, TransactionEffects,
    TransactionEvents,
};
use sui_types::execution::{
    DynamicallyLoadedObjectMetadata, ExecutionResults, ExecutionResultsV2, SharedInput,
};
use sui_types::execution_status::{ExecutionErrorKind, ExecutionStatus};
use sui_types::inner_temporary_store::InnerTemporaryStore;
use sui_types::layout_resolver::LayoutResolver;
use sui_types::object::Data;
use sui_types::storage::{BackingStore, DenyListResult, PackageObject};
use sui_types::sui_system_state::{AdvanceEpochParams, get_sui_system_state_wrapper};
use sui_types::{
    SUI_DENY_LIST_OBJECT_ID,
    base_types::{ObjectID, ObjectRef, SequenceNumber, SuiAddress, TransactionDigest},
    effects::EffectsObjectChange,
    error::{ExecutionError, SuiResult},
    gas::GasCostSummary,
    object::Object,
    object::Owner,
    storage::{BackingPackageStore, ChildObjectResolver, ParentSync, Storage},
    transaction::InputObjects,
};
use sui_types::{SUI_SYSTEM_STATE_OBJECT_ID, TypeTag, is_system_package};

pub struct TemporaryStore<'backing> {
    // The backing store for retrieving Move packages onchain.
    // When executing a Move call, the dependent packages are not going to be
    // in the input objects. They will be fetched from the backing store.
    // Also used for fetching the backing parent_sync to get the last known version for wrapped
    // objects
    store: &'backing dyn BackingStore,
    tx_digest: TransactionDigest,
    input_objects: BTreeMap<ObjectID, Object>,

    /// Store the original versions of the non-exclusive write inputs, in order to detect
    /// mutations (which are illegal, but not prevented by the type system).
    non_exclusive_input_original_versions: BTreeMap<ObjectID, Object>,

    stream_ended_consensus_objects: BTreeMap<ObjectID, SequenceNumber /* start_version */>,
    /// The version to assign to all objects written by the transaction using this store.
    lamport_timestamp: SequenceNumber,
    /// Inputs that will be mutated by the transaction. Does not include NonExclusiveWrite inputs,
    /// which can be taken as `&mut T` but cannot be directly mutated.
    mutable_input_refs: BTreeMap<ObjectID, (VersionDigest, Owner)>,
    execution_results: ExecutionResultsV2,
    /// Objects that were loaded during execution (dynamic fields + received objects).
    loaded_runtime_objects: BTreeMap<ObjectID, DynamicallyLoadedObjectMetadata>,
    /// A map from wrapped object to its container. Used during expensive invariant checks.
    wrapped_object_containers: BTreeMap<ObjectID, ObjectID>,
    protocol_config: &'backing ProtocolConfig,

    /// Every package that was loaded from DB store during execution.
    /// These packages were not previously loaded into the temporary store.
    runtime_packages_loaded_from_db: RwLock<BTreeMap<ObjectID, PackageObject>>,

    /// The set of objects that we may receive during execution. Not guaranteed to receive all, or
    /// any of the objects referenced in this set.
    receiving_objects: Vec<ObjectRef>,

    /// The set of all generated object IDs from the object runtime during the transaction. This includes any
    /// created-and-then-deleted objects in addition to any `new_ids` which contains only the set
    /// of created (but not deleted) IDs in the transaction.
    generated_runtime_ids: BTreeSet<ObjectID>,

    // TODO: Now that we track epoch here, there are a few places we don't need to pass it around.
    /// The current epoch.
    cur_epoch: EpochId,

    /// The set of per-epoch config objects that were loaded during execution, and are not in the
    /// input objects. This allows us to commit them to the effects.
    loaded_per_epoch_config_objects: RwLock<BTreeSet<ObjectID>>,

    /// Index ranges into `execution_results.accumulator_events` for events emitted from PTB
    /// (Move) execution. Each `record_execution_results` call appends a contiguous range
    /// bracketing the merge. Any index outside these ranges was emitted by the runtime outside
    /// of PTB execution (currently only `gas_charger`'s `add_accumulator_event`). Consumed by
    /// `check_sui_address_balance_changes` inside `run_conservation_checks` to gate
    /// non-PTB-emitted events behind input reservations. Cleared on `drop_writes` since the
    /// underlying events are also cleared. A `Vec` is sufficient because real transactions run
    /// the PTB at most a handful of times.
    ptb_emitted_accumulator_event_ranges: Vec<std::ops::Range<usize>>,
}

impl<'backing> TemporaryStore<'backing> {
    /// Creates a new store associated with an authority store, and populates it with
    /// initial objects.
    pub fn new(
        store: &'backing dyn BackingStore,
        input_objects: InputObjects,
        receiving_objects: Vec<ObjectRef>,
        tx_digest: TransactionDigest,
        protocol_config: &'backing ProtocolConfig,
        cur_epoch: EpochId,
    ) -> Self {
        let mutable_input_refs = input_objects.exclusive_mutable_inputs();
        let non_exclusive_input_original_versions = input_objects.non_exclusive_input_objects();

        let lamport_timestamp = input_objects.lamport_timestamp(&receiving_objects);
        let stream_ended_consensus_objects = input_objects.consensus_stream_ended_objects();
        let objects = input_objects.into_object_map();
        #[cfg(debug_assertions)]
        {
            // Ensure that input objects and receiving objects must not overlap.
            assert!(
                objects
                    .keys()
                    .collect::<HashSet<_>>()
                    .intersection(
                        &receiving_objects
                            .iter()
                            .map(|oref| &oref.0)
                            .collect::<HashSet<_>>()
                    )
                    .next()
                    .is_none()
            );
        }
        Self {
            store,
            tx_digest,
            input_objects: objects,
            non_exclusive_input_original_versions,
            stream_ended_consensus_objects,
            lamport_timestamp,
            mutable_input_refs,
            execution_results: ExecutionResultsV2::default(),
            protocol_config,
            loaded_runtime_objects: BTreeMap::new(),
            wrapped_object_containers: BTreeMap::new(),
            runtime_packages_loaded_from_db: RwLock::new(BTreeMap::new()),
            receiving_objects,
            generated_runtime_ids: BTreeSet::new(),
            cur_epoch,
            loaded_per_epoch_config_objects: RwLock::new(BTreeSet::new()),
            ptb_emitted_accumulator_event_ranges: Vec::new(),
        }
    }

    // Helpers to access private fields
    pub fn objects(&self) -> &BTreeMap<ObjectID, Object> {
        &self.input_objects
    }

    pub fn update_object_version_and_prev_tx(&mut self) {
        self.execution_results.update_version_and_previous_tx(
            self.lamport_timestamp,
            self.tx_digest,
            &self.input_objects,
            self.protocol_config.reshare_at_same_initial_version(),
        );

        #[cfg(debug_assertions)]
        {
            self.check_invariants();
        }
    }

    fn calculate_accumulator_running_max_withdraws(&self) -> BTreeMap<AccumulatorObjId, u128> {
        let mut running_net_withdraws: BTreeMap<AccumulatorObjId, i128> = BTreeMap::new();
        let mut running_max_withdraws: BTreeMap<AccumulatorObjId, u128> = BTreeMap::new();
        for event in &self.execution_results.accumulator_events {
            match &event.write.value {
                AccumulatorValue::Integer(amount) => match event.write.operation {
                    AccumulatorOperation::Split => {
                        let entry = running_net_withdraws
                            .entry(event.accumulator_obj)
                            .or_default();
                        *entry += *amount as i128;
                        if *entry > 0 {
                            let max_entry = running_max_withdraws
                                .entry(event.accumulator_obj)
                                .or_default();
                            *max_entry = (*max_entry).max(*entry as u128);
                        }
                    }
                    AccumulatorOperation::Merge => {
                        let entry = running_net_withdraws
                            .entry(event.accumulator_obj)
                            .or_default();
                        *entry -= *amount as i128;
                    }
                },
                AccumulatorValue::IntegerTuple(_, _) | AccumulatorValue::EventDigest(_) => {}
            }
        }
        running_max_withdraws
    }

    /// Ensure that there is one entry for each accumulator object in the accumulator events.
    fn merge_accumulator_events(&mut self) {
        self.execution_results.accumulator_events = self
            .execution_results
            .accumulator_events
            .iter()
            .fold(
                BTreeMap::<AccumulatorObjId, Vec<AccumulatorWriteV1>>::new(),
                |mut map, event| {
                    map.entry(event.accumulator_obj)
                        .or_default()
                        .push(event.write.clone());
                    map
                },
            )
            .into_iter()
            .map(|(obj_id, writes)| {
                AccumulatorEvent::new(obj_id, AccumulatorWriteV1::merge(writes))
            })
            .collect();
    }

    /// Break up the structure and return its internal stores (objects, active_inputs, written, deleted)
    pub fn into_inner(
        self,
        accumulator_running_max_withdraws: BTreeMap<AccumulatorObjId, u128>,
    ) -> InnerTemporaryStore {
        let results = self.execution_results;
        InnerTemporaryStore {
            input_objects: self.input_objects,
            stream_ended_consensus_objects: self.stream_ended_consensus_objects,
            mutable_inputs: self.mutable_input_refs,
            written: results.written_objects,
            events: TransactionEvents {
                data: results.user_events,
            },
            accumulator_events: results.accumulator_events,
            loaded_runtime_objects: self.loaded_runtime_objects,
            runtime_packages_loaded_from_db: self.runtime_packages_loaded_from_db.into_inner(),
            lamport_version: self.lamport_timestamp,
            binary_config: self.protocol_config.binary_config(None),
            accumulator_running_max_withdraws,
        }
    }

    /// For every object from active_inputs (i.e. all mutable objects), if they are not
    /// mutated during the transaction execution, force mutating them by incrementing the
    /// sequence number. This is required to achieve safety.
    pub(crate) fn ensure_active_inputs_mutated(&mut self) {
        let mut to_be_updated = vec![];
        // Note: we do not mutate input objects if they are non-exclusive write
        for id in self.mutable_input_refs.keys() {
            if !self.execution_results.modified_objects.contains(id) {
                // We cannot update here but have to push to `to_be_updated` and update later
                // because the for loop is holding a reference to `self`, and calling
                // `self.mutate_input_object` requires a mutable reference to `self`.
                to_be_updated.push(self.input_objects[id].clone());
            }
        }
        for object in to_be_updated {
            // The object must be mutated as it was present in the input objects
            self.mutate_input_object(object.clone());
        }
    }

    fn get_object_changes(&self) -> BTreeMap<ObjectID, EffectsObjectChange> {
        let results = &self.execution_results;
        let all_ids = results
            .created_object_ids
            .iter()
            .chain(&results.deleted_object_ids)
            .chain(&results.modified_objects)
            .chain(results.written_objects.keys())
            .collect::<BTreeSet<_>>();
        all_ids
            .into_iter()
            .map(|id| {
                (
                    *id,
                    EffectsObjectChange::new(
                        self.get_object_modified_at(id)
                            .map(|metadata| ((metadata.version, metadata.digest), metadata.owner)),
                        results.written_objects.get(id),
                        results.created_object_ids.contains(id),
                        results.deleted_object_ids.contains(id),
                    ),
                )
            })
            .chain(results.accumulator_events.iter().cloned().map(
                |AccumulatorEvent {
                     accumulator_obj,
                     write,
                 }| {
                    (
                        *accumulator_obj.inner(),
                        EffectsObjectChange::new_from_accumulator_write(write),
                    )
                },
            ))
            .collect()
    }

    pub fn into_effects(
        mut self,
        shared_object_refs: Vec<SharedInput>,
        transaction_digest: &TransactionDigest,
        mut transaction_dependencies: BTreeSet<TransactionDigest>,
        gas_cost_summary: GasCostSummary,
        status: ExecutionStatus,
        gas_charger: &mut GasCharger,
        epoch: EpochId,
    ) -> (InnerTemporaryStore, TransactionEffects) {
        self.update_object_version_and_prev_tx();
        // This must happens before merge_accumulator_events.
        let accumulator_running_max_withdraws = self.calculate_accumulator_running_max_withdraws();
        self.merge_accumulator_events();

        // Regardless of execution status (including aborts), we insert the previous transaction
        // for any successfully received objects during the transaction.
        for (id, expected_version, expected_digest) in &self.receiving_objects {
            // If the receiving object is in the loaded runtime objects, then that means that it
            // was actually successfully loaded (so existed, and there was authenticated mutable
            // access to it). So we insert the previous transaction as a dependency.
            if let Some(obj_meta) = self.loaded_runtime_objects.get(id) {
                // Check that the expected version, digest, and owner match the loaded version,
                // digest, and owner. If they don't then don't register a dependency.
                // This is because this could be "spoofed" by loading a dynamic object field.
                let loaded_via_receive = obj_meta.version == *expected_version
                    && obj_meta.digest == *expected_digest
                    && obj_meta.owner.is_address_owned();
                if loaded_via_receive {
                    transaction_dependencies.insert(obj_meta.previous_transaction);
                }
            }
        }

        assert!(self.protocol_config.enable_effects_v2());

        // In the case of special transactions that don't require a gas object,
        // we don't really care about the effects to gas, just use the input for it.
        // Gas coins are guaranteed to be at least size 1 and if more than 1
        // the first coin is where all the others are merged.
        let gas_coin = gas_charger
            .gas_payment_amount()
            .and_then(|gp| match gp.location {
                PaymentLocation::Coin(coin_id) => Some(coin_id),
                PaymentLocation::AddressBalance(_) => None,
            });

        let object_changes = self.get_object_changes();

        let lamport_version = self.lamport_timestamp;
        // TODO: Cleanup this clone. Potentially add unchanged_shraed_objects directly to InnerTempStore.
        let loaded_per_epoch_config_objects = self.loaded_per_epoch_config_objects.read().clone();
        let inner = self.into_inner(accumulator_running_max_withdraws);

        let effects = TransactionEffects::new_from_execution_v2(
            status,
            epoch,
            gas_cost_summary,
            // TODO: Provide the list of read-only shared objects directly.
            shared_object_refs,
            loaded_per_epoch_config_objects,
            *transaction_digest,
            lamport_version,
            object_changes,
            gas_coin,
            if inner.events.data.is_empty() {
                None
            } else {
                Some(inner.events.digest())
            },
            transaction_dependencies.into_iter().collect(),
        );

        (inner, effects)
    }

    /// An internal check of the invariants (will only fire in debug)
    #[cfg(debug_assertions)]
    fn check_invariants(&self) {
        // Check not both deleted and written
        debug_assert!(
            {
                self.execution_results
                    .written_objects
                    .keys()
                    .all(|id| !self.execution_results.deleted_object_ids.contains(id))
            },
            "Object both written and deleted."
        );

        // Check all mutable inputs are modified
        debug_assert!(
            {
                self.mutable_input_refs
                    .keys()
                    .all(|id| self.execution_results.modified_objects.contains(id))
            },
            "Mutable input not modified."
        );

        debug_assert!(
            {
                self.execution_results
                    .written_objects
                    .values()
                    .all(|obj| obj.previous_transaction == self.tx_digest)
            },
            "Object previous transaction not properly set",
        );
    }

    /// Mutate a mutable input object. This is used to mutate input objects outside of PT execution.
    pub fn mutate_input_object(&mut self, object: Object) {
        let id = object.id();
        debug_assert!(self.input_objects.contains_key(&id));
        debug_assert!(!object.is_immutable());
        self.execution_results.modified_objects.insert(id);
        self.execution_results.written_objects.insert(id, object);
    }

    pub fn mutate_new_or_input_object(&mut self, object: Object) {
        let id = object.id();
        debug_assert!(!object.is_immutable());
        if self.input_objects.contains_key(&id) {
            self.execution_results.modified_objects.insert(id);
        }
        self.execution_results.written_objects.insert(id, object);
    }

    /// Mutate a child object outside of PT. This should be used extremely rarely.
    /// Currently it's only used by advance_epoch_safe_mode because it's all native
    /// without PT. This should almost never be used otherwise.
    pub fn mutate_child_object(&mut self, old_object: Object, new_object: Object) {
        let id = new_object.id();
        let old_ref = old_object.compute_object_reference();
        debug_assert_eq!(old_ref.0, id);
        self.loaded_runtime_objects.insert(
            id,
            DynamicallyLoadedObjectMetadata {
                version: old_ref.1,
                digest: old_ref.2,
                owner: old_object.owner.clone(),
                storage_rebate: old_object.storage_rebate,
                previous_transaction: old_object.previous_transaction,
            },
        );
        self.execution_results.modified_objects.insert(id);
        self.execution_results
            .written_objects
            .insert(id, new_object);
    }

    /// Upgrade system package during epoch change. This requires special treatment
    /// since the system package to be upgraded is not in the input objects.
    /// We could probably fix above to make it less special.
    pub fn upgrade_system_package(&mut self, package: Object) {
        let id = package.id();
        assert!(package.is_package() && is_system_package(id));
        self.execution_results.modified_objects.insert(id);
        self.execution_results.written_objects.insert(id, package);
    }

    /// Crate a new objcet. This is used to create objects outside of PT execution.
    pub fn create_object(&mut self, object: Object) {
        // Created mutable objects' versions are set to the store's lamport timestamp when it is
        // committed to effects. Creating an object at a non-zero version risks violating the
        // lamport timestamp invariant (that a transaction's lamport timestamp is strictly greater
        // than all versions witnessed by the transaction).
        debug_assert!(
            object.is_immutable() || object.version() == SequenceNumber::MIN,
            "Created mutable objects should not have a version set",
        );
        let id = object.id();
        self.execution_results.created_object_ids.insert(id);
        self.execution_results.written_objects.insert(id, object);
    }

    /// Delete a mutable input object. This is used to delete input objects outside of PT execution.
    pub fn delete_input_object(&mut self, id: &ObjectID) {
        // there should be no deletion after write
        debug_assert!(!self.execution_results.written_objects.contains_key(id));
        debug_assert!(self.input_objects.contains_key(id));
        self.execution_results.modified_objects.insert(*id);
        self.execution_results.deleted_object_ids.insert(*id);
    }

    pub fn drop_writes(&mut self) {
        self.execution_results.drop_writes();
        // The PTB-emitted ranges pointed into the now-cleared accumulator_events vec.
        self.ptb_emitted_accumulator_event_ranges.clear();
    }

    pub fn read_object(&self, id: &ObjectID) -> Option<&Object> {
        // there should be no read after delete
        debug_assert!(!self.execution_results.deleted_object_ids.contains(id));
        self.execution_results
            .written_objects
            .get(id)
            .or_else(|| self.input_objects.get(id))
    }

    pub fn save_loaded_runtime_objects(
        &mut self,
        loaded_runtime_objects: BTreeMap<ObjectID, DynamicallyLoadedObjectMetadata>,
    ) {
        #[cfg(debug_assertions)]
        {
            for (id, v1) in &loaded_runtime_objects {
                if let Some(v2) = self.loaded_runtime_objects.get(id) {
                    assert_eq!(v1, v2);
                }
            }
            for (id, v1) in &self.loaded_runtime_objects {
                if let Some(v2) = loaded_runtime_objects.get(id) {
                    assert_eq!(v1, v2);
                }
            }
        }
        // Merge the two maps because we may be calling the execution engine more than once
        // (e.g. in advance epoch transaction, where we may be publishing a new system package).
        self.loaded_runtime_objects.extend(loaded_runtime_objects);
    }

    pub fn save_wrapped_object_containers(
        &mut self,
        wrapped_object_containers: BTreeMap<ObjectID, ObjectID>,
    ) {
        #[cfg(debug_assertions)]
        {
            for (id, container1) in &wrapped_object_containers {
                if let Some(container2) = self.wrapped_object_containers.get(id) {
                    assert_eq!(container1, container2);
                }
            }
            for (id, container1) in &self.wrapped_object_containers {
                if let Some(container2) = wrapped_object_containers.get(id) {
                    assert_eq!(container1, container2);
                }
            }
        }
        // Merge the two maps because we may be calling the execution engine more than once
        // (e.g. in advance epoch transaction, where we may be publishing a new system package).
        self.wrapped_object_containers
            .extend(wrapped_object_containers);
    }

    pub fn save_generated_object_ids(&mut self, generated_ids: BTreeSet<ObjectID>) {
        #[cfg(debug_assertions)]
        {
            for id in &self.generated_runtime_ids {
                assert!(!generated_ids.contains(id))
            }
            for id in &generated_ids {
                assert!(!self.generated_runtime_ids.contains(id));
            }
        }
        self.generated_runtime_ids.extend(generated_ids);
    }

    pub fn estimate_effects_size_upperbound(&self) -> usize {
        TransactionEffects::estimate_effects_size_upperbound_v2(
            self.execution_results.written_objects.len(),
            self.execution_results.modified_objects.len(),
            self.input_objects.len(),
        )
    }

    pub fn written_objects_size(&self) -> usize {
        self.execution_results
            .written_objects
            .values()
            .fold(0, |sum, obj| sum + obj.object_size_for_gas_metering())
    }

    /// Validates gasless post-execution invariants:
    /// - No new objects were created or existing objects mutated (written_objects is empty)
    /// - The set of deleted objects exactly equals the set of input Coin objects
    /// - Each recipient receives at least the minimum transfer amount per token type
    /// - Unused withdrawal reservation (reservation - actual split) is 0 or >= min_amount
    pub fn check_gasless_execution_requirements(
        &self,
        withdrawal_reservations: Option<&BTreeMap<(SuiAddress, TypeTag), u64>>,
    ) -> Result<(), String> {
        if !self.execution_results.written_objects.is_empty() {
            return Err("Gasless transactions cannot create or mutate objects".to_string());
        }

        let input_coin_ids: BTreeSet<ObjectID> = self
            .input_objects
            .iter()
            .filter(|(_, obj)| obj.coin_type_maybe().is_some())
            .map(|(id, _)| *id)
            .collect();
        if self.execution_results.deleted_object_ids != input_coin_ids {
            return Err(format!(
                "Gasless transaction must destroy exactly its input Coins. \
                 Expected: {input_coin_ids:?}, deleted: {:?}",
                self.execution_results.deleted_object_ids
            ));
        }

        let allowed_types =
            sui_types::transaction::get_gasless_allowed_token_types(self.protocol_config);

        // Aggregate signed balance changes per (address, token_type).
        // Positive nets are recipient deposits that must meet the minimum transfer amount.
        let net_totals = sui_types::balance_change::signed_balance_changes_from_events(
            &self.execution_results.accumulator_events,
        )
        .fold(
            BTreeMap::<(SuiAddress, TypeTag), i128>::new(),
            |mut totals, (address, token_type, signed_amount)| {
                *totals.entry((address, token_type)).or_default() += signed_amount;
                totals
            },
        );

        for ((recipient, token_type), net_amount) in &net_totals {
            if *net_amount <= 0 {
                continue;
            }
            if let Some(&min_amount) = allowed_types.get(token_type)
                && *net_amount < i128::from(min_amount)
            {
                return Err(format!(
                    "Gasless transfer of {net_amount} to {recipient} is below \
                     minimum {min_amount} for token type {token_type}"
                ));
            }
        }

        if let Some(reservations) = withdrawal_reservations {
            for ((owner, token_type), &reserved) in reservations {
                let net = net_totals
                    .get(&(*owner, token_type.clone()))
                    .copied()
                    .unwrap_or(0);
                let remaining = (reserved as i128).saturating_add(net);
                if remaining > 0
                    && let Some(&min_balance_remaining) = allowed_types.get(token_type)
                    && min_balance_remaining > 0
                    && remaining < min_balance_remaining as i128
                {
                    return Err(format!(
                        "Gasless withdrawal leaves {remaining} unused for {owner}, \
                         below minimum {min_balance_remaining} for token type {token_type}"
                    ));
                }
            }
        }

        Ok(())
    }

    /// If there are unmetered storage rebate (due to system transaction), we put them into
    /// the storage rebate of 0x5 object.
    /// TODO: This will not work for potential future new system transactions if 0x5 is not in the input.
    /// We should fix this.
    pub fn conserve_unmetered_storage_rebate(&mut self, unmetered_storage_rebate: u64) {
        if unmetered_storage_rebate == 0 {
            // If unmetered_storage_rebate is 0, we are most likely executing the genesis transaction.
            // And in that case we cannot mutate the 0x5 object because it's newly created.
            // And there is no storage rebate that needs distribution anyway.
            return;
        }
        tracing::debug!(
            "Amount of unmetered storage rebate from system tx: {:?}",
            unmetered_storage_rebate
        );
        let mut system_state_wrapper = self
            .read_object(&SUI_SYSTEM_STATE_OBJECT_ID)
            .expect("0x5 object must be mutated in system tx with unmetered storage rebate")
            .clone();
        // In unmetered execution, storage_rebate field of mutated object must be 0.
        // If not, we would be dropping SUI on the floor by overriding it.
        assert_eq!(system_state_wrapper.storage_rebate, 0);
        system_state_wrapper.storage_rebate = unmetered_storage_rebate;
        self.mutate_input_object(system_state_wrapper);
    }

    /// Add an accumulator event to the execution results.
    pub fn add_accumulator_event(&mut self, event: AccumulatorEvent) {
        self.execution_results.accumulator_events.push(event);
    }

    /// Given an object ID, if it's not modified, returns None.
    /// Otherwise returns its metadata, including version, digest, owner and storage rebate.
    /// A modified object must be either a mutable input, or a loaded child object.
    /// The only exception is when we upgrade system packages, in which case the upgraded
    /// system packages are not part of input, but are modified.
    fn get_object_modified_at(
        &self,
        object_id: &ObjectID,
    ) -> Option<DynamicallyLoadedObjectMetadata> {
        if self.execution_results.modified_objects.contains(object_id) {
            Some(
                self.mutable_input_refs
                    .get(object_id)
                    .map(
                        |((version, digest), owner)| DynamicallyLoadedObjectMetadata {
                            version: *version,
                            digest: *digest,
                            owner: owner.clone(),
                            // It's guaranteed that a mutable input object is an input object.
                            storage_rebate: self.input_objects[object_id].storage_rebate,
                            previous_transaction: self.input_objects[object_id]
                                .previous_transaction,
                        },
                    )
                    .or_else(|| self.loaded_runtime_objects.get(object_id).cloned())
                    .unwrap_or_else(|| {
                        debug_assert!(is_system_package(*object_id));
                        let package_obj =
                            self.store.get_package_object(object_id).unwrap().unwrap();
                        let obj = package_obj.object();
                        DynamicallyLoadedObjectMetadata {
                            version: obj.version(),
                            digest: obj.digest(),
                            owner: obj.owner.clone(),
                            storage_rebate: obj.storage_rebate,
                            previous_transaction: obj.previous_transaction,
                        }
                    }),
            )
        } else {
            None
        }
    }

    pub fn protocol_config(&self) -> &'backing ProtocolConfig {
        self.protocol_config
    }
}

impl TemporaryStore<'_> {
    // check that every object read is owned directly or indirectly by sender, sponsor,
    // or a shared object input
    pub fn check_ownership_invariants(
        &self,
        sender: &SuiAddress,
        sponsor: &Option<SuiAddress>,
        gas_charger: &mut GasCharger,
        mutable_inputs: &HashSet<ObjectID>,
        is_epoch_change: bool,
    ) -> SuiResult<()> {
        let gas_objs: HashSet<&ObjectID> = gas_charger.used_coins().map(|g| &g.0).collect();
        let gas_owner = sponsor.as_ref().unwrap_or(sender);

        // mark input objects as authenticated
        let mut authenticated_for_mutation: HashSet<_> = self
            .input_objects
            .iter()
            .filter_map(|(id, obj)| {
                match &obj.owner {
                    Owner::AddressOwner(a) => {
                        if gas_objs.contains(id) {
                            // gas object must be owned by sender or sponsor
                            assert!(
                                a == gas_owner,
                                "Gas object must be owned by sender or sponsor"
                            );
                        } else {
                            assert!(sender == a, "Input object must be owned by sender");
                        }
                        Some(id)
                    }
                    Owner::Shared { .. } | Owner::ConsensusAddressOwner { .. } => Some(id),
                    Owner::Immutable => {
                        // object is authenticated, but it cannot own other objects,
                        // so we should not add it to `authenticated_objs`
                        // However, we would definitely want to add immutable objects
                        // to the set of authenticated roots if we were doing runtime
                        // checks inside the VM instead of after-the-fact in the temporary
                        // store. Here, we choose not to add them because this will catch a
                        // bug where we mutate or delete an object that belongs to an immutable
                        // object (though it will show up somewhat opaquely as an authentication
                        // failure), whereas adding the immutable object to the roots will prevent
                        // us from catching this.
                        None
                    }
                    Owner::ObjectOwner(_parent) => {
                        unreachable!(
                            "Input objects must be address owned, shared, consensus, or immutable"
                        )
                    }
                }
            })
            .filter(|id| {
                // remove any non-mutable inputs. This will remove deleted or readonly shared
                // objects
                mutable_inputs.contains(id)
            })
            .copied()
            // Add any object IDs generated in the object runtime during execution to the
            // authenticated set (i.e., new (non-package) objects, and possibly ephemeral UIDs).
            .chain(self.generated_runtime_ids.iter().copied())
            .collect();

        // Add sender and sponsor (if present) to authenticated set
        authenticated_for_mutation.insert((*sender).into());
        if let Some(sponsor) = sponsor {
            authenticated_for_mutation.insert((*sponsor).into());
        }

        // check all modified objects are authenticated
        let mut objects_to_authenticate = self
            .execution_results
            .modified_objects
            .iter()
            .copied()
            .collect::<Vec<_>>();

        while let Some(to_authenticate) = objects_to_authenticate.pop() {
            if authenticated_for_mutation.contains(&to_authenticate) {
                // object has already been authenticated
                continue;
            }

            let parent = if let Some(container_id) =
                self.wrapped_object_containers.get(&to_authenticate)
            {
                // It's a wrapped object, so check that the container is authenticated
                *container_id
            } else {
                // It's non-wrapped, so check the owner -- we can load the object from the
                // store.
                let Some(old_obj) = self.store.get_object(&to_authenticate) else {
                    panic!(
                        "Failed to load object {to_authenticate:?}.\n \
                         If it cannot be loaded, we would expect it to be in the wrapped object map: {:#?}",
                        &self.wrapped_object_containers
                    )
                };

                match &old_obj.owner {
                    // We mutated a dynamic field, we can continue to trace this back to verify
                    // proper ownership.
                    Owner::ObjectOwner(parent) => ObjectID::from(*parent),
                    // We mutated an address owned or sequenced address owned object -- one of two cases apply:
                    // 1) the object is owned by an object or address in the authenticated set,
                    // 2) the object is owned by some other address, in which case we should
                    //    continue to trace this back.
                    Owner::AddressOwner(parent)
                    | Owner::ConsensusAddressOwner { owner: parent, .. } => {
                        // For Receiving<_> objects, the address owner is actually an object.
                        // If it was actually an address, we should have caught it as an input and
                        // it would already have been in authenticated_for_mutation
                        ObjectID::from(*parent)
                    }
                    // We mutated a shared object -- we checked if this object was in the
                    // authenticated set at the top of this loop and it wasn't so this is a failure.
                    owner @ Owner::Shared { .. } => {
                        panic!(
                            "Unauthenticated root at {to_authenticate:?} with owner {owner:?}\n\
                             Potentially covering objects in: {authenticated_for_mutation:#?}"
                        );
                    }
                    Owner::Immutable => {
                        assert!(
                            is_epoch_change,
                            "Immutable objects cannot be written, except for \
                             Sui Framework/Move stdlib upgrades at epoch change boundaries"
                        );
                        // Note: this assumes that the only immutable objects an epoch change
                        // tx can update are system packages,
                        // but in principle we could allow others.
                        assert!(
                            is_system_package(to_authenticate),
                            "Only system packages can be upgraded"
                        );
                        continue;
                    }
                }
            };

            // we now assume the object is authenticated and check the parent
            authenticated_for_mutation.insert(to_authenticate);
            objects_to_authenticate.push(parent);
        }
        Ok(())
    }
}

impl TemporaryStore<'_> {
    /// Track storage gas for each mutable input object (including the gas coin)
    /// and each created object. Compute storage refunds for each deleted object.
    /// Will *not* charge anything, gas status keeps track of storage cost and rebate.
    /// All objects will be updated with their new (current) storage rebate/cost.
    /// `SuiGasStatus` `storage_rebate` and `storage_gas_units` track the transaction
    /// overall storage rebate and cost.
    pub(crate) fn collect_storage_and_rebate(&mut self, gas_charger: &mut GasCharger) {
        // Use two loops because we cannot mut iterate written while calling get_object_modified_at.
        let old_storage_rebates: Vec<_> = self
            .execution_results
            .written_objects
            .keys()
            .map(|object_id| {
                self.get_object_modified_at(object_id)
                    .map(|metadata| metadata.storage_rebate)
                    .unwrap_or_default()
            })
            .collect();
        for (object, old_storage_rebate) in self
            .execution_results
            .written_objects
            .values_mut()
            .zip_debug_eq(old_storage_rebates)
        {
            // new object size
            let new_object_size = object.object_size_for_gas_metering();
            // track changes and compute the new object `storage_rebate`
            let new_storage_rebate = gas_charger.track_storage_mutation(
                object.id(),
                new_object_size,
                old_storage_rebate,
            );
            object.storage_rebate = new_storage_rebate;
        }

        self.collect_rebate(gas_charger);
    }

    pub(crate) fn collect_rebate(&self, gas_charger: &mut GasCharger) {
        for object_id in &self.execution_results.modified_objects {
            if self
                .execution_results
                .written_objects
                .contains_key(object_id)
            {
                continue;
            }
            // get and track the deleted object `storage_rebate`
            let storage_rebate = self
                .get_object_modified_at(object_id)
                // Unwrap is safe because this loop iterates through all modified objects.
                .unwrap()
                .storage_rebate;
            gas_charger.track_storage_mutation(*object_id, 0, storage_rebate);
        }
    }

    pub fn check_execution_results_consistency(&self) -> Result<(), ExecutionError> {
        assert_invariant!(
            self.execution_results
                .created_object_ids
                .iter()
                .all(|id| !self.execution_results.deleted_object_ids.contains(id)
                    && !self.execution_results.modified_objects.contains(id)),
            "Created object IDs cannot also be deleted or modified"
        );
        assert_invariant!(
            self.execution_results.modified_objects.iter().all(|id| {
                self.mutable_input_refs.contains_key(id)
                    || self.loaded_runtime_objects.contains_key(id)
                    || is_system_package(*id)
            }),
            "A modified object must be either a mutable input, a loaded child object, or a system package"
        );
        Ok(())
    }
}
//==============================================================================
// Charge gas current - end
//==============================================================================

impl TemporaryStore<'_> {
    pub fn advance_epoch_safe_mode(
        &mut self,
        params: &AdvanceEpochParams,
        protocol_config: &ProtocolConfig,
    ) {
        let wrapper = get_sui_system_state_wrapper(self.store.as_object_store())
            .expect("System state wrapper object must exist");
        let (old_object, new_object) =
            wrapper.advance_epoch_safe_mode(params, self.store.as_object_store(), protocol_config);
        self.mutate_child_object(old_object, new_object);
    }
}

type ModifiedObjectInfo<'a> = (
    ObjectID,
    // old object metadata, including version, digest, owner, and storage rebate.
    Option<DynamicallyLoadedObjectMetadata>,
    Option<&'a Object>,
);

impl TemporaryStore<'_> {
    fn get_input_sui(
        &self,
        id: &ObjectID,
        expected_version: SequenceNumber,
        layout_resolver: &mut impl LayoutResolver,
    ) -> Result<u64, ExecutionError> {
        if let Some(obj) = self.input_objects.get(id) {
            // the assumption here is that if it is in the input objects must be the right one
            if obj.version() != expected_version {
                invariant_violation!(
                    "Version mismatching when resolving input object to check conservation--\
                     expected {}, got {}",
                    expected_version,
                    obj.version(),
                );
            }
            obj.get_total_sui(layout_resolver).map_err(|e| {
                make_invariant_violation!(
                    "Failed looking up input SUI in SUI conservation checking for input with \
                         type {:?}: {e:#?}",
                    obj.struct_tag(),
                )
            })
        } else {
            // not in input objects, must be a dynamic field
            let Some(obj) = self.store.get_object_by_key(id, expected_version) else {
                invariant_violation!(
                    "Failed looking up dynamic field {id} in SUI conservation checking"
                );
            };
            obj.get_total_sui(layout_resolver).map_err(|e| {
                make_invariant_violation!(
                    "Failed looking up input SUI in SUI conservation checking for type \
                         {:?}: {e:#?}",
                    obj.struct_tag(),
                )
            })
        }
    }

    /// Return the list of all modified objects, for each object, returns
    /// - Object ID,
    /// - Input: If the object existed prior to this transaction, include their version and storage_rebate,
    /// - Output: If a new version of the object is written, include the new object.
    fn get_modified_objects(&self) -> Vec<ModifiedObjectInfo<'_>> {
        self.execution_results
            .modified_objects
            .iter()
            .map(|id| {
                let metadata = self.get_object_modified_at(id);
                let output = self.execution_results.written_objects.get(id);
                (*id, metadata, output)
            })
            .chain(
                self.execution_results
                    .written_objects
                    .iter()
                    .filter_map(|(id, object)| {
                        if self.execution_results.modified_objects.contains(id) {
                            None
                        } else {
                            Some((*id, None, Some(object)))
                        }
                    }),
            )
            .collect()
    }

    /// Check that this transaction neither creates nor destroys SUI. This should hold for all txes
    /// except the epoch change tx, which mints staking rewards equal to the gas fees burned in the
    /// previous epoch.  Specifically, this checks two key invariants about storage
    /// fees and storage rebate:
    ///
    /// 1. all SUI in storage rebate fields of input objects should flow either to the transaction
    ///    storage rebate, or the transaction non-refundable storage rebate
    /// 2. all SUI charged for storage should flow into the storage rebate field of some output
    ///    object
    ///
    /// This function is intended to be called *after* we have charged for
    /// gas + applied the storage rebate to the gas object, but *before* we
    /// have updated object versions.
    pub fn check_sui_conserved(
        &self,
        simple_conservation_checks: bool,
        gas_summary: &GasCostSummary,
    ) -> Result<(), ExecutionError> {
        if !simple_conservation_checks {
            return Ok(());
        }
        // total amount of SUI in storage rebate of input objects
        let mut total_input_rebate = 0;
        // total amount of SUI in storage rebate of output objects
        let mut total_output_rebate = 0;
        for (_id, input, output) in self.get_modified_objects() {
            if let Some(input) = input {
                total_input_rebate += input.storage_rebate;
            }
            if let Some(object) = output {
                total_output_rebate += object.storage_rebate;
            }
        }

        if gas_summary.storage_cost == 0 {
            // this condition is usually true when the transaction went OOG and no
            // gas is left for storage charges.
            // The storage cost has to be there at least for the gas coin which
            // will not be deleted even when going to 0.
            // However if the storage cost is 0 and if there is any object touched
            // or deleted the value in input must be equal to the output plus rebate and
            // non refundable.
            // Rebate and non refundable will be positive when there are object deleted
            // (gas smashing being the primary and possibly only example).
            // A more typical condition is for all storage charges in summary to be 0 and
            // then input and output must be the same value
            if total_input_rebate
                != total_output_rebate
                    + gas_summary.storage_rebate
                    + gas_summary.non_refundable_storage_fee
            {
                return Err(ExecutionError::invariant_violation(format!(
                    "SUI conservation failed -- no storage charges in gas summary \
                        and total storage input rebate {} not equal  \
                        to total storage output rebate {}",
                    total_input_rebate, total_output_rebate,
                )));
            }
        } else {
            // all SUI in storage rebate fields of input objects should flow either to
            // the transaction storage rebate, or the non-refundable storage rebate pool
            if total_input_rebate
                != gas_summary.storage_rebate + gas_summary.non_refundable_storage_fee
            {
                return Err(ExecutionError::invariant_violation(format!(
                    "SUI conservation failed -- {} SUI in storage rebate field of input objects, \
                        {} SUI in tx storage rebate or tx non-refundable storage rebate",
                    total_input_rebate, gas_summary.non_refundable_storage_fee,
                )));
            }

            // all SUI charged for storage should flow into the storage rebate field
            // of some output object
            if gas_summary.storage_cost != total_output_rebate {
                return Err(ExecutionError::invariant_violation(format!(
                    "SUI conservation failed -- {} SUI charged for storage, \
                        {} SUI in storage rebate field of output objects",
                    gas_summary.storage_cost, total_output_rebate
                )));
            }
        }
        Ok(())
    }

    /// Defense-in-depth invariant on funds-accumulator events. Per `(address, type)`:
    /// - If the pair is in `input_reservations`: net withdrawal ≤ budget.
    /// - Else if PTB-emitted events touched it: runtime contribution must not push the net
    ///   below Move's deposit (`actual ≥ min(0, ptb_change)`).
    /// - Else: any event is unauthorized — fatal.
    ///
    /// Currently the only funds-accumulator type is `Balance<T>`, so the check is scoped to
    /// those events. As more accumulator shapes are added the filter and the integer
    /// arithmetic in `check_address_balance_changes_impl` will need to grow with them.
    ///
    /// PTB-emitted events are identified via `ptb_emitted_accumulator_event_ranges`, populated
    /// at `record_execution_results` time. They are trusted because Move enforces `&mut UID`
    /// and the native checks the actual balance.
    ///
    /// `protocol_config.enforce_address_balance_change_invariant()` selects the failure mode:
    /// - On (post-flag): violations are returned as `Err` so the caller's
    ///   conservation-recovery flow can abort the tx cleanly.
    /// - Off (pre-flag): the check still runs, but a violation panics so unexpected
    ///   violations surface loudly during rollout.
    pub fn check_address_balance_changes(
        &self,
        protocol_config: &ProtocolConfig,
        input_reservations: &BTreeMap<(SuiAddress, TypeTag), u64>,
    ) -> Result<(), ExecutionError> {
        let result = self.check_address_balance_changes_impl(input_reservations);
        if protocol_config.enforce_address_balance_change_invariant() {
            result
        } else {
            if let Err(e) = result {
                panic!("address-balance-change invariant violated pre-flag: {e}");
            }
            Ok(())
        }
    }

    fn check_address_balance_changes_impl(
        &self,
        input_reservations: &BTreeMap<(SuiAddress, TypeTag), u64>,
    ) -> Result<(), ExecutionError> {
        use sui_types::balance::Balance;

        let mut actual_changes: BTreeMap<(SuiAddress, TypeTag), i128> = BTreeMap::new();
        let mut ptb_changes: BTreeMap<(SuiAddress, TypeTag), i128> = BTreeMap::new();
        for (idx, event) in self.execution_results.accumulator_events.iter().enumerate() {
            // Filter on the value shape first: only `Integer` carries the funds-flow we care
            // about. Other shapes (e.g. `EventDigest` for event-stream heads) belong to
            // non-Balance accumulators and are out of scope here. If we ever see an `Integer`
            // value at a non-`Balance<T>` type, the accounting invariants below don't apply
            // — debug_fatal so that case is surfaced instead of silently accepted.
            let amount = match event.write.value {
                AccumulatorValue::Integer(amount) => amount as i128,
                AccumulatorValue::IntegerTuple(_, _) | AccumulatorValue::EventDigest(_) => {
                    continue;
                }
            };
            if !Balance::is_balance_type(&event.write.address.ty) {
                debug_fatal!(
                    "Integer accumulator value at non-Balance type: {:?}",
                    event.write.address.ty
                );
                continue;
            }
            let is_ptb_emitted = self
                .ptb_emitted_accumulator_event_ranges
                .iter()
                .any(|range| range.contains(&idx));
            let key = (event.write.address.address, event.write.address.ty.clone());
            let change = match event.write.operation {
                AccumulatorOperation::Split => -amount,
                AccumulatorOperation::Merge => amount,
            };
            *actual_changes.entry(key.clone()).or_insert(0) += change;
            if is_ptb_emitted {
                *ptb_changes.entry(key).or_insert(0) += change;
            }
        }

        for (key, actual) in actual_changes {
            let (address, type_tag) = &key;
            if let Some(budget) = input_reservations.get(&key).copied() {
                let net_withdrawn = -actual.min(0) as u128;
                assert_invariant!(
                    net_withdrawn <= budget as u128,
                    "Balance accumulator withdrawal exceeds reservation budget at address \
                    {address} for type {type_tag}: net Split {net_withdrawn}, budget {budget}"
                );
            } else if let Some(ptb_change) = ptb_changes.get(&key).copied() {
                // Runtime-emitted withdrawals at this (address, type) are bounded by Move's
                // net deposit at the same key: actual ≥ min(0, ptb_change). When Move
                // deposited (ptb_change > 0), the runtime may withdraw down to 0; when Move
                // withdrew (ptb_change < 0), the runtime may not withdraw further.
                assert_invariant!(
                    actual >= ptb_change.min(0),
                    "PTB-emitted Balance accumulator events do not cover runtime withdrawals \
                    at address {address} for type {type_tag}: PTB change {ptb_change}, net \
                    change {actual}"
                );
            } else {
                invariant_violation!(
                    "Unauthorized runtime Balance accumulator event at address {address} for \
                    type {type_tag}: net change {actual} (no input reservation, no PTB-emitted \
                    events)"
                );
            }
        }

        Ok(())
    }

    /// Check that this transaction neither creates nor destroys SUI.
    /// This more expensive check will check a third invariant on top of the 2 performed
    /// by `check_sui_conserved` above:
    ///
    /// * all SUI in input objects (including coins etc in the Move part of an object) should flow
    ///   either to an output object, or be burned as part of computation fees or non-refundable
    ///   storage rebate
    ///
    /// This function is intended to be called *after* we have charged for gas + applied the
    /// storage rebate to the gas object, but *before* we have updated object versions. The
    /// advance epoch transaction would mint `epoch_fees` amount of SUI, and burn `epoch_rebates`
    /// amount of SUI. We need these information for this check.
    pub fn check_sui_conserved_expensive(
        &self,
        gas_summary: &GasCostSummary,
        advance_epoch_gas_summary: Option<(u64, u64)>,
        layout_resolver: &mut impl LayoutResolver,
    ) -> Result<(), ExecutionError> {
        // total amount of SUI in input objects, including both coins and storage rebates
        let mut total_input_sui = 0;
        // total amount of SUI in output objects, including both coins and storage rebates
        let mut total_output_sui = 0;

        // settlement input/output sui is used by the settlement transactions to account for
        // Sui that has been gathered from the accumulator writes of transactions which it is
        // settling.
        total_input_sui += self.execution_results.settlement_input_sui;
        total_output_sui += self.execution_results.settlement_output_sui;

        for (id, input, output) in self.get_modified_objects() {
            if let Some(input) = input {
                total_input_sui += self.get_input_sui(&id, input.version, layout_resolver)?;
            }
            if let Some(object) = output {
                total_output_sui += object.get_total_sui(layout_resolver).map_err(|e| {
                    make_invariant_violation!(
                        "Failed looking up output SUI in SUI conservation checking for \
                         mutated type {:?}: {e:#?}",
                        object.struct_tag(),
                    )
                })?;
            }
        }

        for event in &self.execution_results.accumulator_events {
            let (input, output) = event.total_sui_in_event();
            total_input_sui += input;
            total_output_sui += output;
        }

        // note: storage_cost flows into the storage_rebate field of the output objects, which is
        // why it is not accounted for here.
        // similarly, all of the storage_rebate *except* the storage_fund_rebate_inflow
        // gets credited to the gas coin both computation costs and storage rebate inflow are
        total_output_sui += gas_summary.computation_cost + gas_summary.non_refundable_storage_fee;
        if let Some((epoch_fees, epoch_rebates)) = advance_epoch_gas_summary {
            total_input_sui += epoch_fees;
            total_output_sui += epoch_rebates;
        }
        if total_input_sui != total_output_sui {
            return Err(ExecutionError::invariant_violation(format!(
                "SUI conservation failed: input={}, output={}, \
                    this transaction either mints or burns SUI",
                total_input_sui, total_output_sui,
            )));
        }
        Ok(())
    }
}

impl ChildObjectResolver for TemporaryStore<'_> {
    fn read_child_object(
        &self,
        parent: &ObjectID,
        child: &ObjectID,
        child_version_upper_bound: SequenceNumber,
    ) -> SuiResult<Option<Object>> {
        let obj_opt = self.execution_results.written_objects.get(child);
        if obj_opt.is_some() {
            Ok(obj_opt.cloned())
        } else {
            let _scope = monitored_scope("Execution::read_child_object");
            self.store
                .read_child_object(parent, child, child_version_upper_bound)
        }
    }

    fn get_object_received_at_version(
        &self,
        owner: &ObjectID,
        receiving_object_id: &ObjectID,
        receive_object_at_version: SequenceNumber,
        epoch_id: EpochId,
    ) -> SuiResult<Option<Object>> {
        // You should never be able to try and receive an object after deleting it or writing it in the same
        // transaction since `Receiving` doesn't have copy.
        debug_assert!(
            !self
                .execution_results
                .written_objects
                .contains_key(receiving_object_id)
        );
        debug_assert!(
            !self
                .execution_results
                .deleted_object_ids
                .contains(receiving_object_id)
        );
        self.store.get_object_received_at_version(
            owner,
            receiving_object_id,
            receive_object_at_version,
            epoch_id,
        )
    }
}

/// Compares the owner and payload of an object.
/// This is used to detect illegal writes to non-exclusive write objects.
fn was_object_mutated(object: &Object, original: &Object) -> bool {
    let data_equal = match (&object.data, &original.data) {
        (Data::Move(a), Data::Move(b)) => a.contents_and_type_equal(b),
        // We don't have a use for package content-equality, so we remain as strict as
        // possible for now.
        (Data::Package(a), Data::Package(b)) => a == b,
        _ => false,
    };

    let owner_equal = match (&object.owner, &original.owner) {
        // We don't compare initial shared versions, because re-shared objects do not have the
        // correct initial shared version at this point in time, and this field is not something
        // that can be modified by a single transaction anyway.
        (Owner::Shared { .. }, Owner::Shared { .. }) => true,
        (
            Owner::ConsensusAddressOwner { owner: a, .. },
            Owner::ConsensusAddressOwner { owner: b, .. },
        ) => a == b,
        (Owner::AddressOwner(a), Owner::AddressOwner(b)) => a == b,
        (Owner::Immutable, Owner::Immutable) => true,
        (Owner::ObjectOwner(a), Owner::ObjectOwner(b)) => a == b,

        // Keep the left hand side of the match exhaustive to catch future
        // changes to Owner
        (Owner::AddressOwner(_), _)
        | (Owner::Immutable, _)
        | (Owner::ObjectOwner(_), _)
        | (Owner::Shared { .. }, _)
        | (Owner::ConsensusAddressOwner { .. }, _) => false,
    };

    !data_equal || !owner_equal
}

impl Storage for TemporaryStore<'_> {
    fn reset(&mut self) {
        self.drop_writes();
    }

    fn read_object(&self, id: &ObjectID) -> Option<&Object> {
        TemporaryStore::read_object(self, id)
    }

    /// Take execution results v2, and translate it back to be compatible with effects v1.
    fn record_execution_results(
        &mut self,
        results: ExecutionResults,
    ) -> Result<(), ExecutionError> {
        let ExecutionResults::V2(mut results) = results else {
            panic!("ExecutionResults::V2 expected in sui-execution v1 and above");
        };

        // for all non-exclusive write inputs, remove them from written objects
        let mut to_remove = Vec::new();
        for (id, original) in &self.non_exclusive_input_original_versions {
            // Object must be present in `written_objects` and identical
            if results
                .written_objects
                .get(id)
                .map(|obj| was_object_mutated(obj, original))
                .unwrap_or(true)
            {
                return Err(ExecutionError::new_with_source(
                    ExecutionErrorKind::NonExclusiveWriteInputObjectModified { id: *id },
                    "Non-exclusive write input object has been modified or deleted",
                ));
            }
            to_remove.push(*id);
        }

        for id in to_remove {
            results.written_objects.remove(&id);
            results.modified_objects.remove(&id);
        }

        // It's important to merge instead of override results because it's
        // possible to execute PT more than once during tx execution.
        // Track the index range of accumulator events brought in here as PTB-emitted; the
        // address-balance change invariant (run inside `run_conservation_checks`) uses this
        // set to distinguish trusted PTB-emitted events from runtime-emitted ones.
        let event_start = self.execution_results.accumulator_events.len();
        self.execution_results.merge_results(
            results, /* consistent_merge */ true, /* invariant_checks */ true,
        )?;
        let event_end = self.execution_results.accumulator_events.len();
        debug_assert!(
            event_start <= event_end,
            "merge_results should not shrink accumulator_events"
        );
        let (event_start, event_end) = (event_start.min(event_end), event_start.max(event_end));
        let range = event_start..event_end;
        match self.ptb_emitted_accumulator_event_ranges.last_mut() {
            // Coalesce with the previous PTB range if no runtime events were added in between.
            Some(last) if last.end == range.start => last.end = range.end,
            _ => self.ptb_emitted_accumulator_event_ranges.push(range),
        }

        Ok(())
    }

    fn save_loaded_runtime_objects(
        &mut self,
        loaded_runtime_objects: BTreeMap<ObjectID, DynamicallyLoadedObjectMetadata>,
    ) {
        TemporaryStore::save_loaded_runtime_objects(self, loaded_runtime_objects)
    }

    fn save_wrapped_object_containers(
        &mut self,
        wrapped_object_containers: BTreeMap<ObjectID, ObjectID>,
    ) {
        TemporaryStore::save_wrapped_object_containers(self, wrapped_object_containers)
    }

    fn check_coin_deny_list(
        &self,
        receiving_funds_type_and_owners: BTreeMap<TypeTag, BTreeSet<SuiAddress>>,
    ) -> DenyListResult {
        let result = check_coin_deny_list_v2_during_execution(
            receiving_funds_type_and_owners,
            self.cur_epoch,
            self.store.as_object_store(),
        );
        // The denylist object is only loaded if there are regulated transfers.
        // And also if we already have it in the input there is no need to commit it again in the effects.
        if result.num_non_gas_coin_owners > 0
            && !self.input_objects.contains_key(&SUI_DENY_LIST_OBJECT_ID)
        {
            self.loaded_per_epoch_config_objects
                .write()
                .insert(SUI_DENY_LIST_OBJECT_ID);
        }
        result
    }

    fn record_generated_object_ids(&mut self, generated_ids: BTreeSet<ObjectID>) {
        TemporaryStore::save_generated_object_ids(self, generated_ids)
    }
}

impl BackingPackageStore for TemporaryStore<'_> {
    fn get_package_object(&self, package_id: &ObjectID) -> SuiResult<Option<PackageObject>> {
        // We first check the objects in the temporary store because in non-production code path,
        // it is possible to read packages that are just written in the same transaction.
        // This can happen for example when we run the expensive conservation checks, where we may
        // look into the types of each written object in the output, and some of them need the
        // newly written packages for type checking.
        // In production path though, this should never happen.
        if let Some(obj) = self.execution_results.written_objects.get(package_id) {
            Ok(Some(PackageObject::new(obj.clone())))
        } else {
            self.store.get_package_object(package_id).inspect(|obj| {
                // Track object but leave unchanged
                if let Some(v) = obj
                    && !self
                        .runtime_packages_loaded_from_db
                        .read()
                        .contains_key(package_id)
                {
                    // TODO: Can this lock ever block execution?
                    // TODO: Another way to avoid the cost of maintaining this map is to not
                    // enable it in normal runs, and if a fork is detected, rerun it with a flag
                    // turned on and start populating this field.
                    self.runtime_packages_loaded_from_db
                        .write()
                        .insert(*package_id, v.clone());
                }
            })
        }
    }
}

impl ParentSync for TemporaryStore<'_> {
    fn get_latest_parent_entry_ref_deprecated(&self, _object_id: ObjectID) -> Option<ObjectRef> {
        unreachable!("Never called in newer protocol versions")
    }
}