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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449//! storage-mode validation at open time.//!//! the storage layout has two immutable axes: `only_index_links` (record//! values are bodies vs cids) and `ephemeral` (bodies live only inside the//! TTL-bounded event log). neither is safely convertible in place://!//! - flipping links-only rewrites what `records` values mean, so replay of//! events written before the flip would parse cids as bodies or vice//! versa. opening with a different value than the database was written//! with is rejected with an actionable error.//! - ephemeral databases have no record heads or history at all. interpreting//! one layout as the other would either discard permanent heads or retain//! body bytes past the promised event TTL.//!//! mismatches are detected via a marker in the `filter` keyspace written on//! first open. databases that predate the marker get it written for the//! currently configured layout: a flip that already happened silently before//! the marker existed stays silent, and future flips are protected.
#[cfg(feature = "indexer_stream")]use miette::Context;use miette::{IntoDiagnostic, Result};use serde::{Deserialize, Serialize};
use crate::config::Config;use crate::db::Db;
/// persisted marker of the storage layout the db was written with, kept in/// the filter keyspace next to the other instance-level configpub(super) const STORAGE_MODE_KEY: &[u8] = b"storage_mode";
#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)]struct StorageModeMarker { links_only: bool, ephemeral: bool,}
/// validate an existing marker before migrations can write anything.#[cfg(feature = "indexer")]pub(crate) fn preflight(db: &Db, cfg: &Config) -> Result<bool> { let marker = StorageModeMarker { links_only: cfg.only_index_links, ephemeral: cfg.ephemeral, };
let Some(stored) = db.filter.get(STORAGE_MODE_KEY).into_diagnostic()? else { if let Some(inferred_ephemeral) = infer_existing_ephemeral(db)? && inferred_ephemeral != marker.ephemeral { return Err(miette::miette!( "legacy event bodies imply ephemeral={inferred_ephemeral} but config requests {}: \ refusing to migrate a markerless database under the wrong storage mode. keep the \ previous HYDRANT_EPHEMERAL setting or use an explicit offline migration", marker.ephemeral, )); } return Ok(false); }; let prev: StorageModeMarker = rmp_serde::from_slice(&stored).into_diagnostic()?; if prev.ephemeral && db .indexer .heads .iter() .next() .map(|guard| guard.into_inner()) .transpose() .into_diagnostic()? .is_some() { miette::bail!( "ephemeral storage marker conflicts with persisted record heads; rebuild this unsupported pre-release database from v9" ); } if prev.links_only != marker.links_only { return Err(miette::miette!( "database was written with only_index_links={} but config requests {}: \ links-only and body storage are not convertible in place. keep the previous \ setting or start from a fresh database path", prev.links_only, marker.links_only, )); } if prev.ephemeral != marker.ephemeral { return Err(miette::miette!( "database was written with ephemeral={} but config requests {}: \ permanent and ephemeral storage are not convertible at startup. keep the \ previous HYDRANT_EPHEMERAL setting, use an explicit offline migration, or \ start from a fresh database path", prev.ephemeral, marker.ephemeral, )); } Ok(true)}
/// infer shipped v9 mode from body-bearing stream events. ephemeral writes/// stored `Block`; permanent writes stored `Ptr`. body-less delete events do/// not identify a mode, and seeing both encodings means the database was/// already switched unsafely before markers existed.fn infer_existing_ephemeral(db: &Db) -> Result<Option<bool>> { let inferred = db .indexer .heads .iter() .next() .map(|guard| guard.into_inner()) .transpose() .into_diagnostic()? .map(|_| false);
#[cfg(feature = "indexer_stream")] { let mut inferred = inferred; for guard in db.stream.events.iter() { let value = guard.value().into_diagnostic()?; let event: crate::types::StoredEvent<'_> = rmp_serde::from_slice(&value) .into_diagnostic() .wrap_err("cannot infer storage mode from a legacy event")?; let ephemeral = match event.data { crate::types::StoredData::Block(_) => true, crate::types::StoredData::Ptr(_) => false, crate::types::StoredData::Nothing => continue, }; if inferred.is_some_and(|previous| previous != ephemeral) { miette::bail!( "database contains both inline ephemeral and pointer-based permanent events; use an explicit offline migration" ); } inferred = Some(ephemeral); } Ok(inferred) } #[cfg(not(feature = "indexer_stream"))] { Ok(inferred) }}
/// adopt the configured layout for a database that predates the marker. this/// runs after migrations because legacy record values alone cannot distinguish/// a body-bearing database from links-only storage.#[cfg(feature = "indexer")]pub(crate) fn adopt(db: &Db, cfg: &Config) -> Result<()> { let marker = StorageModeMarker { links_only: cfg.only_index_links, ephemeral: cfg.ephemeral, }; if let Some(inferred_links_only) = infer_existing_links_only(db)? && inferred_links_only != marker.links_only { return Err(miette::miette!( "database record values imply only_index_links={inferred_links_only} but config requests {}: \ refusing to bless an incompatible pre-marker database. keep the existing setting or \ migrate into a fresh database path", marker.links_only, )); } db.filter .insert( STORAGE_MODE_KEY, rmp_serde::to_vec_named(&marker).into_diagnostic()?, ) .into_diagnostic()?; tracing::info!( links_only = marker.links_only, ephemeral = marker.ephemeral, "storage mode marker initialized" ); Ok(())}
/// infer the layout of a database created before `STORAGE_MODE_KEY`. scanning/// every head is intentional: accepting a mixed keyspace would make whichever/// value happens to sort first define the interpretation of all other values.fn infer_existing_links_only(db: &Db) -> Result<Option<bool>> { let mut inferred = None; for guard in db.indexer.heads.iter() { let value = guard.value().into_diagnostic()?; let links_only = crate::db::is_cid_record_value(&value); if inferred.is_some_and(|previous| previous != links_only) { miette::bail!( "database contains mixed record body and cid-only values; refusing to infer only_index_links" ); } inferred = Some(links_only); } Ok(inferred)}
#[cfg(all(test, feature = "indexer"))]mod tests { use super::*; use crate::db::types::{DbAction, DbRkey, DbTid}; use crate::state::AppState; use jacquard_common::types::string::{Did, Tid};
fn did() -> Did { Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap() }
fn config_at(path: &std::path::Path, ephemeral: bool, links_only: bool) -> Config { let mut config = Config::default(); config.database_path = path.to_path_buf(); config.ephemeral = ephemeral; config.only_index_links = links_only; config }
fn write_record(state: &AppState, rkey: &str, body: &[u8]) { let mut txn = crate::db::Txn::new(&state.db); let did = did(); let rev = DbTid::from(&Tid::now_0()); let mut record_txn = txn.records(state, &rev, &did).unwrap(); let block = crate::car::CarBlock::from_body(bytes::Bytes::copy_from_slice(body)); record_txn .put_record( "app.bsky.feed.post", &DbRkey::new(rkey), &block, DbAction::Create, ) .unwrap(); txn.commit().unwrap(); }
#[test] fn storage_mode_marker_persisted_golden_fixture() { let marker = StorageModeMarker { links_only: false, ephemeral: true, }; let encoded = rmp_serde::to_vec_named(&marker).unwrap(); assert_eq!( crate::util::hex::encode(&encoded), "82aa6c696e6b735f6f6e6c79c2a9657068656d6572616cc3" ); assert_eq!( rmp_serde::from_slice::<StorageModeMarker>(&encoded).unwrap(), marker ); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn ephemeral_mode_is_immutable() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); }
let err = match AppState::new(&config_at(tmp.path(), true, false)) { Ok(_) => panic!("permanent to ephemeral flip must be rejected"), Err(err) => err, }; let msg = format!("{err}"); assert!(msg.contains("ephemeral=false"), "{msg}"); assert!(msg.contains("HYDRANT_EPHEMERAL"), "{msg}"); assert!(msg.contains("offline migration"), "{msg}");
AppState::new(&config_at(tmp.path(), false, false)).unwrap(); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn ephemeral_to_permanent_flip_is_rejected() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); }
let err = match AppState::new(&config_at(tmp.path(), false, false)) { Ok(_) => panic!("ephemeral to permanent flip must be rejected"), Err(err) => err, }; assert!(format!("{err}").contains("ephemeral=true"));
AppState::new(&config_at(tmp.path(), true, false)).unwrap(); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn empty_database_still_records_its_ephemeral_identity() { let tmp = tempfile::tempdir().unwrap(); AppState::new(&config_at(tmp.path(), true, false)).unwrap();
assert!(AppState::new(&config_at(tmp.path(), false, false)).is_err()); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn markerless_inline_events_prevent_wrong_permanent_adoption() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); assert!(state.db.indexer.heads.is_empty().unwrap()); state.db.filter.remove(STORAGE_MODE_KEY).unwrap(); state.db.persist().unwrap(); }
let err = match AppState::new(&config_at(tmp.path(), false, false)) { Ok(_) => panic!("inline v9 events must identify ephemeral mode"), Err(err) => err, }; assert!(format!("{err}").contains("imply ephemeral=true")); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn markerless_pointer_events_prevent_wrong_ephemeral_adoption() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); state.db.filter.remove(STORAGE_MODE_KEY).unwrap(); state.db.persist().unwrap(); }
let err = match AppState::new(&config_at(tmp.path(), true, false)) { Ok(_) => panic!("pointer v9 events must identify permanent mode"), Err(err) => err, }; assert!(format!("{err}").contains("imply ephemeral=false")); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn marked_ephemeral_database_rejects_unshipped_record_heads() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), true, false)).unwrap(); let key = crate::db::keys::record_key( &did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"), ); state.db.indexer.heads.insert(key, b"stale head").unwrap(); state.db.persist().unwrap(); }
let err = match AppState::new(&config_at(tmp.path(), true, false)) { Ok(_) => panic!("event-only ephemeral mode must reject persisted heads"), Err(err) => err, }; assert!(format!("{err}").contains("unsupported pre-release")); }
#[test] #[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))] fn rejected_layout_open_cannot_arm_history_retention() { let tmp = tempfile::tempdir().unwrap(); let record_key = crate::db::keys::record_key( &did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"), ); let history_key = crate::db::keys::history_key(&record_key, &DbTid::new_from_bytes(1_u64.to_be_bytes()));
{ let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap(); let mut batch = state.db.inner.batch(); state .db .indexer .stage_history(&mut batch, &history_key, b"must survive".as_slice()); batch.commit().unwrap(); state.db.indexer.history.rotate_memtable_and_wait().unwrap(); state.db.persist().unwrap(); }
let mut wrong = config_at(tmp.path(), false, true); wrong.history_ttl = Some(std::time::Duration::from_secs(1)); assert!(AppState::new(&wrong).is_err());
let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap(); state.db.indexer.history.major_compact().unwrap(); assert_eq!( state .db .indexer .history .get(history_key) .unwrap() .as_deref(), Some(b"must survive".as_slice()) ); }
#[test] fn links_only_flip_is_rejected_actionably() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), false, true)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); }
let err = match AppState::new(&config_at(tmp.path(), false, false)) { Ok(_) => panic!("links-only flip must be rejected"), Err(err) => err, }; let msg = format!("{err}"); assert!( msg.contains("not convertible in place"), "unexpected error: {msg}" ); assert!(msg.contains("only_index_links"), "unexpected error: {msg}"); }
#[test] fn missing_marker_does_not_bless_a_links_only_database_as_body_storage() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), false, true)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); state.db.filter.remove(STORAGE_MODE_KEY).unwrap(); state.db.persist().unwrap(); }
let err = match AppState::new(&config_at(tmp.path(), false, false)) { Ok(_) => panic!("pre-marker links-only mismatch must be rejected"), Err(err) => err, }; let msg = format!("{err}"); assert!(msg.contains("imply only_index_links=true"), "{msg}"); }
#[test] fn missing_marker_does_not_bless_body_storage_as_links_only() { let tmp = tempfile::tempdir().unwrap(); { let state = AppState::new(&config_at(tmp.path(), false, false)).unwrap(); write_record(&state, "3kzbif5moe22m", b"v1"); state.db.filter.remove(STORAGE_MODE_KEY).unwrap(); state.db.persist().unwrap(); }
let err = match AppState::new(&config_at(tmp.path(), false, true)) { Ok(_) => panic!("pre-marker body mismatch must be rejected"), Err(err) => err, }; let msg = format!("{err}"); assert!(msg.contains("imply only_index_links=false"), "{msg}"); }}