gonzalo_core/
collect.rs

1//! Tombstone collection (spec §3.8, gonzalo#203).
2//!
3//! A tombstone is what stops a deleted record from resurrecting on the next
4//! sync, so purging one is only safe once every peer has synced past the
5//! delete. Only the operator knows how long that takes. Collection is
6//! therefore explicit, never automatic, and takes the horizon as a required
7//! argument with no default. Not to be confused with [`gc_blobs`](crate::gc_blobs),
8//! which sweeps orphaned code-graph slices.
9
10use crate::{DeleteResult, KeyPrefix, RecordKey, Result, Store};
11use std::time::Duration;
12
13/// What a collection run did.
14#[derive(Clone, Debug, Default, PartialEq, Eq)]
15#[must_use = "a CollectReport may list conflicts (keys recreated during collection)"]
16pub struct CollectReport {
17    /// Tombstones purged by this run, including any found already gone when
18    /// purged.
19    pub purged: Vec<RecordKey>,
20    /// Tombstones kept because they have no `deleted_at`.
21    pub unstamped: usize,
22    /// Purge lost an OCC race (the key was recreated during collection).
23    pub conflicts: Vec<RecordKey>,
24}
25
26/// Purge tombstones under `prefix` whose `deleted_at` is at least `horizon` old.
27///
28/// Live records are never touched. A tombstone with no `deleted_at` is kept
29/// and counted in [`CollectReport::unstamped`]. A future-dated `deleted_at`
30/// (clock skew) gives a negative age and is kept, so skew can delay collection
31/// but can't trigger it early. A horizon longer than any representable age
32/// purges nothing. `now_ms` is a parameter so tests control time; the CLI
33/// passes the system clock.
34///
35/// Stops at the first store error and returns it; tombstones already purged
36/// stay purged, so a re-run is safe. Over a daemon, purge needs an admin
37/// token: a non-admin token fails on the first eligible tombstone, before
38/// anything is purged. A daemon predating replication fails on `list_raw`
39/// with `DAEMON_PREDATES_REPLICATION`.
40pub async fn collect(
41    store: &dyn Store,
42    prefix: &KeyPrefix,
43    horizon: Duration,
44    now_ms: i64,
45) -> Result<CollectReport> {
46    let horizon_ms = i128::try_from(horizon.as_millis()).unwrap_or(i128::MAX);
47    let mut report = CollectReport::default();
48    for key in store.list_raw(prefix).await? {
49        if !prefix.matches(&key) {
50            continue;
51        }
52        let Some(record) = store.get_raw(&key).await? else {
53            continue;
54        };
55        if !record.is_tombstone() {
56            continue;
57        }
58        let Some(deleted_at) = record.deleted_at else {
59            report.unstamped += 1;
60            continue;
61        };
62        if i128::from(now_ms) - i128::from(deleted_at) < horizon_ms {
63            continue;
64        }
65        match store.purge(&key, record.revision).await? {
66            DeleteResult::Deleted => report.purged.push(key),
67            DeleteResult::Conflict(_) => report.conflicts.push(key),
68        }
69    }
70    Ok(report)
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use crate::PutResult;
77    use crate::memstore::MemStore;
78    use crate::test_support::{Hook, HookedStore, rec, seed};
79    use crate::tombstone::{DEFAULT_ANCESTOR_CAP, tombstone_of};
80    use std::sync::atomic::AtomicBool;
81    use std::time::Duration;
82
83    const DAY_MS: i64 = 86_400_000;
84    const NOW: i64 = 100 * DAY_MS;
85    const THIRTY_DAYS: Duration = Duration::from_secs(30 * 86_400);
86
87    /// Store a tombstone for `ns/col/id` directly, with an arbitrary
88    /// `deleted_at`, the way sync would replicate one into an empty store.
89    async fn seed_tombstone(
90        store: &dyn Store,
91        ns: &str,
92        col: &str,
93        id: &str,
94        deleted_at: Option<i64>,
95    ) {
96        let live = rec(ns, col, id, "was live");
97        let mut tomb = tombstone_of(&live, 0, DEFAULT_ANCESTOR_CAP, None);
98        tomb.deleted_at = deleted_at;
99        // `put_raw` is the replication write: stores the tombstone unchanged.
100        assert!(matches!(
101            store.put_raw(tomb, None).await.unwrap(),
102            PutResult::Committed(_)
103        ));
104    }
105
106    #[tokio::test]
107    async fn purges_only_stamped_tombstones_at_least_horizon_old() {
108        let store = MemStore::new();
109        seed_tombstone(&store, "ns", "col", "old", Some(NOW - 31 * DAY_MS)).await;
110        seed_tombstone(&store, "ns", "col", "exact", Some(NOW - 30 * DAY_MS)).await;
111        seed_tombstone(&store, "ns", "col", "young", Some(NOW - DAY_MS)).await;
112        seed_tombstone(&store, "ns", "col", "future", Some(NOW + DAY_MS)).await;
113        seed_tombstone(&store, "ns", "col", "unstamped", None).await;
114        seed(&store, rec("ns", "col", "live", "content")).await;
115        let live_key = RecordKey::new("ns", "col", "live");
116        let live_before = store.raw_snapshot()[&live_key].clone();
117
118        let report = collect(&store, &KeyPrefix::default(), THIRTY_DAYS, NOW)
119            .await
120            .unwrap();
121
122        assert_eq!(
123            report.purged,
124            vec![
125                RecordKey::new("ns", "col", "exact"),
126                RecordKey::new("ns", "col", "old"),
127            ]
128        );
129        assert_eq!(report.unstamped, 1);
130        assert!(report.conflicts.is_empty());
131
132        let after = store.raw_snapshot();
133        assert!(!after.contains_key(&RecordKey::new("ns", "col", "old")));
134        assert!(!after.contains_key(&RecordKey::new("ns", "col", "exact")));
135        for kept in ["young", "future", "unstamped"] {
136            let key = RecordKey::new("ns", "col", kept);
137            assert!(after[&key].is_tombstone(), "{kept} must be kept");
138        }
139        assert_eq!(after[&live_key], live_before, "live record untouched");
140    }
141
142    #[tokio::test]
143    async fn respects_the_prefix() {
144        let store = MemStore::new();
145        let old = Some(NOW - 60 * DAY_MS);
146        seed_tombstone(&store, "ns", "col", "in", old).await;
147        seed_tombstone(&store, "ns", "sibling", "out1", old).await;
148        seed_tombstone(&store, "other", "col", "out2", old).await;
149
150        let report = collect(
151            &store,
152            &KeyPrefix {
153                namespace: Some("ns".into()),
154                collection: Some("col".into()),
155            },
156            THIRTY_DAYS,
157            NOW,
158        )
159        .await
160        .unwrap();
161
162        assert_eq!(report.purged, vec![RecordKey::new("ns", "col", "in")]);
163        let after = store.raw_snapshot();
164        assert!(after.contains_key(&RecordKey::new("ns", "sibling", "out1")));
165        assert!(after.contains_key(&RecordKey::new("other", "col", "out2")));
166    }
167
168    #[tokio::test]
169    async fn purges_a_tombstone_written_by_delete() {
170        let store = MemStore::new().with_clock(NOW - 40 * DAY_MS);
171        let key = RecordKey::new("ns", "col", "k");
172        seed(&store, rec("ns", "col", "k", "x")).await;
173        assert_eq!(
174            store.delete(&key, None).await.unwrap(),
175            DeleteResult::Deleted
176        );
177
178        let report = collect(&store, &KeyPrefix::default(), THIRTY_DAYS, NOW)
179            .await
180            .unwrap();
181
182        assert_eq!(report.purged, vec![key.clone()]);
183        assert_eq!(store.get_raw(&key).await.unwrap(), None);
184        assert!(
185            store
186                .list_raw(&KeyPrefix::default())
187                .await
188                .unwrap()
189                .is_empty()
190        );
191    }
192
193    #[tokio::test]
194    async fn a_horizon_too_large_for_i64_millis_purges_nothing() {
195        let store = MemStore::new();
196        seed_tombstone(&store, "ns", "col", "ancient", Some(i64::MIN / 2)).await;
197        seed_tombstone(&store, "ns", "col", "floor", Some(i64::MIN)).await;
198
199        let report = collect(&store, &KeyPrefix::default(), Duration::MAX, NOW)
200            .await
201            .unwrap();
202
203        assert!(report.purged.is_empty());
204        assert_eq!(store.raw_snapshot().len(), 2);
205    }
206
207    #[tokio::test]
208    async fn recreation_during_collection_is_a_conflict_and_the_live_record_survives() {
209        let key = RecordKey::new("ns", "col", "k");
210        let store = HookedStore {
211            inner: MemStore::new(),
212            hook: Hook::RecreateBeforePurge(key.clone()),
213            fired: AtomicBool::new(false),
214        };
215        seed_tombstone(&store, "ns", "col", "k", Some(NOW - 60 * DAY_MS)).await;
216        let tomb_rev = store.get_raw(&key).await.unwrap().unwrap().revision;
217
218        let report = collect(&store, &KeyPrefix::default(), THIRTY_DAYS, NOW)
219            .await
220            .unwrap();
221
222        assert!(report.purged.is_empty());
223        assert_eq!(report.conflicts, vec![key.clone()]);
224        let live = store
225            .get(&key)
226            .await
227            .unwrap()
228            .expect("recreated record survives");
229        assert_eq!(live.body.bytes(), b"recreated");
230        assert_eq!(live.revision.counter, tomb_rev.counter + 1);
231    }
232}
gonzalo_core/collect.rs

gonzalo_core/
collect.rs
