From 9af943f284fde7391c90908021c45d4ab01651fa Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Wed, 1 Apr 2026 12:13:52 +0300 Subject: [PATCH] [all] re-organize types and feature gates --- src/api/crawler.rs | 2 +- src/api/debug.rs | 23 +- src/api/ingestion.rs | 8 + src/api/mod.rs | 10 +- src/api/repos.rs | 22 +- src/api/xrpc/mod.rs | 21 +- src/backfill/manager.rs | 11 +- src/backfill/mod.rs | 6 +- src/control/indexer.rs | 75 ++++ src/control/mod.rs | 247 +++++--------- src/control/relay.rs | 40 +++ src/control/{repos.rs => repos/indexer.rs} | 379 ++------------------- src/control/repos/mod.rs | 358 +++++++++++++++++++ src/control/seed.rs | 2 +- src/crawler/worker.rs | 4 +- src/db/indexer.rs | 166 +++++++++ src/db/keys/indexer.rs | 165 +++++++++ src/db/keys/mod.rs | 174 +--------- src/db/mod.rs | 299 +++++----------- src/filter.rs | 39 ++- src/ingest/firehose.rs | 2 +- src/ingest/indexer.rs | 8 +- src/ingest/relay.rs | 5 +- src/lib.rs | 1 + src/ops.rs | 4 +- src/state.rs | 24 +- src/types.rs | 93 ++--- src/util/mod.rs | 2 + 28 files changed, 1188 insertions(+), 1002 deletions(-) create mode 100644 src/control/indexer.rs create mode 100644 src/control/relay.rs rename src/control/{repos.rs => repos/indexer.rs} (63%) create mode 100644 src/control/repos/mod.rs create mode 100644 src/db/indexer.rs create mode 100644 src/db/keys/indexer.rs diff --git a/src/api/crawler.rs b/src/api/crawler.rs index 63283af..85e300a 100644 --- a/src/api/crawler.rs +++ b/src/api/crawler.rs @@ -8,7 +8,7 @@ use serde::Deserialize; use url::Url; use crate::config::{CrawlerMode, CrawlerSource}; -use crate::control::{CrawlerSourceInfo, Hydrant}; +use crate::control::{Hydrant, crawler::CrawlerSourceInfo}; pub fn router() -> Router { Router::new() diff --git a/src/api/debug.rs b/src/api/debug.rs index d494648..f792ef3 100644 --- a/src/api/debug.rs +++ b/src/api/debug.rs @@ -8,26 +8,28 @@ use axum::{ http::StatusCode, }; use jacquard_common::types::cid::Cid; +#[cfg(feature = "indexer")] use jacquard_common::types::ident::AtIdentifier; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::str::FromStr; use std::sync::Arc; +#[cfg(feature = "indexer")] #[derive(Deserialize)] pub struct DebugCountRequest { pub did: String, pub collection: String, } +#[cfg(feature = "indexer")] #[derive(Serialize)] pub struct DebugCountResponse { pub count: usize, } pub fn router() -> axum::Router> { - axum::Router::new() - .route("/debug/count", get(handle_debug_count)) + let r = axum::Router::new() .route("/debug/get", get(handle_debug_get)) .route("/debug/iter", get(handle_debug_iter)) .route("/debug/compact", post(handle_debug_compact)) @@ -35,9 +37,15 @@ pub fn router() -> axum::Router> { "/debug/ephemeral_ttl_tick", post(handle_debug_ephemeral_ttl_tick), ) - .route("/debug/seed_watermark", post(handle_debug_seed_watermark)) + .route("/debug/seed_watermark", post(handle_debug_seed_watermark)); + + #[cfg(feature = "indexer")] + let r = r.route("/debug/count", get(handle_debug_count)); + + r } +#[cfg(feature = "indexer")] pub async fn handle_debug_count( State(state): State>, Query(req): Query, @@ -263,12 +271,17 @@ pub async fn handle_debug_iter( fn get_keyspace_by_name(db: &crate::db::Db, name: &str) -> Result { match name { "repos" => Ok(db.repos.clone()), - "blocks" => Ok(db.blocks.clone()), + "counts" => Ok(db.counts.clone()), "cursors" => Ok(db.cursors.clone()), + #[cfg(feature = "indexer")] + "blocks" => Ok(db.blocks.clone()), + #[cfg(feature = "indexer")] "pending" => Ok(db.pending.clone()), + #[cfg(feature = "indexer")] "resync" => Ok(db.resync.clone()), + #[cfg(feature = "indexer")] "events" => Ok(db.events.clone()), - "counts" => Ok(db.counts.clone()), + #[cfg(feature = "indexer")] "records" => Ok(db.records.clone()), _ => Err(StatusCode::BAD_REQUEST), } diff --git a/src/api/ingestion.rs b/src/api/ingestion.rs index eef678e..9db97da 100644 --- a/src/api/ingestion.rs +++ b/src/api/ingestion.rs @@ -15,25 +15,31 @@ pub fn router() -> Router { #[derive(Serialize)] pub struct IngestionStatus { + #[cfg(feature = "indexer")] pub crawler: bool, pub firehose: bool, + #[cfg(feature = "indexer")] pub backfill: bool, } pub async fn get_ingestion(State(hydrant): State) -> Json { Json(IngestionStatus { + #[cfg(feature = "indexer")] crawler: hydrant.crawler.is_enabled(), firehose: hydrant.firehose.is_enabled(), + #[cfg(feature = "indexer")] backfill: hydrant.backfill.is_enabled(), }) } #[derive(Deserialize)] pub struct IngestionPatch { + #[cfg(feature = "indexer")] #[serde(default)] pub crawler: Option, #[serde(default)] pub firehose: Option, + #[cfg(feature = "indexer")] #[serde(default)] pub backfill: Option, } @@ -42,6 +48,7 @@ pub async fn patch_ingestion( State(hydrant): State, Json(body): Json, ) -> StatusCode { + #[cfg(feature = "indexer")] if let Some(crawler) = body.crawler { if crawler { hydrant.crawler.enable(); @@ -56,6 +63,7 @@ pub async fn patch_ingestion( hydrant.firehose.disable(); } } + #[cfg(feature = "indexer")] if let Some(backfill) = body.backfill { if backfill { hydrant.backfill.enable(); diff --git a/src/api/mod.rs b/src/api/mod.rs index a08cfac..24abcd5 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -5,6 +5,7 @@ use std::{net::SocketAddr, sync::Arc}; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; +#[cfg(feature = "indexer")] mod crawler; mod db; mod debug; @@ -25,16 +26,21 @@ pub async fn serve(hydrant: Hydrant, port: u16) -> miette::Result<()> { .route("/stats", get(stats::get_stats)); #[cfg(feature = "indexer")] let app = app.nest("/stream", stream::router()); - let app = app + #[allow(unused_mut)] + let mut app = app .merge(xrpc::router()) .merge(filter::router()) .merge(pds::router()) .merge(repos::router()) .merge(ingestion::router()) - .merge(crawler::router()) .merge(firehose::router()) .merge(db::router()); + #[cfg(feature = "indexer")] + { + app = app.merge(crawler::router()); + } + #[cfg(feature = "backlinks")] let app = app.merge(crate::backlinks::api::router()); diff --git a/src/api/repos.rs b/src/api/repos.rs index 24b5526..bd7226a 100644 --- a/src/api/repos.rs +++ b/src/api/repos.rs @@ -1,27 +1,36 @@ use std::str::FromStr; use crate::control::{Hydrant, RepoInfo}; +#[cfg(feature = "indexer")] +use axum::routing::{delete, post, put}; use axum::{ Json, Router, body::Body, extract::{Path, Query, State}, http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, - routing::{delete, get, post, put}, + routing::get, }; use jacquard_common::types::did::Did; use miette::IntoDiagnostic; use serde::Deserialize; pub fn router() -> Router { - Router::new() + #[allow(unused_mut)] + let r = Router::new() .route("/repos", get(handle_get_repos)) + .route("/repos/{did}", get(handle_get_repo)); + + #[cfg(feature = "indexer")] + let r = r .route("/repos/resync", post(handle_post_resync)) - .route("/repos/{did}", get(handle_get_repo)) .route("/repos", put(handle_put_repos)) - .route("/repos", delete(handle_delete_repos)) + .route("/repos", delete(handle_delete_repos)); + + r } +#[cfg(feature = "indexer")] #[derive(Deserialize, Debug)] pub struct RepoRequest { pub did: String, @@ -90,6 +99,7 @@ pub async fn handle_get_repo( .ok_or_else(|| (StatusCode::NOT_FOUND, "repository not found".to_string())) } +#[cfg(feature = "indexer")] pub async fn handle_put_repos( State(hydrant): State, headers: HeaderMap, @@ -111,6 +121,7 @@ pub async fn handle_put_repos( Ok(did_list_response(queued, &headers)) } +#[cfg(feature = "indexer")] pub async fn handle_delete_repos( State(hydrant): State, headers: HeaderMap, @@ -132,6 +143,7 @@ pub async fn handle_delete_repos( Ok(did_list_response(untracked, &headers)) } +#[cfg(feature = "indexer")] pub async fn handle_post_resync( State(hydrant): State, headers: HeaderMap, @@ -159,6 +171,7 @@ fn prefers_json(headers: &HeaderMap) -> bool { contains_json(header::ACCEPT) || contains_json(header::CONTENT_TYPE) } +#[cfg(feature = "indexer")] fn did_list_response(dids: Vec>, headers: &HeaderMap) -> Response { if prefers_json(headers) { let body: Vec = dids.into_iter().map(|d| d.to_string()).collect(); @@ -173,6 +186,7 @@ fn did_list_response(dids: Vec>, headers: &HeaderMap) -> Response { } } +#[cfg(feature = "indexer")] async fn parse_body( body: Body, headers: &HeaderMap, diff --git a/src/api/xrpc/mod.rs b/src/api/xrpc/mod.rs index a717555..18dc7d7 100644 --- a/src/api/xrpc/mod.rs +++ b/src/api/xrpc/mod.rs @@ -2,20 +2,22 @@ use crate::control::Hydrant; use axum::extract::FromRequest; use axum::response::IntoResponse; use axum::routing::get; +#[cfg(feature = "relay")] +use axum::routing::post; use axum::{Json, Router, extract::State, http::StatusCode}; use jacquard_api::com_atproto::sync::get_host_status::GetHostStatusRequest; use jacquard_api::com_atproto::sync::get_latest_commit::GetLatestCommitRequest; use jacquard_api::com_atproto::sync::get_repo_status::GetRepoStatusRequest; use jacquard_api::com_atproto::sync::list_hosts::ListHostsRequest; use jacquard_api::com_atproto::sync::list_repos::ListReposRequest; +#[cfg(feature = "indexer")] use jacquard_common::types::ident::AtIdentifier; +#[cfg(feature = "indexer")] +use jacquard_common::types::string::AtUri; use jacquard_common::xrpc::XrpcResp; +use jacquard_common::xrpc::{GenericXrpcError, XrpcError}; use jacquard_common::xrpc::{XrpcEndpoint, XrpcMethod}; use jacquard_common::{IntoStatic, xrpc::XrpcRequest}; -use jacquard_common::{ - types::string::AtUri, - xrpc::{GenericXrpcError, XrpcError}, -}; use serde::{Deserialize, Serialize}; use smol_str::ToSmolStr; use std::fmt::Display; @@ -84,14 +86,8 @@ pub fn router() -> Router { #[cfg(feature = "relay")] let r = r - .route( - SubscribeReposEndpoint::PATH, - axum::routing::get(subscribe_repos::handle), - ) - .route( - RequestCrawlRequest::PATH, - axum::routing::get(subscribe_repos::handle), - ); + .route(SubscribeReposEndpoint::PATH, get(subscribe_repos::handle)) + .route(RequestCrawlRequest::PATH, post(request_crawl::handle)); r } @@ -177,6 +173,7 @@ fn bad_request( } } +#[cfg(feature = "indexer")] fn upstream_error( nsid: &'static str, message: impl Display, diff --git a/src/backfill/manager.rs b/src/backfill/manager.rs index a5e91df..74486f3 100644 --- a/src/backfill/manager.rs +++ b/src/backfill/manager.rs @@ -43,7 +43,7 @@ pub fn queue_gone_backfills(state: &Arc) -> Result<()> { continue; } }; - let mut metadata = crate::db::deser_repo_metadata(&metadata_bytes)?; + let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; // move from resync back into pending batch.remove(&state.db.resync, key.clone()); @@ -58,7 +58,7 @@ pub fn queue_gone_backfills(state: &Arc) -> Result<()> { batch.insert( &state.db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); transitions.push((GaugeState::Resync(None), GaugeState::Pending)); @@ -133,9 +133,8 @@ pub fn retry_worker(state: Arc) { continue; } }; - let mut metadata = match crate::db::deser_repo_metadata( - metadata_bytes.as_ref(), - ) { + let mut metadata = match crate::db::deser_repo_meta(metadata_bytes.as_ref()) + { Ok(m) => m, Err(e) => { error!(did = %did, err = %e, "failed to deserialize repo metadata"); @@ -153,7 +152,7 @@ pub fn retry_worker(state: Arc) { keys::pending_key(metadata.index_id), key.clone(), ); - let serialized_metadata = match crate::db::ser_repo_metadata(&metadata) { + let serialized_metadata = match crate::db::ser_repo_meta(&metadata) { Ok(s) => s, Err(e) => { error!(did = %did, err = %e, "failed to serialize repo metadata"); diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 92a80ac..2418b5b 100644 --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -742,12 +742,12 @@ async fn process_did<'i>( .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let mut metadata = crate::db::deser_repo_metadata(&metadata_bytes)?; + let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; metadata.tracked = true; batch.insert( &app_state.db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); // add the counts @@ -771,7 +771,7 @@ async fn process_did<'i>( .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_metadata(metadata_bytes.as_ref())?; + let metadata = crate::db::deser_repo_meta(metadata_bytes.as_ref())?; let Some((_state, records_cnt_delta, added_blocks, count)) = result else { // signal mode: no signal-matching records found, clean up the optimistically-added repo diff --git a/src/control/indexer.rs b/src/control/indexer.rs new file mode 100644 index 0000000..fc083fb --- /dev/null +++ b/src/control/indexer.rs @@ -0,0 +1,75 @@ +use super::*; + +/// a stream of [`Event`]s. returned by [`Hydrant::subscribe`]. +/// +/// implements [`futures::Stream`] and can be used with `StreamExt::next`, +/// `while let Some(evt) = stream.next().await`, `forward`, etc. +/// the stream terminates when the underlying channel closes (i.e. hydrant shuts down). +pub struct EventStream(mpsc::Receiver); + +impl Stream for EventStream { + type Item = Event; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.0.poll_recv(cx) + } +} + +/// runtime control over the backfill worker component. +/// +/// the backfill worker fetches full repo CAR files from each repo's PDS for any +/// repository in the pending queue, parses the MST, and inserts all matching records +/// into the database. concurrency is bounded by `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`. +#[derive(Clone)] +pub struct BackfillHandle(Arc); + +impl BackfillHandle { + pub(crate) fn new(state: Arc) -> Self { + Self(state) + } + + /// enable the backfill worker, no-op if already enabled. + pub fn enable(&self) { + self.0.backfill_enabled.send_replace(true); + } + /// disable the backfill worker, in-flight repos complete before pausing. + pub fn disable(&self) { + self.0.backfill_enabled.send_replace(false); + } + /// returns the current enabled state of the backfill worker. + pub fn is_enabled(&self) -> bool { + *self.0.backfill_enabled.borrow() + } +} + +impl Hydrant { + /// subscribe to the ordered event stream. + /// + /// returns an [`EventStream`] that implements [`futures::Stream`]. + /// + /// - if `cursor` is `None`, streaming starts from the current head (live tail only). + /// - if `cursor` is `Some(id)`, all persisted `record` events from that ID onward are + /// replayed first, then the stream will switch to live tailing. + /// + /// `identity` and `account` events are ephemeral and are never replayed from a cursor, + /// only live ones are delivered. use [`ReposControl::info`] to fetch current state for + /// a specific repository. + /// + /// multiple concurrent subscribers each receive a full independent copy of the stream. + /// the stream ends when the `EventStream` is dropped. + pub fn subscribe(&self, cursor: Option) -> EventStream { + let (tx, rx) = mpsc::channel(500); + let state = self.state.clone(); + let runtime = tokio::runtime::Handle::current(); + + std::thread::Builder::new() + .name("hydrant-stream".into()) + .spawn(move || { + let _g = runtime.enter(); + event_stream_thread(state, tx, cursor); + }) + .expect("failed to spawn stream thread"); + + EventStream(rx) + } +} diff --git a/src/control/mod.rs b/src/control/mod.rs index 7740d69..255fa47 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -1,5 +1,6 @@ #![allow(unused_imports)] +#[cfg(feature = "indexer")] pub(crate) mod crawler; pub(crate) mod filter; pub(crate) mod firehose; @@ -8,7 +9,16 @@ pub(crate) mod repos; mod seed; pub(crate) mod stream; -pub use crawler::{CrawlerHandle, CrawlerSourceInfo}; +#[cfg(feature = "indexer")] +mod indexer; +#[cfg(feature = "indexer")] +pub use indexer::*; + +#[cfg(feature = "relay")] +mod relay; +#[cfg(feature = "relay")] +pub use relay::*; + pub use filter::{FilterControl, FilterPatch, FilterSnapshot}; pub use firehose::{FirehoseHandle, FirehoseSourceInfo}; pub use pds::{PdsControl, PdsTierAssignment, PdsTierDefinition}; @@ -30,17 +40,15 @@ use tracing::{debug, error, info}; #[cfg(feature = "indexer")] use crate::backfill::BackfillWorker; use crate::config::{Config, SignatureVerification}; -use crate::db::{ - self, filter as db_filter, keys, load_persisted_crawler_sources, - load_persisted_firehose_sources, -}; +#[cfg(feature = "indexer")] +use crate::db::load_persisted_crawler_sources; +use crate::db::{self, filter as db_filter, keys, load_persisted_firehose_sources}; use crate::filter::FilterMode; #[cfg(feature = "indexer")] use crate::ingest::indexer::FirehoseWorker; use crate::state::AppState; use crate::types::MarshallableEvt; -use crawler::{CrawlerShared, spawn_crawler_producer}; use firehose::{FirehoseShared, spawn_firehose_ingestor}; #[cfg(feature = "indexer")] use stream::event_stream_thread; @@ -86,8 +94,10 @@ pub type Event = MarshallableEvt<'static>; /// ``` #[derive(Clone)] pub struct Hydrant { - pub crawler: CrawlerHandle, + #[cfg(feature = "indexer")] + pub crawler: crawler::CrawlerHandle, pub firehose: FirehoseHandle, + #[cfg(feature = "indexer")] pub backfill: BackfillHandle, pub filter: FilterControl, pub pds: PdsControl, @@ -160,26 +170,32 @@ impl Hydrant { state.filter.store(Arc::new(new_filter)); } - // 4. set crawler enabled state from config, evaluated against the post-patch filter - let post_patch_crawler = match config.enable_crawler { - Some(b) => b, - None => { - state.filter.load().mode == FilterMode::Full || !config.crawler_sources.is_empty() - } - }; - state.crawler_enabled.send_replace(post_patch_crawler); + #[cfg(feature = "indexer")] + { + // 4. set crawler enabled state from config, evaluated against the post-patch filter + let post_patch_crawler = match config.enable_crawler { + Some(b) => b, + None => { + state.filter.load().mode == FilterMode::Full + || !config.crawler_sources.is_empty() + } + }; + state.crawler_enabled.send_replace(post_patch_crawler); + } let state = Arc::new(state); Ok(Self { - crawler: CrawlerHandle { + #[cfg(feature = "indexer")] + crawler: crawler::CrawlerHandle { state: state.clone(), shared: Arc::new(std::sync::OnceLock::new()), tasks: Arc::new(scc::HashMap::new()), persisted: Arc::new(scc::HashSet::new()), }, firehose: FirehoseHandle::new(state.clone()), - backfill: BackfillHandle(state.clone()), + #[cfg(feature = "indexer")] + backfill: BackfillHandle::new(state.clone()), filter: FilterControl(state.clone()), pds: pds::PdsControl(state.clone()), repos: ReposControl(state.clone()), @@ -209,6 +225,7 @@ impl Hydrant { pub fn run(&self) -> Result>> { let state = self.state.clone(); let config = self.config.clone(); + #[cfg(feature = "indexer")] let crawler = self.crawler.clone(); let firehose = self.firehose.clone(); @@ -357,13 +374,6 @@ impl Hydrant { let (fatal_tx_inner, mut fatal_rx) = watch::channel(None); let fatal_tx = Arc::new(fatal_tx_inner); - info!( - crawler_enabled = *state.crawler_enabled.borrow(), - firehose_enabled = *state.firehose_enabled.borrow(), - filter_mode = ?state.filter.load().mode, - "starting ingestion" - ); - // 10. set shared and spawn firehose ingestors firehose .shared @@ -516,7 +526,7 @@ impl Hydrant { // set shared objects so CrawlerHandle methods can use them crawler .shared - .set(CrawlerShared { + .set(crawler::CrawlerShared { http, checker, in_flight, @@ -530,7 +540,7 @@ impl Hydrant { // spawn initial sources from config for source in config.crawler_sources.iter() { let enabled_rx = state.crawler_enabled.subscribe(); - let handle = spawn_crawler_producer( + let handle = crawler::spawn_crawler_producer( source, &shared.http, &state, @@ -556,7 +566,7 @@ impl Hydrant { continue; } let enabled_rx = state.crawler_enabled.subscribe(); - let handle = spawn_crawler_producer( + let handle = crawler::spawn_crawler_producer( source, &shared.http, &state, @@ -662,61 +672,6 @@ impl Hydrant { Ok(fut) } - /// subscribe to the ordered event stream. - /// - /// returns an [`EventStream`] that implements [`futures::Stream`]. - /// - /// - if `cursor` is `None`, streaming starts from the current head (live tail only). - /// - if `cursor` is `Some(id)`, all persisted `record` events from that ID onward are - /// replayed first, then the stream will switch to live tailing. - /// - /// `identity` and `account` events are ephemeral and are never replayed from a cursor, - /// only live ones are delivered. use [`ReposControl::info`] to fetch current state for - /// a specific repository. - /// - /// multiple concurrent subscribers each receive a full independent copy of the stream. - /// the stream ends when the `EventStream` is dropped. - #[cfg(feature = "indexer")] - pub fn subscribe(&self, cursor: Option) -> EventStream { - let (tx, rx) = mpsc::channel(500); - let state = self.state.clone(); - let runtime = tokio::runtime::Handle::current(); - - std::thread::Builder::new() - .name("hydrant-stream".into()) - .spawn(move || { - let _g = runtime.enter(); - event_stream_thread(state, tx, cursor); - }) - .expect("failed to spawn stream thread"); - - EventStream(rx) - } - - /// subscribe to the relay's ordered `subscribeRepos` event stream. - /// - /// returns a [`RelayEventStream`] that yields pre-encoded CBOR binary frames - /// ready to forward directly to ATProto clients via WebSocket. - /// - /// - if `cursor` is `None`, streaming starts from the current head (live tail only). - /// - if `cursor` is `Some(seq)`, all persisted events from that seq onward are replayed first. - #[cfg(feature = "relay")] - pub fn subscribe_repos(&self, cursor: Option) -> RelayEventStream { - let (tx, rx) = mpsc::channel(500); - let state = self.state.clone(); - let runtime = tokio::runtime::Handle::current(); - - std::thread::Builder::new() - .name("hydrant-relay-stream".into()) - .spawn(move || { - let _g = runtime.enter(); - relay_stream_thread(state, tx, cursor); - }) - .expect("failed to spawn relay stream thread"); - - RelayEventStream(rx) - } - /// return database counts and on-disk sizes for all keyspaces. /// /// counts include: `repos`, `pending`, `resync`, `records`, `blocks`, `events`, @@ -726,44 +681,64 @@ impl Hydrant { pub async fn stats(&self) -> Result { let state = self.state.clone(); - // todo: update stats, only return necessary info on relay vs indexer modes - // (and ephemeral indexer) - let mut counts: BTreeMap<&'static str, u64> = futures::future::join_all( - [ - "repos", - "pending", - "records", - "blocks", - "resync", - "error_ratelimited", - "error_transport", - "error_generic", - ] - .into_iter() - .map(|name| { + #[allow(unused_mut)] + let mut count_keys = vec![ + "repos", + "error_ratelimited", + "error_transport", + "error_generic", + ]; + + #[cfg(feature = "indexer")] + { + count_keys.push("pending"); + count_keys.push("records"); + count_keys.push("blocks"); + count_keys.push("resync"); + } + + let mut counts: BTreeMap<&'static str, u64> = + futures::future::join_all(count_keys.into_iter().map(|name| { let state = state.clone(); async move { (name, state.db.get_count(name).await) } - }), - ) - .await - .into_iter() - .collect(); + })) + .await + .into_iter() + .collect(); + #[cfg(feature = "indexer")] counts.insert("events", state.db.events.approximate_len() as u64); + #[cfg(feature = "relay")] + counts.insert( + "relay_events", + state.db.relay_events.approximate_len() as u64, + ); + let sizes = tokio::task::spawn_blocking(move || { let mut s = BTreeMap::new(); s.insert("repos", state.db.repos.disk_space()); - s.insert("records", state.db.records.disk_space()); - s.insert("blocks", state.db.blocks.disk_space()); s.insert("cursors", state.db.cursors.disk_space()); - s.insert("pending", state.db.pending.disk_space()); - s.insert("resync", state.db.resync.disk_space()); - s.insert("resync_buffer", state.db.resync_buffer.disk_space()); - s.insert("events", state.db.events.disk_space()); s.insert("counts", state.db.counts.disk_space()); s.insert("filter", state.db.filter.disk_space()); s.insert("crawler", state.db.crawler.disk_space()); + + #[cfg(feature = "indexer")] + { + s.insert("records", state.db.records.disk_space()); + s.insert("blocks", state.db.blocks.disk_space()); + s.insert("pending", state.db.pending.disk_space()); + s.insert("resync", state.db.resync.disk_space()); + s.insert("resync_buffer", state.db.resync_buffer.disk_space()); + s.insert("events", state.db.events.disk_space()); + } + + #[cfg(feature = "relay")] + s.insert("relay_events", state.db.relay_events.disk_space()); + + #[cfg(feature = "backlinks")] + s.insert("backlinks", state.db.backlinks.disk_space()); + s }) .await @@ -890,39 +865,6 @@ impl axum::extract::FromRef for Arc { } } -/// a stream of [`Event`]s. returned by [`Hydrant::subscribe`]. -/// -/// implements [`futures::Stream`] and can be used with `StreamExt::next`, -/// `while let Some(evt) = stream.next().await`, `forward`, etc. -/// the stream terminates when the underlying channel closes (i.e. hydrant shuts down). -#[cfg(feature = "indexer")] -pub struct EventStream(mpsc::Receiver); - -#[cfg(feature = "indexer")] -impl Stream for EventStream { - type Item = Event; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.0.poll_recv(cx) - } -} - -/// the relay event stream produced by [`Hydrant::subscribe_repos`]. -#[cfg(feature = "relay")] -pub struct RelayEventStream(mpsc::Receiver); - -#[cfg(feature = "relay")] -impl futures::Stream for RelayEventStream { - type Item = bytes::Bytes; - - fn poll_next( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - self.0.poll_recv(cx) - } -} - /// database statistics returned by [`Hydrant::stats`]. #[derive(serde::Serialize)] pub struct StatsResponse { @@ -932,29 +874,6 @@ pub struct StatsResponse { pub sizes: BTreeMap<&'static str, u64>, } -/// runtime control over the backfill worker component. -/// -/// the backfill worker fetches full repo CAR files from each repo's PDS for any -/// repository in the pending queue, parses the MST, and inserts all matching records -/// into the database. concurrency is bounded by `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`. -#[derive(Clone)] -pub struct BackfillHandle(Arc); - -impl BackfillHandle { - /// enable the backfill worker, no-op if already enabled. - pub fn enable(&self) { - self.0.backfill_enabled.send_replace(true); - } - /// disable the backfill worker, in-flight repos complete before pausing. - pub fn disable(&self) { - self.0.backfill_enabled.send_replace(false); - } - /// returns the current enabled state of the backfill worker. - pub fn is_enabled(&self) -> bool { - *self.0.backfill_enabled.borrow() - } -} - /// control over database maintenance operations. /// /// all methods pause the crawler, firehose, and backfill worker for the duration diff --git a/src/control/relay.rs b/src/control/relay.rs new file mode 100644 index 0000000..fdb4d25 --- /dev/null +++ b/src/control/relay.rs @@ -0,0 +1,40 @@ +use super::*; + +/// the relay event stream produced by [`Hydrant::subscribe_repos`]. +pub struct RelayEventStream(mpsc::Receiver); + +impl futures::Stream for RelayEventStream { + type Item = bytes::Bytes; + + fn poll_next( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.0.poll_recv(cx) + } +} + +impl Hydrant { + /// subscribe to the relay's ordered `subscribeRepos` event stream. + /// + /// returns a [`RelayEventStream`] that yields pre-encoded CBOR binary frames + /// ready to forward directly to ATProto clients via WebSocket. + /// + /// - if `cursor` is `None`, streaming starts from the current head (live tail only). + /// - if `cursor` is `Some(seq)`, all persisted events from that seq onward are replayed first. + pub fn subscribe_repos(&self, cursor: Option) -> RelayEventStream { + let (tx, rx) = mpsc::channel(500); + let state = self.state.clone(); + let runtime = tokio::runtime::Handle::current(); + + std::thread::Builder::new() + .name("hydrant-relay-stream".into()) + .spawn(move || { + let _g = runtime.enter(); + relay_stream_thread(state, tx, cursor); + }) + .expect("failed to spawn relay stream thread"); + + RelayEventStream(rx) + } +} diff --git a/src/control/repos.rs b/src/control/repos/indexer.rs similarity index 63% rename from src/control/repos.rs rename to src/control/repos/indexer.rs index 237b6be..7c7c09b 100644 --- a/src/control/repos.rs +++ b/src/control/repos/indexer.rs @@ -1,121 +1,15 @@ -use std::collections::HashMap; -use std::sync::Arc; - -use chrono::{DateTime, Utc}; -use fjall::OwnedWriteBatch; -use futures::TryFutureExt; -use jacquard_common::cowstr::ToCowStr; -use jacquard_common::types::cid::{Cid, IpldCid}; -use jacquard_common::types::ident::AtIdentifier; -use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::string::{Did, Handle, Rkey}; -use jacquard_common::types::tid::Tid; -use jacquard_common::{CowStr, Data, IntoStatic}; -use miette::{Context, IntoDiagnostic, Result}; +use futures::{FutureExt, TryFutureExt}; use rand::Rng; -use smol_str::ToSmolStr; -use url::Url; - -use crate::db::types::{DbRkey, DidKey, TrimmedDid}; -use crate::db::{self, Db, keys}; -use crate::state::AppState; -use crate::types::{GaugeState, RepoMetadata, RepoState, RepoStatus}; -use crate::util::invalid_handle; - -/// information about a tracked or known repository. returned by [`ReposControl`] methods. -#[derive(Debug, Clone, serde::Serialize)] -pub struct RepoInfo { - /// the DID of the repository. - pub did: Did<'static>, - /// the status of the repository. - #[serde(serialize_with = "crate::util::repo_status_serialize_str")] - pub status: RepoStatus, - /// whether this repository is tracked or not. - /// untracked repositories are not updated and they stay frozen. - pub tracked: bool, - /// the revision of the root commit of this repository. - #[serde(skip_serializing_if = "Option::is_none")] - pub rev: Option, - /// the CID of the MST root of this repository. - #[serde(serialize_with = "crate::util::opt_cid_serialize_str")] - #[serde(skip_serializing_if = "Option::is_none")] - pub data: Option, - /// the handle for the DID of this repository. - /// - /// note that this handle is not bi-directionally verified. - #[serde(skip_serializing_if = "Option::is_none")] - pub handle: Option>, - /// the URL for the PDS in which this repository is hosted on. - #[serde(skip_serializing_if = "Option::is_none")] - pub pds: Option, - /// ATProto signing key of this repository. - #[serde(serialize_with = "crate::util::opt_did_key_serialize_str")] - #[serde(skip_serializing_if = "Option::is_none")] - pub signing_key: Option>, - /// when this repository was last touched (status update, commit ingested, etc.). - #[serde(skip_serializing_if = "Option::is_none")] - pub last_updated_at: Option>, - /// the time of the last message gotten from the firehose for this repository. - /// this is equal to the `time` field. - #[serde(skip_serializing_if = "Option::is_none")] - pub last_message_at: Option>, -} -/// control over which repositories are tracked and access to their state. -/// -/// in `filter` mode, a repo is only indexed if it either matches a signal or is -/// explicitly tracked via [`ReposControl::track`]. in `full` mode all repos are -/// indexed and tracking is implicit. -/// -/// tracking a DID that hydrant has never seen enqueues an immediate backfill. -/// tracking a DID that hydrant already knows about (but has marked untracked) -/// re-enqueues it for backfill. -#[derive(Clone)] -pub struct ReposControl(pub(super) Arc); +use super::*; impl ReposControl { - pub(crate) fn iter_states( - &self, - cursor: Option<&Did<'_>>, - ) -> impl Iterator, RepoState<'static>, crate::types::RepoMetadata)>> - { - let start_bound = if let Some(cursor) = cursor { - let did_key = keys::repo_key(cursor); - std::ops::Bound::Excluded(did_key) - } else { - std::ops::Bound::Unbounded - }; - - let state = self.0.clone(); - self.0 - .db - .repos - .range((start_bound, std::ops::Bound::Unbounded)) - .map(move |g| { - let (k, v) = g.into_inner().into_diagnostic()?; - let repo_state = crate::db::deser_repo_state(&v)?.into_static(); - let did = TrimmedDid::try_from(k.as_ref())?.to_did(); - let metadata_key = keys::repo_metadata_key(&did); - let metadata = state - .db - .repo_metadata - .get(&metadata_key) - .into_diagnostic()? - .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_metadata(metadata.as_ref())?; - Ok((did, repo_state, metadata)) - }) - } - - /// iterates through all repositories, returning their state. - pub fn iter(&self, cursor: Option<&Did<'_>>) -> impl Iterator> { - self.iter_states(cursor) - .map(|r| r.map(|(did, s, m)| repo_state_to_info(did, s, m.tracked))) - } - - #[allow(dead_code)] /// iterates through pending repositories, returning their state. - fn iter_pending(&self, cursor: Option) -> impl Iterator> { + #[allow(dead_code)] + pub(crate) fn iter_pending( + &self, + cursor: Option, + ) -> impl Iterator> { let start_bound = if let Some(cursor) = cursor { std::ops::Bound::Excluded(cursor.to_be_bytes().to_vec()) } else { @@ -151,7 +45,7 @@ impl ReposControl { .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_metadata(metadata.as_ref())?; + let metadata = crate::db::deser_repo_meta(metadata.as_ref())?; Ok(Some(( id, repo_state_to_info(did, repo_state.into_static(), metadata.tracked), @@ -162,7 +56,10 @@ impl ReposControl { } #[allow(dead_code)] - fn iter_resync(&self, cursor: Option<&Did<'_>>) -> impl Iterator> { + pub(crate) fn iter_resync( + &self, + cursor: Option<&Did<'_>>, + ) -> impl Iterator> { let start_bound = if let Some(cursor) = cursor { let did_key = keys::repo_key(cursor); std::ops::Bound::Excluded(did_key) @@ -192,7 +89,7 @@ impl ReposControl { .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_metadata(metadata.as_ref())?; + let metadata = crate::db::deser_repo_meta(metadata.as_ref())?; Ok(Some(repo_state_to_info( did, repo_state.into_static(), @@ -203,34 +100,10 @@ impl ReposControl { .flatten() } - /// gets a handle for a repository to read from it. - pub fn get<'i>(&self, did: &Did<'i>) -> RepoHandle<'i> { - RepoHandle { - state: self.0.clone(), - did: did.clone(), - } - } - - /// same as [`ReposControl::get`] but allows you to pass in an identifier that can be - /// either a handle or a DID. - pub async fn resolve(&self, repo: &AtIdentifier<'_>) -> Result> { - let did = self.0.resolver.resolve_did(repo).await?; - Ok(RepoHandle { - state: self.0.clone(), - did, - }) - } - - /// fetch the current state of a repository. - /// returns `None` if hydrant has never seen this repository. - pub async fn info(&self, did: &Did<'_>) -> Result> { - self.get(did).info().await - } - - fn _resync( + pub(crate) fn _resync( db: &Db, did: &Did<'_>, - batch: &mut OwnedWriteBatch, + batch: &mut fjall::OwnedWriteBatch, transitions: &mut Vec<(GaugeState, GaugeState)>, ) -> Result { let did_key = keys::repo_key(did); @@ -248,7 +121,7 @@ impl ReposControl { .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let mut metadata = crate::db::deser_repo_metadata(&metadata_bytes)?; + let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; // skip if already in pending queue let is_pending = db @@ -269,7 +142,7 @@ impl ReposControl { batch.insert( &db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); transitions.push((old, GaugeState::Pending)); return Ok(true); @@ -341,7 +214,6 @@ impl ReposControl { let mut added = 0i64; let mut queued: Vec> = Vec::new(); let mut transitions: Vec<(GaugeState, GaugeState)> = Vec::new(); - let mut rng = rand::rng(); for did in dids { let did_key = keys::repo_key(&did); @@ -349,7 +221,7 @@ impl ReposControl { let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; let existing_metadata = metadata_bytes - .map(|b| crate::db::deser_repo_metadata(&b)) + .map(|b| crate::db::deser_repo_meta(&b)) .transpose()?; if let Some(metadata) = existing_metadata { @@ -358,12 +230,12 @@ impl ReposControl { } } else { let repo_state = RepoState::backfilling(); - let metadata = RepoMetadata::backfilling(rng.next_u64()); + let metadata = RepoMetadata::backfilling(rand::random()); batch.insert(&db.repos, &did_key, crate::db::ser_repo_state(&repo_state)?); batch.insert( &db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); batch.insert(&db.pending, keys::pending_key(metadata.index_id), &did_key); added += 1; @@ -418,7 +290,7 @@ impl ReposControl { if let Some(repo_state) = existing { let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; let existing_metadata = metadata_bytes - .map(|b| crate::db::deser_repo_metadata(&b)) + .map(|b| crate::db::deser_repo_meta(&b)) .transpose()?; if let Some(mut metadata) = existing_metadata { @@ -429,7 +301,7 @@ impl ReposControl { batch.insert( &db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); batch.remove(&db.pending, keys::pending_key(metadata.index_id)); batch.remove(&db.resync, &did_key); @@ -459,214 +331,7 @@ impl ReposControl { } } -pub(crate) fn repo_state_to_info(did: Did<'static>, s: RepoState<'_>, tracked: bool) -> RepoInfo { - let (rev, data) = s - .root - .map(|c| (Some(c.rev.to_tid()), Some(c.data))) - .unwrap_or_default(); - RepoInfo { - did, - status: s.status, - tracked, - rev, - data, - handle: s.handle.map(|h| h.into_static()), - pds: s.pds.and_then(|p| p.parse().ok()), - signing_key: s.signing_key.map(|k| k.into_static()), - last_updated_at: DateTime::from_timestamp_secs(s.last_updated_at), - last_message_at: s.last_message_time.and_then(DateTime::from_timestamp_secs), - } -} - -pub struct Record { - pub did: Did<'static>, - pub cid: Cid<'static>, - pub value: Data<'static>, -} - -pub struct ListedRecord { - pub rkey: Rkey<'static>, - pub cid: Cid<'static>, - pub value: Data<'static>, -} - -pub struct RecordList { - pub records: Vec, - pub cursor: Option>, -} - -#[derive(Debug, thiserror::Error)] -pub enum MiniDocError { - #[error("repo is not synced yet")] - NotSynced, - #[error("repo not found")] - RepoNotFound, - #[error("could not resolve identity")] - CouldNotResolveIdentity, - #[error("{0}")] - Other(miette::Error), -} - -/// a mini doc with a bi-directionally verified handle. -pub struct MiniDoc<'i> { - /// the did. - pub did: Did<'i>, - /// the handle. if verification fails or no handle is found, - /// this will be "handle.invalid". - pub handle: Handle<'i>, - /// the url of the PDS of this repo. - pub pds: Url, - /// the atproto signing key of this repo. - pub signing_key: DidKey<'i>, -} - -/// handle to access data related to this repository. -#[derive(Clone)] -pub struct RepoHandle<'i> { - state: Arc, - pub did: Did<'i>, -} - impl<'i> RepoHandle<'i> { - pub(crate) async fn state(&self) -> Result>> { - let did_key = keys::repo_key(&self.did); - let app_state = self.state.clone(); - - tokio::task::spawn_blocking(move || { - let bytes = app_state.db.repos.get(&did_key).into_diagnostic()?; - bytes - .as_deref() - .map(db::deser_repo_state) - .transpose() - .map(|opt| opt.map(IntoStatic::into_static)) - }) - .await - .into_diagnostic()? - } - - /// fetch the current state of this repository. - /// returns `None` if hydrant has never seen this repository. - pub async fn info(&self) -> Result> { - let did = self.did.clone().into_static(); - let did_key = keys::repo_key(&did); - let metadata_key = keys::repo_metadata_key(&did); - let app_state = self.state.clone(); - - tokio::task::spawn_blocking(move || { - let state_bytes = app_state.db.repos.get(&did_key).into_diagnostic()?; - let Some(state_bytes) = state_bytes else { - return Ok(None); - }; - let repo_state = crate::db::deser_repo_state(&state_bytes)?; - - let metadata_bytes = app_state - .db - .repo_metadata - .get(&metadata_key) - .into_diagnostic()? - .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_metadata(&metadata_bytes)?; - - Ok(Some(repo_state_to_info(did, repo_state, metadata.tracked))) - }) - .await - .into_diagnostic()? - } - - /// returns the collections of this repository and the number of records it has in each. - pub async fn collections(&self) -> Result, u64>> { - let did = self.did.clone().into_static(); - let state = self.state.clone(); - - tokio::task::spawn_blocking(move || { - let prefix = keys::did_collection_prefix(&did); - let mut res = HashMap::new(); - for item in state.db.counts.prefix(&prefix) { - let (k, v) = item.into_inner().into_diagnostic()?; - let col = k - .strip_prefix(prefix.as_slice()) - .ok_or_else(|| miette::miette!("invalid collection count key: {k:?}")) - .and_then(|r| std::str::from_utf8(r).into_diagnostic()) - .and_then(|n| Nsid::new(n).into_diagnostic())? - .into_static(); - let count = u64::from_be_bytes( - v.as_ref() - .try_into() - .into_diagnostic() - .wrap_err("expected to be count (8 bytes)")?, - ); - res.insert(col, count); - } - Ok(res) - }) - .await - .into_diagnostic()? - } - - /// returns a bi-directionally validated mini doc. - pub async fn mini_doc(&self) -> Result, MiniDocError> { - let Some(info) = self.info().await.map_err(MiniDocError::Other)? else { - return Err(MiniDocError::RepoNotFound); - }; - - // check if repo is still backfilling (in pending) - let metadata_key = keys::repo_metadata_key(&self.did); - let app_state = self.state.clone(); - - let is_pending = tokio::task::spawn_blocking(move || { - let metadata_bytes = app_state - .db - .repo_metadata - .get(&metadata_key) - .into_diagnostic()?; - let Some(metadata_bytes) = metadata_bytes else { - return Ok::<_, miette::Report>(false); - }; - let metadata = crate::db::deser_repo_metadata(metadata_bytes.as_ref())?; - Ok(app_state - .db - .pending - .get(crate::db::keys::pending_key(metadata.index_id)) - .into_diagnostic()? - .is_some()) - }) - .await - .map_err(|e| MiniDocError::Other(miette::miette!(e)))? - .map_err(MiniDocError::Other)?; - - if is_pending { - return Err(MiniDocError::NotSynced); - } - - let pds = info - .pds - .ok_or_else(|| MiniDocError::CouldNotResolveIdentity)?; - let signing_key = info - .signing_key - .ok_or_else(|| MiniDocError::CouldNotResolveIdentity)? - .into_static(); - - let handle = if let Some(h) = info.handle { - let is_valid = self - .state - .resolver - .verify_handle(&self.did, &h) - .await - .into_diagnostic() - .map_err(MiniDocError::Other)?; - is_valid.then_some(h).unwrap_or_else(invalid_handle) - } else { - invalid_handle() - }; - - Ok(MiniDoc { - did: self.did.clone().into_static(), - handle, - pds, - signing_key, - }) - } - /// gets a record from this repository. pub async fn get_record(&self, collection: &str, rkey: &str) -> Result> { let did = self.did.clone().into_static(); diff --git a/src/control/repos/mod.rs b/src/control/repos/mod.rs new file mode 100644 index 0000000..783c028 --- /dev/null +++ b/src/control/repos/mod.rs @@ -0,0 +1,358 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use chrono::{DateTime, Utc}; +use jacquard_common::cowstr::ToCowStr; +use jacquard_common::types::cid::{Cid, IpldCid}; +use jacquard_common::types::ident::AtIdentifier; +use jacquard_common::types::nsid::Nsid; +use jacquard_common::types::string::{Did, Handle, Rkey}; +use jacquard_common::types::tid::Tid; +use jacquard_common::{CowStr, Data, IntoStatic}; +use miette::{Context, IntoDiagnostic, Result, WrapErr}; +use smol_str::ToSmolStr; +use url::Url; + +use crate::db::types::{DbRkey, DidKey, TrimmedDid}; +use crate::db::{self, Db, keys}; +use crate::state::AppState; +#[cfg(feature = "indexer")] +use crate::types::GaugeState; +use crate::types::{RepoMetadata, RepoState, RepoStatus}; +use crate::util::invalid_handle; + +#[cfg(feature = "indexer")] +mod indexer; + +#[cfg(feature = "indexer")] +pub use indexer::*; + +/// information about a tracked or known repository. returned by [`ReposControl`] methods. +#[derive(Debug, Clone, serde::Serialize)] +pub struct RepoInfo { + /// the DID of the repository. + pub did: Did<'static>, + /// the status of the repository. + #[serde(serialize_with = "crate::util::repo_status_serialize_str")] + pub status: RepoStatus, + /// whether this repository is tracked or not. + /// untracked repositories are not updated and they stay frozen. + pub tracked: bool, + /// the revision of the root commit of this repository. + #[serde(skip_serializing_if = "Option::is_none")] + pub rev: Option, + /// the CID of the MST root of this repository. + #[serde(serialize_with = "crate::util::opt_cid_serialize_str")] + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + /// the handle for the DID of this repository. + /// + /// note that this handle is not bi-directionally verified. + #[serde(skip_serializing_if = "Option::is_none")] + pub handle: Option>, + /// the URL for the PDS in which this repository is hosted on. + #[serde(skip_serializing_if = "Option::is_none")] + pub pds: Option, + /// ATProto signing key of this repository. + #[serde(serialize_with = "crate::util::opt_did_key_serialize_str")] + #[serde(skip_serializing_if = "Option::is_none")] + pub signing_key: Option>, + /// when this repository was last touched (status update, commit ingested, etc.). + #[serde(skip_serializing_if = "Option::is_none")] + pub last_updated_at: Option>, + /// the time of the last message gotten from the firehose for this repository. + /// this is equal to the `time` field. + #[serde(skip_serializing_if = "Option::is_none")] + pub last_message_at: Option>, +} + +/// control over which repositories are tracked and access to their state. +/// +/// in `filter` mode, a repo is only indexed if it either matches a signal or is +/// explicitly tracked via [`ReposControl::track`]. in `full` mode all repos are +/// indexed and tracking is implicit. +/// +/// tracking a DID that hydrant has never seen enqueues an immediate backfill. +/// tracking a DID that hydrant already knows about (but has marked untracked) +/// re-enqueues it for backfill. +#[derive(Clone)] +pub struct ReposControl(pub(super) Arc); + +impl ReposControl { + pub(crate) fn iter_states( + &self, + cursor: Option<&Did<'_>>, + ) -> impl Iterator, RepoState<'static>, crate::types::RepoMetadata)>> + { + let start_bound = if let Some(cursor) = cursor { + let did_key = keys::repo_key(cursor); + std::ops::Bound::Excluded(did_key) + } else { + std::ops::Bound::Unbounded + }; + + let state = self.0.clone(); + self.0 + .db + .repos + .range((start_bound, std::ops::Bound::Unbounded)) + .map(move |g| { + let (k, v) = g.into_inner().into_diagnostic()?; + let repo_state = crate::db::deser_repo_state(&v)?.into_static(); + let did = TrimmedDid::try_from(k.as_ref())?.to_did(); + let metadata_key = keys::repo_metadata_key(&did); + let metadata = state + .db + .repo_metadata + .get(&metadata_key) + .into_diagnostic()? + .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; + let metadata = crate::db::deser_repo_meta(metadata.as_ref())?; + Ok((did, repo_state, metadata)) + }) + } + + /// iterates through all repositories, returning their state. + pub fn iter(&self, cursor: Option<&Did<'_>>) -> impl Iterator> { + self.iter_states(cursor) + .map(|r| r.map(|(did, s, m)| repo_state_to_info(did, s, m.tracked))) + } + + /// gets a handle for a repository to read from it. + pub fn get<'i>(&self, did: &Did<'i>) -> RepoHandle<'i> { + RepoHandle { + state: self.0.clone(), + did: did.clone(), + } + } + + /// same as [`ReposControl::get`] but allows you to pass in an identifier that can be + /// either a handle or a DID. + pub async fn resolve(&self, repo: &AtIdentifier<'_>) -> Result> { + let did = self.0.resolver.resolve_did(repo).await?; + Ok(RepoHandle { + state: self.0.clone(), + did, + }) + } + + /// fetch the current state of a repository. + /// returns `None` if hydrant has never seen this repository. + pub async fn info(&self, did: &Did<'_>) -> Result> { + self.get(did).info().await + } +} + +pub(crate) fn repo_state_to_info(did: Did<'static>, s: RepoState<'_>, tracked: bool) -> RepoInfo { + let (rev, data) = s + .root + .map(|c| (Some(c.rev.to_tid()), Some(c.data))) + .unwrap_or_default(); + RepoInfo { + did, + status: s.status, + tracked, + rev, + data, + handle: s.handle.map(|h| h.into_static()), + pds: s.pds.and_then(|p| p.parse().ok()), + signing_key: s.signing_key.map(|k| k.into_static()), + last_updated_at: DateTime::from_timestamp_secs(s.last_updated_at), + last_message_at: s.last_message_time.and_then(DateTime::from_timestamp_secs), + } +} + +pub struct Record { + pub did: Did<'static>, + pub cid: Cid<'static>, + pub value: Data<'static>, +} + +pub struct ListedRecord { + pub rkey: Rkey<'static>, + pub cid: Cid<'static>, + pub value: Data<'static>, +} + +pub struct RecordList { + pub records: Vec, + pub cursor: Option>, +} + +#[derive(Debug, thiserror::Error)] +pub enum MiniDocError { + #[error("repo is not synced yet")] + NotSynced, + #[error("repo not found")] + RepoNotFound, + #[error("could not resolve identity")] + CouldNotResolveIdentity, + #[error("{0}")] + Other(miette::Error), +} + +/// a mini doc with a bi-directionally verified handle. +pub struct MiniDoc<'i> { + /// the did. + pub did: Did<'i>, + /// the handle. if verification fails or no handle is found, + /// this will be "handle.invalid". + pub handle: Handle<'i>, + /// the url of the PDS of this repo. + pub pds: Url, + /// the atproto signing key of this repo. + pub signing_key: DidKey<'i>, +} + +/// handle to access data related to this repository. +#[derive(Clone)] +pub struct RepoHandle<'i> { + state: Arc, + pub did: Did<'i>, +} + +impl<'i> RepoHandle<'i> { + pub(crate) async fn state(&self) -> Result>> { + let did_key = keys::repo_key(&self.did); + let app_state = self.state.clone(); + + tokio::task::spawn_blocking(move || { + let bytes = app_state.db.repos.get(&did_key).into_diagnostic()?; + bytes + .as_deref() + .map(db::deser_repo_state) + .transpose() + .map(|opt| opt.map(IntoStatic::into_static)) + }) + .await + .into_diagnostic()? + } + + /// fetch the current state of this repository. + /// returns `None` if hydrant has never seen this repository. + pub async fn info(&self) -> Result> { + let did = self.did.clone().into_static(); + let did_key = keys::repo_key(&did); + let metadata_key = keys::repo_metadata_key(&did); + let app_state = self.state.clone(); + + tokio::task::spawn_blocking(move || { + let state_bytes = app_state.db.repos.get(&did_key).into_diagnostic()?; + let Some(state_bytes) = state_bytes else { + return Ok(None); + }; + let repo_state = crate::db::deser_repo_state(&state_bytes)?; + + let metadata_bytes = app_state + .db + .repo_metadata + .get(&metadata_key) + .into_diagnostic()? + .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; + let metadata = crate::db::deser_repo_meta(&metadata_bytes)?; + + Ok(Some(repo_state_to_info(did, repo_state, metadata.tracked))) + }) + .await + .into_diagnostic()? + } + + /// returns the collections of this repository and the number of records it has in each. + pub async fn collections(&self) -> Result, u64>> { + let did = self.did.clone().into_static(); + let state = self.state.clone(); + + tokio::task::spawn_blocking(move || { + let prefix = keys::did_collection_prefix(&did); + let mut res = HashMap::new(); + for item in state.db.counts.prefix(&prefix) { + let (k, v) = item.into_inner().into_diagnostic()?; + let col = k + .strip_prefix(prefix.as_slice()) + .ok_or_else(|| miette::miette!("invalid collection count key: {k:?}")) + .and_then(|r| std::str::from_utf8(r).into_diagnostic()) + .and_then(|n| Nsid::new(n).into_diagnostic())? + .into_static(); + let count = u64::from_be_bytes( + v.as_ref() + .try_into() + .into_diagnostic() + .wrap_err("expected to be count (8 bytes)")?, + ); + res.insert(col, count); + } + Ok(res) + }) + .await + .into_diagnostic()? + } + + /// returns a bi-directionally validated mini doc. + pub async fn mini_doc(&self) -> Result, MiniDocError> { + let Some(info) = self.info().await.map_err(MiniDocError::Other)? else { + return Err(MiniDocError::RepoNotFound); + }; + + // check if repo is still backfilling (in pending) + #[cfg(feature = "indexer")] + let is_pending = { + let metadata_key = keys::repo_metadata_key(&self.did); + let app_state = self.state.clone(); + tokio::task::spawn_blocking(move || { + let metadata_bytes = app_state + .db + .repo_metadata + .get(&metadata_key) + .into_diagnostic()?; + let Some(metadata_bytes) = metadata_bytes else { + return Ok::<_, miette::Report>(false); + }; + + let metadata = crate::db::deser_repo_meta(metadata_bytes.as_ref())?; + return Ok(app_state + .db + .pending + .get(crate::db::keys::pending_key(metadata.index_id)) + .into_diagnostic()? + .is_some()); + }) + .await + .map_err(|e| MiniDocError::Other(miette::miette!(e)))? + .map_err(MiniDocError::Other)? + }; + #[cfg(feature = "relay")] + let is_pending = false; + + if is_pending { + return Err(MiniDocError::NotSynced); + } + + let pds = info + .pds + .ok_or_else(|| MiniDocError::CouldNotResolveIdentity)?; + let signing_key = info + .signing_key + .ok_or_else(|| MiniDocError::CouldNotResolveIdentity)? + .into_static(); + + let handle = if let Some(h) = info.handle { + let is_valid = self + .state + .resolver + .verify_handle(&self.did, &h) + .await + .into_diagnostic() + .map_err(MiniDocError::Other)?; + is_valid.then_some(h).unwrap_or_else(invalid_handle) + } else { + invalid_handle() + }; + + Ok(MiniDoc { + did: self.did.clone().into_static(), + handle, + pds, + signing_key, + }) + } +} diff --git a/src/control/seed.rs b/src/control/seed.rs index cde99ca..e30530c 100644 --- a/src/control/seed.rs +++ b/src/control/seed.rs @@ -59,7 +59,7 @@ async fn seed_one( let ks = state.db.cursors.clone(); let key = cursor_key.clone(); match db::Db::get(ks, key).await { - Ok(Some(b)) => rmp_serde::from_slice::(&b).ok(), + Ok(Some(b)) => rmp_serde::from_slice::(b.as_ref()).ok(), Ok(None) => None, Err(e) => { warn!(err = %e, "failed to load seed cursor, starting from scratch"); diff --git a/src/crawler/worker.rs b/src/crawler/worker.rs index bf81ba3..2267116 100644 --- a/src/crawler/worker.rs +++ b/src/crawler/worker.rs @@ -161,8 +161,9 @@ impl CrawlerWorker { batch.insert( &app_state.db.repo_metadata, &metadata_key, - crate::db::ser_repo_metadata(&metadata)?, + crate::db::ser_repo_meta(&metadata)?, ); + #[cfg(feature = "indexer")] batch.insert( &app_state.db.pending, keys::pending_key(metadata.index_id), @@ -204,6 +205,7 @@ impl CrawlerWorker { .db .update_count_async("pending", count as i64) .await; + #[cfg(feature = "indexer")] self.state.notify_backfill(); } diff --git a/src/db/indexer.rs b/src/db/indexer.rs new file mode 100644 index 0000000..1da986e --- /dev/null +++ b/src/db/indexer.rs @@ -0,0 +1,166 @@ +use crate::types::{GaugeState, RepoStatus, ResyncState}; +use fjall::{Keyspace, OwnedWriteBatch}; +use jacquard_common::IntoStatic; +use jacquard_common::types::string::Did; +use miette::{IntoDiagnostic, Result, WrapErr}; +use url::Url; + +use crate::db::{Db, deser_repo_state, keys, ser_repo_state}; +use crate::types::RepoState; + +impl Db { + pub(crate) fn update_gauge_diff(&self, old: &GaugeState, new: &GaugeState) { + update_gauge_diff_impl!(self, old, new, update_count); + } + + pub(crate) async fn update_gauge_diff_async(&self, old: &GaugeState, new: &GaugeState) { + update_gauge_diff_impl!(self, old, new, update_count_async, await); + } + + pub(crate) fn update_repo_state( + batch: &mut OwnedWriteBatch, + repos: &Keyspace, + did: &Did<'_>, + f: F, + ) -> Result, T)>> + where + F: FnOnce(&mut RepoState, (&[u8], &mut fjall::OwnedWriteBatch)) -> Result<(bool, T)>, + { + let key = keys::repo_key(did); + if let Some(bytes) = repos.get(&key).into_diagnostic()? { + let mut state: RepoState = deser_repo_state(bytes.as_ref())?.into_static(); + let (changed, result) = f(&mut state, (key.as_slice(), batch))?; + if changed { + batch.insert(repos, key, ser_repo_state(&state)?); + } + Ok(Some((state, result))) + } else { + Ok(None) + } + } + + pub(crate) async fn update_repo_state_async( + &self, + did: &Did<'_>, + f: F, + ) -> Result, T)>> + where + F: FnOnce(&mut RepoState, (&[u8], &mut fjall::OwnedWriteBatch)) -> Result<(bool, T)> + + Send + + 'static, + T: Send + 'static, + { + let mut batch = self.inner.batch(); + let repos = self.repos.clone(); + let did = did.clone().into_static(); + + tokio::task::spawn_blocking(move || { + let Some((state, t)) = Self::update_repo_state(&mut batch, &repos, &did, f)? else { + return Ok(None); + }; + batch.commit().into_diagnostic()?; + Ok(Some((state, t))) + }) + .await + .into_diagnostic()? + } + + pub(crate) fn repo_gauge_state( + repo_state: &RepoState, + resync_bytes: Option<&[u8]>, + ) -> GaugeState { + match repo_state.status { + RepoStatus::Synced => GaugeState::Synced, + RepoStatus::Error(_) + | RepoStatus::Deactivated + | RepoStatus::Takendown + | RepoStatus::Suspended + | RepoStatus::Deleted + | RepoStatus::Desynchronized + | RepoStatus::Throttled => resync_bytes + .and_then(|b| rmp_serde::from_slice::(b).ok()) + .and_then(|s| match s { + ResyncState::Error { kind, .. } => Some(GaugeState::Resync(Some(kind))), + _ => None, + }) + .unwrap_or(GaugeState::Resync(None)), + } + } +} + +pub fn set_record_count( + batch: &mut OwnedWriteBatch, + db: &Db, + did: &Did<'_>, + collection: &str, + count: u64, +) { + let key = keys::count_collection_key(did, collection); + batch.insert(&db.counts, key, count.to_be_bytes()); +} + +pub fn update_record_count( + batch: &mut OwnedWriteBatch, + db: &Db, + did: &Did<'_>, + collection: &str, + delta: i64, +) -> Result<()> { + let key = keys::count_collection_key(did, collection); + let count = db + .counts + .get(&key) + .into_diagnostic()? + .map(|v| -> Result<_> { + Ok(u64::from_be_bytes( + v.as_ref() + .try_into() + .into_diagnostic() + .wrap_err("expected to be count (8 bytes)")?, + )) + }) + .transpose()? + .unwrap_or(0); + let new_count = if delta >= 0 { + count.saturating_add(delta as u64) + } else { + count.saturating_sub(delta.unsigned_abs()) + }; + batch.insert(&db.counts, key, new_count.to_be_bytes()); + Ok(()) +} + +pub fn get_record_count(db: &Db, did: &Did<'_>, collection: &str) -> Result { + let key = keys::count_collection_key(did, collection); + let count = db + .counts + .get(&key) + .into_diagnostic()? + .map(|v| -> Result<_> { + Ok(u64::from_be_bytes( + v.as_ref() + .try_into() + .into_diagnostic() + .wrap_err("expected to be count (8 bytes)")?, + )) + }) + .transpose()?; + Ok(count.unwrap_or(0)) +} + +pub fn load_persisted_crawler_sources( + db: &crate::db::Db, +) -> Result> { + use crate::db::keys::CRAWLER_SOURCE_PREFIX; + + let mut sources = Vec::new(); + for entry in db.crawler.prefix(CRAWLER_SOURCE_PREFIX) { + let (key, val) = entry.into_inner().into_diagnostic()?; + let url_bytes = &key[CRAWLER_SOURCE_PREFIX.len()..]; + let url_str = std::str::from_utf8(url_bytes).into_diagnostic()?; + let url = Url::parse(url_str).into_diagnostic()?; + let mode: crate::config::CrawlerMode = rmp_serde::from_slice(&val).into_diagnostic()?; + sources.push(crate::config::CrawlerSource { url, mode }); + } + Ok(sources) +} diff --git a/src/db/keys/indexer.rs b/src/db/keys/indexer.rs new file mode 100644 index 0000000..532b4cb --- /dev/null +++ b/src/db/keys/indexer.rs @@ -0,0 +1,165 @@ +use jacquard_common::types::string::Did; +use smol_str::SmolStr; + +use super::SEP; +use crate::db::types::{DbRkey, DbTid, TrimmedDid}; + +pub const EVENT_WATERMARK_PREFIX: &[u8] = b"ewm|"; + +pub fn pending_key(id: u64) -> [u8; 8] { + id.to_be_bytes() +} + +pub fn event_watermark_key(timestamp_secs: u64) -> Vec { + let mut key = Vec::with_capacity(EVENT_WATERMARK_PREFIX.len() + 8); + key.extend_from_slice(EVENT_WATERMARK_PREFIX); + key.extend_from_slice(×tamp_secs.to_be_bytes()); + key +} + +// prefix format: {DID}| (DID trimmed) +pub fn record_prefix_did(did: &Did) -> Vec { + let repo = TrimmedDid::from(did); + let mut prefix = Vec::with_capacity(repo.len() + 1); + repo.write_to_vec(&mut prefix); + prefix.push(SEP); + prefix +} + +// prefix format: {DID}|{collection}| +pub fn record_prefix_collection(did: &Did, collection: &str) -> Vec { + let repo = TrimmedDid::from(did); + let mut prefix = Vec::with_capacity(repo.len() + 1 + collection.len() + 1); + repo.write_to_vec(&mut prefix); + prefix.push(SEP); + prefix.extend_from_slice(collection.as_bytes()); + prefix.push(SEP); + prefix +} + +// key format: {DID}|{collection}|{rkey} +pub fn record_key(did: &Did, collection: &str, rkey: &DbRkey) -> Vec { + let repo = TrimmedDid::from(did); + let mut key = Vec::with_capacity(repo.len() + 1 + collection.len() + 1 + rkey.len() + 1); + repo.write_to_vec(&mut key); + key.push(SEP); + key.extend_from_slice(collection.as_bytes()); + key.push(SEP); + write_rkey(&mut key, rkey); + key +} + +pub fn write_rkey(buf: &mut Vec, rkey: &DbRkey) { + match rkey { + DbRkey::Tid(tid) => { + buf.push(b't'); + buf.extend_from_slice(tid.as_bytes()); + } + DbRkey::Str(s) => { + buf.push(b's'); + buf.extend_from_slice(s.as_bytes()); + } + } +} + +pub fn parse_rkey(raw: &[u8]) -> miette::Result { + let Some(kind) = raw.first() else { + miette::bail!("record key is empty"); + }; + let rkey = match kind { + b't' => { + DbRkey::Tid(DbTid::new_from_bytes(raw[1..].try_into().map_err(|e| { + miette::miette!("record key '{raw:?}' is invalid: {e}") + })?)) + } + b's' => DbRkey::Str(SmolStr::new( + std::str::from_utf8(&raw[1..]) + .map_err(|e| miette::miette!("record key '{raw:?}' is invalid: {e}"))?, + )), + _ => miette::bail!("invalid record key kind: {}", *kind as char), + }; + Ok(rkey) +} + +// key format: r|{DID}|{collection} (DID trimmed) +pub fn count_collection_key(did: &Did, collection: &str) -> Vec { + let mut key = super::did_collection_prefix(did); + key.extend_from_slice(collection.as_bytes()); + key +} + +// key format: {DID}|{rev} +pub fn resync_buffer_key(did: &Did, rev: DbTid) -> Vec { + let repo = TrimmedDid::from(did); + let mut key = Vec::with_capacity(repo.len() + 1 + 8); + repo.write_to_vec(&mut key); + key.push(SEP); + key.extend_from_slice(&rev.as_bytes()); + key +} + +// prefix format: {DID}| (DID trimmed) +pub fn resync_buffer_prefix(did: &Did) -> Vec { + let repo = TrimmedDid::from(did); + let mut prefix = Vec::with_capacity(repo.len() + 1); + repo.write_to_vec(&mut prefix); + prefix.push(SEP); + prefix +} + +/// key format: `ret|` +pub const CRAWLER_RETRY_PREFIX: &[u8] = b"ret|"; + +pub fn crawler_retry_key(did: &Did) -> Vec { + let repo = TrimmedDid::from(did); + let mut key = Vec::with_capacity(CRAWLER_RETRY_PREFIX.len() + repo.len()); + key.extend_from_slice(CRAWLER_RETRY_PREFIX); + repo.write_to_vec(&mut key); + key +} + +pub fn crawler_retry_parse_key(key: &[u8]) -> miette::Result> { + TrimmedDid::try_from(&key[CRAWLER_RETRY_PREFIX.len()..]) +} + +pub const CRAWLER_CURSOR_PREFIX: &[u8] = b"crawler_cursor|"; + +pub fn crawler_cursor_key(relay: &str) -> Vec { + let mut key = CRAWLER_CURSOR_PREFIX.to_vec(); + key.extend_from_slice(relay.as_bytes()); + key +} + +pub const BY_COLLECTION_CURSOR_PREFIX: &[u8] = b"by_collection_cursor|"; + +/// prefix for all by-collection cursors belonging to a given index URL. +pub fn by_collection_cursor_prefix(url: &str) -> Vec { + let mut prefix = BY_COLLECTION_CURSOR_PREFIX.to_vec(); + prefix.extend_from_slice(url.as_bytes()); + prefix.push(SEP); + prefix +} + +pub fn by_collection_cursor_key(url: &str, collection: &str) -> Vec { + let mut key = by_collection_cursor_prefix(url); + key.extend_from_slice(collection.as_bytes()); + key +} + +pub const CRAWLER_SOURCE_PREFIX: &[u8] = b"src|"; + +pub fn crawler_source_key(url: &str) -> Vec { + let mut key = Vec::with_capacity(CRAWLER_SOURCE_PREFIX.len() + url.len()); + key.extend_from_slice(CRAWLER_SOURCE_PREFIX); + key.extend_from_slice(url.as_bytes()); + key +} + +// key format: {collection}|{cid_bytes} +pub fn block_key(collection: &str, cid: &[u8]) -> Vec { + let mut key = Vec::with_capacity(collection.len() + 1 + cid.len()); + key.extend_from_slice(collection.as_bytes()); + key.push(SEP); + key.extend_from_slice(cid); + key +} diff --git a/src/db/keys/mod.rs b/src/db/keys/mod.rs index 280c5f6..2cae544 100644 --- a/src/db/keys/mod.rs +++ b/src/db/keys/mod.rs @@ -1,7 +1,6 @@ use jacquard_common::types::string::Did; -use smol_str::SmolStr; -use crate::db::types::{DbRkey, DbTid, TrimmedDid}; +use crate::db::types::TrimmedDid; pub mod v1; @@ -11,7 +10,9 @@ pub use v1::{firehose_cursor_key, firehose_cursor_key_from_url}; pub const SEP: u8 = b'|'; #[cfg(feature = "indexer")] -pub const EVENT_WATERMARK_PREFIX: &[u8] = b"ewm|"; +pub mod indexer; +#[cfg(feature = "indexer")] +pub use indexer::*; #[cfg(feature = "relay")] pub const RELAY_EVENT_WATERMARK_PREFIX: &[u8] = b"rwm|"; @@ -35,16 +36,10 @@ pub fn repo_metadata_key<'a>(did: &'a Did) -> Vec { vec } -pub fn pending_key(id: u64) -> [u8; 8] { - id.to_be_bytes() -} - -#[cfg(feature = "indexer")] -pub fn event_watermark_key(timestamp_secs: u64) -> Vec { - let mut key = Vec::with_capacity(EVENT_WATERMARK_PREFIX.len() + 8); - key.extend_from_slice(EVENT_WATERMARK_PREFIX); - key.extend_from_slice(×tamp_secs.to_be_bytes()); - key +#[cfg(feature = "relay")] +/// key format: {SEQ} (u64 big-endian), mirroring event_key +pub fn relay_event_key(seq: u64) -> [u8; 8] { + seq.to_be_bytes() } #[cfg(feature = "relay")] @@ -55,70 +50,6 @@ pub fn relay_event_watermark_key(timestamp_secs: u64) -> Vec { key } -// prefix format: {DID}| (DID trimmed) -pub fn record_prefix_did(did: &Did) -> Vec { - let repo = TrimmedDid::from(did); - let mut prefix = Vec::with_capacity(repo.len() + 1); - repo.write_to_vec(&mut prefix); - prefix.push(SEP); - prefix -} - -// prefix format: {DID}|{collection}| -pub fn record_prefix_collection(did: &Did, collection: &str) -> Vec { - let repo = TrimmedDid::from(did); - let mut prefix = Vec::with_capacity(repo.len() + 1 + collection.len() + 1); - repo.write_to_vec(&mut prefix); - prefix.push(SEP); - prefix.extend_from_slice(collection.as_bytes()); - prefix.push(SEP); - prefix -} - -// key format: {DID}|{collection}|{rkey} -pub fn record_key(did: &Did, collection: &str, rkey: &DbRkey) -> Vec { - let repo = TrimmedDid::from(did); - let mut key = Vec::with_capacity(repo.len() + 1 + collection.len() + 1 + rkey.len() + 1); - repo.write_to_vec(&mut key); - key.push(SEP); - key.extend_from_slice(collection.as_bytes()); - key.push(SEP); - write_rkey(&mut key, rkey); - key -} - -pub fn write_rkey(buf: &mut Vec, rkey: &DbRkey) { - match rkey { - DbRkey::Tid(tid) => { - buf.push(b't'); - buf.extend_from_slice(tid.as_bytes()); - } - DbRkey::Str(s) => { - buf.push(b's'); - buf.extend_from_slice(s.as_bytes()); - } - } -} - -pub fn parse_rkey(raw: &[u8]) -> miette::Result { - let Some(kind) = raw.first() else { - miette::bail!("record key is empty"); - }; - let rkey = match kind { - b't' => { - DbRkey::Tid(DbTid::new_from_bytes(raw[1..].try_into().map_err(|e| { - miette::miette!("record key '{raw:?}' is invalid: {e}") - })?)) - } - b's' => DbRkey::Str(SmolStr::new( - std::str::from_utf8(&raw[1..]) - .map_err(|e| miette::miette!("record key '{raw:?}' is invalid: {e}"))?, - )), - _ => miette::bail!("invalid record key kind: {}", *kind as char), - }; - Ok(rkey) -} - // key format: {SEQ} pub fn event_key(seq: u64) -> [u8; 8] { seq.to_be_bytes() @@ -146,80 +77,6 @@ pub fn did_collection_prefix(did: &Did) -> Vec { key } -// key format: r|{DID}|{collection} (DID trimmed) -pub fn count_collection_key(did: &Did, collection: &str) -> Vec { - let mut key = did_collection_prefix(did); - key.extend_from_slice(collection.as_bytes()); - key -} - -// key format: {DID}|{rev} -pub fn resync_buffer_key(did: &Did, rev: DbTid) -> Vec { - let repo = TrimmedDid::from(did); - let mut key = Vec::with_capacity(repo.len() + 1 + 8); - repo.write_to_vec(&mut key); - key.push(SEP); - key.extend_from_slice(&rev.as_bytes()); - key -} - -// prefix format: {DID}| (DID trimmed) -pub fn resync_buffer_prefix(did: &Did) -> Vec { - let repo = TrimmedDid::from(did); - let mut prefix = Vec::with_capacity(repo.len() + 1); - repo.write_to_vec(&mut prefix); - prefix.push(SEP); - prefix -} - -/// key format: `ret|` -pub const CRAWLER_RETRY_PREFIX: &[u8] = b"ret|"; - -pub fn crawler_retry_key(did: &Did) -> Vec { - let repo = TrimmedDid::from(did); - let mut key = Vec::with_capacity(CRAWLER_RETRY_PREFIX.len() + repo.len()); - key.extend_from_slice(CRAWLER_RETRY_PREFIX); - repo.write_to_vec(&mut key); - key -} - -pub fn crawler_retry_parse_key(key: &[u8]) -> miette::Result> { - TrimmedDid::try_from(&key[CRAWLER_RETRY_PREFIX.len()..]) -} - -pub const CRAWLER_CURSOR_PREFIX: &[u8] = b"crawler_cursor|"; - -pub fn crawler_cursor_key(relay: &str) -> Vec { - let mut key = CRAWLER_CURSOR_PREFIX.to_vec(); - key.extend_from_slice(relay.as_bytes()); - key -} - -pub const BY_COLLECTION_CURSOR_PREFIX: &[u8] = b"by_collection_cursor|"; - -/// prefix for all by-collection cursors belonging to a given index URL. -pub fn by_collection_cursor_prefix(url: &str) -> Vec { - let mut prefix = BY_COLLECTION_CURSOR_PREFIX.to_vec(); - prefix.extend_from_slice(url.as_bytes()); - prefix.push(SEP); - prefix -} - -pub fn by_collection_cursor_key(url: &str, collection: &str) -> Vec { - let mut key = by_collection_cursor_prefix(url); - key.extend_from_slice(collection.as_bytes()); - key -} - -pub const CRAWLER_SOURCE_PREFIX: &[u8] = b"src|"; - -pub fn crawler_source_key(url: &str) -> Vec { - let mut key = Vec::with_capacity(CRAWLER_SOURCE_PREFIX.len() + url.len()); - key.extend_from_slice(CRAWLER_SOURCE_PREFIX); - key.extend_from_slice(url.as_bytes()); - key -} - pub const SEED_CURSOR_PREFIX: &[u8] = b"seed_cursor|"; pub fn seed_cursor_key(url: &str) -> Vec { @@ -242,18 +99,3 @@ pub fn firehose_source_key(url: &str) -> Vec { pub fn pds_account_count_key(host: &str) -> String { format!("p|{host}") } - -#[cfg(feature = "relay")] -/// key format: {SEQ} (u64 big-endian), mirroring event_key -pub fn relay_event_key(seq: u64) -> [u8; 8] { - seq.to_be_bytes() -} - -// key format: {collection}|{cid_bytes} -pub fn block_key(collection: &str, cid: &[u8]) -> Vec { - let mut key = Vec::with_capacity(collection.len() + 1 + cid.len()); - key.extend_from_slice(collection.as_bytes()); - key.push(SEP); - key.extend_from_slice(cid); - key -} diff --git a/src/db/mod.rs b/src/db/mod.rs index 6ce3de8..cc68e54 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -10,8 +10,6 @@ use fjall::config::{BlockSizePolicy, CompressionPolicy, RestartIntervalPolicy}; use fjall::{ CompressionType, Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode, Slice, }; -use jacquard_common::IntoStatic; -use jacquard_common::types::string::Did; use lsm_tree::compaction::Factory; use miette::{Context, IntoDiagnostic, Result}; use scc::HashMap; @@ -43,17 +41,23 @@ pub struct Db { pub inner: Arc, pub path: std::path::PathBuf, pub repos: Keyspace, + pub repo_metadata: Keyspace, + pub cursors: Keyspace, + pub counts: Keyspace, + pub filter: Keyspace, + pub crawler: Keyspace, + #[cfg(feature = "indexer")] pub records: Keyspace, + #[cfg(feature = "indexer")] pub blocks: Keyspace, - pub cursors: Keyspace, + #[cfg(feature = "indexer")] pub pending: Keyspace, + #[cfg(feature = "indexer")] pub resync: Keyspace, + #[cfg(feature = "indexer")] pub resync_buffer: Keyspace, - pub repo_metadata: Keyspace, + #[cfg(feature = "indexer")] pub events: Keyspace, - pub counts: Keyspace, - pub filter: Keyspace, - pub crawler: Keyspace, #[cfg(feature = "backlinks")] pub backlinks: Keyspace, #[cfg(feature = "indexer")] @@ -69,6 +73,7 @@ pub struct Db { pub counts_map: HashMap, } +#[cfg(feature = "indexer")] macro_rules! update_gauge_diff_impl { ($self:ident, $old:ident, $new:ident, $update_method:ident $(, $await:tt)?) => {{ use crate::types::GaugeState; @@ -226,6 +231,7 @@ impl Db { // did plc are random so the interval wont rlly matter .data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4])), )?; + #[cfg(feature = "indexer")] let pending = open_ks( "pending", opts() @@ -239,6 +245,7 @@ impl Db { // ids are sequential and share prefix so we can use large interval to save space .data_block_restart_interval_policy(RestartIntervalPolicy::all(64)), )?; + #[cfg(feature = "indexer")] let resync = open_ks( "resync", opts() @@ -253,6 +260,7 @@ impl Db { .data_block_restart_interval_policy(RestartIntervalPolicy::all(4)), )?; // this is used in non-ephemeral mode + #[cfg(feature = "indexer")] let blocks = open_ks( "blocks", opts() @@ -276,6 +284,7 @@ impl Db { ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])), )?; + #[cfg(feature = "indexer")] let records = open_ks( "records", opts() @@ -302,6 +311,7 @@ impl Db { .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(1)), )?; + #[cfg(feature = "indexer")] let resync_buffer = open_ks( "resync_buffer", opts() @@ -313,6 +323,7 @@ impl Db { .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(16)), )?; + #[cfg(feature = "indexer")] let events = open_ks( "events", opts() @@ -432,13 +443,19 @@ impl Db { inner: db, path: cfg.database_path.clone(), repos, + repo_metadata, + #[cfg(feature = "indexer")] records, + #[cfg(feature = "indexer")] blocks, cursors, + #[cfg(feature = "indexer")] pending, + #[cfg(feature = "indexer")] resync, + #[cfg(feature = "indexer")] resync_buffer, - repo_metadata, + #[cfg(feature = "indexer")] events, counts, filter, @@ -510,7 +527,9 @@ impl Db { pub fn train_dict(&self, ks_name: &str) -> Result<()> { let ks = match ks_name { + #[cfg(feature = "indexer")] "blocks" => &self.blocks, + #[cfg(feature = "indexer")] "events" => &self.events, "repos" => &self.repos, #[cfg(feature = "backlinks")] @@ -527,27 +546,33 @@ impl Db { }; let samples: Vec> = if ks_name == "blocks" { - // sample up to 200 data blocks per collection, discovered lazily in the predicate - let per_collection_limit = 200usize; - let collection_counts: RefCell, usize>> = - RefCell::new(std::collections::HashMap::new()); - - let new = ks - .sample_data_blocks(5000, |first, _last| { - let Some(sep_idx) = first.iter().position(|&b| b == keys::SEP) else { - return false; - }; - let mut counts = collection_counts.borrow_mut(); - let count = counts.entry(first[..sep_idx].to_vec()).or_insert(0); - if *count >= per_collection_limit { - return false; - } - *count += 1; - true - }) - .into_diagnostic()?; + #[cfg(not(feature = "indexer"))] + miette::bail!("indexer feature required for blocks keyspace training"); - new.into_iter().map(|s| s.to_vec()).collect() + #[cfg(feature = "indexer")] + { + // sample up to 200 data blocks per collection, discovered lazily in the predicate + let per_collection_limit = 200usize; + let collection_counts: RefCell, usize>> = + RefCell::new(std::collections::HashMap::new()); + + let new = ks + .sample_data_blocks(5000, |first, _last| { + let Some(sep_idx) = first.iter().position(|&b| b == keys::SEP) else { + return false; + }; + let mut counts = collection_counts.borrow_mut(); + let count = counts.entry(first[..sep_idx].to_vec()).or_insert(0); + if *count >= per_collection_limit { + return false; + } + *count += 1; + true + }) + .into_diagnostic()?; + + new.into_iter().map(|s| s.to_vec()).collect() + } } else { let mut seen_keys = HashSet::new(); let captured_keys = RefCell::new(Vec::new()); @@ -605,24 +630,34 @@ impl Db { .await .into_diagnostic()? }; - tokio::try_join!( + + let mut tasks = vec![ compact(self.repos.clone()), - compact(self.records.clone()), - compact(self.blocks.clone()), compact(self.cursors.clone()), - compact(self.pending.clone()), - compact(self.resync.clone()), - compact(self.resync_buffer.clone()), compact(self.repo_metadata.clone()), - compact(self.events.clone()), compact(self.counts.clone()), compact(self.filter.clone()), compact(self.crawler.clone()), - )?; + ]; + + #[cfg(feature = "indexer")] + { + tasks.push(compact(self.records.clone())); + tasks.push(compact(self.blocks.clone())); + tasks.push(compact(self.pending.clone())); + tasks.push(compact(self.resync.clone())); + tasks.push(compact(self.resync_buffer.clone())); + tasks.push(compact(self.events.clone())); + } + #[cfg(feature = "relay")] - compact(self.relay_events.clone()).await?; + tasks.push(compact(self.relay_events.clone())); + #[cfg(feature = "backlinks")] - compact(self.backlinks.clone()).await?; + tasks.push(compact(self.backlinks.clone())); + + futures::future::try_join_all(tasks).await?; + Ok(()) } @@ -654,6 +689,7 @@ impl Db { .into_diagnostic()? } + #[allow(dead_code)] pub async fn contains_key(ks: Keyspace, key: impl Into) -> Result { let key = key.into(); tokio::task::spawn_blocking(move || ks.contains_key(key).into_diagnostic()) @@ -711,98 +747,18 @@ impl Db { .await .unwrap_or(0) } +} - pub(crate) fn update_gauge_diff( - &self, - old: &crate::types::GaugeState, - new: &crate::types::GaugeState, - ) { - update_gauge_diff_impl!(self, old, new, update_count); - } - - pub(crate) async fn update_gauge_diff_async( - &self, - old: &crate::types::GaugeState, - new: &crate::types::GaugeState, - ) { - update_gauge_diff_impl!(self, old, new, update_count_async, await); - } - - pub(crate) fn update_repo_state( - batch: &mut OwnedWriteBatch, - repos: &Keyspace, - did: &Did<'_>, - f: F, - ) -> Result, T)>> - where - F: FnOnce(&mut RepoState, (&[u8], &mut fjall::OwnedWriteBatch)) -> Result<(bool, T)>, - { - let key = keys::repo_key(did); - if let Some(bytes) = repos.get(&key).into_diagnostic()? { - let mut state: RepoState = deser_repo_state(bytes.as_ref())?.into_static(); - let (changed, result) = f(&mut state, (key.as_slice(), batch))?; - if changed { - batch.insert(repos, key, ser_repo_state(&state)?); - } - Ok(Some((state, result))) - } else { - Ok(None) - } - } +#[cfg(feature = "indexer")] +mod indexer; - pub(crate) async fn update_repo_state_async( - &self, - did: &Did<'_>, - f: F, - ) -> Result, T)>> - where - F: FnOnce(&mut RepoState, (&[u8], &mut fjall::OwnedWriteBatch)) -> Result<(bool, T)> - + Send - + 'static, - T: Send + 'static, - { - let mut batch = self.inner.batch(); - let repos = self.repos.clone(); - let did = did.clone().into_static(); - - tokio::task::spawn_blocking(move || { - let Some((state, t)) = Self::update_repo_state(&mut batch, &repos, &did, f)? else { - return Ok(None); - }; - batch.commit().into_diagnostic()?; - Ok(Some((state, t))) - }) - .await - .into_diagnostic()? - } +#[cfg(feature = "indexer")] +pub use indexer::*; - pub(crate) fn repo_gauge_state( - repo_state: &RepoState, - resync_bytes: Option<&[u8]>, - ) -> crate::types::GaugeState { - match repo_state.status { - crate::types::RepoStatus::Synced => crate::types::GaugeState::Synced, - crate::types::RepoStatus::Error(_) - | crate::types::RepoStatus::Deactivated - | crate::types::RepoStatus::Takendown - | crate::types::RepoStatus::Suspended - | crate::types::RepoStatus::Deleted - | crate::types::RepoStatus::Desynchronized - | crate::types::RepoStatus::Throttled => { - if let Some(resync_bytes) = resync_bytes { - if let Ok(crate::types::ResyncState::Error { kind, .. }) = - rmp_serde::from_slice::(resync_bytes) - { - crate::types::GaugeState::Resync(Some(kind)) - } else { - crate::types::GaugeState::Resync(None) - } - } else { - crate::types::GaugeState::Resync(None) - } - } - } - } +#[derive(serde::Serialize, serde::Deserialize, Default)] +pub(crate) struct FirehoseSourceMeta { + #[serde(default)] + pub(crate) is_pds: bool, } pub fn set_firehose_cursor(db: &Db, relay: &Url, cursor: i64) -> Result<()> { @@ -818,7 +774,7 @@ pub async fn get_firehose_cursor(db: &Db, relay: &Url) -> Result> { let key = keys::firehose_cursor_key_from_url(relay); Db::get(db.cursors.clone(), key) .await? - .map(|v| { + .map(|v: Slice| { Ok(i64::from_be_bytes( v.as_ref() .try_into() @@ -829,11 +785,11 @@ pub async fn get_firehose_cursor(db: &Db, relay: &Url) -> Result> { .transpose() } -pub fn ser_repo_metadata(state: &RepoMetadata) -> Result> { +pub fn ser_repo_meta(state: &RepoMetadata) -> Result> { rmp_serde::to_vec(&state).into_diagnostic() } -pub fn deser_repo_metadata(bytes: &[u8]) -> Result { +pub fn deser_repo_meta(bytes: &[u8]) -> Result { rmp_serde::from_slice(bytes).into_diagnostic() } @@ -873,72 +829,6 @@ pub fn persist_counts(db: &Db) -> Result<()> { batch.commit().into_diagnostic() } -pub fn set_record_count( - batch: &mut OwnedWriteBatch, - db: &Db, - did: &Did<'_>, - collection: &str, - count: u64, -) { - let key = keys::count_collection_key(did, collection); - batch.insert(&db.counts, key, count.to_be_bytes()); -} - -pub fn update_record_count( - batch: &mut OwnedWriteBatch, - db: &Db, - did: &Did<'_>, - collection: &str, - delta: i64, -) -> Result<()> { - let key = keys::count_collection_key(did, collection); - let count = db - .counts - .get(&key) - .into_diagnostic()? - .map(|v| -> Result<_> { - Ok(u64::from_be_bytes( - v.as_ref() - .try_into() - .into_diagnostic() - .wrap_err("expected to be count (8 bytes)")?, - )) - }) - .transpose()? - .unwrap_or(0); - let new_count = if delta >= 0 { - count.saturating_add(delta as u64) - } else { - count.saturating_sub(delta.unsigned_abs()) - }; - batch.insert(&db.counts, key, new_count.to_be_bytes()); - Ok(()) -} - -pub fn get_record_count(db: &Db, did: &Did<'_>, collection: &str) -> Result { - let key = keys::count_collection_key(did, collection); - let count = db - .counts - .get(&key) - .into_diagnostic()? - .map(|v| -> Result<_> { - Ok(u64::from_be_bytes( - v.as_ref() - .try_into() - .into_diagnostic() - .wrap_err("expected to be count (8 bytes)")?, - )) - }) - .transpose()?; - Ok(count.unwrap_or(0)) -} - -#[derive(serde::Serialize, serde::Deserialize, Default)] -pub(crate) struct FirehoseSourceMeta { - #[serde(default)] - pub(crate) is_pds: bool, -} - pub fn load_persisted_firehose_sources( db: &crate::db::Db, ) -> Result> { @@ -959,20 +849,3 @@ pub fn load_persisted_firehose_sources( } Ok(sources) } - -pub fn load_persisted_crawler_sources( - db: &crate::db::Db, -) -> Result> { - use crate::db::keys::CRAWLER_SOURCE_PREFIX; - - let mut sources = Vec::new(); - for entry in db.crawler.prefix(CRAWLER_SOURCE_PREFIX) { - let (key, val) = entry.into_inner().into_diagnostic()?; - let url_bytes = &key[CRAWLER_SOURCE_PREFIX.len()..]; - let url_str = std::str::from_utf8(url_bytes).into_diagnostic()?; - let url = Url::parse(url_str).into_diagnostic()?; - let mode: crate::config::CrawlerMode = rmp_serde::from_slice(&val).into_diagnostic()?; - sources.push(crate::config::CrawlerSource { url, mode }); - } - Ok(sources) -} diff --git a/src/filter.rs b/src/filter.rs index a987c1e..f8f3961 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -30,26 +30,33 @@ impl FilterConfig { collections: Vec::new(), } } +} + +#[cfg(feature = "indexer")] +mod indexer { + use super::*; - pub fn matches_collection(&self, collection: &str) -> bool { - if self.collections.is_empty() { - return true; + impl FilterConfig { + pub fn matches_collection(&self, collection: &str) -> bool { + if self.collections.is_empty() { + return true; + } + self.collections.iter().any(|p| nsid_matches(p, collection)) } - self.collections.iter().any(|p| nsid_matches(p, collection)) - } - pub fn matches_signal(&self, collection: &str) -> bool { - self.signals.iter().any(|p| nsid_matches(p, collection)) - } + pub fn matches_signal(&self, collection: &str) -> bool { + self.signals.iter().any(|p| nsid_matches(p, collection)) + } - pub fn check_signals(&self) -> bool { - self.mode == FilterMode::Filter && !self.signals.is_empty() + pub fn check_signals(&self) -> bool { + self.mode == FilterMode::Filter && !self.signals.is_empty() + } } -} -fn nsid_matches(pattern: &str, col: &str) -> bool { - pattern - .strip_suffix(".*") - .map(|prefix| col == prefix || col.starts_with(prefix)) - .unwrap_or_else(|| col == pattern) + fn nsid_matches(pattern: &str, col: &str) -> bool { + pattern + .strip_suffix(".*") + .map(|prefix| col == prefix || col.starts_with(prefix)) + .unwrap_or_else(|| col == pattern) + } } diff --git a/src/ingest/firehose.rs b/src/ingest/firehose.rs index 77183d6..47e8bc1 100644 --- a/src/ingest/firehose.rs +++ b/src/ingest/firehose.rs @@ -288,7 +288,7 @@ impl FirehoseIngestor { .get(&metadata_key) .into_diagnostic()? { - let metadata = crate::db::deser_repo_metadata(bytes.as_ref())?; + let metadata = crate::db::deser_repo_meta(bytes.as_ref())?; if metadata.tracked { trace!(did = %did, "tracked repo, processing"); diff --git a/src/ingest/indexer.rs b/src/ingest/indexer.rs index fb4388c..91db48c 100644 --- a/src/ingest/indexer.rs +++ b/src/ingest/indexer.rs @@ -1,5 +1,5 @@ use super::*; -use crate::db::{self, keys, ser_repo_metadata}; +use crate::db::{self, keys, ser_repo_meta}; use crate::ingest::stream::{Account, Commit, Identity}; use crate::ingest::validation; use crate::resolver::{NoSigningKeyError, ResolverError}; @@ -434,7 +434,7 @@ impl FirehoseWorker { let metadata_key = keys::repo_metadata_key(did); let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; let is_backfilling = if let Some(metadata_bytes) = metadata_bytes { - let metadata = crate::db::deser_repo_metadata(metadata_bytes.as_ref())?; + let metadata = crate::db::deser_repo_meta(metadata_bytes.as_ref())?; db.pending .get(keys::pending_key(metadata.index_id)) .into_diagnostic()? @@ -616,7 +616,7 @@ impl FirehoseWorker { .repo_metadata .get(&meta_key) .into_diagnostic()? - .map(|b| crate::db::deser_repo_metadata(&b)) + .map(|b| crate::db::deser_repo_meta(&b)) .transpose()?; let had_metadata = existing_metadata.is_some(); let mut metadata = existing_metadata.unwrap_or_else(|| RepoMetadata { @@ -634,7 +634,7 @@ impl FirehoseWorker { metadata.index_id = rand::random::(); batch.insert(&db.pending, keys::pending_key(metadata.index_id), &repo_key); - batch.insert(&db.repo_metadata, &meta_key, ser_repo_metadata(&metadata)?); + batch.insert(&db.repo_metadata, &meta_key, ser_repo_meta(&metadata)?); batch.commit().into_diagnostic()?; if !was_pending { diff --git a/src/ingest/relay.rs b/src/ingest/relay.rs index 39bf7bf..ffc33cb 100644 --- a/src/ingest/relay.rs +++ b/src/ingest/relay.rs @@ -304,6 +304,9 @@ impl RelayWorker { .. } = validated; + #[cfg(not(feature = "indexer"))] + let _ = parsed_blocks; + if chain_break.is_broken() { // chain breaks are not grounds for blocking when acting as a relay debug!(broken = ?chain_break, "chain break, forwarding anyway"); @@ -782,7 +785,7 @@ impl WorkerContext<'_> { .repo_metadata .get(&metadata_key) .into_diagnostic()? - .map(|bytes| db::deser_repo_metadata(&bytes)) + .map(|bytes| db::deser_repo_meta(&bytes)) .transpose()?; if metadata.map_or(false, |m| !m.tracked) { diff --git a/src/lib.rs b/src/lib.rs index 779eb00..572b3bb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,6 +13,7 @@ pub(crate) mod api; pub(crate) mod backfill; #[cfg(feature = "backlinks")] pub(crate) mod backlinks; +#[cfg(feature = "indexer")] pub(crate) mod crawler; pub(crate) mod db; pub(crate) mod ingest; diff --git a/src/ops.rs b/src/ops.rs index 638980b..c775b7f 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -72,7 +72,7 @@ pub fn delete_repo( let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { - let metadata = db::deser_repo_metadata(&metadata_bytes)?; + let metadata = db::deser_repo_meta(&metadata_bytes)?; batch.remove(&db.pending, keys::pending_key(metadata.index_id)); } @@ -131,7 +131,7 @@ pub fn transition_repo<'batch, 's>( let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { - let metadata = db::deser_repo_metadata(&metadata_bytes)?; + let metadata = db::deser_repo_meta(&metadata_bytes)?; let pending_key = keys::pending_key(metadata.index_id); // manage queues diff --git a/src/state.rs b/src/state.rs index 1fb35ac..f602a32 100644 --- a/src/state.rs +++ b/src/state.rs @@ -6,7 +6,9 @@ use std::time::Duration; use arc_swap::ArcSwap; use miette::Result; use smol_str::SmolStr; -use tokio::sync::{Notify, watch}; +#[cfg(feature = "indexer")] +use tokio::sync::Notify; +use tokio::sync::watch; use url::Url; use crate::{ @@ -27,9 +29,12 @@ pub struct AppState { pub(crate) pds_tiers: PdsTierHandle, pub(crate) rate_tiers: HashMap, pub firehose_cursors: scc::HashIndex, + #[cfg(feature = "indexer")] pub backfill_notify: Notify, + #[cfg(feature = "indexer")] pub crawler_enabled: watch::Sender, pub firehose_enabled: watch::Sender, + #[cfg(feature = "indexer")] pub backfill_enabled: watch::Sender, pub ephemeral_ttl: Duration, pub throttler: Throttler, @@ -41,6 +46,7 @@ impl AppState { let resolver = Resolver::new(config.plc_urls.clone(), config.identity_cache_size); let filter_config = crate::db::filter::load(&db.filter)?; + #[cfg(feature = "indexer")] let crawler_default = match config.enable_crawler { Some(b) => b, // default: enabled if full-network mode, or if crawler sources are configured @@ -69,8 +75,10 @@ impl AppState { let relay_cursors = scc::HashIndex::new(); + #[cfg(feature = "indexer")] let (crawler_enabled, _) = watch::channel(crawler_default); let (firehose_enabled, _) = watch::channel(config.enable_firehose); + #[cfg(feature = "indexer")] let (backfill_enabled, _) = watch::channel(true); Ok(Self { @@ -80,15 +88,19 @@ impl AppState { pds_tiers, rate_tiers: config.rate_tiers.clone(), firehose_cursors: relay_cursors, + #[cfg(feature = "indexer")] backfill_notify: Notify::new(), + #[cfg(feature = "indexer")] crawler_enabled, firehose_enabled, + #[cfg(feature = "indexer")] backfill_enabled, ephemeral_ttl: config.ephemeral_ttl.clone(), throttler: Throttler::new(), }) } + #[cfg(feature = "indexer")] pub fn notify_backfill(&self) { self.backfill_notify.notify_one(); } @@ -115,16 +127,26 @@ impl AppState { F: FnOnce() -> Fut, Fut: Future, { + #[cfg(feature = "indexer")] let crawler_was = *self.crawler_enabled.borrow(); let firehose_was = *self.firehose_enabled.borrow(); + #[cfg(feature = "indexer")] let backfill_was = *self.backfill_enabled.borrow(); + + #[cfg(feature = "indexer")] self.crawler_enabled.send_replace(false); self.firehose_enabled.send_replace(false); + #[cfg(feature = "indexer")] self.backfill_enabled.send_replace(false); + let result = f().await; + + #[cfg(feature = "indexer")] self.crawler_enabled.send_replace(crawler_was); self.firehose_enabled.send_replace(firehose_was); + #[cfg(feature = "indexer")] self.backfill_enabled.send_replace(backfill_was); + result } } diff --git a/src/types.rs b/src/types.rs index 4a1759d..fe003fb 100644 --- a/src/types.rs +++ b/src/types.rs @@ -169,15 +169,61 @@ impl Display for RepoStatus { } } -impl RepoMetadata { - pub fn backfilling(index_id: u64) -> Self { - Self { - index_id, - tracked: true, +impl RepoMetadata {} + +#[cfg(feature = "indexer")] +mod indexer { + use super::*; + + impl RepoMetadata { + pub fn backfilling(index_id: u64) -> Self { + Self { + index_id, + tracked: true, + } + } + } + + impl ResyncState { + pub fn next_backoff(retry_count: u32) -> i64 { + // exponential backoff: 1m, 2m, 4m, 8m... up to 1h + let base = 60; + let cap = 3600; + let mult = 2u64.pow(retry_count.min(10)) as i64; + let delay = (base * mult).min(cap); + + // add +/- 10% jitter + let jitter = (rand::random::() * 0.2 - 0.1) * delay as f64; + let delay = (delay as f64 + jitter) as i64; + + chrono::Utc::now().timestamp() + delay + } + } + + #[derive(Clone, Debug)] + pub(crate) enum BroadcastEvent { + #[allow(dead_code)] + Persisted(u64), + Ephemeral(Box>), + } + + #[derive(Debug, PartialEq, Eq, Clone, Copy)] + pub(crate) enum GaugeState { + Synced, + Pending, + Resync(Option), + } + + impl GaugeState { + pub fn is_resync(&self) -> bool { + matches!(self, GaugeState::Resync(_)) } } } +#[cfg(feature = "indexer")] +pub(crate) use indexer::*; + impl<'i> RepoState<'i> { pub fn backfilling() -> Self { Self { @@ -250,22 +296,6 @@ pub(crate) enum ResyncState { }, } -impl ResyncState { - pub fn next_backoff(retry_count: u32) -> i64 { - // exponential backoff: 1m, 2m, 4m, 8m... up to 1h - let base = 60; - let cap = 3600; - let mult = 2u64.pow(retry_count.min(10)) as i64; - let delay = (base * mult).min(cap); - - // add +/- 10% jitter - let jitter = (rand::random::() * 0.2 - 0.1) * delay as f64; - let delay = (delay as f64 + jitter) as i64; - - chrono::Utc::now().timestamp() + delay - } -} - #[derive(Debug, Serialize, Clone)] pub enum EventType { Record, @@ -304,14 +334,6 @@ pub struct MarshallableEvt<'i> { pub account: Option>, } -#[cfg(feature = "indexer")] -#[derive(Clone, Debug)] -pub(crate) enum BroadcastEvent { - #[allow(dead_code)] - Persisted(u64), - Ephemeral(Box>), -} - #[derive(Debug, Serialize, Clone)] pub struct RecordEvt<'i> { pub live: bool, @@ -392,19 +414,6 @@ pub(crate) struct StoredEvent<'i> { pub data: StoredData, } -#[derive(Debug, PartialEq, Eq, Clone, Copy)] -pub(crate) enum GaugeState { - Synced, - Pending, - Resync(Option), -} - -impl GaugeState { - pub fn is_resync(&self) -> bool { - matches!(self, GaugeState::Resync(_)) - } -} - #[cfg(feature = "relay")] #[derive(Clone)] pub(crate) enum RelayBroadcast { diff --git a/src/util/mod.rs b/src/util/mod.rs index 5ed8756..d70e9f6 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -1,3 +1,5 @@ +#![allow(dead_code)] + use std::{hash::Hash, time::Duration}; use jacquard_common::{deps::fluent_uri, types::string::Handle}; -- 2.51.2