fix(log-ingestor): Enforce atomic job status validation using a shared lock. (y-scope#2045)

Author: LinZhihao-723

components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs

@@ -334,7 +334,7 @@ impl ClpIngestionState {
         mut tx: sqlx::Transaction<'_, sqlx::MySql>,
     ) -> anyhow::Result<()> {
         let curr_status: ClpIngestionJobStatus = sqlx::query_scalar(formatcp!(
-            r"SELECT `status` FROM `{table}` WHERE `id` = ?",
+            r"SELECT `status` FROM `{table}` WHERE `id` = ? LOCK IN SHARE MODE;",
             table = INGESTION_JOB_TABLE_NAME,
         ))
         .bind(self.job_id)