use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use uuid::Uuid;
use vcon::Inspection;
use crate::audit::{self, Action, Actor, Entry};
use crate::catalog;
use crate::clock::{Clock, nanos};
use crate::envelope::{self, Context, Key, WrapLimit};
use crate::fs::Fs;
use crate::ids::{KeyId, TenantId, VersionId};
use crate::retention::Retention;
mod examine;
mod legal;
mod list;
mod purge;
mod rekey;
pub use examine::Examination;
pub use legal::{Legal, VersionInfo};
pub use list::{Cursor, ListQuery, Listed, Listing};
pub use purge::{Kept, PurgeReport, due};
const TABLES: &str = "
CREATE TABLE vcon (
uuid BLOB PRIMARY KEY NOT NULL,
current BLOB,
deleted_at INTEGER,
hold INTEGER NOT NULL DEFAULT 0,
stored_at INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX vcon_listed ON vcon (stored_at, uuid);
CREATE TABLE version (
version BLOB PRIMARY KEY NOT NULL,
uuid BLOB NOT NULL,
form TEXT NOT NULL,
gzip INTEGER NOT NULL,
size INTEGER NOT NULL,
stored_at INTEGER NOT NULL,
hidden_at INTEGER,
tenant_key_id BLOB NOT NULL,
wrapped_key BLOB NOT NULL
);
CREATE INDEX version_uuid ON version (uuid);
CREATE TABLE tombstone (
uuid BLOB PRIMARY KEY NOT NULL,
purged_at INTEGER NOT NULL
);
CREATE TABLE key_use (
tenant_key_id BLOB PRIMARY KEY NOT NULL,
wraps INTEGER NOT NULL
);
";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Caller {
pub key: KeyId,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Created {
New(VersionId),
Same(VersionId),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Stored {
pub version: VersionId,
pub form: vcon::Form,
pub gzip: bool,
pub body: Vec<u8>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Counts {
pub versions: u64,
pub held: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreError {
Conflict,
Gone,
NotFound,
NotDeleted,
PreconditionFailed,
Immutable,
LimitReached,
Damaged,
AuditFailed,
Io,
}
pub struct Store {
tenant: TenantId,
tenant_key: Key,
objects: PathBuf,
conn: Mutex<Connection>,
fs: Arc<dyn Fs>,
clock: Arc<dyn Clock>,
#[cfg(test)]
hooks: tests::Hooks,
}
impl Store {
pub fn open(
dir: &Path,
tenant: TenantId,
tenant_key: Key,
fs: Arc<dyn Fs>,
clock: Arc<dyn Clock>,
) -> Result<Store, StoreError> {
let store = Store::open_beside(dir, tenant, tenant_key, fs, clock)?;
store.recover()?;
Ok(store)
}
pub fn open_beside(
dir: &Path,
tenant: TenantId,
tenant_key: Key,
fs: Arc<dyn Fs>,
clock: Arc<dyn Clock>,
) -> Result<Store, StoreError> {
let catalog_key = envelope::catalog_key(&tenant_key).map_err(|_| StoreError::Io)?;
let schema = format!("{TABLES}{}", audit::SCHEMA);
let conn = catalog::open(&dir.join("catalog.sqlite"), &catalog_key, &schema)
.map_err(|_| StoreError::Io)?;
let objects = dir.join("objects");
fs.create_dir_all(&objects).map_err(|_| StoreError::Io)?;
fs.sync_dir(dir).map_err(|_| StoreError::Io)?;
Ok(Store {
tenant,
tenant_key,
objects,
conn: Mutex::new(conn),
fs,
clock,
#[cfg(test)]
hooks: tests::Hooks::default(),
})
}
fn recover(&self) -> Result<(), StoreError> {
let conn = self.conn();
let busy: i64 = conn
.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| r.get(0))
.map_err(io)?;
if busy != 0 {
return Err(StoreError::Io);
}
let mut removed = false;
for name in self.fs.list(&self.objects).map_err(io)? {
let Some(name) = name.to_str() else { continue };
let (id, temporary) = match name.strip_suffix(".tmp") {
Some(id) => (id, true),
None => (name, false),
};
let Some(version) = VersionId::parse(id) else {
continue;
};
let has_row = conn
.query_row(
"SELECT 1 FROM version WHERE version = ?1",
[version.0],
|_| Ok(()),
)
.optional()
.map_err(io)?
.is_some();
if temporary || !has_row {
self.fs.remove(&self.objects.join(name)).map_err(io)?;
removed = true;
}
}
if removed {
self.fs.sync_dir(&self.objects).map_err(io)?;
}
Ok(())
}
pub fn create(
&self,
who: &Caller,
found: &Inspection,
body: &[u8],
) -> Result<Created, StoreError> {
let uuid = found.uuid;
let entry = |status, version| Entry {
actor: Actor::Key(who.key),
action: Action::Create,
target: Some(uuid.to_string()),
version,
status,
};
let key_id = self.tenant_key.id();
{
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
let refused = match state(&tx, uuid)? {
State::Gone => Some((410, None, StoreError::Gone)),
State::Live(current) => match self.read_version(&tx, uuid, current) {
Ok(stored) if stored == body => {
self.record(&tx, &entry(200, Some(current)))?;
commit(tx)?;
return Ok(Created::Same(current));
}
Ok(_) => Some((409, None, StoreError::Conflict)),
Err(StoreError::Damaged) => Some((500, Some(current), StoreError::Damaged)),
Err(e) => return Err(e),
},
State::Absent if envelope::wrap_limit(wraps(&tx, key_id)?) == WrapLimit::Refuse => {
Some((503, None, StoreError::LimitReached))
}
State::Absent => {
add_wrap(&tx, key_id)?;
None
}
};
if let Some((status, version, error)) = refused {
self.record(&tx, &entry(status, version))?;
commit(tx)?;
return Err(error);
}
commit(tx)?;
}
let (version, wrapped, file) = self.write_version(uuid, body)?;
self.pause();
let committed = (|| {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
match state(&tx, uuid)? {
State::Absent => {}
lost => {
let (status, error) = match lost {
State::Gone => (410, StoreError::Gone),
_ => (409, StoreError::Conflict),
};
self.record(&tx, &entry(status, None))?;
commit(tx)?;
return Err(error);
}
}
let stored_at = nanos(self.clock.now());
tx.execute(
"INSERT INTO version
(version, uuid, form, gzip, size, stored_at, hidden_at, tenant_key_id, wrapped_key)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8)",
params![
version.0,
uuid.as_bytes(),
form_name(found.form),
found.gzip,
i64::try_from(body.len()).map_err(io)?,
stored_at,
key_id.bytes(),
wrapped,
],
)
.map_err(io)?;
tx.execute(
"INSERT INTO vcon (uuid, current, stored_at) VALUES (?1, ?2, ?3)",
params![uuid.as_bytes(), version.0, stored_at],
)
.map_err(io)?;
self.record(&tx, &entry(201, Some(version)))?;
commit(tx)?;
Ok(Created::New(version))
})();
if committed.is_err() {
let _ = self.fs.remove(&file);
}
committed
}
pub fn get(&self, who: &Caller, uuid: Uuid) -> Result<Stored, StoreError> {
let entry = |status, version| Entry {
actor: Actor::Key(who.key),
action: Action::Read,
target: Some(uuid.to_string()),
version,
status,
};
let refuse = |tx: Transaction, state: State| {
let (status, error) = match state {
State::Gone => (410, StoreError::Gone),
_ => (404, StoreError::NotFound),
};
self.record(&tx, &entry(status, None))?;
commit(tx)?;
Err(error)
};
loop {
let (version, row) = {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
match state(&tx, uuid)? {
State::Live(version) => (version, version_row(&tx, version)),
other => return refuse(tx, other),
}
};
let opened = row.and_then(|row| {
let body = self.open_version(uuid, version, &row.wrapped)?;
Ok(Stored {
version,
form: row.form,
gzip: row.gzip,
body,
})
});
self.pause();
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
match state(&tx, uuid)? {
State::Live(now) if now == version => {
let status = if opened.is_ok() { 200 } else { 500 };
self.record(&tx, &entry(status, Some(version)))?;
commit(tx)?;
return opened;
}
State::Live(_) => continue,
other => return refuse(tx, other),
}
}
}
pub fn replace(
&self,
who: &Caller,
if_match: VersionId,
found: &Inspection,
body: &[u8],
) -> Result<VersionId, StoreError> {
let uuid = found.uuid;
let entry = |status, version| Entry {
actor: Actor::Key(who.key),
action: Action::Replace,
target: Some(uuid.to_string()),
version,
status,
};
let key_id = self.tenant_key.id();
let refusal = |tx: &Transaction| -> Result<Option<(u16, StoreError)>, StoreError> {
Ok(match state(tx, uuid)? {
State::Absent => Some((404, StoreError::NotFound)),
State::Gone => Some((410, StoreError::Gone)),
State::Live(current) if version_row(tx, current)?.form != vcon::Form::Unsigned => {
Some((409, StoreError::Immutable))
}
State::Live(current) if current != if_match => {
Some((412, StoreError::PreconditionFailed))
}
State::Live(_) => None,
})
};
{
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
let refused = match refusal(&tx)? {
None if envelope::wrap_limit(wraps(&tx, key_id)?) == WrapLimit::Refuse => {
Some((503, StoreError::LimitReached))
}
other => other,
};
if let Some((status, error)) = refused {
self.record(&tx, &entry(status, None))?;
commit(tx)?;
return Err(error);
}
add_wrap(&tx, key_id)?;
commit(tx)?;
}
let (version, wrapped, file) = self.write_version(uuid, body)?;
self.pause();
let committed = (|| {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
if let Some((status, error)) = refusal(&tx)? {
self.record(&tx, &entry(status, None))?;
commit(tx)?;
return Err(error);
}
let now = nanos(self.clock.now());
tx.execute(
"INSERT INTO version
(version, uuid, form, gzip, size, stored_at, hidden_at, tenant_key_id, wrapped_key)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8)",
params![
version.0,
uuid.as_bytes(),
form_name(found.form),
found.gzip,
i64::try_from(body.len()).map_err(io)?,
now,
key_id.bytes(),
wrapped,
],
)
.map_err(io)?;
tx.execute(
"UPDATE version SET hidden_at = ?1 WHERE version = ?2",
params![now, if_match.0],
)
.map_err(io)?;
tx.execute(
"UPDATE vcon SET current = ?1 WHERE uuid = ?2",
params![version.0, uuid.as_bytes()],
)
.map_err(io)?;
self.record(&tx, &entry(200, Some(version)))?;
commit(tx)?;
Ok(version)
})();
if committed.is_err() {
let _ = self.fs.remove(&file);
}
committed
}
pub fn delete(&self, who: &Caller, uuid: Uuid, retention: Retention) -> Result<(), StoreError> {
let entry = |status| Entry {
actor: Actor::Key(who.key),
action: Action::Delete,
target: Some(uuid.to_string()),
version: None,
status,
};
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
let outcome = match state(&tx, uuid)? {
State::Absent => Err(StoreError::NotFound),
State::Gone => Ok(()),
State::Live(_) => {
let now = nanos(self.clock.now());
tx.execute(
"UPDATE vcon SET current = NULL, deleted_at = ?1 WHERE uuid = ?2",
params![now, uuid.as_bytes()],
)
.map_err(io)?;
tx.execute(
"UPDATE version SET hidden_at = ?1 WHERE uuid = ?2 AND hidden_at IS NULL",
params![now, uuid.as_bytes()],
)
.map_err(io)?;
Ok(())
}
};
let status = if outcome.is_ok() { 204 } else { 404 };
self.record(&tx, &entry(status))?;
commit(tx)?;
outcome?;
drop(conn);
self.purge_hidden(Actor::Key(who.key), retention, Some(uuid))?;
Ok(())
}
pub fn check(&self) -> Result<Vec<VersionId>, StoreError> {
let conn = self.conn();
let mut rows = conn
.prepare("SELECT version FROM version ORDER BY version")
.map_err(io)?;
let versions = rows
.query_map([], |r| r.get::<_, [u8; 16]>(0))
.map_err(io)?
.collect::<Result<Vec<_>, _>>()
.map_err(io)?;
let present: std::collections::BTreeSet<_> = self
.fs
.list(&self.objects)
.map_err(io)?
.into_iter()
.collect();
Ok(versions
.into_iter()
.map(VersionId)
.filter(|v| !present.contains(std::ffi::OsStr::new(&v.to_string())))
.collect())
}
fn write_version(
&self,
uuid: Uuid,
body: &[u8],
) -> Result<(VersionId, Vec<u8>, PathBuf), StoreError> {
let version = VersionId::generate();
let ctx = Context::Version {
tenant: self.tenant,
uuid,
version,
};
let data_key = Key::generate().map_err(|_| StoreError::Io)?;
let blob = envelope::seal(&data_key, &ctx, body);
let wrapped = envelope::wrap(&self.tenant_key, &ctx, &data_key);
let file = self.write_object(version, &blob)?;
Ok((version, wrapped, file))
}
fn pause(&self) {
#[cfg(test)]
{
let hook = self.hooks.between.lock().expect("hook").take();
if let Some(hook) = hook {
hook();
}
}
}
fn write_object(&self, version: VersionId, blob: &[u8]) -> Result<PathBuf, StoreError> {
let tmp = self.objects.join(format!("{version}.tmp"));
let file = self.objects.join(version.to_string());
let written = self
.fs
.write_new(&tmp, blob)
.and_then(|()| self.fs.sync_file(&tmp))
.and_then(|()| self.fs.rename(&tmp, &file))
.and_then(|()| self.fs.sync_dir(&self.objects));
if written.is_err() {
let _ = self.fs.remove(&tmp);
let _ = self.fs.remove(&file);
return Err(StoreError::Io);
}
Ok(file)
}
fn read_version(
&self,
tx: &Transaction,
uuid: Uuid,
version: VersionId,
) -> Result<Vec<u8>, StoreError> {
let row = version_row(tx, version)?;
self.open_version(uuid, version, &row.wrapped)
}
fn open_version(
&self,
uuid: Uuid,
version: VersionId,
wrapped: &[u8],
) -> Result<Vec<u8>, StoreError> {
let ctx = Context::Version {
tenant: self.tenant,
uuid,
version,
};
let data_key =
envelope::unwrap(&self.tenant_key, &ctx, wrapped).map_err(|_| StoreError::Damaged)?;
let blob = self
.fs
.read(&self.objects.join(version.to_string()))
.map_err(|_| StoreError::Damaged)?;
envelope::open(&data_key, &ctx, &blob).map_err(|_| StoreError::Damaged)
}
pub fn counts(&self) -> Result<Counts, StoreError> {
let conn = self.conn();
let count = |sql| -> Result<u64, StoreError> {
let n: i64 = conn.query_row(sql, [], |r| r.get(0)).map_err(io)?;
u64::try_from(n).map_err(io)
};
Ok(Counts {
versions: count("SELECT count(*) FROM version")?,
held: count("SELECT count(*) FROM vcon WHERE hold != 0")?,
})
}
pub fn export(
&self,
since: Option<i64>,
out: &mut dyn std::io::Write,
) -> Result<(), audit::AuditError> {
audit::export(&self.conn(), since, out)
}
pub fn audit_page(
&self,
who: &Caller,
query: &audit::AuditQuery,
) -> Result<Vec<u8>, StoreError> {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
let mut out = Vec::new();
let read = audit::page(&tx, query, &mut out);
self.record(
&tx,
&Entry {
actor: Actor::Key(who.key),
action: Action::AuditRead,
target: None,
version: None,
status: if read.is_ok() { 200 } else { 500 },
},
)?;
commit(tx)?;
read.map(|()| out).map_err(io)
}
pub fn log(&self, entry: &Entry) -> Result<(), StoreError> {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
self.record(&tx, entry)?;
commit(tx)
}
#[cfg(test)]
pub(crate) fn break_audit(&self) {
self.conn()
.execute_batch(
"CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'no audit'); END;",
)
.expect("trigger");
}
#[cfg(test)]
pub(crate) fn conn_for_test(&self) -> MutexGuard<'_, Connection> {
self.conn()
}
#[cfg(test)]
pub(crate) fn records(&self) -> Vec<serde_json::Value> {
crate::audit::testing::exported(&self.conn(), None)
}
pub fn count(&self, entry: &Entry) -> Result<(), StoreError> {
let mut conn = self.conn();
let tx = immediate(&mut conn)?;
audit::count(&tx, entry, self.clock.now()).map_err(|_| StoreError::AuditFailed)?;
commit(tx)
}
fn record(&self, tx: &Transaction, entry: &Entry) -> Result<(), StoreError> {
audit::record(tx, entry, self.clock.now()).map_err(|_| StoreError::AuditFailed)
}
fn conn(&self) -> MutexGuard<'_, Connection> {
self.conn
.lock()
.expect("no thread panicked holding the catalog")
}
}
enum State {
Absent,
Live(VersionId),
Gone,
}
fn state(tx: &Transaction, uuid: Uuid) -> Result<State, StoreError> {
let purged = tx
.query_row(
"SELECT 1 FROM tombstone WHERE uuid = ?1",
[uuid.as_bytes()],
|_| Ok(()),
)
.optional()
.map_err(io)?;
if purged.is_some() {
return Ok(State::Gone);
}
let row: Option<(Option<[u8; 16]>, Option<i64>)> = tx
.query_row(
"SELECT current, deleted_at FROM vcon WHERE uuid = ?1",
[uuid.as_bytes()],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.optional()
.map_err(io)?;
match row {
None => Ok(State::Absent),
Some((_, Some(_))) => Ok(State::Gone),
Some((Some(current), None)) => Ok(State::Live(VersionId(current))),
Some((None, None)) => Err(StoreError::Damaged),
}
}
struct VersionRow {
form: vcon::Form,
gzip: bool,
wrapped: Vec<u8>,
}
fn version_row(tx: &Transaction, version: VersionId) -> Result<VersionRow, StoreError> {
let row: Option<(String, bool, Vec<u8>)> = tx
.query_row(
"SELECT form, gzip, wrapped_key FROM version WHERE version = ?1",
[version.0],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)
.optional()
.map_err(io)?;
let (form, gzip, wrapped) = row.ok_or(StoreError::Damaged)?;
let form = match form.as_str() {
"unsigned" => vcon::Form::Unsigned,
"signed" => vcon::Form::Signed,
"encrypted" => vcon::Form::Encrypted,
_ => return Err(StoreError::Damaged),
};
Ok(VersionRow {
form,
gzip,
wrapped,
})
}
fn wraps(tx: &Transaction, key: KeyId) -> Result<u64, StoreError> {
let wraps: Option<i64> = tx
.query_row(
"SELECT wraps FROM key_use WHERE tenant_key_id = ?1",
[key.bytes()],
|r| r.get(0),
)
.optional()
.map_err(io)?;
u64::try_from(wraps.unwrap_or(0)).map_err(io)
}
fn add_wrap(tx: &Transaction, key: KeyId) -> Result<(), StoreError> {
tx.execute(
"INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, 1)
ON CONFLICT (tenant_key_id) DO UPDATE SET wraps = wraps + 1",
[key.bytes()],
)
.map_err(io)?;
Ok(())
}
fn immediate(conn: &mut Connection) -> Result<Transaction<'_>, StoreError> {
conn.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(io)
}
fn commit(tx: Transaction) -> Result<(), StoreError> {
tx.commit().map_err(io)
}
fn io<E>(_: E) -> StoreError {
StoreError::Io
}
fn form_name(form: vcon::Form) -> &'static str {
match form {
vcon::Form::Unsigned => "unsigned",
vcon::Form::Signed => "signed",
vcon::Form::Encrypted => "encrypted",
}
}
#[cfg(test)]
pub(crate) mod testing {
use std::collections::BTreeSet;
use vcon::{Form, Inspection, Limits, inspect};
pub(crate) fn vcon(uuid: &str, subject: &str) -> (Inspection, Vec<u8>) {
let body = format!(
r#"{{"uuid": "{uuid}", "created_at": "2026-10-23T12:00:00Z", "subject": "{subject}", "parties": []}}"#
)
.into_bytes();
let found = inspect(&body, None, &Limits { inflated: 1 << 20 }, &BTreeSet::new())
.expect("a valid vCon");
assert_eq!(found.form, Form::Unsigned);
(found, body)
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeSet;
use serde_json::Value;
use time::macros::datetime;
use vcon::{Limits, inspect};
use super::testing::vcon;
use super::*;
use crate::audit::testing::exported;
use crate::catalog::testing::columns_taking_null;
use crate::clock::ManualClock;
use crate::fs::{CrashFs, Op};
const UUID: &str = "01a07da8-c2bb-83e5-b9a2-279e0d16bc46";
#[derive(Default)]
pub(super) struct Hooks {
pub(super) between: Mutex<Option<Box<dyn FnOnce() + Send>>>,
}
pub(super) struct Fixture {
pub(super) dir: tempfile::TempDir,
pub(super) fs: Arc<CrashFs>,
pub(super) clock: Arc<ManualClock>,
master: Key,
wrapped: Vec<u8>,
pub(super) store: Store,
pub(super) caller: Caller,
}
pub(super) const TENANT: TenantId = TenantId([0x11; 16]);
impl Fixture {
pub(super) fn open(&self) -> Result<Store, StoreError> {
let ctx = Context::Tenant { tenant: TENANT };
let tenant_key = envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps");
Store::open(
self.dir.path(),
TENANT,
tenant_key,
self.fs.clone(),
self.clock.clone(),
)
}
pub(super) fn tenant_key(&self) -> Key {
let ctx = Context::Tenant { tenant: TENANT };
envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps")
}
pub(super) fn tenant(&self) -> TenantId {
TENANT
}
fn open_beside(&self) -> Result<Store, StoreError> {
let ctx = Context::Tenant { tenant: TENANT };
let tenant_key = envelope::unwrap(&self.master, &ctx, &self.wrapped).expect("unwraps");
Store::open_beside(
self.dir.path(),
TENANT,
tenant_key,
self.fs.clone(),
self.clock.clone(),
)
}
pub(super) fn reopen(&mut self) {
let placeholder = self.open().expect("opens");
drop(std::mem::replace(&mut self.store, placeholder));
self.store = self.open().expect("reopens");
}
}
pub(super) fn fixture() -> Fixture {
let dir = tempfile::tempdir().expect("temp dir");
let fs = Arc::new(CrashFs::new(dir.path()));
let clock = Arc::new(ManualClock::at(datetime!(2026-10-23 12:00 UTC)));
let master = Key::generate().expect("master key");
let tenant_key = Key::generate().expect("tenant key");
let wrapped = envelope::wrap(&master, &Context::Tenant { tenant: TENANT }, &tenant_key);
let store =
Store::open(dir.path(), TENANT, tenant_key, fs.clone(), clock.clone()).expect("opens");
let caller = Caller {
key: Key::generate().expect("caller key").id(),
};
Fixture {
dir,
fs,
clock,
master,
wrapped,
store,
caller,
}
}
fn audit_log(store: &Store) -> Vec<Value> {
exported(&store.conn(), None)
}
fn count(store: &Store, sql: &str) -> i64 {
store.conn().query_row(sql, [], |r| r.get(0)).expect(sql)
}
fn wraps(store: &Store) -> i64 {
store
.conn()
.query_row("SELECT coalesce(sum(wraps), 0) FROM key_use", [], |r| {
r.get(0)
})
.expect("wraps")
}
fn stored_body(f: &Fixture, uuid: Uuid, version: VersionId) -> Vec<u8> {
let wrapped: Vec<u8> = f
.store
.conn()
.query_row(
"SELECT wrapped_key FROM version WHERE version = ?1",
[version.0],
|r| r.get(0),
)
.expect("row");
let ctx = Context::Version {
tenant: f.store.tenant,
uuid,
version,
};
let data_key = envelope::unwrap(&f.store.tenant_key, &ctx, &wrapped).expect("unwraps");
let blob =
f.fs.read(&f.store.objects.join(version.to_string()))
.expect("file");
envelope::open(&data_key, &ctx, &blob).expect("opens")
}
fn audit_page(f: &Fixture, after: i64, limit: u32) -> Result<Vec<u8>, StoreError> {
f.store
.audit_page(&f.caller, &audit::AuditQuery { after, limit })
}
fn export_text(store: &Store, since: Option<i64>) -> Vec<u8> {
let mut out = Vec::new();
store.export(since, &mut out).expect("export");
out
}
#[test]
fn an_audit_page_is_export_s_lines_and_records_its_read_after_them() {
let f = fixture();
for n in 0..3 {
let (found, body) = vcon(&format!("01a07da8-c2bb-83e5-b9a2-27900000000{n}"), "x");
f.store.create(&f.caller, &found, &body).expect("stored");
}
let before = export_text(&f.store, None);
assert_eq!(audit_page(&f, 0, 1000), Ok(before.clone()));
let read = audit_log(&f.store);
assert_eq!(read.len(), 4, "three creates and the read");
assert_eq!(read[3]["action"], "audit_read");
assert_eq!(read[3]["actor"], f.caller.key.to_string());
assert_eq!(read[3]["status"], 200);
assert_eq!(read[3]["seq"], 4);
assert_eq!(read[3]["target"], Value::Null);
let lines: Vec<&[u8]> = before.split_inclusive(|b| *b == b'\n').collect();
assert_eq!(audit_page(&f, 1, 2), Ok(lines[1..3].concat()));
let tail = audit_page(&f, 4, 1000).expect("a page");
assert_eq!(tail, export_text(&f.store, Some(4))[..tail.len()]);
assert_eq!(String::from_utf8(tail).expect("UTF-8").lines().count(), 1);
}
#[test]
fn an_audit_page_that_cannot_be_read_is_a_failure_and_recorded() {
let f = fixture();
f.store
.conn_for_test()
.execute(
"INSERT INTO audit (at, actor, action, status, count)
VALUES ('2026-10-23T12:00:00Z', 'cli', 'read', 70000, 1)",
[],
)
.expect("damaged");
assert_eq!(audit_page(&f, 0, 10), Err(StoreError::Io));
let mut recorded = Vec::new();
f.store.export(Some(1), &mut recorded).expect("export");
let recorded: Value = serde_json::from_slice(&recorded).expect("one record");
assert_eq!(recorded["action"], "audit_read");
assert_eq!(recorded["status"], 500);
}
#[test]
fn an_audit_page_whose_record_fails_is_refused() {
let f = fixture();
f.store.break_audit();
assert_eq!(audit_page(&f, 0, 10), Err(StoreError::AuditFailed));
}
#[test]
fn a_new_vcon_is_written_in_order_and_recorded() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let ops = f.fs.log().len();
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
let (tmp, done) = (
f.store.objects.join(format!("{version}.tmp")),
f.store.objects.join(version.to_string()),
);
assert_eq!(
f.fs.log()[ops..],
[
Op::Write(tmp.clone()),
Op::SyncFile(tmp.clone()),
Op::Rename(tmp, done),
Op::SyncDir(f.store.objects.clone()),
]
);
assert_eq!(stored_body(&f, found.uuid, version), body);
assert_eq!(
count(
&f.store,
"SELECT count(*) FROM version WHERE hidden_at IS NULL"
),
1
);
let current: Vec<u8> = f
.store
.conn()
.query_row(
"SELECT current FROM vcon WHERE uuid = ?1",
[found.uuid.as_bytes()],
|r| r.get(0),
)
.expect("vcon row");
assert_eq!(current, version.0);
assert_eq!(wraps(&f.store), 1, "one data key wrapped");
let log = audit_log(&f.store);
assert_eq!(log.len(), 1);
assert_eq!(log[0]["action"], "create");
assert_eq!(log[0]["actor"], f.caller.key.to_string());
assert_eq!(log[0]["target"], UUID);
assert_eq!(log[0]["version"], version.to_string());
assert_eq!(log[0]["status"], 201);
}
#[test]
fn the_same_body_again_is_same_and_writes_nothing() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
let ops = f.fs.log().len();
assert_eq!(
f.store.create(&f.caller, &found, &body),
Ok(Created::Same(version))
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(count(&f.store, "SELECT count(*) FROM version"), 1);
assert_eq!(wraps(&f.store), 1, "nothing wrapped");
let log = audit_log(&f.store);
assert_eq!(
(log[1]["status"].clone(), log[1]["version"].clone()),
(200.into(), version.to_string().into())
);
}
#[test]
fn another_body_for_a_stored_uuid_is_a_conflict_and_writes_nothing() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
f.store.create(&f.caller, &found, &body).expect("stored");
let ops = f.fs.log().len();
let (found, other) = vcon(UUID, "other");
assert_eq!(
f.store.create(&f.caller, &found, &other),
Err(StoreError::Conflict)
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(wraps(&f.store), 1, "nothing wrapped");
assert_eq!(audit_log(&f.store)[1]["status"], 409);
}
#[test]
fn a_deleted_or_purged_uuid_is_gone_and_writes_nothing() {
for sql in [
"INSERT INTO vcon (uuid, current, deleted_at) VALUES (?1, NULL, 1)",
"INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
] {
let f = fixture();
let (found, body) = vcon(UUID, "first");
f.store
.conn()
.execute(sql, [found.uuid.as_bytes()])
.expect(sql);
let ops = f.fs.log().len();
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::Gone),
"{sql}"
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(wraps(&f.store), 0, "nothing wrapped");
assert_eq!(audit_log(&f.store)[0]["status"], 410);
}
}
#[test]
fn counts_include_hidden_versions_and_held_uuids() {
let f = fixture();
let none = Counts {
versions: 0,
held: 0,
};
assert_eq!(f.store.counts(), Ok(none));
let (found, old) = stored(&f, "first");
let (_, newer) = vcon(UUID, "second");
f.store
.replace(&f.caller, old, &found, &newer)
.expect("replaced");
f.store
.delete(&f.caller, found.uuid, Retention::Years(7))
.expect("deleted");
let (other, body) = vcon("01a07da8-c2bb-83e5-b9a2-279e0d16bc47", "other");
f.store.create(&f.caller, &other, &body).expect("stored");
assert_eq!(
f.store.counts(),
Ok(Counts {
versions: 3,
held: 0
}),
"two hidden versions and a live one"
);
f.store
.conn()
.execute(
"UPDATE vcon SET hold = 1 WHERE uuid = ?1",
[found.uuid.as_bytes()],
)
.expect("a hold");
assert_eq!(f.store.counts().map(|c| c.held), Ok(1));
}
#[test]
fn a_negative_wrap_count_is_damage_and_stores_nothing() {
let f = fixture();
f.store
.conn()
.execute(
"INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, -1)",
[f.store.tenant_key.id().bytes()],
)
.expect("a damaged count");
let (found, body) = vcon(UUID, "first");
let ops = f.fs.log().len();
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::Io)
);
assert_eq!(f.fs.log().len(), ops, "no file written");
}
#[test]
fn a_tenant_key_at_its_wrap_limit_refuses_new_vcons() {
let f = fixture();
f.store
.conn()
.execute(
"INSERT INTO key_use (tenant_key_id, wraps) VALUES (?1, ?2)",
params![f.store.tenant_key.id().bytes(), 1_i64 << 32],
)
.expect("at the limit");
let (found, body) = vcon(UUID, "first");
let ops = f.fs.log().len();
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::LimitReached)
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(audit_log(&f.store)[0]["status"], 503);
}
#[test]
fn a_missing_current_file_is_damage_not_a_conflict() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
f.fs.remove(&f.store.objects.join(version.to_string()))
.expect("remove");
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::Damaged)
);
assert_eq!(audit_log(&f.store)[1]["status"], 500);
}
#[test]
fn a_racing_create_that_loses_gets_a_conflict_and_leaves_no_file() {
let f = Arc::new(fixture());
let (found, body) = vcon(UUID, "first");
let winner = f.clone();
*f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
let (found, other) = vcon(UUID, "second");
assert!(matches!(
winner.store.create(&winner.caller, &found, &other),
Ok(Created::New(_))
));
}));
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::Conflict)
);
assert_eq!(
count(&f.store, "SELECT count(*) FROM version"),
1,
"only the winner"
);
assert_eq!(
f.fs.list(&f.store.objects).expect("list").len(),
1,
"the loser's file is gone"
);
assert_eq!(wraps(&f.store), 2, "both wraps are counted");
let statuses: Vec<_> = audit_log(&f.store)
.iter()
.map(|r| r["status"].clone())
.collect();
assert_eq!(statuses, [201, 409]);
}
#[test]
fn a_create_whose_audit_record_fails_leaves_no_row_and_no_file() {
let f = fixture();
f.store
.conn()
.execute_batch(
"CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
)
.expect("trigger");
let (found, body) = vcon(UUID, "first");
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::AuditFailed)
);
assert_eq!(count(&f.store, "SELECT count(*) FROM version"), 0);
assert_eq!(count(&f.store, "SELECT count(*) FROM vcon"), 0);
assert_eq!(
f.fs.list(&f.store.objects).expect("list"),
Vec::<std::ffi::OsString>::new()
);
}
fn get_entry_status(f: &Fixture) -> Value {
audit_log(&f.store).last().expect("a record")["status"].clone()
}
#[test]
fn a_stored_vcon_reads_back_byte_for_byte_and_is_recorded() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
let stored = f.store.get(&f.caller, found.uuid).expect("reads");
assert_eq!(
stored,
Stored {
version,
form: vcon::Form::Unsigned,
gzip: false,
body
}
);
let log = audit_log(&f.store);
let read = log.last().expect("a record");
assert_eq!(read["action"], "read");
assert_eq!(read["status"], 200);
assert_eq!(read["version"], version.to_string());
assert_eq!(read["target"], UUID);
}
#[test]
fn a_uuid_never_stored_is_not_found() {
let f = fixture();
let (found, _) = vcon(UUID, "first");
assert_eq!(
f.store.get(&f.caller, found.uuid),
Err(StoreError::NotFound)
);
assert_eq!(get_entry_status(&f), 404);
}
#[test]
fn a_deleted_or_purged_uuid_is_gone() {
for sql in [
"UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
"DELETE FROM vcon WHERE uuid = ?1; INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
] {
let f = fixture();
let (found, body) = vcon(UUID, "first");
f.store.create(&f.caller, &found, &body).expect("stored");
for statement in sql.split("; ") {
f.store
.conn()
.execute(statement, [found.uuid.as_bytes()])
.expect(statement);
}
assert_eq!(
f.store.get(&f.caller, found.uuid),
Err(StoreError::Gone),
"{sql}"
);
assert_eq!(get_entry_status(&f), 410);
}
}
#[test]
fn a_missing_or_damaged_file_is_a_500_never_an_empty_body() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
let file = f.store.objects.join(version.to_string());
let blob = f.fs.read(&file).expect("file");
f.fs.remove(&file).expect("remove");
assert_eq!(
f.store.get(&f.caller, found.uuid),
Err(StoreError::Damaged),
"missing"
);
assert_eq!(get_entry_status(&f), 500);
let mut bad = blob.clone();
let last = bad.len() - 1;
bad[last] ^= 1;
f.fs.write_new(&file, &bad).expect("write");
assert_eq!(
f.store.get(&f.caller, found.uuid),
Err(StoreError::Damaged),
"tampered"
);
}
#[test]
fn a_read_racing_a_delete_never_returns_the_body() {
let f = Arc::new(fixture());
let (found, body) = vcon(UUID, "first");
f.store.create(&f.caller, &found, &body).expect("stored");
let deleter = f.clone();
*f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
deleter
.store
.conn()
.execute(
"UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
[found.uuid.as_bytes()],
)
.expect("delete");
}));
assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
assert_eq!(get_entry_status(&f), 410);
}
#[test]
fn a_read_racing_a_new_version_starts_again_and_returns_the_new_one() {
let f = Arc::new(fixture());
let (found, body) = vcon(UUID, "first");
f.store.create(&f.caller, &found, &body).expect("stored");
let (_, newer) = vcon(UUID, "newer");
let (version, wrapped, _) = f.store.write_version(found.uuid, &newer).expect("written");
let writer = f.clone();
*f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
let conn = writer.store.conn();
conn.execute(
"INSERT INTO version
(version, uuid, form, gzip, size, stored_at, tenant_key_id, wrapped_key)
VALUES (?1, ?2, 'unsigned', 0, 0, 0, ?3, ?4)",
params![
version.0,
found.uuid.as_bytes(),
writer.store.tenant_key.id().bytes(),
wrapped
],
)
.expect("new version");
conn.execute(
"UPDATE vcon SET current = ?1 WHERE uuid = ?2",
params![version.0, found.uuid.as_bytes()],
)
.expect("made current");
}));
let stored = f.store.get(&f.caller, found.uuid).expect("reads");
assert_eq!((stored.version, stored.body), (version, newer));
let reads = audit_log(&f.store)
.iter()
.filter(|r| r["action"] == "read")
.count();
assert_eq!(reads, 1, "one record for the one request");
}
#[test]
fn every_form_and_the_gzip_flag_read_back_as_stored() {
use base64::Engine;
use std::io::Write;
let f = fixture();
let uuids = [
"01a07da8-c2bb-83e5-b9a2-279e0d16bc01",
"01a07da8-c2bb-83e5-b9a2-279e0d16bc02",
"01a07da8-c2bb-83e5-b9a2-279e0d16bc03",
];
let (_, unsigned) = vcon(uuids[0], "plain");
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
gz.write_all(&unsigned).expect("gzip");
let gzipped = gz.finish().expect("gzip");
let (_, inner) = vcon(uuids[1], "signed");
let signed = format!(
r#"{{"payload": "{}", "signatures": [{{"header": {{}}}}]}}"#,
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(inner)
)
.into_bytes();
let encrypted = format!(
r#"{{"unprotected": {{"uuid": "{}"}}, "ciphertext": "x", "recipients": []}}"#,
uuids[2]
)
.into_bytes();
let cases = [
(gzipped, vcon::Form::Unsigned, true),
(signed, vcon::Form::Signed, false),
(encrypted, vcon::Form::Encrypted, false),
];
for (body, form, gzip) in cases {
let found = inspect(&body, None, &Limits { inflated: 1 << 20 }, &BTreeSet::new())
.expect("a valid vCon");
assert_eq!((found.form, found.gzip), (form, gzip));
f.store.create(&f.caller, &found, &body).expect("stored");
let stored = f.store.get(&f.caller, found.uuid).expect("reads");
assert_eq!((stored.form, stored.gzip, stored.body), (form, gzip, body));
}
}
fn stored(f: &Fixture, subject: &str) -> (Inspection, VersionId) {
let (found, body) = vcon(UUID, subject);
let Ok(Created::New(version)) = f.store.create(&f.caller, &found, &body) else {
panic!("a new vCon")
};
(found, version)
}
#[test]
fn a_replace_makes_a_new_current_version_and_hides_the_old_one() {
let f = fixture();
let (found, old) = stored(&f, "first");
let (_, newer) = vcon(UUID, "second");
let version = f
.store
.replace(&f.caller, old, &found, &newer)
.expect("replaced");
assert_ne!(version, old);
let stored = f.store.get(&f.caller, found.uuid).expect("reads");
assert_eq!((stored.version, stored.body), (version, newer));
let hidden: Option<i64> = f
.store
.conn()
.query_row(
"SELECT hidden_at FROM version WHERE version = ?1",
[old.0],
|r| r.get(0),
)
.expect("the old row is kept");
assert!(hidden.is_some(), "the old version is hidden");
assert!(
f.fs.read(&f.store.objects.join(old.to_string())).is_ok(),
"its file is kept"
);
let record = audit_log(&f.store)
.into_iter()
.find(|r| r["action"] == "replace")
.expect("a replace record");
assert_eq!(
(record["status"].clone(), record["version"].clone()),
(200.into(), version.to_string().into())
);
}
#[test]
fn a_replace_of_a_missing_or_gone_uuid_is_refused() {
let f = fixture();
let (found, body) = vcon(UUID, "first");
let any = VersionId([7; 16]);
assert_eq!(
f.store.replace(&f.caller, any, &found, &body),
Err(StoreError::NotFound)
);
assert_eq!(get_entry_status(&f), 404);
let (found, old) = stored(&f, "first");
f.store
.conn()
.execute(
"UPDATE vcon SET current = NULL, deleted_at = 1 WHERE uuid = ?1",
[found.uuid.as_bytes()],
)
.expect("delete");
assert_eq!(
f.store.replace(&f.caller, old, &found, &body),
Err(StoreError::Gone)
);
assert_eq!(get_entry_status(&f), 410);
}
#[test]
fn a_stale_if_match_is_refused_and_writes_nothing() {
let f = fixture();
let (found, _) = stored(&f, "first");
let ops = f.fs.log().len();
let (_, newer) = vcon(UUID, "second");
assert_eq!(
f.store
.replace(&f.caller, VersionId([7; 16]), &found, &newer),
Err(StoreError::PreconditionFailed)
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(wraps(&f.store), 1, "nothing wrapped");
assert_eq!(get_entry_status(&f), 412);
}
#[test]
fn a_signed_or_encrypted_current_version_is_immutable() {
use base64::Engine;
let f = fixture();
let (found, old) = stored(&f, "first");
let (_, inner) = vcon(UUID, "signed");
let signed = format!(
r#"{{"payload": "{}", "signatures": [{{"header": {{}}}}]}}"#,
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(inner)
)
.into_bytes();
let signed_found = inspect(
&signed,
None,
&Limits { inflated: 1 << 20 },
&BTreeSet::new(),
)
.expect("a signed vCon");
let version = f
.store
.replace(&f.caller, old, &signed_found, &signed)
.expect("an unsigned vCon may be replaced by a signed one");
let (_, newer) = vcon(UUID, "second");
let ops = f.fs.log().len();
assert_eq!(
f.store.replace(&f.caller, version, &found, &newer),
Err(StoreError::Immutable)
);
assert_eq!(f.fs.log().len(), ops, "no file written");
assert_eq!(get_entry_status(&f), 409);
let g = fixture();
let encrypted = format!(
r#"{{"unprotected": {{"uuid": "{UUID}"}}, "ciphertext": "x", "recipients": []}}"#
)
.into_bytes();
let enc_found = inspect(
&encrypted,
None,
&Limits { inflated: 1 << 20 },
&BTreeSet::new(),
)
.expect("an encrypted vCon");
let Ok(Created::New(enc_version)) = g.store.create(&g.caller, &enc_found, &encrypted)
else {
panic!("stored")
};
assert_eq!(
g.store.replace(&g.caller, enc_version, &found, &newer),
Err(StoreError::Immutable)
);
}
#[test]
fn a_replace_at_the_wrap_limit_is_refused() {
let f = fixture();
let (found, old) = stored(&f, "first");
f.store
.conn()
.execute("UPDATE key_use SET wraps = ?1", [1_i64 << 32])
.expect("at the limit");
let (_, newer) = vcon(UUID, "second");
assert_eq!(
f.store.replace(&f.caller, old, &found, &newer),
Err(StoreError::LimitReached)
);
assert_eq!(get_entry_status(&f), 503);
}
#[test]
fn of_two_replaces_racing_on_one_version_the_second_to_commit_loses() {
let f = Arc::new(fixture());
let (found, old) = stored(&f, "first");
let winner = f.clone();
let racer = found;
*f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
let (_, body) = vcon(UUID, "winner");
winner
.store
.replace(&winner.caller, old, &racer, &body)
.expect("wins");
}));
let (_, body) = vcon(UUID, "loser");
assert_eq!(
f.store.replace(&f.caller, old, &found, &body),
Err(StoreError::PreconditionFailed)
);
assert_eq!(
f.fs.list(&f.store.objects).expect("list").len(),
2,
"first and winner only"
);
assert_eq!(wraps(&f.store), 3, "every wrap counted");
assert_eq!(get_entry_status(&f), 412);
}
#[test]
fn two_threads_replacing_with_the_same_if_match_one_wins() {
let f = Arc::new(fixture());
let (_, old) = stored(&f, "first");
let threads: Vec<_> = ["one", "two"]
.into_iter()
.map(|subject| {
let f = f.clone();
std::thread::spawn(move || {
let (found, body) = vcon(UUID, subject);
f.store.replace(&f.caller, old, &found, &body)
})
})
.collect();
let results: Vec<_> = threads
.into_iter()
.map(|t| t.join().expect("thread"))
.collect();
assert_eq!(
results.iter().filter(|r| r.is_ok()).count(),
1,
"{results:?}"
);
assert!(
results.contains(&Err(StoreError::PreconditionFailed)),
"{results:?}"
);
}
#[test]
fn a_replace_whose_audit_record_fails_changes_nothing() {
let f = fixture();
let (found, old) = stored(&f, "first");
f.store
.conn()
.execute_batch(
"CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
)
.expect("trigger");
let (_, newer) = vcon(UUID, "second");
assert_eq!(
f.store.replace(&f.caller, old, &found, &newer),
Err(StoreError::AuditFailed)
);
f.store
.conn()
.execute_batch("DROP TRIGGER no_audit")
.expect("drop trigger");
assert_eq!(
f.store.get(&f.caller, found.uuid).expect("reads").version,
old
);
assert_eq!(
f.fs.list(&f.store.objects).expect("list").len(),
1,
"only the first file"
);
}
fn hidden_versions(f: &Fixture) -> i64 {
count(
&f.store,
"SELECT count(*) FROM version WHERE hidden_at IS NOT NULL",
)
}
#[test]
fn a_deleted_uuid_answers_gone_and_keeps_its_versions_hidden() {
let f = fixture();
let (found, old) = stored(&f, "first");
let (_, newer) = vcon(UUID, "second");
f.clock.set(datetime!(2026-10-23 13:00 UTC));
let current = f
.store
.replace(&f.caller, old, &found, &newer)
.expect("replaced");
f.clock.set(datetime!(2026-10-23 14:00 UTC));
assert_eq!(
f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
Ok(())
);
let record = audit_log(&f.store).pop().expect("a record");
assert_eq!(
(record["action"].clone(), record["status"].clone()),
("delete".into(), 204.into())
);
let hidden_at = |version: VersionId| -> i64 {
f.store
.conn()
.query_row(
"SELECT hidden_at FROM version WHERE version = ?1",
[version.0],
|r| r.get(0),
)
.expect("hidden")
};
assert_eq!(
hidden_at(old),
nanos(datetime!(2026-10-23 13:00 UTC)),
"hidden by the replace"
);
assert_eq!(
hidden_at(current),
nanos(datetime!(2026-10-23 14:00 UTC)),
"hidden by the delete"
);
assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
let (_, body) = vcon(UUID, "third");
assert_eq!(
f.store.create(&f.caller, &found, &body),
Err(StoreError::Gone)
);
assert_eq!(hidden_versions(&f), 2, "both versions hidden");
assert_eq!(
f.fs.list(&f.store.objects).expect("list").len(),
2,
"files kept for legal reads"
);
let (current, deleted_at): (Option<Vec<u8>>, Option<i64>) = f
.store
.conn()
.query_row(
"SELECT current, deleted_at FROM vcon WHERE uuid = ?1",
[found.uuid.as_bytes()],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.expect("row");
assert_eq!(current, None);
assert_eq!(deleted_at, Some(nanos(datetime!(2026-10-23 14:00 UTC))));
}
#[test]
fn deleting_a_deleted_or_purged_uuid_succeeds_again() {
let f = fixture();
let (found, _) = stored(&f, "first");
f.store
.delete(&f.caller, found.uuid, Retention::Years(7))
.expect("deleted");
assert_eq!(
f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
Ok(())
);
assert_eq!(get_entry_status(&f), 204);
let g = fixture();
g.store
.conn()
.execute(
"INSERT INTO tombstone (uuid, purged_at) VALUES (?1, 1)",
[found.uuid.as_bytes()],
)
.expect("purged");
assert_eq!(
g.store.delete(&g.caller, found.uuid, Retention::Years(7)),
Ok(())
);
assert_eq!(get_entry_status(&g), 204);
}
#[test]
fn deleting_a_uuid_never_stored_is_not_found() {
let f = fixture();
let (found, _) = vcon(UUID, "first");
assert_eq!(
f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
Err(StoreError::NotFound)
);
assert_eq!(get_entry_status(&f), 404);
}
#[test]
fn a_hold_does_not_stop_a_delete() {
let f = fixture();
let (found, _) = stored(&f, "first");
f.store
.conn()
.execute(
"UPDATE vcon SET hold = 1 WHERE uuid = ?1",
[found.uuid.as_bytes()],
)
.expect("hold");
assert_eq!(
f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
Ok(())
);
assert_eq!(f.store.get(&f.caller, found.uuid), Err(StoreError::Gone));
}
#[test]
fn a_delete_whose_audit_record_fails_changes_nothing() {
let f = fixture();
let (found, version) = stored(&f, "first");
f.store
.conn()
.execute_batch(
"CREATE TEMP TRIGGER no_audit BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'audit fails'); END;",
)
.expect("trigger");
assert_eq!(
f.store.delete(&f.caller, found.uuid, Retention::Years(7)),
Err(StoreError::AuditFailed)
);
f.store
.conn()
.execute_batch("DROP TRIGGER no_audit")
.expect("drop trigger");
assert_eq!(
f.store
.get(&f.caller, found.uuid)
.expect("still live")
.version,
version
);
assert_eq!(hidden_versions(&f), 0);
}
#[test]
fn only_the_optional_catalog_columns_may_be_empty() {
let f = fixture();
let conn = f.store.conn();
let tables: [(&str, &[(&str, &str)]); 5] = [
(
"vcon",
&[
("uuid", "x'01'"),
("current", "x'02'"),
("deleted_at", "1"),
("hold", "0"),
],
),
(
"version",
&[
("version", "x'02'"),
("uuid", "x'01'"),
("form", "'unsigned'"),
("gzip", "0"),
("size", "1"),
("stored_at", "1"),
("hidden_at", "1"),
("tenant_key_id", "x'03'"),
("wrapped_key", "x'04'"),
],
),
("tombstone", &[("uuid", "x'01'"), ("purged_at", "1")]),
("key_use", &[("tenant_key_id", "x'03'"), ("wraps", "1")]),
(
"audit",
&[
("at", "'2026-10-23T12:00Z'"),
("actor", "'cli'"),
("action", "'create'"),
("target", "'x'"),
("version", "'y'"),
("status", "201"),
("count", "1"),
],
),
];
let may_be_empty: Vec<String> = tables
.iter()
.flat_map(|(table, row)| columns_taking_null(&conn, table, row))
.collect();
assert_eq!(
may_be_empty,
[
"vcon.current",
"vcon.deleted_at",
"version.hidden_at",
"audit.target",
"audit.version"
]
);
}
#[test]
fn a_new_stores_objects_directory_survives_a_crash() {
let f = fixture();
f.fs.crash();
assert_eq!(
f.fs.list(f.dir.path()).expect("list"),
[std::ffi::OsString::from("objects")]
);
}
#[test]
fn opening_removes_temporary_and_orphaned_files_and_nothing_else() {
let mut f = fixture();
let (found, version) = stored(&f, "first");
let orphan = VersionId([0x0e; 16]);
let objects = f.store.objects.clone();
for name in [
format!("{orphan}"),
format!("{orphan}.tmp"),
format!("{version}.tmp"),
"notes.txt".to_owned(),
] {
f.fs.write_new(&objects.join(name), b"x").expect("write");
}
f.reopen();
let names: Vec<String> =
f.fs.list(&objects)
.expect("list")
.into_iter()
.map(|n| n.to_string_lossy().into_owned())
.collect();
assert_eq!(names, [version.to_string(), "notes.txt".to_owned()]);
assert_eq!(
f.store
.get(&f.caller, found.uuid)
.expect("still reads")
.version,
version
);
let log = f.fs.log();
let last_removal = log
.iter()
.rposition(|op| matches!(op, Op::Remove(_)))
.expect("files were removed");
assert!(
log[last_removal..].contains(&Op::SyncDir(objects)),
"the removals are synced"
);
}
#[test]
fn opening_beside_a_server_removes_nothing() {
let f = fixture();
let (found, version) = stored(&f, "first");
let orphan = VersionId([0x0e; 16]);
let objects = f.store.objects.clone();
let mut written = vec![version.to_string()];
for name in [
format!("{orphan}"),
format!("{orphan}.tmp"),
format!("{version}.tmp"),
] {
f.fs.write_new(&objects.join(&name), b"x").expect("write");
written.push(name);
}
written.sort();
let ops = f.fs.log().len();
let beside = f.open_beside().expect("opens beside");
let mut names: Vec<String> =
f.fs.list(&objects)
.expect("list")
.into_iter()
.map(|n| n.to_string_lossy().into_owned())
.collect();
names.sort();
assert_eq!(names, written);
assert!(
!f.fs.log()[ops..]
.iter()
.any(|op| matches!(op, Op::Remove(_))),
"nothing removed"
);
assert_eq!(
beside.get(&f.caller, found.uuid).expect("reads").version,
version
);
}
#[test]
fn opening_checkpoints_the_wal_with_truncation() {
let mut f = fixture();
stored(&f, "first");
let wal = f.dir.path().join("catalog.sqlite-wal");
assert!(
std::fs::metadata(&wal).expect("a WAL").len() > 0,
"the WAL holds the write"
);
f.reopen();
assert_eq!(std::fs::metadata(&wal).expect("a WAL").len(), 0);
}
#[test]
fn check_reports_a_version_whose_file_is_missing() {
let f = fixture();
let (_, version) = stored(&f, "first");
assert_eq!(f.store.check(), Ok(vec![]));
f.fs.remove(&f.store.objects.join(version.to_string()))
.expect("remove");
assert_eq!(f.store.check(), Ok(vec![version]));
}
#[test]
fn opening_refuses_while_a_reader_stops_the_wal_from_being_truncated() {
let f = fixture();
stored(&f, "first");
let ctx = Context::Tenant { tenant: TENANT };
let tenant_key = envelope::unwrap(&f.master, &ctx, &f.wrapped).expect("unwraps");
let catalog_key = envelope::catalog_key(&tenant_key).expect("catalog key");
let mut reader = catalog::open(&f.dir.path().join("catalog.sqlite"), &catalog_key, "")
.expect("a second connection");
let read = reader.transaction().expect("read transaction");
let _: i64 = read
.query_row("SELECT count(*) FROM vcon", [], |r| r.get(0))
.expect("a read holds a snapshot");
assert!(matches!(f.open(), Err(StoreError::Io)));
drop(read);
assert!(f.open().is_ok(), "opens once the reader is done");
}
const WRITE_STEPS: usize = 4;
fn assert_whole(f: &Fixture, step: usize) {
assert_eq!(
f.store.check(),
Ok(vec![]),
"a row without its file at step {step}"
);
for name in f.fs.list(&f.store.objects).expect("list") {
let name = name.to_string_lossy().into_owned();
let version = VersionId::parse(&name)
.unwrap_or_else(|| panic!("{name} left over at step {step}"));
let rows = count(
&f.store,
&format!("SELECT count(*) FROM version WHERE hex(version) = upper('{version}')"),
);
assert_eq!(rows, 1, "{name} has no row at step {step}");
}
}
pub(super) fn block_commits(f: &Fixture) -> Connection {
let ctx = Context::Tenant { tenant: TENANT };
let tenant_key = envelope::unwrap(&f.master, &ctx, &f.wrapped).expect("unwraps");
let catalog_key = envelope::catalog_key(&tenant_key).expect("catalog key");
catalog::open(&f.dir.path().join("catalog.sqlite"), &catalog_key, "")
.expect("second connection")
}
fn die_before_commit(f: &Fixture) {
let other = block_commits(f);
let fs = f.fs.clone();
*f.store.hooks.between.lock().expect("hook") = Some(Box::new(move || {
fs.fail_from(0);
other
.execute_batch(
"CREATE TRIGGER dead BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'the process has died'); END;",
)
.expect("trigger");
}));
}
pub(super) fn restart(f: &mut Fixture) {
f.fs.crash();
f.store
.conn()
.execute_batch("DROP TRIGGER IF EXISTS dead")
.expect("drop trigger");
f.reopen();
}
#[test]
fn a_create_cut_off_at_any_step_leaves_no_vcon_or_the_whole_vcon() {
let (found, body) = vcon(UUID, "first");
for step in 0..=WRITE_STEPS {
let mut f = fixture();
f.fs.fail_from(f.fs.log().len() + step);
let result = f.store.create(&f.caller, &found, &body);
assert_eq!(
result.is_ok(),
step == WRITE_STEPS,
"step {step}: {result:?}"
);
restart(&mut f);
match f.store.get(&f.caller, found.uuid) {
Ok(stored) => assert_eq!(stored.body, body, "step {step}"),
Err(StoreError::NotFound) => {
assert!(result.is_err(), "acknowledged, then lost, at step {step}")
}
other => panic!("half a vCon at step {step}: {other:?}"),
}
assert_whole(&f, step);
}
}
#[test]
fn a_create_cut_off_between_its_file_and_its_commit_leaves_nothing() {
let mut f = fixture();
let (found, body) = vcon(UUID, "first");
die_before_commit(&f);
assert!(f.store.create(&f.caller, &found, &body).is_err());
restart(&mut f);
assert_eq!(
f.store.get(&f.caller, found.uuid),
Err(StoreError::NotFound)
);
assert_whole(&f, WRITE_STEPS);
}
#[test]
fn a_replace_cut_off_at_any_step_leaves_the_old_version_or_the_new() {
let (found, old_body) = vcon(UUID, "first");
let (_, new_body) = vcon(UUID, "second");
for step in 0..=WRITE_STEPS {
let mut f = fixture();
let Ok(Created::New(old)) = f.store.create(&f.caller, &found, &old_body) else {
panic!("stored")
};
f.fs.fail_from(f.fs.log().len() + step);
let result = f.store.replace(&f.caller, old, &found, &new_body);
assert_eq!(
result.is_ok(),
step == WRITE_STEPS,
"step {step}: {result:?}"
);
restart(&mut f);
let stored = f
.store
.get(&f.caller, found.uuid)
.unwrap_or_else(|e| panic!("step {step}: {e:?}"));
match result {
Ok(new) => assert_eq!(
(stored.version, stored.body),
(new, new_body.clone()),
"step {step}"
),
Err(_) => assert_eq!(
(stored.version, stored.body),
(old, old_body.clone()),
"step {step}"
),
}
assert_whole(&f, step);
}
}
#[test]
fn a_replace_cut_off_between_its_file_and_its_commit_keeps_the_old_version() {
let mut f = fixture();
let (found, old) = stored(&f, "first");
let (_, newer) = vcon(UUID, "second");
die_before_commit(&f);
assert!(f.store.replace(&f.caller, old, &found, &newer).is_err());
restart(&mut f);
assert_eq!(
f.store.get(&f.caller, found.uuid).expect("reads").version,
old
);
assert_eq!(hidden_versions(&f), 0);
assert_whole(&f, WRITE_STEPS);
}
#[test]
fn a_delete_cut_off_before_its_commit_leaves_the_vcon_live() {
let mut f = fixture();
let (found, version) = stored(&f, "first");
let other = block_commits(&f);
other
.execute_batch(
"CREATE TRIGGER dead BEFORE INSERT ON audit
BEGIN SELECT RAISE(ABORT, 'the process has died'); END;",
)
.expect("trigger");
assert!(
f.store
.delete(&f.caller, found.uuid, Retention::Years(7))
.is_err()
);
drop(other);
restart(&mut f);
assert_eq!(
f.store
.get(&f.caller, found.uuid)
.expect("still live")
.version,
version
);
assert_eq!(hidden_versions(&f), 0);
}
}