espresso_types/v0/

traits.rs

//! This module contains all the traits used for building the sequencer types.
//! It also includes some trait implementations that cannot be implemented in an external crate.
#[cfg(feature = "node")]
use std::{cmp::max, collections::BTreeMap};
use std::{fmt::Debug, ops::Range, sync::Arc, time::Duration};

use alloy::primitives::Address;
#[cfg(feature = "node")]
use anyhow::{Context, bail, ensure};
use async_trait::async_trait;
use committable::Commitment;
use futures::{FutureExt, TryFutureExt};
#[cfg(feature = "node")]
use hotshot_new_protocol::storage::NewProtocolStorage;
#[cfg(feature = "node")]
use hotshot_types::simple_certificate::{Certificate1, Certificate2};
#[cfg(feature = "node")]
use hotshot_types::{
    data::{DaProposal, DaProposal2, QuorumProposalWrapper, VidCommitment, VidDisperseShare},
    drb::DrbInput,
    event::{HotShotAction, LeafInfo},
    initializer::{HotShotInitializer, InitializerEpochInfo},
    message::Proposal,
    simple_certificate::{
        CertificatePair, NextEpochQuorumCertificate2, QuorumCertificate2, UpgradeCertificate,
    },
    traits::{
        ValidatedState as HotShotState, metrics::Metrics, node_implementation::NodeType,
        storage::Storage,
    },
    vote::HasViewNumber,
};
use hotshot_types::{
    data::{EpochNumber, ViewNumber},
    drb::DrbResult,
    epoch_membership::EpochMembershipCoordinator,
    new_protocol::CoordinatorEvent,
    simple_certificate::LightClientStateUpdateCertificateV2,
};
use indexmap::IndexMap;
use serde::{Serialize, de::DeserializeOwned};
#[cfg(feature = "node")]
use versions::{NEW_PROTOCOL_VERSION, Upgrade};

use super::{
    impls::NodeState,
    utils::BackoffParams,
    v0_3::{EventKey, IndexedStake, StakeTableEvent},
};
use crate::{
    AuthenticatedValidatorMap, BlockMerkleTree, FeeAccount, FeeAccountProof, FeeMerkleCommitment,
    Header, Leaf2, PubKey, SeqTypes,
    v0::impls::StakeTableHash,
    v0_3::{
        ChainConfig, RegisteredValidator, RewardAccountProofV1, RewardAccountV1, RewardAmount,
        RewardMerkleCommitmentV1,
    },
    v0_4::{PermittedRewardMerkleTreeV2, RewardAccountV2, RewardMerkleCommitmentV2},
};
#[cfg(feature = "node")]
use crate::{NetworkConfig, v0::impls::ValidatedState};

#[async_trait]
pub trait StateCatchup: Send + Sync {
    /// Fetch the leaf at the given height without retrying on transient errors.
    ///
    /// `coordinator` resolves the stake tables used to verify the fetched leaf
    /// chain, triggering catchup for epochs whose stake table is not yet
    /// available.
    async fn try_fetch_leaf(
        &self,
        retry: usize,
        coordinator: EpochMembershipCoordinator<SeqTypes>,
        height: u64,
    ) -> anyhow::Result<Leaf2>;

    /// Fetch the leaf at the given height, retrying on transient errors.
    async fn fetch_leaf(
        &self,
        coordinator: EpochMembershipCoordinator<SeqTypes>,
        height: u64,
    ) -> anyhow::Result<Leaf2> {
        self.backoff()
            .retry(self, |provider, retry| {
                let coordinator = coordinator.clone();
                async move { provider.try_fetch_leaf(retry, coordinator, height).await }.boxed()
            })
            .await
    }

    /// Fetch the given list of accounts without retrying on transient errors.
    async fn try_fetch_accounts(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        fee_merkle_tree_root: FeeMerkleCommitment,
        accounts: &[FeeAccount],
    ) -> anyhow::Result<Vec<FeeAccountProof>>;

    /// Fetch the given list of accounts, retrying on transient errors.
    async fn fetch_accounts(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        fee_merkle_tree_root: FeeMerkleCommitment,
        accounts: Vec<FeeAccount>,
    ) -> anyhow::Result<Vec<FeeAccountProof>> {
        self.backoff()
            .retry(self, |provider, retry| {
                let accounts = &accounts;
                async move {
                    provider
                        .try_fetch_accounts(
                            retry,
                            instance,
                            height,
                            view,
                            fee_merkle_tree_root,
                            accounts,
                        )
                        .await
                        .map_err(|err| {
                            err.context(format!(
                                "fetching accounts {accounts:?}, height {height}, view {view}"
                            ))
                        })
                }
                .boxed()
            })
            .await
    }

    /// Fetch and remember the blocks frontier without retrying on transient errors.
    async fn try_remember_blocks_merkle_tree(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        mt: &mut BlockMerkleTree,
    ) -> anyhow::Result<()>;

    /// Fetch and remember the blocks frontier, retrying on transient errors.
    async fn remember_blocks_merkle_tree(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        mt: &mut BlockMerkleTree,
    ) -> anyhow::Result<()> {
        self.backoff()
            .retry(mt, |mt, retry| {
                self.try_remember_blocks_merkle_tree(retry, instance, height, view, mt)
                    .map_err(|err| err.context(format!("fetching frontier using {}", self.name())))
                    .boxed()
            })
            .await
    }

