sui_types/

transaction.rs

// Copyright (c) 2021, Facebook, Inc. and its affiliates
// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use super::{SUI_BRIDGE_OBJECT_ID, base_types::*, error::*};
use crate::accumulator_root::{AccumulatorObjId, AccumulatorValue};
use crate::authenticator_state::ActiveJwk;
use crate::balance::{
    BALANCE_MODULE_NAME, BALANCE_REDEEM_FUNDS_FUNCTION_NAME, BALANCE_SEND_FUNDS_FUNCTION_NAME,
    BALANCE_SPLIT_FUNCTION_NAME, BALANCE_ZERO_FUNCTION_NAME, Balance,
};
use crate::coin::{
    COIN_MODULE_NAME, INTO_BALANCE_FUNC_NAME, PUT_FUNC_NAME, REDEEM_FUNDS_FUNC_NAME,
    SEND_FUNDS_FUNC_NAME,
};
use crate::coin_reservation::{
    CoinReservationResolverTrait, ParsedDigest, ParsedObjectRefWithdrawal,
};
use crate::committee::{Committee, EpochId, ProtocolVersion};
use crate::crypto::{
    AuthoritySignInfo, AuthoritySignInfoTrait, AuthoritySignature, AuthorityStrongQuorumSignInfo,
    DefaultHash, Ed25519SuiSignature, EmptySignInfo, RandomnessRound, Signature, Signer,
    SuiSignatureInner, ToFromBytes, default_hash,
};
use crate::digests::{AdditionalConsensusStateDigest, CertificateDigest, SenderSignedDataDigest};
use crate::digests::{ChainIdentifier, ConsensusCommitDigest, ZKLoginInputsDigest};
use crate::execution::{ExecutionTimeObservationKey, SharedInput};
use crate::funds_accumulator::{FUNDS_ACCUMULATOR_MODULE_NAME, WITHDRAWAL_SPLIT_FUNC_NAME};
use crate::gas_coin::GAS;
use crate::gas_model::gas_predicates::check_for_gas_price_too_high;
use crate::gas_model::gas_v2::SuiCostTable;
use crate::message_envelope::{Envelope, Message, TrustedEnvelope, VerifiedEnvelope};
use crate::messages_checkpoint::CheckpointTimestamp;
use crate::messages_consensus::{
    ConsensusCommitPrologue, ConsensusCommitPrologueV2, ConsensusCommitPrologueV3,
    ConsensusCommitPrologueV4, ConsensusDeterminedVersionAssignments,
};
use crate::object::{MoveObject, Object, Owner};
use crate::programmable_transaction_builder::ProgrammableTransactionBuilder;
use crate::signature::{GenericSignature, VerifyParams};
use crate::signature_verification::{
    VerifiedDigestCache, verify_sender_signed_data_message_signatures,
};
use crate::type_input::TypeInput;
use crate::{
    SUI_ACCUMULATOR_ROOT_OBJECT_ID, SUI_AUTHENTICATOR_STATE_OBJECT_ID, SUI_CLOCK_OBJECT_ID,
    SUI_CLOCK_OBJECT_SHARED_VERSION, SUI_FRAMEWORK_ADDRESS, SUI_FRAMEWORK_PACKAGE_ID,
    SUI_RANDOMNESS_STATE_OBJECT_ID, SUI_SYSTEM_STATE_OBJECT_ID,
    SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
};
use enum_dispatch::enum_dispatch;
use fastcrypto::{encoding::Base64, hash::HashFunction};
use itertools::{Either, Itertools};
use move_core_types::account_address::AccountAddress;
use move_core_types::identifier::IdentStr;
use move_core_types::{ident_str, identifier};
use move_core_types::{identifier::Identifier, language_storage::TypeTag};
use mysten_common::{assert_reachable, debug_fatal};
use nonempty::{NonEmpty, nonempty};
use serde::{Deserialize, Serialize};
use shared_crypto::intent::{Intent, IntentMessage, IntentScope};
use std::fmt::Write;
use std::fmt::{Debug, Display, Formatter};
use std::sync::Arc;
use std::sync::RwLock;
use std::time::Duration;
use std::{
    collections::{BTreeMap, BTreeSet, HashSet},
    hash::Hash,
    iter,
};
use strum::IntoStaticStr;
use sui_protocol_config::{PerObjectCongestionControlMode, ProtocolConfig};
use tap::Pipe;
use tracing::trace;

#[cfg(test)]
#[path = "unit_tests/transaction_serialization_tests.rs"]
mod transaction_serialization_tests;

pub const TEST_ONLY_GAS_UNIT_FOR_TRANSFER: u64 = 10_000;
pub const TEST_ONLY_GAS_UNIT_FOR_OBJECT_BASICS: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_PUBLISH: u64 = 70_000;
pub const TEST_ONLY_GAS_UNIT_FOR_STAKING: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_GENERIC: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_SPLIT_COIN: u64 = 10_000;
// For some transactions we may either perform heavy operations or touch
// objects that are storage expensive. That may happen (and often is the case)
// because the object touched are set up in genesis and carry no storage cost
// (and thus rebate) on first usage.
pub const TEST_ONLY_GAS_UNIT_FOR_HEAVY_COMPUTATION_STORAGE: u64 = 5_000_000;

pub const GAS_PRICE_FOR_SYSTEM_TX: u64 = 1;

pub const DEFAULT_VALIDATOR_GAS_PRICE: u64 = 1000;

const BLOCKED_MOVE_FUNCTIONS: [(ObjectID, &str, &str); 0] = [];

#[cfg(test)]
#[path = "unit_tests/messages_tests.rs"]
mod messages_tests;

#[cfg(test)]
#[path = "unit_tests/balance_withdraw_tests.rs"]
mod balance_withdraw_tests;

#[cfg(test)]
#[path = "unit_tests/address_balance_gas_tests.rs"]
mod address_balance_gas_tests;

#[cfg(test)]
#[path = "unit_tests/transaction_claims_tests.rs"]
mod transaction_claims_tests;

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum CallArg {
    // contains no structs or objects
    Pure(Vec<u8>),
    // an object
    Object(ObjectArg),
    // Reservation to withdraw balance from a funds a accumulator. This will be converted into a
    // `sui::funds_accumulator::Withdrawal` struct and passed into Move.
    // It is allowed to have multiple withdraw arguments even for the same funds type.
    FundsWithdrawal(FundsWithdrawalArg),
}

