forest/daemon/

mod.rs

// Copyright 2019-2026 ChainSafe Systems
// SPDX-License-Identifier: Apache-2.0, MIT

pub mod bundle;
mod context;
pub mod db_util;
pub mod main;

use crate::blocks::TipsetKey;
use crate::chain::ChainStore;
use crate::chain_sync::ChainFollower;
use crate::chain_sync::network_context::SyncNetworkContext;
use crate::cli_shared::snapshot;
use crate::cli_shared::{
    chain_path,
    cli::{CliOpts, Config},
    delete_chain_data,
};
use crate::daemon::{context::AppContext, db_util::import_chain_as_forest_car};
use crate::db::gc::SnapshotGarbageCollector;
use crate::db::ttl::EthMappingCollector;
use crate::libp2p::{Libp2pService, PeerManager};
use crate::message_pool::{MessagePool, MpoolConfig, MpoolLocker, NonceTracker};
use crate::networks::{self, ChainConfig};
use crate::prelude::*;
use crate::rpc::RPCState;
use crate::rpc::eth::filter::EthEventHandler;
use crate::rpc::start_rpc;
use crate::shim::address::Address;
use crate::shim::clock::ChainEpoch;
use crate::shim::state_tree::StateTree;
use crate::shim::version::NetworkVersion;
use crate::state_manager::StateManager;
use crate::utils::misc::env::is_env_truthy;
use crate::utils::{self};
use crate::utils::{proofs_api::ensure_proof_params_downloaded, version::FOREST_VERSION_STRING};
use anyhow::{Context as _, bail};
use backon::{ExponentialBuilder, Retryable};
use dialoguer::theme::ColorfulTheme;
use futures::{Future, FutureExt};
use std::path::Path;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::{Duration, Instant};
use tokio::sync::broadcast::error::RecvError;
use tokio::{
    signal::{
        ctrl_c,
        unix::{SignalKind, signal},
    },
    sync::mpsc,
    task::JoinSet,
};
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

pub static GLOBAL_SNAPSHOT_GC: OnceLock<Arc<SnapshotGarbageCollector>> = OnceLock::new();

/// Increase the file descriptor limit to a reasonable number.
/// This prevents the node from failing if the default soft limit is too low.
/// Note that the value is only increased, never decreased.
fn maybe_increase_fd_limit() -> anyhow::Result<()> {
    static DESIRED_SOFT_LIMIT: u64 = 8192;
    let (soft_before, _) = rlimit::Resource::NOFILE.get()?;

    let soft_after = rlimit::increase_nofile_limit(DESIRED_SOFT_LIMIT)?;
    if soft_before < soft_after {
        debug!("Increased file descriptor limit from {soft_before} to {soft_after}");
    }
    if soft_after < DESIRED_SOFT_LIMIT {
        warn!(
            "File descriptor limit is too low: {soft_after} < {DESIRED_SOFT_LIMIT}. \
            You may encounter 'too many open files' errors.",
        );
    }

    Ok(())
}

// Start the daemon and abort if we're interrupted by ctrl-c, SIGTERM, or `forest-cli shutdown`.
pub async fn start_interruptable(opts: CliOpts, config: Config) -> anyhow::Result<()> {
    let start_time = chrono::Utc::now();
    let mut terminate = signal(SignalKind::terminate())?;
    let (shutdown_send, mut shutdown_recv) = mpsc::channel(1);
    let (rpc_stop_handle, rpc_server_handle) = jsonrpsee::server::stop_channel();
    let result = tokio::select! {
        ret = start(start_time, opts, config, shutdown_send, rpc_stop_handle) => ret,
        _ = ctrl_c() => {
            info!("Keyboard interrupt.");
            Ok(())
        },
        _ = terminate.recv() => {
            info!("Received SIGTERM.");
            Ok(())
        },
        _ = shutdown_recv.recv() => {
            info!("Client requested a shutdown.");
            Ok(())
        },
    };
    _ = rpc_server_handle.stop();
    crate::utils::io::terminal_cleanup();
    result
}

/// This function initialize Forest with below steps
/// - increase file descriptor limit (for parity-db)
/// - setup proofs parameter cache directory
/// - prints Forest version
fn startup_init(config: &Config) -> anyhow::Result<()> {
    maybe_increase_fd_limit()?;
    // Sets proof parameter file download path early, the files will be checked and
    // downloaded later right after snapshot import step
    crate::utils::proofs_api::maybe_set_proofs_parameter_cache_dir_env(&config.client.data_dir);
    info!(
        "Starting Forest daemon, version {}",
        FOREST_VERSION_STRING.as_str()
    );
    info!("Using data directory: {}", config.client.data_dir.display());
    Ok(())
}

