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