#[cfg(feature = "indexer")] use std::collections::HashMap; use std::time::Duration; #[cfg(feature = "indexer")] use bytes::Bytes; #[cfg(feature = "indexer")] use jacquard_common::types::cid::IpldCid; #[cfg(feature = "indexer")] use jacquard_common::types::did::Did; #[cfg(feature = "indexer")] use jacquard_common::types::string::Tid; use miette::{IntoDiagnostic, Result}; #[cfg(feature = "indexer")] use crate::db::types::{DbAction, DbRkey, DbTid}; use crate::db::{CountDeltas, Db}; #[cfg(feature = "indexer")] use crate::db::{LifecycleCountBatch, keys}; #[cfg(feature = "indexer")] use crate::ops::record_events::{EmitOp, RecordEmitter, RecordEventOrigin, RecordEvents}; #[cfg(feature = "indexer")] use crate::state::AppState; #[cfg(feature = "indexer")] use crate::types::{GaugeState, RepoState}; #[cfg(feature = "firehose-diagnostics")] type TxnInstant = std::time::Instant; #[cfg(not(feature = "firehose-diagnostics"))] #[derive(Clone, Copy)] struct TxnInstant; #[cfg(not(feature = "firehose-diagnostics"))] impl TxnInstant { #[inline(always)] fn now() -> Self { Self } #[inline(always)] fn elapsed(&self) -> Duration { Duration::ZERO } } pub(crate) struct TxnCommitTimings { pub(crate) stage_counts: Duration, pub(crate) stage_and_commit: Duration, pub(crate) apply_counts: Duration, } /// one atomic database write, including its in-memory count projections. /// /// count deltas are persisted in the same fjall batch and applied in memory only /// after that batch commits. dropping a transaction discards both. pub(crate) struct Txn<'db> { pub(crate) batch: fjall::OwnedWriteBatch, pub(crate) db: &'db Db, pub(crate) counts: CountDeltas, #[cfg(feature = "indexer")] lifecycle: Option>, /// record-write guards retained from the first record/purge touch until /// this transaction drops after commit. #[cfg(feature = "indexer")] record_lock_guards: Vec>, /// lock indexes already held, so a repeat acquisition cannot self-deadlock. #[cfg(feature = "indexer")] held_record_locks: std::collections::BTreeSet, } /// record-write locks deliberately retained after their transaction commits. /// /// this lets callers linearize commit-coupled side effects, such as publishing /// body-bearing events, against operator redaction. dropping the barrier ends /// that ordering guarantee. #[cfg(feature = "indexer")] #[must_use = "dropping the barrier allows operator deletion to proceed"] pub(crate) struct RecordWriteBarrier<'db> { _guards: Vec>, } impl<'db> Txn<'db> { #[allow(dead_code)] pub(crate) fn new(db: &'db Db) -> Self { Self { batch: db.inner.batch(), db, counts: CountDeltas::default(), #[cfg(feature = "indexer")] lifecycle: None, #[cfg(feature = "indexer")] record_lock_guards: Vec::new(), #[cfg(feature = "indexer")] held_record_locks: std::collections::BTreeSet::new(), } } pub(crate) fn from_parts( db: &'db Db, batch: fjall::OwnedWriteBatch, counts: CountDeltas, ) -> Self { Self { batch, db, counts, #[cfg(feature = "indexer")] lifecycle: None, #[cfg(feature = "indexer")] record_lock_guards: Vec::new(), #[cfg(feature = "indexer")] held_record_locks: std::collections::BTreeSet::new(), } } /// hold one record-write lock until this transaction drops. recover a /// poisoned mutex so a panicking writer cannot wedge the shard forever. #[cfg(feature = "indexer")] pub(crate) fn hold_record_lock_index(&mut self, lock_index: u8) { if self.held_record_locks.contains(&lock_index) { return; } assert!( self.lifecycle.is_none(), "record locks must be acquired before lifecycle counts" ); self.held_record_locks.insert(lock_index); let guard = self.db.record_write_locks[lock_index as usize] .lock() .unwrap_or_else(|e| e.into_inner()); self.record_lock_guards.push(guard); } /// hold one repository's write barrier until this transaction drops. #[cfg(feature = "indexer")] pub(crate) fn hold_repo_write_lock(&mut self, did: &Did<'_>) { self.hold_record_lock_index(crate::db::record_lock_index_for_did(did)); } /// lock this repository, then read its durable exclusion status inside the /// same critical section. #[cfg(feature = "indexer")] pub(crate) fn lock_repo_and_is_excluded(&mut self, did: &Did<'_>) -> Result { self.hold_repo_write_lock(did); self.db .filter .contains_key(crate::db::filter::exclude_key(did.as_str())?) .into_diagnostic() } /// acquire a transaction's complete record-lock set in one canonical order. /// callers that can touch several DIDs must do this before initializing /// lifecycle counts; dynamic encounter-order locking can deadlock against /// another multi-repository operation. #[cfg(feature = "indexer")] pub(crate) fn hold_record_lock_indexes_sorted(&mut self, locks: impl IntoIterator) { assert!( self.lifecycle.is_none(), "record locks must be acquired before lifecycle counts" ); let mut locks: Vec<_> = locks.into_iter().collect(); locks.sort_unstable(); locks.dedup(); for lock in locks { self.hold_record_lock_index(lock); } } #[cfg(feature = "indexer")] pub(crate) fn records<'txn, 'did, 'repo>( &'txn mut self, state: &AppState, commit_rev: &DbTid, did: &'did Did<'repo>, ) -> Result> { self.record_scope(state, commit_rev, did, RecordEventOrigin::Live, false) } #[cfg(feature = "indexer")] pub(crate) fn backfill_records<'txn, 'did, 'repo>( &'txn mut self, state: &AppState, commit_rev: &DbTid, did: &'did Did<'repo>, ) -> Result> { self.record_scope(state, commit_rev, did, RecordEventOrigin::Backfill, true) } #[cfg(feature = "indexer")] fn record_scope<'txn, 'did, 'repo>( &'txn mut self, state: &AppState, commit_rev: &DbTid, did: &'did Did<'repo>, origin: RecordEventOrigin, count_ephemeral_records: bool, ) -> Result> { // ephemeral mode persists no mutable record state, so operator body // deletion cannot race it and there is nothing to lock. permanent // record mutations retain the lock through commit and broadcast. let excluded = if state.ephemeral { self.db .filter .contains_key(crate::db::filter::exclude_key(did.as_str())?) .into_diagnostic()? } else { self.lock_repo_and_is_excluded(did)? }; Ok(RecordTxn { txn: self, emitter: RecordEmitter::new(state, commit_rev, origin), did, rev: *commit_rev, ephemeral: state.ephemeral, only_index_links: state.only_index_links, count_ephemeral_records, records_delta: 0, history_count: 0, collection_deltas: HashMap::new(), excluded, }) } #[cfg(feature = "indexer")] pub(crate) fn transition_lifecycle( &mut self, did: &Did<'_>, gauge: GaugeState, ) -> Result { let lifecycle = self .lifecycle .get_or_insert_with(|| self.db.lifecycle_counts()); lifecycle.transition(&mut self.batch, did, gauge) } #[cfg(feature = "indexer")] pub(crate) fn transition_pending_key( &mut self, did: &Did<'_>, pending_key: &[u8], gauge: GaugeState, ) -> Result { let lifecycle = self .lifecycle .get_or_insert_with(|| self.db.lifecycle_counts()); lifecycle.transition_pending_key(&mut self.batch, did, pending_key, gauge) } #[allow(dead_code)] pub(crate) fn commit(self) -> Result<()> { self.commit_with(|_| Ok(())).map(|_| ()) } /// commit while returning the record-write guards instead of releasing /// them, so post-commit effects remain ordered before operator deletion. #[cfg(feature = "indexer")] pub(crate) fn commit_with_record_barrier(mut self) -> Result> { let guards = std::mem::take(&mut self.record_lock_guards); self.commit()?; Ok(RecordWriteBarrier { _guards: guards }) } /// stages count reservations, lets the caller add commit-coupled writes, /// commits once, then applies the in-memory projections. #[allow(dead_code)] pub(crate) fn commit_with(mut self, stage: F) -> Result<(T, TxnCommitTimings)> where F: FnOnce(&mut fjall::OwnedWriteBatch) -> Result, { let stage_counts_started = TxnInstant::now(); #[cfg(feature = "indexer")] let lifecycle_reservation = self .lifecycle .map(|lifecycle| lifecycle.stage(&mut self.batch)); let count_reservation = self.db.stage_count_deltas(&mut self.batch, &self.counts); let stage_counts = stage_counts_started.elapsed(); let stage_and_commit_started = TxnInstant::now(); let staged = stage(&mut self.batch)?; self.batch.commit().into_diagnostic()?; let stage_and_commit = stage_and_commit_started.elapsed(); let apply_counts_started = TxnInstant::now(); self.db.apply_count_deltas(&self.counts); drop(count_reservation); #[cfg(feature = "indexer")] if let Some(reservation) = lifecycle_reservation { self.db.apply_lifecycle_counts(reservation); } let apply_counts = apply_counts_started.elapsed(); Ok(( staged, TxnCommitTimings { stage_counts, stage_and_commit, apply_counts, }, )) } } /// one commit's record mutations within a larger atomic transaction. #[cfg(feature = "indexer")] pub(crate) struct RecordTxn<'txn, 'db, 'did, 'repo> { txn: &'txn mut Txn<'db>, emitter: RecordEmitter, did: &'did Did<'repo>, /// the rev of the commit being applied; history entries are keyed by it rev: DbTid, ephemeral: bool, only_index_links: bool, count_ephemeral_records: bool, records_delta: i64, history_count: i64, collection_deltas: HashMap, excluded: bool, } #[cfg(feature = "indexer")] impl RecordTxn<'_, '_, '_, '_> { pub(crate) fn is_excluded(&self) -> bool { self.excluded } pub(crate) fn put_record( &mut self, collection: &str, rkey: &DbRkey, claimed_cid: &IpldCid, block: &Bytes, action: DbAction, ) -> Result<()> { if self.is_excluded() { return Ok(()); } // the body is the source of truth; the cid always derives from it. // a mismatch with the claimed cid means this block is not the record // the repo says it is: storing it would poison every downstream // consumer (events address bodies by the claimed cid), so reject // regardless of whether MST verification is enabled let cid = jacquard_repo::mst::util::compute_cid(block.as_ref()).into_diagnostic()?; if cid != *claimed_cid { return Err(miette::miette!( "claimed cid {claimed_cid} does not match body cid {cid} for {}/{collection}/{rkey}", self.did )); } if self.ephemeral { if action == DbAction::Create && self.count_ephemeral_records { self.records_delta += 1; } return self.emitter.emit( &mut self.txn.batch, self.txn.db, EmitOp { did: self.did, collection, rkey, action, cid: Some(cid), block: Some(block), }, ); } let key = keys::record_key(self.did, collection, rkey); let head = self.txn.db.indexer.record(&key).into_diagnostic()?; // counts derive from the actual head transition, never from the // claimed action: replayed creates, updates of records we never // saw, and rewrites all count as the storage transition they // perform, which makes mutation replay idempotent let head_cid_is_current = head .as_ref() .map(|old| crate::db::ipld_cid_from_record_value(old.as_ref())) .transpose()? .is_some_and(|old_cid| old_cid == cid); let version_is_redacted = !self.only_index_links && self .txn .db .indexer .redactions .contains_key(keys::redaction_key(&key, &cid)) .into_diagnostic()?; let store_cid_marker = self.only_index_links || version_is_redacted; let head_is_cid_marker = head .as_ref() .is_some_and(|old| crate::db::is_cid_record_value(old)); let storage_is_current = head_cid_is_current && head_is_cid_marker == store_cid_marker; let cid_bytes = cid.to_bytes(); let value: &[u8] = if store_cid_marker { // to_bytes is [u8; 36]-sized for cids we produce here &cid_bytes } else { block.as_ref() }; match head { _ if storage_is_current => { // identical replay (buffered live events vs an imported // car): no storage change } Some(_) if head_cid_is_current => { // the logical version is unchanged, but its representation is // not: either enforce a durable operator redaction or repair a // non-redaction CID placeholder once the body is available. self.txn .batch .insert(&self.txn.db.indexer.records, key, value); } Some(old) => { if !self.only_index_links && !crate::db::is_cid_record_value(&old) { // a differing head moves to history before being // replaced, so replays of earlier events can still // resolve it self.txn.batch.insert( &self.txn.db.indexer.history, keys::history_key(&key, &self.death_rev()), old.as_ref(), ); self.history_count += 1; } self.txn .batch .insert(&self.txn.db.indexer.records, key, value); } None => { self.records_delta += 1; *self .collection_deltas .entry(collection.to_owned()) .or_default() += 1; self.txn .batch .insert(&self.txn.db.indexer.records, key, value); } } if version_is_redacted { crate::ops::backlink_ops::delete_record( &mut self.txn.batch, self.txn.db, self.did, collection, &rkey.to_smolstr(), )?; } else { crate::ops::backlink_ops::index_record( &mut self.txn.batch, self.txn.db, self.did, collection, &rkey.to_smolstr(), block, )?; } self.emitter.emit( &mut self.txn.batch, self.txn.db, EmitOp { did: self.did, collection, rkey, action, cid: Some(cid), block: (!version_is_redacted).then_some(block), }, ) } /// history death key for this commit: the later of the commit rev and /// the local clock. retention ages history on the same local clock as /// event retention, so delayed commits keep their bodies resolvable for /// the full configured window, and deaths always sort after the events /// that precede them fn death_rev(&self) -> DbTid { self.rev.max(DbTid::from(&Tid::now_0())) } /// true when a death entry already exists for `key` at or after this /// commit's rev: this delete (or a later one) has already been applied fn already_deleted(&self, key: &[u8]) -> Result { let lo = keys::history_key(key, &self.rev); let mut hi = key.to_vec(); hi.push(1); Ok(self .txn .db .indexer .history .range(lo..hi) .next() .map(|guard| guard.into_inner()) .transpose() .into_diagnostic()? .is_some()) } pub(crate) fn delete_record(&mut self, collection: &str, rkey: &DbRkey) -> Result<()> { if self.is_excluded() { return Ok(()); } if self.ephemeral { if self.count_ephemeral_records { self.records_delta -= 1; } return self.emitter.emit( &mut self.txn.batch, self.txn.db, EmitOp { did: self.did, collection, rkey, action: DbAction::Delete, cid: None, block: None, }, ); } let key = keys::record_key(self.did, collection, rkey); // like put_record, counts derive from the actual head transition: // deleting a missing record is not a decrement, and replaying a // delete is a full no-op so the originally preserved body survives match self.txn.db.indexer.record(&key).into_diagnostic()? { Some(old) => { self.records_delta -= 1; *self .collection_deltas .entry(collection.to_owned()) .or_default() -= 1; if !self.only_index_links { // preserve the dying body so replays of earlier events // can still resolve it let history_value: &[u8] = if crate::db::is_cid_record_value(&old) { &[] } else { old.as_ref() }; self.txn.batch.insert( &self.txn.db.indexer.history, keys::history_key(&key, &self.death_rev()), history_value, ); self.history_count += 1; } self.txn.batch.remove(&self.txn.db.indexer.records, key); } None => { if !self.only_index_links && !self.already_deleted(&key)? { // a delete of a record we never saw leaves an empty // tombstone, distinguishing "deleted at rev" from // "never existed" self.txn.batch.insert( &self.txn.db.indexer.history, keys::history_key(&key, &self.death_rev()), [], ); self.history_count += 1; } } } crate::ops::backlink_ops::delete_record( &mut self.txn.batch, self.txn.db, self.did, collection, &rkey.to_smolstr(), )?; self.emitter.emit( &mut self.txn.batch, self.txn.db, EmitOp { did: self.did, collection, rkey, action: DbAction::Delete, cid: None, block: None, }, ) } pub(crate) fn update_repo_state(&mut self, state: &RepoState<'_>) -> Result<()> { if self.is_excluded() { return Ok(()); } self.txn.batch.insert( &self.txn.db.repos, keys::repo_key(self.did), crate::db::ser_repo_state(state)?, ); Ok(()) } pub(crate) fn finish(self) -> Result { for (collection, delta) in &self.collection_deltas { crate::db::update_record_count( &mut self.txn.batch, self.txn.db, self.did, collection, *delta, )?; } self.txn.counts.add_records(self.records_delta); self.txn.counts.add_history(self.history_count); Ok(self.emitter.finish()) } } #[cfg(all(test, feature = "indexer"))] mod tests { use super::*; use crate::db::types::DbAction; use jacquard_common::types::string::Tid; use std::ops::Bound; use tempfile::TempDir; const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz"; fn test_state() -> (TempDir, AppState) { let tmp = tempfile::tempdir().unwrap(); let mut config = crate::config::Config::default(); config.database_path = tmp.path().to_path_buf(); let state = AppState::new(&config).unwrap(); (tmp, state) } fn rev(s: &str) -> DbTid { DbTid::from(&Tid::new(s).unwrap()) } fn body(s: &str) -> Bytes { Bytes::copy_from_slice(s.as_bytes()) } fn did() -> Did<'static> { Did::new(DID).unwrap() } #[test] fn committed_record_barrier_keeps_the_repo_locked() { let (_tmp, state) = test_state(); let did = did(); let lock_index = crate::db::record_lock_index_for_did(&did); let mut txn = Txn::new(&state.db); assert!(!txn.lock_repo_and_is_excluded(&did).unwrap()); let barrier = txn.commit_with_record_barrier().unwrap(); assert!(matches!( state.db.record_write_locks[lock_index as usize].try_lock(), Err(std::sync::TryLockError::WouldBlock) )); drop(barrier); assert!( state.db.record_write_locks[lock_index as usize] .try_lock() .is_ok() ); } fn write( state: &AppState, rev: &DbTid, collection: &str, rkey: &str, record: &Bytes, action: DbAction, ) { let mut txn = Txn::new(&state.db); let did = did(); let mut record_txn = txn.records(state, rev, &did).unwrap(); let cid = jacquard_repo::mst::util::compute_cid(record.as_ref()).unwrap(); record_txn .put_record(collection, &DbRkey::new(rkey), &cid, record, action) .unwrap(); record_txn.finish().unwrap(); txn.commit().unwrap(); } fn delete(state: &AppState, rev: &DbTid, collection: &str, rkey: &str) { let mut txn = Txn::new(&state.db); let did = did(); let mut record_txn = txn.records(state, rev, &did).unwrap(); record_txn .delete_record(collection, &DbRkey::new(rkey)) .unwrap(); record_txn.finish().unwrap(); txn.commit().unwrap(); } fn head(state: &AppState, collection: &str, rkey: &str) -> Option> { let key = keys::record_key(&did(), collection, &DbRkey::new(rkey)); state.db.indexer.record(&key).unwrap().map(|v| v.to_vec()) } fn history_bodies(state: &AppState, collection: &str, rkey: &str) -> Vec> { let key = keys::record_key(&did(), collection, &DbRkey::new(rkey)); let mut end = key.clone(); end.push(0xFF); state .db .indexer .history .range((Bound::Included(key.clone()), Bound::Excluded(end))) .filter_map(|guard| guard.into_inner().ok()) // `{record key} 00 {rev}` for exactly this record (history_hit_for) .filter(|(k, _)| k.len() == key.len() + 9 && k.starts_with(&key)) .map(|(_, v)| v.to_vec()) .collect() } #[test] fn create_writes_head_without_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v1".to_vec()) ); assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty()); } #[test] fn update_moves_superseded_body_to_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v2"), DbAction::Update, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v2".to_vec()) ); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); } #[test] fn identical_update_leaves_no_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Update, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v1".to_vec()) ); assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty()); } #[test] fn delete_moves_body_to_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); delete( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", ); assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); } #[test] fn recreate_after_delete_writes_new_head() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); delete( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", ); write( &state, &rev("3kzbif5mog44m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v3"), DbAction::Create, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v3".to_vec()) ); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); } #[test] fn replayed_create_over_identical_head_is_noop() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); // buffered live event replayed after the car import carried the same body write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v1".to_vec()) ); assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty()); } #[test] fn same_cid_replay_does_not_restore_a_redacted_head() { let (_tmp, state) = test_state(); let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; let body = body("v1"); write( &state, &rev("3kzbif5moe22m"), collection, rkey, &body, DbAction::Create, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::Head, ) .unwrap(); write( &state, &rev("3kzbif5mof33m"), collection, rkey, &body, DbAction::Update, ); let head = head(&state, collection, rkey).unwrap(); assert!(crate::db::is_cid_record_value(&head)); assert!(history_bodies(&state, collection, rkey).is_empty()); } #[test] fn genuinely_new_cid_replaces_a_redacted_head_without_archiving_marker() { let (_tmp, state) = test_state(); let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; write( &state, &rev("3kzbif5moe22m"), collection, rkey, &body("v1"), DbAction::Create, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::Head, ) .unwrap(); write( &state, &rev("3kzbif5mof33m"), collection, rkey, &body("v2"), DbAction::Update, ); assert_eq!(head(&state, collection, rkey), Some(b"v2".to_vec())); assert!(history_bodies(&state, collection, rkey).is_empty()); } #[test] fn redacted_cid_stays_redacted_after_an_intervening_version() { let (_tmp, state) = test_state(); let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; let v1 = body("v1"); write( &state, &rev("3kzbif5moe22m"), collection, rkey, &v1, DbAction::Create, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::Head, ) .unwrap(); write( &state, &rev("3kzbif5mof33m"), collection, rkey, &body("v2"), DbAction::Update, ); write( &state, &rev("3kzbif5mog44m"), collection, rkey, &v1, DbAction::Update, ); let head = head(&state, collection, rkey).unwrap(); assert!(crate::db::is_cid_record_value(&head)); } #[test] fn redacted_cid_stays_redacted_after_reopen() { let tmp = tempfile::tempdir().unwrap(); let config = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; let v1 = body("v1"); { let state = AppState::new(&config).unwrap(); write( &state, &rev("3kzbif5moe22m"), collection, rkey, &v1, DbAction::Create, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::Head, ) .unwrap(); state.db.persist().unwrap(); } let state = AppState::new(&config).unwrap(); write( &state, &rev("3kzbif5mof33m"), collection, rkey, &body("v2"), DbAction::Update, ); write( &state, &rev("3kzbif5mog44m"), collection, rkey, &v1, DbAction::Update, ); assert!(crate::db::is_cid_record_value( &head(&state, collection, rkey).unwrap() )); } #[test] fn protocol_delete_of_redacted_head_writes_only_a_tombstone() { let (_tmp, state) = test_state(); let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; write( &state, &rev("3kzbif5moe22m"), collection, rkey, &body("v1"), DbAction::Create, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::Head, ) .unwrap(); delete(&state, &rev("3kzbif5mof33m"), collection, rkey); assert_eq!(head(&state, collection, rkey), None); assert_eq!( history_bodies(&state, collection, rkey), vec![Vec::::new()] ); write( &state, &rev("3kzbif5mog44m"), collection, rkey, &body("v1"), DbAction::Create, ); assert!(crate::db::is_cid_record_value( &head(&state, collection, rkey).unwrap() )); } #[test] fn history_redaction_suppresses_that_version_when_it_returns_to_head() { let (_tmp, state) = test_state(); let collection = "app.bsky.feed.post"; let rkey = "3kzbif5moe22m"; let v1 = body("v1"); write( &state, &rev("3kzbif5moe22m"), collection, rkey, &v1, DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), collection, rkey, &body("v2"), DbAction::Update, ); crate::db::redact_record_bodies( &state.db, &did(), collection, &DbRkey::new(rkey), crate::types::DeleteBodyTarget::History, ) .unwrap(); write( &state, &rev("3kzbif5mog44m"), collection, rkey, &v1, DbAction::Update, ); assert!(crate::db::is_cid_record_value( &head(&state, collection, rkey).unwrap() )); } #[test] fn create_over_differing_head_preserves_it() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); // out-of-order replay: the old head still belongs in history write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v2"), DbAction::Create, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v2".to_vec()) ); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); } #[test] fn update_of_never_seen_record_just_writes_head() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Update, ); assert_eq!( head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), Some(b"v1".to_vec()) ); assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty()); } #[test] fn delete_of_never_seen_record_writes_empty_tombstone() { let (_tmp, state) = test_state(); delete( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", ); assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None); // an empty entry records "deleted at rev" even when there was no // body to preserve, distinguishing it from "never existed" assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![Vec::::new()] ); // never seen means nothing to decrement assert_eq!(state.db.get_count_sync("records"), 0); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 0 ); } #[test] fn replayed_create_keeps_counts_and_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); assert_eq!(state.db.get_count_sync("records"), 1); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 1 ); assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty()); } #[test] fn update_of_never_seen_record_counts_as_create() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Update, ); assert_eq!(state.db.get_count_sync("records"), 1); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 1 ); } #[test] fn update_of_existing_record_leaves_counts_alone() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v2"), DbAction::Update, ); assert_eq!(state.db.get_count_sync("records"), 1); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 1 ); } #[test] fn duplicate_delete_preserves_original_body() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); delete( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", ); // buffered replay applies the same delete again at the same rev delete( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.post", "3kzbif5moe22m", ); // the original body survives; no empty tombstone overwrites it assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); assert_eq!(state.db.get_count_sync("records"), 0); assert_eq!(state.db.get_count_sync("history"), 1); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 0 ); } #[test] fn duplicate_delete_of_never_seen_writes_one_tombstone() { let (_tmp, state) = test_state(); delete( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", ); delete( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", ); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![Vec::::new()] ); assert_eq!(state.db.get_count_sync("records"), 0); assert_eq!(state.db.get_count_sync("history"), 1); } #[test] fn claimed_cid_mismatch_is_rejected() { let (_tmp, state) = test_state(); let mut txn = Txn::new(&state.db); let did = did(); let mut record_txn = txn.records(&state, &rev("3kzbif5moe22m"), &did).unwrap(); let wrong_cid = jacquard_repo::mst::util::compute_cid(b"not the body").unwrap(); let err = record_txn .put_record( "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"), &wrong_cid, &body("v1"), DbAction::Create, ) .unwrap_err(); assert!(err.to_string().contains("does not match")); drop(record_txn); drop(txn); assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None); assert_eq!(state.db.get_count_sync("records"), 0); } #[test] fn delayed_commit_death_rev_uses_local_clock() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); // a delayed commit arrives with an old rev: the death must still be // keyed on the local clock so event retention and history retention // age out together let before = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_micros() as i64; write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v2"), DbAction::Update, ); let after = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_micros() as i64; let key = keys::record_key(&did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m")); let mut end = key.clone(); end.push(0xFF); let death_micros: Vec = state .db .indexer .history .range((Bound::Included(key.clone()), Bound::Excluded(end))) .filter_map(|guard| guard.into_inner().ok()) .filter(|(k, _)| k.len() == key.len() + 9 && k.starts_with(&key)) .map(|(k, _)| { let raw = u64::from_be_bytes(k[key.len() + 1..].try_into().unwrap()); (raw >> 10) as i64 }) .collect(); assert_eq!(death_micros.len(), 1); assert!( death_micros[0] >= before && death_micros[0] <= after + 1_000_000, "death rev {} should be on the local clock, not the old commit rev", death_micros[0] ); } #[test] fn repo_purge_moves_all_bodies_to_history() { let (_tmp, state) = test_state(); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.like", "3kzbif5mof33m", &body("v2"), DbAction::Create, ); let mut txn = Txn::new(&state.db); crate::db::delete_repo_records(&mut txn, &state.db, &did(), Some(&rev("3kzbif5mog44m"))) .unwrap(); txn.commit().unwrap(); assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None); assert_eq!(head(&state, "app.bsky.feed.like", "3kzbif5mof33m"), None); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); assert_eq!( history_bodies(&state, "app.bsky.feed.like", "3kzbif5mof33m"), vec![b"v2".to_vec()] ); } #[test] fn repo_purge_with_future_skew_root_sorts_death_strictly_after_root_and_updates_counts() { let (_tmp, state) = test_state(); let future_root_rev = rev("3kzbif5mog44m"); write( &state, &rev("3kzbif5moe22m"), "app.bsky.feed.post", "3kzbif5moe22m", &body("v1"), DbAction::Create, ); write( &state, &rev("3kzbif5mof33m"), "app.bsky.feed.like", "3kzbif5mof33m", &body("v2"), DbAction::Create, ); assert_eq!(state.db.get_count_sync("records"), 2); assert_eq!(state.db.get_count_sync("history"), 0); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 1 ); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.like").unwrap(), 1 ); let mut txn = Txn::new(&state.db); crate::db::delete_repo_records(&mut txn, &state.db, &did(), Some(&future_root_rev)) .unwrap(); txn.commit().unwrap(); // heads gone, bodies in history assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None); assert_eq!(head(&state, "app.bsky.feed.like", "3kzbif5mof33m"), None); assert_eq!( history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"), vec![b"v1".to_vec()] ); assert_eq!( history_bodies(&state, "app.bsky.feed.like", "3kzbif5mof33m"), vec![b"v2".to_vec()] ); // count assertions assert_eq!(state.db.get_count_sync("records"), 0); assert_eq!(state.db.get_count_sync("history"), 2); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 0 ); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.like").unwrap(), 0 ); // death rev sorting check: death key in history sorts strictly AFTER future_root_rev let post_key = keys::record_key(&did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m")); // `history_range_after` inline: (Excluded({key} 00 {rev}), Excluded({key} 01)) let lo = std::ops::Bound::Excluded(keys::history_key(&post_key, &future_root_rev)); let mut upper = post_key.clone(); upper.push(1); let hi = std::ops::Bound::Excluded(upper); let history_hit = state.db.indexer.history.range((lo, hi)).next(); assert!( history_hit.is_some(), "history entry death rev must sort strictly after future root rev so event replay resolves it" ); } }