async fn maybe_import_snapshot(
    opts: &CliOpts,
    config: &mut Config,
    ctx: &AppContext,
) -> anyhow::Result<()> {
    let chain_config = ctx.state_manager.chain_config();
    // Sets the latest snapshot if needed for downloading later
    if config.client.snapshot_path.is_none() && !opts.stateless {
        maybe_set_snapshot_path(
            config,
            chain_config,
            ctx.state_manager.chain_store().heaviest_tipset().epoch(),
            opts.auto_download_snapshot,
            &ctx.db_meta_data.get_root_dir(),
        )
        .await?;
    }

    let snapshot_tracker = ctx.snapshot_progress_tracker.clone();
    // Import chain if needed
    if !opts.skip_load.unwrap_or_default()
        && let Some(path) = &config.client.snapshot_path
    {
        let (car_db_path, ts) = import_chain_as_forest_car(
            path,
            &ctx.db_meta_data.get_forest_car_db_dir(),
            config.client.import_mode,
            config.client.rpc_v1_endpoint()?,
            &crate::f3::get_f3_root(config),
            ctx.chain_config(),
            &snapshot_tracker,
        )
        .await?;
        ctx.db
            .read_only_files(std::iter::once(car_db_path.clone()))?;
        let ts_epoch = ts.epoch();
        // Explicitly set heaviest tipset here in case HEAD_KEY has already been set
        // in the current setting store
        ctx.state_manager.chain_store().set_heaviest_tipset(ts)?;
        debug!(
            "Loaded car DB at {} and set current head to epoch {ts_epoch}",
            car_db_path.display(),
        );
    }

    // If the snapshot progress state is not completed,
    // set the state to not required
    if !snapshot_tracker.is_completed() {
        snapshot_tracker.not_required();
    }

    if let Some(validate_from) = config.client.snapshot_height {
        // We've been provided a snapshot and asked to validate it
        ensure_proof_params_downloaded().await?;
        // Use the specified HEAD, otherwise take the current HEAD.
        let current_height = config
            .client
            .snapshot_head
            .unwrap_or_else(|| ctx.state_manager.chain_store().heaviest_tipset().epoch());

        let validation_range = validation_range(current_height, validate_from)?;
        // `validate_range` is CPU-bound (drives rayon-parallel VM execution) and
        // can run for minutes. Safer to spawn it on a blocking thread.
        let state_manager = ctx.state_manager.shallow_clone();
        tokio::task::spawn_blocking(move || {
            state_manager.validate_range_blocking(validation_range)
        })
        .await??;
    }

    Ok(())
}

/// Returns the range of epochs to validate. This includes special handling for negative `from`
/// values, which are interpreted as offsets from the current epoch.
fn validation_range(
    current: ChainEpoch,
    from: ChainEpoch,
) -> anyhow::Result<std::ops::RangeInclusive<ChainEpoch>> {
    anyhow::ensure!(
        current.is_positive(),
        "current head epoch {current} is invalid"
    );

    // Negative values scroll back from the current head (e.g. --height=-1000).
    // `saturating_add` + `.max(0)` keeps extreme negatives from underflowing or
    // wrapping to a huge positive (which would silently produce an empty range).
    let start = if from.is_negative() {
        current.saturating_add(from).max(0)
    } else {
        from
    };

    // An absolute `--height` past the head would otherwise produce an empty
    // range and silently succeed without validating anything.
    anyhow::ensure!(
        start <= current,
        "requested validation start epoch {start} is beyond the current head at epoch {current}",
    );

    Ok(start..=current)
}

