solana_svm/

transaction_processor.rs

use {
    crate::{
        account_loader::{
            AccountLoader, CheckedTransactionDetails, FeesOnlyTransaction, LoadedTransaction,
            NoOpTransaction, TransactionCheckResult, TransactionLoadResult,
            TransactionValidationResult, ValidatedTransactionDetails, load_transaction,
            update_rent_exempt_status_for_account, validate_fee_payer,
        },
        account_overrides::AccountOverrides,
        nonce_info::NonceInfo,
        program_loader::{filter_executable_program_accounts, load_program_with_pubkey},
        rollback_accounts::RollbackAccounts,
        transaction_account_state_info::{
            TransactionAccountStateInfo, get_uninitialized_accounts_size, verify_changes,
        },
        transaction_balances::{BalanceCollectionRoutines, BalanceCollector},
        transaction_error_metrics::TransactionErrorMetrics,
        transaction_execution_result::{
            AccountsDeltas, ExecutedTransaction, TransactionExecutionDetails,
        },
        transaction_processing_result::{ProcessedTransaction, TransactionProcessingResult},
    },
    log::debug,
    solana_account::{AccountSharedData, ReadableAccount, state_traits::StateMutWincode as _},
    solana_clock::{Epoch, Slot},
    solana_hash::Hash,
    solana_instruction::TRANSACTION_LEVEL_STACK_HEIGHT,
    solana_instruction_error::InstructionError,
    solana_message::{
        compiled_instruction::CompiledInstruction,
        inner_instruction::{InnerInstruction, InnerInstructionsList},
    },
    solana_nonce::{
        NONCED_TX_MARKER_IX_INDEX,
        state::{DurableNonce, State as NonceState},
        versions::Versions as NonceVersions,
    },
    solana_nonce_account::verify_nonce_account,
    solana_program_runtime::{
        execution_budget::{
            SVMTransactionExecutionAndFeeBudgetLimits, SVMTransactionExecutionCost,
        },
        invoke_context::{EnvironmentConfig, InvokeContext},
        loaded_programs::{
            EpochBoundaryPreparation, ForkGraph, Percent, ProgramCache, ProgramCacheForTxBatch,
            ProgramRuntimeEnvironment, ProgramRuntimeEnvironments, ProgramToLoad,
        },
        program_cache_entry::{ProgramCacheEntry, ProgramCacheEntryOwner},
        program_metrics::ProgramStatistics,
        solana_sbpf::{program::BuiltinProgram, vm::Config as VmConfig},
        sysvar_cache::SysvarCache,
    },
    solana_pubkey::Pubkey,
    solana_rent::Rent,
    solana_svm_callback::{InvokeContextCallback, TransactionProcessingCallback},
    solana_svm_feature_set::SVMFeatureSet,
    solana_svm_log_collector::LogCollector,
    solana_svm_measure::{measure::Measure, measure_us},
    solana_svm_timings::{ExecuteTimingType, ExecuteTimings},
    solana_svm_transaction::{svm_message::SVMMessage, svm_transaction::SVMTransaction},
    solana_svm_type_overrides::sync::{Arc, RwLock, RwLockReadGuard},
    solana_transaction_context::{
        DropOnBailOut,
        transaction::{ComputeMeter, ExecutionRecord, TransactionContext},
    },
    solana_transaction_error::{TransactionError, TransactionResult},
    std::{
        collections::HashSet,
        fmt::{Debug, Formatter},
        rc::Rc,
    },
};
#[cfg(feature = "dev-context-only-utils")]
use {
    qualifier_attr::{field_qualifiers, qualifiers},
    std::sync::Weak,
};

/// A list of log messages emitted during a transaction
pub type TransactionLogMessages = Vec<String>;

/// The output of the transaction batch processor's
/// `load_and_execute_sanitized_transactions` method.
pub struct LoadAndExecuteSanitizedTransactionsOutput {
    /// Error metrics for transactions that were processed.
    pub error_metrics: TransactionErrorMetrics,
    /// Timings for transaction batch execution.
    pub execute_timings: ExecuteTimings,
    /// Vector of results indicating whether a transaction was processed or
    /// could not be processed. Note processed transactions can still have a
    /// failure result meaning that the transaction will be rolled back.
    pub processing_results: Vec<TransactionProcessingResult>,
    /// Balances accumulated for TransactionStatusSender when
    /// transaction balance recording is enabled.
    pub balance_collector: Option<BalanceCollector>,
}

/// Configuration of the recording capabilities for transaction execution
#[derive(Copy, Clone, Default)]
pub struct ExecutionRecordingConfig {
    pub enable_cpi_recording: bool,
    pub enable_log_recording: bool,
    pub enable_return_data_recording: bool,
    pub enable_transaction_balance_recording: bool,
}

impl ExecutionRecordingConfig {
    pub fn new_single_setting(option: bool) -> Self {
        ExecutionRecordingConfig {
            enable_return_data_recording: option,
            enable_log_recording: option,
            enable_cpi_recording: option,
            enable_transaction_balance_recording: option,
        }
    }
}

/// Configurations for processing transactions.
#[derive(Default)]
pub struct TransactionProcessingConfig<'a> {
    /// Encapsulates overridden accounts, typically used for transaction
    /// simulation.
    pub account_overrides: Option<&'a AccountOverrides>,
    /// The maximum number of bytes that log messages can consume.
    pub log_messages_bytes_limit: Option<usize>,
    /// Whether to limit the number of programs loaded for the transaction
    /// batch.
    pub limit_to_load_programs: bool,
    /// Recording capabilities for transaction execution.
    pub recording_config: ExecutionRecordingConfig,
    /// Should failing transactions within the batch be dropped (no fee charged
    /// & not committed).
    pub drop_on_failure: bool,
    /// If any transaction in the batch is not committed then the entire batch
    /// should not be committed.
    ///
    /// # Note
    ///
    /// Without `drop_on_failure` this flag will still allow processed but
    /// failing transactions to be committed. If both flags are set then any
    /// failing transaction will cause all transactions to be aborted.
    pub all_or_nothing: bool,
    /// Strictly require durable nonce accounts to have the canonical nonce account size.
    ///
    /// This is a leader-side filtering policy. It must not be enabled for replay.
    pub strict_nonce_size_check: bool,
    /// Do not commit unprocessable (no-op) transactions to the ledger.
    ///
    /// This is a leader-side filtering policy. It must not be enabled for replay.
    pub drop_noop_transactions: bool,
    /// Drops transactions which bailed out in the program runtime.
    ///
    /// This is a leader-side filtering policy. It must not be enabled for replay.
    pub drop_bail_out_transactions: bool,
}

/// Runtime environment for transaction batch processing.
pub struct TransactionProcessingEnvironment {
    /// The blockhash to use for the transaction batch.
    pub blockhash: Hash,
    /// Lamports per signature that corresponds to this blockhash.
    ///
    /// Note: This value is primarily used for nonce accounts. If set to zero,
    /// it will disable transaction fees. However, any non-zero value will not
    /// change transaction fees. For this reason, it is recommended to use the
    /// `fee_per_signature` field to adjust transaction fees.
    pub blockhash_lamports_per_signature: u64,
    /// Whether the alpenglow migration has completed for this bank context.
    pub alpenglow_migration_succeeded: bool,
    /// The total stake for the current epoch.
    pub epoch_total_stake: u64,
    /// Runtime feature set to use for the transaction batch.
    pub feature_set: SVMFeatureSet,
    /// Program runtime environments for execution and deployment.
    pub program_runtime_environments: ProgramRuntimeEnvironments,
    /// Rent calculator to use for the transaction batch.
    pub rent: Rent,
}

#[cfg(feature = "dev-context-only-utils")]
pub fn get_mock_transaction_processing_environment() -> TransactionProcessingEnvironment {
    TransactionProcessingEnvironment {
        blockhash: Hash::default(),
        blockhash_lamports_per_signature: 0,
        alpenglow_migration_succeeded: false,
        epoch_total_stake: 0,
        feature_set: SVMFeatureSet::default(),
        program_runtime_environments: ProgramRuntimeEnvironments::mock(),
        rent: Rent::default(),
    }
}

#[cfg_attr(
    feature = "dev-context-only-utils",
    field_qualifiers(slot(pub), epoch(pub), sysvar_cache(pub))
)]
pub struct TransactionBatchProcessor<FG: ForkGraph> {
    /// Bank slot (i.e. block)
    slot: Slot,

    /// Bank epoch
    epoch: Epoch,

    /// SysvarCache is a collection of system variables that are
    /// accessible from on chain programs. It is passed to SVM from
    /// client code (e.g. Bank) and forwarded to process_message.
    sysvar_cache: RwLock<SysvarCache>,

    /// Anticipates the environments of the upcoming epoch
    pub epoch_boundary_preparation: Arc<RwLock<EpochBoundaryPreparation>>,

    /// Programs required for transaction batch processing
    pub global_program_cache: Arc<RwLock<ProgramCache<FG>>>,

    /// ProgramRuntimeEnvironment of the current epoch
    pub program_runtime_environment: ProgramRuntimeEnvironment,

    /// In the final slot of an epoch where we've activated a program runtime
    /// feature, this override protects us from an `epoch_boundary_preparation`
    /// environment that is outdated or otherwise incorrect.
    deployment_env_override: Option<ProgramRuntimeEnvironment>,

    /// Builtin program ids for this fork.
    /// Used to see the `builtin_program_cache` between slots via
    /// `new_from()`.
    pub builtin_program_ids: RwLock<HashSet<Pubkey>>,

    /// Cached ProgramCacheForTxBatch pre-populated with builtin entries.
    /// Populated once per slot in `new_from()` from the global program cache.
    ///
    /// Keeping a dedicated builtin program cache for each slot:
    ///
    /// - Protects different forks from extracting the another fork's version
    ///   of a builtin that may have been recently enabled or migrated to Core
    ///   BPF.
    /// - Avoids re-acquiring the lock and re-running extract() on every batch.
    builtin_program_cache: RwLock<ProgramCacheForTxBatch>,

    execution_cost: SVMTransactionExecutionCost,
}

impl<FG: ForkGraph> Debug for TransactionBatchProcessor<FG> {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TransactionBatchProcessor")
            .field("slot", &self.slot)
            .field("epoch", &self.epoch)
            .field("sysvar_cache", &self.sysvar_cache)
            .field("global_program_cache", &self.global_program_cache)
            .finish()
    }
}

impl<FG: ForkGraph> Default for TransactionBatchProcessor<FG> {
    fn default() -> Self {
        Self {
            slot: Slot::default(),
            epoch: Epoch::default(),
            sysvar_cache: RwLock::<SysvarCache>::default(),
            epoch_boundary_preparation: Arc::new(RwLock::new(EpochBoundaryPreparation::default())),
            global_program_cache: Arc::new(RwLock::new(ProgramCache::new(Slot::default()))),
            program_runtime_environment: ProgramRuntimeEnvironment::from(
                BuiltinProgram::new_loader(VmConfig::default()),
            ),
            deployment_env_override: None,
            builtin_program_ids: RwLock::new(HashSet::new()),
            builtin_program_cache: RwLock::new(ProgramCacheForTxBatch::new(Slot::default())),
            execution_cost: SVMTransactionExecutionCost::default(),
        }
    }
}

impl<FG: ForkGraph> TransactionBatchProcessor<FG> {
    /// Create a new, uninitialized `TransactionBatchProcessor`.
    ///
    /// In this context, uninitialized means that the `TransactionBatchProcessor`
    /// has been initialized with an empty program cache. The cache contains no
    /// programs (including builtins) and has not been configured with a valid
    /// fork graph.
    ///
    /// When using this method, it's advisable to call `set_fork_graph_in_program_cache`
    /// as well as `add_builtin` to configure the cache before using the processor.
    pub fn new_uninitialized(slot: Slot, epoch: Epoch) -> Self {
        let epoch_boundary_preparation =
            Arc::new(RwLock::new(EpochBoundaryPreparation::new(epoch)));
        Self {
            slot,
            epoch,
            epoch_boundary_preparation,
            global_program_cache: Arc::new(RwLock::new(ProgramCache::new(slot))),
            builtin_program_cache: RwLock::new(ProgramCacheForTxBatch::new(slot)),
            ..Self::default()
        }
    }

    /// Create a new `TransactionBatchProcessor`.
    ///
    /// The created processor's program cache is initialized with the provided
    /// fork graph and loaders. If any loaders are omitted, a default "empty"
    /// loader (no syscalls) will be used.
    ///
    /// The cache will still not contain any builtin programs. It's advisable to
    /// call `add_builtin` to add the required builtins before using the processor.
    #[cfg(feature = "dev-context-only-utils")]
    pub fn new(
        slot: Slot,
        epoch: Epoch,
        fork_graph: Weak<RwLock<FG>>,
        program_runtime_environment: Option<ProgramRuntimeEnvironment>,
    ) -> Self {
        let mut processor = Self::new_uninitialized(slot, epoch);
        processor
            .global_program_cache
            .write()
            .unwrap()
            .set_fork_graph(fork_graph);
        let empty_loader = || ProgramRuntimeEnvironment::from(BuiltinProgram::new_mock());
        processor
            .global_program_cache
            .write()
            .unwrap()
            .latest_root_slot = processor.slot;
        processor
            .epoch_boundary_preparation
            .write()
            .unwrap()
            .upcoming_epoch = processor.epoch;
        processor.program_runtime_environment =
            program_runtime_environment.unwrap_or(empty_loader());
        processor
    }

    /// Create a new `TransactionBatchProcessor` from the current instance, but
    /// with the provided slot and epoch.
    ///
    /// * Inherits the program cache and builtin program ids from the current
    ///   instance.
    /// * Resets the sysvar cache.
    pub fn new_from(&self, slot: Slot, epoch: Epoch) -> Self {
        let builtin_program_ids = self.builtin_program_ids.read().unwrap().clone();
        let environments = self.program_runtime_environment.clone();

        // Pre-populate the builtin program cache from the global cache.
        // This is done once per block rather than once per batch.
        //
        // Note: Only the builtins already-listed in this slot's builtin program
        // IDs can be extracted from the global program cache. This is important
        // for cross-fork extraction when a builtin may be activated for the
        // first time or migrated to Core BPF.
        //
        // The `builtin_program_ids` on the batch processor are effectively a
        // guard against cross-fork extraction:
        //
        // - If a builtin appears in one fork's `builtin_program_ids`, it will
        //   never be extracted by a fork whose `builtin_program_ids` _doesn't_
        //   have it.
        // - If a builtin is removed from one fork's `builtin_program_ids`, it
        //   may still be extracted on another fork whose `builtin_program_ids`
        //   still has it.
        //
        // `builtin_program_ids` seeds the `builtin_program_cache` below, which
        // ensures only non-cached program IDs are extracted from the global
        // program cache during `replenish_program_cache`.
        let mut builtin_program_cache = ProgramCacheForTxBatch::new(slot);
        let mut search_for: Vec<ProgramToLoad> = builtin_program_ids
            .iter()
            .map(|program_id| ProgramToLoad {
                program_id,
                loader: ProgramCacheEntryOwner::NativeLoader,
                deployment_slot: 0,
            })
            .collect();
        self.global_program_cache.read().unwrap().extract(
            &mut search_for,
            &mut builtin_program_cache,
            &environments,
            false,
            false,
        );

        // Every registered builtin must be present in the global program cache,
        // since `add_builtin` writes both together.
        debug_assert!(
            search_for.is_empty(),
            "builtin(s) registered on this fork but missing from the global program cache: \
             {search_for:?}"
        );

        Self {
            slot,
            epoch,
            sysvar_cache: RwLock::<SysvarCache>::default(),
            epoch_boundary_preparation: self.epoch_boundary_preparation.clone(),
            global_program_cache: self.global_program_cache.clone(),
            program_runtime_environment: environments,
            deployment_env_override: None,
            builtin_program_ids: RwLock::new(builtin_program_ids),
            builtin_program_cache: RwLock::new(builtin_program_cache),
            execution_cost: self.execution_cost,
        }
    }

