feat(scheduler): bounded retry for transient Threads API errors + carousel failure enrichment (#257)

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;

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<String>) = 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<titen_core::models::Schedule> =
+        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<String>, Option<String>) = 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");
+}

crates/titen-core/src/error.rs

@@ -65,6 +65,43 @@ impl From<reqwest::Error> 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 <status>)" 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<T> = std::result::Result<T, TitenError>;
 
 #[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());

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<String>,
+    /// 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<String>,
     pub created_at: String,
     pub updated_at: String,
 }

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:?}"
                         )));
                     }
                 }

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 {