Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions lib/crates/fabro-db/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ workspace = true
anyhow.workspace = true
sqlx.workspace = true
tokio.workspace = true
tracing.workspace = true

[dev-dependencies]
tempfile = "3"
Expand Down
144 changes: 142 additions & 2 deletions lib/crates/fabro-db/src/lib.rs
Original file line number Diff line number Diff line change
@@ -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;

Expand Down Expand Up @@ -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)
Expand All @@ -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<HashSet<i64>> {
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(())
}
96 changes: 96 additions & 0 deletions lib/crates/fabro-db/tests/sqlite.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<sqlx::SqlitePool> {
Ok(sqlx::SqlitePool::connect(&format!("sqlite://{}?mode=ro", path.display())).await?)
}

async fn table_exists(pool: &sqlx::SqlitePool, table: &str) -> anyhow::Result<bool> {
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)
}
Loading