impl CallArg {
    pub const SUI_SYSTEM_MUT: Self = Self::Object(ObjectArg::SUI_SYSTEM_MUT);
    pub const CLOCK_IMM: Self = Self::Object(ObjectArg::SharedObject {
        id: SUI_CLOCK_OBJECT_ID,
        initial_shared_version: SUI_CLOCK_OBJECT_SHARED_VERSION,
        mutability: SharedObjectMutability::Immutable,
    });
    pub const CLOCK_MUT: Self = Self::Object(ObjectArg::SharedObject {
        id: SUI_CLOCK_OBJECT_ID,
        initial_shared_version: SUI_CLOCK_OBJECT_SHARED_VERSION,
        mutability: SharedObjectMutability::Mutable,
    });
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
pub enum ObjectArg {
    // A Move object from fastpath.
    ImmOrOwnedObject(ObjectRef),
    // A Move object from consensus (historically consensus objects were always shared).
    // SharedObject::mutable controls whether caller asks for a mutable reference to shared object.
    SharedObject {
        id: ObjectID,
        initial_shared_version: SequenceNumber,
        // Note: this used to be a bool, but because true/false encode to 0x00/0x01, we are able to
        // be backward compatible.
        mutability: SharedObjectMutability,
    },
    // A Move object that can be received in this transaction.
    Receiving(ObjectRef),
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum Reservation {
    // Reserve a specific amount of the balance.
    MaxAmountU64(u64),
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum WithdrawalTypeArg {
    Balance(TypeTag),
}

impl WithdrawalTypeArg {
    /// Convert the withdrawal type argument to a full type tag,
    /// e.g. `Balance<T>` -> `0x2::balance::Balance<T>`
    pub fn to_type_tag(&self) -> TypeTag {
        let WithdrawalTypeArg::Balance(type_param) = self;
        Balance::type_tag(type_param.clone())
    }

    /// If this is a Balance accumulator, return the type parameter of `Balance<T>`,
    /// e.g. `Balance<T>` -> `Some(T)`
    /// Otherwise, return `None`. This is not possible today, but in the future we will support other types of accumulators.
    pub fn get_balance_type_param(&self) -> Option<TypeTag> {
        let WithdrawalTypeArg::Balance(type_param) = self;
        Some(type_param.clone())
    }
}

// TODO(address-balances): Rename all the related structs and enums.
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct FundsWithdrawalArg {
    /// The reservation of the funds accumulator to withdraw.
    pub reservation: Reservation,
    /// The type argument of the funds accumulator to withdraw, e.g. `Balance<_>`.
    pub type_arg: WithdrawalTypeArg,
    /// The source of the funds to withdraw.
    pub withdraw_from: WithdrawFrom,
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum WithdrawFrom {
    /// Withdraw from the sender of the transaction.
    Sender,
    /// Withdraw from the sponsor of the transaction (gas owner).
    Sponsor,
    // TODO(address-balances): Add more options here, such as multi-party withdraws.
}

impl FundsWithdrawalArg {
    /// Withdraws from `Balance<balance_type>` in the sender's address.
    pub fn balance_from_sender(amount: u64, balance_type: TypeTag) -> Self {
        Self {
            reservation: Reservation::MaxAmountU64(amount),
            type_arg: WithdrawalTypeArg::Balance(balance_type),
            withdraw_from: WithdrawFrom::Sender,
        }
    }

    /// Withdraws from `Balance<balance_type>` in the sponsor's address (gas owner).
    pub fn balance_from_sponsor(amount: u64, balance_type: TypeTag) -> Self {
        Self {
            reservation: Reservation::MaxAmountU64(amount),
            type_arg: WithdrawalTypeArg::Balance(balance_type),
            withdraw_from: WithdrawFrom::Sponsor,
        }
    }

    pub fn owner_for_withdrawal(&self, tx: &impl TransactionDataAPI) -> SuiAddress {
        match self.withdraw_from {
            WithdrawFrom::Sender => tx.sender(),
            WithdrawFrom::Sponsor => tx.gas_owner(),
        }
    }
}

fn type_input_validity_check(
    tag: &TypeInput,
    config: &ProtocolConfig,
    starting_count: &mut usize,
) -> UserInputResult<()> {
    let mut stack = vec![(tag, 1)];
    while let Some((tag, depth)) = stack.pop() {
        *starting_count += 1;
        fp_ensure!(
            *starting_count < config.max_type_arguments() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum type arguments in a call transaction".to_string(),
                value: config.max_type_arguments().to_string()
            }
        );
        fp_ensure!(
            depth < config.max_type_argument_depth(),
            UserInputError::SizeLimitExceeded {
                limit: "maximum type argument depth in a call transaction".to_string(),
                value: config.max_type_argument_depth().to_string()
            }
        );
        match tag {
            TypeInput::Bool
            | TypeInput::U8
            | TypeInput::U64
            | TypeInput::U128
            | TypeInput::Address
            | TypeInput::Signer
            | TypeInput::U16
            | TypeInput::U32
            | TypeInput::U256 => (),
            TypeInput::Vector(t) => {
                stack.push((t, depth + 1));
            }
            TypeInput::Struct(s) => {
                let next_depth = depth + 1;
                if config.validate_identifier_inputs() {
                    fp_ensure!(
                        identifier::is_valid(&s.module),
                        UserInputError::InvalidIdentifier {
                            error: s.module.clone()
                        }
                    );
                    fp_ensure!(
                        identifier::is_valid(&s.name),
                        UserInputError::InvalidIdentifier {
                            error: s.name.clone()
                        }
                    );
                }
                stack.extend(s.type_params.iter().map(|t| (t, next_depth)));
            }
        }
    }
    Ok(())
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct ChangeEpoch {
    /// The next (to become) epoch ID.
    pub epoch: EpochId,
    /// The protocol version in effect in the new epoch.
    pub protocol_version: ProtocolVersion,
    /// The total amount of gas charged for storage during the epoch.
    pub storage_charge: u64,
    /// The total amount of gas charged for computation during the epoch.
    pub computation_charge: u64,
    /// The amount of storage rebate refunded to the txn senders.
    pub storage_rebate: u64,
    /// The non-refundable storage fee.
    pub non_refundable_storage_fee: u64,
    /// Unix timestamp when epoch started
    pub epoch_start_timestamp_ms: u64,
    /// System packages (specifically framework and move stdlib) that are written before the new
    /// epoch starts. This tracks framework upgrades on chain. When executing the ChangeEpoch txn,
    /// the validator must write out the modules below.  Modules are provided with the version they
    /// will be upgraded to, their modules in serialized form (which include their package ID), and
    /// a list of their transitive dependencies.
    pub system_packages: Vec<(SequenceNumber, Vec<Vec<u8>>, Vec<ObjectID>)>,
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct GenesisTransaction {
    pub objects: Vec<GenesisObject>,
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum GenesisObject {
    RawObject {
        data: crate::object::Data,
        owner: crate::object::Owner,
    },
}

impl GenesisObject {
    pub fn id(&self) -> ObjectID {
        match self {
            GenesisObject::RawObject { data, .. } => data.id(),
        }
    }
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct AuthenticatorStateExpire {
    /// expire JWKs that have a lower epoch than this
    pub min_epoch: u64,
    /// The initial version of the authenticator object that it was shared at.
    pub authenticator_obj_initial_shared_version: SequenceNumber,
}

impl AuthenticatorStateExpire {
    pub fn authenticator_obj_initial_shared_version(&self) -> SequenceNumber {
        self.authenticator_obj_initial_shared_version
    }
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub enum StoredExecutionTimeObservations {
    V1(Vec<(ExecutionTimeObservationKey, Vec<(AuthorityName, Duration)>)>),
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct WriteAccumulatorStorageCost {
    /// Contains the end-of-epoch-computed storage cost for accumulator objects.
    pub storage_cost: u64,
}

impl StoredExecutionTimeObservations {
    pub fn unwrap_v1(self) -> Vec<(ExecutionTimeObservationKey, Vec<(AuthorityName, Duration)>)> {
        match self {
            Self::V1(observations) => observations,
        }
    }

    pub fn filter_and_sort_v1<P>(&self, predicate: P, limit: usize) -> Self
    where
        P: FnMut(&&(ExecutionTimeObservationKey, Vec<(AuthorityName, Duration)>)) -> bool,
    {
        match self {
            Self::V1(observations) => Self::V1(
                observations
                    .iter()
                    .filter(predicate)
                    .sorted_by_key(|(key, _)| key)
                    .take(limit)
                    .cloned()
                    .collect(),
            ),
        }
    }

    /// Split observations into chunks of the specified size.
    /// Returns a vector of chunks, each containing up to `chunk_size` observations.
    pub fn chunk_observations(&self, chunk_size: usize) -> Vec<Self> {
        match self {
            Self::V1(observations) => {
                if chunk_size == 0 {
                    return vec![];
                }
                observations
                    .chunks(chunk_size)
                    .map(|chunk| Self::V1(chunk.to_vec()))
                    .collect()
            }
        }
    }

    /// Merge multiple chunks into a single observation set.
    /// Chunks must be provided in order and already sorted.
    pub fn merge_sorted_chunks(chunks: Vec<Self>) -> Self {
        let mut all_observations = Vec::new();

        for chunk in chunks {
            match chunk {
                Self::V1(observations) => {
                    all_observations.extend(observations);
                }
            }
        }

        Self::V1(all_observations)
    }
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct AuthenticatorStateUpdate {
    /// Epoch of the authenticator state update transaction
    pub epoch: u64,
    /// Consensus round of the authenticator state update
    pub round: u64,
    /// newly active jwks
    pub new_active_jwks: Vec<ActiveJwk>,
    /// The initial version of the authenticator object that it was shared at.
    pub authenticator_obj_initial_shared_version: SequenceNumber,
    // to version this struct, do not add new fields. Instead, add a AuthenticatorStateUpdateV2 to
    // TransactionKind.
}

impl AuthenticatorStateUpdate {
    pub fn authenticator_obj_initial_shared_version(&self) -> SequenceNumber {
        self.authenticator_obj_initial_shared_version
    }
}

#[derive(Debug, Hash, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct RandomnessStateUpdate {
    /// Epoch of the randomness state update transaction
    pub epoch: u64,
    /// Randomness round of the update
    pub randomness_round: RandomnessRound,
    /// Updated random bytes
    pub random_bytes: Vec<u8>,
    /// The initial version of the randomness object that it was shared at.
    pub randomness_obj_initial_shared_version: SequenceNumber,
    // to version this struct, do not add new fields. Instead, add a RandomnessStateUpdateV2 to
    // TransactionKind.
}

impl RandomnessStateUpdate {
    pub fn randomness_obj_initial_shared_version(&self) -> SequenceNumber {
        self.randomness_obj_initial_shared_version
    }
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize, IntoStaticStr)]
pub enum TransactionKind {
    /// A transaction that allows the interleaving of native commands and Move calls
    ProgrammableTransaction(ProgrammableTransaction),
    /// A system transaction that will update epoch information on-chain.
    /// It will only ever be executed once in an epoch.
    /// The argument is the next epoch number, which is critical
    /// because it ensures that this transaction has a unique digest.
    /// This will eventually be translated to a Move call during execution.
    /// It also doesn't require/use a gas object.
    /// A validator will not sign a transaction of this kind from outside. It only
    /// signs internally during epoch changes.
    ///
    /// The ChangeEpoch enumerant is now deprecated (but the ChangeEpoch struct is still used by
    /// EndOfEpochTransaction below).
    ChangeEpoch(ChangeEpoch),
    Genesis(GenesisTransaction),
    ConsensusCommitPrologue(ConsensusCommitPrologue),
    AuthenticatorStateUpdate(AuthenticatorStateUpdate),

    /// EndOfEpochTransaction replaces ChangeEpoch with a list of transactions that are allowed to
    /// run at the end of the epoch.
    EndOfEpochTransaction(Vec<EndOfEpochTransactionKind>),

    RandomnessStateUpdate(RandomnessStateUpdate),
    // V2 ConsensusCommitPrologue also includes the digest of the current consensus output.
    ConsensusCommitPrologueV2(ConsensusCommitPrologueV2),

    ConsensusCommitPrologueV3(ConsensusCommitPrologueV3),
    ConsensusCommitPrologueV4(ConsensusCommitPrologueV4),

    /// A system transaction that is expressed as a PTB
    ProgrammableSystemTransaction(ProgrammableTransaction),
    // .. more transaction types go here
}

/// EndOfEpochTransactionKind
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize, IntoStaticStr)]
pub enum EndOfEpochTransactionKind {
    ChangeEpoch(ChangeEpoch),
    AuthenticatorStateCreate,
    AuthenticatorStateExpire(AuthenticatorStateExpire),
    RandomnessStateCreate,
    DenyListStateCreate,
    BridgeStateCreate(ChainIdentifier),
    BridgeCommitteeInit(SequenceNumber),
    StoreExecutionTimeObservations(StoredExecutionTimeObservations),
    AccumulatorRootCreate,
    CoinRegistryCreate,
    DisplayRegistryCreate,
    AddressAliasStateCreate,
    WriteAccumulatorStorageCost(WriteAccumulatorStorageCost),
}

impl EndOfEpochTransactionKind {
    pub fn new_change_epoch(
        next_epoch: EpochId,
        protocol_version: ProtocolVersion,
        storage_charge: u64,
        computation_charge: u64,
        storage_rebate: u64,
        non_refundable_storage_fee: u64,
        epoch_start_timestamp_ms: u64,
        system_packages: Vec<(SequenceNumber, Vec<Vec<u8>>, Vec<ObjectID>)>,
    ) -> Self {
        Self::ChangeEpoch(ChangeEpoch {
            epoch: next_epoch,
            protocol_version,
            storage_charge,
            computation_charge,
            storage_rebate,
            non_refundable_storage_fee,
            epoch_start_timestamp_ms,
            system_packages,
        })
    }

    pub fn new_authenticator_state_expire(
        min_epoch: u64,
        authenticator_obj_initial_shared_version: SequenceNumber,
    ) -> Self {
        Self::AuthenticatorStateExpire(AuthenticatorStateExpire {
            min_epoch,
            authenticator_obj_initial_shared_version,
        })
    }

    pub fn new_authenticator_state_create() -> Self {
        Self::AuthenticatorStateCreate
    }

    pub fn new_randomness_state_create() -> Self {
        Self::RandomnessStateCreate
    }

    pub fn new_accumulator_root_create() -> Self {
        Self::AccumulatorRootCreate
    }

    pub fn new_coin_registry_create() -> Self {
        Self::CoinRegistryCreate
    }

    pub fn new_display_registry_create() -> Self {
        Self::DisplayRegistryCreate
    }

    pub fn new_deny_list_state_create() -> Self {
        Self::DenyListStateCreate
    }

    pub fn new_address_alias_state_create() -> Self {
        Self::AddressAliasStateCreate
    }

    pub fn new_bridge_create(chain_identifier: ChainIdentifier) -> Self {
        Self::BridgeStateCreate(chain_identifier)
    }

    pub fn init_bridge_committee(bridge_shared_version: SequenceNumber) -> Self {
        Self::BridgeCommitteeInit(bridge_shared_version)
    }

    pub fn new_store_execution_time_observations(
        estimates: StoredExecutionTimeObservations,
    ) -> Self {
        Self::StoreExecutionTimeObservations(estimates)
    }

    pub fn new_write_accumulator_storage_cost(storage_cost: u64) -> Self {
        Self::WriteAccumulatorStorageCost(WriteAccumulatorStorageCost { storage_cost })
    }

    fn input_objects(&self) -> Vec<InputObjectKind> {
        match self {
            Self::ChangeEpoch(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_SYSTEM_STATE_OBJECT_ID,
                    initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::AuthenticatorStateCreate => vec![],
            Self::AuthenticatorStateExpire(expire) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_AUTHENTICATOR_STATE_OBJECT_ID,
                    initial_shared_version: expire.authenticator_obj_initial_shared_version(),
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::RandomnessStateCreate => vec![],
            Self::DenyListStateCreate => vec![],
            Self::BridgeStateCreate(_) => vec![],
            Self::BridgeCommitteeInit(bridge_version) => vec![
                InputObjectKind::SharedMoveObject {
                    id: SUI_BRIDGE_OBJECT_ID,
                    initial_shared_version: *bridge_version,
                    mutability: SharedObjectMutability::Mutable,
                },
                InputObjectKind::SharedMoveObject {
                    id: SUI_SYSTEM_STATE_OBJECT_ID,
                    initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                },
            ],
            Self::StoreExecutionTimeObservations(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_SYSTEM_STATE_OBJECT_ID,
                    initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::AccumulatorRootCreate => vec![],
            Self::CoinRegistryCreate => vec![],
            Self::DisplayRegistryCreate => vec![],
            Self::AddressAliasStateCreate => vec![],
            Self::WriteAccumulatorStorageCost(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_SYSTEM_STATE_OBJECT_ID,
                    initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
        }
    }

    fn shared_input_objects(&self) -> impl Iterator<Item = SharedInputObject> + '_ {
        match self {
            Self::ChangeEpoch(_) => {
                Either::Left(vec![SharedInputObject::SUI_SYSTEM_OBJ].into_iter())
            }
            Self::AuthenticatorStateExpire(expire) => Either::Left(
                vec![SharedInputObject {
                    id: SUI_AUTHENTICATOR_STATE_OBJECT_ID,
                    initial_shared_version: expire.authenticator_obj_initial_shared_version(),
                    mutability: SharedObjectMutability::Mutable,
                }]
                .into_iter(),
            ),
            Self::AuthenticatorStateCreate => Either::Right(iter::empty()),
            Self::RandomnessStateCreate => Either::Right(iter::empty()),
            Self::DenyListStateCreate => Either::Right(iter::empty()),
            Self::BridgeStateCreate(_) => Either::Right(iter::empty()),
            Self::BridgeCommitteeInit(bridge_version) => Either::Left(
                vec![
                    SharedInputObject {
                        id: SUI_BRIDGE_OBJECT_ID,
                        initial_shared_version: *bridge_version,
                        mutability: SharedObjectMutability::Mutable,
                    },
                    SharedInputObject::SUI_SYSTEM_OBJ,
                ]
                .into_iter(),
            ),
            Self::StoreExecutionTimeObservations(_) => {
                Either::Left(vec![SharedInputObject::SUI_SYSTEM_OBJ].into_iter())
            }
            Self::AccumulatorRootCreate => Either::Right(iter::empty()),
            Self::CoinRegistryCreate => Either::Right(iter::empty()),
            Self::DisplayRegistryCreate => Either::Right(iter::empty()),
            Self::AddressAliasStateCreate => Either::Right(iter::empty()),
            Self::WriteAccumulatorStorageCost(_) => {
                Either::Left(vec![SharedInputObject::SUI_SYSTEM_OBJ].into_iter())
            }
        }
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            Self::ChangeEpoch(_) => (),
            Self::AuthenticatorStateCreate | Self::AuthenticatorStateExpire(_) => {
                if !config.enable_jwk_consensus_updates() {
                    return Err(UserInputError::Unsupported(
                        "authenticator state updates not enabled".to_string(),
                    ));
                }
            }
            Self::RandomnessStateCreate => {
                if !config.random_beacon() {
                    return Err(UserInputError::Unsupported(
                        "random beacon not enabled".to_string(),
                    ));
                }
            }
            Self::DenyListStateCreate => {
                if !config.enable_coin_deny_list_v1() {
                    return Err(UserInputError::Unsupported(
                        "coin deny list not enabled".to_string(),
                    ));
                }
            }
            Self::BridgeStateCreate(_) => {
                if !config.enable_bridge() {
                    return Err(UserInputError::Unsupported(
                        "bridge not enabled".to_string(),
                    ));
                }
            }
            Self::BridgeCommitteeInit(_) => {
                if !config.enable_bridge() {
                    return Err(UserInputError::Unsupported(
                        "bridge not enabled".to_string(),
                    ));
                }
                if !config.should_try_to_finalize_bridge_committee() {
                    return Err(UserInputError::Unsupported(
                        "should not try to finalize committee yet".to_string(),
                    ));
                }
            }
            Self::StoreExecutionTimeObservations(_) => {
                if !matches!(
                    config.per_object_congestion_control_mode(),
                    PerObjectCongestionControlMode::ExecutionTimeEstimate(_)
                ) {
                    return Err(UserInputError::Unsupported(
                        "execution time estimation not enabled".to_string(),
                    ));
                }
            }
            Self::AccumulatorRootCreate => {
                if !config.create_root_accumulator_object() {
                    return Err(UserInputError::Unsupported(
                        "accumulators not enabled".to_string(),
                    ));
                }
            }
            Self::CoinRegistryCreate => {
                if !config.enable_coin_registry() {
                    return Err(UserInputError::Unsupported(
                        "coin registry not enabled".to_string(),
                    ));
                }
            }
            Self::DisplayRegistryCreate => {
                if !config.enable_display_registry() {
                    return Err(UserInputError::Unsupported(
                        "display registry not enabled".to_string(),
                    ));
                }
            }
            Self::AddressAliasStateCreate => {
                if !config.address_aliases() {
                    return Err(UserInputError::Unsupported(
                        "address aliases not enabled".to_string(),
                    ));
                }
            }
            Self::WriteAccumulatorStorageCost(_) => {
                if !config.enable_accumulators() {
                    return Err(UserInputError::Unsupported(
                        "accumulators not enabled".to_string(),
                    ));
                }
            }
        }
        Ok(())
    }
}

impl CallArg {
    fn input_objects(&self) -> Vec<InputObjectKind> {
        match self {
            CallArg::Pure(_) => vec![],
            CallArg::Object(ObjectArg::ImmOrOwnedObject(object_ref)) => {
                if ParsedDigest::is_coin_reservation_digest(&object_ref.2) {
                    vec![]
                } else {
                    vec![InputObjectKind::ImmOrOwnedMoveObject(*object_ref)]
                }
            }
            CallArg::Object(ObjectArg::SharedObject {
                id,
                initial_shared_version,
                mutability,
            }) => vec![InputObjectKind::SharedMoveObject {
                id: *id,
                initial_shared_version: *initial_shared_version,
                mutability: *mutability,
            }],
            // Receiving objects are not part of the input objects.
            CallArg::Object(ObjectArg::Receiving(_)) => vec![],
            // While we do read accumulator state when processing withdraws,
            // this really happened at scheduling time instead of execution time.
            // Hence we do not need to depend on the accumulator object in withdraws.
            CallArg::FundsWithdrawal(_) => vec![],
        }
    }

    fn receiving_objects(&self) -> Vec<ObjectRef> {
        match self {
            CallArg::Pure(_) => vec![],
            CallArg::Object(o) => match o {
                ObjectArg::ImmOrOwnedObject(_) => vec![],
                ObjectArg::SharedObject { .. } => vec![],
                ObjectArg::Receiving(obj_ref) => vec![*obj_ref],
            },
            CallArg::FundsWithdrawal(_) => vec![],
        }
    }

    pub fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            CallArg::Pure(p) => {
                fp_ensure!(
                    p.len() < config.max_pure_argument_size() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum pure argument size".to_string(),
                        value: config.max_pure_argument_size().to_string()
                    }
                );
            }
            CallArg::Object(o) => match o {
                ObjectArg::ImmOrOwnedObject(obj_ref)
                    if ParsedDigest::is_coin_reservation_digest(&obj_ref.2) =>
                {
                    if !config.enable_coin_reservation_obj_refs() {
                        return Err(UserInputError::Unsupported(
                            "coin reservation backward compatibility layer is not enabled"
                                .to_string(),
                        ));
                    }
                }
                ObjectArg::ImmOrOwnedObject(_) => (),
                ObjectArg::SharedObject { mutability, .. } => match mutability {
                    SharedObjectMutability::Mutable | SharedObjectMutability::Immutable => (),
                    SharedObjectMutability::NonExclusiveWrite => {
                        if !config.enable_non_exclusive_writes() {
                            return Err(UserInputError::Unsupported(
                                "User transactions cannot use SharedObjectMutability::NonExclusiveWrite".to_string(),
                            ));
                        }
                    }
                },

                ObjectArg::Receiving(_) => {
                    if !config.receiving_objects_supported() {
                        return Err(UserInputError::Unsupported(format!(
                            "receiving objects is not supported at {:?}",
                            config.version
                        )));
                    }
                }
            },
            CallArg::FundsWithdrawal(_) => {}
        }
        Ok(())
    }
}

impl From<bool> for CallArg {
    fn from(b: bool) -> Self {
        // unwrap safe because every u8 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&b).unwrap())
    }
}

impl From<u8> for CallArg {
    fn from(n: u8) -> Self {
        // unwrap safe because every u8 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&n).unwrap())
    }
}

impl From<u16> for CallArg {
    fn from(n: u16) -> Self {
        // unwrap safe because every u16 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&n).unwrap())
    }
}

impl From<u32> for CallArg {
    fn from(n: u32) -> Self {
        // unwrap safe because every u32 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&n).unwrap())
    }
}

impl From<u64> for CallArg {
    fn from(n: u64) -> Self {
        // unwrap safe because every u64 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&n).unwrap())
    }
}

impl From<u128> for CallArg {
    fn from(n: u128) -> Self {
        // unwrap safe because every u128 value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(&n).unwrap())
    }
}

impl From<&Vec<u8>> for CallArg {
    fn from(v: &Vec<u8>) -> Self {
        // unwrap safe because every vec<u8> value is BCS-serializable
        CallArg::Pure(bcs::to_bytes(v).unwrap())
    }
}

impl From<ObjectRef> for CallArg {
    fn from(obj: ObjectRef) -> Self {
        CallArg::Object(ObjectArg::ImmOrOwnedObject(obj))
    }
}

impl ObjectArg {
    pub const SUI_SYSTEM_MUT: Self = Self::SharedObject {
        id: SUI_SYSTEM_STATE_OBJECT_ID,
        initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
        mutability: SharedObjectMutability::Mutable,
    };

    pub fn id(&self) -> ObjectID {
        match self {
            ObjectArg::Receiving((id, _, _))
            | ObjectArg::ImmOrOwnedObject((id, _, _))
            | ObjectArg::SharedObject { id, .. } => *id,
        }
    }
}

// Add package IDs, `ObjectID`, for types defined in modules.
fn add_type_input_packages(packages: &mut BTreeSet<ObjectID>, type_argument: &TypeInput) {
    let mut stack = vec![type_argument];
    while let Some(cur) = stack.pop() {
        match cur {
            TypeInput::Bool
            | TypeInput::U8
            | TypeInput::U64
            | TypeInput::U128
            | TypeInput::Address
            | TypeInput::Signer
            | TypeInput::U16
            | TypeInput::U32
            | TypeInput::U256 => (),
            TypeInput::Vector(inner) => stack.push(inner),
            TypeInput::Struct(struct_tag) => {
                packages.insert(struct_tag.address.into());
                stack.extend(struct_tag.type_params.iter())
            }
        }
    }
}

/// A series of commands where the results of one command can be used in future
/// commands
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct ProgrammableTransaction {
    /// Input objects or primitive values
    pub inputs: Vec<CallArg>,
    /// The commands to be executed sequentially. A failure in any command will
    /// result in the failure of the entire transaction.
    pub commands: Vec<Command>,
}

#[cfg(feature = "testing")]
static GASLESS_TOKENS_FOR_TESTING: RwLock<Vec<(String, u64)>> = RwLock::new(Vec::new());

#[cfg(feature = "testing")]
pub fn add_gasless_token_for_testing(type_string: String, min_transfer: u64) {
    GASLESS_TOKENS_FOR_TESTING
        .write()
        .unwrap()
        .push((type_string, min_transfer));
}

#[cfg(feature = "testing")]
pub fn clear_gasless_tokens_for_testing() {
    GASLESS_TOKENS_FOR_TESTING.write().unwrap().clear();
}

impl ProgrammableTransaction {
    pub fn has_shared_inputs(&self) -> bool {
        self.inputs
            .iter()
            .any(|input| matches!(input, CallArg::Object(ObjectArg::SharedObject { .. })))
    }

    pub fn validate_gasless_transaction(&self, config: &ProtocolConfig) -> UserInputResult {
        fp_ensure!(
            !self.commands.is_empty(),
            UserInputError::Unsupported(
                "Gasless transactions must have at least one command".to_string()
            )
        );

        for input in &self.inputs {
            match input {
                CallArg::Pure(_) | CallArg::FundsWithdrawal(_) => {}
                CallArg::Object(
                    ObjectArg::ImmOrOwnedObject(_) | ObjectArg::SharedObject { .. },
                ) => {}
                CallArg::Object(ObjectArg::Receiving(_)) => {
                    return Err(UserInputError::Unsupported(
                        "Gasless transactions do not support Receiving object inputs".to_string(),
                    ));
                }
            }
        }

        let allowed_token_types = get_gasless_allowed_token_types(config);

        for command in &self.commands {
            command.validate_gasless_transaction(&allowed_token_types)?;
        }

        self.validate_gasless_inputs(config)?;

        Ok(())
    }

    fn validate_gasless_inputs(&self, config: &ProtocolConfig) -> UserInputResult {
        let mut used_inputs = vec![false; self.inputs.len()];
        for idx in self.commands.iter().flat_map(|cmd| cmd.input_arguments()) {
            if let Some(slot) = used_inputs.get_mut(idx as usize) {
                *slot = true;
            }
        }

        let max_unused_pure = config.get_gasless_max_unused_inputs();
        let max_pure_bytes = config.get_gasless_max_pure_input_bytes();
        let mut unused_pure_count = 0u64;

        for (i, input) in self.inputs.iter().enumerate() {
            let is_used = used_inputs[i];
            match input {
                CallArg::Pure(bytes) => {
                    fp_ensure!(
                        bytes.len() as u64 <= max_pure_bytes,
                        UserInputError::Unsupported(format!(
                            "Input {} has size {} bytes, but gasless transactions \
                             allow at most {} bytes per Pure input",
                            i,
                            bytes.len(),
                            max_pure_bytes
                        ))
                    );
                    if !is_used {
                        unused_pure_count += 1;
                    }
                }
                CallArg::Object(_) if !is_used => {
                    return Err(UserInputError::Unsupported(format!(
                        "Gasless transactions do not allow unused Object inputs (input {})",
                        i
                    )));
                }
                CallArg::FundsWithdrawal(_) if !is_used => {
                    return Err(UserInputError::Unsupported(format!(
                        "Gasless transactions do not allow unused FundsWithdrawal inputs (input {})",
                        i
                    )));
                }
                CallArg::Object(_) | CallArg::FundsWithdrawal(_) => {}
            }
        }

        fp_ensure!(
            unused_pure_count <= max_unused_pure,
            UserInputError::Unsupported(format!(
                "Gasless transactions allow at most {} unused Pure inputs, but found {}",
                max_unused_pure, unused_pure_count
            ))
        );

        Ok(())
    }
}

/// Caches gasless allowed token types for the most recently seen protocol version.
pub fn get_gasless_allowed_token_types(config: &ProtocolConfig) -> Arc<BTreeMap<TypeTag, u64>> {
    #[allow(clippy::type_complexity)]
    static CACHE: RwLock<Option<(u64, Arc<BTreeMap<TypeTag, u64>>)>> = RwLock::new(None);

    let version = config.version.as_u64();

    // Fast path: read lock only.
    if let Some((v, map)) = CACHE.read().unwrap().as_ref()
        && *v == version
    {
        return apply_test_token_overrides(Arc::clone(map));
    }

    // Parse from ProtocolConfig if it changed.
    let mut cache = CACHE.write().unwrap();
    if let Some((v, map)) = cache.as_ref()
        && *v == version
    {
        return apply_test_token_overrides(Arc::clone(map));
    }
    let map: BTreeMap<TypeTag, u64> = config
        .gasless_allowed_token_types()
        .iter()
        .map(|(s, min_amount)| {
            let tag: TypeTag = s
                .parse()
                .unwrap_or_else(|e| panic!("invalid gasless token type {s:?}: {e}"));
            (tag, *min_amount)
        })
        .collect();
    let arc = Arc::new(map);
    *cache = Some((version, Arc::clone(&arc)));
    apply_test_token_overrides(arc)
}

fn apply_test_token_overrides(base: Arc<BTreeMap<TypeTag, u64>>) -> Arc<BTreeMap<TypeTag, u64>> {
    #[cfg(feature = "testing")]
    {
        let overrides = GASLESS_TOKENS_FOR_TESTING.read().unwrap();
        if !overrides.is_empty() {
            let mut types = (*base).clone();
            for (s, min_transfer) in overrides.iter() {
                match s.parse() {
                    Ok(tag) => {
                        types.insert(tag, *min_transfer);
                    }
                    Err(e) => {
                        debug_fatal!("invalid gasless token override {s:?}: {e}");
                    }
                }
            }
            return Arc::new(types);
        }
    }
    base
}

/// A single command in a programmable transaction.
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum Command {
    /// A call to either an entry or a public Move function
    MoveCall(Box<ProgrammableMoveCall>),
    /// `(Vec<forall T:key+store. T>, address)`
    /// It sends n-objects to the specified address. These objects must have store
    /// (public transfer) and either the previous owner must be an address or the object must
    /// be newly created.
    TransferObjects(Vec<Argument>, Argument),
    /// `(&mut Coin<T>, Vec<u64>)` -> `Vec<Coin<T>>`
    /// It splits off some amounts into a new coins with those amounts
    SplitCoins(Argument, Vec<Argument>),
    /// `(&mut Coin<T>, Vec<Coin<T>>)`
    /// It merges n-coins into the first coin
    MergeCoins(Argument, Vec<Argument>),
    /// Publishes a Move package. It takes the package bytes and a list of the package's transitive
    /// dependencies to link against on-chain.
    Publish(Vec<Vec<u8>>, Vec<ObjectID>),
    /// `forall T: Vec<T> -> vector<T>`
    /// Given n-values of the same type, it constructs a vector. For non objects or an empty vector,
    /// the type tag must be specified.
    MakeMoveVec(Option<TypeInput>, Vec<Argument>),
    /// Upgrades a Move package
    /// Takes (in order):
    /// 1. A vector of serialized modules for the package.
    /// 2. A vector of object ids for the transitive dependencies of the new package.
    /// 3. The object ID of the package being upgraded.
    /// 4. An argument holding the `UpgradeTicket` that must have been produced from an earlier command in the same
    ///    programmable transaction.
    Upgrade(Vec<Vec<u8>>, Vec<ObjectID>, ObjectID, Argument),
}

/// An argument to a programmable transaction command
#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
pub enum Argument {
    /// The gas coin. The gas coin can only be used by-ref, except for with
    /// `TransferObjects`, which can use it by-value.
    GasCoin,
    /// One of the input objects or primitive values (from
    /// `ProgrammableTransaction` inputs)
    Input(u16),
    /// The result of another command (from `ProgrammableTransaction` commands)
    Result(u16),
    /// Like a `Result` but it accesses a nested result. Currently, the only usage
    /// of this is to access a value from a Move call with multiple return values.
    NestedResult(u16, u16),
}

/// The command for calling a Move function, either an entry function or a public
/// function (which cannot return references).
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct ProgrammableMoveCall {
    /// The package containing the module and function.
    pub package: ObjectID,
    /// The specific module in the package containing the function.
    pub module: String,
    /// The function to be called.
    pub function: String,
    /// The type arguments to the function.
    pub type_arguments: Vec<TypeInput>,
    /// The arguments to the function.
    pub arguments: Vec<Argument>,
}

impl ProgrammableMoveCall {
    fn input_objects(&self) -> Vec<InputObjectKind> {
        let ProgrammableMoveCall {
            package,
            type_arguments,
            ..
        } = self;
        let mut packages = BTreeSet::from([*package]);
        for type_argument in type_arguments {
            add_type_input_packages(&mut packages, type_argument)
        }
        packages
            .into_iter()
            .map(InputObjectKind::MovePackage)
            .collect()
    }

    pub fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        let is_blocked = BLOCKED_MOVE_FUNCTIONS.contains(&(
            self.package,
            self.module.as_str(),
            self.function.as_str(),
        ));
        fp_ensure!(!is_blocked, UserInputError::BlockedMoveFunction);
        let mut type_arguments_count = 0;
        for tag in &self.type_arguments {
            type_input_validity_check(tag, config, &mut type_arguments_count)?;
        }
        fp_ensure!(
            self.arguments.len() < config.max_arguments() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum arguments in a move call".to_string(),
                value: config.max_arguments().to_string()
            }
        );
        if config.validate_identifier_inputs() {
            fp_ensure!(
                identifier::is_valid(&self.module),
                UserInputError::InvalidIdentifier {
                    error: self.module.clone()
                }
            );
            fp_ensure!(
                identifier::is_valid(&self.function),
                UserInputError::InvalidIdentifier {
                    error: self.module.clone()
                }
            );
        }
        Ok(())
    }

    fn validate_gasless_transaction(
        &self,
        allowed_token_types: &BTreeMap<TypeTag, u64>,
    ) -> UserInputResult {
        type FunctionIdent = (AccountAddress, &'static IdentStr, &'static IdentStr);

        enum TypeArgConstraint {
            /// Type arg is the fund type directly (e.g. `send_funds<USDC>`).
            FundType,
            /// Type arg is `Balance<T>`; extract `T` as the fund type.
            BalanceType,
        }
        use TypeArgConstraint::*;

        const SUI_BALANCE_SEND_FUNDS: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            BALANCE_MODULE_NAME,
            BALANCE_SEND_FUNDS_FUNCTION_NAME,
        );
        const SUI_BALANCE_REDEEM_FUNDS: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            BALANCE_MODULE_NAME,
            BALANCE_REDEEM_FUNDS_FUNCTION_NAME,
        );
        const SUI_BALANCE_SPLIT: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            BALANCE_MODULE_NAME,
            BALANCE_SPLIT_FUNCTION_NAME,
        );
        const SUI_BALANCE_ZERO: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            BALANCE_MODULE_NAME,
            BALANCE_ZERO_FUNCTION_NAME,
        );
        const SUI_FUNDS_ACCUMULATOR_WITHDRAWAL_SPLIT: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            FUNDS_ACCUMULATOR_MODULE_NAME,
            WITHDRAWAL_SPLIT_FUNC_NAME,
        );
        const SUI_COIN_INTO_BALANCE: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            COIN_MODULE_NAME,
            INTO_BALANCE_FUNC_NAME,
        );
        const SUI_COIN_REDEEM_FUNDS: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            COIN_MODULE_NAME,
            REDEEM_FUNDS_FUNC_NAME,
        );
        const SUI_COIN_SEND_FUNDS: FunctionIdent = (
            SUI_FRAMEWORK_ADDRESS,
            COIN_MODULE_NAME,
            SEND_FUNDS_FUNC_NAME,
        );
        const SUI_COIN_PUT: FunctionIdent =
            (SUI_FRAMEWORK_ADDRESS, COIN_MODULE_NAME, PUT_FUNC_NAME);

        const GASLESS_FUNCTIONS: &[(FunctionIdent, &[Option<TypeArgConstraint>])] = &[
            (SUI_BALANCE_SEND_FUNDS, &[Some(FundType)]),
            (SUI_BALANCE_REDEEM_FUNDS, &[Some(FundType)]),
            (SUI_BALANCE_SPLIT, &[Some(FundType)]),
            (SUI_BALANCE_ZERO, &[Some(FundType)]),
            (SUI_FUNDS_ACCUMULATOR_WITHDRAWAL_SPLIT, &[Some(BalanceType)]),
            (SUI_COIN_INTO_BALANCE, &[Some(FundType)]),
            (SUI_COIN_REDEEM_FUNDS, &[Some(FundType)]),
            (SUI_COIN_SEND_FUNDS, &[Some(FundType)]),
            (SUI_COIN_PUT, &[Some(FundType)]),
        ];

        let Some((_, type_arg_constraints)) =
            GASLESS_FUNCTIONS
                .iter()
                .find(|((addr, module, function), _)| {
                    *addr == AccountAddress::from(self.package)
                        && module.as_str() == self.module
                        && function.as_str() == self.function
                })
        else {
            return Err(UserInputError::Unsupported(format!(
                "Function {}::{}::{} is not supported in gasless transactions",
                self.package, self.module, self.function
            )));
        };

        fp_ensure!(
            type_arg_constraints.len() == self.type_arguments.len(),
            UserInputError::Unsupported(format!(
                "Function {}::{}::{} requires {} type arguments, but {} were provided",
                self.package,
                self.module,
                self.function,
                type_arg_constraints.len(),
                self.type_arguments.len()
            ))
        );

        for (type_arg_constraint, type_input) in
            type_arg_constraints.iter().zip(&self.type_arguments)
        {
            let Some(type_arg_constraint) = type_arg_constraint else {
                continue;
            };
            let type_arg = type_input.to_type_tag().map_err(|e| {
                UserInputError::Unsupported(format!(
                    "Failed to parse type argument {type_input} as a type tag: {e}"
                ))
            })?;
            let fund_type = match type_arg_constraint {
                TypeArgConstraint::FundType => type_arg,
                TypeArgConstraint::BalanceType => Balance::maybe_get_balance_type_param(&type_arg)
                    .ok_or_else(|| {
                        UserInputError::Unsupported(format!(
                            "Expected a type Balance<_> but got {type_input}",
                        ))
                    })?,
            };
            fp_ensure!(
                allowed_token_types.contains_key(&fund_type),
                UserInputError::Unsupported(format!(
                    "Fund type {fund_type} is not currently allowed in gasless transactions"
                ))
            );
        }
        Ok(())
    }
}

