Something went wrong. Try again.
A public mirror for the whole atmosphere hubble.microcosm.blue
Something went wrong. Try again.
3.8 kB · 119 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120//! minimal CARv1 parser for atproto firehose commits//!//! Blocks are yielded as slices borrowed from the input, so nothing is copied.//! This is not a general CAR implementation, it only needs to work with CARs//! from atproto firehose commits.//!//! bleh. i would not like to keep this but iroh-car staying committed to async//! is unfortunate. probably should ship a proper sans-io carv1 reader...
use serde::Deserialize;
use crate::cid::{Cid, CidParseError};
#[derive(Debug, thiserror::Error)]pub enum CarError { #[error("truncated: wanted {wanted} more bytes, {available} left")] Truncated { wanted: usize, available: usize }, #[error("length prefix is not a valid varint")] BadVarint, #[error("header did not decode: {0}")] BadHeader(String), #[error("CAR version {0}, only v1 is supported")] UnsupportedVersion(u64), #[error("CAR header lists {0} roots, expected at least 1")] RootCount(usize), #[error("invalid CID in block header: {0}")] BadCid(#[from] CidParseError),}
#[derive(Deserialize)]struct Header { roots: Vec<Cid>, version: u64,}
/// Reads the blocks of a CARv1 archive in the order they were written.pub struct CarReader<'a> { root: Cid, rest: &'a [u8],}
impl<'a> CarReader<'a> { /// Parse the header, leaving the blocks to be read by iterating. pub fn new(bytes: &'a [u8]) -> Result<Self, CarError> { let (header, rest) = take_chunk(bytes)?; let header: Header = dasl::drisl::from_slice(header).map_err(|e| CarError::BadHeader(e.to_string()))?; if header.version != 1 { return Err(CarError::UnsupportedVersion(header.version)); } let Some(root) = header.roots.first() else { return Err(CarError::RootCount(header.roots.len())); }; Ok(Self { root: *root, rest }) }
/// The archive's single root: for a commit CAR, the commit object. pub fn root(&self) -> Cid { self.root }}
impl<'a> Iterator for CarReader<'a> { type Item = Result<(Cid, &'a [u8]), CarError>;
fn next(&mut self) -> Option<Self::Item> { if self.rest.is_empty() { return None; } Some(self.read_block()) }}
impl<'a> CarReader<'a> { fn read_block(&mut self) -> Result<(Cid, &'a [u8]), CarError> { let (chunk, rest) = take_chunk(self.rest)?; self.rest = rest; // a block is its CID followed by its bytes, with no length between // them, so the CID's own length has to be worked out from its prefix let cid_len = cid_len(chunk)?; let (cid, block) = chunk.split_at(cid_len); Ok((Cid::from_bytes_raw(cid)?, block)) }}
/// Split off one varint-length-prefixed chunk.fn take_chunk(bytes: &[u8]) -> Result<(&[u8], &[u8]), CarError> { let (len, rest) = unsigned_varint::decode::usize(bytes).map_err(|_| CarError::BadVarint)?; if rest.len() < len { return Err(CarError::Truncated { wanted: len, available: rest.len(), }); } Ok(rest.split_at(len))}
/// Length of the CID at the head of a block frame.////// atproto only uses CIDv1 with a single-byte codec and a single-byte/// multihash code, so this is the 4-byte prefix plus the declared digest/// length. Anything else is not a CID we can read anyway.fn cid_len(bytes: &[u8]) -> Result<usize, CarError> { const PREFIX: usize = 4; let Some(prefix) = bytes.get(..PREFIX) else { return Err(CarError::Truncated { wanted: PREFIX, available: bytes.len(), }); }; let len = PREFIX + usize::from(prefix[3]); if bytes.len() < len { return Err(CarError::Truncated { wanted: len, available: bytes.len(), }); } Ok(len)}