espresso_node/

lib.rs

mod external_event_handler;
mod message_compat_tests;
mod proposal_fetcher;
mod request_response;
mod startup_catchup;

pub mod api;
pub mod catchup;
pub mod consensus_handle;
pub mod context;
pub mod follower;
pub mod genesis;
#[cfg(any(test, feature = "testing"))]
pub mod legacy_chain;
pub use espresso_keyset as keyset;
pub mod options;
pub mod persistence;
pub mod run;
pub mod state;
pub mod state_cert;
pub mod state_signature;
pub mod util;

use std::{fmt::Debug, path::PathBuf, sync::Arc, time::Duration};

use alloy::primitives::U256;
use anyhow::Context;
use async_lock::Mutex;
use catchup::{ParallelStateCatchup, StatePeers};
use context::SequencerContext;
use dyn_clone::clone_box;
use espresso_types::{
    BackoffParams, EpochCommittees, EpochRewardsCalculator, L1ClientOptions, NodeState, SeqTypes,
    ValidatedState,
    traits::{EventConsumer, MembershipPersistence},
    v0::traits::SequencerPersistence,
    v0_1::{ChainId, DECAF_CHAIN_ID, MAINNET_CHAIN_ID},
    v0_3::Fetcher,
};

pub(crate) const MAINNET_TELEMETRY_ENDPOINT: &str = "https://telemetry.main.net.espresso.network";
pub(crate) const DECAF_TELEMETRY_ENDPOINT: &str =
    "https://telemetry.decaf.testnet.espresso.network";

/// Default telemetry endpoint for known networks. Unknown chains have no
/// default; operators set `ESPRESSO_NODE_TELEMETRY_ENDPOINT` explicitly.
pub(crate) fn default_telemetry_endpoint(chain_id: ChainId) -> Option<&'static str> {
    if chain_id == MAINNET_CHAIN_ID {
        Some(MAINNET_TELEMETRY_ENDPOINT)
    } else if chain_id == DECAF_CHAIN_ID {
        Some(DECAF_TELEMETRY_ENDPOINT)
    } else {
        None
    }
}

pub use espresso_types::RECENT_STAKE_TABLES_LIMIT;
use genesis::L1Finalized;
pub use genesis::{Genesis, GenesisSource};
use hotshot_new_protocol::network::{Cliquenet, message_limit};
use hotshot_orchestrator::client::{OrchestratorClient, get_complete_config};
use hotshot_types::{
    ValidatorConfig,
    addr::NetAddr,
    data::ViewNumber,
    epoch_membership::EpochMembershipCoordinator,
    light_client::{StateKeyPair, StateSignKey},
    network::NetworkConfig,
    signature_key::{BLSPrivKey, BLSPubKey},
    traits::{
        metrics::{Metrics, NoMetrics},
        node_implementation::NodeType,
        signature_key::SignatureKey,
        storage::Storage,
    },
    x25519,
};
use moka::future::Cache;
use options::Identity;
pub use options::Options;
use proposal_fetcher::ProposalFetcherConfig;
pub use run::main;
use tracing::info;
use url::Url;
use vbs::version::StaticVersion;

use crate::request_response::data_source::Storage as RequestResponseStorage;

pub type SequencerApiVersion = StaticVersion<0, 1>;

#[derive(Clone, Debug)]
pub struct NetworkParams {
    pub orchestrator_url: Url,
    pub state_relay_server_url: Url,

    /// The URLs of the builders to use for submitting transactions
    pub builder_urls: Vec<Url>,

    pub private_staking_key: BLSPrivKey,
    pub private_state_key: StateSignKey,
    pub config_peers: Option<Vec<Url>>,
    pub catchup: CatchupParams,
    /// Per-step timeout for the startup stake-table catchup walk
    /// (`bootstrap_epoch_window`).
    pub bootstrap_epoch_catchup_timeout: Duration,
    /// The address to advertise as our public API's URL
    pub public_api_url: Option<Url>,
    /// Cliquenet network address.
    pub cliquenet_bind_addr: NetAddr,
    /// Cliquenet address to advertise to other nodes (registered in the stake table).
    pub cliquenet_advertise_addr: Option<NetAddr>,
    /// X25519 secret key.
    pub x25519_secret_key: x25519::SecretKey,
}

pub struct L1Params {
    pub urls: Vec<Url>,
    pub options: L1ClientOptions,
}

#[derive(Clone, Debug)]
pub struct CatchupParams {
    pub state_peers: Vec<Url>,
    pub backoff: BackoffParams,
    pub base_timeout: Duration,
    pub local_timeout: Duration,
    pub http_delay: Duration,
}

pub(crate) enum LoadedNetworkConfig {
    Persisted(NetworkConfig<SeqTypes>),
    /// The caller persists it once genesis overrides are applied.
    Fetched(NetworkConfig<SeqTypes>),
}

pub(crate) struct NodeStateParts<P> {
    pub node_state: NodeState,
    pub state_catchup: ParallelStateCatchup,
    pub persistence: Arc<P>,
}

