//! 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;
#[cfg(feature = "indexer")]
use fjall::config::{FilterPolicy, PinningPolicy};
use fjall::{
CompressionType, Keyspace, KeyspaceCreateOptions,
config::{BlockSizePolicy, CompressionPolicy, RestartIntervalPolicy},
};
use miette::Result;
use serde_json::Value;
use crate::util::hex;
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`].
///
/// one `Keyspace` handle plus the database's poison flag, both cheap to clone.
pub struct Ks<S: Schema> {
inner: Keyspace,
poison: super::Poison,
_s: PhantomData<fn() -> S>,
}
impl<S: Schema> Clone for Ks<S> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
poison: self.poison.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,
poison: cx.poison.clone(),
_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(|e| self.poison.check(e))
}
#[inline]
pub fn contains_key<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<bool> {
self.inner
.contains_key(key)
.inspect_err(|e| self.poison.check(e))
}
#[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(|e| self.poison.check(e))
}
#[inline]
pub fn remove<K: Into<fjall::UserKey>>(&self, key: K) -> fjall::Result<()> {
self.inner.remove(key).inspect_err(|e| self.poison.check(e))
}
}
/// 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 bytes
pub 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 log
pub 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 meta
pub 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 entries
pub 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]))
}
fn parse_debug_key(key: &str) -> Option<Vec<u8>> {
// "did|col|rkey@rev". no segment of a record key can hold an `@`
let (record, rev) = key.rsplit_once('@')?;
let rev = crate::db::types::DbTid::from_s32(rev)?;
Some(super::keys::indexer::history_key(
&Heads::parse_debug_key(record)?,
&rev,
))
}
}
/// 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) -> `{DID}` (trimmed)
#[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
}
}
/// the backfill queue again in did order, `{DID} 00 {ID}` -> empty, so it can be
/// paged by did next to `resync`
#[cfg(feature = "indexer")]
pub struct PendingDids;
#[cfg(feature = "indexer")]
impl Schema for PendingDids {
const NAME: &'static str = "pending_dids";
fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
KeyspaceCreateOptions::default()
// only ever scanned or written blind, so no level needs a filter and
// nothing has to stay pinned outside the block cache
.filter_policy(FilterPolicy::disabled())
.filter_block_pinning_policy(PinningPolicy::disabled())
.index_block_pinning_policy(PinningPolicy::disabled())
// the same churn as `pending`, but only a listing reads it, so it
// doesn't need to hold as much in memory
.max_memtable_size(mb(4))
.data_block_size_policy(BlockSizePolicy::all(kb(8)))
// dids barely compress and entries leave once their backfill is done
.data_block_compression_policy(CompressionPolicy::disabled())
.data_block_restart_interval_policy(RestartIntervalPolicy::all(16))
}
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",
"pending_dids",
"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);
}
}