mark jobs permanently dead if models are gone

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<Utc>,
   pub updated_at: DateTime<Utc>,
+  pub mod_deleted_at: Option<DateTime<Utc>>,
+  pub user_deleted_at: Option<DateTime<Utc>>,
 }
 
+
+#[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<Option<TtsModelRecord2>>
+) -> Result<TtsModelForInferenceRecord, TtsModelForInferenceError>
 {
   // 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)
 }

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<FileMetadata> {
 /// 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<TtsModelForInferenceError> 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<Self> {
+  ) -> Result<Self, ModelStateError> {
     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)