forest/rpc/methods/

chain.rs

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

pub mod types;
use types::*;

#[cfg(test)]
use crate::blocks::RawBlockHeader;
use crate::blocks::{Block, CachingBlockHeader, Tipset, TipsetKey};
use crate::chain::index::{ChainIndex, ResolveNullTipset};
use crate::chain::{ChainStore, ExportOptions, ExportResult, FilecoinSnapshotVersion, HeadChange};
use crate::chain_sync::{get_full_tipset, load_full_tipset};
use crate::cid_collections::{CidHashSet, FileBackedCidHashSet};
use crate::db::car::forest::{
    ASYNC_OPS_TIMEOUT, forest_car_sha256sum_path, forest_car_with_filename_suffix,
    tmp_exporting_forest_car_path,
};
use crate::ipld::DfsIter;
use crate::ipld::{CHAIN_EXPORT_STATUS, ChainExportGuard, ChainExportKind};
use crate::lotus_json::{HasLotusJson, LotusJson, lotus_json_with_self};
#[cfg(test)]
use crate::lotus_json::{assert_all_snapshots, assert_unchanged_via_json};
use crate::message::{ChainMessage, SignedMessage};
use crate::networks::ChainConfig;
use crate::prelude::*;
use crate::rpc::f3::F3ExportLatestSnapshot;
use crate::rpc::types::*;
use crate::rpc::{ApiPaths, Ctx, EthEventHandler, Permission, RpcMethod, ServerError};
use crate::shim::clock::ChainEpoch;
use crate::shim::error::ExitCode;
use crate::shim::executor::Receipt;
use crate::shim::message::Message;
use crate::utils::db::CborStoreExt as _;
use crate::utils::encoding::hex;
use crate::utils::io::VoidAsyncWriter;
use crate::utils::misc::env::is_env_truthy;
use crate::utils::spawn_blocking_with_timeout;
use anyhow::{Context as _, Result};
use digest::Digest as _;
use enumflags2::{BitFlags, make_bitflags};
use fvm_ipld_encoding::{CborStore, RawBytes};
use ipld_core::ipld::Ipld;
use jsonrpsee::types::Params;
use jsonrpsee::types::error::ErrorObjectOwned;
use num::BigInt;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use sha2::Sha256;
use std::convert::Infallible;
use std::fs::File;
use std::{
    collections::VecDeque,
    path::{Path, PathBuf},
    sync::LazyLock,
};
use tokio::sync::broadcast::{self, Receiver as Subscriber};

const HEAD_CHANNEL_CAPACITY: usize = 10;

/// [`SAFE_HEIGHT_DISTANCE`] is the distance from the latest tipset, i.e. "heaviest", that
/// is considered to be safe from re-orgs at an increasingly diminishing
/// probability.
///
/// This is used to determine the safe tipset when using the "safe" tag in
/// [`TipsetSelector`] or via Eth JSON-RPC APIs. Note that "safe" doesn't guarantee
/// finality, but rather a high probability of not being reverted. For guaranteed
/// finality, use the "finalized" tag.
///
/// This constant is experimental and may change in the future.
/// Discussion on this current value and a tracking item to document the
/// probabilistic impact of various values is in
/// https://github.com/filecoin-project/go-f3/issues/944
pub const SAFE_HEIGHT_DISTANCE: ChainEpoch = 200;

pub enum ChainGetFinalizedTipset {}
impl RpcMethod<0> for ChainGetFinalizedTipset {
    const NAME: &'static str = "Filecoin.ChainGetFinalizedTipSet";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::V1);
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the latest F3 finalized tipset, or falls back to EC finality if F3 is not operational on the node or if the F3 finalized tipset is further back than EC finalized tipset.";

    type Params = ();
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(ChainGetTipSetV2::get_latest_finalized_tipset(&ctx).await?)
    }
}