#[allow(clippy::too_many_arguments)]
pub async fn init_node<P>(
    genesis: Genesis,
    network_params: NetworkParams,
    metrics: Box<dyn Metrics>,
    persistence: P,
    l1_params: L1Params,
    storage: Option<RequestResponseStorage>,
    event_consumer: impl EventConsumer + 'static,
    is_da: bool,
    identity: Identity,
    proposal_fetcher_config: ProposalFetcherConfig,
    empty_block_delay: Duration,
    leader_trace_dir: Option<PathBuf>,
) -> anyhow::Result<SequencerContext<P>>
where
    P: SequencerPersistence + MembershipPersistence,
    Arc<P>: Storage<SeqTypes>,
{
    // Expose genesis version fields via the status API.
    metrics
        .text_family(
            "genesis".into(),
            vec![
                "base_version".into(),
                "upgrade_version".into(),
                "genesis_version".into(),
            ],
        )
        .create(vec![
            genesis.base_version.to_string(),
            genesis.upgrade_version.to_string(),
            genesis.genesis_version.to_string(),
        ]);
    let upgrades_family = metrics.text_family("genesis_upgrade".into(), vec!["version".into()]);
    for version in genesis.upgrades.keys() {
        upgrades_family.create(vec![version.to_string()]);
    }

    // Expose git information via status API.
    let info = espresso_utils::build_info!();
    metrics
        .text_family(
            "version".into(),
            vec!["rev".into(), "desc".into(), "timestamp".into()],
        )
        .create(vec![
            info.git_sha.into(),
            info.git_describe.into(),
            info.git_commit_timestamp.into(),
        ]);

    metrics
        .text_family(
            "build_info".into(),
            vec![
                "modified".into(),
                "branch".into(),
                "debug".into(),
                "features".into(),
            ],
        )
        .create(vec![
            info.git_dirty.into(),
            info.git_branch.into(),
            info.is_debug.to_string(),
            env!("VERGEN_CARGO_FEATURES").into(),
        ]);

    // Expose Node Entity Information via the status/metrics API
    metrics
        .text_family(
            "node_identity_general".into(),
            vec![
                "name".into(),
                "description".into(),
                "company_name".into(),
                "company_website".into(),
                "operating_system".into(),
                "node_type".into(),
                "network_type".into(),
            ],
        )
        .create(vec![
            identity.node_name.unwrap_or_default(),
            identity.node_description.unwrap_or_default(),
            identity.company_name.unwrap_or_default(),
            identity
                .company_website
                .map(|u| u.into())
                .unwrap_or_default(),
            identity.operating_system.unwrap_or_default(),
            identity.node_type.unwrap_or_default(),
            identity.network_type.unwrap_or_default(),
        ]);

    // Expose Node Identity Location via the status/metrics API
    metrics
        .text_family(
            "node_identity_location".into(),
            vec!["country".into(), "latitude".into(), "longitude".into()],
        )
        .create(vec![
            identity.country_code.unwrap_or_default(),
            identity.latitude.map(|l| l.to_string()).unwrap_or_default(),
            identity
                .longitude
                .map(|l| l.to_string())
                .unwrap_or_default(),
        ]);

    // Expose icons for node dashboard via the status/metrics API
    metrics
        .text_family(
            "node_identity_icon".into(),
            vec![
                "small_1x".into(),
                "small_2x".into(),
                "small_3x".into(),
                "large_1x".into(),
                "large_2x".into(),
                "large_3x".into(),
            ],
        )
        .create(vec![
            identity
                .icon_14x14_1x
                .map(|u| u.to_string())
                .unwrap_or_default(),
            identity
                .icon_14x14_2x
                .map(|u| u.to_string())
                .unwrap_or_default(),
            identity
                .icon_14x14_3x
                .map(|u| u.to_string())
                .unwrap_or_default(),
            identity
                .icon_24x24_1x
                .map(|u| u.to_string())
                .unwrap_or_default(),
            identity
                .icon_24x24_2x
                .map(|u| u.to_string())
                .unwrap_or_default(),
            identity
                .icon_24x24_3x
                .map(|u| u.to_string())
                .unwrap_or_default(),
        ]);

    // Stick our public key in `metrics` so it is easily accessible via the status API.
    let pub_key = BLSPubKey::from_private(&network_params.private_staking_key);
    metrics
        .text_family("node".into(), vec!["key".into()])
        .create(vec![pub_key.to_string()]);

    // Orchestrator client
    let orchestrator_client = OrchestratorClient::new(network_params.orchestrator_url);

    let validator_config = local_validator_config(
        network_params.private_staking_key,
        network_params.private_state_key,
        &network_params.x25519_secret_key,
        network_params.cliquenet_advertise_addr,
        is_da,
    );

    let catchup_params = network_params.catchup;

    let loaded_network_config = load_or_fetch_network_config(
        &persistence,
        network_params.config_peers,
        &validator_config,
        catchup_params.backoff,
        catchup_params.base_timeout,
    )
    .await?;
    let (mut network_config, orchestrator_peer_config, persist_config) = match loaded_network_config
    {
        Some(LoadedNetworkConfig::Persisted(config)) => (config, None, false),
        Some(LoadedNetworkConfig::Fetched(config)) => (config, None, true),
        // Otherwise, this is a fresh network; load from the orchestrator.
        None => {
            tracing::warn!("loading network config from orchestrator");
            tracing::warn!(
                "waiting for other nodes to connect, DO NOT RESTART until fully connected"
            );

            let registration = orchestrator_registration(&validator_config)?;

            let config = get_complete_config(&orchestrator_client, registration.clone())
                .await?
                .0;

            tracing::warn!(
                node_id = config.node_index,
                stake_table = ?config.config.known_nodes_with_stake,
                "loaded config",
            );
            tracing::warn!("all nodes connected");
            (config, Some(registration.public_config()), true)
        },
    };

    apply_genesis_overrides(&mut network_config, &genesis);

    // Override the builder URLs in the network config with the ones from the command line
    // if any were provided
    if !network_params.builder_urls.is_empty() {
        network_config.config.builder_urls = network_params.builder_urls.try_into().unwrap();
    }

    let version_upgrade = versions::Upgrade::new(genesis.base_version, genesis.upgrade_version);
    tracing::warn!("setting version_upgrade={version_upgrade}");

    // Save *after* the above updates. The orchestrator and peer fetched configs don't include
    // epoch_height, drb_difficulty, etc. those come from genesis and are applied above.
    // Saving before these updates would persist zeros for those values
    if persist_config {
        persistence.save_config(&network_config).await?;
    }

    let NodeStateParts {
        node_state: instance_state,
        state_catchup: state_catchup_providers,
        persistence,
    } = init_node_state(
        &genesis,
        &network_config,
        l1_params,
        catchup_params,
        persistence,
        &*metrics,
    )
    .await?;
    let coordinator = instance_state.coordinator.clone();

    check_cliquenet_info_registered(
        coordinator.membership(),
        &validator_config.public_key,
        genesis.base_version,
        genesis.chain_config.stake_table_contract,
        &instance_state.l1_client,
    )
    .await;

    // Load saved consensus state from storage. It is loaded once here, both to
    // seed consensus in `SequencerContext::init` below and to tell whether the
    // network has already cut over to the new protocol.
    let (initializer, anchor_view) = persistence
        .load_consensus_state(instance_state, version_upgrade)
        .await?;

    let block_sizes = genesis.block_sizes();
    let cliquenet = {
        let metrics = clone_box(&*metrics);
        let secret_key = network_params.x25519_secret_key.into();
        let bind_addr = network_params.cliquenet_bind_addr.clone();
        // Accept the largest configured version's messages from startup.
        let largest = *block_sizes
            .values()
            .max()
            .expect("genesis sets a block size");
        let max_message_size = Some(message_limit(largest));
        let name = format!("espresso-{}", genesis.chain_config.chain_id);
        move |upgrade| {
            Cliquenet::create(
                name,
                pub_key,
                secret_key,
                bind_addr,
                [],
                max_message_size,
                upgrade,
                metrics,
            )
        }
    };

    let mut ctx = SequencerContext::init(
        network_config,
        version_upgrade,
        validator_config,
        coordinator,
        initializer,
        anchor_view,
        storage,
        state_catchup_providers,
        persistence,
        cliquenet,
        Some(network_params.state_relay_server_url),
        &*metrics,
        genesis.stake_table.capacity,
        event_consumer,
        proposal_fetcher_config,
        network_params.bootstrap_epoch_catchup_timeout,
        empty_block_delay,
        block_sizes,
        leader_trace_dir,
    )
    .await?;

    if let Some(peer_config) = orchestrator_peer_config {
        ctx = ctx.wait_for_orchestrator(orchestrator_client, peer_config);
    }

    Ok(ctx)
}

/// This node's own validator config, which `status/keys` reports.
///
/// The x25519 key is the cliquenet identity the node runs with, however it was configured
/// (mnemonic, key file or explicit key), so it is always recorded. When no key is configured,
/// `KeySet` generates a random one at startup, so the reported key is only stable across
/// restarts if the operator configured it.
///
/// `p2p_addr` is the address peers dial, so it is only the configured advertise address. The
/// bind address is not recorded: its default is `0.0.0.0`, which nobody can dial, and on real
/// networks the dialable address lives in the stake table rather than in node config.
///
/// What is published into the stake table is decided separately by
/// [`orchestrator_registration`].
fn local_validator_config(
    staking_key: BLSPrivKey,
    state_key: StateSignKey,
    x25519_key: &x25519::SecretKey,
    advertise_addr: Option<NetAddr>,
    is_da: bool,
) -> ValidatorConfig<SeqTypes> {
    let state_key_pair = StateKeyPair::from_sign_key(state_key);
    ValidatorConfig {
        public_key: BLSPubKey::from_private(&staking_key),
        private_key: staking_key,
        stake_value: U256::ONE,
        state_public_key: state_key_pair.ver_key(),
        state_private_key: state_key_pair.sign_key(),
        is_da,
        x25519_keypair: Some(x25519::Keypair::from(x25519_key)),
        p2p_addr: advertise_addr,
    }
}

/// The config this node registers with the orchestrator. Its `public_config()` is also what
/// `/ready` posts, since the orchestrator equality-checks that against `known_nodes_with_stake`.
/// It carries the node's cliquenet `connect_info` so peers can dial us from the stake table, which
/// requires an advertise address to be configured.
fn orchestrator_registration(
    validator_config: &ValidatorConfig<SeqTypes>,
) -> anyhow::Result<ValidatorConfig<SeqTypes>> {
    validator_config.p2p_addr.as_ref().context(
        "ESPRESSO_NODE_CLIQUENET_ADVERTISE_ADDRESS must be set when bootstrapping a Cliquenet \
         network from the orchestrator",
    )?;
    Ok(validator_config.clone())
}

pub(crate) async fn load_or_fetch_network_config<P: SequencerPersistence>(
    persistence: &P,
    config_peers: Option<Vec<Url>>,
    validator_config: &ValidatorConfig<SeqTypes>,
    backoff: BackoffParams,
    base_timeout: Duration,
) -> anyhow::Result<Option<LoadedNetworkConfig>> {
    if let Some(config) = persistence.load_config().await? {
        tracing::warn!("loaded network config from storage, rejoining existing network");
        return Ok(Some(LoadedNetworkConfig::Persisted(config)));
    }

    let Some(peers) = config_peers else {
        return Ok(None);
    };
    tracing::warn!(?peers, "loading network config from peers");
    let peers =
        StatePeers::<SequencerApiVersion>::from_urls(peers, backoff, base_timeout, &NoMetrics);
    let config = peers.fetch_config(validator_config.clone()).await?;

    tracing::warn!(
        node_id = config.node_index,
        stake_table = ?config.config.known_nodes_with_stake,
        "loaded config",
    );
    Ok(Some(LoadedNetworkConfig::Fetched(config)))
}

/// Orchestrator and peer-fetched configs carry none of these; run before the config is persisted.
pub(crate) fn apply_genesis_overrides(
    network_config: &mut NetworkConfig<SeqTypes>,
    genesis: &Genesis,
) {
    if let Some(upgrade) = genesis.upgrades.get(&genesis.upgrade_version) {
        upgrade.set_hotshot_config_parameters(&mut network_config.config);
    }

    let epoch_height = genesis.epoch_height.unwrap_or_default();
    let drb_difficulty = genesis.drb_difficulty.unwrap_or_default();
    let drb_upgrade_difficulty = genesis.drb_upgrade_difficulty.unwrap_or_default();
    let epoch_start_block = genesis.epoch_start_block.unwrap_or_default();
    let stake_table_capacity = genesis
        .stake_table_capacity
        .unwrap_or(hotshot_types::light_client::DEFAULT_STAKE_TABLE_CAPACITY);

    tracing::warn!("setting epoch_height={epoch_height:?}");
    tracing::warn!("setting drb_difficulty={drb_difficulty:?}");
    tracing::warn!("setting drb_upgrade_difficulty={drb_upgrade_difficulty:?}");
    tracing::warn!("setting epoch_start_block={epoch_start_block:?}");
    tracing::warn!("setting stake_table_capacity={stake_table_capacity:?}");
    network_config.config.epoch_height = epoch_height;
    network_config.config.drb_difficulty = drb_difficulty;
    network_config.config.drb_upgrade_difficulty = drb_upgrade_difficulty;
    network_config.config.epoch_start_block = epoch_start_block;
    network_config.config.stake_table_capacity = stake_table_capacity;

    if let Some(da_committees) = &genesis.da_committees {
        tracing::warn!("setting da_committees from genesis: {da_committees:?}");
        network_config.config.da_committees = da_committees.clone();
    }
}

