feat: add --unordered to forest-cli snapshot export

CHANGELOG.md

@@ -31,6 +31,8 @@
 
 - [#5859](https://github.com/ChainSafe/forest/pull/5859) Added size metrics for zstd frame cache and made max size configurable via `FOREST_ZSTD_FRAME_CACHE_DEFAULT_MAX_SIZE` environment variable.
 
+- [#5867](https://github.com/ChainSafe/forest/pull/5867) Added `--unordered` to `forest-cli snapshot export` for exporting `CAR` blocks in non-deterministic order for better performance with more parallelization.
+
 ### Changed
 
 - [#5869](https://github.com/ChainSafe/forest/pull/5869) Updated `forest-cli snapshot export` to print average speed.

src/blocks/header.rs

@@ -14,10 +14,12 @@ use crate::shim::{
     address::Address, crypto::Signature, econ::TokenAmount, sector::PoStProof,
     version::NetworkVersion,
 };
-use crate::utils::{cid::CidCborExt as _, encoding::blake2b_256};
+use crate::utils::encoding::blake2b_256;
+use crate::utils::multihash::MultihashCode;
 use cid::Cid;
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_encoding::CborStore as _;
+use multihash_derive::MultihashDigest as _;
 use num::BigInt;
 use serde::{Deserialize, Serialize};
 use serde_tuple::{Deserialize_tuple, Serialize_tuple};
@@ -96,7 +98,15 @@ impl Default for RawBlockHeader {
 
 impl RawBlockHeader {
     pub fn cid(&self) -> Cid {
-        Cid::from_cbor_blake2b256(self).unwrap()
+        self.car_block().expect("Infallible").0
+    }
+    pub fn car_block(&self) -> anyhow::Result<(Cid, Vec<u8>)> {
+        let data = fvm_ipld_encoding::to_vec(self)?;
+        let cid = Cid::new_v1(
+            fvm_ipld_encoding::DAG_CBOR,
+            MultihashCode::Blake2b256.digest(&data),
+        );
+        Ok((cid, data))
     }
     pub(super) fn tipset_sort_key(&self) -> Option<([u8; 32], Vec<u8>)> {
         let ticket_hash = blake2b_256(self.ticket.as_ref()?.vrfproof.as_bytes());

src/chain/mod.rs

@@ -6,7 +6,7 @@ use crate::blocks::{Tipset, TipsetKey};
 use crate::cid_collections::CidHashSet;
 use crate::db::car::forest;
 use crate::db::{SettingsStore, SettingsStoreExt};
-use crate::ipld::stream_chain;
+use crate::ipld::{stream_chain, unordered_stream_chain};
 use crate::utils::io::{AsyncWriterWithChecksum, Checksum};
 use crate::utils::stream::par_buffer;
 use anyhow::Context as _;
@@ -17,17 +17,23 @@ use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
 
 pub use self::{store::*, weight::*};
 
+#[derive(Debug, Clone, Default)]
+pub struct ExportOptions {
+    pub skip_checksum: bool,
+    pub unordered: bool,
+    pub seen: CidHashSet,
+}
+
 pub async fn export_from_head<D: Digest>(
     db: &Arc<impl Blockstore + SettingsStore + Send + Sync + 'static>,
     lookup_depth: ChainEpochDelta,
     writer: impl AsyncWrite + Unpin,
-    seen: CidHashSet,
-    skip_checksum: bool,
+    option: Option<ExportOptions>,
 ) -> anyhow::Result<(Tipset, Option<digest::Output<D>>), Error> {
     let head_key = SettingsStoreExt::read_obj::<TipsetKey>(db, crate::db::setting_keys::HEAD_KEY)?
         .context("chain head key not found")?;
     let head_ts = Tipset::load_required(&db, &head_key)?;
-    let digest = export::<D>(db, &head_ts, lookup_depth, writer, seen, skip_checksum).await?;
+    let digest = export::<D>(db, &head_ts, lookup_depth, writer, option).await?;
     Ok((head_ts, digest))
 }
 
@@ -36,9 +42,14 @@ pub async fn export<D: Digest>(
     tipset: &Tipset,
     lookup_depth: ChainEpochDelta,
     writer: impl AsyncWrite + Unpin,
-    seen: CidHashSet,
-    skip_checksum: bool,
+    option: Option<ExportOptions>,
 ) -> anyhow::Result<Option<digest::Output<D>>, Error> {
+    let ExportOptions {
+        skip_checksum,
+        unordered,
+        seen,
+    } = option.unwrap_or_default();
+
     let stateroot_lookup_limit = tipset.epoch() - lookup_depth;
     let roots = tipset.key().to_cids();
 
@@ -52,12 +63,25 @@ pub async fn export<D: Digest>(
         // are small enough that keeping 1k in memory isn't a problem. Average
         // block size is between 1kb and 2kb.
         1024,
-        stream_chain(
-            Arc::clone(db),
-            tipset.clone().chain_owned(Arc::clone(db)),
-            stateroot_lookup_limit,
-        )
-        .with_seen(seen),
+        if unordered {
+            futures::future::Either::Left(
+                unordered_stream_chain(
+                    Arc::clone(db),
+                    tipset.clone().chain_owned(Arc::clone(db)),
+                    stateroot_lookup_limit,
+                )
+                .with_seen(seen),
+            )
+        } else {
+            futures::future::Either::Right(
+                stream_chain(
+                    Arc::clone(db),
+                    tipset.clone().chain_owned(Arc::clone(db)),
+                    stateroot_lookup_limit,
+                )
+                .with_seen(seen),
+            )
+        },
     );
 
     // Encode Ipld key-value pairs in zstd frames

src/cli/subcommands/snapshot_cmd.rs

@@ -6,8 +6,7 @@ use crate::chain_sync::SyncConfig;
 use crate::cli_shared::snapshot::{self, TrustedVendor};
 use crate::db::car::forest::new_forest_car_temp_path_in;
 use crate::networks::calibnet;
-use crate::rpc::types::ApiTipsetKey;
-use crate::rpc::{self, chain::ChainExportParams, prelude::*};
+use crate::rpc::{self, chain::ForestChainExportParams, prelude::*, types::ApiTipsetKey};
 use anyhow::Context as _;
 use chrono::DateTime;
 use clap::Subcommand;
@@ -38,6 +37,9 @@ pub enum SnapshotCommands {
         /// How many state-roots to include. Lower limit is 900 for `calibnet` and `mainnet`.
         #[arg(short, long)]
         depth: Option<crate::chain::ChainEpochDelta>,
+        /// Traverse chain in non-deterministic order for better performance with more parallelization.
+        #[arg(long)]
+        unordered: bool,
     },
 }
 
@@ -50,6 +52,7 @@ impl SnapshotCommands {
                 dry_run,
                 tipset,
                 depth,
+                unordered,
             } => {
                 let chain_head = ChainHead::call(&client, ()).await?;
 
@@ -85,11 +88,12 @@ impl SnapshotCommands {
                 let output_dir = output_path.parent().context("invalid output path")?;
                 let temp_path = new_forest_car_temp_path_in(output_dir)?;
 
-                let params = ChainExportParams {
+                let params = ForestChainExportParams {
                     epoch,
                     recent_roots: depth.unwrap_or(SyncConfig::default().recent_state_roots),
                     output_path: temp_path.to_path_buf(),
                     tipset_keys: ApiTipsetKey(Some(chain_head.key().clone())),
+                    unordered,
                     skip_checksum,
                     dry_run,
                 };
@@ -131,16 +135,18 @@ impl SnapshotCommands {
                 // Manually construct RpcRequest because snapshot export could
                 // take a few hours on mainnet
                 let hash_result = client
-                    .call(ChainExport::request((params,))?.with_timeout(Duration::MAX))
+                    .call(ForestChainExport::request((params,))?.with_timeout(Duration::MAX))
                     .await?;
 
                 handle.abort();
                 let _ = handle.await;
 
-                if let Some(hash) = hash_result {
-                    save_checksum(&output_path, hash).await?;
+                if !dry_run {
+                    if let Some(hash) = hash_result {
+                        save_checksum(&output_path, hash).await?;
+                    }
+                    temp_path.persist(output_path)?;
                 }
-                temp_path.persist(output_path)?;
 
                 println!("Export completed.");
                 Ok(())

src/db/gc/snapshot.rs

@@ -43,7 +43,7 @@
 //!
 
 use crate::blocks::{Tipset, TipsetKey};
-use crate::cid_collections::CidHashSet;
+use crate::chain::ExportOptions;
 use crate::cli_shared::chain_path;
 use crate::db::car::forest::new_forest_car_temp_path_in;
 use crate::db::{
@@ -225,8 +225,10 @@ where
             &db,
             self.recent_state_roots,
             file,
-            CidHashSet::default(),
-            true,
+            Some(ExportOptions {
+                skip_checksum: true,
+                ..Default::default()
+            }),
         )
         .await?;
         let target_path = self.car_db_dir.join(format!(