commit 2a94866a7135c6d2a9ca61e58f62060e8727dab0
parent f687ab0896349ca0ba536e50c316edfa535d9da1
Author: triesap <tyson@radroots.org>
Date: Sun, 2 Aug 2026 19:36:08 +0000
storage-sqlite: implement transactional schema migrations
- bind runtime and private databases to governed application and schema versions
- apply every pending forward migration inside one immediate transaction
- reject read-only upgrades, unknown catalogs, wrong identities, and newer schemas
- prove every supported upgrade suffix, data preservation, and whole-plan rollback
Diffstat:
6 files changed, 739 insertions(+), 8 deletions(-)
diff --git a/contracts/storage/private_schema_v1.toml b/contracts/storage/private_schema_v1.toml
@@ -1,5 +1,6 @@
schema_version = 1
database = "private.sqlite"
+application_id = 1380208722 # ASCII "RDPR"
minimum_version = 1
current_version = 1
migration_name = "private_artifacts"
diff --git a/contracts/storage/runtime_schema_v1.toml b/contracts/storage/runtime_schema_v1.toml
@@ -1,5 +1,6 @@
schema_version = 1
database = "runtime.sqlite"
+application_id = 1380209236 # ASCII "RDRT"
minimum_version = 1
current_version = 5
migration_name = "projection_metadata"
diff --git a/crates/storage_sqlite/src/migration.rs b/crates/storage_sqlite/src/migration.rs
@@ -1,6 +1,653 @@
//! Governed SQLite schema migration boundary.
+use crate::{Error, OpenMode};
+use sqlx::{Connection, Row, SqliteConnection};
+
/// Versioned schema authority for `private.sqlite`.
pub mod private;
/// Versioned schema authority for `runtime.sqlite`.
pub mod runtime;
+
+const RUNTIME_DATABASE: &str = "runtime.sqlite";
+const PRIVATE_DATABASE: &str = "private.sqlite";
+const RUNTIME_APPLICATION_ID: u32 = 1_380_209_236;
+const PRIVATE_APPLICATION_ID: u32 = 1_380_208_722;
+
+const SET_RUNTIME_APPLICATION_ID: &str = "PRAGMA application_id = 1380209236";
+const SET_PRIVATE_APPLICATION_ID: &str = "PRAGMA application_id = 1380208722";
+
+#[derive(Clone, Copy)]
+struct MigrationStep {
+ version: u32,
+ sql: &'static str,
+ owned_objects: &'static [&'static str],
+}
+
+struct MigrationPlan {
+ database: &'static str,
+ application_id: u32,
+ set_application_id_sql: &'static str,
+ minimum_version: u32,
+ current_version: u32,
+ steps: Vec<MigrationStep>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct MigrationReport {
+ initial_version: u32,
+ final_version: u32,
+ applied: u32,
+}
+
+impl MigrationReport {
+ #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint.
+ pub(crate) const fn initial_version(self) -> u32 {
+ self.initial_version
+ }
+
+ #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint.
+ pub(crate) const fn final_version(self) -> u32 {
+ self.final_version
+ }
+
+ #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint.
+ pub(crate) const fn applied(self) -> u32 {
+ self.applied
+ }
+}
+
+#[allow(dead_code)] // Wired into the public open lifecycle in its ordered RCL checkpoint.
+pub(crate) async fn migrate_runtime(
+ connection: &mut SqliteConnection,
+ mode: OpenMode,
+) -> Result<MigrationReport, Error> {
+ let steps = runtime::MIGRATIONS
+ .iter()
+ .map(|migration| {
+ Ok(MigrationStep {
+ version: migration.version(),
+ sql: runtime::migration_sql(migration.version()).ok_or(
+ Error::SchemaMetadataUnavailable {
+ database: RUNTIME_DATABASE,
+ },
+ )?,
+ owned_objects: migration.owned_objects(),
+ })
+ })
+ .collect::<Result<Vec<_>, Error>>()?;
+ migrate(
+ connection,
+ mode,
+ &MigrationPlan {
+ database: RUNTIME_DATABASE,
+ application_id: RUNTIME_APPLICATION_ID,
+ set_application_id_sql: SET_RUNTIME_APPLICATION_ID,
+ minimum_version: runtime::MINIMUM_VERSION,
+ current_version: runtime::CURRENT_VERSION,
+ steps,
+ },
+ )
+ .await
+}
+
+#[allow(dead_code)] // Wired into the public open lifecycle in its ordered RCL checkpoint.
+pub(crate) async fn migrate_private(
+ connection: &mut SqliteConnection,
+ mode: OpenMode,
+) -> Result<MigrationReport, Error> {
+ let steps = private::MIGRATIONS
+ .iter()
+ .map(|migration| {
+ Ok(MigrationStep {
+ version: migration.version(),
+ sql: private::migration_sql(migration.version()).ok_or(
+ Error::SchemaMetadataUnavailable {
+ database: PRIVATE_DATABASE,
+ },
+ )?,
+ owned_objects: migration.owned_objects(),
+ })
+ })
+ .collect::<Result<Vec<_>, Error>>()?;
+ migrate(
+ connection,
+ mode,
+ &MigrationPlan {
+ database: PRIVATE_DATABASE,
+ application_id: PRIVATE_APPLICATION_ID,
+ set_application_id_sql: SET_PRIVATE_APPLICATION_ID,
+ minimum_version: private::MINIMUM_VERSION,
+ current_version: private::CURRENT_VERSION,
+ steps,
+ },
+ )
+ .await
+}
+
+async fn migrate(
+ connection: &mut SqliteConnection,
+ mode: OpenMode,
+ plan: &MigrationPlan,
+) -> Result<MigrationReport, Error> {
+ validate_plan(plan)?;
+ let initial = metadata(connection, plan.database).await?;
+ validate_metadata(plan, initial)?;
+ validate_catalog(connection, plan, initial.version).await?;
+
+ if initial.version == plan.current_version {
+ return Ok(MigrationReport {
+ initial_version: initial.version,
+ final_version: initial.version,
+ applied: 0,
+ });
+ }
+ if !mode.is_writable() {
+ return Err(Error::SchemaMigrationRequired {
+ database: plan.database,
+ current: plan.current_version,
+ actual: initial.version,
+ });
+ }
+
+ let mut transaction = connection
+ .begin_with("BEGIN IMMEDIATE")
+ .await
+ .map_err(|_| Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: initial.version.saturating_add(1),
+ })?;
+ let transactional = match metadata(&mut transaction, plan.database).await {
+ Ok(metadata) => metadata,
+ Err(error) => {
+ let _rollback = transaction.rollback().await;
+ return Err(error);
+ }
+ };
+ if transactional != initial {
+ let _rollback = transaction.rollback().await;
+ return Err(Error::SchemaMetadataUnavailable {
+ database: plan.database,
+ });
+ }
+
+ if initial.version == 0
+ && sqlx::raw_sql(plan.set_application_id_sql)
+ .execute(&mut *transaction)
+ .await
+ .is_err()
+ {
+ let error = Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: 1,
+ };
+ let _rollback = transaction.rollback().await;
+ return Err(error);
+ }
+
+ let mut applied = 0_u32;
+ for step in plan
+ .steps
+ .iter()
+ .filter(|step| step.version > initial.version)
+ {
+ let version_sql =
+ set_user_version_sql(step.version).ok_or(Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: step.version,
+ })?;
+ if sqlx::raw_sql(step.sql)
+ .execute(&mut *transaction)
+ .await
+ .is_err()
+ {
+ let error = Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: step.version,
+ };
+ let _rollback = transaction.rollback().await;
+ return Err(error);
+ }
+ if sqlx::raw_sql(version_sql)
+ .execute(&mut *transaction)
+ .await
+ .is_err()
+ {
+ let error = Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: step.version,
+ };
+ let _rollback = transaction.rollback().await;
+ return Err(error);
+ }
+ if validate_exact_catalog(
+ &mut transaction,
+ plan.database,
+ step.version,
+ step.owned_objects,
+ )
+ .await
+ .is_err()
+ {
+ let error = Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: step.version,
+ };
+ let _rollback = transaction.rollback().await;
+ return Err(error);
+ }
+ applied = applied.saturating_add(1);
+ }
+ transaction
+ .commit()
+ .await
+ .map_err(|_| Error::SchemaMigrationFailed {
+ database: plan.database,
+ target_version: plan.current_version,
+ })?;
+ Ok(MigrationReport {
+ initial_version: initial.version,
+ final_version: plan.current_version,
+ applied,
+ })
+}
+
+#[derive(Clone, Copy, Eq, PartialEq)]
+struct SchemaMetadata {
+ application_id: u32,
+ version: u32,
+}
+
+async fn metadata(
+ connection: &mut SqliteConnection,
+ database: &'static str,
+) -> Result<SchemaMetadata, Error> {
+ let application_id = sqlx::query_scalar::<_, i64>("PRAGMA application_id")
+ .fetch_one(&mut *connection)
+ .await
+ .map_err(|_| Error::SchemaMetadataUnavailable { database })?;
+ let version = sqlx::query_scalar::<_, i64>("PRAGMA user_version")
+ .fetch_one(&mut *connection)
+ .await
+ .map_err(|_| Error::SchemaMetadataUnavailable { database })?;
+ Ok(SchemaMetadata {
+ application_id: u32::try_from(application_id)
+ .map_err(|_| Error::SchemaMetadataUnavailable { database })?,
+ version: u32::try_from(version)
+ .map_err(|_| Error::SchemaMetadataUnavailable { database })?,
+ })
+}
+
+fn validate_plan(plan: &MigrationPlan) -> Result<(), Error> {
+ let valid = plan.minimum_version > 0
+ && plan.minimum_version <= plan.current_version
+ && plan.current_version <= 5
+ && plan.steps.len() == usize::try_from(plan.current_version).unwrap_or(usize::MAX)
+ && plan
+ .steps
+ .iter()
+ .enumerate()
+ .all(|(index, step)| step.version == u32::try_from(index + 1).unwrap_or(u32::MAX));
+ if valid {
+ Ok(())
+ } else {
+ Err(Error::SchemaMetadataUnavailable {
+ database: plan.database,
+ })
+ }
+}
+
+fn validate_metadata(plan: &MigrationPlan, metadata: SchemaMetadata) -> Result<(), Error> {
+ if metadata.version > plan.current_version {
+ return Err(Error::SchemaTooNew {
+ database: plan.database,
+ supported: plan.current_version,
+ actual: metadata.version,
+ });
+ }
+ if metadata.version > 0 && metadata.version < plan.minimum_version {
+ return Err(Error::SchemaTooOld {
+ database: plan.database,
+ minimum: plan.minimum_version,
+ actual: metadata.version,
+ });
+ }
+ let expected_application_id = if metadata.version == 0 {
+ 0
+ } else {
+ plan.application_id
+ };
+ if metadata.application_id != expected_application_id {
+ return Err(Error::SchemaIdentityMismatch {
+ database: plan.database,
+ expected: expected_application_id,
+ actual: metadata.application_id,
+ });
+ }
+ Ok(())
+}
+
+async fn validate_catalog(
+ connection: &mut SqliteConnection,
+ plan: &MigrationPlan,
+ version: u32,
+) -> Result<(), Error> {
+ let expected = if version == 0 {
+ &[][..]
+ } else {
+ plan.steps
+ .get(
+ usize::try_from(version - 1).map_err(|_| Error::SchemaCatalogMismatch {
+ database: plan.database,
+ version,
+ })?,
+ )
+ .ok_or(Error::SchemaCatalogMismatch {
+ database: plan.database,
+ version,
+ })?
+ .owned_objects
+ };
+ validate_exact_catalog(connection, plan.database, version, expected).await
+}
+
+async fn validate_exact_catalog(
+ connection: &mut SqliteConnection,
+ database: &'static str,
+ version: u32,
+ expected: &[&str],
+) -> Result<(), Error> {
+ let rows = sqlx::query(
+ "SELECT name FROM sqlite_schema
+ WHERE name NOT LIKE 'sqlite_%'
+ ORDER BY name",
+ )
+ .fetch_all(&mut *connection)
+ .await
+ .map_err(|_| Error::SchemaCatalogMismatch { database, version })?;
+ let actual = rows
+ .iter()
+ .map(|row| row.get::<String, _>("name"))
+ .collect::<Vec<_>>();
+ if actual
+ .iter()
+ .map(String::as_str)
+ .eq(expected.iter().copied())
+ {
+ Ok(())
+ } else if version == 0 {
+ Err(Error::UnrecognizedSchema { database })
+ } else {
+ Err(Error::SchemaCatalogMismatch { database, version })
+ }
+}
+
+const fn set_user_version_sql(version: u32) -> Option<&'static str> {
+ match version {
+ 1 => Some("PRAGMA user_version = 1"),
+ 2 => Some("PRAGMA user_version = 2"),
+ 3 => Some("PRAGMA user_version = 3"),
+ 4 => Some("PRAGMA user_version = 4"),
+ 5 => Some("PRAGMA user_version = 5"),
+ _ => None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ const TEST_V1_OBJECTS: &[&str] = &["radroots_test_one"];
+ const TEST_V2_OBJECTS: &[&str] = &["radroots_test_one", "radroots_test_two"];
+
+ async fn connection() -> SqliteConnection {
+ SqliteConnection::connect("sqlite::memory:")
+ .await
+ .expect("memory SQLite")
+ }
+
+ async fn pragma(connection: &mut SqliteConnection, name: &str) -> i64 {
+ let sql = match name {
+ "application_id" => "PRAGMA application_id",
+ "user_version" => "PRAGMA user_version",
+ _ => panic!("unsupported test pragma"),
+ };
+ sqlx::query_scalar(sql)
+ .fetch_one(connection)
+ .await
+ .expect("pragma")
+ }
+
+ async fn establish_runtime_version(connection: &mut SqliteConnection, version: u32) {
+ for migration_version in 1..=version {
+ sqlx::raw_sql(
+ runtime::migration_sql(migration_version).expect("registered runtime SQL"),
+ )
+ .execute(&mut *connection)
+ .await
+ .expect("runtime migration");
+ }
+ sqlx::raw_sql(SET_RUNTIME_APPLICATION_ID)
+ .execute(&mut *connection)
+ .await
+ .expect("runtime application id");
+ sqlx::raw_sql(set_user_version_sql(version).expect("version pragma"))
+ .execute(&mut *connection)
+ .await
+ .expect("runtime user version");
+ }
+
+ #[tokio::test]
+ async fn fresh_runtime_and_private_schemas_migrate_to_exact_current_versions() {
+ let mut runtime_connection = connection().await;
+ let runtime_report = migrate_runtime(&mut runtime_connection, OpenMode::Create)
+ .await
+ .expect("runtime migrations");
+ assert_eq!(runtime_report.initial_version(), 0);
+ assert_eq!(runtime_report.final_version(), runtime::CURRENT_VERSION);
+ assert_eq!(runtime_report.applied(), runtime::CURRENT_VERSION);
+ assert_eq!(
+ pragma(&mut runtime_connection, "application_id").await,
+ i64::from(RUNTIME_APPLICATION_ID)
+ );
+ assert_eq!(
+ pragma(&mut runtime_connection, "user_version").await,
+ i64::from(runtime::CURRENT_VERSION)
+ );
+ assert_eq!(
+ migrate_runtime(&mut runtime_connection, OpenMode::ReadOnly)
+ .await
+ .expect("current read-only runtime"),
+ MigrationReport {
+ initial_version: runtime::CURRENT_VERSION,
+ final_version: runtime::CURRENT_VERSION,
+ applied: 0,
+ }
+ );
+
+ let mut private_connection = connection().await;
+ let private_report = migrate_private(&mut private_connection, OpenMode::Create)
+ .await
+ .expect("private migrations");
+ assert_eq!(private_report.initial_version(), 0);
+ assert_eq!(private_report.final_version(), private::CURRENT_VERSION);
+ assert_eq!(private_report.applied(), private::CURRENT_VERSION);
+ assert_eq!(
+ pragma(&mut private_connection, "application_id").await,
+ i64::from(PRIVATE_APPLICATION_ID)
+ );
+ assert_eq!(
+ pragma(&mut private_connection, "user_version").await,
+ i64::from(private::CURRENT_VERSION)
+ );
+ }
+
+ #[tokio::test]
+ async fn recognized_runtime_schema_upgrades_forward_and_preserves_data() {
+ let mut connection = connection().await;
+ establish_runtime_version(&mut connection, 1).await;
+ sqlx::query(
+ "INSERT INTO radroots_runtime_source_generations (
+ generation, state, created_at_unix_ms
+ ) VALUES (?, 'active', 10)",
+ )
+ .bind([7_u8; 32].as_slice())
+ .execute(&mut connection)
+ .await
+ .expect("v1 data");
+
+ let report = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting)
+ .await
+ .expect("forward migration");
+ assert_eq!(report.initial_version(), 1);
+ assert_eq!(report.final_version(), 5);
+ assert_eq!(report.applied(), 4);
+ assert_eq!(
+ sqlx::query_scalar::<_, i64>(
+ "SELECT COUNT(*) FROM radroots_runtime_source_generations",
+ )
+ .fetch_one(&mut connection)
+ .await
+ .expect("preserved data"),
+ 1
+ );
+ }
+
+ #[tokio::test]
+ async fn every_recognized_runtime_version_applies_exactly_the_pending_suffix() {
+ for initial_version in 1..=runtime::CURRENT_VERSION {
+ let mut connection = connection().await;
+ establish_runtime_version(&mut connection, initial_version).await;
+ let report = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting)
+ .await
+ .expect("recognized forward migration");
+ assert_eq!(report.initial_version(), initial_version);
+ assert_eq!(report.final_version(), runtime::CURRENT_VERSION);
+ assert_eq!(report.applied(), runtime::CURRENT_VERSION - initial_version);
+ }
+ }
+
+ #[tokio::test]
+ async fn read_only_old_schema_requires_migration_without_mutating() {
+ let mut connection = connection().await;
+ establish_runtime_version(&mut connection, 1).await;
+ assert!(matches!(
+ migrate_runtime(&mut connection, OpenMode::ReadOnly).await,
+ Err(Error::SchemaMigrationRequired {
+ database: RUNTIME_DATABASE,
+ current: 5,
+ actual: 1,
+ })
+ ));
+ assert_eq!(pragma(&mut connection, "user_version").await, 1);
+ }
+
+ #[tokio::test]
+ async fn newer_wrong_identity_and_unversioned_nonempty_schemas_fail_closed() {
+ let mut newer = connection().await;
+ sqlx::raw_sql(SET_RUNTIME_APPLICATION_ID)
+ .execute(&mut newer)
+ .await
+ .expect("application id");
+ sqlx::raw_sql("PRAGMA user_version = 6")
+ .execute(&mut newer)
+ .await
+ .expect("newer version");
+ assert!(matches!(
+ migrate_runtime(&mut newer, OpenMode::ReadWriteExisting).await,
+ Err(Error::SchemaTooNew {
+ database: RUNTIME_DATABASE,
+ supported: 5,
+ actual: 6,
+ })
+ ));
+ assert_eq!(pragma(&mut newer, "user_version").await, 6);
+
+ let mut wrong_identity = connection().await;
+ establish_runtime_version(&mut wrong_identity, 1).await;
+ assert!(matches!(
+ migrate_private(&mut wrong_identity, OpenMode::ReadWriteExisting).await,
+ Err(Error::SchemaIdentityMismatch {
+ database: PRIVATE_DATABASE,
+ expected: PRIVATE_APPLICATION_ID,
+ actual: RUNTIME_APPLICATION_ID,
+ })
+ ));
+
+ let mut unknown = connection().await;
+ sqlx::query("CREATE TABLE unrelated(value INTEGER)")
+ .execute(&mut unknown)
+ .await
+ .expect("unknown table");
+ assert!(matches!(
+ migrate_runtime(&mut unknown, OpenMode::Create).await,
+ Err(Error::UnrecognizedSchema {
+ database: RUNTIME_DATABASE,
+ })
+ ));
+ assert_eq!(pragma(&mut unknown, "application_id").await, 0);
+ assert_eq!(pragma(&mut unknown, "user_version").await, 0);
+ }
+
+ #[tokio::test]
+ async fn current_version_with_incomplete_catalog_is_rejected() {
+ let mut connection = connection().await;
+ sqlx::raw_sql(SET_PRIVATE_APPLICATION_ID)
+ .execute(&mut connection)
+ .await
+ .expect("private application id");
+ sqlx::raw_sql("PRAGMA user_version = 1")
+ .execute(&mut connection)
+ .await
+ .expect("private version");
+ assert!(matches!(
+ migrate_private(&mut connection, OpenMode::ReadOnly).await,
+ Err(Error::SchemaCatalogMismatch {
+ database: PRIVATE_DATABASE,
+ version: 1,
+ })
+ ));
+ }
+
+ #[tokio::test]
+ async fn any_failed_step_rolls_back_the_entire_pending_plan() {
+ let mut connection = connection().await;
+ let plan = MigrationPlan {
+ database: "test.sqlite",
+ application_id: 4_242,
+ set_application_id_sql: "PRAGMA application_id = 4242",
+ minimum_version: 1,
+ current_version: 2,
+ steps: vec![
+ MigrationStep {
+ version: 1,
+ sql: "CREATE TABLE radroots_test_one(value INTEGER)",
+ owned_objects: TEST_V1_OBJECTS,
+ },
+ MigrationStep {
+ version: 2,
+ sql: "CREATE TABLE radroots_test_two(value INTEGER);
+ INSERT INTO radroots_missing VALUES (1)",
+ owned_objects: TEST_V2_OBJECTS,
+ },
+ ],
+ };
+ assert!(matches!(
+ migrate(&mut connection, OpenMode::Create, &plan).await,
+ Err(Error::SchemaMigrationFailed {
+ database: "test.sqlite",
+ target_version: 2,
+ })
+ ));
+ assert_eq!(pragma(&mut connection, "application_id").await, 0);
+ assert_eq!(pragma(&mut connection, "user_version").await, 0);
+ assert_eq!(
+ sqlx::query_scalar::<_, i64>(
+ "SELECT COUNT(*) FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'",
+ )
+ .fetch_one(&mut connection)
+ .await
+ .expect("rolled-back catalog"),
+ 0
+ );
+ }
+}
diff --git a/crates/storage_sqlite/src/migration/private/mod.rs b/crates/storage_sqlite/src/migration/private/mod.rs
@@ -8,7 +8,6 @@ pub const MINIMUM_VERSION: u32 = 1;
/// Current private schema version created by this package.
pub const CURRENT_VERSION: u32 = 1;
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const PRIVATE_V1_SQL: &str = include_str!("0001_private.up.sql");
/// Stable, non-SQL description of one forward private migration.
@@ -56,7 +55,6 @@ pub const MIGRATIONS: &[MigrationDescriptor] = &[MigrationDescriptor {
owned_objects: PRIVATE_V1_OBJECTS,
}];
-#[allow(dead_code)] // Keeps raw SQL crate-private until the migration executor is installed.
pub(crate) const fn migration_sql(version: u32) -> Option<&'static str> {
match version {
1 => Some(PRIVATE_V1_SQL),
@@ -78,6 +76,7 @@ mod tests {
struct PlanSnapshot {
schema_version: u32,
database: String,
+ application_id: u32,
minimum_version: u32,
current_version: u32,
migration_name: String,
@@ -104,6 +103,7 @@ mod tests {
let migration = MIGRATIONS[0];
assert_eq!(snapshot.schema_version, 1);
assert_eq!(snapshot.database, "private.sqlite");
+ assert_eq!(snapshot.application_id, 1_380_208_722);
assert_eq!(snapshot.minimum_version, MINIMUM_VERSION);
assert_eq!(snapshot.current_version, CURRENT_VERSION);
assert_eq!(snapshot.migration_name, migration.name());
diff --git a/crates/storage_sqlite/src/migration/runtime/mod.rs b/crates/storage_sqlite/src/migration/runtime/mod.rs
@@ -8,15 +8,10 @@ pub const MINIMUM_VERSION: u32 = 1;
/// Current runtime schema version created by this package.
pub const CURRENT_VERSION: u32 = 5;
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const RUNTIME_V1_SQL: &str = include_str!("0001_runtime.up.sql");
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const CANONICAL_EVENT_STORAGE_V2_SQL: &str = include_str!("0002_canonical_event_storage.up.sql");
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const OPERATION_JOURNAL_V3_SQL: &str = include_str!("0003_operation_journal.up.sql");
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const OUTBOX_DELIVERY_EVIDENCE_V4_SQL: &str = include_str!("0004_outbox_delivery_evidence.up.sql");
-#[allow(dead_code)] // Consumed by the migration executor introduced in its ordered RCL step.
const PROJECTION_METADATA_V5_SQL: &str = include_str!("0005_projection_metadata.up.sql");
/// Stable, non-SQL description of one forward runtime migration.
@@ -176,7 +171,6 @@ pub const MIGRATIONS: &[MigrationDescriptor] = &[
},
];
-#[allow(dead_code)] // Keeps raw SQL crate-private until the migration executor is installed.
pub(crate) const fn migration_sql(version: u32) -> Option<&'static str> {
match version {
1 => Some(RUNTIME_V1_SQL),
@@ -202,6 +196,7 @@ mod tests {
struct PlanSnapshot {
schema_version: u32,
database: String,
+ application_id: u32,
minimum_version: u32,
current_version: u32,
migration_name: String,
@@ -230,6 +225,7 @@ mod tests {
let migration = MIGRATIONS[4];
assert_eq!(snapshot.schema_version, 1);
assert_eq!(snapshot.database, "runtime.sqlite");
+ assert_eq!(snapshot.application_id, 1_380_209_236);
assert_eq!(snapshot.minimum_version, MINIMUM_VERSION);
assert_eq!(snapshot.current_version, CURRENT_VERSION);
assert_eq!(snapshot.migration_name, migration.name());
diff --git a/crates/storage_sqlite/src/open.rs b/crates/storage_sqlite/src/open.rs
@@ -181,6 +181,40 @@ pub enum Error {
path: PathBuf,
source: std::io::Error,
},
+ SchemaMetadataUnavailable {
+ database: &'static str,
+ },
+ SchemaIdentityMismatch {
+ database: &'static str,
+ expected: u32,
+ actual: u32,
+ },
+ SchemaTooOld {
+ database: &'static str,
+ minimum: u32,
+ actual: u32,
+ },
+ SchemaTooNew {
+ database: &'static str,
+ supported: u32,
+ actual: u32,
+ },
+ SchemaMigrationRequired {
+ database: &'static str,
+ current: u32,
+ actual: u32,
+ },
+ UnrecognizedSchema {
+ database: &'static str,
+ },
+ SchemaCatalogMismatch {
+ database: &'static str,
+ version: u32,
+ },
+ SchemaMigrationFailed {
+ database: &'static str,
+ target_version: u32,
+ },
}
impl fmt::Display for Error {
@@ -255,6 +289,58 @@ impl fmt::Display for Error {
"failed to release SQLite writer lock: {}",
path.display()
),
+ Self::SchemaMetadataUnavailable { database } => {
+ write!(formatter, "failed to inspect {database} schema metadata")
+ }
+ Self::SchemaIdentityMismatch {
+ database,
+ expected,
+ actual,
+ } => write!(
+ formatter,
+ "{database} application id {actual} does not match required id {expected}"
+ ),
+ Self::SchemaTooOld {
+ database,
+ minimum,
+ actual,
+ } => write!(
+ formatter,
+ "{database} schema version {actual} is older than supported version {minimum}"
+ ),
+ Self::SchemaTooNew {
+ database,
+ supported,
+ actual,
+ } => write!(
+ formatter,
+ "{database} schema version {actual} is newer than supported version {supported}"
+ ),
+ Self::SchemaMigrationRequired {
+ database,
+ current,
+ actual,
+ } => write!(
+ formatter,
+ "{database} schema version {actual} requires writable migration to {current}"
+ ),
+ Self::UnrecognizedSchema { database } => {
+ write!(
+ formatter,
+ "{database} has an unrecognized unversioned schema"
+ )
+ }
+ Self::SchemaCatalogMismatch { database, version } => write!(
+ formatter,
+ "{database} object catalog does not match schema version {version}"
+ ),
+ Self::SchemaMigrationFailed {
+ database,
+ target_version,
+ } => write!(
+ formatter,
+ "{database} migration to schema version {target_version} failed"
+ ),
}
}
}
