Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795use std::sync::atomic::{AtomicU64, Ordering};use std::sync::{Arc, Mutex};use std::time::Duration;
use bobbin_runtime::{Clock, RuntimeHasher};use bobbin_slingshot_client::{SlingshotClient, SlingshotError};use bobbin_types::identity::IdentitySink;use jacquard_common::DefaultStr;use jacquard_common::types::crypto::PublicKey;use jacquard_common::types::did::Did;use jacquard_common::types::ident::AtIdentifier;use jacquard_common::types::string::Handle;use scc::HashMap as SccMap;use scc::HashSet as SccSet;use scc::hash_map::Entry as MapEntry;use serde::{Deserialize, Serialize};use thiserror::Error;use tokio::sync::{OnceCell, Semaphore, mpsc};use tokio::time::Instant;use tokio_util::sync::CancellationToken;use tracing::debug;use url::Url;
use crate::SlingshotProbe;use crate::hydrant::{HydrantClient, RepoIdentity};
const WARM_CONCURRENCY: usize = 32;// how long a failed lookup suppresses another attempt for that didconst WARM_RETRY_AFTER: Duration = Duration::from_secs(300);// identity frames are best effort and can miss a did doc change, so cached knots// expire on their own tooconst KNOT_TTL: Duration = Duration::from_secs(3600);
#[derive(Clone, Copy)]enum WarmKind { Fill, Refresh,}
type WarmItem = (Did<DefaultStr>, WarmKind);
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]pub struct MiniDoc { pub did: Did<DefaultStr>, pub handle: Handle<DefaultStr>, #[serde(skip_serializing_if = "Option::is_none")] pub pds: Option<Url>, #[serde(default, with = "signing_key", skip_serializing_if = "Option::is_none")] pub signing_key: Option<PublicKey<'static>>,}
pub(crate) mod signing_key { use jacquard_common::types::crypto::{KeyCodec, PublicKey, multikey}; use serde::de::Error as _; use serde::{Deserialize, Deserializer, Serializer};
pub fn serialize<S: Serializer>( key: &Option<PublicKey<'static>>, serializer: S, ) -> Result<S::Ok, S::Error> { match key { Some(key) => { // this is here because jacquard doesnt let us use theirs :( let code = match key.codec { KeyCodec::Ed25519 => 0xED, KeyCodec::Secp256k1 => 0xE7, KeyCodec::P256 => 0x1200, KeyCodec::Unknown(code) => code, }; serializer.serialize_str(&multikey(code, &key.bytes)) } None => serializer.serialize_none(), } }
pub fn deserialize<'de, D: Deserializer<'de>>( deserializer: D, ) -> Result<Option<PublicKey<'static>>, D::Error> { let Some(raw) = Option::<std::borrow::Cow<'de, str>>::deserialize(deserializer)? else { return Ok(None); }; PublicKey::decode_owned(raw.as_ref()) .map(Some) .map_err(D::Error::custom) }}
#[derive(Clone, Debug, Error, Eq, PartialEq)]pub enum IdentityResolveError { #[error("identity not found")] NotFound, #[error("identity upstream: {0}")] Upstream(String), #[error("invalid identity response: {0}")] Decode(String),}
impl From<SlingshotError> for IdentityResolveError { fn from(error: SlingshotError) -> Self { match error { SlingshotError::NotFound => Self::NotFound, other => Self::Upstream(other.to_string()), } }}
type ResolveCell = Arc<OnceCell<Result<MiniDoc, IdentityResolveError>>>;
#[derive(Clone)]enum IdentityState { Cached(MiniDoc), Inactive { handle: Option<Handle<DefaultStr>>, }, Revalidating { handle: Option<Handle<DefaultStr>>, cell: ResolveCell, },}
impl IdentityState { fn doc(&self) -> Option<&MiniDoc> { match self { Self::Cached(doc) => Some(doc), Self::Inactive { .. } | Self::Revalidating { .. } => None, } }
fn handle(&self) -> Option<&Handle<DefaultStr>> { match self { Self::Cached(doc) => Some(&doc.handle), Self::Inactive { handle } | Self::Revalidating { handle, .. } => handle.as_ref(), } }}
enum Revalidation { Cached(MiniDoc), Flight(ResolveCell), Missing, Suppressed,}
struct WarmQueueGuard<'a> { queued: &'a SccSet<Did<DefaultStr>, RuntimeHasher>, did: &'a Did<DefaultStr>,}
impl Drop for WarmQueueGuard<'_> { fn drop(&mut self) { self.queued.remove_sync(self.did); }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub struct IdentityResolverStatsSnapshot { pub entries: usize, pub hits: u64, pub misses: u64, pub upstream_requests: u64, pub warm_queued: u64, pub warm_dropped: u64, pub warm_resolved: u64, pub warm_failed: u64,}
#[derive(Default)]struct IdentityResolverStats { hits: AtomicU64, misses: AtomicU64, upstream_requests: AtomicU64, warm_queued: AtomicU64, warm_dropped: AtomicU64, warm_resolved: AtomicU64, warm_failed: AtomicU64,}
#[derive(Clone)]struct KnotEntry { knot: Option<String>, expires: Option<Instant>,}
// in the future this would spill to disk probablypub struct IdentityResolver { by_did: SccMap<Did<DefaultStr>, IdentityState, RuntimeHasher>, by_handle: SccMap<Handle<DefaultStr>, Did<DefaultStr>, RuntimeHasher>, in_flight: SccMap<String, ResolveCell, RuntimeHasher>, // dids queued for background warming, keeps the queue free of duplicates queued: SccSet<Did<DefaultStr>, RuntimeHasher>, // dids whose last warm failed, mapped to when we may try again cooldown: SccMap<Did<DefaultStr>, Instant, RuntimeHasher>, knots: SccMap<Did<DefaultStr>, KnotEntry, RuntimeHasher>, warm_tx: mpsc::UnboundedSender<WarmItem>, warm_rx: Mutex<Option<mpsc::UnboundedReceiver<WarmItem>>>, hydrant: Option<HydrantClient>, probe: Option<SlingshotProbe>, clock: Option<Arc<dyn Clock>>, sink: Option<Arc<dyn IdentitySink>>, mutations: Mutex<()>, stats: IdentityResolverStats,}
impl IdentityResolver { pub fn with_slingshot( client: SlingshotClient, clock: Arc<dyn Clock>, hasher: RuntimeHasher, ) -> Self { Self::new(Some(SlingshotProbe { client, clock }), hasher) }
pub fn detached(hasher: RuntimeHasher) -> Self { Self::new(None, hasher) }
pub fn with_hydrant(mut self, client: HydrantClient) -> Self { self.hydrant = Some(client); self }
pub fn with_sink(mut self, sink: Arc<dyn IdentitySink>) -> Self { self.sink = Some(sink); self }
fn new(probe: Option<SlingshotProbe>, hasher: RuntimeHasher) -> Self { let clock = probe.as_ref().map(|probe| probe.clock.clone()); let (warm_tx, warm_rx) = mpsc::unbounded_channel(); Self { by_did: SccMap::with_hasher(hasher.clone()), by_handle: SccMap::with_hasher(hasher.clone()), in_flight: SccMap::with_hasher(hasher.clone()), queued: SccSet::with_hasher(hasher.clone()), cooldown: SccMap::with_hasher(hasher.clone()), knots: SccMap::with_hasher(hasher), warm_tx, warm_rx: Mutex::new(Some(warm_rx)), hydrant: None, probe, clock, sink: None, mutations: Mutex::new(()), stats: IdentityResolverStats::default(), } }
pub fn stats(&self) -> IdentityResolverStatsSnapshot { IdentityResolverStatsSnapshot { entries: self.by_did.len(), hits: self.stats.hits.load(Ordering::Relaxed), misses: self.stats.misses.load(Ordering::Relaxed), upstream_requests: self.stats.upstream_requests.load(Ordering::Relaxed), warm_queued: self.stats.warm_queued.load(Ordering::Relaxed), warm_dropped: self.stats.warm_dropped.load(Ordering::Relaxed), warm_resolved: self.stats.warm_resolved.load(Ordering::Relaxed), warm_failed: self.stats.warm_failed.load(Ordering::Relaxed), } }
pub fn warm(&self, did: &Did<DefaultStr>) { let kind = match self.by_did.get_sync(did).as_deref() { Some(IdentityState::Cached(_)) => return, Some(IdentityState::Inactive { .. } | IdentityState::Revalidating { .. }) => { WarmKind::Refresh } None => WarmKind::Fill, }; self.enqueue(did, kind); }
pub fn refresh(&self, did: &Did<DefaultStr>) { self.knots.remove_sync(did); self.enqueue(did, WarmKind::Refresh); }
/// the knot host a repo did names, `None` when it names none or hydrant can't say pub async fn repo_knot(&self, did: &Did<DefaultStr>) -> Option<String> { let now = self.clock.as_ref().map(|clock| clock.now_instant()); let cached = self.knots.read_sync(did, |_, entry| entry.clone()); if let Some(entry) = cached && entry .expires .zip(now) .is_none_or(|(expires, now)| now < expires) { return entry.knot; } let knot = self .hydrant .as_ref()? .repo_knot(did) .await .inspect_err(|error| debug!(did = did.as_str(), %error, "knot lookup failed")) .ok()?; let expires = now.map(|now| now + KNOT_TTL); self.knots.upsert_sync( did.clone(), KnotEntry { knot: knot.clone(), expires, }, ); knot }
fn enqueue(&self, did: &Did<DefaultStr>, kind: WarmKind) { if self.cooling_down(did) { return; } if self.queued.insert_sync(did.clone()).is_err() { return; } if self.warm_tx.send((did.clone(), kind)).is_err() { self.queued.remove_sync(did); self.stats.warm_dropped.fetch_add(1, Ordering::Relaxed); return; } self.stats.warm_queued.fetch_add(1, Ordering::Relaxed); }
fn cooling_down(&self, did: &Did<DefaultStr>) -> bool { let Some(clock) = self.clock.as_ref() else { return false; }; self.cooldown .read_sync(did, |_, until| *until > clock.now_instant()) .unwrap_or(false) }
fn arm_cooldown(&self, did: &Did<DefaultStr>) { if let Some(clock) = self.clock.as_ref() { self.cooldown .upsert_sync(did.clone(), clock.now_instant() + WARM_RETRY_AFTER); } }
pub async fn run_warming(self: Arc<Self>, cancel: CancellationToken) { let Some(mut rx) = self .warm_rx .lock() .expect("identity warm receiver mutex poisoned") .take() else { return; }; let permits = Arc::new(Semaphore::new(WARM_CONCURRENCY)); loop { let (did, kind) = tokio::select! { biased; _ = cancel.cancelled() => break, next = rx.recv() => match next { Some(next) => next, None => break, }, }; let Ok(permit) = permits.clone().acquire_owned().await else { break; }; let resolver = self.clone(); tokio::spawn(async move { let _permit = permit; resolver.warm_one(did, kind).await; }); } }
async fn warm_one(&self, did: Did<DefaultStr>, kind: WarmKind) { let _queued = WarmQueueGuard { queued: &self.queued, did: &did, }; let (outcome, arm_on_error) = match kind { _ if self.needs_revalidation(&did) => { (self.revalidate_did(&did).await.map(drop), false) } WarmKind::Refresh => ( match self.fetch_upstream(&AtIdentifier::Did(did.clone())).await { Ok(doc) => self.accept_fetched(doc, true).await.map(drop), Err(error) => Err(error), }, true, ), WarmKind::Fill => (self.fill_one(&did).await, true), };
match outcome { Ok(()) => { self.cooldown.remove_sync(&did); self.stats.warm_resolved.fetch_add(1, Ordering::Relaxed); } Err(error) => { if arm_on_error { self.arm_cooldown(&did); } self.stats.warm_failed.fetch_add(1, Ordering::Relaxed); debug!(did = did.as_ref(), %error, "identity warm failed"); } } }
async fn fill_one(&self, did: &Did<DefaultStr>) -> Result<(), IdentityResolveError> { let known = match self.hydrant.as_ref() { Some(hydrant) => hydrant.repo_identity(did).await.unwrap_or_else(|error| { debug!(did = did.as_ref(), %error, "hydrant warm failed"); None }), None => None, }; match known { Some(RepoIdentity::Live(doc)) => self.accept_fetched(doc, true).await.map(drop), Some(RepoIdentity::Dead) => { self.deactivate(did.clone()); Ok(()) } None => self .resolve_minidoc(&AtIdentifier::Did(did.clone())) .await .map(drop), } }
pub fn observe(&self, did: Did<DefaultStr>, handle: Handle<DefaultStr>) { self.knots.remove_sync(&did); let observed = MiniDoc { did: did.clone(), handle, pds: None, signing_key: None, }; let refresh = { let _mutation = self .mutations .lock() .expect("identity mutation mutex poisoned"); let state = self.by_did.get_sync(&did).map(|state| state.get().clone()); match state { Some(IdentityState::Inactive { .. }) => true, Some(IdentityState::Revalidating { .. }) => true, Some(IdentityState::Cached(doc)) => { let incomplete = doc.pds.is_none(); self.observe_doc_locked(observed); incomplete } None => { self.observe_doc_locked(observed); false } } }; if refresh { self.refresh(&did); } }
fn observe_doc_locked(&self, doc: MiniDoc) { let (did, handle) = (doc.did.clone(), doc.handle.clone()); let mut previous_handle = None; match self.by_did.entry_sync(did.clone()) { MapEntry::Occupied(mut occupied) => { previous_handle = occupied.get().handle().cloned(); let merged = match occupied.get().doc() { Some(previous) => MiniDoc { pds: doc.pds.or_else(|| previous.pds.clone()), signing_key: doc.signing_key.or_else(|| previous.signing_key.clone()), ..doc }, None => doc, }; occupied.insert(IdentityState::Cached(merged)); } MapEntry::Vacant(vacant) => { vacant.insert_entry(IdentityState::Cached(doc)); } } self.remove_by_handle_if_owned(&did, previous_handle.as_ref()); self.insert_by_handle(did.clone(), handle.clone()); if let Some(sink) = self.sink.as_ref() { sink.identity_changed(&did, previous_handle.as_ref(), &handle); } }
pub fn deactivate(&self, did: Did<DefaultStr>) { self.knots.remove_sync(&did); let _mutation = self .mutations .lock() .expect("identity mutation mutex poisoned"); let previous_handle = self .by_did .get_sync(&did) .and_then(|state| state.get().handle().cloned()); self.by_did.upsert_sync( did.clone(), IdentityState::Inactive { handle: previous_handle.clone(), }, ); if let Some(sink) = self.sink.as_ref() { sink.identity_removed(&did, previous_handle.as_ref()); } }
fn remove_by_handle_if_owned( &self, did: &Did<DefaultStr>, handle: Option<&Handle<DefaultStr>>, ) { let Some(handle) = handle else { return; }; self.by_handle.remove_if_sync(handle, |owner| owner == did); }
fn remove_by_handle_for_removed_did(&self, removed: Option<(Did<DefaultStr>, IdentityState)>) { let Some((did, state)) = removed else { return; }; let Some(handle) = state.handle() else { return; }; self.remove_by_handle_if_owned(&did, Some(handle)); }
fn insert_by_handle(&self, did: Did<DefaultStr>, handle: Handle<DefaultStr>) { if let Some(displaced_did) = self .by_handle .upsert_sync(handle.clone(), did.clone()) .filter(|displaced_did| displaced_did != &did) { let removed = self.by_did.remove_if_sync(&displaced_did, |state| { state.doc().is_some_and(|doc| doc.handle == handle) }); self.remove_by_handle_for_removed_did(removed); }
if self.by_did_matches_handle(&did, &handle) { return; } self.remove_by_handle_if_owned(&did, Some(&handle)); }
fn by_did_matches_handle(&self, did: &Did<DefaultStr>, handle: &Handle<DefaultStr>) -> bool { self.by_did .get_sync(did) .is_some_and(|state| state.get().doc().is_some_and(|doc| doc.handle == *handle)) }
fn cached( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<Option<MiniDoc>, IdentityResolveError> { match identifier { AtIdentifier::Did(did) => Ok(self .by_did .get_sync(did) .and_then(|state| state.get().doc().cloned())), AtIdentifier::Handle(handle) => { let Some(did) = self .by_handle .get_sync(handle) .map(|entry| entry.get().clone()) else { return Ok(None); }; let state = self.by_did.get_sync(&did).map(|state| state.get().clone()); match state.as_ref() { Some(IdentityState::Cached(doc)) if doc.handle == *handle => { Ok(Some(doc.clone())) } Some(state) if state.handle() == Some(handle) => Ok(None), _ => { self.remove_by_handle_if_owned(&did, Some(handle)); Ok(None) } } } } }
fn get_cached( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<Option<MiniDoc>, IdentityResolveError> { let result = self.cached(identifier); match &result { Ok(Some(_)) | Err(_) => self.stats.hits.fetch_add(1, Ordering::Relaxed), Ok(None) => self.stats.misses.fetch_add(1, Ordering::Relaxed), }; result }
fn needs_revalidation(&self, did: &Did<DefaultStr>) -> bool { self.by_did.get_sync(did).is_some_and(|state| { matches!( state.get(), IdentityState::Inactive { .. } | IdentityState::Revalidating { .. } ) }) }
fn inactive_did_for_handle(&self, handle: &Handle<DefaultStr>) -> Option<Did<DefaultStr>> { let did = self.by_handle.get_sync(handle)?.get().clone(); self.by_did .get_sync(&did) .is_some_and(|state| { matches!( state.get(), IdentityState::Inactive { .. } | IdentityState::Revalidating { .. } ) && state.get().handle() == Some(handle) }) .then_some(did) }
/// reads the cache without waiting, a miss queues a background warm pub fn get_by_did(&self, did: &Did<DefaultStr>) -> Result<MiniDoc, IdentityResolveError> { match self.get_cached(&AtIdentifier::Did(did.clone()))? { Some(doc) => Ok(doc), None => { self.warm(did); Err(IdentityResolveError::NotFound) } } }
/// resolve a minidoc, waiting on slingshot when the cache misses pub async fn resolve_minidoc( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<MiniDoc, IdentityResolveError> { if let Some(doc) = self.get_cached(identifier)? { return Ok(doc); } match identifier { AtIdentifier::Did(did) if self.needs_revalidation(did) => { self.revalidate_did(did).await } AtIdentifier::Handle(handle) => { if let Some(did) = self.inactive_did_for_handle(handle) { let doc = self.revalidate_did(&did).await?; return (doc.handle == *handle) .then_some(doc) .ok_or(IdentityResolveError::NotFound); } self.resolve_uncached(identifier).await } AtIdentifier::Did(_) => self.resolve_uncached(identifier).await, } }
async fn resolve_uncached( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<MiniDoc, IdentityResolveError> { let key = identifier.as_str().to_owned(); let cell = 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_minidoc(identifier).await }) .await .clone(); self.in_flight .remove_if_async(&key, |current| Arc::ptr_eq(current, &cell)) .await; result }
async fn fetch_upstream( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<MiniDoc, IdentityResolveError> { let probe = self.probe.as_ref().ok_or(IdentityResolveError::NotFound)?; self.stats.upstream_requests.fetch_add(1, Ordering::Relaxed); let bytes = probe .client .resolve_mini_doc(identifier) .await .map_err(IdentityResolveError::from)?; serde_json::from_slice::<MiniDoc>(&bytes) .map_err(|error| IdentityResolveError::Decode(error.to_string())) }
async fn fetch_minidoc( &self, identifier: &AtIdentifier<DefaultStr>, ) -> Result<MiniDoc, IdentityResolveError> { let doc = self.fetch_upstream(identifier).await?; self.accept_fetched(doc, false).await }
async fn accept_fetched( &self, doc: MiniDoc, overwrite_cached: bool, ) -> Result<MiniDoc, IdentityResolveError> { { let _mutation = self .mutations .lock() .expect("identity mutation mutex poisoned"); let state = self .by_did .get_sync(&doc.did) .map(|state| state.get().clone()); match state { Some(IdentityState::Inactive { .. } | IdentityState::Revalidating { .. }) => {} Some(IdentityState::Cached(current)) if !overwrite_cached => return Ok(current), Some(IdentityState::Cached(_)) | None => { self.observe_doc_locked(doc.clone()); return Ok(doc); } } } self.revalidate_did(&doc.did).await }
fn start_revalidation(&self, did: &Did<DefaultStr>) -> Revalidation { let _mutation = self .mutations .lock() .expect("identity mutation mutex poisoned"); match self.by_did.entry_sync(did.clone()) { MapEntry::Occupied(mut occupied) => match occupied.get() { IdentityState::Cached(doc) => Revalidation::Cached(doc.clone()), IdentityState::Inactive { .. } if self.cooling_down(did) => { Revalidation::Suppressed } IdentityState::Inactive { handle } => { let handle = handle.clone(); let cell = Arc::new(OnceCell::new()); occupied.insert(IdentityState::Revalidating { handle, cell: cell.clone(), }); Revalidation::Flight(cell) } IdentityState::Revalidating { cell, .. } => Revalidation::Flight(cell.clone()), }, MapEntry::Vacant(_) => Revalidation::Missing, } }
async fn revalidate_did(&self, did: &Did<DefaultStr>) -> Result<MiniDoc, IdentityResolveError> { match self.start_revalidation(did) { Revalidation::Cached(doc) => Ok(doc), Revalidation::Flight(cell) => { let owner = cell.clone(); // share whether the fetch was accepted, not just what it returned cell.get_or_init(|| async { let result = self.fetch_upstream(&AtIdentifier::Did(did.clone())).await; self.finish_revalidation(did, &owner, result) }) .await .clone() } Revalidation::Missing | Revalidation::Suppressed => Err(IdentityResolveError::NotFound), } }
fn finish_revalidation( &self, did: &Did<DefaultStr>, owner: &ResolveCell, result: Result<MiniDoc, IdentityResolveError>, ) -> Result<MiniDoc, IdentityResolveError> { let _mutation = self .mutations .lock() .expect("identity mutation mutex poisoned"); let state = self.by_did.get_sync(did).map(|state| state.get().clone()); let handle = match state { Some(IdentityState::Revalidating { handle, cell }) if Arc::ptr_eq(&cell, owner) => { handle } Some(IdentityState::Cached(doc)) => return Ok(doc), Some(IdentityState::Inactive { .. } | IdentityState::Revalidating { .. }) | None => { return Err(IdentityResolveError::NotFound); } }; match result { Ok(doc) if doc.did != *did => { self.by_did .upsert_sync(did.clone(), IdentityState::Inactive { handle }); self.arm_cooldown(did); Err(IdentityResolveError::Decode(format!( "resolved {} while revalidating {}", doc.did.as_str(), did.as_str() ))) } Ok(doc) => { self.cooldown.remove_sync(did); self.observe_doc_locked(doc.clone()); Ok(doc) } Err(error) => { self.by_did .upsert_sync(did.clone(), IdentityState::Inactive { handle }); self.arm_cooldown(did); Err(error) } } }}
#[cfg(test)]mod tests { use super::*; use bobbin_runtime::{ReqwestHttp, SystemClock}; use bobbin_slingshot_client::default_http_client; use url::Url; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate};
fn hasher() -> RuntimeHasher { RuntimeHasher::from_seeds(1, 2, 3, 4) }
fn resolver() -> IdentityResolver { IdentityResolver::detached(hasher()) }
fn did(value: &str) -> Did<DefaultStr> { Did::new_owned(value).unwrap() }
fn handle(value: &str) -> Handle<DefaultStr> { Handle::new_owned(value).unwrap() }
#[test] fn observed_identity_resolves_by_did_without_upstream() { let resolver = resolver(); resolver.observe(did("did:plc:dawn"), handle("ptr.pet"));
let doc = resolver.get_by_did(&did("did:plc:dawn")).unwrap(); assert_eq!(doc.handle, handle("ptr.pet")); assert_eq!(doc.pds, None); assert_eq!(resolver.stats().hits, 1); }
#[test] fn observed_identity_updates_existing_did_and_removes_old_handle() { let resolver = resolver(); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet")); resolver.observe(identity.clone(), handle("new.ptr.pet"));
assert!( resolver .cached(&AtIdentifier::Handle(handle("ptr.pet"))) .unwrap() .is_none() ); let updated = resolver.get_by_did(&identity).unwrap(); assert_eq!(updated.handle, handle("new.ptr.pet")); }
#[test] fn handle_reassignment_keeps_the_new_owner() { let resolver = resolver(); let first = did("did:plc:first"); let second = did("did:plc:second"); let shared = handle("shared.example.com"); resolver.observe(first.clone(), shared.clone()); resolver.observe(second.clone(), shared.clone()); resolver.observe(first, handle("first.example.com"));
let cached = resolver .cached(&AtIdentifier::Handle(shared)) .unwrap() .unwrap(); assert_eq!(cached.did, second); }
#[test] fn handle_lookup_discards_an_unvalidated_reverse_hint() { let resolver = resolver(); let identity = did("did:plc:dawn"); let stale = handle("stale.example.com"); resolver.observe(identity.clone(), handle("current.example.com")); resolver .by_handle .upsert_sync(stale.clone(), identity.clone());
assert!( resolver .cached(&AtIdentifier::Handle(stale.clone())) .unwrap() .is_none() ); assert!(resolver.by_handle.get_sync(&stale).is_none()); }
// the codec mapping is ours, a wrong code re-encodes to a different key #[test] fn signing_key_survives_a_json_round_trip() { let encoded = "zQ3shSiLsnqpyQ4SfDTT1D8qzFEoeYT8rSDXW6o8pVY7VcRBJ"; let doc = MiniDoc { did: did("did:plc:dawn"), handle: handle("ptr.pet"), pds: Some(Url::parse("https://gaze.systems/").unwrap()), signing_key: Some(PublicKey::decode_owned(encoded).unwrap()), };
let json = serde_json::to_value(&doc).unwrap(); assert_eq!(json["signing_key"], serde_json::json!(encoded)); assert_eq!(json["pds"], serde_json::json!("https://gaze.systems/")); assert_eq!(serde_json::from_value::<MiniDoc>(json).unwrap(), doc); }
#[test] fn absent_optional_fields_round_trip_as_absent() { let doc = MiniDoc { did: did("did:plc:dawn"), handle: handle("ptr.pet"), pds: None, signing_key: None, };
let json = serde_json::to_value(&doc).unwrap(); assert!(json.get("signing_key").is_none()); assert!(json.get("pds").is_none()); assert_eq!(serde_json::from_value::<MiniDoc>(json).unwrap(), doc); }
#[test] fn handle_lookup_keeps_a_reverse_hint_the_forward_entry_agrees_with() { let resolver = resolver(); let identity = did("did:plc:dawn"); let handle = handle("ptr.pet"); resolver.observe(identity.clone(), handle.clone());
let found = resolver .cached(&AtIdentifier::Handle(handle.clone())) .unwrap() .unwrap(); assert_eq!(found.did, identity); assert_eq!( resolver .by_handle .get_sync(&handle) .map(|entry| entry.get().clone()), Some(identity) ); }
#[tokio::test] async fn resolve_serves_a_hydrant_observed_entry_from_cache() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .respond_with(ResponseTemplate::new(500)) .expect(0) .mount(&server) .await; let client = SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = IdentityResolver::with_slingshot(client, Arc::new(SystemClock::new()), hasher()); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet"));
let doc = resolver .resolve_minidoc(&AtIdentifier::Did(identity)) .await .unwrap(); assert_eq!(doc.handle, handle("ptr.pet")); assert_eq!(doc.pds, None); assert_eq!(resolver.stats().upstream_requests, 0); }
#[test] fn deactivation_cancels_an_in_flight_revalidation() { let resolver = resolver(); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet")); resolver.deactivate(identity.clone()); let Revalidation::Flight(owner) = resolver.start_revalidation(&identity) else { panic!("inactive identity did not start revalidation"); }; resolver.deactivate(identity.clone());
assert_eq!( resolver.finish_revalidation( &identity, &owner, Ok(MiniDoc { did: identity.clone(), handle: handle("ptr.pet"), pds: Some(Url::parse("https://pds.example.com").unwrap()), signing_key: None, }) ), Err(IdentityResolveError::NotFound) ); assert!( resolver .cached(&AtIdentifier::Handle(handle("ptr.pet"))) .unwrap() .is_none() ); }
#[test] fn an_identity_frame_does_not_cancel_a_live_revalidation() { let resolver = resolver(); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet")); resolver.deactivate(identity.clone()); let Revalidation::Flight(owner) = resolver.start_revalidation(&identity) else { panic!("inactive identity did not start revalidation"); };
resolver.observe(identity.clone(), handle("ptr.pet")); let doc = MiniDoc { did: identity.clone(), handle: handle("ptr.pet"), pds: Some(Url::parse("https://pds.example.com").unwrap()), signing_key: None, }; assert_eq!( resolver.finish_revalidation(&identity, &owner, Ok(doc.clone())), Ok(doc) ); assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("ptr.pet") ); }
#[tokio::test] async fn warming_resolves_a_did_we_only_saw_on_a_record() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "ptr.pet", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await; let client = SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(); let resolver = Arc::new(IdentityResolver::with_slingshot( client, Arc::new(SystemClock::new()), hasher(), )); let identity = did("did:plc:dawn");
// a plain cache read is what feeds and enrich do, and it must not block assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound) ); assert_eq!(resolver.stats().warm_queued, 1);
let cancel = CancellationToken::new(); let warmer = tokio::spawn(resolver.clone().run_warming(cancel.clone())); let warmed = tokio::time::timeout(Duration::from_secs(5), async { loop { if let Ok(doc) = resolver.get_by_did(&identity) { return doc; } tokio::task::yield_now().await; } }) .await .expect("warmer should resolve the queued did"); assert_eq!(warmed.handle, handle("ptr.pet")); assert_eq!( warmed.pds, Some(Url::parse("https://pds.example.com").unwrap()) );
cancel.cancel(); warmer.await.unwrap(); assert_eq!(resolver.stats().warm_resolved, 1); }
async fn drain_warming(resolver: &Arc<IdentityResolver>, did: &Did<DefaultStr>) { let settled_before = resolver.stats().warm_resolved + resolver.stats().warm_failed; let cancel = CancellationToken::new(); let warmer = tokio::spawn(resolver.clone().run_warming(cancel.clone())); tokio::time::timeout(Duration::from_secs(5), async { while resolver.stats().warm_resolved + resolver.stats().warm_failed == settled_before { tokio::task::yield_now().await; } }) .await .unwrap_or_else(|_| panic!("warmer never settled {did}")); cancel.cancel(); warmer.await.unwrap(); }
fn hydrant_backed(server: &MockServer) -> Arc<IdentityResolver> { let base = Url::parse(&server.uri()).unwrap(); Arc::new( IdentityResolver::with_slingshot( SlingshotClient::with_default_http(base.clone()).unwrap(), Arc::new(SystemClock::new()), hasher(), ) .with_hydrant( HydrantClient::new(base, ReqwestHttp::shared(default_http_client().unwrap())) .unwrap(), ), ) }
#[tokio::test] async fn warming_prefers_hydrant_over_slingshot() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "status": "synced", "tracked": true, "handle": "ptr.pet", }))) .expect(1) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .respond_with(ResponseTemplate::new(500)) .expect(0) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound) );
drain_warming(&resolver, &identity).await; assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("ptr.pet") ); assert_eq!(resolver.stats().upstream_requests, 0); }
#[tokio::test] async fn repo_knot_is_cached_until_an_identity_change() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.describeRepo")) .and(query_param("repo", "did:plc:kelp")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "didDoc": { "id": "did:plc:kelp", "service": [{ "id": "#tangled_knot", "type": "TangledKnot", "serviceEndpoint": "https://knot.example.com/repo/abc", }], }, }))) .expect(2) .mount(&server) .await;
let resolver = hydrant_backed(&server); let repo = did("did:plc:kelp"); for _ in 0..2 { assert_eq!( resolver.repo_knot(&repo).await.as_deref(), Some("knot.example.com") ); } resolver.refresh(&repo); assert_eq!( resolver.repo_knot(&repo).await.as_deref(), Some("knot.example.com") ); }
#[tokio::test] async fn warming_falls_back_to_slingshot_for_dids_hydrant_lacks() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:stranger")) .respond_with(ResponseTemplate::new(404)) .expect(1) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:stranger")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:stranger", "handle": "stranger.example.com", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:stranger"); resolver.warm(&identity); drain_warming(&resolver, &identity).await;
assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("stranger.example.com") ); assert_eq!(resolver.stats().warm_resolved, 1); }
#[tokio::test] async fn warming_falls_back_when_hydrant_errors() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:dawn")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "ptr.pet", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); resolver.warm(&identity); drain_warming(&resolver, &identity).await;
assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("ptr.pet") ); }
#[tokio::test] async fn refresh_re_resolves_upstream_and_skips_hydrant() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:dawn")) .respond_with(ResponseTemplate::new(500)) .expect(0) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "new.ptr.pet", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("old.ptr.pet")); resolver.refresh(&identity);
assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("old.ptr.pet"), "the cached handle keeps serving while the refresh is in flight" );
drain_warming(&resolver, &identity).await; assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("new.ptr.pet") ); assert!( resolver .by_handle .get_sync(&handle("old.ptr.pet")) .is_none() ); }
#[tokio::test] async fn a_failed_refresh_leaves_the_cached_handle_alone() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet")); resolver.refresh(&identity); drain_warming(&resolver, &identity).await;
assert_eq!( resolver.get_by_did(&identity).unwrap().handle, handle("ptr.pet") ); assert_eq!(resolver.stats().warm_failed, 1); }
#[tokio::test] async fn a_deactivated_did_reverifies_on_read() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with( ResponseTemplate::new(200) .set_delay(Duration::from_millis(100)) .set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "ptr.pet", "pds": "https://pds.example.com" })), ) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); resolver.deactivate(identity.clone());
assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound), "the read queues a revalidation instead of waiting on it" ); assert_eq!(resolver.stats().warm_queued, 1);
let cancel = CancellationToken::new(); let warmer = tokio::spawn(resolver.clone().run_warming(cancel.clone())); tokio::time::timeout(Duration::from_secs(5), async { while !matches!( resolver.by_did.get_sync(&identity).as_deref(), Some(IdentityState::Revalidating { .. }) ) { tokio::task::yield_now().await; } }) .await .expect("background revalidation never started");
let doc = resolver .resolve_minidoc(&AtIdentifier::Did(identity.clone())) .await .unwrap(); tokio::time::timeout(Duration::from_secs(5), async { while resolver.stats().warm_resolved == 0 { tokio::task::yield_now().await; } }) .await .expect("background revalidation never settled"); cancel.cancel(); warmer.await.unwrap();
assert_eq!(doc.handle, handle("ptr.pet")); assert_eq!(resolver.stats().upstream_requests, 1); assert_eq!( doc.pds, Some(Url::parse("https://pds.example.com").unwrap()) ); }
#[tokio::test] async fn a_deactivated_did_that_stays_dead_is_not_refetched_per_read() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .respond_with(ResponseTemplate::new(404)) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:gone"); resolver.deactivate(identity.clone());
for _ in 0..3 { assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound) ); } assert_eq!(resolver.stats().warm_queued, 1); drain_warming(&resolver, &identity).await;
for _ in 0..3 { assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound) ); } let stats = resolver.stats(); assert_eq!(stats.warm_queued, 1); assert_eq!(stats.warm_failed, 1); }
#[tokio::test] async fn failed_foreground_revalidation_obeys_the_cooldown() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:gone")) .respond_with(ResponseTemplate::new(404)) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:gone"); resolver.deactivate(identity.clone()); let identifier = AtIdentifier::Did(identity);
for _ in 0..3 { assert_eq!( resolver.resolve_minidoc(&identifier).await, Err(IdentityResolveError::NotFound) ); } assert_eq!(resolver.stats().upstream_requests, 1); }
#[tokio::test] async fn did_and_handle_lookups_share_an_inactive_revalidation() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "dawn.example.com")) .respond_with(ResponseTemplate::new(404)) .expect(0) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "dawn.example.com", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("dawn.example.com")); resolver.deactivate(identity.clone()); let by_handle = AtIdentifier::Handle(handle("dawn.example.com")); let by_did = AtIdentifier::Did(identity);
let (handle_result, did_result) = tokio::join!( resolver.resolve_minidoc(&by_handle), resolver.resolve_minidoc(&by_did) ); assert_eq!(handle_result.unwrap().handle, handle("dawn.example.com")); assert_eq!(did_result.unwrap().handle, handle("dawn.example.com")); assert_eq!(resolver.stats().upstream_requests, 1); assert_eq!(resolver.stats().warm_queued, 0); }
#[tokio::test] async fn an_inactive_handle_hint_cannot_claim_a_renamed_identity() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:dawn")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dawn", "handle": "new.example.com", "pds": "https://pds.example.com" }))) .expect(1) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:dawn"); let old_handle = handle("old.example.com"); resolver.observe(identity.clone(), old_handle.clone()); resolver.deactivate(identity);
assert_eq!( resolver .resolve_minidoc(&AtIdentifier::Handle(old_handle.clone())) .await, Err(IdentityResolveError::NotFound) ); assert!(resolver.by_handle.get_sync(&old_handle).is_none()); assert_eq!( resolver .cached(&AtIdentifier::Handle(handle("new.example.com"))) .unwrap() .unwrap() .handle, handle("new.example.com") ); }
#[tokio::test] async fn warming_a_dead_repo_caches_no_handle() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:gone")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:gone", "status": "deleted", "tracked": true, "handle": "gone.example.com", }))) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .respond_with(ResponseTemplate::new(500)) .expect(0) .mount(&server) .await;
let resolver = hydrant_backed(&server); let identity = did("did:plc:gone"); resolver.warm(&identity); drain_warming(&resolver, &identity).await;
assert_eq!( resolver.get_by_did(&identity), Err(IdentityResolveError::NotFound) ); resolver.warm(&identity); assert_eq!(resolver.stats().warm_queued, 1); }
#[tokio::test] async fn warming_queues_dids_without_dropping() { let resolver = Arc::new(IdentityResolver::with_slingshot( SlingshotClient::with_default_http(Url::parse("http://127.0.0.1:1").unwrap()).unwrap(), Arc::new(SystemClock::new()), hasher(), )); for n in 0..10_000 { resolver.warm(&did(&format!("did:plc:bulk{n}"))); } assert_eq!(resolver.stats().warm_queued, 10_000); assert_eq!(resolver.stats().warm_dropped, 0); }
#[test] fn warming_skips_dids_we_already_know() { let resolver = resolver(); let identity = did("did:plc:dawn"); resolver.observe(identity.clone(), handle("ptr.pet")); resolver.warm(&identity); assert_eq!(resolver.stats().warm_queued, 0); }
#[derive(Clone, Debug, Eq, PartialEq)] enum SinkEvent { Changed { did: Did<DefaultStr>, previous_handle: Option<Handle<DefaultStr>>, handle: Handle<DefaultStr>, }, Removed { did: Did<DefaultStr>, previous_handle: Option<Handle<DefaultStr>>, }, }
#[derive(Default)] struct RecordingSink { events: Mutex<Vec<SinkEvent>>, }
impl IdentitySink for RecordingSink { fn identity_changed( &self, did: &Did<DefaultStr>, previous_handle: Option<&Handle<DefaultStr>>, handle: &Handle<DefaultStr>, ) { self.events.lock().unwrap().push(SinkEvent::Changed { did: did.clone(), previous_handle: previous_handle.cloned(), handle: handle.clone(), }); }
fn identity_removed( &self, did: &Did<DefaultStr>, previous_handle: Option<&Handle<DefaultStr>>, ) { self.events.lock().unwrap().push(SinkEvent::Removed { did: did.clone(), previous_handle: previous_handle.cloned(), }); } }
struct IdentityCheckingSink { resolver: Arc<OnceCell<Arc<IdentityResolver>>>, events: Mutex<Vec<SinkEvent>>, }
impl IdentitySink for IdentityCheckingSink { fn identity_changed( &self, did: &Did<DefaultStr>, previous_handle: Option<&Handle<DefaultStr>>, handle: &Handle<DefaultStr>, ) { let resolver = self.resolver.get().expect("resolver initialized"); let doc = resolver.get_by_did(did).expect("did must resolve"); assert_eq!(&doc.handle, handle); let by_handle = resolver .cached(&AtIdentifier::Handle(handle.clone())) .expect("cache lookup must succeed") .expect("handle must resolve"); assert_eq!(&by_handle.did, did); if let Some(prev) = previous_handle && prev != handle { assert!( resolver .cached(&AtIdentifier::Handle(prev.clone())) .expect("cache lookup must succeed") .is_none() ); } self.events.lock().unwrap().push(SinkEvent::Changed { did: did.clone(), previous_handle: previous_handle.cloned(), handle: handle.clone(), }); }
fn identity_removed( &self, did: &Did<DefaultStr>, previous_handle: Option<&Handle<DefaultStr>>, ) { let resolver = self.resolver.get().expect("resolver initialized"); assert_eq!( resolver.get_by_did(did), Err(IdentityResolveError::NotFound) ); if let Some(prev) = previous_handle { assert!( resolver .cached(&AtIdentifier::Handle(prev.clone())) .expect("cache lookup must succeed") .is_none() ); } self.events.lock().unwrap().push(SinkEvent::Removed { did: did.clone(), previous_handle: previous_handle.cloned(), }); } }
#[test] fn callback_runs_after_identity_maps_are_updated() { let cell = Arc::new(OnceCell::new()); let sink = Arc::new(IdentityCheckingSink { resolver: cell.clone(), events: Mutex::new(Vec::new()), }); let resolver = Arc::new(resolver().with_sink(sink.clone())); cell.set(resolver.clone()).ok().expect("cell set");
let identity = did("did:plc:dawn"); let initial_handle = handle("old.ptr.pet"); let updated_handle = handle("new.ptr.pet");
// 1. Initial observation resolver.observe(identity.clone(), initial_handle.clone()); // 2. Rename resolver.observe(identity.clone(), updated_handle.clone()); // 3. Deactivate resolver.deactivate(identity.clone());
let checked = sink.events.lock().unwrap().clone(); assert_eq!(checked.len(), 3); assert_eq!( checked[0], SinkEvent::Changed { did: identity.clone(), previous_handle: None, handle: initial_handle.clone(), } ); assert_eq!( checked[1], SinkEvent::Changed { did: identity.clone(), previous_handle: Some(initial_handle), handle: updated_handle.clone(), } ); assert_eq!( checked[2], SinkEvent::Removed { did: identity, previous_handle: Some(updated_handle), } ); }
#[tokio::test] async fn hydrant_results_are_sent_to_identity_sink() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/repos/did:plc:live")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:live", "status": "active", "tracked": true, "handle": "live.example.com", }))) .mount(&server) .await; Mock::given(method("GET")) .and(path("/repos/did:plc:dead")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "did": "did:plc:dead", "status": "deleted", "tracked": true, "handle": "dead.example.com", }))) .mount(&server) .await;
let sink = Arc::new(RecordingSink::default()); let base = Url::parse(&server.uri()).unwrap(); let resolver = Arc::new( IdentityResolver::with_slingshot( SlingshotClient::with_default_http(base.clone()).unwrap(), Arc::new(SystemClock::new()), hasher(), ) .with_hydrant( HydrantClient::new(base, ReqwestHttp::shared(default_http_client().unwrap())) .unwrap(), ) .with_sink(sink.clone()), );
let live_did = did("did:plc:live"); resolver.fill_one(&live_did).await.unwrap();
let dead_did = did("did:plc:dead"); resolver.fill_one(&dead_did).await.unwrap();
let events = sink.events.lock().unwrap().clone(); assert_eq!( events, vec![ SinkEvent::Changed { did: live_did, previous_handle: None, handle: handle("live.example.com"), }, SinkEvent::Removed { did: dead_did, previous_handle: None, }, ] ); }}