fix: ensure migration progress is not lost for PG, mysql and sqlite (#…

…1991)

* fix: ensure migration progress is not lost for PG

Fixes #1966.

* fix: ensure migration progress is not lost for sqlite

This is similar to #1966.

* fix: ensure reverse migration progress is not lost for PG

See #1966.

* fix: ensure reverse migration progress is not lost for sqlite

See #1966.

* fix: ensure migration progress is not lost for mysql

This is similar to #1966.

* fix: ensure reverse migration progress is not lost for mysql

See #1966.

* test: check migration type as well

* test: extend migrations testing

* fix: work around MySQL implicit commits

* refactor: simplify migration testing

Cargo.toml

@@ -212,6 +212,11 @@ name = "sqlite-test-attr"
 path = "tests/sqlite/test-attr.rs"
 required-features = ["sqlite", "macros", "migrate"]
 
+[[test]]
+name = "sqlite-migrate"
+path = "tests/sqlite/migrate.rs"
+required-features = ["sqlite", "macros", "migrate"]
+
 #
 # MySQL
 #
@@ -241,6 +246,11 @@ name = "mysql-test-attr"
 path = "tests/mysql/test-attr.rs"
 required-features = ["mysql", "macros", "migrate"]
 
+[[test]]
+name = "mysql-migrate"
+path = "tests/mysql/migrate.rs"
+required-features = ["mysql", "macros", "migrate"]
+
 #
 # PostgreSQL
 #
@@ -275,6 +285,11 @@ name = "postgres-test-attr"
 path = "tests/postgres/test-attr.rs"
 required-features = ["postgres", "macros", "migrate"]
 
+[[test]]
+name = "postgres-migrate"
+path = "tests/postgres/migrate.rs"
+required-features = ["postgres", "macros", "migrate"]
+
 #
 # Microsoft SQL Server (MSSQL)
 #

sqlx-core/src/migrate/migration_type.rs

@@ -1,5 +1,5 @@
 /// Migration Type represents the type of migration
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, PartialEq)]
 pub enum MigrationType {
     /// Simple migration are single file migrations with no up / down queries
     Simple,

sqlx-core/src/mysql/migrate.rs

@@ -1,4 +1,4 @@
-use crate::connection::ConnectOptions;
+use crate::connection::{ConnectOptions, Connection};
 use crate::error::Error;
 use crate::executor::Executor;
 use crate::migrate::MigrateError;
@@ -209,29 +209,67 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
         migration: &'m Migration,
     ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
         Box::pin(async move {
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
+            // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
+            // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
+            // and update it once the actual transaction completed.
+            let mut tx = self.begin().await?;
             let start = Instant::now();
 
-            let res = self.execute(&*migration.sql).await;
-
-            let elapsed = start.elapsed();
-
+            // For MySQL we cannot really isolate migrations due to implicit commits caused by table modification, see
+            // https://dev.mysql.com/doc/refman/8.0/en/implicit-commit.html
+            //
+            // To somewhat try to detect this, we first insert the migration into the migration table with
+            // `success=FALSE` and later modify the flag.
+            //
             // language=MySQL
             let _ = query(
                 r#"
     INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
-    VALUES ( ?, ?, ?, ?, ? )
+    VALUES ( ?, ?, FALSE, ?, -1 )
                 "#,
             )
             .bind(migration.version)
             .bind(&*migration.description)
-            .bind(res.is_ok())
             .bind(&*migration.checksum)
+            .execute(&mut tx)
+            .await?;
+
+            let _ = tx.execute(&*migration.sql).await?;
+
+            // language=MySQL
+            let _ = query(
+                r#"
+    UPDATE _sqlx_migrations
+    SET success = TRUE
+    WHERE version = ?
+                "#,
+            )
+            .bind(migration.version)
+            .execute(&mut tx)
+            .await?;
+
+            tx.commit().await?;
+
+            // Update `elapsed_time`.
+            // NOTE: The process may disconnect/die at this point, so the elapsed time value might be lost. We accept
+            //       this small risk since this value is not super important.
+
+            let elapsed = start.elapsed();
+
+            let _ = query(
+                r#"
+    UPDATE _sqlx_migrations
+    SET execution_time = ?
+    WHERE version = ?
+                "#,
+            )
             .bind(elapsed.as_nanos() as i64)
+            .bind(migration.version)
             .execute(self)
             .await?;
 
-            res?;
-
             Ok(elapsed)
         })
     }