    /// Sets the base execution cost for the transactions that this instance of transaction processor
    /// will execute.
    pub fn set_execution_cost(&mut self, cost: SVMTransactionExecutionCost) {
        self.execution_cost = cost;
    }

    /// Updates the environments when entering a new Epoch.
    pub fn set_program_runtime_environment(&mut self, new_environment: ProgramRuntimeEnvironment) {
        // First update the environment only if it is different
        if *self.program_runtime_environment != *new_environment {
            self.program_runtime_environment = new_environment;
        }
        // Then try to consolidate with the upcoming environment (to reuse the address)
        if let Some(upcoming_environment) = &self
            .epoch_boundary_preparation
            .read()
            .unwrap()
            .upcoming_environment
        {
            let upcoming_environment = ProgramRuntimeEnvironment::clone(upcoming_environment);
            if self.program_runtime_environment != upcoming_environment
                && *self.program_runtime_environment == *upcoming_environment
            {
                // Use the prediction if equal but not identical
                self.program_runtime_environment = upcoming_environment;
            }
        }
    }

    /// Returns the current environments depending on the given epoch
    /// Returns None if the call could result in a deadlock
    pub fn program_runtime_environment_for_epoch(&self, epoch: Epoch) -> ProgramRuntimeEnvironment {
        self.epoch_boundary_preparation
            .read()
            .unwrap()
            .get_upcoming_environment_for_epoch(epoch)
            .unwrap_or_else(|| ProgramRuntimeEnvironment::clone(&self.program_runtime_environment))
    }

    pub fn deployment_env_override(&self) -> Option<&ProgramRuntimeEnvironment> {
        self.deployment_env_override.as_ref()
    }

    pub fn set_deployment_env_override(&mut self, environment: ProgramRuntimeEnvironment) {
        self.deployment_env_override = Some(environment);
    }

