From 01a4dc50b4b33120063c67ec645f3ff45b60af6e Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sat, 16 May 2026 12:16:31 -0400 Subject: [PATCH] job --- .../gmicloud/list_pending_gmicloud_jobs.rs | 98 +++++++++ .../queries/generic_inference/gmicloud/mod.rs | 1 + crates/service/job/gmicloud_job/Cargo.toml | 53 +++++ .../config/gmicloud-job.development.env | 1 + .../endpoints/health_check_handler.rs | 101 ++++++++++ .../src/http_server/endpoints/mod.rs | 1 + .../http_server/http_server_shared_state.rs | 7 + .../job/gmicloud_job/src/http_server/mod.rs | 3 + .../src/http_server/run_http_server.rs | 66 ++++++ .../job/gmicloud_job/src/job_dependencies.rs | 26 +++ crates/service/job/gmicloud_job/src/main.rs | 140 +++++++++++++ .../service/job/gmicloud_job/src/main_loop.rs | 93 +++++++++ .../job/gmicloud_job/src/process_job/mod.rs | 2 + .../src/process_job/process_failed_job.rs | 113 +++++++++++ .../src/process_job/process_successful_job.rs | 189 ++++++++++++++++++ .../src/configs/omni_gen/video_models.rs | 156 +++++++-------- 16 files changed, 972 insertions(+), 78 deletions(-) create mode 100644 crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/list_pending_gmicloud_jobs.rs create mode 100644 crates/service/job/gmicloud_job/Cargo.toml create mode 100644 crates/service/job/gmicloud_job/config/gmicloud-job.development.env create mode 100644 crates/service/job/gmicloud_job/src/http_server/endpoints/health_check_handler.rs create mode 100644 crates/service/job/gmicloud_job/src/http_server/endpoints/mod.rs create mode 100644 crates/service/job/gmicloud_job/src/http_server/http_server_shared_state.rs create mode 100644 crates/service/job/gmicloud_job/src/http_server/mod.rs create mode 100644 crates/service/job/gmicloud_job/src/http_server/run_http_server.rs create mode 100644 crates/service/job/gmicloud_job/src/job_dependencies.rs create mode 100644 crates/service/job/gmicloud_job/src/main.rs create mode 100644 crates/service/job/gmicloud_job/src/main_loop.rs create mode 100644 crates/service/job/gmicloud_job/src/process_job/mod.rs create mode 100644 crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs create mode 100644 crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/list_pending_gmicloud_jobs.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/list_pending_gmicloud_jobs.rs new file mode 100644 index 0000000000..4ec2d8f79a --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/list_pending_gmicloud_jobs.rs @@ -0,0 +1,98 @@ +use anyhow::anyhow; +use log::warn; +use sqlx::MySqlPool; + +use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty; +use enums::common::visibility::Visibility; +use errors::AnyhowResult; +use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::prompts::PromptToken; +use tokens::tokens::users::UserToken; +use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken; + +/// A GmiCloud job that is waiting for a result from the external API. +#[derive(Debug)] +pub struct PendingGmiCloudJob { + pub job_token: InferenceJobToken, + + /// The external request_id from GmiCloud. + pub request_id: String, + + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + pub creator_ip_address: String, + pub creator_set_visibility: Visibility, + + pub maybe_prompt_token: Option, + + pub maybe_wallet_ledger_entry_token: Option, +} + +#[derive(Debug, Default)] +struct RawRecord { + job_token: InferenceJobToken, + request_id: Option, + maybe_creator_user_token: Option, + maybe_creator_anonymous_visitor_token: Option, + creator_ip_address: String, + creator_set_visibility: Visibility, + maybe_prompt_token: Option, + maybe_wallet_ledger_entry_token: Option, +} + +/// Returns all non-terminal GmiCloud jobs that have an associated request_id. +pub async fn list_pending_gmicloud_jobs(pool: &MySqlPool) -> AnyhowResult> { + let records = sqlx::query_as!( + RawRecord, + r#" +SELECT + jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`, + jobs.maybe_external_third_party_id as `request_id`, + jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`, + jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`, + jobs.creator_ip_address, + jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`, + jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`, + jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken` + +FROM generic_inference_jobs as jobs + +WHERE jobs.maybe_external_third_party = ? + AND jobs.status NOT IN ('complete_success', 'complete_failure') + AND jobs.maybe_external_third_party_id IS NOT NULL + +LIMIT 25000 + "#, + InferenceJobExternalThirdParty::GmiCloud.to_str(), + ) + .fetch_all(pool) + .await + .map_err(|err| anyhow!("error querying pending gmicloud jobs: {:?}", err))?; + + let jobs = records + .into_iter() + .filter_map(|record| { + let request_id = match record.request_id { + Some(id) => id, + None => { + warn!("PendingGmiCloudJob has no request_id, skipping"); + return None; + } + }; + + Some(PendingGmiCloudJob { + job_token: record.job_token, + request_id, + maybe_creator_user_token: record.maybe_creator_user_token, + maybe_creator_anonymous_visitor_token: record.maybe_creator_anonymous_visitor_token, + creator_ip_address: record.creator_ip_address, + creator_set_visibility: record.creator_set_visibility, + maybe_prompt_token: record.maybe_prompt_token, + maybe_wallet_ledger_entry_token: record.maybe_wallet_ledger_entry_token, + }) + }) + .collect(); + + Ok(jobs) +} diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/mod.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/mod.rs index ac61f12ef8..c4b4b9faf5 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/mod.rs @@ -1 +1,2 @@ pub mod insert_generic_inference_job_for_gmicloud; +pub mod list_pending_gmicloud_jobs; diff --git a/crates/service/job/gmicloud_job/Cargo.toml b/crates/service/job/gmicloud_job/Cargo.toml new file mode 100644 index 0000000000..4253d843d8 --- /dev/null +++ b/crates/service/job/gmicloud_job/Cargo.toml @@ -0,0 +1,53 @@ +[package] +name = "gmicloud-job" +edition = "2021" +version = "0.0.1" +publish = false + +[[bin]] +name = "gmicloud-job" +path = "src/main.rs" + +[package.metadata.cargo-udeps.ignore] +normal = ["workspace-build-acceleration"] + +[dependencies] +# Workspace hack (faster compile times) +workspace-build-acceleration = { workspace = true } + +# Internal +actix_helpers.workspace = true +bootstrap.workspace = true +bucket_paths.workspace = true +cloud_storage.workspace = true +concurrency.workspace = true +easyenv.workspace = true +enums.workspace = true +errors.workspace = true +gmicloud_client.workspace = true +hashing = { path = "../../../lib/files/hashing" } +jobs_common.workspace = true +mysql_queries.workspace = true +server_environment = { path = "../../../lib/server_environment" } +shared_env_var_config.workspace = true +tokens.workspace = true + +# External +anyhow.workspace = true +log.workspace = true + +sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] } + +# Runtime +tokio = { version = "1.40.0", features = ["macros", "sync"] } + +# HTTP Client +reqwest.workspace = true + +# HTTP Server +actix-web.workspace = true + +# Serialization +serde.workspace = true +serde_derive.workspace = true +serde_json.workspace = true diff --git a/crates/service/job/gmicloud_job/config/gmicloud-job.development.env b/crates/service/job/gmicloud_job/config/gmicloud-job.development.env new file mode 100644 index 0000000000..7c0ff2ef1c --- /dev/null +++ b/crates/service/job/gmicloud_job/config/gmicloud-job.development.env @@ -0,0 +1 @@ +SERVER_ENVIRONMENT='development' diff --git a/crates/service/job/gmicloud_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/gmicloud_job/src/http_server/endpoints/health_check_handler.rs new file mode 100644 index 0000000000..419bcba0cd --- /dev/null +++ b/crates/service/job/gmicloud_job/src/http_server/endpoints/health_check_handler.rs @@ -0,0 +1,101 @@ +use std::sync::Arc; + +use actix_web::error::ResponseError; +use actix_web::http::header::ContentType; +use actix_web::http::StatusCode; +use actix_web::{web, HttpRequest, HttpResponse}; +use log::error; + +use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response; + +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +#[derive(Serialize)] +pub struct HealthCheckResponse { + pub success: bool, + pub is_healthy: bool, + pub consecutive_failure_count: u64, + pub consecutive_success_count: u64, + pub total_failure_count: u64, + pub total_success_count: u64, + pub total_failure_ratio: f32, + pub total_success_ratio: f32, +} + +#[derive(Debug, Serialize)] +pub enum HealthCheckError { + ServerError, +} + +impl ResponseError for HealthCheckError { + fn status_code(&self) -> StatusCode { + match *self { + HealthCheckError::ServerError => StatusCode::INTERNAL_SERVER_ERROR, + } + } + + fn error_response(&self) -> HttpResponse { + error_to_json_http_response(self) + } +} + +impl std::fmt::Display for HealthCheckError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self) + } +} + +pub async fn get_health_check_handler( + _http_request: HttpRequest, + server_state: web::Data>, +) -> Result { + let job_stats = server_state + .job_stats + .get_status() + .map_err(|e| { + error!("Error serving health check status: {:?}", e); + HealthCheckError::ServerError + })?; + + let total_tries = job_stats + .total_failure_count + .saturating_add(job_stats.total_success_count); + + let total_success_ratio = if total_tries > 0 { + (job_stats.total_success_count as f32) / (total_tries as f32) + } else { + 0.0 + }; + + let total_failure_ratio = if total_tries > 0 { + 1.0 - total_success_ratio + } else { + 0.0 + }; + + let is_healthy = + job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold; + + let response = HealthCheckResponse { + success: true, + is_healthy, + consecutive_failure_count: job_stats.consecutive_failure_count, + consecutive_success_count: job_stats.consecutive_success_count, + total_failure_count: job_stats.total_failure_count, + total_success_count: job_stats.total_success_count, + total_failure_ratio, + total_success_ratio, + }; + + let body = serde_json::to_string(&response).map_err(|_e| HealthCheckError::ServerError)?; + + if is_healthy { + Ok(HttpResponse::Ok() + .content_type(ContentType::json()) + .body(body)) + } else { + Ok(HttpResponse::InternalServerError() + .content_type(ContentType::json()) + .body(body)) + } +} diff --git a/crates/service/job/gmicloud_job/src/http_server/endpoints/mod.rs b/crates/service/job/gmicloud_job/src/http_server/endpoints/mod.rs new file mode 100644 index 0000000000..0f408270bf --- /dev/null +++ b/crates/service/job/gmicloud_job/src/http_server/endpoints/mod.rs @@ -0,0 +1 @@ +pub mod health_check_handler; diff --git a/crates/service/job/gmicloud_job/src/http_server/http_server_shared_state.rs b/crates/service/job/gmicloud_job/src/http_server/http_server_shared_state.rs new file mode 100644 index 0000000000..534526580f --- /dev/null +++ b/crates/service/job/gmicloud_job/src/http_server/http_server_shared_state.rs @@ -0,0 +1,7 @@ +use jobs_common::job_stats::JobStats; + +#[derive(Clone)] +pub struct HttpServerSharedState { + pub job_stats: JobStats, + pub consecutive_failure_unhealthy_threshold: u64, +} diff --git a/crates/service/job/gmicloud_job/src/http_server/mod.rs b/crates/service/job/gmicloud_job/src/http_server/mod.rs new file mode 100644 index 0000000000..5d888824cc --- /dev/null +++ b/crates/service/job/gmicloud_job/src/http_server/mod.rs @@ -0,0 +1,3 @@ +pub mod endpoints; +pub mod http_server_shared_state; +pub mod run_http_server; diff --git a/crates/service/job/gmicloud_job/src/http_server/run_http_server.rs b/crates/service/job/gmicloud_job/src/http_server/run_http_server.rs new file mode 100644 index 0000000000..1ae97a6eb1 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/http_server/run_http_server.rs @@ -0,0 +1,66 @@ +use std::sync::Arc; + +use actix_web::dev::Server; +use actix_web::middleware::Logger; +use actix_web::{web, App, HttpResponse, HttpServer}; +use log::info; + +use bootstrap::bootstrap::ContainerEnvironment; +use errors::AnyhowResult; +use jobs_common::job_stats::JobStats; + +use crate::http_server::endpoints::health_check_handler::get_health_check_handler; +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:12345"; +const DEFAULT_NUM_WORKERS: usize = 2; + +pub struct CreateServerArgs { + pub container_environment: ContainerEnvironment, + pub job_stats: JobStats, +} + +pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult { + let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS); + let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?; + + let server_state = HttpServerSharedState { + job_stats: args.job_stats.clone(), + consecutive_failure_unhealthy_threshold: easyenv::get_env_num( + "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", + 3, + )?, + }; + + let server_state_arc = web::Data::new(Arc::new(server_state)); + + info!("Starting HTTP service (for k8s health checking)."); + + let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; + + let handle = HttpServer::new(move || { + App::new() + .app_data(server_state_arc.clone()) + .wrap(Logger::new(&log_format).exclude("/_status")) + .service( + web::resource("/") + .route(web::get().to(|| HttpResponse::Ok())) + .route(web::head().to(|| HttpResponse::Ok())), + ) + .service( + web::resource("/_status") + .route(web::get().to(get_health_check_handler)) + .route(web::head().to(|| HttpResponse::Ok())), + ) + }) + .bind(bind_address)? + .workers(num_workers) + .run(); + + Ok(handle) +} + +pub async fn launch_http_server(args: CreateServerArgs) -> AnyhowResult<()> { + run_http_server(args)?.await?; + Ok(()) +} diff --git a/crates/service/job/gmicloud_job/src/job_dependencies.rs b/crates/service/job/gmicloud_job/src/job_dependencies.rs new file mode 100644 index 0000000000..f6d8979ac0 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/job_dependencies.rs @@ -0,0 +1,26 @@ +use cloud_storage::bucket_client::BucketClient; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey; +use jobs_common::job_stats::JobStats; +use server_environment::ServerEnvironment; +use sqlx::MySqlPool; + +pub struct JobDependencies { + pub mysql_pool: MySqlPool, + + /// Public GCS/S3 bucket for storing generated videos. + pub public_bucket_client: BucketClient, + + /// API key for GmiCloud. + pub gmicloud_api_key: GmiCloudApiKey, + + pub server_environment: ServerEnvironment, + + pub job_stats: JobStats, + + /// How long to sleep between poll iterations (milliseconds). + pub poll_interval_millis: u64, + + /// Set to `true` from another thread to trigger graceful shutdown. + pub application_shutdown: RelaxedAtomicBool, +} diff --git a/crates/service/job/gmicloud_job/src/main.rs b/crates/service/job/gmicloud_job/src/main.rs new file mode 100644 index 0000000000..3a22a6e64e --- /dev/null +++ b/crates/service/job/gmicloud_job/src/main.rs @@ -0,0 +1,140 @@ +// Never allow these +#![forbid(private_bounds)] +#![forbid(private_interfaces)] +#![forbid(unused_must_use)] + +// Always allow +#![allow(dead_code)] +#![allow(non_snake_case)] + +#[macro_use] extern crate serde_derive; + +use std::time::Duration; + +use anyhow::anyhow; +use log::{info, warn}; +use sqlx::mysql::MySqlPoolOptions; + +use bootstrap::bootstrap::{bootstrap, BootstrapArgs}; +use cloud_storage::bucket_client::BucketClient; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use shared_env_var_config::logging::DEFAULT_RUST_LOG; +use errors::AnyhowResult; +use gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey; +use jobs_common::job_stats::JobStats; +use server_environment::ServerEnvironment; +use shared_env_var_config::mysql::env_get_mysql_connection_string_or_default; + +use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs}; +use crate::main_loop::main_loop; +use crate::job_dependencies::JobDependencies; + +pub mod http_server; +pub mod job_dependencies; +pub mod main_loop; +pub mod process_job; + +// Bucket config +const ENV_ACCESS_KEY: &str = "ACCESS_KEY"; +const ENV_SECRET_KEY: &str = "SECRET_KEY"; +const ENV_REGION_NAME: &str = "REGION_NAME"; +const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME"; +const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL"; +const ENV_GMICLOUD_API_KEY: &str = "GMICLOUD_API_KEY"; + +#[actix_web::main] +async fn main() -> AnyhowResult<()> { + + let container_environment = bootstrap(BootstrapArgs { + app_name: "gmicloud-job", + default_logging_override: Some(DEFAULT_RUST_LOG), + config_search_directories: &[".", "./config", "crates/service/job/gmicloud_job/config"], + ignore_legacy_dot_env_file: true, + })?; + + info!("Hostname: {}", &container_environment.hostname); + + let _k8s_node_name = easyenv::get_env_string_optional("K8S_NODE_NAME"); + let _k8s_pod_name = easyenv::get_env_string_optional("K8S_POD_NAME"); + + let db_connection_string = env_get_mysql_connection_string_or_default(); + + info!("Connecting to database..."); + + let mysql_pool = MySqlPoolOptions::new() + .max_connections(2) + .connect(&db_connection_string) + .await?; + + info!("Connected to MySQL."); + + let server_environment = ServerEnvironment::from_str( + &easyenv::get_env_string_required("SERVER_ENVIRONMENT")?, + ) + .ok_or(anyhow!("invalid server environment"))?; + + // Bucket setup + let access_key = easyenv::get_env_string_required(ENV_ACCESS_KEY)?; + let secret_key = easyenv::get_env_string_required(ENV_SECRET_KEY)?; + let region_name = easyenv::get_env_string_required(ENV_REGION_NAME)?; + let public_bucket_name = easyenv::get_env_string_required(ENV_PUBLIC_BUCKET_NAME)?; + let s3_compatible_endpoint_url = easyenv::get_env_string_required(ENV_S3_ENDPOINT)?; + + let bucket_timeout = easyenv::get_env_duration_seconds_or_default( + "BUCKET_TIMEOUT_SECONDS", + Duration::from_secs(60 * 5), + ); + + let public_bucket_client = BucketClient::create( + &access_key, + &secret_key, + ®ion_name, + &public_bucket_name, + &s3_compatible_endpoint_url, + None, + Some(bucket_timeout), + )?; + + // GmiCloud API key + let gmicloud_api_key_str = easyenv::get_env_string_required(ENV_GMICLOUD_API_KEY)?; + let gmicloud_api_key = GmiCloudApiKey::new(gmicloud_api_key_str); + + // How often to poll for results (default: 15 seconds) + let poll_interval_millis: u64 = easyenv::get_env_num( + "GMICLOUD_POLL_INTERVAL_MILLIS", + 15_000, + )?; + + let application_shutdown = RelaxedAtomicBool::new(false); + let job_stats = JobStats::new(); + + let create_server_args = CreateServerArgs { + container_environment: container_environment.clone(), + job_stats: job_stats.clone(), + }; + + let job_dependencies = JobDependencies { + mysql_pool, + public_bucket_client, + gmicloud_api_key, + server_environment, + job_stats, + poll_interval_millis, + application_shutdown: application_shutdown.clone(), + }; + + std::thread::spawn(move || { + let actix_runtime = actix_web::rt::System::new(); + let http_server_handle = launch_http_server(create_server_args); + + actix_runtime.block_on(http_server_handle) + .expect("HTTP server should not exit."); + + warn!("Server thread is shut down."); + application_shutdown.set(true); + }); + + main_loop(job_dependencies).await; + + Ok(()) +} diff --git a/crates/service/job/gmicloud_job/src/main_loop.rs b/crates/service/job/gmicloud_job/src/main_loop.rs new file mode 100644 index 0000000000..fb6616303b --- /dev/null +++ b/crates/service/job/gmicloud_job/src/main_loop.rs @@ -0,0 +1,93 @@ +use std::time::Duration; + +use log::{error, info, warn}; +use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::list_pending_gmicloud_jobs; +use gmicloud_client::requests::poll_request_queue::poll_request_queue::poll_gmicloud_request; + +use crate::process_job::process_failed_job::process_failed_job; +use crate::process_job::process_successful_job::process_successful_job; +use crate::job_dependencies::JobDependencies; + +pub async fn main_loop(job_dependencies: JobDependencies) { + while !job_dependencies.application_shutdown.get() { + let result = run_poll_iteration(&job_dependencies).await; + + if let Err(err) = result { + error!("Error in poll iteration: {:?}", err); + let _ = job_dependencies.job_stats.increment_failure_count(); + } + + tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await; + } + + warn!("GmiCloud job runner main loop is shut down."); +} + +async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { + let pending_jobs = list_pending_gmicloud_jobs(&deps.mysql_pool).await?; + + if pending_jobs.is_empty() { + return Ok(()); + } + + info!("Found {} pending GmiCloud job(s).", pending_jobs.len()); + + for job in &pending_jobs { + let poll_result = match poll_gmicloud_request(&deps.gmicloud_api_key, &job.request_id).await { + Ok(response) => response, + Err(err) => { + warn!( + "Error polling GmiCloud request {} for job {}: {:?}", + job.request_id, job.job_token.as_str(), err + ); + continue; + } + }; + + if poll_result.is_failed() { + let reason = "GmiCloud generation failed"; + info!( + "Request {} failed. Processing job {} as failed.", + job.request_id, job.job_token.as_str() + ); + process_failed_job(deps, job, reason).await; + continue; + } + + if poll_result.is_in_progress() { + continue; + } + + if poll_result.is_success() { + let video_url = match poll_result.video_url() { + Some(url) => url.to_string(), + None => { + warn!( + "Request {} succeeded but has no video_url for job {}. Skipping.", + job.request_id, job.job_token.as_str() + ); + continue; + } + }; + + let thumbnail_url = poll_result.thumbnail_url().map(|s| s.to_string()); + + info!( + "Request {} completed, processing job {}.", + job.request_id, job.job_token.as_str() + ); + + if let Err(err) = process_successful_job(deps, job, &video_url, thumbnail_url.as_deref()).await { + warn!( + "Error processing completed request {} for job {}: {:?}", + job.request_id, job.job_token.as_str(), err + ); + let _ = deps.job_stats.increment_failure_count(); + } else { + let _ = deps.job_stats.increment_success_count(); + } + } + } + + Ok(()) +} diff --git a/crates/service/job/gmicloud_job/src/process_job/mod.rs b/crates/service/job/gmicloud_job/src/process_job/mod.rs new file mode 100644 index 0000000000..357e8d5ae1 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/process_job/mod.rs @@ -0,0 +1,2 @@ +pub mod process_failed_job; +pub mod process_successful_job; diff --git a/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs b/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs new file mode 100644 index 0000000000..42a5547258 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs @@ -0,0 +1,113 @@ +use log::{error, info, warn}; + +use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; +use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs}; +use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob; +use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome}; + +use crate::job_dependencies::JobDependencies; + +pub async fn process_failed_job( + deps: &JobDependencies, + job: &PendingGmiCloudJob, + reason: &str, +) { + // --- Step 1: Attempt the refund before touching the job status. --- + // + // Refund first so a crash between refund and status update never leaves + // a failed job without a refund. If the refund fails, bail out early: + // the job stays pending and will be retried on the next poll cycle. + + match &job.maybe_wallet_ledger_entry_token { + None => { + warn!( + "Job {} has no wallet ledger entry token; skipping refund.", + job.job_token.as_str() + ); + } + Some(ledger_token) => { + let mut transaction = match deps.mysql_pool.begin().await { + Ok(tx) => tx, + Err(err) => { + error!( + "Failed to begin refund transaction for job {} (ledger {}): {:?}. \ + Job will NOT be marked failed yet and will be retried next poll.", + job.job_token.as_str(), ledger_token.as_str(), err + ); + return; + } + }; + + match try_to_refund_ledger_entry(ledger_token, &mut transaction).await { + Ok(WalletRefundOutcome::Refunded(summary)) => { + info!( + "Refunded {} credits for failed job {} (ledger {} -> refund ledger {}).", + summary.refund_amount, + job.job_token.as_str(), + ledger_token.as_str(), + summary.refund_ledger_entry_token.as_str(), + ); + if let Err(err) = transaction.commit().await { + error!( + "Failed to commit refund transaction for job {} (ledger {}): {:?}. \ + Job will NOT be marked failed yet and will be retried next poll.", + job.job_token.as_str(), ledger_token.as_str(), err + ); + return; + } + } + Ok(WalletRefundOutcome::AlreadyRefunded) => { + info!( + "Ledger entry {} for job {} was already refunded; proceeding to mark job failed.", + ledger_token.as_str(), job.job_token.as_str(), + ); + let _ = transaction.rollback().await; + } + Err(err) => { + error!( + "Failed to refund ledger entry {} for job {}: {:?}. \ + Job will NOT be marked failed yet and will be retried next poll.", + ledger_token.as_str(), job.job_token.as_str(), err, + ); + let _ = transaction.rollback().await; + return; + } + } + } + } + + // --- Step 2: Mark the job record as failed. --- + + let reason_lower = reason.to_lowercase(); + + let platform_rules_violation = reason_lower.contains("violates") + || reason_lower.contains("platform rules") + || reason_lower.contains("please modify") + || reason_lower.contains("content policy"); + + let frontend_failure_category = if platform_rules_violation { + Some(FrontendFailureCategory::ModelRulesViolation) + } else { + None + }; + + warn!( + "Request for job {} failed: {}. Marking job failed.", + job.job_token.as_str(), reason + ); + + let mark_failed_result = mark_job_failed_by_token(MarkJobFailedByTokenArgs { + pool: &deps.mysql_pool, + job_token: &job.job_token, + maybe_public_failure_reason: Some(reason), + internal_debugging_failure_reason: reason, + maybe_frontend_failure_category: frontend_failure_category, + }).await; + + if let Err(err) = mark_failed_result { + error!( + "Error marking job {} as failed: {:?}", + job.job_token.as_str(), err + ); + } +} diff --git a/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs b/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs new file mode 100644 index 0000000000..76573dbf14 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs @@ -0,0 +1,189 @@ +use anyhow::anyhow; +use log::{error, info, warn}; + +use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath; +use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType; +use enums::by_table::media_files::media_file_class::MediaFileClass; +use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory; +use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory; +use enums::by_table::media_files::media_file_type::MediaFileType; +use errors::AnyhowResult; +use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes; +use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob; +use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder; +use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_successfully_done_by_token::mark_generic_inference_job_successfully_done_by_token; +use tokens::tokens::media_files::MediaFileToken; + +use crate::job_dependencies::JobDependencies; + +const VIDEO_PREFIX: &str = "artcraft_"; +const VIDEO_SUFFIX: &str = ".mp4"; + +const THUMB_PREFIX: &str = "artcraft_"; + +/// Download the completed video, upload to bucket, create media file record, and mark job done. +pub async fn process_successful_job( + deps: &JobDependencies, + job: &PendingGmiCloudJob, + video_url: &str, + maybe_thumbnail_url: Option<&str>, +) -> AnyhowResult<()> { + + // --- Step 1: Download and upload thumbnail as cover image (if available). --- + + let maybe_cover_token = match maybe_thumbnail_url { + Some(thumbnail_url) => { + match download_and_upload_thumbnail(deps, job, thumbnail_url).await { + Ok(token) => Some(token), + Err(err) => { + warn!( + "Failed to create thumbnail for job {}: {:?}. Continuing without cover.", + job.job_token.as_str(), err + ); + None + } + } + } + None => None, + }; + + // --- Step 2: Download and upload the video. --- + + info!("Downloading video for job {} from: {}", job.job_token.as_str(), video_url); + + let video_bytes: Vec = reqwest::get(video_url) + .await + .map_err(|err| anyhow!("reqwest error downloading video: {:?}", err))? + .bytes() + .await + .map_err(|err| anyhow!("error reading video bytes: {:?}", err))? + .to_vec(); + + info!("Downloaded {} bytes for job {}", video_bytes.len(), job.job_token.as_str()); + + let checksum = sha256_hash_bytes(&video_bytes) + .map_err(|err| anyhow!("error hashing video: {:?}", err))?; + + let bucket_path = MediaFileBucketPath::generate_new(Some(VIDEO_PREFIX), Some(VIDEO_SUFFIX)); + let object_path = bucket_path.get_full_object_path_str(); + + info!("Uploading video to public bucket at path: {}", object_path); + + deps + .public_bucket_client + .upload_file_with_content_type_process(object_path, &video_bytes, "video/mp4") + .await + .map_err(|err| anyhow!("error uploading video to bucket: {:?}", err))?; + + info!("Uploaded video for job {}. Creating media file record.", job.job_token.as_str()); + + let media_file_token = MediaFileInsertBuilder::new() + .maybe_creator_user(job.maybe_creator_user_token.as_ref()) + .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) + .creator_ip_address(&job.creator_ip_address) + .creator_set_visibility(job.creator_set_visibility) + .media_file_class(MediaFileClass::Video) + .media_file_type(MediaFileType::Mp4) + .media_file_origin_category(MediaFileOriginCategory::Inference) + .media_file_origin_product_category(MediaFileOriginProductCategory::VideoGeneration) + .mime_type("video/mp4") + .file_size_bytes(video_bytes.len() as u64) + .checksum_sha2(&checksum) + .maybe_prompt_token(job.maybe_prompt_token.as_ref()) + .maybe_cover_image_media_file_token(maybe_cover_token.as_ref()) + .public_bucket_directory_hash(&bucket_path) + .insert_pool(&deps.mysql_pool) + .await + .map_err(|err| anyhow!("error inserting media file record: {:?}", err))?; + + info!( + "Created media file {} for job {}. Marking job complete.", + media_file_token.as_str(), job.job_token.as_str() + ); + + mark_generic_inference_job_successfully_done_by_token( + &deps.mysql_pool, + &job.job_token, + Some(InferenceResultType::MediaFile), + Some(media_file_token.as_str()), + None, + None, + ) + .await + .map_err(|err| { + error!("Error marking job {} done: {:?}", job.job_token.as_str(), err); + anyhow!("error marking job done: {:?}", err) + })?; + + info!("Job {} completed successfully.", job.job_token.as_str()); + + Ok(()) +} + +fn guess_image_format_from_url(url: &str) -> (&'static str, &'static str, MediaFileType) { + let path = url.split('?').next().unwrap_or(url); + let path = path.split('#').next().unwrap_or(path); + + if let Some(dot) = path.rfind('.') { + match &path[dot..] { + ".jpg" | ".jpeg" => (".jpg", "image/jpeg", MediaFileType::Jpg), + ".png" => (".png", "image/png", MediaFileType::Png), + _ => (".png", "image/png", MediaFileType::Png), + } + } else { + (".png", "image/png", MediaFileType::Png) + } +} + +async fn download_and_upload_thumbnail( + deps: &JobDependencies, + job: &PendingGmiCloudJob, + thumbnail_url: &str, +) -> AnyhowResult { + info!("Downloading thumbnail for job {} from: {}", job.job_token.as_str(), thumbnail_url); + + let (suffix, mime_type, media_file_type) = guess_image_format_from_url(thumbnail_url); + + let thumb_bytes: Vec = reqwest::get(thumbnail_url) + .await + .map_err(|err| anyhow!("reqwest error downloading thumbnail: {:?}", err))? + .bytes() + .await + .map_err(|err| anyhow!("error reading thumbnail bytes: {:?}", err))? + .to_vec(); + + let checksum = sha256_hash_bytes(&thumb_bytes) + .map_err(|err| anyhow!("error hashing thumbnail: {:?}", err))?; + + let bucket_path = MediaFileBucketPath::generate_new(Some(THUMB_PREFIX), Some(suffix)); + let object_path = bucket_path.get_full_object_path_str(); + + deps + .public_bucket_client + .upload_file_with_content_type_process(object_path, &thumb_bytes, mime_type) + .await + .map_err(|err| anyhow!("error uploading thumbnail to bucket: {:?}", err))?; + + let thumb_token = MediaFileInsertBuilder::new() + .maybe_creator_user(job.maybe_creator_user_token.as_ref()) + .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) + .creator_ip_address(&job.creator_ip_address) + .creator_set_visibility(job.creator_set_visibility) + .media_file_class(MediaFileClass::Image) + .media_file_type(media_file_type) + .is_intermediate_system_file(true) + .media_file_origin_category(MediaFileOriginCategory::Inference) + .media_file_origin_product_category(MediaFileOriginProductCategory::VideoGeneration) + .mime_type(mime_type) + .file_size_bytes(thumb_bytes.len() as u64) + .checksum_sha2(&checksum) + .maybe_prompt_token(job.maybe_prompt_token.as_ref()) + .public_bucket_directory_hash(&bucket_path) + .insert_pool(&deps.mysql_pool) + .await + .map_err(|err| anyhow!("error inserting thumbnail media file record: {:?}", err))?; + + info!("Created thumbnail media file {} for job {}.", thumb_token.as_str(), job.job_token.as_str()); + + Ok(thumb_token) +} diff --git a/crates/service/web/storyteller_web/src/configs/omni_gen/video_models.rs b/crates/service/web/storyteller_web/src/configs/omni_gen/video_models.rs index 104ae1a45e..f5f98bc71d 100644 --- a/crates/service/web/storyteller_web/src/configs/omni_gen/video_models.rs +++ b/crates/service/web/storyteller_web/src/configs/omni_gen/video_models.rs @@ -155,84 +155,84 @@ fn build_omni_gen_video_models() -> Vec { ..Default::default() }); -// models.push(OmniGenVideoModelDetails { -// model: CommonVideoModel::Seedance2p0Global, -// model_creator: Some(ModelCreator::Bytedance), -// full_name: Some("Seedance 2.0 (Global)".to_string()), -// text_prompt_supported: Some(true), -// starting_keyframe_supported: Some(true), -// ending_keyframe_supported: Some(true), -// image_references_supported: Some(true), -// image_references_max: Some(9), -// audio_references_supported: Some(true), -// audio_references_max: Some(3), -// audio_references_max_total_duration_seconds: Some(15), -// video_references_supported: Some(true), -// video_references_max: Some(3), -// video_references_max_total_duration_seconds: Some(15), -// character_references_supported: Some(true), -// character_references_max: Some(9), -// aspect_ratio_options: Some(vec![ -// CommonAspectRatio::WideTwentyOneByNine, -// CommonAspectRatio::WideSixteenByNine, -// CommonAspectRatio::WideFourByThree, -// CommonAspectRatio::Square, -// CommonAspectRatio::TallThreeByFour, -// CommonAspectRatio::TallNineBySixteen, -// ]), -// aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine), -// resolution_options: Some(vec![ -// CommonResolution::FourEightyP, -// CommonResolution::SevenTwentyP, -// CommonResolution::TenEightyP, -// ]), -// resolution_default: Some(CommonResolution::SevenTwentyP), -// batch_size_options: Some(vec![1, 2, 4]), -// batch_size_default: Some(1), -// duration_seconds_min: Some(4), -// duration_seconds_max: Some(15), -// duration_seconds_default: Some(5), -// ..Default::default() -// }); -// -// models.push(OmniGenVideoModelDetails { -// model: CommonVideoModel::Seedance2p0FastGlobal, -// model_creator: Some(ModelCreator::Bytedance), -// full_name: Some("Seedance 2.0 Fast (Global)".to_string()), -// text_prompt_supported: Some(true), -// starting_keyframe_supported: Some(true), -// ending_keyframe_supported: Some(true), -// image_references_supported: Some(true), -// image_references_max: Some(9), -// audio_references_supported: Some(true), -// audio_references_max: Some(3), -// audio_references_max_total_duration_seconds: Some(15), -// video_references_supported: Some(true), -// video_references_max: Some(3), -// video_references_max_total_duration_seconds: Some(15), -// character_references_supported: Some(true), -// character_references_max: Some(9), -// aspect_ratio_options: Some(vec![ -// CommonAspectRatio::WideTwentyOneByNine, -// CommonAspectRatio::WideSixteenByNine, -// CommonAspectRatio::WideFourByThree, -// CommonAspectRatio::Square, -// CommonAspectRatio::TallThreeByFour, -// CommonAspectRatio::TallNineBySixteen, -// ]), -// aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine), -// resolution_options: Some(vec![ -// CommonResolution::FourEightyP, -// CommonResolution::SevenTwentyP, -// ]), -// resolution_default: Some(CommonResolution::SevenTwentyP), -// batch_size_options: Some(vec![1, 2, 4]), -// batch_size_default: Some(1), -// duration_seconds_min: Some(4), -// duration_seconds_max: Some(15), -// duration_seconds_default: Some(5), -// ..Default::default() -// }); + models.push(OmniGenVideoModelDetails { + model: CommonVideoModel::Seedance2p0Global, + model_creator: Some(ModelCreator::Bytedance), + full_name: Some("Seedance 2.0 (Global)".to_string()), + text_prompt_supported: Some(true), + starting_keyframe_supported: Some(true), + ending_keyframe_supported: Some(true), + image_references_supported: Some(true), + image_references_max: Some(9), + audio_references_supported: Some(true), + audio_references_max: Some(3), + audio_references_max_total_duration_seconds: Some(15), + video_references_supported: Some(true), + video_references_max: Some(3), + video_references_max_total_duration_seconds: Some(15), + character_references_supported: Some(true), + character_references_max: Some(9), + aspect_ratio_options: Some(vec![ + CommonAspectRatio::WideTwentyOneByNine, + CommonAspectRatio::WideSixteenByNine, + CommonAspectRatio::WideFourByThree, + CommonAspectRatio::Square, + CommonAspectRatio::TallThreeByFour, + CommonAspectRatio::TallNineBySixteen, + ]), + aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine), + resolution_options: Some(vec![ + CommonResolution::FourEightyP, + CommonResolution::SevenTwentyP, + CommonResolution::TenEightyP, + ]), + resolution_default: Some(CommonResolution::SevenTwentyP), + batch_size_options: Some(vec![1, 2, 4]), + batch_size_default: Some(1), + duration_seconds_min: Some(4), + duration_seconds_max: Some(15), + duration_seconds_default: Some(5), + ..Default::default() + }); + + models.push(OmniGenVideoModelDetails { + model: CommonVideoModel::Seedance2p0FastGlobal, + model_creator: Some(ModelCreator::Bytedance), + full_name: Some("Seedance 2.0 Fast (Global)".to_string()), + text_prompt_supported: Some(true), + starting_keyframe_supported: Some(true), + ending_keyframe_supported: Some(true), + image_references_supported: Some(true), + image_references_max: Some(9), + audio_references_supported: Some(true), + audio_references_max: Some(3), + audio_references_max_total_duration_seconds: Some(15), + video_references_supported: Some(true), + video_references_max: Some(3), + video_references_max_total_duration_seconds: Some(15), + character_references_supported: Some(true), + character_references_max: Some(9), + aspect_ratio_options: Some(vec![ + CommonAspectRatio::WideTwentyOneByNine, + CommonAspectRatio::WideSixteenByNine, + CommonAspectRatio::WideFourByThree, + CommonAspectRatio::Square, + CommonAspectRatio::TallThreeByFour, + CommonAspectRatio::TallNineBySixteen, + ]), + aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine), + resolution_options: Some(vec![ + CommonResolution::FourEightyP, + CommonResolution::SevenTwentyP, + ]), + resolution_default: Some(CommonResolution::SevenTwentyP), + batch_size_options: Some(vec![1, 2, 4]), + batch_size_default: Some(1), + duration_seconds_min: Some(4), + duration_seconds_max: Some(15), + duration_seconds_default: Some(5), + ..Default::default() + }); // TODO(bt,2026-04-10): Veo 2 image-to-video doesn't support aspect ratio models.push(OmniGenVideoModelDetails {