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