Skip to main content

vcon_server/
store.rs

1//! One tenant's vCons: a sealed object file per kept version and a catalog
2//! row for it. The catalog is the truth: a version exists once its row is
3//! committed, and an object file with no row is garbage.
4
5use std::path::{Path, PathBuf};
6use std::sync::{Arc, Mutex, MutexGuard};
7
8use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
9use uuid::Uuid;
10use vcon::Inspection;
11
12use crate::audit::{self, Action, Actor, Entry};
13use crate::catalog;
14use crate::clock::{Clock, nanos};
15use crate::envelope::{self, Context, Key, WrapLimit};
16use crate::fs::Fs;
17use crate::ids::{KeyId, TenantId, VersionId};
18use crate::retention::Retention;
19
20mod examine;
21mod legal;
22mod list;
23mod purge;
24mod rekey;
25
26pub use examine::Examination;
27pub use legal::{Legal, VersionInfo};
28pub use list::{Cursor, ListQuery, Listed, Listing};
29pub use purge::{Kept, PurgeReport, due};
30
31/// The tables of a tenant's catalog, besides its audit log. A key column
32/// says NOT NULL because SQLite lets a PRIMARY KEY other than an INTEGER
33/// one hold NULL.
34const TABLES: &str = "
35CREATE TABLE vcon (
36    uuid BLOB PRIMARY KEY NOT NULL,
37    current BLOB,
38    deleted_at INTEGER,
39    hold INTEGER NOT NULL DEFAULT 0,
40    stored_at INTEGER NOT NULL DEFAULT 0
41);
42CREATE INDEX vcon_listed ON vcon (stored_at, uuid);
43CREATE TABLE version (
44    version BLOB PRIMARY KEY NOT NULL,
45    uuid BLOB NOT NULL,
46    form TEXT NOT NULL,
47    gzip INTEGER NOT NULL,
48    size INTEGER NOT NULL,
49    stored_at INTEGER NOT NULL,
50    hidden_at INTEGER,
51    tenant_key_id BLOB NOT NULL,
52    wrapped_key BLOB NOT NULL
53);
54CREATE INDEX version_uuid ON version (uuid);
55CREATE TABLE tombstone (
56    uuid BLOB PRIMARY KEY NOT NULL,
57    purged_at INTEGER NOT NULL
58);
59CREATE TABLE key_use (
60    tenant_key_id BLOB PRIMARY KEY NOT NULL,
61    wraps INTEGER NOT NULL
62);
63";
64
65/// Who is calling, for the audit record.
66#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67pub struct Caller {
68    pub key: KeyId,
69}
70
71/// What `create` did.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub enum Created {
74    /// Stored as a new vCon.
75    New(VersionId),
76    /// The same bytes were already current; nothing was written.
77    Same(VersionId),
78}
79
80/// A version as `get` returns it.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct Stored {
83    pub version: VersionId,
84    pub form: vcon::Form,
85    /// Whether the body is gzipped, as it arrived.
86    pub gzip: bool,
87    /// The body exactly as it arrived.
88    pub body: Vec<u8>,
89}
90
91/// What a tenant's Store keeps, as removal checks it.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub struct Counts {
94    /// Versions kept, hidden ones included.
95    pub versions: u64,
96    /// Uuids under a legal hold.
97    pub held: u64,
98}
99
100/// Why a Store call was refused or failed.
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub enum StoreError {
103    Conflict,
104    Gone,
105    NotFound,
106    NotDeleted,
107    PreconditionFailed,
108    Immutable,
109    /// The tenant key is at its wrap limit and must be rotated.
110    LimitReached,
111    /// A row's object file is missing or does not open.
112    Damaged,
113    /// The audit record could not be written, so nothing was done.
114    AuditFailed,
115    Io,
116}
117
118/// One tenant's vCons.
119pub struct Store {
120    tenant: TenantId,
121    tenant_key: Key,
122    objects: PathBuf,
123    conn: Mutex<Connection>,
124    fs: Arc<dyn Fs>,
125    clock: Arc<dyn Clock>,
126    #[cfg(test)]
127    hooks: tests::Hooks,
128}
129
130impl Store {
131    /// Opens the tenant's catalog under its catalog key, making it if new,
132    /// then runs start-up recovery.
133    pub fn open(
134        dir: &Path,
135        tenant: TenantId,
136        tenant_key: Key,
137        fs: Arc<dyn Fs>,
138        clock: Arc<dyn Clock>,
139    ) -> Result<Store, StoreError> {
140        let store = Store::open_beside(dir, tenant, tenant_key, fs, clock)?;
141        store.recover()?;
142        Ok(store)
143    }
144
145    /// Opens the store as a command running beside a server does: without
146    /// start-up recovery, which would take the server's writes in progress
147    /// for a crash's leftovers.
148    pub fn open_beside(
149        dir: &Path,
150        tenant: TenantId,
151        tenant_key: Key,
152        fs: Arc<dyn Fs>,
153        clock: Arc<dyn Clock>,
154    ) -> Result<Store, StoreError> {
155        let catalog_key = envelope::catalog_key(&tenant_key).map_err(|_| StoreError::Io)?;
156        let schema = format!("{TABLES}{}", audit::SCHEMA);
157        let conn = catalog::open(&dir.join("catalog.sqlite"), &catalog_key, &schema)
158            .map_err(|_| StoreError::Io)?;
159        let objects = dir.join("objects");
160        fs.create_dir_all(&objects).map_err(|_| StoreError::Io)?;
161        // A new directory survives a crash only once the directory that
162        // holds it is synced; without this a crash could take the objects
163        // directory, synced files and all, while the catalog keeps its rows.
164        fs.sync_dir(dir).map_err(|_| StoreError::Io)?;
165        Ok(Store {
166            tenant,
167            tenant_key,
168            objects,
169            conn: Mutex::new(conn),
170            fs,
171            clock,
172            #[cfg(test)]
173            hooks: tests::Hooks::default(),
174        })
175    }
176
177    /// Start-up recovery: the WAL is checkpointed with truncation, and
178    /// temporary files and object files with no row are removed. A name
179    /// that is not a version id, with or without `.tmp`, is not ours and
180    /// is left alone.
181    fn recover(&self) -> Result<(), StoreError> {
182        let conn = self.conn();
183        let busy: i64 = conn
184            .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| r.get(0))
185            .map_err(io)?;
186        if busy != 0 {
187            return Err(StoreError::Io);
188        }
189        let mut removed = false;
190        for name in self.fs.list(&self.objects).map_err(io)? {
191            let Some(name) = name.to_str() else { continue };
192            let (id, temporary) = match name.strip_suffix(".tmp") {
193                Some(id) => (id, true),
194                None => (name, false),
195            };
196            let Some(version) = VersionId::parse(id) else {
197                continue;
198            };
199            let has_row = conn
200                .query_row(
201                    "SELECT 1 FROM version WHERE version = ?1",
202                    [version.0],
203                    |_| Ok(()),
204                )
205                .optional()
206                .map_err(io)?
207                .is_some();
208            if temporary || !has_row {
209                self.fs.remove(&self.objects.join(name)).map_err(io)?;
210                removed = true;
211            }
212        }
213        if removed {
214            self.fs.sync_dir(&self.objects).map_err(io)?;
215        }
216        Ok(())
217    }
218
219    /// Stores a new vCon. The uuid's state is checked before anything is
220    /// sealed, the body is written in the order below, and the
221    /// commit checks the state again, so of two racing writers one wins
222    /// and the other's file is removed.
223    pub fn create(
224        &self,
225        who: &Caller,
226        found: &Inspection,
227        body: &[u8],
228    ) -> Result<Created, StoreError> {
229        let uuid = found.uuid;
230        let entry = |status, version| Entry {
231            actor: Actor::Key(who.key),
232            action: Action::Create,
233            target: Some(uuid.to_string()),
234            version,
235            status,
236        };
237        let key_id = self.tenant_key.id();
238        {
239            let mut conn = self.conn();
240            let tx = immediate(&mut conn)?;
241            let refused = match state(&tx, uuid)? {
242                State::Gone => Some((410, None, StoreError::Gone)),
243                State::Live(current) => match self.read_version(&tx, uuid, current) {
244                    Ok(stored) if stored == body => {
245                        self.record(&tx, &entry(200, Some(current)))?;
246                        commit(tx)?;
247                        return Ok(Created::Same(current));
248                    }
249                    Ok(_) => Some((409, None, StoreError::Conflict)),
250                    Err(StoreError::Damaged) => Some((500, Some(current), StoreError::Damaged)),
251                    Err(e) => return Err(e),
252                },
253                State::Absent if envelope::wrap_limit(wraps(&tx, key_id)?) == WrapLimit::Refuse => {
254                    Some((503, None, StoreError::LimitReached))
255                }
256                // The wrap is counted before it is made, so every wrap is
257                // counted once whether this write wins or loses.
258                State::Absent => {
259                    add_wrap(&tx, key_id)?;
260                    None
261                }
262            };
263            if let Some((status, version, error)) = refused {
264                self.record(&tx, &entry(status, version))?;
265                commit(tx)?;
266                return Err(error);
267            }
268            commit(tx)?;
269        }
270
271        let (version, wrapped, file) = self.write_version(uuid, body)?;
272        self.pause();
273
274        let committed = (|| {
275            let mut conn = self.conn();
276            let tx = immediate(&mut conn)?;
277            match state(&tx, uuid)? {
278                State::Absent => {}
279                lost => {
280                    let (status, error) = match lost {
281                        State::Gone => (410, StoreError::Gone),
282                        _ => (409, StoreError::Conflict),
283                    };
284                    self.record(&tx, &entry(status, None))?;
285                    commit(tx)?;
286                    return Err(error);
287                }
288            }
289            // The version's time, and the uuid's first: the list keeps
290            // its place by it through replace, delete and restore.
291            let stored_at = nanos(self.clock.now());
292            tx.execute(
293                "INSERT INTO version
294                     (version, uuid, form, gzip, size, stored_at, hidden_at, tenant_key_id, wrapped_key)
295                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8)",
296                params![
297                    version.0,
298                    uuid.as_bytes(),
299                    form_name(found.form),
300                    found.gzip,
301                    i64::try_from(body.len()).map_err(io)?,
302                    stored_at,
303                    key_id.bytes(),
304                    wrapped,
305                ],
306            )
307            .map_err(io)?;
308            tx.execute(
309                "INSERT INTO vcon (uuid, current, stored_at) VALUES (?1, ?2, ?3)",
310                params![uuid.as_bytes(), version.0, stored_at],
311            )
312            .map_err(io)?;
313            self.record(&tx, &entry(201, Some(version)))?;
314            commit(tx)?;
315            Ok(Created::New(version))
316        })();
317        if committed.is_err() {
318            // Best effort: start-up recovery removes any file left over.
319            let _ = self.fs.remove(&file);
320        }
321        committed
322    }
323
324    /// Reads the current version of `uuid`. The body is decrypted whole,
325    /// then one transaction checks that the version is still current and
326    /// commits the audit record; only then is the body returned. If the
327    /// version changed meanwhile, the read starts again.
328    pub fn get(&self, who: &Caller, uuid: Uuid) -> Result<Stored, StoreError> {
329        let entry = |status, version| Entry {
330            actor: Actor::Key(who.key),
331            action: Action::Read,
332            target: Some(uuid.to_string()),
333            version,
334            status,
335        };
336        let refuse = |tx: Transaction, state: State| {
337            let (status, error) = match state {
338                State::Gone => (410, StoreError::Gone),
339                _ => (404, StoreError::NotFound),
340            };
341            self.record(&tx, &entry(status, None))?;
342            commit(tx)?;
343            Err(error)
344        };
345        loop {
346            let (version, row) = {
347                let mut conn = self.conn();
348                let tx = immediate(&mut conn)?;
349                match state(&tx, uuid)? {
350                    State::Live(version) => (version, version_row(&tx, version)),
351                    other => return refuse(tx, other),
352                }
353            };
354            let opened = row.and_then(|row| {
355                let body = self.open_version(uuid, version, &row.wrapped)?;
356                Ok(Stored {
357                    version,
358                    form: row.form,
359                    gzip: row.gzip,
360                    body,
361                })
362            });
363            self.pause();
364            let mut conn = self.conn();
365            let tx = immediate(&mut conn)?;
366            match state(&tx, uuid)? {
367                State::Live(now) if now == version => {
368                    let status = if opened.is_ok() { 200 } else { 500 };
369                    self.record(&tx, &entry(status, Some(version)))?;
370                    commit(tx)?;
371                    return opened;
372                }
373                State::Live(_) => continue,
374                other => return refuse(tx, other),
375            }
376        }
377    }
378
379    /// Replaces the current version of `found.uuid`, if it is `if_match`
380    /// and neither signed nor encrypted. The caller has checked that the
381    /// body's uuid is the one asked for. The old version is hidden, and
382    /// kept for legal reads until it is purged. As in `create`, the state
383    /// is checked before sealing and again at the commit, so of two racing
384    /// replaces one wins and the other's file is removed.
385    pub fn replace(
386        &self,
387        who: &Caller,
388        if_match: VersionId,
389        found: &Inspection,
390        body: &[u8],
391    ) -> Result<VersionId, StoreError> {
392        let uuid = found.uuid;
393        let entry = |status, version| Entry {
394            actor: Actor::Key(who.key),
395            action: Action::Replace,
396            target: Some(uuid.to_string()),
397            version,
398            status,
399        };
400        let key_id = self.tenant_key.id();
401        // Why the state does not allow this replace, if it does not.
402        let refusal = |tx: &Transaction| -> Result<Option<(u16, StoreError)>, StoreError> {
403            Ok(match state(tx, uuid)? {
404                State::Absent => Some((404, StoreError::NotFound)),
405                State::Gone => Some((410, StoreError::Gone)),
406                State::Live(current) if version_row(tx, current)?.form != vcon::Form::Unsigned => {
407                    Some((409, StoreError::Immutable))
408                }
409                State::Live(current) if current != if_match => {
410                    Some((412, StoreError::PreconditionFailed))
411                }
412                State::Live(_) => None,
413            })
414        };
415        {
416            let mut conn = self.conn();
417            let tx = immediate(&mut conn)?;
418            let refused = match refusal(&tx)? {
419                None if envelope::wrap_limit(wraps(&tx, key_id)?) == WrapLimit::Refuse => {
420                    Some((503, StoreError::LimitReached))
421                }
422                other => other,
423            };
424            if let Some((status, error)) = refused {
425                self.record(&tx, &entry(status, None))?;
426                commit(tx)?;
427                return Err(error);
428            }
429            add_wrap(&tx, key_id)?;
430            commit(tx)?;
431        }
432
433        let (version, wrapped, file) = self.write_version(uuid, body)?;
434        self.pause();
435
436        let committed = (|| {
437            let mut conn = self.conn();
438            let tx = immediate(&mut conn)?;
439            if let Some((status, error)) = refusal(&tx)? {
440                self.record(&tx, &entry(status, None))?;
441                commit(tx)?;
442                return Err(error);
443            }
444            let now = nanos(self.clock.now());
445            tx.execute(
446                "INSERT INTO version
447                     (version, uuid, form, gzip, size, stored_at, hidden_at, tenant_key_id, wrapped_key)
448                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8)",
449                params![
450                    version.0,
451                    uuid.as_bytes(),
452                    form_name(found.form),
453                    found.gzip,
454                    i64::try_from(body.len()).map_err(io)?,
455                    now,
456                    key_id.bytes(),
457                    wrapped,
458                ],
459            )
460            .map_err(io)?;
461            tx.execute(
462                "UPDATE version SET hidden_at = ?1 WHERE version = ?2",
463                params![now, if_match.0],
464            )
465            .map_err(io)?;
466            tx.execute(
467                "UPDATE vcon SET current = ?1 WHERE uuid = ?2",
468                params![version.0, uuid.as_bytes()],
469            )
470            .map_err(io)?;
471            self.record(&tx, &entry(200, Some(version)))?;
472            commit(tx)?;
473            Ok(version)
474        })();
475        if committed.is_err() {
476            // Best effort: start-up recovery removes any file left over.
477            let _ = self.fs.remove(&file);
478        }
479        committed
480    }
481
482    /// Deletes `uuid`: one transaction marks it deleted, empties its
483    /// current version and hides every kept version, with the audit
484    /// record. Its versions stay for legal reads until they are purged;
485    /// any already as old as `retention` is purged at once, so a zero
486    /// period erases the vCon inside the request.
487    /// Deleting a deleted or purged uuid succeeds again. A legal hold does
488    /// not stop a delete, only a purge.
489    pub fn delete(&self, who: &Caller, uuid: Uuid, retention: Retention) -> Result<(), StoreError> {
490        let entry = |status| Entry {
491            actor: Actor::Key(who.key),
492            action: Action::Delete,
493            target: Some(uuid.to_string()),
494            version: None,
495            status,
496        };
497        let mut conn = self.conn();
498        let tx = immediate(&mut conn)?;
499        let outcome = match state(&tx, uuid)? {
500            State::Absent => Err(StoreError::NotFound),
501            State::Gone => Ok(()),
502            State::Live(_) => {
503                let now = nanos(self.clock.now());
504                tx.execute(
505                    "UPDATE vcon SET current = NULL, deleted_at = ?1 WHERE uuid = ?2",
506                    params![now, uuid.as_bytes()],
507                )
508                .map_err(io)?;
509                tx.execute(
510                    "UPDATE version SET hidden_at = ?1 WHERE uuid = ?2 AND hidden_at IS NULL",
511                    params![now, uuid.as_bytes()],
512                )
513                .map_err(io)?;
514                Ok(())
515            }
516        };
517        let status = if outcome.is_ok() { 204 } else { 404 };
518        self.record(&tx, &entry(status))?;
519        commit(tx)?;
520        outcome?;
521        drop(conn);
522        // A version already as old as `retention` is erased now, not at
523        // the next sweep.
524        self.purge_hidden(Actor::Key(who.key), retention, Some(uuid))?;
525        Ok(())
526    }
527
528    /// Every version whose object file is missing. Such a version answers
529    /// 500 when read; it is reported, never dropped.
530    pub fn check(&self) -> Result<Vec<VersionId>, StoreError> {
531        let conn = self.conn();
532        let mut rows = conn
533            .prepare("SELECT version FROM version ORDER BY version")
534            .map_err(io)?;
535        let versions = rows
536            .query_map([], |r| r.get::<_, [u8; 16]>(0))
537            .map_err(io)?
538            .collect::<Result<Vec<_>, _>>()
539            .map_err(io)?;
540        let present: std::collections::BTreeSet<_> = self
541            .fs
542            .list(&self.objects)
543            .map_err(io)?
544            .into_iter()
545            .collect();
546        Ok(versions
547            .into_iter()
548            .map(VersionId)
549            .filter(|v| !present.contains(std::ffi::OsStr::new(&v.to_string())))
550            .collect())
551    }
552
553    /// Seals `body` as a new version of `uuid` under a new data key, wraps
554    /// that key under the tenant key, and writes the object file. Returns
555    /// the version id, the wrapped key and the file's path.
556    fn write_version(
557        &self,
558        uuid: Uuid,
559        body: &[u8],
560    ) -> Result<(VersionId, Vec<u8>, PathBuf), StoreError> {
561        let version = VersionId::generate();
562        let ctx = Context::Version {
563            tenant: self.tenant,
564            uuid,
565            version,
566        };
567        let data_key = Key::generate().map_err(|_| StoreError::Io)?;
568        let blob = envelope::seal(&data_key, &ctx, body);
569        let wrapped = envelope::wrap(&self.tenant_key, &ctx, &data_key);
570        let file = self.write_object(version, &blob)?;
571        Ok((version, wrapped, file))
572    }
573
574    /// In tests, runs the hook set for the gap between work done outside
575    /// the catalog lock and the transaction that confirms it.
576    fn pause(&self) {
577        #[cfg(test)]
578        {
579            let hook = self.hooks.between.lock().expect("hook").take();
580            if let Some(hook) = hook {
581                hook();
582            }
583        }
584    }
585
586    /// Writes a sealed body in order: a temporary file, synced,
587    /// renamed to its version id, and the directory synced. Returns the
588    /// file's path.
589    fn write_object(&self, version: VersionId, blob: &[u8]) -> Result<PathBuf, StoreError> {
590        let tmp = self.objects.join(format!("{version}.tmp"));
591        let file = self.objects.join(version.to_string());
592        let written = self
593            .fs
594            .write_new(&tmp, blob)
595            .and_then(|()| self.fs.sync_file(&tmp))
596            .and_then(|()| self.fs.rename(&tmp, &file))
597            .and_then(|()| self.fs.sync_dir(&self.objects));
598        if written.is_err() {
599            // Best effort: start-up recovery removes any file left over.
600            let _ = self.fs.remove(&tmp);
601            let _ = self.fs.remove(&file);
602            return Err(StoreError::Io);
603        }
604        Ok(file)
605    }
606
607    /// The plaintext of a stored version. A missing row or file, or one
608    /// that does not open, is damage.
609    fn read_version(
610        &self,
611        tx: &Transaction,
612        uuid: Uuid,
613        version: VersionId,
614    ) -> Result<Vec<u8>, StoreError> {
615        let row = version_row(tx, version)?;
616        self.open_version(uuid, version, &row.wrapped)
617    }
618
619    /// Opens a version's object file with its wrapped data key.
620    fn open_version(
621        &self,
622        uuid: Uuid,
623        version: VersionId,
624        wrapped: &[u8],
625    ) -> Result<Vec<u8>, StoreError> {
626        let ctx = Context::Version {
627            tenant: self.tenant,
628            uuid,
629            version,
630        };
631        let data_key =
632            envelope::unwrap(&self.tenant_key, &ctx, wrapped).map_err(|_| StoreError::Damaged)?;
633        let blob = self
634            .fs
635            .read(&self.objects.join(version.to_string()))
636            .map_err(|_| StoreError::Damaged)?;
637        envelope::open(&data_key, &ctx, &blob).map_err(|_| StoreError::Damaged)
638    }
639
640    /// How many versions the Store keeps and how many uuids are held.
641    pub fn counts(&self) -> Result<Counts, StoreError> {
642        let conn = self.conn();
643        let count = |sql| -> Result<u64, StoreError> {
644            let n: i64 = conn.query_row(sql, [], |r| r.get(0)).map_err(io)?;
645            u64::try_from(n).map_err(io)
646        };
647        Ok(Counts {
648            versions: count("SELECT count(*) FROM version")?,
649            held: count("SELECT count(*) FROM vcon WHERE hold != 0")?,
650        })
651    }
652
653    /// Writes the tenant's log after `since` as JSON lines, oldest first.
654    /// Not recorded: reading the log changes nothing.
655    pub fn export(
656        &self,
657        since: Option<i64>,
658        out: &mut dyn std::io::Write,
659    ) -> Result<(), audit::AuditError> {
660        audit::export(&self.conn(), since, out)
661    }
662
663    /// A page of the tenant's log as JSON lines, the lines `export` writes
664    /// for the same records. The read is recorded as audit_read in the same
665    /// log and transaction, after the page is read, so a page never holds
666    /// its own record.
667    pub fn audit_page(
668        &self,
669        who: &Caller,
670        query: &audit::AuditQuery,
671    ) -> Result<Vec<u8>, StoreError> {
672        let mut conn = self.conn();
673        let tx = immediate(&mut conn)?;
674        let mut out = Vec::new();
675        let read = audit::page(&tx, query, &mut out);
676        self.record(
677            &tx,
678            &Entry {
679                actor: Actor::Key(who.key),
680                action: Action::AuditRead,
681                target: None,
682                version: None,
683                status: if read.is_ok() { 200 } else { 500 },
684            },
685        )?;
686        commit(tx)?;
687        read.map(|()| out).map_err(io)
688    }
689
690    /// Records, in a transaction of its own, an action on the tenant taken
691    /// elsewhere, such as a change to its retention period.
692    pub fn log(&self, entry: &Entry) -> Result<(), StoreError> {
693        let mut conn = self.conn();
694        let tx = immediate(&mut conn)?;
695        self.record(&tx, entry)?;
696        commit(tx)
697    }
698
699    /// Makes every audit insert on this Store's connection fail, for tests.
700    #[cfg(test)]
701    pub(crate) fn break_audit(&self) {
702        self.conn()
703            .execute_batch(
704                "CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
705                 BEGIN SELECT RAISE(ABORT, 'no audit'); END;",
706            )
707            .expect("trigger");
708    }
709
710    /// The catalog connection, for tests that damage or inspect it.
711    #[cfg(test)]
712    pub(crate) fn conn_for_test(&self) -> MutexGuard<'_, Connection> {
713        self.conn()
714    }
715
716    /// The tenant's log as `export` writes it, for tests.
717    #[cfg(test)]
718    pub(crate) fn records(&self) -> Vec<serde_json::Value> {
719        crate::audit::testing::exported(&self.conn(), None)
720    }
721
722    /// Counts, in a transaction of its own, an event the tenant's log
723    /// keeps one record per minute for, such as a failed check of one of
724    /// its keys.
725    pub fn count(&self, entry: &Entry) -> Result<(), StoreError> {
726        let mut conn = self.conn();
727        let tx = immediate(&mut conn)?;
728        audit::count(&tx, entry, self.clock.now()).map_err(|_| StoreError::AuditFailed)?;
729        commit(tx)
730    }
731
732    /// Writes an audit record in `tx`; if it cannot, nothing is done.
733    fn record(&self, tx: &Transaction, entry: &Entry) -> Result<(), StoreError> {
734        audit::record(tx, entry, self.clock.now()).map_err(|_| StoreError::AuditFailed)
735    }
736
737    fn conn(&self) -> MutexGuard<'_, Connection> {
738        self.conn
739            .lock()
740            .expect("no thread panicked holding the catalog")
741    }
742}
743
744/// What the catalog says about a uuid.
745enum State {
746    /// Never stored.
747    Absent,
748    /// Stored, with this current version.
749    Live(VersionId),
750    /// Deleted or purged.
751    Gone,
752}
753
754fn state(tx: &Transaction, uuid: Uuid) -> Result<State, StoreError> {
755    let purged = tx
756        .query_row(
757            "SELECT 1 FROM tombstone WHERE uuid = ?1",
758            [uuid.as_bytes()],
759            |_| Ok(()),
760        )
761        .optional()
762        .map_err(io)?;
763    if purged.is_some() {
764        return Ok(State::Gone);
765    }
766    let row: Option<(Option<[u8; 16]>, Option<i64>)> = tx
767        .query_row(
768            "SELECT current, deleted_at FROM vcon WHERE uuid = ?1",
769            [uuid.as_bytes()],
770            |r| Ok((r.get(0)?, r.get(1)?)),
771        )
772        .optional()
773        .map_err(io)?;
774    match row {
775        None => Ok(State::Absent),
776        Some((_, Some(_))) => Ok(State::Gone),
777        Some((Some(current), None)) => Ok(State::Live(VersionId(current))),
778        // A live row always names its current version.
779        Some((None, None)) => Err(StoreError::Damaged),
780    }
781}
782
783/// What a version row holds that a read needs.
784struct VersionRow {
785    form: vcon::Form,
786    gzip: bool,
787    wrapped: Vec<u8>,
788}
789
790/// A version's row. One that is missing or unreadable is damage.
791fn version_row(tx: &Transaction, version: VersionId) -> Result<VersionRow, StoreError> {
792    let row: Option<(String, bool, Vec<u8>)> = tx
793        .query_row(
794            "SELECT form, gzip, wrapped_key FROM version WHERE version = ?1",
795            [version.0],
796            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
797        )
798        .optional()
799        .map_err(io)?;
800    let (form, gzip, wrapped) = row.ok_or(StoreError::Damaged)?;
801    let form = match form.as_str() {
802        "unsigned" => vcon::Form::Unsigned,
803        "signed" => vcon::Form::Signed,
804        "encrypted" => vcon::Form::Encrypted,
805        _ => return Err(StoreError::Damaged),
806    };
807    Ok(VersionRow {
808        form,
809        gzip,
810        wrapped,
811    })
812}
813
814/// Data keys wrapped so far under the tenant key `key`.
815fn wraps(tx: &Transaction, key: KeyId) -> Result<u64, StoreError> {
816    let wraps: Option<i64> = tx
817        .query_row(
818            "SELECT wraps FROM key_use WHERE tenant_key_id = ?1",
819            [key.bytes()],
820            |r| r.get(0),
821        )
822        .optional()
823        .map_err(io)?;
824    u64::try_from(wraps.unwrap_or(0)).map_err(io)
825}
826
827fn add_wrap(tx: &Transaction, key: KeyId) -> Result<(), StoreError> {
828    tx.execute(
829        "INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, 1)
830         ON CONFLICT (tenant_key_id) DO UPDATE SET wraps = wraps + 1",
831        [key.bytes()],
832    )
833    .map_err(io)?;
834    Ok(())
835}
836
837fn immediate(conn: &mut Connection) -> Result<Transaction<'_>, StoreError> {
838    conn.transaction_with_behavior(TransactionBehavior::Immediate)
839        .map_err(io)
840}
841
842fn commit(tx: Transaction) -> Result<(), StoreError> {
843    tx.commit().map_err(io)
844}
845
846fn io<E>(_: E) -> StoreError {
847    StoreError::Io
848}
849
850fn form_name(form: vcon::Form) -> &'static str {
851    match form {
852        vcon::Form::Unsigned => "unsigned",
853        vcon::Form::Signed => "signed",
854        vcon::Form::Encrypted => "encrypted",
855    }
856}
857
858/// Stored vCons, for tests.
859#[cfg(test)]
860pub(crate) mod testing {
861    use std::collections::BTreeSet;
862
863    use vcon::{Form, Inspection, Limits, inspect};
864
865    /// An unsigned vCon with this uuid, and what `inspect` makes of it.
866    pub(crate) fn vcon(uuid: &str, subject: &str) -> (Inspection, Vec<u8>) {
867        let body = format!(
868            r#"{{"uuid": "{uuid}", "created_at": "2026-10-23T12:00:00Z", "subject": "{subject}", "parties": []}}"#
869        )
870        .into_bytes();
871        let found = inspect(&body, None, &Limits { inflated: 1 << 20 }, &BTreeSet::new())
872            .expect("a valid vCon");
873        assert_eq!(found.form, Form::Unsigned);
874        (found, body)
875    }
876}
877
878#[cfg(test)]
879mod tests {
880    use std::collections::BTreeSet;
881
882    use serde_json::Value;
883    use time::macros::datetime;
884    use vcon::{Limits, inspect};
885
886    use super::testing::vcon;
887    use super::*;
888    use crate::audit::testing::exported;
889    use crate::catalog::testing::columns_taking_null;
890    use crate::clock::ManualClock;
891    use crate::fs::{CrashFs, Op};
892
893    const UUID: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc46";
894
895    /// Test-only points where a test can step in.
896    #[derive(Default)]
897    pub(super) struct Hooks {
898        /// Runs once, between work done outside the catalog lock (a
899        /// write's files, a read's decryption) and the transaction that
900        /// confirms it.
901        pub(super) between: Mutex<Option<Box<dyn FnOnce() + Send>>>,
902    }
903
904    pub(super) struct Fixture {
905        pub(super) dir: tempfile::TempDir,
906        pub(super) fs: Arc<CrashFs>,
907        pub(super) clock: Arc<ManualClock>,
908        /// The tenant key, wrapped under `master`, so a test can open the
909        /// store again with the same key.
910        master: Key,
911        wrapped: Vec<u8>,
912        pub(super) store: Store,
913        pub(super) caller: Caller,
914    }
915
916    pub(super) const TENANT: TenantId = TenantId([0x11; 16]);
917
918    impl Fixture {
919        pub(super) fn open(&self) -> Result<Store, StoreError> {
920            let ctx = Context::Tenant { tenant: TENANT };
921            let tenant_key = envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps");
922            Store::open(
923                self.dir.path(),
924                TENANT,
925                tenant_key,
926                self.fs.clone(),
927                self.clock.clone(),
928            )
929        }
930
931        /// The tenant key the Store was made under.
932        pub(super) fn tenant_key(&self) -> Key {
933            let ctx = Context::Tenant { tenant: TENANT };
934            envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps")
935        }
936
937        pub(super) fn tenant(&self) -> TenantId {
938            TENANT
939        }
940
941        /// Opens the store again as a command beside a server would.
942        fn open_beside(&self) -> Result<Store, StoreError> {
943            let ctx = Context::Tenant { tenant: TENANT };
944            let tenant_key = envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps");
945            Store::open_beside(
946                self.dir.path(),
947                TENANT,
948                tenant_key,
949                self.fs.clone(),
950                self.clock.clone(),
951            )
952        }
953
954        /// Closes the store and opens it again, as a restart would.
955        pub(super) fn reopen(&mut self) {
956            let placeholder = self.open().expect("opens");
957            drop(std::mem::replace(&mut self.store, placeholder));
958            self.store = self.open().expect("reopens");
959        }
960    }
961
962    pub(super) fn fixture() -> Fixture {
963        let dir = tempfile::tempdir().expect("temp dir");
964        let fs = Arc::new(CrashFs::new(dir.path()));
965        let clock = Arc::new(ManualClock::at(datetime!(2026-10-23 12:00 UTC)));
966        let master = Key::generate().expect("master key");
967        let tenant_key = Key::generate().expect("tenant key");
968        let wrapped = envelope::wrap(&master, &Context::Tenant { tenant: TENANT }, &tenant_key);
969        let store =
970            Store::open(dir.path(), TENANT, tenant_key, fs.clone(), clock.clone()).expect("opens");
971        let caller = Caller {
972            key: Key::generate().expect("caller key").id(),
973        };
974        Fixture {
975            dir,
976            fs,
977            clock,
978            master,
979            wrapped,
980            store,
981            caller,
982        }
983    }
984
985    fn audit_log(store: &Store) -> Vec<Value> {
986        exported(&store.conn(), None)
987    }
988
989    fn count(store: &Store, sql: &str) -> i64 {
990        store.conn().query_row(sql, [], |r| r.get(0)).expect(sql)
991    }
992
993    fn wraps(store: &Store) -> i64 {
994        store
995            .conn()
996            .query_row("SELECT coalesce(sum(wraps), 0) FROM key_use", [], |r| {
997                r.get(0)
998            })
999            .expect("wraps")
1000    }
1001
1002    /// The body the store holds for `version`, opened as `get` will.
1003    fn stored_body(f: &Fixture, uuid: Uuid, version: VersionId) -> Vec<u8> {
1004        let wrapped: Vec<u8> = f
1005            .store
1006            .conn()
1007            .query_row(
1008                "SELECT wrapped_key FROM version WHERE version = ?1",
1009                [version.0],
1010                |r| r.get(0),
1011            )
1012            .expect("row");
1013        let ctx = Context::Version {
1014            tenant: f.store.tenant,
1015            uuid,
1016            version,
1017        };
1018        let data_key = envelope::unwrap(&f.store.tenant_key, &ctx, &wrapped).expect("unwraps");
1019        let blob =
1020            f.fs.read(&f.store.objects.join(version.to_string()))
1021                .expect("file");
1022        envelope::open(&data_key, &ctx, &blob).expect("opens")
1023    }
1024
1025    fn audit_page(f: &Fixture, after: i64, limit: u32) -> Result<Vec<u8>, StoreError> {
1026        f.store
1027            .audit_page(&f.caller, &audit::AuditQuery { after, limit })
1028    }
1029
1030    fn export_text(store: &Store, since: Option<i64>) -> Vec<u8> {
1031        let mut out = Vec::new();
1032        store.export(since, &mut out).expect("export");
1033        out
1034    }
1035
1036    #[test]
1037    fn an_audit_page_is_export_s_lines_and_records_its_read_after_them() {
1038        let f = fixture();
1039        for n in 0..3 {
1040            let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-27900000000{n}"), "x");
1041            f.store.create(&f.caller, &found, &body).expect("stored");
1042        }
1043        let before = export_text(&f.store, None);
1044        assert_eq!(audit_page(&f, 0, 1000), Ok(before.clone()));
1045        let read = audit_log(&f.store);
1046        assert_eq!(read.len(), 4, "three creates and the read");
1047        assert_eq!(read[3]["action"], "audit_read");
1048        assert_eq!(read[3]["actor"], f.caller.key.to_string());
1049        assert_eq!(read[3]["status"], 200);
1050        assert_eq!(read[3]["seq"], 4);
1051        assert_eq!(read[3]["target"], Value::Null);
1052        // A page of two from seq 1: the second and third creates.
1053        let lines: Vec<&[u8]> = before.split_inclusive(|b| *b == b'\n').collect();
1054        assert_eq!(audit_page(&f, 1, 2), Ok(lines[1..3].concat()));
1055        // Past the end: nothing but the earlier reads' records.
1056        let tail = audit_page(&f, 4, 1000).expect("a page");
1057        assert_eq!(tail, export_text(&f.store, Some(4))[..tail.len()]);
1058        assert_eq!(String::from_utf8(tail).expect("UTF-8").lines().count(), 1);
1059    }
1060
1061    /// A record that cannot be read back (a status past any HTTP status)
1062    /// fails the page, and the read is still recorded, as 500.
1063    #[test]
1064    fn an_audit_page_that_cannot_be_read_is_a_failure_and_recorded() {
1065        let f = fixture();
1066        f.store
1067            .conn_for_test()
1068            .execute(
1069                "INSERT INTO audit (at, actor, action, status, count)
1070                 VALUES ('2026-10-23T12:00:00Z', 'cli', 'read', 70000, 1)",
1071                [],
1072            )
1073            .expect("damaged");
1074        assert_eq!(audit_page(&f, 0, 10), Err(StoreError::Io));
1075        let mut recorded = Vec::new();
1076        f.store.export(Some(1), &mut recorded).expect("export");
1077        let recorded: Value = serde_json::from_slice(&recorded).expect("one record");
1078        assert_eq!(recorded["action"], "audit_read");
1079        assert_eq!(recorded["status"], 500);
1080    }
1081
1082    #[test]
1083    fn an_audit_page_whose_record_fails_is_refused() {
1084        let f = fixture();
1085        f.store.break_audit();
1086        assert_eq!(audit_page(&f, 0, 10), Err(StoreError::AuditFailed));
1087    }
1088
1089    #[test]
1090    fn a_new_vcon_is_written_in_order_and_recorded() {
1091        let f = fixture();
1092        let (found, body) = vcon(UUID, "first");
1093        let ops = f.fs.log().len();
1094        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1095            panic!("a new vCon")
1096        };
1097        let (tmp, done) = (
1098            f.store.objects.join(format!("{version}.tmp")),
1099            f.store.objects.join(version.to_string()),
1100        );
1101        assert_eq!(
1102            f.fs.log()[ops..],
1103            [
1104                Op::Write(tmp.clone()),
1105                Op::SyncFile(tmp.clone()),
1106                Op::Rename(tmp, done),
1107                Op::SyncDir(f.store.objects.clone()),
1108            ]
1109        );
1110        assert_eq!(stored_body(&f, found.uuid, version), body);
1111        assert_eq!(
1112            count(
1113                &f.store,
1114                "SELECT count(*) FROM version WHERE hidden_at IS NULL"
1115            ),
1116            1
1117        );
1118        let current: Vec<u8> = f
1119            .store
1120            .conn()
1121            .query_row(
1122                "SELECT current FROM vcon WHERE uuid = ?1",
1123                [found.uuid.as_bytes()],
1124                |r| r.get(0),
1125            )
1126            .expect("vcon row");
1127        assert_eq!(current, version.0);
1128        assert_eq!(wraps(&f.store), 1, "one data key wrapped");
1129        let log = audit_log(&f.store);
1130        assert_eq!(log.len(), 1);
1131        assert_eq!(log[0]["action"], "create");
1132        assert_eq!(log[0]["actor"], f.caller.key.to_string());
1133        assert_eq!(log[0]["target"], UUID);
1134        assert_eq!(log[0]["version"], version.to_string());
1135        assert_eq!(log[0]["status"], 201);
1136    }
1137
1138    #[test]
1139    fn the_same_body_again_is_same_and_writes_nothing() {
1140        let f = fixture();
1141        let (found, body) = vcon(UUID, "first");
1142        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1143            panic!("a new vCon")
1144        };
1145        let ops = f.fs.log().len();
1146        assert_eq!(
1147            f.store.create(&f.caller, &found, &body),
1148            Ok(Created::Same(version))
1149        );
1150        assert_eq!(f.fs.log().len(), ops, "no file written");
1151        assert_eq!(count(&f.store, "SELECT count(*) FROM version"), 1);
1152        assert_eq!(wraps(&f.store), 1, "nothing wrapped");
1153        let log = audit_log(&f.store);
1154        assert_eq!(
1155            (log[1]["status"].clone(), log[1]["version"].clone()),
1156            (200.into(), version.to_string().into())
1157        );
1158    }
1159
1160    #[test]
1161    fn another_body_for_a_stored_uuid_is_a_conflict_and_writes_nothing() {
1162        let f = fixture();
1163        let (found, body) = vcon(UUID, "first");
1164        f.store.create(&f.caller, &found, &body).expect("stored");
1165        let ops = f.fs.log().len();
1166        // The same length as the first body, so only the bytes differ.
1167        let (found, other) = vcon(UUID, "other");
1168        assert_eq!(
1169            f.store.create(&f.caller, &found, &other),
1170            Err(StoreError::Conflict)
1171        );
1172        assert_eq!(f.fs.log().len(), ops, "no file written");
1173        assert_eq!(wraps(&f.store), 1, "nothing wrapped");
1174        assert_eq!(audit_log(&f.store)[1]["status"], 409);
1175    }
1176
1177    #[test]
1178    fn a_deleted_or_purged_uuid_is_gone_and_writes_nothing() {
1179        for sql in [
1180            "INSERT INTO vcon (uuid, current, deleted_at) VALUES (?1, NULL, 1)",
1181            "INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
1182        ] {
1183            let f = fixture();
1184            let (found, body) = vcon(UUID, "first");
1185            f.store
1186                .conn()
1187                .execute(sql, [found.uuid.as_bytes()])
1188                .expect(sql);
1189            let ops = f.fs.log().len();
1190            assert_eq!(
1191                f.store.create(&f.caller, &found, &body),
1192                Err(StoreError::Gone),
1193                "{sql}"
1194            );
1195            assert_eq!(f.fs.log().len(), ops, "no file written");
1196            assert_eq!(wraps(&f.store), 0, "nothing wrapped");
1197            assert_eq!(audit_log(&f.store)[0]["status"], 410);
1198        }
1199    }
1200
1201    #[test]
1202    fn counts_include_hidden_versions_and_held_uuids() {
1203        let f = fixture();
1204        let none = Counts {
1205            versions: 0,
1206            held: 0,
1207        };
1208        assert_eq!(f.store.counts(), Ok(none));
1209        let (found, old) = stored(&f, "first");
1210        let (_, newer) = vcon(UUID, "second");
1211        f.store
1212            .replace(&f.caller, old, &found, &newer)
1213            .expect("replaced");
1214        f.store
1215            .delete(&f.caller, found.uuid, Retention::Years(7))
1216            .expect("deleted");
1217        let (other, body) = vcon("01a07da8-c2bb-83e5-b9a2-279e0d16bc47", "other");
1218        f.store.create(&f.caller, &other, &body).expect("stored");
1219        assert_eq!(
1220            f.store.counts(),
1221            Ok(Counts {
1222                versions: 3,
1223                held: 0
1224            }),
1225            "two hidden versions and a live one"
1226        );
1227        f.store
1228            .conn()
1229            .execute(
1230                "UPDATE vcon SET hold = 1 WHERE uuid = ?1",
1231                [found.uuid.as_bytes()],
1232            )
1233            .expect("a hold");
1234        assert_eq!(f.store.counts().map(|c| c.held), Ok(1));
1235    }
1236
1237    #[test]
1238    fn a_negative_wrap_count_is_damage_and_stores_nothing() {
1239        let f = fixture();
1240        f.store
1241            .conn()
1242            .execute(
1243                "INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, -1)",
1244                [f.store.tenant_key.id().bytes()],
1245            )
1246            .expect("a damaged count");
1247        let (found, body) = vcon(UUID, "first");
1248        let ops = f.fs.log().len();
1249        assert_eq!(
1250            f.store.create(&f.caller, &found, &body),
1251            Err(StoreError::Io)
1252        );
1253        assert_eq!(f.fs.log().len(), ops, "no file written");
1254    }
1255
1256    #[test]
1257    fn a_tenant_key_at_its_wrap_limit_refuses_new_vcons() {
1258        let f = fixture();
1259        f.store
1260            .conn()
1261            .execute(
1262                "INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, ?2)",
1263                params![f.store.tenant_key.id().bytes(), 1_i64 << 32],
1264            )
1265            .expect("at the limit");
1266        let (found, body) = vcon(UUID, "first");
1267        let ops = f.fs.log().len();
1268        assert_eq!(
1269            f.store.create(&f.caller, &found, &body),
1270            Err(StoreError::LimitReached)
1271        );
1272        assert_eq!(f.fs.log().len(), ops, "no file written");
1273        assert_eq!(audit_log(&f.store)[0]["status"], 503);
1274    }
1275
1276    #[test]
1277    fn a_missing_current_file_is_damage_not_a_conflict() {
1278        let f = fixture();
1279        let (found, body) = vcon(UUID, "first");
1280        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1281            panic!("a new vCon")
1282        };
1283        f.fs.remove(&f.store.objects.join(version.to_string()))
1284            .expect("remove");
1285        assert_eq!(
1286            f.store.create(&f.caller, &found, &body),
1287            Err(StoreError::Damaged)
1288        );
1289        assert_eq!(audit_log(&f.store)[1]["status"], 500);
1290    }
1291
1292    #[test]
1293    fn a_racing_create_that_loses_gets_a_conflict_and_leaves_no_file() {
1294        let f = Arc::new(fixture());
1295        let (found, body) = vcon(UUID, "first");
1296        let winner = f.clone();
1297        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
1298            let (found, other) = vcon(UUID, "second");
1299            assert!(matches!(
1300                winner.store.create(&winner.caller, &found, &other),
1301                Ok(Created::New(_))
1302            ));
1303        }));
1304        assert_eq!(
1305            f.store.create(&f.caller, &found, &body),
1306            Err(StoreError::Conflict)
1307        );
1308        assert_eq!(
1309            count(&f.store, "SELECT count(*) FROM version"),
1310            1,
1311            "only the winner"
1312        );
1313        assert_eq!(
1314            f.fs.list(&f.store.objects).expect("list").len(),
1315            1,
1316            "the loser's file is gone"
1317        );
1318        assert_eq!(wraps(&f.store), 2, "both wraps are counted");
1319        let statuses: Vec<_> = audit_log(&f.store)
1320            .iter()
1321            .map(|r| r["status"].clone())
1322            .collect();
1323        assert_eq!(statuses, [201, 409]);
1324    }
1325
1326    #[test]
1327    fn a_create_whose_audit_record_fails_leaves_no_row_and_no_file() {
1328        let f = fixture();
1329        // Every audit insert now fails, as a full disk or a broken log would.
1330        f.store
1331            .conn()
1332            .execute_batch(
1333                "CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
1334                 BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
1335            )
1336            .expect("trigger");
1337        let (found, body) = vcon(UUID, "first");
1338        assert_eq!(
1339            f.store.create(&f.caller, &found, &body),
1340            Err(StoreError::AuditFailed)
1341        );
1342        assert_eq!(count(&f.store, "SELECT count(*) FROM version"), 0);
1343        assert_eq!(count(&f.store, "SELECT count(*) FROM vcon"), 0);
1344        assert_eq!(
1345            f.fs.list(&f.store.objects).expect("list"),
1346            Vec::<std::ffi::OsString>::new()
1347        );
1348    }
1349
1350    fn get_entry_status(f: &Fixture) -> Value {
1351        audit_log(&f.store).last().expect("a record")["status"].clone()
1352    }
1353
1354    #[test]
1355    fn a_stored_vcon_reads_back_byte_for_byte_and_is_recorded() {
1356        let f = fixture();
1357        let (found, body) = vcon(UUID, "first");
1358        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1359            panic!("a new vCon")
1360        };
1361        let stored = f.store.get(&f.caller, found.uuid).expect("reads");
1362        assert_eq!(
1363            stored,
1364            Stored {
1365                version,
1366                form: vcon::Form::Unsigned,
1367                gzip: false,
1368                body
1369            }
1370        );
1371        let log = audit_log(&f.store);
1372        let read = log.last().expect("a record");
1373        assert_eq!(read["action"], "read");
1374        assert_eq!(read["status"], 200);
1375        assert_eq!(read["version"], version.to_string());
1376        assert_eq!(read["target"], UUID);
1377    }
1378
1379    #[test]
1380    fn a_uuid_never_stored_is_not_found() {
1381        let f = fixture();
1382        let (found, _) = vcon(UUID, "first");
1383        assert_eq!(
1384            f.store.get(&f.caller, found.uuid),
1385            Err(StoreError::NotFound)
1386        );
1387        assert_eq!(get_entry_status(&f), 404);
1388    }
1389
1390    #[test]
1391    fn a_deleted_or_purged_uuid_is_gone() {
1392        for sql in [
1393            "UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
1394            "DELETE FROM vcon WHERE uuid = ?1; INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
1395        ] {
1396            let f = fixture();
1397            let (found, body) = vcon(UUID, "first");
1398            f.store.create(&f.caller, &found, &body).expect("stored");
1399            for statement in sql.split("; ") {
1400                f.store
1401                    .conn()
1402                    .execute(statement, [found.uuid.as_bytes()])
1403                    .expect(statement);
1404            }
1405            assert_eq!(
1406                f.store.get(&f.caller, found.uuid),
1407                Err(StoreError::Gone),
1408                "{sql}"
1409            );
1410            assert_eq!(get_entry_status(&f), 410);
1411        }
1412    }
1413
1414    #[test]
1415    fn a_missing_or_damaged_file_is_a_500_never_an_empty_body() {
1416        let f = fixture();
1417        let (found, body) = vcon(UUID, "first");
1418        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1419            panic!("a new vCon")
1420        };
1421        let file = f.store.objects.join(version.to_string());
1422        let blob = f.fs.read(&file).expect("file");
1423        f.fs.remove(&file).expect("remove");
1424        assert_eq!(
1425            f.store.get(&f.caller, found.uuid),
1426            Err(StoreError::Damaged),
1427            "missing"
1428        );
1429        assert_eq!(get_entry_status(&f), 500);
1430        let mut bad = blob.clone();
1431        let last = bad.len() - 1;
1432        bad[last] ^= 1;
1433        f.fs.write_new(&file, &bad).expect("write");
1434        assert_eq!(
1435            f.store.get(&f.caller, found.uuid),
1436            Err(StoreError::Damaged),
1437            "tampered"
1438        );
1439    }
1440
1441    #[test]
1442    fn a_read_racing_a_delete_never_returns_the_body() {
1443        let f = Arc::new(fixture());
1444        let (found, body) = vcon(UUID, "first");
1445        f.store.create(&f.caller, &found, &body).expect("stored");
1446        let deleter = f.clone();
1447        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
1448            deleter
1449                .store
1450                .conn()
1451                .execute(
1452                    "UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
1453                    [found.uuid.as_bytes()],
1454                )
1455                .expect("delete");
1456        }));
1457        assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
1458        assert_eq!(get_entry_status(&f), 410);
1459    }
1460
1461    #[test]
1462    fn a_read_racing_a_new_version_starts_again_and_returns_the_new_one() {
1463        let f = Arc::new(fixture());
1464        let (found, body) = vcon(UUID, "first");
1465        f.store.create(&f.caller, &found, &body).expect("stored");
1466        let (_, newer) = vcon(UUID, "newer");
1467        let (version, wrapped, _) = f.store.write_version(found.uuid, &newer).expect("written");
1468        let writer = f.clone();
1469        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
1470            let conn = writer.store.conn();
1471            conn.execute(
1472                "INSERT INTO version
1473                     (version, uuid, form, gzip, size, stored_at, tenant_key_id, wrapped_key)
1474                 VALUES (?1, ?2, 'unsigned', 0, 0, 0, ?3, ?4)",
1475                params![
1476                    version.0,
1477                    found.uuid.as_bytes(),
1478                    writer.store.tenant_key.id().bytes(),
1479                    wrapped
1480                ],
1481            )
1482            .expect("new version");
1483            conn.execute(
1484                "UPDATE vcon SET current = ?1 WHERE uuid = ?2",
1485                params![version.0, found.uuid.as_bytes()],
1486            )
1487            .expect("made current");
1488        }));
1489        let stored = f.store.get(&f.caller, found.uuid).expect("reads");
1490        assert_eq!((stored.version, stored.body), (version, newer));
1491        let reads = audit_log(&f.store)
1492            .iter()
1493            .filter(|r| r["action"] == "read")
1494            .count();
1495        assert_eq!(reads, 1, "one record for the one request");
1496    }
1497
1498    #[test]
1499    fn every_form_and_the_gzip_flag_read_back_as_stored() {
1500        use base64::Engine;
1501        use std::io::Write;
1502        let f = fixture();
1503        let uuids = [
1504            "01a07da8-c2bb-83e5-b9a2-279e0d16bc01",
1505            "01a07da8-c2bb-83e5-b9a2-279e0d16bc02",
1506            "01a07da8-c2bb-83e5-b9a2-279e0d16bc03",
1507        ];
1508        let (_, unsigned) = vcon(uuids[0], "plain");
1509        let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
1510        gz.write_all(&unsigned).expect("gzip");
1511        let gzipped = gz.finish().expect("gzip");
1512        let (_, inner) = vcon(uuids[1], "signed");
1513        let signed = format!(
1514            r#"{{"payload": "{}", "signatures": [{{"header": {{}}}}]}}"#,
1515            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(inner)
1516        )
1517        .into_bytes();
1518        let encrypted = format!(
1519            r#"{{"unprotected": {{"uuid": "{}"}}, "ciphertext": "x", "recipients": []}}"#,
1520            uuids[2]
1521        )
1522        .into_bytes();
1523        let cases = [
1524            (gzipped, vcon::Form::Unsigned, true),
1525            (signed, vcon::Form::Signed, false),
1526            (encrypted, vcon::Form::Encrypted, false),
1527        ];
1528        for (body, form, gzip) in cases {
1529            let found = inspect(&body, None, &Limits { inflated: 1 << 20 }, &BTreeSet::new())
1530                .expect("a valid vCon");
1531            assert_eq!((found.form, found.gzip), (form, gzip));
1532            f.store.create(&f.caller, &found, &body).expect("stored");
1533            let stored = f.store.get(&f.caller, found.uuid).expect("reads");
1534            assert_eq!((stored.form, stored.gzip, stored.body), (form, gzip, body));
1535        }
1536    }
1537
1538    /// Stores `subject` under UUID and returns its version.
1539    fn stored(f: &Fixture, subject: &str) -> (Inspection, VersionId) {
1540        let (found, body) = vcon(UUID, subject);
1541        let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
1542            panic!("a new vCon")
1543        };
1544        (found, version)
1545    }
1546
1547    #[test]
1548    fn a_replace_makes_a_new_current_version_and_hides_the_old_one() {
1549        let f = fixture();
1550        let (found, old) = stored(&f, "first");
1551        let (_, newer) = vcon(UUID, "second");
1552        let version = f
1553            .store
1554            .replace(&f.caller, old, &found, &newer)
1555            .expect("replaced");
1556        assert_ne!(version, old);
1557        let stored = f.store.get(&f.caller, found.uuid).expect("reads");
1558        assert_eq!((stored.version, stored.body), (version, newer));
1559        let hidden: Option<i64> = f
1560            .store
1561            .conn()
1562            .query_row(
1563                "SELECT hidden_at FROM version WHERE version = ?1",
1564                [old.0],
1565                |r| r.get(0),
1566            )
1567            .expect("the old row is kept");
1568        assert!(hidden.is_some(), "the old version is hidden");
1569        assert!(
1570            f.fs.read(&f.store.objects.join(old.to_string())).is_ok(),
1571            "its file is kept"
1572        );
1573        let record = audit_log(&f.store)
1574            .into_iter()
1575            .find(|r| r["action"] == "replace")
1576            .expect("a replace record");
1577        assert_eq!(
1578            (record["status"].clone(), record["version"].clone()),
1579            (200.into(), version.to_string().into())
1580        );
1581    }
1582
1583    #[test]
1584    fn a_replace_of_a_missing_or_gone_uuid_is_refused() {
1585        let f = fixture();
1586        let (found, body) = vcon(UUID, "first");
1587        let any = VersionId([7; 16]);
1588        assert_eq!(
1589            f.store.replace(&f.caller, any, &found, &body),
1590            Err(StoreError::NotFound)
1591        );
1592        assert_eq!(get_entry_status(&f), 404);
1593        let (found, old) = stored(&f, "first");
1594        f.store
1595            .conn()
1596            .execute(
1597                "UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
1598                [found.uuid.as_bytes()],
1599            )
1600            .expect("delete");
1601        assert_eq!(
1602            f.store.replace(&f.caller, old, &found, &body),
1603            Err(StoreError::Gone)
1604        );
1605        assert_eq!(get_entry_status(&f), 410);
1606    }
1607
1608    #[test]
1609    fn a_stale_if_match_is_refused_and_writes_nothing() {
1610        let f = fixture();
1611        let (found, _) = stored(&f, "first");
1612        let ops = f.fs.log().len();
1613        let (_, newer) = vcon(UUID, "second");
1614        assert_eq!(
1615            f.store
1616                .replace(&f.caller, VersionId([7; 16]), &found, &newer),
1617            Err(StoreError::PreconditionFailed)
1618        );
1619        assert_eq!(f.fs.log().len(), ops, "no file written");
1620        assert_eq!(wraps(&f.store), 1, "nothing wrapped");
1621        assert_eq!(get_entry_status(&f), 412);
1622    }
1623
1624    #[test]
1625    fn a_signed_or_encrypted_current_version_is_immutable() {
1626        use base64::Engine;
1627        let f = fixture();
1628        let (found, old) = stored(&f, "first");
1629        let (_, inner) = vcon(UUID, "signed");
1630        let signed = format!(
1631            r#"{{"payload": "{}", "signatures": [{{"header": {{}}}}]}}"#,
1632            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(inner)
1633        )
1634        .into_bytes();
1635        let signed_found = inspect(
1636            &signed,
1637            None,
1638            &Limits { inflated: 1 << 20 },
1639            &BTreeSet::new(),
1640        )
1641        .expect("a signed vCon");
1642        let version = f
1643            .store
1644            .replace(&f.caller, old, &signed_found, &signed)
1645            .expect("an unsigned vCon may be replaced by a signed one");
1646        let (_, newer) = vcon(UUID, "second");
1647        let ops = f.fs.log().len();
1648        assert_eq!(
1649            f.store.replace(&f.caller, version, &found, &newer),
1650            Err(StoreError::Immutable)
1651        );
1652        assert_eq!(f.fs.log().len(), ops, "no file written");
1653        assert_eq!(get_entry_status(&f), 409);
1654
1655        let g = fixture();
1656        let encrypted = format!(
1657            r#"{{"unprotected": {{"uuid": "{UUID}"}}, "ciphertext": "x", "recipients": []}}"#
1658        )
1659        .into_bytes();
1660        let enc_found = inspect(
1661            &encrypted,
1662            None,
1663            &Limits { inflated: 1 << 20 },
1664            &BTreeSet::new(),
1665        )
1666        .expect("an encrypted vCon");
1667        let Ok(Created::New(enc_version)) = g.store.create(&g.caller, &enc_found, &encrypted)
1668        else {
1669            panic!("stored")
1670        };
1671        assert_eq!(
1672            g.store.replace(&g.caller, enc_version, &found, &newer),
1673            Err(StoreError::Immutable)
1674        );
1675    }
1676
1677    #[test]
1678    fn a_replace_at_the_wrap_limit_is_refused() {
1679        let f = fixture();
1680        let (found, old) = stored(&f, "first");
1681        f.store
1682            .conn()
1683            .execute("UPDATE key_use SET wraps = ?1", [1_i64 << 32])
1684            .expect("at the limit");
1685        let (_, newer) = vcon(UUID, "second");
1686        assert_eq!(
1687            f.store.replace(&f.caller, old, &found, &newer),
1688            Err(StoreError::LimitReached)
1689        );
1690        assert_eq!(get_entry_status(&f), 503);
1691    }
1692
1693    #[test]
1694    fn of_two_replaces_racing_on_one_version_the_second_to_commit_loses() {
1695        let f = Arc::new(fixture());
1696        let (found, old) = stored(&f, "first");
1697        let winner = f.clone();
1698        let racer = found;
1699        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
1700            let (_, body) = vcon(UUID, "winner");
1701            winner
1702                .store
1703                .replace(&winner.caller, old, &racer, &body)
1704                .expect("wins");
1705        }));
1706        let (_, body) = vcon(UUID, "loser");
1707        assert_eq!(
1708            f.store.replace(&f.caller, old, &found, &body),
1709            Err(StoreError::PreconditionFailed)
1710        );
1711        assert_eq!(
1712            f.fs.list(&f.store.objects).expect("list").len(),
1713            2,
1714            "first and winner only"
1715        );
1716        assert_eq!(wraps(&f.store), 3, "every wrap counted");
1717        assert_eq!(get_entry_status(&f), 412);
1718    }
1719
1720    #[test]
1721    fn two_threads_replacing_with_the_same_if_match_one_wins() {
1722        let f = Arc::new(fixture());
1723        let (_, old) = stored(&f, "first");
1724        let threads: Vec<_> = ["one", "two"]
1725            .into_iter()
1726            .map(|subject| {
1727                let f = f.clone();
1728                std::thread::spawn(move || {
1729                    let (found, body) = vcon(UUID, subject);
1730                    f.store.replace(&f.caller, old, &found, &body)
1731                })
1732            })
1733            .collect();
1734        let results: Vec<_> = threads
1735            .into_iter()
1736            .map(|t| t.join().expect("thread"))
1737            .collect();
1738        assert_eq!(
1739            results.iter().filter(|r| r.is_ok()).count(),
1740            1,
1741            "{results:?}"
1742        );
1743        assert!(
1744            results.contains(&Err(StoreError::PreconditionFailed)),
1745            "{results:?}"
1746        );
1747    }
1748
1749    #[test]
1750    fn a_replace_whose_audit_record_fails_changes_nothing() {
1751        let f = fixture();
1752        let (found, old) = stored(&f, "first");
1753        f.store
1754            .conn()
1755            .execute_batch(
1756                "CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
1757                 BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
1758            )
1759            .expect("trigger");
1760        let (_, newer) = vcon(UUID, "second");
1761        assert_eq!(
1762            f.store.replace(&f.caller, old, &found, &newer),
1763            Err(StoreError::AuditFailed)
1764        );
1765        f.store
1766            .conn()
1767            .execute_batch("DROP TRIGGER no_audit")
1768            .expect("drop trigger");
1769        assert_eq!(
1770            f.store.get(&f.caller, found.uuid).expect("reads").version,
1771            old
1772        );
1773        assert_eq!(
1774            f.fs.list(&f.store.objects).expect("list").len(),
1775            1,
1776            "only the first file"
1777        );
1778    }
1779
1780    fn hidden_versions(f: &Fixture) -> i64 {
1781        count(
1782            &f.store,
1783            "SELECT count(*) FROM version WHERE hidden_at IS NOT NULL",
1784        )
1785    }
1786
1787    #[test]
1788    fn a_deleted_uuid_answers_gone_and_keeps_its_versions_hidden() {
1789        let f = fixture();
1790        let (found, old) = stored(&f, "first");
1791        let (_, newer) = vcon(UUID, "second");
1792        f.clock.set(datetime!(2026-10-23 13:00 UTC));
1793        let current = f
1794            .store
1795            .replace(&f.caller, old, &found, &newer)
1796            .expect("replaced");
1797        f.clock.set(datetime!(2026-10-23 14:00 UTC));
1798        assert_eq!(
1799            f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
1800            Ok(())
1801        );
1802        let record = audit_log(&f.store).pop().expect("a record");
1803        assert_eq!(
1804            (record["action"].clone(), record["status"].clone()),
1805            ("delete".into(), 204.into())
1806        );
1807        let hidden_at = |version: VersionId| -> i64 {
1808            f.store
1809                .conn()
1810                .query_row(
1811                    "SELECT hidden_at FROM version WHERE version = ?1",
1812                    [version.0],
1813                    |r| r.get(0),
1814                )
1815                .expect("hidden")
1816        };
1817        assert_eq!(
1818            hidden_at(old),
1819            nanos(datetime!(2026-10-23 13:00 UTC)),
1820            "hidden by the replace"
1821        );
1822        assert_eq!(
1823            hidden_at(current),
1824            nanos(datetime!(2026-10-23 14:00 UTC)),
1825            "hidden by the delete"
1826        );
1827        assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
1828        let (_, body) = vcon(UUID, "third");
1829        assert_eq!(
1830            f.store.create(&f.caller, &found, &body),
1831            Err(StoreError::Gone)
1832        );
1833        assert_eq!(hidden_versions(&f), 2, "both versions hidden");
1834        assert_eq!(
1835            f.fs.list(&f.store.objects).expect("list").len(),
1836            2,
1837            "files kept for legal reads"
1838        );
1839        let (current, deleted_at): (Option<Vec<u8>>, Option<i64>) = f
1840            .store
1841            .conn()
1842            .query_row(
1843                "SELECT current, deleted_at FROM vcon WHERE uuid = ?1",
1844                [found.uuid.as_bytes()],
1845                |r| Ok((r.get(0)?, r.get(1)?)),
1846            )
1847            .expect("row");
1848        assert_eq!(current, None);
1849        assert_eq!(deleted_at, Some(nanos(datetime!(2026-10-23 14:00 UTC))));
1850    }
1851
1852    #[test]
1853    fn deleting_a_deleted_or_purged_uuid_succeeds_again() {
1854        let f = fixture();
1855        let (found, _) = stored(&f, "first");
1856        f.store
1857            .delete(&f.caller, found.uuid, Retention::Years(7))
1858            .expect("deleted");
1859        assert_eq!(
1860            f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
1861            Ok(())
1862        );
1863        assert_eq!(get_entry_status(&f), 204);
1864        let g = fixture();
1865        g.store
1866            .conn()
1867            .execute(
1868                "INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
1869                [found.uuid.as_bytes()],
1870            )
1871            .expect("purged");
1872        assert_eq!(
1873            g.store.delete(&g.caller, found.uuid, Retention::Years(7)),
1874            Ok(())
1875        );
1876        assert_eq!(get_entry_status(&g), 204);
1877    }
1878
1879    #[test]
1880    fn deleting_a_uuid_never_stored_is_not_found() {
1881        let f = fixture();
1882        let (found, _) = vcon(UUID, "first");
1883        assert_eq!(
1884            f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
1885            Err(StoreError::NotFound)
1886        );
1887        assert_eq!(get_entry_status(&f), 404);
1888    }
1889
1890    #[test]
1891    fn a_hold_does_not_stop_a_delete() {
1892        let f = fixture();
1893        let (found, _) = stored(&f, "first");
1894        f.store
1895            .conn()
1896            .execute(
1897                "UPDATE vcon SET hold = 1 WHERE uuid = ?1",
1898                [found.uuid.as_bytes()],
1899            )
1900            .expect("hold");
1901        assert_eq!(
1902            f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
1903            Ok(())
1904        );
1905        assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
1906    }
1907
1908    #[test]
1909    fn a_delete_whose_audit_record_fails_changes_nothing() {
1910        let f = fixture();
1911        let (found, version) = stored(&f, "first");
1912        f.store
1913            .conn()
1914            .execute_batch(
1915                "CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
1916                 BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
1917            )
1918            .expect("trigger");
1919        assert_eq!(
1920            f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
1921            Err(StoreError::AuditFailed)
1922        );
1923        f.store
1924            .conn()
1925            .execute_batch("DROP TRIGGER no_audit")
1926            .expect("drop trigger");
1927        assert_eq!(
1928            f.store
1929                .get(&f.caller, found.uuid)
1930                .expect("still live")
1931                .version,
1932            version
1933        );
1934        assert_eq!(hidden_versions(&f), 0);
1935    }
1936
1937    #[test]
1938    fn only_the_optional_catalog_columns_may_be_empty() {
1939        let f = fixture();
1940        let conn = f.store.conn();
1941        let tables: [(&str, &[(&str, &str)]); 5] = [
1942            (
1943                "vcon",
1944                &[
1945                    ("uuid", "x'01'"),
1946                    ("current", "x'02'"),
1947                    ("deleted_at", "1"),
1948                    ("hold", "0"),
1949                ],
1950            ),
1951            (
1952                "version",
1953                &[
1954                    ("version", "x'02'"),
1955                    ("uuid", "x'01'"),
1956                    ("form", "'unsigned'"),
1957                    ("gzip", "0"),
1958                    ("size", "1"),
1959                    ("stored_at", "1"),
1960                    ("hidden_at", "1"),
1961                    ("tenant_key_id", "x'03'"),
1962                    ("wrapped_key", "x'04'"),
1963                ],
1964            ),
1965            ("tombstone", &[("uuid", "x'01'"), ("purged_at", "1")]),
1966            ("key_use", &[("tenant_key_id", "x'03'"), ("wraps", "1")]),
1967            (
1968                "audit",
1969                &[
1970                    ("at", "'2026-10-23T12:00Z'"),
1971                    ("actor", "'cli'"),
1972                    ("action", "'create'"),
1973                    ("target", "'x'"),
1974                    ("version", "'y'"),
1975                    ("status", "201"),
1976                    ("count", "1"),
1977                ],
1978            ),
1979        ];
1980        let may_be_empty: Vec<String> = tables
1981            .iter()
1982            .flat_map(|(table, row)| columns_taking_null(&conn, table, row))
1983            .collect();
1984        assert_eq!(
1985            may_be_empty,
1986            [
1987                "vcon.current",
1988                "vcon.deleted_at",
1989                "version.hidden_at",
1990                "audit.target",
1991                "audit.version"
1992            ]
1993        );
1994    }
1995
1996    #[test]
1997    fn a_new_stores_objects_directory_survives_a_crash() {
1998        let f = fixture();
1999        f.fs.crash();
2000        assert_eq!(
2001            f.fs.list(f.dir.path()).expect("list"),
2002            [std::ffi::OsString::from("objects")]
2003        );
2004    }
2005
2006    #[test]
2007    fn opening_removes_temporary_and_orphaned_files_and_nothing_else() {
2008        let mut f = fixture();
2009        let (found, version) = stored(&f, "first");
2010        let orphan = VersionId([0x0e; 16]);
2011        let objects = f.store.objects.clone();
2012        for name in [
2013            format!("{orphan}"),
2014            format!("{orphan}.tmp"),
2015            format!("{version}.tmp"),
2016            "notes.txt".to_owned(),
2017        ] {
2018            f.fs.write_new(&objects.join(name), b"x").expect("write");
2019        }
2020        f.reopen();
2021        let names: Vec<String> =
2022            f.fs.list(&objects)
2023                .expect("list")
2024                .into_iter()
2025                .map(|n| n.to_string_lossy().into_owned())
2026                .collect();
2027        assert_eq!(names, [version.to_string(), "notes.txt".to_owned()]);
2028        assert_eq!(
2029            f.store
2030                .get(&f.caller, found.uuid)
2031                .expect("still reads")
2032                .version,
2033            version
2034        );
2035        let log = f.fs.log();
2036        let last_removal = log
2037            .iter()
2038            .rposition(|op| matches!(op, Op::Remove(_)))
2039            .expect("files were removed");
2040        assert!(
2041            log[last_removal..].contains(&Op::SyncDir(objects)),
2042            "the removals are synced"
2043        );
2044    }
2045
2046    /// A command beside a running server must not take the server's
2047    /// writes in progress for a crash's leftovers.
2048    #[test]
2049    fn opening_beside_a_server_removes_nothing() {
2050        let f = fixture();
2051        let (found, version) = stored(&f, "first");
2052        let orphan = VersionId([0x0e; 16]);
2053        let objects = f.store.objects.clone();
2054        let mut written = vec![version.to_string()];
2055        for name in [
2056            format!("{orphan}"),
2057            format!("{orphan}.tmp"),
2058            format!("{version}.tmp"),
2059        ] {
2060            f.fs.write_new(&objects.join(&name), b"x").expect("write");
2061            written.push(name);
2062        }
2063        written.sort();
2064        let ops = f.fs.log().len();
2065        let beside = f.open_beside().expect("opens beside");
2066        let mut names: Vec<String> =
2067            f.fs.list(&objects)
2068                .expect("list")
2069                .into_iter()
2070                .map(|n| n.to_string_lossy().into_owned())
2071                .collect();
2072        names.sort();
2073        assert_eq!(names, written);
2074        assert!(
2075            !f.fs.log()[ops..]
2076                .iter()
2077                .any(|op| matches!(op, Op::Remove(_))),
2078            "nothing removed"
2079        );
2080        assert_eq!(
2081            beside.get(&f.caller, found.uuid).expect("reads").version,
2082            version
2083        );
2084    }
2085
2086    #[test]
2087    fn opening_checkpoints_the_wal_with_truncation() {
2088        let mut f = fixture();
2089        stored(&f, "first");
2090        let wal = f.dir.path().join("catalog.sqlite-wal");
2091        assert!(
2092            std::fs::metadata(&wal).expect("a WAL").len() > 0,
2093            "the WAL holds the write"
2094        );
2095        f.reopen();
2096        assert_eq!(std::fs::metadata(&wal).expect("a WAL").len(), 0);
2097    }
2098
2099    #[test]
2100    fn check_reports_a_version_whose_file_is_missing() {
2101        let f = fixture();
2102        let (_, version) = stored(&f, "first");
2103        assert_eq!(f.store.check(), Ok(vec![]));
2104        f.fs.remove(&f.store.objects.join(version.to_string()))
2105            .expect("remove");
2106        assert_eq!(f.store.check(), Ok(vec![version]));
2107    }
2108
2109    #[test]
2110    fn opening_refuses_while_a_reader_stops_the_wal_from_being_truncated() {
2111        let f = fixture();
2112        stored(&f, "first");
2113        let ctx = Context::Tenant { tenant: TENANT };
2114        let tenant_key = envelope::unwrap(&f.master, &ctx, &f.wrapped).expect("unwraps");
2115        let catalog_key = envelope::catalog_key(&tenant_key).expect("catalog key");
2116        let mut reader = catalog::open(&f.dir.path().join("catalog.sqlite"), &catalog_key, "")
2117            .expect("a second connection");
2118        let read = reader.transaction().expect("read transaction");
2119        let _: i64 = read
2120            .query_row("SELECT count(*) FROM vcon", [], |r| r.get(0))
2121            .expect("a read holds a snapshot");
2122        assert!(matches!(f.open(), Err(StoreError::Io)));
2123        drop(read);
2124        assert!(f.open().is_ok(), "opens once the reader is done");
2125    }
2126
2127    /// The filesystem changes a create or replace makes when nothing
2128    /// fails: write, sync, rename, sync the directory.
2129    const WRITE_STEPS: usize = 4;
2130
2131    /// After a restart, no row lacks its file and every file has a row.
2132    fn assert_whole(f: &Fixture, step: usize) {
2133        assert_eq!(
2134            f.store.check(),
2135            Ok(vec![]),
2136            "a row without its file at step {step}"
2137        );
2138        for name in f.fs.list(&f.store.objects).expect("list") {
2139            let name = name.to_string_lossy().into_owned();
2140            let version = VersionId::parse(&name)
2141                .unwrap_or_else(|| panic!("{name} left over at step {step}"));
2142            let rows = count(
2143                &f.store,
2144                &format!("SELECT count(*) FROM version WHERE hex(version) = upper('{version}')"),
2145            );
2146            assert_eq!(rows, 1, "{name} has no row at step {step}");
2147        }
2148    }
2149
2150    /// Makes the next audit insert fail for every connection, as if the
2151    /// process died before its commit; `unblock` removes it again.
2152    pub(super) fn block_commits(f: &Fixture) -> Connection {
2153        let ctx = Context::Tenant { tenant: TENANT };
2154        let tenant_key = envelope::unwrap(&f.master, &ctx, &f.wrapped).expect("unwraps");
2155        let catalog_key = envelope::catalog_key(&tenant_key).expect("catalog key");
2156        catalog::open(&f.dir.path().join("catalog.sqlite"), &catalog_key, "")
2157            .expect("second connection")
2158    }
2159
2160    fn die_before_commit(f: &Fixture) {
2161        let other = block_commits(f);
2162        let fs = f.fs.clone();
2163        *f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
2164            fs.fail_from(0);
2165            other
2166                .execute_batch(
2167                    "CREATE TRIGGER dead BEFORE INSERT ON audit
2168                     BEGIN SELECT RAISE(ABORT, 'the process has died'); END;",
2169                )
2170                .expect("trigger");
2171        }));
2172    }
2173
2174    pub(super) fn restart(f: &mut Fixture) {
2175        f.fs.crash();
2176        f.store
2177            .conn()
2178            .execute_batch("DROP TRIGGER IF EXISTS dead")
2179            .expect("drop trigger");
2180        f.reopen();
2181    }
2182
2183    #[test]
2184    fn a_create_cut_off_at_any_step_leaves_no_vcon_or_the_whole_vcon() {
2185        let (found, body) = vcon(UUID, "first");
2186        for step in 0..=WRITE_STEPS {
2187            let mut f = fixture();
2188            f.fs.fail_from(f.fs.log().len() + step);
2189            let result = f.store.create(&f.caller, &found, &body);
2190            assert_eq!(
2191                result.is_ok(),
2192                step == WRITE_STEPS,
2193                "step {step}: {result:?}"
2194            );
2195            restart(&mut f);
2196            match f.store.get(&f.caller, found.uuid) {
2197                Ok(stored) => assert_eq!(stored.body, body, "step {step}"),
2198                Err(StoreError::NotFound) => {
2199                    assert!(result.is_err(), "acknowledged, then lost, at step {step}")
2200                }
2201                other => panic!("half a vCon at step {step}: {other:?}"),
2202            }
2203            assert_whole(&f, step);
2204        }
2205    }
2206
2207    #[test]
2208    fn a_create_cut_off_between_its_file_and_its_commit_leaves_nothing() {
2209        let mut f = fixture();
2210        let (found, body) = vcon(UUID, "first");
2211        die_before_commit(&f);
2212        assert!(f.store.create(&f.caller, &found, &body).is_err());
2213        restart(&mut f);
2214        assert_eq!(
2215            f.store.get(&f.caller, found.uuid),
2216            Err(StoreError::NotFound)
2217        );
2218        assert_whole(&f, WRITE_STEPS);
2219    }
2220
2221    #[test]
2222    fn a_replace_cut_off_at_any_step_leaves_the_old_version_or_the_new() {
2223        let (found, old_body) = vcon(UUID, "first");
2224        let (_, new_body) = vcon(UUID, "second");
2225        for step in 0..=WRITE_STEPS {
2226            let mut f = fixture();
2227            let Ok(Created::New(old)) = f.store.create(&f.caller, &found, &old_body) else {
2228                panic!("stored")
2229            };
2230            f.fs.fail_from(f.fs.log().len() + step);
2231            let result = f.store.replace(&f.caller, old, &found, &new_body);
2232            assert_eq!(
2233                result.is_ok(),
2234                step == WRITE_STEPS,
2235                "step {step}: {result:?}"
2236            );
2237            restart(&mut f);
2238            let stored = f
2239                .store
2240                .get(&f.caller, found.uuid)
2241                .unwrap_or_else(|e| panic!("step {step}: {e:?}"));
2242            match result {
2243                Ok(new) => assert_eq!(
2244                    (stored.version, stored.body),
2245                    (new, new_body.clone()),
2246                    "step {step}"
2247                ),
2248                Err(_) => assert_eq!(
2249                    (stored.version, stored.body),
2250                    (old, old_body.clone()),
2251                    "step {step}"
2252                ),
2253            }
2254            assert_whole(&f, step);
2255        }
2256    }
2257
2258    #[test]
2259    fn a_replace_cut_off_between_its_file_and_its_commit_keeps_the_old_version() {
2260        let mut f = fixture();
2261        let (found, old) = stored(&f, "first");
2262        let (_, newer) = vcon(UUID, "second");
2263        die_before_commit(&f);
2264        assert!(f.store.replace(&f.caller, old, &found, &newer).is_err());
2265        restart(&mut f);
2266        assert_eq!(
2267            f.store.get(&f.caller, found.uuid).expect("reads").version,
2268            old
2269        );
2270        assert_eq!(hidden_versions(&f), 0);
2271        assert_whole(&f, WRITE_STEPS);
2272    }
2273
2274    #[test]
2275    fn a_delete_cut_off_before_its_commit_leaves_the_vcon_live() {
2276        let mut f = fixture();
2277        let (found, version) = stored(&f, "first");
2278        let other = block_commits(&f);
2279        other
2280            .execute_batch(
2281                "CREATE TRIGGER dead BEFORE INSERT ON audit
2282                 BEGIN SELECT RAISE(ABORT, 'the process has died'); END;",
2283            )
2284            .expect("trigger");
2285        assert!(
2286            f.store
2287                .delete(&f.caller, found.uuid, Retention::Years(7))
2288                .is_err()
2289        );
2290        drop(other);
2291        restart(&mut f);
2292        assert_eq!(
2293            f.store
2294                .get(&f.caller, found.uuid)
2295                .expect("still live")
2296                .version,
2297            version
2298        );
2299        assert_eq!(hidden_versions(&f), 0);
2300    }
2301}