@@ -241,18 +279,41 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
         migration: &'m Migration,
     ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
         Box::pin(async move {
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
+            let mut tx = self.begin().await?;
             let start = Instant::now();
 
-            self.execute(&*migration.sql).await?;
+            // For MySQL we cannot really isolate migrations due to implicit commits caused by table modification, see
+            // https://dev.mysql.com/doc/refman/8.0/en/implicit-commit.html
+            //
+            // To somewhat try to detect this, we first insert the migration into the migration table with
+            // `success=FALSE` and later remove the migration altogether.
+            //
+            // language=MySQL
+            let _ = query(
+                r#"
+    UPDATE _sqlx_migrations
+    SET success = FALSE
+    WHERE version = ?
+                "#,
+            )
+            .bind(migration.version)
+            .execute(&mut tx)
+            .await?;
 
-            let elapsed = start.elapsed();
+            tx.execute(&*migration.sql).await?;
 
             // language=SQL
             let _ = query(r#"DELETE FROM _sqlx_migrations WHERE version = ?"#)
                 .bind(migration.version)
-                .execute(self)
+                .execute(&mut tx)
                 .await?;
 
+            tx.commit().await?;
+
+            let elapsed = start.elapsed();
+
             Ok(elapsed)
         })
     }

sqlx-core/src/postgres/migrate.rs

@@ -222,23 +222,44 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
             let mut tx = self.begin().await?;
             let start = Instant::now();
 
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
+            // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
+            // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
+            // and update it once the actual transaction completed.
             let _ = tx.execute(&*migration.sql).await?;
 
-            tx.commit().await?;
-
-            let elapsed = start.elapsed();
-
             // language=SQL
             let _ = query(
                 r#"
     INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
-    VALUES ( $1, $2, TRUE, $3, $4 )
+    VALUES ( $1, $2, TRUE, $3, -1 )
                 "#,
             )
             .bind(migration.version)
             .bind(&*migration.description)
             .bind(&*migration.checksum)
+            .execute(&mut tx)
+            .await?;
+
+            tx.commit().await?;
+
+            // Update `elapsed_time`.
+            // NOTE: The process may disconnect/die at this point, so the elapsed time value might be lost. We accept
+            //       this small risk since this value is not super important.
+
+            let elapsed = start.elapsed();
+
+            // language=SQL
+            let _ = query(
+                r#"
+    UPDATE _sqlx_migrations
+    SET execution_time = $1
+    WHERE version = $2
+                "#,
+            )
             .bind(elapsed.as_nanos() as i64)
+            .bind(migration.version)
             .execute(self)
             .await?;
 
@@ -251,21 +272,23 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
         migration: &'m Migration,
     ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
         Box::pin(async move {
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
             let mut tx = self.begin().await?;
             let start = Instant::now();
 
             let _ = tx.execute(&*migration.sql).await?;
 
-            tx.commit().await?;
-
-            let elapsed = start.elapsed();
-
             // language=SQL
             let _ = query(r#"DELETE FROM _sqlx_migrations WHERE version = $1"#)
                 .bind(migration.version)
-                .execute(self)
+                .execute(&mut tx)
                 .await?;
 
+            tx.commit().await?;
+
+            let elapsed = start.elapsed();
+
             Ok(elapsed)
         })
     }

sqlx-core/src/sqlite/migrate.rs

@@ -173,23 +173,44 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
             let mut tx = self.begin().await?;
             let start = Instant::now();
 
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
+            // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
+            // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
+            // and update it once the actual transaction completed.
             let _ = tx.execute(&*migration.sql).await?;
 
-            tx.commit().await?;
-
-            let elapsed = start.elapsed();
-
             // language=SQL
             let _ = query(
                 r#"
     INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
-    VALUES ( ?1, ?2, TRUE, ?3, ?4 )
+    VALUES ( ?1, ?2, TRUE, ?3, -1 )
                 "#,
             )
             .bind(migration.version)
             .bind(&*migration.description)
             .bind(&*migration.checksum)
