fix(compactor): re-arm finished merge jobs so the hour-end merge still runs (#14543)

src/infra/src/file_list/mod.rs

@@ -891,7 +891,7 @@ pub struct MergeJobPendingRecord {
 }
 
 #[derive(Debug, Clone, sqlx::Type, PartialEq, Default)]
-#[repr(i64)]
+#[repr(i32)]
 pub enum FileListJobStatus {
     #[default]
     Pending,
@@ -899,8 +899,8 @@ pub enum FileListJobStatus {
     Done,
 }
 
-impl From<i64> for FileListJobStatus {
-    fn from(status: i64) -> Self {
+impl From<i32> for FileListJobStatus {
+    fn from(status: i32) -> Self {
         match status {
             0 => Self::Pending,
             1 => Self::Running,
@@ -976,10 +976,10 @@ mod tests {
         assert!(parse_stream_key("org/logs/stream/extra").is_none());
     }
 
-    // ── FileListJobStatus::from(i64) ──────────────────────────────────────────
+    // ── FileListJobStatus::from(i32) ──────────────────────────────────────────
 
     #[test]
-    fn test_file_list_job_status_from_i64() {
+    fn test_file_list_job_status_from_i32() {
         assert_eq!(FileListJobStatus::from(0), FileListJobStatus::Pending);
         assert_eq!(FileListJobStatus::from(1), FileListJobStatus::Running);
         assert_eq!(FileListJobStatus::from(2), FileListJobStatus::Done);

src/infra/src/file_list/postgres.rs

@@ -1340,8 +1340,19 @@ DO UPDATE SET
                 return Err(e.into());
             }
         };
-        let id = ret.try_get::<i64, &str>("id").unwrap_or_default();
-        let status = ret.try_get::<i64, &str>("status").unwrap_or_default();
+        // status is an INT column: decoding it as i64 fails and must not be read as Pending
+        let (id, status) = match ret
+            .try_get::<i64, &str>("id")
+            .and_then(|id| Ok((id, ret.try_get::<i32, &str>("status")?)))
+        {
+            Ok(v) => v,
+            Err(e) => {
+                if let Err(e) = tx.rollback().await {
+                    log::error!("[POSTGRES] rollback add job error: {e}");
+                }
+                return Err(e.into());
+            }
+        };
         if id > 0
             && super::FileListJobStatus::from(status) == super::FileListJobStatus::Done
             && let Err(e) =
@@ -3525,7 +3536,7 @@ mod tests {
     use tokio::sync::OnceCell;
 
     use super::*;
-    use crate::file_list::FileList;
+    use crate::file_list::{FileList, FileListJobStatus};
 
     static _INIT: Once = Once::new();
     static DB_POOL: OnceCell<PgPool> = OnceCell::const_new();
@@ -3635,6 +3646,11 @@ mod tests {
         )
         .execute(pool)
         .await?;
+        sqlx::query(
+            "CREATE UNIQUE INDEX IF NOT EXISTS file_list_jobs_stream_offsets_idx ON file_list_jobs (stream, offsets)",
+        )
+        .execute(pool)
+        .await?;
 
         sqlx::query(
             r#"
@@ -5409,5 +5425,31 @@ mod tests {
                 .any(|j| j.stream == free_stream && j.id == rowlock_max[&free_stream]),
             "the unlocked stream's max-id job must still be claimed in the same call"
         );
+
+        // ---- add_job re-arms a Done job so the hour-end pass still runs ----
+        let rearm_stream = format!("rearm_{pid}");
+        let rearm_id = postgres_list
+            .add_job("fix4_org", StreamType::Logs, &rearm_stream, 3_600_000_000)
+            .await
+            .unwrap();
+        postgres_list.set_job_done(&[rearm_id]).await.unwrap();
+        let again_id = postgres_list
+            .add_job("fix4_org", StreamType::Logs, &rearm_stream, 3_600_000_000)
+            .await
+            .unwrap();
+        assert_eq!(
+            again_id, rearm_id,
+            "add_job must reuse the existing job row"
+        );
+        let status: i32 = sqlx::query_scalar("SELECT status FROM file_list_jobs WHERE id = $1")
+            .bind(rearm_id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(
+            status,
+            FileListJobStatus::Pending as i32,
+            "add_job must re-arm a Done job to Pending"
+        );
     }
 }

src/infra/src/file_list/sqlite.rs

@@ -1046,8 +1046,18 @@ DO UPDATE SET
                 return Err(e.into());
             }
         };
-        let id = ret.try_get::<i64, &str>("id").unwrap_or_default();
-        let status = ret.try_get::<i64, &str>("status").unwrap_or_default();
+        let (id, status) = match ret
+            .try_get::<i64, &str>("id")
+            .and_then(|id| Ok((id, ret.try_get::<i32, &str>("status")?)))
+        {
+            Ok(v) => v,
+            Err(e) => {
+                if let Err(e) = tx.rollback().await {
+                    log::error!("[SQLITE] rollback add job error: {e}");
+                }
+                return Err(e.into());
+            }
+        };
         if id > 0
             && super::FileListJobStatus::from(status) == super::FileListJobStatus::Done
             && let Err(e) =