iota_core/checkpoints/checkpoint_executor/

mod.rs

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

//! CheckpointExecutor is a Node component that executes all checkpoints for the
//! given epoch. It acts as a Consumer to StateSync
//! for newly synced checkpoints, taking these checkpoints and
//! scheduling and monitoring their execution. Its primary goal is to allow
//! for catching up to the current checkpoint sequence number of the network
//! as quickly as possible so that a newly joined, or recovering Node can
//! participate in a timely manner. To that end, CheckpointExecutor attempts
//! to saturate the CPU with executor tasks (one per checkpoint), each of which
//! handle scheduling and awaiting checkpoint transaction execution.
//!
//! CheckpointExecutor is made recoverable in the event of Node shutdown by way
//! of a watermark, highest_executed_checkpoint, which is guaranteed to be
//! updated sequentially in order, despite checkpoints themselves potentially
//! being executed nonsequentially and in parallel. CheckpointExecutor
//! parallelizes checkpoints of the same epoch as much as possible.
//! CheckpointExecutor enforces the invariant that if `run` returns
//! successfully, we have reached the end of epoch. This allows us to use it as
//! a signal for reconfig.

use std::{future::Future, sync::Arc, time::Instant};

use futures::StreamExt;
use iota_common::{debug_fatal, fatal};
use iota_config::node::{CheckpointExecutorConfig, RunWithRange};
use iota_macros::fail_point;
use iota_sdk_types::{
    CheckpointContents, RandomnessRound, TransactionDigest, TransactionEffects,
    TransactionEffectsDigest, TransactionKind,
};
use iota_types::{
    base_types::ExecutionData,
    effects::TransactionEffectsAPI,
    executable_transaction::VerifiedExecutableTransaction,
    full_checkpoint_content::CheckpointData,
    global_state_hash::GlobalStateHash,
    messages_checkpoint::{
        CheckpointContentsExt, CheckpointSequenceNumber, CheckpointSummaryExt,
        FullCheckpointContents, VerifiedCheckpoint,
    },
    transaction::{SenderSignedTransactionAPI, TransactionAPI, VerifiedTransaction},
};
use parking_lot::Mutex;
use tap::{TapFallible, TapOptional};
use tracing::{debug, info, instrument};

use crate::{
    authority::{
        AuthorityState, ExecutionEnv, authority_per_epoch_store::AuthorityPerEpochStore,
        backpressure::BackpressureManager,
        shared_object_version_manager::gas_object_cancellation_version_from_effects,
    },
    checkpoint_progress_tracker::CheckpointProgressTracker,
    checkpoints::CheckpointStore,
    execution_cache::{ObjectCacheRead, TransactionCacheRead},
    execution_scheduler::{ExecutionSchedulerAPI, ExecutionSchedulerWrapper},
    global_state_hasher::GlobalStateHasher,
};

mod data_ingestion_handler;
pub mod metrics;
pub(crate) mod utils;

#[cfg(test)]
pub(crate) mod tests;

use data_ingestion_handler::{load_checkpoint_data, store_checkpoint_locally};
use metrics::CheckpointExecutorMetrics;
use utils::*;

type CheckpointDataSender = Box<dyn Fn(&CheckpointData) + Send + Sync>;

#[derive(PartialEq, Eq, Debug)]
pub enum StopReason {
    EpochComplete,
    RunWithRangeCondition,
}

pub(crate) struct CheckpointExecutionData {
    pub checkpoint: VerifiedCheckpoint,
    pub checkpoint_contents: CheckpointContents,
    pub tx_digests: Vec<TransactionDigest>,
    pub fx_digests: Vec<TransactionEffectsDigest>,
}

pub(crate) struct CheckpointTransactionData {
    pub transactions: Vec<VerifiedExecutableTransaction>,
    pub effects: Vec<TransactionEffects>,
}

pub(crate) struct CheckpointExecutionState {
    pub data: CheckpointExecutionData,
    state_hash: Option<GlobalStateHash>,
    full_data: Option<CheckpointData>,
}

impl CheckpointExecutionState {
    pub fn new(data: CheckpointExecutionData) -> Self {
        Self {
            data,
            state_hash: None,
            full_data: None,
        }
    }

    pub fn new_with_global_state_hash(
        data: CheckpointExecutionData,
        hash: GlobalStateHash,
    ) -> Self {
        Self {
            data,
            state_hash: Some(hash),
            full_data: None,
        }
    }
}

macro_rules! finish_stage {
    ($handle:expr, $stage:ident) => {
        $handle.finish_stage(PipelineStage::$stage).await;
    };
}

pub struct CheckpointExecutor {
    epoch_store: Arc<AuthorityPerEpochStore>,
    state: Arc<AuthorityState>,
    checkpoint_store: Arc<CheckpointStore>,
    object_cache_reader: Arc<dyn ObjectCacheRead>,
    transaction_cache_reader: Arc<dyn TransactionCacheRead>,
    execution_scheduler: Arc<ExecutionSchedulerWrapper>,
    global_state_hasher: Arc<GlobalStateHasher>,
    backpressure_manager: Arc<BackpressureManager>,
    config: CheckpointExecutorConfig,
    metrics: Arc<CheckpointExecutorMetrics>,
    tps_estimator: Mutex<TPSEstimator>,
    checkpoint_progress_tracker: Option<Arc<CheckpointProgressTracker>>,
    data_sender: Option<CheckpointDataSender>,
}

