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