gonzalo_core/
reset.rs

1//! Namespace / collection reset (spec §3.7, gonzalo#203).
2//!
3//! Reset tombstones every live record under a prefix by issuing ordinary
4//! OCC-guarded `delete` calls, so it replicates like any other delete and
5//! needs only `write` on the namespace. It is **not atomic**: no substrate
6//! offers multi-key transactions. It is **idempotent** instead. A second run
7//! tombstones whatever the first run missed (a key that lost a race) and
8//! skips everything already gone.
9
10use crate::{CoreError, DeleteResult, Identity, KeyPrefix, RecordKey, Result, Store};
11
12/// What a reset run did.
13#[derive(Clone, Debug, Default, PartialEq, Eq)]
14#[must_use = "a ResetReport may list conflicted keys that were not deleted"]
15pub struct ResetReport {
16    /// Keys tombstoned by this run, including any found already deleted at
17    /// delete time, in `list` order.
18    pub deleted: Vec<RecordKey>,
19    /// Keys edited between this run's `get` and its `delete`. Left live, not
20    /// retried. Run reset again to delete them.
21    pub conflicts: Vec<RecordKey>,
22}
23
24/// Tombstone every live record under `prefix`. `prefix.namespace` is required.
25/// Resetting every namespace in a store is not a reset, so a caller who wants
26/// that must loop over namespaces explicitly.
27///
28/// For each live key (consumer `list`), reads its current revision and issues
29/// `delete(key, Some(revision))`. A key that disappears between `list` and
30/// `get` is skipped. A `Conflict` goes into [`ResetReport::conflicts`].
31///
32/// Stops at the first store error and returns it; keys already tombstoned
33/// stay tombstoned, so a re-run is safe. Against a daemon whose backing store
34/// predates tombstones, deletes are physical and do not replicate (spec
35/// §8.1).
36pub async fn reset(store: &dyn Store, prefix: &KeyPrefix) -> Result<ResetReport> {
37    reset_as(store, prefix, None).await
38}
39
40/// As [`reset`], stamping `author` as each tombstone's deleter
41/// (`Store::delete_as`). `None` keeps each record's own author, exactly as
42/// [`reset`] does. Over a daemon the deleter follows spec §3.6: a non-admin
43/// token is stamped as itself.
44///
45/// Stops at the first store error and returns it; keys already tombstoned
46/// stay tombstoned, so a re-run is safe. Against a daemon whose backing store
47/// predates tombstones, deletes are physical and do not replicate (spec
48/// §8.1).
49pub async fn reset_as(
50    store: &dyn Store,
51    prefix: &KeyPrefix,
52    author: Option<Identity>,
53) -> Result<ResetReport> {
54    if prefix.namespace.is_none() {
55        return Err(CoreError::Backend("reset requires a namespace".into()));
56    }
57    let mut report = ResetReport::default();
58    for key in store.list(prefix).await? {
59        // Belt-and-braces: reset is destructive, so never act on a key a
60        // misbehaving store returned from outside the prefix.
61        if !prefix.matches(&key) {
62            continue;
63        }
64        let Some(current) = store.get(&key).await? else {
65            continue;
66        };
67        match store
68            .delete_as(&key, Some(current.revision), author.clone())
69            .await?
70        {
71            DeleteResult::Deleted => report.deleted.push(key),
72            DeleteResult::Conflict(_) => report.conflicts.push(key),
73        }
74    }
75    Ok(report)
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use crate::memstore::MemStore;
82    use crate::test_support::{Hook, HookedStore, rec, seed};
83    use std::sync::atomic::AtomicBool;
84
85    const CLOCK: i64 = 1_000;
86
87    fn ns(namespace: &str) -> KeyPrefix {
88        KeyPrefix {
89            namespace: Some(namespace.into()),
90            collection: None,
91        }
92    }
93
94    #[tokio::test]
95    async fn refuses_a_prefix_without_a_namespace() {
96        let store = MemStore::new().with_clock(CLOCK);
97        seed(&store, rec("ns", "col", "a", "x")).await;
98
99        let err = reset(
100            &store,
101            &KeyPrefix {
102                namespace: None,
103                collection: Some("col".into()),
104            },
105        )
106        .await
107        .unwrap_err();
108
109        assert!(
110            matches!(&err, CoreError::Backend(m) if m == "reset requires a namespace"),
111            "got {err:?}"
112        );
113        // Nothing was touched.
114        assert!(store.raw_snapshot().values().all(|r| !r.is_tombstone()));
115    }
116
117    #[tokio::test]
118    async fn tombstones_every_live_record_in_the_namespace_only() {
119        let store = MemStore::new().with_clock(CLOCK);
120        seed(&store, rec("ns", "a", "1", "x")).await;
121        seed(&store, rec("ns", "b", "2", "y")).await;
122        seed(&store, rec("other", "a", "3", "z")).await;
123
124        let report = reset(&store, &ns("ns")).await.unwrap();
125
126        assert_eq!(
127            report.deleted,
128            vec![
129                RecordKey::new("ns", "a", "1"),
130                RecordKey::new("ns", "b", "2")
131            ]
132        );
133        assert!(report.conflicts.is_empty());
134        for key in &report.deleted {
135            assert_eq!(store.get(key).await.unwrap(), None, "{key} hidden");
136            let raw = store.get_raw(key).await.unwrap().expect("tombstone kept");
137            assert!(raw.is_tombstone());
138            assert_eq!(raw.deleted_at, Some(CLOCK));
139        }
140        let other = RecordKey::new("other", "a", "3");
141        assert!(
142            store.get(&other).await.unwrap().is_some(),
143            "sibling namespace untouched"
144        );
145    }
146
147    #[tokio::test]
148    async fn collection_scope_leaves_sibling_collections_alone() {
149        let store = MemStore::new().with_clock(CLOCK);
150        seed(&store, rec("ns", "col", "1", "x")).await;
151        seed(&store, rec("ns", "sibling", "2", "y")).await;
152
153        let report = reset(
154            &store,
155            &KeyPrefix {
156                namespace: Some("ns".into()),
157                collection: Some("col".into()),
158            },
159        )
160        .await
161        .unwrap();
162
163        assert_eq!(report.deleted, vec![RecordKey::new("ns", "col", "1")]);
164        assert!(report.conflicts.is_empty());
165        assert!(
166            store
167                .get(&RecordKey::new("ns", "sibling", "2"))
168                .await
169                .unwrap()
170                .is_some()
171        );
172    }
173
174    #[tokio::test]
175    async fn already_deleted_keys_are_skipped_and_their_chain_does_not_advance() {
176        let store = MemStore::new().with_clock(CLOCK);
177        let gone = RecordKey::new("ns", "col", "gone");
178        seed(&store, rec("ns", "col", "gone", "x")).await;
179        seed(&store, rec("ns", "col", "live", "y")).await;
180        assert_eq!(
181            store.delete(&gone, None).await.unwrap(),
182            DeleteResult::Deleted
183        );
184        let before = store.get_raw(&gone).await.unwrap().unwrap().revision;
185
186        let report = reset(&store, &ns("ns")).await.unwrap();
187
188        assert_eq!(report.deleted, vec![RecordKey::new("ns", "col", "live")]);
189        assert_eq!(
190            store.get_raw(&gone).await.unwrap().unwrap().revision,
191            before
192        );
193    }
194
195    #[tokio::test]
196    async fn concurrent_edit_is_a_conflict_and_rerunning_is_idempotent() {
197        let raced = RecordKey::new("ns", "col", "b");
198        let store = HookedStore {
199            inner: MemStore::new().with_clock(CLOCK),
200            hook: Hook::EditBeforeDelete(raced.clone()),
201            fired: AtomicBool::new(false),
202        };
203        for id in ["a", "b", "c"] {
204            seed(&store, rec("ns", "col", id, id)).await;
205        }
206
207        // Run 1: the edit wins the race on `b`; reset reports it, doesn't retry.
208        let first = reset(&store, &ns("ns")).await.unwrap();
209        assert_eq!(
210            first.deleted,
211            vec![
212                RecordKey::new("ns", "col", "a"),
213                RecordKey::new("ns", "col", "c")
214            ]
215        );
216        assert_eq!(first.conflicts, vec![raced.clone()]);
217        let survivor = store.get(&raced).await.unwrap().expect("edit survives");
218        assert_eq!(survivor.body.bytes(), b"edited concurrently");
219
220        // Run 2: tombstones exactly what run 1 missed, and conflicts on nothing.
221        let second = reset(&store, &ns("ns")).await.unwrap();
222        assert_eq!(second.deleted, vec![raced.clone()]);
223        assert!(second.conflicts.is_empty());
224        assert_eq!(store.get(&raced).await.unwrap(), None);
225
226        // Run 3: nothing left to do.
227        let third = reset(&store, &ns("ns")).await.unwrap();
228        assert_eq!(third, ResetReport::default());
229    }
230
231    #[tokio::test]
232    async fn reset_as_stamps_the_deleter() {
233        let store = MemStore::new().with_clock(CLOCK);
234        let key = RecordKey::new("ns", "col", "a");
235        seed(&store, rec("ns", "col", "a", "x")).await;
236
237        let report = reset_as(&store, &ns("ns"), Some(Identity::new("x")))
238            .await
239            .unwrap();
240
241        assert_eq!(report.deleted, vec![key.clone()]);
242        let raw = store.get_raw(&key).await.unwrap().expect("tombstone kept");
243        assert!(raw.is_tombstone());
244        assert_eq!(raw.meta.author, Identity::new("x"));
245    }
246
247    #[tokio::test]
248    async fn a_key_that_vanishes_between_list_and_get_is_skipped() {
249        let vanished = RecordKey::new("ns", "col", "b");
250        let store = HookedStore {
251            inner: MemStore::new().with_clock(CLOCK),
252            hook: Hook::VanishBeforeGet(vanished),
253            fired: AtomicBool::new(false),
254        };
255        seed(&store, rec("ns", "col", "a", "x")).await;
256        seed(&store, rec("ns", "col", "b", "y")).await;
257
258        let report = reset(&store, &ns("ns")).await.unwrap();
259
260        assert_eq!(report.deleted, vec![RecordKey::new("ns", "col", "a")]);
261        assert!(report.conflicts.is_empty());
262    }
263}
gonzalo_core/reset.rs

gonzalo_core/
reset.rs
