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

vcon_server/store/
purge.rs
