refactor job

This commit is contained in:
Brandon Thomas committed 2026-02-25 23:52:03 -08:00
1 parent e8d88ca318
commit f092034bec
7 files changed
+66 -65

No files matched your search

@@ -1,2 +0,0 @@
pub mod main_loop;
pub mod process_completed_order;
@@ -1 +0,0 @@
pub mod job_loop;
@@ -26,12 +26,13 @@ use seedance2pro::creds::seedance2pro_session::Seedance2ProSession;
use server_environment::ServerEnvironment;
use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
use crate::job::job_loop::main_loop::main_loop;
use crate::main_loop::main_loop;
use crate::job_dependencies::JobDependencies;
pub mod http_server;
pub mod job;
pub mod job_dependencies;
pub mod main_loop;
pub mod process_job;
// Bucket config
const ENV_ACCESS_KEY: &str = "ACCESS_KEY";
@@ -2,12 +2,11 @@ use std::collections::HashMap;
use std::time::Duration;
use log::{error, info, warn};
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
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::list_pending_seedance2pro_jobs;
use seedance2pro::requests::poll_orders::poll_orders::{poll_orders, PollOrdersArgs, TaskStatus};
use crate::job::job_loop::process_completed_order::{mark_job_failed, process_completed_order};
use crate::process_job::process_failed_job::process_failed_job;
use crate::process_job::process_successful_job::process_successful_job;
use crate::job_dependencies::JobDependencies;
pub async fn main_loop(job_dependencies: JobDependencies) {
@@ -80,7 +79,7 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
order.order_id,
job.job_token.as_str()
);
if let Err(err) = process_completed_order(deps, &job, order).await {
if let Err(err) = process_successful_job(deps, &job, order).await {
warn!(
"Error processing completed order {}: {:?}",
order.order_id, err
@@ -91,43 +90,7 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
}
}
TaskStatus::Failed => {
let reason = order
.fail_reason
.as_deref()
.unwrap_or("unknown failure reason");
let reason_lower = reason.to_lowercase();
let platform_rules_violation = reason_lower.contains("violates") ||
reason_lower.contains("platform rules") ||
reason_lower.contains("please modify");
let frontend_failure_category = if platform_rules_violation {
Some(FrontendFailureCategory::ModelRulesViolation)
} else {
None
};
warn!(
"Order {} failed: {}. Marking job {} failed.",
order.order_id, reason, job.job_token.as_str()
);
let mark_failed_result = mark_job_failed_by_token(MarkJobFailedByTokenArgs {
pool: &deps.mysql_pool,
job_token: &job.job_token,
maybe_public_failure_reason: Some(reason),
internal_debugging_failure_reason: reason,
maybe_frontend_failure_category: frontend_failure_category,
}).await;
if let Err(err) = mark_failed_result {
error!(
"Error marking job {} as failed: {:?}",
job.job_token.as_str(),
err
);
}
process_failed_job(deps, &job, order).await;
}
TaskStatus::Pending | TaskStatus::Processing => {
// Still in progress — check again next poll.
@@ -0,0 +1,2 @@
pub mod process_failed_job;
pub mod process_successful_job;
@@ -0,0 +1,54 @@
use log::{error, warn};
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
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 seedance2pro::requests::poll_orders::poll_orders::OrderStatus;
use crate::job_dependencies::JobDependencies;
pub async fn process_failed_job(
deps: &JobDependencies,
job: &PendingSeedance2ProJob,
order: &OrderStatus,
) {
let reason = order
.fail_reason
.as_deref()
.unwrap_or("unknown failure reason");
let reason_lower = reason.to_lowercase();
let platform_rules_violation = reason_lower.contains("violates") ||
reason_lower.contains("platform rules") ||
reason_lower.contains("please modify");
let frontend_failure_category = if platform_rules_violation {
Some(FrontendFailureCategory::ModelRulesViolation)
} else {
None
};
warn!(
"Order {} failed: {}. Marking job {} failed.",
order.order_id, reason, job.job_token.as_str()
);
let mark_failed_result = mark_job_failed_by_token(MarkJobFailedByTokenArgs {
pool: &deps.mysql_pool,
job_token: &job.job_token,
maybe_public_failure_reason: Some(reason),
internal_debugging_failure_reason: reason,
maybe_frontend_failure_category: frontend_failure_category,
}).await;
if let Err(err) = mark_failed_result {
error!(
"Error marking job {} as failed: {:?}",
job.job_token.as_str(),
err
);
}
// TODO: Refund credits to the user for the failed generation.
}
@@ -9,18 +9,17 @@ use enums::by_table::media_files::media_file_origin_product_category::MediaFileO
use enums::by_table::media_files::media_file_type::MediaFileType;
use errors::AnyhowResult;
use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
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::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::requests::poll_orders::poll_orders::OrderStatus;
use crate::job_dependencies::JobDependencies;
const PREFIX : &str = "artcraft_";
const SUFFIX : &str = ".mp4";
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_completed_order(
pub async fn process_successful_job(
deps: &JobDependencies,
job: &PendingSeedance2ProJob,
order: &OrderStatus,
@@ -143,18 +142,3 @@ pub async fn process_completed_order(
Ok(())
}
/// Mark a job as permanently failed.
pub async fn mark_job_failed(
deps: &JobDependencies,
job: &PendingSeedance2ProJob,
fail_reason: &str,
) -> AnyhowResult<()> {
mark_job_failed_by_token(MarkJobFailedByTokenArgs {
pool: &deps.mysql_pool,
job_token: &job.job_token,
maybe_public_failure_reason: Some(fail_reason),
internal_debugging_failure_reason: fail_reason,
maybe_frontend_failure_category: None,
}).await
}