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