vcon_server/store/
examine.rs

1//! What `check` reads of one tenant's Store: every kept version opened,
2//! the files no row names, versions due for purge, and purges since the
3//! tenant key was last rotated. Reading changes nothing.
4
5use std::collections::BTreeSet;
6
7use time::OffsetDateTime;
8use time::format_description::well_known::Rfc3339;
9
10use super::{Kept, Store, StoreError, Unopened, due, storage_failed};
11use crate::backend::ObjectError;
12use crate::clock::from_nanos;
13use crate::ids::VersionId;
14use crate::retention::Retention;
15
16/// One Store as `check` found it.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct Examination {
19    /// Every kept version, live or hidden.
20    pub versions: u64,
21    /// Those packed into pack files.
22    pub packed: u64,
23    /// Versions whose object file is missing or does not open.
24    pub damaged: Vec<VersionId>,
25    /// Files under `objects/` that are a version's, or its temporary, and
26    /// that no row names: what start-up recovery removes.
27    pub leftovers: Vec<String>,
28    /// Versions due for purge now.
29    pub due: u64,
30    /// When the longest-waiting of them fell due.
31    pub oldest_due_since: Option<OffsetDateTime>,
32    /// Purges recorded since `key_since`: their keys are destroyed only by
33    /// the next rotation.
34    pub purges_since_key: u64,
35    /// When the oldest of those purges was made.
36    pub oldest_purge_since_key: Option<OffsetDateTime>,
37}
38
39/// Why a Store could not be examined.
40#[derive(Debug, Clone, Copy, PartialEq, Eq)]
41pub enum ExamineError {
42    /// The storage failed to read a version's object or pack file, or to
43    /// list the objects: the error's kind and the system's error number,
44    /// as the server logs them. The error's text is not kept, since a
45    /// service's message may name an object's key.
46    Objects {
47        kind: std::io::ErrorKind,
48        os_error: Option<i32>,
49    },
50    /// Anything else, such as a catalog read that failed.
51    Failed(StoreError),
52}
53
54impl From<StoreError> for ExamineError {
55    fn from(e: StoreError) -> ExamineError {
56        ExamineError::Failed(e)
57    }
58}
59
60impl ExamineError {
61    /// A failure of the storage, logged as a read's is.
62    fn storage(e: &std::io::Error) -> ExamineError {
63        storage_failed(e);
64        ExamineError::Objects {
65            kind: e.kind(),
66            os_error: e.raw_os_error(),
67        }
68    }
69}
70
71/// A listing of the objects that failed: the storage's failure, unless it
72/// says there is nothing of the name, which a listing never means.
73fn unlisted(e: ObjectError) -> ExamineError {
74    match e {
75        ObjectError::Io(e) => ExamineError::storage(&e),
76        ObjectError::Missing => ExamineError::Failed(StoreError::Io),
77    }
78}
79
80impl Store {
81    /// Examines the Store under `retention` at `now`, counting purges made
82    /// since `key_since`.
83    pub fn examine(
84        &self,
85        retention: Retention,
86        key_since: OffsetDateTime,
87        now: OffsetDateTime,
88    ) -> Result<Examination, ExamineError> {
89        let mut tx = self.catalog.snapshot()?;
90        let rows = tx.examined()?;
91        let time = |nanos| from_nanos(nanos).ok_or(StoreError::Damaged);
92        let mut found = Examination {
93            versions: rows.len() as u64,
94            packed: tx.count_packed()?,
95            damaged: Vec::new(),
96            leftovers: Vec::new(),
97            due: 0,
98            oldest_due_since: None,
99            purges_since_key: 0,
100            oldest_purge_since_key: None,
101        };
102        for row in &rows {
103            let (version, stored_at, held) = (row.version, row.stored_at, row.held);
104            match self.read_or_why(&mut tx, row.uuid, version) {
105                Ok(_) => {}
106                Err(Unopened::Store(StoreError::Damaged)) => found.damaged.push(version),
107                Err(Unopened::Store(e)) => return Err(e.into()),
108                Err(Unopened::Storage(e)) => return Err(ExamineError::storage(&e)),
109            }
110            let Some(hidden_at) = row.hidden_at else {
111                continue;
112            };
113            let stored_at = time(stored_at)?;
114            let kept = Kept {
115                stored_at,
116                hidden: true,
117                held,
118            };
119            if !due(&kept, retention, now) {
120                continue;
121            }
122            // Due once its period ended, or from its hiding if that came
123            // later.
124            let end = retention.after(stored_at).ok_or(StoreError::Damaged)?;
125            let since = end.max(time(hidden_at)?);
126            found.due += 1;
127            found.oldest_due_since = Some(found.oldest_due_since.map_or(since, |o| o.min(since)));
128        }
129        // The objects the catalog names, as start-up recovery reads them: a
130        // version's object, unless its row says the version is packed.
131        let named: BTreeSet<String> = tx
132            .places()?
133            .into_iter()
134            .filter(|(_, place)| super::names_its_object(place))
135            .map(|(version, _)| version.to_string())
136            .collect();
137        for name in self.objects.list().map_err(unlisted)? {
138            let name = name.as_str();
139            let (id, temporary) = match name.strip_suffix(".tmp") {
140                Some(id) => (id, true),
141                None => (name, false),
142            };
143            if VersionId::parse(id).is_some() && (temporary || !named.contains(id)) {
144                found.leftovers.push(name.to_owned());
145            }
146        }
147        found.leftovers.sort();
148        drop(tx);
149        (found.purges_since_key, found.oldest_purge_since_key) = self.purges_since(key_since)?;
150        Ok(found)
151    }
152
153    /// How many purges were recorded since `since`, and when the oldest of
154    /// them was made: work only the tenant key's next rotation finishes.
155    pub fn purges_since(
156        &self,
157        since: OffsetDateTime,
158    ) -> Result<(u64, Option<OffsetDateTime>), StoreError> {
159        let purged = self.catalog.snapshot()?.purge_times()?;
160        let (mut count, mut oldest) = (0, None::<OffsetDateTime>);
161        for at in purged {
162            let at = OffsetDateTime::parse(&at, &Rfc3339).map_err(|_| StoreError::Damaged)?;
163            if at >= since {
164                count += 1;
165                oldest = Some(oldest.map_or(at, |o| o.min(at)));
166            }
167        }
168        Ok((count, oldest))
169    }
170}
171
172#[cfg(test)]
173mod examine_tests {
174    use time::macros::datetime;
175
176    use super::super::testing::vcon;
177    use super::super::tests::{Fixture, fixture};
178    use super::*;
179    use crate::fs::Fs;
180    use crate::store::Created;
181
182    const START: OffsetDateTime = datetime!(2026-10-23 12:00 UTC);
183
184    fn stored(f: &Fixture, n: u8) -> (uuid::Uuid, VersionId) {
185        let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-279e0d16bc4{n}"), "kept");
186        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
187            panic!("a new vCon")
188        };
189        (found.uuid, version)
190    }
191
192    fn examine(f: &Fixture, retention: Retention, now: OffsetDateTime) -> Examination {
193        f.store.examine(retention, START, now).expect("examined")
194    }
195
196    #[test]
197    fn a_sound_store_reports_its_versions_and_nothing_else() {
198        let f = fixture();
199        stored(&f, 0);
200        stored(&f, 1);
201        assert_eq!(
202            examine(&f, Retention::Years(7), START),
203            Examination {
204                versions: 2,
205                packed: 0,
206                damaged: vec![],
207                leftovers: vec![],
208                due: 0,
209                oldest_due_since: None,
210                purges_since_key: 0,
211                oldest_purge_since_key: None,
212            }
213        );
214    }
215
216    #[test]
217    fn a_missing_or_unreadable_object_is_damage() {
218        let f = fixture();
219        let (_, missing) = stored(&f, 0);
220        let (_, altered) = stored(&f, 1);
221        stored(&f, 2);
222        let objects = f.store.objects_dir.clone();
223        f.fs.remove(&objects.join(missing.to_string()))
224            .expect("remove");
225        f.fs.remove(&objects.join(altered.to_string()))
226            .expect("remove");
227        f.fs.write_new(&objects.join(altered.to_string()), b"not a sealed body")
228            .expect("write");
229        let mut expected = vec![missing, altered];
230        expected.sort_by_key(|v| v.0);
231        assert_eq!(examine(&f, Retention::Years(7), START).damaged, expected);
232    }
233
234    #[test]
235    fn files_no_row_names_are_leftovers_and_foreign_files_are_not_ours() {
236        let f = fixture();
237        let (_, version) = stored(&f, 0);
238        let orphan = VersionId([0x0e; 16]);
239        let objects = f.store.objects_dir.clone();
240        for name in [
241            orphan.to_string(),
242            format!("{orphan}.tmp"),
243            format!("{version}.tmp"),
244            "notes.txt".to_owned(),
245        ] {
246            f.fs.write_new(&objects.join(name), b"x").expect("write");
247        }
248        let mut expected = vec![
249            orphan.to_string(),
250            format!("{orphan}.tmp"),
251            format!("{version}.tmp"),
252        ];
253        expected.sort();
254        let found = examine(&f, Retention::Years(7), START);
255        assert_eq!(found.leftovers, expected);
256        assert_eq!(found.damaged, Vec::<VersionId>::new());
257    }
258
259    #[test]
260    fn versions_due_are_counted_from_when_each_fell_due_and_a_hold_keeps_them() {
261        let f = fixture();
262        let (deleted_early, _) = stored(&f, 0);
263        let (deleted_late, _) = stored(&f, 1);
264        let (held, _) = stored(&f, 2);
265        stored(&f, 3);
266        f.store
267            .delete(&f.caller, deleted_early, Retention::Years(7))
268            .expect("deleted");
269        f.store
270            .delete(&f.caller, held, Retention::Years(7))
271            .expect("deleted");
272        f.store.set_hold(&f.caller, held, true).expect("held");
273        // Deleted after its period had already run out: due from the delete.
274        f.clock.set(datetime!(2026-10-26 09:00 UTC));
275        f.store
276            .delete(&f.caller, deleted_late, Retention::Years(7))
277            .expect("deleted");
278        let found = examine(&f, Retention::Days(1), datetime!(2026-10-27 12:00 UTC));
279        assert_eq!(found.due, 2, "the held one is not due");
280        assert_eq!(
281            found.oldest_due_since,
282            Some(datetime!(2026-10-24 12:00 UTC)),
283            "a day after it was stored"
284        );
285        let later = examine(&f, Retention::Years(7), datetime!(2026-10-27 12:00 UTC));
286        assert_eq!((later.due, later.oldest_due_since), (0, None));
287        assert_eq!(later.versions, 4);
288    }
289
290    /// A version hidden after its period had run out became due when it was
291    /// hidden, not when the period ended.
292    #[test]
293    fn a_version_hidden_late_is_due_from_its_hiding() {
294        let f = fixture();
295        let (uuid, _) = stored(&f, 0);
296        f.clock.set(datetime!(2026-10-26 09:00 UTC));
297        f.store
298            .delete(&f.caller, uuid, Retention::Years(7))
299            .expect("deleted");
300        let found = examine(&f, Retention::Days(1), datetime!(2026-10-27 12:00 UTC));
301        assert_eq!(found.due, 1);
302        assert_eq!(
303            found.oldest_due_since,
304            Some(datetime!(2026-10-26 09:00 UTC))
305        );
306    }
307
308    #[test]
309    fn purges_since_the_key_was_rotated_are_counted() {
310        let f = fixture();
311        let (first, _) = stored(&f, 0);
312        let (second, _) = stored(&f, 1);
313        f.clock.set(datetime!(2026-10-24 12:00 UTC));
314        f.store
315            .delete(&f.caller, first, Retention::Days(0))
316            .expect("deleted and purged");
317        f.clock.set(datetime!(2026-10-25 12:00 UTC));
318        f.store
319            .delete(&f.caller, second, Retention::Days(0))
320            .expect("deleted and purged");
321        // A purge that failed was recorded too; it erased nothing.
322        f.store
323            .conn()
324            .execute(
325                "INSERT INTO audit (at, actor, action, target, version, status, count)
326                 VALUES ('2026-10-23T13:00:00Z', 'sweep', 'purge', NULL, NULL, 500, 1)",
327                [],
328            )
329            .expect("a failed purge's record");
330        let now = datetime!(2026-10-26 12:00 UTC);
331        let all = f
332            .store
333            .examine(Retention::Days(0), START, now)
334            .expect("examined");
335        assert_eq!(all.purges_since_key, 2);
336        assert_eq!(
337            all.oldest_purge_since_key,
338            Some(datetime!(2026-10-24 12:00 UTC))
339        );
340        let since = f
341            .store
342            .examine(Retention::Days(0), datetime!(2026-10-25 00:00 UTC), now)
343            .expect("examined");
344        assert_eq!(since.purges_since_key, 1);
345        assert_eq!(
346            since.oldest_purge_since_key,
347            Some(datetime!(2026-10-25 12:00 UTC))
348        );
349        assert_eq!(since.versions, 0);
350    }
351}
vcon_server/store/examine.rs

vcon_server/store/
examine.rs
