diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index d7dd9bef80..05eae95988 100644 --- a/src/infra/src/file_list/mod.rs +++ b/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 for FileListJobStatus { - fn from(status: i64) -> Self { +impl From 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); diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 44357d9d58..1e471a91b5 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -1340,8 +1340,19 @@ DO UPDATE SET return Err(e.into()); } }; - let id = ret.try_get::("id").unwrap_or_default(); - let status = ret.try_get::("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::("id") + .and_then(|id| Ok((id, ret.try_get::("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 = 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" + ); } } diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index 216ef261df..042144f242 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -1046,8 +1046,18 @@ DO UPDATE SET return Err(e.into()); } }; - let id = ret.try_get::("id").unwrap_or_default(); - let status = ret.try_get::("status").unwrap_or_default(); + let (id, status) = match ret + .try_get::("id") + .and_then(|id| Ok((id, ret.try_get::("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) =