better logging

This commit is contained in:
Brandon Thomas committed 2026-06-23 17:21:02 -04:00
1 parent 654f9164d3
commit d398cf4a37
3 files changed
+49 -9

No files matched your search

@@ -1,6 +1,6 @@
{
"db_name": "MySQL",
"query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n jobs.maybe_external_third_party_id as `order_id`,\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,\n jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken`,\n jobs.platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`\n\nFROM generic_inference_jobs as jobs\n\nWHERE jobs.maybe_external_third_party = ?\n AND jobs.job_type = ?\n AND jobs.status NOT IN ('complete_success', 'complete_failure')\n AND jobs.maybe_external_third_party_id IS NOT NULL\n\nLIMIT 25000\n ",
"query": "\nSELECT\n jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n jobs.maybe_external_third_party_id as `order_id`,\n jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n jobs.creator_ip_address,\n jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,\n jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken`,\n jobs.platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,\n jobs.created_at\n\nFROM generic_inference_jobs as jobs\n\nWHERE jobs.maybe_external_third_party = ?\n AND jobs.job_type = ?\n AND jobs.status NOT IN ('complete_success', 'complete_failure')\n AND jobs.maybe_external_third_party_id IS NOT NULL\n\nLIMIT 25000\n ",
"describe": {
"columns": [
{
@@ -92,6 +92,16 @@
"char_set": 224,
"max_size": 64
}
},
{
"ordinal": 9,
"name": "created_at",
"type_info": {
"type": "Timestamp",
"flags": "NOT_NULL | MULTIPLE_KEY | BINARY | TIMESTAMP",
"char_set": 63,
"max_size": 19
}
}
],
"parameters": {
@@ -106,8 +116,9 @@
false,
true,
true,
true
true,
false
]
},
"hash": "ddbbd3886c00268edd3993508f635549fc8ddc877bb059fd58f14b8c07586fc9"
"hash": "1703bc63ce65d98cbcad81d515cbcf211a943c533e38e887d2c12713c78b687d"
}
@@ -1,3 +1,4 @@
use chrono::{DateTime, Utc};
use log::warn;
use sqlx::MySqlPool;
@@ -29,9 +30,13 @@ pub struct PendingSeedance2ProJob {
pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
pub maybe_platform_type: Option<PlatformType>,
/// When the job row was created. Useful for surfacing how long the oldest
/// pending job has been waiting.
pub created_at: DateTime<Utc>,
}
#[derive(Debug, Default)]
#[derive(Debug)]
struct RawRecord {
job_token: InferenceJobToken,
order_id: Option<String>,
@@ -42,6 +47,7 @@ struct RawRecord {
maybe_prompt_token: Option<PromptToken>,
maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
maybe_platform_type: Option<PlatformType>,
created_at: DateTime<Utc>,
}
/// Returns all non-terminal Seedance2Pro video jobs that have an associated order_id.
@@ -62,7 +68,8 @@ SELECT
jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,
jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,
jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken`,
jobs.platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`
jobs.platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,
jobs.created_at
FROM generic_inference_jobs as jobs
@@ -100,6 +107,7 @@ LIMIT 25000
maybe_prompt_token: record.maybe_prompt_token,
maybe_wallet_ledger_entry_token: record.maybe_wallet_ledger_entry_token,
maybe_platform_type: record.maybe_platform_type,
created_at: record.created_at,
})
})
.collect();
@@ -143,6 +143,7 @@ async fn walk_orders(
.collect();
let pending_db_job_count = job_by_order_id.len();
let oldest_pending_created_at = job_by_order_id.values().map(|job| job.created_at).min();
let iteration_start = Instant::now();
let mut result = WalkResult::default();
@@ -159,18 +160,27 @@ async fn walk_orders(
// itself fails, we still have the surrounding context in the logs.
let iteration_elapsed = iteration_start.elapsed();
let pending_db_jobs_display = match oldest_pending_created_at {
Some(oldest) if pending_db_job_count > 0 => format!(
"{} (oldest: {} ago)",
pending_db_job_count,
format_duration_ago(Utc::now() - oldest),
),
_ => pending_db_job_count.to_string(),
};
info!(
"Requesting Kinovi page:\n \
page: {page}\n \
cursor: {cursor:?}\n \
kinovi page: {page}\n \
kinovi cursor: {cursor:?}\n \
orders seen so far: {orders_seen}\n \
pending DB jobs: {pending}\n \
pending DB jobs: {pending_db}\n \
reconciler backlog: {backlog} order(s) awaiting processing\n \
iteration elapsed: {secs}s ({millis}ms)",
page = result.pages_seen,
cursor = cursor,
orders_seen = result.orders_seen,
pending = pending_db_job_count,
pending_db = pending_db_jobs_display,
backlog = deps.order_reconciler.len(),
secs = iteration_elapsed.as_secs(),
millis = iteration_elapsed.as_millis(),
@@ -309,6 +319,17 @@ fn stage_finished_orders(
PageStageSummary { staged }
}
/// Format a positive elapsed duration as `"{h} hours {m} minutes {s} seconds"`
/// (hours are unbounded; e.g. a 50-hour-old job reads `"50 hours ..."`).
/// Negative durations clamp to zero.
fn format_duration_ago(elapsed: chrono::Duration) -> String {
let total_seconds = elapsed.num_seconds().max(0);
let hours = total_seconds / 3600;
let minutes = (total_seconds % 3600) / 60;
let seconds = total_seconds % 60;
format!("{} hours {} minutes {} seconds", hours, minutes, seconds)
}
/// Poll orders from Kinovi with retries. On transient failures, waits with
/// increasing delay (attempt × 2s, capped at `poll_retry_max_delay_millis`).
/// After exhausting retries, alerts the pager and returns an error.