new RVC dispatching

_sql/actions_and_analytics/generic_inference/find_non_model_weights_tokens_in_jobs.sql

@@ -0,0 +1,16 @@
+-- Find vcm_* tokens that are still in use
+select
+  job_token,
+  model_token,
+  user_token
+from (
+  select token as job_token,
+  maybe_model_token as model_token,
+  maybe_creator_user_token as user_token
+  from generic_inference_jobs
+  order by id
+  desc limit 1000000
+) as x
+where model_token NOT LIKE "TM:%"
+and model_token NOT LIKE "weight_%"
+limit 50;

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

@@ -24,6 +24,7 @@ use crate::job::job_types::mocap::process_single_mc_job::process_single_mc_job;
 use crate::job::job_types::render_engine_scene::process_render_engine_scene::process_single_render_engine_scene_job;
 use crate::job::job_types::tts::process_single_tts_job::process_single_tts_job;
 use crate::job::job_types::vc::process_single_vc_job::process_single_vc_job;
+use crate::job::job_types::vc::rvc_v2::dispatch_rvc_v2_job::dispatch_rvc_v2_job;
 use crate::job::job_types::videofilter::process_single_vf_job::process_single_vf_job;
 use crate::job::job_types::workflow::process_single_workflow_job::process_single_workflow_job;
 use crate::state::job_dependencies::JobDependencies;
@@ -275,6 +276,9 @@ async fn new_dispatch(
     InferenceJobType::GptSovits => {
       process_single_gpt_sovits_job(job_dependencies, job).await?
     },
+    InferenceJobType::RvcV2 => {
+      dispatch_rvc_v2_job(job_dependencies, job).await?
+    },
     _ => {
       return Err(ProcessSingleJobError::InvalidJob(
         anyhow!("invalid job type for dispatch: {:?}", job.job_type)))

crates/service/job/inference_job/src/job/job_types/gpt_sovits/process_single_gpt_sovits_job.rs

@@ -27,7 +27,7 @@ pub async fn process_single_gpt_sovits_job(
       }
     }
     _ => {
-      Err(ProcessSingleJobError::Other(anyhow!("tts model type not set")))
+      Err(ProcessSingleJobError::Other(anyhow!("job type not set")))
     }
   }
 }

crates/service/job/inference_job/src/job/job_types/vc/process_single_vc_job.rs

@@ -1,5 +1,5 @@
 use anyhow::anyhow;
-use log::{error, info};
+use log::{error, info, warn};
 
 use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType;
 use migration::voice_conversion::query_vc_model_for_migration::{query_vc_model_for_migration, VcModelType};
@@ -13,11 +13,15 @@ use crate::job::job_loop::job_success_result::JobSuccessResult;
 use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
 use crate::job::job_types::vc::{rvc_v2, so_vits_svc};
 use crate::job::job_types::vc::media_for_inference::MediaForInference;
-use crate::job::job_types::vc::rvc_v2::process_job::RvcV2ProcessJobArgs;
+use crate::job::job_types::vc::rvc_v2::process_rvc_job::RvcV2ProcessJobArgs;
 use crate::job::job_types::vc::so_vits_svc::process_job::SoVitsSvcProcessJobArgs;
 use crate::state::job_dependencies::JobDependencies;
 
-pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError> {
+pub async fn process_single_vc_job(
+  job_dependencies: &JobDependencies,
+  job: &AvailableInferenceJob
+) -> Result<JobSuccessResult, ProcessSingleJobError> {
+
   let model_token = match job.maybe_model_token.as_deref() {
     Some(token) => token,
     None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No model token for job: {:?}", job.inference_job_token))),
@@ -113,7 +117,8 @@ pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &Ava
 
   let job_success_result = match model.get_model_type() {
     VcModelType::RvcV2 => {
-      rvc_v2::process_job::process_job(RvcV2ProcessJobArgs {
+      warn!("OLD/LEGACY RVC code path. We should instead dispatch on the basis of job_type. See `dispatch_rvc_v2_job` and migrate to that code path.");
+      rvc_v2::process_rvc_job::process_rvc_job(RvcV2ProcessJobArgs {
         job_dependencies,
         job,
         vc_model: &model,
@@ -134,5 +139,3 @@ pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &Ava
 
   Ok(job_success_result)
 }
-
-

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

@@ -0,0 +1,38 @@
+use anyhow::anyhow;
+use log::info;
+use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
+use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
+
+use crate::job::job_loop::job_success_result::JobSuccessResult;
+use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
+use crate::job::job_types::vc::rvc_v2::process_rvc_job_old_entrypoint_adapter::process_rvc_job_old_entrypoint_adapter;
+use crate::state::job_dependencies::JobDependencies;
+
+pub async fn dispatch_rvc_v2_job(
+  job_dependencies: &JobDependencies,
+  job: &AvailableInferenceJob
+) -> Result<JobSuccessResult, ProcessSingleJobError> {
+
+  info!("New dispatch route(2024-09-04): dispatch_rvc_v2_job instead of process_single_vc_job");
+
+  return match job.job_type {
+    InferenceJobType::RvcV2 => {
+      match job.maybe_download_url {
+        Some(_) => {
+          // TODO(bt,2024-09-04): Support RVC uploads
+          Err(ProcessSingleJobError::Other(anyhow!("RVC uploads not yet supported!")))
+        }
+        None => {
+          let job_success_result = process_rvc_job_old_entrypoint_adapter(
+            job_dependencies,
+            job
+          ).await?;
+          Ok(job_success_result)
+        }
+      }
+    }
+    _ => {
+      Err(ProcessSingleJobError::Other(anyhow!("job type not set (new rvc code path)")))
+    }
+  }
+}

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

@@ -1,5 +1,7 @@
+pub mod dispatch_rvc_v2_job;
 pub mod model_downloaders;
 pub mod pretrained_hubert_model;
-pub mod process_job;
+pub mod process_rvc_job;
+pub mod process_rvc_job_old_entrypoint_adapter;
 pub mod rvc_v2_dependencies;
 pub mod rvc_v2_inference_command;

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

@@ -42,7 +42,7 @@ pub struct RvcV2ProcessJobArgs<'a> {
   pub inference_media: &'a MediaForInference,
 }
 
-pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
+pub async fn process_rvc_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
   let job = args.job;
   let vc_model = args.vc_model;
 

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

@@ -0,0 +1,118 @@
+use anyhow::anyhow;
+use log::{error, info};
+
+use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType;
+use migration::voice_conversion::query_vc_model_for_migration::{query_vc_model_for_migration, VcModelType};
+use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
+use mysql_queries::queries::media_files::get::get_media_file_for_inference::get_media_file_for_inference;
+use mysql_queries::queries::media_uploads::get_media_upload_for_inference::get_media_upload_for_inference;
+use tokens::tokens::media_files::MediaFileToken;
+use tokens::tokens::media_uploads::MediaUploadToken;
+
+use crate::job::job_loop::job_success_result::JobSuccessResult;
+use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
+use crate::job::job_types::vc::{rvc_v2, so_vits_svc};
+use crate::job::job_types::vc::media_for_inference::MediaForInference;
+use crate::job::job_types::vc::rvc_v2::process_rvc_job::RvcV2ProcessJobArgs;
+use crate::job::job_types::vc::so_vits_svc::process_job::SoVitsSvcProcessJobArgs;
+use crate::state::job_dependencies::JobDependencies;
+
+/// This is meant to take over from the `process_single_vc_job` function for a short bit
+pub async fn process_rvc_job_old_entrypoint_adapter(
+  job_dependencies: &JobDependencies,
+  job: &AvailableInferenceJob
+) -> Result<JobSuccessResult, ProcessSingleJobError> {
+
+  let model_token = match job.maybe_model_token.as_deref() {
+    Some(token) => token,
+    None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No model token for job: {:?}", job.inference_job_token))),
+  };
+
+  // TODO: Interrogate a db result cache + filesystem model file cache
+
+  let maybe_model = query_vc_model_for_migration(model_token, &job_dependencies.db.mysql_pool)
+      .await
+      .map_err(|err| {
+        error!("error querying vc model {model_token} : {:?}", err);
+        ProcessSingleJobError::Other(anyhow!("error querying vc model {model_token} : {:?}", err))
+      })?;
+
+  let model = match maybe_model {
+    None => return Err(ProcessSingleJobError::Other(anyhow!("model weights not found: {:?}", model_token))),
+    Some(model) => model,
+  };
+
+  let media_token = job.maybe_input_source_token
+      .as_deref()
+      .ok_or_else(|| ProcessSingleJobError::Other(anyhow!(
+        "no associated media token for vc job: {:?}", job.inference_job_token)))?;
+
+  let token_type = job.maybe_input_source_token_type
+      .ok_or_else(|| ProcessSingleJobError::Other(anyhow!(
+        "no associated media token type for vc job: {:?}", job.inference_job_token)))?;
+
+  let inference_media = match token_type {
+    InferenceInputSourceTokenType::MediaFile => {
+      // media_files case
+      let media_file_token = MediaFileToken::new_from_str(media_token);
+      let maybe_media_file = get_media_file_for_inference(&media_file_token, &job_dependencies.db.mysql_pool).await;
+
+      let media_file = match maybe_media_file {
+        Ok(Some(media_file)) => media_file,
+        Ok(None) => {
+          error!("no media file record found for token: {:?}", media_token);
+          return Err(ProcessSingleJobError::Other(
+            anyhow!("no media file record found for token: {:?}", media_token)));
+        }
+        Err(err) => {
+          error!("error fetching media file record from db: {:?}", err);
+          return Err(ProcessSingleJobError::Other(err));
+        }
+      };
+
+      MediaForInference::MediaFile(media_file)
+    }
+    InferenceInputSourceTokenType::MediaUpload => {
+      // media_uploads case
+      let media_upload_token = MediaUploadToken::new_from_str(media_token);
+      let maybe_media_upload_result =
+          get_media_upload_for_inference(&media_upload_token, &job_dependencies.db.mysql_pool).await;
+
+      let media_upload = match maybe_media_upload_result {
+        Ok(Some(media_upload)) => media_upload,
+        Ok(None) => {
+          error!("no media upload record found for token: {:?}", media_token);
+          return Err(ProcessSingleJobError::Other(
+            anyhow!("no media upload record found for token: {:?}", media_token)));
+        }
+        Err(err) => {
+          error!("error fetching media upload record from db: {:?}", err);
+          return Err(ProcessSingleJobError::Other(err));
+        }
+      };
+
+      MediaForInference::LegacyMediaUpload(media_upload)
+    }
+  };
+
+  info!("Source media upload file size (bytes): {}", inference_media.file_size_bytes());
+  info!("Source media upload duration (millis): {:?}", inference_media.maybe_duration_millis());
+  info!("Source media upload duration (seconds): {:?}", (inference_media.maybe_duration_millis()
+    .map(|d| d as f32 / 1000.0)));
+
+  let job_success_result = match model.get_model_type() {
+    VcModelType::RvcV2 => {
+      rvc_v2::process_rvc_job::process_rvc_job(RvcV2ProcessJobArgs {
+        job_dependencies,
+        job,
+        vc_model: &model,
+        inference_media: &inference_media,
+      }).await?
+    }
+    _ => return Err(ProcessSingleJobError::InvalidJob(anyhow!("only RVC models should be dispatched this way"))),
+  };
+
+  Ok(job_success_result)
+}
+
+

crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/enqueue_voice_conversion_inference_handler.rs

@@ -281,25 +281,6 @@ pub async fn enqueue_voice_conversion_inference_handler(
         "input token is not a media_upload or media_file token".to_string()));
     };
 
-  // NB: This is better, but we'd need to change the API and migrate callers.
-  // let (input_token, input_token_type) =
-  //     match (&request.source_media_upload_token, &request.source_media_file_token) {
-  //       (Some(media_upload), None) => {
-  //         (media_upload.to_string(), InferenceInputSourceTokenType::MediaUpload)
-  //       },
-  //       (None, Some(media_file)) => {
-  //         (media_file.to_string(), InferenceInputSourceTokenType::MediaFile)
-  //       },
-  //       (Some(_media_upload), Some(_media_file)) => {
-  //         return Err(EnqueueVoiceConversionInferenceError::BadInput(
-  //           "both source_media_upload_token and source_media_file_token provided".to_string()));
-  //       },
-  //       (None, None) => {
-  //         return Err(EnqueueVoiceConversionInferenceError::BadInput(
-  //           "neither source_media_upload_token nor source_media_file_token provided".to_string()));
-  //       },
-  //     };
-
   // ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== //
 
   let mut redis = server_state.redis_pool