//! a fetched repo's records as both passes of a backfill read them: the check before anything
//! is sent, and the merge against what's stored
use std::collections::HashMap;
use std::sync::Arc;
use bytes::Bytes;
use fjall::Slice;
use jacquard_common::types::cid::IpldCid;
use miette::Result;
use smol_str::{SmolStr, ToSmolStr};
use tracing::debug;
use crate::backfill::spill::SpilledCar;
use crate::car::{Blocks, CarBlock, CarBlocks, CommitBlocks};
use crate::filter::{FilterConfig, FilterMode};
use crate::mst::{BlockSource, MstKey};
use crate::ops;
use crate::sparse_mst::SortedLeaves;
/// a fetched repo's records, as leaves in key order and the blocks they name. the leaves are
/// read again for each pass, so a whole car is never copied into a list of its records
pub(super) enum RepoRecords {
Whole {
blocks: WholeBlocks,
mst_root: IpldCid,
},
Filtered(FilteredRecords),
}
/// what a filtered car's or a sparse fetch's scan found, with the blocks it brought back for
/// what the car dropped
pub(crate) struct FilteredRecords {
pub(super) leaves: SortedLeaves,
pub(super) blocks: Blocks<CommitBlocks>,
pub(super) fetched: Blocks<CommitBlocks>,
pub(super) whole: Option<Blocks<CommitBlocks>>,
}
impl FilteredRecords {
pub(crate) fn new(
leaves: SortedLeaves,
blocks: Blocks<CommitBlocks>,
fetched: Blocks<CommitBlocks>,
) -> Self {
Self {
leaves,
blocks,
fetched,
whole: None,
}
}
}
impl RepoRecords {
fn leaves(&self) -> Box<dyn Iterator<Item = Result<(MstKey, IpldCid)>> + '_> {
match self {
RepoRecords::Whole { blocks, mst_root } => {
Box::new(crate::mst::walk(blocks, *mst_root))
}
RepoRecords::Filtered(repo) => Box::new(repo.leaves.iter().cloned().map(Ok)),
}
}
/// the block under `cid`. a spilled car reads and hashes it now, which fails if the disk
/// changed it since it landed
fn block(&self, cid: &IpldCid) -> Result<Option<CarBlock>> {
match self {
RepoRecords::Whole { blocks, .. } => blocks.block(cid),
RepoRecords::Filtered(repo) => Ok(repo
.blocks
.block(cid)
.or_else(|| repo.fetched.block(cid))
.or_else(|| repo.whole.as_ref().and_then(|whole| whole.block(cid)))),
}
}
/// whether the car has the block under `cid`, without reading it off disk
fn contains(&self, cid: &IpldCid) -> bool {
match self {
RepoRecords::Whole { blocks, .. } => blocks.contains(cid),
RepoRecords::Filtered(repo) => {
repo.blocks.get(cid).is_some()
|| repo.fetched.get(cid).is_some()
|| repo
.whole
.as_ref()
.is_some_and(|whole| whole.get(cid).is_some())
}
}
}
}
/// a whole car's blocks, in memory or spilled to disk
pub(super) enum WholeBlocks {
Memory(Blocks<CarBlocks>),
Spilled(SpilledCar),
}
impl WholeBlocks {
fn block(&self, cid: &IpldCid) -> Result<Option<CarBlock>> {
match self {
WholeBlocks::Memory(blocks) => Ok(blocks.block(cid)),
WholeBlocks::Spilled(spilled) => spilled.block(cid),
}
}
fn contains(&self, cid: &IpldCid) -> bool {
match self {
WholeBlocks::Memory(blocks) => blocks.get(cid).is_some(),
WholeBlocks::Spilled(spilled) => spilled.contains(cid),
}
}
pub(super) fn len(&self) -> usize {
match self {
WholeBlocks::Memory(blocks) => blocks.map().len(),
WholeBlocks::Spilled(spilled) => spilled.len(),
}
}
}
impl BlockSource for WholeBlocks {
fn read(&self, cid: &IpldCid) -> Result<Option<Bytes>> {
match self {
WholeBlocks::Memory(blocks) => Ok(blocks.get(cid).cloned()),
WholeBlocks::Spilled(spilled) => spilled.read(cid),
}
}
}
/// what the first pass over a repo's records found
pub(super) enum CheckOutcome {
/// every present leaf parsed as `collection/rkey` and the repo matched any signal filter
Keep(CheckedRecords),
/// a signal filter found nothing it wants, so nothing is written
Discarded,
}
/// a repo's records after [`check_records`] verified every present leaf's path and matched
/// any configured signal. only [`CheckedRecords`] can yield [`NewRecord`]s for
/// [`super::apply::write`], so the second pass cannot run on unchecked paths or a different
/// filter
pub(super) struct CheckedRecords {
records: RepoRecords,
filter: Arc<FilterConfig>,
collection_counts: HashMap<SmolStr, u64>,
/// leaves whose block the car didn't have, which the write skips
pub(super) left_out: usize,
}
impl CheckedRecords {
pub(super) fn new_records(
&self,
) -> impl Iterator<Item = Result<(MstKey, IpldCid, NewRecord)>> + '_ {
self.records
.leaves()
.filter_map(|leaf| new_record(leaf, &self.records, &self.filter).transpose())
}
pub(super) fn take_collection_counts(&mut self) -> HashMap<SmolStr, u64> {
std::mem::take(&mut self.collection_counts)
}
}
/// reads every record the way the write will, without writing anything
pub(super) fn check_records(
records: RepoRecords,
filter: Arc<FilterConfig>,
) -> Result<CheckOutcome> {
let mut signal_seen = filter.mode == FilterMode::Full || filter.signals.is_empty();
let mut collection_counts = HashMap::new();
let mut left_out = 0;
for leaf in records.leaves() {
let (key, cid) = leaf?;
if !records.contains(&cid) {
left_out += 1;
continue;
}
let (collection, _) = ops::parse_path(&key)?;
if !filter.matches_collection(collection) {
continue;
}
if !signal_seen && filter.matches_signal(collection) {
debug!(collection = %collection, "signal matched");
signal_seen = true;
}
*collection_counts.entry(collection.to_smolstr()).or_default() += 1;
}
if !signal_seen {
return Ok(CheckOutcome::Discarded);
}
Ok(CheckOutcome::Keep(CheckedRecords {
records,
filter,
collection_counts,
left_out,
}))
}
/// a record of the new repo that the backfill keeps, with its block already cid-verified
pub(super) struct NewRecord {
/// where the collection ends in the record's key
pub(super) collection_len: usize,
pub(super) block: CarBlock,
}
/// `leaf` as the diff takes it, or none if the car left its block out or the filter doesn't
/// want its collection
fn new_record(
leaf: Result<(MstKey, IpldCid)>,
records: &RepoRecords,
filter: &FilterConfig,
) -> Result<Option<(MstKey, IpldCid, NewRecord)>> {
let (key, cid) = leaf?;
let Some(block) = records.block(&cid)? else {
return Ok(None);
};
let block = block
.checked()
.map_err(|e| miette::miette!("record {key}: {e}"))?;
// check_records already ran parse_path on every leaf that has a block, over the same
// immutable blocks, so its regexes needn't run twice
let (collection, _) = key
.split_once('/')
.ok_or_else(|| miette::miette!("record key {key} has no collection"))?;
if !filter.matches_collection(collection) {
return Ok(None);
}
let collection_len = collection.len();
Ok(Some((
key,
cid,
NewRecord {
collection_len,
block,
},
)))
}
/// a stored record's key, which the diff reads from after the did prefix
pub(super) struct StoredKey {
pub(super) key: Slice,
pub(super) suffix: usize,
}
impl AsRef<[u8]> for StoredKey {
fn as_ref(&self) -> &[u8] {
&self.key[self.suffix..]
}
}
#[cfg(test)]
mod tests {
use super::*;
/// a filtered car's records with every block present
fn filtered(leaves: &[(&str, &[u8])]) -> RepoRecords {
let cid = |body: &[u8]| crate::car::compute_cid(body);
RepoRecords::Filtered(FilteredRecords {
leaves: SortedLeaves::from_vec(
leaves
.iter()
.map(|(key, body)| (MstKey::new(key), cid(body)))
.collect(),
),
blocks: Blocks::unverified(
leaves
.iter()
.map(|(_, body)| (cid(body), Bytes::copy_from_slice(body)))
.collect(),
),
fetched: Blocks::default(),
whole: None,
})
}
// the scan only finds leaves inside the filter's ranges, so this is the only place these
// checks get an unwanted record, and a break in them would go unnoticed without it
#[test]
fn records_outside_the_filter_are_neither_counted_nor_written() {
let mut filter = crate::filter::FilterConfig::new(FilterMode::Filter);
filter.collections = vec![SmolStr::new("sh.tangled.repo")];
let records = filtered(&[
("app.bsky.feed.post/a", b"post"),
("sh.tangled.repo/core", b"core"),
]);
let CheckOutcome::Keep(mut checked) =
check_records(records, Arc::new(filter)).unwrap()
else {
panic!("expected records to pass check");
};
let kept = checked
.new_records()
.map(|record| record.unwrap().0)
.collect::<Vec<_>>();
assert_eq!(kept, ["sh.tangled.repo/core"]);
assert_eq!(
checked.take_collection_counts(),
HashMap::from([(SmolStr::new("sh.tangled.repo"), 1)])
);
}
}