From 8393095be267e1d7fb1687feb42d011a84cd8713 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 8 Jun 2021 03:16:04 -0400 Subject: [PATCH] job dead letter queueing (attempt count + config) --- src/job_queries/tts_download_job_queries.rs | 17 ++++++++++----- src/job_queries/w2l_download_job_queries.rs | 22 +++++++++++++------- src/job_queries/w2l_inference_job_queries.rs | 22 +++++++++++++------- src/tts_download_job.rs | 12 ++++++++++- src/tts_inference_job.rs | 4 ++-- src/w2l_download_job.rs | 10 +++++++-- src/w2l_inference_job.rs | 10 +++++++-- 7 files changed, 71 insertions(+), 26 deletions(-) diff --git a/src/job_queries/tts_download_job_queries.rs b/src/job_queries/tts_download_job_queries.rs index 86d0e8f771..aebab0804f 100644 --- a/src/job_queries/tts_download_job_queries.rs +++ b/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, ) diff --git a/src/job_queries/w2l_download_job_queries.rs b/src/job_queries/w2l_download_job_queries.rs index 8df5dc5bba..93c5577d96 100644 --- a/src/job_queries/w2l_download_job_queries.rs +++ b/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, ) diff --git a/src/job_queries/w2l_inference_job_queries.rs b/src/job_queries/w2l_inference_job_queries.rs index 66bad12dc2..d17c198c2a 100644 --- a/src/job_queries/w2l_inference_job_queries.rs +++ b/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, ) diff --git a/src/tts_download_job.rs b/src/tts_download_job.rs index b220c29c3b..cc5eaf6d2a 100644 --- a/src/tts_download_job.rs +++ b/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) -> 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; } } } diff --git a/src/tts_inference_job.rs b/src/tts_inference_job.rs index 3d4778999c..5a955a831d 100644 --- a/src/tts_inference_job.rs +++ b/src/tts_inference_job.rs @@ -270,8 +270,8 @@ async fn process_jobs(inferencer: &Inferencer, jobs: Vec) &inferencer.mysql_pool, &job, failure_reason, - inferencer.job_max_attempts) - .await; + inferencer.job_max_attempts + ).await; } } } diff --git a/src/w2l_download_job.rs b/src/w2l_download_job.rs index 1307714918..eb8f82c001 100644 --- a/src/w2l_download_job.rs +++ b/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 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) let _r = mark_w2l_inference_job_failure( &inferencer.mysql_pool, &job, - failure_reason) - .await; + failure_reason, + inferencer.job_max_attempts + ).await; } } }