Cleanup Inference Job startup

- Don't require all env vars for all job types

Cargo.lock

@@ -3453,7 +3453,7 @@ checksum = "adcf93614601c8129ddf72e2d5633df827ba6551541c6d8c59520a371475be1f"
 dependencies = [
  "hermit-abi 0.3.1",
  "io-lifetimes",
- "rustix 0.37.13",
+ "rustix 0.37.27",
  "windows-sys 0.48.0",
 ]
 

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

@@ -209,6 +209,15 @@ async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableInfere
         None, // Obviously don't tell the user about errors even we're not sure about
         Some(FrontendFailureCategory::RetryableWorkerError),
       ),
+
+    ProcessSingleJobError::JobSystemMisconfiguration(ref maybe_reason) =>
+      (
+        JobFailureClass::TransientFailure,
+        ContainerHealth::IncrementContainerFailCount,
+        format!("job system misconfiguration error: {:?}", maybe_reason),
+        None, // Obviously don't tell the user about errors even we're not sure about
+        Some(FrontendFailureCategory::RetryableWorkerError),
+      ),
   };
 
   if container_health_report == ContainerHealth::IncrementContainerFailCount {
@@ -251,6 +260,7 @@ async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableInfere
     ProcessSingleJobError::KeepAliveElapsed => {}
     ProcessSingleJobError::NotYetImplemented => {}
     ProcessSingleJobError::FaceDetectionFailure => {}
+    ProcessSingleJobError::JobSystemMisconfiguration(_) => {}
   }
 
   Ok(())

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