/// The epoch layout is read from `genesis`, not `network_config`, so a caller that skips
/// [`apply_genesis_overrides`] cannot silently start with `epoch_height = 0`.
pub(crate) async fn init_node_state<P>(
    genesis: &Genesis,
    network_config: &NetworkConfig<SeqTypes>,
    l1_params: L1Params,
    catchup: CatchupParams,
    mut persistence: P,
    metrics: &dyn Metrics,
) -> anyhow::Result<NodeStateParts<P>>
where
    P: SequencerPersistence + MembershipPersistence,
    Arc<P>: Storage<SeqTypes>,
{
    // Before the L1 work below, which can wait a long time for L1.
    let first_epoch = genesis.first_epoch()?;

    let l1_client = l1_params
        .options
        .with_metrics(metrics)
        .connect(l1_params.urls)
        .with_context(|| "failed to create L1 client")?;

    info!("Validating fee contract");

    genesis.validate_fee_contract(&l1_client).await?;

    info!("Fee contract validated. Spawning L1 tasks");

    l1_client.spawn_tasks().await;

    info!(
        "L1 tasks spawned. Waiting for L1 genesis: {:?}",
        genesis.l1_finalized
    );

    let l1_genesis = match &genesis.l1_finalized {
        L1Finalized::Block(b) => *b,
        L1Finalized::Number { number } => l1_client.wait_for_finalized_block(*number).await,
        L1Finalized::Timestamp { timestamp } => {
            l1_client
                .wait_for_finalized_block_with_timestamp(U256::from(timestamp.unix_timestamp()))
                .await
        },
    };

    info!("L1 genesis found: {:?}", l1_genesis);

    let genesis_chain_config = genesis.header.chain_config;
    let mut genesis_state = ValidatedState {
        chain_config: genesis_chain_config.into(),
        ..Default::default()
    };
    for (address, amount) in &genesis.accounts {
        tracing::warn!(%address, %amount, "Prefunding account for demo");
        genesis_state.prefund_account(*address, *amount);
    }

    let state_catchup_providers = ParallelStateCatchup::new(&[], catchup.local_timeout)
        .with_fallback_delay(catchup.http_delay);
    let state_peers = StatePeers::<SequencerApiVersion>::from_urls(
        catchup.state_peers,
        catchup.backoff,
        catchup.base_timeout,
        metrics,
    );
    state_catchup_providers.add_fallback_provider(Arc::new(state_peers));
    // Local is a single attempt per call; remote providers retry.
    match persistence
        .clone()
        .into_catchup_provider(BackoffParams::disabled())
    {
        Ok(provider) => {
            state_catchup_providers.add_provider(Arc::new(provider));
        },
        Err(e) => {
            tracing::warn!(
                "Failed to create local catchup provider: {e:#}. Only using remote catchup."
            );
        },
    };

    persistence.enable_metrics(metrics);

    let fetcher = Fetcher::new(
        Arc::new(state_catchup_providers.clone()),
        Arc::new(Mutex::new(persistence.clone())),
        l1_client.clone(),
        genesis.chain_config,
    );

    // Before `spawn_update_loop`, so the cold full-history scan runs once, not twice.
    prefetch_stake_table_events(
        &fetcher,
        &l1_client,
        &l1_genesis,
        genesis.chain_config.stake_table_contract,
    )
    .await?;

    info!("Spawning update loop");

    fetcher.spawn_update_loop().await;
    info!("Update loop spawned. Fetching block reward");

    let block_reward = fetcher.fetch_fixed_block_reward().await.ok();
    info!("Block reward fetched: {:?}", block_reward);
    let epoch_height = genesis.epoch_height.unwrap_or_default();
    let mut membership = EpochCommittees::new(
        network_config.config.known_nodes_with_stake.clone(),
        network_config.config.known_da_nodes.clone(),
        block_reward,
        fetcher,
        epoch_height,
        first_epoch,
    );
    info!("Membership created. Reloading stake");
    membership.reload_stake(RECENT_STAKE_TABLES_LIMIT).await;
    info!("Stake reloaded");

    let persistence = Arc::new(persistence);
    let coordinator = EpochMembershipCoordinator::new(membership, epoch_height, &persistence);

    let epoch_rewards_calculator = Arc::new(Mutex::new(EpochRewardsCalculator::new()));

    let node_state = NodeState {
        chain_config: genesis.chain_config,
        genesis_chain_config,
        l1_client,
        genesis_header: genesis.header.clone(),
        genesis_state,
        l1_genesis: Some(l1_genesis),
        node_id: network_config.node_index,
        upgrades: genesis.upgrades.clone(),
        current_version: genesis.base_version,
        epoch_height: Some(epoch_height),
        state_catchup: Arc::new(state_catchup_providers.clone()),
        coordinator,
        genesis_version: genesis.genesis_version,
        epoch_start_block: genesis.epoch_start_block.unwrap_or_default(),
        epoch_rewards_calculator,
        light_client_contract_address: Cache::builder().max_capacity(1).build(),
        token_contract_address: Cache::builder().max_capacity(1).build(),
        finalized_hotshot_height: Cache::builder()
            .max_capacity(1)
            .time_to_live(Duration::from_secs(30))
            .build(),
    };

    Ok(NodeStateParts {
        node_state,
        state_catchup: state_catchup_providers,
        persistence,
    })
}

/// Scans the whole L1 contract history when nothing is persisted yet, keeping every
/// `bootstrap_epoch_window` step bounded to one epoch instead of loading that scan onto
/// the first step, which silently downgrades the node to a stale epoch window when it
/// exceeds its timeout. Untimed: an unreachable L1 panics after
/// `l1_events_max_retry_duration`.
pub(crate) async fn prefetch_stake_table_events(
    fetcher: &Fetcher,
    l1_client: &espresso_types::v0::L1Client,
    l1_genesis: &espresso_types::v0::L1BlockInfo,
    stake_table_contract: Option<alloy::primitives::Address>,
) -> anyhow::Result<()> {
    let Some(addr) = stake_table_contract else {
        return Ok(());
    };

    // A pinned genesis block never waits on the L1 client, so the snapshot may be empty here.
    // Does not fetch the genesis block itself: RPCs that prune old history no longer serve it.
    let finalized = l1_client
        .wait_for_latest_finalized_at_least(l1_genesis.number)
        .await;

    tracing::info!(
        target: "announce",
        %addr,
        to_block = finalized.number,
        "prefetching stake table events",
    );
    let events = fetcher
        .fetch_and_store_stake_table_events(addr, finalized.number)
        .await
        .context("prefetching stake table events")?;
    tracing::info!(target: "announce", events = events.len(), "prefetched stake table events");
    Ok(())
}

/// On the version immediately preceding CLIQUENET, log an error if this
/// validator has a stake-table entry without an x25519 key or p2p address.
/// Skipped unless StakeTableV3 is deployed (otherwise the operator has no
/// actionable path), detected via `getVersion()` on the proxy.
async fn check_cliquenet_info_registered(
    membership: &EpochCommittees,
    pub_key: &BLSPubKey,
    current_version: vbs::version::Version,
    stake_table_contract: Option<alloy::primitives::Address>,
    l1_client: &espresso_types::v0::L1Client,
) {
    if current_version != versions::EPOCH_REWARD_VERSION {
        return;
    }
    let Some(addr) = stake_table_contract else {
        return;
    };
    let stake_table =
        hotshot_contract_adapter::sol_types::StakeTableV3::new(addr, l1_client.provider.clone());
    let mut major = None;
    let mut last_err = None;
    for attempt in 1..=3 {
        match stake_table.getVersion().call().await {
            Ok(v) => {
                major = Some(v.majorVersion);
                break;
            },
            Err(e) => {
                tracing::warn!(attempt, %e, "failed to read StakeTable getVersion(), retrying");
                last_err = Some(e);
                if attempt < 3 {
                    tokio::time::sleep(Duration::from_secs(1)).await;
                }
            },
        }
    }
    let Some(major) = major else {
        tracing::warn!(
            err = ?last_err,
            "could not read StakeTable getVersion() after 3 attempts; skipping check"
        );
        return;
    };
    if major < 3 {
        tracing::info!(
            major,
            "StakeTableV3 not deployed; skipping network-info registration check"
        );
        return;
    }
    let Some(cfg) = membership.latest_peer_config(pub_key) else {
        return;
    };
    if cfg.connect_info.is_some() {
        return;
    }
    tracing::error!(
        bls_key = %pub_key,
        "Validator has no x25519 key or p2p address registered on-chain. After the CLIQUENET \
         upgrade activates, the validator will be excluded from the active set and stop earning \
         rewards. To fix: (1) generate an x25519 keypair if needed (`keygen --scheme x25519 \
         --out keys.env`), then (2) register it on-chain (`staking-cli update-network-config \
         --x25519-key <ESPRESSO_NODE_PUBLIC_X25519_KEY> --p2p-addr <host:port>`)."
    );
}

