iota_core/checkpoints/

mod.rs

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

mod causal_order;
pub mod checkpoint_executor;
mod checkpoint_output;
mod epoch_info;
mod full_checkpoint_contents_cache;
mod metrics;

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
    fs::File,
    future::Future,
    io::Write,
    path::Path,
    pin::Pin,
    sync::{Arc, Weak},
    task::{Context, Poll},
    time::{Duration, SystemTime},
};

use diffy::create_patch;
use iota_common::{
    debug_fatal, fatal,
    random::get_rng,
    sync::notify_read::{CHECKPOINT_BUILDER_NOTIFY_READ_TASK_NAME, NotifyRead},
};
use iota_metrics::{MonitoredFutureExt, monitored_future, monitored_scope};
use iota_network::default_iota_network_config;
use iota_sdk_types::{
    CheckpointContentsDigest, CheckpointDigest, GasCostSummary, TransactionDigest, TransactionKind,
    UserSignature,
    checkpoint::{CheckpointCommitment, CheckpointContents, CheckpointSummary, EndOfEpochData},
};
use iota_types::{
    base_types::{AuthorityName, ConciseableName, EpochId, ExecutionData},
    committee::StakeUnit,
    crypto::AuthorityStrongQuorumSignInfo,
    effects::{TransactionEffects, TransactionEffectsAPI, TransactionEffectsExt},
    error::{IotaError, IotaResult},
    event::SystemEpochInfoEvent,
    iota_system_state::{
        IotaSystemState, IotaSystemStateTrait,
        epoch_start_iota_system_state::EpochStartSystemStateTrait,
    },
    messages_checkpoint::{
        CertifiedCheckpointSummary, CheckpointContentsExt, CheckpointRequest, CheckpointResponse,
        CheckpointSequenceNumber, CheckpointSignatureMessage, CheckpointSummaryExt,
        CheckpointSummaryResponse, CheckpointTimestamp, FullCheckpointContents,
        SignedCheckpointSummary, TrustedCheckpoint, VerifiedCheckpoint, VerifiedCheckpointContents,
    },
    messages_consensus::ConsensusTransactionKey,
    storage::EpochInfoV2,
    transaction::{TransactionDataAPI, TransactionEnvelope, TransactionKey},
};
use itertools::Itertools;
use nonempty::NonEmpty;
use parking_lot::Mutex;
use pin_project_lite::pin_project;
use rand::seq::SliceRandom;
use serde::{Deserialize, Serialize};
use tokio::{
    sync::{Notify, mpsc, watch},
    task::JoinSet,
    time::timeout,
};
use tracing::{debug, error, info, instrument, trace, warn};
use typed_store::{
    DBMapUtils, Map, TypedStoreError,
    rocks::{DBMap, MetricConf},
};

pub use crate::checkpoints::{
    checkpoint_output::{
        LogCheckpointOutput, SendCheckpointToStateSync, SubmitCheckpointToConsensus,
    },
    full_checkpoint_contents_cache::{
        FullCheckpointContentsCache, FullCheckpointContentsCacheMetrics,
    },
    metrics::CheckpointMetrics,
};
use crate::{
    authority::{
        AuthorityState,
        authority_per_epoch_store::{AuthorityPerEpochStore, scorer::MAX_SCORE},
    },
    authority_client::{
        make_network_authority_clients_with_network_config, validator_peer::ValidatorPeerAPI,
    },
    checkpoints::{
        causal_order::CausalOrder,
        checkpoint_output::{CertifiedCheckpointOutput, CheckpointOutput},
    },
    consensus_handler::SequencedConsensusTransactionKey,
    consensus_manager::ReplayWaiter,
    execution_cache::TransactionCacheRead,
    global_state_hasher::GlobalStateHasher,
    stake_aggregator::{InsertResult, MultiStakeAggregator},
};

pub type CheckpointHeight = u64;

pub struct EpochStats {
    pub checkpoint_count: u64,
    pub transaction_count: u64,
    pub total_gas_reward: u64,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PendingCheckpointInfo {
    pub timestamp_ms: CheckpointTimestamp,
    pub last_of_epoch: bool,
    pub checkpoint_height: CheckpointHeight,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum PendingCheckpoint {
    // This is an enum for future upgradability, though at the moment there is only one variant.
    V1(PendingCheckpointContentsV1),
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PendingCheckpointContentsV1 {
    pub roots: Vec<TransactionKey>,
    pub details: PendingCheckpointInfo,
}

impl PendingCheckpoint {
    pub fn as_v1(&self) -> &PendingCheckpointContentsV1 {
        match self {
            PendingCheckpoint::V1(contents) => contents,
        }
    }

    pub fn into_v1(self) -> PendingCheckpointContentsV1 {
        match self {
            PendingCheckpoint::V1(contents) => contents,
        }
    }

    pub fn roots(&self) -> &Vec<TransactionKey> {
        &self.as_v1().roots
    }

    pub fn details(&self) -> &PendingCheckpointInfo {
        &self.as_v1().details
    }

    pub fn height(&self) -> CheckpointHeight {
        self.details().checkpoint_height
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct BuilderCheckpointSummary {
    pub summary: CheckpointSummary,
    // Height at which this checkpoint summary was built. None for genesis checkpoint
    pub checkpoint_height: Option<CheckpointHeight>,
    pub position_in_commit: usize,
}

#[derive(DBMapUtils)]
pub struct CheckpointStoreTables {
    /// Maps checkpoint contents digest to checkpoint contents
    pub(crate) checkpoint_content: DBMap<CheckpointContentsDigest, CheckpointContents>,

    /// Deprecated: the contents-digest to sequence-number mapping moved to
    /// the in-memory [`FullCheckpointContentsCache`]. Dropped on open; not
    /// migrated (entries were a short-lived cache).
    #[allow(dead_code)]
    #[deprecated_db_map]
    checkpoint_sequence_by_contents_digest: Option<DBMap<(), ()>>,

    /// Deprecated: full checkpoint contents moved to the in-memory
    /// [`FullCheckpointContentsCache`]. Dropped on open; not migrated
    /// (entries were a short-lived cache, and readers can reconstruct full
    /// contents from `checkpoint_content` plus the transaction and effects
    /// stores).
    #[allow(dead_code)]
    #[deprecated_db_map]
    full_checkpoint_content: Option<DBMap<(), ()>>,

    /// Stores certified checkpoints
    pub(crate) certified_checkpoints: DBMap<CheckpointSequenceNumber, TrustedCheckpoint>,
    /// Map from checkpoint digest to certified checkpoint
    pub(crate) checkpoint_by_digest: DBMap<CheckpointDigest, TrustedCheckpoint>,

    /// Store locally computed checkpoint summaries so that we can detect forks
    /// and log useful information. Can be pruned as soon as we verify that
    /// we are in agreement with the latest certified checkpoint.
    pub(crate) locally_computed_checkpoints: DBMap<CheckpointSequenceNumber, CheckpointSummary>,

    /// A map from epoch ID to the sequence number of the last checkpoint in
    /// that epoch.
    ///
    /// Written at certification time, before the boundary executes, so it
    /// records the closing sequence number independently of the later-finalized
    /// `epoch_info` row. Never pruned.
    epoch_last_checkpoint_map: DBMap<EpochId, CheckpointSequenceNumber>,

    /// Per-epoch verified data (start-of-epoch identity plus its close-of-epoch
    /// proof bundle ([`EpochInfoV2`])) keyed by epoch ID. Populated at every
    /// epoch boundary by the checkpoint executor and seeded from a formal
    /// snapshot's `EPOCH_INFO` on restore.
    ///
    /// Intentionally not pruned: callers (the snapshot writer, the gRPC API,
    /// etc.) need full `[0, snapshot_epoch]` coverage, so
    /// this table grows unboundedly with epoch count (one row per epoch, ever)
    /// by design. Do not add it to `prune_checkpoints`.
    ///
    /// Completeness is tracked by `epoch_info_watermark`.
    epoch_info: DBMap<EpochId, EpochInfoV2>,

    /// Highest epoch whose `epoch_info` row is finalized (close-of-epoch proof
    /// present), as a contiguous prefix `[0, watermark]`. Singleton keyed by
    /// `()`. Advanced atomically with the close-of-epoch upsert in
    /// `index_epoch`, and recomputed over a seeded prefix by
    /// `insert_epoch_info`; only ever raised. Never pruned.
    epoch_info_watermark: DBMap<(), EpochId>,

    /// Watermarks used to determine the highest verified, fully synced, and
    /// fully executed checkpoints
    pub(crate) watermarks: DBMap<CheckpointWatermark, (CheckpointSequenceNumber, CheckpointDigest)>,
}

impl CheckpointStoreTables {
    pub fn new(path: &Path, metric_name: &'static str) -> Self {
        Self::open_tables_read_write(path.to_path_buf(), MetricConf::new(metric_name), None, None)
    }
    pub fn open_readonly(path: &Path) -> CheckpointStoreTablesReadOnly {
        Self::get_read_only_handle(
            path.to_path_buf(),
            None,
            None,
            MetricConf::new("checkpoint_readonly"),
        )
    }
}

pub struct CheckpointStore {
    pub(crate) tables: CheckpointStoreTables,
    full_checkpoint_contents_cache: FullCheckpointContentsCache,
    synced_checkpoint_notify_read: NotifyRead<CheckpointSequenceNumber, VerifiedCheckpoint>,
    executed_checkpoint_notify_read: NotifyRead<CheckpointSequenceNumber, VerifiedCheckpoint>,
}

impl CheckpointStore {
    pub fn new(path: &Path) -> Arc<Self> {
        Self::new_with_contents_cache(path, FullCheckpointContentsCache::default())
    }

    pub fn new_with_contents_cache(
        path: &Path,
        contents_cache: FullCheckpointContentsCache,
    ) -> Arc<Self> {
        let tables = CheckpointStoreTables::new(path, "checkpoint");
        Arc::new(Self {
            tables,
            full_checkpoint_contents_cache: contents_cache,
            synced_checkpoint_notify_read: NotifyRead::new(),
            executed_checkpoint_notify_read: NotifyRead::new(),
        })
    }

    pub fn new_for_tests() -> Arc<Self> {
        let storage_dir = iota_common::tempdir().keep();
        CheckpointStore::new(storage_dir.as_path())
    }

    pub fn open_readonly(path: &Path) -> CheckpointStoreTablesReadOnly {
        CheckpointStoreTables::open_readonly(path)
    }

    #[instrument(level = "info", skip_all)]
    pub fn insert_genesis_checkpoint(
        &self,
        checkpoint: VerifiedCheckpoint,
        contents: CheckpointContents,
        epoch_store: &AuthorityPerEpochStore,
    ) {
        assert_eq!(
            checkpoint.epoch(),
            0,
            "can't call insert_genesis_checkpoint with a checkpoint not in epoch 0"
        );
        assert_eq!(
            checkpoint.sequence_number(),
            0,
            "can't call insert_genesis_checkpoint with a checkpoint that doesn't have a sequence number of 0"
        );

        // Only insert the genesis checkpoint if the DB is empty and doesn't have it
        // already
        if self
            .get_checkpoint_by_digest(checkpoint.digest())
            .unwrap()
            .is_none()
        {
            if epoch_store.epoch() == checkpoint.epoch {
                epoch_store
                    .put_genesis_checkpoint_in_builder(checkpoint.data(), &contents)
                    .unwrap();
            } else {
                debug!(
                    validator_epoch =% epoch_store.epoch(),
                    genesis_epoch =% checkpoint.epoch(),
                    "Not inserting checkpoint builder data for genesis checkpoint",
                );
            }
            self.insert_checkpoint_contents(contents).unwrap();
            self.insert_verified_checkpoint(&checkpoint).unwrap();
            self.update_highest_synced_checkpoint(&checkpoint).unwrap();
        }
    }

    pub fn get_checkpoint_by_digest(
        &self,
        digest: &CheckpointDigest,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        self.tables
            .checkpoint_by_digest
            .get(digest)
            .map(|maybe_checkpoint| maybe_checkpoint.map(|c| c.into()))
    }

    pub fn get_checkpoint_by_sequence_number(
        &self,
        sequence_number: CheckpointSequenceNumber,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        self.tables
            .certified_checkpoints
            .get(&sequence_number)
            .map(|maybe_checkpoint| maybe_checkpoint.map(|c| c.into()))
    }

    pub fn get_locally_computed_checkpoint(
        &self,
        sequence_number: CheckpointSequenceNumber,
    ) -> Result<Option<CheckpointSummary>, TypedStoreError> {
        self.tables
            .locally_computed_checkpoints
            .get(&sequence_number)
    }

    /// Get full checkpoint contents by contents digest from the in-memory
    /// contents cache.
    ///
    /// Returns `None` once the entry has been evicted; callers reconstruct
    /// full contents from `checkpoint_content` and the transaction and
    /// effects stores instead.
    pub fn get_full_checkpoint_contents_by_digest(
        &self,
        digest: &CheckpointContentsDigest,
    ) -> Option<Arc<FullCheckpointContents>> {
        self.full_checkpoint_contents_cache.get_by_digest(digest)
    }

    pub fn get_latest_certified_checkpoint(
        &self,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        Ok(self
            .tables
            .certified_checkpoints
            .safe_range_iter_reversed(..)
            .next()
            .transpose()?
            .map(|(_, v)| v.into()))
    }

    pub fn get_latest_locally_computed_checkpoint(
        &self,
    ) -> Result<Option<CheckpointSummary>, TypedStoreError> {
        Ok(self
            .tables
            .locally_computed_checkpoints
            .safe_range_iter_reversed(..)
            .next()
            .transpose()?
            .map(|(_, v)| v))
    }

    pub fn multi_get_checkpoint_by_sequence_number(
        &self,
        sequence_numbers: &[CheckpointSequenceNumber],
    ) -> Result<Vec<Option<VerifiedCheckpoint>>, TypedStoreError> {
        let checkpoints = self
            .tables
            .certified_checkpoints
            .multi_get(sequence_numbers)?
            .into_iter()
            .map(|maybe_checkpoint| maybe_checkpoint.map(|c| c.into()))
            .collect();

        Ok(checkpoints)
    }

    pub fn multi_get_checkpoint_content(
        &self,
        contents_digest: &[CheckpointContentsDigest],
    ) -> Result<Vec<Option<CheckpointContents>>, TypedStoreError> {
        self.tables.checkpoint_content.multi_get(contents_digest)
    }

    pub fn get_highest_verified_checkpoint(
        &self,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        let highest_verified = if let Some(highest_verified) = self
            .tables
            .watermarks
            .get(&CheckpointWatermark::HighestVerified)?
        {
            highest_verified
        } else {
            return Ok(None);
        };
        self.get_checkpoint_by_digest(&highest_verified.1)
    }

    pub fn get_highest_synced_checkpoint(
        &self,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        let highest_synced = if let Some(highest_synced) = self
            .tables
            .watermarks
            .get(&CheckpointWatermark::HighestSynced)?
        {
            highest_synced
        } else {
            return Ok(None);
        };
        self.get_checkpoint_by_digest(&highest_synced.1)
    }

    pub fn get_highest_synced_checkpoint_seq_number(
        &self,
    ) -> Result<Option<CheckpointSequenceNumber>, TypedStoreError> {
        if let Some(highest_synced) = self
            .tables
            .watermarks
            .get(&CheckpointWatermark::HighestSynced)?
        {
            Ok(Some(highest_synced.0))
        } else {
            Ok(None)
        }
    }

    pub fn get_highest_executed_checkpoint_seq_number(
        &self,
    ) -> Result<Option<CheckpointSequenceNumber>, TypedStoreError> {
        if let Some(highest_executed) = self
            .tables
            .watermarks
            .get(&CheckpointWatermark::HighestExecuted)?
        {
            Ok(Some(highest_executed.0))
        } else {
            Ok(None)
        }
    }

    pub fn get_highest_executed_checkpoint(
        &self,
    ) -> Result<Option<VerifiedCheckpoint>, TypedStoreError> {
        let highest_executed = if let Some(highest_executed) = self
            .tables
            .watermarks
            .get(&CheckpointWatermark::HighestExecuted)?
        {
            highest_executed
        } else {
            return Ok(None);
        };
        self.get_checkpoint_by_digest(&highest_executed.1)
    }

    pub fn get_highest_pruned_checkpoint_seq_number(
        &self,
    ) -> Result<Option<CheckpointSequenceNumber>, TypedStoreError> {
        self.tables
            .watermarks
            .get(&CheckpointWatermark::HighestPruned)
            .map(|watermark| watermark.map(|w| w.0))
    }

    pub fn get_checkpoint_contents(
        &self,
        digest: &CheckpointContentsDigest,
    ) -> Result<Option<CheckpointContents>, TypedStoreError> {
        self.tables.checkpoint_content.get(digest)
    }

    /// Get full checkpoint contents from the in-memory contents cache.
    ///
    /// Returns `None` once the entry has been evicted; callers fall back to
    /// loading the individual transactions and effects from their stores.
    pub fn get_full_checkpoint_contents_by_sequence_number(
        &self,
        seq: CheckpointSequenceNumber,
    ) -> Option<Arc<FullCheckpointContents>> {
        self.full_checkpoint_contents_cache.get_by_seq(seq)
    }

    fn prune_local_summaries(&self) -> IotaResult {
        if let Some((last_local_summary, _)) = self
            .tables
            .locally_computed_checkpoints
            .safe_range_iter_reversed(..)
            .next()
            .transpose()?
        {
            let mut batch = self.tables.locally_computed_checkpoints.batch();
            batch.schedule_delete_range(
                &self.tables.locally_computed_checkpoints,
                &0,
                &last_local_summary,
            )?;
            batch.write()?;
            info!("Pruned local summaries up to {:?}", last_local_summary);
        }
        Ok(())
    }

    #[instrument(level = "trace", skip_all)]
    fn check_for_checkpoint_fork(
        &self,
        local_checkpoint: &CheckpointSummary,
        verified_checkpoint: &VerifiedCheckpoint,
    ) {
        if local_checkpoint != verified_checkpoint.data() {
            let verified_contents = self
                .get_checkpoint_contents(&verified_checkpoint.contents_digest)
                .map(|opt_contents| {
                    opt_contents
                        .map(|contents| format!("{contents:?}"))
                        .unwrap_or_else(|| {
                            format!(
                                "Verified checkpoint contents not found, digest: {:?}",
                                verified_checkpoint.contents_digest,
                            )
                        })
                })
                .map_err(|e| {
                    format!(
                        "Failed to get verified checkpoint contents, digest: {:?} error: {:?}",
                        verified_checkpoint.contents_digest, e
                    )
                })
                .unwrap_or_else(|err_msg| err_msg);

            let local_contents = self
                .get_checkpoint_contents(&local_checkpoint.contents_digest)
                .map(|opt_contents| {
                    opt_contents
                        .map(|contents| format!("{contents:?}"))
                        .unwrap_or_else(|| {
                            format!(
                                "Local checkpoint contents not found, digest: {:?}",
                                local_checkpoint.contents_digest
                            )
                        })
                })
                .map_err(|e| {
                    format!(
                        "Failed to get local checkpoint contents, digest: {:?} error: {:?}",
                        local_checkpoint.contents_digest, e
                    )
                })
                .unwrap_or_else(|err_msg| err_msg);

            // checkpoint contents may be too large for panic message.
            error!(
                verified_checkpoint = ?verified_checkpoint.data(),
                ?verified_contents,
                ?local_checkpoint,
                ?local_contents,
                "Local checkpoint fork detected!",
            );
            fatal!(
                "Local checkpoint fork detected for sequence number: {}",
                local_checkpoint.sequence_number()
            );
        }
    }

    // Called by consensus (ConsensusAggregator).
    // Different from `insert_verified_checkpoint`, it does not touch
    // the highest_verified_checkpoint watermark such that state sync
    // will have a chance to process this checkpoint and perform some
    // state-sync only things.
    pub fn insert_certified_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        debug!(
            checkpoint_seq = checkpoint.sequence_number(),
            "Inserting certified checkpoint",
        );
        let mut batch = self.tables.certified_checkpoints.batch();
        batch
            .insert_batch(
                &self.tables.certified_checkpoints,
                [(checkpoint.sequence_number(), checkpoint.serializable_ref())],
            )?
            .insert_batch(
                &self.tables.checkpoint_by_digest,
                [(checkpoint.digest(), checkpoint.serializable_ref())],
            )?;
        if checkpoint.next_epoch_committee().is_some() {
            batch.insert_batch(
                &self.tables.epoch_last_checkpoint_map,
                [(&checkpoint.epoch(), checkpoint.sequence_number())],
            )?;
        }
        batch.write()?;

        if let Some(local_checkpoint) = self
            .tables
            .locally_computed_checkpoints
            .get(&checkpoint.sequence_number())?
        {
            self.check_for_checkpoint_fork(&local_checkpoint, checkpoint);
        }

        Ok(())
    }

    // Called by state sync, apart from inserting the checkpoint and updating
    // related tables, it also bumps the highest_verified_checkpoint watermark.
    #[instrument(level = "trace", skip_all)]
    pub fn insert_verified_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        self.insert_certified_checkpoint(checkpoint)?;
        self.update_highest_verified_checkpoint(checkpoint)
    }

    pub fn update_highest_verified_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        if Some(checkpoint.sequence_number())
            > self
                .get_highest_verified_checkpoint()?
                .map(|x| x.sequence_number())
        {
            debug!(
                checkpoint_seq = checkpoint.sequence_number(),
                "Updating highest verified checkpoint",
            );
            self.tables.watermarks.insert(
                &CheckpointWatermark::HighestVerified,
                &(checkpoint.sequence_number(), *checkpoint.digest()),
            )?;
        }

        Ok(())
    }

    pub fn update_highest_synced_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        let seq = checkpoint.sequence_number();
        debug!(checkpoint_seq = seq, "Updating highest synced checkpoint",);
        self.tables.watermarks.insert(
            &CheckpointWatermark::HighestSynced,
            &(seq, *checkpoint.digest()),
        )?;
        self.synced_checkpoint_notify_read.notify(&seq, checkpoint);
        Ok(())
    }

    async fn notify_read_checkpoint_watermark<F>(
        &self,
        notify_read: &NotifyRead<CheckpointSequenceNumber, VerifiedCheckpoint>,
        seq: CheckpointSequenceNumber,
        get_watermark: F,
    ) -> VerifiedCheckpoint
    where
        F: Fn() -> Option<CheckpointSequenceNumber>,
    {
        type ReadResult = Result<Vec<Option<VerifiedCheckpoint>>, TypedStoreError>;

        notify_read
            .read("notify_read_checkpoint_watermark", &[seq], |seqs| {
                let seq = seqs[0];
                let Some(highest) = get_watermark() else {
                    return Ok(vec![None]) as ReadResult;
                };
                if highest < seq {
                    return Ok(vec![None]) as ReadResult;
                }
                let checkpoint = self
                    .get_checkpoint_by_sequence_number(seq)
                    .expect("db error")
                    .expect("checkpoint not found");
                Ok(vec![Some(checkpoint)]) as ReadResult
            })
            .await
            .unwrap()
            .into_iter()
            .next()
            .unwrap()
    }

    pub async fn notify_read_synced_checkpoint(
        &self,
        seq: CheckpointSequenceNumber,
    ) -> VerifiedCheckpoint {
        self.notify_read_checkpoint_watermark(&self.synced_checkpoint_notify_read, seq, || {
            self.get_highest_synced_checkpoint_seq_number()
                .expect("db error")
        })
        .await
    }

    pub async fn notify_read_executed_checkpoint(
        &self,
        seq: CheckpointSequenceNumber,
    ) -> VerifiedCheckpoint {
        self.notify_read_checkpoint_watermark(&self.executed_checkpoint_notify_read, seq, || {
            self.get_highest_executed_checkpoint_seq_number()
                .expect("db error")
        })
        .await
    }

    pub fn update_highest_executed_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        if let Some(seq_number) = self.get_highest_executed_checkpoint_seq_number()? {
            if seq_number >= checkpoint.sequence_number() {
                return Ok(());
            }
            assert_eq!(
                seq_number + 1,
                checkpoint.sequence_number(),
                "Cannot update highest executed checkpoint to {} when current highest executed checkpoint is {}",
                checkpoint.sequence_number(),
                seq_number
            );
        }
        let seq = checkpoint.sequence_number();
        debug!(checkpoint_seq = seq, "Updating highest executed checkpoint",);
        self.tables.watermarks.insert(
            &CheckpointWatermark::HighestExecuted,
            &(seq, *checkpoint.digest()),
        )?;
        self.executed_checkpoint_notify_read
            .notify(&seq, checkpoint);
        Ok(())
    }