    /// Fetch the chain config without retrying on transient errors.
    async fn try_fetch_chain_config(
        &self,
        retry: usize,
        commitment: Commitment<ChainConfig>,
    ) -> anyhow::Result<ChainConfig>;

    /// Fetch the chain config, retrying on transient errors.
    async fn fetch_chain_config(
        &self,
        commitment: Commitment<ChainConfig>,
    ) -> anyhow::Result<ChainConfig> {
        self.backoff()
            .retry(self, |provider, retry| {
                provider
                    .try_fetch_chain_config(retry, commitment)
                    .map_err(|err| err.context("fetching chain config"))
                    .boxed()
            })
            .await
    }

    /// Fetch the given reward merkle tree without retrying on transient errors.
    async fn try_fetch_reward_merkle_tree_v2(
        &self,
        retry: usize,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV2,
        accounts: Arc<Vec<RewardAccountV2>>,
    ) -> anyhow::Result<PermittedRewardMerkleTreeV2>;

    async fn fetch_reward_merkle_tree_v2(
        &self,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV2,
        accounts: Arc<Vec<RewardAccountV2>>,
    ) -> anyhow::Result<PermittedRewardMerkleTreeV2> {
        self.backoff()
            .retry(self, |provider, retry| {
                let accounts = accounts.clone();
                async move {
                    provider
                        .try_fetch_reward_merkle_tree_v2(
                            retry,
                            height,
                            view,
                            reward_merkle_tree_root,
                            accounts,
                        )
                        .await
                        .map_err(|err| {
                            err.context(format!("fetching reward merkle tree for height {height}"))
                        })
                }
                .boxed()
            })
            .await
    }

    /// Fetch the given list of reward accounts without retrying on transient errors.
    async fn try_fetch_reward_accounts_v1(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV1,
        accounts: &[RewardAccountV1],
    ) -> anyhow::Result<Vec<RewardAccountProofV1>>;

    /// Fetch the given list of reward accounts, retrying on transient errors.
    async fn fetch_reward_accounts_v1(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV1,
        accounts: Vec<RewardAccountV1>,
    ) -> anyhow::Result<Vec<RewardAccountProofV1>> {
        self.backoff()
            .retry(self, |provider, retry| {
                let accounts = &accounts;
                async move {
                    provider
                        .try_fetch_reward_accounts_v1(
                            retry,
                            instance,
                            height,
                            view,
                            reward_merkle_tree_root,
                            accounts,
                        )
                        .await
                        .map_err(|err| {
                            err.context(format!(
                                "fetching v1 reward accounts {accounts:?}, height {height}, view \
                                 {view}"
                            ))
                        })
                }
                .boxed()
            })
            .await
    }

    /// Fetch the state certificate for a given epoch without retrying on transient errors.
    async fn try_fetch_state_cert(
        &self,
        retry: usize,
        epoch: u64,
    ) -> anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>>;

    /// Fetch the state certificate for a given epoch, retrying on transient errors.
    async fn fetch_state_cert(
        &self,
        epoch: u64,
    ) -> anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>> {
        self.backoff()
            .retry(self, |provider, retry| {
                provider
                    .try_fetch_state_cert(retry, epoch)
                    .map_err(|err| err.context(format!("fetching state cert for epoch {epoch}")))
                    .boxed()
            })
            .await
    }

    /// Returns true if the catchup provider is local (e.g. does not make calls to remote resources).
    fn is_local(&self) -> bool;

    /// Returns the backoff parameters for the catchup provider.
    fn backoff(&self) -> &BackoffParams;

    /// Returns the name of the catchup provider.
    fn name(&self) -> String;

    /// A copy of this provider that waits up to `local_timeout` on its local providers, or `None`
    /// if it has no such timeout.
    fn with_local_timeout(&self, _local_timeout: Duration) -> Option<Arc<dyn StateCatchup>> {
        None
    }
}

#[async_trait]
impl<T: StateCatchup + ?Sized> StateCatchup for Arc<T> {
    async fn try_fetch_leaf(
        &self,
        retry: usize,
        coordinator: EpochMembershipCoordinator<SeqTypes>,
        height: u64,
    ) -> anyhow::Result<Leaf2> {
        (**self).try_fetch_leaf(retry, coordinator, height).await
    }

    async fn fetch_leaf(
        &self,
        coordinator: EpochMembershipCoordinator<SeqTypes>,
        height: u64,
    ) -> anyhow::Result<Leaf2> {
        (**self).fetch_leaf(coordinator, height).await
    }

    async fn try_fetch_accounts(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        fee_merkle_tree_root: FeeMerkleCommitment,
        accounts: &[FeeAccount],
    ) -> anyhow::Result<Vec<FeeAccountProof>> {
        (**self)
            .try_fetch_accounts(
                retry,
                instance,
                height,
                view,
                fee_merkle_tree_root,
                accounts,
            )
            .await
    }

