Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035mod legacy_upgrade;mod normalize;
pub use legacy_upgrade::{ DecodedRecord, decode_canon_or_upgrade, decode_canon_or_upgrade_bytes, normalize_record_fields, scrub_record_bytes, synthesize_created_at, upgrade, upgrade_wire_bytes,};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::edges::{ExtractError, Record};use bobbin_types::ids::{RepoIdent, nsid_static};use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::recordkey::Rkey;use scc::HashMap as SccMap;use tokio::sync::OnceCell;use tracing::warn;
const REPO_COLLECTION: &str = "sh.tangled.repo";const TRANSIENT_TTL: Duration = Duration::from_secs(60);
#[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)]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), } }}
struct SlingshotProbe { client: SlingshotClient, clock: Arc<dyn Clock>,}
pub struct RepoIdResolver { cache: SccMap<RepoIdent, CacheEntry, RuntimeHasher>, by_repo_did: SccMap<Did<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()), 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()), 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().clone()) }
pub async fn observe( &self, owner: Did<DefaultStr>, rkey: Rkey<DefaultStr>, repo_did: Option<Did<DefaultStr>>, ) -> Option<RepoIdent> { let ident = RepoIdent::new(owner, rkey); 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 repo_did = repo_did?; let mut prior: Option<RepoIdent> = None; self.by_repo_did .entry_async(repo_did) .await .and_modify(|existing| { if *existing != ident { prior = Some(existing.clone()); *existing = ident.clone(); } }) .or_insert(ident); prior }
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, |existing| *existing == 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 { self.fetch_repo_did(owner, rkey, &key).await }) .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(&nsid, &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 { .. }, )}
fn repo_did_from_body( nsid: &Nsid<DefaultStr>, body: &[u8],) -> Result<Option<Did<DefaultStr>>, ExtractError> { match DecodedRecord::try_decode(nsid, body)? { DecodedRecord::Canon(Record::Repo(repo)) => Ok(repo.repo_did), DecodedRecord::Canon(_) | DecodedRecord::Legacy(_) => Ok(None), }}
#[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()) }
#[tokio::test] async fn observation_returns_prior_ident_when_repo_did_moves() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let prior = resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), Some(did("did:plc:clam")), ) .await; assert!(prior.is_none(), "first observation has no prior");
let prior = resolver .observe(did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam"))) .await; assert_eq!( prior, 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 prior = resolver .observe(did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam"))) .await; assert!(prior.is_none(), "re-observing the same ident is a no-op"); }
#[tokio::test] async fn observation_without_repo_did_does_not_track_reverse() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let prior = resolver .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None) .await; assert!(prior.is_none()); }
#[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")), ) .await; resolver .observe(did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam"))) .await;
resolver .forget(&did("did:plc:nel"), &rkey("3liuighjy2h22")) .await;
let prior = resolver .observe( did("did:plc:nel"), rkey("core-renamed"), Some(did("did:plc:clam")), ) .await; assert_eq!( prior, 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 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")), ) .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) .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")), ) .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) .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")), ) .await; resolver .observe( did("did:plc:olaren"), rkey("xyzxyzxyzxyzx"), Some(did("did:plc:limpet")), ) .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")), ) .await; resolver .observe( did("did:plc:nel"), rkey("abcabcabcabcz"), Some(did("did:plc:uni")), ) .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"))) .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).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_unparseable_repo_value_caches_as_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"} }); 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, "a repo body that fails lexicon validation must not be conflated with NoRepoDid", ); assert_eq!(second, Resolution::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).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"))) .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", ); }}