diff --git a/CHANGELOG.md b/CHANGELOG.md index d48ff90f2925..b73faa3045b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ ### Added +- [#6802](https://github.com/ChainSafe/forest/issues/6802): When importing a snapshot from a URL, Forest now checks the snapshot's `.metadata.json` and, if extended snapshots are advertised, downloads and imports the receipts, events, and the tipset lookup alongside it. + ### Changed - [#7702](https://github.com/ChainSafe/forest/issues/7702): `eth_estimateGas` matches Lotus v1.38.0 and reth. The estimate never exceeds the call's `gas` or, with `gasPrice` or `maxFeePerGas` set, the gas the sender can pay for after the value; without a price the sender's balance does not limit the gas. A revert that more gas fixes is searched past, and failures use the Ethereum codes: `-32000` when the cap is below the message inclusion cost or the sender affords no gas, `-32003` when the call runs out of gas within the cap or, with a price, the value exceeds the balance, `-32602` when both prices are set. See [lotus#13911](https://github.com/filecoin-project/lotus/pull/13911). diff --git a/scripts/tests/external-rpc-checks/init.sh b/scripts/tests/external-rpc-checks/init.sh index de1b89a51221..d032c3f6b182 100755 --- a/scripts/tests/external-rpc-checks/init.sh +++ b/scripts/tests/external-rpc-checks/init.sh @@ -1,14 +1,11 @@ #!/usr/bin/env bash -# Runs in the `init` service: imports the snapshot ./resolve.rb picked and -# back-fills the chain index, all before the daemon starts. +# Runs in the `init` service: imports the snapshot ./resolve.rb picked, along +# with its extended snapshots (receipts and events, tipset lookup), before the +# daemon starts. set -euo pipefail url=$(< /data/snapshot-url) -epoch=$(< /data/snapshot-epoch) -read -r chain start _ < /data/check-target +read -r chain _ < /data/check-target forest --chain="${chain}" --encrypt-keystore=false --import-snapshot="${url}" --halt-after-import - -# inclusive on both ends -forest-tool index backfill --chain="${chain}" --from="${epoch}" --to="${start}" diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index 2895c2c86fec..39c9151d20ff 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -1,12 +1,13 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT -use crate::blocks::Tipset; -use crate::db::SettingsStoreExt; +use crate::blocks::{Tipset, TipsetKey}; +use crate::chain::{TIPSET_LOOKUP_HAMT_BIT_WIDTH, index::ChainIndex}; 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::db::{EthMappingsStore, MemoryDB, SettingsStoreExt}; use crate::ipld::ChainExportState; use crate::message::SignedMessage; use crate::networks::ChainConfig; @@ -14,10 +15,11 @@ use crate::prelude::*; use crate::rpc::sync::SnapshotProgressTracker; use crate::shim::clock::ChainEpoch; use crate::state_manager::StateManager; -use crate::utils::db::car_stream::CarStream; +use crate::utils::db::car_stream::{CarBlock, CarStream}; use crate::utils::io::EitherMmapOrRandomAccessFile; use crate::utils::net::{DownloadFileOption, download_to}; use anyhow::{Context, bail}; +use fil_actors_shared::fvm_ipld_hamt::Hamt; use futures::TryStreamExt; use serde::{Deserialize, Serialize}; use std::sync::LazyLock; @@ -37,9 +39,6 @@ use walkdir::WalkDir; #[cfg(doc)] use crate::rpc::eth::types::EthHash; -#[cfg(doc)] -use crate::blocks::TipsetKey; - #[cfg(doc)] use cid::Cid; @@ -333,6 +332,222 @@ async fn transcode_into_forest_car(from: &Path, to: &Path) -> anyhow::Result<()> Ok(()) } +/// The subset of the snapshot `.metadata.json` file published in the Forest archive +/// that is needed to locate the extended snapshots. +#[derive(Debug, Deserialize)] +struct SnapshotMetadataFile { + #[serde(rename = "Snapshot")] + snapshot: SnapshotMetadataSection, +} + +#[derive(Debug, Deserialize)] +struct SnapshotMetadataSection { + #[serde(rename = "Extended", default)] + extended: Option, +} + +/// Auxiliary snapshots that enrich a base snapshot, see `Forest.ChainExport`. +#[derive(Debug, Default, Deserialize, PartialEq)] +struct ExtendedSnapshots { + /// Message receipts and events + #[serde(rename = "Augmented", default)] + receipts_events: Option, + /// Tipset lookup table (checkpoint epoch to tipset key) + #[serde(rename = "Tipset lookup", default)] + tipset_lookup: Option, +} + +#[derive(Debug, Deserialize, PartialEq)] +struct ExtendedSnapshotFile { + #[serde(rename = "Filename")] + filename: String, + #[serde(rename = "Sha256")] + sha256: String, +} + +/// Fetches the `.metadata.json` published next to `snapshot_url` and returns its extended +/// snapshots, if any. +async fn fetch_extended_snapshots(snapshot_url: &Url) -> anyhow::Result> { + let metadata_url = Url::parse(&format!("{snapshot_url}.metadata.json"))?; + let response = crate::utils::net::global_http_client() + .get(metadata_url.clone()) + .send() + .await?; + if response.status() == reqwest::StatusCode::NOT_FOUND { + debug!("snapshot metadata not found at {metadata_url}"); + return Ok(None); + } + let metadata: SnapshotMetadataFile = response + .error_for_status()? + .json() + .await + .with_context(|| format!("failed to parse snapshot metadata at {metadata_url}"))?; + Ok(metadata.snapshot.extended) +} + +/// Number of blocks or tipset lookup entries written to the database per commit when importing +/// extended snapshots. +const EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE: usize = 1_000; + +async fn download_extended_snapshot( + snapshot_url: &Url, + file: &ExtendedSnapshotFile, + temp_dir: &Path, +) -> anyhow::Result { + use sha2::{Digest as _, Sha256}; + + anyhow::ensure!( + !file.filename.is_empty() && !file.filename.contains(['/', '\\', '?', '#']), + "invalid extended snapshot filename: {}", + file.filename + ); + let url = snapshot_url.join(&file.filename)?; + let path = new_forest_car_temp_path_in(temp_dir)?; + info!("Downloading extended snapshot: {url}"); + download_to(&url, &path, DownloadFileOption::Resumable, None).await?; + + let actual = crate::utils::encoding::hex::encode(Sha256::digest(tokio::fs::read(&path).await?)); + anyhow::ensure!( + actual.eq_ignore_ascii_case(&file.sha256), + "checksum mismatch for {}, expected: {}, actual: {actual}", + file.filename, + file.sha256 + ); + Ok(path) +} + +/// Imports message receipts and events from a receipts and events snapshot into `db`, +/// committing [`EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE`] blocks at a time. +async fn import_receipts_events_snapshot(db: &Arc, path: &Path) -> anyhow::Result +where + DB: Blockstore + Send + Sync + 'static, +{ + async fn flush( + db: &Arc, + batch: Vec<(Cid, bytes::Bytes)>, + ) -> anyhow::Result<()> { + let db = db.clone(); + tokio::task::spawn_blocking(move || db.put_many_keyed(batch)).await? + } + + let mut car_stream = CarStream::new_from_path(path).await?; + let mut n_blocks = 0; + let mut batch = Vec::with_capacity(EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE); + while let Some(CarBlock { cid, data }) = car_stream.try_next().await? { + batch.push((cid, data)); + n_blocks += 1; + if batch.len() == EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE { + flush( + db, + std::mem::replace( + &mut batch, + Vec::with_capacity(EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE), + ), + ) + .await?; + } + } + if !batch.is_empty() { + flush(db, batch).await?; + } + Ok(n_blocks) +} + +/// Populates the tipset lookup table in `db` from a tipset lookup snapshot. +async fn import_tipset_lookup_snapshot( + db: &Arc, + path: &Path, + head_epoch: ChainEpoch, +) -> anyhow::Result +where + DB: EthMappingsStore + Send + Sync + 'static, +{ + let store = MemoryDB::default(); + let mut car_stream = CarStream::new_from_path(path).await?; + let hamt_root = *car_stream.header_v1.roots.first(); + while let Some(CarBlock { cid, data }) = car_stream.try_next().await? { + store.put_keyed(&cid, &data)?; + } + + let db = db.clone(); + tokio::task::spawn_blocking(move || { + let hamt: Hamt<_, TipsetKey, ChainEpoch> = + Hamt::load_with_bit_width(&hamt_root, &store, TIPSET_LOOKUP_HAMT_BIT_WIDTH) + .context("failed to load tipset lookup HAMT")?; + // Validate all entries before writing any of them. + let mut entries = vec![]; + hamt.for_each_cacheless(|&epoch, tsk| { + anyhow::ensure!( + ChainIndex::is_tipset_lookup_checkpoint(epoch) && epoch <= head_epoch, + "invalid tipset lookup entry at epoch {epoch}, head epoch: {head_epoch}", + ); + entries.push((epoch, tsk.clone())); + Ok(()) + })?; + for batch in entries.chunks(EXTENDED_SNAPSHOT_IMPORT_BATCH_SIZE) { + db.set_tipset_keys_at_epochs_raw(batch)?; + } + Ok(entries.len()) + }) + .await? +} + +/// Downloads the extended snapshots (message receipts and events, tipset lookup table) +/// advertised in the metadata of the snapshot at `snapshot_url` and imports them into `db`. +pub async fn maybe_import_extended_snapshots( + db: &Arc, + snapshot_url: &Url, + head: &Tipset, + temp_dir: &Path, +) where + DB: Blockstore + EthMappingsStore + Send + Sync + 'static, +{ + let extended = match fetch_extended_snapshots(snapshot_url).await { + Ok(Some(extended)) => extended, + Ok(None) => { + info!("No extended snapshots advertised for {snapshot_url}"); + return; + } + Err(e) => { + warn!("Failed to fetch snapshot metadata for {snapshot_url}: {e:#}"); + return; + } + }; + + if let Some(file) = &extended.receipts_events { + let result = async { + let path = download_extended_snapshot(snapshot_url, file, temp_dir).await?; + import_receipts_events_snapshot(db, &path).await + } + .await; + match result { + Ok(n) => info!( + "Imported {n} message receipt and event blocks from {}", + file.filename + ), + Err(e) => warn!( + "Failed to import receipts and events snapshot {}: {e:#}", + file.filename + ), + } + } + + if let Some(file) = &extended.tipset_lookup { + let result = async { + let path = download_extended_snapshot(snapshot_url, file, temp_dir).await?; + import_tipset_lookup_snapshot(db, &path, head.epoch()).await + } + .await; + match result { + Ok(n) => info!("Imported {n} tipset lookup entries from {}", file.filename), + Err(e) => warn!( + "Failed to import tipset lookup snapshot {}: {e:#}", + file.filename + ), + } + } +} + /// 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"; @@ -862,6 +1077,44 @@ pub async fn run_backfill( #[cfg(test)] mod test { use super::*; + + #[test] + fn parse_snapshot_metadata_with_extended() { + let json = r#"{ + "Snapshot": { + "Snapshot version": "2", + "Epoch": 4137600, + "Extended": { + "Augmented": { + "Filename": "forest_snapshot_calibnet_2026-10-08_height_4137600_receipts_events.forest.car.zst", + "Sha256": "9585fa94f3c84a117929dd1942014a6f4f79f47dbfdbf693122ff9acfa1311cc" + }, + "Tipset lookup": { + "Filename": "forest_snapshot_calibnet_2026-10-08_height_4137600_tipset_lookup.forest.car.zst", + "Sha256": "710262581a4fdea6a8cd162e4a65b42727475b83c44b0b0fe05aff72ce70c2f2" + } + } + }, + "Build Information": {} + }"#; + let metadata: SnapshotMetadataFile = serde_json::from_str(json).unwrap(); + let extended = metadata.snapshot.extended.unwrap(); + assert_eq!( + extended.receipts_events.unwrap().filename, + "forest_snapshot_calibnet_2026-10-08_height_4137600_receipts_events.forest.car.zst" + ); + assert_eq!( + extended.tipset_lookup.unwrap().sha256, + "710262581a4fdea6a8cd162e4a65b42727475b83c44b0b0fe05aff72ce70c2f2" + ); + } + + #[test] + fn parse_snapshot_metadata_without_extended() { + let json = r#"{ "Snapshot": { "Snapshot version": "2", "Epoch": 4137600 } }"#; + let metadata: SnapshotMetadataFile = serde_json::from_str(json).unwrap(); + assert!(metadata.snapshot.extended.is_none()); + } use rstest::rstest; // The backfill guard shares the chain-export single-flight slot, so serialize with the export diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 82064cfc0f11..3b3933d43a79 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -16,7 +16,10 @@ use crate::cli_shared::{ cli::{CliOpts, Config}, delete_chain_data, }; -use crate::daemon::{context::AppContext, db_util::import_chain_as_forest_car}; +use crate::daemon::{ + context::AppContext, + db_util::{import_chain_as_forest_car, maybe_import_extended_snapshots}, +}; use crate::db::gc::SnapshotGarbageCollector; use crate::db::ttl::EthMappingCollector; use crate::libp2p::{Libp2pService, PeerManager}; @@ -155,6 +158,15 @@ async fn maybe_import_snapshot( .await?; ctx.db .read_only_files(std::iter::once(car_db_path.clone()))?; + if let Ok(snapshot_url) = url::Url::parse(&path.display().to_string()) { + maybe_import_extended_snapshots( + &ctx.db, + &snapshot_url, + &ts, + &ctx.db_meta_data.get_forest_car_db_dir(), + ) + .await; + } let ts_epoch = ts.epoch(); // Explicitly set heaviest tipset here in case HEAD_KEY has already been set // in the current setting store diff --git a/src/db/car/many.rs b/src/db/car/many.rs index 2e870be35bee..96a3a44aee75 100644 --- a/src/db/car/many.rs +++ b/src/db/car/many.rs @@ -321,6 +321,13 @@ impl EthMappingsStore for ManyCar { ) -> anyhow::Result<()> { EthMappingsStore::set_tipset_key_at_epoch_raw(self.writer(), epoch, tsk) } + + fn set_tipset_keys_at_epochs_raw( + &self, + entries: &[(ChainEpoch, TipsetKey)], + ) -> anyhow::Result<()> { + EthMappingsStore::set_tipset_keys_at_epochs_raw(self.writer(), entries) + } } impl EthBlockBloomStore for ManyCar { diff --git a/src/db/mod.rs b/src/db/mod.rs index 90c0f466423a..4d6c53d018ca 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -123,6 +123,18 @@ pub trait EthMappingsStore { fn set_tipset_key_at_epoch_raw(&self, epoch: ChainEpoch, tsk: &TipsetKey) -> anyhow::Result<()>; + /// Writes the tipset keys for the given epochs(heights) to the store. Implementations + /// should commit them in a single batch. + fn set_tipset_keys_at_epochs_raw( + &self, + entries: &[(ChainEpoch, TipsetKey)], + ) -> anyhow::Result<()> { + for (epoch, tsk) in entries { + EthMappingsStore::set_tipset_key_at_epoch_raw(self, *epoch, tsk)?; + } + Ok(()) + } + /// Writes the tipset key for a given epoch(height) to the store. fn set_tipset_key_at_epoch(&self, ts: &Tipset) -> anyhow::Result<()> { EthMappingsStore::set_tipset_key_at_epoch_raw(self, ts.epoch(), ts.key()) diff --git a/src/db/parity_db.rs b/src/db/parity_db.rs index 684ba2f11c92..33dda17f76e6 100644 --- a/src/db/parity_db.rs +++ b/src/db/parity_db.rs @@ -263,6 +263,25 @@ impl EthMappingsStore for ParityDb { let bytes = fvm_ipld_encoding::to_vec(tsk)?; self.write_to_column(key, bytes, DbColumn::EthMappings) } + + fn set_tipset_keys_at_epochs_raw( + &self, + entries: &[(ChainEpoch, TipsetKey)], + ) -> anyhow::Result<()> { + let tx = entries + .iter() + .map(|(epoch, tsk)| { + Ok(( + DbColumn::EthMappings as u8, + epoch.to_le_bytes(), + Some(fvm_ipld_encoding::to_vec(tsk)?), + )) + }) + .collect::>>()?; + self.db + .commit(tx) + .with_context(|| format!("error writing to column {}", DbColumn::EthMappings)) + } } impl EthBlockBloomStore for ParityDb { @@ -636,4 +655,26 @@ mod test { assert!(db.write_ops_broadcast_tx.read().is_none()); } + + #[test] + fn set_tipset_keys_at_epochs_raw_test() { + let db = TempParityDB::new(); + let tsk = |n: u64| { + TipsetKey::from(nunny::vec![Cid::new_v1( + DAG_CBOR, + MultihashCode::Blake2b256.digest(&n.to_le_bytes()) + )]) + }; + let entries = [(20, tsk(1)), (40, tsk(2)), (60, tsk(3))]; + db.set_tipset_keys_at_epochs_raw(&entries).unwrap(); + for (epoch, expected) in &entries { + assert_eq!( + db.tipset_key_by_epoch(*epoch).unwrap().as_ref(), + Some(expected) + ); + } + assert!(db.tipset_key_by_epoch(80).unwrap().is_none()); + // Empty batches are a no-op. + db.set_tipset_keys_at_epochs_raw(&[]).unwrap(); + } } diff --git a/src/db/parity_db/gc.rs b/src/db/parity_db/gc.rs index bbecc51284d1..757160cc2414 100644 --- a/src/db/parity_db/gc.rs +++ b/src/db/parity_db/gc.rs @@ -167,6 +167,13 @@ impl EthMappingsStore for GarbageCollectableParityDb { ) -> anyhow::Result<()> { EthMappingsStore::set_tipset_key_at_epoch_raw(&*self.db.read(), epoch, tsk) } + + fn set_tipset_keys_at_epochs_raw( + &self, + entries: &[(ChainEpoch, TipsetKey)], + ) -> anyhow::Result<()> { + EthMappingsStore::set_tipset_keys_at_epochs_raw(&*self.db.read(), entries) + } } impl EthBlockBloomStore for GarbageCollectableParityDb {