shouldn't be impacting

This commit is contained in:
Brandon Thomas committed 2026-02-21 09:55:00 +07:00
1 parent 43ec1ae5d7
commit 1f0eb35ff1
7 files changed
+289 -1

No files matched your search

@@ -86,7 +86,6 @@ pub async fn insert_generic_inference_job_for_fal_queue_with_apriori_job_token<'
let maybe_external_third_party = InferenceJobExternalThirdParty::Fal;
let maybe_external_third_party_id = "";
const STATUS : JobStatusPlus = JobStatusPlus::Pending;
let query = sqlx::query!(
@@ -0,0 +1,56 @@
use anyhow::anyhow;
use sqlx::MySqlPool;
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use enums::common::job_status_plus::JobStatusPlus;
use errors::AnyhowResult;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
pub struct MarkJobFailedByTokenArgs<'a> {
pub pool: &'a MySqlPool,
pub job_token: &'a InferenceJobToken,
pub maybe_public_failure_reason: Option<&'a str>,
pub internal_debugging_failure_reason: &'a str,
pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
}
/// Permanently mark an inference job as failed, looked up by its token.
/// Unlike `mark_generic_inference_job_failure`, this does not allow retries —
/// the job will always land in `complete_failure`.
pub async fn mark_job_failed_by_token(args: MarkJobFailedByTokenArgs<'_>) -> AnyhowResult<()> {
let maybe_public_failure_reason = args.maybe_public_failure_reason.map(|reason| {
let mut reason = reason.trim().to_string();
reason.truncate(512);
reason
});
let mut internal_debugging_failure_reason = args.internal_debugging_failure_reason.trim().to_string();
internal_debugging_failure_reason.truncate(512);
const FAILURE_STATUS: &str = JobStatusPlus::CompleteFailure.to_str();
let query_result = sqlx::query!(
r#"
UPDATE generic_inference_jobs
SET
status = ?,
failure_reason = ?,
internal_debugging_failure_reason = ?,
frontend_failure_category = ?,
retry_at = NULL
WHERE token = ?
"#,
FAILURE_STATUS,
maybe_public_failure_reason.as_deref(),
&internal_debugging_failure_reason,
args.maybe_frontend_failure_category,
args.job_token.as_str()
)
.execute(args.pool)
.await;
match query_result {
Err(err) => Err(anyhow!("error with query: {:?}", err)),
Ok(_r) => Ok(()),
}
}
@@ -3,6 +3,7 @@ pub mod count_untried_jobs_of_type;
pub mod list_available_generic_inference_jobs;
pub mod mark_generic_inference_job_completely_failed;
pub mod mark_generic_inference_job_failure;
pub mod mark_job_failed_by_token;
pub mod mark_generic_inference_job_pending_and_grab_lock;
pub mod mark_generic_inference_job_successfully_done;
pub mod reap_stale_fakeyou_jobs;
@@ -1,3 +1,4 @@
pub mod fal;
pub mod job;
pub mod seedance2pro;
pub mod web;
@@ -0,0 +1,137 @@
use anyhow::anyhow;
use log::info;
use sqlx::{Executor, MySql};
use std::marker::PhantomData;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty;
use enums::by_table::generic_inference_jobs::inference_job_product_category::InferenceJobProductCategory;
use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
use enums::common::job_status_plus::JobStatusPlus;
use enums::common::visibility::Visibility;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::prompts::PromptToken;
use tokens::tokens::users::UserToken;
use crate::errors::database_query_error::DatabaseQueryError;
use crate::payloads::generic_inference_args::generic_inference_args::GenericInferenceArgs;
pub struct InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs<'e, 'c, E>
where E: 'e + Executor<'c, Database = MySql>
{
pub uuid_idempotency_token: &'e str,
// NOTE: We'll generate this ahead of time so we can save it with billing info!
pub apriori_job_token: &'e InferenceJobToken,
/// The external order_id returned by the seedance2pro generate_video call.
pub maybe_external_third_party_id: &'e str,
pub maybe_inference_args: Option<GenericInferenceArgs>,
pub maybe_prompt_token: Option<&'e PromptToken>,
pub maybe_creator_user_token: Option<&'e UserToken>,
pub maybe_avt_token: Option<&'e AnonymousVisitorTrackingToken>,
pub creator_ip_address: &'e str,
pub creator_set_visibility: Visibility,
pub mysql_executor: E,
pub phantom: PhantomData<&'c E>,
}
pub async fn insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token<'e, 'c : 'e, E>(
args: InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs<'e, 'c, E>
) -> Result<InferenceJobToken, DatabaseQueryError>
where E: 'e + Executor<'c, Database = MySql>
{
let serialized_args_payload = serde_json::ser::to_string(&args.maybe_inference_args)
.map_err(|_e| anyhow!("could not encode inference args"))?;
const JOB_TYPE: InferenceJobType = InferenceJobType::Seedance2ProQueue;
const INFERENCE_CATEGORY: InferenceCategory = InferenceCategory::VideoGeneration;
const PRODUCT_CATEGORY: InferenceJobProductCategory = InferenceJobProductCategory::Seedance2ProVideo;
const EXTERNAL_THIRD_PARTY: InferenceJobExternalThirdParty = InferenceJobExternalThirdParty::Seedance2Pro;
const STATUS: JobStatusPlus = JobStatusPlus::Pending;
let query = sqlx::query!(
r#"
INSERT INTO generic_inference_jobs
SET
token = ?,
uuid_idempotency_token = ?,
job_type = ?,
maybe_external_third_party = ?,
maybe_external_third_party_id = ?,
product_category = ?,
inference_category = ?,
maybe_model_type = NULL,
maybe_model_token = NULL,
maybe_input_source_token = NULL,
maybe_input_source_token_type = NULL,
maybe_download_url = NULL,
maybe_cover_image_media_file_token = NULL,
maybe_prompt_token = ?,
maybe_raw_inference_text = NULL,
maybe_inference_args = ?,
maybe_creator_user_token = ?,
maybe_creator_anonymous_visitor_token = ?,
creator_ip_address = ?,
creator_set_visibility = ?,
priority_level = 0,
is_keepalive_required = FALSE,
max_duration_seconds = 0,
is_debug_request = FALSE,
maybe_routing_tag = NULL,
status = ?
"#,
args.apriori_job_token.as_str(),
args.uuid_idempotency_token,
JOB_TYPE.to_str(),
EXTERNAL_THIRD_PARTY.to_str(),
args.maybe_external_third_party_id,
PRODUCT_CATEGORY.to_str(),
INFERENCE_CATEGORY.to_str(),
args.maybe_prompt_token.map(|t| t.to_string()),
serialized_args_payload,
args.maybe_creator_user_token.map(|t| t.to_string()),
args.maybe_avt_token.map(|t| t.to_string()),
args.creator_ip_address,
args.creator_set_visibility.to_str(),
STATUS.to_str(),
);
let query_result = query.execute(args.mysql_executor)
.await;
let record_id = match query_result {
Err(err) => return Err(DatabaseQueryError::from(err)),
Ok(res) => res.last_insert_id(),
};
info!("Insert generic inference job for Seedance 2 Pro queue: {} with record ID {}", args.apriori_job_token, record_id);
Ok(args.apriori_job_token.clone())
}
@@ -0,0 +1,92 @@
use anyhow::anyhow;
use log::warn;
use sqlx::MySqlPool;
use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty;
use enums::common::visibility::Visibility;
use errors::AnyhowResult;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::prompts::PromptToken;
use tokens::tokens::users::UserToken;
/// A Seedance2Pro job that is waiting for a result from the external API.
#[derive(Debug)]
pub struct PendingSeedance2ProJob {
pub job_token: InferenceJobToken,
/// The external order_id from seedance2-pro.com.
pub order_id: String,
pub maybe_creator_user_token: Option<UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
pub creator_ip_address: String,
pub creator_set_visibility: Visibility,
pub maybe_prompt_token: Option<PromptToken>,
}
#[derive(Debug, Default)]
struct RawRecord {
job_token: InferenceJobToken,
order_id: Option<String>,
maybe_creator_user_token: Option<UserToken>,
maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
creator_ip_address: String,
creator_set_visibility: Visibility,
maybe_prompt_token: Option<PromptToken>,
}
/// Returns all non-terminal Seedance2Pro jobs that have an associated order_id.
pub async fn list_pending_seedance2pro_jobs(pool: &MySqlPool) -> AnyhowResult<Vec<PendingSeedance2ProJob>> {
let records = sqlx::query_as!(
RawRecord,
r#"
SELECT
jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,
jobs.maybe_external_third_party_id as `order_id`,
jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,
jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,
jobs.creator_ip_address,
jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,
jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`
FROM generic_inference_jobs as jobs
WHERE jobs.maybe_external_third_party = ?
AND jobs.status NOT IN ('complete_success', 'complete_failure')
AND jobs.maybe_external_third_party_id IS NOT NULL
LIMIT 25000
"#,
InferenceJobExternalThirdParty::Seedance2Pro.to_str(),
)
.fetch_all(pool)
.await
.map_err(|err| anyhow!("error querying pending seedance2pro jobs: {:?}", err))?;
let jobs = records
.into_iter()
.filter_map(|record| {
let order_id = match record.order_id {
Some(id) => id,
None => {
warn!("PendingSeedance2ProJob has no order_id, skipping");
return None;
}
};
Some(PendingSeedance2ProJob {
job_token: record.job_token,
order_id,
maybe_creator_user_token: record.maybe_creator_user_token,
maybe_creator_anonymous_visitor_token: record.maybe_creator_anonymous_visitor_token,
creator_ip_address: record.creator_ip_address,
creator_set_visibility: record.creator_set_visibility,
maybe_prompt_token: record.maybe_prompt_token,
})
})
.collect();
Ok(jobs)
}
@@ -0,0 +1,2 @@
pub mod insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token;
pub mod list_pending_seedance2pro_jobs;