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 metadata(&self) -> Option<&FilecoinSnapshotMetadata> {
177        self.metadata
178            .get_or_init(|| {
179                if self.header_v1.roots.len() == super::V2_SNAPSHOT_ROOT_COUNT {
180                    let maybe_metadata_cid = self.header_v1.roots.first();
181                    if let Ok(Some(metadata)) =
182                        self.get_cbor::<FilecoinSnapshotMetadata>(maybe_metadata_cid)
183                    {
184                        return Some(metadata);
185                    }
186                }
187                None
188            })
189            .as_ref()
190    }
191
192    pub fn head_tipset_key(&self) -> &NonEmpty<Cid> {
193        // head tipset key is stored in v2 snapshot metadata
194        // See <https://github.com/filecoin-project/FIPs/blob/98e33b9fa306959aa0131519eb4cc155522b2081/FRCs/frc-0108.md#v2-specification>
195        if let Some(metadata) = self.metadata() {
196            &metadata.head_tipset_key
197        } else {
198            &self.header_v1.roots
199        }
200    }
201
202    pub fn version(&self) -> u64 {
203        self.version
204    }
205
206    pub fn heaviest_tipset_key(&self) -> TipsetKey {
207        TipsetKey::from(self.head_tipset_key().clone())
208    }
209
210    pub fn heaviest_tipset(&self) -> anyhow::Result<Tipset> {
211        Tipset::load_required(self, &self.heaviest_tipset_key())
212    }
213
214    /// In an arbitrary order
215    #[cfg(test)]
216    pub fn cids(&self) -> Vec<Cid> {
217        self.index.read().keys().collect()
218    }
219
220    pub fn into_dyn(self) -> PlainCar<Box<dyn super::RandomAccessFileReader>> {
221        PlainCar {
222            reader: Box::new(self.reader),
223            index: self.index,
224            version: self.version,
225            header_v1: self.header_v1,
226            header_v2: self.header_v2,
227            metadata: self.metadata,
228        }
229    }
230
231    /// Gets a reader of the block data by its `Cid`
232    pub fn get_reader(&self, k: Cid) -> Option<impl Read> {
233        self.index
234            .read()
235            .get(&k)
236            .map(|UncompressedBlockDataLocation { offset, length }| {
237                positioned_io::Cursor::new_pos(&self.reader, *offset).take(u64::from(*length))
238            })
239    }
240}
241
242impl TryFrom<&'static [u8]> for PlainCar<&'static [u8]> {
243    type Error = io::Error;
244    fn try_from(bytes: &'static [u8]) -> io::Result<Self> {
245        PlainCar::new(bytes)
246    }
247}
248
249/// If you seek to `offset` (from the start of the file), and read `length` bytes,
250/// you should get data that corresponds to a [`Cid`] (but NOT the [`Cid`] itself).
251#[derive(Debug, serde::Serialize, serde::Deserialize)]
252pub struct UncompressedBlockDataLocation {
253    offset: u64,
254    length: u32,
255}
256
257impl<ReaderT> Blockstore for PlainCar<ReaderT>
258where
259    ReaderT: ReadAt,
260{
261    #[tracing::instrument(level = "trace", skip(self))]
262    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
263        match self.index.read().get(k) {
264            Some(UncompressedBlockDataLocation { offset, length }) => {
265                trace!("fetching from disk");
266                let mut data = vec![0; usize::try_from(*length).unwrap()];
267                self.reader.read_exact_at(*offset, &mut data)?;
268                Ok(Some(data))
269            }
270            None => {
271                trace!("not found");
272                Ok(None)
273            }
274        }
275    }
276
277    /// Not supported, use [`super::ManyCar`] instead.
278    fn put_keyed(&self, _: &Cid, _: &[u8]) -> anyhow::Result<()> {
279        unreachable!("PlainCar is read-only, use ManyCar instead");
280    }
281}
282
283impl<ReaderT> PersistentStore for PlainCar<ReaderT>
284where
285    ReaderT: ReadAt,
286{
287    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
288        self.put_keyed(k, block)
289    }
290}
291
292pub async fn write_skip_frame_header_async(
293    mut writer: impl AsyncWrite + Unpin,
294    data_len: u32,
295) -> std::io::Result<()> {
296    writer
297        .write_all(&super::forest::ZSTD_SKIPPABLE_FRAME_MAGIC_HEADER)
298        .await?;
299    writer.write_all(&data_len.to_le_bytes()).await?;
300    Ok(())
301}
302
303fn cid_error_to_io_error(cid_error: cid::Error) -> io::Error {
304    match cid_error {
305        cid::Error::Io(io_error) => io_error,
306        other => io::Error::new(InvalidData, other),
307    }
308}
309
310/// <https://ipld.io/specs/transport/car/carv2/#header>
311/// ```text
312/// start ►│    reader end ►│
313///        ├──────┬─────────┤
314///        │pragma│v2 header│
315///        └──────┴─────────┘
316/// ```
317pub fn read_v2_header(mut reader: impl Read) -> io::Result<Option<CarV2Header>> {
318    /// <https://ipld.io/specs/transport/car/carv2/#pragma>
319    const CAR_V2_PRAGMA: [u8; 10] = [0xa1, 0x67, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x02];
320
321    let len = reader.read_fixedint::<u8>()? as usize;
322    if len == CAR_V2_PRAGMA.len() {
323        let mut buffer = vec![0; len];
324        reader.read_exact(&mut buffer)?;
325        if buffer[..] == CAR_V2_PRAGMA {
326            let mut characteristics = [0; 16];
327            reader.read_exact(&mut characteristics)?;
328            let data_offset: i64 = reader.read_fixedint()?;
329            let data_size: i64 = reader.read_fixedint()?;
330            let index_offset: i64 = reader.read_fixedint()?;
331            return Ok(Some(CarV2Header {
332                characteristics,
333                data_offset,
334                data_size,
335                index_offset,
336            }));
337        }
338    }
339    Ok(None)
340}
341
342/// ```text
343/// start ►│         reader end ►│
344///        ├───────────┬─────────┤
345///        │body length│v1 header│
346///        └───────────┴─────────┘
347/// ```
348/// Maximum CARv1 header length we will allocate a buffer for up-front. Matches
349/// go-car's `DefaultMaxAllowedHeaderSize`. Without this bound, a bad CAR
350/// could declare a huge header length via the leading varint and trigger an
351/// unbounded allocation.
352const MAX_ALLOWED_HEADER_SIZE: usize = 32 << 20; // 32 MiB
353
354#[tracing::instrument(level = "trace", skip_all, ret)]
355fn read_v1_header(mut reader: impl Read) -> io::Result<CarV1Header> {
356    let header_len = reader.read_varint()?;
357    if header_len > MAX_ALLOWED_HEADER_SIZE {
358        return Err(io::Error::new(
359            InvalidData,
360            format!(
361                "CAR header length {header_len} exceeds maximum allowed {MAX_ALLOWED_HEADER_SIZE}"
362            ),
363        ));
364    }
365    let mut buffer = vec![0; header_len];
366    reader.read_exact(&mut buffer)?;
367    let header: CarV1Header =
368        from_slice_with_fallback(&buffer).map_err(|e| io::Error::new(InvalidData, e))?;
369    if header.version == 1 {
370        Ok(header)
371    } else {
372        Err(io::Error::new(
373            Unsupported,
374            format!("unsupported CAR version {}", header.version),
375        ))
376    }
377}
378
379/// Returns ([`Cid`], the `block data offset` and `block data length`)
380/// ```text
381/// start ►│              reader end ►│
382///        ├───────────┬───┬──────────┤
383///        │body length│cid│block data│
384///        └───────────┴───┼──────────┤
385///                        │◄────────►│
386///                        │  =block data length
387///            block data  │
388///                offset ►│
389/// ```
390/// Importantly, we seek `block data length`, rather than read any in.
391/// This allows us to keep indexing fast.
392///
393/// [`Ok(None)`] on EOF
394#[tracing::instrument(level = "trace", skip_all, ret)]
395fn read_block_data_location_and_skip(
396    mut reader: impl Read + Seek,
397    limit_position: Option<u64>,
398) -> io::Result<Option<(Cid, UncompressedBlockDataLocation)>> {
399    if let Some(limit_position) = limit_position
400        && reader.stream_position()? >= limit_position
401    {
402        return Ok(None);
403    }
404    let Some(body_length) = read_varint_body_length_or_eof(&mut reader)? else {
405        return Ok(None);
406    };
407    let frame_body_offset = reader.stream_position()?;
408    let mut reader = CountRead::new(&mut reader);
409    let cid = Cid::read_bytes(&mut reader).map_err(cid_error_to_io_error)?;
410
411    // counting the read bytes saves us a syscall for finding block data offset
412    let cid_length = reader.bytes_read();
413    let block_data_offset = frame_body_offset + u64::try_from(cid_length).unwrap();
414    let next_frame_offset = frame_body_offset + u64::from(body_length);
415    // A malformed CAR frame can declare a `body_length` smaller than the CID it
416    // encodes, which would underflow here.
417    let block_data_length = next_frame_offset
418        .checked_sub(block_data_offset)
419        .and_then(|len| u32::try_from(len).ok())
420        .ok_or_else(|| {
421            io::Error::new(
422                InvalidData,
423                format!(
424                    "invalid CAR frame: body length ({body_length}) is smaller than the encoded CID ({cid_length} bytes)"
425                ),
426            )
427        })?;
428    reader
429        .into_inner()
430        .seek(SeekFrom::Start(next_frame_offset))?;
431    Ok(Some((
432        cid,
433        UncompressedBlockDataLocation {
434            offset: block_data_offset,
435            length: block_data_length,
436        },
437    )))
438}
439
440/// Reads `body length`, leaving the reader at the start of a varint frame,
441/// or returns [`Ok(None)`] if we've reached EOF
442/// ```text
443/// start ►│
444///        ├───────────┬─────────────┐
445///        │varint:    │             │
446///        │body length│frame body   │
447///        └───────────┼─────────────┘
448///        reader end ►│
449/// ```
450fn read_varint_body_length_or_eof(mut reader: impl Read) -> io::Result<Option<u32>> {
451    let mut byte = [0u8; 1]; // detect EOF
452    match reader.read(&mut byte)? {
453        0 => Ok(None),
454        1 => (byte.chain(reader)).read_varint().map(Some),
455        _ => unreachable!(),
456    }
457}
458
459/// A reader that keeps track of how many bytes it has read.
460///
461/// This is useful for calculating the _block data length_ when the (_varint frame_) _body length_ is known.
462struct CountRead<ReadT> {
463    inner: ReadT,
464    count: usize,
465}
466
467impl<ReadT> CountRead<ReadT> {
468    pub fn new(inner: ReadT) -> Self {
469        Self { inner, count: 0 }
470    }
471    pub fn bytes_read(&self) -> usize {
472        self.count
473    }
474    pub fn into_inner(self) -> ReadT {
475        self.inner
476    }
477}
478
479impl<ReadT> Read for CountRead<ReadT>
480where
481    ReadT: Read,
482{
483    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
484        let n = self.inner.read(buf)?;
485        self.count += n;
486        Ok(n)
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493    use crate::utils::db::{
494        car_stream::{CarStream, CarV1Header},
495        car_util::load_car,
496    };
497    use futures::TryStreamExt as _;
498    use fvm_ipld_blockstore::{Blockstore, MemoryBlockstore};
499    use std::io::Cursor;
500    use std::sync::LazyLock;
501    use tokio::io::{AsyncBufRead, AsyncSeek, BufReader};
502    use tokio_test::block_on;
503
504    #[quickcheck_macros::quickcheck]
505    fn plain_car_new_no_panic(junk: Vec<u8>) {
506        // Reading an arbitrary/malicious byte stream must never panic (only
507        // return `Err`). Exercises `read_v2_header`, `read_varint_body_length`,
508        // and the frame-length arithmetic in `read_block_data_location_and_skip`.
509        let _ = PlainCar::new(junk);
510    }
511
512    #[test]
513    fn test_uncompressed_v1() {
514        let car = chain4_car();
515        let car_backed = PlainCar::new(car).unwrap();
516
517        assert_eq!(car_backed.version(), 1);
518        assert_eq!(car_backed.head_tipset_key().len(), 1);
519        assert_eq!(car_backed.cids().len(), 1222);
520
521        let reference_car = reference(Cursor::new(car));
522        let reference_car_zst = reference(Cursor::new(chain4_car_zst()));
523        let reference_car_zst_unsafe = reference_unsafe(chain4_car_zst());
524        for cid in car_backed.cids() {
525            let expected = reference_car.get(&cid).unwrap().unwrap();
526            let expected2 = reference_car_zst.get(&cid).unwrap().unwrap();
527            let expected3 = reference_car_zst_unsafe.get(&cid).unwrap().unwrap();
528            let mut expected4 = vec![];
529            car_backed
530                .get_reader(cid)
531                .unwrap()
532                .read_to_end(&mut expected4)
533                .unwrap();
534            let actual = car_backed.get(&cid).unwrap().unwrap();
535            assert_eq!(expected, actual);
536            assert_eq!(expected2, actual);
537            assert_eq!(expected3, actual);
538            assert_eq!(expected4, actual);
539        }
540    }
541
542    #[test]
543    fn test_uncompressed_v2() {
544        let car = carv2_car();
545        let car_backed = PlainCar::new(car).unwrap();
546
547        assert_eq!(car_backed.version(), 2);
548        assert_eq!(car_backed.head_tipset_key().len(), 1);
549        assert_eq!(car_backed.cids().len(), 7153);
550
551        let reference_car = reference(Cursor::new(car));
552        let reference_car_zst = reference(Cursor::new(carv2_car_zst()));
553        let reference_car_zst_unsafe = reference_unsafe(carv2_car_zst());
554        for cid in car_backed.cids() {
555            let expected = reference_car.get(&cid).unwrap().unwrap();
556            let expected2 = reference_car_zst.get(&cid).unwrap().unwrap();
557            let expected3 = reference_car_zst_unsafe.get(&cid).unwrap().unwrap();
558            let actual = car_backed.get(&cid).unwrap().unwrap();
559            assert_eq!(expected, actual);
560            assert_eq!(expected2, actual);
561            assert_eq!(expected3, actual);
562        }
563    }
564
565    fn reference(reader: impl AsyncBufRead + AsyncSeek + Unpin) -> MemoryBlockstore {
566        let blockstore = MemoryBlockstore::new();
567        block_on(load_car(&blockstore, reader)).unwrap();
568        blockstore
569    }
570
571    fn reference_unsafe(reader: impl AsyncBufRead + Unpin) -> MemoryBlockstore {
572        let blockstore = MemoryBlockstore::new();
573        block_on(load_car_unsafe(&blockstore, reader)).unwrap();
574        blockstore
575    }
576
577    pub async fn load_car_unsafe<R>(db: &impl Blockstore, reader: R) -> anyhow::Result<CarV1Header>
578    where
579        R: AsyncBufRead + Unpin,
580    {
581        let mut stream = CarStream::new_unsafe(BufReader::new(reader)).await?;
582        while let Some(block) = stream.try_next().await? {
583            db.put_keyed(&block.cid, &block.data)?;
584        }
585        Ok(stream.header_v1)
586    }
587
588    fn chain4_car_zst() -> &'static [u8] {
589        include_bytes!("../../../test-snapshots/chain4.car.zst")
590    }
591
592    fn chain4_car() -> &'static [u8] {
593        static CAR: LazyLock<Vec<u8>> =
594            LazyLock::new(|| zstd::decode_all(chain4_car_zst()).unwrap());
595        CAR.as_slice()
596    }
597
598    fn carv2_car_zst() -> &'static [u8] {
599        include_bytes!("../../../test-snapshots/carv2.car.zst")
600    }
601
602    fn carv2_car() -> &'static [u8] {
603        static CAR: LazyLock<Vec<u8>> =
604            LazyLock::new(|| zstd::decode_all(carv2_car_zst()).unwrap());
605        CAR.as_slice()
606    }
607}
forest/db/car/plain.rs

forest/db/car/
plain.rs
