mj_controller/database/
schema.rs

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
32/// The revision, ledger, and compatibility floor must describe one snapshot.
33/// A missing floor is only legitimate before compatibility was introduced.
34pub(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                    // Missing tables/columns and invalid field types are
58                    // structural. Busy, I/O, and interruption errors are not
59                    // evidence of an incompatible migration.
60                    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/// Open an existing database without permitting schema or data mutation.
120/// Client processes use this path so an accidental write fails locally
121/// instead of competing with the daemon's writer.
122#[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    // Path-taking database helpers are migration fixtures in unit tests: they
130    // intentionally open old or not-yet-created schemas. Production query
131    // entry points compile against the strict reader above.
132    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
151/// Databases this process has already migrated. A controller owns its store
152/// exclusively (`ControllerStoreGuard`), so a schema verified once stays
153/// verified and later connections skip the migration probes entirely.
154fn 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
159/// Stable cache identity for a database. The file itself may not exist yet, so
160/// the canonicalized parent directory carries the identity.
161fn 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
174/// Run the migration ladder the first time this process opens a database.
175/// Later opens confirm compatibility without repeating schema repairs. A
176/// database behind this build is migrated again, so a recreated file under a
177/// reused path still converges. Compatible future stores are never repaired.
178fn 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        // An older build must never run its repairs against a newer schema.
188        return state.ensure_supported();
189    }
190    // Holding the lock across the ladder keeps two first opens of the same
191    // database from running the additive migration steps against each other.
192    migrate_schema(connection)?;
193    read_schema_state(connection)?.ensure_supported()?;
194    verified.insert(key);
195    Ok(())
196}
197
198/// Forget that this process verified a database's schema. Only tests need it:
199/// they simulate a store written by an older build by editing the schema of a
200/// database this process has already opened, which no controller can do.
201#[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
209/// The oldest revision this build upgrades in place: the revision Mjolnir 2.7.2
210/// shipped. A new store is created directly at this revision from
211/// `baseline.sql`; stores written by older builds are refused.
212const BASELINE_SCHEMA_VERSION: i64 = 33;
213
214/// The compatibility floor a baseline store records. Migration 32 (ZCode) was
215/// the last breaking change before the baseline.
216const 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        // Every release from 2.7.2 through 2.9.x still carries the migration
228        // chain below the baseline; every later release refuses, as this one
229        // does. That range is closed, so the advice never goes stale.
230        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    // Compatible: adds one table. Older readers ignore it and treat read-write
238    // mounts as copy-on-write, a behaviour difference rather than lost data.
239    // Older writers rewrite `session_mounts` but never touch this table, so its
240    // rows survive their updates; a row only applies while a mount with the same
241    // source and destination is still not read-only, so an older build that
242    // makes the mount read-only or removes it keeps that choice. The
243    // compatibility floor stays where it is.
244    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    // Compatible: adds one nullable column. Older readers ignore it, and the
261    // older writer's session upsert lists columns explicitly, so it preserves
262    // the value. An older executable launching such a session uses the shared
263    // `/workspace` instead of the recorded per-session path, which is a
264    // behaviour difference, not data loss. The compatibility floor stays where
265    // it is.
266    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    // Compatible: adds one nullable column. Older readers ignore it, and the
277    // older writer's session upsert lists columns explicitly, so it preserves
278    // the value. An older executable launching such a session runs it without
279    // the mbx build cache, which is a behaviour difference, not data loss. The
280    // compatibility floor stays where it is.
281    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    // Compatible: adds one nullable column to `session_targets`. Only a
292    // container sub-agent child row ever carries a value, and older builds
293    // could never start such a child, so an older update that rewrites the row
294    // without the column loses nothing usable. Older readers ignore it. The
295    // compatibility floor stays where it is.
296    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    // Compatible: adds one table holding the dashboard's conversation pane
307    // arrangement per workspace. Older readers never select from it and older
308    // writers never touch it, so their updates leave its rows intact; a
309    // workspace deleted by an older build still removes them through the
310    // foreign key. Losing the table only means the conversation area opens as
311    // a single pane. The compatibility floor stays where it is.
312    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    // Breaking: persisted input_required API events can now omit the structured
326    // request. Older readers require it and fail to deserialize the event log;
327    // the shared read/write compatibility floor must advance with the revision.
328    if version < 39 {
329        connection.execute_batch(
330            "BEGIN IMMEDIATE;
331             UPDATE schema_compatibility SET minimum_compatible_version = 39 WHERE singleton = 1;
332             INSERT INTO schema_migrations(version, applied_at)
333                 VALUES (39, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
334             PRAGMA user_version = 39;
335             COMMIT;",
336        )?;
337    }
338    // Breaking: native child projections and relay observations must be preserved
339    // by every reader/writer; older builds cannot interpret their lifecycle.
340    if version < 40 {
341        connection.execute_batch(
342            "BEGIN IMMEDIATE;
343             CREATE TABLE native_agents (
344                 owner TEXT NOT NULL REFERENCES sessions(session_id) ON DELETE CASCADE,
345                 child TEXT NOT NULL,
346                 staging INTEGER NOT NULL CHECK(staging IN (0,1)),
347                 body TEXT NOT NULL CHECK(json_valid(body)),
348                 PRIMARY KEY(owner, child, staging)
349             ) STRICT;
350             CREATE TABLE native_agent_transcript (
351                 owner TEXT NOT NULL,
352                 child TEXT NOT NULL,
353                 staging INTEGER NOT NULL,
354                 stable_id TEXT NOT NULL,
355                 position INTEGER NOT NULL,
356                 body TEXT NOT NULL CHECK(json_valid(body)),
357                 PRIMARY KEY(owner, child, staging, stable_id),
358                 FOREIGN KEY(owner, child, staging) REFERENCES native_agents(owner, child, staging)
359                     ON DELETE CASCADE ON UPDATE CASCADE
360             ) STRICT;
361             CREATE INDEX native_agent_transcript_position ON native_agent_transcript(owner, child, staging, position);
362             CREATE TABLE native_agent_replay (
363                 owner TEXT PRIMARY KEY REFERENCES sessions(session_id) ON DELETE CASCADE
364             ) STRICT;
365             UPDATE schema_compatibility SET minimum_compatible_version = 40 WHERE singleton = 1;
366             INSERT INTO schema_migrations(version, applied_at)
367                 VALUES (40, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
368             PRAGMA user_version = 40;
369             COMMIT;",
370        )?;
371    }
372    // Breaking: clear-context relay commands/outcomes are persisted in event
373    // JSON. Older readers cannot decode them or honor the context boundary.
374    if version < 41 {
375        connection.execute_batch(
376            "BEGIN IMMEDIATE;
377             UPDATE schema_compatibility SET minimum_compatible_version = 41 WHERE singleton = 1;
378             INSERT INTO schema_migrations(version, applied_at)
379                 VALUES (41, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
380             PRAGMA user_version = 41;
381             COMMIT;",
382        )?;
383    }
384
385    // Breaking: older layout writers discard Browse identity and pin badges.
386    if version < 42 {
387        connection.execute_batch(
388            "BEGIN IMMEDIATE;
389             UPDATE schema_compatibility SET minimum_compatible_version = 42 WHERE singleton = 1;
390             INSERT INTO schema_migrations(version, applied_at)
391                 VALUES (42, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
392             PRAGMA user_version = 42;
393             COMMIT;",
394        )?;
395    }
396
397    // Breaking: durable steering commands/observations and native availability
398    // cannot be interpreted or preserved by older readers and writers.
399    if version < 43 {
400        connection.execute_batch("BEGIN IMMEDIATE;
401            UPDATE schema_compatibility SET minimum_compatible_version = 43 WHERE singleton = 1;
402            INSERT INTO schema_migrations(version, applied_at) VALUES (43, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
403            PRAGMA user_version = 43;
404            COMMIT;")?;
405    }
406
407    let recorded: Option<i64> =
408        connection.query_row("SELECT max(version) FROM schema_migrations", [], |row| {
409            row.get(0)
410        })?;
411    if recorded != Some(SCHEMA_VERSION) {
412        bail!(
413            "Mjolnir database migration ledger {:?} does not match schema {}",
414            recorded,
415            SCHEMA_VERSION
416        );
417    }
418    Ok(())
419}
420
421/// Create an empty store at the baseline revision in one immediate transaction.
422/// The revision is read again under the write lock, so a second process that
423/// raced to create the same store finds it already created.
424fn create_baseline_schema(connection: &Connection) -> Result<()> {
425    connection.execute_batch("BEGIN IMMEDIATE;")?;
426    let created = (|| -> Result<()> {
427        let version: i64 = connection.query_row("PRAGMA user_version", [], |row| row.get(0))?;
428        if version != 0 {
429            return Ok(());
430        }
431        connection.execute_batch(include_str!("baseline.sql"))?;
432        connection.execute(
433            "INSERT INTO schema_compatibility(singleton, minimum_compatible_version) VALUES (1, ?1)",
434            [BASELINE_MINIMUM_COMPATIBLE_VERSION],
435        )?;
436        connection.execute(
437            "INSERT INTO schema_migrations(version, applied_at)
438             VALUES (?1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
439            [BASELINE_SCHEMA_VERSION],
440        )?;
441        connection.pragma_update(None, "user_version", BASELINE_SCHEMA_VERSION)?;
442        Ok(())
443    })();
444    match created {
445        Ok(()) => connection
446            .execute_batch("COMMIT;")
447            .context("commit baseline database schema"),
448        Err(error) => {
449            if let Err(rollback) = connection.execute_batch("ROLLBACK;") {
450                tracing::warn!(%rollback, "could not roll back a failed baseline schema");
451            }
452            Err(error.context("create baseline database schema"))
453        }
454    }
455}
456
457#[cfg(test)]
458pub(super) fn advance_test_schema(path: &Path, revision: i64, minimum_compatible: i64) {
459    let connection = Connection::open(path).unwrap();
460    let transaction = connection.unchecked_transaction().unwrap();
461    transaction
462        .execute(
463            "UPDATE schema_compatibility SET minimum_compatible_version = ?1",
464            [minimum_compatible],
465        )
466        .unwrap();
467    transaction
468        .execute(
469            "INSERT INTO schema_migrations(version, applied_at) VALUES (?1, 'test')",
470            [revision],
471        )
472        .unwrap();
473    transaction
474        .pragma_update(None, "user_version", revision)
475        .unwrap();
476    transaction.commit().unwrap();
477    forget_verified_schema(path);
478}
479
480#[cfg(test)]
481mod reader_tests {
482    use super::*;
483
484    /// The oldest executable revision that can still read and write a store at
485    /// `SCHEMA_VERSION`. Migration 43 adds durable steering and availability.
486    const MINIMUM_COMPATIBLE_VERSION: i64 = 43;
487
488    /// Rewrites a store's recorded schema version the way another build's
489    /// migration ladder would, and forgets that this process verified it.
490    fn stamp_schema_version(path: &Path, version: i64) {
491        if version > SCHEMA_VERSION {
492            advance_test_schema(path, version, version);
493            return;
494        }
495        let connection = Connection::open(path).unwrap();
496        connection
497            .execute_batch(&format!("PRAGMA user_version = {version};"))
498            .unwrap();
499        connection
500            .execute(
501                "DELETE FROM schema_migrations WHERE version > ?1",
502                [version],
503            )
504            .unwrap();
505        if version == 30 {
506            connection
507                .execute(
508                    "UPDATE schema_compatibility SET minimum_compatible_version = 30 WHERE singleton = 1",
509                    [],
510                )
511                .unwrap();
512        }
513        drop(connection);
514        forget_verified_schema(path);
515    }
516
517    #[test]
518    fn native_agents_and_unstructured_input_raise_the_store_compatibility_floor() {
519        let directory = tempfile::tempdir().unwrap();
520        let path = directory.path().join("mj.sqlite3");
521        let connection = open_writer(&path).unwrap();
522        connection
523            .execute_batch(
524                "BEGIN IMMEDIATE;
525             DROP TABLE native_agent_transcript;
526             DROP TABLE native_agents;
527             DROP TABLE native_agent_replay;
528             DELETE FROM schema_migrations WHERE version >= 39;
529             UPDATE schema_compatibility SET minimum_compatible_version = 32;
530             PRAGMA user_version = 38;
531             COMMIT;",
532            )
533            .unwrap();
534        migrate_schema(&connection).unwrap();
535        let state = read_schema_state(&connection).unwrap();
536        assert_eq!(state.revision, SCHEMA_VERSION);
537        assert_eq!(state.minimum_compatible, Some(MINIMUM_COMPATIBLE_VERSION));
538        let event = ApiEventData::InputRequired {
539            request: None,
540            turn_id: Some(1),
541        };
542        #[derive(serde::Deserialize)]
543        struct LegacyInputEvent {
544            #[serde(rename = "request")]
545            _request: mj_core::elicitation::ElicitationRequest,
546        }
547        let encoded = serde_json::to_value(&event).unwrap();
548        assert!(serde_json::from_value::<LegacyInputEvent>(encoded["data"].clone()).is_err());
549    }
550
551    #[test]
552    fn older_readers_and_reopened_writers_preserve_a_compatible_future_schema() {
553        let directory = tempfile::tempdir().unwrap();
554        let path = directory.path().join("mj.sqlite3");
555        let connection = open_writer(&path).unwrap();
556        connection
557            .execute_batch(
558                "CREATE TABLE future_feature(value TEXT NOT NULL);
559                 INSERT INTO future_feature VALUES ('preserve me');",
560            )
561            .unwrap();
562        drop(connection);
563        advance_test_schema(&path, SCHEMA_VERSION + 1, SCHEMA_VERSION);
564
565        let reader = open_reader_strict(&path).unwrap();
566        assert_eq!(
567            reader
568                .query_row("SELECT value FROM future_feature", [], |row| row
569                    .get::<_, String>(0))
570                .unwrap(),
571            "preserve me"
572        );
573        assert!(reader.execute("DELETE FROM future_feature", []).is_err());
574        drop(reader);
575
576        // A repair would recreate this deliberately removed trigger. A future
577        // schema is authoritative even when it differs from our own repairs.
578        let raw = Connection::open(&path).unwrap();
579        raw.execute_batch("DROP TRIGGER api_session_error_updated;")
580            .unwrap();
581        drop(raw);
582        let writer = open_writer(&path).unwrap();
583        assert!(!writer.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name = 'api_session_error_updated')", [], |row| row.get::<_, bool>(0)).unwrap());
584        assert_eq!(
585            writer
586                .query_row("SELECT value FROM future_feature", [], |row| row
587                    .get::<_, String>(0))
588                .unwrap(),
589            "preserve me"
590        );
591        let state = read_schema_state(&writer).unwrap();
592        assert_eq!(state.revision, SCHEMA_VERSION + 1);
593        assert_eq!(state.minimum_compatible, Some(SCHEMA_VERSION));
594    }
595
596    #[test]
597    fn invalid_compatibility_metadata_refuses_readers_and_writers() {
598        for alteration in [
599            "DROP TABLE schema_compatibility",
600            "DELETE FROM schema_compatibility",
601            "PRAGMA ignore_check_constraints = ON; UPDATE schema_compatibility SET minimum_compatible_version = 0",
602            "UPDATE schema_compatibility SET minimum_compatible_version = 99999",
603            "PRAGMA ignore_check_constraints = ON; UPDATE schema_compatibility SET singleton = 2",
604            "PRAGMA ignore_check_constraints = ON; INSERT INTO schema_compatibility VALUES (2, 30)",
605            "DROP TABLE schema_compatibility; CREATE TABLE schema_compatibility(singleton, minimum_compatible_version); INSERT INTO schema_compatibility VALUES (1, 'invalid')",
606            "DELETE FROM schema_migrations WHERE version = (SELECT max(version) FROM schema_migrations)",
607        ] {
608            for future in [false, true] {
609                let directory = tempfile::tempdir().unwrap();
610                let path = directory.path().join("mj.sqlite3");
611                drop(open_writer(&path).unwrap());
612                if future {
613                    advance_test_schema(&path, SCHEMA_VERSION + 1, SCHEMA_VERSION);
614                }
615                let raw = Connection::open(&path).unwrap();
616                raw.execute_batch(alteration).unwrap();
617                let before: i64 = raw
618                    .query_row("PRAGMA schema_version", [], |row| row.get(0))
619                    .unwrap();
620                // Exercise the cached path as well as a fresh writer open.
621                for error in [
622                    open_reader_strict(&path).unwrap_err(),
623                    open_writer(&path).unwrap_err(),
624                ] {
625                    let mismatch = error.downcast_ref::<StoreSchemaMismatch>().unwrap();
626                    assert_eq!(
627                        mismatch.reason,
628                        StoreSchemaMismatchReason::InvalidCompatibilityMetadata,
629                        "{alteration}"
630                    );
631                }
632                forget_verified_schema(&path);
633                assert!(open_writer(&path).is_err(), "{alteration}");
634                let after: i64 = raw
635                    .query_row("PRAGMA schema_version", [], |row| row.get(0))
636                    .unwrap();
637                assert_eq!(
638                    before, after,
639                    "a rejected open repaired schema: {alteration}"
640                );
641            }
642        }
643    }
644
645    #[test]
646    fn a_failed_baseline_leaves_an_empty_store_that_a_retry_creates() {
647        let directory = tempfile::tempdir().unwrap();
648        let path = directory.path().join("mj.sqlite3");
649        let connection = Connection::open(&path).unwrap();
650        // A table the baseline also creates makes its batch fail part way.
651        connection
652            .execute_batch("CREATE TABLE workspaces(conflict TEXT)")
653            .unwrap();
654
655        let error = migrate_schema(&connection).unwrap_err();
656
657        assert!(format!("{error:#}").contains("create baseline database schema"));
658        assert!(
659            connection.is_autocommit(),
660            "the failed baseline left a transaction open"
661        );
662        assert_eq!(read_schema_state(&connection).unwrap().revision, 0);
663        let tables: i64 = connection
664            .query_row(
665                "SELECT count(*) FROM sqlite_schema WHERE type = 'table'",
666                [],
667                |row| row.get(0),
668            )
669            .unwrap();
670        assert_eq!(tables, 1, "only the conflicting table remains");
671
672        connection.execute_batch("DROP TABLE workspaces").unwrap();
673        drop(connection);
674        let writer = open_writer(&path).unwrap();
675        let state = read_schema_state(&writer).unwrap();
676        assert_eq!(state.revision, SCHEMA_VERSION);
677        assert_eq!(state.minimum_compatible, Some(43));
678    }
679
680    #[test]
681    fn a_store_from_before_the_baseline_is_refused_with_upgrade_advice() {
682        let directory = tempfile::tempdir().unwrap();
683        let path = directory.path().join("mj.sqlite3");
684        let connection = Connection::open(&path).unwrap();
685        connection
686            .execute_batch(&format!(
687                "PRAGMA user_version = {};",
688                COMPATIBILITY_METADATA_VERSION - 1
689            ))
690            .unwrap();
691        drop(connection);
692
693        let error = open_writer(&path).unwrap_err();
694
695        let message = format!("{error:#}");
696        assert!(message.contains("older than 2.7.2"), "{message}");
697        assert!(
698            message.contains("from 2.7.2 through 2.9.x"),
699            "advice must name the closed range of releases that can migrate: {message}"
700        );
701    }
702
703    /// A store ahead of this build cannot be fixed by starting a daemon of
704    /// this build, so the reader must not say so. This is the message the
705    /// incident in #24 printed twice a second for an hour.
706    #[test]
707    fn strict_reader_reports_a_newer_store_without_blaming_the_daemon() {
708        let directory = tempfile::tempdir().unwrap();
709        let path = directory.path().join("mj.sqlite3");
710        drop(open_writer(&path).unwrap());
711        stamp_schema_version(&path, SCHEMA_VERSION + 1);
712
713        let error = open_reader_strict(&path).unwrap_err();
714
715        let mismatch = error
716            .chain()
717            .find_map(|cause| cause.downcast_ref::<StoreSchemaMismatch>())
718            .expect("the reader reports the mismatch as a typed cause");
719        assert_eq!(mismatch.found, SCHEMA_VERSION + 1);
720        assert_eq!(mismatch.supported, SCHEMA_VERSION);
721        let message = mismatch.to_string();
722        assert!(message.contains("upgrade Mjolnir"), "got {message}");
723        assert!(
724            !message.contains("start the Mjolnir daemon"),
725            "got {message}"
726        );
727    }
728
729    /// A store behind this build keeps the advice that works, verbatim, so
730    /// existing log greps and runbooks keep matching.
731    #[test]
732    fn strict_reader_keeps_the_migrate_advice_when_the_store_is_behind() {
733        let directory = tempfile::tempdir().unwrap();
734        let path = directory.path().join("mj.sqlite3");
735        drop(open_writer(&path).unwrap());
736        let raw = Connection::open(&path).unwrap();
737        raw.execute_batch(&format!(
738            "UPDATE schema_compatibility SET minimum_compatible_version = {0};
739             DELETE FROM schema_migrations WHERE version > {0};
740             INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES ({0}, 'test');
741             PRAGMA user_version = {0};",
742            SCHEMA_VERSION - 1
743        ))
744        .unwrap();
745        drop(raw);
746
747        let error = open_reader_strict(&path).unwrap_err();
748
749        let mismatch = error
750            .chain()
751            .find_map(|cause| cause.downcast_ref::<StoreSchemaMismatch>())
752            .expect("the reader reports the mismatch as a typed cause");
753        assert_eq!(
754            mismatch.to_string(),
755            format!(
756                "Mjolnir database schema {} is not the supported schema {SCHEMA_VERSION}; \
757                 start the Mjolnir daemon to migrate it",
758                SCHEMA_VERSION - 1
759            )
760        );
761    }
762
763    #[test]
764    fn strict_reader_rejects_mutation() {
765        let directory = tempfile::tempdir().unwrap();
766        let path = directory.path().join("mj.sqlite3");
767        drop(open_writer(&path).unwrap());
768
769        let reader = open_reader_strict(&path).unwrap();
770        let error = reader
771            .execute("CREATE TABLE forbidden(value TEXT)", [])
772            .unwrap_err();
773        assert!(
774            matches!(
775                error.sqlite_error_code(),
776                Some(rusqlite::ErrorCode::ReadOnly)
777            ),
778            "unexpected mutation error: {error}"
779        );
780    }
781}
mj_controller/database/schema.rs

mj_controller/database/
schema.rs
