diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/inference/get_voice_conversion_model_for_inference.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/inference/get_voice_conversion_model_for_inference.rs index 815b4519cb..db1384150a 100644 --- a/crates/lib/mysql_queries/src/queries/voice_conversion/inference/get_voice_conversion_model_for_inference.rs +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/inference/get_voice_conversion_model_for_inference.rs @@ -92,7 +92,7 @@ WHERE vc.token = ? return Ok(None); }, _ => { - warn!("tts model query error: {:?}", err); + warn!("voice conversion model query error: {:?}", err); return Err(VoiceConversionModelForInferenceError::DatabaseError { reason: format!("Mysql error: {:?}", err) }); diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/mod.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/mod.rs index e5a97c880c..962214e104 100644 --- a/crates/lib/mysql_queries/src/queries/voice_conversion/mod.rs +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/mod.rs @@ -1,3 +1,4 @@ pub mod inference; +pub mod model_info_lite; pub mod models; pub mod results; \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/get_voice_conversion_model_info_lite.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/get_voice_conversion_model_info_lite.rs new file mode 100644 index 0000000000..e7b7d4b6ac --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/get_voice_conversion_model_info_lite.rs @@ -0,0 +1,61 @@ +use anyhow::anyhow; +use crate::queries::voice_conversion::model_info_lite::model_info_lite::VoiceConversionModelInfoLite; +use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; +use errors::AnyhowResult; +use log::warn; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; +use tokens::voice_conversion::model::VoiceConversionModelToken; + +pub async fn get_voice_conversion_model_info_lite( + token: &str, + mysql_pool: &MySqlPool, +) -> AnyhowResult> { + let mut connection = mysql_pool.acquire().await?; + get_voice_conversion_model_info_lite_with_connection(token, &mut connection).await +} + +pub async fn get_voice_conversion_model_info_lite_with_connection( + token: &str, + mysql_connection: &mut PoolConnection, +) -> AnyhowResult> { + + let maybe_model= sqlx::query_as!( + RawVoiceConversionModelInfoLite, + r#" +SELECT + vc.token as `token: tokens::voice_conversion::model::VoiceConversionModelToken`, + vc.model_type as `model_type: enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType` +FROM voice_conversion_models as vc +WHERE vc.token = ? + "#, + token + ) + .fetch_one(mysql_connection) + .await; + + let model : RawVoiceConversionModelInfoLite = match maybe_model { + Ok(model) => model, + Err(err) => { + return match err { + sqlx::Error::RowNotFound => { + Ok(None) + }, + _ => { + warn!("voice conversion model query error: {:?}", err); + Err(anyhow!("error querying voce conversion model info lite: {:?}", err)) + } + } + } + }; + + Ok(Some(VoiceConversionModelInfoLite { + token: model.token, + model_type: model.model_type, + })) +} + +struct RawVoiceConversionModelInfoLite { + pub token: VoiceConversionModelToken, + pub model_type: VoiceConversionModelType, +} diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/models/list_voice_conversion_model_info_lite.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/list_voice_conversion_model_info_lite.rs similarity index 83% rename from crates/lib/mysql_queries/src/queries/voice_conversion/models/list_voice_conversion_model_info_lite.rs rename to crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/list_voice_conversion_model_info_lite.rs index 74c9d58019..d4f263002a 100644 --- a/crates/lib/mysql_queries/src/queries/voice_conversion/models/list_voice_conversion_model_info_lite.rs +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/list_voice_conversion_model_info_lite.rs @@ -1,17 +1,10 @@ +use crate::queries::voice_conversion::model_info_lite::model_info_lite::VoiceConversionModelInfoLite; use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; use errors::AnyhowResult; use sqlx::pool::PoolConnection; use sqlx::{MySql, MySqlPool}; use tokens::voice_conversion::model::VoiceConversionModelToken; -/// This is meant to be used for quick in-memory caches, -/// particularly the one that serves the voice conversion enqueue API. -#[derive(Serialize)] -pub struct VoiceConversionModelInfoLite { - pub token: VoiceConversionModelToken, - pub model_type: VoiceConversionModelType, -} - pub async fn list_voice_conversion_model_info_lite( mysql_pool: &MySqlPool, ) -> AnyhowResult> { diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/mod.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/mod.rs new file mode 100644 index 0000000000..c5096011bd --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/mod.rs @@ -0,0 +1,3 @@ +pub mod get_voice_conversion_model_info_lite; +pub mod list_voice_conversion_model_info_lite; +pub mod model_info_lite; diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/model_info_lite.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/model_info_lite.rs new file mode 100644 index 0000000000..a52fa55ef8 --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/model_info_lite/model_info_lite.rs @@ -0,0 +1,10 @@ +use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; +use tokens::voice_conversion::model::VoiceConversionModelToken; + +/// This is meant to be used for quick in-memory caches, +/// particularly the one that serves the voice conversion enqueue API. +#[derive(Serialize)] +pub struct VoiceConversionModelInfoLite { + pub token: VoiceConversionModelToken, + pub model_type: VoiceConversionModelType, +} diff --git a/crates/lib/mysql_queries/src/queries/voice_conversion/models/mod.rs b/crates/lib/mysql_queries/src/queries/voice_conversion/models/mod.rs index d05753506c..3a04e54d4f 100644 --- a/crates/lib/mysql_queries/src/queries/voice_conversion/models/mod.rs +++ b/crates/lib/mysql_queries/src/queries/voice_conversion/models/mod.rs @@ -1,3 +1,2 @@ pub mod insert_voice_conversion_model_from_download_job; -pub mod list_voice_conversion_model_info_lite; pub mod list_voice_conversion_models; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories/list_fully_computed_assigned_tts_categories.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories/list_fully_computed_assigned_tts_categories.rs index d00b43a1a1..2cdcdb9bf5 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories/list_fully_computed_assigned_tts_categories.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories/list_fully_computed_assigned_tts_categories.rs @@ -40,7 +40,7 @@ pub async fn list_fully_computed_assigned_tts_categories_handler( _http_request: HttpRequest, server_state: web::Data>) -> Result { - let maybe_category_assignments = server_state.caches.tts_model_category_assignments.grab_copy_without_bump_if_unexpired() + let maybe_category_assignments = server_state.caches.ephemeral.tts_model_category_assignments.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("Error consulting cache: {:?}", e); ListFullyComputedAssignedTtsCategoriesError::ServerError @@ -53,11 +53,11 @@ pub async fn list_fully_computed_assigned_tts_categories_handler( }, None => { let category_assignments = query_and_construct_payload( - &server_state.caches.database_tts_category_list, + &server_state.caches.ephemeral.database_tts_category_list, &server_state.mysql_pool) .await?; - server_state.caches.tts_model_category_assignments.store_copy(&category_assignments) + server_state.caches.ephemeral.tts_model_category_assignments.store_copy(&category_assignments) .map_err(|e| { error!("Error storing cache: {:?}", e); ListFullyComputedAssignedTtsCategoriesError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_tts_categories.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_tts_categories.rs index 29d69ba918..0764b5745d 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_tts_categories.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_tts_categories.rs @@ -95,7 +95,7 @@ pub async fn list_tts_categories_handler( server_state: web::Data>) -> Result { let database_categories = list_cached_tts_categories_for_public_dropdown_db_pool( - &server_state.caches.database_tts_category_list, &server_state.mysql_pool) + &server_state.caches.ephemeral.database_tts_category_list, &server_state.mysql_pool) .await .map_err(|e| { error!("error querying categories from db/cache: {:?}", e); diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_pending_inference_job_count.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_pending_inference_job_count.rs index 22042ec2b8..edac18447c 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_pending_inference_job_count.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_pending_inference_job_count.rs @@ -65,7 +65,7 @@ pub async fn get_pending_inference_job_count_handler( }); } - let maybe_cached = server_state.caches.inference_queue_length.grab_copy_without_bump_if_unexpired() + let maybe_cached = server_state.caches.ephemeral.inference_queue_length.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("error consulting cache: {:?}", e); GetPendingInferenceJobCountError::ServerError @@ -88,7 +88,7 @@ pub async fn get_pending_inference_job_count_handler( Err(err) => { warn!("error querying database / inserting into cache: {:?}", err); - let maybe_cached = server_state.caches.inference_queue_length.grab_even_expired_and_bump() + let maybe_cached = server_state.caches.ephemeral.inference_queue_length.grab_even_expired_and_bump() .map_err(|err| { error!("error consulting cache (even expired): {:?}", err); GetPendingInferenceJobCountError::ServerError @@ -102,7 +102,7 @@ pub async fn get_pending_inference_job_count_handler( // Happy path... Ok(count_result) => { - server_state.caches.inference_queue_length.store_copy(&count_result) + server_state.caches.ephemeral.inference_queue_length.store_copy(&count_result) .map_err(|e| { error!("error storing cache: {:?}", e); GetPendingInferenceJobCountError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/leaderboard/get_leaderboard.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/leaderboard/get_leaderboard.rs index cd79268407..11013282c2 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/leaderboard/get_leaderboard.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/leaderboard/get_leaderboard.rs @@ -84,7 +84,7 @@ pub async fn leaderboard_handler( server_state: web::Data> ) -> Result { - let maybe_cached = server_state.caches.leaderboard.grab_copy_without_bump_if_unexpired() + let maybe_cached = server_state.caches.ephemeral.leaderboard.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("error consulting cache: {:?}", e); LeaderboardErrorResponse::server_error() @@ -108,7 +108,7 @@ pub async fn leaderboard_handler( Err(err) => { warn!("error querying database / inserting into cache: {:?}", err); - let maybe_cached = server_state.caches.leaderboard.grab_even_expired_and_bump() + let maybe_cached = server_state.caches.ephemeral.leaderboard.grab_even_expired_and_bump() .map_err(|err| { error!("error consulting cache (even expired): {:?}", err); LeaderboardErrorResponse::server_error() @@ -122,7 +122,7 @@ pub async fn leaderboard_handler( // Happy path... Ok(leaderboard_info) => { - server_state.caches.leaderboard.store_copy(&leaderboard_info) + server_state.caches.ephemeral.leaderboard.store_copy(&leaderboard_info) .map_err(|e| { error!("error storing cache: {:?}", e); LeaderboardErrorResponse::server_error() diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/stats/get_unified_queue_stats.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/stats/get_unified_queue_stats.rs index 347f985321..b5f7ae9dbf 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/stats/get_unified_queue_stats.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/stats/get_unified_queue_stats.rs @@ -79,7 +79,7 @@ pub async fn get_unified_queue_stats_handler( }); } - let maybe_cached = server_state.caches.queue_stats.grab_copy_without_bump_if_unexpired() + let maybe_cached = server_state.caches.ephemeral.queue_stats.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("error consulting cache: {:?}", e); GetUnifiedQueueStatsError::ServerError @@ -102,7 +102,7 @@ pub async fn get_unified_queue_stats_handler( Err(err) => { warn!("error querying database / inserting into cache: {:?}", err); - let maybe_cached = server_state.caches.queue_stats.grab_even_expired_and_bump() + let maybe_cached = server_state.caches.ephemeral.queue_stats.grab_even_expired_and_bump() .map_err(|err| { error!("error consulting cache (even expired): {:?}", err); GetUnifiedQueueStatsError::ServerError @@ -116,7 +116,7 @@ pub async fn get_unified_queue_stats_handler( // Happy path... Ok(stats_result) => { - server_state.caches.queue_stats.store_copy(&stats_result) + server_state.caches.ephemeral.queue_stats.store_copy(&stats_result) .map_err(|e| { error!("error storing cache: {:?}", e); GetUnifiedQueueStatsError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_pending_tts_inference_job_count.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_pending_tts_inference_job_count.rs index 5860e49ae0..d1192cd430 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_pending_tts_inference_job_count.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_pending_tts_inference_job_count.rs @@ -66,7 +66,7 @@ pub async fn get_pending_tts_inference_job_count_handler( }); } - let maybe_cached = server_state.caches.tts_queue_length.grab_copy_without_bump_if_unexpired() + let maybe_cached = server_state.caches.ephemeral.tts_queue_length.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("error consulting cache: {:?}", e); GetPendingTtsInferenceJobCountError::ServerError @@ -89,7 +89,7 @@ pub async fn get_pending_tts_inference_job_count_handler( Err(err) => { warn!("error querying database / inserting into cache: {:?}", err); - let maybe_cached = server_state.caches.tts_queue_length.grab_even_expired_and_bump() + let maybe_cached = server_state.caches.ephemeral.tts_queue_length.grab_even_expired_and_bump() .map_err(|err| { error!("error consulting cache (even expired): {:?}", err); GetPendingTtsInferenceJobCountError::ServerError @@ -103,7 +103,7 @@ pub async fn get_pending_tts_inference_job_count_handler( // Happy path... Ok(count_result) => { - server_state.caches.tts_queue_length.store_copy(&count_result) + server_state.caches.ephemeral.tts_queue_length.store_copy(&count_result) .map_err(|e| { error!("error storing cache: {:?}", e); GetPendingTtsInferenceJobCountError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/list_tts_models.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/list_tts_models.rs index 5ff71188b5..b9d98839a4 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/list_tts_models.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/list_tts_models.rs @@ -105,7 +105,7 @@ pub async fn list_tts_models_handler( }); } - let maybe_models = server_state.caches.tts_model_list.grab_copy_without_bump_if_unexpired() + let maybe_models = server_state.caches.ephemeral.tts_model_list.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("Error consulting cache: {:?}", e); ListTtsModelsError::ServerError @@ -139,7 +139,7 @@ pub async fn list_tts_models_handler( maybe_mysql_connection = Some(mysql_connection); - server_state.caches.tts_model_list.store_copy(&models) + server_state.caches.ephemeral.tts_model_list.store_copy(&models) .map_err(|e| { error!("Error storing cache: {:?}", e); ListTtsModelsError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/inference/enqueue_voice_conversion_inference.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/inference/enqueue_voice_conversion_inference.rs index 633e3ba996..c44f72793b 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/inference/enqueue_voice_conversion_inference.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/inference/enqueue_voice_conversion_inference.rs @@ -9,20 +9,25 @@ use actix_web::{web, HttpResponse, HttpRequest}; use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session; use crate::http_server::endpoints::investor_demo::demo_cookie::request_has_demo_cookie; use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; +use crate::memory_cache::model_token_to_info_cache::{ModelInfoLite, ModelTokenToInfoCache}; use crate::server_state::ServerState; use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType; +use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType; +use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; use enums::common::visibility::Visibility; use http_server_common::request::get_request_header_optional::get_request_header_optional; use http_server_common::request::get_request_ip::get_request_ip; -use log::{info, warn}; +use log::{error, info, warn}; use mysql_queries::payloads::generic_inference_args::{FundamentalFrequencyMethodForJob, GenericInferenceArgs, InferenceCategoryAbbreviated, PolymorphicInferenceArgs}; use mysql_queries::queries::generic_inference::web::insert_generic_inference_job::{InsertGenericInferenceArgs, insert_generic_inference_job}; +use mysql_queries::queries::voice_conversion::model_info_lite::get_voice_conversion_model_info_lite::get_voice_conversion_model_info_lite_with_connection; use r2d2_redis::redis::Commands; use redis_common::redis_keys::RedisKeys; +use sqlx::MySql; +use sqlx::pool::PoolConnection; use std::fmt; use std::sync::Arc; -use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType; -use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType; use tokens::files::media_upload::MediaUploadToken; use tokens::jobs::inference::InferenceJobToken; use tokens::users::user::UserToken; @@ -213,16 +218,24 @@ pub async fn enqueue_voice_conversion_inference_handler( } } - // ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== // + // ==================== LOOK UP MODEL INFO ==================== // // TODO(bt): CHECK DATABASE! let model_token = request.voice_conversion_model_token.clone(); let media_token = request.source_media_upload_token.clone(); + let model_info_lite = lookup_model_info( + &model_token, + &server_state.caches.durable.model_token_info, + &mut mysql_connection + ).await?; + + // ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== // + let mut redis = server_state.redis_pool .get() .map_err(|e| { - warn!("redis error: {:?}", e); + error!("redis error: {:?}", e); EnqueueVoiceConversionInferenceError::ServerError })?; @@ -268,7 +281,7 @@ pub async fn enqueue_voice_conversion_inference_handler( let query_result = insert_generic_inference_job(InsertGenericInferenceArgs { uuid_idempotency_token: &request.uuid_idempotency_token, inference_category: InferenceCategory::VoiceConversion, - maybe_model_type: Some(InferenceModelType::SoVitsSvc), // TODO/FIXME: This is stupidly wrong. + maybe_model_type: Some(model_info_lite.model_type), maybe_model_token: Some(model_token.as_str()), maybe_input_source_token: Some(media_token.as_str()), maybe_input_source_token_type: Some(InferenceInputSourceTokenType::MediaUpload), @@ -316,3 +329,64 @@ pub async fn enqueue_voice_conversion_inference_handler( .content_type("application/json") .body(body)) } + +async fn lookup_model_info( + model_token: &VoiceConversionModelToken, + cache: &ModelTokenToInfoCache, + mysql_connection: &mut PoolConnection, +) -> Result { + let maybe_model_token_info = cache + .get_info(model_token.as_str()) + .map_err(|err| { + error!("in-memory cache error: {:?}", err); + EnqueueVoiceConversionInferenceError::ServerError + })?; + + let model_token_info = match maybe_model_token_info { + Some(info) => info, + None => { + let result = get_voice_conversion_model_info_lite_with_connection( + model_token.as_str(), mysql_connection) + .await + .map_err(|err| { + error!("model lookup error: {:?}", err); + EnqueueVoiceConversionInferenceError::ServerError + })?; + + match result { + Some(info) => { + let model_type = match info.model_type { + VoiceConversionModelType::RvcV2 => InferenceModelType::RvcV2, + VoiceConversionModelType::SoVitsSvc => InferenceModelType::SoVitsSvc, + VoiceConversionModelType::SoftVc => { + // SoftVC is unsupported + return Err(EnqueueVoiceConversionInferenceError::BadInput("wrong model type".to_string())); + } + }; + + let model_info = ModelInfoLite { + inference_category: InferenceCategory::VoiceConversion, + model_type, + }; + + cache.insert_one(model_token.as_str(), &model_info) + .map_err(|err| { + error!("cache insertion error: {:?}", err); + EnqueueVoiceConversionInferenceError::ServerError + })?; + + model_info + } + None => { + // NB: Fail open for now. + ModelInfoLite { + inference_category: InferenceCategory::VoiceConversion, + model_type: InferenceModelType::SoVitsSvc, + } + } + } + } + }; + + Ok(model_token_info) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/models/list_voice_conversion_models.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/models/list_voice_conversion_models.rs index dce48e5b7a..186b17548b 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/models/list_voice_conversion_models.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/models/list_voice_conversion_models.rs @@ -113,7 +113,7 @@ pub async fn list_voice_conversion_models_handler( }); } - let maybe_models = server_state.caches.voice_conversion_model_list.grab_copy_without_bump_if_unexpired() + let maybe_models = server_state.caches.ephemeral.voice_conversion_model_list.grab_copy_without_bump_if_unexpired() .map_err(|e| { error!("Error consulting cache: {:?}", e); ListVoiceConversionModelsError::ServerError @@ -146,7 +146,7 @@ pub async fn list_voice_conversion_models_handler( maybe_mysql_connection = Some(mysql_connection); - server_state.caches.voice_conversion_model_list.store_copy(&models) + server_state.caches.ephemeral.voice_conversion_model_list.store_copy(&models) .map_err(|e| { error!("Error storing cache: {:?}", e); ListVoiceConversionModelsError::ServerError diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/w2l/list_w2l_templates.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/w2l/list_w2l_templates.rs index 6ade45c837..1bd5691643 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/w2l/list_w2l_templates.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/w2l/list_w2l_templates.rs @@ -61,6 +61,7 @@ pub async fn list_w2l_templates_handler( let maybe_templates = server_state .caches + .ephemeral .w2l_template_list .grab_copy_without_bump_if_unexpired() .map_err(|e| { @@ -89,7 +90,7 @@ pub async fn list_w2l_templates_handler( } }; - server_state.caches.w2l_template_list.store_copy(&templates) + server_state.caches.ephemeral.w2l_template_list.store_copy(&templates) .map_err(|e| { error!("error storing cache: {:?}", e); ListW2lTemplatesError::ServerError diff --git a/crates/service/web/storyteller_web/src/main.rs b/crates/service/web/storyteller_web/src/main.rs index 387e47edb6..c197340d52 100644 --- a/crates/service/web/storyteller_web/src/main.rs +++ b/crates/service/web/storyteller_web/src/main.rs @@ -62,7 +62,7 @@ use crate::configs::static_api_tokens::{StaticApiTokenConfig, StaticApiTokens, S use crate::http_server::middleware::pushback_filter_middleware::PushbackFilter; use crate::http_server::web_utils::redis_rate_limiter::RedisRateLimiter; use crate::routes::add_routes; -use crate::server_state::{ServerState, EnvConfig, TwitchOauthSecrets, TwitchOauth, RedisRateLimiters, InMemoryCaches, StripeSettings, ServerInfo, StaticFeatureFlags, TrollBans}; +use crate::server_state::{ServerState, EnvConfig, TwitchOauthSecrets, TwitchOauth, RedisRateLimiters, InMemoryCaches, StripeSettings, ServerInfo, StaticFeatureFlags, TrollBans, DurableInMemoryCaches, EphemeralInMemoryCaches}; use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus; use crate::threads::db_health_checker_thread::db_health_checker_thread::db_health_checker_thread; use crate::threads::poll_ip_banlist_thread::poll_ip_bans; @@ -95,6 +95,8 @@ use twitch_common::twitch_secrets::TwitchSecrets; use url_config::third_party_url_redirector::ThirdPartyUrlRedirector; use users_component::utils::session_checker::SessionChecker; use users_component::utils::session_cookie_manager::SessionCookieManager; +use crate::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache; +use crate::threads::poll_model_token_info_thread::poll_model_token_info_thread; const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345"; @@ -330,12 +332,16 @@ async fn main() -> AnyhowResult<()> { let user_token_troll_bans = load_troll_user_token_bans(); let ip_address_troll_bans = load_ip_address_troll_bans(); + let model_token_info_cache = ModelTokenToInfoCache::new(); + let model_token_info_cache2 = model_token_info_cache.clone(); + // Background jobs. let health_check_status = HealthCheckStatus::new(); let health_check_status2 = health_check_status.clone(); let mysql_pool3 = pool.clone(); let mysql_pool4 = pool.clone(); + let mysql_pool5 = pool.clone(); let tokio_runtime = Runtime::new()?; @@ -355,6 +361,13 @@ async fn main() -> AnyhowResult<()> { poll_ip_bans(ip_ban_list2, mysql_pool4).await; }); + info!("Spawning token info cache polling thread."); + + tokio_runtime.spawn(async { + poll_model_token_info_thread(model_token_info_cache2, mysql_pool5).await; + }); + + let stripe_configs = StripeConfig { checkout: StripeCheckoutConfigs { success_url: FullUrlOrPath::Path(easyenv::get_env_string_required("STRIPE_CHECKOUT_SUCCESS_URL_PATH")?), @@ -448,21 +461,26 @@ async fn main() -> AnyhowResult<()> { sort_key_crypto, static_api_token_set, caches: InMemoryCaches { - tts_model_list: voice_list_cache, - voice_conversion_model_list: SingleItemTtlCache::create_with_duration( - easyenv::get_env_duration_seconds_or_default( - "VOICE_CONVERSION_MODEL_LIST_CACHE_TTL_SECONDS", - Duration::from_secs(60))), - w2l_template_list: w2l_template_cache, - database_tts_category_list: database_tts_category_list_cache, - tts_queue_length: tts_queue_length_cache, - tts_model_category_assignments: tts_model_category_assignments_cache, - leaderboard: leaderboard_cache, - inference_queue_length: inference_queue_length_cache, - queue_stats: SingleItemTtlCache::create_with_duration( - easyenv::get_env_duration_seconds_or_default( - "QUEUE_STATS_CACHE_TTL_SECONDS", - Duration::from_secs(60))), + durable: DurableInMemoryCaches { + model_token_info: model_token_info_cache, + }, + ephemeral: EphemeralInMemoryCaches { + tts_model_list: voice_list_cache, + voice_conversion_model_list: SingleItemTtlCache::create_with_duration( + easyenv::get_env_duration_seconds_or_default( + "VOICE_CONVERSION_MODEL_LIST_CACHE_TTL_SECONDS", + Duration::from_secs(60))), + w2l_template_list: w2l_template_cache, + database_tts_category_list: database_tts_category_list_cache, + tts_queue_length: tts_queue_length_cache, + tts_model_category_assignments: tts_model_category_assignments_cache, + leaderboard: leaderboard_cache, + inference_queue_length: inference_queue_length_cache, + queue_stats: SingleItemTtlCache::create_with_duration( + easyenv::get_env_duration_seconds_or_default( + "QUEUE_STATS_CACHE_TTL_SECONDS", + Duration::from_secs(60))), + } }, twitch_oauth: TwitchOauth { secrets: TwitchOauthSecrets { diff --git a/crates/service/web/storyteller_web/src/memory_cache/model_token_to_info_cache.rs b/crates/service/web/storyteller_web/src/memory_cache/model_token_to_info_cache.rs index 77f400289d..aa17a163ef 100644 --- a/crates/service/web/storyteller_web/src/memory_cache/model_token_to_info_cache.rs +++ b/crates/service/web/storyteller_web/src/memory_cache/model_token_to_info_cache.rs @@ -25,22 +25,22 @@ impl ModelTokenToInfoCache { } } - pub fn insert_one(&self, token: &str, info: ModelInfoLite) -> AnyhowResult<()> { + pub fn insert_one(&self, token: &str, info: &ModelInfoLite) -> AnyhowResult<()> { match self.database.write() { Err(err) => return Err(anyhow!("lock err: {:?}", err)), Ok(mut lock) => { - lock.insert(token.to_string(), info); + lock.insert(token.to_string(), info.clone()); } } Ok(()) } - pub fn insert_many(&self, records: Vec<(&str, ModelInfoLite)>) -> AnyhowResult<()> { + pub fn insert_many(&self, records: Vec<(String, ModelInfoLite)>) -> AnyhowResult<()> { match self.database.write() { Err(err) => return Err(anyhow!("lock err: {:?}", err)), Ok(mut lock) => { for (token, info) in records.into_iter() { - lock.insert(token.to_string(), info); + lock.insert(token, info); } } } diff --git a/crates/service/web/storyteller_web/src/server_state.rs b/crates/service/web/storyteller_web/src/server_state.rs index 045dfc23d9..9ccb1edd5d 100644 --- a/crates/service/web/storyteller_web/src/server_state.rs +++ b/crates/service/web/storyteller_web/src/server_state.rs @@ -27,6 +27,7 @@ use sqlx::MySqlPool; use url_config::third_party_url_redirector::ThirdPartyUrlRedirector; use users_component::utils::session_checker::SessionChecker; use users_component::utils::session_cookie_manager::SessionCookieManager; +use crate::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache; /// State that is injected into every endpoint. pub struct ServerState { @@ -134,9 +135,23 @@ pub struct RedisRateLimiters { pub model_upload: RedisRateLimiter, } -/// In-memory caches with TTL-based eviction. +/// In-memory caches of several types. #[derive(Clone)] pub struct InMemoryCaches { + pub durable: DurableInMemoryCaches, + pub ephemeral: EphemeralInMemoryCaches, +} + +/// Durable caches +#[derive(Clone)] +pub struct DurableInMemoryCaches { + /// Lightweight model token (for generic inference) metadata + pub model_token_info: ModelTokenToInfoCache, +} + +/// In-memory caches with TTL-based eviction. +#[derive(Clone)] +pub struct EphemeralInMemoryCaches { /// Contains a list of all TTS models. pub tts_model_list: SingleItemTtlCache>, @@ -179,7 +194,6 @@ pub struct StripeSettings { pub client: stripe::Client, } - /// Flags set at service startup #[derive(Clone)] pub struct StaticFeatureFlags { diff --git a/crates/service/web/storyteller_web/src/threads/mod.rs b/crates/service/web/storyteller_web/src/threads/mod.rs index 5273919ae2..5804badbad 100644 --- a/crates/service/web/storyteller_web/src/threads/mod.rs +++ b/crates/service/web/storyteller_web/src/threads/mod.rs @@ -1,2 +1,3 @@ pub mod db_health_checker_thread; -pub mod poll_ip_banlist_thread; \ No newline at end of file +pub mod poll_ip_banlist_thread; +pub mod poll_model_token_info_thread; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/threads/poll_model_token_info_thread.rs b/crates/service/web/storyteller_web/src/threads/poll_model_token_info_thread.rs new file mode 100644 index 0000000000..65a924d5d5 --- /dev/null +++ b/crates/service/web/storyteller_web/src/threads/poll_model_token_info_thread.rs @@ -0,0 +1,64 @@ +use crate::memory_cache::model_token_to_info_cache::{ModelInfoLite, ModelTokenToInfoCache}; +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType; +use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; +use log::{debug, error, info}; +use mysql_queries::queries::voice_conversion::model_info_lite::list_voice_conversion_model_info_lite::list_voice_conversion_model_info_lite; +use sqlx::MySqlPool; +use std::time::Duration; + +pub async fn poll_model_token_info_thread( + model_token_info_cache: ModelTokenToInfoCache, + mysql_pool: MySqlPool, +) { + let startup_wait = easyenv::get_env_duration_seconds_or_default( + "POLL_MODEL_TOKEN_STARTUP_WAIT_DURATION_SECS", Duration::from_secs(5)); + + let interval_wait = easyenv::get_env_duration_seconds_or_default( + "POLL_MODEL_TOKEN_INTERVAL_SECS", Duration::from_secs(10 * 60)); + + std::thread::sleep(startup_wait); + + loop { + debug!("Job fetching token info..."); + + let token_infos= + match list_voice_conversion_model_info_lite(&mysql_pool).await { + Ok(infos) => infos, + Err(err) => { + error!("Error polling model token info: {:?}", err); + std::thread::sleep(interval_wait); + continue; + } + }; + + let mut token_info_items = Vec::with_capacity(token_infos.len()); + + for token_info in token_infos.into_iter() { + let model_type = match token_info.model_type { + VoiceConversionModelType::RvcV2 => InferenceModelType::RvcV2, + VoiceConversionModelType::SoVitsSvc => InferenceModelType::SoVitsSvc, + VoiceConversionModelType::SoftVc => { + continue // NB: SoftVC is not supported. + }, + }; + + let info = ModelInfoLite { + inference_category: InferenceCategory::VoiceConversion, + model_type: InferenceModelType::RvcV2, + }; + + token_info_items.push((token_info.token.to_string(), info)); + } + + let database_count = token_info_items.len(); + + info!("Job found {} token info items from database.", database_count); + + if let Err(err) = model_token_info_cache.insert_many(token_info_items) { + error!("Error inserting model token info: {:?}", err); + } + + std::thread::sleep(interval_wait); + } +}