vcon_server/store/
purge.rs

1//! When a kept version is erased: the purge rule, apart from the files and
2//! rows it touches, so its boundaries can be tested directly.
3
4use time::OffsetDateTime;
5use uuid::Uuid;
6
7use super::{Store, StoreError, commit, io};
8use crate::audit::{Action, Actor, Entry};
9use crate::clock::{from_nanos, nanos};
10use crate::ids::VersionId;
11use crate::retention::Retention;
12
13/// What the purge rule reads of one version.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub struct Kept {
16    /// When the version was stored; retention counts from here.
17    pub stored_at: OffsetDateTime,
18    /// Whether it was replaced or its uuid deleted. A current version is
19    /// never purged.
20    pub hidden: bool,
21    /// Whether its uuid is under a legal hold.
22    pub held: bool,
23}
24
25/// Whether `version` is due for purge at `now` under `retention`: it is
26/// hidden, its uuid has no hold, and `retention` has passed since it was
27/// stored, counted in calendar time in UTC. A period that ends past the
28/// last time this build can hold never passes. A version hidden after it
29/// reached that age is due at once.
30pub fn due(version: &Kept, retention: Retention, now: OffsetDateTime) -> bool {
31    version.hidden
32        && !version.held
33        && retention
34            .after(version.stored_at)
35            .is_some_and(|end| now >= end)
36}
37
38/// What a purge did.
39#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
40pub struct PurgeReport {
41    /// Versions erased.
42    pub purged: u64,
43}
44
45impl Store {
46    /// Erases every version due under `retention`. Each goes in a
47    /// transaction of its own that checks the rule again, so a restore or
48    /// a hold that commits first keeps it. The transaction deletes the
49    /// version's row, and with it the only copy of its wrapped data key,
50    /// and records the purge; a deleted uuid left with no version gives up
51    /// its row for a tombstone. The object file is removed after the
52    /// commit. Last, the WAL is checkpointed with truncation, tried again
53    /// while readers finish, so no page that held a purged key is left in
54    /// it.
55    pub fn purge_due(&self, retention: Retention) -> Result<PurgeReport, StoreError> {
56        self.purge_hidden(Actor::Sweep, retention, None)
57    }
58
59    /// Purges, as `purge_due` does, the hidden versions that are due, with
60    /// `who` as the actor of each record; with `uuid`, only that uuid's.
61    pub(super) fn purge_hidden(
62        &self,
63        who: Actor,
64        retention: Retention,
65        uuid: Option<Uuid>,
66    ) -> Result<PurgeReport, StoreError> {
67        let hidden = self.catalog.snapshot()?.hidden_versions(uuid)?;
68        // A restore, hold or read may commit here, between the listing and
69        // each version's own transaction, which checks the rule again.
70        self.pause();
71        let mut report = PurgeReport::default();
72        for version in hidden {
73            if self.purge(who, version, retention)? {
74                report.purged += 1;
75            }
76        }
77        if report.purged > 0 {
78            self.catalog.after_purge()?;
79        }
80        Ok(report)
81    }
82
83    /// Purges `version` if it is due; false if it is not, or is gone.
84    fn purge(
85        &self,
86        who: Actor,
87        version: VersionId,
88        retention: Retention,
89    ) -> Result<bool, StoreError> {
90        let mut tx = self.catalog.immediate()?;
91        let Some(row) = tx.hidden_row(version)? else {
92            return Ok(false);
93        };
94        let now = self.clock.now();
95        let kept = Kept {
96            stored_at: from_nanos(row.stored_at).ok_or(StoreError::Damaged)?,
97            hidden: true,
98            held: row.held,
99        };
100        if !due(&kept, retention, now) {
101            return Ok(false);
102        }
103        tx.delete_version(version)?;
104        if row.deleted && tx.versions_of(row.uuid)? == 0 {
105            tx.tombstone(row.uuid, nanos(now))?;
106        }
107        let entry = Entry {
108            actor: who,
109            action: Action::Purge,
110            target: Some(row.uuid.to_string()),
111            version: Some(version),
112            status: 200,
113        };
114        self.record(&mut tx, &entry)?;
115        commit(tx)?;
116        self.objects.remove(version).map_err(io)?;
117        Ok(true)
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use time::Duration;
124    use time::macros::datetime;
125
126    use super::*;
127
128    const SECOND: Duration = Duration::seconds(1);
129
130    fn hidden(stored_at: OffsetDateTime) -> Kept {
131        Kept {
132            stored_at,
133            hidden: true,
134            held: false,
135        }
136    }
137
138    #[test]
139    fn a_hidden_version_is_due_from_the_second_its_period_ends() {
140        let seven = Retention::Years(7);
141        let cases = [
142            // Stored on an ordinary day: due seven calendar years later.
143            (
144                datetime!(2026-03-15 10:00 UTC),
145                datetime!(2033-03-15 10:00 UTC),
146            ),
147            // Stored on 29 February: due on 28 February seven years later,
148            // not on 1 March.
149            (
150                datetime!(2028-02-29 12:00 UTC),
151                datetime!(2035-02-28 12:00 UTC),
152            ),
153        ];
154        for (stored_at, end) in cases {
155            let version = hidden(stored_at);
156            assert!(
157                !due(&version, seven, end - SECOND),
158                "{stored_at}: a second short"
159            );
160            assert!(due(&version, seven, end), "{stored_at}: at the end");
161            assert!(
162                due(&version, seven, end + SECOND),
163                "{stored_at}: a second past"
164            );
165        }
166    }
167
168    #[test]
169    fn a_hold_keeps_a_version_past_its_end() {
170        let stored_at = datetime!(2028-02-29 12:00 UTC);
171        let end = datetime!(2035-02-28 12:00 UTC);
172        let version = Kept {
173            held: true,
174            ..hidden(stored_at)
175        };
176        for now in [end - SECOND, end, end + SECOND, end + Duration::days(3650)] {
177            assert!(!due(&version, Retention::Years(7), now), "{now}");
178        }
179    }
180
181    #[test]
182    fn a_current_version_is_never_due() {
183        let version = Kept {
184            hidden: false,
185            ..hidden(datetime!(2026-03-15 10:00 UTC))
186        };
187        let end = datetime!(2033-03-15 10:00 UTC);
188        for now in [end - SECOND, end, end + SECOND] {
189            assert!(!due(&version, Retention::Years(7), now), "{now}");
190        }
191    }
192
193    #[test]
194    fn a_zero_period_makes_a_hidden_version_due_at_once() {
195        let stored_at = datetime!(2026-03-15 10:00 UTC);
196        let version = hidden(stored_at);
197        assert!(due(&version, Retention::Days(0), stored_at));
198        assert!(!due(&version, Retention::Days(0), stored_at - SECOND));
199    }
200
201    #[test]
202    fn a_period_ending_past_the_last_time_held_is_never_due() {
203        let version = hidden(datetime!(9998-06-01 00:00 UTC));
204        let last = datetime!(9999-12-31 23:59:59 UTC);
205        assert!(!due(&version, Retention::Years(7), last));
206    }
207}
208
209#[cfg(test)]
210mod purge_tests {
211    use std::path::Path;
212    use std::sync::{Arc, mpsc};
213    use std::time::Duration as Wait;
214
215    use aws_lc_rs::cipher::{AES_256, DecryptingKey, DecryptionContext, UnboundCipherKey};
216    use aws_lc_rs::iv::FixedLength;
217    use serde_json::Value;
218    use time::Duration;
219    use uuid::Uuid;
220
221    use super::super::testing::vcon;
222    use super::super::tests::{Fixture, fixture};
223    use super::*;
224    use crate::catalog;
225    use crate::clock::Clock;
226    use crate::envelope::{self, Key};
227    use crate::fs::Fs;
228    use crate::ids::VersionId;
229    use crate::store::{Created, StoreError};
230
231    const UUID: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc46";
232    const DAY: Retention = Retention::Days(1);
233
234    /// Stores a vCon and returns its uuid and version.
235    fn stored(f: &Fixture, subject: &str) -> (Uuid, VersionId) {
236        let (found, body) = vcon(UUID, subject);
237        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
238            panic!("a new vCon")
239        };
240        (found.uuid, version)
241    }
242
243    fn later(f: &Fixture, by: Duration) {
244        f.clock.set(f.clock.now() + by);
245    }
246
247    fn count(f: &Fixture, sql: &str, uuid: Uuid) -> i64 {
248        f.store
249            .conn()
250            .query_row(sql, [uuid.as_bytes()], |r| r.get(0))
251            .expect(sql)
252    }
253
254    fn wrapped_key(f: &Fixture, version: VersionId) -> Vec<u8> {
255        f.store
256            .conn()
257            .query_row(
258                "SELECT wrapped_key FROM version WHERE version = ?1",
259                [version.0],
260                |r| r.get(0),
261            )
262            .expect("row")
263    }
264
265    fn object_exists(f: &Fixture, version: VersionId) -> bool {
266        f.fs.read(&f.store.objects_dir.join(version.to_string()))
267            .is_ok()
268    }
269
270    fn purges(f: &Fixture) -> Vec<Value> {
271        f.store
272            .records()
273            .into_iter()
274            .filter(|r| r["action"] == "purge")
275            .collect()
276    }
277
278    #[test]
279    fn a_purge_erases_a_deleted_vcon_and_leaves_a_tombstone() {
280        let f = fixture();
281        let (uuid, version) = stored(&f, "first");
282        f.store
283            .delete(&f.caller, uuid, Retention::Years(7))
284            .expect("deletes");
285        later(&f, Duration::days(2));
286
287        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 1 }));
288
289        assert_eq!(
290            count(&f, "SELECT count(*) FROM version WHERE uuid = ?1", uuid),
291            0
292        );
293        assert_eq!(
294            count(&f, "SELECT count(*) FROM vcon WHERE uuid = ?1", uuid),
295            0
296        );
297        assert_eq!(
298            count(&f, "SELECT count(*) FROM tombstone WHERE uuid = ?1", uuid),
299            1
300        );
301        let purged_at: i64 = f
302            .store
303            .conn()
304            .query_row(
305                "SELECT purged_at FROM tombstone WHERE uuid = ?1",
306                [uuid.as_bytes()],
307                |r| r.get(0),
308            )
309            .expect("tombstone");
310        assert_eq!(
311            purged_at,
312            nanos(f.clock.now()),
313            "the tombstone holds the purge time"
314        );
315        assert!(!object_exists(&f, version), "the object file is removed");
316        let records = purges(&f);
317        assert_eq!(records.len(), 1);
318        assert_eq!(records[0]["actor"], "sweep");
319        assert_eq!(records[0]["target"], uuid.to_string());
320        assert_eq!(records[0]["version"], version.to_string());
321        assert_eq!(f.store.get(&f.caller, uuid), Err(StoreError::Gone));
322        let (found, body) = vcon(UUID, "again");
323        assert_eq!(
324            f.store.create(&f.caller, &found, &body),
325            Err(StoreError::Gone),
326            "a purged uuid cannot be stored again"
327        );
328    }
329
330    #[test]
331    fn a_purge_of_a_replaced_version_keeps_the_live_vcon() {
332        let f = fixture();
333        let (uuid, old) = stored(&f, "first");
334        let (found, body) = vcon(UUID, "second");
335        let new = f
336            .store
337            .replace(&f.caller, old, &found, &body)
338            .expect("replaces");
339        later(&f, Duration::days(2));
340
341        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 1 }));
342
343        assert_eq!(
344            count(&f, "SELECT count(*) FROM version WHERE uuid = ?1", uuid),
345            1
346        );
347        assert_eq!(
348            count(&f, "SELECT count(*) FROM tombstone WHERE uuid = ?1", uuid),
349            0
350        );
351        assert!(!object_exists(&f, old));
352        assert!(object_exists(&f, new));
353        assert_eq!(f.store.get(&f.caller, uuid).expect("reads").body, body);
354    }
355
356    #[test]
357    fn versions_not_due_are_kept() {
358        let f = fixture();
359        // Current: never due.
360        let (current, _) = stored(&f, "current");
361        // Deleted, but held.
362        let (found, body) = vcon("0199e5c6-0d7a-7c11-8000-000000000001", "held");
363        f.store.create(&f.caller, &found, &body).expect("stores");
364        f.store
365            .delete(&f.caller, found.uuid, Retention::Years(7))
366            .expect("deletes");
367        f.store
368            .conn()
369            .execute(
370                "UPDATE vcon SET hold = 1 WHERE uuid = ?1",
371                [found.uuid.as_bytes()],
372            )
373            .expect("holds");
374        later(&f, Duration::days(2));
375        // Deleted, but stored too recently.
376        let (young, body) = vcon("0199e5c6-0d7a-7c11-8000-000000000002", "young");
377        f.store.create(&f.caller, &young, &body).expect("stores");
378        f.store
379            .delete(&f.caller, young.uuid, Retention::Years(7))
380            .expect("deletes");
381
382        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 0 }));
383
384        for uuid in [current, found.uuid, young.uuid] {
385            assert_eq!(
386                count(&f, "SELECT count(*) FROM version WHERE uuid = ?1", uuid),
387                1
388            );
389        }
390        assert!(purges(&f).is_empty());
391    }
392
393    #[test]
394    fn deleting_an_old_vcon_erases_it_at_once() {
395        let f = fixture();
396        let (uuid, version) = stored(&f, "first");
397        later(&f, Duration::days(2));
398
399        assert_eq!(f.store.delete(&f.caller, uuid, DAY), Ok(()));
400
401        assert_eq!(
402            count(&f, "SELECT count(*) FROM version WHERE uuid = ?1", uuid),
403            0
404        );
405        assert_eq!(
406            count(&f, "SELECT count(*) FROM tombstone WHERE uuid = ?1", uuid),
407            1
408        );
409        assert!(!object_exists(&f, version));
410        let records = purges(&f);
411        assert_eq!(records.len(), 1);
412        assert_eq!(
413            records[0]["actor"],
414            f.caller.key.to_string(),
415            "the delete that caused the purge is its actor"
416        );
417    }
418
419    #[test]
420    fn deleting_a_young_vcon_keeps_it_hidden() {
421        let f = fixture();
422        let (uuid, version) = stored(&f, "first");
423        later(&f, Duration::hours(23));
424
425        assert_eq!(f.store.delete(&f.caller, uuid, DAY), Ok(()));
426
427        assert_eq!(
428            count(
429                &f,
430                "SELECT count(*) FROM version WHERE uuid = ?1 AND hidden_at IS NOT NULL",
431                uuid
432            ),
433            1
434        );
435        assert!(object_exists(&f, version));
436        assert!(purges(&f).is_empty());
437    }
438
439    #[test]
440    fn a_zero_period_erases_inside_the_delete() {
441        let f = fixture();
442        let (uuid, version) = stored(&f, "first");
443
444        assert_eq!(f.store.delete(&f.caller, uuid, Retention::Days(0)), Ok(()));
445
446        assert_eq!(
447            count(&f, "SELECT count(*) FROM tombstone WHERE uuid = ?1", uuid),
448            1
449        );
450        assert!(!object_exists(&f, version));
451    }
452
453    #[test]
454    fn a_delete_purges_only_its_own_uuid() {
455        let f = fixture();
456        let (uuid, _) = stored(&f, "first");
457        let (other, body) = vcon("0199e5c6-0d7a-7c11-8000-000000000003", "other");
458        f.store.create(&f.caller, &other, &body).expect("stores");
459        f.store
460            .delete(&f.caller, other.uuid, Retention::Years(7))
461            .expect("deletes");
462        later(&f, Duration::days(2));
463
464        f.store.delete(&f.caller, uuid, DAY).expect("deletes");
465
466        assert_eq!(
467            count(
468                &f,
469                "SELECT count(*) FROM version WHERE uuid = ?1",
470                other.uuid
471            ),
472            1,
473            "the other uuid waits for the sweep"
474        );
475    }
476
477    #[test]
478    fn a_delete_whose_purge_fails_stands_and_answers_the_failure() {
479        let f = fixture();
480        let (uuid, version) = stored(&f, "first");
481        later(&f, Duration::days(2));
482        f.store
483            .conn()
484            .execute_batch(
485                "CREATE TEMP TRIGGER no_purge BEFORE DELETE ON version
486                 BEGIN SELECT RAISE(ABORT, 'the disk has failed'); END;",
487            )
488            .expect("trigger");
489
490        assert_eq!(
491            f.store.delete(&f.caller, uuid, DAY),
492            Err(StoreError::Io),
493            "the delete answers the failed purge: the store's I/O error, not a refusal"
494        );
495
496        assert_eq!(f.store.get(&f.caller, uuid), Err(StoreError::Gone));
497        assert_eq!(
498            count(
499                &f,
500                "SELECT count(*) FROM version WHERE uuid = ?1 AND hidden_at IS NOT NULL",
501                uuid
502            ),
503            1,
504            "kept hidden for the sweep to purge"
505        );
506        assert!(object_exists(&f, version));
507    }
508
509    #[test]
510    fn deleting_again_erases_what_has_since_come_due() {
511        let f = fixture();
512        let (uuid, version) = stored(&f, "first");
513        f.store
514            .delete(&f.caller, uuid, Retention::Years(7))
515            .expect("deletes");
516        later(&f, Duration::days(2));
517
518        assert_eq!(f.store.delete(&f.caller, uuid, DAY), Ok(()));
519
520        assert!(!object_exists(&f, version));
521    }
522
523    /// Stores a vCon, deletes it while seven years apply, and moves the
524    /// clock two days on, so under one day it is due for purge.
525    fn due_for_purge(f: &Fixture) -> (Uuid, VersionId, Vec<u8>) {
526        let (found, body) = vcon(UUID, "first");
527        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
528            panic!("a new vCon")
529        };
530        f.store
531            .delete(&f.caller, found.uuid, Retention::Years(7))
532            .expect("deleted");
533        later(f, Duration::days(2));
534        (found.uuid, version, body)
535    }
536
537    #[test]
538    fn a_purge_cut_off_before_its_commit_leaves_the_version_whole() {
539        let mut f = fixture();
540        let (uuid, version, body) = due_for_purge(&f);
541        let other = super::super::tests::second_connection(&f);
542        other
543            .execute_batch(
544                "CREATE TRIGGER dead BEFORE INSERT ON audit
545                 BEGIN SELECT RAISE(ABORT, 'the process has died'); END;",
546            )
547            .expect("trigger");
548
549        assert!(f.store.purge_due(DAY).is_err());
550        drop(other);
551        super::super::tests::restart(&mut f);
552
553        let read = f.store.legal_get(&f.caller, uuid, None).expect("whole");
554        assert_eq!((read.stored.version, read.stored.body), (version, body));
555        assert!(object_exists(&f, version));
556        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 1 }));
557        assert!(!object_exists(&f, version));
558    }
559
560    #[test]
561    fn a_purge_cut_off_at_its_file_step_is_finished_by_recovery() {
562        // Step 0 fails the removal itself; step 1 removes it unsynced, so a
563        // crash brings the file back.
564        for step in 0..=1 {
565            let mut f = fixture();
566            let (uuid, version, _) = due_for_purge(&f);
567            f.fs.fail_from(f.fs.log().len() + step);
568            let _ = f.store.purge_due(DAY);
569            super::super::tests::restart(&mut f);
570
571            assert_eq!(
572                count(&f, "SELECT count(*) FROM version WHERE uuid = ?1", uuid),
573                0,
574                "step {step}: the commit stands"
575            );
576            assert_eq!(
577                count(&f, "SELECT count(*) FROM tombstone WHERE uuid = ?1", uuid),
578                1,
579                "step {step}"
580            );
581            assert!(
582                !object_exists(&f, version),
583                "step {step}: recovery removes the file"
584            );
585            assert_eq!(
586                f.store.legal_get(&f.caller, uuid, None),
587                Err(StoreError::Gone),
588                "step {step}"
589            );
590        }
591    }
592
593    /// Runs `legal` on the store between the purge's listing of hidden
594    /// versions and its transaction for each, as a racing request would.
595    fn between_listing_and_purge(f: &Arc<Fixture>, legal: impl FnOnce(&Fixture) + Send + 'static) {
596        let racer = f.clone();
597        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || legal(&racer)));
598    }
599
600    #[test]
601    fn a_restore_that_commits_before_the_purge_keeps_the_vcon_live() {
602        let f = Arc::new(fixture());
603        let (uuid, version, body) = due_for_purge(&f);
604        between_listing_and_purge(&f, move |racer| {
605            racer.store.restore(&racer.caller, uuid).expect("restored");
606        });
607
608        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 0 }));
609
610        let read = f.store.get(&f.caller, uuid).expect("live");
611        assert_eq!((read.version, read.body), (version, body));
612    }
613
614    #[test]
615    fn a_hold_that_commits_before_the_purge_keeps_the_version() {
616        let f = Arc::new(fixture());
617        let (uuid, version, body) = due_for_purge(&f);
618        between_listing_and_purge(&f, move |racer| {
619            racer
620                .store
621                .set_hold(&racer.caller, uuid, true)
622                .expect("held");
623        });
624
625        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 0 }));
626
627        let read = f.store.legal_get(&f.caller, uuid, None).expect("kept");
628        assert_eq!((read.stored.version, read.stored.body), (version, body));
629    }
630
631    #[test]
632    fn a_legal_read_racing_a_purge_gets_the_whole_body_or_gone() {
633        let f = Arc::new(fixture());
634        let (uuid, _, body) = due_for_purge(&f);
635        let (seen_tx, seen) = mpsc::channel();
636        between_listing_and_purge(&f, move |racer| {
637            seen_tx
638                .send(racer.store.legal_get(&racer.caller, uuid, None))
639                .expect("send");
640        });
641
642        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 1 }));
643
644        let before = seen
645            .recv_timeout(Wait::from_secs(5))
646            .expect("the racing read ran");
647        assert_eq!(before.expect("before the purge").stored.body, body);
648        assert_eq!(
649            f.store.legal_get(&f.caller, uuid, None),
650            Err(StoreError::Gone),
651            "after it"
652        );
653    }
654
655    /// The plain contents of every page of the catalog at `path` and of
656    /// every frame in its WAL, decrypted as SQLCipher 4 writes them:
657    /// AES-256-CBC under the raw catalog key, the last 80 bytes of each
658    /// page reserved for its 16-byte IV and 64-byte HMAC, and the first 16
659    /// bytes of page 1 holding the salt in the clear.
660    fn plain_pages(path: &Path, key: &Key, page_size: usize) -> Vec<Vec<u8>> {
661        const RESERVED: usize = 80;
662        const IV: usize = 16;
663        const SALT: usize = 16;
664        let raw = key.sqlcipher_raw();
665        let hex = &raw[2..66];
666        let key: Vec<u8> = (0..32)
667            .map(|i| u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).expect("hex"))
668            .collect();
669        let decrypt = |number: u32, page: &[u8]| {
670            let start = if number == 1 { SALT } else { 0 };
671            let end = page_size - RESERVED;
672            let iv: [u8; IV] = page[end..end + IV].try_into().expect("an IV");
673            let mut text = page[start..end].to_vec();
674            let cipher = DecryptingKey::cbc(UnboundCipherKey::new(&AES_256, &key).expect("key"))
675                .expect("CBC");
676            cipher
677                .decrypt(&mut text, DecryptionContext::Iv128(FixedLength::from(iv)))
678                .expect("decrypts");
679            text
680        };
681        let mut pages = Vec::new();
682        let db = std::fs::read(path).expect("catalog");
683        for (i, page) in db.chunks_exact(page_size).enumerate() {
684            pages.push(decrypt(u32::try_from(i + 1).expect("page number"), page));
685        }
686        let mut wal_path = path.as_os_str().to_owned();
687        wal_path.push("-wal");
688        let wal = std::fs::read(&wal_path).unwrap_or_default();
689        if let Some(frames) = wal.get(32..) {
690            for frame in frames.chunks_exact(24 + page_size) {
691                let number = u32::from_be_bytes(frame[..4].try_into().expect("4 bytes"));
692                pages.push(decrypt(number, &frame[24..]));
693            }
694        }
695        pages
696    }
697
698    fn holds(pages: &[Vec<u8>], needle: &[u8]) -> bool {
699        pages
700            .iter()
701            .any(|page| page.windows(needle.len()).any(|w| w == needle))
702    }
703
704    #[test]
705    fn no_copy_of_a_purged_wrapped_key_is_left_in_the_catalog() {
706        let f = fixture();
707        let (uuid, version) = stored(&f, "first");
708        let wrapped = wrapped_key(&f, version);
709        f.store
710            .delete(&f.caller, uuid, Retention::Years(7))
711            .expect("deletes");
712        let path = f.dir.path().join("catalog.sqlite");
713        let key = envelope::catalog_key(&f.store.tenant_key).expect("catalog key");
714        let page_size = crate::catalog::tests::cipher_setting(&f.store.conn(), "cipher_page_size");
715        let page_size: usize = page_size.parse().expect("a page size");
716        // The scan must see the key before the purge, or finding nothing
717        // afterwards would prove nothing.
718        assert!(holds(&plain_pages(&path, &key, page_size), &wrapped));
719        later(&f, Duration::days(2));
720
721        assert_eq!(f.store.purge_due(DAY), Ok(PurgeReport { purged: 1 }));
722
723        assert!(!holds(&plain_pages(&path, &key, page_size), &wrapped));
724        let mut wal = path.as_os_str().to_owned();
725        wal.push("-wal");
726        assert_eq!(
727            std::fs::metadata(&wal).expect("WAL").len(),
728            0,
729            "WAL truncated"
730        );
731    }
732
733    /// Lets the test's reader go when dropped.
734    struct LetGo(mpsc::Sender<()>);
735
736    impl Drop for LetGo {
737        fn drop(&mut self) {
738            let _ = self.0.send(());
739        }
740    }
741
742    #[test]
743    fn the_checkpoint_waits_for_a_reader_without_holding_the_catalog() {
744        let f = fixture();
745        let (uuid, _) = stored(&f, "first");
746        f.store
747            .delete(&f.caller, uuid, Retention::Years(7))
748            .expect("deletes");
749        later(&f, Duration::days(2));
750        let path = f.dir.path().join("catalog.sqlite");
751        let key = envelope::catalog_key(&f.store.tenant_key).expect("catalog key");
752        let mut reader = catalog::open(&path, &key, "").expect("a second connection");
753        // The reader holds its snapshot until this thread lets it go,
754        // after the checks below, so no timing decides what they see.
755        let (reading, started) = mpsc::channel();
756        let (release, released) = mpsc::channel::<()>();
757        let held = std::thread::spawn(move || {
758            let tx = reader.transaction().expect("begin");
759            let _: i64 = tx
760                .query_row("SELECT count(*) FROM version", [], |r| r.get(0))
761                .expect("reads");
762            reading.send(()).expect("signal");
763            released.recv().expect("let go");
764            drop(tx);
765        });
766        started.recv().expect("the reader started");
767        let (busy, waiting) = mpsc::channel();
768        f.store.catalog.on_checkpoint_wait(busy);
769
770        std::thread::scope(|s| {
771            let purging = s.spawn(|| f.store.purge_due(DAY));
772            // Dropped on the way out of this block, also when a check below
773            // fails, so the purge never waits for the reader forever and the
774            // failure is reported instead of hanging the test.
775            let let_go = LetGo(release);
776            waiting
777                .recv_timeout(Wait::from_secs(20))
778                .expect("the purge reaches its checkpoint and waits for the reader");
779            // In a thread of its own, so a catalog held by the checkpoint
780            // fails here instead of hanging the test.
781            let (answer, answered) = mpsc::channel();
782            let store = &f.store;
783            let counting = s.spawn(move || answer.send(store.counts().is_ok()));
784            assert_eq!(
785                answered.recv_timeout(Wait::from_secs(20)),
786                Ok(true),
787                "the catalog answers another call while the checkpoint waits"
788            );
789            counting.join().expect("the count finished").expect("sent");
790            drop(let_go);
791            assert_eq!(
792                purging.join().expect("the purge finished"),
793                Ok(PurgeReport { purged: 1 }),
794                "the purge completes once the reader is gone"
795            );
796        });
797        held.join().expect("the reader finished");
798        let timeout: i64 = f
799            .store
800            .conn()
801            .pragma_query_value(None, "busy_timeout", |r| r.get(0))
802            .expect("busy timeout");
803        assert_eq!(timeout, 5000, "the catalog's busy timeout is restored");
804        let mut wal = path.as_os_str().to_owned();
805        wal.push("-wal");
806        assert_eq!(
807            std::fs::metadata(&wal).expect("WAL").len(),
808            0,
809            "WAL truncated"
810        );
811    }
812}
vcon_server/store/purge.rs

vcon_server/store/
purge.rs