    pub fn update_highest_pruned_checkpoint(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        self.tables.watermarks.insert(
            &CheckpointWatermark::HighestPruned,
            &(checkpoint.sequence_number(), *checkpoint.digest()),
        )
    }

    /// Sets highest executed checkpoint to any value.
    ///
    /// WARNING: This method is very subtle and can corrupt the database if used
    /// incorrectly. It should only be used in one-off cases or tests after
    /// fully understanding the risk.
    pub fn set_highest_executed_checkpoint_subtle(
        &self,
        checkpoint: &VerifiedCheckpoint,
    ) -> Result<(), TypedStoreError> {
        self.tables.watermarks.insert(
            &CheckpointWatermark::HighestExecuted,
            &(checkpoint.sequence_number(), *checkpoint.digest()),
        )
    }

    pub fn insert_checkpoint_contents(
        &self,
        contents: CheckpointContents,
    ) -> Result<(), TypedStoreError> {
        debug!(
            checkpoint_seq = ?contents.digest(),
            "Inserting checkpoint contents",
        );
        self.tables
            .checkpoint_content
            .insert(&contents.digest(), &contents)
    }

    /// Persists the checkpoint contents in digest form and caches the full
    /// contents in memory, where they serve the checkpoint executor's bulk
    /// transaction loads and contents requests from state-sync peers.
    ///
    /// INVARIANT: See [`Self::cache_full_checkpoint_contents`].
    pub fn insert_verified_checkpoint_contents(
        &self,
        checkpoint: &VerifiedCheckpoint,
        full_contents: VerifiedCheckpointContents,
    ) -> Result<(), TypedStoreError> {
        let full_contents = full_contents.into_inner();
        let contents = full_contents.checkpoint_contents();
        assert_eq!(checkpoint.contents_digest, contents.digest());

        self.tables
            .checkpoint_content
            .insert(&contents.digest(), &contents)?;

        self.cache_full_checkpoint_contents(
            checkpoint.sequence_number(),
            checkpoint.contents_digest,
            full_contents,
        );
        Ok(())
    }

    /// Whether [`Self::cache_full_checkpoint_contents`] would retain contents
    /// for this sequence number, so callers can skip assembling contents that
    /// the cache would reject (disabled cache, or an entry the lowest-seq
    /// eviction would remove immediately).
    pub fn should_cache_full_checkpoint_contents(&self, seq: CheckpointSequenceNumber) -> bool {
        self.full_checkpoint_contents_cache.should_cache(seq)
    }

    /// Caches full checkpoint contents in memory without writing anything to
    /// disk, so state-sync peers can be served without reconstructing the
    /// contents. `contents_digest` must be the digest of `full_contents`.
    ///
    /// INVARIANT: The caller must have durably written the matching
    /// `checkpoint_content` row (and the contained transactions and
    /// effects) first for two reasons:
    /// 1. once the cache evicts the entry (or after a restart), readers
    ///    reconstruct the full contents from those stores.
    /// 2. state-sync treats available contents as proof of that row and skips
    ///    its own durable write, and the checkpoint executor panics on a
    ///    missing row.
    ///
    /// Best-effort: a serialization failure is logged and the insert skipped;
    /// readers fall back to reconstructing the contents from the durable
    /// stores.
    pub fn cache_full_checkpoint_contents(
        &self,
        sequence_number: CheckpointSequenceNumber,
        contents_digest: CheckpointContentsDigest,
        full_contents: FullCheckpointContents,
    ) {
        let size = match bcs::serialized_size(&full_contents) {
            Ok(size) => size,
            Err(e) => {
                warn!(
                    sequence_number,
                    "failed to serialize full checkpoint contents for caching: {e}"
                );
                return;
            }
        };
        self.full_checkpoint_contents_cache.insert(
            sequence_number,
            contents_digest,
            Arc::new(full_contents),
            size,
        );
    }

    pub fn get_epoch_last_checkpoint(
        &self,
        epoch_id: EpochId,
    ) -> IotaResult<Option<VerifiedCheckpoint>> {
        let seq = self.get_epoch_last_checkpoint_seq_number(epoch_id)?;
        let checkpoint = match seq {
            Some(seq) => self.get_checkpoint_by_sequence_number(seq)?,
            None => None,
        };
        Ok(checkpoint)
    }

    /// Sequence number of `epoch_id`'s last checkpoint. Unlike
    /// [`Self::get_epoch_last_checkpoint`], this does not require the summary
    /// itself to still be present: the underlying map is never pruned.
    pub fn get_epoch_last_checkpoint_seq_number(
        &self,
        epoch_id: EpochId,
    ) -> Result<Option<CheckpointSequenceNumber>, TypedStoreError> {
        self.tables.epoch_last_checkpoint_map.get(&epoch_id)
    }

    pub fn insert_epoch_last_checkpoint(
        &self,
        epoch_id: EpochId,
        checkpoint: &VerifiedCheckpoint,
    ) -> IotaResult {
        self.tables
            .epoch_last_checkpoint_map
            .insert(&epoch_id, &checkpoint.sequence_number())?;
        Ok(())
    }

    pub fn get_epoch_state_commitments(
        &self,
        epoch: EpochId,
    ) -> IotaResult<Option<Vec<CheckpointCommitment>>> {
        let commitments = self.get_epoch_last_checkpoint(epoch)?.map(|checkpoint| {
            checkpoint
                .end_of_epoch_data
                .as_ref()
                .expect("Last checkpoint of epoch expected to have EndOfEpochData")
                .epoch_commitments
                .clone()
        });
        Ok(commitments)
    }

    /// Given the epoch ID, and the last checkpoint of the epoch, derive a few
    /// statistics of the epoch.
    pub fn get_epoch_stats(
        &self,
        epoch: EpochId,
        last_checkpoint: &CheckpointSummary,
    ) -> Option<EpochStats> {
        let (first_checkpoint, prev_epoch_network_transactions) = if epoch == 0 {
            (0, 0)
        } else if let Ok(Some(checkpoint)) = self.get_epoch_last_checkpoint(epoch - 1) {
            (
                checkpoint.sequence_number + 1,
                checkpoint.network_total_transactions,
            )
        } else {
            return None;
        };
        Some(EpochStats {
            checkpoint_count: last_checkpoint.sequence_number - first_checkpoint + 1,
            transaction_count: last_checkpoint.network_total_transactions
                - prev_epoch_network_transactions,
            total_gas_reward: last_checkpoint
                .epoch_rolling_gas_cost_summary
                .computation_cost,
        })
    }
}

#[derive(Copy, Clone, Debug, Serialize, Deserialize)]
pub enum CheckpointWatermark {
    HighestVerified,
    HighestSynced,
    HighestExecuted,
    HighestPruned,
}

struct CheckpointStateHasher {
    epoch_store: Arc<AuthorityPerEpochStore>,
    hasher: Weak<GlobalStateHasher>,
    receive_from_builder: mpsc::Receiver<(CheckpointSequenceNumber, Vec<TransactionEffects>)>,
}

impl CheckpointStateHasher {
    fn new(
        epoch_store: Arc<AuthorityPerEpochStore>,
        hasher: Weak<GlobalStateHasher>,
        receive_from_builder: mpsc::Receiver<(CheckpointSequenceNumber, Vec<TransactionEffects>)>,
    ) -> Self {
        Self {
            epoch_store,
            hasher,
            receive_from_builder,
        }
    }

