use 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(())
}
}