jake/Lumotia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(rb-02): migrations run inside a single transaction

Browse Source 
Every multi-statement migration and its matching schema_version insert
now execute on the same sqlx Transaction. A failure anywhere — a bad
statement, the version insert, or the commit itself — rolls the
database back to its previous state, so the next startup replays the
migration against a clean schema rather than a half-mutated one.

Extracted run_migrations_slice(pool, migrations) as the single apply
path. run_migrations delegates to it with MIGRATIONS; the test helper
run_migrations_up_to now filters MIGRATIONS by target and delegates to
the same code, eliminating the duplicated loop that previously lived
in the test module.

Regression test multi_statement_migration_rolls_back_on_failure
injects a poisoned v9 migration (valid CREATE followed by a bogus
function call) and asserts neither the partial schema change nor the
schema_version row persists after the failure.

SQLite DDL participates in transactions, so this is sufficient. Any
future migration that needs an implicitly-committing statement
(VACUUM / REINDEX / ATTACH — none today) must be its own
non-transactional migration; that's a reviewer responsibility.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jake
2026-04-22 10:27:29 +01:00
parent b48d39bfb1
commit 05d823bf05
3 changed files with 135 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
crates/storage/src/migrations.rs
View File

@@ -261,6 +261,30 @@ fn split_statements(sql: &str) -> Vec<String> {
/// Ensure the schema_version table exists and run any pending migrations.
pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
run_migrations_slice(pool, MIGRATIONS).await
}
/// Apply the pending prefix of `migrations`, each inside its own
/// transaction along with the matching `schema_version` row insert.
///
/// Atomicity was added in response to the 2026-04-22 review (RB-02):
/// the previous implementation executed statements individually against
/// the pool and only recorded the new version after all statements had
/// succeeded. A multi-statement migration that failed midway therefore
/// left the schema partially changed but still appearing unapplied —
/// the next startup would replay the migration against a mutated DB
/// and fail in surprising ways.
///
/// Wrapping both the statements and the version record in a single
/// `Transaction` is sufficient for SQLite (DDL participates in
/// transactions there). If a future migration needs an operation that
/// implicitly commits (`VACUUM`, `REINDEX`, `ATTACH`), it must be split
/// out into its own non-transactional migration — reviewer's job to
/// flag.
async fn run_migrations_slice(
pool: &SqlitePool,
migrations: &[(i64, &str, &str)],
) -> Result<()> {
sqlx::query(
"CREATE TABLE IF NOT EXISTS schema_version (
version INTEGER PRIMARY KEY,
@@ -277,27 +301,36 @@ pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
.await
.map_err(|e| KonError::StorageError(format!("Schema version query failed: {e}")))?;
for (version, description, sql) in MIGRATIONS {
for (version, description, sql) in migrations {
if *version > current {
log::info!("Running migration {}: {}", version, description);
let statements = split_statements(sql);
let mut tx = pool.begin().await.map_err(|e| {
KonError::StorageError(format!("Migration {} tx begin failed: {e}", version))
})?;
for statement in &statements {
sqlx::query(statement).execute(pool).await.map_err(|e| {
KonError::StorageError(format!("Migration {} failed: {e}", version))
})?;
for statement in split_statements(sql) {
sqlx::query(&statement)
.execute(&mut *tx)
.await
.map_err(|e| {
KonError::StorageError(format!("Migration {} failed: {e}", version))
})?;
}
sqlx::query("INSERT INTO schema_version (version, description) VALUES (?, ?)")
.bind(version)
.bind(description)
.execute(pool)
.execute(&mut *tx)
.await
.map_err(|e| {
KonError::StorageError(format!("Migration version record failed: {e}"))
})?;
tx.commit().await.map_err(|e| {
KonError::StorageError(format!("Migration {} commit failed: {e}", version))
})?;
log::info!("Migration {} complete", version);
}
}
@@ -463,47 +496,15 @@ mod tests {
}
/// Test-only helper: run migrations only up to (and including) `target_version`.
/// Mirrors `run_migrations` but stops early — used by the v6 upgrade-path test
/// to seed a v5 schema with dictionary rows before applying v6.
/// Used by the v6 upgrade-path test to seed a v5 schema with
/// dictionary rows before applying v6.
async fn run_migrations_up_to(pool: &SqlitePool, target_version: i64) -> Result<()> {
sqlx::query(
"CREATE TABLE IF NOT EXISTS schema_version (
version INTEGER PRIMARY KEY,
description TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
)",
)
.execute(pool)
.await
.map_err(|e| {
KonError::StorageError(format!("Schema version table creation failed: {e}"))
})?;
let current: i64 =
sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
.fetch_one(pool)
.await
.map_err(|e| KonError::StorageError(format!("Schema version query failed: {e}")))?;
for (version, description, sql) in MIGRATIONS {
if *version > current && *version <= target_version {
let statements = split_statements(sql);
for statement in &statements {
sqlx::query(statement).execute(pool).await.map_err(|e| {
KonError::StorageError(format!("Migration {} failed: {e}", version))
})?;
}
sqlx::query("INSERT INTO schema_version (version, description) VALUES (?, ?)")
.bind(version)
.bind(description)
.execute(pool)
.await
.map_err(|e| {
KonError::StorageError(format!("Migration version record failed: {e}"))
})?;
}
}
Ok(())
let filtered: Vec<(i64, &str, &str)> = MIGRATIONS
.iter()
.filter(|(v, _, _)| *v <= target_version)
.copied()
.collect();
run_migrations_slice(pool, &filtered).await
}
#[tokio::test]
@@ -632,4 +633,63 @@ mod tests {
let err = result.unwrap_err().to_string().to_lowercase();
assert!(err.contains("no such table"), "got: {err}");
}
// RB-02 regression: a multi-statement migration that fails part-way
// through must leave no trace on disk — the transaction rolls back
// both the partial schema change and (implicitly) the `schema_version`
// row that the pre-fix implementation would have recorded after
// statement-level success.
//
// The poisoned migration below first creates `poison_marker`
// (syntactically valid, would succeed against any SQLite) and then
// runs a guaranteed-invalid function call. Under the new atomic
// implementation, neither `poison_marker` nor the v9 row should
// survive the failed call.
#[tokio::test]
async fn multi_statement_migration_rolls_back_on_failure() {
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect("sqlite::memory:")
.await
.expect("pool");
run_migrations(&pool).await.expect("baseline migrate");
const POISON: &[(i64, &str, &str)] = &[(
9,
"rb-02 atomicity poison",
r#"
CREATE TABLE poison_marker (id INTEGER PRIMARY KEY);
SELECT this_function_does_not_exist();
"#,
)];
let result = run_migrations_slice(&pool, POISON).await;
assert!(
result.is_err(),
"poisoned migration must return Err, got: {result:?}"
);
// `poison_marker` must not be on disk — transaction rolled back.
let marker: std::result::Result<i64, sqlx::Error> =
sqlx::query_scalar("SELECT COUNT(*) FROM poison_marker")
.fetch_one(&pool)
.await;
assert!(
marker.is_err(),
"poison_marker must not exist; got: {marker:?}"
);
// `schema_version` must not include v9 — version insert is part
// of the same transaction that rolled back.
let max: i64 =
sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
.fetch_one(&pool)
.await
.expect("read schema_version");
assert_eq!(
max, 8,
"schema_version must not advance past the failed migration"
);
}
}