    async fn fetch_accounts(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        fee_merkle_tree_root: FeeMerkleCommitment,
        accounts: Vec<FeeAccount>,
    ) -> anyhow::Result<Vec<FeeAccountProof>> {
        (**self)
            .fetch_accounts(instance, height, view, fee_merkle_tree_root, accounts)
            .await
    }

    async fn try_remember_blocks_merkle_tree(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        mt: &mut BlockMerkleTree,
    ) -> anyhow::Result<()> {
        (**self)
            .try_remember_blocks_merkle_tree(retry, instance, height, view, mt)
            .await
    }

    async fn remember_blocks_merkle_tree(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        mt: &mut BlockMerkleTree,
    ) -> anyhow::Result<()> {
        (**self)
            .remember_blocks_merkle_tree(instance, height, view, mt)
            .await
    }

    async fn try_fetch_chain_config(
        &self,
        retry: usize,
        commitment: Commitment<ChainConfig>,
    ) -> anyhow::Result<ChainConfig> {
        (**self).try_fetch_chain_config(retry, commitment).await
    }

    async fn fetch_chain_config(
        &self,
        commitment: Commitment<ChainConfig>,
    ) -> anyhow::Result<ChainConfig> {
        (**self).fetch_chain_config(commitment).await
    }

    async fn try_fetch_reward_merkle_tree_v2(
        &self,
        retry: usize,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV2,
        accounts: Arc<Vec<RewardAccountV2>>,
    ) -> anyhow::Result<PermittedRewardMerkleTreeV2> {
        (**self)
            .try_fetch_reward_merkle_tree_v2(retry, height, view, reward_merkle_tree_root, accounts)
            .await
    }

    async fn fetch_reward_merkle_tree_v2(
        &self,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV2,
        accounts: Arc<Vec<RewardAccountV2>>,
    ) -> anyhow::Result<PermittedRewardMerkleTreeV2> {
        (**self)
            .fetch_reward_merkle_tree_v2(height, view, reward_merkle_tree_root, accounts)
            .await
    }

    async fn try_fetch_reward_accounts_v1(
        &self,
        retry: usize,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV1,
        accounts: &[RewardAccountV1],
    ) -> anyhow::Result<Vec<RewardAccountProofV1>> {
        (**self)
            .try_fetch_reward_accounts_v1(
                retry,
                instance,
                height,
                view,
                reward_merkle_tree_root,
                accounts,
            )
            .await
    }

    async fn fetch_reward_accounts_v1(
        &self,
        instance: &NodeState,
        height: u64,
        view: ViewNumber,
        reward_merkle_tree_root: RewardMerkleCommitmentV1,
        accounts: Vec<RewardAccountV1>,
    ) -> anyhow::Result<Vec<RewardAccountProofV1>> {
        (**self)
            .fetch_reward_accounts_v1(instance, height, view, reward_merkle_tree_root, accounts)
            .await
    }

    async fn try_fetch_state_cert(
        &self,
        retry: usize,
        epoch: u64,
    ) -> anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>> {
        (**self).try_fetch_state_cert(retry, epoch).await
    }

    async fn fetch_state_cert(
        &self,
        epoch: u64,
    ) -> anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>> {
        (**self).fetch_state_cert(epoch).await
    }

    fn backoff(&self) -> &BackoffParams {
        (**self).backoff()
    }

    fn name(&self) -> String {
        (**self).name()
    }

    fn is_local(&self) -> bool {
        (**self).is_local()
    }

    fn with_local_timeout(&self, local_timeout: Duration) -> Option<Arc<dyn StateCatchup>> {
        (**self).with_local_timeout(local_timeout)
    }
}

#[cfg(feature = "node")]
#[async_trait]
pub trait PersistenceOptions: Clone + Send + Sync + Debug + 'static {
    type Persistence: SequencerPersistence + MembershipPersistence;

    fn set_view_retention(&mut self, view_retention: u64);
    fn set_consensus_only(&mut self);
    async fn create(&mut self) -> anyhow::Result<Self::Persistence>;
    async fn reset(self) -> anyhow::Result<()>;
}

/// Determine the read state based on the queried block range.
// - If the persistence returned events up to the requested block, the read is complete.
/// - Otherwise, indicate that the read is up to the last processed block.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum EventsPersistenceRead {
    Complete,
    UntilL1Block(u64),
}

/// Tuple type for stake table data: (validators, block_reward, stake_table_hash)
pub type StakeTuple = (
    AuthenticatedValidatorMap,
    Option<RewardAmount>,
    Option<StakeTableHash>,
);

