feat(daemon): import enriched snapshots

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).

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}"

src/daemon/db_util.rs

@@ -1,23 +1,25 @@
 // 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;
 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<ExtendedSnapshots>,
+}
+
+/// 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<ExtendedSnapshotFile>,
+    /// Tipset lookup table (checkpoint epoch to tipset key)
+    #[serde(rename = "Tipset lookup", default)]
+    tipset_lookup: Option<ExtendedSnapshotFile>,
+}
+
+#[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<Option<ExtendedSnapshots>> {
+    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<tempfile::TempPath> {
+    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>(db: &Arc<DB>, path: &Path) -> anyhow::Result<usize>
+where
+    DB: Blockstore + Send + Sync + 'static,
+{
+    async fn flush<DB: Blockstore + Send + Sync + 'static>(
+        db: &Arc<DB>,
+        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>(
+    db: &Arc<DB>,
+    path: &Path,
+    head_epoch: ChainEpoch,
+) -> anyhow::Result<usize>
+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>(
+    db: &Arc<DB>,
+    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

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

src/db/car/many.rs

@@ -321,6 +321,13 @@ impl<WriterT: EthMappingsStore> EthMappingsStore for ManyCar<WriterT> {
     ) -> 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<WriterT: EthBlockBloomStore> EthBlockBloomStore for ManyCar<WriterT> {

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())