record frontend_failure_category, emit SadTalker errors

_sql/migrations_squashed/generic_inference_jobs.sql

@@ -173,7 +173,14 @@ CREATE TABLE generic_inference_jobs (
   -- An enum-like key to present the frontend with a failure class in an
   -- i18n-friendly way. The javascript/frontend can use these as indices
   -- into descriptive, broadly-localized error messages.
-  --   * 'face_not_detected' for images or videos that do not have a detectable face (SadTalker, Wav2Lip, etc.)
+  --
+  --   * 'face_not_detected' for images or videos that do not have a detectable
+  --                         face (SadTalker, Wav2Lip, etc.)
+  --
+  --   * 'retryable_worker_error' for generic worker errors (full filesystem,
+  --                              etc.) that can be retried. (We don't need to
+  --                              tell the user why the job failed.)
+  --
   frontend_failure_category VARCHAR(32) DEFAULT NULL,
 
   -- On success, we populate how long the job execution took in milliseconds.

crates/_types/enums/src/by_table/generic_inference_jobs/frontend_failure_category.rs

@@ -26,6 +26,9 @@ pub enum FrontendFailureCategory {
   /// For SadTalker (and possibly Wav2Lip)
   #[serde(rename = "face_not_detected")]
   FaceNotDetected,
+
+  #[serde(rename = "retryable_worker_error")]
+  RetryableWorkerError,
 }
 
 // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
@@ -37,12 +40,14 @@ impl FrontendFailureCategory {
   pub fn to_str(&self) -> &'static str {
     match self {
       Self::FaceNotDetected => "face_not_detected",
+      Self::RetryableWorkerError => "retryable_worker_error",
     }
   }
 
   pub fn from_str(value: &str) -> Result<Self, String> {
     match value {
       "face_not_detected" => Ok(Self::FaceNotDetected),
+      "retryable_worker_error" => Ok(Self::RetryableWorkerError),
       _ => Err(format!("invalid value: {:?}", value)),
     }
   }
@@ -52,6 +57,7 @@ impl FrontendFailureCategory {
     // NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc.
     BTreeSet::from([
       Self::FaceNotDetected,
+      Self::RetryableWorkerError,
     ])
   }
 }
@@ -67,23 +73,27 @@ mod tests {
     #[test]
     fn test_serialization() {
       assert_serialization(FrontendFailureCategory::FaceNotDetected, "face_not_detected");
+      assert_serialization(FrontendFailureCategory::RetryableWorkerError, "retryable_worker_error");
     }
 
     #[test]
     fn to_str() {
       assert_eq!(FrontendFailureCategory::FaceNotDetected.to_str(), "face_not_detected");
+      assert_eq!(FrontendFailureCategory::RetryableWorkerError.to_str(), "retryable_worker_error");
     }
 
     #[test]
     fn from_str() {
       assert_eq!(FrontendFailureCategory::from_str("face_not_detected").unwrap(), FrontendFailureCategory::FaceNotDetected);
+      assert_eq!(FrontendFailureCategory::from_str("retryable_worker_error").unwrap(), FrontendFailureCategory::RetryableWorkerError);
     }
 
     #[test]
     fn all_variants() {
       let mut variants = FrontendFailureCategory::all_variants();
-      assert_eq!(variants.len(), 1);
+      assert_eq!(variants.len(), 2);
       assert_eq!(variants.pop_first(), Some(FrontendFailureCategory::FaceNotDetected));
+      assert_eq!(variants.pop_first(), Some(FrontendFailureCategory::RetryableWorkerError));
       assert_eq!(variants.pop_first(), None);
     }
   }

crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_completely_failed.rs

@@ -1,6 +1,7 @@
 use anyhow::anyhow;
 use sqlx::MySqlPool;
 
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
 use errors::AnyhowResult;
 
 use crate::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
