Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177use super::*;
#[cfg(feature = "indexer_stream")]/// a stream of [`Event`]s. returned by [`Hydrant::subscribe`].////// implements [`futures::Stream`] and can be used with `StreamExt::next`,/// `while let Some(item) = stream.next().await`, `forward`, etc./// the stream terminates when the underlying channel closes (i.e. hydrant shuts down).pub struct EventStream { receiver: mpsc::Receiver<Result<Event, StreamError>>, wake: tokio::sync::broadcast::Sender<crate::types::BroadcastEvent>,}
#[cfg(feature = "indexer_stream")]#[derive(Debug, Clone, thiserror::Error)]pub enum StreamError { #[error("stream consumer too slow: {reason}")] ConsumerTooSlow { reason: String },}
#[cfg(feature = "indexer_stream")]impl Drop for EventStream { fn drop(&mut self) { // Wake the blocking stream thread so it can observe that its receiver // has gone away instead of retaining the Hydrant database indefinitely. let _ = self.wake.send(crate::types::BroadcastEvent::Persisted(0)); }}
#[cfg(feature = "indexer_stream")]impl StreamError { pub fn code(&self) -> &'static str { match self { Self::ConsumerTooSlow { .. } => "ConsumerTooSlow", } }}
#[cfg(feature = "indexer_stream")]impl Stream for EventStream { type Item = Result<Event, StreamError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { self.receiver.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<AppState>);
impl BackfillHandle { pub(crate) fn new(state: Arc<AppState>) -> 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() }}
#[cfg(feature = "indexer_stream")]impl Hydrant { /// Return the highest committed durable indexer-stream event ID. /// /// Consumers can snapshot this before subscribing and report readiness only /// after their committed cursor reaches the snapshot. pub fn stream_head(&self) -> miette::Result<Option<u64>> { self.state.db.stream.event_head() }
/// 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. slow consumers receive /// [`StreamError::ConsumerTooSlow`] before the stream terminates when possible. pub fn subscribe(&self, cursor: Option<u64>) -> EventStream { let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY); let state = self.state.clone(); let runtime = tokio::runtime::Handle::current(); let opts = stream::StreamOptions::from_config(&self.config);
std::thread::Builder::new() .name("hydrant-stream".into()) .spawn(move || { let _g = runtime.enter(); event_stream_thread(state, tx, cursor, opts); }) .expect("failed to spawn stream thread");
EventStream { receiver: rx, wake: self.state.db.stream.event_tx.clone(), } }
#[cfg(feature = "indexer_stream")] #[doc(hidden)] pub fn seed_events_for_bench(&self, count: usize) -> usize { use crate::db::keys; use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid}; use crate::types::{StoredData, StoredEvent}; use jacquard_common::types::did::Did; use jacquard_common::{CowStr, IntoStatic}; use std::sync::atomic::Ordering;
let db = &self.state.db; let mut batch = db.inner.batch(); let mut total_bytes = 0usize;
let did_str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaaaaaa"; let did: Did = Did::new_owned(did_str).expect("valid plc did"); let trimmed = TrimmedDid::from(&did).into_static(); let rev = DbTid::new_from_bytes([0u8; 8]); let collection = CowStr::Borrowed("app.bsky.feed.post").into_static(); let block = bytes::Bytes::from( serde_ipld_dagcbor::to_vec(&serde_json::json!({ "$type": "app.bsky.feed.post", "text": "seeded event", "createdAt": "2026-01-01T00:00:00Z" })) .expect("fixture record must encode as DAG-CBOR"), );
for i in 0..count { let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst); let rkey = DbRkey::Str(smol_str::format_smolstr!("r{i}")); let cid = jacquard_repo::mst::util::compute_cid(block.as_ref()) .expect("fixture record must hash"); db.indexer.stage_record( &mut batch, crate::db::keys::record_key(&did, &collection, &rkey), block.as_ref(), ); let evt = StoredEvent { live: false, did: trimmed.clone(), rev, collection: collection.clone(), rkey, action: DbAction::Create, data: StoredData::Ptr(cid), }; let bytes = rmp_serde::to_vec(&evt).expect("msgpack serialization cannot fail"); total_bytes += bytes.len(); db.stream .stage_event(&mut batch, keys::event_key(event_id), bytes); }
batch.commit().expect("failed to commit events batch"); total_bytes }}