sui_core/authority/

shared_object_version_manager.rs

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

use mysten_common::ZipDebugEqIteratorExt;
use mysten_common::debug_fatal;

use crate::authority::AuthorityPerEpochStore;
use crate::authority::authority_per_epoch_store::CancelConsensusCertificateReason;
use crate::execution_cache::ObjectCacheRead;
use either::Either;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use sui_types::SUI_ACCUMULATOR_ROOT_OBJECT_ID;
use sui_types::SUI_CLOCK_OBJECT_ID;
use sui_types::SUI_CLOCK_OBJECT_SHARED_VERSION;
use sui_types::SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID;
use sui_types::base_types::ConsensusObjectSequenceKey;
use sui_types::base_types::ConsensusObjectVersion;
use sui_types::base_types::ObjectID;
use sui_types::base_types::SystemObjectVersions;
use sui_types::base_types::TransactionDigest;
use sui_types::committee::EpochId;
use sui_types::crypto::RandomnessRound;
use sui_types::effects::{TransactionEffects, TransactionEffectsAPI};
use sui_types::executable_transaction::VerifiedExecutableTransaction;
use sui_types::executable_transaction::VerifiedExecutableTransactionWithAliases;
use sui_types::storage::{
    ObjectKey, transaction_non_shared_input_object_keys, transaction_receiving_object_keys,
};
use sui_types::transaction::SharedObjectMutability;
use sui_types::transaction::{SharedInputObject, TransactionDataAPI, TransactionKey};
use sui_types::{
    IMPLICITLY_READ_SYSTEM_OBJECTS, SUI_RANDOMNESS_STATE_OBJECT_ID, base_types::SequenceNumber,
    error::SuiResult,
};
use tracing::trace;

pub struct SharedObjVerManager {}

/// Version assignments for a single transaction
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AssignedVersions {
    pub shared_object_versions: Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
    /// Versions of system objects, keyed by object ID, that this transaction may read during
    /// execution but that are not part of its declared shared inputs. Each version is assigned
    /// deterministically during consensus sequencing, so that every validator reads the same
    /// version of the object.
    ///
    /// Today this holds the accumulator root version (as of the beginning of the consensus
    /// commit this transaction belongs to) and the forwarding address registry version (as of the
    /// point in the commit at which this transaction is sequenced). The accumulator root qualifies
    /// because it is written at the end of every commit, so there is always a well-defined prior
    /// version to read from.
    pub system_object_versions: SystemObjectVersions,
}

impl AssignedVersions {
    pub fn new(
        shared_object_versions: Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
        system_object_versions: SystemObjectVersions,
    ) -> Self {
        debug_assert!(
            shared_object_versions
                .iter()
                .filter(|((id, _), version)| {
                    id.is_implicitly_read_system_object() && !version.is_cancelled()
                })
                .all(|((id, _), version)| {
                    system_object_versions
                        .get(id)
                        .is_none_or(|system_version| system_version.version == *version)
                }),
            "declared versions {shared_object_versions:?} disagree with {system_object_versions:?}"
        );
        Self {
            shared_object_versions,
            system_object_versions,
        }
    }

    pub fn empty() -> Self {
        Self::new(vec![], SystemObjectVersions::empty())
    }

    /// Sets the accumulator root as a system object read during execution. For test paths
    /// that execute directly: the test version-assignment helper assigns no root version,
    /// but execution reads the root implicitly for any object funds withdraw.
    pub fn with_accumulator_version_for_testing(mut self, version: ConsensusObjectVersion) -> Self {
        self.system_object_versions = SystemObjectVersions::new(
            Some(version),
            self.system_object_versions
                .get(&sui_types::SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID),
        );
        self
    }

    /// Construct with only the accumulator root as the system object read during execution.
    /// Production callers build the full `system_object_versions` map directly.
    #[cfg(test)]
    pub fn new_for_testing(
        shared_object_versions: Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
        accumulator_version: Option<SequenceNumber>,
    ) -> Self {
        Self::new(
            shared_object_versions,
            SystemObjectVersions::new(
                accumulator_version.map(|v| ConsensusObjectVersion {
                    initial_shared_version: sui_types::object::OBJECT_START_VERSION,
                    version: v,
                }),
                None,
            ),
        )
    }

    /// The accumulator root version this transaction reads, if any.
    pub fn accumulator_version(&self) -> Option<SequenceNumber> {
        self.system_object_versions
            .get(&SUI_ACCUMULATOR_ROOT_OBJECT_ID)
            .map(|v| v.version)
    }

    pub fn iter(&self) -> impl Iterator<Item = &(ConsensusObjectSequenceKey, SequenceNumber)> {
        self.shared_object_versions.iter()
    }

    pub fn as_slice(&self) -> &[(ConsensusObjectSequenceKey, SequenceNumber)] {
        &self.shared_object_versions
    }
}

#[derive(Default, Debug, PartialEq, Eq)]
pub struct AssignedTxAndVersions(pub Vec<(TransactionKey, AssignedVersions)>);

impl AssignedTxAndVersions {
    pub fn new(assigned_versions: Vec<(TransactionKey, AssignedVersions)>) -> Self {
        Self(assigned_versions)
    }

    pub fn into_map(self) -> HashMap<TransactionKey, AssignedVersions> {
        self.0.into_iter().collect()
    }
}

/// A wrapper around things that can be scheduled for execution by the assigning of
/// shared object versions.
#[derive(Clone)]
pub enum Schedulable<T = VerifiedExecutableTransaction> {
    Transaction(T),
    RandomnessStateUpdate(EpochId, RandomnessRound),
    AccumulatorSettlement(EpochId, u64 /* checkpoint height */),
    ConsensusCommitPrologue(EpochId, u64 /* round */, u32 /* sub_dag_index */),
}

impl From<VerifiedExecutableTransaction> for Schedulable<VerifiedExecutableTransaction> {
    fn from(tx: VerifiedExecutableTransaction) -> Self {
        Schedulable::Transaction(tx)
    }
}

impl From<Schedulable<VerifiedExecutableTransactionWithAliases>>
    for Schedulable<VerifiedExecutableTransaction>
{
    fn from(schedulable: Schedulable<VerifiedExecutableTransactionWithAliases>) -> Self {
        match schedulable {
            Schedulable::Transaction(tx) => Schedulable::Transaction(tx.into_tx()),
            Schedulable::RandomnessStateUpdate(epoch, round) => {
                Schedulable::RandomnessStateUpdate(epoch, round)
            }
            Schedulable::AccumulatorSettlement(epoch, checkpoint_height) => {
                Schedulable::AccumulatorSettlement(epoch, checkpoint_height)
            }
            Schedulable::ConsensusCommitPrologue(epoch, round, sub_dag_index) => {
                Schedulable::ConsensusCommitPrologue(epoch, round, sub_dag_index)
            }
        }
    }
}