@@ -10,6 +11,7 @@ pub async fn mark_generic_inference_job_completely_failed(
   job: &AvailableInferenceJob,
   maybe_public_failure_reason: Option<&str>,
   maybe_internal_debugging_failure_reason: Option<&str>,
+  maybe_frontend_failure_category: Option<FrontendFailureCategory>,
 ) -> AnyhowResult<()>
 {
   let mut maybe_public_failure_reason
@@ -33,11 +35,13 @@ SET
   status = "complete_failure",
   failure_reason = ?,
   internal_debugging_failure_reason = ?,
+  frontend_failure_category = ?,
   retry_at = NULL
 WHERE id = ?
         "#,
         maybe_public_failure_reason.as_deref(),
         maybe_internal_debugging_failure_reason.as_deref(),
+        maybe_frontend_failure_category,
         job.id.0,
     )
       .execute(pool)

crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_failure.rs

@@ -2,6 +2,7 @@ use anyhow::anyhow;
 use sqlx;
 use sqlx::MySqlPool;
 
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
 use errors::AnyhowResult;
 
 use crate::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
@@ -12,6 +13,7 @@ pub async fn mark_generic_inference_job_failure(
   job: &AvailableInferenceJob,
   maybe_public_failure_reason: Option<&str>,
   internal_debugging_failure_reason: &str,
+  maybe_frontend_failure_category: Option<FrontendFailureCategory>,
   max_attempts: u16
 ) -> AnyhowResult<()> {
 
@@ -40,12 +42,14 @@ SET
   status = ?,
   failure_reason = ?,
   internal_debugging_failure_reason = ?,
+  frontend_failure_category = ?,
   retry_at = NOW() + interval 2 minute
 WHERE id = ?
         "#,
         next_status,
         maybe_public_failure_reason.as_deref(),
         &internal_debugging_failure_reason,
+        maybe_frontend_failure_category,
         job.id.0,
     )
       .execute(pool)

crates/service/job/inference_job/config/inference-job.development.env

@@ -21,6 +21,7 @@ VALL_E_X_INFERENCE_ROOT_DIRECTORY = "/home/tensor/code/Storyteller/storyteller-m
 VALL_E_X_INFERENCE_COMMAND = "python3 main.py"
 VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND = "source .env/bin/activate"
 VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE = "" # fill this out for dev?
+
 # ================================= #
 #                                   #
 #             RVC (v2)              #

crates/service/job/inference_job/src/job/job_loop/main_loop.rs

@@ -1,6 +1,7 @@
 use std::time::Duration;
 
 use log::{error, info, warn};
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
 
 use errors::AnyhowResult;
 use filesys::file_exists::file_exists;
@@ -22,6 +23,22 @@ const INCREASE_TIMEOUT_MILLIS : u64 = 1000;
 /// Pause file millis
 const PAUSE_FILE_EXISTS_WAIT_MILLIS : u64 = 1000 * 30;
 