async fn maybe_start_metrics_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    config: &Config,
    ctx: &AppContext,
) -> anyhow::Result<()> {
    if config.client.enable_metrics_endpoint {
        let prometheus_listener =
            crate::utils::net::bind_tcp_listener(config.client.metrics_address, 0).await?;
        info!(
            "Prometheus server started at {}",
            config.client.metrics_address
        );
        let db_directory = crate::db::db_engine::db_root(&chain_path(config))?;
        let db = ctx.db.writer().clone();

        let get_chain_head_height = Arc::new({
            let cs = ctx.state_manager.chain_store().shallow_clone();
            move || cs.heaviest_tipset().epoch()
        });
        let get_chain_head_actor_version = Arc::new({
            let cs = ctx.state_manager.chain_store().shallow_clone();
            move || {
                if let Ok(state) =
                    StateTree::new_from_root(cs.db(), cs.heaviest_tipset().parent_state())
                    && let Ok(bundle_meta) = state.get_actor_bundle_metadata()
                    && let Ok(actor_version) = bundle_meta.actor_major_version()
                {
                    actor_version
                } else {
                    0
                }
            }
        });
        services.spawn({
            let chain_config = ctx.chain_config().clone();
            let get_chain_head_height = get_chain_head_height.clone();
            async {
                crate::metrics::init_prometheus(
                    prometheus_listener,
                    db_directory,
                    db,
                    chain_config,
                    get_chain_head_height,
                    get_chain_head_actor_version,
                )
                .await
                .context("Failed to initiate prometheus server")
            }
        });

        crate::metrics::register_collector(Box::new(
            networks::metrics::NetworkHeightCollector::new(
                ctx.state_manager.chain_config().block_delay_secs,
                ctx.state_manager
                    .chain_store()
                    .genesis_block_header()
                    .timestamp,
                get_chain_head_height,
            ),
        ));
    }
    Ok(())
}

async fn create_p2p_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    config: &mut Config,
    ctx: &AppContext,
) -> anyhow::Result<Libp2pService> {
    // if bootstrap peers are not set, set them
    if config.network.bootstrap_peers.is_empty() {
        config.network.bootstrap_peers = ctx.state_manager.chain_config().bootstrap_peers.clone();
    }

    let peer_manager = Arc::new(PeerManager::default());
    services.spawn(peer_manager.clone().peer_operation_event_loop_task());
    // Libp2p service setup
    let p2p_service = Libp2pService::new(
        config.network.clone(),
        ctx.state_manager.chain_store().shallow_clone(),
        peer_manager.clone(),
        ctx.net_keypair.clone(),
        config.chain.genesis_name(),
        *ctx.state_manager.chain_store().genesis_block_header().cid(),
    )
    .await?;
    Ok(p2p_service)
}

fn create_mpool(
    services: &mut JoinSet<anyhow::Result<()>>,
    p2p_service: &Libp2pService,
    ctx: &AppContext,
) -> anyhow::Result<MessagePool<ChainStore>> {
    Ok(MessagePool::new(
        ctx.state_manager.chain_store().shallow_clone(),
        p2p_service.network_sender().clone(),
        MpoolConfig::load_config(ctx.db.writer().as_ref())?,
        ctx.state_manager.chain_config().clone(),
        services,
    )?)
}

fn create_chain_follower(
    opts: &CliOpts,
    p2p_service: &Libp2pService,
    mpool: MessagePool<ChainStore>,
    ctx: &AppContext,
) -> anyhow::Result<ChainFollower> {
    let network_send = p2p_service.network_sender().clone();
    let peer_manager = p2p_service.peer_manager().clone();
    let network = SyncNetworkContext::new(network_send, peer_manager, ctx.db.clone().into());
    Ok(ChainFollower::new(
        ctx.state_manager.shallow_clone(),
        network,
        ctx.state_manager.chain_store().genesis_tipset(),
        p2p_service.network_receiver(),
        opts.stateless,
        mpool,
    ))
}

fn start_chain_follower_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    opts: &CliOpts,
    config: &Config,
    chain_follower: ChainFollower,
) {
    services.spawn({
        let chain_follower = chain_follower.shallow_clone();
        async move { chain_follower.run().await }
    });
    maybe_prefill_rpc_caches(services, opts, config, chain_follower);
}

fn maybe_prefill_rpc_caches(
    services: &mut JoinSet<anyhow::Result<()>>,
    opts: &CliOpts,
    config: &Config,
    chain_follower: ChainFollower,
) {
    // Prefill RPC method caches for newly validated tipsets to speed up subsequent RPC calls.
    if config.client.enable_rpc && !opts.stateless {
        let sync_status = chain_follower.sync_status.shallow_clone();
        let state_manager = chain_follower.state_manager.shallow_clone();
        let mut validated_tipset_rx = chain_follower.subscribe_validated_tipset();
        services.spawn(async move {
            let cancellation_token = CancellationToken::new();
            let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref();
            loop {
                match validated_tipset_rx.recv().await {
                    Ok(_) if !sync_status.load().is_synced() => {
                        // Skip if the node is catching up to avoid unnecessary work, as the head may be changing rapidly.
                        continue;
                    }
                    Ok(tsk) => {
                        let state_manager = state_manager.shallow_clone();
                        let cancellation_token = cancellation_token.clone();
                        tokio::spawn(async move {
                            cancellation_token
                                .run_until_cancelled(prefill_rpc_caches_for_tipset(
                                    state_manager,
                                    tsk,
                                ))
                                .await
                        });
                    }
                    Err(RecvError::Lagged(n)) => {
                        warn!("validated tipset broadcast lagged: skipped {n} tipsets")
                    }
                    Err(RecvError::Closed) => break Ok(()),
                }
            }
        });
    }
}

