forest/chain/
mod.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4pub mod ec_finality;
5mod snapshot_format;
6pub mod store;
7#[cfg(test)]
8mod tests;
9mod weight;
10
11pub use self::{snapshot_format::*, store::*, weight::*};
12
13use crate::blocks::{Tipset, TipsetKey};
14use crate::chain::index::ChainIndex;
15use crate::cid_collections::CidHashSetLike;
16use crate::db::IndexMapBlockstore;
17use crate::db::car::forest::{self, ForestCarFrame, finalize_frame};
18use crate::ipld::stream_chain;
19use crate::prelude::*;
20use crate::utils::db::car_stream::{CarBlock, CarBlockWrite};
21use crate::utils::io::{AsyncWriterWithChecksum, Checksum};
22use crate::utils::multihash::MultihashCode;
23use crate::utils::stream::par_buffer;
24use fil_actors_shared::fvm_ipld_hamt::Hamt;
25use futures::StreamExt as _;
26use fvm_ipld_encoding::DAG_CBOR;
27use multihash_derive::MultihashDigest as _;
28use nunny::Vec as NonEmpty;
29use sha2::digest::{self, Digest};
30use std::io::{Read, Seek, SeekFrom};
31use std::time::Instant;
32use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
33use tokio_util::task::AbortOnDropHandle;
34
35const TIPSET_LOOKUP_HAMT_BIT_WIDTH: u32 = 5;
36
37pub struct ExportOptions<S> {
38    pub skip_checksum: bool,
39    pub include_receipts: bool,
40    pub include_events: bool,
41    pub include_tipset_keys: bool,
42    pub include_tipset_lookup: bool,
43    pub seen: S,
44}
45
46impl<S: Default> Default for ExportOptions<S> {
47    fn default() -> Self {
48        Self {
49            skip_checksum: Default::default(),
50            include_receipts: Default::default(),
51            include_events: Default::default(),
52            include_tipset_keys: Default::default(),
53            include_tipset_lookup: Default::default(),
54            seen: Default::default(),
55        }
56    }
57}
58
59pub struct ExportResult<D: Digest> {
60    pub checksum: Option<digest::Output<D>>,
61    #[allow(dead_code)]
62    pub tipset_lookup: Option<anyhow::Result<Hamt<IndexMapBlockstore, TipsetKey, ChainEpoch>>>,
63}
64
65/// Exports a Filecoin snapshot in v1 format
66/// See <https://github.com/filecoin-project/FIPs/blob/98e33b9fa306959aa0131519eb4cc155522b2081/FRCs/frc-0108.md#v1-specification>
67pub async fn export<D: Digest, S: CidHashSetLike + Send + Sync + 'static>(
68    db: &(impl Blockstore + ShallowClone + Unpin + Send + Sync + 'static),
69    tipset: &Tipset,
70    lookup_depth: ChainEpochDelta,
71    writer: impl AsyncWrite + Unpin,
72    options: ExportOptions<S>,
73) -> anyhow::Result<ExportResult<D>> {
74    let roots = tipset.key().to_cids();
75    export_to_forest_car::<D, S>(roots, None, db, tipset, lookup_depth, writer, options).await
76}
77
78/// Exports a Filecoin snapshot in v2 format
79/// See <https://github.com/filecoin-project/FIPs/blob/98e33b9fa306959aa0131519eb4cc155522b2081/FRCs/frc-0108.md#v2-specification>
80pub async fn export_v2<D: Digest, F: Seek + Read, S: CidHashSetLike + Send + Sync + 'static>(
81    db: &(impl Blockstore + ShallowClone + Unpin + Send + Sync + 'static),
82    mut f3: Option<(Cid, F)>,
83    tipset: &Tipset,
84    lookup_depth: ChainEpochDelta,
85    writer: impl AsyncWrite + Unpin,
86    options: ExportOptions<S>,
87) -> anyhow::Result<ExportResult<D>> {
88    // validate f3 data
89    if let Some((f3_cid, f3_data)) = &mut f3 {
90        f3_data.seek(SeekFrom::Start(0))?;
91        let expected_cid = crate::f3::snapshot::get_f3_snapshot_cid(f3_data)?;
92        anyhow::ensure!(
93            f3_cid == &expected_cid,
94            "f3 snapshot integrity check failed, actual cid: {f3_cid}, expected cid: {expected_cid}"
95        );
96    }
97
98    let head = tipset.key().to_cids();
99    let f3_cid = f3.as_ref().map(|(cid, _)| *cid);
100    let snap_meta = FilecoinSnapshotMetadata::new_v2(head, f3_cid);
101    let snap_meta_cbor_encoded = fvm_ipld_encoding::to_vec(&snap_meta)?;
102    let snap_meta_block = CarBlock {
103        cid: Cid::new_v1(
104            DAG_CBOR,
105            MultihashCode::Blake2b256.digest(&snap_meta_cbor_encoded),
106        ),
107        data: snap_meta_cbor_encoded.into(),
108    };
109    let roots = nunny::vec![snap_meta_block.cid];
110    let mut prefix_data_frames = vec![{
111        let mut encoder = forest::new_encoder(forest::DEFAULT_FOREST_CAR_COMPRESSION_LEVEL)?;
112        snap_meta_block.write(&mut encoder)?;
113        anyhow::Ok((
114            vec![snap_meta_block.cid],
115            finalize_frame(forest::DEFAULT_FOREST_CAR_COMPRESSION_LEVEL, &mut encoder)?,
116        ))
117    }];
118
119    if let Some((f3_cid, mut f3_data)) = f3 {
120        let f3_data_len = f3_data.seek(SeekFrom::End(0))?;
121        f3_data.seek(SeekFrom::Start(0))?;
122        prefix_data_frames.push({
123            let mut encoder = forest::new_encoder(forest::DEFAULT_FOREST_CAR_COMPRESSION_LEVEL)?;
124            encoder.write_car_block(f3_cid, f3_data_len, &mut f3_data)?;
125            anyhow::Ok((
126                vec![f3_cid],
127                finalize_frame(forest::DEFAULT_FOREST_CAR_COMPRESSION_LEVEL, &mut encoder)?,
128            ))
129        });
130    }
131
132    export_to_forest_car::<D, S>(
133        roots,
134        Some(prefix_data_frames),
135        db,
136        tipset,
137        lookup_depth,
138        writer,
139        options,
140    )
141    .await
142}
143
144#[allow(clippy::too_many_arguments)]
145async fn export_to_forest_car<D: Digest, S: CidHashSetLike + Send + Sync + 'static>(
146    roots: NonEmpty<Cid>,
147    prefix_data_frames: Option<Vec<anyhow::Result<ForestCarFrame>>>,
148    db: &(impl Blockstore + ShallowClone + Unpin + Send + Sync + 'static),
149    tipset: &Tipset,
150    lookup_depth: ChainEpochDelta,
151    writer: impl AsyncWrite + Unpin,
152    ExportOptions {
153        skip_checksum,
154        include_receipts,
155        include_events,
156        include_tipset_keys,
157        include_tipset_lookup,
158        seen,
159    }: ExportOptions<S>,
160) -> anyhow::Result<ExportResult<D>> {
161    if include_events && !include_receipts {
162        anyhow::bail!("message receipts must be included when events are included");
163    }
164
165    let start = Instant::now();
166    tracing::info!(
167        "Exporting snapshot, epoch={}, depth={lookup_depth}, prefix_frames={}",
168        tipset.epoch(),
169        prefix_data_frames.as_ref().map(|v| v.len()).unwrap_or(0)
170    );
171
172    let stateroot_lookup_limit = tipset.epoch() - lookup_depth;
173
174    // Wrap writer in optional checksum calculator
175    let mut writer = AsyncWriterWithChecksum::<D, _>::new(BufWriter::new(writer), !skip_checksum);
176
177    let (ts_lookup_tx, ts_lookup_handle) = if include_tipset_lookup {
178        let (ts_lookup_tx, ts_lookup_rx) = flume::bounded::<(ChainEpoch, TipsetKey)>(1024);
179        let handle = AbortOnDropHandle::new(tokio::spawn(async move {
180            let mut hamt = Hamt::new_with_bit_width(
181                IndexMapBlockstore::default(),
182                TIPSET_LOOKUP_HAMT_BIT_WIDTH,
183            );
184            while let Ok((epoch, tsk)) = ts_lookup_rx.recv_async().await {
185                hamt.set(epoch, tsk)?;
186            }
187            hamt.flush()?;
188            anyhow::Ok(hamt)
189        }));
190        (Some(ts_lookup_tx), Some(handle))
191    } else {
192        (None, None)
193    };
194
195    // Stream stateroots in range (stateroot_lookup_limit+1)..=tipset.epoch(). Also
196    // stream all block headers until genesis.
197    let (blocks, _drop_guard) = par_buffer(
198        // Queue 1k blocks. This is enough to saturate the compressor and blocks
199        // are small enough that keeping 1k in memory isn't a problem. Average
200        // block size is between 1kb and 2kb.
201        1024,
202        stream_chain(
203            db.shallow_clone(),
204            tipset
205                .shallow_clone()
206                .chain_owned(db.shallow_clone())
207                .inspect(move |ts| {
208                    if let Some(ts_lookup_tx) = &ts_lookup_tx
209                        && ChainIndex::is_tipset_lookup_checkpoint(ts.epoch())
210                    {
211                        _ = ts_lookup_tx.send((ts.epoch(), ts.key().clone()));
212                    }
213                }),
214            stateroot_lookup_limit,
215            seen,
216        )
217        .with_message_receipts(include_receipts)
218        .with_events(include_events)
219        .with_tipset_keys(include_tipset_keys)
220        .track_progress(true),
221    );
222
223    // Encode Ipld key-value pairs in zstd frames
224    let block_frames = forest::Encoder::compress_stream_default(blocks);
225    let frames = futures::stream::iter(prefix_data_frames.unwrap_or_default()).chain(block_frames);
226
227    // Write zstd frames and include a skippable index
228    forest::Encoder::write(&mut writer, roots, frames).await?;
229
230    // Flush to ensure everything has been successfully written
231    tokio::time::timeout(forest::ASYNC_OPS_TIMEOUT, writer.flush())
232        .await
233        .context("`writer.flush` timed out")??;
234
235    let digest = writer.finalize().map_err(|e| Error::Other(e.to_string()))?;
236
237    let tipset_lookup = if let Some(ts_lookup_handle) = ts_lookup_handle {
238        // This join is not I/O-bound: the task finishes once the `par_buffer` producer
239        // exits and drops `ts_lookup_tx`, which `Encoder::write` guarantees by exhausting
240        // the frame stream. The timeout guards against a pipeline lifecycle bug keeping a
241        // sender alive, not against slowness.
242        Some(
243            tokio::time::timeout(forest::ASYNC_OPS_TIMEOUT, ts_lookup_handle)
244                .await
245                .context(
246                    "tipset-lookup task did not finish; is a `ts_lookup_tx` sender still alive?",
247                )??,
248        )
249    } else {
250        None
251    };
252
253    tracing::info!(
254        "Exported snapshot, took {}",
255        humantime::format_duration(start.elapsed())
256    );
257
258    Ok(ExportResult {
259        checksum: digest,
260        tipset_lookup,
261    })
262}
forest/chain/mod.rs

forest/chain/
mod.rs