+#[derive(Eq,PartialEq)]
+enum JobFailureClass {
+  // Jobs that can be retried
+  TransientFailure,
+  // Jobs that cannot be retried and must be marked dead
+  PermanentFailure,
+}
+
+#[derive(Eq,PartialEq)]
+enum ContainerHealth {
+  // No impact to container health
+  Ignore,
+  // Increment the container health failure counter
+  IncrementContainerFailCount,
+}
+
 pub async fn main_loop(job_dependencies: JobDependencies) {
   let mut noop_logger = NoOpLogger::new(job_dependencies.no_op_logger_millis as i64);
 
@@ -126,46 +143,92 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<Availab
       Err(e) => {
         warn!("Failure to process job: {:?}", e);
 
-        let (permanent_failure, increment_fail_count, internal_failure_reason, maybe_public_failure_reason) =
-            match e {
-              // Permanent failures
-              ProcessSingleJobError::KeepAliveElapsed =>
-                (true, false, "keepalive elapsed".to_string(), Some("keepalive elapsed")),
-              ProcessSingleJobError::InvalidJob(ref err) =>
-                (true, false, format!("InvalidJob: {:?}", err), Some("invalid job")),
-              ProcessSingleJobError::NotYetImplemented =>
-                (true, false, "not yet implemented".to_string(), Some("not yet implemented")),
-              ProcessSingleJobError::FaceDetectionFailure =>
-                (true, false, "face not detected".to_string(), Some("face not detected")),
+        let (
+          job_failure_class,
+          container_health_report,
+          internal_failure_reason,
+          maybe_public_failure_reason, // TODO(bt,2023-10-11): Remove this column in favor of "frontend_failure_category".
+          maybe_frontend_failure_category
+        ) = match e {
+          // Permanent failures
+          ProcessSingleJobError::KeepAliveElapsed =>
+            (
+              JobFailureClass::PermanentFailure,
+              ContainerHealth::Ignore,
+              "keepalive elapsed".to_string(),
+              Some("keepalive elapsed"),
+              None,
+            ),
+          ProcessSingleJobError::InvalidJob(ref err) =>
+            (
+              JobFailureClass::PermanentFailure,
+              ContainerHealth::Ignore,
+              format!("InvalidJob: {:?}", err),
+              Some("invalid job"),
+              None,
+            ),
+          ProcessSingleJobError::NotYetImplemented =>
+            (
+              JobFailureClass::PermanentFailure,
+              ContainerHealth::Ignore,
+              "not yet implemented".to_string(),
+              Some("not yet implemented"),
+              None,
+            ),
+          ProcessSingleJobError::FaceDetectionFailure =>
+            (
+              JobFailureClass::PermanentFailure,
+              ContainerHealth::Ignore,
+              "face not detected".to_string(),
+              Some("face not detected"),
+              Some(FrontendFailureCategory::FaceNotDetected),
+            ),
 
-              // Non-permanent failures
-              ProcessSingleJobError::FilesystemFull =>
-                (false, true, "worker filesystem full".to_string(), Some("worker filesystem full")),
-              ProcessSingleJobError::Other(ref err) =>
-                (false, true, format!("OtherErr: {:?}", err), None),
-            };
+          // Non-permanent failures
+          ProcessSingleJobError::FilesystemFull =>
+            (
+              JobFailureClass::TransientFailure,
+              ContainerHealth::IncrementContainerFailCount,
+              "worker filesystem full".to_string(),
+              Some("worker filesystem full"),
+              Some(FrontendFailureCategory::RetryableWorkerError),
+            ),
+          ProcessSingleJobError::Other(ref err) =>
+            (
+              JobFailureClass::TransientFailure,
+              ContainerHealth::IncrementContainerFailCount,
+              format!("OtherErr: {:?}", err),
+              None,
+              Some(FrontendFailureCategory::RetryableWorkerError),
+            ),
+        };
 
-        if increment_fail_count {
+        if container_health_report == ContainerHealth::IncrementContainerFailCount {
           // NB: We only increment the fail count for events that may indicate the job server is stuck.
           let stats = job_dependencies.job_stats.increment_failure_count().ok();
           warn!("Failure stats: {:?}", stats);
         }
 
-        if permanent_failure {
-          let _r = mark_generic_inference_job_completely_failed(
-            &job_dependencies.mysql_pool,
-            &job,
-            maybe_public_failure_reason,
-            Some(&internal_failure_reason),
-          ).await;
-        } else {
-          let _r = mark_generic_inference_job_failure(
-            &job_dependencies.mysql_pool,
-            &job,
-            maybe_public_failure_reason,
-            &internal_failure_reason,
-            job_dependencies.job_max_attempts
-          ).await;
+        match job_failure_class {
+          JobFailureClass::PermanentFailure => {
+            let _r = mark_generic_inference_job_completely_failed(
+              &job_dependencies.mysql_pool,
+              &job,
+              maybe_public_failure_reason,
+              Some(&internal_failure_reason),
+              maybe_frontend_failure_category,
+            ).await;
+          }
+          JobFailureClass::TransientFailure => {
+            let _r = mark_generic_inference_job_failure(
+              &job_dependencies.mysql_pool,
+              &job,
+              maybe_public_failure_reason,
+              &internal_failure_reason,
+              maybe_frontend_failure_category,
+              job_dependencies.job_max_attempts
+            ).await;
+          }
         }
 
         match e {

crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs

@@ -1,5 +1,6 @@
 use std::fs::read_to_string;
-use std::time::Instant;
+use std::thread;
+use std::time::{Duration, Instant};
 
 use anyhow::anyhow;
 use log::{error, info, warn};
@@ -209,6 +210,8 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
       if contents.contains("can not detect the landmark from source image") {
         warn!("Face not detected in source image");
         error = ProcessSingleJobError::FaceDetectionFailure;
+      } else {
+        warn!("Captured stderr output: {}", contents);
       }
     }