feat: add --augmented-data and --tipset-lookup to forest-cli snapshot export

CHANGELOG.md

@@ -31,6 +31,8 @@
 
 ### Added
 
+- [#7376](https://github.com/ChainSafe/forest/pull/7376): Added `--augmented-data` and `--tipset-lookup` to `forest-cli snapshot export`.
+
 ### Changed
 
 ### Removed

docs/docs/users/reference/cli.sh

@@ -112,6 +112,7 @@ generate_markdown_section "forest-tool" "snapshot"
 generate_markdown_section "forest-tool" "snapshot fetch"
 generate_markdown_section "forest-tool" "snapshot validate-diffs"
 generate_markdown_section "forest-tool" "snapshot validate"
+generate_markdown_section "forest-tool" "snapshot validate-extended"
 generate_markdown_section "forest-tool" "snapshot compress"
 generate_markdown_section "forest-tool" "snapshot compute-state"
 

scripts/tests/calibnet_export_check.sh

@@ -61,7 +61,7 @@ echo "Exporting zstd compressed snapshot at genesis"
 $FOREST_CLI_PATH snapshot export --tipset 0 --format "$format"
 
 echo "Exporting zstd compressed snapshot in $format format"
-$FOREST_CLI_PATH snapshot export --format "$format" &
+$FOREST_CLI_PATH snapshot export --format "$format" --tipset-lookup -o snapshot.forest.car.zst &
 EXPORT_CMD_PID=$!
 sleep 5
 # another export job should be disallowed
@@ -91,6 +91,19 @@ $FOREST_CLI_PATH snapshot export-status --wait
 
 $FOREST_CLI_PATH shutdown --force
 
+# Check file sizes
+ls -ahl *.forest.car.zst
+
+echo "Verifying snapshot checksum"
+sha256sum --check ./*.sha256sum
+
+# Validate tipset lookup snapshots
+# export and check augmented data once we have receipts and events tipset published and imported
+# for now there's no receipts and events on a freshly bootstrapped node.
+$FOREST_TOOL_PATH snapshot validate-extended --base snapshot.forest.car.zst --tipset-lookup snapshot_tipset_lookup.forest.car.zst
+# Remove tipset lookup snapshots before proceeding
+rm *_tipset_lookup.forest.car.zst*
+
 for f in *.car.zst; do
   echo "Inspecting archive info $f"
   $FOREST_TOOL_PATH archive info "$f"
@@ -101,9 +114,6 @@ done
 echo "Testing snapshot validity"
 zstd --test ./*.car.zst
 
-echo "Verifying snapshot checksum"
-sha256sum --check ./*.sha256sum
-
 for f in *.car.zst; do
   echo "Validating CAR file $f"
   $FOREST_TOOL_PATH snapshot validate "$f"

src/chain/mod.rs

@@ -12,11 +12,12 @@ pub use self::{snapshot_format::*, store::*, weight::*};
 
 use crate::blocks::{Tipset, TipsetKey};
 use crate::chain::index::ChainIndex;
-use crate::cid_collections::CidHashSetLike;
+use crate::cid_collections::{CidHashSet, CidHashSetLike};
 use crate::db::IndexMapBlockstore;
 use crate::db::car::forest::{self, ForestCarFrame, finalize_frame};
-use crate::ipld::stream_chain;
+use crate::ipld::{IpldStream, stream_chain};
 use crate::prelude::*;
+use crate::shim::executor::Receipt;
 use crate::utils::db::car_stream::{CarBlock, CarBlockWrite};
 use crate::utils::io::{AsyncWriterWithChecksum, Checksum};
 use crate::utils::multihash::MultihashCode;
@@ -32,7 +33,7 @@ use std::time::Instant;
 use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
 use tokio_util::task::AbortOnDropHandle;
 
-const TIPSET_LOOKUP_HAMT_BIT_WIDTH: u32 = 5;
+pub const TIPSET_LOOKUP_HAMT_BIT_WIDTH: u32 = 5;
 
 pub struct ExportOptions<S> {
     pub skip_checksum: bool,
@@ -260,3 +261,78 @@ async fn export_to_forest_car<D: Digest, S: CidHashSetLike + Send + Sync + 'stat
         tipset_lookup,
     })
 }
+
+pub async fn export_receipts_events_to_forest_car(
+    db: &(impl Blockstore + ShallowClone + Unpin + Send + Sync + 'static),
+    tipset: &Tipset,
+    lookup_depth: ChainEpochDelta,
+    writer: impl AsyncWrite + Unpin,
+) -> anyhow::Result<()> {
+    let start = Instant::now();
+    tracing::info!(
+        "Exporting message receipts and events snapshot, epoch={}, depth={lookup_depth}",
+        tipset.epoch(),
+    );
+
+    let min_lookup_epoch_exclusive = tipset.epoch() - lookup_depth;
+    let ipld_roots = tokio::task::spawn_blocking({
+        let tipset = tipset.shallow_clone();
+        let db = db.shallow_clone();
+        move || {
+            let mut ipld_roots = vec![];
+            for ts in tipset
+                .chain(&db)
+                .take_while(|ts| ts.epoch() > min_lookup_epoch_exclusive)
+            {
+                let message_receipts_root = *ts.parent_message_receipts();
+                ipld_roots.push(message_receipts_root);
+                let receipts = Receipt::get_receipts(&db, message_receipts_root).with_context(|| {
+                    format!(
+                        "failed to get receipts, root: {message_receipts_root}, epoch: {}, tipset key: {}",
+                        ts.epoch(),
+                        ts.key(),
+                    )
+                })?;
+                ipld_roots.extend(receipts.into_iter().filter_map(|r| r.events_root()));
+            }
+            anyhow::Ok(ipld_roots)
+        }
+    })
+    .await??;
+
+    let stream = IpldStream::new(db.shallow_clone(), ipld_roots, CidHashSet::default());
+
+    let mut writer = BufWriter::new(writer);
+
+    // Stream message receipts and events in range (stateroot_lookup_limit+1)..=tipset.epoch().
+    let (blocks, _drop_guard) = par_buffer(
+        // Queue 1k blocks. This is enough to saturate the compressor and blocks
+        // are small enough that keeping 1k in memory isn't a problem. Average
+        // block size is between 1kb and 2kb.
+        1024, stream,
+    );
+
+    // Encode Ipld key-value pairs in zstd frames
+    let block_frames = forest::Encoder::compress_stream_default(blocks);
+
+    // There's no data root for this snapshot, use a default CID as placeholder.
+    // Note that the output CAR could only be validated with `forest-tool snapshot validate-extended`.
+    // Another option could be including chain spine in the CAR and use head tipset key as root,
+    // whose downside would be bloating the CAR.
+    let roots = nunny::vec![Cid::default()];
+
+    // Write zstd frames and include a skippable index
+    forest::Encoder::write(&mut writer, roots, block_frames).await?;
+
+    // Flush to ensure everything has been successfully written
+    tokio::time::timeout(forest::ASYNC_OPS_TIMEOUT, writer.flush())
+        .await
+        .context("`writer.flush` timed out")??;
+
+    tracing::info!(
+        "Exported message receipts and events snapshot, took {}",
+        humantime::format_duration(start.elapsed())
+    );
+
+    Ok(())
+}

src/chain/store/index.rs

@@ -443,7 +443,7 @@ impl ChainIndex {
 }
 
 #[cfg(test)]
-mod tests {
+pub mod tests {
     use super::*;
     use crate::blocks::{CachingBlockHeader, RawBlockHeader};
     use crate::chain::store::ChainStore;
@@ -457,20 +457,20 @@ mod tests {
         atomic::{AtomicU64, Ordering},
     };
 
-    fn persist_tipset(tipset: &Tipset, db: &impl Blockstore) {
+    pub fn persist_tipset(tipset: &Tipset, db: &impl Blockstore) {
         for block in tipset.block_headers() {
             db.put_cbor_default(block).unwrap();
         }
     }
 
-    fn genesis_tipset() -> Tipset {
+    pub fn genesis_tipset() -> Tipset {
         Tipset::from(CachingBlockHeader::new(RawBlockHeader {
             ticket: dummy_ticket(0),
             ..Default::default()
         }))
     }
 
-    fn tipset_child(parent: &Tipset, epoch: ChainEpoch) -> Tipset {
+    pub fn tipset_child(parent: &Tipset, epoch: ChainEpoch) -> Tipset {
         // Use a static counter to give all tipsets a unique timestamp
         static COUNTER: AtomicU64 = AtomicU64::new(0);
         let n = COUNTER.fetch_add(1, Ordering::Relaxed);

src/cli/subcommands/snapshot_cmd.rs

@@ -45,6 +45,12 @@ pub enum SnapshotCommands {
         /// Snapshot format to export.
         #[arg(long, value_enum, default_value_t = FilecoinSnapshotVersion::V2)]
         format: FilecoinSnapshotVersion,
+        /// Also exports an augmented data snapshot that contains message receipts and events
+        #[arg(long)]
+        augmented_data: bool,
+        /// Also exports a tipset lookup HAMT snapshot
+        #[arg(long)]
+        tipset_lookup: bool,
     },
     /// Show status of the current export.
     ExportStatus {
@@ -84,6 +90,8 @@ impl SnapshotCommands {
                 tipset,
                 depth,
                 format,
+                augmented_data: augmented_snapshot,
+                tipset_lookup,
             } => {
                 anyhow::ensure!(
                     depth >= 0,
@@ -141,7 +149,8 @@ impl SnapshotCommands {
                     include_receipts: false,
                     include_events: false,
                     include_tipset_keys: false,
-                    include_tipset_lookup: false,
+                    augmented_snapshot,
+                    tipset_lookup,
                     skip_checksum,
                     dry_run,
                 };

src/db/car/forest.rs

@@ -530,6 +530,21 @@ pub fn forest_car_sha256sum_path(output_path: &Path) -> PathBuf {
     p
 }
 
+pub fn forest_car_with_filename_suffix(path: &Path, suffix: &str) -> anyhow::Result<PathBuf> {
+    anyhow::ensure!(!suffix.is_empty(), "suffix cannot be empty");
+    let file_name = path.file_name().and_then(|n| n.to_str()).with_context(|| {
+        format!(
+            "failed to extract filename from the given path: {}",
+            path.display()
+        )
+    })?;
+    let new_name = match file_name.split_once('.') {
+        Some((stem, rest)) => format!("{stem}{suffix}.{rest}"),
+        None => format!("{file_name}{suffix}"),
+    };
+    Ok(path.with_file_name(new_name))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -656,11 +671,14 @@ mod tests {
 
     #[rstest]
     #[case(
-        Path::new("/tmp/a.forst.car.zst"),
-        Path::new("/tmp/a.forst.car.zst.tmp")
+        Path::new("/tmp/a.forest.car.zst"),
+        Path::new("/tmp/a.forest.car.zst.tmp")
+    )]
+    #[case(
+        Path::new("tmp/a.forest.car.zst"),
+        Path::new("tmp/a.forest.car.zst.tmp")
     )]
-    #[case(Path::new("tmp/a.forst.car.zst"), Path::new("tmp/a.forst.car.zst.tmp"))]
-    #[case(Path::new("a.forst.car.zst"), Path::new("a.forst.car.zst.tmp"))]
+    #[case(Path::new("a.forest.car.zst"), Path::new("a.forest.car.zst.tmp"))]
     #[case(Path::new(""), Path::new(""))]
     #[case(Path::new("."), Path::new("."))]
     fn test_tmp_exporting_forest_car_path(#[case] input: &Path, #[case] output: &Path) {
@@ -669,17 +687,52 @@ mod tests {
 
     #[rstest]
     #[case(
-        Path::new("/tmp/a.forst.car.zst"),
-        Path::new("/tmp/a.forst.car.zst.sha256sum")
+        Path::new("/tmp/a.forest.car.zst"),
+        Path::new("/tmp/a.forest.car.zst.sha256sum")
     )]
     #[case(
-        Path::new("tmp/a.forst.car.zst"),
-        Path::new("tmp/a.forst.car.zst.sha256sum")
+        Path::new("tmp/a.forest.car.zst"),
+        Path::new("tmp/a.forest.car.zst.sha256sum")
     )]
-    #[case(Path::new("a.forst.car.zst"), Path::new("a.forst.car.zst.sha256sum"))]
+    #[case(Path::new("a.forest.car.zst"), Path::new("a.forest.car.zst.sha256sum"))]
     #[case(Path::new(""), Path::new(""))]
     #[case(Path::new("."), Path::new("."))]
     fn test_forest_car_sha256sum_path(#[case] input: &Path, #[case] output: &Path) {
         assert_eq!(forest_car_sha256sum_path(input), output);
     }
+
+    #[rstest]
+    #[case(
+        Path::new("/tmp/a.forest.car.zst"),
+        "_suffix",
+        Path::new("/tmp/a_suffix.forest.car.zst")
+    )]
+    #[case(
+        Path::new("a.forest.car.zst"),
+        "_suffix",
+        Path::new("a_suffix.forest.car.zst")
+    )]
+    #[case(Path::new("a"), "_suffix", Path::new("a_suffix"))]
+    fn test_forest_car_with_filename_suffix_valid(
+        #[case] input: &Path,
+        #[case] suffix: &str,
+        #[case] output: &Path,
+    ) {
+        assert_eq!(
+            forest_car_with_filename_suffix(input, suffix).unwrap(),
+            output
+        );
+    }
+
+    #[rstest]
+    #[case(Path::new("/tmp/a.forest.car.zst"), "", "suffix cannot be empty")]
+    #[case(Path::new("."), "_suffix", "failed to extract filename")]
+    fn test_forest_car_with_filename_suffix_invalid(
+        #[case] input: &Path,
+        #[case] suffix: &str,
+        #[case] reason: &str,
+    ) {
+        let e = forest_car_with_filename_suffix(input, suffix).unwrap_err();
+        assert!(e.to_string().contains(reason));
+    }
 }

src/db/memory.rs

@@ -17,6 +17,7 @@ use ahash::HashMap;
 use indexmap::IndexMap;
 use nunny::Vec as NonEmpty;
 use parking_lot::RwLock;
+use tokio::io::AsyncWriteExt as _;
 
 #[derive(Debug, Default)]
 pub struct MemoryDB {
@@ -243,7 +244,6 @@ pub struct IndexMapBlockstore {
 }
 
 impl IndexMapBlockstore {
-    #[allow(dead_code)]
     pub async fn export_forest_car<W: tokio::io::AsyncWrite + Unpin>(
         &self,
         roots: NonEmpty<Cid>,
@@ -271,7 +271,9 @@ impl IndexMapBlockstore {
         };
         let frames =
             crate::db::car::forest::Encoder::compress_stream_default(futures::stream::iter(blocks));
-        crate::db::car::forest::Encoder::write(writer, roots, frames).await
+        crate::db::car::forest::Encoder::write(&mut *writer, roots, frames).await?;
+        writer.flush().await?;
+        Ok(())
     }
 }