impl CheckpointExecutor {
    pub fn new(
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_store: Arc<CheckpointStore>,
        state: Arc<AuthorityState>,
        global_state_hasher: Arc<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        config: CheckpointExecutorConfig,
        metrics: Arc<CheckpointExecutorMetrics>,
        data_sender: Option<CheckpointDataSender>,
        checkpoint_progress_tracker: Option<Arc<CheckpointProgressTracker>>,
    ) -> Self {
        Self {
            epoch_store,
            state: state.clone(),
            checkpoint_store,
            object_cache_reader: state.get_object_cache_reader().clone(),
            transaction_cache_reader: state.get_transaction_cache_reader().clone(),
            execution_scheduler: state.execution_scheduler().clone(),
            global_state_hasher,
            backpressure_manager,
            config,
            metrics,
            tps_estimator: Mutex::new(TPSEstimator::default()),
            checkpoint_progress_tracker,
            data_sender,
        }
    }

    pub fn new_for_tests(
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_store: Arc<CheckpointStore>,
        state: Arc<AuthorityState>,
        global_state_hasher: Arc<GlobalStateHasher>,
    ) -> Self {
        Self::new(
            epoch_store,
            checkpoint_store.clone(),
            state,
            global_state_hasher,
            BackpressureManager::new_from_checkpoint_store(&checkpoint_store),
            Default::default(),
            CheckpointExecutorMetrics::new_for_tests(),
            None, // No callback for data
            None, // No progress tracker for tests
        )
    }

    // Gets the next checkpoint to schedule for execution. If the epoch is already
    // completed, returns None.
    fn get_next_to_schedule(&self) -> Option<CheckpointSequenceNumber> {
        // Decide the first checkpoint to schedule for execution.
        // If we haven't executed anything in the past, we schedule checkpoint 0.
        // Otherwise we schedule the one after highest executed.
        let highest_executed = self
            .checkpoint_store
            .get_highest_executed_checkpoint()
            .unwrap();

        if let Some(highest_executed) = &highest_executed {
            if self.epoch_store.epoch() == highest_executed.epoch()
                && highest_executed.is_last_checkpoint_of_epoch()
            {
                // We can arrive at this point if we bump the highest_executed_checkpoint
                // watermark, and then crash before completing reconfiguration.
                info!(seq = ?highest_executed.sequence_number, "final checkpoint of epoch has already been executed");
                return None;
            }
        }

        Some(
            highest_executed
                .as_ref()
                .map(|c| c.sequence_number() + 1)
                .unwrap_or_else(|| {
                    // TODO this invariant may no longer hold once we introduce snapshots
                    assert_eq!(self.epoch_store.epoch(), 0);
                    // we need to execute the genesis checkpoint
                    0
                }),
        )
    }

    /// Execute all checkpoints for the current epoch, ensuring that the node
    /// has not forked, and return when finished.
    /// If `run_with_range` is set, execution will stop early.
    #[instrument(level = "error", skip_all, fields(epoch = ?self.epoch_store.epoch()))]
    pub async fn run_epoch(self, run_with_range: Option<RunWithRange>) -> StopReason {
        let _metrics_scope = iota_metrics::monitored_scope("CheckpointExecutor::run_epoch");
        info!(?run_with_range, "CheckpointExecutor::run_epoch");
        debug!(
            "Checkpoint executor running for epoch {:?}",
            self.epoch_store.epoch(),
        );

        // check if we want to run this epoch based on RunWithRange condition value
        // we want to be inclusive of the defined RunWithRangeEpoch::Epoch
        // i.e Epoch(N) means we will execute epoch N and stop when reaching N+1
        if run_with_range.is_some_and(|rwr| rwr.is_epoch_gt(self.epoch_store.epoch())) {
            info!("RunWithRange condition satisfied at {:?}", run_with_range,);
            return StopReason::RunWithRangeCondition;
        };

        self.metrics
            .checkpoint_exec_epoch
            .set(self.epoch_store.epoch() as i64);

        let Some(next_to_schedule) = self.get_next_to_schedule() else {
            return StopReason::EpochComplete;
        };

        let this = Arc::new(self);

        let concurrency = std::env::var("IOTA_CHECKPOINT_EXECUTION_MAX_CONCURRENCY")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(this.config.checkpoint_execution_max_concurrency);

        let pipeline_stages = PipelineStages::new(
            next_to_schedule,
            this.metrics.clone(),
            this.checkpoint_progress_tracker.clone(),
        );

        let final_checkpoint_executed = stream_synced_checkpoints(
            this.checkpoint_store.clone(),
            next_to_schedule,
            run_with_range.and_then(|rwr| rwr.into_checkpoint_bound()),
        )
        // Checkpoint loading and execution is parallelized: each
        // checkpoint's task loads its transaction data before awaiting
        // admission to the ordered pipeline stages.
        .map(|checkpoint| {
            let this = this.clone();
            let pipeline_stages = pipeline_stages.clone();
            async move {
                let seq = checkpoint.sequence_number();
                let pipeline_handle = async move { pipeline_stages.handle(seq).await };
                tokio::spawn(this.execute_checkpoint(checkpoint, pipeline_handle))
                    .await
                    .unwrap()
            }
        })
        .buffered(concurrency)
        // Take the last value from the stream to determine if we completed the epoch
        .fold(false, |state, is_final_checkpoint| async move {
            assert!(!state, "Cannot execute checkpoint after epoch end");
            is_final_checkpoint
        })
        .await;

        if final_checkpoint_executed {
            StopReason::EpochComplete
        } else {
            StopReason::RunWithRangeCondition
        }
    }
}