// AsTx is like Deref, in that it allows us to use either refs or values in Schedulable.
// Deref does not work because it conflicts with the impl of Deref for VerifiedExecutableTransaction.
pub trait AsTx {
    fn as_tx(&self) -> &VerifiedExecutableTransaction;
}

impl AsTx for VerifiedExecutableTransaction {
    fn as_tx(&self) -> &VerifiedExecutableTransaction {
        self
    }
}

impl AsTx for &'_ VerifiedExecutableTransaction {
    fn as_tx(&self) -> &VerifiedExecutableTransaction {
        self
    }
}

impl AsTx for VerifiedExecutableTransactionWithAliases {
    fn as_tx(&self) -> &VerifiedExecutableTransaction {
        self.tx()
    }
}

impl AsTx for &'_ VerifiedExecutableTransactionWithAliases {
    fn as_tx(&self) -> &VerifiedExecutableTransaction {
        self.tx()
    }
}

impl Schedulable<&'_ VerifiedExecutableTransaction> {
    // Cannot use the blanket ToOwned trait impl because it just calls clone.
    pub fn to_owned_schedulable(&self) -> Schedulable<VerifiedExecutableTransaction> {
        match self {
            Schedulable::Transaction(tx) => Schedulable::Transaction((*tx).clone()),
            Schedulable::RandomnessStateUpdate(epoch, round) => {
                Schedulable::RandomnessStateUpdate(*epoch, *round)
            }
            Schedulable::AccumulatorSettlement(epoch, checkpoint_height) => {
                Schedulable::AccumulatorSettlement(*epoch, *checkpoint_height)
            }
            Schedulable::ConsensusCommitPrologue(epoch, round, sub_dag_index) => {
                Schedulable::ConsensusCommitPrologue(*epoch, *round, *sub_dag_index)
            }
        }
    }
}

impl<T> Schedulable<T> {
    pub fn as_tx(&self) -> Option<&VerifiedExecutableTransaction>
    where
        T: AsTx,
    {
        match self {
            Schedulable::Transaction(tx) => Some(tx.as_tx()),
            Schedulable::RandomnessStateUpdate(_, _) => None,
            Schedulable::AccumulatorSettlement(_, _) => None,
            Schedulable::ConsensusCommitPrologue(_, _, _) => None,
        }
    }

    pub fn shared_input_objects(
        &self,
        epoch_store: &AuthorityPerEpochStore,
    ) -> impl Iterator<Item = SharedInputObject> + '_
    where
        T: AsTx,
    {
        match self {
            Schedulable::Transaction(tx) => Either::Left(tx.as_tx().shared_input_objects()),
            Schedulable::RandomnessStateUpdate(_, _) => {
                Either::Right(std::iter::once(SharedInputObject {
                    id: SUI_RANDOMNESS_STATE_OBJECT_ID,
                    initial_shared_version: epoch_store
                        .epoch_start_config()
                        .randomness_obj_initial_shared_version()
                        .expect("randomness obj initial shared version should be set"),
                    mutability: SharedObjectMutability::Mutable,
                }))
            }
            Schedulable::AccumulatorSettlement(_, _) => {
                Either::Right(std::iter::once(SharedInputObject {
                    id: SUI_ACCUMULATOR_ROOT_OBJECT_ID,
                    initial_shared_version: epoch_store
                        .epoch_start_config()
                        .accumulator_root_obj_initial_shared_version()
                        .expect("accumulator root obj initial shared version should be set"),
                    mutability: SharedObjectMutability::Mutable,
                }))
            }
            Schedulable::ConsensusCommitPrologue(_, _, _) => {
                Either::Right(std::iter::once(SharedInputObject {
                    id: SUI_CLOCK_OBJECT_ID,
                    initial_shared_version: SUI_CLOCK_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }))
            }
        }
    }

    pub fn non_shared_input_object_keys(&self) -> Vec<ObjectKey>
    where
        T: AsTx,
    {
        match self {
            Schedulable::Transaction(tx) => transaction_non_shared_input_object_keys(tx.as_tx())
                .expect("Transaction input should have been verified"),
            Schedulable::RandomnessStateUpdate(_, _) => vec![],
            Schedulable::AccumulatorSettlement(_, _) => vec![],
            Schedulable::ConsensusCommitPrologue(_, _, _) => vec![],
        }
    }

    pub fn receiving_object_keys(&self) -> Vec<ObjectKey>
    where
        T: AsTx,
    {
        match self {
            Schedulable::Transaction(tx) => transaction_receiving_object_keys(tx.as_tx()),
            Schedulable::RandomnessStateUpdate(_, _) => vec![],
            Schedulable::AccumulatorSettlement(_, _) => vec![],
            Schedulable::ConsensusCommitPrologue(_, _, _) => vec![],
        }
    }

    pub fn key(&self) -> TransactionKey
    where
        T: AsTx,
    {
        match self {
            Schedulable::Transaction(tx) => tx.as_tx().key(),
            Schedulable::RandomnessStateUpdate(epoch, round) => {
                TransactionKey::RandomnessRound(*epoch, *round)
            }
            Schedulable::AccumulatorSettlement(epoch, checkpoint_height) => {
                TransactionKey::AccumulatorSettlement(*epoch, *checkpoint_height)
            }
            Schedulable::ConsensusCommitPrologue(epoch, round, sub_dag_index) => {
                TransactionKey::ConsensusCommitPrologue(*epoch, *round, *sub_dag_index)
            }
        }
    }
}

#[must_use]
#[derive(Default, Eq, PartialEq, Debug)]
pub struct ConsensusSharedObjVerAssignment {
    pub shared_input_next_versions: HashMap<ConsensusObjectSequenceKey, SequenceNumber>,
    pub assigned_versions: AssignedTxAndVersions,
}

