seedance2-pro-job alerting

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

@@ -5,6 +5,7 @@ use log::{error, info, warn};
 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;
 
@@ -14,6 +15,7 @@ pub async fn main_loop(job_dependencies: JobDependencies) {
 
     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();
     }
 
@@ -25,7 +27,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.");
@@ -60,16 +68,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());