Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416use futures::FutureExt;use miette::{IntoDiagnostic, Result};use std::future::Future;use std::sync::Arc;use std::sync::atomic::{AtomicBool, Ordering};use tokio::sync::watch;use tracing::{debug, error, info, warn};use url::Url;
use crate::config::{Config, SignatureVerification};use crate::db::{self, filter as db_filter};use crate::filter::FilterMode;use crate::state::AppState;
use super::firehose::{FirehoseHandle, FirehoseShared};use super::{DbControl, FilterControl, ReposControl, pds::PdsControl};
#[cfg(feature = "indexer")]use super::{BackfillHandle, crawler};#[cfg(feature = "indexer")]use crate::backfill::BackfillWorker;#[cfg(feature = "indexer")]use crate::db::load_persisted_crawler_sources;#[cfg(feature = "indexer")]use crate::ingest::indexer::FirehoseWorker;
#[cfg(feature = "backlinks")]use crate::backlinks::BacklinksControl;
use super::seed;
pub(crate) mod run;
/// the top-level handle to a hydrant instance.////// `Hydrant` is cheaply cloneable. all sub-handles share the same underlying state./// construct it via [`Hydrant::new`] or [`Hydrant::from_env`], configure the filter/// and repos as needed, then call [`Hydrant::run`] to start all background components.#[derive(Clone)]pub struct Hydrant { #[cfg(feature = "indexer")] pub crawler: crawler::CrawlerHandle, pub firehose: FirehoseHandle, #[cfg(feature = "indexer")] pub backfill: BackfillHandle, pub filter: FilterControl, pub pds: PdsControl, pub repos: ReposControl, pub db: DbControl, #[cfg(feature = "backlinks")] pub backlinks: crate::backlinks::BacklinksControl, pub(crate) state: Arc<AppState>, pub(crate) config: Arc<Config>, pub(crate) started: Arc<AtomicBool>, _priv: (),}
impl Hydrant { /// Returns a reference to the internal DID resolver. pub fn resolver(&self) -> &crate::resolver::Resolver { &self.state.resolver }
#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] pub(crate) fn stream_send_timeout(&self) -> std::time::Duration { self.config.stream_send_timeout }
/// open the database and configure hydrant from `config`. /// /// this sets up the database, applies any filter configuration from `config`, and /// initializes all sub-handles. no background tasks are started yet: call /// [`run`](Self::run) to start all components and drive the instance. pub async fn new(config: Config) -> Result<Self> { Self::new_inner(config, None).await }
#[cfg(all(test, feature = "indexer_stream"))] pub(crate) async fn new_with_resolver( config: Config, resolver: crate::resolver::Resolver, ) -> Result<Self> { Self::new_inner(config, Some(resolver)).await }
async fn new_inner( config: Config, resolver: Option<crate::resolver::Resolver>, ) -> Result<Self> { info!("{config}");
#[cfg(feature = "relay")] if config.only_index_links { miette::bail!("HYDRANT_ONLY_INDEX_LINKS is not supported in relay mode"); }
// 1. open database and construct AppState let mut state = AppState::new(&config)?; if let Some(resolver) = resolver { state.resolver = resolver; }
// 2. apply any filter config from env variables if config.full_network || config.filter_signals.is_some() || config.filter_collections.is_some() || config.filter_excludes.is_some() { let mode = config.full_network.then_some(FilterMode::Full); let signals = config .filter_signals .clone() .map(crate::patch::SetUpdate::Set); let collections = config .filter_collections .clone() .map(crate::patch::SetUpdate::Set); let excludes = config .filter_excludes .clone() .map(crate::patch::SetUpdate::Set);
state .db .run(move |db| { let mut batch = db.inner.batch(); db_filter::apply_patch( &mut batch, &db.filter, mode, signals, collections, excludes, )?; batch.commit().into_diagnostic() }) .await?;
// 3. reload the live filter into the hot-path arc-swap let new_filter = state.db.run(move |db| db_filter::load(&db.filter)).await?; state.filter.store(Arc::new(new_filter)); }
#[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 { firehose: FirehoseHandle::new(state.clone()), filter: FilterControl(state.clone()), pds: PdsControl(state.clone()), repos: ReposControl(state.clone()), db: DbControl(state.clone()), #[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()), }, #[cfg(feature = "indexer")] backfill: BackfillHandle::new(state.clone()), #[cfg(feature = "backlinks")] backlinks: crate::backlinks::BacklinksControl(state.clone()), state, config: Arc::new(config), started: Arc::new(AtomicBool::new(false)), _priv: (), }) }
/// reads config from environment variables and calls [`Hydrant::new`]. pub async fn from_env() -> Result<Self> { Self::new(Config::from_env()?).await }
/// returns a future that runs the HTTP management API server on the given bind addresses. /// /// the server exposes all management endpoints (`/filter`, `/repos`, `/ingestion`, /// `/stream`, `/stats`, `/db/*`, `/xrpc/*`). it runs indefinitely and resolves /// only on error. /// /// intended for `tokio::spawn` or inclusion in a `select!` / task list. the clone /// of `self` is deferred until the future is first polled. /// /// to disable the HTTP API entirely, simply don't call this method. pub fn serve(&self, binds: super::ApiBinds) -> impl Future<Output = Result<()>> { let hydrant = self.clone(); async move { crate::api::serve(hydrant, binds).await } }
/// returns a future that runs the debug HTTP API server on `127.0.0.1:{port}`. /// /// exposes internal inspection endpoints (`/debug/get`, `/debug/iter`, etc.). /// binds only to loopback. pub fn serve_debug(&self, port: u16) -> impl Future<Output = Result<()>> { let state = self.state.clone(); async move { crate::api::serve_debug(state, port).await } }
/// iterate sequentially over the current head body of every record in /// `collection`, ordered by (repo, rkey). /// /// values are raw DAG-CBOR. records are keyed repo-major, so this walks /// the whole records keyspace and filters; superseded bodies live in the /// history keyspace and are not included. #[cfg(feature = "indexer")] pub fn scan_collection_blocks( &self, collection: &str, ) -> impl Iterator<Item = Result<ScannedBlock>> + 'static { let db = self.state.db.indexer.clone(); let want = collection.to_string();
let iter = db.record_range::<&[u8], _>(..); iter.filter_map(move |item| { let (key, value) = match item.into_inner() { Ok(kv) => kv, Err(e) => return Some(Err(miette::miette!("{e}"))), }; let Some(did_len) = crate::db::keys::indexer::trimmed_did_len(&key) else { return None; }; let Ok((col, _)) = crate::db::keys::indexer::split_record_suffix(&key[did_len + 1..]) else { return None; }; if col != want { return None; } // links-only mode stores raw cids, not bodies; there is nothing // to scan (a dag-cbor record is a map and never carries the // cidv1-dagcbor-sha256 prefix) if crate::db::is_cid_record_value(&value) { return None; } // the scanned key keeps the legacy `{collection}|{cid}` shape; the // cid is recomputed from the body let cid = match jacquard_repo::mst::util::compute_cid(value.as_ref()) { Ok(c) => c, Err(e) => return Some(Err(miette::miette!("{e}"))), }; let key = crate::db::keys::indexer::block_key(&want, &cid.to_bytes()); Some(Ok(ScannedBlock { key: fjall::Slice::from(key.as_slice()), value, })) }) }}
#[cfg(feature = "indexer")]/// a raw block entry yielded by [`Hydrant::scan_collection_blocks`].////// key and value are kept as raw slices; use the accessor methods for lazy parsing.#[derive(Clone, Debug)]pub struct ScannedBlock { pub(crate) key: fjall::Slice, pub(crate) value: fjall::Slice,}
#[cfg(feature = "indexer")]impl ScannedBlock { /// the collection name part of the block's key. pub fn collection(&self) -> &str { let sep_idx = self .key .iter() .position(|&b| b == b'|') .expect("malformed block key: missing separator"); std::str::from_utf8(&self.key[..sep_idx]).expect("collection name must be valid UTF-8") }
/// the raw CID bytes after the separator. pub fn cid(&self) -> &[u8] { let sep_idx = self .key .iter() .position(|&b| b == b'|') .expect("malformed block key: missing separator"); &self.key[sep_idx + 1..] }
/// the key's CID, parsed. pub fn parsed_cid(&self) -> Result<cid::Cid, miette::Report> { cid::Cid::read_bytes(self.cid()).map_err(|e| miette::miette!("failed to parse CID: {e}")) }
/// the key's CID as a jacquard CID. pub fn jacquard_cid(&self) -> Result<jacquard_common::types::cid::Cid, miette::Report> { let parsed = cid::Cid::read_bytes(self.cid()) .map_err(|e| miette::miette!("failed to parse CID: {e}"))?; Ok(jacquard_common::types::cid::Cid::ipld(parsed)) }
/// the raw DAG-CBOR block bytes. pub fn value(&self) -> &[u8] { &self.value }}
impl axum::extract::FromRef<Hydrant> for Arc<AppState> { fn from_ref(h: &Hydrant) -> Self { h.state.clone() }}
#[cfg(all(test, feature = "indexer"))]mod tests { use super::*; use crate::config::Config; use crate::db::keys; use jacquard_common::IntoStatic; use tempfile::tempdir;
fn test_config(path: &std::path::Path) -> Config { Config { database_path: path.to_path_buf(), ..Default::default() } }
#[tokio::test] async fn test_scan_collection_blocks() -> Result<()> { use crate::db::types::DbRkey; use jacquard_common::types::string::Did;
let tmp = tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(tmp.path())).await?; let state = hydrant.state.clone();
let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let post_body1: &[u8] = b"post_cbor_value"; let like_body: &[u8] = b"like_cbor_value"; let post_body2: &[u8] = b"post_cbor_value_2"; // the scanned cid is recomputed from the body let cid1 = jacquard_repo::mst::util::compute_cid(post_body1).unwrap(); let cid2 = jacquard_repo::mst::util::compute_cid(like_body).unwrap(); let cid3 = jacquard_repo::mst::util::compute_cid(post_body2).unwrap();
state .db .run(move |db| -> Result<()> { let mut batch = db.inner.batch(); db.indexer.stage_record( &mut batch, keys::record_key(&did, "app.bsky.feed.post", &DbRkey::new("a")), post_body1, ); db.indexer.stage_record( &mut batch, keys::record_key(&did, "app.bsky.feed.like", &DbRkey::new("b")), like_body, ); db.indexer.stage_record( &mut batch, keys::record_key(&did, "app.bsky.feed.post", &DbRkey::new("c")), post_body2, ); batch.commit().into_diagnostic()?; Ok(()) }) .await?;
// scan posts: ordered by (repo, rkey), so rkey "a" then "c" let post_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.post") .collect::<Result<Vec<_>>>()?;
assert_eq!(post_blocks.len(), 2); assert_eq!(post_blocks[0].collection(), "app.bsky.feed.post"); assert_eq!(post_blocks[0].cid(), cid1.to_bytes()); assert_eq!(post_blocks[0].parsed_cid().unwrap(), cid1); assert_eq!( post_blocks[0].jacquard_cid().unwrap(), jacquard_common::types::cid::Cid::ipld(cid1) ); assert_eq!(post_blocks[0].value(), post_body1);
assert_eq!(post_blocks[1].collection(), "app.bsky.feed.post"); assert_eq!(post_blocks[1].cid(), cid3.to_bytes()); assert_eq!(post_blocks[1].parsed_cid().unwrap(), cid3); assert_eq!(post_blocks[1].value(), post_body2);
// scan likes let like_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.like") .collect::<Result<Vec<_>>>()?;
assert_eq!(like_blocks.len(), 1); assert_eq!(like_blocks[0].collection(), "app.bsky.feed.like"); assert_eq!(like_blocks[0].cid(), cid2.to_bytes()); assert_eq!(like_blocks[0].parsed_cid().unwrap(), cid2); assert_eq!(like_blocks[0].value(), like_body);
// scan non-existent let empty_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.repost") .collect::<Result<Vec<_>>>()?; assert!(empty_blocks.is_empty());
Ok(()) }}