Skip to main content

vcon_server/
store.rs

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