impl CheckpointExecutor {
    /// Load all data for a checkpoint, ensure all transactions are executed,
    /// and check for forks.
    #[instrument(level = "debug", skip_all, fields(seq = ?checkpoint.sequence_number()))]
    async fn execute_checkpoint(
        self: Arc<Self>,
        checkpoint: VerifiedCheckpoint,
        pipeline_handle: impl Future<Output = PipelineHandle> + Send,
    ) -> bool /* is final checkpoint */ {
        debug!("executing checkpoint");
        let sequence_number = checkpoint.sequence_number;
        let is_last_checkpoint_of_epoch = checkpoint.is_last_checkpoint_of_epoch();

        checkpoint.report_checkpoint_age(&self.metrics.checkpoint_contents_age);
        self.backpressure_manager
            .update_highest_certified_checkpoint(sequence_number);

        // Load the transaction data before awaiting admission to the
        // pipeline: it reads only state-sync-written data (contents,
        // transactions, and expected effects), so it can run in parallel
        // across the buffered checkpoints instead of inside the ordered
        // ExecuteTransactions stage. The clone is needed because the
        // locally-built-checkpoint path below wants the checkpoint too.
        let loaded = (self.state.is_fullnode(&self.epoch_store) || is_last_checkpoint_of_epoch)
            .then(|| {
                let _scope = iota_metrics::monitored_scope(
                    "CheckpointExecutor::load_checkpoint_transactions",
                );
                self.load_checkpoint_transactions(checkpoint.clone())
            });

        let mut pipeline_handle = pipeline_handle.await;

        if is_last_checkpoint_of_epoch && sequence_number > 0 {
            let _wait_for_previous_checkpoints_guard =
                iota_metrics::monitored_scope("CheckpointExecutor::wait_for_previous_checkpoints");

            info!(
                "Reached end of epoch checkpoint, waiting for all previous checkpoints to be executed"
            );
            self.checkpoint_store
                .notify_read_executed_checkpoint(sequence_number - 1)
                .await;
        }

        let _parallel_step_guard =
            iota_metrics::monitored_scope("CheckpointExecutor::parallel_step");

        // Note: only `execute_transactions_from_synced_checkpoint` has end-of-epoch
        // logic.
        let exec_start = Instant::now();
        let ckpt_state = match loaded {
            Some((ckpt_state, tx_data)) => {
                self.execute_transactions_from_synced_checkpoint(
                    ckpt_state,
                    tx_data,
                    &mut pipeline_handle,
                )
                .await
            }
            None => {
                self.verify_locally_built_checkpoint(checkpoint, &mut pipeline_handle)
                    .await
            }
        };

        let tps = self.tps_estimator.lock().update(
            Instant::now(),
            ckpt_state.data.checkpoint.network_total_transactions,
        );
        self.metrics.checkpoint_exec_sync_tps.set(tps as i64);

        self.backpressure_manager
            .update_highest_executed_checkpoint(ckpt_state.data.checkpoint.sequence_number());

        let is_final_checkpoint = ckpt_state.data.checkpoint.is_last_checkpoint_of_epoch();

        let seq = ckpt_state.data.checkpoint.sequence_number;

        let batch = self.state.get_cache_commit().build_db_batch(
            self.epoch_store.epoch(),
            seq,
            &ckpt_state.data.tx_digests,
        );

        finish_stage!(pipeline_handle, BuildDbBatch);

        let mut ckpt_state = tokio::task::spawn_blocking({
            let this = self.clone();
            move || {
                // Commit all transaction effects to disk
                let cache_commit = this.state.get_cache_commit();
                debug!(?seq, "committing checkpoint transactions to disk");
                cache_commit.commit_transaction_outputs(
                    this.epoch_store.epoch(),
                    batch,
                    &ckpt_state.data.tx_digests,
                );
                ckpt_state
            }
        })
        .await
        .unwrap();

        finish_stage!(pipeline_handle, CommitTransactionOutputs);

        self.epoch_store
            .handle_finalized_checkpoint(&ckpt_state.data.checkpoint, &ckpt_state.data.tx_digests)
            .expect("cannot fail");

        let randomness_rounds = self.extract_randomness_rounds(
            &ckpt_state.data.checkpoint,
            &ckpt_state.data.checkpoint_contents,
        );

        // Once the checkpoint is finalized, we know that any randomness contained in
        // this checkpoint has been successfully included in a checkpoint
        // certified by quorum of validators. (RandomnessManager/
        // RandomnessReporter is only present on validators.)
        if let Some(randomness_reporter) = self.epoch_store.randomness_reporter() {
            for round in randomness_rounds {
                debug!(
                    ?round,
                    "notifying RandomnessReporter that randomness update was executed in checkpoint"
                );
                randomness_reporter
                    .notify_randomness_in_checkpoint(round)
                    .expect("epoch cannot have ended");
            }
        }

        finish_stage!(pipeline_handle, FinalizeCheckpoint);

        if let Some(checkpoint_data) = ckpt_state.full_data.take() {
            self.commit_index_updates(checkpoint_data);
        }

        finish_stage!(pipeline_handle, UpdateRpcIndex);

        self.global_state_hasher
            .accumulate_running_root(&self.epoch_store, seq, ckpt_state.state_hash)
            .expect("Failed to accumulate running root");

        if is_final_checkpoint {
            self.checkpoint_store
                .insert_epoch_last_checkpoint(self.epoch_store.epoch(), &ckpt_state.data.checkpoint)
                .expect("Failed to insert epoch last checkpoint");

            self.global_state_hasher
                .accumulate_epoch(self.epoch_store.clone(), seq)
                .expect("Accumulating epoch cannot fail");

            self.checkpoint_store
                .prune_local_summaries()
                .tap_err(|e| debug_fatal!("Failed to prune local summaries: {}", e))
                .ok();
        }

        fail_point!("crash");

        self.bump_highest_executed_checkpoint(&ckpt_state.data.checkpoint);

        self.broadcast_checkpoint(&ckpt_state.data, ckpt_state.full_data.as_ref());

        // Nudge the pruner now that this checkpoint is executed and available;
        // pruning of aged-out data runs off the propagation path.
        self.state
            .pruner()
            .nudge(ckpt_state.data.checkpoint.sequence_number());

        finish_stage!(pipeline_handle, BumpHighestExecutedCheckpoint);

        if let Some(tracker) = &self.checkpoint_progress_tracker {
            tracker.add_execution_time(exec_start.elapsed());
        }

        // Important: code after the last pipeline stage is finished can run out of
        // checkpoint order.

        ckpt_state.data.checkpoint.is_last_checkpoint_of_epoch()
    }

