diff --git a/CHANGELOG.md b/CHANGELOG.md index 7eff8e97de6..91670cc7cb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,8 @@ ### Added +- [#7408](https://github.com/ChainSafe/forest/pull/7408): Impl `Forest.IndexBackfill`, `Forest.IndexBackfillStatus` and `Forest.IndexBackfillCancel` RPC methods and the `forest-cli index backfill` command back-fill the chain index (Ethereum mappings, events, block blooms) through the running daemon, so the node no longer needs to be stopped. + ### Changed ### Removed diff --git a/docs/docs/users/reference/cli.md b/docs/docs/users/reference/cli.md index e17e2102aab..f552175ad45 100644 --- a/docs/docs/users/reference/cli.md +++ b/docs/docs/users/reference/cli.md @@ -797,6 +797,78 @@ Options: -h, --help Print help ``` +### `forest-cli index` + +``` +Manage the chain index + +Usage: forest-cli index + +Commands: + backfill Backfill the chain index (Ethereum mappings, events, block blooms) using the running node + backfill-status Show the status of the current (or last) index backfill + backfill-cancel Cancel the in-progress index backfill + help Print this message or the help of the given subcommand(s) + +Options: + -h, --help Print help +``` + +### `forest-cli index backfill` + +``` +Backfill the chain index (Ethereum mappings, events, block blooms) using the running node. + +Unlike `forest-tool index backfill`, this does not require the node to be stopped: the running daemon performs the backfill through its own database handle. + +Usage: forest-cli index backfill [OPTIONS] + +Options: + --from + Starting tipset epoch for back-filling (inclusive). Defaults to the persisted resume checkpoint if present, otherwise the chain head + + --to + Ending tipset epoch for back-filling (inclusive) + + --n-tipsets + Number of tipsets to back-fill + + --recompute + Recompute missing tipset state (expensive) instead of skipping it. Without this, tipsets whose state has been garbage-collected are skipped and reported + + --allow-near-head + Also index revert-prone tipsets within `CHAIN_FINALITY` of the head. By default the walk is clamped below finality + + --no-wait + Trigger the backfill and return immediately without waiting for completion + + -h, --help + Print help (see a summary with '-h') +``` + +### `forest-cli index backfill-status` + +``` +Show the status of the current (or last) index backfill + +Usage: forest-cli index backfill-status [OPTIONS] + +Options: + --wait Wait until the backfill completes, showing progress + -h, --help Print help +``` + +### `forest-cli index backfill-cancel` + +``` +Cancel the in-progress index backfill + +Usage: forest-cli index backfill-cancel + +Options: + -h, --help Print help +``` + ### `forest-cli send` ``` diff --git a/docs/docs/users/reference/cli.sh b/docs/docs/users/reference/cli.sh index 077383d7110..6b8ad8b7b21 100755 --- a/docs/docs/users/reference/cli.sh +++ b/docs/docs/users/reference/cli.sh @@ -73,6 +73,11 @@ generate_markdown_section "forest-cli" "config" generate_markdown_section "forest-cli" "snapshot" generate_markdown_section "forest-cli" "snapshot export" +generate_markdown_section "forest-cli" "index" +generate_markdown_section "forest-cli" "index backfill" +generate_markdown_section "forest-cli" "index backfill-status" +generate_markdown_section "forest-cli" "index backfill-cancel" + generate_markdown_section "forest-cli" "send" generate_markdown_section "forest-cli" "info" generate_markdown_section "forest-cli" "shutdown" diff --git a/docs/openrpc-specs/v0.json b/docs/openrpc-specs/v0.json index 55a2765963b..c0377d234ee 100644 --- a/docs/openrpc-specs/v0.json +++ b/docs/openrpc-specs/v0.json @@ -6024,6 +6024,53 @@ }, "paramStructure": "by-position" }, + { + "name": "Forest.IndexBackfill", + "description": "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.", + "params": [ + { + "name": "params", + "required": true, + "schema": { + "$ref": "#/components/schemas/IndexBackfillParams" + } + } + ], + "result": { + "name": "Forest.IndexBackfill.Result", + "required": true, + "schema": { + "type": "null" + } + }, + "paramStructure": "by-position" + }, + { + "name": "Forest.IndexBackfillCancel", + "description": "Cancels the in-progress index backfill, returning whether one was running.", + "params": [], + "result": { + "name": "Forest.IndexBackfillCancel.Result", + "required": true, + "schema": { + "type": "boolean" + } + }, + "paramStructure": "by-position" + }, + { + "name": "Forest.IndexBackfillStatus", + "description": "Returns the progress and status of the in-progress (or last) index backfill.", + "params": [], + "result": { + "name": "Forest.IndexBackfillStatus.Result", + "required": true, + "schema": { + "$ref": "#/components/schemas/ApiIndexBackfillStatus" + } + }, + "paramStructure": "by-position" + }, { "name": "Forest.NetChainExchange", "description": "Internal API for debugging chain exchange.", @@ -7878,6 +7925,63 @@ "ValidityTerm" ] }, + "ApiIndexBackfillStatus": { + "description": "Progress and status of the in-daemon index backfill, returned by [`IndexBackfillStatus`].", + "type": "object", + "properties": { + "currentEpoch": { + "type": "integer", + "format": "int64" + }, + "error": { + "type": [ + "string", + "null" + ] + }, + "indexed": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "progress": { + "type": "number", + "format": "double" + }, + "skipped": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "startEpoch": { + "type": "integer", + "format": "int64" + }, + "startTime": { + "type": [ + "string", + "null" + ], + "format": "date-time" + }, + "state": { + "$ref": "#/components/schemas/ChainExportState" + }, + "targetEpoch": { + "type": "integer", + "format": "int64" + } + }, + "required": [ + "state", + "progress", + "startEpoch", + "currentEpoch", + "targetEpoch", + "indexed", + "skipped" + ] + }, "ApiInvocResult": { "type": "object", "properties": { @@ -8475,6 +8579,11 @@ "description": "A lite snapshot export performed by the automatic snapshot GC.", "type": "string", "const": "SnapshotGc" + }, + { + "description": "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap.", + "type": "string", + "const": "IndexBackfill" } ] }, @@ -10542,6 +10651,47 @@ "RebroadcastBackoffMax" ] }, + "IndexBackfillParams": { + "description": "Parameters for [`IndexBackfill`].", + "type": "object", + "properties": { + "allowNearHead": { + "description": "Allow indexing revert-prone tipsets within `CHAIN_FINALITY` of the head.", + "type": "boolean", + "default": false + }, + "from": { + "description": "Starting epoch, inclusive. Defaults to the persisted resume checkpoint if present,\notherwise the chain head.", + "type": [ + "integer", + "null" + ], + "format": "int64" + }, + "nTipsets": { + "description": "Number of tipsets to backfill. Mutually exclusive with `to`.", + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + }, + "recompute": { + "description": "Recompute missing tipset state (expensive) instead of skipping it.", + "type": "boolean", + "default": false + }, + "to": { + "description": "Ending epoch, inclusive. Mutually exclusive with `n_tipsets`.", + "type": [ + "integer", + "null" + ], + "format": "int64" + } + } + }, "KeyInfo": { "type": "object", "properties": { diff --git a/docs/openrpc-specs/v1.json b/docs/openrpc-specs/v1.json index 5959330e00a..5082ed0f763 100644 --- a/docs/openrpc-specs/v1.json +++ b/docs/openrpc-specs/v1.json @@ -6044,6 +6044,53 @@ }, "paramStructure": "by-position" }, + { + "name": "Forest.IndexBackfill", + "description": "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.", + "params": [ + { + "name": "params", + "required": true, + "schema": { + "$ref": "#/components/schemas/IndexBackfillParams" + } + } + ], + "result": { + "name": "Forest.IndexBackfill.Result", + "required": true, + "schema": { + "type": "null" + } + }, + "paramStructure": "by-position" + }, + { + "name": "Forest.IndexBackfillCancel", + "description": "Cancels the in-progress index backfill, returning whether one was running.", + "params": [], + "result": { + "name": "Forest.IndexBackfillCancel.Result", + "required": true, + "schema": { + "type": "boolean" + } + }, + "paramStructure": "by-position" + }, + { + "name": "Forest.IndexBackfillStatus", + "description": "Returns the progress and status of the in-progress (or last) index backfill.", + "params": [], + "result": { + "name": "Forest.IndexBackfillStatus.Result", + "required": true, + "schema": { + "$ref": "#/components/schemas/ApiIndexBackfillStatus" + } + }, + "paramStructure": "by-position" + }, { "name": "Forest.NetChainExchange", "description": "Internal API for debugging chain exchange.", @@ -8017,6 +8064,63 @@ "ValidityTerm" ] }, + "ApiIndexBackfillStatus": { + "description": "Progress and status of the in-daemon index backfill, returned by [`IndexBackfillStatus`].", + "type": "object", + "properties": { + "currentEpoch": { + "type": "integer", + "format": "int64" + }, + "error": { + "type": [ + "string", + "null" + ] + }, + "indexed": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "progress": { + "type": "number", + "format": "double" + }, + "skipped": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "startEpoch": { + "type": "integer", + "format": "int64" + }, + "startTime": { + "type": [ + "string", + "null" + ], + "format": "date-time" + }, + "state": { + "$ref": "#/components/schemas/ChainExportState" + }, + "targetEpoch": { + "type": "integer", + "format": "int64" + } + }, + "required": [ + "state", + "progress", + "startEpoch", + "currentEpoch", + "targetEpoch", + "indexed", + "skipped" + ] + }, "ApiInvocResult": { "type": "object", "properties": { @@ -8614,6 +8718,11 @@ "description": "A lite snapshot export performed by the automatic snapshot GC.", "type": "string", "const": "SnapshotGc" + }, + { + "description": "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap.", + "type": "string", + "const": "IndexBackfill" } ] }, @@ -10919,6 +11028,47 @@ "RebroadcastBackoffMax" ] }, + "IndexBackfillParams": { + "description": "Parameters for [`IndexBackfill`].", + "type": "object", + "properties": { + "allowNearHead": { + "description": "Allow indexing revert-prone tipsets within `CHAIN_FINALITY` of the head.", + "type": "boolean", + "default": false + }, + "from": { + "description": "Starting epoch, inclusive. Defaults to the persisted resume checkpoint if present,\notherwise the chain head.", + "type": [ + "integer", + "null" + ], + "format": "int64" + }, + "nTipsets": { + "description": "Number of tipsets to backfill. Mutually exclusive with `to`.", + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + }, + "recompute": { + "description": "Recompute missing tipset state (expensive) instead of skipping it.", + "type": "boolean", + "default": false + }, + "to": { + "description": "Ending epoch, inclusive. Mutually exclusive with `n_tipsets`.", + "type": [ + "integer", + "null" + ], + "format": "int64" + } + } + }, "KeyInfo": { "type": "object", "properties": { diff --git a/docs/openrpc-specs/v2.json b/docs/openrpc-specs/v2.json index f3108107d91..e49c23dfee8 100644 --- a/docs/openrpc-specs/v2.json +++ b/docs/openrpc-specs/v2.json @@ -3158,6 +3158,11 @@ "description": "A lite snapshot export performed by the automatic snapshot GC.", "type": "string", "const": "SnapshotGc" + }, + { + "description": "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap.", + "type": "string", + "const": "IndexBackfill" } ] }, diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 71cb6249ab2..b68ac96d37a 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -305,6 +305,19 @@ impl ChainStore { Ok(()) } + /// Like [`Self::put_mapping`], but only overwrites an existing entry when the incoming + /// `timestamp` is strictly newer than the stored one. Used by index backfill (which runs + /// concurrently with the live head indexer) so that walking historical tipsets cannot + /// clobber a mapping written for a newer tipset with an older one. + pub fn put_mapping_if_newer(&self, k: EthHash, v: Cid, timestamp: u64) -> Result<(), Error> { + if let Some((_, existing_timestamp)) = self.db().read_obj::<(Cid, u64)>(&k)? + && existing_timestamp >= timestamp + { + return Ok(()); + } + self.put_mapping(k, v, timestamp) + } + /// Reads the `Cid` from the blockstore for `EthAPI` queries. pub fn get_mapping(&self, hash: &EthHash) -> Result, Error> { Ok(self.db().read_obj::<(Cid, u64)>(hash)?.map(|(cid, _)| cid)) @@ -572,7 +585,16 @@ impl ChainStore { } /// Filter [`SignedMessage`]'s to keep only the most recent ones, then write corresponding entries to the Ethereum mapping. - pub fn process_signed_messages(&self, messages: &[(SignedMessage, u64)]) -> anyhow::Result<()> { + /// + /// When `compare_timestamps` is `true`, existing entries are only overwritten by strictly + /// newer ones (see [`Self::put_mapping_if_newer`]). The live head indexer passes `false` to + /// keep its blind-write fast path, while index backfill passes `true` so that historical + /// writes do not clobber newer mappings written concurrently by the head indexer. + pub fn process_signed_messages( + &self, + messages: &[(SignedMessage, u64)], + compare_timestamps: bool, + ) -> anyhow::Result<()> { let eth_txs: Vec<(EthHash, Cid, u64, usize)> = messages .iter() .enumerate() @@ -597,7 +619,11 @@ impl ChainStore { // write back for (k, v, timestamp) in filtered.into_iter() { trace!("Insert mapping {} => {}", k, v); - self.put_mapping(k, v, timestamp)?; + if compare_timestamps { + self.put_mapping_if_newer(k, v, timestamp)?; + } else { + self.put_mapping(k, v, timestamp)?; + } } trace!("Wrote {} entries in Ethereum mapping", num_entries); Ok(()) @@ -860,6 +886,42 @@ mod tests { assert!(cs.is_block_validated(&cid)); } + #[test] + fn put_mapping_if_newer_keeps_newest() { + let db = DbImpl::from(Arc::new(crate::db::MemoryDB::default())); + let chain_config = Arc::new(ChainConfig::default()); + let gen_block = CachingBlockHeader::new(RawBlockHeader { + miner_address: Address::new_id(0), + ..Default::default() + }); + let cs = ChainStore::new(db, chain_config, gen_block).unwrap(); + + let hash = EthHash::default(); + let older = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[1])); + let newer = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[2])); + + // Seed with the newer (higher timestamp) entry, as the live head indexer would. + cs.put_mapping(hash, newer, 100).unwrap(); + + // A backfill writing an older (lower timestamp) entry must not clobber it. + cs.put_mapping_if_newer(hash, older, 50).unwrap(); + assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newer)); + + // An equal timestamp must also not overwrite. + cs.put_mapping_if_newer(hash, older, 100).unwrap(); + assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newer)); + + // A strictly newer entry wins. + let newest = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&[3])); + cs.put_mapping_if_newer(hash, newest, 200).unwrap(); + assert_eq!(cs.get_mapping(&hash).unwrap(), Some(newest)); + + // Writing into an empty slot always succeeds. + let fresh_hash = EthHash(ethereum_types::H256::repeat_byte(0xab)); + cs.put_mapping_if_newer(fresh_hash, older, 1).unwrap(); + assert_eq!(cs.get_mapping(&fresh_hash).unwrap(), Some(older)); + } + #[test] fn test_messages_in_tipset_cache() { let cache = MessagesInTipsetCache::new(nonzero!(2_usize)); diff --git a/src/cli/subcommands/index_cmd.rs b/src/cli/subcommands/index_cmd.rs new file mode 100644 index 00000000000..c4b2d532e81 --- /dev/null +++ b/src/cli/subcommands/index_cmd.rs @@ -0,0 +1,145 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use crate::ipld::ChainExportState; +use crate::rpc::chain::{ + ApiIndexBackfillStatus, IndexBackfill, IndexBackfillCancel, IndexBackfillParams, + IndexBackfillStatus, +}; +use crate::rpc::{self, prelude::*}; +use crate::shim::clock::ChainEpoch; +use clap::Subcommand; +use indicatif::{ProgressBar, ProgressStyle}; +use std::time::Duration; + +#[derive(Debug, Subcommand)] +pub enum IndexCommands { + /// Backfill the chain index (Ethereum mappings, events, block blooms) using the running node. + /// + /// Unlike `forest-tool index backfill`, this does not require the node to be stopped: the + /// running daemon performs the backfill through its own database handle. + Backfill { + /// Starting tipset epoch for back-filling (inclusive). Defaults to the persisted resume + /// checkpoint if present, otherwise the chain head. + #[arg(long)] + from: Option, + /// Ending tipset epoch for back-filling (inclusive). + #[arg(long)] + to: Option, + /// Number of tipsets to back-fill. + #[arg(long, conflicts_with = "to")] + n_tipsets: Option, + /// Recompute missing tipset state (expensive) instead of skipping it. Without this, + /// tipsets whose state has been garbage-collected are skipped and reported. + #[arg(long)] + recompute: bool, + /// Also index revert-prone tipsets within `CHAIN_FINALITY` of the head. By default the + /// walk is clamped below finality. + #[arg(long)] + allow_near_head: bool, + /// Trigger the backfill and return immediately without waiting for completion. + #[arg(long)] + no_wait: bool, + }, + /// Show the status of the current (or last) index backfill. + BackfillStatus { + /// Wait until the backfill completes, showing progress. + #[arg(long)] + wait: bool, + }, + /// Cancel the in-progress index backfill. + BackfillCancel {}, +} + +impl IndexCommands { + pub async fn run(self, client: rpc::Client) -> anyhow::Result<()> { + match self { + Self::Backfill { + from, + to, + n_tipsets, + recompute, + allow_near_head, + no_wait, + } => { + let params = IndexBackfillParams { + from, + to, + n_tipsets, + recompute, + allow_near_head, + }; + client + .call(IndexBackfill::request((params,))?.with_timeout(Duration::from_secs(30))) + .await?; + println!("Index backfill started."); + if no_wait { + println!("Use `forest-cli index backfill-status` to monitor progress."); + return Ok(()); + } + wait_for_backfill(&client).await + } + Self::BackfillStatus { wait } => { + let status = client + .call(IndexBackfillStatus::request(())?.with_timeout(Duration::from_secs(30))) + .await?; + if !wait || status.state != ChainExportState::Running { + println!("{status}"); + return Ok(()); + } + wait_for_backfill(&client).await + } + Self::BackfillCancel {} => { + let cancelled = client + .call(IndexBackfillCancel::request(())?.with_timeout(Duration::from_secs(30))) + .await?; + if cancelled { + println!("Index backfill cancelled."); + } else { + println!("No index backfill in progress to cancel."); + } + Ok(()) + } + } + } +} + +/// Polls `Forest.IndexBackfillStatus` until the backfill reaches a terminal state, rendering a +/// progress bar. +async fn wait_for_backfill(client: &rpc::Client) -> anyhow::Result<()> { + let pb = ProgressBar::new(10000).with_message("Backfilling index"); + pb.set_style( + ProgressStyle::with_template("[{elapsed_precise}] [{wide_bar}] {percent}% {msg}") + .expect("indicatif template must be valid") + .progress_chars("#>-"), + ); + let last: ApiIndexBackfillStatus = loop { + let status = client + .call(IndexBackfillStatus::request(())?.with_timeout(Duration::from_secs(30))) + .await?; + let position = (status.progress.clamp(0.0, 1.0) * 10000.0).trunc() as u64; + pb.set_position(position); + if status.state != ChainExportState::Running { + break status; + } + tokio::time::sleep(Duration::from_millis(500)).await; + }; + match last.state { + ChainExportState::Succeeded => pb.finish_with_message(format!( + "Backfill completed (indexed {}, skipped {})", + last.indexed, last.skipped + )), + ChainExportState::Cancelled => pb.abandon_with_message(format!( + "Backfill cancelled (indexed {}, skipped {})", + last.indexed, last.skipped + )), + _ => { + pb.abandon_with_message("Backfill failed"); + anyhow::bail!( + "index backfill failed: {}", + last.error.as_deref().unwrap_or("unknown error") + ); + } + } + Ok(()) +} diff --git a/src/cli/subcommands/mod.rs b/src/cli/subcommands/mod.rs index 29546f14b62..3151507164b 100644 --- a/src/cli/subcommands/mod.rs +++ b/src/cli/subcommands/mod.rs @@ -11,6 +11,7 @@ mod chain_cmd; mod config_cmd; mod f3_cmd; mod healthcheck_cmd; +mod index_cmd; mod info_cmd; mod mpool_cmd; mod net_cmd; @@ -22,9 +23,10 @@ mod wait_api_cmd; pub(super) use self::{ auth_cmd::AuthCommands, chain_cmd::ChainCommands, config_cmd::ConfigCommands, - f3_cmd::F3Commands, healthcheck_cmd::HealthcheckCommand, mpool_cmd::MpoolCommands, - net_cmd::NetCommands, shutdown_cmd::ShutdownCommand, snapshot_cmd::SnapshotCommands, - state_cmd::StateCommands, sync_cmd::SyncCommands, wait_api_cmd::WaitApiCommand, + f3_cmd::F3Commands, healthcheck_cmd::HealthcheckCommand, index_cmd::IndexCommands, + mpool_cmd::MpoolCommands, net_cmd::NetCommands, shutdown_cmd::ShutdownCommand, + snapshot_cmd::SnapshotCommands, state_cmd::StateCommands, sync_cmd::SyncCommands, + wait_api_cmd::WaitApiCommand, }; use crate::cli::subcommands::info_cmd::InfoCommand; pub(crate) use crate::cli_shared::cli::Config; @@ -84,6 +86,10 @@ pub enum Subcommand { #[command(subcommand)] Snapshot(SnapshotCommands), + /// Manage the chain index + #[command(subcommand)] + Index(IndexCommands), + /// Print node info #[command(subcommand)] Info(InfoCommand), diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index e01a82af6ce..16853322a45 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -2,14 +2,18 @@ // SPDX-License-Identifier: Apache-2.0, MIT use crate::blocks::Tipset; +use crate::db::SettingsStoreExt; use crate::db::car::forest::{ FOREST_CAR_FILE_EXTENSION, TEMP_FOREST_CAR_FILE_EXTENSION, new_forest_car_temp_path_in, }; use crate::db::car::{ForestCar, ManyCar}; +use crate::ipld::ChainExportState; +use crate::message::SignedMessage; use crate::networks::ChainConfig; use crate::prelude::*; use crate::rpc::sync::SnapshotProgressTracker; use crate::shim::clock::ChainEpoch; +use crate::shim::policy::policy_constants::CHAIN_FINALITY; use crate::state_manager::StateManager; use crate::utils::db::car_stream::CarStream; use crate::utils::io::EitherMmapOrRandomAccessFile; @@ -17,6 +21,8 @@ use crate::utils::net::{DownloadFileOption, download_to}; use anyhow::{Context, bail}; use futures::TryStreamExt; use serde::{Deserialize, Serialize}; +use std::sync::LazyLock; +use std::sync::atomic::{AtomicI64, Ordering}; use std::{ ffi::OsStr, fs, @@ -24,6 +30,8 @@ use std::{ time, }; use tokio::io::AsyncWriteExt; +use tokio::sync::broadcast::error::TryRecvError; +use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; use url::Url; use walkdir::WalkDir; @@ -327,15 +335,304 @@ async fn transcode_into_forest_car(from: &Path, to: &Path) -> anyhow::Result<()> Ok(()) } +/// Settings-store key under which index backfill persists the epoch of the last committed +/// batch, so an interrupted backfill can be resumed from where it left off. +pub const BACKFILL_CHECKPOINT_KEY: &str = "/index/backfill/checkpoint"; + +/// Outcome of indexing a single tipset during backfill. +enum ProcessOutcome { + /// The tipset was indexed. + Indexed, + /// The tipset was skipped because its state output was unavailable and recomputation was + /// disabled (see [`BackfillOptions::allow_recompute`]). + Skipped, +} + +/// Options controlling a backfill run. [`Default`] matches the historical offline behavior: +/// recompute missing state, allow indexing right up to the head, and commit in modest batches. +#[derive(Debug, Clone, Copy)] +pub struct BackfillOptions { + /// When `true`, missing tipset state is recomputed (expensive); when `false`, such tipsets + /// are skipped and reported. Online backfill default this to `false` to avoid starving sync. + pub allow_recompute: bool, + /// When `false`, the walk start is clamped to `head - CHAIN_FINALITY` so that revert-prone + /// near-head tipsets are not indexed. + pub allow_near_head: bool, + /// Number of tipsets to process between commits/checkpoints. + pub batch_size: usize, +} + +impl Default for BackfillOptions { + fn default() -> Self { + Self { + allow_recompute: true, + allow_near_head: true, + batch_size: 1000, + } + } +} + +/// Report returned by [`run_backfill`]. +#[derive(Debug, Clone, Copy, Default)] +pub struct BackfillReport { + pub indexed: u64, + pub skipped: u64, + pub cancelled: bool, +} + +/// Lock-free counters for backfill progress; the epoch counters are hot on the walk. +#[derive(Default)] +struct BackfillCounters { + start_epoch: AtomicI64, + current_epoch: AtomicI64, + target_epoch: AtomicI64, + indexed: AtomicI64, + skipped: AtomicI64, +} + +#[derive(Default)] +struct BackfillStatusInner { + /// `None` while running and before the first run. + outcome: Option, + error: Option, + start_time: Option>, + cancellation_token: Option, + counters: Arc, +} + +impl BackfillStatusInner { + fn is_running(&self) -> bool { + self.cancellation_token.is_some() + } +} + +/// Status of the in-daemon index backfill, surfaced by the `Forest.IndexBackfillStatus` RPC and +/// driven only through [`BackfillGuard`]. Mirrors the life-cycle of [`ChainExportState`]. +#[derive(Default)] +pub struct BackfillStatus { + inner: parking_lot::Mutex, +} + +/// A consistent snapshot of [`BackfillStatus`], read under a single lock. +#[derive(Debug, Clone)] +pub struct BackfillStatusSnapshot { + pub state: ChainExportState, + pub error: Option, + pub start_time: Option>, + pub start_epoch: ChainEpoch, + pub current_epoch: ChainEpoch, + pub target_epoch: ChainEpoch, + pub indexed: u64, + pub skipped: u64, +} + +impl BackfillStatus { + pub fn snapshot(&self) -> BackfillStatusSnapshot { + let inner = self.inner.lock(); + let c = &inner.counters; + BackfillStatusSnapshot { + state: if inner.is_running() { + ChainExportState::Running + } else { + inner.outcome.unwrap_or(ChainExportState::Idle) + }, + error: inner.error.clone(), + start_time: inner.start_time, + start_epoch: c.start_epoch.load(Ordering::Relaxed), + current_epoch: c.current_epoch.load(Ordering::Relaxed), + target_epoch: c.target_epoch.load(Ordering::Relaxed), + indexed: c.indexed.load(Ordering::Relaxed).max(0) as u64, + skipped: c.skipped.load(Ordering::Relaxed).max(0) as u64, + } + } + + /// Cancels the running backfill, if any, returning whether one was running. + pub fn cancel_running(&self) -> bool { + if let Some(token) = &self.inner.lock().cancellation_token { + token.cancel(); + true + } else { + false + } + } + + fn try_begin( + &self, + cancellation_token: CancellationToken, + ) -> anyhow::Result> { + let mut inner = self.inner.lock(); + anyhow::ensure!( + !inner.is_running(), + "an index backfill is already running; check `forest-cli index backfill --status`", + ); + let counters = Arc::new(BackfillCounters::default()); + *inner = BackfillStatusInner { + outcome: None, + error: None, + start_time: Some(chrono::Utc::now()), + cancellation_token: Some(cancellation_token), + counters: counters.clone(), + }; + Ok(counters) + } + + fn record_outcome(&self, outcome: ChainExportState, error: Option) { + let mut inner = self.inner.lock(); + if inner.outcome.is_none() { + inner.outcome = Some(outcome); + inner.error = error; + } + } + + fn end(&self) { + let mut inner = self.inner.lock(); + if inner.outcome.is_none() { + inner.outcome = Some(ChainExportState::Failed); + } + inner.cancellation_token = None; + } +} + +/// Global status of the in-daemon index backfill. +pub static BACKFILL_STATUS: LazyLock = LazyLock::new(BackfillStatus::default); + +/// Single-flight guard for an index backfill. Holds the [`BACKFILL_STATUS`] slot for progress and +/// cancellation, and nests a [`ChainExportGuard`] so backfill never overlaps snapshot exports or +/// the snapshot GC (all three share the chain-export slot). +pub struct BackfillGuard { + cancellation_token: CancellationToken, + counters: Arc, + // Held for the lifetime of the backfill to exclude exports and snapshot GC. + _export_guard: crate::ipld::ChainExportGuard, +} + +impl BackfillGuard { + pub fn try_start() -> anyhow::Result { + // Acquire the shared chain-export slot first so backfill excludes GC/exports. + let export_guard = crate::ipld::ChainExportGuard::try_start_export( + crate::ipld::ChainExportKind::IndexBackfill, + )?; + let cancellation_token = CancellationToken::new(); + let counters = match BACKFILL_STATUS.try_begin(cancellation_token.clone()) { + Ok(counters) => counters, + Err(e) => { + // Roll back the export slot if another backfill is somehow already tracked. + drop(export_guard); + return Err(e); + } + }; + Ok(Self { + cancellation_token, + counters, + _export_guard: export_guard, + }) + } + + pub fn cancellation_token(&self) -> CancellationToken { + self.cancellation_token.clone() + } + + /// Records the terminal outcome for the backfill. + pub fn finish(self, result: anyhow::Result) -> anyhow::Result { + match &result { + Ok(_) => BACKFILL_STATUS.record_outcome(ChainExportState::Succeeded, None), + Err(e) => { + BACKFILL_STATUS.record_outcome(ChainExportState::Failed, Some(format!("{e:#}"))) + } + } + result + } + + fn reset_counters(&self, start_epoch: ChainEpoch, target_epoch: ChainEpoch) { + self.counters + .start_epoch + .store(start_epoch, Ordering::Relaxed); + self.counters + .current_epoch + .store(start_epoch, Ordering::Relaxed); + self.counters + .target_epoch + .store(target_epoch, Ordering::Relaxed); + self.counters.indexed.store(0, Ordering::Relaxed); + self.counters.skipped.store(0, Ordering::Relaxed); + } + + fn set_current(&self, epoch: ChainEpoch) { + self.counters.current_epoch.store(epoch, Ordering::Relaxed); + } + + fn inc_indexed(&self) { + self.counters.indexed.fetch_add(1, Ordering::Relaxed); + } + + fn inc_skipped(&self) { + self.counters.skipped.fetch_add(1, Ordering::Relaxed); + } + + fn record_cancelled(&self) { + BACKFILL_STATUS.record_outcome(ChainExportState::Cancelled, None); + } +} + +impl Drop for BackfillGuard { + fn drop(&mut self) { + self.cancellation_token.cancel(); + BACKFILL_STATUS.end(); + } +} + +/// Reads the persisted backfill checkpoint epoch, if any. A cleared checkpoint (see +/// [`clear_backfill_checkpoint`]) is reported as `None`. +pub fn read_backfill_checkpoint( + state_manager: &StateManager, +) -> anyhow::Result> { + Ok(state_manager + .db() + .read_obj::(BACKFILL_CHECKPOINT_KEY)? + .filter(|epoch| *epoch != ChainEpoch::MIN)) +} + +fn write_backfill_checkpoint( + state_manager: &StateManager, + epoch: ChainEpoch, +) -> anyhow::Result<()> { + state_manager + .db() + .write_obj(BACKFILL_CHECKPOINT_KEY, &epoch) +} + +fn clear_backfill_checkpoint(state_manager: &StateManager) -> anyhow::Result<()> { + // Persist a sentinel rather than deleting: the settings store has no typed delete here and a + // stale checkpoint is only used as a resume hint, which the caller validates against the range. + state_manager + .db() + .write_obj(BACKFILL_CHECKPOINT_KEY, &ChainEpoch::MIN) +} + async fn process_ts( ts: &Tipset, state_manager: &StateManager, - delegated_messages: &mut Vec<(crate::message::SignedMessage, u64)>, -) -> anyhow::Result<()> { + delegated_messages: &mut Vec<(SignedMessage, u64)>, + allow_recompute: bool, +) -> anyhow::Result { let epoch = ts.epoch(); let tsk = ts.key().clone(); - let executed = state_manager.load_executed_tipset(ts).await?; + let executed = match state_manager + .load_executed_tipset_for_backfill(ts, allow_recompute) + .await + { + Ok(executed) => executed, + // With recomputation allowed, a load failure is a real error. With it disabled, a missing + // state output (e.g. reclaimed by GC) is expected: skip and report rather than fail. + Err(e) if allow_recompute => return Err(e), + Err(e) => { + tracing::warn!( + "skipping tipset @{epoch} during backfill (state unavailable, recomputation disabled): {e:#}" + ); + return Ok(ProcessOutcome::Skipped); + } + }; crate::rpc::eth::store_block_logs_bloom( state_manager, ts, @@ -351,9 +648,10 @@ async fn process_ts( tracing::trace!("Indexing tipset @{}: {}", epoch, &tsk); tsk.save(state_manager.db())?; - Ok(()) + Ok(ProcessOutcome::Indexed) } +#[derive(Clone, Copy)] pub enum RangeSpec { To(ChainEpoch), NumTipsets(usize), @@ -375,54 +673,249 @@ impl std::fmt::Display for RangeSpec { /// - [`struct@EthHash`] -> [`TipsetKey`], /// - [`struct@EthHash`] -> Delegated message [`Cid`]. /// -/// This function traverses the chain store and populates these columns accordingly. +/// This function traverses the chain store and populates these columns accordingly. It is a thin +/// wrapper over [`run_backfill`] with the historical (offline) options and no cancellation. pub async fn backfill_db( state_manager: &StateManager, head_ts: &Tipset, spec: RangeSpec, ) -> anyhow::Result<()> { + let guard = BackfillGuard::try_start()?; + let result = run_backfill( + state_manager, + head_ts, + spec, + BackfillOptions::default(), + &guard, + ) + .await; + let report = guard.finish(result)?; + tracing::info!( + "Total successful backfills: {} (skipped: {})", + report.indexed, + report.skipped + ); + Ok(()) +} + +/// Hardened index backfill core shared by the offline `forest-tool index backfill` command and the +/// online `Forest.IndexBackfill` RPC method. +/// +/// Beyond the plain chain walk it: +/// - clamps the start below `CHAIN_FINALITY` unless [`BackfillOptions::allow_near_head`] is set, +/// - commits and checkpoints in batches of [`BackfillOptions::batch_size`] so a large range is not +/// a single transaction and can be resumed, +/// - honors `cancel` between tipsets, +/// - writes Ethereum mappings with newest-wins semantics so it does not clobber the live head +/// indexer, and +/// - re-indexes tipsets applied while the walk was running (revert-awareness). +/// +/// Progress is published to [`BACKFILL_STATUS`] via `guard`. +pub async fn run_backfill( + state_manager: &StateManager, + from_ts: &Tipset, + spec: RangeSpec, + options: BackfillOptions, + guard: &BackfillGuard, +) -> anyhow::Result { tracing::info!("Starting index backfill..."); - let mut delegated_messages = vec![]; + let cancel = guard.cancellation_token(); - let mut num_backfills = 0; + // Subscribe before the walk so applies/reverts that happen during it are observed. + let mut head_rx = state_manager.chain_store().subscribe_head_changes(); - match spec { - RangeSpec::To(to_epoch) => { - for ts in head_ts - .shallow_clone() - .chain(&state_manager.chain_store().db()) - .take_while(|ts| ts.epoch() >= to_epoch) - { - process_ts(&ts, state_manager, &mut delegated_messages).await?; - num_backfills += 1; - } + // Optionally clamp the start below finality to avoid indexing revert-prone near-head tipsets. + let start_ts = if options.allow_near_head { + from_ts.shallow_clone() + } else { + let head_epoch = state_manager.heaviest_tipset().epoch(); + let safe_epoch = head_epoch.saturating_sub(CHAIN_FINALITY); + if from_ts.epoch() > safe_epoch { + state_manager + .chain_index() + .load_required_tipset_by_height( + safe_epoch, + from_ts.shallow_clone(), + crate::chain::index::ResolveNullTipset::TakeOlder, + ) + .await? + } else { + from_ts.shallow_clone() } - RangeSpec::NumTipsets(n_tipsets) => { - for ts in head_ts - .shallow_clone() - .chain(&state_manager.chain_store().db()) - .take(n_tipsets) - { - process_ts(&ts, state_manager, &mut delegated_messages).await?; - num_backfills += 1; + }; + + let target_epoch = match spec { + RangeSpec::To(to_epoch) => to_epoch, + // Not known exactly ahead of time; approximate for progress reporting. + RangeSpec::NumTipsets(n) => start_ts.epoch().saturating_sub(n as ChainEpoch), + }; + guard.reset_counters(start_ts.epoch(), target_epoch); + + let mut batch: Vec<(SignedMessage, u64)> = vec![]; + let mut report = BackfillReport::default(); + let mut processed_since_flush = 0usize; + let mut lowest_epoch = start_ts.epoch(); + + for (count, ts) in start_ts + .shallow_clone() + .chain(&state_manager.chain_store().db()) + .enumerate() + { + match spec { + RangeSpec::To(to_epoch) if ts.epoch() < to_epoch => break, + RangeSpec::NumTipsets(n) if count >= n => break, + _ => {} + } + + if cancel.is_cancelled() { + report.cancelled = true; + break; + } + + guard.set_current(ts.epoch()); + lowest_epoch = ts.epoch(); + match process_ts(&ts, state_manager, &mut batch, options.allow_recompute).await? { + ProcessOutcome::Indexed => { + report.indexed += 1; + guard.inc_indexed(); + } + ProcessOutcome::Skipped => { + report.skipped += 1; + guard.inc_skipped(); } } + processed_since_flush += 1; + + if processed_since_flush >= options.batch_size { + state_manager + .chain_store() + .process_signed_messages(&batch, true)?; + batch.clear(); + write_backfill_checkpoint(state_manager, ts.epoch())?; + processed_since_flush = 0; + } } + // Final commit of the trailing batch. state_manager .chain_store() - .process_signed_messages(&delegated_messages)?; + .process_signed_messages(&batch, true)?; + batch.clear(); - tracing::info!("Total successful backfills: {}", num_backfills); + // Revert-awareness: re-index tipsets applied while walking so the canonical mapping wins + // under newest-wins semantics. Only relevant for the range we just covered. + if !report.cancelled { + let mut extra: Vec<(SignedMessage, u64)> = vec![]; + loop { + match head_rx.try_recv() { + Ok(changes) => { + for ts in changes.applies { + if ts.epoch() >= lowest_epoch && ts.epoch() <= start_ts.epoch() { + tracing::debug!( + "re-indexing tipset @{} applied during backfill", + ts.epoch() + ); + if let Err(e) = + process_ts(&ts, state_manager, &mut extra, options.allow_recompute) + .await + { + tracing::warn!( + "failed to re-index applied tipset @{}: {e:#}", + ts.epoch() + ); + } + } + } + } + Err(TryRecvError::Empty) | Err(TryRecvError::Closed) => break, + Err(TryRecvError::Lagged(n)) => { + tracing::warn!("backfill head-change listener lagged: skipped {n} events"); + continue; + } + } + } + if !extra.is_empty() { + state_manager + .chain_store() + .process_signed_messages(&extra, true)?; + } + } - Ok(()) + if report.cancelled { + // Persist where we stopped so the run can be resumed, and reflect the cancellation. + write_backfill_checkpoint(state_manager, lowest_epoch)?; + guard.record_cancelled(); + tracing::info!( + "Index backfill cancelled after {} tipsets (skipped: {})", + report.indexed, + report.skipped + ); + } else { + // Successful completion: clear the resume checkpoint. + clear_backfill_checkpoint(state_manager)?; + } + + Ok(report) } #[cfg(test)] mod test { use super::*; + // The backfill guard shares the chain-export single-flight slot, so serialize with the export + // tests that also touch it. + #[test] + #[serial_test::serial(chain_export)] + fn backfill_guard_is_single_flight_and_records_outcomes() { + let g = BackfillGuard::try_start().unwrap(); + assert_eq!(BACKFILL_STATUS.snapshot().state, ChainExportState::Running); + + // A second concurrent backfill is rejected while the first is running. + assert!(BackfillGuard::try_start().is_err()); + + // Succeeded is recorded via `finish`. + g.finish(anyhow::Ok(())).unwrap(); + assert_eq!( + BACKFILL_STATUS.snapshot().state, + ChainExportState::Succeeded + ); + + // A new run can start once the previous one finished, and failures are recorded. + let g = BackfillGuard::try_start().unwrap(); + g.finish(anyhow::Result::<()>::Err(anyhow::anyhow!("boom"))) + .unwrap_err(); + let snapshot = BACKFILL_STATUS.snapshot(); + assert_eq!(snapshot.state, ChainExportState::Failed); + assert_eq!(snapshot.error.as_deref(), Some("boom")); + + // A guard dropped without `finish` lands in `Failed`. + let g = BackfillGuard::try_start().unwrap(); + drop(g); + assert_eq!(BACKFILL_STATUS.snapshot().state, ChainExportState::Failed); + } + + #[test] + #[serial_test::serial(chain_export)] + fn backfill_cancellation_is_observable_and_wins() { + let g = BackfillGuard::try_start().unwrap(); + // The cancel handler cancels the running backfill. + assert!(BACKFILL_STATUS.cancel_running()); + assert!(g.cancellation_token().is_cancelled()); + + // The cooperative-cancel path records `Cancelled`, and that terminal state wins over a + // subsequent `finish(Ok(..))` (as happens when the walk returns a cancelled report). + g.record_cancelled(); + g.finish(anyhow::Ok(())).unwrap(); + assert_eq!( + BACKFILL_STATUS.snapshot().state, + ChainExportState::Cancelled + ); + + // With no backfill running, cancel is a no-op. + assert!(!BACKFILL_STATUS.cancel_running()); + } + #[tokio::test] async fn import_snapshot_from_file_valid() { for import_mode in [ImportMode::Auto, ImportMode::Copy, ImportMode::Move] { diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index ad491f3e7ff..7ca6cbe5ec2 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -729,7 +729,9 @@ fn maybe_start_indexer_service( tracing::debug!("Indexing tipset {}", ts.key()); let delegated_messages = chain_store .headers_delegated_messages(ts.block_headers().iter())?; - chain_store.process_signed_messages(&delegated_messages)?; + // Head indexing writes the newest tipset, so use the blind-write + // fast path (no read-before-write timestamp comparison). + chain_store.process_signed_messages(&delegated_messages, false)?; } } Err(RecvError::Lagged(n)) => { diff --git a/src/db/gc/snapshot.rs b/src/db/gc/snapshot.rs index 1654dcbba94..e28e0b501c0 100644 --- a/src/db/gc/snapshot.rs +++ b/src/db/gc/snapshot.rs @@ -182,6 +182,12 @@ impl SnapshotGarbageCollector { } } + /// Whether a snapshot GC run is currently in progress. Index backfill checks this to avoid + /// reading historical state/blocks while the GC is reclaiming graph columns. + pub fn is_running(&self) -> bool { + self.running.load(Ordering::Relaxed) + } + pub fn trigger(&self) -> anyhow::Result>> { if self.running.load(Ordering::Relaxed) { anyhow::bail!("snap gc has already been running"); diff --git a/src/ipld/export_status.rs b/src/ipld/export_status.rs index ecc58028047..d651fb9acf2 100644 --- a/src/ipld/export_status.rs +++ b/src/ipld/export_status.rs @@ -30,6 +30,9 @@ pub enum ChainExportKind { DiffSnapshot, /// A lite snapshot export performed by the automatic snapshot GC. SnapshotGc, + /// An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight + /// slot as exports and the snapshot GC so that these heavy DB operations never overlap. + IndexBackfill, } /// Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index a3c93387ae2..c5063a0962b 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -601,6 +601,233 @@ impl RpcMethod<0> for ForestChainExportCancel { } } +/// Parameters for [`IndexBackfill`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct IndexBackfillParams { + /// Starting epoch, inclusive. Defaults to the persisted resume checkpoint if present, + /// otherwise the chain head. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub from: Option, + /// Ending epoch, inclusive. Mutually exclusive with `n_tipsets`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub to: Option, + /// Number of tipsets to backfill. Mutually exclusive with `to`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub n_tipsets: Option, + /// Recompute missing tipset state (expensive) instead of skipping it. + #[serde(default)] + pub recompute: bool, + /// Allow indexing revert-prone tipsets within `CHAIN_FINALITY` of the head. + #[serde(default)] + pub allow_near_head: 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, + 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>, +} +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::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 { + // Validate synchronously so bad requests surface immediately to the caller. + let spec = index_backfill_range_spec(¶ms)?; + + // Refuse to run while the snapshot GC is active: it may reclaim historical graph columns + // that the backfill needs to read. + 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 { + use crate::daemon::db_util::RangeSpec; + match (params.to, params.n_tipsets) { + (Some(x), None) => Ok(RangeSpec::To(x)), + (None, Some(x)) => Ok(RangeSpec::NumTipsets(x as usize)), + (None, None) => Err(anyhow::anyhow!("You must provide either 'to' or 'n_tipsets'.").into()), + (Some(_), Some(_)) => { + Err(anyhow::anyhow!("'to' and 'n_tipsets' are mutually exclusive.").into()) + } + } +} + +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 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) = read_backfill_checkpoint(&ctx.state_manager)? { + 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::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 { + 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::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 { + Ok(crate::daemon::db_util::BACKFILL_STATUS.cancel_running()) + } +} + pub enum ForestChainExportDiff {} impl RpcMethod<1> for ForestChainExportDiff { const NAME: &'static str = "Forest.ChainExportDiff"; diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index a390df63e6a..2463e3d913b 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -98,6 +98,9 @@ macro_rules! for_each_rpc_method { $callback!($crate::rpc::chain::ForestChainExportDiff); $callback!($crate::rpc::chain::ForestChainExportStatus); $callback!($crate::rpc::chain::ForestChainExportCancel); + $callback!($crate::rpc::chain::IndexBackfill); + $callback!($crate::rpc::chain::IndexBackfillStatus); + $callback!($crate::rpc::chain::IndexBackfillCancel); $callback!($crate::rpc::chain::ChainGetTipsetByParentState); // common vertical diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap index a744533b49f..ef2ffe1c6e6 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap @@ -435,6 +435,37 @@ methods: schema: type: boolean paramStructure: by-position + - name: Forest.IndexBackfill + description: "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." + params: + - name: params + required: true + schema: + $ref: "#/components/schemas/IndexBackfillParams" + result: + name: Forest.IndexBackfill.Result + required: true + schema: + type: "null" + paramStructure: by-position + - name: Forest.IndexBackfillStatus + description: Returns the progress and status of the in-progress (or last) index backfill. + params: [] + result: + name: Forest.IndexBackfillStatus.Result + required: true + schema: + $ref: "#/components/schemas/ApiIndexBackfillStatus" + paramStructure: by-position + - name: Forest.IndexBackfillCancel + description: "Cancels the in-progress index backfill, returning whether one was running." + params: [] + result: + name: Forest.IndexBackfillCancel.Result + required: true + schema: + type: boolean + paramStructure: by-position - name: Forest.ChainGetTipsetByParentState description: "Returns the tipset whose parent state root matches the given CID, or null if none is found." params: @@ -4915,6 +4946,49 @@ components: - MinerID - FromInstance - ValidityTerm + ApiIndexBackfillStatus: + description: "Progress and status of the in-daemon index backfill, returned by [`IndexBackfillStatus`]." + type: object + properties: + currentEpoch: + type: integer + format: int64 + error: + type: + - string + - "null" + indexed: + type: integer + format: uint64 + minimum: 0 + progress: + type: number + format: double + skipped: + type: integer + format: uint64 + minimum: 0 + startEpoch: + type: integer + format: int64 + startTime: + type: + - string + - "null" + format: date-time + state: + $ref: "#/components/schemas/ChainExportState" + targetEpoch: + type: integer + format: int64 + required: + - state + - progress + - startEpoch + - currentEpoch + - targetEpoch + - indexed + - skipped ApiInvocResult: type: object properties: @@ -5336,6 +5410,9 @@ components: - description: A lite snapshot export performed by the automatic snapshot GC. type: string const: SnapshotGc + - description: "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap." + type: string + const: IndexBackfill ChainExportParams: type: object properties: @@ -6746,6 +6823,37 @@ components: - RebroadcastBackoffExponent - RebroadcastBackoffSpread - RebroadcastBackoffMax + IndexBackfillParams: + description: "Parameters for [`IndexBackfill`]." + type: object + properties: + allowNearHead: + description: "Allow indexing revert-prone tipsets within `CHAIN_FINALITY` of the head." + type: boolean + default: false + from: + description: "Starting epoch, inclusive. Defaults to the persisted resume checkpoint if present,\notherwise the chain head." + type: + - integer + - "null" + format: int64 + nTipsets: + description: "Number of tipsets to backfill. Mutually exclusive with `to`." + type: + - integer + - "null" + format: uint64 + minimum: 0 + recompute: + description: Recompute missing tipset state (expensive) instead of skipping it. + type: boolean + default: false + to: + description: "Ending epoch, inclusive. Mutually exclusive with `n_tipsets`." + type: + - integer + - "null" + format: int64 KeyInfo: type: object properties: diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap index ec075912fae..59fa9cf802a 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap @@ -430,6 +430,37 @@ methods: schema: type: boolean paramStructure: by-position + - name: Forest.IndexBackfill + description: "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." + params: + - name: params + required: true + schema: + $ref: "#/components/schemas/IndexBackfillParams" + result: + name: Forest.IndexBackfill.Result + required: true + schema: + type: "null" + paramStructure: by-position + - name: Forest.IndexBackfillStatus + description: Returns the progress and status of the in-progress (or last) index backfill. + params: [] + result: + name: Forest.IndexBackfillStatus.Result + required: true + schema: + $ref: "#/components/schemas/ApiIndexBackfillStatus" + paramStructure: by-position + - name: Forest.IndexBackfillCancel + description: "Cancels the in-progress index backfill, returning whether one was running." + params: [] + result: + name: Forest.IndexBackfillCancel.Result + required: true + schema: + type: boolean + paramStructure: by-position - name: Forest.ChainGetTipsetByParentState description: "Returns the tipset whose parent state root matches the given CID, or null if none is found." params: @@ -4984,6 +5015,49 @@ components: - MinerID - FromInstance - ValidityTerm + ApiIndexBackfillStatus: + description: "Progress and status of the in-daemon index backfill, returned by [`IndexBackfillStatus`]." + type: object + properties: + currentEpoch: + type: integer + format: int64 + error: + type: + - string + - "null" + indexed: + type: integer + format: uint64 + minimum: 0 + progress: + type: number + format: double + skipped: + type: integer + format: uint64 + minimum: 0 + startEpoch: + type: integer + format: int64 + startTime: + type: + - string + - "null" + format: date-time + state: + $ref: "#/components/schemas/ChainExportState" + targetEpoch: + type: integer + format: int64 + required: + - state + - progress + - startEpoch + - currentEpoch + - targetEpoch + - indexed + - skipped ApiInvocResult: type: object properties: @@ -5405,6 +5479,9 @@ components: - description: A lite snapshot export performed by the automatic snapshot GC. type: string const: SnapshotGc + - description: "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap." + type: string + const: IndexBackfill ChainExportParams: type: object properties: @@ -6958,6 +7035,37 @@ components: - RebroadcastBackoffExponent - RebroadcastBackoffSpread - RebroadcastBackoffMax + IndexBackfillParams: + description: "Parameters for [`IndexBackfill`]." + type: object + properties: + allowNearHead: + description: "Allow indexing revert-prone tipsets within `CHAIN_FINALITY` of the head." + type: boolean + default: false + from: + description: "Starting epoch, inclusive. Defaults to the persisted resume checkpoint if present,\notherwise the chain head." + type: + - integer + - "null" + format: int64 + nTipsets: + description: "Number of tipsets to backfill. Mutually exclusive with `to`." + type: + - integer + - "null" + format: uint64 + minimum: 0 + recompute: + description: Recompute missing tipset state (expensive) instead of skipping it. + type: boolean + default: false + to: + description: "Ending epoch, inclusive. Mutually exclusive with `n_tipsets`." + type: + - integer + - "null" + format: int64 KeyInfo: type: object properties: diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap index 4a908fd7e2c..ed4cf5408a9 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap @@ -1946,6 +1946,9 @@ components: - description: A lite snapshot export performed by the automatic snapshot GC. type: string const: SnapshotGc + - description: "An index backfill requested via `Forest.IndexBackfill`. It holds the same single-flight\nslot as exports and the snapshot GC so that these heavy DB operations never overlap." + type: string + const: IndexBackfill ChainExportState: description: "Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then\nexactly one terminal state once it drops. `Idle`: no export has run since node start." type: string diff --git a/src/state_manager/state_computation.rs b/src/state_manager/state_computation.rs index 7228961528d..f153ec3bca7 100644 --- a/src/state_manager/state_computation.rs +++ b/src/state_manager/state_computation.rs @@ -101,6 +101,23 @@ impl StateManager { .await } + /// Load an executed tipset for index backfill. When `allow_state_compute` is `false`, + /// tipsets whose state output is missing (e.g. reclaimed by GC) return an error instead of + /// being recomputed, so a live backfill does not starve chain sync of CPU; the caller is + /// expected to skip such tipsets. + pub async fn load_executed_tipset_for_backfill( + &self, + ts: &Tipset, + allow_state_compute: bool, + ) -> anyhow::Result { + let policy = if allow_state_compute { + StateRecomputePolicy::Allowed + } else { + StateRecomputePolicy::Disallowed + }; + self.load_executed_tipset_with_cache(ts, policy).await + } + async fn load_executed_tipset_with_cache( &self, ts: &Tipset,