page on seedance jobs (#1266)

also fix threading on video-thumbnail-job

Cargo.lock

@@ -10199,12 +10199,16 @@ dependencies = [
  "jobs_common",
  "log",
  "mysql_queries",
+ "pager",
  "reqwest 0.12.23",
+ "rootly_client",
+ "rootly_config",
  "seedance2pro_client",
  "serde",
  "serde_derive",
  "serde_json",
  "server_environment",
+ "shared_env_var_config",
  "sqlx",
  "tokens",
  "tokio",

crates/schema/config/rootly_config/src/services.rs

@@ -1,2 +1,3 @@
 pub const ROOTLY_SERVICE_ID_STORYTELLER_WEB: &str = "9ebebb09-0c56-4c8a-8f57-d5f0f85f3f16";
+pub const ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB: &str = "2a78f884-c42c-4745-b826-b7a572780ea0";
 pub const ROOTLY_SERVICE_ID_VIDEO_THUMBNAIL_JOB: &str = "9d22997c-4474-47cb-af7b-53119eed1a88";

crates/service/job/seedance2_pro_job/Cargo.toml

@@ -32,8 +32,12 @@ errors.workspace = true
 hashing = { path = "../../../lib/files/hashing" }
 jobs_common = { path = "../../../lib/jobs_common" }
 mysql_queries = { path = "../../../schema/database/mysql_queries" }
+pager = { path = "../../../lib/pager" }
+rootly_client.workspace = true
+rootly_config.workspace = true
 seedance2pro_client = { path = "../../../api_clients/seedance2pro_client" }
 server_environment = { path = "../../../lib/server_environment" }
+shared_env_var_config.workspace = true
 tokens.workspace = true
 
 # External

crates/service/job/seedance2_pro_job/src/http_server/endpoints/health_check_handler.rs

@@ -7,6 +7,8 @@ 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 pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 
 use crate::http_server::http_server_shared_state::HttpServerSharedState;
 
@@ -46,7 +48,7 @@ impl std::fmt::Display for HealthCheckError {
 }
 
 pub async fn get_health_check_handler(
-  _http_request: HttpRequest,
+  http_request: HttpRequest,
   server_state: web::Data<Arc<HttpServerSharedState>>,
 ) -> Result<HttpResponse, HealthCheckError> {
   let job_stats = server_state
@@ -76,6 +78,30 @@ pub async fn get_health_check_handler(
   let is_healthy =
     job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold;
 
+  if !is_healthy {
+    let notification = NotificationDetailsBuilder::from_title(
+          format!("Health check unhealthy on {}", server_state.hostname))
+        .set_description(Some(format!(
+          "Health check returned unhealthy.\n\n\
+             Hostname: {}\n\
+             Consecutive failure count: {}\n\
+             Total failure count: {}\n\
+             Total success count: {}",
+          server_state.hostname,
+          job_stats.consecutive_failure_count,
+          job_stats.total_failure_count,
+          job_stats.total_success_count,
+        )))
+        .set_urgency(Some(NotificationUrgency::High))
+        .set_http_method(Some(http_request.method().to_string()))
+        .set_http_path(Some(http_request.path().to_string()))
+        .build();
+
+    if let Err(err) = server_state.pager.enqueue_page(notification) {
+      error!("Failed to enqueue health check alert: {:?}", err);
+    }
+  }
+
   let response = HealthCheckResponse {
     success: true,
     is_healthy,

crates/service/job/seedance2_pro_job/src/http_server/http_server_shared_state.rs

@@ -1,7 +1,10 @@
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;
 
 #[derive(Clone)]
 pub struct HttpServerSharedState {
   pub job_stats: JobStats,
   pub consecutive_failure_unhealthy_threshold: u64,
+  pub pager: Pager,
+  pub hostname: String,
 }

crates/service/job/seedance2_pro_job/src/http_server/run_http_server.rs

@@ -8,6 +8,7 @@ use log::info;
 use bootstrap::bootstrap::ContainerEnvironment;
 use errors::AnyhowResult;
 use jobs_common::job_stats::JobStats;
+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;
@@ -18,12 +19,13 @@ const DEFAULT_NUM_WORKERS: usize = 2;
 pub struct CreateServerArgs {
   pub container_environment: ContainerEnvironment,
   pub job_stats: JobStats,
+  pub pager: Pager,
 }
 
 pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
   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 _hostname = args.container_environment.hostname.clone();
+  let hostname = args.container_environment.hostname.clone();
 
   let server_state = HttpServerSharedState {
     job_stats: args.job_stats.clone(),
@@ -31,6 +33,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
       "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD",
       3,
     )?,
+    pager: args.pager,
+    hostname,
   };
 
   let server_state_arc = web::Data::new(Arc::new(server_state));

crates/service/job/seedance2_pro_job/src/job_dependencies.rs

@@ -1,6 +1,7 @@
 use cloud_storage::bucket_client::BucketClient;
 use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;
 use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
 use server_environment::ServerEnvironment;
 use sqlx::MySqlPool;
@@ -28,4 +29,7 @@ pub struct JobDependencies {
 
   /// Set to `true` from another thread to trigger graceful shutdown.
   pub application_shutdown: RelaxedAtomicBool,
+
+  /// Pager client for sending alerts.
+  pub pager: Pager,
 }

crates/service/job/seedance2_pro_job/src/jobs/alert_on_error.rs

@@ -0,0 +1,22 @@
+use log::error;
+use pager::client::pager::Pager;
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
+
+/// Enqueue a pager alert for the error, then return it as `Err`.
+pub fn alert_pager_and_return_err<T>(
+  pager: &Pager,
+  title: &str,
+  err: anyhow::Error,
+) -> anyhow::Result<T> {
+  let notification = NotificationDetailsBuilder::from_error(&err)
+      .set_title(title.to_string())
+      .set_urgency(Some(NotificationUrgency::Medium))
+      .build();
+
+  if let Err(pager_err) = pager.enqueue_page(notification) {
+    error!("Failed to enqueue pager alert: {:?}", pager_err);
+  }
+
+  Err(err)
+}

crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs

@@ -1,22 +1,47 @@
 use std::collections::HashMap;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use log::{error, info, warn};
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::list_pending_seedance2pro_jobs;
 use seedance2pro_client::requests::poll_orders::poll_orders::{poll_orders, OrderStatus, PollOrdersArgs, TaskStatus};
 
+use crate::jobs::alert_on_error::alert_pager_and_return_err;
 use crate::jobs::process_page_batch::process_page_batch;
 use crate::job_dependencies::JobDependencies;
 
+const POLL_ALERT_THRESHOLD: Duration = Duration::from_secs(600);
+
 pub async fn main_loop(job_dependencies: JobDependencies) {
   while !job_dependencies.application_shutdown.get() {
+    let start = Instant::now();
     let result = run_poll_iteration(&job_dependencies).await;
+    let elapsed = start.elapsed();
 
     if let Err(err) = result {
       error!("Error in poll iteration: {:?}", err);
+      let _ = alert_pager_and_return_err::<()>(&job_dependencies.pager, "Seedance2Pro poll iteration error", err);
       let _ = job_dependencies.job_stats.increment_failure_count();
     }
 
+    if elapsed > POLL_ALERT_THRESHOLD {
+      warn!("Poll iteration took {:.1}s (threshold: {}s)", elapsed.as_secs_f64(), POLL_ALERT_THRESHOLD.as_secs());
+
+      let notification = NotificationDetailsBuilder::from_title(
+            "Seedance2Pro poll iteration slow".to_string())
+          .set_description(Some(format!(
+            "Poll iteration took {:.1} seconds, exceeding the 10-minute threshold.",
+            elapsed.as_secs_f64(),
+          )))
+          .set_urgency(Some(NotificationUrgency::Medium))
+          .build();
+
+      if let Err(pager_err) = job_dependencies.pager.enqueue_page(notification) {
+        error!("Failed to enqueue slow iteration alert: {:?}", pager_err);
+      }
+    }
+
     tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await;
   }
 
@@ -25,7 +50,13 @@ pub async fn main_loop(job_dependencies: JobDependencies) {
 
 async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
   // 1. Query all non-terminal Seedance2Pro jobs from DB.
-  let pending_jobs = list_pending_seedance2pro_jobs(&deps.mysql_pool).await?;
+  let pending_jobs = match list_pending_seedance2pro_jobs(&deps.mysql_pool).await {
+    Ok(jobs) => jobs,
+    Err(err) => {
+      error!("Failed to list pending seedance2pro jobs: {:?}", err);
+      return alert_pager_and_return_err(&deps.pager, "Seedance2Pro DB query failed", err.into());
+    }
+  };
 
   if pending_jobs.is_empty() {
     info!("No pending Seedance2Pro jobs.");
@@ -52,6 +83,11 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
   let mut pages_in_current_batch: u32 = 0;
 
   loop {
+    if deps.application_shutdown.get() {
+      info!("Shutdown requested during pagination. Stopping early.");
+      break;
+    }
+
     total_page_number += 1;
     pages_in_current_batch += 1;
 
@@ -60,16 +96,21 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
       total_page_number, cursor, total_orders_seen
     );
 
-    let response = poll_orders(PollOrdersArgs {
+    let response = match poll_orders(PollOrdersArgs {
       session: &deps.seedance2pro_session,
       cursor,
       host_override: None,
-    })
-      .await
-      .map_err(|err| {
+    }).await {
+      Ok(r) => r,
+      Err(err) => {
         warn!("Error polling seedance2pro orders: {:?}", err);
-        anyhow::anyhow!("poll_orders failed: {:?}", err)
-      })?;
+        return alert_pager_and_return_err(
+          &deps.pager,
+          "Seedance2Pro API poll failed",
+          anyhow::anyhow!("poll_orders failed: {:?}", err),
+        );
+      }
+    };
 
     let page_count = response.orders.len() as u32;
     info!("Done polling page {}. Got {} orders on this page.", total_page_number, page_count);

crates/service/job/seedance2_pro_job/src/jobs/mod.rs

@@ -1,3 +1,4 @@
+pub mod alert_on_error;
 pub mod main_loop;
 pub mod process_job;
 pub mod process_page_batch;

crates/service/job/seedance2_pro_job/src/jobs/process_job/process_failed_job.rs

@@ -4,6 +4,8 @@ use enums::by_table::generic_inference_jobs::frontend_failure_category::Frontend
 use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
 use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::PendingSeedance2ProJob;
 use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome};
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 use seedance2pro_client::requests::poll_orders::failure_type::FailureType;
 use seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus;
 
@@ -82,6 +84,16 @@ pub async fn process_failed_job(
             job.job_token.as_str(),
             err,
           );
+
+          let notification = NotificationDetailsBuilder::from_error(&err)
+              .set_title("Seedance2Pro refund failed".to_string())
+              .set_urgency(Some(NotificationUrgency::Medium))
+              .build();
+
+          if let Err(pager_err) = deps.pager.enqueue_page(notification) {
+            error!("Failed to enqueue pager alert: {:?}", pager_err);
+          }
+
           let _ = transaction.rollback().await;
           return;
         }
@@ -124,5 +136,14 @@ pub async fn process_failed_job(
       job.job_token.as_str(),
       err
     );
+
+    let notification = NotificationDetailsBuilder::from_error(&err)
+        .set_title("Seedance2Pro mark job failed error".to_string())
+        .set_urgency(Some(NotificationUrgency::Medium))
+        .build();
+
+    if let Err(pager_err) = deps.pager.enqueue_page(notification) {
+      error!("Failed to enqueue pager alert: {:?}", pager_err);
+    }
   }
 }

crates/service/job/seedance2_pro_job/src/jobs/process_job/process_successful_job.rs

@@ -13,6 +13,8 @@ use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedan
 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 seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus;
+
+use crate::jobs::alert_on_error::alert_pager_and_return_err;
 use crate::job_dependencies::JobDependencies;
 
 const PREFIX: &str = "artcraft_";
@@ -47,13 +49,27 @@ pub async fn process_successful_job(
   );
 
   // Download the video bytes.
-  let video_bytes: Vec<u8> = 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();
+  let video_bytes: Vec<u8> = match reqwest::get(video_url).await {
+    Ok(resp) => match resp.bytes().await {
+      Ok(bytes) => bytes.to_vec(),
+      Err(err) => {
+        error!("Error reading video bytes for order {}: {:?}", order.order_id, err);
+        return alert_pager_and_return_err(
+          &deps.pager,
+          "Seedance2Pro video download failed",
+          anyhow!("error reading video bytes: {:?}", err),
+        );
+      }
+    },
+    Err(err) => {
+      error!("Error downloading video for order {}: {:?}", order.order_id, err);
+      return alert_pager_and_return_err(
+        &deps.pager,
+        "Seedance2Pro video download failed",
+        anyhow!("reqwest error downloading video: {:?}", err),
+      );
+    }
+  };
 
   info!(
     "Downloaded {} bytes for order {}",
@@ -76,11 +92,18 @@ pub async fn process_successful_job(
   );
 
   // Upload to public bucket.
-  deps
+  if let Err(err) = 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))?;
+  {
+    error!("Error uploading video for order {}: {:?}", order.order_id, err);
+    return alert_pager_and_return_err(
+      &deps.pager,
+      "Seedance2Pro bucket upload failed",
+      anyhow!("error uploading video to bucket: {:?}", err),
+    );
+  }
 
   info!(
     "Uploaded video for order {}. Creating media file record.",
@@ -92,7 +115,7 @@ pub async fn process_successful_job(
   let maybe_frame_height = order.results.first().map(|r| r.height);
 
   // Insert media file record.
-  let media_file_token = MediaFileInsertBuilder::new()
+  let media_file_token = match 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)
@@ -110,7 +133,17 @@ pub async fn process_successful_job(
     .public_bucket_directory_hash(&bucket_path)
     .insert_pool(&deps.mysql_pool)
     .await
-    .map_err(|err| anyhow!("error inserting media file record: {:?}", err))?;
+  {
+    Ok(token) => token,
+    Err(err) => {
+      error!("Error inserting media file record for order {}: {:?}", order.order_id, err);
+      return alert_pager_and_return_err(
+        &deps.pager,
+        "Seedance2Pro media file insert failed",
+        anyhow!("error inserting media file record: {:?}", err),
+      );
+    }
+  };
 
   info!(
     "Created media file {} for order {}. Marking job {} complete.",
@@ -120,23 +153,21 @@ pub async fn process_successful_job(
   );
 
   // Mark inference job as successfully completed.
-  mark_generic_inference_job_successfully_done_by_token(
+  if let Err(err) = 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)
-    })?;
+  ).await {
+    error!("Error marking job {} done: {:?}", job.job_token.as_str(), err);
+    return alert_pager_and_return_err(
+      &deps.pager,
+      "Seedance2Pro job completion update failed",
+      anyhow!("error marking job done: {:?}", err),
+    );
+  }
 
   info!("Job {} completed successfully.", job.job_token.as_str());
 

crates/service/job/seedance2_pro_job/src/main.rs

@@ -28,10 +28,12 @@ use server_environment::ServerEnvironment;
 use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
 use crate::jobs::main_loop::main_loop;
 use crate::job_dependencies::JobDependencies;
+use crate::startup::build_pager::build_pager;
 
 pub mod http_server;
 pub mod job_dependencies;
 pub mod jobs;
+pub mod startup;
 
 // Bucket config
 const ENV_ACCESS_KEY: &str = "ACCESS_KEY";
@@ -41,7 +43,7 @@ const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME";
 const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL";
 const ENV_SEEDANCE2PRO_COOKIES : &str = "SEEDANCE2PRO_COOKIES";
 
-#[actix_web::main]
+#[tokio::main]
 async fn main() -> AnyhowResult<()> {
 
   let container_environment = bootstrap(BootstrapArgs {
@@ -113,12 +115,27 @@ async fn main() -> AnyhowResult<()> {
     info!("Batch page count: {}", count);
   }
 
+  let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname);
+
+  info!("Spawning pager worker.");
+
+  // NB: The pager worker uses Condvar::wait() which is a blocking syscall.
+  // It must run on a dedicated OS thread, not a tokio task, to avoid blocking
+  // the tokio runtime.
+  std::thread::spawn(move || {
+    let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime");
+    rt.block_on(pager_worker.run());
+  });
+
   let application_shutdown = RelaxedAtomicBool::new(false);
   let job_stats = JobStats::new();
 
+  let pager_for_shutdown = pager.clone();
+
   let create_server_args = CreateServerArgs {
     container_environment: container_environment.clone(),
     job_stats: job_stats.clone(),
+    pager: pager.clone(),
   };
 
   let job_dependencies = JobDependencies {
@@ -130,8 +147,10 @@ async fn main() -> AnyhowResult<()> {
     poll_interval_millis,
     maybe_pages_per_batch,
     application_shutdown: application_shutdown.clone(),
+    pager,
   };
 
+  // HTTP server runs on a separate OS thread with its own actix System.
   std::thread::spawn(move || {
     let actix_runtime = actix_web::rt::System::new();
     let http_server_handle = launch_http_server(create_server_args);
@@ -139,11 +158,30 @@ async fn main() -> AnyhowResult<()> {
     actix_runtime.block_on(http_server_handle)
       .expect("HTTP server should not exit.");
 
-    warn!("Server thread is shut down.");
-    application_shutdown.set(true);
+    warn!("HTTP server thread is shut down.");
+  });
+
+  // Listen for SIGTERM / Ctrl-C to trigger graceful shutdown.
+  let application_shutdown_for_signal = application_shutdown.clone();
+
+  tokio::spawn(async move {
+    match tokio::signal::ctrl_c().await {
+      Ok(()) => {
+        info!("Received shutdown signal. Shutting down...");
+        application_shutdown_for_signal.set(true);
+      }
+      Err(err) => {
+        warn!("Error listening for shutdown signal: {:?}", err);
+      }
+    }
   });
 
   main_loop(job_dependencies).await;
 
+  info!("Shutting down pager worker...");
+  pager_for_shutdown.shutdown_worker();
+
+  info!("Seedance2Pro job exiting.");
+
   Ok(())
 }

crates/service/job/seedance2_pro_job/src/startup/build_pager.rs

@@ -0,0 +1,65 @@
+use log::{info, warn};
+use pager::client::pager::Pager;
+use pager::client::pager_builder::PagerBuilder;
+use pager::worker::pager_worker::PagerWorker;
+use rootly_client::creds::rootly_api_key::RootlyApiKey;
+use rootly_config::services::ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB;
+use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM};
+use shared_env_var_config::paging::{env_enable_paging_default_false, env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type};
+
+pub fn build_pager(
+  server_environment: server_environment::ServerEnvironment,
+  hostname: &str,
+) -> (Pager, PagerWorker) {
+  let is_paging_enabled = env_enable_paging_default_false();
+
+  info!("Paging enabled: {}", is_paging_enabled);
+
+  let environment = if server_environment.is_deployed_in_production() {
+    "production"
+  } else {
+    "development"
+  };
+
+  let builder = PagerBuilder::new()
+      .application_name("seedance2-pro-job".to_string())
+      .environment(environment.to_string())
+      .hostname(hostname.to_string())
+      .service_id(ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB.to_string());
+
+  // If paging is globally disabled, use a NoOp pager regardless of API key.
+  if !is_paging_enabled {
+    warn!("ENABLE_PAGING is false. Pager will be NoOp.");
+    return builder.build_with_worker();
+  }
+
+  let maybe_api_key = env_optional_rootly_api_key();
+
+  match maybe_api_key {
+    Some(api_key) => {
+      info!("Rootly API key found. Configuring pager with Rootly backend.");
+      build_rootly_pager(builder, api_key)
+    }
+    None => {
+      warn!("ROOTLY_API_KEY not set. Pager will not send real pages.");
+      builder.build_with_worker()
+    }
+  }
+}
+
+fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) {
+  let mut rootly_builder = builder
+      .rootly(RootlyApiKey::new(api_key))
+      .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string())
+      .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string())
+      .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string());
+
+  let target_type = env_optional_rootly_notification_target_type();
+  let target_id = env_optional_rootly_notification_target_id();
+
+  if let (Some(t_type), Some(t_id)) = (target_type, target_id) {
+    rootly_builder = rootly_builder.notification_target(t_type, t_id);
+  }
+
+  rootly_builder.build_with_worker()
+}