    // On validators, checkpoints have often already been constructed locally, in
    // which case we can skip many steps of the checkpoint execution process.
    #[instrument(level = "info", skip_all)]
    async fn verify_locally_built_checkpoint(
        &self,
        checkpoint: VerifiedCheckpoint,
        pipeline_handle: &mut PipelineHandle,
    ) -> CheckpointExecutionState {
        assert!(
            !checkpoint.is_last_checkpoint_of_epoch(),
            "only fullnode path has end-of-epoch logic"
        );

        let sequence_number = checkpoint.sequence_number;
        let locally_built_checkpoint = self
            .checkpoint_store
            .get_locally_computed_checkpoint(sequence_number)
            .expect("db error");

        let Some(locally_built_checkpoint) = locally_built_checkpoint else {
            // fall back to tx-by-tx execution path if we are catching up.
            let (ckpt_state, tx_data) = self.load_checkpoint_transactions(checkpoint);
            return self
                .execute_transactions_from_synced_checkpoint(ckpt_state, tx_data, pipeline_handle)
                .await;
        };

        self.metrics.checkpoint_executor_validator_path.inc();

        // Check for fork
        assert_checkpoint_not_forked(
            &locally_built_checkpoint,
            &checkpoint,
            &self.checkpoint_store,
        );

        // Checkpoint builder triggers accumulation of the checkpoint, so this is
        // guaranteed to finish.
        let state_hash = {
            let _metrics_scope =
                iota_metrics::monitored_scope("CheckpointExecutor::notify_read_state_hash");
            self.epoch_store
                .notify_read_checkpoint_state_hasher(&[sequence_number])
                .await
                .unwrap()
                .pop()
                .unwrap()
        };

        // Checkpoint builder triggers accumulation of the checkpoint, so this is
        // guaranteed to finish.

        let checkpoint_contents = self
            .checkpoint_store
            .get_checkpoint_contents(&checkpoint.contents_digest)
            .expect("db error")
            .expect("checkpoint contents not found");

        let (tx_digests, fx_digests): (Vec<_>, Vec<_>) = checkpoint_contents
            .iter()
            .map(|digests| (digests.transaction, digests.effects))
            .unzip();

        pipeline_handle
            .skip_to(PipelineStage::FinalizeTransactions)
            .await;

        // Currently this code only runs on validators, where this method call does
        // nothing. But in the future, fullnodes may follow the consensus dag
        // and build their own checkpoints.
        self.insert_finalized_transactions(&tx_digests, sequence_number, checkpoint.timestamp_ms);

        pipeline_handle.skip_to(PipelineStage::BuildDbBatch).await;

        CheckpointExecutionState::new_with_global_state_hash(
            CheckpointExecutionData {
                checkpoint,
                checkpoint_contents,
                tx_digests,
                fx_digests,
            },
            state_hash,
        )
    }