#[cfg(any(test, feature = "testing"))]
pub mod testing {
    use std::{
        cmp::max,
        collections::{BTreeMap, HashMap, HashSet},
        net::Ipv4Addr,
        time::Duration,
    };

    use alloy::{
        network::EthereumWallet,
        node_bindings::{Anvil, AnvilInstance},
        primitives::{Address, U256},
        providers::{
            Provider, ProviderBuilder, RootProvider, WalletProvider,
            fillers::{
                BlobGasFiller, ChainIdFiller, FillProvider, GasFiller, JoinFill, NonceFiller,
            },
            layers::AnvilProvider,
        },
        signers::{k256::ecdsa::SigningKey, local::LocalSigner},
    };
    use catchup::NullStateCatchup;
    use committable::Committable;
    use espresso_contract_deployer::{
        Contract, Contracts, DEFAULT_EXIT_ESCROW_PERIOD_SECONDS,
        builder::{DeployerArgs, DeployerArgsBuilder},
        network_config::light_client_genesis_from_stake_table,
    };
    use espresso_types::{
        ChainConfig, FeeAccount, L1Client, NetworkConfig, PubKey, SeqTypes, Transaction, Upgrade,
        UpgradeMap, UpgradeMode,
        eth_signature_key::EthKeyPair,
        v0::traits::{EventConsumer, NullEventConsumer, PersistenceOptions, StateCatchup},
    };
    use futures::{
        future::join_all,
        stream::{Stream, StreamExt},
    };
    use hotshot_contract_adapter::stake_table::StakeTableContractVersion;
    use hotshot_types::{
        HotShotConfig, PeerConfig, PeerConnectInfo,
        data::EpochNumber,
        event::LeafInfo,
        light_client::StateKeyPair,
        new_protocol::CoordinatorEvent,
        traits::{
            BlockPayload, EncodeBytes, block_contents::BlockHeader, metrics::NoMetrics,
            signature_key::BuilderSignatureKey,
        },
        utils::epoch_from_block_number,
    };
    use rand::SeedableRng as _;
    use rand_chacha::ChaCha20Rng;
    use staking_cli::demo::{DelegationConfig, StakingKeySet, StakingTransactions};
    use test_utils::reserve_tcp_port;
    use tokio::time::timeout;
    use vbs::version::Version;
    use versions::{EPOCH_VERSION, LARGE_BLOCK_VERSION};

    use super::*;
    use crate::{
        catchup::ParallelStateCatchup,
        persistence::no_storage::{self, NoStorage},
    };

    const STAKE_TABLE_CAPACITY_FOR_TEST: usize = 10;
    type AnvilFillProvider = AnvilProvider<
        FillProvider<
            JoinFill<
                alloy::providers::Identity,
                JoinFill<GasFiller, JoinFill<BlobGasFiller, JoinFill<NonceFiller, ChainIdFiller>>>,
            >,
            RootProvider,
        >,
    >;

    pub struct TestConfigBuilder<const NUM_NODES: usize> {
        config: HotShotConfig<SeqTypes>,
        priv_keys: Vec<BLSPrivKey>,
        state_key_pairs: Vec<StateKeyPair>,
        l1_url: Url,
        l1_opt: L1ClientOptions,
        anvil_provider: Option<AnvilFillProvider>,
        signer: LocalSigner<SigningKey>,
        state_relay_url: Option<Url>,
        upgrades: BTreeMap<Version, Upgrade>,
        coordinator_addrs: Vec<NetAddr>,
        contracts: Option<Contracts>,
    }

    /// Picks the key sets at `indices` out of the full deterministic sequence,
    /// preserving the order of `indices`. Panics on duplicate or out-of-range
    /// indices.
    fn select_staking_key_sets(all: Vec<StakingKeySet>, indices: &[usize]) -> Vec<StakingKeySet> {
        let mut by_index: Vec<Option<StakingKeySet>> = all.into_iter().map(Some).collect();
        indices
            .iter()
            .map(|&i| {
                by_index
                    .get_mut(i)
                    .unwrap_or_else(|| panic!("validator index {i} out of range"))
                    .take()
                    .unwrap_or_else(|| panic!("duplicate validator index {i}"))
            })
            .collect()
    }

    /// Deploys the contract suite up to and including the requested stake
    /// table version, recording addresses in `contracts`.
    pub async fn deploy_stake_table(
        args: &DeployerArgs<impl Provider + WalletProvider>,
        version: StakeTableContractVersion,
        contracts: &mut Contracts,
    ) -> anyhow::Result<()> {
        match version {
            StakeTableContractVersion::V1 => args.deploy_to_stake_table_v1(contracts).await,
            StakeTableContractVersion::V2 => args.deploy_to_stake_table_v2(contracts).await,
            StakeTableContractVersion::V3 => args.deploy_to_stake_table_v3(contracts).await,
        }
    }

    pub fn staking_priv_keys(
        priv_keys: &[BLSPrivKey],
        state_key_pairs: &[StateKeyPair],
        coordinator_addrs: &[NetAddr],
        num_nodes: usize,
    ) -> Vec<StakingKeySet> {
        let seed = [42u8; 32];
        let mut rng = ChaCha20Rng::from_seed(seed); // Create a deterministic RNG
        let eth_key_pairs = (0..num_nodes).map(|_| SigningKey::random(&mut rng).into());
        eth_key_pairs
            .zip(priv_keys.iter())
            .zip(state_key_pairs.iter())
            .enumerate()
            .map(|(i, ((eth, bls), state))| StakingKeySet {
                signer: eth,
                bls: bls.clone().into(),
                state: state.clone(),
                // x25519 key derived from the BLS key, matching `init_node`'s
                // cliquenet bind.
                x25519: x25519::Keypair::derive_from::<PubKey>(bls)
                    .expect("x25519 keypair derivation should succeed"),
                p2p_addr: coordinator_addrs
                    .get(i)
                    .cloned()
                    .unwrap_or_else(|| "127.0.0.1:8080".parse().unwrap()),
                metadata_uri: None,
            })
            .collect()
    }

    impl<const NUM_NODES: usize> TestConfigBuilder<NUM_NODES> {
        pub fn state_relay_url(mut self, url: Url) -> Self {
            self.state_relay_url = Some(url);
            self
        }

        /// Sets the Anvil provider, constructed using the Anvil instance.
        /// Also sets the L1 URL based on the Anvil endpoint.
        /// The `AnvilProvider` can be used to configure the Anvil, for example,
        /// by enabling interval mining after the test network is initialized.
        pub fn anvil_provider(mut self, anvil: AnvilInstance) -> Self {
            self.l1_url = anvil.endpoint().parse().unwrap();
            let l1_client = L1Client::anvil(&anvil).expect("create l1 client");
            let anvil_provider = AnvilProvider::new(l1_client.provider, Arc::new(anvil));
            self.anvil_provider = Some(anvil_provider);
            self
        }

        /// Sets a custom L1 URL, overriding any previously set Anvil instance URL.
        /// This removes the anvil provider, as well as it is no longer needed
        pub fn l1_url(mut self, l1_url: Url) -> Self {
            self.anvil_provider = None;
            self.l1_url = l1_url;
            self
        }

        pub fn l1_opt(mut self, opt: L1ClientOptions) -> Self {
            self.l1_opt = opt;
            self
        }

        pub fn signer(mut self, signer: LocalSigner<SigningKey>) -> Self {
            self.signer = signer;
            self
        }

        pub fn upgrades(mut self, v: Version, upgrades: BTreeMap<Version, Upgrade>) -> Self {
            let upgrade = upgrades.get(&v).unwrap();
            upgrade.set_hotshot_config_parameters(&mut self.config);
            self.upgrades = upgrades;
            self
        }

        /// Version specific upgrade setup. Extend to future upgrades
        /// by adding a branch to the `match` statement.
        pub async fn set_upgrades(self, version: Version) -> Self {
            let registered: Vec<usize> = (0..NUM_NODES).collect();
            self.set_upgrades_with(version, StakeTableContractVersion::V3, &registered)
                .await
        }