    async fn run(self) {
        let Self {
            epoch_store,
            hasher,
            mut receive_from_builder,
        } = self;
        while let Some((seq, effects)) = receive_from_builder.recv().await {
            let Some(hasher) = hasher.upgrade() else {
                info!("GlobalStateHash was dropped, stopping checkpoint accumulation");
                break;
            };
            hasher
                .accumulate_checkpoint(&effects, seq, &epoch_store)
                .expect("epoch ended while accumulating checkpoint");
        }
    }
}

/// A checkpoint produced by [`CheckpointBuilder::create_checkpoints`],
/// written and cached by [`CheckpointBuilder::write_checkpoints`].
struct BuiltCheckpoint {
    summary: CheckpointSummary,
    contents: CheckpointContents,
    /// Full contents for the in-memory cache, assembled only when the cache
    /// would retain them.
    full_contents: Option<FullCheckpointContents>,
}

#[derive(Debug)]
pub enum CheckpointBuilderError {
    ChangeEpochTxAlreadyExecuted,
    SystemPackagesMissing,
    Retry(anyhow::Error),
}

impl<IotaError: std::error::Error + Send + Sync + 'static> From<IotaError>
    for CheckpointBuilderError
{
    fn from(e: IotaError) -> Self {
        Self::Retry(e.into())
    }
}

pub type CheckpointBuilderResult<T = ()> = Result<T, CheckpointBuilderError>;

pub struct CheckpointBuilder {
    state: Arc<AuthorityState>,
    store: Arc<CheckpointStore>,
    epoch_store: Arc<AuthorityPerEpochStore>,
    notify: Arc<Notify>,
    notify_aggregator: Arc<Notify>,
    last_built: watch::Sender<CheckpointSequenceNumber>,
    effects_store: Arc<dyn TransactionCacheRead>,
    global_state_hasher: Weak<GlobalStateHasher>,
    send_to_hasher: mpsc::Sender<(CheckpointSequenceNumber, Vec<TransactionEffects>)>,
    output: Box<dyn CheckpointOutput>,
    metrics: Arc<CheckpointMetrics>,
    max_transactions_per_checkpoint: usize,
    max_checkpoint_size_bytes: usize,
}

pub struct CheckpointAggregator {
    store: Arc<CheckpointStore>,
    epoch_store: Arc<AuthorityPerEpochStore>,
    notify: Arc<Notify>,
    current: Option<CheckpointSignatureAggregator>,
    output: Box<dyn CertifiedCheckpointOutput>,
    state: Arc<AuthorityState>,
    metrics: Arc<CheckpointMetrics>,
}

// This holds information to aggregate signatures for one checkpoint
pub struct CheckpointSignatureAggregator {
    next_index: u64,
    summary: CheckpointSummary,
    digest: CheckpointDigest,
    /// Aggregates voting stake for each signed checkpoint proposal by authority
    signatures_by_digest: MultiStakeAggregator<CheckpointDigest, CheckpointSummary, true>,
    store: Arc<CheckpointStore>,
    state: Arc<AuthorityState>,
    metrics: Arc<CheckpointMetrics>,
}

impl CheckpointBuilder {
    fn new(
        state: Arc<AuthorityState>,
        store: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        notify: Arc<Notify>,
        effects_store: Arc<dyn TransactionCacheRead>,
        // for synchronous accumulation of end-of-epoch checkpoint
        global_state_hasher: Weak<GlobalStateHasher>,
        // for asynchronous/concurrent accumulation of regular checkpoints
        send_to_hasher: mpsc::Sender<(CheckpointSequenceNumber, Vec<TransactionEffects>)>,
        output: Box<dyn CheckpointOutput>,
        notify_aggregator: Arc<Notify>,
        last_built: watch::Sender<CheckpointSequenceNumber>,
        metrics: Arc<CheckpointMetrics>,
        max_transactions_per_checkpoint: usize,
        max_checkpoint_size_bytes: usize,
    ) -> Self {
        Self {
            state,
            store,
            epoch_store,
            notify,
            effects_store,
            global_state_hasher,
            send_to_hasher,
            output,
            notify_aggregator,
            last_built,
            metrics,
            max_transactions_per_checkpoint,
            max_checkpoint_size_bytes,
        }
    }

    /// This function first waits for ConsensusCommitHandler to finish
    /// reprocessing commits that have been processed before the last
    /// restart, if consensus_replay_waiter is supplied. Then it starts
    /// building checkpoints in a loop.
    ///
    /// It is optional to pass in consensus_replay_waiter, to make it easier to
    /// attribute if slow recovery of previously built checkpoints is due to
    /// consensus replay or checkpoint building.
    async fn run(mut self, consensus_replay_waiter: Option<ReplayWaiter>) {
        if let Some(replay_waiter) = consensus_replay_waiter {
            info!("Waiting for consensus commits to replay ...");
            replay_waiter.wait_for_replay().await;
            info!("Consensus commits finished replaying");
        }
        info!("Starting CheckpointBuilder");
        loop {
            match self.maybe_build_checkpoints().await {
                Ok(()) => {}
                err @ Err(
                    CheckpointBuilderError::ChangeEpochTxAlreadyExecuted
                    | CheckpointBuilderError::SystemPackagesMissing,
                ) => {
                    info!("CheckpointBuilder stopping: {:?}", err);
                    return;
                }
                Err(CheckpointBuilderError::Retry(inner)) => {
                    let msg = format!("{inner:?}");
                    debug_fatal!("Error while making checkpoint, will retry in 1s: {}", msg);
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    self.metrics.checkpoint_errors.inc();
                    continue;
                }
            }

            self.notify.notified().await;
        }
    }

    async fn maybe_build_checkpoints(&mut self) -> CheckpointBuilderResult {
        let _scope = monitored_scope("BuildCheckpoints");

        // Collect info about the most recently built checkpoint.
        let summary = self
            .epoch_store
            .last_built_checkpoint_builder_summary()
            .expect("epoch should not have ended");
        let mut last_height = summary.as_ref().and_then(|s| s.checkpoint_height);
        let mut last_timestamp = summary.as_ref().map(|s| s.summary.timestamp_ms);
        let mut last_seq = summary.map(|s| s.summary.sequence_number);

        let min_checkpoint_interval_ms = self
            .epoch_store
            .protocol_config()
            .min_checkpoint_interval_ms_as_option()
            .unwrap_or_default();
        // When set, the interval may additionally be amortized over this many
        // recent checkpoints of the current epoch.
        let checkpoint_rate_window_size = self
            .epoch_store
            .protocol_config()
            .checkpoint_rate_window_size_as_option();
        let mut grouped_pending_checkpoints = Vec::new();
        let mut checkpoints_iter = self
            .epoch_store
            .get_pending_checkpoints(last_height)
            .expect("unexpected epoch store error")
            .into_iter()
            .peekable();
        while let Some((height, pending)) = checkpoints_iter.next() {
            let current_timestamp = pending.details().timestamp_ms;
            // Strict interval against the immediately preceding checkpoint.
            let adjacent_interval_elapsed = match last_timestamp {
                Some(last_timestamp) => {
                    current_timestamp >= last_timestamp + min_checkpoint_interval_ms
                }
                None => true,
            };
            // Windowed arm: also allow building once the checkpoint `window`
            // back is at least `window * interval` older. This recycles the
            // slack the strict arm loses to discrete commit timestamps,
            // holding the sustained rate at the ceiling, while the strict arm
            // keeps every quiet gap — and thus checkpoint sizes — within one
            // interval. Only checkpoints built in the current epoch are
            // consulted: every validator builds all of them, so the look-back
            // resolves identically everywhere and the gate stays
            // deterministic. Until the window fills (epoch start, chain
            // genesis) the arm is inert.
            let interval_elapsed = adjacent_interval_elapsed
                || checkpoint_rate_window_size.is_some_and(|window| {
                    last_seq
                        .and_then(|seq| (seq + 1).checked_sub(window))
                        .and_then(|window_start_seq| {
                            self.epoch_store
                                .get_built_checkpoint_summary(window_start_seq)
                                .expect("epoch store should not error reading a built checkpoint summary")
                        })
                        .is_some_and(|window_start| {
                            current_timestamp
                                >= window_start.timestamp_ms + window * min_checkpoint_interval_ms
                        })
                });
            // Group PendingCheckpoints until:
            // - the minimum interval has elapsed ...
            let can_build = interval_elapsed
                // - or, next PendingCheckpoint is last-of-epoch (since the last-of-epoch checkpoint
                //   should be written separately) ...
                || checkpoints_iter
                .peek()
                .is_some_and(|(_, next_pending)| next_pending.details().last_of_epoch)
                // - or, we have reached end of epoch.
                || pending.details().last_of_epoch;
            grouped_pending_checkpoints.push(pending);
            if !can_build {
                debug!(
                    checkpoint_commit_height = height,
                    ?last_timestamp,
                    ?current_timestamp,
                    "waiting for more PendingCheckpoints: minimum interval not yet elapsed"
                );
                continue;
            }

            // Min interval has elapsed, we can now coalesce and build a checkpoint.
            last_height = Some(height);
            last_timestamp = Some(current_timestamp);
            let commits_in_checkpoint = grouped_pending_checkpoints.len();
            debug!(
                checkpoint_commit_height_from = grouped_pending_checkpoints
                    .first()
                    .unwrap()
                    .details()
                    .checkpoint_height,
                checkpoint_commit_height_to = last_height,
                "Making checkpoint with commit height range"
            );

            let seq = self
                .make_checkpoint(std::mem::take(&mut grouped_pending_checkpoints))
                .await?;

            // Count only on success; a failed build retries the same
            // group and would otherwise double-count it.
            self.metrics
                .commits_per_checkpoint
                .observe(commits_in_checkpoint as f64);
            // Advance the window anchor to the highest checkpoint just
            // built (a single call may emit several when chunked).
            last_seq = Some(seq);
            self.last_built.send_if_modified(|cur| {
                // when rebuilding checkpoints at startup, seq can be for an old checkpoint
                if seq > *cur {
                    *cur = seq;
                    true
                } else {
                    false
                }
            });

            // Ensure that the task can be cancelled at end of epoch, even if no other await
            // yields execution.
            tokio::task::yield_now().await;
        }
        debug!(
            "Waiting for more checkpoints from consensus after processing {last_height:?}; {} pending checkpoints left unprocessed until next interval",
            grouped_pending_checkpoints.len(),
        );

        Ok(())
    }