    #[instrument(level = "info", skip_all)]
    async fn execute_transactions_from_synced_checkpoint(
        &self,
        mut ckpt_state: CheckpointExecutionState,
        tx_data: CheckpointTransactionData,
        pipeline_handle: &mut PipelineHandle,
    ) -> CheckpointExecutionState {
        let sequence_number = ckpt_state.data.checkpoint.sequence_number;

        let (unexecuted_tx_digests, unexecuted_expected_fx_digests) = {
            let _scope = iota_metrics::monitored_scope("CheckpointExecutor::execute_transactions");
            self.schedule_transaction_execution(&ckpt_state, &tx_data)
        };

        finish_stage!(pipeline_handle, ExecuteTransactions);

        {
            let actual_fx_digests = self
                .transaction_cache_reader
                .notify_read_executed_effects_digests(
                    "CheckpointExecutor::notify_read_executed_effects_digests",
                    &unexecuted_tx_digests,
                )
                .await;

            // A transaction scheduled here can also be scheduled from consensus,
            // and whichever enqueue arrives second is dropped along with its
            // environment — so the expected digest passed above may never have
            // reached execution. Compare here, where the checkpoint's digests are
            // known, so a fork is caught regardless of which enqueue won.
            for (tx_digest, expected, actual) in itertools::izip!(
                unexecuted_tx_digests.iter(),
                unexecuted_expected_fx_digests.iter(),
                actual_fx_digests.iter()
            ) {
                assert_not_forked(
                    &ckpt_state.data.checkpoint,
                    tx_digest,
                    expected,
                    actual,
                    &*self.transaction_cache_reader,
                );
            }
        }

        finish_stage!(pipeline_handle, WaitForTransactions);

        if ckpt_state.data.checkpoint.is_last_checkpoint_of_epoch() {
            self.execute_change_epoch_tx(&tx_data).await;
        }

        let _scope = iota_metrics::monitored_scope("CheckpointExecutor::finalize_checkpoint");

        if self.state.is_fullnode(&self.epoch_store) {
            self.state.congestion_tracker.process_checkpoint_effects(
                &*self.transaction_cache_reader,
                &ckpt_state.data.checkpoint,
                &tx_data.effects,
            );
        }

        self.insert_finalized_transactions(
            &ckpt_state.data.tx_digests,
            sequence_number,
            ckpt_state.data.checkpoint.timestamp_ms,
        );

        // The early versions of the hasher (prior to effectsv2) rely on db
        // state, so we must wait until all transactions have been executed
        // before accumulating the checkpoint.
        ckpt_state.state_hash = Some(
            self.global_state_hasher
                .accumulate_checkpoint(&tx_data.effects, sequence_number, &self.epoch_store)
                .expect("epoch cannot have ended"),
        );

        finish_stage!(pipeline_handle, FinalizeTransactions);

        ckpt_state.full_data = self.process_checkpoint_data(&ckpt_state.data, &tx_data);

        finish_stage!(pipeline_handle, ProcessCheckpointData);

        ckpt_state
    }

    fn checkpoint_data_enabled(&self) -> bool {
        self.state.grpc_indexes_store.is_some()
            || self.config.data_ingestion_dir.is_some()
            || self.data_sender.is_some()
    }

    fn insert_finalized_transactions(
        &self,
        tx_digests: &[TransactionDigest],
        sequence_number: CheckpointSequenceNumber,
        timestamp_ms: u64,
    ) {
        self.epoch_store
            .insert_finalized_transactions(tx_digests, sequence_number, timestamp_ms)
            .expect("failed to insert finalized transactions");

        if self.state.is_fullnode(&self.epoch_store) {
            // TODO remove once we no longer need to support this table for read RPC
            self.state
                .get_checkpoint_cache()
                .insert_finalized_transactions_perpetual_checkpoints(
                    tx_digests,
                    self.epoch_store.epoch(),
                    sequence_number,
                );
        }
    }

    #[instrument(level = "info", skip_all)]
    fn process_checkpoint_data(
        &self,
        ckpt_data: &CheckpointExecutionData,
        tx_data: &CheckpointTransactionData,
    ) -> Option<CheckpointData> {
        let is_checkpoint_data_enabled = self.checkpoint_data_enabled();
        // Boundaries always need full `CheckpointData` to persist `epoch_info`,
        // even when no other consumer is configured.
        let is_last_checkpoint_of_epoch = ckpt_data.checkpoint.is_last_checkpoint_of_epoch();
        if !is_checkpoint_data_enabled && !is_last_checkpoint_of_epoch {
            return None;
        }

        let checkpoint_data = load_checkpoint_data(
            ckpt_data,
            tx_data,
            self.state.get_object_store(),
            &*self.transaction_cache_reader,
        )
        .expect("failed to load checkpoint data");

        // Persist the boundary's `epoch_info` row eagerly. Two properties make
        // that safe. Boundaries run in epoch order: the boundary waits for every
        // earlier checkpoint (`notify_read_executed_checkpoint(seq - 1)` in
        // `execute_checkpoint`) and the stream stops at epoch end, so two
        // boundaries are never in flight. And `index_epoch_boundary` is
        // idempotent — a row upsert plus a contiguous +1 watermark guard — so a
        // crash before the executed watermark advances just re-applies it on the
        // next run.
        if is_last_checkpoint_of_epoch {
            self.checkpoint_store
                .index_epoch_boundary(&checkpoint_data)
                .expect("failed to persist epoch info at boundary");
        }

        if !is_checkpoint_data_enabled {
            // Data was built solely to seed `epoch_info`; skip the other consumers.
            return None;
        }

        // Index the checkpoint. The grpc indexes accumulate non-idempotent state
        // (owner indexes, live-object sets), so each update must land exactly
        // once. Indexing runs here out of order (checkpoints execute
        // concurrently), so the write is only staged now and committed later, in
        // sequence order, via `commit_update_for_checkpoint` — keeping the grpc
        // watermark consistent with the executed checkpoint and crash-safe.
        if let Some(grpc_indexes_store) = &self.state.grpc_indexes_store {
            grpc_indexes_store.index_checkpoint(&checkpoint_data);
        }

        if let Some(path) = &self.config.data_ingestion_dir {
            store_checkpoint_locally(path, &checkpoint_data)
                .expect("failed to store checkpoint locally");
        }

        Some(checkpoint_data)
    }