+            .execute(&mut tx)
+            .await?;
+
+            tx.commit().await?;
+
+            // Update `elapsed_time`.
+            // NOTE: The process may disconnect/die at this point, so the elapsed time value might be lost. We accept
+            //       this small risk since this value is not super important.
+
+            let elapsed = start.elapsed();
+
+            // language=SQL
+            let _ = query(
+                r#"
+    UPDATE _sqlx_migrations
+    SET execution_time = ?1
+    WHERE version = ?2
+                "#,
+            )
             .bind(elapsed.as_nanos() as i64)
+            .bind(migration.version)
             .execute(self)
             .await?;
 
@@ -202,21 +223,23 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
         migration: &'m Migration,
     ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
         Box::pin(async move {
+            // Use a single transaction for the actual migration script and the essential bookeeping so we never
+            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
             let mut tx = self.begin().await?;
             let start = Instant::now();
 
             let _ = tx.execute(&*migration.sql).await?;
 
-            tx.commit().await?;
-
-            let elapsed = start.elapsed();
-
             // language=SQL
             let _ = query(r#"DELETE FROM _sqlx_migrations WHERE version = ?1"#)
                 .bind(migration.version)
-                .execute(self)
+                .execute(&mut tx)
                 .await?;
 
+            tx.commit().await?;
+
+            let elapsed = start.elapsed();
+
             Ok(elapsed)
         })
     }

tests/migrate/macro.rs

@@ -1,20 +1,29 @@
 use sqlx::migrate::Migrator;
 use std::path::Path;
 
-static EMBEDDED: Migrator = sqlx::migrate!("tests/migrate/migrations");
+static EMBEDDED_SIMPLE: Migrator = sqlx::migrate!("tests/migrate/migrations_simple");
+static EMBEDDED_REVERSIBLE: Migrator = sqlx::migrate!("tests/migrate/migrations_reversible");
 
 #[sqlx_macros::test]
 async fn same_output() -> anyhow::Result<()> {
-    let runtime = Migrator::new(Path::new("tests/migrate/migrations")).await?;
+    let runtime_simple = Migrator::new(Path::new("tests/migrate/migrations_simple")).await?;
+    let runtime_reversible =
+        Migrator::new(Path::new("tests/migrate/migrations_reversible")).await?;
 
-    assert_eq!(runtime.migrations.len(), EMBEDDED.migrations.len());
+    assert_same(&EMBEDDED_SIMPLE, &runtime_simple);
+    assert_same(&EMBEDDED_REVERSIBLE, &runtime_reversible);
 
-    for (e, r) in EMBEDDED.iter().zip(runtime.iter()) {
+    Ok(())
+}
+
+fn assert_same(embedded: &Migrator, runtime: &Migrator) {
+    assert_eq!(runtime.migrations.len(), embedded.migrations.len());
+
+    for (e, r) in embedded.iter().zip(runtime.iter()) {
         assert_eq!(e.version, r.version);
         assert_eq!(e.description, r.description);
+        assert_eq!(e.migration_type, r.migration_type);
         assert_eq!(e.sql, r.sql);
         assert_eq!(e.checksum, r.checksum);
     }
-
-    Ok(())
 }

tests/migrate/migrations_reversible/20220721124650_add_table.down.sql

@@ -0,0 +1 @@
+DROP TABLE migrations_reversible_test;

tests/migrate/migrations_reversible/20220721124650_add_table.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE migrations_reversible_test (
+    some_id BIGINT NOT NULL PRIMARY KEY,
+    some_payload BIGINT NOT NUll
+);
+
+INSERT INTO migrations_reversible_test (some_id, some_payload)
+VALUES (1, 100);

tests/migrate/migrations_reversible/20220721125033_modify_column.down.sql

@@ -0,0 +1,2 @@
+UPDATE migrations_reversible_test
+SET some_payload = some_payload - 1;

tests/migrate/migrations_reversible/20220721125033_modify_column.up.sql

@@ -0,0 +1,2 @@
+UPDATE migrations_reversible_test
+SET some_payload = some_payload + 1;