//! v10: inline record bodies into the compressed `heads` keyspace, and retire
//! the legacy `records` store and `blocks` CAS.
//!
//! legacy `{DID}|{COL}|{tag}{rkey}` -> cid entries in `records` become
//! `{DID} 00 {COL} 2f {rkey text}` -> body in `heads`, resolved from the
//! blocks CAS. the schema drops `records` from the registry for this: v9
//! persisted "never compress" into its on-disk config (fair when values were
//! cids), and fjall reads keyspace options from disk at open, so bodies could
//! only ever be compressed in a fresh keyspace. the pass scans `records`
//! through the on-disk fallback and stages into a different keyspace than it
//! scans, so chunks never re-observe their own output and re-inserting the
//! same key/value is a fixpoint.
//!
//! the same bounded migration archives otherwise-unresolved legacy permanent
//! pointer bodies before deleting `blocks` and `records`. ephemeral inline
//! events stay inline so their TTL still removes body bytes atomically. no
//! ingestion can run during `Db::open`, so every pointer visible to this
//! migration is necessarily legacy; an event watermark would only model
//! versions that never shipped.
mod events;
#[cfg(feature = "indexer")]
use fjall::OwnedWriteBatch;
#[cfg(feature = "indexer")]
use miette::{IntoDiagnostic, Result, WrapErr};
#[cfg(feature = "indexer")]
use crate::db::keys::{self, indexer::REC_PATH_SEP, indexer::REC_SEP};
#[cfg(feature = "indexer")]
use crate::db::types::{DbRkey, DbTid, TrimmedDid};
#[cfg(feature = "indexer")]
use crate::db::{Db, keys::indexer as idx};
#[cfg(feature = "indexer")]
use super::ChunkBudget;
use super::Pass;
/// Default byte budget per chunk for v10 record migration (64 MiB).
///
/// Bounds staged output payload memory when resolving full record bodies from blocks.
#[cfg(feature = "indexer")]
pub(super) const DEFAULT_RECORD_CHUNK_BYTES: usize = 64 * 1024 * 1024;
pub(super) const PASSES: &[Pass] = &[
#[cfg(feature = "indexer")]
Pass {
name: "finish_v9_filter_layout",
scan: "filter",
visit: rewrite_filter_key,
budget: ChunkBudget::DEFAULT,
optional_on_absent: false,
},
#[cfg(feature = "indexer")]
Pass {
name: "records",
scan: "records",
visit: rewrite_record,
budget: ChunkBudget {
entries: 100_000,
bytes: DEFAULT_RECORD_CHUNK_BYTES,
},
// v12 databases register `heads` instead and never had `records`
optional_on_absent: true,
},
#[cfg(feature = "indexer_stream")]
events::POINTER_EVENT_BODIES,
];
#[cfg(feature = "indexer")]
/// is this key already in the current (post-v10) format?
///
/// decided by the byte after the trimmed DID: NUL now, `|` before. new keys
/// sort before legacy ones within a repo, so a resuming scan never meets
/// them; this is defense against out-of-order surprises only.
fn is_current_key(key: &[u8]) -> bool {
let Some(did_len) = idx::trimmed_did_len(key) else {
return false;
};
key.get(did_len) == Some(&REC_SEP)
}
#[cfg(feature = "indexer")]
fn rewrite_filter_key(
_db: &Db,
batch: &mut OwnedWriteBatch,
scanned: &fjall::Keyspace,
key: &[u8],
value: &[u8],
) -> Result<usize> {
let exclude_prefix = [crate::db::filter::EXCLUDE_PREFIX, crate::db::keys::SEP];
if let Some(raw) = key.strip_prefix(&exclude_prefix)
&& let Ok(did) = std::str::from_utf8(raw)
&& did.starts_with("did:")
{
let did = jacquard_common::types::did::Did::new_owned(did).into_diagnostic()?;
let trimmed = TrimmedDid::from(&did);
let mut new_key = exclude_prefix.to_vec();
trimmed.write_to_vec(&mut new_key);
let staged = new_key.len();
batch.insert(scanned, new_key, []);
batch.remove(scanned, key);
return Ok(staged);
}
if key.starts_with(b"pds|status|") || key.starts_with(b"pds|tier|") {
return Ok(0);
}
let (legacy_suffix, current_prefix) = if key.ends_with(b"|status") {
(b"|status".as_slice(), b"pds|status|".as_slice())
} else if key.ends_with(b"|tier") {
(b"|tier".as_slice(), b"pds|tier|".as_slice())
} else {
return Ok(0);
};
let host = &key[..key.len() - legacy_suffix.len()];
std::str::from_utf8(host).into_diagnostic()?;
let mut new_key = Vec::with_capacity(current_prefix.len() + host.len());
new_key.extend_from_slice(current_prefix);
new_key.extend_from_slice(host);
let staged = new_key.len() + value.len();
batch.insert(scanned, new_key, value);
batch.remove(scanned, key);
Ok(staged)
}
#[cfg(feature = "indexer")]
/// delete the legacy `records` keyspace only after every copy chunk and its
/// resume cursor are durable. chunk commits use the data WAL while keyspace
/// deletion updates fjall's metadata independently, so force the copy durable
/// before making the old store unreachable. fjall removes the on-disk
/// directory on the next open, when recovery finds it unreferenced.
pub(super) fn retire_legacy_stores(db: &crate::db::Db) -> miette::Result<bool> {
if !events::retire_blocks(db)? {
return Ok(false);
}
if let Some(records) = super::legacy_records(db)? {
db.persist()?;
db.inner.delete_keyspace(records).into_diagnostic()?;
db.persist()?;
}
Ok(true)
}
#[cfg(not(feature = "indexer"))]
pub(super) fn retire_legacy_stores(db: &crate::db::Db) -> miette::Result<bool> {
events::retire_blocks(db)
}
#[cfg(feature = "indexer")]
/// parse a legacy `{DID}|{COL}|{tag}{rkey}` key.
///
/// returns the trimmed-DID length, the collection, and the rkey. legacy
/// encoding knowledge lives here, frozen; the live codec is in `db::keys`.
fn parse_legacy_key(key: &[u8]) -> Result<(usize, &str, DbRkey)> {
let did_len = idx::trimmed_did_len(key)
.ok_or_else(|| miette::miette!("record key has no DID terminator: {key:?}"))?;
if key.get(did_len) != Some(&keys::SEP) {
miette::bail!("record key is not in the legacy format: {key:?}");
}
let rest = &key[did_len + 1..];
let sep = rest
.iter()
.position(|&b| b == keys::SEP)
.ok_or_else(|| miette::miette!("legacy record key missing collection separator"))?;
let collection = std::str::from_utf8(&rest[..sep])
.into_diagnostic()
.wrap_err("legacy record key has invalid collection utf8")?;
let rkey = match rest[sep + 1..].split_first() {
Some((b't', bytes)) => {
let bytes: [u8; 8] = bytes
.try_into()
.into_diagnostic()
.wrap_err("legacy record key has invalid tid rkey length")?;
DbRkey::Tid(DbTid::new_from_bytes(bytes))
}
Some((b's', bytes)) => {
let s = std::str::from_utf8(bytes)
.into_diagnostic()
.wrap_err("legacy record key has invalid string rkey")?;
DbRkey::new(s)
}
_ => miette::bail!("legacy record key has unknown rkey type tag"),
};
Ok((did_len, collection, rkey))
}
#[cfg(feature = "indexer")]
fn rewrite_record(
db: &Db,
batch: &mut OwnedWriteBatch,
_scanned: &fjall::Keyspace,
key: &[u8],
value: &[u8],
) -> Result<usize> {
// already-current key (defense against out-of-order surprises): the body
// is final, copy it verbatim into heads
if is_current_key(key) {
batch.insert(&db.indexer.heads, key, value);
return Ok(key.len() + value.len());
}
let (did_len, collection, rkey) = parse_legacy_key(key)?;
// keep the trimmed-DID bytes, swap the separators, re-encode the rkey as text
let mut new_key = Vec::with_capacity(key.len() + 2);
new_key.extend_from_slice(&key[..did_len]);
new_key.push(REC_SEP);
new_key.extend_from_slice(collection.as_bytes());
new_key.push(REC_PATH_SEP);
new_key.extend_from_slice(rkey.to_smolstr().as_bytes());
// pre-v10 values are always raw cid bytes. parse them even when no block
// exists so malformed storage cannot be blessed as the current layout.
let cid = cid::Cid::read_bytes(value)
.into_diagnostic()
.wrap_err("v10: legacy record value is not a cid")?;
if cid.to_bytes().as_slice() != value {
miette::bail!("v10: legacy record value is not a canonical cid");
}
// resolve body-bearing databases from the legacy CAS. links-only
// databases have no block for these cids and keep the cid value;
// `cid_from_record_value` disambiguates the two shapes on read.
let block_key = keys::block_key(collection, value);
let new_value = match super::legacy_block(db, &block_key)? {
Some(body) => {
let body_cid = crate::car::compute_cid(&body);
if body_cid != cid {
miette::bail!(
"v10: legacy record body cid {body_cid} does not match stored pointer {cid}"
);
}
body
}
None => {
tracing::debug!("v10: no block for record, keeping cid value (links-only database?)");
fjall::Slice::from(value)
}
};
let staged_bytes = new_value.len();
batch.insert(&db.indexer.heads, new_key, new_value);
Ok(staged_bytes)
}
#[cfg(all(test, feature = "indexer"))]
mod tests {
use crate::config::Config;
use crate::db::Db;
use crate::db::keys::{self};
#[cfg(feature = "indexer_stream")]
use crate::db::types::DbAction;
use crate::db::types::{DbRkey, DbTid, TrimmedDid};
#[cfg(feature = "indexer_stream")]
use crate::types::{StoredData, StoredEvent};
#[cfg(feature = "indexer_stream")]
use jacquard_common::CowStr;
use jacquard_common::types::string::{Did, Tid};
use miette::{IntoDiagnostic, Result};
use tempfile::tempdir;
const DID: &str = "did:plc:yk4q3id7id6p5z3bypvshc64";
fn did() -> Did {
Did::new_static(DID).unwrap()
}
/// legacy `{DID}|{COL}|{tag}{rkey}` key, built by hand: the live codec
/// only writes the current format.
fn legacy_record_key(did: &Did, collection: &str, rkey: &DbRkey) -> Vec<u8> {
let mut key = Vec::new();
TrimmedDid::from(did).write_to_vec(&mut key);
key.push(keys::SEP);
key.extend_from_slice(collection.as_bytes());
key.push(keys::SEP);
match rkey {
DbRkey::Tid(t) => {
key.push(b't');
key.extend_from_slice(t.as_bytes());
}
DbRkey::Str(s) => {
key.push(b's');
key.extend_from_slice(s.as_bytes());
}
}
key
}
#[test]
fn v10_rewrites_records_and_retires_legacy_storage() -> Result<()> {
let tmp = tempdir().into_diagnostic()?;
let cfg = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let tid_rkey = DbRkey::Tid(DbTid::from(&Tid::new("3kzbif5moe22m").unwrap()));
let str_rkey = DbRkey::new("self");
let body1: &[u8] = b"post body v1";
let body2: &[u8] = b"profile body";
let cid1 = crate::car::compute_cid(body1).to_bytes();
let cid2 = crate::car::compute_cid(body2).to_bytes();
// a cid with no block, the links-only shape: the value must survive
let cid3 = crate::car::compute_cid(b"missing links-only body").to_bytes();
{
let db = Db::open(&cfg)?;
let mut batch = db.inner.batch();
let blocks = super::super::legacy_blocks_for_test(&db)?;
let records = super::super::legacy_records_for_test(&db)?;
let post_legacy = legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey);
let profile_legacy = legacy_record_key(&did(), "app.bsky.actor.profile", &str_rkey);
let orphan_legacy = legacy_record_key(&did(), "app.bsky.feed.like", &tid_rkey);
batch.insert(&records, &post_legacy, &cid1);
batch.insert(&records, &profile_legacy, &cid2);
batch.insert(&records, &orphan_legacy, &cid3);
// bodies live in the blocks cas
batch.insert(&blocks, keys::block_key("app.bsky.feed.post", &cid1), body1);
batch.insert(
&blocks,
keys::block_key("app.bsky.actor.profile", &cid2),
body2,
);
// an already-current record: the pass must leave it alone
let current_key = keys::record_key(&did(), "app.bsky.feed.repost", &str_rkey);
batch.insert(&records, ¤t_key, b"repost body");
// a body-less legacy event remains valid through the event passes
#[cfg(feature = "indexer_stream")]
{
let event = StoredEvent {
live: false,
did: TrimmedDid::from(&did()).into_static(),
rev: DbTid::from(&Tid::new("3kzbif5moe22m").unwrap()),
collection: CowStr::Borrowed("app.bsky.feed.post"),
rkey: tid_rkey.clone(),
action: DbAction::Delete,
data: StoredData::Nothing,
};
db.stream.stage_event(
&mut batch,
keys::event_key(41),
rmp_serde::to_vec(&event).into_diagnostic()?,
);
}
batch.commit().into_diagnostic()?;
crate::db::migration::rewind_version_for_test(&db, 9)?;
db.persist()?;
}
let db = Db::open(&cfg)?;
// bodies resolved from blocks, under current keys
let post_key = keys::record_key(&did(), "app.bsky.feed.post", &tid_rkey);
assert_eq!(
db.indexer.head(&post_key).into_diagnostic()?.as_deref(),
Some(body1)
);
let profile_key = keys::record_key(&did(), "app.bsky.actor.profile", &str_rkey);
assert_eq!(
db.indexer.head(&profile_key).into_diagnostic()?.as_deref(),
Some(body2)
);
// a record without a block keeps its cid value under the new key
let orphan_key = keys::record_key(&did(), "app.bsky.feed.like", &tid_rkey);
assert_eq!(
db.indexer.head(&orphan_key).into_diagnostic()?.as_deref(),
Some(cid3.as_slice())
);
// legacy keys are gone, the current-format record is untouched
assert!(
db.indexer
.head(legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey))
.into_diagnostic()?
.is_none()
);
let current_key = keys::record_key(&did(), "app.bsky.feed.repost", &str_rkey);
assert_eq!(
db.indexer.head(¤t_key).into_diagnostic()?.as_deref(),
Some(b"repost body".as_slice())
);
// v10 retires the frozen cas after event compatibility is established
assert!(!db.inner.keyspace_exists("blocks"));
// and the legacy records store, whose contents moved to heads
assert!(!db.inner.keyspace_exists("records"));
Ok(())
}
#[test]
fn v10_finishes_v9_filter_keys_skipped_by_a_reduced_feature_build() -> Result<()> {
let tmp = tempdir().into_diagnostic()?;
let cfg = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let current = b"pds|tier|status";
let legacy = b"example.com|tier";
{
let db = Db::open(&cfg)?;
let mut batch = db.inner.batch();
batch.insert(&db.filter, current, b"trusted");
batch.insert(&db.filter, legacy, b"slow");
batch.commit().into_diagnostic()?;
crate::db::migration::rewind_version_for_test(&db, 9)?;
db.persist()?;
}
let db = Db::open(&cfg)?;
assert_eq!(
db.filter.get(current).into_diagnostic()?.as_deref(),
Some(b"trusted".as_slice())
);
assert!(db.filter.get(legacy).into_diagnostic()?.is_none());
assert_eq!(
db.filter
.get(b"pds|tier|example.com")
.into_diagnostic()?
.as_deref(),
Some(b"slow".as_slice())
);
Ok(())
}
#[test]
fn v10_rejects_a_legacy_block_under_the_wrong_cid() -> Result<()> {
let tmp = tempdir().into_diagnostic()?;
let cfg = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let rkey = DbRkey::new("self");
let expected = b"expected body";
let wrong = b"wrong body";
let cid = crate::car::compute_cid(expected).to_bytes();
let legacy_key = legacy_record_key(&did(), "app.bsky.actor.profile", &rkey);
{
let db = Db::open(&cfg)?;
let blocks = super::super::legacy_blocks_for_test(&db)?;
let records = super::super::legacy_records_for_test(&db)?;
let mut batch = db.inner.batch();
batch.insert(&records, &legacy_key, &cid);
batch.insert(
&blocks,
keys::block_key("app.bsky.actor.profile", &cid),
wrong,
);
batch.commit().into_diagnostic()?;
crate::db::migration::rewind_version_for_test(&db, 9)?;
db.persist()?;
}
let err = match Db::open(&cfg) {
Ok(_) => panic!("mismatched legacy body must abort v10"),
Err(err) => err,
};
let rendered = format!("{err:?}");
assert!(
rendered.contains("legacy record body cid") && rendered.contains("stored pointer"),
"unexpected error: {err:?}"
);
let raw = fjall::Database::builder(tmp.path())
.open()
.into_diagnostic()?;
let records = raw
.keyspace("records", fjall::KeyspaceCreateOptions::default)
.into_diagnostic()?;
assert!(records.contains_key(legacy_key).into_diagnostic()?);
assert!(raw.keyspace_exists("blocks"));
Ok(())
}
#[test]
fn v10_heads_are_actually_compressed() -> Result<()> {
// v9's records keyspace persisted "never compress" into its on-disk
// config; the migrated bodies must land in a store that compresses
let tmp = tempdir().into_diagnostic()?;
let cfg = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
// highly compressible bodies, larger than one data block
let body = vec![7u8; 4096];
let count = 2000u32;
let raw: u64;
{
let db = Db::open(&cfg)?;
let records = super::super::legacy_records_for_test(&db)?;
let mut batch = db.inner.batch();
for n in 0..count {
let key = keys::record_key(
&Did::new_owned(format!("did:plc:{n:016x}")).unwrap(),
"app.bsky.feed.post",
&DbRkey::new(&format!("r{n}")),
);
batch.insert(&records, key, &body);
}
batch.commit().into_diagnostic()?;
raw = (0..count)
.map(|n| {
(keys::record_key(
&Did::new_owned(format!("did:plc:{n:016x}")).unwrap(),
"app.bsky.feed.post",
&DbRkey::new(&format!("r{n}")),
)
.len()
+ body.len()) as u64
})
.sum();
crate::db::migration::rewind_version_for_test(&db, 9)?;
db.persist()?;
}
let db = Db::open(&cfg)?;
db.indexer.heads.major_compact().into_diagnostic()?;
db.persist()?;
let stored = db.indexer.heads.disk_space();
assert!(
stored < raw / 2,
"heads should compress repeated bodies: stored {stored} vs raw {raw}"
);
Ok(())
}
#[test]
fn v10_byte_budget_chunking_limits_output_bytes() -> Result<()> {
let tmp = tempdir().into_diagnostic()?;
let cfg = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let db = Db::open(&cfg)?;
let mut batch = db.inner.batch();
let blocks = super::super::legacy_blocks_for_test(&db)?;
let records = super::super::legacy_records_for_test(&db)?;
let large_body = vec![7u8; 10_000]; // 10 KB body
// tid alphabet is base32-sortable (no 0/1/8/9 digits)
const TID_CHARS: &[u8] = b"234567abcdefghij";
for i in 0..10 {
let c = TID_CHARS[i] as char;
let tid_rkey = DbRkey::Tid(DbTid::from(
&Tid::new(&format!("3kzbif5moe2{c}{c}")).unwrap(),
));
let cid = crate::car::compute_cid(&large_body).to_bytes();
let legacy_key = legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey);
batch.insert(&records, &legacy_key, &cid);
batch.insert(
&blocks,
keys::block_key("app.bsky.feed.post", &cid),
&large_body,
);
}
batch.commit().into_diagnostic()?;
let pass = super::Pass {
name: "records",
scan: "records",
visit: super::rewrite_record,
budget: super::ChunkBudget {
entries: usize::MAX,
bytes: 25_000, // 25 KB budget
},
optional_on_absent: true,
};
let cursor_key = super::super::pass_cursor_key(9, pass.name);
let ks = super::super::legacy_records_for_test(&db)?;
let outcome = super::super::run_chunk(&db, &cursor_key, &ks, &pass)?;
assert_eq!(outcome.seen, 3);
assert!(!outcome.exhausted);
Ok(())
}
}