    pub fn sysvar_cache(&self) -> RwLockReadGuard<'_, SysvarCache> {
        self.sysvar_cache.read().unwrap()
    }

    /// Main entrypoint to the SVM.
    pub fn load_and_execute_sanitized_transactions<
        CB: TransactionProcessingCallback + InvokeContextCallback,
    >(
        &self,
        callbacks: &CB,
        sanitized_txs: &[impl SVMTransaction],
        check_results: Vec<TransactionCheckResult>,
        environment: &TransactionProcessingEnvironment,
        config: &TransactionProcessingConfig,
    ) -> LoadAndExecuteSanitizedTransactionsOutput {
        // If `check_results` does not have the same length as `sanitized_txs`,
        // transactions could be truncated as a result of `.iter().zip()` in
        // many of the below methods.
        // See <https://doc.rust-lang.org/std/iter/trait.Iterator.html#method.zip>.
        debug_assert_eq!(
            sanitized_txs.len(),
            check_results.len(),
            "Length of check_results does not match length of sanitized_txs"
        );

        // Initialize metrics.
        let mut error_metrics = TransactionErrorMetrics::default();
        let mut execute_timings = ExecuteTimings::default();
        let mut processing_results = Vec::with_capacity(sanitized_txs.len());

        // Determine a capacity for the internal account cache. This
        // over-allocates but avoids ever reallocating, and spares us from
        // deduplicating the account keys lists.
        let account_keys_in_batch = sanitized_txs.iter().map(|tx| tx.account_keys().len()).sum();

        // Create the account loader, which wraps all external account fetching.
        let mut account_loader = AccountLoader::new_with_loaded_accounts_capacity(
            config.account_overrides,
            callbacks,
            &environment.feature_set,
            account_keys_in_batch,
        );

        // Create the transaction balance collector if recording is enabled.
        let mut balance_collector = config
            .recording_config
            .enable_transaction_balance_recording
            .then(|| BalanceCollector::new_with_transaction_count(sanitized_txs.len()));

        // Clone the batch-local program cache (builtins already populated in new_from()).
        // User-deployed programs are loaded per-transaction via replenish_program_cache
        // in the transaction loop below.
        let mut program_cache_for_tx_batch = self.builtin_program_cache.read().unwrap().clone();

        if program_cache_for_tx_batch.hit_max_limit {
            return LoadAndExecuteSanitizedTransactionsOutput {
                error_metrics,
                execute_timings,
                processing_results: (0..sanitized_txs.len())
                    .map(|_| Err(TransactionError::ProgramCacheHitMaxLimit))
                    .collect(),
                // If we abort the batch and balance recording is enabled, no balances should be
                // collected. If this is a leader thread, no batch will be committed.
                balance_collector: None,
            };
        }

        let (mut load_us, mut execution_us): (u64, u64) = (0, 0);
        let sysvar_cache = self.sysvar_cache();

        // Validate, execute, and collect results from each transaction in order.
        // Transactions must be executed in order, because transactions
        // in the same batch may modify the same accounts. Transaction order is
        // preserved within entries written to the ledger.
        for (tx, check_result) in sanitized_txs.iter().zip(check_results) {
            let (validate_result, validate_fees_us) = measure_us!(match check_result {
                Ok(tx_details) => Self::validate_transaction_nonce_and_fee_payer(
                    &mut account_loader,
                    tx,
                    tx_details,
                    &environment.blockhash,
                    environment.blockhash_lamports_per_signature,
                    &environment.rent,
                    environment.feature_set.relax_post_exec_min_balance_check,
                    config.strict_nonce_size_check,
                    &mut error_metrics,
                ),
                Err(e) => TransactionValidationResult::Unprocessable(e),
            });
            execute_timings
                .saturating_add_in_place(ExecuteTimingType::ValidateFeesUs, validate_fees_us);

            let (load_result, single_load_us) = measure_us!(load_transaction(
                &mut account_loader,
                tx,
                validate_result,
                &mut error_metrics,
                &environment.rent,
            ));
            load_us = load_us.saturating_add(single_load_us);

            let ((), collect_balances_us) =
                measure_us!(balance_collector.collect_pre_balances(&mut account_loader, tx));
            execute_timings
                .saturating_add_in_place(ExecuteTimingType::CollectBalancesUs, collect_balances_us);

            let (processing_result, single_execution_us) = measure_us!(match load_result {
                // Unprocessable transactions always result in an error
                TransactionLoadResult::Unprocessable(e) => Err(e),

                // Validation failures that would be no-ops become errors with `drop_on_failure` or
                // `drop_noop_transactions`. These may be produced by SIMD-0290 (fee-payer failure)
                // or SIMD-0297 (nonce failure), but are dropped during block production to avoid commit.
                TransactionLoadResult::NoOp(NoOpTransaction {
                    validation_error: e,
                    ..
                }) if config.drop_on_failure || config.drop_noop_transactions => Err(e),

                // SIMD-0290 (fee-payer failure) or SIMD-0297 (nonce failure) is a non-error no-op on replay
                TransactionLoadResult::NoOp(no_op_tx) =>
                    Ok(ProcessedTransaction::NoOp(Box::new(no_op_tx))),

                // Loading failures that would be fee-only become errors with `drop_on_failure`
                TransactionLoadResult::FeesOnly(FeesOnlyTransaction { load_error: e, .. })
                    if config.drop_on_failure =>
                    Err(e),

                // Transactions that fail at account loading charge fees and roll nonces
                TransactionLoadResult::FeesOnly(fees_only_tx) => {
                    account_loader
                        .update_accounts_for_failed_tx(&fees_only_tx.rollback_accounts, self.slot);
                    Ok(ProcessedTransaction::FeesOnly(Box::new(fees_only_tx)))
                }

                // Transaction is able to be executed
                TransactionLoadResult::Loaded(loaded_transaction) => {
                    let (missing_programs, filter_executable_us) =
                        measure_us!(filter_executable_program_accounts(
                            &account_loader,
                            &program_cache_for_tx_batch,
                            tx.account_keys().iter(),
                        ));
                    execute_timings.saturating_add_in_place(
                        ExecuteTimingType::FilterExecutableUs,
                        filter_executable_us,
                    );

                    let ((), program_cache_us) = measure_us!({
                        self.replenish_program_cache(
                            &account_loader,
                            missing_programs,
                            environment
                                .program_runtime_environments
                                .get_env_for_execution(),
                            &mut program_cache_for_tx_batch,
                            &mut execute_timings,
                            config.limit_to_load_programs,
                            true, // increment_usage_counter
                        );
                    });
                    execute_timings.saturating_add_in_place(
                        ExecuteTimingType::ProgramCacheUs,
                        program_cache_us,
                    );

                    if program_cache_for_tx_batch.hit_max_limit {
                        return LoadAndExecuteSanitizedTransactionsOutput {
                            error_metrics,
                            execute_timings,
                            processing_results: (0..sanitized_txs.len())
                                .map(|_| Err(TransactionError::ProgramCacheHitMaxLimit))
                                .collect(),
                            // If we abort the batch and balance recording is enabled, no balances should be
                            // collected. If this is a leader thread, no batch will be committed.
                            balance_collector: None,
                        };
                    }

                    let executed_tx = self.execute_loaded_transaction(
                        callbacks,
                        tx,
                        &sysvar_cache,
                        loaded_transaction,
                        &mut execute_timings,
                        &mut error_metrics,
                        &mut program_cache_for_tx_batch,
                        environment,
                        config,
                    );

                    match (
                        &executed_tx.execution_details.status,
                        config.drop_on_failure,
                    ) {
                        // Successful transactions need to update the account loader cache as future
                        // transactions in the batch may depend on them.
                        (Ok(_), _) => {
                            account_loader.update_accounts_for_successful_tx(
                                tx,
                                &executed_tx.loaded_transaction.accounts,
                                &executed_tx.loaded_transaction.touched_flags,
                                self.slot,
                            );
                            // Also update local program cache with modifications made by the
                            // transaction, if it executed successfully.
                            program_cache_for_tx_batch.merge(&executed_tx.programs_modified_by_tx);

                            Ok(ProcessedTransaction::Executed(Box::new(executed_tx)))
                        }
                        // If it bailed out then this transaction will be dropped from the batch.
                        (
                            Err(TransactionError::InstructionError(_, InstructionError::BailOut)),
                            _,
                        ) => Err(TransactionError::BailOut),
                        // If the transaction failed & drop on failure is set then we don't want to
                        // update the accounts as this transaction will be dropped from the batch.
                        (Err(err), true) => Err(err.clone()),
                        // Unsuccessful transactions will still update rollback accounts (fee payer,
                        // nonce, etc).
                        (Err(_), false) => {
                            account_loader.update_accounts_for_failed_tx(
                                &executed_tx.loaded_transaction.rollback_accounts,
                                self.slot,
                            );

                            Ok(ProcessedTransaction::Executed(Box::new(executed_tx)))
                        }
                    }
                }
            });
            execution_us = execution_us.saturating_add(single_execution_us);

            let ((), collect_balances_us) =
                measure_us!(balance_collector.collect_post_balances(&mut account_loader, tx));
            execute_timings
                .saturating_add_in_place(ExecuteTimingType::CollectBalancesUs, collect_balances_us);

            // If this is an all or nothing batch and we failed to process this transaction then we
            // must abort all prior/remaining transactions.
            if config.all_or_nothing && processing_result.is_err() {
                // Abort prior transactions.
                for res in processing_results.iter_mut() {
                    *res = Err(TransactionError::CommitCancelled);
                }

                // Preserve the failure that triggered the batch to abort.
                processing_results.push(processing_result);

                // Abort remaining transactions.
                processing_results.extend(
                    (0..sanitized_txs.len() - processing_results.len())
                        .map(|_| Err(TransactionError::CommitCancelled)),
                );

                return LoadAndExecuteSanitizedTransactionsOutput {
                    error_metrics,
                    execute_timings,
                    processing_results,
                    // If we abort the batch and balance recording is enabled, no balances should be
                    // collected. If this is a leader thread, no batch will be committed.
                    balance_collector: None,
                };
            }

            processing_results.push(processing_result);
        }

        // Skip eviction when there's no chance this particular tx batch has increased the size of
        // ProgramCache entries. Note that loaded_missing is deliberately defined, so that there's
        // still at least one other batch, which will evict the program cache, even after the
        // occurrences of cooperative loading.
        if program_cache_for_tx_batch.loaded_missing || program_cache_for_tx_batch.merged_modified {
            // NOTE: this is a percentage; do not set above 100.
            const SHRINK_LOADED_PROGRAMS_TO_PERCENTAGE: Percent = 90;
            self.global_program_cache
                .write()
                .unwrap()
                .evict_using_random_selection(SHRINK_LOADED_PROGRAMS_TO_PERCENTAGE, self.slot);
        }

        debug!(
            "load: {}us execute: {}us txs_len={}",
            load_us,
            execution_us,
            sanitized_txs.len(),
        );
        execute_timings.saturating_add_in_place(ExecuteTimingType::LoadUs, load_us);
        execute_timings.saturating_add_in_place(ExecuteTimingType::ExecuteUs, execution_us);

        if let Some(ref balance_collector) = balance_collector {
            debug_assert!(balance_collector.lengths_match_expected(sanitized_txs.len()));
        }

        LoadAndExecuteSanitizedTransactionsOutput {
            error_metrics,
            execute_timings,
            processing_results,
            balance_collector,
        }
    }

    fn validate_transaction_nonce_and_fee_payer<CB: TransactionProcessingCallback>(
        account_loader: &mut AccountLoader<CB>,
        message: &impl SVMMessage,
        checked_details: CheckedTransactionDetails,
        environment_blockhash: &Hash,
        next_lamports_per_signature: u64,
        rent: &Rent,
        relax_post_exec_min_balance_check: bool,
        strict_nonce_size_check: bool,
        error_counters: &mut TransactionErrorMetrics,
    ) -> TransactionValidationResult {
        let CheckedTransactionDetails {
            nonce_address,
            compute_budget_and_limits,
        } = checked_details;

        // If this is a nonce transaction, validate the nonce info.
        // This must be done for every transaction because the nonce account
        // may have changed due to use, authorization, or deallocation.
        let nonce_info = if let Some(ref nonce_address) = nonce_address {
            let next_durable_nonce = DurableNonce::from_blockhash(environment_blockhash);
            let nonce_result = Self::validate_transaction_nonce(
                account_loader,
                message,
                nonce_address,
                &next_durable_nonce,
                next_lamports_per_signature,
                strict_nonce_size_check,
                error_counters,
            );

            match nonce_result {
                Ok(nonce_info) => Some(nonce_info),
                Err(e) => return TransactionValidationResult::Unprocessable(e),
            }
        } else {
            None
        };

        let is_blockhash_transaction = nonce_info.is_none();

        // Now validate the fee-payer for the transaction unconditionally.
        let fee_payer_result = Self::validate_transaction_fee_payer(
            account_loader,
            message,
            nonce_info,
            compute_budget_and_limits,
            rent,
            relax_post_exec_min_balance_check,
            error_counters,
        );

        // With SIMD-0290 enabled, a blockhash transaction with an invalid fee-payer can
        // be processed as a no-op. However, nonce transactions must ALWAYS be discarded,
        // regardless of whether they use a standard or nonce-based fee-payer.
        // Fee-payer failure for nonces will be handled the same as nonce failures
        // themselves when we implement SIMD-0297.
        let allow_noop =
            account_loader.feature_set.relax_fee_payer_constraint && is_blockhash_transaction;

        match fee_payer_result {
            Ok(details) => TransactionValidationResult::Loadable(details),
            Err(e) if allow_noop => {
                // We correctly report the fee-payer balance if the account exists but is invalid.
                let fee_payer_balance = account_loader
                    .load_account(message.fee_payer())
                    .map(|account| account.lamports());

                // Per SIMD-0290, we report the maximum allowed compute and data usage.
                // In essence the intent is if you pack a block with n non-paying transactions,
                // you could have packed n or more of the same paying transactions, rather than
                // being able to pack some multiple more non-paying than paying.
                TransactionValidationResult::NoOp(NoOpTransaction {
                    validation_error: e,
                    fee_payer_balance,
                    compute_unit_limit: compute_budget_and_limits.budget.compute_unit_limit,
                    loaded_accounts_bytes_limit: compute_budget_and_limits
                        .loaded_accounts_data_size_limit,
                    nonce_address,
                })
            }
            Err(e) => TransactionValidationResult::Unprocessable(e),
        }
    }

    // Loads transaction fee payer, collects rent if necessary, then calculates
    // transaction fees, and deducts them from the fee payer balance. If the
    // account is not found or has insufficient funds, an error is returned.
    fn validate_transaction_fee_payer<CB: TransactionProcessingCallback>(
        account_loader: &mut AccountLoader<CB>,
        message: &impl SVMMessage,
        nonce_info: Option<NonceInfo>,
        compute_budget_and_limits: SVMTransactionExecutionAndFeeBudgetLimits,
        rent: &Rent,
        relax_post_exec_min_balance_check: bool,
        error_counters: &mut TransactionErrorMetrics,
    ) -> TransactionResult<ValidatedTransactionDetails> {
        let fee_payer_address = message.fee_payer();

        // We *must* use load_transaction_account() here because *this* is when the fee-payer
        // is loaded for the transaction. Transaction loading skips the first account and
        // loads (and thus inspects) all others normally.
        let Some(mut loaded_fee_payer) =
            account_loader.load_transaction_account(fee_payer_address, true)
        else {
            error_counters.account_not_found += 1;
            return Err(TransactionError::AccountNotFound);
        };

        let fee_payer_loaded_rent_epoch = loaded_fee_payer.account.rent_epoch();
        update_rent_exempt_status_for_account(rent, &mut loaded_fee_payer.account);

        let fee_payer_index = 0;
        validate_fee_payer(
            &mut loaded_fee_payer.account,
            fee_payer_index,
            error_counters,
            rent,
            compute_budget_and_limits.fee_details.total_fee(),
            relax_post_exec_min_balance_check,
        )?;

        // Capture fee-subtracted fee payer account and next nonce account state
        // to commit if transaction execution fails.
        let rollback_accounts = RollbackAccounts::new(
            nonce_info,
            *fee_payer_address,
            loaded_fee_payer.account.clone(),
            fee_payer_loaded_rent_epoch,
        );

        Ok(ValidatedTransactionDetails {
            fee_details: compute_budget_and_limits.fee_details,
            rollback_accounts,
            loaded_accounts_bytes_limit: compute_budget_and_limits.loaded_accounts_data_size_limit,
            compute_budget: compute_budget_and_limits.budget,
            loaded_fee_payer_account: loaded_fee_payer,
        })
    }

    fn validate_transaction_nonce<CB: TransactionProcessingCallback>(
        account_loader: &mut AccountLoader<CB>,
        message: &impl SVMMessage,
        nonce_address: &Pubkey,
        next_durable_nonce: &DurableNonce,
        next_lamports_per_signature: u64,
        strict_nonce_size_check: bool,
        error_counters: &mut TransactionErrorMetrics,
    ) -> TransactionResult<NonceInfo> {
        // If the nonce has been used in this batch already, we must drop the transaction.
        // This is the same as if it was used in different batches in the same slot.
        // It is possible that the nonce account was used, closed, closed and reopened, closed and
        // spoofed by a non-system program, or had its authority changed. Such a transaction cannot
        // be processed, even as fee-only.

        // Check if nonce is writable after write-lock demotion, which runtime may not enforce.
        if !message
            .account_keys()
            .iter()
            .position(|key| key == nonce_address)
            .is_some_and(|index| message.is_writable(index))
        {
            error_counters.blockhash_not_found += 1;
            return Err(TransactionError::BlockhashNotFound);
        }

        let Some(mut nonce_account) = account_loader
            .load_transaction_account(nonce_address, true)
            .map(|loaded| loaded.account)
        else {
            error_counters.account_not_found += 1;
            return Err(TransactionError::AccountNotFound);
        };

        if strict_nonce_size_check && nonce_account.data().len() != NonceState::size() {
            error_counters.blockhash_not_found += 1;
            return Err(TransactionError::BlockhashNotFound);
        }

        // This function verifies:
        // * Nonce account owner is SystemProgram
        // * Nonce account parses as State::Initialized
        // * Stored durable nonce matches the message blockhash
        let Some(nonce_data) = verify_nonce_account(&nonce_account, message.recent_blockhash())
        else {
            error_counters.blockhash_not_found += 1;
            return Err(TransactionError::BlockhashNotFound);
        };

        // We must still check that the nonce account is usable and that its authority has signed.
        let nonce_can_be_advanced = &nonce_data.durable_nonce != next_durable_nonce;
        let nonce_authority_is_valid = message
            .get_ix_signers(NONCED_TX_MARKER_IX_INDEX as usize)
            .any(|signer| signer == &nonce_data.authority);

        if nonce_can_be_advanced && nonce_authority_is_valid {
            let next_nonce_state = NonceState::new_initialized(
                &nonce_data.authority,
                *next_durable_nonce,
                next_lamports_per_signature,
            );
            nonce_account
                .set_state(&NonceVersions::new(next_nonce_state))
                .expect("Serializing into a validated nonce account cannot fail");

            Ok(NonceInfo::new(*nonce_address, nonce_account))
        } else {
            error_counters.blockhash_not_found += 1;
            Err(TransactionError::BlockhashNotFound)
        }
    }

    #[cfg_attr(feature = "dev-context-only-utils", qualifiers(pub))]
    fn replenish_program_cache<CB: TransactionProcessingCallback>(
        &self,
        account_loader: &AccountLoader<CB>,
        mut missing_programs: Vec<ProgramToLoad>,
        program_runtime_environment_for_execution: &ProgramRuntimeEnvironment,
        program_cache_for_tx_batch: &mut ProgramCacheForTxBatch,
        execute_timings: &mut ExecuteTimings,
        limit_to_load_programs: bool,
        increment_usage_counter: bool,
    ) {
        if missing_programs.is_empty() {
            // Nothing to load, so skip the global cache and fork graph locks.
            // Program-cache hit/miss counters are unchanged for empty work.
            return;
        }
        let mut count_hits_and_misses = true;
        loop {
            // Lock the global cache.
            let global_program_cache = self.global_program_cache.read().unwrap();
            // Figure out which program needs to be loaded next.
            let program_to_load = global_program_cache.extract(
                &mut missing_programs,
                program_cache_for_tx_batch,
                program_runtime_environment_for_execution,
                increment_usage_counter,
                count_hits_and_misses,
            );
            count_hits_and_misses = false;
            let task_waiter = Arc::clone(&global_program_cache.loading_task_waiter);
            let task_cookie = task_waiter.cookie();
            // Unlock the global cache again.
            drop(global_program_cache);

            let program_to_store = program_to_load.map(|key| {
                // Load, verify and compile one program.
                let program = load_program_with_pubkey(
                    account_loader,
                    program_runtime_environment_for_execution,
                    &key,
                    self.slot,
                    execute_timings,
                )
                .expect("called load_program_with_pubkey() with nonexistent account");
                (key, program)
            });

            if let Some((key, program)) = program_to_store {
                program_cache_for_tx_batch.loaded_missing = true;
                let mut global_program_cache = self.global_program_cache.write().unwrap();
                // Submit our last completed loading task.
                if global_program_cache.finish_cooperative_loading_task(
                    program_runtime_environment_for_execution,
                    self.slot,
                    key,
                    program,
                ) && limit_to_load_programs
                {
                    // This branch is taken when there is an error in assigning a program to a
                    // cache slot. It is not possible to mock this error for SVM unit
                    // tests purposes.
                    *program_cache_for_tx_batch = ProgramCacheForTxBatch::new(self.slot);
                    program_cache_for_tx_batch.hit_max_limit = true;
                    return;
                }
            } else if missing_programs.is_empty() {
                break;
            } else {
                // Remember: there are multiple transaction processor threads running concurrently
                // and those other threads may be loading this or other programs.
                //
                // So, sleep until some other thread submits a program with their
                // `finish_cooperative_loading_task` call. We'll then wake up and try to load the
                // missing programs inside the tx batch again.
                let _new_cookie = task_waiter.wait(task_cookie);
            }
        }
    }

    /// Similar to replenish_program_cache() but only used in Bank::prepare_program_cache_for_upcoming_feature_set().
    pub fn prepare_one_program_for_upcoming_feature_set<CB: TransactionProcessingCallback>(
        &self,
        account_loader: &CB,
        upcoming_environment: &ProgramRuntimeEnvironment,
        key: &Pubkey,
        stats_of_enqueued_program: &ProgramStatistics,
    ) {
        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(self.slot);
        let mut missing_programs = filter_executable_program_accounts(
            account_loader,
            &program_cache_for_tx_batch,
            std::iter::once(key),
        );
        if missing_programs.is_empty() {
            // Program account was closed
            return;
        }
        let program_to_load = {
            let program_cache_guard = self.global_program_cache.read().unwrap();
            program_cache_guard.extract(
                &mut missing_programs,
                &mut program_cache_for_tx_batch,
                upcoming_environment,
                false, // increment_usage_counter
                false, // count_hits_and_misses
            )
            // Unlock again because load_program_with_pubkey() might take a while.
        };
        // Maybe the enqueued program was already loaded and can be skipped.
        if let Some(key) = program_to_load {
            // Load, verify and compile one program.
            let recompiled = load_program_with_pubkey(
                account_loader,
                upcoming_environment,
                &key,
                self.slot,
                &mut ExecuteTimings::default(),
            )
            .expect("called load_program_with_pubkey() with nonexistent account");
            recompiled.stats.merge_from(stats_of_enqueued_program);
            // Lock the global cache as writable this time.
            let mut program_cache_guard = self.global_program_cache.write().unwrap();
            // Submit our last completed loading task.
            program_cache_guard.finish_cooperative_loading_task(
                upcoming_environment,
                self.slot,
                key,
                recompiled,
            );
        }
    }

    /// Execute a transaction using the provided loaded accounts and update
    /// the executors cache if the transaction was successful.
    #[allow(clippy::too_many_arguments)]
    fn execute_loaded_transaction<CB: InvokeContextCallback>(
        &self,
        callback: &CB,
        tx: &impl SVMTransaction,
        sysvar_cache: &SysvarCache,
        mut loaded_transaction: LoadedTransaction,
        execute_timings: &mut ExecuteTimings,
        error_metrics: &mut TransactionErrorMetrics,
        program_cache_for_tx_batch: &mut ProgramCacheForTxBatch,
        environment: &TransactionProcessingEnvironment,
        config: &TransactionProcessingConfig,
    ) -> ExecutedTransaction {
        let transaction_accounts = std::mem::take(&mut loaded_transaction.accounts);

        // Ensure the length of accounts matches the expected length from tx.account_keys().
        // This is a sanity check in case that someone starts adding some additional accounts
        // since this has been done before. See discussion in PR #4497 for details
        debug_assert!(transaction_accounts.len() == tx.account_keys().len());

        fn transaction_accounts_lamports_sum(
            accounts: &[(Pubkey, AccountSharedData)],
        ) -> Option<u128> {
            accounts.iter().try_fold(0u128, |sum, (_, account)| {
                sum.checked_add(u128::from(account.lamports()))
            })
        }

        let lamports_before_tx =
            transaction_accounts_lamports_sum(&transaction_accounts).unwrap_or(0);

        let compute_budget = loaded_transaction.compute_budget;

        let mut transaction_context = TransactionContext::new_with_feature_flags(
            transaction_accounts,
            environment.rent.clone(),
            compute_budget.max_instruction_stack_depth,
            compute_budget.max_instruction_trace_length,
            tx.num_instructions(),
            ComputeMeter::new(compute_budget.compute_unit_limit),
            if config.drop_bail_out_transactions {
                DropOnBailOut::Enabled
            } else {
                DropOnBailOut::Disabled
            },
        );

        let relax_post_exec_min_balance_check =
            environment.feature_set.relax_post_exec_min_balance_check;
        let pre_account_state_info = TransactionAccountStateInfo::new_pre_exec(
            &transaction_context,
            tx,
            &environment.rent,
            relax_post_exec_min_balance_check,
        );

        let log_collector = if config.recording_config.enable_log_recording {
            match config.log_messages_bytes_limit {
                None => Some(LogCollector::new_ref()),
                Some(log_messages_bytes_limit) => Some(LogCollector::new_ref_with_limit(Some(
                    log_messages_bytes_limit,
                ))),
            }
        } else {
            None
        };

        let mut executed_units = 0u64;

        let mut invoke_context = InvokeContext::new(
            &mut transaction_context,
            program_cache_for_tx_batch,
            EnvironmentConfig::new(
                environment.blockhash,
                environment.blockhash_lamports_per_signature,
                environment.alpenglow_migration_succeeded,
                callback,
                &environment.feature_set,
                &environment.program_runtime_environments,
                sysvar_cache,
            ),
            log_collector.clone(),
            compute_budget,
            self.execution_cost,
        );

        let mut process_message_time = Measure::start("process_message_time");
        let process_result = invoke_context
            .process_message(tx, execute_timings, &mut executed_units)
            .map_err(|(index, err)| TransactionError::InstructionError(index, err));
        process_message_time.stop();

        drop(invoke_context);

        execute_timings.execute_accessories.process_message_us += process_message_time.as_us();

        let mut post_account_state_info_result = process_result
            .and_then(|_info| {
                let post_account_state_info = TransactionAccountStateInfo::new_post_exec(
                    &transaction_context,
                    tx,
                    &pre_account_state_info,
                    &environment.rent,
                    relax_post_exec_min_balance_check,
                );
                verify_changes(
                    &pre_account_state_info,
                    &post_account_state_info,
                    &transaction_context,
                )
                .map(|_| post_account_state_info)
            })
            .map_err(|err| {
                match err {
                    TransactionError::InvalidRentPayingAccount
                    | TransactionError::InsufficientFundsForRent { .. } => {
                        error_metrics.invalid_rent_paying_account += 1;
                    }
                    TransactionError::InvalidAccountIndex => {
                        error_metrics.invalid_account_index += 1;
                    }
                    _ => {
                        error_metrics.instruction_error += 1;
                    }
                }
                err
            });

        let log_messages: Option<TransactionLogMessages> =
            log_collector.and_then(|log_collector| {
                Rc::try_unwrap(log_collector)
                    .map(|log_collector| log_collector.into_inner().into_messages())
                    .ok()
            });

        let (execution_record, inner_instructions) = Self::deconstruct_transaction(
            transaction_context,
            config.recording_config.enable_cpi_recording,
        );

        let ExecutionRecord {
            accounts,
            return_data,
            mut touched_flags,
            accounts_resize_delta,
        } = execution_record;

        // The fee payer (account index 0) is debited during loading, outside the
        // VM, so it carries no VM touch flag but must still be written back.
        if let Some(fee_payer_touched) = touched_flags.first_mut() {
            *fee_payer_touched = true;
        }

        // changed_account_count reflects every account that will be written back,
        // including the fee payer marked above.
        let touched_account_count = touched_flags.iter().filter(|touched| **touched).count();

        if post_account_state_info_result.is_ok()
            && transaction_accounts_lamports_sum(&accounts)
                .filter(|lamports_after_tx| lamports_before_tx == *lamports_after_tx)
                .is_none()
        {
            post_account_state_info_result = Err(TransactionError::UnbalancedTransaction);
        }

        // accounts_resize_delta and accounts_uninitialized_size must be set to None
        // in the result if status is an error
        let (status, accounts_deltas) = post_account_state_info_result
            .map(|post_state_info| {
                (
                    Ok(()),
                    Some(AccountsDeltas {
                        accounts_resize_delta,
                        accounts_uninitialized_size: get_uninitialized_accounts_size(
                            &post_state_info,
                        ),
                    }),
                )
            })
            .unwrap_or_else(|err| (Err(err), None));

        loaded_transaction.accounts = accounts;
        loaded_transaction.touched_flags = touched_flags;
        execute_timings.details.total_account_count += loaded_transaction.accounts.len() as u64;
        execute_timings.details.changed_account_count += touched_account_count as u64;

        let return_data = if config.recording_config.enable_return_data_recording
            && !return_data.data.is_empty()
        {
            Some(return_data)
        } else {
            None
        };

        ExecutedTransaction {
            execution_details: TransactionExecutionDetails {
                status,
                log_messages,
                inner_instructions,
                return_data,
                executed_units,
                accounts_deltas,
            },
            loaded_transaction,
            programs_modified_by_tx: program_cache_for_tx_batch.drain_modified_entries(),
        }
    }

    /// Extract an ExecutionRecord and an InnerInstructionsList from a TransactionContext
    fn deconstruct_transaction(
        mut transaction_context: TransactionContext,
        record_inner_instructions: bool,
    ) -> (ExecutionRecord, Option<InnerInstructionsList>) {
        let inner_ix = if record_inner_instructions {
            debug_assert!(
                transaction_context
                    .get_instruction_context_at_index_in_trace(0)
                    .map(|instruction_context| instruction_context.get_stack_height()
                        == TRANSACTION_LEVEL_STACK_HEIGHT)
                    .unwrap_or(true)
            );

            let top_level_ixs_num = transaction_context
                .get_instruction_trace_length()
                .saturating_sub(transaction_context.number_of_cpis_in_trace());
            // This vector is a map between CPI number in trace (not counting top level
            // instructions) and the top level caller index.
            // In TransactionContext, caller instructions always precede callee instructions, so
            // we can use it to avoid backtracking on instructions callers to
            // find the top level instruction that started the call chain.
            let mut parent_positions: Vec<usize> =
                vec![usize::MAX; transaction_context.number_of_cpis_in_trace()];
            let (ix_trace, accounts, ix_data_trace) = transaction_context.take_instruction_trace();
            let mut outer_instructions: Vec<Vec<InnerInstruction>> =
                vec![Vec::new(); top_level_ixs_num];
            for (cpi_num, ((ix_in_trace, ix_data), ix_accounts)) in ix_trace
                .into_iter()
                .zip(ix_data_trace)
                .zip(accounts)
                .skip(top_level_ixs_num)
                .enumerate()
            {
                let caller_ix = ix_in_trace.index_of_caller_instruction;
                debug_assert_ne!(caller_ix, u16::MAX, "Instruction is not a CPI");

                // If the caller index is less than the number of top level instructions,
                // it directly represents a top level instruction index.
                // Top level instructions precede all CPIs in the instruction trace.
                let outer_index = if (caller_ix as usize) < top_level_ixs_num {
                    *parent_positions.get_mut(cpi_num).unwrap() = caller_ix as usize;
                    caller_ix as usize
                // If the above condition was false, we are dealing with a nested CPI.
                // The caller_ix represents the CPI index in the instruction trace.
                // To calculate its cpi_number (i.e. the index in `parent_positions)`
                // we subtract is from the number of top level instructions.
                } else if let Some(caller_index) = parent_positions
                    .get((caller_ix as usize).saturating_sub(top_level_ixs_num))
                    .copied()
                    && caller_index != usize::MAX
                {
                    *parent_positions.get_mut(cpi_num).unwrap() = caller_index;
                    caller_index
                } else {
                    // This case shall never happen. Program runtime always executes caller before
                    // callees, so the if-statement can only be broken into two different cases:
                    // 1. Top-level instructions doing a CPI
                    // 2. A nested CPI.
                    debug_assert!(false);
                    usize::MAX
                };

                if let Some(inner_instructions) = outer_instructions.get_mut(outer_index) {
                    let stack_height = ix_in_trace.nesting_level.saturating_add(1);
                    let stack_height = u8::try_from(stack_height).unwrap_or(u8::MAX);
                    inner_instructions.push(InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(
                            ix_in_trace.program_account_index_in_tx as u8,
                            ix_data.into_owned(),
                            ix_accounts
                                .iter()
                                .map(|acc| acc.index_in_transaction as u8)
                                .collect(),
                        ),
                        stack_height,
                    });
                } else {
                    debug_assert!(false);
                }
            }

            Some(outer_instructions)
        } else {
            None
        };

        let record: ExecutionRecord = transaction_context.into();

        (record, inner_ix)
    }

    pub fn fill_missing_sysvar_cache_entries<CB: TransactionProcessingCallback>(
        &self,
        callbacks: &CB,
    ) {
        let mut sysvar_cache = self.sysvar_cache.write().unwrap();
        Self::fill_missing_sysvar_cache_entries_from_accounts(&mut sysvar_cache, callbacks);
    }

    pub fn reset_and_fill_sysvar_cache_entries<CB: TransactionProcessingCallback>(
        &self,
        callbacks: &CB,
    ) {
        let mut sysvar_cache = self.sysvar_cache.write().unwrap();
        sysvar_cache.reset();
        Self::fill_missing_sysvar_cache_entries_from_accounts(&mut sysvar_cache, callbacks);
    }

    fn fill_missing_sysvar_cache_entries_from_accounts<CB: TransactionProcessingCallback>(
        sysvar_cache: &mut SysvarCache,
        callbacks: &CB,
    ) {
        sysvar_cache.fill_missing_entries(|pubkey, set_sysvar| {
            if let Some(account) = callbacks.get_account_shared_data(pubkey) {
                set_sysvar(account.data());
            }
        });
    }

    pub fn reset_sysvar_cache(&self) {
        let mut sysvar_cache = self.sysvar_cache.write().unwrap();
        sysvar_cache.reset();
    }

    pub fn get_sysvar_cache_for_tests(&self) -> SysvarCache {
        self.sysvar_cache.read().unwrap().clone()
    }

    /// Add a built-in program to this fork.
    ///
    /// A builtin program should _never_ be inserted directly into the global
    /// program cache without updating the fork guard: `builtin_program_ids`
    /// and `builtin_program_cache`.
    pub fn add_builtin(&self, program_id: Pubkey, builtin: ProgramCacheEntry) {
        self.builtin_program_ids.write().unwrap().insert(program_id);
        let entry = Arc::new(builtin);
        self.global_program_cache.write().unwrap().assign_program(
            &self.program_runtime_environment,
            program_id,
            0,
            Arc::clone(&entry),
        );
        self.builtin_program_cache
            .write()
            .unwrap()
            .replenish(program_id, entry);
    }

    /// Remove a builtin-program from this fork.
    ///
    /// This removes the builtin from the fork guard (`builtin_program_ids` and
    /// `builtin_program_cache`), but it does not remove it from the global
    /// program cache. Another fork could be relying on the global entry.
    pub fn remove_builtin(&self, program_id: &Pubkey) {
        self.builtin_program_cache
            .write()
            .unwrap()
            .remove_entry(program_id);
        self.builtin_program_ids.write().unwrap().remove(program_id);
    }

    #[cfg(feature = "dev-context-only-utils")]
    #[cfg_attr(feature = "dev-context-only-utils", qualifiers(pub))]
    fn writable_sysvar_cache(&self) -> &RwLock<SysvarCache> {
        &self.sysvar_cache
    }
}

