From a838c54eed3317ff9ecd6bc671fac71d1a6656df Mon Sep 17 00:00:00 2001 From: Lewis Date: Wed, 06 May 2026 13:38:19 +0000 Subject: [PATCH] refactor: RuntimeHasher, SearchReader, Clock-driven writer everywhere Lewis: May this revision serve well! --- crates/bobbin/Cargo.toml | 4 ++++ crates/edge-index/Cargo.toml | 1 + crates/search/Cargo.toml | 1 + crates/xrpc/Cargo.toml | 1 + crates/bobbin/src/main.rs | 132 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------------- crates/edge-index/src/lib.rs | 70 +++++++++++++++++++++++++++++++++++++++++----------------------------- crates/search/src/lib.rs | 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------- crates/xrpc/src/lib.rs | 51 ++++++++++++++++++++++++++++++++++++++++----------- crates/xrpc/tests/aggregation.rs | 23 +++++++++++++++++------ crates/xrpc/tests/cold_start.rs | 22 ++++++++++++++++------ crates/xrpc/tests/extended.rs | 23 +++++++++++++++++------ crates/xrpc/tests/knot_proxy.rs | 33 +++++++++++++++++++++++++-------- crates/xrpc/tests/search.rs | 25 ++++++++++++++++++------- 13 file(s) changed, 330 insertion(s)(+), 142 deletion(s)(-) diff --git a/crates/bobbin/Cargo.toml b/crates/bobbin/Cargo.toml --- a/crates/bobbin/Cargo.toml +++ b/crates/bobbin/Cargo.toml @@ -14,6 +14,7 @@ bobbin-ingest = { workspace = true } bobbin-knot-proxy = { workspace = true } bobbin-record-lru = { workspace = true } +bobbin-runtime = { workspace = true } bobbin-search = { workspace = true } bobbin-slingshot-client = { workspace = true } bobbin-xrpc = { workspace = true } @@ -31,3 +32,6 @@ confique = { workspace = true } thiserror = { workspace = true } toml = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/edge-index/Cargo.toml b/crates/edge-index/Cargo.toml --- a/crates/edge-index/Cargo.toml +++ b/crates/edge-index/Cargo.toml @@ -6,6 +6,7 @@ rust-version.workspace = true [dependencies] +bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } jacquard-common = { workspace = true } lasso = { workspace = true } diff --git a/crates/search/Cargo.toml b/crates/search/Cargo.toml --- a/crates/search/Cargo.toml +++ b/crates/search/Cargo.toml @@ -6,6 +6,7 @@ rust-version.workspace = true [dependencies] +bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } jacquard-common = { workspace = true } diff --git a/crates/xrpc/Cargo.toml b/crates/xrpc/Cargo.toml --- a/crates/xrpc/Cargo.toml +++ b/crates/xrpc/Cargo.toml @@ -24,6 +24,7 @@ tracing = { workspace = true } [dev-dependencies] +bobbin-runtime = { workspace = true } http = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tower = { workspace = true } diff --git a/crates/bobbin/src/main.rs b/crates/bobbin/src/main.rs --- a/crates/bobbin/src/main.rs +++ b/crates/bobbin/src/main.rs @@ -7,9 +7,10 @@ use anyhow::{Context, anyhow}; use bobbin_edge_index::{CoverageWatch, EdgeStore, HydrantCursor}; use bobbin_ingest::{IngestConfig, IngestRuntime, RepoIdResolver, run as run_ingest}; -use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore}; -use bobbin_search::SearchIndex; +use bobbin_runtime::{Clock, OsEntropy, RuntimeHasher, SystemClock, TungsteniteWs}; +use bobbin_search::{SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use clap::{Parser, Subcommand}; @@ -62,12 +63,11 @@ } }; - if let Err(e) = init_tracing(&cfg) { - eprintln!("failed to install tracing subscriber: {e}"); - return ExitCode::FAILURE; - } - if matches!(cli.command, Some(Command::Validate)) { + if let Err(e) = init_tracing(&cfg, Arc::new(SystemClock::new())) { + eprintln!("failed to install tracing subscriber: {e}"); + return ExitCode::FAILURE; + } println!("configuration is valid"); return ExitCode::SUCCESS; } @@ -81,34 +81,72 @@ } } -fn init_tracing(cfg: &BobbinConfig) -> Result<(), String> { +struct ClockTimer(Arc); + +impl tracing_subscriber::fmt::time::FormatTime for ClockTimer { + fn format_time(&self, w: &mut tracing_subscriber::fmt::format::Writer<'_>) -> std::fmt::Result { + write!(w, "{}", self.0.now_unix_micros().raw()) + } +} + +fn init_tracing(cfg: &BobbinConfig, clock: Arc) -> Result<(), String> { let combined = format!("{},{}", LevelFilter::INFO, cfg.log.filter); - let filter = EnvFilter::try_new(&combined) - .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; let format: LogFormat = cfg.log.format.parse()?; - let builder = tracing_subscriber::fmt().with_env_filter(filter); + let timer = ClockTimer(clock); match format { - LogFormat::Text => builder.try_init().map_err(|e| e.to_string()), - LogFormat::Json => builder.json().try_init().map_err(|e| e.to_string()), + LogFormat::Text => { + let filter = EnvFilter::try_new(&combined) + .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_timer(timer) + .try_init() + .map_err(|e| e.to_string()) + } + LogFormat::Json => { + let filter = EnvFilter::try_new(&combined) + .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; + tracing_subscriber::fmt() + .with_env_filter(filter) + .json() + .with_timer(timer) + .try_init() + .map_err(|e| e.to_string()) + } } } async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { + let clock: Arc = Arc::new(SystemClock::new()); + init_tracing(&cfg, clock.clone()) + .map_err(|e| anyhow!("failed to install tracing subscriber: {e}"))?; + let entropy = Arc::new(OsEntropy); + let hasher = RuntimeHasher::from_entropy(&*entropy); + let ws = TungsteniteWs::shared(); + let records: Arc = Arc::new(LruRecordStore::new(CacheCapacity::from_bytes( cfg.record_cache.lru_bytes, ))); - let slingshot = SlingshotClient::new(cfg.slingshot.url.clone())?; - let resolver = Arc::new(RepoIdResolver::with_slingshot(slingshot.clone())); - let edges = Arc::new(EdgeStore::new()); + let slingshot = SlingshotClient::with_default_http(cfg.slingshot.url.clone())?; + let resolver = Arc::new(RepoIdResolver::with_slingshot( + slingshot.clone(), + hasher.clone(), + )); + let edges = Arc::new(EdgeStore::new(hasher.clone())); let coverage = Arc::new(CoverageWatch::new()); - let knots = Arc::new(KnotProxy::new(KnotProxyConfig { - allow_private_hosts: cfg.knot.allow_private, - require_https: cfg.knot.require_https, - ..KnotProxyConfig::default() - })?); + let knots = Arc::new(KnotProxy::new( + KnotProxyConfig { + allow_private_hosts: cfg.knot.allow_private, + require_https: cfg.knot.require_https, + ..KnotProxyConfig::default() + }, + KnotHttpConfig::default(), + clock.clone(), + hasher, + )?); let search_heap = usize::try_from(cfg.search.heap_bytes) .with_context(|| format!("search.heap_bytes {} exceeds usize", cfg.search.heap_bytes))?; - let search = Arc::new(SearchIndex::new(search_heap)?); + let search = Arc::new(SearchIndex::new(search_heap, clock.clone())?); let ingest_cfg = IngestConfig { hydrant_base: cfg.hydrant.url.clone(), @@ -122,11 +160,21 @@ search: search.clone(), records: records.clone(), resolver: resolver.clone(), + clock: clock.clone(), + entropy, + ws, cancel: cancel.clone(), }; let mut ingest_handle = tokio::spawn(run_ingest(ingest_cfg, ingest_runtime)); - let state = AppState::new(records, slingshot, edges, coverage, knots, search); + let state = AppState::new( + records, + slingshot, + edges, + coverage, + knots, + search as Arc, + ); let app = router(state); let binds = cfg.server.binds.clone(); @@ -148,7 +196,7 @@ cancel.cancel(); let cursor = ingest_coverage.snapshot().last_cursor().raw(); tracing::info!(grace_secs = grace.as_secs(), cursor, "draining ingest"); - drain_with_grace("ingest", grace, &mut ingest_handle).await; + drain_with_grace("ingest", grace, &mut ingest_handle, clock.as_ref()).await; match res { Ok(Ok(())) => Ok(()), Ok(Err(e)) => Err(anyhow::Error::from(e)).context("axum server failed"), @@ -159,7 +207,7 @@ cancel.cancel(); let cursor = ingest_coverage.snapshot().last_cursor().raw(); tracing::info!(grace_secs = grace.as_secs(), cursor, "draining server"); - drain_with_grace("server", grace, &mut server_handle).await; + drain_with_grace("server", grace, &mut server_handle, clock.as_ref()).await; match res { Ok(Ok(())) => Err(anyhow!("ingest run loop exited; loop is supposed to be infinite")), Ok(Err(e)) => Err(anyhow::Error::from(e)).context("ingest exited"), @@ -169,14 +217,22 @@ } } -async fn drain_with_grace(label: &'static str, grace: Duration, handle: &mut JoinHandle) { - if tokio::time::timeout(grace, &mut *handle).await.is_err() { - tracing::warn!( - grace_secs = grace.as_secs(), - label, - "task did not stop within grace, aborting" - ); - handle.abort(); +async fn drain_with_grace( + label: &'static str, + grace: Duration, + handle: &mut JoinHandle, + clock: &dyn Clock, +) { + tokio::select! { + _ = &mut *handle => {} + _ = clock.sleep(grace) => { + tracing::warn!( + grace_secs = grace.as_secs(), + label, + "task did not stop within grace, aborting" + ); + handle.abort(); + } } } @@ -259,7 +315,13 @@ let mut handle = tokio::spawn(async { 7u32 }); tokio::time::advance(Duration::from_millis(1)).await; let start = tokio::time::Instant::now(); - drain_with_grace("test", Duration::from_secs(60), &mut handle).await; + drain_with_grace( + "test", + Duration::from_secs(60), + &mut handle, + &SystemClock::new(), + ) + .await; assert!(start.elapsed() < Duration::from_millis(10)); } @@ -270,7 +332,7 @@ }); let grace = Duration::from_secs(5); let start = tokio::time::Instant::now(); - drain_with_grace("test", grace, &mut handle).await; + drain_with_grace("test", grace, &mut handle, &SystemClock::new()).await; assert!(start.elapsed() >= grace); let outcome = handle.await; assert!(outcome.is_err() && outcome.unwrap_err().is_cancelled()); diff --git a/crates/edge-index/src/lib.rs b/crates/edge-index/src/lib.rs --- a/crates/edge-index/src/lib.rs +++ b/crates/edge-index/src/lib.rs @@ -2,6 +2,7 @@ use std::num::NonZeroU32; use std::sync::{Arc, Mutex}; +use bobbin_runtime::RuntimeHasher; use bobbin_types::edges::Edge; use bobbin_types::ids::EdgeKey; use jacquard_common::DefaultStr; @@ -122,33 +123,37 @@ pub next: Option, } -#[derive(Default)] struct EdgeBucket { sources: RoaringBitmap, - author_refs: HashMap, + author_refs: HashMap, } -pub struct EdgeStore { - source_interner: Arc>, - did_interner: Arc>, - forward: SccMap, - reverse: SccMap>, - writer: Mutex<()>, -} - -impl Default for EdgeStore { - fn default() -> Self { - Self::new() +impl EdgeBucket { + fn with_hasher(hasher: RuntimeHasher) -> Self { + Self { + sources: RoaringBitmap::new(), + author_refs: HashMap::with_hasher(hasher), + } } } +pub struct EdgeStore { + source_interner: Arc>, + did_interner: Arc>, + forward: SccMap, + reverse: SccMap, RuntimeHasher>, + hasher: RuntimeHasher, + writer: Mutex<()>, +} + impl EdgeStore { - pub fn new() -> Self { + pub fn new(hasher: RuntimeHasher) -> Self { Self { - source_interner: Arc::new(ThreadedRodeo::new()), - did_interner: Arc::new(ThreadedRodeo::new()), - forward: SccMap::new(), - reverse: SccMap::new(), + source_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), + did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), + forward: SccMap::with_hasher(hasher.clone()), + reverse: SccMap::with_hasher(hasher.clone()), + hasher, writer: Mutex::new(()), } } @@ -184,7 +189,10 @@ let author = self.intern_author(edge.source.as_ref()); let key = EdgeKey::new(edge.kind, edge.subject); - let mut entry = self.forward.entry_sync(key.clone()).or_default(); + let mut entry = self + .forward + .entry_sync(key.clone()) + .or_insert_with(|| EdgeBucket::with_hasher(self.hasher.clone())); let bucket = entry.get_mut(); let inserted = bucket.sources.insert(id.raw()); if inserted && let Some(author) = author { @@ -284,13 +292,13 @@ rest.split('/').next() } -fn bump_ref(refs: &mut HashMap, author: AuthorId) { +fn bump_ref(refs: &mut HashMap, author: AuthorId) { refs.entry(author) .and_modify(|c| *c = c.saturating_add(1)) .or_insert(NonZeroU32::MIN); } -fn drop_ref(refs: &mut HashMap, author: AuthorId) { +fn drop_ref(refs: &mut HashMap, author: AuthorId) { let Some(counter) = refs.get(&author) else { return; }; @@ -309,6 +317,10 @@ use super::*; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri; + + fn store() -> EdgeStore { + EdgeStore::new(RuntimeHasher::default()) + } fn nsid(s: &'static str) -> Nsid { Nsid::new_static(s).unwrap() @@ -334,7 +346,7 @@ #[test] fn add_then_count() { - let store = EdgeStore::new(); + let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); store.add(star_edge( @@ -356,7 +368,7 @@ #[test] fn duplicate_add_is_idempotent() { - let store = EdgeStore::new(); + let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); let edge = star_edge( "at://did:plc:nel/sh.tangled.feed.star/r1", @@ -370,7 +382,7 @@ #[test] fn remove_source_clears_all_keys_for_that_source() { - let store = EdgeStore::new(); + let store = store(); let star_key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); let follow_key = EdgeKey::new(nsid("sh.tangled.graph.follow"), at("at://did:plc:lyna")); let source = "at://did:plc:nel/sh.tangled.feed.star/r1"; @@ -396,7 +408,7 @@ #[test] fn upsert_source_replaces_old_edges() { - let store = EdgeStore::new(); + let store = store(); let source = at("at://did:plc:teq/sh.tangled.feed.star/r1"); let old_subject = at("at://did:plc:abalone"); let new_subject = at("at://did:plc:uni"); @@ -429,7 +441,7 @@ #[test] fn list_pages_in_id_order() { - let store = EdgeStore::new(); + let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); (0..5).for_each(|i| { store.add(star_edge( @@ -456,7 +468,7 @@ #[test] fn list_exact_fill_signals_exhaustion() { - let store = EdgeStore::new(); + let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); (0..2).for_each(|i| { store.add(star_edge( @@ -472,7 +484,7 @@ #[test] fn list_on_unknown_key_is_empty() { - let store = EdgeStore::new(); + let store = store(); let page = store.list( &EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:periwinkle")), PageCursor::Start, @@ -484,7 +496,7 @@ #[test] fn distinct_authors_decreases_when_last_source_from_author_removed() { - let store = EdgeStore::new(); + let store = store(); let key = EdgeKey::new(nsid("sh.tangled.feed.star"), at("at://did:plc:abalone")); let s1 = "at://did:plc:nel/sh.tangled.feed.star/r1"; let s2 = "at://did:plc:nel/sh.tangled.feed.star/r2"; diff --git a/crates/search/src/lib.rs b/crates/search/src/lib.rs --- a/crates/search/src/lib.rs +++ b/crates/search/src/lib.rs @@ -1,6 +1,9 @@ +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::time::Duration; +use bobbin_runtime::Clock; use bobbin_types::search::{SearchDoc, SearchSink}; use jacquard_common::DefaultStr; use jacquard_common::types::nsid::Nsid; @@ -13,7 +16,6 @@ use tantivy::{Index, IndexReader, IndexWriter, ReloadPolicy, TantivyDocument, TantivyError, Term}; use thiserror::Error; use tokio::sync::{mpsc, oneshot}; -use tokio::time::timeout; use tracing::warn; pub const DEFAULT_WRITER_HEAP_BYTES: usize = 50_000_000; @@ -101,8 +103,6 @@ InvalidNsid(String), #[error("indexed document missing field: {0}")] MissingField(&'static str), - #[error("failed to spawn search writer thread: {0}")] - ThreadSpawn(String), #[error("search blocking task cancelled: {0}")] Cancelled(String), } @@ -133,7 +133,7 @@ } impl SearchIndex { - pub fn new(heap_bytes: usize) -> Result { + pub fn new(heap_bytes: usize, clock: Arc) -> Result { let mut sb = Schema::builder(); let uri = sb.add_text_field("uri", STRING | STORED); let nsid = sb.add_text_field("nsid", STRING | STORED); @@ -162,10 +162,7 @@ }; let (tx, rx) = mpsc::channel(WRITE_QUEUE_CAPACITY); let writer_reader = reader.clone(); - std::thread::Builder::new() - .name("bobbin-search-writer".into()) - .spawn(move || run_writer_thread(writer, fields, writer_reader, rx)) - .map_err(|e| SearchError::ThreadSpawn(e.to_string()))?; + tokio::spawn(writer_loop(writer, fields, writer_reader, rx, clock)); Ok(Self { inner: Arc::new(Inner { fields, @@ -215,6 +212,31 @@ Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()), Err(e) => Err(SearchError::Cancelled(e.to_string())), } + } +} + +pub type SearchReadFuture<'a> = + Pin> + Send + 'a>>; + +pub trait SearchReader: Send + Sync + 'static { + fn search<'a>( + &'a self, + query: &'a str, + nsid_filter: Option<&'a Nsid>, + cursor: SearchCursor, + limit: u32, + ) -> SearchReadFuture<'a>; +} + +impl SearchReader for SearchIndex { + fn search<'a>( + &'a self, + query: &'a str, + nsid_filter: Option<&'a Nsid>, + cursor: SearchCursor, + limit: u32, + ) -> SearchReadFuture<'a> { + Box::pin(SearchIndex::search(self, query, nsid_filter, cursor, limit)) } } @@ -279,56 +301,46 @@ .map(SearchOffset::new) } -fn run_writer_thread( - writer: IndexWriter, - fields: Fields, - reader: IndexReader, - rx: mpsc::Receiver, -) { - let runtime = match tokio::runtime::Builder::new_current_thread() - .enable_time() - .build() - { - Ok(rt) => rt, - Err(e) => { - warn!(?e, "search writer runtime build failed"); - return; - } - }; - runtime.block_on(writer_loop(writer, fields, reader, rx)); -} - async fn writer_loop( mut writer: IndexWriter, fields: Fields, reader: IndexReader, mut rx: mpsc::Receiver, + clock: Arc, ) { let mut pending: usize = 0; + let mut deadline = clock.now_instant() + BATCH_INTERVAL; loop { - match timeout(BATCH_INTERVAL, rx.recv()).await { - Ok(Some(WriteOp::Upsert(doc))) => { + let outcome = tokio::select! { + biased; + msg = rx.recv() => RecvOutcome::Message(msg), + _ = clock.sleep_until(deadline) => RecvOutcome::Idle, + }; + match outcome { + RecvOutcome::Message(Some(WriteOp::Upsert(doc))) => { apply_upsert(&mut writer, fields, doc); pending += 1; } - Ok(Some(WriteOp::Remove(uri))) => { + RecvOutcome::Message(Some(WriteOp::Remove(uri))) => { apply_remove(&mut writer, fields, &uri); pending += 1; } - Ok(Some(WriteOp::Flush(done))) => { + RecvOutcome::Message(Some(WriteOp::Flush(done))) => { if pending > 0 { commit_and_reload(&mut writer, &reader); pending = 0; } let _ = done.send(()); + deadline = clock.now_instant() + BATCH_INTERVAL; continue; } - Ok(None) => break, - Err(_) => { + RecvOutcome::Message(None) => break, + RecvOutcome::Idle => { if pending > 0 { commit_and_reload(&mut writer, &reader); pending = 0; } + deadline = clock.now_instant() + BATCH_INTERVAL; continue; } } @@ -340,6 +352,11 @@ if pending > 0 { commit_and_reload(&mut writer, &reader); } +} + +enum RecvOutcome { + Message(Option), + Idle, } fn apply_upsert(writer: &mut IndexWriter, fields: Fields, doc: SearchDoc) { @@ -404,6 +421,7 @@ #[cfg(test)] mod tests { use super::*; + use bobbin_runtime::SystemClock; use bobbin_types::search::SearchDoc; use jacquard_common::types::nsid::Nsid as NsidType; @@ -425,7 +443,7 @@ } fn build() -> SearchIndex { - SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap() + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap() } #[tokio::test] diff --git a/crates/xrpc/src/lib.rs b/crates/xrpc/src/lib.rs --- a/crates/xrpc/src/lib.rs +++ b/crates/xrpc/src/lib.rs @@ -22,7 +22,7 @@ }; use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; use bobbin_record_lru::RecordStore; -use bobbin_search::{SearchCursor, SearchError, SearchHit, SearchIndex, SearchOffset}; +use bobbin_search::{SearchCursor, SearchError, SearchHit, SearchOffset, SearchReader}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::ids::{EdgeKey, nsid_static}; use bobbin_types::record::RecordBody; @@ -60,9 +60,12 @@ use jacquard_common::xrpc::XrpcResp; use jacquard_common::{DefaultStr, IntoStatic}; use serde::{Deserialize, Serialize}; +use std::time::Duration; use thiserror::Error; -use tower_http::trace::{DefaultMakeSpan, DefaultOnFailure, DefaultOnResponse, TraceLayer}; -use tracing::Level; + +use tower_http::classify::ServerErrorsFailureClass; +use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer}; +use tracing::{Level, Span}; const DEFAULT_LIMIT: u32 = 50; const FETCH_CONCURRENCY: usize = 8; @@ -74,7 +77,7 @@ pub edges: Arc, pub coverage: Arc, pub knots: Arc, - pub search: Arc, + pub search: Arc, } impl AppState { @@ -84,7 +87,7 @@ edges: Arc, coverage: Arc, knots: Arc, - search: Arc, + search: Arc, ) -> Self { Self { records, @@ -172,10 +175,35 @@ TraceLayer::new_for_http() .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) .on_request(()) - .on_response(DefaultOnResponse::new().level(Level::INFO)) - .on_failure(DefaultOnFailure::new().level(Level::WARN)), + .on_response(LatencyFreeTrace) + .on_failure(LatencyFreeTrace), ) .with_state(state) +} + +#[derive(Clone, Copy, Debug)] +struct LatencyFreeTrace; + +impl OnResponse for LatencyFreeTrace { + fn on_response(self, response: &Response, _latency: Duration, _span: &Span) { + tracing::event!( + target: "tower_http::trace::on_response", + Level::INFO, + status = response.status().as_u16(), + "request completed", + ); + } +} + +impl OnFailure for LatencyFreeTrace { + fn on_failure(&mut self, error: ServerErrorsFailureClass, _latency: Duration, _span: &Span) { + tracing::event!( + target: "tower_http::trace::on_failure", + Level::WARN, + error = %error, + "request failed", + ); + } } const REPO_PROXIED_NSIDS: &[&str] = &[ @@ -439,9 +467,11 @@ | E::InvalidAtUri(_) | E::InvalidCid(_) | E::UriMismatch { .. }) => XrpcError::InvalidRecord(e.to_string()), - e @ (E::Transport(_) | E::Upstream(_) | E::BodyTooLarge { .. } | E::BadScheme(_)) => { - XrpcError::UpstreamUnavailable(e.to_string()) - } + e @ (E::Network(_) + | E::Build(_) + | E::Upstream(_) + | E::BodyTooLarge { .. } + | E::BadScheme(_)) => XrpcError::UpstreamUnavailable(e.to_string()), } } @@ -1033,7 +1063,6 @@ | E::InvalidUri(_) | E::InvalidNsid(_) | E::MissingField(_) - | E::ThreadSpawn(_) | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")), } } diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -2,9 +2,10 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, SourceId}; -use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; -use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; use bobbin_xrpc::{AppState, router}; @@ -40,15 +41,25 @@ impl Harness { async fn new() -> Self { let server = MockServer::start().await; - let edges = Arc::new(EdgeStore::new()); + let edges = Arc::new(EdgeStore::new(RuntimeHasher::default())); let coverage = Arc::new(CoverageWatch::new()); let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), - SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(), + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), edges.clone(), coverage.clone(), - Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), - Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), + Arc::new( + KnotProxy::new( + KnotProxyConfig::default(), + KnotHttpConfig::default(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new( + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), + ) as Arc, ); Self { server, diff --git a/crates/xrpc/tests/cold_start.rs b/crates/xrpc/tests/cold_start.rs --- a/crates/xrpc/tests/cold_start.rs +++ b/crates/xrpc/tests/cold_start.rs @@ -2,9 +2,10 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; -use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; -use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use futures::stream::{self, StreamExt}; @@ -21,11 +22,20 @@ async fn fresh_app(server_uri: &str) -> AppState { AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), - SlingshotClient::new(Url::parse(server_uri).unwrap()).unwrap(), - Arc::new(EdgeStore::new()), + SlingshotClient::with_default_http(Url::parse(server_uri).unwrap()).unwrap(), + Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(CoverageWatch::new()), - Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), - Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), + Arc::new( + KnotProxy::new( + KnotProxyConfig::default(), + KnotHttpConfig::default(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap()) + as Arc, ) } diff --git a/crates/xrpc/tests/extended.rs b/crates/xrpc/tests/extended.rs --- a/crates/xrpc/tests/extended.rs +++ b/crates/xrpc/tests/extended.rs @@ -2,9 +2,10 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; -use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; -use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; use bobbin_xrpc::{AppState, router}; @@ -40,15 +41,25 @@ impl Harness { async fn new() -> Self { let server = MockServer::start().await; - let edges = Arc::new(EdgeStore::new()); + let edges = Arc::new(EdgeStore::new(RuntimeHasher::default())); let coverage = Arc::new(CoverageWatch::new()); let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), - SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(), + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), edges.clone(), coverage, - Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), - Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), + Arc::new( + KnotProxy::new( + KnotProxyConfig::default(), + KnotHttpConfig::default(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new( + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), + ) as Arc, ); Self { server, diff --git a/crates/xrpc/tests/knot_proxy.rs b/crates/xrpc/tests/knot_proxy.rs --- a/crates/xrpc/tests/knot_proxy.rs +++ b/crates/xrpc/tests/knot_proxy.rs @@ -3,9 +3,10 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; -use bobbin_knot_proxy::{FailureThreshold, KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{FailureThreshold, KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; -use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; @@ -22,10 +23,15 @@ KnotProxyConfig { failure_threshold: FailureThreshold::new(2).unwrap(), cooldown: Duration::from_millis(80), - connect_timeout: Duration::from_millis(500), - read_timeout: Duration::from_secs(2), allow_private_hosts: true, require_https: false, + } +} + +fn test_http_config() -> KnotHttpConfig { + KnotHttpConfig { + connect_timeout: Duration::from_millis(500), + read_timeout: Duration::from_secs(2), } } @@ -45,11 +51,22 @@ let knot_server = MockServer::start().await; let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), - SlingshotClient::new(Url::parse(&slingshot_server.uri()).unwrap()).unwrap(), - Arc::new(EdgeStore::new()), + SlingshotClient::with_default_http(Url::parse(&slingshot_server.uri()).unwrap()) + .unwrap(), + Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(CoverageWatch::new()), - Arc::new(KnotProxy::new(config).unwrap()), - Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), + Arc::new( + KnotProxy::new( + config, + test_http_config(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new( + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), + ) as Arc, ); Self { slingshot: slingshot_server, diff --git a/crates/xrpc/tests/search.rs b/crates/xrpc/tests/search.rs --- a/crates/xrpc/tests/search.rs +++ b/crates/xrpc/tests/search.rs @@ -2,9 +2,10 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor}; -use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; -use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; +use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::search::{SearchDoc, SearchSink}; use bobbin_xrpc::{AppState, router}; @@ -44,14 +45,24 @@ async fn new() -> Self { let server = MockServer::start().await; let coverage = Arc::new(CoverageWatch::new()); - let search = Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()); + let search = Arc::new( + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), + ); let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), - SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(), - Arc::new(EdgeStore::new()), + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), + Arc::new(EdgeStore::new(RuntimeHasher::default())), coverage.clone(), - Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), - search.clone(), + Arc::new( + KnotProxy::new( + KnotProxyConfig::default(), + KnotHttpConfig::default(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(), + ), + search.clone() as Arc, ); Self { server, -- tangled.sh