Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816//! keyspace schemas: the single table of truth for every keyspace.//!//! each keyspace is declared as one type implementing [`Schema`], carrying its//! on-disk name, open-time tuning, optional zstd dictionary training config,//! and debug key/value rendering. the registry (`db::registry`) derives the//! by-name lookup, `/stats` enumeration, `/debug/*` dispatch, and dictionary//! training from these types, so a keyspace cannot be half-registered.
use std::marker::PhantomData;
use fjall::{ CompressionType, Keyspace, KeyspaceCreateOptions, config::{BlockSizePolicy, CompressionPolicy, RestartIntervalPolicy},};use miette::Result;use serde_json::Value;
use super::keyspaces::OpenCx;
pub(super) const fn kb(v: u32) -> u32 { v * 1024}
pub(super) const fn mb(v: u64) -> u64 { v * 1024 * 1024}
/// zstd dictionary training configuration for a keyspace.#[derive(Debug, Clone, Copy)]pub struct TrainCfg { /// maximum dictionary size in bytes. pub dict_size: u32, /// number of data blocks to sample. pub samples: usize,}
/// a keyspace schema: name, tuning, training, and debug rendering.pub trait Schema { /// on-disk keyspace name. THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE. const NAME: &'static str;
/// zstd dictionary training config; `None` disables training and /// dictionary loading for this keyspace. const TRAIN: Option<TrainCfg> = None;
/// per-keyspace tuning applied at open. fn options(cx: &OpenCx) -> KeyspaceCreateOptions;
/// render a raw value for `/debug/*` endpoints. fn render_value(value: &[u8]) -> Value { Value::String(hex::encode(value)) }
/// parse a user-supplied `/debug/*` key string into key bytes. /// `None` means the string is not a valid key for this keyspace. fn parse_debug_key(key: &str) -> Option<Vec<u8>> { Some(key.as_bytes().to_vec()) }
/// render raw key bytes for `/debug/iter` output. fn render_debug_key(key: &[u8]) -> String { String::from_utf8_lossy(key).into_owned() }}
/// typed handle over a [`Keyspace`], parameterized by its [`Schema`].////// zero-cost: one `Keyspace` handle plus a marker. cloning is as cheap as/// cloning the underlying handle.pub struct Ks<S: Schema> { inner: Keyspace, _s: PhantomData<fn() -> S>,}
impl<S: Schema> Clone for Ks<S> { fn clone(&self) -> Self { Self { inner: self.inner.clone(), _s: PhantomData, } }}
impl<S: Schema> Ks<S> { /// open this keyspace with its schema tuning, recording the name for the /// open-time registry completeness check. pub(super) fn open(cx: &OpenCx) -> Result<Self> { let inner = cx.open_ks(S::NAME, S::options(cx))?; cx.record_opened(S::NAME); Ok(Self { inner, _s: PhantomData, }) }
/// an owned clone of the underlying keyspace handle. pub fn keyspace(&self) -> Keyspace { self.inner.clone() }
/// the raw keyspace handle. escape hatch for migrations and debug paths. pub fn raw(&self) -> &Keyspace { &self.inner }
/// point reads/writes mirror [`Keyspace`]'s signatures exactly, adding /// the fatal poison check every callsite previously had to remember. /// inherent methods win over the deref bridge, so all `db.<ks>.<op>()` /// callsites route through these. #[inline] pub fn get<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> { self.inner.get(key).inspect_err(super::check_poisoned) }
#[inline] pub fn contains_key<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<bool> { self.inner .contains_key(key) .inspect_err(super::check_poisoned) }
#[inline] pub fn insert<K: Into<fjall::UserKey>, V: Into<fjall::UserValue>>( &self, key: K, value: V, ) -> fjall::Result<()> { self.inner .insert(key, value) .inspect_err(super::check_poisoned) }
#[inline] pub fn remove<K: Into<fjall::UserKey>>(&self, key: K) -> fjall::Result<()> { self.inner.remove(key).inspect_err(super::check_poisoned) }}
/// phase-1 migration bridge: existing callsites use raw `Keyspace` methods/// directly. removed once callsites are ported to typed accessors.impl<S: Schema> std::ops::Deref for Ks<S> { type Target = Keyspace;
fn deref(&self) -> &Keyspace { &self.inner }}
fn msgpack_or_hex<T: serde::de::DeserializeOwned + serde::Serialize>(value: &[u8]) -> Value { rmp_serde::from_slice::<T>(value) .ok() .and_then(|v| serde_json::to_value(v).ok()) .unwrap_or_else(|| Value::String(hex::encode(value)))}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]fn u64_be_key_render(key: &[u8]) -> String { key.try_into() .map(|arr| u64::from_be_bytes(arr).to_string()) .unwrap_or_else(|_| "invalid_u64".to_string())}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]fn u64_be_key_parse(key: &str) -> Option<Vec<u8>> { key.parse::<u64>().ok().map(|id| id.to_be_bytes().to_vec())}
/// maps `{DID}` -> `RepoState` (msgpack)pub struct Repos;
impl Schema for Repos { const NAME: &'static str = "repos"; const TRAIN: Option<TrainCfg> = Some(TrainCfg { dict_size: kb(64), samples: 1000, });
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // crawler checks if a repo doesn't exist .expect_point_read_hits(false) .max_memtable_size(mb(cx.cfg.db_repos_memtable_size_mb)) // did -> repo state, not gonna be compressable well because dids are random // and repo state doesnt have repeats really.. // these block sizes work fine since we insert into repos constantly anyway // whenever we update anything related to a repo, so the repos that arent // being updated will be compacted away! .data_block_size_policy(BlockSizePolicy::new([kb(4), kb(4), kb(16), kb(64)])) .data_block_compression_policy(CompressionPolicy::new([ CompressionType::None, CompressionType::None, (cx.compression)("repos", 3), (cx.compression)("repos", 5), ])) // did plc are random so the interval wont rlly matter .data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4])) }
fn render_value(value: &[u8]) -> Value { // RepoState borrows from the input, so it can't go through the // DeserializeOwned helper rmp_serde::from_slice::<crate::types::RepoState>(value) .ok() .and_then(|v| serde_json::to_value(&v).ok()) .unwrap_or_else(|| Value::String(hex::encode(value))) }}
/// maps `rm|{DID}` -> `RepoMetadata` (msgpack)pub struct RepoMetadata;
impl Schema for RepoMetadata { const NAME: &'static str = "repo_metadata";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() .expect_point_read_hits(true) .max_memtable_size(mb(cx.cfg.db_repos_memtable_size_mb / 2)) // its did -> random u64 id + bool, not much to compress, very small .data_block_size_policy(BlockSizePolicy::new([kb(2), kb(4), kb(8)])) .data_block_compression_policy(CompressionPolicy::new([ CompressionType::None, CompressionType::None, (cx.compression)("repos", 3), ])) // did plc are random so the interval wont rlly matter .data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4])) }
fn render_value(value: &[u8]) -> Value { msgpack_or_hex::<crate::types::RepoMetadata>(value) }}
/// per-relay cursor keys -> u64/i64 BE bytespub struct Cursors;
impl Schema for Cursors { const NAME: &'static str = "cursors";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // cursor point reads hit almost 100% of the time .expect_point_read_hits(true) .max_memtable_size(mb(4)) // its just cursors... .data_block_size_policy(BlockSizePolicy::all(kb(1))) .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(1)) }
fn render_value(value: &[u8]) -> Value { render_counter(value) }}
/// maps `k|{NAME}` or `r|{DID}|{COL}` -> count (u64 BE), plus delta logpub struct Counts;
impl Schema for Counts { const NAME: &'static str = "counts";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // count increments hit because counters are mostly pre-initialized .expect_point_read_hits(true) .max_memtable_size(mb(16)) // the data is very small // this is at worst did|col -> u64, so its tiny (40 bytes) .data_block_size_policy(BlockSizePolicy::all(kb(2))) .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(7)) }
fn render_value(value: &[u8]) -> Value { render_counter(value) }}
fn render_counter(value: &[u8]) -> Value { if let Ok(arr) = value.try_into() { return Value::Number(u64::from_be_bytes(arr).into()); } std::str::from_utf8(value) .map(|s| Value::String(s.to_owned())) .unwrap_or_else(|_| Value::String(hex::encode(value)))}
/// filter config: mode key, signal/collection/exclude sets, pds metapub struct Filter;
impl Schema for Filter { const NAME: &'static str = "filter";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { // filter handles high-volume point reads (repo excludes) so it needs the bloom filter KeyspaceCreateOptions::default() .max_memtable_size(mb(16)) // dids arent compressable so this is fine, and we have nothing as the value .data_block_size_policy(BlockSizePolicy::all(kb(1))) .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(2)) }}
/// crawler + firehose source urls and retry entriespub struct Crawler;
impl Schema for Crawler { const NAME: &'static str = "crawler";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // only iterators are used here .expect_point_read_hits(true) .max_memtable_size(mb(8)) // did -> failed state, not very compressable .data_block_size_policy(BlockSizePolicy::all(kb(2))) .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(2)) }}
/// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker.#[cfg(feature = "indexer")]pub struct Heads;
#[cfg(feature = "indexer")]impl Schema for Heads { const NAME: &'static str = "heads";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // point reads might miss when using getRecord // but we assume thats not going to happen often... // since this keyspace is big, turning off bloom filters will help a lot with memory/disk space, // but leaves point reads vulnerable to disk I/O misses under public query load .expect_point_read_hits(!cx.cfg.db_records_bloom_filters) .max_memtable_size(mb(cx.cfg.db_records_memtable_size_mb)) // did 00 col 2f rkey -> record body (dag-cbor). bodies compress well // grouped by repo: author/time locality beats the old col|cid // content addressing (-56% measured on a live corpus, zstd:3@64k) .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64), kb(128)])) // compress from the first level: bulk loads (migrations) write in // sorted order, so their tables never overlap and compaction // trivial-moves them without ever re-compressing — a None first // level sticks to the whole tree (measured: 71MB stored vs 16MB // zstd:3 standalone on the same bodies). heads stay hot in the // memtable regardless. v10 migrates pre-v10 databases into this // keyspace because v9's `records` persisted "never compress" into // fjall's on-disk config, which no open-time schema change can // undo. .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("heads", 3), (cx.compression)("heads", 3), (cx.compression)("heads", 3), (cx.compression)("heads", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])) }
fn render_value(value: &[u8]) -> Value { serde_ipld_dagcbor::from_slice::<Value>(value) .unwrap_or_else(|_| Value::String(hex::encode(value))) }
fn render_debug_key(key: &[u8]) -> String { // key is {trimmed did} 00 {collection} 2f {rkey}, show as "did|col|rkey" let Some(did_len) = super::keys::indexer::trimmed_did_len(key) else { return String::from_utf8_lossy(key).into_owned(); }; let Ok(did) = crate::db::types::TrimmedDid::try_from(&key[..did_len]) else { return String::from_utf8_lossy(key).into_owned(); }; match super::keys::indexer::split_record_suffix(&key[did_len + 1..]) { Ok((col, rkey)) => format!("{did}|{col}|{rkey}"), Err(_) => String::from_utf8_lossy(key).into_owned(), } }
fn parse_debug_key(key: &str) -> Option<Vec<u8>> { // accept "did|col|rkey" and encode as a current record key. the // rendered did is trimmed (no `did:` prefix), so restore it let (did, rest) = key.split_once('|')?; let (col, rkey) = rest.split_once('|')?; let did_str = format!("did:{did}"); let did = jacquard_common::types::string::Did::<&str>::new(&did_str).ok()?; let trimmed = crate::db::types::TrimmedDid::from_did_str(did.as_str()); Some(super::keys::indexer::record_key_trimmed( &trimmed, col, &crate::db::types::DbRkey::new(rkey), )) }}
/// superseded records: `{record key} 00 {death_rev}` -> body, CID marker, or tombstone.////// written when an update/delete (live commit, backfill resync diff, or repo/// removal) replaces or removes a head body in `records`; read only by stream/// inflation of pre-deletion events. retention is the deployment's config:/// permanent indexers keep everything, bounded retention is drop_range-shaped.#[cfg(feature = "indexer")]pub struct History;
#[cfg(feature = "indexer")]impl Schema for History { const NAME: &'static str = "history";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // seeks here are speculative and usually miss, then fall through to // the records head; bloom filters make those misses cheap .expect_point_read_hits(false) .max_memtable_size(mb(16)) // entries are written once and only read on replay; big blocks and // compression at every level .data_block_size_policy(BlockSizePolicy::new([kb(64), kb(128)])) .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("history", 3), (cx.compression)("history", 3), (cx.compression)("history", 3), (cx.compression)("history", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])) }
fn render_value(value: &[u8]) -> Value { serde_ipld_dagcbor::from_slice::<Value>(value) .unwrap_or_else(|_| Value::String(hex::encode(value))) }
fn render_debug_key(key: &[u8]) -> String { // key is {record key} 00 {death_rev: 8 BE}; the rev is always the tail if key.len() < 9 || key[key.len() - 9] != super::keys::indexer::REC_SEP { return String::from_utf8_lossy(key).into_owned(); } let rev_bytes: [u8; 8] = key[key.len() - 8..].try_into().expect("length checked"); let rev = crate::db::types::DbTid::new_from_bytes(rev_bytes); format!("{}@{rev}", Heads::render_debug_key(&key[..key.len() - 9])) }}
/// durable operator-redaction identities: `{record key} 00 {cid}` -> empty.////// a marker prevents a later replay or recurrence of the same record version/// from restoring bytes that an operator explicitly erased. entries are sparse/// and permanent; unlike ordinary history they are never TTL-filtered.#[cfg(feature = "indexer")]pub struct Redactions;
#[cfg(feature = "indexer")]impl Schema for Redactions { const NAME: &'static str = "redactions";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() .expect_point_read_hits(false) .max_memtable_size(mb(4)) .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64)])) .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("redactions", 3), (cx.compression)("redactions", 3), (cx.compression)("redactions", 3), (cx.compression)("redactions", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])) }
fn render_value(_value: &[u8]) -> Value { Value::Null }
fn render_debug_key(key: &[u8]) -> String { let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else { return String::from_utf8_lossy(key).into_owned(); }; format!("{}#{cid}", Heads::render_debug_key(record_key)) }}
/// record-scoped bodies retained solely for legacy stream events:/// `{record key} 00 {cid}` -> dag-cbor body.////// v10 materializes these from the frozen blocks CAS before deleting it. new/// events never write this keyspace: their bodies resolve through the current/// record head or ordinary death-keyed history.#[cfg(feature = "indexer_stream")]pub struct EventBodies;
#[cfg(feature = "indexer_stream")]impl Schema for EventBodies { const NAME: &'static str = "event_bodies";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() .expect_point_read_hits(false) .max_memtable_size(mb(16)) .data_block_size_policy(BlockSizePolicy::new([kb(64), kb(128)])) .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("event_bodies", 3), (cx.compression)("event_bodies", 3), (cx.compression)("event_bodies", 3), (cx.compression)("event_bodies", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])) }
fn render_value(value: &[u8]) -> Value { serde_ipld_dagcbor::from_slice::<Value>(value) .unwrap_or_else(|_| Value::String(hex::encode(value))) }
fn render_debug_key(key: &[u8]) -> String { let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else { return String::from_utf8_lossy(key).into_owned(); }; format!("{}#{cid}", Heads::render_debug_key(record_key)) }}
/// backfill queue of `{ID}` (u64 BE) -> empty#[cfg(feature = "indexer")]pub struct Pending;
#[cfg(feature = "indexer")]impl Schema for Pending { const NAME: &'static str = "pending";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // iterated over as a queue, no point reads are used so bloom filters are disabled .expect_point_read_hits(true) .max_memtable_size(mb(8)) // its just index of id (int) -> did, and dids arent compressable (especially with the ids being random) .data_block_size_policy(BlockSizePolicy::all(kb(8))) // and we'll transition from pending to synced anyway, no point trying to compress .data_block_compression_policy(CompressionPolicy::disabled()) // ids are sequential and share prefix so we can use large interval to save space .data_block_restart_interval_policy(RestartIntervalPolicy::all(64)) }
fn render_value(_value: &[u8]) -> Value { Value::Null }}
/// per-repo resync/retry state, `{DID}` -> `ResyncState` (msgpack)#[cfg(feature = "indexer")]pub struct Resync;
#[cfg(feature = "indexer")]impl Schema for Resync { const NAME: &'static str = "resync";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // we only point read in backfill when we check for existing resync state // ...and also in repos api. so we can disable bloom filters .expect_point_read_hits(true) .max_memtable_size(mb(8)) // did -> error state, so its gonna be basically random, cant compress well .data_block_size_policy(BlockSizePolicy::all(kb(4))) // and we arent going to have many of these anyway, no point trying .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(4)) }
fn render_value(value: &[u8]) -> Value { msgpack_or_hex::<crate::types::ResyncState>(value) }}
/// live events buffered during backfill, `{DID}|{Rev}` -> commit (msgpack)#[cfg(feature = "indexer")]pub struct ResyncBuffer;
#[cfg(feature = "indexer")]impl Schema for ResyncBuffer { const NAME: &'static str = "resync_buffer";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // iterated during backfill, no point reads .expect_point_read_hits(true) .max_memtable_size(mb(16)) .data_block_size_policy(BlockSizePolicy::all(kb(32))) // dont have to compress here since resync buffer will be emptied at some point anyway .data_block_compression_policy(CompressionPolicy::disabled()) .data_block_restart_interval_policy(RestartIntervalPolicy::all(16)) }}
/// maps `{ID}` (u64 BE) -> `StoredEvent`, the source for the json stream api#[cfg(feature = "indexer_stream")]pub struct Events;
#[cfg(feature = "indexer_stream")]impl Schema for Events { const NAME: &'static str = "events"; const TRAIN: Option<TrainCfg> = Some(TrainCfg { dict_size: kb(64), samples: 1000, });
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // only iterators are used here, no point reads .expect_point_read_hits(true) .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb)) // the compression here wont be quite as good since events are quite random // eg. by many different repos and different records etc. // since its sequential we should still go with bigger block size though // backfills will be sequential though... .data_block_size_policy( cx.cfg .ephemeral .then(|| BlockSizePolicy::new([kb(64), kb(128), kb(256)])) .unwrap_or_else(|| BlockSizePolicy::new([kb(128), kb(256)])), ) // monotonic {ID} keys are trivial-moved down (never rewritten), so the L0 // policy is permanent; the live tail is served uncompressed from the memtable .data_block_compression_policy( cx.cfg .ephemeral .then(|| { CompressionPolicy::new([ CompressionType::None, (cx.compression)("events", 3), ]) }) .unwrap_or_else(|| { CompressionPolicy::new([ (cx.compression)("events", 3), (cx.compression)("events", 3), (cx.compression)("events", 3), (cx.compression)("events", 5), ]) }), ) // ids are int, we can prefix truncate a lot .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128])) }
fn render_value(value: &[u8]) -> Value { // StoredEvent borrows from the input, so it can't go through the // DeserializeOwned helper rmp_serde::from_slice::<crate::types::StoredEvent>(value) .ok() .and_then(|v| serde_json::to_value(&v).ok()) .unwrap_or_else(|| Value::String(hex::encode(value))) }
fn parse_debug_key(key: &str) -> Option<Vec<u8>> { u64_be_key_parse(key) }
fn render_debug_key(key: &[u8]) -> String { u64_be_key_render(key) }}
/// maps `{time_us}|{ID}` (16 bytes) -> jetstream event data#[cfg(feature = "jetstream")]pub struct JetstreamEvents;
#[cfg(feature = "jetstream")]impl Schema for JetstreamEvents { const NAME: &'static str = "jetstream_events"; const TRAIN: Option<TrainCfg> = Some(TrainCfg { dict_size: kb(64), samples: 1000, });
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // time-ordered append-only stream metadata, only iterated for replay. .expect_point_read_hits(true) .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb)) .data_block_size_policy(BlockSizePolicy::new([kb(128), kb(256)])) .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("jetstream_events", 3), (cx.compression)("jetstream_events", 3), (cx.compression)("jetstream_events", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128])) }
fn render_value(value: &[u8]) -> Value { // StoredJetstreamEvent borrows from the input, so it can't go through // the DeserializeOwned helper rmp_serde::from_slice::<crate::types::StoredJetstreamEvent>(value) .ok() .and_then(|v| serde_json::to_value(&v).ok()) .unwrap_or_else(|| Value::String(hex::encode(value))) }}
/// maps `{SEQ}` (u64 BE) -> re-encoded relay frame#[cfg(feature = "relay")]pub struct RelayEvents;
#[cfg(feature = "relay")]impl Schema for RelayEvents { const NAME: &'static str = "relay_events";
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // only iterated for cursor replay .expect_point_read_hits(true) .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb)) .data_block_size_policy(BlockSizePolicy::new([kb(128), kb(256)])) // monotonic {SEQ} keys are trivial-moved down (never rewritten), so L0 // compression is permanent; the live tail is served from the memtable .data_block_compression_policy(CompressionPolicy::new([ (cx.compression)("relay_events", 3), (cx.compression)("relay_events", 3), (cx.compression)("relay_events", 3), (cx.compression)("relay_events", 5), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128])) }
fn parse_debug_key(key: &str) -> Option<Vec<u8>> { u64_be_key_parse(key) }
fn render_debug_key(key: &[u8]) -> String { u64_be_key_render(key) }}
/// reverse index of record references#[cfg(feature = "backlinks")]pub struct Backlinks;
#[cfg(feature = "backlinks")]impl Schema for Backlinks { const NAME: &'static str = "backlinks"; const TRAIN: Option<TrainCfg> = Some(TrainCfg { dict_size: kb(64), samples: 1000, });
fn options(cx: &OpenCx) -> KeyspaceCreateOptions { KeyspaceCreateOptions::default() // lets assume we hit backlinks, getBacklinks will use iterator anyway // so we can disable bloom filter okay .expect_point_read_hits(true) .max_memtable_size(mb(cx.cfg.db_records_memtable_size_mb)) // same as records basically .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(32)])) .data_block_compression_policy(CompressionPolicy::new([ CompressionType::None, (cx.compression)("backlinks", 3), ])) .data_block_restart_interval_policy(RestartIntervalPolicy::new([16, 32])) }}
#[cfg(test)]mod tests { use std::collections::BTreeSet;
#[test] fn keyspace_names_match_the_compiled_storage_schema() { let tmp = tempfile::tempdir().unwrap(); let db = crate::db::Db::open(&crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }) .unwrap();
let expected = BTreeSet::from([ "repos", "repo_metadata", "cursors", "counts", "filter", "crawler", ]); #[cfg(any( feature = "backlinks", feature = "indexer", feature = "indexer_stream", feature = "jetstream", feature = "relay" ))] let mut expected = expected; #[cfg(feature = "backlinks")] expected.insert("backlinks"); #[cfg(feature = "indexer")] expected.extend([ "heads", "history", "redactions", "pending", "resync", "resync_buffer", ]); #[cfg(feature = "indexer_stream")] expected.extend(["events", "event_bodies"]); #[cfg(feature = "jetstream")] expected.insert("jetstream_events"); #[cfg(feature = "relay")] expected.insert("relay_events");
let actual = db .all_keyspaces() .into_iter() .map(|(name, _)| name) .collect::<BTreeSet<_>>(); assert_eq!(actual, expected); }}