From f23b4fc37537696a892aa18c2c855315257d949d Mon Sep 17 00:00:00 2001 From: Oliver Eyton-Williams Date: Tue, 10 Nov 2026 18:42:15 +0100 Subject: [PATCH 1/2] refactor(cdb): create separate migration module The idea is basically to create some separation between migration and the logic that deals with the migrated database. This did involve messing around with the privacy rules which may not be entirely ideal --- databases/curriculum/src/bin/cdb-migrate.rs | 6 +- databases/curriculum/src/edd/exams.rs | 20 + databases/curriculum/src/edd/mod.rs | 2 + databases/curriculum/src/lib.rs | 582 +------------------- databases/curriculum/src/migrations.rs | 211 +++++++ databases/curriculum/tests/common/mod.rs | 60 ++ databases/curriculum/tests/exams.rs | 52 ++ databases/curriculum/tests/migrations.rs | 299 ++++++++++ 8 files changed, 652 insertions(+), 580 deletions(-) create mode 100644 databases/curriculum/src/edd/exams.rs create mode 100644 databases/curriculum/src/migrations.rs create mode 100644 databases/curriculum/tests/common/mod.rs create mode 100644 databases/curriculum/tests/exams.rs create mode 100644 databases/curriculum/tests/migrations.rs diff --git a/databases/curriculum/src/bin/cdb-migrate.rs b/databases/curriculum/src/bin/cdb-migrate.rs index ad02d9a..a895485 100644 --- a/databases/curriculum/src/bin/cdb-migrate.rs +++ b/databases/curriculum/src/bin/cdb-migrate.rs @@ -7,6 +7,8 @@ //! Env: `CDB_MIGRATOR_URL`, naming the database (`.../curriculum`), never a `db/branch` revision. use std::{process::ExitCode, time::Duration}; +use cdb::migrations; + /// Bounds the whole run: waiting for another run's lock, then applying and merging. The deploy /// Action's MIGRATE_SECONDS must stay above this plus the image pull. A run cut short ends its /// session, which releases the lock; anything it left is on the scratch branch, which the next run @@ -24,13 +26,13 @@ async fn main() -> ExitCode { return ExitCode::from(2); }; - match tokio::time::timeout(TIMEOUT, cdb::migrate(&url)).await { + match tokio::time::timeout(TIMEOUT, migrations::migrate(&url)).await { Ok(Ok(applied)) if applied.is_empty() => { println!("cDb up to date"); ExitCode::SUCCESS } Ok(Ok(applied)) => { - println!("cDb migrations applied to {}: {applied:?}", cdb::DEV); + println!("cDb migrations applied to {}: {applied:?}", migrations::DEV); ExitCode::SUCCESS } Ok(Err(err)) => { diff --git a/databases/curriculum/src/edd/exams.rs b/databases/curriculum/src/edd/exams.rs new file mode 100644 index 0000000..45ad85e --- /dev/null +++ b/databases/curriculum/src/edd/exams.rs @@ -0,0 +1,20 @@ +use crate::Database; + +pub async fn create( + db: &Database, + id: &str, + slug: &str, + title: &str, + instructions: &str, +) -> sqlx::Result<()> { + sqlx::query( + "INSERT INTO exams (id, slug, title, instructions, status) VALUES (?, ?, ?, ?, 'disabled')", + ) + .bind(id) + .bind(slug) + .bind(title) + .bind(instructions) + .execute(&db.pool) + .await?; + Ok(()) +} diff --git a/databases/curriculum/src/edd/mod.rs b/databases/curriculum/src/edd/mod.rs index ff1a03e..12fac1b 100644 --- a/databases/curriculum/src/edd/mod.rs +++ b/databases/curriculum/src/edd/mod.rs @@ -1 +1,3 @@ //! Schema `edd`: eDd's tables. aAPI has no access. + +pub mod exams; diff --git a/databases/curriculum/src/lib.rs b/databases/curriculum/src/lib.rs index a6ce8e7..532ca03 100644 --- a/databases/curriculum/src/lib.rs +++ b/databases/curriculum/src/lib.rs @@ -1,227 +1,15 @@ //! cDb: versioned curriculum content. Dolt. -use std::{collections::HashMap, time::Duration}; +use std::time::Duration; -use sqlx::{ - Connection, MySqlConnection, MySqlPool, Row, - migrate::{Migrate, MigrateError, Migrator}, - mysql::MySqlPoolOptions, -}; +use sqlx::{MySqlPool, mysql::MySqlPoolOptions}; mod aapi; -mod edd; +pub mod edd; +pub mod migrations; mod shared; -/// The only branch migrations reach directly. -pub const DEV: &str = "dev"; -/// Where `cdb-migrate` applies migrations before merging them into [`DEV`]. Recreated from `dev` -/// on every run, so a failed run leaves `dev` untouched and nothing to clean up. -pub const SCRATCH: &str = "cdb-migrate"; -/// Author of every commit `cdb-migrate` makes. -pub const AUTHOR: &str = "cdb-migrate "; - -#[derive(Debug, thiserror::Error)] -pub enum Error { - #[error(transparent)] - Migrate(#[from] MigrateError), - #[error(transparent)] - Sqlx(#[from] sqlx::Error), - #[error("lock {0} not acquired")] - Lock(String), - #[error("expected branch {expected} checked out, found {actual}")] - WrongBranch { - expected: &'static str, - actual: String, - }, - #[error("merging {SCRATCH} into {DEV} conflicted ({conflicts}): aborted, {DEV} unchanged")] - MergeConflict { conflicts: i64 }, -} - -/// Every migration in `migrations/`, tracked in `_sqlx_migrations` (`sqlx.toml`). -/// -/// Ignores applied migrations it does not embed: those come from a newer release, and this one is a -/// rollback. Safe only because migrations stay backward compatible. Checksums of the migrations it -/// does embed are still verified. -pub fn migrator() -> Migrator { - let mut migrator = sqlx::migrate!(); - migrator.set_ignore_missing(true); - migrator -} - -/// Connects to `url` (a database name, never a `db/branch` revision) and applies pending migrations -/// to [`DEV`], one Dolt commit each. Returns the versions applied. -pub async fn migrate(url: &str) -> Result, Error> { - let mut conn = MySqlConnection::connect(url).await?; - let result = migrate_with(&mut conn, &migrator()).await; - // A failed run may still hold the lock - it goes with the session. - let _ = conn.close().await; - result -} - -/// [`migrate`] on an open connection. Holds a named lock throughout, so concurrent runs apply each -/// migration once. -/// -/// 1. Recreate [`SCRATCH`] from `dev`'s HEAD and check it out. -/// 2. Apply each pending migration there, then `dolt_commit` it. Dolt commits the SQL transaction -/// on DDL, so a migration is never atomic: a failed one is left on the scratch branch only. -/// 3. Merge into `dev`: a fast-forward unless `dev` moved meanwhile, then a merge commit. -/// 4. Delete [`SCRATCH`]. -pub(crate) async fn migrate_with( - conn: &mut MySqlConnection, - migrator: &Migrator, -) -> Result, Error> { - let lock = lock(conn).await?; - let result = async { - let applied = apply_on_scratch(conn, migrator).await?; - if !applied.is_empty() { - merge_into_dev(conn, &applied).await?; - } - drop_scratch(conn).await?; - Ok(applied) - } - .await; - let _ = unlock(conn, &lock).await; - result -} - -async fn lock(conn: &mut MySqlConnection) -> Result { - let database: String = sqlx::query_scalar("SELECT DATABASE()") - .fetch_one(&mut *conn) - .await?; - let name = format!("cdb-migrate:{database}"); - // Waits for as long as it takes: cdb-migrate's own timeout bounds the run. - let acquired: Option = sqlx::query_scalar("SELECT GET_LOCK(?, -1)") - .bind(&name) - .fetch_one(&mut *conn) - .await?; - if acquired != Some(1) { - return Err(Error::Lock(name)); - } - Ok(name) -} - -async fn unlock(conn: &mut MySqlConnection, name: &str) -> Result<(), Error> { - sqlx::query("SELECT RELEASE_LOCK(?)") - .bind(name) - .execute(conn) - .await?; - Ok(()) -} - -async fn checkout(conn: &mut MySqlConnection, branch: &'static str) -> Result<(), Error> { - sqlx::query("CALL DOLT_CHECKOUT(?)") - .bind(branch) - .execute(&mut *conn) - .await?; - let actual: String = sqlx::query_scalar("SELECT ACTIVE_BRANCH()") - .fetch_one(&mut *conn) - .await?; - if actual != branch { - return Err(Error::WrongBranch { - expected: branch, - actual, - }); - } - Ok(()) -} - -/// Steps 1 and 2 of [`migrate_with`]. Leaves [`SCRATCH`] checked out. -pub(crate) async fn apply_on_scratch( - conn: &mut MySqlConnection, - migrator: &Migrator, -) -> Result, Error> { - // Off the scratch branch first: `-f` cannot reset the branch a session has checked out. - checkout(conn, DEV).await?; - sqlx::query("CALL DOLT_BRANCH('-f', ?, ?)") - .bind(SCRATCH) - .bind(DEV) - .execute(&mut *conn) - .await?; - checkout(conn, SCRATCH).await?; - - let table = &*migrator.table_name; - conn.ensure_migrations_table(table).await?; - // Only a migration that failed on `dev` itself could leave one: scratch runs never reach it. - if let Some(version) = conn.dirty_version(table).await? { - return Err(MigrateError::Dirty(version).into()); - } - let applied: HashMap<_, _> = conn - .list_applied_migrations(table) - .await? - .into_iter() - .map(|m| (m.version, m.checksum)) - .collect(); - if !migrator.ignore_missing - && let Some(&version) = applied.keys().find(|v| !migrator.version_exists(**v)) - { - return Err(MigrateError::VersionMissing(version).into()); - } - - let mut versions = Vec::new(); - for migration in migrator - .iter() - .filter(|m| !m.migration_type.is_down_migration()) - { - match applied.get(&migration.version) { - Some(checksum) if *checksum != migration.checksum => { - return Err(MigrateError::VersionMismatch(migration.version).into()); - } - Some(_) => {} - None => { - conn.apply(table, migration).await?; - // After the migration's transaction, never inside it: a dolt_commit in an open - // transaction can commit a stale snapshot. - sqlx::query("CALL DOLT_COMMIT('-A', '-m', ?, '--author', ?)") - .bind(format!( - "cdb-migrate: {:04} {}", - migration.version, migration.description - )) - .bind(AUTHOR) - .execute(&mut *conn) - .await?; - versions.push(migration.version); - } - } - } - Ok(versions) -} - -/// Step 3 of [`migrate_with`]. -pub(crate) async fn merge_into_dev( - conn: &mut MySqlConnection, - applied: &[i64], -) -> Result<(), Error> { - let (first, last) = (applied[0], applied[applied.len() - 1]); - checkout(conn, DEV).await?; - // Fails, and rolls back, on conflicts: `dolt_allow_commit_conflicts` is off (config/server.yaml). - let row = sqlx::query("CALL DOLT_MERGE(?, '-m', ?, '--author', ?)") - .bind(SCRATCH) - .bind(format!("cdb-migrate: merge {first:04}..{last:04}")) - .bind(AUTHOR) - .fetch_one(&mut *conn) - .await?; - let conflicts: i64 = row.try_get("conflicts")?; - if conflicts != 0 { - sqlx::query("CALL DOLT_MERGE('--abort')") - .execute(&mut *conn) - .await?; - return Err(Error::MergeConflict { conflicts }); - } - Ok(()) -} - -/// Step 4 of [`migrate_with`]. Leaves [`DEV`] checked out. -pub(crate) async fn drop_scratch(conn: &mut MySqlConnection) -> Result<(), Error> { - checkout(conn, DEV).await?; - sqlx::query("CALL DOLT_BRANCH('-D', ?)") - .bind(SCRATCH) - .execute(conn) - .await?; - Ok(()) -} - #[derive(Debug, Clone)] pub struct Database { - #[expect(dead_code, reason = "no queries yet")] pool: MySqlPool, } @@ -246,365 +34,3 @@ impl Database { Self { pool } } } - -#[cfg(test)] -pub(crate) mod testing { - use std::str::FromStr; - - use sqlx::{ - ConnectOptions, Connection, Executor, MySqlConnection, migrate::Migrator, - mysql::MySqlConnectOptions, - }; - - use crate::{DEV, Error}; - - /// A database of its own on the Dolt server at `CDB_MIGRATOR_URL`, with branch `dev`. - pub struct TestDb { - pub name: String, - options: MySqlConnectOptions, - } - - impl TestDb { - /// `None` when `CDB_MIGRATOR_URL` is unset: the test then passes without running. - pub async fn new() -> sqlx::Result> { - let Some(url) = std::env::var("CDB_MIGRATOR_URL") - .ok() - .filter(|v| !v.is_empty()) - else { - panic!( - "CDB_MIGRATOR_URL must be set for testing. Hint: copy the value from .env.example into .env" - ); - }; - let options = MySqlConnectOptions::from_str(&url)?; - let name = format!("cdb_test_{}", uuid::Uuid::new_v4().simple()); - let mut conn = options.connect().await?; - conn.execute(sqlx::AssertSqlSafe(format!("CREATE DATABASE `{name}`"))) - .await?; - let db = Self { - options: options.database(&name), - name, - }; - // The server's `init.defaultbranch` may not be `dev`. - let mut conn = db.connect().await?; - let has_dev: i64 = - sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") - .bind(DEV) - .fetch_one(&mut conn) - .await?; - if has_dev == 0 { - sqlx::query("CALL DOLT_BRANCH(?)") - .bind(DEV) - .execute(&mut conn) - .await?; - } - conn.close().await?; - Ok(Some(db)) - } - - pub async fn connect(&self) -> sqlx::Result { - self.options.connect().await - } - - /// Runs `migrator` as `cdb-migrate` does, on a connection of its own. - pub async fn migrate_with(&self, migrator: &Migrator) -> Result, Error> { - let mut conn = self.connect().await?; - let result = crate::migrate_with(&mut conn, migrator).await; - let _ = conn.close().await; - result - } - - pub async fn setup(&self) -> Result, Error> { - self.migrate_with(&crate::migrator()).await - } - - /// Runs `sql` on `dev` and commits it there, as eDd would. - pub async fn commit_on_dev(&self, sql: &'static str) -> sqlx::Result<()> { - let mut conn = self.connect().await?; - sqlx::query("CALL DOLT_CHECKOUT(?)") - .bind(DEV) - .execute(&mut conn) - .await?; - sqlx::raw_sql(sql).execute(&mut conn).await?; - sqlx::query("CALL DOLT_COMMIT('-A', '-m', 'test: on dev')") - .execute(&mut conn) - .await?; - conn.close().await - } - - pub async fn hash_of(&self, branch: &str) -> sqlx::Result { - let mut conn = self.connect().await?; - let hash = sqlx::query_scalar("SELECT HASHOF(?)") - .bind(branch) - .fetch_one(&mut conn) - .await?; - conn.close().await?; - Ok(hash) - } - - /// Successful migrations recorded on `dev`. - pub async fn applied_versions(&self) -> sqlx::Result> { - let mut conn = self.connect().await?; - let versions = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( - "SELECT version FROM `{}/{DEV}`._sqlx_migrations WHERE success ORDER BY version", - self.name - ))) - .fetch_all(&mut conn) - .await?; - conn.close().await?; - Ok(versions) - } - - pub async fn destroy(self) -> sqlx::Result<()> { - let mut conn = self.connect().await?; - conn.execute(sqlx::AssertSqlSafe(format!( - "DROP DATABASE `{}`", - self.name - ))) - .await?; - conn.close().await - } - } - - /// Versions of every up migration `migrator` embeds. - pub fn up_versions(migrator: &Migrator) -> Vec { - migrator - .iter() - .filter(|m| !m.migration_type.is_down_migration()) - .map(|m| m.version) - .collect() - } -} - -#[cfg(test)] -mod tests { - use std::borrow::Cow; - - use sqlx::migrate::{Migration, MigrationType}; - - use super::{testing::*, *}; - - /// [`migrator`] plus `sql` as migration 9998. - fn migrator_with(sql: &'static str) -> Migrator { - let mut migrator = migrator(); - migrator.migrations.to_mut().push(Migration::new( - 9998, - Cow::Borrowed("added by a test"), - MigrationType::Simple, - sqlx::SqlStr::from_static(sql), - false, - )); - migrator - } - - /// A newer release applied migration 9999 on `dev`, then rolled back to this release. - async fn apply_newer_migration(db: &TestDb) -> sqlx::Result<()> { - db.commit_on_dev( - "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) - VALUES (9999, 'from a newer release', true, X'00', 0)", - ) - .await - } - - async fn scratch_exists(db: &TestDb) -> sqlx::Result { - let mut conn = db.connect().await?; - let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") - .bind(SCRATCH) - .fetch_one(&mut conn) - .await?; - conn.close().await?; - Ok(count > 0) - } - - #[tokio::test] - async fn each_migration_is_one_dolt_commit_on_dev() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - let expected = up_versions(&migrator()); - - assert_eq!(db.setup().await?, expected); - - let mut conn = db.connect().await?; - let log: Vec<(String, String)> = sqlx::query_as(sqlx::AssertSqlSafe(format!( - "SELECT message, author FROM dolt_log('{DEV}') LIMIT {}", - expected.len() - ))) - .fetch_all(&mut conn) - .await?; - let messages: Vec<_> = migrator() - .iter() - .rev() - .map(|m| format!("cdb-migrate: {:04} {}", m.version, m.description)) - .collect(); - assert_eq!( - log.iter().map(|(m, _)| m).collect::>(), - messages.iter().collect::>() - ); - assert!(log.iter().all(|(_, c)| c == "cdb-migrate"), "{log:?}"); - let dirty: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( - "SELECT COUNT(*) FROM `{}/{DEV}`.dolt_status", - db.name - ))) - .fetch_one(&mut conn) - .await?; - assert_eq!(dirty, 0); - conn.close().await?; - assert_eq!(db.applied_versions().await?, expected); - assert!(!scratch_exists(&db).await?); - Ok(db.destroy().await?) - } - - #[tokio::test] - async fn rerun_is_a_no_op() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - db.setup().await?; - let before = db.hash_of(DEV).await?; - - assert_eq!(db.setup().await?, Vec::::new()); - - assert_eq!(db.hash_of(DEV).await?, before); - assert!(!scratch_exists(&db).await?); - Ok(db.destroy().await?) - } - - /// Two runs at once: the named lock serializes them. - #[tokio::test] - async fn concurrent_runs_apply_each_migration_once() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - let migrator = migrator(); - - let (a, b) = tokio::join!(db.migrate_with(&migrator), db.migrate_with(&migrator)); - let (mut a, b) = (a?, b?); - - a.extend(b); - a.sort_unstable(); - assert_eq!(a, up_versions(&migrator)); - assert_eq!(db.applied_versions().await?, up_versions(&migrator)); - Ok(db.destroy().await?) - } - - #[tokio::test] - async fn older_bundle_runs_against_newer_migrations() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - db.setup().await?; - apply_newer_migration(&db).await?; - - db.setup().await?; - - Ok(db.destroy().await?) - } - - #[tokio::test] - async fn strict_migrator_rejects_newer_migrations() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - db.setup().await?; - apply_newer_migration(&db).await?; - let mut migrator = migrator(); - migrator.set_ignore_missing(false); - - let result = db.migrate_with(&migrator).await; - - assert!( - matches!( - result, - Err(Error::Migrate(MigrateError::VersionMissing(9999))) - ), - "{result:?}" - ); - Ok(db.destroy().await?) - } - - #[tokio::test] - async fn rejects_changed_migration() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - db.setup().await?; - db.commit_on_dev("UPDATE _sqlx_migrations SET checksum = X'00' WHERE version = 1") - .await?; - - let result = db.migrate_with(&migrator()).await; - - assert!( - matches!( - result, - Err(Error::Migrate(MigrateError::VersionMismatch(1))) - ), - "{result:?}" - ); - Ok(db.destroy().await?) - } - - /// The failed migration stays on the scratch branch, half applied; the next run starts over. - #[tokio::test] - async fn failed_migration_leaves_dev_untouched() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - db.setup().await?; - let before = db.hash_of(DEV).await?; - - let result = db - .migrate_with(&migrator_with( - "CREATE TABLE t (id int); CREATE TABLE t (id int);", - )) - .await; - - assert!(result.is_err(), "{result:?}"); - assert_eq!(db.hash_of(DEV).await?, before); - assert_eq!( - db.migrate_with(&migrator_with("CREATE TABLE t (id int)")) - .await?, - vec![9998] - ); - Ok(db.destroy().await?) - } - - /// eDd commits content to `dev` while migrations apply: the run ends in a merge commit with both. - #[tokio::test] - async fn merges_when_dev_moved() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - let migrator = migrator(); - let mut conn = db.connect().await?; - - let applied = apply_on_scratch(&mut conn, &migrator).await?; - db.commit_on_dev( - "CREATE TABLE content (id int PRIMARY KEY); INSERT INTO content VALUES (1)", - ) - .await?; - merge_into_dev(&mut conn, &applied).await?; - drop_scratch(&mut conn).await?; - - let message: String = sqlx::query_scalar("SELECT message FROM dolt_log(?) LIMIT 1") - .bind(DEV) - .fetch_one(&mut conn) - .await?; - let versions = up_versions(&migrator); - assert_eq!( - message, - format!( - "cdb-migrate: merge {:04}..{:04}", - versions[0], - versions[versions.len() - 1] - ) - ); - let content: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM content") - .fetch_one(&mut conn) - .await?; - assert_eq!(content, 1); - conn.close().await?; - assert_eq!(db.applied_versions().await?, up_versions(&migrator)); - Ok(db.destroy().await?) - } - - #[tokio::test] - async fn never_touches_other_branches() -> Result<(), Error> { - let db = TestDb::new().await?.unwrap(); - let mut conn = db.connect().await?; - sqlx::query("CALL DOLT_BRANCH('prd', ?)") - .bind(DEV) - .execute(&mut conn) - .await?; - conn.close().await?; - let before = db.hash_of("prd").await?; - - db.setup().await?; - - assert_eq!(db.hash_of("prd").await?, before); - Ok(db.destroy().await?) - } -} diff --git a/databases/curriculum/src/migrations.rs b/databases/curriculum/src/migrations.rs new file mode 100644 index 0000000..b9316dd --- /dev/null +++ b/databases/curriculum/src/migrations.rs @@ -0,0 +1,211 @@ +use std::collections::HashMap; + +use sqlx::{ + Connection, MySqlConnection, Row, + migrate::{Migrate, MigrateError, Migrator}, +}; + +/// The only branch migrations reach directly. +pub const DEV: &str = "dev"; +/// Where `cdb-migrate` applies migrations before merging them into [`DEV`]. Recreated from `dev` +/// on every run, so a failed run leaves `dev` untouched and nothing to clean up. +const SCRATCH: &str = "cdb-migrate"; +/// Author of every commit `cdb-migrate` makes. +const AUTHOR: &str = "cdb-migrate "; + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error(transparent)] + Migrate(#[from] MigrateError), + #[error(transparent)] + Sqlx(#[from] sqlx::Error), + #[error("lock {0} not acquired")] + Lock(String), + #[error("expected branch {expected} checked out, found {actual}")] + WrongBranch { + expected: &'static str, + actual: String, + }, + #[error("merging {SCRATCH} into {DEV} conflicted ({conflicts}): aborted, {DEV} unchanged")] + MergeConflict { conflicts: i64 }, +} + +/// Every migration in `migrations/`, tracked in `_sqlx_migrations` (`sqlx.toml`). +/// +/// Ignores applied migrations it does not embed: those come from a newer release, and this one is a +/// rollback. Safe only because migrations stay backward compatible. Checksums of the migrations it +/// does embed are still verified. +pub fn migrator() -> Migrator { + let mut migrator = sqlx::migrate!(); + migrator.set_ignore_missing(true); + migrator +} + +/// Connects to `url` (a database name, never a `db/branch` revision) and applies pending migrations +/// to [`DEV`], one Dolt commit each. Returns the versions applied. +pub async fn migrate(url: &str) -> Result, Error> { + let mut conn = MySqlConnection::connect(url).await?; + let result = migrate_with(&mut conn, &migrator()).await; + // A failed run may still hold the lock - it goes with the session. + let _ = conn.close().await; + result +} + +/// [`migrate`] on an open connection. Holds a named lock throughout, so concurrent runs apply each +/// migration once. +/// +/// 1. Recreate [`SCRATCH`] from `dev`'s HEAD and check it out. +/// 2. Apply each pending migration there, then `dolt_commit` it. Dolt commits the SQL transaction +/// on DDL, so a migration is never atomic: a failed one is left on the scratch branch only. +/// 3. Merge into `dev`: a fast-forward unless `dev` moved meanwhile, then a merge commit. +/// 4. Delete [`SCRATCH`]. +pub async fn migrate_with( + conn: &mut MySqlConnection, + migrator: &Migrator, +) -> Result, Error> { + let lock = lock(conn).await?; + let result = async { + let applied = apply_on_scratch(conn, migrator).await?; + if !applied.is_empty() { + merge_into_dev(conn, &applied).await?; + } + drop_scratch(conn).await?; + Ok(applied) + } + .await; + let _ = unlock(conn, &lock).await; + result +} + +async fn lock(conn: &mut MySqlConnection) -> Result { + let database: String = sqlx::query_scalar("SELECT DATABASE()") + .fetch_one(&mut *conn) + .await?; + let name = format!("cdb-migrate:{database}"); + // Waits for as long as it takes: cdb-migrate's own timeout bounds the run. + let acquired: Option = sqlx::query_scalar("SELECT GET_LOCK(?, -1)") + .bind(&name) + .fetch_one(&mut *conn) + .await?; + if acquired != Some(1) { + return Err(Error::Lock(name)); + } + Ok(name) +} + +async fn unlock(conn: &mut MySqlConnection, name: &str) -> Result<(), Error> { + sqlx::query("SELECT RELEASE_LOCK(?)") + .bind(name) + .execute(conn) + .await?; + Ok(()) +} + +async fn checkout(conn: &mut MySqlConnection, branch: &'static str) -> Result<(), Error> { + sqlx::query("CALL DOLT_CHECKOUT(?)") + .bind(branch) + .execute(&mut *conn) + .await?; + let actual: String = sqlx::query_scalar("SELECT ACTIVE_BRANCH()") + .fetch_one(&mut *conn) + .await?; + if actual != branch { + return Err(Error::WrongBranch { + expected: branch, + actual, + }); + } + Ok(()) +} + +/// Steps 1 and 2 of [`migrate_with`]. Leaves [`SCRATCH`] checked out. +pub async fn apply_on_scratch( + conn: &mut MySqlConnection, + migrator: &Migrator, +) -> Result, Error> { + // Off the scratch branch first: `-f` cannot reset the branch a session has checked out. + checkout(conn, DEV).await?; + sqlx::query("CALL DOLT_BRANCH('-f', ?, ?)") + .bind(SCRATCH) + .bind(DEV) + .execute(&mut *conn) + .await?; + checkout(conn, SCRATCH).await?; + + let table = &*migrator.table_name; + conn.ensure_migrations_table(table).await?; + // Only a migration that failed on `dev` itself could leave one: scratch runs never reach it. + if let Some(version) = conn.dirty_version(table).await? { + return Err(MigrateError::Dirty(version).into()); + } + let applied: HashMap<_, _> = conn + .list_applied_migrations(table) + .await? + .into_iter() + .map(|m| (m.version, m.checksum)) + .collect(); + if !migrator.ignore_missing + && let Some(&version) = applied.keys().find(|v| !migrator.version_exists(**v)) + { + return Err(MigrateError::VersionMissing(version).into()); + } + + let mut versions = Vec::new(); + for migration in migrator + .iter() + .filter(|m| !m.migration_type.is_down_migration()) + { + match applied.get(&migration.version) { + Some(checksum) if *checksum != migration.checksum => { + return Err(MigrateError::VersionMismatch(migration.version).into()); + } + Some(_) => {} + None => { + conn.apply(table, migration).await?; + // After the migration's transaction, never inside it: a dolt_commit in an open + // transaction can commit a stale snapshot. + sqlx::query("CALL DOLT_COMMIT('-A', '-m', ?, '--author', ?)") + .bind(format!( + "cdb-migrate: {:04} {}", + migration.version, migration.description + )) + .bind(AUTHOR) + .execute(&mut *conn) + .await?; + versions.push(migration.version); + } + } + } + Ok(versions) +} + +/// Step 3 of [`migrate_with`]. +pub async fn merge_into_dev(conn: &mut MySqlConnection, applied: &[i64]) -> Result<(), Error> { + let (first, last) = (applied[0], applied[applied.len() - 1]); + checkout(conn, DEV).await?; + // Fails, and rolls back, on conflicts: `dolt_allow_commit_conflicts` is off (config/server.yaml). + let row = sqlx::query("CALL DOLT_MERGE(?, '-m', ?, '--author', ?)") + .bind(SCRATCH) + .bind(format!("cdb-migrate: merge {first:04}..{last:04}")) + .bind(AUTHOR) + .fetch_one(&mut *conn) + .await?; + let conflicts: i64 = row.try_get("conflicts")?; + if conflicts != 0 { + sqlx::query("CALL DOLT_MERGE('--abort')") + .execute(&mut *conn) + .await?; + return Err(Error::MergeConflict { conflicts }); + } + Ok(()) +} + +/// Step 4 of [`migrate_with`]. Leaves [`DEV`] checked out. +pub async fn drop_scratch(conn: &mut MySqlConnection) -> Result<(), Error> { + checkout(conn, DEV).await?; + sqlx::query("CALL DOLT_BRANCH('-D', ?)") + .bind(SCRATCH) + .execute(conn) + .await?; + Ok(()) +} diff --git a/databases/curriculum/tests/common/mod.rs b/databases/curriculum/tests/common/mod.rs new file mode 100644 index 0000000..4453fb8 --- /dev/null +++ b/databases/curriculum/tests/common/mod.rs @@ -0,0 +1,60 @@ +use std::str::FromStr; + +use sqlx::{ConnectOptions, Connection, Executor, MySqlConnection, mysql::MySqlConnectOptions}; + +/// A database of its own on the Dolt server at `CDB_MIGRATOR_URL`, with branch `dev`. +pub struct TestDb { + pub name: String, + pub(crate) options: MySqlConnectOptions, +} + +impl TestDb { + pub async fn new() -> sqlx::Result> { + let Some(url) = std::env::var("CDB_MIGRATOR_URL") + .ok() + .filter(|v| !v.is_empty()) + else { + panic!( + "CDB_MIGRATOR_URL must be set for testing. Hint: copy the value from .env.example into .env" + ); + }; + let options = MySqlConnectOptions::from_str(&url)?; + let name = format!("cdb_test_{}", uuid::Uuid::new_v4().simple()); + let mut conn = options.connect().await?; + conn.execute(sqlx::AssertSqlSafe(format!("CREATE DATABASE `{name}`"))) + .await?; + let db = Self { + options: options.database(&name), + name, + }; + // The server's `init.defaultbranch` may not be `dev`. + let mut conn = db.connect().await?; + let has_dev: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") + .bind("dev") + .fetch_one(&mut conn) + .await?; + if has_dev == 0 { + sqlx::query("CALL DOLT_BRANCH(?)") + .bind("dev") + .execute(&mut conn) + .await?; + } + conn.close().await?; + Ok(Some(db)) + } + + pub async fn connect(&self) -> sqlx::Result { + self.options.connect().await + } + + pub async fn destroy(self) -> sqlx::Result<()> { + let mut conn = self.connect().await?; + conn.execute(sqlx::AssertSqlSafe(format!( + "DROP DATABASE `{}`", + self.name + ))) + .await?; + conn.close().await + } +} diff --git a/databases/curriculum/tests/exams.rs b/databases/curriculum/tests/exams.rs new file mode 100644 index 0000000..32c11bc --- /dev/null +++ b/databases/curriculum/tests/exams.rs @@ -0,0 +1,52 @@ +mod common; + +use cdb::{ + Database, + edd::exams::create, + migrations::{self, DEV, Error}, +}; +use common::TestDb; +use sqlx::ConnectOptions; + +impl TestDb { + pub async fn migrated() -> Result { + let db = Self::new().await?.unwrap(); + migrations::migrate(db.options.to_url_lossy().as_str()).await?; + Ok(db) + } + + pub async fn pool(&self) -> sqlx::Result { + sqlx::mysql::MySqlPoolOptions::new() + .connect_with(self.options.clone().database(&format!("{}/{DEV}", self.name))) + .await + } +} + +#[tokio::test] +async fn creates_exam_in_migrated_database() -> Result<(), Box> { + let fixture = TestDb::migrated().await?; + let pool = fixture.pool().await?; + let db = Database::from_pool(pool.clone()); + + create(&db, "exam-id", "sample-exam", "Sample exam", "Read carefully").await?; + + let row: (String, String, String, String, String) = sqlx::query_as( + "SELECT id, slug, title, instructions, status FROM exams WHERE id = ?", + ) + .bind("exam-id") + .fetch_one(&pool) + .await?; + assert_eq!( + row, + ( + "exam-id".into(), + "sample-exam".into(), + "Sample exam".into(), + "Read carefully".into(), + "disabled".into(), + ) + ); + pool.close().await; + fixture.destroy().await?; + Ok(()) +} diff --git a/databases/curriculum/tests/migrations.rs b/databases/curriculum/tests/migrations.rs new file mode 100644 index 0000000..a00c27d --- /dev/null +++ b/databases/curriculum/tests/migrations.rs @@ -0,0 +1,299 @@ +mod common; + +use std::borrow::Cow; + +use cdb::migrations::{ + self, DEV, Error, apply_on_scratch, drop_scratch, merge_into_dev, migrator, +}; +use common::TestDb; +use sqlx::{ + ConnectOptions, Connection, + migrate::{MigrateError, Migration, MigrationType, Migrator}, +}; + +const SCRATCH: &str = "cdb-migrate"; + +impl TestDb { + /// Runs `migrator` as `cdb-migrate` does, on a connection of its own. + pub async fn migrate_with(&self, migrator: &Migrator) -> Result, Error> { + let mut conn = self.connect().await?; + let result = migrations::migrate_with(&mut conn, migrator).await; + let _ = conn.close().await; + result + } + + pub async fn setup(&self) -> Result, Error> { + migrations::migrate(self.options.to_url_lossy().as_str()).await + } + + /// Runs `sql` on `dev` and commits it there, as eDd would. + pub async fn commit_on_dev(&self, sql: &'static str) -> sqlx::Result<()> { + let mut conn = self.connect().await?; + sqlx::query("CALL DOLT_CHECKOUT(?)") + .bind(DEV) + .execute(&mut conn) + .await?; + sqlx::raw_sql(sql).execute(&mut conn).await?; + sqlx::query("CALL DOLT_COMMIT('-A', '-m', 'test: on dev')") + .execute(&mut conn) + .await?; + conn.close().await + } + + pub async fn hash_of(&self, branch: &str) -> sqlx::Result { + let mut conn = self.connect().await?; + let hash = sqlx::query_scalar("SELECT HASHOF(?)") + .bind(branch) + .fetch_one(&mut conn) + .await?; + conn.close().await?; + Ok(hash) + } + + /// Successful migrations recorded on `dev`. + pub async fn applied_versions(&self) -> sqlx::Result> { + let mut conn = self.connect().await?; + let versions = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT version FROM `{}/{DEV}`._sqlx_migrations WHERE success ORDER BY version", + self.name + ))) + .fetch_all(&mut conn) + .await?; + conn.close().await?; + Ok(versions) + } +} + +/// Versions of every up migration `migrator` embeds. +pub fn up_versions(migrator: &Migrator) -> Vec { + migrator + .iter() + .filter(|m| !m.migration_type.is_down_migration()) + .map(|m| m.version) + .collect() +} + +/// [`migrator`] plus `sql` as migration 9998. +fn migrator_with(sql: &'static str) -> Migrator { + let mut migrator = migrator(); + migrator.migrations.to_mut().push(Migration::new( + 9998, + Cow::Borrowed("added by a test"), + MigrationType::Simple, + sqlx::SqlStr::from_static(sql), + false, + )); + migrator +} + +/// A newer release applied migration 9999 on `dev`, then rolled back to this release. +async fn apply_newer_migration(db: &TestDb) -> sqlx::Result<()> { + db.commit_on_dev( + "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) + VALUES (9999, 'from a newer release', true, X'00', 0)", + ) + .await +} + +async fn scratch_exists(db: &TestDb) -> sqlx::Result { + let mut conn = db.connect().await?; + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") + .bind(SCRATCH) + .fetch_one(&mut conn) + .await?; + conn.close().await?; + Ok(count > 0) +} + +#[tokio::test] +async fn each_migration_is_one_dolt_commit_on_dev() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + let expected = up_versions(&migrator()); + + assert_eq!(db.setup().await?, expected); + + let mut conn = db.connect().await?; + let log: Vec<(String, String)> = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT message, author FROM dolt_log('{DEV}') LIMIT {}", + expected.len() + ))) + .fetch_all(&mut conn) + .await?; + let messages: Vec<_> = migrator() + .iter() + .rev() + .map(|m| format!("cdb-migrate: {:04} {}", m.version, m.description)) + .collect(); + assert_eq!( + log.iter().map(|(m, _)| m).collect::>(), + messages.iter().collect::>() + ); + assert!(log.iter().all(|(_, c)| c == "cdb-migrate"), "{log:?}"); + let dirty: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT COUNT(*) FROM `{}/{DEV}`.dolt_status", + db.name + ))) + .fetch_one(&mut conn) + .await?; + assert_eq!(dirty, 0); + conn.close().await?; + assert_eq!(db.applied_versions().await?, expected); + assert!(!scratch_exists(&db).await?); + Ok(db.destroy().await?) +} + +#[tokio::test] +async fn rerun_is_a_no_op() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + db.setup().await?; + let before = db.hash_of(DEV).await?; + + assert_eq!(db.setup().await?, Vec::::new()); + + assert_eq!(db.hash_of(DEV).await?, before); + assert!(!scratch_exists(&db).await?); + Ok(db.destroy().await?) +} + +/// Two runs at once: the named lock serializes them. +#[tokio::test] +async fn concurrent_runs_apply_each_migration_once() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + let migrator = migrator(); + + let (a, b) = tokio::join!(db.migrate_with(&migrator), db.migrate_with(&migrator)); + let (mut a, b) = (a?, b?); + + a.extend(b); + a.sort_unstable(); + assert_eq!(a, up_versions(&migrator)); + assert_eq!(db.applied_versions().await?, up_versions(&migrator)); + Ok(db.destroy().await?) +} + +#[tokio::test] +async fn older_bundle_runs_against_newer_migrations() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + db.setup().await?; + apply_newer_migration(&db).await?; + + db.setup().await?; + + Ok(db.destroy().await?) +} + +#[tokio::test] +async fn strict_migrator_rejects_newer_migrations() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + db.setup().await?; + apply_newer_migration(&db).await?; + let mut migrator = migrator(); + migrator.set_ignore_missing(false); + + let result = db.migrate_with(&migrator).await; + + assert!( + matches!( + result, + Err(Error::Migrate(MigrateError::VersionMissing(9999))) + ), + "{result:?}" + ); + Ok(db.destroy().await?) +} + +#[tokio::test] +async fn rejects_changed_migration() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + db.setup().await?; + db.commit_on_dev("UPDATE _sqlx_migrations SET checksum = X'00' WHERE version = 1") + .await?; + + let result = db.migrate_with(&migrator()).await; + + assert!( + matches!( + result, + Err(Error::Migrate(MigrateError::VersionMismatch(1))) + ), + "{result:?}" + ); + Ok(db.destroy().await?) +} + +/// The failed migration stays on the scratch branch, half applied; the next run starts over. +#[tokio::test] +async fn failed_migration_leaves_dev_untouched() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + db.setup().await?; + let before = db.hash_of(DEV).await?; + + let result = db + .migrate_with(&migrator_with( + "CREATE TABLE t (id int); CREATE TABLE t (id int);", + )) + .await; + + assert!(result.is_err(), "{result:?}"); + assert_eq!(db.hash_of(DEV).await?, before); + assert_eq!( + db.migrate_with(&migrator_with("CREATE TABLE t (id int)")) + .await?, + vec![9998] + ); + Ok(db.destroy().await?) +} + +/// eDd commits content to `dev` while migrations apply: the run ends in a merge commit with both. +#[tokio::test] +async fn merges_when_dev_moved() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + let migrator = migrator(); + let mut conn = db.connect().await?; + + let applied = apply_on_scratch(&mut conn, &migrator).await?; + db.commit_on_dev( + "CREATE TABLE content (id int PRIMARY KEY); INSERT INTO content VALUES (1)", + ) + .await?; + merge_into_dev(&mut conn, &applied).await?; + drop_scratch(&mut conn).await?; + + let message: String = sqlx::query_scalar("SELECT message FROM dolt_log(?) LIMIT 1") + .bind(DEV) + .fetch_one(&mut conn) + .await?; + let versions = up_versions(&migrator); + assert_eq!( + message, + format!( + "cdb-migrate: merge {:04}..{:04}", + versions[0], + versions[versions.len() - 1] + ) + ); + let content: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM content") + .fetch_one(&mut conn) + .await?; + assert_eq!(content, 1); + conn.close().await?; + assert_eq!(db.applied_versions().await?, up_versions(&migrator)); + Ok(db.destroy().await?) +} + +#[tokio::test] +async fn never_touches_other_branches() -> Result<(), Error> { + let db = TestDb::new().await?.unwrap(); + let mut conn = db.connect().await?; + sqlx::query("CALL DOLT_BRANCH('prd', ?)") + .bind(DEV) + .execute(&mut conn) + .await?; + conn.close().await?; + let before = db.hash_of("prd").await?; + + db.setup().await?; + + assert_eq!(db.hash_of("prd").await?, before); + Ok(db.destroy().await?) +} From ab49496cef8659d5319e8eddef102bfcbb2e4cac Mon Sep 17 00:00:00 2001 From: Oliver Eyton-Williams Date: Tue, 10 Nov 2026 18:52:20 +0100 Subject: [PATCH 2/2] refactor: format --- databases/curriculum/tests/common/mod.rs | 9 ++++----- databases/curriculum/tests/exams.rs | 24 +++++++++++++++++------- databases/curriculum/tests/migrations.rs | 10 +++------- 3 files changed, 24 insertions(+), 19 deletions(-) diff --git a/databases/curriculum/tests/common/mod.rs b/databases/curriculum/tests/common/mod.rs index 4453fb8..849455c 100644 --- a/databases/curriculum/tests/common/mod.rs +++ b/databases/curriculum/tests/common/mod.rs @@ -29,11 +29,10 @@ impl TestDb { }; // The server's `init.defaultbranch` may not be `dev`. let mut conn = db.connect().await?; - let has_dev: i64 = - sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") - .bind("dev") - .fetch_one(&mut conn) - .await?; + let has_dev: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dolt_branches WHERE name = ?") + .bind("dev") + .fetch_one(&mut conn) + .await?; if has_dev == 0 { sqlx::query("CALL DOLT_BRANCH(?)") .bind("dev") diff --git a/databases/curriculum/tests/exams.rs b/databases/curriculum/tests/exams.rs index 32c11bc..0dde4e1 100644 --- a/databases/curriculum/tests/exams.rs +++ b/databases/curriculum/tests/exams.rs @@ -17,7 +17,11 @@ impl TestDb { pub async fn pool(&self) -> sqlx::Result { sqlx::mysql::MySqlPoolOptions::new() - .connect_with(self.options.clone().database(&format!("{}/{DEV}", self.name))) + .connect_with( + self.options + .clone() + .database(&format!("{}/{DEV}", self.name)), + ) .await } } @@ -28,14 +32,20 @@ async fn creates_exam_in_migrated_database() -> Result<(), Box Result<(), Error> { let mut conn = db.connect().await?; let applied = apply_on_scratch(&mut conn, &migrator).await?; - db.commit_on_dev( - "CREATE TABLE content (id int PRIMARY KEY); INSERT INTO content VALUES (1)", - ) - .await?; + db.commit_on_dev("CREATE TABLE content (id int PRIMARY KEY); INSERT INTO content VALUES (1)") + .await?; merge_into_dev(&mut conn, &applied).await?; drop_scratch(&mut conn).await?;