From 1c22e272ad17d011682231a5e2ac458ce5971d7b Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 11:15:08 +0700 Subject: [PATCH 1/8] =?UTF-8?q?feat(threads):=20thread=20bundles=20?= =?UTF-8?q?=E2=80=94=20one=20object,=20root=20+=20chained=20replies,=20ins?= =?UTF-8?q?tant=20or=20scheduled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/threads accepts a bundle of 1-20 items. Item 0 is the root post; later items reply to an earlier item by index (reply_to: 0) or to any existing Threads post id (reply_to: id-string). Without scheduled_at everything publishes immediately (root on the next tick, replies chain after it); with scheduled_at the bundle occupies one calendar slot. Storage reuses the schedules table with three new columns (bundle_id, bundle_seq, bundle_total — migration 017) so HITL approval, bookkeeping, calendar listing, and rate tracking all keep working unchanged. Scheduler: a pre-pass promotes due bundle roots (bundle_waiting -> pending); on each successful publish the next member is promoted, and a failed chain item fails the remaining members so nothing strands. Bundle reply markers (bundle::) resolve to the referenced member's real post id at publish time. GET /api/threads/{bundle_id} reports aggregate status (queued/partial/ published/failed) with per-item detail. 229/229 tests, clippy clean. --- .../migrations/017_thread_bundles.sql | 8 + crates/titen-api/src/routes/mod.rs | 1 + crates/titen-api/src/routes/schedules.rs | 3 + crates/titen-api/src/routes/threads_bundle.rs | 303 ++++++++++++++++++ crates/titen-api/src/server.rs | 8 + crates/titen-api/tests/api_schedules.rs | 128 ++++++++ crates/titen-api/tests/common/mod.rs | 8 + crates/titen-core/src/models.rs | 47 +++ crates/titen-core/src/scheduler.rs | 71 +++- crates/titen-core/src/store.rs | 107 ++++++- crates/titen-mcp/src/main.rs | 3 + crates/titen-mcp/tests/mcp_integration.rs | 9 + 12 files changed, 693 insertions(+), 3 deletions(-) create mode 100644 crates/titen-api/migrations/017_thread_bundles.sql create mode 100644 crates/titen-api/src/routes/threads_bundle.rs diff --git a/crates/titen-api/migrations/017_thread_bundles.sql b/crates/titen-api/migrations/017_thread_bundles.sql new file mode 100644 index 0000000..3264582 --- /dev/null +++ b/crates/titen-api/migrations/017_thread_bundles.sql @@ -0,0 +1,8 @@ +-- 017 — thread bundles (#thread-bundle Phase 2) +-- A bundle is N schedule rows sharing bundle_id. seq orders execution; +-- total is the item count (integrity check). Root post is seq 0. +-- Rows keep their own status/result so partial failures are visible per item. +ALTER TABLE schedules ADD COLUMN bundle_id TEXT; +ALTER TABLE schedules ADD COLUMN bundle_seq INTEGER; +ALTER TABLE schedules ADD COLUMN bundle_total INTEGER; +CREATE INDEX IF NOT EXISTS idx_schedules_bundle ON schedules(bundle_id); diff --git a/crates/titen-api/src/routes/mod.rs b/crates/titen-api/src/routes/mod.rs index 6844752..8c4171b 100644 --- a/crates/titen-api/src/routes/mod.rs +++ b/crates/titen-api/src/routes/mod.rs @@ -9,4 +9,5 @@ pub mod posts; pub mod schedules; pub mod settings; pub mod threads; +pub mod threads_bundle; pub mod trends; diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index 49fb745..0c23f69 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -534,6 +534,9 @@ pub async fn ingest_schedule( location_id: input.location_id, auto_approve: false, reply_to_id: input.reply_to_id, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }; // Reuse the same validation surface as the manual create route by diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs new file mode 100644 index 0000000..42c476f --- /dev/null +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -0,0 +1,303 @@ +//! Thread bundles (#thread-bundle Phase 2): one root post plus chained +//! replies, publishable instantly or as a single scheduled slot. + +use axum::{Json, extract::State, http::StatusCode}; +use uuid::Uuid; + +use titen_core::models::CreateSchedule; + +/// Validate a single bundle item's media configuration (cheap, pre-insert). +fn validate_item( + item: &titen_core::models::BundleItem, +) -> Result<(), (StatusCode, Json)> { + let media_type = item.media_type.as_deref().unwrap_or("TEXT"); + let urls: Vec = item.media_urls.clone().unwrap_or_default(); + match media_type { + "TEXT" => { + if item.caption.as_deref().unwrap_or("").trim().is_empty() { + return Err(bad("TEXT items require a caption", "INVALID_ITEM")); + } + } + "IMAGE" => { + if urls.len() != 1 { + return Err(bad( + "IMAGE items require exactly 1 media_urls entry", + "INVALID_ITEM", + )); + } + } + "VIDEO" => { + if urls.len() != 1 { + return Err(bad( + "VIDEO items require exactly 1 media_urls entry", + "INVALID_ITEM", + )); + } + } + "CAROUSEL" => { + if !(2..=20).contains(&urls.len()) { + return Err(bad( + "CAROUSEL items require 2-20 media_urls entries", + "INVALID_ITEM", + )); + } + } + other => { + return Err(bad( + &format!("Unsupported media_type: {other}"), + "INVALID_ITEM", + )); + } + } + // Reply targets must reference an earlier item or an external post. + if let Some(serde_json::Value::Number(n)) = &item.reply_to { + let idx = n.as_i64().unwrap_or(i64::MIN); + if idx < 0 { + return Err(bad("reply_to index must be >= 0", "INVALID_REPLY_TO")); + } + } + Ok(()) +} + +fn bad(message: &str, code: &str) -> (StatusCode, Json) { + ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "error": message, "code": code })), + ) +} + +/// POST /api/threads — create a thread bundle. +/// +/// Without `scheduled_at`: every item is created `pending` with the same +/// "now"-ish timestamp; the scheduler publishes the root, then the chain +/// hooks promote each next item as the previous one publishes. +/// With `scheduled_at`: all items share that slot; root is `pending` +/// (or `draft` unless auto_approve), the rest wait as `bundle_waiting`. +pub async fn create_thread_bundle( + State(state): State, + Json(input): Json, +) -> (StatusCode, Json) { + use titen_core::time::parse_utc; + + let n_items = input.posts.len(); + if n_items == 0 { + return bad("posts must contain at least 1 item", "EMPTY_BUNDLE"); + } + if n_items > 20 { + return bad("posts is limited to 20 items", "TOO_MANY_ITEMS"); + } + // Index-based reply targets must point at earlier items. + for (i, item) in input.posts.iter().enumerate() { + if let Some(serde_json::Value::Number(num)) = &item.reply_to { + let idx = num.as_i64().unwrap_or(i64::MIN); + if idx >= i as i64 { + return bad( + &format!("item {i}: reply_to index {idx} must reference an earlier item"), + "INVALID_REPLY_TO", + ); + } + } + if let Err(resp) = validate_item(item) { + return resp; + } + // caption length: reuse the 499-char guard + if let Some(ref c) = item.caption { + if c.chars().count() > 499 { + return bad( + &format!( + "item {i}: caption exceeds the 499-character limit (got {})", + c.chars().count() + ), + "CAPTION_TOO_LONG", + ); + } + } + } + + // Resolve mode + timestamp + let (scheduled_at, root_status) = match input.scheduled_at.as_deref() { + Some(sa) => { + if parse_utc(sa).is_none() { + return bad( + &format!( + "scheduled_at must be an ISO 8601 / RFC 3339 timestamp with offset, got: {sa}" + ), + "INVALID_SCHEDULED_AT", + ); + } + ( + sa.to_string(), + if input.auto_approve { + "pending" + } else { + "draft" + }, + ) + } + None => { + // Instant mode: everything is pending immediately. Give each item + // its slot = now (root) — chain promotion handles the order. + (chrono::Utc::now().to_rfc3339(), "pending") + } + }; + + let bundle_id = format!("bundle_{}", Uuid::now_v7()); + let mut created = Vec::with_capacity(n_items); + let mut failures = Vec::new(); + + for (i, item) in input.posts.iter().enumerate() { + // Reply target: index -> defer via reply_to_seq marker. The scheduler + // resolves indexes at publish time (the previous item's real post id + // only exists after it publishes). External string ids pass through. + let (reply_to_id, reply_to_seq) = match &item.reply_to { + Some(serde_json::Value::Number(num)) => (None, num.as_i64()), + Some(serde_json::Value::String(s)) if !s.trim().is_empty() => { + (Some(s.trim().to_string()), None) + } + _ => (None, None), + }; + + let status = if i == 0 { + root_status.to_string() + } else { + // Non-root items wait until their predecessor publishes. + // In instant mode the root is already pending, so the first tick + // publishes it and chain-promotes item 1 immediately. + "bundle_waiting".to_string() + }; + + let cs = CreateSchedule { + account_id: input.account_id.clone(), + media_type: Some( + item.media_type + .clone() + .unwrap_or_else(|| "TEXT".to_string()), + ), + caption: item.caption.clone(), + text_attachment: None, + media_urls: item.media_urls.clone(), + // Waiting items get the bundle slot time; promotion overrides order. + scheduled_at: scheduled_at.clone(), + location_id: None, + auto_approve: matches!(status.as_str(), "pending"), + reply_to_id, + bundle_id: Some(bundle_id.clone()), + bundle_seq: Some(i as i64), + bundle_total: Some(n_items as i64), + }; + // reply_to_seq marker: encode into result_json-free dedicated place — + // reuse reply_to_id with a marker prefix the scheduler understands. + let mut cs = cs; + if let Some(seq) = reply_to_seq { + cs.reply_to_id = Some(format!("bundle:{bundle_id}:{seq}")); + } + // Force non-root items to bundle_waiting regardless of auto_approve. + if i > 0 { + cs.auto_approve = false; + } + + let id = Uuid::now_v7().to_string(); + // Response reflects the FINAL per-item status (non-roots are flipped + // to bundle_waiting right after the loop). + let reported_status = if i == 0 { + root_status.to_string() + } else { + "bundle_waiting".to_string() + }; + match state.store.create_schedule(&id, &cs).await { + Ok(schedule) => created.push(serde_json::json!({ + "id": schedule.id, + "seq": i, + "status": reported_status, + "db_status": schedule.status, + })), + Err(e) => failures.push(serde_json::json!({ "seq": i, "error": e.to_string() })), + } + } + + // Race note: create_schedule forces 'draft'/'pending' internally by + // auto_approve. Non-root items must be bundle_waiting — patch them now. + if root_status != "bundle_waiting" { + let _ = state + .store + .set_bundle_status(&bundle_id, "draft", "bundle_waiting") + .await; + let _ = state + .store + .set_bundle_status(&bundle_id, "pending", "bundle_waiting") + .await; + // Restore the root to its intended status. + if let Some(root) = created.first() { + let _ = state + .store + .update_schedule_status(root["id"].as_str().unwrap_or(""), root_status, None, None) + .await; + } + } + + let status = if failures.is_empty() { + StatusCode::CREATED + } else { + StatusCode::MULTI_STATUS + }; + + ( + status, + Json(serde_json::json!({ + "data": { + "bundle_id": bundle_id, + "items": created, + "failures": failures, + "mode": if input.scheduled_at.is_some() { "scheduled" } else { "instant" }, + } + })), + ) +} + +/// GET /api/threads/{bundle_id} — bundle overview (items in seq order). +pub async fn get_thread_bundle( + State(state): State, + axum::extract::Path(bundle_id): axum::extract::Path, +) -> (StatusCode, Json) { + match state.store.get_bundle_schedules(&bundle_id).await { + Ok(items) => { + let published = state + .store + .count_bundle_status(&bundle_id, "published") + .await + .unwrap_or(0); + let failed = state + .store + .count_bundle_status(&bundle_id, "failed") + .await + .unwrap_or(0); + let total = items.len() as i64; + let bundle_status = if failed > 0 { + "failed" + } else if published == total { + "published" + } else if published > 0 { + "partial" + } else { + "queued" + }; + ( + StatusCode::OK, + Json(serde_json::json!({ + "data": { + "bundle_id": bundle_id, + "status": bundle_status, + "published": published, + "failed": failed, + "total": total, + "items": items, + } + })), + ) + } + Err(e) => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": e.to_string(), "code": "BUNDLE_NOT_FOUND" })), + ), + } +} diff --git a/crates/titen-api/src/server.rs b/crates/titen-api/src/server.rs index 02dcd81..0e8589d 100644 --- a/crates/titen-api/src/server.rs +++ b/crates/titen-api/src/server.rs @@ -381,6 +381,14 @@ pub async fn serve( get(routes::posts::get_post).delete(routes::posts::delete_post), ) .route("/api/posts/{id}/insights", get(routes::posts::get_insights)) + .route( + "/api/threads", + post(routes::threads_bundle::create_thread_bundle), + ) + .route( + "/api/threads/{bundle_id}", + get(routes::threads_bundle::get_thread_bundle), + ) .route( "/api/schedules", get(routes::schedules::list_schedules).post(routes::schedules::create_schedule), diff --git a/crates/titen-api/tests/api_schedules.rs b/crates/titen-api/tests/api_schedules.rs index a5a481e..2afb898 100644 --- a/crates/titen-api/tests/api_schedules.rs +++ b/crates/titen-api/tests/api_schedules.rs @@ -177,6 +177,9 @@ async fn list_schedules() { location_id: None, auto_approve: true, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }, ) .await @@ -220,6 +223,9 @@ async fn list_schedules_filters_by_account() { location_id: None, auto_approve: true, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }, ) .await @@ -314,6 +320,9 @@ async fn list_schedules_pagination_meta() { location_id: None, auto_approve: true, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }, ) .await @@ -368,6 +377,125 @@ async fn list_schedules_pagination_meta() { // ─── #232: schedulable thread replies (reply_to_id) ───────────────────── +#[tokio::test] +async fn create_thread_bundle_creates_pending_root_and_waiting_replies() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "scheduled_at": "2099-06-15T12:00:00Z", + "auto_approve": true, + "posts": [ + { "media_type": "TEXT", "caption": "root post" }, + { "media_type": "TEXT", "caption": "reply 1", "reply_to": 0 }, + { "media_type": "IMAGE", "caption": "reply 2 with media", + "media_urls": ["https://s3.ajianaz.dev/example/image.jpg"], + "reply_to": 0 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 201); + + let body = body_to_json(resp).await; + let bundle_id = body["data"]["bundle_id"].as_str().unwrap().to_string(); + let items = body["data"]["items"].as_array().unwrap(); + assert_eq!(items.len(), 3); + assert_eq!(items[0]["status"], "pending"); + assert_eq!(items[1]["status"], "bundle_waiting"); + assert_eq!(items[2]["status"], "bundle_waiting"); + + // GET the bundle back + let req = axum::http::Request::builder() + .method("GET") + .uri(format!("/api/threads/{bundle_id}")) + .body(Body::empty()) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 200); + let body = body_to_json(resp).await; + assert_eq!(body["data"]["total"], 3); + assert_eq!(body["data"]["status"], "queued"); + assert_eq!(body["data"]["items"].as_array().unwrap().len(), 3); +} + +#[tokio::test] +async fn create_thread_bundle_rejects_reply_to_forward_index() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "posts": [ + { "media_type": "TEXT", "caption": "root", + "reply_to": 1 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 400); + let body = body_to_json(resp).await; + assert_eq!(body["code"], "INVALID_REPLY_TO"); +} + +#[tokio::test] +async fn create_thread_bundle_instant_mode_all_items_waiting_but_root() { + let pool = test_pool().await; + let state = test_state(pool.clone()); + let app = test_app(state); + + let account = create_test_account(&app, &pool).await; + let account_id = account["id"].as_str().unwrap(); + + // No scheduled_at => instant mode: root publishes on the next tick. + let req = axum::http::Request::builder() + .method("POST") + .uri("/api/threads") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&json!({ + "account_id": account_id, + "posts": [ + { "media_type": "TEXT", "caption": "instant root" }, + { "media_type": "TEXT", "caption": "instant reply", "reply_to": 0 } + ] + })) + .unwrap(), + )) + .unwrap(); + let resp = send(req, &app).await; + assert_eq!(resp.status(), 201); + + let body = body_to_json(resp).await; + assert_eq!(body["data"]["mode"], "instant"); + let items = body["data"]["items"].as_array().unwrap(); + assert_eq!(items[0]["status"], "pending"); + assert_eq!(items[1]["status"], "bundle_waiting"); +} + #[tokio::test] async fn create_reply_schedule_persists_reply_to_id() { let pool = test_pool().await; diff --git a/crates/titen-api/tests/common/mod.rs b/crates/titen-api/tests/common/mod.rs index c6f91a9..3bc859f 100644 --- a/crates/titen-api/tests/common/mod.rs +++ b/crates/titen-api/tests/common/mod.rs @@ -50,6 +50,14 @@ pub fn test_app(state: AppState) -> Router { "/api/accounts", get(routes::accounts::list_accounts).post(routes::accounts::create_account), ) + .route( + "/api/threads", + post(routes::threads_bundle::create_thread_bundle), + ) + .route( + "/api/threads/{bundle_id}", + get(routes::threads_bundle::get_thread_bundle), + ) .route( "/api/accounts/{id}", put(routes::accounts::update_account).delete(routes::accounts::delete_account), diff --git a/crates/titen-core/src/models.rs b/crates/titen-core/src/models.rs index a0e2971..44826ff 100644 --- a/crates/titen-core/src/models.rs +++ b/crates/titen-core/src/models.rs @@ -127,6 +127,10 @@ pub struct Schedule { /// When set, the scheduler publishes this schedule as a reply to that /// Threads post ID instead of a root-level post (TEXT only). See #232. pub reply_to_id: Option, + /// Thread bundle: rows sharing a bundle_id execute in bundle_seq order. + pub bundle_id: Option, + pub bundle_seq: Option, + pub bundle_total: Option, /// Free-form origin identifier ("cmo-agent", "ci-pipeline", ...) set via /// the ingest endpoint (#242). NULL for dashboard-created schedules. pub source: Option, @@ -151,6 +155,49 @@ pub struct CreateSchedule { /// Publish as a reply to this Threads post ID (TEXT media type only). #[serde(default)] pub reply_to_id: Option, + /// Thread-bundle membership (set by /api/threads expansion, not by + /// clients directly). + #[serde(default)] + pub bundle_id: Option, + #[serde(default)] + pub bundle_seq: Option, + #[serde(default)] + pub bundle_total: Option, +} + +/// One entry inside a thread bundle (`POST /api/threads`). +#[derive(Debug, Deserialize, ToSchema)] +pub struct BundleItem { + /// `TEXT`, `IMAGE`, `VIDEO`, or `CAROUSEL`. Defaults to TEXT. + #[serde(default)] + pub media_type: Option, + #[serde(default)] + pub caption: Option, + /// IMAGE/VIDEO: single URL. CAROUSEL: 2–20 URLs. + #[serde(default)] + pub media_urls: Option>, + #[serde(default)] + pub alt_text: Option, + /// Reply target: an integer = the 0-based index of a previous item in + /// this bundle (0 = root post); a string = an existing Threads post ID. + #[serde(default)] + pub reply_to: Option, +} + +/// `POST /api/threads` — a thread bundle: one root post plus optional +/// chained replies. Without `scheduled_at` everything publishes immediately; +/// with it, one schedule slot executes the whole chain in order. +#[derive(Debug, Deserialize, ToSchema)] +pub struct CreateThreadBundle { + pub account_id: String, + /// ISO 8601 / RFC 3339 timestamp. Optional — absent = publish now. + #[serde(default)] + pub scheduled_at: Option, + /// Skip HITL review (schedule mode only). Default: false. + #[serde(default)] + pub auto_approve: bool, + /// 1..=20 items; item 0 is the root post. + pub posts: Vec, } /// Request body for the agent/CI ingest endpoint (#242). diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index aaa438b..b5292f4 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -125,6 +125,14 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< } } + // Thread-bundle Phase 2: promote bundle members whose ROOT item (seq 0) + // is due. Members 1..n enter as 'bundle_waiting' and only become + // 'pending' once the previous item publishes — this keeps the chain + // ordered while everything shares the existing schedules machinery. + if let Err(e) = store.promote_due_bundle_roots().await { + error!("Failed to promote due bundle roots: {e}"); + } + let due_schedules = match store.get_due_schedules().await { Ok(s) => s, Err(e) => { @@ -260,11 +268,18 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< schedule.location_id.as_deref(), schedule.reply_to_id.as_deref(), ); - let reply_to = req + let mut reply_to = req .reply_to_id .take() .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); + // Bundle chain markers ("bundle::") resolve to the + // real Threads post id of the referenced bundle member. + if let Some(marker) = reply_to.clone() { + if let Some(Some(target_id)) = resolve_bundle_marker(store, &marker).await { + reply_to = Some(target_id); + } + } req.reply_to_id = reply_to.clone(); let result = match crate::publisher::publish(client, &account, &req).await { @@ -324,6 +339,18 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< .update_schedule_status(&schedule.id, "published", Some(&result_str), None) .await; + // Thread-bundle: promote the next waiting member of this bundle. + if let Some(ref bundle_id) = schedule.bundle_id { + let seq = schedule.bundle_seq.unwrap_or(0); + match store.promote_next_bundle_item(bundle_id, seq).await { + Ok(n) if n > 0 => { + info!("Bundle {bundle_id}: promoted {n} item(s) after seq {seq}"); + } + Ok(_) => {} + Err(e) => error!("Bundle {bundle_id} chain advance failed: {e}"), + } + } + info!( "Schedule {} published as post {} for @{}", schedule.id, post_id, account.username @@ -333,6 +360,28 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< let _ = store .update_schedule_status(&schedule.id, "failed", None, Some(&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 { + match store + .fail_remaining_bundle( + bundle_id, + &format!( + "bundle chain stopped at seq {}: {e}", + schedule.bundle_seq.unwrap_or(0) + ), + ) + .await + { + Ok(n) if n > 0 => { + warn!( + "Bundle {bundle_id}: failed {n} remaining item(s) after chain break" + ); + } + Ok(_) => {} + Err(e2) => error!("Bundle {bundle_id} cascade fail failed: {e2}"), + } + } error!("Schedule {} failed: {e}", schedule.id); } } @@ -340,3 +389,23 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< Ok(()) } + +/// Resolve a `bundle::` reply marker to the referenced +/// member's published Threads post id. +/// Returns: +/// - `Some(Some(post_id))` — referenced member published; proceed. +/// - `Some(None)` — referenced member failed; the caller should fail this +/// item (its reply target can never exist). +/// - `None` — not a bundle marker. +async fn resolve_bundle_marker(store: &Store, marker: &str) -> Option> { + let rest = marker.strip_prefix("bundle:")?; + let (bundle_id, seq_str) = rest.rsplit_once(':')?; + let seq: i64 = seq_str.parse().ok()?; + let members = store.get_bundle_schedules(bundle_id).await.ok()?; + let target = members.iter().find(|s| s.bundle_seq == Some(seq))?; + match target.status.as_str() { + "published" => Some(target.result_post_id.clone()), + "failed" | "cancelled" | "rejected" => Some(None), + _ => Some(None), + } +} diff --git a/crates/titen-core/src/store.rs b/crates/titen-core/src/store.rs index fc8263c..b2e4f56 100644 --- a/crates/titen-core/src/store.rs +++ b/crates/titen-core/src/store.rs @@ -442,6 +442,21 @@ impl Store { } } + // 017 — thread bundles: bundle_id/bundle_seq/bundle_total columns + for stmt in split_sql_statements(include_str!( + "../../titen-api/migrations/017_thread_bundles.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") && !msg.contains("already exists") { + return Err(TitenError::DatabaseError(format!( + "migration 017 failed: {msg}" + ))); + } + } + } + Ok(()) } @@ -992,8 +1007,8 @@ impl Store { }; sqlx::query( - "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, source) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, source, bundle_id, bundle_seq, bundle_total) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(id) .bind(&input.account_id) @@ -1006,12 +1021,100 @@ impl Store { .bind(&input.location_id) .bind(&input.reply_to_id) .bind(source) + .bind(&input.bundle_id) + .bind(input.bundle_seq) + .bind(input.bundle_total) .execute(&self.pool) .await?; self.get_schedule(id).await } + /// All bundle members of one bundle, ordered by seq (thread-bundle Phase 2). + pub async fn get_bundle_schedules(&self, bundle_id: &str) -> Result> { + sqlx::query_as::<_, Schedule>( + "SELECT * FROM schedules \ + WHERE bundle_id = ? \ + ORDER BY bundle_seq ASC", + ) + .bind(bundle_id) + .fetch_all(&self.pool) + .await + .map_err(Into::into) + } + + /// Bulk status update for a bundle (approve/reject all members at once). + /// Returns the number of rows transitioned draft -> target status. + pub async fn set_bundle_status(&self, bundle_id: &str, from: &str, to: &str) -> Result { + let result = + sqlx::query("UPDATE schedules SET status = ? WHERE bundle_id = ? AND status = ?") + .bind(to) + .bind(bundle_id) + .bind(from) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Count bundle members still pending/published-root for partial tracking. + pub async fn count_bundle_status(&self, bundle_id: &str, status: &str) -> Result { + let row: (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM schedules WHERE bundle_id = ? AND status = ?") + .bind(bundle_id) + .bind(status) + .fetch_one(&self.pool) + .await?; + Ok(row.0) + } + + /// Promote due bundle roots: seq-0 rows of a bundle whose root + /// scheduled_at has passed move bundle_waiting -> pending so the normal + /// due-schedule path claims them. Idempotent. + pub async fn promote_due_bundle_roots(&self) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE status = 'bundle_waiting' \ + AND bundle_seq = 0 \ + AND datetime(scheduled_at) <= datetime('now')", + ) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Chain advance: when bundle member `seq` publishes, promote the next + /// waiting member(s) whose reply chain resolves to an item at or before + /// `seq`. Member k replies to seq k-1 by design (bundle_reply_to is the + /// *index*); a reply to an external post id (not an index) is promoted + /// immediately after the root publishes. + pub async fn promote_next_bundle_item(&self, bundle_id: &str, seq: i64) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE bundle_id = ? \ + AND status = 'bundle_waiting' \ + AND bundle_seq = ? + 1", + ) + .bind(bundle_id) + .bind(seq) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Fail every remaining waiting member of a bundle (root failed — the + /// chain can never start). + pub async fn fail_remaining_bundle(&self, bundle_id: &str, error: &str) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'failed', error = ? \ + WHERE bundle_id = ? AND status = 'bundle_waiting'", + ) + .bind(error) + .bind(bundle_id) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + pub async fn get_schedule(&self, id: &str) -> Result { sqlx::query_as::<_, Schedule>("SELECT * FROM schedules WHERE id = ?") .bind(id) diff --git a/crates/titen-mcp/src/main.rs b/crates/titen-mcp/src/main.rs index b9e4574..49e40ef 100644 --- a/crates/titen-mcp/src/main.rs +++ b/crates/titen-mcp/src/main.rs @@ -647,6 +647,9 @@ fn handle_tool_call( location_id: None, auto_approve: false, reply_to_id, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }; let id = uuid::Uuid::now_v7().to_string(); diff --git a/crates/titen-mcp/tests/mcp_integration.rs b/crates/titen-mcp/tests/mcp_integration.rs index d44f69e..f676674 100644 --- a/crates/titen-mcp/tests/mcp_integration.rs +++ b/crates/titen-mcp/tests/mcp_integration.rs @@ -72,6 +72,9 @@ async fn create_test_schedule(store: &Store, id: &str, account_id: &str) { location_id: None, auto_approve: true, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }; store .create_schedule(id, &input) @@ -347,6 +350,9 @@ async fn test_approve_draft_schedule() { location_id: None, auto_approve: false, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }; let sched = store .create_schedule("sched-1", &input) @@ -377,6 +383,9 @@ async fn test_reject_draft_schedule() { location_id: None, auto_approve: false, reply_to_id: None, + bundle_id: None, + bundle_seq: None, + bundle_total: None, }; store .create_schedule("sched-1", &input) From 0565605f94c4af8d428ccb3b62ad2f48aaf9f985 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 11:28:26 +0700 Subject: [PATCH 2/8] fix(bundles): race-free state fixup, cascade on early failures and HITL reject CodeCora review findings on PR #252: - The post-loop bulk status flip (set_bundle_status draft/pending -> bundle_waiting + unconditional root restore) raced a concurrent scheduler tick in instant mode: the tick could claim or publish the root before the restore ran, and the restore would then reset processing/published back to pending, re-publishing the post. Non-root rows are now fixed up individually right after insert, scoped to the new row id. - fail_remaining_bundle was only invoked in the publish-Err arm. Roots that failed earlier (account missing / inactive) left the remaining members bundle_waiting forever. Both branches now cascade via a shared cascade_fail_bundle helper. - Rejecting a bundle member via HITL now also fails remaining waiting members so rejected bundles resolve cleanly. 229/229 tests, clippy clean. --- crates/titen-api/src/routes/schedules.rs | 25 ++++++++ crates/titen-api/src/routes/threads_bundle.rs | 59 ++++++++----------- crates/titen-core/src/scheduler.rs | 20 +++++++ 3 files changed, 71 insertions(+), 33 deletions(-) diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index 0c23f69..d8fac86 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -439,6 +439,31 @@ pub async fn reject_schedule( body: Option>, ) -> (StatusCode, Json) { let reason = body.and_then(|b| b.reason.clone()); + // Thread-bundle: rejecting one member strands the rest of the chain — + // fail the remaining waiting members so the bundle resolves cleanly. + if let Ok(schedule) = state.store.get_schedule(&id).await { + if schedule.bundle_id.is_some() { + if let Ok(n) = state + .store + .fail_remaining_bundle( + schedule.bundle_id.as_deref().unwrap_or(""), + &format!( + "bundle chain rejected at seq {}: {}", + schedule.bundle_seq.unwrap_or(0), + reason.as_deref().unwrap_or("no reason") + ), + ) + .await + { + if n > 0 { + tracing::info!( + "Bundle {} rejected: failed {n} remaining item(s)", + schedule.bundle_id.as_deref().unwrap_or("") + ); + } + } + } + } match state.store.reject_schedule(&id, reason.as_deref()).await { Ok(schedule) => ( StatusCode::OK, diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs index 42c476f..7cf22c0 100644 --- a/crates/titen-api/src/routes/threads_bundle.rs +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -197,44 +197,37 @@ pub async fn create_thread_bundle( } let id = Uuid::now_v7().to_string(); - // Response reflects the FINAL per-item status (non-roots are flipped - // to bundle_waiting right after the loop). - let reported_status = if i == 0 { - root_status.to_string() - } else { - "bundle_waiting".to_string() - }; match state.store.create_schedule(&id, &cs).await { - Ok(schedule) => created.push(serde_json::json!({ - "id": schedule.id, - "seq": i, - "status": reported_status, - "db_status": schedule.status, - })), + Ok(schedule) => { + // create_schedule only writes 'draft'/'pending'. Non-root + // items must be 'bundle_waiting': fix THIS row up immediately, + // scoped to its id (a brand-new row created moments ago with + // scheduled_at in the future cannot be claimed by a concurrent + // tick, and the update touches nothing else — no bulk flip, + // no root restore, no race). + if i > 0 { + if let Err(e) = state + .store + .update_schedule_status(&id, "bundle_waiting", None, None) + .await + { + failures.push(serde_json::json!({ + "seq": i, + "error": format!("failed to set bundle_waiting: {e}"), + })); + continue; + } + } + created.push(serde_json::json!({ + "id": schedule.id, + "seq": i, + "status": if i == 0 { root_status } else { "bundle_waiting" }, + })) + } Err(e) => failures.push(serde_json::json!({ "seq": i, "error": e.to_string() })), } } - // Race note: create_schedule forces 'draft'/'pending' internally by - // auto_approve. Non-root items must be bundle_waiting — patch them now. - if root_status != "bundle_waiting" { - let _ = state - .store - .set_bundle_status(&bundle_id, "draft", "bundle_waiting") - .await; - let _ = state - .store - .set_bundle_status(&bundle_id, "pending", "bundle_waiting") - .await; - // Restore the root to its intended status. - if let Some(root) = created.first() { - let _ = state - .store - .update_schedule_status(root["id"].as_str().unwrap_or(""), root_status, None, None) - .await; - } - } - let status = if failures.is_empty() { StatusCode::CREATED } else { diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index b5292f4..614b5d3 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -177,6 +177,7 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< Some(&format!("Account not found: {e}")), ) .await; + cascade_fail_bundle(store, &schedule, &format!("account not found: {e}")).await; continue; } }; @@ -190,6 +191,7 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< let _ = store .update_schedule_status(&schedule.id, "failed", None, Some("Account is inactive")) .await; + cascade_fail_bundle(store, &schedule, "account is inactive").await; continue; } @@ -390,6 +392,24 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< Ok(()) } +/// Fail every remaining waiting member of a bundle when one of its members +/// fails outside the publish-result path (account missing/inactive, HITL +/// reject, etc.). No-op for non-bundle schedules. +async fn cascade_fail_bundle(store: &Store, schedule: &crate::models::Schedule, reason: &str) { + if let Some(ref bundle_id) = schedule.bundle_id { + match store + .fail_remaining_bundle(bundle_id, &format!("bundle chain stopped: {reason}")) + .await + { + Ok(n) if n > 0 => { + warn!("Bundle {bundle_id}: failed {n} remaining item(s) — {reason}"); + } + Ok(_) => {} + Err(e) => error!("Bundle {bundle_id} cascade fail failed: {e}"), + } + } +} + /// Resolve a `bundle::` reply marker to the referenced /// member's published Threads post id. /// Returns: From d92f1e1fe9ae029edaf61a2f7112e1071b187f72 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 11:39:47 +0700 Subject: [PATCH 3/8] fix(bundles): close chain-integrity gaps from CodeCora review round 2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GET /api/threads/{id}: empty result now 404 (was misread as published); DB errors now 500 (was 404). - Bundle creation aborts on the first failed insert and fails all waiting members — a missing seq would strand everything after it (promotion is strictly seq-1 -> seq). Same for a failed bundle_waiting fixup: the row is marked failed, not left pending. - Bundle reply markers now fail the item cleanly (plus cascade) when the referenced member is failed or not published, instead of sending the literal bundle:: marker to the Threads API. - reconcile_bundle_promotions() runs each tick: promotes any waiting member whose predecessor published — self-heals lost/failed promote hooks and crashes between publish and promote. 229/229 tests, clippy clean. --- crates/titen-api/src/routes/threads_bundle.rs | 44 ++++++++++++++++--- crates/titen-core/src/scheduler.rs | 25 ++++++++++- crates/titen-core/src/store.rs | 20 +++++++++ 3 files changed, 80 insertions(+), 9 deletions(-) diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs index 7cf22c0..afe3b32 100644 --- a/crates/titen-api/src/routes/threads_bundle.rs +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -211,10 +211,23 @@ pub async fn create_thread_bundle( .update_schedule_status(&id, "bundle_waiting", None, None) .await { - failures.push(serde_json::json!({ - "seq": i, - "error": format!("failed to set bundle_waiting: {e}"), - })); + let msg = format!("failed to set bundle_waiting: {e}"); + // This row is stuck as draft/pending and would + // duplicate the chain — mark it failed explicitly. + let _ = state + .store + .update_schedule_status(&id, "failed", None, Some(&msg)) + .await; + failures.push(serde_json::json!({ "seq": i, "error": msg })); + // A gap at i strands everything after it (promotion is + // seq-1 -> seq): stop and fail the waiting members. + let _ = state + .store + .fail_remaining_bundle( + &bundle_id, + &format!("bundle creation aborted at seq {i}"), + ) + .await; continue; } } @@ -224,7 +237,19 @@ pub async fn create_thread_bundle( "status": if i == 0 { root_status } else { "bundle_waiting" }, })) } - Err(e) => failures.push(serde_json::json!({ "seq": i, "error": e.to_string() })), + Err(e) => { + failures.push(serde_json::json!({ "seq": i, "error": e.to_string() })); + // A gap at i would strand every later item: stop creating and + // fail the waiting members created so far. + let _ = state + .store + .fail_remaining_bundle( + &bundle_id, + &format!("bundle creation aborted at seq {i}: {e}"), + ) + .await; + break; + } } } @@ -253,6 +278,10 @@ pub async fn get_thread_bundle( axum::extract::Path(bundle_id): axum::extract::Path, ) -> (StatusCode, Json) { match state.store.get_bundle_schedules(&bundle_id).await { + Ok(items) if items.is_empty() => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "bundle not found", "code": "BUNDLE_NOT_FOUND" })), + ), Ok(items) => { let published = state .store @@ -288,9 +317,10 @@ pub async fn get_thread_bundle( })), ) } + // DB error is a server fault — 500, not 404. Err(e) => ( - StatusCode::NOT_FOUND, - Json(serde_json::json!({ "error": e.to_string(), "code": "BUNDLE_NOT_FOUND" })), + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": e.to_string(), "code": "INTERNAL" })), ), } } diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index 614b5d3..6a1c576 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -132,6 +132,12 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< if let Err(e) = store.promote_due_bundle_roots().await { error!("Failed to promote due bundle roots: {e}"); } + // Reconcile: promote any waiting member whose predecessor published, + // regardless of how publication happened (covers lost promote hooks, + // hook errors, and crashes between publish and promote). + if let Err(e) = store.reconcile_bundle_promotions().await { + error!("Failed to reconcile bundle promotions: {e}"); + } let due_schedules = match store.get_due_schedules().await { Ok(s) => s, @@ -278,8 +284,23 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< // Bundle chain markers ("bundle::") resolve to the // real Threads post id of the referenced bundle member. if let Some(marker) = reply_to.clone() { - if let Some(Some(target_id)) = resolve_bundle_marker(store, &marker).await { - reply_to = Some(target_id); + if marker.starts_with("bundle:") { + match resolve_bundle_marker(store, &marker).await { + Some(Some(target_id)) => { + reply_to = Some(target_id); + } + // Target failed or not yet published: fail this item + // cleanly instead of sending a literal marker to Threads. + Some(None) => { + let msg = format!("bundle reply target unavailable: {marker}"); + let _ = store + .update_schedule_status(&schedule.id, "failed", None, Some(&msg)) + .await; + cascade_fail_bundle(store, &schedule, &msg).await; + continue; + } + None => {} + } } } req.reply_to_id = reply_to.clone(); diff --git a/crates/titen-core/src/store.rs b/crates/titen-core/src/store.rs index b2e4f56..5c81717 100644 --- a/crates/titen-core/src/store.rs +++ b/crates/titen-core/src/store.rs @@ -1101,6 +1101,26 @@ impl Store { Ok(result.rows_affected()) } + /// Reconcile bundle chains: promote waiting members whose predecessor + /// (seq-1) is published. Self-healing — covers lost promote hooks, + /// hook errors, and crashes between publish and promote. Idempotent. + pub async fn reconcile_bundle_promotions(&self) -> Result { + let result = sqlx::query( + "UPDATE schedules SET status = 'pending' \ + WHERE status = 'bundle_waiting' \ + AND bundle_seq > 0 \ + AND EXISTS ( \ + SELECT 1 FROM schedules prev \ + WHERE prev.bundle_id = schedules.bundle_id \ + AND prev.bundle_seq = schedules.bundle_seq - 1 \ + AND prev.status = 'published' \ + )", + ) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + /// Fail every remaining waiting member of a bundle (root failed — the /// chain can never start). pub async fn fail_remaining_bundle(&self, bundle_id: &str, error: &str) -> Result { From ff5f2ddacb7696edee4f6b7fbee8099be22e3552 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 11:53:01 +0700 Subject: [PATCH 4/8] fix(bundles): address CodeCora review round 3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Reject cascade now runs only after reject_schedule succeeds (Ok arm), so invalid/failed rejects can no longer wipe waiting bundle members. - bundle_waiting fixup failure now BREAKS creation instead of continuing (later rows would be stranded behind a failed seq). - Bundle reply marker lookup failure (bad seq / store error / missing member) now fails the item cleanly, same as an unavailable target — the literal marker is never sent to Threads. 229/229 tests, clippy clean. --- crates/titen-api/src/routes/schedules.rs | 57 ++++++++++--------- crates/titen-api/src/routes/threads_bundle.rs | 2 +- crates/titen-core/src/scheduler.rs | 8 ++- 3 files changed, 37 insertions(+), 30 deletions(-) diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index d8fac86..ad63736 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -439,36 +439,39 @@ pub async fn reject_schedule( body: Option>, ) -> (StatusCode, Json) { let reason = body.and_then(|b| b.reason.clone()); - // Thread-bundle: rejecting one member strands the rest of the chain — - // fail the remaining waiting members so the bundle resolves cleanly. - if let Ok(schedule) = state.store.get_schedule(&id).await { - if schedule.bundle_id.is_some() { - if let Ok(n) = state - .store - .fail_remaining_bundle( - schedule.bundle_id.as_deref().unwrap_or(""), - &format!( - "bundle chain rejected at seq {}: {}", - schedule.bundle_seq.unwrap_or(0), - reason.as_deref().unwrap_or("no reason") - ), - ) - .await - { - if n > 0 { - tracing::info!( - "Bundle {} rejected: failed {n} remaining item(s)", - schedule.bundle_id.as_deref().unwrap_or("") - ); + match state.store.reject_schedule(&id, reason.as_deref()).await { + Ok(schedule) => { + // Thread-bundle: rejecting one member strands the rest of the + // chain. The cascade runs only after reject succeeds (reject + // itself fails for non-rejectable states like published), so an + // invalid request cannot destroy scheduled content. + if let Some(ref bundle_id) = schedule.bundle_id { + match state + .store + .fail_remaining_bundle( + bundle_id, + &format!( + "bundle chain rejected at seq {}: {}", + schedule.bundle_seq.unwrap_or(0), + reason.as_deref().unwrap_or("no reason") + ), + ) + .await + { + Ok(n) if n > 0 => { + tracing::info!("Bundle {bundle_id} rejected: failed {n} remaining item(s)"); + } + Ok(_) => {} + Err(e) => { + tracing::error!("Bundle {bundle_id} cascade fail failed: {e}") + } } } + ( + StatusCode::OK, + Json(serde_json::json!({ "data": schedule })), + ) } - } - match state.store.reject_schedule(&id, reason.as_deref()).await { - Ok(schedule) => ( - StatusCode::OK, - Json(serde_json::json!({ "data": schedule })), - ), Err(e) => { let code = if matches!(e, titen_core::TitenError::ScheduleNotFound(_)) { StatusCode::NOT_FOUND diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs index afe3b32..2f6f00b 100644 --- a/crates/titen-api/src/routes/threads_bundle.rs +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -228,7 +228,7 @@ pub async fn create_thread_bundle( &format!("bundle creation aborted at seq {i}"), ) .await; - continue; + break; } } created.push(serde_json::json!({ diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index 6a1c576..12c1ad0 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -291,7 +291,12 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< } // Target failed or not yet published: fail this item // cleanly instead of sending a literal marker to Threads. - Some(None) => { + Some(None) | None => { + // Some(None): target failed or not yet published. + // None: marker present but lookup failed (bad seq, + // store error, missing member) — either way the + // marker can never resolve to a real post id, so + // fail the item instead of publishing garbage. let msg = format!("bundle reply target unavailable: {marker}"); let _ = store .update_schedule_status(&schedule.id, "failed", None, Some(&msg)) @@ -299,7 +304,6 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< cascade_fail_bundle(store, &schedule, &msg).await; continue; } - None => {} } } } From 608088497cd778e5d741f72a49bcb9672eaf987a Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 13:16:18 +0700 Subject: [PATCH 5/8] refactor(bundles): atomic transactional creation + serde isolation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeCora review round 4: - Bundle creation is now a single SQLite transaction (Store::begin + create_bundle_item_tx): either all rows exist or none — eliminates orphaned drafts, seq gaps, and stranded waiting rows from mid-loop failures or crashes. - Bundle membership fields (bundle_id/seq/total) removed from the serde-exposed CreateSchedule: clients could previously inject them via POST /api/schedules and hijack bundle promote/cascade logic. Bundle rows are created through an internal CreateBundleItem struct that no request body deserializes into. 229/229 tests, clippy clean. --- crates/titen-api/src/routes/schedules.rs | 3 - crates/titen-api/src/routes/threads_bundle.rs | 148 +++++++----------- crates/titen-api/tests/api_schedules.rs | 9 -- crates/titen-core/src/models.rs | 25 ++- crates/titen-core/src/store.rs | 55 ++++++- crates/titen-mcp/src/main.rs | 3 - crates/titen-mcp/tests/mcp_integration.rs | 9 -- 7 files changed, 122 insertions(+), 130 deletions(-) diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index ad63736..d0ac8fd 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -562,9 +562,6 @@ pub async fn ingest_schedule( location_id: input.location_id, auto_approve: false, reply_to_id: input.reply_to_id, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }; // Reuse the same validation surface as the manual create route by diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs index 2f6f00b..d02780d 100644 --- a/crates/titen-api/src/routes/threads_bundle.rs +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -4,8 +4,6 @@ use axum::{Json, extract::State, http::StatusCode}; use uuid::Uuid; -use titen_core::models::CreateSchedule; - /// Validate a single bundle item's media configuration (cheap, pre-insert). fn validate_item( item: &titen_core::models::BundleItem, @@ -142,122 +140,85 @@ pub async fn create_thread_bundle( }; let bundle_id = format!("bundle_{}", Uuid::now_v7()); + + // Single transaction: either the whole bundle exists or nothing does. + // This kills every partial-state class (orphaned drafts, gaps in seq, + // waiting rows stranded by a mid-loop failure). + let mut tx = match state.store.begin().await { + Ok(tx) => tx, + Err(e) => { + return bad( + &format!("failed to begin transaction: {e}"), + "CREATE_FAILED", + ); + } + }; + let mut created = Vec::with_capacity(n_items); - let mut failures = Vec::new(); for (i, item) in input.posts.iter().enumerate() { - // Reply target: index -> defer via reply_to_seq marker. The scheduler - // resolves indexes at publish time (the previous item's real post id - // only exists after it publishes). External string ids pass through. - let (reply_to_id, reply_to_seq) = match &item.reply_to { - Some(serde_json::Value::Number(num)) => (None, num.as_i64()), + // Reply target: index -> defer via bundle marker (the referenced + // item's real post id only exists after it publishes). External + // string ids pass through as-is. + let reply_to_id = match &item.reply_to { + Some(serde_json::Value::Number(num)) => { + let seq = num.as_i64().unwrap_or(i64::MIN); + Some(format!("bundle:{bundle_id}:{seq}")) + } Some(serde_json::Value::String(s)) if !s.trim().is_empty() => { - (Some(s.trim().to_string()), None) + Some(s.trim().to_string()) } - _ => (None, None), + _ => None, }; let status = if i == 0 { root_status.to_string() } else { - // Non-root items wait until their predecessor publishes. - // In instant mode the root is already pending, so the first tick - // publishes it and chain-promotes item 1 immediately. + // Non-root items wait until their predecessor publishes. In + // instant mode the root is pending, so the next tick publishes + // it and chain-promotes item 1. "bundle_waiting".to_string() }; - let cs = CreateSchedule { + let row = titen_core::models::CreateBundleItem { account_id: input.account_id.clone(), - media_type: Some( - item.media_type - .clone() - .unwrap_or_else(|| "TEXT".to_string()), - ), + media_type: item + .media_type + .clone() + .unwrap_or_else(|| "TEXT".to_string()), caption: item.caption.clone(), - text_attachment: None, media_urls: item.media_urls.clone(), - // Waiting items get the bundle slot time; promotion overrides order. scheduled_at: scheduled_at.clone(), - location_id: None, - auto_approve: matches!(status.as_str(), "pending"), reply_to_id, - bundle_id: Some(bundle_id.clone()), - bundle_seq: Some(i as i64), - bundle_total: Some(n_items as i64), + bundle_id: bundle_id.clone(), + bundle_seq: i as i64, + bundle_total: n_items as i64, + status: status.clone(), }; - // reply_to_seq marker: encode into result_json-free dedicated place — - // reuse reply_to_id with a marker prefix the scheduler understands. - let mut cs = cs; - if let Some(seq) = reply_to_seq { - cs.reply_to_id = Some(format!("bundle:{bundle_id}:{seq}")); - } - // Force non-root items to bundle_waiting regardless of auto_approve. - if i > 0 { - cs.auto_approve = false; - } let id = Uuid::now_v7().to_string(); - match state.store.create_schedule(&id, &cs).await { - Ok(schedule) => { - // create_schedule only writes 'draft'/'pending'. Non-root - // items must be 'bundle_waiting': fix THIS row up immediately, - // scoped to its id (a brand-new row created moments ago with - // scheduled_at in the future cannot be claimed by a concurrent - // tick, and the update touches nothing else — no bulk flip, - // no root restore, no race). - if i > 0 { - if let Err(e) = state - .store - .update_schedule_status(&id, "bundle_waiting", None, None) - .await - { - let msg = format!("failed to set bundle_waiting: {e}"); - // This row is stuck as draft/pending and would - // duplicate the chain — mark it failed explicitly. - let _ = state - .store - .update_schedule_status(&id, "failed", None, Some(&msg)) - .await; - failures.push(serde_json::json!({ "seq": i, "error": msg })); - // A gap at i strands everything after it (promotion is - // seq-1 -> seq): stop and fail the waiting members. - let _ = state - .store - .fail_remaining_bundle( - &bundle_id, - &format!("bundle creation aborted at seq {i}"), - ) - .await; - break; - } - } - created.push(serde_json::json!({ - "id": schedule.id, - "seq": i, - "status": if i == 0 { root_status } else { "bundle_waiting" }, - })) - } + let create_result = state.store.create_bundle_item_tx(&mut tx, &id, &row).await; + match create_result { + Ok(schedule) => created.push(serde_json::json!({ + "id": schedule.id, + "seq": i, + "status": schedule.status, + })), Err(e) => { - failures.push(serde_json::json!({ "seq": i, "error": e.to_string() })); - // A gap at i would strand every later item: stop creating and - // fail the waiting members created so far. - let _ = state - .store - .fail_remaining_bundle( - &bundle_id, - &format!("bundle creation aborted at seq {i}: {e}"), - ) - .await; - break; + // tx drops here → rollback: nothing partial survives. + return bad( + &format!("failed to create item {i}: {e} — bundle rolled back"), + "CREATE_FAILED", + ); } } } - let status = if failures.is_empty() { - StatusCode::CREATED - } else { - StatusCode::MULTI_STATUS - }; + if let Err(e) = tx.commit().await { + return bad(&format!("failed to commit bundle: {e}"), "CREATE_FAILED"); + } + + let status = StatusCode::CREATED; ( status, @@ -265,8 +226,7 @@ pub async fn create_thread_bundle( "data": { "bundle_id": bundle_id, "items": created, - "failures": failures, - "mode": if input.scheduled_at.is_some() { "scheduled" } else { "instant" }, + "mode": if input.scheduled_at.is_some() { "scheduled" } else { "instant" }, } })), ) diff --git a/crates/titen-api/tests/api_schedules.rs b/crates/titen-api/tests/api_schedules.rs index 2afb898..324671c 100644 --- a/crates/titen-api/tests/api_schedules.rs +++ b/crates/titen-api/tests/api_schedules.rs @@ -177,9 +177,6 @@ async fn list_schedules() { location_id: None, auto_approve: true, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }, ) .await @@ -223,9 +220,6 @@ async fn list_schedules_filters_by_account() { location_id: None, auto_approve: true, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }, ) .await @@ -320,9 +314,6 @@ async fn list_schedules_pagination_meta() { location_id: None, auto_approve: true, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }, ) .await diff --git a/crates/titen-core/src/models.rs b/crates/titen-core/src/models.rs index 44826ff..6df241a 100644 --- a/crates/titen-core/src/models.rs +++ b/crates/titen-core/src/models.rs @@ -155,14 +155,23 @@ pub struct CreateSchedule { /// Publish as a reply to this Threads post ID (TEXT media type only). #[serde(default)] pub reply_to_id: Option, - /// Thread-bundle membership (set by /api/threads expansion, not by - /// clients directly). - #[serde(default)] - pub bundle_id: Option, - #[serde(default)] - pub bundle_seq: Option, - #[serde(default)] - pub bundle_total: Option, +} + +/// Internal bundle row creator (Phase 2). Not serde-deserializable from any +/// request body: only /api/threads constructs it, so clients can never +/// inject bundle membership into普通 schedules. +#[derive(Debug)] +pub struct CreateBundleItem { + pub account_id: String, + pub media_type: String, + pub caption: Option, + pub media_urls: Option>, + pub scheduled_at: String, + pub reply_to_id: Option, + pub bundle_id: String, + pub bundle_seq: i64, + pub bundle_total: i64, + pub status: String, } /// One entry inside a thread bundle (`POST /api/threads`). diff --git a/crates/titen-core/src/store.rs b/crates/titen-core/src/store.rs index 5c81717..811bbff 100644 --- a/crates/titen-core/src/store.rs +++ b/crates/titen-core/src/store.rs @@ -1007,8 +1007,8 @@ impl Store { }; sqlx::query( - "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, source, bundle_id, bundle_seq, bundle_total) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, source) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(id) .bind(&input.account_id) @@ -1021,13 +1021,60 @@ impl Store { .bind(&input.location_id) .bind(&input.reply_to_id) .bind(source) + .execute(&self.pool) + .await?; + + self.get_schedule(id).await + } + + /// Begin a transaction (thread-bundle atomic creation). + pub async fn begin(&self) -> Result> { + self.pool.begin().await.map_err(Into::into) + } + + /// Create one thread-bundle member inside `tx` (Phase 2). Bundle + /// membership is internal-only: CreateBundleItem is not serde-compatible + /// with any request body, so clients cannot inject it. + pub async fn create_bundle_item_tx( + &self, + tx: &mut sqlx::SqliteConnection, + id: &str, + input: &crate::models::CreateBundleItem, + ) -> Result { + let media_urls = input + .media_urls + .as_ref() + .map(|urls| serde_json::to_string(urls).unwrap_or_default()); + let caption = input + .caption + .as_deref() + .map(crate::models::sanitize_caption); + + sqlx::query( + "INSERT INTO schedules (id, account_id, media_type, caption, text_attachment, media_urls, scheduled_at, status, location_id, reply_to_id, bundle_id, bundle_seq, bundle_total) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(id) + .bind(&input.account_id) + .bind(&input.media_type) + .bind(&caption) + .bind(None::) // text_attachment + .bind(&media_urls) + .bind(&input.scheduled_at) + .bind(&input.status) + .bind(None::) // location_id + .bind(&input.reply_to_id) .bind(&input.bundle_id) .bind(input.bundle_seq) .bind(input.bundle_total) - .execute(&self.pool) + .execute(&mut *tx) .await?; - self.get_schedule(id).await + sqlx::query_as::<_, Schedule>("SELECT * FROM schedules WHERE id = ?") + .bind(id) + .fetch_one(&mut *tx) + .await + .map_err(|_| TitenError::ScheduleNotFound(id.to_string())) } /// All bundle members of one bundle, ordered by seq (thread-bundle Phase 2). diff --git a/crates/titen-mcp/src/main.rs b/crates/titen-mcp/src/main.rs index 49e40ef..b9e4574 100644 --- a/crates/titen-mcp/src/main.rs +++ b/crates/titen-mcp/src/main.rs @@ -647,9 +647,6 @@ fn handle_tool_call( location_id: None, auto_approve: false, reply_to_id, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }; let id = uuid::Uuid::now_v7().to_string(); diff --git a/crates/titen-mcp/tests/mcp_integration.rs b/crates/titen-mcp/tests/mcp_integration.rs index f676674..d44f69e 100644 --- a/crates/titen-mcp/tests/mcp_integration.rs +++ b/crates/titen-mcp/tests/mcp_integration.rs @@ -72,9 +72,6 @@ async fn create_test_schedule(store: &Store, id: &str, account_id: &str) { location_id: None, auto_approve: true, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }; store .create_schedule(id, &input) @@ -350,9 +347,6 @@ async fn test_approve_draft_schedule() { location_id: None, auto_approve: false, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }; let sched = store .create_schedule("sched-1", &input) @@ -383,9 +377,6 @@ async fn test_reject_draft_schedule() { location_id: None, auto_approve: false, reply_to_id: None, - bundle_id: None, - bundle_seq: None, - bundle_total: None, }; store .create_schedule("sched-1", &input) From 717b47cea089bb840188246b5f0c38e61407cb32 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 13:26:05 +0700 Subject: [PATCH 6/8] fix(bundles): reject alt_text explicitly; wire bundle-level HITL approve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeCora review round 5: - Bundle items accepted alt_text but the schedule path discarded it silently — publishes without the accessibility text the caller set. Bundle items with alt_text now get an explicit 400 (ALT_TEXT_UNSUPPORTED) until alt text is plumbed through. - Approving a bundle root via HITL now approves the whole bundle: members 1..n (draft) flip to bundle_waiting and run after the root publishes. Previously they were never reviewed at all, contradicting the auto_approve contract. set_bundle_status (previously unused) is the mechanism. 229/229 tests, clippy clean. --- crates/titen-api/src/routes/schedules.rs | 29 ++++++++++++++++--- crates/titen-api/src/routes/threads_bundle.rs | 15 ++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/crates/titen-api/src/routes/schedules.rs b/crates/titen-api/src/routes/schedules.rs index d0ac8fd..c751912 100644 --- a/crates/titen-api/src/routes/schedules.rs +++ b/crates/titen-api/src/routes/schedules.rs @@ -394,10 +394,31 @@ pub async fn approve_schedule( ) -> (StatusCode, Json) { // TODO: extract approver identity from auth context when per-user auth is added match state.store.approve_schedule(&id, Some("api")).await { - Ok(schedule) => ( - StatusCode::OK, - Json(serde_json::json!({ "data": schedule })), - ), + Ok(schedule) => { + // Thread-bundle HITL: approving the ROOT approves the whole + // bundle — members 1..n (draft) become pending-waiting so the + // chain runs after the root publishes. Approving a non-root + // item only affects that item. + if let Some(ref bundle_id) = schedule.bundle_id { + if schedule.bundle_seq == Some(0) { + match state + .store + .set_bundle_status(bundle_id, "draft", "bundle_waiting") + .await + { + Ok(n) if n > 0 => tracing::info!( + "Bundle {bundle_id} approved: {n} waiting item(s) queued" + ), + Ok(_) => {} + Err(e) => tracing::error!("Bundle {bundle_id} approve cascade failed: {e}"), + } + } + } + ( + StatusCode::OK, + Json(serde_json::json!({ "data": schedule })), + ) + } Err(e) => { let code = if matches!(e, titen_core::TitenError::ScheduleNotFound(_)) { StatusCode::NOT_FOUND diff --git a/crates/titen-api/src/routes/threads_bundle.rs b/crates/titen-api/src/routes/threads_bundle.rs index d02780d..f145a78 100644 --- a/crates/titen-api/src/routes/threads_bundle.rs +++ b/crates/titen-api/src/routes/threads_bundle.rs @@ -110,6 +110,21 @@ pub async fn create_thread_bundle( ); } } + // alt_text: not plumbed through the schedule path yet — reject + // explicitly instead of silently discarding accessibility text. + if item + .alt_text + .as_deref() + .map(str::trim) + .map(str::len) + .unwrap_or(0) + > 0 + { + return bad( + &format!("item {i}: alt_text is not yet supported on bundles; omit it for now"), + "ALT_TEXT_UNSUPPORTED", + ); + } } // Resolve mode + timestamp From b00f0ea0b725a4bd27c95f9eb976467b5812ecbe Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 13:37:02 +0700 Subject: [PATCH 7/8] fix(bundles): transient marker lookup errors retry instead of destroy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeCora review round 4 final finding: resolve_bundle_marker mapped a transient DB error to None, and the caller treated every None as a permanent failure — one SQLite hiccup would permanently kill the schedule and cascade-fail the rest of the bundle. resolve_bundle_marker now returns Result: malformed markers and failed targets still fail the item permanently, but store errors reset the claimed schedule to pending so the next tick retries the lookup. Bundle creation is also fully transactional now (Store::begin + create_bundle_item_tx in one commit) — a mid-loop failure rolls back every row instead of leaving orphaned drafts or seq gaps. alt_text on bundle items is rejected explicitly (ALT_TEXT_UNSUPPORTED) until plumbed through, and bundle-level HITL approve flips all draft members to bundle_waiting via set_bundle_status. 229/229 tests, clippy clean. --- crates/titen-core/src/scheduler.rs | 70 +++++++++++++++++++++--------- 1 file changed, 49 insertions(+), 21 deletions(-) diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index 12c1ad0..bd5c6ec 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -286,16 +286,14 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< if let Some(marker) = reply_to.clone() { if marker.starts_with("bundle:") { match resolve_bundle_marker(store, &marker).await { - Some(Some(target_id)) => { + Ok(Some(Some(target_id))) => { reply_to = Some(target_id); } - // Target failed or not yet published: fail this item - // cleanly instead of sending a literal marker to Threads. - Some(None) | None => { - // Some(None): target failed or not yet published. - // None: marker present but lookup failed (bad seq, - // store error, missing member) — either way the - // marker can never resolve to a real post id, so + // Target failed or not yet published / malformed marker: + // fail this item cleanly instead of sending a literal + // marker to Threads. + Ok(Some(None)) | Ok(None) => { + // Ok(None): malformed marker — permanent problem, // fail the item instead of publishing garbage. let msg = format!("bundle reply target unavailable: {marker}"); let _ = store @@ -304,6 +302,20 @@ async fn process_due_schedules(store: &Store, client: &ThreadsClient) -> Result< cascade_fail_bundle(store, &schedule, &msg).await; continue; } + Err(e) => { + // Transient store error (lock timeout, pool + // exhaustion): NOT fatal — reset to pending so the + // next tick retries the lookup instead of destroying + // publishable content. + warn!( + "Bundle marker lookup errored for schedule {} — retrying next tick: {e}", + schedule.id + ); + let _ = store + .update_schedule_status(&schedule.id, "pending", None, None) + .await; + continue; + } } } } @@ -438,19 +450,35 @@ async fn cascade_fail_bundle(store: &Store, schedule: &crate::models::Schedule, /// Resolve a `bundle::` reply marker to the referenced /// member's published Threads post id. /// Returns: -/// - `Some(Some(post_id))` — referenced member published; proceed. -/// - `Some(None)` — referenced member failed; the caller should fail this -/// item (its reply target can never exist). -/// - `None` — not a bundle marker. -async fn resolve_bundle_marker(store: &Store, marker: &str) -> Option> { - let rest = marker.strip_prefix("bundle:")?; - let (bundle_id, seq_str) = rest.rsplit_once(':')?; - let seq: i64 = seq_str.parse().ok()?; - let members = store.get_bundle_schedules(bundle_id).await.ok()?; - let target = members.iter().find(|s| s.bundle_seq == Some(seq))?; +/// - `Ok(Some(Some(post_id)))` — referenced member published; proceed. +/// - `Ok(Some(None))` — referenced member failed/cancelled; its reply can +/// never exist, so the caller should fail this item permanently. +/// - `Ok(None)` — malformed marker; permanent, fail the item. +/// - `Err(e)` — transient store error; the caller should retry next tick +/// instead of destroying scheduled content. +async fn resolve_bundle_marker( + store: &Store, + marker: &str, +) -> crate::Result>> { + let rest = marker + .strip_prefix("bundle:") + .ok_or_else(|| crate::TitenError::InvalidRequest("not a bundle marker".to_string()))?; + let (bundle_id, seq_str) = rest + .rsplit_once(':') + .ok_or_else(|| crate::TitenError::InvalidRequest("malformed bundle marker".to_string()))?; + let seq: i64 = seq_str.parse().map_err(|_| { + crate::TitenError::InvalidRequest("malformed bundle marker seq".to_string()) + })?; + let members = store.get_bundle_schedules(bundle_id).await?; + let target = members + .iter() + .find(|s| s.bundle_seq == Some(seq)) + .ok_or_else(|| { + crate::TitenError::InvalidRequest(format!("bundle member {bundle_id}:{seq} not found")) + })?; match target.status.as_str() { - "published" => Some(target.result_post_id.clone()), - "failed" | "cancelled" | "rejected" => Some(None), - _ => Some(None), + "published" => Ok(Some(target.result_post_id.clone())), + "failed" | "cancelled" | "rejected" => Ok(Some(None)), + _ => Ok(Some(None)), } } From 157c5b273ce71087348f982f1387d1f2bed3199a Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 8 Sep 2026 13:47:10 +0700 Subject: [PATCH 8/8] fix(bundles): malformed markers fail permanently; only store errors retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeCora review round 5: resolve_bundle_marker returned Err(InvalidRequest) for permanent conditions (malformed marker, unparseable seq, missing member), which the caller treated as transient and reset to pending — a user-supplied reply_to string starting with 'bundle:' would retry forever, spamming logs every tick. The marker parser now returns Ok(None) for permanent conditions (item fails immediately) and Err only for transient store errors (item resets to pending for retry next tick). --- crates/titen-core/src/scheduler.rs | 36 +++++++++++++++++------------- 1 file changed, 21 insertions(+), 15 deletions(-) diff --git a/crates/titen-core/src/scheduler.rs b/crates/titen-core/src/scheduler.rs index bd5c6ec..f75c458 100644 --- a/crates/titen-core/src/scheduler.rs +++ b/crates/titen-core/src/scheduler.rs @@ -460,22 +460,28 @@ async fn resolve_bundle_marker( store: &Store, marker: &str, ) -> crate::Result>> { - let rest = marker - .strip_prefix("bundle:") - .ok_or_else(|| crate::TitenError::InvalidRequest("not a bundle marker".to_string()))?; - let (bundle_id, seq_str) = rest - .rsplit_once(':') - .ok_or_else(|| crate::TitenError::InvalidRequest("malformed bundle marker".to_string()))?; - let seq: i64 = seq_str.parse().map_err(|_| { - crate::TitenError::InvalidRequest("malformed bundle marker seq".to_string()) - })?; + // Malformed markers and missing members are PERMANENT (client-supplied + // input) -> Ok(None): the caller fails the item. Only store errors are + // Err: transient, the caller retries next tick. + let rest = match marker.strip_prefix("bundle:") { + Some(r) => r, + None => return Ok(None), // unreachable — caller pre-gates on prefix + }; + let (bundle_id, seq_str) = match rest.rsplit_once(':') { + Some(pair) => pair, + None => return Ok(None), + }; + let seq: i64 = match seq_str.parse() { + Ok(s) => s, + Err(_) => return Ok(None), + }; let members = store.get_bundle_schedules(bundle_id).await?; - let target = members - .iter() - .find(|s| s.bundle_seq == Some(seq)) - .ok_or_else(|| { - crate::TitenError::InvalidRequest(format!("bundle member {bundle_id}:{seq} not found")) - })?; + let target = match members.iter().find(|s| s.bundle_seq == Some(seq)) { + Some(t) => t, + // Referenced member does not exist (deleted bundle, bad index): + // permanent — its post id can never appear. + None => return Ok(None), + }; match target.status.as_str() { "published" => Ok(Some(target.result_post_id.clone())), "failed" | "cancelled" | "rejected" => Ok(Some(None)),