diff --git a/crates/service/job/seedance2_pro_job/src/jobs/alert_on_error.rs b/crates/service/job/seedance2_pro_job/src/jobs/alert_on_error.rs new file mode 100644 index 0000000000..9dd1816b3c --- /dev/null +++ b/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( + pager: &Pager, + title: &str, + err: anyhow::Error, +) -> anyhow::Result { + 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) +} diff --git a/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs b/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs index ee42ac595d..e62afcfcd2 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs +++ b/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); diff --git a/crates/service/job/seedance2_pro_job/src/jobs/mod.rs b/crates/service/job/seedance2_pro_job/src/jobs/mod.rs index 17abce4012..f3011e4bc1 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/mod.rs +++ b/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; diff --git a/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_failed_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_failed_job.rs index 80c9ecfbbc..c0c20d7a28 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_failed_job.rs +++ b/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); + } } } diff --git a/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_successful_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_successful_job.rs index 9531bb9e91..419630176b 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/process_job/process_successful_job.rs +++ b/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 = 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 = 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());