diff --git a/crates/api_clients/kinovi_web_client/src/billing/get_billing_payments_history.rs b/crates/api_clients/kinovi_web_client/src/billing/get_billing_payments_history.rs index 78ff5e9031..94c4d2e7d0 100644 --- a/crates/api_clients/kinovi_web_client/src/billing/get_billing_payments_history.rs +++ b/crates/api_clients/kinovi_web_client/src/billing/get_billing_payments_history.rs @@ -10,6 +10,7 @@ use crate::error::kinovi_web_error::KinoviWebError; use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError; use crate::requests::kinovi_host::{KinoviHost, resolve_host}; use crate::utils::common_headers::FIREFOX_USER_AGENT; +use crate::utils::request_timeouts::KINOVI_REQUEST_TIMEOUT; use crate::utils::number_coercion::de_u64_int_or_float; /// Page size the kinovi.ai billing dashboard uses. @@ -125,6 +126,7 @@ pub async fn get_billing_payments_history( let client = Client::builder() .emulation(Emulation::Firefox143) + .timeout(KINOVI_REQUEST_TIMEOUT) .build() .map_err(|err| KinoviWebClientError::WreqClientError(err))?; diff --git a/crates/api_clients/kinovi_web_client/src/billing/get_credits_history.rs b/crates/api_clients/kinovi_web_client/src/billing/get_credits_history.rs index 69ddbe6fb4..1468df40d3 100644 --- a/crates/api_clients/kinovi_web_client/src/billing/get_credits_history.rs +++ b/crates/api_clients/kinovi_web_client/src/billing/get_credits_history.rs @@ -10,6 +10,7 @@ use crate::error::kinovi_web_error::KinoviWebError; use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError; use crate::requests::kinovi_host::{KinoviHost, resolve_host}; use crate::utils::common_headers::FIREFOX_USER_AGENT; +use crate::utils::request_timeouts::KINOVI_REQUEST_TIMEOUT; use crate::utils::number_coercion::{de_i64_int_or_float, de_u64_int_or_float}; /// Page size the kinovi.ai billing dashboard uses. @@ -117,6 +118,7 @@ pub async fn get_credits_history( let client = Client::builder() .emulation(Emulation::Firefox143) + .timeout(KINOVI_REQUEST_TIMEOUT) .build() .map_err(|err| KinoviWebClientError::WreqClientError(err))?; diff --git a/crates/api_clients/kinovi_web_client/src/requests/create_material/create_material.rs b/crates/api_clients/kinovi_web_client/src/requests/create_material/create_material.rs index fbd663dbc7..8828e231e4 100644 --- a/crates/api_clients/kinovi_web_client/src/requests/create_material/create_material.rs +++ b/crates/api_clients/kinovi_web_client/src/requests/create_material/create_material.rs @@ -7,6 +7,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError; use crate::requests::create_material::request_types::*; use crate::requests::kinovi_host::{KinoviHost, resolve_host}; use crate::utils::common_headers::FIREFOX_USER_AGENT; +use crate::utils::request_timeouts::KINOVI_REQUEST_TIMEOUT; use log::info; use wreq::Client; use wreq_util::Emulation; @@ -215,6 +216,7 @@ pub async fn create_material(args: CreateMaterialArgs<'_>) -> Result) -> Result) -> Result) -> Result) -> Result) -> Result) -> Result) -> Result Result( let client = Client::builder() .emulation(Emulation::Firefox143) + .timeout(KINOVI_REQUEST_TIMEOUT) .build() .map_err(|err| KinoviWebClientError::WreqClientError(err))?; diff --git a/crates/api_clients/kinovi_web_client/src/utils/mod.rs b/crates/api_clients/kinovi_web_client/src/utils/mod.rs index df975fb934..579b7e7964 100644 --- a/crates/api_clients/kinovi_web_client/src/utils/mod.rs +++ b/crates/api_clients/kinovi_web_client/src/utils/mod.rs @@ -1,3 +1,4 @@ pub mod categorize_kinovi_web_error; pub mod common_headers; pub mod number_coercion; +pub mod request_timeouts; diff --git a/crates/api_clients/kinovi_web_client/src/utils/request_timeouts.rs b/crates/api_clients/kinovi_web_client/src/utils/request_timeouts.rs new file mode 100644 index 0000000000..2af0aab423 --- /dev/null +++ b/crates/api_clients/kinovi_web_client/src/utils/request_timeouts.rs @@ -0,0 +1,13 @@ +use std::time::Duration; + +/// Total request deadline (connect + send + receive) for ordinary Kinovi +/// API calls: tRPC queries, order/character polling, generation submits. +/// +/// Without a deadline a peer that accepts the connection and then never +/// answers parks the caller forever — the job loops that poll Kinovi have no +/// other way to recover from that short of a pod restart. +pub const KINOVI_REQUEST_TIMEOUT: Duration = Duration::from_secs(60); + +/// Deadline for file uploads to Kinovi's storage, which legitimately take +/// longer than an API round trip for large reference videos. +pub const KINOVI_UPLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60); diff --git a/crates/service/job/seedance2_pro_job/Cargo.toml b/crates/service/job/seedance2_pro_job/Cargo.toml index 7628ecf5db..fa8f0699a4 100644 --- a/crates/service/job/seedance2_pro_job/Cargo.toml +++ b/crates/service/job/seedance2_pro_job/Cargo.toml @@ -48,7 +48,7 @@ rand.workspace = true sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] } # Runtime -tokio = { version = "1.40.0", features = ["macros", "sync"] } +tokio = { version = "1.40.0", features = ["macros", "sync", "time"] } # HTTP Client reqwest.workspace = true diff --git a/crates/service/job/seedance2_pro_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/seedance2_pro_job/src/http_server/endpoints/health_check_handler.rs index 9c5f4a8b8f..4c2c49b8eb 100644 --- a/crates/service/job/seedance2_pro_job/src/http_server/endpoints/health_check_handler.rs +++ b/crates/service/job/seedance2_pro_job/src/http_server/endpoints/health_check_handler.rs @@ -11,6 +11,7 @@ use pager::notification::notification_details_builder::NotificationDetailsBuilde use pager::notification::notification_urgency::NotificationUrgency; use crate::http_server::http_server_shared_state::HttpServerSharedState; +use crate::loop_heartbeats::HeartbeatAge; #[derive(Serialize)] pub struct HealthCheckResponse { @@ -22,6 +23,16 @@ pub struct HealthCheckResponse { pub total_success_count: u64, pub total_failure_ratio: f32, pub total_success_ratio: f32, + /// Seconds since each loop last made progress, oldest first. + pub loop_heartbeat_ages_seconds: Vec, + /// Loops whose heartbeat is older than the staleness threshold. + pub stale_loops: Vec, +} + +#[derive(Serialize)] +pub struct LoopHeartbeatAge { + pub loop_name: String, + pub age_seconds: u64, } #[derive(Debug, Serialize)] @@ -75,8 +86,21 @@ pub async fn get_health_check_handler( 0.0 }; - let is_healthy = - job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold; + let heartbeat_ages = server_state.heartbeats.ages(); + let stale_loops: Vec = server_state.heartbeats + .stale(server_state.heartbeat_stale_threshold) + .into_iter() + .map(|entry| format!("{} ({}s)", entry.loop_name, entry.age.as_secs())) + .collect(); + + let too_many_failures = + job_stats.consecutive_failure_count >= server_state.consecutive_failure_unhealthy_threshold; + + let is_healthy = !too_many_failures && stale_loops.is_empty(); + + if !stale_loops.is_empty() { + error!("Health check: stale loop heartbeat(s): {:?}", stale_loops); + } if !is_healthy { let notification = NotificationDetailsBuilder::from_title( @@ -86,11 +110,13 @@ pub async fn get_health_check_handler( Hostname: {}\n\ Consecutive failure count: {}\n\ Total failure count: {}\n\ - Total success count: {}", + Total success count: {}\n\ + Stale loops: {:?}", server_state.hostname, job_stats.consecutive_failure_count, job_stats.total_failure_count, job_stats.total_success_count, + stale_loops, ))) .set_urgency(Some(NotificationUrgency::High)) .set_http_method(Some(http_request.method().to_string())) @@ -111,6 +137,8 @@ pub async fn get_health_check_handler( total_success_count: job_stats.total_success_count, total_failure_ratio, total_success_ratio, + loop_heartbeat_ages_seconds: heartbeat_ages.into_iter().map(to_loop_heartbeat_age).collect(), + stale_loops, }; let body = serde_json::to_string(&response).map_err(|_e| HealthCheckError::ServerError)?; @@ -125,3 +153,10 @@ pub async fn get_health_check_handler( .body(body)) } } + +fn to_loop_heartbeat_age(entry: HeartbeatAge) -> LoopHeartbeatAge { + LoopHeartbeatAge { + loop_name: entry.loop_name.to_string(), + age_seconds: entry.age.as_secs(), + } +} diff --git a/crates/service/job/seedance2_pro_job/src/http_server/http_server_shared_state.rs b/crates/service/job/seedance2_pro_job/src/http_server/http_server_shared_state.rs index ef3e9fe208..7128062611 100644 --- a/crates/service/job/seedance2_pro_job/src/http_server/http_server_shared_state.rs +++ b/crates/service/job/seedance2_pro_job/src/http_server/http_server_shared_state.rs @@ -1,10 +1,17 @@ +use std::time::Duration; + use jobs_common::job_stats::JobStats; use pager::client::pager::Pager; +use crate::loop_heartbeats::LoopHeartbeats; + #[derive(Clone)] pub struct HttpServerSharedState { pub job_stats: JobStats, pub consecutive_failure_unhealthy_threshold: u64, + pub heartbeats: LoopHeartbeats, + /// A loop whose last heartbeat is older than this makes the pod unhealthy. + pub heartbeat_stale_threshold: Duration, pub pager: Pager, pub hostname: String, } diff --git a/crates/service/job/seedance2_pro_job/src/http_server/run_http_server.rs b/crates/service/job/seedance2_pro_job/src/http_server/run_http_server.rs index b43fbca523..4110424707 100644 --- a/crates/service/job/seedance2_pro_job/src/http_server/run_http_server.rs +++ b/crates/service/job/seedance2_pro_job/src/http_server/run_http_server.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::time::Duration; use actix_web::dev::Server; use actix_web::middleware::Logger; @@ -12,13 +13,19 @@ use pager::client::pager::Pager; use crate::http_server::endpoints::health_check_handler::get_health_check_handler; use crate::http_server::http_server_shared_state::HttpServerSharedState; +use crate::loop_heartbeats::LoopHeartbeats; const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:11223"; const DEFAULT_NUM_WORKERS: usize = 2; +/// Loops heartbeat at least once per polled page (sub-second) and once per +/// idle tick, so anything quieter than this for five minutes is wedged. +const DEFAULT_HEARTBEAT_STALE_THRESHOLD_SECONDS: u64 = 5 * 60; + pub struct CreateServerArgs { pub container_environment: ContainerEnvironment, pub job_stats: JobStats, + pub heartbeats: LoopHeartbeats, pub pager: Pager, } @@ -33,6 +40,11 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult { "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", 3, )?, + heartbeats: args.heartbeats, + heartbeat_stale_threshold: Duration::from_secs(easyenv::get_env_num( + "HEARTBEAT_STALE_THRESHOLD_SECONDS", + DEFAULT_HEARTBEAT_STALE_THRESHOLD_SECONDS, + )?), pager: args.pager, hostname, }; diff --git a/crates/service/job/seedance2_pro_job/src/job_dependencies.rs b/crates/service/job/seedance2_pro_job/src/job_dependencies.rs index f4acfbe33c..0c724fa23b 100644 --- a/crates/service/job/seedance2_pro_job/src/job_dependencies.rs +++ b/crates/service/job/seedance2_pro_job/src/job_dependencies.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use crate::kinovi_version::KinoviVersion; +use crate::loop_heartbeats::LoopHeartbeats; use crate::order_reconciler::OrderReconciler; use chrono::Duration; use cloud_storage::legacy_bucket_client::LegacyBucketClient; @@ -19,6 +20,10 @@ pub struct JobDependencies { /// Public GCS/S3 bucket for storing generated videos. pub public_bucket_client: LegacyBucketClient, + /// HTTP client for downloading finished media from Kinovi's CDN. Built once + /// with a request deadline so a stalled download can't park a loop forever. + pub download_client: reqwest::Client, + /// Session credentials for polling seedance2-pro.com. pub kinovi_web_session: KinoviWebSession, @@ -61,4 +66,8 @@ pub struct JobDependencies { /// Hand-off between the polling loop (producer) and the processing loop /// (consumer): finished Kinovi orders staged for reconciliation into our DB. pub order_reconciler: OrderReconciler, + + /// Liveness signal from each loop, surfaced by the health check so a + /// wedged loop gets the pod restarted instead of sitting silent. + pub heartbeats: LoopHeartbeats, } diff --git a/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/character_polling_main_loop.rs b/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/character_polling_main_loop.rs index 678b1d1ec5..280b6442a7 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/character_polling_main_loop.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/character_polling_main_loop.rs @@ -9,13 +9,22 @@ use kinovi_web_client::requests::poll_characters::poll_characters::{ }; use crate::job_dependencies::JobDependencies; +use crate::with_deadline::{with_deadline, DATABASE_DEADLINE}; use super::process_failed_character::process_failed_character; use super::process_successful_character::process_successful_character; const POLL_ALERT_THRESHOLD: Duration = Duration::from_secs(600); +/// Hard ceiling on reconciling one finished character (download + DB writes). +const CHARACTER_PROCESSING_DEADLINE: Duration = Duration::from_secs(15 * 60); + +/// Name under which this loop reports liveness to the health check. +pub const HEARTBEAT_NAME: &str = "character_polling"; + pub async fn character_polling_main_loop(deps: JobDependencies) { while !deps.application_shutdown.get() { + deps.heartbeats.beat(HEARTBEAT_NAME); + let start = Instant::now(); let result = run_poll_iteration(&deps).await; let elapsed = start.elapsed(); @@ -40,7 +49,11 @@ pub async fn character_polling_main_loop(deps: JobDependencies) { async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { // 1. Query all pending character creation jobs from DB. - let pending_jobs = list_pending_kinovi_web_character_jobs(&deps.mysql_pool).await?; + let pending_jobs = with_deadline( + "pending character jobs query", + DATABASE_DEADLINE, + list_pending_kinovi_web_character_jobs(&deps.mysql_pool), + ).await?; if pending_jobs.is_empty() { return Ok(()); @@ -72,6 +85,8 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { break; } + deps.heartbeats.beat(HEARTBEAT_NAME); + let job = match job_by_character_id.get(&character.character_id) { Some(j) => j, None => continue, // Not one of our pending jobs. @@ -83,7 +98,13 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { "Character {} completed successfully, processing job {}", character.character_id, job.job_token, ); - match process_successful_character(deps, job, character).await { + let outcome = with_deadline( + "character processing", + CHARACTER_PROCESSING_DEADLINE, + process_successful_character(deps, job, character), + ).await; + + match outcome { Ok(()) => { let _ = deps.job_stats.increment_success_count(); } @@ -101,7 +122,15 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { "Character {} failed, processing job {}", character.character_id, job.job_token, ); - process_failed_character(deps, job, character).await; + let outcome = with_deadline( + "failed character processing", + CHARACTER_PROCESSING_DEADLINE, + async { Ok::<(), anyhow::Error>(process_failed_character(deps, job, character).await) }, + ).await; + + if let Err(err) = outcome { + warn!("Error processing failed character {}: {:?}", character.character_id, err); + } } CharacterCreationStatus::Pending => { // Still in progress — check again next poll. diff --git a/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/process_successful_character.rs b/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/process_successful_character.rs index 425e4f2d8a..173c660fca 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/process_successful_character.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/process_successful_character.rs @@ -117,7 +117,8 @@ async fn download_and_create_media_file( ) -> AnyhowResult { info!("Downloading character image: {}", image_url); - let image_bytes: Vec = reqwest::get(image_url) + let image_bytes: Vec = deps.download_client.get(image_url) + .send() .await .map_err(|err| anyhow!("reqwest error downloading image: {:?}", err))? .bytes() diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_polling_job/order_polling_main_loop.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_polling_job/order_polling_main_loop.rs index 88fdf4e792..66af7c1f22 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_polling_job/order_polling_main_loop.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_polling_job/order_polling_main_loop.rs @@ -13,9 +13,13 @@ use kinovi_web_client::requests::poll_orders::poll_orders::{poll_orders, OrderSt use crate::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; use crate::kinovi_version::KinoviVersion; +use crate::with_deadline::{with_deadline, DATABASE_DEADLINE}; const POLL_ALERT_THRESHOLD: Duration = Duration::from_mins(6); +/// Name under which this loop reports liveness to the health check. +pub const HEARTBEAT_NAME: &str = "order_polling"; + /// How many of the newest order pages to re-scan when we periodically revisit /// the head of the list. Recently-submitted jobs finish soonest, so checking /// the first few pages more often shortens their tail latency without waiting @@ -37,6 +41,8 @@ const DEEP_PAGES_BETWEEN_HEAD_RECHECKS: u32 = 5; /// batch (or the whole walk) to complete. pub async fn order_polling_main_loop(job_dependencies: JobDependencies) { while !job_dependencies.application_shutdown.get() { + job_dependencies.heartbeats.beat(HEARTBEAT_NAME); + let start = Instant::now(); let result = run_poll_iteration(&job_dependencies).await; @@ -94,11 +100,15 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { }; // 1. Query all (limit 25,000) non-terminal KinoviWeb jobs from the DB. - let pending_jobs = match list_pending_kinovi_web_video_jobs(&deps.mysql_pool, third_party, job_type).await { + let pending_jobs = match with_deadline( + "pending jobs query", + DATABASE_DEADLINE, + list_pending_kinovi_web_video_jobs(&deps.mysql_pool, third_party, job_type), + ).await { Ok(jobs) => jobs, Err(err) => { error!("Failed to list pending database jobs: {:?}", err); - return alert_pager_and_return_err(&deps.pager, "Jobs DB query failed", err.into(), None); + return alert_pager_and_return_err(&deps.pager, "Jobs DB query failed", err, None); } }; @@ -188,6 +198,8 @@ async fn walk_orders( let response = poll_orders_with_retry(deps, cursor).await?; + deps.heartbeats.beat(HEARTBEAT_NAME); + let page_summary = stage_finished_orders(deps, &response.orders, &job_by_order_id); result.pages_seen += 1; @@ -262,6 +274,8 @@ async fn recheck_head_pages( let response = poll_orders_with_retry(deps, cursor).await?; + deps.heartbeats.beat(HEARTBEAT_NAME); + let page_summary = stage_finished_orders(deps, &response.orders, job_by_order_id); result.pages_seen += 1; diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/order_processing_main_loop.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/order_processing_main_loop.rs index 1435063980..857af54e17 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/order_processing_main_loop.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/order_processing_main_loop.rs @@ -6,6 +6,7 @@ use mysql_queries::queries::generic_inference::web::get_inference_job_status_val use crate::job_dependencies::JobDependencies; use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal; use crate::jobs::order_processing_job::process_one_order::process_one_order; +use crate::with_deadline::{with_deadline, DATABASE_DEADLINE}; /// How long to nap when the reconciler is empty before peeking again. const IDLE_SLEEP: Duration = Duration::from_millis(500); @@ -13,6 +14,14 @@ const IDLE_SLEEP: Duration = Duration::from_millis(500); /// How long to back off after a transient error before retrying. const ERROR_BACKOFF: Duration = Duration::from_secs(2); +/// Hard ceiling on reconciling one order end to end (download, upload, DB +/// finalize). Each I/O step has its own deadline; this is the backstop that +/// guarantees the consumer can never be parked on a single order forever. +const ORDER_PROCESSING_DEADLINE: Duration = Duration::from_secs(15 * 60); + +/// Name under which this loop reports liveness to the health check. +pub const HEARTBEAT_NAME: &str = "order_processing"; + /// The consumer half of the pipeline. /// /// Drains finished orders staged by the polling loop and reconciles each into @@ -23,6 +32,8 @@ const ERROR_BACKOFF: Duration = Duration::from_secs(2); /// soon as it's staged rather than after a whole batch. pub async fn order_processing_main_loop(deps: JobDependencies) { while !deps.application_shutdown.get() { + deps.heartbeats.beat(HEARTBEAT_NAME); + let details = match deps.order_reconciler.peek_random() { Some(details) => details, None => { @@ -36,7 +47,13 @@ pub async fn order_processing_main_loop(deps: JobDependencies) { // Lightweight pre-check: if the DB already considers the job terminal, drop // the staged order without doing any expensive download/upload work. - match get_inference_job_status_value(&deps.mysql_pool, &job_token).await { + let status_lookup = with_deadline( + "job status pre-check", + DATABASE_DEADLINE, + get_inference_job_status_value(&deps.mysql_pool, &job_token), + ).await; + + match status_lookup { Ok(Some(status)) if is_job_status_terminal(status) => { info!( "Job {} is already terminal ({:?}); dropping staged order {}.", @@ -69,7 +86,13 @@ pub async fn order_processing_main_loop(deps: JobDependencies) { // Commit to this order: pop it so we don't peek it again while working. deps.order_reconciler.remove(&order_id); - match process_one_order(&deps, &details).await { + let outcome = with_deadline( + "order processing", + ORDER_PROCESSING_DEADLINE, + process_one_order(&deps, &details), + ).await; + + match outcome { Ok(()) => { let _ = deps.job_stats.increment_success_count(); } diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_image_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_image_job.rs index d85593c009..b50c84365e 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_image_job.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_image_job.rs @@ -186,7 +186,8 @@ async fn download_and_store_one_image( index + 1, order.results.len(), order.order_id, image_url, ); - let image_bytes: Vec = reqwest::get(image_url) + let image_bytes: Vec = deps.download_client.get(image_url) + .send() .await .map_err(|err| anyhow!("reqwest error downloading image {}: {:?}", index, err))? .bytes() diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs index 50c2107ba4..43761fe1cb 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs @@ -52,7 +52,7 @@ pub async fn process_successful_video_job( ); // Download the video bytes. - let video_bytes: Vec = match reqwest::get(video_url).await { + let video_bytes: Vec = match deps.download_client.get(video_url).send().await { Ok(resp) => match resp.bytes().await { Ok(bytes) => bytes.to_vec(), Err(err) => { diff --git a/crates/service/job/seedance2_pro_job/src/loop_heartbeats.rs b/crates/service/job/seedance2_pro_job/src/loop_heartbeats.rs new file mode 100644 index 0000000000..f394998dbf --- /dev/null +++ b/crates/service/job/seedance2_pro_job/src/loop_heartbeats.rs @@ -0,0 +1,107 @@ +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +/// Liveness signal for the job's long-running loops. +/// +/// Each loop calls [`beat`](Self::beat) whenever it makes progress (an +/// iteration started, a page polled, an idle tick). The health check compares +/// the age of every registered heartbeat against a staleness threshold, so a +/// loop that is parked forever inside an await with no deadline — the failure +/// mode that previously went unnoticed for hours — turns the pod unhealthy and +/// lets Kubernetes restart it. +/// +/// Cheaply [`Clone`]able: clones share the same underlying map. +#[derive(Clone)] +pub struct LoopHeartbeats { + // NB: A std (sync) RwLock is intentional. Critical sections are tiny map + // operations with no `.await` held across the guard. + inner: Arc>>, +} + +/// One loop's heartbeat age, as reported by the health check. +#[derive(Debug, Clone)] +pub struct HeartbeatAge { + pub loop_name: &'static str, + pub age: Duration, +} + +impl LoopHeartbeats { + pub fn new() -> Self { + Self { + inner: Arc::new(RwLock::new(HashMap::new())), + } + } + + /// Record that the named loop made progress just now. The first call for a + /// name registers it; unregistered loops are never considered stale. + pub fn beat(&self, loop_name: &'static str) { + let mut map = self.inner.write().expect("loop heartbeats lock poisoned"); + map.insert(loop_name, Instant::now()); + } + + /// Age of every registered heartbeat, oldest first. + pub fn ages(&self) -> Vec { + let map = self.inner.read().expect("loop heartbeats lock poisoned"); + let now = Instant::now(); + + let mut ages: Vec = map + .iter() + .map(|(loop_name, last_beat)| HeartbeatAge { + loop_name, + age: now.saturating_duration_since(*last_beat), + }) + .collect(); + + ages.sort_by(|a, b| b.age.cmp(&a.age)); + ages + } + + /// Every registered loop whose last heartbeat is older than `threshold`. + pub fn stale(&self, threshold: Duration) -> Vec { + self.ages().into_iter().filter(|entry| entry.age > threshold).collect() + } +} + +impl Default for LoopHeartbeats { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unregistered_loops_are_never_stale() { + let heartbeats = LoopHeartbeats::new(); + assert!(heartbeats.stale(Duration::ZERO).is_empty()); + } + + #[test] + fn fresh_beat_is_not_stale() { + let heartbeats = LoopHeartbeats::new(); + heartbeats.beat("polling"); + assert!(heartbeats.stale(Duration::from_secs(60)).is_empty()); + assert_eq!(heartbeats.ages().len(), 1); + } + + #[test] + fn beat_older_than_threshold_is_stale() { + let heartbeats = LoopHeartbeats::new(); + heartbeats.beat("polling"); + std::thread::sleep(Duration::from_millis(5)); + let stale = heartbeats.stale(Duration::from_millis(1)); + assert_eq!(stale.len(), 1); + assert_eq!(stale[0].loop_name, "polling"); + } + + #[test] + fn clones_share_state() { + let heartbeats = LoopHeartbeats::new(); + let clone = heartbeats.clone(); + clone.beat("processing"); + assert_eq!(heartbeats.ages().len(), 1); + } +} diff --git a/crates/service/job/seedance2_pro_job/src/main.rs b/crates/service/job/seedance2_pro_job/src/main.rs index 01611b443f..45bbac1a7b 100644 --- a/crates/service/job/seedance2_pro_job/src/main.rs +++ b/crates/service/job/seedance2_pro_job/src/main.rs @@ -13,7 +13,7 @@ use std::sync::Arc; use std::time::Duration; use anyhow::anyhow; -use log::{info, warn}; +use log::{error, info, warn}; use sqlx::mysql::MySqlPoolOptions; use tokio::sync::Notify; @@ -33,6 +33,7 @@ use crate::jobs::credits_checking_job::credits_checking_main_loop::credits_check use crate::jobs::order_polling_job::order_polling_main_loop::order_polling_main_loop; use crate::jobs::order_processing_job::order_processing_main_loop::order_processing_main_loop; use crate::job_dependencies::JobDependencies; +use crate::loop_heartbeats::LoopHeartbeats; use crate::order_reconciler::OrderReconciler; use crate::startup::build_pager::build_pager; use crate::startup::kinovi_setup::{get_kinovi_session, get_kinovi_version}; @@ -42,8 +43,10 @@ pub mod http_server; pub mod job_dependencies; pub mod jobs; pub mod kinovi_version; +pub mod loop_heartbeats; pub mod order_reconciler; pub mod startup; +pub mod with_deadline; // Bucket config const ENV_ACCESS_KEY: &str = "ACCESS_KEY"; @@ -54,6 +57,20 @@ const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL"; const ENV_MAX_JOB_AGE_THRESHOLD_HOURS: &str = "MAX_JOB_AGE_THRESHOLD_HOURS"; +/// Bound on acquiring a pooled MySQL connection (the query itself is bounded +/// separately by `with_deadline`). +const MYSQL_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Recycle idle MySQL connections well inside any middlebox idle cutoff so a +/// silently-dropped socket is far less likely to be handed to a query. +const MYSQL_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); + +/// Deadline for a single media download from Kinovi's CDN. +const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60); + +/// How long to wait for the remaining loops to wind down once one has exited. +const LOOP_SHUTDOWN_GRACE: Duration = Duration::from_secs(30); + #[tokio::main] async fn main() -> AnyhowResult<()> { @@ -75,6 +92,8 @@ async fn main() -> AnyhowResult<()> { let mysql_pool = MySqlPoolOptions::new() .max_connections(2) + .acquire_timeout(MYSQL_ACQUIRE_TIMEOUT) + .idle_timeout(Some(MYSQL_IDLE_TIMEOUT)) .connect(&db_connection_string) .await?; @@ -107,6 +126,10 @@ async fn main() -> AnyhowResult<()> { Some(bucket_timeout), )?; + let download_client = reqwest::Client::builder() + .timeout(DOWNLOAD_TIMEOUT) + .build()?; + let kinovi_version = get_kinovi_version()?; let kinovi_session = get_kinovi_session(kinovi_version)?; @@ -161,17 +184,22 @@ async fn main() -> AnyhowResult<()> { // Shared hand-off between the polling loop (producer) and processing loop (consumer). let order_reconciler = OrderReconciler::new(); + // Liveness signal from every loop, read by the health check. + let heartbeats = LoopHeartbeats::new(); + let pager_for_shutdown = pager.clone(); let create_server_args = CreateServerArgs { container_environment: container_environment.clone(), job_stats: job_stats.clone(), + heartbeats: heartbeats.clone(), pager: pager.clone(), }; let job_dependencies = JobDependencies { mysql_pool, public_bucket_client, + download_client, kinovi_web_session: kinovi_session, kinovi_version, server_environment, @@ -185,6 +213,7 @@ async fn main() -> AnyhowResult<()> { shutdown_notify: shutdown_notify.clone(), pager, order_reconciler, + heartbeats, }; // HTTP server runs on a separate OS thread with its own actix System. @@ -233,16 +262,63 @@ async fn main() -> AnyhowResult<()> { credits_checking_main_loop(credits_deps).await; }); - if kinovi_version.has_characters() { + let maybe_character_handle = if kinovi_version.has_characters() { let character_deps = job_dependencies; - let character_handle = tokio::spawn(async move { + Some(tokio::spawn(async move { character_polling_main_loop(character_deps).await; - }); - let _ = tokio::join!(polling_handle, processing_handle, character_handle, credits_handle); + })) } else { // skip character polling entirely. info!("Alternate mode: character polling is disabled."); - let _ = tokio::join!(polling_handle, processing_handle, credits_handle); + None + }; + + // Wait for the FIRST loop to exit. If shutdown wasn't requested, that loop + // died (panicked, or returned unexpectedly) and the job is silently crippled + // — the other loops would happily keep the process alive forever. Fail fast + // so Kubernetes restarts us instead. + let mut polling_handle = polling_handle; + let mut processing_handle = processing_handle; + let mut credits_handle = credits_handle; + let mut character_handle = maybe_character_handle; + + let (first_exited, first_result) = tokio::select! { + result = &mut polling_handle => ("order polling", result), + result = &mut processing_handle => ("order processing", result), + result = &mut credits_handle => ("credits checking", result), + result = async { + match character_handle.as_mut() { + Some(handle) => handle.await, + None => std::future::pending().await, + } + } => ("character polling", result), + }; + + let was_shutdown_requested = application_shutdown.get(); + + if let Err(join_err) = &first_result { + error!("The {} loop panicked or was cancelled: {:?}", first_exited, join_err); + } else if !was_shutdown_requested { + error!("The {} loop exited without a shutdown request.", first_exited); + } else { + info!("The {} loop exited after shutdown request.", first_exited); + } + + // Make sure the remaining loops wind down, then give them a bounded grace period. + application_shutdown.set(true); + shutdown_notify.notify_waiters(); + + let remaining = async { + let _ = polling_handle.await; + let _ = processing_handle.await; + let _ = credits_handle.await; + if let Some(handle) = character_handle { + let _ = handle.await; + } + }; + + if tokio::time::timeout(LOOP_SHUTDOWN_GRACE, remaining).await.is_err() { + warn!("Remaining loops did not exit within {}s; exiting anyway.", LOOP_SHUTDOWN_GRACE.as_secs()); } info!("Shutting down pager worker..."); @@ -250,6 +326,10 @@ async fn main() -> AnyhowResult<()> { info!("KinoviWeb job exiting."); + if first_result.is_err() || !was_shutdown_requested { + return Err(anyhow!("{} loop terminated unexpectedly; exiting so the pod restarts", first_exited)); + } + Ok(()) } diff --git a/crates/service/job/seedance2_pro_job/src/with_deadline.rs b/crates/service/job/seedance2_pro_job/src/with_deadline.rs new file mode 100644 index 0000000000..6146e0049f --- /dev/null +++ b/crates/service/job/seedance2_pro_job/src/with_deadline.rs @@ -0,0 +1,74 @@ +use std::future::Future; +use std::time::Duration; + +use anyhow::anyhow; +use log::error; + +/// Deadline for a single database round trip (pool acquire + query). +/// +/// sqlx's `acquire_timeout` only bounds acquiring a connection; the query +/// itself has no read timeout and the socket has no TCP keepalive, so a peer +/// that vanishes mid-query (managed-DB failover, a NAT entry expiring) parks +/// the caller forever. This is the outage that stalled the order poller for +/// hours with no log line and no alert. +pub const DATABASE_DEADLINE: Duration = Duration::from_secs(60); + +/// Run `future` with a hard deadline. On timeout, logs with `label` and +/// returns an error that reads like any other failure, so the caller's usual +/// error path (alert, back off, retry next iteration) handles it. +pub async fn with_deadline( + label: &str, + deadline: Duration, + future: F, +) -> anyhow::Result + where + E: Into, + F: Future>, +{ + match tokio::time::timeout(deadline, future).await { + Ok(Ok(value)) => Ok(value), + Ok(Err(err)) => Err(err.into()), + Err(_elapsed) => { + error!("{} did not complete within {}s; abandoning it.", label, deadline.as_secs()); + Err(anyhow!("{} timed out after {}s", label, deadline.as_secs())) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn passes_through_ok() { + let result: anyhow::Result = with_deadline( + "ok", + Duration::from_secs(1), + async { Ok::(7) }, + ).await; + assert_eq!(result.unwrap(), 7); + } + + #[tokio::test] + async fn passes_through_err() { + let result: anyhow::Result = with_deadline( + "err", + Duration::from_secs(1), + async { Err::(anyhow!("boom")) }, + ).await; + assert_eq!(result.unwrap_err().to_string(), "boom"); + } + + #[tokio::test] + async fn reports_timeout() { + let result: anyhow::Result = with_deadline( + "slow query", + Duration::from_millis(10), + async { + tokio::time::sleep(Duration::from_secs(5)).await; + Ok::(1) + }, + ).await; + assert_eq!(result.unwrap_err().to_string(), "slow query timed out after 0s"); + } +}