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

vcon_server/store/
purge.rs
