Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106use std::ops::Bound;
use fjall::{Keyspace, OwnedWriteBatch, Readable};use miette::{Context, IntoDiagnostic, Result};
use crate::db::Db;use crate::db::keys::{SEP, VERSIONING_KEY};
#[cfg(feature = "indexer")]const LEGACY_BLOCKS_KEYSPACE: &str = "blocks";
/// open the pre-v10 CAS only when it actually exists. keeping this out of the/// live keyspace registry means a fresh database never recreates the retired/// layout merely so a migration can observe its absence.#[cfg(feature = "indexer")]fn legacy_blocks(db: &Db) -> Result<Option<Keyspace>> { if !db.inner.keyspace_exists(LEGACY_BLOCKS_KEYSPACE) { return Ok(None); } db.inner .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default) .into_diagnostic() .map(Some)}
#[cfg(feature = "indexer")]fn legacy_block(db: &Db, key: &[u8]) -> Result<Option<fjall::Slice>> { legacy_blocks(db)? .map(|blocks| blocks.get(key).into_diagnostic()) .transpose() .map(Option::flatten)}
#[cfg(all(test, feature = "indexer"))]fn legacy_blocks_for_test(db: &Db) -> Result<Keyspace> { db.inner .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default) .into_diagnostic()}
/// the pre-v12 record store. v12's schema drops `records` from the registry,/// but its on-disk tables persist until the `recompress_record_heads`/// migration copies them into `heads` and deletes the keyspace.#[cfg(feature = "indexer")]const LEGACY_RECORDS_KEYSPACE: &str = "records";
#[cfg(all(test, feature = "indexer"))]fn legacy_records_for_test(db: &Db) -> Result<Keyspace> { db.inner .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default) .into_diagnostic()}
#[cfg(feature = "indexer")]fn legacy_records(db: &Db) -> Result<Option<Keyspace>> { if !db.inner.keyspace_exists(LEGACY_RECORDS_KEYSPACE) { return Ok(None); } db.inner .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default) .into_diagnostic() .map(Some)}
mod v1;mod v10;mod v11;mod v2;mod v3;mod v4;mod v5;mod v6;mod v7;mod v8;mod v9;
/// a migration whose entire write set is buffered into one batch./// returns `Ok(true)` if applied, or `Ok(false)` if skipped due to missing keyspaces in this build.type AtomicFn = fn(&Db, &mut OwnedWriteBatch) -> Result<bool>;
/// transform one scanned entry. staged writes may target any keyspace./// returns the number of output bytes staged into the write batch for this entry./// visit one scanned entry, staging its rewrite into the chunk batch, and/// return the staged byte count for the chunk budget.////// the third argument is the keyspace being scanned: in-place passes stage/// into it, so a pass keeps writing the same store it reads even when that/// store is a legacy keyspace the registry no longer knows.type VisitFn = fn(&Db, &mut OwnedWriteBatch, &Keyspace, &[u8], &[u8]) -> Result<usize>;
/// destructive or otherwise non-batched work that runs only after every/// chunked pass is durable. returning false means this build does not contain/// the keyspaces needed by the migration.type FinalizeFn = fn(&Db) -> Result<bool>;
/// how a migration is applied.enum Migration { /// the whole write set is buffered into one batch, committed atomically with /// the version bump. /// /// a batch is held entirely in memory, so this is only for migrations whose /// write set is small and bounded. it is *not* bounded by how much the /// migration reads: streaming a large keyspace to accumulate a handful of /// counters is fine, staging one write per entry of a large keyspace is not. Atomic(AtomicFn), /// ordered resumable passes, each committed in chunks with bounded memory. /// /// the only way to rewrite a keyspace larger than memory. passes run in /// order, and each runs to completion before the next starts. Chunked { passes: &'static [Pass], finalize: Option<FinalizeFn>, },}
/// per-chunk limits. both bound peak batch memory; whichever is hit first ends/// the chunk. they also bound how much work a crash re-does.#[derive(Clone, Copy)]struct ChunkBudget { /// entries scanned per chunk. entries: usize, /// staged output bytes per chunk. /// /// bounds peak write batch memory when converting small input keys/references /// into large staged bodies (e.g. v10 record body expansion). bytes: usize,}
impl ChunkBudget { /// what a pass should use unless it has a measured reason not to. #[allow(dead_code)] const DEFAULT: Self = Self { entries: 100_000, bytes: 64 * 1024 * 1024, };}
/// one resumable key-ordered scan over a single keyspace.////// the framework owns iteration, chunk commits, and the resume cursor; a pass/// supplies only [`Pass::visit`].////// # visit must be idempotent////// the resume cursor advances only on a committed chunk, so a crash re-visits/// every entry of the in-flight chunk.////// more subtly, a pass that writes into the keyspace it scans can observe its/// own output. each chunk reads a fresh snapshot, so output that sorts *after*/// the resume cursor is scanned again in a later chunk. `visit` must therefore/// be safe to apply twice, and re-visiting its own output must reach a fixpoint/// — normally by making the migrated form recognizable and doing nothing when it/// is seen. a transform whose output sorts before its input (as key-shortening/// rewrites usually do) is never rescanned, but that is a property of the/// encoding, not a guarantee of the framework.struct Pass { /// stable cursor identity. THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE. name: &'static str, /// on-disk name of the keyspace to scan. /// /// the registry resolves names from this build first; a name it no longer /// knows falls back to opening an existing on-disk keyspace, so a pass can /// still finish migrating data out of a store a later schema dropped. /// /// a pass whose keyspace exists nowhere is skipped, so a migration can /// name mode-specific keyspaces unconditionally — unless /// `optional_on_absent` is set, in which case absence means the database /// never had the source layout and the pass completes as a no-op. /// /// a pass scanning `counts` also sees its own resume cursor, which sorts /// between the `k|` and `r|` prefixes; `visit` must ignore keys under /// [`MIGRATION_CURSOR_PREFIX`]. scan: &'static str, visit: VisitFn, budget: ChunkBudget, optional_on_absent: bool,}
/// ordered list of schema migrations.////// invariants:/// - migration `vN` upgrades on-disk data from schema version `N-1` to `N`/// - once a migration ships, its input/output wire types are frozen/// - never "fix up" an older migration to match a newer live type; add a new/// schema version and a new migration instead/// - if a stored type changes shape, define a new versioned type in/// `src/types.rs`, export the newest one for live code, and have the new/// migration explicitly deserialize the previous version and write the new one/// - a migration that stages one write per entry of a high-cardinality keyspace/// must be [`Migration::Chunked`]; [`Migration::Atomic`] would buffer the/// whole rewrite in memory////// example: if `RepoState` changes again after `types::v7::RepoState`, add a new/// `types::v8::RepoState`, keep `v7` frozen, and add a migration that rewrites/// `v7 -> v8`./// ordered list of migrations. migration at index `i` upgrades the schema from version `i` to `i+1`.const MIGRATIONS: &[(&str, Migration)] = &[ ( "stable_firehose_cursors", Migration::Atomic(v1::stable_firehose_cursors), ), ( "repo_state_root_commit", Migration::Atomic(v2::repo_state_root_commit), ), ( "firehose_source_is_pds", Migration::Atomic(v3::firehose_source_is_pds), ), ( "repo_state_active", Migration::Atomic(v4::repo_state_active), ), ("pds_meta_layout", Migration::Atomic(v5::pds_meta_layout)), ( "rebuild_pds_account_counts", Migration::Atomic(v6::rebuild_pds_account_counts), ), ( "repo_state_event_clocks", Migration::Atomic(v7::repo_state_event_clocks), ), ( "rebuild_lifecycle_counts", Migration::Atomic(v8::rebuild_lifecycle_counts), ), ( "migrate_excludes_and_pds_keys", Migration::Atomic(v9::migrate_v9), ), ( "inline_record_bodies", Migration::Chunked { passes: v10::PASSES, finalize: Some(v10::retire_legacy_stores), }, ), ( "canonicalize_pds_status_hosts", Migration::Chunked { passes: v11::PASSES, finalize: None, }, ),];
pub(crate) const LATEST_VERSION: u64 = MIGRATIONS.len() as u64;
/// resume cursors for chunked passes, in the `counts` keyspace:/// `mig|{version}|{pass}` -> last committed key.////// the `counts` compaction filter only drops `r|` and old `d|` entries, so these/// survive compaction.const MIGRATION_CURSOR_PREFIX: &[u8] = b"mig|";
/// applied markers for migrations, in the `counts` keyspace:/// `mig_applied|{name}` -> `b"1"`.const MIGRATION_APPLIED_PREFIX: &[u8] = b"mig_applied|";
fn pass_cursor_key(version: usize, pass: &str) -> Vec<u8> { let mut key = Vec::with_capacity(MIGRATION_CURSOR_PREFIX.len() + 8 + 1 + pass.len()); key.extend_from_slice(MIGRATION_CURSOR_PREFIX); key.extend_from_slice(&(version as u64).to_be_bytes()); key.push(SEP); key.extend_from_slice(pass.as_bytes()); key}
fn migration_applied_key(name: &str) -> Vec<u8> { let mut key = Vec::with_capacity(MIGRATION_APPLIED_PREFIX.len() + name.len()); key.extend_from_slice(MIGRATION_APPLIED_PREFIX); key.extend_from_slice(name.as_bytes()); key}
/// test helper: rewind the db to `version` as if later migrations never ran./// opening a fresh test db already runs (and marks) every migration, so/// simulating an old database means clearing the applied markers too.#[cfg(test)]pub(crate) fn rewind_version_for_test(db: &Db, version: u64) -> Result<()> { let mut batch = db.inner.batch(); for (name, _) in MIGRATIONS { batch.remove(&db.counts, migration_applied_key(name)); } batch.insert(&db.counts, VERSIONING_KEY, version.to_be_bytes()); batch.commit().into_diagnostic()}
fn is_migration_applied(db: &Db, index: usize, name: &str, stored_version: u64) -> Result<bool> { let key = migration_applied_key(name); if db.counts.contains_key(&key).into_diagnostic()? { return Ok(true); }
// Legacy versions predate per-migration applied keys. Their global version // proves ordinary migrations ran, while feature-gated migrations need a // cheap migration-specific proof: reduced-feature builds advanced the old // version even when they skipped the work. if (index as u64) < stored_version { return match name { "rebuild_lifecycle_counts" => v8::legacy_migration_applied(db), // The old v9 marker is trusted; v10 safely and boundedly finishes // any work a reduced-feature open skipped. "migrate_excludes_and_pds_keys" => Ok(true), // no intermediate v10 layout shipped. accepting a v10 global // version without this migration's marker would silently bless a // branch-local draft as the final schema. "inline_record_bodies" => { miette::bail!("unsupported pre-release v10 database; rebuild it from v9") } _ => Ok(true), }; }
Ok(false)}
fn read_version(db: &Db) -> Result<u64> { db.counts .get(VERSIONING_KEY) .into_diagnostic()? .map(|r| { r.as_ref() .try_into() .into_diagnostic() .wrap_err("db version key expected to be 8 bytes") .map(u64::from_be_bytes) }) .transpose() .map(|v| v.unwrap_or(0))}
struct ChunkOutcome { /// entries visited and committed. seen: usize, /// the scan reached the end of the keyspace. exhausted: bool,}
/// visit one chunk of `pass`, committing the transformed entries and the/// advanced resume cursor as one atomic batch.fn run_chunk(db: &Db, cursor_key: &[u8], ks: &Keyspace, pass: &Pass) -> Result<ChunkOutcome> { let resume = db.counts.get(cursor_key).into_diagnostic()?; let start = resume .as_ref() .map_or(Bound::Unbounded, |k| Bound::Excluded(k.to_vec()));
let mut batch = db.inner.batch(); let mut seen = 0usize; let mut bytes = 0usize; let mut last = None; let mut exhausted = true;
{ // a fresh snapshot per chunk keeps the scan consistent while the batch is // staged, without pinning superseded versions for the whole pass. let snapshot = db.inner.snapshot(); for guard in snapshot.range(ks, (start, Bound::<Vec<u8>>::Unbounded)) { // checked before visiting, so breaking here proves an entry remains. // a chunk always visits at least one entry, so neither a degenerate // budget nor an entry larger than the byte budget can stall the pass // or make an unfinished scan look exhausted. if seen > 0 && (seen >= pass.budget.entries || bytes >= pass.budget.bytes) { exhausted = false; break; } let (key, value) = guard.into_inner().into_diagnostic()?; let staged_bytes = (pass.visit)(db, &mut batch, ks, &key, &value) .wrap_err_with(|| format!("migration pass {} failed on an entry", pass.name))?; bytes += staged_bytes; last = Some(key); seen += 1; } }
let Some(last) = last else { // nothing left to do; leave the cursor for `run` to clear with the // version bump return Ok(ChunkOutcome { seen: 0, exhausted: true, }); };
batch.insert(&db.counts, cursor_key, last); batch.commit().into_diagnostic()?;
Ok(ChunkOutcome { seen, exhausted })}
/// run one pass to completion, resuming from its stored cursor if present.fn run_pass(db: &Db, version: usize, pass: &Pass) -> Result<bool> { let ks = match db.keyspace_by_name(pass.scan) { Some(ks) => Some(ks), // a keyspace dropped from the registry by a later schema may still // exist on disk; open it so the pass can finish migrating its data None if db.inner.keyspace_exists(pass.scan) => Some( db.inner .keyspace(pass.scan, Default::default) .into_diagnostic()?, ), None => None, }; let Some(ks) = ks else { if pass.optional_on_absent { // the source layout never existed on this database, so there is // genuinely nothing to migrate tracing::debug!( "db: migration pass {} found no {} keyspace, nothing to migrate", pass.name, pass.scan ); return Ok(true); } // keyspace is not compiled into this build, so there is nothing to migrate tracing::debug!( "db: migration pass {} skipped, keyspace {} not in this build", pass.name, pass.scan ); return Ok(false); }; let cursor_key = pass_cursor_key(version, pass.name);
if db.counts.contains_key(&cursor_key).into_diagnostic()? { tracing::info!("db: resuming migration pass {} from its cursor", pass.name); }
let mut total: u64 = 0; loop { let outcome = run_chunk(db, &cursor_key, &ks, pass)?; total += outcome.seen as u64; if outcome.seen > 0 { tracing::info!( "db: migration pass {} committed {total} entries so far", pass.name ); } if outcome.exhausted { break; } }
tracing::info!( "db: migration pass {} complete ({total} entries)", pass.name ); Ok(true)}
/// run all pending database migrations in order.////// an atomic migration commits its write set and applied marker together. a/// chunked migration commits each chunk with its resume cursor, durably runs/// any finalizer, then commits cursor cleanup with the applied marker. the/// global version is derived from the contiguous markers afterward. a crash/// before the marker therefore resumes from the durable pass cursors.////// safe to run on a fresh database (all migrations are no-ops when no relevant/// data exists). called during [`Db::open`].pub(super) fn run(db: &Db) -> Result<()> { let stored_version = read_version(db)?; if stored_version > LATEST_VERSION { miette::bail!( "database schema version {stored_version} is newer than supported version {LATEST_VERSION}" ); } for (i, (name, migration)) in MIGRATIONS.iter().enumerate() { if is_migration_applied(db, i, name, stored_version)? { let key = migration_applied_key(name); if !db.counts.contains_key(&key).into_diagnostic()? { db.counts.insert(key, b"1").into_diagnostic()?; tracing::info!("db: adopted legacy applied marker for {name}"); } tracing::debug!("db: migration {name} already applied"); continue; }
tracing::info!("db: running migration {name} (v{i} -> v{})", i + 1); match migration { Migration::Atomic(f) => { let mut batch = db.inner.batch(); if f(db, &mut batch)? { batch.insert(&db.counts, migration_applied_key(name), b"1"); batch.commit().into_diagnostic()?; tracing::info!("db: migration {name} complete"); } else { tracing::info!( "db: migration {name} skipped (required keyspace not in this build)" ); } } Migration::Chunked { passes, finalize } => { let mut all_passed = !(passes.is_empty() && finalize.is_none()); for pass in *passes { if !run_pass(db, i, pass)? { all_passed = false; break; } } if all_passed && let Some(finalize) = finalize && !finalize(db)? { all_passed = false; } if all_passed { let mut batch = db.inner.batch(); for pass in *passes { batch.remove(&db.counts, pass_cursor_key(i, pass.name)); } batch.insert(&db.counts, migration_applied_key(name), b"1"); batch.commit().into_diagnostic()?; tracing::info!("db: migration {name} complete"); } else { tracing::info!( "db: migration {name} skipped (required keyspace not in this build)" ); } } } }
// Advance VERSIONING_KEY to highest contiguous applied migration version let mut max_contiguous_version = 0u64; for (i, (name, _)) in MIGRATIONS.iter().enumerate() { if is_migration_applied(db, i, name, stored_version)? { max_contiguous_version = (i + 1) as u64; } else { break; } } if max_contiguous_version != stored_version { let mut batch = db.inner.batch(); batch.insert( &db.counts, VERSIONING_KEY, max_contiguous_version.to_be_bytes(), ); batch.commit().into_diagnostic()?; }
Ok(())}
#[cfg(test)]mod tests { use super::*;
const TEST_PREFIX: &[u8] = b"chunktest|"; const MIGRATED_PREFIX: &[u8] = b"chunkdone|";
/// rewrites `chunktest|{n}` to `chunkdone|{n}`, keeping the value. /// /// idempotent: already-migrated keys carry the new prefix and are ignored. /// note the output sorts *before* the input (`chunkd` < `chunkt`), which is /// the usual shape for a key rewrite and means output is never rescanned. fn rewrite_prefix( db: &Db, batch: &mut OwnedWriteBatch, scanned: &Keyspace, key: &[u8], value: &[u8], ) -> Result<usize> { let Some(suffix) = key.strip_prefix(TEST_PREFIX) else { return Ok(0); }; let mut new_key = MIGRATED_PREFIX.to_vec(); new_key.extend_from_slice(suffix); let _ = db; let staged_bytes = new_key.len() + value.len(); batch.insert(scanned, new_key, value); batch.remove(scanned, key); Ok(staged_bytes) }
fn test_pass(entries: usize) -> Pass { Pass { name: "rewrite_prefix", scan: "cursors", visit: rewrite_prefix, budget: ChunkBudget { entries, bytes: usize::MAX, }, optional_on_absent: false, } }
fn open_db() -> Result<(tempfile::TempDir, Db)> { let tmp = tempfile::tempdir().into_diagnostic()?; let cfg = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let db = Db::open(&cfg)?; Ok((tmp, db)) }
fn seed(db: &Db, count: usize) -> Result<()> { let mut batch = db.inner.batch(); for i in 0..count { let mut key = TEST_PREFIX.to_vec(); key.extend_from_slice(&(i as u64).to_be_bytes()); batch.insert(&db.cursors, key, (i as u64).to_be_bytes()); } batch.commit().into_diagnostic()?; Ok(()) }
fn migrated(db: &Db) -> Result<Vec<(u64, u64)>> { db.cursors .prefix(MIGRATED_PREFIX) .map(|guard| { let (k, v) = guard.into_inner().into_diagnostic()?; let n = u64::from_be_bytes(k[MIGRATED_PREFIX.len()..].try_into().into_diagnostic()?); let v = u64::from_be_bytes(v.as_ref().try_into().into_diagnostic()?); Ok((n, v)) }) .collect() }
fn expected(count: usize) -> Vec<(u64, u64)> { (0..count as u64).map(|i| (i, i)).collect() }
#[cfg(feature = "indexer")] #[test] fn v0_database_migrates_all_early_representation_boundaries() -> Result<()> { use crate::config::Config; use crate::db::{ keys, types::{DbTid, DidKey}, }; use crate::types::v2::RepoStatus as V2RepoStatus; use jacquard_common::CowStr; use jacquard_common::types::{ string::{Did, Handle}, tid::Tid, };
let tmp = tempfile::tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let did = Did::new_static("did:web:migration.test").into_diagnostic()?; let repo_key = keys::repo_key(&did); let old_cursor_url = url::Url::parse("wss://relay.example:7443/x").into_diagnostic()?; let old_cursor_key = [ keys::FIREHOSE_CURSOR_PREFIX, old_cursor_url.as_str().as_bytes(), ] .concat(); let new_cursor_key = keys::v1::firehose_cursor_key_from_url(&old_cursor_url); let relay_source = "wss://bsky.network"; let pds_source = "wss://pds.example";
{ let db = Db::open(&cfg)?; let old_state = v2::OldRepoState { status: V2RepoStatus::Error("stinkpot".into()), rev: Some(DbTid::from(&Tid::new("3jzfcijpj2z2a").unwrap())), data: Some(jacquard_repo::mst::util::compute_cid(b"root")?), last_message_time: Some(-1), last_updated_at: 42, tracked: true, index_id: 7, signing_key: Some(DidKey(std::borrow::Cow::Borrowed(&[0xed, 1, 2, 3]))), pds: Some(CowStr::Borrowed("https://pds.example")), handle: Some(Handle::new_static("alice.test").into_diagnostic()?), }; let mut batch = db.inner.batch(); batch.insert( &db.repos, &repo_key, rmp_serde::to_vec(&old_state).into_diagnostic()?, ); batch.insert(&db.cursors, &old_cursor_key, 42i64.to_be_bytes()); batch.insert(&db.crawler, keys::firehose_source_key(relay_source), []); batch.insert(&db.crawler, keys::firehose_source_key(pds_source), []); batch.insert(&db.filter, b"pt|pds.example", b"trusted"); batch.insert(&db.filter, b"pb|banned.example", []); batch.insert( &db.counts, keys::count_keyspace_key("p|stale.example"), 99u64.to_be_bytes(), ); batch.commit().into_diagnostic()?; rewind_version_for_test(&db, 0)?; db.persist()?; }
let db = Db::open(&cfg)?; assert!(!db.cursors.contains_key(&old_cursor_key).into_diagnostic()?); let cursor = db .cursors .get(&new_cursor_key) .into_diagnostic()? .expect("v1 cursor"); assert_eq!( i64::from_be_bytes(cursor.as_ref().try_into().into_diagnostic()?), 42 );
for (source, is_pds) in [(relay_source, false), (pds_source, true)] { let encoded = db .crawler .get(keys::firehose_source_key(source)) .into_diagnostic()? .expect("v3 source metadata"); let metadata: crate::db::FirehoseSourceMeta = rmp_serde::from_slice(&encoded).into_diagnostic()?; assert_eq!(metadata.is_pds, is_pds); }
let state_bytes = db .repos .get(&repo_key) .into_diagnostic()? .expect("migrated repo state"); let state = crate::db::deser_repo_state(&state_bytes)?; assert!(state.active); assert!(matches!( &state.status, crate::types::RepoStatus::Error(error) if error == "stinkpot" )); let root = state.root.as_ref().expect("migrated commit"); assert_eq!(root.version, -1); assert_eq!(root.rev, DbTid::from(&Tid::new("3jzfcijpj2z2a").unwrap())); assert_eq!(root.data, jacquard_repo::mst::util::compute_cid(b"root")?); assert!(root.prev.is_none()); assert!(root.sig.is_empty()); assert_eq!(state.last_message_time, Some(-1)); assert_eq!(state.last_identity_time, None); assert_eq!(state.last_account_time, None); assert_eq!(state.last_updated_at, 42); assert_eq!( state.signing_key.as_ref().map(|key| key.0.as_ref()), Some([0xed, 1, 2, 3].as_slice()) ); assert_eq!(state.pds.as_deref(), Some("https://pds.example")); assert_eq!( state.handle.as_ref().map(Handle::as_str), Some("alice.test") ); let metadata = crate::db::deser_repo_meta( &db.repo_metadata .get(keys::repo_metadata_key(&did)) .into_diagnostic()? .expect("v4 repo metadata"), )?; assert!(metadata.tracked); assert_eq!(metadata.index_id, 7);
assert_eq!( db.filter .get(crate::db::pds_meta::pds_tier_key("pds.example")) .into_diagnostic()? .as_deref(), Some(b"trusted".as_slice()) ); assert!( !db.filter .contains_key(crate::db::pds_meta::pds_status_key("banned.example")) .into_diagnostic()? ); assert!( db.filter .contains_key(crate::db::pds_meta::pds_ban_key("banned.example")) .into_diagnostic()? ); assert!( !db.counts .contains_key(keys::count_keyspace_key("p|stale.example")) .into_diagnostic()? ); let account_count = db .counts .get(keys::count_keyspace_key("p|pds.example")) .into_diagnostic()? .expect("v6 PDS account count"); assert_eq!( u64::from_be_bytes(account_count.as_ref().try_into().into_diagnostic()?), 1 );
let version = db .counts .get(VERSIONING_KEY) .into_diagnostic()? .expect("schema version"); assert_eq!( u64::from_be_bytes(version.as_ref().try_into().into_diagnostic()?), LATEST_VERSION ); Ok(()) }
#[test] fn chunked_pass_rewrites_every_entry_across_many_chunks() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 50)?;
let pass = test_pass(7); run_pass(&db, 99, &pass)?;
assert_eq!(migrated(&db)?, expected(50)); assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 0); Ok(()) }
#[test] fn interrupted_pass_resumes_and_matches_an_uninterrupted_run() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 50)?;
let pass = test_pass(7); let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
// one chunk only, standing in for a crash mid-pass let first = run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(first.seen, 7); assert!(!first.exhausted); assert_eq!(migrated(&db)?, expected(7)); assert!(db.counts.contains_key(&cursor_key).into_diagnostic()?);
// reopening resumes from the cursor rather than restarting run_pass(&db, 99, &pass)?;
assert_eq!(migrated(&db)?, expected(50)); assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 0); Ok(()) }
#[test] fn completed_pass_is_a_cheap_no_op_when_rerun() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 20)?;
let pass = test_pass(7); run_pass(&db, 99, &pass)?; let after_first = migrated(&db)?;
// the cursor survives until the version bump, so a rerun scans nothing let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace"); let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(outcome.seen, 0); assert!(outcome.exhausted);
assert_eq!(migrated(&db)?, after_first); Ok(()) }
#[test] fn visiting_migrated_entries_again_is_a_fixpoint() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 20)?;
let pass = test_pass(7); run_pass(&db, 99, &pass)?; let after_first = migrated(&db)?;
// drop the cursor to force a full rescan over already-migrated entries, // the worst case the idempotency contract has to survive db.counts .remove(pass_cursor_key(99, pass.name)) .into_diagnostic()?; run_pass(&db, 99, &pass)?;
assert_eq!(migrated(&db)?, after_first); Ok(()) }
/// fails partway through a chunk, standing in for a crash mid-chunk. fn fail_partway( db: &Db, batch: &mut OwnedWriteBatch, scanned: &Keyspace, key: &[u8], value: &[u8], ) -> Result<usize> { let Some(suffix) = key.strip_prefix(TEST_PREFIX) else { return Ok(0); }; let n = u64::from_be_bytes(suffix.try_into().into_diagnostic()?); if n == 4 { miette::bail!("simulated failure at entry 4"); } rewrite_prefix(db, batch, scanned, key, value) } #[test] fn a_chunk_that_fails_partway_commits_nothing() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 50)?;
let pass = Pass { visit: fail_partway, ..test_pass(7) }; let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
assert!(run_chunk(&db, &cursor_key, &ks, &pass).is_err());
// entries 0..=3 were staged before the failure, but the batch is dropped // uncommitted, so neither they nor the cursor survive assert_eq!(migrated(&db)?, vec![]); assert!(!db.counts.contains_key(&cursor_key).into_diagnostic()?); assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 50);
// retrying with a working visit starts over from the beginning run_pass(&db, 99, &test_pass(7))?; assert_eq!(migrated(&db)?, expected(50)); Ok(()) }
#[test] fn byte_budget_ends_a_chunk_before_the_entry_budget() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 50)?;
// each entry is 26 scanned bytes (10-byte prefix + 8-byte key + 8-byte // value), and the budget is checked before visiting, so a 60-byte budget // admits 3 entries (26, 52, 78) before it trips let pass = Pass { budget: ChunkBudget { entries: usize::MAX, bytes: 60, }, ..test_pass(0) }; let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(outcome.seen, 3); assert!(!outcome.exhausted);
run_pass(&db, 99, &pass)?; assert_eq!(migrated(&db)?, expected(50)); Ok(()) }
#[test] fn a_degenerate_budget_still_makes_progress() -> Result<()> { let (_tmp, db) = open_db()?; seed(&db, 5)?;
// a zero budget would otherwise trip before visiting anything, leaving // the chunk with no cursor to advance and reporting itself exhausted let pass = Pass { budget: ChunkBudget { entries: 0, bytes: 0, }, ..test_pass(0) }; let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(outcome.seen, 1); assert!(!outcome.exhausted);
run_pass(&db, 99, &pass)?; assert_eq!(migrated(&db)?, expected(5)); Ok(()) }
#[test] fn pass_on_a_keyspace_missing_from_this_build_is_skipped() -> Result<()> { let (_tmp, db) = open_db()?;
let pass = Pass { scan: "not_a_keyspace", ..test_pass(7) }; run_pass(&db, 99, &pass)?;
let cursor_key = pass_cursor_key(99, pass.name); assert!(!db.counts.contains_key(&cursor_key).into_diagnostic()?); Ok(()) }
#[test] fn cursor_keys_are_distinct_per_version_and_pass() { assert_ne!(pass_cursor_key(1, "a"), pass_cursor_key(2, "a")); assert_ne!(pass_cursor_key(1, "a"), pass_cursor_key(1, "b")); assert!(pass_cursor_key(1, "a").starts_with(MIGRATION_CURSOR_PREFIX)); }
#[test] fn staged_byte_budget_bounds_chunk_write_size() -> Result<()> { let (_tmp, db) = open_db()?; let mut batch = db.inner.batch(); // Insert small input keys/values that expand into large staged outputs for i in 0..10 { let mut key = TEST_PREFIX.to_vec(); key.extend_from_slice(&(i as u64).to_be_bytes()); batch.insert(&db.cursors, key, vec![0u8; 1000]); // 1 KB payload } batch.commit().into_diagnostic()?;
// Set byte budget to 2500 bytes. Each output staged is ~1018 bytes. // First entry: 1018 bytes. Second entry: 2036 bytes. Third entry: 3054 bytes (>= 2500). // 4th check trips because 3054 >= 2500. So outcome.seen == 3. let pass = Pass { budget: ChunkBudget { entries: usize::MAX, bytes: 2500, }, ..test_pass(0) }; let cursor_key = pass_cursor_key(99, pass.name); let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?; assert_eq!(outcome.seen, 3); assert!(!outcome.exhausted); Ok(()) }
#[test] fn applied_migration_keys_track_per_migration_status() -> Result<()> { let (_tmp, db) = open_db()?; assert!( db.counts .contains_key(&migration_applied_key("stable_firehose_cursors")) .into_diagnostic()? ); Ok(()) }
#[test] fn branch_local_v10_without_the_final_marker_is_rejected() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let cfg = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; { let db = Db::open(&cfg)?; rewind_version_for_test(&db, 10)?; db.persist()?; }
let err = match Db::open(&cfg) { Ok(_) => panic!("an unmarked branch-local v10 must not be adopted"), Err(err) => err, }; assert!(format!("{err:?}").contains("unsupported pre-release v10")); Ok(()) }
#[cfg(feature = "indexer")] #[test] fn legacy_marker_adoption_does_not_rerun_lifecycle_rebuild() -> Result<()> { use jacquard_common::types::string::Did;
let tmp = tempfile::tempdir().into_diagnostic()?; let cfg = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let repo = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?; let stale = Did::new_static("did:plc:yk4q3id7id6p5z3bypvshc64").into_diagnostic()?; let mut stale_gauge = crate::db::keys::COUNT_GAUGE_PREFIX.to_vec(); stale_gauge.extend_from_slice(&crate::db::keys::repo_key(&stale));
{ let db = Db::open(&cfg)?; let mut batch = db.inner.batch(); batch.insert( &db.repos, crate::db::keys::repo_key(&repo), crate::db::ser_repo_state(&crate::types::RepoState::synced())?, ); let mut gauge = crate::db::keys::COUNT_GAUGE_PREFIX.to_vec(); gauge.extend_from_slice(&crate::db::keys::repo_key(&repo)); let synced = crate::db::lifecycle_counts::encode_gauge(crate::types::GaugeState::Synced); batch.insert(&db.counts, gauge, synced); // v8 would clear this stale-but-valid row. retaining it proves the // legacy global version was adopted rather than rebuilt. batch.insert(&db.counts, &stale_gauge, synced); batch.commit().into_diagnostic()?; // v9 is the last schema that existed before this change. rewind_version_for_test(&db, 9)?; db.persist()?; }
let db = Db::open(&cfg)?; assert!(db.counts.contains_key(stale_gauge).into_diagnostic()?); assert!( db.counts .contains_key(migration_applied_key("rebuild_lifecycle_counts")) .into_diagnostic()? ); Ok(()) }}