    // Load all required transaction and effects data for the checkpoint.
    #[instrument(level = "info", skip_all)]
    fn load_checkpoint_transactions(
        &self,
        checkpoint: VerifiedCheckpoint,
    ) -> (CheckpointExecutionState, CheckpointTransactionData) {
        let seq = checkpoint.sequence_number;
        let epoch = checkpoint.epoch;

        let checkpoint_contents = self
            .checkpoint_store
            .get_checkpoint_contents(&checkpoint.contents_digest)
            .expect("db error")
            .expect("checkpoint contents not found");

        // attempt to load full checkpoint contents in bulk
        if let Some(full_contents) = self
            .checkpoint_store
            .get_full_checkpoint_contents_by_sequence_number(seq)
            .tap_some(|_| debug!("loaded full checkpoint contents in bulk for sequence {seq}"))
        {
            let num_txns = full_contents.size();
            let mut tx_digests = Vec::with_capacity(num_txns);
            let mut transactions = Vec::with_capacity(num_txns);
            let mut effects = Vec::with_capacity(num_txns);
            let mut fx_digests = Vec::with_capacity(num_txns);

            full_contents
                .iter()
                .zip(checkpoint_contents.iter())
                .for_each(|(execution_data, digests)| {
                    let tx_digest = digests.transaction;
                    let fx_digest = digests.effects;
                    debug_assert_eq!(tx_digest, *execution_data.transaction.digest());
                    debug_assert_eq!(fx_digest, execution_data.effects.digest());

                    tx_digests.push(tx_digest);
                    transactions.push(VerifiedExecutableTransaction::new_from_checkpoint(
                        VerifiedTransaction::new_unchecked(execution_data.transaction.clone()),
                        epoch,
                        seq,
                    ));
                    effects.push(execution_data.effects.clone());
                    fx_digests.push(fx_digest);
                });

            (
                CheckpointExecutionState::new(CheckpointExecutionData {
                    checkpoint,
                    checkpoint_contents,
                    tx_digests,
                    fx_digests,
                }),
                CheckpointTransactionData {
                    transactions,
                    effects,
                },
            )
        } else {
            // load items one-by-one

            let digests = checkpoint_contents.transactions();

            let (tx_digests, fx_digests): (Vec<_>, Vec<_>) =
                digests.iter().map(|d| (d.transaction, d.effects)).unzip();
            let verified_transactions: Vec<VerifiedTransaction> = self
                .transaction_cache_reader
                .multi_get_transaction_blocks(&tx_digests)
                .into_iter()
                .enumerate()
                .map(|(i, tx)| {
                    let tx = tx
                        .unwrap_or_else(|| fatal!("transaction not found for {:?}", tx_digests[i]));
                    Arc::try_unwrap(tx).unwrap_or_else(|tx| (*tx).clone())
                })
                .collect();
            let effects: Vec<TransactionEffects> = self
                .transaction_cache_reader
                .multi_get_effects(&fx_digests)
                .into_iter()
                .enumerate()
                .map(|(i, effect)| {
                    effect.unwrap_or_else(|| {
                        fatal!("checkpoint effect not found for {:?}", digests[i])
                    })
                })
                .collect();

            // Reached when the contents were not already cached: a fullnode
            // syncing from a peer, or a node catching up. Cache the assembled
            // contents so this node can in turn serve state-sync peers without
            // reconstruction. (Validators' locally built checkpoints are cached
            // by the checkpoint builder and take the bulk path above instead.)
            //
            // The assembly clones every transaction and effect, so skip it
            // when the cache wouldn't retain the entry (see `should_cache`).
            if self
                .checkpoint_store
                .should_cache_full_checkpoint_contents(seq)
            {
                let execution_data = verified_transactions
                    .iter()
                    .zip(effects.iter())
                    .map(|(tx, fx)| ExecutionData::new(tx.clone().into_inner(), fx.clone()));
                let full_contents = FullCheckpointContents::from_contents_and_execution_data(
                    checkpoint_contents.clone(),
                    execution_data,
                )
                // Both were loaded for this checkpoint's own contents above,
                // one transaction and one effects entry per digest.
                .expect("checkpoint has one executed transaction per pinned signature set");
                self.checkpoint_store.cache_full_checkpoint_contents(
                    seq,
                    checkpoint.contents_digest,
                    full_contents,
                );
            }

            let transactions = verified_transactions
                .into_iter()
                .map(|tx| VerifiedExecutableTransaction::new_from_checkpoint(tx, epoch, seq))
                .collect();

            (
                CheckpointExecutionState::new(CheckpointExecutionData {
                    checkpoint,
                    checkpoint_contents,
                    tx_digests,
                    fx_digests,
                }),
                CheckpointTransactionData {
                    transactions,
                    effects,
                },
            )
        }
    }

