server/artcraft
1
0
Fork 0
mirror of https://github.com/storytold/artcraft.git synced 2026-10-20 17:02:53 -04:00
Code Issues Packages Projects Releases Wiki Activity

kinovi jobs - better logging (#1351)

Browse Source
This commit is contained in:
Brandon Thomas
2026-04-12 21:24:36 -04:00
committed by GitHub
parent 07d9d537dd
commit cba8e2c074
4 changed files with 138 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/list_pending_seedance2pro_video_jobs.rs
+1 -1
View File
@@ -15,7 +15,7 @@ use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
pub struct PendingSeedance2ProJob {
pub job_token: InferenceJobToken,
/// The external order_id from seedance2-pro.com.
/// Database column `maybe_external_third_party_id` is Kinovi's `order_id`.
pub order_id: String,
pub maybe_creator_user_token: Option<UserToken>,
crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
pub mod alert_on_error;
pub mod video_polling_main_loop;
pub mod process_job;
pub mod process_page_batch;
pub mod process_orders_batch;
crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_page_batch.rs → crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_orders_batch.rs
+37 -5
View File
@@ -12,17 +12,22 @@ use crate::jobs::video_polling_job::process_job::process_successful_job::process
///
/// Removes matched order_ids from `job_by_order_id` so they are not
/// processed again in subsequent batches.
pub async fn process_page_batch(
pub async fn process_orders_batch(
deps: &JobDependencies,
orders: &[OrderStatus],
kinovi_orders: &[OrderStatus],
job_by_order_id: &mut HashMap<String, PendingSeedance2ProJob>,
pages_in_current_batch: u32,
) {
log_batch_summary(&kinovi_orders, pages_in_current_batch);
let pending_job_count = job_by_order_id.len();
let mut batch_succeeded = 0u32;
let mut batch_failed = 0u32;
let mut batch_in_progress = 0u32;
let mut batch_matched = 0u32;
for order in orders {
for order in kinovi_orders {
let job = match job_by_order_id.remove(&order.order_id) {
Some(j) => j,
None => continue, // Not one of our pending jobs.
@@ -64,8 +69,14 @@ pub async fn process_page_batch(
}
info!(
"Batch processing done: {} orders in batch, {} matched pending jobs (succeeded={}, failed={}, in_progress={}), {} pending jobs remaining",
orders.len(),
"Batch processing done: \
{} kinovi orders in polled batch, \
{} pending db jobs in batch at start, \
{} kinovi orders in batch matched pending jobs \
(orders succeeded={}, orders failed={}, orders in_progress={}), \
{} pending db jobs in batch remaining",
kinovi_orders.len(),
pending_job_count,
batch_matched,
batch_succeeded,
batch_failed,
@@ -73,3 +84,24 @@ pub async fn process_page_batch(
job_by_order_id.len(),
);
}
fn log_batch_summary(orders: &[OrderStatus], pages_in_batch: u32) {
let mut succeeded = 0u32;
let mut failed = 0u32;
let mut in_progress = 0u32;
let mut unknown = 0u32;
for order in orders {
match &order.task_status {
TaskStatus::Completed => succeeded += 1,
TaskStatus::Failed => failed += 1,
TaskStatus::Pending | TaskStatus::Processing => in_progress += 1,
TaskStatus::Unknown(_) => unknown += 1,
}
}
info!(
"Processing batch of {} Kinovi order pages, {} total orders (succeeded={}, failed={}, in_progress={}, unknown={})",
pages_in_batch, orders.len(), succeeded, failed, in_progress, unknown
);
}
crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/video_polling_main_loop.rs
+99 -70
View File
@@ -5,24 +5,30 @@ use chrono::Utc;
use log::{error, info, warn};
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
use pager::notification::notification_urgency::NotificationUrgency;
use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_video_jobs::list_pending_seedance2pro_video_jobs;
use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_video_jobs::{list_pending_seedance2pro_video_jobs, PendingSeedance2ProJob};
use seedance2pro_client::requests::poll_orders::poll_orders::{poll_orders, OrderStatus, PollOrdersArgs, TaskStatus};
use crate::jobs::video_polling_job::alert_on_error::alert_pager_and_return_err;
use crate::jobs::video_polling_job::process_page_batch::process_page_batch;
use crate::jobs::video_polling_job::process_orders_batch::process_orders_batch;
use crate::job_dependencies::JobDependencies;
const POLL_ALERT_THRESHOLD: Duration = Duration::from_secs(600);
const POLL_ALERT_THRESHOLD: Duration = Duration::from_mins(6);
pub async fn video_polling_main_loop(job_dependencies: JobDependencies) {
while !job_dependencies.application_shutdown.get() {
let start = Instant::now();
//
// Run a single polling iteration and alert if it takes too long
//
let result = run_poll_iteration(&job_dependencies).await;
let elapsed = start.elapsed();
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, None);
let _ = alert_pager_and_return_err::<()>(&job_dependencies.pager, "Kinovi poll iteration error", err, None);
let _ = job_dependencies.job_stats.increment_failure_count();
}
@@ -30,7 +36,7 @@ pub async fn video_polling_main_loop(job_dependencies: JobDependencies) {
warn!("Poll iteration took {:.1}s (threshold: {}s)", elapsed.as_secs_f64(), POLL_ALERT_THRESHOLD.as_secs());
let notification = NotificationDetailsBuilder::from_title(
"Seedance2Pro poll iteration slow".to_string())
"Kinovi poll iteration slow".to_string())
.set_description(Some(format!(
"Poll iteration took {:.1} seconds, exceeding the 10-minute threshold.",
elapsed.as_secs_f64(),
@@ -46,36 +52,64 @@ pub async fn video_polling_main_loop(job_dependencies: JobDependencies) {
tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await;
}
warn!("Seedance2Pro job runner main loop is shut down.");
warn!("Kinovi job runner main loop is shut down.");
}
async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
// 1. Query all non-terminal Seedance2Pro jobs from DB.
// 1. Query all (limit 25,000) non-terminal Seedance2Pro jobs from DB.
// This is all non-(complete_success, complete_failure) jobs.
let pending_jobs = match list_pending_seedance2pro_video_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(), None);
error!("Failed to list pending database jobs: {:?}", err);
return alert_pager_and_return_err(&deps.pager, "Jobs DB query failed", err.into(), None);
}
};
let total_pending_jobs = pending_jobs.len();
if pending_jobs.is_empty() {
info!("No pending Seedance2Pro jobs.");
info!("No pending database jobs found in the database.");
return Ok(());
}
info!("Found {} pending Seedance2Pro job(s).", pending_jobs.len());
info!("Found {} pending database job(s).", pending_jobs.len());
let result = website_polling_loop(&deps, pending_jobs).await?;
info!(
"Database + Kinovi poll iteration complete: \
{} total database jobs, \
{} total Kinovi pages, \
{} total Kinovi orders seen.",
total_pending_jobs,
result.total_pages_seen,
result.total_orders_seen
);
Ok(())
}
struct WebsitePollingResult {
total_pages_seen: u32,
total_orders_seen: u32,
}
async fn website_polling_loop(
deps: &&JobDependencies,
pending_jobs: Vec<PendingSeedance2ProJob>
) -> anyhow::Result<WebsitePollingResult> {
// Build a lookup: order_id -> job. This is mutated as batches are processed
// so that each order is handled at most once.
let mut job_by_order_id: HashMap<String, _> = pending_jobs
.into_iter()
.map(|job| (job.order_id.clone(), job))
.collect();
.into_iter()
.map(|job| (job.order_id.clone(), job))
.collect();
// 2. Poll orders from seedance2pro API, optionally processing in chunks.
// 2. Poll orders from Kinovi API, optionally processing in chunks.
let mut cursor: Option<u64> = None;
let mut total_page_number: u32 = 0;
let mut total_pages_seen: u32 = 0;
let mut total_orders_seen: u32 = 0;
// Accumulates orders for the current batch (when chunking is enabled)
@@ -89,12 +123,9 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
break;
}
total_page_number += 1;
pages_in_current_batch += 1;
info!(
"Beginning request for page {}... (cursor={:?}, total_orders_so_far={})",
total_page_number, cursor, total_orders_seen
"Beginning request for Kinovi page {}... (cursor: {:?}, total orders seen thus far: {})",
total_pages_seen, cursor, total_orders_seen
);
let response = match poll_orders(PollOrdersArgs {
@@ -104,38 +135,53 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
}).await {
Ok(r) => r,
Err(err) => {
warn!("Error polling seedance2pro orders: {:?}", err);
warn!("Error polling Kinovi orders: {:?}", err);
return alert_pager_and_return_err(
&deps.pager,
"Seedance2Pro API poll failed",
"Kinovi API polling failed",
anyhow::anyhow!("poll_orders failed: {:?}", err),
None,
);
}
};
let page_count = response.orders.len() as u32;
info!("Done polling page {}. Got {} orders on this page.", total_page_number, page_count);
let page_orders_count = response.orders.len() as u32;
total_orders_seen += page_count;
total_pages_seen += 1;
total_orders_seen += page_orders_count;
pages_in_current_batch += 1;
let maybe_last_order_created_at = response.orders
.iter()
.filter(|order| order.created_at_utc.is_some())
.last()
.and_then(|order| order.created_at_utc);
info!(
"Done polling Kinovi orders page {}. \
Got {} orders on this page (oldest on page created at {:?}).",
total_pages_seen,
page_orders_count,
maybe_last_order_created_at,
);
// Check if the last (oldest) order in this page exceeds the max age threshold.
// Orders are returned newest-first, so the last order is the oldest.
let mut exceeded_max_age = false;
if let Some(ref max_age) = deps.maybe_max_job_age {
let maybe_last_order_created= response.orders
.iter()
.filter(|order| order.created_at_utc.is_some())
.last()
.and_then(|order| order.created_at_utc);
if let Some(last_order_created) = maybe_last_order_created {
let order_age = Utc::now() - last_order_created;
if let Some(last_order_created_at) = maybe_last_order_created_at {
let order_age = Utc::now() - last_order_created_at;
let too_old = order_age > *max_age;
if too_old {
info!("Last order on page {} is {} hours old (threshold: {} hours). Stopping pagination.",
total_page_number, order_age.num_hours(), max_age.num_hours());
info!(
"Last order on Kinovi order page {} is {} hours old \
(staleness threshold: {} hours). Stopping Kinovi pagination.",
total_pages_seen,
order_age.num_hours(),
max_age.num_hours()
);
exceeded_max_age = true;
}
}
@@ -144,25 +190,30 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
batch_orders.extend(response.orders);
cursor = response.next_cursor;
let reached_end = cursor.is_none() || exceeded_max_age;
// Determine if we should process the current batch now.
let should_process_batch = reached_end
|| deps.maybe_pages_per_batch
.map(|limit| pages_in_current_batch >= limit)
let batch_is_full = deps.maybe_pages_per_batch
.map(|batch_page_limit| pages_in_current_batch >= batch_page_limit)
.unwrap_or(false);
if should_process_batch && !batch_orders.is_empty() {
log_batch_summary(&batch_orders, pages_in_current_batch);
// Determine if we should process the current batch now.
let should_process_batch_now = reached_end || batch_is_full;
process_page_batch(deps, &batch_orders, &mut job_by_order_id).await;
if should_process_batch_now && !batch_orders.is_empty() {
process_orders_batch(
deps,
&batch_orders,
&mut job_by_order_id,
pages_in_current_batch
).await;
batch_orders.clear();
pages_in_current_batch = 0;
// If all pending jobs have been matched, no need to keep paging.
if job_by_order_id.is_empty() {
info!("All pending jobs have been matched. Stopping pagination early.");
info!("All pending database jobs have been matched to Kinovi orders and processed. Stopping pagination early.");
break;
}
}
@@ -172,31 +223,9 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
}
}
info!(
"Poll iteration complete: {} total pages, {} total orders seen.",
total_page_number, total_orders_seen
);
Ok(())
Ok(WebsitePollingResult {
total_pages_seen,
total_orders_seen
})
}
fn log_batch_summary(orders: &[OrderStatus], pages_in_batch: u32) {
let mut succeeded = 0u32;
let mut failed = 0u32;
let mut in_progress = 0u32;
let mut unknown = 0u32;
for order in orders {
match &order.task_status {
TaskStatus::Completed => succeeded += 1,
TaskStatus::Failed => failed += 1,
TaskStatus::Pending | TaskStatus::Processing => in_progress += 1,
TaskStatus::Unknown(_) => unknown += 1,
}
}
info!(
"Processing batch of {} pages, {} orders (succeeded={}, failed={}, in_progress={}, unknown={})",
pages_in_batch, orders.len(), succeeded, failed, in_progress, unknown
);
}