From b9e4087e3ea4df4ec652f4ceb7aa73c1d2175964 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 19 Sep 2021 20:20:06 -0400 Subject: [PATCH] mark jobs permanently dead if models are gone --- src/job_queries/tts_inference_job_queries.rs | 76 +++++++++++-- src/tts_inference_job.rs | 109 +++++++++++++++---- 2 files changed, 151 insertions(+), 34 deletions(-) diff --git a/src/job_queries/tts_inference_job_queries.rs b/src/job_queries/tts_inference_job_queries.rs index 63ff083d13..55f17f8efa 100644 --- a/src/job_queries/tts_inference_job_queries.rs +++ b/src/job_queries/tts_inference_job_queries.rs @@ -11,6 +11,7 @@ use crate::util::anyhow_result::AnyhowResult; use crate::util::random_prefix_crockford_token::random_prefix_crockford_token; use log::{warn, info}; use sqlx::{MySqlPool, Transaction, MySql}; +use std::error; use std::path::Path; /// table: tts_inference_jobs @@ -192,6 +193,30 @@ WHERE id = ? Ok(()) } + +pub async fn mark_tts_inference_job_permanently_dead( + pool: &MySqlPool, + job: &TtsInferenceJobRecord, + failure_reason: &str, +) -> AnyhowResult<()> { + let query_result = sqlx::query!( + r#" +UPDATE tts_inference_jobs +SET + status = "dead", + failure_reason = ?, + retry_at = NULL +WHERE id = ? + "#, + failure_reason.to_string(), + job.id, + ) + .execute(pool) + .await?; + + Ok(()) +} + pub async fn mark_tts_inference_job_done( pool: &MySqlPool, job: &TtsInferenceJobRecord, @@ -374,7 +399,7 @@ SET Ok((record_id, inference_result_token.clone())) } -pub struct TtsModelRecord2 { +pub struct TtsModelForInferenceRecord { pub model_token: String, pub tts_model_type: String, @@ -388,17 +413,40 @@ pub struct TtsModelRecord2 { pub private_bucket_hash: String, pub created_at: DateTime, pub updated_at: DateTime, + pub mod_deleted_at: Option>, + pub user_deleted_at: Option>, } + +#[derive(Clone, Debug)] +pub enum TtsModelForInferenceError { + ModelNotFound, + ModelDeleted, + DatabaseError { reason: String }, +} + +impl std::fmt::Display for TtsModelForInferenceError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TtsModelForInferenceError::ModelNotFound => write!(f, "ModelNotFound"), + TtsModelForInferenceError::ModelDeleted => write!(f, "ModelDeleted"), + TtsModelForInferenceError::DatabaseError { reason} => + write!(f, "Database error: {:?}", reason), + } + } +} + +impl std::error::Error for TtsModelForInferenceError {} + pub async fn get_tts_model_by_token( pool: &MySqlPool, model_token: &str -) -> AnyhowResult> +) -> Result { // NB: Lookup failure is Err(RowNotFound). // NB: Since this is publicly exposed, we don't query sensitive data. let maybe_model = sqlx::query_as!( - TtsModelRecord2, + TtsModelForInferenceRecord, r#" SELECT tts.token as model_token, @@ -410,31 +458,39 @@ SELECT tts.title, tts.private_bucket_hash, tts.created_at, - tts.updated_at + tts.updated_at, + tts.user_deleted_at, + tts.mod_deleted_at FROM tts_models as tts JOIN users ON users.token = tts.creator_user_token WHERE tts.token = ? -AND tts.user_deleted_at IS NULL -AND tts.mod_deleted_at IS NULL "#, &model_token ) .fetch_one(pool) .await; // TODO: This will return error if it doesn't exist - match maybe_model { - Ok(model) => Ok(Some(model)), + let model : TtsModelForInferenceRecord = match maybe_model { + Ok(model) => model, Err(err) => { match err { RowNotFound => { - Ok(None) + return Err(TtsModelForInferenceError::ModelNotFound); }, _ => { warn!("tts model query error: {:?}", err); - Err(anyhow!("Mysql error: {:?}", err)) + return Err(TtsModelForInferenceError::DatabaseError { + reason: format!("Mysql error: {:?}", err) + }); } } } + }; + + if model.mod_deleted_at.is_some() || model.user_deleted_at.is_some() { + return Err(TtsModelForInferenceError::ModelDeleted); } + + Ok(model) } diff --git a/src/tts_inference_job.rs b/src/tts_inference_job.rs index 03e182a1c5..5f67dfd22b 100644 --- a/src/tts_inference_job.rs +++ b/src/tts_inference_job.rs @@ -24,13 +24,16 @@ use crate::common_env::CommonEnv; use crate::database::enums::vocoder_type::VocoderType; use crate::database::mediators::firehose_publisher::FirehosePublisher; use crate::http_clients::tts_inference_sidecar_client::TtsInferenceSidecarClient; +use crate::job_queries::tts_inference_job_queries::TtsInferenceJobRecord; +use crate::job_queries::tts_inference_job_queries::TtsModelForInferenceError; +use crate::job_queries::tts_inference_job_queries::TtsModelForInferenceRecord; use crate::job_queries::tts_inference_job_queries::get_tts_model_by_token; use crate::job_queries::tts_inference_job_queries::grab_job_lock_and_mark_pending; use crate::job_queries::tts_inference_job_queries::insert_tts_result; use crate::job_queries::tts_inference_job_queries::mark_tts_inference_job_done; use crate::job_queries::tts_inference_job_queries::mark_tts_inference_job_failure; +use crate::job_queries::tts_inference_job_queries::mark_tts_inference_job_permanently_dead; use crate::job_queries::tts_inference_job_queries::query_tts_inference_job_records; -use crate::job_queries::tts_inference_job_queries::{TtsInferenceJobRecord, TtsModelRecord2}; use crate::script_execution::tacotron_inference_command::TacotronInferenceCommand; use crate::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING; use crate::shared_constants::DEFAULT_RUST_LOG; @@ -399,15 +402,40 @@ async fn process_jobs( let model_state = match model_state_result { Ok(model_state) => model_state, Err(e) => { - warn!("Failure to check model state: {:?}", e); + error!("TTS model fetch and state check error: {}, reason: {:?}", &job.model_token, &e); + + let (failure_reason, permanent_failure) = match e { + ModelStateError::ModelNotFound => ("model was not found", true), + ModelStateError::ModelDeleted => ("model has been deleted", true), + ModelStateError::CacheError { .. } => ("internal cache error", false), + ModelStateError::DatabaseError { .. } => ("unknown database error", false), + }; + + + let mut redis = inferencer.redis_pool.get()?; + let mut redis_logger = RedisJobStatusLogger::new_tts_inference( + &mut redis, + &job.inference_job_token); + + redis_logger.log_status(failure_reason)?; + + if permanent_failure { + warn!("Marking job permanently dead: {} because: {:?}", job.inference_job_token, &e); + + let _r = mark_tts_inference_job_permanently_dead( + &inferencer.mysql_pool, + &job, + failure_reason + ).await; + } else { + let _r = mark_tts_inference_job_failure( + &inferencer.mysql_pool, + &job, + failure_reason, + inferencer.job_max_attempts + ).await; + } - let failure_reason = ""; - let _r = mark_tts_inference_job_failure( - &inferencer.mysql_pool, - &job, - failure_reason, - inferencer.job_max_attempts - ).await; continue; } }; @@ -486,11 +514,42 @@ fn read_metadata_file(filename: &PathBuf) -> AnyhowResult { /// We check both of these in one go so that we can reuse the ModelRecord later /// without another DB query. struct ModelState { - pub model_record: TtsModelRecord2, + pub model_record: TtsModelForInferenceRecord, pub is_downloaded_to_filesystem: bool, pub is_in_memory_cache: bool, } +#[derive(Debug, Clone)] +enum ModelStateError { + ModelNotFound, + ModelDeleted, + CacheError { reason: String }, + DatabaseError { reason: String }, +} + +impl std::fmt::Display for ModelStateError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ModelStateError::ModelNotFound => write!(f, "ModelNotFound"), + ModelStateError::ModelDeleted => write!(f, "ModelDeleted"), + ModelStateError::CacheError { reason} => write!(f, "Cache error: {:?}", reason), + ModelStateError::DatabaseError { reason} => write!(f, "Database error: {:?}", reason), + } + } +} + +impl From for ModelStateError { + fn from(error: TtsModelForInferenceError) -> Self { + match error { + TtsModelForInferenceError::ModelNotFound => ModelStateError::ModelNotFound, + TtsModelForInferenceError::ModelDeleted => ModelStateError::ModelDeleted, + TtsModelForInferenceError::DatabaseError { reason } => ModelStateError::DatabaseError { reason } + } + } +} + +impl std::error::Error for ModelStateError {} + impl ModelState { /// Query the model details and see if the model file is on the filesystem in one go. pub async fn query_model_and_check_filesystem( @@ -498,35 +557,37 @@ impl ModelState { mysql_pool: &MySqlPool, semi_persistent_cache: &SemiPersistentCacheDir, virtual_cache: &SyncVirtualLfuCache, - ) -> AnyhowResult { + ) -> Result { info!("Looking up TTS model by token: {}", &job.model_token); - let query_result = get_tts_model_by_token( + let tts_model = get_tts_model_by_token( &mysql_pool, - &job.model_token).await?; + &job.model_token + ).await?; - let model_record = match query_result { - Some(model) => model, - None => { - error!("TTS model not found: {}", &job.model_token); - return Err(anyhow!("Model not found!")) + /*let model_record = match query_result { + Ok(model) => model, + Err(e) => { + error!("TTS model lookup error: {}, reason: {:?}", &job.model_token, &e); + return Err(e) }, - }; + };*/ let tts_synthesizer_fs_path = semi_persistent_cache.tts_synthesizer_model_path( - &model_record.model_token); + &tts_model.model_token); let is_downloaded_to_filesystem = tts_synthesizer_fs_path.exists(); let path = tts_synthesizer_fs_path .to_str() - .ok_or(anyhow!("could not make path"))? + .ok_or(ModelStateError::CacheError { reason: "could not make path".to_string() })? .to_string(); - let is_in_memory_cache = virtual_cache.in_cache(&path)?; + let is_in_memory_cache = virtual_cache.in_cache(&path) + .map_err(|e| ModelStateError::CacheError { reason: format!("Model cache error: {:?}", e) })?; Ok(Self { - model_record, + model_record: tts_model, is_downloaded_to_filesystem, is_in_memory_cache, }) @@ -536,7 +597,7 @@ impl ModelState { async fn process_job( inferencer: &Inferencer, job: &TtsInferenceJobRecord, - model_record: &TtsModelRecord2, + model_record: &TtsModelForInferenceRecord, ) -> AnyhowResult<(Span, Span)> { // TODO 1. Mark processing (DONE)