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