1use std::collections::{BTreeMap, HashMap, HashSet};
2use std::fmt;
3use std::fs;
4use std::io;
5use std::path::{Path, PathBuf};
6use std::time::Duration;
7
8use serde_json::Value;
9
10use crate::atomic::AtomicTemporaryFileError;
11use crate::catalog::{
12 check_integrity, declare_relation as declare_relation_in, empty_catalog, expand_indexes, index_identity,
13 parse_catalog, read_catalog_text, reference_of, relations_touching, write_catalog, write_catalog_sealed, Catalog,
14 CatalogCompaction, CatalogLayout, ExpandedIndex, IntegrityReport, ReadCatalogError, Relation, RelationAction,
15 TableEntry, TableOptions, TriggerEntry,
16};
17use crate::checkpoint::{read_snapshot, RestoreOutcome};
18use std::sync::Arc;
19
20use crate::crypto::{CipherOptions, FieldCipher, PasswordHashParams, Sealer};
21use crate::errors::{HaysonCipherError, HaysonRelationRestrictError, HaysonRelationViolationError, HaysonSchemaError};
22use crate::lock::{DirLock, DirLockError, LockOptions};
23use crate::query::builder::QueryBuilder;
24use crate::schema::{is_partitioned, parse_schema, FieldSpec, HaysonRecord, TableSchema};
25use crate::storage::{partition_names_in, LOG_SUFFIX};
26use crate::table::{describe_table_document, CachePolicy, FollowPolicy, Table, TableError};
27use crate::txn::{apply_prepared, recover_incomplete_sealed, PreparedUnit, TxnError, UnitOfWork, UnitOfWorkResult};
28
29use super::layout::{resolve_kind, stored_kind_with, Layout, LayoutKind, LayoutOptions, DEFAULT_METADATA_FILE};
30const DEFAULT_SNAPSHOT_EVERY: u64 = 1000;
33const DEFAULT_COMPACTION: CatalogCompaction = CatalogCompaction::Enabled {
36 dead_bytes_ratio: 0.4,
37 min_dead_bytes: 64 * 1024,
38};
39const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
40const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
41
42#[derive(Debug)]
43pub enum DatabaseError {
44 Lock(DirLockError),
45 Schema(HaysonSchemaError),
46 ReadOnly {
49 dir: PathBuf,
50 operation: String,
51 },
52 FieldInUse {
55 table: String,
56 field: String,
57 records: usize,
58 },
59 Catalog(String),
60 Io(io::Error),
61 Log(PathBuf, io::Error),
64 Table(String, TableError),
65 Txn(TxnError),
66 Cipher(Box<HaysonCipherError>),
69}
70
71impl fmt::Display for DatabaseError {
72 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
73 match self {
74 DatabaseError::Lock(error) => error.fmt(f),
75 DatabaseError::Schema(error) => error.fmt(f),
76 DatabaseError::ReadOnly { dir, operation } => {
77 write!(f, "database \"{}\" is readonly — {operation} is not available", dir.display())
78 }
79 DatabaseError::FieldInUse { table, field, records } => write!(
80 f,
81 "field \"{field}\" of table \"{table}\" still holds a value in {records} live record(s) — pass force to drop it anyway"
82 ),
83 DatabaseError::Catalog(message) => write!(f, "{message}"),
84 DatabaseError::Io(error) => error.fmt(f),
85 DatabaseError::Log(path, error) => write!(f, "{error}, open \"{}\"", path.display()),
86 DatabaseError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
87 DatabaseError::Txn(error) => error.fmt(f),
88 DatabaseError::Cipher(error) => error.fmt(f),
89 }
90 }
91}
92
93impl std::error::Error for DatabaseError {}
94
95impl From<io::Error> for DatabaseError {
96 fn from(value: io::Error) -> Self {
97 DatabaseError::Io(value)
98 }
99}
100
101impl From<DirLockError> for DatabaseError {
102 fn from(value: DirLockError) -> Self {
103 DatabaseError::Lock(value)
104 }
105}
106
107impl From<HaysonSchemaError> for DatabaseError {
108 fn from(value: HaysonSchemaError) -> Self {
109 DatabaseError::Schema(value)
110 }
111}
112
113impl From<TxnError> for DatabaseError {
114 fn from(value: TxnError) -> Self {
115 DatabaseError::Txn(value)
116 }
117}
118
119impl From<HaysonCipherError> for DatabaseError {
120 fn from(value: HaysonCipherError) -> Self {
121 DatabaseError::Cipher(Box::new(value))
122 }
123}
124
125#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
133pub enum FollowOption {
134 #[default]
135 Default,
136 Off,
138 Every(Duration),
140}
141
142pub const DEFAULT_FOLLOW_INTERVAL: Duration = Duration::ZERO;
145
146impl FollowOption {
147 fn resolve(self, readonly: bool) -> Option<FollowPolicy> {
148 match self {
149 FollowOption::Off => None,
150 FollowOption::Every(interval) => Some(FollowPolicy { interval }),
151 FollowOption::Default if readonly => Some(FollowPolicy {
152 interval: DEFAULT_FOLLOW_INTERVAL,
153 }),
154 FollowOption::Default => None,
155 }
156 }
157}
158
159#[derive(Debug, Clone, Copy)]
166pub enum LockMode {
167 Enabled(LockOptions),
169 Skip,
171}
172
173impl Default for LockMode {
174 fn default() -> Self {
175 LockMode::Enabled(LockOptions::default())
176 }
177}
178
179impl LockMode {
180 pub fn options(&self) -> Option<&LockOptions> {
183 match self {
184 LockMode::Enabled(options) => Some(options),
185 LockMode::Skip => None,
186 }
187 }
188
189 pub fn options_mut(&mut self) -> Option<&mut LockOptions> {
191 match self {
192 LockMode::Enabled(options) => Some(options),
193 LockMode::Skip => None,
194 }
195 }
196}
197
198#[derive(Debug, Clone)]
200pub struct OpenOptions {
201 pub readonly: bool,
202 pub lock: LockMode,
203 pub snapshot_every: u64,
206 pub compaction: Option<CatalogCompaction>,
211 pub cypher: Option<CipherOptions>,
213 pub follow: FollowOption,
215 pub cache: CachePolicy,
220 pub max_image_bytes: Option<u64>,
224 pub password_hash: PasswordHashParams,
228 pub encryption: Option<CipherOptions>,
233 pub layout: LayoutOptions,
238}
239
240impl Default for OpenOptions {
241 fn default() -> Self {
242 OpenOptions {
243 readonly: false,
244 lock: LockMode::default(),
245 snapshot_every: DEFAULT_SNAPSHOT_EVERY,
246 compaction: None,
247 cypher: None,
248 follow: FollowOption::default(),
249 cache: CachePolicy::default(),
250 max_image_bytes: None,
251 password_hash: PasswordHashParams::default(),
252 encryption: None,
253 layout: LayoutOptions::default(),
254 }
255 }
256}
257
258fn now_ms() -> u64 {
259 chrono::Utc::now().timestamp_millis().max(0) as u64
260}
261
262fn scalar_as_text(value: &Value) -> String {
267 match value {
268 Value::String(s) => s.clone(),
269 other => other.to_string(),
270 }
271}
272
273fn fnv1a(bytes: &[u8]) -> String {
274 let mut hash = FNV_OFFSET;
275 for byte in bytes {
276 hash ^= u64::from(*byte);
277 hash = hash.wrapping_mul(FNV_PRIME);
278 }
279 format!("{hash:016x}")
280}
281
282fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
290 let schema = format!("{schema:?}").replace(", password_hash: false", "");
292 if declared.is_empty() {
293 return fnv1a(schema.as_bytes());
294 }
295 let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
296 described.sort();
297 fnv1a(format!("{schema}|{}", described.join("|")).as_bytes())
298}
299
300fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
305 let fields: BTreeMap<String, FieldSpec> = entry
306 .fields
307 .iter()
308 .map(|(field, spec)| (field.clone(), spec.spec.clone()))
309 .collect();
310 TableSchema {
311 table: name.to_string(),
312 revision: entry.revision as u32,
313 primary_key: entry.primary_key.clone(),
314 partition_by: entry.partition_by.clone(),
315 fields,
316 }
317}
318
319fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
322 let Some(declared) = catalog.triggers.as_ref() else {
323 return Vec::new();
324 };
325 declared
326 .iter()
327 .filter(|trigger| trigger.table == name)
328 .cloned()
329 .collect()
330}
331
332fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
338 for (name, table) in tables.iter_mut() {
339 table.set_triggers(triggers_of(catalog, name));
340 }
341}
342
343fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
347 entry
348 .options
349 .as_ref()
350 .and_then(&pick)
351 .or_else(|| catalog.defaults.as_ref().and_then(&pick))
352}
353
354fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
358 if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
359 table.set_image_dir(dir.join(image_dir));
360 }
361}
362
363fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
367 if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
368 table.set_sync(sync);
369 }
370}
371
372fn apply_validation(entry: &TableEntry, table: &mut Table) {
380 if let Some(validation) = entry.validation {
381 table.set_validation(validation);
382 }
383}
384
385fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
392 let marked: BTreeMap<String, String> = entry
393 .fields
394 .iter()
395 .filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
396 .collect();
397 if !marked.is_empty() {
398 table.set_deprecated_fields(marked);
399 }
400}
401
402fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
406 if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
407 table.set_link_parsing(link_parsing);
408 }
409}
410
411fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
415 named
416 .or_else(|| {
417 entry
418 .and_then(|entry| entry.options.as_ref())
419 .and_then(|options| options.compaction)
420 })
421 .or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
422 .unwrap_or(DEFAULT_COMPACTION)
423}
424
425fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
428 table.set_compaction(compaction);
429}
430
431fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
435 table.set_cipher(cipher.cloned());
436}
437
438fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
442 let all = open_table
443 .query(&QueryBuilder::new().spec())
444 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
445 let mut residual = 0usize;
446 for key in &all.keys {
447 let record = open_table
448 .get(key)
449 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
450 if record.is_some_and(|record| record.contains_key(field)) {
451 residual += 1;
452 }
453 }
454 if residual > 0 {
455 return Err(DatabaseError::FieldInUse {
456 table: name.to_string(),
457 field: field.to_string(),
458 records: residual,
459 });
460 }
461 Ok(())
462}
463
464fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
470 match error.downcast::<io::Error>() {
471 Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
472 Ok(io) => DatabaseError::Catalog(io.to_string()),
473 Err(other) => DatabaseError::Catalog(other.to_string()),
474 }
475}
476
477fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
480 layout.tables().join(table)
481}
482
483fn on_disk(path: &Path) -> bool {
486 fs::symlink_metadata(path).is_ok()
487}
488
489fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
497 let flat = layout.log_path(&schema.table);
498 if !is_partitioned(schema) {
499 return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
500 }
501 let dir = partition_dir_of(layout, &schema.table);
502 let present = partition_names_in(&dir);
503 if present.is_empty() && on_disk(&flat) {
505 return Ok(Vec::new());
506 }
507 let claimed = read_snapshot(index_path)?
508 .and_then(|snapshot| snapshot.header.parts)
509 .unwrap_or_default();
510 Ok(claimed
511 .into_iter()
512 .filter(|part| !present.contains(&part.part))
513 .map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
514 .collect())
515}
516
517fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
519 let flat = layout.log_path(&schema.table);
520 if !is_partitioned(schema) {
521 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
522 }
523 let dir = partition_dir_of(layout, &schema.table);
524 let present = partition_names_in(&dir);
525 if present.is_empty() {
526 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
527 }
528 present
529 .iter()
530 .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
531 .collect()
532}
533
534enum TableDrift {
537 Absent,
538 Schema { revision: u64 },
539 Indexes { unserved: Vec<String>, stale: Vec<String> },
540}
541
542struct DriftedTable {
543 name: String,
544 drift: TableDrift,
545}
546
547fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
550 let identities: HashSet<String> = present.iter().map(index_identity).collect();
551 let mut names: Vec<String> = expected
552 .iter()
553 .filter(|index| !identities.contains(&index_identity(index)))
554 .map(|index| index.name.clone())
555 .collect();
556 names.sort();
557 names
558}
559
560fn shape_of(schema: &TableSchema) -> TableSchema {
569 let mut bare = schema.clone();
570 for spec in bare.fields.values_mut() {
571 spec.index = None;
572 spec.unique = false;
573 spec.fulltext = false;
574 }
575 bare
576}
577
578fn quoted(names: &[String]) -> String {
579 names
580 .iter()
581 .map(|name| format!("\"{name}\""))
582 .collect::<Vec<_>>()
583 .join(", ")
584}
585
586fn stored_side_of(drift: &TableDrift) -> String {
587 match drift {
588 TableDrift::Absent => "declares no table at all".to_string(),
589 TableDrift::Schema { revision } => format!("declares revision {revision}"),
590 TableDrift::Indexes { unserved, stale } => {
591 let mut halves = Vec::new();
592 if !unserved.is_empty() {
593 halves.push(format!(
594 "declares indexes this handle does not serve: {}",
595 quoted(unserved)
596 ));
597 }
598 if !stale.is_empty() {
599 halves.push(format!(
600 "no longer declares indexes this handle still serves: {}",
601 quoted(stale)
602 ));
603 }
604 halves.join(", and ")
605 }
606 }
607}
608
609fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
610 let listed = missing
611 .iter()
612 .map(|path| format!("\"{}\"", path.display()))
613 .collect::<Vec<_>>()
614 .join(", ");
615 let one = missing.len() == 1;
616 let (its, is, it, file, log) = if one {
617 ("its log".to_string(), "is", "it", "file", "log")
618 } else {
619 (format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
620 };
621 HaysonSchemaError(format!(
622 "the catalog of \"{}\" declares table \"{name}\", but {its} {listed} {is} not on disk — this handle is readonly, so nothing here will write {it}; restore the {file}, or open the database for writing to have the {log} created",
623 dir.display()
624 ))
625}
626
627fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
638 let schema = table_schema_of(name, entry);
639 let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
640 if !missing.is_empty() {
641 return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
642 }
643 for path in present_logs_of(layout, &schema) {
644 fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
645 }
646 Ok(())
647}
648
649#[allow(clippy::too_many_arguments)]
657fn open_catalog_table(
658 layout: &Layout,
659 name: &str,
660 catalog: &Catalog,
661 entry: &TableEntry,
662 snapshot_every: u64,
663 cipher: Option<&FieldCipher>,
664 follow: Option<FollowPolicy>,
665 readonly: bool,
666 compaction: CatalogCompaction,
667 cache: CachePolicy,
668 max_image_bytes: Option<u64>,
669 password_hash: PasswordHashParams,
670 sealer: Option<Arc<Sealer>>,
671) -> Result<(Table, RestoreOutcome), DatabaseError> {
672 let schema = table_schema_of(name, entry);
673 let declared = expand_indexes(name, entry);
674 let hash = schema_fingerprint(&schema, &declared);
675 let (mut table, outcome) = Table::open_sealed(
676 schema,
677 layout.log_path(name),
678 layout.index_path(name),
679 &hash,
680 snapshot_every,
681 readonly,
682 &declared,
683 sealer,
684 )
685 .map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
686 apply_image_dir(layout.root(), catalog, entry, &mut table);
687 apply_link_parsing(catalog, entry, &mut table);
688 apply_sync(catalog, entry, &mut table);
689 apply_validation(entry, &mut table);
690 apply_deprecated(entry, &mut table);
691 apply_compaction(compaction, &mut table);
692 table.set_cache(cache);
693 table.set_max_image_bytes(max_image_bytes);
694 table.set_password_hash_params(password_hash);
695 apply_cipher(cipher, &mut table);
696 table.set_follow(follow);
697 table.set_triggers(triggers_of(catalog, name));
698 Ok((table, outcome))
699}
700
701fn first_table_log(layout: &Layout) -> Option<PathBuf> {
705 let has_log_suffix = |path: &Path| {
706 path.extension()
707 .is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
708 };
709 let entries_of = |dir: &Path| -> Vec<PathBuf> {
710 let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
711 let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
712 found.sort();
713 found
714 };
715
716 let top = entries_of(layout.tables());
717 if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
718 return Some(log.clone());
719 }
720 top.iter().filter(|path| path.is_dir()).find_map(|dir| {
721 entries_of(dir)
722 .into_iter()
723 .find(|path| path.is_file() && has_log_suffix(path))
724 })
725}
726
727fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
735 let Some(log) = first_table_log(layout) else {
736 return Ok(());
737 };
738 Err(HaysonSchemaError(format!(
739 "\"{}\" holds table data but no \"{}\" — refusing to open it as an empty database, which would write a fresh catalog next to \"{}\" and append records of a foreign schema onto it; restore \"{}\" from a backup, or move the logs aside to start over",
740 layout.root().display(),
741 layout.metadata_file(),
742 log.display(),
743 layout.metadata_file(),
744 )))
745}
746
747fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
748 read_catalog_with(path, None)
749}
750
751fn read_catalog_with(path: &Path, sealer: Option<&Sealer>) -> Result<Option<Catalog>, DatabaseError> {
752 let raw = match read_catalog_text(path, sealer) {
753 Ok(Some(raw)) => raw,
754 Ok(None) => return Ok(None),
755 Err(ReadCatalogError::Io(error)) => return Err(error.into()),
756 Err(
757 ReadCatalogError::Cipher(error)
758 | ReadCatalogError::Encrypted(error)
759 | ReadCatalogError::NotEncrypted(error),
760 ) => return Err(DatabaseError::Cipher(Box::new(error))),
761 };
762 let value: Value = serde_json::from_str(&raw)
763 .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
764 Ok(Some(parse_catalog(&value)?))
765}
766
767pub(super) struct HeldLock(pub(super) Option<DirLock>);
778
779impl HeldLock {
780 fn into_held(mut self) -> Option<DirLock> {
781 self.0.take()
782 }
783}
784
785impl Drop for HeldLock {
786 fn drop(&mut self) {
787 if let Some(lock) = self.0.as_mut() {
788 lock.release();
789 }
790 }
791}
792
793pub struct Database {
797 layout: Layout,
798 lock: Option<DirLock>,
799 catalog: Catalog,
800 tables: HashMap<String, Table>,
801 restore_outcomes: HashMap<String, RestoreOutcome>,
802 readonly: bool,
803 snapshot_every: u64,
804 compaction: Option<CatalogCompaction>,
807 cipher: Option<FieldCipher>,
808 follow: Option<FollowPolicy>,
809 cache: CachePolicy,
812 max_image_bytes: Option<u64>,
813 password_hash: PasswordHashParams,
814 sealer: Option<Arc<Sealer>>,
816}
817
818impl Database {
819 pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
820 Database::open_with(dir, OpenOptions::default())
821 }
822
823 pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
837 let sealer = options.encryption.as_ref().map(Sealer::new).transpose()?.map(Arc::new);
838 let (layout, lock) = match options.lock {
839 LockMode::Enabled(lock) if !options.readonly => {
840 Database::hold_for_open(dir, &options.layout, lock, sealer.as_deref())?
841 }
842 LockMode::Enabled(_) | LockMode::Skip => (
843 Database::resolve_layout_sealed(dir, &options.layout, sealer.as_deref())?,
844 HeldLock(None),
845 ),
846 };
847
848 if !options.readonly {
849 for directory in layout.directories() {
850 fs::create_dir_all(directory)?;
851 }
852 }
853
854 Database::open_locked(layout, options, lock, sealer)
855 }
856
857 fn hold_for_open(
878 dir: &Path,
879 options: &LayoutOptions,
880 lock: LockOptions,
881 sealer: Option<&Sealer>,
882 ) -> Result<(Layout, HeldLock), DatabaseError> {
883 const SETTLE_ATTEMPTS: usize = 3;
884 let mut probed = Database::resolve_layout_sealed(dir, options, sealer)?;
885 for _ in 0..SETTLE_ATTEMPTS {
886 fs::create_dir_all(probed.runtime())?;
887 let held = HeldLock(Some(DirLock::acquire(probed.lock_path(), lock)?));
888 let settled = Database::resolve_layout_sealed(dir, options, sealer)?;
889 if settled.kind() == probed.kind() {
890 return Ok((settled, held));
891 }
892 drop(held);
894 if probed.kind() == LayoutKind::Nested {
895 let _ = fs::remove_dir(probed.runtime());
896 }
897 probed = settled;
898 }
899 Err(HaysonSchemaError(format!(
900 "\"{}\" changed arrangement {SETTLE_ATTEMPTS} times while its writer lock was being taken — something keeps moving it without the lock; open it again once the moves are over",
901 dir.display()
902 ))
903 .into())
904 }
905
906 pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
920 Database::resolve_layout_sealed(dir, options, None)
921 }
922
923 pub fn resolve_layout_sealed(
925 dir: &Path,
926 options: &LayoutOptions,
927 sealer: Option<&Sealer>,
928 ) -> Result<Layout, DatabaseError> {
929 let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
930 let stored = stored_kind_with(&dir.join(metadata_file), sealer);
931 let kind = resolve_kind(dir, stored, options.kind)?;
932 Ok(Layout::new(dir, kind, metadata_file))
933 }
934
935 fn open_locked(
936 layout: Layout,
937 options: OpenOptions,
938 lock: HeldLock,
939 sealer: Option<Arc<Sealer>>,
940 ) -> Result<Database, DatabaseError> {
941 let found = read_catalog_with(layout.metadata_path(), sealer.as_deref())?;
942 let catalog = match found {
943 Some(catalog) => catalog,
944 None => {
945 refuse_orphan_logs(&layout)?;
946 let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
947 if !options.readonly {
948 write_catalog_sealed(layout.metadata_path(), &catalog, sealer.as_deref())
949 .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
950 }
951 catalog
952 }
953 };
954
955 let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
956 options.password_hash.validate()?;
957 let follow = options.follow.resolve(options.readonly);
958
959 let mut tables = HashMap::new();
960 let mut restore_outcomes = HashMap::new();
961 for (name, entry) in &catalog.tables {
962 if options.readonly {
963 refuse_unreadable_logs(&layout, name, entry)?;
964 }
965 let (table, outcome) = open_catalog_table(
966 &layout,
967 name,
968 &catalog,
969 entry,
970 options.snapshot_every,
971 cipher.as_ref(),
972 follow,
973 options.readonly,
974 compaction_of(options.compaction, &catalog, Some(entry)),
975 options.cache,
976 options.max_image_bytes,
977 options.password_hash,
978 sealer.clone(),
979 )?;
980 tables.insert(name.clone(), table);
981 restore_outcomes.insert(name.clone(), outcome);
982 }
983
984 if !options.readonly {
985 recover_incomplete_sealed(&layout.journal_path(), &mut tables, sealer.as_deref())?;
986 }
987
988 sync_table_triggers(&catalog, &mut tables);
989
990 Ok(Database {
991 layout,
992 lock: lock.into_held(),
993 catalog,
994 tables,
995 restore_outcomes,
996 readonly: options.readonly,
997 snapshot_every: options.snapshot_every,
998 compaction: options.compaction,
999 cipher,
1000 follow,
1001 cache: options.cache,
1002 max_image_bytes: options.max_image_bytes,
1003 password_hash: options.password_hash,
1004 sealer,
1005 })
1006 }
1007
1008 pub fn dir(&self) -> &Path {
1009 self.layout.root()
1010 }
1011
1012 pub fn layout(&self) -> &Layout {
1015 &self.layout
1016 }
1017
1018 pub fn follow(&self) -> Option<FollowPolicy> {
1021 self.follow
1022 }
1023
1024 pub fn table_names(&self) -> Vec<&str> {
1025 let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
1026 names.sort_unstable();
1027 names
1028 }
1029
1030 pub fn table(&self, name: &str) -> Option<&Table> {
1031 self.tables.get(name)
1032 }
1033
1034 pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
1040 let _ = self.renew_lock();
1041 self.tables.get_mut(name)
1042 }
1043
1044 fn assert_relations_hold(&self, table: &str, values: &Value) -> Result<(), DatabaseError> {
1062 for relation in relations_touching(&self.catalog, Some(table)) {
1063 if relation.from.table != table || relation.enforce != Some(true) {
1064 continue;
1065 }
1066 let Some(value) = reference_of(values.get(&relation.from.field)) else {
1067 continue;
1068 };
1069 let Some(to_table) = self.tables.get(&relation.to.table) else {
1070 continue;
1071 };
1072 if !to_table.contains_value(&relation.to.field, &value) {
1073 return Err(DatabaseError::Table(
1074 table.to_string(),
1075 TableError::Relation(HaysonRelationViolationError {
1076 relation: relation.name.clone(),
1077 table: table.to_string(),
1078 field: relation.from.field.clone(),
1079 value: scalar_as_text(&value),
1080 }),
1081 ));
1082 }
1083 }
1084 Ok(())
1085 }
1086
1087 pub fn put_checked(&mut self, table: &str, document: &Value, force: bool) -> Result<HaysonRecord, DatabaseError> {
1092 self.assert_relations_hold(table, document)?;
1093 let root = self.layout.root().display().to_string();
1094 let handle = self
1095 .table_mut(table)
1096 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1097 handle
1098 .put(document, force)
1099 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1100 }
1101
1102 pub fn patch_checked(
1105 &mut self,
1106 table: &str,
1107 key: &str,
1108 changes: &Value,
1109 force: bool,
1110 ) -> Result<HaysonRecord, DatabaseError> {
1111 self.assert_relations_hold(table, changes)?;
1112 let root = self.layout.root().display().to_string();
1113 let handle = self
1114 .table_mut(table)
1115 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1116 handle
1117 .patch(key, changes, force)
1118 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1119 }
1120
1121 pub fn stage_put_checked(
1126 &mut self,
1127 table: &str,
1128 document: &Value,
1129 force: bool,
1130 ) -> Result<HaysonRecord, DatabaseError> {
1131 self.assert_relations_hold(table, document)?;
1132 let root = self.layout.root().display().to_string();
1133 let handle = self
1134 .table_mut(table)
1135 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1136 handle
1137 .stage_put(document, force)
1138 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1139 }
1140
1141 pub fn stage_patch_checked(
1144 &mut self,
1145 table: &str,
1146 key: &str,
1147 changes: &Value,
1148 force: bool,
1149 ) -> Result<HaysonRecord, DatabaseError> {
1150 self.assert_relations_hold(table, changes)?;
1151 let root = self.layout.root().display().to_string();
1152 let handle = self
1153 .table_mut(table)
1154 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1155 handle
1156 .stage_patch(key, changes, force)
1157 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1158 }
1159
1160 pub fn delete_checked(&mut self, table: &str, key: &str, force: bool) -> Result<bool, DatabaseError> {
1175 if self.readonly {
1176 return Err(DatabaseError::ReadOnly {
1177 dir: self.layout.root().to_path_buf(),
1178 operation: "delete".to_string(),
1179 });
1180 }
1181 let root = self.layout.root().display().to_string();
1182 if self.tables.get(table).is_none() {
1183 return Err(DatabaseError::Catalog(format!(
1184 "table \"{table}\" does not exist in \"{root}\""
1185 )));
1186 }
1187 let incoming = relations_touching(&self.catalog, Some(table))
1188 .into_iter()
1189 .any(|relation| relation.to.table == table);
1190 if !incoming {
1191 self.renew_lock()?;
1192 let handle = self.tables.get_mut(table).expect("checked above");
1193 return handle
1194 .delete(key, force)
1195 .map_err(|error| DatabaseError::Table(table.to_string(), error));
1196 }
1197
1198 let mut unit = UnitOfWork::new();
1199 let mut visited: HashSet<(String, String)> = HashSet::new();
1200 let existed = self.plan_delete_cascade(table, key, &mut visited, &mut unit)?;
1201 if !existed {
1202 return Ok(false);
1203 }
1204 self.commit(&mut unit)?;
1205 Ok(true)
1206 }
1207
1208 fn plan_delete_cascade(
1218 &self,
1219 table: &str,
1220 key: &str,
1221 visited: &mut HashSet<(String, String)>,
1222 unit: &mut UnitOfWork,
1223 ) -> Result<bool, DatabaseError> {
1224 if !visited.insert((table.to_string(), key.to_string())) {
1225 return Ok(true);
1226 }
1227 let handle = self
1228 .tables
1229 .get(table)
1230 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist")))?;
1231 let Some(record) = handle
1232 .get(key)
1233 .map_err(|error| DatabaseError::Table(table.to_string(), error))?
1234 else {
1235 return Ok(false);
1236 };
1237
1238 for relation in relations_touching(&self.catalog, Some(table)) {
1239 if relation.to.table != table {
1240 continue;
1241 }
1242 let action = relation.on_delete.unwrap_or(RelationAction::NoAction);
1243 if matches!(action, RelationAction::NoAction) {
1244 continue;
1245 }
1246 let Some(value) = reference_of(record.get(&relation.to.field)) else {
1247 continue;
1248 };
1249 let Some(from_table) = self.tables.get(&relation.from.table) else {
1250 continue;
1251 };
1252 let referencing = from_table
1253 .keys_referencing(&relation.from.field, &value)
1254 .map_err(|error| DatabaseError::Table(relation.from.table.clone(), error))?;
1255 if referencing.is_empty() {
1256 continue;
1257 }
1258
1259 match action {
1260 RelationAction::Restrict => {
1261 return Err(DatabaseError::Table(
1262 table.to_string(),
1263 TableError::RelationRestrict(HaysonRelationRestrictError {
1264 relation: relation.name.clone(),
1265 table: table.to_string(),
1266 key: key.to_string(),
1267 referencing_table: relation.from.table.clone(),
1268 referencing_field: relation.from.field.clone(),
1269 referencing: referencing.len(),
1270 }),
1271 ));
1272 }
1273 RelationAction::Cascade => {
1274 for referencing_key in &referencing {
1275 self.plan_delete_cascade(&relation.from.table, referencing_key, visited, unit)?;
1276 }
1277 }
1278 RelationAction::SetNull => {
1279 for referencing_key in &referencing {
1280 unit.patch(
1281 relation.from.table.as_str(),
1282 referencing_key.clone(),
1283 Value::Object(serde_json::Map::from_iter([(relation.from.field.clone(), Value::Null)])),
1284 );
1285 }
1286 }
1287 RelationAction::NoAction => unreachable!("filtered above"),
1288 }
1289 }
1290
1291 unit.delete(table, key.to_string());
1292 Ok(true)
1293 }
1294
1295 pub fn tick(&self) -> Result<(), DatabaseError> {
1300 self.renew_lock()
1301 }
1302
1303 fn renew_lock(&self) -> Result<(), DatabaseError> {
1304 match &self.lock {
1305 Some(lock) => Ok(lock.tick()?),
1306 None => Ok(()),
1307 }
1308 }
1309
1310 pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
1313 self.restore_outcomes.get(name).copied()
1314 }
1315
1316 pub fn catalog_unchecked(&self) -> &Catalog {
1320 &self.catalog
1321 }
1322
1323 pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
1331 let drifted = self.drifted_tables();
1332 if drifted.is_empty() {
1333 return Ok(&self.catalog);
1334 }
1335 Err(DatabaseError::Schema(self.drift_error(&drifted)))
1336 }
1337
1338 pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
1350 if let Some(drift) = self.drift_of(name) {
1351 let drifted = [DriftedTable {
1352 name: name.to_string(),
1353 drift,
1354 }];
1355 return Err(DatabaseError::Schema(self.drift_error(&drifted)));
1356 }
1357 let table = self.tables.get(name).ok_or_else(|| {
1358 DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
1359 })?;
1360 Ok(describe_table_document(table))
1361 }
1362
1363 fn drift_of(&self, name: &str) -> Option<TableDrift> {
1364 let table = self.tables.get(name)?;
1365 let Some(entry) = self.catalog.tables.get(name) else {
1366 return Some(TableDrift::Absent);
1367 };
1368 if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
1369 return Some(TableDrift::Schema {
1370 revision: entry.revision,
1371 });
1372 }
1373 let declared = expand_indexes(name, entry);
1374 let served = table.declared_indexes();
1375 let unserved = index_names_missing_from(&declared, served);
1376 let stale = index_names_missing_from(served, &declared);
1377 if unserved.is_empty() && stale.is_empty() {
1378 return None;
1379 }
1380 Some(TableDrift::Indexes { unserved, stale })
1381 }
1382
1383 fn drifted_tables(&self) -> Vec<DriftedTable> {
1384 let mut names: Vec<&String> = self.tables.keys().collect();
1385 names.sort();
1386 names
1387 .into_iter()
1388 .filter_map(|name| {
1389 self.drift_of(name).map(|drift| DriftedTable {
1390 name: name.clone(),
1391 drift,
1392 })
1393 })
1394 .collect()
1395 }
1396
1397 fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
1398 let detail = drifted
1399 .iter()
1400 .map(|DriftedTable { name, drift }| {
1401 let revision = self.tables.get(name).map(|table| table.schema().revision);
1402 let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
1403 format!(
1404 "\"{name}\" is open here at {open} while the catalog {}",
1405 stored_side_of(drift)
1406 )
1407 })
1408 .collect::<Vec<_>>()
1409 .join("; ");
1410 HaysonSchemaError(format!(
1411 "the catalog of \"{}\" moved on under this handle: {detail} — an open table keeps the schema and the index set it was opened with, so describing it from the catalog would describe a table this handle does not serve; reopen the database to work with the catalog on disk",
1412 self.layout.root().display()
1413 ))
1414 }
1415
1416 pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
1420 relations_touching(&self.catalog, table)
1421 }
1422
1423 pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
1426 let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
1427 check_integrity(&self.catalog, &refs)
1428 }
1429
1430 pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
1440 let mut changed = false;
1441 if let Some(catalog) = read_catalog_with(self.layout.metadata_path(), self.sealer.as_deref())? {
1442 if catalog != self.catalog {
1443 self.catalog = catalog;
1444 changed = true;
1445 }
1446 }
1447
1448 let missing: Vec<String> = self
1449 .catalog
1450 .tables
1451 .keys()
1452 .filter(|name| !self.tables.contains_key(*name))
1453 .cloned()
1454 .collect();
1455 for name in missing {
1456 let Some(entry) = self.catalog.tables.get(&name) else {
1457 continue;
1458 };
1459 let (table, outcome) = open_catalog_table(
1460 &self.layout,
1461 &name,
1462 &self.catalog,
1463 entry,
1464 self.snapshot_every,
1465 self.cipher.as_ref(),
1466 self.follow,
1467 self.readonly,
1468 compaction_of(self.compaction, &self.catalog, Some(entry)),
1469 self.cache,
1470 self.max_image_bytes,
1471 self.password_hash,
1472 self.sealer.clone(),
1473 )?;
1474 self.tables.insert(name.clone(), table);
1475 self.restore_outcomes.insert(name, outcome);
1476 changed = true;
1477 }
1478
1479 for (name, table) in self.tables.iter_mut() {
1480 let outcome = table
1481 .refresh()
1482 .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
1483 changed |= outcome.changed();
1484 }
1485
1486 sync_table_triggers(&self.catalog, &mut self.tables);
1488 Ok(changed)
1489 }
1490
1491 pub fn unit_of_work(&mut self) -> UnitOfWork {
1494 UnitOfWork::new()
1495 }
1496
1497 pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
1504 let prepared = self.prepare_commit(unit)?;
1505 unit.journal_sealed(&prepared, &self.layout.journal_path(), self.sealer.as_deref())?;
1506 self.apply_commit(prepared)
1507 }
1508
1509 pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
1518 if self.readonly {
1519 return Err(DatabaseError::ReadOnly {
1520 dir: self.layout.root().to_path_buf(),
1521 operation: "commit".to_string(),
1522 });
1523 }
1524 self.renew_lock()?;
1525 Ok(unit.prepare(&mut self.tables)?)
1526 }
1527
1528 pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
1532 let journal_path = self.layout.journal_path();
1533 Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
1534 }
1535
1536 fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
1540 if self.readonly {
1541 return Err(DatabaseError::ReadOnly {
1542 dir: self.layout.root().to_path_buf(),
1543 operation: operation.to_string(),
1544 });
1545 }
1546 self.renew_lock()
1547 }
1548
1549 fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
1550 write_catalog_sealed(self.layout.metadata_path(), catalog, self.sealer.as_deref()).map_err(catalog_write_error)
1551 }
1552
1553 pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
1568 self.assert_writable("createTable")?;
1569 let schema = parse_schema(&schema_spec)?;
1570 if schema.table != name {
1571 return Err(DatabaseError::Schema(HaysonSchemaError(format!(
1572 "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
1573 schema.table
1574 ))));
1575 }
1576 let now = now_ms();
1577 let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
1578 self.write_catalog_to_disk(&next_catalog)?;
1579
1580 let next_entry = &next_catalog.tables[name];
1581 let (table, outcome) = open_catalog_table(
1582 &self.layout,
1583 name,
1584 &next_catalog,
1585 next_entry,
1586 self.snapshot_every,
1587 self.cipher.as_ref(),
1588 self.follow,
1589 self.readonly,
1590 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1591 self.cache,
1592 self.max_image_bytes,
1593 self.password_hash,
1594 self.sealer.clone(),
1595 )?;
1596
1597 self.catalog = next_catalog;
1598 self.tables.insert(name.to_string(), table);
1599 self.restore_outcomes.insert(name.to_string(), outcome);
1600 Ok(())
1601 }
1602
1603 pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
1616 self.assert_writable("dropField")?;
1617 self.flush_table(table)?;
1618 let existing = self.tables.get(table).ok_or_else(|| {
1619 DatabaseError::Catalog(format!(
1620 "table \"{table}\" does not exist in \"{}\"",
1621 self.layout.root().display()
1622 ))
1623 })?;
1624
1625 let now = now_ms();
1626 let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
1627
1628 if !force {
1629 assert_no_residual_values(table, existing, field)?;
1630 }
1631
1632 self.write_catalog_to_disk(&next_catalog)?;
1633
1634 let (mut table_handle, _) = open_catalog_table(
1635 &self.layout,
1636 table,
1637 &next_catalog,
1638 &next_entry,
1639 self.snapshot_every,
1640 self.cipher.as_ref(),
1641 self.follow,
1642 self.readonly,
1643 compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
1644 self.cache,
1645 self.max_image_bytes,
1646 self.password_hash,
1647 self.sealer.clone(),
1648 )?;
1649 table_handle
1651 .snapshot()
1652 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1653
1654 self.catalog = next_catalog;
1655 self.tables.insert(table.to_string(), table_handle);
1656 Ok(())
1657 }
1658
1659 pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
1668 self.assert_writable("declareIndex")?;
1669 self.flush_table(table)?;
1670 let existing = self.tables.get(table).ok_or_else(|| {
1671 DatabaseError::Catalog(format!(
1672 "table \"{table}\" does not exist in \"{}\"",
1673 self.layout.root().display()
1674 ))
1675 })?;
1676
1677 let now = now_ms();
1678 let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
1679 self.write_catalog_to_disk(&next_catalog)?;
1680
1681 let next_entry = &next_catalog.tables[table];
1682 let (mut table_handle, _) = open_catalog_table(
1683 &self.layout,
1684 table,
1685 &next_catalog,
1686 next_entry,
1687 self.snapshot_every,
1688 self.cipher.as_ref(),
1689 self.follow,
1690 self.readonly,
1691 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1692 self.cache,
1693 self.max_image_bytes,
1694 self.password_hash,
1695 self.sealer.clone(),
1696 )?;
1697 table_handle
1699 .snapshot()
1700 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1701
1702 self.catalog = next_catalog;
1703 self.tables.insert(table.to_string(), table_handle);
1704 Ok(())
1705 }
1706
1707 pub fn declare_relation(&mut self, relation_spec: Value, force: bool) -> Result<(), DatabaseError> {
1717 self.assert_writable("declareRelation")?;
1718
1719 let from_table = relation_spec
1720 .get("from")
1721 .and_then(|end| end.get("table"))
1722 .and_then(Value::as_str)
1723 .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"from.table\"".to_string()))?
1724 .to_string();
1725 let to_table = relation_spec
1726 .get("to")
1727 .and_then(|end| end.get("table"))
1728 .and_then(Value::as_str)
1729 .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"to.table\"".to_string()))?
1730 .to_string();
1731
1732 self.flush_table(&from_table)?;
1733 self.flush_table(&to_table)?;
1734 let from_handle = self.tables.get(&from_table).ok_or_else(|| {
1735 DatabaseError::Catalog(format!(
1736 "table \"{from_table}\" does not exist in \"{}\"",
1737 self.layout.root().display()
1738 ))
1739 })?;
1740 let to_handle = self.tables.get(&to_table).ok_or_else(|| {
1741 DatabaseError::Catalog(format!(
1742 "table \"{to_table}\" does not exist in \"{}\"",
1743 self.layout.root().display()
1744 ))
1745 })?;
1746
1747 let now = now_ms();
1748 let next_catalog = declare_relation_in(&self.catalog, &relation_spec, from_handle, to_handle, now, force)?;
1749 self.write_catalog_to_disk(&next_catalog)?;
1750 self.catalog = next_catalog;
1751 Ok(())
1752 }
1753
1754 pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
1760 self.assert_writable("declareTrigger")?;
1761
1762 let now = now_ms();
1763 let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
1764 self.write_catalog_to_disk(&next_catalog)?;
1765
1766 self.catalog = next_catalog;
1767 sync_table_triggers(&self.catalog, &mut self.tables);
1768 Ok(())
1769 }
1770
1771 pub fn flush(&mut self) -> Result<(), DatabaseError> {
1775 for (name, table) in &mut self.tables {
1776 table
1777 .flush()
1778 .map_err(|error| DatabaseError::Table(name.clone(), error))?;
1779 }
1780 Ok(())
1781 }
1782
1783 fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
1787 let Some(handle) = self.tables.get_mut(table) else {
1788 return Ok(());
1789 };
1790 handle
1791 .flush()
1792 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1793 }
1794
1795 pub fn close(mut self) -> io::Result<()> {
1805 let flushed = if self.readonly {
1806 Ok(())
1807 } else {
1808 self.flush_and_snapshot_all()
1809 };
1810 if let Some(mut lock) = self.lock.take() {
1811 lock.release();
1812 }
1813 flushed
1814 }
1815
1816 fn flush_and_snapshot_all(&mut self) -> io::Result<()> {
1817 let mut first_refusal = None;
1818 for table in self.tables.values_mut() {
1819 let outcome = table.flush().map_err(io::Error::from).and_then(|()| {
1821 if table.needs_snapshot() {
1822 table.snapshot()
1823 } else {
1824 Ok(())
1825 }
1826 });
1827 if let Err(error) = outcome {
1828 first_refusal.get_or_insert(error);
1829 }
1830 }
1831 match first_refusal {
1832 Some(error) => Err(error),
1833 None => Ok(()),
1834 }
1835 }
1836}
1837
1838#[cfg(test)]
1839mod tests {
1840 use serde_json::json;
1841 use tempfile::tempdir;
1842
1843 use super::*;
1844 use crate::catalog::create_table;
1845 use crate::catalog::types::{CatalogIndexKind, IndexEntry};
1846 use crate::checkpoint::write_snapshot;
1847 use crate::registry::table_registry::TableRegistry;
1848 use crate::schema::parse_schema;
1849 use crate::storage::{serialize_entry, LogMeta, LogOp};
1850 use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
1851 use crate::txn::JOURNAL_FILE;
1852
1853 const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
1854
1855 fn flat(dir: &Path) -> Layout {
1856 Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
1857 }
1858
1859 fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
1860 flat(dir).log_path(table)
1861 }
1862
1863 #[test]
1867 fn a_database_and_its_tables_can_be_shared_across_threads() {
1868 fn shareable<T: Send + Sync>() {}
1869 shareable::<Database>();
1870 shareable::<Table>();
1871 shareable::<crate::registry::fulltext_index::FulltextIndex>();
1872 }
1873
1874 fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
1875 flat(dir).index_path(table)
1876 }
1877
1878 fn users_schema() -> TableSchema {
1879 parse_schema(&json!({
1880 "schemaVersion": 1,
1881 "table": "users",
1882 "primaryKey": "id",
1883 "fields": {
1884 "id": { "type": "string" },
1885 "name": { "type": "string", "required": true },
1886 },
1887 }))
1888 .unwrap()
1889 }
1890
1891 fn orders_schema() -> TableSchema {
1892 parse_schema(&json!({
1893 "schemaVersion": 1,
1894 "table": "orders",
1895 "primaryKey": "id",
1896 "fields": {
1897 "id": { "type": "string" },
1898 "userId": { "type": "string", "required": true },
1899 },
1900 }))
1901 .unwrap()
1902 }
1903
1904 fn catalog_of(schemas: &[TableSchema]) -> Catalog {
1905 let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
1906 for schema in schemas {
1907 catalog = create_table(&catalog, &schema.table, schema, 1_000);
1908 }
1909 catalog
1910 }
1911
1912 fn write_catalog_to(dir: &Path, catalog: &Catalog) {
1913 write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
1914 }
1915
1916 #[test]
1917 fn open_builds_a_database_with_every_declared_table_available() {
1918 let dir = tempdir().unwrap();
1919 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1920
1921 let mut db = Database::open(dir.path()).unwrap();
1922 assert_eq!(db.table_names(), vec!["orders", "users"]);
1923
1924 db.table_mut("users")
1925 .unwrap()
1926 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1927 .unwrap();
1928 assert_eq!(
1929 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1930 Some(&json!("Ania"))
1931 );
1932 db.close().unwrap();
1933 }
1934
1935 #[test]
1936 fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
1937 let dir = tempdir().unwrap();
1938 let catalog = catalog_of(&[users_schema()]);
1939 write_catalog_to(dir.path(), &catalog);
1940
1941 let schema = table_schema_of("users", &catalog.tables["users"]);
1942 let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
1943 let log_path = log_path(dir.path(), "users");
1944 let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
1945 log.append(&[serialize_entry(
1946 &LogMeta {
1947 key: "u1".into(),
1948 op: LogOp::Put,
1949 seq: 1,
1950 ts: 1,
1951 },
1952 Some(&{
1953 let mut record = crate::schema::HaysonRecord::new();
1954 record.insert("id".into(), json!("u1"));
1955 record.insert("name".into(), json!("Ania"));
1956 record
1957 }),
1958 )])
1959 .unwrap();
1960 let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
1961 write_snapshot(&index_path(dir.path(), "users"), ®istry, "users", &hash, 0, &log).unwrap();
1962
1963 let db = Database::open(dir.path()).unwrap();
1964 assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
1965 assert_eq!(
1966 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1967 Some(&json!("Ania"))
1968 );
1969 db.close().unwrap();
1970 }
1971
1972 #[test]
1973 fn open_finishes_an_unfinished_journal_before_returning() {
1974 let dir = tempdir().unwrap();
1975 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1976
1977 {
1978 let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
1979 let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
1980 let user_entries = users
1981 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
1982 .unwrap();
1983 let order_entries = orders
1984 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
1985 .unwrap();
1986
1987 let mut entries = Vec::new();
1988 for entry in &user_entries {
1989 entries.push(JournalEntry {
1990 t: "users".into(),
1991 s: entry.seq(),
1992 l: entry.line().to_string(),
1993 });
1994 }
1995 for entry in &order_entries {
1996 entries.push(JournalEntry {
1997 t: "orders".into(),
1998 s: entry.seq(),
1999 l: entry.line().to_string(),
2000 });
2001 }
2002 write_journal(
2003 &dir.path().join(JOURNAL_FILE),
2004 &Journal {
2005 header: JournalHeader {
2006 journal_version: JOURNAL_VERSION,
2007 kind: "unit".into(),
2008 id: "crash".into(),
2009 started_at: 0,
2010 tables: vec!["users".into(), "orders".into()],
2011 },
2012 entries,
2013 },
2014 )
2015 .unwrap();
2016 }
2018
2019 assert!(dir.path().join(JOURNAL_FILE).exists());
2020 let db = Database::open(dir.path()).unwrap();
2021 assert!(!dir.path().join(JOURNAL_FILE).exists());
2022 assert_eq!(
2023 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2024 Some(&json!("Ania"))
2025 );
2026 assert_eq!(
2027 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2028 Some(&json!("u1"))
2029 );
2030 db.close().unwrap();
2031 }
2032
2033 #[test]
2034 fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
2035 let dir = tempdir().unwrap();
2036 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2037
2038 let _first = Database::open(dir.path()).unwrap();
2039 let second = Database::open_with(
2040 dir.path(),
2041 OpenOptions {
2042 readonly: false,
2043 lock: LockMode::Enabled(LockOptions {
2044 timeout_ms: Some(0),
2045 ..Default::default()
2046 }),
2047 ..Default::default()
2048 },
2049 );
2050 assert!(matches!(second, Err(DatabaseError::Lock(_))));
2051 }
2052
2053 fn lock_path_of(dir: &Path) -> std::path::PathBuf {
2054 dir.join(crate::lock::LOCK_FILE)
2055 }
2056
2057 fn open_refusing(dir: &Path) -> DatabaseError {
2058 Database::open_with(
2059 dir,
2060 OpenOptions {
2061 lock: LockMode::Enabled(LockOptions {
2062 timeout_ms: Some(0),
2063 ..Default::default()
2064 }),
2065 ..Default::default()
2066 },
2067 )
2068 .map(|_| ())
2069 .expect_err("this directory must not open")
2070 }
2071
2072 #[test]
2073 fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
2074 let dir = tempdir().unwrap();
2075 let mut db = Database::open(dir.path()).unwrap();
2076 db.create_table(
2077 "users",
2078 json!({
2079 "schemaVersion": 1,
2080 "table": "users",
2081 "primaryKey": "id",
2082 "fields": {
2083 "id": { "type": "string" },
2084 "name": { "type": "string", "required": true },
2085 },
2086 }),
2087 )
2088 .unwrap();
2089 db.table_mut("users")
2090 .unwrap()
2091 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2092 .unwrap();
2093 db.close().unwrap();
2094
2095 let log = fs::read(dir.path().join("users.jsonl")).unwrap();
2096 fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2097
2098 let refusal = open_refusing(dir.path()).to_string();
2099 assert!(
2100 refusal.contains("users.jsonl"),
2101 "the refusal has to name the log it found: {refusal}"
2102 );
2103 assert!(
2104 refusal.contains(MANIFEST_FILE),
2105 "the refusal has to name the catalog: {refusal}"
2106 );
2107
2108 assert!(
2110 !dir.path().join(MANIFEST_FILE).exists(),
2111 "a refused open must not leave a catalog"
2112 );
2113 assert_eq!(
2114 fs::read(dir.path().join("users.jsonl")).unwrap(),
2115 log,
2116 "the log must be untouched"
2117 );
2118 }
2119
2120 #[test]
2121 fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
2122 let empty = tempdir().unwrap();
2123 assert!(
2124 Database::open(empty.path()).is_ok(),
2125 "an empty directory is still a new database"
2126 );
2127
2128 let littered = tempdir().unwrap();
2129 fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
2130 fs::create_dir(littered.path().join("users-backup")).unwrap();
2131 assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
2132 }
2133
2134 #[test]
2135 fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
2136 let dir = tempdir().unwrap();
2137 fs::create_dir_all(dir.path().join("notes")).unwrap();
2138 fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
2139
2140 let refusal = open_refusing(dir.path()).to_string();
2141 assert!(
2142 refusal.contains("note.jsonl"),
2143 "a partitioned log still counts as data: {refusal}"
2144 );
2145 }
2146
2147 #[test]
2151 fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
2152 for (case, document) in [
2153 ("not JSON at all", "{ this is not json".to_string()),
2154 (
2155 "no formatVersion",
2156 json!({ "catalogVersion": 2, "tables": {} }).to_string(),
2157 ),
2158 (
2159 "the format-1 marker",
2160 json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
2161 ),
2162 (
2163 "a formatVersion from the future",
2164 json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
2165 ),
2166 ] {
2167 let dir = tempdir().unwrap();
2168 std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
2169
2170 open_refusing(dir.path());
2171
2172 assert!(
2173 !lock_path_of(dir.path()).exists(),
2174 "{case}: the refused open left its lock file behind"
2175 );
2176 }
2177 }
2178
2179 #[test]
2182 fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
2183 let dir = tempdir().unwrap();
2184 std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
2185
2186 open_refusing(dir.path());
2187
2188 std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2189 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2190
2191 let mut db = Database::open(dir.path()).unwrap();
2192 db.table_mut("users")
2193 .unwrap()
2194 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2195 .unwrap();
2196 db.close().unwrap();
2197 }
2198
2199 #[test]
2202 fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
2203 let dir = tempdir().unwrap();
2204 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2205
2206 let error = Database::open_with(
2207 dir.path(),
2208 OpenOptions {
2209 cypher: Some(CipherOptions {
2210 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
2211 key_id: None,
2212 on_missing_key: None,
2213 }),
2214 ..OpenOptions::default()
2215 },
2216 )
2217 .map(|_| ())
2218 .unwrap_err();
2219
2220 assert!(matches!(error, DatabaseError::Cipher(_)));
2221 assert!(!lock_path_of(dir.path()).exists());
2222 Database::open(dir.path()).unwrap().close().unwrap();
2223 }
2224
2225 #[test]
2228 fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
2229 let dir = tempdir().unwrap();
2230 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2231
2232 let held = Database::open(dir.path()).unwrap();
2233 assert!(lock_path_of(dir.path()).exists());
2234 assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
2235
2236 held.close().unwrap();
2237 assert!(!lock_path_of(dir.path()).exists());
2238 }
2239
2240 fn open_skipping_the_lock(dir: &Path) -> Database {
2243 Database::open_with(
2244 dir,
2245 OpenOptions {
2246 lock: LockMode::Skip,
2247 ..Default::default()
2248 },
2249 )
2250 .unwrap()
2251 }
2252
2253 fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
2254 let catalog = read_catalog(&dir.join(MANIFEST_FILE))
2255 .unwrap()
2256 .expect("a catalog on disk");
2257 let mut names: Vec<String> = catalog.tables[table]
2258 .indexes
2259 .iter()
2260 .flatten()
2261 .map(|index| index.name.clone())
2262 .collect();
2263 names.sort();
2264 names
2265 }
2266
2267 fn two_tables_with_records(dir: &Path) {
2271 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
2272 let mut db = Database::open(dir).unwrap();
2273 db.table_mut("users")
2274 .unwrap()
2275 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2276 .unwrap();
2277 db.table_mut("users")
2278 .unwrap()
2279 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
2280 .unwrap();
2281 db.close().unwrap();
2282 }
2283
2284 #[test]
2285 fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
2286 let dir = tempdir().unwrap();
2287 two_tables_with_records(dir.path());
2288
2289 let mut locked = Database::open(dir.path()).unwrap();
2290 let mut unlocked = open_skipping_the_lock(dir.path());
2291
2292 locked
2294 .table_mut("users")
2295 .unwrap()
2296 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
2297 .unwrap();
2298 unlocked
2299 .table_mut("orders")
2300 .unwrap()
2301 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2302 .unwrap();
2303
2304 let third = Database::open_with(
2306 dir.path(),
2307 OpenOptions {
2308 lock: LockMode::Enabled(LockOptions {
2309 timeout_ms: Some(0),
2310 ..Default::default()
2311 }),
2312 ..Default::default()
2313 },
2314 );
2315 assert!(matches!(third, Err(DatabaseError::Lock(_))));
2316
2317 unlocked.close().unwrap();
2318 locked.close().unwrap();
2319 }
2320
2321 #[test]
2326 fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
2327 let dir = tempdir().unwrap();
2328 two_tables_with_records(dir.path());
2329
2330 let mut locked = Database::open(dir.path()).unwrap();
2331 let mut unlocked = open_skipping_the_lock(dir.path());
2332
2333 locked
2334 .declare_index(
2335 "users",
2336 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2337 )
2338 .unwrap();
2339 unlocked
2340 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2341 .unwrap();
2342
2343 unlocked.close().unwrap();
2344 locked.close().unwrap();
2345
2346 assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
2347 }
2348
2349 #[test]
2350 fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
2351 let dir = tempdir().unwrap();
2352 two_tables_with_records(dir.path());
2353
2354 let mut locked = Database::open(dir.path()).unwrap();
2355 let mut unlocked = open_skipping_the_lock(dir.path());
2356
2357 locked
2358 .declare_index(
2359 "users",
2360 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2361 )
2362 .unwrap();
2363 assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
2364 unlocked
2365 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2366 .unwrap();
2367
2368 unlocked.close().unwrap();
2369 locked.close().unwrap();
2370
2371 assert_eq!(
2372 declared_on_disk(dir.path(), "users"),
2373 vec!["by_id".to_string(), "by_name".to_string()]
2374 );
2375 }
2376
2377 fn open_follower(dir: &Path) -> Database {
2380 Database::open_with(
2381 dir,
2382 OpenOptions {
2383 readonly: true,
2384 lock: LockMode::default(),
2385 ..Default::default()
2386 },
2387 )
2388 .unwrap()
2389 }
2390
2391 #[test]
2392 fn refresh_with_nothing_new_on_disk_reports_no_change() {
2393 let dir = tempdir().unwrap();
2394 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2395 let mut writer = Database::open(dir.path()).unwrap();
2396 writer
2397 .table_mut("users")
2398 .unwrap()
2399 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2400 .unwrap();
2401 let mut follower = open_follower(dir.path());
2402 follower.refresh().unwrap();
2403
2404 assert!(!follower.refresh().unwrap());
2405 writer.close().unwrap();
2406 }
2407
2408 #[test]
2409 fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
2410 let dir = tempdir().unwrap();
2411 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2412 let mut writer = Database::open(dir.path()).unwrap();
2413 let mut follower = open_follower(dir.path());
2414 assert_eq!(follower.table_names(), vec!["users"]);
2415
2416 writer
2417 .table_mut("users")
2418 .unwrap()
2419 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2420 .unwrap();
2421 writer
2422 .create_table(
2423 "orders",
2424 json!({
2425 "schemaVersion": 1,
2426 "table": "orders",
2427 "primaryKey": "id",
2428 "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
2429 }),
2430 )
2431 .unwrap();
2432 writer
2433 .table_mut("orders")
2434 .unwrap()
2435 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2436 .unwrap();
2437
2438 assert!(follower.refresh().unwrap());
2439
2440 assert_eq!(follower.table_names(), vec!["orders", "users"]);
2441 assert_eq!(
2442 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2443 Some(&json!("Ania"))
2444 );
2445 assert_eq!(
2446 follower
2447 .table("orders")
2448 .unwrap()
2449 .get("o1")
2450 .unwrap()
2451 .unwrap()
2452 .get("userId"),
2453 Some(&json!("u1"))
2454 );
2455 writer.close().unwrap();
2456 }
2457
2458 #[test]
2459 fn refresh_survives_a_compaction_run_by_the_writer_process() {
2460 let dir = tempdir().unwrap();
2461 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2462 let mut writer = Database::open(dir.path()).unwrap();
2463 let users = writer.table_mut("users").unwrap();
2464 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
2465 users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
2466 users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
2467 let mut follower = open_follower(dir.path());
2468 assert_eq!(follower.table("users").unwrap().len(), 2);
2469
2470 writer.table_mut("users").unwrap().compact().unwrap();
2471 assert!(follower.refresh().unwrap());
2472
2473 assert_eq!(follower.table("users").unwrap().len(), 2);
2474 assert_eq!(
2475 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2476 Some(&json!("Anna"))
2477 );
2478 assert_eq!(follower.table("users").unwrap().stats().gen, 1);
2479 writer.close().unwrap();
2480 }
2481
2482 #[test]
2483 fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
2484 let dir = tempdir().unwrap();
2485 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2486 let mut db = Database::open(dir.path()).unwrap();
2487
2488 let mut unit = db.unit_of_work();
2489 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
2490 unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
2491 let result = db.commit(&mut unit).unwrap();
2492 assert_eq!(result.operations, 2);
2493
2494 assert_eq!(
2495 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2496 Some(&json!("Ania"))
2497 );
2498 assert_eq!(
2499 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2500 Some(&json!("u1"))
2501 );
2502 db.close().unwrap();
2503 }
2504
2505 #[test]
2506 fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
2507 let dir = tempdir().unwrap();
2508 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2509 let mut db = Database::open(dir.path()).unwrap();
2510
2511 db.create_table(
2512 "orders",
2513 json!({
2514 "schemaVersion": 1,
2515 "table": "orders",
2516 "primaryKey": "id",
2517 "fields": {
2518 "id": { "type": "string" },
2519 "userId": { "type": "string", "required": true },
2520 },
2521 }),
2522 )
2523 .unwrap();
2524
2525 assert_eq!(db.table_names(), vec!["orders", "users"]);
2526 assert!(db.catalog.tables.contains_key("orders"));
2527
2528 db.table_mut("orders")
2529 .unwrap()
2530 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2531 .unwrap();
2532 assert_eq!(
2533 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2534 Some(&json!("u1"))
2535 );
2536
2537 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2538 assert!(reread.tables.contains_key("orders"));
2539 db.close().unwrap();
2540 }
2541
2542 #[test]
2543 fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
2544 let dir = tempdir().unwrap();
2545 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2546 let mut db = Database::open(dir.path()).unwrap();
2547 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2548
2549 let error = db
2550 .create_table(
2551 "orders",
2552 json!({
2553 "schemaVersion": 1,
2554 "table": "invoices",
2555 "primaryKey": "id",
2556 "fields": { "id": { "type": "string" } },
2557 }),
2558 )
2559 .expect_err("a schema naming another table is refused");
2560
2561 assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
2562 let message = error.to_string();
2563 assert!(
2564 message.contains("invoices"),
2565 "the refusal names the schema's table: {message}"
2566 );
2567 assert!(
2568 message.contains("orders"),
2569 "the refusal names the key it was created under: {message}"
2570 );
2571
2572 assert_eq!(db.table_names(), vec!["users"]);
2573 assert!(!db.catalog.tables.contains_key("orders"));
2574 assert!(!db.catalog.tables.contains_key("invoices"));
2575 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2576 db.close().unwrap();
2577 }
2578
2579 #[test]
2580 fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
2581 let dir = tempdir().unwrap();
2582 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2583 let mut db = Database::open(dir.path()).unwrap();
2584
2585 db.create_table(
2586 "orders",
2587 json!({
2588 "schemaVersion": 1,
2589 "table": "orders",
2590 "primaryKey": "id",
2591 "fields": { "id": { "type": "string" } },
2592 }),
2593 )
2594 .unwrap();
2595
2596 assert_eq!(db.table("orders").unwrap().schema().table, "orders");
2597 db.close().unwrap();
2598 }
2599
2600 #[test]
2601 fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
2602 let dir = tempdir().unwrap();
2603 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2604 Database::open(dir.path()).unwrap().close().unwrap();
2606 let mut db = Database::open_with(
2607 dir.path(),
2608 OpenOptions {
2609 readonly: true,
2610 lock: LockMode::default(),
2611 ..Default::default()
2612 },
2613 )
2614 .unwrap();
2615
2616 let error = db
2617 .create_table(
2618 "orders",
2619 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2620 )
2621 .unwrap_err();
2622 assert!(
2623 matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
2624 "expected a typed readonly refusal, got {error:?}"
2625 );
2626 assert_eq!(db.table_names(), vec!["users"]);
2627 }
2628
2629 #[test]
2630 fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
2631 let dir = tempdir().unwrap();
2632 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2633 let mut db = Database::open(dir.path()).unwrap();
2634 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2635 let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
2637 fs::create_dir(&temporary).unwrap();
2638
2639 let error = db
2640 .create_table(
2641 "orders",
2642 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2643 )
2644 .unwrap_err();
2645
2646 let refusal = match &error {
2647 DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
2648 .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
2649 other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
2650 };
2651 assert_eq!(refusal.path, temporary);
2652 assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
2653 assert_eq!(db.table_names(), vec!["users"]);
2654 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2655 }
2656
2657 #[test]
2658 fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
2659 let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
2660
2661 assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
2662 assert!(matches!(
2663 catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
2664 DatabaseError::Catalog(_)
2665 ));
2666 }
2667
2668 #[test]
2669 fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
2670 let dir = tempdir().unwrap();
2671 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2672 let mut db = Database::open(dir.path()).unwrap();
2673 db.table_mut("users")
2674 .unwrap()
2675 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2676 .unwrap();
2677 db.table_mut("users").unwrap().delete("u1", false).unwrap();
2678
2679 db.drop_field("users", "name", false).unwrap();
2680
2681 assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
2682 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2683
2684 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2685 assert!(!reread.tables["users"].fields.contains_key("name"));
2686 db.close().unwrap();
2687 }
2688
2689 #[test]
2690 fn drop_field_refuses_a_live_residual_field_without_force() {
2691 let dir = tempdir().unwrap();
2692 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2693 let mut db = Database::open(dir.path()).unwrap();
2694 db.table_mut("users")
2695 .unwrap()
2696 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2697 .unwrap();
2698
2699 let error = db.drop_field("users", "name", false);
2700 assert!(error.is_err());
2701 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2702 assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
2703
2704 db.drop_field("users", "name", true).unwrap();
2705 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2706 db.close().unwrap();
2707 }
2708
2709 fn users_catalog_indexing_name() -> Catalog {
2710 let mut catalog = catalog_of(&[users_schema()]);
2711 catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
2712 name: "by_name".to_string(),
2713 fields: vec!["name".to_string()],
2714 kind: CatalogIndexKind::Exact,
2715 unique: false,
2716 where_: None,
2717 expire_after_seconds: None,
2718 }]);
2719 catalog
2720 }
2721
2722 #[test]
2723 fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
2724 let dir = tempdir().unwrap();
2725 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2726 let mut db = Database::open(dir.path()).unwrap();
2727 db.table_mut("users")
2728 .unwrap()
2729 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2730 .unwrap();
2731
2732 let error = db.drop_field("users", "name", false).unwrap_err();
2733
2734 assert!(
2735 matches!(error, DatabaseError::Schema(_)),
2736 "expected a schema refusal, got {error:?}"
2737 );
2738 let message = error.to_string();
2739 assert!(message.contains("by_name"), "{message}");
2740 assert!(!message.contains("live record(s)"), "{message}");
2741 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2742 db.close().unwrap();
2743 }
2744
2745 #[test]
2746 fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
2747 let dir = tempdir().unwrap();
2748 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2749 let mut db = Database::open(dir.path()).unwrap();
2750
2751 let error = db.drop_field("users", "name", false).unwrap_err();
2752
2753 assert!(
2754 matches!(error, DatabaseError::Schema(_)),
2755 "expected a schema refusal, got {error:?}"
2756 );
2757 assert!(error.to_string().contains("by_name"));
2758 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2759 db.close().unwrap();
2760 }
2761
2762 #[test]
2763 fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
2764 let dir = tempdir().unwrap();
2765 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2766 let mut db = Database::open(dir.path()).unwrap();
2767 db.table_mut("users")
2768 .unwrap()
2769 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2770 .unwrap();
2771
2772 let error = db.drop_field("users", "name", false).unwrap_err();
2773
2774 assert!(
2775 matches!(
2776 &error,
2777 DatabaseError::FieldInUse { table, field, records }
2778 if table == "users" && field == "name" && *records == 1
2779 ),
2780 "expected a typed field-in-use refusal, got {error:?}"
2781 );
2782 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2783 db.close().unwrap();
2784 }
2785
2786 #[test]
2787 fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
2788 let dir = tempdir().unwrap();
2789 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2790 let mut db = Database::open(dir.path()).unwrap();
2791 db.table_mut("users")
2792 .unwrap()
2793 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2794 .unwrap();
2795
2796 let error = db.drop_field("users", "name", true).unwrap_err();
2797
2798 assert!(
2799 matches!(error, DatabaseError::Schema(_)),
2800 "expected a schema refusal, got {error:?}"
2801 );
2802 assert!(error.to_string().contains("by_name"));
2803 db.close().unwrap();
2804 }
2805
2806 #[test]
2807 fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
2808 let dir = tempdir().unwrap();
2809 let schema = parse_schema(&json!({
2810 "schemaVersion": 1,
2811 "table": "users",
2812 "primaryKey": "id",
2813 "fields": {
2814 "id": { "type": "string" },
2815 "email": { "type": "string" },
2816 },
2817 }))
2818 .unwrap();
2819 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2820 let mut db = Database::open(dir.path()).unwrap();
2821 db.table_mut("users")
2822 .unwrap()
2823 .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
2824 .unwrap();
2825 db.table_mut("users")
2826 .unwrap()
2827 .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
2828 .unwrap();
2829
2830 let error = db.declare_index(
2831 "users",
2832 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2833 );
2834 assert!(error.is_err());
2835 assert!(db.catalog.tables["users"].indexes.is_none());
2836
2837 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2838 assert!(reread.tables["users"].indexes.is_none());
2839 db.close().unwrap();
2840 }
2841
2842 fn orders_user_relation(enforce: bool) -> Value {
2843 json!({
2844 "name": "orders_user",
2845 "from": { "table": "orders", "field": "userId" },
2846 "to": { "table": "users", "field": "id" },
2847 "cardinality": "many-to-one",
2848 "enforce": enforce,
2849 })
2850 }
2851
2852 #[test]
2853 fn declare_relation_adds_a_relation_to_the_catalog_and_persists_it() {
2854 let dir = tempdir().unwrap();
2855 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2856 let mut db = Database::open(dir.path()).unwrap();
2857
2858 db.declare_relation(orders_user_relation(false), false).unwrap();
2859
2860 assert_eq!(db.catalog.relations.as_ref().unwrap()[0].name, "orders_user");
2861 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2862 assert_eq!(reread.relations.unwrap()[0].name, "orders_user");
2863 db.close().unwrap();
2864 }
2865
2866 #[test]
2867 fn declare_relation_with_enforce_true_refuses_over_a_live_orphan_and_force_overrides() {
2868 let dir = tempdir().unwrap();
2869 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2870 let mut db = Database::open(dir.path()).unwrap();
2871 db.table_mut("orders")
2872 .unwrap()
2873 .put(&json!({ "id": "o1", "userId": "ghost" }), false)
2874 .unwrap();
2875
2876 let error = db.declare_relation(orders_user_relation(true), false).unwrap_err();
2877 assert!(error.to_string().contains("orders_user"));
2878 assert!(db.catalog.relations.is_none());
2879
2880 db.declare_relation(orders_user_relation(true), true).unwrap();
2881 assert_eq!(db.catalog.relations.as_ref().unwrap()[0].enforce, Some(true));
2882 db.close().unwrap();
2883 }
2884
2885 #[test]
2886 fn put_checked_refuses_a_value_with_no_counterpart_on_an_enforced_relation() {
2887 let dir = tempdir().unwrap();
2888 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2889 let mut db = Database::open(dir.path()).unwrap();
2890 db.declare_relation(orders_user_relation(true), false).unwrap();
2891
2892 let error = db
2893 .put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2894 .unwrap_err();
2895
2896 assert!(error.to_string().contains("orders_user"));
2897 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
2898 db.close().unwrap();
2899 }
2900
2901 #[test]
2902 fn put_checked_allows_a_value_with_a_live_counterpart() {
2903 let dir = tempdir().unwrap();
2904 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2905 let mut db = Database::open(dir.path()).unwrap();
2906 db.declare_relation(orders_user_relation(true), false).unwrap();
2907 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2908 .unwrap();
2909
2910 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2911 .unwrap();
2912
2913 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2914 db.close().unwrap();
2915 }
2916
2917 #[test]
2918 fn put_checked_ignores_a_relation_that_does_not_declare_enforce() {
2919 let dir = tempdir().unwrap();
2920 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2921 let mut db = Database::open(dir.path()).unwrap();
2922 db.declare_relation(orders_user_relation(false), false).unwrap();
2923
2924 db.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2925 .unwrap();
2926
2927 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2928 db.close().unwrap();
2929 }
2930
2931 #[test]
2932 fn patch_checked_refuses_changing_the_field_to_a_value_with_no_counterpart() {
2933 let dir = tempdir().unwrap();
2934 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2935 let mut db = Database::open(dir.path()).unwrap();
2936 db.declare_relation(orders_user_relation(true), false).unwrap();
2937 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2938 .unwrap();
2939 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2940 .unwrap();
2941
2942 let error = db
2943 .patch_checked("orders", "o1", &json!({ "userId": "ghost" }), false)
2944 .unwrap_err();
2945
2946 assert!(error.to_string().contains("orders_user"));
2947 assert_eq!(
2948 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2949 Some(&json!("u1"))
2950 );
2951 db.close().unwrap();
2952 }
2953
2954 #[test]
2955 fn patch_checked_leaves_the_field_alone_when_changes_does_not_touch_it() {
2956 let dir = tempdir().unwrap();
2957 let orders_named = parse_schema(&json!({
2958 "schemaVersion": 1,
2959 "table": "orders",
2960 "primaryKey": "id",
2961 "fields": {
2962 "id": { "type": "string" },
2963 "userId": { "type": "string", "required": true },
2964 "note": { "type": "string" },
2965 },
2966 }))
2967 .unwrap();
2968 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_named]));
2969 let mut db = Database::open(dir.path()).unwrap();
2970 db.declare_relation(orders_user_relation(true), false).unwrap();
2971 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2972 .unwrap();
2973 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2974 .unwrap();
2975
2976 db.patch_checked("orders", "o1", &json!({ "note": "gift-wrapped" }), false)
2977 .unwrap();
2978
2979 assert_eq!(
2980 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("note"),
2981 Some(&json!("gift-wrapped"))
2982 );
2983 db.close().unwrap();
2984 }
2985
2986 #[test]
2987 fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
2988 let dir = tempdir().unwrap();
2989 let schema = parse_schema(&json!({
2990 "schemaVersion": 1,
2991 "table": "users",
2992 "primaryKey": "id",
2993 "fields": {
2994 "id": { "type": "string" },
2995 "email": { "type": "string" },
2996 },
2997 }))
2998 .unwrap();
2999 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3000 let mut db = Database::open(dir.path()).unwrap();
3001 db.table_mut("users")
3002 .unwrap()
3003 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
3004 .unwrap();
3005 db.table_mut("users")
3006 .unwrap()
3007 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
3008 .unwrap();
3009 db.table_mut("users")
3010 .unwrap()
3011 .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
3012 .unwrap();
3013
3014 db.declare_index(
3015 "users",
3016 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
3017 )
3018 .unwrap();
3019
3020 assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
3021
3022 let spec = QueryBuilder::new()
3023 .r#where(
3024 "email",
3025 crate::query::predicates::ComparisonOp::Eq,
3026 json!("piotr@example.com"),
3027 )
3028 .spec();
3029 let found = db.table("users").unwrap().query(&spec).unwrap();
3030 assert_eq!(found.keys, vec!["u2".to_string()]);
3031
3032 let conflicting = db
3033 .table_mut("users")
3034 .unwrap()
3035 .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
3036 assert!(conflicting.is_err());
3037 db.close().unwrap();
3038 }
3039
3040 #[test]
3041 fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
3042 let dir = tempdir().unwrap();
3043 let schema = parse_schema(&json!({
3044 "schemaVersion": 1,
3045 "table": "users",
3046 "primaryKey": "id",
3047 "fields": {
3048 "id": { "type": "string" },
3049 "email": { "type": "string" },
3050 },
3051 }))
3052 .unwrap();
3053 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3054 let mut db = Database::open(dir.path()).unwrap();
3055 db.table_mut("users")
3056 .unwrap()
3057 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
3058 .unwrap();
3059
3060 db.declare_index(
3061 "users",
3062 json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
3063 )
3064 .unwrap();
3065
3066 assert!(
3067 index_path(dir.path(), "users").exists(),
3068 "declare_index owes the table a snapshot carrying the new index's projection"
3069 );
3070
3071 db.table_mut("users")
3072 .unwrap()
3073 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
3074 .unwrap();
3075 assert!(
3076 db.table("users").unwrap().needs_snapshot(),
3077 "the handle declare_index left behind must still be able to snapshot itself"
3078 );
3079 db.close().unwrap();
3080
3081 let reopened = Database::open(dir.path()).unwrap();
3082 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3083 let spec = QueryBuilder::new()
3084 .r#where(
3085 "email",
3086 crate::query::predicates::ComparisonOp::Eq,
3087 json!("piotr@example.com"),
3088 )
3089 .spec();
3090 assert_eq!(
3091 reopened.table("users").unwrap().query(&spec).unwrap().keys,
3092 vec!["u2".to_string()]
3093 );
3094 assert_eq!(
3095 declared_on_disk(dir.path(), "users"),
3096 vec!["users_email_exact".to_string()]
3097 );
3098 reopened.close().unwrap();
3099 }
3100
3101 #[test]
3102 fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
3103 let dir = tempdir().unwrap();
3104 let mut catalog = catalog_of(&[users_schema()]);
3105 catalog.defaults = Some(crate::catalog::TableOptions {
3106 sync: Some(crate::catalog::CatalogSyncMode::Never),
3107 ..Default::default()
3108 });
3109 write_catalog_to(dir.path(), &catalog);
3110
3111 let db = Database::open(dir.path()).unwrap();
3112 assert_eq!(
3113 db.table("users").unwrap().sync_mode(),
3114 crate::catalog::CatalogSyncMode::Never
3115 );
3116 db.close().unwrap();
3117 }
3118
3119 #[test]
3120 fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
3121 let dir = tempdir().unwrap();
3122 let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
3123 catalog.defaults = Some(crate::catalog::TableOptions {
3124 sync: Some(crate::catalog::CatalogSyncMode::Never),
3125 ..Default::default()
3126 });
3127 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3128 sync: Some(crate::catalog::CatalogSyncMode::Commit),
3129 ..Default::default()
3130 });
3131 write_catalog_to(dir.path(), &catalog);
3132
3133 let db = Database::open(dir.path()).unwrap();
3134 assert_eq!(
3135 db.table("users").unwrap().sync_mode(),
3136 crate::catalog::CatalogSyncMode::Commit
3137 );
3138 assert_eq!(
3139 db.table("orders").unwrap().sync_mode(),
3140 crate::catalog::CatalogSyncMode::Never
3141 );
3142 db.close().unwrap();
3143 }
3144
3145 #[test]
3146 fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
3147 let dir = tempdir().unwrap();
3148 let mut catalog = catalog_of(&[users_schema()]);
3149 catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
3150 write_catalog_to(dir.path(), &catalog);
3151
3152 let mut db = Database::open(dir.path()).unwrap();
3153 let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
3154 let table = db.table_mut("users").unwrap();
3155 let record = table.put(&document, false).unwrap();
3156
3157 assert_eq!(record.get("nickname"), None);
3158 assert_eq!(
3159 table.dropped_field_warnings(&document),
3160 vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
3161 );
3162 db.close().unwrap();
3163
3164 let reopened = Database::open(dir.path()).unwrap();
3165 let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
3166 assert_eq!(stored.get("name"), Some(&json!("Ania")));
3167 assert_eq!(stored.get("nickname"), None);
3168 reopened.close().unwrap();
3169 }
3170
3171 #[test]
3172 fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
3173 let dir = tempdir().unwrap();
3174 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3175
3176 let mut db = Database::open(dir.path()).unwrap();
3177 assert_eq!(
3178 db.table("users").unwrap().validation(),
3179 crate::catalog::CatalogValidation::Strict
3180 );
3181 let error = db
3182 .table_mut("users")
3183 .unwrap()
3184 .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
3185 .unwrap_err();
3186
3187 assert!(error.to_string().contains("is not declared in the schema"), "{error}");
3188 db.close().unwrap();
3189 }
3190
3191 fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
3192 let mut catalog = catalog_of(&[users_with_age_schema()]);
3193 let entry = catalog.tables.get_mut("users").unwrap();
3194 entry.validation = validation;
3195 entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
3196 write_catalog_to(dir, &catalog);
3197 }
3198
3199 fn users_with_age_schema() -> TableSchema {
3200 parse_schema(&json!({
3201 "schemaVersion": 1,
3202 "table": "users",
3203 "primaryKey": "id",
3204 "fields": {
3205 "id": { "type": "string" },
3206 "name": { "type": "string", "required": true },
3207 "age": { "type": "int" },
3208 },
3209 }))
3210 .unwrap()
3211 }
3212
3213 #[test]
3214 fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
3215 let dir = tempdir().unwrap();
3216 deprecating_age(dir.path(), None, "use \"bornAt\" instead");
3217
3218 let mut db = Database::open(dir.path()).unwrap();
3219 let table = db.table_mut("users").unwrap();
3220 let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
3221 let record = table.put(&touching, false).unwrap();
3222
3223 assert_eq!(record.get("age"), Some(&json!(30)));
3224 assert_eq!(
3225 table.deprecated_field_warnings(&touching),
3226 vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
3227 );
3228 assert!(table
3229 .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
3230 .is_empty());
3231 db.close().unwrap();
3232 }
3233
3234 #[test]
3235 fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
3236 let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
3237 let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
3238
3239 for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
3240 let dir = tempdir().unwrap();
3241 deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
3242
3243 let mut db = Database::open(dir.path()).unwrap();
3244 let table = db.table_mut("users").unwrap();
3245 table.put(&document, false).unwrap();
3246 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3247 table.put(&document, false).unwrap();
3248 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3249 db.close().unwrap();
3250
3251 let reopened = Database::open(dir.path()).unwrap();
3252 assert_eq!(
3253 reopened.table("users").unwrap().deprecated_field_warnings(&document),
3254 expected
3255 );
3256 reopened.close().unwrap();
3257 }
3258 }
3259
3260 #[test]
3261 fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
3262 let dir = tempdir().unwrap();
3263 write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
3264
3265 let db = Database::open(dir.path()).unwrap();
3266 let table = db.table("users").unwrap();
3267
3268 assert!(table.deprecated_fields().is_empty());
3269 assert!(table
3270 .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
3271 .is_empty());
3272 db.close().unwrap();
3273 }
3274
3275 #[test]
3276 fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
3277 let dir = tempdir().unwrap();
3278 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3279
3280 let db = Database::open(dir.path()).unwrap();
3281 assert_eq!(
3282 db.table("users").unwrap().sync_mode(),
3283 crate::catalog::CatalogSyncMode::Commit
3284 );
3285 db.close().unwrap();
3286 }
3287
3288 #[test]
3289 fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
3290 let dir = tempdir().unwrap();
3291 let mut catalog = catalog_of(&[users_schema()]);
3292 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3293 sync: Some(crate::catalog::CatalogSyncMode::Never),
3294 ..Default::default()
3295 });
3296 write_catalog_to(dir.path(), &catalog);
3297
3298 {
3299 let mut db = Database::open(dir.path()).unwrap();
3300 db.table_mut("users")
3301 .unwrap()
3302 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3303 .unwrap();
3304 db.close().unwrap();
3305 }
3306
3307 let db = Database::open(dir.path()).unwrap();
3308 assert_eq!(
3309 db.table("users").unwrap().sync_mode(),
3310 crate::catalog::CatalogSyncMode::Never
3311 );
3312 assert_eq!(
3313 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3314 Some(&json!("Ania"))
3315 );
3316 db.close().unwrap();
3317 }
3318
3319 #[test]
3320 fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
3321 let dir = tempdir().unwrap();
3322 let pages_schema = parse_schema(&json!({
3323 "schemaVersion": 1,
3324 "table": "pages",
3325 "primaryKey": "id",
3326 "fields": {
3327 "id": { "type": "string" },
3328 "page": { "type": "markdown" },
3329 },
3330 }))
3331 .unwrap();
3332 let mut catalog = catalog_of(&[pages_schema]);
3333 catalog.defaults = Some(crate::catalog::TableOptions {
3334 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3335 ..Default::default()
3336 });
3337 write_catalog_to(dir.path(), &catalog);
3338
3339 let db = Database::open(dir.path()).unwrap();
3340 assert_eq!(
3341 db.table("pages").unwrap().link_parsing(),
3342 crate::catalog::LinkParsing::Lazy
3343 );
3344 db.close().unwrap();
3345 }
3346
3347 #[test]
3348 fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
3349 let dir = tempdir().unwrap();
3350 let pages_schema = parse_schema(&json!({
3351 "schemaVersion": 1,
3352 "table": "pages",
3353 "primaryKey": "id",
3354 "fields": {
3355 "id": { "type": "string" },
3356 "page": { "type": "markdown" },
3357 },
3358 }))
3359 .unwrap();
3360 let mut catalog = catalog_of(&[pages_schema]);
3361 catalog.defaults = Some(crate::catalog::TableOptions {
3362 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3363 ..Default::default()
3364 });
3365 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3366 link_parsing: Some(crate::catalog::LinkParsing::Eager),
3367 ..Default::default()
3368 });
3369 write_catalog_to(dir.path(), &catalog);
3370
3371 let db = Database::open(dir.path()).unwrap();
3372 assert_eq!(
3373 db.table("pages").unwrap().link_parsing(),
3374 crate::catalog::LinkParsing::Eager
3375 );
3376 db.close().unwrap();
3377 }
3378
3379 #[test]
3380 fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
3381 let dir = tempdir().unwrap();
3382 let docs_schema = parse_schema(&json!({
3383 "schemaVersion": 1,
3384 "table": "docs",
3385 "primaryKey": "id",
3386 "fields": {
3387 "id": { "type": "string" },
3388 "cover": { "type": "image" },
3389 },
3390 }))
3391 .unwrap();
3392 let mut catalog = catalog_of(&[docs_schema]);
3393 catalog.defaults = Some(crate::catalog::TableOptions {
3394 image_dir: Some("shared-images".to_string()),
3395 ..Default::default()
3396 });
3397 write_catalog_to(dir.path(), &catalog);
3398
3399 let mut db = Database::open(dir.path()).unwrap();
3400 let record = db
3401 .table_mut("docs")
3402 .unwrap()
3403 .put(
3404 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3405 false,
3406 )
3407 .unwrap();
3408 let physical_path = record
3409 .get("cover")
3410 .unwrap()
3411 .get("physicalPath")
3412 .unwrap()
3413 .as_str()
3414 .unwrap();
3415
3416 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3417 assert!(dir.path().join("shared-images").join(physical_path).exists());
3418 db.close().unwrap();
3419 }
3420
3421 #[test]
3422 fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
3423 let dir = tempdir().unwrap();
3424 let schema = parse_schema(&json!({
3425 "schemaVersion": 1,
3426 "table": "orders",
3427 "primaryKey": "id",
3428 "fields": {
3429 "id": { "type": "string" },
3430 "userId": { "type": "string" },
3431 "status": { "type": "string" },
3432 },
3433 }))
3434 .unwrap();
3435 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3436 let mut db = Database::open(dir.path()).unwrap();
3437 db.table_mut("orders")
3438 .unwrap()
3439 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3440 .unwrap();
3441 db.table_mut("orders")
3442 .unwrap()
3443 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3444 .unwrap();
3445 db.table_mut("orders")
3446 .unwrap()
3447 .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
3448 .unwrap();
3449
3450 db.declare_index(
3451 "orders",
3452 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3453 )
3454 .unwrap();
3455
3456 let spec = QueryBuilder::new()
3457 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3458 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
3459 .spec();
3460 let found = db.table("orders").unwrap().query(&spec).unwrap();
3461 assert_eq!(found.keys, vec!["o1".to_string()]);
3462 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3463 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3464 assert_eq!(found.plan.candidates, 1);
3465 db.close().unwrap();
3466 }
3467
3468 #[test]
3469 fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
3470 let dir = tempdir().unwrap();
3471 let schema = parse_schema(&json!({
3472 "schemaVersion": 1,
3473 "table": "users",
3474 "primaryKey": "id",
3475 "fields": {
3476 "id": { "type": "string" },
3477 "role": { "type": "string" },
3478 "status": { "type": "string" },
3479 },
3480 }))
3481 .unwrap();
3482 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3483 let mut db = Database::open(dir.path()).unwrap();
3484 db.table_mut("users")
3485 .unwrap()
3486 .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
3487 .unwrap();
3488 db.table_mut("users")
3489 .unwrap()
3490 .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
3491 .unwrap();
3492
3493 db.declare_index(
3494 "users",
3495 json!({
3496 "name": "users_role_active",
3497 "fields": ["role"],
3498 "kind": "exact",
3499 "where": [{ "field": "status", "op": "=", "value": "active" }],
3500 }),
3501 )
3502 .unwrap();
3503
3504 let eq = crate::query::predicates::ComparisonOp::Eq;
3505 let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
3506 let found = db.table("users").unwrap().query(&omitted).unwrap();
3507 assert_eq!(
3508 found.keys,
3509 vec!["u1".to_string(), "u2".to_string()],
3510 "the index holds u1 alone, but the query never asked for active records only"
3511 );
3512 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
3513
3514 let repeated = QueryBuilder::new()
3515 .r#where("role", eq, json!("admin"))
3516 .r#where("status", eq, json!("active"))
3517 .spec();
3518 let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
3519 assert_eq!(narrowed.keys, vec!["u1".to_string()]);
3520 assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3521 assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
3522 assert_eq!(narrowed.plan.candidates, 1);
3523 db.close().unwrap();
3524 }
3525
3526 #[test]
3527 fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
3528 let dir = tempdir().unwrap();
3529 let schema = parse_schema(&json!({
3530 "schemaVersion": 1,
3531 "table": "orders",
3532 "primaryKey": "id",
3533 "fields": {
3534 "id": { "type": "string" },
3535 "userId": { "type": "string" },
3536 "status": { "type": "string" },
3537 },
3538 }))
3539 .unwrap();
3540 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3541 let mut db = Database::open(dir.path()).unwrap();
3542 db.table_mut("orders")
3543 .unwrap()
3544 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3545 .unwrap();
3546 db.table_mut("orders")
3547 .unwrap()
3548 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3549 .unwrap();
3550 db.declare_index(
3551 "orders",
3552 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3553 )
3554 .unwrap();
3555 db.close().unwrap();
3556
3557 let reopened = Database::open(dir.path()).unwrap();
3558 let spec = QueryBuilder::new()
3559 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3560 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
3561 .spec();
3562 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3563 assert_eq!(found.keys, vec!["o2".to_string()]);
3564 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3565 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3566 reopened.close().unwrap();
3567 }
3568
3569 fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
3570 let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
3571 let entry = catalog.tables.get_mut(table).unwrap();
3572 entry.indexes = Some(vec![index]);
3573 write_catalog_to(dir, &catalog);
3574 }
3575
3576 fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
3577 IndexEntry {
3578 name: name.to_string(),
3579 fields: fields.iter().map(|field| field.to_string()).collect(),
3580 kind: CatalogIndexKind::Sorted,
3581 unique: false,
3582 where_: None,
3583 expire_after_seconds: None,
3584 }
3585 }
3586
3587 #[test]
3592 fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
3593 let dir = tempdir().unwrap();
3594 let schema = parse_schema(&json!({
3595 "schemaVersion": 1,
3596 "table": "orders",
3597 "primaryKey": "id",
3598 "fields": {
3599 "id": { "type": "string" },
3600 "userId": { "type": "string" },
3601 "total": { "type": "int" },
3602 },
3603 }))
3604 .unwrap();
3605 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3606 let mut db = Database::open(dir.path()).unwrap();
3607 let orders = db.table_mut("orders").unwrap();
3608 orders
3609 .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
3610 .unwrap();
3611 orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
3612 orders
3613 .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
3614 .unwrap();
3615 db.close().unwrap();
3616 assert!(
3617 index_path(dir.path(), "orders").exists(),
3618 "the fixture needs a snapshot to restore from"
3619 );
3620
3621 add_index_to_catalog_on_disk(
3622 dir.path(),
3623 "orders",
3624 sorted_index_entry("orders_user_total", &["userId", "total"]),
3625 );
3626
3627 let reopened = Database::open(dir.path()).unwrap();
3628 assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
3629
3630 let spec = QueryBuilder::new()
3631 .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
3632 .spec();
3633 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3634 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
3635 assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
3636 let mut keys = found.keys.clone();
3637 keys.sort();
3638 assert_eq!(
3639 keys,
3640 vec!["a".to_string(), "b".to_string()],
3641 "b has no total, but it still has a userId"
3642 );
3643 reopened.close().unwrap();
3644 }
3645
3646 #[test]
3649 fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
3650 let dir = tempdir().unwrap();
3651 let schema = parse_schema(&json!({
3652 "schemaVersion": 1,
3653 "table": "tickets",
3654 "primaryKey": "id",
3655 "partitionBy": "kind",
3656 "fields": {
3657 "id": { "type": "string" },
3658 "kind": { "type": "string", "required": true },
3659 "userId": { "type": "string", "required": true },
3660 "orderId": { "type": "int" },
3661 },
3662 }))
3663 .unwrap();
3664 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3665 let mut db = Database::open(dir.path()).unwrap();
3666 let tickets = db.table_mut("tickets").unwrap();
3667 tickets
3668 .put(
3669 &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
3670 false,
3671 )
3672 .unwrap();
3673 tickets
3674 .put(
3675 &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
3676 false,
3677 )
3678 .unwrap();
3679 tickets
3680 .put(
3681 &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
3682 false,
3683 )
3684 .unwrap();
3685 db.close().unwrap();
3686
3687 add_index_to_catalog_on_disk(
3688 dir.path(),
3689 "tickets",
3690 IndexEntry {
3691 name: "tickets_user_order".to_string(),
3692 fields: vec!["userId".to_string(), "orderId".to_string()],
3693 kind: CatalogIndexKind::Exact,
3694 unique: true,
3695 where_: None,
3696 expire_after_seconds: None,
3697 },
3698 );
3699
3700 let mut reopened = Database::open(dir.path()).unwrap();
3701 let tickets = reopened.table_mut("tickets").unwrap();
3702 let clash = tickets.put(
3704 &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
3705 false,
3706 );
3707 match clash {
3708 Err(TableError::Unique(violation)) => {
3709 assert_eq!(violation.field, "userId, orderId");
3710 assert_eq!(violation.holder, "t2");
3711 }
3712 other => panic!("a repeated composite key must be refused, got {other:?}"),
3713 }
3714 tickets
3715 .put(
3716 &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
3717 false,
3718 )
3719 .unwrap();
3720 reopened.close().unwrap();
3721 }
3722
3723 #[test]
3724 fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
3725 let dir = tempdir().unwrap();
3726 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3727 let mut db = Database::open(dir.path()).unwrap();
3728
3729 db.table_mut("users")
3731 .unwrap()
3732 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3733 .unwrap();
3734 db.table_mut("users")
3735 .unwrap()
3736 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3737 .unwrap();
3738 assert!(!index_path(dir.path(), "users").exists());
3739
3740 db.close().unwrap();
3741 assert!(index_path(dir.path(), "users").exists());
3742
3743 let reopened = Database::open(dir.path()).unwrap();
3744 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3745 assert_eq!(
3746 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3747 Some(&json!("Ania"))
3748 );
3749 assert_eq!(
3750 reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3751 Some(&json!("Piotr"))
3752 );
3753 reopened.close().unwrap();
3754 }
3755
3756 #[test]
3757 fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
3758 let dir = tempdir().unwrap();
3759 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3760 let mut db = Database::open(dir.path()).unwrap();
3761 db.table_mut("users")
3762 .unwrap()
3763 .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
3764 .unwrap();
3765 assert_eq!(
3766 db.table("users").unwrap().len(),
3767 0,
3768 "nothing is committed before the close"
3769 );
3770
3771 db.close().unwrap();
3772
3773 let reopened = Database::open(dir.path()).unwrap();
3774 assert_eq!(
3775 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3776 Some(&json!("Ania"))
3777 );
3778 reopened.close().unwrap();
3779 }
3780
3781 #[cfg(unix)]
3784 fn swap_file_in_place(path: &Path) {
3785 let swapped = path.with_extension("swap");
3786 std::fs::copy(path, &swapped).unwrap();
3787 std::fs::rename(&swapped, path).unwrap();
3788 }
3789
3790 fn log_replaced_of(error: &io::Error) -> Option<&crate::errors::HaysonLogReplacedError> {
3791 error.get_ref()?.downcast_ref::<crate::errors::HaysonLogReplacedError>()
3792 }
3793
3794 #[cfg(unix)]
3798 #[test]
3799 fn a_close_refused_by_a_swapped_log_still_releases_the_directory_lock() {
3800 let dir = tempdir().unwrap();
3801 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3802 let mut db = Database::open(dir.path()).unwrap();
3803 db.table_mut("users")
3804 .unwrap()
3805 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3806 .unwrap();
3807 db.table_mut("users")
3808 .unwrap()
3809 .stage_put(&json!({ "id": "u2", "name": "Piotr" }), false)
3810 .unwrap();
3811 swap_file_in_place(&log_path(dir.path(), "users"));
3812
3813 let error = db.close().unwrap_err();
3814
3815 assert!(log_replaced_of(&error).is_some(), "{error}");
3816 assert!(!lock_path_of(dir.path()).exists(), "the lock file outlived the handle");
3817 let started = std::time::Instant::now();
3818 let reopened = Database::open(dir.path()).unwrap();
3819 assert!(
3820 started.elapsed() < Duration::from_secs(1),
3821 "the reopen waited {:?} on a lock nobody holds",
3822 started.elapsed()
3823 );
3824 assert_eq!(
3825 reopened.table("users").unwrap().get("u2").unwrap(),
3826 None,
3827 "the staged write never reached the log"
3828 );
3829 reopened.close().unwrap();
3830 }
3831
3832 #[cfg(unix)]
3835 fn swap_older_version_in(path: &Path) {
3836 let current = std::fs::read_to_string(path).unwrap();
3837 let older = format!("{}\n", current.lines().next().unwrap());
3838 let swapped = path.with_extension("swap");
3839 std::fs::write(&swapped, older).unwrap();
3840 std::fs::rename(&swapped, path).unwrap();
3841 }
3842
3843 #[cfg(unix)]
3847 #[test]
3848 fn a_close_after_a_swapped_log_leaves_no_snapshot_so_the_reopen_reads_the_new_file() {
3849 let dir = tempdir().unwrap();
3850 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3851 let mut db = Database::open(dir.path()).unwrap();
3852 db.table_mut("users")
3853 .unwrap()
3854 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3855 .unwrap();
3856 db.table_mut("users")
3857 .unwrap()
3858 .put(&json!({ "id": "u1", "name": "Anna" }), false)
3859 .unwrap();
3860 swap_older_version_in(&log_path(dir.path(), "users"));
3861
3862 let error = db.close().unwrap_err();
3863
3864 assert!(log_replaced_of(&error).is_some(), "{error}");
3865 assert!(
3866 !index_path(dir.path(), "users").exists(),
3867 "a snapshot describing the orphaned inode was written"
3868 );
3869 let reopened = Database::open(dir.path()).unwrap();
3870 assert_eq!(
3871 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3872 Some(&json!("Ania")),
3873 "the reopen reads what the checkout put at the path"
3874 );
3875 reopened.close().unwrap();
3876 }
3877
3878 #[cfg(unix)]
3881 #[test]
3882 fn refresh_after_a_swapped_log_takes_the_new_file_and_writes_land_in_it_again() {
3883 let dir = tempdir().unwrap();
3884 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3885 let mut db = Database::open(dir.path()).unwrap();
3886 db.table_mut("users")
3887 .unwrap()
3888 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3889 .unwrap();
3890 db.table_mut("users")
3891 .unwrap()
3892 .put(&json!({ "id": "u1", "name": "Anna" }), false)
3893 .unwrap();
3894 let log = log_path(dir.path(), "users");
3895 swap_older_version_in(&log);
3896 assert!(db
3897 .table_mut("users")
3898 .unwrap()
3899 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3900 .is_err());
3901
3902 assert!(db.refresh().unwrap(), "the swap counts as a change");
3903
3904 let users = db.table_mut("users").unwrap();
3905 assert_eq!(users.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
3906 users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
3907 assert_eq!(std::fs::read_to_string(&log).unwrap().lines().count(), 2);
3908 db.close().unwrap();
3909
3910 let reopened = Database::open(dir.path()).unwrap();
3911 assert_eq!(
3912 reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3913 Some(&json!("Piotr"))
3914 );
3915 reopened.close().unwrap();
3916 }
3917
3918 #[test]
3919 fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
3920 let dir = tempdir().unwrap();
3921 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3922 let mut db = Database::open_with(
3923 dir.path(),
3924 OpenOptions {
3925 snapshot_every: 2,
3926 ..Default::default()
3927 },
3928 )
3929 .unwrap();
3930
3931 db.table_mut("users")
3932 .unwrap()
3933 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3934 .unwrap();
3935 assert!(!index_path(dir.path(), "users").exists());
3936 db.table_mut("users")
3937 .unwrap()
3938 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3939 .unwrap();
3940 assert!(index_path(dir.path(), "users").exists());
3941
3942 db.close().unwrap();
3943 }
3944
3945 #[test]
3946 fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
3947 let dir = tempdir().unwrap();
3948 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3949 let mut db = Database::open_with(
3950 dir.path(),
3951 OpenOptions {
3952 snapshot_every: 2,
3953 ..Default::default()
3954 },
3955 )
3956 .unwrap();
3957
3958 db.table_mut("users")
3959 .unwrap()
3960 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3961 .unwrap();
3962 db.table_mut("users")
3963 .unwrap()
3964 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3965 .unwrap();
3966 db.table_mut("users")
3967 .unwrap()
3968 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
3969 .unwrap();
3970 assert_eq!(
3971 db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
3972 Some(&json!("Kasia"))
3973 );
3974 db.close().unwrap();
3975 }
3976
3977 fn users_insert_audits_into_orders() -> Value {
3978 json!({
3979 "name": "users_insert_audit",
3980 "table": "users",
3981 "event": "insert",
3982 "effect": {
3983 "table": "orders",
3984 "fields": [
3985 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3986 { "target": "userId", "source": { "kind": "new", "field": "id" } },
3987 ],
3988 },
3989 })
3990 }
3991
3992 #[test]
3993 fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
3994 let dir = tempdir().unwrap();
3995 let docs_schema = parse_schema(&json!({
3996 "schemaVersion": 1,
3997 "table": "docs",
3998 "primaryKey": "id",
3999 "fields": {
4000 "id": { "type": "string" },
4001 "cover": { "type": "image" },
4002 },
4003 }))
4004 .unwrap();
4005 let mut catalog = catalog_of(&[docs_schema]);
4006 catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
4007 image_dir: Some("shared-images".to_string()),
4008 ..Default::default()
4009 });
4010 write_catalog_to(dir.path(), &catalog);
4011
4012 let mut db = Database::open(dir.path()).unwrap();
4013 let record = db
4014 .table_mut("docs")
4015 .unwrap()
4016 .put(
4017 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
4018 false,
4019 )
4020 .unwrap();
4021 let physical_path = record
4022 .get("cover")
4023 .unwrap()
4024 .get("physicalPath")
4025 .unwrap()
4026 .as_str()
4027 .unwrap();
4028
4029 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
4030 assert!(dir.path().join("shared-images").join(physical_path).exists());
4031 db.close().unwrap();
4032 }
4033
4034 #[test]
4035 fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
4036 let dir = tempdir().unwrap();
4037 let pages_schema = parse_schema(&json!({
4038 "schemaVersion": 1,
4039 "table": "pages",
4040 "primaryKey": "id",
4041 "fields": {
4042 "id": { "type": "string" },
4043 "page": { "type": "markdown" },
4044 },
4045 }))
4046 .unwrap();
4047 let mut catalog = catalog_of(&[pages_schema]);
4048 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
4049 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
4050 ..Default::default()
4051 });
4052 write_catalog_to(dir.path(), &catalog);
4053
4054 let mut db = Database::open(dir.path()).unwrap();
4055 assert_eq!(
4056 db.table("pages").unwrap().link_parsing(),
4057 crate::catalog::LinkParsing::Lazy
4058 );
4059
4060 let outcome = db
4061 .table_mut("pages")
4062 .unwrap()
4063 .put_with_warnings(
4064 &json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
4065 false,
4066 )
4067 .unwrap();
4068 assert!(outcome.warnings.is_empty());
4069 db.close().unwrap();
4070 }
4071
4072 #[test]
4073 fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
4074 let dir = tempdir().unwrap();
4075 let pages_schema = parse_schema(&json!({
4076 "schemaVersion": 1,
4077 "table": "pages",
4078 "primaryKey": "id",
4079 "fields": {
4080 "id": { "type": "string" },
4081 "page": { "type": "markdown" },
4082 },
4083 }))
4084 .unwrap();
4085 write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
4086
4087 let db = Database::open(dir.path()).unwrap();
4088 assert_eq!(
4089 db.table("pages").unwrap().link_parsing(),
4090 crate::catalog::LinkParsing::Eager
4091 );
4092 }
4093
4094 #[test]
4095 fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
4096 let dir = tempdir().unwrap();
4097 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4098 let mut db = Database::open(dir.path()).unwrap();
4099
4100 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4101
4102 assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
4103 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4104 assert!(db.table("orders").unwrap().triggers().is_empty());
4105
4106 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4107 assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
4108 assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
4109
4110 db.close().unwrap();
4111 }
4112
4113 #[test]
4114 fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
4115 let dir = tempdir().unwrap();
4116 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4117 let mut db = Database::open(dir.path()).unwrap();
4118 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4119
4120 let mut unit = db.unit_of_work();
4121 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4122 db.commit(&mut unit).unwrap();
4123
4124 assert_eq!(
4125 db.table("orders")
4126 .unwrap()
4127 .get("o-audit")
4128 .unwrap()
4129 .unwrap()
4130 .get("userId"),
4131 Some(&json!("u1"))
4132 );
4133 db.close().unwrap();
4134 }
4135
4136 #[test]
4137 fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
4138 let dir = tempdir().unwrap();
4139 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4140 let mut db = Database::open(dir.path()).unwrap();
4141 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4142 db.close().unwrap();
4143
4144 let mut reopened = Database::open(dir.path()).unwrap();
4145 let mut unit = reopened.unit_of_work();
4146 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4147 reopened.commit(&mut unit).unwrap();
4148
4149 assert_eq!(
4150 reopened
4151 .table("orders")
4152 .unwrap()
4153 .get("o-audit")
4154 .unwrap()
4155 .unwrap()
4156 .get("userId"),
4157 Some(&json!("u1"))
4158 );
4159 reopened.close().unwrap();
4160 }
4161
4162 #[test]
4163 fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
4164 let dir = tempdir().unwrap();
4165 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4166 let mut db = Database::open(dir.path()).unwrap();
4167 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4168
4169 db.declare_index(
4170 "users",
4171 json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
4172 )
4173 .unwrap();
4174
4175 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4176
4177 let mut unit = db.unit_of_work();
4178 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4179 db.commit(&mut unit).unwrap();
4180
4181 assert_eq!(
4182 db.table("orders")
4183 .unwrap()
4184 .get("o-audit")
4185 .unwrap()
4186 .unwrap()
4187 .get("userId"),
4188 Some(&json!("u1"))
4189 );
4190 db.close().unwrap();
4191 }
4192
4193 #[test]
4194 fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
4195 let dir = tempdir().unwrap();
4196 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4197 let mut db = Database::open(dir.path()).unwrap();
4198 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4199
4200 db.drop_field("users", "name", false).unwrap();
4201
4202 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4203
4204 let mut unit = db.unit_of_work();
4205 unit.put("users", json!({ "id": "u1" }));
4206 db.commit(&mut unit).unwrap();
4207
4208 assert_eq!(
4209 db.table("orders")
4210 .unwrap()
4211 .get("o-audit")
4212 .unwrap()
4213 .unwrap()
4214 .get("userId"),
4215 Some(&json!("u1"))
4216 );
4217 db.close().unwrap();
4218 }
4219
4220 #[test]
4221 fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
4222 let dir = tempdir().unwrap();
4223 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4224 let mut db = Database::open(dir.path()).unwrap();
4225 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4226
4227 db.create_table(
4228 "users",
4229 json!({
4230 "schemaVersion": 1,
4231 "table": "users",
4232 "revision": 2,
4233 "primaryKey": "id",
4234 "fields": {
4235 "id": { "type": "string" },
4236 "name": { "type": "string", "required": true },
4237 "city": { "type": "string" },
4238 },
4239 }),
4240 )
4241 .unwrap();
4242
4243 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4244
4245 let mut unit = db.unit_of_work();
4246 unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
4247 db.commit(&mut unit).unwrap();
4248
4249 assert_eq!(
4250 db.table("orders")
4251 .unwrap()
4252 .get("o-audit")
4253 .unwrap()
4254 .unwrap()
4255 .get("userId"),
4256 Some(&json!("u1"))
4257 );
4258 db.close().unwrap();
4259 }
4260
4261 #[test]
4262 fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
4263 let dir = tempdir().unwrap();
4264 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4265 let mut db = Database::open(dir.path()).unwrap();
4266
4267 let spec = json!({
4268 "name": "users_insert_audit",
4269 "table": "users",
4270 "event": "insert",
4271 "effect": {
4272 "table": "ghost",
4273 "fields": [
4274 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
4275 ],
4276 },
4277 });
4278 let error = db.declare_trigger(spec);
4279 assert!(error.is_err());
4280 assert!(db.catalog.triggers.is_none());
4281
4282 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4283 assert!(reread.triggers.is_none());
4284 db.close().unwrap();
4285 }
4286
4287 mod automatic_compaction {
4288 use super::*;
4289 use crate::catalog::TableOptions;
4290
4291 fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
4292 CatalogCompaction::Enabled {
4293 dead_bytes_ratio,
4294 min_dead_bytes,
4295 }
4296 }
4297
4298 fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
4299 Database::open_with(
4300 dir,
4301 OpenOptions {
4302 compaction,
4303 ..Default::default()
4304 },
4305 )
4306 .unwrap()
4307 }
4308
4309 fn lines_of(path: &Path) -> Vec<String> {
4310 fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
4311 }
4312
4313 fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
4316 let users = db.table_mut(table).unwrap();
4317 users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
4318 for round in 0..overwrites {
4319 users
4320 .patch(key, &json!({ "name": format!("Ania {round}") }), false)
4321 .unwrap();
4322 }
4323 }
4324
4325 #[test]
4326 fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
4327 let dir = tempdir().unwrap();
4328 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4329 let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
4330
4331 churn(&mut db, "users", "u1", 12);
4332
4333 let stats = db.table("users").unwrap().stats();
4334 assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
4335 assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
4336 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4337 db.close().unwrap();
4338 }
4339
4340 #[test]
4341 fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
4342 let dir = tempdir().unwrap();
4343 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4344 let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
4345
4346 churn(&mut db, "users", "u1", 4);
4347
4348 let stats = db.table("users").unwrap().stats();
4349 assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
4350 assert!(stats.dead_bytes > 1);
4351 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
4352 db.close().unwrap();
4353 }
4354
4355 #[test]
4356 fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
4357 let dir = tempdir().unwrap();
4358 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4359 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4360
4361 churn(&mut db, "users", "u1", 12);
4362
4363 let stats = db.table("users").unwrap().stats();
4364 assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
4365 assert_eq!(stats.dead_bytes, 0);
4366 assert_eq!(stats.records, 1);
4367 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4368 assert_eq!(
4369 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4370 Some(&json!("Ania 11"))
4371 );
4372 db.close().unwrap();
4373 }
4374
4375 #[test]
4376 fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
4377 let dir = tempdir().unwrap();
4378 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4379 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4380 churn(&mut db, "users", "u1", 12);
4381 db.table_mut("users")
4382 .unwrap()
4383 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
4384 .unwrap();
4385 let before = db.table("users").unwrap().stats();
4386 db.close().unwrap();
4387
4388 let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
4389
4390 let after = reopened.table("users").unwrap().stats();
4391 assert_eq!(
4392 after.gen, before.gen,
4393 "the generation is carried by the snapshot, not reset"
4394 );
4395 assert_eq!(after.records, 2);
4396 assert_eq!(after.dead_bytes, 0);
4397 let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
4398 keys.sort();
4399 assert_eq!(keys, vec!["u1", "u2"]);
4400 assert_eq!(
4401 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4402 Some(&json!("Ania 11"))
4403 );
4404 reopened.close().unwrap();
4405 }
4406
4407 #[test]
4408 fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
4409 let dir = tempdir().unwrap();
4410 let schema = parse_schema(&json!({
4411 "schemaVersion": 1,
4412 "table": "tickets",
4413 "primaryKey": "id",
4414 "partitionBy": "kind",
4415 "fields": {
4416 "id": { "type": "string" },
4417 "kind": { "type": "string", "required": true },
4418 "note": { "type": "string" },
4419 },
4420 }))
4421 .unwrap();
4422 write_catalog_to(dir.path(), &catalog_of(&[schema]));
4423 let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
4425
4426 let tickets = db.table_mut("tickets").unwrap();
4427 tickets
4428 .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
4429 .unwrap();
4430 let refund = dir.path().join("tickets").join("refund.jsonl");
4431 let quiet_size = fs::metadata(&refund).unwrap().len();
4432
4433 tickets
4434 .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
4435 .unwrap();
4436 for round in 0..12 {
4437 tickets
4438 .patch("s1", &json!({ "note": format!("n{round}") }), false)
4439 .unwrap();
4440 }
4441
4442 let sale = dir.path().join("tickets").join("sale.jsonl");
4443 assert!(
4444 lines_of(&sale).len() < 13,
4445 "the partition over its thresholds was rewritten, so not all 13 lines are still there"
4446 );
4447 assert_eq!(
4448 fs::metadata(&refund).unwrap().len(),
4449 quiet_size,
4450 "the partition holding no garbage was left byte for byte"
4451 );
4452 assert_eq!(lines_of(&refund).len(), 1);
4453 let stats = db.table("tickets").unwrap().stats();
4454 assert_eq!(
4455 stats.gen, 1,
4456 "one pass moves the generation once, not once per partition"
4457 );
4458 assert_eq!(stats.records, 2);
4459 db.close().unwrap();
4460 }
4461
4462 #[test]
4463 fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
4464 let dir = tempdir().unwrap();
4465 let mut catalog = catalog_of(&[users_schema()]);
4466 catalog.defaults = Some(TableOptions {
4467 compaction: Some(CatalogCompaction::Disabled),
4468 ..Default::default()
4469 });
4470 write_catalog_to(dir.path(), &catalog);
4471 let mut db = opened(dir.path(), None);
4472
4473 churn(&mut db, "users", "u1", 12);
4474
4475 assert_eq!(db.table("users").unwrap().stats().gen, 0);
4476 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4477 db.close().unwrap();
4478 }
4479
4480 #[test]
4481 fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
4482 let dir = tempdir().unwrap();
4483 let mut catalog = catalog_of(&[users_schema()]);
4484 catalog.defaults = Some(TableOptions {
4485 compaction: Some(CatalogCompaction::Disabled),
4486 ..Default::default()
4487 });
4488 catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
4489 compaction: Some(enabled(0.5, 200)),
4490 ..Default::default()
4491 });
4492 write_catalog_to(dir.path(), &catalog);
4493
4494 let db = opened(dir.path(), None);
4495 assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
4496 db.close().unwrap();
4497
4498 let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
4499 assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
4500 db.close().unwrap();
4501 }
4502
4503 #[test]
4504 fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
4505 let dir = tempdir().unwrap();
4506 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4507 db.create_table(
4508 "users",
4509 json!({
4510 "schemaVersion": 1,
4511 "table": "users",
4512 "primaryKey": "id",
4513 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4514 }),
4515 )
4516 .unwrap();
4517
4518 churn(&mut db, "users", "u1", 12);
4519
4520 assert!(db.table("users").unwrap().stats().gen > 0);
4521 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4522 db.close().unwrap();
4523 }
4524
4525 #[test]
4526 fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
4527 let dir = tempdir().unwrap();
4528 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4529 let db = opened(dir.path(), None);
4530
4531 assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
4532 db.close().unwrap();
4533 }
4534 }
4535
4536 mod read_cache {
4539 use super::*;
4540
4541 fn schema() -> Value {
4542 json!({
4543 "schemaVersion": 1,
4544 "table": "users",
4545 "primaryKey": "id",
4546 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4547 })
4548 }
4549
4550 fn opened(dir: &Path, cache: CachePolicy) -> Database {
4551 Database::open_with(
4552 dir,
4553 OpenOptions {
4554 cache,
4555 ..Default::default()
4556 },
4557 )
4558 .unwrap()
4559 }
4560
4561 #[test]
4562 fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
4563 let dir = tempdir().unwrap();
4564 let mut db = Database::open(dir.path()).unwrap();
4565 db.create_table("users", schema()).unwrap();
4566 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4567 db.close().unwrap();
4568
4569 let db = Database::open(dir.path()).unwrap();
4570 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4571 db.close().unwrap();
4572 }
4573
4574 #[test]
4575 fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
4576 let dir = tempdir().unwrap();
4577 let mut db = opened(dir.path(), CachePolicy::with_entries(64));
4578 db.create_table("users", schema()).unwrap();
4579 assert_eq!(
4580 db.table("users").unwrap().cache_policy(),
4581 CachePolicy::with_entries(64),
4582 "a table created under the handle is armed on the same terms"
4583 );
4584 db.close().unwrap();
4585
4586 let db = opened(dir.path(), CachePolicy::with_entries(64));
4587 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
4588 db.close().unwrap();
4589 }
4590
4591 #[test]
4592 fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
4593 let dir = tempdir().unwrap();
4594 let mut db = opened(dir.path(), CachePolicy::with_entries(8));
4595 db.create_table("users", schema()).unwrap();
4596 let users = db.table_mut("users").unwrap();
4597 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
4598
4599 let first = users.get("u1").unwrap();
4600 let second = users.get("u1").unwrap();
4601
4602 assert_eq!(first, second);
4603 assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
4604 db.close().unwrap();
4605 }
4606 }
4607
4608 mod cypher {
4609 use super::*;
4610 use crate::crypto::{CipherOptions, MissingKeyPolicy};
4611 use std::collections::HashMap;
4612
4613 fn key(byte: u8) -> Vec<u8> {
4614 vec![byte; 32]
4615 }
4616
4617 fn secrets_schema_json() -> Value {
4618 json!({
4619 "schemaVersion": 1,
4620 "table": "secrets",
4621 "primaryKey": "id",
4622 "fields": {
4623 "id": { "type": "string" },
4624 "token": { "type": "string", "cypher": true },
4625 },
4626 })
4627 }
4628
4629 fn secrets_schema_json_with_note() -> Value {
4630 json!({
4631 "schemaVersion": 1,
4632 "table": "secrets",
4633 "primaryKey": "id",
4634 "fields": {
4635 "id": { "type": "string" },
4636 "token": { "type": "string", "cypher": true },
4637 "note": { "type": "string" },
4638 },
4639 })
4640 }
4641
4642 fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
4643 Database::open_with(
4644 dir,
4645 OpenOptions {
4646 cypher: Some(CipherOptions {
4647 keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
4648 key_id: key_id.map(str::to_string),
4649 on_missing_key: None,
4650 }),
4651 ..OpenOptions::default()
4652 },
4653 )
4654 .unwrap()
4655 }
4656
4657 #[test]
4658 fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
4659 let dir = tempdir().unwrap();
4660 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4661 db.create_table("secrets", secrets_schema_json()).unwrap();
4662 db.table_mut("secrets")
4663 .unwrap()
4664 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4665 .unwrap();
4666 db.close().unwrap();
4667
4668 let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
4669 assert_eq!(
4670 reopened
4671 .table("secrets")
4672 .unwrap()
4673 .get("s1")
4674 .unwrap()
4675 .unwrap()
4676 .get("token"),
4677 Some(&json!("hunter2"))
4678 );
4679 reopened.close().unwrap();
4680 }
4681
4682 #[test]
4683 fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
4684 let dir = tempdir().unwrap();
4685 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4686 db.create_table("secrets", secrets_schema_json()).unwrap();
4687 db.table_mut("secrets")
4688 .unwrap()
4689 .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
4690 .unwrap();
4691 db.close().unwrap();
4692
4693 let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
4694 assert_eq!(
4695 rotated
4696 .table("secrets")
4697 .unwrap()
4698 .get("s1")
4699 .unwrap()
4700 .unwrap()
4701 .get("token"),
4702 Some(&json!("written-with-k1"))
4703 );
4704 rotated
4705 .table_mut("secrets")
4706 .unwrap()
4707 .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
4708 .unwrap();
4709 rotated.close().unwrap();
4710
4711 let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
4712 assert!(log.contains("$cypher$v1$k1$"));
4713 assert!(log.contains("$cypher$v1$k2$"));
4714 }
4715
4716 #[test]
4717 fn keeps_the_keyring_across_drop_field_reopening_the_table() {
4718 let dir = tempdir().unwrap();
4719 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4720 db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
4721 db.table_mut("secrets")
4722 .unwrap()
4723 .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
4724 .unwrap();
4725
4726 db.drop_field("secrets", "note", true).unwrap();
4727 assert_eq!(
4728 db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
4729 Some(&json!("hunter2"))
4730 );
4731 db.close().unwrap();
4732 }
4733
4734 #[test]
4735 fn open_refuses_a_key_that_is_not_32_bytes() {
4736 let dir = tempdir().unwrap();
4737 let error = Database::open_with(
4738 dir.path(),
4739 OpenOptions {
4740 cypher: Some(CipherOptions {
4741 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
4742 key_id: None,
4743 on_missing_key: None,
4744 }),
4745 ..OpenOptions::default()
4746 },
4747 )
4748 .map(|_| ())
4749 .unwrap_err();
4750 assert!(matches!(error, DatabaseError::Cipher(_)));
4751 }
4752
4753 #[test]
4754 fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
4755 let dir = tempdir().unwrap();
4756 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4757 db.create_table("secrets", secrets_schema_json()).unwrap();
4758 db.table_mut("secrets")
4759 .unwrap()
4760 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4761 .unwrap();
4762 db.close().unwrap();
4763
4764 let raw = Database::open_with(
4765 dir.path(),
4766 OpenOptions {
4767 cypher: Some(CipherOptions {
4768 keys: HashMap::from([("other".to_string(), key(2))]),
4769 key_id: None,
4770 on_missing_key: Some(MissingKeyPolicy::Raw),
4771 }),
4772 ..OpenOptions::default()
4773 },
4774 )
4775 .unwrap();
4776 let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
4777 assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
4778 raw.close().unwrap();
4779 }
4780 }
4781
4782 mod partitions {
4783 use super::*;
4784
4785 fn notes_schema_json(partitioned: bool) -> Value {
4786 let mut spec = json!({
4787 "schemaVersion": 1,
4788 "table": "notes",
4789 "primaryKey": "id",
4790 "fields": {
4791 "id": { "type": "string" },
4792 "kind": { "type": "string", "required": true },
4793 "title": { "type": "string", "required": true },
4794 "body": { "type": "string" },
4795 },
4796 });
4797 if partitioned {
4798 spec.as_object_mut()
4799 .unwrap()
4800 .insert("partitionBy".into(), json!("kind"));
4801 }
4802 spec
4803 }
4804
4805 fn lines_in(dir: &Path, part: &str) -> usize {
4806 fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
4807 .unwrap()
4808 .trim()
4809 .lines()
4810 .filter(|line| !line.is_empty())
4811 .count()
4812 }
4813
4814 #[test]
4815 fn a_partitioned_table_survives_the_catalog_round_trip() {
4816 let dir = tempdir().unwrap();
4817 {
4818 let mut db = Database::open(dir.path()).unwrap();
4819 db.create_table("notes", notes_schema_json(true)).unwrap();
4820 let notes = db.table_mut("notes").unwrap();
4821 notes
4822 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4823 .unwrap();
4824 notes
4825 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4826 .unwrap();
4827 db.close().unwrap();
4828 }
4829
4830 let db = Database::open(dir.path()).unwrap();
4831 let notes = db.table("notes").unwrap();
4832 assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
4833 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4834 assert_eq!(lines_in(dir.path(), "note"), 1);
4835 assert_eq!(lines_in(dir.path(), "adr"), 1);
4836 assert!(!dir.path().join("notes.jsonl").exists());
4837 db.close().unwrap();
4838 }
4839
4840 #[test]
4841 fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
4842 let dir = tempdir().unwrap();
4843 {
4844 let mut db = Database::open(dir.path()).unwrap();
4845 db.create_table("notes", notes_schema_json(false)).unwrap();
4846 let notes = db.table_mut("notes").unwrap();
4847 notes
4848 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4849 .unwrap();
4850 notes
4851 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4852 .unwrap();
4853 notes.delete("n1", false).unwrap();
4854 db.close().unwrap();
4855 }
4856
4857 let mut db = Database::open(dir.path()).unwrap();
4858 let mut partitioned = notes_schema_json(true);
4859 partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
4860 db.create_table("notes", partitioned).unwrap();
4861
4862 let notes = db.table("notes").unwrap();
4863 assert_eq!(notes.get("n1").unwrap(), None);
4864 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4865 assert!(!dir.path().join("notes.jsonl").exists());
4866 assert_eq!(lines_in(dir.path(), "adr"), 1);
4867 db.close().unwrap();
4868 }
4869
4870 #[test]
4871 fn dropping_the_field_the_table_partitions_by_is_refused() {
4872 let dir = tempdir().unwrap();
4873 let mut db = Database::open(dir.path()).unwrap();
4874 db.create_table("notes", notes_schema_json(true)).unwrap();
4875 db.table_mut("notes")
4876 .unwrap()
4877 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4878 .unwrap();
4879
4880 let error = db.drop_field("notes", "kind", true).unwrap_err();
4881 assert!(error.to_string().contains("partitioned by"));
4882 assert_eq!(
4883 db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
4884 Some(&json!("note"))
4885 );
4886 db.close().unwrap();
4887 }
4888
4889 #[test]
4890 fn dropping_an_unrelated_field_keeps_every_partition_readable() {
4891 let dir = tempdir().unwrap();
4892 let mut db = Database::open(dir.path()).unwrap();
4893 db.create_table("notes", notes_schema_json(true)).unwrap();
4894 {
4895 let notes = db.table_mut("notes").unwrap();
4896 notes
4897 .put(
4898 &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
4899 false,
4900 )
4901 .unwrap();
4902 notes
4903 .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
4904 .unwrap();
4905 }
4906
4907 db.drop_field("notes", "body", true).unwrap();
4908
4909 let notes = db.table("notes").unwrap();
4910 assert!(!notes.schema().fields.contains_key("body"));
4911 assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
4912 assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
4913 assert_eq!(lines_in(dir.path(), "note"), 1);
4914 assert_eq!(lines_in(dir.path(), "adr"), 1);
4915 db.close().unwrap();
4916 }
4917
4918 #[test]
4919 fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
4920 let dir = tempdir().unwrap();
4921 let mut db = Database::open(dir.path()).unwrap();
4922 db.create_table("notes", notes_schema_json(true)).unwrap();
4923
4924 let mut unit = db.unit_of_work();
4925 unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
4926 unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
4927 let result = db.commit(&mut unit).unwrap();
4928 assert_eq!(result.operations, 2);
4929
4930 let notes = db.table("notes").unwrap();
4931 assert_eq!(
4932 notes.get("n1").unwrap().unwrap().get("title"),
4933 Some(&json!("from unit"))
4934 );
4935 assert_eq!(
4936 notes.get("a1").unwrap().unwrap().get("title"),
4937 Some(&json!("from unit too"))
4938 );
4939 assert_eq!(lines_in(dir.path(), "note"), 1);
4940 assert_eq!(lines_in(dir.path(), "adr"), 1);
4941 db.close().unwrap();
4942 }
4943 }
4944 mod following_the_log {
4945 use super::*;
4946
4947 fn readonly(follow: FollowOption) -> OpenOptions {
4948 OpenOptions {
4949 readonly: true,
4950 follow,
4951 ..Default::default()
4952 }
4953 }
4954
4955 fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
4959 let mut writer = Database::open(dir).unwrap();
4960 writer.create_table("users", users_schema_spec()).unwrap();
4961 writer
4962 .table_mut("users")
4963 .unwrap()
4964 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4965 .unwrap();
4966 let reader = Database::open_with(dir, readonly(follow)).unwrap();
4967 (writer, reader)
4968 }
4969
4970 fn users_schema_spec() -> Value {
4971 json!({
4972 "schemaVersion": 1,
4973 "table": "users",
4974 "primaryKey": "id",
4975 "fields": {
4976 "id": { "type": "string" },
4977 "name": { "type": "string", "required": true },
4978 },
4979 })
4980 }
4981
4982 fn put(writer: &mut Database, key: &str, name: &str) {
4983 writer
4984 .table_mut("users")
4985 .unwrap()
4986 .put(&json!({ "id": key, "name": name }), false)
4987 .unwrap();
4988 }
4989
4990 #[test]
4991 fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
4992 let dir = tempdir().unwrap();
4993 let mut writer = Database::open(dir.path()).unwrap();
4994 writer.create_table("users", users_schema_spec()).unwrap();
4995
4996 assert_eq!(writer.follow(), None);
4997 assert_eq!(writer.table("users").unwrap().follow(), None);
4998
4999 let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
5000 assert_eq!(
5001 reader.follow(),
5002 Some(FollowPolicy {
5003 interval: DEFAULT_FOLLOW_INTERVAL
5004 })
5005 );
5006 assert_eq!(
5007 reader.table("users").unwrap().follow(),
5008 Some(FollowPolicy {
5009 interval: DEFAULT_FOLLOW_INTERVAL
5010 })
5011 );
5012 }
5013
5014 #[test]
5015 fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
5016 let dir = tempdir().unwrap();
5017 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5018
5019 for index in 2..=5 {
5020 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
5021 let found = reader
5022 .table_mut("users")
5023 .unwrap()
5024 .get_following(&format!("u{index}"))
5025 .unwrap();
5026 assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
5027 }
5028 }
5029
5030 #[test]
5031 fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
5032 let dir = tempdir().unwrap();
5033 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5034
5035 let spec = QueryBuilder::new().spec();
5036 assert_eq!(
5037 reader
5038 .table_mut("users")
5039 .unwrap()
5040 .query_following(&spec)
5041 .unwrap()
5042 .matched,
5043 1
5044 );
5045
5046 put(&mut writer, "u2", "Osoba 2");
5047
5048 assert_eq!(
5049 reader
5050 .table_mut("users")
5051 .unwrap()
5052 .query_following(&spec)
5053 .unwrap()
5054 .matched,
5055 2
5056 );
5057 }
5058
5059 #[test]
5060 fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
5061 let dir = tempdir().unwrap();
5062 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
5063 put(&mut writer, "u2", "Osoba 2");
5064
5065 assert!(reader
5066 .table_mut("users")
5067 .unwrap()
5068 .get_following("u2")
5069 .unwrap()
5070 .is_none());
5071 assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
5072 assert!(reader
5073 .table_mut("users")
5074 .unwrap()
5075 .get_following("u2")
5076 .unwrap()
5077 .is_some());
5078 }
5079
5080 #[test]
5081 fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
5082 let dir = tempdir().unwrap();
5083 let (mut writer, mut reader) =
5084 writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
5085
5086 assert!(reader
5088 .table_mut("users")
5089 .unwrap()
5090 .get_following("u1")
5091 .unwrap()
5092 .is_some());
5093 put(&mut writer, "u2", "Osoba 2");
5094 assert!(reader
5095 .table_mut("users")
5096 .unwrap()
5097 .get_following("u2")
5098 .unwrap()
5099 .is_none());
5100
5101 std::thread::sleep(Duration::from_millis(200));
5102
5103 assert!(reader
5104 .table_mut("users")
5105 .unwrap()
5106 .get_following("u2")
5107 .unwrap()
5108 .is_some());
5109 }
5110
5111 #[test]
5112 fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
5113 let dir = tempdir().unwrap();
5114 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5115 put(&mut writer, "u2", "Osoba 2");
5116
5117 let keys = vec!["u1".to_string(), "u2".to_string()];
5118 let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
5119 assert!(rows.iter().all(Option::is_some), "{rows:?}");
5120
5121 put(&mut writer, "u3", "Osoba 3");
5122 let spec = QueryBuilder::new().spec();
5123 let aggregate = crate::query::aggregate::AggregateSpec {
5124 count: true,
5125 ..Default::default()
5126 };
5127 let counted = reader
5128 .table_mut("users")
5129 .unwrap()
5130 .aggregate_following(&spec, &aggregate)
5131 .unwrap();
5132 assert_eq!(counted.count, 3);
5133 }
5134
5135 #[test]
5136 fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
5137 let dir = tempdir().unwrap();
5138 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5139 for index in 2..=6 {
5140 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
5141 }
5142 writer.table_mut("users").unwrap().delete("u5", false).unwrap();
5143 assert!(reader
5144 .table_mut("users")
5145 .unwrap()
5146 .get_following("u6")
5147 .unwrap()
5148 .is_some());
5149
5150 writer.table_mut("users").unwrap().compact().unwrap();
5151
5152 let users = reader.table_mut("users").unwrap();
5153 assert_eq!(
5154 users.get_following("u3").unwrap().unwrap().get("name"),
5155 Some(&json!("Osoba 3"))
5156 );
5157 assert!(users.get_following("u5").unwrap().is_none());
5158 assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
5159 }
5160 }
5161
5162 mod where_the_files_land {
5163 use super::*;
5164 use crate::database::LayoutKind;
5165
5166 fn notes_schema_spec() -> Value {
5167 json!({
5168 "schemaVersion": 1,
5169 "table": "notes",
5170 "primaryKey": "id",
5171 "partitionBy": "kind",
5172 "fields": {
5173 "id": { "type": "string" },
5174 "kind": { "type": "string", "required": true, "index": "exact" },
5175 "title": { "type": "string", "required": true },
5176 },
5177 })
5178 }
5179
5180 fn nested(metadata_file: Option<&str>) -> OpenOptions {
5181 OpenOptions {
5182 layout: LayoutOptions {
5183 kind: Some(LayoutKind::Nested),
5184 metadata_file: metadata_file.map(str::to_string),
5185 ..Default::default()
5186 },
5187 ..Default::default()
5188 }
5189 }
5190
5191 #[test]
5192 fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
5193 let dir = tempdir().unwrap();
5194 let mut db = Database::open(dir.path()).unwrap();
5195 db.create_table(
5196 "users",
5197 json!({
5198 "schemaVersion": 1,
5199 "table": "users",
5200 "primaryKey": "id",
5201 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5202 }),
5203 )
5204 .unwrap();
5205 db.table_mut("users")
5206 .unwrap()
5207 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5208 .unwrap();
5209 db.close().unwrap();
5210
5211 assert!(dir.path().join("hayson.json").is_file());
5212 assert!(dir.path().join("users.jsonl").is_file());
5213 assert!(!dir.path().join("tables").exists());
5214 assert!(!dir.path().join("runtime").exists());
5215
5216 let db = Database::open(dir.path()).unwrap();
5217 assert_eq!(db.layout().kind(), LayoutKind::Flat);
5218 assert_eq!(db.layout().metadata_file(), "hayson.json");
5219 assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
5220 }
5221
5222 #[test]
5223 fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
5224 let dir = tempdir().unwrap();
5225 let options = OpenOptions {
5226 layout: LayoutOptions {
5227 metadata_file: Some("cerebro.json".into()),
5228 ..Default::default()
5229 },
5230 ..Default::default()
5231 };
5232
5233 let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
5234 db.create_table(
5235 "users",
5236 json!({
5237 "schemaVersion": 1,
5238 "table": "users",
5239 "primaryKey": "id",
5240 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5241 }),
5242 )
5243 .unwrap();
5244 db.table_mut("users")
5245 .unwrap()
5246 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5247 .unwrap();
5248 db.close().unwrap();
5249
5250 assert!(dir.path().join("cerebro.json").is_file());
5251 assert!(!dir.path().join("hayson.json").exists());
5252
5253 let db = Database::open_with(dir.path(), options).unwrap();
5254 assert_eq!(db.layout().metadata_file(), "cerebro.json");
5255 assert_eq!(db.table_names(), vec!["users"]);
5256 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5257 }
5258
5259 #[test]
5260 fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
5261 let dir = tempdir().unwrap();
5262 let legacy = OpenOptions {
5263 layout: LayoutOptions {
5264 manifest: Some("cerebro.json".into()),
5265 ..Default::default()
5266 },
5267 ..Default::default()
5268 };
5269 let mut db = Database::open_with(dir.path(), legacy).unwrap();
5270 db.create_table(
5271 "users",
5272 json!({
5273 "schemaVersion": 1,
5274 "table": "users",
5275 "primaryKey": "id",
5276 "fields": { "id": { "type": "string" } },
5277 }),
5278 )
5279 .unwrap();
5280 db.close().unwrap();
5281
5282 let current = OpenOptions {
5283 layout: LayoutOptions {
5284 metadata_file: Some("cerebro.json".into()),
5285 ..Default::default()
5286 },
5287 ..Default::default()
5288 };
5289 let db = Database::open_with(dir.path(), current).unwrap();
5290 assert_eq!(db.table_names(), vec!["users"]);
5291 assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
5292 }
5293
5294 #[test]
5295 fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
5296 let dir = tempdir().unwrap();
5297 let options = OpenOptions {
5298 layout: LayoutOptions {
5299 metadata_file: Some("cerebro.json".into()),
5300 manifest: Some("hayson.json".into()),
5301 ..Default::default()
5302 },
5303 ..Default::default()
5304 };
5305
5306 let Err(error) = Database::open_with(dir.path(), options) else {
5307 panic!("two disagreeing names for the metadata document were accepted")
5308 };
5309
5310 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5311 let message = error.to_string();
5312 assert!(
5313 message.contains("metadataFile") && message.contains("manifest"),
5314 "{message}"
5315 );
5316 assert!(!dir.path().join("hayson.json").exists());
5317 assert!(!dir.path().join("cerebro.json").exists());
5318 assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
5319 }
5320
5321 #[test]
5322 fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
5323 let dir = tempdir().unwrap();
5324 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5325 db.create_table(
5326 "users",
5327 json!({
5328 "schemaVersion": 1,
5329 "table": "users",
5330 "primaryKey": "id",
5331 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5332 }),
5333 )
5334 .unwrap();
5335 db.table_mut("users")
5336 .unwrap()
5337 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5338 .unwrap();
5339
5340 assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
5341 db.close().unwrap();
5342
5343 assert!(dir.path().join("hayson.json").is_file());
5344 assert!(dir.path().join("tables").join("users.jsonl").is_file());
5345 assert!(dir.path().join("index").join("users.idx").is_file());
5346 assert!(!dir.path().join("users.jsonl").exists());
5347 }
5348
5349 #[test]
5350 fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
5351 let dir = tempdir().unwrap();
5352 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5353 db.create_table("notes", notes_schema_spec()).unwrap();
5354 db.table_mut("notes")
5355 .unwrap()
5356 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5357 .unwrap();
5358 db.close().unwrap();
5359
5360 assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
5361 }
5362
5363 #[test]
5364 fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
5365 let dir = tempdir().unwrap();
5366 let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
5367 db.create_table(
5368 "users",
5369 json!({
5370 "schemaVersion": 1,
5371 "table": "users",
5372 "primaryKey": "id",
5373 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5374 }),
5375 )
5376 .unwrap();
5377 db.table_mut("users")
5378 .unwrap()
5379 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5380 .unwrap();
5381 db.close().unwrap();
5382
5383 let named_only = OpenOptions {
5384 layout: LayoutOptions {
5385 metadata_file: Some("cerebro.json".into()),
5386 ..Default::default()
5387 },
5388 ..Default::default()
5389 };
5390 let db = Database::open_with(dir.path(), named_only).unwrap();
5391 assert_eq!(db.layout().kind(), LayoutKind::Nested);
5392 assert_eq!(
5393 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
5394 Some(&json!("Ania"))
5395 );
5396 db.close().unwrap();
5397
5398 let as_flat = OpenOptions {
5399 layout: LayoutOptions {
5400 kind: Some(LayoutKind::Flat),
5401 metadata_file: Some("cerebro.json".into()),
5402 ..Default::default()
5403 },
5404 ..Default::default()
5405 };
5406 let Err(error) = Database::open_with(dir.path(), as_flat) else {
5407 panic!("a nested database was opened as flat")
5408 };
5409 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5410 assert!(error.to_string().contains("nested"), "{error}");
5411 }
5412
5413 #[test]
5414 fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
5415 let dir = tempdir().unwrap();
5416 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5417 db.create_table(
5418 "users",
5419 json!({
5420 "schemaVersion": 1,
5421 "table": "users",
5422 "primaryKey": "id",
5423 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5424 }),
5425 )
5426 .unwrap();
5427
5428 let mut unit = db.unit_of_work();
5429 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
5430 unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
5431 db.commit(&mut unit).unwrap();
5432 db.close().unwrap();
5433
5434 assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
5435 let db = Database::open_with(dir.path(), nested(None)).unwrap();
5436 assert_eq!(db.table("users").unwrap().len(), 2);
5437 }
5438 }
5439
5440 mod a_declared_table_whose_log_is_not_on_disk {
5443 use super::*;
5444
5445 fn notes_spec() -> Value {
5446 json!({
5447 "schemaVersion": 1,
5448 "table": "notes",
5449 "primaryKey": "id",
5450 "partitionBy": "kind",
5451 "fields": {
5452 "id": { "type": "string" },
5453 "kind": { "type": "string", "required": true },
5454 "title": { "type": "string", "required": true },
5455 },
5456 })
5457 }
5458
5459 fn readonly() -> OpenOptions {
5460 OpenOptions {
5461 readonly: true,
5462 ..Default::default()
5463 }
5464 }
5465
5466 fn users_with_a_record(dir: &Path) {
5467 let mut db = Database::open(dir).unwrap();
5468 db.create_table(
5469 "users",
5470 json!({
5471 "schemaVersion": 1,
5472 "table": "users",
5473 "primaryKey": "id",
5474 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5475 }),
5476 )
5477 .unwrap();
5478 db.table_mut("users")
5479 .unwrap()
5480 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5481 .unwrap();
5482 db.close().unwrap();
5483 }
5484
5485 fn notes_with_a_record(dir: &Path) {
5486 let mut db = Database::open(dir).unwrap();
5487 db.create_table("notes", notes_spec()).unwrap();
5488 db.table_mut("notes")
5489 .unwrap()
5490 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5491 .unwrap();
5492 db.close().unwrap();
5493 }
5494
5495 fn refusal(dir: &Path) -> DatabaseError {
5496 match Database::open_with(dir, readonly()) {
5497 Err(error) => error,
5498 Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
5499 }
5500 }
5501
5502 fn schema_refusal(error: DatabaseError) -> String {
5503 match error {
5504 DatabaseError::Schema(HaysonSchemaError(message)) => message,
5505 other => panic!("expected a schema refusal, got {other:?}"),
5506 }
5507 }
5508
5509 #[test]
5510 fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
5511 let dir = tempdir().unwrap();
5512 users_with_a_record(dir.path());
5513 let log = log_path(dir.path(), "users");
5514 fs::remove_file(&log).unwrap();
5515
5516 let message = schema_refusal(refusal(dir.path()));
5517
5518 assert!(message.contains("\"users\""), "{message}");
5519 assert!(message.contains(&log.display().to_string()), "{message}");
5520 assert!(message.contains("readonly"), "{message}");
5521 assert!(!message.contains("createTable"), "{message}");
5522 assert!(!log.exists(), "the refused open must not have created the log");
5523 }
5524
5525 #[test]
5526 fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
5527 let dir = tempdir().unwrap();
5528 users_with_a_record(dir.path());
5529 fs::remove_file(log_path(dir.path(), "users")).unwrap();
5530
5531 let mut db = Database::open(dir.path()).unwrap();
5532
5533 assert_eq!(db.table_names(), vec!["users"]);
5534 assert!(log_path(dir.path(), "users").is_file());
5535 assert_eq!(db.table("users").unwrap().len(), 0);
5536 db.table_mut("users")
5537 .unwrap()
5538 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5539 .unwrap();
5540 assert_eq!(db.table("users").unwrap().len(), 1);
5541 db.close().unwrap();
5542 }
5543
5544 #[test]
5546 fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
5547 let dir = tempdir().unwrap();
5548 users_with_a_record(dir.path());
5549 fs::write(log_path(dir.path(), "users"), "").unwrap();
5550
5551 let db = Database::open_with(dir.path(), readonly()).unwrap();
5552
5553 assert_eq!(db.table_names(), vec!["users"]);
5554 assert_eq!(db.table("users").unwrap().len(), 0);
5555 }
5556
5557 #[test]
5558 fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
5559 let dir = tempdir().unwrap();
5560 notes_with_a_record(dir.path());
5561 let partition = dir.path().join("notes").join("note.jsonl");
5562 assert!(
5563 index_path(dir.path(), "notes").is_file(),
5564 "the refusal reads the snapshot"
5565 );
5566 fs::remove_dir_all(dir.path().join("notes")).unwrap();
5567
5568 let message = schema_refusal(refusal(dir.path()));
5569
5570 assert!(message.contains("\"notes\""), "{message}");
5571 assert!(message.contains(&partition.display().to_string()), "{message}");
5572 assert!(message.contains("readonly"), "{message}");
5573 assert!(!message.contains("createTable"), "{message}");
5574 }
5575
5576 #[test]
5577 fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
5578 let dir = tempdir().unwrap();
5579 let mut db = Database::open(dir.path()).unwrap();
5580 db.create_table("notes", notes_spec()).unwrap();
5581 db.close().unwrap();
5582
5583 let db = Database::open_with(dir.path(), readonly()).unwrap();
5584
5585 assert_eq!(db.table_names(), vec!["notes"]);
5586 assert_eq!(db.table("notes").unwrap().len(), 0);
5587 }
5588
5589 #[cfg(unix)]
5592 #[test]
5593 fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
5594 let dir = tempdir().unwrap();
5595 notes_with_a_record(dir.path());
5596 let partition = dir.path().join("notes").join("note.jsonl");
5597 fs::remove_file(&partition).unwrap();
5598 std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
5599
5600 let error = refusal(dir.path());
5601
5602 match error {
5603 DatabaseError::Log(path, inner) => {
5604 assert_eq!(path, partition);
5605 assert_eq!(inner.kind(), io::ErrorKind::NotFound);
5606 }
5607 other => panic!("expected a log refusal naming the partition, got {other:?}"),
5608 }
5609 assert!(!dir.path().join("notes").join("gone.jsonl").exists());
5610 }
5611 }
5612
5613 mod cascading_delete {
5614 use super::*;
5615
5616 fn orders_schema_optional_user() -> TableSchema {
5617 parse_schema(&json!({
5618 "schemaVersion": 1,
5619 "table": "orders",
5620 "primaryKey": "id",
5621 "fields": {
5622 "id": { "type": "string" },
5623 "userId": { "type": "string" },
5624 },
5625 }))
5626 .unwrap()
5627 }
5628
5629 fn order_items_schema() -> TableSchema {
5630 parse_schema(&json!({
5631 "schemaVersion": 1,
5632 "table": "order_items",
5633 "primaryKey": "id",
5634 "fields": {
5635 "id": { "type": "string" },
5636 "orderId": { "type": "string", "required": true },
5637 },
5638 }))
5639 .unwrap()
5640 }
5641
5642 fn orders_user_relation_on_delete(action: &str) -> Value {
5643 json!({
5644 "name": "orders_user",
5645 "from": { "table": "orders", "field": "userId" },
5646 "to": { "table": "users", "field": "id" },
5647 "cardinality": "many-to-one",
5648 "onDelete": action,
5649 })
5650 }
5651
5652 fn order_items_order_relation_cascade() -> Value {
5653 json!({
5654 "name": "order_items_order",
5655 "from": { "table": "order_items", "field": "orderId" },
5656 "to": { "table": "orders", "field": "id" },
5657 "cardinality": "many-to-one",
5658 "onDelete": "cascade",
5659 })
5660 }
5661
5662 #[test]
5663 fn a_table_no_relation_names_as_to_goes_straight_through_plain_delete() {
5664 let dir = tempdir().unwrap();
5665 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5666 let mut db = Database::open(dir.path()).unwrap();
5667 db.table_mut("users")
5668 .unwrap()
5669 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5670 .unwrap();
5671
5672 let deleted = db.delete_checked("users", "u1", false).unwrap();
5673
5674 assert!(deleted);
5675 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5676 db.close().unwrap();
5677 }
5678
5679 #[test]
5680 fn a_missing_key_answers_false_without_staging_anything() {
5681 let dir = tempdir().unwrap();
5682 write_catalog_to(
5683 dir.path(),
5684 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5685 );
5686 let mut db = Database::open(dir.path()).unwrap();
5687 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5688 .unwrap();
5689
5690 let deleted = db.delete_checked("users", "ghost", false).unwrap();
5691
5692 assert!(!deleted);
5693 db.close().unwrap();
5694 }
5695
5696 #[test]
5697 fn restrict_refuses_when_a_referencing_record_exists_and_leaves_everything_untouched() {
5698 let dir = tempdir().unwrap();
5699 write_catalog_to(
5700 dir.path(),
5701 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5702 );
5703 let mut db = Database::open(dir.path()).unwrap();
5704 db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5705 .unwrap();
5706 db.table_mut("users")
5707 .unwrap()
5708 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5709 .unwrap();
5710 db.table_mut("orders")
5711 .unwrap()
5712 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5713 .unwrap();
5714
5715 let error = db.delete_checked("users", "u1", false).unwrap_err();
5716
5717 match error {
5718 DatabaseError::Table(table, TableError::RelationRestrict(inner)) => {
5719 assert_eq!(table, "users");
5720 assert_eq!(inner.relation, "orders_user");
5721 assert_eq!(inner.referencing_table, "orders");
5722 assert_eq!(inner.referencing, 1);
5723 }
5724 other => panic!("expected a RelationRestrict refusal, got {other:?}"),
5725 }
5726 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5727 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5728 db.close().unwrap();
5729 }
5730
5731 #[test]
5732 fn restrict_allows_the_delete_once_nothing_references_it() {
5733 let dir = tempdir().unwrap();
5734 write_catalog_to(
5735 dir.path(),
5736 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5737 );
5738 let mut db = Database::open(dir.path()).unwrap();
5739 db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5740 .unwrap();
5741 db.table_mut("users")
5742 .unwrap()
5743 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5744 .unwrap();
5745
5746 let deleted = db.delete_checked("users", "u1", false).unwrap();
5747
5748 assert!(deleted);
5749 db.close().unwrap();
5750 }
5751
5752 #[test]
5753 fn cascade_deletes_every_referencing_record() {
5754 let dir = tempdir().unwrap();
5755 write_catalog_to(
5756 dir.path(),
5757 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5758 );
5759 let mut db = Database::open(dir.path()).unwrap();
5760 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5761 .unwrap();
5762 db.table_mut("users")
5763 .unwrap()
5764 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5765 .unwrap();
5766 db.table_mut("orders")
5767 .unwrap()
5768 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5769 .unwrap();
5770 db.table_mut("orders")
5771 .unwrap()
5772 .put(&json!({ "id": "o2", "userId": "u1" }), false)
5773 .unwrap();
5774 db.table_mut("orders")
5775 .unwrap()
5776 .put(&json!({ "id": "o3", "userId": "other" }), false)
5777 .unwrap();
5778
5779 let deleted = db.delete_checked("users", "u1", false).unwrap();
5780
5781 assert!(deleted);
5782 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5783 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5784 assert!(db.table("orders").unwrap().get("o2").unwrap().is_none());
5785 assert!(db.table("orders").unwrap().get("o3").unwrap().is_some());
5786 db.close().unwrap();
5787 }
5788
5789 #[test]
5790 fn cascade_recurses_through_a_chain_of_relations() {
5791 let dir = tempdir().unwrap();
5792 write_catalog_to(
5793 dir.path(),
5794 &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5795 );
5796 let mut db = Database::open(dir.path()).unwrap();
5797 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5798 .unwrap();
5799 db.declare_relation(order_items_order_relation_cascade(), false)
5800 .unwrap();
5801 db.table_mut("users")
5802 .unwrap()
5803 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5804 .unwrap();
5805 db.table_mut("orders")
5806 .unwrap()
5807 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5808 .unwrap();
5809 db.table_mut("order_items")
5810 .unwrap()
5811 .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5812 .unwrap();
5813
5814 let deleted = db.delete_checked("users", "u1", false).unwrap();
5815
5816 assert!(deleted);
5817 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5818 assert!(db.table("order_items").unwrap().get("i1").unwrap().is_none());
5819 db.close().unwrap();
5820 }
5821
5822 #[test]
5823 fn set_null_clears_the_referencing_field_instead_of_deleting_the_row() {
5824 let dir = tempdir().unwrap();
5825 write_catalog_to(
5826 dir.path(),
5827 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5828 );
5829 let mut db = Database::open(dir.path()).unwrap();
5830 db.declare_relation(orders_user_relation_on_delete("set-null"), false)
5831 .unwrap();
5832 db.table_mut("users")
5833 .unwrap()
5834 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5835 .unwrap();
5836 db.table_mut("orders")
5837 .unwrap()
5838 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5839 .unwrap();
5840
5841 let deleted = db.delete_checked("users", "u1", false).unwrap();
5842
5843 assert!(deleted);
5844 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5845 let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5846 assert_eq!(order.get("userId"), Some(&Value::Null));
5847 db.close().unwrap();
5848 }
5849
5850 #[test]
5851 fn no_action_and_undeclared_leave_referencing_records_untouched() {
5852 let dir = tempdir().unwrap();
5853 write_catalog_to(
5854 dir.path(),
5855 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5856 );
5857 let mut db = Database::open(dir.path()).unwrap();
5858 db.declare_relation(orders_user_relation_on_delete("no-action"), false)
5859 .unwrap();
5860 db.table_mut("users")
5861 .unwrap()
5862 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5863 .unwrap();
5864 db.table_mut("orders")
5865 .unwrap()
5866 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5867 .unwrap();
5868
5869 let deleted = db.delete_checked("users", "u1", false).unwrap();
5870
5871 assert!(deleted);
5872 let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5873 assert_eq!(order.get("userId"), Some(&Value::String("u1".to_string())));
5874 db.close().unwrap();
5875 }
5876
5877 #[test]
5878 fn a_restrict_deep_in_the_cascade_aborts_the_whole_delete_atomically() {
5879 let dir = tempdir().unwrap();
5880 write_catalog_to(
5881 dir.path(),
5882 &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5883 );
5884 let mut db = Database::open(dir.path()).unwrap();
5885 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5886 .unwrap();
5887 db.declare_relation(
5888 json!({
5889 "name": "order_items_order",
5890 "from": { "table": "order_items", "field": "orderId" },
5891 "to": { "table": "orders", "field": "id" },
5892 "cardinality": "many-to-one",
5893 "onDelete": "restrict",
5894 }),
5895 false,
5896 )
5897 .unwrap();
5898 db.table_mut("users")
5899 .unwrap()
5900 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5901 .unwrap();
5902 db.table_mut("orders")
5903 .unwrap()
5904 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5905 .unwrap();
5906 db.table_mut("order_items")
5907 .unwrap()
5908 .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5909 .unwrap();
5910
5911 let error = db.delete_checked("users", "u1", false).unwrap_err();
5912
5913 assert!(matches!(
5914 error,
5915 DatabaseError::Table(_, TableError::RelationRestrict(_))
5916 ));
5917 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5918 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5919 assert!(db.table("order_items").unwrap().get("i1").unwrap().is_some());
5920 db.close().unwrap();
5921 }
5922
5923 #[test]
5924 fn a_cycle_of_cascading_relations_does_not_recurse_forever() {
5925 let dir = tempdir().unwrap();
5926 let a_schema = parse_schema(&json!({
5927 "schemaVersion": 1,
5928 "table": "a",
5929 "primaryKey": "id",
5930 "fields": { "id": { "type": "string" }, "bId": { "type": "string" } },
5931 }))
5932 .unwrap();
5933 let b_schema = parse_schema(&json!({
5934 "schemaVersion": 1,
5935 "table": "b",
5936 "primaryKey": "id",
5937 "fields": { "id": { "type": "string" }, "aId": { "type": "string" } },
5938 }))
5939 .unwrap();
5940 write_catalog_to(dir.path(), &catalog_of(&[a_schema, b_schema]));
5941 let mut db = Database::open(dir.path()).unwrap();
5942 db.declare_relation(
5943 json!({
5944 "name": "a_b",
5945 "from": { "table": "a", "field": "bId" },
5946 "to": { "table": "b", "field": "id" },
5947 "cardinality": "many-to-one",
5948 "onDelete": "cascade",
5949 }),
5950 false,
5951 )
5952 .unwrap();
5953 db.declare_relation(
5954 json!({
5955 "name": "b_a",
5956 "from": { "table": "b", "field": "aId" },
5957 "to": { "table": "a", "field": "id" },
5958 "cardinality": "many-to-one",
5959 "onDelete": "cascade",
5960 }),
5961 false,
5962 )
5963 .unwrap();
5964 db.table_mut("a")
5965 .unwrap()
5966 .put(&json!({ "id": "a1", "bId": "b1" }), false)
5967 .unwrap();
5968 db.table_mut("b")
5969 .unwrap()
5970 .put(&json!({ "id": "b1", "aId": "a1" }), false)
5971 .unwrap();
5972
5973 let deleted = db.delete_checked("a", "a1", false).unwrap();
5974
5975 assert!(deleted);
5976 assert!(db.table("a").unwrap().get("a1").unwrap().is_none());
5977 assert!(db.table("b").unwrap().get("b1").unwrap().is_none());
5978 db.close().unwrap();
5979 }
5980
5981 #[test]
5982 fn delete_checked_refuses_on_a_readonly_handle() {
5983 let dir = tempdir().unwrap();
5984 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5985 {
5986 let mut db = Database::open(dir.path()).unwrap();
5987 db.table_mut("users")
5988 .unwrap()
5989 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5990 .unwrap();
5991 db.close().unwrap();
5992 }
5993 let mut db = Database::open_with(
5994 dir.path(),
5995 OpenOptions {
5996 readonly: true,
5997 ..OpenOptions::default()
5998 },
5999 )
6000 .unwrap();
6001
6002 let error = db.delete_checked("users", "u1", false).unwrap_err();
6003
6004 assert!(matches!(error, DatabaseError::ReadOnly { .. }));
6005 }
6006 }
6007
6008 mod describing_one_table {
6009 use super::*;
6010
6011 fn users_with_city() -> TableSchema {
6012 parse_schema(&json!({
6013 "schemaVersion": 1,
6014 "table": "users",
6015 "revision": 2,
6016 "primaryKey": "id",
6017 "fields": {
6018 "id": { "type": "string" },
6019 "name": { "type": "string", "required": true },
6020 "city": { "type": "string", "index": "exact" },
6021 },
6022 }))
6023 .unwrap()
6024 }
6025
6026 fn users_drifted_under(dir: &Path) -> Database {
6029 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
6030 let mut db = Database::open(dir).unwrap();
6031 write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
6032 assert!(db.refresh().unwrap());
6033 db
6034 }
6035
6036 fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
6037 match answered {
6038 Err(DatabaseError::Schema(error)) => error.to_string(),
6039 other => panic!("expected a schema refusal, got {other:?}"),
6040 }
6041 }
6042
6043 #[test]
6044 fn describes_a_table_of_its_own_while_another_one_has_drifted() {
6045 let dir = tempdir().unwrap();
6046 let db = users_drifted_under(dir.path());
6047
6048 let document = db.describe_table("orders").unwrap();
6049
6050 assert_eq!(document["table"], json!("orders"));
6051 assert_eq!(document["revision"], json!(1));
6052 assert!(db.catalog().is_err());
6053 }
6054
6055 #[test]
6056 fn refuses_the_table_that_drifted_itself() {
6057 let dir = tempdir().unwrap();
6058 let db = users_drifted_under(dir.path());
6059
6060 let message = refusal_of(db.describe_table("users"));
6061
6062 assert!(message.contains("\"users\""), "{message}");
6063 assert!(message.contains("revision 2"), "{message}");
6064 assert_eq!(db.table("users").unwrap().schema().revision, 1);
6066 }
6067
6068 #[test]
6069 fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
6070 let dir = tempdir().unwrap();
6071 let mut catalog = catalog_of(&[users_schema()]);
6072 let entry = catalog.tables.get_mut("users").unwrap();
6073 entry.indexes = Some(vec![IndexEntry {
6074 name: "by_name".to_string(),
6075 fields: vec!["name".to_string()],
6076 kind: crate::catalog::CatalogIndexKind::Exact,
6077 unique: false,
6078 where_: None,
6079 expire_after_seconds: None,
6080 }]);
6081 write_catalog_to(dir.path(), &catalog);
6082 let db = Database::open(dir.path()).unwrap();
6083
6084 let document = db.describe_table("users").unwrap();
6085
6086 assert_eq!(
6087 document["fields"]["name"],
6088 json!({ "type": "string", "required": true })
6089 );
6090 assert!(document.as_object().unwrap().get("indexes").is_none());
6091 assert!(db.catalog().is_ok());
6092 }
6093
6094 #[test]
6095 fn refuses_a_table_this_handle_holds_nowhere() {
6096 let dir = tempdir().unwrap();
6097 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
6098 let db = Database::open(dir.path()).unwrap();
6099
6100 let message = refusal_of(db.describe_table("ghosts"));
6101
6102 assert!(message.contains("declares no table \"ghosts\""), "{message}");
6103 }
6104 }
6105}