From 22ef4a0f116cf56825565d9680914778cda253e5 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 22 Jun 2026 20:32:25 -0400 Subject: [PATCH] refactor --- .../is_job_status_terminal.rs | 8 +++ .../src/jobs/order_processing_job/mod.rs | 12 +--- .../order_processing_main_loop.rs | 2 +- .../process_failed_job.rs | 57 +++++++++++-------- .../order_processing_job/process_one_order.rs | 6 +- .../process_successful_image_job.rs | 39 ++++++++----- ...job.rs => process_successful_video_job.rs} | 39 ++++++++----- 7 files changed, 96 insertions(+), 67 deletions(-) create mode 100644 crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/is_job_status_terminal.rs rename crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/{process_successful_job.rs => process_successful_video_job.rs} (88%) diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/is_job_status_terminal.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/is_job_status_terminal.rs new file mode 100644 index 0000000000..aaeeb4e8cf --- /dev/null +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/is_job_status_terminal.rs @@ -0,0 +1,8 @@ +use enums::common::job_status_plus::JobStatusPlus; + +/// Whether a job's status means it has already been fully settled and must not +/// be re-processed. Matches the terminal set excluded by the pending-jobs +/// lister (`complete_success` / `complete_failure`). +pub fn is_job_status_terminal(status: JobStatusPlus) -> bool { + matches!(status, JobStatusPlus::CompleteSuccess | JobStatusPlus::CompleteFailure) +} diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs index 860df0357e..40ba7f3b20 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs @@ -1,14 +1,6 @@ +pub mod is_job_status_terminal; pub mod order_processing_main_loop; pub mod process_one_order; pub mod process_failed_job; pub mod process_successful_image_job; -pub mod process_successful_job; - -use enums::common::job_status_plus::JobStatusPlus; - -/// Whether a job's status means it has already been fully settled and must not -/// be re-processed. Matches the terminal set excluded by the pending-jobs -/// lister (`complete_success` / `complete_failure`). -pub fn is_job_status_terminal(status: JobStatusPlus) -> bool { - matches!(status, JobStatusPlus::CompleteSuccess | JobStatusPlus::CompleteFailure) -} +pub mod process_successful_video_job; 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 7b0ec34b46..1e39775a6b 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 @@ -4,7 +4,7 @@ use log::{error, info, warn}; use mysql_queries::queries::generic_inference::web::get_inference_job_status_value::get_inference_job_status_value; use crate::job_dependencies::JobDependencies; -use crate::jobs::order_processing_job::is_job_status_terminal; +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; /// How long to nap when the reconciler is empty before peeking again. diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_failed_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_failed_job.rs index 9957e414fb..18bbee0983 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_failed_job.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_failed_job.rs @@ -11,7 +11,7 @@ use seedance2pro_client::requests::poll_orders::failure_type::FailureType; use seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus; use crate::job_dependencies::JobDependencies; -use crate::jobs::order_processing_job::is_job_status_terminal; +use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal; /// Reconcile a failed order: refund the user (if billed) and mark the job /// failed — atomically, in one transaction. @@ -58,28 +58,16 @@ pub async fn process_failed_job( } }; - // Re-check under a row lock: another finalizer may have already settled this - // job between the processing loop's pre-check and now. - match select_inference_job_status_for_update(&mut *transaction, &job.job_token).await { - Ok(Some(status)) if is_job_status_terminal(status) => { - info!( - "Job {} already terminal ({:?}) before failure finalize (order {}); skipping.", - job.job_token.as_str(), status, order.order_id, - ); - let _ = transaction.rollback().await; - return; - } - Ok(Some(_)) => { - // Still pending — proceed. - } - Ok(None) => { - warn!( - "Job {} vanished before failure finalize (order {}); skipping.", - job.job_token.as_str(), order.order_id, - ); - let _ = transaction.rollback().await; - return; - } + // ── Terminal-state guard (do NOT remove) ── + // + // Bail unless the job is still pending. Another finalizer may have already + // settled (and refunded) this job between the processing loop's pre-check and + // this locked re-read. If we don't stop here we could double-refund the user. + // This is the single most important check in this function, so it's a + // discrete, early-returning step. + + let maybe_status = match select_inference_job_status_for_update(&mut *transaction, &job.job_token).await { + Ok(maybe_status) => maybe_status, Err(err) => { error!( "Failed to lock job {} for failure finalize: {:?}. Will retry next poll.", @@ -88,8 +76,31 @@ pub async fn process_failed_job( let _ = transaction.rollback().await; return; } + }; + + let status = match maybe_status { + Some(status) => status, + None => { + warn!( + "Job {} vanished before failure finalize (order {}); skipping.", + job.job_token.as_str(), order.order_id, + ); + let _ = transaction.rollback().await; + return; + } + }; + + if is_job_status_terminal(status) { + info!( + "Job {} is already terminal ({:?}) before failure finalize (order {}); skipping.", + job.job_token.as_str(), status, order.order_id, + ); + let _ = transaction.rollback().await; + return; } + // Still pending — refund and mark failed within the locked transaction. + // // Refund first (within the same transaction), so the refund and the failure // write commit or roll back together. match &job.maybe_wallet_ledger_entry_token { diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs index 64a07986d5..b24d5da19d 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs @@ -6,7 +6,7 @@ use seedance2pro_client::requests::poll_orders::poll_orders::{OrderMediaType, Ta use crate::job_dependencies::JobDependencies; use crate::jobs::order_processing_job::process_failed_job::process_failed_job; use crate::jobs::order_processing_job::process_successful_image_job::process_successful_image_job; -use crate::jobs::order_processing_job::process_successful_job::process_successful_job; +use crate::jobs::order_processing_job::process_successful_video_job::process_successful_video_job; use crate::order_reconciler::OrderDetails; /// Reconcile a single finished order into our system, dispatching on its @@ -26,13 +26,13 @@ pub async fn process_one_order( // both fall to the video handler for back-compat. match &order.media_type { Some(OrderMediaType::Image) => process_successful_image_job(deps, job, order).await, - Some(OrderMediaType::Video) | None => process_successful_job(deps, job, order).await, + Some(OrderMediaType::Video) | None => process_successful_video_job(deps, job, order).await, Some(OrderMediaType::Unknown(other)) => { warn!( "Order {} has unrecognised media_type {:?}; treating as video.", order.order_id, other, ); - process_successful_job(deps, job, order).await + process_successful_video_job(deps, job, order).await } } } 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 2c8a6c3094..b8415feec2 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 @@ -20,7 +20,7 @@ use tokens::tokens::media_files::MediaFileToken; use crate::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; -use crate::jobs::order_processing_job::is_job_status_terminal; +use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal; const PREFIX: &str = "artcraft_"; const SUFFIX: &str = ".png"; @@ -100,7 +100,7 @@ pub async fn process_successful_image_job( ); // Finalize inside a transaction: re-check the job is still pending under a row - // lock, then mark it complete. See `process_successful_job` for the rationale. + // lock, then mark it complete. See `process_successful_video_job` for the rationale. let mut transaction = deps.mysql_pool.begin().await.map_err(|err| { anyhow!("error beginning finalize transaction for job {}: {:?}", job.job_token.as_str(), err) })?; @@ -109,16 +109,16 @@ pub async fn process_successful_image_job( .await .map_err(|err| anyhow!("error locking job {} for finalize: {:?}", job.job_token.as_str(), err))?; - match maybe_status { - Some(status) if is_job_status_terminal(status) => { - warn!( - "Job {} became terminal ({:?}) before finalize; skipping mark-done (order {}). \ - {} media file(s) may be orphaned.", - job.job_token.as_str(), status, order.order_id, created_tokens.len(), - ); - let _ = transaction.rollback().await; - return Ok(()); - } + // ── Terminal-state guard (do NOT remove) ── + // + // Bail unless the job is still pending. A concurrent finalizer (another poll, + // a web cancel) may have settled it between the processing loop's pre-check + // and this locked re-read. If we don't stop here we'd re-mark a finished job + // and leak the just-uploaded media files. This is the single most important + // check in this function, so it's a discrete, early-returning step. + + let status = match maybe_status { + Some(status) => status, None => { let _ = transaction.rollback().await; return Err(anyhow!( @@ -126,11 +126,20 @@ pub async fn process_successful_image_job( job.job_token.as_str(), order.order_id, )); } - Some(_) => { - // Still pending — proceed. - } + }; + + if is_job_status_terminal(status) { + warn!( + "Job {} is already terminal ({:?}); skipping mark-done (order {}). \ + {} media file(s) may be orphaned.", + job.job_token.as_str(), status, order.order_id, created_tokens.len(), + ); + let _ = transaction.rollback().await; + return Ok(()); } + // Still pending — mark it done within the locked transaction. + if let Err(err) = mark_generic_inference_job_successfully_done_by_token_with_executor( &mut *transaction, &job.job_token, diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs similarity index 88% rename from crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_job.rs rename to crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs index bea0dd36f7..d0929884ee 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_job.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_video_job.rs @@ -18,13 +18,13 @@ use seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus; use crate::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; -use crate::jobs::order_processing_job::is_job_status_terminal; +use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal; const PREFIX: &str = "artcraft_"; const SUFFIX: &str = ".mp4"; /// Download the completed video, upload to bucket, create media file record, and mark job done. -pub async fn process_successful_job( +pub async fn process_successful_video_job( deps: &JobDependencies, job: &PendingSeedance2ProJob, order: &OrderStatus, @@ -185,16 +185,16 @@ async fn finalize_success( .await .map_err(|err| anyhow!("error locking job {} for finalize: {:?}", job.job_token.as_str(), err))?; - match maybe_status { - Some(status) if is_job_status_terminal(status) => { - warn!( - "Job {} became terminal ({:?}) before finalize; skipping mark-done (order {}). \ - Media file {} may be orphaned.", - job.job_token.as_str(), status, order.order_id, media_file_token, - ); - let _ = transaction.rollback().await; - return Ok(()); - } + // ── Terminal-state guard (do NOT remove) ── + // + // Bail unless the job is still pending. A concurrent finalizer (another poll, + // a web cancel) may have settled it between the processing loop's pre-check + // and this locked re-read. If we don't stop here we'd re-mark a finished job + // and leak the just-uploaded media file. This is the single most important + // check in this function, so it's a discrete, early-returning step. + + let status = match maybe_status { + Some(status) => status, None => { let _ = transaction.rollback().await; return Err(anyhow!( @@ -202,11 +202,20 @@ async fn finalize_success( job.job_token.as_str(), order.order_id, )); } - Some(_) => { - // Still pending — proceed to mark it done within the locked transaction. - } + }; + + if is_job_status_terminal(status) { + warn!( + "Job {} is already terminal ({:?}); skipping mark-done (order {}). \ + Media file {} may be orphaned.", + job.job_token.as_str(), status, order.order_id, media_file_token, + ); + let _ = transaction.rollback().await; + return Ok(()); } + // Still pending — mark it done within the locked transaction. + if let Err(err) = mark_generic_inference_job_successfully_done_by_token_with_executor( &mut *transaction, &job.job_token,