async fn prefill_rpc_caches_for_tipset(state_manager: StateManager, tsk: TipsetKey) {
    match state_manager.chain_index().load_required_tipset(&tsk) {
        Ok(ts) => {
            {
                // First, compute state for the ts as it's disallowed for RPC methods by default
                if let Err(e) = state_manager.load_executed_tipset(&ts).await {
                    warn!("failed to load executed tipset for cache warmup: {e:#}");
                    return; // Skip when state computation fails
                }
            }
            for tx_info in [crate::rpc::eth::TxInfo::Full, crate::rpc::eth::TxInfo::Hash] {
                if let Err(e) = crate::rpc::eth::Block::from_filecoin_tipset(
                    &state_manager,
                    ts.shallow_clone(),
                    tx_info,
                )
                .await
                {
                    warn!("failed to call `Block::from_filecoin_tipset` for cache warmup: {e:#}");
                }
            }
            {
                // Warms both the FVM-replay cache and the parity-trace cache,
                // since `eth_trace_block` calls `execution_trace` internally.
                if let Err(e) = crate::rpc::eth::eth_trace_block(&state_manager, &ts).await {
                    warn!("failed to call `eth_trace_block` for cache warmup: {e:#}");
                }
            }
            {
                use crate::rpc::eth::filter::{Matcher, SkipEvent};
                struct CollectEventsCachePrefillingMatcher;
                impl Matcher for CollectEventsCachePrefillingMatcher {
                    fn msg_cid_filter(&self) -> Option<&Cid> {
                        None
                    }
                    fn matches(
                        &self,
                        _: &Address,
                        _: &[crate::shim::executor::Entry],
                    ) -> anyhow::Result<bool> {
                        Ok(false)
                    }
                }
                let mut collected_events = vec![];
                if let Err(e) = EthEventHandler::collect_events(
                    &state_manager,
                    &ts,
                    Some(&CollectEventsCachePrefillingMatcher),
                    SkipEvent::OnUnresolvedAddress,
                    &mut collected_events,
                )
                .await
                {
                    warn!("failed to collect events for cache warmup: {e:#}");
                }
            }
        }
        Err(e) => {
            warn!("failed to load tipset for cache warmup: {e:#}");
        }
    }
}

async fn maybe_start_health_check_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    config: &Config,
    p2p_service: &Libp2pService,
    chain_follower: &ChainFollower,
    ctx: &AppContext,
) -> anyhow::Result<()> {
    if config.client.enable_health_check {
        let forest_state = crate::health::ForestState {
            config: config.clone(),
            chain_config: ctx.state_manager.chain_config().clone(),
            genesis_timestamp: ctx
                .state_manager
                .chain_store()
                .genesis_block_header()
                .timestamp,
            sync_status: chain_follower.sync_status.clone(),
            peer_manager: p2p_service.peer_manager().clone(),
        };
        let healthcheck_address = forest_state.config.client.healthcheck_address;
        info!("Healthcheck endpoint will listen at {healthcheck_address}");
        let listener = crate::utils::net::bind_tcp_listener(healthcheck_address, 0).await?;
        services.spawn(async move {
            crate::health::init_healthcheck_server(forest_state, listener)
                .await
                .context("Failed to initiate healthcheck server")
        });
    } else {
        info!("Healthcheck service is disabled");
    }
    Ok(())
}

