//! a full repo car read block by block, keeping only what a filtered backfill can use. a
//! repo's car can be gigabytes of records in collections the filter doesn't want, and parsing
//! it whole means holding every one of them at once.
use std::collections::HashSet;
use std::sync::Arc;
use bytes::Bytes;
use cid::Cid as IpldCid;
use miette::Result;
use serde::Deserialize;
use crate::car::{Blocks, CarStream, CommitBlocks};
use crate::filter::FilterConfig;
use crate::sparse_mst::{SPARSE_PARKED_BUDGET, SparseScanner, sparse_ranges};
pub(crate) struct FilteredCar {
stream: CarStream,
keep: Keep,
}
struct Keep {
wanted: Wanted,
blocks: CommitBlocks,
stats: FilteredStats,
}
/// which blocks of a car are kept, besides its commit
enum Wanted {
/// the records `filter` wants, walking the mst through `scanner` as the car streams
Filter {
expected_root: Option<IpldCid>,
filter: Option<Arc<FilterConfig>>,
scanner: SparseScanner,
},
/// these and nothing else from a car of commit `root`, for going back for what an earlier
/// car of it left behind
Only {
root: IpldCid,
cids: HashSet<IpldCid>,
},
}
#[derive(Debug, Default, Clone, Copy)]
pub(crate) struct FilteredStats {
pub(crate) blocks: usize,
pub(crate) bytes: usize,
pub(crate) kept_blocks: usize,
pub(crate) kept_bytes: usize,
}
/// what a filtered car left behind: its commit, the records the filter wants and the mst
/// scanner that walked its nodes on the way
#[derive(Debug)]
pub(crate) struct FilteredRepo {
pub(crate) root: IpldCid,
pub(crate) blocks: Blocks<CommitBlocks>,
pub(crate) scanner: SparseScanner,
pub(crate) stats: FilteredStats,
}
impl FilteredCar {
/// `None` when `filter` takes every collection, since then every block is wanted anyway
pub(crate) fn new(filter: Arc<FilterConfig>, verify_cids: bool) -> Option<Self> {
Self::with_parked_budget(filter, SPARSE_PARKED_BUDGET, verify_cids)
}
pub(crate) fn with_parked_budget(
filter: Arc<FilterConfig>,
parked_budget: usize,
verify_cids: bool,
) -> Option<Self> {
let ranges = sparse_ranges(&filter.collections);
if ranges.is_empty() {
return None;
}
Some(Self::keeping(
Wanted::Filter {
expected_root: None,
filter: Some(filter),
scanner: SparseScanner::with_parked_budget(ranges, parked_budget),
},
verify_cids,
))
}
/// continues `scanner` over a second car of commit `root`, for resolving mst nodes the
/// parked budget evicted when the pds has no `getBlocks`
pub(crate) fn resume(root: IpldCid, mut scanner: SparseScanner, verify_cids: bool) -> Self {
scanner.prepare_pass();
Self::keeping(
Wanted::Filter {
expected_root: Some(root),
filter: None,
scanner,
},
verify_cids,
)
}
/// keeps the commit and `cids`, so a second car costs what's missing rather than the
/// whole repo
pub(crate) fn only(root: IpldCid, cids: HashSet<IpldCid>, verify_cids: bool) -> Self {
Self::keeping(Wanted::Only { root, cids }, verify_cids)
}
fn keeping(wanted: Wanted, verify_cids: bool) -> Self {
Self {
stream: CarStream::new(verify_cids),
keep: Keep {
wanted,
blocks: CommitBlocks::new(),
stats: FilteredStats::default(),
},
}
}
pub(crate) fn push(&mut self, chunk: &[u8]) -> Result<()> {
let keep = &mut self.keep;
self.stream
.push(chunk, |root, cid, bytes| keep.offer(root, cid, bytes))
}
pub(crate) fn finish(self) -> Result<FilteredRepo> {
let Keep {
wanted,
blocks,
stats,
} = self.keep;
let car = self.stream.finish_with(blocks)?;
let mut scanner = match wanted {
Wanted::Filter {
expected_root,
scanner,
..
} => {
if let Some(root) = expected_root {
moved_on(&root, &car.root)?;
}
scanner
}
Wanted::Only { root, .. } => {
moved_on(&root, &car.root)?;
SparseScanner::with_parked_budget(Vec::new(), 0)
}
};
scanner.finish_stream();
Ok(FilteredRepo {
root: car.root,
blocks: car.blocks,
scanner,
stats,
})
}
}
impl Keep {
fn offer(&mut self, root: &IpldCid, cid: IpldCid, bytes: &[u8]) -> Result<()> {
// the header comes first, so a car of another commit stops at its first block
// rather than after downloading the whole repo
if let Some(expected) = self.wanted.expected_root() {
moved_on(expected, root)?;
}
self.stats.blocks += 1;
self.stats.bytes += bytes.len();
let wanted = cid == *root || self.wanted.keeps(&cid, bytes)?;
if wanted {
self.stats.kept_blocks += 1;
self.stats.kept_bytes += bytes.len();
// copied, so a kept block doesn't pin the chunk it arrived in
self.blocks.insert(cid, Bytes::copy_from_slice(bytes));
}
if cid == *root
&& let Wanted::Filter { scanner, .. } = &mut self.wanted
&& let Ok(commit) = crate::mst::AtpCommit::from_cbor(bytes)
{
scanner.start(commit.data)?;
}
Ok(())
}
}
fn moved_on(expected: &IpldCid, root: &IpldCid) -> Result<()> {
if root != expected {
miette::bail!("repo moved on: this car is of commit {root}, not {expected}");
}
Ok(())
}
impl Wanted {
fn expected_root(&self) -> Option<&IpldCid> {
match self {
Wanted::Filter { expected_root, .. } => expected_root.as_ref(),
Wanted::Only { root, .. } => Some(root),
}
}
fn keeps(&mut self, cid: &IpldCid, bytes: &[u8]) -> Result<bool> {
Ok(match self {
Wanted::Only { cids, .. } => cids.contains(cid),
Wanted::Filter {
filter, scanner, ..
} => {
if let Some(collection) = record_type(bytes) {
// a record filed under a collection other than its `$type` is dropped here,
// and the scan finds it missing later
filter
.as_ref()
.is_some_and(|filter| filter.matches_collection(collection))
} else {
if !scanner.is_complete() {
scanner.offer_block(*cid, bytes)?;
}
false
}
}
})
}
}
/// a record's `$type`. commits, mst nodes and anything that isn't a map with a string `$type`
/// have none
fn record_type(bytes: &[u8]) -> Option<&str> {
#[derive(Deserialize)]
struct Typed<'a> {
#[serde(rename = "$type", borrow)]
ty: Option<&'a str>,
}
// in dag-cbor the 5-byte text key `"$type"` is always encoded as `0x65 "$type"`;
// skipping blocks without it avoids a full cbor map walk on every mst node
if !bytes.windows(6).any(|w| w == b"\x65$type") {
return None;
}
serde_ipld_dagcbor::from_slice::<Typed>(bytes).ok()?.ty
}
#[cfg(test)]
mod tests {
use super::*;
use crate::car::compute_cid;
use crate::filter::FilterMode;
use smol_str::SmolStr;
fn record(collection: &str, n: usize) -> Vec<u8> {
serde_ipld_dagcbor::to_vec(&serde_json::json!({ "$type": collection, "n": n })).unwrap()
}
/// a whole repo car: the commit, every mst node and every record, keyed `collection/rkey`
async fn repo_car(records: &[(String, Vec<u8>)]) -> (Bytes, IpldCid) {
let (mst_root, mst_nodes) = crate::mst::build_tree(
records
.iter()
.map(|(key, body)| (key.as_str(), compute_cid(body))),
)
.unwrap();
let commit = crate::mst::AtpCommit {
did: jacquard_common::types::did::Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")
.unwrap(),
version: 3,
data: mst_root,
rev: "3l2abcdefgh22".parse().unwrap(),
prev: None,
sig: Bytes::new(),
};
let commit_cid = commit.to_cid().unwrap();
let mut car = Vec::new();
let mut writer =
crate::car::CarWriter::new(crate::car::CarHeader::new_v1(vec![commit_cid]), &mut car);
writer
.write(commit_cid, commit.to_cbor().unwrap())
.await
.unwrap();
for (cid, bytes) in mst_nodes {
writer.write(cid, &bytes).await.unwrap();
}
for (_, body) in records {
writer.write(compute_cid(body), body).await.unwrap();
}
writer.finish().await.unwrap();
(car.into(), mst_root)
}
fn filter(collections: &[&str]) -> Arc<FilterConfig> {
let mut filter = FilterConfig::new(FilterMode::Filter);
filter.collections = collections.iter().map(SmolStr::new).collect();
Arc::new(filter)
}
fn stream(car: &[u8], collections: &[&str], chunk: usize) -> FilteredRepo {
let mut filtered = FilteredCar::new(filter(collections), true).unwrap();
for piece in car.chunks(chunk) {
filtered.push(piece).unwrap();
}
filtered.finish().unwrap()
}
fn scan(
mut repo: FilteredRepo,
mst_root: IpldCid,
) -> (Vec<(SmolStr, IpldCid)>, Blocks<CommitBlocks>) {
let leaves = repo
.scanner
.scan(mst_root)
.unwrap()
.unwrap()
.leaves
.iter()
.map(|(k, cid)| (SmolStr::new(k.as_str()), *cid))
.collect();
(leaves, repo.blocks)
}
fn mixed_repo() -> Vec<(String, Vec<u8>)> {
(0..2000usize)
.map(|i| {
let collection = match i % 400 {
0 => "sh.tangled.repo",
200 => "sh.tangled.repo.collaborator",
n if n % 3 == 0 => "app.bsky.feed.post",
_ => "app.bsky.feed.like",
};
(format!("{collection}/{i:013}"), record(collection, i))
})
.collect()
}
#[tokio::test]
async fn only_keeps_the_commit_and_the_blocks_it_was_asked_for() {
let records = mixed_repo();
let (car, _) = repo_car(&records).await;
let asked = [&records[3].1, &records[1999].1].map(|body| compute_cid(body));
let root = crate::car::parse_car(car.clone(), false).unwrap().root;
let mut only = FilteredCar::only(root, asked.into_iter().collect(), true);
for piece in car.chunks(64) {
only.push(piece).unwrap();
}
let repo = only.finish().unwrap();
let mut kept = repo.blocks.map().keys().copied().collect::<HashSet<_>>();
assert!(kept.remove(&repo.root), "the commit wasn't kept");
assert_eq!(kept, asked.into_iter().collect());
assert_eq!(repo.stats.kept_blocks, 3);
assert_eq!(repo.scanner.ruled_out_len(), 0);
}
#[tokio::test]
async fn only_stops_at_the_first_block_of_another_commit() {
let (car, _) = repo_car(&mixed_repo()).await;
let other = compute_cid(b"another commit");
let mut only = FilteredCar::only(other, HashSet::new(), true);
let pieces = car.chunks(64).collect::<Vec<_>>();
let failed_at = pieces
.iter()
.position(|piece| only.push(piece).is_err())
.expect("a car of another commit is refused");
assert!(
failed_at < pieces.len() / 100,
"stopped at piece {failed_at}"
);
}
#[test]
fn takes_every_block_when_the_filter_takes_every_collection() {
assert!(FilteredCar::new(filter(&[]), true).is_none());
}
#[tokio::test]
async fn keeps_only_the_commit_wanted_records_and_nodes_that_can_reach_them() {
let records = mixed_repo();
let (car, mst_root) = repo_car(&records).await;
let wanted = ["sh.tangled.repo", "sh.tangled.repo.collaborator"];
let want = records
.iter()
.filter(|(key, _)| wanted.iter().any(|c| key.split('/').next() == Some(*c)))
.map(|(key, body)| (SmolStr::new(key), compute_cid(body)))
.collect::<std::collections::BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
assert_eq!(want.len(), 10);
for chunk in [1, 7, 4096, car.len()] {
let repo = stream(&car, &wanted, chunk);
let stats = repo.stats;
assert!(
repo.blocks.get(&repo.root).is_some(),
"chunk {chunk}: commit kept"
);
// only the commit and the wanted records stay in `blocks`; mst nodes are walked
// and dropped as the car streams
assert_eq!(
stats.kept_blocks,
1 + want.len(),
"chunk {chunk}: kept {} blocks",
stats.kept_blocks
);
assert!(repo.scanner.is_complete(), "chunk {chunk}");
assert_eq!(repo.scanner.parked_bytes(), 0, "chunk {chunk}");
assert_eq!(repo.scanner.ruled_out_len(), 0, "chunk {chunk}");
let (leaves, blocks) = scan(repo, mst_root);
assert_eq!(leaves, want, "chunk {chunk}");
for (key, cid) in &leaves {
let body = &records.iter().find(|(k, _)| k == key).unwrap().1;
assert_eq!(
blocks.get(cid).unwrap().as_ref(),
body.as_slice(),
"chunk {chunk}"
);
}
for (key, body) in records
.iter()
.filter(|(key, _)| key.starts_with("app.bsky."))
{
let cid = compute_cid(body);
assert!(blocks.get(&cid).is_none(), "chunk {chunk}: kept {key}");
}
}
}
#[tokio::test]
async fn a_repo_without_wanted_records_keeps_none_of_them() {
let records = (0..2000usize)
.map(|i| {
(
format!("io.atcr.hold.layer/{i:013}"),
record("io.atcr.hold.layer", i),
)
})
.collect::<Vec<_>>();
let (car, mst_root) = repo_car(&records).await;
let repo = stream(&car, &["sh.tangled.repo"], 64);
let stats = repo.stats;
let (leaves, blocks) = scan(repo, mst_root);
assert!(leaves.is_empty());
for (key, body) in &records {
assert!(
blocks.get(&compute_cid(body)).is_none(),
"kept {key}"
);
}
assert!(
stats.kept_bytes * 3 < stats.bytes,
"kept {} of {} bytes",
stats.kept_bytes,
stats.bytes
);
}
#[tokio::test]
async fn a_record_filed_under_another_type_is_left_for_the_scan_to_find_missing() {
let mut records = mixed_repo();
let odd = record("app.bsky.feed.post", 9999);
records.push(("sh.tangled.repo/odd".to_string(), odd.clone()));
let (car, mst_root) = repo_car(&records).await;
let (leaves, blocks) = scan(stream(&car, &["sh.tangled.repo"], 512), mst_root);
let odd_cid = compute_cid(&odd);
assert!(leaves.contains(&(SmolStr::new("sh.tangled.repo/odd"), odd_cid)));
assert!(blocks.get(&odd_cid).is_none());
}
#[tokio::test]
async fn out_of_order_nodes_drop_when_complete_and_resume_when_evicted() {
let records = mixed_repo();
let (car, mst_root) = repo_car(&records).await;
let parsed = crate::car::parse_car(car, true).unwrap();
let commit_bytes = parsed.blocks.get(&parsed.root).unwrap().clone();
// an unreachable historical mst node whose open left subtree could still reach
// `sh.tangled.repo` without ancestor bounds
let historical = serde_ipld_dagcbor::to_vec(&crate::mst::tests::node(
vec![("zz.historical/1", parsed.root, None)],
Some(parsed.root),
))
.unwrap();
let historical_cid = compute_cid(&historical);
// write every non-commit block before the commit so all mst nodes arrive before their
// parents and the walk only starts at the very end of the car
let mut reversed_car = Vec::new();
let mut writer = crate::car::CarWriter::new(
crate::car::CarHeader::new_v1(vec![parsed.root]),
&mut reversed_car,
);
writer.write(historical_cid, &historical).await.unwrap();
for (cid, body) in parsed.blocks.map() {
if *cid != parsed.root {
writer.write(*cid, body).await.unwrap();
}
}
writer.write(parsed.root, &commit_bytes).await.unwrap();
writer.finish().await.unwrap();
// with the normal budget, out-of-order nodes resolve when the commit arrives and the
// historical node is dropped once the walk completes
let mut full_budget =
FilteredCar::new(filter(&["sh.tangled.repo"]), true).unwrap();
full_budget.push(&reversed_car).unwrap();
let repo = full_budget.finish().unwrap();
assert!(repo.scanner.is_complete());
assert!(!repo.scanner.evicted_any());
assert_eq!(repo.scanner.parked_bytes(), 0);
assert_eq!(repo.scanner.ruled_out_len(), 0);
let (leaves, _) = scan(repo, mst_root);
assert_eq!(leaves.len(), 5);
// with a 1-byte budget, out-of-order nodes are evicted during the first pass and
// resolved level by level across `FilteredCar::resume` passes
let mut tiny_budget =
FilteredCar::with_parked_budget(filter(&["sh.tangled.repo"]), 1, true).unwrap();
tiny_budget.push(&reversed_car).unwrap();
let mut current = tiny_budget.finish().unwrap();
assert!(current.scanner.evicted_any());
assert!(current.scanner.scan(mst_root).unwrap().is_err());
let mut passes = 0usize;
let resumed_leaves = loop {
passes += 1;
assert!(passes <= 4, "took too many passes to resolve the tree");
let mut next_pass = FilteredCar::resume(current.root, current.scanner, true);
next_pass.push(&reversed_car).unwrap();
current = next_pass.finish().unwrap();
assert_eq!(current.stats.kept_blocks, 1);
if let Ok(scan) = current.scanner.scan(mst_root).unwrap() {
break scan.leaves;
}
};
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 = crate::car::CarWriter::new(
crate::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})"
);
}
}
}
}
}