From ca3ef49ac1a6c9d9e619f8d91bfb8506bb48621b Mon Sep 17 00:00:00 2001 From: dawn Date: Tue, 8 Sep 2026 02:25:16 +0900 Subject: [PATCH] bobbin/{ingest,resolver}: settle repoDid claims by record createdAt Signed-off-by: dawn --- bobbin/crates/ingest/src/lib.rs | 168 ++++++++++++-- bobbin/crates/resolver/src/legacy_upgrade.rs | 5 + bobbin/crates/resolver/src/lib.rs | 220 ++++++++++++++++--- bobbin/crates/xrpc/src/codesearch.rs | 2 + bobbin/crates/xrpc/tests/aggregation.rs | 2 + bobbin/crates/xrpc/tests/bulk.rs | 4 + bobbin/crates/xrpc/tests/cold_start.rs | 9 +- bobbin/crates/xrpc/tests/trending.rs | 2 + 8 files changed, 355 insertions(+), 57 deletions(-) diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs index 0458596bf..fdc4de479 100644 --- a/bobbin/crates/ingest/src/lib.rs +++ b/bobbin/crates/ingest/src/lib.rs @@ -12,7 +12,8 @@ use bobbin_edge_index::{ use bobbin_knot_ingest::{CapabilityGate, KnotRegistry}; use bobbin_record_lru::RecordStore; use bobbin_resolver::{ - IdentityResolver, NormalizeRepoRefs, decode_canon_or_upgrade_bytes, synthesize_created_at, + IdentityResolver, NormalizeRepoRefs, RepoClaim, decode_canon_or_upgrade_bytes, + synthesize_created_at, }; use bobbin_runtime::{ Clock, Entropy, NetworkError, RuntimeHasher, UnixMicros, WsConn, WsMessage, WsStream, @@ -886,15 +887,18 @@ enum PendingOp { source: AtUri, nsid: Nsid, }, - // a native knot is authoritative for this record. any copy indexed - // while the knot was still legacy has to go too - KnotOwned { + // somebody else owns this record, so any copy we indexed earlier has to go too + NotIndexed { source: AtUri, nsid: Nsid, cid: Option>, + detail: &'static str, }, } +const KNOT_AUTHORITATIVE: &str = "the knot is authoritative for this record"; +const STALE_RENAME_ALIAS: &str = "a newer sh.tangled.repo record owns this repoDid"; + struct Prepared { pending: Pending, prepare_start: Instant, @@ -912,7 +916,7 @@ struct Resolved { fn pending_nsid(op: &PendingOp) -> Option<&Nsid> { match op { PendingOp::Upsert(pieces) => Some(&pieces.nsid), - PendingOp::Delete { nsid, .. } | PendingOp::KnotOwned { nsid, .. } => Some(nsid), + PendingOp::Delete { nsid, .. } | PendingOp::NotIndexed { nsid, .. } => Some(nsid), PendingOp::Parked { nsid, .. } => Some(nsid), PendingOp::Noop | PendingOp::Identity(_) @@ -964,11 +968,22 @@ async fn claim_pending( mut pending: Pending, ctx: &PipelineCtx<'_, S>, ) -> Pending { + let mut refused = None; match &mut pending.op { PendingOp::Upsert(pieces) => { evict_from_buffer(ctx.buffer, &pieces.source).await; if let Record::Repo(repo) = &pieces.parsed { - pieces.supersedes = claim_repo(&pieces.source, repo, ctx).await; + match claim_repo(&pieces.source, repo, ctx).await { + RepoClaim::Current { displaced } => pieces.supersedes = displaced, + RepoClaim::Superseded { .. } => { + refused = Some(PendingOp::NotIndexed { + source: pieces.source.clone(), + nsid: pieces.nsid.clone(), + cid: pieces.cid.clone(), + detail: STALE_RENAME_ALIAS, + }); + } + } } } PendingOp::Rejected { source, .. } => evict_from_buffer(ctx.buffer, source).await, @@ -980,35 +995,42 @@ async fn claim_pending( ctx.resolver.forget(&ident.owner, &ident.rkey).await; } } - PendingOp::KnotOwned { source, .. } => evict_from_buffer(ctx.buffer, source).await, + PendingOp::NotIndexed { source, .. } => evict_from_buffer(ctx.buffer, source).await, PendingOp::Noop | PendingOp::Identity(_) | PendingOp::Account(_) | PendingOp::Parked { .. } => {} } + if let Some(op) = refused { + pending.op = op; + } pending } -/// takes the rkey for this repo and hands back the ident it displaced, if any async fn claim_repo( source: &AtUri, repo: &RepoRecord, ctx: &PipelineCtx<'_, S>, -) -> Option { - let ident = repo_ident_of(source)?; +) -> RepoClaim { + let Some(ident) = repo_ident_of(source) else { + return RepoClaim::Current { displaced: None }; + }; if let Some(shadow) = ctx.shadow { shadow.note_observed(&ident.owner, &ident.rkey).await; } - let supersedes = ctx + let claim = ctx .resolver .observe( ident.owner.clone(), ident.rkey.clone(), repo.repo_did.clone(), repo.name.clone(), + &repo.created_at, ) .await; - if let Some(prior) = supersedes.as_ref() + if let RepoClaim::Current { + displaced: Some(prior), + } = &claim && let Some(prior_uri) = repo_ident_uri(prior) { evict_from_buffer(ctx.buffer, &prior_uri).await; @@ -1021,7 +1043,7 @@ async fn claim_repo( None => registry.observe_host(&host), } } - supersedes + claim } async fn resolve_stage( @@ -1180,7 +1202,12 @@ async fn prepare_record( if let Some(registry) = ctx.knot_registry { registry.forget_legacy_member(&source); } - return PendingOp::KnotOwned { source, nsid, cid }; + return PendingOp::NotIndexed { + source, + nsid, + cid, + detail: KNOT_AUTHORITATIVE, + }; } AclDisposition::LegacyMember { host } => { if let Some(registry) = ctx.knot_registry { @@ -1573,17 +1600,19 @@ async fn commit_pending( outcome: RecordOutcome::Deleted, }) } - PendingOp::KnotOwned { source, nsid, cid } => { + PendingOp::NotIndexed { + source, + nsid, + cid, + detail, + } => { delete_record_indexes(store, issue_states, pull_statuses, &source, &nsid); records.remove(&source); search.remove(&source).await; Some(Settlement { source, cid: settle_cid(cid.as_ref()), - outcome: RecordOutcome::rejected( - Rejection::NotIndexed, - Some("the knot is authoritative for this record"), - ), + outcome: RecordOutcome::rejected(Rejection::NotIndexed, Some(detail)), }) } }; @@ -1772,6 +1801,7 @@ mod tests { use bobbin_record_lru::{CacheCapacity, LruRecordStore, NoopRecordStore, RecordStore}; use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, WsConnectFuture, WsTransport}; use bobbin_types::search::NoopSearchSink; + use jacquard_common::types::datetime::Datetime; use jacquard_common::types::handle::Handle; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::tid::Tid; @@ -1989,7 +2019,7 @@ mod tests { let native = prepare_frame(member_frame(1, "aaaaaaaaaaaaz", &native_host), &ctx, now()).await; assert!( - matches!(native.op, PendingOp::KnotOwned { .. }), + matches!(native.op, PendingOp::NotIndexed { .. }), "member record for a native knot must be dropped" ); @@ -3329,6 +3359,92 @@ mod tests { ); } + #[tokio::test] + async fn a_backfilled_rename_alias_cannot_steal_the_repo_did() { + let (store, issue_states, pull_statuses, cov, resolver) = fresh(); + let identity = detached_identity(); + let search = NoopSearchSink; + let records = NoopRecordStore; + let ctx = PipelineCtx { + resolver: &resolver, + identity: &identity, + store: &store, + issue_states: &issue_states, + pull_statuses: &pull_statuses, + coverage: &cov, + records: &records, + search: &search, + shadow: None, + buffer: None, + knot_registry: None, + knot_gate: None, + settlements: None, + }; + let repo_frame = |id: u64, rkey: &str, created_at: &str| { + parse_frame(json!({ + "id": id, + "type": "record", + "record": { + "live": false, + "did": "did:plc:oppi", + "rev": fresh_tid().as_str(), + "collection": "sh.tangled.repo", + "rkey": rkey, + "action": "create", + "record": { + "$type": "sh.tangled.repo", + "createdAt": created_at, + "knot": "knot1.tangled.sh", + "repoDid": "did:plc:abalone" + } + } + })) + }; + // a backfill replays a repo in rkey order, so the alias trails what replaced it + let mut settled = Vec::new(); + for frame in [ + repo_frame(1, "stinkpot", "2026-07-24T14:33:33Z"), + repo_frame(2, "tortu", "2026-07-20T19:26:39Z"), + ] { + let pending = prepare_frame(frame, &ctx, now()).await; + let pending = claim_pending(pending, &ctx).await; + let pending = resolve_pending(pending, &ctx).await; + settled.push(commit_pending(pending, &ctx).await); + } + + let owner = Did::new_owned("did:plc:oppi").unwrap(); + let stinkpot = RepoIdent::new(owner.clone(), rkey("stinkpot")); + assert_eq!( + resolver + .lookup_by_repo_did(&Did::new_owned("did:plc:abalone").unwrap()) + .await, + Some(stinkpot.clone()), + "the newest record owns the repoDid, whichever rkey the replay ends on", + ); + assert_eq!( + resolver.lookup_by_name(&owner, "tortu").await, + Some(stinkpot), + "the old name has to answer with the repo as it stands now", + ); + let key = bobbin_types::ids::EdgeKey::new( + Nsid::new_static("sh.tangled.repo").unwrap(), + did_subj("did:plc:oppi"), + ); + assert_eq!( + store.sources_for(&key), + vec![AtUri::new_owned("at://did:plc:oppi/sh.tangled.repo/stinkpot").unwrap()], + "only the renamed repo belongs in the owner's index", + ); + assert_eq!( + settled.pop().flatten().map(|s| s.outcome), + Some(RecordOutcome::rejected( + Rejection::NotIndexed, + Some(STALE_RENAME_ALIAS) + )), + "the alias is refused on purpose, so a writer polling for it is told", + ); + } + #[tokio::test] async fn unresolvable_repo_subject_drops_edge() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); @@ -3514,6 +3630,7 @@ mod tests { Rkey::new_owned("abcabcabcabcz").unwrap(), Some(Did::new_owned("did:plc:abalone").unwrap()), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let issue: HydrantFrame = parse_frame(json!({ @@ -3583,6 +3700,7 @@ mod tests { Rkey::new_owned("abcabcabcabcz").unwrap(), Some(Did::new_owned("did:plc:abalone").unwrap()), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let search = RecordingSearchSink::default(); @@ -3758,6 +3876,7 @@ mod tests { Rkey::new_owned("core").unwrap(), Some(Did::new_owned("did:plc:abalone").unwrap()), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; handle_frame( @@ -3862,6 +3981,7 @@ mod tests { rkey.clone(), Some(Did::new_owned("did:plc:abalone").unwrap()), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; resolver.observe_rkey(owner.clone(), rkey.clone()).await; @@ -4034,7 +4154,13 @@ mod tests { let slow_rkeys: [Rkey; 2] = [rkey("slowrkeyaa01"), rkey("slowrkeyaa02")]; for r in &fast_rkeys { resolver - .observe(owner.clone(), r.clone(), Some(abalone.clone()), None) + .observe( + owner.clone(), + r.clone(), + Some(abalone.clone()), + None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), + ) .await; } diff --git a/bobbin/crates/resolver/src/legacy_upgrade.rs b/bobbin/crates/resolver/src/legacy_upgrade.rs index 8aa9b7804..0771c7715 100644 --- a/bobbin/crates/resolver/src/legacy_upgrade.rs +++ b/bobbin/crates/resolver/src/legacy_upgrade.rs @@ -500,6 +500,7 @@ mod tests { use bobbin_runtime::RuntimeHasher; use bobbin_types::edges::Record; use jacquard_common::DefaultStr; + use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::recordkey::Rkey; @@ -777,6 +778,7 @@ mod tests { rkey("limpet"), Some(did("did:plc:scallop")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let canon = match decoded { @@ -813,6 +815,7 @@ mod tests { key.clone(), Some(did("did:plc:scallop")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let json = br#"{"$type":"sh.tangled.repo.issue","repo":"at://did:plc:nel/sh.tangled.repo/abcabcabcabcz","title":"t","createdAt":"2026-05-01T00:00:00Z"}"#; @@ -951,6 +954,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:scallop")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let json = br#"{"$type":"sh.tangled.repo.pull","title":"t","createdAt":"2026-05-01T00:00:00Z","rounds":[],"target":{"branch":"main","repo":"at://did:plc:nel/sh.tangled.repo/abcabcabcabcz","repoDid":""}}"#; @@ -1053,6 +1057,7 @@ mod tests { key.clone(), Some(did("did:plc:scallop")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let json = br#"{"$type":"sh.tangled.feed.star","createdAt":"2026-05-01T00:00:00Z","subject":"at://did:plc:nel/sh.tangled.repo/abcabcabcabcz"}"#; diff --git a/bobbin/crates/resolver/src/lib.rs b/bobbin/crates/resolver/src/lib.rs index 81ef3e688..29bad13b6 100644 --- a/bobbin/crates/resolver/src/lib.rs +++ b/bobbin/crates/resolver/src/lib.rs @@ -24,6 +24,7 @@ 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::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; @@ -56,6 +57,25 @@ impl AuthoritativeResolution { } } +#[derive(Clone, Debug, Eq, PartialEq)] +struct RepoDidHolder { + ident: RepoIdent, + created_at: Datetime, +} + +impl RepoDidHolder { + // a backfill replays a repo in rkey order, so the stamp decides and not arrival + fn yields_to(&self, incoming: &Self) -> bool { + self.ident == incoming.ident || incoming.created_at >= self.created_at + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum RepoClaim { + Current { displaced: Option }, + Superseded { by: RepoIdent }, +} + #[derive(Clone, Debug, Eq, PartialEq)] enum CacheEntry { Authoritative(AuthoritativeResolution), @@ -174,7 +194,7 @@ pub(crate) struct SlingshotProbe { pub struct RepoIdResolver { cache: SccMap, - by_repo_did: SccMap, RepoIdent, RuntimeHasher>, + by_repo_did: SccMap, RepoDidHolder, RuntimeHasher>, by_rkey: SccSet, by_name: SccMap<(Did, DefaultStr), RepoIdent, RuntimeHasher>, in_flight: SccMap>, RuntimeHasher>, @@ -235,7 +255,7 @@ impl RepoIdResolver { self.by_repo_did .get_async(repo_did) .await - .map(|e| e.get().clone()) + .map(|e| e.get().ident.clone()) } pub async fn lookup_by_name(&self, owner: &Did, name: &str) -> Option { @@ -243,13 +263,25 @@ impl RepoIdResolver { if let Ok(rkey) = Rkey::new_owned(name) { let ident = RepoIdent::new(owner.clone(), rkey); if self.by_rkey.contains_async(&ident).await { - return Some(ident); + return Some(self.current_record(ident).await); } } - self.by_name + let named = self + .by_name .get_async(&(owner.clone(), DefaultStr::from(name))) .await - .map(|e| e.get().clone()) + .map(|e| e.get().clone())?; + Some(self.current_record(named).await) + } + + // the names a rename leaves behind still have to answer with the repo as it stands + async fn current_record(&self, ident: RepoIdent) -> RepoIdent { + let Some(Resolution::Mapped(repo_did)) = + self.cached_resolution(&ident.owner, &ident.rkey).await + else { + return ident; + }; + self.lookup_by_repo_did(&repo_did).await.unwrap_or(ident) } pub async fn observe_rkey(&self, owner: Did, rkey: Rkey) { @@ -262,7 +294,8 @@ impl RepoIdResolver { rkey: Rkey, repo_did: Option>, name: Option, - ) -> Option { + created_at: &Datetime, + ) -> RepoClaim { let ident = RepoIdent::new(owner, rkey); self.observe_rkey(ident.owner.clone(), ident.rkey.clone()) .await; @@ -280,19 +313,32 @@ impl RepoIdResolver { .and_modify(|existing| *existing = entry.clone()) .or_insert(entry); - let repo_did = repo_did?; - let mut prior: Option = None; + let Some(repo_did) = repo_did else { + return RepoClaim::Current { displaced: None }; + }; + let incoming = RepoDidHolder { + ident, + created_at: created_at.clone(), + }; + let mut outcome = RepoClaim::Current { displaced: None }; self.by_repo_did .entry_async(repo_did) .await - .and_modify(|existing| { - if *existing != ident { - prior = Some(existing.clone()); - *existing = ident.clone(); + .and_modify(|held| { + let yields = held.yields_to(&incoming); + outcome = yields + .then(|| RepoClaim::Current { + displaced: (held.ident != incoming.ident).then(|| held.ident.clone()), + }) + .unwrap_or_else(|| RepoClaim::Superseded { + by: held.ident.clone(), + }); + if yields { + *held = incoming.clone(); } }) - .or_insert(ident); - prior + .or_insert(incoming); + outcome } pub async fn forget(&self, owner: &Did, rkey: &Rkey) { @@ -304,7 +350,7 @@ impl RepoIdResolver { .map(|(_, entry)| entry.into_resolution()); if let Some(Resolution::Mapped(repo_did)) = prior_resolution { self.by_repo_did - .remove_if_async(&repo_did, |existing| *existing == ident) + .remove_if_async(&repo_did, |held| held.ident == ident) .await; } self.by_rkey.remove_async(&ident).await; @@ -489,51 +535,117 @@ mod tests { Arc::new(SystemClock::new()) } + fn made(day: u32) -> Datetime { + Datetime::raw_str(format!("2026-05-{day:02}T00:00:00Z")) + } + #[tokio::test] async fn observation_returns_prior_ident_when_repo_did_moves() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); - let prior = resolver + let claim = resolver .observe( did("did:plc:nel"), rkey("3liuighjy2h22"), Some(did("did:plc:clam")), None, + &made(1), ) .await; - assert!(prior.is_none(), "first observation has no prior"); + assert_eq!( + claim, + RepoClaim::Current { displaced: None }, + "first observation has no prior" + ); - let prior = resolver + let claim = resolver .observe( did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam")), None, + &made(2), ) .await; assert_eq!( - prior, - Some(RepoIdent::new(did("did:plc:nel"), rkey("3liuighjy2h22"))), + claim, + RepoClaim::Current { + displaced: Some(RepoIdent::new(did("did:plc:nel"), rkey("3liuighjy2h22"))), + }, "same repoDID at a new (owner, rkey) returns the prior ident so callers can evict the stale at-uri", ); - let prior = resolver + let claim = resolver .observe( did("did:plc:nel"), rkey("core"), Some(did("did:plc:clam")), None, + &made(2), + ) + .await; + assert_eq!( + claim, + RepoClaim::Current { displaced: None }, + "re-observing the same ident is a no-op" + ); + } + + #[tokio::test] + async fn a_record_older_than_the_holder_cannot_take_the_repo_did() { + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); + let owner = did("did:plc:oppi"); + let repo_did = did("did:plc:clam"); + resolver + .observe( + owner.clone(), + rkey("stinkpot"), + Some(repo_did.clone()), + None, + &made(24), + ) + .await; + let claim = resolver + .observe( + owner.clone(), + rkey("tortu"), + Some(repo_did.clone()), + None, + &made(20), ) .await; - assert!(prior.is_none(), "re-observing the same ident is a no-op"); + + let stinkpot = RepoIdent::new(owner.clone(), rkey("stinkpot")); + assert_eq!( + claim, + RepoClaim::Superseded { + by: stinkpot.clone() + }, + "the older record is the alias a rename left behind", + ); + assert_eq!( + resolver.lookup_by_repo_did(&repo_did).await, + Some(stinkpot.clone()), + ); + assert_eq!( + resolver.lookup_by_name(&owner, "tortu").await, + Some(stinkpot), + "the name a rename left behind has to answer with the repo as it stands now", + ); } #[tokio::test] async fn observation_without_repo_did_does_not_track_reverse() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); - let prior = resolver - .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None, None) + let claim = resolver + .observe( + did("did:plc:nel"), + rkey("abcabcabcabcz"), + None, + None, + &made(1), + ) .await; - assert!(prior.is_none()); + assert_eq!(claim, RepoClaim::Current { displaced: None }); } #[tokio::test] @@ -545,6 +657,7 @@ mod tests { rkey("3liuighjy2h22"), Some(did("did:plc:clam")), None, + &made(1), ) .await; resolver @@ -553,6 +666,7 @@ mod tests { rkey("core"), Some(did("did:plc:clam")), None, + &made(1), ) .await; @@ -560,17 +674,20 @@ mod tests { .forget(&did("did:plc:nel"), &rkey("3liuighjy2h22")) .await; - let prior = resolver + let claim = resolver .observe( did("did:plc:nel"), rkey("core-renamed"), Some(did("did:plc:clam")), None, + &made(1), ) .await; assert_eq!( - prior, - Some(RepoIdent::new(did("did:plc:nel"), rkey("core"))), + claim, + RepoClaim::Current { + displaced: Some(RepoIdent::new(did("did:plc:nel"), rkey("core"))), + }, "stale at-uri's forget must not displace the live owner of did:plc:clam", ); } @@ -622,6 +739,7 @@ mod tests { rkey("3liuighjy2h22"), Some(did("did:plc:clam")), Some(DefaultStr::from("ark")), + &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:nel"), "ark").await; @@ -640,6 +758,7 @@ mod tests { rkey("3liuighjy2h22"), None, Some(DefaultStr::from("ark")), + &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:olaren"), "ark").await; @@ -653,7 +772,7 @@ mod tests { async fn lookup_by_name_prefers_rkey_over_claimed_name() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver - .observe(did("did:plc:nel"), rkey("core"), None, None) + .observe(did("did:plc:nel"), rkey("core"), None, None, &made(1)) .await; resolver .observe( @@ -661,6 +780,7 @@ mod tests { rkey("3liuighjy2h22"), None, Some(DefaultStr::from("core")), + &made(1), ) .await; let got = resolver.lookup_by_name(&did("did:plc:nel"), "core").await; @@ -691,6 +811,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:clam")), None, + &made(1), ) .await; let got = resolver @@ -703,7 +824,13 @@ mod tests { async fn observation_without_repo_did_resolves_no_repo_did() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver - .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None, None) + .observe( + did("did:plc:nel"), + rkey("abcabcabcabcz"), + None, + None, + &made(1), + ) .await; let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) @@ -733,6 +860,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:limpet")), None, + &made(1), ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; @@ -753,7 +881,13 @@ mod tests { async fn lookup_by_repo_did_misses_when_repo_did_was_none() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver - .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None, None) + .observe( + did("did:plc:nel"), + rkey("abcabcabcabcz"), + None, + None, + &made(1), + ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; assert_eq!(got, None); @@ -768,6 +902,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:limpet")), None, + &made(1), ) .await; resolver @@ -776,6 +911,7 @@ mod tests { rkey("xyzxyzxyzxyzx"), Some(did("did:plc:limpet")), None, + &made(1), ) .await; let got = resolver.lookup_by_repo_did(&did("did:plc:limpet")).await; @@ -794,6 +930,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:clam")), None, + &made(1), ) .await; resolver @@ -802,6 +939,7 @@ mod tests { rkey("abcabcabcabcz"), Some(did("did:plc:uni")), None, + &made(1), ) .await; let got = resolver @@ -816,7 +954,13 @@ mod tests { let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver - .observe(owner.clone(), key.clone(), Some(did("did:plc:clam")), None) + .observe( + owner.clone(), + key.clone(), + Some(did("did:plc:clam")), + None, + &made(1), + ) .await; resolver .fill_provisional( @@ -838,7 +982,7 @@ mod tests { let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver - .observe(owner.clone(), key.clone(), None, None) + .observe(owner.clone(), key.clone(), None, None, &made(1)) .await; resolver .fill_provisional( @@ -1180,7 +1324,7 @@ mod tests { ) .await; resolver - .observe(owner.clone(), key.clone(), None, None) + .observe(owner.clone(), key.clone(), None, None, &made(1)) .await; let got = resolver.resolve(&owner, &key).await; assert_eq!( @@ -1196,7 +1340,13 @@ mod tests { let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver - .observe(owner.clone(), key.clone(), Some(did("did:plc:clam")), None) + .observe( + owner.clone(), + key.clone(), + Some(did("did:plc:clam")), + None, + &made(1), + ) .await; assert_eq!( resolver.cached_resolution(&owner, &key).await, diff --git a/bobbin/crates/xrpc/src/codesearch.rs b/bobbin/crates/xrpc/src/codesearch.rs index b068aec49..1b4086c97 100644 --- a/bobbin/crates/xrpc/src/codesearch.rs +++ b/bobbin/crates/xrpc/src/codesearch.rs @@ -215,6 +215,7 @@ mod tests { use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; + use jacquard_common::types::datetime::Datetime; use jacquard_common::types::recordkey::Rkey; use serde_json::{Value, json}; use std::sync::Arc; @@ -342,6 +343,7 @@ mod tests { Rkey::new_owned("r1").unwrap(), Some(did(REPO)), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; } diff --git a/bobbin/crates/xrpc/tests/aggregation.rs b/bobbin/crates/xrpc/tests/aggregation.rs index f0696f33b..9035da0f6 100644 --- a/bobbin/crates/xrpc/tests/aggregation.rs +++ b/bobbin/crates/xrpc/tests/aggregation.rs @@ -18,6 +18,7 @@ use chrono::Utc; use futures::stream::{self, StreamExt}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; @@ -2177,6 +2178,7 @@ async fn profile_activity_returns_one_hydrated_aggregate() { target_rkey, Some(target_repo_did.clone()), Some(DefaultStr::from("core")), + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; h.mount( diff --git a/bobbin/crates/xrpc/tests/bulk.rs b/bobbin/crates/xrpc/tests/bulk.rs index 7e57e906c..6ebf46a89 100644 --- a/bobbin/crates/xrpc/tests/bulk.rs +++ b/bobbin/crates/xrpc/tests/bulk.rs @@ -11,6 +11,7 @@ use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::handle::Handle; use jacquard_common::types::nsid::Nsid; @@ -558,6 +559,7 @@ async fn get_repos_by_repo_dids_returns_resolved_repos() { rkey("abalone"), Some(did("did:plc:limpet")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; h.state @@ -567,6 +569,7 @@ async fn get_repos_by_repo_dids_returns_resolved_repos() { rkey("coral"), Some(did("did:plc:coral")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; h.mount( @@ -615,6 +618,7 @@ async fn get_repos_by_repo_dids_skips_unobserved_dids() { rkey("abalone"), Some(did("did:plc:limpet")), None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; h.mount( diff --git a/bobbin/crates/xrpc/tests/cold_start.rs b/bobbin/crates/xrpc/tests/cold_start.rs index 73191213a..35c343d25 100644 --- a/bobbin/crates/xrpc/tests/cold_start.rs +++ b/bobbin/crates/xrpc/tests/cold_start.rs @@ -12,6 +12,7 @@ use bobbin_xrpc::{AppState, router}; use futures::stream::{self, StreamExt}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; @@ -648,7 +649,13 @@ async fn get_repo_by_repo_did_returns_observed_record() { let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; state .resolver - .observe(owner_did.clone(), rk.clone(), Some(repo_did.clone()), None) + .observe( + owner_did.clone(), + rk.clone(), + Some(repo_did.clone()), + None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), + ) .await; let app = router(state); diff --git a/bobbin/crates/xrpc/tests/trending.rs b/bobbin/crates/xrpc/tests/trending.rs index 561424108..a673a180e 100644 --- a/bobbin/crates/xrpc/tests/trending.rs +++ b/bobbin/crates/xrpc/tests/trending.rs @@ -14,6 +14,7 @@ use bobbin_types::ids::SubjectRef; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; +use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; @@ -114,6 +115,7 @@ impl Harness { record_key.clone(), Some(repo_did.clone()), Some(name.into()), + &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; let uri = format!( -- 2.51.2