impl Command {
    pub fn move_call(
        package: ObjectID,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        arguments: Vec<Argument>,
    ) -> Self {
        let module = module.to_string();
        let function = function.to_string();
        let type_arguments = type_arguments.into_iter().map(TypeInput::from).collect();
        Command::MoveCall(Box::new(ProgrammableMoveCall {
            package,
            module,
            function,
            type_arguments,
            arguments,
        }))
    }

    pub fn make_move_vec(ty: Option<TypeTag>, args: Vec<Argument>) -> Self {
        Command::MakeMoveVec(ty.map(TypeInput::from), args)
    }

    fn input_objects(&self) -> Vec<InputObjectKind> {
        match self {
            Command::Upgrade(_, deps, package_id, _) => deps
                .iter()
                .map(|id| InputObjectKind::MovePackage(*id))
                .chain(Some(InputObjectKind::MovePackage(*package_id)))
                .collect(),
            Command::Publish(_, deps) => deps
                .iter()
                .map(|id| InputObjectKind::MovePackage(*id))
                .collect(),
            Command::MoveCall(c) => c.input_objects(),
            Command::MakeMoveVec(Some(t), _) => {
                let mut packages = BTreeSet::new();
                add_type_input_packages(&mut packages, t);
                packages
                    .into_iter()
                    .map(InputObjectKind::MovePackage)
                    .collect()
            }
            Command::MakeMoveVec(None, _)
            | Command::TransferObjects(_, _)
            | Command::SplitCoins(_, _)
            | Command::MergeCoins(_, _) => vec![],
        }
    }

    fn non_system_packages_to_be_published(&self) -> Option<&Vec<Vec<u8>>> {
        match self {
            Command::Upgrade(v, _, _, _) => Some(v),
            Command::Publish(v, _) => Some(v),
            Command::MoveCall(_)
            | Command::TransferObjects(_, _)
            | Command::SplitCoins(_, _)
            | Command::MergeCoins(_, _)
            | Command::MakeMoveVec(_, _) => None,
        }
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            Command::MoveCall(call) => call.validity_check(config)?,
            Command::TransferObjects(args, _)
            | Command::MergeCoins(_, args)
            | Command::SplitCoins(_, args) => {
                fp_ensure!(!args.is_empty(), UserInputError::EmptyCommandInput);
                fp_ensure!(
                    args.len() < config.max_arguments() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum arguments in a programmable transaction command"
                            .to_string(),
                        value: config.max_arguments().to_string()
                    }
                );
            }
            Command::MakeMoveVec(ty_opt, args) => {
                // ty_opt.is_none() ==> !args.is_empty()
                fp_ensure!(
                    ty_opt.is_some() || !args.is_empty(),
                    UserInputError::EmptyCommandInput
                );
                if let Some(ty) = ty_opt {
                    let mut type_arguments_count = 0;
                    type_input_validity_check(ty, config, &mut type_arguments_count)?;
                }
                fp_ensure!(
                    args.len() < config.max_arguments() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum arguments in a programmable transaction command"
                            .to_string(),
                        value: config.max_arguments().to_string()
                    }
                );
            }
            Command::Publish(modules, deps) | Command::Upgrade(modules, deps, _, _) => {
                fp_ensure!(!modules.is_empty(), UserInputError::EmptyCommandInput);
                fp_ensure!(
                    modules.len() < config.max_modules_in_publish() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum modules in a programmable transaction upgrade command"
                            .to_string(),
                        value: config.max_modules_in_publish().to_string()
                    }
                );
                if let Some(max_package_dependencies) = config.max_package_dependencies_as_option()
                {
                    fp_ensure!(
                        deps.len() < max_package_dependencies as usize,
                        UserInputError::SizeLimitExceeded {
                            limit: "maximum package dependencies".to_string(),
                            value: max_package_dependencies.to_string()
                        }
                    );
                };
            }
        };
        Ok(())
    }

    fn validate_gasless_transaction(
        &self,
        allowed_token_types: &BTreeMap<TypeTag, u64>,
    ) -> UserInputResult {
        match self {
            Command::MoveCall(call) => call.validate_gasless_transaction(allowed_token_types),
            Command::MergeCoins(_, _) | Command::SplitCoins(_, _) => Ok(()),
            _ => Err(UserInputError::Unsupported(
                "Gasless transactions only support MoveCall, MergeCoins, and SplitCoins commands"
                    .to_string(),
            )),
        }
    }

    fn is_input_arg_used(&self, input_arg: u16) -> bool {
        self.is_argument_used(Argument::Input(input_arg))
    }

    pub fn is_gas_coin_used(&self) -> bool {
        self.is_argument_used(Argument::GasCoin)
    }

    pub fn is_argument_used(&self, argument: Argument) -> bool {
        self.arguments().any(|a| a == &argument)
    }

    fn input_arguments(&self) -> impl Iterator<Item = u16> + '_ {
        self.arguments().filter_map(|arg| match arg {
            Argument::Input(i) => Some(*i),
            _ => None,
        })
    }

    fn arguments(&self) -> impl Iterator<Item = &Argument> + '_ {
        let (args, single): (&[Argument], Option<&Argument>) = match self {
            Command::MoveCall(c) => (&c.arguments, None),
            Command::TransferObjects(args, arg)
            | Command::MergeCoins(arg, args)
            | Command::SplitCoins(arg, args) => (args, Some(arg)),
            Command::MakeMoveVec(_, args) => (args, None),
            Command::Upgrade(_, _, _, arg) => (&[], Some(arg)),
            Command::Publish(_, _) => (&[], None),
        };
        args.iter().chain(single)
    }
}

pub fn write_sep<T: Display>(
    f: &mut Formatter<'_>,
    items: impl IntoIterator<Item = T>,
    sep: &str,
) -> std::fmt::Result {
    let mut xs = items.into_iter();
    let Some(x) = xs.next() else {
        return Ok(());
    };
    write!(f, "{x}")?;
    for x in xs {
        write!(f, "{sep}{x}")?;
    }
    Ok(())
}

impl ProgrammableTransaction {
    pub fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let ProgrammableTransaction { inputs, commands } = self;
        let input_arg_objects = inputs
            .iter()
            .flat_map(|arg| arg.input_objects())
            .collect::<Vec<_>>();
        // all objects, not just mutable, must be unique
        let mut used = HashSet::new();
        if !input_arg_objects.iter().all(|o| used.insert(o.object_id())) {
            return Err(UserInputError::DuplicateObjectRefInput);
        }
        // do not duplicate packages referred to in commands
        let command_input_objects: BTreeSet<InputObjectKind> = commands
            .iter()
            .flat_map(|command| command.input_objects())
            .collect();
        Ok(input_arg_objects
            .into_iter()
            .chain(command_input_objects)
            .collect())
    }

    fn receiving_objects(&self) -> Vec<ObjectRef> {
        let ProgrammableTransaction { inputs, .. } = self;
        inputs
            .iter()
            .flat_map(|arg| arg.receiving_objects())
            .collect()
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        let ProgrammableTransaction { inputs, commands } = self;
        fp_ensure!(
            commands.len() < config.max_programmable_tx_commands() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum commands in a programmable transaction".to_string(),
                value: config.max_programmable_tx_commands().to_string()
            }
        );
        let total_inputs = self.input_objects()?.len() + self.receiving_objects().len();
        fp_ensure!(
            total_inputs <= config.max_input_objects() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum input + receiving objects in a transaction".to_string(),
                value: config.max_input_objects().to_string()
            }
        );
        for input in inputs {
            input.validity_check(config)?
        }
        if let Some(max_publish_commands) = config.max_publish_or_upgrade_per_ptb_as_option() {
            let publish_count = commands
                .iter()
                .filter(|c| matches!(c, Command::Publish(_, _) | Command::Upgrade(_, _, _, _)))
                .count() as u64;
            fp_ensure!(
                publish_count <= max_publish_commands,
                UserInputError::MaxPublishCountExceeded {
                    max_publish_commands,
                    publish_count,
                }
            );
        }
        for command in commands {
            command.validity_check(config)?;
        }

        // If randomness is used, it must be enabled by protocol config.
        // A command that uses Random can only be followed by TransferObjects or MergeCoins.
        if let Some(random_index) = inputs.iter().position(|obj| {
            matches!(
                obj,
                CallArg::Object(ObjectArg::SharedObject { id, .. }) if *id == SUI_RANDOMNESS_STATE_OBJECT_ID
            )
        }) {
            fp_ensure!(
                config.random_beacon(),
                UserInputError::Unsupported(
                    "randomness is not enabled on this network".to_string(),
                )
            );
            let mut used_random_object = false;
            let random_index = random_index.try_into().unwrap();
            for command in commands {
                if !used_random_object {
                    used_random_object = command.is_input_arg_used(random_index);
                } else {
                    fp_ensure!(
                        matches!(
                            command,
                            Command::TransferObjects(_, _) | Command::MergeCoins(_, _)
                        ),
                        UserInputError::PostRandomCommandRestrictions
                    );
                }
            }
        }

        Ok(())
    }

    /// Return all coin reservation object references used by the transaction inputs.
    pub fn coin_reservation_obj_refs(&self) -> impl Iterator<Item = ObjectRef> + '_ {
        self.inputs.iter().filter_map(|arg| match arg {
            CallArg::Object(ObjectArg::ImmOrOwnedObject(obj_ref))
                if ParsedDigest::is_coin_reservation_digest(&obj_ref.2) =>
            {
                Some(*obj_ref)
            }
            _ => None,
        })
    }

    pub fn shared_input_objects(&self) -> impl Iterator<Item = SharedInputObject> + '_ {
        self.inputs.iter().filter_map(|arg| match arg {
            CallArg::Pure(_)
            | CallArg::Object(ObjectArg::Receiving(_))
            | CallArg::Object(ObjectArg::ImmOrOwnedObject(_))
            | CallArg::FundsWithdrawal(_) => None,
            CallArg::Object(ObjectArg::SharedObject {
                id,
                initial_shared_version,
                mutability,
            }) => Some(SharedInputObject {
                id: *id,
                initial_shared_version: *initial_shared_version,
                mutability: *mutability,
            }),
        })
    }

    fn move_calls(&self) -> Vec<(usize, &ObjectID, &str, &str)> {
        self.commands
            .iter()
            .enumerate()
            .filter_map(|(idx, command)| match command {
                Command::MoveCall(m) => {
                    Some((idx, &m.package, m.module.as_str(), m.function.as_str()))
                }
                _ => None,
            })
            .collect()
    }

    pub fn non_system_packages_to_be_published(&self) -> impl Iterator<Item = &Vec<Vec<u8>>> + '_ {
        self.commands
            .iter()
            .filter_map(|q| q.non_system_packages_to_be_published())
    }
}

impl Display for Argument {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Argument::GasCoin => write!(f, "GasCoin"),
            Argument::Input(i) => write!(f, "Input({i})"),
            Argument::Result(i) => write!(f, "Result({i})"),
            Argument::NestedResult(i, j) => write!(f, "NestedResult({i},{j})"),
        }
    }
}

impl Display for ProgrammableMoveCall {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let ProgrammableMoveCall {
            package,
            module,
            function,
            type_arguments,
            arguments,
        } = self;
        write!(f, "{package}::{module}::{function}")?;
        if !type_arguments.is_empty() {
            write!(f, "<")?;
            write_sep(f, type_arguments, ",")?;
            write!(f, ">")?;
        }
        write!(f, "(")?;
        write_sep(f, arguments, ",")?;
        write!(f, ")")
    }
}

impl Display for Command {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Command::MoveCall(p) => {
                write!(f, "MoveCall({p})")
            }
            Command::MakeMoveVec(ty_opt, elems) => {
                write!(f, "MakeMoveVec(")?;
                if let Some(ty) = ty_opt {
                    write!(f, "Some{ty}")?;
                } else {
                    write!(f, "None")?;
                }
                write!(f, ",[")?;
                write_sep(f, elems, ",")?;
                write!(f, "])")
            }
            Command::TransferObjects(objs, addr) => {
                write!(f, "TransferObjects([")?;
                write_sep(f, objs, ",")?;
                write!(f, "],{addr})")
            }
            Command::SplitCoins(coin, amounts) => {
                write!(f, "SplitCoins({coin}")?;
                write_sep(f, amounts, ",")?;
                write!(f, ")")
            }
            Command::MergeCoins(target, coins) => {
                write!(f, "MergeCoins({target},")?;
                write_sep(f, coins, ",")?;
                write!(f, ")")
            }
            Command::Publish(_bytes, deps) => {
                write!(f, "Publish(_,")?;
                write_sep(f, deps, ",")?;
                write!(f, ")")
            }
            Command::Upgrade(_bytes, deps, current_package_id, ticket) => {
                write!(f, "Upgrade(_,")?;
                write_sep(f, deps, ",")?;
                write!(f, ", {current_package_id}")?;
                write!(f, ", {ticket}")?;
                write!(f, ")")
            }
        }
    }
}

impl Display for ProgrammableTransaction {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let ProgrammableTransaction { inputs, commands } = self;
        writeln!(f, "Inputs: {inputs:?}")?;
        writeln!(f, "Commands: [")?;
        for c in commands {
            writeln!(f, "  {c},")?;
        }
        writeln!(f, "]")
    }
}

#[derive(Debug, PartialEq, Eq)]
pub struct SharedInputObject {
    pub id: ObjectID,
    pub initial_shared_version: SequenceNumber,
    pub mutability: SharedObjectMutability,
}

impl SharedInputObject {
    pub const SUI_SYSTEM_OBJ: Self = Self {
        id: SUI_SYSTEM_STATE_OBJECT_ID,
        initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
        mutability: SharedObjectMutability::Mutable,
    };

    pub fn id(&self) -> ObjectID {
        self.id
    }

    pub fn id_and_version(&self) -> (ObjectID, SequenceNumber) {
        (self.id, self.initial_shared_version)
    }

    pub fn into_id_and_version(self) -> (ObjectID, SequenceNumber) {
        (self.id, self.initial_shared_version)
    }

    pub fn is_accessed_exclusively(&self) -> bool {
        self.mutability.is_exclusive()
    }
}

impl TransactionKind {
    /// present to make migrations to programmable transactions eaier.
    /// Will be removed
    pub fn programmable(pt: ProgrammableTransaction) -> Self {
        TransactionKind::ProgrammableTransaction(pt)
    }

    pub fn is_system_tx(&self) -> bool {
        // Keep this as an exhaustive match so that we can't forget to update it.
        match self {
            TransactionKind::ChangeEpoch(_)
            | TransactionKind::Genesis(_)
            | TransactionKind::ConsensusCommitPrologue(_)
            | TransactionKind::ConsensusCommitPrologueV2(_)
            | TransactionKind::ConsensusCommitPrologueV3(_)
            | TransactionKind::ConsensusCommitPrologueV4(_)
            | TransactionKind::AuthenticatorStateUpdate(_)
            | TransactionKind::RandomnessStateUpdate(_)
            | TransactionKind::EndOfEpochTransaction(_)
            | TransactionKind::ProgrammableSystemTransaction(_) => true,
            TransactionKind::ProgrammableTransaction(_) => false,
        }
    }

    pub fn is_end_of_epoch_tx(&self) -> bool {
        matches!(
            self,
            TransactionKind::EndOfEpochTransaction(_) | TransactionKind::ChangeEpoch(_)
        )
    }

    pub fn is_accumulator_barrier_settle_tx(&self) -> bool {
        matches!(self, TransactionKind::ProgrammableSystemTransaction(_))
            && self.shared_input_objects().any(|obj| {
                obj.id == SUI_ACCUMULATOR_ROOT_OBJECT_ID
                    && obj.mutability == SharedObjectMutability::Mutable
            })
    }

