[mst] inline mst keys, decode into scratch, and cut sparse scan allocations
36 unmodified lines37383940414142434445131 unmodified linesExpand all177178179180181181182183184185389 unmodified linesExpand all36 unmodified linesuse cid::Cid as IpldCid;use smol_str::SmolStr;
use car::{CarBlock, CarBlocks};use mst::FlatEntry;use mst::{FlatEntry, MstKey};use sparse_mst::{SparseScanner, sparse_ranges};
#[cfg(feature = "alloc-mimalloc")]static INNER: mimalloc::MiMalloc = mimalloc::MiMalloc;131 unmodified linesExpand all #[cfg(feature = "indexer")] reverse_car: OnceLock<Bytes>, mst_root: IpldCid, mst_nodes: Vec<(IpldCid, Bytes)>, leaves: Vec<(SmolStr, IpldCid)>, leaves: Vec<(MstKey, IpldCid)>,}
impl Repo { fn parse(&self, verify_cids: bool) -> car::ParsedCar<CarBlocks> {389 unmodified linesExpand all42 unmodified lines4344454647474849505115 unmodified lines67686970717172737475433 unmodified linesExpand all42 unmodified lines};use smol_str::SmolStr;
use car::{CarBlock, CarBlocks};use mst::FlatEntry;use mst::{FlatEntry, MstKey};use sparse_mst::{SparseScanner, sparse_ranges};
/// one repo that fits in any cache, one that fits a big l3, and one that fits none.const REPOS: [&str; 3] = ["atproto.com", "jay.bsky.team", "pfrazee.com"];15 unmodified lines mst_root: IpldCid, /// every reachable mst node in pre-order mst_nodes: Vec<(IpldCid, Bytes)>, /// every record in the repo, in key order leaves: Vec<(SmolStr, IpldCid)>, leaves: Vec<(MstKey, IpldCid)>,}
impl Repo { /// a parse for setting a benchmark up, outside what it measures433 unmodified linesExpand all12345678910111213814151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701714 unmodified lines17617717817918018118218318418518618718818919019119219319429195196197198199343520020120220320438394041205206207208209210211212213214215216217218219220221222223224225444546474849505152535422622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625756575859602582592602616263262263264265266267268656667686970269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317743183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034048540540640740840910 unmodified lines42042142242310442442542642710842842943043143210 unmodified lines4434444454461274474484491294504514524534544554564574581384594604614624638 unmodified lines47247347447515547647747847948037 unmodified lines51851952052120152252352452552613 unmodified lines54054154254322354454554654754854955055155255355455530 unmodified lines58658758858926259059159259359475 unmodified lines6706716726736746756766776786796806816826836846856866876886896906916926 unmodified lines69970070170270370470570670770870926 unmodified lines//! mst node decoding, shared by the full-repo walk here and the sparse scanner in//! `sparse_mst`.
use std::borrow::Borrow;use std::cmp::Ordering;use std::collections::HashSet;use std::fmt;use std::hash::{Hash, Hasher};use std::ops::Deref;use std::sync::Arc;
use bytes::Bytes;use cid::Cid as IpldCid;use jacquard_repo::mst::NodeData;use miette::{IntoDiagnostic, Result, WrapErr};use serde::Deserialize;use serde::de::{DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};use smol_str::SmolStr;
use crate::car::BlockMap;
/// inline capacity for mst keys (`collection/rkey`). 100% of keys across the benchmark/// corpus are <= 62 bytes (`size_of::<MstKey>() == 64`), so normal keys never touch the heap.const INLINE_KEY_CAP: usize = 62;
#[derive(Clone)]pub(crate) enum MstKey { Inline { len: u8, buf: [u8; INLINE_KEY_CAP] }, Heap(Arc<str>),}
impl MstKey { #[inline] pub(crate) fn new(s: impl AsRef<str>) -> Self { let s = s.as_ref(); let bytes = s.as_bytes(); if bytes.len() <= INLINE_KEY_CAP { let mut buf = [0u8; INLINE_KEY_CAP]; buf[..bytes.len()].copy_from_slice(bytes); Self::Inline { len: bytes.len() as u8, buf, } } else { Self::Heap(Arc::from(s)) } }
#[inline] pub(crate) fn as_bytes(&self) -> &[u8] { match self { Self::Inline { len, buf } => &buf[..usize::from(*len)], Self::Heap(s) => s.as_bytes(), } }
#[inline] pub(crate) fn as_str(&self) -> &str { match self { Self::Inline { len, buf } => { std::str::from_utf8(&buf[..usize::from(*len)]).expect("MstKey holds valid utf8") } Self::Heap(s) => s, } }}
impl Deref for MstKey { type Target = str;
#[inline] fn deref(&self) -> &str { self.as_str() }}
impl AsRef<str> for MstKey { #[inline] fn as_ref(&self) -> &str { self.as_str() }}
impl AsRef<[u8]> for MstKey { #[inline] fn as_ref(&self) -> &[u8] { self.as_bytes() }}
impl Borrow<str> for MstKey { #[inline] fn borrow(&self) -> &str { self.as_str() }}
impl PartialEq for MstKey { #[inline] fn eq(&self, other: &Self) -> bool { self.as_bytes() == other.as_bytes() }}
impl Eq for MstKey {}
impl PartialOrd for MstKey { #[inline] fn partial_cmp(&self, other: &Self) -> Option<Ordering> { Some(self.cmp(other)) }}
impl Ord for MstKey { #[inline] fn cmp(&self, other: &Self) -> Ordering { self.as_bytes().cmp(other.as_bytes()) }}
impl Hash for MstKey { #[inline] fn hash<H: Hasher>(&self, state: &mut H) { self.as_str().hash(state); }}
impl fmt::Debug for MstKey { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Debug::fmt(self.as_str(), f) }}
impl fmt::Display for MstKey { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Display::fmt(self.as_str(), f) }}
impl PartialEq<str> for MstKey { #[inline] fn eq(&self, other: &str) -> bool { self.as_bytes() == other.as_bytes() }}
impl PartialEq<&str> for MstKey { #[inline] fn eq(&self, other: &&str) -> bool { self.as_bytes() == other.as_bytes() }}
impl PartialEq<SmolStr> for MstKey { #[inline] fn eq(&self, other: &SmolStr) -> bool { self.as_bytes() == other.as_bytes() }}
#[cfg(test)]impl From<SmolStr> for MstKey { #[inline] fn from(s: SmolStr) -> Self { Self::new(s.as_str()) }}
/// where [`walk`] reads nodes from. a read hands back owned bytes and can fail, so a car/// that isn't held in memory can serve the walk too.pub(crate) trait BlockSource { fn read(&self, cid: &IpldCid) -> Result<Option<Bytes>>;4 unmodified lines Ok(self.get(cid).cloned()) }}
#[derive(Deserialize)]struct RawTreeEntry<'a> { #[serde(rename = "p")] prefix_len: u8, #[serde(rename = "k", with = "serde_bytes", borrow)] key_suffix: &'a [u8], #[serde(rename = "v")] value: IpldCid, #[serde(rename = "t")] tree: Option<IpldCid>,}
#[derive(Debug, Clone)]pub(crate) enum FlatEntry { Tree { cid: IpldCid }, Leaf { key: SmolStr, cid: IpldCid }, Leaf { key: MstKey, cid: IpldCid },}
/// a node's subtrees and leaves in key order, with full keys rebuilt from prefix compression.pub(crate) fn decode_node(bytes: &[u8]) -> Result<Vec<FlatEntry>> { let node: NodeData = serde_ipld_dagcbor::from_slice(bytes).into_diagnostic()?; decode_node_entries(&node) let mut entries = Vec::new(); decode_node_into(bytes, &mut entries)?; Ok(entries)}
fn decode_node_entries(node: &NodeData) -> Result<Vec<FlatEntry>> { let mut entries = Vec::with_capacity(node.entries.len() * 2 + 1); if let Some(cid) = node.left { entries.push(FlatEntry::Tree { cid });/// appends a decoded node's entries onto `entries`, reusing its allocation when empty.pub(crate) fn decode_node_into(bytes: &[u8], entries: &mut Vec<FlatEntry>) -> Result<()> { let mut de = serde_ipld_dagcbor::de::Deserializer::from_slice(bytes); NodeSeed(entries).deserialize(&mut de).into_diagnostic()?; de.end().into_diagnostic()?; Ok(())}
struct NodeSeed<'a>(&'a mut Vec<FlatEntry>);
impl<'de> DeserializeSeed<'de> for NodeSeed<'_> { type Value = ();
fn deserialize<D>(self, deserializer: D) -> std::result::Result<Self::Value, D::Error> where D: serde::Deserializer<'de>, { deserializer.deserialize_map(self) }}
// each key is stored as a prefix length into the previous key plus a suffix let mut key = String::new(); 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); if !key.is_char_boundary(prefix_len) { return Err(miette::miette!( "invalid mst key prefix length {prefix_len}" ));impl<'de> Visitor<'de> for NodeSeed<'_> { type Value = ();
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("an mst node map") }
fn visit_map<A>(self, mut map: A) -> std::result::Result<Self::Value, A::Error> where A: MapAccess<'de>, { let base = self.0.len(); let mut left: Option<IpldCid> = None; let mut seen_entries = false;
while let Some(field) = map.next_key::<&str>()? { match field { "l" => { left = map.next_value()?; } "e" => { if seen_entries { return Err(serde::de::Error::duplicate_field("e")); } seen_entries = true; map.next_value_seed(EntriesSeed(&mut *self.0))?; } _ => { map.next_value::<IgnoredAny>()?; } } } 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}");
if !seen_entries { return Err(serde::de::Error::missing_field("e")); } key.truncate(prefix_len); key.push_str(suffix); if let Some(cid) = left { self.0.insert(base, FlatEntry::Tree { cid }); } Ok(()) }}
entries.push(FlatEntry::Leaf { key: SmolStr::new(&key), cid: entry.value, }); if let Some(cid) = entry.tree { entries.push(FlatEntry::Tree { cid });struct EntriesSeed<'a>(&'a mut Vec<FlatEntry>);
impl<'de> DeserializeSeed<'de> for EntriesSeed<'_> { type Value = ();
fn deserialize<D>(self, deserializer: D) -> std::result::Result<Self::Value, D::Error> where D: serde::Deserializer<'de>, { deserializer.deserialize_seq(self) }}
impl<'de> Visitor<'de> for EntriesSeed<'_> { type Value = ();
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("an mst entry list") }
fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error> where A: SeqAccess<'de>, { let count = seq.size_hint().unwrap_or(0); if count > 0 { self.0.reserve(count * 2 + 1); }
// each key is stored as a prefix length into the previous key plus a suffix let mut key = KeyBuilder::new(); let mut idx = 0usize; while let Some(entry) = seq.next_element::<RawTreeEntry<'de>>()? { let leaf_key = key .advance(idx > 0, usize::from(entry.prefix_len), entry.key_suffix) .map_err(serde::de::Error::custom)?; self.0.push(FlatEntry::Leaf { key: leaf_key, cid: entry.value, }); if let Some(cid) = entry.tree { self.0.push(FlatEntry::Tree { cid }); } idx += 1; } Ok(()) }}
Ok(entries)enum KeyBuilder { Inline { len: usize, buf: [u8; INLINE_KEY_CAP], }, Heap(String),}
impl KeyBuilder { #[inline] fn new() -> Self { Self::Inline { len: 0, buf: [0u8; INLINE_KEY_CAP], } }
#[inline] fn advance( &mut self, check_order: bool, prefix_len: usize, suffix_bytes: &[u8], ) -> std::result::Result<MstKey, String> { let suffix = std::str::from_utf8(suffix_bytes) .map_err(|err| format!("invalid utf8 in mst key suffix: {err}"))?; let prev_bytes = match self { Self::Inline { len, buf } => &buf[..*len], Self::Heap(s) => s.as_bytes(), }; if !is_utf8_char_boundary(prev_bytes, prefix_len) { return Err(format!("invalid mst key prefix length {prefix_len}")); } if check_order && suffix.as_bytes() <= &prev_bytes[prefix_len..] { let prev = std::str::from_utf8(prev_bytes).expect("KeyBuilder holds valid utf8"); let cur = format!("{}{suffix}", &prev[..prefix_len]); return Err(format!("mst key {cur} does not sort after {prev}")); }
let new_len = prefix_len + suffix.len(); match self { Self::Inline { len, buf } if new_len <= INLINE_KEY_CAP => { buf[prefix_len..new_len].copy_from_slice(suffix.as_bytes()); *len = new_len; Ok(MstKey::Inline { len: new_len as u8, buf: *buf, }) } Self::Inline { buf, .. } => { let mut s = String::with_capacity(new_len.max(128)); s.push_str( std::str::from_utf8(&buf[..prefix_len]).expect("KeyBuilder holds valid utf8"), ); s.push_str(suffix); let key = MstKey::Heap(Arc::from(s.as_str())); *self = Self::Heap(s); Ok(key) } Self::Heap(s) => { s.truncate(prefix_len); s.push_str(suffix); Ok(MstKey::new(s.as_str())) } } }}
#[inline]fn is_utf8_char_boundary(bytes: &[u8], index: usize) -> bool { if index == bytes.len() { return true; } match bytes.get(index) { None => false, Some(&b) => (b as i8) >= -0x40, }}
/// every leaf of the mst rooted at `root`, in key order.////// the tree comes from an untrusted car, so anything a valid mst cannot contain is an error:/// a missing node, a node reached twice (which also ends cid cycles), or keys that are not/// strictly increasing. the walk keeps its own stack, so tree depth cannot overflow the/// thread's.// backfill walks with `walk`, so only tests and the benches collect every leaf at once#[cfg_attr(not(test), allow(dead_code))]pub(crate) fn leaves(blocks: &impl BlockSource, root: IpldCid) -> Result<Vec<(SmolStr, IpldCid)>> {pub(crate) fn leaves(blocks: &impl BlockSource, root: IpldCid) -> Result<Vec<(MstKey, IpldCid)>> { walk(blocks, root).collect()}
/// [`leaves`] one at a time, holding only the path being walked rather than every leaf. it10 unmodified linespub(crate) struct Walk<'a, B: ?Sized> { blocks: &'a B, visited: HashSet<IpldCid, ahash::RandomState>, stack: Vec<FlatEntry>, last: Option<SmolStr>, last: Option<MstKey>,}
impl<B: BlockSource + ?Sized> Walk<'_, B> { fn step(&mut self) -> Result<Option<(SmolStr, IpldCid)>> { fn step(&mut self) -> Result<Option<(MstKey, IpldCid)>> { while let Some(entry) = self.stack.pop() { match entry { FlatEntry::Leaf { key, cid } => { if let Some(previous) = self.last.as_ref().filter(|previous| **previous >= key)10 unmodified lines let bytes = self .blocks .read(&cid)? .ok_or_else(|| miette::miette!("mst node {cid} missing from car"))?; let entries = decode_node(&bytes) let start = self.stack.len(); decode_node_into(&bytes, &mut self.stack) .wrap_err_with(|| format!("failed to decode mst node {cid}"))?; self.stack.extend(entries.into_iter().rev()); self.stack[start..].reverse(); } } } Ok(None) }}
impl<B: BlockSource + ?Sized> Iterator for Walk<'_, B> { type Item = Result<(SmolStr, IpldCid)>; type Item = Result<(MstKey, IpldCid)>;
fn next(&mut self) -> Option<Self::Item> { let step = self.step(); if step.is_err() {8 unmodified lines use super::*; use crate::car::CarBlocks; use cid::multihash::Multihash; use jacquard_common::types::crypto::{DAG_CBOR, SHA2_256}; use jacquard_repo::mst::TreeEntry; use jacquard_repo::mst::{NodeData, TreeEntry}; use jacquard_repo::mst::util::compute_cid;
pub(crate) fn cid(byte: u8) -> IpldCid { let mh = Multihash::<64>::wrap(SHA2_256, &[byte; 32]).unwrap();37 unmodified lines .map(|(cid, node)| (cid, serde_ipld_dagcbor::to_vec(&node).unwrap().into())) .collect() }
fn keys(leaves: &[(SmolStr, IpldCid)]) -> Vec<&str> { fn keys(leaves: &[(MstKey, IpldCid)]) -> Vec<&str> { leaves.iter().map(|(key, _)| key.as_str()).collect() }
#[tokio::test]13 unmodified lines let (root, nodes) = mst.collect_blocks().await.unwrap(); assert!(nodes.len() > 1, "test tree should span several nodes");
let ours = leaves(&nodes.into_iter().collect::<CarBlocks>(), root).unwrap(); assert_eq!(ours, mst.leaves().await.unwrap()); let theirs: Vec<(MstKey, IpldCid)> = mst .leaves() .await .unwrap() .into_iter() .map(|(k, c)| (MstKey::from(k), c)) .collect(); assert_eq!(ours, theirs); }
#[test] fn walks_an_empty_tree() {30 unmodified lines
let mut walked = walk(&tree, root); assert_eq!( walked.next().unwrap().unwrap(), (SmolStr::new("a/1"), cid(9)) (MstKey::new("a/1"), cid(9)) ); let err = walked.next().unwrap().unwrap_err(); assert!(err.to_string().contains("missing from car"), "{err}"); // b/1 is still on the stack, but nothing after an error counts75 unmodified lines }
#[test] fn rebuilds_prefix_compressed_keys() { let long_prefix = "io.atcr.hold.extremely.long.collection.name.exceeding.inline/"; let long_a = format!("{long_prefix}3k2b1f5moe22a"); let long_b = format!("{long_prefix}3k2b1f5moe22b"); assert!(long_a.len() > INLINE_KEY_CAP); let root = cid(1); let tree = blocks(vec![( root, node( vec![ ("app.bsky.feed.like/1", cid(9), None), ("app.bsky.feed.post/1", cid(9), None), ("app.bsky.feed.post/2", cid(9), None), (long_a.as_str(), cid(9), None), (long_b.as_str(), cid(9), None), ("io.atcr.hold.short/1", cid(9), None), // both start with byte 0xc3, so they share no whole-char prefix ("è/1", cid(9), None), ("é/1", cid(9), None), ],6 unmodified lines [ "app.bsky.feed.like/1", "app.bsky.feed.post/1", "app.bsky.feed.post/2", long_a.as_str(), long_b.as_str(), "io.atcr.hold.short/1", "è/1", "é/1" ] );26 unmodified lines294 unmodified linesExpand all295296297298299299300301302303304373 unmodified linesExpand all678679680681682683684685686687688689157 unmodified linesExpand all84784884985085185285385485546 unmodified lines294 unmodified linesExpand allcodec-call:src/ingest/stream/codec.rs::encode_error_frame::serde_ipld_dagcbor::to_vec rust:src/ingest/stream/codec.rs::firehose_frame_fixture_decodes_identitycodec-call:src/ingest/stream/codec.rs::encode_frame::serde_ipld_dagcbor::to_vec rust:src/ingest/stream/codec.rs::firehose_frame_fixture_decodes_identitycodec-call:src/ingest/validation.rs::validate_commit::serde_ipld_dagcbor::from_slice rust:src/ingest/validation.rs::record_dag_cbor_projection_accepts_data_and_rejects_invalid_bytescodec-call:src/jetstream.rs::JetstreamEphemeral::to_json::serde_json::to_vec rust:src/control/stream/interleaving.rs::live_json_is_what_a_replay_renderscodec-call:src/mst.rs::decode_node::serde_ipld_dagcbor::from_slice rust:src/mst.rs::leaves_match_jacquard_walkcodec-call:src/mst.rs::decode_node_into::.deserialize rust:src/mst.rs::leaves_match_jacquard_walkcodec-call:src/mst.rs::decode_node_into::serde_ipld_dagcbor::de::Deserializer::from_slice rust:src/mst.rs::leaves_match_jacquard_walkcodec-call:src/ops.rs::persist_to_resync_buffer::rmp_serde::to_vec_named rust:src/ops.rs::ephemeral_identity_and_account_json_contractscodec-call:src/ops.rs::stage_gone::rmp_serde::to_vec rust:src/ingest/indexer/shard.rs::backfill_finished_parks_any_inactive_status_the_relay_writescodec-call:src/ops.rs::transition_repo::rmp_serde::to_vec rust:src/ops.rs::ephemeral_identity_and_account_json_contractscodec-call:src/ops/backlink_ops.rs::index_record::serde_ipld_dagcbor::from_slice nu:tests/backlinks.nu373 unmodified linesExpand allserde-attribute:src/ingest/stream/types.rs::RepoOp::serde(rename_all="camelCase") rust:src/ingest/stream.rs::test_decode_commit_empty_sinceserde-attribute:src/ingest/stream/types.rs::Sync::blocks::serde(with="jacquard_common::serde_bytes_helper") rust:src/ingest/stream/codec.rs::firehose_frame_mock_decodes_sync_binary_blocksserde-attribute:src/ingest/stream/types.rs::Sync::rev::serde(borrow) rust:src/ingest/stream/codec.rs::firehose_frame_mock_decodes_sync_binary_blocksserde-attribute:src/ingest/stream/types.rs::Sync::serde(rename_all="camelCase") rust:src/ingest/stream/codec.rs::firehose_frame_mock_decodes_sync_binary_blocksserde-attribute:src/mst.rs::RawTreeEntry::key_suffix::serde(rename="k",with="serde_bytes",borrow) rust:src/mst.rs::leaves_match_jacquard_walkserde-attribute:src/mst.rs::RawTreeEntry::prefix_len::serde(rename="p") rust:src/mst.rs::leaves_match_jacquard_walkserde-attribute:src/mst.rs::RawTreeEntry::tree::serde(rename="t") rust:src/mst.rs::leaves_match_jacquard_walkserde-attribute:src/mst.rs::RawTreeEntry::value::serde(rename="v") rust:src/mst.rs::leaves_match_jacquard_walkserde-attribute:src/patch.rs::SetUpdate::serde(untagged) rust:src/patch.rs::set_update_json_markers_are_structurally_disjointserde-attribute:src/types.rs::BulkJobV1::Apply::pending_key::serde(with="serde_bytes") rust:src/types.rs::bulk_job_golden_fixturesserde-attribute:src/types.rs::DeleteBodyTarget::serde(rename_all="snake_case") rust:src/types.rs::operator_redaction_json_contractsserde-attribute:src/types/event/jetstream.rs::StoredJetstreamEvent::Account::did::serde(borrow) rust:src/types/event/jetstream.rs::common_jetstream_event_golden_fixtures157 unmodified linesExpand allserde-type:src/ingest/stream/types.rs::Identity rust:src/ingest/stream/codec.rs::firehose_frame_fixture_decodes_identityserde-type:src/ingest/stream/types.rs::Info rust:src/ingest/stream/codec.rs::firehose_frame_preserves_outdated_cursor_infoserde-type:src/ingest/stream/types.rs::RepoOp rust:src/ingest/stream.rs::test_decode_commit_empty_sinceserde-type:src/ingest/stream/types.rs::Sync rust:src/ingest/stream/codec.rs::firehose_frame_mock_decodes_sync_binary_blocksserde-type:src/mst.rs::RawTreeEntry rust:src/mst.rs::leaves_match_jacquard_walkserde-type:src/patch.rs::SetUpdate rust:src/patch.rs::set_update_json_markers_are_structurally_disjointserde-type:src/pds_meta.rs::HostStatus rust:src/pds_meta.rs::host_status_persisted_golden_fixturesserde-type:src/resolver.rs::ResolveHandleOutput rust:src/resolver.rs::the_xrpc_source_answers_with_the_did_it_encodesserde-type:src/types.rs::BulkJob rust:src/types.rs::bulk_job_golden_fixtures46 unmodified lines4 unmodified lines5678991011121314151616171818192021224 unmodified lines272829303131323334358 unmodified lines444546474848495050515253545556565758596032 unmodified lines93949596979798991001015 unmodified lines10710810911011111111211311411511 unmodified lines12712812913013113113213313413539 unmodified lines175176177178179179180181182183103 unmodified linesExpand all2872882892902912912922932942956 unmodified lines4 unmodified linesuse std::iter::Fuse;
use cid::Cid as IpldCid;use miette::Result;use smol_str::SmolStr;use crate::mst::MstKey;use tracing::trace;
/// one write that brings the stored records in line with the new repo#[derive(Debug)]pub(crate) enum RecordOp<K, T> { /// a record that `stored` does not have Create { key: SmolStr, record: T }, Create { key: MstKey, record: T }, /// a record whose cid differs from the stored one Update { key: SmolStr, record: T }, Update { key: MstKey, record: T }, /// a stored record the new repo doesn't have Delete(K),}
4 unmodified lines/// `stored` yields each stored record's key after the did prefix and its cid. a heads key/// ends in the same `collection/rkey` text as the mst key, so both sides sort the same way.pub(crate) fn diff<N, S, K, T>(new: N, stored: S) -> Diff<N, S, K, T>where N: Iterator<Item = Result<(SmolStr, IpldCid, T)>>, N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ Diff {8 unmodified lines
pub(crate) struct Diff<N: Iterator, S: Iterator, K, T> { new: Fuse<N>, stored: Fuse<S>, next_new: Option<(SmolStr, IpldCid, T)>, next_new: Option<(MstKey, IpldCid, T)>, next_stored: Option<(K, IpldCid)>, last_new: Option<SmolStr>, last_new: Option<MstKey>, failed: bool,}
impl<N, S, K, T> Diff<N, S, K, T>where N: Iterator<Item = Result<(SmolStr, IpldCid, T)>>, N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ fn step(&mut self) -> Result<Option<RecordOp<K, T>>> {32 unmodified lines }
/// the next new record. a merge against records out of order would delete and recreate /// them, so that's an error instead fn pull_new(&mut self) -> Result<Option<(SmolStr, IpldCid, T)>> { fn pull_new(&mut self) -> Result<Option<(MstKey, IpldCid, T)>> { let Some(next) = self.new.next().transpose()? else { return Ok(None); }; if let Some(last) = self.last_new.as_ref().filter(|last| **last >= next.0) {5 unmodified lines}
impl<N, S, K, T> Iterator for Diff<N, S, K, T>where N: Iterator<Item = Result<(SmolStr, IpldCid, T)>>, N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ type Item = Result<RecordOp<K, T>>;11 unmodified lines#[cfg(test)]mod tests { use std::collections::HashMap;
use smol_str::ToSmolStr; use smol_str::{SmolStr, ToSmolStr};
use super::*; use crate::db::keys; use crate::db::types::{DbAction, DbRkey, DbTid};39 unmodified lines stored: &[(String, IpldCid)], ) -> Vec<(String, &'static str)> { let new = new .iter() .map(|(key, cid)| Ok((SmolStr::new(key), *cid, ()))); .map(|(key, cid)| Ok((MstKey::new(key), *cid, ()))); let stored = stored.iter().map(|(key, cid)| Ok((key.clone(), *cid))); diff(new, stored) .map(|op| match op.unwrap() { RecordOp::Create { key, .. } => (key.to_string(), "create"),103 unmodified linesExpand all
#[test] fn new_records_out_of_order_are_an_error() { let new = [("a.b.c/2", cid(1)), ("a.b.c/1", cid(1))] .map(|(key, cid)| Ok((SmolStr::new(key), cid, ()))); .map(|(key, cid)| Ok((MstKey::new(key), cid, ()))); let mut ops = diff( new.into_iter(), std::iter::empty::<Result<(Vec<u8>, IpldCid)>>(), );6 unmodified lines300 unmodified linesExpand all301302303304305305306307308309310311312313314315316317347 unmodified linesExpand all300 unmodified linesExpand all fn scan( mut repo: FilteredRepo, mst_root: IpldCid, ) -> (Vec<(SmolStr, IpldCid)>, Blocks<CommitBlocks>) { let leaves = repo.scanner.scan(mst_root).unwrap().unwrap().leaves.to_vec(); 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>)> {347 unmodified linesExpand all12 unmodified lines1314151617171819202130 unmodified lines5253545556565758596090 unmodified lines15115215315415515515615715815947 unmodified lines20720820921021121121221321421421521621721838 unmodified lines25725825926026126126226326426536 unmodified lines12 unmodified lines
use crate::backfill::spill::SpilledCar;use crate::car::{Blocks, CarBlock, CarBlocks, CommitBlocks};use crate::filter::{FilterConfig, FilterMode};use crate::mst::BlockSource;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 are30 unmodified lines }}
impl RepoRecords { fn leaves(&self) -> Box<dyn Iterator<Item = Result<(SmolStr, IpldCid)>> + '_> { 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)) }90 unmodified lines
impl CheckedRecords { pub(super) fn new_records( &self, ) -> impl Iterator<Item = Result<(SmolStr, IpldCid, NewRecord)>> + '_ { ) -> impl Iterator<Item = Result<(MstKey, IpldCid, NewRecord)>> + '_ { self.records .leaves() .filter_map(|leaf| new_record(leaf, &self.records, &self.filter).transpose()) }47 unmodified lines
/// `leaf` as the diff takes it, or none if the car left its block out or the filter doesn't/// want its collectionfn new_record( leaf: Result<(SmolStr, IpldCid)>, leaf: Result<(MstKey, IpldCid)>, records: &RepoRecords, filter: &FilterConfig,) -> Result<Option<(SmolStr, IpldCid, NewRecord)>> {) -> Result<Option<(MstKey, IpldCid, NewRecord)>> { let (key, cid) = leaf?; let Some(block) = records.block(&cid)? else { return Ok(None); };38 unmodified lines RepoRecords::Filtered(FilteredRecords { leaves: SortedLeaves::from_vec( leaves .iter() .map(|(key, body)| (SmolStr::new(key), cid(body))) .map(|(key, body)| (MstKey::new(key), cid(body))) .collect(), ), blocks: Blocks::unverified( leaves36 unmodified lines