fn maybe_start_gc_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    opts: &CliOpts,
    config: &Config,
    chain_follower: ChainFollower,
) -> anyhow::Result<()> {
    // If the node is stateless, GC shouldn't get triggered even on demand.
    if opts.stateless {
        return Ok(());
    }

    let snap_gc = Arc::new(SnapshotGarbageCollector::new(chain_follower, config)?);

    GLOBAL_SNAPSHOT_GC
        .set(snap_gc.clone())
        .ok()
        .context("failed to set GLOBAL_SNAPSHOT_GC")?;

    services.spawn({
        let snap_gc = snap_gc.clone();
        async move {
            snap_gc.event_loop().await;
            Ok(())
        }
    });

    // GC shouldn't run periodically if the node is stateless or if the user has disabled it.
    if !opts.no_gc {
        services.spawn({
            let snap_gc = snap_gc.clone();
            async move {
                snap_gc.scheduler_loop().await;
                Ok(())
            }
        });
    }

    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn maybe_start_rpc_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    config: &Config,
    mpool: MessagePool<ChainStore>,
    chain_follower: &ChainFollower,
    start_time: chrono::DateTime<chrono::Utc>,
    shutdown: mpsc::Sender<()>,
    rpc_stop_handle: jsonrpsee::server::StopHandle,
    ctx: &AppContext,
) -> anyhow::Result<()> {
    if config.client.enable_rpc {
        let rpc_address = config.client.rpc_address;
        let metrics_mode = crate::rpc::MetricsMode::from(config.client.enable_metrics_endpoint);
        let filter_list = config
            .client
            .rpc_filter_list
            .as_ref()
            .map(|path| crate::rpc::FilterList::new_from_file(path).map(Arc::new))
            .transpose()?;
        info!("JSON-RPC endpoint will listen at {rpc_address}");
        let eth_event_handler = Arc::new(EthEventHandler::from_config(
            &config.events,
            ctx.chain_config().eth_chain_id,
            mpool.subscriber(),
        ));
        if is_env_truthy("FOREST_JWT_DISABLE_EXP_VALIDATION") {
            warn!(
                "JWT expiration validation is disabled; this significantly weakens security and should only be used in tightly controlled environments"
            );
        }
        services.spawn({
            let state_manager = ctx.state_manager.shallow_clone();
            let bad_blocks = chain_follower.bad_blocks.shallow_clone();
            let sync_status = chain_follower.sync_status.shallow_clone();
            let sync_network_context = chain_follower.network.shallow_clone();
            let tipset_send = chain_follower.tipset_sender.clone();
            let keystore = ctx.keystore.shallow_clone();
            let snapshot_progress_tracker = ctx.snapshot_progress_tracker.clone();
            let nonce_tracker = NonceTracker::new();
            let mpool_locker = MpoolLocker::new();
            let temp_dir = Arc::new(ctx.temp_dir.clone());
            async move {
                let rpc_listener = crate::utils::net::bind_tcp_listener(
                    rpc_address,
                    crate::rpc::default_max_connections(),
                )
                .await?;
                start_rpc(
                    RPCState {
                        state_manager,
                        keystore,
                        mpool,
                        bad_blocks,
                        sync_status,
                        eth_event_handler,
                        eth_logs_feed: Default::default(),
                        sync_network_context,
                        start_time,
                        shutdown,
                        tipset_send,
                        snapshot_progress_tracker,
                        mpool_locker,
                        nonce_tracker,
                        temp_dir,
                    },
                    rpc_listener,
                    rpc_stop_handle,
                    filter_list,
                    metrics_mode,
                )
                .await
            }
        });
    } else {
        debug!("RPC disabled.");
    };
    Ok(())
}