    /// If this is an accumulator barrier settlement transaction, returns its
    /// `AccumulatorSettlement` transaction key by extracting epoch and
    /// checkpoint_height from the prologue call arguments.
    pub fn accumulator_barrier_settlement_key(&self) -> Option<TransactionKey> {
        let TransactionKind::ProgrammableSystemTransaction(pt) = self else {
            return None;
        };
        let has_mutable_acc_root = pt.inputs.iter().any(|input| {
            matches!(
                input,
                CallArg::Object(ObjectArg::SharedObject {
                    id,
                    mutability: SharedObjectMutability::Mutable,
                    ..
                }) if *id == SUI_ACCUMULATOR_ROOT_OBJECT_ID
            )
        });
        if !has_mutable_acc_root {
            return None;
        }
        // The prologue embeds epoch as Input(1) and checkpoint_height as Input(2),
        // both as BCS-encoded u64 pure values.
        let epoch = pt.inputs.get(1).and_then(|arg| match arg {
            CallArg::Pure(bytes) => bcs::from_bytes::<u64>(bytes).ok(),
            _ => None,
        })?;
        let checkpoint_height = pt.inputs.get(2).and_then(|arg| match arg {
            CallArg::Pure(bytes) => bcs::from_bytes::<u64>(bytes).ok(),
            _ => None,
        })?;
        Some(TransactionKey::AccumulatorSettlement(
            epoch,
            checkpoint_height,
        ))
    }

    /// If this is advance epoch transaction, returns (total gas charged, total gas rebated).
    /// TODO: We should use GasCostSummary directly in ChangeEpoch struct, and return that
    /// directly.
    pub fn get_advance_epoch_tx_gas_summary(&self) -> Option<(u64, u64)> {
        let e = match self {
            Self::ChangeEpoch(e) => e,
            Self::EndOfEpochTransaction(txns) => {
                if let EndOfEpochTransactionKind::ChangeEpoch(e) =
                    txns.last().expect("at least one end-of-epoch txn required")
                {
                    e
                } else {
                    panic!("final end-of-epoch txn must be ChangeEpoch")
                }
            }
            _ => return None,
        };

        Some((e.computation_charge + e.storage_charge, e.storage_rebate))
    }

    /// Returns an iterator of all shared input objects used by this transaction.
    /// It covers both Call and ChangeEpoch transaction kind, because both makes Move calls.
    pub fn shared_input_objects(&self) -> impl Iterator<Item = SharedInputObject> + '_ {
        match &self {
            Self::ChangeEpoch(_) => {
                Either::Left(Either::Left(iter::once(SharedInputObject::SUI_SYSTEM_OBJ)))
            }

            Self::ConsensusCommitPrologue(_)
            | Self::ConsensusCommitPrologueV2(_)
            | Self::ConsensusCommitPrologueV3(_)
            | Self::ConsensusCommitPrologueV4(_) => {
                Either::Left(Either::Left(iter::once(SharedInputObject {
                    id: SUI_CLOCK_OBJECT_ID,
                    initial_shared_version: SUI_CLOCK_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                })))
            }
            Self::AuthenticatorStateUpdate(update) => {
                Either::Left(Either::Left(iter::once(SharedInputObject {
                    id: SUI_AUTHENTICATOR_STATE_OBJECT_ID,
                    initial_shared_version: update.authenticator_obj_initial_shared_version,
                    mutability: SharedObjectMutability::Mutable,
                })))
            }
            Self::RandomnessStateUpdate(update) => {
                Either::Left(Either::Left(iter::once(SharedInputObject {
                    id: SUI_RANDOMNESS_STATE_OBJECT_ID,
                    initial_shared_version: update.randomness_obj_initial_shared_version,
                    mutability: SharedObjectMutability::Mutable,
                })))
            }
            Self::EndOfEpochTransaction(txns) => Either::Left(Either::Right(
                txns.iter().flat_map(|txn| txn.shared_input_objects()),
            )),
            Self::ProgrammableTransaction(pt) | Self::ProgrammableSystemTransaction(pt) => {
                Either::Right(Either::Left(pt.shared_input_objects()))
            }
            Self::Genesis(_) => Either::Right(Either::Right(iter::empty())),
        }
    }

    fn move_calls(&self) -> Vec<(usize, &ObjectID, &str, &str)> {
        match &self {
            Self::ProgrammableTransaction(pt) => pt.move_calls(),
            _ => vec![],
        }
    }

    pub fn receiving_objects(&self) -> Vec<ObjectRef> {
        match &self {
            TransactionKind::ChangeEpoch(_)
            | TransactionKind::Genesis(_)
            | TransactionKind::ConsensusCommitPrologue(_)
            | TransactionKind::ConsensusCommitPrologueV2(_)
            | TransactionKind::ConsensusCommitPrologueV3(_)
            | TransactionKind::ConsensusCommitPrologueV4(_)
            | TransactionKind::AuthenticatorStateUpdate(_)
            | TransactionKind::RandomnessStateUpdate(_)
            | TransactionKind::EndOfEpochTransaction(_)
            | TransactionKind::ProgrammableSystemTransaction(_) => vec![],
            TransactionKind::ProgrammableTransaction(pt) => pt.receiving_objects(),
        }
    }

    /// Return the metadata of each of the input objects for the transaction.
    /// For a Move object, we attach the object reference;
    /// for a Move package, we provide the object id only since they never change on chain.
    /// TODO: use an iterator over references here instead of a Vec to avoid allocations.
    pub fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let input_objects = match &self {
            Self::ChangeEpoch(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_SYSTEM_STATE_OBJECT_ID,
                    initial_shared_version: SUI_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::Genesis(_) => {
                vec![]
            }
            Self::ConsensusCommitPrologue(_)
            | Self::ConsensusCommitPrologueV2(_)
            | Self::ConsensusCommitPrologueV3(_)
            | Self::ConsensusCommitPrologueV4(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_CLOCK_OBJECT_ID,
                    initial_shared_version: SUI_CLOCK_OBJECT_SHARED_VERSION,
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::AuthenticatorStateUpdate(update) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_AUTHENTICATOR_STATE_OBJECT_ID,
                    initial_shared_version: update.authenticator_obj_initial_shared_version(),
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::RandomnessStateUpdate(update) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: SUI_RANDOMNESS_STATE_OBJECT_ID,
                    initial_shared_version: update.randomness_obj_initial_shared_version(),
                    mutability: SharedObjectMutability::Mutable,
                }]
            }
            Self::EndOfEpochTransaction(txns) => {
                // Dedup since transactions may have a overlap in input objects.
                // Note: it's critical to ensure the order of inputs are deterministic.
                let before_dedup: Vec<_> =
                    txns.iter().flat_map(|txn| txn.input_objects()).collect();
                let mut has_seen = HashSet::new();
                let mut after_dedup = vec![];
                for obj in before_dedup {
                    if has_seen.insert(obj) {
                        after_dedup.push(obj);
                    }
                }
                after_dedup
            }
            Self::ProgrammableTransaction(p) | Self::ProgrammableSystemTransaction(p) => {
                return p.input_objects();
            }
        };
        // Ensure that there are no duplicate inputs. This cannot be removed because:
        // In [`AuthorityState::check_locks`], we check that there are no duplicate mutable
        // input objects, which would have made this check here unnecessary. However we
        // do plan to allow shared objects show up more than once in multiple single
        // transactions down the line. Once we have that, we need check here to make sure
        // the same shared object doesn't show up more than once in the same single
        // transaction.
        let mut used = HashSet::new();
        if !input_objects.iter().all(|o| used.insert(o.object_id())) {
            return Err(UserInputError::DuplicateObjectRefInput);
        }
        Ok(input_objects)
    }

    fn get_funds_withdrawals<'a>(&'a self) -> impl Iterator<Item = &'a FundsWithdrawalArg> + 'a {
        let TransactionKind::ProgrammableTransaction(pt) = &self else {
            return Either::Left(iter::empty());
        };
        Either::Right(pt.inputs.iter().filter_map(|input| {
            if let CallArg::FundsWithdrawal(withdraw) = input {
                Some(withdraw)
            } else {
                None
            }
        }))
    }

    pub fn get_coin_reservation_obj_refs(&self) -> impl Iterator<Item = ObjectRef> + '_ {
        let TransactionKind::ProgrammableTransaction(pt) = &self else {
            return Either::Left(iter::empty());
        };
        Either::Right(pt.coin_reservation_obj_refs())
    }

    pub fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            TransactionKind::ProgrammableTransaction(p) => p.validity_check(config)?,
            // All transactiond kinds below are assumed to be system,
            // and no validity or limit checks are performed.
            TransactionKind::ChangeEpoch(_)
            | TransactionKind::Genesis(_)
            | TransactionKind::ConsensusCommitPrologue(_) => (),
            TransactionKind::ConsensusCommitPrologueV2(_) => {
                if !config.include_consensus_digest_in_prologue() {
                    return Err(UserInputError::Unsupported(
                        "ConsensusCommitPrologueV2 is not supported".to_string(),
                    ));
                }
            }
            TransactionKind::ConsensusCommitPrologueV3(_) => {
                if !config.record_consensus_determined_version_assignments_in_prologue() {
                    return Err(UserInputError::Unsupported(
                        "ConsensusCommitPrologueV3 is not supported".to_string(),
                    ));
                }
            }
            TransactionKind::ConsensusCommitPrologueV4(_) => {
                if !config.record_additional_state_digest_in_prologue() {
                    return Err(UserInputError::Unsupported(
                        "ConsensusCommitPrologueV4 is not supported".to_string(),
                    ));
                }
            }
            TransactionKind::EndOfEpochTransaction(txns) => {
                if !config.end_of_epoch_transaction_supported() {
                    return Err(UserInputError::Unsupported(
                        "EndOfEpochTransaction is not supported".to_string(),
                    ));
                }

                for tx in txns {
                    tx.validity_check(config)?;
                }
            }

            TransactionKind::AuthenticatorStateUpdate(_) => {
                if !config.enable_jwk_consensus_updates() {
                    return Err(UserInputError::Unsupported(
                        "authenticator state updates not enabled".to_string(),
                    ));
                }
            }
            TransactionKind::RandomnessStateUpdate(_) => {
                if !config.random_beacon() {
                    return Err(UserInputError::Unsupported(
                        "randomness state updates not enabled".to_string(),
                    ));
                }
            }
            TransactionKind::ProgrammableSystemTransaction(_) => {
                if !config.enable_accumulators() {
                    return Err(UserInputError::Unsupported(
                        "accumulators not enabled".to_string(),
                    ));
                }
            }
        };
        Ok(())
    }

    /// number of commands, or 0 if it is a system transaction
    pub fn num_commands(&self) -> usize {
        match self {
            TransactionKind::ProgrammableTransaction(pt) => pt.commands.len(),
            _ => 0,
        }
    }

    pub fn iter_commands(&self) -> impl Iterator<Item = &Command> {
        match self {
            TransactionKind::ProgrammableTransaction(pt) => pt.commands.iter(),
            _ => [].iter(),
        }
    }

    /// number of transactions, or 1 if it is a system transaction
    pub fn tx_count(&self) -> usize {
        match self {
            TransactionKind::ProgrammableTransaction(pt) => pt.commands.len(),
            _ => 1,
        }
    }

    pub fn name(&self) -> &'static str {
        match self {
            Self::ChangeEpoch(_) => "ChangeEpoch",
            Self::Genesis(_) => "Genesis",
            Self::ConsensusCommitPrologue(_) => "ConsensusCommitPrologue",
            Self::ConsensusCommitPrologueV2(_) => "ConsensusCommitPrologueV2",
            Self::ConsensusCommitPrologueV3(_) => "ConsensusCommitPrologueV3",
            Self::ConsensusCommitPrologueV4(_) => "ConsensusCommitPrologueV4",
            Self::ProgrammableTransaction(_) => "ProgrammableTransaction",
            Self::ProgrammableSystemTransaction(_) => "ProgrammableSystemTransaction",
            Self::AuthenticatorStateUpdate(_) => "AuthenticatorStateUpdate",
            Self::RandomnessStateUpdate(_) => "RandomnessStateUpdate",
            Self::EndOfEpochTransaction(_) => "EndOfEpochTransaction",
        }
    }
}

impl Display for TransactionKind {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let mut writer = String::new();
        match &self {
            Self::ChangeEpoch(e) => {
                writeln!(writer, "Transaction Kind : Epoch Change")?;
                writeln!(writer, "New epoch ID : {}", e.epoch)?;
                writeln!(writer, "Storage gas reward : {}", e.storage_charge)?;
                writeln!(writer, "Computation gas reward : {}", e.computation_charge)?;
                writeln!(writer, "Storage rebate : {}", e.storage_rebate)?;
                writeln!(writer, "Timestamp : {}", e.epoch_start_timestamp_ms)?;
            }
            Self::Genesis(_) => {
                writeln!(writer, "Transaction Kind : Genesis")?;
            }
            Self::ConsensusCommitPrologue(p) => {
                writeln!(writer, "Transaction Kind : Consensus Commit Prologue")?;
                writeln!(writer, "Timestamp : {}", p.commit_timestamp_ms)?;
            }
            Self::ConsensusCommitPrologueV2(p) => {
                writeln!(writer, "Transaction Kind : Consensus Commit Prologue V2")?;
                writeln!(writer, "Timestamp : {}", p.commit_timestamp_ms)?;
                writeln!(writer, "Consensus Digest: {}", p.consensus_commit_digest)?;
            }
            Self::ConsensusCommitPrologueV3(p) => {
                writeln!(writer, "Transaction Kind : Consensus Commit Prologue V3")?;
                writeln!(writer, "Timestamp : {}", p.commit_timestamp_ms)?;
                writeln!(writer, "Consensus Digest: {}", p.consensus_commit_digest)?;
                writeln!(
                    writer,
                    "Consensus determined version assignment: {:?}",
                    p.consensus_determined_version_assignments
                )?;
            }
            Self::ConsensusCommitPrologueV4(p) => {
                writeln!(writer, "Transaction Kind : Consensus Commit Prologue V4")?;
                writeln!(writer, "Timestamp : {}", p.commit_timestamp_ms)?;
                writeln!(writer, "Consensus Digest: {}", p.consensus_commit_digest)?;
                writeln!(
                    writer,
                    "Consensus determined version assignment: {:?}",
                    p.consensus_determined_version_assignments
                )?;
                writeln!(
                    writer,
                    "Additional State Digest: {}",
                    p.additional_state_digest
                )?;
            }
            Self::ProgrammableTransaction(p) => {
                writeln!(writer, "Transaction Kind : Programmable")?;
                write!(writer, "{p}")?;
            }
            Self::ProgrammableSystemTransaction(p) => {
                writeln!(writer, "Transaction Kind : Programmable System")?;
                write!(writer, "{p}")?;
            }
            Self::AuthenticatorStateUpdate(_) => {
                writeln!(writer, "Transaction Kind : Authenticator State Update")?;
            }
            Self::RandomnessStateUpdate(_) => {
                writeln!(writer, "Transaction Kind : Randomness State Update")?;
            }
            Self::EndOfEpochTransaction(_) => {
                writeln!(writer, "Transaction Kind : End of Epoch Transaction")?;
            }
        }
        write!(f, "{}", writer)
    }
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct GasData {
    pub payment: Vec<ObjectRef>,
    pub owner: SuiAddress,
    pub price: u64,
    pub budget: u64,
}

impl GasData {
    pub fn is_unmetered(&self) -> bool {
        self.payment.len() == 1
            && self.payment[0].0 == ObjectID::ZERO
            && self.payment[0].1 == SequenceNumber::default()
            && self.payment[0].2 == ObjectDigest::MIN
    }
}

pub fn is_gas_paid_from_address_balance(
    gas_data: &GasData,
    transaction_kind: &TransactionKind,
) -> bool {
    gas_data.payment.is_empty()
        && matches!(
            transaction_kind,
            TransactionKind::ProgrammableTransaction(_)
        )
}

