From 1e11fdb35d05d015af415e554ecd2552f917fbf2 Mon Sep 17 00:00:00 2001 From: kasisnu <5044120+kasisnu@users.noreply.github.com> Date: Wed, 29 Nov 2023 12:43:49 +0530 Subject: [PATCH] fail model deleted inference jobs --- .../job/job_loop/determine_dependency_status.rs | 14 +++++++++----- .../inference_job/src/job/job_loop/main_loop.rs | 11 +++++++++-- .../src/job/job_loop/process_single_job.rs | 3 +-- .../src/job/job_loop/process_single_job_error.rs | 3 +++ 4 files changed, 22 insertions(+), 9 deletions(-) diff --git a/crates/service/job/inference_job/src/job/job_loop/determine_dependency_status.rs b/crates/service/job/inference_job/src/job/job_loop/determine_dependency_status.rs index 37914cd5a4..a42aff7b6e 100644 --- a/crates/service/job/inference_job/src/job/job_loop/determine_dependency_status.rs +++ b/crates/service/job/inference_job/src/job/job_loop/determine_dependency_status.rs @@ -8,8 +8,9 @@ use enums::by_table::voice_conversion_models::voice_conversion_model_type::Voice use errors::AnyhowResult; use filesys::file_exists::file_exists; use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; -use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference_improved::{get_tts_model_for_inference_improved, TtsModelForInferenceRecord}; +use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference_improved::{get_tts_model_for_inference_improved, TtsModelForInferenceError, TtsModelForInferenceRecord}; use mysql_queries::queries::voice_conversion::inference::get_voice_conversion_model_for_inference::{get_voice_conversion_model_for_inference, VoiceConversionModelForInference}; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; use crate::job_dependencies::JobDependencies; @@ -31,7 +32,7 @@ pub enum MaybeInferenceModel { VcModel(VoiceConversionModelForInference) } -pub async fn determine_dependency_status(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> AnyhowResult { +pub async fn determine_dependency_status(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result { let maybe_model = get_model_record_from_cacheable_query(job_dependencies, job).await?; let maybe_token_and_path = get_model_token_and_path(job_dependencies, &maybe_model); @@ -105,7 +106,7 @@ fn get_model_token_and_path(job_dependencies: &JobDependencies, maybe_model: &Ma } } -async fn get_model_record_from_cacheable_query(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> AnyhowResult { +async fn get_model_record_from_cacheable_query(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result { let model = match (job.inference_category, job.maybe_model_token.as_deref()) { (InferenceCategory::TextToSpeech, Some(token)) => { let maybe_model = job_dependencies.job.info.caches.tts_model_record_cache.copy_without_bump_if_unexpired(token.to_string())?; @@ -116,8 +117,11 @@ async fn get_model_record_from_cacheable_query(job_dependencies: &JobDependencie let maybe_tts_model = get_tts_model_for_inference_improved( &job_dependencies.db.mysql_pool, token) .await - .map_err(|err| anyhow!("database error: {:?}", err))?; - + .map_err(|err| + match err { + TtsModelForInferenceError::ModelDeleted => ProcessSingleJobError::ModelDeleted, + _ => ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)) + })?; match maybe_tts_model { Some(model) => { let token = token.to_string(); diff --git a/crates/service/job/inference_job/src/job/job_loop/main_loop.rs b/crates/service/job/inference_job/src/job/job_loop/main_loop.rs index 5a36700594..8add384db2 100644 --- a/crates/service/job/inference_job/src/job/job_loop/main_loop.rs +++ b/crates/service/job/inference_job/src/job/job_loop/main_loop.rs @@ -125,7 +125,7 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec { - warn!("Failure to process job: {:?}", err); + warn!("Failure to process job: {} - {:?}",job.token, err); let _r = handle_error(&job_dependencies, &job, err).await?; } } @@ -191,7 +191,14 @@ async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableInfere None, Some(FrontendFailureCategory::FaceNotDetected), ), - + ProcessSingleJobError::ModelDeleted => + ( + JobFailureClass::PermanentFailure, + ContainerHealth::Ignore, + "model deleted".to_string(), + None, + None, + ), // Non-permanent failures ProcessSingleJobError::FilesystemFull => ( diff --git a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs index 6d79e6164d..401fdd76dd 100644 --- a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs +++ b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs @@ -45,8 +45,7 @@ pub async fn process_single_job( // TODO(bt,2023-07-23): Re-review the following. It looks sus. // TODO(bt,2023-07-23): Re-review the following. It looks sus. let dependency_status = determine_dependency_status(job_dependencies, job) - .await - .map_err(|err| ProcessSingleJobError::Other(anyhow!("database or cache error: {:?}", err)))?; + .await?; if !force_execution && !dependency_status.models_already_on_filesystem { match dependency_status.maybe_model_token { diff --git a/crates/service/job/inference_job/src/job/job_loop/process_single_job_error.rs b/crates/service/job/inference_job/src/job/job_loop/process_single_job_error.rs index 560255651d..7277671180 100644 --- a/crates/service/job/inference_job/src/job/job_loop/process_single_job_error.rs +++ b/crates/service/job/inference_job/src/job/job_loop/process_single_job_error.rs @@ -25,6 +25,9 @@ pub enum ProcessSingleJobError { /// It might be important, we just haven't special cased it yet. Other(anyhow::Error), + /// Job underlying resources are missing - model deleted + ModelDeleted, + /// The job system is misconfigured. /// Retry the job, but mark as a job runner system failure (health check failure). JobSystemMisconfiguration(Option),