fn maybe_start_f3_service(opts: &CliOpts, config: &Config, ctx: &AppContext) -> anyhow::Result<()> {
    // already running
    if crate::rpc::f3::F3_LEASE_MANAGER.get().is_some() {
        return Ok(());
    }

    if !config.client.enable_rpc {
        if crate::f3::is_sidecar_ffi_enabled(ctx.state_manager.chain_config()) {
            tracing::warn!("F3 sidecar is enabled but not run because RPC is disabled. ")
        }
        return Ok(());
    }

    if !opts.halt_after_import && !opts.stateless {
        let rpc_endpoint = config.client.rpc_v1_endpoint()?;
        let state_manager = &ctx.state_manager;
        let p2p_peer_id = ctx.p2p_peer_id;
        let admin_jwt = ctx.admin_jwt.clone();
        tokio::task::spawn_blocking({
            crate::rpc::f3::F3_LEASE_MANAGER
                .set(crate::rpc::f3::F3LeaseManager::new(
                    state_manager.chain_config().network.clone(),
                    p2p_peer_id,
                ))
                .expect("F3 lease manager should not have been initialized before");
            let chain_config = state_manager.chain_config().clone();
            let f3_root = crate::f3::get_f3_root(config);
            let crate::f3::F3Options {
                chain_finality,
                bootstrap_epoch,
                initial_power_table,
            } = crate::f3::get_f3_sidecar_params(&chain_config);
            move || {
                crate::f3::run_f3_sidecar_if_enabled(
                    &chain_config,
                    rpc_endpoint.to_string(),
                    admin_jwt,
                    crate::rpc::f3::get_f3_rpc_endpoint().to_string(),
                    initial_power_table
                        .map(|i| i.to_string())
                        .unwrap_or_default(),
                    bootstrap_epoch,
                    chain_finality,
                    f3_root.display().to_string(),
                );
            }
        });
        tokio::task::spawn({
            let chain_store = ctx.chain_store().shallow_clone();
            async move {
                // wait 1s to let F3 RPC server start
                tokio::time::sleep(Duration::from_secs(1)).await;
                match (|| crate::rpc::f3::F3GetLatestCertificate::get())
                    .retry(ExponentialBuilder::default())
                    .await
                {
                    Ok(f3_finalized_cert) => {
                        let f3_finalized_head = f3_finalized_cert.chain_head();
                        match chain_store
                            .chain_index()
                            .load_required_tipset(&f3_finalized_head.key)
                        {
                            Ok(ts) => {
                                chain_store.set_f3_finalized_tipset(ts);
                                tracing::info!(
                                    "Set F3 finalized tipset to epoch {} and key {}",
                                    f3_finalized_head.epoch,
                                    f3_finalized_head.key,
                                );
                            }
                            Err(e) => {
                                tracing::error!(
                                    "Failed to get F3 finalized tipset epoch {} and key {}: {e}",
                                    f3_finalized_head.epoch,
                                    f3_finalized_head.key
                                );
                            }
                        }
                    }
                    Err(e) => {
                        tracing::error!("Failed to get F3 latest certificate: {e:#}");
                    }
                }
            }
        });
    }

    Ok(())
}

fn maybe_start_indexer_service(
    services: &mut JoinSet<anyhow::Result<()>>,
    opts: &CliOpts,
    config: &Config,
    ctx: &AppContext,
) {
    if config.chain_indexer.enable_indexer
        && !opts.stateless
        && !ctx.state_manager.chain_config().is_devnet()
    {
        let mut head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes();
        let chain_store = ctx.state_manager.chain_store().shallow_clone();
        services.spawn(async move {
            tracing::info!("Starting indexer service");

            // Continuously listen for head changes
            loop {
                match head_changes_rx.recv().await {
                    Ok(changes) => {
                        for ts in changes.applies {
                            tracing::debug!("Indexing tipset {}", ts.key());
                            let delegated_messages = chain_store
                                .headers_delegated_messages(ts.block_headers().iter())?;
                            chain_store.process_signed_messages(&delegated_messages)?;
                        }
                    }
                    Err(RecvError::Lagged(n)) => {
                        warn!("indexer service lagged: skipping {n} events")
                    }
                    Err(RecvError::Closed) => break Ok(()),
                }
            }
        });

        // Run the collector only if chain indexer is enabled
        if let Some(retention_epochs) = config.chain_indexer.gc_retention_epochs {
            let chain_store = ctx.state_manager.chain_store().shallow_clone();
            let chain_config = ctx.state_manager.chain_config().clone();
            services.spawn(async move {
                tracing::info!("Starting collector for eth_mappings");
                let mut collector = EthMappingCollector::new(
                    chain_store.db_owned(),
                    chain_config.eth_chain_id,
                    retention_epochs.into(),
                );
                collector.run().await
            });
        }
    }
}

/// Starts daemon process
pub(super) async fn start(
    start_time: chrono::DateTime<chrono::Utc>,
    opts: CliOpts,
    config: Config,
    shutdown_send: mpsc::Sender<()>,
    rpc_stop_handle: jsonrpsee::server::StopHandle,
) -> anyhow::Result<()> {
    startup_init(&config)?;
    if opts.remove_existing_chain {
        warn!("Deleting existing chain data for {}", config.chain());
        delete_chain_data(&config)?;
    }
    start_services(
        start_time,
        &opts,
        config.clone(),
        shutdown_send.clone(),
        rpc_stop_handle,
    )
    .await
}