        /// Like [`Self::set_upgrades`], but deploys the requested stake table
        /// contract version and registers only the validators at the
        /// `registered` node indices. The deployed [`Contracts`] registry is
        /// retained and available via [`TestConfig::contracts`].
        pub async fn set_upgrades_with(
            mut self,
            version: Version,
            stake_table_version: StakeTableContractVersion,
            registered: &[usize],
        ) -> Self {
            let upgrade = match version {
                version if version >= EPOCH_VERSION => {
                    tracing::debug!(?version, "upgrade version");
                    let blocks_per_epoch = self.config.epoch_height;
                    let epoch_start_block = self.config.epoch_start_block;

                    let (genesis_state, genesis_stake) = light_client_genesis_from_stake_table(
                        &self.config.hotshot_stake_table(),
                        STAKE_TABLE_CAPACITY_FOR_TEST,
                    )
                    .unwrap();

                    let validators = select_staking_key_sets(
                        staking_priv_keys(
                            &self.priv_keys,
                            &self.state_key_pairs,
                            &self.coordinator_addrs,
                            NUM_NODES,
                        ),
                        registered,
                    );

                    let deployer = ProviderBuilder::new()
                        .wallet(EthereumWallet::from(self.signer.clone()))
                        .connect_http(self.l1_url.clone());

                    let mut contracts = Contracts::new();
                    let args = DeployerArgsBuilder::default()
                        .deployer(deployer.clone())
                        .rpc_url(self.l1_url.clone())
                        .mock_light_client(true)
                        .genesis_lc_state(genesis_state)
                        .genesis_st_state(genesis_stake)
                        .blocks_per_epoch(blocks_per_epoch)
                        .epoch_start_block(epoch_start_block)
                        .exit_escrow_period(U256::from(max(
                            blocks_per_epoch * 15 + 100,
                            DEFAULT_EXIT_ESCROW_PERIOD_SECONDS,
                        )))
                        .multisig_pauser(self.signer.address())
                        .token_name("Espresso".to_string())
                        .token_symbol("ESP".to_string())
                        .initial_token_supply(U256::from(3590000000u64))
                        .ops_timelock_delay(U256::from(0))
                        .ops_timelock_admin(self.signer.address())
                        .ops_timelock_proposers(vec![self.signer.address()])
                        .ops_timelock_executors(vec![self.signer.address()])
                        .safe_exit_timelock_delay(U256::from(10))
                        .safe_exit_timelock_admin(self.signer.address())
                        .safe_exit_timelock_proposers(vec![self.signer.address()])
                        .safe_exit_timelock_executors(vec![self.signer.address()])
                        .build()
                        .unwrap();
                    deploy_stake_table(&args, stake_table_version, &mut contracts)
                        .await
                        .expect("failed to deploy all contracts");

                    let st_addr = contracts
                        .address(Contract::StakeTableProxy)
                        .expect("StakeTableProxy address not found");
                    StakingTransactions::create(
                        self.l1_url.clone(),
                        &deployer,
                        st_addr,
                        validators,
                        None,
                        DelegationConfig::default(),
                    )
                    .await
                    .expect("stake table setup failed")
                    .apply_all()
                    .await
                    .expect("send all txns failed");

                    self.contracts = Some(contracts);

                    if version >= LARGE_BLOCK_VERSION {
                        Upgrade::large_block_view_based(st_addr)
                    } else {
                        Upgrade::pos_view_based(st_addr)
                    }
                },
                _ => panic!("Upgrade not configured for version {version:?}"),
            };

            let mut upgrades = std::collections::BTreeMap::new();
            upgrade.set_hotshot_config_parameters(&mut self.config);
            upgrades.insert(version, upgrade);

            self.upgrades = upgrades;
            self
        }

        pub fn epoch_height(mut self, epoch_height: u64) -> Self {
            self.config.epoch_height = epoch_height;
            self
        }

        pub fn epoch_start_block(mut self, start_block: u64) -> Self {
            self.config.epoch_start_block = start_block;
            self
        }

        pub fn builder_timeout(mut self, timeout: Duration) -> Self {
            self.config.builder_timeout = timeout;
            self
        }

        /// Override the base next-view (failure) timeout. Raise it when using a large
        /// [`Self::builder_timeout`] so a slow-but-healthy view isn't mistaken for a failed one.
        pub fn next_view_timeout(mut self, timeout: Duration) -> Self {
            self.config.next_view_timeout = timeout.as_millis() as u64;
            self
        }

        /// Override the views during which the upgrade is proposed. Call after `set_upgrades`.
        pub fn upgrade_proposing_views(mut self, start: u64, stop: u64) -> Self {
            for upgrade in self.upgrades.values_mut() {
                if let UpgradeMode::View(v) = &mut upgrade.mode {
                    v.start_proposing_view = start;
                    v.stop_proposing_view = stop;
                }
            }
            self.config.start_proposing_view = start;
            self.config.stop_proposing_view = stop;
            self
        }

        pub fn build(self) -> TestConfig<NUM_NODES> {
            TestConfig {
                config: self.config,
                priv_keys: self.priv_keys,
                state_key_pairs: self.state_key_pairs,
                l1_url: self.l1_url,
                l1_opt: self.l1_opt,
                signer: self.signer,
                state_relay_url: self.state_relay_url,
                upgrades: self.upgrades,
                anvil_provider: self.anvil_provider,
                coordinator_addrs: self.coordinator_addrs,
                contracts: self.contracts,
                genesis_chain_config: None,
            }
        }

        pub fn stake_table_capacity(mut self, stake_table_capacity: usize) -> Self {
            self.config.stake_table_capacity = stake_table_capacity;
            self
        }
    }

    impl<const NUM_NODES: usize> Default for TestConfigBuilder<NUM_NODES> {
        fn default() -> Self {
            let num_nodes = NUM_NODES;

            // Generate keys for the nodes.
            let seed = [0; 32];
            let (pub_keys, priv_keys): (Vec<_>, Vec<_>) = (0..num_nodes)
                .map(|i| <PubKey as SignatureKey>::generated_from_seed_indexed(seed, i as u64))
                .unzip();
            let state_key_pairs = (0..num_nodes)
                .map(|i| StateKeyPair::generate_from_seed_indexed(seed, i as u64))
                .collect::<Vec<_>>();

            // Reserve one cliquenet coordinator port per node.
            let coordinator_addrs: Vec<NetAddr> = (0..num_nodes)
                .map(|_| {
                    let port =
                        reserve_tcp_port().expect("OS should have ephemeral ports available");
                    NetAddr::Inet(Ipv4Addr::LOCALHOST.into(), port)
                })
                .collect();

            let known_nodes_with_stake = pub_keys
                .iter()
                .zip(&priv_keys)
                .zip(&state_key_pairs)
                .zip(&coordinator_addrs)
                .map(
                    |(((pub_key, priv_key), state_key_pair), addr)| PeerConfig::<SeqTypes> {
                        stake_table_entry: pub_key.stake_table_entry(U256::from(1)),
                        state_ver_key: state_key_pair.ver_key(),
                        connect_info: Some(PeerConnectInfo {
                            x25519_key: x25519::Keypair::derive_from::<PubKey>(priv_key)
                                .expect("x25519 keypair derivation should succeed")
                                .public_key(),
                            p2p_addr: addr.clone(),
                        }),
                    },
                )
                .collect::<Vec<_>>();

            let builder_port = reserve_tcp_port().unwrap();

            let config: HotShotConfig<SeqTypes> = HotShotConfig {
                fixed_leader_for_gpuvid: 0,
                num_nodes_with_stake: num_nodes.try_into().unwrap(),
                known_da_nodes: known_nodes_with_stake.clone(),
                da_committees: Default::default(),
                known_nodes_with_stake: known_nodes_with_stake.clone(),
                next_view_timeout: Duration::from_secs(5).as_millis() as u64,
                num_bootstrap: 1usize,
                da_staked_committee_size: num_nodes,
                view_sync_timeout: Duration::from_secs(1),
                data_request_delay: Duration::from_secs(1),
                builder_urls: vec1::vec1![
                    Url::parse(&format!("http://localhost:{builder_port}")).unwrap()
                ],
                builder_timeout: Duration::from_secs(1),
                start_threshold: (
                    known_nodes_with_stake.clone().len() as u64,
                    known_nodes_with_stake.clone().len() as u64,
                ),
                start_proposing_view: 0,
                stop_proposing_view: 0,
                start_voting_view: 0,
                stop_voting_view: 0,
                start_proposing_time: 0,
                start_voting_time: 0,
                stop_proposing_time: 0,
                stop_voting_time: 0,
                epoch_height: 30,
                epoch_start_block: 1,
                stake_table_capacity: hotshot_types::light_client::DEFAULT_STAKE_TABLE_CAPACITY,
                drb_difficulty: 10,
                drb_upgrade_difficulty: 20,
            };

            let anvil = Anvil::new()
                .args(["--slots-in-an-epoch", "0", "--balance", "1000000"])
                .spawn();

            let l1_client = L1Client::anvil(&anvil).expect("failed to create l1 client");
            let anvil_provider = AnvilProvider::new(l1_client.provider, Arc::new(anvil));

            let l1_signer_key = anvil_provider.anvil().keys()[0].clone();
            let signer = LocalSigner::from(l1_signer_key);

            Self {
                config,
                priv_keys,
                state_key_pairs,
                l1_url: anvil_provider.anvil().endpoint().parse().unwrap(),
                l1_opt: L1ClientOptions {
                    stake_table_update_interval: Duration::from_secs(5),
                    l1_events_max_block_range: 1000,
                    l1_polling_interval: Duration::from_secs(1),
                    subscription_timeout: Duration::from_secs(5),
                    ..Default::default()
                },
                anvil_provider: Some(anvil_provider),
                signer,
                state_relay_url: None,
                upgrades: Default::default(),
                coordinator_addrs,
                contracts: None,
            }
        }
    }