    /// Enqueues the checkpoint's not-yet-executed transactions, and returns
    /// their digests together with the effects digest the checkpoint expects
    /// for each, so the caller can check for a fork once they execute.
    #[instrument(level = "info", skip_all)]
    fn schedule_transaction_execution(
        &self,
        ckpt_state: &CheckpointExecutionState,
        tx_data: &CheckpointTransactionData,
    ) -> (Vec<TransactionDigest>, Vec<TransactionEffectsDigest>) {
        // Which transactions have already been executed must be read here, in
        // the ordered pipeline stage, and not when the checkpoint data is
        // loaded: execution progresses in between, and a stale answer would
        // skip the fork check for a transaction that executed since loading.
        let executed_fx_digests = self
            .transaction_cache_reader
            .multi_get_executed_effects_digests(&ckpt_state.data.tx_digests);

        // Find unexecuted transactions and their expected effects digests
        let (unexecuted_tx_digests, unexecuted_expected_fx_digests, unexecuted_txns): (
            Vec<_>,
            Vec<_>,
            Vec<_>,
        ) = itertools::multiunzip(
            itertools::izip!(
                tx_data.transactions.iter(),
                ckpt_state.data.tx_digests.iter(),
                ckpt_state.data.fx_digests.iter(),
                tx_data.effects.iter(),
                executed_fx_digests.iter()
            )
            .filter_map(
                |(txn, tx_digest, expected_fx_digest, effects, executed_fx_digest)| {
                    if let Some(executed_fx_digest) = executed_fx_digest {
                        assert_not_forked(
                            &ckpt_state.data.checkpoint,
                            tx_digest,
                            expected_fx_digest,
                            executed_fx_digest,
                            &*self.transaction_cache_reader,
                        );
                        None
                    } else if txn.transaction().is_end_of_epoch_tx() {
                        None
                    } else {
                        // A transaction without shared inputs normally needs no version
                        // assignments, but one cancelled for congestion carries its
                        // cancellation version on the gas object, which re-execution
                        // must reproduce.
                        let assigned_versions = if txn.contains_shared_object()
                            || gas_object_cancellation_version_from_effects(effects).is_some()
                        {
                            self.epoch_store
                                .acquire_shared_version_assignments_from_effects(
                                    txn,
                                    effects,
                                    &*self.object_cache_reader,
                                )
                                .expect("failed to initialize shared object versions from effects")
                        } else {
                            Default::default()
                        };

                        let env = ExecutionEnv::new()
                            .with_assigned_versions(assigned_versions)
                            .with_expected_effects_digest(*expected_fx_digest);

                        Some((tx_digest, *expected_fx_digest, (txn.clone(), env)))
                    }
                },
            ),
        );

        // Enqueue unexecuted transactions with their expected effects digests
        self.execution_scheduler
            .enqueue_transactions(unexecuted_txns, &self.epoch_store);

        (unexecuted_tx_digests, unexecuted_expected_fx_digests)
    }

    // Execute the change epoch txn
    #[instrument(level = "error", skip_all)]
    async fn execute_change_epoch_tx(&self, tx_data: &CheckpointTransactionData) {
        let change_epoch_tx = tx_data.transactions.last().unwrap();
        let change_epoch_fx = tx_data.effects.last().unwrap();
        assert_eq!(
            change_epoch_tx.digest(),
            change_epoch_fx.transaction_digest()
        );
        assert!(
            change_epoch_tx.transaction().is_end_of_epoch_tx(),
            "final txn must be an end of epoch txn"
        );

        // Ordinarily we would assert that the change epoch txn has not been executed
        // yet. However, during crash recovery, it is possible that we already
        // passed this point and the txn has been executed. You can uncomment
        // this assert if you are debugging a problem related to reconfig. If
        // you hit this assert and it is not because of crash-recovery,
        // it may indicate a bug in the checkpoint executor.
        //
        //     if self
        //         .transaction_cache_reader
        //         .get_executed_effects(change_epoch_tx.digest())
        //         .is_some()
        //     {
        //         fatal!(
        //             "end of epoch txn must not have been executed: {:?}",
        //             change_epoch_tx.digest()
        //         );
        //     }

        let assigned_versions = self
            .epoch_store
            .acquire_shared_version_assignments_from_effects(
                change_epoch_tx,
                change_epoch_fx,
                self.object_cache_reader.as_ref(),
            )
            .expect("failed to initialize shared object versions for the change epoch transaction");

        info!(
            "scheduling change epoch txn with digest: {:?}, expected effects digest: {:?}, \
            assigned versions: {:?}",
            change_epoch_tx.digest(),
            change_epoch_fx.digest(),
            assigned_versions
        );
        self.execution_scheduler.enqueue_transactions(
            vec![(
                change_epoch_tx.clone(),
                ExecutionEnv::new()
                    .with_assigned_versions(assigned_versions)
                    .with_expected_effects_digest(change_epoch_fx.digest()),
            )],
            &self.epoch_store,
        );

        self.transaction_cache_reader
            .notify_read_executed_effects_digests(
                "CheckpointExecutor::notify_read_advance_epoch_tx",
                &[*change_epoch_tx.digest()],
            )
            .await;
    }

