fix(schema-engine): report rolled back migrations as unapplied (#5817)

## Summary

Fixes prisma/prisma#29601.

`diagnoseMigrationHistory` was matching a migration from the filesystem
with any migration table row that had the same name. That included rows
marked with `rolled_back_at`, even though those rows are ignored by
migrate when they are not null.

This changes the history comparison to ignore rolled-back rows when
deciding which migrations are already present in the database. A
rolled-back migration that still exists on disk is now reported as
unapplied, which lets `migrate status` return a non-clean result instead
of reporting the schema as up to date.

## Tests

```sh
cargo fmt --check
git diff --check
make dev-sqlite
set -a; . ./.test_database_urls/sqlite; set +a
CARGO_BUILD_JOBS=2 RUST_TEST_THREADS=1 cargo test -p sql-migration-tests diagnose_migrations_history_reports_rolled_back_migration_as_unapplied -- --test-threads 1
CARGO_BUILD_JOBS=2 RUST_TEST_THREADS=1 cargo test -p sql-migration-tests diagnose_migration_history_tests -- --test-threads 1
CARGO_BUILD_JOBS=2 cargo check -p schema-core -p schema-commands -p sql-migration-tests
```

Author: goutamadwant

schema-engine/commands/src/commands/diagnose_migration_history.rs

@@ -90,9 +90,9 @@ pub async fn diagnose_migration_history(
 
     // Check filesystem history against database history.
     for (index, fs_migration) in migrations_from_filesystem.migration_directories.iter().enumerate() {
-        let corresponding_db_migration = migrations_from_database
-            .iter()
-            .find(|db_migration| db_migration.migration_name == fs_migration.migration_name());
+        let corresponding_db_migration = migrations_from_database.iter().find(|db_migration| {
+            db_migration.migration_name == fs_migration.migration_name() && db_migration.rolled_back_at.is_none()
+        });
 
         match corresponding_db_migration {
             Some(db_migration)
@@ -107,13 +107,17 @@ pub async fn diagnose_migration_history(
         }
     }
 
-    for (index, db_migration) in migrations_from_database.iter().enumerate() {
+    let active_migrations_from_database = migrations_from_database
+        .iter()
+        .filter(|migration| migration.rolled_back_at.is_none());
+
+    for (index, db_migration) in active_migrations_from_database.enumerate() {
         let corresponding_fs_migration = migrations_from_filesystem
             .migration_directories
             .iter()
             .find(|fs_migration| db_migration.migration_name == fs_migration.migration_name());
 
-        if db_migration.finished_at.is_none() && db_migration.rolled_back_at.is_none() {
+        if db_migration.finished_at.is_none() {
             diagnostics.failed_migrations.push(db_migration);
         }
 

schema-engine/core/src/commands/diagnose_migration_history_cli.rs

@@ -30,9 +30,9 @@ pub async fn diagnose_migration_history_cli(
 
     // Check filesystem history against database history.
     for (index, fs_migration) in migrations_from_filesystem.migration_directories.iter().enumerate() {
-        let corresponding_db_migration = migrations_from_database
-            .iter()
-            .find(|db_migration| db_migration.migration_name == fs_migration.migration_name());
+        let corresponding_db_migration = migrations_from_database.iter().find(|db_migration| {
+            db_migration.migration_name == fs_migration.migration_name() && db_migration.rolled_back_at.is_none()
+        });
 
         match corresponding_db_migration {
             Some(db_migration)
@@ -47,13 +47,17 @@ pub async fn diagnose_migration_history_cli(
         }
     }
 
-    for (index, db_migration) in migrations_from_database.iter().enumerate() {
+    let active_migrations_from_database = migrations_from_database
+        .iter()
+        .filter(|migration| migration.rolled_back_at.is_none());
+
+    for (index, db_migration) in active_migrations_from_database.enumerate() {
         let corresponding_fs_migration = migrations_from_filesystem
             .migration_directories
             .iter()
             .find(|fs_migration| db_migration.migration_name == fs_migration.migration_name());
 
-        if db_migration.finished_at.is_none() && db_migration.rolled_back_at.is_none() {
+        if db_migration.finished_at.is_none() {
             diagnostics.failed_migrations.push(db_migration);
         }
 

schema-engine/sql-migration-tests/tests/migrations/diagnose_migration_history_tests.rs

@@ -282,6 +282,67 @@ fn diagnose_migrations_history_can_detect_when_the_database_is_behind(api: TestA
     assert!(error_in_unapplied_migration.is_none());
 }
 
+#[test_connector]
+fn diagnose_migrations_history_reports_rolled_back_migration_as_unapplied(api: TestApi) {
+    let directory = api.create_migrations_directory();
+
+    let dm1 = api.datamodel_with_provider(
+        r#"
+        model Cat {
+            id   Int @id
+            name String
+        }
+    "#,
+    );
+
+    api.create_migration("initial", &dm1, &directory).send_sync();
+    api.apply_migrations(&directory).send_sync();
+
+    let dm2 = api.datamodel_with_provider(
+        r#"
+        model Cat {
+            id          Int @id
+            name        String
+            fluffiness  Float
+        }
+    "#,
+    );
+
+    let rolled_back_migration_name = api
+        .create_migration("02_add_fluffiness", &dm2, &directory)
+        .send_sync()
+        .modify_migration(|script| {
+            script.clear();
+            script.push_str("SELECT YOLO;");
+        })
+        .into_output()
+        .generated_migration_name;
+
+    api.apply_migrations(&directory).send_unwrap_err();
+    api.mark_migration_rolled_back(&rolled_back_migration_name).send();
+
+    let DiagnoseMigrationHistoryOutput {
+        drift,
+        history,
+        failed_migration_names,
+        edited_migration_names,
+        has_migrations_table,
+        error_in_unapplied_migration,
+    } = api.diagnose_migration_history(&directory).send_sync().into_output();
+
+    assert!(drift.is_none());
+    assert!(failed_migration_names.is_empty());
+    assert!(edited_migration_names.is_empty());
+    assert_eq!(
+        history,
+        Some(HistoryDiagnostic::DatabaseIsBehind {
+            unapplied_migration_names: vec![rolled_back_migration_name],
+        })
+    );
+    assert!(has_migrations_table);
+    assert!(error_in_unapplied_migration.is_none());
+}
+
 #[test_connector]
 fn diagnose_migrations_history_can_detect_when_the_folder_is_behind(api: TestApi) {
     let directory = api.create_migrations_directory();