pub(super) async fn start_services(
    start_time: chrono::DateTime<chrono::Utc>,
    opts: &CliOpts,
    mut config: Config,
    shutdown_send: mpsc::Sender<()>,
    rpc_stop_handle: jsonrpsee::server::StopHandle,
) -> anyhow::Result<()> {
    // Cleanup the collector prometheus metrics registry on start
    crate::metrics::reset_collector_registry();
    let mut services = JoinSet::new();
    let network = config.chain();
    let ctx = AppContext::init(opts, &config).await?;
    info!("Using network :: {network}");
    utils::misc::display_chain_logo(config.chain());
    if opts.exit_after_init {
        return Ok(());
    }
    if !opts.stateless
        && !opts.skip_load_actors
        && let Err(e) = ctx.state_manager.maybe_rewind_heaviest_tipset().await
    {
        tracing::warn!("error in maybe_rewind_heaviest_tipset: {e:#}");
    }

    let p2p_service = create_p2p_service(&mut services, &mut config, &ctx).await?;
    let mpool = create_mpool(&mut services, &p2p_service, &ctx)?;
    let chain_follower = create_chain_follower(opts, &p2p_service, mpool.shallow_clone(), &ctx)?;

    maybe_start_rpc_service(
        &mut services,
        &config,
        mpool.shallow_clone(),
        &chain_follower,
        start_time,
        shutdown_send.clone(),
        rpc_stop_handle,
        &ctx,
    )?;

    maybe_import_snapshot(opts, &mut config, &ctx).await?;
    if opts.halt_after_import {
        // Cancel all async services
        services.shutdown().await;
        return Ok(());
    }

    warmup_in_background(&ctx);
    maybe_start_gc_service(&mut services, opts, &config, chain_follower.shallow_clone())?;
    maybe_start_metrics_service(&mut services, &config, &ctx).await?;
    maybe_start_f3_service(opts, &config, &ctx)?;
    maybe_start_health_check_service(&mut services, &config, &p2p_service, &chain_follower, &ctx)
        .await?;
    maybe_start_indexer_service(&mut services, opts, &config, &ctx);
    if !opts.stateless {
        ensure_proof_params_downloaded().await?;
    }
    services.spawn(p2p_service.run());
    start_chain_follower_service(&mut services, opts, &config, chain_follower);
    // blocking until any of the services returns an error,
    propagate_error(&mut services)
        .await
        .context("services failure")
        .map(|_| {})
}

fn warmup_in_background(ctx: &AppContext) {
    // Verify and re-populate the `tipset_by_height` lookup table over the whole chain.
    let cs = ctx.chain_store().shallow_clone();
    tokio::task::spawn_blocking(move || {
        let start = Instant::now();
        let head = cs.heaviest_tipset();
        match cs.chain_index().repair_tipset_lookup_window(
            &head,
            head.epoch(),
            cs.ec_calculator_finalized_epoch(),
        ) {
            Ok(n_repaired) => tracing::info!(
                "Successfully verified tipset lookup table, {n_repaired} entries repaired, took {}.",
                humantime::format_duration(start.elapsed()),
            ),
            Err(e) => warn!("failed to verify tipset lookup table: {e:#?}"),
        }
    });
}

