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