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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365use std::collections::{HashMap, HashSet, VecDeque};
use bytes::Bytes;use cid::Cid as IpldCid;use miette::{Result, WrapErr};use smol_str::SmolStr;
use crate::car::{Blocks, CommitBlocks};use crate::mst::{FlatEntry, MstKey, decode_node, decode_node_into, layer_for_key};
#[derive(Debug, Clone, PartialEq, Eq)]pub(crate) struct KeyRange { start: SmolStr, end: Option<SmolStr>,}
impl KeyRange { #[inline] pub(crate) fn contains(&self, key: impl AsRef<[u8]>) -> bool { let key = key.as_ref(); key >= self.start.as_bytes() && self .end .as_ref() .map(|end| key < end.as_bytes()) .unwrap_or(true) }
#[inline] fn intersects_subtree( &self, lower_exclusive: Option<&[u8]>, upper_exclusive: Option<&[u8]>, ) -> bool { if let Some(upper) = upper_exclusive { if self.start.as_bytes() >= upper { return false; } }
if let (Some(end), Some(lower)) = (self.end.as_ref(), lower_exclusive) { if end.as_bytes() <= lower { return false; } }
true }}
pub(crate) fn sparse_ranges(patterns: &[SmolStr]) -> Vec<KeyRange> { patterns .iter() .filter_map(|pattern| { let prefix = pattern .strip_suffix(".*") .map(|prefix| SmolStr::new(format!("{prefix}."))) .unwrap_or_else(|| SmolStr::new(format!("{pattern}/"))); Some(KeyRange { end: prefix_upper_bound(prefix.as_str()), start: prefix, }) }) .collect()}
pub(crate) fn sparse_probe_collection(patterns: &[SmolStr]) -> Option<SmolStr> { patterns.iter().find_map(|pattern| { pattern .strip_suffix(".*") .map(|prefix| format!("{prefix}.probe")) .unwrap_or_else(|| pattern.to_string()) .parse::<jacquard_common::types::string::Nsid>() .ok() .map(|_| { pattern .strip_suffix(".*") .map(|prefix| SmolStr::new(format!("{prefix}.probe"))) .unwrap_or_else(|| pattern.clone()) }) })}
fn prefix_upper_bound(prefix: &str) -> Option<SmolStr> { let mut bytes = prefix.as_bytes().to_vec(); for i in (0..bytes.len()).rev() { if bytes[i] < u8::MAX { bytes[i] += 1; bytes.truncate(i + 1); return String::from_utf8(bytes).ok().map(SmolStr::new); } } None}
pub(crate) fn mst_node_layer(bytes: &[u8]) -> Result<Option<usize>> { let entries = decode_node(bytes).wrap_err("failed to decode mst node for layer estimate")?; Ok(entries.into_iter().find_map(|entry| match entry { FlatEntry::Leaf { key, .. } => Some(layer_for_key(key.as_str())), FlatEntry::Tree { .. } => None, }))}
/// leaves from a completed [`SparseScanner::scan`], guaranteed sorted and deduplicated by/// strictly increasing mst key#[derive(Debug, Clone, Default, PartialEq, Eq)]pub(crate) struct SortedLeaves(Vec<(MstKey, IpldCid)>);
impl SortedLeaves { fn from_raw(mut leaves: Vec<(MstKey, IpldCid)>) -> Self { leaves.sort_unstable_by(|a, b| a.0.cmp(&b.0)); leaves.dedup_by(|a, b| a.0 == b.0); Self(leaves) }
#[cfg(test)] pub(crate) fn from_vec(leaves: Vec<(MstKey, IpldCid)>) -> Self { Self::from_raw(leaves) }}
impl std::ops::Deref for SortedLeaves { type Target = [(MstKey, IpldCid)];
fn deref(&self) -> &Self::Target { &self.0 }}
impl<'a> IntoIterator for &'a SortedLeaves { type Item = &'a (MstKey, IpldCid); type IntoIter = std::slice::Iter<'a, (MstKey, IpldCid)>;
fn into_iter(self) -> Self::IntoIter { self.0.iter() }}
impl PartialEq<Vec<(MstKey, IpldCid)>> for SortedLeaves { fn eq(&self, other: &Vec<(MstKey, IpldCid)>) -> bool { &self.0 == other }}
impl PartialEq<[(MstKey, IpldCid)]> for SortedLeaves { fn eq(&self, other: &[(MstKey, IpldCid)]) -> bool { self.0.as_slice() == other }}
impl PartialEq<Vec<(SmolStr, IpldCid)>> for SortedLeaves { fn eq(&self, other: &Vec<(SmolStr, IpldCid)>) -> bool { self.0.len() == other.len() && self .0 .iter() .zip(other.iter()) .all(|((k1, c1), (k2, c2))| k1 == k2 && c1 == c2) }}
#[derive(Debug, Clone)]pub(crate) struct SparseScanOutput { pub(crate) leaves: SortedLeaves, pub(crate) node_blocks_seen: usize, pub(crate) node_bytes_seen: usize,}
#[derive(Debug, Clone, PartialEq, Eq)]struct NodeBounds { lower: Option<MstKey>, upper: Option<MstKey>,}
/// almost every mst node is reached through a single parent with a single bound; keeping the/// first bound inline avoids a heap allocation per node in `pending_missing` and `visited`#[derive(Debug, Clone)]struct BoundsSet { first: NodeBounds, rest: Vec<NodeBounds>,}
impl BoundsSet { fn new(first: NodeBounds) -> Self { Self { first, rest: Vec::new(), } }
fn covers(&self, bounds: &NodeBounds) -> bool { self.first.covers(bounds) || self.rest.iter().any(|prior| prior.covers(bounds)) }
fn insert(&mut self, bounds: NodeBounds) { if !self.covers(&bounds) { self.rest.push(bounds); } }
fn iter(&self) -> std::iter::Chain<std::iter::Once<&NodeBounds>, std::slice::Iter<'_, NodeBounds>> { std::iter::once(&self.first).chain(self.rest.iter()) }
fn into_iter(self) -> std::iter::Chain<std::iter::Once<NodeBounds>, std::vec::IntoIter<NodeBounds>> { std::iter::once(self.first).chain(self.rest) }}
type CidMap<V> = HashMap<IpldCid, V, ahash::RandomState>;type CidSet = HashSet<IpldCid, ahash::RandomState>;
impl NodeBounds { fn unbounded() -> Self { Self { lower: None, upper: None, } }
fn covers(&self, other: &Self) -> bool { let lower_ok = match (&self.lower, &other.lower) { (None, _) => true, (Some(_), None) => false, (Some(a), Some(b)) => a <= b, }; let upper_ok = match (&self.upper, &other.upper) { (None, _) => true, (Some(_), None) => false, (Some(a), Some(b)) => a >= b, }; lower_ok && upper_ok }
#[inline] fn contains_leaf(&self, key: impl AsRef<[u8]>) -> bool { let key = key.as_ref(); self.lower .as_ref() .is_none_or(|lower| key > lower.as_bytes()) && self .upper .as_ref() .is_none_or(|upper| key < upper.as_bytes()) }
#[inline] fn subtree_bounds<'a>( &'a self, lower_leaf: Option<&'a MstKey>, upper_leaf: Option<&'a MstKey>, ranges: &[KeyRange], ) -> Option<(Option<&'a MstKey>, Option<&'a MstKey>)> { let lower = match (self.lower.as_ref(), lower_leaf) { (Some(a), Some(b)) => Some(a.max(b)), (a, b) => a.or(b), }; let upper = match (self.upper.as_ref(), upper_leaf) { (Some(a), Some(b)) => Some(a.min(b)), (a, b) => a.or(b), }; if lower.zip(upper).is_some_and(|(l, u)| l >= u) { return None; } if !ranges.iter().any(|range| { range.intersects_subtree(lower.map(MstKey::as_bytes), upper.map(MstKey::as_bytes)) }) { return None; } Some((lower, upper)) }
/// narrows `self` to a child subtree's `(lower_leaf, upper_leaf)` interval, returning /// `None` if the resulting interval is inverted/empty or cannot intersect `ranges` fn narrow( &self, lower_leaf: Option<&MstKey>, upper_leaf: Option<&MstKey>, ranges: &[KeyRange], ) -> Option<Self> { let (lower, upper) = self.subtree_bounds(lower_leaf, upper_leaf, ranges)?; Some(Self { lower: lower.cloned(), upper: upper.cloned(), }) }}
/// whether a node seen on its own, its place in the tree unknown, could hold a key in/// `ranges`: a leaf in one, or a subtree that could reach one whatever the node's outer bounds/// turn out to be. a scan of `ranges` takes nothing from a node this rules out, from any/// ancestor, because real bounds only narrow the intervals checked herepub(crate) fn node_may_reach(ranges: &[KeyRange], entries: &[FlatEntry]) -> bool { node_may_reach_within(&NodeBounds::unbounded(), ranges, entries)}
fn entries_within_bounds(bounds: &NodeBounds, entries: &[FlatEntry]) -> bool { let first = entries.iter().find_map(|entry| match entry { FlatEntry::Leaf { key, .. } => Some(key), FlatEntry::Tree { .. } => None, }); let last = entries.iter().rev().find_map(|entry| match entry { FlatEntry::Leaf { key, .. } => Some(key), FlatEntry::Tree { .. } => None, }); first.is_none_or(|key| bounds.contains_leaf(key)) && last.is_none_or(|key| bounds.contains_leaf(key))}
fn node_may_reach_within( bounds: &NodeBounds, ranges: &[KeyRange], entries: &[FlatEntry],) -> bool { entries.iter().enumerate().any(|(idx, entry)| match entry { FlatEntry::Leaf { key, .. } => { bounds.contains_leaf(key) && ranges.iter().any(|range| range.contains(key)) } FlatEntry::Tree { .. } => bounds .subtree_bounds(previous_leaf(entries, idx), next_leaf(entries, idx), ranges) .is_some(), })}
/// mst nodes [`node_may_reach`] ruled out, kept as bare sha-256 digests since a car can rule/// out most of its nodes and a whole cid is several times bigger. a digest names the same/// bytes whatever codec a cid puts in front of it/// how many bytes of out-of-order mst nodes a streamed car parks while waiting for their/// parents. a node evicted past this is fetched again if an ancestor turns out to want itpub(crate) const SPARSE_PARKED_BUDGET: usize = 64 * 1024 * 1024;
#[derive(Debug, Default)]pub(crate) struct RuledOut(HashSet<[u8; 32], ahash::RandomState>);
impl RuledOut { fn with_hasher(hasher: ahash::RandomState) -> Self { Self(HashSet::with_hasher(hasher)) }
/// remembers `cid` as ruled out. false for a cid that isn't sha-256, which can't be ruled out pub(crate) fn insert(&mut self, cid: &IpldCid) -> bool { sha256_digest(cid) .map(|digest| self.0.insert(digest)) .is_some() }
#[inline] pub(crate) fn contains(&self, cid: &IpldCid) -> bool { !self.0.is_empty() && sha256_digest(cid).is_some_and(|digest| self.0.contains(&digest)) }
pub(crate) fn len(&self) -> usize { self.0.len() }}
fn sha256_digest(cid: &IpldCid) -> Option<[u8; 32]> { let hash = cid.hash(); if hash.code() != jacquard_common::types::crypto::SHA2_256 { return None; } hash.digest().try_into().ok()}
#[derive(Debug)]pub(crate) struct SparseScanner { ranges: Vec<KeyRange>, blocks: Blocks<CommitBlocks>, hasher: ahash::RandomState, parked: CidMap<Bytes>, unparked: CidMap<Bytes>, parked_order: VecDeque<IpldCid>, parked_bytes: usize, parked_budget: usize, evicted: CidSet, corrupt: CidMap<String>, ruled_out: RuledOut, root: Option<IpldCid>, root_bounds: Vec<NodeBounds>, pending_missing: CidMap<BoundsSet>, visited: CidMap<BoundsSet>, leaves: Vec<(MstKey, IpldCid)>, scratch: Vec<FlatEntry>, stack: Vec<(IpldCid, NodeBounds)>, stats: ScanStats,}
impl SparseScanner { pub(crate) fn new(ranges: Vec<KeyRange>, blocks: Blocks<CommitBlocks>) -> Self { let hasher = ahash::RandomState::new(); Self { ranges, blocks, parked: HashMap::with_hasher(hasher.clone()), unparked: HashMap::with_hasher(hasher.clone()), parked_order: VecDeque::new(), parked_bytes: 0, parked_budget: usize::MAX, evicted: HashSet::with_hasher(hasher.clone()), corrupt: HashMap::with_hasher(hasher.clone()), ruled_out: RuledOut::with_hasher(hasher.clone()), root: None, root_bounds: Vec::new(), pending_missing: HashMap::with_hasher(hasher.clone()), visited: HashMap::with_hasher(hasher.clone()), hasher, leaves: Vec::new(), scratch: Vec::new(), stack: Vec::new(), stats: ScanStats::default(), } }
pub(crate) fn with_parked_budget(ranges: Vec<KeyRange>, parked_budget: usize) -> Self { Self { parked_budget, ..Self::new(ranges, Blocks::default()) } }
pub(crate) fn insert_blocks(&mut self, blocks: Blocks<CommitBlocks>) { self.blocks.extend(blocks); }
/// nodes a streamed car already ruled out, which the scan then reads as empty instead of /// missing #[cfg(test)] pub(crate) fn rule_out(&mut self, ruled_out: RuledOut) { self.ruled_out = ruled_out; }
pub(crate) fn is_complete(&self) -> bool { self.root.is_some() && self.pending_missing.is_empty() }
pub(crate) fn was_evicted(&self, cid: &IpldCid) -> bool { !self.evicted.is_empty() && self.evicted.contains(cid) }
#[cfg(test)] pub(crate) fn evicted_any(&self) -> bool { !self.evicted.is_empty() }
pub(crate) fn is_pending(&self, cid: &IpldCid) -> bool { self.pending_missing.contains_key(cid) }
pub(crate) fn ruled_out_len(&self) -> usize { self.ruled_out.len() }
#[cfg(test)] pub(crate) fn parked_bytes(&self) -> usize { self.parked_bytes }
/// resets parked state before re-streaming a whole car for nodes evicted on an earlier pass pub(crate) fn prepare_pass(&mut self) { if !self.parked.is_empty() { self.parked = HashMap::with_hasher(self.hasher.clone()); self.parked_order = VecDeque::new(); self.parked_bytes = 0; } if !self.unparked.is_empty() { self.unparked = HashMap::with_hasher(self.hasher.clone()); } if !self.evicted.is_empty() { self.evicted = HashSet::with_hasher(self.hasher.clone()); } if !self.corrupt.is_empty() { self.corrupt = HashMap::with_hasher(self.hasher.clone()); } }
/// drops parked nodes and ruled-out digests once the stream ends and no evicted node is pending pub(crate) fn finish_stream(&mut self) { if self.root.is_some() && (self.evicted.is_empty() || self.is_complete()) { self.clear_parked(); } }
/// sets the mst root once the car's commit block arrives and resolves any parked nodes /// reachable from it pub(crate) fn start(&mut self, root: IpldCid) -> Result<()> { if let Some(existing_root) = self.root { if existing_root != root { return Err(miette::miette!( "sparse scanner root changed from {existing_root} to {root}" )); } return Ok(()); } self.root = Some(root); self.scan_subtree(root, NodeBounds::unbounded())?; if self.is_complete() { self.clear_parked(); } Ok(()) }
/// offers a non-record block from a streaming car. a node the walk already wants is /// resolved with its real bounds; one that arrives before its parent is parked if it /// could reach `ranges`, or remembered in `ruled_out` as a digest; once the walk is /// complete, everything else is dropped unparked. pub(crate) fn offer_block(&mut self, cid: IpldCid, bytes: &[u8]) -> Result<()> { if self.is_complete() { return Ok(()); } if let Some(bounds_set) = self.pending_missing.remove(&cid) { self.evicted.remove(&cid); self.scratch.clear(); decode_node_into(bytes, &mut self.scratch) .wrap_err_with(|| format!("failed to decode mst node {cid}"))?; let scratch = std::mem::take(&mut self.scratch); let mut stack = std::mem::take(&mut self.stack); stack.clear(); for bounds in bounds_set.into_iter() { if !self.is_visited(&cid, &bounds) { self.visit_entries(cid, bytes.len(), &scratch, bounds, &mut stack); } } self.scratch = scratch; let res = self.drain_stack(&mut stack); self.stack = stack; res?; if self.is_complete() { self.clear_parked(); } return Ok(()); } if self.ruled_out.contains(&cid) || self.visited.contains_key(&cid) || (!self.parked.is_empty() && self.parked.contains_key(&cid)) { return Ok(()); } self.scratch.clear(); if let Err(err) = decode_node_into(bytes, &mut self.scratch) { self.corrupt.insert(cid, format!("{err:#}")); return Ok(()); } // a node whose cid can't be remembered as ruled out has to be parked instead if self.scratch_may_reach() || !self.ruled_out.insert(&cid) { self.park(cid, Bytes::copy_from_slice(bytes)); } Ok(()) }
pub(crate) fn scan(&mut self, root: IpldCid) -> Result<Result<SparseScanOutput, Vec<IpldCid>>> { if let Some(existing_root) = self.root { if existing_root != root { return Err(miette::miette!( "sparse scanner root changed from {existing_root} to {root}" )); } } else { self.root = Some(root); self.scan_subtree(root, NodeBounds::unbounded())?; }
if !self.pending_missing.is_empty() { let mut ready = self .pending_missing .iter() .filter(|(cid, _)| { self.blocks.get(cid).is_some() || (!self.parked.is_empty() && self.parked.contains_key(cid)) || (!self.unparked.is_empty() && self.unparked.contains_key(cid)) }) .map(|(cid, bounds)| (*cid, bounds.clone())) .collect::<Vec<_>>(); ready.sort_unstable_by_key(|(cid, _)| *cid); for (cid, bounds_set) in ready { self.pending_missing.remove(&cid); for bounds in bounds_set.into_iter() { self.scan_subtree(cid, bounds)?; } } }
if self.pending_missing.is_empty() { self.clear_parked(); Ok(Ok(SparseScanOutput { leaves: SortedLeaves::from_raw(std::mem::take(&mut self.leaves)), node_blocks_seen: self.stats.node_blocks_seen, node_bytes_seen: self.stats.node_bytes_seen, })) } else { let mut missing = self.pending_missing.keys().copied().collect::<Vec<_>>(); missing.sort_unstable(); Ok(Err(missing)) } }
pub(crate) fn take_blocks(self) -> Blocks<CommitBlocks> { self.blocks }
fn root_visited(&self) -> bool { self.root.is_some_and(|root| self.visited.contains_key(&root)) }
fn scratch_may_reach(&self) -> bool { if !node_may_reach(&self.ranges, &self.scratch) { return false; } if !self.root_visited() { return true; } if self.pending_missing.len() <= 8 { return self.pending_missing.values().any(|bounds_set| { bounds_set.iter().any(|bounds| { entries_within_bounds(bounds, &self.scratch) && node_may_reach_within(bounds, &self.ranges, &self.scratch) }) }); } self.root_bounds.iter().any(|bounds| { entries_within_bounds(bounds, &self.scratch) && node_may_reach_within(bounds, &self.ranges, &self.scratch) }) }
fn is_visited(&self, cid: &IpldCid, bounds: &NodeBounds) -> bool { self.visited .get(cid) .is_some_and(|seen| seen.covers(bounds)) }
fn record_visited(&mut self, cid: IpldCid, byte_len: usize, bounds: NodeBounds) { if !self.evicted.is_empty() { self.evicted.remove(&cid); } match self.visited.entry(cid) { std::collections::hash_map::Entry::Occupied(mut occupied) => { occupied.get_mut().insert(bounds); } std::collections::hash_map::Entry::Vacant(vacant) => { self.stats.node_blocks_seen += 1; self.stats.node_bytes_seen += byte_len; vacant.insert(BoundsSet::new(bounds)); } } }
fn record_pending(&mut self, cid: IpldCid, bounds: NodeBounds) { match self.pending_missing.entry(cid) { std::collections::hash_map::Entry::Occupied(mut occupied) => { occupied.get_mut().insert(bounds); } std::collections::hash_map::Entry::Vacant(vacant) => { vacant.insert(BoundsSet::new(bounds)); } } }
fn park(&mut self, cid: IpldCid, bytes: Bytes) { if !self.parked.is_empty() && self.parked.contains_key(&cid) { return; } if !self.evicted.is_empty() { self.evicted.remove(&cid); } self.parked_bytes += bytes.len(); self.parked.insert(cid, bytes); self.parked_order.push_back(cid); while self.parked_bytes > self.parked_budget { let Some(oldest) = self.parked_order.pop_front() else { break; }; if let Some(evicted) = self.parked.remove(&oldest) { self.parked_bytes -= evicted.len(); self.evicted.insert(oldest); } } }
fn unpark(&mut self, cid: &IpldCid) -> Option<(Bytes, usize)> { if !self.parked.is_empty() && let Some(bytes) = self.parked.remove(cid) { let byte_len = bytes.len(); self.parked_bytes -= byte_len; if self.parked.is_empty() { self.parked_order.clear(); } else { while self .parked_order .front() .is_some_and(|front| !self.parked.contains_key(front)) { self.parked_order.pop_front(); } } self.unparked.insert(*cid, bytes.clone()); return Some((bytes, byte_len)); } if !self.unparked.is_empty() && let Some(bytes) = self.unparked.get(cid).cloned() { return Some((bytes, 0)); } None }
fn clear_parked(&mut self) { if !self.parked.is_empty() { self.parked = HashMap::with_hasher(self.hasher.clone()); self.parked_order = VecDeque::new(); self.parked_bytes = 0; } if !self.unparked.is_empty() { self.unparked = HashMap::with_hasher(self.hasher.clone()); } if !self.corrupt.is_empty() { self.corrupt = HashMap::with_hasher(self.hasher.clone()); } if self.ruled_out.len() > 0 { self.ruled_out = RuledOut::with_hasher(self.hasher.clone()); } self.scratch = Vec::new(); self.stack = Vec::new(); }
/// scans the subtree under `root`. proof cars come from the pds, so the scan keeps its own /// stack rather than recursing once per tree level. fn scan_subtree(&mut self, root: IpldCid, bounds: NodeBounds) -> Result<()> { let mut stack = std::mem::take(&mut self.stack); stack.clear(); stack.push((root, bounds)); let res = self.drain_stack(&mut stack); self.stack = stack; res }
fn drain_stack(&mut self, stack: &mut Vec<(IpldCid, NodeBounds)>) -> Result<()> { while let Some((cid, bounds)) = stack.pop() { let mut bounds_set = BoundsSet::new(bounds); while let Some(pos) = stack.iter().rposition(|(other, _)| *other == cid) { let (_, other_bounds) = stack.remove(pos); bounds_set.insert(other_bounds); } if bounds_set.iter().all(|b| self.is_visited(&cid, b)) || self.ruled_out.contains(&cid) { continue; } let (bytes, byte_len) = if let Some((bytes, byte_len)) = self.unpark(&cid) { (bytes, byte_len) } else if let Some(bytes) = self.blocks.get(&cid).cloned() { let len = bytes.len(); (bytes, len) } else if !self.corrupt.is_empty() && let Some(err) = self.corrupt.get(&cid) { return Err( miette::miette!("{err}").wrap_err(format!("failed to decode mst node {cid}")) ); } else { for b in bounds_set.into_iter() { if !self.is_visited(&cid, &b) { self.record_pending(cid, b); } } continue; };
self.scratch.clear(); decode_node_into(&bytes, &mut self.scratch) .wrap_err_with(|| format!("failed to decode mst node {cid}"))?; let scratch = std::mem::take(&mut self.scratch); for b in bounds_set.into_iter() { if !self.is_visited(&cid, &b) { self.visit_entries(cid, byte_len, &scratch, b, stack); } } self.scratch = scratch; }
Ok(()) }
fn visit_entries( &mut self, cid: IpldCid, byte_len: usize, entries: &[FlatEntry], bounds: NodeBounds, stack: &mut Vec<(IpldCid, NodeBounds)>, ) { let is_root = self.root == Some(cid) && self.root_bounds.is_empty(); self.record_visited(cid, byte_len, bounds.clone());
let stack_start = stack.len(); for idx in 0..entries.len() { match &entries[idx] { FlatEntry::Leaf { key, cid } => { if bounds.contains_leaf(key) && self.ranges.iter().any(|range| range.contains(key)) { self.leaves.push((key.clone(), *cid)); } } FlatEntry::Tree { cid } => { if let Some(child) = bounds.narrow( previous_leaf(entries, idx), next_leaf(entries, idx), &self.ranges, ) { if is_root { self.root_bounds.push(child.clone()); } stack.push((*cid, child)); } } } } // reversed, so subtrees are scanned left to right, depth first stack[stack_start..].reverse(); }}
#[derive(Debug, Default)]struct ScanStats { node_blocks_seen: usize, node_bytes_seen: usize,}
fn previous_leaf(entries: &[FlatEntry], idx: usize) -> Option<&MstKey> { match idx.checked_sub(1).and_then(|prev| entries.get(prev)) { Some(FlatEntry::Leaf { key, .. }) => Some(key), _ => None, }}
fn next_leaf(entries: &[FlatEntry], idx: usize) -> Option<&MstKey> { match entries.get(idx + 1) { Some(FlatEntry::Leaf { key, .. }) => Some(key), _ => None, }}
#[cfg(test)]mod tests { use std::collections::BTreeMap;
use super::*; use crate::mst::RawNodeData; use crate::mst::tests::{cid, left_chain, node}; use bytes::Bytes;
#[test] fn builds_exact_and_wildcard_ranges() { let ranges = sparse_ranges(&[SmolStr::new("sh.tangled.repo"), SmolStr::new("app.bsky.*")]);
assert!(ranges[0].contains("sh.tangled.repo/abc")); assert!(!ranges[0].contains("sh.tangled.repo.comment/abc")); assert!(ranges[1].contains("app.bsky.feed.post/abc")); assert!(!ranges[1].contains("app.bskyfoo.feed.post/abc")); assert!(!ranges[1].contains("app.csky.feed.post/abc")); }
#[test] fn synthesizes_probe_collection_for_wildcard() { assert_eq!( sparse_probe_collection(&[SmolStr::new("sh.tangled.*")]).as_deref(), Some("sh.tangled.probe") ); assert_eq!( sparse_probe_collection(&[SmolStr::new("sh.tangled.repo")]).as_deref(), Some("sh.tangled.repo") ); }
#[test] fn reports_missing_intersecting_subtrees_only() { let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]); let left = cid(1); let middle = cid(2); let right = cid(3); let root = cid(4); let root_node = node( vec![ ("app.bsky.feed.post/1", cid(5), Some(middle)), ("zz.example.record/1", cid(6), Some(right)), ], Some(left), ); let root_bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap(); let mut scanner = SparseScanner::new( wanted, Blocks::unverified(BTreeMap::from([(root, root_bytes.into())])), );
let missing = scanner.scan(root).unwrap().unwrap_err();
assert_eq!(missing, vec![middle]); }
#[test] fn resumes_without_rescanning_known_nodes() { let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]); let middle = cid(1); let root = cid(2); let record = cid(3); let root_node = node( vec![ ("app.bsky.feed.post/1", cid(4), Some(middle)), ("zz.example.record/1", cid(5), None), ], None, ); let middle_node = node(vec![("sh.tangled.repo/1", record, None)], None); let mut scanner = SparseScanner::new( wanted, Blocks::unverified(BTreeMap::from([( root, serde_ipld_dagcbor::to_vec(&root_node).unwrap().into(), )])), );
assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![middle]);
scanner.insert_blocks(Blocks::unverified(BTreeMap::from([( middle, serde_ipld_dagcbor::to_vec(&middle_node).unwrap().into(), )]))); let scan = scanner.scan(root).unwrap().unwrap();
assert_eq!( scan.leaves, vec![(SmolStr::new("sh.tangled.repo/1"), record)] ); assert_eq!(scan.node_blocks_seen, 2); }
#[test] fn node_may_reach_checks_leaves_and_every_subtree_interval() { let wanted = sparse_ranges(&[SmolStr::new("mm.x")]); let leaf = |key: &str| FlatEntry::Leaf { key: MstKey::new(key), cid: cid(9), }; let tree = FlatEntry::Tree { cid: cid(8) };
assert!(node_may_reach(&wanted, &[leaf("mm.x/1")])); assert!(!node_may_reach(&wanted, &[leaf("aa.a/1"), leaf("zz.z/1")])); assert!(!node_may_reach(&wanted, &[])); // between two keys the interval is known exactly assert!(node_may_reach( &wanted, &[leaf("aa.a/1"), tree.clone(), leaf("zz.z/1")] )); assert!(!node_may_reach( &wanted, &[leaf("aa.a/1"), tree.clone(), leaf("bb.b/1")] )); // at either edge the node's real bound is unknown, so it could reach anything past it assert!(node_may_reach(&wanted, &[tree.clone(), leaf("zz.z/1")])); assert!(!node_may_reach(&wanted, &[tree.clone(), leaf("aa.a/1")])); assert!(node_may_reach(&wanted, &[leaf("aa.a/1"), tree.clone()])); assert!(!node_may_reach(&wanted, &[leaf("zz.z/1"), tree])); }
#[test] fn scan_over_kept_nodes_finds_every_wanted_leaf() { let leaves = (0..3000usize).map(|i| { // mostly likes and posts with a few records of everything else, the way a real // account's repo looks let collection = match i % 300 { 0 => "sh.tangled.repo", 150 => "sh.tangled.repo.collaborator", 7 | 107 | 207 => "io.atcr.hold.layer", n if n % 4 == 0 => "app.bsky.feed.post", n if n % 20 == 1 => "zz.example.record", _ => "app.bsky.feed.like", }; (format!("{collection}/{i:013}"), cid((i % 251) as u8)) }); let (root, nodes) = crate::mst::build_tree(leaves).unwrap(); let all: crate::car::CarBlocks = nodes.into_iter().collect(); let every_leaf = crate::mst::leaves(&all, root).unwrap();
// whether the patterns want few enough keys that the leaf-level nodes, most of a tree, // should all be ruled out for (patterns, selective) in [ ( vec!["sh.tangled.repo", "sh.tangled.repo.collaborator"], true, ), (vec!["sh.tangled.*"], true), (vec!["app.bsky.feed.post"], false), (vec!["io.atcr.*", "zz.example.record"], true), (vec!["nothing.here"], true), (vec!["app.bsky.*", "sh.tangled.repo"], false), ] { let ranges = sparse_ranges(&patterns.iter().map(SmolStr::new).collect::<Vec<_>>()); let mut kept = BTreeMap::new(); let mut ruled_out = RuledOut::default(); for (cid, bytes) in &all { if node_may_reach(&ranges, &decode_node(bytes).unwrap()) { kept.insert(*cid, bytes.clone()); } else { assert!(ruled_out.insert(cid)); } } let mut scanner = SparseScanner::new(ranges.clone(), Blocks::unverified(kept.clone())); scanner.rule_out(ruled_out);
let scan = scanner .scan(root) .unwrap() .unwrap_or_else(|missing| panic!("{patterns:?}: missing {missing:?}"));
let want = every_leaf .iter() .filter(|(key, _)| ranges.iter().any(|range| range.contains(key))) .cloned() .collect::<Vec<_>>(); assert_eq!(scan.leaves, want, "{patterns:?}"); assert!( !selective || kept.len() * 3 < all.len(), "{patterns:?}: kept {} of {} nodes", kept.len(), all.len() );
// streaming the same nodes in either order resolves the walk on the fly and drops // everything it parked once complete let ordered = all.iter().map(|(cid, bytes)| (*cid, bytes.clone())).collect::<Vec<_>>(); for (label, start_first, pass) in [ ("top-down", true, ordered.clone()), ("bottom-up", false, ordered.into_iter().rev().collect()), ] { let mut streamed = SparseScanner::with_parked_budget(ranges.clone(), usize::MAX); if start_first { streamed.start(root).unwrap(); } for (cid, bytes) in &pass { streamed.offer_block(*cid, bytes).unwrap(); } if !start_first { streamed.start(root).unwrap(); } assert!(streamed.is_complete(), "{patterns:?} ({label})"); assert_eq!(streamed.parked_bytes(), 0, "{patterns:?} ({label})"); assert_eq!(streamed.ruled_out_len(), 0, "{patterns:?} ({label})"); let scan = streamed.scan(root).unwrap().unwrap(); assert_eq!(scan.leaves, want, "{patterns:?} ({label})"); } } }
#[test] fn estimates_node_layer_from_first_leaf() { let root_node = node(vec![("sh.tangled.repo/1", cid(1), None)], None); let bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap();
assert_eq!( mst_node_layer(&bytes).unwrap(), Some(layer_for_key("sh.tangled.repo/1")) ); }
fn car_bytes(node: &RawNodeData<'_>) -> Bytes { serde_ipld_dagcbor::to_vec(node).unwrap().into() }
#[test] fn ancestor_bounds_prune_out_of_range_boundary_subtrees() { // two disjoint ranges; the fetched middle subtree's leftmost child sits // entirely between them once the ancestor lower bound is applied. let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]); let boundary_child = cid(1); let middle = cid(2); let root = cid(3); let root_node = node( vec![ ("aa.x/1", cid(4), None), ("bb.q/1", cid(5), Some(middle)), ("zz.y/5", cid(6), None), ], None, ); // boundary_child spans ("bb.q/1", "mm.a/1") given the ancestor bound, // which intersects neither range; without the bound its interval is // (-inf, "mm.a/1"), which falsely intersects the aa.x range. let middle_node = node( vec![("mm.a/1", cid(7), None), ("zz.y/1", cid(8), None)], Some(boundary_child), ); let mut scanner = SparseScanner::new( wanted, Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])), );
assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![middle]);
scanner.insert_blocks(Blocks::unverified(BTreeMap::from([( middle, car_bytes(&middle_node), )]))); let scan = scanner.scan(root).unwrap().unwrap();
assert_eq!( scan.leaves, vec![ (SmolStr::new("aa.x/1"), cid(4)), (SmolStr::new("zz.y/1"), cid(8)), (SmolStr::new("zz.y/5"), cid(6)), ] ); }
#[test] fn scans_a_tree_deeper_than_the_thread_stack() { // proof cars come from the pds, so their depth is not bounded by anything we trust let depth = 100_000; let (top, nodes) = left_chain(depth, "sh.tangled.repo/1"); let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]); let mut scanner = SparseScanner::new(wanted, Blocks::unverified(nodes.into_iter().collect()));
let scan = scanner.scan(top).unwrap().unwrap();
assert_eq!( scan.leaves, vec![(SmolStr::new("sh.tangled.repo/1"), cid(9))] ); assert_eq!(scan.node_blocks_seen, depth + 1); }
#[test] fn stops_on_a_node_that_is_its_own_left_subtree() { // an unverified car can claim any cid; the narrower child bounds are already covered let root = cid(1); let root_node = node(vec![("sh.tangled.repo/1", cid(9), None)], Some(root)); let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]); let mut scanner = SparseScanner::new( wanted, Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])), );
let scan = scanner.scan(root).unwrap().unwrap();
assert_eq!( scan.leaves, vec![(SmolStr::new("sh.tangled.repo/1"), cid(9))] ); assert_eq!(scan.node_blocks_seen, 1); }
#[test] fn duplicate_pending_cids_fetch_once_and_scan_deterministically() { // the same missing child CID is referenced from two positions with // different ancestor bounds; it must appear once in the missing list // and both contexts must be scanned after a single insert. let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]); let shared = cid(1); let root = cid(2); let root_node = node( vec![ ("aa.w/1", cid(3), Some(shared)), ("bb.q/1", cid(4), None), ("zz.x/1", cid(5), Some(shared)), ], None, ); let shared_node = node( vec![("aa.x/1", cid(6), None), ("zz.y/1", cid(7), None)], None, ); let mut scanner = SparseScanner::new( wanted, Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])), );
assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![shared]);
scanner.insert_blocks(Blocks::unverified(BTreeMap::from([( shared, car_bytes(&shared_node), )]))); let scan = scanner.scan(root).unwrap().unwrap();
assert_eq!( scan.leaves, vec![ (SmolStr::new("aa.x/1"), cid(6)), (SmolStr::new("zz.y/1"), cid(7)), ] ); assert_eq!(scan.node_blocks_seen, 2);
// the same shared node arriving before its parents in a streamed car (including across // distinct intermediate parent branches) must be visited under both ancestor bounds let left_branch = cid(8); let right_branch = cid(9); let left_branch_node = node(vec![("aa.w/1", cid(3), Some(shared))], None); let right_branch_node = node(vec![("zz.x/1", cid(5), Some(shared))], None); let diamond_root = node( vec![("bb.q/1", cid(4), Some(right_branch))], Some(left_branch), ); let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]); let mut streamed = SparseScanner::with_parked_budget(wanted, usize::MAX); streamed .offer_block(shared, &car_bytes(&shared_node)) .unwrap(); streamed .offer_block(left_branch, &car_bytes(&left_branch_node)) .unwrap(); streamed .offer_block(right_branch, &car_bytes(&right_branch_node)) .unwrap(); streamed .offer_block(root, &car_bytes(&diamond_root)) .unwrap(); streamed.start(root).unwrap(); let scan = streamed.scan(root).unwrap().unwrap(); assert_eq!( scan.leaves, vec![ (SmolStr::new("aa.x/1"), cid(6)), (SmolStr::new("zz.y/1"), cid(7)), ] ); assert_eq!(scan.node_blocks_seen, 4); assert_eq!(streamed.parked_bytes(), 0); }
#[test] fn enforces_intra_node_key_order_ancestor_bounds_and_pending_decode_errors() { let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]); let root = cid(1); let unsorted = node( vec![ ("sh.tangled.repo/b", cid(2), None), ("sh.tangled.repo/a", cid(3), None), ], None, ); let mut scanner = SparseScanner::new( wanted.clone(), Blocks::unverified(BTreeMap::from([(root, car_bytes(&unsorted))])), ); let err = scanner.scan(root).unwrap_err(); assert!( format!("{err:?}").contains("does not sort after"), "{err:?}" );
// a child cannot overwrite its parent's boundary key or smuggle out-of-bounds leaves let right = cid(4); let left = cid(5); let bounded_root = node( vec![ ("sh.tangled.repo/b", cid(2), Some(right)), ("sh.tangled.repo/d", cid(3), None), ], Some(left), ); let out_of_bounds_left = node(vec![("sh.tangled.repo/z", cid(8), None)], None); let dup_and_valid_right = node( vec![ ("sh.tangled.repo/b", cid(9), None), ("sh.tangled.repo/c", cid(6), None), ("sh.tangled.repo/e", cid(7), None), ], None, ); let mut scanner = SparseScanner::new( wanted.clone(), Blocks::unverified(BTreeMap::from([ (root, car_bytes(&bounded_root)), (left, car_bytes(&out_of_bounds_left)), (right, car_bytes(&dup_and_valid_right)), ])), ); let scan = scanner.scan(root).unwrap().unwrap(); assert_eq!( scan.leaves, vec![ (SmolStr::new("sh.tangled.repo/b"), cid(2)), (SmolStr::new("sh.tangled.repo/c"), cid(6)), (SmolStr::new("sh.tangled.repo/d"), cid(3)), ] );
// wanted pending node whose block fails to decode errors immediately in offer_block, // whether it arrives after its parent or out of order before its parent let mut streamed = SparseScanner::with_parked_budget(wanted.clone(), usize::MAX); streamed .offer_block(root, &car_bytes(&bounded_root)) .unwrap(); streamed.start(root).unwrap(); assert!(streamed.is_pending(&right)); let err = streamed.offer_block(right, b"not cbor").unwrap_err(); assert!( err.to_string().contains("failed to decode mst node"), "{err}" );
let mut out_of_order = SparseScanner::with_parked_budget(wanted, usize::MAX); out_of_order.offer_block(right, b"not cbor").unwrap(); out_of_order .offer_block(left, &car_bytes(&out_of_bounds_left)) .unwrap(); out_of_order .offer_block(root, &car_bytes(&bounded_root)) .unwrap(); let err = out_of_order.start(root).unwrap_err(); assert!( err.to_string().contains("failed to decode mst node"), "{err}" ); }
#[test] fn evicts_oldest_parked_nodes_fifo_and_resumes_on_second_pass() { let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.repo")]); let child = cid(1); let stale = cid(2); let root = cid(3); let record = cid(4); let child_bytes = car_bytes(&node(vec![("sh.tangled.repo/1", record, None)], None)); // an unreachable historical node with an open left subtree that `node_may_reach` // cannot rule out without ancestor bounds let stale_bytes = car_bytes(&node(vec![("zz.other/1", cid(5), None)], Some(cid(6)))); let root_bytes = car_bytes(&node( vec![("app.bsky.feed.post/1", cid(7), Some(child))], None, ));
// budget fits one parked node at a time, so parking `stale` evicts `child` let mut scanner = SparseScanner::with_parked_budget(wanted, stale_bytes.len()); scanner.start(root).unwrap(); scanner.offer_block(child, &child_bytes).unwrap(); assert_eq!(scanner.parked_bytes(), child_bytes.len()); assert!(!scanner.evicted_any());
scanner.offer_block(stale, &stale_bytes).unwrap(); assert!(scanner.evicted_any()); assert_eq!(scanner.parked_bytes(), stale_bytes.len());
scanner.offer_block(root, &root_bytes).unwrap(); assert!(!scanner.is_complete()); assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![child]);
// a second pass resolves `child` directly because `root` already queued it in // `pending_missing`, without re-visiting `root` scanner.prepare_pass(); assert!(!scanner.evicted_any()); scanner.offer_block(child, &child_bytes).unwrap(); assert!(scanner.is_complete()); assert_eq!(scanner.parked_bytes(), 0); let scan = scanner.scan(root).unwrap().unwrap(); assert_eq!( scan.leaves, vec![(SmolStr::new("sh.tangled.repo/1"), record)] ); assert_eq!(scan.node_blocks_seen, 2); }}