pub fn is_gasless_transaction(gas_data: &GasData, transaction_kind: &TransactionKind) -> bool {
    is_gas_paid_from_address_balance(gas_data, transaction_kind) && gas_data.price == 0
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
pub enum TransactionExpiration {
    /// The transaction has no expiration
    None,
    /// Validators wont sign a transaction unless the expiration Epoch
    /// is greater than or equal to the current epoch
    Epoch(EpochId),
    /// ValidDuring enables gas payments from address balances.
    ///
    /// When transactions use address balances for gas payment instead of explicit gas coins,
    /// we lose the natural transaction uniqueness and replay prevention that comes from
    /// mutation of gas coin objects.
    ///
    /// By bounding expiration and providing a nonce, validators must only retain
    /// executed digests for the maximum possible expiry range to differentiate
    /// retries from unique transactions with otherwise identical inputs.
    ValidDuring {
        /// Transaction invalid before this epoch. Must equal current epoch.
        min_epoch: Option<EpochId>,
        /// Transaction expires after this epoch. Must equal current epoch
        max_epoch: Option<EpochId>,
        /// Future support for sub-epoch timing (not yet implemented)
        min_timestamp: Option<u64>,
        /// Future support for sub-epoch timing (not yet implemented)
        max_timestamp: Option<u64>,
        /// Network identifier to prevent cross-chain replay
        chain: ChainIdentifier,
        /// User-provided uniqueness identifier to differentiate otherwise identical transactions
        nonce: u32,
    },
}

impl TransactionExpiration {
    /// Validators remember all executed transaction digests from the current and previous
    /// epoch. Therefore, ValidDuring with a one or two epoch range provides replay protection.
    /// Either the transaction is statically invalid (current epoch not within range) or the
    /// validator will remember if the transaction was already executed.
    pub fn is_replay_protected(&self) -> bool {
        matches!(self, TransactionExpiration::ValidDuring {
                min_epoch: Some(min_epoch),
                max_epoch: Some(max_epoch),
                ..
            } if *max_epoch == *min_epoch || *max_epoch == min_epoch.saturating_add(1))
    }
}

#[enum_dispatch(TransactionDataAPI)]
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub enum TransactionData {
    V1(TransactionDataV1),
    // When new variants are introduced, it is important that we check version support
    // in the validity_check function based on the protocol config.
}

#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
pub struct TransactionDataV1 {
    pub kind: TransactionKind,
    pub sender: SuiAddress,
    pub gas_data: GasData,
    pub expiration: TransactionExpiration,
}

impl TransactionData {
    pub fn as_v1(&self) -> &TransactionDataV1 {
        match self {
            TransactionData::V1(v1) => v1,
        }
    }
    fn new_system_transaction(kind: TransactionKind) -> Self {
        // assert transaction kind if a system transaction
        assert!(kind.is_system_tx());
        let sender = SuiAddress::default();
        TransactionData::V1(TransactionDataV1 {
            kind,
            sender,
            gas_data: GasData {
                price: GAS_PRICE_FOR_SYSTEM_TX,
                owner: sender,
                payment: vec![(ObjectID::ZERO, SequenceNumber::default(), ObjectDigest::MIN)],
                budget: 0,
            },
            expiration: TransactionExpiration::None,
        })
    }

    pub fn new(
        kind: TransactionKind,
        sender: SuiAddress,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        TransactionData::V1(TransactionDataV1 {
            kind,
            sender,
            gas_data: GasData {
                price: gas_price,
                owner: sender,
                payment: vec![gas_payment],
                budget: gas_budget,
            },
            expiration: TransactionExpiration::None,
        })
    }

    pub fn new_with_gas_coins(
        kind: TransactionKind,
        sender: SuiAddress,
        gas_payment: Vec<ObjectRef>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        Self::new_with_gas_coins_allow_sponsor(
            kind,
            sender,
            gas_payment,
            gas_budget,
            gas_price,
            sender,
        )
    }

    pub fn new_with_gas_coins_allow_sponsor(
        kind: TransactionKind,
        sender: SuiAddress,
        gas_payment: Vec<ObjectRef>,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: SuiAddress,
    ) -> Self {
        TransactionData::V1(TransactionDataV1 {
            kind,
            sender,
            gas_data: GasData {
                price: gas_price,
                owner: gas_sponsor,
                payment: gas_payment,
                budget: gas_budget,
            },
            expiration: TransactionExpiration::None,
        })
    }

    pub fn new_with_gas_data(kind: TransactionKind, sender: SuiAddress, gas_data: GasData) -> Self {
        TransactionData::V1(TransactionDataV1 {
            kind,
            sender,
            gas_data,
            expiration: TransactionExpiration::None,
        })
    }

    pub fn new_with_gas_data_and_expiration(
        kind: TransactionKind,
        sender: SuiAddress,
        gas_data: GasData,
        expiration: TransactionExpiration,
    ) -> Self {
        TransactionData::V1(TransactionDataV1 {
            kind,
            sender,
            gas_data,
            expiration,
        })
    }

    pub fn new_move_call(
        sender: SuiAddress,
        package: ObjectID,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: ObjectRef,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Self> {
        Self::new_move_call_with_gas_coins(
            sender,
            package,
            module,
            function,
            type_arguments,
            vec![gas_payment],
            arguments,
            gas_budget,
            gas_price,
        )
    }

    pub fn new_move_call_with_gas_coins(
        sender: SuiAddress,
        package: ObjectID,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: Vec<ObjectRef>,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Self> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.move_call(package, module, function, type_arguments, arguments)?;
            builder.finish()
        };
        Ok(Self::new_programmable(
            sender,
            gas_payment,
            pt,
            gas_budget,
            gas_price,
        ))
    }

    pub fn new_transfer(
        recipient: SuiAddress,
        full_object_ref: FullObjectRef,
        sender: SuiAddress,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.transfer_object(recipient, full_object_ref).unwrap();
            builder.finish()
        };
        Self::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    pub fn new_transfer_sui(
        recipient: SuiAddress,
        sender: SuiAddress,
        amount: Option<u64>,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        Self::new_transfer_sui_allow_sponsor(
            recipient,
            sender,
            amount,
            gas_payment,
            gas_budget,
            gas_price,
            sender,
        )
    }

    pub fn new_transfer_sui_allow_sponsor(
        recipient: SuiAddress,
        sender: SuiAddress,
        amount: Option<u64>,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: SuiAddress,
    ) -> Self {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.transfer_sui(recipient, amount);
            builder.finish()
        };
        Self::new_programmable_allow_sponsor(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
            gas_sponsor,
        )
    }

    pub fn new_pay(
        sender: SuiAddress,
        coins: Vec<ObjectRef>,
        recipients: Vec<SuiAddress>,
        amounts: Vec<u64>,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Self> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay(coins, recipients, amounts)?;
            builder.finish()
        };
        Ok(Self::new_programmable(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
        ))
    }

    pub fn new_pay_sui(
        sender: SuiAddress,
        mut coins: Vec<ObjectRef>,
        recipients: Vec<SuiAddress>,
        amounts: Vec<u64>,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Self> {
        coins.insert(0, gas_payment);
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay_sui(recipients, amounts)?;
            builder.finish()
        };
        Ok(Self::new_programmable(
            sender, coins, pt, gas_budget, gas_price,
        ))
    }

    pub fn new_pay_all_sui(
        sender: SuiAddress,
        mut coins: Vec<ObjectRef>,
        recipient: SuiAddress,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        coins.insert(0, gas_payment);
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay_all_sui(recipient);
            builder.finish()
        };
        Self::new_programmable(sender, coins, pt, gas_budget, gas_price)
    }

    pub fn new_split_coin(
        sender: SuiAddress,
        coin: ObjectRef,
        amounts: Vec<u64>,
        gas_payment: ObjectRef,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.split_coin(sender, coin, amounts);
            builder.finish()
        };
        Self::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    pub fn new_module(
        sender: SuiAddress,
        gas_payment: ObjectRef,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectID>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            let upgrade_cap = builder.publish_upgradeable(modules, dep_ids);
            builder.transfer_arg(sender, upgrade_cap);
            builder.finish()
        };
        Self::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    pub fn new_upgrade(
        sender: SuiAddress,
        gas_payment: ObjectRef,
        package_id: ObjectID,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectID>,
        (upgrade_capability, capability_owner): (ObjectRef, Owner),
        upgrade_policy: u8,
        digest: Vec<u8>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Self> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            let capability_arg = match capability_owner {
                Owner::AddressOwner(_) => ObjectArg::ImmOrOwnedObject(upgrade_capability),
                Owner::Shared {
                    initial_shared_version,
                }
                | Owner::ConsensusAddressOwner {
                    start_version: initial_shared_version,
                    ..
                } => ObjectArg::SharedObject {
                    id: upgrade_capability.0,
                    initial_shared_version,
                    mutability: SharedObjectMutability::Mutable,
                },
                Owner::Immutable => {
                    return Err(anyhow::anyhow!(
                        "Upgrade capability is stored immutably and cannot be used for upgrades"
                    ));
                }
                // If the capability is owned by an object, then the module defining the owning
                // object gets to decide how the upgrade capability should be used.
                Owner::ObjectOwner(_) => {
                    return Err(anyhow::anyhow!("Upgrade capability controlled by object"));
                }
            };
            builder.obj(capability_arg).unwrap();
            let upgrade_arg = builder.pure(upgrade_policy).unwrap();
            let digest_arg = builder.pure(digest).unwrap();
            let upgrade_ticket = builder.programmable_move_call(
                SUI_FRAMEWORK_PACKAGE_ID,
                ident_str!("package").to_owned(),
                ident_str!("authorize_upgrade").to_owned(),
                vec![],
                vec![Argument::Input(0), upgrade_arg, digest_arg],
            );
            let upgrade_receipt = builder.upgrade(package_id, upgrade_ticket, dep_ids, modules);

            builder.programmable_move_call(
                SUI_FRAMEWORK_PACKAGE_ID,
                ident_str!("package").to_owned(),
                ident_str!("commit_upgrade").to_owned(),
                vec![],
                vec![Argument::Input(0), upgrade_receipt],
            );

            builder.finish()
        };
        Ok(Self::new_programmable(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
        ))
    }

    pub fn new_programmable(
        sender: SuiAddress,
        gas_payment: Vec<ObjectRef>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
    ) -> Self {
        Self::new_programmable_allow_sponsor(sender, gas_payment, pt, gas_budget, gas_price, sender)
    }

    pub fn new_programmable_allow_sponsor(
        sender: SuiAddress,
        gas_payment: Vec<ObjectRef>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
        sponsor: SuiAddress,
    ) -> Self {
        let kind = TransactionKind::ProgrammableTransaction(pt);
        Self::new_with_gas_coins_allow_sponsor(
            kind,
            sender,
            gas_payment,
            gas_budget,
            gas_price,
            sponsor,
        )
    }

    pub fn new_programmable_with_address_balance_gas(
        sender: SuiAddress,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
        chain_identifier: ChainIdentifier,
        current_epoch: EpochId,
        nonce: u32,
    ) -> Self {
        TransactionData::V1(TransactionDataV1 {
            kind: TransactionKind::ProgrammableTransaction(pt),
            sender,
            gas_data: GasData {
                payment: vec![],
                owner: sender,
                price: gas_price,
                budget: gas_budget,
            },
            expiration: TransactionExpiration::ValidDuring {
                min_epoch: Some(current_epoch),
                max_epoch: Some(current_epoch + 1),
                min_timestamp: None,
                max_timestamp: None,
                chain: chain_identifier,
                nonce,
            },
        })
    }

    pub fn message_version(&self) -> u64 {
        match self {
            TransactionData::V1(_) => 1,
        }
    }

    pub fn execution_parts(&self) -> (TransactionKind, SuiAddress, GasData) {
        (self.kind().clone(), self.sender(), self.gas_data().clone())
    }

    pub fn uses_randomness(&self) -> bool {
        self.kind()
            .shared_input_objects()
            .any(|obj| obj.id() == SUI_RANDOMNESS_STATE_OBJECT_ID)
    }

    pub fn digest(&self) -> TransactionDigest {
        TransactionDigest::new(default_hash(self))
    }
}

#[enum_dispatch]
pub trait TransactionDataAPI {
    fn sender(&self) -> SuiAddress;

    // Note: this implies that SingleTransactionKind itself must be versioned, so that it can be
    // shared across versions. This will be easy to do since it is already an enum.
    fn kind(&self) -> &TransactionKind;

    // Used by programmable_transaction_builder
    fn kind_mut(&mut self) -> &mut TransactionKind;

    // kind is moved out of often enough that this is worth it to special case.
    fn into_kind(self) -> TransactionKind;

    /// Transaction signer and Gas owner
    fn required_signers(&self) -> NonEmpty<SuiAddress>;

    fn gas_data(&self) -> &GasData;

    fn gas_owner(&self) -> SuiAddress;

    fn gas(&self) -> &[ObjectRef];

    fn gas_price(&self) -> u64;

    fn gas_budget(&self) -> u64;

    fn expiration(&self) -> &TransactionExpiration;

    fn expiration_mut(&mut self) -> &mut TransactionExpiration;

    fn move_calls(&self) -> Vec<(usize, &ObjectID, &str, &str)>;

    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>>;

    fn shared_input_objects(&self) -> Vec<SharedInputObject>;

    fn receiving_objects(&self) -> Vec<ObjectRef>;

    // Dependency (input, package & receiving) objects that already have a version,
    // and do not require version assignment from consensus.
    // Returns move objects, package objects and receiving objects.
    fn fastpath_dependency_objects(
        &self,
    ) -> UserInputResult<(Vec<ObjectRef>, Vec<ObjectID>, Vec<ObjectRef>)>;

    /// Processes funds withdraws and returns a map from funds account object ID to (total
    /// reserved amount, type tag). This method aggregates all withdraw operations for the same
    /// account by merging their reservations. Each account object ID is derived from the type
    /// parameter of each withdraw operation.
    ///
    /// This method is used at signing time, and can reject a transaction if it contains
    /// invalid reservations.
    fn process_funds_withdrawals_for_signing(
        &self,
        chain_identifier: ChainIdentifier,
        coin_resolver: &dyn CoinReservationResolverTrait,
    ) -> UserInputResult<BTreeMap<AccumulatorObjId, (u64, TypeTag)>>;

    /// Like `process_funds_withdrawals_for_signing`, but must only be called on a certified
    /// transaction, i.e. one that is known to be valid.
    fn process_funds_withdrawals_for_execution(
        &self,
        chain_identifier: ChainIdentifier,
    ) -> BTreeMap<AccumulatorObjId, u64>;

    // A cheap way to quickly check if the transaction has funds withdraws.
    fn has_funds_withdrawals(&self) -> bool;

    fn coin_reservation_obj_refs(
        &self,
        chain_identifier: ChainIdentifier,
    ) -> Vec<ParsedObjectRefWithdrawal>;

    fn validity_check(&self, context: &TxValidityCheckContext<'_>) -> SuiResult;

    fn validity_check_no_gas_check(&self, config: &ProtocolConfig) -> UserInputResult;

    /// Check if the transaction is compliant with sponsorship.
    fn check_sponsorship(&self) -> UserInputResult;

    fn is_system_tx(&self) -> bool;
    fn is_genesis_tx(&self) -> bool;

    /// returns true if the transaction is one that is specially sequenced to run at the very end
    /// of the epoch
    fn is_end_of_epoch_tx(&self) -> bool;

    fn is_consensus_commit_prologue(&self) -> bool;

    /// Check if the transaction is sponsored (namely gas owner != sender)
    fn is_sponsored_tx(&self) -> bool;

    fn is_gas_paid_from_address_balance(&self) -> bool;

    fn is_gasless_transaction(&self) -> bool;

    fn sender_mut_for_testing(&mut self) -> &mut SuiAddress;

    fn gas_data_mut(&mut self) -> &mut GasData;

    // This should be used in testing only.
    fn expiration_mut_for_testing(&mut self) -> &mut TransactionExpiration;
}

impl TransactionDataAPI for TransactionDataV1 {
    fn sender(&self) -> SuiAddress {
        self.sender
    }

    fn kind(&self) -> &TransactionKind {
        &self.kind
    }

    fn kind_mut(&mut self) -> &mut TransactionKind {
        &mut self.kind
    }

    fn into_kind(self) -> TransactionKind {
        self.kind
    }

    /// Transaction signer and Gas owner
    fn required_signers(&self) -> NonEmpty<SuiAddress> {
        let mut signers = nonempty![self.sender];
        if self.gas_owner() != self.sender {
            signers.push(self.gas_owner());
        }
        signers
    }

    fn gas_data(&self) -> &GasData {
        &self.gas_data
    }

    fn gas_owner(&self) -> SuiAddress {
        self.gas_data.owner
    }

    fn gas(&self) -> &[ObjectRef] {
        &self.gas_data.payment
    }

    fn gas_price(&self) -> u64 {
        self.gas_data.price
    }

    fn gas_budget(&self) -> u64 {
        self.gas_data.budget
    }

    fn expiration(&self) -> &TransactionExpiration {
        &self.expiration
    }

    fn expiration_mut(&mut self) -> &mut TransactionExpiration {
        &mut self.expiration
    }

    fn move_calls(&self) -> Vec<(usize, &ObjectID, &str, &str)> {
        self.kind.move_calls()
    }

    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let mut inputs = self.kind.input_objects()?;

        if !self.kind.is_system_tx() {
            inputs.extend(
                self.gas()
                    .iter()
                    .filter(|obj_ref| !ParsedDigest::is_coin_reservation_digest(&obj_ref.2))
                    .map(|obj_ref| InputObjectKind::ImmOrOwnedMoveObject(*obj_ref)),
            );
        }
        Ok(inputs)
    }

    fn shared_input_objects(&self) -> Vec<SharedInputObject> {
        self.kind.shared_input_objects().collect()
    }

    fn receiving_objects(&self) -> Vec<ObjectRef> {
        self.kind.receiving_objects()
    }

    fn fastpath_dependency_objects(
        &self,
    ) -> UserInputResult<(Vec<ObjectRef>, Vec<ObjectID>, Vec<ObjectRef>)> {
        let mut move_objects = vec![];
        let mut packages = vec![];
        let mut receiving_objects = vec![];
        self.input_objects()?.iter().for_each(|o| match o {
            InputObjectKind::ImmOrOwnedMoveObject(object_ref) => {
                move_objects.push(*object_ref);
            }
            InputObjectKind::MovePackage(package_id) => {
                packages.push(*package_id);
            }
            InputObjectKind::SharedMoveObject { .. } => {}
        });
        self.receiving_objects().iter().for_each(|object_ref| {
            receiving_objects.push(*object_ref);
        });
        Ok((move_objects, packages, receiving_objects))
    }

    fn process_funds_withdrawals_for_signing(
        &self,
        chain_identifier: ChainIdentifier,
        coin_resolver: &dyn CoinReservationResolverTrait,
    ) -> UserInputResult<BTreeMap<AccumulatorObjId, (u64, TypeTag)>> {
        let mut withdraws: Vec<_> = self.get_funds_withdrawals().collect();

        for withdraw in self.parsed_coin_reservations(chain_identifier) {
            // During signing, use latest accumulator version (None).
            let withdrawal_arg =
                coin_resolver.resolve_funds_withdrawal(self.sender(), withdraw, None)?;
            withdraws.push(withdrawal_arg);
        }

        withdraws.extend(self.get_funds_withdrawal_for_gas_payment());

        // Accumulate all withdraws per account.
        let mut withdraw_map: BTreeMap<AccumulatorObjId, (u64, TypeTag)> = BTreeMap::new();
        for withdraw in withdraws {
            let reserved_amount = match &withdraw.reservation {
                Reservation::MaxAmountU64(amount) => {
                    assert!(*amount > 0, "verified in validity check");
                    *amount
                }
            };

            let account_address = withdraw.owner_for_withdrawal(self);
            let type_tag = withdraw.type_arg.to_type_tag();
            let account_id =
                AccumulatorValue::get_field_id(account_address, &type_tag).map_err(|e| {
                    UserInputError::InvalidWithdrawReservation {
                        error: e.to_string(),
                    }
                })?;

            let (current_amount, _) = withdraw_map
                .entry(account_id)
                .or_insert_with(|| (0, type_tag));
            *current_amount = current_amount.checked_add(reserved_amount).ok_or(
                UserInputError::InvalidWithdrawReservation {
                    error: "Balance withdraw reservation overflow".to_string(),
                },
            )?;
        }

        Ok(withdraw_map)
    }

    fn process_funds_withdrawals_for_execution(
        &self,
        chain_identifier: ChainIdentifier,
    ) -> BTreeMap<AccumulatorObjId, u64> {
        let mut withdraws: Vec<_> = self.get_funds_withdrawals().collect();
        withdraws.extend(self.get_funds_withdrawal_for_gas_payment());

        // Accumulate all withdraws per account.
        let mut withdraw_map: BTreeMap<AccumulatorObjId, u64> = BTreeMap::new();
        for withdraw in withdraws {
            let reserved_amount = match &withdraw.reservation {
                Reservation::MaxAmountU64(amount) => {
                    assert!(*amount > 0, "verified in validity check");
                    *amount
                }
            };

            let withdrawal_owner = withdraw.owner_for_withdrawal(self);

            // unwrap checked at signing time
            let account_id =
                AccumulatorValue::get_field_id(withdrawal_owner, &withdraw.type_arg.to_type_tag())
                    .unwrap();

            let value = withdraw_map.entry(account_id).or_default();
            // overflow checked at signing time
            *value = value.checked_add(reserved_amount).unwrap();
        }

        // It is not necessarily possible to construct a FundsWithdrawalArg for coin reservations, because
        // the accumulator object may not exist any more. This is okay, as the scheduler will simply
        // cancel the transaction if there are no funds available.
        for obj in self.coin_reservation_obj_refs() {
            assert_reachable!("processing coin reservation withdrawal");
            // unwrap safe because of signing time checks
            let parsed = ParsedObjectRefWithdrawal::parse(&obj, chain_identifier).unwrap();
            let value = withdraw_map
                // new_unchecked is safe because we verify that this is a valid accumulator object id
                // at signing time
                // The underlying object may have been deleted by now - this is okay. We don't need type information
                // here, we only need the accumulator object id.
                .entry(AccumulatorObjId::new_unchecked(parsed.unmasked_object_id))
                .or_default();
            // overflow checked at signing time
            *value = value.checked_add(parsed.reservation_amount()).unwrap();
        }

        withdraw_map
    }

    fn has_funds_withdrawals(&self) -> bool {
        if self.is_gas_paid_from_address_balance() && self.gas_data().budget > 0 {
            return true;
        }
        if let TransactionKind::ProgrammableTransaction(pt) = &self.kind {
            for input in &pt.inputs {
                if matches!(input, CallArg::FundsWithdrawal(_)) {
                    return true;
                }
            }
        }
        if self.coin_reservation_obj_refs().next().is_some() {
            return true;
        }
        false
    }

    fn coin_reservation_obj_refs(
        &self,
        chain_identifier: ChainIdentifier,
    ) -> Vec<ParsedObjectRefWithdrawal> {
        self.coin_reservation_obj_refs()
            .filter_map(|obj_ref| ParsedObjectRefWithdrawal::parse(&obj_ref, chain_identifier))
            .collect()
    }