#[cfg(test)]
mod tests {
    #[allow(deprecated)]
    use solana_sysvar::fees::Fees;
    use {
        super::*,
        crate::{
            account_loader::{
                LoadedTransactionAccount, TRANSACTION_ACCOUNT_BASE_SIZE,
                ValidatedTransactionDetails,
            },
            nonce_info::NonceInfo,
            program_loader::test_utils::*,
            rent_calculator::RENT_EXEMPT_RENT_EPOCH,
            rollback_accounts::RollbackAccounts,
        },
        solana_account::WritableAccount,
        solana_clock::Clock,
        solana_compute_budget_interface::ComputeBudgetInstruction,
        solana_epoch_schedule::EpochSchedule,
        solana_fee_calculator::FeeCalculator,
        solana_fee_structure::FeeDetails,
        solana_hash::Hash,
        solana_loader_v4_interface::state::{LoaderV4State, LoaderV4Status},
        solana_message::{LegacyMessage, Message, MessageHeader, SanitizedMessage},
        solana_nonce as nonce,
        solana_program_runtime::{
            execution_budget::{
                SVMTransactionExecutionAndFeeBudgetLimits, SVMTransactionExecutionBudget,
            },
            invoke_context::BuiltinFunctionRegisterer,
            loaded_programs::BlockRelation,
            program_cache_entry::ProgramCacheEntryType,
        },
        solana_rent::Rent,
        solana_sbpf::vm,
        solana_sdk_ids::{
            bpf_loader, bpf_loader_upgradeable, loader_v4, native_loader, system_program, sysvar,
        },
        solana_signature::Signature,
        solana_svm_callback::{AccountState, InvokeContextCallback},
        solana_svm_type_overrides::sync::atomic::Ordering,
        solana_system_interface::instruction as system_instruction,
        solana_sysvar_id::SysvarId,
        solana_transaction::sanitized::SanitizedTransaction,
        solana_transaction_context::transaction::TransactionContext,
        solana_transaction_error::TransactionError,
        std::{borrow::Cow, collections::HashMap},
        test_case::test_case,
    };

    fn create_sysvar_account<T>(value: &T) -> AccountSharedData
    where
        T: wincode::Serialize<Src = T> + SysvarId,
    {
        let serialized_len = wincode::serialized_size(value).unwrap() as usize;
        let canonical_data_len = match T::id() {
            sysvar::clock::ID => solana_clock::SIZE,
            sysvar::epoch_schedule::ID => solana_epoch_schedule::SIZE,
            sysvar::fees::ID => solana_sysvar::fees::SIZE,
            sysvar::rent::ID => solana_rent::SIZE,
            id => panic!("unsupported sysvar: {id}"),
        };
        let required_data_len = canonical_data_len.max(serialized_len);
        let mut account = AccountSharedData::new(1, required_data_len, &sysvar::id());
        wincode::serialize_into(account.data_as_mut_slice(), value).unwrap();
        account
    }

    fn new_unchecked_sanitized_message(message: Message) -> SanitizedMessage {
        SanitizedMessage::Legacy(LegacyMessage::new(message, &HashSet::new()))
    }

    struct TestForkGraph {}

    impl ForkGraph for TestForkGraph {
        fn relationship(&self, _a: Slot, _b: Slot) -> BlockRelation {
            BlockRelation::Unknown
        }
    }

    #[derive(Clone)]
    struct MockBankCallback {
        account_shared_data: Arc<RwLock<HashMap<Pubkey, AccountSharedData>>>,
        #[allow(clippy::type_complexity)]
        inspected_accounts:
            Arc<RwLock<HashMap<Pubkey, Vec<(Option<AccountSharedData>, /* is_writable */ bool)>>>>,
        feature_set: SVMFeatureSet,
    }

    impl Default for MockBankCallback {
        fn default() -> Self {
            Self {
                account_shared_data: Arc::default(),
                inspected_accounts: Arc::default(),
                feature_set: SVMFeatureSet::all_enabled(),
            }
        }
    }

    impl InvokeContextCallback for MockBankCallback {}

    impl TransactionProcessingCallback for MockBankCallback {
        fn get_account_shared_data(&self, pubkey: &Pubkey) -> Option<AccountSharedData> {
            self.account_shared_data
                .read()
                .unwrap()
                .get(pubkey)
                .cloned()
        }

        fn inspect_account(
            &self,
            address: &Pubkey,
            account_state: AccountState,
            is_writable: bool,
        ) {
            let account = match account_state {
                AccountState::Dead => None,
                AccountState::Alive(account) => Some(account.clone()),
            };
            self.inspected_accounts
                .write()
                .unwrap()
                .entry(*address)
                .or_default()
                .push((account, is_writable));
        }
    }

    impl MockBankCallback {
        pub fn calculate_fee_details(
            message: &impl SVMMessage,
            lamports_per_signature: u64,
            prioritization_fee: u64,
        ) -> FeeDetails {
            let signature_count = message
                .num_transaction_signatures()
                .saturating_add(message.num_ed25519_signatures())
                .saturating_add(message.num_secp256k1_signatures())
                .saturating_add(message.num_secp256r1_signatures());

            FeeDetails::new(
                signature_count.saturating_mul(lamports_per_signature),
                prioritization_fee,
            )
        }
    }

    impl<'a> From<&'a MockBankCallback> for AccountLoader<'a, MockBankCallback> {
        fn from(callbacks: &'a MockBankCallback) -> AccountLoader<'a, MockBankCallback> {
            AccountLoader::new_with_loaded_accounts_capacity(
                None,
                callbacks,
                &callbacks.feature_set,
                0,
            )
        }
    }

    #[test_case(1; "Check results too small")]
    #[test_case(3; "Check results too large")]
    #[should_panic(expected = "Length of check_results does not match length of sanitized_txs")]
    fn test_check_results_txs_length_mismatch(check_results_len: usize) {
        let sanitized_message = new_unchecked_sanitized_message(Message {
            account_keys: vec![Pubkey::new_from_array([0; 32])],
            header: MessageHeader::default(),
            instructions: vec![CompiledInstruction {
                program_id_index: 0,
                accounts: vec![],
                data: vec![],
            }],
            recent_blockhash: Hash::default(),
        });

        // Transactions, length 2.
        let sanitized_txs = vec![
            SanitizedTransaction::new_for_tests(
                sanitized_message,
                vec![Signature::new_unique()],
                false,
            );
            2
        ];

        let check_results = vec![
            TransactionCheckResult::Ok(CheckedTransactionDetails::default());
            check_results_len
        ];

        let batch_processor = TransactionBatchProcessor::<TestForkGraph>::default();
        let callback = MockBankCallback::default();

        batch_processor.load_and_execute_sanitized_transactions(
            &callback,
            &sanitized_txs,
            check_results,
            &get_mock_transaction_processing_environment(),
            &TransactionProcessingConfig::default(),
        );
    }