    #[instrument(level = "debug", skip_all, fields(last_height = pendings.last().unwrap().details().checkpoint_height
    ))]
    async fn make_checkpoint(
        &self,
        pendings: Vec<PendingCheckpoint>,
    ) -> CheckpointBuilderResult<CheckpointSequenceNumber> {
        let _scope = monitored_scope("CheckpointBuilder::make_checkpoint");
        let last_details = pendings.last().unwrap().details().clone();

        // Stores the transactions that should be included in the checkpoint.
        // Transactions will be recorded in the checkpoint in this order.
        let highest_executed_sequence = self
            .store
            .get_highest_executed_checkpoint_seq_number()
            .expect("db error")
            .unwrap_or(0);

        let (poll_count, result) = poll_count(self.resolve_checkpoint_transactions(pendings)).await;
        let (sorted_tx_effects_included_in_checkpoint, all_roots) = result?;

        let new_checkpoints = self
            .create_checkpoints(
                sorted_tx_effects_included_in_checkpoint,
                &last_details,
                &all_roots,
            )
            .await?;
        let highest_sequence = new_checkpoints.last().summary.sequence_number();
        if highest_sequence <= highest_executed_sequence && poll_count > 1 {
            debug_fatal!(
                "resolve_checkpoint_transactions should be instantaneous when executed checkpoint is ahead of checkpoint builder"
            );
        }

        self.write_checkpoints(last_details.checkpoint_height, new_checkpoints)
            .await?;
        Ok(highest_sequence)
    }

    // Given the root transactions of the pending checkpoints, resolve the
    // transactions that should be included in the checkpoint, and return them in
    // the order they should be included in the checkpoint, together with all the
    // resolved roots.
    #[instrument(level = "debug", skip_all)]
    async fn resolve_checkpoint_transactions(
        &self,
        pending_checkpoints: Vec<PendingCheckpoint>,
    ) -> IotaResult<(Vec<TransactionEffects>, HashSet<TransactionDigest>)> {
        let _scope = monitored_scope("CheckpointBuilder::resolve_checkpoint_transactions");

        // Keeps track of the effects that are already included in the current
        // checkpoint. This is used when there are multiple pending checkpoints
        // to create a single checkpoint because in such scenarios, dependencies
        // of a transaction may in earlier created checkpoints, or in earlier
        // pending checkpoints.
        let mut effects_in_current_checkpoint = BTreeSet::new();

        let mut tx_effects = Vec::new();
        let mut tx_roots = HashSet::new();

        for pending_checkpoint in pending_checkpoints.into_iter() {
            let pending = pending_checkpoint.into_v1();
            debug!(
                checkpoint_commit_height = pending.details.checkpoint_height,
                roots = ?pending.roots,
                "Resolving checkpoint transactions for pending checkpoint.",
            );

            let roots = &pending.roots;

            self.metrics
                .checkpoint_roots_count
                .inc_by(roots.len() as u64);

            let root_digests = self
                .epoch_store
                .notify_read_executed_digests(roots)
                .in_monitored_scope("CheckpointNotifyDigests")
                .await?;
            let root_effects = self
                .effects_store
                .try_notify_read_executed_effects(
                    CHECKPOINT_BUILDER_NOTIFY_READ_TASK_NAME,
                    &root_digests,
                )
                .in_monitored_scope("CheckpointNotifyRead")
                .await?;

            let consensus_commit_prologue = {
                // If the roots contains consensus commit prologue transaction, we want to
                // extract it, and put it to the front of the checkpoint.

                let consensus_commit_prologue = self
                    .extract_consensus_commit_prologue(&root_digests, &root_effects)
                    .await?;

                // Get the un-included dependencies of the consensus commit prologue. We should
                // expect no other dependencies that haven't been included in
                // any previous checkpoints.
                if let Some((ccp_digest, ccp_effects)) = &consensus_commit_prologue {
                    let unsorted_ccp = self.complete_checkpoint_effects(
                        vec![ccp_effects.clone()],
                        &mut effects_in_current_checkpoint,
                    )?;

                    // No other dependencies of this consensus commit prologue that haven't been
                    // included in any previous checkpoint.
                    if unsorted_ccp.len() != 1 {
                        fatal!(
                            "Expected 1 consensus commit prologue, got {:?}",
                            unsorted_ccp
                                .iter()
                                .map(|e| e.transaction_digest())
                                .collect::<Vec<_>>()
                        );
                    }
                    assert_eq!(unsorted_ccp.len(), 1);
                    assert_eq!(unsorted_ccp[0].transaction_digest(), ccp_digest);
                }
                consensus_commit_prologue
            };

            let unsorted =
                self.complete_checkpoint_effects(root_effects, &mut effects_in_current_checkpoint)?;

            let _scope = monitored_scope("CheckpointBuilder::causal_sort");
            let mut sorted: Vec<TransactionEffects> = Vec::with_capacity(unsorted.len() + 1);
            if let Some((_ccp_digest, ccp_effects)) = consensus_commit_prologue {
                #[cfg(debug_assertions)]
                {
                    // When consensus_commit_prologue is extracted, it should not be included in the
                    // `unsorted`.
                    for tx in unsorted.iter() {
                        assert!(tx.transaction_digest() != &_ccp_digest);
                    }
                }
                sorted.push(ccp_effects);
            }
            sorted.extend(CausalOrder::causal_sort(unsorted));

            #[cfg(msim)]
            {
                // Check consensus commit prologue invariants in sim test.
                self.expensive_consensus_commit_prologue_invariants_check(&root_digests, &sorted);
            }

            tx_effects.extend(sorted);
            tx_roots.extend(root_digests);
        }

        Ok((tx_effects, tx_roots))
    }

    // This function is used to extract the consensus commit prologue digest and
    // effects from the root transactions.
    // The consensus commit prologue is expected to be the first transaction in the
    // roots.
    async fn extract_consensus_commit_prologue(
        &self,
        root_digests: &[TransactionDigest],
        root_effects: &[TransactionEffects],
    ) -> IotaResult<Option<(TransactionDigest, TransactionEffects)>> {
        let _scope = monitored_scope("CheckpointBuilder::extract_consensus_commit_prologue");
        if root_digests.is_empty() {
            return Ok(None);
        }

        // Reads the first transaction in the roots, and checks whether it is a
        // consensus commit prologue transaction.
        // The consensus commit prologue transaction should be the first transaction
        // in the roots written by the consensus handler.
        let first_tx = self
            .state
            .get_transaction_cache_reader()
            .try_get_transaction_block(&root_digests[0])?
            .expect("Transaction block must exist");

        Ok(match first_tx.transaction().kind() {
            TransactionKind::ConsensusCommitPrologueV1(_) => {
                assert_eq!(first_tx.digest(), root_effects[0].transaction_digest());
                Some((*first_tx.digest(), root_effects[0].clone()))
            }
            _ => None,
        })
    }

    /// Writes the new checkpoints to the DB storage and processes them.
    #[instrument(level = "debug", skip_all)]
    async fn write_checkpoints(
        &self,
        height: CheckpointHeight,
        mut new_checkpoints: NonEmpty<BuiltCheckpoint>,
    ) -> IotaResult {
        let _scope = monitored_scope("CheckpointBuilder::write_checkpoints");
        let mut batch = self.store.tables.checkpoint_content.batch();
        let mut all_tx_digests =
            Vec::with_capacity(new_checkpoints.iter().map(|c| c.contents.len()).sum());

        // Write the new checkpoints to the DB storage.
        for BuiltCheckpoint {
            summary, contents, ..
        } in &new_checkpoints
        {
            debug!(
                checkpoint_commit_height = height,
                checkpoint_seq = summary.sequence_number,
                contents_digest = ?contents.digest(),
                "writing checkpoint",
            );

            if let Some(previously_computed_summary) = self
                .store
                .tables
                .locally_computed_checkpoints
                .get(&summary.sequence_number)?
            {
                if previously_computed_summary != *summary {
                    // Panic so that we don't send out an equivocating checkpoint sig.
                    fatal!(
                        "Checkpoint {} was previously built with a different result: {previously_computed_summary:?} vs {summary:?}",
                        summary.sequence_number,
                    );
                }
            }

            all_tx_digests.extend(contents.iter().map(|digests| digests.transaction));

            self.metrics
                .transactions_included_in_checkpoint
                .inc_by(contents.len() as u64);
            let sequence_number = summary.sequence_number;
            self.metrics
                .last_constructed_checkpoint
                .set(sequence_number as i64);

            batch.insert_batch(
                &self.store.tables.checkpoint_content,
                [(contents.digest(), contents)],
            )?;

            batch.insert_batch(
                &self.store.tables.locally_computed_checkpoints,
                [(sequence_number, summary)],
            )?;
        }

        batch.write()?;

        // Cache the full contents only now that the checkpoint_content rows
        // are durable
        for checkpoint in new_checkpoints.iter_mut() {
            if let Some(full_contents) = checkpoint.full_contents.take() {
                self.store.cache_full_checkpoint_contents(
                    checkpoint.summary.sequence_number,
                    checkpoint.summary.contents_digest,
                    full_contents,
                );
            }
        }

        // Send all checkpoint sigs to consensus. The messages including
        // MisbehaviorReports are also sent in this step.
        for BuiltCheckpoint {
            summary, contents, ..
        } in &new_checkpoints
        {
            self.output
                .checkpoint_created(summary, contents, &self.epoch_store, &self.store)
                .await?;
        }

        for BuiltCheckpoint {
            summary: local_checkpoint,
            ..
        } in &new_checkpoints
        {
            if let Some(certified_checkpoint) = self
                .store
                .tables
                .certified_checkpoints
                .get(&local_checkpoint.sequence_number())?
            {
                self.store
                    .check_for_checkpoint_fork(local_checkpoint, &certified_checkpoint.into());
            }
        }

        self.notify_aggregator.notify_one();
        self.epoch_store.process_constructed_checkpoint(
            height,
            new_checkpoints.map(|c| (c.summary, c.contents)),
        );
        Ok(())
    }

    #[expect(clippy::type_complexity)]
    fn split_checkpoint_chunks(
        &self,
        transactions_effects_and_sizes: Vec<(TransactionEnvelope, TransactionEffects, usize)>,
        signatures: Vec<Vec<UserSignature>>,
    ) -> CheckpointBuilderResult<
        Vec<Vec<(TransactionEnvelope, TransactionEffects, Vec<UserSignature>)>>,
    > {
        let _guard = monitored_scope("CheckpointBuilder::split_checkpoint_chunks");
        let mut chunks = Vec::new();
        let mut chunk = Vec::new();
        let mut chunk_size: usize = 0;
        for ((transaction, effects, transaction_size), signatures) in
            transactions_effects_and_sizes.into_iter().zip(signatures)
        {
            // Roll over to a new chunk after either max count or max size is reached.
            // The size calculation here is intended to estimate the size of the
            // FullCheckpointContents struct. If this code is modified, that struct
            // should also be updated accordingly.
            let size = transaction_size
                + bcs::serialized_size(&effects)?
                + bcs::serialized_size(&signatures)?;
            if chunk.len() == self.max_transactions_per_checkpoint
                || (chunk_size + size) > self.max_checkpoint_size_bytes
            {
                if chunk.is_empty() {
                    // Always allow at least one tx in a checkpoint.
                    warn!(
                        "Size of single transaction ({size}) exceeds max checkpoint size ({}); allowing excessively large checkpoint to go through.",
                        self.max_checkpoint_size_bytes
                    );
                } else {
                    chunks.push(chunk);
                    chunk = Vec::new();
                    chunk_size = 0;
                }
            }

            chunk.push((transaction, effects, signatures));
            chunk_size += size;
        }

        if !chunk.is_empty() || chunks.is_empty() {
            // We intentionally create an empty checkpoint if there is no content provided
            // to make a 'heartbeat' checkpoint.
            // Important: if some conditions are added here later, we need to make sure we
            // always have at least one chunk if last_pending_of_epoch is set
            chunks.push(chunk);
            // Note: empty checkpoints are ok - they shouldn't happen at all on
            // a network with even modest load. Even if they do
            // happen, it is still useful as it allows fullnodes to
            // distinguish between "no transactions have happened" and "i am not
            // receiving new checkpoints".
        }
        Ok(chunks)
    }

    /// Creates checkpoints using the provided transaction effects and pending
    /// checkpoint information.
    fn load_last_built_checkpoint_summary(
        epoch_store: &AuthorityPerEpochStore,
        store: &CheckpointStore,
    ) -> IotaResult<Option<(CheckpointSequenceNumber, CheckpointSummary)>> {
        let mut last_checkpoint = epoch_store.last_built_checkpoint_summary()?;
        if last_checkpoint.is_none() {
            let epoch = epoch_store.epoch();
            if epoch > 0 {
                let previous_epoch = epoch - 1;
                let last_verified = store.get_epoch_last_checkpoint(previous_epoch)?;
                last_checkpoint = last_verified.map(VerifiedCheckpoint::into_summary_and_sequence);
                if let Some((ref seq, _)) = last_checkpoint {
                    debug!(
                        "No checkpoints in builder DB, taking checkpoint from previous epoch with sequence {seq}"
                    );
                } else {
                    // This is some serious bug with when CheckpointBuilder started so surfacing it
                    // via panic
                    panic!("Can not find last checkpoint for previous epoch {previous_epoch}");
                }
            }
        }
        Ok(last_checkpoint)
    }

    #[instrument(level = "debug", skip_all)]
    async fn create_checkpoints(
        &self,
        all_effects: Vec<TransactionEffects>,
        details: &PendingCheckpointInfo,
        all_roots: &HashSet<TransactionDigest>,
    ) -> CheckpointBuilderResult<NonEmpty<BuiltCheckpoint>> {
        let _scope = monitored_scope("CheckpointBuilder::create_checkpoints");

        let total = all_effects.len();
        let mut last_checkpoint =
            Self::load_last_built_checkpoint_summary(&self.epoch_store, &self.store)?;
        let last_checkpoint_seq = last_checkpoint.as_ref().map(|(seq, _)| *seq);
        debug!(
            checkpoint_commit_height = details.checkpoint_height,
            next_checkpoint_seq = last_checkpoint_seq.unwrap_or_default() + 1,
            checkpoint_timestamp = details.timestamp_ms,
            "Creating checkpoint(s) for {} transactions",
            all_effects.len(),
        );

        let all_digests: Vec<_> = all_effects
            .iter()
            .map(|effect| *effect.transaction_digest())
            .collect();
        let transactions_and_sizes = self
            .state
            .get_transaction_cache_reader()
            .try_get_transactions_and_serialized_sizes(&all_digests)?;
        let mut all_effects_and_transaction_sizes = Vec::with_capacity(all_effects.len());
        let mut transactions = Vec::with_capacity(all_effects.len());
        let mut transaction_keys = Vec::with_capacity(all_effects.len());
        let mut randomness_rounds = BTreeMap::new();
        {
            let _guard = monitored_scope("CheckpointBuilder::wait_for_transactions_sequenced");
            debug!(
                ?last_checkpoint_seq,
                "Waiting for {:?} certificates to appear in consensus",
                all_effects.len()
            );

            for (effects, transaction_and_size) in all_effects
                .into_iter()
                .zip(transactions_and_sizes.into_iter())
            {
                let (transaction, size) = transaction_and_size
                    .unwrap_or_else(|| panic!("Could not find executed transaction {effects:?}"));
                match transaction.inner().transaction().kind() {
                    #[allow(deprecated)]
                    TransactionKind::ConsensusCommitPrologueV1(_)
                    | TransactionKind::AuthenticatorStateUpdateV1Deprecated => {
                        // ConsensusCommitPrologue is guaranteed to be
                        // processed before we reach here.
                        //
                        // Deprecated: Authenticator state (JWK) is deprecated
                        // and was never enabled.
                        // These transaction kinds are retained
                        // only for BCS enum variant compatibility.
                    }
                    TransactionKind::RandomnessStateUpdate(rsu) => {
                        randomness_rounds
                            .insert(*effects.transaction_digest(), rsu.randomness_round);
                    }
                    _ => {
                        // Only transactions that are not roots should be included in the call to
                        // `consensus_messages_processed_notify`. roots come directly from the
                        // consensus commit and so are known to be processed
                        // already.
                        let digest = *effects.transaction_digest();
                        if !all_roots.contains(&digest) {
                            // In P-COOL flow, user transactions are submitted as
                            // UserTransaction, not as Certificate.
                            let key = if self.epoch_store.protocol_config().enable_pcool_flow() {
                                ConsensusTransactionKey::UserTransaction(digest)
                            } else {
                                ConsensusTransactionKey::Certificate(digest)
                            };
                            transaction_keys.push(SequencedConsensusTransactionKey::External(key));
                        }
                    }
                }
                transactions.push(transaction);
                all_effects_and_transaction_sizes.push((effects, size));
            }

            self.epoch_store
                .consensus_messages_processed_notify(transaction_keys)
                .await?;
        }

        let signatures = self
            .epoch_store
            .user_signatures_for_checkpoint(&transactions, &all_digests)?;
        debug!(
            ?last_checkpoint_seq,
            "Received {} checkpoint user signatures from consensus",
            signatures.len()
        );

        let transactions_effects_and_sizes = transactions
            .into_iter()
            .zip(all_effects_and_transaction_sizes)
            .map(|(transaction, (effects, size))| (transaction.into_inner(), effects, size))
            .collect();
        let chunks = self.split_checkpoint_chunks(transactions_effects_and_sizes, signatures)?;
        let chunks_count = chunks.len();

        let mut checkpoints = Vec::with_capacity(chunks_count);
        debug!(
            ?last_checkpoint_seq,
            "Creating {} checkpoints with {} transactions", chunks_count, total,
        );

        let epoch = self.epoch_store.epoch();
        for (index, chunk) in chunks.into_iter().enumerate() {
            let first_checkpoint_of_epoch = index == 0
                && last_checkpoint
                    .as_ref()
                    .map(|(_, c)| c.epoch != epoch)
                    .unwrap_or(true);
            if first_checkpoint_of_epoch {
                self.epoch_store
                    .record_epoch_first_checkpoint_creation_time_metric();
            }
            let last_checkpoint_of_epoch = details.last_of_epoch && index == chunks_count - 1;

            let sequence_number = last_checkpoint
                .as_ref()
                .map(|(_, c)| c.sequence_number + 1)
                .unwrap_or_default();
            let mut timestamp_ms = details.timestamp_ms;
            if let Some((_, last_checkpoint)) = &last_checkpoint {
                if last_checkpoint.timestamp_ms > timestamp_ms {
                    // The first consensus commit of an epoch can have zero timestamp.
                    debug!(
                        "Decrease of checkpoint timestamp, possibly due to epoch change. Sequence: {}, previous: {}, current: {}",
                        sequence_number, last_checkpoint.timestamp_ms, timestamp_ms,
                    );
                    timestamp_ms = last_checkpoint.timestamp_ms;
                }
            }

            let (chunk_transactions, mut effects, mut signatures): (
                Vec<TransactionEnvelope>,
                Vec<TransactionEffects>,
                Vec<Vec<UserSignature>>,
            ) = chunk.into_iter().multiunzip();
            let epoch_rolling_gas_cost_summary =
                self.get_epoch_total_gas_cost(last_checkpoint.as_ref().map(|(_, c)| c), &effects);

            let end_of_epoch_data = if last_checkpoint_of_epoch {
                let scores: Vec<u64> = if self
                    .epoch_store
                    .protocol_config()
                    .pass_calculated_validator_scores_to_advance_epoch()
                {
                    self.epoch_store.scoreboard.current_scores()
                } else {
                    // Give everyone in the committee the max score
                    vec![MAX_SCORE; self.epoch_store.committee().num_members()]
                };

                let (system_state_obj, system_epoch_info_event) = self
                    .augment_epoch_last_checkpoint(
                        &epoch_rolling_gas_cost_summary,
                        timestamp_ms,
                        &mut effects,
                        &mut signatures,
                        sequence_number,
                        scores,
                    )
                    .await?;

                // The system epoch info event can be `None` in case if the `advance_epoch`
                // Move function call failed and was executed in the safe mode.
                // In this case, the tokens supply should be unchanged.
                //
                // SAFETY: The number of minted and burnt tokens easily fit into an i64 and due
                // to those small numbers, no overflows will occur during conversion or
                // subtraction.
                let epoch_supply_change =
                    system_epoch_info_event.map_or(0, |event| event.supply_change());

                let committee = system_state_obj
                    .get_current_epoch_committee()
                    .committee()
                    .clone();

                // This must happen after the call to augment_epoch_last_checkpoint,
                // otherwise we will not capture the change_epoch tx.
                let root_state_digest = {
                    let state_acc = self
                        .global_state_hasher
                        .upgrade()
                        .expect("No checkpoints should be getting built after local configuration");
                    let acc = state_acc.accumulate_checkpoint(
                        &effects,
                        sequence_number,
                        &self.epoch_store,
                    )?;

                    state_acc
                        .wait_for_previous_running_root(&self.epoch_store, sequence_number)
                        .await?;

                    state_acc.accumulate_running_root(
                        &self.epoch_store,
                        sequence_number,
                        Some(acc),
                    )?;
                    state_acc
                        .digest_epoch(self.epoch_store.clone(), sequence_number)
                        .await?
                };
                self.metrics.highest_accumulated_epoch.set(epoch as i64);
                info!("Epoch {epoch} root state hash digest: {root_state_digest:?}");

                let epoch_commitments = vec![CheckpointCommitment::EcmhLiveObjectSet {
                    digest: root_state_digest.digest,
                }];

                Some(EndOfEpochData {
                    next_epoch_committee: committee.committee_members(),
                    next_epoch_protocol_version: system_state_obj.protocol_version(),
                    epoch_commitments,
                    epoch_supply_change,
                })
            } else {
                self.send_to_hasher
                    .send((sequence_number, effects.clone()))
                    .await?;
                None
            };
            let contents = CheckpointContents::new_with_digests_and_signatures(
                effects.iter().map(TransactionEffects::execution_digests),
                signatures,
            );

            let num_txns = contents.len() as u64;

            let network_total_transactions = last_checkpoint
                .as_ref()
                .map(|(_, c)| c.network_total_transactions + num_txns)
                .unwrap_or(num_txns);

            let previous_digest = last_checkpoint.as_ref().map(|(_, c)| c.digest());

            let matching_randomness_rounds: Vec<_> = effects
                .iter()
                .filter_map(|e| randomness_rounds.get(e.transaction_digest()))
                .copied()
                .collect();

            let summary = CheckpointSummary::new_with_protocol_config(
                self.epoch_store.protocol_config(),
                epoch,
                sequence_number,
                network_total_transactions,
                &contents,
                previous_digest,
                epoch_rolling_gas_cost_summary,
                end_of_epoch_data,
                timestamp_ms,
                matching_randomness_rounds,
            );
            summary.report_checkpoint_age(&self.metrics.last_created_checkpoint_age);
            if last_checkpoint_of_epoch {
                info!(
                    checkpoint_seq = sequence_number,
                    "creating last checkpoint of epoch {}", epoch
                );
                if let Some(stats) = self.store.get_epoch_stats(epoch, &summary) {
                    self.epoch_store
                        .report_epoch_metrics_at_last_checkpoint(stats);
                }
            }

            // Assemble the full contents for faster checkpoint propagation to
            // peers. End-of-epoch checkpoints carry an appended change-epoch
            // transaction not tracked here; the executor caches those via its
            // synced path.
            let full_contents = (!last_checkpoint_of_epoch
                && self
                    .store
                    .should_cache_full_checkpoint_contents(sequence_number))
            .then(|| {
                let execution_data = chunk_transactions
                    .into_iter()
                    .zip(effects.iter().cloned())
                    .map(|(transaction, effects)| ExecutionData::new(transaction, effects));
                FullCheckpointContents::from_contents_and_execution_data(
                    contents.clone(),
                    execution_data,
                )
            });

            last_checkpoint = Some((sequence_number, summary.clone()));
            checkpoints.push(BuiltCheckpoint {
                summary,
                contents,
                full_contents,
            });
        }

        Ok(NonEmpty::from_vec(checkpoints).expect("at least one checkpoint"))
    }

    fn get_epoch_total_gas_cost(
        &self,
        last_checkpoint: Option<&CheckpointSummary>,
        cur_checkpoint_effects: &[TransactionEffects],
    ) -> GasCostSummary {
        let (previous_epoch, previous_gas_costs) = last_checkpoint
            .map(|c| (c.epoch, c.epoch_rolling_gas_cost_summary.clone()))
            .unwrap_or_default();
        let current_gas_costs =
            checked::new_gas_cost_summary_from_txn_effects(cur_checkpoint_effects.iter());
        if previous_epoch == self.epoch_store.epoch() {
            // sum only when we are within the same epoch
            GasCostSummary::new(
                previous_gas_costs.computation_cost + current_gas_costs.computation_cost,
                previous_gas_costs.computation_cost_burned
                    + current_gas_costs.computation_cost_burned,
                previous_gas_costs.storage_cost + current_gas_costs.storage_cost,
                previous_gas_costs.storage_rebate + current_gas_costs.storage_rebate,
                previous_gas_costs.non_refundable_storage_fee
                    + current_gas_costs.non_refundable_storage_fee,
            )
        } else {
            current_gas_costs
        }
    }

    /// Augments the last checkpoint of the epoch by creating and executing an
    /// advance epoch transaction.
    #[instrument(level = "error", skip_all)]
    async fn augment_epoch_last_checkpoint(
        &self,
        epoch_total_gas_cost: &GasCostSummary,
        epoch_start_timestamp_ms: CheckpointTimestamp,
        checkpoint_effects: &mut Vec<TransactionEffects>,
        signatures: &mut Vec<Vec<UserSignature>>,
        checkpoint: CheckpointSequenceNumber,
        scores: Vec<u64>,
    ) -> CheckpointBuilderResult<(IotaSystemState, Option<SystemEpochInfoEvent>)> {
        let (system_state, system_epoch_info_event, effects) = self
            .state
            .create_and_execute_advance_epoch_tx(
                &self.epoch_store,
                epoch_total_gas_cost,
                checkpoint,
                epoch_start_timestamp_ms,
                scores,
            )
            .await?;
        checkpoint_effects.push(effects);
        signatures.push(vec![]);
        Ok((system_state, system_epoch_info_event))
    }

    /// For the given roots return complete list of effects to include in
    /// checkpoint This list includes the roots and all their dependencies,
    /// which are not part of checkpoint already. Note that this function
    /// may be called multiple times to construct the checkpoint.
    /// `existing_tx_digests_in_checkpoint` is used to track the transactions
    /// that are already included in the checkpoint. Txs in `roots` that
    /// need to be included in the checkpoint will be added to
    /// `existing_tx_digests_in_checkpoint` after the call of this function.
    #[instrument(level = "debug", skip_all)]
    fn complete_checkpoint_effects(
        &self,
        mut roots: Vec<TransactionEffects>,
        existing_tx_digests_in_checkpoint: &mut BTreeSet<TransactionDigest>,
    ) -> IotaResult<Vec<TransactionEffects>> {
        let _scope = monitored_scope("CheckpointBuilder::complete_checkpoint_effects");
        let mut results = vec![];
        let mut seen = HashSet::new();
        loop {
            let mut pending = HashSet::new();

            let transactions_included = self
                .epoch_store
                .builder_included_transactions_in_checkpoint(
                    roots.iter().map(|e| e.transaction_digest()),
                )?;

            for (effect, tx_included) in roots.into_iter().zip(transactions_included.into_iter()) {
                let digest = effect.transaction_digest();
                // Unnecessary to read effects of a dependency if the effect is already
                // processed.
                seen.insert(*digest);

                // Skip roots that are already included in the checkpoint.
                if existing_tx_digests_in_checkpoint.contains(effect.transaction_digest()) {
                    continue;
                }

                // Skip roots already included in checkpoints or roots from previous epochs
                if tx_included || effect.epoch() < self.epoch_store.epoch() {
                    continue;
                }

                let existing_effects = self
                    .epoch_store
                    .transactions_executed_in_cur_epoch(effect.dependencies())?;

                for (dependency, effects_signature_exists) in
                    effect.dependencies().iter().zip(existing_effects.iter())
                {
                    // Skip here if dependency not executed in the current epoch.
                    // Note that the existence of an effects signature in the
                    // epoch store for the given digest indicates that the transaction
                    // was locally executed in the current epoch
                    if !effects_signature_exists {
                        continue;
                    }
                    if seen.insert(*dependency) {
                        pending.insert(*dependency);
                    }
                }
                results.push(effect);
            }
            if pending.is_empty() {
                break;
            }
            let pending = pending.into_iter().collect::<Vec<_>>();
            let effects = self
                .effects_store
                .try_multi_get_executed_effects(&pending)?;
            let effects = effects
                .into_iter()
                .zip(pending)
                .map(|(opt, digest)| match opt {
                    Some(x) => x,
                    None => panic!(
                        "Can not find effect for transaction {digest}, however transaction that depend on it was already executed"
                    ),
                })
                .collect::<Vec<_>>();
            roots = effects;
        }

        existing_tx_digests_in_checkpoint.extend(results.iter().map(|e| e.transaction_digest()));
        Ok(results)
    }

    // This function is used to check the invariants of the consensus commit
    // prologue transactions in the checkpoint in simtest.
    #[cfg(msim)]
    fn expensive_consensus_commit_prologue_invariants_check(
        &self,
        root_digests: &[TransactionDigest],
        sorted: &[TransactionEffects],
    ) {
        // Gets all the consensus commit prologue transactions from the roots.
        let root_txs = self
            .state
            .get_transaction_cache_reader()
            .multi_get_transaction_blocks(root_digests);
        let ccps = root_txs
            .iter()
            .filter_map(|tx| {
                tx.as_ref().filter(|tx| {
                    matches!(
                        tx.transaction().kind(),
                        TransactionKind::ConsensusCommitPrologueV1(_)
                    )
                })
            })
            .collect::<Vec<_>>();

        // There should be at most one consensus commit prologue transaction in the
        // roots.
        assert!(ccps.len() <= 1);

        // Get all the transactions in the checkpoint.
        let txs = self
            .state
            .get_transaction_cache_reader()
            .multi_get_transaction_blocks(
                &sorted
                    .iter()
                    .map(|tx| *tx.transaction_digest())
                    .collect::<Vec<_>>(),
            );

        if ccps.is_empty() {
            // If there is no consensus commit prologue transaction in the roots, then there
            // should be no consensus commit prologue transaction in the
            // checkpoint.
            for tx in txs.iter().flatten() {
                assert!(!matches!(
                    tx.transaction().kind(),
                    TransactionKind::ConsensusCommitPrologueV1(_)
                ));
            }
        } else {
            // If there is one consensus commit prologue, it must be the first one in the
            // checkpoint.
            assert!(matches!(
                txs[0].as_ref().unwrap().transaction().kind(),
                TransactionKind::ConsensusCommitPrologueV1(_)
            ));

            assert_eq!(ccps[0].digest(), txs[0].as_ref().unwrap().digest());

            for tx in txs.iter().skip(1).flatten() {
                assert!(!matches!(
                    tx.transaction().kind(),
                    TransactionKind::ConsensusCommitPrologueV1(_)
                ));
            }
        }
    }
}