/// If our current chain is below a supported height, we need a snapshot to bring it up
/// to a supported height. If we've not been given a snapshot by the user, get one.
///
/// An [`Err`] should be considered fatal.
async fn maybe_set_snapshot_path(
    config: &mut Config,
    chain_config: &ChainConfig,
    epoch: ChainEpoch,
    auto_download_snapshot: bool,
    download_directory: &Path,
) -> anyhow::Result<()> {
    if !download_directory.is_dir() {
        anyhow::bail!(
            "`download_directory` does not exist: {}",
            download_directory.display()
        );
    }

    let vendor = snapshot::TrustedVendor::default();
    let chain = config.chain();

    // What height is our chain at right now, and what network version does that correspond to?
    let network_version = chain_config.network_version(epoch);
    let network_version_is_small = network_version < NetworkVersion::V16;

    // We don't support small network versions (we can't validate from e.g genesis).
    // So we need a snapshot (which will be from a recent network version)
    let require_a_snapshot = network_version_is_small;
    let have_a_snapshot = config.client.snapshot_path.is_some();

    match (require_a_snapshot, have_a_snapshot, auto_download_snapshot) {
        (false, _, _) => {}   // noop - don't need a snapshot
        (true, true, _) => {} // noop - we need a snapshot, and we have one
        (true, false, true) => {
            const AUTO_SNAPSHOT_PATH_ENV_KEY: &str = "FOREST_AUTO_DOWNLOAD_SNAPSHOT_PATH";
            match std::env::var(AUTO_SNAPSHOT_PATH_ENV_KEY) {
                Ok(path) if !path.is_empty() => {
                    tracing::info!(
                        "importing snapshot from {path} set by `{AUTO_SNAPSHOT_PATH_ENV_KEY}`"
                    );
                    config.client.snapshot_path = Some(path.into());
                }
                _ => {
                    // Resolve the redirect URL to get the actual snapshot URL
                    // This ensures all chunks download from the same snapshot even if
                    // a new snapshot is published during the download
                    let (resolved_url, _num_bytes, filename) =
                        crate::cli_shared::snapshot::peek(vendor, chain).await?;
                    tracing::info!("Downloading snapshot: {filename}");
                    config.client.snapshot_path = Some(resolved_url.to_string().into());
                }
            }
        }
        (true, false, false) => {
            // we need a snapshot, don't have one, and don't have permission to download one, so ask the user
            let (url, num_bytes, filename) = crate::cli_shared::snapshot::peek(vendor, chain)
                .await
                .context("couldn't get snapshot size")?;
            // dialoguer will double-print long lines, so manually print the first clause ourselves,
            // then let `Confirm` handle the second.
            println!(
                "Forest requires a snapshot to sync with the network, but automatic fetching is disabled."
            );
            let message = format!(
                "Fetch a {} snapshot? (denying will exit the program). ",
                indicatif::HumanBytes(num_bytes)
            );
            let have_permission = asyncify(|| {
                dialoguer::Confirm::with_theme(&ColorfulTheme::default())
                    .with_prompt(message)
                    .default(false)
                    .interact()
                    // e.g not a tty (or some other error), so haven't got permission.
                    .unwrap_or(false)
            })
            .await;
            if !have_permission {
                bail!(
                    "Forest requires a snapshot to sync with the network, but automatic fetching is disabled."
                )
            }
            tracing::info!("Downloading snapshot: {filename}");
            config.client.snapshot_path = Some(url.to_string().into());
        }
    };

    Ok(())
}

/// returns the first error with which any of the services end, or never returns at all
// This should return anyhow::Result<!> once the `Never` type is stabilized
async fn propagate_error(
    services: &mut JoinSet<anyhow::Result<()>>,
) -> anyhow::Result<std::convert::Infallible> {
    while let Some(result) = services.join_next().await {
        if let Ok(Err(error_message)) = result {
            return Err(error_message);
        }
    }
    std::future::pending().await
}

/// Run the closure on a thread where blocking is allowed
///
/// # Panics
/// If the closure panics
fn asyncify<T>(f: impl FnOnce() -> T + Send + 'static) -> impl Future<Output = T>
where
    T: Send + 'static,
{
    tokio::task::spawn_blocking(f).then(|res| async { res.expect("spawned task panicked") })
}

#[cfg(test)]
mod tests {
    use rstest::rstest;

    use super::*;

    #[rstest]
    #[case::current_non_positive(0, 1, anyhow::Result::Err(anyhow::anyhow!(
        "current head epoch 0 is invalid"
    )))]
    #[case::current_non_positive(-1, 1, anyhow::Result::Err(anyhow::anyhow!(
        "current head epoch 0 is invalid"
    )))]
    #[case::from_positive_beyond_head(10, 11, anyhow::Result::Err(anyhow::anyhow!(
        "requested validation start epoch 11 is beyond the current head at epoch 10"
    )))]
    #[case::from_positive_within_range(10, 5, anyhow::Result::Ok(5..=10))]
    #[case::from_zero(10, 0, anyhow::Result::Ok(0..=10))]
    #[case::from_negative_within_range(10, -5, anyhow::Result::Ok(5..=10))]
    #[case::from_negative_beyond_range(10, -15, anyhow::Result::Ok(0..=10))]
    fn test_validation_range(
        #[case] current: ChainEpoch,
        #[case] from: ChainEpoch,
        #[case] expected: anyhow::Result<std::ops::RangeInclusive<ChainEpoch>>,
    ) {
        let result = validation_range(current, from);
        match expected {
            Ok(expected_range) => {
                assert_eq!(result.unwrap(), expected_range);
            }
            Err(_) => {
                assert!(result.is_err());
            }
        }
    }
}