Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388mod decode;mod hydrant;mod identity;mod normalize;
pub use decode::{decode_record, normalize_record_fields, synthesize_created_at};pub use hydrant::{HydrantClient, HydrantError, StreamChunk, StreamManifest};pub use identity::{ IdentityResolveError, IdentityResolver, IdentityResolverStatsSnapshot, MiniDoc,};pub use normalize::NormalizeRepoRefs;
use std::sync::Arc;use std::sync::atomic::{AtomicU64, Ordering};use std::time::Duration;
use tokio::time::Instant;
use bobbin_runtime::{Clock, RuntimeHasher};use bobbin_slingshot_client::{SlingshotClient, SlingshotError};use bobbin_types::ids::{RepoIdent, nsid_static};use jacquard_common::DefaultStr;use jacquard_common::types::datetime::Datetime;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::recordkey::Rkey;use scc::{HashMap as SccMap, HashSet as SccSet};use tokio::sync::OnceCell;use tracing::warn;
const REPO_COLLECTION: &str = "sh.tangled.repo";const TRANSIENT_TTL: Duration = Duration::from_secs(60);const RESOLVE_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Clone, Debug, Eq, PartialEq)]pub enum Resolution { Mapped(Did<DefaultStr>), NoRepoDid, Unresolvable,}
#[derive(Clone, Debug, Eq, PartialEq)]enum AuthoritativeResolution { Mapped(Did<DefaultStr>), NoRepoDid,}
impl AuthoritativeResolution { fn from_repo_did(repo_did: Option<Did<DefaultStr>>) -> Self { match repo_did { Some(did) => Self::Mapped(did), None => Self::NoRepoDid, } }}
#[derive(Clone, Debug, Eq, PartialEq)]struct RepoDidHolder { ident: RepoIdent, created_at: Datetime,}
impl RepoDidHolder { // a backfill replays a repo in rkey order, so the stamp decides and not arrival fn yields_to(&self, incoming: &Self) -> bool { self.ident == incoming.ident || incoming.created_at >= self.created_at }}
#[derive(Clone, Debug, Eq, PartialEq)]pub enum RepoClaim { Current { displaced: Option<RepoIdent> }, Superseded { by: RepoIdent },}
#[derive(Clone, Debug, Eq, PartialEq)]enum CacheEntry { Authoritative(AuthoritativeResolution), Provisional(Resolution), Transient { expires_at: Instant },}
impl CacheEntry { fn into_resolution(self) -> Resolution { match self { Self::Authoritative(AuthoritativeResolution::Mapped(did)) => Resolution::Mapped(did), Self::Authoritative(AuthoritativeResolution::NoRepoDid) => Resolution::NoRepoDid, Self::Provisional(r) => r, Self::Transient { .. } => Resolution::Unresolvable, } }
fn is_expired_transient(&self, now: Instant) -> bool { matches!(self, Self::Transient { expires_at } if *expires_at <= now) }}
#[derive(Default)]pub struct ResolverStats { hits: AtomicU64, misses_mapped: AtomicU64, misses_no_repo_did: AtomicU64, misses_unresolvable: AtomicU64, misses_transient: AtomicU64, misses_no_client: AtomicU64, miss_latency_micros_sum: AtomicU64, miss_latency_micros_max: AtomicU64,}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]pub struct ResolverStatsSnapshot { pub hits: u64, pub misses_mapped: u64, pub misses_no_repo_did: u64, pub misses_unresolvable: u64, pub misses_transient: u64, pub misses_no_client: u64, pub miss_latency_micros_sum: u64, pub miss_latency_micros_max: u64,}
impl ResolverStatsSnapshot { pub fn miss_count(&self) -> u64 { self.misses_mapped + self.misses_no_repo_did + self.misses_unresolvable + self.misses_transient + self.misses_no_client }
pub fn total(&self) -> u64 { self.hits + self.miss_count() }
pub fn miss_latency_micros_avg(&self) -> Option<u64> { let misses = self.miss_count() - self.misses_no_client; (misses > 0).then(|| self.miss_latency_micros_sum / misses) }}
#[derive(Clone, Copy)]enum MissKind { Mapped, NoRepoDid, Unresolvable, Transient, NoClient,}
impl ResolverStats { fn record_hit(&self) { self.hits.fetch_add(1, Ordering::Relaxed); }
fn record_miss(&self, kind: MissKind, latency: Option<Duration>) { let counter = match kind { MissKind::Mapped => &self.misses_mapped, MissKind::NoRepoDid => &self.misses_no_repo_did, MissKind::Unresolvable => &self.misses_unresolvable, MissKind::Transient => &self.misses_transient, MissKind::NoClient => &self.misses_no_client, }; counter.fetch_add(1, Ordering::Relaxed); if let Some(latency) = latency { let micros = u64::try_from(latency.as_micros()).unwrap_or(u64::MAX); self.miss_latency_micros_sum .fetch_add(micros, Ordering::Relaxed); self.miss_latency_micros_max .fetch_max(micros, Ordering::Relaxed); } }
pub fn snapshot(&self) -> ResolverStatsSnapshot { ResolverStatsSnapshot { hits: self.hits.load(Ordering::Relaxed), misses_mapped: self.misses_mapped.load(Ordering::Relaxed), misses_no_repo_did: self.misses_no_repo_did.load(Ordering::Relaxed), misses_unresolvable: self.misses_unresolvable.load(Ordering::Relaxed), misses_transient: self.misses_transient.load(Ordering::Relaxed), misses_no_client: self.misses_no_client.load(Ordering::Relaxed), miss_latency_micros_sum: self.miss_latency_micros_sum.load(Ordering::Relaxed), miss_latency_micros_max: self.miss_latency_micros_max.load(Ordering::Relaxed), } }}
pub(crate) struct SlingshotProbe { pub(crate) client: SlingshotClient, pub(crate) clock: Arc<dyn Clock>,}
pub struct RepoIdResolver { cache: SccMap<RepoIdent, CacheEntry, RuntimeHasher>, by_repo_did: SccMap<Did<DefaultStr>, RepoDidHolder, RuntimeHasher>, by_rkey: SccSet<RepoIdent, RuntimeHasher>, by_name: SccMap<(Did<DefaultStr>, DefaultStr), RepoIdent, RuntimeHasher>, in_flight: SccMap<RepoIdent, Arc<OnceCell<Resolution>>, RuntimeHasher>, probe: Option<SlingshotProbe>, stats: ResolverStats,}
impl RepoIdResolver { pub fn with_slingshot( client: SlingshotClient, clock: Arc<dyn Clock>, hasher: RuntimeHasher, ) -> Self { Self { cache: SccMap::with_hasher(hasher.clone()), by_repo_did: SccMap::with_hasher(hasher.clone()), by_rkey: SccSet::with_hasher(hasher.clone()), by_name: SccMap::with_hasher(hasher.clone()), in_flight: SccMap::with_hasher(hasher), probe: Some(SlingshotProbe { client, clock }), stats: ResolverStats::default(), } }
pub fn detached(hasher: RuntimeHasher) -> Self { Self { cache: SccMap::with_hasher(hasher.clone()), by_repo_did: SccMap::with_hasher(hasher.clone()), by_rkey: SccSet::with_hasher(hasher.clone()), by_name: SccMap::with_hasher(hasher.clone()), in_flight: SccMap::with_hasher(hasher), probe: None, stats: ResolverStats::default(), } }
pub fn stats(&self) -> ResolverStatsSnapshot { self.stats.snapshot() }
pub async fn cached_resolution( &self, owner: &Did<DefaultStr>, rkey: &Rkey<DefaultStr>, ) -> Option<Resolution> { let key = RepoIdent::new(owner.clone(), rkey.clone()); let entry = self.cache.get_async(&key).await?; let now = self.probe.as_ref().map(|p| p.clock.now_instant()); if let Some(now) = now && entry.get().is_expired_transient(now) { return None; } Some(entry.get().clone().into_resolution()) }
pub async fn lookup_by_repo_did(&self, repo_did: &Did<DefaultStr>) -> Option<RepoIdent> { self.by_repo_did .get_async(repo_did) .await .map(|e| e.get().ident.clone()) }
pub async fn lookup_by_name(&self, owner: &Did<DefaultStr>, name: &str) -> Option<RepoIdent> { // rkey addressing wins so a record can't hijack another repo's rkey via its name if let Ok(rkey) = Rkey::new_owned(name) { let ident = RepoIdent::new(owner.clone(), rkey); if self.by_rkey.contains_async(&ident).await { return Some(self.current_record(ident).await); } } let named = self .by_name .get_async(&(owner.clone(), DefaultStr::from(name))) .await .map(|e| e.get().clone())?; Some(self.current_record(named).await) }
// the names a rename leaves behind still have to answer with the repo as it stands async fn current_record(&self, ident: RepoIdent) -> RepoIdent { let Some(Resolution::Mapped(repo_did)) = self.cached_resolution(&ident.owner, &ident.rkey).await else { return ident; }; self.lookup_by_repo_did(&repo_did).await.unwrap_or(ident) }
pub async fn observe_rkey(&self, owner: Did<DefaultStr>, rkey: Rkey<DefaultStr>) { let _ = self.by_rkey.insert_async(RepoIdent::new(owner, rkey)).await; }
pub async fn observe( &self, owner: Did<DefaultStr>, rkey: Rkey<DefaultStr>, repo_did: Option<Did<DefaultStr>>, name: Option<DefaultStr>, created_at: &Datetime, ) -> RepoClaim { let ident = RepoIdent::new(owner, rkey); self.observe_rkey(ident.owner.clone(), ident.rkey.clone()) .await; if let Some(name) = name { let _ = self .by_name .insert_async((ident.owner.clone(), name), ident.clone()) .await; } let entry = CacheEntry::Authoritative(AuthoritativeResolution::from_repo_did(repo_did.clone())); self.cache .entry_async(ident.clone()) .await .and_modify(|existing| *existing = entry.clone()) .or_insert(entry);
let Some(repo_did) = repo_did else { return RepoClaim::Current { displaced: None }; }; let incoming = RepoDidHolder { ident, created_at: created_at.clone(), }; let mut outcome = RepoClaim::Current { displaced: None }; self.by_repo_did .entry_async(repo_did) .await .and_modify(|held| { let yields = held.yields_to(&incoming); outcome = yields .then(|| RepoClaim::Current { displaced: (held.ident != incoming.ident).then(|| held.ident.clone()), }) .unwrap_or_else(|| RepoClaim::Superseded { by: held.ident.clone(), }); if yields { *held = incoming.clone(); } }) .or_insert(incoming); outcome }
pub async fn forget(&self, owner: &Did<DefaultStr>, rkey: &Rkey<DefaultStr>) { let ident = RepoIdent::new(owner.clone(), rkey.clone()); let prior_resolution = self .cache .remove_async(&ident) .await .map(|(_, entry)| entry.into_resolution()); if let Some(Resolution::Mapped(repo_did)) = prior_resolution { self.by_repo_did .remove_if_async(&repo_did, |held| held.ident == ident) .await; } self.by_rkey.remove_async(&ident).await; }
async fn fill_provisional(&self, key: RepoIdent, resolution: Resolution) { let entry = CacheEntry::Provisional(resolution); self.cache .entry_async(key) .await .and_modify(|existing| { if matches!(existing, CacheEntry::Authoritative(_)) { return; } *existing = entry.clone(); }) .or_insert(entry); }
async fn fill_transient(&self, key: RepoIdent, expires_at: Instant) { let entry = CacheEntry::Transient { expires_at }; self.cache .entry_async(key) .await .and_modify(|existing| { if matches!(existing, CacheEntry::Authoritative(_)) { return; } *existing = entry.clone(); }) .or_insert(entry); }
pub async fn resolve(&self, owner: &Did<DefaultStr>, rkey: &Rkey<DefaultStr>) -> Resolution { let key = RepoIdent::new(owner.clone(), rkey.clone());
let Some(probe) = self.probe.as_ref() else { if let Some(entry) = self.cache.get_async(&key).await { self.stats.record_hit(); return entry.get().clone().into_resolution(); } self.stats.record_miss(MissKind::NoClient, None); return Resolution::Unresolvable; };
let now = probe.clock.now_instant(); if let Some(entry) = self.cache.get_async(&key).await && !entry.get().is_expired_transient(now) { self.stats.record_hit(); return entry.get().clone().into_resolution(); }
let cell: Arc<OnceCell<Resolution>> = self .in_flight .entry_async(key.clone()) .await .or_insert_with(|| Arc::new(OnceCell::new())) .get() .clone();
let result = cell .get_or_init(|| async { match tokio::time::timeout(RESOLVE_TIMEOUT, self.fetch_repo_did(owner, rkey, &key)) .await { Ok(resolution) => resolution, Err(_elapsed) => { warn!( owner = owner.as_ref(), rkey = rkey.as_ref(), "repo resolution hung, caching as unresolvable under short TTL", ); let probe = self .probe .as_ref() .expect("fetch_repo_did is only called when a probe is present"); self.stats .record_miss(MissKind::Transient, Some(RESOLVE_TIMEOUT)); let expires_at = probe.clock.now_instant() + TRANSIENT_TTL; self.fill_transient(key.clone(), expires_at).await; Resolution::Unresolvable } } }) .await .clone();
self.in_flight.remove_async(&key).await;
result }
async fn fetch_repo_did( &self, owner: &Did<DefaultStr>, rkey: &Rkey<DefaultStr>, key: &RepoIdent, ) -> Resolution { let probe = self .probe .as_ref() .expect("fetch_repo_did is only called when a probe is present"); let started = probe.clock.now_instant(); let nsid: Nsid<DefaultStr> = nsid_static(REPO_COLLECTION); let provisional = match probe.client.get_record(owner, &nsid, rkey).await { Ok(body) => match repo_did_from_body(&body.value) { Ok(Some(did)) => Resolution::Mapped(did), Ok(None) => Resolution::NoRepoDid, Err(e) => { warn!( error = ?e, owner = owner.as_ref(), rkey = rkey.as_ref(), "slingshot returned unparseable repo body, caching as unresolvable", ); Resolution::Unresolvable } }, Err(SlingshotError::NotFound) => { warn!( owner = owner.as_ref(), rkey = rkey.as_ref(), "no repo record on slingshot, caching as unresolvable", ); Resolution::Unresolvable } Err(ref e) if is_garbage_response(e) => { warn!( error = ?e, owner = owner.as_ref(), rkey = rkey.as_ref(), "slingshot returned malformed response, caching as unresolvable", ); Resolution::Unresolvable } Err(e) => { warn!( error = ?e, owner = owner.as_ref(), rkey = rkey.as_ref(), "caching transient slingshot failure for repoDID lookup under short TTL", ); let elapsed = probe.clock.now_instant().duration_since(started); self.stats.record_miss(MissKind::Transient, Some(elapsed)); let expires_at = probe.clock.now_instant() + TRANSIENT_TTL; self.fill_transient(key.clone(), expires_at).await; return Resolution::Unresolvable; } }; let elapsed = probe.clock.now_instant().duration_since(started); let kind = match &provisional { Resolution::Mapped(_) => MissKind::Mapped, Resolution::NoRepoDid => MissKind::NoRepoDid, Resolution::Unresolvable => MissKind::Unresolvable, }; self.stats.record_miss(kind, Some(elapsed)); self.fill_provisional(key.clone(), provisional.clone()) .await; provisional }}
fn is_garbage_response(err: &SlingshotError) -> bool { matches!( err, SlingshotError::Decode(_) | SlingshotError::MissingField(_) | SlingshotError::InvalidAtUri(_) | SlingshotError::InvalidCid(_) | SlingshotError::UriMismatch { .. }, )}
// legacy at-uri refs still point at sh.tangled.repo, and only its repoDid matters herefn repo_did_from_body(body: &[u8]) -> Result<Option<Did<DefaultStr>>, serde_json::Error> { #[derive(serde::Deserialize)] struct RepoDid { #[serde(rename = "repoDid")] repo_did: Option<Did<DefaultStr>>, } serde_json::from_slice::<RepoDid>(body).map(|r| r.repo_did)}
#[cfg(test)]mod tests { use super::*; use bobbin_runtime::SystemClock; use jacquard_common::types::did::Did; use jacquard_common::types::recordkey::Rkey;
fn did(s: &str) -> Did<DefaultStr> { Did::new_owned(s).unwrap() }
fn rkey(s: &str) -> Rkey<DefaultStr> { Rkey::new_owned(s).unwrap() }
fn test_clock() -> Arc<dyn Clock> { Arc::new(SystemClock::new()) }
fn made(day: u32) -> Datetime { Datetime::raw_str(format!("2026-05-{day:02}T00:00:00Z")) }
#[tokio::test] async fn observation_returns_prior_ident_when_repo_did_moves() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let claim = resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), Some(did("did:plc:clam")), None, &made(1), ) .await; assert_eq!( claim, RepoClaim::Current { displaced: None }, "first observation has no prior" );
let claim = resolver .observe( did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam")), None, &made(2), ) .await; assert_eq!( claim, RepoClaim::Current { displaced: Some(RepoIdent::new(did("did:plc:nel"), rkey("3liuighjy2h22"))), }, "same repoDID at a new (owner, rkey) returns the prior ident so callers can evict the stale at-uri", );
let claim = resolver .observe( did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam")), None, &made(2), ) .await; assert_eq!( claim, RepoClaim::Current { displaced: None }, "re-observing the same ident is a no-op" ); }
#[tokio::test] async fn a_record_older_than_the_holder_cannot_take_the_repo_did() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:oppi"); let repo_did = did("did:plc:clam"); resolver .observe( owner.clone(), rkey("stinkpot"), Some(repo_did.clone()), None, &made(24), ) .await; let claim = resolver .observe( owner.clone(), rkey("tortu"), Some(repo_did.clone()), None, &made(20), ) .await;
let stinkpot = RepoIdent::new(owner.clone(), rkey("stinkpot")); assert_eq!( claim, RepoClaim::Superseded { by: stinkpot.clone() }, "the older record is the alias a rename left behind", ); assert_eq!( resolver.lookup_by_repo_did(&repo_did).await, Some(stinkpot.clone()), ); assert_eq!( resolver.lookup_by_name(&owner, "tortu").await, Some(stinkpot), "the name a rename left behind has to answer with the repo as it stands now", ); }
#[tokio::test] async fn observation_without_repo_did_does_not_track_reverse() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let claim = resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), None, None, &made(1), ) .await; assert_eq!(claim, RepoClaim::Current { displaced: None }); }
struct HangingHttp;
impl bobbin_runtime::HttpTransport for HangingHttp { fn execute( &self, _request: bobbin_runtime::HttpRequest, ) -> bobbin_runtime::HttpResponseFuture { Box::pin(std::future::pending()) } }
#[tokio::test(start_paused = true)] async fn resolve_returns_even_if_transport_doesnt() { let client = SlingshotClient::new( url::Url::parse("http://127.0.0.1:9").unwrap(), Arc::new(HangingHttp), ) .unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let resolution = tokio::time::timeout( Duration::from_secs(60), resolver.resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")), ) .await .expect("resolve must return even when transport doesn't"); assert_eq!(resolution, Resolution::Unresolvable); assert_eq!(resolver.stats().misses_transient, 1);
let again = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; assert_eq!(again, Resolution::Unresolvable); assert_eq!( resolver.stats().hits, 1, "transient entry must stop retries during TTL such that wedgy upstream can't stall every frame refing to ident" ); }
#[tokio::test] async fn forget_clears_reverse_only_when_still_owned() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), Some(did("did:plc:clam")), None, &made(1), ) .await; resolver .observe( did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam")), None, &made(1), ) .await;
resolver .forget(&did("did:plc:nel"), &rkey("3liuighjy2h22")) .await;
let claim = resolver .observe( did("did:plc:nel"), rkey("core-renamed"), Some(did("did:plc:clam")), None, &made(1), ) .await; assert_eq!( claim, RepoClaim::Current { displaced: Some(RepoIdent::new(did("did:plc:nel"), rkey("core"))), }, "stale at-uri's forget must not displace the live owner of did:plc:clam", ); }
#[tokio::test] async fn lookup_by_name_finds_observed_rkey() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe_rkey(did("did:plc:nel"), rkey("3liuighjy2h22")) .await; let got = resolver .lookup_by_name(&did("did:plc:nel"), "3liuighjy2h22") .await; assert_eq!( got, Some(RepoIdent::new(did("did:plc:nel"), rkey("3liuighjy2h22"))), ); }
#[tokio::test] async fn lookup_by_name_is_scoped_to_the_owner() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe_rkey(did("did:plc:nel"), rkey("3liuighjy2h22")) .await; let got = resolver .lookup_by_name(&did("did:plc:olaren"), "3liuighjy2h22") .await; assert_eq!(got, None, "one owner's rkey must not answer for another's"); }
#[tokio::test] async fn lookup_by_name_rejects_non_rkey() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); resolver .observe_rkey(owner.clone(), rkey("3liuighjy2h22")) .await;
assert_eq!(resolver.lookup_by_name(&owner, "my repo").await, None); }
#[tokio::test] async fn lookup_by_name_finds_observed_record_name() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), Some(did("did:plc:clam")), Some(DefaultStr::from("ark")), &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:nel"), "ark").await; assert_eq!( got, Some(RepoIdent::new(did("did:plc:nel"), rkey("3liuighjy2h22"))), ); }
#[tokio::test] async fn lookup_by_name_record_name_is_scoped_to_the_owner() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), None, Some(DefaultStr::from("ark")), &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:olaren"), "ark").await; assert_eq!( got, None, "one owner's repo name must not answer for another's" ); }
#[tokio::test] async fn lookup_by_name_prefers_rkey_over_claimed_name() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe(did("did:plc:nel"), rkey("core"), None, None, &made(1)) .await; resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), None, Some(DefaultStr::from("core")), &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:nel"), "core").await; assert_eq!( got, Some(RepoIdent::new(did("did:plc:nel"), rkey("core"))), "a record naming itself after a live rkey must not shadow it", ); }
#[tokio::test] async fn forget_clears_the_rkey() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); resolver .observe_rkey(owner.clone(), rkey("3liuighjy2h22")) .await; resolver.forget(&owner, &rkey("3liuighjy2h22")).await; assert_eq!(resolver.lookup_by_name(&owner, "3liuighjy2h22").await, None); }
#[tokio::test] async fn observation_with_repo_did_resolves_mapped() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:clam")), None, &made(1), ) .await; let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; assert_eq!(got, Resolution::Mapped(did("did:plc:clam"))); }
#[tokio::test] async fn observation_without_repo_did_resolves_no_repo_did() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), None, None, &made(1), ) .await; let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; assert_eq!( got, Resolution::NoRepoDid, "observed but empty repoDID is a definitive answer not a lookup failure", ); }
#[tokio::test] async fn cache_miss_without_client_is_unresolvable() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; assert_eq!(got, Resolution::Unresolvable); }
#[tokio::test] async fn lookup_by_repo_did_finds_observed_ident() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:limpet")), None, &made(1), ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; assert_eq!( got, Some(RepoIdent::new(did("did:plc:nel"), rkey("abcabcabcabcz"))), ); }
#[tokio::test] async fn lookup_by_repo_did_misses_when_unobserved() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; assert_eq!(got, None); }
#[tokio::test] async fn lookup_by_repo_did_misses_when_repo_did_was_none() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), None, None, &made(1), ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; assert_eq!(got, None); }
#[tokio::test] async fn lookup_by_repo_did_follows_move_to_new_ident() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:limpet")), None, &made(1), ) .await; resolver .observe( did("did:plc:olaren"), rkey("xyzxyzxyzxyzx"), Some(did("did:plc:limpet")), None, &made(1), ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; assert_eq!( got, Some(RepoIdent::new(did("did:plc:olaren"), rkey("xyzxyzxyzxyzx"))), ); }
#[tokio::test] async fn observation_overwrites_prior_value() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:clam")), None, &made(1), ) .await; resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:uni")), None, &made(1), ) .await; let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; assert_eq!(got, Resolution::Mapped(did("did:plc:uni"))); }
#[tokio::test] async fn fill_provisional_does_not_downgrade_authoritative_mapped() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver .observe( owner.clone(), key.clone(), Some(did("did:plc:clam")), None, &made(1), ) .await; resolver .fill_provisional( RepoIdent::new(owner.clone(), key.clone()), Resolution::Unresolvable, ) .await; let got = resolver.resolve(&owner, &key).await; assert_eq!( got, Resolution::Mapped(did("did:plc:clam")), "firehose-observed mapping must outrank provisional slingshot info", ); }
#[tokio::test] async fn fill_provisional_does_not_downgrade_authoritative_no_repo_did() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver .observe(owner.clone(), key.clone(), None, None, &made(1)) .await; resolver .fill_provisional( RepoIdent::new(owner.clone(), key.clone()), Resolution::Mapped(did("did:plc:clam")), ) .await; let got = resolver.resolve(&owner, &key).await; assert_eq!( got, Resolution::NoRepoDid, "an authoritative empty observation must outrank provisional slingshot info even when slingshot disagrees", ); }
#[tokio::test] async fn slingshot_404_caches_as_unresolvable() { let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(404)) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let first = resolver.resolve(&owner, &key).await; let second = resolver.resolve(&owner, &key).await; assert_eq!(first, Resolution::Unresolvable); assert_eq!(second, Resolution::Unresolvable); }
#[tokio::test] async fn slingshot_malformed_envelope_caches_as_unresolvable() { let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with( wiremock::ResponseTemplate::new(200) .insert_header("content-type", "application/json") .set_body_string("not json"), ) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let first = resolver.resolve(&owner, &key).await; let second = resolver.resolve(&owner, &key).await; assert_eq!(first, Resolution::Unresolvable); assert_eq!( second, Resolution::Unresolvable, "garbage envelopes are stable across retries, so caching avoids hammering slingshot", ); }
#[tokio::test] async fn slingshot_uri_mismatch_caches_as_unresolvable() { let server = wiremock::MockServer::start().await; let body = serde_json::json!({ "uri": "at://did:plc:limpet/sh.tangled.repo/elsewhere", "cid": "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i", "value": {"$type": "sh.tangled.repo", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z"} }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body)) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let first = resolver.resolve(&owner, &key).await; let second = resolver.resolve(&owner, &key).await; assert_eq!(first, Resolution::Unresolvable); assert_eq!(second, Resolution::Unresolvable); }
#[tokio::test] async fn slingshot_legacy_repo_body_resolves_no_repo_did_not_unresolvable() { let server = wiremock::MockServer::start().await; let body = serde_json::json!({ "uri": "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", "cid": "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i", "value": { "$type": "sh.tangled.repo", "addedAt": "2025-03-07T21:47:53Z", "knot": "knot1.tangled.sh", "name": "scallop", "owner": "did:plc:nel", }, }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body)) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let got = resolver.resolve(&owner, &key).await; assert_eq!( got, Resolution::NoRepoDid, "legacy repo wires without a repo_did parse via legacy upgrade and resolve as NoRepoDid, not Unresolvable", ); }
#[tokio::test] async fn slingshot_transport_error_caches_with_short_ttl() { let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(503)) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let first = resolver.resolve(&owner, &key).await; let second = resolver.resolve(&owner, &key).await; assert_eq!(first, Resolution::Unresolvable); assert_eq!( second, Resolution::Unresolvable, "transient TTL must suppress immediate re-hammering of a sick upstream", ); }
#[tokio::test] async fn slingshot_transient_recorded_separately_from_unresolvable() { let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(503)) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver.resolve(&owner, &key).await; resolver.resolve(&owner, &key).await;
let snap = resolver.stats(); assert_eq!( snap.misses_transient, 1, "second resolve must hit the short-TTL cache instead of re-firing the transient miss", ); assert_eq!(snap.hits, 1, "second call hits cached transient entry"); assert_eq!( snap.misses_unresolvable, 0, "canonical unresolvable counter is reserved for cached terminal answers", ); assert!( snap.miss_latency_micros_sum > 0, "transient misses still have latency contributions", ); assert_eq!(snap.miss_count(), 1); }
#[tokio::test] async fn slingshot_in_flight_requests_coalesce() { let server = wiremock::MockServer::start().await; let body = serde_json::json!({ "uri": "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", "cid": "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i", "value": {"$type": "sh.tangled.repo", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z", "repoDid": "did:plc:limpet"} }); wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with( wiremock::ResponseTemplate::new(200) .set_body_json(body) .set_delay(Duration::from_millis(200)), ) .expect(1) .mount(&server) .await;
let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = Arc::new(RepoIdResolver::with_slingshot( client, test_clock(), RuntimeHasher::default(), ));
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); let r0 = resolver.clone(); let r1 = resolver.clone(); let r2 = resolver.clone(); let o0 = owner.clone(); let o1 = owner.clone(); let o2 = owner.clone(); let k0 = key.clone(); let k1 = key.clone(); let k2 = key.clone(); let (a, b, c) = tokio::join!( tokio::spawn(async move { r0.resolve(&o0, &k0).await }), tokio::spawn(async move { r1.resolve(&o1, &k1).await }), tokio::spawn(async move { r2.resolve(&o2, &k2).await }), ); let expected = Resolution::Mapped(did("did:plc:limpet")); assert_eq!(a.unwrap(), expected); assert_eq!(b.unwrap(), expected); assert_eq!(c.unwrap(), expected);
let snap = resolver.stats(); assert_eq!( snap.misses_mapped, 1, "only the winning task pays the slingshot RTT", ); }
#[tokio::test] async fn stats_count_hits_misses_and_latency() { let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with(wiremock::ResponseTemplate::new(404)) .mount(&server) .await; let client = SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver.resolve(&owner, &key).await; resolver.resolve(&owner, &key).await;
let snap = resolver.stats(); assert_eq!( snap.misses_unresolvable, 1, "first call is the slingshot miss" ); assert_eq!(snap.hits, 1, "second call hits the unresolvable cache"); assert_eq!(snap.miss_count(), 1); assert_eq!(snap.total(), 2); assert!( snap.miss_latency_micros_sum > 0, "latency recorded for slingshot miss" ); assert!(snap.miss_latency_micros_avg().unwrap() > 0); }
#[tokio::test] async fn stats_no_client_miss_recorded_without_latency() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; let snap = resolver.stats(); assert_eq!(snap.misses_no_client, 1); assert_eq!(snap.miss_latency_micros_sum, 0); assert_eq!(snap.miss_latency_micros_avg(), None); }
#[tokio::test] async fn firehose_observe_can_demote_provisional() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver .fill_provisional( RepoIdent::new(owner.clone(), key.clone()), Resolution::Mapped(did("did:plc:clam")), ) .await; resolver .observe(owner.clone(), key.clone(), None, None, &made(1)) .await; let got = resolver.resolve(&owner, &key).await; assert_eq!( got, Resolution::NoRepoDid, "firehose update is canonical and may legitimately remove repoDID", ); }
#[tokio::test] async fn forget_removes_cache_entry() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver .observe( owner.clone(), key.clone(), Some(did("did:plc:clam")), None, &made(1), ) .await; assert_eq!( resolver.cached_resolution(&owner, &key).await, Some(Resolution::Mapped(did("did:plc:clam"))), ); resolver.forget(&owner, &key).await; assert_eq!( resolver.cached_resolution(&owner, &key).await, None, "forget must drop the entry entirely so a subsequent observe can supply fresh state", ); }}