impl CheckpointAggregator {
    fn new(
        tables: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        notify: Arc<Notify>,
        output: Box<dyn CertifiedCheckpointOutput>,
        state: Arc<AuthorityState>,
        metrics: Arc<CheckpointMetrics>,
    ) -> Self {
        let current = None;
        Self {
            store: tables,
            epoch_store,
            notify,
            current,
            output,
            state,
            metrics,
        }
    }

    /// Runs the `CheckpointAggregator` in an asynchronous loop, managing the
    /// aggregation of checkpoints.
    /// The function ensures continuous aggregation of checkpoints, handling
    /// errors and retries gracefully, and allowing for proper shutdown on
    /// receiving an exit signal.
    async fn run(mut self) {
        info!("Starting CheckpointAggregator");
        loop {
            if let Err(e) = self.run_and_notify().await {
                error!(
                    "Error while aggregating checkpoint, will retry in 1s: {:?}",
                    e
                );
                self.metrics.checkpoint_errors.inc();
                tokio::time::sleep(Duration::from_secs(1)).await;
                continue;
            }

            let _ = timeout(Duration::from_secs(1), self.notify.notified()).await;
        }
    }

    async fn run_and_notify(&mut self) -> IotaResult {
        let summaries = self.run_inner()?;
        for summary in summaries {
            self.output.certified_checkpoint_created(&summary).await?;
        }
        Ok(())
    }

    fn run_inner(&mut self) -> IotaResult<Vec<CertifiedCheckpointSummary>> {
        let _scope = monitored_scope("CheckpointAggregator");
        let mut result = vec![];
        'outer: loop {
            let next_to_certify = self.next_checkpoint_to_certify()?;
            let current = if let Some(current) = &mut self.current {
                // It's possible that the checkpoint was already certified by
                // the rest of the network and we've already received the
                // certified checkpoint via StateSync. In this case, we reset
                // the current signature aggregator to the next checkpoint to
                // be certified
                if current.summary.sequence_number < next_to_certify {
                    self.current = None;
                    continue;
                }
                current
            } else {
                let Some(summary) = self
                    .epoch_store
                    .get_built_checkpoint_summary(next_to_certify)?
                else {
                    return Ok(result);
                };
                self.current = Some(CheckpointSignatureAggregator {
                    next_index: 0,
                    digest: summary.digest(),
                    summary,
                    signatures_by_digest: MultiStakeAggregator::new(
                        self.epoch_store.committee().clone(),
                    ),
                    store: self.store.clone(),
                    state: self.state.clone(),
                    metrics: self.metrics.clone(),
                });
                self.current.as_mut().unwrap()
            };

            let epoch_tables = self
                .epoch_store
                .tables()
                .expect("should not run past end of epoch");
            let iter = epoch_tables
                .pending_checkpoint_signatures
                .safe_iter_with_bounds(
                    Some((current.summary.sequence_number, current.next_index)),
                    None,
                );
            for item in iter {
                let ((seq, index), data) = item?;
                if seq != current.summary.sequence_number {
                    trace!(
                        checkpoint_seq =? current.summary.sequence_number,
                        "Not enough checkpoint signatures",
                    );
                    // No more signatures (yet) for this checkpoint
                    return Ok(result);
                }
                trace!(
                    checkpoint_seq = current.summary.sequence_number,
                    "Processing signature for checkpoint (digest: {:?}) from {:?}",
                    current.summary.digest(),
                    data.summary.auth_sig().authority.concise()
                );
                self.metrics
                    .checkpoint_participation
                    .with_label_values(&[&format!(
                        "{:?}",
                        data.summary.auth_sig().authority.concise()
                    )])
                    .inc();
                if let Ok(auth_signature) = current.try_aggregate(data) {
                    debug!(
                        checkpoint_seq = current.summary.sequence_number,
                        "Successfully aggregated signatures for checkpoint (digest: {:?})",
                        current.summary.digest(),
                    );
                    let summary = VerifiedCheckpoint::new_unchecked(
                        CertifiedCheckpointSummary::new_from_data_and_sig(
                            current.summary.clone(),
                            auth_signature,
                        ),
                    );

                    self.store.insert_certified_checkpoint(&summary)?;
                    self.metrics
                        .last_certified_checkpoint
                        .set(current.summary.sequence_number as i64);
                    current
                        .summary
                        .report_checkpoint_age(&self.metrics.last_certified_checkpoint_age);
                    result.push(summary.into_inner());
                    self.current = None;
                    continue 'outer;
                } else {
                    current.next_index = index + 1;
                }
            }
            break;
        }
        Ok(result)
    }

    fn next_checkpoint_to_certify(&self) -> IotaResult<CheckpointSequenceNumber> {
        Ok(self
            .store
            .tables
            .certified_checkpoints
            .safe_range_iter_reversed(..)
            .next()
            .transpose()?
            .map(|(seq, _)| seq + 1)
            .unwrap_or_default())
    }
}

impl CheckpointSignatureAggregator {
    #[expect(clippy::result_unit_err)]
    pub fn try_aggregate(
        &mut self,
        data: CheckpointSignatureMessage,
    ) -> Result<AuthorityStrongQuorumSignInfo, ()> {
        let their_digest = *data.summary.digest();
        let (_, signature) = data.summary.into_data_and_sig();
        let author = signature.authority;
        let envelope =
            SignedCheckpointSummary::new_from_data_and_sig(self.summary.clone(), signature);
        match self.signatures_by_digest.insert(their_digest, envelope) {
            // ignore repeated signatures
            InsertResult::Failed {
                error:
                    IotaError::StakeAggregatorRepeatedSigner {
                        conflicting_sig: false,
                        ..
                    },
            } => Err(()),
            InsertResult::Failed { error } => {
                warn!(
                    checkpoint_seq = self.summary.sequence_number,
                    "Failed to aggregate new signature from validator {:?}: {:?}",
                    author.concise(),
                    error
                );
                self.check_for_split_brain();
                Err(())
            }
            InsertResult::QuorumReached(cert) => {
                // It is not guaranteed that signature.authority == consensus_cert.author, but
                // we do verify the signature so we know that the author signed
                // the message at some point.
                if their_digest != self.digest {
                    self.metrics.remote_checkpoint_forks.inc();
                    warn!(
                        checkpoint_seq = self.summary.sequence_number,
                        "Validator {:?} has mismatching checkpoint digest {}, we have digest {}",
                        author.concise(),
                        their_digest,
                        self.digest
                    );
                    return Err(());
                }
                Ok(cert)
            }
            InsertResult::NotEnoughVotes {
                bad_votes: _,
                bad_authorities: _,
            } => {
                self.check_for_split_brain();
                Err(())
            }
        }
    }