    #[derive(Clone)]
    pub struct TestConfig<const NUM_NODES: usize> {
        config: HotShotConfig<SeqTypes>,
        priv_keys: Vec<BLSPrivKey>,
        state_key_pairs: Vec<StateKeyPair>,
        l1_url: Url,
        l1_opt: L1ClientOptions,
        anvil_provider: Option<AnvilFillProvider>,
        signer: LocalSigner<SigningKey>,
        state_relay_url: Option<Url>,
        upgrades: BTreeMap<Version, Upgrade>,
        /// Per-node cliquenet coordinator bind addresses, indexed by node.
        coordinator_addrs: Vec<NetAddr>,
        /// Contracts deployed by [`TestConfigBuilder::set_upgrades_with`], if any.
        contracts: Option<Contracts>,
        /// See [`Self::set_genesis_chain_config`].
        genesis_chain_config: Option<ChainConfig>,
    }

    impl<const NUM_NODES: usize> TestConfig<NUM_NODES> {
        pub fn num_nodes(&self) -> usize {
            self.priv_keys.len()
        }

        pub fn hotshot_config(&self) -> &HotShotConfig<SeqTypes> {
            &self.config
        }

        pub fn signer(&self) -> LocalSigner<SigningKey> {
            self.signer.clone()
        }

        pub fn l1_url(&self) -> Url {
            self.l1_url.clone()
        }

        pub fn l1_opt(&self) -> L1ClientOptions {
            self.l1_opt.clone()
        }

        pub fn anvil(&self) -> Option<&AnvilFillProvider> {
            self.anvil_provider.as_ref()
        }

        pub fn get_upgrade_map(&self) -> UpgradeMap {
            self.upgrades.clone().into()
        }

        pub fn upgrades(&self) -> BTreeMap<Version, Upgrade> {
            self.upgrades.clone()
        }

        pub fn staking_priv_keys(&self) -> Vec<StakingKeySet> {
            staking_priv_keys(
                &self.priv_keys,
                &self.state_key_pairs,
                &self.coordinator_addrs,
                self.num_nodes(),
            )
        }

        /// Key sets for the given node indices, aligned with
        /// [`Self::staking_priv_keys`]: the full deterministic sequence is
        /// generated first and then filtered, so a subset keeps the same keys
        /// per node index.
        pub fn staking_key_sets(&self, indices: &[usize]) -> Vec<StakingKeySet> {
            select_staking_key_sets(self.staking_priv_keys(), indices)
        }

        /// The cliquenet coordinator address assigned to node `i`.
        pub fn coordinator_addr(&self, i: usize) -> NetAddr {
            self.coordinator_addrs[i].clone()
        }

        /// Rebinds node `i`'s cliquenet coordinator to `addr`, taking effect
        /// the next time the node is initialized. Peers learn the address
        /// from the stake table contract, so the caller must also publish it
        /// there (`updateP2pAddr` or `updateNetworkConfig`).
        pub fn set_coordinator_addr(&mut self, i: usize, addr: NetAddr) {
            self.coordinator_addrs[i] = addr;
        }

        /// Replaces node `i`'s consensus keys, taking effect the next time
        /// the node is initialized: it then signs with the new keys and
        /// derives its cliquenet x25519 identity from the new BLS key (see
        /// [`Self::init_node`]). The genesis stake table intentionally keeps
        /// the old keys — it must stay identical across nodes — so the new
        /// keys only become effective once the caller publishes the same
        /// rotation on-chain (`updateConsensusKeysV2`, plus
        /// `updateNetworkConfig` with the newly derived x25519 key) and the
        /// change reaches an epoch's stake table snapshot.
        ///
        /// Caveat: the light-client state signer checks membership against
        /// the genesis stake table, so a rotated node stops contributing
        /// state-relay signatures. No test combines rotation with a state
        /// relay yet.
        pub fn set_consensus_keys(&mut self, i: usize, bls: BLSPrivKey, state: StateKeyPair) {
            self.priv_keys[i] = bls;
            self.state_key_pairs[i] = state;
        }

        /// Builds every node's genesis header from `chain_config`, as a real
        /// network's shared genesis file does. Otherwise a node takes it from
        /// its own starting state, so nodes that start from only a commitment
        /// build a different genesis leaf, and the new protocol rejects each
        /// other's genesis QC.
        pub fn set_genesis_chain_config(&mut self, chain_config: ChainConfig) {
            self.genesis_chain_config = Some(chain_config);
        }

        /// Contracts deployed by [`TestConfigBuilder::set_upgrades_with`], if
        /// that was used to set up this config.
        pub fn contracts(&self) -> Option<Contracts> {
            self.contracts.clone()
        }

        pub fn validator_providers(
            &self,
        ) -> Vec<(Address, impl Provider + Clone + use<NUM_NODES>)> {
            self.staking_priv_keys()
                .into_iter()
                .map(|key_set| {
                    (
                        key_set.signer.address(),
                        ProviderBuilder::new()
                            .wallet(EthereumWallet::from(key_set.signer))
                            .connect_http(self.l1_url.clone()),
                    )
                })
                .collect()
        }

        pub async fn init_nodes(
            &self,
            upgrade: versions::Upgrade,
        ) -> Vec<SequencerContext<NoStorage>> {
            join_all((0..self.num_nodes()).map(|i| async move {
                self.init_node(
                    i,
                    ValidatedState::default(),
                    no_storage::Options,
                    Some(NullStateCatchup::default()),
                    None,
                    &NoMetrics,
                    STAKE_TABLE_CAPACITY_FOR_TEST,
                    NullEventConsumer,
                    upgrade,
                    Default::default(),
                )
                .await
            }))
            .await
        }

        pub fn known_nodes_with_stake(&self) -> &[PeerConfig<SeqTypes>] {
            &self.config.known_nodes_with_stake
        }

        #[allow(clippy::too_many_arguments)]
        pub async fn init_node<P: PersistenceOptions>(
            &self,
            i: usize,
            state: ValidatedState,
            persistence_opt: P,
            state_peers: Option<impl StateCatchup + 'static>,
            storage: Option<RequestResponseStorage>,
            metrics: &dyn Metrics,
            stake_table_capacity: usize,
            event_consumer: impl EventConsumer + 'static,
            upgrade: versions::Upgrade,
            upgrades: BTreeMap<Version, Upgrade>,
        ) -> SequencerContext<P::Persistence> {
            self.try_init_node(
                i,
                state,
                persistence_opt,
                state_peers,
                storage,
                metrics,
                stake_table_capacity,
                event_consumer,
                upgrade,
                upgrades,
            )
            .await
            .expect("node initializes")
        }