impl SharedObjVerManager {
    pub fn assign_versions_from_consensus<'a, T>(
        epoch_store: &AuthorityPerEpochStore,
        cache_reader: &dyn ObjectCacheRead,
        assignables: impl Iterator<Item = &'a Schedulable<T>> + Clone,
        cancelled_txns: &BTreeMap<TransactionDigest, CancelConsensusCertificateReason>,
    ) -> SuiResult<ConsensusSharedObjVerAssignment>
    where
        T: AsTx + 'a,
    {
        let mut shared_input_next_versions = get_or_init_versions(
            assignables
                .clone()
                .flat_map(|a| a.shared_input_objects(epoch_store)),
            epoch_store,
            cache_reader,
        )?;
        let mut assigned_versions = Vec::new();
        for assignable in assignables {
            assert!(
                !matches!(assignable, Schedulable::AccumulatorSettlement(_, _))
                    || epoch_store.accumulators_enabled(),
                "AccumulatorSettlement should not be scheduled when accumulators are disabled"
            );

            let cert_assigned_versions = Self::assign_versions_for_certificate(
                epoch_store,
                assignable,
                &mut shared_input_next_versions,
                cancelled_txns,
            );
            assigned_versions.push((assignable.key(), cert_assigned_versions));
        }

        Ok(ConsensusSharedObjVerAssignment {
            shared_input_next_versions,
            assigned_versions: AssignedTxAndVersions::new(assigned_versions),
        })
    }

    pub fn assign_versions_from_effects(
        certs_and_effects: &[(
            &VerifiedExecutableTransaction,
            &TransactionEffects,
            // Accumulator version
            Option<SequenceNumber>,
        )],
        epoch_store: &AuthorityPerEpochStore,
        cache_reader: &dyn ObjectCacheRead,
    ) -> AssignedTxAndVersions {
        // We don't care about the results since we can use effects to assign versions.
        // But we must call it to make sure whenever a consensus object is touched the first time
        // during an epoch, either through consensus or through checkpoint executor,
        // its next version must be initialized. This is because we initialize the next version
        // of a consensus object in an epoch by reading the current version from the object store.
        // This must be done before we mutate it the first time, otherwise we would be initializing
        // it with the wrong version.
        let _ = get_or_init_versions(
            certs_and_effects.iter().flat_map(|(cert, _, _)| {
                cert.transaction_data().shared_input_objects().into_iter()
            }),
            epoch_store,
            cache_reader,
        );
        let mut assigned_versions = Vec::new();
        for (cert, effects, accumulator_version) in certs_and_effects {
            let initial_version_map: BTreeMap<_, _> = cert
                .transaction_data()
                .shared_input_objects()
                .into_iter()
                .map(|input| input.into_id_and_version())
                .collect();
            // Implicitly read system objects retrieve their accessed versions from this map.
            let accessed_versions: BTreeMap<ObjectID, SequenceNumber> = effects
                .accessed_consensus_objects()
                .into_iter()
                .map(|iso| iso.id_and_version())
                .collect();
            let cert_assigned_versions: Vec<_> = accessed_versions
                .iter()
                .filter_map(|(id, version)| {
                    let v = initial_version_map
                        .get(id)
                        .map(|initial_version| ((*id, *initial_version), *version));
                    if v.is_none() {
                        debug_assert!(
                            IMPLICITLY_READ_SYSTEM_OBJECTS.contains(id),
                            "accessed consensus object is neither a declared input nor a known implicitly read system object: \
                             accessed={accessed_versions:?} declared={initial_version_map:?}"
                        );
                    }
                    v
                })
                .collect();
            if let (Some(effects_version), Some(sequenced_version)) = (
                accessed_versions.get(&SUI_ACCUMULATOR_ROOT_OBJECT_ID),
                accumulator_version,
            ) && !effects_version.is_cancelled()
                && effects_version != sequenced_version
            {
                debug_fatal!(
                    "accumulator root version from effects {:?} disagrees \
                        with the reconstructed accumulator version {:?} for tx {:?}",
                    effects_version,
                    sequenced_version,
                    cert.digest()
                );
            }
            let mut system_object_versions: BTreeMap<_, _> =
                implicitly_read_system_objects(epoch_store)
                    .into_iter()
                    .filter_map(|(id, initial_shared_version)| {
                        let version = *accessed_versions.get(&id)?;
                        // A cancelled sentinel only appears when the object was also a declared
                        // input of a cancelled transaction; it is not a version to pin.
                        (!version.is_cancelled()).then_some((
                            id,
                            ConsensusObjectVersion {
                                initial_shared_version,
                                version,
                            },
                        ))
                    })
                    .collect();
            // Note that for accumulator version, we cannot rely on the one from effects yet, since it won't
            // be produced until implicitly read system objects are fully shipped. But the old object funds withdraw
            // still need it. Hence we always use the one provided from the caller (i.e. checkpoint executor).
            // TODO: Once check_object_funds_withdraw_in_execution is enabled everywhere and the
            // post-execution ObjectFundsChecker is removed, drop this override (and the
            // accumulator version threaded from the checkpoint executor) so the accumulator root
            // is reconstructed from effects exactly like the forwarding address registry.
            if let Some(version) = *accumulator_version {
                let initial_shared_version = epoch_store
                    .epoch_start_config()
                    .accumulator_root_obj_initial_shared_version()
                    .expect(
                        "initial shared version must be known for an implicitly read system object",
                    );
                system_object_versions.insert(
                    SUI_ACCUMULATOR_ROOT_OBJECT_ID,
                    ConsensusObjectVersion {
                        initial_shared_version,
                        version,
                    },
                );
            }
            let system_object_versions = SystemObjectVersions::from_map(system_object_versions);
            let tx_key = cert.key();
            trace!(
                ?tx_key,
                ?cert_assigned_versions,
                ?system_object_versions,
                "assigned consensus object versions from effects"
            );
            assigned_versions.push((
                tx_key,
                AssignedVersions::new(cert_assigned_versions, system_object_versions),
            ));
        }
        AssignedTxAndVersions::new(assigned_versions)
    }

    pub fn assign_versions_for_certificate(
        epoch_store: &AuthorityPerEpochStore,
        assignable: &Schedulable<impl AsTx>,
        shared_input_next_versions: &mut HashMap<ConsensusObjectSequenceKey, SequenceNumber>,
        cancelled_txns: &BTreeMap<TransactionDigest, CancelConsensusCertificateReason>,
    ) -> AssignedVersions {
        let shared_input_objects: Vec<_> = assignable.shared_input_objects(epoch_store).collect();

        let system_object_versions = SystemObjectVersions::from_map(
            implicitly_read_system_objects(epoch_store)
                .into_iter()
                .map(|(id, initial_shared_version)| {
                    let version = *shared_input_next_versions
                        .get(&(id, initial_shared_version))
                        .expect(
                            "implicitly read system objects must be in shared_input_next_versions",
                        );
                    (
                        id,
                        ConsensusObjectVersion {
                            initial_shared_version,
                            version,
                        },
                    )
                })
                .collect(),
        );

        if shared_input_objects.is_empty() {
            // No shared object used by this transaction. No need to assign versions.
            return AssignedVersions::new(vec![], system_object_versions);
        }

        let tx_key = assignable.key();

        // Check if the transaction is cancelled due to congestion.
        let cancellation_info = tx_key
            .as_digest()
            .and_then(|tx_digest| cancelled_txns.get(tx_digest));
        let congested_objects_info: Option<HashSet<_>> =
            if let Some(CancelConsensusCertificateReason::CongestionOnObjects(congested_objects)) =
                &cancellation_info
            {
                Some(congested_objects.iter().cloned().collect())
            } else {
                None
            };
        let txn_cancelled = cancellation_info.is_some();

        let mut input_object_keys = assignable.non_shared_input_object_keys();
        let mut assigned_versions = Vec::with_capacity(shared_input_objects.len());
        let mut is_exclusively_accessed_input = Vec::with_capacity(shared_input_objects.len());
        // Record receiving object versions towards the shared version computation.
        let receiving_object_keys = assignable.receiving_object_keys();
        input_object_keys.extend(receiving_object_keys);

        if txn_cancelled {
            // For cancelled transaction due to congestion, assign special versions to all shared objects.
            // Note that new lamport version does not depend on any shared objects.
            for SharedInputObject {
                id,
                initial_shared_version,
                ..
            } in shared_input_objects.iter()
            {
                let assigned_version = match cancellation_info {
                    Some(CancelConsensusCertificateReason::CongestionOnObjects(_)) => {
                        if congested_objects_info
                            .as_ref()
                            .is_some_and(|info| info.contains(id))
                        {
                            SequenceNumber::CONGESTED
                        } else {
                            SequenceNumber::CANCELLED_READ
                        }
                    }
                    Some(CancelConsensusCertificateReason::DkgFailed) => {
                        if id == &SUI_RANDOMNESS_STATE_OBJECT_ID {
                            SequenceNumber::RANDOMNESS_UNAVAILABLE
                        } else {
                            SequenceNumber::CANCELLED_READ
                        }
                    }
                    None => unreachable!("cancelled transaction should have cancellation info"),
                };
                assigned_versions.push(((*id, *initial_shared_version), assigned_version));
                is_exclusively_accessed_input.push(false);
            }
        } else {
            for (
                SharedInputObject {
                    id,
                    initial_shared_version,
                    mutability,
                },
                assigned_version,
            ) in shared_input_objects.iter().map(|obj| {
                (
                    obj,
                    *shared_input_next_versions
                        .get(&obj.id_and_version())
                        .unwrap(),
                )
            }) {
                assigned_versions.push(((*id, *initial_shared_version), assigned_version));
                input_object_keys.push(ObjectKey(*id, assigned_version));
                is_exclusively_accessed_input.push(mutability.is_exclusive());
            }
        }

        let next_version =
            SequenceNumber::lamport_increment(input_object_keys.iter().map(|obj| obj.1));
        assert!(
            next_version.is_valid(),
            "Assigned version must be valid. Got {:?}",
            next_version
        );

        if !txn_cancelled {
            // Update the next version for the shared objects.
            assigned_versions
                .iter()
                .zip_debug_eq(is_exclusively_accessed_input)
                .filter_map(|((id, _), mutable)| {
                    if mutable {
                        Some((*id, next_version))
                    } else {
                        None
                    }
                })
                .for_each(|(id, version)| {
                    assert!(
                        version.is_valid(),
                        "Assigned version must be a valid version."
                    );
                    shared_input_next_versions
                        .insert(id, version)
                        .expect("Object must exist in shared_input_next_versions.");
                });
        }

        trace!(
            ?tx_key,
            ?assigned_versions,
            ?next_version,
            ?txn_cancelled,
            "locking shared objects"
        );

        AssignedVersions::new(assigned_versions, system_object_versions)
    }
}