    /// Check if there is a split brain condition in checkpoint signature
    /// aggregation, defined as any state wherein it is no longer possible
    /// to achieve quorum on a checkpoint proposal, irrespective of the
    /// outcome of any outstanding votes.
    fn check_for_split_brain(&self) {
        debug!(
            checkpoint_seq = self.summary.sequence_number,
            "Checking for split brain condition"
        );
        if self.signatures_by_digest.quorum_unreachable() {
            // TODO: at this point we should immediately halt processing
            // of new transaction certificates to avoid building on top of
            // forked output
            // self.halt_all_execution();

            let digests_by_stake_messages = self
                .signatures_by_digest
                .get_all_unique_values()
                .into_iter()
                .sorted_by_key(|(_, (_, stake))| -(*stake as i64))
                .map(|(digest, (_authorities, total_stake))| {
                    format!("{digest} (total stake: {total_stake})")
                })
                .collect::<Vec<String>>();
            debug_fatal!(
                "Split brain detected in checkpoint signature aggregation for checkpoint {:?}. Remaining stake: {:?}, Digests by stake: {:?}",
                self.summary.sequence_number,
                self.signatures_by_digest.uncommitted_stake(),
                digests_by_stake_messages,
            );
            self.metrics.split_brain_checkpoint_forks.inc();

            let all_unique_values = self.signatures_by_digest.get_all_unique_values();
            let local_summary = self.summary.clone();
            let state = self.state.clone();
            let tables = self.store.clone();

            tokio::spawn(async move {
                diagnose_split_brain(all_unique_values, local_summary, state, tables).await;
            });
        }
    }
}

/// Create data dump containing relevant data for diagnosing cause of the
/// split brain by querying one disagreeing validator for full checkpoint
/// contents. To minimize peer chatter, we only query one validator at random
/// from each disagreeing faction, as all honest validators that participated in
/// this round may inevitably run the same process.
async fn diagnose_split_brain(
    all_unique_values: BTreeMap<CheckpointDigest, (Vec<AuthorityName>, StakeUnit)>,
    local_summary: CheckpointSummary,
    state: Arc<AuthorityState>,
    tables: Arc<CheckpointStore>,
) {
    debug!(
        checkpoint_seq = local_summary.sequence_number,
        "Running split brain diagnostics..."
    );
    let time = SystemTime::now();
    // collect one random disagreeing validator per differing digest
    let digest_to_validator = all_unique_values
        .iter()
        .filter_map(|(digest, (validators, _))| {
            if *digest != local_summary.digest() {
                let random_validator = validators.choose(&mut get_rng()).unwrap();
                Some((*digest, *random_validator))
            } else {
                None
            }
        })
        .collect::<HashMap<_, _>>();
    if digest_to_validator.is_empty() {
        panic!(
            "Given split brain condition, there should be at \
                least one validator that disagrees with local signature"
        );
    }

    let epoch_store = state.load_epoch_store_one_call_per_task();
    let committee = epoch_store
        .epoch_start_state()
        .get_iota_committee_with_network_metadata();
    let network_config = default_iota_network_config();
    let network_clients =
        make_network_authority_clients_with_network_config(&committee, &network_config);

    // Query all disagreeing validators
    let response_futures = digest_to_validator
        .values()
        .cloned()
        .map(|validator| {
            let client = network_clients
                .get(&validator)
                .expect("Failed to get network client");
            let request = CheckpointRequest {
                sequence_number: Some(local_summary.sequence_number),
                request_content: true,
                certified: false,
            };
            client.get_checkpoint_v2(request)
        })
        .collect::<Vec<_>>();

    let digest_name_pair = digest_to_validator.iter();
    let response_data = futures::future::join_all(response_futures)
        .await
        .into_iter()
        .zip(digest_name_pair)
        .filter_map(|(response, (digest, name))| match response {
            Ok(response) => match response {
                CheckpointResponse {
                    checkpoint: Some(CheckpointSummaryResponse::Pending(summary)),
                    contents: Some(contents),
                } => Some((*name, *digest, summary, contents)),
                CheckpointResponse {
                    checkpoint: Some(CheckpointSummaryResponse::Certified(_)),
                    contents: _,
                } => {
                    panic!("Expected pending checkpoint, but got certified checkpoint");
                }
                CheckpointResponse {
                    checkpoint: None,
                    contents: _,
                } => {
                    error!(
                        "Summary for checkpoint {:?} not found on validator {:?}",
                        local_summary.sequence_number, name
                    );
                    None
                }
                CheckpointResponse {
                    checkpoint: _,
                    contents: None,
                } => {
                    error!(
                        "Contents for checkpoint {:?} not found on validator {:?}",
                        local_summary.sequence_number, name
                    );
                    None
                }
            },
            Err(e) => {
                error!(
                    "Failed to get checkpoint contents from validator for fork diagnostics: {:?}",
                    e
                );
                None
            }
        })
        .collect::<Vec<_>>();

    let local_checkpoint_contents = tables
        .get_checkpoint_contents(&local_summary.contents_digest)
        .unwrap_or_else(|_| {
            panic!(
                "Could not find checkpoint contents for digest {:?}",
                local_summary.digest()
            )
        })
        .unwrap_or_else(|| {
            panic!(
                "Could not find local full checkpoint contents for checkpoint {:?}, digest {:?}",
                local_summary.sequence_number,
                local_summary.digest()
            )
        });
    let local_contents_text = format!("{local_checkpoint_contents:?}");

    let local_summary_text = format!("{local_summary:?}");
    let local_validator = state.name.concise();
    let diff_patches = response_data
        .iter()
        .map(|(name, other_digest, other_summary, contents)| {
            let other_contents_text = format!("{contents:?}");
            let other_summary_text = format!("{other_summary:?}");
            let (local_transactions, local_effects): (Vec<_>, Vec<_>) = local_checkpoint_contents
                .enumerate_transactions(&local_summary)
                .map(|(_, exec_digest)| (exec_digest.transaction, exec_digest.effects))
                .unzip();
            let (other_transactions, other_effects): (Vec<_>, Vec<_>) = contents
                .enumerate_transactions(other_summary)
                .map(|(_, exec_digest)| (exec_digest.transaction, exec_digest.effects))
                .unzip();
            let summary_patch = create_patch(&local_summary_text, &other_summary_text);
            let contents_patch = create_patch(&local_contents_text, &other_contents_text);
            let local_transactions_text = format!("{local_transactions:#?}");
            let other_transactions_text = format!("{other_transactions:#?}");
            let transactions_patch =
                create_patch(&local_transactions_text, &other_transactions_text);
            let local_effects_text = format!("{local_effects:#?}");
            let other_effects_text = format!("{other_effects:#?}");
            let effects_patch = create_patch(&local_effects_text, &other_effects_text);
            let seq_number = local_summary.sequence_number;
            let local_digest = local_summary.digest();
            let other_validator = name.concise();
            format!(
                "Checkpoint: {seq_number:?}\n\
                Local validator (original): {local_validator:?}, digest: {local_digest}\n\
                Other validator (modified): {other_validator:?}, digest: {other_digest}\n\n\
                Summary Diff: \n{summary_patch}\n\n\
                Contents Diff: \n{contents_patch}\n\n\
                Transactions Diff: \n{transactions_patch}\n\n\
                Effects Diff: \n{effects_patch}",
            )
        })
        .collect::<Vec<_>>()
        .join("\n\n\n");

    let header = format!(
        "Checkpoint Fork Dump - Authority {local_validator:?}: \n\
        Datetime: {time:?}"
    );
    let fork_logs_text = format!("{header}\n\n{diff_patches}\n\n");
    let checkpoint_fork_dir = iota_common::tempdir().keep();
    let checkpoint_fork_file_path = checkpoint_fork_dir.join(Path::new("checkpoint_fork_dump.txt"));
    let mut file = File::create(checkpoint_fork_file_path).unwrap();
    write!(file, "{fork_logs_text}").unwrap();
    debug!("{}", fork_logs_text);
}

pub trait CheckpointServiceNotify {
    fn notify_checkpoint_signature(
        &self,
        epoch_store: &AuthorityPerEpochStore,
        info: &CheckpointSignatureMessage,
    ) -> IotaResult;

    fn notify_checkpoint(&self) -> IotaResult;
}

enum CheckpointServiceState {
    Unstarted(
        Box<(
            CheckpointBuilder,
            CheckpointAggregator,
            CheckpointStateHasher,
        )>,
    ),
    Started,
}

impl CheckpointServiceState {
    fn take_unstarted(
        &mut self,
    ) -> (
        CheckpointBuilder,
        CheckpointAggregator,
        CheckpointStateHasher,
    ) {
        let mut state = CheckpointServiceState::Started;
        std::mem::swap(self, &mut state);

        match state {
            CheckpointServiceState::Unstarted(tup) => (tup.0, tup.1, tup.2),
            CheckpointServiceState::Started => panic!("CheckpointServiceState is already started"),
        }
    }
}

pub struct CheckpointService {
    tables: Arc<CheckpointStore>,
    notify_builder: Arc<Notify>,
    notify_aggregator: Arc<Notify>,
    last_signature_index: Mutex<u64>,
    // A notification for the current highest built sequence number.
    highest_currently_built_seq_tx: watch::Sender<CheckpointSequenceNumber>,
    // The highest sequence number that had already been built at the time CheckpointService
    // was constructed
    highest_previously_built_seq: CheckpointSequenceNumber,
    metrics: Arc<CheckpointMetrics>,
    state: Mutex<CheckpointServiceState>,
}

impl CheckpointService {
    /// Spawns the checkpoint service, initializing and starting the checkpoint
    /// builder and aggregator tasks.
    /// Constructs a new CheckpointService in an un-started state.
    pub fn build(
        state: Arc<AuthorityState>,
        checkpoint_store: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        effects_store: Arc<dyn TransactionCacheRead>,
        global_state_hasher: Weak<GlobalStateHasher>,
        checkpoint_output: Box<dyn CheckpointOutput>,
        certified_checkpoint_output: Box<dyn CertifiedCheckpointOutput>,
        metrics: Arc<CheckpointMetrics>,
        max_transactions_per_checkpoint: usize,
        max_checkpoint_size_bytes: usize,
    ) -> Arc<Self> {
        info!(
            "Starting checkpoint service with {max_transactions_per_checkpoint} max_transactions_per_checkpoint and {max_checkpoint_size_bytes} max_checkpoint_size_bytes"
        );
        let notify_builder = Arc::new(Notify::new());
        let notify_aggregator = Arc::new(Notify::new());

        // We may have built higher checkpoint numbers before restarting.
        let highest_previously_built_seq = checkpoint_store
            .get_latest_locally_computed_checkpoint()
            .expect("failed to get latest locally computed checkpoint")
            .map(|s| s.sequence_number)
            .unwrap_or(0);

        let highest_currently_built_seq =
            CheckpointBuilder::load_last_built_checkpoint_summary(&epoch_store, &checkpoint_store)
                .expect("epoch should not have ended")
                .map(|(seq, _)| seq)
                .unwrap_or(0);

        let (highest_currently_built_seq_tx, _) = watch::channel(highest_currently_built_seq);

        let aggregator = CheckpointAggregator::new(
            checkpoint_store.clone(),
            epoch_store.clone(),
            notify_aggregator.clone(),
            certified_checkpoint_output,
            state.clone(),
            metrics.clone(),
        );

        let (send_to_hasher, receive_from_builder) = mpsc::channel(16);

        let ckpt_state_hasher = CheckpointStateHasher::new(
            epoch_store.clone(),
            global_state_hasher.clone(),
            receive_from_builder,
        );

        let builder = CheckpointBuilder::new(
            state,
            checkpoint_store.clone(),
            epoch_store.clone(),
            notify_builder.clone(),
            effects_store,
            global_state_hasher,
            send_to_hasher,
            checkpoint_output,
            notify_aggregator.clone(),
            highest_currently_built_seq_tx.clone(),
            metrics.clone(),
            max_transactions_per_checkpoint,
            max_checkpoint_size_bytes,
        );

        let last_signature_index = epoch_store
            .get_last_checkpoint_signature_index()
            .expect("should not cross end of epoch");
        let last_signature_index = Mutex::new(last_signature_index);

        Arc::new(Self {
            tables: checkpoint_store,
            notify_builder,
            notify_aggregator,
            last_signature_index,
            highest_currently_built_seq_tx,
            highest_previously_built_seq,
            metrics,
            state: Mutex::new(CheckpointServiceState::Unstarted(Box::new((
                builder,
                aggregator,
                ckpt_state_hasher,
            )))),
        })
    }

    /// Starts the CheckpointService.
    ///
    /// This function blocks until the CheckpointBuilder re-builds all
    /// checkpoints that had been built before the most recent restart. You
    /// can think of this as a WAL replay operation. Upon startup, we may
    /// have a number of consensus commits and resulting checkpoints that
    /// were built but not committed to disk. We want to reprocess the
    /// commits and rebuild the checkpoints before starting normal operation.
    pub async fn spawn(&self, consensus_replay_waiter: Option<ReplayWaiter>) -> JoinSet<()> {
        let mut tasks = JoinSet::new();

        let (builder, aggregator, state_hasher) = self.state.lock().take_unstarted();
        tasks.spawn(monitored_future!(builder.run(consensus_replay_waiter)));
        tasks.spawn(monitored_future!(aggregator.run()));
        tasks.spawn(monitored_future!(state_hasher.run()));

        // If this times out, the validator may still start up. The worst that can
        // happen is that we will crash later on instead of immediately. The eventual
        // crash would occur because we may be missing transactions that are below the
        // highest_synced_checkpoint watermark, which can cause a crash in
        // `CheckpointExecutor::extract_randomness_rounds`.
        if tokio::time::timeout(
            Duration::from_secs(120),
            self.wait_for_rebuilt_checkpoints(),
        )
        .await
        .is_err()
        {
            debug_fatal!("Timed out waiting for checkpoints to be rebuilt");
        }

        tasks
    }
}

impl CheckpointService {
    /// Waits until all checkpoints had been built before the node restarted
    /// are rebuilt. This is required to preserve the invariant that all
    /// checkpoints (and their transactions) below the
    /// highest_synced_checkpoint watermark are available. Once the
    /// checkpoints are constructed, we can be sure that the transactions
    /// have also been executed.
    pub async fn wait_for_rebuilt_checkpoints(&self) {
        let highest_previously_built_seq = self.highest_previously_built_seq;
        let mut rx = self.highest_currently_built_seq_tx.subscribe();
        let mut highest_currently_built_seq = *rx.borrow_and_update();
        info!(
            "Waiting for checkpoints to be rebuilt, previously built seq: \
            {highest_previously_built_seq}, currently built seq: {highest_currently_built_seq}"
        );
        loop {
            if highest_currently_built_seq >= highest_previously_built_seq {
                info!("Checkpoint rebuild complete");
                break;
            }
            rx.changed().await.unwrap();
            highest_currently_built_seq = *rx.borrow_and_update();
        }
    }

    #[cfg(test)]
    fn write_and_notify_checkpoint_for_testing(
        &self,
        epoch_store: &AuthorityPerEpochStore,
        checkpoint: PendingCheckpoint,
    ) -> IotaResult {
        use crate::authority::authority_per_epoch_store::consensus_quarantine::ConsensusCommitOutput;

        let mut output = ConsensusCommitOutput::new(0);
        epoch_store.write_pending_checkpoint(&mut output, &checkpoint)?;
        output.set_default_commit_stats_for_testing();
        epoch_store.push_consensus_output_for_tests(output);
        self.notify_checkpoint()?;
        Ok(())
    }
}