crates/service/job/seedance2_pro_job/src/startup/mod.rs

@@ -0,0 +1 @@
+pub mod build_pager;

crates/service/job/video_thumbnail_job/src/main.rs

@@ -42,7 +42,7 @@ 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";
 
-#[actix_web::main]
+#[tokio::main]
 async fn main() -> AnyhowResult<()> {
 
   let container_environment = bootstrap(BootstrapArgs {
@@ -151,15 +151,21 @@ async fn main() -> AnyhowResult<()> {
 
   let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname);
 
-  info!("Spawning pager worker thread.");
+  info!("Spawning pager worker.");
 
-  tokio::spawn(async move {
-    pager_worker.run().await;
+  // NB: The pager worker uses Condvar::wait() which is a blocking syscall.
+  // It must run on a dedicated OS thread, not a tokio task, to avoid blocking
+  // the tokio runtime.
+  std::thread::spawn(move || {
+    let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime");
+    rt.block_on(pager_worker.run());
   });
 
   let application_shutdown = RelaxedAtomicBool::new(false);
   let job_stats = JobStats::new();
 
+  let pager_for_shutdown = pager.clone();
+
   let create_server_args = CreateServerArgs {
     container_environment: container_environment.clone(),
     job_stats: job_stats.clone(),
@@ -182,6 +188,7 @@ async fn main() -> AnyhowResult<()> {
     pager,
   };
 
+  // HTTP server runs on a separate OS thread with its own actix System.
   std::thread::spawn(move || {
     let actix_runtime = actix_web::rt::System::new();
     let http_server_handle = launch_http_server(create_server_args);
@@ -189,11 +196,30 @@ async fn main() -> AnyhowResult<()> {
     actix_runtime.block_on(http_server_handle)
       .expect("HTTP server should not exit.");
 
-    warn!("Server thread is shut down.");
-    application_shutdown.set(true);
+    warn!("HTTP server thread is shut down.");
+  });
+
+  // Listen for SIGTERM / Ctrl-C to trigger graceful shutdown.
+  let application_shutdown_for_signal = application_shutdown.clone();
+
+  tokio::spawn(async move {
+    match tokio::signal::ctrl_c().await {
+      Ok(()) => {
+        info!("Received shutdown signal. Shutting down...");
+        application_shutdown_for_signal.set(true);
+      }
+      Err(err) => {
+        warn!("Error listening for shutdown signal: {:?}", err);
+      }
+    }
   });
 
   main_loop(job_dependencies).await;
 
+  info!("Shutting down pager worker...");
+  pager_for_shutdown.shutdown_worker();
+
+  info!("Video thumbnail job exiting.");
+
   Ok(())
 }