[backfill] enforce mst key order and ancestor bounds in sparse scans
41 unmodified lines424344454646474849501 unmodified line5253545556575859606162636465243 unmodified linesExpand all30931031131230731331431531631731831932077 unmodified lines41 unmodified lines }
// each key is stored as a prefix length into the previous key plus a suffix let mut key = String::new(); for entry in &node.entries { for (idx, entry) in node.entries.iter().enumerate() { let suffix = std::str::from_utf8(&entry.key_suffix) .into_diagnostic() .wrap_err("invalid utf8 in mst key suffix")?; let prefix_len = usize::from(entry.prefix_len);1 unmodified line return Err(miette::miette!( "invalid mst key prefix length {prefix_len}" )); } if idx > 0 && suffix <= &key[prefix_len..] { let previous = key.clone(); key.truncate(prefix_len); key.push_str(suffix); miette::bail!("mst key {key} does not sort after {previous}"); } key.truncate(prefix_len); key.push_str(suffix);
entries.push(FlatEntry::Leaf {243 unmodified linesExpand all node(vec![("b/1", cid(9), None), ("a/1", cid(9), None)], None), )]);
let err = leaves(&tree, root).unwrap_err(); assert!(err.to_string().contains("does not sort after"), "{err}"); assert!( format!("{err:?}").contains("does not sort after"), "{err:?}" ); }
#[test] fn rejects_a_duplicate_key_across_subtrees() {77 unmodified lines164 unmodified linesExpand all165166167168169170171172173174175176177178179180181182183184174 unmodified linesExpand all359360361362351352353354363364365366367368195 unmodified linesExpand all564565566567558568569570571572573574575576577565566567578579580581582583584585398 unmodified linesExpand all984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106341 unmodified lines164 unmodified linesExpand all }; lower_ok && upper_ok }
fn contains_leaf(&self, key: &str) -> bool { self.lower.as_deref().is_none_or(|lower| key > lower) && self.upper.as_deref().is_none_or(|upper| key < upper) }
fn is_non_empty(&self) -> bool { match (&self.lower, &self.upper) { (Some(lower), Some(upper)) => lower < upper, _ => true, } }
fn child(&self, lower_leaf: Option<&str>, upper_leaf: Option<&str>) -> Self { let lower = match (self.lower.as_deref(), lower_leaf) { (Some(a), Some(b)) => Some(SmolStr::new(a.max(b))), (a, b) => a.or(b).map(SmolStr::new),174 unmodified linesExpand all if self.is_complete() { return Ok(()); } if let Some(bounds_set) = self.pending_missing.remove(&cid) { let Ok(entries) = decode_node(bytes) else { self.pending_missing.insert(cid, bounds_set); return Ok(()); }; let entries = decode_node(bytes).wrap_err_with(|| format!("failed to decode mst node {cid}"))?; let mut stack = Vec::new(); for bounds in bounds_set.into_iter() { if !self.is_visited(&cid, &bounds) { self.visit_entries(cid, bytes.len(), &entries, bounds, &mut stack);195 unmodified linesExpand all let mut children = Vec::new(); for idx in 0..entries.len() { match &entries[idx] { FlatEntry::Leaf { key, cid } => { if self.ranges.iter().any(|range| range.contains(key)) { if bounds.contains_leaf(key) && self.ranges.iter().any(|range| range.contains(key)) { self.leaves.insert(key.clone(), *cid); } } FlatEntry::Tree { cid } => { let child = bounds.child(previous_leaf(entries, idx), next_leaf(entries, idx)); if child.is_non_empty() if self.ranges.iter().any(|range| { range.intersects_subtree(child.lower.as_deref(), child.upper.as_deref()) }) { && self.ranges.iter().any(|range| { range.intersects_subtree(child.lower.as_deref(), child.upper.as_deref()) }) { children.push((*cid, child)); } } }398 unmodified linesExpand all 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 let mut streamed = SparseScanner::with_parked_budget(wanted, 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}" ); }
#[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);41 unmodified lines521 unmodified linesExpand all522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648521 unmodified linesExpand all } }; assert_eq!(resumed_leaves, leaves); }
#[tokio::test] async fn whole_car_filtered_car_and_sparse_scanner_agree_across_permutations_and_evictions() { let collections = [ "app.bsky.feed.like", "app.bsky.feed.post", "io.atcr.hold.layer", "sh.tangled.repo", "sh.tangled.repo.collaborator", "zz.example.record", ]; let mut records = Vec::new(); for i in 0..360usize { let collection = collections[i % collections.len()]; records.push((format!("{collection}/{i:013}"), record(collection, i))); } let (car, mst_root) = repo_car(&records).await; let parsed = crate::car::parse_car(car, true).unwrap(); let all_leaves = crate::mst::leaves(parsed.blocks.map(), mst_root).unwrap();
let filter_sets: &[&[&str]] = &[ &["sh.tangled.repo"], &["sh.tangled.repo", "sh.tangled.repo.collaborator"], &["sh.tangled.*"], &["io.atcr.hold.layer", "zz.example.record"], &["nothing.here"], ];
for patterns in filter_sets { let cfg = filter(patterns); let want = all_leaves .iter() .filter(|(key, _)| { let (collection, _) = key.split_once('/').unwrap(); cfg.matches_collection(collection) }) .cloned() .collect::<Vec<_>>();
// 1. multi-round SparseScanner (simulating incremental getBlocks rounds) let ranges = sparse_ranges(&cfg.collections); let mut root_only = std::collections::BTreeMap::new(); root_only.insert( mst_root, parsed.blocks.get(&mst_root).unwrap().clone(), ); let mut sparse = SparseScanner::new(ranges, Blocks::unverified(root_only)); let sparse_leaves = loop { match sparse.scan(mst_root).unwrap() { Ok(out) => break out.leaves, Err(missing) => { let round_blocks = missing .into_iter() .take(2) .map(|cid| (cid, parsed.blocks.get(&cid).unwrap().clone())) .collect(); sparse.insert_blocks(Blocks::unverified(round_blocks)); } } }; assert_eq!(sparse_leaves, want, "sparse rounds for {patterns:?}");
// 2. FilteredCar under deterministic block shuffles and a tiny 64-byte parked budget let base_blocks = parsed .blocks .map() .iter() .map(|(cid, bytes)| (*cid, bytes.clone())) .collect::<Vec<_>>(); for seed in 1..=6u64 { let mut state = seed.wrapping_mul(0x9e37_79b9_7f4a_7c15); let mut shuffled = base_blocks.clone(); for i in (1..shuffled.len()).rev() { state ^= state << 13; state ^= state >> 7; state ^= state << 17; let j = (state as usize) % (i + 1); shuffled.swap(i, j); } let mut car_bytes = Vec::new(); let mut writer = iroh_car::CarWriter::new( iroh_car::CarHeader::new_v1(vec![parsed.root]), &mut car_bytes, ); for (cid, bytes) in &shuffled { writer.write(*cid, bytes).await.unwrap(); } writer.finish().await.unwrap();
let mut stream_car = FilteredCar::with_parked_budget(cfg.clone(), 64, true).unwrap(); stream_car.push(&car_bytes).unwrap(); let FilteredRepo { blocks: kept_records, mut scanner, .. } = stream_car.finish().unwrap(); let streamed_leaves = loop { match scanner.scan(mst_root).unwrap() { Ok(out) => break out.leaves, Err(_) => { assert!(scanner.evicted_any()); let mut pass = FilteredCar::resume(parsed.root, scanner, true); pass.push(&car_bytes).unwrap(); scanner = pass.finish().unwrap().scanner; } } }; assert_eq!( streamed_leaves, want, "shuffled filtered car (seed {seed}) for {patterns:?}" ); for (_, cid) in &streamed_leaves { assert_eq!( kept_records.get(cid), parsed.blocks.get(cid), "kept record {cid} (seed {seed})" ); } } } }}