    #[test]
    fn test_inner_instructions_list_from_instruction_trace() {
        let mut transaction_context = TransactionContext::new(
            vec![(
                Pubkey::new_unique(),
                AccountSharedData::new(1, 1, &bpf_loader::ID),
            )],
            Rent::default(),
            4,
            11,
            4,
        );
        let num_transaction_accounts = usize::from(transaction_context.get_number_of_accounts());

        // Four top level instructions
        for i in 0..4 {
            transaction_context
                .configure_instruction_at_index(
                    i,
                    0,
                    vec![],
                    vec![u8::MAX; num_transaction_accounts],
                    Cow::Owned(vec![i as u8]),
                    None,
                )
                .unwrap();
        }

        // Execute ix #0
        transaction_context.push().unwrap();
        // ix #0 does a CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![0, 0])
            .unwrap();
        transaction_context.push().unwrap();
        // Returning from everything
        transaction_context.pop().unwrap();
        transaction_context.pop().unwrap();
        // Execute ix #1
        transaction_context.push().unwrap();
        transaction_context.pop().unwrap();
        // Execute ix #2
        transaction_context.push().unwrap();
        // ix #2 does a CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![2, 0])
            .unwrap();
        transaction_context.push().unwrap();
        // A nested CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![2, 1])
            .unwrap();
        transaction_context.push().unwrap();
        // Return from nested CPI
        transaction_context.pop().unwrap();
        // Return from CPI
        transaction_context.pop().unwrap();
        // ix #2 does another CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![2, 2])
            .unwrap();
        transaction_context.push().unwrap();
        // Return from everything related to ix #2
        transaction_context.pop().unwrap();
        transaction_context.pop().unwrap();
        // Execute ix #3
        transaction_context.push().unwrap();
        // ix #3 does a CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![3, 0])
            .unwrap();
        transaction_context.push().unwrap();
        // ix #3 does a nested CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![3, 1])
            .unwrap();
        transaction_context.push().unwrap();
        // ix #3 does a second nested CPI
        transaction_context
            .configure_next_cpi_for_tests(0, vec![], vec![3, 2])
            .unwrap();
        transaction_context.push().unwrap();
        // Return from everything related to ix #3
        transaction_context.pop().unwrap();
        transaction_context.pop().unwrap();
        transaction_context.pop().unwrap();
        transaction_context.pop().unwrap();

        let inner_instructions =
            TransactionBatchProcessor::<TestForkGraph>::deconstruct_transaction(
                transaction_context,
                true,
            )
            .1
            .unwrap();

        assert_eq!(
            inner_instructions,
            vec![
                vec![InnerInstruction {
                    instruction: CompiledInstruction::new_from_raw_parts(0, vec![0, 0], vec![]),
                    stack_height: 2,
                }],
                vec![],
                vec![
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![2, 0], vec![]),
                        stack_height: 2,
                    },
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![2, 1], vec![]),
                        stack_height: 3,
                    },
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![2, 2], vec![]),
                        stack_height: 2,
                    },
                ],
                vec![
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![3, 0], vec![]),
                        stack_height: 2,
                    },
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![3, 1], vec![]),
                        stack_height: 3,
                    },
                    InnerInstruction {
                        instruction: CompiledInstruction::new_from_raw_parts(0, vec![3, 2], vec![]),
                        stack_height: 4,
                    },
                ]
            ]
        );
    }

    #[test]
    fn test_execute_loaded_transaction_recordings() {
        // Setting all the arguments correctly is too burdensome for testing
        // execute_loaded_transaction separately.This function will be tested in an integration
        // test with load_and_execute_sanitized_transactions
        let message = Message {
            account_keys: vec![Pubkey::new_from_array([0; 32])],
            header: MessageHeader::default(),
            instructions: vec![CompiledInstruction {
                program_id_index: 0,
                accounts: vec![],
                data: vec![],
            }],
            recent_blockhash: Hash::default(),
        };

        let sanitized_message = new_unchecked_sanitized_message(message);
        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::default();
        let batch_processor = TransactionBatchProcessor::<TestForkGraph>::default();

        let sanitized_transaction = SanitizedTransaction::new_for_tests(
            sanitized_message,
            vec![Signature::new_unique()],
            false,
        );

        let loaded_transaction = LoadedTransaction {
            accounts: vec![(Pubkey::new_unique(), AccountSharedData::default())],
            touched_flags: Box::default(),
            fee_details: FeeDetails::default(),
            rollback_accounts: RollbackAccounts::default(),
            compute_budget: SVMTransactionExecutionBudget::default(),
            loaded_accounts_data_size: 32,
        };

        let processing_environment = get_mock_transaction_processing_environment();

        let mut processing_config = TransactionProcessingConfig::default();
        processing_config.recording_config.enable_log_recording = true;

        let mock_bank = MockBankCallback::default();
        let sysvar_cache = batch_processor.sysvar_cache();

        let executed_tx = batch_processor.execute_loaded_transaction(
            &mock_bank,
            &sanitized_transaction,
            &sysvar_cache,
            loaded_transaction.clone(),
            &mut ExecuteTimings::default(),
            &mut TransactionErrorMetrics::default(),
            &mut program_cache_for_tx_batch,
            &processing_environment,
            &processing_config,
        );
        assert!(executed_tx.execution_details.log_messages.is_some());

        processing_config.log_messages_bytes_limit = Some(2);

        let executed_tx = batch_processor.execute_loaded_transaction(
            &mock_bank,
            &sanitized_transaction,
            &sysvar_cache,
            loaded_transaction.clone(),
            &mut ExecuteTimings::default(),
            &mut TransactionErrorMetrics::default(),
            &mut program_cache_for_tx_batch,
            &processing_environment,
            &processing_config,
        );
        assert!(executed_tx.execution_details.log_messages.is_some());
        assert!(executed_tx.execution_details.inner_instructions.is_none());

        processing_config.recording_config.enable_log_recording = false;
        processing_config.recording_config.enable_cpi_recording = true;
        processing_config.log_messages_bytes_limit = None;

        let executed_tx = batch_processor.execute_loaded_transaction(
            &mock_bank,
            &sanitized_transaction,
            &sysvar_cache,
            loaded_transaction,
            &mut ExecuteTimings::default(),
            &mut TransactionErrorMetrics::default(),
            &mut program_cache_for_tx_batch,
            &processing_environment,
            &processing_config,
        );

        assert!(executed_tx.execution_details.log_messages.is_none());
        assert!(executed_tx.execution_details.inner_instructions.is_some());
    }

    #[test]
    fn test_execute_loaded_transaction_error_metrics() {
        // Setting all the arguments correctly is too burdensome for testing
        // execute_loaded_transaction separately.This function will be tested in an integration
        // test with load_and_execute_sanitized_transactions
        let key1 = Pubkey::new_unique();
        let key2 = Pubkey::new_unique();
        let message = Message {
            account_keys: vec![key1, key2],
            header: MessageHeader::default(),
            instructions: vec![CompiledInstruction {
                program_id_index: 0,
                accounts: vec![1],
                data: vec![],
            }],
            recent_blockhash: Hash::default(),
        };

        let sanitized_message = new_unchecked_sanitized_message(message);
        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::default();
        let batch_processor = TransactionBatchProcessor::<TestForkGraph>::default();

        let sanitized_transaction = SanitizedTransaction::new_for_tests(
            sanitized_message,
            vec![Signature::new_unique()],
            false,
        );

        let loaded_transaction = LoadedTransaction {
            accounts: vec![
                (key1, AccountSharedData::default()),
                (key2, AccountSharedData::default()),
            ],
            touched_flags: Box::default(),
            fee_details: FeeDetails::default(),
            rollback_accounts: RollbackAccounts::default(),
            compute_budget: SVMTransactionExecutionBudget::default(),
            loaded_accounts_data_size: 0,
        };

        let processing_config = TransactionProcessingConfig {
            recording_config: ExecutionRecordingConfig::new_single_setting(false),
            ..Default::default()
        };
        let mut error_metrics = TransactionErrorMetrics::new();
        let mock_bank = MockBankCallback::default();
        let sysvar_cache = batch_processor.sysvar_cache();

        let _ = batch_processor.execute_loaded_transaction(
            &mock_bank,
            &sanitized_transaction,
            &sysvar_cache,
            loaded_transaction,
            &mut ExecuteTimings::default(),
            &mut error_metrics,
            &mut program_cache_for_tx_batch,
            &get_mock_transaction_processing_environment(),
            &processing_config,
        );

        assert_eq!(error_metrics.instruction_error.0, 1);
    }

    #[test]
    #[allow(deprecated)]
    fn test_sysvar_cache_initialization1() {
        let mock_bank = MockBankCallback::default();

        let clock = Clock {
            slot: 1,
            epoch_start_timestamp: 2,
            epoch: 3,
            leader_schedule_epoch: 4,
            unix_timestamp: 5,
        };
        let clock_account = create_sysvar_account(&clock);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::clock::id(), clock_account);

        let epoch_schedule = EpochSchedule::custom(64, 2, true);
        let epoch_schedule_account = create_sysvar_account(&epoch_schedule);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::epoch_schedule::id(), epoch_schedule_account);

        let fees = Fees {
            fee_calculator: FeeCalculator {
                lamports_per_signature: 123,
            },
        };
        let fees_account = create_sysvar_account(&fees);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::fees::id(), fees_account);

        let rent = Rent::default();
        let rent_account = create_sysvar_account(&rent);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::rent::id(), rent_account);

        let transaction_processor = TransactionBatchProcessor::<TestForkGraph>::default();
        transaction_processor.fill_missing_sysvar_cache_entries(&mock_bank);

        let sysvar_cache = transaction_processor.sysvar_cache.read().unwrap();
        let cached_clock = sysvar_cache.get_clock();
        let cached_epoch_schedule = sysvar_cache.get_epoch_schedule();
        let cached_fees = sysvar_cache.get_fees();
        let cached_rent = sysvar_cache.get_rent();

        assert_eq!(
            cached_clock.expect("clock sysvar missing in cache"),
            clock.into()
        );
        assert_eq!(
            cached_epoch_schedule.expect("epoch_schedule sysvar missing in cache"),
            epoch_schedule.into()
        );
        assert_eq!(
            cached_fees.expect("fees sysvar missing in cache"),
            fees.into()
        );
        assert_eq!(
            cached_rent.expect("rent sysvar missing in cache"),
            rent.into()
        );
        assert!(sysvar_cache.get_slot_hashes().is_err());
        assert!(sysvar_cache.get_epoch_rewards().is_err());
    }

    #[test]
    #[allow(deprecated)]
    fn test_reset_and_fill_sysvar_cache() {
        let mock_bank = MockBankCallback::default();

        let clock = Clock {
            slot: 1,
            epoch_start_timestamp: 2,
            epoch: 3,
            leader_schedule_epoch: 4,
            unix_timestamp: 5,
        };
        let clock_account = create_sysvar_account(&clock);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::clock::id(), clock_account);

        let epoch_schedule = EpochSchedule::custom(64, 2, true);
        let epoch_schedule_account = create_sysvar_account(&epoch_schedule);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::epoch_schedule::id(), epoch_schedule_account);

        let fees = Fees {
            fee_calculator: FeeCalculator {
                lamports_per_signature: 123,
            },
        };
        let fees_account = create_sysvar_account(&fees);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::fees::id(), fees_account);

        let rent = Rent::default();
        let rent_account = create_sysvar_account(&rent);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::rent::id(), rent_account);

        let transaction_processor = TransactionBatchProcessor::<TestForkGraph>::default();
        // Fill the sysvar cache
        transaction_processor.fill_missing_sysvar_cache_entries(&mock_bank);

        let updated_clock = Clock {
            slot: 6,
            epoch_start_timestamp: 7,
            epoch: 8,
            leader_schedule_epoch: 9,
            unix_timestamp: 10,
        };
        let updated_clock_account = create_sysvar_account(&updated_clock);
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(sysvar::clock::id(), updated_clock_account);
        transaction_processor.reset_and_fill_sysvar_cache_entries(&mock_bank);
        {
            let sysvar_cache = transaction_processor.sysvar_cache.read().unwrap();
            assert_eq!(
                sysvar_cache
                    .get_clock()
                    .expect("clock sysvar missing in cache"),
                updated_clock.clone().into()
            );
        }

        // Reset the sysvar cache
        transaction_processor.reset_sysvar_cache();

        {
            let sysvar_cache = transaction_processor.sysvar_cache.read().unwrap();
            // Test that sysvar cache is empty and none of the values are found
            assert!(sysvar_cache.get_clock().is_err());
            assert!(sysvar_cache.get_epoch_schedule().is_err());
            assert!(sysvar_cache.get_fees().is_err());
            assert!(sysvar_cache.get_epoch_rewards().is_err());
            assert!(sysvar_cache.get_rent().is_err());
            assert!(sysvar_cache.get_epoch_rewards().is_err());
        }

        // Refill the cache and test the values are available.
        transaction_processor.fill_missing_sysvar_cache_entries(&mock_bank);

        let sysvar_cache = transaction_processor.sysvar_cache.read().unwrap();
        let cached_clock = sysvar_cache.get_clock();
        let cached_epoch_schedule = sysvar_cache.get_epoch_schedule();
        let cached_fees = sysvar_cache.get_fees();
        let cached_rent = sysvar_cache.get_rent();

        assert_eq!(
            cached_clock.expect("clock sysvar missing in cache"),
            updated_clock.into()
        );
        assert_eq!(
            cached_epoch_schedule.expect("epoch_schedule sysvar missing in cache"),
            epoch_schedule.into()
        );
        assert_eq!(
            cached_fees.expect("fees sysvar missing in cache"),
            fees.into()
        );
        assert_eq!(
            cached_rent.expect("rent sysvar missing in cache"),
            rent.into()
        );
        assert!(sysvar_cache.get_slot_hashes().is_err());
        assert!(sysvar_cache.get_epoch_rewards().is_err());
    }

    #[test]
    fn test_add_builtin() {
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(0, 0, Arc::downgrade(&fork_graph), None);

        let key = Pubkey::new_unique();
        let register_fn: BuiltinFunctionRegisterer = |p, n| {
            p.register_function(
                n,
                (
                    |_invoke_context, _param0, _param1, _param2, _param3, _param4| {},
                    |_| {},
                ),
            )
        };
        let program = ProgramCacheEntry::new_builtin(register_fn);
        batch_processor.add_builtin(key, program);

        let mut loaded_programs_for_tx_batch = ProgramCacheForTxBatch::new(0);
        let program_runtime_environment =
            batch_processor.program_runtime_environment_for_epoch(batch_processor.epoch);
        batch_processor
            .global_program_cache
            .write()
            .unwrap()
            .extract(
                &mut vec![ProgramToLoad {
                    program_id: &key,
                    loader: ProgramCacheEntryOwner::NativeLoader,
                    deployment_slot: 0,
                }],
                &mut loaded_programs_for_tx_batch,
                &program_runtime_environment,
                true,
                true,
            );
        let entry = loaded_programs_for_tx_batch.find(&key).unwrap();

        // Repeating code because ProgramCacheEntry does not implement clone.
        let program = ProgramCacheEntry::new_builtin(register_fn);
        assert_eq!(entry, Arc::new(program));
    }

    #[test]
    fn test_builtin_in_global_cache_but_not_in_fork_is_not_extracted() {
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(0, 0, Arc::downgrade(&fork_graph), None);

        let key = Pubkey::new_unique();
        let register_fn: BuiltinFunctionRegisterer = |p, n| {
            p.register_function(
                n,
                (
                    |_invoke_context, _param0, _param1, _param2, _param3, _param4| {},
                    |_| {},
                ),
            )
        };

        // Insert the builtin into the global program cache only, but bypass the
        // `builtin_program_ids` fork guard.
        batch_processor
            .global_program_cache
            .write()
            .unwrap()
            .assign_program(
                &batch_processor.program_runtime_environment,
                key,
                0,
                Arc::new(ProgramCacheEntry::new_builtin(register_fn)),
            );
        assert!(
            !batch_processor
                .builtin_program_ids
                .read()
                .unwrap()
                .contains(&key)
        );

        // Roll over into another slot.
        // Assert this bank does not pick up the inserted builtin.
        let next_slot = batch_processor.new_from(1, 0);
        assert!(!next_slot.builtin_program_ids.read().unwrap().contains(&key));
        assert!(
            next_slot
                .builtin_program_cache
                .read()
                .unwrap()
                .find(&key)
                .is_none()
        );

        // Now register the address on this fork.
        // Assert the builtin is still not available without a slot rollover.
        batch_processor
            .builtin_program_ids
            .write()
            .unwrap()
            .insert(key);
        assert!(
            next_slot
                .builtin_program_cache
                .read()
                .unwrap()
                .find(&key)
                .is_none()
        );

        // Now roll over into another slot.
        // Assert the builtin was extracted from the global cache.
        let next_slot = batch_processor.new_from(2, 0);
        let entry = next_slot
            .builtin_program_cache
            .read()
            .unwrap()
            .find(&key)
            .unwrap();
        assert_eq!(entry, Arc::new(ProgramCacheEntry::new_builtin(register_fn)));
    }

    #[test]
    fn test_builtin_not_in_fork_is_never_searched_for() {
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(0, 0, Arc::downgrade(&fork_graph), None);

        let key = Pubkey::new_unique();
        let register_fn: BuiltinFunctionRegisterer = |p, n| {
            p.register_function(
                n,
                (
                    |_invoke_context, _param0, _param1, _param2, _param3, _param4| {},
                    |_| {},
                ),
            )
        };

        // Insert a builtin into the global program cache.
        // This is a builtin added by another fork, but not by ours.
        batch_processor
            .global_program_cache
            .write()
            .unwrap()
            .assign_program(
                &batch_processor.program_runtime_environment,
                key,
                0,
                Arc::new(ProgramCacheEntry::new_builtin(register_fn)),
            );

        // Assert that a transaction naming the builtin's address does not
        // tamper with the global program cache.
        let mock_bank = MockBankCallback::default();
        for account_exists in [false, true] {
            if account_exists {
                // Even if the account exists owned by the native loader,
                // the invariant should hold.
                mock_bank
                    .account_shared_data
                    .write()
                    .unwrap()
                    .insert(key, AccountSharedData::new(1, 1, &native_loader::id()));
            }

            let program_cache_for_tx_batch = batch_processor
                .builtin_program_cache
                .read()
                .unwrap()
                .clone();
            assert!(program_cache_for_tx_batch.find(&key).is_none());

            let search_for = filter_executable_program_accounts(
                &mock_bank,
                &program_cache_for_tx_batch,
                std::iter::once(&key),
            );
            assert!(search_for.is_empty());
        }
    }

    #[test]
    fn test_core_bpf_migration_of_a_builtin() {
        const BUILTIN_SLOT: Slot = 0;
        const MIGRATION_SLOT: Slot = 10;
        const NEXT_SLOT: Slot = 11;

        struct SingleChainForkGraph;
        impl ForkGraph for SingleChainForkGraph {
            fn relationship(&self, a: Slot, b: Slot) -> BlockRelation {
                if a == b {
                    BlockRelation::Equal
                } else if a < b {
                    BlockRelation::Ancestor
                } else {
                    BlockRelation::Descendant
                }
            }
        }

        let fork_graph = Arc::new(RwLock::new(SingleChainForkGraph));
        let batch_processor =
            TransactionBatchProcessor::new(BUILTIN_SLOT, 0, Arc::downgrade(&fork_graph), None);

        let key = Pubkey::new_unique();
        let register_fn: BuiltinFunctionRegisterer = |p, n| {
            p.register_function(
                n,
                (
                    |_invoke_context, _param0, _param1, _param2, _param3, _param4| {},
                    |_| {},
                ),
            )
        };

        // Start with a properly guarded builtin: registered on the fork and
        // present in the global program cache.
        batch_processor.add_builtin(key, ProgramCacheEntry::new_builtin(register_fn));
        assert!(
            batch_processor
                .builtin_program_ids
                .read()
                .unwrap()
                .contains(&key)
        );

        // Simulate migrating the builtin program to Core BPF: "deploy" the
        // program by assigning it a new Unloaded entry owned by Loader V3 in
        // the global program cache, then remove it from `builtin_program_ids`.
        batch_processor
            .global_program_cache
            .write()
            .unwrap()
            .assign_program(
                &batch_processor.program_runtime_environment,
                key,
                MIGRATION_SLOT,
                Arc::new(ProgramCacheEntry::new_unloaded(
                    MIGRATION_SLOT,
                    ProgramCacheEntryOwner::LoaderV3,
                    batch_processor.program_runtime_environment.clone(),
                )),
            );
        batch_processor.remove_builtin(&key);

        // The builtin leaves the batch cache in the same slot.
        let program_cache_for_tx_batch = batch_processor
            .builtin_program_cache
            .read()
            .unwrap()
            .clone();
        assert!(program_cache_for_tx_batch.find(&key).is_none());

        // The new account state seeds the extraction search, and since the
        // program was just "deployed" in this slot, we get a delayed visibility
        // tombstone.
        let mut search_for = vec![ProgramToLoad {
            program_id: &key,
            loader: ProgramCacheEntryOwner::LoaderV3,
            deployment_slot: MIGRATION_SLOT,
        }];
        let mut extracted = ProgramCacheForTxBatch::new(MIGRATION_SLOT);
        batch_processor
            .global_program_cache
            .read()
            .unwrap()
            .extract(
                &mut search_for,
                &mut extracted,
                &batch_processor.program_runtime_environment,
                false,
                false,
            );
        assert!(matches!(
            extracted.find(&key).unwrap().program,
            ProgramCacheEntryType::DelayVisibility
        ));

        // Rolling to the next slot rebuilds the batch cache from the fork guard,
        // which no longer lists the address, so the builtin is gone.
        let next_slot = batch_processor.new_from(NEXT_SLOT, 0);
        assert!(
            next_slot
                .builtin_program_cache
                .read()
                .unwrap()
                .find(&key)
                .is_none()
        );

        // A caller reporting the deployed slot now reaches the migrated program's
        // entry, which is unloaded, so the cache signals a reload from account
        // state. Crucially the builtin is not served in its place.
        let mut search_for = vec![ProgramToLoad {
            program_id: &key,
            loader: ProgramCacheEntryOwner::LoaderV3,
            deployment_slot: MIGRATION_SLOT,
        }];
        let mut extracted = ProgramCacheForTxBatch::new(NEXT_SLOT);
        next_slot.global_program_cache.read().unwrap().extract(
            &mut search_for,
            &mut extracted,
            &next_slot.program_runtime_environment,
            false,
            false,
        );
        assert!(extracted.find(&key).is_none());
        assert_eq!(search_for.len(), 1);

        // The builtin entry does survive in the global program cache, but keyed at
        // slot 0 under `NativeLoader`, which the lookup above cannot reach.
        let mut search_for = vec![ProgramToLoad {
            program_id: &key,
            loader: ProgramCacheEntryOwner::NativeLoader,
            deployment_slot: BUILTIN_SLOT,
        }];
        let mut extracted = ProgramCacheForTxBatch::new(NEXT_SLOT);
        next_slot.global_program_cache.read().unwrap().extract(
            &mut search_for,
            &mut extracted,
            &next_slot.program_runtime_environment,
            false,
            false,
        );
        assert!(matches!(
            extracted.find(&key).unwrap().program,
            ProgramCacheEntryType::Builtin(_)
        ));
    }

    #[test]
    #[should_panic(expected = "missing from the global program cache")]
    fn test_builtin_in_fork_but_not_in_global_cache_panics() {
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(0, 0, Arc::downgrade(&fork_graph), None);

        // Register a builtin without going through `add_builtin()` and just
        // add it to `builtin_program_ids` directly. This will trip the debug
        // assertion for the fork guard's `search_for`.
        batch_processor
            .builtin_program_ids
            .write()
            .unwrap()
            .insert(Pubkey::new_unique());
        batch_processor.new_from(1, 0);
    }

    #[test]
    fn test_validate_transaction_fee_payer_exact_balance() {
        let lamports_per_signature = 5000;
        let message = new_unchecked_sanitized_message(Message::new_with_blockhash(
            &[
                ComputeBudgetInstruction::set_compute_unit_limit(2000u32),
                ComputeBudgetInstruction::set_compute_unit_price(1_000_000_000),
            ],
            Some(&Pubkey::new_unique()),
            &Hash::new_unique(),
        ));
        let fee_payer_address = message.fee_payer();
        let current_epoch = 42;
        let rent = Rent::default();
        let min_balance = rent.minimum_balance(nonce::state::State::size());
        let transaction_fee = lamports_per_signature;
        let priority_fee = 2_000_000u64;
        let starting_balance = transaction_fee + priority_fee;
        assert!(
            starting_balance > min_balance,
            "we're testing that a rent exempt fee payer can be fully drained, so ensure that the \
             starting balance is more than the min balance"
        );

        let fee_payer_rent_epoch = current_epoch;
        let fee_payer_account = AccountSharedData::new_rent_epoch(
            starting_balance,
            0,
            &Pubkey::default(),
            fee_payer_rent_epoch,
        );
        let mut mock_accounts = HashMap::new();
        mock_accounts.insert(*fee_payer_address, fee_payer_account.clone());
        let mock_bank = MockBankCallback {
            account_shared_data: Arc::new(RwLock::new(mock_accounts)),
            ..Default::default()
        };
        let mut account_loader = (&mock_bank).into();

        let mut error_counters = TransactionErrorMetrics::default();
        let compute_budget_and_limits = SVMTransactionExecutionAndFeeBudgetLimits {
            budget: SVMTransactionExecutionBudget {
                compute_unit_limit: 2000,
                ..SVMTransactionExecutionBudget::default()
            },
            fee_details: FeeDetails::new(transaction_fee, priority_fee),
            ..SVMTransactionExecutionAndFeeBudgetLimits::default()
        };
        let result =
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(None, compute_budget_and_limits),
                &Hash::default(),
                lamports_per_signature,
                &rent,
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

        let post_validation_fee_payer_account = {
            let mut account = fee_payer_account.clone();
            account.set_rent_epoch(RENT_EXEMPT_RENT_EPOCH);
            account.set_lamports(0);
            account
        };

        assert_eq!(
            result,
            TransactionValidationResult::Loadable(ValidatedTransactionDetails {
                rollback_accounts: RollbackAccounts::new(
                    None, // nonce
                    *fee_payer_address,
                    post_validation_fee_payer_account.clone(),
                    fee_payer_rent_epoch
                ),
                compute_budget: compute_budget_and_limits.budget,
                loaded_accounts_bytes_limit: compute_budget_and_limits
                    .loaded_accounts_data_size_limit,
                fee_details: FeeDetails::new(transaction_fee, priority_fee),
                loaded_fee_payer_account: LoadedTransactionAccount {
                    loaded_size: TRANSACTION_ACCOUNT_BASE_SIZE + fee_payer_account.data().len(),
                    account: post_validation_fee_payer_account,
                },
            })
        );
    }

    #[test_case(false; "strict_fee_payer")]
    #[test_case(true; "relaxed_fee_payer")]
    fn test_validate_transaction_fee_payer_not_found(relax_fee_payer_constraint: bool) {
        let lamports_per_signature = 5000;
        let message =
            new_unchecked_sanitized_message(Message::new(&[], Some(&Pubkey::new_unique())));

        let fee_and_limits = SVMTransactionExecutionAndFeeBudgetLimits::default();
        let mut mock_bank = MockBankCallback::default();
        mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
        let mut account_loader = (&mock_bank).into();
        let mut error_counters = TransactionErrorMetrics::default();

        let expected_error = TransactionError::AccountNotFound;
        let expected_result = if relax_fee_payer_constraint {
            TransactionValidationResult::NoOp(NoOpTransaction {
                validation_error: expected_error,
                fee_payer_balance: None,
                compute_unit_limit: fee_and_limits.budget.compute_unit_limit,
                loaded_accounts_bytes_limit: fee_and_limits.loaded_accounts_data_size_limit,
                nonce_address: None,
            })
        } else {
            TransactionValidationResult::Unprocessable(expected_error)
        };

        let actual_result =
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(None, fee_and_limits),
                &Hash::default(),
                lamports_per_signature,
                &Rent::default(),
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

        assert_eq!(error_counters.account_not_found.0, 1);
        assert_eq!(actual_result, expected_result);
    }

    #[test_case(false; "strict_fee_payer")]
    #[test_case(true; "relaxed_fee_payer")]
    fn test_validate_transaction_fee_payer_insufficient_funds(relax_fee_payer_constraint: bool) {
        let lamports_per_signature = 5000;
        let fee_payer_balance = 1;
        let message =
            new_unchecked_sanitized_message(Message::new(&[], Some(&Pubkey::new_unique())));
        let fee_details =
            MockBankCallback::calculate_fee_details(&message, lamports_per_signature, 0);
        let fee_and_limits = SVMTransactionExecutionAndFeeBudgetLimits::with_fee(fee_details);
        let fee_payer_address = message.fee_payer();
        let fee_payer_account = AccountSharedData::new(fee_payer_balance, 0, &Pubkey::default());
        let mut mock_accounts = HashMap::new();
        mock_accounts.insert(*fee_payer_address, fee_payer_account);
        let mut mock_bank = MockBankCallback {
            account_shared_data: Arc::new(RwLock::new(mock_accounts)),
            ..Default::default()
        };
        mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
        let mut account_loader = (&mock_bank).into();

        let mut error_counters = TransactionErrorMetrics::default();

        let expected_error = TransactionError::InsufficientFundsForFee;
        let expected_result = if relax_fee_payer_constraint {
            TransactionValidationResult::NoOp(NoOpTransaction {
                validation_error: expected_error,
                fee_payer_balance: Some(fee_payer_balance),
                compute_unit_limit: fee_and_limits.budget.compute_unit_limit,
                loaded_accounts_bytes_limit: fee_and_limits.loaded_accounts_data_size_limit,
                nonce_address: None,
            })
        } else {
            TransactionValidationResult::Unprocessable(expected_error)
        };

        let actual_result =
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(None, fee_and_limits),
                &Hash::default(),
                lamports_per_signature,
                &Rent::default(),
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

        assert_eq!(error_counters.insufficient_funds.0, 1);
        assert_eq!(actual_result, expected_result);
    }

    #[test_case(false; "strict_fee_payer")]
    #[test_case(true; "relaxed_fee_payer")]
    fn test_validate_transaction_fee_payer_insufficient_rent(relax_fee_payer_constraint: bool) {
        let lamports_per_signature = 5000;
        let message =
            new_unchecked_sanitized_message(Message::new(&[], Some(&Pubkey::new_unique())));
        let fee_details =
            MockBankCallback::calculate_fee_details(&message, lamports_per_signature, 0);
        let fee_and_limits = SVMTransactionExecutionAndFeeBudgetLimits::with_fee(fee_details);
        let fee_payer_address = message.fee_payer();
        let transaction_fee = lamports_per_signature;
        let rent = Rent::default();
        let min_balance = rent.minimum_balance(0);
        let starting_balance = min_balance + transaction_fee - 1;
        let fee_payer_account = AccountSharedData::new(starting_balance, 0, &Pubkey::default());
        let mut mock_accounts = HashMap::new();
        mock_accounts.insert(*fee_payer_address, fee_payer_account);
        let mut mock_bank = MockBankCallback {
            account_shared_data: Arc::new(RwLock::new(mock_accounts)),
            ..Default::default()
        };
        mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
        let mut account_loader = (&mock_bank).into();

        let mut error_counters = TransactionErrorMetrics::default();

        let expected_error = TransactionError::InsufficientFundsForRent { account_index: 0 };
        let expected_result = if relax_fee_payer_constraint {
            TransactionValidationResult::NoOp(NoOpTransaction {
                validation_error: expected_error,
                fee_payer_balance: Some(starting_balance),
                compute_unit_limit: fee_and_limits.budget.compute_unit_limit,
                loaded_accounts_bytes_limit: fee_and_limits.loaded_accounts_data_size_limit,
                nonce_address: None,
            })
        } else {
            TransactionValidationResult::Unprocessable(expected_error)
        };

        let actual_result =
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(None, fee_and_limits),
                &Hash::default(),
                lamports_per_signature,
                &rent,
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

        assert_eq!(actual_result, expected_result);
    }

    #[test_case(false; "strict_fee_payer")]
    #[test_case(true; "relaxed_fee_payer")]
    fn test_validate_transaction_fee_payer_invalid(relax_fee_payer_constraint: bool) {
        let lamports_per_signature = 5000;
        let fee_payer_balance = 1_000_000;
        let message =
            new_unchecked_sanitized_message(Message::new(&[], Some(&Pubkey::new_unique())));
        let fee_details =
            MockBankCallback::calculate_fee_details(&message, lamports_per_signature, 0);
        let fee_and_limits = SVMTransactionExecutionAndFeeBudgetLimits::with_fee(fee_details);
        let fee_payer_address = message.fee_payer();
        let fee_payer_account = AccountSharedData::new(fee_payer_balance, 0, &Pubkey::new_unique());
        let mut mock_accounts = HashMap::new();
        mock_accounts.insert(*fee_payer_address, fee_payer_account);
        let mut mock_bank = MockBankCallback {
            account_shared_data: Arc::new(RwLock::new(mock_accounts)),
            ..Default::default()
        };
        mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
        let mut account_loader = (&mock_bank).into();

        let mut error_counters = TransactionErrorMetrics::default();

        let expected_error = TransactionError::InvalidAccountForFee;
        let expected_result = if relax_fee_payer_constraint {
            TransactionValidationResult::NoOp(NoOpTransaction {
                validation_error: expected_error,
                fee_payer_balance: Some(fee_payer_balance),
                compute_unit_limit: fee_and_limits.budget.compute_unit_limit,
                loaded_accounts_bytes_limit: fee_and_limits.loaded_accounts_data_size_limit,
                nonce_address: None,
            })
        } else {
            TransactionValidationResult::Unprocessable(expected_error)
        };

        let actual_result =
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(None, fee_and_limits),
                &Hash::default(),
                lamports_per_signature,
                &Rent::default(),
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

        assert_eq!(error_counters.invalid_account_for_fee.0, 1);
        assert_eq!(actual_result, expected_result);
    }

    #[derive(Debug, PartialEq, Eq)]
    enum ValidateNonce {
        Success,
        NoAccount,
        BadOwner,
        BlockhashMismatch,
        AlreadyUsed,
        BadSigner,
        NotWritable,
    }

    #[test_case(ValidateNonce::Success)]
    #[test_case(ValidateNonce::NoAccount)]
    #[test_case(ValidateNonce::BadOwner)]
    #[test_case(ValidateNonce::BlockhashMismatch)]
    #[test_case(ValidateNonce::AlreadyUsed)]
    #[test_case(ValidateNonce::BadSigner)]
    #[test_case(ValidateNonce::NotWritable)]
    fn test_validate_transaction_nonce(case: ValidateNonce) {
        let lamports_per_signature = 5000;
        let previous_durable_nonce = DurableNonce::from_blockhash(&Hash::new_unique());
        let nonce_address = Pubkey::new_unique();
        let authority_address = Pubkey::new_unique();

        let message_blockhash = if case == ValidateNonce::BlockhashMismatch {
            Hash::new_unique()
        } else {
            *previous_durable_nonce.as_hash()
        };

        let message_authority = if case == ValidateNonce::BadSigner {
            Pubkey::new_unique()
        } else {
            authority_address
        };

        let mut advance_nonce_instruction =
            system_instruction::advance_nonce_account(&nonce_address, &message_authority);
        if case == ValidateNonce::NotWritable {
            advance_nonce_instruction.accounts[0].is_writable = false;
        }

        let message = new_unchecked_sanitized_message(Message::new_with_blockhash(
            &[advance_nonce_instruction],
            Some(&Pubkey::new_unique()),
            &message_blockhash,
        ));

        let environment_blockhash = Hash::new_unique();
        let next_durable_nonce = DurableNonce::from_blockhash(&environment_blockhash);

        let stored_durable_nonce = if case == ValidateNonce::AlreadyUsed {
            next_durable_nonce
        } else {
            previous_durable_nonce
        };

        let nonce_versions = nonce::versions::Versions::new(nonce::state::State::Initialized(
            nonce::state::Data::new(
                authority_address,
                stored_durable_nonce,
                lamports_per_signature,
            ),
        ));

        let nonce_owner = if case == ValidateNonce::BadOwner {
            Pubkey::new_unique()
        } else {
            system_program::id()
        };

        let nonce_account = AccountSharedData::new_data(1, &nonce_versions, &nonce_owner).unwrap();

        let mut mock_accounts = HashMap::new();

        if case != ValidateNonce::NoAccount {
            mock_accounts.insert(nonce_address, nonce_account.clone());
        }

        let mock_bank = MockBankCallback {
            account_shared_data: Arc::new(RwLock::new(mock_accounts)),
            ..Default::default()
        };
        let mut account_loader = (&mock_bank).into();

        let mut error_counters = TransactionErrorMetrics::default();
        let result = TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce(
            &mut account_loader,
            &message,
            &nonce_address,
            &next_durable_nonce,
            lamports_per_signature,
            false,
            &mut error_counters,
        );

        match case {
            ValidateNonce::Success => {
                let mut future_nonce_info = NonceInfo::new(nonce_address, nonce_account);
                future_nonce_info
                    .try_advance_nonce(next_durable_nonce, lamports_per_signature)
                    .unwrap();

                assert_eq!(result, Ok(future_nonce_info));
            }
            ValidateNonce::NoAccount => {
                assert_eq!(error_counters.account_not_found.0, 1);
                assert_eq!(result, Err(TransactionError::AccountNotFound));
            }
            _ => {
                assert_eq!(error_counters.blockhash_not_found.0, 1);
                assert_eq!(result, Err(TransactionError::BlockhashNotFound));
            }
        }
    }

    #[test_case(false; "strict_fee_payer")]
    #[test_case(true; "relaxed_fee_payer")]
    fn test_validate_transaction_fee_payer_is_nonce(relax_fee_payer_constraint: bool) {
        let lamports_per_signature = 5000;
        let rent = Rent::default();
        let compute_unit_limit = 1000u64;
        let previous_durable_nonce = DurableNonce::from_blockhash(&Hash::new_unique());
        let fee_payer_address = &Pubkey::new_unique();
        let message = new_unchecked_sanitized_message(Message::new_with_blockhash(
            &[
                system_instruction::advance_nonce_account(fee_payer_address, fee_payer_address),
                ComputeBudgetInstruction::set_compute_unit_limit(compute_unit_limit as u32),
                ComputeBudgetInstruction::set_compute_unit_price(1_000_000),
            ],
            Some(fee_payer_address),
            previous_durable_nonce.as_hash(),
        ));
        let transaction_fee = lamports_per_signature;
        let compute_budget_and_limits = SVMTransactionExecutionAndFeeBudgetLimits {
            fee_details: FeeDetails::new(transaction_fee, compute_unit_limit),
            ..SVMTransactionExecutionAndFeeBudgetLimits::default()
        };
        let min_balance = Rent::default().minimum_balance(nonce::state::State::size());
        let priority_fee = compute_unit_limit;

        let nonce_versions = nonce::versions::Versions::new(nonce::state::State::Initialized(
            nonce::state::Data::new(
                *fee_payer_address,
                previous_durable_nonce,
                lamports_per_signature,
            ),
        ));

        let environment_blockhash = Hash::new_unique();
        let next_durable_nonce = DurableNonce::from_blockhash(&environment_blockhash);

        // Sufficient Fees
        {
            let fee_payer_account = AccountSharedData::new_data(
                min_balance + transaction_fee + priority_fee,
                &nonce_versions,
                &system_program::id(),
            )
            .unwrap();

            let mut future_nonce = NonceInfo::new(*fee_payer_address, fee_payer_account.clone());
            future_nonce
                .try_advance_nonce(next_durable_nonce, lamports_per_signature)
                .unwrap();

            let mut mock_accounts = HashMap::new();
            mock_accounts.insert(*fee_payer_address, fee_payer_account.clone());
            let mut mock_bank = MockBankCallback {
                account_shared_data: Arc::new(RwLock::new(mock_accounts)),
                ..Default::default()
            };
            mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
            let mut account_loader = (&mock_bank).into();

            let mut error_counters = TransactionErrorMetrics::default();

            let tx_details =
                CheckedTransactionDetails::new(Some(*fee_payer_address), compute_budget_and_limits);

            let result = TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                tx_details,
                &environment_blockhash,
                lamports_per_signature,
                &rent,
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

            let post_validation_fee_payer_account = {
                let mut account = fee_payer_account.clone();
                account.set_rent_epoch(RENT_EXEMPT_RENT_EPOCH);
                account.set_lamports(min_balance);
                account
            };

            assert_eq!(
                result,
                TransactionValidationResult::Loadable(ValidatedTransactionDetails {
                    rollback_accounts: RollbackAccounts::new(
                        Some(future_nonce),
                        *fee_payer_address,
                        post_validation_fee_payer_account.clone(),
                        0, // fee_payer_rent_epoch
                    ),
                    compute_budget: compute_budget_and_limits.budget,
                    loaded_accounts_bytes_limit: compute_budget_and_limits
                        .loaded_accounts_data_size_limit,
                    fee_details: FeeDetails::new(transaction_fee, priority_fee),
                    loaded_fee_payer_account: LoadedTransactionAccount {
                        loaded_size: TRANSACTION_ACCOUNT_BASE_SIZE + fee_payer_account.data().len(),
                        account: post_validation_fee_payer_account,
                    }
                })
            );
        }

        // Insufficient Fees
        {
            let fee_payer_account = AccountSharedData::new_data(
                transaction_fee + priority_fee, // no min_balance this time
                &nonce_versions,
                &system_program::id(),
            )
            .unwrap();

            let mut mock_accounts = HashMap::new();
            mock_accounts.insert(*fee_payer_address, fee_payer_account);
            let mut mock_bank = MockBankCallback {
                account_shared_data: Arc::new(RwLock::new(mock_accounts)),
                ..Default::default()
            };
            mock_bank.feature_set.relax_fee_payer_constraint = relax_fee_payer_constraint;
            let mut account_loader = (&mock_bank).into();

            let mut error_counters = TransactionErrorMetrics::default();

            let tx_details =
                CheckedTransactionDetails::new(Some(*fee_payer_address), compute_budget_and_limits);

            let result = TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                tx_details,
                &environment_blockhash,
                lamports_per_signature,
                &rent,
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut error_counters,
            );

            // nonce transactions are never processable under SIMD-0290 rules
            assert_eq!(error_counters.insufficient_funds.0, 1);
            assert_eq!(
                result,
                TransactionValidationResult::Unprocessable(
                    TransactionError::InsufficientFundsForFee
                )
            );
        }
    }

    // Ensure `TransactionProcessingCallback::inspect_account()` is called when
    // validating the fee payer, since that's when the fee payer account is loaded.
    #[test]
    fn test_inspect_account_fee_payer() {
        let lamports_per_signature = 5000;
        let fee_payer_address = Pubkey::new_unique();
        let fee_payer_account = AccountSharedData::new_rent_epoch(
            123_000_000_000,
            0,
            &Pubkey::default(),
            RENT_EXEMPT_RENT_EPOCH,
        );
        let mock_bank = MockBankCallback::default();
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(fee_payer_address, fee_payer_account.clone());
        let mut account_loader = (&mock_bank).into();

        let message = new_unchecked_sanitized_message(Message::new_with_blockhash(
            &[
                ComputeBudgetInstruction::set_compute_unit_limit(2000u32),
                ComputeBudgetInstruction::set_compute_unit_price(1_000_000_000),
            ],
            Some(&fee_payer_address),
            &Hash::new_unique(),
        ));

        assert!(matches!(
            TransactionBatchProcessor::<TestForkGraph>::validate_transaction_nonce_and_fee_payer(
                &mut account_loader,
                &message,
                CheckedTransactionDetails::new(
                    None,
                    SVMTransactionExecutionAndFeeBudgetLimits::with_fee(
                        MockBankCallback::calculate_fee_details(&message, 5000, 0),
                    ),
                ),
                &Hash::default(),
                lamports_per_signature,
                &Rent::default(),
                mock_bank.feature_set.relax_post_exec_min_balance_check,
                false,
                &mut TransactionErrorMetrics::default(),
            ),
            TransactionValidationResult::Loadable(_)
        ));

        // ensure the fee payer is an inspected account
        let actual_inspected_accounts: Vec<_> = mock_bank
            .inspected_accounts
            .read()
            .unwrap()
            .iter()
            .map(|(k, v)| (*k, v.clone()))
            .collect();
        assert_eq!(
            actual_inspected_accounts.as_slice(),
            &[(fee_payer_address, vec![(Some(fee_payer_account), true)])],
        );
    }

    #[test]
    fn test_set_program_runtime_environment() {
        let mut transaction_processor = TransactionBatchProcessor::<TestForkGraph>::default();
        let current_environment =
            ProgramRuntimeEnvironment::clone(&transaction_processor.program_runtime_environment);
        let new_environment = ProgramRuntimeEnvironment::from(BuiltinProgram::new_mock());
        let config = vm::Config {
            enable_symbol_and_section_labels: true,
            ..vm::Config::default()
        };
        let new_environment2 = ProgramRuntimeEnvironment::from(BuiltinProgram::new_loader(config));
        assert_ne!(current_environment, new_environment);
        assert_ne!(current_environment, new_environment2);
        assert_ne!(new_environment, new_environment2);
        // Assign an equal and identical environment: No changes
        transaction_processor.set_program_runtime_environment(ProgramRuntimeEnvironment::clone(
            &current_environment,
        ));
        assert_eq!(
            transaction_processor.program_runtime_environment,
            current_environment,
        );
        // Assign an equal but not identical environment: No changes
        transaction_processor
            .set_program_runtime_environment(ProgramRuntimeEnvironment::clone(&new_environment));
        assert_eq!(
            transaction_processor.program_runtime_environment,
            current_environment,
        );
        // Assign a different and not identical environment: Overwritten
        transaction_processor
            .set_program_runtime_environment(ProgramRuntimeEnvironment::clone(&new_environment2));
        assert_eq!(
            transaction_processor.program_runtime_environment,
            new_environment2,
        );
        // Assign an environment which is equal to the upcoming_environment: Overwritten
        transaction_processor
            .epoch_boundary_preparation
            .write()
            .unwrap()
            .upcoming_environment = Some(ProgramRuntimeEnvironment::clone(&new_environment));
        transaction_processor.set_program_runtime_environment(ProgramRuntimeEnvironment::clone(
            &current_environment,
        ));
        assert_eq!(
            transaction_processor.program_runtime_environment,
            new_environment,
        );
    }

    fn catch_panic(f: impl FnOnce()) -> Option<String> {
        let previous_hook = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {}));
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
        std::panic::set_hook(previous_hook);
        result.err().map(|payload| {
            payload
                .downcast_ref::<&str>()
                .map(|s| s.to_string())
                .or_else(|| payload.downcast_ref::<String>().cloned())
                .unwrap_or_else(|| "<non-string panic payload>".to_string())
        })
    }

    #[test_case(ProgramCacheEntryOwner::LoaderV1)]
    #[test_case(ProgramCacheEntryOwner::LoaderV2)]
    #[test_case(ProgramCacheEntryOwner::LoaderV3)]
    #[test_case(ProgramCacheEntryOwner::LoaderV4)]
    fn test_replenish_program_cache_success(loader: ProgramCacheEntryOwner) {
        const BATCH_SLOT: u64 = 200;
        const DEPLOYMENT_SLOT: u64 = 10;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);

        // Set up a valid program, with a valid ELF, for each loader. Note V1
        // and V2 do not retain a deployment slot, so they report zero.
        let expected_deployment_slot = match loader {
            ProgramCacheEntryOwner::LoaderV1 | ProgramCacheEntryOwner::LoaderV2 => {
                let mut account = AccountSharedData::default();
                account.set_owner(Pubkey::from(loader));
                account.set_data_from_slice(&load_test_program());
                mock_bank
                    .account_shared_data
                    .write()
                    .unwrap()
                    .insert(program_id, account);
                0
            }
            ProgramCacheEntryOwner::LoaderV3 => {
                let programdata_id = Pubkey::new_unique();
                mock_bank
                    .account_shared_data
                    .write()
                    .unwrap()
                    .insert(program_id, loader_v3_program_account(programdata_id));
                mock_bank.account_shared_data.write().unwrap().insert(
                    programdata_id,
                    loader_v3_programdata_account(DEPLOYMENT_SLOT, &load_test_program()),
                );
                DEPLOYMENT_SLOT
            }
            ProgramCacheEntryOwner::LoaderV4 => {
                mock_bank.account_shared_data.write().unwrap().insert(
                    program_id,
                    loader_v4_account(
                        DEPLOYMENT_SLOT,
                        LoaderV4Status::Deployed,
                        &load_test_program(),
                    ),
                );
                DEPLOYMENT_SLOT
            }
            ProgramCacheEntryOwner::NativeLoader => unreachable!(),
        };

        // The program should be identified for extraction.
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert_eq!(
            missing_programs,
            &[ProgramToLoad {
                program_id: &program_id,
                loader,
                deployment_slot: expected_deployment_slot,
            }]
        );

        // The load succeeds, leaving a `Loaded` entry in the batch cache.
        batch_processor.replenish_program_cache(
            &account_loader,
            missing_programs,
            &environment,
            &mut program_cache_for_tx_batch,
            &mut ExecuteTimings::default(),
            true,
            true,
        );
        let entry = program_cache_for_tx_batch.find(&program_id).unwrap();
        assert!(matches!(entry.program, ProgramCacheEntryType::Loaded(_)));
        assert_eq!(entry.deployment_slot, expected_deployment_slot);
        assert_eq!(entry.account_owner, loader);
    }

    #[test_case(ProgramCacheEntryOwner::LoaderV1)]
    #[test_case(ProgramCacheEntryOwner::LoaderV2)]
    #[test_case(ProgramCacheEntryOwner::LoaderV3)]
    #[test_case(ProgramCacheEntryOwner::LoaderV4)]
    #[test_case(ProgramCacheEntryOwner::NativeLoader)]
    fn test_replenish_program_cache_program_account_not_found(loader: ProgramCacheEntryOwner) {
        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(0, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);

        // The program should not be identified as a program we must extract.
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert!(missing_programs.is_empty());

        // But still, even if we did happen to try to extract this program, it
        // should panic.
        let panicked = catch_panic(|| {
            batch_processor.replenish_program_cache(
                &account_loader,
                vec![ProgramToLoad {
                    program_id: &program_id,
                    loader,
                    deployment_slot: 0,
                }],
                &environment,
                &mut program_cache_for_tx_batch,
                &mut ExecuteTimings::default(),
                true,
                true,
            )
        });
        assert_eq!(
            panicked.as_deref(),
            Some("called load_program_with_pubkey() with nonexistent account")
        );
    }

    #[test_case(ProgramCacheEntryOwner::LoaderV1)]
    #[test_case(ProgramCacheEntryOwner::LoaderV2)]
    #[test_case(ProgramCacheEntryOwner::LoaderV3)]
    #[test_case(ProgramCacheEntryOwner::LoaderV4)]
    #[test_case(ProgramCacheEntryOwner::NativeLoader)]
    fn test_replenish_program_cache_program_account_empty(loader: ProgramCacheEntryOwner) {
        const BATCH_SLOT: u64 = 200;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);

        // Insert a program account owned by the proper loader, but with no
        // state in it at all.
        let mut program_account = AccountSharedData::default();
        program_account.set_owner(Pubkey::from(loader));
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(program_id, program_account);

        // The program should not be identified as a program we must extract.
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert!(missing_programs.is_empty());

        // Try to extract it anyway.
        let panicked = catch_panic(|| {
            batch_processor.replenish_program_cache(
                &account_loader,
                vec![ProgramToLoad {
                    program_id: &program_id,
                    loader,
                    deployment_slot: 0,
                }],
                &environment,
                &mut program_cache_for_tx_batch,
                &mut ExecuteTimings::default(),
                true,
                true,
            )
        });

        match loader {
            ProgramCacheEntryOwner::LoaderV1 | ProgramCacheEntryOwner::LoaderV2 => {
                // For Loader V1 & V2, the entire account contents is taken
                // as-is and passed to `load`. Empty bytes == invalid ELF.
                assert_eq!(
                    panicked.as_deref(),
                    None, // <-- no panic
                );
                let entry = program_cache_for_tx_batch.find(&program_id).unwrap();
                assert_eq!(
                    entry,
                    Arc::new(ProgramCacheEntry::new_failed_verification_tombstone(
                        0, // <-- V1/V2 programs are always deployment slot 0
                        loader,
                        environment,
                    ))
                );
            }
            ProgramCacheEntryOwner::LoaderV3 | ProgramCacheEntryOwner::LoaderV4 => {
                // For Loader V3 & V4, data is too small for the respective
                // metadata section, so the program is considered *closed*.
                //
                // We observe a panic here from `assign_program`, as extraction
                // enters into a **continuous loop** of inserting a `Closed`
                // tombstone for the batch slot and missing it on extraction,
                // which targets the *deployment slot*.
                //
                // More information on this loop condition is provided below in
                // `test_replenish_program_cache_loader_v3_closed`.
                assert_eq!(
                    panicked.as_deref(),
                    Some("Unexpected replacement of an entry")
                );
            }
            ProgramCacheEntryOwner::NativeLoader => {
                // Native loader-owned accounts panic, since the call to
                // `load_program_with_pubkey` just returns an error which gets
                // unwrapped via `expect` in `replenish_program_cache`.
                assert_eq!(
                    panicked.as_deref(),
                    Some("called load_program_with_pubkey() with nonexistent account")
                );
            }
        }
    }

    enum ProgramDataCase {
        DoesNotExist,
        OwnedButEmpty,
        Uninitialized,
    }

    #[test_case(ProgramDataCase::DoesNotExist)]
    #[test_case(ProgramDataCase::OwnedButEmpty)]
    #[test_case(ProgramDataCase::Uninitialized)]
    fn test_replenish_program_cache_loader_v3_closed(case: ProgramDataCase) {
        const BATCH_SLOT: u64 = 200;
        const DEPLOYMENT_SLOT: u64 = 10;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();
        let programdata_id = Pubkey::new_unique();

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);

        // Insert a legit program account.
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(program_id, loader_v3_program_account(programdata_id));

        let programdata_account = match case {
            ProgramDataCase::DoesNotExist => {
                // Case: Programdata does not exist
                AccountSharedData::default()
            }
            ProgramDataCase::OwnedButEmpty => {
                // Case: Programdata is owned by Loader V3, but no state.
                let mut account = AccountSharedData::default();
                account.set_owner(bpf_loader_upgradeable::id());
                account
            }
            ProgramDataCase::Uninitialized => {
                // Case: Programdata is `Uninitialized`.
                let mut account = AccountSharedData::default();
                account.set_owner(bpf_loader_upgradeable::id());
                account.set_data_from_slice(&[0u8; 4]);
                account
            }
        };
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(programdata_id, programdata_account);

        // The program should not be identified as a program we must extract.
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert!(missing_programs.is_empty());

        // `filter_executable_program_accounts` is the guard. As long as we
        // don't place a closed/non-existant program in the search list for
        // extraction, we correctly treat it as a `Closed` program in the batch
        // cache.
        //
        // If this ever goes wrong, and a closed/non-existant program appears
        // in the search list for extraction, we can end up with a **continuous
        // loop**.
        //
        // The loop works like this:
        //
        // - Closed program is requested for extraction. Global cache doesn't
        //   have it, so it signals for reload.
        // - Reloading via `load_program_with_pubkey` returns a `Closed`
        //   tombstone. This tombstone is inserted into the global program
        //   cache via `finish_cooperative_loading_task`, which calls into
        //   `assign_program`.
        // - After this step, extraction is *supposed* to run again, so the
        //   program key is still present in the search list. However, each
        //   subsequent call to `extract` is going to MISS. This is due to the
        //   fact that extraction MUST match on the deployment slot, and a
        //   `Closed` tombstone holds the **batch slot**.
        //
        // The loop can only happen in production. In debug mode, we trip on
        // the `debug_assert!` inside of `assign_program`, which traps on an
        // invalid `Closed`-`Closed` transition. We observe this below.
        let panicked = catch_panic(|| {
            batch_processor.replenish_program_cache(
                &account_loader,
                vec![ProgramToLoad {
                    program_id: &program_id,
                    loader: ProgramCacheEntryOwner::LoaderV3,
                    deployment_slot: DEPLOYMENT_SLOT,
                }],
                &environment,
                &mut program_cache_for_tx_batch,
                &mut ExecuteTimings::default(),
                true,
                true,
            )
        });
        assert_eq!(
            panicked.as_deref(),
            Some("Unexpected replacement of an entry")
        );
    }

    #[test_case(true)]
    #[test_case(false)]
    fn test_replenish_program_cache_loader_v4_retracted(just_zeroes: bool) {
        const BATCH_SLOT: u64 = 200;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);

        let program_account = if just_zeroes {
            // Case: "gifted" state
            // Sized correctly, all-zeroes. Since `LoaderV4Status::Retracted`
            // holds variant `0`, we check this case, too.
            let mut account = AccountSharedData::default();
            account.set_owner(loader_v4::id());
            account.set_data_from_slice(&[0u8; LoaderV4State::program_data_offset()]);
            account
        } else {
            // Case: status is Retracted
            loader_v4_account(9, LoaderV4Status::Retracted, &load_test_program())
        };
        mock_bank
            .account_shared_data
            .write()
            .unwrap()
            .insert(program_id, program_account);

        // The program should not be identified as a program we must extract.
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert!(missing_programs.is_empty());

        // As in `test_replenish_program_cache_loader_v3_closed`,
        // `filter_executable_program_accounts` is the guard. Forcing the
        // program into the search list anyway walks into the continuous loop
        // described there, which debug mode traps on.
        let panicked = catch_panic(|| {
            batch_processor.replenish_program_cache(
                &account_loader,
                vec![ProgramToLoad {
                    program_id: &program_id,
                    loader: ProgramCacheEntryOwner::LoaderV4,
                    deployment_slot: if just_zeroes { 0 } else { 9 },
                }],
                &environment,
                &mut program_cache_for_tx_batch,
                &mut ExecuteTimings::default(),
                true,
                true,
            )
        });
        assert_eq!(
            panicked.as_deref(),
            Some("Unexpected replacement of an entry")
        );
    }

    fn store_loader_v3_program(
        mock_bank: &MockBankCallback,
        program_id: &Pubkey,
        deployment_slot: Slot,
        elf: &[u8],
    ) {
        let programdata_id = Pubkey::new_unique();
        let mut accounts = mock_bank.account_shared_data.write().unwrap();
        accounts.insert(*program_id, loader_v3_program_account(programdata_id));
        accounts.insert(
            programdata_id,
            loader_v3_programdata_account(deployment_slot, elf),
        );
    }

    #[test_case(false)]
    #[test_case(true)]
    fn test_replenish_program_cache_delay_visibility(already_cached: bool) {
        // Tests delay visibility behavior in `replenish_program_cache`.
        //
        // Deployments currently insert an unloaded entry into the cache, so we
        // should always see an existing entry. If, for some reason, the
        // program is not already cached as unloaded, we end up doing a
        // cooperative load one slot too early.
        const BATCH_SLOT: u64 = 200;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        // Deployed in the batch's own slot, so it is not effective yet.
        store_loader_v3_program(&mock_bank, &program_id, BATCH_SLOT, &load_test_program());

        if already_cached {
            batch_processor
                .global_program_cache
                .write()
                .unwrap()
                .assign_program(
                    &environment,
                    program_id,
                    BATCH_SLOT,
                    Arc::new(ProgramCacheEntry::new_unloaded(
                        BATCH_SLOT,
                        ProgramCacheEntryOwner::LoaderV3,
                        environment.clone(),
                    )),
                );
            batch_processor
                .global_program_cache
                .write()
                .unwrap()
                .stats
                .reset();
        }

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);
        let keys = [program_id];
        batch_processor.replenish_program_cache(
            &account_loader,
            filter_executable_program_accounts(
                &mock_bank,
                &program_cache_for_tx_batch,
                keys.iter(),
            ),
            &environment,
            &mut program_cache_for_tx_batch,
            &mut ExecuteTimings::default(),
            true,
            true,
        );

        // Either way the batch is handed a tombstone.
        // Here we read the entry directly.
        let stored = program_cache_for_tx_batch
            .get_entry_for_tests(&program_id)
            .unwrap();
        assert!(matches!(
            stored.program,
            ProgramCacheEntryType::DelayVisibility
        ));
        assert_eq!(stored.deployment_slot, BATCH_SLOT);

        // And here we query with `::find`, and see the same thing.
        let entry = program_cache_for_tx_batch.find(&program_id).unwrap();
        assert!(matches!(
            entry.program,
            ProgramCacheEntryType::DelayVisibility
        ));
        assert_eq!(entry.deployment_slot, BATCH_SLOT);

        // The global cache should just have one entry for this program.
        // What that entry looks like will differ, though.
        let global_program_cache = batch_processor.global_program_cache.read().unwrap();
        let slot_versions = global_program_cache.get_slot_versions_for_tests(&program_id);
        assert_eq!(slot_versions.len(), 1);

        if already_cached {
            // If the program was already cached as unloaded, we did not waste
            // a load here, and it remains unloaded until effective.
            assert!(matches!(
                slot_versions[0].program,
                ProgramCacheEntryType::Unloaded(_)
            ));
            assert!(!program_cache_for_tx_batch.loaded_missing);
            assert_eq!(global_program_cache.stats.hits.load(Ordering::Relaxed), 1);
            assert_eq!(global_program_cache.stats.misses.load(Ordering::Relaxed), 0);
        } else {
            // If the program was not cached, we actually loaded it, even
            // though we couldn't use it.
            //
            // It's fine, since the batch cache still returns the tombstone.
            assert!(matches!(
                slot_versions[0].program,
                ProgramCacheEntryType::Loaded(_)
            ));
            assert!(program_cache_for_tx_batch.loaded_missing);
            assert_eq!(global_program_cache.stats.hits.load(Ordering::Relaxed), 0);
            assert_eq!(global_program_cache.stats.misses.load(Ordering::Relaxed), 1);
        }
    }

    #[test]
    fn test_replenish_program_cache_failed_verification_is_cached() {
        const BATCH_SLOT: u64 = 200;
        const DEPLOYMENT_SLOT: u64 = 10;

        let mock_bank = MockBankCallback::default();
        let account_loader = (&mock_bank).into();
        let fork_graph = Arc::new(RwLock::new(TestForkGraph {}));
        let batch_processor =
            TransactionBatchProcessor::new(BATCH_SLOT, 0, Arc::downgrade(&fork_graph), None);
        let environment = batch_processor.program_runtime_environment_for_epoch(0);
        let program_id = Pubkey::new_unique();

        // Valid programdata, bad ELF.
        store_loader_v3_program(&mock_bank, &program_id, DEPLOYMENT_SLOT, &[1u8; 64]);

        let mut program_cache_for_tx_batch = ProgramCacheForTxBatch::new(batch_processor.slot);
        let keys = [program_id];
        let missing_programs = filter_executable_program_accounts(
            &mock_bank,
            &program_cache_for_tx_batch,
            keys.iter(),
        );
        assert_eq!(missing_programs.len(), 1);

        batch_processor.replenish_program_cache(
            &account_loader,
            missing_programs,
            &environment,
            &mut program_cache_for_tx_batch,
            &mut ExecuteTimings::default(),
            true,
            true,
        );

        // Unlike a closed program, a program which fails verification is
        // tombstoned at its own deployment slot, so the extraction which
        // follows the load finds it.
        let entry = program_cache_for_tx_batch.find(&program_id).unwrap();
        assert!(matches!(
            entry.program,
            ProgramCacheEntryType::FailedVerification(_)
        ));
        assert_eq!(entry.deployment_slot, DEPLOYMENT_SLOT);

        // It is cached globally too, so the next batch does not recompile it.
        let global_program_cache = batch_processor.global_program_cache.read().unwrap();
        let slot_versions = global_program_cache.get_slot_versions_for_tests(&program_id);
        assert_eq!(slot_versions.len(), 1);
        assert!(matches!(
            slot_versions[0].program,
            ProgramCacheEntryType::FailedVerification(_)
        ));
        assert!(matches!(slot_versions[0].deployment_slot, DEPLOYMENT_SLOT));
    }
}