From 9fc067052b6652de2049e98b6aa4bb0469ec4678 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Thu, 10 Sep 2026 15:18:10 +0700 Subject: [PATCH] feat(scheduler): bounded retry for transient Threads API errors + carousel failure enrichment (#257) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What - MIGRATION 018: schedules.attempt_count (int, default 0) + schedules.next_due_at (nullable RFC3339 UTC eligibility gate). - SCHEDULER: on publish failure, classify the error via TitenError::is_transient_message. Transient + non-bundle + attempts < 2 defer the schedule 10 minutes (pending + [retry N] error marker); otherwise fail permanently with attempt_count persisted. - CLASSIFIER: transient = OAuthException #1 (Meta catch-all, killed carousel child creation on 2026-09-08 and 2026-09-10 while TEXT/IMAGE kept publishing), OAuthException #24 (observed transient 2026-08-25), HTTP 429/5xx. Permanent = #100, #190, validation errors. - GET_DUE_SCHEDULES: next_due_at overrides scheduled_at as the eligibility gate when set — otherwise a retry would fire on the very next tick and the backoff would never hold. NULL keeps legacy behavior. - PUBLISHER: carousel child-container failures now surface WHICH child failed (idx/total), the offending URL, and orphaned child IDs — every prior failure looked identical and required guesswork. - Bundle members are excluded from retry: a chain that breaks mid-flight fails the whole chain (existing semantics preserved). ## Why Production audit of 2026-09-08 + 2026-09-10 carousel failures (OAuthException #1 at child-container creation) showed Meta-side transient errors permanently kill a scheduled slot with zero recovery — the identical content republishes fine on a later attempt. Two back-to-back carousel slots died silently; the same rows failed on two different Titen builds, ruling out a code regression. ## Testing - 7 new unit tests for is_transient_message (transient + permanent cases) - 4 new integration tests (store_retry.rs): deferral sets a future gate and hides the row from get_due_schedules; gate expiry re-exposes it; legacy rows (NULL gate) keep scheduled_at semantics; final failure persists attempt_count and clears the gate. - cargo test --workspace green; clippy -D warnings clean; fmt clean. --- .../migrations/018_schedule_retry.sql | 10 ++ crates/titen-api/tests/store_retry.rs | 135 ++++++++++++++++++ crates/titen-core/src/error.rs | 85 +++++++++++ crates/titen-core/src/models.rs | 8 ++ crates/titen-core/src/publisher.rs | 20 ++- crates/titen-core/src/scheduler.rs | 42 +++++- crates/titen-core/src/store.rs | 76 +++++++++- 7 files changed, 365 insertions(+), 11 deletions(-) create mode 100644 crates/titen-api/migrations/018_schedule_retry.sql create mode 100644 crates/titen-api/tests/store_retry.rs diff --git a/crates/titen-api/migrations/018_schedule_retry.sql b/crates/titen-api/migrations/018_schedule_retry.sql new file mode 100644 index 0000000..9f43fc3 --- /dev/null +++ b/crates/titen-api/migrations/018_schedule_retry.sql @@ -0,0 +1,10 @@ +-- 018 — Transient-error retry for scheduled publishes (#257) +-- +-- Two retry-planning columns on schedules: +-- attempt_count — how many publish attempts have been made (0 = never tried) +-- next_due_at — when the schedule becomes eligible again. NULL = use +-- scheduled_at (original behavior). On transient failure the +-- scheduler backfills this with a +10min delay and resets the +-- row to 'pending'; on permanent failure it stays NULL. +ALTER TABLE schedules ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0; +ALTER TABLE schedules ADD COLUMN next_due_at TEXT; diff --git a/crates/titen-api/tests/store_retry.rs b/crates/titen-api/tests/store_retry.rs new file mode 100644 index 0000000..6a28e19 --- /dev/null +++ b/crates/titen-api/tests/store_retry.rs @@ -0,0 +1,135 @@ +//! #257 — transient-error retry mechanics at the store level. +//! +//! The scheduler's decision logic (which errors are transient, when to give +//! up) is covered by unit tests in `titen-core/src/error.rs`. These tests +//! verify the DB half: deferral backfills `next_due_at` (eligibility gate), +//! `get_due_schedules` honors that gate, and final failures persist the +//! attempt count. + +use sqlx::{ + SqlitePool, + sqlite::{SqliteConnectOptions, SqlitePoolOptions}, +}; +use std::str::FromStr; + +async fn pool() -> SqlitePool { + let options = SqliteConnectOptions::from_str("sqlite::memory:").expect("in-memory sqlite"); + let pool = SqlitePoolOptions::new() + .max_connections(1) // single connection = single in-memory DB + .connect_with(options) + .await + .expect("pool"); + let store = titen_core::Store::new(pool.clone()); + store.migrate().await.expect("migrations"); + pool +} + +async fn seed_pending(pool: &SqlitePool, id: &str) { + // schedules.account_id has an FK to accounts — seed the parent first. + sqlx::query( + "INSERT INTO accounts (id, username, user_id, access_token, expires_at, is_active, created_at, updated_at) \ + VALUES ('acc-1', 'retrytest', '987654321', 'FAKE_TEST_TOKEN', '2026-12-01T00:00:00+00:00', 1, datetime('now'), datetime('now')) \ + ON CONFLICT(id) DO NOTHING", + ) + .execute(pool) + .await + .expect("seed account"); + + // Insert directly: the retry columns are scheduler-owned, and the API + // route has no reason to set them. + sqlx::query( + "INSERT INTO schedules (id, account_id, media_type, caption, scheduled_at, status, attempt_count, created_at, updated_at) \ + VALUES (?, 'acc-1', 'CAROUSEL', 'retry test', '2026-09-09T10:00:00+07:00', 'pending', 0, datetime('now'), datetime('now'))", + ) + .bind(id) + .execute(pool) + .await + .expect("seed schedule"); +} + +#[tokio::test] +async fn defer_sets_future_gate_and_resets_pending() { + let pool = pool().await; + let store = titen_core::Store::new(pool.clone()); + seed_pending(&pool, "sched-defer-1").await; + + store + .defer_failed_schedule("sched-defer-1", 1, 600, "transient boom") + .await + .expect("defer"); + + let row: (String, i64, Option) = sqlx::query_as( + "SELECT status, attempt_count, next_due_at FROM schedules WHERE id = 'sched-defer-1'", + ) + .fetch_one(&pool) + .await + .expect("row"); + + assert_eq!(row.0, "pending"); + assert_eq!(row.1, 1); + let gate = row.2.expect("next_due_at backfilled"); + // Gate must be in the future relative to now (10 min backoff). + let still_due: Vec = + store.get_due_schedules().await.expect("due query"); + assert!( + !still_due.iter().any(|s| s.id == "sched-defer-1"), + "deferred schedule must NOT be due while the gate is in the future (gate={gate})" + ); +} + +#[tokio::test] +async fn deferred_schedule_becomes_due_after_gate_passes() { + let pool = pool().await; + let store = titen_core::Store::new(pool.clone()); + seed_pending(&pool, "sched-defer-2").await; + + store + .defer_failed_schedule("sched-defer-2", 1, 600, "transient boom") + .await + .expect("defer"); + + // Simulate the 10 minutes elapsing. + sqlx::query("UPDATE schedules SET next_due_at = datetime('now', '-1 second') WHERE id = 'sched-defer-2'") + .execute(&pool) + .await + .expect("age gate"); + + let due = store.get_due_schedules().await.expect("due query"); + assert!(due.iter().any(|s| s.id == "sched-defer-2")); +} + +#[tokio::test] +async fn schedules_without_gate_still_use_scheduled_at() { + // Regression guard: rows created before migration 018 (next_due_at NULL) + // must keep the original eligibility semantics. + let pool = pool().await; + let store = titen_core::Store::new(pool.clone()); + seed_pending(&pool, "sched-legacy").await; + + let due = store.get_due_schedules().await.expect("due query"); + assert!(due.iter().any(|s| s.id == "sched-legacy")); +} + +#[tokio::test] +async fn final_failure_persists_attempt_count() { + let pool = pool().await; + let store = titen_core::Store::new(pool.clone()); + seed_pending(&pool, "sched-final-1").await; + + store + .fail_schedule_final("sched-final-1", 2, "gave up after retry") + .await + .expect("fail final"); + + let row: (String, i64, Option, Option) = sqlx::query_as( + "SELECT status, attempt_count, next_due_at, error FROM schedules WHERE id = 'sched-final-1'", + ) + .fetch_one(&pool) + .await + .expect("row"); + + assert_eq!(row.0, "failed"); + assert_eq!(row.1, 2); + assert_eq!(row.2, None, "final failure must not set a retry gate"); + assert_eq!(row.3.expect("error"), "gave up after retry"); +} diff --git a/crates/titen-core/src/error.rs b/crates/titen-core/src/error.rs index 846fda5..b4532d0 100644 --- a/crates/titen-core/src/error.rs +++ b/crates/titen-core/src/error.rs @@ -65,6 +65,43 @@ impl From for TitenError { } } +impl TitenError { + /// Heuristic: does this error message look like a TRANSIENT Threads API + /// failure worth a bounded scheduler retry (#257)? + /// + /// Message-based (not enum-based) because transient errors arrive wrapped: + /// the publisher wraps carousel child-container failures into + /// `InvalidRequest("Failed to create carousel item: Threads API error: …")`, + /// so the OAuthException code only survives in the rendered text. + /// + /// Transient: OAuthException #1 (Meta catch-all "unknown error" — observed + /// killing carousel child creation on 2026-09-08 and 2026-09-10 while TEXT + /// and IMAGE kept publishing), OAuthException #24 ("resource does not + /// exist" — observed transient on 2026-08-25), and server-side HTTP + /// statuses (429/5xx). + /// Permanent: #100 (invalid parameter — our payload is wrong), #190 + /// (auth/token), validation and caption errors. + pub fn is_transient_message(msg: &str) -> bool { + if msg.contains("[OAuthException #1]") || msg.contains("[OAuthException #24]") { + return true; + } + // Server-side HTTP failures surfaced by the threads_post/threads_get + // helpers as "(HTTP )" in the message. + for status in [ + "(HTTP 429", + "(HTTP 500", + "(HTTP 502", + "(HTTP 503", + "(HTTP 504", + ] { + if msg.contains(status) { + return true; + } + } + false + } +} + pub type Result = std::result::Result; #[cfg(test)] @@ -81,6 +118,54 @@ mod tests { assert!(msg.contains("user123")); } + // ─── #257 transient-error classifier ────────────────── + + #[test] + fn transient_oauth_exception_1() { + let msg = "Failed to create carousel item 1/5 (https://x/slide-1.jpg): \ + Threads API error: An unknown error occurred [OAuthException #1] \ + (HTTP 400 Bad Request)"; + assert!(TitenError::is_transient_message(msg)); + } + + #[test] + fn transient_oauth_exception_24() { + let msg = "Threads API error: The requested resource does not exist [OAuthException #24] \ + (HTTP 400 Bad Request)"; + assert!(TitenError::is_transient_message(msg)); + } + + #[test] + fn transient_http_500() { + let msg = "Publish failed: Threads API error: (HTTP 500 Internal Server Error)"; + assert!(TitenError::is_transient_message(msg)); + } + + #[test] + fn transient_http_429() { + let msg = "Threads API error: rate limited (HTTP 429 Too Many Requests)"; + assert!(TitenError::is_transient_message(msg)); + } + + #[test] + fn permanent_oauth_exception_100() { + let msg = "Threads API error: Invalid parameter [OAuthException #100] \ + (HTTP 400 Bad Request)"; + assert!(!TitenError::is_transient_message(msg)); + } + + #[test] + fn permanent_validation_error() { + let msg = "image_url is required for IMAGE posts"; + assert!(!TitenError::is_transient_message(msg)); + } + + #[test] + fn permanent_caption_error() { + let msg = "caption exceeds the 499-character limit (CAPTION_TOO_LONG)"; + assert!(!TitenError::is_transient_message(msg)); + } + #[test] fn display_account_already_exists() { let err = TitenError::AccountAlreadyExists("user123".into()); diff --git a/crates/titen-core/src/models.rs b/crates/titen-core/src/models.rs index 6df241a..2392bfb 100644 --- a/crates/titen-core/src/models.rs +++ b/crates/titen-core/src/models.rs @@ -134,6 +134,14 @@ pub struct Schedule { /// Free-form origin identifier ("cmo-agent", "ci-pipeline", ...) set via /// the ingest endpoint (#242). NULL for dashboard-created schedules. pub source: Option, + /// Number of publish attempts made by the scheduler (#257 retry). + /// 0 for rows created before migration 018 / non-scheduler paths. + #[serde(default)] + pub attempt_count: i64, + /// Retry eligibility timestamp (RFC3339 UTC). NULL = eligibility follows + /// `scheduled_at` (original behavior). Set by the scheduler when a + /// transient publish failure is deferred 10 minutes (#257). + pub next_due_at: Option, pub created_at: String, pub updated_at: String, } diff --git a/crates/titen-core/src/publisher.rs b/crates/titen-core/src/publisher.rs index 7a552f2..8252799 100644 --- a/crates/titen-core/src/publisher.rs +++ b/crates/titen-core/src/publisher.rs @@ -148,21 +148,29 @@ pub async fn publish( .await } "CAROUSEL" => { - let mut children_ids = Vec::with_capacity(req.media_urls.len()); - for url in &req.media_urls { + let total = req.media_urls.len(); + let mut children_ids = Vec::with_capacity(total); + for (i, url) in req.media_urls.iter().enumerate() { match client .create_carousel_item(account, "IMAGE", Some(url.as_str()), None, None) .await { Ok(id) => children_ids.push(id), Err(e) => { + let idx = i + 1; + let ok = children_ids.len(); + // #257: surface WHICH child failed, the offending URL, + // and which child containers already exist. Without + // this every carousel failure looks identical and + // production incidents (2026-09-08, 2026-09-10) require + // guesswork about the failure point. tracing::error!( - "Partial carousel failure after {n} children. \ - Orphaned children IDs (manual cleanup needed): {children_ids:?}", - n = children_ids.len() + "Partial carousel failure at child {idx}/{total} after {ok} ok. \ + Orphaned children IDs (manual cleanup needed): {children_ids:?}" ); return Err(TitenError::InvalidRequest(format!( - "Failed to create carousel item: {e}" + "Failed to create carousel item {idx}/{total} ({url}): {e}. \ + {ok} child container(s) already created: {children_ids:?}" ))); } } diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index f75c458..4237a09 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -1,10 +1,17 @@ -use crate::error::Result; +use crate::error::{Result, TitenError}; use crate::store::Store; use crate::threads_client::ThreadsClient; use std::sync::Arc; use tokio_cron_scheduler::{Job, JobScheduler}; use tracing::{debug, error, info, warn}; +/// #257: how many publish attempts a schedule gets before it is marked +/// permanently `failed`. 2 = initial attempt + one retry. +const MAX_PUBLISH_ATTEMPTS: i64 = 2; + +/// #257: seconds to wait before the transient-error retry becomes eligible. +const RETRY_DELAY_SECS: i64 = 600; + /// Scheduler that ticks every N seconds to check for due schedules pub struct TitenScheduler { scheduler: JobScheduler, @@ -396,9 +403,36 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< ); } Err(e) => { - let _ = store - .update_schedule_status(&schedule.id, "failed", None, Some(&e)) - .await; + // #257: bounded retry for TRANSIENT Threads API failures. + // Production data (2026-09-08 + 2026-09-10 carousel child + // creation, OAuthException #1) shows Meta-side hiccups kill a + // slot permanently while the identical content republishes + // fine later. One retry after 10 minutes, then give up. + let attempt = schedule.attempt_count + 1; + if attempt <= MAX_PUBLISH_ATTEMPTS + && TitenError::is_transient_message(&e) + && schedule.bundle_id.is_none() + { + let delay = RETRY_DELAY_SECS; + warn!( + "Schedule {} hit transient error (attempt {attempt}/{MAX_PUBLISH_ATTEMPTS}) — retrying in {delay}s: {e}", + schedule.id + ); + if let Err(defer_err) = store + .defer_failed_schedule(&schedule.id, attempt, delay, &e) + .await + { + // Deferral failed — fall through to permanent failure + // so the slot never loops on a broken row. + error!( + "Schedule {} defer_failed_schedule failed: {defer_err}", + schedule.id + ); + let _ = store.fail_schedule_final(&schedule.id, attempt, &e).await; + } + } else { + let _ = store.fail_schedule_final(&schedule.id, attempt, &e).await; + } // Thread-bundle: a failed chain item strands the rest — fail // every remaining waiting member so nothing hangs forever. if let Some(ref bundle_id) = schedule.bundle_id { diff --git a/crates/titen-core/src/store.rs b/crates/titen-core/src/store.rs index 811bbff..b2ba60d 100644 --- a/crates/titen-core/src/store.rs +++ b/crates/titen-core/src/store.rs @@ -457,6 +457,21 @@ impl Store { } } + // 018 — transient-error retry: attempt_count + next_due_at columns (#257) + for stmt in split_sql_statements(include_str!( + "../../titen-api/migrations/018_schedule_retry.sql" + )) { + let result = sqlx::query(&stmt).execute(&self.pool).await; + if let Err(e) = result { + let msg = e.to_string(); + if !msg.contains("duplicate column") { + return Err(TitenError::DatabaseError(format!( + "migration 018 failed: {msg}" + ))); + } + } + } + Ok(()) } @@ -964,10 +979,17 @@ impl Store { // // Fix: use datetime(scheduled_at) to parse ISO 8601 → comparable format, // then compare against datetime('now'). + // + // #257: when a transient failure deferred this schedule, next_due_at is + // set and overrides scheduled_at as the eligibility gate — otherwise a + // retry would fire on the very next tick (10min backoff would never hold). sqlx::query_as::<_, Schedule>( "SELECT * FROM schedules \ WHERE status = 'pending' \ - AND datetime(scheduled_at) <= datetime('now') \ + AND (CASE \ + WHEN next_due_at IS NOT NULL THEN datetime(next_due_at) <= datetime('now') \ + ELSE datetime(scheduled_at) <= datetime('now') \ + END) \ ORDER BY scheduled_at ASC", ) .fetch_all(&self.pool) @@ -975,6 +997,58 @@ impl Store { .map_err(Into::into) } + /// Defer a schedule that just hit a transient publish error (#257). + /// + /// Resets the row to `pending` with a +`delay_secs` eligibility gate and + /// increments `attempt_count`. The caller decides retryability first — + /// this method unconditionally defers, so it must only be called when + /// `attempt_count` is still under the cap. `error` is stored on the row + /// (surfaced in the dashboard) with a `[retry N]` marker. + pub async fn defer_failed_schedule( + &self, + id: &str, + attempt: i64, + delay_secs: i64, + error: &str, + ) -> Result<()> { + let marker = format!("[retry {attempt}] {error}"); + sqlx::query( + "UPDATE schedules \ + SET status = 'pending', \ + attempt_count = ?, \ + next_due_at = datetime('now', ?), \ + error = ?, \ + updated_at = datetime('now') \ + WHERE id = ?", + ) + .bind(attempt) + .bind(format!("+{delay_secs} seconds")) + .bind(marker) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Record a final (non-retryable) publish failure (#257). + /// Persists `attempt_count` so the dashboard shows how many tries were made. + pub async fn fail_schedule_final(&self, id: &str, attempt: i64, error: &str) -> Result<()> { + sqlx::query( + "UPDATE schedules \ + SET status = 'failed', \ + attempt_count = ?, \ + error = ?, \ + updated_at = datetime('now') \ + WHERE id = ?", + ) + .bind(attempt) + .bind(error) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + pub async fn create_schedule(&self, id: &str, input: &CreateSchedule) -> Result { self.create_schedule_with_source(id, input, None).await }