    fn validity_check(&self, context: &TxValidityCheckContext<'_>) -> SuiResult {
        let config = context.config;

        // Checks to see if the transaction has expired
        match self.expiration() {
            TransactionExpiration::None => (), // always valid
            TransactionExpiration::Epoch(max_epoch) => {
                if context.epoch > *max_epoch {
                    return Err(SuiErrorKind::TransactionExpired.into());
                }
            }
            TransactionExpiration::ValidDuring {
                min_epoch,
                max_epoch,
                min_timestamp,
                max_timestamp,
                chain,
                nonce: _,
            } => {
                if min_timestamp.is_some() || max_timestamp.is_some() {
                    return Err(UserInputError::Unsupported(
                        "Timestamp-based transaction expiration is not yet supported".to_string(),
                    )
                    .into());
                }

                // Legacy behavior: If ValidDuring is present, it must have either one- or two-epoch
                // validity, even if the transaction is has other replay-protection.
                // New behavior: ValidDuring can specify any epoch range. Replay protection is enforced
                // by sui_transaction_checks::check_replay_protection.
                match (min_epoch, max_epoch) {
                    _ if config.relax_valid_during_for_owned_inputs() => (),
                    (Some(min), Some(max)) => {
                        if config.enable_multi_epoch_transaction_expiration() {
                            if !(*max == *min || *max == min.saturating_add(1)) {
                                return Err(UserInputError::Unsupported(
                                    "max_epoch must be at most min_epoch + 1".to_string(),
                                )
                                .into());
                            }
                        } else if min != max {
                            return Err(UserInputError::Unsupported(
                                "min_epoch must equal max_epoch".to_string(),
                            )
                            .into());
                        }
                    }
                    _ => {
                        return Err(UserInputError::Unsupported(
                            "Both min_epoch and max_epoch must be specified".to_string(),
                        )
                        .into());
                    }
                }

                if *chain != context.chain_identifier {
                    return Err(UserInputError::InvalidChainId {
                        provided: format!("{:?}", chain),
                        expected: format!("{:?}", context.chain_identifier),
                    }
                    .into());
                }

                if let Some(min) = min_epoch
                    && context.epoch < *min
                {
                    return Err(SuiErrorKind::TransactionExpired.into());
                }
                if let Some(max) = max_epoch
                    && context.epoch > *max
                {
                    return Err(SuiErrorKind::TransactionExpired.into());
                }
            }
        }

        if self.has_funds_withdrawals() {
            // TODO: this check is incorrect, we should only require this if there are zero owned
            // inputs
            fp_ensure!(
                !self.gas().is_empty() || config.enable_address_balance_gas_payments(),
                UserInputError::MissingGasPayment.into()
            );

            fp_ensure!(
                config.enable_accumulators(),
                UserInputError::Unsupported("Address balance withdraw is not enabled".to_string())
                    .into()
            );

            // TODO(address-balances): Use a protocol config parameter for max_withdraws.
            let max_withdraws = 10;
            let mut num_reservations = 0;

            for withdraw in self.kind.get_funds_withdrawals() {
                num_reservations += 1;
                match withdraw.withdraw_from {
                    WithdrawFrom::Sender => (),
                    WithdrawFrom::Sponsor => {
                        return Err(UserInputError::InvalidWithdrawReservation {
                            error: "Explicit sponsor withdrawals are not yet supported".to_string(),
                        }
                        .into());
                    }
                }

                match withdraw.reservation {
                    Reservation::MaxAmountU64(amount) => {
                        fp_ensure!(
                            amount > 0,
                            UserInputError::InvalidWithdrawReservation {
                                error: "Balance withdraw reservation amount must be non-zero"
                                    .to_string(),
                            }
                            .into()
                        );
                    }
                };
            }

            for parsed in self.parsed_coin_reservations(context.chain_identifier) {
                num_reservations += 1;
                // coin reservations are valid for the current and next epoch, just as transactions that
                // specify a TransactionDuring are.
                // TODO: this check can be skipped if the transaction contains any address owned inputs.
                if parsed.epoch_id() != context.epoch && parsed.epoch_id() + 1 != context.epoch {
                    return Err(SuiErrorKind::TransactionExpired.into());
                }
                if parsed.reservation_amount() == 0 {
                    return Err(UserInputError::InvalidWithdrawReservation {
                        error: "Balance withdraw reservation amount must be non-zero".to_string(),
                    }
                    .into());
                }
            }

            // Count implicit gas budget as a withdrawal when gas is paid from address balance
            if config.enable_address_balance_gas_payments()
                && self.is_gas_paid_from_address_balance()
            {
                num_reservations += 1;
            }

            fp_ensure!(
                num_reservations <= max_withdraws,
                UserInputError::InvalidWithdrawReservation {
                    error: format!(
                        "Maximum number of balance withdraw reservations is {max_withdraws}"
                    ),
                }
                .into()
            );
        }

        if config.enable_accumulators()
            && config.enable_address_balance_gas_payments()
            && self.is_gas_paid_from_address_balance()
        {
            if config.address_balance_gas_reject_gas_coin_arg()
                && let TransactionKind::ProgrammableTransaction(pt) = &self.kind
            {
                fp_ensure!(
                    !pt.commands.iter().any(|cmd| cmd.is_gas_coin_used()),
                    UserInputError::Unsupported(
                        "Argument::GasCoin is not supported with address balance gas payments"
                            .to_string(),
                    )
                    .into()
                );
            }

            let is_gasless = config.enable_gasless() && self.is_gasless_transaction();
            if config.address_balance_gas_check_rgp_at_signing() && !is_gasless {
                fp_ensure!(
                    self.gas_data.price >= context.reference_gas_price,
                    UserInputError::GasPriceUnderRGP {
                        gas_price: self.gas_data.price,
                        reference_gas_price: context.reference_gas_price,
                    }
                    .into()
                );
            }

            // Legacy behavior: when paying gas from address balance, we require ValidDuring expiration
            // even if the transaction has other replay-protected inputs.
            // New behavior: the check is done in `check_address_balance_replay_protection`, which only
            // requires two-epoch ValidDuring if there are no replay-protected inputs.
            if !config.relax_valid_during_for_owned_inputs() {
                if matches!(self.expiration(), TransactionExpiration::None) {
                    // To avoid changing error behavior unnecessarily, we flag this as a missing gas payment error
                    // instead of a missing expiration error.
                    return Err(UserInputError::MissingGasPayment.into());
                }

                if !self.expiration().is_replay_protected() {
                    return Err(UserInputError::InvalidExpiration {
                        error: "Address balance gas payments require ValidDuring expiration"
                            .to_string(),
                    }
                    .into());
                }
            }
        } else {
            fp_ensure!(
                !self.gas().is_empty(),
                UserInputError::MissingGasPayment.into()
            );
        }

        let gas_len = self.gas().len();
        let max_gas_objects = config.max_gas_payment_objects() as usize;

        let within_limit = if config.correct_gas_payment_limit_check() {
            gas_len <= max_gas_objects
        } else {
            gas_len < max_gas_objects
        };

        fp_ensure!(
            within_limit,
            UserInputError::SizeLimitExceeded {
                limit: "maximum number of gas payment objects".to_string(),
                value: config.max_gas_payment_objects().to_string()
            }
            .into()
        );

        if !config.enable_coin_reservation_obj_refs() {
            for (_, _, gas_digest) in self.gas() {
                fp_ensure!(
                    !ParsedDigest::is_coin_reservation_digest(gas_digest),
                    UserInputError::GasObjectNotOwnedObject {
                        owner: Owner::AddressOwner(self.sender)
                    }
                    .into()
                );
            }
        } else {
            // When coin reservations are enabled, validate that gas coin reservations are for SUI,
            // and that they are owned by the sender. (Sponsorship via coin reservations is not supported.)
            let sui_accumulator_id =
                *AccumulatorValue::get_field_id(self.sender, &Balance::type_tag(GAS::type_tag()))?
                    .inner();

            for gas_ref in self.gas() {
                if let Some(parsed) =
                    ParsedObjectRefWithdrawal::parse(gas_ref, context.chain_identifier)
                {
                    // Coin reservations draw from the sender's address balance, so they cannot
                    // be used in sponsored transactions where gas is paid by someone else.
                    fp_ensure!(
                        self.gas_owner() == self.sender,
                        UserInputError::GasObjectNotOwnedObject {
                            owner: Owner::AddressOwner(self.sender)
                        }
                        .into()
                    );
                    fp_ensure!(
                        parsed.unmasked_object_id == sui_accumulator_id,
                        UserInputError::GasObjectNotOwnedObject {
                            owner: Owner::AddressOwner(self.sender)
                        }
                        .into()
                    );
                }
            }
        }

        if !self.is_system_tx() {
            fp_ensure!(
                !check_for_gas_price_too_high(config.gas_model_version())
                    || self.gas_data.price < config.max_gas_price(),
                UserInputError::GasPriceTooHigh {
                    max_gas_price: config.max_gas_price(),
                }
                .into()
            );
            let cost_table = SuiCostTable::new(config, self.gas_data.price);

            fp_ensure!(
                self.gas_data.budget <= cost_table.max_gas_budget,
                UserInputError::GasBudgetTooHigh {
                    gas_budget: self.gas_data().budget,
                    max_budget: cost_table.max_gas_budget,
                }
                .into()
            );
            let is_gasless = config.enable_gasless() && self.is_gasless_transaction();
            if is_gasless {
                fp_ensure!(
                    self.gas_data.budget == 0,
                    UserInputError::Unsupported(
                        "gas_budget must be 0 for gasless transactions".to_string()
                    )
                    .into()
                );
            } else {
                fp_ensure!(
                    self.gas_data.budget >= cost_table.min_transaction_cost,
                    UserInputError::GasBudgetTooLow {
                        gas_budget: self.gas_data.budget,
                        min_budget: cost_table.min_transaction_cost,
                    }
                    .into()
                );
            }
        }

        self.validity_check_no_gas_check(config)?;
        Ok(())
    }

    // Keep all the logic for validity here, we need this for dry run where the gas
    // may not be provided and created "on the fly"
    fn validity_check_no_gas_check(&self, config: &ProtocolConfig) -> UserInputResult {
        self.kind().validity_check(config)?;

        if config.enable_gasless() && self.is_gasless_transaction() {
            let TransactionKind::ProgrammableTransaction(pt) = &self.kind else {
                debug_fatal!("gasless transaction is not a ProgrammableTransaction");
                return Err(UserInputError::Unsupported(
                    "Gasless transactions must be programmable transactions".to_string(),
                ));
            };
            pt.validate_gasless_transaction(config)?;
        }

        self.check_sponsorship()
    }

    /// Check if the transaction is sponsored (namely gas owner != sender)
    fn is_sponsored_tx(&self) -> bool {
        self.gas_owner() != self.sender
    }

    // Note: it is possible to pay gas from a coin reservation, which ultimately draws from
    // the address balance. This function still returns false in that case. In other words,
    // it indicates use of the first-class API for address balance gas payments, not the legacy API.
    fn is_gas_paid_from_address_balance(&self) -> bool {
        is_gas_paid_from_address_balance(&self.gas_data, &self.kind)
    }

    fn is_gasless_transaction(&self) -> bool {
        is_gasless_transaction(&self.gas_data, &self.kind)
    }

    /// Check if the transaction is compliant with sponsorship.
    fn check_sponsorship(&self) -> UserInputResult {
        // Not a sponsored transaction, nothing to check
        if self.gas_owner() == self.sender() {
            return Ok(());
        }
        if matches!(&self.kind, TransactionKind::ProgrammableTransaction(_)) {
            return Ok(());
        }
        Err(UserInputError::UnsupportedSponsoredTransactionKind)
    }

    fn is_end_of_epoch_tx(&self) -> bool {
        matches!(
            self.kind,
            TransactionKind::ChangeEpoch(_) | TransactionKind::EndOfEpochTransaction(_)
        )
    }

    fn is_consensus_commit_prologue(&self) -> bool {
        match &self.kind {
            TransactionKind::ConsensusCommitPrologue(_)
            | TransactionKind::ConsensusCommitPrologueV2(_)
            | TransactionKind::ConsensusCommitPrologueV3(_)
            | TransactionKind::ConsensusCommitPrologueV4(_) => true,

            TransactionKind::ProgrammableTransaction(_)
            | TransactionKind::ProgrammableSystemTransaction(_)
            | TransactionKind::ChangeEpoch(_)
            | TransactionKind::Genesis(_)
            | TransactionKind::AuthenticatorStateUpdate(_)
            | TransactionKind::EndOfEpochTransaction(_)
            | TransactionKind::RandomnessStateUpdate(_) => false,
        }
    }

    fn is_system_tx(&self) -> bool {
        self.kind.is_system_tx()
    }

    fn is_genesis_tx(&self) -> bool {
        matches!(self.kind, TransactionKind::Genesis(_))
    }

    fn sender_mut_for_testing(&mut self) -> &mut SuiAddress {
        &mut self.sender
    }

    fn gas_data_mut(&mut self) -> &mut GasData {
        &mut self.gas_data
    }

    fn expiration_mut_for_testing(&mut self) -> &mut TransactionExpiration {
        &mut self.expiration
    }
}

impl TransactionDataV1 {
    fn get_funds_withdrawal_for_gas_payment(&self) -> Option<FundsWithdrawalArg> {
        if self.is_gas_paid_from_address_balance() && self.gas_data().budget > 0 {
            Some(if self.sender() != self.gas_owner() {
                FundsWithdrawalArg::balance_from_sponsor(self.gas_data().budget, GAS::type_tag())
            } else {
                FundsWithdrawalArg::balance_from_sender(self.gas_data().budget, GAS::type_tag())
            })
        } else {
            None
        }
    }

    fn get_funds_withdrawals(&self) -> impl Iterator<Item = FundsWithdrawalArg> + '_ {
        self.kind.get_funds_withdrawals().cloned()
    }

    fn coin_reservation_obj_refs(&self) -> impl Iterator<Item = ObjectRef> {
        self.kind
            .get_coin_reservation_obj_refs()
            .chain(self.gas().iter().filter_map(|gas_ref| {
                if ParsedDigest::is_coin_reservation_digest(&gas_ref.2) {
                    Some(*gas_ref)
                } else {
                    None
                }
            }))
    }

    fn parsed_coin_reservations(
        &self,
        chain_identifier: ChainIdentifier,
    ) -> impl Iterator<Item = ParsedObjectRefWithdrawal> {
        self.coin_reservation_obj_refs().map(move |obj_ref| {
            ParsedObjectRefWithdrawal::parse(&obj_ref, chain_identifier).unwrap()
        })
    }
}

pub struct TxValidityCheckContext<'a> {
    pub config: &'a ProtocolConfig,
    pub epoch: EpochId,
    pub chain_identifier: ChainIdentifier,
    pub reference_gas_price: u64,
}

impl<'a> TxValidityCheckContext<'a> {
    pub fn from_cfg_for_testing(config: &'a ProtocolConfig) -> Self {
        Self {
            config,
            epoch: 0,
            chain_identifier: ChainIdentifier::default(),
            reference_gas_price: 1000,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct SenderSignedData(SizeOneVec<SenderSignedTransaction>);

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct SenderSignedTransaction {
    pub intent_message: IntentMessage<TransactionData>,
    /// A list of signatures signed by all transaction participants.
    /// 1. non participant signature must not be present.
    /// 2. signature order does not matter.
    pub tx_signatures: Vec<GenericSignature>,
}

impl Serialize for SenderSignedTransaction {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        #[derive(Serialize)]
        #[serde(rename = "SenderSignedTransaction")]
        struct SignedTxn<'a> {
            intent_message: &'a IntentMessage<TransactionData>,
            tx_signatures: &'a Vec<GenericSignature>,
        }

        if self.intent_message().intent != Intent::sui_transaction() {
            return Err(serde::ser::Error::custom("invalid Intent for Transaction"));
        }

        let txn = SignedTxn {
            intent_message: self.intent_message(),
            tx_signatures: &self.tx_signatures,
        };
        txn.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for SenderSignedTransaction {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(rename = "SenderSignedTransaction")]
        struct SignedTxn {
            intent_message: IntentMessage<TransactionData>,
            tx_signatures: Vec<GenericSignature>,
        }

        let SignedTxn {
            intent_message,
            tx_signatures,
        } = Deserialize::deserialize(deserializer)?;

        if intent_message.intent != Intent::sui_transaction() {
            return Err(serde::de::Error::custom("invalid Intent for Transaction"));
        }

        Ok(Self {
            intent_message,
            tx_signatures,
        })
    }
}

impl SenderSignedTransaction {
    /// Returns a mapping from signer address to the signature and its index in `tx_signatures`.
    pub(crate) fn get_signer_sig_mapping(
        &self,
        verify_legacy_zklogin_address: bool,
    ) -> SuiResult<BTreeMap<SuiAddress, (u8, &GenericSignature)>> {
        let mut mapping = BTreeMap::new();
        for (idx, sig) in self.tx_signatures.iter().enumerate() {
            if verify_legacy_zklogin_address && let GenericSignature::ZkLoginAuthenticator(z) = sig
            {
                // Try deriving the address from the legacy padded way.
                mapping.insert(SuiAddress::try_from_padded(&z.inputs)?, (idx as u8, sig));
            }
            let address = sig.try_into()?;
            mapping.insert(address, (idx as u8, sig));
        }
        Ok(mapping)
    }

    pub fn intent_message(&self) -> &IntentMessage<TransactionData> {
        &self.intent_message
    }
}

impl SenderSignedData {
    pub fn new(tx_data: TransactionData, tx_signatures: Vec<GenericSignature>) -> Self {
        Self(SizeOneVec::new(SenderSignedTransaction {
            intent_message: IntentMessage::new(Intent::sui_transaction(), tx_data),
            tx_signatures,
        }))
    }

    pub fn new_from_sender_signature(tx_data: TransactionData, tx_signature: Signature) -> Self {
        Self(SizeOneVec::new(SenderSignedTransaction {
            intent_message: IntentMessage::new(Intent::sui_transaction(), tx_data),
            tx_signatures: vec![tx_signature.into()],
        }))
    }

    pub fn inner(&self) -> &SenderSignedTransaction {
        self.0.element()
    }

    pub fn into_inner(self) -> SenderSignedTransaction {
        self.0.into_inner()
    }

    pub fn inner_mut(&mut self) -> &mut SenderSignedTransaction {
        self.0.element_mut()
    }

    // This function does not check validity of the signature
    // or perform any de-dup checks.
    pub fn add_signature(&mut self, new_signature: Signature) {
        self.inner_mut().tx_signatures.push(new_signature.into());
    }

    pub(crate) fn get_signer_sig_mapping(
        &self,
        verify_legacy_zklogin_address: bool,
    ) -> SuiResult<BTreeMap<SuiAddress, (u8, &GenericSignature)>> {
        self.inner()
            .get_signer_sig_mapping(verify_legacy_zklogin_address)
    }

    pub fn transaction_data(&self) -> &TransactionData {
        &self.intent_message().value
    }

    pub fn intent_message(&self) -> &IntentMessage<TransactionData> {
        self.inner().intent_message()
    }

    pub fn tx_signatures(&self) -> &[GenericSignature] {
        &self.inner().tx_signatures
    }

    pub fn has_zklogin_sig(&self) -> bool {
        self.tx_signatures().iter().any(|sig| sig.is_zklogin())
    }

    pub fn has_upgraded_multisig(&self) -> bool {
        self.tx_signatures()
            .iter()
            .any(|sig| sig.is_upgraded_multisig())
    }

    #[cfg(test)]
    pub fn intent_message_mut_for_testing(&mut self) -> &mut IntentMessage<TransactionData> {
        &mut self.inner_mut().intent_message
    }

    // used cross-crate, so cannot be #[cfg(test)]
    pub fn tx_signatures_mut_for_testing(&mut self) -> &mut Vec<GenericSignature> {
        &mut self.inner_mut().tx_signatures
    }

    /// Includes alias_versions to ensure cache invalidation when aliases change.
    pub fn full_message_digest_with_alias_versions(
        &self,
        alias_versions: &Vec<(SuiAddress, Option<SequenceNumber>)>,
    ) -> SenderSignedDataDigest {
        let mut digest = DefaultHash::default();
        bcs::serialize_into(&mut digest, self).expect("serialization should not fail");
        bcs::serialize_into(&mut digest, alias_versions).expect("serialization should not fail");
        let hash = digest.finalize();
        SenderSignedDataDigest::new(hash.into())
    }

    pub fn serialized_size(&self) -> SuiResult<usize> {
        bcs::serialized_size(self).map_err(|e| {
            SuiErrorKind::TransactionSerializationError {
                error: e.to_string(),
            }
            .into()
        })
    }

    fn check_user_signature_protocol_compatibility(&self, config: &ProtocolConfig) -> SuiResult {
        for sig in &self.inner().tx_signatures {
            match sig {
                GenericSignature::MultiSig(_) => {
                    if !config.supports_upgraded_multisig() {
                        return Err(SuiErrorKind::UserInputError {
                            error: UserInputError::Unsupported(
                                "upgraded multisig format not enabled on this network".to_string(),
                            ),
                        }
                        .into());
                    }
                }
                GenericSignature::ZkLoginAuthenticator(_) => {
                    if !config.zklogin_auth() {
                        return Err(SuiErrorKind::UserInputError {
                            error: UserInputError::Unsupported(
                                "zklogin is not enabled on this network".to_string(),
                            ),
                        }
                        .into());
                    }
                }
                GenericSignature::PasskeyAuthenticator(_) => {
                    if !config.passkey_auth() {
                        return Err(SuiErrorKind::UserInputError {
                            error: UserInputError::Unsupported(
                                "passkey is not enabled on this network".to_string(),
                            ),
                        }
                        .into());
                    }
                }
                GenericSignature::Signature(_) | GenericSignature::MultiSigLegacy(_) => (),
            }
        }

        Ok(())
    }

    /// Validate untrusted user transaction, including its size, input count, command count, etc.
    /// Returns the certificate serialised bytes size.
    pub fn validity_check(&self, context: &TxValidityCheckContext<'_>) -> Result<usize, SuiError> {
        // Check that the features used by the user signatures are enabled on the network.
        self.check_user_signature_protocol_compatibility(context.config)?;

        // TODO: The following checks can be moved to TransactionData, if we pass context into it.

        // CRITICAL!!
        // Users cannot send system transactions.
        let tx_data = &self.transaction_data();
        fp_ensure!(
            !tx_data.is_system_tx(),
            SuiErrorKind::UserInputError {
                error: UserInputError::Unsupported(
                    "SenderSignedData must not contain system transaction".to_string()
                )
            }
            .into()
        );

        // Enforce overall transaction size limit.
        let tx_size = self.serialized_size()?;
        let max_tx_size_bytes = context.config.max_tx_size_bytes();
        fp_ensure!(
            tx_size as u64 <= max_tx_size_bytes,
            SuiErrorKind::UserInputError {
                error: UserInputError::SizeLimitExceeded {
                    limit: format!(
                        "serialized transaction size exceeded maximum of {max_tx_size_bytes}"
                    ),
                    value: tx_size.to_string(),
                }
            }
            .into()
        );

        tx_data.validity_check(context)?;

        Ok(tx_size)
    }
}

impl Message for SenderSignedData {
    type DigestType = TransactionDigest;
    const SCOPE: IntentScope = IntentScope::SenderSignedTransaction;

    /// Computes the tx digest that encodes the Rust type prefix from Signable trait.
    fn digest(&self) -> Self::DigestType {
        self.intent_message().value.digest()
    }
}

impl<S> Envelope<SenderSignedData, S> {
    pub fn sender_address(&self) -> SuiAddress {
        self.data().intent_message().value.sender()
    }

    pub fn gas_owner(&self) -> SuiAddress {
        self.data().intent_message().value.gas_owner()
    }

    pub fn gas(&self) -> &[ObjectRef] {
        self.data().intent_message().value.gas()
    }

    pub fn is_consensus_tx(&self) -> bool {
        self.transaction_data().has_funds_withdrawals()
            || self.shared_input_objects().next().is_some()
    }

    pub fn shared_input_objects(&self) -> impl Iterator<Item = SharedInputObject> + '_ {
        self.data()
            .inner()
            .intent_message
            .value
            .shared_input_objects()
            .into_iter()
    }

    // Returns the primary key for this transaction.
    pub fn key(&self) -> TransactionKey {
        match &self.data().intent_message().value.kind() {
            TransactionKind::RandomnessStateUpdate(rsu) => {
                TransactionKey::RandomnessRound(rsu.epoch, rsu.randomness_round)
            }
            _ => TransactionKey::Digest(*self.digest()),
        }
    }

    // Returns non-Digest keys that could be used to refer to this transaction.
    //
    // At the moment this returns a single Option for efficiency, but if more key types are added,
    // the return type could change to Vec<TransactionKey>.
    pub fn non_digest_key(&self) -> Option<TransactionKey> {
        match &self.data().intent_message().value.kind() {
            TransactionKind::RandomnessStateUpdate(rsu) => Some(TransactionKey::RandomnessRound(
                rsu.epoch,
                rsu.randomness_round,
            )),
            _ => None,
        }
    }

    pub fn is_system_tx(&self) -> bool {
        self.data().intent_message().value.is_system_tx()
    }

    pub fn is_sponsored_tx(&self) -> bool {
        self.data().intent_message().value.is_sponsored_tx()
    }
}

