Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
52 kB · 1581 lines
Rust
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582use std::collections::{BTreeSet, HashMap};use std::marker::PhantomData;use std::num::NonZeroU32;use std::ops::{Bound, ControlFlow};use std::sync::atomic::{AtomicU32, Ordering};use std::sync::{Arc, Mutex};
const FILTER_SCAN_MULTIPLIER: usize = 64;const FILTER_SCAN_FLOOR: usize = 512;
struct ScanState { matched: Vec<(BucketKey, AtUri<DefaultStr>)>, last_scanned: Option<BucketKey>, scanned: usize,}
impl ScanState { fn with_capacity(cap: usize) -> Self { Self { matched: Vec::with_capacity(cap), last_scanned: None, scanned: 0, } }}
use bobbin_runtime::RuntimeHasher;use bobbin_types::edges::Edge;use bobbin_types::ids::EdgeKey;use either::Either;use jacquard_common::DefaultStr;use jacquard_common::types::string::AtUri;use lasso::{Key, Spur, ThreadedRodeo};use scc::HashMap as SccMap;use scc::hash_map::Entry;use smallvec::SmallVec;use thiserror::Error;
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]struct SortMicros(u64);
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]struct BucketKey { micros: SortMicros, source: SourceId,}
impl BucketKey { fn new(micros: u64, source: SourceId) -> Self { Self { micros: SortMicros(micros), source, } }
fn token(self) -> PageToken { PageToken::new(self.micros.0, self.source.index()) }
fn from_token(tok: PageToken) -> Self { Self { micros: SortMicros(tok.micros), source: SourceId::from_raw(tok.source), } }}
pub mod coverage;pub mod state_index;pub use coverage::{Coverage, CoverageWatch, HydrantCursor, PromotionSignal};pub use state_index::{ ApplyOutcome, IssueStateKind, PullStatusKind, StateIndex, StateKind, apply_record_state,};
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct SourceTag;#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]struct AuthorTag;#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]struct CollectionTag;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct Interned<T>(u32, PhantomData<T>);
impl<T: Copy> Interned<T> { fn from_spur(spur: Spur) -> Self { Self(spur.into_usize() as u32, PhantomData) }
fn from_raw(raw: u32) -> Self { Self(raw, PhantomData) }
fn index(self) -> u32 { self.0 }
fn to_spur(self) -> Option<Spur> { Spur::try_from_usize(self.0 as usize) }}
pub type SourceId = Interned<SourceTag>;type AuthorId = Interned<AuthorTag>;type CollectionId = Interned<CollectionTag>;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct PageToken { micros: u64, source: u32,}
impl PageToken { pub fn new(micros: u64, source: u32) -> Self { Self { micros, source } }
pub fn micros(self) -> u64 { self.micros }
pub fn source(self) -> u32 { self.source }
pub fn encode_token(self) -> String { let mut bytes = [0u8; 12]; bytes[..8].copy_from_slice(&self.micros.to_be_bytes()); bytes[8..].copy_from_slice(&self.source.to_be_bytes()); encode_hex(&bytes) }
pub fn decode_token(token: &str) -> Result<Self, CursorParseError> { let bytes: [u8; 12] = decode_hex_array(token).ok_or(CursorParseError::Malformed)?; let micros = u64::from_be_bytes(bytes[..8].try_into().unwrap()); let source = u32::from_be_bytes(bytes[8..].try_into().unwrap()); Ok(Self { micros, source }) }}
fn encode_hex(bytes: &[u8]) -> String { bytes .iter() .fold(String::with_capacity(bytes.len() * 2), |mut acc, b| { acc.push(char::from_digit((b >> 4) as u32, 16).unwrap()); acc.push(char::from_digit((b & 0x0f) as u32, 16).unwrap()); acc })}
fn decode_hex_array<const N: usize>(token: &str) -> Option<[u8; N]> { if token.len() != N * 2 { return None; } let parsed: Vec<u8> = token .as_bytes() .chunks_exact(2) .map(|pair| { let hi = (pair[0] as char).to_digit(16)?; let lo = (pair[1] as char).to_digit(16)?; Some(((hi << 4) | lo) as u8) }) .collect::<Option<Vec<u8>>>()?; parsed.try_into().ok()}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub enum PageCursor { Start, After(PageToken),}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]pub enum SortDir { Asc, #[default] Desc,}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]pub enum CursorParseError { #[error("cursor token must be a valid TID")] Malformed,}
impl PageCursor { pub fn from_token(raw: Option<&str>) -> Result<Self, CursorParseError> { raw.map_or(Ok(Self::Start), |t| { PageToken::decode_token(t).map(Self::After) }) }}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct PageLimit(u32);
#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]pub enum PageLimitError { #[error("page limit {value} below minimum {min}")] TooSmall { value: u32, min: u32 }, #[error("page limit {value} above maximum {max}")] TooLarge { value: u32, max: u32 },}
impl PageLimit { pub const MIN: u32 = 1; pub const MAX: u32 = 1000;
pub fn new(value: u32) -> Result<Self, PageLimitError> { match value { v if v < Self::MIN => Err(PageLimitError::TooSmall { value: v, min: Self::MIN, }), v if v > Self::MAX => Err(PageLimitError::TooLarge { value: v, max: Self::MAX, }), v => Ok(Self(v)), } }
pub const fn get(self) -> u32 { self.0 }}
#[derive(Debug)]pub struct EdgePage { pub items: Vec<AtUri<DefaultStr>>, pub next: Option<PageToken>,}
#[derive(Clone, Copy, Debug, Default)]pub struct EdgeMemReport { pub key_count: u64, pub source_count: u64, pub source_interner_bytes: u64, pub did_interner_bytes: u64, pub collection_interner_bytes: u64, pub key_interner_bytes: u64, pub edges_total: u64, pub author_refs_total: u64, pub reverse_entries: u64, pub reverse_cap: u64, pub forward_struct_bytes: u64, pub reverse_struct_bytes: u64, pub bucket_struct_bytes: u64, pub max_bucket: u64, pub bucket_size_classes: [u64; BUCKET_CLASS_COUNT],}
const BUCKET_CLASS_BOUNDS: [u64; 16] = [ 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 8192, 32768, 131072, u64::MAX,];const BUCKET_CLASS_COUNT: usize = BUCKET_CLASS_BOUNDS.len();
fn bucket_class(n: u64) -> usize { BUCKET_CLASS_BOUNDS .iter() .position(|&bound| n <= bound) .unwrap_or(BUCKET_CLASS_COUNT - 1)}
impl EdgeMemReport { pub fn bucket_histogram(&self) -> impl Iterator<Item = (u64, u64)> { BUCKET_CLASS_BOUNDS .into_iter() .zip(self.bucket_size_classes) }}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]struct EdgeKeyId(u32);
fn bump_author(authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>, author: AuthorId) { authors .entry(author) .and_modify(|c| *c = c.saturating_add(1)) .or_insert(NonZeroU32::MIN);}
fn drop_author(authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>, author: AuthorId) { match authors.get(&author).map(|c| c.get() - 1) { Some(0) | None => { authors.remove(&author); } Some(next) => { authors.insert(author, NonZeroU32::new(next).unwrap()); } }}
struct LargeBucket { keys: BTreeSet<BucketKey>, authors: HashMap<AuthorId, NonZeroU32, RuntimeHasher>,}
const BUCKET_PROMOTE_AT: usize = 256;const SOURCE_BYTES: u64 = 16;
enum Sources { Small(SmallVec<[BucketKey; 2]>), Large(Box<LargeBucket>),}
impl Default for Sources { fn default() -> Self { Self::Small(SmallVec::new()) }}
impl Sources { fn len(&self) -> usize { match self { Self::Small(v) => v.len(), Self::Large(big) => big.keys.len(), } }
fn is_empty(&self) -> bool { self.len() == 0 }
fn insert(&mut self, key: BucketKey, author: Option<AuthorId>) -> bool { match self { Self::Small(v) => match v.binary_search(&key) { Ok(_) => false, Err(pos) => { v.insert(pos, key); true } }, Self::Large(big) => { let inserted = big.keys.insert(key); if inserted && let Some(a) = author { bump_author(&mut big.authors, a); } inserted } } }
fn remove(&mut self, key: &BucketKey, author: Option<AuthorId>) { match self { Self::Small(v) => { if let Ok(pos) = v.binary_search(key) { v.remove(pos); } } Self::Large(big) => { if big.keys.remove(key) && let Some(a) = author { drop_author(&mut big.authors, a); } } } }
fn directed( &self, cursor: PageCursor, dir: SortDir, ) -> Box<dyn Iterator<Item = BucketKey> + '_> { match self { Self::Small(v) => Box::new(directed_slice(v, cursor, dir)), Self::Large(big) => Box::new(directed_tree(&big.keys, cursor, dir)), } }
fn heap_bytes(&self) -> u64 { const BTREE_BYTES_PER_KEY: u64 = 32; const HASHMAP_FIXED: u64 = 48; const HASHMAP_PER_CAP: u64 = 9; match self { Self::Small(v) if v.spilled() => v.capacity() as u64 * SOURCE_BYTES, Self::Small(_) => 0, Self::Large(big) => { std::mem::size_of::<LargeBucket>() as u64 + big.keys.len() as u64 * BTREE_BYTES_PER_KEY + HASHMAP_FIXED + big.authors.capacity() as u64 * HASHMAP_PER_CAP } } }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]struct ReverseEntry { key_id: EdgeKeyId, sort_micros: u64,}
pub struct EdgeStore { source_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, did_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, collection_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, key_ids: SccMap<EdgeKey, EdgeKeyId, RuntimeHasher>, next_key_id: AtomicU32, forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>, reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>, hasher: RuntimeHasher, writer: Mutex<()>,}
impl EdgeStore { pub fn new(hasher: RuntimeHasher) -> Self { Self { source_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), collection_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), key_ids: SccMap::with_hasher(hasher.clone()), next_key_id: AtomicU32::new(0), forward: SccMap::with_hasher(hasher.clone()), reverse: SccMap::with_hasher(hasher.clone()), hasher, writer: Mutex::new(()), } }
fn intern_key(&self, key: EdgeKey) -> EdgeKeyId { match self.key_ids.entry_sync(key) { Entry::Occupied(e) => *e.get(), Entry::Vacant(e) => { let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed)); e.insert_entry(id); id } } }
fn lookup_key(&self, key: &EdgeKey) -> Option<EdgeKeyId> { self.key_ids.read_sync(key, |_, id| *id) }
pub fn add(&self, edge: Edge) { let _w = self .writer .lock() .expect("edge-store writer mutex poisoned"); self.add_locked(edge); }
pub fn intern_source(&self, source: &AtUri<DefaultStr>) -> SourceId { let author = self.intern_author(source); SourceId::from_spur( self.source_interner .get_or_intern(self.source_key(source, author)), ) }
pub fn upsert_source(&self, source: &AtUri<DefaultStr>, edges: Vec<Edge>) { let _w = self .writer .lock() .expect("edge-store writer mutex poisoned"); self.clear_source_locked(source); edges.into_iter().for_each(|e| self.add_locked(e)); }
pub fn remove_source(&self, source: &AtUri<DefaultStr>) { let _w = self .writer .lock() .expect("edge-store writer mutex poisoned"); self.clear_source_locked(source); }
fn add_locked(&self, edge: Edge) { let author = self.intern_author(&edge.source); let source_key = self.source_key(&edge.source, author); let id = SourceId::from_spur(self.source_interner.get_or_intern(source_key)); let sort_micros = edge.sort_micros; let key_id = self.intern_key(EdgeKey::new(edge.kind, edge.subject));
let key = BucketKey::new(sort_micros, id); let mut entry = self.forward.entry_sync(key_id).or_default(); let inserted = entry.get_mut().insert(key, author); let promote_keys = match entry.get() { Sources::Small(v) if v.len() > BUCKET_PROMOTE_AT => { Some(v.iter().copied().collect::<Vec<_>>()) } _ => None, }; drop(entry);
if let Some(keys) = promote_keys { let authors = self.build_author_map(&keys); let large = LargeBucket { keys: keys.into_iter().collect(), authors, }; if let Entry::Occupied(mut e) = self.forward.entry_sync(key_id) { *e.get_mut() = Sources::Large(Box::new(large)); } }
if inserted { let mut rev = self.reverse.entry_sync(id).or_default(); rev.get_mut().push(ReverseEntry { key_id, sort_micros, }); } }
fn clear_source_locked(&self, source: &AtUri<DefaultStr>) { let author = source_authority_did(source) .and_then(|s| self.did_interner.get(s)) .map(AuthorId::from_spur); let Some(source_spur) = self.source_interner.get(self.source_key(source, author)) else { return; }; let id = SourceId::from_spur(source_spur); let Some((_, entries)) = self.reverse.remove_sync(&id) else { return; }; entries.into_iter().for_each( |ReverseEntry { key_id, sort_micros, }| { self.forward.update_sync(&key_id, |_, sources| { sources.remove(&BucketKey::new(sort_micros, id), author); }); self.forward .remove_if_sync(&key_id, |sources| sources.is_empty()); }, ); }
fn intern_author(&self, source: &AtUri<DefaultStr>) -> Option<AuthorId> { let did = source_authority_did(source)?; Some(AuthorId::from_spur(self.did_interner.get_or_intern(did))) }
fn source_key(&self, source: &AtUri<DefaultStr>, author: Option<AuthorId>) -> String { match (split_record_uri(source.as_ref()), author) { (Some((_, collection, rkey)), Some(author)) => { let collection = CollectionId::from_spur(self.collection_interner.get_or_intern(collection)); format!("{}/{}/{}", author.index(), collection.index(), rkey) } _ => source.as_ref().to_owned(), } }
fn decode_source(&self, stored: &str) -> Option<String> { if stored.starts_with("at://") { return Some(stored.to_owned()); } let mut parts = stored.splitn(3, '/'); let author = AuthorId::from_raw(parts.next()?.parse().ok()?); let collection = CollectionId::from_raw(parts.next()?.parse().ok()?); let rkey = parts.next()?; let did = self.did_interner.try_resolve(&author.to_spur()?)?; let collection = self .collection_interner .try_resolve(&collection.to_spur()?)?; Some(format!("at://{did}/{collection}/{rkey}")) }
fn author_of_stored(&self, stored: &str) -> Option<AuthorId> { match stored.strip_prefix("at://") { Some(rest) => { let authority = rest.split('/').next().unwrap_or(rest); authority .starts_with("did:") .then(|| self.did_interner.get(authority).map(AuthorId::from_spur)) .flatten() } None => stored .split('/') .next()? .parse::<u32>() .ok() .map(AuthorId::from_raw), } }
fn author_of(&self, source: SourceId) -> Option<AuthorId> { let spur = source.to_spur()?; let stored = self.source_interner.try_resolve(&spur)?; self.author_of_stored(stored) }
fn distinct_authors_small(&self, keys: &[BucketKey]) -> u64 { keys.iter() .filter_map(|key| self.author_of(key.source)) .collect::<std::collections::HashSet<AuthorId>>() .len() as u64 }
fn build_author_map(&self, keys: &[BucketKey]) -> HashMap<AuthorId, NonZeroU32, RuntimeHasher> { keys.iter().fold( HashMap::with_hasher(self.hasher.clone()), |mut authors, key| { if let Some(a) = self.author_of(key.source) { bump_author(&mut authors, a); } authors }, ) }
pub fn count(&self, key: &EdgeKey) -> u64 { self.lookup_key(key) .and_then(|id| { self.forward .read_sync(&id, |_, sources| sources.len() as u64) }) .unwrap_or(0) }
pub fn count_distinct_authors(&self, key: &EdgeKey) -> u64 { self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| match sources { Sources::Large(big) => big.authors.len() as u64, Sources::Small(v) => self.distinct_authors_small(v), }) }) .unwrap_or(0) }
pub fn list( &self, key: &EdgeKey, cursor: PageCursor, limit: PageLimit, dir: SortDir, ) -> EdgePage { let limit_usize = limit.get() as usize; self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| { let iter = sources.directed(cursor, dir); let entries: Vec<BucketKey> = iter.take(limit_usize + 1).collect(); let has_more = entries.len() > limit_usize; let page = &entries[..entries.len().min(limit_usize)]; let items = page .iter() .filter_map(|&key| { let spur = key.source.to_spur()?; let stored = self.source_interner.try_resolve(&spur)?; AtUri::new_owned(self.decode_source(stored)?).ok() }) .collect(); let next = has_more .then(|| page.last().copied()) .flatten() .map(BucketKey::token); EdgePage { items, next } }) }) .unwrap_or(EdgePage { items: Vec::new(), next: None, }) }
pub fn list_filtered<F>( &self, key: &EdgeKey, cursor: PageCursor, limit: PageLimit, dir: SortDir, predicate: F, ) -> EdgePage where F: Fn(&AtUri<DefaultStr>) -> bool, { let limit_usize = limit.get() as usize; let scan_cap = limit_usize .saturating_mul(FILTER_SCAN_MULTIPLIER) .max(FILTER_SCAN_FLOOR); self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| { let init = ScanState::with_capacity(limit_usize + 1); let outcome = sources .directed(cursor, dir) .try_fold(init, |mut state, key| { if state.scanned >= scan_cap && state.matched.len() <= limit_usize { return ControlFlow::Break(state); } state.scanned += 1; state.last_scanned = Some(key); if let Some(spur) = key.source.to_spur() && let Some(stored) = self.source_interner.try_resolve(&spur) && let Some(decoded) = self.decode_source(stored) && let Ok(uri) = AtUri::new_owned(decoded) && predicate(&uri) { state.matched.push((key, uri)); if state.matched.len() > limit_usize { return ControlFlow::Break(state); } } ControlFlow::Continue(state) }); let (state, bucket_exhausted) = match outcome { ControlFlow::Continue(s) => (s, true), ControlFlow::Break(s) => (s, false), }; let has_more_matches = state.matched.len() > limit_usize; let visible_len = state.matched.len().min(limit_usize); let next = if has_more_matches && visible_len > 0 { Some(state.matched[visible_len - 1].0.token()) } else if !bucket_exhausted { state.last_scanned.map(BucketKey::token) } else { None }; let items = state .matched .into_iter() .take(visible_len) .map(|(_, uri)| uri) .collect(); EdgePage { items, next } }) }) .unwrap_or(EdgePage { items: Vec::new(), next: None, }) }
pub fn key_count(&self) -> usize { self.forward.len() }
pub fn source_count(&self) -> usize { self.reverse.len() }
pub fn mem_report(&self) -> EdgeMemReport { const SCC_SLOT: u64 = 32; let edge_key = std::mem::size_of::<EdgeKey>() as u64; let rev_entry = std::mem::size_of::<ReverseEntry>() as u64; let rev_smallvec = std::mem::size_of::<SmallVec<[ReverseEntry; 1]>>() as u64; let bucket_struct = std::mem::size_of::<Sources>() as u64;
let mut edges_total = 0u64; let mut author_refs_total = 0u64; let mut forward_struct_bytes = 0u64; let mut max_bucket = 0u64; let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT]; self.forward.iter_sync(|_, sources| { let bucket_len = sources.len() as u64; edges_total += bucket_len; max_bucket = max_bucket.max(bucket_len); bucket_size_classes[bucket_class(bucket_len)] += 1; if let Sources::Large(big) = sources { author_refs_total += big.authors.len() as u64; } forward_struct_bytes += SCC_SLOT + bucket_struct + sources.heap_bytes(); true }); let key_interner_bytes = self.key_ids.len() as u64 * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT);
let mut reverse_entries = 0u64; let mut reverse_cap = 0u64; let mut reverse_struct_bytes = 0u64; self.reverse.iter_sync(|_, refs| { let cap = refs.capacity() as u64; reverse_entries += refs.len() as u64; reverse_cap += cap; let heap = if refs.spilled() { cap * rev_entry } else { 0 }; reverse_struct_bytes += SCC_SLOT + rev_smallvec + heap; true });
EdgeMemReport { key_count: self.forward.len() as u64, source_count: self.reverse.len() as u64, source_interner_bytes: self.source_interner.current_memory_usage() as u64, did_interner_bytes: self.did_interner.current_memory_usage() as u64, collection_interner_bytes: self.collection_interner.current_memory_usage() as u64, key_interner_bytes, edges_total, author_refs_total, reverse_entries, reverse_cap, forward_struct_bytes, reverse_struct_bytes, bucket_struct_bytes: bucket_struct, max_bucket, bucket_size_classes, } }}
fn directed_slice( sources: &[BucketKey], cursor: PageCursor, dir: SortDir,) -> impl Iterator<Item = BucketKey> + '_ { match dir { SortDir::Asc => { let start = match cursor { PageCursor::Start => 0, PageCursor::After(tok) => { sources.partition_point(|k| *k <= BucketKey::from_token(tok)) } }; Either::Left(sources[start..].iter().copied()) } SortDir::Desc => { let end = match cursor { PageCursor::Start => sources.len(), PageCursor::After(tok) => { sources.partition_point(|k| *k < BucketKey::from_token(tok)) } }; Either::Right(sources[..end].iter().rev().copied()) } }}
fn directed_tree( keys: &BTreeSet<BucketKey>, cursor: PageCursor, dir: SortDir,) -> impl Iterator<Item = BucketKey> + '_ { match dir { SortDir::Asc => { let lower = match cursor { PageCursor::Start => Bound::Unbounded, PageCursor::After(tok) => Bound::Excluded(BucketKey::from_token(tok)), }; Either::Left(keys.range((lower, Bound::Unbounded)).copied()) } SortDir::Desc => { let upper = match cursor { PageCursor::Start => Bound::Unbounded, PageCursor::After(tok) => Bound::Excluded(BucketKey::from_token(tok)), }; Either::Right(keys.range((Bound::Unbounded, upper)).rev().copied()) } }}
fn split_record_uri(source: &str) -> Option<(&str, &str, &str)> { let rest = source.strip_prefix("at://")?; let mut parts = rest.split('/'); let authority = parts.next()?; let collection = parts.next()?; let rkey = parts.next()?; if parts.next().is_some() { return None; } (authority.starts_with("did:") && !collection.is_empty() && !rkey.is_empty()) .then_some((authority, collection, rkey))}
fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<&str> { let rest = source.as_ref().strip_prefix("at://")?; let end = rest.find('/').unwrap_or(rest.len()); let candidate = &rest[..end]; candidate.starts_with("did:").then_some(candidate)}
#[cfg(test)]mod tests { use super::*; use bobbin_types::ids::SubjectRef; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri;
fn store() -> EdgeStore { EdgeStore::new(RuntimeHasher::default()) }
fn nsid(s: &'static str) -> Nsid<DefaultStr> { Nsid::new_static(s).unwrap() }
fn at(s: &str) -> AtUri<DefaultStr> { AtUri::new_owned(s).unwrap() }
fn did(s: &str) -> Did<DefaultStr> { Did::new_owned(s).unwrap() }
fn did_subj(s: &str) -> SubjectRef { SubjectRef::Did(did(s)) }
fn limit(n: u32) -> PageLimit { PageLimit::new(n).unwrap() }
const NAMES: [&str; 5] = ["nel", "olaren", "teq", "lyna", "bailey"];
fn star_edge(source: AtUri<DefaultStr>, subject: Did<DefaultStr>) -> Edge { star_edge_at(source, subject, 0) }
fn star_edge_at(source: AtUri<DefaultStr>, subject: Did<DefaultStr>, sort_micros: u64) -> Edge { Edge { kind: nsid("sh.tangled.feed.star"), subject: SubjectRef::Did(subject), source, sort_micros, } }
fn shuffled_micros(i: usize) -> u64 { (i as u64).wrapping_mul(2_654_435_761) % 64 }
fn source_uri(i: usize) -> String { format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i % NAMES.len()] ) }
fn fill_subject(store: &EdgeStore, n: usize) -> EdgeKey { let subject = did("did:plc:squid"); (0..n).for_each(|i| { store.add(star_edge_at( at(&source_uri(i)), subject.clone(), shuffled_micros(i), )); }); EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:squid")) }
fn reference(kept: impl Iterator<Item = usize>) -> Vec<String> { let mut rows: Vec<(u64, usize)> = kept.map(|i| (shuffled_micros(i), i)).collect(); rows.sort_unstable(); rows.into_iter().map(|(_, i)| source_uri(i)).collect() }
fn paginate_all(store: &EdgeStore, key: &EdgeKey, dir: SortDir, page: u32) -> Vec<String> { std::iter::successors( Some(store.list(key, PageCursor::Start, limit(page), dir)), |prev| { prev.next .map(|tok| store.list(key, PageCursor::After(tok), limit(page), dir)) }, ) .flat_map(|p| { p.items .into_iter() .map(|u| u.as_ref().to_owned()) .collect::<Vec<_>>() }) .collect() }
#[test] fn pagination_matches_reference_across_small_and_large() { [64usize, 1000].into_iter().for_each(|n| { let store = store(); let key = fill_subject(&store, n); assert_eq!(store.count(&key), n as u64, "count mismatch at n={n}"); assert_eq!( store.count_distinct_authors(&key), NAMES.len() as u64, "distinct authors at n={n}" );
let asc = reference(0..n); let desc: Vec<String> = asc.iter().rev().cloned().collect();
[3u32, 7, 50].into_iter().for_each(|page| { assert_eq!( paginate_all(&store, &key, SortDir::Asc, page), asc, "asc mismatch n={n} page={page}" ); assert_eq!( paginate_all(&store, &key, SortDir::Desc, page), desc, "desc mismatch n={n} page={page}" ); }); }); }
#[test] fn remove_from_large_bucket_keeps_pagination_exact() { let store = store(); let n = 1000usize; let key = fill_subject(&store, n); (0..n).step_by(3).for_each(|i| { store.remove_source(&at(&source_uri(i))); }); let expected = reference((0..n).filter(|i| i % 3 != 0)); assert_eq!(store.count(&key), expected.len() as u64); assert_eq!( store.count_distinct_authors(&key), NAMES.len() as u64, "every author keeps sources after partial removal" ); assert_eq!(paginate_all(&store, &key, SortDir::Asc, 7), expected); assert_eq!( paginate_all(&store, &key, SortDir::Desc, 11), expected.iter().rev().cloned().collect::<Vec<_>>() ); }
#[test] fn add_then_count() { let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
store.add(star_edge( at("at://did:plc:nel/sh.tangled.feed.star/r1"), did("did:plc:abalone"), )); store.add(star_edge( at("at://did:plc:olaren/sh.tangled.feed.star/r2"), did("did:plc:abalone"), )); store.add(star_edge( at("at://did:plc:nel/sh.tangled.feed.star/r3"), did("did:plc:abalone"), ));
assert_eq!(store.count(&key), 3); assert_eq!(store.count_distinct_authors(&key), 2); }
#[test] fn duplicate_add_is_idempotent() { let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); let edge = star_edge( at("at://did:plc:nel/sh.tangled.feed.star/r1"), did("did:plc:abalone"), ); store.add(edge.clone()); store.add(edge); assert_eq!(store.count(&key), 1); assert_eq!(store.count_distinct_authors(&key), 1); }
#[test] fn remove_source_clears_all_keys_for_that_source() { let store = store(); let star_key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); let follow_key = EdgeKey::new(nsid("sh.tangled.graph.follow"), did_subj("did:plc:lyna")); let source = "at://did:plc:nel/sh.tangled.feed.star/r1";
store.add(Edge { kind: nsid("sh.tangled.feed.star"), subject: did_subj("did:plc:abalone"), source: at(source), sort_micros: 0, }); store.add(Edge { kind: nsid("sh.tangled.graph.follow"), subject: did_subj("did:plc:lyna"), source: at(source), sort_micros: 0, });
assert_eq!(store.count(&star_key), 1); assert_eq!(store.count(&follow_key), 1);
store.remove_source(&at(source)); assert_eq!(store.count(&star_key), 0); assert_eq!(store.count(&follow_key), 0); }
#[test] fn upsert_source_replaces_old_edges() { let store = store(); let source = at("at://did:plc:teq/sh.tangled.feed.star/r1"); let old_subject = did_subj("did:plc:abalone"); let new_subject = did_subj("did:plc:uni"); let kind = nsid("sh.tangled.feed.star");
store.upsert_source( &source, vec![Edge { kind: kind.clone(), subject: old_subject.clone(), source: source.clone(), sort_micros: 0, }], ); assert_eq!( store.count(&EdgeKey::new(kind.clone(), old_subject.clone())), 1 );
store.upsert_source( &source, vec![Edge { kind: kind.clone(), subject: new_subject.clone(), source: source.clone(), sort_micros: 0, }], ); assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0); assert_eq!(store.count(&EdgeKey::new(kind, new_subject)), 1); }
#[test] fn list_pages_in_sort_order() { let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), did("did:plc:abalone"), 1_000_000 + i as u64 * 1_000_000, )); });
let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc); assert_eq!(page1.items.len(), 2); let cursor = PageCursor::After(page1.next.expect("has next"));
let page2 = store.list(&key, cursor, limit(2), SortDir::Asc); assert_eq!(page2.items.len(), 2);
let cursor2 = PageCursor::After(page2.next.expect("has next")); let page3 = store.list(&key, cursor2, limit(2), SortDir::Asc); assert_eq!(page3.items.len(), 1); assert!( page3.next.is_none(), "final partial page should signal exhaustion" ); }
#[test] fn list_exact_fill_signals_exhaustion() { let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); (0..2).for_each(|i| { store.add(star_edge( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), did("did:plc:abalone"), )); });
let page = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc); assert_eq!(page.items.len(), 2); assert!(page.next.is_none(), "exact-fill page must not promise more"); }
#[test] fn list_on_unknown_key_is_empty() { let store = store(); let page = store.list( &EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:periwinkle")), PageCursor::Start, limit(10), SortDir::Asc, ); assert!(page.items.is_empty()); assert!(page.next.is_none()); }
#[test] fn distinct_authors_decreases_when_last_source_from_author_removed() { let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); let s1 = at("at://did:plc:nel/sh.tangled.feed.star/r1"); let s2 = at("at://did:plc:nel/sh.tangled.feed.star/r2"); let s3 = at("at://did:plc:olaren/sh.tangled.feed.star/r3");
store.add(star_edge(s1.clone(), did("did:plc:abalone"))); store.add(star_edge(s2.clone(), did("did:plc:abalone"))); store.add(star_edge(s3, did("did:plc:abalone"))); assert_eq!(store.count_distinct_authors(&key), 2);
store.remove_source(&s1); assert_eq!(store.count_distinct_authors(&key), 2, "user1 still has s2");
store.remove_source(&s2); assert_eq!(store.count_distinct_authors(&key), 1, "user1 fully gone"); }
#[test] fn page_limit_rejects_zero_and_oversize() { assert!(matches!( PageLimit::new(0), Err(PageLimitError::TooSmall { value: 0, min: 1 }) )); assert!(matches!( PageLimit::new(PageLimit::MAX + 1), Err(PageLimitError::TooLarge { .. }) )); assert_eq!(PageLimit::new(50).unwrap().get(), 50); }
#[test] fn cursor_token_round_trip() { let original = PageToken::new(1_730_000_000_000_000, 0x1234_abcd); let token = original.encode_token(); assert_eq!(token.len(), 24, "12-byte cursor encodes to 24 hex chars"); assert_eq!(PageToken::decode_token(&token).unwrap(), original); }
#[test] fn from_token_none_yields_start() { assert_eq!(PageCursor::from_token(None).unwrap(), PageCursor::Start); }
#[test] fn from_token_some_yields_after() { let original = PageToken::new(1_730_000_000_000_000, 42); let token = original.encode_token(); assert_eq!( PageCursor::from_token(Some(&token)).unwrap(), PageCursor::After(original), ); }
#[test] fn cursor_decode_rejects_malformed() { let bad = [ "", "deadbeef", "no-hex!!aaaaaaaaaaaaaaaa", "12345", "this-string-is-way-too-long-to-be-a-valid-cursor", ]; bad.into_iter().for_each(|s| { assert!( matches!(PageToken::decode_token(s), Err(CursorParseError::Malformed)), "expected malformed for {s:?}", ); }); }
#[test] fn cursor_token_is_hex_shape() { let token = PageToken::new(1_730_000_000_000_000, 7).encode_token(); assert_eq!(token.len(), 24); assert!(token.chars().all(|c| c.is_ascii_hexdigit())); }
#[test] fn list_does_not_drop_entries_at_same_sort_micros_across_pages() { let store = store(); let subject_did = did("did:plc:limpet"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..4).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), subject_did.clone(), 42, )); });
let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc); assert_eq!(page1.items.len(), 2); let token = page1.next.expect("cursor must continue across ties");
let page2 = store.list(&key, PageCursor::After(token), limit(2), SortDir::Asc); assert_eq!( page2.items.len(), 2, "remaining ties must surface on next page" ); assert!(page2.next.is_none()); let combined: std::collections::HashSet<String> = page1 .items .iter() .chain(page2.items.iter()) .map(|u| u.as_ref().to_owned()) .collect(); assert_eq!(combined.len(), 4, "every tied entry visible exactly once"); }
#[test] fn list_filtered_narrows_by_predicate_and_paginates() { let store = store(); let subject_did = did("did:plc:limpet"); let key = EdgeKey::new( nsid("sh.tangled.repo.issue"), SubjectRef::Did(subject_did.clone()), ); let by_nel = (0..3).map(|i| { star_edge_at( at(&format!("at://did:plc:nel/sh.tangled.repo.issue/r{i}")), subject_did.clone(), 100 + i as u64, ) }); let by_olaren = (0..2).map(|i| { star_edge_at( at(&format!("at://did:plc:olaren/sh.tangled.repo.issue/o{i}")), subject_did.clone(), 500 + i as u64, ) }); by_nel .chain(by_olaren) .map(|mut e| { e.kind = nsid("sh.tangled.repo.issue"); e }) .for_each(|e| store.add(e));
let only_nel = |u: &AtUri<DefaultStr>| u.as_ref().starts_with("at://did:plc:nel/"); let page1 = store.list_filtered(&key, PageCursor::Start, limit(2), SortDir::Asc, only_nel); assert_eq!(page1.items.len(), 2); assert!(page1.next.is_some(), "cursor must allow more nel matches");
let page2 = store.list_filtered( &key, PageCursor::After(page1.next.unwrap()), limit(2), SortDir::Asc, only_nel, ); assert_eq!(page2.items.len(), 1, "only one nel issue left"); assert!(page2.next.is_none(), "tail page must not promise more"); }
#[test] fn list_descending_returns_newest_first() { let store = store(); let subject_did = did("did:plc:limpet"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), subject_did.clone(), 1_000_000 + i as u64 * 1_000_000, )); });
let asc = store.list(&key, PageCursor::Start, limit(5), SortDir::Asc); let desc = store.list(&key, PageCursor::Start, limit(5), SortDir::Desc); assert_eq!(asc.items.len(), 5); assert_eq!(desc.items.len(), 5); let asc_uris: Vec<_> = asc.items.iter().map(|u| u.as_ref().to_owned()).collect(); let mut reversed = asc_uris.clone(); reversed.reverse(); let desc_uris: Vec<_> = desc.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc"); }
#[test] fn list_descending_paginates_with_cursor() { let store = store(); let subject_did = did("did:plc:whelk"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), subject_did.clone(), 1_000_000 + i as u64 * 1_000_000, )); });
let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Desc); assert_eq!(page1.items.len(), 2); let cursor = PageCursor::After(page1.next.expect("desc page1 must continue")); let page2 = store.list(&key, cursor, limit(2), SortDir::Desc); assert_eq!(page2.items.len(), 2); let cursor2 = PageCursor::After(page2.next.expect("desc page2 must continue")); let page3 = store.list(&key, cursor2, limit(2), SortDir::Desc); assert_eq!(page3.items.len(), 1); assert!(page3.next.is_none());
let combined: std::collections::HashSet<String> = page1 .items .iter() .chain(page2.items.iter()) .chain(page3.items.iter()) .map(|u| u.as_ref().to_owned()) .collect(); assert_eq!( combined.len(), 5, "desc pagination visits each item exactly once" ); }
#[test] fn list_filtered_descending_respects_predicate() { let store = store(); let subject_did = did("did:plc:scallop"); let key = EdgeKey::new( nsid("sh.tangled.repo.issue"), SubjectRef::Did(subject_did.clone()), ); let by_nel = (0..3).map(|i| { star_edge_at( at(&format!("at://did:plc:nel/sh.tangled.repo.issue/r{i}")), subject_did.clone(), 100 + i as u64, ) }); let by_olaren = (0..2).map(|i| { star_edge_at( at(&format!("at://did:plc:olaren/sh.tangled.repo.issue/o{i}")), subject_did.clone(), 500 + i as u64, ) }); by_nel .chain(by_olaren) .map(|mut e| { e.kind = nsid("sh.tangled.repo.issue"); e }) .for_each(|e| store.add(e));
let only_nel = |u: &AtUri<DefaultStr>| u.as_ref().starts_with("at://did:plc:nel/"); let page = store.list_filtered(&key, PageCursor::Start, limit(5), SortDir::Desc, only_nel); assert_eq!(page.items.len(), 3, "all three nel issues visible"); let last = page.items.last().unwrap().as_ref(); let first = page.items.first().unwrap().as_ref(); assert!( first > last, "desc order: first item rkey must be greater than last (got first={first}, last={last})", ); }
#[test] fn non_did_source_round_trips_via_raw_fallback() { let store = store(); let subject = did("did:plc:limpet"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); let did_source = at("at://did:plc:nel/sh.tangled.feed.star/r1"); let handle_source = at("at://witchcraft.systems/sh.tangled.feed.star/r2"); store.add(star_edge_at(did_source.clone(), subject.clone(), 1)); store.add(star_edge_at(handle_source.clone(), subject.clone(), 2));
let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc); let got: std::collections::HashSet<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert!( got.contains(did_source.as_ref()), "did source must decode exactly" ); assert!( got.contains(handle_source.as_ref()), "non-did authority must round-trip through the raw fallback" );
store.remove_source(&handle_source); assert_eq!( store.count(&key), 1, "raw-keyed source removable by its uri" ); }
#[test] fn distinct_collections_decode_with_their_own_collection() { let store = store(); let subject = did("did:plc:limpet"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); let star_src = at("at://did:plc:nel/sh.tangled.feed.star/aaa"); let issue_src = at("at://did:plc:nel/sh.tangled.repo.issue/bbb"); store.add(Edge { kind: nsid("sh.tangled.feed.star"), subject: SubjectRef::Did(subject.clone()), source: star_src.clone(), sort_micros: 1, }); store.add(Edge { kind: nsid("sh.tangled.feed.star"), subject: SubjectRef::Did(subject.clone()), source: issue_src.clone(), sort_micros: 2, });
let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc); let got: std::collections::HashSet<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert!( got.contains(star_src.as_ref()), "star-collection source decodes exactly" ); assert!( got.contains(issue_src.as_ref()), "issue-collection source must keep its own collection, not borrow the star one" ); }
#[test] fn author_refs_spill_preserves_distinct_count() { let store = store(); let subject = did("did:plc:scallop"); let key = EdgeKey::new( nsid("sh.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i] )), subject.clone(), i as u64, )); }); assert_eq!( store.count_distinct_authors(&key), 5, "five authors exceed the inline cap and spill to the map" );
store.add(star_edge_at( at("at://did:plc:nel/sh.tangled.feed.star/r99"), subject.clone(), 99, )); assert_eq!( store.count_distinct_authors(&key), 5, "second nel source adds no author" );
store.remove_source(&at("at://did:plc:nel/sh.tangled.feed.star/r0")); assert_eq!( store.count_distinct_authors(&key), 5, "nel still present via r99" ); store.remove_source(&at("at://did:plc:nel/sh.tangled.feed.star/r99")); assert_eq!(store.count_distinct_authors(&key), 4, "nel fully removed"); }}