This commit is contained in:
Brandon Thomas committed 2026-06-22 20:32:25 -04:00
1 parent 9aab743116
commit 22ef4a0f11
7 files changed
+96 -67

No files matched your search

@@ -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)
}
@@ -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;
@@ -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.
@@ -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 {
@@ -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
}
}
}
@@ -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,
@@ -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,