Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972use std::collections::{BTreeMap, HashMap};use std::io::Cursor;
use bytes::{Buf, Bytes, BytesMut};use cid::Cid as IpldCid;use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256};use miette::{IntoDiagnostic, Result};use sha2::{Digest, Sha256};
/// CIDv1 (dag-cbor, sha2-256) of a raw DAG-CBOR block.pub(crate) fn compute_cid(bytes: &[u8]) -> IpldCid { let digest = Sha256::digest(bytes); let mh = cid::multihash::Multihash::<64>::wrap(SHA2_256, &digest) .expect("sha2-256 digest fits a 64-byte multihash"); IpldCid::new_v1(DAG_CBOR, mh)}
/// blocks of a whole-repo CAR. hashed rather than ordered: the MST walk looks up every/// block, and cids are chosen by the PDS, so the hasher is randomly seeded.pub(crate) type CarBlocks = HashMap<IpldCid, Bytes, ahash::RandomState>;
/// blocks of a commit-sized car (firehose `#commit` and `#sync`, `getRecord` proofs,/// `getBlocks`), ordered: the shape jacquard's `MemoryBlockStore` takes.pub(crate) type CommitBlocks = BTreeMap<IpldCid, Bytes>;
pub(crate) type CommitCar = ParsedCar<CommitBlocks>;
/// repo blocks average a few hundred bytes; sizing for that avoids most rehashing.const EXPECTED_SECTION_BYTES: usize = 256;
/// the biggest block a streamed car may hold. a stream keeps a block whole until it ends, so/// without a cap a pds could park the whole car in memory as one block, past a spill. the/// firehose stops records at 1 MiB but sends bigger updates as a #sync, which ends up here, so/// this leaves some room past that and still turns away a broken carconst MAX_STREAMED_SECTION_BYTES: usize = 2 * 1024 * 1024;
#[derive(Debug)]pub(crate) struct ParsedCar<M> { pub root: IpldCid, pub blocks: Blocks<M>,}
/// a car's blocks by cid, and whether the parse hashed every one of them against its cid. only/// a verifying parse sets that, and only other `Blocks` can be added afterwards, so a block/// handed out as checked was checked.#[derive(Debug)]pub(crate) struct Blocks<M> { map: M, cids_checked: bool,}
/// the lookup [`Blocks`] needs from the map behind it.pub(crate) trait BlockMap { fn get(&self, cid: &IpldCid) -> Option<&Bytes>;}
impl<S: std::hash::BuildHasher> BlockMap for HashMap<IpldCid, Bytes, S> { fn get(&self, cid: &IpldCid) -> Option<&Bytes> { HashMap::get(self, cid) }}
impl BlockMap for CommitBlocks { fn get(&self, cid: &IpldCid) -> Option<&Bytes> { BTreeMap::get(self, cid) }}
impl<M: BlockMap> Blocks<M> { /// blocks nothing has checked yet; each one is hashed when it is verified. #[cfg(test)] pub(crate) fn unverified(map: M) -> Self { Self { map, cids_checked: false, } }
/// the bytes under `cid`, for reading them; storing them takes [`Blocks::block`]. pub(crate) fn get(&self, cid: &IpldCid) -> Option<&Bytes> { self.map.get(cid) }
#[cfg_attr(not(feature = "indexer"), allow(dead_code))] pub(crate) fn block(&self, cid: &IpldCid) -> Option<CarBlock> { self.map.get(cid).map(|bytes| CarBlock { cid: *cid, bytes: bytes.clone(), cid_checked: self.cids_checked, }) }
pub(crate) fn map(&self) -> &M { &self.map }}
/// no blocks, so every one of them is checked.impl<M: Default> Default for Blocks<M> { fn default() -> Self { Self { map: M::default(), cids_checked: true, } }}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl Blocks<CommitBlocks> { /// adds `other`'s blocks, which replace any with the same cid. the result is checked only /// if both sides were. pub(crate) fn extend(&mut self, other: Self) { self.map.extend(other.map); self.cids_checked &= other.cids_checked; }}
/// a block as a car delivered it. its bytes are only reachable through [`CarBlock::verify`],/// which costs a hash only if the parse did not check it already.#[cfg_attr(not(feature = "indexer"), allow(dead_code))]#[derive(Debug)]pub(crate) struct CarBlock { cid: IpldCid, bytes: Bytes, cid_checked: bool,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl CarBlock { /// a block for `bytes`, under the cid they hash to. #[cfg(test)] pub(crate) fn from_body(bytes: impl Into<Bytes>) -> Self { let bytes = bytes.into(); let cid = compute_cid(&bytes); Self { cid, bytes, cid_checked: true, } }
/// a block whose bytes were just hashed against `cid` pub(crate) fn hashed(cid: IpldCid, bytes: Bytes) -> Self { Self { cid, bytes, cid_checked: true, } }
/// how many bytes the block holds, which needs no check pub(crate) fn size(&self) -> usize { self.bytes.len() }
/// the cid the car claims for this block. #[cfg(test)] pub(crate) fn cid(&self) -> &IpldCid { &self.cid }
/// the same block, hashed now if the parse didn't, so verifying it later costs nothing pub(crate) fn checked(self) -> Result<Self> { self.verify()?; Ok(Self { cid_checked: true, ..self }) }
pub(crate) fn verify(&self) -> Result<VerifiedBlock<'_>> { if !self.cid_checked { validate_block_cid(&self.cid, &self.bytes)?; } Ok(VerifiedBlock { cid: &self.cid, bytes: &self.bytes, }) }}
/// a block whose bytes hash to its cid (sha2-256, dag-cbor). record storage reads bodies only/// through one: events and history find bodies by cid, so a body kept under a cid it does not/// hash to would be served as that record to every reader.#[cfg_attr(not(feature = "indexer"), allow(dead_code))]#[derive(Debug)]pub(crate) struct VerifiedBlock<'a> { cid: &'a IpldCid, bytes: &'a Bytes,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl<'a> VerifiedBlock<'a> { pub(crate) fn cid(&self) -> &'a IpldCid { self.cid }
pub(crate) fn bytes(&self) -> &'a Bytes { self.bytes }}
// backfill streams whole cars with WholeCar, so only tests and the benches parse one in a piece#[cfg_attr(not(test), allow(dead_code))]pub(crate) fn parse_car(data: Bytes, verify_cids: bool) -> Result<ParsedCar<CarBlocks>> { let (root, blocks_start) = car_root(&data)?; let mut map = CarBlocks::with_capacity_and_hasher( data.len() / EXPECTED_SECTION_BYTES, ahash::RandomState::new(), ); for_each_block(&data, blocks_start, verify_cids, |cid, payload| { map.insert(cid, payload); })?; let blocks = Blocks { map, cids_checked: verify_cids, };
Ok(ParsedCar { root, blocks })}
pub(crate) fn parse_commit_car(data: Bytes, verify_cids: bool) -> Result<CommitCar> { let (root, blocks_start) = car_root(&data)?; let blocks = ordered_blocks(&data, blocks_start, verify_cids)?; Ok(ParsedCar { root, blocks })}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) fn parse_car_blocks(data: Bytes, verify_cids: bool) -> Result<Blocks<CommitBlocks>> { let (_, blocks_start) = car_header_bounds(&data)?; ordered_blocks(&data, blocks_start, verify_cids)}
fn ordered_blocks( data: &Bytes, blocks_start: usize, verify_cids: bool,) -> Result<Blocks<CommitBlocks>> { let mut map = BTreeMap::new(); for_each_block(data, blocks_start, verify_cids, |cid, payload| { map.insert(cid, payload); })?; Ok(Blocks { map, cids_checked: verify_cids, })}
fn car_root(data: &[u8]) -> Result<(IpldCid, usize)> { let (header_start, blocks_start) = car_header_bounds(data)?; Ok(( header_root(&data[header_start..blocks_start])?, blocks_start, ))}
#[derive(serde::Serialize, serde::Deserialize)]pub(crate) struct CarHeader { roots: Vec<IpldCid>, version: u64,}
impl CarHeader { #[cfg_attr(not(feature = "indexer"), allow(dead_code))] pub(crate) fn new_v1(roots: Vec<IpldCid>) -> Self { Self { roots, version: 1 } }
#[cfg_attr(not(feature = "indexer"), allow(dead_code))] pub(crate) fn encode(&self) -> std::io::Result<Vec<u8>> { serde_ipld_dagcbor::to_vec(self).map_err(std::io::Error::other) }}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) struct CarWriter<W> { header: Option<CarHeader>, writer: W, cid_buf: Vec<u8>,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl<W: tokio::io::AsyncWrite + Unpin> CarWriter<W> { pub(crate) fn new(header: CarHeader, writer: W) -> Self { Self { header: Some(header), writer, cid_buf: Vec::new(), } }
pub(crate) async fn write(&mut self, cid: IpldCid, data: impl AsRef<[u8]>) -> std::io::Result<()> { use tokio::io::AsyncWriteExt as _; if let Some(header) = self.header.take() { let header_bytes = header.encode()?; let mut varint = [0u8; 10]; let n = encode_uvarint(header_bytes.len(), &mut varint); self.writer.write_all(&varint[..n]).await?; self.writer.write_all(&header_bytes).await?; } self.cid_buf.clear(); cid.write_bytes(&mut self.cid_buf) .map_err(std::io::Error::other)?; let data = data.as_ref(); let mut varint = [0u8; 10]; let n = encode_uvarint(self.cid_buf.len() + data.len(), &mut varint); self.writer.write_all(&varint[..n]).await?; self.writer.write_all(&self.cid_buf).await?; self.writer.write_all(data).await?; Ok(()) }
pub(crate) async fn finish(mut self) -> std::io::Result<W> { use tokio::io::AsyncWriteExt as _; self.writer.flush().await?; Ok(self.writer) }}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]fn encode_uvarint(mut value: usize, buf: &mut [u8; 10]) -> usize { let mut i = 0; while value >= 0x80 { buf[i] = (value as u8) | 0x80; value >>= 7; i += 1; } buf[i] = value as u8; i + 1}
fn header_root(header: &[u8]) -> Result<IpldCid> { let header: CarHeader = serde_ipld_dagcbor::from_slice(header).into_diagnostic()?; if header.version != 1 { miette::bail!("Only CAR file version 1 is supported"); } header .roots .first() .copied() .ok_or_else(|| miette::miette!("CAR file has no roots"))}
fn car_header_bounds(data: &[u8]) -> Result<(usize, usize)> { let mut offset = 0; let Some(header_len) = read_uvarint(data, &mut offset)? else { return Err(miette::miette!("empty CAR file")); }; let header_start = offset; let header_end = header_start .checked_add(header_len) .ok_or_else(|| miette::miette!("CAR header length overflow"))?; if header_end > data.len() { return Err(miette::miette!("truncated CAR header")); }
Ok((header_start, header_end))}
/// yields each section's cid and zero-copy payload in file order. with `verify_cids`, every/// section is hashed as it is read, while its bytes are still in cache; checking a finished/// map afterwards would walk memory in cid order.fn for_each_block( data: &Bytes, mut offset: usize, verify_cids: bool, mut on_block: impl FnMut(IpldCid, Bytes),) -> Result<()> { while let Some(section_len) = read_uvarint(data, &mut offset)? { let section_end = offset .checked_add(section_len) .ok_or_else(|| miette::miette!("CAR block length overflow"))?; if section_end > data.len() { return Err(miette::miette!("truncated CAR block")); }
let section = data.slice(offset..section_end); offset = section_end;
let (cid, block_start) = split_section(§ion)?; let payload = section.slice(block_start..); if verify_cids { validate_block_cid(&cid, &payload)?; } on_block(cid, payload); }
Ok(())}
/// a section's cid, and where its payload startsfn split_section(section: &[u8]) -> Result<(IpldCid, usize)> { let mut cursor = Cursor::new(section); let cid = IpldCid::read_bytes(&mut cursor).into_diagnostic()?; let block_start = cursor.position() as usize; if block_start >= section.len() { return Err(miette::miette!("CAR block has no payload for {cid}")); } Ok((cid, block_start))}
/// a car read as its chunks arrive, so a caller can copy out the blocks it wants without the/// whole file ever being in memory as one piece. besides what the caller copies out it holds/// one unfinished section at most. blocks come out in file order, hashed against their cids when/// `verify_cids` is set, the same as [`parse_car`].#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) struct CarStream { pending: BytesMut, root: Option<IpldCid>, verify_cids: bool,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl CarStream { pub(crate) fn new(verify_cids: bool) -> Self { Self { pending: BytesMut::new(), root: None, verify_cids, } }
/// adds the next chunk and hands each block it completes to `on_block`, with the car's /// root, since the header always comes first pub(crate) fn push( &mut self, chunk: &[u8], mut on_block: impl FnMut(&IpldCid, IpldCid, &[u8]) -> Result<()>, ) -> Result<()> { self.pending.extend_from_slice(chunk); let mut consumed = 0; while let Some((len, len_bytes)) = try_read_uvarint(&self.pending[consumed..])? { if len > MAX_STREAMED_SECTION_BYTES { miette::bail!("CAR section of {len} bytes is over {MAX_STREAMED_SECTION_BYTES}"); } let start = consumed + len_bytes; let end = start .checked_add(len) .ok_or_else(|| miette::miette!("CAR section length overflow"))?; if end > self.pending.len() { break; } let section = &self.pending[start..end]; match self.root { None => self.root = Some(header_root(section)?), Some(root) => { let (cid, block_start) = split_section(section)?; let payload = §ion[block_start..]; if self.verify_cids { validate_block_cid(&cid, payload)?; } on_block(&root, cid, payload)?; } } consumed = end; } self.pending.advance(consumed); Ok(()) }
/// hashes every block from here on against its cid, whatever the car was made with pub(crate) fn check_cids(&mut self) { self.verify_cids = true; }
/// ends the car, with `kept` holding the blocks the caller copied out of it. a car that /// stops partway through its header or a block is truncated, not just short pub(crate) fn finish_with<M>(self, kept: M) -> Result<ParsedCar<M>> { let root = match (self.root, self.pending.is_empty()) { (None, true) => return Err(miette::miette!("empty CAR file")), (None, false) => return Err(miette::miette!("truncated CAR header")), (Some(_), false) => return Err(miette::miette!("truncated CAR block")), (Some(root), true) => root, }; Ok(ParsedCar { root, blocks: Blocks { map: kept, cids_checked: self.verify_cids, }, }) }}
/// a whole car read as its chunks arrive, every block copied out on its own. the body is never/// held in one piece, and blocks the mst walk leaves behind are freed with the map instead of/// pinning the body until the last record is stored#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) struct WholeCar { stream: CarStream, blocks: CarBlocks,}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]impl WholeCar { pub(crate) fn new(verify_cids: bool) -> Self { Self { stream: CarStream::new(verify_cids), blocks: CarBlocks::default(), } }
/// makes room for a car of about `car_bytes`, so the map isn't rehashed as it fills pub(crate) fn reserve(&mut self, car_bytes: usize) { self.blocks.reserve(car_bytes / EXPECTED_SECTION_BYTES); }
pub(crate) fn push(&mut self, chunk: &[u8]) -> Result<()> { let blocks = &mut self.blocks; self.stream.push(chunk, |_, cid, bytes| { blocks.insert(cid, Bytes::copy_from_slice(bytes)); Ok(()) }) }
pub(crate) fn finish(self) -> Result<ParsedCar<CarBlocks>> { self.stream.finish_with(self.blocks) }
/// the stream partway through the car, and the blocks it has read so far pub(crate) fn into_parts(self) -> (CarStream, CarBlocks) { (self.stream, self.blocks) }}
/// checks that a block's payload hashes to its claimed CID (sha2-256, dag-cbor).pub(crate) fn validate_block_cid(claimed_cid: &IpldCid, bytes: &[u8]) -> Result<()> { let computed_cid = compute_cid(bytes); if computed_cid != *claimed_cid { return Err(miette::miette!( "CAR block CID mismatch: claimed {claimed_cid}, computed {computed_cid}" )); } Ok(())}
/// a uvarint at the start of `data` and how many bytes it took, or `None` while `data` ends/// before it doesfn try_read_uvarint(data: &[u8]) -> Result<Option<(usize, usize)>> { let mut value = 0u64; for (idx, byte) in data.iter().take(10).enumerate() { value |= u64::from(byte & 0x7f) << (7 * idx); if byte & 0x80 == 0 { return Ok(Some((usize::try_from(value).into_diagnostic()?, idx + 1))); } } if data.len() >= 10 { return Err(miette::miette!("uvarint overflow in CAR file")); } Ok(None)}
fn read_uvarint(data: &[u8], offset: &mut usize) -> Result<Option<usize>> { let rest = &data[(*offset).min(data.len())..]; if rest.is_empty() { return Ok(None); } let (value, len) = try_read_uvarint(rest)?.ok_or_else(|| miette::miette!("truncated uvarint in CAR file"))?; *offset += len; Ok(Some(value))}
#[cfg(test)]mod tests { use super::*; use cid::multihash::Multihash; use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256};
fn cid(byte: u8) -> IpldCid { let hash = [byte; 32]; let mh = Multihash::<64>::wrap(SHA2_256, &hash).unwrap(); IpldCid::new_v1(DAG_CBOR, mh) }
#[tokio::test] async fn parses_rootless_block_car() { let cid = cid(1); let mut buf = Vec::new(); let header = CarHeader::new_v1(Vec::new()); let mut writer = CarWriter::new(header, &mut buf); writer.write(cid, b"block".to_vec()).await.unwrap(); writer.finish().await.unwrap();
let blocks = parse_car_blocks(buf.into(), false).unwrap(); assert_eq!(blocks.get(&cid).unwrap().as_ref(), b"block"); }
#[tokio::test] async fn parses_rooted_car_with_zero_copy_payloads() { let root = cid(1); let block = cid(2); let mut buf = Vec::new(); let header = CarHeader::new_v1(vec![root]); let mut writer = CarWriter::new(header, &mut buf); writer.write(block, b"record".to_vec()).await.unwrap(); writer.finish().await.unwrap();
let data = Bytes::from(buf); let data_range = data.as_ptr() as usize..data.as_ptr() as usize + data.len(); let parsed = parse_car(data, false).unwrap(); let payload = parsed.blocks.get(&block).unwrap(); let payload_range = payload.as_ptr() as usize..payload.as_ptr() as usize + payload.len();
assert_eq!(parsed.root, root); assert_eq!(payload.as_ref(), b"record"); assert!(data_range.start <= payload_range.start); assert!(payload_range.end <= data_range.end); }
async fn car_with_blocks(blocks: &[(IpldCid, &[u8])]) -> Bytes { let mut buf = Vec::new(); let header = CarHeader::new_v1(vec![blocks[0].0]); let mut writer = CarWriter::new(header, &mut buf); for (cid, bytes) in blocks { writer.write(*cid, bytes.to_vec()).await.unwrap(); } writer.finish().await.unwrap(); buf.into() }
/// a rooted car whose blocks are big enough for some sections to need a two-byte length async fn stream_fixture() -> (IpldCid, Vec<(IpldCid, Vec<u8>)>, Bytes) { let blocks = [ b"a".to_vec(), vec![7u8; 300], b"record".to_vec(), vec![1u8; 129], ] .into_iter() .map(|bytes| (compute_cid(&bytes), bytes)) .collect::<Vec<_>>(); let root = blocks[0].0; let mut buf = Vec::new(); let mut writer = CarWriter::new(CarHeader::new_v1(vec![root]), &mut buf); for (cid, bytes) in &blocks { writer.write(*cid, bytes).await.unwrap(); } writer.finish().await.unwrap(); (root, blocks, buf.into()) }
fn stream_in_chunks( car: &[u8], chunk: usize, verify_cids: bool, ) -> Result<(CommitCar, Vec<IpldCid>)> { let mut stream = CarStream::new(verify_cids); let mut kept = CommitBlocks::new(); let mut order = Vec::new(); for piece in car.chunks(chunk) { stream.push(piece, |_, cid, bytes| { order.push(cid); kept.insert(cid, Bytes::copy_from_slice(bytes)); Ok(()) })?; } Ok((stream.finish_with(kept)?, order)) }
#[tokio::test] async fn stream_yields_what_a_whole_parse_does_however_the_car_is_chunked() { let (root, blocks, car) = stream_fixture().await; let whole = parse_car(car.clone(), true).unwrap(); for chunk in [1, 2, 3, 7, 64, 333, car.len()] { let (streamed, order) = stream_in_chunks(&car, chunk, true).unwrap();
assert_eq!(streamed.root, root, "chunk {chunk}"); assert_eq!( order, blocks.iter().map(|(cid, _)| *cid).collect::<Vec<_>>(), "chunk {chunk}" ); for (cid, bytes) in &blocks { assert_eq!( streamed.blocks.get(cid).unwrap().as_ref(), bytes.as_slice(), "chunk {chunk}" ); assert_eq!( whole.blocks.get(cid), streamed.blocks.get(cid), "chunk {chunk}" ); } assert!(streamed.blocks.block(&root).unwrap().verify().is_ok()); } }
#[tokio::test] async fn a_stream_turns_a_huge_block_away_before_it_arrives() { let (_, _, car) = stream_fixture().await; let header_len = { let mut offset = 0; let len = read_uvarint(&car, &mut offset).unwrap().unwrap(); offset + len }; // the header, then a section that only says how long it'll be let mut start = car[..header_len].to_vec(); let mut len = MAX_STREAMED_SECTION_BYTES + 1; while len >= 0x80 { start.push((len as u8 & 0x7f) | 0x80); len >>= 7; } start.push(len as u8);
let mut stream = CarStream::new(true); let err = stream.push(&start, |_, _, _| Ok(())).unwrap_err(); assert!(err.to_string().contains("CAR section of"), "{err}"); }
#[tokio::test] async fn whole_car_keeps_every_block_a_whole_parse_does() { let (root, _, car) = stream_fixture().await; let parsed = parse_car(car.clone(), true).unwrap(); for chunk in [1, 3, 64, car.len()] { let mut whole = WholeCar::new(true); whole.reserve(car.len()); for piece in car.chunks(chunk) { whole.push(piece).unwrap(); } let streamed = whole.finish().unwrap(); assert_eq!(streamed.root, root, "chunk {chunk}"); assert_eq!(streamed.blocks.map(), parsed.blocks.map(), "chunk {chunk}"); assert!(streamed.blocks.block(&root).unwrap().cid_checked); }
let real = compute_cid(b"trusted"); let forged = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await; let mut whole = WholeCar::new(true); let err = whole.push(&forged).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch"), "{err}"); }
#[tokio::test] async fn stream_cut_short_is_truncated() { let (_, blocks, car) = stream_fixture().await; let err = stream_in_chunks(&[], 1, true).unwrap_err(); assert!(err.to_string().contains("empty CAR file"), "{err}");
// cutting exactly between sections leaves a complete, shorter car let mut boundaries = Vec::new(); let mut offset = 0; while let Some(len) = read_uvarint(&car, &mut offset).unwrap() { offset += len; boundaries.push(offset); } assert_eq!(boundaries.len(), blocks.len() + 1); for cut in 1..car.len() { let result = stream_in_chunks(&car[..cut], 5, true); if let Some(sections) = boundaries.iter().position(|&end| end == cut) { let (streamed, order) = result.unwrap(); assert_eq!(order.len(), sections, "cut at {cut}"); assert_eq!(streamed.blocks.map().len(), sections, "cut at {cut}"); continue; } let err = result .err() .unwrap_or_else(|| panic!("a car cut at {cut} parsed")); let want = if cut < boundaries[0] { "truncated CAR header" } else { "truncated CAR block" }; assert!(err.to_string().contains(want), "cut at {cut}: {err}"); } }
#[tokio::test] async fn stream_rejects_a_forged_block_when_verifying() { let real = compute_cid(b"trusted"); let car = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await;
let err = stream_in_chunks(&car, 3, true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch"), "{err}"); let (unverified, _) = stream_in_chunks(&car, 3, false).unwrap(); assert_eq!(unverified.blocks.get(&cid(1)).unwrap().as_ref(), b"forged"); assert!(unverified.blocks.block(&cid(1)).unwrap().verify().is_err()); }
#[tokio::test] async fn stream_rejects_a_rootless_car() { let mut buf = Vec::new(); let mut writer = CarWriter::new(CarHeader::new_v1(Vec::new()), &mut buf); writer.write(cid(1), b"block".to_vec()).await.unwrap(); writer.finish().await.unwrap();
let streamed = stream_in_chunks(&buf, 4, false).unwrap_err(); let whole = parse_car(buf.into(), false).unwrap_err(); assert_eq!(streamed.to_string(), whole.to_string()); }
#[test] fn partial_uvarint_waits_for_more_bytes() { assert_eq!(try_read_uvarint(&[]).unwrap(), None); assert_eq!(try_read_uvarint(&[0x80]).unwrap(), None); assert_eq!(try_read_uvarint(&[0xac, 0x02]).unwrap(), Some((300, 2))); assert_eq!(try_read_uvarint(&[0x05, 0xff]).unwrap(), Some((5, 1))); assert!(try_read_uvarint(&[0x80; 10]).is_err()); }
#[tokio::test] async fn verifying_parse_rejects_forged_block() { let real = compute_cid(b"trusted"); let car = car_with_blocks(&[(real, b"trusted"), (cid(1), b"forged")]).await;
assert!(parse_car(car.clone(), false).is_ok()); let err = parse_car(car.clone(), true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); let err = parse_commit_car(car.clone(), true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); let err = parse_car_blocks(car, true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); }
#[tokio::test] async fn verifying_parse_accepts_matching_blocks() { let root = compute_cid(b"root"); let leaf = compute_cid(b"leaf"); let car = car_with_blocks(&[(root, b"root"), (leaf, b"leaf")]).await;
let parsed = parse_car(car.clone(), true).unwrap(); assert_eq!(parsed.root, root); assert_eq!(parsed.blocks.get(&leaf).unwrap().as_ref(), b"leaf"); let commit = parse_commit_car(car, true).unwrap(); assert_eq!(commit.root, root); assert_eq!(commit.blocks.get(&leaf).unwrap().as_ref(), b"leaf"); }
#[tokio::test] async fn verifying_parse_rejects_forged_copy_shadowed_by_valid_duplicate() { // the map keeps only the last copy, so the forged one is caught only while parsing; // no honest pds sends it, so the whole car is rejected let real = compute_cid(b"trusted"); let car = car_with_blocks(&[(real, b"forged"), (real, b"trusted")]).await;
let err = parse_car(car.clone(), true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); let err = parse_commit_car(car, true).unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch")); }
#[tokio::test] async fn every_parse_keeps_the_last_duplicate_copy() { let car = car_with_blocks(&[(cid(1), b"first"), (cid(1), b"second")]).await;
let repo = parse_car(car.clone(), false).unwrap(); assert_eq!(repo.blocks.get(&cid(1)).unwrap().as_ref(), b"second"); let commit = parse_commit_car(car.clone(), false).unwrap(); assert_eq!(commit.blocks.get(&cid(1)).unwrap().as_ref(), b"second"); let blocks = parse_car_blocks(car, false).unwrap(); assert_eq!(blocks.get(&cid(1)).unwrap().as_ref(), b"second"); }
// iroh-car's reader, behind jacquard's `parse_car_bytes`, accepts each of the next three
#[tokio::test] async fn rejects_a_trailing_partial_varint() { // iroh-car takes a varint cut off by the end of the car for a clean end let mut car = car_with_blocks(&[(cid(1), b"block")]).await.to_vec(); car.push(0x80);
let err = parse_commit_car(car.into(), false).unwrap_err(); assert!(err.to_string().contains("truncated uvarint")); }
#[tokio::test] async fn rejects_a_section_longer_than_the_car() { // iroh-car zero-fills a buffer of the claimed length before reading into it let mut car = car_with_blocks(&[(cid(1), b"block")]).await.to_vec(); car.extend_from_slice(&[0xff, 0xff, 0xff, 0x01, 0x00]);
let err = parse_commit_car(car.into(), false).unwrap_err(); assert!(err.to_string().contains("truncated CAR block")); }
#[tokio::test] async fn rejects_a_block_without_payload() { // iroh-car accepts it, and the cid of empty bytes even passes verification let empty = compute_cid(b""); let car = car_with_blocks(&[(empty, b"")]).await;
let err = parse_commit_car(car, true).unwrap_err(); assert!(err.to_string().contains("has no payload")); }
fn forged(claimed: IpldCid) -> Blocks<CommitBlocks> { Blocks::unverified(BTreeMap::from([(claimed, Bytes::from_static(b"forged"))])) }
#[test] fn unverified_blocks_are_hashed_when_verified() { let block = forged(cid(1)).block(&cid(1)).unwrap(); let err = block.verify().unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch"));
let real = compute_cid(b"forged"); let block = forged(real).block(&real).unwrap(); let verified = block.verify().unwrap(); assert_eq!(verified.cid(), &real); assert_eq!(verified.bytes().as_ref(), b"forged"); }
#[test] fn checking_a_block_hashes_it_once() { let block = forged(cid(1)).block(&cid(1)).unwrap(); let err = block.checked().unwrap_err(); assert!(err.to_string().contains("CAR block CID mismatch"));
let real = compute_cid(b"forged"); let checked = forged(real).block(&real).unwrap().checked().unwrap(); assert!(checked.cid_checked); assert_eq!(checked.verify().unwrap().bytes().as_ref(), b"forged"); }
#[tokio::test] async fn blocks_from_a_non_verifying_parse_are_hashed_when_verified() { let car = car_with_blocks(&[(cid(1), b"forged")]).await;
let repo = parse_car(car.clone(), false).unwrap(); assert!(repo.blocks.block(&cid(1)).unwrap().verify().is_err()); let commit = parse_commit_car(car.clone(), false).unwrap(); assert!(commit.blocks.block(&cid(1)).unwrap().verify().is_err()); let blocks = parse_car_blocks(car, false).unwrap(); assert!(blocks.block(&cid(1)).unwrap().verify().is_err()); }
#[tokio::test] async fn blocks_from_a_verifying_parse_verify_as_parsed() { let leaf = compute_cid(b"leaf"); let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let parsed = parse_commit_car(car, true).unwrap(); let block = parsed.blocks.block(&leaf).unwrap(); let verified = block.verify().unwrap(); assert_eq!(verified.cid(), &leaf); assert_eq!(verified.bytes().as_ref(), b"leaf"); }
#[tokio::test] async fn merged_blocks_stay_checked_only_if_both_sides_were() { let leaf = compute_cid(b"leaf"); let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let mut checked = parse_car_blocks(car.clone(), true).unwrap(); checked.extend(forged(cid(1))); assert!(checked.block(&cid(1)).unwrap().verify().is_err());
let mut unchecked = forged(cid(1)); unchecked.extend(parse_car_blocks(car, true).unwrap()); assert!(unchecked.block(&cid(1)).unwrap().verify().is_err());
let mut empty = Blocks::default(); empty.extend(forged(cid(1))); assert!(empty.block(&cid(1)).unwrap().verify().is_err()); }
#[tokio::test] async fn an_empty_default_keeps_merged_blocks_checked() { // sparse backfill folds verified fetches into a default, so a default // counted as unchecked would hash every one of them again let leaf = compute_cid(b"leaf"); let car = car_with_blocks(&[(leaf, b"leaf")]).await;
let mut merged = Blocks::default(); merged.extend(parse_car_blocks(car, true).unwrap()); let block = merged.block(&leaf).unwrap(); assert!(block.cid_checked); assert_eq!(block.cid(), &leaf); }}