impl CheckpointServiceNotify for CheckpointService {
    fn notify_checkpoint_signature(
        &self,
        epoch_store: &AuthorityPerEpochStore,
        info: &CheckpointSignatureMessage,
    ) -> IotaResult {
        let sequence = info.summary.sequence_number;
        let signer = info.summary.auth_sig().authority.concise();

        if let Some(highest_verified_checkpoint) = self
            .tables
            .get_highest_verified_checkpoint()?
            .map(|x| x.sequence_number())
        {
            if sequence <= highest_verified_checkpoint {
                trace!(
                    checkpoint_seq = sequence,
                    "Ignore checkpoint signature from {} - already certified", signer,
                );
                self.metrics
                    .last_ignored_checkpoint_signature_received
                    .set(sequence as i64);
                return Ok(());
            }
        }
        trace!(
            checkpoint_seq = sequence,
            "Received checkpoint signature, digest {} from {}",
            info.summary.digest(),
            signer,
        );
        self.metrics
            .last_received_checkpoint_signatures
            .with_label_values(&[&signer.to_string()])
            .set(sequence as i64);
        // While it can be tempting to make last_signature_index into AtomicU64, this
        // won't work We need to make sure we write to `pending_signatures` and
        // trigger `notify_aggregator` without race conditions
        let mut index = self.last_signature_index.lock();
        *index += 1;
        epoch_store.insert_checkpoint_signature(sequence, *index, info)?;
        self.notify_aggregator.notify_one();
        Ok(())
    }

    fn notify_checkpoint(&self) -> IotaResult {
        self.notify_builder.notify_one();
        Ok(())
    }
}

#[iota_macros::with_checked_arithmetic]
mod checked {
    use iota_sdk_types::GasCostSummary;
    use iota_types::effects::{TransactionEffects, TransactionEffectsAPI};
    use itertools::MultiUnzip;

    #[expect(clippy::type_complexity)]
    pub fn new_gas_cost_summary_from_txn_effects<'a>(
        transactions: impl Iterator<Item = &'a TransactionEffects>,
    ) -> GasCostSummary {
        let (
            storage_costs,
            computation_costs,
            computation_costs_burned,
            storage_rebates,
            non_refundable_storage_fee,
        ): (Vec<u64>, Vec<u64>, Vec<u64>, Vec<u64>, Vec<u64>) = transactions
            .map(|e| {
                (
                    e.gas_cost_summary().storage_cost,
                    e.gas_cost_summary().computation_cost,
                    e.gas_cost_summary().computation_cost_burned,
                    e.gas_cost_summary().storage_rebate,
                    e.gas_cost_summary().non_refundable_storage_fee,
                )
            })
            .multiunzip();

        GasCostSummary::new(
            computation_costs.iter().sum(),
            computation_costs_burned.iter().sum(),
            storage_costs.iter().sum(),
            storage_rebates.iter().sum(),
            non_refundable_storage_fee.iter().sum(),
        )
    }
}
// test helper
pub struct CheckpointServiceNoop {}
impl CheckpointServiceNotify for CheckpointServiceNoop {
    fn notify_checkpoint_signature(
        &self,
        _: &AuthorityPerEpochStore,
        _: &CheckpointSignatureMessage,
    ) -> IotaResult {
        Ok(())
    }

    fn notify_checkpoint(&self) -> IotaResult {
        Ok(())
    }
}

pin_project! {
    pub struct PollCounter<Fut> {
        #[pin]
        future: Fut,
        count: usize,
    }
}

impl<Fut> PollCounter<Fut> {
    pub fn new(future: Fut) -> Self {
        Self { future, count: 0 }
    }

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

impl<Fut: Future> Future for PollCounter<Fut> {
    type Output = (usize, Fut::Output);

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        *this.count += 1;
        match this.future.poll(cx) {
            Poll::Ready(output) => Poll::Ready((*this.count, output)),
            Poll::Pending => Poll::Pending,
        }
    }
}

fn poll_count<Fut>(future: Fut) -> PollCounter<Fut> {
    PollCounter::new(future)
}

/// A verified checkpoint over the given contents at the given sequence
/// number, with a placeholder signature; usable wherever verification is
/// not re-run and no committee is needed.
#[cfg(test)]
pub(crate) fn test_checkpoint_with_contents(
    sequence_number: CheckpointSequenceNumber,
    full_contents: &FullCheckpointContents,
) -> VerifiedCheckpoint {
    let contents = full_contents.checkpoint_contents();
    let summary = CheckpointSummary {
        epoch: 0,
        sequence_number,
        network_total_transactions: full_contents.size() as u64,
        contents_digest: contents.digest(),
        previous_digest: None,
        epoch_rolling_gas_cost_summary: GasCostSummary::default(),
        end_of_epoch_data: None,
        timestamp_ms: 0,
        version_specific_data: Vec::new(),
        checkpoint_commitments: Vec::new(),
    };
    let sig = AuthorityStrongQuorumSignInfo {
        epoch: 0,
        signature: Default::default(),
        signers_map: Default::default(),
    };
    VerifiedCheckpoint::new_unchecked(
        iota_types::message_envelope::Envelope::new_from_data_and_sig(summary, sig),
    )
}

#[cfg(test)]
mod tests {
    use std::{
        collections::{BTreeMap, HashMap},
        ops::Deref,
    };

    use futures::{FutureExt as _, future::BoxFuture};
    use iota_macros::sim_test;
    use iota_protocol_config::{Chain, ProtocolConfig, ProtocolVersion};
    use iota_sdk_types::{
        GenesisObject, Identifier, ObjectData, ObjectId, Owner, TransactionEffectsDigest, Version,
        move_package::MovePackage,
    };
    use iota_types::{
        effects::{
            TransactionEffects, TransactionEffectsAPIForTesting, TransactionEffectsExtForTesting,
            TransactionEvents,
        },
        messages_checkpoint::SignedCheckpointSummary,
        transaction::VerifiedTransaction,
    };
    use tokio::sync::mpsc;

    use super::*;
    use crate::authority::test_authority_builder::TestAuthorityBuilder;

    #[tokio::test]
    async fn insert_verified_checkpoint_contents_persists_digests_and_caches_full_contents() {
        let tempdir = iota_common::tempdir();
        let path = tempdir.path();

        let full_contents = FullCheckpointContents::random_for_testing();
        let checkpoint = test_checkpoint_with_contents(0, &full_contents);
        let contents_digest = checkpoint.contents_digest;

        {
            let store = CheckpointStore::new(path);
            store
                .insert_verified_checkpoint_contents(
                    &checkpoint,
                    VerifiedCheckpointContents::new_unchecked(full_contents.clone()),
                )
                .unwrap();

            // Full contents are served from the in-memory cache.
            assert_eq!(
                store
                    .get_full_checkpoint_contents_by_sequence_number(0)
                    .unwrap()
                    .as_ref(),
                &full_contents
            );
            assert_eq!(
                store
                    .get_full_checkpoint_contents_by_digest(&contents_digest)
                    .unwrap()
                    .as_ref(),
                &full_contents
            );
            // The digest-form contents are durable.
            assert_eq!(
                store
                    .get_checkpoint_contents(&contents_digest)
                    .unwrap()
                    .map(|c| c.digest()),
                Some(contents_digest)
            );
        }

        // Reopening drops the in-memory cache, but the digest-form contents
        // survive for readers to reconstruct full contents from.
        let store = CheckpointStore::new(path);
        assert!(
            store
                .get_full_checkpoint_contents_by_sequence_number(0)
                .is_none()
        );
        assert!(
            store
                .get_full_checkpoint_contents_by_digest(&contents_digest)
                .is_none()
        );
        assert!(
            store
                .get_checkpoint_contents(&contents_digest)
                .unwrap()
                .is_some()
        );
    }

    #[tokio::test]
    async fn cache_full_checkpoint_contents_serves_reads_without_disk_writes() {
        let store = CheckpointStore::new_for_tests();
        let full_contents = FullCheckpointContents::random_for_testing();
        let checkpoint = test_checkpoint_with_contents(0, &full_contents);
        let contents_digest = checkpoint.contents_digest;

        store.cache_full_checkpoint_contents(
            checkpoint.sequence_number(),
            contents_digest,
            full_contents.clone(),
        );

        assert_eq!(
            store
                .get_full_checkpoint_contents_by_sequence_number(0)
                .unwrap()
                .as_ref(),
            &full_contents
        );
        assert_eq!(
            store
                .get_full_checkpoint_contents_by_digest(&contents_digest)
                .unwrap()
                .as_ref(),
            &full_contents
        );
        // Cache-only: the digest-form contents row is untouched.
        assert!(
            store
                .get_checkpoint_contents(&contents_digest)
                .unwrap()
                .is_none()
        );
    }

    // State-sync skips its own durable `checkpoint_content` write when the
    // full contents are already available, and the checkpoint executor panics
    // on a missing row. So the builder must not serve contents from the cache
    // before the row is durable.
    #[tokio::test]
    async fn builder_caches_full_contents_only_after_durable_contents_write() {
        let state = TestAuthorityBuilder::new().build().await;

        let tx = VerifiedTransaction::new_genesis_transaction(vec![], vec![]);
        let digest = *tx.digest();
        state
            .database_for_testing()
            .perpetual_tables
            .transactions
            .insert(&digest, tx.serializable_ref())
            .unwrap();

        let mut effects_map = HashMap::new();
        commit_cert_for_test(
            &mut effects_map,
            state.clone(),
            digest,
            vec![],
            GasCostSummary::new(1, 1, 1, 1, 1),
        );
        let effects = effects_map[&digest].clone();

        let signature = iota_types::crypto::zero_ed25519_signature().into();
        state
            .epoch_store_for_testing()
            .test_insert_user_signature(digest, vec![signature]);

        let (output, _result) = mpsc::channel::<(CheckpointContents, CheckpointSummary)>(10);
        let (certified_output, _certified_result) = mpsc::channel::<CertifiedCheckpointSummary>(10);

        let tmp_dir = iota_common::tempdir();
        let checkpoint_store = CheckpointStore::new(tmp_dir.path());
        let epoch_store = state.epoch_store_for_testing();

        let global_state_hasher = Arc::new(GlobalStateHasher::new_for_tests(
            state.get_global_state_hash_store().clone(),
        ));

        let checkpoint_service = CheckpointService::build(
            state.clone(),
            checkpoint_store.clone(),
            epoch_store.clone(),
            Arc::new(effects_map),
            Arc::downgrade(&global_state_hasher),
            Box::new(output),
            Box::new(certified_output),
            CheckpointMetrics::new_for_tests(),
            3,
            100_000,
        );
        // Drive the builder manually instead of spawning it, to observe the
        // store between checkpoint creation and the durable batch write. The
        // state hasher must stay alive: `create_checkpoints` sends to it.
        let (builder, _aggregator, _hasher) = checkpoint_service.state.lock().take_unstarted();

        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(0, vec![digest], 0))
            .unwrap();

        let details = PendingCheckpointInfo {
            timestamp_ms: 0,
            last_of_epoch: false,
            checkpoint_height: 0,
        };
        let new_checkpoints = builder
            .create_checkpoints(vec![effects], &details, &HashSet::from([digest]))
            .await
            .unwrap();
        let summary = new_checkpoints.first().summary.clone();

        assert!(
            checkpoint_store
                .get_full_checkpoint_contents_by_sequence_number(summary.sequence_number)
                .is_none(),
            "full contents must not be served before the checkpoint_content row is durable"
        );
        assert!(
            checkpoint_store
                .get_checkpoint_contents(&summary.contents_digest)
                .unwrap()
                .is_none()
        );

        builder
            .write_checkpoints(details.checkpoint_height, new_checkpoints)
            .await
            .unwrap();

