retry kinovi polling with backoff

This commit is contained in:
Brandon Thomas committed 2026-04-28 16:49:09 -04:00
1 parent 7f35eec2b4
commit 7d6b99c16e
3 files changed
+69 -17

No files matched your search

@@ -37,6 +37,14 @@ pub struct JobDependencies {
/// ancient orders that will never match a pending job.
pub maybe_max_job_age: Option<Duration>,
/// Maximum number of retries when a poll_orders request fails before
/// alerting the pager and aborting the iteration.
pub poll_max_retries: u32,
/// Maximum delay in milliseconds between poll retries. The delay increases
/// with each attempt up to this cap.
pub poll_retry_max_delay_millis: u64,
/// Page an alert when available Kinovi credits fall below this threshold.
pub credits_alert_threshold: u64,
@@ -6,7 +6,7 @@ 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, PendingSeedance2ProJob};
use seedance2pro_client::requests::poll_orders::poll_orders::{poll_orders, OrderStatus, PollOrdersArgs, TaskStatus};
use seedance2pro_client::requests::poll_orders::poll_orders::{poll_orders, OrderStatus, PollOrdersArgs, PollOrdersResponse, TaskStatus};
use crate::jobs::video_polling_job::alert_on_error::alert_pager_and_return_err;
use crate::jobs::video_polling_job::process_orders_batch::process_orders_batch;
@@ -131,22 +131,7 @@ async fn website_polling_loop(
total_pages_seen, cursor, total_orders_seen
);
let response = match poll_orders(PollOrdersArgs {
session: &deps.seedance2pro_session,
cursor,
host_override: None,
}).await {
Ok(r) => r,
Err(err) => {
warn!("Error polling Kinovi orders: {:?}", err);
return alert_pager_and_return_err(
&deps.pager,
"Kinovi API polling failed",
anyhow::anyhow!("poll_orders failed: {:?}", err),
None,
);
}
};
let response = poll_orders_with_retry(deps, cursor).await?;
let page_orders_count = response.orders.len() as u32;
@@ -232,3 +217,48 @@ async fn website_polling_loop(
})
}
/// 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.
async fn poll_orders_with_retry(
deps: &JobDependencies,
cursor: Option<u64>,
) -> anyhow::Result<PollOrdersResponse> {
let max_retries = deps.poll_max_retries;
for attempt in 1..=max_retries {
match poll_orders(PollOrdersArgs {
session: &deps.seedance2pro_session,
cursor,
host_override: None,
}).await {
Ok(response) => return Ok(response),
Err(err) => {
warn!(
"Error polling Kinovi orders (attempt {}/{}): {:?}",
attempt, max_retries, err
);
if attempt >= max_retries {
return alert_pager_and_return_err(
&deps.pager,
"Kinovi API polling failed after retries",
anyhow::anyhow!("poll_orders failed after {} attempts: {:?}", attempt, err),
None,
);
}
let delay_millis = (attempt as u64 * 2_000).min(deps.poll_retry_max_delay_millis);
tokio::time::sleep(Duration::from_millis(delay_millis)).await;
}
}
}
alert_pager_and_return_err(
&deps.pager,
"Kinovi API polling failed after max retries",
anyhow::anyhow!("poll_orders failed after {} attempts", max_retries),
None,
)
}
@@ -124,6 +124,18 @@ async fn main() -> AnyhowResult<()> {
info!("Max job age threshold: {} hours", duration.num_hours());
}
let poll_max_retries: u32 = easyenv::get_env_num(
"POLL_MAX_RETRIES",
3,
)?;
let poll_retry_max_delay_millis: u64 = easyenv::get_env_num(
"POLL_RETRY_MAX_DELAY_MILLIS",
10_000,
)?;
info!("Poll max retries: {}, max retry delay: {}ms", poll_max_retries, poll_retry_max_delay_millis);
let credits_alert_threshold: u64 = easyenv::get_env_num(
"CREDITS_ALERT_THRESHOLD",
10_000,
@@ -164,6 +176,8 @@ async fn main() -> AnyhowResult<()> {
poll_interval_millis,
maybe_pages_per_batch,
maybe_max_job_age,
poll_max_retries,
poll_retry_max_delay_millis,
credits_alert_threshold,
application_shutdown: application_shutdown.clone(),
shutdown_notify: shutdown_notify.clone(),