From 9467e445ede52bfa1c3a24e8b1c06ce0f9c6ff6d Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 7 Oct 2026 00:41:31 +0300 Subject: [PATCH] [stats] report history as the keyspace's approximate size counts.history was a durable write counter. the history retention filter destroys expired entries during compaction with no way to decrement it, so with HYDRANT_HISTORY_TTL set it only ever grew while the keyspace stayed bounded. it also counted a history key twice when a second version was replaced at the same death rev (a commit rev ahead of the local clock), which no decrement could undo. /stats now reads counts.history from the history keyspace's table and memtable metadata, the same way events and relay_events are already counted, so retention, repo erase and overwrites all show up once compaction reaches them. the counter and its per commit delta write are gone, which takes one count delta row off every commit that writes history. nothing outside hydrant reads the field, and the name stays the same. the old persisted counter key stays behind in existing databases but nothing reads it. history_count_drops_what_retention_expires and history_count_follows_overwrites_and_purges go through Hydrant::stats. against the old counter code the first reports 2 after the entries expired and were compacted (want 0) and the second reports 2 after a same key overwrite (want 1). docs/concepts/retention.md says what the number means, and HISTORY_TTL gets its missing row in docs/configuration.md. fixes hydrant-2xc and hydrant-aze. --- docs/concepts/README.md | 1 + docs/concepts/retention.md | 13 ++++ docs/configuration.md | 1 + src/control/stats.rs | 126 +++++++++++++++++++++++++++++++++++-- src/db/counts.rs | 5 -- src/db/indexer.rs | 15 +---- src/db/keyspaces.rs | 6 ++ src/db/txn.rs | 14 ++--- 8 files changed, 148 insertions(+), 33 deletions(-) create mode 100644 docs/concepts/retention.md diff --git a/docs/concepts/README.md b/docs/concepts/README.md index 8ba1d89..bb9e6a8 100644 --- a/docs/concepts/README.md +++ b/docs/concepts/README.md @@ -4,3 +4,4 @@ title: concepts - [hydrant vs tap](vs-tap.md): design comparison, stream behavior - [relay & seeding](relay.md): multi-relay support, firehose seeding, crawler sources +- [retention](retention.md): how event and record history ttls behave, and what `/stats` counts under them diff --git a/docs/concepts/retention.md b/docs/concepts/retention.md new file mode 100644 index 0000000..0325917 --- /dev/null +++ b/docs/concepts/retention.md @@ -0,0 +1,13 @@ +--- +title: retention +--- + +hydrant keeps two things on a clock: stream events, bounded by `EPHEMERAL_TTL`, and in permanent databases the replaced record versions in `history`, bounded by `HISTORY_TTL`. + +## record history + +when a record is updated or deleted, its previous body moves to the `history` keyspace, keyed by the record and the time it was replaced, so replaying an older event can still show the version it was about. a delete of a record hydrant never saw leaves an empty entry, so "deleted" and "never existed" stay different. + +without `HISTORY_TTL` history is kept forever. with it, entries older than the ttl are dropped by a compaction filter, so they go when compaction next rewrites their part of the keyspace, not right at the deadline. replaying an event whose version was trimmed gives the event without its record body, and `/stats` counts those as `counts.history_trim_misses`. + +`counts.history` in `/stats` is approximate. it's read from the storage engine's table metadata rather than kept as a counter, so it does go down when retention or a repo erase removes entries, but only once compaction has rewritten the tables holding them. until then removed and overwritten entries still count. read it as about how many versions are on disk, not an exact number. diff --git a/docs/configuration.md b/docs/configuration.md index fd6c8f8..2a4b29e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -23,6 +23,7 @@ hydrant is configured via environment variables, all prefixed with `HYDRANT_` (e | `FULL_NETWORK` | `false` (indexer), `true` (relay) | if `true`, discover and index all repos in the network | | `EPHEMERAL` | `false` (indexer), `true` (relay) | if enabled, no records are stored (in indexer mode). events are deleted after a certain duration (`EPHEMERAL_TTL`). immutable per database: the setting is persisted on first open and a later flip is rejected at startup — keep the previous setting or use a fresh database path | | `EPHEMERAL_TTL` | `60min`, `3d` (relay) | how long to keep events before deletion. when built with `jetstream`, retained Jetstream replay metadata is pruned on the same schedule | +| `HISTORY_TTL` | unset | indexer only, permanent databases only. how long replaced record versions stay in `history` so replays of older events can still show them. humantime duration like `2h`, unset keeps them forever. see [retention](concepts/retention.md) | | `ONLY_INDEX_LINKS` | `false` | indexer only. if enabled, record blocks are not stored, only the index (records, counts, events) is kept. `getRecord`, `listRecords`, and `getRepo` will return errors. the event stream and Jetstream stream still work, but create/update events will not include record values. immutable per database: the setting is persisted on first open and a later flip is rejected at startup — keep the previous setting or use a fresh database path | ## filter diff --git a/src/control/stats.rs b/src/control/stats.rs index 666890b..2e0c71b 100644 --- a/src/control/stats.rs +++ b/src/control/stats.rs @@ -15,7 +15,10 @@ impl Hydrant { /// return database counts and on-disk sizes for all keyspaces. /// /// counts include: `repos`, `pending`, `resync`, `records`, `history`, `events`, - /// `error_ratelimited`, `error_transport`, `error_generic`. + /// `error_ratelimited`, `error_transport`, `error_generic`. `history`, `events` + /// and the relay and jetstream event counts are approximate: they come from + /// the keyspace's table metadata, so deleted or expired entries still count + /// until compaction drops them. /// /// sizes are in bytes, reported per keyspace. pub async fn stats(&self) -> Result { @@ -33,14 +36,10 @@ impl Hydrant { { count_keys.push("pending"); count_keys.push("records"); - count_keys.push("history"); count_keys.push("resync"); } - #[cfg_attr( - not(any(feature = "indexer_stream", feature = "relay")), - allow(unused_mut) - )] + #[cfg_attr(not(any(feature = "indexer", feature = "relay")), allow(unused_mut))] let mut counts: BTreeMap<&'static str, u64> = futures::future::join_all(count_keys.into_iter().map(|name| { let state = state.clone(); @@ -50,6 +49,14 @@ impl Hydrant { .into_iter() .collect(); + // a write counter can't see what the retention filter drops, so this + // reads the keyspace itself + #[cfg(feature = "indexer")] + counts.insert( + "history", + state.db.indexer.approximate_history_count() as u64, + ); + #[cfg(feature = "indexer_stream")] counts.insert("events", state.db.stream.approximate_event_count() as u64); @@ -89,3 +96,110 @@ impl Hydrant { Ok(StatsResponse { counts, sizes }) } } + +#[cfg(all(test, feature = "indexer"))] +mod tests { + use super::*; + use crate::config::Config; + use crate::db::types::{DbAction, DbRkey, DbTid}; + use crate::state::AppState; + use jacquard_common::types::string::Did; + use std::time::Duration; + + const COLLECTION: &str = "app.bsky.feed.post"; + + fn did() -> Did { + Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap() + } + + fn rev_at(micros: i64) -> DbTid { + DbTid::new_from_bytes((((micros as u64) << 10) | 0x2A).to_be_bytes()) + } + + async fn hydrant(dir: &std::path::Path, history_ttl: Option) -> Result { + Hydrant::new(Config { + database_path: dir.to_path_buf(), + history_ttl, + ..Config::default() + }) + .await + } + + fn put(state: &AppState, rev: &DbTid, rkey: &str, body: &'static str) -> Result<()> { + let did = did(); + let mut txn = crate::db::Txn::new(&state.db); + let mut records = txn.records(state, rev, &did)?; + let block = crate::car::CarBlock::from_body(body.as_bytes()); + records.put_record(COLLECTION, &DbRkey::new(rkey), &block, DbAction::Update)?; + records.finish()?; + txn.commit() + } + + fn delete(state: &AppState, rev: &DbTid, rkey: &str) -> Result<()> { + let did = did(); + let mut txn = crate::db::Txn::new(&state.db); + let mut records = txn.records(state, rev, &did)?; + records.delete_record(COLLECTION, &DbRkey::new(rkey))?; + records.finish()?; + txn.commit() + } + + fn compact_history(state: &AppState) -> Result<()> { + let history = state + .db + .keyspace_by_name("history") + .ok_or_else(|| miette::miette!("no history keyspace"))?; + history.rotate_memtable_and_wait().into_diagnostic()?; + history.major_compact().into_diagnostic() + } + + async fn history_count(hydrant: &Hydrant) -> Result { + Ok(hydrant.stats().await?.counts["history"]) + } + + #[tokio::test] + async fn history_count_drops_what_retention_expires() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let hydrant = hydrant(tmp.path(), Some(Duration::from_secs(1))).await?; + let state = &hydrant.state; + let now = chrono::Utc::now().timestamp_micros(); + + put(state, &rev_at(now), "3kzbif5moe22m", "v1")?; + put(state, &rev_at(now + 1), "3kzbif5moe22m", "v2")?; + delete(state, &rev_at(now + 2), "3kzbif5moe22m")?; + assert_eq!(history_count(&hydrant).await?, 2); + + tokio::time::sleep(Duration::from_millis(1100)).await; + compact_history(state)?; + assert_eq!(history_count(&hydrant).await?, 0); + Ok(()) + } + + #[tokio::test] + async fn history_count_follows_overwrites_and_purges() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let hydrant = hydrant(tmp.path(), None).await?; + let state = &hydrant.state; + + // a rev ahead of the local clock is its own death rev, so a second + // version replaced at that rev lands on the same history key + let ahead = rev_at(chrono::Utc::now().timestamp_micros() + 3_600_000_000); + put(state, &ahead, "3kzbif5moe22m", "v1")?; + put(state, &ahead, "3kzbif5moe22m", "v2")?; + put(state, &ahead, "3kzbif5moe22m", "v3")?; + compact_history(state)?; + assert_eq!(history_count(&hydrant).await?, 1); + + put(state, &ahead, "3kzbif5mof33m", "other")?; + let mut txn = crate::db::Txn::new(&state.db); + crate::db::delete_repo_records(&mut txn, &state.db, &did(), None)?; + txn.commit()?; + compact_history(state)?; + assert_eq!(history_count(&hydrant).await?, 3); + + crate::db::erase_repo(&state.db, &did())?; + compact_history(state)?; + assert_eq!(history_count(&hydrant).await?, 0); + Ok(()) + } +} diff --git a/src/db/counts.rs b/src/db/counts.rs index 487a60a..d1ab0dc 100644 --- a/src/db/counts.rs +++ b/src/db/counts.rs @@ -37,11 +37,6 @@ impl CountDeltas { self.add("records", delta); } - #[cfg(feature = "indexer")] - pub(crate) fn add_history(&mut self, delta: i64) { - self.add("history", delta); - } - pub(crate) fn add_pds_account(&mut self, host: &str, delta: i64) { self.add(&keys::pds_account_count_key(host), delta); } diff --git a/src/db/indexer.rs b/src/db/indexer.rs index fbdb8f8..d6faedd 100644 --- a/src/db/indexer.rs +++ b/src/db/indexer.rs @@ -307,15 +307,9 @@ pub(crate) fn erase_repo(db: &Db, did: &Did) -> Result { report.heads_redacted = count; report.body_bytes_redacted += bytes; - let (count, bytes) = erase_prefix( - db, - db.indexer.history.raw(), - &prefix, - |_, counts, _, value| { - counts.add_history(-1); - Ok(body_len(value)) - }, - )?; + let (count, bytes) = erase_prefix(db, db.indexer.history.raw(), &prefix, |_, _, _, value| { + Ok(body_len(value)) + })?; report.history_redacted = count; report.body_bytes_redacted += bytes; @@ -440,7 +434,6 @@ pub(crate) fn delete_repo_records( }; let mut moved_heads: i64 = 0; - let mut moved_bodies: i64 = 0; let mut collection_deltas: BTreeMap = BTreeMap::new(); let prefix = keys::record_prefix_did(did); @@ -457,7 +450,6 @@ pub(crate) fn delete_repo_records( // resolve it. in links-only mode the value is a cid, not a body, but // that mode's replay cannot resolve records either way if !value.is_empty() { - moved_bodies += 1; txn.batch.insert( &db.indexer.history, keys::history_key(&key, &death_rev), @@ -474,7 +466,6 @@ pub(crate) fn delete_repo_records( // deferred to the txn's commit so the in-memory projection never leads // the durable batch txn.counts.add_records(-moved_heads); - txn.counts.add_history(moved_bodies); Ok(()) } diff --git a/src/db/keyspaces.rs b/src/db/keyspaces.rs index 0d71a63..bd3ffbd 100644 --- a/src/db/keyspaces.rs +++ b/src/db/keyspaces.rs @@ -122,6 +122,12 @@ impl IndexerDb { self.history.range(range) } + /// read from table and memtable metadata, so entries that expired, were + /// erased or overwritten still count until compaction reaches them + pub(crate) fn approximate_history_count(&self) -> usize { + self.history.approximate_len() + } + #[cfg(any(test, feature = "indexer_stream"))] pub(crate) fn stage_record(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where diff --git a/src/db/txn.rs b/src/db/txn.rs index c8a4d03..8e5ce50 100644 --- a/src/db/txn.rs +++ b/src/db/txn.rs @@ -218,7 +218,6 @@ impl<'db> Txn<'db> { only_index_links: state.only_index_links, count_ephemeral_records, records_delta: 0, - history_count: 0, collection_deltas: HashMap::new(), excluded, }) @@ -418,7 +417,6 @@ pub(crate) struct RecordTxn<'txn, 'db, 'did> { only_index_links: bool, count_ephemeral_records: bool, records_delta: i64, - history_count: i64, collection_deltas: HashMap, excluded: bool, } @@ -519,7 +517,6 @@ impl RecordTxn<'_, '_, '_> { keys::history_key(&key, &self.death_rev()), old.as_ref(), ); - self.history_count += 1; } self.txn .batch @@ -643,7 +640,6 @@ impl RecordTxn<'_, '_, '_> { keys::history_key(&key, &self.death_rev()), history_value, ); - self.history_count += 1; } self.txn.batch.remove(&self.txn.db.indexer.records, key); } @@ -657,7 +653,6 @@ impl RecordTxn<'_, '_, '_> { keys::history_key(&key, &self.death_rev()), [], ); - self.history_count += 1; } } } @@ -706,7 +701,6 @@ impl RecordTxn<'_, '_, '_> { )?; } self.txn.counts.add_records(self.records_delta); - self.txn.counts.add_history(self.history_count); Ok(()) } } @@ -1482,7 +1476,7 @@ mod tests { 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!(state.db.indexer.approximate_history_count(), 1); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 0 @@ -1510,7 +1504,7 @@ mod tests { vec![Vec::::new()] ); assert_eq!(state.db.get_count_sync("records"), 0); - assert_eq!(state.db.get_count_sync("history"), 1); + assert_eq!(state.db.indexer.approximate_history_count(), 1); } #[test] @@ -1654,7 +1648,7 @@ mod tests { ); assert_eq!(state.db.get_count_sync("records"), 2); - assert_eq!(state.db.get_count_sync("history"), 0); + assert_eq!(state.db.indexer.approximate_history_count(), 0); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 1 @@ -1683,7 +1677,7 @@ mod tests { // count assertions assert_eq!(state.db.get_count_sync("records"), 0); - assert_eq!(state.db.get_count_sync("history"), 2); + assert_eq!(state.db.indexer.approximate_history_count(), 2); assert_eq!( crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(), 0 -- 2.51.2