From 922e76eeb11a09535e910da57a332e39a9df1605 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 19 Dec 2023 21:53:04 -0500 Subject: [PATCH 1/2] migration fix wip --- .../copy_cloud_files.rs | 4 +- .../voice_conversion_to_weights/migrate.rs | 60 +++++++++++++++++-- ...odel_weight_from_voice_conversion_model.rs | 9 ++- 3 files changed, 61 insertions(+), 12 deletions(-) diff --git a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs index 83d9c16f42..9d5f3468de 100644 --- a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs +++ b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs @@ -43,7 +43,7 @@ async fn copy_model(model: &WholeVoiceConversionModelRecord, deps: &Deps) -> Any VoiceConversionModelType::SoftVc => return Err(anyhow!("we never built softvc models")), }; - deps.bucket_development_public.upload_filename_with_content_type( + deps.bucket_production_public.upload_filename_with_content_type( &new_model_bucket_path.get_full_object_path_str(), &model_temp_fs_path, "application/octet-stream").await?; @@ -68,7 +68,7 @@ async fn copy_index_file(model: &WholeVoiceConversionModelRecord, deps: &Deps, b let new_model_bucket_path = WeightFileBucketPath::rvc_index_file_from_object_hash(bucket_path.get_object_hash()); - deps.bucket_development_public.upload_filename_with_content_type( + deps.bucket_production_public.upload_filename_with_content_type( &new_model_bucket_path.get_full_object_path_str(), &model_temp_fs_path, "application/octet-stream").await?; diff --git a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs index 869808a36e..4263697209 100644 --- a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs +++ b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs @@ -1,9 +1,15 @@ use std::thread; use std::time::Duration; +use log::{info, warn}; + +use buckets::public::weight_files::bucket_file_path::WeightFileBucketPath; +use enums::by_table::model_weights::weights_types::WeightsType; use errors::AnyhowResult; -use mysql_queries::queries::model_weights::migration::upsert_model_weight_from_voice_conversion_model::upsert_model_weight_from_voice_conversion_model; -use mysql_queries::queries::voice_conversion::migration::list_whole_voice_conversion_models_using_cursor::list_whole_voice_conversion_models_using_cursor; +use mysql_queries::queries::model_weights::get_weight::get_weight_by_token; +use mysql_queries::queries::model_weights::migration::upsert_model_weight_from_voice_conversion_model::{CopiedFileData, upsert_model_weight_from_voice_conversion_model}; +use mysql_queries::queries::voice_conversion::migration::list_whole_voice_conversion_models_using_cursor::{list_whole_voice_conversion_models_using_cursor, WholeVoiceConversionModelRecord}; +use tokens::tokens::model_weights::ModelWeightToken; use crate::deps::Deps; use crate::migrations::voice_conversion_to_weights::copy_cloud_files::copy_cloud_files; @@ -15,17 +21,19 @@ pub async fn migrate_voice_conversion_to_weights(deps: &Deps) -> AnyhowResult<() let mut cursor = 0; loop { + info!("Querying {PAGE_SIZE} models at cursor = {cursor}"); + let results = list_whole_voice_conversion_models_using_cursor(&deps.mysql_production, PAGE_SIZE, cursor).await?; if results.is_empty() { + warn!("No more results found; exiting."); break; } for result in results.iter() { - println!("result: {:?}\n\n", result); - let copied_data = copy_cloud_files(result, &deps).await?; - upsert_model_weight_from_voice_conversion_model(result, &deps.mysql_development, &copied_data).await?; + println!("\n\nmigrating over result {:?} : {:?} ...", result.token, result); + copy_model_record_and_bucket_files(result, &deps).await?; } if let Some(last_id) = results.last().map(|result| result.id) { @@ -37,3 +45,45 @@ pub async fn migrate_voice_conversion_to_weights(deps: &Deps) -> AnyhowResult<() Ok(()) } + +async fn copy_model_record_and_bucket_files(record: &WholeVoiceConversionModelRecord, deps: &Deps) -> AnyhowResult<()> { + let mut maybe_copied_data = None; + + match &record.maybe_migration_new_model_weights_token { + None => { + info!("Copying bucket files for record {:?} ...", record.token); + let result = copy_cloud_files(record, &deps).await?; + maybe_copied_data = Some(result); + } + Some(token) => { + info!("Existing migrated record; skipping cloud bucket copy..."); + maybe_copied_data = lookup_bucket_details_from_record(token, &deps).await?; + } + } + + upsert_model_weight_from_voice_conversion_model(record, &deps.mysql_production, maybe_copied_data.as_ref()).await?; + + Ok(()) +} + +/// Look up the bucket storage details from a previously-migrated record +async fn lookup_bucket_details_from_record(model_weight_token: &ModelWeightToken, deps: &Deps) -> AnyhowResult> { + let maybe_model_weight_record = + get_weight_by_token(model_weight_token, true, &deps.mysql_production).await?; + + let model_weight_record = match maybe_model_weight_record { + None => return Ok(None), + Some(record) => record, + }; + + let bucket_path = match model_weight_record.weights_type { + WeightsType::RvcV2 => WeightFileBucketPath::rvc_model_file_from_object_hash(&model_weight_record.private_bucket_hash), + WeightsType::SoVitsSvc => WeightFileBucketPath::svc_model_file_from_object_hash(&model_weight_record.private_bucket_hash), + _ => return Ok(None), // NB: Stable Diffusion, etc. aren't valid for migration + }; + + Ok(Some(CopiedFileData { + bucket_path, + file_sha_hash: model_weight_record.file_checksum_sha2, + })) +} \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs b/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs index 5c480ce813..bbcc71323e 100644 --- a/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs +++ b/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs @@ -22,18 +22,17 @@ pub struct CopiedFileData { pub async fn upsert_model_weight_from_voice_conversion_model( record: &WholeVoiceConversionModelRecord, mysql_pool: &MySqlPool, - copied_data: &CopiedFileData, + maybe_copied_data: Option<&CopiedFileData>, ) -> AnyhowResult<()> { let mut transaction = mysql_pool.begin().await?; let model_weight_token = create_or_generate_token(record); - upsert_model_weights_record(record, &model_weight_token, copied_data, &mut transaction).await?; + upsert_model_weights_record(record, &model_weight_token, maybe_copied_data, &mut transaction).await?; upsert_model_weights_extension_record(record, &model_weight_token, &mut transaction).await?; - // TODO: Don't update source record with new token *unless* we're backfilling the same database. - // update_original_record(record, &model_weight_token, &mut transaction).await?; + update_original_record(record, &model_weight_token, &mut transaction).await?; transaction.commit().await?; @@ -50,7 +49,7 @@ pub fn create_or_generate_token(record: &WholeVoiceConversionModelRecord) -> Mod pub async fn upsert_model_weights_record( record: &WholeVoiceConversionModelRecord, model_weight_token: &ModelWeightToken, - copied_data: &CopiedFileData, + maybe_copied_data: Option<&CopiedFileData>, mut transaction: &mut Transaction<'_, MySql>, ) -> AnyhowResult<()> { From 44d75baeec4bbaee2bb0d770ef27911755c0a252 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 19 Dec 2023 22:26:04 -0500 Subject: [PATCH 2/2] update old record with token --- Cargo.lock | 1 + crates/cli/migration-tool/Cargo.toml | 1 + .../copy_cloud_files.rs | 8 ++++++++ .../voice_conversion_to_weights/migrate.rs | 5 ++++- ...model_weight_from_voice_conversion_model.rs | 18 +++++++++++++++--- 5 files changed, 29 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 840d827860..4fd0695c9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3519,6 +3519,7 @@ dependencies = [ "easyenv", "enums", "errors", + "filesys", "hashing", "log", "my-workspace-hack", diff --git a/crates/cli/migration-tool/Cargo.toml b/crates/cli/migration-tool/Cargo.toml index 26e243858e..bf2cc11e90 100644 --- a/crates/cli/migration-tool/Cargo.toml +++ b/crates/cli/migration-tool/Cargo.toml @@ -20,6 +20,7 @@ config = { path = "../../lib/deprecated/config" } easyenv = { path = "../../lib/easyenv" } enums = { path = "../../_types/enums" } errors = { path = "../../lib/errors" } +filesys = { path = "../../lib/files/filesys" } hashing = { path = "../../lib/files/hashing" } mysql_queries = { path = "../../lib/mysql_queries" } storyteller_root = { path = "../../lib/storyteller_root" } diff --git a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs index 9d5f3468de..ebc35303ce 100644 --- a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs +++ b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/copy_cloud_files.rs @@ -4,6 +4,8 @@ use buckets::public::weight_files::bucket_file_path::WeightFileBucketPath; use cloud_storage::bucket_path_unifier::BucketPathUnifier; use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType; use errors::{anyhow, AnyhowResult}; +use filesys::safe_delete_temp_directory::safe_delete_temp_directory; +use filesys::safe_delete_temp_file::safe_delete_temp_file; use hashing::sha256::sha256_hash_file::sha256_hash_file; use mysql_queries::queries::model_weights::migration::upsert_model_weight_from_voice_conversion_model::CopiedFileData; use mysql_queries::queries::voice_conversion::migration::list_whole_voice_conversion_models_using_cursor::WholeVoiceConversionModelRecord; @@ -48,6 +50,9 @@ async fn copy_model(model: &WholeVoiceConversionModelRecord, deps: &Deps) -> Any &model_temp_fs_path, "application/octet-stream").await?; + safe_delete_temp_file(&model_temp_fs_path); + safe_delete_temp_directory(&temp_dir); + Ok(CopiedFileData { bucket_path: new_model_bucket_path, file_sha_hash: file_checksum, @@ -73,5 +78,8 @@ async fn copy_index_file(model: &WholeVoiceConversionModelRecord, deps: &Deps, b &model_temp_fs_path, "application/octet-stream").await?; + safe_delete_temp_file(&model_temp_fs_path); + safe_delete_temp_directory(&temp_dir); + Ok(()) // NB: We don't care about the path of the index file. } diff --git a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs index 4263697209..6c9beacc14 100644 --- a/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs +++ b/crates/cli/migration-tool/src/migrations/voice_conversion_to_weights/migrate.rs @@ -61,7 +61,10 @@ async fn copy_model_record_and_bucket_files(record: &WholeVoiceConversionModelRe } } - upsert_model_weight_from_voice_conversion_model(record, &deps.mysql_production, maybe_copied_data.as_ref()).await?; + if let Some(copied_data) = &maybe_copied_data { + info!("Upserting records..."); + upsert_model_weight_from_voice_conversion_model(record, &deps.mysql_production, copied_data).await?; + } Ok(()) } diff --git a/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs b/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs index bbcc71323e..76799a629c 100644 --- a/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs +++ b/crates/lib/mysql_queries/src/queries/model_weights/migration/upsert_model_weight_from_voice_conversion_model.rs @@ -22,14 +22,14 @@ pub struct CopiedFileData { pub async fn upsert_model_weight_from_voice_conversion_model( record: &WholeVoiceConversionModelRecord, mysql_pool: &MySqlPool, - maybe_copied_data: Option<&CopiedFileData>, + copied_data: &CopiedFileData, ) -> AnyhowResult<()> { let mut transaction = mysql_pool.begin().await?; let model_weight_token = create_or_generate_token(record); - upsert_model_weights_record(record, &model_weight_token, maybe_copied_data, &mut transaction).await?; + upsert_model_weights_record(record, &model_weight_token, copied_data, &mut transaction).await?; upsert_model_weights_extension_record(record, &model_weight_token, &mut transaction).await?; update_original_record(record, &model_weight_token, &mut transaction).await?; @@ -49,7 +49,7 @@ pub fn create_or_generate_token(record: &WholeVoiceConversionModelRecord) -> Mod pub async fn upsert_model_weights_record( record: &WholeVoiceConversionModelRecord, model_weight_token: &ModelWeightToken, - maybe_copied_data: Option<&CopiedFileData>, + copied_data: &CopiedFileData, mut transaction: &mut Transaction<'_, MySql>, ) -> AnyhowResult<()> { @@ -226,6 +226,18 @@ pub async fn update_original_record( model_weight_token: &ModelWeightToken, transaction: &mut Transaction<'_, MySql> ) -> AnyhowResult<()> { + let query = sqlx::query!( + r#" +UPDATE voice_conversion_models +SET + maybe_migration_new_model_weights_token = ? +WHERE token = ? + "#, + model_weight_token, + record.token, + ); + + let _r = query.execute(&mut **transaction).await?; Ok(()) }