impl Transaction {
    pub fn from_data_and_signer(
        data: TransactionData,
        signers: Vec<&dyn Signer<Signature>>,
    ) -> Self {
        let signatures = {
            let intent_msg = IntentMessage::new(Intent::sui_transaction(), &data);
            signers
                .into_iter()
                .map(|s| Signature::new_secure(&intent_msg, s))
                .collect()
        };
        Self::from_data(data, signatures)
    }

    // TODO: Rename this function and above to make it clearer.
    pub fn from_data(data: TransactionData, signatures: Vec<Signature>) -> Self {
        Self::from_generic_sig_data(data, signatures.into_iter().map(|s| s.into()).collect())
    }

    pub fn signature_from_signer(
        data: TransactionData,
        intent: Intent,
        signer: &dyn Signer<Signature>,
    ) -> Signature {
        let intent_msg = IntentMessage::new(intent, data);
        Signature::new_secure(&intent_msg, signer)
    }

    pub fn from_generic_sig_data(data: TransactionData, signatures: Vec<GenericSignature>) -> Self {
        Self::new(SenderSignedData::new(data, signatures))
    }

    /// Returns the Base64 encoded tx_bytes
    /// and a list of Base64 encoded [enum GenericSignature].
    pub fn to_tx_bytes_and_signatures(&self) -> (Base64, Vec<Base64>) {
        (
            Base64::from_bytes(&bcs::to_bytes(&self.data().intent_message().value).unwrap()),
            self.data()
                .inner()
                .tx_signatures
                .iter()
                .map(|s| Base64::from_bytes(s.as_ref()))
                .collect(),
        )
    }
}

impl VerifiedTransaction {
    pub fn new_change_epoch(
        next_epoch: EpochId,
        protocol_version: ProtocolVersion,
        storage_charge: u64,
        computation_charge: u64,
        storage_rebate: u64,
        non_refundable_storage_fee: u64,
        epoch_start_timestamp_ms: u64,
        system_packages: Vec<(SequenceNumber, Vec<Vec<u8>>, Vec<ObjectID>)>,
    ) -> Self {
        ChangeEpoch {
            epoch: next_epoch,
            protocol_version,
            storage_charge,
            computation_charge,
            storage_rebate,
            non_refundable_storage_fee,
            epoch_start_timestamp_ms,
            system_packages,
        }
        .pipe(TransactionKind::ChangeEpoch)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_genesis_transaction(objects: Vec<GenesisObject>) -> Self {
        GenesisTransaction { objects }
            .pipe(TransactionKind::Genesis)
            .pipe(Self::new_system_transaction)
    }

    pub fn new_consensus_commit_prologue(
        epoch: u64,
        round: u64,
        commit_timestamp_ms: CheckpointTimestamp,
    ) -> Self {
        ConsensusCommitPrologue {
            epoch,
            round,
            commit_timestamp_ms,
        }
        .pipe(TransactionKind::ConsensusCommitPrologue)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_consensus_commit_prologue_v2(
        epoch: u64,
        round: u64,
        commit_timestamp_ms: CheckpointTimestamp,
        consensus_commit_digest: ConsensusCommitDigest,
    ) -> Self {
        ConsensusCommitPrologueV2 {
            epoch,
            round,
            commit_timestamp_ms,
            consensus_commit_digest,
        }
        .pipe(TransactionKind::ConsensusCommitPrologueV2)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_consensus_commit_prologue_v3(
        epoch: u64,
        round: u64,
        commit_timestamp_ms: CheckpointTimestamp,
        consensus_commit_digest: ConsensusCommitDigest,
        consensus_determined_version_assignments: ConsensusDeterminedVersionAssignments,
    ) -> Self {
        ConsensusCommitPrologueV3 {
            epoch,
            round,
            // sub_dag_index is reserved for when we have multi commits per round.
            sub_dag_index: None,
            commit_timestamp_ms,
            consensus_commit_digest,
            consensus_determined_version_assignments,
        }
        .pipe(TransactionKind::ConsensusCommitPrologueV3)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_consensus_commit_prologue_v4(
        epoch: u64,
        round: u64,
        commit_timestamp_ms: CheckpointTimestamp,
        consensus_commit_digest: ConsensusCommitDigest,
        consensus_determined_version_assignments: ConsensusDeterminedVersionAssignments,
        additional_state_digest: AdditionalConsensusStateDigest,
    ) -> Self {
        ConsensusCommitPrologueV4 {
            epoch,
            round,
            // sub_dag_index is reserved for when we have multi commits per round.
            sub_dag_index: None,
            commit_timestamp_ms,
            consensus_commit_digest,
            consensus_determined_version_assignments,
            additional_state_digest,
        }
        .pipe(TransactionKind::ConsensusCommitPrologueV4)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_authenticator_state_update(
        epoch: u64,
        round: u64,
        new_active_jwks: Vec<ActiveJwk>,
        authenticator_obj_initial_shared_version: SequenceNumber,
    ) -> Self {
        AuthenticatorStateUpdate {
            epoch,
            round,
            new_active_jwks,
            authenticator_obj_initial_shared_version,
        }
        .pipe(TransactionKind::AuthenticatorStateUpdate)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_randomness_state_update(
        epoch: u64,
        randomness_round: RandomnessRound,
        random_bytes: Vec<u8>,
        randomness_obj_initial_shared_version: SequenceNumber,
    ) -> Self {
        RandomnessStateUpdate {
            epoch,
            randomness_round,
            random_bytes,
            randomness_obj_initial_shared_version,
        }
        .pipe(TransactionKind::RandomnessStateUpdate)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_end_of_epoch_transaction(txns: Vec<EndOfEpochTransactionKind>) -> Self {
        TransactionKind::EndOfEpochTransaction(txns).pipe(Self::new_system_transaction)
    }

    pub fn new_system_transaction(system_transaction: TransactionKind) -> Self {
        system_transaction
            .pipe(TransactionData::new_system_transaction)
            .pipe(|data| {
                SenderSignedData::new_from_sender_signature(
                    data,
                    Ed25519SuiSignature::from_bytes(&[0; Ed25519SuiSignature::LENGTH])
                        .unwrap()
                        .into(),
                )
            })
            .pipe(Transaction::new)
            .pipe(Self::new_from_verified)
    }
}

impl VerifiedSignedTransaction {
    /// Use signing key to create a signed object.
    pub fn new(
        epoch: EpochId,
        transaction: VerifiedTransaction,
        authority: AuthorityName,
        secret: &dyn Signer<AuthoritySignature>,
    ) -> Self {
        Self::new_from_verified(SignedTransaction::new(
            epoch,
            transaction.into_inner().into_data(),
            secret,
            authority,
        ))
    }
}

/// A transaction that is signed by a sender but not yet by an authority.
pub type Transaction = Envelope<SenderSignedData, EmptySignInfo>;
pub type VerifiedTransaction = VerifiedEnvelope<SenderSignedData, EmptySignInfo>;
pub type TrustedTransaction = TrustedEnvelope<SenderSignedData, EmptySignInfo>;

/// A transaction that is signed by a sender and also by an authority.
pub type SignedTransaction = Envelope<SenderSignedData, AuthoritySignInfo>;
pub type VerifiedSignedTransaction = VerifiedEnvelope<SenderSignedData, AuthoritySignInfo>;

impl Transaction {
    pub fn verify_signature_for_testing(
        &self,
        current_epoch: EpochId,
        verify_params: &VerifyParams,
    ) -> SuiResult {
        verify_sender_signed_data_message_signatures(
            self.data(),
            current_epoch,
            verify_params,
            Arc::new(VerifiedDigestCache::new_empty()),
            vec![],
        )?;
        Ok(())
    }

    pub fn try_into_verified_for_testing(
        self,
        current_epoch: EpochId,
        verify_params: &VerifyParams,
    ) -> SuiResult<VerifiedTransaction> {
        self.verify_signature_for_testing(current_epoch, verify_params)?;
        Ok(VerifiedTransaction::new_from_verified(self))
    }
}

impl SignedTransaction {
    pub fn verify_signatures_authenticated_for_testing(
        &self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> SuiResult {
        verify_sender_signed_data_message_signatures(
            self.data(),
            committee.epoch(),
            verify_params,
            Arc::new(VerifiedDigestCache::new_empty()),
            vec![],
        )?;

        self.auth_sig().verify_secure(
            self.data(),
            Intent::sui_app(IntentScope::SenderSignedTransaction),
            committee,
        )
    }

    pub fn try_into_verified_for_testing(
        self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> SuiResult<VerifiedSignedTransaction> {
        self.verify_signatures_authenticated_for_testing(committee, verify_params)?;
        Ok(VerifiedSignedTransaction::new_from_verified(self))
    }
}

pub type CertifiedTransaction = Envelope<SenderSignedData, AuthorityStrongQuorumSignInfo>;

impl CertifiedTransaction {
    pub fn certificate_digest(&self) -> CertificateDigest {
        let mut digest = DefaultHash::default();
        bcs::serialize_into(&mut digest, self).expect("serialization should not fail");
        let hash = digest.finalize();
        CertificateDigest::new(hash.into())
    }

    pub fn gas_price(&self) -> u64 {
        self.data().transaction_data().gas_price()
    }

    // TODO: Eventually we should remove all calls to verify_signature
    // and make sure they all call verify to avoid repeated verifications.
    pub fn verify_signatures_authenticated(
        &self,
        committee: &Committee,
        verify_params: &VerifyParams,
        zklogin_inputs_cache: Arc<VerifiedDigestCache<ZKLoginInputsDigest>>,
    ) -> SuiResult {
        verify_sender_signed_data_message_signatures(
            self.data(),
            committee.epoch(),
            verify_params,
            zklogin_inputs_cache,
            vec![],
        )?;
        self.auth_sig().verify_secure(
            self.data(),
            Intent::sui_app(IntentScope::SenderSignedTransaction),
            committee,
        )
    }

    pub fn try_into_verified_for_testing(
        self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> SuiResult<VerifiedCertificate> {
        self.verify_signatures_authenticated(
            committee,
            verify_params,
            Arc::new(VerifiedDigestCache::new_empty()),
        )?;
        Ok(VerifiedCertificate::new_from_verified(self))
    }

    pub fn verify_committee_sigs_only(&self, committee: &Committee) -> SuiResult {
        self.auth_sig().verify_secure(
            self.data(),
            Intent::sui_app(IntentScope::SenderSignedTransaction),
            committee,
        )
    }
}

pub type VerifiedCertificate = VerifiedEnvelope<SenderSignedData, AuthorityStrongQuorumSignInfo>;
pub type TrustedCertificate = TrustedEnvelope<SenderSignedData, AuthorityStrongQuorumSignInfo>;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct WithAliases<T>(
    T,
    #[serde(with = "nonempty_as_vec")] NonEmpty<(u8, Option<SequenceNumber>)>,
);

impl<T> WithAliases<T> {
    pub fn new(tx: T, aliases: NonEmpty<(u8, Option<SequenceNumber>)>) -> Self {
        Self(tx, aliases)
    }

    pub fn tx(&self) -> &T {
        &self.0
    }

    pub fn aliases(&self) -> &NonEmpty<(u8, Option<SequenceNumber>)> {
        &self.1
    }

    pub fn into_tx(self) -> T {
        self.0
    }

    pub fn into_aliases(self) -> NonEmpty<(u8, Option<SequenceNumber>)> {
        self.1
    }

    pub fn into_inner(self) -> (T, NonEmpty<(u8, Option<SequenceNumber>)>) {
        (self.0, self.1)
    }
}

impl<T: Message, S> WithAliases<VerifiedEnvelope<T, S>> {
    /// Analogous to VerifiedEnvelope::serializable.
    pub fn serializable(self) -> WithAliases<TrustedEnvelope<T, S>> {
        WithAliases(self.0.serializable(), self.1)
    }
}

impl<S> WithAliases<Envelope<SenderSignedData, S>> {
    /// Creates a WithAliases where each required signer is mapped to its corresponding
    /// signature index (assuming 1:1 correspondence) with no alias object version.
    pub fn no_aliases(tx: Envelope<SenderSignedData, S>) -> Self {
        let required_signers = tx.intent_message().value.required_signers();
        assert_eq!(required_signers.len(), tx.tx_signatures().len());
        let no_aliases = required_signers
            .iter()
            .enumerate()
            .map(|(idx, _)| (idx as u8, None))
            .collect::<Vec<_>>();
        Self::new(
            tx,
            NonEmpty::from_vec(no_aliases).expect("must have at least one required_signer"),
        )
    }
}

impl<S> WithAliases<VerifiedEnvelope<SenderSignedData, S>> {
    /// Creates a WithAliases where each required signer is mapped to its corresponding
    /// signature index (assuming 1:1 correspondence) with no alias object version.
    pub fn no_aliases(tx: VerifiedEnvelope<SenderSignedData, S>) -> Self {
        let required_signers = tx.intent_message().value.required_signers();
        assert_eq!(required_signers.len(), tx.tx_signatures().len());
        let no_aliases = required_signers
            .iter()
            .enumerate()
            .map(|(idx, _)| (idx as u8, None))
            .collect::<Vec<_>>();
        Self::new(
            tx,
            NonEmpty::from_vec(no_aliases).expect("must have at least one required_signer"),
        )
    }
}

pub type TransactionWithAliases = WithAliases<Transaction>;
pub type VerifiedTransactionWithAliases = WithAliases<VerifiedTransaction>;
pub type TrustedTransactionWithAliases = WithAliases<TrustedTransaction>;

/// Deprecated version of WithAliases that uses SuiAddress instead of u8.
/// This is needed to read data from deferred_transactions_with_aliases_v2 table
/// which was written with the old format before the type was changed.
// TODO: Delete this after all production networks are on the latest table.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DeprecatedWithAliases<T>(
    T,
    #[serde(with = "nonempty_as_vec")] NonEmpty<(SuiAddress, Option<SequenceNumber>)>,
);

impl<T> DeprecatedWithAliases<T> {
    pub fn into_inner(self) -> (T, NonEmpty<(SuiAddress, Option<SequenceNumber>)>) {
        (self.0, self.1)
    }
}

impl<T: Message, S> From<WithAliases<VerifiedEnvelope<T, S>>> for WithAliases<Envelope<T, S>> {
    fn from(value: WithAliases<VerifiedEnvelope<T, S>>) -> Self {
        Self(value.0.into(), value.1)
    }
}

impl<T: Message, S> From<WithAliases<TrustedEnvelope<T, S>>>
    for WithAliases<VerifiedEnvelope<T, S>>
{
    fn from(value: WithAliases<TrustedEnvelope<T, S>>) -> Self {
        Self(value.0.into(), value.1)
    }
}

mod nonempty_as_vec {
    use super::*;
    use serde::{Deserialize, Deserializer, Serialize, Serializer};

    pub fn serialize<S, T>(value: &NonEmpty<T>, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
        T: Serialize,
    {
        let vec: Vec<&T> = value.iter().collect();
        vec.serialize(serializer)
    }

    pub fn deserialize<'de, D, T>(deserializer: D) -> Result<NonEmpty<T>, D::Error>
    where
        D: Deserializer<'de>,
        T: Deserialize<'de> + Clone,
    {
        use serde::de::{SeqAccess, Visitor};
        use std::fmt;
        use std::marker::PhantomData;

        struct NonEmptyVisitor<T>(PhantomData<T>);

        impl<'de, T> Visitor<'de> for NonEmptyVisitor<T>
        where
            T: Deserialize<'de> + Clone,
        {
            type Value = NonEmpty<T>;

            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                formatter.write_str("a non-empty sequence")
            }

            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
            where
                A: SeqAccess<'de>,
            {
                let head = seq
                    .next_element()?
                    .ok_or_else(|| serde::de::Error::custom("empty vector"))?;

                let mut tail = Vec::new();
                while let Some(elem) = seq.next_element()? {
                    tail.push(elem);
                }

                Ok(NonEmpty { head, tail })
            }
        }

        deserializer.deserialize_seq(NonEmptyVisitor(PhantomData))
    }
}

// =============================================================================
// TransactionWithClaims - Generalized claim system for consensus messages
// =============================================================================

/// Claims that can be attached to a transaction for consensus validation.
/// Each claim type represents a piece of information that:
/// 1. The submitting validator includes in the consensus message
/// 2. Voting validators verify before accepting
/// 3. The consensus handler can use deterministically
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub enum TransactionClaim {
    /// DEPRECATED. Do not use.
    #[deprecated(note = "Use AddressAliasesV2")]
    AddressAliases(
        #[serde(with = "nonempty_as_vec")] NonEmpty<(SuiAddress, Option<SequenceNumber>)>,
    ),

    /// Object IDs that are claimed to be immutable.
    /// Used to filter out immutable objects from lock acquisition in consensus handler.
    ImmutableInputObjects(Vec<ObjectID>),

    /// Address aliases used for signature verification.
    /// Length must equal the number of `required_signers`. Each element maps the corresponding
    /// signer to the signature index and alias object version (if any) used to verify it.
    AddressAliasesV2(#[serde(with = "nonempty_as_vec")] NonEmpty<(u8, Option<SequenceNumber>)>),
}

/// A transaction with attached claims that have been verified by voting validators.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TransactionWithClaims<T> {
    tx: T,
    claims: Vec<TransactionClaim>,
}

impl<T> TransactionWithClaims<T> {
    pub fn new(tx: T, claims: Vec<TransactionClaim>) -> Self {
        Self { tx, claims }
    }

    /// Create from a transaction with only address aliases.
    pub fn from_aliases(tx: T, aliases: NonEmpty<(u8, Option<SequenceNumber>)>) -> Self {
        Self {
            tx,
            claims: vec![TransactionClaim::AddressAliasesV2(aliases)],
        }
    }

    /// Creates from a transaction without any aliases attached.
    pub fn no_aliases(tx: T) -> Self {
        Self { tx, claims: vec![] }
    }

    pub fn tx(&self) -> &T {
        &self.tx
    }

    pub fn into_tx(self) -> T {
        self.tx
    }

    /// Get the address aliases V2 claim. Differentiate between empty and not present for validation.
    pub fn aliases(&self) -> Option<NonEmpty<(u8, Option<SequenceNumber>)>> {
        self.claims
            .iter()
            .find_map(|c| match c {
                TransactionClaim::AddressAliasesV2(aliases) => Some(aliases),
                _ => None,
            })
            .cloned()
    }

    // TODO: Remove once `fix_checkpoint_signature_mapping` flag is enabled in testnet.
    #[allow(deprecated)]
    pub fn aliases_v1(&self) -> Option<NonEmpty<(SuiAddress, Option<SequenceNumber>)>> {
        self.claims
            .iter()
            .find_map(|c| match c {
                TransactionClaim::AddressAliases(aliases) => Some(aliases),
                _ => None,
            })
            .cloned()
    }

    /// Get the immutable input objects claim. Returns empty vector if not present.
    pub fn get_immutable_objects(&self) -> Vec<ObjectID> {
        self.claims
            .iter()
            .find_map(|c| match c {
                TransactionClaim::ImmutableInputObjects(objs) => Some(objs.clone()),
                _ => None,
            })
            .unwrap_or_default()
    }
}

pub type PlainTransactionWithClaims = TransactionWithClaims<Transaction>;

/// Convert from `WithAliases<VerifiedEnvelope>` to `TransactionWithClaims<Envelope>`.
/// Used when feature flag is off to convert existing WithAliases to the new type.
impl<T: Message, S> From<WithAliases<VerifiedEnvelope<T, S>>>
    for TransactionWithClaims<Envelope<T, S>>
{
    fn from(value: WithAliases<VerifiedEnvelope<T, S>>) -> Self {
        let (tx, aliases) = value.into_inner();
        Self::from_aliases(tx.into(), aliases)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize, PartialOrd, Ord, Hash)]
pub enum InputObjectKind {
    // A Move package, must be immutable.
    MovePackage(ObjectID),
    // A Move object, either immutable, or owned mutable.
    ImmOrOwnedMoveObject(ObjectRef),
    // A Move object that's shared and mutable.
    SharedMoveObject {
        id: ObjectID,
        initial_shared_version: SequenceNumber,
        mutability: SharedObjectMutability,
    },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize, PartialOrd, Ord, Hash)]
pub enum SharedObjectMutability {
    // The "classic" mutable/immutable modes.
    Immutable,
    Mutable,
    // Non-exclusive write is used to allow multiple transactions to
    // simultaneously add disjoint dynamic fields to an object.
    // (Currently only used by settlement transactions).
    NonExclusiveWrite,
}

impl SharedObjectMutability {
    pub fn is_exclusive(&self) -> bool {
        match self {
            SharedObjectMutability::Mutable => true,
            SharedObjectMutability::Immutable => false,
            SharedObjectMutability::NonExclusiveWrite => false,
        }
    }
}

impl InputObjectKind {
    pub fn object_id(&self) -> ObjectID {
        self.full_object_id().id()
    }

    pub fn full_object_id(&self) -> FullObjectID {
        match self {
            Self::MovePackage(id) => FullObjectID::Fastpath(*id),
            Self::ImmOrOwnedMoveObject((id, _, _)) => FullObjectID::Fastpath(*id),
            Self::SharedMoveObject {
                id,
                initial_shared_version,
                ..
            } => FullObjectID::Consensus((*id, *initial_shared_version)),
        }
    }

    pub fn version(&self) -> Option<SequenceNumber> {
        match self {
            Self::MovePackage(..) => None,
            Self::ImmOrOwnedMoveObject((_, version, _)) => Some(*version),
            Self::SharedMoveObject { .. } => None,
        }
    }

    pub fn object_not_found_error(&self) -> UserInputError {
        match *self {
            Self::MovePackage(package_id) => {
                UserInputError::DependentPackageNotFound { package_id }
            }
            Self::ImmOrOwnedMoveObject((object_id, version, _)) => UserInputError::ObjectNotFound {
                object_id,
                version: Some(version),
            },
            Self::SharedMoveObject { id, .. } => UserInputError::ObjectNotFound {
                object_id: id,
                version: None,
            },
        }
    }

    pub fn is_shared_object(&self) -> bool {
        matches!(self, Self::SharedMoveObject { .. })
    }
}

/// The result of reading an object for execution. Because shared objects may be deleted, one
/// possible result of reading a shared object is that ObjectReadResultKind::Deleted is returned.
#[derive(Clone, Debug)]
pub struct ObjectReadResult {
    pub input_object_kind: InputObjectKind,
    pub object: ObjectReadResultKind,
}

#[derive(Clone)]
pub enum ObjectReadResultKind {
    Object(Object),
    // The version of the object that the transaction intended to read, and the digest of the tx
    // that removed it from consensus.
    ObjectConsensusStreamEnded(SequenceNumber, TransactionDigest),
    // A shared object in a cancelled transaction. The sequence number embeds cancellation reason.
    CancelledTransactionSharedObject(SequenceNumber),
}

impl ObjectReadResultKind {
    pub fn is_cancelled(&self) -> bool {
        matches!(
            self,
            ObjectReadResultKind::CancelledTransactionSharedObject(_)
        )
    }

    pub fn version(&self) -> SequenceNumber {
        match self {
            ObjectReadResultKind::Object(object) => object.version(),
            ObjectReadResultKind::ObjectConsensusStreamEnded(seq, _) => *seq,
            ObjectReadResultKind::CancelledTransactionSharedObject(seq) => *seq,
        }
    }
}

impl std::fmt::Debug for ObjectReadResultKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ObjectReadResultKind::Object(obj) => {
                write!(f, "Object({:?})", obj.compute_object_reference())
            }
            ObjectReadResultKind::ObjectConsensusStreamEnded(seq, digest) => {
                write!(f, "ObjectConsensusStreamEnded({}, {:?})", seq, digest)
            }
            ObjectReadResultKind::CancelledTransactionSharedObject(seq) => {
                write!(f, "CancelledTransactionSharedObject({})", seq)
            }
        }
    }
}

impl From<Object> for ObjectReadResultKind {
    fn from(object: Object) -> Self {
        Self::Object(object)
    }
}

impl ObjectReadResult {
    pub fn new(input_object_kind: InputObjectKind, object: ObjectReadResultKind) -> Self {
        if let (
            InputObjectKind::ImmOrOwnedMoveObject(_),
            ObjectReadResultKind::ObjectConsensusStreamEnded(_, _),
        ) = (&input_object_kind, &object)
        {
            panic!("only consensus objects can be ObjectConsensusStreamEnded");
        }

        if let (
            InputObjectKind::ImmOrOwnedMoveObject(_),
            ObjectReadResultKind::CancelledTransactionSharedObject(_),
        ) = (&input_object_kind, &object)
        {
            panic!("only consensus objects can be CancelledTransactionSharedObject");
        }

        Self {
            input_object_kind,
            object,
        }
    }

    pub fn id(&self) -> ObjectID {
        self.input_object_kind.object_id()
    }

    pub fn as_object(&self) -> Option<&Object> {
        match &self.object {
            ObjectReadResultKind::Object(object) => Some(object),
            ObjectReadResultKind::ObjectConsensusStreamEnded(_, _) => None,
            ObjectReadResultKind::CancelledTransactionSharedObject(_) => None,
        }
    }

    pub fn new_from_gas_object(gas: &Object) -> Self {
        let objref = gas.compute_object_reference();
        Self {
            input_object_kind: InputObjectKind::ImmOrOwnedMoveObject(objref),
            object: ObjectReadResultKind::Object(gas.clone()),
        }
    }

    pub fn is_mutable(&self) -> bool {
        match (&self.input_object_kind, &self.object) {
            (InputObjectKind::MovePackage(_), _) => false,
            (InputObjectKind::ImmOrOwnedMoveObject(_), ObjectReadResultKind::Object(object)) => {
                !object.is_immutable()
            }
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::ObjectConsensusStreamEnded(_, _),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::CancelledTransactionSharedObject(_),
            ) => unreachable!(),
            (InputObjectKind::SharedMoveObject { mutability, .. }, _) => match mutability {
                SharedObjectMutability::Mutable => true,
                SharedObjectMutability::Immutable => false,
                SharedObjectMutability::NonExclusiveWrite => false,
            },
        }
    }

    pub fn is_shared_object(&self) -> bool {
        self.input_object_kind.is_shared_object()
    }

    pub fn is_consensus_stream_ended(&self) -> bool {
        self.consensus_stream_end_info().is_some()
    }

    pub fn consensus_stream_end_info(&self) -> Option<(SequenceNumber, TransactionDigest)> {
        match &self.object {
            ObjectReadResultKind::ObjectConsensusStreamEnded(v, tx) => Some((*v, *tx)),
            _ => None,
        }
    }

    /// Return the object ref iff the object is an address-owned object (i.e. not shared, not immutable).
    pub fn get_address_owned_objref(&self) -> Option<ObjectRef> {
        match (&self.input_object_kind, &self.object) {
            (InputObjectKind::MovePackage(_), _) => None,
            (
                InputObjectKind::ImmOrOwnedMoveObject(objref),
                ObjectReadResultKind::Object(object),
            ) => {
                if object.is_immutable() {
                    None
                } else {
                    Some(*objref)
                }
            }
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::ObjectConsensusStreamEnded(_, _),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::CancelledTransactionSharedObject(_),
            ) => unreachable!(),
            (InputObjectKind::SharedMoveObject { .. }, _) => None,
        }
    }

    pub fn is_address_owned(&self) -> bool {
        self.get_address_owned_objref().is_some()
    }

    pub fn is_replay_protected_input(&self) -> bool {
        if let InputObjectKind::ImmOrOwnedMoveObject(obj_ref) = &self.input_object_kind
            && ParsedDigest::is_coin_reservation_digest(&obj_ref.2)
        {
            true
        } else {
            self.is_address_owned()
        }
    }

    pub fn to_shared_input(&self) -> Option<SharedInput> {
        match self.input_object_kind {
            InputObjectKind::MovePackage(_) => None,
            InputObjectKind::ImmOrOwnedMoveObject(_) => None,
            InputObjectKind::SharedMoveObject { id, mutability, .. } => Some(match &self.object {
                ObjectReadResultKind::Object(obj) => {
                    SharedInput::Existing(obj.compute_object_reference())
                }
                ObjectReadResultKind::ObjectConsensusStreamEnded(seq, digest) => {
                    SharedInput::ConsensusStreamEnded((id, *seq, mutability, *digest))
                }
                ObjectReadResultKind::CancelledTransactionSharedObject(seq) => {
                    SharedInput::Cancelled((id, *seq))
                }
            }),
        }
    }

    pub fn get_previous_transaction(&self) -> Option<TransactionDigest> {
        match &self.object {
            ObjectReadResultKind::Object(obj) => Some(obj.previous_transaction),
            ObjectReadResultKind::ObjectConsensusStreamEnded(_, digest) => Some(*digest),
            ObjectReadResultKind::CancelledTransactionSharedObject(_) => None,
        }
    }
}

#[derive(Clone)]
pub struct InputObjects {
    objects: Vec<ObjectReadResult>,
}

impl std::fmt::Debug for InputObjects {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_list().entries(self.objects.iter()).finish()
    }
}

// An InputObjects new-type that has been verified by sui-transaction-checks, and can be
// safely passed to execution.
#[derive(Clone)]
pub struct CheckedInputObjects(InputObjects);

// DO NOT CALL outside of sui-transaction-checks, genesis, or replay.
//
// CheckedInputObjects should really be defined in sui-transaction-checks so that we can
// make public construction impossible. But we can't do that because it would result in circular
// dependencies.
impl CheckedInputObjects {
    // Only called by sui-transaction-checks.
    pub fn new_with_checked_transaction_inputs(inputs: InputObjects) -> Self {
        Self(inputs)
    }

    // Only called when building the genesis transaction
    pub fn new_for_genesis(input_objects: Vec<ObjectReadResult>) -> Self {
        Self(InputObjects::new(input_objects))
    }

    // Only called from the replay tool.
    pub fn new_for_replay(input_objects: InputObjects) -> Self {
        Self(input_objects)
    }

    pub fn inner(&self) -> &InputObjects {
        &self.0
    }

    pub fn into_inner(self) -> InputObjects {
        self.0
    }
}

impl From<Vec<ObjectReadResult>> for InputObjects {
    fn from(objects: Vec<ObjectReadResult>) -> Self {
        Self::new(objects)
    }
}

impl InputObjects {
    pub fn new(objects: Vec<ObjectReadResult>) -> Self {
        Self { objects }
    }

    pub fn len(&self) -> usize {
        self.objects.len()
    }

    pub fn is_empty(&self) -> bool {
        self.objects.is_empty()
    }

    pub fn contains_consensus_stream_ended_objects(&self) -> bool {
        self.objects
            .iter()
            .any(|obj| obj.is_consensus_stream_ended())
    }

    // Returns IDs of objects responsible for a transaction being cancelled, and the corresponding
    // reason for cancellation.
    pub fn get_cancelled_objects(&self) -> Option<(Vec<ObjectID>, SequenceNumber)> {
        let mut contains_cancelled = false;
        let mut cancel_reason = None;
        let mut cancelled_objects = Vec::new();
        for obj in &self.objects {
            if let ObjectReadResultKind::CancelledTransactionSharedObject(version) = obj.object {
                contains_cancelled = true;
                if version == SequenceNumber::CONGESTED
                    || version == SequenceNumber::RANDOMNESS_UNAVAILABLE
                {
                    // Verify we don't have multiple cancellation reasons.
                    assert!(cancel_reason.is_none() || cancel_reason == Some(version));
                    cancel_reason = Some(version);
                    cancelled_objects.push(obj.id());
                }
            }
        }

        if !cancelled_objects.is_empty() {
            Some((
                cancelled_objects,
                cancel_reason
                    .expect("there should be a cancel reason if there are cancelled objects"),
            ))
        } else {
            assert!(!contains_cancelled);
            None
        }
    }

    pub fn filter_owned_objects(&self) -> Vec<ObjectRef> {
        let owned_objects: Vec<_> = self
            .objects
            .iter()
            .filter_map(|obj| obj.get_address_owned_objref())
            .collect();

        trace!(
            num_mutable_objects = owned_objects.len(),
            "Checked locks and found mutable objects"
        );

        owned_objects
    }

    pub fn filter_shared_objects(&self) -> Vec<SharedInput> {
        self.objects
            .iter()
            .filter(|obj| obj.is_shared_object())
            .map(|obj| {
                obj.to_shared_input()
                    .expect("already filtered for shared objects")
            })
            .collect()
    }

    pub fn transaction_dependencies(&self) -> BTreeSet<TransactionDigest> {
        self.objects
            .iter()
            .filter_map(|obj| obj.get_previous_transaction())
            .collect()
    }

    /// All inputs that will be directly mutated by the transaction. This does
    /// not include SharedObjectMutability::NonExclusiveWrite inputs.
    pub fn exclusive_mutable_inputs(&self) -> BTreeMap<ObjectID, (VersionDigest, Owner)> {
        self.mutables_with_input_kinds()
            .filter_map(|(id, (version, owner, kind))| match kind {
                InputObjectKind::SharedMoveObject { mutability, .. } => match mutability {
                    SharedObjectMutability::Mutable => Some((id, (version, owner))),
                    SharedObjectMutability::Immutable => None,
                    SharedObjectMutability::NonExclusiveWrite => None,
                },
                _ => Some((id, (version, owner))),
            })
            .collect()
    }

    pub fn non_exclusive_input_objects(&self) -> BTreeMap<ObjectID, Object> {
        self.objects
            .iter()
            .filter_map(|read_result| {
                match (read_result.as_object(), read_result.input_object_kind) {
                    (
                        Some(object),
                        InputObjectKind::SharedMoveObject {
                            mutability: SharedObjectMutability::NonExclusiveWrite,
                            ..
                        },
                    ) => Some((read_result.id(), object.clone())),
                    _ => None,
                }
            })
            .collect()
    }

    /// All inputs that can be taken as &mut T, which includes both
    /// SharedObjectMutability::Mutable and SharedObjectMutability::NonExclusiveWrite inputs.
    pub fn all_mutable_inputs(&self) -> BTreeMap<ObjectID, (VersionDigest, Owner)> {
        self.mutables_with_input_kinds()
            .filter_map(|(id, (version, owner, kind))| match kind {
                InputObjectKind::SharedMoveObject { mutability, .. } => match mutability {
                    SharedObjectMutability::Mutable => Some((id, (version, owner))),
                    SharedObjectMutability::Immutable => None,
                    SharedObjectMutability::NonExclusiveWrite => Some((id, (version, owner))),
                },
                _ => Some((id, (version, owner))),
            })
            .collect()
    }

    fn mutables_with_input_kinds(
        &self,
    ) -> impl Iterator<Item = (ObjectID, (VersionDigest, Owner, InputObjectKind))> + '_ {
        self.objects.iter().filter_map(
            |ObjectReadResult {
                 input_object_kind,
                 object,
             }| match (input_object_kind, object) {
                (InputObjectKind::MovePackage(_), _) => None,
                (
                    InputObjectKind::ImmOrOwnedMoveObject(object_ref),
                    ObjectReadResultKind::Object(object),
                ) => {
                    if object.is_immutable() {
                        None
                    } else {
                        Some((
                            object_ref.0,
                            (
                                (object_ref.1, object_ref.2),
                                object.owner.clone(),
                                *input_object_kind,
                            ),
                        ))
                    }
                }
                (
                    InputObjectKind::ImmOrOwnedMoveObject(_),
                    ObjectReadResultKind::ObjectConsensusStreamEnded(_, _),
                ) => {
                    unreachable!()
                }
                (
                    InputObjectKind::SharedMoveObject { .. },
                    ObjectReadResultKind::ObjectConsensusStreamEnded(_, _),
                ) => None,
                (
                    InputObjectKind::SharedMoveObject { mutability, .. },
                    ObjectReadResultKind::Object(object),
                ) => match *mutability {
                    SharedObjectMutability::Mutable => {
                        let oref = object.compute_object_reference();
                        Some((
                            oref.0,
                            ((oref.1, oref.2), object.owner.clone(), *input_object_kind),
                        ))
                    }
                    SharedObjectMutability::Immutable => None,
                    SharedObjectMutability::NonExclusiveWrite => {
                        let oref = object.compute_object_reference();
                        Some((
                            oref.0,
                            ((oref.1, oref.2), object.owner.clone(), *input_object_kind),
                        ))
                    }
                },
                (
                    InputObjectKind::ImmOrOwnedMoveObject(_),
                    ObjectReadResultKind::CancelledTransactionSharedObject(_),
                ) => {
                    unreachable!()
                }
                (
                    InputObjectKind::SharedMoveObject { .. },
                    ObjectReadResultKind::CancelledTransactionSharedObject(_),
                ) => None,
            },
        )
    }

    /// The version to set on objects created by the computation that `self` is input to.
    /// Guaranteed to be strictly greater than the versions of all input objects and objects
    /// received in the transaction.
    pub fn lamport_timestamp(&self, receiving_objects: &[ObjectRef]) -> SequenceNumber {
        let input_versions = self
            .objects
            .iter()
            .filter_map(|object| match &object.object {
                ObjectReadResultKind::Object(object) => {
                    object.data.try_as_move().map(MoveObject::version)
                }
                ObjectReadResultKind::ObjectConsensusStreamEnded(v, _) => Some(*v),
                ObjectReadResultKind::CancelledTransactionSharedObject(_) => None,
            })
            .chain(receiving_objects.iter().map(|object_ref| object_ref.1));

        SequenceNumber::lamport_increment(input_versions)
    }

    pub fn object_kinds(&self) -> impl Iterator<Item = &InputObjectKind> {
        self.objects.iter().map(
            |ObjectReadResult {
                 input_object_kind, ..
             }| input_object_kind,
        )
    }

    pub fn consensus_stream_ended_objects(&self) -> BTreeMap<ObjectID, SequenceNumber> {
        self.objects
            .iter()
            .filter_map(|obj| {
                if let InputObjectKind::SharedMoveObject {
                    id,
                    initial_shared_version,
                    ..
                } = obj.input_object_kind
                {
                    obj.is_consensus_stream_ended()
                        .then_some((id, initial_shared_version))
                } else {
                    None
                }
            })
            .collect()
    }

    pub fn into_object_map(self) -> BTreeMap<ObjectID, Object> {
        self.objects
            .into_iter()
            .filter_map(|o| o.as_object().map(|object| (o.id(), object.clone())))
            .collect()
    }

    pub fn push(&mut self, object: ObjectReadResult) {
        self.objects.push(object);
    }

    pub fn iter(&self) -> impl Iterator<Item = &ObjectReadResult> {
        self.objects.iter()
    }

    pub fn iter_objects(&self) -> impl Iterator<Item = &Object> {
        self.objects.iter().filter_map(|o| o.as_object())
    }

    pub fn non_exclusive_mutable_inputs(
        &self,
    ) -> impl Iterator<Item = (ObjectID, SequenceNumber)> + '_ {
        self.objects.iter().filter_map(
            |ObjectReadResult {
                 input_object_kind,
                 object,
             }| match input_object_kind {
                // TODO: this is not exercised yet since settlement transactions cannot be
                // cancelled, but if/when we expose non-exclusive writes to users,
                // a cancelled transaction should not be considered to have done any writes.
                InputObjectKind::SharedMoveObject {
                    id,
                    mutability: SharedObjectMutability::NonExclusiveWrite,
                    ..
                } if !object.is_cancelled() => Some((*id, object.version())),
                _ => None,
            },
        )
    }
}

