use 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 v12;
mod v13;
mod v14;
mod v15;
mod v16;
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,
},
),
(
"pending_dids",
Migration::Chunked {
passes: v12::PASSES,
finalize: None,
},
),
(
"recount_pds_account_counts",
Migration::Atomic(v13::recount_pds_account_counts),
),
(
"repo_metadata_v14",
Migration::Atomic(v14::repo_metadata_v14),
),
(
FENCE,
Migration::Atomic(v15::repo_state_identity_resolved_at),
),
(
"repo_metadata_refetch",
Migration::Atomic(v16::repo_metadata_refetch),
),
];
/// from this migration on, the applied markers are the whole truth. a relay-only build skips v8
/// and the indexer passes for good, so a contiguous version stays at 7 there and would never
/// keep an older build out of rows it can't read
const FENCE: &str = "repo_state_identity_resolved_at";
/// the version to store, from which migrations are applied in order. before the fence it's the
/// end of the unbroken run from the start, like it always was. once fenced it's the newest
/// applied one, holes or not, so every migration a build applies keeps the builds before it out
fn schema_version(applied: &[bool], fenced: bool) -> u64 {
let version = match fenced {
true => applied
.iter()
.rposition(|applied| *applied)
.map_or(0, |i| i + 1),
false => applied.iter().take_while(|applied| **applied).count(),
};
version as u64
}
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. 9 is
/// as far forward as this can go, because a v10 database without its marker
/// is rejected on purpose.
#[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,
fenced: bool,
) -> 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. a fenced database had markers
// for everything already, and its version may be past what it applied.
if !fenced && (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}"
);
}
let fenced = db
.counts
.contains_key(migration_applied_key(FENCE))
.into_diagnostic()?;
for (i, (name, migration)) in MIGRATIONS.iter().enumerate() {
if is_migration_applied(db, i, name, stored_version, fenced)? {
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)"
);
}
}
}
}
let applied = MIGRATIONS
.iter()
.enumerate()
.map(|(i, (name, _))| is_migration_applied(db, i, name, stored_version, fenced))
.collect::<Result<Vec<_>>>()?;
let fenced = db
.counts
.contains_key(migration_applied_key(FENCE))
.into_diagnostic()?;
let version = schema_version(&applied, fenced);
if version != stored_version {
let mut batch = db.inner.batch();
batch.insert(&db.counts, VERSIONING_KEY, 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()
}
#[test]
fn stored_schema_version_is_read_big_endian() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let db = Db::open(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
db.counts
.insert(VERSIONING_KEY, [0, 0, 0, 0, 0, 0, 1, 2])
.into_diagnostic()?;
assert_eq!(read_version(&db)?, 258);
db.counts.insert(VERSIONING_KEY, [0; 7]).into_diagnostic()?;
assert!(read_version(&db).is_err());
db.counts.remove(VERSIONING_KEY).into_diagnostic()?;
assert_eq!(read_version(&db)?, 0);
Ok(())
}
#[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(crate::car::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, crate::car::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 the_stored_version_follows_the_markers_once_fenced() {
// a legacy database, everything applied in order up to v12
let legacy = [[true; 12].as_slice(), &[false; 3]].concat();
assert_eq!(schema_version(&legacy, false), 12);
// unfenced holes still stop the version, so a skipped migration runs before anything
// treats the database as past it
assert_eq!(schema_version(&[true, true, false, true], false), 2);
// a relay-only build: v8 and the indexer passes never apply, the fence (v15) does
let mut relay = [true; 15];
for skipped in [7, 9, 10, 11] {
relay[skipped] = false;
}
assert_eq!(schema_version(&relay, false), 7);
assert_eq!(schema_version(&relay, true), 15);
// and a v16 every build runs moves it on, so a v15 build is kept out of its changes
let after = [relay.as_slice(), &[true]].concat();
assert_eq!(schema_version(&after, true), 16);
let not_yet = [relay.as_slice(), &[false]].concat();
assert_eq!(schema_version(¬_yet, true), 15);
}
#[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(())
}
}