pub enum ChainGetMessage {}
impl RpcMethod<1> for ChainGetMessage {
    const NAME: &'static str = "Filecoin.ChainGetMessage";
    const PARAM_NAMES: [&'static str; 1] = ["messageCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the message with the specified CID.";

    type Params = (Cid,);
    type Ok = Message;

    async fn handle(
        ctx: Ctx,
        (message_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let chain_message: ChainMessage = ctx
            .db()
            .get_cbor(&message_cid)?
            .with_context(|| format!("can't find message with cid {message_cid}"))?;
        let message = match chain_message {
            ChainMessage::Signed(m) => Arc::unwrap_or_clone(m).into_message(),
            ChainMessage::Unsigned(m) => Arc::unwrap_or_clone(m),
        };

        Ok(message)
    }
}

/// Returns the events stored under the given event AMT root CID.
/// Errors if the root CID cannot be found in the blockstore.
pub enum ChainGetEvents {}
impl RpcMethod<1> for ChainGetEvents {
    const NAME: &'static str = "Filecoin.ChainGetEvents";
    const PARAM_NAMES: [&'static str; 1] = ["rootCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the events under the given event AMT root CID.";

    type Params = (Cid,);
    type Ok = Vec<Event>;
    async fn handle(
        ctx: Ctx,
        (root_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let events = EthEventHandler::get_events_by_event_root(&ctx, &root_cid)?;
        Ok(events)
    }
}

pub enum ChainGetParentMessages {}
impl RpcMethod<1> for ChainGetParentMessages {
    const NAME: &'static str = "Filecoin.ChainGetParentMessages";
    const PARAM_NAMES: [&'static str; 1] = ["blockCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns the messages included in the blocks of the parent tipset.";

    type Params = (Cid,);
    type Ok = Vec<ApiMessage>;

    async fn handle(
        ctx: Ctx,
        (block_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let store = ctx.db();
        let block_header: CachingBlockHeader = store
            .get_cbor(&block_cid)?
            .with_context(|| format!("can't find block header with cid {block_cid}"))?;
        if block_header.epoch == 0 {
            Ok(vec![])
        } else {
            let parent_tipset = ctx
                .chain_index()
                .load_required_tipset(&block_header.parents)?;
            load_api_messages_from_tipset(&ctx, parent_tipset.key()).await
        }
    }
}

pub enum ChainGetParentReceipts {}
impl RpcMethod<1> for ChainGetParentReceipts {
    const NAME: &'static str = "Filecoin.ChainGetParentReceipts";
    const PARAM_NAMES: [&'static str; 1] = ["blockCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns the message receipts included in the blocks of the parent tipset.";

    type Params = (Cid,);
    type Ok = Vec<ApiReceipt>;

    async fn handle(
        ctx: Ctx,
        (block_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let store = ctx.db();
        let block_header: CachingBlockHeader = store
            .get_cbor(&block_cid)?
            .with_context(|| format!("can't find block header with cid {block_cid}"))?;
        if block_header.epoch == 0 {
            return Ok(vec![]);
        }
        let receipts = Receipt::get_receipts(store, block_header.message_receipts)
            .map_err(|_| {
                ErrorObjectOwned::owned::<()>(
                    1,
                    format!(
                        "failed to root: ipld: could not find {}",
                        block_header.message_receipts
                    ),
                    None,
                )
            })?
            .iter()
            .map(|r| ApiReceipt {
                exit_code: r.exit_code().into(),
                return_data: r.return_data(),
                gas_used: r.gas_used(),
                events_root: r.events_root(),
            })
            .collect_vec();

        Ok(receipts)
    }
}

pub enum ChainGetMessagesInTipset {}
impl RpcMethod<1> for ChainGetMessagesInTipset {
    const NAME: &'static str = "Filecoin.ChainGetMessagesInTipset";
    const PARAM_NAMES: [&'static str; 1] = ["tipsetKey"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns all messages included in the tipset with the given key.";

    type Params = (ApiTipsetKey,);
    type Ok = Vec<ApiMessage>;

    async fn handle(
        ctx: Ctx,
        (ApiTipsetKey(tipset_key),): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let tipset = ctx
            .chain_store()
            .load_required_tipset_or_heaviest(&tipset_key)?;
        load_api_messages_from_tipset(&ctx, tipset.key()).await
    }
}

pub enum ChainPruneSnapshot {}
impl RpcMethod<1> for ChainPruneSnapshot {
    const NAME: &'static str = "Forest.SnapshotGC";
    const PARAM_NAMES: [&'static str; 1] = ["blocking"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Admin;
    const DESCRIPTION: &'static str =
        "Triggers database garbage collection, optionally blocking until it completes.";

    type Params = (bool,);
    type Ok = ();

    async fn handle(
        _ctx: Ctx,
        (blocking,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        if let Some(gc) = crate::daemon::GLOBAL_SNAPSHOT_GC.get() {
            let progress_rx = gc.trigger()?;
            if blocking {
                progress_rx.recv_async().await.map_err(|_| {
                    anyhow::anyhow!("snapshot GC ended without reporting an outcome")
                })??;
            }
            Ok(())
        } else {
            Err(anyhow::anyhow!("snapshot gc is not enabled").into())
        }
    }
}

pub enum ForestChainExport {}
impl RpcMethod<1> for ForestChainExport {
    const NAME: &'static str = "Forest.ChainExport";
    const PARAM_NAMES: [&'static str; 1] = ["params"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all_with_v2();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Exports a chain snapshot to a CAR file from the given epoch; only one export may run at a time.";

    type Params = (ForestChainExportParams,);
    type Ok = ApiExportResult;

    async fn handle(
        ctx: Ctx,
        (params,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        // Spawn a task so it's not cancelled when CLI client is disconnected.
        // So do not wrap this with `AbortOnDropHandle`
        let handle = tokio::spawn(async move {
            let chain_export_guard = ChainExportGuard::try_start_export(ChainExportKind::Snapshot)?;
            let result = export_chain_inner(&ctx, params, &chain_export_guard).await;
            chain_export_guard.finish(result)
        });
        Ok(handle.await??)
    }
}

fn save_checksum(
    checksum: digest::Output<Sha256>,
    snapshot_output_path: &Path,
) -> anyhow::Result<()> {
    let path = forest_car_sha256sum_path(snapshot_output_path);
    std::fs::write(
        path,
        format!(
            "{} {}\n",
            hex::encode(checksum),
            snapshot_output_path
                .file_name()
                .and_then(std::ffi::OsStr::to_str)
                .context("Failed to retrieve file name while saving checksum")?
        ),
    )?;
    Ok(())
}

async fn export_chain_inner(
    ctx: &Ctx,
    params: ForestChainExportParams,
    chain_export_guard: &ChainExportGuard,
) -> anyhow::Result<ApiExportResult> {
    let ForestChainExportParams {
        version,
        epoch,
        recent_roots,
        output_path,
        tipset_keys: ApiTipsetKey(tsk),
        include_receipts,
        include_events,
        include_tipset_keys,
        augmented_snapshot,
        tipset_lookup,
        skip_checksum,
        dry_run,
    } = params;

    anyhow::ensure!(
        recent_roots >= 0,
        "recentRoots must not be negative, got {recent_roots}."
    );

    let head = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?;
    let start_ts = ctx
        .chain_index()
        .load_required_tipset_by_height(epoch, head, ResolveNullTipset::TakeOlder)
        .await?;

    let options = ExportOptions {
        skip_checksum,
        include_receipts,
        include_events,
        include_tipset_keys,
        include_tipset_lookup: tipset_lookup,
        seen: FileBackedCidHashSet::new(ctx.temp_dir.as_path())?,
    };
    let tmp_path = tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?;
    let writer = if dry_run {
        tokio_util::either::Either::Left(VoidAsyncWriter)
    } else {
        tokio_util::either::Either::Right(tokio::fs::File::create(&tmp_path).await?)
    };
    let chain_export = match version {
        FilecoinSnapshotVersion::V1 => {
            crate::chain::export::<Sha256, _>(ctx.db(), &start_ts, recent_roots, writer, options)
                .boxed()
        }
        FilecoinSnapshotVersion::V2 => {
            let f3_snap_tmp_path = {
                let mut f3_snap_dir = output_path.clone();
                let mut builder = tempfile::Builder::new();
                let with_suffix = builder.suffix(".f3snap.bin");
                if f3_snap_dir.pop() {
                    with_suffix.tempfile_in(&f3_snap_dir)
                } else {
                    with_suffix.tempfile_in(".")
                }?
                .into_temp_path()
            };
            let f3_snap = {
                match F3ExportLatestSnapshot::run(f3_snap_tmp_path.display().to_string()).await {
                    Ok(cid) => Some((cid, File::open(&f3_snap_tmp_path)?)),
                    Err(e) => {
                        tracing::error!("Failed to export F3 snapshot: {e:#}");
                        None
                    }
                }
            };
            crate::chain::export_v2::<Sha256, _, _>(
                ctx.db(),
                f3_snap,
                &start_ts,
                recent_roots,
                writer,
                options,
            )
            .boxed()
        }
    };
    match chain_export_guard.run_cancellable(chain_export).await {
        Some(result) => {
            let ExportResult {
                checksum,
                tipset_lookup: hamt,
            } = result?;
            if !dry_run {
                let output_path = output_path.clone();
                spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || {
                    tmp_path.persist(&output_path)?;
                    // The snapshot at `output_path` is usable from this point on;
                    // a checksum-file failure is not worth failing the export over.
                    if let Some(checksum) = checksum
                        && let Err(e) = save_checksum(checksum, &output_path)
                    {
                        tracing::warn!(
                            "failed to save the checksum file for {}: {e:#}",
                            output_path.display()
                        );
                    }
                    Ok(())
                })
                .await
                .context("failed to persist the exported snapshot")?;
            }
            // The auxiliary snapshots must stay cancellable: the guard is still held,
            // so an unguarded await here would accept a cancel yet ignore it.
            let auxiliary_exports = async {
                match (tipset_lookup, hamt) {
                    (true, Some(hamt)) => {
                        let mut hamt = hamt.context("failed to generate tipset lookup snapshot")?;
                        let roots = nunny::vec![hamt.flush()?];
                        let hamt_output_path =
                            forest_car_with_filename_suffix(&output_path, "_tipset_lookup")?;
                        let (mut writer, hamt_output_tmp_path) = if dry_run {
                            (tokio_util::either::Either::Left(VoidAsyncWriter), None)
                        } else {
                            let tmp_path = tempfile::TempPath::try_from_path(
                                tmp_exporting_forest_car_path(&hamt_output_path),
                            )?;
                            (
                                tokio_util::either::Either::Right(
                                    tokio::fs::File::create(&tmp_path).await?,
                                ),
                                Some(tmp_path),
                            )
                        };
                        hamt.into_store()
                            .export_forest_car(roots, &mut writer)
                            .await
                            .context("failed to write tipset lookup snapshot")?;
                        if let Some(hamt_output_tmp_path) = hamt_output_tmp_path {
                            hamt_output_tmp_path.persist(&hamt_output_path)?;
                            if !skip_checksum {
                                // No need to generate checksum on the fly for small snapshots
                                save_checksum(
                                    Sha256::digest(std::fs::read(&hamt_output_path)?),
                                    &hamt_output_path,
                                )?;
                            }
                        }
                    }
                    (true, None) => {
                        anyhow::bail!("requested tipset lookup snapshot is missing")
                    }
                    (false, Some(_)) => {
                        anyhow::bail!(
                            "tipset lookup snapshot should not be generated when it's not requested"
                        )
                    }
                    _ => {}
                }
                if augmented_snapshot {
                    // It takes <10s on mainnet so export it in sequence for simplicity.
                    // Some stats:
                    // calibnet 3895470+2000: 5s  5.4MiB
                    // mainnet  6193120+2000: 5s  12MiB
                    let augmented_snapshot_output_path =
                        forest_car_with_filename_suffix(&output_path, "_receipts_events")?;
                    let (writer, augmented_snapshot_output_tmp_path) = if dry_run {
                        (tokio_util::either::Either::Left(VoidAsyncWriter), None)
                    } else {
                        let tmp_path = tempfile::TempPath::try_from_path(
                            tmp_exporting_forest_car_path(&augmented_snapshot_output_path),
                        )?;
                        (
                            tokio_util::either::Either::Right(
                                tokio::fs::File::create(&tmp_path).await?,
                            ),
                            Some(tmp_path),
                        )
                    };
                    crate::chain::export_receipts_events_to_forest_car(
                        ctx.db(),
                        &start_ts,
                        recent_roots,
                        writer,
                    )
                    .await
                    .context("failed to export message receipts and events snapshot")?;
                    if let Some(augmented_snapshot_output_tmp_path) =
                        augmented_snapshot_output_tmp_path
                    {
                        augmented_snapshot_output_tmp_path
                            .persist(&augmented_snapshot_output_path)?;
                        if !skip_checksum {
                            // No need to generate checksum on the fly for small snapshots
                            save_checksum(
                                Sha256::digest(std::fs::read(&augmented_snapshot_output_path)?),
                                &augmented_snapshot_output_path,
                            )?;
                        }
                    }
                }
                anyhow::Ok(())
            };
            match chain_export_guard.run_cancellable(auxiliary_exports).await {
                Some(result) => {
                    result?;
                    Ok(ApiExportResult::Done)
                }
                None => {
                    tracing::warn!("Auxiliary snapshot exports were cancelled");
                    Ok(ApiExportResult::Cancelled)
                }
            }
        }
        None => {
            tracing::warn!("Snapshot export was cancelled");
            Ok(ApiExportResult::Cancelled)
        }
    }
}

pub enum ForestChainExportStatus {}
impl RpcMethod<0> for ForestChainExportStatus {
    const NAME: &'static str = "Forest.ChainExportStatus";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all_with_v2();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns the progress and status of the in-progress chain export.";

    type Params = ();
    type Ok = ApiExportStatus;

    async fn handle(
        _ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let snapshot = CHAIN_EXPORT_STATUS.snapshot();
        let initial_epoch = snapshot.initial_epoch;
        let epoch = snapshot.epoch;
        let progress = if initial_epoch == 0 {
            0.0
        } else {
            let p = 1.0 - ((epoch as f64) / (initial_epoch as f64));
            if p.is_finite() {
                p.clamp(0.0, 1.0)
            } else {
                0.0
            }
        };
        // only two decimal places
        let progress = (progress * 100.0).round() / 100.0;

        Ok(ApiExportStatus {
            state: snapshot.state,
            kind: snapshot.kind,
            error: snapshot.error,
            progress,
            start_time: snapshot.start_time,
            current_epoch: epoch,
            start_epoch: initial_epoch,
        })
    }
}

pub enum ForestChainExportCancel {}
impl RpcMethod<0> for ForestChainExportCancel {
    const NAME: &'static str = "Forest.ChainExportCancel";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all_with_v2();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Cancels the in-progress chain export, returning whether one was running.";

    type Params = ();
    type Ok = bool;

    async fn handle(
        _ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(CHAIN_EXPORT_STATUS.cancel_running())
    }
}

/// Parameters for [`IndexBackfill`].
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct IndexBackfillParams {
    /// Starting epoch, inclusive. Defaults to the chain head, unless `resume` is set and a
    /// persisted resume checkpoint exists.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub from: Option<ChainEpoch>,
    /// Ending epoch, inclusive. Mutually exclusive with `n_tipsets`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub to: Option<ChainEpoch>,
    /// Number of tipsets to backfill. Mutually exclusive with `to`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub n_tipsets: Option<u64>,
    /// Recompute missing tipset state (expensive) instead of skipping it; tipsets that still can't
    /// be computed are skipped and reported rather than aborting the run.
    #[serde(default)]
    pub recompute: bool,
    /// Also index revert-prone tipsets newer than the EC-finalized epoch (up to the head). By
    /// default the walk is clamped to the EC-finalized epoch.
    #[serde(default)]
    pub allow_near_head: bool,
    /// Resume from the persisted checkpoint of a previous run instead of starting at the chain
    /// head. Ignored when `from` is given.
    #[serde(default)]
    pub resume: bool,
}
lotus_json_with_self!(IndexBackfillParams);

/// Progress and status of the in-daemon index backfill, returned by [`IndexBackfillStatus`].
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ApiIndexBackfillStatus {
    pub state: crate::ipld::ChainExportState,
    pub error: Option<String>,
    pub progress: f64,
    pub start_epoch: ChainEpoch,
    pub current_epoch: ChainEpoch,
    pub target_epoch: ChainEpoch,
    pub indexed: u64,
    pub skipped: u64,
    pub start_time: Option<chrono::DateTime<chrono::Utc>>,
}
lotus_json_with_self!(ApiIndexBackfillStatus);

impl std::fmt::Display for ApiIndexBackfillStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        use crate::ipld::ChainExportState::*;
        match self.state {
            Running => write!(
                f,
                "Backfilling: {:.1}% (walk at epoch {}, from {} down to {}; indexed {}, skipped {})",
                self.progress.clamp(0.0, 1.0) * 100.0,
                self.current_epoch,
                self.start_epoch,
                self.target_epoch,
                self.indexed,
                self.skipped,
            ),
            Idle => write!(f, "No index backfill in progress"),
            Succeeded => write!(
                f,
                "No index backfill in progress (last run succeeded: indexed {}, skipped {})",
                self.indexed, self.skipped
            ),
            Cancelled => write!(
                f,
                "No index backfill in progress (last run was cancelled: indexed {}, skipped {})",
                self.indexed, self.skipped
            ),
            Failed => write!(
                f,
                "No index backfill in progress (last run failed: {})",
                self.error.as_deref().unwrap_or("unknown error")
            ),
        }
    }
}

pub enum IndexBackfill {}
impl RpcMethod<1> for IndexBackfill {
    const NAME: &'static str = "Forest.IndexBackfill";
    const PARAM_NAMES: [&'static str; 1] = ["params"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Admin;
    const DESCRIPTION: &'static str = "Starts a chain index backfill (Ethereum mappings, events, block blooms) over an epoch range using the running node, returning immediately. Poll `Forest.IndexBackfillStatus` for progress. Only one backfill may run at a time, and it never overlaps snapshot export or the snapshot GC.";

    type Params = (IndexBackfillParams,);
    type Ok = ();

    async fn handle(
        ctx: Ctx,
        (params,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        // Validate synchronously so bad requests surface immediately to the caller.
        let spec = index_backfill_range_spec(&params)?;

        // Best-effort early rejection while snapshot GC runs; mutual exclusion is actually enforced
        // by the shared chain-export slot and `BackfillGuard::try_start` below.
        if crate::daemon::GLOBAL_SNAPSHOT_GC
            .get()
            .is_some_and(|gc| gc.is_running())
        {
            return Err(anyhow::anyhow!(
                "snapshot GC is currently running; retry the backfill once it completes"
            )
            .into());
        }

        // Acquire the single-flight guard now so "already running" is reported immediately.
        let guard = crate::daemon::db_util::BackfillGuard::try_start()?;

        // Run detached so the backfill is not cancelled when the CLI client disconnects.
        // So do not wrap this with `AbortOnDropHandle`.
        tokio::spawn(async move {
            let result = run_index_backfill_inner(&ctx, params, spec, &guard).await;
            let _ = guard.finish(result);
        });
        Ok(())
    }
}

fn index_backfill_range_spec(
    params: &IndexBackfillParams,
) -> Result<crate::daemon::db_util::RangeSpec, ServerError> {
    let n_tipsets = params.n_tipsets.map(|n| n as usize);
    Ok(crate::daemon::db_util::RangeSpec::new(
        params.to, n_tipsets,
    )?)
}

async fn run_index_backfill_inner(
    ctx: &Ctx,
    params: IndexBackfillParams,
    spec: crate::daemon::db_util::RangeSpec,
    guard: &crate::daemon::db_util::BackfillGuard,
) -> anyhow::Result<()> {
    use crate::daemon::db_util::{BackfillOptions, read_backfill_checkpoint, run_backfill};

    let head_ts = ctx.chain_store().heaviest_tipset();
    let checkpoint = if params.resume {
        read_backfill_checkpoint(&ctx.state_manager)?
    } else {
        None
    };
    let from_ts = if let Some(from) = params.from {
        let from = from.min(head_ts.epoch());
        ctx.chain_index()
            .load_required_tipset_by_height(from, head_ts, ResolveNullTipset::TakeOlder)
            .await?
    } else if let Some(checkpoint) = checkpoint {
        let checkpoint = checkpoint.min(head_ts.epoch());
        tracing::info!("Resuming index backfill from checkpoint epoch {checkpoint}");
        ctx.chain_index()
            .load_required_tipset_by_height(checkpoint, head_ts, ResolveNullTipset::TakeOlder)
            .await?
    } else {
        head_ts
    };

    let options = BackfillOptions {
        allow_recompute: params.recompute,
        allow_near_head: params.allow_near_head,
        ..BackfillOptions::default()
    };

    run_backfill(&ctx.state_manager, &from_ts, spec, options, guard).await?;
    Ok(())
}

pub enum IndexBackfillStatus {}
impl RpcMethod<0> for IndexBackfillStatus {
    const NAME: &'static str = "Forest.IndexBackfillStatus";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns the progress and status of the in-progress (or last) index backfill.";

    type Params = ();
    type Ok = ApiIndexBackfillStatus;

    async fn handle(
        _ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let snapshot = crate::daemon::db_util::BACKFILL_STATUS.snapshot();
        // Progress is the fraction of the epoch span walked (the backfill counts downward).
        let span = snapshot.start_epoch - snapshot.target_epoch;
        let progress = if span > 0 {
            let walked = snapshot.start_epoch - snapshot.current_epoch;
            ((walked as f64) / (span as f64)).clamp(0.0, 1.0)
        } else {
            0.0
        };
        let progress = (progress * 100.0).round() / 100.0;
        Ok(ApiIndexBackfillStatus {
            state: snapshot.state,
            error: snapshot.error,
            progress,
            start_epoch: snapshot.start_epoch,
            current_epoch: snapshot.current_epoch,
            target_epoch: snapshot.target_epoch,
            indexed: snapshot.indexed,
            skipped: snapshot.skipped,
            start_time: snapshot.start_time,
        })
    }
}

pub enum IndexBackfillCancel {}
impl RpcMethod<0> for IndexBackfillCancel {
    const NAME: &'static str = "Forest.IndexBackfillCancel";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Admin;
    const DESCRIPTION: &'static str =
        "Cancels the in-progress index backfill, returning whether one was running.";

    type Params = ();
    type Ok = bool;

    async fn handle(
        _ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(crate::daemon::db_util::BACKFILL_STATUS.cancel_running())
    }
}

pub enum ForestChainExportDiff {}
impl RpcMethod<1> for ForestChainExportDiff {
    const NAME: &'static str = "Forest.ChainExportDiff";
    const PARAM_NAMES: [&'static str; 1] = ["params"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all_with_v2();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Exports a differential snapshot covering the given epoch range to a CAR file.";

    type Params = (ForestChainExportDiffParams,);
    type Ok = ApiExportResult;

    async fn handle(
        ctx: Ctx,
        (params,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        // Spawn a task so it's not cancelled when CLI client is disconnected
        // So do not wrap this with `AbortOnDropHandle`
        let handle = tokio::spawn(async move {
            let chain_export_guard =
                ChainExportGuard::try_start_export(ChainExportKind::DiffSnapshot)?;
            let result = export_diff_inner(&ctx, params, &chain_export_guard).await;
            chain_export_guard.finish(result)
        });
        Ok(handle.await??)
    }
}

async fn export_diff_inner(
    ctx: &Ctx,
    params: ForestChainExportDiffParams,
    chain_export_guard: &ChainExportGuard,
) -> anyhow::Result<ApiExportResult> {
    let ForestChainExportDiffParams {
        from,
        to,
        depth,
        output_path,
    } = params;

    let chain_finality = ctx.chain_config().policy.chain_finality;
    anyhow::ensure!(
        depth >= chain_finality,
        "depth {depth} must be greater than or equal to chain_finality {chain_finality}"
    );

    let head = ctx.chain_store().heaviest_tipset();
    let start_ts = ctx
        .chain_index()
        .load_required_tipset_by_height(from, head, ResolveNullTipset::TakeOlder)
        .await?;
    let tmp_path = tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?;
    let chain_export = crate::tool::subcommands::archive_cmd::do_export(
        ctx.chain_index().db(),
        start_ts,
        Some(ctx.chain_store().genesis_tipset()),
        tmp_path.to_path_buf(),
        None,
        depth,
        Some(to),
        Some(chain_finality),
        true,
    );

    match chain_export_guard.run_cancellable(chain_export).await {
        Some(result) => {
            result?;
            spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || {
                Ok(tmp_path.persist(&output_path)?)
            })
            .await
            .context("failed to persist the exported snapshot")?;
            Ok(ApiExportResult::Done)
        }
        None => {
            tracing::warn!("Diff snapshot export was cancelled");
            Ok(ApiExportResult::Cancelled)
        }
    }
}

pub enum ChainExport {}
impl RpcMethod<1> for ChainExport {
    const NAME: &'static str = "Filecoin.ChainExport";
    const PARAM_NAMES: [&'static str; 1] = ["params"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Exports a v1 chain snapshot to a CAR file from the given epoch.";

    type Params = (ChainExportParams,);
    type Ok = ApiExportResult;

    async fn handle(
        ctx: Ctx,
        (ChainExportParams {
            epoch,
            recent_roots,
            output_path,
            tipset_keys,
            skip_checksum,
            dry_run,
        },): Self::Params,
        ext: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        ForestChainExport::handle(
            ctx,
            (ForestChainExportParams {
                version: FilecoinSnapshotVersion::V1,
                epoch,
                recent_roots,
                output_path,
                tipset_keys,
                include_receipts: false,
                include_events: false,
                include_tipset_keys: false,
                augmented_snapshot: false,
                tipset_lookup: false,
                skip_checksum,
                dry_run,
            },),
            ext,
        )
        .await
    }
}

pub enum ChainReadObj {}
impl RpcMethod<1> for ChainReadObj {
    const NAME: &'static str = "Filecoin.ChainReadObj";
    const PARAM_NAMES: [&'static str; 1] = ["cid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Reads IPLD nodes referenced by the specified CID from the chain blockstore and returns raw bytes.";

    type Params = (Cid,);
    type Ok = Vec<u8>;

    async fn handle(
        ctx: Ctx,
        (cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let bytes = ctx
            .db()
            .get(&cid)?
            .with_context(|| format!("can't find object with cid={cid}"))?;
        Ok(bytes)
    }
}

pub enum ChainHasObj {}
impl RpcMethod<1> for ChainHasObj {
    const NAME: &'static str = "Filecoin.ChainHasObj";
    const PARAM_NAMES: [&'static str; 1] = ["cid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Checks if a given CID exists in the chain blockstore.";

    type Params = (Cid,);
    type Ok = bool;

    async fn handle(
        ctx: Ctx,
        (cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(ctx.db().get(&cid)?.is_some())
    }
}

/// Returns statistics about the graph referenced by 'obj'.
/// If 'base' is also specified, then the returned stat will be a diff between the two objects.
pub enum ChainStatObj {}
impl RpcMethod<2> for ChainStatObj {
    const NAME: &'static str = "Filecoin.ChainStatObj";
    const PARAM_NAMES: [&'static str; 2] = ["objCid", "baseCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the size and link count of the IPLD graph under the given CID, or the difference relative to an optional base CID.";

    type Params = (Cid, Option<Cid>);
    type Ok = ObjStat;

    async fn handle(
        ctx: Ctx,
        (obj_cid, base_cid): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let mut stats = ObjStat::default();
        let mut seen = CidHashSet::default();
        let mut walk = |cid, collect| {
            let mut queue = VecDeque::new();
            queue.push_back(cid);
            while let Some(link_cid) = queue.pop_front() {
                if !seen.insert(link_cid) {
                    continue;
                }
                let data = ctx.db().get(&link_cid)?;
                if let Some(data) = data {
                    if collect {
                        stats.links += 1;
                        stats.size += data.len();
                    }
                    if matches!(link_cid.codec(), fvm_ipld_encoding::DAG_CBOR)
                        && let Ok(ipld) =
                            crate::utils::encoding::from_slice_with_fallback::<Ipld>(&data)
                    {
                        for ipld in DfsIter::new(ipld) {
                            if let Ipld::Link(cid) = ipld {
                                queue.push_back(cid);
                            }
                        }
                    }
                }
            }
            anyhow::Ok(())
        };
        if let Some(base_cid) = base_cid {
            walk(base_cid, false)?;
        }
        walk(obj_cid, true)?;
        Ok(stats)
    }
}

pub enum ChainGetBlockMessages {}
impl RpcMethod<1> for ChainGetBlockMessages {
    const NAME: &'static str = "Filecoin.ChainGetBlockMessages";
    const PARAM_NAMES: [&'static str; 1] = ["blockCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns all messages from the specified block.";

    type Params = (Cid,);
    type Ok = BlockMessages;

    async fn handle(
        ctx: Ctx,
        (block_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let blk: CachingBlockHeader = ctx.db().get_cbor_required(&block_cid)?;
        let (unsigned_cids, signed_cids) = crate::chain::read_msg_cids(ctx.db(), &blk)?;
        let (bls_msg, secp_msg) =
            crate::chain::block_messages_from_cids(ctx.db(), &unsigned_cids, &signed_cids)?;
        let cids = unsigned_cids.into_iter().chain(signed_cids).collect();

        let ret = BlockMessages {
            bls_msg,
            secp_msg,
            cids,
        };
        Ok(ret)
    }
}

pub enum ChainGetPath {}
impl RpcMethod<2> for ChainGetPath {
    const NAME: &'static str = "Filecoin.ChainGetPath";
    const PARAM_NAMES: [&'static str; 2] = ["from", "to"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the path between the two specified tipsets.";

    type Params = (TipsetKey, TipsetKey);
    type Ok = Vec<PathChange>;

    async fn handle(
        ctx: Ctx,
        (from, to): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(chain_get_path(ctx.chain_store(), &from, &to)?.into_change_vec())
    }
}

/// Find the path between two tipsets, as [`PathChanges`].
///
/// ```text
/// 0 - A - B - C - D
///     ^~~~~~~~> apply B, C
///
/// 0 - A - B - C - D
///     <~~~~~~~^ revert C, B
///
///     <~~~~~~~~ revert C, B
/// 0 - A - B  - C
///     |
///      -- B' - C'
///      ~~~~~~~~> then apply B', C'
/// ```
///
/// Exposes errors from the [`Blockstore`], and returns an error if there is no common ancestor.
pub fn chain_get_path(
    chain_store: &ChainStore,
    from: &TipsetKey,
    to: &TipsetKey,
) -> anyhow::Result<PathChanges> {
    let finality = chain_store.chain_config().policy.chain_finality;
    let mut to_revert = chain_store
        .load_required_tipset_or_heaviest(from)
        .context("couldn't load `from`")?;
    let mut to_apply = chain_store
        .load_required_tipset_or_heaviest(to)
        .context("couldn't load `to`")?;

    anyhow::ensure!(
        (to_apply.epoch() - to_revert.epoch()).abs() <= finality,
        "the gap between the new head ({}) and the old head ({}) is larger than chain finality ({finality})",
        to_apply.epoch(),
        to_revert.epoch()
    );

    let mut reverts = vec![];
    let mut applies = vec![];

    // This loop is guaranteed to terminate if the blockstore contain no cycles.
    // This is currently computationally infeasible.
    while to_revert != to_apply {
        if to_revert.epoch() > to_apply.epoch() {
            let next = chain_store
                .load_required_tipset_or_heaviest(to_revert.parents())
                .context("couldn't load ancestor of `from`")?;
            reverts.push(to_revert);
            to_revert = next;
        } else {
            let next = chain_store
                .load_required_tipset_or_heaviest(to_apply.parents())
                .context("couldn't load ancestor of `to`")?;
            applies.push(to_apply);
            to_apply = next;
        }
    }
    applies.reverse();
    Ok(PathChanges { reverts, applies })
}

/// Get tipset at epoch. Pick younger tipset if epoch points to a
/// null-tipset. Only tipsets below the given `head` are searched. If `head`
/// is null, the node will use the heaviest tipset.
pub enum ChainGetTipSetByHeight {}
impl RpcMethod<2> for ChainGetTipSetByHeight {
    const NAME: &'static str = "Filecoin.ChainGetTipSetByHeight";
    const PARAM_NAMES: [&'static str; 2] = ["height", "tipsetKey"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the tipset at the specified height.";

    type Params = (ChainEpoch, ApiTipsetKey);
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (height, ApiTipsetKey(tipset_key)): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let ts = ctx
            .chain_store()
            .load_required_tipset_or_heaviest(&tipset_key)?;
        let tss = ctx
            .chain_index()
            .load_required_tipset_by_height(height, ts, ResolveNullTipset::TakeOlder)
            .await?;
        Ok(tss)
    }
}

pub enum ChainGetTipSetAfterHeight {}
impl RpcMethod<2> for ChainGetTipSetAfterHeight {
    const NAME: &'static str = "Filecoin.ChainGetTipSetAfterHeight";
    const PARAM_NAMES: [&'static str; 2] = ["height", "tipsetKey"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Looks back and returns the tipset at the specified epoch.
    If there are no blocks at the given epoch,
    returns the first non-nil tipset at a later epoch.";

    type Params = (ChainEpoch, ApiTipsetKey);
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (height, ApiTipsetKey(tipset_key)): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let ts = ctx
            .chain_store()
            .load_required_tipset_or_heaviest(&tipset_key)?;
        let tss = ctx
            .chain_index()
            .load_required_tipset_by_height(height, ts, ResolveNullTipset::TakeNewer)
            .await?;
        Ok(tss)
    }
}

pub enum ChainGetGenesis {}
impl RpcMethod<0> for ChainGetGenesis {
    const NAME: &'static str = "Filecoin.ChainGetGenesis";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the genesis tipset of the chain.";

    type Params = ();
    type Ok = Option<Tipset>;

    async fn handle(
        ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let genesis = ctx.chain_store().genesis_block_header();
        Ok(Some(Tipset::from(genesis)))
    }
}

pub enum ChainHead {}
impl RpcMethod<0> for ChainHead {
    const NAME: &'static str = "Filecoin.ChainHead";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the chain head (heaviest tipset).";

    type Params = ();
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let heaviest = ctx.chain_store().heaviest_tipset();
        Ok(heaviest)
    }
}

pub enum ChainGetBlock {}
impl RpcMethod<1> for ChainGetBlock {
    const NAME: &'static str = "Filecoin.ChainGetBlock";
    const PARAM_NAMES: [&'static str; 1] = ["blockCid"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the block with the specified CID.";

    type Params = (Cid,);
    type Ok = CachingBlockHeader;

    async fn handle(
        ctx: Ctx,
        (block_cid,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let blk: CachingBlockHeader = ctx.db().get_cbor_required(&block_cid)?;
        Ok(blk)
    }
}

pub enum ChainGetTipSet {}

impl RpcMethod<1> for ChainGetTipSet {
    const NAME: &'static str = "Filecoin.ChainGetTipSet";
    const PARAM_NAMES: [&'static str; 1] = ["tipsetKey"];
    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::{ V0 | V1 });
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the tipset with the specified CID.";

    type Params = (ApiTipsetKey,);
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (ApiTipsetKey(tsk),): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        if let Some(tsk) = &tsk {
            let ts = ctx.chain_index().load_required_tipset(tsk)?;
            Ok(ts)
        } else {
            // It contains Lotus error message `NewTipSet called with zero length array of blocks` for parity tests
            Err(anyhow::anyhow!(
                "TipsetKey cannot be empty (NewTipSet called with zero length array of blocks)"
            )
            .into())
        }
    }
}

pub enum ChainGetTipSetV2 {}

impl ChainGetTipSetV2 {
    pub async fn get_tipset_by_anchor(
        ctx: &Ctx,
        anchor: Option<&TipsetAnchor>,
    ) -> anyhow::Result<Tipset> {
        if let Some(anchor) = anchor {
            match (&anchor.key.0, &anchor.tag) {
                // Anchor is zero-valued. Fall back to heaviest tipset.
                (None, None) => Ok(ctx.state_manager.heaviest_tipset()),
                // Get tipset at the specified key.
                (Some(tsk), None) => Ok(ctx.chain_index().load_required_tipset(tsk)?),
                (None, Some(tag)) => Self::get_tipset_by_tag(ctx, *tag).await,
                _ => {
                    anyhow::bail!("invalid anchor")
                }
            }
        } else {
            // No anchor specified. Fall back to finalized tipset.
            Self::get_tipset_by_tag(ctx, TipsetTag::Finalized).await
        }
    }

    pub async fn get_tipset_by_tag(ctx: &Ctx, tag: TipsetTag) -> anyhow::Result<Tipset> {
        match tag {
            TipsetTag::Latest => Ok(ctx.state_manager.heaviest_tipset()),
            TipsetTag::Finalized => Self::get_latest_finalized_tipset(ctx).await,
            TipsetTag::Safe => Self::get_latest_safe_tipset(ctx).await,
        }
    }

    pub async fn get_latest_safe_tipset(ctx: &Ctx) -> anyhow::Result<Tipset> {
        let finalized = Self::get_latest_finalized_tipset(ctx).await?;
        let head = ctx.chain_store().heaviest_tipset();
        let safe_height = (head.epoch() - SAFE_HEIGHT_DISTANCE).max(0);
        if finalized.epoch() >= safe_height {
            Ok(finalized)
        } else {
            Ok(ctx
                .chain_index()
                .load_required_tipset_by_height(safe_height, head, ResolveNullTipset::TakeOlder)
                .await?)
        }
    }

    pub async fn get_latest_finalized_tipset(ctx: &Ctx) -> anyhow::Result<Tipset> {
        ChainGetTipSetFinalityStatus::get_finality_status(ctx)
            .await?
            .finalized_tip_set
            .context("failed to resolve finalized tipset")
    }

    pub async fn get_tipset(ctx: &Ctx, selector: &TipsetSelector) -> anyhow::Result<Tipset> {
        selector.validate()?;
        // Get tipset by key.
        if let ApiTipsetKey(Some(tsk)) = &selector.key {
            let ts = ctx.chain_index().load_required_tipset(tsk)?;
            return Ok(ts);
        }
        // Get tipset by height.
        if let Some(height) = &selector.height {
            let anchor = Self::get_tipset_by_anchor(ctx, height.anchor.as_ref()).await?;
            let ts = ctx
                .chain_index()
                .load_required_tipset_by_height(
                    height.at,
                    anchor,
                    height.resolve_null_tipset_policy(),
                )
                .await?;
            return Ok(ts);
        }
        // Get tipset by tag, either latest or finalized.
        if let Some(tag) = &selector.tag {
            let ts = Self::get_tipset_by_tag(ctx, *tag).await?;
            return Ok(ts);
        }
        anyhow::bail!("no tipset found for selector")
    }
}

impl RpcMethod<1> for ChainGetTipSetV2 {
    const NAME: &'static str = "Filecoin.ChainGetTipSet";
    const PARAM_NAMES: [&'static str; 1] = ["tipsetSelector"];
    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::{ V2 });
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the tipset with the specified CID.";

    type Params = (TipsetSelector,);
    type Ok = Tipset;

    async fn handle(
        ctx: Ctx,
        (selector,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(Self::get_tipset(&ctx, &selector).await?)
    }
}

pub enum ChainGetTipSetFinalityStatus {}

const EC_CALCULATOR_FINALITY_CACHE_SIZE: usize = 4;
impl ChainGetTipSetFinalityStatus {
    pub async fn get_finality_status(ctx: &Ctx) -> anyhow::Result<ChainFinalityStatus> {
        let head = ctx.chain_store().heaviest_tipset();
        let (ec_finality_threshold_depth, ec_finalized_tip_set) =
            Self::get_ec_finality_threshold_depth_and_tipset_with_cache(ctx, head.shallow_clone())
                .await?;
        let f3_finalized_tip_set = ctx.chain_store().f3_finalized_tipset();
        let finalized_tip_set = match (&ec_finalized_tip_set, &f3_finalized_tip_set) {
            (Some(ec), Some(f3)) => {
                if ec.epoch() >= f3.epoch() {
                    Some(ec.shallow_clone())
                } else {
                    Some(f3.shallow_clone())
                }
            }
            (Some(ec), None) => Some(ec.shallow_clone()),
            (None, Some(f3)) => Some(f3.shallow_clone()),
            (None, None) => None,
        };
        Ok(ChainFinalityStatus {
            ec_finality_threshold_depth,
            ec_finalized_tip_set,
            f3_finalized_tip_set,
            finalized_tip_set,
            head,
        })
    }

    pub async fn get_ec_finality_threshold_depth_and_tipset_with_cache(
        ctx: &Ctx,
        head: Tipset,
    ) -> anyhow::Result<(i64, Option<Tipset>)> {
        static CACHE: LazyLock<quick_cache::sync::Cache<TipsetKey, (i64, Option<Tipset>)>> =
            LazyLock::new(|| quick_cache::sync::Cache::new(EC_CALCULATOR_FINALITY_CACHE_SIZE));
        CACHE
            .get_or_insert_async(
                head.shallow_clone().key(),
                Self::get_ec_finality_threshold_depth_and_tipset(ctx, head),
            )
            .await
    }

    pub fn get_ec_finality_epoch(
        chain_index: &ChainIndex,
        chain_config: &ChainConfig,
        head: &Tipset,
    ) -> i64 {
        let depth =
            Self::get_ec_finality_threshold_depth_with_cache(chain_index, chain_config, head);
        Self::get_ec_finality_epoch_by_depth(chain_config, head, depth)
    }

    fn get_ec_finality_epoch_by_depth(
        chain_config: &ChainConfig,
        head: &Tipset,
        depth: i64,
    ) -> i64 {
        let depth = if depth >= 0 {
            depth
        } else {
            chain_config.policy.chain_finality
        };
        (head.epoch() - depth).max(0)
    }

    fn get_ec_finality_threshold_depth_with_cache(
        chain_index: &ChainIndex,
        chain_config: &ChainConfig,
        head: &Tipset,
    ) -> i64 {
        static CACHE: LazyLock<quick_cache::sync::Cache<TipsetKey, i64>> =
            LazyLock::new(|| quick_cache::sync::Cache::new(EC_CALCULATOR_FINALITY_CACHE_SIZE));
        CACHE
            .get_or_insert_with(head.key(), move || -> Result<i64, Infallible> {
                Ok(Self::get_ec_finality_threshold_depth(
                    chain_index,
                    chain_config,
                    head,
                ))
            })
            .expect("infallible")
    }

    fn get_ec_finality_threshold_depth(
        chain_index: &ChainIndex,
        chain_config: &ChainConfig,
        head: &Tipset,
    ) -> i64 {
        use crate::chain::ec_finality::calculator::{
            DEFAULT_BLOCKS_PER_EPOCH, DEFAULT_BYZANTINE_FRACTION, DEFAULT_GUARANTEE,
            find_threshold_depth,
        };

        /// Number of extra epochs to fetch beyond [`chain_finality`] when
        /// building the chain sample for [`find_threshold_depth`].
        ///
        /// The extra 5 epochs act as a tail buffer to prevent out-of-bounds access,
        /// particularly when null rounds (epochs with zero blocks) are present, since
        /// they consume array slots without advancing the meaningful epoch count.
        const FINALITY_CHAIN_EXTRA_EPOCHS: usize = 5;

        let finality = chain_config.policy.chain_finality;
        let chain_len = finality as usize + FINALITY_CHAIN_EXTRA_EPOCHS;
        let mut chain = Vec::with_capacity(chain_len);
        let mut ts = head.shallow_clone();
        while chain.len() < chain_len {
            chain.push(ts.len() as i64);
            if let Ok(parent) = chain_index.load_required_tipset(ts.parents()) {
                // insert 0 for null rounds
                let pad = usize::try_from(ts.epoch() - parent.epoch() - 1).unwrap_or_default();
                chain.resize(chain.len().saturating_add(pad).min(chain_len), 0);
                ts = parent;
            } else {
                break;
            }
        }
        // Reverse to chronological order (oldest first).
        chain.reverse();
        match find_threshold_depth(
            &chain,
            finality,
            DEFAULT_BLOCKS_PER_EPOCH,
            DEFAULT_BYZANTINE_FRACTION,
            *DEFAULT_GUARANTEE,
        ) {
            Ok(threshold) => threshold,
            Err(e) => {
                tracing::error!(
                    "Failed to calculate EC finality threshold depth: {e:#}, chain: {chain:?}"
                );
                -1
            }
        }
    }

    async fn get_ec_finality_threshold_depth_and_tipset(
        ctx: &Ctx,
        head: Tipset,
    ) -> anyhow::Result<(i64, Option<Tipset>)> {
        let depth = Self::get_ec_finality_threshold_depth_with_cache(
            ctx.chain_index(),
            ctx.chain_config(),
            &head,
        );
        let ec_finality_epoch =
            Self::get_ec_finality_epoch_by_depth(ctx.chain_config(), &head, depth);
        let finalized = ctx
            .chain_index()
            .tipset_by_height(ec_finality_epoch, head, ResolveNullTipset::TakeOlder)
            .await?;
        Ok((depth, finalized))
    }
}

impl RpcMethod<0> for ChainGetTipSetFinalityStatus {
    const NAME: &'static str = "Filecoin.ChainGetTipSetFinalityStatus";
    const PARAM_NAMES: [&'static str; 0] = [];
    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::{ V2 });
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns a breakdown of how the node is currently determining finality.";

    type Params = ();
    type Ok = ChainFinalityStatus;

    async fn handle(
        ctx: Ctx,
        (): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(Self::get_finality_status(&ctx).await?)
    }
}

pub enum ChainSetHead {}
impl RpcMethod<1> for ChainSetHead {
    const NAME: &'static str = "Filecoin.ChainSetHead";
    const PARAM_NAMES: [&'static str; 1] = ["tsk"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Admin;
    const DESCRIPTION: &'static str =
        "Forcibly sets the chain head to the tipset with the given key.";

    type Params = (TipsetKey,);
    type Ok = ();

    async fn handle(
        ctx: Ctx,
        (tsk,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        // This is basically a port of the reference implementation at
        // https://github.com/filecoin-project/lotus/blob/v1.23.0/node/impl/full/chain.go#L321

        let new_head = ctx.chain_index().load_required_tipset(&tsk)?;
        let mut current = ctx.chain_store().heaviest_tipset();
        while current.epoch() >= new_head.epoch() {
            for cid in current.key().to_cids() {
                ctx.chain_store().unmark_block_as_validated(&cid);
            }
            let parents = &current.block_headers().first().parents;
            current = ctx.chain_index().load_required_tipset(parents)?;
        }
        ctx.chain_store()
            .set_heaviest_tipset(new_head)
            .map_err(Into::into)
    }
}

pub enum ChainGetMinBaseFee {}
impl RpcMethod<1> for ChainGetMinBaseFee {
    const NAME: &'static str = "Forest.ChainGetMinBaseFee";
    const PARAM_NAMES: [&'static str; 1] = ["lookback"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str =
        "Returns the minimum base fee across the given number of lookback tipsets, in attoFIL.";

    type Params = (u32,);
    type Ok = String;

    async fn handle(
        ctx: Ctx,
        (lookback,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let mut current = ctx.chain_store().heaviest_tipset();
        let mut min_base_fee = current.block_headers().first().parent_base_fee.clone();

        for _ in 0..lookback {
            let parents = &current.block_headers().first().parents;
            current = ctx.chain_index().load_required_tipset(parents)?;

            min_base_fee =
                min_base_fee.min(current.block_headers().first().parent_base_fee.to_owned());
        }

        Ok(min_base_fee.atto().to_string())
    }
}

pub enum ChainTipSetWeight {}
impl RpcMethod<1> for ChainTipSetWeight {
    const NAME: &'static str = "Filecoin.ChainTipSetWeight";
    const PARAM_NAMES: [&'static str; 1] = ["tipsetKey"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the weight of the specified tipset.";

    type Params = (ApiTipsetKey,);
    type Ok = BigInt;

    async fn handle(
        ctx: Ctx,
        (ApiTipsetKey(tipset_key),): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        let ts = ctx
            .chain_store()
            .load_required_tipset_or_heaviest(&tipset_key)?;
        let weight = crate::fil_cns::weight(ctx.db(), &ts)?;
        Ok(weight)
    }
}

pub enum ChainGetTipsetByParentState {}
impl RpcMethod<1> for ChainGetTipsetByParentState {
    const NAME: &'static str = "Forest.ChainGetTipsetByParentState";
    const PARAM_NAMES: [&'static str; 1] = ["parentState"];
    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
    const PERMISSION: Permission = Permission::Read;
    const DESCRIPTION: &'static str = "Returns the tipset whose parent state root matches the given CID, or null if none is found.";

    type Params = (Cid,);
    type Ok = Option<Tipset>;

    async fn handle(
        ctx: Ctx,
        (parent_state,): Self::Params,
        _: &http::Extensions,
    ) -> Result<Self::Ok, ServerError> {
        Ok(ctx
            .chain_store()
            .heaviest_tipset()
            .chain(ctx.db())
            .find(|ts| ts.parent_state() == &parent_state)
            .shallow_clone())
    }
}

pub const CHAIN_NOTIFY: &str = "Filecoin.ChainNotify";
pub(crate) fn chain_notify(
    _params: Params<'_>,
    data: &crate::rpc::RPCState,
) -> Subscriber<Vec<ApiHeadChange>> {
    chain_notify_inner(data.chain_store())
}

fn chain_notify_inner(chain_store: &ChainStore) -> Subscriber<Vec<ApiHeadChange>> {
    let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY);

    // Subscribe before sampling the head, else a change landing in between is lost.
    let head_changes_rx = chain_store.subscribe_head_changes();
    let current = chain_store.heaviest_tipset();
    sender
        .send(vec![ApiHeadChange {
            change: HeadChangeType::Current,
            tipset: current,
        }])
        .expect("receiver is not dropped");

    tokio::spawn(async move {
        while let Ok(changes) = head_changes_rx.recv_async().await {
            let api_changes = changes.into_change_vec().into_iter().map_into().collect();
            if sender.send(api_changes).is_err() {
                tracing::info!("chain notify subscribers are all closed");
                break;
            }
        }
        tracing::info!("head changes channel closed");
    });
    receiver
}

async fn load_api_messages_from_tipset(
    ctx: &crate::rpc::RPCState,
    tipset_keys: &TipsetKey,
) -> Result<Vec<ApiMessage>, ServerError> {
    static SHOULD_BACKFILL: LazyLock<bool> = LazyLock::new(|| {
        let enabled = is_env_truthy("FOREST_RPC_BACKFILL_FULL_TIPSET_FROM_NETWORK");
        if enabled {
            tracing::warn!(
                "Full tipset backfilling from network is enabled via FOREST_RPC_BACKFILL_FULL_TIPSET_FROM_NETWORK, excessive disk and bandwidth usage is expected."
            );
        }
        enabled
    });
    let full_tipset = if *SHOULD_BACKFILL {
        get_full_tipset(
            &ctx.sync_network_context,
            ctx.chain_store(),
            None,
            tipset_keys,
        )
        .await?
    } else {
        load_full_tipset(ctx.chain_store(), tipset_keys)?
    };
    let blocks = full_tipset.into_blocks();
    let mut messages = vec![];
    let mut seen = CidHashSet::default();
    for Block {
        bls_messages,
        secp_messages,
        ..
    } in blocks
    {
        for message in bls_messages {
            let cid = message.cid();
            if seen.insert(cid) {
                messages.push(ApiMessage { cid, message });
            }
        }

        for msg in secp_messages {
            let cid = msg.cid();
            if seen.insert(cid) {
                messages.push(ApiMessage {
                    cid,
                    message: msg.message,
                });
            }
        }
    }

    Ok(messages)
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct BlockMessages {
    #[serde(rename = "BlsMessages", with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Vec<Message>>")]
    pub bls_msg: Vec<Message>,
    #[serde(rename = "SecpkMessages", with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Vec<SignedMessage>>")]
    pub secp_msg: Vec<SignedMessage>,
    #[serde(rename = "Cids", with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Vec<Cid>>")]
    pub cids: Vec<Cid>,
}
lotus_json_with_self!(BlockMessages);

#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, JsonSchema)]
#[serde(rename_all = "PascalCase")]
pub struct ApiReceipt {
    // Exit status of message execution
    pub exit_code: ExitCode,
    // `Return` value if the exit code is zero
    #[serde(rename = "Return", with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<RawBytes>")]
    pub return_data: RawBytes,
    // Non-negative value of GasUsed
    pub gas_used: u64,
    #[serde(with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Option<Cid>>")]
    pub events_root: Option<Cid>,
}

lotus_json_with_self!(ApiReceipt);

#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug, Eq, PartialEq)]
#[serde(rename_all = "PascalCase")]
pub struct ApiMessage {
    #[serde(with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Cid>")]
    pub cid: Cid,
    #[serde(with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Message>")]
    pub message: Message,
}

lotus_json_with_self!(ApiMessage);

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ForestChainExportParams {
    pub version: FilecoinSnapshotVersion,
    pub epoch: ChainEpoch,
    pub recent_roots: i64,
    pub output_path: PathBuf,
    #[schemars(with = "LotusJson<ApiTipsetKey>")]
    #[serde(with = "crate::lotus_json", default)]
    pub tipset_keys: ApiTipsetKey,
    /// Include message receipts in the output snapshot
    #[serde(default)]
    pub include_receipts: bool,
    /// Include events in the output snapshot
    #[serde(default)]
    pub include_events: bool,
    #[serde(default)]
    pub include_tipset_keys: bool,
    /// Generate a separate snapshot that contains augmented data (message receipts and events)
    #[serde(default)]
    pub augmented_snapshot: bool,
    /// Generate a separate snapshot that contains tipset lookup
    #[serde(default)]
    pub tipset_lookup: bool,
    pub skip_checksum: bool,
    pub dry_run: bool,
}
lotus_json_with_self!(ForestChainExportParams);

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ForestChainExportDiffParams {
    pub from: ChainEpoch,
    pub to: ChainEpoch,
    pub depth: i64,
    pub output_path: PathBuf,
}
lotus_json_with_self!(ForestChainExportDiffParams);

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ChainExportParams {
    pub epoch: ChainEpoch,
    pub recent_roots: i64,
    pub output_path: PathBuf,
    #[schemars(with = "LotusJson<ApiTipsetKey>")]
    #[serde(with = "crate::lotus_json")]
    pub tipset_keys: ApiTipsetKey,
    pub skip_checksum: bool,
    pub dry_run: bool,
}
lotus_json_with_self!(ChainExportParams);

/// The kind of head change delivered by `Filecoin.ChainNotify`.
#[derive(PartialEq, Eq, Debug, Clone, Copy, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "lowercase")]
pub enum HeadChangeType {
    Current,
    Apply,
    Revert,
}

#[derive(PartialEq, Debug, Serialize, Deserialize, Clone, JsonSchema)]
#[serde(rename_all = "PascalCase")]
pub struct ApiHeadChange {
    #[serde(rename = "Type")]
    pub change: HeadChangeType,
    #[serde(rename = "Val", with = "crate::lotus_json")]
    #[schemars(with = "LotusJson<Tipset>")]
    pub tipset: Tipset,
}
lotus_json_with_self!(ApiHeadChange);

impl From<HeadChange> for ApiHeadChange {
    fn from(change: HeadChange) -> Self {
        match change {
            HeadChange::Apply(tipset) => Self {
                change: HeadChangeType::Apply,
                tipset,
            },
            HeadChange::Revert(tipset) => Self {
                change: HeadChangeType::Revert,
                tipset,
            },
        }
    }
}

#[derive(PartialEq, Debug, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "Type", content = "Val", rename_all = "snake_case")]
pub enum PathChange<T = Tipset> {
    Revert(T),
    Apply(T),
}

impl<T: Clone> Clone for PathChange<T> {
    fn clone(&self) -> Self {
        match self {
            Self::Revert(i) => Self::Revert(i.clone()),
            Self::Apply(i) => Self::Apply(i.clone()),
        }
    }
}

impl<T> PathChange<T> {
    pub fn tipset(&self) -> &T {
        match self {
            Self::Revert(ts) | Self::Apply(ts) => ts,
        }
    }
}

impl HasLotusJson for PathChange {
    type LotusJson = PathChange<<Tipset as HasLotusJson>::LotusJson>;

    #[cfg(test)]
    fn snapshots() -> Vec<(serde_json::Value, Self)> {
        use crate::test_utils::dummy_ticket;
        use serde_json::json;
        let header = CachingBlockHeader::new(RawBlockHeader {
            ticket: dummy_ticket(0),
            ..Default::default()
        });
        let header_cid = *header.cid();
        vec![(
            json!({
                "Type": "revert",
                "Val": {
                    "Blocks": [
                        {
                            "BeaconEntries": null,
                            "ForkSignaling": 0,
                            "Height": 0,
                            "Messages": { "/": "baeaaaaa" },
                            "Miner": "f00",
                            "ParentBaseFee": "0",
                            "ParentMessageReceipts": { "/": "baeaaaaa" },
                            "ParentStateRoot": { "/":"baeaaaaa" },
                            "ParentWeight": "0",
                            "Parents": [{"/":"bafyreiaqpwbbyjo4a42saasj36kkrpv4tsherf2e7bvezkert2a7dhonoi"}],
                            "Ticket": { "VRFProof": "AA==" },
                            "Timestamp": 0,
                            "WinPoStProof": null
                        }
                    ],
                    "Cids": [
                        { "/": header_cid.to_string() }
                    ],
                    "Height": 0
                }
            }),
            Self::Revert(Tipset::from(header)),
        )]
    }

    fn into_lotus_json(self) -> Self::LotusJson {
        match self {
            PathChange::Revert(it) => PathChange::Revert(it.into_lotus_json()),
            PathChange::Apply(it) => PathChange::Apply(it.into_lotus_json()),
        }
    }

    fn from_lotus_json(lotus_json: Self::LotusJson) -> Self {
        match lotus_json {
            PathChange::Revert(it) => PathChange::Revert(Tipset::from_lotus_json(it)),
            PathChange::Apply(it) => PathChange::Apply(Tipset::from_lotus_json(it)),
        }
    }
}

#[derive(Debug)]
pub struct PathChanges<T = Tipset> {
    pub reverts: Vec<T>,
    pub applies: Vec<T>,
}

impl<T: Clone> Clone for PathChanges<T> {
    fn clone(&self) -> Self {
        let Self { reverts, applies } = self;
        Self {
            reverts: reverts.clone(),
            applies: applies.clone(),
        }
    }
}

impl<T> PathChanges<T> {
    pub fn is_empty(&self) -> bool {
        self.reverts.is_empty() && self.applies.is_empty()
    }

    pub fn into_change_vec(self) -> Vec<PathChange<T>> {
        let Self { reverts, applies } = self;
        reverts
            .into_iter()
            .map(PathChange::Revert)
            .chain(applies.into_iter().map(PathChange::Apply))
            .collect_vec()
    }
}

#[cfg(test)]
impl<T> quickcheck::Arbitrary for PathChange<T>
where
    T: quickcheck::Arbitrary + ShallowClone,
{
    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
        let inner = T::arbitrary(g);
        g.choose(&[PathChange::Apply(inner.clone()), PathChange::Revert(inner)])
            .unwrap()
            .clone()
    }
}

#[test]
fn snapshots() {
    assert_all_snapshots::<PathChange>()
}

#[cfg(test)]
#[quickcheck_macros::quickcheck]
fn quickcheck(val: PathChange) {
    assert_unchanged_via_json(val)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::daemon::db_util::RangeSpec;
    use crate::{
        blocks::{Chain4U, HeaderBuilder, RawBlockHeader, chain4u},
        db::{
            MemoryDB,
            car::{AnyCar, ManyCar},
        },
        networks::{self, ChainConfig},
    };
    use PathChange::{Apply, Revert};
    use rstest::rstest;
    use std::sync::Arc;
    use std::time::Duration;

    #[rstest]
    #[case(Some(0), None, Some(RangeSpec::To(0)))]
    #[case(Some(-1), None, None)]
    #[case(None, Some(10), Some(RangeSpec::NumTipsets(10)))]
    #[case(None, None, None)]
    #[case(Some(10), Some(10), None)]
    fn index_backfill_range_spec_validates_params(
        #[case] to: Option<ChainEpoch>,
        #[case] n_tipsets: Option<u64>,
        #[case] expected: Option<RangeSpec>,
    ) {
        let params = IndexBackfillParams {
            to,
            n_tipsets,
            ..Default::default()
        };
        assert_eq!(index_backfill_range_spec(&params).ok(), expected);
    }

    #[test]
    fn revert_to_ancestor_linear() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c, d] -> [e]
        };

        // simple
        assert_path_change(&cs, b, a, [Revert(&[b])]);

        // from multi-member tipset
        assert_path_change(&cs, [c, d], a, [Revert(&[c, d][..]), Revert(&[b])]);

        // to multi-member tipset
        assert_path_change(&cs, e, [c, d], [Revert(e)]);

        // over multi-member tipset
        assert_path_change(&cs, e, b, [Revert(&[e][..]), Revert(&[c, d])]);
    }

    /// Mirror how lotus handles passing an incomplete `TipsetKey`s.
    /// Tested on lotus `1.23.2`
    #[test]
    fn incomplete_tipsets() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a, b] -> [c] -> [d, _e] // this pattern 2 -> 1 -> 2 can be found at calibnet epoch 1369126
        };

        // apply to descendant with incomplete `from`
        assert_path_change(
            &cs,
            a,
            c,
            [
                Revert(&[a][..]), // revert the incomplete tipset
                Apply(&[a, b]),   // apply the complete one
                Apply(&[c]),      // apply the destination
            ],
        );

        // apply to descendant with incomplete `to`
        assert_path_change(&cs, c, d, [Apply(d)]);

        // revert to ancestor with incomplete `from`
        assert_path_change(&cs, d, c, [Revert(d)]);

        // revert to ancestor with incomplete `to`
        assert_path_change(
            &cs,
            c,
            a,
            [
                Revert(&[c][..]),
                Revert(&[a, b]), // revert the complete tipset
                Apply(&[a]),     // apply the incomplete one
            ],
        );
    }

    #[test]
    fn apply_to_descendant_linear() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c, d] -> [e]
        };

        // simple
        assert_path_change(&cs, a, b, [Apply(&[b])]);

        // from multi-member tipset
        assert_path_change(&cs, [c, d], e, [Apply(e)]);

        // to multi-member tipset
        assert_path_change(&cs, b, [c, d], [Apply([c, d])]);

        // over multi-member tipset
        assert_path_change(&cs, b, e, [Apply(&[c, d][..]), Apply(&[e])]);
    }

    #[test]
    fn cross_fork_simple() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b1] -> [c1]
        };
        chain4u! {
            from [a] in db;
            [b2] -> [c2]
        };

        // same height
        assert_path_change(&cs, b1, b2, [Revert(b1), Apply(b2)]);

        // different height
        assert_path_change(&cs, b1, c2, [Revert(b1), Apply(b2), Apply(c2)]);

        let _ = (a, c1);
    }

    #[test]
    fn head_changes_published_deduped_and_ordered() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c, d] -> [e]
        };

        let rx = cs.subscribe_head_changes();

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        // Re-setting the same head must not publish a change.
        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        cs.set_heaviest_tipset(b.make_tipset()).unwrap();
        cs.set_heaviest_tipset([c, d].make_tipset()).unwrap();
        cs.set_heaviest_tipset(e.make_tipset()).unwrap();

        let drained = rx.try_iter().collect_vec();
        let applied = drained.iter().map(|c| c.applies.clone()).collect_vec();
        assert_eq!(
            applied,
            vec![
                vec![a.make_tipset()],
                vec![b.make_tipset()],
                vec![[c, d].make_tipset()],
                vec![e.make_tipset()],
            ]
        );
        assert!(drained.iter().all(|c| c.reverts.is_empty()));
    }

    #[test]
    fn head_changes_publishes_reverts_on_reorg() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b1]
        };
        chain4u! {
            from [a] in db;
            [b2]
        };

        let rx = cs.subscribe_head_changes();
        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        cs.set_heaviest_tipset(b1.make_tipset()).unwrap();
        cs.set_heaviest_tipset(b2.make_tipset()).unwrap(); // reorg b1 -> b2

        let last = rx.try_iter().last().unwrap();
        assert_eq!(last.reverts, vec![b1.make_tipset()]);
        assert_eq!(last.applies, vec![b2.make_tipset()]);
    }

    #[tokio::test]
    async fn chain_notify_delivers_every_apply_from_subscription() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c]
        };

        let mut rx = chain_notify_inner(&cs);

        // First message is the current head.
        let first = next_batch(&mut rx).await;
        assert_eq!(first.len(), 1);
        assert_eq!(first[0].change, HeadChangeType::Current);

        for ts in [&a, &b, &c] {
            cs.set_heaviest_tipset(ts.make_tipset()).unwrap();
        }

        // Every applied tipset must be delivered, in order, with none dropped.
        let mut applied = vec![];
        for _ in 0..3 {
            applied.extend(
                next_batch(&mut rx)
                    .await
                    .into_iter()
                    .filter(|c| c.change == HeadChangeType::Apply)
                    .map(|c| c.tipset),
            );
        }
        assert_eq!(
            applied,
            vec![a.make_tipset(), b.make_tipset(), c.make_tipset()]
        );
    }

    impl ChainStore {
        fn _load(genesis_car: &'static [u8], genesis_cid: Cid) -> Self {
            let db = Arc::new(
                ManyCar::new(MemoryDB::default())
                    .with_read_only(AnyCar::new(genesis_car).unwrap())
                    .unwrap(),
            );
            let genesis_block_header: CachingBlockHeader =
                db.get_cbor(&genesis_cid).unwrap().unwrap();
            ChainStore::new(db, Arc::new(ChainConfig::calibnet()), genesis_block_header).unwrap()
        }
        pub fn calibnet() -> Self {
            Self::_load(
                networks::calibnet::DEFAULT_GENESIS,
                *networks::calibnet::GENESIS_CID,
            )
        }
    }

    /// Utility for writing ergonomic tests
    trait MakeTipset {
        fn make_tipset(self) -> Tipset;
    }

    impl MakeTipset for &RawBlockHeader {
        fn make_tipset(self) -> Tipset {
            Tipset::from(CachingBlockHeader::new(self.clone()))
        }
    }

    impl<const N: usize> MakeTipset for [&RawBlockHeader; N] {
        fn make_tipset(self) -> Tipset {
            self.as_slice().make_tipset()
        }
    }

    impl<const N: usize> MakeTipset for &[&RawBlockHeader; N] {
        fn make_tipset(self) -> Tipset {
            self.as_slice().make_tipset()
        }
    }

    impl MakeTipset for &[&RawBlockHeader] {
        fn make_tipset(self) -> Tipset {
            Tipset::new(self.iter().cloned().cloned()).unwrap()
        }
    }

    #[track_caller]
    fn assert_path_change<T: MakeTipset>(
        store: &ChainStore,
        from: impl MakeTipset,
        to: impl MakeTipset,
        expected: impl IntoIterator<Item = PathChange<T>>,
    ) {
        fn print(path_change: &PathChange) {
            let it = match path_change {
                Revert(it) => {
                    print!("Revert(");
                    it
                }
                Apply(it) => {
                    print!(" Apply(");
                    it
                }
            };
            println!(
                "epoch = {}, key.cid = {})",
                it.epoch(),
                it.key().cid().unwrap()
            )
        }

        let actual = chain_get_path(store, from.make_tipset().key(), to.make_tipset().key())
            .unwrap()
            .into_change_vec();
        let expected = expected
            .into_iter()
            .map(|change| match change {
                PathChange::Revert(it) => PathChange::Revert(it.make_tipset()),
                PathChange::Apply(it) => PathChange::Apply(it.make_tipset()),
            })
            .collect_vec();
        if expected != actual {
            println!("SUMMARY");
            println!("=======");
            println!("expected:");
            for it in &expected {
                print(it)
            }
            println!();
            println!("actual:");
            for it in &actual {
                print(it)
            }
            println!("=======\n")
        }
        assert_eq!(
            expected, actual,
            "expected change (left) does not match actual change (right)"
        )
    }

    const BATCH_TIMEOUT: Duration = Duration::from_secs(1);

    async fn next_batch(rx: &mut Subscriber<Vec<ApiHeadChange>>) -> Vec<ApiHeadChange> {
        tokio::time::timeout(BATCH_TIMEOUT, rx.recv())
            .await
            .expect("timed out waiting for a head-change batch")
            .expect("chain notify channel closed")
    }

    /// Open a ChainNotify subscription and consume the immediate `current`
    /// event, asserting it matches the store head.
    async fn open_notify(cs: &ChainStore) -> Subscriber<Vec<ApiHeadChange>> {
        let mut rx = chain_notify_inner(cs);
        assert_eq!(
            next_batch(&mut rx).await,
            vec![ApiHeadChange {
                change: HeadChangeType::Current,
                tipset: cs.heaviest_tipset(),
            }]
        );
        rx
    }

    fn applied(ts: impl MakeTipset) -> ApiHeadChange {
        ApiHeadChange {
            change: HeadChangeType::Apply,
            tipset: ts.make_tipset(),
        }
    }

    fn reverted(ts: impl MakeTipset) -> ApiHeadChange {
        ApiHeadChange {
            change: HeadChangeType::Revert,
            tipset: ts.make_tipset(),
        }
    }

    #[tokio::test]
    async fn current_first_matches_store_head() {
        let cs = ChainStore::calibnet();

        let mut rx = chain_notify_inner(&cs);

        assert_eq!(
            next_batch(&mut rx).await,
            vec![ApiHeadChange {
                change: HeadChangeType::Current,
                tipset: cs.heaviest_tipset(),
            }]
        );
    }

    #[tokio::test]
    async fn linear_applies_chain() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c] -> [d]
        };

        let mut rx = open_notify(&cs).await;

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        cs.set_heaviest_tipset(b.make_tipset()).unwrap();
        cs.set_heaviest_tipset(c.make_tipset()).unwrap();
        cs.set_heaviest_tipset(d.make_tipset()).unwrap();

        // one single-apply batch per head move, in chain order; equality with
        // the constructed tipsets pins increasing heights and parent linkage
        assert_eq!(next_batch(&mut rx).await, vec![applied(a)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(b)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(c)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(d)]);
    }

    #[tokio::test]
    async fn fork_switch_one_batch_reverts_then_applies() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c]
        };
        chain4u! {
            from [a] in db;
            [b2] -> [c2] -> [d2]
        };

        let mut rx = open_notify(&cs).await;

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(a)]);

