vcon_server/tenants/
examine.rs

1//! What `check` reads of the registry: each tenant's Store examined under
2//! its own retention period, rotations left part way, and tenant removals
3//! since the master key was last rotated. Reading changes nothing.
4
5use time::OffsetDateTime;
6use time::format_description::well_known::Rfc3339;
7
8use super::{OpenFailure, TenantError, Tenants, io, stored_retention};
9use crate::audit::Action;
10use crate::clock::from_nanos;
11use crate::envelope::{self, Context};
12use crate::ids::TenantId;
13use crate::retention::Retention;
14use crate::store::{Examination, ExamineError};
15
16/// One tenant as `check` found it.
17#[derive(Debug)]
18pub struct TenantExamination {
19    pub id: TenantId,
20    pub name: String,
21    /// A key rotation of this tenant was stopped part way.
22    pub rotation_stopped: bool,
23    /// Its Store examined, or why it was not.
24    pub store: Result<Examination, Unexamined>,
25}
26
27/// Why a tenant's Store was not examined: it did not open, or it opened
28/// and the examination failed, as when the objects cannot be read.
29#[derive(Debug)]
30pub enum Unexamined {
31    /// The Store did not open.
32    Open(OpenFailure),
33    /// The Store opened, and examining it failed.
34    Examine(ExamineError),
35}
36
37/// The registry as `check` found it.
38#[derive(Debug)]
39pub struct RegistryExamination {
40    pub tenants: Vec<TenantExamination>,
41    /// Tenants removed since the master key was last rotated: their keys
42    /// are destroyed only by the next rotation.
43    pub removed_since_master: u64,
44    /// When the oldest of those removals was made.
45    pub oldest_removed_since_master: Option<OffsetDateTime>,
46}
47
48/// Whether every tenant's data is under its current key, as `serve`
49/// requires before it starts: counts of tenants only, never which, and no
50/// path, since a path names a tenant.
51#[derive(Debug, Clone, PartialEq, Eq, Default)]
52pub struct KeyOrder {
53    /// Tenants whose key does not unwrap under the master key.
54    pub wrapped_elsewhere: u64,
55    /// Tenants with a key rotation stopped part way.
56    pub rotations_stopped: u64,
57    /// Tenants whose catalog does not open under their key, or with a
58    /// version row naming another key.
59    pub data_under_other_keys: u64,
60    /// Tenants whose Store did not open for a reason other than its key,
61    /// such as files this process may not use.
62    pub catalogs_unopened: u64,
63    /// The system's error for the first of those that has one, such as
64    /// `Permission denied (os error 13)`.
65    pub unopened_error: Option<String>,
66}
67
68impl KeyOrder {
69    pub fn is_sound(&self) -> bool {
70        *self == KeyOrder::default()
71    }
72}
73
74/// Work that only a key rotation finishes: purges since each tenant key's
75/// rotation, and tenant removals since the master key's. Read from the
76/// audit logs only.
77#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
78pub struct RotationBacklog {
79    pub tenants_with_purges: u64,
80    pub purges: u64,
81    pub oldest_purge: Option<OffsetDateTime>,
82    pub removals: u64,
83    pub oldest_removal: Option<OffsetDateTime>,
84}
85
86impl Tenants {
87    /// Counts the tenants whose data is not all under their current key.
88    pub fn key_order(&self) -> Result<KeyOrder, TenantError> {
89        let rows = self.registry.snapshot()?.tenants()?;
90        let mut order = KeyOrder::default();
91        for row in rows {
92            let (id, wrapped, pending) = (row.id, row.wrapped, row.pending.is_some());
93            if envelope::unwrap(&self.master, &Context::Tenant { tenant: id }, &wrapped).is_err() {
94                order.wrapped_elsewhere += 1;
95                continue;
96            }
97            if pending {
98                order.rotations_stopped += 1;
99            }
100            let elsewhere = match self.store_or_why(id) {
101                Ok(store) => store.rows_under_other_keys().map_err(io)? > 0,
102                Err(OpenFailure::WrongKey) => true,
103                Err(failure) => {
104                    order.catalogs_unopened += 1;
105                    if let OpenFailure::Access { error, .. } = failure {
106                        order.unopened_error.get_or_insert(error.to_string());
107                    }
108                    false
109                }
110            };
111            if elsewhere {
112                order.data_under_other_keys += 1;
113            }
114        }
115        Ok(order)
116    }
117
118    /// How many tenants have a version packed into a pack file: with any,
119    /// `serve` needs `[storage.parquet]`.
120    pub fn tenants_with_packed_versions(&self) -> Result<u64, TenantError> {
121        let rows = self.registry.snapshot()?.tenants()?;
122        let mut count = 0;
123        for row in rows {
124            if self.store(row.id)?.packed_versions().map_err(io)? > 0 {
125                count += 1;
126            }
127        }
128        Ok(count)
129    }
130
131    /// The rotation backlog: purges per tenant since its key's rotation,
132    /// and removals since the master key's.
133    pub fn rotation_backlog(&self) -> Result<RotationBacklog, TenantError> {
134        let rows = self.registry.snapshot()?.tenants()?;
135        let mut backlog = RotationBacklog::default();
136        for row in rows {
137            let key_since = from_nanos(row.key_since).ok_or(TenantError::Io)?;
138            let (purges, oldest) = self.store(row.id)?.purges_since(key_since).map_err(io)?;
139            if purges > 0 {
140                backlog.tenants_with_purges += 1;
141                backlog.purges += purges;
142                backlog.oldest_purge = earliest(backlog.oldest_purge, oldest);
143            }
144        }
145        (backlog.removals, backlog.oldest_removal) = self.removals_since_master()?;
146        Ok(backlog)
147    }
148
149    /// Tenant removals since the master key's rotation, and the oldest.
150    fn removals_since_master(&self) -> Result<(u64, Option<OffsetDateTime>), TenantError> {
151        let (since, removals) = {
152            let mut tx = self.registry.snapshot()?;
153            let since = tx.master_since(self.master.id())?;
154            let removals = tx.record_times(Action::TenantRemove, 204)?;
155            (since, removals)
156        };
157        let since = since
158            .map(|s| from_nanos(s).ok_or(TenantError::Io))
159            .transpose()?;
160        let (mut count, mut oldest) = (0, None);
161        for at in removals {
162            let at = OffsetDateTime::parse(&at, &Rfc3339).map_err(io)?;
163            if since.is_none_or(|since| at >= since) {
164                count += 1;
165                oldest = earliest(oldest, Some(at));
166            }
167        }
168        Ok((count, oldest))
169    }
170
171    /// Examines every tenant at `now`, under its own retention period or
172    /// `default`.
173    pub fn examine(
174        &self,
175        default: Retention,
176        now: OffsetDateTime,
177    ) -> Result<RegistryExamination, TenantError> {
178        let rows = self.registry.snapshot()?.tenants()?;
179        let time = |nanos| from_nanos(nanos).ok_or(TenantError::Io);
180        let mut tenants = Vec::new();
181        for row in rows {
182            let (id, name, retention, key_since) = (row.id, row.name, row.retention, row.key_since);
183            let rotation_stopped = row.pending.is_some();
184            let retention = super::kept_for(stored_retention(retention)?, default);
185            let key_since = time(key_since)?;
186            let store = self
187                .store_or_why(id)
188                .map_err(Unexamined::Open)
189                .and_then(|store| {
190                    store
191                        .examine(retention, key_since, now)
192                        .map_err(Unexamined::Examine)
193                });
194            tenants.push(TenantExamination {
195                id,
196                name,
197                rotation_stopped,
198                store,
199            });
200        }
201        let (removed_since_master, oldest_removed_since_master) = self.removals_since_master()?;
202        let found = RegistryExamination {
203            tenants,
204            removed_since_master,
205            oldest_removed_since_master,
206        };
207        Ok(found)
208    }
209}
210
211/// The earlier of two times, either of which may be absent.
212fn earliest(a: Option<OffsetDateTime>, b: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
213    match (a, b) {
214        (Some(a), Some(b)) => Some(a.min(b)),
215        (a, b) => a.or(b),
216    }
217}
218
219#[cfg(test)]
220mod examine_tests {
221    use time::macros::datetime;
222
223    use super::super::tests::fixture;
224    use super::*;
225    use crate::audit::Actor;
226    use crate::store::Caller;
227    use crate::store::testing::vcon;
228
229    const CALLER: Caller = Caller {
230        key: crate::ids::KeyId([7; 8]),
231    };
232
233    #[test]
234    fn each_tenant_is_examined_under_its_own_period_or_the_default() {
235        let f = fixture();
236        let own = f
237            .tenants
238            .create(Actor::Cli, "own", Some(Retention::Days(1)))
239            .expect("made");
240        let default = f.tenants.create(Actor::Cli, "default", None).expect("made");
241        for id in [own, default] {
242            let store = f.tenants.store(id).expect("store");
243            let (found, body) = vcon("01a07da8-c2bb-83e5-b9a2-279e0d16bc46", "kept");
244            store.create(&CALLER, &found, &body).expect("stored");
245            store
246                .delete(&CALLER, found.uuid, Retention::Years(7))
247                .expect("deleted");
248        }
249        let found = f
250            .tenants
251            .examine(Retention::Years(7), datetime!(2026-10-30 12:00 UTC))
252            .expect("examined");
253        let by_name = |name: &str| found.tenants.iter().find(|t| t.name == name).expect(name);
254        let own = by_name("own");
255        assert_eq!(
256            own.store.as_ref().expect("opens").due,
257            1,
258            "a day has passed"
259        );
260        assert!(!own.rotation_stopped);
261        assert_eq!(by_name("default").store.as_ref().expect("opens").due, 0);
262        assert_eq!(found.tenants.len(), 2);
263    }
264
265    #[test]
266    fn a_rotation_stopped_part_way_is_reported_and_a_catalog_that_does_not_open_too() {
267        let f = fixture();
268        let id = f.tenants.create(Actor::Cli, "acme", None).expect("made");
269        *f.tenants.hooks.rekey_stop.lock().expect("hook") =
270            Some(super::super::rekey::RekeyStep::Reencrypted);
271        assert_eq!(f.tenants.rekey_tenant(Actor::Cli, id), Err(TenantError::Io));
272        let found = f
273            .tenants
274            .examine(Retention::Years(7), datetime!(2026-10-23 12:00 UTC))
275            .expect("examined");
276        let acme = &found.tenants[0];
277        assert!(acme.rotation_stopped);
278        assert!(
279            matches!(acme.store, Err(Unexamined::Open(OpenFailure::WrongKey))),
280            "its catalog is under the pending key: {:?}",
281            acme.store
282        );
283    }
284
285    /// A tenant's files this process may not write, such as ones a command
286    /// run as root left, are reported as that, with the path and the
287    /// system's error; only a key that does not open the catalog is a key
288    /// problem.
289    #[test]
290    fn a_catalog_the_process_may_not_open_is_an_access_failure_not_a_key_one() {
291        use std::os::unix::fs::PermissionsExt;
292        let f = fixture();
293        // Made, but not yet opened through the registry, as at start-up.
294        let id = f.tenants.create(Actor::Cli, "acme", None).expect("made");
295        let dir = f.dir.path().join("tenants").join(id.to_string());
296        let mode =
297            |m| std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(m)).expect("chmod");
298        mode(0o500);
299        let order = f.tenants.key_order();
300        let found = f
301            .tenants
302            .examine(Retention::Years(7), datetime!(2026-10-23 12:00 UTC));
303        mode(0o700);
304        let order = order.expect("read");
305        assert_eq!(order.data_under_other_keys, 0, "{order:?}");
306        assert_eq!(order.catalogs_unopened, 1, "{order:?}");
307        assert_eq!(
308            order.unopened_error.as_deref(),
309            Some("Permission denied (os error 13)")
310        );
311        assert!(!order.is_sound());
312        match &found.expect("examined").tenants[0].store {
313            Err(Unexamined::Open(OpenFailure::Access { path, error })) => {
314                assert_eq!(path, &dir);
315                assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
316            }
317            other => panic!("{other:?}"),
318        }
319        assert!(
320            f.tenants.key_order().expect("read").is_sound(),
321            "once allowed"
322        );
323    }
324
325    /// The tenants with a version whose catalog row names a pack file, as
326    /// a pack run leaves them: counted per tenant, not per version, and
327    /// none while every version is in its object.
328    #[test]
329    fn tenants_with_packed_versions_are_counted() {
330        let f = fixture();
331        let mut ids = Vec::new();
332        for name in ["acme", "beta", "gamma"] {
333            let id = f.tenants.create(Actor::Cli, name, None).expect("made");
334            let store = f.tenants.store(id).expect("store");
335            for n in 0..2 {
336                let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-279e0d16bc4{n}"), "k");
337                store.create(&CALLER, &found, &body).expect("stored");
338            }
339            ids.push(id);
340        }
341        assert_eq!(
342            f.tenants.tenants_with_packed_versions(),
343            Ok(0),
344            "every version in its object"
345        );
346        // Both versions of the first tenant, and one of the second.
347        for (id, limit) in [(ids[0], 2_i64), (ids[1], 1)] {
348            let store = f.tenants.store(id).expect("store");
349            let changed = store
350                .conn_for_test()
351                .execute(
352                    "UPDATE version SET place_file = randomblob(16), place_group = 0, place_row = 0
353                     WHERE version IN (SELECT version FROM version ORDER BY version LIMIT ?1)",
354                    [limit],
355                )
356                .expect("placed in a pack file");
357            assert_eq!(i64::try_from(changed), Ok(limit), "placed");
358        }
359        assert_eq!(
360            f.tenants.tenants_with_packed_versions(),
361            Ok(2),
362            "two tenants, one with a single packed version"
363        );
364    }
365
366    mod key_order {
367        use super::*;
368        use crate::envelope::{self, Context, Key};
369        use crate::tenants::rekey::RekeyStep;
370
371        fn with_vcons(f: &super::super::super::tests::Fixture) -> TenantId {
372            let id = f.tenants.create(Actor::Cli, "acme", None).expect("made");
373            let store = f.tenants.store(id).expect("store");
374            for n in 0..3 {
375                let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-279e0d16bc4{n}"), "k");
376                store.create(&CALLER, &found, &body).expect("stored");
377            }
378            id
379        }
380
381        #[test]
382        fn a_sound_registry_is_in_order() {
383            let f = fixture();
384            with_vcons(&f);
385            assert!(f.tenants.key_order().expect("read").is_sound());
386        }
387
388        #[test]
389        fn a_rotation_stopped_at_any_step_is_out_of_order() {
390            for (step, data) in [
391                (RekeyStep::Pending, 0),
392                (RekeyStep::Batch, 1),
393                (RekeyStep::Rewrapped, 1),
394                (RekeyStep::Reencrypted, 1),
395            ] {
396                let f = fixture();
397                let id = with_vcons(&f);
398                *f.tenants.hooks.rekey_stop.lock().expect("hook") = Some(step);
399                assert!(f.tenants.rekey_in_batches(Actor::Cli, id, 2).is_err());
400                assert_eq!(
401                    f.tenants.key_order().expect("read"),
402                    KeyOrder {
403                        wrapped_elsewhere: 0,
404                        rotations_stopped: 1,
405                        data_under_other_keys: data,
406                        ..KeyOrder::default()
407                    },
408                    "{step:?}"
409                );
410                f.tenants.rekey_tenant(Actor::Cli, id).expect("finished");
411                assert!(f.tenants.key_order().expect("read").is_sound(), "{step:?}");
412            }
413        }
414
415        #[test]
416        fn a_tenant_key_wrapped_under_another_key_is_out_of_order() {
417            let f = fixture();
418            let id = with_vcons(&f);
419            let other = Key::generate().expect("key");
420            let foreign =
421                envelope::wrap(&other, &Context::Tenant { tenant: id }, &f.tenant_key(id));
422            f.conn()
423                .execute(
424                    "UPDATE tenant SET wrapped_tenant_key = ?1 WHERE id = ?2",
425                    rusqlite::params![foreign, id.0],
426                )
427                .expect("rewritten");
428            f.tenants.close(id);
429            assert_eq!(f.tenants.key_order().expect("read").wrapped_elsewhere, 1);
430        }
431    }
432
433    #[test]
434    fn the_rotation_backlog_counts_purges_and_removals_from_the_logs() {
435        let f = fixture();
436        let purging = f
437            .tenants
438            .create(Actor::Cli, "purging", Some(Retention::Days(0)))
439            .expect("made");
440        f.tenants.create(Actor::Cli, "quiet", None).expect("made");
441        let gone = f.tenants.create(Actor::Cli, "gone", None).expect("made");
442        let store = f.tenants.store(purging).expect("store");
443        for (n, at) in [
444            (0, datetime!(2026-10-24 10:00 UTC)),
445            (1, datetime!(2026-10-24 11:00 UTC)),
446        ] {
447            f.clock.set(at);
448            let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-279e0d16bc4{n}"), "k");
449            store.create(&CALLER, &found, &body).expect("stored");
450            store
451                .delete(&CALLER, found.uuid, Retention::Days(0))
452                .expect("deleted and purged");
453        }
454        drop(store);
455        f.clock.set(datetime!(2026-10-24 12:00 UTC));
456        f.tenants
457            .remove(Actor::Cli, gone, crate::tenants::Removal::Empty)
458            .expect("removed");
459        assert_eq!(
460            f.tenants.rotation_backlog().expect("read"),
461            RotationBacklog {
462                tenants_with_purges: 1,
463                purges: 2,
464                oldest_purge: Some(datetime!(2026-10-24 10:00 UTC)),
465                removals: 1,
466                oldest_removal: Some(datetime!(2026-10-24 12:00 UTC)),
467            }
468        );
469    }
470
471    #[test]
472    fn removals_since_the_master_key_was_rotated_are_counted() {
473        let f = fixture();
474        let gone = f.tenants.create(Actor::Cli, "gone", None).expect("made");
475        f.tenants.create(Actor::Cli, "kept", None).expect("made");
476        f.clock.set(datetime!(2026-10-24 08:00 UTC));
477        f.tenants
478            .remove(Actor::Cli, gone, crate::tenants::Removal::Empty)
479            .expect("removed");
480        // A removal from before the master key's rotation, and a refused one.
481        f.conn()
482            .execute_batch(
483                "INSERT INTO audit (at, actor, action, target, version, status, count)
484                 VALUES ('2026-10-01T00:00:00Z', 'cli', 'tenant_remove', NULL, NULL, 204, 1),
485                        ('2026-10-24T09:00:00Z', 'cli', 'tenant_remove', NULL, NULL, 409, 1);",
486            )
487            .expect("older records");
488        let found = f
489            .tenants
490            .examine(Retention::Years(7), datetime!(2026-10-25 12:00 UTC))
491            .expect("examined");
492        assert_eq!(found.removed_since_master, 1);
493        assert_eq!(
494            found.oldest_removed_since_master,
495            Some(datetime!(2026-10-24 08:00 UTC))
496        );
497        assert_eq!(found.tenants.len(), 1);
498    }
499}
vcon_server/tenants/examine.rs

vcon_server/tenants/
examine.rs
