diff --git a/crates/bobbin-sim/src/workloads/concurrent_reads_during_replay.rs b/crates/bobbin-sim/src/workloads/concurrent_reads_during_replay.rs index fab4fd3..42a8ffa 100644 --- a/crates/bobbin-sim/src/workloads/concurrent_reads_during_replay.rs +++ b/crates/bobbin-sim/src/workloads/concurrent_reads_during_replay.rs @@ -7,9 +7,9 @@ use bobbin_edge_index::{PageCursor, PageLimit}; use bobbin_runtime::{ Clock, MemHttpResponder, MemWsResponder, MemWsServerFuture, WsMessage, }; -use bobbin_types::ids::EdgeKey; +use bobbin_types::ids::{EdgeKey, SubjectRef}; +use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::string::AtUri; use tokio::sync::mpsc; use url::Url; @@ -87,7 +87,7 @@ impl Workload for ConcurrentReadsDuringReplay { let read_key = EdgeKey::new( Nsid::new_static("sh.tangled.graph.follow").unwrap(), - AtUri::new_owned(format!("at://{TARGET_DID}")).unwrap(), + SubjectRef::Did(Did::new_owned(TARGET_DID).unwrap()), ); let final_key = read_key.clone(); diff --git a/crates/bobbin/src/main.rs b/crates/bobbin/src/main.rs index b1af496..b2c747c 100644 --- a/crates/bobbin/src/main.rs +++ b/crates/bobbin/src/main.rs @@ -185,6 +185,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { coverage, knots, search as Arc, + resolver, ); let app = router(state); diff --git a/crates/edge-index/Cargo.toml b/crates/edge-index/Cargo.toml index dfefa86..0653df9 100644 --- a/crates/edge-index/Cargo.toml +++ b/crates/edge-index/Cargo.toml @@ -10,7 +10,6 @@ bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } jacquard-common = { workspace = true } lasso = { workspace = true } -roaring = { workspace = true } scc = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } diff --git a/crates/edge-index/src/lib.rs b/crates/edge-index/src/lib.rs index cc86274..29a231c 100644 --- a/crates/edge-index/src/lib.rs +++ b/crates/edge-index/src/lib.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::num::NonZeroU32; use std::sync::{Arc, Mutex}; @@ -7,8 +7,8 @@ use bobbin_types::edges::Edge; use bobbin_types::ids::EdgeKey; use jacquard_common::DefaultStr; use jacquard_common::types::string::AtUri; +use jacquard_common::types::tid::Tid; use lasso::{Key, Spur, ThreadedRodeo}; -use roaring::RoaringBitmap; use scc::HashMap as SccMap; use thiserror::Error; @@ -30,17 +30,27 @@ impl SourceId { pub fn raw(self) -> u32 { self.0 } +} + +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub struct PageToken(u64); + +impl PageToken { + pub fn new(micros: u64) -> Self { + Self(micros) + } + + pub fn micros(self) -> u64 { + self.0 + } pub fn encode_token(self) -> String { - format!("{:0width$x}", self.0, width = CURSOR_TOKEN_LEN) + Tid::from_time(self.0, 0).as_str().to_owned() } pub fn decode_token(token: &str) -> Result { - if token.len() != CURSOR_TOKEN_LEN || !token.bytes().all(|b| b.is_ascii_hexdigit()) { - return Err(CursorParseError::Malformed); - } - u32::from_str_radix(token, 16) - .map(Self) + Tid::new(token) + .map(|t| Self(t.timestamp())) .map_err(|_| CursorParseError::Malformed) } } @@ -57,28 +67,26 @@ impl AuthorId { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum PageCursor { Start, - After(SourceId), + After(PageToken), } -const CURSOR_TOKEN_LEN: usize = 8; - #[derive(Clone, Copy, Debug, Eq, PartialEq, Error)] pub enum CursorParseError { - #[error("cursor token must be {CURSOR_TOKEN_LEN} hex characters")] + #[error("cursor token must be a valid TID")] Malformed, } impl PageCursor { - fn skip_until(self) -> Option { + fn skip_until(self) -> Option { match self { Self::Start => None, - Self::After(id) => Some(id.raw().saturating_add(1)), + Self::After(tok) => Some(tok.micros().saturating_add(1)), } } pub fn from_token(raw: Option<&str>) -> Result { raw.map_or(Ok(Self::Start), |t| { - SourceId::decode_token(t).map(Self::After) + PageToken::decode_token(t).map(Self::After) }) } } @@ -120,28 +128,34 @@ impl PageLimit { #[derive(Debug)] pub struct EdgePage { pub items: Vec>, - pub next: Option, + pub next: Option, } struct EdgeBucket { - sources: RoaringBitmap, + sources: BTreeSet<(u64, u32)>, author_refs: HashMap, } impl EdgeBucket { fn with_hasher(hasher: RuntimeHasher) -> Self { Self { - sources: RoaringBitmap::new(), + sources: BTreeSet::new(), author_refs: HashMap::with_hasher(hasher), } } } +#[derive(Clone, Debug, Eq, PartialEq)] +struct ReverseEntry { + key: EdgeKey, + sort_micros: u64, +} + pub struct EdgeStore { source_interner: Arc>, did_interner: Arc>, forward: SccMap, - reverse: SccMap, RuntimeHasher>, + reverse: SccMap, RuntimeHasher>, hasher: RuntimeHasher, writer: Mutex<()>, } @@ -191,6 +205,7 @@ impl EdgeStore { let source_spur = self.source_interner.get_or_intern(edge.source.as_ref()); let id = SourceId::from_spur(source_spur); let author = self.intern_author(edge.source.as_ref()); + let sort_micros = edge.sort_micros; let key = EdgeKey::new(edge.kind, edge.subject); let mut entry = self @@ -198,7 +213,7 @@ impl EdgeStore { .entry_sync(key.clone()) .or_insert_with(|| EdgeBucket::with_hasher(self.hasher.clone())); let bucket = entry.get_mut(); - let inserted = bucket.sources.insert(id.raw()); + let inserted = bucket.sources.insert((sort_micros, id.raw())); if inserted && let Some(author) = author { bump_ref(&mut bucket.author_refs, author); } @@ -206,7 +221,7 @@ impl EdgeStore { if inserted { let mut rev = self.reverse.entry_sync(id).or_default(); - rev.get_mut().push(key); + rev.get_mut().push(ReverseEntry { key, sort_micros }); } } @@ -218,12 +233,12 @@ impl EdgeStore { let author = author_str(source.as_ref()) .and_then(|s| self.did_interner.get(s)) .map(AuthorId::from_spur); - let Some((_, keys)) = self.reverse.remove_sync(&id) else { + let Some((_, entries)) = self.reverse.remove_sync(&id) else { return; }; - keys.into_iter().for_each(|key| { + entries.into_iter().for_each(|ReverseEntry { key, sort_micros }| { self.forward.update_sync(&key, |_, bucket| { - bucket.sources.remove(id.raw()); + bucket.sources.remove(&(sort_micros, id.raw())); if let Some(author) = author { drop_ref(&mut bucket.author_refs, author); } @@ -240,7 +255,7 @@ impl EdgeStore { pub fn count(&self, key: &EdgeKey) -> u64 { self.forward - .read_sync(key, |_, bucket| bucket.sources.len()) + .read_sync(key, |_, bucket| bucket.sources.len() as u64) .unwrap_or(0) } @@ -254,16 +269,18 @@ impl EdgeStore { let limit_usize = limit.get() as usize; self.forward .read_sync(key, |_, bucket| { - let mut iter = bucket.sources.iter(); - if let Some(skip_to) = cursor.skip_until() { - iter.advance_to(skip_to); - } - let raw_ids: Vec = iter.take(limit_usize + 1).collect(); - let has_more = raw_ids.len() > limit_usize; - let page_ids = &raw_ids[..raw_ids.len().min(limit_usize)]; - let items = page_ids + let start_micros = cursor.skip_until().unwrap_or(0); + let entries: Vec<(u64, u32)> = bucket + .sources + .range((start_micros, 0)..) + .take(limit_usize + 1) + .copied() + .collect(); + let has_more = entries.len() > limit_usize; + let page = &entries[..entries.len().min(limit_usize)]; + let items = page .iter() - .filter_map(|&raw| { + .filter_map(|&(_, raw)| { let spur = SourceId(raw).to_spur()?; self.source_interner .try_resolve(&spur) @@ -271,9 +288,9 @@ impl EdgeStore { }) .collect(); let next = has_more - .then(|| page_ids.last().copied()) + .then(|| page.last().copied()) .flatten() - .map(SourceId); + .map(|(m, _)| PageToken(m)); EdgePage { items, next } }) .unwrap_or(EdgePage { @@ -319,6 +336,8 @@ fn drop_ref(refs: &mut HashMap, author: Aut #[cfg(test)] mod tests { use super::*; + use bobbin_types::ids::SubjectRef; + use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri; @@ -334,36 +353,45 @@ mod tests { AtUri::new_owned(s).unwrap() } + fn did_subj(s: &str) -> SubjectRef { + SubjectRef::Did(Did::new_owned(s).unwrap()) + } + fn limit(n: u32) -> PageLimit { PageLimit::new(n).unwrap() } const NAMES: [&str; 5] = ["nel", "olaren", "teq", "lyna", "bailey"]; - fn star_edge(source: &str, subject: &str) -> Edge { + fn star_edge(source: &str, subject_did: &str) -> Edge { + star_edge_at(source, subject_did, 0) + } + + fn star_edge_at(source: &str, subject_did: &str, sort_micros: u64) -> Edge { Edge { kind: nsid("sh.tangled.feed.star"), - subject: at(subject), + subject: did_subj(subject_did), source: at(source), + sort_micros, } } #[test] fn add_then_count() { let store = store(); - let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); + let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); store.add(star_edge( "at://did:plc:nel/sh.tangled.feed.star/r1", - "at://did:plc:abalone", + "did:plc:abalone", )); store.add(star_edge( "at://did:plc:olaren/sh.tangled.feed.star/r2", - "at://did:plc:abalone", + "did:plc:abalone", )); store.add(star_edge( "at://did:plc:nel/sh.tangled.feed.star/r3", - "at://did:plc:abalone", + "did:plc:abalone", )); assert_eq!(store.count(&key), 3); @@ -373,10 +401,10 @@ mod tests { #[test] fn duplicate_add_is_idempotent() { let store = store(); - let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); + let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); let edge = star_edge( "at://did:plc:nel/sh.tangled.feed.star/r1", - "at://did:plc:abalone", + "did:plc:abalone", ); store.add(edge.clone()); store.add(edge); @@ -387,19 +415,21 @@ mod tests { #[test] fn remove_source_clears_all_keys_for_that_source() { let store = store(); - let star_key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); - let follow_key = EdgeKey::new(nsid("sh.tangled.graph.follow"), at("at://did:plc:lyna")); + let star_key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); + let follow_key = EdgeKey::new(nsid("sh.tangled.graph.follow"), did_subj("did:plc:lyna")); let source = "at://did:plc:nel/sh.tangled.feed.star/r1"; store.add(Edge { kind: nsid("sh.tangled.feed.star"), - subject: at("at://did:plc:abalone"), + subject: did_subj("did:plc:abalone"), source: at(source), + sort_micros: 0, }); store.add(Edge { kind: nsid("sh.tangled.graph.follow"), - subject: at("at://did:plc:lyna"), + subject: did_subj("did:plc:lyna"), source: at(source), + sort_micros: 0, }); assert_eq!(store.count(&star_key), 1); @@ -414,8 +444,8 @@ mod tests { fn upsert_source_replaces_old_edges() { let store = store(); let source = at("at://did:plc:teq/sh.tangled.feed.star/r1"); - let old_subject = at("at://did:plc:abalone"); - let new_subject = at("at://did:plc:uni"); + let old_subject = did_subj("did:plc:abalone"); + let new_subject = did_subj("did:plc:uni"); let kind = nsid("sh.tangled.feed.star"); store.upsert_source( @@ -424,6 +454,7 @@ mod tests { kind: kind.clone(), subject: old_subject.clone(), source: source.clone(), + sort_micros: 0, }], ); assert_eq!( @@ -437,6 +468,7 @@ mod tests { kind: kind.clone(), subject: new_subject.clone(), source: source.clone(), + sort_micros: 0, }], ); assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0); @@ -444,13 +476,14 @@ mod tests { } #[test] - fn list_pages_in_id_order() { + fn list_pages_in_sort_order() { let store = store(); - let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); + let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); (0..5).for_each(|i| { - store.add(star_edge( + store.add(star_edge_at( &format!("at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i]), - "at://did:plc:abalone", + "did:plc:abalone", + 1_000_000 + i as u64 * 1_000_000, )); }); @@ -473,11 +506,11 @@ mod tests { #[test] fn list_exact_fill_signals_exhaustion() { let store = store(); - let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); + let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); (0..2).for_each(|i| { store.add(star_edge( &format!("at://did:plc:{}/sh.tangled.feed.star/r{i}", NAMES[i]), - "at://did:plc:abalone", + "did:plc:abalone", )); }); @@ -490,7 +523,7 @@ mod tests { fn list_on_unknown_key_is_empty() { let store = store(); let page = store.list( - &EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:periwinkle")), + &EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:periwinkle")), PageCursor::Start, limit(10), ); @@ -501,14 +534,14 @@ mod tests { #[test] fn distinct_authors_decreases_when_last_source_from_author_removed() { let store = store(); - let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); + let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone")); let s1 = "at://did:plc:nel/sh.tangled.feed.star/r1"; let s2 = "at://did:plc:nel/sh.tangled.feed.star/r2"; let s3 = "at://did:plc:olaren/sh.tangled.feed.star/r3"; - store.add(star_edge(s1, "at://did:plc:abalone")); - store.add(star_edge(s2, "at://did:plc:abalone")); - store.add(star_edge(s3, "at://did:plc:abalone")); + store.add(star_edge(s1, "did:plc:abalone")); + store.add(star_edge(s2, "did:plc:abalone")); + store.add(star_edge(s3, "did:plc:abalone")); assert_eq!(store.count_distinct_authors(&key), 2); store.remove_source(&at(s1)); @@ -533,10 +566,10 @@ mod tests { #[test] fn cursor_token_round_trip() { - let original = SourceId(0xdead_beef); + let original = PageToken::new(1_730_000_000_000_000); let token = original.encode_token(); - assert_eq!(token.len(), CURSOR_TOKEN_LEN); - assert_eq!(SourceId::decode_token(&token).unwrap(), original); + assert_eq!(token.len(), 13, "TID encoding is 13 base32 chars"); + assert_eq!(PageToken::decode_token(&token).unwrap(), original); } #[test] @@ -546,28 +579,29 @@ mod tests { #[test] fn from_token_some_yields_after() { - let token = SourceId(7).encode_token(); + let original = PageToken::new(1_730_000_000_000_000); + let token = original.encode_token(); assert_eq!( PageCursor::from_token(Some(&token)).unwrap(), - PageCursor::After(SourceId(7)), + PageCursor::After(original), ); } #[test] fn cursor_decode_rejects_malformed() { - let bad = ["", "deadbeef0", "no-hex!!", "zzzzzzzz", "1234567"]; + let bad = ["", "deadbeef0", "no-hex!!", "1234567", "way-too-long-not-a-tid"]; bad.into_iter().for_each(|s| { assert!( - matches!(SourceId::decode_token(s), Err(CursorParseError::Malformed)), - "expected malformed for {s:?}" + matches!(PageToken::decode_token(s), Err(CursorParseError::Malformed)), + "expected malformed for {s:?}", ); }); } #[test] - fn cursor_token_does_not_leak_raw_integer() { - let token = SourceId(42).encode_token(); - assert_ne!(token, "42"); - assert_eq!(token.len(), CURSOR_TOKEN_LEN); + fn cursor_token_is_tid_shape() { + let token = PageToken::new(1_730_000_000_000_000).encode_token(); + assert_eq!(token.len(), 13); + assert!(token.chars().all(|c| c.is_ascii_alphanumeric())); } } diff --git a/crates/ingest/Cargo.toml b/crates/ingest/Cargo.toml index 974afa1..2ecc8a2 100644 --- a/crates/ingest/Cargo.toml +++ b/crates/ingest/Cargo.toml @@ -9,6 +9,7 @@ rust-version.workspace = true bobbin-types = { workspace = true } bobbin-edge-index = { workspace = true } bobbin-record-lru = { workspace = true } +bobbin-resolver = { workspace = true } bobbin-runtime = { workspace = true } bobbin-slingshot-client = { workspace = true } jacquard-common = { workspace = true } diff --git a/crates/ingest/src/lib.rs b/crates/ingest/src/lib.rs index d898b9d..a1fe7d6 100644 --- a/crates/ingest/src/lib.rs +++ b/crates/ingest/src/lib.rs @@ -11,8 +11,9 @@ use bobbin_runtime::{ Clock, Entropy, NetworkError, RuntimeHasher, UnixMicros, WsConn, WsMessage, WsStream, WsTransport, }; +use bobbin_resolver::{NormalizeRepoRefs, decode_canon_or_upgrade_bytes}; use bobbin_types::edges::{Edge, ExtractError, Record}; -use bobbin_types::ids::RepoIdent; +use bobbin_types::ids::{RepoIdent, SubjectRef}; use bobbin_types::record::RecordBody; use bobbin_types::search::{SearchSink, SearchableRecord}; use bytes::Bytes; @@ -897,7 +898,15 @@ async fn commit_stage( resolve_end, } = staged; let commit_start = rt.clock.now_instant(); - commit_pending(pending, &rt.store, &rt.coverage, &*rt.search, &*rt.records).await; + commit_pending( + pending, + &rt.store, + &rt.coverage, + &*rt.search, + &*rt.records, + &rt.resolver, + ) + .await; let commit_end = rt.clock.now_instant(); tracing::trace!( target: "bobbin_ingest::stage", @@ -970,9 +979,21 @@ async fn prepare_record( debug!(collection = %nsid, "create/update missing record body, clearing cache"); return PendingOp::ClearCache { source }; }; - let bytes = Bytes::copy_from_slice(raw.get().as_bytes()); - let parsed = match Record::from_json_bytes(record.collection.as_ref(), &bytes) { - Ok(r) => r, + let wire_bytes = Bytes::copy_from_slice(raw.get().as_bytes()); + let (parsed, bytes) = match decode_canon_or_upgrade_bytes( + record.collection.as_ref(), + &wire_bytes, + ctx.resolver, + ) + .await + { + Ok((parsed, canon_bytes)) => { + let bytes = match canon_bytes { + std::borrow::Cow::Borrowed(_) => wire_bytes, + std::borrow::Cow::Owned(v) => Bytes::from(v), + }; + (parsed, bytes) + } Err(ExtractError::UnknownCollection(name)) => { debug!(collection = %name, "unknown sh.tangled.* collection, clearing cache"); return PendingOp::ClearCache { source }; @@ -986,13 +1007,27 @@ async fn prepare_record( if let Some(shadow) = ctx.shadow { shadow.note_observed(&record.did, &record.rkey).await; } - ctx.resolver + let superseded = ctx + .resolver .observe( record.did.clone(), record.rkey.clone(), repo.repo_did.clone(), ) .await; + if let Some(prior) = superseded { + let prior_uri = format!( + "at://{}/sh.tangled.repo/{}", + prior.owner.as_ref(), + prior.rkey.as_ref(), + ); + if let Ok(prior_at_uri) = AtUri::::new_owned(&prior_uri) { + evict_from_buffer(ctx.buffer, &prior_at_uri).await; + ctx.store.remove_source(&prior_at_uri); + ctx.records.remove(&prior_at_uri); + ctx.search.remove(&prior_at_uri).await; + } + } if let Some(buffer) = ctx.buffer { let drained = buffer.take_observed(&record.did, &record.rkey).await; if !drained.is_empty() { @@ -1019,6 +1054,9 @@ async fn prepare_record( } RecordAction::Delete => { evict_from_buffer(ctx.buffer, &source).await; + if nsid.as_ref() == "sh.tangled.repo" { + ctx.resolver.forget(&record.did, &record.rkey).await; + } PendingOp::Delete { source, nsid } } RecordAction::Other => { @@ -1161,7 +1199,10 @@ async fn collect_unresolved_deps( ) -> Vec { futures::stream::iter(edges) .fold(Vec::new(), |mut acc, edge| async move { - let Some((owner, rkey)) = parse_repo_subject(&edge.subject) else { + let Some(uri) = edge.subject.as_uri() else { + return acc; + }; + let Some((owner, rkey)) = parse_repo_subject_uri(uri) else { return acc; }; if resolver.cached_resolution(&owner, &rkey).await.is_some() { @@ -1193,9 +1234,7 @@ async fn finalize_drained( let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; cache_body(ctx.records, &source, cid, bytes); ctx.store.upsert_source(&source, edges); - if let Some(searchable) = SearchableRecord::try_from_record(parsed) { - ctx.search.upsert(searchable.to_search_doc(&source)).await; - } + index_search(ctx.search, ctx.resolver, &source, parsed).await; } } @@ -1205,6 +1244,7 @@ async fn commit_pending( coverage: &CoverageWatch, search: &S, records: &dyn RecordStore, + resolver: &RepoIdResolver, ) { let Pending { cursor, @@ -1225,9 +1265,7 @@ async fn commit_pending( } => { cache_body(records, &source, cid, bytes); store.upsert_source(&source, edges); - if let Some(searchable) = SearchableRecord::try_from_record(parsed) { - search.upsert(searchable.to_search_doc(&source)).await; - } + index_search(search, resolver, &source, parsed).await; } PendingOp::Delete { source, nsid: _ } => { store.remove_source(&source); @@ -1238,6 +1276,21 @@ async fn commit_pending( coverage.update(|c| c.advance(cursor).maybe_promote(signal)); } +async fn index_search( + search: &S, + resolver: &RepoIdResolver, + source: &AtUri, + parsed: Record, +) { + let Some(searchable) = SearchableRecord::try_from_record(parsed) else { + return; + }; + let Some(searchable) = searchable.normalize(resolver).await else { + return; + }; + search.upsert(searchable.to_search_doc(source)).await; +} + #[cfg(test)] async fn handle_frame( frame: HydrantFrame, @@ -1261,7 +1314,7 @@ async fn handle_frame( }; let pending = prepare_frame(frame, &ctx, now).await; let pending = resolve_pending(pending, &ctx).await; - commit_pending(pending, store, coverage, search, records).await; + commit_pending(pending, store, coverage, search, records, resolver).await; } fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { @@ -1300,7 +1353,10 @@ async fn normalize_subjects( let warming = shadow.is_some() && !coverage.snapshot().is_ready(); futures::stream::iter(edges) .filter_map(|edge| async move { - let Some((owner, rkey)) = parse_repo_subject(&edge.subject) else { + let Some(uri) = edge.subject.as_uri() else { + return Some(edge); + }; + let Some((owner, rkey)) = parse_repo_subject_uri(uri) else { return Some(edge); }; if warming @@ -1313,7 +1369,7 @@ async fn normalize_subjects( } match resolver.resolve(&owner, &rkey).await { Resolution::Mapped(repo_did) => Some(Edge { - subject: owned_did_aturi(&repo_did), + subject: SubjectRef::Did(repo_did), ..edge }), Resolution::NoRepoDid => { @@ -1327,14 +1383,24 @@ async fn normalize_subjects( ); None } - Resolution::Unresolvable => Some(edge), + Resolution::Unresolvable => { + warn!( + target: "bobbin_ingest::normalize", + kind = %edge.kind, + owner = owner.as_ref(), + rkey = rkey.as_ref(), + source = edge.source.as_ref(), + "dropping edge: repo unresolvable, rkey-form subject will not match bare-DID queries", + ); + None + } } }) .collect() .await } -fn parse_repo_subject(uri: &AtUri) -> Option<(Did, Rkey)> { +fn parse_repo_subject_uri(uri: &AtUri) -> Option<(Did, Rkey)> { let collection = uri.collection()?; if collection.as_ref() != "sh.tangled.repo" { return None; @@ -1348,11 +1414,6 @@ fn parse_repo_subject(uri: &AtUri) -> Option<(Did, Rkey< Some((owner, rkey)) } -fn owned_did_aturi(did: &Did) -> AtUri { - AtUri::new_owned(format!("at://{}", did.as_ref())) - .expect("DID concatenated with at:// scheme is a valid AT-URI") -} - fn build_source_uri(r: &RecordFrame) -> Result, IngestError> { Ok(AtUri::from_parts_owned( r.did.as_ref(), @@ -1374,6 +1435,14 @@ mod tests { const VALID_CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; + fn did_subj(s: &str) -> SubjectRef { + SubjectRef::Did(Did::new_owned(s).unwrap()) + } + + fn uri_subj(s: &str) -> SubjectRef { + SubjectRef::Uri(AtUri::new_owned(s).unwrap()) + } + fn fresh() -> (Arc, Arc, Arc) { ( Arc::new(EdgeStore::new(RuntimeHasher::default())), @@ -1463,7 +1532,7 @@ mod tests { .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("sh.tangled.feed.star").unwrap(), - AtUri::new_owned("at://did:plc:abalone").unwrap(), + did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); @@ -1540,12 +1609,8 @@ mod tests { .await; let kind = Nsid::new_static("sh.tangled.feed.star").unwrap(); - let old = bobbin_types::ids::EdgeKey::new( - kind.clone(), - AtUri::new_owned("at://did:plc:abalone").unwrap(), - ); - let new = - bobbin_types::ids::EdgeKey::new(kind, AtUri::new_owned("at://did:plc:uni").unwrap()); + let old = bobbin_types::ids::EdgeKey::new(kind.clone(), did_subj("did:plc:abalone")); + let new = bobbin_types::ids::EdgeKey::new(kind, did_subj("did:plc:uni")); assert_eq!(store.count(&old), 0); assert_eq!(store.count(&new), 1); } @@ -1624,7 +1689,11 @@ mod tests { let cached = lru.get(&source).expect("hydrant cid must seed the lru"); assert_eq!(cached.cid.as_ref(), VALID_CID); let parsed: serde_json::Value = serde_json::from_slice(&cached.value).unwrap(); - assert_eq!(parsed["subjectDid"], "did:plc:abalone"); + assert_eq!( + parsed["subject"]["did"], + "did:plc:abalone", + "legacy wire is upgraded to canon shape before caching so downstream readers see canonical fields" + ); } #[tokio::test] @@ -2268,12 +2337,8 @@ mod tests { .await; let nsid = Nsid::new_static("sh.tangled.feed.star").unwrap(); - let repo_keyed = bobbin_types::ids::EdgeKey::new( - nsid.clone(), - AtUri::new_owned("at://did:plc:abalone").unwrap(), - ); - let owner_keyed = - bobbin_types::ids::EdgeKey::new(nsid, AtUri::new_owned("at://did:plc:nel").unwrap()); + let repo_keyed = bobbin_types::ids::EdgeKey::new(nsid.clone(), did_subj("did:plc:abalone")); + let owner_keyed = bobbin_types::ids::EdgeKey::new(nsid, did_subj("did:plc:nel")); assert_eq!( store.count(&repo_keyed), 1, @@ -2287,7 +2352,7 @@ mod tests { } #[tokio::test] - async fn unresolvable_repo_subject_keeps_at_uri() { + async fn unresolvable_repo_subject_drops_edge() { let (store, cov, resolver) = fresh(); let star: HydrantFrame = parse_frame(json!({ "id": 1, @@ -2319,13 +2384,10 @@ mod tests { .await; let nsid = Nsid::new_static("sh.tangled.feed.star").unwrap(); - let owner_keyed = bobbin_types::ids::EdgeKey::new( - nsid.clone(), - AtUri::new_owned("at://did:plc:nel").unwrap(), - ); + let owner_keyed = bobbin_types::ids::EdgeKey::new(nsid.clone(), did_subj("did:plc:nel")); let uri_keyed = bobbin_types::ids::EdgeKey::new( nsid, - AtUri::new_owned("at://did:plc:nel/sh.tangled.repo/abcabcabcabcz").unwrap(), + uri_subj("at://did:plc:nel/sh.tangled.repo/abcabcabcabcz"), ); assert_eq!( store.count(&owner_keyed), @@ -2334,8 +2396,8 @@ mod tests { ); assert_eq!( store.count(&uri_keyed), - 1, - "an unresolvable repo subject keeps the at-uri so the edge stays queryable", + 0, + "unresolvable rkey-form subject must drop the edge; keeping it would index against a key that never matches bare-DID queries", ); } @@ -2404,10 +2466,9 @@ mod tests { let nsid = Nsid::new_static("sh.tangled.feed.star").unwrap(); let uri_keyed = bobbin_types::ids::EdgeKey::new( nsid.clone(), - AtUri::new_owned("at://did:plc:nel/sh.tangled.repo/abcabcabcabcz").unwrap(), + uri_subj("at://did:plc:nel/sh.tangled.repo/abcabcabcabcz"), ); - let owner_keyed = - bobbin_types::ids::EdgeKey::new(nsid, AtUri::new_owned("at://did:plc:nel").unwrap()); + let owner_keyed = bobbin_types::ids::EdgeKey::new(nsid, did_subj("did:plc:nel")); assert_eq!( store.count(&uri_keyed), 0, @@ -2453,7 +2514,7 @@ mod tests { .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("sh.tangled.feed.star").unwrap(), - AtUri::new_owned("at://did:plc:abalone").unwrap(), + did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); } @@ -2499,11 +2560,117 @@ mod tests { .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("sh.tangled.repo.issue").unwrap(), - AtUri::new_owned("at://did:plc:abalone").unwrap(), + did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); } + #[derive(Default)] + struct RecordingSearchSink { + docs: tokio::sync::Mutex>, + } + + impl SearchSink for RecordingSearchSink { + async fn upsert(&self, doc: bobbin_types::search::SearchDoc) { + self.docs.lock().await.push(doc); + } + async fn remove(&self, _uri: &AtUri) {} + } + + #[tokio::test] + async fn search_index_hydrates_repo_did_via_resolver() { + let (store, cov, resolver) = fresh(); + resolver + .observe( + Did::new_owned("did:plc:nel").unwrap(), + Rkey::new_owned("abcabcabcabcz").unwrap(), + Some(Did::new_owned("did:plc:abalone").unwrap()), + ) + .await; + let search = RecordingSearchSink::default(); + let issue: HydrantFrame = parse_frame(json!({ + "id": 1, + "type": "record", + "record": { + "live": false, + "did": "did:plc:olaren", + "rev": fresh_tid().as_str(), + "collection": "sh.tangled.repo.issue", + "rkey": "abcabcabcabcz", + "action": "create", + "record": { + "$type": "sh.tangled.repo.issue", + "createdAt": "2026-05-01T00:00:00Z", + "title": "bug", + "repo": "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz" + } + } + })); + handle_frame( + issue, + &store, + &cov, + &search, + &NoopRecordStore, + &resolver, + &sys_clock(), + now(), + ) + .await; + let docs = search.docs.lock().await; + assert_eq!(docs.len(), 1, "issue should produce one search doc"); + assert_eq!( + docs[0].repo, + Some(Did::new_owned("did:plc:abalone").unwrap()), + "search doc repo field must be resolved from the observed repo, not left as None", + ); + } + + #[tokio::test] + async fn delete_repo_record_evicts_resolver_cache() { + let (store, cov, resolver) = fresh(); + let owner = Did::new_owned("did:plc:nel").unwrap(); + let rkey = Rkey::new_owned("abcabcabcabcz").unwrap(); + resolver + .observe( + owner.clone(), + rkey.clone(), + Some(Did::new_owned("did:plc:abalone").unwrap()), + ) + .await; + assert!( + resolver.cached_resolution(&owner, &rkey).await.is_some(), + "observe must seed the cache", + ); + let delete: HydrantFrame = parse_frame(json!({ + "id": 1, + "type": "record", + "record": { + "live": false, + "did": owner.as_ref(), + "rev": fresh_tid().as_str(), + "collection": "sh.tangled.repo", + "rkey": rkey.as_ref(), + "action": "delete" + } + })); + handle_frame( + delete, + &store, + &cov, + &NoopSearchSink, + &NoopRecordStore, + &resolver, + &sys_clock(), + now(), + ) + .await; + assert!( + resolver.cached_resolution(&owner, &rkey).await.is_none(), + "deleting the repo record must clear the resolver cache so future observes are not blocked by a stale Authoritative entry", + ); + } + #[tokio::test] async fn cancel_short_circuits_a_hung_ws_connect() { let _listener = tokio::net::TcpListener::bind("127.0.0.1:0") diff --git a/crates/ingest/src/resolver.rs b/crates/ingest/src/resolver.rs index 7895afa..20186a7 100644 --- a/crates/ingest/src/resolver.rs +++ b/crates/ingest/src/resolver.rs @@ -1,656 +1 @@ -use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Duration; - -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 tracing::warn; - -const REPO_COLLECTION: &str = "sh.tangled.repo"; - -#[derive(Clone, Debug, Eq, PartialEq)] -pub enum Resolution { - Mapped(Did), - NoRepoDid, - Unresolvable, -} - -#[derive(Clone, Debug, Eq, PartialEq)] -enum AuthoritativeResolution { - Mapped(Did), - NoRepoDid, -} - -impl AuthoritativeResolution { - fn from_repo_did(repo_did: Option>) -> Self { - match repo_did { - Some(did) => Self::Mapped(did), - None => Self::NoRepoDid, - } - } -} - -#[derive(Clone, Debug, Eq, PartialEq)] -enum CacheEntry { - Authoritative(AuthoritativeResolution), - Provisional(Resolution), -} - -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, - } - } -} - -#[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 { - 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) { - 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, -} - -pub struct RepoIdResolver { - cache: SccMap, - probe: Option, - stats: ResolverStats, -} - -impl RepoIdResolver { - pub fn with_slingshot( - client: SlingshotClient, - clock: Arc, - hasher: RuntimeHasher, - ) -> Self { - Self { - cache: SccMap::with_hasher(hasher), - probe: Some(SlingshotProbe { client, clock }), - stats: ResolverStats::default(), - } - } - - pub fn detached(hasher: RuntimeHasher) -> Self { - Self { - cache: 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, - rkey: &Rkey, - ) -> Option { - let key = RepoIdent::new(owner.clone(), rkey.clone()); - self.cache - .get_async(&key) - .await - .map(|entry| entry.get().clone().into_resolution()) - } - - pub async fn observe( - &self, - owner: Did, - rkey: Rkey, - repo_did: Option>, - ) { - let entry = CacheEntry::Authoritative(AuthoritativeResolution::from_repo_did(repo_did)); - self.cache - .entry_async(RepoIdent::new(owner, rkey)) - .await - .and_modify(|existing| *existing = entry.clone()) - .or_insert(entry); - } - - 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); - } - - pub async fn resolve(&self, owner: &Did, rkey: &Rkey) -> Resolution { - let key = RepoIdent::new(owner.clone(), rkey.clone()); - if let Some(entry) = self.cache.get_async(&key).await { - self.stats.record_hit(); - return entry.get().clone().into_resolution(); - } - let Some(probe) = self.probe.as_ref() else { - self.stats.record_miss(MissKind::NoClient, None); - return Resolution::Unresolvable; - }; - let started = probe.clock.now_instant(); - let nsid: Nsid = 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(), - "slingshot transient failure during repoDID lookup, will retry", - ); - let elapsed = probe.clock.now_instant().duration_since(started); - self.stats - .record_miss(MissKind::Transient, Some(elapsed)); - 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, 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(body: &[u8]) -> Result>, ExtractError> { - match Record::from_json_bytes(REPO_COLLECTION, body)? { - Record::Repo(repo) => Ok(repo.repo_did), - _ => 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 { - Did::new_owned(s).unwrap() - } - - fn rkey(s: &str) -> Rkey { - Rkey::new_owned(s).unwrap() - } - - fn test_clock() -> Arc { - Arc::new(SystemClock::new()) - } - - #[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:abalone")), - ) - .await; - let got = resolver - .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) - .await; - assert_eq!(got, Resolution::Mapped(did("did:plc:abalone"))); - } - - #[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 observation_overwrites_prior_value() { - let resolver = RepoIdResolver::detached(RuntimeHasher::default()); - resolver - .observe( - did("did:plc:nel"), - rkey("abcabcabcabcz"), - Some(did("did:plc:abalone")), - ) - .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:abalone"))) - .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:abalone")), - "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:abalone")), - ) - .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_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_does_not_cache() { - 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(2) - .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_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(2) - .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, 2, - "transport-error retries must count as transient, not as canonical unresolvable", - ); - 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(), 2); - } - - #[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:abalone")), - ) - .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", - ); - } -} +pub use bobbin_resolver::*; diff --git a/crates/ingest/src/warming.rs b/crates/ingest/src/warming.rs index 7c2d5e3..3f6dbc9 100644 --- a/crates/ingest/src/warming.rs +++ b/crates/ingest/src/warming.rs @@ -321,7 +321,7 @@ mod tests { fn make_upsert(source: &str) -> ParkedUpsert { let source_uri = at(source); - let star_json = br#"{"$type":"sh.tangled.feed.star","createdAt":"2026-05-01T00:00:00Z","subjectDid":"did:plc:abalone"}"#; + let star_json = br#"{"$type":"sh.tangled.feed.star","createdAt":"2026-05-01T00:00:00Z","subject":{"$type":"sh.tangled.feed.star#repo","did":"did:plc:abalone"}}"#; let parsed = Record::from_json_bytes("sh.tangled.feed.star", star_json) .expect("star fixture parses"); ParkedUpsert { diff --git a/crates/xrpc/Cargo.toml b/crates/xrpc/Cargo.toml index 1d3eeb4..cd3668d 100644 --- a/crates/xrpc/Cargo.toml +++ b/crates/xrpc/Cargo.toml @@ -9,6 +9,8 @@ rust-version.workspace = true bobbin-types = { workspace = true } bobbin-edge-index = { workspace = true } bobbin-record-lru = { workspace = true } +bobbin-resolver = { workspace = true } +bobbin-runtime = { workspace = true } bobbin-search = { workspace = true } bobbin-slingshot-client = { workspace = true } bobbin-knot-proxy = { workspace = true } diff --git a/crates/xrpc/src/lib.rs b/crates/xrpc/src/lib.rs index 09ceb50..f8f5b81 100644 --- a/crates/xrpc/src/lib.rs +++ b/crates/xrpc/src/lib.rs @@ -1,6 +1,8 @@ use std::future::Future; use std::sync::Arc; +use bobbin_resolver::{NormalizeRepoRefs, upgrade_wire_bytes}; + use axum::{ Router, body::Body, @@ -18,15 +20,17 @@ use axum::{ routing::get, }; use bobbin_edge_index::{ - Coverage, CoverageWatch, CursorParseError, EdgePage, EdgeStore, PageCursor, PageLimit, SourceId, + Coverage, CoverageWatch, CursorParseError, EdgePage, EdgeStore, PageCursor, PageLimit, + PageToken, }; use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; use bobbin_record_lru::RecordStore; +use bobbin_resolver::RepoIdResolver; use bobbin_search::{ SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchReader, }; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; -use bobbin_types::ids::{EdgeKey, nsid_static}; +use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static}; use bobbin_types::record::RecordBody; use bobbin_types::search::SearchableRecord; use bobbin_types::sh_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; @@ -97,6 +101,7 @@ pub struct AppState { pub coverage: Arc, pub knots: Arc, pub search: Arc, + pub resolver: Arc, } impl AppState { @@ -107,6 +112,7 @@ impl AppState { coverage: Arc, knots: Arc, search: Arc, + resolver: Arc, ) -> Self { Self { records, @@ -115,6 +121,7 @@ impl AppState { coverage, knots, search, + resolver, } } } @@ -673,13 +680,6 @@ fn parse_uri(raw: &str) -> Result, XrpcError> { AtUri::::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}"))) } -fn parse_subject_uri(raw: &str) -> Result, XrpcError> { - if let Ok(did) = Did::::new_owned(raw) { - return AtUri::::new_owned(format!("at://{}", did.as_ref())) - .map_err(|e| XrpcError::InvalidParams(format!("uri: {e}"))); - } - parse_uri(raw) -} #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum SubjectShape { @@ -812,16 +812,16 @@ impl MirrorOf for SpindleMemberBy { impl HasSubject for StarRecord { const SHAPE: SubjectShape = - SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string", "sh.tangled.repo"]); + SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]); } impl HasSubject for FollowRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for IssueRecord { - const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); + const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for PullRecord { - const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); + const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for IssueCommentRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue"); @@ -843,7 +843,7 @@ impl HasSubject for PipelineStatusRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.pipeline"); } impl HasSubject for ArtifactRecord { - const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); + const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for KnotMemberRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; @@ -861,7 +861,7 @@ impl HasSubject for RefUpdateRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for CollaboratorRecord { - const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); + const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for IssueStateRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue"); @@ -885,52 +885,64 @@ impl HasSubject for VouchRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; } -fn parse_subject(raw: &RawAtUriParam, shape: SubjectShape) -> Result, XrpcError> { - let uri = parse_subject_uri(raw.as_str())?; +fn parse_subject(raw: &RawAtUriParam, shape: SubjectShape) -> Result { + if let Ok(did) = Did::::new_owned(raw.as_str()) { + return match shape { + SubjectShape::BareDid | SubjectShape::BareDidOrOneOfCollections(_) => { + Ok(SubjectRef::Did(did)) + } + SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!( + "subject must be at:///{expected}/, got bare did" + ))), + SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( + "subject must be at://// with nsid in [{}], got bare did", + allowed.join(", "), + ))), + SubjectShape::AnyAtUri => Err(XrpcError::InvalidParams( + "subject must be at-uri form, got bare did".into(), + )), + }; + } + let uri = parse_uri(raw.as_str())?; if matches!(uri.authority(), AtIdentifier::Handle(_)) { return Err(XrpcError::InvalidParams( "subject authority must be a did, not a handle".into(), )); } - let collection = uri.collection(); - match (shape, collection.as_ref().map(|c| c.as_ref())) { - (SubjectShape::BareDid, None) => Ok(uri), - (SubjectShape::BareDid, Some(c)) => Err(XrpcError::InvalidParams(format!( - "subject must be a bare did at-uri like at://; got path collection {c}" + let Some(collection) = uri.collection() else { + return Err(XrpcError::InvalidParams( + "subject must be a bare did or full at:////".into(), + )); + }; + let c = collection.as_ref(); + match shape { + SubjectShape::BareDid => Err(XrpcError::InvalidParams(format!( + "subject must be a bare did, got at-uri with collection {c}" ))), - (SubjectShape::Collection(expected), Some(c)) if c == expected => { + SubjectShape::Collection(expected) if c == expected => { require_rkey(&uri, expected)?; - Ok(uri) + Ok(SubjectRef::Uri(uri)) } - (SubjectShape::Collection(expected), _) => Err(XrpcError::InvalidParams(format!( - "subject must be at:///{expected}/" + SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!( + "subject must be at:///{expected}/, got collection {c}" ))), - (SubjectShape::BareDidOrOneOfCollections(_), None) => Ok(uri), - (SubjectShape::BareDidOrOneOfCollections(allowed), Some(c)) if allowed.contains(&c) => { + SubjectShape::OneOfCollections(allowed) if allowed.contains(&c) => { require_rkey(&uri, c)?; - Ok(uri) + Ok(SubjectRef::Uri(uri)) } - (SubjectShape::BareDidOrOneOfCollections(allowed), Some(c)) => { - Err(XrpcError::InvalidParams(format!( - "subject must be at:// or at://// with nsid in [{}]; got collection {c}", - allowed.join(", "), - ))) - } - (SubjectShape::OneOfCollections(allowed), Some(c)) if allowed.contains(&c) => { + SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( + "subject must be at://// with nsid in [{}], got collection {c}", + allowed.join(", "), + ))), + SubjectShape::BareDidOrOneOfCollections(allowed) if allowed.contains(&c) => { require_rkey(&uri, c)?; - Ok(uri) + Ok(SubjectRef::Uri(uri)) } - (SubjectShape::OneOfCollections(allowed), Some(c)) => { - Err(XrpcError::InvalidParams(format!( - "subject must be at://// with nsid in [{}]; got collection {c}", - allowed.join(", "), - ))) - } - (SubjectShape::OneOfCollections(allowed), None) => Err(XrpcError::InvalidParams(format!( - "subject must be at://// with nsid in [{}]", + SubjectShape::BareDidOrOneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( + "subject must be a bare did or at://// with nsid in [{}], got collection {c}", allowed.join(", "), ))), - (SubjectShape::AnyAtUri, _) => Ok(uri), + SubjectShape::AnyAtUri => Ok(SubjectRef::Uri(uri)), } } @@ -1027,18 +1039,39 @@ fn verify_type_tag(body: &RecordBody, expected: ExpectedNsid<'_>) -> Result<(), Ok(()) } +async fn deserialize_or_upgrade( + state: &AppState, + nsid: &str, + bytes: &[u8], +) -> Result +where + V: serde::de::DeserializeOwned, +{ + match serde_json::from_slice::(bytes) { + Ok(v) => Ok(v), + Err(canon_err) => match upgrade_wire_bytes(nsid, bytes, &state.resolver).await { + Ok(canon_bytes) => serde_json::from_slice(&canon_bytes) + .map_err(|e| XrpcError::InvalidRecord(e.to_string())), + Err(_) => Err(XrpcError::InvalidRecord(canon_err.to_string())), + }, + } +} + async fn fetch( state: &AppState, uri: &RawAtUriParam, ) -> Result<(Arc, V), XrpcError> where R: XrpcResp, - V: serde::de::DeserializeOwned, + V: serde::de::DeserializeOwned + NormalizeRepoRefs, { let parsed = parse_uri(uri.as_str())?; let (body, _did) = resolve(state, ExpectedNsid::new(R::NSID), parsed).await?; - let value: V = - serde_json::from_slice(&body.value).map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; + let value: V = deserialize_or_upgrade(state, R::NSID, &body.value).await?; + let value = value + .normalize(&state.resolver) + .await + .ok_or_else(|| XrpcError::UpstreamGone(uri.as_str().to_owned()))?; Ok((body, value)) } @@ -1140,7 +1173,7 @@ fn collect_repeated(query: Option<&str>, key: &str) -> Vec { async fn bulk_fetch(state: &AppState, uris: Vec) -> Result, XrpcError> where R: XrpcResp, - V: serde::de::DeserializeOwned + Serialize, + V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs, { if uris.is_empty() { return Err(XrpcError::InvalidParams("at least one uri required".into())); @@ -1157,12 +1190,17 @@ where let items: Vec> = stream::iter(parsed) .map(|uri| async move { match resolve(state, ExpectedNsid::new(R::NSID), uri).await { - Ok((body, _)) => match serde_json::from_slice::(&body.value) { - Ok(value) => Ok(Some(RecordView { - uri: body.uri.clone(), - cid: Some(body.cid.clone()), - value, - })), + Ok((body, _)) => match deserialize_or_upgrade::(state, R::NSID, &body.value).await { + Ok(value) => { + let Some(value) = value.normalize(&state.resolver).await else { + return Ok(None); + }; + Ok(Some(RecordView { + uri: body.uri.clone(), + cid: Some(body.cid.clone()), + value, + })) + } Err(_) => Ok(None), }, Err(XrpcError::NotFound | XrpcError::UpstreamGone(_) | XrpcError::InvalidRecord(_)) => { @@ -1181,7 +1219,7 @@ where async fn list_records(state: &AppState, q: ListQuery) -> Result, XrpcError> where R: XrpcResp + HasSubject, - V: serde::de::DeserializeOwned + Serialize, + V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs, { let subject = parse_subject(&q.subject, R::SHAPE)?; let cursor = parse_cursor(q.cursor.as_deref())?; @@ -1191,20 +1229,26 @@ where let items = stream::iter(items) .map(|uri| async move { let body = resolve_for_view(state, R::NSID, uri).await?; - let value: V = serde_json::from_slice(&body.value) - .map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; - Ok::<_, XrpcError>(RecordView { + let value: V = match deserialize_or_upgrade::(state, R::NSID, &body.value).await { + Ok(v) => v, + Err(_) => return Ok::<_, XrpcError>(None), + }; + let Some(value) = value.normalize(&state.resolver).await else { + return Ok::<_, XrpcError>(None); + }; + Ok::<_, XrpcError>(Some(RecordView { uri: body.uri.clone(), cid: Some(body.cid.clone()), value, - }) + })) }) .buffered(FETCH_CONCURRENCY) + .try_filter_map(|opt| async move { Ok(opt) }) .try_collect() .await?; Ok(ListResponse { items, - cursor: next.map(SourceId::encode_token), + cursor: next.map(PageToken::encode_token), }) } @@ -1223,7 +1267,7 @@ fn count_for( async fn list_mirror(state: &AppState, q: ListQuery) -> Result, XrpcError> where M: MirrorOf, - V: serde::de::DeserializeOwned + Serialize, + V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs, { let subject = parse_subject(&q.subject, M::SHAPE)?; let cursor = parse_cursor(q.cursor.as_deref())?; @@ -1232,21 +1276,28 @@ where let EdgePage { items, next } = state.edges.list(&key, cursor, limit); let items = stream::iter(items) .map(|uri| async move { - let body = resolve_for_view(state, ::NSID, uri).await?; - let value: V = serde_json::from_slice(&body.value) - .map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; - Ok::<_, XrpcError>(RecordView { + let nsid = ::NSID; + let body = resolve_for_view(state, nsid, uri).await?; + let value: V = match deserialize_or_upgrade::(state, nsid, &body.value).await { + Ok(v) => v, + Err(_) => return Ok::<_, XrpcError>(None), + }; + let Some(value) = value.normalize(&state.resolver).await else { + return Ok::<_, XrpcError>(None); + }; + Ok::<_, XrpcError>(Some(RecordView { uri: body.uri.clone(), cid: Some(body.cid.clone()), value, - }) + })) }) .buffered(FETCH_CONCURRENCY) + .try_filter_map(|opt| async move { Ok(opt) }) .try_collect() .await?; Ok(ListResponse { items, - cursor: next.map(SourceId::encode_token), + cursor: next.map(PageToken::encode_token), }) } @@ -1973,8 +2024,19 @@ async fn hydrate_search_hit( Err(XrpcError::UpstreamGone(_)) => return Ok(None), Err(other) => return Err(other), }; - let value = SearchableRecord::from_json_bytes(nsid.as_ref(), &body.value) - .map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; + let value = match SearchableRecord::from_json_bytes(nsid.as_ref(), &body.value) { + Ok(v) => v, + Err(canon_err) => { + let canon_bytes = upgrade_wire_bytes(nsid.as_ref(), &body.value, &state.resolver) + .await + .map_err(|_| XrpcError::InvalidRecord(canon_err.to_string()))?; + SearchableRecord::from_json_bytes(nsid.as_ref(), &canon_bytes) + .map_err(|e| XrpcError::InvalidRecord(e.to_string()))? + } + }; + let Some(value) = value.normalize(&state.resolver).await else { + return Ok(None); + }; Ok(Some(SearchHitView { uri: body.uri.clone(), cid: Some(body.cid.clone()), @@ -2048,7 +2110,11 @@ async fn resolve_knot_target( .map_err(|e| XrpcError::InvalidRecord(format!("decode repo record: {e}")))?; let host = KnotHost::parse(value.knot.as_ref()) .map_err(|e| XrpcError::InvalidRecord(format!("knot field: {e}")))?; - let slug = RepoSlug::new(did.as_ref(), value.name.as_ref()) + let name = value + .name + .as_deref() + .ok_or_else(|| XrpcError::InvalidRecord("repo record missing name".to_string()))?; + let slug = RepoSlug::new(did.as_ref(), name) .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?; Ok((host, slug)) } diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs index 72c8fde..abab05f 100644 --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -1,17 +1,20 @@ use std::sync::Arc; use axum::body::{Body, to_bytes}; -use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, SourceId}; +use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, PageToken}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; +use bobbin_types::ids::SubjectRef; use bobbin_xrpc::{AppState, router}; use futures::stream::{self, StreamExt}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri; use serde_json::{Value, json}; @@ -31,6 +34,12 @@ fn nsid(s: &'static str) -> Nsid { Nsid::new_static(s).unwrap() } +fn subj(s: &str) -> SubjectRef { + Did::::new_owned(s) + .map(SubjectRef::Did) + .unwrap_or_else(|_| SubjectRef::Uri(AtUri::new_owned(s).unwrap())) +} + struct Harness { server: MockServer, edges: Arc, @@ -38,6 +47,12 @@ struct Harness { state: AppState, } +static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + +fn next_sort_micros() -> u64 { + EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed) +} + impl Harness { async fn new() -> Self { let server = MockServer::start().await; @@ -60,6 +75,7 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { server, @@ -72,8 +88,9 @@ impl Harness { fn add_edge(&self, kind: &'static str, subject: &str, source: &str) { self.edges.add(Edge { kind: nsid(kind), - subject: at(subject), + subject: subj(subject), source: at(source), + sort_micros: next_sort_micros(), }); } @@ -133,7 +150,7 @@ async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { fn issue_body(repo_did: &str, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.issue", - "repoDid": repo_did, + "repo": repo_did, "title": title, "createdAt": "2026-05-01T00:00:00Z" }) @@ -143,7 +160,10 @@ fn star_body(subject_did: &str) -> Value { json!({ "$type": "sh.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", - "subjectDid": subject_did + "subject": { + "$type": "sh.tangled.feed.star#repo", + "did": subject_did + } }) } @@ -358,8 +378,8 @@ async fn list_paginates_via_cursor() { .expect("first page must yield a cursor") .to_owned(); assert!( - SourceId::decode_token(&cursor).is_ok(), - "cursor must be an opaque token decodable by SourceId" + PageToken::decode_token(&cursor).is_ok(), + "cursor must be a TID-shaped token" ); let (_, page2) = json_response( @@ -802,7 +822,7 @@ async fn list_item_with_wrong_type_tag_propagates_invalid_record() { } #[tokio::test] -async fn bare_did_endpoints_reject_subject_with_path() { +async fn bare_did_endpoints_reject_at_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ @@ -828,7 +848,7 @@ async fn bare_did_endpoints_reject_subject_with_path() { body["message"] .as_str() .unwrap_or_default() - .contains("bare did at-uri"), + .contains("bare did"), "{endpoint}: {}", body["message"], ); @@ -838,7 +858,7 @@ async fn bare_did_endpoints_reject_subject_with_path() { } #[tokio::test] -async fn repo_pointing_endpoints_accept_repo_uri_subject() { +async fn repo_pointing_endpoints_reject_at_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ @@ -861,11 +881,19 @@ async fn repo_pointing_endpoints_accept_repo_uri_subject() { )) .await .unwrap(); - let (status, _body) = json_response(resp).await; + let (status, body) = json_response(resp).await; assert_eq!( status, - StatusCode::OK, - "{endpoint} must accept a sh.tangled.repo path subject as input shape, even when the index keys edges on the repoDID", + StatusCode::BAD_REQUEST, + "{endpoint} must reject rkey-form subjects since rkeys are unstable; clients must send the repoDID", + ); + assert!( + body["message"] + .as_str() + .unwrap_or_default() + .contains("bare did"), + "{endpoint}: {}", + body["message"], ); } }) @@ -873,36 +901,27 @@ async fn repo_pointing_endpoints_accept_repo_uri_subject() { } #[tokio::test] -async fn repo_pointing_endpoints_reject_unrelated_collection() { +async fn repo_pointing_endpoints_accept_bare_did() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ "sh.tangled.repo.listIssues", + "sh.tangled.repo.countIssues", "sh.tangled.repo.listPulls", + "sh.tangled.repo.countPulls", "sh.tangled.repo.listArtifacts", + "sh.tangled.repo.countArtifacts", ]; stream::iter(cases) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app - .oneshot(list_request( - endpoint, - "at://did:plc:abalone/sh.tangled.knot/k1", - &[], - )) + .oneshot(list_request(endpoint, "did:plc:abalone", &[])) .await .unwrap(); - let (status, body) = json_response(resp).await; - assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); - assert!( - body["message"] - .as_str() - .unwrap_or_default() - .contains("sh.tangled.repo"), - "{endpoint}: {}", - body["message"], - ); + let (status, _body) = json_response(resp).await; + assert_eq!(status, StatusCode::OK, "{endpoint} must accept bare did"); } }) .await; @@ -968,7 +987,7 @@ async fn star_endpoints_reject_unrelated_collection() { assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); let msg = body["message"].as_str().unwrap_or_default(); assert!( - msg.contains("sh.tangled.string") && msg.contains("sh.tangled.repo"), + msg.contains("sh.tangled.string"), "{endpoint}: {msg}", ); } @@ -977,7 +996,7 @@ async fn star_endpoints_reject_unrelated_collection() { } #[tokio::test] -async fn star_endpoints_accept_repo_subject_form() { +async fn star_endpoints_reject_repo_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app @@ -989,8 +1008,13 @@ async fn star_endpoints_accept_repo_subject_form() { .await .unwrap(); let (status, body) = json_response(resp).await; - assert_eq!(status, StatusCode::OK); - assert_eq!(body["count"], 0); + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "rkey-form repo URI must be rejected; clients must send the repoDID directly", + ); + let msg = body["message"].as_str().unwrap_or_default(); + assert!(msg.contains("sh.tangled.string"), "{msg}"); } #[tokio::test] @@ -1051,7 +1075,7 @@ async fn list_pulls_hydrates_via_slingshot_when_edges_present() { "title": "ship it", "createdAt": "2026-05-01T00:00:00Z", "rounds": [], - "target": {"repoDid": target_did, "branch": "main"}, + "target": {"repo": target_did, "branch": "main"}, }), ) .await; @@ -1066,7 +1090,7 @@ async fn list_pulls_hydrates_via_slingshot_when_edges_present() { let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["title"], json!("ship it")); - assert_eq!(items[0]["value"]["target"]["repoDid"], json!(target_did)); + assert_eq!(items[0]["value"]["target"]["repo"], json!(target_did)); } #[tokio::test] @@ -1136,5 +1160,5 @@ async fn extractor_to_xrpc_round_trip_for_star() { ); let items = json["items"].as_array().unwrap(); assert_eq!(items.len(), 1, "expected exactly one star edge"); - assert_eq!(items[0]["value"]["subjectDid"], json!(subject_did)); + assert_eq!(items[0]["value"]["subject"]["did"], json!(subject_did)); } diff --git a/crates/xrpc/tests/bulk.rs b/crates/xrpc/tests/bulk.rs index 37c563c..837edd2 100644 --- a/crates/xrpc/tests/bulk.rs +++ b/crates/xrpc/tests/bulk.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; @@ -44,6 +45,7 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { server, state } } @@ -104,7 +106,7 @@ async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { fn issue_body(repo_did: &str, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.issue", - "repoDid": repo_did, + "repo": repo_did, "title": title, "createdAt": "2026-05-01T00:00:00Z" }) @@ -248,7 +250,7 @@ async fn get_issues_returns_all_resolved_issues() { #[tokio::test] async fn get_pulls_returns_all_resolved_pulls() { let h = Harness::new().await; - let target = "at://did:plc:abalone/sh.tangled.repo/abalone"; + let target = "did:plc:abalone"; h.mount( "did:plc:nel", "sh.tangled.repo.pull", diff --git a/crates/xrpc/tests/cold_start.rs b/crates/xrpc/tests/cold_start.rs index d1165fc..1b62af5 100644 --- a/crates/xrpc/tests/cold_start.rs +++ b/crates/xrpc/tests/cold_start.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; @@ -36,6 +37,7 @@ async fn fresh_app(server_uri: &str) -> AppState { ), Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap()) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ) } @@ -106,7 +108,7 @@ async fn cold_start_serves_all_four_point_lookups() { "i1", json!({ "$type": "sh.tangled.repo.issue", - "repoDid": "did:plc:limpet", + "repo": "did:plc:limpet", "title": "broken", "createdAt": "2026-05-01T00:00:00Z" }), @@ -123,7 +125,7 @@ async fn cold_start_serves_all_four_point_lookups() { "title": "ship", "createdAt": "2026-05-01T00:00:00Z", "rounds": [], - "target": {"repoDid": "did:plc:limpet", "branch": "main"} + "target": {"repo": "did:plc:limpet", "branch": "main"} }), ) .await; diff --git a/crates/xrpc/tests/coverage.rs b/crates/xrpc/tests/coverage.rs index b33119b..a3c8cc3 100644 --- a/crates/xrpc/tests/coverage.rs +++ b/crates/xrpc/tests/coverage.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; @@ -40,6 +41,7 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { coverage, state } } diff --git a/crates/xrpc/tests/extended.rs b/crates/xrpc/tests/extended.rs index 0517e1c..dd67007 100644 --- a/crates/xrpc/tests/extended.rs +++ b/crates/xrpc/tests/extended.rs @@ -4,13 +4,16 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; +use bobbin_types::ids::SubjectRef; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri; use serde_json::{Value, json}; @@ -32,6 +35,12 @@ fn nsid(s: &'static str) -> Nsid { Nsid::new_static(s).unwrap() } +fn subj(s: &str) -> SubjectRef { + Did::::new_owned(s) + .map(SubjectRef::Did) + .unwrap_or_else(|_| SubjectRef::Uri(AtUri::new_owned(s).unwrap())) +} + struct Harness { server: MockServer, edges: Arc, @@ -60,6 +69,7 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { server, @@ -69,10 +79,12 @@ impl Harness { } fn add_edge(&self, kind: &'static str, subject: &str, source: &str) { + static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); self.edges.add(Edge { kind: nsid(kind), - subject: at(subject), + subject: subj(subject), source: at(source), + sort_micros: EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed), }); } @@ -725,7 +737,7 @@ async fn extractor_to_xrpc_round_trip_for_pipeline() { } #[tokio::test] -async fn list_pipelines_falls_back_to_owner_did_for_legacy_records() { +async fn list_pipelines_drops_records_without_resolvable_repo_did() { let h = Harness::new().await; let owner_did = "did:plc:nel"; let spindle_did = "did:plc:lyna"; @@ -754,22 +766,10 @@ async fn list_pipelines_falls_back_to_owner_did_for_legacy_records() { .unwrap(), ) .await; - assert_eq!( - status, - StatusCode::OK, - "owner-DID fallback must hydrate, body was {json}" - ); + assert_eq!(status, StatusCode::OK, "body was {json}"); let items = json["items"].as_array().unwrap(); - assert_eq!(items.len(), 1); - assert_eq!( - items[0]["value"]["triggerMetadata"]["repo"]["did"], - json!(owner_did) - ); assert!( - items[0]["value"]["triggerMetadata"]["repo"] - .get("repoDid") - .is_none(), - "fixture must omit repoDid but got {}", - items[0]["value"]["triggerMetadata"]["repo"] + items.is_empty(), + "pipeline without resolvable repoDid must be dropped, got {json}" ); } diff --git a/crates/xrpc/tests/knot_proxy.rs b/crates/xrpc/tests/knot_proxy.rs index 5097833..47b4a04 100644 --- a/crates/xrpc/tests/knot_proxy.rs +++ b/crates/xrpc/tests/knot_proxy.rs @@ -5,6 +5,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{FailureThreshold, KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; @@ -67,6 +68,7 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { slingshot: slingshot_server, diff --git a/crates/xrpc/tests/search.rs b/crates/xrpc/tests/search.rs index 84b0f89..4c63ea2 100644 --- a/crates/xrpc/tests/search.rs +++ b/crates/xrpc/tests/search.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor}; use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; @@ -63,6 +64,7 @@ impl Harness { .unwrap(), ), search.clone() as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { server, @@ -122,7 +124,7 @@ impl Harness { let uri = format!("at://{did}/sh.tangled.repo.issue/{rkey}"); let value = json!({ "$type": "sh.tangled.repo.issue", - "repoDid": "did:plc:abalone", + "repo": "did:plc:abalone", "title": title, "body": body, "createdAt": "2026-05-01T00:00:00Z" @@ -246,7 +248,7 @@ async fn indexed_issue_hydrates_typed_value() { hits[0]["value"]["title"], json!("barnacle pagination overflow"), ); - assert_eq!(hits[0]["value"]["repoDid"], json!("did:plc:abalone")); + assert_eq!(hits[0]["value"]["repo"], json!("did:plc:abalone")); assert!(hits[0]["score"].as_f64().unwrap() > 0.0); } @@ -482,7 +484,7 @@ async fn second_query_short_circuits_via_lru_without_re_querying_slingshot() { "cid": CID, "value": { "$type": "sh.tangled.repo.issue", - "repoDid": "did:plc:abalone", + "repo": "did:plc:abalone", "title": "kelp", "body": "ocean", "createdAt": "2026-05-01T00:00:00Z"