        assert!(
            checkpoint_store
                .get_checkpoint_contents(&summary.contents_digest)
                .unwrap()
                .is_some()
        );
        let cached = checkpoint_store
            .get_full_checkpoint_contents_by_sequence_number(summary.sequence_number)
            .expect("builder should cache full contents once the row is durable");
        assert_eq!(
            cached.checkpoint_contents().digest(),
            summary.contents_digest
        );
    }

    #[sim_test]
    pub async fn checkpoint_builder_test() {
        telemetry_subscribers::init_for_testing();

        let mut protocol_config =
            ProtocolConfig::get_for_version(ProtocolVersion::max(), Chain::Unknown);
        protocol_config.set_min_checkpoint_interval_ms_for_testing(100);
        // This test exercises the strict adjacent-checkpoint interval.
        protocol_config.disable_checkpoint_rate_window_size_for_testing();
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(protocol_config)
            .build()
            .await;

        // Build distinct genesis transactions and assign their digests to
        // indices so that, within any pending checkpoint, digest order matches
        // index order: `CausalOrder` orders non-dependent transactions by
        // digest, and the assertions below rely on that order. Transactions
        // 15..20 carry a large payload to exercise size-based checkpoint
        // splitting.
        let make_tx = |seed: u8, payload_size: usize| {
            let mut id = [0u8; 32];
            id[0] = seed;
            VerifiedTransaction::new_genesis_transaction(
                vec![GenesisObject::new(
                    ObjectData::Package(
                        MovePackage::new(
                            ObjectId::new(id),
                            Version::default(),
                            BTreeMap::from([(
                                Identifier::new_unchecked("m"),
                                vec![0u8; payload_size],
                            )]),
                            100_000,
                            // no modules so empty type_origin_table as no types are defined in
                            // this package
                            Vec::new(),
                            // no modules so empty linkage_table as no dependencies of this package
                            // exist
                            BTreeMap::new(),
                        )
                        .unwrap(),
                    ),
                    Owner::Immutable,
                )],
                vec![],
            )
        };

        let mut small: Vec<_> = (0..15).map(|seed| make_tx(seed, 1)).collect();
        small.sort_by_key(|tx| *tx.digest());

        let mut large: Vec<_> = (0..5).map(|seed| make_tx(100 + seed, 40000)).collect();
        large.sort_by_key(|tx| *tx.digest());

        let txns: Vec<_> = small.into_iter().chain(large).collect();
        let digests: Vec<TransactionDigest> = txns.iter().map(|tx| *tx.digest()).collect();

        // Digest for test index `i`; ascending within the small (0..15) and
        // large (15..20) pools, so index order implies digest order per pool.
        let d = |i: u8| digests[i as usize];

        for (tx, digest) in txns.iter().zip(&digests) {
            state
                .database_for_testing()
                .perpetual_tables
                .transactions
                .insert(digest, tx.serializable_ref())
                .unwrap();
        }

        let mut store = HashMap::<TransactionDigest, TransactionEffects>::new();
        commit_cert_for_test(
            &mut store,
            state.clone(),
            d(1),
            vec![d(2), d(3)],
            GasCostSummary::new(11, 11, 12, 11, 1),
        );
        commit_cert_for_test(
            &mut store,
            state.clone(),
            d(2),
            vec![d(3), d(4)],
            GasCostSummary::new(21, 21, 22, 21, 1),
        );
        commit_cert_for_test(
            &mut store,
            state.clone(),
            d(3),
            vec![],
            GasCostSummary::new(31, 31, 32, 31, 1),
        );
        commit_cert_for_test(
            &mut store,
            state.clone(),
            d(4),
            vec![],
            GasCostSummary::new(41, 41, 42, 41, 1),
        );
        for i in [5, 6, 7, 10, 11, 12, 13] {
            commit_cert_for_test(
                &mut store,
                state.clone(),
                d(i),
                vec![],
                GasCostSummary::new(41, 41, 42, 41, 1),
            );
        }
        for i in [15, 16, 17] {
            commit_cert_for_test(
                &mut store,
                state.clone(),
                d(i),
                vec![],
                GasCostSummary::new(51, 51, 52, 51, 1),
            );
        }
        let all_digests: Vec<_> = store.keys().copied().collect();
        for digest in all_digests {
            let signature = iota_types::crypto::zero_ed25519_signature().into();
            state
                .epoch_store_for_testing()
                .test_insert_user_signature(digest, vec![signature]);
        }

        let (output, mut result) = mpsc::channel::<(CheckpointContents, CheckpointSummary)>(10);
        let (certified_output, mut certified_result) =
            mpsc::channel::<CertifiedCheckpointSummary>(10);
        let store = Arc::new(store);

        let tmp_dir = iota_common::tempdir();
        let checkpoint_store = CheckpointStore::new(tmp_dir.path());
        let epoch_store = state.epoch_store_for_testing();

        let global_state_hasher = Arc::new(GlobalStateHasher::new_for_tests(
            state.get_global_state_hash_store().clone(),
        ));

        let checkpoint_service = CheckpointService::build(
            state.clone(),
            checkpoint_store.clone(),
            epoch_store.clone(),
            store,
            Arc::downgrade(&global_state_hasher),
            Box::new(output),
            Box::new(certified_output),
            CheckpointMetrics::new_for_tests(),
            3,
            100_000,
        );
        let _tasks = checkpoint_service.spawn(None).await;

        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(0, vec![d(4)], 0))
            .unwrap();
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(1, vec![d(1), d(3)], 2000))
            .unwrap();
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(
                &epoch_store,
                p(2, vec![d(10), d(11), d(12), d(13)], 3000),
            )
            .unwrap();
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(
                &epoch_store,
                p(3, vec![d(15), d(16), d(17)], 4000),
            )
            .unwrap();
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(4, vec![d(5)], 4001))
            .unwrap();
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(5, vec![d(6)], 5000))
            .unwrap();

        let (c1c, c1s) = result.recv().await.unwrap();
        let (c2c, c2s) = result.recv().await.unwrap();

        let c1t = c1c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        let c2t = c2c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        assert_eq!(c1t, vec![d(4)]);
        assert_eq!(c1s.previous_digest, None);
        assert_eq!(c1s.sequence_number, 0);
        assert_eq!(
            c1s.epoch_rolling_gas_cost_summary,
            GasCostSummary::new(41, 41, 42, 41, 1)
        );

        assert_eq!(c2t, vec![d(3), d(2), d(1)]);
        assert_eq!(c2s.previous_digest, Some(c1s.digest()));
        assert_eq!(c2s.sequence_number, 1);
        assert_eq!(
            c2s.epoch_rolling_gas_cost_summary,
            GasCostSummary::new(104, 104, 108, 104, 4)
        );

        // The builder caches the full contents of each locally built checkpoint.
        // The cached contents must match the checkpoint's transactions and hash
        // to its content digest, so state-sync peers can be served from memory.
        for (summary, digest_contents) in [(&c1s, &c1c), (&c2s, &c2c)] {
            let cached = checkpoint_store
                .get_full_checkpoint_contents_by_sequence_number(summary.sequence_number)
                .expect("builder should cache full contents of a locally built checkpoint");
            assert_eq!(&cached.checkpoint_contents(), digest_contents);
            assert_eq!(
                cached.checkpoint_contents().digest(),
                summary.contents_digest
            );
            // A cached entry must imply a durable checkpoint_content row:
            // state-sync skips its own durable write when contents are
            // already available.
            assert!(
                checkpoint_store
                    .get_checkpoint_contents(&summary.contents_digest)
                    .unwrap()
                    .is_some()
            );
        }

        // Pending at index 2 had 4 transactions, and we configured 3 transactions max.
        // Verify that we split into 2 checkpoints.
        let (c3c, c3s) = result.recv().await.unwrap();
        let c3t = c3c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        let (c4c, c4s) = result.recv().await.unwrap();
        let c4t = c4c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        assert_eq!(c3s.sequence_number, 2);
        assert_eq!(c3s.previous_digest, Some(c2s.digest()));
        assert_eq!(c4s.sequence_number, 3);
        assert_eq!(c4s.previous_digest, Some(c3s.digest()));
        assert_eq!(c3t, vec![d(10), d(11), d(12)]);
        assert_eq!(c4t, vec![d(13)]);

        // Pending at index 3 had 3 transactions of 40K size, and we configured 100K
        // max. Verify that we split into 2 checkpoints.
        let (c5c, c5s) = result.recv().await.unwrap();
        let c5t = c5c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        let (c6c, c6s) = result.recv().await.unwrap();
        let c6t = c6c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        assert_eq!(c5s.sequence_number, 4);
        assert_eq!(c5s.previous_digest, Some(c4s.digest()));
        assert_eq!(c6s.sequence_number, 5);
        assert_eq!(c6s.previous_digest, Some(c5s.digest()));
        assert_eq!(c5t, vec![d(15), d(16)]);
        assert_eq!(c6t, vec![d(17)]);

        // Pending at index 4 was too soon after the prior one and should be coalesced
        // into the next one.
        let (c7c, c7s) = result.recv().await.unwrap();
        let c7t = c7c.iter().map(|d| d.transaction).collect::<Vec<_>>();
        assert_eq!(c7t, vec![d(5), d(6)]);
        assert_eq!(c7s.previous_digest, Some(c6s.digest()));
        assert_eq!(c7s.sequence_number, 6);

        let c1ss = SignedCheckpointSummary::new(c1s.epoch, c1s, state.secret.deref(), state.name);
        let c2ss = SignedCheckpointSummary::new(c2s.epoch, c2s, state.secret.deref(), state.name);

        checkpoint_service
            .notify_checkpoint_signature(
                &epoch_store,
                &CheckpointSignatureMessage { summary: c2ss },
            )
            .unwrap();
        checkpoint_service
            .notify_checkpoint_signature(
                &epoch_store,
                &CheckpointSignatureMessage { summary: c1ss },
            )
            .unwrap();

        let c1sc = certified_result.recv().await.unwrap();
        let c2sc = certified_result.recv().await.unwrap();
        assert_eq!(c1sc.sequence_number, 0);
        assert_eq!(c2sc.sequence_number, 1);
    }

    #[sim_test]
    pub async fn checkpoint_builder_windowed_interval_test() {
        telemetry_subscribers::init_for_testing();

        // 100ms interval with a window of 3 checkpoints: a checkpoint is built
        // when either 100ms passed since the previous one or the checkpoint 3
        // back is >= 300ms older. The windowed arm permits sub-interval bursts
        // while banked budget lasts; the adjacent arm caps every quiet gap at
        // one interval regardless of window state.
        let mut protocol_config =
            ProtocolConfig::get_for_version(ProtocolVersion::max(), Chain::Unknown);
        protocol_config.set_min_checkpoint_interval_ms_for_testing(100);
        protocol_config.set_checkpoint_rate_window_size_for_testing(3);
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(protocol_config)
            .build()
            .await;

        // Distinct real transactions: only build timing is under test, but the
        // builder pairs each stored transaction with its effects by digest, so
        // effects must reference the digests of actually stored transactions.
        let make_tx = |seed: u8| {
            let mut id = [0u8; 32];
            id[0] = seed;
            VerifiedTransaction::new_genesis_transaction(
                vec![GenesisObject::new(
                    ObjectData::Package(
                        MovePackage::new(
                            ObjectId::new(id),
                            Version::default(),
                            BTreeMap::from([(Identifier::new_unchecked("m"), vec![0u8; 1])]),
                            100_000,
                            // no modules so empty type_origin_table as no types are defined in
                            // this package
                            Vec::new(),
                            // no modules so empty linkage_table as no dependencies of this package
                            // exist
                            BTreeMap::new(),
                        )
                        .unwrap(),
                    ),
                    Owner::Immutable,
                )],
                vec![],
            )
        };
        let txns: Vec<_> = (1..=8).map(make_tx).collect();
        let digests: Vec<TransactionDigest> = txns.iter().map(|tx| *tx.digest()).collect();
        // Digest for test index `i` (1-based).
        let d = |i: u8| digests[(i - 1) as usize];

        for tx in &txns {
            state
                .database_for_testing()
                .perpetual_tables
                .transactions
                .insert(tx.digest(), tx.serializable_ref())
                .unwrap();
        }

        let mut store = HashMap::<TransactionDigest, TransactionEffects>::new();
        for i in 1..=8 {
            commit_cert_for_test(
                &mut store,
                state.clone(),
                d(i),
                vec![],
                GasCostSummary::new(11, 11, 12, 11, 1),
            );
        }
        let all_digests: Vec<_> = store.keys().copied().collect();
        for digest in all_digests {
            let signature = iota_types::crypto::zero_ed25519_signature().into();
            state
                .epoch_store_for_testing()
                .test_insert_user_signature(digest, vec![signature]);
        }

        let (output, mut result) = mpsc::channel::<(CheckpointContents, CheckpointSummary)>(10);
        let (certified_output, _certified_result) = mpsc::channel::<CertifiedCheckpointSummary>(10);
        let store = Arc::new(store);

        let tmp_dir = iota_common::tempdir();
        let checkpoint_store = CheckpointStore::new(tmp_dir.path());
        let epoch_store = state.epoch_store_for_testing();

        let global_state_hasher = Arc::new(GlobalStateHasher::new_for_tests(
            state.get_global_state_hash_store().clone(),
        ));

        let checkpoint_service = CheckpointService::build(
            state.clone(),
            checkpoint_store,
            epoch_store.clone(),
            store,
            Arc::downgrade(&global_state_hasher),
            Box::new(output),
            Box::new(certified_output),
            CheckpointMetrics::new_for_tests(),
            3,
            100_000,
        );
        let _tasks = checkpoint_service.spawn(None).await;

        // The windowed arm is inert until 3 checkpoints exist; the first three
        // build via the adjacent arm, which their spread-out timestamps
        // (0, 1000, 2000) satisfy while accruing burst budget.
        // window_ms = 3 * 100 = 300.
        for (height, root, timestamp_ms) in [(0, 1u8, 0), (1, 2, 1000), (2, 3, 2000)] {
            checkpoint_service
                .write_and_notify_checkpoint_for_testing(
                    &epoch_store,
                    p(height, vec![d(root)], timestamp_ms),
                )
                .unwrap();
        }
        // p3 @ 2010: adjacent red (< C2 + 100 = 2100) but windowed green
        // (>= C0(0) + 300) -> builds, only 10ms after C2. A strict adjacent
        // rule would instead coalesce it.
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(3, vec![d(4)], 2010))
            .unwrap();
        // p4 @ 2020: windowed green (>= C1(1000) + 300) -> builds, another
        // sub-interval burst.
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(4, vec![d(5)], 2020))
            .unwrap();
        // Budget now spent: the window covers the tight 2000-2020 cluster.
        // p5 @ 2030: adjacent red (< C4 + 100 = 2120), windowed red
        // (< C2(2000) + 300 = 2300) -> coalesced.
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(5, vec![d(6)], 2030))
            .unwrap();
        // p6 @ 2130: windowed still red (< 2300), but adjacent green
        // (>= 2120) -> builds, coalescing p5 and p6. A window-only gate would
        // keep coalescing until 2300; the adjacent arm caps the gap.
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(6, vec![d(7)], 2130))
            .unwrap();
        // p7 @ 2300: windowed red (< C3(2010) + 300 = 2310), adjacent green
        // (>= C5 + 100 = 2230) -> builds.
        checkpoint_service
            .write_and_notify_checkpoint_for_testing(&epoch_store, p(7, vec![d(8)], 2300))
            .unwrap();

        let mut built = Vec::new();
        for _ in 0..7 {
            let (contents, summary) = result.recv().await.unwrap();
            built.push((summary.sequence_number, contents.iter().count()));
        }

        // Seven checkpoints, contiguous sequence numbers.
        let sequence_numbers: Vec<_> = built.iter().map(|(seq, _)| *seq).collect();
        assert_eq!(sequence_numbers, vec![0, 1, 2, 3, 4, 5, 6]);
        // Bursts (seq 3, 4) stand alone despite sub-interval spacing; once the
        // budget is spent, pendings coalesce only until the adjacent interval
        // elapses (seq 5), never longer. A strict adjacent rule would build
        // [0,1,2] and then coalesce the whole 2010-2130 cluster into one
        // checkpoint; a window-only rule would coalesce p5-p7 into one
        // checkpoint at 2300.
        let sizes: Vec<_> = built.iter().map(|(_, size)| *size).collect();
        assert_eq!(sizes, vec![1, 1, 1, 1, 1, 2, 1]);
    }

    impl TransactionCacheRead for HashMap<TransactionDigest, TransactionEffects> {
        fn try_notify_read_executed_effects(
            &self,
            _: &str,
            digests: &[TransactionDigest],
        ) -> BoxFuture<'_, IotaResult<Vec<TransactionEffects>>> {
            std::future::ready(Ok(digests
                .iter()
                .map(|d| self.get(d).expect("effects not found").clone())
                .collect()))
            .boxed()
        }

        fn try_notify_read_executed_effects_digests(
            &self,
            _: &str,
            digests: &[TransactionDigest],
        ) -> BoxFuture<'_, IotaResult<Vec<TransactionEffectsDigest>>> {
            std::future::ready(Ok(digests
                .iter()
                .map(|d| {
                    self.get(d)
                        .map(|fx| fx.digest())
                        .expect("effects not found")
                })
                .collect()))
            .boxed()
        }

        fn try_multi_get_executed_effects(
            &self,
            digests: &[TransactionDigest],
        ) -> IotaResult<Vec<Option<TransactionEffects>>> {
            Ok(digests.iter().map(|d| self.get(d).cloned()).collect())
        }

        // Unimplemented methods - its unfortunate to have this big blob of useless
        // code, but it wasn't worth it to keep EffectsNotifyRead around just
        // for these tests, as it caused a ton of complication in non-test code.
        // (e.g. had to implement EFfectsNotifyRead for all ExecutionCacheRead
        // implementors).

        fn try_multi_get_transaction_blocks(
            &self,
            _: &[TransactionDigest],
        ) -> IotaResult<Vec<Option<Arc<VerifiedTransaction>>>> {
            unimplemented!()
        }

        fn try_multi_get_executed_effects_digests(
            &self,
            _: &[TransactionDigest],
        ) -> IotaResult<Vec<Option<TransactionEffectsDigest>>> {
            unimplemented!()
        }

        fn try_multi_get_effects(
            &self,
            _: &[TransactionEffectsDigest],
        ) -> IotaResult<Vec<Option<TransactionEffects>>> {
            unimplemented!()
        }

        fn try_multi_get_events(
            &self,
            _: &[TransactionDigest],
        ) -> IotaResult<Vec<Option<TransactionEvents>>> {
            unimplemented!()
        }
    }

    #[async_trait::async_trait]
    impl CheckpointOutput for mpsc::Sender<(CheckpointContents, CheckpointSummary)> {
        async fn checkpoint_created(
            &self,
            summary: &CheckpointSummary,
            contents: &CheckpointContents,
            _epoch_store: &Arc<AuthorityPerEpochStore>,
            _checkpoint_store: &Arc<CheckpointStore>,
        ) -> IotaResult {
            self.try_send((contents.clone(), summary.clone())).unwrap();
            Ok(())
        }
    }

    #[async_trait::async_trait]
    impl CertifiedCheckpointOutput for mpsc::Sender<CertifiedCheckpointSummary> {
        async fn certified_checkpoint_created(
            &self,
            summary: &CertifiedCheckpointSummary,
        ) -> IotaResult {
            self.try_send(summary.clone()).unwrap();
            Ok(())
        }
    }

    fn p(i: u64, roots: Vec<TransactionDigest>, timestamp_ms: u64) -> PendingCheckpoint {
        PendingCheckpoint::V1(PendingCheckpointContentsV1 {
            roots: roots.into_iter().map(TransactionKey::Digest).collect(),
            details: PendingCheckpointInfo {
                timestamp_ms,
                last_of_epoch: false,
                checkpoint_height: i,
            },
        })
    }

    fn e(
        transaction_digest: TransactionDigest,
        dependencies: Vec<TransactionDigest>,
        gas_cost_summary: GasCostSummary,
    ) -> TransactionEffects {
        let mut effects = TransactionEffects::new_empty_v1_for_testing(transaction_digest);
        *effects.dependencies_mut_for_testing() = dependencies;
        *effects.gas_cost_summary_mut_for_testing() = gas_cost_summary;
        effects
    }

    fn commit_cert_for_test(
        store: &mut HashMap<TransactionDigest, TransactionEffects>,
        state: Arc<AuthorityState>,
        digest: TransactionDigest,
        dependencies: Vec<TransactionDigest>,
        gas_cost_summary: GasCostSummary,
    ) {
        let epoch_store = state.epoch_store_for_testing();
        let effects = e(digest, dependencies, gas_cost_summary);
        store.insert(digest, effects);
        epoch_store
            .insert_tx_key_and_digest(&TransactionKey::Digest(digest), &digest)
            .expect("Inserting cert fx and sigs should not fail");
    }
}