@@ -20,7 +20,11 @@ pub enum ProcessSingleJobError {
   /// This is any other kind of error.
   /// It might be important, we just haven't special cased it yet.
   Other(anyhow::Error),
- 
+
+  /// The job system is misconfigured.
+  /// Retry the job, but mark as a job runner system failure (health check failure).
+  JobSystemMisconfiguration(Option<String>),
+
   /// We hit a feature or a path for this job that has not yet been implemented.
   /// Permanently fail the job.
   NotYetImplemented

crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/mod.rs

@@ -1,4 +1,5 @@
 pub mod model_downloaders;
 pub mod pretrained_hubert_model;
 pub mod process_job;
+pub mod rvc_v2_dependencies;
 pub mod rvc_v2_inference_command;

crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_job.rs

@@ -49,11 +49,18 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
       .new_generic_inference(job.inference_job_token.as_str())
       .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?;
 
+  let model_dependencies = args
+      .job_dependencies
+      .job_specific_dependencies
+      .maybe_rvc_v2_dependencies
+      .as_ref()
+      .ok_or_else(|| ProcessSingleJobError::JobSystemMisconfiguration(Some("missing RVC dependencies".to_string())))?;
+
   // ==================== DOWNLOAD HUBERT ==================== //
 
   info!("Download RVC hubert model (if not present)...");
 
-  args.job_dependencies.pretrained_models.rvc_v2_hubert.download_if_not_on_filesystem(
+  model_dependencies.pretrained_hubert_model.download_if_not_on_filesystem(
     &args.job_dependencies.private_bucket_client,
     &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads)
       .await
@@ -96,7 +103,7 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
 
   let mut i : usize = 0;
 
-  for downloader in args.job_dependencies.job_type_details.rvc_v2.downloaders.all_downloaders() {
+  for downloader in model_dependencies.downloaders.all_downloaders() {
 
     // Temporary debugging
     info!("Downloader {}", i);
@@ -239,15 +246,13 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
 
   let inference_start_time = Instant::now();
 
-  let command_exit_status = args.job_dependencies
-      .job_type_details
-      .rvc_v2
+  let command_exit_status = model_dependencies
       .inference_command
       .execute_inference(InferenceArgs {
         model_path: &rvc_v2_model_fs_path,
         maybe_model_index_path: maybe_rvc_v2_model_index_fs_path,
-        hubert_path: &args.job_dependencies.pretrained_models.rvc_v2_hubert.filesystem_path,
-        rmvpe_path: &args.job_dependencies.job_type_details.rvc_v2.downloaders.rmvpe.filesystem_path,
+        hubert_path: &model_dependencies.pretrained_hubert_model.filesystem_path,
+        rmvpe_path: &model_dependencies.downloaders.rmvpe.filesystem_path,
         input_path: &input_wav_path,
         output_path: &output_audio_fs_path,
         maybe_override_f0_method,

crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/rvc_v2_dependencies.rs

@@ -0,0 +1,20 @@
+use errors::AnyhowResult;
+use crate::job::job_types::vc::rvc_v2::model_downloaders::RvcV2Downloaders;
+use crate::job::job_types::vc::rvc_v2::pretrained_hubert_model::PretrainedHubertModel;
+use crate::job::job_types::vc::rvc_v2::rvc_v2_inference_command::RvcV2InferenceCommand;
+
+pub struct RvcV2Dependencies {
+  pub inference_command: RvcV2InferenceCommand,
+  pub pretrained_hubert_model: PretrainedHubertModel,
+  pub downloaders: RvcV2Downloaders,
+}
+
+impl RvcV2Dependencies {
+  pub fn setup() -> AnyhowResult<Self> {
+    Ok(Self {
+      inference_command: RvcV2InferenceCommand::from_env()?,
+      pretrained_hubert_model: PretrainedHubertModel::from_env(),
+      downloaders: RvcV2Downloaders::build_all_from_env(),
+    })
+  }
+}

crates/service/job/inference_job/src/job_dependencies.rs

@@ -1,3 +1,8 @@
+#![forbid(unreachable_patterns)]
+#![forbid(unused_imports)]
+#![forbid(unused_mut)]
+#![forbid(unused_variables)]
+
 use std::path::PathBuf;
 
 use r2d2_redis::r2d2;
@@ -23,15 +28,11 @@ use crate::job::job_types::lipsync::sad_talker::model_downloaders::SadTalkerDown
 use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::SadTalkerInferenceCommand;
 use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::tacotron2_inference_command::Tacotron2InferenceCommand;
 use crate::job::job_types::tts::vall_e_x::model_downloaders::VallEXDownloaders;
-use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
-
-use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
 use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXCreateEmbeddingCommand;
-use crate::job::job_types::vc::rvc_v2::model_downloaders::RvcV2Downloaders;
-
-use crate::job::job_types::vc::rvc_v2::pretrained_hubert_model::PretrainedHubertModel;
-use crate::job::job_types::vc::rvc_v2::rvc_v2_inference_command::RvcV2InferenceCommand;
+use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
+use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
 use crate::job::job_types::vc::so_vits_svc::so_vits_svc_inference_command::SoVitsSvcInferenceCommand;
+use crate::job_specific_dependencies::JobSpecificDependencies;
 use crate::util::common_commands::ffmpeg_logo_watermark_command::FfmpegLogoWatermarkCommand;
 use crate::util::scoped_execution::ScopedExecution;
 use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
@@ -41,6 +42,11 @@ pub struct JobDependencies {
   /// This is provided at job start.
   pub scoped_execution: ScopedExecution,
 
+  /// Specific dependencies for the various job types.
+  /// They're only loaded if that type of job is configured to run.
+  /// (See "scoped execution")
+  pub job_specific_dependencies: JobSpecificDependencies,
+
   /// Filesystem info and utils
   pub fs: FileSystemDetails,
 
@@ -105,8 +111,6 @@ pub struct JobDependencies {
   // Details for each job type (grouped by the job type)
   pub job_type_details: JobTypeDetails,
 
-  pub pretrained_models: PretrainedModels,
-
   pub container: ContainerEnvironment,
   pub container_db: ContainerEnvironmentArg, // Same info, but for database.
 
@@ -156,15 +160,10 @@ pub struct JobCaches {
   pub model_cache_counter: TtlKeyCounter,
 }
 
-pub struct PretrainedModels {
-  pub rvc_v2_hubert: PretrainedHubertModel,
-}
-
 /// Per-job type details
 pub struct JobTypeDetails {
   pub tacotron2_old_vocodes: Tacotron2VocodesDetails,
   pub vits: VitsDetails,
-  pub rvc_v2: RvcV2Details,
   pub so_vits_svc: SoVitsSvcDetails,
   pub sad_talker: SadTalkerDetails,
   pub vall_e_x:  VallEXDetails
@@ -190,11 +189,6 @@ pub struct VitsDetails {
   pub inference_command: VitsInferenceCommand,
 }
 
-pub struct RvcV2Details {
-  pub inference_command: RvcV2InferenceCommand,
-  pub downloaders: RvcV2Downloaders,
-}
-
 pub struct SoVitsSvcDetails {
   pub inference_command: SoVitsSvcInferenceCommand,
 }

crates/service/job/inference_job/src/job_specific_dependencies.rs

@@ -0,0 +1,25 @@
+use log::info;
+use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
+use errors::AnyhowResult;
+use crate::job::job_types::vc::rvc_v2::rvc_v2_dependencies::RvcV2Dependencies;
+use crate::util::scoped_execution::ScopedExecution;
+
+pub struct JobSpecificDependencies {
+  pub maybe_rvc_v2_dependencies: Option<RvcV2Dependencies>,
+}
+
+impl JobSpecificDependencies {
+
+  pub fn setup_for_jobs(scoped_execution: &ScopedExecution) -> AnyhowResult<Self> {
+    let mut maybe_rvc_v2_dependencies = None;
+
+    if scoped_execution.can_run_job(InferenceModelType::RvcV2) {
+      info!("Setting RVCv2 dependencies...");
+      maybe_rvc_v2_dependencies = Some(RvcV2Dependencies::setup()?);
+    }
+
+    Ok(JobSpecificDependencies {
+      maybe_rvc_v2_dependencies,
+    })
+  }
+}

crates/service/job/inference_job/src/main.rs

@@ -53,19 +53,13 @@ use crate::job::job_loop::main_loop::main_loop;
 use crate::job::job_types::lipsync::sad_talker::model_downloaders::SadTalkerDownloaders;
 use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::SadTalkerInferenceCommand;
 use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::tacotron2_inference_command::Tacotron2InferenceCommand;
-
-use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
-
-use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
-use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXCreateEmbeddingCommand;
 use crate::job::job_types::tts::vall_e_x::model_downloaders::VallEXDownloaders;
-use crate::job::job_types::vc::rvc_v2::model_downloaders::RvcV2Downloaders;
-
-use crate::job::job_types::vc::rvc_v2::pretrained_hubert_model::PretrainedHubertModel;
-use crate::job::job_types::vc::rvc_v2::rvc_v2_inference_command::RvcV2InferenceCommand;
+use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXCreateEmbeddingCommand;
+use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
+use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
 use crate::job::job_types::vc::so_vits_svc::so_vits_svc_inference_command::SoVitsSvcInferenceCommand;
-
-use crate::job_dependencies::{FileSystemDetails, JobCaches, JobDependencies, JobTypeDetails, JobWorkerDetails, PretrainedModels, RvcV2Details, SadTalkerDetails, SoVitsSvcDetails, Tacotron2VocodesDetails, VitsDetails, VallEXDetails};
+use crate::job_dependencies::{FileSystemDetails, JobCaches, JobDependencies, JobTypeDetails, JobWorkerDetails, SadTalkerDetails, SoVitsSvcDetails, Tacotron2VocodesDetails, VallEXDetails, VitsDetails};
+use crate::job_specific_dependencies::JobSpecificDependencies;
 use crate::util::common_commands::ffmpeg_logo_watermark_command::FfmpegLogoWatermarkCommand;
 use crate::util::scoped_execution::ScopedExecution;
 use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
@@ -73,6 +67,7 @@ use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
 pub mod http_server;
 pub mod job;
 pub mod job_dependencies;
+pub mod job_specific_dependencies;
 pub mod util;
 
 // Buckets (shared config)
@@ -255,8 +250,13 @@ async fn main() -> AnyhowResult<()> {
     job_stats: job_stats.clone(),
   };
 
+  let scoped_execution = ScopedExecution::new_from_env()?;
+
+  let job_specific_dependencies = JobSpecificDependencies::setup_for_jobs(&scoped_execution)?;
+
   let job_dependencies = JobDependencies {
-    scoped_execution: ScopedExecution::new_from_env()?,
+    scoped_execution,
+    job_specific_dependencies,
     fs: FileSystemDetails {
       temp_directory_downloads: temp_directory_downloads.clone(),
       temp_directory_work: temp_directory_work.clone(),
@@ -326,19 +326,12 @@ async fn main() -> AnyhowResult<()> {
       so_vits_svc: SoVitsSvcDetails {
         inference_command: SoVitsSvcInferenceCommand::from_env()?,
       },
-      rvc_v2: RvcV2Details {
-        inference_command: RvcV2InferenceCommand::from_env()?,
-        downloaders: RvcV2Downloaders::build_all_from_env(),
-      },
       sad_talker: SadTalkerDetails {
         downloaders: SadTalkerDownloaders::build_all_from_env(),
         inference_command: SadTalkerInferenceCommand::from_env()?,
         ffmpeg_watermark_command: FfmpegLogoWatermarkCommand::from_env()?,
       },
     },
-    pretrained_models: PretrainedModels {
-      rvc_v2_hubert: PretrainedHubertModel::from_env(),
-    },
     container: container_environment.clone(),
     container_db: ContainerEnvironmentArg {
       hostname: container_environment.hostname,