ensure job status consistency (#1576)

libsql-server/src/schema/db.rs

@@ -8,6 +8,7 @@ use crate::connection::program::Program;
 use crate::namespace::NamespaceName;
 use crate::schema::status::{MigrationJobProgress, MigrationJobSummary};
 
+use super::status::MigrationProgress;
 use super::{
     status::{MigrationJob, MigrationTask},
     Error, MigrationDetails, MigrationJobStatus, MigrationSummary, MigrationTaskStatus,
@@ -354,11 +355,44 @@ pub(super) fn get_next_pending_migration_job(
             Ok(())
         })?
         .collect::<Result<(), rusqlite::Error>>()?;
+
+        // if a crash happened before we could update the job status, we need to update the job
+        // status
+        let actual_status = status_from_progress(&job.progress);
+        if actual_status != *job.status() {
+            *job.status_mut() = actual_status;
+        }
     }
 
     Ok(job)
 }
 
+/// infer the status from the migration progress
+fn status_from_progress(progress: &MigrationProgress) -> MigrationJobStatus {
+    use MigrationTaskStatus::*;
+    if progress[DryRunFailure as usize] != 0 {
+        return MigrationJobStatus::DryRunFailure;
+    }
+
+    if progress[Failure as usize] != 0 {
+        return MigrationJobStatus::RunFailure;
+    }
+
+    if progress[Enqueued as usize] != 0 {
+        return MigrationJobStatus::WaitingDryRun;
+    }
+
+    if progress[DryRunSuccess as usize] != 0 || progress[Run as usize] != 0 {
+        return MigrationJobStatus::WaitingRun;
+    }
+
+    if progress[Success as usize] != 0 {
+        return MigrationJobStatus::RunSuccess;
+    }
+
+    MigrationJobStatus::WaitingDryRun
+}
+
 pub fn get_migration_details(
     conn: &mut rusqlite::Connection,
     schema: NamespaceName,

libsql-server/src/schema/status.rs

@@ -34,13 +34,15 @@ impl MigrationTask {
     }
 }
 
+pub(crate) type MigrationProgress = [usize; MigrationTaskStatus::num_variants()];
+
 #[derive(Debug, Clone)]
 pub struct MigrationJob {
     pub(super) schema: NamespaceName,
     pub(super) status: MigrationJobStatus,
     pub(super) job_id: i64,
     pub(super) migration: Arc<Program>,
-    pub(super) progress: [usize; MigrationTaskStatus::num_variants()],
+    pub(super) progress: MigrationProgress,
     /// error info for the task that failed the job
     pub(super) task_error: Option<(i64, String, NamespaceName)>,
 }