        /// [`Self::init_node`], returning the error `SequencerContext::init` refuses with.
        #[allow(clippy::too_many_arguments)]
        pub async fn try_init_node<P: PersistenceOptions>(
            &self,
            i: usize,
            mut state: ValidatedState,
            mut persistence_opt: P,
            state_peers: Option<impl StateCatchup + 'static>,
            storage: Option<RequestResponseStorage>,
            metrics: &dyn Metrics,
            stake_table_capacity: usize,
            event_consumer: impl EventConsumer + 'static,
            upgrade: versions::Upgrade,
            upgrades: BTreeMap<Version, Upgrade>,
        ) -> anyhow::Result<SequencerContext<P::Persistence>> {
            let config = self.config.clone();
            let my_peer_config = &config.known_nodes_with_stake[i];
            let is_da = config.known_da_nodes.contains(my_peer_config);

            // The new-protocol (cliquenet) coordinator network identifies this
            // node by an x25519 key derived from its BLS key, reachable at its
            // pre-assigned coordinator address. These must match what is
            // registered on-chain for this validator (see `staking_priv_keys`),
            // so peers can resolve and dial each other from the stake table.
            let x25519_keypair = x25519::Keypair::derive_from::<PubKey>(&self.priv_keys[i])
                .expect("x25519 keypair derivation should succeed");
            let coordinator_addr = self.coordinator_addrs[i].clone();

            let pub_key = PubKey::from_private(&self.priv_keys[i]);

            let validator_config = ValidatorConfig {
                public_key: pub_key,
                private_key: self.priv_keys[i].clone(),
                stake_value: my_peer_config.stake_table_entry.stake_amount,
                state_public_key: self.state_key_pairs[i].ver_key(),
                state_private_key: self.state_key_pairs[i].sign_key(),
                is_da,
                x25519_keypair: Some(x25519_keypair.clone()),
                p2p_addr: Some(coordinator_addr.clone()),
            };

            // Make sure the builder account is funded.
            let builder_account = Self::builder_key().fee_account();
            tracing::info!(%builder_account, "prefunding builder account");
            state.prefund_account(builder_account, U256::MAX.into());

            let persistence = persistence_opt.create().await.unwrap();

            let chain_config = state.chain_config.resolve().unwrap_or_default();

            // Create an empty list of catchup providers
            let catchup_providers = ParallelStateCatchup::new(&[], Duration::from_secs(5));

            // If we have the state peers, add them
            if let Some(state_peers) = state_peers {
                catchup_providers.add_provider(Arc::new(state_peers));
            }

            // If we have a working local catchup provider, add it
            // Local is a single attempt per call; remote providers retry.
            match persistence
                .clone()
                .into_catchup_provider(BackoffParams::disabled())
            {
                Ok(local_catchup) => {
                    catchup_providers.add_provider(local_catchup);
                },
                Err(e) => {
                    tracing::warn!(
                        "Failed to create local catchup provider: {e:#}. Only using remote \
                         catchup."
                    );
                },
            };

            let l1_client = self
                .l1_opt
                .clone()
                .connect(vec![self.l1_url.clone()])
                .expect("failed to create L1 client");
            l1_client.spawn_tasks().await;

            let fetcher = Fetcher::new(
                Arc::new(catchup_providers.clone()),
                Arc::new(Mutex::new(persistence.clone())),
                l1_client.clone(),
                chain_config,
            );
            fetcher.spawn_update_loop().await;

            let block_reward = fetcher.fetch_fixed_block_reward().await.ok();
            let mut membership = EpochCommittees::new(
                config.known_nodes_with_stake.clone(),
                config.known_da_nodes.clone(),
                block_reward,
                fetcher,
                config.epoch_height,
                EpochNumber::new(epoch_from_block_number(
                    config.epoch_start_block,
                    config.epoch_height,
                )),
            );
            membership.reload_stake(50).await;

            let membership = Arc::new(membership);
            let persistence = Arc::new(persistence);

            let coordinator = EpochMembershipCoordinator::new(
                membership,
                config.epoch_height,
                &persistence.clone(),
            );

            let max_block_size = *chain_config.max_block_size;
            let mut node_state = NodeState::new(
                i as u64,
                chain_config,
                l1_client,
                Arc::new(catchup_providers.clone()),
                upgrade.base,
                coordinator.clone(),
                upgrade.base,
            )
            .with_current_version(upgrade.base)
            .with_genesis(state)
            .with_epoch_height(config.epoch_height)
            .with_upgrades(upgrades)
            .with_epoch_start_block(config.epoch_start_block);
            // Only the genesis header: `chain_config` stays what this node resolves from its
            // state, so a node that starts from a commitment still has to fetch the config.
            if let Some(genesis_chain_config) = self.genesis_chain_config {
                node_state.genesis_header.chain_config = genesis_chain_config;
                node_state.genesis_chain_config = genesis_chain_config;
            }

            tracing::info!(
                i,
                key = %pub_key,
                state_key = %self.state_key_pairs[i].ver_key(),
                "starting node",
            );

            let coordinator_network = move |upgrade| {
                Cliquenet::create(
                    "test-coordinator",
                    pub_key,
                    x25519_keypair,
                    coordinator_addr,
                    [],
                    Some(message_limit(max_block_size)),
                    upgrade,
                    Box::new(NoMetrics),
                )
            };

            let (initializer, anchor_view) = persistence
                .load_consensus_state(node_state, upgrade)
                .await
                .unwrap();

            SequencerContext::init(
                NetworkConfig {
                    config,
                    // For testing, we use a fake network, so the rest of the network config beyond
                    // the base consensus config does not matter.
                    ..Default::default()
                },
                upgrade,
                validator_config,
                coordinator,
                initializer,
                anchor_view,
                storage,
                catchup_providers,
                persistence,
                coordinator_network,
                self.state_relay_url.clone(),
                metrics,
                stake_table_capacity,
                event_consumer,
                Default::default(),
                Duration::from_secs(2),
                Duration::from_millis(500),
                BTreeMap::from([(upgrade.base, max_block_size)]),
                None,
            )
            .await
        }

        pub fn builder_key() -> EthKeyPair {
            FeeAccount::generated_from_seed_indexed([1; 32], 0).1
        }
    }

    /// The leaves a decide event finalizes, newest first.
    pub fn decided_leaves(event: &CoordinatorEvent<SeqTypes>) -> Option<&[LeafInfo<SeqTypes>]> {
        match event {
            CoordinatorEvent::NewDecide { leaf_infos, .. } => Some(leaf_infos),
            _ => None,
        }
    }

    /// Wait for `submitted_txn` to be decided; returns its block number and payload size.
    pub async fn wait_for_decide_on_handle(
        events: &mut (impl Stream<Item = CoordinatorEvent<SeqTypes>> + Unpin),
        submitted_txn: &Transaction,
    ) -> (u64, usize) {
        let commitment = submitted_txn.commit();

        // At 0.6 a decide carries the block payload if the node built or
        // reconstructed it by then; otherwise the payload follows in a
        // separate `BlockPayloadReconstructed` event. Pair the two by view so
        // the transaction is only reported once its block is decided.
        let mut reconstructed = HashMap::new();
        let mut decided_without_payload = HashSet::new();

        let (height, size) = timeout(Duration::from_secs(120), async {
            loop {
                let event = events.next().await.unwrap();
                tracing::info!("Received event from handle: {event:?}");

                if let CoordinatorEvent::BlockPayloadReconstructed {
                    view,
                    header,
                    payload,
                } = &event
                {
                    if payload
                        .transaction_commitments(header.metadata())
                        .contains(&commitment)
                    {
                        let found = (header.block_number(), payload.encode().len());
                        if decided_without_payload.contains(view) {
                            return found;
                        }
                        reconstructed.insert(*view, found);
                    }
                    continue;
                }

                let Some(leaf_chain) = decided_leaves(&event) else {
                    continue;
                };
                for LeafInfo { leaf, .. } in leaf_chain {
                    let Some(payload) = leaf.block_payload() else {
                        let view = leaf.view_number();
                        if let Some(found) = reconstructed.remove(&view) {
                            return found;
                        }
                        decided_without_payload.insert(view);
                        continue;
                    };
                    if payload
                        .transaction_commitments(leaf.block_header().metadata())
                        .contains(&commitment)
                    {
                        return (leaf.block_header().block_number(), payload.encode().len());
                    }
                }
            }
        })
        .await
        .unwrap_or_else(|_| {
            panic!("transaction {commitment} was not sequenced within the timeout")
        });
        tracing::info!(height, "transaction {commitment} sequenced");
        (height, size)
    }

    /// Waits until a node has reached the given target epoch (exclusive).
    /// The function returns once the first event indicates an epoch higher than `target_epoch`.
    /// Panics if the event stream ends before reaching `target_epoch`.
    pub async fn wait_for_epochs(
        events: &mut (impl futures::Stream<Item = CoordinatorEvent<SeqTypes>> + std::marker::Unpin),
        epoch_height: u64,
        target_epoch: u64,
    ) {
        tracing::info!(target_epoch, "waiting for epoch");
        let mut last_seen = None;
        while let Some(event) = events.next().await {
            // A decide carries its most recent leaf first.
            let leaf = match event {
                CoordinatorEvent::NewDecide { leaf_infos, .. } => leaf_infos[0].leaf.clone(),
                _ => continue,
            };
            let epoch = leaf.epoch(epoch_height);
            tracing::debug!(
                "Node decided at height: {}, epoch: {epoch:?}",
                leaf.height(),
            );

            if epoch > Some(EpochNumber::new(target_epoch)) {
                tracing::info!(target_epoch, "epoch started");
                return;
            }
            last_seen = Some((leaf.height(), epoch));
        }
        panic!("event stream ended before target epoch {target_epoch}, last decide: {last_seen:?}");
    }
}

#[cfg(test)]
mod test {
    use alloy::signers::local::coins_bip39::{English, Mnemonic};
    use espresso_keyset::KeySet;
    use espresso_types::{Header, NamespaceId, TEST_UPGRADE, Transaction};
    use futures::StreamExt;
    use hotshot_types::{
        PeerConnectInfo, addr::NetAddr, event::LeafInfo, traits::block_contents::BlockHeader,
        x25519,
    };
    use test_utils::reserve_tcp_port;
    use testing::{TestConfigBuilder, decided_leaves, wait_for_decide_on_handle};