    // Increment the highest executed checkpoint watermark
    #[instrument(level = "debug", skip_all)]
    fn bump_highest_executed_checkpoint(&self, checkpoint: &VerifiedCheckpoint) {
        // Ensure that we are not skipping checkpoints at any point
        let seq = checkpoint.sequence_number();
        debug!("Bumping highest_executed_checkpoint watermark to {seq:?}");
        if let Some(prev_highest) = self
            .checkpoint_store
            .get_highest_executed_checkpoint_seq_number()
            .unwrap()
        {
            assert_eq!(prev_highest + 1, seq);
        } else {
            assert_eq!(seq, 0);
        }
        fail_point!("highest-executed-checkpoint");

        self.checkpoint_store
            .update_highest_executed_checkpoint(checkpoint)
            .unwrap();
        self.metrics.last_executed_checkpoint.set(seq as i64);

        self.metrics
            .last_executed_checkpoint_timestamp_ms
            .set(checkpoint.timestamp_ms as i64);
        checkpoint.report_checkpoint_age(&self.metrics.last_executed_checkpoint_age);
    }

    /// Helper to broadcast checkpoint summary and data if the
    /// channels are set.
    fn broadcast_checkpoint(
        &self,
        checkpoint_exec_data: &CheckpointExecutionData,
        checkpoint_data: Option<&CheckpointData>,
    ) {
        if let Some(data_sender) = &self.data_sender {
            let checkpoint_data = if let Some(data) = checkpoint_data {
                data.clone()
            } else {
                // Reconstruct checkpoint data if needed (rare case: data_sender configured but
                // checkpoint_data_enabled is false)
                let (_, tx_data) =
                    self.load_checkpoint_transactions(checkpoint_exec_data.checkpoint.clone());
                load_checkpoint_data(
                    checkpoint_exec_data,
                    &tx_data,
                    self.state.get_object_store(),
                    self.transaction_cache_reader.as_ref(),
                )
                .expect("Failed to load full CheckpointData")
            };
            data_sender(&checkpoint_data);
        }

        debug!(
            "[Fullnode] Full CheckpointData is available: seq={}",
            checkpoint_exec_data.checkpoint.sequence_number()
        );
    }

    /// If configured, commit the pending index updates for the provided
    /// checkpoint
    #[instrument(level = "info", skip_all)]
    fn commit_index_updates(&self, checkpoint: CheckpointData) {
        if let Some(grpc_indexes_store) = &self.state.grpc_indexes_store {
            grpc_indexes_store
                .commit_update_for_checkpoint(checkpoint.checkpoint_summary.sequence_number)
                .expect("failed to update gRPC indexes");
        }
    }

    // Extract randomness rounds from the checkpoint version-specific data (if
    // available). Otherwise, extract randomness rounds from the first
    // transaction in the checkpoint
    #[instrument(level = "debug", skip_all)]
    fn extract_randomness_rounds(
        &self,
        checkpoint: &VerifiedCheckpoint,
        checkpoint_contents: &CheckpointContents,
    ) -> Vec<RandomnessRound> {
        if let Some(version_specific_data) = checkpoint
            .parse_version_specific_data(self.epoch_store.protocol_config())
            .expect("unable to get version_specific_data")
        {
            // With version-specific data, randomness rounds are stored in checkpoint
            // summary.
            version_specific_data.into_v1().randomness_rounds
        } else {
            // Before version-specific data, checkpoint batching must be disabled. In this
            // case, randomness state update tx must be first if it exists,
            // because all other transactions in a checkpoint that includes a
            // randomness state update are causally dependent on it.
            assert_eq!(
                0,
                self.epoch_store
                    .protocol_config()
                    .min_checkpoint_interval_ms_as_option()
                    .unwrap_or_default(),
            );
            if let Some(first_digest) = checkpoint_contents.transactions().first() {
                let maybe_randomness_tx = self.transaction_cache_reader.get_transaction_block(&first_digest.transaction)
                .unwrap_or_else(||
                    fatal!(
                        "state-sync should have ensured that transaction with digests {first_digest:?} exists for checkpoint: {}",
                        checkpoint.sequence_number()
                    )
                );
                if let TransactionKind::RandomnessStateUpdate(rsu) =
                    maybe_randomness_tx.data().transaction().kind()
                {
                    vec![rsu.randomness_round]
                } else {
                    Vec::new()
                }
            } else {
                Vec::new()
            }
        }
    }
}