diff --git a/Cargo.lock b/Cargo.lock index 242b12da7f..44c3c565be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2563,6 +2563,7 @@ dependencies = [ "sqlx", "tempfile", "tokio", + "tracing", ] [[package]] diff --git a/lib/crates/fabro-db/Cargo.toml b/lib/crates/fabro-db/Cargo.toml index c29362b0fc..34963bb7a6 100644 --- a/lib/crates/fabro-db/Cargo.toml +++ b/lib/crates/fabro-db/Cargo.toml @@ -16,6 +16,7 @@ workspace = true anyhow.workspace = true sqlx.workspace = true tokio.workspace = true +tracing.workspace = true [dev-dependencies] tempfile = "3" diff --git a/lib/crates/fabro-db/src/lib.rs b/lib/crates/fabro-db/src/lib.rs index fcf5fa8753..f895ee07a0 100644 --- a/lib/crates/fabro-db/src/lib.rs +++ b/lib/crates/fabro-db/src/lib.rs @@ -1,10 +1,12 @@ -use std::path::Path; +use std::collections::HashSet; +use std::path::{Path, PathBuf}; use std::time::Duration; use anyhow::Context as _; -use sqlx::migrate::Migrator; +use sqlx::migrate::{Migrate as _, Migrator}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; use tokio::fs; +use tracing::info; pub type DbPool = sqlx::SqlitePool; @@ -41,12 +43,95 @@ impl Database { } pub async fn migrate(&self) -> anyhow::Result<()> { + self.snapshot_before_new_migrations() + .await + .context("snapshotting SQLite database before migrations")?; MIGRATOR .run(&self.pool) .await .context("running SQLite migrations") } + /// Copy the database aside before applying migrations it has not seen. + /// + /// A binary downgrade after new migrations have been applied fails sqlx's + /// startup validation (`migration N was previously applied but is missing + /// in the resolved migrations`), so the snapshot written to + /// [`pre_migration_snapshot_path`] is the operator's rollback artifact: + /// stop the server, replace the database file with the snapshot (and + /// delete any `-wal`/`-shm` siblings), and the previous binary boots + /// again. Writes made after the upgrade are lost on rollback, as with any + /// point-in-time restore. + /// + /// The snapshot is only taken when the database has applied migrations + /// before (a fresh database has nothing worth preserving) and at least + /// one bundled migration is pending, so the file always holds the state + /// from immediately before the most recent schema change. Failing to + /// write the snapshot fails the migration: no rollback artifact, no + /// schema change. + async fn snapshot_before_new_migrations(&self) -> anyhow::Result<()> { + let applied = applied_migration_versions(&self.pool).await?; + let has_pending = MIGRATOR + .iter() + .any(|migration| !applied.contains(&migration.version)); + if applied.is_empty() || !has_pending { + return Ok(()); + } + + let connect_options = self.pool.connect_options(); + let database_path = connect_options.get_filename(); + let snapshot_path = pre_migration_snapshot_path(database_path); + + // VACUUM INTO produces a consistent single-file copy from the live + // pool, so the snapshot needs no -wal/-shm siblings to restore. It + // writes to a staging file that is renamed into place afterwards, so + // a failure mid-copy never leaves a partial file at the snapshot + // path. + let staging_path = append_to_path(&snapshot_path, ".tmp"); + remove_file_if_exists(&staging_path) + .await + .with_context(|| { + format!( + "removing stale snapshot staging file {}", + staging_path.display() + ) + })?; + let staging_target = staging_path + .to_str() + .context("snapshot staging path is not valid UTF-8")?; + sqlx::query("VACUUM INTO ?") + .bind(staging_target) + .execute(&self.pool) + .await + .with_context(|| { + format!("writing pre-migration snapshot {}", staging_path.display()) + })?; + set_private_permissions(&staging_path).await?; + remove_file_if_exists(&snapshot_path) + .await + .with_context(|| { + format!( + "removing stale pre-migration snapshot {}", + snapshot_path.display() + ) + })?; + fs::rename(&staging_path, &snapshot_path) + .await + .with_context(|| { + format!( + "publishing pre-migration snapshot {}", + snapshot_path.display() + ) + })?; + + info!( + database = %database_path.display(), + snapshot = %snapshot_path.display(), + "Snapshotted SQLite database before applying new migrations" + ); + Ok(()) + } + pub async fn health_check(&self) -> anyhow::Result<()> { sqlx::query("SELECT 1") .execute(&self.pool) @@ -63,3 +148,58 @@ impl Database { self.pool.clone() } } + +/// Rollback artifact written by [`Database::migrate`] before applying new +/// migrations: the database path with `.pre-migration.bak` appended. +pub fn pre_migration_snapshot_path(database_path: &Path) -> PathBuf { + append_to_path(database_path, ".pre-migration.bak") +} + +fn append_to_path(path: &Path, suffix: &str) -> PathBuf { + let mut path = path.as_os_str().to_os_string(); + path.push(suffix); + PathBuf::from(path) +} + +async fn applied_migration_versions(pool: &DbPool) -> anyhow::Result> { + let mut conn = pool + .acquire() + .await + .context("acquiring a SQLite connection")?; + // Migrator::run performs this same ensure + list as its first step, so + // asking the Migrate trait (rather than querying sqlx's bookkeeping + // table by hand) cannot drift from what it will actually apply. + conn.ensure_migrations_table(&MIGRATOR.table_name) + .await + .context("ensuring the sqlx migrations table exists")?; + let applied = conn + .list_applied_migrations(&MIGRATOR.table_name) + .await + .context("listing applied migration versions")?; + Ok(applied + .into_iter() + .map(|migration| migration.version) + .collect()) +} + +async fn remove_file_if_exists(path: &Path) -> std::io::Result<()> { + match fs::remove_file(path).await { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(err) => Err(err), + } +} + +#[cfg(unix)] +async fn set_private_permissions(path: &Path) -> anyhow::Result<()> { + use std::os::unix::fs::PermissionsExt; + + fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) + .await + .with_context(|| format!("setting permissions on {}", path.display())) +} + +#[cfg(not(unix))] +async fn set_private_permissions(_path: &Path) -> anyhow::Result<()> { + Ok(()) +} diff --git a/lib/crates/fabro-db/tests/sqlite.rs b/lib/crates/fabro-db/tests/sqlite.rs index a030dafa77..28af2aadd3 100644 --- a/lib/crates/fabro-db/tests/sqlite.rs +++ b/lib/crates/fabro-db/tests/sqlite.rs @@ -123,3 +123,99 @@ async fn variables_schema_enforces_env_style_names() -> anyhow::Result<()> { Ok(()) } + +#[tokio::test] +async fn fresh_database_migrate_takes_no_snapshot() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let db_path = dir.path().join("fabro.sqlite3"); + + let database = fabro_db::Database::connect(&db_path).await?; + database.migrate().await?; + + assert!( + !fabro_db::pre_migration_snapshot_path(&db_path).exists(), + "a fresh database has no pre-migration state worth snapshotting" + ); + Ok(()) +} + +// Simulates a binary upgrade: a database whose `_sqlx_migrations` table is +// missing an entry for a bundled migration is exactly what an older binary +// leaves behind for a newer one. The environments migration is pure CREATE +// TABLE, so dropping the table and deleting its version row makes it pending +// again without violating checksums. +#[tokio::test] +async fn migrate_snapshots_database_before_applying_new_migrations() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let db_path = dir.path().join("fabro.sqlite3"); + let snapshot_path = fabro_db::pre_migration_snapshot_path(&db_path); + + let database = fabro_db::Database::connect(&db_path).await?; + database.migrate().await?; + sqlx::query( + "INSERT INTO variables (name, value, created_at, updated_at) \ + VALUES ('SNAPSHOT_MARKER', 'kept', '2026-07-22T00:00:00Z', '2026-07-22T00:00:00Z')", + ) + .execute(database.pool()) + .await?; + sqlx::query("DROP TABLE environments") + .execute(database.pool()) + .await?; + sqlx::query("DELETE FROM _sqlx_migrations WHERE version = 2026063002") + .execute(database.pool()) + .await?; + + database.migrate().await?; + + assert!( + snapshot_path.exists(), + "pending migration must snapshot first" + ); + let snapshot = connect_read_only(&snapshot_path).await?; + assert!( + !table_exists(&snapshot, "environments").await?, + "snapshot must hold the pre-migration schema" + ); + let snapshot_marker: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM variables WHERE name = 'SNAPSHOT_MARKER'") + .fetch_one(&snapshot) + .await?; + assert_eq!(snapshot_marker, 1, "snapshot must preserve row data"); + snapshot.close().await; + + assert!( + table_exists(database.pool(), "environments").await?, + "migration must still apply" + ); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&snapshot_path)?.permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "snapshot must be private"); + } + + // With nothing pending, migrate must not rewrite the snapshot: it still + // holds the state from before the most recent schema change. + database.migrate().await?; + let snapshot = connect_read_only(&snapshot_path).await?; + assert!( + !table_exists(&snapshot, "environments").await?, + "no-pending migrate must leave the snapshot untouched" + ); + snapshot.close().await; + Ok(()) +} + +async fn connect_read_only(path: &std::path::Path) -> anyhow::Result { + Ok(sqlx::SqlitePool::connect(&format!("sqlite://{}?mode=ro", path.display())).await?) +} + +async fn table_exists(pool: &sqlx::SqlitePool, table: &str) -> anyhow::Result { + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?") + .bind(table) + .fetch_one(pool) + .await?; + Ok(count == 1) +}