fn implicitly_read_system_objects(
    epoch_store: &AuthorityPerEpochStore,
) -> Vec<ConsensusObjectSequenceKey> {
    let epoch_start_config = epoch_store.epoch_start_config();
    let mut objects = Vec::new();
    if epoch_store.accumulators_enabled() {
        objects.push((
            SUI_ACCUMULATOR_ROOT_OBJECT_ID,
            epoch_start_config
                .accumulator_root_obj_initial_shared_version()
                .expect("accumulator root obj initial shared version should be set"),
        ));
    }
    if epoch_store.protocol_config().enable_forwarding_addresses()
        && let Some(initial_shared_version) =
            epoch_start_config.forwarding_address_registry_obj_initial_shared_version()
    {
        objects.push((
            SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
            initial_shared_version,
        ));
    }
    objects
}

fn get_or_init_versions<'a>(
    shared_input_objects: impl Iterator<Item = SharedInputObject> + 'a,
    epoch_store: &AuthorityPerEpochStore,
    cache_reader: &dyn ObjectCacheRead,
) -> SuiResult<HashMap<ConsensusObjectSequenceKey, SequenceNumber>> {
    let mut shared_input_objects: Vec<_> = shared_input_objects
        .map(|so| so.into_id_and_version())
        .collect();

    shared_input_objects.extend(implicitly_read_system_objects(epoch_store));

    shared_input_objects.sort();
    shared_input_objects.dedup();

    epoch_store.get_or_init_next_object_versions(&shared_input_objects, cache_reader)
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::authority::AuthorityState;
    use crate::authority::shared_object_version_manager::{
        ConsensusSharedObjVerAssignment, SharedObjVerManager,
    };
    use crate::authority::test_authority_builder::TestAuthorityBuilder;
    use std::collections::{BTreeMap, HashMap};
    use std::sync::Arc;
    use sui_protocol_config::ProtocolConfig;
    use sui_test_transaction_builder::TestTransactionBuilder;
    use sui_types::base_types::{ObjectID, SequenceNumber, SuiAddress};
    use sui_types::crypto::{RandomnessRound, get_account_key_pair};
    use sui_types::digests::ObjectDigest;
    use sui_types::effects::TestEffectsBuilder;
    use sui_types::executable_transaction::{
        CertificateProof, ExecutableTransaction, VerifiedExecutableTransaction,
    };

    use sui_types::object::Object;
    use sui_types::transaction::{ObjectArg, SenderSignedData, VerifiedTransaction};

    use sui_types::gas_coin::GAS;
    use sui_types::transaction::FundsWithdrawalArg;
    use sui_types::{SUI_ACCUMULATOR_ROOT_OBJECT_ID, SUI_RANDOMNESS_STATE_OBJECT_ID};

    fn assigned_versions_for_testing(
        shared_object_versions: Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
        accumulator_version: Option<SequenceNumber>,
        forwarding_address_registry_version: Option<SequenceNumber>,
    ) -> AssignedVersions {
        let at_start_version = |v| ConsensusObjectVersion {
            initial_shared_version: sui_types::object::OBJECT_START_VERSION,
            version: v,
        };
        AssignedVersions::new(
            shared_object_versions,
            SystemObjectVersions::new(
                accumulator_version.map(at_start_version),
                forwarding_address_registry_version.map(at_start_version),
            ),
        )
    }

    #[tokio::test]
    async fn test_assign_versions_from_consensus_basic() {
        let shared_object = Object::shared_for_testing();
        let id = shared_object.id();
        let init_shared_version = shared_object.owner.start_version().unwrap();
        let authority = TestAuthorityBuilder::new()
            .with_starting_objects(std::slice::from_ref(&shared_object))
            .build()
            .await;
        let certs = [
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 3),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, false)], 5),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 9),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 11),
        ];
        let epoch_store = authority.epoch_store_for_testing();
        let assignables = certs
            .iter()
            .map(Schedulable::Transaction)
            .collect::<Vec<_>>();
        let ConsensusSharedObjVerAssignment {
            shared_input_next_versions,
            assigned_versions,
        } = SharedObjVerManager::assign_versions_from_consensus(
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
            assignables.iter(),
            &BTreeMap::new(),
        )
        .unwrap();
        // Check that the shared object's next version is always initialized in the epoch store.
        assert_eq!(
            epoch_store
                .get_next_object_version(&id, init_shared_version)
                .unwrap(),
            init_shared_version
        );
        // Check that the final version of the shared object is the lamport version of the last
        // transaction.
        assert_eq!(
            *shared_input_next_versions
                .get(&(id, init_shared_version))
                .unwrap(),
            SequenceNumber::from_u64(12)
        );
        // Check that the version assignment for each transaction is correct.
        // For a transaction that uses the shared object with mutable=false, it won't update the version
        // using lamport version, hence the next transaction will use the same version number.
        // In the following case, certs[2] has the same assignment as certs[1] for this reason.
        let expected_accumulator_version = SequenceNumber::from_u64(1);
        let expected_registry_version = SequenceNumber::from_u64(1);
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), init_shared_version)],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[1].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(4))],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[2].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(4))],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[3].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(10))],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
            ]
        );
    }

    #[tokio::test]
    async fn test_forwarding_registry_version_follows_consensus_order() {
        let authority = TestAuthorityBuilder::new().build().await;
        let epoch_store = authority.epoch_store_for_testing();
        let registry_initial_version = epoch_store
            .epoch_start_config()
            .forwarding_address_registry_obj_initial_shared_version()
            .unwrap();
        let registry_key = (
            SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
            registry_initial_version,
        );
        let certs = [
            generate_shared_objs_tx_with_gas_version(&[], 3),
            generate_shared_objs_tx_with_gas_version(
                &[(
                    SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
                    registry_initial_version,
                    true,
                )],
                5,
            ),
            generate_shared_objs_tx_with_gas_version(&[], 1),
        ];
        let assignables = certs
            .iter()
            .map(Schedulable::Transaction)
            .collect::<Vec<_>>();
        let ConsensusSharedObjVerAssignment {
            shared_input_next_versions,
            assigned_versions,
        } = SharedObjVerManager::assign_versions_from_consensus(
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
            assignables.iter(),
            &BTreeMap::new(),
        )
        .unwrap();
        let registry_version = |version| ConsensusObjectVersion {
            initial_shared_version: registry_initial_version,
            version,
        };
        // The registration is sequenced second, so only the last transaction reads the
        // registry at its post-registration version.
        assert_eq!(
            assigned_versions
                .0
                .iter()
                .map(|(_, versions)| {
                    (
                        versions.shared_object_versions.clone(),
                        versions
                            .system_object_versions
                            .get(&SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID)
                            .unwrap(),
                    )
                })
                .collect::<Vec<_>>(),
            vec![
                (vec![], registry_version(registry_initial_version)),
                (
                    vec![(registry_key, registry_initial_version)],
                    registry_version(registry_initial_version),
                ),
                (vec![], registry_version(SequenceNumber::from_u64(6))),
            ]
        );
        assert_eq!(
            shared_input_next_versions.get(&registry_key),
            Some(&SequenceNumber::from_u64(6))
        );
    }

    #[tokio::test]
    async fn test_forwarding_registry_declared_by_cancelled_transaction() {
        let authority = TestAuthorityBuilder::new().build().await;
        let epoch_store = authority.epoch_store_for_testing();
        let registry_initial_version = epoch_store
            .epoch_start_config()
            .forwarding_address_registry_obj_initial_shared_version()
            .unwrap();
        let registry_key = (
            SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
            registry_initial_version,
        );
        let certs = [
            generate_shared_objs_tx_with_gas_version(
                &[(
                    SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
                    registry_initial_version,
                    true,
                )],
                5,
            ),
            generate_shared_objs_tx_with_gas_version(&[], 3),
        ];
        let cancelled_txns = BTreeMap::from([(
            *certs[0].digest(),
            CancelConsensusCertificateReason::CongestionOnObjects(vec![
                SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
            ]),
        )]);
        let assignables = certs
            .iter()
            .map(Schedulable::Transaction)
            .collect::<Vec<_>>();
        let ConsensusSharedObjVerAssignment {
            shared_input_next_versions,
            assigned_versions,
        } = SharedObjVerManager::assign_versions_from_consensus(
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
            assignables.iter(),
            &cancelled_txns,
        )
        .unwrap();
        // The cancelled declaration carries the sentinel, while the implicit pin keeps the real
        // version and the registry does not advance for the transaction after it.
        let expected_version = SequenceNumber::from_u64(1);
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    assigned_versions_for_testing(
                        vec![(registry_key, SequenceNumber::CONGESTED)],
                        Some(expected_version),
                        Some(expected_version)
                    )
                ),
                (
                    certs[1].key(),
                    assigned_versions_for_testing(
                        vec![],
                        Some(expected_version),
                        Some(expected_version)
                    )
                ),
            ]
        );
        assert_eq!(
            shared_input_next_versions.get(&registry_key),
            Some(&registry_initial_version)
        );
    }

    #[tokio::test]
    async fn test_forwarding_registry_version_from_effects() {
        let authority = TestAuthorityBuilder::new().build().await;
        let epoch_store = authority.epoch_store_for_testing();
        let registry_initial_version = epoch_store
            .epoch_start_config()
            .forwarding_address_registry_obj_initial_shared_version()
            .unwrap();
        let registry_version = SequenceNumber::from_u64(4);
        let certs = [
            generate_shared_objs_tx_with_gas_version(&[], 3),
            generate_shared_objs_tx_with_gas_version(
                &[(
                    SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
                    registry_initial_version,
                    false,
                )],
                5,
            ),
        ];
        let effects = certs
            .iter()
            .map(|cert| {
                TestEffectsBuilder::new(cert.data())
                    .with_shared_input_versions(BTreeMap::from([(
                        SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
                        registry_version,
                    )]))
                    .build()
            })
            .collect::<Vec<_>>();
        let assigned_versions = SharedObjVerManager::assign_versions_from_effects(
            certs
                .iter()
                .zip_debug_eq(effects.iter())
                .map(|(cert, effect)| (cert, effect, None))
                .collect::<Vec<_>>()
                .as_slice(),
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
        );
        let system_object_versions = SystemObjectVersions::new(
            None,
            Some(ConsensusObjectVersion {
                initial_shared_version: registry_initial_version,
                version: registry_version,
            }),
        );
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    AssignedVersions::new(vec![], system_object_versions)
                ),
                (
                    certs[1].key(),
                    AssignedVersions::new(
                        vec![(
                            (
                                SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
                                registry_initial_version
                            ),
                            registry_version
                        )],
                        system_object_versions
                    )
                ),
            ]
        );
    }

    #[tokio::test]
    async fn test_assign_versions_from_consensus_with_randomness() {
        let authority = TestAuthorityBuilder::new().build().await;
        let epoch_store = authority.epoch_store_for_testing();
        let randomness_obj_version = epoch_store
            .epoch_start_config()
            .randomness_obj_initial_shared_version()
            .unwrap();
        let certs = [
            VerifiedExecutableTransaction::new_system(
                VerifiedTransaction::new_randomness_state_update(
                    epoch_store.epoch(),
                    RandomnessRound::new(1),
                    vec![],
                    randomness_obj_version,
                ),
                epoch_store.epoch(),
            ),
            generate_shared_objs_tx_with_gas_version(
                &[(
                    SUI_RANDOMNESS_STATE_OBJECT_ID,
                    randomness_obj_version,
                    // This can only be false since it's not allowed to use randomness object with mutable=true.
                    false,
                )],
                3,
            ),
            generate_shared_objs_tx_with_gas_version(
                &[(
                    SUI_RANDOMNESS_STATE_OBJECT_ID,
                    randomness_obj_version,
                    false,
                )],
                5,
            ),
        ];
        let assignables = certs
            .iter()
            .map(Schedulable::Transaction)
            .collect::<Vec<_>>();
        let ConsensusSharedObjVerAssignment {
            shared_input_next_versions,
            assigned_versions,
        } = SharedObjVerManager::assign_versions_from_consensus(
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
            assignables.iter(),
            &BTreeMap::new(),
        )
        .unwrap();
        // Check that the randomness object's next version is initialized.
        assert_eq!(
            epoch_store
                .get_next_object_version(&SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version)
                .unwrap(),
            randomness_obj_version
        );
        let next_randomness_obj_version = randomness_obj_version.next();
        assert_eq!(
            *shared_input_next_versions
                .get(&(SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version))
                .unwrap(),
            // Randomness object's version is only incremented by 1 regardless of lamport version.
            next_randomness_obj_version
        );
        let expected_accumulator_version = SequenceNumber::from_u64(1);
        let expected_registry_version = SequenceNumber::from_u64(1);
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    assigned_versions_for_testing(
                        vec![(
                            (SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version),
                            randomness_obj_version
                        )],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[1].key(),
                    // It is critical that the randomness object version is updated before the assignment.
                    assigned_versions_for_testing(
                        vec![(
                            (SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version),
                            next_randomness_obj_version
                        )],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[2].key(),
                    // It is critical that the randomness object version is updated before the assignment.
                    assigned_versions_for_testing(
                        vec![(
                            (SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version),
                            next_randomness_obj_version
                        )],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
            ]
        );
    }

    // Tests shared object version assignment for cancelled transaction.
    #[tokio::test]
    async fn test_assign_versions_from_consensus_with_cancellation() {
        let shared_object_1 = Object::shared_for_testing();
        let shared_object_2 = Object::shared_for_testing();
        let id1 = shared_object_1.id();
        let id2 = shared_object_2.id();
        let init_shared_version_1 = shared_object_1.owner.start_version().unwrap();
        let init_shared_version_2 = shared_object_2.owner.start_version().unwrap();
        let authority = TestAuthorityBuilder::new()
            .with_starting_objects(&[shared_object_1.clone(), shared_object_2.clone()])
            .build()
            .await;
        let randomness_obj_version = authority
            .epoch_store_for_testing()
            .epoch_start_config()
            .randomness_obj_initial_shared_version()
            .unwrap();

        // Generate 5 transactions for testing.
        //   tx1: shared_object_1, shared_object_2, owned_object_version = 3
        //   tx2: shared_object_1, shared_object_2, owned_object_version = 5
        //   tx3: shared_object_1, owned_object_version = 1
        //   tx4: shared_object_1, shared_object_2, owned_object_version = 9
        //   tx5: shared_object_1, shared_object_2, owned_object_version = 11
        //
        // Later, we cancel transaction 2 and 4 due to congestion, and 5 due to DKG failure.
        // Expected outcome:
        //   tx1: both shared objects assign version 1, lamport version = 4
        //   tx2: shared objects assign cancelled version, lamport version = 6 due to gas object version = 5
        //   tx3: shared object 1 assign version 4, lamport version = 5
        //   tx4: shared objects assign cancelled version, lamport version = 10 due to gas object version = 9
        //   tx5: shared objects assign cancelled version, lamport version = 12 due to gas object version = 11
        let certs = [
            generate_shared_objs_tx_with_gas_version(
                &[
                    (id1, init_shared_version_1, true),
                    (id2, init_shared_version_2, true),
                ],
                3,
            ),
            generate_shared_objs_tx_with_gas_version(
                &[
                    (id1, init_shared_version_1, true),
                    (id2, init_shared_version_2, true),
                ],
                5,
            ),
            generate_shared_objs_tx_with_gas_version(&[(id1, init_shared_version_1, true)], 1),
            generate_shared_objs_tx_with_gas_version(
                &[
                    (id1, init_shared_version_1, true),
                    (id2, init_shared_version_2, true),
                ],
                9,
            ),
            generate_shared_objs_tx_with_gas_version(
                &[
                    (
                        SUI_RANDOMNESS_STATE_OBJECT_ID,
                        randomness_obj_version,
                        false,
                    ),
                    (id2, init_shared_version_2, true),
                ],
                11,
            ),
        ];
        let epoch_store = authority.epoch_store_for_testing();

        // Cancel transactions 2 and 4 due to congestion.
        let cancelled_txns: BTreeMap<TransactionDigest, CancelConsensusCertificateReason> = [
            (
                *certs[1].digest(),
                CancelConsensusCertificateReason::CongestionOnObjects(vec![id1]),
            ),
            (
                *certs[3].digest(),
                CancelConsensusCertificateReason::CongestionOnObjects(vec![id2]),
            ),
            (
                *certs[4].digest(),
                CancelConsensusCertificateReason::DkgFailed,
            ),
        ]
        .into_iter()
        .collect();

        let assignables = certs
            .iter()
            .map(Schedulable::Transaction)
            .collect::<Vec<_>>();

        // Run version assignment logic.
        let ConsensusSharedObjVerAssignment {
            mut shared_input_next_versions,
            assigned_versions,
        } = SharedObjVerManager::assign_versions_from_consensus(
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
            assignables.iter(),
            &cancelled_txns,
        )
        .unwrap();

        // Check that the final version of the shared object is the lamport version of the last
        // transaction.
        shared_input_next_versions
            .remove(&(SUI_ACCUMULATOR_ROOT_OBJECT_ID, SequenceNumber::from_u64(1)));
        shared_input_next_versions.remove(&(
            SUI_FORWARDING_ADDRESS_REGISTRY_OBJECT_ID,
            SequenceNumber::from_u64(1),
        ));
        assert_eq!(
            shared_input_next_versions,
            HashMap::from([
                ((id1, init_shared_version_1), SequenceNumber::from_u64(5)), // determined by tx3
                ((id2, init_shared_version_2), SequenceNumber::from_u64(4)), // determined by tx1
                (
                    (SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version),
                    SequenceNumber::from_u64(1)
                ), // not mutable
            ])
        );

        // Check that the version assignment for each transaction is correct.
        let expected_accumulator_version = SequenceNumber::from_u64(1);
        let expected_registry_version = SequenceNumber::from_u64(1);
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    assigned_versions_for_testing(
                        vec![
                            ((id1, init_shared_version_1), init_shared_version_1),
                            ((id2, init_shared_version_2), init_shared_version_2)
                        ],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[1].key(),
                    assigned_versions_for_testing(
                        vec![
                            ((id1, init_shared_version_1), SequenceNumber::CONGESTED),
                            ((id2, init_shared_version_2), SequenceNumber::CANCELLED_READ),
                        ],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[2].key(),
                    assigned_versions_for_testing(
                        vec![((id1, init_shared_version_1), SequenceNumber::from_u64(4))],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[3].key(),
                    assigned_versions_for_testing(
                        vec![
                            ((id1, init_shared_version_1), SequenceNumber::CANCELLED_READ),
                            ((id2, init_shared_version_2), SequenceNumber::CONGESTED)
                        ],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
                (
                    certs[4].key(),
                    assigned_versions_for_testing(
                        vec![
                            (
                                (SUI_RANDOMNESS_STATE_OBJECT_ID, randomness_obj_version),
                                SequenceNumber::RANDOMNESS_UNAVAILABLE
                            ),
                            ((id2, init_shared_version_2), SequenceNumber::CANCELLED_READ)
                        ],
                        Some(expected_accumulator_version),
                        Some(expected_registry_version)
                    )
                ),
            ]
        );
    }

    #[tokio::test]
    async fn test_assign_versions_from_effects() {
        let shared_object = Object::shared_for_testing();
        let id = shared_object.id();
        let init_shared_version = shared_object.owner.start_version().unwrap();
        let authority = TestAuthorityBuilder::new()
            .with_starting_objects(std::slice::from_ref(&shared_object))
            .build()
            .await;
        let certs = [
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 3),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, false)], 5),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 9),
            generate_shared_objs_tx_with_gas_version(&[(id, init_shared_version, true)], 11),
        ];
        let effects = [
            TestEffectsBuilder::new(certs[0].data()).build(),
            TestEffectsBuilder::new(certs[1].data())
                .with_shared_input_versions(BTreeMap::from([(id, SequenceNumber::from_u64(4))]))
                .build(),
            TestEffectsBuilder::new(certs[2].data())
                .with_shared_input_versions(BTreeMap::from([(id, SequenceNumber::from_u64(4))]))
                .build(),
            TestEffectsBuilder::new(certs[3].data())
                .with_shared_input_versions(BTreeMap::from([(id, SequenceNumber::from_u64(10))]))
                .build(),
        ];
        let epoch_store = authority.epoch_store_for_testing();
        let assigned_versions = SharedObjVerManager::assign_versions_from_effects(
            certs
                .iter()
                .zip_debug_eq(effects.iter())
                .map(|(cert, effect)| (cert, effect, None))
                .collect::<Vec<_>>()
                .as_slice(),
            &epoch_store,
            authority.get_object_cache_reader().as_ref(),
        );
        // Check that the shared object's next version is always initialized in the epoch store.
        assert_eq!(
            epoch_store
                .get_next_object_version(&id, init_shared_version)
                .unwrap(),
            init_shared_version
        );
        assert_eq!(
            assigned_versions.0,
            vec![
                (
                    certs[0].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), init_shared_version)],
                        None,
                        None
                    )
                ),
                (
                    certs[1].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(4))],
                        None,
                        None
                    )
                ),
                (
                    certs[2].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(4))],
                        None,
                        None
                    )
                ),
                (
                    certs[3].key(),
                    assigned_versions_for_testing(
                        vec![((id, init_shared_version), SequenceNumber::from_u64(10))],
                        None,
                        None
                    )
                ),
            ]
        );
    }

    /// Generate a transaction that uses shared objects as specified in the parameters.
    /// Also uses a gas object with specified version.
    /// The version of the gas object is used to manipulate the lamport version of this transaction.
    fn generate_shared_objs_tx_with_gas_version(
        shared_objects: &[(ObjectID, SequenceNumber, bool)],
        gas_object_version: u64,
    ) -> VerifiedExecutableTransaction {
        let mut tx_builder = TestTransactionBuilder::new(
            SuiAddress::ZERO,
            (
                ObjectID::random(),
                SequenceNumber::from_u64(gas_object_version),
                ObjectDigest::random(),
            ),
            0,
        );
        let tx_data = {
            let builder = tx_builder.ptb_builder_mut();
            for (shared_object_id, shared_object_init_version, shared_object_mutable) in
                shared_objects
            {
                builder
                    .obj(ObjectArg::SharedObject {
                        id: *shared_object_id,
                        initial_shared_version: *shared_object_init_version,
                        mutability: if *shared_object_mutable {
                            SharedObjectMutability::Mutable
                        } else {
                            SharedObjectMutability::Immutable
                        },
                    })
                    .unwrap();
            }
            tx_builder.build()
        };
        let tx = SenderSignedData::new(tx_data, vec![]);
        VerifiedExecutableTransaction::new_unchecked(ExecutableTransaction::new_from_data_and_sig(
            tx,
            CertificateProof::new_system(0),
        ))
    }

    struct WithdrawTestContext {
        authority: Arc<AuthorityState>,
        assignables: Vec<Schedulable<VerifiedExecutableTransaction>>,
        shared_objects: Vec<Object>,
    }

    impl WithdrawTestContext {
        pub async fn new() -> Self {
            // Create a shared object for testing
            let shared_objects = vec![Object::shared_for_testing()];
            let mut config = ProtocolConfig::get_for_max_version_UNSAFE();
            config.set_enable_accumulators_for_testing(true);
            config.set_enable_forwarding_addresses_for_testing(false);
            let authority = TestAuthorityBuilder::new()
                .with_starting_objects(&shared_objects)
                .with_protocol_config(config)
                .build()
                .await;
            Self {
                authority,
                assignables: vec![],
                shared_objects,
            }
        }

        pub fn add_withdraw_transaction(&mut self) -> TransactionKey {
            // Generate random sender and gas object for each transaction
            let (sender, keypair) = get_account_key_pair();
            let gas_object = Object::with_owner_for_testing(sender);
            let gas_object_ref = gas_object.compute_object_reference();
            // Generate a unique gas price to make the transaction unique.
            let gas_price = (self.assignables.len() + 1) as u64;
            let mut tx_builder = TestTransactionBuilder::new(sender, gas_object_ref, gas_price);
            let tx_data = {
                let ptb_builder = tx_builder.ptb_builder_mut();
                ptb_builder
                    .funds_withdrawal(FundsWithdrawalArg::balance_from_sender(
                        200,
                        GAS::type_tag(),
                    ))
                    .unwrap();
                tx_builder.build()
            };
            let cert = VerifiedExecutableTransaction::new_for_testing(tx_data, &keypair);
            let key = cert.key();
            self.assignables.push(Schedulable::Transaction(cert));
            key
        }

        pub fn add_settlement_transaction(&mut self) -> TransactionKey {
            let height = (self.assignables.len() + 1) as u64;
            let settlement = Schedulable::AccumulatorSettlement(0, height);
            let key = settlement.key();
            self.assignables.push(settlement);
            key
        }

        pub fn add_withdraw_with_shared_object_transaction(&mut self) -> TransactionKey {
            // Generate random sender and gas object for each transaction
            let (sender, keypair) = get_account_key_pair();
            let gas_object = Object::with_owner_for_testing(sender);
            let gas_object_ref = gas_object.compute_object_reference();
            // Generate a unique gas price to make the transaction unique.
            let gas_price = (self.assignables.len() + 1) as u64;
            let mut tx_builder = TestTransactionBuilder::new(sender, gas_object_ref, gas_price);
            let tx_data = {
                let ptb_builder = tx_builder.ptb_builder_mut();
                // Add shared object to the transaction
                if let Some(shared_obj) = self.shared_objects.first() {
                    let id = shared_obj.id();
                    let init_version = shared_obj.owner.start_version().unwrap();
                    ptb_builder
                        .obj(ObjectArg::SharedObject {
                            id,
                            initial_shared_version: init_version,
                            mutability: SharedObjectMutability::Mutable,
                        })
                        .unwrap();
                }
                // Add balance withdraw
                ptb_builder
                    .funds_withdrawal(FundsWithdrawalArg::balance_from_sender(
                        200,
                        GAS::type_tag(),
                    ))
                    .unwrap();
                tx_builder.build()
            };
            let cert = VerifiedExecutableTransaction::new_for_testing(tx_data, &keypair);
            let key = cert.key();
            self.assignables.push(Schedulable::Transaction(cert));
            key
        }

        pub fn assign_versions_from_consensus(&self) -> ConsensusSharedObjVerAssignment {
            let epoch_store = self.authority.epoch_store_for_testing();
            SharedObjVerManager::assign_versions_from_consensus(
                &epoch_store,
                self.authority.get_object_cache_reader().as_ref(),
                self.assignables.iter(),
                &BTreeMap::new(),
            )
            .unwrap()
        }
    }

    #[tokio::test]
    async fn test_assign_versions_from_consensus_with_withdraws_simple() {
        // Note that we don't need a shared object to trigger withdraw version assignment.
        // In fact it is important that this works without a shared object.
        let mut ctx = WithdrawTestContext::new().await;

        let acc_version = ctx
            .authority
            .get_object(&SUI_ACCUMULATOR_ROOT_OBJECT_ID)
            .unwrap()
            .version();

        let withdraw_key = ctx.add_withdraw_transaction();
        let settlement_key = ctx.add_settlement_transaction();

        let assigned_versions = ctx.assign_versions_from_consensus();
        assert_eq!(
            assigned_versions,
            ConsensusSharedObjVerAssignment {
                assigned_versions: AssignedTxAndVersions::new(vec![
                    (
                        withdraw_key,
                        assigned_versions_for_testing(vec![], Some(acc_version), None)
                    ),
                    (
                        settlement_key,
                        assigned_versions_for_testing(
                            vec![((SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version), acc_version)],
                            Some(acc_version),
                            None
                        )
                    ),
                ]),
                shared_input_next_versions: HashMap::from([(
                    (SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version),
                    acc_version.next()
                )]),
            }
        );
    }

    #[tokio::test]
    async fn test_assign_versions_from_consensus_with_multiple_withdraws_and_settlements() {
        // Test with multiple withdrawals and multiple settlements, with settlement as the last transaction
        let mut ctx = WithdrawTestContext::new().await;

        let acc_version = ctx
            .authority
            .get_object(&SUI_ACCUMULATOR_ROOT_OBJECT_ID)
            .unwrap()
            .version();

        // First withdrawal and settlement
        let withdraw_key1 = ctx.add_withdraw_transaction();
        let settlement_key1 = ctx.add_settlement_transaction();

        // Second withdrawal and settlement
        let withdraw_key2 = ctx.add_withdraw_transaction();
        let settlement_key2 = ctx.add_settlement_transaction();

        // Third withdrawal and final settlement
        let withdraw_key3 = ctx.add_withdraw_transaction();
        let settlement_key3 = ctx.add_settlement_transaction();

        let assigned_versions = ctx.assign_versions_from_consensus();
        assert_eq!(
            assigned_versions,
            ConsensusSharedObjVerAssignment {
                assigned_versions: AssignedTxAndVersions::new(vec![
                    (
                        withdraw_key1,
                        assigned_versions_for_testing(vec![], Some(acc_version), None)
                    ),
                    (
                        settlement_key1,
                        assigned_versions_for_testing(
                            vec![((SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version), acc_version)],
                            Some(acc_version),
                            None
                        )
                    ),
                    (
                        withdraw_key2,
                        assigned_versions_for_testing(vec![], Some(acc_version.next()), None)
                    ),
                    (
                        settlement_key2,
                        assigned_versions_for_testing(
                            vec![(
                                (SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version),
                                acc_version.next()
                            )],
                            Some(acc_version.next()),
                            None
                        )
                    ),
                    (
                        withdraw_key3,
                        assigned_versions_for_testing(
                            vec![],
                            Some(acc_version.next().next()),
                            None
                        )
                    ),
                    (
                        settlement_key3,
                        assigned_versions_for_testing(
                            vec![(
                                (SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version),
                                acc_version.next().next()
                            )],
                            Some(acc_version.next().next()),
                            None
                        )
                    ),
                ]),
                shared_input_next_versions: HashMap::from([(
                    (SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version),
                    acc_version.next().next().next()
                )]),
            }
        );
    }

    #[tokio::test]
    async fn test_assign_versions_from_consensus_with_withdraw_and_shared_object() {
        // Test that a transaction can have both a withdrawal and use a shared object
        let mut ctx = WithdrawTestContext::new().await;

        // Get the shared object info from the context
        let shared_obj_id = ctx.shared_objects[0].id();
        let shared_obj_version = ctx.shared_objects[0].owner.start_version().unwrap();

        let acc_version = ctx
            .authority
            .get_object(&SUI_ACCUMULATOR_ROOT_OBJECT_ID)
            .unwrap()
            .version();

        let withdraw_with_shared_key = ctx.add_withdraw_with_shared_object_transaction();
        let settlement_key = ctx.add_settlement_transaction();

        let assigned_versions = ctx.assign_versions_from_consensus();
        assert_eq!(
            assigned_versions,
            ConsensusSharedObjVerAssignment {
                assigned_versions: AssignedTxAndVersions::new(vec![
                    (
                        withdraw_with_shared_key,
                        assigned_versions_for_testing(
                            vec![((shared_obj_id, shared_obj_version), shared_obj_version)],
                            Some(acc_version),
                            None
                        )
                    ),
                    (
                        settlement_key,
                        assigned_versions_for_testing(
                            vec![((SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version), acc_version)],
                            Some(acc_version),
                            None
                        )
                    ),
                ]),
                shared_input_next_versions: HashMap::from([
                    (
                        (SUI_ACCUMULATOR_ROOT_OBJECT_ID, acc_version),
                        acc_version.next()
                    ),
                    (
                        (shared_obj_id, shared_obj_version),
                        shared_obj_version.next()
                    ),
                ]),
            }
        );
    }
}