Skip to main content

forest/db/car/
forest.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! # Forest CAR format
5//!
6//! See [`crate::db::car::plain`] for details on the CAR format.
7//!
8//! The `forest.car.zst` format wraps multiple CAR blocks in small (usually 8 KiB)
9//! zstd frames, and has an index in one ore more skippable zstd frames (each
10//! skippable frame contains up to `u32::MAX` bytes). At the end of the data, there
11//! has to be a fixed-size skippable frame containing magic numbers and meta
12//! information about the archive. CAR blocks may not span multiple z-frames
13//! and the CAR header is kept it a separate z-frame.
14//!
15//! Imagine a `forest.car.zst` archive with 5 blocks. They could be arranged in
16//! z-frames as drawn below:
17//!
18//! ```text
19//!  Z-Frame 1   Z-Frame 2   Z-Frame 3   Skip Frames    Skip Frame
20//! ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌───────────┐ ┌────────────┐
21//! │┌──────┐ │ │┌───────┐│ │┌───────┐│ │Offsets    │ │Index offset│
22//! ││Header│ │ ││Block 1││ ││Block 4││ │ Z-Frame 2 │ │Magic number│
23//! │└──────┘ │ │└───────┘│ │└───────┘│ │ Z-Frame 2 │ │Version info|
24//! └─────────┘ │┌───────┐│ │┌───────┐│ │ Z-Frame 2 │ └────────────┘
25//!             ││Block 2││ ││Block 5││ │ Z-Frame 3 │
26//!             │└───────┘│ │└───────┘│ │ Z-Frame 3 │
27//!             │┌───────┐│ └─────────┘ └───────────┘
28//!             ││Block 3││
29//!             │└───────┘│
30//!             └─────────┘
31//! ```
32//!
33//! Looking up a block uses an [`index::Reader`] to find
34//! the right z-frame. The frame is then decoded and each block is linearly
35//! scanned until a match is found. Decoded (and scanned) z-frames are stored in
36//! a cache for faster repeat retrievals.
37//!
38//! `forest.car.zst` files are backward compatible with Lotus (and all other
39//! tools that consume compressed CAR files). All Forest-specifc information is
40//! encoded as skippable frames that are (as the name suggests) skipped by tools
41//! that don't understand them.
42//!
43//! # Additional reading
44//!
45//! `zstd` frame format: <https://github.com/facebook/zstd/blob/dev/doc/zstd_compression_format.md>
46//! skippable `zstd` frames: <https://github.com/facebook/zstd/blob/dev/doc/zstd_compression_format.md#skippable-frames>
47//! CARv1 specification: <https://ipld.io/specs/transport/car/carv1/>
48//!
49
50use super::{CacheKey, ZstdFrameCache};
51use crate::blocks::{Tipset, TipsetKey};
52use crate::chain::FilecoinSnapshotMetadata;
53use crate::db::car::RandomAccessFileReader;
54use crate::db::car::forest::index::ZstdSkipFramesEncodedDataReader;
55use crate::prelude::*;
56use crate::utils::db::car_stream::{CarBlock, CarV1Header, uvi_bytes};
57use crate::utils::encoding::from_slice_with_fallback;
58use crate::utils::io::EitherMmapOrRandomAccessFile;
59use bytes::{BufMut as _, Bytes, BytesMut, buf::Writer};
60use futures::{Stream, TryStreamExt as _};
61use fvm_ipld_encoding::CborStore as _;
62use integer_encoding::VarIntReader;
63use nunny::Vec as NonEmpty;
64use positioned_io::{Cursor, ReadAt, Size as _, SizeCursor};
65use std::io::{self, Read, Seek, SeekFrom, Write};
66use std::path::{Path, PathBuf};
67use std::sync::OnceLock;
68use std::task::Poll;
69use std::time::Duration;
70use tokio::io::{AsyncWrite, AsyncWriteExt};
71use tokio_util::codec::{Decoder, Encoder as _};
72
73#[cfg(feature = "benchmark-private")]
74pub mod index;
75#[cfg(not(feature = "benchmark-private"))]
76mod index;
77
78pub const FOREST_CAR_FILE_EXTENSION: &str = ".forest.car.zst";
79pub const TEMP_FOREST_CAR_FILE_EXTENSION: &str = ".forest.car.zst.tmp";
80/// <https://github.com/facebook/zstd/blob/dev/doc/zstd_compression_format.md#skippable-frames>
81pub const ZSTD_SKIPPABLE_FRAME_MAGIC_HEADER: [u8; 4] = [0x50, 0x2A, 0x4D, 0x18];
82pub const DEFAULT_FOREST_CAR_FRAME_SIZE: usize = 8000_usize.next_power_of_two();
83pub const DEFAULT_FOREST_CAR_COMPRESSION_LEVEL: u16 = zstd::DEFAULT_COMPRESSION_LEVEL as _;
84pub const ZSTD_SKIP_FRAME_LEN: u64 = 8;
85
86/// `zstd` frame of Forest CAR
87pub type ForestCarFrame = (Vec<Cid>, Bytes);
88
89pub struct ForestCar<ReaderT> {
90    // Multiple `ForestCar` structures may share the same cache. The cache key is used to identify
91    // the origin of a cached z-frame.
92    cache_key: CacheKey,
93    indexed: index::Reader<index::ZstdSkipFramesEncodedDataReader<positioned_io::Slice<ReaderT>>>,
94    index_size_bytes: u64,
95    frame_cache: ZstdFrameCache,
96    header: CarV1Header,
97    metadata: OnceLock<Option<FilecoinSnapshotMetadata>>,
98}
99
100impl<ReaderT: super::RandomAccessFileReader> ForestCar<ReaderT> {
101    pub fn new(reader: ReaderT) -> io::Result<ForestCar<ReaderT>> {
102        let validation_result = Self::validate_car(&reader)?;
103        Self::new_from_validation_result(reader, validation_result)
104    }
105
106    pub(super) fn new_from_validation_result(
107        reader: ReaderT,
108        (header, index_start_pos, index_size_bytes): (CarV1Header, u64, u64),
109    ) -> io::Result<ForestCar<ReaderT>> {
110        let indexed = index::Reader::new(index::ZstdSkipFramesEncodedDataReader::new(
111            positioned_io::Slice::new(reader, index_start_pos, Some(index_size_bytes)),
112        ))?;
113        Ok(ForestCar {
114            cache_key: 0,
115            indexed,
116            index_size_bytes,
117            frame_cache: ZstdFrameCache::default(),
118            header,
119            metadata: OnceLock::new(),
120        })
121    }
122
123    pub fn header_v1(&self) -> &CarV1Header {
124        &self.header
125    }
126
127    pub fn metadata(&self) -> Option<&FilecoinSnapshotMetadata> {
128        self.metadata
129            .get_or_init(|| {
130                if self.header.roots.len() == super::V2_SNAPSHOT_ROOT_COUNT {
131                    let maybe_metadata_cid = self.header.roots.first();
132                    if let Ok(Some(metadata)) =
133                        self.get_cbor::<FilecoinSnapshotMetadata>(maybe_metadata_cid)
134                    {
135                        return Some(metadata);
136                    }
137                }
138                None
139            })
140            .as_ref()
141    }
142
143    pub fn is_valid(reader: &ReaderT) -> bool {
144        Self::validate_car(reader).is_ok()
145    }
146
147    pub(super) fn validate_car(reader: &ReaderT) -> io::Result<(CarV1Header, u64, u64)> {
148        let mut cursor = SizeCursor::new(&reader);
149        cursor.seek(SeekFrom::End(-(ForestCarFooter::SIZE as i64)))?;
150        let index_end_pos = cursor.position();
151
152        let mut footer_buffer = [0; ForestCarFooter::SIZE];
153        cursor.read_exact(&mut footer_buffer)?;
154
155        let footer = ForestCarFooter::try_from_le_bytes(footer_buffer).ok_or_else(|| {
156            invalid_data(format!(
157                "not recognizable as a `{FOREST_CAR_FILE_EXTENSION}` file"
158            ))
159        })?;
160        let index_start_pos = footer.index.checked_sub(ZSTD_SKIP_FRAME_LEN).ok_or_else(||
161            invalid_data(format!(
162                "unexpected error: footer.index({}) < ZSTD_SKIP_FRAME_LEN({ZSTD_SKIP_FRAME_LEN})",
163                footer.index
164            )),
165        )?;
166        let index_len = index_end_pos.checked_sub(index_start_pos).ok_or_else(||
167            invalid_data(format!("unexpected error: index_end_pos({index_end_pos}) < index_start_pos({index_start_pos})"))
168        )?;
169
170        let cursor = Cursor::new_pos(&reader, 0);
171        let mut header_zstd_frame = decode_zstd_single_frame(cursor)?.into();
172        let block_frame = uvi_bytes()
173            .decode(&mut header_zstd_frame)?
174            .ok_or_else(|| invalid_data("malformed uvibytes"))?;
175        let header = from_slice_with_fallback::<CarV1Header>(&block_frame)
176            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
177
178        Ok((header, index_start_pos, index_len))
179    }
180
181    pub fn head_tipset_key(&self) -> &NonEmpty<Cid> {
182        // head tipset key is stored in v2 snapshot metadata
183        // See <https://github.com/filecoin-project/FIPs/blob/98e33b9fa306959aa0131519eb4cc155522b2081/FRCs/frc-0108.md#v2-specification>
184        if let Some(metadata) = self.metadata() {
185            &metadata.head_tipset_key
186        } else {
187            &self.header.roots
188        }
189    }
190
191    pub fn index_size_bytes(&self) -> u64 {
192        self.index_size_bytes
193    }
194
195    pub fn heaviest_tipset_key(&self) -> TipsetKey {
196        TipsetKey::from(self.head_tipset_key().clone())
197    }
198
199    pub fn heaviest_tipset(&self) -> anyhow::Result<Tipset> {
200        Tipset::load_required(self, &self.heaviest_tipset_key())
201    }
202
203    pub fn into_dyn(self) -> ForestCar<Box<dyn super::RandomAccessFileReader>> {
204        ForestCar {
205            cache_key: self.cache_key,
206            indexed: self.indexed.map(|slice| {
207                let offset = slice.inner().offset();
208                let size = slice.inner().size().ok().flatten();
209                ZstdSkipFramesEncodedDataReader::new(positioned_io::Slice::new(
210                    Box::new(slice.into_inner().into_inner()) as Box<dyn RandomAccessFileReader>,
211                    offset,
212                    size,
213                ))
214            }),
215            index_size_bytes: self.index_size_bytes,
216            frame_cache: self.frame_cache,
217            header: self.header,
218            metadata: self.metadata,
219        }
220    }
221
222    pub fn with_cache(self, frame_cache: ZstdFrameCache, key: CacheKey) -> Self {
223        Self {
224            cache_key: key,
225            frame_cache,
226            ..self
227        }
228    }
229
230    /// Gets a reader of the block data by its `Cid`
231    pub fn get_reader(&self, k: Cid) -> anyhow::Result<Option<impl Read>> {
232        for position in self.indexed.get(k)? {
233            // escape the positioned_io::Slice
234            let entire_file = self.indexed.reader().inner().get_ref();
235            // `position` is the frame start offset.
236            let cursor = Cursor::new_pos(entire_file, position);
237            let mut decoder = zstd::Decoder::new(cursor)?.single_frame();
238            while let Ok(car_block_len) = decoder.read_varint::<usize>() {
239                let cid = Cid::read_bytes(&mut decoder)?;
240                let data_len = car_block_len.saturating_sub(cid.encoded_len()) as u64;
241                if cid == k {
242                    // return the reader instead of decoding the entire data block into memory
243                    return Ok(Some(decoder.take(data_len)));
244                }
245                // Discard data bytes
246                io::copy(&mut decoder.by_ref().take(data_len), &mut io::sink())?;
247            }
248        }
249        Ok(None)
250    }
251}
252
253impl TryFrom<&Path> for ForestCar<EitherMmapOrRandomAccessFile> {
254    type Error = std::io::Error;
255    fn try_from(path: &Path) -> std::io::Result<Self> {
256        ForestCar::new(EitherMmapOrRandomAccessFile::open(path)?)
257    }
258}
259
260impl<ReaderT> Blockstore for ForestCar<ReaderT>
261where
262    ReaderT: ReadAt,
263{
264    #[tracing::instrument(level = "trace", skip(self))]
265    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
266        let indexed = &self.indexed;
267        for position in indexed.get(*k)?.into_iter() {
268            let cache_query = self.frame_cache.get(position, self.cache_key, *k);
269            match cache_query {
270                // Frame cache hit, found value.
271                Some(Some(val)) => return Ok(Some(val.to_vec())),
272                // Frame cache hit, no value. This only happens when hashes collide
273                Some(None) => {}
274                None => {
275                    // Decode entire frame into memory, "position" arg is the frame start offset.
276                    let entire_file = indexed.reader().inner().get_ref(); // escape the positioned_io::Slice
277                    let cursor = Cursor::new_pos(entire_file, position);
278                    let mut zstd_frame = decode_zstd_single_frame(cursor)?.into();
279                    // Parse all key-value pairs and insert them into a map
280                    let mut block_map = hashbrown::HashMap::new();
281                    while let Some(block_frame) = uvi_bytes().decode_eof(&mut zstd_frame)? {
282                        let CarBlock { cid, data } = CarBlock::from_bytes(block_frame)?;
283                        block_map.insert(cid.into(), data);
284                    }
285                    let get_result = block_map.get(k).cloned();
286                    self.frame_cache.put(position, self.cache_key, block_map);
287
288                    // This lookup only fails in case of a hash collision
289                    if let Some(value) = get_result {
290                        return Ok(Some(value.to_vec()));
291                    }
292                }
293            }
294        }
295        Ok(None)
296    }
297
298    /// Not supported, use [`super::ManyCar`] instead.
299    fn put_keyed(&self, _: &Cid, _: &[u8]) -> anyhow::Result<()> {
300        anyhow::bail!("ForestCar is read-only, use ManyCar instead");
301    }
302}
303
304fn decode_zstd_single_frame<ReaderT: Read>(reader: ReaderT) -> io::Result<Bytes> {
305    let mut zstd_frame = vec![];
306    zstd::Decoder::new(reader)?
307        .single_frame()
308        .read_to_end(&mut zstd_frame)?;
309    Ok(zstd_frame.into())
310}
311
312/// Timeout applied to each async I/O operation of the snapshot export pipeline so that a
313/// stalled reader or writer surfaces as an error instead of wedging the export forever
314/// while `Forest.ChainExportStatus` keeps reporting it as in progress.
315pub(crate) const ASYNC_OPS_TIMEOUT: Duration = Duration::from_mins(5);
316
317pub struct Encoder {}
318
319impl Encoder {
320    pub async fn write(
321        mut sink: impl AsyncWrite + Unpin,
322        roots: NonEmpty<Cid>,
323        mut stream: impl Stream<Item = anyhow::Result<ForestCarFrame>> + Unpin,
324    ) -> anyhow::Result<()> {
325        let mut offset = 0;
326
327        // Write CARv1 header
328        let mut header_encoder = new_encoder(DEFAULT_FOREST_CAR_COMPRESSION_LEVEL)?;
329
330        let header = CarV1Header { roots, version: 1 };
331        let mut header_uvi_frame = BytesMut::new();
332        uvi_bytes().encode(
333            Bytes::from(fvm_ipld_encoding::to_vec(&header)?),
334            &mut header_uvi_frame,
335        )?;
336        header_encoder.write_all(&header_uvi_frame)?;
337        let header_bytes = header_encoder.finish()?.into_inner().freeze();
338
339        tokio::time::timeout(ASYNC_OPS_TIMEOUT, sink.write_all(&header_bytes))
340            .await
341            .context("header `sink.write_all` timed out")??;
342        let header_len = header_bytes.len();
343
344        offset += header_len;
345
346        // Write seekable zstd and collect a mapping of CIDs to frame_offset+data_offset.
347        let mut builder = index::Builder::new();
348        let mut n_frames = 0;
349        while let Some((cids, zstd_frame)) =
350            tokio::time::timeout(ASYNC_OPS_TIMEOUT, stream.try_next())
351                .await
352                .with_context(|| {
353                    format!("`stream.try_next` timed out, offset={offset}, n_frames={n_frames}")
354                })??
355        {
356            builder.extend(cids.into_iter().map(|cid| (cid, offset as u64)));
357            tokio::time::timeout(ASYNC_OPS_TIMEOUT, sink.write_all(&zstd_frame))
358                .await
359                .with_context(|| format!("`sink.write_all` timed out, offset={offset}, n_frames={n_frames}, zstd_frame_len={}", zstd_frame.len()))??;
360            offset += zstd_frame.len();
361            n_frames += 1;
362        }
363
364        tracing::info!("Finished writing {n_frames} zstd CAR frames");
365
366        // Create index
367        let writer = builder.into_writer();
368        tokio::time::timeout(
369            ASYNC_OPS_TIMEOUT,
370            writer.write_zstd_skip_frames_into(&mut sink),
371        )
372        .await
373        .context("`writer.write_zstd_skip_frames_into` timed out")??;
374        tracing::info!("Finished writing zstd CAR index frames");
375        // Write ForestCAR.zst footer, it's a valid ZSTD skip-frame
376        let footer = ForestCarFooter {
377            index: offset as u64 + ZSTD_SKIP_FRAME_LEN,
378        };
379        tokio::time::timeout(ASYNC_OPS_TIMEOUT, sink.write_all(&footer.to_le_bytes()))
380            .await
381            .context("footer `sink.write_all` timed out")??;
382        tracing::info!("Finished writing zstd CAR footer frame");
383        Ok(())
384    }
385
386    /// `compress_stream` with [`DEFAULT_FOREST_CAR_FRAME_SIZE`] as default frame size and [`DEFAULT_FOREST_CAR_COMPRESSION_LEVEL`] as default compression level.
387    pub fn compress_stream_default(
388        stream: impl Stream<Item = anyhow::Result<CarBlock>>,
389    ) -> impl Stream<Item = anyhow::Result<ForestCarFrame>> {
390        Self::compress_stream(
391            DEFAULT_FOREST_CAR_FRAME_SIZE,
392            DEFAULT_FOREST_CAR_COMPRESSION_LEVEL,
393            stream,
394        )
395    }
396
397    /// Consume stream of blocks, emit a new position of each block and a stream
398    /// of zstd frames.
399    pub fn compress_stream(
400        zstd_frame_size_tripwire: usize,
401        zstd_compression_level: u16,
402        stream: impl Stream<Item = anyhow::Result<CarBlock>>,
403    ) -> impl Stream<Item = anyhow::Result<ForestCarFrame>> {
404        let mut encoder_store = new_encoder(zstd_compression_level);
405        let mut frame_cids = vec![];
406
407        let mut stream = Box::pin(stream.into_stream());
408        futures::stream::poll_fn(move |cx| {
409            let encoder = match encoder_store.as_mut() {
410                Err(e) => {
411                    let dummy_error = io::Error::other("Error already consumed.");
412                    return Poll::Ready(Some(Err(anyhow::Error::from(std::mem::replace(
413                        e,
414                        dummy_error,
415                    )))));
416                }
417                Ok(encoder) => encoder,
418            };
419            loop {
420                // Emit frame if compressed_len > zstd_frame_size_tripwire
421                if compressed_len(encoder) > zstd_frame_size_tripwire {
422                    let cids = std::mem::take(&mut frame_cids);
423                    let frame = finalize_frame(zstd_compression_level, encoder)?;
424                    return Poll::Ready(Some(Ok((cids, frame))));
425                }
426                // No frame to emit, let's get another block
427                let ret = futures::ready!(stream.as_mut().poll_next(cx));
428                match ret {
429                    // End-of-stream
430                    None => {
431                        // If there's anything in the zstd buffer, emit it.
432                        if compressed_len(encoder) > 0 {
433                            let cids = std::mem::take(&mut frame_cids);
434                            let frame = finalize_frame(zstd_compression_level, encoder)?;
435                            return Poll::Ready(Some(Ok((cids, frame))));
436                        } else {
437                            // Otherwise we're all done.
438                            return Poll::Ready(None);
439                        }
440                    }
441                    // Pass errors through
442                    Some(Err(e)) => {
443                        return Poll::Ready(Some(Err(anyhow::anyhow!(
444                            "error polling CarBlock from stream: {e:#}"
445                        ))));
446                    }
447                    // Got element, add to encoder and emit block position
448                    Some(Ok(block)) => {
449                        frame_cids.push(block.cid);
450                        block.write(encoder)?;
451                        encoder.flush()?;
452                    }
453                }
454            }
455        })
456    }
457}
458
459fn invalid_data(inner: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> io::Error {
460    io::Error::new(io::ErrorKind::InvalidData, inner)
461}
462
463fn compressed_len(encoder: &zstd::Encoder<'static, Writer<BytesMut>>) -> usize {
464    encoder.get_ref().get_ref().len()
465}
466
467pub fn finalize_frame(
468    zstd_compression_level: u16,
469    encoder: &mut zstd::Encoder<'static, Writer<BytesMut>>,
470) -> io::Result<Bytes> {
471    let prev_encoder = std::mem::replace(encoder, new_encoder(zstd_compression_level)?);
472    Ok(prev_encoder.finish()?.into_inner().freeze())
473}
474
475pub fn new_encoder(
476    zstd_compression_level: u16,
477) -> io::Result<zstd::Encoder<'static, Writer<BytesMut>>> {
478    zstd::Encoder::new(BytesMut::new().writer(), i32::from(zstd_compression_level))
479}
480
481#[derive(Debug, Clone, Eq, PartialEq)]
482#[cfg_attr(test, derive(derive_quickcheck_arbitrary::Arbitrary))]
483struct ForestCarFooter {
484    index: u64,
485}
486
487impl ForestCarFooter {
488    pub const SIZE: usize = 16;
489
490    pub fn to_le_bytes(&self) -> [u8; Self::SIZE] {
491        let mut buffer = [0; 16];
492        // Skippable frames start with 50 2A 4D 18
493        buffer[0..4].copy_from_slice(&ZSTD_SKIPPABLE_FRAME_MAGIC_HEADER);
494        // Then a u32 containing the length of the data in the frame
495        buffer[4..8].copy_from_slice(&(std::mem::size_of_val(&self.index) as u32).to_le_bytes());
496        // And finally the metadata we want to store
497        buffer[8..16].copy_from_slice(&self.index.to_le_bytes());
498        buffer
499    }
500
501    pub fn try_from_le_bytes(bytes: [u8; Self::SIZE]) -> Option<ForestCarFooter> {
502        let index = u64::from_le_bytes(bytes[8..16].try_into().expect("infallible"));
503        let footer = ForestCarFooter { index };
504        if bytes == footer.to_le_bytes() {
505            Some(footer)
506        } else {
507            None
508        }
509    }
510}
511
512pub fn new_forest_car_temp_path_in(
513    output_dir: impl AsRef<Path>,
514) -> std::io::Result<tempfile::TempPath> {
515    Ok(tempfile::Builder::new()
516        .suffix(TEMP_FOREST_CAR_FILE_EXTENSION)
517        .tempfile_in(output_dir)?
518        .into_temp_path())
519}
520
521pub fn tmp_exporting_forest_car_path(output_path: &Path) -> PathBuf {
522    let mut p = output_path.to_owned();
523    p.add_extension("tmp");
524    p
525}
526
527pub fn forest_car_sha256sum_path(output_path: &Path) -> PathBuf {
528    let mut p = output_path.to_owned();
529    p.add_extension("sha256sum");
530    p
531}
532
533pub fn forest_car_with_filename_suffix(path: &Path, suffix: &str) -> anyhow::Result<PathBuf> {
534    anyhow::ensure!(!suffix.is_empty(), "suffix cannot be empty");
535    let file_name = path.file_name().and_then(|n| n.to_str()).with_context(|| {
536        format!(
537            "failed to extract filename from the given path: {}",
538            path.display()
539        )
540    })?;
541    let new_name = match file_name.split_once('.') {
542        Some((stem, rest)) => format!("{stem}{suffix}.{rest}"),
543        None => format!("{file_name}{suffix}"),
544    };
545    Ok(path.with_file_name(new_name))
546}
547
548#[cfg(test)]
549mod tests {
550    use super::*;
551    use nunny::vec as nonempty;
552    use quickcheck_macros::quickcheck;
553    use rstest::rstest;
554    use tokio_test::block_on;
555
556    fn mk_encoded_car(
557        zstd_frame_size_tripwire: usize,
558        zstd_compression_level: u16,
559        roots: NonEmpty<Cid>,
560        blocks: NonEmpty<CarBlock>,
561    ) -> Vec<u8> {
562        block_on(async {
563            let frame_stream = Encoder::compress_stream(
564                zstd_frame_size_tripwire,
565                zstd_compression_level,
566                futures::stream::iter(blocks.into_iter().map(Ok)),
567            );
568            let mut encoded = vec![];
569            Encoder::write(&mut encoded, roots, frame_stream)
570                .await
571                .unwrap();
572            encoded
573        })
574    }
575
576    #[quickcheck]
577    fn forest_car_create_basic(blocks: nunny::Vec<CarBlock>) {
578        let roots = nonempty!(blocks.first().cid);
579        let forest_car =
580            ForestCar::new(mk_encoded_car(1024 * 4, 3, roots.clone(), blocks.clone())).unwrap();
581        assert_eq!(forest_car.head_tipset_key(), &roots);
582        for block in blocks {
583            assert_eq!(forest_car.get(&block.cid).unwrap().unwrap(), block.data);
584            let mut buf = vec![];
585            forest_car
586                .get_reader(block.cid)
587                .unwrap()
588                .unwrap()
589                .read_to_end(&mut buf)
590                .unwrap();
591            assert_eq!(buf, block.data);
592        }
593    }
594
595    #[quickcheck]
596    fn forest_car_create_options(
597        blocks: nunny::Vec<CarBlock>,
598        frame_size: usize,
599        mut compression_level: u16,
600    ) {
601        compression_level %= 15;
602        let roots = nonempty!(blocks.first().cid);
603
604        let forest_car = ForestCar::new(mk_encoded_car(
605            frame_size,
606            compression_level.max(1),
607            roots.clone(),
608            blocks.clone(),
609        ))
610        .unwrap();
611        assert_eq!(forest_car.head_tipset_key(), &roots);
612        for block in blocks {
613            assert_eq!(
614                forest_car.get(&block.cid).unwrap().map(Bytes::from),
615                Some(block.data)
616            );
617        }
618    }
619
620    #[quickcheck]
621    fn forest_car_open_invalid(junk: Vec<u8>) {
622        // The chance of thinking random data is a valid ForestCar should be practically zero.
623        assert!(ForestCar::new(junk).is_err());
624    }
625
626    #[quickcheck]
627    fn forest_footer_roundtrip(footer: ForestCarFooter) {
628        let footer_recoded = ForestCarFooter::try_from_le_bytes(footer.to_le_bytes());
629        assert_eq!(footer_recoded, Some(footer));
630    }
631
632    // Two colliding hashes in separate zstd-frames should not affect each other.
633    #[test]
634    fn encode_hash_collisions() {
635        use crate::utils::multihash::prelude::*;
636
637        // Distinct CIDs may map to the same hash value
638        let cid_a = Cid::new_v1(0, MultihashCode::Identity.digest(&[10]));
639        let cid_b = Cid::new_v1(0, MultihashCode::Identity.digest(&[0]));
640        // A and B are _not_ the same...
641        assert_ne!(cid_a, cid_b);
642        // ... but they map to the same hash:
643        assert_eq!(index::hash::summary(&cid_a), index::hash::summary(&cid_b));
644
645        // For testing purposes, we ignore that the data doesn't map to the
646        // CIDs.
647        let blocks = nonempty![
648            CarBlock {
649                cid: cid_a,
650                data: "bill and ben".into(),
651            },
652            CarBlock {
653                cid: cid_b,
654                data: "the flowerpot men".into(),
655            },
656        ];
657
658        // Setting the desired frame size to 0 means each block will be put in a separate frame.
659        let forest_car = ForestCar::new(mk_encoded_car(
660            0,
661            3,
662            nonempty![blocks.first().cid],
663            blocks.clone(),
664        ))
665        .unwrap();
666
667        // Even with colliding hashes, the CIDs can still be queried:
668        assert_eq!(forest_car.get(&cid_a).unwrap().unwrap(), blocks[0].data);
669        assert_eq!(forest_car.get(&cid_b).unwrap().unwrap(), blocks[1].data);
670    }
671
672    #[rstest]
673    #[case(
674        Path::new("/tmp/a.forest.car.zst"),
675        Path::new("/tmp/a.forest.car.zst.tmp")
676    )]
677    #[case(
678        Path::new("tmp/a.forest.car.zst"),
679        Path::new("tmp/a.forest.car.zst.tmp")
680    )]
681    #[case(Path::new("a.forest.car.zst"), Path::new("a.forest.car.zst.tmp"))]
682    #[case(Path::new(""), Path::new(""))]
683    #[case(Path::new("."), Path::new("."))]
684    fn test_tmp_exporting_forest_car_path(#[case] input: &Path, #[case] output: &Path) {
685        assert_eq!(tmp_exporting_forest_car_path(input), output);
686    }
687
688    #[rstest]
689    #[case(
690        Path::new("/tmp/a.forest.car.zst"),
691        Path::new("/tmp/a.forest.car.zst.sha256sum")
692    )]
693    #[case(
694        Path::new("tmp/a.forest.car.zst"),
695        Path::new("tmp/a.forest.car.zst.sha256sum")
696    )]
697    #[case(Path::new("a.forest.car.zst"), Path::new("a.forest.car.zst.sha256sum"))]
698    #[case(Path::new(""), Path::new(""))]
699    #[case(Path::new("."), Path::new("."))]
700    fn test_forest_car_sha256sum_path(#[case] input: &Path, #[case] output: &Path) {
701        assert_eq!(forest_car_sha256sum_path(input), output);
702    }
703
704    #[rstest]
705    #[case(
706        Path::new("/tmp/a.forest.car.zst"),
707        "_suffix",
708        Path::new("/tmp/a_suffix.forest.car.zst")
709    )]
710    #[case(
711        Path::new("a.forest.car.zst"),
712        "_suffix",
713        Path::new("a_suffix.forest.car.zst")
714    )]
715    #[case(Path::new("a"), "_suffix", Path::new("a_suffix"))]
716    fn test_forest_car_with_filename_suffix_valid(
717        #[case] input: &Path,
718        #[case] suffix: &str,
719        #[case] output: &Path,
720    ) {
721        assert_eq!(
722            forest_car_with_filename_suffix(input, suffix).unwrap(),
723            output
724        );
725    }
726
727    #[rstest]
728    #[case(Path::new("/tmp/a.forest.car.zst"), "", "suffix cannot be empty")]
729    #[case(Path::new("."), "_suffix", "failed to extract filename")]
730    fn test_forest_car_with_filename_suffix_invalid(
731        #[case] input: &Path,
732        #[case] suffix: &str,
733        #[case] reason: &str,
734    ) {
735        let e = forest_car_with_filename_suffix(input, suffix).unwrap_err();
736        assert!(e.to_string().contains(reason));
737    }
738}