forest/db/car/
plain.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! # Varint frames
5//!
6//! CARs are made of concatenations of _varint frames_. Each varint frame is a concatenation of the
7//! _body length_ as an
8//! [varint](https://docs.rs/integer-encoding/4.0.0/integer_encoding/trait.VarInt.html), and the
9//! _frame body_ itself. [`unsigned_varint::codec::UviBytes`] can be used to read frames
10//! piecewise into memory.
11//!
12//! ```text
13//!        varint frame
14//! │◄───────────────────────►│
15//! │                         │
16//! ├───────────┬─────────────┤
17//! │varint:    │             │
18//! │body length│frame body   │
19//! └───────────┼─────────────┤
20//!             │             │
21//! frame body ►│◄───────────►│
22//!     offset     =body length
23//! ```
24//!
25//! # CARv1 layout and seeking
26//!
27//! The first varint frame is a _header frame_, where the frame body is a [`CarHeader`] encoded
28//! using [`ipld_dagcbor`](serde_ipld_dagcbor).
29//!
30//! Subsequent varint frames are _block frames_, where the frame body is a concatenation of a
31//! [`Cid`] and the _block data_ addressed by that CID.
32//!
33//! ```text
34//! block frame ►│
35//! body offset  │
36//!              │  =body length
37//!              │◄────────────►│
38//!  ┌───────────┼───┬──────────┤
39//!  │body length│cid│block data│
40//!  └───────────┴───┼──────────┤
41//!                  │◄────────►│
42//!                  │  =block data length
43//!      block data  │
44//!          offset ►│
45//! ```
46//!
47//! ## Block ordering
48//! > _... a filecoin-deterministic car-file is currently implementation-defined as containing all
49//! > DAG-forming blocks in first-seen order, as a result of a depth-first DAG traversal starting
50//! > from a single root._
51//! - [CAR documentation](https://ipld.io/specs/transport/car/carv1/#determinism)
52//!
53//! # Future work
54//! - [`fadvise`](https://linux.die.net/man/2/posix_fadvise)-based APIs to pre-fetch parts of the
55//!   file, to improve random access performance.
56//! - Use an inner [`Blockstore`] for writes.
57//! - Use safe arithmetic for all operations - a malicious frame shouldn't cause a crash.
58//! - Theoretically, file-backed blockstores should be clonable (or even [`Sync`]) with very low
59//!   overhead, so that multiple threads could perform operations concurrently.
60//! - CARv2 support
61//! - A wrapper that abstracts over car formats for reading.
62
63use crate::chain::FilecoinSnapshotMetadata;
64use crate::cid_collections::CidHashMap;
65use crate::db::PersistentStore;
66use crate::utils::db::car_stream::{CarV1Header, CarV2Header};
67use crate::{
68    blocks::{Tipset, TipsetKey},
69    utils::encoding::from_slice_with_fallback,
70};
71use cid::Cid;
72use fvm_ipld_blockstore::Blockstore;
73use fvm_ipld_encoding::CborStore as _;
74use integer_encoding::{FixedIntReader, VarIntReader};
75use nunny::Vec as NonEmpty;
76use parking_lot::RwLock;
77use positioned_io::ReadAt;
78use std::{
79    io::{
80        self, BufReader,
81        ErrorKind::{InvalidData, Unsupported},
82        Read, Seek, SeekFrom,
83    },
84    iter,
85    sync::OnceLock,
86};
87use tokio::io::{AsyncWrite, AsyncWriteExt};
88use tracing::{debug, trace};
89
90/// **Note that all operations on this store are blocking**.
91///
92/// It can often be time, memory, or disk prohibitive to read large snapshots into a database like
93/// [`ParityDb`](crate::db::parity_db::ParityDb).
94///
95/// This is an implementer of [`Blockstore`] that simply wraps an uncompressed [CARv1
96/// file](https://ipld.io/specs/transport/car/carv1).
97///
98/// On creation, [`PlainCar`] builds an in-memory index of the [`Cid`]s in the file,
99/// and their offsets into that file.
100/// Note that it prepares its own buffer for doing so.
101///
102/// When a block is requested, [`PlainCar`] scrolls to that offset, and reads the block, on-demand.
103///
104/// Writes for new blocks (which don't exist in the CAR already) are not supported.
105///
106/// Random-access performance is expected to be poor, as the OS will have to load separate parts of
107/// the file from disk, and flush it for each read. However, (near) linear access should be pretty
108/// good, as file chunks will be pre-fetched.
109///
110/// See [module documentation](mod@self) for more.
111pub struct PlainCar<ReaderT> {
112    reader: ReaderT,
113    index: RwLock<CidHashMap<UncompressedBlockDataLocation>>,
114    version: u64,
115    header_v1: CarV1Header,
116    header_v2: Option<CarV2Header>,
117    metadata: OnceLock<Option<FilecoinSnapshotMetadata>>,
118}
119
120impl<ReaderT: super::RandomAccessFileReader> PlainCar<ReaderT> {
121    /// To be correct:
122    /// - `reader` must read immutable data. e.g if it is a file, it should be
123    ///   [`flock`](https://linux.die.net/man/2/flock)ed.
124    ///   [`Blockstore`] API calls may panic if this is not upheld.
125    #[tracing::instrument(level = "debug", skip_all)]
126    pub fn new(reader: ReaderT) -> io::Result<Self> {
127        let mut cursor = positioned_io::Cursor::new(&reader);
128        let position = cursor.position();
129        let header_v2 = read_v2_header(&mut cursor)?;
130        let (limit_position, version) =
131            if let Some(header_v2) = &header_v2 {
132                cursor.set_position(position.saturating_add(
133                    u64::try_from(header_v2.data_offset).map_err(io::Error::other)?,
134                ));
135                (
136                    Some(cursor.stream_position()?.saturating_add(
137                        u64::try_from(header_v2.data_size).map_err(io::Error::other)?,
138                    )),
139                    2,
140                )
141            } else {
142                cursor.set_position(position);
143                (None, 1)
144            };
145
146        let header_v1 = read_v1_header(&mut cursor)?;
147        // When indexing, we perform small reads of the length and CID before seeking
148        // Buffering these gives us a ~50% speedup (n=10): https://github.com/ChainSafe/forest/pull/3085#discussion_r1246897333
149        let mut buf_reader = BufReader::with_capacity(1024, cursor);
150
151        // now create the index
152        let index = iter::from_fn(|| {
153            read_block_data_location_and_skip(&mut buf_reader, limit_position).transpose()
154        })
155        .collect::<Result<CidHashMap<_>, _>>()?;
156
157        match index.len() {
158            0 => Err(io::Error::new(
159                InvalidData,
160                "CARv1 files must contain at least one block",
161            )),
162            num_blocks => {
163                debug!(num_blocks, "indexed CAR");
164                Ok(Self {
165                    reader,
166                    index: RwLock::new(index),
167                    version,
168                    header_v1,
169                    header_v2,
170                    metadata: OnceLock::new(),
171                })
172            }
173        }
174    }
175
176    pub fn header_v1(&self) -> &CarV1Header {
177        &self.header_v1
178    }
179
180    pub fn metadata(&self) -> Option<&FilecoinSnapshotMetadata> {
181        self.metadata
182            .get_or_init(|| {
183                if self.header_v1.roots.len() == super::V2_SNAPSHOT_ROOT_COUNT {
184                    let maybe_metadata_cid = self.header_v1.roots.first();
185                    if let Ok(Some(metadata)) =
186                        self.get_cbor::<FilecoinSnapshotMetadata>(maybe_metadata_cid)
187                    {
188                        return Some(metadata);
189                    }
190                }
191                None
192            })
193            .as_ref()
194    }
195
196    pub fn head_tipset_key(&self) -> &NonEmpty<Cid> {
197        // head tipset key is stored in v2 snapshot metadata
198        // See <https://github.com/filecoin-project/FIPs/blob/98e33b9fa306959aa0131519eb4cc155522b2081/FRCs/frc-0108.md#v2-specification>
199        if let Some(metadata) = self.metadata() {
200            &metadata.head_tipset_key
201        } else {
202            &self.header_v1.roots
203        }
204    }
205
206    pub fn version(&self) -> u64 {
207        self.version
208    }
209
210    pub fn heaviest_tipset_key(&self) -> TipsetKey {
211        TipsetKey::from(self.head_tipset_key().clone())
212    }
213
214    pub fn heaviest_tipset(&self) -> anyhow::Result<Tipset> {
215        Tipset::load_required(self, &self.heaviest_tipset_key())
216    }
217
218    /// In an arbitrary order
219    #[cfg(test)]
220    pub fn cids(&self) -> Vec<Cid> {
221        self.index.read().keys().collect()
222    }
223
224    pub fn into_dyn(self) -> PlainCar<Box<dyn super::RandomAccessFileReader>> {
225        PlainCar {
226            reader: Box::new(self.reader),
227            index: self.index,
228            version: self.version,
229            header_v1: self.header_v1,
230            header_v2: self.header_v2,
231            metadata: self.metadata,
232        }
233    }
234
235    /// Gets a reader of the block data by its `Cid`
236    pub fn get_reader(&self, k: Cid) -> Option<impl Read> {
237        self.index
238            .read()
239            .get(&k)
240            .map(|UncompressedBlockDataLocation { offset, length }| {
241                positioned_io::Cursor::new_pos(&self.reader, *offset).take(u64::from(*length))
242            })
243    }
244}
245
246impl TryFrom<&'static [u8]> for PlainCar<&'static [u8]> {
247    type Error = io::Error;
248    fn try_from(bytes: &'static [u8]) -> io::Result<Self> {
249        PlainCar::new(bytes)
250    }
251}
252
253/// If you seek to `offset` (from the start of the file), and read `length` bytes,
254/// you should get data that corresponds to a [`Cid`] (but NOT the [`Cid`] itself).
255#[derive(Debug, serde::Serialize, serde::Deserialize)]
256pub struct UncompressedBlockDataLocation {
257    offset: u64,
258    length: u32,
259}
260
261impl<ReaderT> Blockstore for PlainCar<ReaderT>
262where
263    ReaderT: ReadAt,
264{
265    #[tracing::instrument(level = "trace", skip(self))]
266    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
267        match self.index.read().get(k) {
268            Some(UncompressedBlockDataLocation { offset, length }) => {
269                trace!("fetching from disk");
270                let mut data = vec![0; usize::try_from(*length).expect("u32 must fit in usize")];
271                self.reader.read_exact_at(*offset, &mut data)?;
272                Ok(Some(data))
273            }
274            None => {
275                trace!("not found");
276                Ok(None)
277            }
278        }
279    }
280
281    /// Not supported, use [`super::ManyCar`] instead.
282    fn put_keyed(&self, _: &Cid, _: &[u8]) -> anyhow::Result<()> {
283        anyhow::bail!("PlainCar is read-only, use ManyCar instead");
284    }
285}
286
287impl<ReaderT> PersistentStore for PlainCar<ReaderT>
288where
289    ReaderT: ReadAt,
290{
291    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
292        self.put_keyed(k, block)
293    }
294}
295
296pub async fn write_skip_frame_header_async(
297    mut writer: impl AsyncWrite + Unpin,
298    data_len: u32,
299) -> std::io::Result<()> {
300    writer
301        .write_all(&super::forest::ZSTD_SKIPPABLE_FRAME_MAGIC_HEADER)
302        .await?;
303    writer.write_all(&data_len.to_le_bytes()).await?;
304    Ok(())
305}
306
307fn cid_error_to_io_error(cid_error: cid::Error) -> io::Error {
308    match cid_error {
309        cid::Error::Io(io_error) => io_error,
310        other => io::Error::new(InvalidData, other),
311    }
312}
313
314/// <https://ipld.io/specs/transport/car/carv2/#header>
315/// ```text
316/// start ►│    reader end ►│
317///        ├──────┬─────────┤
318///        │pragma│v2 header│
319///        └──────┴─────────┘
320/// ```
321pub fn read_v2_header(mut reader: impl Read) -> io::Result<Option<CarV2Header>> {
322    /// <https://ipld.io/specs/transport/car/carv2/#pragma>
323    const CAR_V2_PRAGMA: [u8; 10] = [0xa1, 0x67, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x02];
324
325    let len = reader.read_fixedint::<u8>()? as usize;
326    if len == CAR_V2_PRAGMA.len() {
327        let mut buffer = vec![0; len];
328        reader.read_exact(&mut buffer)?;
329        if buffer[..] == CAR_V2_PRAGMA {
330            let mut characteristics = [0; 16];
331            reader.read_exact(&mut characteristics)?;
332            let data_offset: i64 = reader.read_fixedint()?;
333            let data_size: i64 = reader.read_fixedint()?;
334            let index_offset: i64 = reader.read_fixedint()?;
335            return Ok(Some(CarV2Header {
336                characteristics,
337                data_offset,
338                data_size,
339                index_offset,
340            }));
341        }
342    }
343    Ok(None)
344}
345
346/// ```text
347/// start ►│         reader end ►│
348///        ├───────────┬─────────┤
349///        │body length│v1 header│
350///        └───────────┴─────────┘
351/// ```
352/// Maximum CARv1 header length we will allocate a buffer for up-front. Matches
353/// go-car's `DefaultMaxAllowedHeaderSize`. Without this bound, a bad CAR
354/// could declare a huge header length via the leading varint and trigger an
355/// unbounded allocation.
356const MAX_ALLOWED_HEADER_SIZE: usize = 32 << 20; // 32 MiB
357
358#[tracing::instrument(level = "trace", skip_all, ret)]
359fn read_v1_header(mut reader: impl Read) -> io::Result<CarV1Header> {
360    let header_len = reader.read_varint()?;
361    if header_len > MAX_ALLOWED_HEADER_SIZE {
362        return Err(io::Error::new(
363            InvalidData,
364            format!(
365                "CAR header length {header_len} exceeds maximum allowed {MAX_ALLOWED_HEADER_SIZE}"
366            ),
367        ));
368    }
369    let mut buffer = vec![0; header_len];
370    reader.read_exact(&mut buffer)?;
371    let header: CarV1Header =
372        from_slice_with_fallback(&buffer).map_err(|e| io::Error::new(InvalidData, e))?;
373    if header.version == 1 {
374        Ok(header)
375    } else {
376        Err(io::Error::new(
377            Unsupported,
378            format!("unsupported CAR version {}", header.version),
379        ))
380    }
381}
382
383/// Returns ([`Cid`], the `block data offset` and `block data length`)
384/// ```text
385/// start ►│              reader end ►│
386///        ├───────────┬───┬──────────┤
387///        │body length│cid│block data│
388///        └───────────┴───┼──────────┤
389///                        │◄────────►│
390///                        │  =block data length
391///            block data  │
392///                offset ►│
393/// ```
394/// Importantly, we seek `block data length`, rather than read any in.
395/// This allows us to keep indexing fast.
396///
397/// [`Ok(None)`] on EOF
398#[tracing::instrument(level = "trace", skip_all, ret)]
399fn read_block_data_location_and_skip(
400    mut reader: impl Read + Seek,
401    limit_position: Option<u64>,
402) -> io::Result<Option<(Cid, UncompressedBlockDataLocation)>> {
403    if let Some(limit_position) = limit_position
404        && reader.stream_position()? >= limit_position
405    {
406        return Ok(None);
407    }
408    let Some(body_length) = read_varint_body_length_or_eof(&mut reader)? else {
409        return Ok(None);
410    };
411    let frame_body_offset = reader.stream_position()?;
412    let mut reader = CountRead::new(&mut reader);
413    let cid = Cid::read_bytes(&mut reader).map_err(cid_error_to_io_error)?;
414
415    // counting the read bytes saves us a syscall for finding block data offset
416    let cid_length = reader.bytes_read();
417    let block_data_offset =
418        frame_body_offset + u64::try_from(cid_length).expect("usize must fit in u64");
419    let next_frame_offset = frame_body_offset + u64::from(body_length);
420    // A malformed CAR frame can declare a `body_length` smaller than the CID it
421    // encodes, which would underflow here.
422    let block_data_length = next_frame_offset
423        .checked_sub(block_data_offset)
424        .and_then(|len| u32::try_from(len).ok())
425        .ok_or_else(|| {
426            io::Error::new(
427                InvalidData,
428                format!(
429                    "invalid CAR frame: body length ({body_length}) is smaller than the encoded CID ({cid_length} bytes)"
430                ),
431            )
432        })?;
433    reader
434        .into_inner()
435        .seek(SeekFrom::Start(next_frame_offset))?;
436    Ok(Some((
437        cid,
438        UncompressedBlockDataLocation {
439            offset: block_data_offset,
440            length: block_data_length,
441        },
442    )))
443}
444
445/// Reads `body length`, leaving the reader at the start of a varint frame,
446/// or returns [`Ok(None)`] if we've reached EOF
447/// ```text
448/// start ►│
449///        ├───────────┬─────────────┐
450///        │varint:    │             │
451///        │body length│frame body   │
452///        └───────────┼─────────────┘
453///        reader end ►│
454/// ```
455fn read_varint_body_length_or_eof(mut reader: impl Read) -> io::Result<Option<u32>> {
456    let mut byte = [0u8; 1]; // detect EOF
457    match reader.read(&mut byte)? {
458        0 => Ok(None),
459        // `Read::read` into a 1-byte buffer reads at most one byte, so any non-zero count is one.
460        _ => (byte.chain(reader)).read_varint().map(Some),
461    }
462}
463
464/// A reader that keeps track of how many bytes it has read.
465///
466/// This is useful for calculating the _block data length_ when the (_varint frame_) _body length_ is known.
467struct CountRead<ReadT> {
468    inner: ReadT,
469    count: usize,
470}
471
472impl<ReadT> CountRead<ReadT> {
473    pub fn new(inner: ReadT) -> Self {
474        Self { inner, count: 0 }
475    }
476    pub fn bytes_read(&self) -> usize {
477        self.count
478    }
479    pub fn into_inner(self) -> ReadT {
480        self.inner
481    }
482}
483
484impl<ReadT> Read for CountRead<ReadT>
485where
486    ReadT: Read,
487{
488    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
489        let n = self.inner.read(buf)?;
490        self.count += n;
491        Ok(n)
492    }
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498    use crate::utils::db::{
499        car_stream::{CarStream, CarV1Header},
500        car_util::load_car,
501    };
502    use futures::TryStreamExt as _;
503    use fvm_ipld_blockstore::{Blockstore, MemoryBlockstore};
504    use std::io::Cursor;
505    use std::sync::LazyLock;
506    use tokio::io::{AsyncBufRead, AsyncSeek, BufReader};
507    use tokio_test::block_on;
508
509    #[quickcheck_macros::quickcheck]
510    fn plain_car_new_no_panic(junk: Vec<u8>) {
511        // Reading an arbitrary/malicious byte stream must never panic (only
512        // return `Err`). Exercises `read_v2_header`, `read_varint_body_length`,
513        // and the frame-length arithmetic in `read_block_data_location_and_skip`.
514        let _ = PlainCar::new(junk);
515    }
516
517    #[test]
518    fn test_uncompressed_v1() {
519        let car = chain4_car();
520        let car_backed = PlainCar::new(car).unwrap();
521
522        assert_eq!(car_backed.version(), 1);
523        assert_eq!(car_backed.head_tipset_key().len(), 1);
524        assert_eq!(car_backed.cids().len(), 1222);
525
526        let reference_car = reference(Cursor::new(car));
527        let reference_car_zst = reference(Cursor::new(chain4_car_zst()));
528        let reference_car_zst_unsafe = reference_unsafe(chain4_car_zst());
529        for cid in car_backed.cids() {
530            let expected = reference_car.get(&cid).unwrap().unwrap();
531            let expected2 = reference_car_zst.get(&cid).unwrap().unwrap();
532            let expected3 = reference_car_zst_unsafe.get(&cid).unwrap().unwrap();
533            let mut expected4 = vec![];
534            car_backed
535                .get_reader(cid)
536                .unwrap()
537                .read_to_end(&mut expected4)
538                .unwrap();
539            let actual = car_backed.get(&cid).unwrap().unwrap();
540            assert_eq!(expected, actual);
541            assert_eq!(expected2, actual);
542            assert_eq!(expected3, actual);
543            assert_eq!(expected4, actual);
544        }
545    }
546
547    #[test]
548    fn test_uncompressed_v2() {
549        let car = carv2_car();
550        let car_backed = PlainCar::new(car).unwrap();
551
552        assert_eq!(car_backed.version(), 2);
553        assert_eq!(car_backed.head_tipset_key().len(), 1);
554        assert_eq!(car_backed.cids().len(), 7153);
555
556        let reference_car = reference(Cursor::new(car));
557        let reference_car_zst = reference(Cursor::new(carv2_car_zst()));
558        let reference_car_zst_unsafe = reference_unsafe(carv2_car_zst());
559        for cid in car_backed.cids() {
560            let expected = reference_car.get(&cid).unwrap().unwrap();
561            let expected2 = reference_car_zst.get(&cid).unwrap().unwrap();
562            let expected3 = reference_car_zst_unsafe.get(&cid).unwrap().unwrap();
563            let actual = car_backed.get(&cid).unwrap().unwrap();
564            assert_eq!(expected, actual);
565            assert_eq!(expected2, actual);
566            assert_eq!(expected3, actual);
567        }
568    }
569
570    fn reference(reader: impl AsyncBufRead + AsyncSeek + Unpin) -> MemoryBlockstore {
571        let blockstore = MemoryBlockstore::new();
572        block_on(load_car(&blockstore, reader)).unwrap();
573        blockstore
574    }
575
576    fn reference_unsafe(reader: impl AsyncBufRead + Unpin) -> MemoryBlockstore {
577        let blockstore = MemoryBlockstore::new();
578        block_on(load_car_unsafe(&blockstore, reader)).unwrap();
579        blockstore
580    }
581
582    pub async fn load_car_unsafe<R>(db: &impl Blockstore, reader: R) -> anyhow::Result<CarV1Header>
583    where
584        R: AsyncBufRead + Unpin,
585    {
586        let mut stream = CarStream::new_unsafe(BufReader::new(reader)).await?;
587        while let Some(block) = stream.try_next().await? {
588            db.put_keyed(&block.cid, &block.data)?;
589        }
590        Ok(stream.header_v1)
591    }
592
593    fn chain4_car_zst() -> &'static [u8] {
594        include_bytes!("../../../test-snapshots/chain4.car.zst")
595    }
596
597    fn chain4_car() -> &'static [u8] {
598        static CAR: LazyLock<Vec<u8>> =
599            LazyLock::new(|| zstd::decode_all(chain4_car_zst()).unwrap());
600        CAR.as_slice()
601    }
602
603    fn carv2_car_zst() -> &'static [u8] {
604        include_bytes!("../../../test-snapshots/carv2.car.zst")
605    }
606
607    fn carv2_car() -> &'static [u8] {
608        static CAR: LazyLock<Vec<u8>> =
609            LazyLock::new(|| zstd::decode_all(carv2_car_zst()).unwrap());
610        CAR.as_slice()
611    }
612}
forest/db/car/plain.rs

forest/db/car/
plain.rs