// Result of attempting to read a receiving object (currently only at signing time).
// Because an object may have been previously received and deleted, the result may be
// ReceivingObjectReadResultKind::PreviouslyReceivedObject.
#[derive(Clone, Debug)]
pub enum ReceivingObjectReadResultKind {
    Object(Object),
    // The object was received by some other transaction, and we were not able to read it
    PreviouslyReceivedObject,
}

impl ReceivingObjectReadResultKind {
    pub fn as_object(&self) -> Option<&Object> {
        match &self {
            Self::Object(object) => Some(object),
            Self::PreviouslyReceivedObject => None,
        }
    }
}

pub struct ReceivingObjectReadResult {
    pub object_ref: ObjectRef,
    pub object: ReceivingObjectReadResultKind,
}

impl ReceivingObjectReadResult {
    pub fn new(object_ref: ObjectRef, object: ReceivingObjectReadResultKind) -> Self {
        Self { object_ref, object }
    }

    pub fn is_previously_received(&self) -> bool {
        matches!(
            self.object,
            ReceivingObjectReadResultKind::PreviouslyReceivedObject
        )
    }
}

impl From<Object> for ReceivingObjectReadResultKind {
    fn from(object: Object) -> Self {
        Self::Object(object)
    }
}

pub struct ReceivingObjects {
    pub objects: Vec<ReceivingObjectReadResult>,
}

impl ReceivingObjects {
    pub fn iter(&self) -> impl Iterator<Item = &ReceivingObjectReadResult> {
        self.objects.iter()
    }

    pub fn iter_objects(&self) -> impl Iterator<Item = &Object> {
        self.objects.iter().filter_map(|o| o.object.as_object())
    }
}

impl From<Vec<ReceivingObjectReadResult>> for ReceivingObjects {
    fn from(objects: Vec<ReceivingObjectReadResult>) -> Self {
        Self { objects }
    }
}

impl Display for CertifiedTransaction {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let mut writer = String::new();
        writeln!(writer, "Transaction Hash: {:?}", self.digest())?;
        writeln!(
            writer,
            "Signed Authorities Bitmap : {:?}",
            self.auth_sig().signers_map
        )?;
        write!(writer, "{}", &self.data().intent_message().value.kind())?;
        write!(f, "{}", writer)
    }
}

/// TransactionKey uniquely identifies a transaction across all epochs.
/// Note that a single transaction may have multiple keys, for example a RandomnessStateUpdate
/// could be identified by both `Digest` and `RandomnessRound`.
#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum TransactionKey {
    Digest(TransactionDigest),
    RandomnessRound(EpochId, RandomnessRound),
    AccumulatorSettlement(EpochId, u64 /* checkpoint height */),
    ConsensusCommitPrologue(EpochId, u64 /* round */, u32 /* sub_dag_index */),
}

impl TransactionKey {
    pub fn unwrap_digest(&self) -> &TransactionDigest {
        match self {
            TransactionKey::Digest(d) => d,
            _ => panic!("called unwrap_digest on a non-Digest TransactionKey: {self:?}"),
        }
    }

    pub fn as_digest(&self) -> Option<&TransactionDigest> {
        match self {
            TransactionKey::Digest(d) => Some(d),
            _ => None,
        }
    }
}