#[async_trait]
/// Trait used by `Memberships` implementations to interact with persistence layer.
pub trait MembershipPersistence: Send + Sync + 'static {
    /// Load stake table for epoch from storage
    async fn load_stake(&self, epoch: EpochNumber) -> anyhow::Result<Option<StakeTuple>>;

    /// Load stake tables for storage for latest `n` known epochs
    async fn load_latest_stake(&self, limit: u64) -> anyhow::Result<Option<Vec<IndexedStake>>>;

    /// Load the DRB result for `epoch`.
    async fn load_drb_result(&self, epoch: EpochNumber) -> anyhow::Result<Option<DrbResult>>;

    /// Load the epoch root block header for `epoch`.
    async fn load_epoch_root(&self, epoch: EpochNumber) -> anyhow::Result<Option<Header>>;

    /// Store the epoch root block header for `epoch`.
    async fn store_epoch_root(
        &self,
        epoch: EpochNumber,
        block_header: Header,
    ) -> anyhow::Result<()>;

    /// Store stake table at `epoch` in the persistence layer
    async fn store_stake(
        &self,
        epoch: EpochNumber,
        stake: AuthenticatedValidatorMap,
        block_reward: Option<RewardAmount>,
        stake_table_hash: Option<StakeTableHash>,
    ) -> anyhow::Result<()>;

    async fn store_events(
        &self,
        l1_finalized: u64,
        events: Vec<(EventKey, StakeTableEvent)>,
    ) -> anyhow::Result<()>;
    async fn load_events(
        &self,
        from_l1_block: u64,
        l1_finalized: u64,
    ) -> anyhow::Result<(
        Option<EventsPersistenceRead>,
        Vec<(EventKey, StakeTableEvent)>,
    )>;

    /// Delete all stake table events, the L1 block tracker, and the epoch DRB and root data.
    async fn delete_stake_tables(&self) -> anyhow::Result<()>;

    async fn store_all_validators(
        &self,
        epoch: EpochNumber,
        all_validators: IndexMap<Address, RegisteredValidator<PubKey>>,
    ) -> anyhow::Result<()>;

    async fn load_all_validators(
        &self,
        epoch: EpochNumber,
        offset: u64,
        limit: u64,
    ) -> anyhow::Result<Vec<RegisteredValidator<PubKey>>>;
}

/// State certificates by epoch, apart from [`SequencerPersistence`] so that the API can hold them as
/// a trait object.
#[cfg(feature = "node")]
#[async_trait]
pub trait StateCertPersistence: Send + Sync + 'static {
    /// Get a state certificate for an epoch.
    async fn get_state_cert_by_epoch(
        &self,
        epoch: u64,
    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>>;

    /// Insert a state certificate for a given epoch.
    async fn insert_state_cert(
        &self,
        epoch: u64,
        cert: LightClientStateUpdateCertificateV2<SeqTypes>,
    ) -> anyhow::Result<()>;
}

