From 807c8fd4f8a241fa34ebe31c699db25f4c0a70b4 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Fri, 5 Jul 2024 15:23:00 -0400 Subject: [PATCH] add progress estimation to job status endpoints (this is incomplete) (#353) This adds a progress indicator to the three job status endpoints. There's a function to calculate progress, which is mostly left as a stub for now. In the future we may even want to do dynamically updating averages. --- ...040a50ff2942224dc70b0faa0ec988aee7cf.json} | 17 +++- .../test_search_media_files_documents.rs | 2 + .../web/batch_get_inference_job_status.rs | 82 ++-------------- .../web/get_inference_job_status.rs | 84 ++--------------- .../generic_inference/web/job_status.rs | 85 +++++++++++++++++ .../web/list_session_jobs.rs | 82 ++-------------- .../src/queries/generic_inference/web/mod.rs | 1 + .../batch_get_inference_job_status_handler.rs | 12 ++- .../get_inference_job_status_handler.rs | 12 ++- .../list_session_jobs_handler.rs | 13 ++- .../endpoints/inference_job/mod.rs | 3 +- .../utils/estimate_job_progress.rs | 94 +++++++++++++++++++ .../endpoints/inference_job/utils/mod.rs | 1 + .../helpers/get_scoped_engine_categories.rs | 2 +- .../helpers/get_scoped_media_classes.rs | 2 +- .../helpers/get_scoped_media_types.rs | 2 +- 16 files changed, 259 insertions(+), 235 deletions(-) rename .sqlx/{query-7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d.json => query-e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf.json} (90%) create mode 100644 crates/lib/mysql_queries/src/queries/generic_inference/web/job_status.rs create mode 100644 crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/estimate_job_progress.rs create mode 100644 crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/mod.rs diff --git a/.sqlx/query-7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d.json b/.sqlx/query-e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf.json similarity index 90% rename from .sqlx/query-7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d.json rename to .sqlx/query-e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf.json index 322567181b..fd60892c18 100644 --- a/.sqlx/query-7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d.json +++ b/.sqlx/query-e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf.json @@ -1,6 +1,6 @@ { "db_name": "MySQL", - "query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n\n jobs.status as `status: enums::common::job_status_plus::JobStatusPlus`,\n jobs.attempt_count,\n\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n\n jobs.inference_category as `inference_category: enums::by_table::generic_inference_jobs::inference_category::InferenceCategory`,\n jobs.maybe_model_type,\n jobs.maybe_model_token,\n jobs.maybe_raw_inference_text,\n jobs.maybe_inference_args,\n\n jobs.frontend_failure_category as `maybe_frontend_failure_category: enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory`,\n\n jobs.on_success_result_entity_type as maybe_result_entity_type,\n jobs.on_success_result_entity_token as maybe_result_entity_token,\n\n model_weights.title as maybe_model_weights_title,\n tts_models.title as maybe_tts_model_title,\n voice_conversion_models.title as maybe_voice_conversion_model_title,\n\n tts_results.public_bucket_wav_audio_path as maybe_tts_public_bucket_path,\n voice_conversion_results.public_bucket_hash as maybe_voice_conversion_public_bucket_hash,\n\n media_files.public_bucket_directory_hash as maybe_media_file_public_bucket_directory_hash,\n media_files.maybe_public_bucket_prefix as maybe_media_file_public_bucket_prefix,\n media_files.maybe_public_bucket_extension as maybe_media_file_public_bucket_extension,\n\n jobs.assigned_worker as maybe_assigned_worker,\n jobs.assigned_cluster as maybe_assigned_cluster,\n\n jobs.is_keepalive_required,\n\n jobs.created_at,\n jobs.updated_at,\n\n jobs.first_started_at as maybe_first_started_at,\n jobs.successfully_completed_at as maybe_successfully_completed_at\n\nFROM generic_inference_jobs as jobs\n\nLEFT OUTER JOIN model_weights ON jobs.maybe_model_token = model_weights.token\nLEFT OUTER JOIN tts_models ON jobs.maybe_model_token = tts_models.token\nLEFT OUTER JOIN voice_conversion_models ON jobs.maybe_model_token = voice_conversion_models.token\n\nLEFT OUTER JOIN tts_results ON jobs.on_success_result_entity_token = tts_results.token\nLEFT OUTER JOIN voice_conversion_results ON jobs.on_success_result_entity_token = voice_conversion_results.token\nLEFT OUTER JOIN media_files ON jobs.on_success_result_entity_token = media_files.token\n\nWHERE jobs.token = ?\n ", + "query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n\n jobs.status as `status: enums::common::job_status_plus::JobStatusPlus`,\n jobs.attempt_count,\n\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n\n jobs.inference_category as `inference_category: enums::by_table::generic_inference_jobs::inference_category::InferenceCategory`,\n jobs.maybe_model_type,\n jobs.maybe_model_token,\n jobs.maybe_raw_inference_text,\n jobs.maybe_inference_args,\n\n jobs.frontend_failure_category as `maybe_frontend_failure_category: enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory`,\n\n jobs.on_success_result_entity_type as maybe_result_entity_type,\n jobs.on_success_result_entity_token as maybe_result_entity_token,\n\n model_weights.title as maybe_model_weights_title,\n tts_models.title as maybe_tts_model_title,\n voice_conversion_models.title as maybe_voice_conversion_model_title,\n\n tts_results.public_bucket_wav_audio_path as maybe_tts_public_bucket_path,\n voice_conversion_results.public_bucket_hash as maybe_voice_conversion_public_bucket_hash,\n\n media_files.public_bucket_directory_hash as maybe_media_file_public_bucket_directory_hash,\n media_files.maybe_public_bucket_prefix as maybe_media_file_public_bucket_prefix,\n media_files.maybe_public_bucket_extension as maybe_media_file_public_bucket_extension,\n\n jobs.assigned_worker as maybe_assigned_worker,\n jobs.assigned_cluster as maybe_assigned_cluster,\n\n jobs.is_keepalive_required,\n\n jobs.created_at,\n jobs.updated_at,\n\n jobs.first_started_at as maybe_first_started_at,\n jobs.successfully_completed_at as maybe_successfully_completed_at,\n\n NOW() as database_clock\n\nFROM generic_inference_jobs as jobs\n\nLEFT OUTER JOIN model_weights ON jobs.maybe_model_token = model_weights.token\nLEFT OUTER JOIN tts_models ON jobs.maybe_model_token = tts_models.token\nLEFT OUTER JOIN voice_conversion_models ON jobs.maybe_model_token = voice_conversion_models.token\n\nLEFT OUTER JOIN tts_results ON jobs.on_success_result_entity_token = tts_results.token\nLEFT OUTER JOIN voice_conversion_results ON jobs.on_success_result_entity_token = voice_conversion_results.token\nLEFT OUTER JOIN media_files ON jobs.on_success_result_entity_token = media_files.token\n\nWHERE jobs.token = ?\n ", "describe": { "columns": [ { @@ -292,6 +292,16 @@ "char_set": 63, "max_size": 19 } + }, + { + "ordinal": 29, + "name": "database_clock", + "type_info": { + "type": "Datetime", + "flags": "NOT_NULL | BINARY", + "char_set": 63, + "max_size": 19 + } } ], "parameters": { @@ -326,8 +336,9 @@ false, false, true, - true + true, + false ] }, - "hash": "7d5030b221e9b470ee7af4b73539fd2ef477e6f3c98c56422016449c6775222d" + "hash": "e5121d132d93ec43e48197e00b19040a50ff2942224dc70b0faa0ec988aee7cf" } diff --git a/crates/cli/production/elasticsearch-cli/src/plans/media_files/test_search_media_files_documents.rs b/crates/cli/production/elasticsearch-cli/src/plans/media_files/test_search_media_files_documents.rs index 3c11b92fbb..9315a97fbb 100644 --- a/crates/cli/production/elasticsearch-cli/src/plans/media_files/test_search_media_files_documents.rs +++ b/crates/cli/production/elasticsearch-cli/src/plans/media_files/test_search_media_files_documents.rs @@ -11,6 +11,8 @@ pub async fn test_search_media_files(client: &Elasticsearch) -> AnyhowResult<()> let results = search_media_files(SearchArgs { search_term: "cac", + is_featured: None, + maybe_creator_user_token: None, maybe_media_classes: Some(HashSet::from_iter(vec![ MediaFileClass::Dimensional, MediaFileClass::Image, diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/batch_get_inference_job_status.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/batch_get_inference_job_status.rs index 621f42e8a7..f592bd9b4d 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/web/batch_get_inference_job_status.rs +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/batch_get_inference_job_status.rs @@ -7,7 +7,6 @@ use sqlx::pool::PoolConnection; use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; use enums::common::job_status_plus::JobStatusPlus; -use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; use enums::traits::mysql_from_row::MySqlFromRow; use errors::AnyhowResult; use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; @@ -17,78 +16,7 @@ use tokens::traits::mysql_token_from_row::MySqlTokenFromRow; use crate::helpers::boolean_converters::i8_to_bool; use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs}; - -// NB: Serialization is for internal endpoints, not for returning to end-users. -#[derive(Debug, Default, Serialize)] -pub struct GenericInferenceJobStatus { - pub job_token: InferenceJobToken, - - pub status: JobStatusPlus, - pub attempt_count: u16, - - pub maybe_assigned_worker: Option, - pub maybe_assigned_cluster: Option, - - pub maybe_first_started_at: Option>, - - pub maybe_frontend_failure_category: Option, - - pub request_details: RequestDetails, - pub maybe_result_details: Option, - pub user_details: UserDetails, - - pub is_keepalive_required: bool, - - pub created_at: DateTime, - pub updated_at: DateTime, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the user's original inference request -/// (We may want to present it in the "pending" UI.) -#[derive(Debug, Default, Serialize)] -pub struct RequestDetails { - pub inference_category: InferenceCategory, - pub maybe_model_type: Option, // TODO: Strongly type - pub maybe_model_token: Option, - pub maybe_model_title: Option, - - /// TTS input. In the future, perhaps voice conversion SST - pub maybe_raw_inference_text: Option, - - /// For Comfy / Video Style Transfer jobs, this might include - /// the name of the selected style. - pub maybe_style_name: Option, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the generated result -#[derive(Debug, Default, Serialize)] -pub struct ResultDetails { - pub entity_type: String, - pub entity_token: String, - - /// The bucket storage hash (for vc and media_files) or full path (for tts) - pub public_bucket_location_or_hash: String, - pub maybe_media_file_public_bucket_prefix: Option, - pub maybe_media_file_public_bucket_extension: Option, - - /// Whether the location is a full path (for tts) or a hash (for vc) that - /// needs to be reconstructed into a path. - pub public_bucket_location_is_hash: bool, - - pub maybe_successfully_completed_at: Option>, -} - -/// NB: DO NOT EXPOSE TO FRONTEND. -/// NB: Serialize is for internal moderator-only endpoints -/// This is used to gate access to job termination -#[derive(Debug, Default, Serialize)] -pub struct UserDetails { - pub maybe_creator_user_token: Option, - pub maybe_creator_anonymous_visitor_token: Option, - pub creator_ip_address: String, -} +use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails}; /// Look up job status. /// Returns Ok(Vec::new()) when the records cannot be found. @@ -187,7 +115,9 @@ SELECT jobs.updated_at, jobs.first_started_at as maybe_first_started_at, - jobs.successfully_completed_at as maybe_successfully_completed_at + jobs.successfully_completed_at as maybe_successfully_completed_at, + + NOW() as database_clock FROM generic_inference_jobs as jobs @@ -313,6 +243,7 @@ fn raw_records_to_public_result(records: Vec) -> V is_keepalive_required: i8_to_bool(record.is_keepalive_required), created_at: record.created_at, updated_at: record.updated_at, + database_clock: record.database_clock, } }) .collect::>() @@ -360,6 +291,8 @@ struct RawGenericInferenceJobStatus { pub maybe_first_started_at: Option>, pub maybe_successfully_completed_at: Option>, + + pub database_clock: DateTime, } // NB(bt,2023-12-05): There's an issue with type hinting in the `as` clauses with QueryBuilder (or @@ -406,6 +339,7 @@ impl FromRow<'_, MySqlRow> for RawGenericInferenceJobStatus { maybe_successfully_completed_at: row.try_get("maybe_successfully_completed_at")?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, + database_clock: row.try_get("database_clock")?, }) } } diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs index d0a355c5b0..4f435e91cf 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs @@ -1,5 +1,5 @@ use anyhow::anyhow; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, NaiveDateTime, TimeZone, Utc}; use log::warn; use sqlx::{MySql, MySqlPool}; use sqlx::pool::PoolConnection; @@ -15,78 +15,7 @@ use tokens::tokens::users::UserToken; use crate::helpers::boolean_converters::i8_to_bool; use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs}; - -// NB: Serialization is for internal endpoints, not for returning to end-users. -#[derive(Debug, Default, Serialize)] -pub struct GenericInferenceJobStatus { - pub job_token: InferenceJobToken, - - pub status: JobStatusPlus, - pub attempt_count: u16, - - pub maybe_assigned_worker: Option, - pub maybe_assigned_cluster: Option, - - pub maybe_first_started_at: Option>, - - pub maybe_frontend_failure_category: Option, - - pub request_details: RequestDetails, - pub maybe_result_details: Option, - pub user_details: UserDetails, - - pub is_keepalive_required: bool, - - pub created_at: DateTime, - pub updated_at: DateTime, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the user's original inference request -/// (We may want to present it in the "pending" UI.) -#[derive(Debug, Default, Serialize)] -pub struct RequestDetails { - pub inference_category: InferenceCategory, - pub maybe_model_type: Option, // TODO: Strongly type - pub maybe_model_token: Option, - pub maybe_model_title: Option, - - /// TTS input. In the future, perhaps voice conversion SST - pub maybe_raw_inference_text: Option, - - /// For Comfy / Video Style Transfer jobs, this might include - /// the name of the selected style. - pub maybe_style_name: Option, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the generated result -#[derive(Debug, Default, Serialize)] -pub struct ResultDetails { - pub entity_type: String, - pub entity_token: String, - - /// The bucket storage hash (for vc and media_files) or full path (for tts) - pub public_bucket_location_or_hash: String, - pub maybe_media_file_public_bucket_prefix: Option, - pub maybe_media_file_public_bucket_extension: Option, - - /// Whether the location is a full path (for tts) or a hash (for vc) that - /// needs to be reconstructed into a path. - pub public_bucket_location_is_hash: bool, - - pub maybe_successfully_completed_at: Option>, -} - -/// NB: DO NOT EXPOSE TO FRONTEND. -/// NB: Serialize is for internal moderator-only endpoints -/// This is used to gate access to job termination -#[derive(Debug, Default, Serialize)] -pub struct UserDetails { - pub maybe_creator_user_token: Option, - pub maybe_creator_anonymous_visitor_token: Option, - pub creator_ip_address: String, -} +use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails}; /// Look up job status. /// Returns Ok(None) when the record cannot be found. @@ -151,7 +80,9 @@ SELECT jobs.updated_at, jobs.first_started_at as maybe_first_started_at, - jobs.successfully_completed_at as maybe_successfully_completed_at + jobs.successfully_completed_at as maybe_successfully_completed_at, + + NOW() as database_clock FROM generic_inference_jobs as jobs @@ -294,6 +225,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI is_keepalive_required: i8_to_bool(record.is_keepalive_required), created_at: record.created_at, updated_at: record.updated_at, + database_clock: DateTime::from_naive_utc_and_offset(record.database_clock, Utc), } } @@ -339,6 +271,10 @@ struct RawGenericInferenceJobStatus { pub maybe_first_started_at: Option>, pub maybe_successfully_completed_at: Option>, + + // NB: Typed query can't convert to Utc, so we use NaiveDateTime and do the type conversion ourselves. + // The database server *should* be reporting in UTC. + pub database_clock: NaiveDateTime, } #[cfg(test)] diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/job_status.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/job_status.rs new file mode 100644 index 0000000000..e87bb67648 --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/job_status.rs @@ -0,0 +1,85 @@ +use chrono::{DateTime, Utc}; + +use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::common::job_status_plus::JobStatusPlus; +use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; +use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::users::UserToken; + +/// Shared inference job status struct for several queries (list, get, batch get) +/// NB: Serialization is for internal endpoints, not for returning to end-users. +#[derive(Debug, Default, Serialize)] +pub struct GenericInferenceJobStatus { + pub job_token: InferenceJobToken, + + pub status: JobStatusPlus, + pub attempt_count: u16, + + pub maybe_assigned_worker: Option, + pub maybe_assigned_cluster: Option, + + pub maybe_first_started_at: Option>, + + pub maybe_frontend_failure_category: Option, + + pub request_details: RequestDetails, + pub maybe_result_details: Option, + pub user_details: UserDetails, + + pub is_keepalive_required: bool, + + pub created_at: DateTime, + pub updated_at: DateTime, + + /// This is MySQL's NOW(), which will let us do duration math without clock skew. + pub database_clock: DateTime, +} + +/// NB: Serialize is for internal moderator-only endpoints +/// Details about the user's original inference request +/// (We may want to present it in the "pending" UI.) +#[derive(Debug, Default, Serialize)] +pub struct RequestDetails { + pub inference_category: InferenceCategory, + pub maybe_model_type: Option, // TODO: Strongly type + pub maybe_model_token: Option, + pub maybe_model_title: Option, + + /// TTS input. In the future, perhaps voice conversion SST + pub maybe_raw_inference_text: Option, + + /// For Comfy / Video Style Transfer jobs, this might include + /// the name of the selected style. + pub maybe_style_name: Option, +} + +/// NB: Serialize is for internal moderator-only endpoints +/// Details about the generated result +#[derive(Debug, Default, Serialize)] +pub struct ResultDetails { + pub entity_type: String, + pub entity_token: String, + + /// The bucket storage hash (for vc and media_files) or full path (for tts) + pub public_bucket_location_or_hash: String, + pub maybe_media_file_public_bucket_prefix: Option, + pub maybe_media_file_public_bucket_extension: Option, + + /// Whether the location is a full path (for tts) or a hash (for vc) that + /// needs to be reconstructed into a path. + pub public_bucket_location_is_hash: bool, + + pub maybe_successfully_completed_at: Option>, +} + +/// NB: DO NOT EXPOSE TO FRONTEND. +/// NB: Serialize is for internal moderator-only endpoints +/// This is used to gate access to job termination +#[derive(Debug, Default, Serialize)] +pub struct UserDetails { + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + pub creator_ip_address: String, +} diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/list_session_jobs.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/list_session_jobs.rs index 21cc634155..93ee11027e 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/web/list_session_jobs.rs +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/list_session_jobs.rs @@ -9,7 +9,6 @@ use sqlx::pool::PoolConnection; use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; use enums::common::job_status_plus::JobStatusPlus; -use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; use enums::traits::mysql_from_row::MySqlFromRow; use errors::AnyhowResult; use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; @@ -19,78 +18,7 @@ use tokens::traits::mysql_token_from_row::MySqlTokenFromRow; use crate::helpers::boolean_converters::i8_to_bool; use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs}; - -// NB: Serialization is for internal endpoints, not for returning to end-users. -#[derive(Debug, Default, Serialize)] -pub struct GenericInferenceJobStatus { - pub job_token: InferenceJobToken, - - pub status: JobStatusPlus, - pub attempt_count: u16, - - pub maybe_assigned_worker: Option, - pub maybe_assigned_cluster: Option, - - pub maybe_first_started_at: Option>, - - pub maybe_frontend_failure_category: Option, - - pub request_details: RequestDetails, - pub maybe_result_details: Option, - pub user_details: UserDetails, - - pub is_keepalive_required: bool, - - pub created_at: DateTime, - pub updated_at: DateTime, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the user's original inference request -/// (We may want to present it in the "pending" UI.) -#[derive(Debug, Default, Serialize)] -pub struct RequestDetails { - pub inference_category: InferenceCategory, - pub maybe_model_type: Option, // TODO: Strongly type - pub maybe_model_token: Option, - pub maybe_model_title: Option, - - /// TTS input. In the future, perhaps voice conversion SST - pub maybe_raw_inference_text: Option, - - /// For Comfy / Video Style Transfer jobs, this might include - /// the name of the selected style. - pub maybe_style_name: Option, -} - -/// NB: Serialize is for internal moderator-only endpoints -/// Details about the generated result -#[derive(Debug, Default, Serialize)] -pub struct ResultDetails { - pub entity_type: String, - pub entity_token: String, - - /// The bucket storage hash (for vc and media_files) or full path (for tts) - pub public_bucket_location_or_hash: String, - pub maybe_media_file_public_bucket_prefix: Option, - pub maybe_media_file_public_bucket_extension: Option, - - /// Whether the location is a full path (for tts) or a hash (for vc) that - /// needs to be reconstructed into a path. - pub public_bucket_location_is_hash: bool, - - pub maybe_successfully_completed_at: Option>, -} - -/// NB: DO NOT EXPOSE TO FRONTEND. -/// NB: Serialize is for internal moderator-only endpoints -/// This is used to gate access to job termination -#[derive(Debug, Default, Serialize)] -pub struct UserDetails { - pub maybe_creator_user_token: Option, - pub maybe_creator_anonymous_visitor_token: Option, - pub creator_ip_address: String, -} +use crate::queries::generic_inference::web::job_status::{GenericInferenceJobStatus, RequestDetails, ResultDetails, UserDetails}; pub struct ListSessionJobsForUserArgs<'a> { // TODO(bt,2024-04-23): Support AnonymousVisitorToken @@ -239,7 +167,9 @@ SELECT jobs.updated_at, jobs.first_started_at as maybe_first_started_at, - jobs.successfully_completed_at as maybe_successfully_completed_at + jobs.successfully_completed_at as maybe_successfully_completed_at, + + NOW() as database_clock FROM generic_inference_jobs as jobs @@ -365,6 +295,7 @@ fn raw_records_to_public_result(records: Vec) -> V is_keepalive_required: i8_to_bool(record.is_keepalive_required), created_at: record.created_at, updated_at: record.updated_at, + database_clock: record.database_clock, } }) .collect::>() @@ -412,6 +343,8 @@ struct RawGenericInferenceJobStatus { pub maybe_first_started_at: Option>, pub maybe_successfully_completed_at: Option>, + + pub database_clock: DateTime, } // NB(bt,2023-12-05): There's an issue with type hinting in the `as` clauses with QueryBuilder (or @@ -458,6 +391,7 @@ impl FromRow<'_, MySqlRow> for RawGenericInferenceJobStatus { maybe_successfully_completed_at: row.try_get("maybe_successfully_completed_at")?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, + database_clock: row.try_get("database_clock")?, }) } } diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/mod.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/mod.rs index 63efd4b7aa..4daa07efdd 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/web/mod.rs +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/mod.rs @@ -3,6 +3,7 @@ pub mod dismiss_finished_jobs_for_user; pub mod get_inference_job_status; pub mod get_pending_inference_job_count; pub mod insert_generic_inference_job; +pub mod job_status; pub mod kill_generic_inference_jobs; pub mod list_session_jobs; pub mod mark_generic_inference_job_cancelled_by_user; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/batch_get_inference_job_status_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/batch_get_inference_job_status_handler.rs index cc6917e2fb..33c02e23a8 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/batch_get_inference_job_status_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/batch_get_inference_job_status_handler.rs @@ -18,11 +18,13 @@ use enums::by_table::generic_inference_jobs::frontend_failure_category::Frontend use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; use enums::common::job_status_plus::JobStatusPlus; use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; -use mysql_queries::queries::generic_inference::web::batch_get_inference_job_status::{batch_get_inference_job_status, GenericInferenceJobStatus}; +use mysql_queries::queries::generic_inference::web::batch_get_inference_job_status::batch_get_inference_job_status; +use mysql_queries::queries::generic_inference::web::job_status::GenericInferenceJobStatus; use redis_common::redis_keys::RedisKeys; use tokens::tokens::generic_inference_jobs::InferenceJobToken; use tokens::tokens::media_files::MediaFileToken; +use crate::http_server::endpoints::inference_job::utils::estimate_job_progress::estimate_job_progress; use crate::http_server::endpoints::media_files::get::batch_get_media_files_handler::BatchGetMediaFilesQueryParams; use crate::http_server::responses::filter_model_name::maybe_filter_model_name; use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; @@ -106,7 +108,10 @@ pub struct BatchStatusDetailsResponse { /// An enum the frontend can use to display localized/I18N error /// messages. These pertain to both transient and permanent failures. - pub maybe_failure_category: Option + pub maybe_failure_category: Option, + + /// This is an integer number between 0 and 100 that reports the completeness. + pub progress_percentage: u8, } /// Details about the completed result (if any) @@ -256,6 +261,8 @@ fn db_record_to_response_payload( ) -> BatchInferenceJobStatusResponsePayload { let inference_category = record.request_details.inference_category; + let progress_percentage = estimate_job_progress(&record); + BatchInferenceJobStatusResponsePayload { job_token: record.job_token, request: BatchRequestDetailsResponse { @@ -275,6 +282,7 @@ fn db_record_to_response_payload( attempt_count: record.attempt_count as u8, requires_keepalive: record.is_keepalive_required, maybe_failure_category: record.maybe_frontend_failure_category, + progress_percentage, }, maybe_result: record.maybe_result_details.map(|result_details| { // NB: Be careful here, because this varies based on the type of inference result. diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status_handler.rs index 9eb96cb1f9..79af386148 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status_handler.rs @@ -16,10 +16,12 @@ use enums::by_table::generic_inference_jobs::frontend_failure_category::Frontend use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; use enums::common::job_status_plus::JobStatusPlus; use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; -use mysql_queries::queries::generic_inference::web::get_inference_job_status::{GenericInferenceJobStatus, get_inference_job_status}; +use mysql_queries::queries::generic_inference::web::get_inference_job_status::get_inference_job_status; +use mysql_queries::queries::generic_inference::web::job_status::GenericInferenceJobStatus; use redis_common::redis_keys::RedisKeys; use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use crate::http_server::endpoints::inference_job::utils::estimate_job_progress::estimate_job_progress; use crate::http_server::responses::filter_model_name::maybe_filter_model_name; use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; use crate::server_state::ServerState; @@ -94,7 +96,10 @@ pub struct StatusDetailsResponse { /// An enum the frontend can use to display localized/I18N error /// messages. These pertain to both transient and permanent failures. - pub maybe_failure_category: Option + pub maybe_failure_category: Option, + + /// This is an integer number between 0 and 100 that reports the completeness. + pub progress_percentage: u8, } /// Details about the completed result (if any) @@ -247,6 +252,8 @@ fn record_to_payload( ) -> InferenceJobStatusResponsePayload { let inference_category = record.request_details.inference_category; + let progress_percentage = estimate_job_progress(&record); + InferenceJobStatusResponsePayload { job_token: record.job_token, request: RequestDetailsResponse { @@ -266,6 +273,7 @@ fn record_to_payload( attempt_count: record.attempt_count as u8, requires_keepalive: record.is_keepalive_required, maybe_failure_category: record.maybe_frontend_failure_category, + progress_percentage, }, maybe_result: record.maybe_result_details.map(|result_details| { // NB: Be careful here, because this varies based on the type of inference result. diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/list_session_jobs_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/list_session_jobs_handler.rs index 038437cd85..9824a557ca 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/list_session_jobs_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/list_session_jobs_handler.rs @@ -18,12 +18,14 @@ use enums::by_table::generic_inference_jobs::frontend_failure_category::Frontend use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; use enums::common::job_status_plus::JobStatusPlus; use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; -use mysql_queries::queries::generic_inference::web::list_session_jobs::{GenericInferenceJobStatus, list_session_jobs, list_session_jobs_from_connection, ListSessionJobsForUserArgs}; +use mysql_queries::queries::generic_inference::web::job_status::GenericInferenceJobStatus; +use mysql_queries::queries::generic_inference::web::list_session_jobs::{list_session_jobs_from_connection, ListSessionJobsForUserArgs}; use redis_common::redis_keys::RedisKeys; use tokens::tokens::generic_inference_jobs::InferenceJobToken; use tokens::tokens::media_files::MediaFileToken; use tokens::tokens::users::UserToken; +use crate::http_server::endpoints::inference_job::utils::estimate_job_progress::estimate_job_progress; use crate::http_server::responses::filter_model_name::maybe_filter_model_name; use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; use crate::server_state::ServerState; @@ -110,7 +112,10 @@ pub struct ListSessionStatusDetailsResponse { /// An enum the frontend can use to display localized/I18N error /// messages. These pertain to both transient and permanent failures. - pub maybe_failure_category: Option + pub maybe_failure_category: Option, + + /// This is an integer number between 0 and 100 that reports the completeness. + pub progress_percentage: u8, } /// Details about the completed result (if any) @@ -312,6 +317,8 @@ fn db_record_to_response_payload( ) -> ListSessionJobsItem { let inference_category = record.request_details.inference_category; + let progress_percentage = estimate_job_progress(&record); + ListSessionJobsItem { job_token: record.job_token, request: ListSessionRequestDetailsResponse { @@ -331,6 +338,7 @@ fn db_record_to_response_payload( attempt_count: record.attempt_count as u8, requires_keepalive: record.is_keepalive_required, maybe_failure_category: record.maybe_frontend_failure_category, + progress_percentage, }, maybe_result: record.maybe_result_details.map(|result_details| { // NB: Be careful here, because this varies based on the type of inference result. @@ -427,3 +435,4 @@ fn db_record_to_response_payload( updated_at: record.updated_at, } } + diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/mod.rs index 0bcb5d539c..03ca033562 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/mod.rs @@ -4,4 +4,5 @@ pub mod get_inference_job_status_handler; pub mod get_pending_inference_job_count_handler; pub mod kill_inference_jobs_handler; pub mod list_session_jobs_handler; -pub mod terminate_inference_job_handler; \ No newline at end of file +pub mod terminate_inference_job_handler; +pub mod utils; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/estimate_job_progress.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/estimate_job_progress.rs new file mode 100644 index 0000000000..75dfec7a2e --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/estimate_job_progress.rs @@ -0,0 +1,94 @@ +use std::cmp::min; + +use chrono::Utc; + +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::common::job_status_plus::JobStatusPlus; +use mysql_queries::queries::generic_inference::web::job_status::GenericInferenceJobStatus; + +// TODO: These numbers are made up. We should measure the average job durations. +const COMFY_JOB_AVERAGE_SECONDS : u64 = 60 * 4; +const LIPSYNC_JOB_AVERAGE_SECONDS : u64 = 60 * 3; +const TTS_JOB_AVERAGE_SECONDS : u64 = 7; +const VC_JOB_AVERAGE_SECONDS : u64 = 90; + +pub fn estimate_job_progress(job: &GenericInferenceJobStatus) -> u8 { + match job.status { + // Jobs that haven't started + JobStatusPlus::Pending + | JobStatusPlus::AttemptFailed => return 0, + // Jobs that are "done" + JobStatusPlus::CompleteSuccess + | JobStatusPlus::CompleteFailure + | JobStatusPlus::Dead + | JobStatusPlus::CancelledByUser + | JobStatusPlus::CancelledBySystem => return 100, + // Fall through + JobStatusPlus::Started => {} + } + + // TODO: This is inaccurate under retries + // "mark_generic_inference_job_pending_and_grab_lock" only sets "maybe_first_started_at" + // for the first execution, not retries. It's not clear that this is a valuable behavior, + // and perhaps the code should be updated to always write every job attempt's start time. + let started_at= job.maybe_first_started_at.unwrap_or(job.created_at); + + let now = job.database_clock; + + if started_at < now { + return 0; // We shouldn't see clock skew unless we read from a DB replica. + } + + let duration = now.signed_duration_since(started_at); + let duration_seconds = duration.num_seconds().abs() as u64; + + // TODO: Estimate functions based on duration for each job type + // This might also take parameters: the job arguments, text length, + // video length (currently unavailable - we'll need to put that in the + // job metadata), etc. + let progress = match job.request_details.inference_category { + InferenceCategory::LipsyncAnimation => percent(duration_seconds, LIPSYNC_JOB_AVERAGE_SECONDS), + InferenceCategory::TextToSpeech => percent(duration_seconds, TTS_JOB_AVERAGE_SECONDS), + InferenceCategory::VoiceConversion => percent(duration_seconds, VC_JOB_AVERAGE_SECONDS), + + // TODO: Better estimate using video duration, params, etc. + InferenceCategory::Workflow => percent(duration_seconds, COMFY_JOB_AVERAGE_SECONDS), + + // NB: We don't run these job types anymore. + InferenceCategory::FormatConversion => 0, + InferenceCategory::Mocap => 0, + InferenceCategory::ImageGeneration => 0, + InferenceCategory::VideoFilter => 0, + InferenceCategory::ConvertBvhToWorkflow => 0, + }; + + // We shouldn't show 100% if the job isn't complete. + min(progress, 95) +} + +fn percent(numerator: u64, denominator: u64) -> u8 { + if denominator == 0 { + return 0; + } + let percent = (numerator as f64 / denominator as f64) * 100.0; + let percent = percent as u8; + + min(percent, 100) +} + +#[cfg(test)] +mod tests { + #[test] + fn test_percent() { + assert_eq!(0, super::percent(0, 0)); + assert_eq!(0, super::percent(0, 100)); + assert_eq!(50, super::percent(50, 100)); + assert_eq!(100, super::percent(100, 100)); + assert_eq!(100, super::percent(500, 100)); + assert_eq!(25, super::percent(1, 4)); + assert_eq!(50, super::percent(2, 4)); + assert_eq!(75, super::percent(3, 4)); + assert_eq!(100, super::percent(4, 4)); + assert_eq!(100, super::percent(5, 4)); + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/mod.rs new file mode 100644 index 0000000000..dd5037d43c --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/utils/mod.rs @@ -0,0 +1 @@ +pub mod estimate_job_progress; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_engine_categories.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_engine_categories.rs index e81e8c38ef..6aa7b48ee4 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_engine_categories.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_engine_categories.rs @@ -30,7 +30,7 @@ mod test { use enums::by_table::media_files::media_file_engine_category::MediaFileEngineCategory; - use crate::http_server::endpoints::media_files::list::helpers::get_scoped_engine_categories::get_scoped_engine_categories; + use crate::http_server::endpoints::media_files::helpers::get_scoped_engine_categories::get_scoped_engine_categories; #[test] fn none() { diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_classes.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_classes.rs index a2307c7591..66f43b6c4f 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_classes.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_classes.rs @@ -30,7 +30,7 @@ mod test { use enums::by_table::media_files::media_file_class::MediaFileClass; - use crate::http_server::endpoints::media_files::list::helpers::get_scoped_media_classes::get_scoped_media_classes; + use crate::http_server::endpoints::media_files::helpers::get_scoped_media_classes::get_scoped_media_classes; #[test] fn none() { diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_types.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_types.rs index d4ef4dccfb..65ad137a79 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_types.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/media_files/helpers/get_scoped_media_types.rs @@ -30,7 +30,7 @@ mod test { use enums::by_table::media_files::media_file_type::MediaFileType; - use crate::http_server::endpoints::media_files::list::helpers::get_scoped_media_types::get_scoped_media_types; + use crate::http_server::endpoints::media_files::helpers::get_scoped_media_types::get_scoped_media_types; #[test] fn none() {