Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257use std::collections::{HashMap, HashSet};use std::marker::PhantomData;use std::num::NonZeroU32;use std::sync::atomic::{AtomicU32, Ordering};use std::sync::{Arc, Mutex};
use bobbin_runtime::RuntimeHasher;use bobbin_types::edges::{Edge, Record, TICKET_ISSUE_KIND, TICKET_PULL_KIND};use bobbin_types::ids::{EdgeKey, SubjectRef};use either::Either;use itertools::Itertools;use jacquard_common::DefaultStr;use jacquard_common::deps::smol_str::SmolStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;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;
#[repr(C)]#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]pub(crate) struct SortMicros { high: u32, low: u32,}
impl SortMicros { const fn new(value: u64) -> Self { Self { high: (value >> 32) as u32, low: value as u32, } }
const fn get(self) -> u64 { ((self.high as u64) << 32) | self.low as u64 }}
#[repr(C)]#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]pub(crate) struct BucketKey { micros: SortMicros, source: SourceId,}
impl BucketKey { pub(crate) const fn new(micros: u64, source: SourceId) -> Self { Self { micros: SortMicros::new(micros), source, } }
fn token(self) -> PageToken { PageToken::new(self.micros.get(), self.source.index()) }
fn from_token(tok: PageToken) -> Self { Self::new(tok.micros, SourceId::from_raw(tok.source)) }}
pub mod claims;pub mod coverage;pub mod invites;pub mod settle;mod sorted_blocks;pub mod state_index;mod state_view;pub mod stream;pub use claims::{ClaimDelta, ClaimIndex, RepoTransition, Rewrite};pub use coverage::{Coverage, CoverageWatch, HydrantCursor, PromotionSignal};pub use invites::{InviteIndex, InviteScope, Offer};pub use settle::{ParsedCid, RecordOutcome, Rejection, Settlement, Settlements, Watch};use sorted_blocks::{PageKey, SortedBlocks};pub use state_index::{StateKind, TicketStateKind};use state_view::{ProjectedState, StateViewIndex};pub use stream::{StreamHealth, StreamHealthSnapshot, StreamState};
#[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>;pub(crate) 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, Hash, Ord, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize,)]#[serde(transparent)]pub struct PageOffset(u64);
impl PageOffset { pub const ZERO: Self = Self(0);
pub const fn new(value: u64) -> Self { Self(value) }
pub const fn get(self) -> u64 { self.0 }
pub const fn as_usize(self) -> usize { self.0 as usize }}
impl From<u64> for PageOffset { fn from(v: u64) -> Self { Self(v) }}
impl From<usize> for PageOffset { fn from(v: usize) -> Self { Self(v as u64) }}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct PageRank(usize);
impl PageRank { pub const ZERO: Self = Self(0);
pub const fn new(value: usize) -> Self { Self(value) }
pub const fn get(self) -> usize { self.0 }}
impl From<usize> for PageRank { fn from(v: usize) -> Self { Self(v) }}
#[derive( Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize,)]#[serde(transparent)]pub struct TotalCount(u64);
impl TotalCount { pub const ZERO: Self = Self(0);
pub const fn new(value: u64) -> Self { Self(value) }
pub const fn get(self) -> u64 { self.0 }}
impl From<u64> for TotalCount { fn from(v: u64) -> Self { Self(v) }}
impl From<usize> for TotalCount { fn from(v: usize) -> Self { Self(v as u64) }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub enum PageStart { Cursor(PageCursor), Offset(PageOffset),}
impl From<PageCursor> for PageStart { fn from(c: PageCursor) -> Self { Self::Cursor(c) }}
impl From<PageOffset> for PageStart { fn from(o: PageOffset) -> Self { Self::Offset(o) }}
impl From<u64> for PageStart { fn from(v: u64) -> Self { Self::Offset(PageOffset::new(v)) }}
#[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<EdgeItem>, pub next: Option<PageToken>, pub total: Option<TotalCount>,}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct EdgeItem { pub uri: AtUri<DefaultStr>, pub sort_micros: u64, pub sort_source: u32,}
impl EdgeItem { pub fn sort_token(&self) -> PageToken { PageToken::new(self.sort_micros, self.sort_source) }}
impl AsRef<str> for EdgeItem { fn as_ref(&self) -> &str { self.uri.as_ref() }}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]pub struct FilteredCount { pub count: Count, pub distinct_authors: DistinctAuthorCount,}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]pub struct Count(u64);
impl Count { pub const fn new(value: u64) -> Self { Self(value) }
pub const fn get(self) -> u64 { self.0 }}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]pub struct DistinctAuthorCount(u64);
impl DistinctAuthorCount { pub const fn new(value: u64) -> Self { Self(value) }
pub const fn get(self) -> u64 { self.0 }}
impl FilteredCount { pub const fn new(count: u64, distinct_authors: u64) -> Self { Self { count: Count::new(count), distinct_authors: DistinctAuthorCount::new(distinct_authors), } }}
#[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)]pub(crate) struct EdgeKeyId(u32);
pub(crate) 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);}
pub(crate) 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()); } }}
enum AuthorHistory { Small(SmallVec<[BucketKey; 4]>), Large(BlockStore),}
impl AuthorHistory { fn pair(left: BucketKey, right: BucketKey) -> Self { let mut keys = SmallVec::from_slice(&[left, right]); keys.sort_unstable(); Self::Small(keys) }
fn len(&self) -> usize { match self { Self::Small(keys) => keys.len(), Self::Large(keys) => keys.len(), } }
fn newest(&self) -> BucketKey { match self { Self::Small(keys) => *keys.last().unwrap(), Self::Large(keys) => keys.last().expect("nonempty author history"), } }
fn insert(&mut self, key: BucketKey) -> bool { match self { Self::Small(keys) => match keys.binary_search(&key) { Ok(_) => false, Err(position) if keys.len() < keys.inline_size() => { keys.insert(position, key); true } Err(position) => { let mut promoted = keys.to_vec(); promoted.insert(position, key); *self = Self::Large(BlockStore::from_sorted(promoted)); true } }, Self::Large(keys) => keys.insert(key), } }
fn remove(&mut self, key: &BucketKey) -> bool { match self { Self::Small(keys) => keys .binary_search(key) .map(|position| { keys.remove(position); }) .is_ok(), Self::Large(keys) => { if !keys.remove(key) { return false; } if keys.len() <= 4 { *self = Self::Small(keys.directed(PageCursor::Start, SortDir::Asc).collect()); } true } } }
fn heap_bytes(&self) -> u64 { match self { Self::Small(keys) if keys.spilled() => keys.capacity() as u64 * BUCKET_KEY_BYTES, Self::Small(_) => 0, Self::Large(keys) => keys.heap_bytes(), } }}
fn estimated_hashmap_heap_bytes<K, V, S>(map: &HashMap<K, V, S>) -> u64 { let slot = std::mem::size_of::<(K, V)>() as u64 + 1; map.capacity() as u64 * slot}
struct AuthorIndex { newest: HashMap<AuthorId, BucketKey, RuntimeHasher>, histories: HashMap<AuthorId, Box<AuthorHistory>, RuntimeHasher>, representatives: BlockStore,}
impl AuthorIndex { fn new(hasher: RuntimeHasher) -> Self { Self { newest: HashMap::with_hasher(hasher.clone()), histories: HashMap::with_hasher(hasher), representatives: BlockStore::from_sorted(Vec::new()), } }
fn len(&self) -> usize { self.newest.len() }
fn contains(&self, author: &AuthorId) -> bool { self.newest.contains_key(author) }
fn count(&self, author: &AuthorId) -> usize { self.newest .contains_key(author) .then(|| { self.histories .get(author) .map_or(1, |history| history.len()) }) .unwrap_or(0) }
fn replace_representative(&mut self, previous: BucketKey, next: BucketKey) { if self.representatives.replace_last(previous, next) { return; } let removed = self.representatives.remove(&previous); let inserted = self.representatives.insert(next); debug_assert!(removed && inserted); }
fn insert(&mut self, author: AuthorId, key: BucketKey) { let Some(previous) = self.newest.get(&author).copied() else { let inserted = self.representatives.insert(key); debug_assert!(inserted); self.newest.insert(author, key); return; }; let inserted = match self.histories.get_mut(&author) { Some(history) => history.insert(key), None => { self.histories .insert(author, Box::new(AuthorHistory::pair(previous, key))); true } }; debug_assert!(inserted); if key > previous { self.replace_representative(previous, key); self.newest.insert(author, key); } }
fn remove(&mut self, author: AuthorId, key: &BucketKey) { let Some(newest) = self.newest.get(&author).copied() else { debug_assert!(false, "large bucket missing author"); return; }; let Some(history) = self.histories.get_mut(&author) else { debug_assert_eq!(*key, newest, "single author key is not newest"); self.newest.remove(&author); let removed = self.representatives.remove(&newest); debug_assert!(removed); return; }; let removed = history.remove(key); debug_assert!(removed); let next = history.newest(); if history.len() == 1 { self.histories.remove(&author); } if *key == newest { self.replace_representative(newest, next); self.newest.insert(author, next); } }
fn heap_bytes(&self) -> u64 { estimated_hashmap_heap_bytes(&self.newest) + estimated_hashmap_heap_bytes(&self.histories) + self.representatives.heap_bytes() + self .histories .values() .map(|history| std::mem::size_of::<AuthorHistory>() as u64 + history.heap_bytes()) .sum::<u64>() }}
struct LargeBucket { keys: BlockStore, authors: AuthorIndex,}
impl LargeBucket { fn insert(&mut self, key: BucketKey, author: Option<AuthorId>) -> bool { if !self.keys.insert(key) { return false; } if let Some(author) = author { self.authors.insert(author, key); } true }
fn remove(&mut self, key: &BucketKey, author: Option<AuthorId>) -> bool { if !self.keys.remove(key) { return false; } if let Some(author) = author { self.authors.remove(author, key); } true }}
const BUCKET_PROMOTE_AT: usize = 256;const BUCKET_KEY_BYTES: u64 = std::mem::size_of::<BucketKey>() as u64;
type BlockStore = SortedBlocks<BucketKey>;
impl PageKey for BucketKey { fn token(self) -> PageToken { self.token() }
fn from_token(tok: PageToken) -> Self { Self::from_token(tok) }}
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) => big.insert(key, author), } }
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) => { big.remove(key, author); } } }
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) => big.keys.directed(cursor, dir), } }
fn select(&self, rank: PageRank) -> Option<BucketKey> { match self { Self::Small(v) => v.get(rank.get()).copied(), Self::Large(big) => big.keys.select(rank), } }
fn resolve_start(&self, start: PageStart, dir: SortDir) -> Option<PageCursor> { match start { PageStart::Cursor(c) => Some(c), PageStart::Offset(o) if o.get() == 0 => Some(PageCursor::Start), PageStart::Offset(o) => { let len = self.len() as u64; let o = o.get(); if o >= len { return None; } let pred_rank = match dir { SortDir::Asc => o - 1, SortDir::Desc => len - o, }; self.select(PageRank::new(pred_rank as usize)) .map(|k| PageCursor::After(k.token())) } } }
fn heap_bytes(&self) -> u64 { match self { Self::Small(v) if v.spilled() => v.capacity() as u64 * BUCKET_KEY_BYTES, Self::Small(_) => 0, Self::Large(big) => { std::mem::size_of::<LargeBucket>() as u64 + big.keys.heap_bytes() + big.authors.heap_bytes() } } }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]struct ReverseEntry { key_id: EdgeKeyId, sort_micros: SortMicros,}
#[cfg(target_pointer_width = "64")]const _: () = { assert!(std::mem::size_of::<SortMicros>() == 8); assert!(std::mem::align_of::<SortMicros>() == 4); assert!(std::mem::size_of::<BucketKey>() == 12); assert!(std::mem::align_of::<BucketKey>() == 4); assert!(std::mem::size_of::<ReverseEntry>() == 12); assert!(std::mem::size_of::<(AuthorId, BucketKey)>() == 16); assert!(std::mem::size_of::<(AuthorId, Box<AuthorHistory>)>() == 16);};
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>, keys: SccMap<EdgeKeyId, EdgeKey, RuntimeHasher>, next_key_id: AtomicU32, forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>, reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>, source_collections: SccMap<SourceId, Vec<SmolStr>, RuntimeHasher>, ticket_states: StateViewIndex<TicketStateKind>, 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()), keys: SccMap::with_hasher(hasher.clone()), next_key_id: AtomicU32::new(0), forward: SccMap::with_hasher(hasher.clone()), reverse: SccMap::with_hasher(hasher.clone()), source_collections: SccMap::with_hasher(hasher.clone()), ticket_states: StateViewIndex::new(hasher.clone()), hasher, writer: Mutex::new(()), } }
fn intern_key(&self, key: EdgeKey) -> EdgeKeyId { match self.key_ids.entry_sync(key.clone()) { Entry::Occupied(e) => *e.get(), Entry::Vacant(e) => { let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed)); e.insert_entry(id); let _ = self.keys.insert_sync(id, key); 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); }
/// Store collection filters for a source. None or empty = all collections (removes entry). /// Call only after the source has been interned via add/upsert_source. pub fn set_source_collections( &self, source: &AtUri<DefaultStr>, collections: Option<Vec<SmolStr>>, ) { let author = source_authority_did(source) .and_then(|s| self.did_interner.get(s)) .map(AuthorId::from_spur); let Some(spur) = self.source_interner.get(self.source_key(source, author)) else { return; }; let id = SourceId::from_spur(spur); if let Some(cols) = collections && !cols.is_empty() { let _ = self.source_collections.insert_sync(id, cols); } else { self.source_collections.remove_sync(&id); } }
/// Returns None if no collection filter is stored (meaning all collections). pub fn source_collections_for(&self, source: &AtUri<DefaultStr>) -> Option<Vec<SmolStr>> { let author = source_authority_did(source) .and_then(|s| self.did_interner.get(s)) .map(AuthorId::from_spur); let spur = self.source_interner.get(self.source_key(source, author))?; let id = SourceId::from_spur(spur); self.source_collections .read_sync(&id, |_, cols| cols.clone()) } 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_index(&keys); let large = LargeBucket { keys: BlockStore::from_sorted(keys), 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: SortMicros::new(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); self.source_collections.remove_sync(&id); 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.get(), 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 edge_item(&self, key: BucketKey) -> Option<EdgeItem> { let spur = key.source.to_spur()?; let stored = self.source_interner.try_resolve(&spur)?; let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?; Some(EdgeItem { uri, sort_micros: key.micros.get(), sort_source: key.token().source(), }) }
fn edge_page( &self, total: usize, keys: impl Iterator<Item = BucketKey>, limit: PageLimit, ) -> EdgePage { let limit = limit.get() as usize; let entries = keys.take(limit + 1).collect::<Vec<_>>(); let has_more = entries.len() > limit; let page = &entries[..entries.len().min(limit)]; let items = page.iter().filter_map(|&key| self.edge_item(key)).collect(); let next = has_more .then(|| page.last().copied()) .flatten() .map(BucketKey::token); EdgePage { items, next, total: Some(TotalCount::from(total)), } }
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_index(&self, keys: &[BucketKey]) -> AuthorIndex { debug_assert!(keys.windows(2).all(|pair| pair[0] < pair[1])); let mut authors = AuthorIndex::new(self.hasher.clone()); keys.iter().for_each(|&key| { if let Some(author) = self.author_of(key.source) { authors.insert(author, key); } }); authors } fn distinct_authors(&self, sources: &Sources) -> u64 { match sources { Sources::Large(big) => big.authors.len() as u64, Sources::Small(keys) => self.distinct_authors_small(keys), } }
fn distinct_authors_since(&self, sources: &Sources, since_micros: u64) -> u64 { sources .directed(PageCursor::Start, SortDir::Asc) .filter(|key| key.micros.get() >= since_micros) .filter_map(|key| self.author_of(key.source)) .collect::<std::collections::HashSet<AuthorId>>() .len() as u64 }
fn ranked_dids(&self, kind: &str, since_micros: Option<u64>, limit: usize) -> Vec<Did> { let mut ranked = Vec::with_capacity(self.forward.len()); self.forward.iter_sync(|key_id, sources| { let Some(key) = self.keys.read_sync(key_id, |_, key| key.clone()) else { return true; }; let SubjectRef::Did(did) = key.subject else { return true; }; if key.kind.as_ref() != kind { return true; } let count = since_micros.map_or_else( || self.distinct_authors(sources), |since| self.distinct_authors_since(sources, since), ); if count > 0 { ranked.push((count, did)); } true }); ranked.sort_unstable_by(|(left_count, left_did), (right_count, right_did)| { right_count .cmp(left_count) .then_with(|| left_did.as_str().cmp(right_did.as_str())) }); ranked.into_iter().take(limit).map(|(_, did)| did).collect() }
pub fn most_followed(&self, limit: usize) -> Vec<Did> { self.ranked_dids("org.tangled.graph.follow", None, limit) }
pub fn top_starred_since(&self, since_micros: u64, limit: usize) -> Vec<Did> { self.ranked_dids("org.tangled.feed.star", Some(since_micros), limit) }
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| self.distinct_authors(sources)) }) .unwrap_or(0) }
pub fn count_by_author(&self, key: &EdgeKey, author: &Did<DefaultStr>) -> u64 { let Some(author) = self .did_interner .get(author.as_ref()) .map(AuthorId::from_spur) else { return 0; }; self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| match sources { Sources::Large(big) => big.authors.count(&author) as u64, Sources::Small(keys) => keys .iter() .filter(|key| self.author_of(key.source) == Some(author)) .count() as u64, }) }) .unwrap_or(0) }
pub fn refresh_ticket_counts(&self, entity: &AtUri<DefaultStr>, kind: TicketStateKind) { // compute under the write lock so concurrent updates dont write a stale projection let mut inner = self.ticket_states.write_inner(); let source = self.intern_source(entity); let author = self.author_of(source); let reverse = self .reverse .read_sync(&source, |_, entries| entries.clone()) .unwrap_or_default(); let current = reverse .into_iter() .filter_map(|entry| { let key = self.keys.read_sync(&entry.key_id, |_, key| key.clone())?; [TICKET_ISSUE_KIND, TICKET_PULL_KIND] .contains(&key.kind.as_ref()) .then(|| ProjectedState { key: entry.key_id, kind, author, micros: entry.sort_micros.get(), // tickets are authored by the repo, so author sub-views would mirror the bucket by_bucket: true, }) }) .collect(); StateViewIndex::apply_projection(&mut inner, source, ¤t); }
pub fn count_ticket_state(&self, key: &EdgeKey, kind: TicketStateKind) -> FilteredCount { let Some(key) = self.lookup_key(key) else { return FilteredCount::default(); }; let (count, distinct) = self.ticket_states.counts(key, kind, None); FilteredCount::new(count, distinct) }
pub fn page_ticket_state( &self, key: &EdgeKey, kind: TicketStateKind, start: impl Into<PageStart>, limit: PageLimit, dir: SortDir, ) -> EdgePage { let page = self.lookup_key(key).and_then(|key_id| { self.ticket_states .page(key_id, Some(kind), start.into(), limit, dir, None) }); let Some(page) = page else { return EdgePage { items: Vec::new(), next: None, total: Some(TotalCount::ZERO), }; }; let items = page .keys .iter() .filter_map(|&key| self.edge_item(key)) .collect(); EdgePage { items, next: page.next, total: Some(page.total), } }
/// answers "did the viewer star/follow/etc. this subject, and with what rkey" pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option<AtUri<DefaultStr>> { let author_spur = self.did_interner.get(viewer)?; let author_id = AuthorId::from_spur(author_spur); let key_id = self.lookup_key(key)?;
self.forward .read_sync(&key_id, |_, sources| { // large buckets track authors, so a missing author fast-fails the scan if let Sources::Large(big) = sources && !big.authors.contains(&author_id) { return None; } sources .directed(PageCursor::Start, SortDir::Desc) .find_map(|bucket| { let spur = bucket.source.to_spur()?; let stored = self.source_interner.try_resolve(&spur)?; let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?; (source_authority_did(&uri) == Some(viewer)).then_some(uri) }) }) .flatten() }
pub fn sources_for(&self, key: &EdgeKey) -> Vec<AtUri<DefaultStr>> { self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| { sources .directed(PageCursor::Start, SortDir::Desc) .filter_map(|bucket| { let spur = bucket.source.to_spur()?; let stored = self.source_interner.try_resolve(&spur)?; AtUri::new_owned(self.decode_source(stored)?).ok() }) .collect::<Vec<_>>() }) }) .unwrap_or_default() }
pub fn list( &self, key: &EdgeKey, start: impl Into<PageStart>, limit: PageLimit, dir: SortDir, ) -> EdgePage { let start = start.into(); self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| { let total = sources.len(); let Some(cursor) = sources.resolve_start(start, dir) else { return EdgePage { items: Vec::new(), next: None, total: Some(TotalCount::from(total)), }; }; self.edge_page(total, sources.directed(cursor, dir), limit) }) }) .unwrap_or(EdgePage { items: Vec::new(), next: None, total: None, }) }
pub fn list_distinct_authors( &self, key: &EdgeKey, start: impl Into<PageStart>, limit: PageLimit, dir: SortDir, ) -> EdgePage { let start = start.into(); self.lookup_key(key) .and_then(|id| { self.forward.read_sync(&id, |_, sources| match sources { Sources::Small(keys) => { let mut seen = HashSet::with_hasher(self.hasher.clone()); let mut representatives = keys .iter() .rev() .filter(|key| { self.author_of(key.source) .is_some_and(|author| seen.insert(author)) }) .copied() .collect::<SmallVec<[BucketKey; 2]>>(); representatives.reverse(); let representatives = Sources::Small(representatives); let total = representatives.len(); let Some(cursor) = representatives.resolve_start(start, dir) else { return EdgePage { items: Vec::new(), next: None, total: Some(TotalCount::from(total)), }; }; self.edge_page(total, representatives.directed(cursor, dir), limit) } Sources::Large(big) => { let total = big.authors.representatives.len(); debug_assert_eq!(total, big.authors.len()); let Some(cursor) = big.authors.representatives.resolve_start(start, dir) else { return EdgePage { items: Vec::new(), next: None, total: Some(TotalCount::from(total)), }; }; self.edge_page( total, big.authors.representatives.directed(cursor, dir), limit, ) } }) }) .unwrap_or(EdgePage { items: Vec::new(), next: None, total: None, }) }
/// Merge a page across many buckets in one global time order. pub fn list_multi( &self, keys: &[EdgeKey], cursor: PageCursor, limit: PageLimit, dir: SortDir, ) -> EdgePage { let n = limit.get() as usize + 1; let is_first = |x: &BucketKey, y: &BucketKey| match dir { SortDir::Asc => x <= y, SortDir::Desc => x >= y, }; let mut top: Vec<BucketKey> = Vec::with_capacity(n); for key in keys { let Some(id) = self.lookup_key(key) else { continue; }; let bucket: Vec<BucketKey> = self .forward .read_sync(&id, |_, sources| { sources.directed(cursor, dir).take(n).collect() }) .unwrap_or_default(); top = top .into_iter() .merge_by(bucket, &is_first) .take(n) .collect(); } top.dedup_by_key(|k| k.source);
let has_more = top.len() > limit.get() as usize; let page = &top[..top.len().min(limit.get() as usize)]; let items = page.iter().filter_map(|&key| self.edge_item(key)).collect(); let next = has_more .then(|| page.last().copied()) .flatten() .map(BucketKey::token); EdgePage { items, next, total: 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 = self.ticket_states.heap_bytes(); 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) + self.keys.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, } }}
pub fn upsert_record_indexes( edges: &EdgeStore, claims: &ClaimIndex, source: &AtUri<DefaultStr>, record_edges: Vec<Edge>, record: &Record,) -> Vec<RepoTransition> { // manifests only get their edges once claimed, and a declaration can flip a manifest it points at let transitions = match record { Record::Repo(m) => apply_claims( edges, claims.upsert_manifest( source, m.declaration.owner.as_ref(), m.declaration.name.0.as_str(), record_edges, ), ), Record::Declaration(d) => { apply_claims(edges, claims.upsert_declaration(source, d.repo.as_ref())) } _ => { edges.upsert_source(source, record_edges); Vec::new() } }; if let Record::Ticket(t) = record { // an unknown ticket state counts as open edges.refresh_ticket_counts( source, TicketStateKind::from_ticket(t.state.as_ref()).unwrap_or_default(), ); } transitions}
fn apply_claims(edges: &EdgeStore, delta: ClaimDelta) -> Vec<RepoTransition> { for rewrite in delta.rewrites { edges.upsert_source(&rewrite.source, rewrite.edges); } delta.transitions}
pub fn delete_record_indexes( edges: &EdgeStore, claims: &ClaimIndex, source: &AtUri<DefaultStr>, nsid: &Nsid<DefaultStr>,) { edges.remove_source(source); match nsid.as_ref() { "org.tangled.repo.manifest" | "org.tangled.repo.declaration" => { apply_claims(edges, claims.delete(source)); } "org.tangled.track.ticket" => edges.refresh_ticket_counts(source, TicketStateKind::Open), _ => {} }}
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 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("org.tangled.feed.star"), subject: SubjectRef::Did(subject), source, sort_micros, } }
fn follow_edge_at( source: AtUri<DefaultStr>, subject: Did<DefaultStr>, sort_micros: u64, ) -> Edge { Edge { kind: nsid("org.tangled.graph.follow"), 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:{}/org.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("org.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: PageLimit, ) -> Vec<String> { std::iter::successors( Some(store.list(key, PageCursor::Start, page, dir)), |prev| { prev.next .map(|tok| store.list(key, PageCursor::After(tok), page, dir)) }, ) .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect() }
fn paginate_distinct_authors( store: &EdgeStore, key: &EdgeKey, dir: SortDir, page: PageLimit, ) -> Vec<String> { std::iter::successors( Some(store.list_distinct_authors(key, PageCursor::Start, page, dir)), |prev| { prev.next.map(|token| { store.list_distinct_authors(key, PageCursor::After(token), page, dir) }) }, ) .flat_map(|page| page.items.into_iter().map(|item| item.as_ref().to_owned())) .collect() }
fn newest_author_reference(rows: impl IntoIterator<Item = String>) -> Vec<String> { let mut authors = HashSet::new(); let mut newest = rows .into_iter() .collect::<Vec<_>>() .into_iter() .rev() .filter(|uri| { let author = split_record_uri(uri).unwrap().0; authors.insert(author.to_owned()) }) .collect::<Vec<_>>(); newest.reverse(); newest }
#[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, limit(page)), asc, "asc mismatch n={n} page={page}" ); assert_eq!( paginate_all(&store, &key, SortDir::Desc, limit(page)), desc, "desc mismatch n={n} page={page}" ); }); }); }
fn paginate_by_offsets( store: &EdgeStore, key: &EdgeKey, dir: SortDir, page: PageLimit, ) -> Vec<String> { std::iter::successors( Some(( PageOffset::ZERO, store.list(key, PageOffset::ZERO, page, dir), )), |(offset, prev)| { prev.next.map(|_| { let next = PageOffset::from(offset.get() + page.get() as u64); (next, store.list(key, next, page, dir)) }) }, ) .flat_map(|(_, p)| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect() }
#[test] fn distinct_author_pagination_matches_reference_across_small_and_large() { [64usize, 1000].into_iter().for_each(|n| { let store = store(); let key = fill_subject(&store, n); let asc = newest_author_reference(reference(0..n)); let desc = asc.iter().rev().cloned().collect::<Vec<_>>();
[1u32, 2, 5].into_iter().for_each(|page| { assert_eq!( paginate_distinct_authors(&store, &key, SortDir::Asc, limit(page)), asc, "asc distinct mismatch n={n} page={page}", ); assert_eq!( paginate_distinct_authors(&store, &key, SortDir::Desc, limit(page)), desc, "desc distinct mismatch n={n} page={page}", ); });
let offset = store.list_distinct_authors(&key, PageOffset::new(2), limit(2), SortDir::Asc); assert_eq!( offset .items .iter() .map(|item| item.as_ref().to_owned()) .collect::<Vec<_>>(), asc.iter().skip(2).take(2).cloned().collect::<Vec<_>>(), ); assert_eq!( offset.total, Some(TotalCount::from(NAMES.len())), "distinct total n={n}", ); }); }
#[test] fn large_distinct_index_tracks_newest_representative() { let store = store(); let subject = did("did:plc:large-subject"); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); (0..255).for_each(|index| { store.add(star_edge_at( at(&format!( "at://did:plc:author{index}/org.tangled.feed.star/r{index}" )), subject.clone(), index as u64, )); }); let older = at("at://did:plc:repeat/org.tangled.feed.star/older"); let newest = at("at://did:plc:repeat/org.tangled.feed.star/newest"); store.add(star_edge_at(older.clone(), subject.clone(), 1_000)); store.add(star_edge_at(newest.clone(), subject.clone(), 2_000));
let listed = || { store .list_distinct_authors( &key, PageCursor::Start, limit(PageLimit::MAX), SortDir::Desc, ) .items .into_iter() .map(|item| item.uri) .collect::<HashSet<_>>() }; assert_eq!(store.count(&key), 257); assert_eq!(store.count_distinct_authors(&key), 256); assert!(listed().contains(&newest)); assert!(!listed().contains(&older));
store.remove_source(&newest); assert!(listed().contains(&older));
store.add(star_edge_at(newest.clone(), subject, 3_000)); store.remove_source(&older); assert!(listed().contains(&newest)); assert_eq!(store.count_distinct_authors(&key), 256); }
#[test] fn author_history_promotes_and_shrinks_without_losing_the_newest() { let store = store(); let subject = did("did:plc:large-subject"); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); (0..255).for_each(|index| { store.add(star_edge_at( at(&format!( "at://did:plc:author{index}/org.tangled.feed.star/r{index}" )), subject.clone(), index as u64, )); }); let repeated = (0..6) .map(|index| { at(&format!( "at://did:plc:repeat/org.tangled.feed.star/r{index}" )) }) .collect::<Vec<_>>(); repeated.iter().enumerate().for_each(|(index, source)| { store.add(star_edge_at( source.clone(), subject.clone(), 1_000 + index as u64, )); }); let newest = || { store .list_distinct_authors(&key, PageCursor::Start, limit(1), SortDir::Desc) .items[0] .uri .clone() };
repeated .iter() .enumerate() .rev() .for_each(|(index, source)| { assert_eq!(newest(), *source); store.remove_source(source); assert_eq!( store.count_distinct_authors(&key), 255 + u64::from(index != 0), ); }); }
#[test] fn offset_pagination_matches_reference_across_small_and_large() { [64usize, 1000].into_iter().for_each(|n| { let store = store(); let key = fill_subject(&store, 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_by_offsets(&store, &key, SortDir::Asc, limit(page)), asc, "asc offset walk mismatch n={n} page={page}" ); assert_eq!( paginate_by_offsets(&store, &key, SortDir::Desc, limit(page)), desc, "desc offset walk mismatch n={n} page={page}" ); });
for o in [1u64, 5, 63, 255, 256, 257, 999] { let o = o.min(n as u64 - 1); let page = store.list( &key, PageStart::Offset(PageOffset::from(o)), limit(10), SortDir::Asc, ); let want: Vec<String> = asc.iter().skip(o as usize).take(10).cloned().collect(); let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!(got, want, "asc offset {o} mismatch n={n}"); assert_eq!(page.total, Some(TotalCount::from(n as u64)), "total n={n}"); }
let page = store.list( &key, PageStart::Offset(PageOffset::from(n as u64)), limit(10), SortDir::Asc, ); assert!(page.items.is_empty(), "offset==len must be empty n={n}"); assert!(page.next.is_none()); assert_eq!(page.total, Some(TotalCount::from(n as u64))); }); }
#[test] fn offset_page_cursor_continues_sequentially() { let n = 1000usize; let store = store(); let key = fill_subject(&store, n); let asc = reference(0..n);
let first = store.list( &key, PageStart::Offset(PageOffset::new(600)), limit(7), SortDir::Asc, ); let got: Vec<String> = std::iter::successors(Some(first), |prev| { prev.next .map(|tok| store.list(&key, PageCursor::After(tok), limit(7), SortDir::Asc)) }) .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect(); assert_eq!(got, asc.into_iter().skip(600).collect::<Vec<_>>()); }
#[test] fn offset_desc_matches_cursor_desc_small_bucket() { let store = store(); let key = fill_subject(&store, 5); let desc: Vec<String> = reference(0..5).into_iter().rev().collect();
let first = store.list( &key, PageStart::Offset(PageOffset::new(2)), limit(3), SortDir::Desc, ); let got: Vec<String> = std::iter::successors(Some(first), |prev| { prev.next .map(|tok| store.list(&key, PageCursor::After(tok), limit(3), SortDir::Desc)) }) .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect(); assert_eq!(got, desc.into_iter().skip(2).collect::<Vec<_>>()); }
#[test] fn offset_desc_matches_cursor_desc_large_bucket() { let n = 1000usize; let store = store(); let key = fill_subject(&store, n); let desc: Vec<String> = reference(0..n).into_iter().rev().collect();
let first = store.list( &key, PageStart::Offset(PageOffset::new(600)), limit(7), SortDir::Desc, ); let got: Vec<String> = std::iter::successors(Some(first), |prev| { prev.next .map(|tok| store.list(&key, PageCursor::After(tok), limit(7), SortDir::Desc)) }) .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect(); assert_eq!(got, desc.into_iter().skip(600).collect::<Vec<_>>()); }
#[test] fn offset_boundary_matrix_asc_and_desc() { // offsets {0, 1, len-1, len} across asc and desc pin resolve_start pred_rank math for &n in &[5usize, 300] { let store = store(); let key = fill_subject(&store, n); let asc = reference(0..n); let desc: Vec<String> = asc.iter().rev().cloned().collect();
for &(dir, ref_order, label) in &[(SortDir::Asc, &asc, "asc"), (SortDir::Desc, &desc, "desc")] { let p = store.list( &key, PageStart::Offset(PageOffset::ZERO), limit(n as u32), dir, ); let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!(got, ref_order[..n].to_vec(), "n={n} {label} offset=0");
let p = store.list( &key, PageStart::Offset(PageOffset::new(1)), limit(n as u32), dir, ); let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!(got, ref_order[1..].to_vec(), "n={n} {label} offset=1");
let p = store.list( &key, PageStart::Offset(PageOffset::from(n as u64 - 1)), limit(n as u32), dir, ); let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!( got, vec![ref_order[n - 1].clone()], "n={n} {label} offset=len-1" ); assert!(p.next.is_none(), "last item has no continuation");
let p = store.list( &key, PageStart::Offset(PageOffset::from(n as u64)), limit(n as u32), dir, ); assert!(p.items.is_empty(), "n={n} {label} offset=len is empty"); assert_eq!( p.total, Some(TotalCount::from(n as u64)), "past-end page still reports total" ); } } } #[test] fn descending_cursor_block_boundary_edges() { let store = store(); let n = 600usize; let key = fill_subject(&store, n); let asc = reference(0..n); let desc: Vec<String> = asc.iter().rev().cloned().collect();
for target_idx in [255usize, 256, 257, 511, 512] { let p_single = store.list( &key, PageStart::Offset(PageOffset::from(target_idx as u64)), limit(1), SortDir::Asc, ); let tok = p_single.next.unwrap(); let page = store.list( &key, PageStart::Cursor(PageCursor::After(tok)), limit(600), SortDir::Desc, ); let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect(); let expected = &desc[n - target_idx..]; assert_eq!(got, expected, "descending cursor after index {target_idx}"); }
let page = store.list( &key, PageStart::Offset(PageOffset::new(256)), limit(10), SortDir::Desc, ); let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect(); assert_eq!(got, desc[256..266].to_vec(), "descending offset 256"); } #[test] fn empty_store_descending_after_cursor_does_not_underflow() { let store = store(); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(did("did:plc:empty_test")), ); let tok = PageToken::new(12345, 1); let page = store.list( &key, PageStart::Cursor(PageCursor::After(tok)), limit(10), SortDir::Desc, ); assert!(page.items.is_empty()); }
#[test] fn property_random_offsets_match_reference() { // seeded random inserts with duplicate attempts swept across offsets fn xorshift(mut x: u64) -> u64 { x ^= x << 13; x ^= x >> 7; x ^= x << 17; x }
let store = store(); let key = EdgeKey::new( nsid("sh.tangled.repo.issue"), SubjectRef::Did(did("did:plc:prop")), );
let mut rng = 0xdead_beef_cafe_babe_u64; let mut order: Vec<usize> = (0..500).collect(); for i in (1..order.len()).rev() { rng = xorshift(rng); let j = rng as usize % (i + 1); order.swap(i, j); } for _ in 0..100 { rng = xorshift(rng); order.push(rng as usize % 500); }
for &i in &order { store.add(Edge { kind: nsid("sh.tangled.repo.issue"), subject: SubjectRef::Did(did("did:plc:prop")), source: at(&format!("at://did:plc:prop/sh.tangled.repo.issue/i{i}")), sort_micros: (i as u64 + 1) * 1000, }); }
let asc: Vec<String> = (0..500) .map(|i| format!("at://did:plc:prop/sh.tangled.repo.issue/i{i}")) .collect(); let desc: Vec<String> = asc.iter().rev().cloned().collect(); let n = 500usize;
let mut offsets: Vec<u64> = (0..n as u64).step_by(50).collect(); offsets.extend([n as u64 - 1, n as u64]); offsets.sort(); offsets.dedup();
for o in offsets { for &(dir, ref_order, label) in &[(SortDir::Asc, &asc, "asc"), (SortDir::Desc, &desc, "desc")] { let first = store.list(&key, PageStart::Offset(PageOffset::from(o)), limit(25), dir); let got: Vec<String> = std::iter::successors(Some(first), |prev| { prev.next .map(|tok| store.list(&key, PageCursor::After(tok), limit(25), dir)) }) .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned())) .collect(); let expected = if (o as usize) < n { ref_order[o as usize..].to_vec() } else { vec![] }; assert_eq!(got, expected, "random n=500 {label} offset={o}"); } } }
#[test] fn block_store_matches_sorted_vec_oracle() { let mut blocks = BlockStore::from_sorted(Vec::new()); let mut oracle: Vec<BucketKey> = Vec::new(); let key_at = |i: usize| BucketKey::new(shuffled_micros(i), SourceId::from_raw(i as u32));
let mut inserted: Vec<usize> = (0..1500).collect(); inserted.sort_by_key(|i| shuffled_micros(*i ^ 0x5a5a)); for i in inserted { let key = key_at(i); assert!(blocks.insert(key), "insert {i}"); let pos = oracle.partition_point(|k| *k < key); oracle.insert(pos, key); } assert_eq!(blocks.len(), oracle.len());
// remove every third key in a different shuffled order let mut removed: Vec<BucketKey> = oracle.iter().step_by(3).copied().collect(); removed.sort_by_key(|k| shuffled_micros(k.source.index() as usize ^ 0x33)); for key in &removed { assert!(blocks.remove(key), "remove present key"); let pos = oracle.partition_point(|k| *k < *key); oracle.remove(pos); } assert_eq!(blocks.len(), oracle.len()); assert!(!blocks.remove(&removed[0]), "double remove must be false");
for rank in [ 0usize, 1, 255, 256, 300, 511, oracle.len() / 2, oracle.len() - 1, ] { assert_eq!( blocks.select(PageRank::new(rank)), Some(oracle[rank]), "select rank {rank}" ); } assert_eq!( blocks.select(PageRank::new(oracle.len())), None, "select past end" );
let asc: Vec<BucketKey> = blocks.directed(PageCursor::Start, SortDir::Asc).collect(); assert_eq!(asc, oracle, "asc full iteration"); let desc: Vec<BucketKey> = blocks.directed(PageCursor::Start, SortDir::Desc).collect(); assert_eq!( desc, oracle.iter().rev().copied().collect::<Vec<_>>(), "desc full iteration" );
// cursor-resume from a mid-block token in both directions let pivot = oracle[oracle.len() / 2]; let after_asc: Vec<BucketKey> = blocks .directed(PageCursor::After(pivot.token()), SortDir::Asc) .collect(); let want_asc: Vec<BucketKey> = oracle.iter().copied().filter(|k| *k > pivot).collect(); assert_eq!(after_asc, want_asc, "asc after pivot"); let after_desc: Vec<BucketKey> = blocks .directed(PageCursor::After(pivot.token()), SortDir::Desc) .collect(); let want_desc: Vec<BucketKey> = oracle .iter() .rev() .copied() .filter(|k| *k < pivot) .collect(); assert_eq!(after_desc, want_desc, "desc after pivot"); }
// Build three subject buckets with interleaved micros so the global time // order crosses buckets, and return the three keys. fn fill_three_subjects(store: &EdgeStore) -> Vec<EdgeKey> { [ ("alpha", "r0", 10u64), ("bravo", "r1", 50), ("charlie", "r2", 20), ("alpha", "r3", 40), ("bravo", "r4", 30), ("charlie", "r5", 60), ] .into_iter() .map(|(subj, rkey, micros)| { star_edge_at( at(&format!("at://did:plc:x/org.tangled.feed.star/{rkey}")), did(&format!("did:plc:{subj}")), micros, ) }) .for_each(|edge| store.add(edge)); ["alpha", "bravo", "charlie"] .iter() .map(|s| { EdgeKey::new( nsid("org.tangled.feed.star"), did_subj(&format!("did:plc:{s}")), ) }) .collect() }
#[test] fn list_multi_merges_subjects_newest_first() { let store = store(); let keys = fill_three_subjects(&store); let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc); let micros: Vec<u64> = page.items.iter().map(|it| it.sort_micros).collect(); assert_eq!(micros, vec![60, 50, 40, 30, 20, 10]); assert!(page.next.is_none()); }
#[test] fn list_multi_paginates_disjoint_across_buckets() { let store = store(); let keys = fill_three_subjects(&store); // limit=4 with 6 items across 3 buckets exercises the bounded top-N merge. let p1 = store.list_multi(&keys, PageCursor::Start, limit(4), SortDir::Desc); let m1: Vec<u64> = p1.items.iter().map(|it| it.sort_micros).collect(); assert_eq!(m1, vec![60, 50, 40, 30]); let tok = p1.next.expect("first page has more"); let p2 = store.list_multi(&keys, PageCursor::After(tok), limit(4), SortDir::Desc); let m2: Vec<u64> = p2.items.iter().map(|it| it.sort_micros).collect(); assert_eq!(m2, vec![20, 10]); assert!(p2.next.is_none()); }
#[test] fn list_multi_dedups_source_under_two_keys() { let store = store(); // Same record (source) indexed under two subject keys, same createdAt. let shared = at("at://did:plc:x/org.tangled.feed.star/shared"); store.add(star_edge_at(shared.clone(), did("did:plc:alpha"), 100)); store.add(star_edge_at(shared, did("did:plc:bravo"), 100)); let keys = vec![ EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:alpha")), EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:bravo")), ]; let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc); assert_eq!(page.items.len(), 1); assert!(page.next.is_none()); }
#[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, limit(7)), expected); assert_eq!( paginate_all(&store, &key, SortDir::Desc, limit(11)), expected.iter().rev().cloned().collect::<Vec<_>>() ); }
#[test] fn add_then_count() { let store = store(); let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
store.add(star_edge( at("at://did:plc:nel/org.tangled.feed.star/r1"), did("did:plc:abalone"), )); store.add(star_edge( at("at://did:plc:olaren/org.tangled.feed.star/r2"), did("did:plc:abalone"), )); store.add(star_edge( at("at://did:plc:nel/org.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("org.tangled.feed.star"), did_subj("did:plc:abalone")); let edge = star_edge( at("at://did:plc:nel/org.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("org.tangled.feed.star"), did_subj("did:plc:abalone")); let follow_key = EdgeKey::new(nsid("org.tangled.graph.follow"), did_subj("did:plc:lyna")); let source = "at://did:plc:nel/org.tangled.feed.star/r1";
store.add(Edge { kind: nsid("org.tangled.feed.star"), subject: did_subj("did:plc:abalone"), source: at(source), sort_micros: 0, }); store.add(Edge { kind: nsid("org.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/org.tangled.feed.star/r1"); let old_subject = did_subj("did:plc:abalone"); let new_subject = did_subj("did:plc:uni"); let kind = nsid("org.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); }
fn seed_tickets(store: &EdgeStore, repo: &str, micros: &[u64]) -> Vec<AtUri<DefaultStr>> { let kind = nsid(TICKET_ISSUE_KIND); micros .iter() .map(|micros| { let ticket = at(&format!("at://{repo}/org.tangled.track.ticket/{micros}")); store.upsert_source( &ticket, vec![Edge { kind: kind.clone(), subject: SubjectRef::Did(did(repo)), source: ticket.clone(), sort_micros: *micros, }], ); store.refresh_ticket_counts(&ticket, TicketStateKind::Open); ticket }) .collect() }
#[test] fn state_view_pages_default_open_by_rank_with_total() { let store = store(); let key = EdgeKey::new( nsid(TICKET_ISSUE_KIND), SubjectRef::Did(did("did:plc:limpet")), ); let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2, 3, 4, 5]);
let page = store.page_ticket_state( &key, TicketStateKind::Open, PageStart::Offset(PageOffset::new(2)), limit(2), SortDir::Asc, ); assert_eq!(page.total, Some(TotalCount::new(5)),); assert_eq!( page.items .iter() .map(|i| i.uri.as_ref()) .collect::<Vec<_>>(), vec![issues[2].as_ref(), issues[3].as_ref()], ); let follow = store.page_ticket_state( &key, TicketStateKind::Open, PageStart::Cursor(PageCursor::After(page.next.unwrap())), limit(2), SortDir::Asc, ); assert_eq!(follow.items.len(), 1); assert_eq!(follow.items[0].uri, issues[4]); }
#[test] fn state_view_transitions_repage_across_kinds() { let store = store(); let key = EdgeKey::new( nsid(TICKET_ISSUE_KIND), SubjectRef::Did(did("did:plc:limpet")), ); let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2]);
store.refresh_ticket_counts(&issues[0], TicketStateKind::Closed);
let open = store.page_ticket_state( &key, TicketStateKind::Open, PageStart::Offset(PageOffset::ZERO), limit(10), SortDir::Asc, ); assert_eq!(open.total, Some(TotalCount::new(1))); assert_eq!(open.items[0].uri, issues[1]); let closed = store.page_ticket_state( &key, TicketStateKind::Closed, PageStart::Offset(PageOffset::ZERO), limit(10), SortDir::Asc, ); assert_eq!(closed.total, Some(TotalCount::new(1))); assert_eq!(closed.items[0].uri, issues[0]);
store.refresh_ticket_counts(&issues[1], TicketStateKind::Completed); let completed = store.count_ticket_state(&key, TicketStateKind::Completed); assert_eq!(completed.count.get(), 1); let open = store.count_ticket_state(&key, TicketStateKind::Open); assert_eq!(open.count.get(), 0); }
#[test] fn state_view_offset_desc_matches_cursor_desc() { // views resolve offsets through SortedBlocks::resolve_start directly: // issues i1..i5 ascending page as i5 i4 i3 i2 i1 descending, offset 2 // starts at i3 let store = store(); let key = EdgeKey::new( nsid(TICKET_ISSUE_KIND), SubjectRef::Did(did("did:plc:limpet")), ); let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2, 3, 4, 5]);
let page = store.page_ticket_state( &key, TicketStateKind::Open, PageStart::Offset(PageOffset::new(2)), limit(2), SortDir::Desc, ); assert_eq!(page.total, Some(TotalCount::new(5))); assert_eq!( page.items .iter() .map(|i| i.uri.as_ref()) .collect::<Vec<_>>(), vec![issues[2].as_ref(), issues[1].as_ref()], "desc offset starts at the third-newest", );
let tail = store.page_ticket_state( &key, TicketStateKind::Open, PageCursor::After(page.next.unwrap()), limit(5), SortDir::Desc, ); assert_eq!( tail.items .iter() .map(|i| i.uri.as_ref()) .collect::<Vec<_>>(), vec![issues[0].as_ref()], "cursor continuation finishes the desc walk", ); assert!(tail.next.is_none()); }
#[test] fn record_index_mutations_move_materialized_state_counts() { let store = store(); let ticket = at("at://did:plc:limpet/org.tangled.track.ticket/1"); let nsid_ticket = nsid("org.tangled.track.ticket"); let repo = || SubjectRef::Did(did("did:plc:limpet")); let issues = EdgeKey::new(nsid(TICKET_ISSUE_KIND), repo()); let pulls = EdgeKey::new(nsid(TICKET_PULL_KIND), repo()); let upsert = |state: &str, target: &str| { let body = format!( r#"{{"$type":"org.tangled.track.ticket","title":{{"text":"bug"}},"state":"{state}",{target}"createdAt":"2026-05-01T00:00:00Z"}}"# ); let record = Record::from_json_bytes(&nsid_ticket, body.as_bytes()).unwrap(); let edges = record.extract_edges(&ticket).unwrap(); upsert_record_indexes(&store, &ClaimIndex::new(), &ticket, edges, &record); }; let count = |key, kind| store.count_ticket_state(key, kind).count.get(); use TicketStateKind::{Closed, Completed, Open};
upsert("closed", ""); assert_eq!((count(&issues, Open), count(&issues, Closed)), (0, 1)); upsert("open", ""); assert_eq!((count(&issues, Open), count(&issues, Closed)), (1, 0));
// gaining a target branch moves the ticket from issues to pulls upsert("completed", r#""targetBranch":"main","#); assert_eq!(count(&issues, Open), 0); assert_eq!(count(&pulls, Completed), 1);
delete_record_indexes(&store, &ClaimIndex::new(), &ticket, &nsid_ticket); assert_eq!(count(&pulls, Completed), 0); }
#[test] fn list_pages_in_sort_order() { let store = store(); let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone")); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/org.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("org.tangled.feed.star"), did_subj("did:plc:abalone")); (0..2).for_each(|i| { store.add(star_edge( at(&format!( "at://did:plc:{}/org.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("org.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("org.tangled.feed.star"), did_subj("did:plc:abalone")); let s1 = at("at://did:plc:nel/org.tangled.feed.star/r1"); let s2 = at("at://did:plc:nel/org.tangled.feed.star/r2"); let s3 = at("at://did:plc:olaren/org.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 packed_sort_micros_preserves_u64_ordering_and_cursor_values() { let values = [ 0, u32::MAX as u64, u32::MAX as u64 + 1, 1_730_000_000_000_000, u64::MAX, ]; let keys = values.map(|micros| BucketKey::new(micros, SourceId::from_raw(7))); assert!(keys.windows(2).all(|pair| pair[0] < pair[1])); keys.into_iter().zip(values).for_each(|(key, micros)| { assert_eq!(key.micros.get(), micros); assert_eq!(key.token(), PageToken::new(micros, 7)); }); }
#[test] fn author_index_separates_and_rejoins_history() { let author = AuthorId::from_raw(1); let older = BucketKey::new(10, SourceId::from_raw(10)); let middle = BucketKey::new(20, SourceId::from_raw(20)); let newest = BucketKey::new(30, SourceId::from_raw(30)); let mut index = AuthorIndex::new(RuntimeHasher::default());
index.insert(author, newest); assert_eq!(index.newest[&author], newest); assert!(index.histories.is_empty());
index.insert(author, older); index.insert(author, middle); assert_eq!(index.newest[&author], newest); assert_eq!(index.histories[&author].len(), 3);
index.remove(author, &older); assert_eq!(index.histories[&author].len(), 2); assert_eq!(index.newest[&author], newest);
index.remove(author, &newest); assert!(index.histories.is_empty()); assert_eq!(index.newest[&author], middle); assert_eq!(index.representatives.last(), Some(middle));
index.remove(author, &middle); assert!(index.newest.is_empty()); assert!(index.representatives.last().is_none()); }
#[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("org.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..4).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/org.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_descending_returns_newest_first() { let store = store(); let subject_did = did("did:plc:limpet"); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/org.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("org.tangled.feed.star"), SubjectRef::Did(subject_did.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/org.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 non_did_source_round_trips_via_raw_fallback() { let store = store(); let subject = did("did:plc:limpet"); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); let did_source = at("at://did:plc:nel/org.tangled.feed.star/r1"); let handle_source = at("at://witchcraft.systems/org.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("org.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); let star_src = at("at://did:plc:nel/org.tangled.feed.star/aaa"); let issue_src = at("at://did:plc:nel/sh.tangled.repo.issue/bbb"); store.add(Edge { kind: nsid("org.tangled.feed.star"), subject: SubjectRef::Did(subject.clone()), source: star_src.clone(), sort_micros: 1, }); store.add(Edge { kind: nsid("org.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 most_followed_ranks_distinct_current_followers() { let store = store(); let alpha = did("did:plc:alpha"); let beta = did("did:plc:beta"); let gamma = did("did:plc:gamma"); [ ("did:plc:nel", "a1", alpha.clone()), ("did:plc:nel", "a2", alpha.clone()), ("did:plc:olaren", "a3", alpha.clone()), ("did:plc:nel", "b1", beta.clone()), ("did:plc:olaren", "b2", beta.clone()), ("did:plc:teq", "b3", beta.clone()), ("did:plc:nel", "g1", gamma.clone()), ("did:plc:olaren", "g2", gamma.clone()), ] .into_iter() .enumerate() .for_each(|(index, (author, rkey, subject))| { store.add(follow_edge_at( at(&format!("at://{author}/org.tangled.graph.follow/{rkey}")), subject, index as u64, )); });
assert_eq!(store.most_followed(3), vec![beta, alpha, gamma]); }
#[test] fn top_starred_since_counts_distinct_recent_starrers() { let store = store(); let recent = did("did:plc:recent"); let one = did("did:plc:one"); let old = did("did:plc:old"); [ ("did:plc:nel", "r1", recent.clone(), 100), ("did:plc:nel", "r2", recent.clone(), 110), ("did:plc:olaren", "r3", recent.clone(), 120), ("did:plc:teq", "o1", one.clone(), 130), ("did:plc:nel", "x1", old.clone(), 1), ("did:plc:olaren", "x2", old.clone(), 2), ("did:plc:teq", "x3", old.clone(), 3), ] .into_iter() .for_each(|(author, rkey, subject, micros)| { store.add(star_edge_at( at(&format!("at://{author}/org.tangled.feed.star/{rkey}")), subject, micros, )); });
assert_eq!(store.top_starred_since(100, 5), vec![recent, one]); assert_eq!(store.top_starred_since(100, 1).len(), 1); }
#[test] fn author_refs_spill_preserves_distinct_count() { let store = store(); let subject = did("did:plc:scallop"); let key = EdgeKey::new( nsid("org.tangled.feed.star"), SubjectRef::Did(subject.clone()), ); (0..5).for_each(|i| { store.add(star_edge_at( at(&format!( "at://did:plc:{}/org.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/org.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/org.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/org.tangled.feed.star/r99")); assert_eq!(store.count_distinct_authors(&key), 4, "nel fully removed"); }
#[test] fn source_collections_set_and_clear() { let store = store(); let source = at("at://did:plc:bob/org.tangled.feed.subscription/r1"); let edge = Edge { kind: nsid("org.tangled.feed.subscription"), subject: did_subj("did:plc:repo"), source: source.clone(), sort_micros: 1, }; store.add(edge);
// Initially no filter assert_eq!(store.source_collections_for(&source), None);
// Set a filter store.set_source_collections( &source, Some(vec![SmolStr::new_static("sh.tangled.repo.issue")]), ); let stored = store.source_collections_for(&source).unwrap(); assert_eq!(stored, vec![SmolStr::new_static("sh.tangled.repo.issue")]);
// Update to empty = unrestricted store.set_source_collections(&source, Some(vec![])); assert_eq!(store.source_collections_for(&source), None);
// Set a different filter store.set_source_collections( &source, Some(vec![SmolStr::new_static("sh.tangled.repo.pull")]), ); let stored = store.source_collections_for(&source).unwrap(); assert_eq!(stored, vec![SmolStr::new_static("sh.tangled.repo.pull")]);
// Set to None = unrestricted store.set_source_collections(&source, None); assert_eq!(store.source_collections_for(&source), None); }
#[test] fn remove_source_cleans_up_collection_filter() { let store = store(); let source = at("at://did:plc:bob/org.tangled.feed.subscription/r1"); let edge = Edge { kind: nsid("org.tangled.feed.subscription"), subject: did_subj("did:plc:repo"), source: source.clone(), sort_micros: 1, }; store.add(edge); store.set_source_collections( &source, Some(vec![SmolStr::new_static("sh.tangled.repo.issue")]), ); assert!(store.source_collections_for(&source).is_some());
store.remove_source(&source); assert_eq!(store.source_collections_for(&source), None); }}