[backfill] encode step and scan invariants in types
11234585 unmodified lines9192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421439714414514614714839 unmodified lines18818919019119219319419519619719819920020120220320 unmodified lines22422522622717417517617717817918018122822923023123223323423523618318423723823924018718824124224319024424524624724824925025125225325425525625754 unmodified lines3123133143152533163173183193202572583213223233243253264 unmodified lines33133233333427133533633733833934027634134234334434518 unmodified lines36436536636736836937037137237330437437537637737812 unmodified lines39139239339432539539639739839940033040140240340440524 unmodified lines4304314324333633644344354364374384394404414424434444454464474484 unmodified lines45345445545638338445745845946046146246346446546635 unmodified lines5025035045054284294304314325065075085095109 unmodified lines5205215225234505245255265275285295304565315325335345354615365375385395405415425435445455465475485495505515525537 unmodified lines5615625635645655665675685691 unmodified line57157257357449057557657757857917 unmodified lines5975985996006016026036046056066076086096 unmodified lines6166176186196206216226236246256266276286296305335345355365375386316326336346356366376386396406416426435465476446456466476486496506516527 unmodified lines6606616626635646646656666676686696705716716726736745755765775785795805815826756766776786796806816826836846855886866876886896902 unmodified lines693694695696599600601602697698699700701702703606607608609704705706707708709710711712713714715716717344 unmodified linesExpand all106210631064106596696710661067106810691070107110721073107410751076107710781079108097310811082108310841085108610871088108910901091109210931 unmodified line10951096109710989831099110011011102110353 unmodified lines11571158115911601045104611611162116311641165116611672 unmodified lines117011711172117311741175117611771178117911801181118211831184118511861187118811891190119144 unmodified linesuse std::collections::{BTreeMap, HashMap, HashSet, VecDeque};use std::collections::{HashMap, HashSet, VecDeque};use cid::Cid as IpldCid;use jacquard_repo::mst::util::layer_for_key;use miette::{Result, WrapErr};use smol_str::SmolStr;85 unmodified lines 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<(SmolStr, IpldCid)>);
impl SortedLeaves { fn from_raw(mut leaves: Vec<(SmolStr, 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<(SmolStr, IpldCid)>) -> Self { Self::from_raw(leaves) }}
impl std::ops::Deref for SortedLeaves { type Target = [(SmolStr, IpldCid)];
fn deref(&self) -> &Self::Target { &self.0 }}
impl<'a> IntoIterator for &'a SortedLeaves { type Item = &'a (SmolStr, IpldCid); type IntoIter = std::slice::Iter<'a, (SmolStr, IpldCid)>;
fn into_iter(self) -> Self::IntoIter { self.0.iter() }}
impl PartialEq<Vec<(SmolStr, IpldCid)>> for SortedLeaves { fn eq(&self, other: &Vec<(SmolStr, IpldCid)>) -> bool { &self.0 == other }}
impl PartialEq<[(SmolStr, IpldCid)]> for SortedLeaves { fn eq(&self, other: &[(SmolStr, IpldCid)]) -> bool { self.0.as_slice() == other }}
#[derive(Debug, Clone)]pub(crate) struct SparseScanOutput { pub(crate) leaves: Vec<(SmolStr, IpldCid)>, pub(crate) leaves: SortedLeaves, pub(crate) node_blocks_seen: usize, pub(crate) node_bytes_seen: usize,}
39 unmodified lines byte_len: usize, entries: Box<[FlatEntry]>,}
#[derive(Debug, Clone)]struct VisitedNode { bounds: BoundsSet, entries: Option<Box<[FlatEntry]>>,}
type CidMap<V> = HashMap<IpldCid, V, ahash::RandomState>;type CidSet = HashSet<IpldCid, ahash::RandomState>;
impl NodeBounds { fn unbounded() -> Self { Self {20 unmodified lines 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 { /// 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<&str>, upper_leaf: Option<&str>, ranges: &[KeyRange], ) -> Option<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), (Some(a), Some(b)) => Some(a.max(b)), (a, b) => a.or(b), }; let upper = match (self.upper.as_deref(), upper_leaf) { (Some(a), Some(b)) => Some(SmolStr::new(a.min(b))), (a, b) => a.or(b).map(SmolStr::new), (Some(a), Some(b)) => Some(a.min(b)), (a, b) => a.or(b), }; Self { lower, upper } if lower.zip(upper).is_some_and(|(l, u)| l >= u) { return None; } if !ranges.iter().any(|range| range.intersects_subtree(lower, upper)) { return None; } Some(Self { lower: lower.map(SmolStr::new), upper: upper.map(SmolStr::new), }) }}
/// whether a node seen on its own, its place in the tree unknown, could hold a key in54 unmodified lines parked: CidMap<ParkedNode>, parked_order: VecDeque<IpldCid>, parked_bytes: usize, parked_budget: usize, evicted: bool, evicted: CidSet, corrupt: CidMap<String>, ruled_out: RuledOut, root: Option<IpldCid>, pending_missing: CidMap<BoundsSet>, visited: CidMap<BoundsSet>, leaves: BTreeMap<SmolStr, IpldCid>, visited: CidMap<VisitedNode>, leaves: Vec<(SmolStr, IpldCid)>, stats: ScanStats,}
impl SparseScanner {4 unmodified lines parked: CidMap::default(), parked_order: VecDeque::new(), parked_bytes: 0, parked_budget: usize::MAX, evicted: false, evicted: CidSet::default(), corrupt: CidMap::default(), ruled_out: RuledOut::default(), root: None, pending_missing: CidMap::default(), visited: CidMap::default(), leaves: BTreeMap::new(), leaves: Vec::new(), stats: ScanStats::default(), } }
18 unmodified lines 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.contains(cid) }
#[cfg(test)] pub(crate) fn evicted_any(&self) -> bool { self.evicted !self.evicted.is_empty() }
pub(crate) fn is_pending(&self, cid: &IpldCid) -> bool { self.pending_missing.contains_key(cid)12 unmodified lines pub(crate) fn prepare_pass(&mut self) { self.parked = CidMap::default(); self.parked_order = VecDeque::new(); self.parked_bytes = 0; self.evicted = false; self.evicted = CidSet::default(); self.corrupt = CidMap::default(); }
/// 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 || self.is_complete()) { if self.root.is_some() && (self.evicted.is_empty() || self.is_complete()) { self.clear_parked(); } }
24 unmodified lines if self.is_complete() { return Ok(()); } if let Some(bounds_set) = self.pending_missing.remove(&cid) { let entries = decode_node(bytes).wrap_err_with(|| format!("failed to decode mst node {cid}"))?; self.evicted.remove(&cid); let entries = decode_node(bytes) .wrap_err_with(|| format!("failed to decode mst node {cid}"))? .into_boxed_slice(); 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); } } self.store_visited_entries(cid, entries); self.drain_stack(&mut stack)?; if self.is_complete() { self.clear_parked(); }4 unmodified lines || self.parked.contains_key(&cid) { return Ok(()); } let Ok(entries) = decode_node(bytes) else { return Ok(()); let entries = match decode_node(bytes) { Ok(entries) => entries, Err(err) => { 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 node_may_reach(&self.ranges, &entries) || !self.ruled_out.insert(&cid) { self.park(35 unmodified lines
if self.pending_missing.is_empty() { self.clear_parked(); Ok(Ok(SparseScanOutput { leaves: self .leaves .iter() .map(|(key, cid)| (key.clone(), *cid)) .collect(), 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 {9 unmodified lines
fn is_visited(&self, cid: &IpldCid, bounds: &NodeBounds) -> bool { self.visited .get(cid) .is_some_and(|seen| seen.covers(bounds)) .is_some_and(|seen| seen.bounds.covers(bounds)) }
fn record_visited(&mut self, cid: IpldCid, byte_len: usize, bounds: NodeBounds) { self.evicted.remove(&cid); match self.visited.entry(cid) { std::collections::hash_map::Entry::Occupied(mut occupied) => { occupied.get_mut().insert(bounds); occupied.get_mut().bounds.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)); vacant.insert(VisitedNode { bounds: BoundsSet::new(bounds), entries: None, }); } } }
fn store_visited_entries(&mut self, cid: IpldCid, entries: Box<[FlatEntry]>) { if let Some(visited) = self.visited.get_mut(&cid) { visited.entries = Some(entries); } }
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);7 unmodified lines fn park(&mut self, cid: IpldCid, node: ParkedNode) { if self.parked.contains_key(&cid) { return; } self.evicted.remove(&cid); self.parked_bytes += node.byte_len; self.parked.insert(cid, node); self.parked_order.push_back(cid); while self.parked_bytes > self.parked_budget {1 unmodified line break; }; if let Some(evicted) = self.parked.remove(&oldest) { self.parked_bytes -= evicted.byte_len; self.evicted = true; self.evicted.insert(oldest); } } }
17 unmodified lines fn clear_parked(&mut self) { self.parked = CidMap::default(); self.parked_order = VecDeque::new(); self.parked_bytes = 0; self.corrupt = CidMap::default(); self.ruled_out = RuledOut::default(); for visited in self.visited.values_mut() { visited.entries = None; } }
/// 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.6 unmodified lines while let Some((cid, bounds)) = stack.pop() { if self.is_visited(&cid, &bounds) || self.ruled_out.contains(&cid) { continue; } if let Some(entries) = self .visited .get_mut(&cid) .and_then(|visited| visited.entries.take()) { self.visit_entries(cid, 0, &entries, bounds, stack); self.store_visited_entries(cid, entries); continue; } if let Some(node) = self.unpark(&cid) { self.visit_entries(cid, node.byte_len, &node.entries, bounds, stack); while let Some(pos) = stack.iter().rposition(|(other, _)| *other == cid) { let (_, other_bounds) = stack.remove(pos); if !self.is_visited(&cid, &other_bounds) { self.visit_entries(cid, node.byte_len, &node.entries, other_bounds, stack); } } self.store_visited_entries(cid, node.entries); continue; } if let Some(err) = self.corrupt.get(&cid) { return Err( miette::miette!("{err}").wrap_err(format!("failed to decode mst node {cid}")) ); } let Some(bytes) = self.blocks.get(&cid).cloned() else { self.record_pending(cid, bounds); continue; };
let entries = decode_node(&bytes).wrap_err_with(|| format!("failed to decode mst node {cid}"))?; let entries = decode_node(&bytes) .wrap_err_with(|| format!("failed to decode mst node {cid}"))? .into_boxed_slice(); self.visit_entries(cid, bytes.len(), &entries, bounds, stack); self.store_visited_entries(cid, entries); }
Ok(()) }7 unmodified lines stack: &mut Vec<(IpldCid, NodeBounds)>, ) { self.record_visited(cid, byte_len, bounds.clone());
let mut children = Vec::new(); 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.insert(key.clone(), *cid); self.leaves.push((key.clone(), *cid)); } } FlatEntry::Tree { cid } => { let child = bounds.child(previous_leaf(entries, idx), next_leaf(entries, idx)); if child.is_non_empty() && self.ranges.iter().any(|range| { range.intersects_subtree(child.lower.as_deref(), child.upper.as_deref()) }) { children.push((*cid, child)); if let Some(child) = bounds.narrow( previous_leaf(entries, idx), next_leaf(entries, idx), &self.ranges, ) { stack.push((*cid, child)); } } } } // reversed, so subtrees are scanned left to right, depth first stack.extend(children.into_iter().rev()); stack[stack_start..].reverse(); }}
#[derive(Debug, Default)]2 unmodified lines node_bytes_seen: usize,}
fn previous_leaf(entries: &[FlatEntry], idx: usize) -> Option<&str> { entries[..idx].iter().rev().find_map(|entry| match entry { FlatEntry::Leaf { key, .. } => Some(key.as_str()), FlatEntry::Tree { .. } => None, }) match idx.checked_sub(1).and_then(|prev| entries.get(prev)) { Some(FlatEntry::Leaf { key, .. }) => Some(key.as_str()), _ => None, }}
fn next_leaf(entries: &[FlatEntry], idx: usize) -> Option<&str> { entries[idx + 1..].iter().find_map(|entry| match entry { FlatEntry::Leaf { key, .. } => Some(key.as_str()), FlatEntry::Tree { .. } => None, }) match entries.get(idx + 1) { Some(FlatEntry::Leaf { key, .. }) => Some(key.as_str()), _ => None, }}
#[cfg(test)]mod tests { use std::collections::BTreeMap;
use super::*; use crate::mst::tests::{cid, left_chain, node}; use bytes::Bytes; use jacquard_repo::mst::NodeData;344 unmodified linesExpand all ] ); assert_eq!(scan.node_blocks_seen, 2);
// the same shared node arriving before its parent in a streamed car must still be // visited under both ancestor bounds before it is unparked for good // 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(root, &car_bytes(&root_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,1 unmodified line (SmolStr::new("aa.x/1"), cid(6)), (SmolStr::new("zz.y/1"), cid(7)), ] ); assert_eq!(scan.node_blocks_seen, 2); assert_eq!(scan.node_blocks_seen, 4); assert_eq!(streamed.parked_bytes(), 0); }
#[test]53 unmodified lines (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); // 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();2 unmodified lines 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() {44 unmodified lines7 unmodified lines8910111213121314171819202122151617181920212243 unmodified lines66676869747071727374757677788179808182838485868887888990919212 unmodified lines10510610710810911011111211311411511610911011111218 unmodified lines13113213313414313513613713813939 unmodified lines179180181182191183184185186187188103 unmodified linesExpand all2922932942953032962972982993001 unmodified line7 unmodified linesuse miette::Result;use smol_str::SmolStr;use tracing::trace;
use crate::db::types::DbAction;
/// one write that brings the stored records in line with the new repo#[derive(Debug)]pub(crate) enum RecordOp<K, T> { /// a record that's new, or whose cid differs from the stored one Put { key: SmolStr, record: T, action: DbAction, }, /// a record that `stored` does not have Create { key: SmolStr, record: T }, /// a record whose cid differs from the stored one Update { key: SmolStr, record: T }, /// a stored record the new repo doesn't have Delete(K),}
43 unmodified lines self.next_stored = self.stored.next().transpose()?; } let (new, stored) = match (self.next_new.take(), self.next_stored.take()) { (None, None) => return Ok(None), (Some(new), None) => return Ok(Some(put(new, DbAction::Create))), (Some((key, _, record)), None) => { return Ok(Some(RecordOp::Create { key, record })); } (None, Some((stored, _))) => return Ok(Some(RecordOp::Delete(stored))), (Some(new), Some(stored)) => (new, stored), }; match new.0.as_bytes().cmp(stored.0.as_ref()) { Ordering::Less => { self.next_stored = Some(stored); return Ok(Some(put(new, DbAction::Create))); let (key, _, record) = new; return Ok(Some(RecordOp::Create { key, record })); } Ordering::Greater => { self.next_new = Some(new); return Ok(Some(RecordOp::Delete(stored.0))); } Ordering::Equal if new.1 != stored.1 => { return Ok(Some(put(new, DbAction::Update))); let (key, _, record) = new; return Ok(Some(RecordOp::Update { key, record })); } Ordering::Equal => trace!(key = %new.0, cid = %new.1, "skip unchanged record"), } }12 unmodified lines Ok(Some(next)) }}
fn put<K, T>((key, _, record): (SmolStr, IpldCid, T), action: DbAction) -> RecordOp<K, T> { RecordOp::Put { key, record, action, }}
impl<N, S, K, T> Iterator for Diff<N, S, K, T>where N: Iterator<Item = Result<(SmolStr, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>,18 unmodified lines use smol_str::ToSmolStr;
use super::*; use crate::db::keys; use crate::db::types::{DbRkey, DbTid}; use crate::db::types::{DbAction, DbRkey, DbTid}; use crate::mst::tests::cid;
/// every stored record in a map, each new record taken out of it, and whatever stays behind /// deleted. returns (key, action) in key order39 unmodified lines .map(|(key, cid)| Ok((SmolStr::new(key), *cid, ()))); let stored = stored.iter().map(|(key, cid)| Ok((key.clone(), *cid))); diff(new, stored) .map(|op| match op.unwrap() { RecordOp::Put { key, action, .. } => (key.to_string(), action.as_str()), RecordOp::Create { key, .. } => (key.to_string(), "create"), RecordOp::Update { key, .. } => (key.to_string(), "update"), RecordOp::Delete(key) => (key, "delete"), }) .collect() }103 unmodified linesExpand all let mut ops = diff( new.into_iter(), std::iter::empty::<Result<(Vec<u8>, IpldCid)>>(), ); assert!(matches!(ops.next(), Some(Ok(RecordOp::Put { .. })))); assert!(matches!(ops.next(), Some(Ok(RecordOp::Create { .. })))); let err = ops.next().unwrap().unwrap_err(); assert!(err.to_string().contains("does not sort after"), "{err}"); assert!(ops.next().is_none()); }1 unmodified line208 unmodified linesExpand all2092102112122132132142152162172182197 unmodified lines22722822923023123223323423523623723823924060 unmodified lines301302303304297305306307308309347 unmodified linesExpand all208 unmodified linesExpand all filter .as_ref() .is_some_and(|filter| filter.matches_collection(collection)) } else { if !scanner.is_complete() { scanner.offer_block(*cid, bytes)?; scanner.offer_block(*cid, bytes)?; } false } } })7 unmodified lines 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)]60 unmodified lines fn scan( mut repo: FilteredRepo, mst_root: IpldCid, ) -> (Vec<(SmolStr, IpldCid)>, Blocks<CommitBlocks>) { let leaves = repo.scanner.scan(mst_root).unwrap().unwrap().leaves; let leaves = repo.scanner.scan(mst_root).unwrap().unwrap().leaves.to_vec(); (leaves, repo.blocks) }
fn mixed_repo() -> Vec<(String, Vec<u8>)> {347 unmodified linesExpand all147 unmodified linesExpand all148149150151152152153154155156157158159159160160161162163164165166167168169170171223 unmodified linesExpand all3953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454461504 unmodified linesExpand all147 unmodified linesExpand all )) }
/// a backfill's records for one of its steps. its shard read the exclusion once, at the /// first step, so this never reads it /// first step, and [`BackfillStepTxn::apply`] only accepts a [`DiffStep`] minted by /// `diff(new, StoredHeads)` #[cfg(feature = "indexer")] pub(crate) fn backfill_step_records<'txn, 'did>( &'txn mut self, state: &AppState, commit_rev: &DbTid, did: &'did Did, ) -> Result<RecordTxn<'txn, 'db, 'did>> { Ok(self.record_txn(state, commit_rev, did, RecordEventOrigin::Backfill, false)) ) -> BackfillStepTxn<'txn, 'db, 'did> { BackfillStepTxn(self.record_txn( state, commit_rev, did, RecordEventOrigin::Backfill, false, )) }
#[cfg(feature = "indexer")] fn record_txn<'txn, 'did>(223 unmodified linesExpand all count_ephemeral_records: bool, excluded: bool,}
/// a backfill step's record transaction on its owner shard, which only accepts a [`DiffStep`]/// minted by `diff(new, StoredHeads)` where keys are strictly increasing and `Create` is/// proven absent from `heads`#[cfg(feature = "indexer")]pub(crate) struct BackfillStepTxn<'txn, 'db, 'did>(RecordTxn<'txn, 'db, 'did>);
#[cfg(feature = "indexer")]impl BackfillStepTxn<'_, '_, '_> { pub(crate) fn apply( &mut self, step: crate::backfill::worker::apply::DiffStep, ) -> Result<()> { use crate::ingest::indexer::message::RecordWrite;
for write in step.into_writes() { match write { RecordWrite::Create { collection, rkey, block, } => self.0.put_record(&collection, &rkey, &block, DbAction::Create)?, RecordWrite::Update { collection, rkey, block, } => self.0.put_record(&collection, &rkey, &block, DbAction::Update)?, RecordWrite::Delete { collection, rkey } => { self.0.delete_record(&collection, &rkey)? } } } Ok(()) }
pub(crate) fn read_repo(&mut self) -> Result<Option<RepoState<'static>>> { self.0.read_repo() }
pub(crate) fn update_repo_state(&mut self, state: &RepoState<'_>) -> Result<()> { self.0.update_repo_state(state) }}
#[cfg(feature = "indexer")]impl RecordTxn<'_, '_, '_> { pub(crate) fn is_excluded(&self) -> bool { self.excluded }1504 unmodified linesExpand all10 unmodified lines1112131415151617181921 unmodified lines41424344454647484950515253545556575859606162636465665051526768697071727374757660617778798065666781828384857 unmodified lines939495968397989910010110210321 unmodified lines12512612712811312913013113213339 unmodified lines173174175176161162163164165177178167179180181182171183184185186187188189190191192193194195196102 unmodified linesExpand all10 unmodified linesuse jacquard_common::types::did::Did;use miette::{IntoDiagnostic, Result};use tracing::trace;
use super::records::{NewRecord, RepoRecords, StoredKey, new_record};use super::records::{CheckedRecords, NewRecord, StoredKey};use crate::backfill::diff::{RecordOp, diff};use crate::db::keys;use crate::db::types::{DbRkey, DbTid};use crate::filter::FilterConfig;21 unmodified lines /// the pds says the repo is inactive, which the fetch wrote down instead of any records Inactive,}
/// one step of writes minted by [`write`] from [`diff`] against [`StoredHeads`]: keys are/// strictly increasing, blocks are cid-verified, and `Create` means `StoredHeads` saw no/// stored head for the key#[derive(Debug)]pub(crate) struct DiffStep(Vec<RecordWrite>);
impl DiffStep { pub(crate) fn into_writes(self) -> Vec<RecordWrite> { self.0 }
#[cfg(test)] pub(crate) fn empty() -> Self { Self(Vec::new()) }}
/// record bytes a step stops at, short of its chunk of records. the records come from the pds,/// so a chunk of big ones would otherwise put far more than a step's share of the budget on the/// heap and into one shard commitpub(super) const STEP_BYTES: usize = 4 * 1024 * 1024;
/// sends `records`' writes to `did`'s shard, a chunk of records or [`STEP_BYTES`] per step, and/// waits for each answer. the last step also carries `tail`, so a repo whose writes fit in one/// step stores its records in one batch/// sends `checked`'s writes to `did`'s shard, a chunk of records or [`STEP_BYTES`] per step,/// and waits for each answer. the last step also carries `tail`, so a repo whose writes fit in/// one step stores its records in one batch#[allow(clippy::too_many_arguments)]pub(super) fn write( app_state: &AppState, indexer: &IndexerTx, did: &Did, pending_key: &Slice, rev: DbTid, records: &RepoRecords, filter: &FilterConfig, checked: &CheckedRecords, tail: BackfillTail,) -> Result<Outcome> { let chunk = app_state.backfill_chunk_records; let new = records .leaves() .filter_map(|leaf| new_record(leaf, records, filter).transpose()); let new = checked.new_records(); let stored = StoredHeads::new(app_state, did, chunk, tail.fetched_only.clone()); let mut writes = diff(new, stored) .map(|op| op.and_then(record_write)) .peekable();7 unmodified lines && let Some(write) = writes.next() { match write { Ok(write) => { if let RecordWrite::Put { block, .. } = &write { if let RecordWrite::Create { block, .. } | RecordWrite::Update { block, .. } = &write { step_bytes += block.size(); } step.push(write); }21 unmodified lines let chunk = BackfillChunk { did: did.clone(), pending_key: pending_key.clone(), rev, writes: step, writes: DiffStep(step), tail: if last { tail.take().map(Box::new) } else { None39 unmodified lines}
fn record_write(op: RecordOp<StoredKey, NewRecord>) -> Result<RecordWrite> { Ok(match op { RecordOp::Put { key, record, action, } => { RecordOp::Create { key, record } => { let (collection, rkey) = key.split_at(record.collection_len); RecordWrite::Put { RecordWrite::Create { collection: collection.into(), rkey: DbRkey::new(&rkey[1..]), block: record.block, action, } } RecordOp::Update { key, record } => { let (collection, rkey) = key.split_at(record.collection_len); RecordWrite::Update { collection: collection.into(), rkey: DbRkey::new(&rkey[1..]), block: record.block, } } RecordOp::Delete(stored) => { let (collection, rkey) = keys::split_record_suffix(stored.as_ref()) .map_err(|e| miette::miette!("invalid record key '{:?}': {e}", stored.key))?; RecordWrite::Delete {102 unmodified linesExpand all22 unmodified lines23242526272728293031399 unmodified linesExpand all431432433434435435436437438439440441442443444445446447442443444445448449450449451452453452453454455456457458459460461454455456457458317 unmodified linesExpand all776777778779787788780781782783784785780 unmodified linesExpand all1566156715681569157715701571157215731574157515761577157815791580158115821583158415841585158615871588158915901591159215931594159515961593159715981599159616001601160216031604160518 unmodified lines22 unmodified lines fetch_blocks, process_did_sparse,};use crate::backfill::spill::{SpillFile, SpillingCar, Spills};use crate::backfill::worker::apply::{self, Outcome};use crate::backfill::worker::records::{FilteredRecords, RepoRecords, WholeBlocks, check_records};use crate::backfill::worker::records::{CheckOutcome, FilteredRecords, RepoRecords, WholeBlocks, check_records};use crate::car::{Blocks, ParsedCar, WholeCar};use crate::config::{BackfillStrategy, RateTier};use crate::db::{Txn as DbTxn, keys};use crate::filter::FilterConfig;399 unmodified linesExpand all
tokio::task::spawn_blocking(move || { // a first pass over every record before anything is sent, so a broken mst or a key // that doesn't parse fails with nothing written let checked = check_records(&records, &filter)?; let mut checked = match check_records(records, filter)? { CheckOutcome::Keep(checked) => checked, CheckOutcome::Discarded => { trace!("no signal-matching records found, discarding repo"); return Ok(Outcome::Discarded); } }; if checked.left_out > 0 { debug!( records = checked.left_out, "skipping records whose block the car left out" ); } if !checked.signal_seen { trace!(signals = ?filter.signals, "no signal-matching records found, discarding repo"); return Ok(Outcome::Discarded); } let tail = BackfillTail { before, after, collection_counts: checked.collection_counts, collection_counts: checked.take_collection_counts(), fetched_only, }; apply::write( &app_state, &indexer, &did, &pending_key, rev, &records, &filter, tail, ) apply::write(&app_state, &indexer, &did, &pending_key, rev, &checked, tail) }) .await .into_diagnostic()?? };317 unmodified linesExpand all let missing = match scan { Ok(scan) => break scan.leaves, Err(missing) => missing, }; if !scanner.evicted_any() { return Err(miette::miette!("mst node {} missing from car", missing[0]).into()); if let Some(cid) = missing.iter().find(|cid| !scanner.was_evicted(cid)) { return Err(miette::miette!("mst node {cid} missing from car").into()); } scan_rounds += 1; if scan_rounds > SPARSE_MAX_SCAN_ROUNDS { return Err(780 unmodified linesExpand all usize::from(!serve_get_blocks) ); }
// a missing mst node when nothing was evicted fails immediately without a refetch // a missing mst node (even when another out-of-order node was evicted) fails immediately // without calling getBlocks or refetching the car let stale_bytes = serde_ipld_dagcbor::to_vec(&crate::mst::tests::node( vec![("zz.other/1", record_cid, None)], Some(record_cid), )) .into_diagnostic()?; let stale_cid = jacquard_repo::mst::util::compute_cid(&stale_bytes).into_diagnostic()?; let mut missing_car = Vec::new(); let mut missing_writer = iroh_car::CarWriter::new( iroh_car::CarHeader::new_v1(vec![commit_cid]), &mut missing_car, ); missing_writer .write(mst_root, &root_bytes) .write(stale_cid, &stale_bytes) .await .into_diagnostic()?; missing_writer .write(commit_cid, &commit_bytes) .await .into_diagnostic()?; missing_writer .write(mst_root, &root_bytes) .await .into_diagnostic()?; missing_writer.finish().await.into_diagnostic()?;
let mut filtered = FilteredCar::new(filter, true).unwrap(); let mut filtered = FilteredCar::with_parked_budget(filter, 0, true).unwrap(); filtered.push(&missing_car)?; let repo = filtered.finish()?; assert!(!repo.scanner.evicted_any()); assert!(repo.scanner.evicted_any()); assert!(!repo.scanner.was_evicted(&child_cid));
let addr = spawn_get_repo(StatusCode::OK, car_bytes.to_vec()).await; let throttler = crate::util::throttle::Throttler::new(10); let (public_pds, http) = public_client(addr, &throttler);18 unmodified lines11 unmodified lines121314151616171819202122234 unmodified linesExpand all25725825926026026126226326426529 unmodified lines295296297298298299300301302303304305306307308309310311299300301302303304479 unmodified linesExpand all784785786787799788789790791792126 unmodified linesExpand all11 unmodified linesuse tokio::sync::oneshot;use tracing::{debug, error};
use super::message::{ BackfillChunk, BackfillTail, BulkMessage, ChunkReply, EraseRequest, RecordWrite, shard_of, BackfillChunk, BackfillTail, BulkMessage, ChunkReply, EraseRequest, shard_of,};use crate::backfill::worker::apply::DiffStep;use super::shard::WorkerContext;use super::worker::FirehoseWorker;use crate::backfill::backfilled_state;use crate::db::types::DbTid;234 unmodified linesExpand all shard: &mut ShardState, did: &Did, pending_key: &Slice, rev: DbTid, writes: Vec<RecordWrite>, writes: DiffStep, tail: Option<Box<BackfillTail>>, ) -> Result<Written> { let db = &state.db; let mut txn = Txn::new(db);29 unmodified lines }, )?; }
let mut records = txn.backfill_step_records(state, &rev, did)?; for write in writes { match write { RecordWrite::Put { collection, rkey, block, action, } => records.put_record(&collection, &rkey, &block, action)?, RecordWrite::Delete { collection, rkey } => { records.delete_record(&collection, &rkey)? } } } let mut records = txn.backfill_step_records(state, &rev, did); records.apply(writes)?; let Some(tail) = tail else { txn.commit()?; return Ok(Written::Records); };479 unmodified linesExpand all let chunk = BackfillChunk { did: did.clone(), pending_key: keys::pending_key(7).to_vec().into(), rev: DbTid::new_from_bytes(9_u64.to_be_bytes()), writes: Vec::new(), writes: DiffStep::empty(), tail: None, reply, }; assert!(FirehoseWorker::write_chunk(&state, &mut shard, chunk).is_none());126 unmodified linesExpand all