    use super::{local_validator_config, orchestrator_registration};
    use crate::api::{
        Options,
        test_helpers::{TestNetwork, TestNetworkConfigBuilder},
    };

    fn test_keys() -> KeySet {
        let mnemonic = Mnemonic::<English>::new_from_phrase(
            "test test test test test test test test test test test junk",
        )
        .unwrap();
        KeySet::from_mnemonic(mnemonic, Some(7)).unwrap()
    }

    /// `status/keys` reports the x25519 key and address from the validator config, so the config
    /// must hold the identity cliquenet runs with. The address is only the advertise address:
    /// without one, nothing dialable is known, so no `connect_info` is reported. This covers the
    /// helper only: `init_node` is not exercised in-process, so nothing here catches a
    /// config-source path that stops using the helper.
    #[test]
    fn local_validator_config_records_cliquenet_identity() {
        let keys = test_keys();
        let advertise: NetAddr = "node.example.com:9977".parse().unwrap();
        let x25519_key = x25519::Keypair::from(&keys.x25519).public_key();

        let with_addr = local_validator_config(
            keys.staking.clone(),
            keys.state.clone(),
            &keys.x25519,
            Some(advertise.clone()),
            false,
        );
        assert_eq!(
            with_addr.public_config().connect_info,
            Some(PeerConnectInfo {
                x25519_key,
                p2p_addr: advertise,
            })
        );

        let without_addr =
            local_validator_config(keys.staking, keys.state, &keys.x25519, None, false);
        assert_eq!(
            without_addr
                .x25519_keypair
                .as_ref()
                .map(|kp| kp.public_key()),
            Some(x25519_key)
        );
        assert!(without_addr.public_config().connect_info.is_none());
    }

    #[test]
    fn orchestrator_registration_publishes_connect_info() {
        let keys = test_keys();
        let advertise: NetAddr = "node.example.com:9977".parse().unwrap();
        let config = local_validator_config(
            keys.staking.clone(),
            keys.state.clone(),
            &keys.x25519,
            Some(advertise.clone()),
            false,
        );

        let registration = orchestrator_registration(&config).unwrap();
        assert_eq!(
            registration
                .public_config()
                .connect_info
                .map(|i| i.p2p_addr),
            Some(advertise)
        );

        let unconfigured =
            local_validator_config(keys.staking, keys.state, &keys.x25519, None, false);
        assert!(orchestrator_registration(&unconfigured).is_err());
    }

    #[test]
    fn telemetry_endpoint_defaults_by_chain() {
        use super::{
            ChainId, DECAF_TELEMETRY_ENDPOINT, MAINNET_CHAIN_ID, MAINNET_TELEMETRY_ENDPOINT, U256,
            default_telemetry_endpoint,
        };

        assert_eq!(
            default_telemetry_endpoint(MAINNET_CHAIN_ID),
            Some(MAINNET_TELEMETRY_ENDPOINT)
        );
        assert_eq!(
            default_telemetry_endpoint(DECAF_CHAIN_ID),
            Some(DECAF_TELEMETRY_ENDPOINT)
        );
        // Unknown chains (e.g. the demo chain) get no default.
        assert_eq!(
            default_telemetry_endpoint(ChainId(U256::from(999999999u64))),
            None
        );
    }

    #[test]
    fn apply_genesis_overrides_writes_genesis_values() {
        use hotshot_types::light_client::DEFAULT_STAKE_TABLE_CAPACITY;
        use vbs::version::Version;

        use crate::genesis::StakeTableConfig;

        let version = Version { major: 0, minor: 1 };
        let mut genesis = Genesis {
            chain_config: Default::default(),
            stake_table: StakeTableConfig { capacity: 10 },
            accounts: Default::default(),
            l1_finalized: L1Finalized::Number { number: 0 },
            header: Default::default(),
            upgrades: Default::default(),
            base_version: version,
            upgrade_version: version,
            genesis_version: version,
            epoch_height: Some(30),
            drb_difficulty: Some(10),
            drb_upgrade_difficulty: Some(20),
            epoch_start_block: Some(100),
            stake_table_capacity: None,
            da_committees: None,
        };
        let mut config = NetworkConfig::<SeqTypes>::default();

        apply_genesis_overrides(&mut config, &genesis);
        assert_eq!(config.config.epoch_height, 30);
        assert_eq!(config.config.drb_difficulty, 10);
        assert_eq!(config.config.drb_upgrade_difficulty, 20);
        assert_eq!(config.config.epoch_start_block, 100);
        assert_eq!(
            config.config.stake_table_capacity,
            DEFAULT_STAKE_TABLE_CAPACITY
        );

        genesis.stake_table_capacity = Some(50);
        apply_genesis_overrides(&mut config, &genesis);
        assert_eq!(config.config.stake_table_capacity, 50);
    }

    use super::*;

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_skeleton_instantiation() {
        let port = reserve_tcp_port().expect("OS should have ephemeral ports available");
        let config = TestNetworkConfigBuilder::default()
            .api_config(Options::with_port(port))
            .network_config(TestConfigBuilder::default().build())
            .build();
        let network = TestNetwork::new(config, TEST_UPGRADE).await;
        let mut events = network.server.event_stream();

        let txn = Transaction::new(NamespaceId::from(1_u32), vec![1, 2, 3]);
        network
            .server
            .submit_transaction(txn.clone())
            .await
            .expect("Failed to submit transaction");
        tracing::info!("Submitted transaction to handle: {txn:?}");

        wait_for_decide_on_handle(&mut events, &txn).await;
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_header_invariants() {
        let success_height = 30;
        let port = reserve_tcp_port().expect("OS should have ephemeral ports available");
        let config = TestNetworkConfigBuilder::default()
            .api_config(Options::with_port(port))
            .network_config(TestConfigBuilder::default().build())
            .build();
        let network = TestNetwork::new(config, TEST_UPGRADE).await;
        let mut events = network.server.event_stream();

        // The network is already running, so start from the first header it
        // decides after we subscribe.
        let mut parent: Option<Header> = None;

        loop {
            let event = events.next().await.unwrap();
            tracing::info!("Received event from handle: {event:?}");
            let Some(leaf_chain) = decided_leaves(&event) else {
                continue;
            };
            tracing::info!("Got decide {leaf_chain:?}");

            // Check that each successive header satisfies invariants relative to its parent: all
            // the fields which should be monotonic are.
            for LeafInfo { leaf, .. } in leaf_chain.iter().rev() {
                let header = leaf.block_header().clone();
                if let Some(parent) = &parent {
                    assert_eq!(header.height(), parent.height() + 1);
                    assert!(header.timestamp() >= parent.timestamp());
                    assert!(header.l1_head() >= parent.l1_head());
                    assert!(header.l1_finalized() >= parent.l1_finalized());
                }
                parent = Some(header);
            }

            if parent
                .as_ref()
                .is_some_and(|p| p.height() >= success_height)
            {
                break;
            }
        }
    }

    #[rstest::rstest]
    #[case::missing(None)]
    #[case::zero(Some(0))]
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn init_node_state_rejects_epochless_genesis_before_l1(
        #[case] epoch_height: Option<u64>,
    ) {
        use espresso_types::v0::traits::PersistenceOptions;

        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../../../data/genesis/mainnet.toml");
        let mut genesis = Genesis::from_file(&path).unwrap();
        genesis.epoch_height = epoch_height;
        // Nothing listens on the L1 port and this block is never finalized, so L1 work cannot finish.
        genesis.l1_finalized = L1Finalized::Number { number: u64::MAX };
        let port = reserve_tcp_port().expect("OS should have ephemeral ports available");
        let l1_params = L1Params {
            urls: vec![format!("http://127.0.0.1:{port}").parse().unwrap()],
            options: L1ClientOptions::default(),
        };
        let catchup = CatchupParams {
            state_peers: vec![],
            backoff: Default::default(),
            base_timeout: Duration::from_secs(2),
            local_timeout: Duration::from_secs(5),
            http_delay: Duration::ZERO,
        };
        let tmp = tempfile::TempDir::new().unwrap();
        let persistence = persistence::fs::Options::new(tmp.path().into())
            .create()
            .await
            .unwrap();

        let result = tokio::time::timeout(
            Duration::from_secs(10),
            init_node_state(
                &genesis,
                &NetworkConfig::<SeqTypes>::default(),
                l1_params,
                catchup,
                persistence,
                &NoMetrics,
            ),
        )
        .await;

        let err = match result {
            Ok(Err(err)) => err,
            Ok(Ok(_)) => panic!("init_node_state accepted epoch_height={epoch_height:?}"),
            Err(_) => panic!("init_node_state did not return within 10s: it is doing L1 work"),
        };
        assert!(
            format!("{err:#}").contains("epoch_height"),
            "expected an epoch_height error, got: {err:#}"
        );
    }
}