diff --git a/Cargo.lock b/Cargo.lock index 0dbea2f09..6ec0cc283 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -845,6 +845,7 @@ name = "bobbin-ingest" version = "0.0.1" dependencies = [ "bobbin-edge-index", + "bobbin-knot-proxy", "bobbin-record-lru", "bobbin-resolver", "bobbin-runtime", @@ -886,8 +887,11 @@ dependencies = [ "jacquard-common", "reqwest 0.13.5", "scc", + "serde", + "serde_json", "thiserror 2.0.18", "tokio", + "tracing", "url", "wiremock", ] diff --git a/bobbin/crates/bobbin-sim/src/runtime.rs b/bobbin/crates/bobbin-sim/src/runtime.rs index 786095a1f..7da72adcd 100644 --- a/bobbin/crates/bobbin-sim/src/runtime.rs +++ b/bobbin/crates/bobbin-sim/src/runtime.rs @@ -127,6 +127,7 @@ impl Sim { warming_buffer: warming_buffer_enabled.then(|| warming_buffer.clone()), settlements: None, analytics: None, + ticket_authorship: None, }; let ingest_config = IngestConfig { hydrant_base, diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs index f57f36d54..51dbdbf21 100644 --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -27,6 +27,9 @@ const KNOWN_KEYS: &[&str] = &[ "hydrant.start_cursor", "ingest.parallelism", "ingest.idle_promote_min_events", + "ingest.author_fill_concurrency", + "ingest.author_fill_max_hold_secs", + "ingest.author_held_docs_max", "backpressure.per_request_anon_bytes", "backpressure.adjust_interval_ms", "backpressure.relieve_below_ratio", @@ -65,6 +68,9 @@ const KNOWN_ENVS: &[&str] = &[ "BOBBIN_START_CURSOR", "BOBBIN_INGEST_PARALLELISM", "BOBBIN_INGEST_IDLE_PROMOTE_MIN_EVENTS", + "BOBBIN_INGEST_AUTHOR_FILL_CONCURRENCY", + "BOBBIN_INGEST_AUTHOR_FILL_MAX_HOLD_SECS", + "BOBBIN_INGEST_AUTHOR_HELD_DOCS_MAX", "BOBBIN_BACKPRESSURE_PER_REQUEST_ANON_BYTES", "BOBBIN_BACKPRESSURE_ADJUST_INTERVAL_MS", "BOBBIN_BACKPRESSURE_RELIEVE_BELOW_RATIO", @@ -241,6 +247,21 @@ pub struct IngestConfig { /// said count. #[config(env = "BOBBIN_INGEST_IDLE_PROMOTE_MIN_EVENTS", default = 256)] pub idle_promote_min_events: u64, + + /// How many repos look up their ticket openers at once. A backfilled ticket + /// only names its latest editor, so the mirror is the only way to learn who opened it. + #[config(env = "BOBBIN_INGEST_AUTHOR_FILL_CONCURRENCY", default = 8)] + pub author_fill_concurrency: usize, + + /// Seconds to keep retrying an unreachable mirror. A starting bobbin waits this + /// long at most, then goes ready with each ticket's latest editor as its author. + #[config(env = "BOBBIN_INGEST_AUTHOR_FILL_MAX_HOLD_SECS", default = 30)] + pub author_fill_max_hold_secs: u64, + + /// How many search docs can wait for their ticket's opener. Past this cap they + /// index under the latest editor, so a mirror outage can't use unbounded memory. + #[config(env = "BOBBIN_INGEST_AUTHOR_HELD_DOCS_MAX", default = 20_000)] + pub author_held_docs_max: usize, } #[derive(Debug, Config)] diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index b156cc3fa..8ddedc987 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -11,9 +11,9 @@ use bobbin_edge_index::{ StreamHealth, }; use bobbin_ingest::{ - IngestConfig, IngestRuntime, RepoIdResolver, WarmingBuffer, run as run_ingest, + IngestConfig, IngestRuntime, RepoIdResolver, TicketAuthorship, WarmingBuffer, run as run_ingest, }; -use bobbin_knot_proxy::{FailureThreshold, MirrorProxy}; +use bobbin_knot_proxy::{FailureThreshold, FillConfig, MirrorProxy, TicketAuthors}; use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore}; use bobbin_resolver::{HydrantClient, IdentityResolver}; use bobbin_runtime::{ @@ -409,6 +409,16 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { config::Moderation::Off => None, }; + let ticket_authors = TicketAuthors::new( + mirror.clone(), + clock.clone(), + hasher.clone(), + FillConfig { + concurrency: cfg.ingest.author_fill_concurrency, + max_hold: Duration::from_secs(cfg.ingest.author_fill_max_hold_secs), + ..FillConfig::default() + }, + ); let ingest_runtime = IngestRuntime { store: edges.clone(), claims: claims.clone(), @@ -428,6 +438,12 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { warming_buffer: Some(warming_buffer), settlements: Some(settlements.clone()), analytics, + ticket_authorship: Some(TicketAuthorship::new( + ticket_authors.clone(), + edges.clone(), + clock.clone(), + cfg.ingest.author_held_docs_max, + )), }; let mut ingest_handle = tokio::spawn(run_ingest(ingest_cfg, ingest_runtime)); let _identity_warmer = tokio::spawn(identity.clone().run_warming(cancel.clone())); @@ -485,6 +501,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { .with_settlements(settlements) .with_invites(invites) .with_claims(claims) + .with_ticket_authors(ticket_authors) .with_stream_health(stream_health) .with_moderation(moderation); let app = router(state).layer(axum::middleware::from_fn(metrics::xrpc)); diff --git a/bobbin/crates/ingest/Cargo.toml b/bobbin/crates/ingest/Cargo.toml index 6551903d4..c7ae7c73f 100644 --- a/bobbin/crates/ingest/Cargo.toml +++ b/bobbin/crates/ingest/Cargo.toml @@ -8,6 +8,7 @@ rust-version.workspace = true [dependencies] bobbin-types = { workspace = true, features = ["streaming"] } bobbin-edge-index = { workspace = true } +bobbin-knot-proxy = { workspace = true } bobbin-record-lru = { workspace = true } bobbin-resolver = { workspace = true } bobbin-runtime = { workspace = true } diff --git a/bobbin/crates/ingest/examples/smoke.rs b/bobbin/crates/ingest/examples/smoke.rs index 49c01a914..7263adbff 100644 --- a/bobbin/crates/ingest/examples/smoke.rs +++ b/bobbin/crates/ingest/examples/smoke.rs @@ -53,6 +53,7 @@ async fn main() { warming_buffer: None, settlements: None, analytics: None, + ticket_authorship: None, }; let task = tokio::spawn(async move { let _ = run(cfg, runtime).await; diff --git a/bobbin/crates/ingest/src/authorship.rs b/bobbin/crates/ingest/src/authorship.rs new file mode 100644 index 000000000..2b441b158 --- /dev/null +++ b/bobbin/crates/ingest/src/authorship.rs @@ -0,0 +1,226 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::Mutex as StdMutex; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use bobbin_edge_index::EdgeStore; +use bobbin_knot_proxy::TicketAuthors; +use bobbin_runtime::Clock; +use bobbin_types::edges::{Edge, ticket_by_edge}; +use bobbin_types::search::{SearchDoc, SearchSink}; +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::string::AtUri; +use tokio::sync::Mutex; +use tokio::time::Instant; + +struct HeldDoc { + doc: SearchDoc, + origin_edge: Option, +} + +#[derive(Default)] +struct Held { + by_repo: HashMap>, +} + +impl Held { + fn remove(&mut self, repo: &str, uri: &str) -> bool { + let Some(docs) = self.by_repo.get_mut(repo) else { + return false; + }; + let removed = docs.remove(uri).is_some(); + if docs.is_empty() { + self.by_repo.remove(repo); + } + removed + } +} + +#[derive(Default)] +struct Deferral { + since: Option, + wanted: bool, +} + +pub struct TicketAuthorship { + authors: Arc, + store: Arc, + clock: Arc, + max_held: usize, + // search writes happen under this lock, or flushing a held doc could overwrite a newer version + held: Mutex, + held_count: AtomicUsize, + deferral: StdMutex, +} + +impl TicketAuthorship { + pub fn new( + authors: Arc, + store: Arc, + clock: Arc, + max_held: usize, + ) -> Arc { + Arc::new(Self { + authors, + store, + clock, + max_held, + held: Mutex::new(Held::default()), + held_count: AtomicUsize::new(0), + deferral: StdMutex::new(Deferral::default()), + }) + } + + pub fn authors(&self) -> &Arc { + &self.authors + } + + /// `editor` is the knot's x-tngl-editor, none for records no knot hosted + pub async fn index( + &self, + search: &S, + source: &AtUri, + mut doc: SearchDoc, + editor: Option>, + first_version: bool, + origin_edge: Option, + ) { + let uri = source.as_ref(); + let Some(repo) = doc.repo.clone() else { + search.upsert(doc).await; + return; + }; + let mut held = self.held.lock().await; + self.unhold(&mut held, repo.as_ref(), uri); + let known = self.authors.get(uri).or_else(|| { + let editor = editor.clone().filter(|_| first_version)?; + self.authors.remember(uri, editor.clone()); + Some(editor) + }); + if let Some(author) = known { + self.list_under(origin_edge.as_ref(), author.clone()); + doc.author = Some(author); + search.upsert(doc).await; + return; + } + if editor.is_none() { + // without an editor the repo did is the only author we have + self.list_under(origin_edge.as_ref(), repo); + search.upsert(doc).await; + return; + } + let filling = self.authors.kick(repo.as_ref()); + if filling && self.held_count.load(Ordering::Acquire) < self.max_held { + held.by_repo + .entry(repo.as_ref().to_owned()) + .or_default() + .insert(uri.to_owned(), HeldDoc { doc, origin_edge }); + self.held_count.fetch_add(1, Ordering::AcqRel); + } else { + search.upsert(doc).await; + } + } + + fn list_under(&self, origin_edge: Option<&Edge>, author: Did) { + if let Some(edge) = origin_edge.and_then(|edge| ticket_by_edge(edge, author)) { + self.store.add(edge); + } + } + + pub async fn remove(&self, search: &S, source: &AtUri) { + let uri = source.as_ref(); + let mut held = self.held.lock().await; + if let Some(repo) = bobbin_types::ids::owner_did_from_aturi(source) { + self.unhold(&mut held, repo.as_ref(), uri); + } + self.authors.forget(uri); + // earlier flushes may have listed it under its author, so clear the edges too + self.store.remove_source(source); + search.remove(source).await; + } + + pub async fn flush(&self, search: &S, repo: &str) { + let mut held = self.held.lock().await; + if self.authors.is_running(repo) { + return; + } + let Some(docs) = held.by_repo.remove(repo) else { + return; + }; + self.held_count.fetch_sub(docs.len(), Ordering::AcqRel); + for ( + uri, + HeldDoc { + mut doc, + origin_edge, + }, + ) in docs + { + if let Some(author) = self.authors.get(&uri) { + self.list_under(origin_edge.as_ref(), author.clone()); + doc.author = Some(author); + } + search.upsert(doc).await; + } + } + + pub async fn flush_idle(&self, search: &S) { + let repos: Vec = self.held.lock().await.by_repo.keys().cloned().collect(); + for repo in repos { + self.flush(search, &repo).await; + } + } + + fn unhold(&self, held: &mut Held, repo: &str, uri: &str) { + if held.remove(repo, uri) { + self.held_count.fetch_sub(1, Ordering::AcqRel); + } + } + + fn blocked(&self) -> bool { + self.authors.any_running() || self.held_count.load(Ordering::Acquire) > 0 + } + + fn expired(&self, since: Instant) -> bool { + self.clock.now_instant().duration_since(since) >= self.authors.max_hold() + } + + /// stops waiting one max hold after the first time it was blocked + pub fn defer_promotion(&self) -> bool { + let mut deferral = self.deferral.lock().expect("deferral mutex poisoned"); + if !self.blocked() { + deferral.since = None; + return false; + } + let since = *deferral + .since + .get_or_insert_with(|| self.clock.now_instant()); + if self.expired(since) { + return false; + } + deferral.wanted = true; + true + } + + pub fn release_promotion(&self) -> bool { + let mut deferral = self.deferral.lock().expect("deferral mutex poisoned"); + if !deferral.wanted { + return false; + } + let blocked = self.blocked() && !deferral.since.is_some_and(|since| self.expired(since)); + if blocked { + return false; + } + deferral.wanted = false; + true + } + + pub fn promotion_deadline(&self) -> Option { + let deferral = self.deferral.lock().expect("deferral mutex poisoned"); + deferral + .since + .filter(|_| deferral.wanted) + .map(|since| since + self.authors.max_hold()) + } +} diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs index 6ba363326..fe9a68456 100644 --- a/bobbin/crates/ingest/src/lib.rs +++ b/bobbin/crates/ingest/src/lib.rs @@ -17,7 +17,7 @@ use bobbin_runtime::{ Clock, Entropy, NetworkError, RuntimeHasher, UnixMicros, WsConn, WsMessage, WsStream, WsTransport, }; -use bobbin_types::edges::{Edge, ExtractError, Record}; +use bobbin_types::edges::{Edge, ExtractError, Record, ticket_edge, ticket_editor}; use bobbin_types::ids::{RepoIdent, SubjectRef, owner_did_from_aturi}; use bobbin_types::org_tangled::repo::manifest::Manifest; use bobbin_types::record::RecordBody; @@ -32,6 +32,7 @@ use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::{AtStrError, AtUri, Cid}; use jacquard_common::types::tid::Tid; use thiserror::Error; +use tokio::sync::broadcast; use tokio::time::Instant; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; @@ -39,6 +40,7 @@ use tracing::{debug, info, warn}; use url::Url; mod analytics; +mod authorship; mod frame; mod metrics; pub mod moderation; @@ -77,6 +79,7 @@ pub const DEFAULT_IDLE_PROMOTE_MIN_EVENTS: NonZeroU64 = match NonZeroU64::new(25 }; pub use analytics::{Analytics, PosthogConfig}; +pub use authorship::TicketAuthorship; #[derive(Clone, Debug)] pub struct IngestConfig { @@ -264,6 +267,7 @@ pub struct IngestRuntime { pub warming_buffer: Option>, pub settlements: Option>, pub analytics: Option>, + pub ticket_authorship: Option>, } impl Clone for IngestRuntime { @@ -287,6 +291,7 @@ impl Clone for IngestRuntime { warming_buffer: self.warming_buffer.clone(), settlements: self.settlements.clone(), analytics: self.analytics.clone(), + ticket_authorship: self.ticket_authorship.clone(), } } } @@ -305,6 +310,7 @@ impl IngestRuntime { buffer: self.warming_buffer.as_deref(), settlements: self.settlements.as_deref(), analytics: self.analytics.as_deref(), + authorship: self.ticket_authorship.as_deref(), } } } @@ -321,6 +327,7 @@ struct PipelineCtx<'a, S: SearchSink + 'static> { buffer: Option<&'a WarmingBuffer>, settlements: Option<&'a Settlements>, analytics: Option<&'a Analytics>, + authorship: Option<&'a TicketAuthorship>, } pub async fn run( @@ -329,7 +336,11 @@ pub async fn run( ) -> Result<(), IngestError> { let frames = FrameCounter::default(); let warming_flusher = spawn_warming_flusher(&runtime); + let author_flusher = spawn_author_flusher(&runtime); let result = run_inner(config, &runtime, &frames).await; + if let Some(handle) = author_flusher { + handle.abort(); + } if let Some(handle) = warming_flusher && let Err(join) = handle.await { @@ -437,6 +448,37 @@ fn spawn_warming_flusher( })) } +fn spawn_author_flusher( + runtime: &IngestRuntime, +) -> Option> { + let rt = runtime.clone(); + let authorship = rt.ticket_authorship.clone()?; + Some(tokio::spawn(async move { + let mut done = authorship.authors().subscribe(); + loop { + tokio::select! { + biased; + _ = rt.cancel.cancelled() => return, + _ = wait_until(authorship.promotion_deadline(), rt.clock.as_ref()) => {} + repo = done.recv() => match repo { + Ok(repo) => authorship.flush(&*rt.search, &repo).await, + Err(broadcast::error::RecvError::Lagged(_)) => { + authorship.flush_idle(&*rt.search).await; + } + Err(broadcast::error::RecvError::Closed) => return, + }, + } + if authorship.release_promotion() { + rt.coverage.update(|c| c.force_ready()); + info!( + target: "bobbin_ingest::coverage", + "ticket authors landed or ran out of hold time, promoting coverage to ready", + ); + } + } + })) +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum FlushReason { Ready, @@ -527,7 +569,10 @@ fn promote_if_caught_up( delivered: FramesDelivered, ) { let snap = runtime.coverage.snapshot(); - if snap.is_ready() || snap.events_processed() < min_events.get() { + if snap.is_ready() + || snap.events_processed() < min_events.get() + || authors_hold_promotion(runtime) + { return; } runtime.coverage.update(|c| c.force_ready()); @@ -578,9 +623,16 @@ async fn fetch_replay_head( } } +fn authors_hold_promotion(runtime: &IngestRuntime) -> bool { + runtime + .ticket_authorship + .as_deref() + .is_some_and(TicketAuthorship::defer_promotion) +} + fn promote_at_head(runtime: &IngestRuntime, head: HydrantCursor) { let snap = runtime.coverage.snapshot(); - if snap.is_ready() || snap.last_cursor() < head { + if snap.is_ready() || snap.last_cursor() < head || authors_hold_promotion(runtime) { return; } runtime.coverage.update(|c| c.force_ready()); @@ -1286,6 +1338,7 @@ async fn prepare_record( cid: record.cid, rev: record.rev, edges, + first_version: record.action == RecordAction::Create && record.live, })) } RecordAction::Delete => PendingOp::Delete { @@ -1385,6 +1438,7 @@ struct UpsertPieces { cid: Option>, rev: Tid, edges: Vec, + first_version: bool, } impl From for UpsertPieces { @@ -1397,6 +1451,7 @@ impl From for UpsertPieces { cid: u.cid, rev: u.rev, edges: u.edges, + first_version: false, } } } @@ -1496,7 +1551,7 @@ async fn finalize_drained( cache_body(ctx.records, &source, cid, bytes); // parked records wait on legacy repo at-uris, so they are never manifests or declarations upsert_record_indexes(ctx.store, ctx.claims, &source, edges, &parsed); - index_search(ctx.search, ctx.resolver, &source, parsed).await; + index_upsert(ctx, &source, parsed, false, None).await; let settled = Settlement { source, cid: settled_cid, @@ -1527,7 +1582,7 @@ async fn commit_pending( let claims = ctx.claims; let mut transitions = Vec::new(); let (coverage, search, records) = (ctx.coverage, ctx.search, ctx.records); - let (resolver, identity) = (ctx.resolver, ctx.identity); + let identity = ctx.identity; let Pending { cursor, signal, @@ -1576,11 +1631,13 @@ async fn commit_pending( cid, rev: _, edges, + first_version, } = *pieces; let settled_cid = settle_cid(cid.as_ref()); cache_body(records, &source, cid, bytes); + let origin_edge = ticket_edge(&edges).cloned(); transitions = upsert_record_indexes(store, claims, &source, edges, &parsed); - index_search(search, resolver, &source, parsed).await; + index_upsert(ctx, &source, parsed, first_version, origin_edge).await; Some(Settlement { source, cid: settled_cid, @@ -1590,7 +1647,10 @@ async fn commit_pending( PendingOp::Delete { source, nsid, .. } => { delete_record_indexes(store, claims, &source, &nsid); records.remove(&source); - search.remove(&source).await; + match ctx.authorship { + Some(authorship) => authorship.remove(search, &source).await, + None => search.remove(&source).await, + } Some(Settlement { source, cid: None, @@ -1598,23 +1658,48 @@ async fn commit_pending( }) } }; - coverage.update(|c| c.advance(cursor).maybe_promote(signal)); + let hold = signal.caught_up() + && !coverage.snapshot().is_ready() + && ctx + .authorship + .is_some_and(TicketAuthorship::defer_promotion); + coverage.update(|c| { + let advanced = c.advance(cursor); + if hold { + advanced + } else { + advanced.maybe_promote(signal) + } + }); (settled, transitions) } -async fn index_search( - search: &S, - resolver: &RepoIdResolver, +async fn index_upsert( + ctx: &PipelineCtx<'_, S>, source: &AtUri, parsed: Record, + first_version: bool, + origin_edge: Option, ) { + let editor = match &parsed { + Record::Ticket(ticket) => Some(ticket_editor(ticket)), + _ => None, + }; let Some(searchable) = SearchableRecord::try_from_record(parsed) else { return; }; - let Some(searchable) = searchable.normalize(resolver).await else { + let Some(searchable) = searchable.normalize(ctx.resolver).await else { return; }; - search.upsert(searchable.to_search_doc(source)).await; + let doc = searchable.to_search_doc(source); + match (ctx.authorship, editor) { + (Some(authorship), Some(editor)) => { + authorship + .index(ctx.search, source, doc, editor, first_version, origin_edge) + .await; + } + _ => ctx.search.upsert(doc).await, + } } #[cfg(test)] @@ -1644,6 +1729,7 @@ async fn handle_frame( buffer: None, settlements: None, analytics: None, + authorship: None, }; let pending = claim_pending(prepare_frame(frame, &ctx, now).await, &ctx).await; let pending = resolve_pending(pending, &ctx, None).await; @@ -1761,6 +1847,7 @@ fn build_source_uri(r: &RecordFrame) -> Result, IngestError> { mod tests { use super::*; use bobbin_edge_index::{Coverage, TicketStateKind}; + use bobbin_knot_proxy::TicketAuthors; use bobbin_record_lru::{CacheCapacity, LruRecordStore, NoopRecordStore, RecordStore}; use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, WsConnectFuture, WsTransport}; use bobbin_slingshot_client::SlingshotClient; @@ -2086,6 +2173,7 @@ mod tests { buffer: None, settlements: None, analytics: None, + authorship: None, }; let mk = |live: bool| -> HydrantFrame { parse_frame(json!({ @@ -2759,6 +2847,7 @@ mod tests { buffer: None, settlements: None, analytics: None, + authorship: None, }; let apply = async |frame: HydrantFrame| { let pending = prepare_frame(frame, &ctx, now()).await; @@ -2822,6 +2911,7 @@ mod tests { buffer: None, settlements: None, analytics: None, + authorship: None, }; let olaren = Did::new_static("did:plc:olaren").unwrap(); let apply = async |frame: HydrantFrame| { @@ -2897,6 +2987,7 @@ mod tests { buffer: None, settlements: None, analytics: None, + authorship: None, }; let star = |id: u64, rkey: &str| { parse_frame(json!({ @@ -2968,6 +3059,7 @@ mod tests { buffer: None, settlements: None, analytics: None, + authorship: None, }; let pending = prepare_frame( parse_frame(json!({ @@ -3060,6 +3152,7 @@ mod tests { warming_buffer: None, settlements: None, analytics: None, + ticket_authorship: None, } } @@ -3989,6 +4082,7 @@ mod tests { warming_buffer: None, settlements: None, analytics: None, + ticket_authorship: None, }; let parallelism = 4usize; @@ -4063,6 +4157,7 @@ mod tests { IngestRuntime { settlements: Some(watch.clone()), analytics: None, + ticket_authorship: None, ..fresh_runtime(CancellationToken::new()) } } @@ -4227,4 +4322,494 @@ mod tests { run_stages(repo, &rt).await; assert_eq!(published(&watch, STAR_URI), Some(RecordOutcome::Indexed)); } + + #[derive(Debug, PartialEq)] + enum IndexOp { + Upsert { + uri: String, + author: Option, + state: String, + }, + Remove(String), + } + + #[derive(Default)] + struct LedgerSink { + ops: std::sync::Mutex>, + } + + impl LedgerSink { + fn ops(&self) -> Vec { + std::mem::take(&mut *self.ops.lock().unwrap()) + } + + fn upserts(&self) -> usize { + let ops = self.ops.lock().unwrap(); + ops.iter() + .filter(|op| matches!(op, IndexOp::Upsert { .. })) + .count() + } + } + + impl SearchSink for LedgerSink { + async fn upsert(&self, doc: bobbin_types::search::SearchDoc) { + self.ops.lock().unwrap().push(IndexOp::Upsert { + uri: doc.uri.to_string(), + author: doc.author.map(|d| d.to_string()), + state: doc.ticket.map(|t| t.state).unwrap_or_default(), + }); + } + + async fn remove(&self, uri: &AtUri) { + self.ops + .lock() + .unwrap() + .push(IndexOp::Remove(uri.to_string())); + } + } + + const TICKET_REPO: &str = "did:plc:abalone"; + const MAINTAINER: &str = "did:plc:maintainer"; + + fn ticket_uri(rkey: &str) -> String { + format!("at://{TICKET_REPO}/org.tangled.track.ticket/{rkey}") + } + + struct Frames { + id: u64, + live: bool, + action: &'static str, + editor: Option<&'static str>, + state: &'static str, + } + + impl Frames { + fn backfilled_create(editor: &'static str) -> Self { + Self { + id: 1, + live: false, + action: "create", + editor: Some(editor), + state: "open", + } + } + + fn live_create(editor: &'static str) -> Self { + Self { + live: true, + ..Self::backfilled_create(editor) + } + } + + fn update(editor: &'static str, state: &'static str) -> Self { + Self { + id: 2, + live: true, + action: "update", + editor: Some(editor), + state, + } + } + + fn frame(&self, rkey: &str) -> HydrantFrame { + let mut record = json!({ + "$type": "org.tangled.track.ticket", + "createdAt": "2026-05-01T00:00:00Z", + "title": { "$type": "org.tangled.markup.markdown#inline", "text": "bug" }, + "state": self.state, + }); + if let Some(editor) = self.editor { + record["x-tngl-editor"] = json!(editor); + } + parse_frame(json!({ + "id": self.id, + "type": "record", + "record": { + "live": self.live, + "did": TICKET_REPO, + "rev": fresh_tid().as_str(), + "collection": "org.tangled.track.ticket", + "rkey": rkey, + "action": self.action, + "cid": VALID_CID, + "record": record, + } + })) + } + + fn delete(id: u64) -> Self { + Self { + id, + live: true, + action: "delete", + editor: None, + state: "open", + } + } + } + + struct Authored { + rt: IngestRuntime, + sink: Arc, + authors: Arc, + mirror: MockServer, + } + + fn origin_page(origins: &[(&str, &str)]) -> serde_json::Value { + json!({ + "repo": TICKET_REPO, + "collection": "org.tangled.track.ticket", + "origins": origins + .iter() + .map(|(rkey, author)| json!({ + "rkey": rkey, + "author": author, + "changeId": "abc", + "createdAt": "2026-05-01T00:00:00Z", + })) + .collect::>(), + }) + } + + impl Authored { + async fn new(max_hold: Duration, max_held: usize) -> Self { + let mirror = MockServer::start().await; + let clock: Arc = Arc::new(SystemClock::new()); + let proxy = bobbin_knot_proxy::MirrorProxy::new( + &Url::parse(&mirror.uri()).unwrap(), + clock.clone(), + RuntimeHasher::default(), + ) + .unwrap(); + let authors = TicketAuthors::new( + Some(Arc::new(proxy)), + clock.clone(), + RuntimeHasher::default(), + bobbin_knot_proxy::FillConfig { + max_hold, + ..Default::default() + }, + ); + let sink = Arc::new(LedgerSink::default()); + let fresh = fresh_runtime(CancellationToken::new()); + let rt = IngestRuntime { + store: fresh.store.clone(), + claims: fresh.claims, + coverage: fresh.coverage, + search: sink.clone(), + records: fresh.records, + resolver: fresh.resolver, + identity: fresh.identity, + clock: clock.clone(), + entropy: fresh.entropy, + stream_health: fresh.stream_health, + ws: fresh.ws, + hydrant: None, + cancel: fresh.cancel, + disconnects: None, + warming_shadow: None, + warming_buffer: None, + settlements: None, + analytics: None, + ticket_authorship: Some(TicketAuthorship::new( + authors.clone(), + fresh.store.clone(), + clock, + max_held, + )), + }; + Self { + rt, + sink, + authors, + mirror, + } + } + + async fn mirror_answers(&self, response: ResponseTemplate) { + Mock::given(method("GET")) + .and(path("/xrpc/org.tangled.temp.repo.listRecordOrigins")) + .respond_with(response) + .mount(&self.mirror) + .await; + } + + async fn mirror_calls(&self) -> usize { + self.mirror.received_requests().await.unwrap().len() + } + + async fn feed(&self, rkey: &str, frames: Frames) { + self.feed_at(frames.frame(rkey), None).await; + } + + async fn feed_at(&self, frame: HydrantFrame, head: Option) { + let staged = prep_stage(frame, self.rt.clone()).await; + let staged = claim_stage(staged, self.rt.clone()).await; + let staged = resolve_stage(staged, self.rt.clone(), head).await; + commit_stage(staged, self.rt.clone(), 1, head).await; + } + + async fn wait_for_upserts(&self, count: usize) { + tokio::time::timeout(Duration::from_secs(10), async { + while self.sink.upserts() < count { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the held docs never reached the index"); + } + } + + fn listed_under(t: &Authored, author: &str) -> u64 { + let key = bobbin_types::ids::EdgeKey::new( + Nsid::new_static(bobbin_types::edges::TICKET_ISSUE_BY_KIND).unwrap(), + did_subj(author), + ); + t.rt.store.count(&key) + } + + fn upsert(rkey: &str, author: &str, state: &str) -> IndexOp { + IndexOp::Upsert { + uri: ticket_uri(rkey), + author: Some(author.to_owned()), + state: state.to_owned(), + } + } + + #[tokio::test] + async fn a_live_create_names_its_author_without_asking_the_mirror() { + let t = Authored::new(Duration::from_secs(1), 100).await; + t.feed("t1", Frames::live_create("did:plc:alice")).await; + assert_eq!(t.sink.ops(), vec![upsert("t1", "did:plc:alice", "open")]); + assert_eq!( + t.authors.get(&ticket_uri("t1")).unwrap().as_ref(), + "did:plc:alice" + ); + assert_eq!(t.mirror_calls().await, 0); + assert_eq!(listed_under(&t, "did:plc:alice"), 1); + assert_eq!(listed_under(&t, TICKET_REPO), 0); + } + + #[tokio::test] + async fn a_maintainers_update_keeps_the_original_author_in_search() { + let t = Authored::new(Duration::from_secs(1), 100).await; + t.feed("t1", Frames::live_create("did:plc:alice")).await; + t.feed("t1", Frames::update(MAINTAINER, "closed")).await; + assert_eq!( + t.sink.ops(), + vec![ + upsert("t1", "did:plc:alice", "open"), + upsert("t1", "did:plc:alice", "closed"), + ] + ); + assert_eq!(t.mirror_calls().await, 0); + assert_eq!(listed_under(&t, "did:plc:alice"), 1); + assert_eq!(listed_under(&t, MAINTAINER), 0, "the closer opened nothing"); + } + + #[tokio::test] + async fn a_ticket_without_an_editor_is_authored_by_its_repo() { + let t = Authored::new(Duration::from_secs(1), 100).await; + let frames = Frames { + editor: None, + ..Frames::backfilled_create("unused") + }; + t.feed("t1", frames).await; + assert_eq!(t.sink.ops(), vec![upsert("t1", TICKET_REPO, "open")]); + assert_eq!( + t.mirror_calls().await, + 0, + "nothing to fill for a pds ticket" + ); + assert_eq!(listed_under(&t, TICKET_REPO), 1); + } + + #[tokio::test] + async fn backfilled_tickets_wait_for_one_fill_per_repo() { + let t = Authored::new(Duration::from_secs(5), 100).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[ + ("t1", "did:plc:alice"), + ("t2", "did:plc:bob"), + ])) + .set_delay(Duration::from_millis(200)), + ) + .await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + t.feed("t2", Frames::backfilled_create(MAINTAINER)).await; + assert!(t.sink.ops().is_empty(), "held until the fill lands"); + assert_eq!( + listed_under(&t, MAINTAINER), + 0, + "never under a provisional author" + ); + + t.wait_for_upserts(2).await; + let mut ops = t.sink.ops(); + ops.sort_by_key(|op| format!("{op:?}")); + assert_eq!( + ops, + vec![ + upsert("t1", "did:plc:alice", "open"), + upsert("t2", "did:plc:bob", "open"), + ] + ); + assert_eq!(t.mirror_calls().await, 1); + assert_eq!(listed_under(&t, "did:plc:alice"), 1); + assert_eq!(listed_under(&t, "did:plc:bob"), 1); + assert_eq!(listed_under(&t, MAINTAINER), 0); + flusher.abort(); + } + + #[tokio::test] + async fn a_delete_unlists_the_ticket_from_its_author() { + let t = Authored::new(Duration::from_secs(1), 100).await; + t.feed("t1", Frames::live_create("did:plc:alice")).await; + assert_eq!(listed_under(&t, "did:plc:alice"), 1); + t.feed("t1", Frames::delete(2)).await; + assert_eq!(listed_under(&t, "did:plc:alice"), 0); + } + + #[tokio::test] + async fn a_newer_version_replaces_the_held_doc() { + let t = Authored::new(Duration::from_secs(5), 100).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[("t1", "did:plc:alice")])) + .set_delay(Duration::from_millis(200)), + ) + .await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + t.feed("t1", Frames::update(MAINTAINER, "closed")).await; + + t.wait_for_upserts(1).await; + assert_eq!(t.sink.ops(), vec![upsert("t1", "did:plc:alice", "closed")]); + flusher.abort(); + } + + #[tokio::test] + async fn a_version_indexed_straight_away_drops_the_stale_held_doc() { + let t = Authored::new(Duration::from_secs(5), 100).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[("t1", "did:plc:alice")])) + .set_delay(Duration::from_millis(300)), + ) + .await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + t.authors + .remember(&ticket_uri("t1"), Did::new_owned("did:plc:alice").unwrap()); + t.feed("t1", Frames::update(MAINTAINER, "closed")).await; + + t.authors.wait(TICKET_REPO, Duration::from_secs(5)).await; + tokio::time::sleep(Duration::from_millis(150)).await; + assert_eq!(t.sink.ops(), vec![upsert("t1", "did:plc:alice", "closed")]); + flusher.abort(); + } + + #[tokio::test] + async fn a_delete_drops_the_held_doc() { + let t = Authored::new(Duration::from_secs(5), 100).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[("t1", "did:plc:alice")])) + .set_delay(Duration::from_millis(200)), + ) + .await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + t.feed("t1", Frames::delete(2)).await; + + t.authors.wait(TICKET_REPO, Duration::from_secs(5)).await; + tokio::time::sleep(Duration::from_millis(150)).await; + assert_eq!(t.sink.ops(), vec![IndexOp::Remove(ticket_uri("t1"))]); + flusher.abort(); + } + + #[tokio::test] + async fn a_failing_mirror_leaves_provisional_authors_after_the_hold() { + let t = Authored::new(Duration::from_millis(500), 100).await; + t.mirror_answers(ResponseTemplate::new(503)).await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + assert!(t.sink.ops().is_empty()); + + t.wait_for_upserts(1).await; + assert_eq!(t.sink.ops(), vec![upsert("t1", MAINTAINER, "open")]); + assert!(t.authors.get(&ticket_uri("t1")).is_none()); + flusher.abort(); + } + + #[tokio::test] + async fn past_the_held_cap_a_provisional_doc_is_indexed_at_once() { + let t = Authored::new(Duration::from_secs(5), 1).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[ + ("t1", "did:plc:alice"), + ("t2", "did:plc:bob"), + ])) + .set_delay(Duration::from_millis(300)), + ) + .await; + t.feed("t1", Frames::backfilled_create(MAINTAINER)).await; + t.feed("t2", Frames::backfilled_create(MAINTAINER)).await; + assert_eq!(t.sink.ops(), vec![upsert("t2", MAINTAINER, "open")]); + } + + fn warming_head_frame(rkey: &str) -> HydrantFrame { + Frames::backfilled_create(MAINTAINER).frame(rkey) + } + + #[tokio::test] + async fn promotion_waits_for_a_running_fill() { + let t = Authored::new(Duration::from_secs(5), 100).await; + t.mirror_answers( + ResponseTemplate::new(200) + .set_body_json(origin_page(&[("t1", "did:plc:alice")])) + .set_delay(Duration::from_millis(300)), + ) + .await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed_at(warming_head_frame("t1"), Some(HydrantCursor::new(1))) + .await; + assert!( + !t.rt.coverage.snapshot().is_ready(), + "the replay reached the head but a fill is running" + ); + + let mut coverage = t.rt.coverage.subscribe(); + tokio::time::timeout(Duration::from_secs(10), coverage.wait_for(|c| c.is_ready())) + .await + .expect("coverage never promoted") + .unwrap(); + assert_eq!(t.sink.ops(), vec![upsert("t1", "did:plc:alice", "open")]); + flusher.abort(); + } + + #[tokio::test] + async fn promotion_gives_up_after_the_hold() { + let t = Authored::new(Duration::from_millis(600), 100).await; + t.mirror_answers(ResponseTemplate::new(503)).await; + let flusher = spawn_author_flusher(&t.rt).unwrap(); + t.feed_at(warming_head_frame("t1"), Some(HydrantCursor::new(1))) + .await; + assert!(!t.rt.coverage.snapshot().is_ready()); + + let started = Instant::now(); + let mut coverage = t.rt.coverage.subscribe(); + tokio::time::timeout(Duration::from_secs(10), coverage.wait_for(|c| c.is_ready())) + .await + .expect("coverage never promoted") + .unwrap(); + assert!( + started.elapsed() >= Duration::from_millis(200), + "retries ran first" + ); + flusher.abort(); + } } diff --git a/bobbin/crates/knot-proxy/Cargo.toml b/bobbin/crates/knot-proxy/Cargo.toml index f2dc6f64e..06ebec7a1 100644 --- a/bobbin/crates/knot-proxy/Cargo.toml +++ b/bobbin/crates/knot-proxy/Cargo.toml @@ -13,7 +13,10 @@ http = { workspace = true } jacquard-common = { workspace = true } reqwest = { workspace = true } scc = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } thiserror = { workspace = true } +tracing = { workspace = true } tokio = { workspace = true, features = ["net"] } url = { workspace = true } diff --git a/bobbin/crates/knot-proxy/src/lib.rs b/bobbin/crates/knot-proxy/src/lib.rs index 505849b66..68d4927b8 100644 --- a/bobbin/crates/knot-proxy/src/lib.rs +++ b/bobbin/crates/knot-proxy/src/lib.rs @@ -21,11 +21,13 @@ mod breaker; mod dns; mod host; mod mirror; +mod origins; pub use breaker::{Breaker, BreakerPermit, CircuitOpen, FailureThreshold, ThresholdError}; pub use dns::PrivateAddressFilter; pub use host::{KnotHost, KnotHostError, PrivateHostReason, classify_ip}; pub use mirror::{MirrorProxy, MirrorProxyError}; +pub use origins::{FillConfig, TicketAuthors}; const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION")); const HTTPS_SCHEME: &str = "https"; diff --git a/bobbin/crates/knot-proxy/src/origins.rs b/bobbin/crates/knot-proxy/src/origins.rs new file mode 100644 index 000000000..e5c1f0a8b --- /dev/null +++ b/bobbin/crates/knot-proxy/src/origins.rs @@ -0,0 +1,450 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bobbin_runtime::{Clock, RuntimeHasher}; +use futures::StreamExt; +use http::{HeaderMap, StatusCode}; +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::nsid::Nsid; +use scc::HashMap as SccMap; +use serde::Deserialize; +use tokio::sync::{Semaphore, broadcast}; +use tokio::time::Instant; + +use crate::{KnotProxyError, MirrorProxy}; + +const LIST_RECORD_ORIGINS_NSID: &str = "org.tangled.temp.repo.listRecordOrigins"; +const TICKET_COLLECTION: &str = "org.tangled.track.ticket"; +// about 35 KiB a page, well under MAX_PAGE_BYTES +const PAGE_LIMIT: &str = "200"; +const MAX_PAGE_BYTES: usize = 256 * 1024; +const RETRY_INITIAL: Duration = Duration::from_millis(250); +const RETRY_MAX: Duration = Duration::from_secs(4); +const DONE_CHANNEL_DEPTH: usize = 256; + +#[derive(Clone, Copy, Debug)] +pub struct FillConfig { + pub concurrency: usize, + pub max_hold: Duration, + pub refill_interval: Duration, +} + +impl Default for FillConfig { + fn default() -> Self { + Self { + concurrency: 8, + max_hold: Duration::from_secs(30), + refill_interval: Duration::from_secs(10), + } + } +} + +#[derive(Default)] +struct RepoFill { + running: bool, + last_started: Option, +} + +/// backfills only see the latest editor, so the opener comes from the mirror +pub struct TicketAuthors { + memo: SccMap, RuntimeHasher>, + repos: Mutex>, + running: AtomicUsize, + permits: Semaphore, + done: broadcast::Sender, + mirror: Option>, + clock: Arc, + config: FillConfig, +} + +#[derive(Deserialize)] +struct OriginsPage { + origins: Vec, + #[serde(default)] + cursor: Option, +} + +#[derive(Deserialize)] +struct Origin { + rkey: String, + author: Did, +} + +enum PageError { + GiveUp(String), + Retry(String), +} + +impl TicketAuthors { + pub fn new( + mirror: Option>, + clock: Arc, + hasher: RuntimeHasher, + config: FillConfig, + ) -> Arc { + Arc::new(Self { + memo: SccMap::with_hasher(hasher), + repos: Mutex::new(HashMap::new()), + running: AtomicUsize::new(0), + permits: Semaphore::new(config.concurrency.max(1)), + done: broadcast::channel(DONE_CHANNEL_DEPTH).0, + mirror, + clock, + config, + }) + } + + pub fn max_hold(&self) -> Duration { + self.config.max_hold + } + + pub fn get(&self, uri: &str) -> Option> { + self.memo.read_sync(uri, |_, author| author.clone()) + } + + pub fn remember(&self, uri: &str, author: Did) { + let _ = self.memo.insert_sync(uri.to_owned(), author); + } + + pub fn forget(&self, uri: &str) { + self.memo.remove_sync(uri); + } + + pub fn any_running(&self) -> bool { + self.running.load(Ordering::Acquire) > 0 + } + + pub fn is_running(&self, repo: &str) -> bool { + self.repos + .lock() + .expect("ticket authors mutex poisoned") + .get(repo) + .is_some_and(|fill| fill.running) + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.done.subscribe() + } + + /// true means a done send is coming for the repo, false means don't wait on it + pub fn kick(self: &Arc, repo: &str) -> bool { + let Some(mirror) = self.mirror.clone() else { + return false; + }; + let now = self.clock.now_instant(); + { + let mut repos = self.repos.lock().expect("ticket authors mutex poisoned"); + let fill = repos.entry(repo.to_owned()).or_default(); + if fill.running { + return true; + } + let too_soon = fill + .last_started + .is_some_and(|at| now < at + self.config.refill_interval); + if too_soon { + return false; + } + fill.running = true; + fill.last_started = Some(now); + } + self.running.fetch_add(1, Ordering::AcqRel); + let this = Arc::clone(self); + let repo = repo.to_owned(); + tokio::spawn(async move { + this.fill(&mirror, &repo).await; + this.finish(repo); + }); + true + } + + pub async fn wait(&self, repo: &str, cap: Duration) { + let mut done = self.done.subscribe(); + let deadline = self.clock.now_instant() + cap; + while self.is_running(repo) { + tokio::select! { + biased; + _ = self.clock.sleep_until(deadline) => return, + _ = done.recv() => {} + } + } + } + + fn finish(&self, repo: String) { + if let Some(fill) = self + .repos + .lock() + .expect("ticket authors mutex poisoned") + .get_mut(&repo) + { + fill.running = false; + } + self.running.fetch_sub(1, Ordering::AcqRel); + let _ = self.done.send(repo); + } + + async fn fill(&self, mirror: &MirrorProxy, repo: &str) { + let Ok(_permit) = self.permits.acquire().await else { + return; + }; + let deadline = self.clock.now_instant() + self.config.max_hold; + let mut cursor: Option = None; + let mut delay = RETRY_INITIAL; + loop { + match fetch_page(mirror, repo, cursor.as_deref()).await { + Ok(page) => { + delay = RETRY_INITIAL; + for origin in page.origins { + let uri = format!("at://{repo}/{TICKET_COLLECTION}/{}", origin.rkey); + self.remember(&uri, origin.author); + } + match page.cursor { + Some(next) => cursor = Some(next), + None => return, + } + } + Err(PageError::GiveUp(why)) => { + tracing::debug!(repo, why, "ticket authors fill gave up"); + return; + } + Err(PageError::Retry(why)) => { + if self.clock.now_instant() + delay > deadline { + tracing::warn!(repo, why, "ticket authors fill ran out of hold time"); + return; + } + self.clock.sleep(delay).await; + delay = (delay * 2).min(RETRY_MAX); + } + } + } + } +} + +async fn fetch_page( + mirror: &MirrorProxy, + repo: &str, + cursor: Option<&str>, +) -> Result { + let nsid = + Nsid::new_static(LIST_RECORD_ORIGINS_NSID).expect("list record origins NSID is valid"); + let mut query = vec![ + ("repo", repo), + ("collection", TICKET_COLLECTION), + ("limit", PAGE_LIMIT), + ]; + if let Some(cursor) = cursor { + query.push(("cursor", cursor)); + } + let response = mirror + .forward_raw(&nsid, &query, HeaderMap::new()) + .await + .map_err(|error| match error { + KnotProxyError::PlaintextHttp { .. } | KnotProxyError::BlockedHost { .. } => { + PageError::GiveUp(error.to_string()) + } + _ => PageError::Retry(error.to_string()), + })?; + let status = response.status(); + if !status.is_success() { + response.discard().await; + let why = format!("mirror returned {status}"); + let retryable = matches!( + status, + StatusCode::TOO_MANY_REQUESTS | StatusCode::REQUEST_TIMEOUT + ); + return Err(if retryable { + PageError::Retry(why) + } else { + PageError::GiveUp(why) + }); + } + let mut body = response.into_body_stream(); + let mut bytes = Vec::new(); + while let Some(chunk) = body.next().await { + let chunk = chunk.map_err(|error| PageError::Retry(error.to_string()))?; + if bytes.len().saturating_add(chunk.len()) > MAX_PAGE_BYTES { + return Err(PageError::GiveUp("origins page is too large".to_owned())); + } + bytes.extend_from_slice(&chunk); + } + serde_json::from_slice(&bytes).map_err(|error| PageError::GiveUp(error.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + use bobbin_runtime::SystemClock; + use serde_json::json; + use url::Url; + use wiremock::matchers::{method, path, query_param}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + const REPO: &str = "did:plc:repo"; + + fn authors_for(server: &MockServer, config: FillConfig) -> Arc { + let clock: Arc = Arc::new(SystemClock::new()); + let mirror = MirrorProxy::new( + &Url::parse(&server.uri()).unwrap(), + clock.clone(), + RuntimeHasher::default(), + ) + .unwrap(); + TicketAuthors::new( + Some(Arc::new(mirror)), + clock, + RuntimeHasher::default(), + config, + ) + } + + fn uri(rkey: &str) -> String { + format!("at://{REPO}/{TICKET_COLLECTION}/{rkey}") + } + + fn origin(rkey: &str, author: &str) -> serde_json::Value { + json!({ "rkey": rkey, "author": author, "changeId": "abc", "createdAt": "2026-05-01T00:00:00Z" }) + } + + fn page(origins: Vec, cursor: Option<&str>) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({ + "repo": REPO, + "collection": TICKET_COLLECTION, + "cursor": cursor, + "origins": origins, + })) + } + + fn origins_route() -> wiremock::MockBuilder { + Mock::given(method("GET")).and(path("/xrpc/org.tangled.temp.repo.listRecordOrigins")) + } + + #[tokio::test] + async fn a_fill_pages_through_every_origin_and_announces_itself() { + let server = MockServer::start().await; + origins_route() + .and(query_param("repo", REPO)) + .and(query_param("collection", TICKET_COLLECTION)) + .and(query_param("cursor", "a")) + .respond_with(page(vec![origin("b", "did:plc:bob")], None)) + .mount(&server) + .await; + origins_route() + .respond_with(page(vec![origin("a", "did:plc:alice")], Some("a"))) + .mount(&server) + .await; + let authors = authors_for(&server, FillConfig::default()); + let mut done = authors.subscribe(); + + assert!(authors.kick(REPO)); + assert_eq!(done.recv().await.unwrap(), REPO); + + assert_eq!(authors.get(&uri("a")).unwrap().as_ref(), "did:plc:alice"); + assert_eq!(authors.get(&uri("b")).unwrap().as_ref(), "did:plc:bob"); + assert!(!authors.is_running(REPO) && !authors.any_running()); + } + + #[tokio::test] + async fn a_second_kick_while_running_is_single_flight() { + let server = MockServer::start().await; + origins_route() + .respond_with( + page(vec![origin("a", "did:plc:alice")], None) + .set_delay(Duration::from_millis(100)), + ) + .expect(1) + .mount(&server) + .await; + let authors = authors_for(&server, FillConfig::default()); + let mut done = authors.subscribe(); + + assert!(authors.kick(REPO)); + assert!(authors.kick(REPO), "joins the running fill"); + done.recv().await.unwrap(); + assert!( + !authors.kick(REPO), + "the refill interval holds a repo back after a fill" + ); + } + + #[tokio::test] + async fn a_missing_repo_gives_up_without_retrying() { + let server = MockServer::start().await; + origins_route() + .respond_with( + ResponseTemplate::new(404).set_body_json(json!({ "error": "RepoNotFound" })), + ) + .expect(1) + .mount(&server) + .await; + let authors = authors_for(&server, FillConfig::default()); + let mut done = authors.subscribe(); + authors.kick(REPO); + done.recv().await.unwrap(); + assert!(authors.get(&uri("a")).is_none()); + } + + #[tokio::test] + async fn a_failing_mirror_is_retried_until_the_hold_runs_out() { + let server = MockServer::start().await; + origins_route() + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + let authors = authors_for( + &server, + FillConfig { + max_hold: Duration::from_millis(700), + ..FillConfig::default() + }, + ); + let mut done = authors.subscribe(); + authors.kick(REPO); + done.recv().await.unwrap(); + let attempts = server.received_requests().await.unwrap().len(); + assert!(attempts >= 2, "retried {attempts} times"); + } + + #[tokio::test] + async fn no_mirror_means_no_fill() { + let authors = TicketAuthors::new( + None, + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + FillConfig::default(), + ); + assert!(!authors.kick(REPO)); + assert!(!authors.any_running()); + } + + #[tokio::test] + async fn wait_returns_once_the_fill_lands() { + let server = MockServer::start().await; + origins_route() + .respond_with( + page(vec![origin("a", "did:plc:alice")], None) + .set_delay(Duration::from_millis(100)), + ) + .mount(&server) + .await; + let authors = authors_for(&server, FillConfig::default()); + authors.kick(REPO); + authors.wait(REPO, Duration::from_secs(5)).await; + assert_eq!(authors.get(&uri("a")).unwrap().as_ref(), "did:plc:alice"); + } + + #[test] + fn forget_drops_a_remembered_author() { + let authors = TicketAuthors::new( + None, + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + FillConfig::default(), + ); + authors.remember(&uri("a"), Did::new_owned("did:plc:alice").unwrap()); + assert!(authors.get(&uri("a")).is_some()); + authors.forget(&uri("a")); + assert!(authors.get(&uri("a")).is_none()); + } +} diff --git a/bobbin/crates/search/src/lib.rs b/bobbin/crates/search/src/lib.rs index 3cc2bc60a..7a9556c21 100644 --- a/bobbin/crates/search/src/lib.rs +++ b/bobbin/crates/search/src/lib.rs @@ -207,7 +207,7 @@ pub enum FacetKind { Kind(Vec), /// the rkeys of label definitions Label(Vec), - /// the author or the editor of a ticket + /// the account that opened a ticket, or the owner of any other record Author(Vec>), } @@ -240,7 +240,6 @@ struct Fields { kind: Field, state: Field, label: Field, - editor: Field, } impl SearchIndex { @@ -262,7 +261,6 @@ impl SearchIndex { let kind = sb.add_text_field("kind", STRING); let state = sb.add_text_field("state", STRING); let label = sb.add_text_field("label", STRING); - let editor = sb.add_text_field("editor", STRING); let schema = sb.build(); let index = Index::create_in_ram(schema); let analyzer = index.tokenizer_for_field(title)?; @@ -282,7 +280,6 @@ impl SearchIndex { kind, state, label, - editor, }; let (tx, rx) = mpsc::channel(WRITE_QUEUE_CAPACITY); let writer_reader = reader.clone(); @@ -624,16 +621,7 @@ impl Inner { FacetKind::State(states) => self.any_of(f.state, states.iter().map(|s| s.as_str())), FacetKind::Kind(kinds) => self.any_of(f.kind, kinds.iter().map(|k| k.as_str())), FacetKind::Label(keys) => self.any_of(f.label, keys.iter().map(String::as_str)), - FacetKind::Author(dids) => Box::new(BooleanQuery::new(vec![ - ( - Occur::Should, - self.any_of(f.editor, dids.iter().map(|d| d.as_str())), - ), - ( - Occur::Should, - self.any_of(f.author, dids.iter().map(|d| d.as_str())), - ), - ])), + FacetKind::Author(dids) => self.any_of(f.author, dids.iter().map(|d| d.as_str())), } } @@ -793,9 +781,6 @@ fn apply_upsert(writer: &mut IndexWriter, fields: Fields, doc: SearchDoc) { for label in &ticket.labels { td.add_text(fields.label, label); } - if let Some(editor) = &ticket.editor { - td.add_text(fields.editor, editor.as_ref()); - } } if let Err(e) = writer.add_document(td) { warn!(?e, "search add_document failed"); @@ -901,7 +886,7 @@ mod tests { state: &'a str, pull: bool, labels: &'a [&'a str], - editor: Option<&'a str>, + author: Option<&'a str>, } impl Default for T<'_> { @@ -913,21 +898,20 @@ mod tests { state: "open", pull: false, labels: &[], - editor: None, + author: None, } } } fn ticket(t: T<'_>) -> SearchDoc { SearchDoc { - author: Some(did(REPO)), + author: Some(did(t.author.unwrap_or(REPO))), created_at: Some(t.created_at), repo: Some(did(REPO)), ticket: Some(TicketFacets { kind: ticket_facet_kind(t.pull), state: t.state.to_owned(), labels: t.labels.iter().map(|l| (*l).to_owned()).collect(), - editor: t.editor.map(did), }), ..doc( at(&format!("at://{REPO}/{TICKET}/{}", t.rkey)), @@ -1110,12 +1094,12 @@ mod tests { } #[tokio::test] - async fn facets_filter_state_kind_label_and_editor() { + async fn facets_filter_state_kind_label_and_author() { let idx = seeded(vec![ ticket(T { rkey: "open-issue", labels: &["bug"], - editor: Some("did:plc:alice"), + author: Some("did:plc:alice"), created_at: 5, ..T::default() }), @@ -1131,7 +1115,7 @@ mod tests { state: "completed", pull: true, labels: &["docs"], - editor: Some("did:plc:bob"), + author: Some("did:plc:bob"), created_at: 3, ..T::default() }), @@ -1188,6 +1172,11 @@ mod tests { run(vec![Facet::is(FacetKind::Author(vec![did("did:plc:bob")]))]).await, vec!["merged-pull"] ); + assert_eq!( + run(vec![Facet::is(FacetKind::Author(vec![did(REPO)]))]).await, + vec!["closed-issue", "open-pull"], + "the repo only counts as an author for the tickets it opened itself" + ); assert!( run(vec![Facet::is(FacetKind::Label(vec![]))]) .await diff --git a/bobbin/crates/types/src/edges.rs b/bobbin/crates/types/src/edges.rs index c2da4c409..02c7c157a 100644 --- a/bobbin/crates/types/src/edges.rs +++ b/bobbin/crates/types/src/edges.rs @@ -456,6 +456,30 @@ fn vouch_edges( pub const TICKET_ISSUE_KIND: &str = "org.tangled.track.ticket.issue"; pub const TICKET_PULL_KIND: &str = "org.tangled.track.ticket.pull"; +pub const TICKET_ISSUE_BY_KIND: &str = "sh.tangled.repo.issue.by"; +pub const TICKET_PULL_BY_KIND: &str = "sh.tangled.repo.pull.by"; + +pub fn ticket_edge(edges: &[Edge]) -> Option<&Edge> { + edges + .iter() + .find(|e| matches!(e.kind.as_ref(), TICKET_ISSUE_KIND | TICKET_PULL_KIND)) +} + +/// extraction can't build this, a record only names its latest editor +pub fn ticket_by_edge(ticket_edge: &Edge, author: Did) -> Option { + let kind = match ticket_edge.kind.as_ref() { + TICKET_ISSUE_KIND => TICKET_ISSUE_BY_KIND, + TICKET_PULL_KIND => TICKET_PULL_BY_KIND, + _ => return None, + }; + Some(Edge { + kind: nsid_static(kind), + subject: SubjectRef::Did(author), + source: ticket_edge.source.clone(), + sort_micros: ticket_edge.sort_micros, + }) +} + /// a ticket with a target branch is considered as a pull request pub fn ticket_kind(ticket: &Ticket) -> &'static str { if ticket.target_branch.is_some() { @@ -904,6 +928,33 @@ mod tests { ); } + #[test] + fn a_ticket_is_listed_under_its_author_not_its_repo() { + let edges = extract( + "org.tangled.track.ticket", + "at://did:plc:abalone/org.tangled.track.ticket/t1", + json!({ + "$type": "org.tangled.track.ticket", + "title": { "text": "t" }, + "state": "open", + "targetBranch": "main", + "createdAt": "2026-05-01T00:00:00Z", + "x-tngl-editor": "did:plc:closer", + }), + ); + assert!( + edges.iter().all(|e| !e.kind.as_ref().ends_with(".by")), + "extraction doesn't know the opener" + ); + let primary = ticket_edge(&edges).expect("a pull makes its ticket edge"); + let author = Did::new_owned("did:plc:opener").unwrap(); + let by = ticket_by_edge(primary, author.clone()).expect("pulls have a by edge"); + assert_eq!(by.kind.as_ref(), TICKET_PULL_BY_KIND); + assert_eq!(by.subject, SubjectRef::Did(author)); + assert_eq!(by.source, primary.source); + assert_eq!(by.sort_micros, primary.sort_micros); + } + #[test] fn patch_keys_on_its_ticket_in_the_same_repo() { let oid = json!({"oid": "0123456789abcdef0123456789abcdef01234567"}); diff --git a/bobbin/crates/types/src/search.rs b/bobbin/crates/types/src/search.rs index 3339af285..e8b4f16d7 100644 --- a/bobbin/crates/types/src/search.rs +++ b/bobbin/crates/types/src/search.rs @@ -38,8 +38,6 @@ pub struct TicketFacets { pub state: String, /// the rkeys of label definitions pub labels: Vec, - /// the editor of a ticket that a repo owns - pub editor: Option>, } pub const TICKET_FACET_ISSUE: &str = "issue"; @@ -224,8 +222,9 @@ fn ticket_doc(source: &AtUri, r: &Ticket) -> SearchDoc { kind: ticket_facet_kind(ticket_kind(r) == TICKET_PULL_KIND), state: r.state.as_str().to_owned(), labels, - editor: ticket_editor(r), }), + // the editor stands in for the opener until ingest swaps it + author: ticket_editor(r).or_else(|| author_of(source)), ..doc( source, "org.tangled.track.ticket", @@ -358,7 +357,7 @@ mod tests { assert_eq!(facets.state, "completed"); assert_eq!(facets.labels, vec!["bug".to_owned(), "3lxyz".to_owned()]); assert_eq!( - facets.editor.as_ref().map(|d| d.as_str()), + doc.author.as_ref().map(|d| d.as_str()), Some("did:plc:octopus") ); } @@ -379,7 +378,11 @@ mod tests { assert_eq!(facets.kind, TICKET_FACET_ISSUE); assert_eq!(facets.state, "open"); assert!(facets.labels.is_empty()); - assert!(facets.editor.is_none()); + assert_eq!( + doc.author.as_ref().map(|d| d.as_str()), + Some("did:plc:squid"), + "a ticket without an editor isn't knot hosted, so its repo is its author" + ); } #[test] diff --git a/bobbin/crates/xrpc/src/authors.rs b/bobbin/crates/xrpc/src/authors.rs new file mode 100644 index 000000000..711392971 --- /dev/null +++ b/bobbin/crates/xrpc/src/authors.rs @@ -0,0 +1,33 @@ +use std::time::Duration; + +use bobbin_types::ids::owner_did_from_aturi; +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::string::AtUri; + +use crate::AppState; + +const FILL_WAIT: Duration = Duration::from_secs(2); + +/// the opener comes from the cache, `editor` is just the latest writer as a fallback +pub(crate) async fn ticket_author( + state: &AppState, + uri: &AtUri, + editor: Option>, +) -> Option> { + let authors = &state.ticket_authors; + if let Some(author) = authors.get(uri.as_ref()) { + return Some(author); + } + let repo = owner_did_from_aturi(uri)?; + let Some(editor) = editor else { + return Some(repo); + }; + if authors.kick(repo.as_ref()) { + authors.wait(repo.as_ref(), FILL_WAIT).await; + if let Some(author) = authors.get(uri.as_ref()) { + return Some(author); + } + } + Some(editor) +} diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 6fb39cbbe..27d7b40cb 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -27,14 +27,14 @@ use bobbin_edge_index::{ ParsedCid, Rejection, Settlement, Settlements, SortDir, StateKind, StreamHealth, StreamHealthSnapshot, TicketStateKind, TotalCount, Watch, }; -use bobbin_knot_proxy::{KnotProxyError, MirrorProxy, ProxyResponse}; +use bobbin_knot_proxy::{FillConfig, KnotProxyError, MirrorProxy, ProxyResponse, TicketAuthors}; use bobbin_record_lru::RecordStore; use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver}; use bobbin_runtime::{ReqwestHttp, UnixMicros}; use bobbin_search::{ActorIndex, ActorIndexError, SearchReader}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::edges::REPO_SOURCE_EDGE_KIND; -use bobbin_types::edges::{TICKET_ISSUE_KIND, TICKET_PULL_KIND}; +use bobbin_types::edges::{TICKET_ISSUE_KIND, TICKET_PULL_KIND, ticket_editor}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::org_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; use bobbin_types::org_tangled::feed::comment::{ @@ -86,6 +86,7 @@ use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer}; use tracing::{Level, Span}; mod actor; +mod authors; mod backpressure; mod change_ids; mod client_address; @@ -214,6 +215,7 @@ pub struct AppState { pub settlements: Arc, pub invites: Arc, pub claims: Arc, + pub ticket_authors: Arc, /// the ingest stream's liveness, absent when nothing publishes it pub stream_health: Option>, pub moderation: Option, @@ -260,6 +262,12 @@ impl AppState { settlements: Arc::new(Settlements::new()), invites: Arc::new(InviteIndex::new()), claims: Arc::new(ClaimIndex::new()), + ticket_authors: TicketAuthors::new( + None, + Arc::new(bobbin_runtime::SystemClock::new()), + bobbin_runtime::RuntimeHasher::default(), + FillConfig::default(), + ), stream_health: None, moderation: None, cors: Cors::Off, @@ -328,6 +336,11 @@ impl AppState { self } + pub fn with_ticket_authors(mut self, authors: Arc) -> Self { + self.ticket_authors = authors; + self + } + pub fn with_moderation(mut self, moderation: Option) -> Self { self.moderation = moderation; self @@ -1068,6 +1081,8 @@ struct StatefulItem { view: RecordView, state: &'static str, comment_count: u64, + #[serde(skip_serializing_if = "Option::is_none")] + author: Option>, } pub(crate) fn source_authority_did(source: &AtUri) -> Option> { @@ -1077,7 +1092,7 @@ pub(crate) fn source_authority_did(source: &AtUri) -> Option>, ) -> StatefulItem> { @@ -1086,10 +1101,12 @@ fn enrich_ticket_view( SubjectRef::Uri(view.uri.clone()), )); let kind = TicketStateKind::from_ticket(view.value.state.as_ref()).unwrap_or_default(); + let author = authors::ticket_author(state, &view.uri, ticket_editor(&view.value)).await; StatefulItem { view, state: kind.wire(), comment_count, + author, } } @@ -2269,7 +2286,15 @@ fn list_tickets( page.items, HitProvenance::Indexed, ) - .map(move |view| view.map(|v| enrich_ticket_view(&owned, v))); + .then(move |view| { + let owned = owned.clone(); + async move { + match view { + Ok(v) => Ok(enrich_ticket_view(&owned, v).await), + Err(e) => Err(e), + } + } + }); Ok(json_stream::>, _>( "items", items, diff --git a/bobbin/crates/xrpc/src/profile_activity.rs b/bobbin/crates/xrpc/src/profile_activity.rs index 263467910..76bca36a5 100644 --- a/bobbin/crates/xrpc/src/profile_activity.rs +++ b/bobbin/crates/xrpc/src/profile_activity.rs @@ -2,7 +2,9 @@ use std::collections::{BTreeMap, HashMap}; use axum::{extract::State, response::Json}; use bobbin_edge_index::{EdgeItem, EdgeStore, PageCursor, PageLimit, SortDir}; -use bobbin_types::edges::{TICKET_PULL_KIND, ticket_kind}; +use bobbin_types::edges::{ + TICKET_ISSUE_BY_KIND, TICKET_PULL_BY_KIND, TICKET_PULL_KIND, ticket_kind, +}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::org_tangled::actor::get_profile_activity::{ ActivityMonth, GetProfileActivityOutput, IssueActivity, PullActivity, RepoActivity, @@ -30,10 +32,6 @@ const PAGE_SIZE: u32 = 1_000; const MAX_ACTIVITY_ITEMS: usize = 1_000; const HYDRATE_CONCURRENCY: usize = 8; -// TODO: update index names -const ISSUE_BY_NSID: &str = "sh.tangled.repo.issue.by"; -const PULL_BY_NSID: &str = "sh.tangled.repo.pull.by"; - #[derive(Debug, Deserialize)] pub(crate) struct GetProfileActivityQuery { actor: Did, @@ -117,8 +115,14 @@ fn activity_keys(subject: &Did) -> [EdgeKey; 3] { nsid_static(ManifestRecord::NSID), SubjectRef::Did(subject.clone()), ), - EdgeKey::new(nsid_static(ISSUE_BY_NSID), SubjectRef::Did(subject.clone())), - EdgeKey::new(nsid_static(PULL_BY_NSID), SubjectRef::Did(subject.clone())), + EdgeKey::new( + nsid_static(TICKET_ISSUE_BY_KIND), + SubjectRef::Did(subject.clone()), + ), + EdgeKey::new( + nsid_static(TICKET_PULL_BY_KIND), + SubjectRef::Did(subject.clone()), + ), ] } @@ -403,12 +407,12 @@ mod tests { "2026-09-01T00:00:00Z", ), ( - ISSUE_BY_NSID, + TICKET_ISSUE_BY_KIND, "org.tangled.track.ticket", "2026-08-01T00:00:00Z", ), ( - PULL_BY_NSID, + TICKET_PULL_BY_KIND, "org.tangled.track.ticket", "2026-03-01T00:00:00Z", ), diff --git a/bobbin/crates/xrpc/src/repo.rs b/bobbin/crates/xrpc/src/repo.rs index acd48424a..f7bd67bc2 100644 --- a/bobbin/crates/xrpc/src/repo.rs +++ b/bobbin/crates/xrpc/src/repo.rs @@ -1,5 +1,5 @@ use crate::{ - AppState, XrpcError, fetch, + AppState, XrpcError, authors, fetch, view::{self, build_profile_basic, build_repo_view_basic}, }; use axum::extract::State; @@ -67,7 +67,9 @@ pub(crate) async fn get_ticket_view( .ok_or_else(|| XrpcError::InvalidParams("ticket uri authority must be a did".into()))?; let repo = build_repo_view_basic(&state, viewer, &repo_did).await?; - let author_did = ticket_editor(&ticket).unwrap_or_else(|| repo_did.clone()); + let author_did = authors::ticket_author(&state, ¶ms.uri, ticket_editor(&ticket)) + .await + .unwrap_or_else(|| repo_did.clone()); let author = build_profile_basic(&state, viewer, &author_did).await; Ok(XrpcResponse(get_ticket_view::GetTicketViewOutput { diff --git a/bobbin/crates/xrpc/src/search.rs b/bobbin/crates/xrpc/src/search.rs index 81c661f25..c89635955 100644 --- a/bobbin/crates/xrpc/src/search.rs +++ b/bobbin/crates/xrpc/src/search.rs @@ -8,6 +8,7 @@ use bobbin_search::{ Facet, FacetKind, SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchRequest, Sort, StateCounts, TextTerm, TicketKind, }; +use bobbin_types::edges::ticket_editor; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static}; use bobbin_types::search::SearchableRecord; use jacquard_common::DefaultStr; @@ -18,7 +19,7 @@ use serde::Serialize; use url::form_urlencoded; use crate::{ - AppState, HitProvenance, XrpcError, drop_unhydratable, hydrate_stream, json_stream, + AppState, HitProvenance, XrpcError, authors, drop_unhydratable, hydrate_stream, json_stream, parse_limit, resolve_for_view, }; @@ -338,6 +339,8 @@ struct SearchHitView { state: Option, #[serde(skip_serializing_if = "Option::is_none")] comment_count: Option, + #[serde(skip_serializing_if = "Option::is_none")] + author: Option>, } async fn hydrate_search_hit( @@ -364,6 +367,12 @@ async fn hydrate_search_hit( ), _ => (None, None), }; + let author = match &value { + SearchableRecord::Ticket(t) => { + authors::ticket_author(state, &body.uri, ticket_editor(t)).await + } + _ => None, + }; Ok(Some(SearchHitView { uri: body.uri.clone(), cid: Some(body.cid.clone()), @@ -372,6 +381,7 @@ async fn hydrate_search_hit( value, state: ticket_state, comment_count, + author, })) } diff --git a/bobbin/crates/xrpc/tests/aggregation.rs b/bobbin/crates/xrpc/tests/aggregation.rs index 6a23bea54..9d84a42c3 100644 --- a/bobbin/crates/xrpc/tests/aggregation.rs +++ b/bobbin/crates/xrpc/tests/aggregation.rs @@ -5,7 +5,7 @@ use bobbin_edge_index::{ Coverage, CoverageWatch, EdgeStore, HydrantCursor, InviteIndex, InviteScope, Offer, PageToken, TicketStateKind, }; -use bobbin_knot_proxy::MirrorProxy; +use bobbin_knot_proxy::{FillConfig, MirrorProxy, TicketAuthors}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; use bobbin_resolver::RepoIdResolver; use bobbin_runtime::{RuntimeHasher, SystemClock, UnixMicros}; @@ -2680,9 +2680,11 @@ async fn flood_of_offers_is_limited_and_says_so() { ); } -#[tokio::test] -async fn ticket_view_names_the_last_editor_as_author() { - let h = Harness::new().await; +const CLOSER: &str = "did:plc:teq"; +const OPENER: &str = "did:plc:opener"; + +/// the record only names the closer, the opener comes from the mirror +async fn mount_closed_ticket(h: &Harness) -> AtUri { let repo = did("did:plc:limpet"); h.mount( &repo, @@ -2696,7 +2698,7 @@ async fn ticket_view_names_the_last_editor_as_author() { ) .await; let mut ticket = pull_body("ship", "completed"); - ticket["x-tngl-editor"] = json!("did:plc:teq"); + ticket["x-tngl-editor"] = json!(CLOSER); h.mount( &repo, &nsid("org.tangled.track.ticket"), @@ -2704,24 +2706,119 @@ async fn ticket_view_names_the_last_editor_as_author() { ticket, ) .await; + at("at://did:plc:limpet/org.tangled.track.ticket/t1") +} - let (status, body) = json_response( - router(h.state.clone()) +async fn with_mirror_naming_the_opener(h: &Harness) -> (AppState, MockServer) { + let mirror = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/xrpc/org.tangled.temp.repo.listRecordOrigins")) + .and(query_param("repo", "did:plc:limpet")) + .and(query_param("collection", "org.tangled.track.ticket")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "repo": "did:plc:limpet", + "collection": "org.tangled.track.ticket", + "origins": [{ + "rkey": "t1", + "author": OPENER, + "changeId": "abc", + "createdAt": "2026-05-01T00:00:00Z", + }], + }))) + .mount(&mirror) + .await; + let clock = Arc::new(SystemClock::new()); + let proxy = MirrorProxy::new( + &Url::parse(&mirror.uri()).unwrap(), + clock.clone(), + RuntimeHasher::default(), + ) + .unwrap(); + let authors = TicketAuthors::new( + Some(Arc::new(proxy)), + clock, + RuntimeHasher::default(), + FillConfig::default(), + ); + (h.state.clone().with_ticket_authors(authors), mirror) +} + +async fn ticket_view(state: AppState, uri: &AtUri) -> (StatusCode, Value) { + json_response( + router(state) .oneshot( Request::builder() - .uri("/xrpc/org.tangled.track.getTicketView?uri=at://did:plc:limpet/org.tangled.track.ticket/t1") + .uri(format!( + "/xrpc/org.tangled.track.getTicketView?uri={}", + uri.as_ref() + )) .body(Body::empty()) .unwrap(), ) .await .unwrap(), ) - .await; + .await +} + +#[tokio::test] +async fn ticket_view_names_the_original_author_not_the_closer() { + let h = Harness::new().await; + let uri = mount_closed_ticket(&h).await; + let (state, mirror) = with_mirror_naming_the_opener(&h).await; + + let (status, body) = ticket_view(state.clone(), &uri).await; + assert_eq!(status, StatusCode::OK, "{body}"); - assert_eq!(body["author"]["did"], "did:plc:teq"); + assert_eq!(body["author"]["did"], OPENER); assert_eq!(body["repo"]["did"], "did:plc:limpet"); assert_eq!(body["state"], "completed"); assert_eq!(body["targetBranch"], "main"); + ticket_view(state, &uri).await; + assert_eq!( + mirror.received_requests().await.unwrap().len(), + 1, + "the second view reads the memo" + ); +} + +#[tokio::test] +async fn ticket_view_falls_back_to_the_last_editor_when_the_mirror_cannot_say() { + let h = Harness::new().await; + let uri = mount_closed_ticket(&h).await; + + let (status, body) = ticket_view(h.state.clone(), &uri).await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["author"]["did"], CLOSER); +} + +#[tokio::test] +async fn list_pulls_names_the_original_author_on_each_item() { + let h = Harness::new().await; + let uri = mount_closed_ticket(&h).await; + let repo = at("at://did:plc:limpet"); + h.add_edge(&nsid(TICKET_PULL_KIND), &repo, &uri); + let (state, _mirror) = with_mirror_naming_the_opener(&h).await; + + let (status, body) = json_response( + router(state) + .oneshot(list_request( + "org.tangled.repo.listPulls", + repo.as_ref(), + &[], + )) + .await + .unwrap(), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["items"][0]["author"], OPENER); + assert_eq!( + body["items"][0]["value"]["x-tngl-editor"], CLOSER, + "the record itself still carries its latest editor" + ); } #[tokio::test] diff --git a/bobbin/crates/xrpc/tests/await_record_e2e.rs b/bobbin/crates/xrpc/tests/await_record_e2e.rs index b8fc0bff7..68ea88b4a 100644 --- a/bobbin/crates/xrpc/tests/await_record_e2e.rs +++ b/bobbin/crates/xrpc/tests/await_record_e2e.rs @@ -109,6 +109,7 @@ async fn a_record_off_the_stream_answers_an_await() { warming_buffer: None, settlements: Some(settlements.clone()), analytics: None, + ticket_authorship: None, }; let state = AppState::new( diff --git a/bobbin/crates/xrpc/tests/enrich.rs b/bobbin/crates/xrpc/tests/enrich.rs index 432ea4885..2e361dc9e 100644 --- a/bobbin/crates/xrpc/tests/enrich.rs +++ b/bobbin/crates/xrpc/tests/enrich.rs @@ -634,6 +634,63 @@ async fn minidoc_payloads_use_observed_record_authors() { assert_eq!(h.state.identity.stats().upstream_requests, 0); } +#[tokio::test] +async fn minidocs_follow_the_author_field_of_ticket_items() { + let h = Harness::new().await; + let repo_did = did("did:plc:limpet"); + let opener = did("did:plc:opener"); + let ticket = ticket_uri(&repo_did); + h.add_edge( + "org.tangled.track.ticket.pull", + SubjectRef::Did(repo_did.clone()), + &ticket, + ); + h.mount( + &repo_did, + "org.tangled.track.ticket", + "i0", + json!({ + "$type": "org.tangled.track.ticket", + "title": { "$type": "org.tangled.markup.markdown#inline", "text": "ship" }, + "state": "completed", + "targetBranch": "main", + "createdAt": "2026-05-01T00:00:00Z", + "x-tngl-editor": "did:plc:closer", + }), + ) + .await; + h.state + .ticket_authors + .remember(ticket.as_ref(), opener.clone()); + h.state + .identity + .observe(opener.clone(), handle("opener.example.com")); + + let (status, body) = json_response( + router(h.state.clone()) + .oneshot(enrich_request(json!({ + "xrpc": "org.tangled.repo.listPulls", + "params": { "subject": repo_did.as_ref() }, + "enrich": [{ + "source": "org.tangled.graph.follow:.repo", + "type": MINIDOC, + "targets": ["items[].author"] + }] + }))) + .await + .unwrap(), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["output"]["items"][0]["author"], json!(opener.as_ref())); + assert_eq!( + body["data"]["did:plc:opener"]["org.tangled.graph.follow:.repo"][MINIDOC]["handle"], + json!("opener.example.com"), + "{body}" + ); +} + #[tokio::test] async fn minidoc_repo_sources_skip_the_author_index() { let h = Harness::new().await; diff --git a/bobbin/crates/xrpc/tests/search.rs b/bobbin/crates/xrpc/tests/search.rs index 66efc0cf5..b2671e5c7 100644 --- a/bobbin/crates/xrpc/tests/search.rs +++ b/bobbin/crates/xrpc/tests/search.rs @@ -304,7 +304,6 @@ impl Harness { kind: ticket_facet_kind(t.pull), state: t.state.to_owned(), labels: t.labels.iter().map(|l| (*l).to_owned()).collect(), - editor: None, }), }) .await; @@ -1239,6 +1238,29 @@ async fn qualifiers_filter_tickets_and_hits_carry_state() { } } +#[tokio::test] +async fn ticket_hits_name_the_original_author() { + let h = Harness::new().await; + seed_reef(&h).await; + h.state.ticket_authors.remember( + &format!("at://{REEF}/org.tangled.track.ticket/open-bug"), + did("did:plc:opener"), + ); + let (status, page) = json_response( + router(h.state.clone()) + .oneshot(search_request(&[("q", "kelp type:issue"), ("repo", REEF)])) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{page}"); + for hit in page["hits"].as_array().unwrap() { + let remembered = hit["uri"].as_str().unwrap().ends_with("/open-bug"); + let want = if remembered { "did:plc:opener" } else { REEF }; + assert_eq!(hit["author"], json!(want), "{hit}"); + } +} + #[tokio::test] async fn state_qualifier_and_counts_drive_tabs() { let h = Harness::new().await; diff --git a/bobbin/example.toml b/bobbin/example.toml index 5ca8bc4e3..81b8c4524 100644 --- a/bobbin/example.toml +++ b/bobbin/example.toml @@ -88,6 +88,30 @@ # Default value: 256 #idle_promote_min_events = 256 +# How many repos look up their ticket openers at once. A backfilled ticket +# only names its latest editor, so the mirror is the only way to learn who opened it. +# +# Can also be specified via environment variable `BOBBIN_INGEST_AUTHOR_FILL_CONCURRENCY`. +# +# Default value: 8 +#author_fill_concurrency = 8 + +# Seconds to keep retrying an unreachable mirror. A starting bobbin waits this +# long at most, then goes ready with each ticket's latest editor as its author. +# +# Can also be specified via environment variable `BOBBIN_INGEST_AUTHOR_FILL_MAX_HOLD_SECS`. +# +# Default value: 30 +#author_fill_max_hold_secs = 30 + +# How many search docs can wait for their ticket's opener. Past this cap they +# index under the latest editor, so a mirror outage can't use unbounded memory. +# +# Can also be specified via environment variable `BOBBIN_INGEST_AUTHOR_HELD_DOCS_MAX`. +# +# Default value: 20000 +#author_held_docs_max = 20000 + [backpressure] # Estimated peak transient anonymous bytes a single hydrating request holds: # roughly the page of decoded records plus its serialized copy. Combined with