Fix build errors. This is largely because inference_job is not done

crates/service/job/inference_job/src/caching/cache_miss_strategizer.rs

@@ -122,7 +122,7 @@ impl <T: Hash + Eq> CacheMissStrategizer<T> {
 #[cfg(test)]
 mod tests {
   use chrono::{Duration, Utc, DateTime};
-  use crate::{CacheMissStrategizer, CacheMissStrategy};
+  use crate::caching::cache_miss_strategizer::{CacheMissStrategizer, CacheMissStrategy};
 
   fn get_date(datetime: &str) -> DateTime<Utc> {
     let datetime = DateTime::parse_from_rfc3339(datetime).unwrap();

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

@@ -13,6 +13,8 @@ use jobs_common::job_progress_reporter::job_progress_reporter::JobProgressReport
 use jobs_common::semi_persistent_cache_dir::SemiPersistentCacheDir;
 use memory_caching::multi_item_ttl_cache::MultiItemTtlCache;
 use newrelic_telemetry::Client as NewRelicClient;
+use r2d2_redis::RedisConnectionManager;
+use r2d2_redis::r2d2;
 use sqlx::MySqlPool;
 use std::path::PathBuf;
 
@@ -22,6 +24,13 @@ pub struct JobDependencies {
 
   pub mysql_pool: MySqlPool,
 
+  // TODO(2023-01-11): We don't always connect to a Redis
+  //  Typically this is for job status reporting, but we might also report on when users leave the
+  //  site to proactively kill their inference jobs and save on worker quota.
+  //  On local dev we probably don't care about Redis at all, and on on-prem workers, we cannot
+  //  connect to production Redis easily (requires lots of setup - ghosttunnel or something + IP rules)
+  pub maybe_redis_pool: Option<r2d2::Pool<RedisConnectionManager>>,
+
   pub job_progress_reporter: Box<dyn JobProgressReporterBuilder>,
 
   pub private_bucket_client: BucketClient,
@@ -64,8 +73,7 @@ pub struct JobDependencies {
   pub job_batch_wait_millis: u64,
 
   // Max job attempts before failure.
-  // NB: This is an i32 so we don't need to convert to db column type.
-  pub job_max_attempts: i32,
+  pub job_max_attempts: u16,
 
   // Number of jobs to dequeue at once.
   pub job_batch_size: u32,

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

@@ -1,5 +1,4 @@
 use crate::job_steps::job_dependencies::JobDependencies;
-use crate::job_steps::process_single_job_old::process_single_job_old;
 use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::{AvailableInferenceJob, list_available_generic_inference_jobs, ListAvailableGenericInferenceJobArgs};
 use database_queries::queries::generic_inference::job::mark_generic_inference_job_failure::mark_generic_inference_job_failure;
 use errors::AnyhowResult;
@@ -77,19 +76,20 @@ pub async fn main_loop(job_dependencies: JobDependencies) {
 // TODO: A common interface/trait for each submodule (tts, webvc) to declare how to determine if the job is "ready".
 //  This probably returns a struct or enum with some measure of how many GB need to be downloaded.
 
-async fn process_job_batch(job_state: &JobState, jobs: Vec<AvailableInferenceJob>) -> AnyhowResult<()> {
+async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<AvailableInferenceJob>) -> AnyhowResult<()> {
   for job in jobs.into_iter() {
-    let result = process_single_job(job_state, &job).await;
+    let result = process_single_job(job_dependencies, &job).await;
     match result {
       Ok(_) => {},
       Err(e) => {
         warn!("Failure to process job: {:?}", e);
         let failure_reason = "";
         let _r = mark_generic_inference_job_failure(
-          &job_state.mysql_pool,
+          &job_dependencies.mysql_pool,
           &job,
           failure_reason,
-          job_state.job_max_attempts
+          failure_reason,
+          job_dependencies.job_max_attempts
         ).await;
       }
     }

crates/service/job/inference_job/src/job_steps/mod.rs

@@ -1,9 +1,7 @@
 pub mod job_dependencies;
 pub mod job_stats;
 pub mod main_loop;
-pub mod main_loop_old;
 pub mod process_single_job;
 pub mod process_single_job_error;
-pub mod process_single_job_old;
 pub mod select_job_batch;
 pub mod sidecar_health_check_trap;

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

@@ -1,19 +1,15 @@
-use anyhow::anyhow;
-use config::is_bad_download_url::is_bad_download_url;
-use container_common::filesystem::safe_delete_temp_directory::safe_delete_temp_directory;
 use crate::job_steps::job_dependencies::JobDependencies;
 use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
 use database_queries::queries::generic_inference::job::mark_generic_inference_job_pending_and_grab_lock::mark_generic_inference_job_pending_and_grab_lock;
 use database_queries::queries::generic_inference::job::mark_generic_inference_job_successfully_done::mark_generic_inference_job_successfully_done;
 use enums::workers::generic_inference_type::GenericInferenceType;
 use errors::AnyhowResult;
-use jobs_common::redis_job_status_logger::RedisJobStatusLogger;
 use log::{info, warn};
-use tempdir::TempDir;
 
 pub async fn process_single_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> AnyhowResult<()> {
-  let mut redis = job_dependencies.redis_pool.get()?;
-  let mut redis_logger = RedisJobStatusLogger::new_generic_download(&mut redis, job.download_job_token.as_str());
+  // TODO(bt, 2023-01-11): Restore an optional status logger
+  //let mut redis = job_dependencies.redis_pool.get()?;
+  //let mut redis_logger = RedisJobStatusLogger::new_generic_download(&mut redis, job.download_job_token.as_str());
 
   // ==================== ATTEMPT TO GRAB JOB LOCK ==================== //
 
@@ -24,13 +20,15 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa
     return Ok(())
   }
 
-  info!("Beginning work on {:?} = {} | {}", job.download_type, job.download_job_token, job.download_url);
+  info!("Beginning work on {:?} = {}", job.inference_type, job.inference_job_token);
 
   // ==================== SETUP TEMP DIRS ==================== //
 
   let temp_dir = format!("temp_{}", job.id.0);
   let temp_dir = job_dependencies.scoped_temp_dir_creator.new_tempdir(&temp_dir)?;
 
+  let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove.
+
   // ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== //
 
   let mut entity_token : Option<String> = None;
@@ -39,6 +37,8 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa
   match job.inference_type {
     GenericInferenceType::TextToSpeech => {
       // TODO
+      entity_type = Some("todo".to_string()); // TODO
+      entity_token = Some("todo".to_string()); // TODO
     }
     GenericInferenceType::VoiceConversion => {
       // TODO
@@ -101,16 +101,19 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa
 
   info!("Saved model record: {} - {}", job.id.0, &job.inference_job_token);
 
-  job_dependencies.firehose_publisher.publish_generic_download_finished(
-    &job.creator_user_token,
-    entity_token.as_deref())
-      .await
-      .map_err(|e| {
-        warn!("error publishing event: {:?}", e);
-        anyhow!("error publishing event")
-      })?;
+  // TODO(bt, 2023-01-11): Need to publish that the job finished.
+  //  Publish the *correct type* of event.
+  //job_dependencies.firehose_publisher.publish_generic_download_finished(
+  //  &job.maybe_creator_user_token,
+  //  entity_token.as_deref())
+  //    .await
+  //    .map_err(|e| {
+  //      warn!("error publishing event: {:?}", e);
+  //      anyhow!("error publishing event")
+  //    })?;
 
-  redis_logger.log_status("done")?;
+  // TODO(bt, 2023-01-11): Restore optional Redis logging
+  //redis_logger.log_status("done")?;
 
   info!("Job done: {}", job.id.0);
 

crates/service/job/inference_job/src/job_types/tts/handle_tts_inference.rs

@@ -1,18 +1,16 @@
-use anyhow::anyhow;
-use log::{info, warn};
-use tempdir::TempDir;
 use crate::job_steps::job_dependencies::JobDependencies;
-use crate::job_steps::process_single_job_error::ProcessSingleJobError;
 use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
 use errors::AnyhowResult;
-use tts_common::clean_symbols::clean_symbols;
-use crate::util::download_file_from_bucket::maybe_download_file_from_bucket;
 
 pub async fn handle_tts_inference(
-  job_dependencies: &JobDependencies,
-  job: &AvailableInferenceJob,
+  _job_dependencies: &JobDependencies,
+  _job: &AvailableInferenceJob,
 ) -> AnyhowResult<()> {
 
+  /*
+  TODO(bt, 2023-01-11): This was just copied verbatim from tts-inference-job.
+   It's not ready to run from inference-job yet as it needs adjustments to the new schema.
+
   let mut job_progress_reporter = job_dependencies
       .job_progress_reporter
       .new_tts_inference(job.inference_job_token.as_str())
@@ -339,6 +337,6 @@ pub async fn handle_tts_inference(
   job_iteration_span.set_attribute("status", "success");
   job_iteration_span.set_duration(duration);
 
-  Ok((since_creation_span, job_iteration_span))
+   */
   Ok(())
 }

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

@@ -38,7 +38,6 @@ use crate::http_clients::tts_sidecar_health_check_client::TtsSidecarHealthCheckC
 use crate::job_steps::job_dependencies::JobWorkerDetails;
 use crate::job_steps::job_dependencies::{JobDependencies, JobCaches, JobHttpClients};
 use crate::job_steps::job_stats::JobStats;
-use crate::job_steps::main_loop_old::main_loop_old;
 use crate::script_execution::tacotron_inference_command::TacotronInferenceCommand;
 use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
 use database_queries::mediators::firehose_publisher::FirehosePublisher;
@@ -286,10 +285,11 @@ async fn main() -> AnyhowResult<()> {
     }
   };
 
-  let inferencer = JobDependencies {
+  let _job_dependencies = JobDependencies {
     scoped_temp_dir_creator: ScopedTempDirCreator::for_directory(&temp_directory),
     download_temp_directory: temp_directory,
     mysql_pool,
+    maybe_redis_pool: None, // TODO(bt, 2023-01-11): See note in JobDependencies
     job_progress_reporter,
     public_bucket_client,
     private_bucket_client,
@@ -320,7 +320,7 @@ async fn main() -> AnyhowResult<()> {
     hifigan_vocoder_model_filename,
     hifigan_superres_vocoder_model_filename,
     job_batch_wait_millis: common_env.job_batch_wait_millis,
-    job_max_attempts: common_env.job_max_attempts as i32,
+    job_max_attempts: common_env.job_max_attempts as u16,
     job_batch_size: common_env.job_batch_size,
     no_op_logger_millis: common_env.no_op_logger_millis,
     sidecar_max_synthesizer_models,
@@ -328,7 +328,8 @@ async fn main() -> AnyhowResult<()> {
     maybe_minimum_priority,
   };
 
-  main_loop_old(inferencer).await;
+  // TODO: Main loop start
+  //main_loop_old(inferencer).await;
 
   Ok(())
 }

crates/service/web/storyteller_web/src/http_server/endpoints/media_uploads/upload_media.rs

@@ -14,6 +14,7 @@ use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
 use crate::server_state::ServerState;
 use crate::validations::model_uploads::validate_model_title;
 use crate::validations::validate_idempotency_token_format::validate_idempotency_token_format;
+use database_queries::payloads::media_upload_details::MediaUploadDetails;
 use database_queries::queries::media_uploads::insert_media_upload::{Args, insert_media_upload};
 use enums::core::visibility::Visibility;
 use enums::files::media_upload_type::MediaUploadType;
@@ -24,7 +25,6 @@ use http_server_common::response::serialize_as_json_error::serialize_as_json_err
 use log::{info, warn, log};
 use media::decode_basic_audio_info::decode_basic_audio_info;
 use regex::Regex;
-use reusable_types::db::payloads::MediaUploadDetails;
 use sqlx::error::DatabaseError;
 use sqlx::error::Error::Database;
 use sqlx::mysql::MySqlDatabaseError;