job dead letter queueing (attempt count + config)

src/job_queries/tts_download_job_queries.rs

@@ -124,6 +124,7 @@ FOR UPDATE
 UPDATE tts_model_upload_jobs
 SET
   status = 'started',
+  attempt_count = attempt_count + 1,
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
@@ -140,11 +141,17 @@ WHERE id = ?
 pub async fn mark_tts_upload_job_failure(
   pool: &MySqlPool,
   job: &TtsUploadJobRecord,
-  failure_reason: &str
-) -> AnyhowResult<()>
-{
+  failure_reason: &str,
+  max_attempts: i32
+) -> AnyhowResult<()> {
+
   // statuses: "attempt_failed", "complete_failure", "dead"
-  let status = "attempt_failed";
+  let mut next_status = "attempt_failed";
+
+  if job.attempt_count >= max_attempts {
+    // NB: Job attempt count is incremented at start
+    next_status = "dead";
+  }
 
   let query_result = sqlx::query!(
         r#"
@@ -155,7 +162,7 @@ SET
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
-        status,
+        next_status,
         failure_reason.to_string(),
         job.id,
     )

src/job_queries/w2l_download_job_queries.rs

@@ -123,6 +123,7 @@ FOR UPDATE
 UPDATE w2l_template_upload_jobs
 SET
   status = 'started',
+  attempt_count = attempt_count + 1,
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
@@ -136,13 +137,20 @@ WHERE id = ?
   Ok(true)
 }
 
-pub async fn mark_w2l_template_upload_job_failure(pool: &MySqlPool,
-                                                  job: &W2lTemplateUploadJobRecord,
-                                                  failure_reason: &str)
-  -> AnyhowResult<()>
-{
+pub async fn mark_w2l_template_upload_job_failure(
+  pool: &MySqlPool,
+  job: &W2lTemplateUploadJobRecord,
+  failure_reason: &str,
+  max_attempts: i32
+) -> AnyhowResult<()> {
+
   // statuses: "attempt_failed", "complete_failure", "dead"
-  let status = "attempt_failed";
+  let mut next_status = "attempt_failed";
+
+  if job.attempt_count >= max_attempts {
+    // NB: Job attempt count is incremented at start
+    next_status = "dead";
+  }
 
   let query_result = sqlx::query!(
         r#"
@@ -153,7 +161,7 @@ SET
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
-        status,
+        next_status,
         failure_reason.to_string(),
         job.id,
     )

src/job_queries/w2l_inference_job_queries.rs

@@ -165,6 +165,7 @@ FOR UPDATE
 UPDATE w2l_inference_jobs
 SET
   status = 'started',
+  attempt_count = attempt_count + 1,
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
@@ -179,13 +180,20 @@ WHERE id = ?
 }
 
 
-pub async fn mark_w2l_inference_job_failure(pool: &MySqlPool,
-                                                  job: &W2lInferenceJobRecord,
-                                                  failure_reason: &str)
-                                                  -> AnyhowResult<()>
-{
+pub async fn mark_w2l_inference_job_failure(
+  pool: &MySqlPool,
+  job: &W2lInferenceJobRecord,
+  failure_reason: &str,
+  max_attempts: i32
+) -> AnyhowResult<()> {
+
   // statuses: "attempt_failed", "complete_failure", "dead"
-  let status = "attempt_failed";
+  let mut next_status = "attempt_failed";
+
+  if job.attempt_count >= max_attempts {
+    // NB: Job attempt count is incremented at start
+    next_status = "dead";
+  }
 
   let query_result = sqlx::query!(
         r#"
@@ -196,7 +204,7 @@ SET
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
-        status,
+        next_status,
         failure_reason.to_string(),
         job.id,
     )

src/tts_download_job.rs

@@ -81,6 +81,10 @@ struct Downloader {
 
   // Sleep between batches
   pub job_batch_wait_millis: u64,
+
+  // Max job attempts before failure.
+  // NB: This is an i32 so we don't need to convert to db column type.
+  pub job_max_attempts: i32,
 }
 
 #[tokio::main]
@@ -164,6 +168,7 @@ async fn main() -> AnyhowResult<()> {
     firehose_publisher,
     tts_check: tts_model_check_command,
     job_batch_wait_millis: common_env.job_batch_wait_millis,
+    job_max_attempts: common_env.job_max_attempts as i32,
   };
 
   main_loop(downloader).await;
@@ -223,7 +228,12 @@ async fn process_jobs(downloader: &Downloader, jobs: Vec<TtsUploadJobRecord>) ->
       Err(e) => {
         warn!("Failure to process job: {:?}", e);
         let failure_reason = "";
-        let _r = mark_tts_upload_job_failure(&downloader.mysql_pool, &job, failure_reason).await;
+        let _r = mark_tts_upload_job_failure(
+          &downloader.mysql_pool,
+          &job,
+          failure_reason,
+          downloader.job_max_attempts
+        ).await;
       }
     }
   }

src/tts_inference_job.rs

@@ -270,8 +270,8 @@ async fn process_jobs(inferencer: &Inferencer, jobs: Vec<TtsInferenceJobRecord>)
           &inferencer.mysql_pool,
           &job,
           failure_reason,
-          inferencer.job_max_attempts)
-            .await;
+          inferencer.job_max_attempts
+        ).await;
       }
     }
   }

src/w2l_download_job.rs

@@ -103,6 +103,10 @@ struct Downloader {
 
   // Sleep between batches
   pub job_batch_wait_millis: u64,
+
+  // Max job attempts before failure.
+  // NB: This is an i32 so we don't need to convert to db column type.
+  pub job_max_attempts: i32,
 }
 
 #[tokio::main]
@@ -209,6 +213,7 @@ async fn main() -> AnyhowResult<()> {
     debug_face_detect_sleep_millis: easyenv::get_env_num("DEBUG_FACE_DETECT_SLEEP_MILLIS", 0)?,
     debug_job_end_sleep_millis: easyenv::get_env_num("DEBUG_JOB_END_SLEEP_MILLIS", 0)?,
     job_batch_wait_millis: common_env.job_batch_wait_millis,
+    job_max_attempts: common_env.job_max_attempts as i32,
   };
 
   main_loop(downloader).await;
@@ -275,8 +280,9 @@ async fn process_jobs(downloader: &Downloader, jobs: Vec<W2lTemplateUploadJobRec
         let _r = mark_w2l_template_upload_job_failure(
           &downloader.mysql_pool,
           &job,
-          failure_reason)
-          .await;
+          failure_reason,
+          downloader.job_max_attempts
+        ).await;
       }
     }
   }

src/w2l_inference_job.rs

@@ -102,6 +102,10 @@ struct Inferencer {
 
   // Sleep between batches
   pub job_batch_wait_millis: u64,
+
+  // Max job attempts before failure.
+  // NB: This is an i32 so we don't need to convert to db column type.
+  pub job_max_attempts: i32,
 }
 
 #[tokio::main]
@@ -225,6 +229,7 @@ async fn main() -> AnyhowResult<()> {
     w2l_model_filename,
     w2l_end_bump_filename,
     job_batch_wait_millis: common_env.job_batch_wait_millis,
+    job_max_attempts: common_env.job_max_attempts as i32,
   };
 
   main_loop(inferencer).await;
@@ -291,8 +296,9 @@ async fn process_jobs(inferencer: &Inferencer, jobs: Vec<W2lInferenceJobRecord>)
         let _r = mark_w2l_inference_job_failure(
           &inferencer.mysql_pool,
           &job,
-          failure_reason)
-          .await;
+          failure_reason,
+          inferencer.job_max_attempts
+        ).await;
       }
     }
   }