Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434//! per-mode keyspace groups. this is the single place where mode-specific//! database state is declared: each feature contributes one group struct,//! opened and initialized here, and `Db` embeds each group behind one cfg.//!//! keyspace names, tuning, and debug rendering live in `db::schema`; the//! by-name and enumeration tables are generated in `db::registry`.
use fjall::{CompressionType, Database, Keyspace, KeyspaceCreateOptions};use miette::{IntoDiagnostic, Result};use std::cell::RefCell;use std::sync::Arc;
use crate::config::Config;#[cfg(feature = "indexer")]use jacquard_common::types::did::Did;
#[cfg(any( feature = "indexer", feature = "indexer_stream", feature = "jetstream", feature = "relay"))]use super::schema::{self, Ks};
#[cfg(feature = "jetstream")]use super::sequencer::Clock;#[cfg(any(feature = "indexer_stream", feature = "relay"))]use super::sequencer::Sequence;#[cfg(any(feature = "indexer_stream", feature = "jetstream", feature = "relay"))]use miette::Context;
/// everything a keyspace needs to open itself.////// nameable so `Schema::options` can reference it in its signature, but all/// fields and methods are `pub(super)`: it cannot be constructed or used/// outside `db`.pub struct OpenCx<'a> { pub(super) db: &'a Arc<Database>, pub(super) cfg: &'a Config, pub(super) compression: &'a dyn Fn(&str, i32) -> CompressionType, /// names opened so far, checked against the registry at the end of open. pub(super) opened: RefCell<Vec<&'static str>>, pub(super) poison: super::Poison,}
impl OpenCx<'_> { pub(super) fn open_ks(&self, name: &str, opts: KeyspaceCreateOptions) -> Result<Keyspace> { self.db.keyspace(name, move || opts).into_diagnostic() }
pub(super) fn record_opened(&self, name: &'static str) { self.opened.borrow_mut().push(name); }}
#[cfg(feature = "indexer")]#[derive(Clone)]pub struct IndexerDb { /// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker pub(super) heads: Ks<schema::Heads>, /// superseded bodies/markers: `{record key} 00 {death_rev}` -> body/CID/empty pub(super) history: Ks<schema::History>, /// operator-redacted versions: `{record key} 00 {cid}` -> empty pub(super) redactions: Ks<schema::Redactions>, /// backfill queue of `{ID}` -> `{DID}` pub(super) pending: Ks<schema::Pending>, /// the same queue in did order, `{DID} 00 {ID}` -> empty. both are written /// only by `stage_pending_insert` and `stage_pending_remove`, which keep them /// in step pub(super) pending_dids: Ks<schema::PendingDids>, /// per-repo resync/retry state pub resync: Ks<schema::Resync>, /// live events buffered during backfill pub resync_buffer: Ks<schema::ResyncBuffer>,}
#[cfg(feature = "indexer")]impl IndexerDb { pub(super) fn open(cx: &OpenCx) -> Result<Self> { Ok(Self { heads: Ks::open(cx)?, history: Ks::open(cx)?, redactions: Ks::open(cx)?, pending: Ks::open(cx)?, pending_dids: Ks::open(cx)?, resync: Ks::open(cx)?, resync_buffer: Ks::open(cx)?, }) } pub(crate) fn head<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> { self.heads.get(key) }
pub(crate) fn head_prefix<K: AsRef<[u8]>>(&self, prefix: K) -> fjall::Iter { self.heads.prefix(prefix) }
pub(crate) fn head_range<K, R>(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds<K>, { self.heads.range(range) }
// read-side only: stream inflation is today's sole history reader #[cfg(feature = "indexer_stream")] pub(crate) fn history_range<K, R>(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds<K>, { self.history.range(range) }
/// read from table and memtable metadata, so entries that expired, were /// erased or overwritten still count until compaction reaches them pub(crate) fn approximate_history_count(&self) -> usize { self.history.approximate_len() }
#[cfg(any(test, feature = "indexer_stream"))] pub(crate) fn stage_head<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into<fjall::UserKey>, V: Into<fjall::UserValue>, { batch.insert(&self.heads, key, value); }
#[cfg(test)] #[allow(dead_code)] pub(crate) fn stage_history<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into<fjall::UserKey>, V: Into<fjall::UserValue>, { batch.insert(&self.history, key, value); }
/// queues a backfill of `did` under `pending_key` pub(crate) fn stage_pending_insert( &self, batch: &mut fjall::OwnedWriteBatch, did: &Did, pending_key: &[u8], ) { let repo_key = super::keys::repo_key(did); let did_key = super::keys::pending_did_key(&repo_key, pending_key); batch.insert(&self.pending_dids, did_key, []); batch.insert(&self.pending, pending_key, repo_key); }
/// drops `did`'s queue entry under `pending_key`. the did has to be the /// entry's own, or its did-order copy stays behind pub(crate) fn stage_pending_remove( &self, batch: &mut fjall::OwnedWriteBatch, did: &Did, pending_key: &[u8], ) { let repo_key = super::keys::repo_key(did); batch.remove( &self.pending_dids, super::keys::pending_did_key(&repo_key, pending_key), ); batch.remove(&self.pending, pending_key); }
/// panics unless `pending_dids` holds exactly the did-order copy of every /// `pending` entry #[cfg(test)] pub(crate) fn assert_pending_dids_match(&self) { let expected = self .pending .iter() .map(|guard| { let (pending_key, repo_key) = guard.into_inner().unwrap(); super::keys::pending_did_key(&repo_key, &pending_key) }) .collect::<std::collections::BTreeSet<_>>(); let actual = self .pending_dids .iter() .map(|guard| guard.key().unwrap().to_vec()) .collect::<std::collections::BTreeSet<_>>(); assert_eq!(actual, expected, "pending_dids drifted from pending"); }
pub(crate) fn is_pending(&self, pending_key: &[u8]) -> fjall::Result<bool> { self.pending.contains_key(pending_key) }
/// the queue in `pending_key` order, each entry's value is its repo key pub(crate) fn pending_iter(&self) -> fjall::Iter { self.pending.iter() }
/// the queue in did order, keys are `pending_did_key`s pub(crate) fn pending_did_range<K, R>(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds<K>, { self.pending_dids.range(range) }}
#[cfg(feature = "indexer_stream")]#[derive(Clone)]pub struct StreamDb { /// maps `{ID}` (u64 BE) -> `StoredEvent`, the source for the json stream api pub(super) events: Ks<schema::Events>, /// bodies needed only by legacy events whose record is no longer the /// current head: `{record key} 00 {cid}` -> raw DAG-CBOR body pub(crate) event_bodies: Ks<schema::EventBodies>, pub(crate) event_tx: tokio::sync::broadcast::Sender<crate::types::BroadcastEvent>, /// event ids, taken only by sequenced commits pub(crate) ids: Arc<Sequence>, /// a wiped db or another node has its own id, and their cursors never match here. id: Arc<std::sync::OnceLock<String>>, chunks: Arc<std::sync::OnceLock<super::stream_chunks::StreamChunks>>,}
#[cfg(feature = "indexer_stream")]impl StreamDb { pub(super) fn open(cx: &OpenCx) -> Result<Self> { let (event_tx, _) = tokio::sync::broadcast::channel(512);
Ok(Self { events: Ks::open(cx)?, event_bodies: Ks::open(cx)?, event_tx, ids: Arc::new(Sequence::new(super::keys::STREAM_EVENT_IDS_MARK)), id: Arc::default(), chunks: Arc::default(), }) }
pub(crate) fn id(&self) -> &str { self.id.get().map_or("", String::as_str) }
pub(crate) fn chunks(&self) -> Option<&super::stream_chunks::StreamChunks> { self.chunks.get() }
pub(crate) fn event<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> { self.events.get(key) }
pub(crate) fn event_range<K, R>(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds<K>, { self.events.range(range) }
/// write an event row with a chosen id, for test fixtures. real events /// take their ids from the outbox when they commit. #[cfg(test)] pub(crate) fn stage_event<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into<fjall::UserKey>, V: Into<fjall::UserValue>, { batch.insert(&self.events, key, value); }
pub(crate) fn approximate_event_count(&self) -> usize { self.events.approximate_len() }
pub(crate) fn event_head(&self) -> Result<Option<u64>> { let Some(guard) = self.events.iter().next_back() else { return Ok(None); }; let key = guard.key().into_diagnostic()?; let id = u64::from_be_bytes( key.as_ref() .try_into() .into_diagnostic() .wrap_err("expected stream event ID to be 8 bytes")?, ); Ok(Some(id)) }
pub(super) fn init( &self, marks: &Ks<schema::Cursors>, cfg: &crate::config::Config, ) -> Result<()> { self.ids.resume_after(marks, self.event_head()?)?; let id = match marks.get(super::keys::STREAM_ID).into_diagnostic()? { Some(id) => String::from_utf8_lossy(&id).into_owned(), None => { let id = crate::util::hex::encode(rand::random::<[u8; 16]>()); marks .insert(super::keys::STREAM_ID, id.as_bytes()) .into_diagnostic()?; id } }; // sealed chunks can't follow ephemeral rows or trimmed history. let dir = cfg.database_path.join("stream-chunks"); if cfg.stream_chunk_span > 0 && !cfg.ephemeral && cfg.history_ttl.is_none() { let chunks = super::stream_chunks::StreamChunks::open( dir, cfg.stream_chunk_span, &id, marks.clone(), )?; let _ = self.chunks.set(chunks); } else if dir.exists() { // nothing marks what a redaction reaches, and old chunks would miss it. std::fs::remove_dir_all(&dir).into_diagnostic()?; } let _ = self.id.set(id); Ok(()) }}
#[cfg(feature = "jetstream")]#[derive(Clone)]pub(crate) struct JetstreamDb { /// maps `{time_us}|{ID}` (16 bytes) -> jetstream event data pub(crate) events: Ks<schema::JetstreamEvents>, pub(crate) tx: tokio::sync::broadcast::Sender<crate::types::JetstreamBroadcast>, /// key ids and timestamps, taken only by sequenced commits pub(crate) ids: Arc<Sequence>, pub(crate) clock: Arc<Clock>,}
#[cfg(feature = "jetstream")]impl JetstreamDb { pub(super) fn open(cx: &OpenCx) -> Result<Self> { let (tx, _) = tokio::sync::broadcast::channel(512);
Ok(Self { events: Ks::open(cx)?, tx, ids: Arc::new(Sequence::new(super::keys::JETSTREAM_IDS_MARK)), clock: Arc::new(Clock::new(super::keys::JETSTREAM_TIME_US_MARK)), }) }
/// the last committed event's position. pub(crate) fn head(&self) -> Result<Option<crate::types::JetstreamPosition>> { let Some(guard) = self.events.iter().next_back() else { return Ok(None); }; let key = guard.key().into_diagnostic()?; let (time_us, id) = super::keys::parse_jetstream_event_key(&key)?; Ok(Some(crate::types::JetstreamPosition { time_us, id })) }
/// resume jetstream ids and time watermark after the last stored event, /// or the last ones taken. pub(super) fn init(&self, marks: &Ks<schema::Cursors>) -> Result<()> { let head = self.head()?; self.ids.resume_after(marks, head.map(|head| head.id))?; self.clock .resume_after(marks, head.map(|head| head.time_us)) }}
#[cfg(feature = "relay")]#[derive(Clone)]pub(crate) struct RelayDb { /// maps `{SEQ}` (u64 BE) -> re-encoded relay frame pub(crate) events: Ks<schema::RelayEvents>, /// relay seqs, taken only by sequenced commits pub(crate) seqs: Arc<Sequence>, pub(crate) broadcast_tx: tokio::sync::broadcast::Sender<crate::types::RelayBroadcast>,}
#[cfg(feature = "relay")]impl RelayDb { pub(super) fn open(cx: &OpenCx) -> Result<Self> { let (broadcast_tx, _) = tokio::sync::broadcast::channel(512);
Ok(Self { events: Ks::open(cx)?, seqs: Arc::new(Sequence::new(super::keys::RELAY_SEQS_MARK)), broadcast_tx, }) }
/// resume relay sequence numbers after the last stored frame, or the /// last one taken. pub(super) fn init(&self, marks: &Ks<schema::Cursors>) -> Result<()> { let last_relay_seq = self .events .iter() .next_back() .map(|guard| { let key = guard.key().into_diagnostic()?; key.as_ref() .try_into() .map(u64::from_be_bytes) .into_diagnostic() .wrap_err("relay_events: invalid key length") }) .transpose()?; self.seqs.resume_after(marks, last_relay_seq) }}
#[cfg(all(test, feature = "relay"))]mod tests { use crate::config::Config; use crate::db::Db;
#[test] fn relay_seqs_resume_after_the_last_big_endian_frame_key() { let tmp = tempfile::tempdir().unwrap(); let cfg = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; { let db = Db::open(&cfg).unwrap(); db.relay .events .insert([0, 0, 0, 0, 0, 0, 1, 2], b"frame") .unwrap(); db.persist().unwrap(); } let db = Db::open(&cfg).unwrap(); assert_eq!(db.relay.seqs.next(), 259); }}