#[cfg(feature = "node")]
#[async_trait]
pub trait SequencerPersistence:
    Sized + Send + Sync + Clone + 'static + MembershipPersistence + StateCertPersistence
{
    /// Use this storage as a state catchup backend, if supported.
    fn into_catchup_provider(
        self,
        _backoff: BackoffParams,
    ) -> anyhow::Result<Arc<dyn StateCatchup>> {
        bail!("state catchup is not implemented for this persistence type");
    }

    /// Load the orchestrator config from storage.
    ///
    /// Returns `None` if no config exists (we are joining a network for the first time). Fails with
    /// `Err` if it could not be determined whether a config exists or not.
    async fn load_config(&self) -> anyhow::Result<Option<NetworkConfig>>;

    /// Save the orchestrator config to storage.
    async fn save_config(&self, cfg: &NetworkConfig) -> anyhow::Result<()>;

    /// Load the highest view saved with [`save_voted_view`](Self::save_voted_view).
    async fn load_latest_acted_view(&self) -> anyhow::Result<Option<ViewNumber>>;

    /// Load the view to restart from.
    async fn load_restart_view(&self) -> anyhow::Result<Option<ViewNumber>>;

    /// Load the proposals saved by consensus
    async fn load_quorum_proposals(
        &self,
    ) -> anyhow::Result<BTreeMap<ViewNumber, Proposal<SeqTypes, QuorumProposalWrapper<SeqTypes>>>>;

    async fn load_quorum_proposal(
        &self,
        view: ViewNumber,
    ) -> anyhow::Result<Proposal<SeqTypes, QuorumProposalWrapper<SeqTypes>>>;

    async fn load_vid_share(
        &self,
        view: ViewNumber,
    ) -> anyhow::Result<Option<Proposal<SeqTypes, VidDisperseShare<SeqTypes>>>>;
    async fn load_da_proposal(
        &self,
        view: ViewNumber,
    ) -> anyhow::Result<Option<Proposal<SeqTypes, DaProposal2<SeqTypes>>>>;
    async fn load_upgrade_certificate(
        &self,
    ) -> anyhow::Result<Option<UpgradeCertificate<SeqTypes>>>;
    async fn load_start_epoch_info(&self) -> anyhow::Result<Vec<InitializerEpochInfo<SeqTypes>>>;
    async fn load_state_cert(
        &self,
    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>>;

    /// Load the latest known consensus state.
    ///
    /// Returns an initializer to resume HotShot from the latest saved state (or start from genesis,
    /// if there is no saved state). Also returns the anchor view number, which can be used as a
    /// reference point to process any events which were not processed before a previous shutdown,
    /// if applicable,.
    async fn load_consensus_state(
        &self,
        state: NodeState,
        upgrade: Upgrade,
    ) -> anyhow::Result<(HotShotInitializer<SeqTypes>, Option<ViewNumber>)> {
        let genesis_validated_state = ValidatedState::genesis(&state).0;
        let highest_voted_view = match self
            .load_latest_acted_view()
            .await
            .context("loading last voted view")?
        {
            Some(view) => {
                tracing::info!(?view, "starting with last actioned view");
                view
            },
            None => {
                tracing::info!("no saved view, starting from genesis");
                ViewNumber::genesis()
            },
        };

        let restart_view = match self
            .load_restart_view()
            .await
            .context("loading restart view")?
        {
            Some(view) => {
                tracing::info!(?view, "starting from saved view");
                view
            },
            None => {
                tracing::info!("no saved view, starting from genesis");
                ViewNumber::genesis()
            },
        };
        let config = self.load_config().await.context("loading config")?;
        // Use epoch height from node state. Node state gets the epoch height from the genesis file.
        let epoch_height = state.epoch_height.unwrap_or_else(|| {
            config
                .as_ref()
                .map(|c| c.config.epoch_height)
                .unwrap_or_default()
        });
        let (leaf, cert_pair, anchor_view) = match self
            .load_anchor_leaf()
            .await
            .context("loading anchor leaf")?
        {
            Some((leaf, cert_pair)) => {
                tracing::info!(?leaf, ?cert_pair, "starting from saved leaf");
                let high_qc = cert_pair.qc().clone();
                let leaf_view = leaf.view_number();
                ensure!(
                    leaf_view == high_qc.view_number,
                    format!(
                        "loaded anchor leaf from view {}, but high QC is from view {}",
                        leaf_view, high_qc.view_number
                    )
                );
                if leaf.block_header().version() < NEW_PROTOCOL_VERSION {
                    ensure!(
                        epoch_height == 0
                            || cert_pair.block_number().is_none()
                            || cert_pair.verify_next_epoch_qc(epoch_height).is_ok(),
                        format!(
                            "Next epoch QC is required but it's not present or doesn't match \
                             primary QC\nPrimary QC: {:?}\nNext epoch QC: {:?}",
                            cert_pair.qc(),
                            cert_pair.next_epoch_qc()
                        )
                    );
                }

                let anchor_view = leaf.view_number();
                (leaf, cert_pair, Some(anchor_view))
            },
            None => {
                tracing::info!("no saved leaf, starting from genesis leaf");
                let genesis_qc =
                    QuorumCertificate2::genesis(&genesis_validated_state, &state, upgrade).await;
                (
                    hotshot_types::data::Leaf2::genesis(
                        &genesis_validated_state,
                        &state,
                        upgrade.base,
                    )
                    .await,
                    CertificatePair::new(genesis_qc, None),
                    None,
                )
            },
        };

        let mut high_qc = cert_pair.qc().clone();
        if let Some((extended_high_qc, _)) = self.load_eqc().await
            && extended_high_qc.view_number() > high_qc.view_number()
        {
            high_qc = extended_high_qc;
        }

        if let Some(running_high_qc) = self
            .load_high_qc2()
            .await
            .context("loading persisted high qc")?
            && running_high_qc.view_number() > high_qc.view_number()
        {
            high_qc = running_high_qc;
        }

        let validated_state = if leaf.block_header().height() == 0 {
            // If we are starting from genesis, we can provide the full state.
            genesis_validated_state
        } else {
            // Otherwise, we will have to construct a sparse state and fetch missing data during
            // catchup.
            ValidatedState::from_header(leaf.block_header())
        };

        // If we are not starting from genesis, we start from the view following the maximum view
        // between `highest_voted_view` and `leaf.view_number`. This prevents double votes from
        // starting in a view in which we had already voted before the restart, and prevents
        // unnecessary catchup from starting in a view earlier than the anchor leaf.
        let restart_view = max(restart_view, leaf.view_number());

        let saved_proposals = self
            .load_quorum_proposals()
            .await
            .context("loading saved proposals")?;

        let upgrade_certificate = self
            .load_upgrade_certificate()
            .await
            .context("loading upgrade certificate")?;

        let start_epoch_info = self
            .load_start_epoch_info()
            .await
            .context("loading start epoch info")?;

        let state_cert = self
            .load_state_cert()
            .await
            .context("loading light client state update certificate")?;

        tracing::warn!(
            ?leaf,
            ?restart_view,
            ?high_qc,
            ?validated_state,
            ?state_cert,
            "loaded consensus state"
        );

        Ok((
            HotShotInitializer::load(
                state,
                epoch_height,
                start_epoch_info,
                (leaf, Some(Arc::new(validated_state))),
                restart_view,
                high_qc,
                highest_voted_view,
                saved_proposals,
                upgrade_certificate,
                state_cert,
            ),
            anchor_view,
        ))
    }

    /// Decode a consensus decide event and persist its leaves, for the consensus event loop.
    /// Returns the decided view on a decide so the caller can wake a background task to run
    /// [`process_decided_events`](Self::process_decided_events); `None` otherwise.
    ///
    /// This is the persist-only half of a decide: query-service ingestion and GC are deferred to
    /// [`process_decided_events`](Self::process_decided_events). Tests that want the synchronous
    /// persist-then-process behavior use [`append_decided_leaves`](Self::append_decided_leaves).
    async fn persist_event(
        &self,
        event: &CoordinatorEvent<SeqTypes>,
        consumer: &(impl EventConsumer + 'static),
    ) -> Option<ViewNumber> {
        match event {
            CoordinatorEvent::NewDecide {
                leaf_infos,
                cert1,
                cert2,
            } => {
                let first = leaf_infos.first()?;
                let decided_view = first.leaf.view_number();

                // `cert1` certifies the newest leaf; each newer leaf's justify_qc certifies the
                // next older leaf.
                let certifying_qcs = std::iter::once(cert1.clone())
                    .chain(leaf_infos.iter().map(|info| info.leaf.justify_qc().clone()))
                    .take(leaf_infos.len())
                    .map(CertificatePair::non_epoch_change);

                if let Err(err) = self
                    .persist_decided_leaves(
                        leaf_infos.iter().zip(certifying_qcs),
                        cert2.as_ref(),
                        consumer,
                    )
                    .await
                {
                    tracing::error!(
                        "failed to save decided leaves from new protocol, chain may not be up to \
                         date: {err:#}"
                    );
                    return None;
                }
                Some(decided_view)
            },
            _ => None,
        }
    }

    /// Append decided leaves to persistent storage and emit a corresponding event.
    ///
    /// `consumer` will be sent a `Decide` event containing all decided leaves in persistent storage
    /// up to and including `view`. If available in persistent storage, full block payloads and VID
    /// info will also be included for each leaf.
    ///
    /// Once the new decided leaves have been processed, old data up to `view` will be garbage
    /// collected The consumer's handling of this event is a prerequisite for the completion of
    /// garbage collection: if the consumer fails to process the event, no data is deleted. This
    /// ensures that, if called repeatedly, all decided leaves ever recorded in consensus storage
    /// will eventually be passed to the consumer.
    ///
    /// Note that the converse is not true: if garbage collection fails, it is not guaranteed that
    /// the consumer hasn't processed the decide event. Thus, in rare cases, some events may be
    /// processed twice, or the consumer may get two events which share a subset of their data.
    /// Thus, it is the consumer's responsibility to make sure its handling of each leaf is
    /// idempotent.
    ///
    /// If the consumer fails to handle the new decide event, it may be retried, or simply postponed
    /// until the next decide, at which point all persisted leaves from the failed GC run will be
    /// included in the event along with subsequently decided leaves.
    ///
    /// This functionality is useful for keeping a separate view of the blockchain in sync with the
    /// consensus storage. For example, the `consumer` could be used for moving data from consensus
    /// storage to long-term archival storage.
    ///
    /// Convenience combinator: [`persist_decided_leaves`](Self::persist_decided_leaves) then
    /// [`process_decided_events`](Self::process_decided_events). Production drives the two halves on
    /// separate tasks; tests and back-compat callers use this synchronous form.
    async fn append_decided_leaves(
        &self,
        decided_view: ViewNumber,
        leaf_chain: impl IntoIterator<Item = (&LeafInfo<SeqTypes>, CertificatePair<SeqTypes>)> + Send,
        consumer: &(impl EventConsumer + 'static),
    ) -> anyhow::Result<()> {
        self.persist_decided_leaves(leaf_chain, None, consumer)
            .await?;
        // Leaves are persisted; processing failures are non-fatal here and retried in production.
        if let Err(err) = self.process_decided_events(decided_view, consumer).await {
            tracing::warn!(?decided_view, "decide event processing failed: {err:#}");
        }
        Ok(())
    }

    /// Persist decided leaves only (the critical, must-not-lag half of a decide; also the
    /// anchor for restart recovery). Query-service ingestion and GC are deferred to
    /// [`process_decided_events`](Self::process_decided_events). Backends with no replayable storage
    /// (e.g. `NoStorage`) may instead forward decide events to `consumer` here.
    ///
    /// `cert2` finalizes the newest leaf of `leaf_chain`. It is persisted atomically with the
    /// leaves, so [`process_decided_events`](Self::process_decided_events) never sees that leaf
    /// without it.
    async fn persist_decided_leaves(
        &self,
        leaf_chain: impl IntoIterator<Item = (&LeafInfo<SeqTypes>, CertificatePair<SeqTypes>)> + Send,
        cert2: Option<&Certificate2<SeqTypes>>,
        consumer: &(impl EventConsumer + 'static),
    ) -> anyhow::Result<()>;

    /// Generate decide events for `consumer` from persisted leaves, then GC processed data.
    /// Cursor-driven (e.g. `last_processed_view`): advances only on success, so it may lag
    /// consensus without losing data.
    ///
    /// Returns the highest view confirmed processed (the cursor), or `None` if nothing was
    /// processed, so the caller can track real progress. Errors are propagated; the failed range
    /// is retried on the next call.
    ///
    /// Default returns `Some(decided_view)`: backends with no replayable storage (e.g. `NoStorage`)
    /// forward events synchronously in `persist_decided_leaves` and are always caught up here.
    async fn process_decided_events(
        &self,
        decided_view: ViewNumber,
        _consumer: &(impl EventConsumer + 'static),
    ) -> anyhow::Result<Option<ViewNumber>> {
        Ok(Some(decided_view))
    }

    async fn load_anchor_leaf(&self) -> anyhow::Result<Option<(Leaf2, CertificatePair<SeqTypes>)>>;
    async fn append_vid(
        &self,
        proposal: &Proposal<SeqTypes, VidDisperseShare<SeqTypes>>,
    ) -> anyhow::Result<()>;
    async fn append_da(
        &self,
        proposal: &Proposal<SeqTypes, DaProposal<SeqTypes>>,
        vid_commit: VidCommitment,
    ) -> anyhow::Result<()>;
    async fn record_action(&self, view: ViewNumber, action: HotShotAction) -> anyhow::Result<()>;

    async fn append_quorum_proposal2(
        &self,
        proposal: &Proposal<SeqTypes, QuorumProposalWrapper<SeqTypes>>,
    ) -> anyhow::Result<()>;

    /// Persist cert2 for the given view.
    async fn append_cert2(
        &self,
        _view: ViewNumber,
        _cert2: Certificate2<SeqTypes>,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    /// Load a persisted cert2 by view, if any.
    async fn load_cert2(
        &self,
        _view: ViewNumber,
    ) -> anyhow::Result<Option<Certificate2<SeqTypes>>> {
        Ok(None)
    }

    /// Persist the new protocol's locked QC, written before each phase-2 vote.
    ///
    /// Implementations must apply this as an atomic monotonic compare-and-set: a
    /// write whose view is not newer than the stored one is a no-op. This lets
    /// concurrent, retried writes race without ever regressing the persisted lock.
    async fn append_high_qc2(&self, _high_qc: QuorumCertificate2<SeqTypes>) -> anyhow::Result<()> {
        Ok(())
    }

    /// Load the persisted locked QC, if any.
    async fn load_high_qc2(&self) -> anyhow::Result<Option<QuorumCertificate2<SeqTypes>>> {
        Ok(None)
    }

    /// Persist the new protocol's latest epoch boundary certificate: the own
    /// `Certificate1` of the last block of an epoch, at the block's view.
    ///
    /// Implementations must apply this as an atomic monotonic compare-and-set,
    /// ordered by epoch, then view, like [`Self::append_high_qc2`].
    async fn append_boundary_qc2(&self, _qc: QuorumCertificate2<SeqTypes>) -> anyhow::Result<()> {
        Ok(())
    }

    /// Load the persisted boundary certificate, if any.
    async fn load_boundary_qc2(&self) -> anyhow::Result<Option<QuorumCertificate2<SeqTypes>>> {
        Ok(None)
    }

    /// Update the current eQC in storage.
    async fn store_eqc(
        &self,
        _high_qc: QuorumCertificate2<SeqTypes>,
        _next_epoch_high_qc: NextEpochQuorumCertificate2<SeqTypes>,
    ) -> anyhow::Result<()>;

    /// Load the current eQC from storage.
    async fn load_eqc(
        &self,
    ) -> Option<(
        QuorumCertificate2<SeqTypes>,
        NextEpochQuorumCertificate2<SeqTypes>,
    )>;

    async fn store_upgrade_certificate(
        &self,
        decided_upgrade_certificate: Option<UpgradeCertificate<SeqTypes>>,
    ) -> anyhow::Result<()>;

    async fn load_anchor_view(&self) -> anyhow::Result<ViewNumber> {
        match self.load_anchor_leaf().await? {
            Some((leaf, _)) => Ok(leaf.view_number()),
            None => Ok(ViewNumber::genesis()),
        }
    }

    async fn append_next_epoch_high_qc2(
        &self,
        _next_epoch_high_qc: NextEpochQuorumCertificate2<SeqTypes>,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    async fn append_da2(
        &self,
        proposal: &Proposal<SeqTypes, DaProposal2<SeqTypes>>,
        vid_commit: VidCommitment,
    ) -> anyhow::Result<()>;

    async fn store_drb_result(
        &self,
        epoch: EpochNumber,
        drb_result: DrbResult,
    ) -> anyhow::Result<()>;
    async fn store_drb_input(&self, drb_input: DrbInput) -> anyhow::Result<()>;
    async fn load_drb_input(&self, epoch: u64) -> anyhow::Result<DrbInput>;
    async fn add_state_cert(
        &self,
        state_cert: LightClientStateUpdateCertificateV2<SeqTypes>,
    ) -> anyhow::Result<()>;

    fn enable_metrics(&mut self, metrics: &dyn Metrics);
}

#[async_trait]
pub trait EventConsumer: Debug + Send + Sync {
    async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()>;
}

#[async_trait]
impl<T> EventConsumer for Box<T>
where
    T: EventConsumer + ?Sized,
{
    async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
        (**self).handle_event(event).await
    }
}

#[derive(Clone, Copy, Debug)]
pub struct NullEventConsumer;

#[async_trait]
impl EventConsumer for NullEventConsumer {
    async fn handle_event(&self, _event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
        Ok(())
    }
}

#[cfg(feature = "node")]
#[async_trait]
impl<P: SequencerPersistence> Storage<SeqTypes> for Arc<P> {
    async fn append_vid(
        &self,
        proposal: &Proposal<SeqTypes, VidDisperseShare<SeqTypes>>,
    ) -> anyhow::Result<()> {
        (**self).append_vid(proposal).await
    }

    async fn append_da(
        &self,
        proposal: &Proposal<SeqTypes, DaProposal<SeqTypes>>,
        vid_commit: VidCommitment,
    ) -> anyhow::Result<()> {
        (**self).append_da(proposal, vid_commit).await
    }

    async fn append_da2(
        &self,
        proposal: &Proposal<SeqTypes, DaProposal2<SeqTypes>>,
        vid_commit: VidCommitment,
    ) -> anyhow::Result<()> {
        (**self).append_da2(proposal, vid_commit).await
    }

    async fn record_action(&self, view: ViewNumber, action: HotShotAction) -> anyhow::Result<()> {
        (**self).record_action(view, action).await
    }

    async fn append_proposal_wrapper(
        &self,
        proposal: &Proposal<SeqTypes, QuorumProposalWrapper<SeqTypes>>,
    ) -> anyhow::Result<()> {
        (**self).append_quorum_proposal2(proposal).await
    }

    async fn update_decided_upgrade_certificate(
        &self,
        decided_upgrade_certificate: Option<UpgradeCertificate<SeqTypes>>,
    ) -> anyhow::Result<()> {
        (**self)
            .store_upgrade_certificate(decided_upgrade_certificate)
            .await
    }

    async fn store_drb_result(
        &self,
        epoch: EpochNumber,
        drb_result: DrbResult,
    ) -> anyhow::Result<()> {
        (**self).store_drb_result(epoch, drb_result).await
    }

    async fn store_epoch_root(
        &self,
        epoch: EpochNumber,
        block_header: <SeqTypes as NodeType>::BlockHeader,
    ) -> anyhow::Result<()> {
        (**self).store_epoch_root(epoch, block_header).await
    }

    async fn store_drb_input(&self, drb_input: DrbInput) -> anyhow::Result<()> {
        (**self).store_drb_input(drb_input).await
    }

    async fn load_drb_input(&self, epoch: u64) -> anyhow::Result<DrbInput> {
        (**self).load_drb_input(epoch).await
    }

    async fn update_state_cert(
        &self,
        state_cert: LightClientStateUpdateCertificateV2<SeqTypes>,
    ) -> anyhow::Result<()> {
        (**self).add_state_cert(state_cert).await
    }
}

#[cfg(feature = "node")]
#[async_trait]
impl<P: SequencerPersistence> NewProtocolStorage<SeqTypes> for Arc<P> {
    async fn append_high_qc2(&self, high_qc: Certificate1<SeqTypes>) -> anyhow::Result<()> {
        // Writes are spawned concurrently and retried, so a stale write can land
        // after a newer one. The backend applies a monotonic compare-and-set
        // atomically, so such a stale write is a no-op and never regresses the
        // persisted view.
        (**self).append_high_qc2(high_qc).await
    }

    async fn load_high_qc2(&self) -> anyhow::Result<Option<Certificate1<SeqTypes>>> {
        (**self).load_high_qc2().await
    }

    async fn append_boundary_qc2(&self, qc: Certificate1<SeqTypes>) -> anyhow::Result<()> {
        (**self).append_boundary_qc2(qc).await
    }

    async fn load_boundary_qc2(&self) -> anyhow::Result<Option<Certificate1<SeqTypes>>> {
        (**self).load_boundary_qc2().await
    }
}

/// Data that can be deserialized from a subslice of namespace payload bytes.
///
/// Companion trait for [`NsPayloadBytesRange`], which specifies the subslice of
/// namespace payload bytes to read.
pub trait FromNsPayloadBytes<'a> {
    /// Deserialize `Self` from namespace payload bytes.
    fn from_payload_bytes(bytes: &'a [u8]) -> Self;
}

/// Specifies a subslice of namespace payload bytes to read.
///
/// Companion trait for [`FromNsPayloadBytes`], which holds data that can be
/// deserialized from that subslice of bytes.
pub trait NsPayloadBytesRange<'a> {
    type Output: FromNsPayloadBytes<'a>;

    /// Range relative to this ns payload
    fn ns_payload_range(&self) -> Range<usize>;
}

/// Types which can be deserialized from either integers or strings.
///
/// Some types can be represented as an integer or a string in human-readable formats like JSON or
/// TOML. For example, 1 GWEI might be represented by the integer `1000000000` or the string `"1
/// gwei"`. Such types can implement `FromStringOrInteger` and then use [`impl_string_or_integer`]
/// to derive this user-friendly serialization.
///
/// These types are assumed to have an efficient representation as an integral type in Rust --
/// [`Self::Binary`] -- and will be serialized to and from this type when using a non-human-readable
/// encoding. With human readable encodings, serialization is always to a string.
pub trait FromStringOrInteger: Sized {
    type Binary: Serialize + DeserializeOwned;
    type Integer: Serialize + DeserializeOwned;

    fn from_binary(b: Self::Binary) -> anyhow::Result<Self>;
    fn from_string(s: String) -> anyhow::Result<Self>;
    fn from_integer(i: Self::Integer) -> anyhow::Result<Self>;

    fn to_binary(&self) -> anyhow::Result<Self::Binary>;
    fn to_string(&self) -> anyhow::Result<String>;
}