mitigations to prevent stuck jobs

- timeouts on db - timeouts on http - job health check loop staleness - fail fast
2026-10-20 17:02:53 -04:00 · 2026-08-28 17:07:07 -04:00
parent 44ff9ce338
commit 9523604a74
29 changed files with 452 additions and 20 deletions
@@ -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))?;

@@ -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))?;

@@ -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<CreateMater

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -5,6 +5,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError;
 use crate::requests::generate_character::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;
@@ -75,6 +76,7 @@ pub async fn generate_character(args: GenerateCharacterArgs<'_>) -> Result<Gener

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -8,6 +8,7 @@ use crate::requests::generate_image::request_types::*;
 use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::utils::categorize_kinovi_web_error::categorize_kinovi_web_error;
 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;
@@ -253,6 +254,7 @@ pub async fn generate_image(args: GenerateImageArgs<'_>) -> Result<GenerateImage

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -7,6 +7,7 @@ use crate::requests::generate_video::request_types::*;
 use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::utils::categorize_kinovi_web_error::categorize_kinovi_web_error;
 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;
@@ -364,6 +365,7 @@ pub async fn generate_video(args: GenerateVideoArgs<'_>) -> Result<GenerateVideo

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -5,6 +5,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError;
 use crate::requests::get_user_auth_details::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;
@@ -42,6 +43,7 @@ pub async fn get_user_auth_details(

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -5,6 +5,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError;
 use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::requests::poll_characters::request_types::*;
 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;
@@ -139,6 +140,7 @@ pub async fn poll_characters(args: PollCharactersArgs<'_>) -> Result<PollCharact

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -6,6 +6,7 @@ use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::requests::poll_orders::failure_reason::FailureReason;
 use crate::requests::poll_orders::request_types::*;
 use crate::utils::common_headers::FIREFOX_USER_AGENT;
+use crate::utils::request_timeouts::KINOVI_REQUEST_TIMEOUT;
 use chrono::{DateTime, Utc};
 use log::info;
 use wreq::Client;
@@ -155,6 +156,7 @@ pub async fn poll_orders(args: PollOrdersArgs<'_>) -> Result<PollOrdersResponse,

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -5,6 +5,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError;
 use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::requests::prepare_file_upload::request_types::*;
 use crate::utils::common_headers::FIREFOX_USER_AGENT;
+use crate::utils::request_timeouts::KINOVI_REQUEST_TIMEOUT;
 use chrono::Utc;
 use log::info;
 use rand::Rng;
@@ -51,6 +52,7 @@ pub async fn prepare_file_upload(args: PrepareFileUploadArgs<'_>) -> Result<Prep

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -5,6 +5,7 @@ use crate::error::kinovi_web_generic_api_error::KinoviWebGenericApiError;
 use crate::requests::update_character::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;
@@ -58,6 +59,7 @@ pub async fn update_character(args: UpdateCharacterArgs<'_>) -> Result<UpdateCha

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -3,6 +3,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_UPLOAD_TIMEOUT;
 use log::info;
 use url::Url;
 use wreq::Client;
@@ -43,6 +44,7 @@ pub async fn upload_file(args: UploadFileArgs) -> Result<UploadFileResponse, Kin

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_UPLOAD_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -7,6 +7,7 @@ use crate::requests::workflow_run_task::request_types::*;
 use crate::requests::kinovi_host::{KinoviHost, resolve_host};
 use crate::utils::categorize_kinovi_web_error::categorize_kinovi_web_error;
 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;
@@ -380,6 +381,7 @@ async fn send_run_task_request<B: serde::Serialize>(

  let client = Client::builder()
    .emulation(Emulation::Firefox143)
+    .timeout(KINOVI_REQUEST_TIMEOUT)
    .build()
    .map_err(|err| KinoviWebClientError::WreqClientError(err))?;

@@ -1,3 +1,4 @@
 pub mod categorize_kinovi_web_error;
 pub mod common_headers;
 pub mod number_coercion;
+pub mod request_timeouts;
@@ -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);
@@ -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
@@ -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<LoopHeartbeatAge>,
+  /// Loops whose heartbeat is older than the staleness threshold.
+  pub stale_loops: Vec<String>,
+}
+
+#[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<String> = 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(),
+  }
+}
@@ -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,
 }
@@ -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<Server> {
      "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,
  };
@@ -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,
 }
@@ -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.
@@ -117,7 +117,8 @@ async fn download_and_create_media_file(
 ) -> AnyhowResult<tokens::tokens::media_files::MediaFileToken> {
  info!("Downloading character image: {}", image_url);

-  let image_bytes: Vec<u8> = reqwest::get(image_url)
+  let image_bytes: Vec<u8> = deps.download_client.get(image_url)
+    .send()
    .await
    .map_err(|err| anyhow!("reqwest error downloading image: {:?}", err))?
    .bytes()
@@ -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;
@@ -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();
      }
@@ -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<u8> = reqwest::get(image_url)
+  let image_bytes: Vec<u8> = deps.download_client.get(image_url)
+    .send()
    .await
    .map_err(|err| anyhow!("reqwest error downloading image {}: {:?}", index, err))?
    .bytes()
@@ -52,7 +52,7 @@ pub async fn process_successful_video_job(
  );

  // Download the video bytes.
-  let video_bytes: Vec<u8> = match reqwest::get(video_url).await {
+  let video_bytes: Vec<u8> = match deps.download_client.get(video_url).send().await {
    Ok(resp) => match resp.bytes().await {
      Ok(bytes) => bytes.to_vec(),
      Err(err) => {
@@ -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<RwLock<HashMap<&'static str, Instant>>>,
+}
+
+/// 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<HeartbeatAge> {
+    let map = self.inner.read().expect("loop heartbeats lock poisoned");
+    let now = Instant::now();
+
+    let mut ages: Vec<HeartbeatAge> = 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<HeartbeatAge> {
+    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);
+  }
+}
@@ -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(())
 }

@@ -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<T, E, F>(
+  label: &str,
+  deadline: Duration,
+  future: F,
+) -> anyhow::Result<T>
+  where
+    E: Into<anyhow::Error>,
+    F: Future<Output = Result<T, E>>,
+{
+  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<u8> = with_deadline(
+      "ok",
+      Duration::from_secs(1),
+      async { Ok::<u8, anyhow::Error>(7) },
+    ).await;
+    assert_eq!(result.unwrap(), 7);
+  }
+
+  #[tokio::test]
+  async fn passes_through_err() {
+    let result: anyhow::Result<u8> = with_deadline(
+      "err",
+      Duration::from_secs(1),
+      async { Err::<u8, anyhow::Error>(anyhow!("boom")) },
+    ).await;
+    assert_eq!(result.unwrap_err().to_string(), "boom");
+  }
+
+  #[tokio::test]
+  async fn reports_timeout() {
+    let result: anyhow::Result<u8> = with_deadline(
+      "slow query",
+      Duration::from_millis(10),
+      async {
+        tokio::time::sleep(Duration::from_secs(5)).await;
+        Ok::<u8, anyhow::Error>(1)
+      },
+    ).await;
+    assert_eq!(result.unwrap_err().to_string(), "slow query timed out after 0s");
+  }
+}