diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f5e7f6a348..e6c1dd5f108 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,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 + - [#7802](https://github.com/ChainSafe/forest/pull/7802): Added `--target ` to `forest-cli sync wait`. The command exits once the node head reaches the given epoch. - [#7781](https://github.com/ChainSafe/forest/pull/7781): Actor bundles are now verified against their CIDs, and missing or tampered blocks are replaced. diff --git a/docs/docs/users/knowledge_base/enriched_snapshots.md b/docs/docs/users/knowledge_base/enriched_snapshots.md index 2160bc68c6c..abfcdc37d50 100644 --- a/docs/docs/users/knowledge_base/enriched_snapshots.md +++ b/docs/docs/users/knowledge_base/enriched_snapshots.md @@ -29,9 +29,9 @@ forest_snapshot_mainnet_2026-08-14_height_6279840_tipset_lookup.forest.car.zst ``` :::info -Enriched snapshots are a recent addition and are not yet listed by the snapshot -service. Discovery is still to be defined; for now the companion files follow the -suffix convention above next to their base snapshot. +The companion files are listed, with their SHA-256, under `Snapshot.Extended` in +the base snapshot's `.metadata.json`, published next to it as +`.metadata.json`. ::: ## Generating @@ -61,6 +61,17 @@ at least one must be given. ## Importing into Forest +### Automatically on startup + +When Forest imports a snapshot from a URL on startup, with `--auto-download-snapshot` or +`--import-snapshot `, it imports the enriched snapshot too, then indexes the +imported receipts and events (block `logsBloom` and Ethereum mappings), so +`FOREST_ETH_RPC_COMPUTE_BLOOM_ON_MISS=1` is only needed for the manual recipe below. +This step is best-effort: if a companion file can't be downloaded, verified, or +imported, Forest logs a warning and starts without it. + +### Manually + `--import-snapshot` accepts a single file and the companion CARs have no chain spine of their own, so only the base goes through it. The companions are loaded with `forest-tool db`, which writes directly into the node's key-value store and diff --git a/scripts/tests/external-rpc-checks/init.sh b/scripts/tests/external-rpc-checks/init.sh index de1b89a5122..d032c3f6b18 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/scripts/tests/external-rpc-checks/resolve.rb b/scripts/tests/external-rpc-checks/resolve.rb index de5121039af..ab8762518ca 100644 --- a/scripts/tests/external-rpc-checks/resolve.rb +++ b/scripts/tests/external-rpc-checks/resolve.rb @@ -1,8 +1,9 @@ # frozen_string_literal: true # Picks the newest ${FOREST_CHAIN} snapshot whose head lies within the first 2000 epochs of -# the UTC day: it holds the state the backfill needs to reach the day's start, -# and the node syncs the rest. Writes its URL, head epoch and the day's epoch range under /data. +# the UTC day: importing it indexes its receipts and events back to the day's start, +# and the node syncs the rest. Writes its URL and the day's epoch range under /data, and fails +# if the snapshot advertises no receipts and events snapshot. require 'date' require 'json' @@ -47,7 +48,9 @@ def fetch(uri, attempts: 3) exit 2 end +metadata = JSON.parse(fetch(URI("#{url}.metadata.json"))) +abort "#{url} advertises no receipts and events snapshot" unless metadata.dig('Snapshot', 'Extended', 'Augmented') + File.write('/data/snapshot-url', "#{url}\n") -File.write('/data/snapshot-epoch', "#{epoch}\n") File.write('/data/check-target', "#{chain} #{start} #{finish}\n") puts "#{chain} #{day.iso8601} is epochs #{start}..#{finish}; snapshot at #{epoch}: #{url}" diff --git a/scripts/tests/external-rpc-checks/verify.rb b/scripts/tests/external-rpc-checks/verify.rb index 57987db1c5f..7894a0fda39 100644 --- a/scripts/tests/external-rpc-checks/verify.rb +++ b/scripts/tests/external-rpc-checks/verify.rb @@ -1,31 +1,35 @@ # frozen_string_literal: true -# Confirms the node indexed the epochs the checks query. +# Confirms the node indexed the epochs the checks query: the tipset is served and the receipts +# of its messages are in the store, read back through the standard chain API. require 'json' require 'net/http' url = URI("http://#{ENV.fetch('FOREST_RPC_URL')}") -ARGV.each do |argument| - epoch = Integer(argument) +def rpc(url, method, params) request = Net::HTTP::Post.new(url, 'Content-Type' => 'application/json') - request.body = { - jsonrpc: '2.0', - id: 1, - method: 'eth_getBlockByNumber', - params: ["0x#{epoch.to_s(16)}", false] - }.to_json + request.body = { jsonrpc: '2.0', id: 1, method: method, params: params }.to_json + JSON.parse(Net::HTTP.start(url.hostname, url.port) { |http| http.request(request) }.body) +end - response = Net::HTTP.start(url.hostname, url.port) { |http| http.request(request) } - body = JSON.parse(response.body) +ARGV.each do |argument| + epoch = Integer(argument) + block = rpc(url, 'eth_getBlockByNumber', ["0x#{epoch.to_s(16)}", false]) # Error code 12 denotes a null round across implementations. # https://github.com/ChainSafe/forest/blob/8c980e679b6606c534c6eb5bd8aa03ff8cc6f5c0/src/rpc/methods/eth/errors.rs#L22 - null_round = body.dig('error', 'code') == 12 - - unless body.dig('result', 'number') || null_round - abort "epoch #{epoch} is not indexed: #{response.code} #{response.body}" + if block.dig('error', 'code') == 12 + puts "epoch #{epoch} was a null round" + next end + abort "epoch #{epoch} is not indexed: #{block}" unless block.dig('result', 'number') - puts "epoch #{epoch} #{null_round ? 'was a null round' : 'is indexed'}" + # The receipts of an epoch's messages are stored under its child tipset, the first non-null + # tipset above it. Reading them back is the same check any client would make. + child = rpc(url, 'Filecoin.ChainGetTipSetAfterHeight', [epoch + 1, []]) + abort "epoch #{epoch} has no tipset above it: #{child['error']}" if child['error'] + receipts = rpc(url, 'Filecoin.ChainGetParentReceipts', [child.dig('result', 'Cids', 0)]) + abort "epoch #{epoch} has no receipts: #{receipts['error']}" if receipts['error'] + puts "epoch #{epoch} is indexed" end diff --git a/src/chain/mod.rs b/src/chain/mod.rs index ccc7aa5a430..4e37c84e0c9 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -13,8 +13,9 @@ pub use self::{snapshot_format::*, store::*, weight::*}; use crate::blocks::{Tipset, TipsetKey}; use crate::chain::index::{ChainIndex, ResolveNullTipset}; use crate::cid_collections::{CidHashSet, CidHashSetLike, FileBackedCidHashSet}; -use crate::db::IndexMapBlockstore; +use crate::db::car::AnyCar; use crate::db::car::forest::{self, ForestCarFrame, finalize_frame}; +use crate::db::{EthMappingsStore, IndexMapBlockstore}; use crate::ipld::{IpldStream, should_save_block_to_snapshot, stream_chain}; use crate::prelude::*; use crate::shim::executor::Receipt; @@ -481,3 +482,62 @@ pub async fn export_receipts_events_to_forest_car( Ok(()) } + +/// Check that the entry is at a checkpoint epoch and that the stored tipset is at that epoch. +pub fn validate_tipset_lookup_entry( + db: &DB, + epoch: ChainEpoch, + tsk: &TipsetKey, +) -> anyhow::Result +where + DB: Blockstore, +{ + anyhow::ensure!( + ChainIndex::is_tipset_lookup_checkpoint(epoch), + "epoch {epoch} is not a tipset lookup checkpoint" + ); + let ts: Tipset = Tipset::load_required(db, tsk)?; + anyhow::ensure!( + ts.epoch() == epoch, + "epochs do not match, {epoch} in hamt, {} in database", + ts.epoch() + ); + Ok(ts) +} + +/// Imports a tipset lookup snapshot into `db`, calling `on_imported` after each entry is written. +/// Every entry is validated against the stored chain before the first write. +pub fn import_tipset_lookup_snapshot( + db: &DB, + path: &Path, + mut on_imported: impl FnMut(), +) -> anyhow::Result<()> +where + DB: Blockstore + EthMappingsStore, +{ + let started = Instant::now(); + let hamt_car = AnyCar::try_from(path)?; + anyhow::ensure!( + hamt_car.header_v1().roots.len() == 1, + "hamt car should have a single root" + ); + let hamt_root = *hamt_car.header_v1().roots.first(); + let hamt: Hamt<_, TipsetKey, ChainEpoch> = + Hamt::load_with_bit_width(&hamt_root, hamt_car, TIPSET_LOOKUP_HAMT_BIT_WIDTH) + .context("failed to load tipset lookup HAMT")?; + tracing::info!("Validating tipset lookup entries against the chain, this can take a while"); + hamt.for_each_cacheless(|&epoch, tsk| { + validate_tipset_lookup_entry(db, epoch, tsk)?; + Ok(()) + })?; + hamt.for_each_cacheless(|&epoch, tsk| { + db.set_tipset_key_at_epoch_raw(epoch, tsk)?; + on_imported(); + Ok(()) + })?; + tracing::info!( + "Imported tipset lookup entries in {:.1}s", + started.elapsed().as_secs_f64() + ); + Ok(()) +} diff --git a/src/chain/tests.rs b/src/chain/tests.rs index b552e173e3c..c34045896f3 100644 --- a/src/chain/tests.rs +++ b/src/chain/tests.rs @@ -604,3 +604,73 @@ async fn ensure_recent_roots_present_checks_oldest_walked_tipset( expected ); } + +// Imports a lookup table with the given `hamt_epoch -> tipset at tipset_epoch` entries into a +// chain of tipsets at epochs 20, 21 and 40; `None` points at a tipset missing from the chain. +// On an error nothing may have been written, so one invalid entry rejects the valid ones too. +#[rstest] +#[case::checkpoint(&[(40, Some(40))], None)] +#[case::non_checkpoint_epoch( + &[(20, Some(20)), (21, Some(21)), (40, Some(40))], + Some("is not a tipset lookup checkpoint") +)] +#[case::epoch_mismatch(&[(40, Some(20))], Some("epochs do not match"))] +#[case::unknown_tipset(&[(40, None)], Some("Required tipset missing"))] +#[tokio::test] +async fn import_tipset_lookup( + #[case] entries: &[(ChainEpoch, Option)], + #[case] expected_err: Option<&str>, +) { + use crate::chain::index::tests::{genesis_tipset, persist_tipset, tipset_child}; + + let db = MemoryDB::default(); + let mut parent = genesis_tipset(); + persist_tipset(&parent, &db); + let mut chain = vec![]; + for epoch in [20, 21, 40] { + let ts = tipset_child(&parent, epoch); + persist_tipset(&ts, &db); + chain.push(ts.shallow_clone()); + parent = ts; + } + + let hamt_store = MemoryDB::default(); + let mut hamt: Hamt<_, TipsetKey, ChainEpoch> = + Hamt::new_with_bit_width(&hamt_store, TIPSET_LOOKUP_HAMT_BIT_WIDTH); + let mut expected = vec![]; + for &(hamt_epoch, tipset_epoch) in entries { + let tsk = match tipset_epoch { + Some(epoch) => chain + .iter() + .find(|ts| ts.epoch() == epoch) + .unwrap() + .key() + .clone(), + None => tipset_child(&parent, 40).key().clone(), + }; + hamt.set(hamt_epoch, tsk.clone()).unwrap(); + expected.push((hamt_epoch, tsk)); + } + let root = hamt.flush().unwrap(); + let mut car = vec![]; + hamt_store + .export_forest_car_with_roots(nunny::vec![root], &mut car) + .await + .unwrap(); + let temp_dir = tempfile::tempdir().unwrap(); + let car_path = temp_dir.path().join("tipset_lookup.forest.car.zst"); + std::fs::write(&car_path, car).unwrap(); + + let result = import_tipset_lookup_snapshot(&db, &car_path, || {}); + match expected_err { + None => result.unwrap(), + Some(expected) => { + let err = format!("{:#}", result.unwrap_err()); + assert!(err.contains(expected), "{err}"); + } + } + for (hamt_epoch, tsk) in expected { + let written = expected_err.is_none().then_some(tsk); + assert_eq!(db.tipset_key_by_epoch(hamt_epoch).unwrap(), written); + } +} diff --git a/src/cli_shared/snapshot.rs b/src/cli_shared/snapshot.rs index 68059207492..a321764bdef 100644 --- a/src/cli_shared/snapshot.rs +++ b/src/cli_shared/snapshot.rs @@ -5,6 +5,7 @@ use std::{ fmt::Display, path::{Path, PathBuf}, str::FromStr, + time::Duration, }; use crate::prelude::*; @@ -112,6 +113,24 @@ pub async fn peek( )) } +/// Returns the final URL after redirection(s). +/// if server rejects `HEAD` requests returns the `url` itself +pub async fn peek_url(url: Url) -> Url { + let response = crate::utils::net::global_http_client() + .head(url.clone()) + .timeout(Duration::from_secs(30)) + .send() + .await + .and_then(|response| response.error_for_status()); + match response { + Ok(response) => response.url().clone(), + Err(e) => { + tracing::debug!("Couldn't follow the redirects of {url}: {e:#}"); + url + } + } +} + // Extract file paths from content-disposition values: // "attachment; filename=\"911520_2023_09_14T06_13_00Z.car.zst\"" // => "911520_2023_09_14T06_13_00Z.car.zst" @@ -434,8 +453,41 @@ mod parse { #[cfg(test)] mod tests { - use super::parse_content_disposition; + use super::{parse_content_disposition, peek_url}; use reqwest::header::HeaderValue; + use url::Url; + + #[rstest::rstest] + #[case::follows_redirect("/latest", "/snapshot.forest.car.zst")] + #[case::head_rejected("/post-only", "/post-only")] + #[case::not_found("/missing", "/missing")] + #[case::redirect_to_rejected("/stale", "/stale")] + #[tokio::test] + async fn peek_url_falls_back_to_given_url(#[case] path: &str, #[case] expected: &str) { + use axum::{ + Router, + response::Redirect, + routing::{get, post}, + }; + + let app = Router::new() + .route( + "/latest", + get(|| async { Redirect::temporary("/snapshot.forest.car.zst") }), + ) + .route("/snapshot.forest.car.zst", get(|| async { "snapshot" })) + .route("/post-only", post(|| async { "post" })) + .route( + "/stale", + get(|| async { Redirect::temporary("/post-only") }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + + let url = Url::parse(&format!("http://{addr}{path}")).unwrap(); + assert_eq!(peek_url(url).await.path(), expected); + } #[test] fn content_disposition_forest() { diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index 32fa4ea8ed2..5e2f818798e 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -154,7 +154,7 @@ pub async fn import_chain_as_forest_car( let forest_car_db_path = forest_car_db_path.clone(); async move { let downloaded_car_temp_path = new_forest_car_temp_path_in(forest_car_db_dir)?; - if let Ok(url) = Url::parse(&from_path.display().to_string()) { + if let Some(url) = path_as_url(from_path) { download_to( &url, &downloaded_car_temp_path, @@ -194,7 +194,7 @@ pub async fn import_chain_as_forest_car( match import_mode { ImportMode::Auto => { - if Url::parse(&from_path.display().to_string()).is_ok() { + if path_as_url(from_path).is_some() { // Fallback to move if from_path is url move_or_copy(ImportMode::Move).await?; } else if ForestCar::is_valid(&EitherMmapOrRandomAccessFile::open(from_path)?) { @@ -292,6 +292,11 @@ pub async fn import_chain_as_forest_car( Ok((forest_car_db_path, ts)) } +/// Parses `path` as a URL, returning `None` for local paths. +pub(super) fn path_as_url(path: &Path) -> Option { + path.to_str().and_then(|s| Url::parse(s).ok()) +} + fn move_or_copy_file(from: &Path, to: &Path, import_mode: ImportMode) -> anyhow::Result<()> { match import_mode { ImportMode::Move => { @@ -859,6 +864,9 @@ pub async fn run_backfill( Ok(report) } +#[cfg(test)] +pub(crate) use test::test_state_manager; + #[cfg(test)] mod test { use super::*; @@ -934,7 +942,7 @@ mod test { drop(bf); } - fn test_state_manager() -> StateManager { + pub(crate) fn test_state_manager() -> StateManager { use crate::blocks::{CachingBlockHeader, RawBlockHeader}; use crate::chain::ChainStore; use crate::networks::ChainConfig; diff --git a/src/daemon/extended_snapshot.rs b/src/daemon/extended_snapshot.rs new file mode 100644 index 00000000000..c8d61c02aa2 --- /dev/null +++ b/src/daemon/extended_snapshot.rs @@ -0,0 +1,610 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +//! Download and import of the extended snapshots (receipts/events and tipset lookup) +//! published next to a snapshot in the Forest archive. + +use crate::blocks::Tipset; +use crate::chain::{ChainEpochDelta, import_tipset_lookup_snapshot}; +use crate::daemon::db_util::{ + BackfillGuard, BackfillOptions, BackfillReport, RangeSpec, run_backfill, +}; +use crate::db::car::forest::new_forest_car_temp_path_in; +use crate::prelude::*; +use crate::state_manager::StateManager; +use crate::utils::db::car_util::import_car_blocks; +use crate::utils::io::progress_log::Progress; +use crate::utils::net::{DownloadFileOption, download_to_with_timeout}; +use anyhow::Context; +use regex::Regex; +use serde::Deserialize; +use sha2::Sha256; +use std::path::Path; +use std::time::{Duration, Instant}; +use tracing::{info, warn}; +use url::Url; + +/// 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 { + /// Number of epochs below the head with state, which the receipts and events also cover + #[serde(rename = "State-roots")] + state_roots: ChainEpochDelta, + #[serde(rename = "Extended")] + extended: Option, +} + +/// Auxiliary snapshots that enrich a base snapshot, see `Forest.ChainExport`. +#[derive(Debug, Deserialize)] +struct ExtendedSnapshots { + /// Message receipts and events + #[serde(rename = "Augmented")] + receipts_events: Option, + /// Tipset lookup table (checkpoint epoch to tipset key) + #[serde(rename = "Tipset lookup")] + tipset_lookup: Option, +} + +#[derive(Debug, Deserialize)] +struct ExtendedSnapshotFile { + #[serde(rename = "Filename")] + filename: String, + #[serde(rename = "Sha256")] + #[serde(with = "crate::utils::encoding::hex::serde")] + sha256: [u8; 32], +} + +/// Fetches the `.metadata.json` published next to `snapshot_url`, or `None` when the host has +/// none (a 404), which is normal for a snapshot without extended snapshots. +async fn fetch_metadata_snapshots_info( + snapshot_url: &Url, +) -> anyhow::Result> { + // Append to the path, not the whole URL, so dont affect any query string. + let mut metadata_url = snapshot_url.clone(); + metadata_url.set_path(&format!("{}.metadata.json", snapshot_url.path())); + let response = crate::utils::net::global_http_client() + .get(metadata_url.clone()) + .timeout(Duration::from_secs(30)) + .send() + .await?; + if response.status() != reqwest::StatusCode::OK { + return Ok(None); + } + let metadata: SnapshotMetadataFile = response + .error_for_status()? + .json() + .await + .with_context(|| format!("failed to parse snapshot metadata at {metadata_url}"))?; + Ok(Some(metadata.snapshot)) +} + +/// Downloads the extended snapshot `file`, published next to `snapshot_url`, into `temp_dir` +/// and verifies it against the SHA-256 from the snapshot metadata. +async fn download_extended_snapshot( + snapshot_url: &Url, + file: &ExtendedSnapshotFile, + temp_dir: &Path, +) -> anyhow::Result { + let filename_re = Regex::new(r"^[A-Za-z0-9_.-]+\.forest\.car\.zst$")?; + anyhow::ensure!( + filename_re.is_match(&file.filename), + "invalid extended snapshot filename: {}", + file.filename + ); + let mut url = snapshot_url.join(&file.filename)?; + url.set_query(snapshot_url.query()); + + let path = new_forest_car_temp_path_in(temp_dir)?; + info!("Downloading extended snapshot: {url}"); + download_to_with_timeout( + &url, + &path, + DownloadFileOption::Resumable, + Duration::from_mins(10), + None, + ) + .await + .with_context(|| format!("failed to download {}", file.filename))?; + + let path_for_hash = path.to_path_buf(); + let actual: [u8; 32] = tokio::task::spawn_blocking(move || { + crate::utils::hash::digest_file::(&path_for_hash) + }) + .await + .with_context(|| format!("failed to hash {}", file.filename))? + .with_context(|| format!("failed to read {} for hashing", file.filename))? + .into(); + anyhow::ensure!( + actual == file.sha256, + "checksum mismatch for {}", + file.filename + ); + Ok(path) +} + +/// Imports message receipts and events from a receipts and events snapshot into `db`. +async fn import_receipts_events_snapshot(db: &DB, path: &Path) -> anyhow::Result<()> +where + DB: Blockstore, +{ + let started = Instant::now(); + import_car_blocks(db, path, true, || {}).await?; + info!( + "Imported message receipts and events in {:.1}s", + started.elapsed().as_secs_f64() + ); + Ok(()) +} + +/// Indexes the tipsets whose receipts and events were just imported (block logs blooms and +/// Ethereum mappings), as `forest-tool index backfill` does. +async fn backfill_receipts_events( + state_manager: &StateManager, + head: &Tipset, + depth: ChainEpochDelta, +) -> anyhow::Result { + let from = Tipset::load_required(state_manager.db(), head.parents())?; + // the lowest tipset that can be indexed is at `head - depth + 1` + let to = head.epoch() - depth.clamp(0, head.epoch()) + 1; + let started = Instant::now(); + let guard = BackfillGuard::try_start()?; + let options = BackfillOptions { + allow_recompute: false, + ..Default::default() + }; + let result = run_backfill(state_manager, &from, RangeSpec::To(to), options, &guard).await; + let report = guard.finish(result)?; + + // index ethereum events from the head + let chain_store = state_manager.chain_store(); + let head_delegated_messages = + chain_store.headers_delegated_messages(head.block_headers().iter())?; + chain_store.process_signed_messages(&head_delegated_messages, true)?; + if report.skipped > 0 { + warn!( + "Indexed {} tipsets from epoch {} to {to}, {} could not be indexed", + report.indexed, + from.epoch(), + report.skipped + ); + } else { + info!( + "Indexed {} tipsets from epoch {} to {to} in {:.1}s", + report.indexed, + from.epoch(), + started.elapsed().as_secs_f64() + ); + } + Ok(report) +} + +/// 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 the +/// database of `state_manager`. +pub(super) async fn maybe_import_extended_snapshots( + state_manager: &StateManager, + head: &Tipset, + snapshot_url: &Url, + temp_dir: &Path, +) { + let metadata = match fetch_metadata_snapshots_info(snapshot_url).await { + Ok(Some(metadata)) => metadata, + Ok(None) => { + warn!("No snapshot metadata at {snapshot_url}, skipping extended snapshots"); + return; + } + Err(e) => { + warn!( + "Failed to fetch extended snapshots for {snapshot_url}: {e:#}; continuing without \ + receipts and events, import them manually, see the enriched snapshots docs" + ); + return; + } + }; + let Some(extended) = metadata.extended else { + info!("No extended snapshots advertised for {snapshot_url}"); + return; + }; + + let receipts_events_imported = match &extended.receipts_events { + Some(file) => { + let result = async { + let path = download_extended_snapshot(snapshot_url, file, temp_dir).await?; + import_receipts_events_snapshot(state_manager.db(), &path).await + } + .await; + if let Err(e) = &result { + warn!( + "Failed to import receipts and events snapshot {}: {e:#}; continuing without \ + them, import them manually, see the enriched snapshots docs", + file.filename + ); + } + result.is_ok() + } + None => false, + }; + + if let Some(file) = &extended.tipset_lookup { + let result = async { + let path = download_extended_snapshot(snapshot_url, file, temp_dir).await?; + let db = state_manager.db_owned(); + tokio::task::spawn_blocking(move || { + let mut progress = Progress::new("Importing tipset lookup entries"); + import_tipset_lookup_snapshot(&db, &path, || progress.inc(1)) + }) + .await? + } + .await; + if let Err(e) = result { + warn!( + "Failed to import tipset lookup snapshot {}: {e:#}; continuing without the tipset \ + lookup table, import it manually, see the enriched snapshots docs", + file.filename + ); + } + } + + // Index once both snapshots are imported, so each import's logs stay together. + if receipts_events_imported + && let Err(e) = backfill_receipts_events(state_manager, head, metadata.state_roots).await + { + warn!( + "Failed to index imported receipts and events: {e:#}; continuing without the bloom \ + index and Ethereum mappings, run forest-tool index backfill, see the enriched \ + snapshots docs" + ); + } +} + +#[cfg(test)] +mod test { + use super::*; + use crate::blocks::{CachingBlockHeader, RawBlockHeader}; + use crate::chain::index::tests::{genesis_tipset, persist_tipset, tipset_child}; + use crate::chain_sync::TipsetValidator; + use crate::rpc::eth::eth_tx_from_signed_eth_message; + use crate::test_utils::{construct_eth_messages, dummy_ticket}; + use crate::utils::db::CborStoreExt as _; + use crate::utils::db::car_stream::CarBlock; + use crate::utils::encoding::hex; + use crate::utils::multihash::prelude::*; + use fvm_ipld_encoding::DAG_CBOR; + use nunny::vec as nonempty; + use rstest::rstest; + + // The default fixture: a head with a single parent. + fn state_manager_with_delegated_head() -> ( + StateManager, + Tipset, + crate::message::SignedMessage, + crate::rpc::eth::types::EthHash, + ) { + state_manager_with_delegated_head_chain(1) + } + + // Builds a chain of `parents_below` tipsets ending in a head that carries a single delegated + // message, and returns the state manager, the head, the message and the message's Ethereum + // hash. + fn state_manager_with_delegated_head_chain( + parents_below: ChainEpoch, + ) -> ( + StateManager, + Tipset, + crate::message::SignedMessage, + crate::rpc::eth::types::EthHash, + ) { + let sm = crate::daemon::db_util::test_state_manager(); + let db = sm.db(); + + // A delegated message carried by the head block. + let (_, smsg) = construct_eth_messages(0); + db.put_cbor_default(&smsg).unwrap(); + + // The fixture uses mainnet, which forks to Hygge at 2_683_348; stay above it so + // `headers_delegated_messages` does not filter the head out. + let head_epoch = 2_700_000; + let mut parent = tipset_child(&genesis_tipset(), head_epoch - parents_below); + persist_tipset(&parent, db); + for epoch in (head_epoch - parents_below + 1)..head_epoch { + parent = tipset_child(&parent, epoch); + persist_tipset(&parent, db); + } + let head = Tipset::from(CachingBlockHeader::new(RawBlockHeader { + parents: parent.key().clone(), + epoch: head_epoch, + ticket: dummy_ticket(1), + messages: TipsetValidator::compute_msg_root( + db, + std::iter::empty(), + [smsg.signed_cid()], + ) + .unwrap(), + ..Default::default() + })); + persist_tipset(&head, db); + + let (_, tx) = + eth_tx_from_signed_eth_message(&smsg, sm.chain_store().chain_config().eth_chain_id) + .unwrap(); + let hash = tx.eth_hash().unwrap().into(); + (sm, head, smsg, hash) + } + + // The walk covers the head's parent down to `head - depth + 1`, both inclusive. The fixture + // tipsets carry no executed state, so each one is skipped and the counters give the range. + // The head is the one tipset the walk cannot reach, so its delegated message is indexed + // separately. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn backfill_receipts_events_indexes_range_and_head() { + let (sm, head, smsg, hash) = state_manager_with_delegated_head_chain(3); + + let report = backfill_receipts_events(&sm, &head, 3).await.unwrap(); + + assert_eq!(report.skipped, 2); + assert_eq!(report.indexed, 0); + assert_eq!( + sm.chain_store().get_mapping(&hash).unwrap(), + Some(smsg.cid()) + ); + } + + // A receipts and events CAR with one valid block. The orchestration tests only need the import + // to succeed; the fixture's tipsets carry no executed state, so the walk indexes no tipset + // keys and the head mapping is the observable effect of the backfill. + async fn receipts_events_car() -> Vec { + let data = b"receipts and events".to_vec(); + let cid = Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(&data)); + let block = CarBlock { + cid, + data: data.into(), + }; + let frames = + crate::db::car::forest::Encoder::compress_stream_default(futures::stream::iter([ + Ok::<_, anyhow::Error>(block), + ])); + let mut writer = vec![]; + crate::db::car::forest::Encoder::write(&mut writer, nonempty![cid], frames) + .await + .unwrap(); + writer + } + + fn extended_metadata_json(state_roots: ChainEpoch, sha256_hex: &str) -> String { + format!( + r#"{{ + "Snapshot": {{ + "State-roots": {state_roots}, + "Extended": {{ + "Augmented": {{ + "Filename": + "receipts.forest.car.zst", + "Sha256": "{sha256_hex}" + }} + }} + }} + }}"# + ) + } + + // Serves `.metadata.json` and the receipts companion next to `snapshot_url`. Both + // handlers require the `token=x` query, so a dropped query string fails the import. + async fn serve_extended_snapshot(metadata: String, car: Vec) -> Url { + use axum::response::IntoResponse as _; + + let app = axum::Router::new() + .route( + "/base.forest.car.zst.metadata.json", + axum::routing::get(move |uri: axum::http::Uri| { + let metadata = metadata.clone(); + async move { + if uri.query() != Some("token=x") { + return (axum::http::StatusCode::FORBIDDEN, String::new()) + .into_response(); + } + (axum::http::StatusCode::OK, metadata).into_response() + } + }), + ) + .route( + "/receipts.forest.car.zst", + axum::routing::get(move |uri: axum::http::Uri| { + let car = car.clone(); + async move { + if uri.query() != Some("token=x") { + return (axum::http::StatusCode::FORBIDDEN, Vec::new()).into_response(); + } + (axum::http::StatusCode::OK, car).into_response() + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + Url::parse(&format!("http://{addr}/base.forest.car.zst?token=x")).unwrap() + } + + // The whole orchestration: metadata lookup and companion download keep the snapshot's query + // string, the receipts import gates the backfill, and the head tipset's delegated message + // ends up indexed. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn maybe_import_extended_snapshots_indexes_head() { + use sha2::{Digest as _, Sha256}; + + let (sm, head, smsg, hash) = state_manager_with_delegated_head(); + let car = receipts_events_car().await; + let metadata = extended_metadata_json(1, &hex::encode(Sha256::digest(&car))); + let snapshot_url = serve_extended_snapshot(metadata, car).await; + + let temp_dir = tempfile::tempdir().unwrap(); + maybe_import_extended_snapshots(&sm, &head, &snapshot_url, temp_dir.path()).await; + + assert_eq!( + sm.chain_store().get_mapping(&hash).unwrap(), + Some(smsg.cid()) + ); + } + + // A failed receipts import must skip the backfill, so the head's mapping is not written. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn maybe_import_extended_snapshots_skips_backfill_without_receipts() { + let (sm, head, _, hash) = state_manager_with_delegated_head(); + let car = receipts_events_car().await; + // A wrong checksum makes the companion import fail. + let metadata = extended_metadata_json(1, &hex::encode([0u8; 32])); + let snapshot_url = serve_extended_snapshot(metadata, car).await; + + let temp_dir = tempfile::tempdir().unwrap(); + maybe_import_extended_snapshots(&sm, &head, &snapshot_url, temp_dir.path()).await; + + assert_eq!(sm.chain_store().get_mapping(&hash).unwrap(), None); + } + + #[rstest] + #[case::ok("companion.forest.car.zst", true, true)] + #[case::checksum_mismatch("companion.forest.car.zst", false, false)] + #[case::path_traversal("../companion.forest.car.zst", true, false)] + #[tokio::test] + async fn download_extended_snapshot_verifies( + #[case] filename: &str, + #[case] matching_checksum: bool, + #[case] succeeds: bool, + ) { + use sha2::{Digest as _, Sha256}; + let dummy_content: &'static [u8] = b"companion snapshot bytes"; + + let app = axum::Router::new().route( + "/companion.forest.car.zst", + axum::routing::get(move || async move { dummy_content }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + + let snapshot_url = Url::parse(&format!("http://{addr}/base.forest.car.zst")).unwrap(); + let file = ExtendedSnapshotFile { + filename: filename.into(), + sha256: if matching_checksum { + Sha256::digest(dummy_content).into() + } else { + [0; 32] + }, + }; + let temp_dir = tempfile::tempdir().unwrap(); + let result = download_extended_snapshot(&snapshot_url, &file, temp_dir.path()).await; + assert_eq!(result.is_ok(), succeeds, "{result:?}"); + if let Ok(path) = result { + assert_eq!(std::fs::read(path).unwrap(), dummy_content); + } + } + + // A stalled companion download must surface as a timeout instead of blocking startup. + #[tokio::test(start_paused = true)] + async fn download_extended_snapshot_times_out() { + let app = axum::Router::new().route( + "/companion.forest.car.zst", + axum::routing::get(|| async { std::future::pending::<()>().await }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + + let snapshot_url = Url::parse(&format!("http://{addr}/base.forest.car.zst")).unwrap(); + let file = ExtendedSnapshotFile { + filename: "companion.forest.car.zst".into(), + sha256: [0; 32], + }; + let temp_dir = tempfile::tempdir().unwrap(); + let err = download_extended_snapshot(&snapshot_url, &file, temp_dir.path()) + .await + .unwrap_err(); + assert_eq!( + err.downcast_ref(), + Some(&crate::utils::RetryError::TimeoutExceeded) + ); + } + + #[test] + fn parse_snapshot_metadata_with_extended() { + let json = r#"{ + "Snapshot": { + "Snapshot version": "2", + "Epoch": 4137600, + "State-roots": 2000, + "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(); + assert_eq!(metadata.snapshot.state_roots, 2000); + 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!( + hex::encode(extended.tipset_lookup.unwrap().sha256), + "710262581a4fdea6a8cd162e4a65b42727475b83c44b0b0fe05aff72ce70c2f2" + ); + } + + #[test] + fn parse_snapshot_metadata_without_extended() { + let json = + r#"{ "Snapshot": { "Snapshot version": "2", "Epoch": 4137600, "State-roots": 2000 } }"#; + let metadata: SnapshotMetadataFile = serde_json::from_str(json).unwrap(); + assert!(metadata.snapshot.extended.is_none()); + } + + // A host without a `.metadata.json` (a 404) is normal, not a warning. Any other metadata + // response is still parsed, and a real failure stays an error. + #[tokio::test] + async fn fetch_metadata_snapshots_info_missing_is_none() { + let app = axum::Router::new() + .route( + "/base.forest.car.zst.metadata.json", + axum::routing::get(|| async { r#"{ "Snapshot": { "State-roots": 2000 } }"# }), + ) + .route( + "/missing.forest.car.zst.metadata.json", + axum::routing::get(|| async { axum::http::StatusCode::NOT_FOUND }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + + let present = Url::parse(&format!("http://{addr}/base.forest.car.zst")).unwrap(); + let metadata = fetch_metadata_snapshots_info(&present) + .await + .unwrap() + .unwrap(); + assert_eq!(metadata.state_roots, 2000); + + let missing = Url::parse(&format!("http://{addr}/missing.forest.car.zst")).unwrap(); + assert!( + fetch_metadata_snapshots_info(&missing) + .await + .unwrap() + .is_none() + ); + } +} diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 82064cfc0f1..6d39e9355f7 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -4,6 +4,7 @@ pub mod bundle; mod context; pub mod db_util; +mod extended_snapshot; pub mod main; use crate::blocks::TipsetKey; @@ -16,7 +17,11 @@ 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, path_as_url}, + extended_snapshot::maybe_import_extended_snapshots, +}; use crate::db::gc::SnapshotGarbageCollector; use crate::db::ttl::EthMappingCollector; use crate::libp2p::{Libp2pService, PeerManager}; @@ -38,7 +43,7 @@ use anyhow::{Context as _, bail}; use backon::{ExponentialBuilder, Retryable}; use dialoguer::theme::ColorfulTheme; use futures::{Future, FutureExt}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::OnceLock; use std::time::{Duration, Instant}; @@ -143,6 +148,16 @@ async fn maybe_import_snapshot( if !opts.skip_load.unwrap_or_default() && let Some(path) = &config.client.snapshot_path { + // Resolve a moving alias once so the base and its extended snapshots come from the same + // snapshot. + let url = match path_as_url(path) { + Some(url) => Some(snapshot::peek_url(url).await), + None => None, + }; + let path = &match &url { + Some(url) => PathBuf::from(url.as_str()), + None => path.clone(), + }; let (car_db_path, ts) = import_chain_as_forest_car( path, &ctx.db_meta_data.get_forest_car_db_dir(), @@ -158,11 +173,22 @@ async fn maybe_import_snapshot( let ts_epoch = ts.epoch(); // Explicitly set heaviest tipset here in case HEAD_KEY has already been set // in the current setting store - ctx.state_manager.chain_store().set_heaviest_tipset(ts)?; + ctx.state_manager + .chain_store() + .set_heaviest_tipset(ts.shallow_clone())?; debug!( "Loaded car DB at {} and set current head to epoch {ts_epoch}", car_db_path.display(), ); + if let Some(url) = &url { + maybe_import_extended_snapshots( + &ctx.state_manager, + &ts, + url, + &ctx.db_meta_data.get_forest_car_db_dir(), + ) + .await; + } } // If the snapshot progress state is not completed, diff --git a/src/db/blockstore_with_write_buffer.rs b/src/db/blockstore_with_write_buffer.rs index afb4b99acf6..7c3585e7a94 100644 --- a/src/db/blockstore_with_write_buffer.rs +++ b/src/db/blockstore_with_write_buffer.rs @@ -41,7 +41,9 @@ impl BlockstoreWithWriteBuffer { } } - fn flush_buffer(&self) -> anyhow::Result<()> { + /// Writes the buffered blocks to the inner store. Dropping the buffer also flushes, but only + /// logs a failure, so callers that must know the blocks were written call this first. + pub fn flush(&self) -> anyhow::Result<()> { let records = { let mut buffer = self.buffer.write(); buffer.drain().collect_vec() @@ -51,7 +53,7 @@ impl BlockstoreWithWriteBuffer { fn flush_buffer_if_needed(&self) -> anyhow::Result<()> { if self.buffer.read().len() >= self.buffer_capacity { - self.flush_buffer() + self.flush() } else { Ok(()) } @@ -60,7 +62,7 @@ impl BlockstoreWithWriteBuffer { impl Drop for BlockstoreWithWriteBuffer { fn drop(&mut self) { - if let Err(e) = self.flush_buffer() { + if let Err(e) = self.flush() { tracing::warn!("{e:#}"); } } diff --git a/src/tool/subcommands/db_cmd.rs b/src/tool/subcommands/db_cmd.rs index b2e10d98198..a208ff37db5 100644 --- a/src/tool/subcommands/db_cmd.rs +++ b/src/tool/subcommands/db_cmd.rs @@ -4,21 +4,18 @@ use std::path::PathBuf; use std::time::Instant; -use crate::blocks::{Tipset, TipsetKey}; -use crate::chain::TIPSET_LOOKUP_HAMT_BIT_WIDTH; +use crate::chain::import_tipset_lookup_snapshot; use crate::cli::subcommands::prompt_confirm; use crate::cli_shared::{chain_path, delete_chain_data, read_config}; use crate::daemon::db_util::load_all_forest_cars; -use crate::db::car::{AnyCar, ManyCar}; +use crate::db::CAR_DB_DIR_NAME; +use crate::db::car::ManyCar; use crate::db::db_engine::{db_root, open_db}; -use crate::db::{BlockstoreWithWriteBuffer, CAR_DB_DIR_NAME, EthMappingsStore}; use crate::networks::NetworkChain; use crate::prelude::*; -use crate::utils::db::car_stream::CarStream; +use crate::utils::db::car_util::import_car_blocks; use clap::Subcommand; -use fil_actors_shared::fvm_ipld_hamt::Hamt; use indicatif::{ProgressBar, ProgressStyle}; -use tokio_stream::StreamExt; #[derive(Debug, Subcommand)] pub enum DBCommands { @@ -123,8 +120,6 @@ impl DBCommands { db, skip_validation: no_validation, } => { - const DB_WRITE_BUFFER_CAPACITY: usize = 10000; - let db_root_path = if let Some(db) = db { db } else { @@ -132,10 +127,7 @@ impl DBCommands { db_root(&chain_path(&config))? }; println!("Opening parity-db at {}", db_root_path.display()); - let db_writer = BlockstoreWithWriteBuffer::new_with_capacity( - open_db(db_root_path, &Default::default())?, - DB_WRITE_BUFFER_CAPACITY, - ); + let db = open_db(db_root_path, &Default::default())?; let pb = ProgressBar::new_spinner().with_style( ProgressStyle::with_template("{spinner} {msg}") @@ -145,18 +137,12 @@ impl DBCommands { let mut total = 0; for snap in snapshot_files { - let mut car = CarStream::new_from_path(&snap).await?; - while let Some(b) = car.try_next().await? { - if !no_validation { - b.validate()?; - } - db_writer.put_keyed(&b.cid, &b.data)?; + import_car_blocks(&db, &snap, !no_validation, || { total += 1; - let text = format!("{total} blocks imported"); - pb.set_message(text); - } + pb.set_message(format!("{total} blocks imported")); + }) + .await?; } - drop(db_writer); pb.finish(); Ok(()) } @@ -166,17 +152,6 @@ impl DBCommands { db, } => { let start = Instant::now(); - println!("Loading tipset lookup hamt..."); - let hamt_car = AnyCar::try_from(snapshot.as_path())?; - anyhow::ensure!( - hamt_car.header_v1().roots.len() == 1, - "hamt car should have a single root" - ); - let hamt_root = *hamt_car.header_v1().roots.first(); - let hamt: Hamt<_, TipsetKey, ChainEpoch> = - Hamt::load_with_bit_width(&hamt_root, hamt_car, TIPSET_LOOKUP_HAMT_BIT_WIDTH)?; - println!("Loaded tipset lookup hamt"); - let db_root_path = if let Some(db) = db { db } else { @@ -190,27 +165,13 @@ impl DBCommands { load_all_forest_cars(&db, &forest_car_db_dir)?; db }; - println!("Validating tipset lookup hamt..."); - hamt.for_each_cacheless(|&epoch, tsk| { - let ts = Tipset::load_required(&db, tsk)?; - anyhow::ensure!( - ts.epoch() == epoch, - "epochs do not match, {epoch} in hamt, {} in database", - ts.epoch() - ); - anyhow::Ok(()) - })?; - println!("Successfully validated tipset lookup hamt"); + println!("Validating and importing tipset lookup hamt..."); let pb = ProgressBar::new_spinner().with_style( ProgressStyle::with_template("{spinner} {pos} entries imported") .expect("indicatif template must be valid"), ); pb.enable_steady_tick(std::time::Duration::from_millis(100)); - hamt.for_each_cacheless(|&epoch, tsk| { - db.set_tipset_key_at_epoch_raw(epoch, tsk)?; - pb.inc(1); - anyhow::Ok(()) - })?; + import_tipset_lookup_snapshot(&db, &snapshot, || pb.inc(1))?; drop(db); let total = pb.position(); pb.finish(); diff --git a/src/tool/subcommands/snapshot_cmd.rs b/src/tool/subcommands/snapshot_cmd.rs index 1e66fc51a4f..871efc806e1 100644 --- a/src/tool/subcommands/snapshot_cmd.rs +++ b/src/tool/subcommands/snapshot_cmd.rs @@ -9,6 +9,7 @@ use crate::blocks::{Tipset, TipsetKey}; use crate::chain::{ TIPSET_LOOKUP_HAMT_BIT_WIDTH, index::{ChainIndex, ResolveNullTipset}, + validate_tipset_lookup_entry, }; use crate::cid_collections::CidHashSet; use crate::cli_shared::snapshot; @@ -386,14 +387,9 @@ fn validate_tipset_lookup_hamt( { if let Some(tsk) = hamt.get(&checkpoint)? { n_checkpoints += 1; - let ts = Tipset::load_required(&store, tsk).with_context(|| { + let ts = validate_tipset_lookup_entry(&store, checkpoint, tsk).with_context(|| { format!("failed to lookup tipset at checkpoint, epoch: {checkpoint}, key: {tsk}") })?; - anyhow::ensure!( - ts.epoch() == checkpoint, - "tipset mismatch, checkpoint epoch: {checkpoint}, tipset epoch: {}, tipset key: {tsk}", - ts.epoch() - ); null_checkpoint_search_from_ts = ts; } else { n_null_checkpoints += 1; diff --git a/src/tool/subcommands/snapshot_cmd/tests.rs b/src/tool/subcommands/snapshot_cmd/tests.rs index 016af522afe..7d0720d6d97 100644 --- a/src/tool/subcommands/snapshot_cmd/tests.rs +++ b/src/tool/subcommands/snapshot_cmd/tests.rs @@ -18,12 +18,12 @@ enum HamtEntries { #[case::bad_checkpoint_entry( HamtEntries::Checkpoints, Some(40), - Some("tipset mismatch, checkpoint epoch: 40") + Some("epochs do not match, 40 in hamt") )] #[case::null_checkpoint_entry( HamtEntries::Checkpoints, Some(20), - Some("tipset mismatch, checkpoint epoch: 20") + Some("epochs do not match, 20 in hamt") )] fn test_validate_tipset_lookup_hamt( #[case] entries: HamtEntries, diff --git a/src/utils/db/car_util.rs b/src/utils/db/car_util.rs index 7b7002c9a93..9f53cef918f 100644 --- a/src/utils/db/car_util.rs +++ b/src/utils/db/car_util.rs @@ -1,11 +1,14 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +use std::path::Path; + use crate::prelude::*; use futures::{Stream, StreamExt as _, TryStreamExt as _}; use tokio::io::{AsyncBufRead, AsyncSeek, BufReader}; use crate::cid_collections::CidHashSet; +use crate::db::BlockstoreWithWriteBuffer; use crate::utils::db::car_stream::{CarBlock, CarStream, CarV1Header}; /// Stream key-value pairs from a CAR archive into a block store. @@ -21,6 +24,26 @@ where Ok(stream.header_v1) } +/// Streams the blocks of the CAR archive at `path` into `db` through a write buffer, validating +/// each block when `validate` is set and calling `on_block` after each block is buffered. +pub async fn import_car_blocks( + db: &impl Blockstore, + path: &Path, + validate: bool, + mut on_block: impl FnMut(), +) -> anyhow::Result<()> { + let db = BlockstoreWithWriteBuffer::new_with_capacity(db, 10_000); + let mut car_stream = CarStream::new_from_path(path).await?; + while let Some(block) = car_stream.try_next().await? { + if validate { + block.validate()?; + } + db.put_keyed(&block.cid, &block.data)?; + on_block(); + } + db.flush() +} + pub fn merge_car_streams( car_streams: Vec>, ) -> impl Stream> @@ -196,4 +219,52 @@ mod tests { Ok::<_, anyhow::Error>(()) }) } + + #[tokio::test] + async fn import_car_blocks_writes_blocks_and_reports_each() { + use crate::db::MemoryDB; + + let blocks = Blocks(vec![ + CarBlock { + cid: Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(b"one")), + data: b"one".to_vec().into(), + }, + CarBlock { + cid: Cid::new_v1(DAG_CBOR, MultihashCode::Blake2b256.digest(b"two")), + data: b"two".to_vec().into(), + }, + ]); + let car = blocks.clone().into_forest_car_zst_bytes().await; + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("blocks.forest.car.zst"); + std::fs::write(&path, car).unwrap(); + + let db = MemoryDB::default(); + let mut imported = 0; + import_car_blocks(&db, &path, true, || imported += 1) + .await + .unwrap(); + + assert_eq!(imported, 2); + for block in &blocks.0 { + assert_eq!(db.get(&block.cid).unwrap(), Some(block.data.to_vec())); + } + + // The trailing buffer is flushed before returning, so a failed write is an error. + import_car_blocks(&FailingBlockstore, &path, true, || {}) + .await + .unwrap_err(); + } + + struct FailingBlockstore; + + impl Blockstore for FailingBlockstore { + fn get(&self, _k: &Cid) -> anyhow::Result>> { + Ok(None) + } + + fn put_keyed(&self, _k: &Cid, _block: &[u8]) -> anyhow::Result<()> { + anyhow::bail!("write failed") + } + } } diff --git a/src/utils/io/progress_log.rs b/src/utils/io/progress_log.rs index 0b65864ec49..a7dbaf8e3f0 100644 --- a/src/utils/io/progress_log.rs +++ b/src/utils/io/progress_log.rs @@ -111,7 +111,7 @@ enum ItemType { } impl Progress { - fn new(message: &str) -> Self { + pub(crate) fn new(message: &str) -> Self { let now = Instant::now(); Self { completed_items: 0, @@ -135,7 +135,7 @@ impl Progress { self } - fn inc(&mut self, value: u64) { + pub(crate) fn inc(&mut self, value: u64) { self.completed_items += value; self.emit_log_if_required(); diff --git a/src/utils/net/download_file.rs b/src/utils/net/download_file.rs index 75414881a83..0a07d72fa56 100644 --- a/src/utils/net/download_file.rs +++ b/src/utils/net/download_file.rs @@ -91,6 +91,22 @@ fn call_progress_callback( } } +fn log_download_progress(downloaded: u64, total_size: u64, bytes_per_sec: f64, elapsed: Duration) { + let percent = downloaded + .checked_mul(100) + .and_then(|v| v.checked_div(total_size)) + .unwrap_or(0); + tracing::info!( + target: "forest::progress", + "Loading {} / {}, {}%, {}/s, elapsed time: {}", + downloaded.human_count_bytes(), + total_size.human_count_bytes(), + percent, + bytes_per_sec.human_count_bytes(), + format_duration(Duration::from_secs(elapsed.as_secs())) + ); +} + #[derive(Debug, Copy, Clone)] pub enum DownloadFileOption { NonResumable, @@ -304,6 +320,8 @@ async fn download_http_parallel( // Store elapsed millis since start_time to avoid needing a Mutex. let last_logged_millis = Arc::new(std::sync::atomic::AtomicU64::new(0)); let start_time = Instant::now(); + // Log the start and the end too, so a download shorter than the update frequency is visible. + log_download_progress(0, total_size, 0.0, Duration::ZERO); const UPDATE_FREQUENCY: Duration = Duration::from_secs(5); const UPDATE_FREQUENCY_MS: u64 = UPDATE_FREQUENCY.as_millis() as u64; @@ -377,24 +395,14 @@ async fn download_http_parallel( .is_ok() { let last_bytes = last_logged_bytes.load(Ordering::Relaxed); - let elapsed_secs = elapsed_ms as f64 / 1000.0; let seconds_since_last = (elapsed_ms - prev_ms) as f64 / 1000.0; let speed = downloaded.saturating_sub(last_bytes) as f64 / seconds_since_last.max(0.1); - let percent = downloaded - .checked_mul(100) - .and_then(|v| v.checked_div(total_size)) - .unwrap_or(0); - tracing::info!( - target: "forest::progress", - "Loading {} / {}, {}%, {}/s, elapsed time: {}", - downloaded.human_count_bytes(), - total_size.human_count_bytes(), - percent, - speed.human_count_bytes(), - format_duration(Duration::from_secs( - elapsed_secs as u64 - )) + log_download_progress( + downloaded, + total_size, + speed, + Duration::from_millis(elapsed_ms), ); last_logged_bytes.store(downloaded, Ordering::Relaxed); @@ -440,6 +448,13 @@ async fn download_http_parallel( for (i, result) in results.into_iter().enumerate() { result.with_context(|| format!("Chunk {i} failed"))?; } + let elapsed = start_time.elapsed(); + log_download_progress( + total_size, + total_size, + total_size as f64 / elapsed.as_secs_f64().max(0.001), + elapsed, + ); // Rename to final destination tokio::fs::rename(&tmp_dst_path, &dst_path) @@ -552,26 +567,45 @@ pub async fn download_to( option: DownloadFileOption, callback: Option>, ) -> anyhow::Result<()> { - download_file_with_retry( - url, - destination.parent().with_context(|| { - format!( - "Error getting the parent directory of {}", - destination.display() - ) - })?, - destination - .file_name() - .and_then(OsStr::to_str) - .with_context(|| format!("Error getting the file name of {}", destination.display()))?, - option, - callback, + let (directory, filename) = destination_parts(destination)?; + download_file_with_retry(url, directory, filename, option, callback).await?; + Ok(()) +} + +/// Like [`download_to`], but with `timeout` argument, retries included. +pub async fn download_to_with_timeout( + url: &Url, + destination: &Path, + option: DownloadFileOption, + timeout: Duration, + callback: Option>, +) -> anyhow::Result<()> { + let (directory, filename) = destination_parts(destination)?; + retry( + RetryArgs { + timeout: Some(timeout), + ..Default::default() + }, + || download_http(url, directory, filename, option, callback.clone()), ) .await?; - Ok(()) } +fn destination_parts(destination: &Path) -> anyhow::Result<(&Path, &str)> { + let directory = destination.parent().with_context(|| { + format!( + "Error getting the parent directory of {}", + destination.display() + ) + })?; + let filename = destination + .file_name() + .and_then(OsStr::to_str) + .with_context(|| format!("Error getting the file name of {}", destination.display()))?; + Ok((directory, filename)) +} + #[cfg(test)] mod test { use super::*;