From f1fd99be15e38796dd777be041deee836cc5de68 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Thu, 12 Oct 2023 01:03:36 -0400 Subject: [PATCH] record frontend_failure_category, emit SadTalker errors --- .../generic_inference_jobs.sql | 9 +- .../frontend_failure_category.rs | 12 +- ...generic_inference_job_completely_failed.rs | 4 + .../job/mark_generic_inference_job_failure.rs | 4 + .../config/inference-job.development.env | 1 + .../src/job/job_loop/main_loop.rs | 129 +++++++++++++----- .../lipsync/sad_talker/process_job.rs | 5 +- 7 files changed, 128 insertions(+), 36 deletions(-) diff --git a/_sql/migrations_squashed/generic_inference_jobs.sql b/_sql/migrations_squashed/generic_inference_jobs.sql index 675002dacd..f23c77525d 100644 --- a/_sql/migrations_squashed/generic_inference_jobs.sql +++ b/_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. diff --git a/crates/_types/enums/src/by_table/generic_inference_jobs/frontend_failure_category.rs b/crates/_types/enums/src/by_table/generic_inference_jobs/frontend_failure_category.rs index d59313120c..b4cca23a60 100644 --- a/crates/_types/enums/src/by_table/generic_inference_jobs/frontend_failure_category.rs +++ b/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 { 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); } } diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_completely_failed.rs b/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_completely_failed.rs index c4445ddbca..2a2452d080 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_completely_failed.rs +++ b/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, ) -> 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) diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_failure.rs b/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_failure.rs index ae0b012878..a1df90a95a 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/job/mark_generic_inference_job_failure.rs +++ b/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, 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) diff --git a/crates/service/job/inference_job/config/inference-job.development.env b/crates/service/job/inference_job/config/inference-job.development.env index 1b14ed66b9..df798ca0e6 100644 --- a/crates/service/job/inference_job/config/inference-job.development.env +++ b/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) # 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 f99d376c74..a5375921a0 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 @@ -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 { 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 { diff --git a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs index 6e3e2ba80e..f2f67d8874 100644 --- a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs +++ b/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