From 2638e8fd673bb85ad0738e163ed5b5380aeae3d7 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 05 Aug 2026 15:13:23 +0000 Subject: [PATCH] [db] retire legacy blocks keyspace in v10 the chunked inline_record_bodies migration materializes unresolved legacy event bodies into the event_bodies compatibility archive, verifies every record body against its claimed cid, and deletes the blocks keyspace. event_bodies is a finite archive: new writes never add entries, and the retention filter never applies to it. pre-release v10 databases without the final per-migration marker are rejected rather than silently adopted (hydrant-ksf: the v11 history watermark is preserved as counts data, so marker adoption cannot orphan it). issue: hydrant-6vo, hydrant-ksf, hydrant-pkq --- src/db/migration/mod.rs | 71 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-- src/db/migration/v10.rs | 509 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/db/migration/v10/events.rs | 680 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 file(s) changed, 1258 insertion(s)(+), 2 deletion(s)(-) diff --git a/src/db/migration/mod.rs b/src/db/migration/mod.rs --- a/src/db/migration/mod.rs +++ b/src/db/migration/mod.rs @@ -6,7 +6,40 @@ 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> { + 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> { + 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 { + db.inner + .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default) + .into_diagnostic() +} + mod v1; +mod v10; mod v2; mod v3; mod v4; @@ -43,8 +76,6 @@ /// /// the only way to rewrite a keyspace larger than memory. passes run in /// order, and each runs to completion before the next starts. - // first user lands with `inline_record_bodies` - #[allow(dead_code)] Chunked { passes: &'static [Pass], finalize: Option, @@ -159,6 +190,13 @@ "migrate_excludes_and_pds_keys", Migration::Atomic(v9::migrate_v9), ), + ( + "inline_record_bodies", + Migration::Chunked { + passes: v10::PASSES, + finalize: Some(v10::retire_blocks), + }, + ), ]; pub(crate) const LATEST_VERSION: u64 = MIGRATIONS.len() as u64; @@ -216,7 +254,15 @@ 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), }; } @@ -743,6 +789,27 @@ .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(()) } diff --git a/src/db/migration/v10.rs b/src/db/migration/v10.rs new file mode 100644 --- /dev/null +++ b/src/db/migration/v10.rs @@ -0,0 +1,509 @@ +//! v10: inline record bodies and retire the legacy blocks CAS. +//! +//! legacy `{DID}|{COL}|{tag}{rkey}` -> cid entries become +//! `{DID} 00 {COL} 2f {rkey text}` -> body, resolved from the blocks CAS. +//! new keys sort before legacy keys within a repo (NUL < `|` after the +//! trimmed DID), so the resuming chunked scan only ever meets legacy keys. +//! +//! the same bounded migration archives otherwise-unresolved legacy permanent +//! pointer bodies before deleting `blocks`. 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, + }, + #[cfg(feature = "indexer")] + Pass { + name: "records", + scan: "records", + visit: rewrite_record, + budget: ChunkBudget { + entries: 100_000, + bytes: DEFAULT_RECORD_CHUNK_BYTES, + }, + }, + #[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, + key: &[u8], + value: &[u8], +) -> Result { + 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(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(&db.filter, new_key, []); + batch.remove(&db.filter, 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(&db.filter, new_key, value); + batch.remove(&db.filter, key); + Ok(staged) +} + +pub(super) fn retire_blocks(db: &crate::db::Db) -> miette::Result { + 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, key: &[u8], value: &[u8]) -> Result { + if is_current_key(key) { + return Ok(0); + } + 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 = jacquard_repo::mst::util::compute_cid(&body) + .into_diagnostic() + .wrap_err("v10: cannot hash legacy record 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.records, new_key, new_value); + batch.remove(&db.indexer.records, key); + 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<'static> { + Did::new(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 { + 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 = jacquard_repo::mst::util::compute_cid(body1) + .into_diagnostic()? + .to_bytes(); + let cid2 = jacquard_repo::mst::util::compute_cid(body2) + .into_diagnostic()? + .to_bytes(); + // a cid with no block, the links-only shape: the value must survive + let cid3 = jacquard_repo::mst::util::compute_cid(b"missing links-only body") + .into_diagnostic()? + .to_bytes(); + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + let blocks = super::super::legacy_blocks_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(&db.indexer.records, &post_legacy, &cid1); + batch.insert(&db.indexer.records, &profile_legacy, &cid2); + batch.insert(&db.indexer.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(&db.indexer.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.record(&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 + .record(&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.record(&orphan_key).into_diagnostic()?.as_deref(), + Some(cid3.as_slice()) + ); + + // legacy keys are gone, the current-format record is untouched + assert!( + db.indexer + .record(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 + .record(¤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")); + + 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 = jacquard_repo::mst::util::compute_cid(expected) + .into_diagnostic()? + .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 mut batch = db.inner.batch(); + batch.insert(&db.indexer.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_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 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 = jacquard_repo::mst::util::compute_cid(&large_body) + .into_diagnostic()? + .to_bytes(); + let legacy_key = legacy_record_key(&did(), "app.bsky.feed.post", &tid_rkey); + batch.insert(&db.indexer.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 + }, + }; + let cursor_key = super::super::pass_cursor_key(9, pass.name); + let ks = db.keyspace_by_name("records").expect("records keyspace"); + + let outcome = super::super::run_chunk(&db, &cursor_key, &ks, &pass)?; + assert_eq!(outcome.seen, 3); + assert!(!outcome.exhausted); + Ok(()) + } +} diff --git a/src/db/migration/v10/events.rs b/src/db/migration/v10/events.rs new file mode 100644 --- /dev/null +++ b/src/db/migration/v10/events.rs @@ -0,0 +1,680 @@ +//! legacy event-body passes for v10. +//! +//! v10 moved permanent current heads into `records`, but legacy permanent +//! stream events can still point into `blocks`. materialize only otherwise- +//! unresolved pointer versions into the record-scoped `event_bodies` archive, +//! then clear the legacy CAS exactly. ephemeral `Block` events remain inline: +//! event TTL must delete their body bytes with the event. + +#[cfg(feature = "indexer_stream")] +use fjall::OwnedWriteBatch; +#[cfg(feature = "indexer_stream")] +use miette::WrapErr; +#[cfg(feature = "indexer")] +use miette::{IntoDiagnostic, Result}; + +#[cfg(feature = "indexer_stream")] +use crate::db::{Db, keys}; +#[cfg(feature = "indexer_stream")] +use crate::types::{StoredData, StoredEvent}; + +#[cfg(feature = "indexer_stream")] +use super::super::{ChunkBudget, Pass}; + +#[cfg(feature = "indexer_stream")] +pub(super) const POINTER_EVENT_BODIES: Pass = Pass { + name: "pointer_event_bodies", + scan: "events", + visit: materialize_pointer_event, + budget: ChunkBudget::DEFAULT, +}; + +#[cfg(feature = "indexer_stream")] +fn body_matches(value: &[u8], expected: &jacquard_common::types::cid::IpldCid) -> bool { + !crate::db::indexer::is_cid_record_value(value) + && jacquard_repo::mst::util::compute_cid(value).is_ok_and(|computed| computed == *expected) +} + +#[cfg(feature = "indexer_stream")] +fn stage_archive_body( + db: &Db, + batch: &mut OwnedWriteBatch, + record_key: &[u8], + cid: &jacquard_common::types::cid::IpldCid, + body: &[u8], +) -> Result { + let computed = jacquard_repo::mst::util::compute_cid(body) + .into_diagnostic() + .wrap_err("v10: cannot hash legacy event body")?; + if computed != *cid { + miette::bail!("v10: legacy body cid {computed} does not match event pointer {cid}"); + } + + let key = keys::event_body_key(record_key, cid); + if let Some(existing) = db.stream.event_bodies.get(&key).into_diagnostic()? { + if body_matches(&existing, cid) { + return Ok(0); + } + miette::bail!("v10: compatibility body at {cid} is corrupt"); + } + + batch.insert(&db.stream.event_bodies, key, body); + Ok(body.len()) +} + +#[cfg(feature = "indexer_stream")] +fn decode_event<'a>(value: &'a [u8]) -> Result> { + rmp_serde::from_slice(value) + .into_diagnostic() + .wrap_err("v10: unreadable stored event") +} + +#[cfg(feature = "indexer_stream")] +fn materialize_pointer_event( + db: &Db, + batch: &mut OwnedWriteBatch, + _key: &[u8], + value: &[u8], +) -> Result { + let event = decode_event(value)?; + let StoredData::Ptr(cid) = &event.data else { + return Ok(0); + }; + + let record_key = keys::record_key_trimmed(&event.did, event.collection.as_str(), &event.rkey); + if db + .resolve_event_record_body(&record_key, &event.rev, cid)? + .is_some() + { + return Ok(0); + } + + // migration runs synchronously inside `Db::open`, before ingestion can add + // events. every unresolved pointer here therefore predates v10 and must be + // resolved from the legacy CAS. + let block_key = keys::block_key(event.collection.as_str(), &cid.to_bytes()); + let body = super::super::legacy_block(db, &block_key)? + .ok_or_else(|| miette::miette!("v10: legacy event body {cid} is missing"))?; + stage_archive_body(db, batch, &record_key, cid, &body) +} + +/// delete the legacy CAS only after every event-materialization chunk and its +/// resume cursor are durable. a crash before the migration marker merely makes +/// this idempotent finalizer run again. +#[cfg(feature = "indexer")] +pub(super) fn retire_blocks(db: &crate::db::Db) -> Result { + #[cfg(not(feature = "indexer_stream"))] + if db.inner.keyspace_exists("events") { + let events = db + .inner + .keyspace("events", Default::default) + .into_diagnostic()?; + if !events.is_empty().into_diagnostic()? { + tracing::warn!( + "v10: legacy events exist but indexer_stream is disabled; retaining blocks until a stream-enabled open" + ); + return Ok(false); + } + } + + // chunk commits and their resume cursor use the data WAL, while keyspace + // deletion updates fjall's metadata independently. force the completed + // archive durable before making the old CAS unreachable. + db.persist()?; + if let Some(blocks) = super::super::legacy_blocks(db)? { + db.inner.delete_keyspace(blocks).into_diagnostic()?; + db.persist()?; + } + Ok(true) +} + +#[cfg(not(feature = "indexer"))] +pub(super) fn retire_blocks(_db: &crate::db::Db) -> miette::Result { + Ok(false) +} + +#[cfg(all(test, feature = "indexer", not(feature = "indexer_stream")))] +mod indexer_only_tests { + use super::*; + use crate::config::Config; + + fn config(path: &std::path::Path) -> Config { + Config { + database_path: path.to_path_buf(), + ..Default::default() + } + } + + fn seed_dormant_events(db: &crate::db::Db, nonempty: bool) -> Result<()> { + let events = db + .inner + .keyspace("events", Default::default) + .into_diagnostic()?; + if nonempty { + events + .insert(1_u64.to_be_bytes(), b"legacy event") + .into_diagnostic()?; + } + let blocks = super::super::super::legacy_blocks_for_test(db)?; + blocks.insert(b"sentinel", b"body").into_diagnostic()?; + crate::db::migration::rewind_version_for_test(db, 9)?; + db.persist() + } + + #[test] + fn empty_dormant_events_do_not_block_retirement() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + { + let db = crate::db::Db::open(&cfg)?; + seed_dormant_events(&db, false)?; + } + + let db = crate::db::Db::open(&cfg)?; + assert!(!db.inner.keyspace_exists("blocks")); + Ok(()) + } + + #[test] + fn nonempty_dormant_events_retain_blocks_for_a_stream_build() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + { + let db = crate::db::Db::open(&cfg)?; + seed_dormant_events(&db, true)?; + } + + let db = crate::db::Db::open(&cfg)?; + assert!(db.inner.keyspace_exists("blocks")); + Ok(()) + } +} + +#[cfg(all(test, feature = "indexer_stream"))] +mod tests { + use super::*; + use crate::config::Config; + use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid}; + use jacquard_common::types::string::{Did, Tid}; + use jacquard_common::{CowStr, IntoStatic}; + use miette::IntoDiagnostic; + + const DID: &str = "did:plc:yk4q3id7id6p5z3bypvshc64"; + const COLLECTION: &str = "app.bsky.feed.post"; + const RKEY: &str = "3kzbif5moe22m"; + + fn config(path: &std::path::Path) -> Config { + Config { + database_path: path.to_path_buf(), + ..Default::default() + } + } + + fn did() -> Did<'static> { + Did::new(DID).unwrap() + } + + fn rev() -> DbTid { + DbTid::from(&Tid::new("3kzbif5moe22m").unwrap()) + } + + fn body(text: &str) -> Vec { + serde_ipld_dagcbor::to_vec(&serde_json::json!({ + "$type": COLLECTION, + "text": text, + })) + .unwrap() + } + + fn event(data: StoredData) -> StoredEvent<'static> { + StoredEvent { + live: false, + did: TrimmedDid::from(&did()).into_static(), + rev: rev(), + collection: CowStr::Borrowed(COLLECTION).into_static(), + rkey: DbRkey::new(RKEY), + action: DbAction::Create, + data, + } + } + + fn record_key() -> Vec { + keys::record_key(&did(), COLLECTION, &DbRkey::new(RKEY)) + } + + fn stage_event(db: &Db, batch: &mut OwnedWriteBatch, id: u64, event: &StoredEvent<'_>) { + db.stream.stage_event( + batch, + keys::event_key(id), + rmp_serde::to_vec(event).unwrap(), + ); + } + + fn stage_legacy_block( + db: &Db, + batch: &mut OwnedWriteBatch, + cid: &jacquard_common::types::cid::IpldCid, + body: &[u8], + ) -> Result<()> { + let blocks = super::super::super::legacy_blocks_for_test(db)?; + batch.insert(&blocks, keys::block_key(COLLECTION, &cid.to_bytes()), body); + Ok(()) + } + + fn rewind_to_v9(db: &Db) -> Result<()> { + crate::db::migration::rewind_version_for_test(db, 9)?; + db.persist() + } + + #[test] + fn pointer_body_is_materialized_before_blocks_are_cleared() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body = body("legacy"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + stage_legacy_block(&db, &mut batch, &cid, &body)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + assert!(!db.inner.keyspace_exists("blocks")); + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + Ok(()) + } + + #[test] + fn inline_ephemeral_body_stays_owned_by_its_event() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let mut cfg = config(tmp.path()); + cfg.ephemeral = true; + let body = body("inline"); + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event( + &db, + &mut batch, + 7, + &event(StoredData::Block(bytes::Bytes::copy_from_slice(&body))), + ); + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + let stored = db + .stream + .events + .get(keys::event_key(7)) + .into_diagnostic()? + .expect("event should remain"); + let stored: StoredEvent<'_> = rmp_serde::from_slice(&stored).into_diagnostic()?; + assert!(matches!(stored.data, StoredData::Block(found) if found.as_ref() == body)); + assert!(db.stream.event_bodies.is_empty().into_diagnostic()?); + Ok(()) + } + + #[test] + fn same_record_and_revision_can_archive_distinct_cids() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body_a = body("fork a"); + let body_b = body("fork b"); + let cid_a = jacquard_repo::mst::util::compute_cid(&body_a).into_diagnostic()?; + let cid_b = jacquard_repo::mst::util::compute_cid(&body_b).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid_a))); + stage_event(&db, &mut batch, 2, &event(StoredData::Ptr(cid_b))); + stage_legacy_block(&db, &mut batch, &cid_a, &body_a)?; + stage_legacy_block(&db, &mut batch, &cid_b, &body_b)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid_a)) + .into_diagnostic()? + .as_deref(), + Some(body_a.as_slice()) + ); + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid_b)) + .into_diagnostic()? + .as_deref(), + Some(body_b.as_slice()) + ); + Ok(()) + } + + #[test] + fn current_head_is_not_duplicated_in_the_archive() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body = body("head"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + db.indexer.stage_record(&mut batch, record_key(), &body); + stage_legacy_block(&db, &mut batch, &cid, &body)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + assert!(!db.inner.keyspace_exists("blocks")); + assert!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .is_none() + ); + Ok(()) + } + + #[test] + fn history_before_the_event_does_not_suppress_archival() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body = body("legacy"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + db.indexer.stage_history( + &mut batch, + keys::history_key(&record_key(), &DbTid::new_from_bytes(0_u64.to_be_bytes())), + &body, + ); + stage_legacy_block(&db, &mut batch, &cid, &body)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + Ok(()) + } + + #[test] + fn preexisting_archive_makes_pointer_migration_idempotent() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body = body("already archived"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + batch.insert( + &db.stream.event_bodies, + keys::event_body_key(&record_key(), &cid), + &body, + ); + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let db = Db::open(&cfg)?; + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + assert!(!db.inner.keyspace_exists("blocks")); + Ok(()) + } + + #[test] + fn missing_body_aborts_without_clearing_blocks() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let missing = body("missing"); + let cid = jacquard_repo::mst::util::compute_cid(&missing).into_diagnostic()?; + let sentinel = body("sentinel"); + let sentinel_cid = jacquard_repo::mst::util::compute_cid(&sentinel).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + stage_legacy_block(&db, &mut batch, &sentinel_cid, &sentinel)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let err = match Db::open(&cfg) { + Ok(_) => panic!("missing sole body must abort migration"), + Err(err) => err, + }; + assert!(format!("{err:?}").contains("is missing")); + + let raw = fjall::Database::builder(tmp.path()) + .open() + .into_diagnostic()?; + let blocks = raw + .keyspace("blocks", fjall::KeyspaceCreateOptions::default) + .into_diagnostic()?; + assert!(!blocks.is_empty().into_diagnostic()?); + Ok(()) + } + + #[test] + fn cid_mismatch_aborts_without_clearing_blocks() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let expected_body = body("expected"); + let wrong_body = body("wrong"); + let cid = jacquard_repo::mst::util::compute_cid(&expected_body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + stage_legacy_block(&db, &mut batch, &cid, &wrong_body)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + let err = match Db::open(&cfg) { + Ok(_) => panic!("cid mismatch must abort migration"), + Err(err) => err, + }; + assert!( + format!("{err:?}").contains("legacy body cid"), + "unexpected error: {err:?}" + ); + + let raw = fjall::Database::builder(tmp.path()) + .open() + .into_diagnostic()?; + let blocks = raw + .keyspace("blocks", fjall::KeyspaceCreateOptions::default) + .into_diagnostic()?; + assert!(!blocks.is_empty().into_diagnostic()?); + Ok(()) + } + + #[test] + fn malformed_event_aborts_without_deleting_legacy_blocks() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let sentinel = body("sentinel"); + let sentinel_cid = jacquard_repo::mst::util::compute_cid(&sentinel).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + db.stream + .stage_event(&mut batch, keys::event_key(1), b"not msgpack"); + stage_legacy_block(&db, &mut batch, &sentinel_cid, &sentinel)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + } + + assert!(Db::open(&cfg).is_err()); + let raw = fjall::Database::builder(tmp.path()) + .open() + .into_diagnostic()?; + let blocks = raw + .keyspace("blocks", fjall::KeyspaceCreateOptions::default) + .into_diagnostic()?; + assert!(!blocks.is_empty().into_diagnostic()?); + Ok(()) + } + + #[test] + fn migration_resumes_after_a_partial_event_chunk() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body_a = body("a"); + let body_b = body("b"); + let cid_a = jacquard_repo::mst::util::compute_cid(&body_a).into_diagnostic()?; + let cid_b = jacquard_repo::mst::util::compute_cid(&body_b).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid_a))); + stage_event(&db, &mut batch, 2, &event(StoredData::Ptr(cid_b))); + stage_legacy_block(&db, &mut batch, &cid_a, &body_a)?; + stage_legacy_block(&db, &mut batch, &cid_b, &body_b)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + + let pass = Pass { + name: "pointer_event_bodies", + scan: "events", + visit: materialize_pointer_event, + budget: ChunkBudget { + entries: 1, + bytes: usize::MAX, + }, + }; + let cursor = super::super::super::pass_cursor_key(9, pass.name); + let events = db.keyspace_by_name("events").expect("events keyspace"); + let outcome = super::super::super::run_chunk(&db, &cursor, &events, &pass)?; + assert_eq!(outcome.seen, 1); + assert!(!outcome.exhausted); + db.persist()?; + } + + let db = Db::open(&cfg)?; + assert!(!db.inner.keyspace_exists("blocks")); + for (cid, body) in [(cid_a, body_a), (cid_b, body_b)] { + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + } + Ok(()) + } + + #[test] + fn migration_retries_after_keyspace_delete_before_marker() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let cfg = config(tmp.path()); + let body = body("legacy"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + + { + let db = Db::open(&cfg)?; + let mut batch = db.inner.batch(); + stage_event(&db, &mut batch, 1, &event(StoredData::Ptr(cid))); + stage_legacy_block(&db, &mut batch, &cid, &body)?; + batch.commit().into_diagnostic()?; + rewind_to_v9(&db)?; + + for pass in super::super::PASSES { + assert!(super::super::super::run_pass(&db, 9, pass)?); + } + assert!(retire_blocks(&db)?); + assert!(!db.inner.keyspace_exists("blocks")); + // deliberately do not write v10's applied marker + } + + let db = Db::open(&cfg)?; + assert!(!db.inner.keyspace_exists("blocks")); + assert_eq!( + db.stream + .event_bodies + .get(keys::event_body_key(&record_key(), &cid)) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + Ok(()) + } + + #[test] + fn history_ttl_filter_never_applies_to_compatibility_bodies() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let mut cfg = config(tmp.path()); + cfg.history_ttl = Some(std::time::Duration::from_secs(1)); + let db = Db::open(&cfg)?; + let body = body("compatibility"); + let cid = jacquard_repo::mst::util::compute_cid(&body).into_diagnostic()?; + let key = keys::event_body_key(&record_key(), &cid); + db.stream + .event_bodies + .insert(&key, &body) + .into_diagnostic()?; + db.stream + .event_bodies + .rotate_memtable_and_wait() + .into_diagnostic()?; + db.stream.event_bodies.major_compact().into_diagnostic()?; + assert_eq!( + db.stream + .event_bodies + .get(key) + .into_diagnostic()? + .as_deref(), + Some(body.as_slice()) + ); + Ok(()) + } + +} -- tangled.sh