        // advancing over several epochs delivers one multi-apply batch
        cs.set_heaviest_tipset(c.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(b), applied(c)]);

        // switching to the longer fork delivers a single batch: reverts
        // newest-first down to the common ancestor, then applies oldest-first
        cs.set_heaviest_tipset(d2.make_tipset()).unwrap();
        assert_eq!(
            next_batch(&mut rx).await,
            vec![
                reverted(c),
                reverted(b),
                applied(b2),
                applied(c2),
                applied(d2)
            ]
        );
    }

    #[tokio::test]
    async fn rewind_pure_revert_batch() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c] -> [d]
        };

        let mut rx = open_notify(&cs).await;

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(a)]);

        cs.set_heaviest_tipset(d.make_tipset()).unwrap();
        assert_eq!(
            next_batch(&mut rx).await,
            vec![applied(b), applied(c), applied(d)]
        );

        // rewinding head to an ancestor (what the admin `ChainSetHead` does)
        // delivers one pure-revert batch, newest-first, with no applies
        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        assert_eq!(
            next_batch(&mut rx).await,
            vec![reverted(d), reverted(c), reverted(b)]
        );
    }

    #[tokio::test]
    async fn null_round_height_gaps() {
        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b]
            -> [c = HeaderBuilder::new().with_epoch(5)] // epochs 3 and 4 are null rounds
            -> [d]
        };
        assert_eq!(c.epoch, 5);
        assert_eq!(c.parents, b.make_tipset().key().clone());

        let mut rx = open_notify(&cs).await;

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        cs.set_heaviest_tipset(b.make_tipset()).unwrap();
        cs.set_heaviest_tipset(c.make_tipset()).unwrap();
        cs.set_heaviest_tipset(d.make_tipset()).unwrap();

        // consecutive applies jump from epoch 2 to epoch 5 over the null
        // rounds; no filler events are emitted for the missing epochs
        assert_eq!(next_batch(&mut rx).await, vec![applied(a)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(b)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(c)]);
        assert_eq!(next_batch(&mut rx).await, vec![applied(d)]);
    }

    #[tokio::test]
    async fn over_finality_fallback_single_apply() {
        let cs = ChainStore::calibnet();
        let finality = cs.chain_config().policy.chain_finality;
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c]
        };
        chain4u! {
            from [a] in db;
            [far = HeaderBuilder::new().with_epoch(finality + 4)]
        };

        let mut rx = open_notify(&cs).await;

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(a)]);

        cs.set_heaviest_tipset(c.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(b), applied(c)]);

        // `far` forks off `a`, so a real path from `c` would start with
        // reverts of `c` and `b`. But the head jump is wider than chain
        // finality, so the path computation bails out and the batch degrades
        // to a single apply with no reverts.
        cs.set_heaviest_tipset(far.make_tipset()).unwrap();
        assert_eq!(next_batch(&mut rx).await, vec![applied(far)]);
    }

    /// End-to-end weld: a real jsonrpsee server serving just the pubsub
    /// module (no auth stack), a raw WebSocket client, and a real
    /// `ChainStore`. A fork switch must arrive as one correctly framed
    /// `xrpc.ch.val` notification carrying the revert+apply batch.
    #[tokio::test]
    async fn ws_weld_end_to_end() {
        use crate::rpc::channel::{NOTIF_METHOD_NAME, RpcModule as FilRpcModule};
        use futures::{SinkExt, StreamExt};
        use serde_json::Value;
        use tokio_tungstenite::tungstenite::Message;

        /// Assert `xrpc.ch.val` framing (positional `[channelId, payload]`
        /// params) and decode the payload.
        fn decode_val_frame(frame: Value) -> (u64, Vec<ApiHeadChange>) {
            assert_eq!(
                frame.get("method").and_then(Value::as_str),
                Some(NOTIF_METHOD_NAME),
                "not an xrpc.ch.val frame: {frame}"
            );
            let params = frame
                .get("params")
                .and_then(Value::as_array)
                .unwrap_or_else(|| panic!("params must be a positional array: {frame}"));
            let [channel_id, payload] = params.as_slice() else {
                panic!("params must be [channelId, payload]: {frame}");
            };
            let channel_id = channel_id.as_u64().expect("channel id must be a u64");
            let batch = serde_json::from_value(payload.clone())
                .expect("payload must decode as Vec<ApiHeadChange>");
            (channel_id, batch)
        }

        let cs = ChainStore::calibnet();
        let db = Chain4U::with_blockstore(cs.db_owned());
        chain4u! {
            in db;
            [_genesis = cs.genesis_block_header()]
            -> [a] -> [b] -> [c]
        };
        chain4u! {
            from [a] in db;
            [b2] -> [c2] -> [d2]
        };

        // serve only the pubsub module (no auth stack)
        let mut pubsub = FilRpcModule::default();
        pubsub
            .register_channel("Filecoin.ChainNotify", {
                let chain_store = cs.shallow_clone();
                move |_params| chain_notify_inner(&chain_store)
            })
            .unwrap();
        let server = jsonrpsee::server::Server::builder()
            .build("127.0.0.1:0")
            .await
            .unwrap();
        let addr = server.local_addr().unwrap();
        let _server_handle = server.start(pubsub);

        let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://{addr}"))
            .await
            .unwrap();

        async fn next_ws_json(
            ws: &mut (
                     impl StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
                     + Unpin
                 ),
        ) -> Value {
            loop {
                let message = tokio::time::timeout(BATCH_TIMEOUT, ws.next())
                    .await
                    .expect("timed out waiting for a websocket frame")
                    .expect("websocket closed")
                    .unwrap();
                if message.is_text() {
                    return serde_json::from_str(message.into_text().unwrap().as_str()).unwrap();
                }
            }
        }

        // subscribe: the response carries a bare u64 channel id
        ws.send(Message::text(
            r#"{"jsonrpc":"2.0","id":1,"method":"Filecoin.ChainNotify","params":[]}"#,
        ))
        .await
        .unwrap();
        let response = next_ws_json(&mut ws).await;
        let channel_id = response
            .get("result")
            .and_then(Value::as_u64)
            .unwrap_or_else(|| panic!("channel id must be a bare u64: {response}"));

        // the first frame is the `current` event for the store head
        let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await);
        assert_eq!(frame_channel, channel_id);
        assert_eq!(
            batch,
            vec![ApiHeadChange {
                change: HeadChangeType::Current,
                tipset: cs.heaviest_tipset(),
            }]
        );

        cs.set_heaviest_tipset(a.make_tipset()).unwrap();
        let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await);
        assert_eq!(frame_channel, channel_id);
        assert_eq!(batch, vec![applied(a)]);

        cs.set_heaviest_tipset(c.make_tipset()).unwrap();
        let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await);
        assert_eq!(frame_channel, channel_id);
        assert_eq!(batch, vec![applied(b), applied(c)]);

        // fork switch: one frame carrying reverts-then-applies
        cs.set_heaviest_tipset(d2.make_tipset()).unwrap();
        let (frame_channel, batch) = decode_val_frame(next_ws_json(&mut ws).await);
        assert_eq!(frame_channel, channel_id);
        assert_eq!(
            batch,
            vec![
                reverted(c),
                reverted(b),
                applied(b2),
                applied(c2),
                applied(d2)
            ]
        );
    }
}