1use super::*;
2use rusqlite::OpenFlags;
3
4const COMPATIBILITY_METADATA_VERSION: i64 = 30;
5
6pub(super) struct SchemaState {
7 pub(super) revision: i64,
8 minimum_compatible: Option<i64>,
9}
10
11impl SchemaState {
12 pub(super) fn ensure_supported(&self) -> Result<()> {
13 let reason = if self.revision < SCHEMA_VERSION {
14 StoreSchemaMismatchReason::NeedsMigration
15 } else if let Some(minimum_compatible) = self.minimum_compatible {
16 if minimum_compatible <= SCHEMA_VERSION {
17 return Ok(());
18 }
19 StoreSchemaMismatchReason::Incompatible { minimum_compatible }
20 } else {
21 StoreSchemaMismatchReason::InvalidCompatibilityMetadata
22 };
23 Err(StoreSchemaMismatch {
24 found: self.revision,
25 supported: SCHEMA_VERSION,
26 reason,
27 }
28 .into())
29 }
30}
31
32pub(super) fn read_schema_state(connection: &Connection) -> Result<SchemaState> {
35 let snapshot = connection
36 .unchecked_transaction()
37 .context("start database compatibility snapshot")?;
38 let revision: i64 = snapshot
39 .query_row("PRAGMA user_version", [], |row| row.get(0))
40 .context("read database migration revision")?;
41 let minimum_compatible = if revision >= COMPATIBILITY_METADATA_VERSION {
42 let invalid = || StoreSchemaMismatch {
43 found: revision,
44 supported: SCHEMA_VERSION,
45 reason: StoreSchemaMismatchReason::InvalidCompatibilityMetadata,
46 };
47 let (count, singleton, floor, recorded): (i64, Option<i64>, Option<i64>, Option<i64>) =
48 snapshot
49 .query_row(
50 "SELECT count(*), min(singleton), min(minimum_compatible_version),
51 (SELECT max(version) FROM schema_migrations)
52 FROM schema_compatibility",
53 [],
54 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
55 )
56 .map_err(|error| {
57 let structural = match &error {
61 rusqlite::Error::SqliteFailure(code, _) => {
62 code.code == rusqlite::ErrorCode::Unknown
63 }
64 _ => true,
65 };
66 let error = anyhow::Error::new(error);
67 if structural {
68 error.context(invalid())
69 } else {
70 error.context("read database compatibility metadata")
71 }
72 })?;
73 if count != 1
74 || singleton != Some(1)
75 || recorded != Some(revision)
76 || !floor
77 .is_some_and(|floor| (COMPATIBILITY_METADATA_VERSION..=revision).contains(&floor))
78 {
79 return Err(invalid().into());
80 }
81 floor
82 } else {
83 None
84 };
85 snapshot
86 .commit()
87 .context("finish database compatibility snapshot")?;
88 Ok(SchemaState {
89 revision,
90 minimum_compatible,
91 })
92}
93
94pub fn database_path() -> PathBuf {
95 data_dir().join("mj.sqlite3")
96}
97
98pub(super) fn open_writer(path: &Path) -> Result<Connection> {
99 if let Some(parent) = path.parent() {
100 fs::create_dir_all(parent)
101 .with_context(|| format!("create Mjolnir data directory {}", parent.display()))?;
102 }
103 let connection = Connection::open(path)
104 .with_context(|| format!("open Mjolnir database {}", path.display()))?;
105 connection.busy_timeout(Duration::from_secs(5))?;
106 connection.execute_batch(
107 "PRAGMA foreign_keys = ON;
108 PRAGMA journal_mode = WAL;
109 PRAGMA synchronous = FULL;",
110 )?;
111 verify_schema_once(path, &connection)?;
112 Ok(connection)
113}
114
115pub(super) fn open(path: &Path) -> Result<Connection> {
116 open_writer(path)
117}
118
119#[cfg(not(test))]
123pub(super) fn open_reader(path: &Path) -> Result<Connection> {
124 open_reader_strict(path)
125}
126
127#[cfg(test)]
128pub(super) fn open_reader(path: &Path) -> Result<Connection> {
129 open_writer(path)
133}
134
135#[cfg_attr(test, allow(dead_code))]
136fn open_reader_strict(path: &Path) -> Result<Connection> {
137 let connection = Connection::open_with_flags(
138 path,
139 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
140 )
141 .with_context(|| format!("open Mjolnir database read-only {}", path.display()))?;
142 connection.busy_timeout(Duration::from_secs(5))?;
143 connection.execute_batch(
144 "PRAGMA foreign_keys = ON;
145 PRAGMA query_only = ON;",
146 )?;
147 read_schema_state(&connection)?.ensure_supported()?;
148 Ok(connection)
149}
150
151fn verified_schemas() -> &'static Mutex<HashSet<PathBuf>> {
155 static VERIFIED: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
156 VERIFIED.get_or_init(|| Mutex::new(HashSet::new()))
157}
158
159fn schema_cache_key(path: &Path) -> PathBuf {
162 let Some(parent) = path
163 .parent()
164 .filter(|parent| !parent.as_os_str().is_empty())
165 else {
166 return path.to_owned();
167 };
168 match (fs::canonicalize(parent), path.file_name()) {
169 (Ok(canonical), Some(name)) => canonical.join(name),
170 _ => path.to_owned(),
171 }
172}
173
174fn verify_schema_once(path: &Path, connection: &Connection) -> Result<()> {
179 let key = schema_cache_key(path);
180 let mut verified = verified_schemas()
181 .lock()
182 .unwrap_or_else(PoisonError::into_inner);
183 let state = read_schema_state(connection)?;
184 if state.revision > SCHEMA_VERSION
185 || (state.revision == SCHEMA_VERSION && verified.contains(&key))
186 {
187 return state.ensure_supported();
189 }
190 migrate_schema(connection)?;
193 read_schema_state(connection)?.ensure_supported()?;
194 verified.insert(key);
195 Ok(())
196}
197
198#[cfg(test)]
202pub(super) fn forget_verified_schema(path: &Path) {
203 verified_schemas()
204 .lock()
205 .unwrap_or_else(PoisonError::into_inner)
206 .remove(&schema_cache_key(path));
207}
208
209const BASELINE_SCHEMA_VERSION: i64 = 33;
213
214const BASELINE_MINIMUM_COMPATIBLE_VERSION: i64 = 32;
217
218fn migrate_schema(connection: &Connection) -> Result<()> {
219 let state = read_schema_state(connection)?;
220 let version = state.revision;
221 if version > SCHEMA_VERSION {
222 return state.ensure_supported();
223 }
224 if version == 0 {
225 create_baseline_schema(connection)?;
226 } else if version < BASELINE_SCHEMA_VERSION {
227 bail!(
231 "Mjolnir database schema {version} was written by a Mjolnir release older than 2.7.2, \
232 which this build cannot upgrade; upgrade through any Mjolnir release from 2.7.2 \
233 through 2.9.x first, or start with a fresh data directory (--instance NAME or \
234 MJ_DATA_DIR)"
235 );
236 }
237 if version < 34 {
245 connection.execute_batch(
246 "BEGIN IMMEDIATE;
247 CREATE TABLE IF NOT EXISTS session_mount_access (
248 session_id TEXT NOT NULL REFERENCES sessions(session_id) ON DELETE CASCADE,
249 source BLOB NOT NULL,
250 destination BLOB NOT NULL,
251 access TEXT NOT NULL CHECK(access IN ('rw')),
252 PRIMARY KEY(session_id, destination)
253 ) STRICT;
254 INSERT INTO schema_migrations(version, applied_at)
255 VALUES (34, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
256 PRAGMA user_version = 34;
257 COMMIT;",
258 )?;
259 }
260 if version < 35 {
267 connection.execute_batch(
268 "BEGIN IMMEDIATE;
269 ALTER TABLE sessions ADD COLUMN container_workspace TEXT;
270 INSERT INTO schema_migrations(version, applied_at)
271 VALUES (35, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
272 PRAGMA user_version = 35;
273 COMMIT;",
274 )?;
275 }
276 if version < 36 {
282 connection.execute_batch(
283 "BEGIN IMMEDIATE;
284 ALTER TABLE sessions ADD COLUMN build_cache_json TEXT;
285 INSERT INTO schema_migrations(version, applied_at)
286 VALUES (36, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
287 PRAGMA user_version = 36;
288 COMMIT;",
289 )?;
290 }
291 if version < 37 {
297 connection.execute_batch(
298 "BEGIN IMMEDIATE;
299 ALTER TABLE session_targets ADD COLUMN borrowed_from TEXT;
300 INSERT INTO schema_migrations(version, applied_at)
301 VALUES (37, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
302 PRAGMA user_version = 37;
303 COMMIT;",
304 )?;
305 }
306 if version < 38 {
313 connection.execute_batch(
314 "BEGIN IMMEDIATE;
315 CREATE TABLE IF NOT EXISTS workspace_layouts (
316 workspace_id TEXT PRIMARY KEY REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
317 layout TEXT NOT NULL
318 ) STRICT;
319 INSERT INTO schema_migrations(version, applied_at)
320 VALUES (38, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
321 PRAGMA user_version = 38;
322 COMMIT;",
323 )?;
324 }
325 let recorded: Option<i64> =
326 connection.query_row("SELECT max(version) FROM schema_migrations", [], |row| {
327 row.get(0)
328 })?;
329 if recorded != Some(SCHEMA_VERSION) {
330 bail!(
331 "Mjolnir database migration ledger {:?} does not match schema {}",
332 recorded,
333 SCHEMA_VERSION
334 );
335 }
336 Ok(())
337}
338
339fn create_baseline_schema(connection: &Connection) -> Result<()> {
343 connection.execute_batch("BEGIN IMMEDIATE;")?;
344 let created = (|| -> Result<()> {
345 let version: i64 = connection.query_row("PRAGMA user_version", [], |row| row.get(0))?;
346 if version != 0 {
347 return Ok(());
348 }
349 connection.execute_batch(include_str!("baseline.sql"))?;
350 connection.execute(
351 "INSERT INTO schema_compatibility(singleton, minimum_compatible_version) VALUES (1, ?1)",
352 [BASELINE_MINIMUM_COMPATIBLE_VERSION],
353 )?;
354 connection.execute(
355 "INSERT INTO schema_migrations(version, applied_at)
356 VALUES (?1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
357 [BASELINE_SCHEMA_VERSION],
358 )?;
359 connection.pragma_update(None, "user_version", BASELINE_SCHEMA_VERSION)?;
360 Ok(())
361 })();
362 match created {
363 Ok(()) => connection
364 .execute_batch("COMMIT;")
365 .context("commit baseline database schema"),
366 Err(error) => {
367 if let Err(rollback) = connection.execute_batch("ROLLBACK;") {
368 tracing::warn!(%rollback, "could not roll back a failed baseline schema");
369 }
370 Err(error.context("create baseline database schema"))
371 }
372 }
373}
374
375#[cfg(test)]
376pub(super) fn advance_test_schema(path: &Path, revision: i64, minimum_compatible: i64) {
377 let connection = Connection::open(path).unwrap();
378 let transaction = connection.unchecked_transaction().unwrap();
379 transaction
380 .execute(
381 "UPDATE schema_compatibility SET minimum_compatible_version = ?1",
382 [minimum_compatible],
383 )
384 .unwrap();
385 transaction
386 .execute(
387 "INSERT INTO schema_migrations(version, applied_at) VALUES (?1, 'test')",
388 [revision],
389 )
390 .unwrap();
391 transaction
392 .pragma_update(None, "user_version", revision)
393 .unwrap();
394 transaction.commit().unwrap();
395 forget_verified_schema(path);
396}
397
398#[cfg(test)]
399mod reader_tests {
400 use super::*;
401
402 const MINIMUM_COMPATIBLE_VERSION: i64 = 32;
406
407 fn stamp_schema_version(path: &Path, version: i64) {
410 if version > SCHEMA_VERSION {
411 advance_test_schema(path, version, version);
412 return;
413 }
414 let connection = Connection::open(path).unwrap();
415 connection
416 .execute_batch(&format!("PRAGMA user_version = {version};"))
417 .unwrap();
418 connection
419 .execute(
420 "DELETE FROM schema_migrations WHERE version > ?1",
421 [version],
422 )
423 .unwrap();
424 if version == 30 {
425 connection
426 .execute(
427 "UPDATE schema_compatibility SET minimum_compatible_version = 30 WHERE singleton = 1",
428 [],
429 )
430 .unwrap();
431 }
432 drop(connection);
433 forget_verified_schema(path);
434 }
435
436 #[test]
437 fn older_readers_and_reopened_writers_preserve_a_compatible_future_schema() {
438 let directory = tempfile::tempdir().unwrap();
439 let path = directory.path().join("mj.sqlite3");
440 let connection = open_writer(&path).unwrap();
441 connection
442 .execute_batch(
443 "CREATE TABLE future_feature(value TEXT NOT NULL);
444 INSERT INTO future_feature VALUES ('preserve me');",
445 )
446 .unwrap();
447 drop(connection);
448 advance_test_schema(&path, SCHEMA_VERSION + 1, SCHEMA_VERSION);
449
450 let reader = open_reader_strict(&path).unwrap();
451 assert_eq!(
452 reader
453 .query_row("SELECT value FROM future_feature", [], |row| row
454 .get::<_, String>(0))
455 .unwrap(),
456 "preserve me"
457 );
458 assert!(reader.execute("DELETE FROM future_feature", []).is_err());
459 drop(reader);
460
461 let raw = Connection::open(&path).unwrap();
464 raw.execute_batch("DROP TRIGGER api_session_error_updated;")
465 .unwrap();
466 drop(raw);
467 let writer = open_writer(&path).unwrap();
468 assert!(!writer.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name = 'api_session_error_updated')", [], |row| row.get::<_, bool>(0)).unwrap());
469 assert_eq!(
470 writer
471 .query_row("SELECT value FROM future_feature", [], |row| row
472 .get::<_, String>(0))
473 .unwrap(),
474 "preserve me"
475 );
476 let state = read_schema_state(&writer).unwrap();
477 assert_eq!(state.revision, SCHEMA_VERSION + 1);
478 assert_eq!(state.minimum_compatible, Some(SCHEMA_VERSION));
479 }
480
481 #[test]
482 fn invalid_compatibility_metadata_refuses_readers_and_writers() {
483 for alteration in [
484 "DROP TABLE schema_compatibility",
485 "DELETE FROM schema_compatibility",
486 "PRAGMA ignore_check_constraints = ON; UPDATE schema_compatibility SET minimum_compatible_version = 0",
487 "UPDATE schema_compatibility SET minimum_compatible_version = 99999",
488 "PRAGMA ignore_check_constraints = ON; UPDATE schema_compatibility SET singleton = 2",
489 "PRAGMA ignore_check_constraints = ON; INSERT INTO schema_compatibility VALUES (2, 30)",
490 "DROP TABLE schema_compatibility; CREATE TABLE schema_compatibility(singleton, minimum_compatible_version); INSERT INTO schema_compatibility VALUES (1, 'invalid')",
491 "DELETE FROM schema_migrations WHERE version = (SELECT max(version) FROM schema_migrations)",
492 ] {
493 for future in [false, true] {
494 let directory = tempfile::tempdir().unwrap();
495 let path = directory.path().join("mj.sqlite3");
496 drop(open_writer(&path).unwrap());
497 if future {
498 advance_test_schema(&path, SCHEMA_VERSION + 1, SCHEMA_VERSION);
499 }
500 let raw = Connection::open(&path).unwrap();
501 raw.execute_batch(alteration).unwrap();
502 let before: i64 = raw
503 .query_row("PRAGMA schema_version", [], |row| row.get(0))
504 .unwrap();
505 for error in [
507 open_reader_strict(&path).unwrap_err(),
508 open_writer(&path).unwrap_err(),
509 ] {
510 let mismatch = error.downcast_ref::<StoreSchemaMismatch>().unwrap();
511 assert_eq!(
512 mismatch.reason,
513 StoreSchemaMismatchReason::InvalidCompatibilityMetadata,
514 "{alteration}"
515 );
516 }
517 forget_verified_schema(&path);
518 assert!(open_writer(&path).is_err(), "{alteration}");
519 let after: i64 = raw
520 .query_row("PRAGMA schema_version", [], |row| row.get(0))
521 .unwrap();
522 assert_eq!(
523 before, after,
524 "a rejected open repaired schema: {alteration}"
525 );
526 }
527 }
528 }
529
530 #[test]
531 fn a_failed_baseline_leaves_an_empty_store_that_a_retry_creates() {
532 let directory = tempfile::tempdir().unwrap();
533 let path = directory.path().join("mj.sqlite3");
534 let connection = Connection::open(&path).unwrap();
535 connection
537 .execute_batch("CREATE TABLE workspaces(conflict TEXT)")
538 .unwrap();
539
540 let error = migrate_schema(&connection).unwrap_err();
541
542 assert!(format!("{error:#}").contains("create baseline database schema"));
543 assert!(
544 connection.is_autocommit(),
545 "the failed baseline left a transaction open"
546 );
547 assert_eq!(read_schema_state(&connection).unwrap().revision, 0);
548 let tables: i64 = connection
549 .query_row(
550 "SELECT count(*) FROM sqlite_schema WHERE type = 'table'",
551 [],
552 |row| row.get(0),
553 )
554 .unwrap();
555 assert_eq!(tables, 1, "only the conflicting table remains");
556
557 connection.execute_batch("DROP TABLE workspaces").unwrap();
558 drop(connection);
559 let writer = open_writer(&path).unwrap();
560 let state = read_schema_state(&writer).unwrap();
561 assert_eq!(state.revision, SCHEMA_VERSION);
562 assert_eq!(state.minimum_compatible, Some(MINIMUM_COMPATIBLE_VERSION));
563 }
564
565 #[test]
566 fn a_store_from_before_the_baseline_is_refused_with_upgrade_advice() {
567 let directory = tempfile::tempdir().unwrap();
568 let path = directory.path().join("mj.sqlite3");
569 let connection = Connection::open(&path).unwrap();
570 connection
571 .execute_batch(&format!(
572 "PRAGMA user_version = {};",
573 COMPATIBILITY_METADATA_VERSION - 1
574 ))
575 .unwrap();
576 drop(connection);
577
578 let error = open_writer(&path).unwrap_err();
579
580 let message = format!("{error:#}");
581 assert!(message.contains("older than 2.7.2"), "{message}");
582 assert!(
583 message.contains("from 2.7.2 through 2.9.x"),
584 "advice must name the closed range of releases that can migrate: {message}"
585 );
586 }
587
588 #[test]
592 fn strict_reader_reports_a_newer_store_without_blaming_the_daemon() {
593 let directory = tempfile::tempdir().unwrap();
594 let path = directory.path().join("mj.sqlite3");
595 drop(open_writer(&path).unwrap());
596 stamp_schema_version(&path, SCHEMA_VERSION + 1);
597
598 let error = open_reader_strict(&path).unwrap_err();
599
600 let mismatch = error
601 .chain()
602 .find_map(|cause| cause.downcast_ref::<StoreSchemaMismatch>())
603 .expect("the reader reports the mismatch as a typed cause");
604 assert_eq!(mismatch.found, SCHEMA_VERSION + 1);
605 assert_eq!(mismatch.supported, SCHEMA_VERSION);
606 let message = mismatch.to_string();
607 assert!(message.contains("upgrade Mjolnir"), "got {message}");
608 assert!(
609 !message.contains("start the Mjolnir daemon"),
610 "got {message}"
611 );
612 }
613
614 #[test]
617 fn strict_reader_keeps_the_migrate_advice_when_the_store_is_behind() {
618 let directory = tempfile::tempdir().unwrap();
619 let path = directory.path().join("mj.sqlite3");
620 drop(open_writer(&path).unwrap());
621 let raw = Connection::open(&path).unwrap();
622 raw.execute_batch(&format!(
623 "UPDATE schema_compatibility SET minimum_compatible_version = {0};
624 DELETE FROM schema_migrations WHERE version > {0};
625 INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES ({0}, 'test');
626 PRAGMA user_version = {0};",
627 SCHEMA_VERSION - 1
628 ))
629 .unwrap();
630 drop(raw);
631
632 let error = open_reader_strict(&path).unwrap_err();
633
634 let mismatch = error
635 .chain()
636 .find_map(|cause| cause.downcast_ref::<StoreSchemaMismatch>())
637 .expect("the reader reports the mismatch as a typed cause");
638 assert_eq!(
639 mismatch.to_string(),
640 format!(
641 "Mjolnir database schema {} is not the supported schema {SCHEMA_VERSION}; \
642 start the Mjolnir daemon to migrate it",
643 SCHEMA_VERSION - 1
644 )
645 );
646 }
647
648 #[test]
649 fn strict_reader_rejects_mutation() {
650 let directory = tempfile::tempdir().unwrap();
651 let path = directory.path().join("mj.sqlite3");
652 drop(open_writer(&path).unwrap());
653
654 let reader = open_reader_strict(&path).unwrap();
655 let error = reader
656 .execute("CREATE TABLE forbidden(value TEXT)", [])
657 .unwrap_err();
658 assert!(
659 matches!(
660 error.sqlite_error_code(),
661 Some(rusqlite::ErrorCode::ReadOnly)
662 ),
663 "unexpected mutation error: {error}"
664 );
665 }
666}