use 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 here
pub(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 it
pub(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);
}
}