Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225use crate::db::keys;use fjall::compaction::filter::Context;use lsm_tree::compaction::{CompactionFilter, Factory};use lsm_tree::compaction::{ItemAccessor, Verdict};use std::sync::Arc;use std::sync::OnceLock;use std::sync::atomic::{AtomicU64, Ordering};
pub struct CountsGcFilterFactory { pub delta_gc_watermark: Arc<AtomicU64>,}
struct CountsGcFilter { delta_gc_watermark: Arc<AtomicU64>,}
impl CompactionFilter for CountsGcFilter { fn filter_item(&mut self, item: ItemAccessor<'_>, _: &Context) -> lsm_tree::Result<Verdict> { let key = item.key();
if key.starts_with(keys::COUNT_DELTA_PREFIX) && let Ok((id, _)) = keys::parse_count_delta_key(key) && id <= self.delta_gc_watermark.load(Ordering::Relaxed) { return Ok(Verdict::Destroy); }
Ok(Verdict::Keep) }}
impl Factory for CountsGcFilterFactory { fn name(&self) -> &str { "counts_gc" }
fn make_filter(&self, _: &Context) -> Box<dyn CompactionFilter> { Box::new(CountsGcFilter { delta_gc_watermark: self.delta_gc_watermark.clone(), }) }}
/// drops history entries whose death rev is older than the configured/// retention window. the rev is a TID in the key tail, so the verdict is/// purely key-local: age is decodable without touching the value, and a/// filter on `history` can never see a head, so there is nothing to/// disambiguate. reclamation is lazy - it lands when compaction next covers/// the range.pub struct HistoryRetentionFilterFactory { pub ttl: Arc<OnceLock<Option<std::time::Duration>>>,}
struct HistoryRetentionFilter { cutoff_micros: Option<i64>,}
/// decodes the timestamp micros out of the trailing 8-byte big-endian TID:/// 53 bits of micros over 10 bits of clock id (jacquard encodes/// `(micros << 10) | clkid`; shifting by 11 would halve the timestamp and/// make fresh entries look decades old)fn tid_micros(key: &[u8]) -> Option<i64> { let rev: [u8; 8] = key.get(key.len().checked_sub(8)?..)?.try_into().ok()?; Some((u64::from_be_bytes(rev) >> 10) as i64)}
fn retention_cutoff_micros(now_micros: i64, ttl: std::time::Duration) -> i64 { let ttl_micros = i64::try_from(ttl.as_micros()).unwrap_or(i64::MAX); now_micros.saturating_sub(ttl_micros)}
impl CompactionFilter for HistoryRetentionFilter { fn filter_item(&mut self, item: ItemAccessor<'_>, _: &Context) -> lsm_tree::Result<Verdict> { match (tid_micros(item.key()), self.cutoff_micros) { (Some(micros), Some(cutoff_micros)) if micros < cutoff_micros => Ok(Verdict::Destroy), _ => Ok(Verdict::Keep), } }}
impl Factory for HistoryRetentionFilterFactory { fn name(&self) -> &str { "history_retention" }
fn make_filter(&self, _: &Context) -> Box<dyn CompactionFilter> { let cutoff_micros = self .ttl .get() .copied() .flatten() .map(|ttl| retention_cutoff_micros(chrono::Utc::now().timestamp_micros(), ttl)); Box::new(HistoryRetentionFilter { cutoff_micros }) }}
#[cfg(all(test, feature = "indexer"))]mod tests { use super::*; use miette::{IntoDiagnostic, Result}; use std::time::Duration;
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
fn tid_at(micros: i64) -> crate::db::types::DbTid { // same layout jacquard uses: (micros << 10) | clkid crate::db::types::DbTid::new_from_bytes((((micros as u64) << 10) | 0x2A).to_be_bytes()) }
#[test] fn tid_micros_decodes_real_jacquard_tids() { // regression: the filter must decode tids produced by jacquard, not // by a private encoder that could drift from it let tid = jacquard_common::types::string::Tid::now_0(); let db_tid = crate::db::types::DbTid::from(&tid); let mut key = b"record-key".to_vec(); key.extend_from_slice(db_tid.as_bytes()); let decoded = tid_micros(&key).unwrap(); let expected = chrono::Utc::now().timestamp_micros(); assert!( (decoded - expected).abs() < 5_000_000, "decoded {decoded} should be within 5s of now ({expected})" ); }
#[test] fn tid_micros_decodes_the_key_tail() { let micros = 1_754_000_000_123_456i64; let mut key = b"record-key".to_vec(); key.extend_from_slice((((micros as u64) << 10) | 0x3FF).to_be_bytes().as_ref()); assert_eq!(tid_micros(&key), Some(micros)); assert_eq!(tid_micros(b"short"), None); assert_eq!(tid_micros(b""), None); }
#[test] fn huge_ttl_cannot_wrap_the_cutoff_into_the_future() { let now = chrono::Utc::now().timestamp_micros(); assert!(retention_cutoff_micros(now, Duration::MAX) < 0); }
fn record_key() -> Result<Vec<u8>> { Ok(crate::db::keys::record_key( &jacquard_common::types::string::Did::new_static(DID).into_diagnostic()?, "app.bsky.feed.post", &crate::db::types::DbRkey::new("3kzbif5moe22m"), )) }
fn db_with_history_ttl(ttl: Option<Duration>, tmp: &std::path::Path) -> Result<crate::db::Db> { let mut cfg = crate::config::Config::default(); cfg.database_path = tmp.to_path_buf(); cfg.history_ttl = ttl; crate::db::Db::open(&cfg) }
#[test] fn retention_filter_drops_expired_history_entries() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let db = db_with_history_ttl(Some(Duration::from_secs(3600)), tmp.path())?;
let now_micros = chrono::Utc::now().timestamp_micros(); let expired = tid_at(now_micros - 2 * 3600 * 1_000_000); let fresh = tid_at(now_micros); let key = record_key()?;
let mut batch = db.inner.batch(); db.indexer.stage_history( &mut batch, crate::db::keys::history_key(&key, &expired), b"old".as_slice(), ); db.indexer.stage_history( &mut batch, crate::db::keys::history_key(&key, &fresh), b"new".as_slice(), ); batch.commit().into_diagnostic()?;
db.indexer .history .rotate_memtable_and_wait() .into_diagnostic()?; db.indexer.history.major_compact().into_diagnostic()?;
let entries: Vec<_> = db .indexer .history .iter() .filter_map(|g| g.into_inner().ok()) .collect(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].1.as_ref(), b"new");
Ok(()) }
#[test] fn without_retention_config_history_survives_compaction() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let db = db_with_history_ttl(None, tmp.path())?;
let ancient = tid_at(1_000_000_000); let key = record_key()?;
let mut batch = db.inner.batch(); db.indexer.stage_history( &mut batch, crate::db::keys::history_key(&key, &ancient), b"old".as_slice(), ); batch.commit().into_diagnostic()?;
db.indexer .history .rotate_memtable_and_wait() .into_diagnostic()?; db.indexer.history.major_compact().into_diagnostic()?;
assert_eq!(db.indexer.history.iter().count(), 1);
Ok(()) }}