From bcbb7a6229291f84635f1b3a6686da8941602b30 Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Fri, 24 Jul 2026 16:23:47 -0600 Subject: [PATCH] feat(server): add background scheduling engine (COD-377) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spawn a tokio task during server startup that polls list_pending_schedules() every 30 seconds and publishes due content via Iris. - scheduler.rs (new): poll loop with tokio::select for shutdown signal, per-schedule processing with atomic claim, Iris timeout, variant selection - storage.rs: add claim_schedule (atomic Pending→InProgress with status guard), update_schedule_status, get_schedule_status - lib.rs: spawn scheduler after listener bind, graceful shutdown via watch channel driven by SIGTERM + SIGINT, bounded sched.await with timeout - api.rs: add build_router_from_arc for Arc sharing Co-authored-by: Archon --- crates/postghost-server/src/api.rs | 7 +- crates/postghost-server/src/lib.rs | 59 +++- crates/postghost-server/src/scheduler.rs | 334 +++++++++++++++++++++++ crates/postghost-server/src/storage.rs | 73 ++++- 4 files changed, 467 insertions(+), 6 deletions(-) create mode 100644 crates/postghost-server/src/scheduler.rs diff --git a/crates/postghost-server/src/api.rs b/crates/postghost-server/src/api.rs index 604def1..f78e7c4 100644 --- a/crates/postghost-server/src/api.rs +++ b/crates/postghost-server/src/api.rs @@ -23,7 +23,12 @@ pub struct AppState { } pub fn build_router(state: AppState) -> Router { - let state = Arc::new(state); + build_router_from_arc(Arc::new(state)) +} + +/// Build a router from an already-Arc'd state. Used when the server shares +/// the state with the background scheduler task. +pub fn build_router_from_arc(state: Arc) -> Router { Router::new() .route("/health", get(health)) .route("/api/v1/content", get(list_content).post(create_content)) diff --git a/crates/postghost-server/src/lib.rs b/crates/postghost-server/src/lib.rs index fbd4d2c..91fb16c 100644 --- a/crates/postghost-server/src/lib.rs +++ b/crates/postghost-server/src/lib.rs @@ -1,11 +1,13 @@ pub mod api; pub mod config; pub mod iris; +pub mod scheduler; pub mod storage; use std::net::SocketAddr; use anyhow::Context; +use tokio::sync::watch; pub async fn run(config: config::ServerConfig) -> anyhow::Result<()> { tracing_subscriber::fmt() @@ -27,17 +29,66 @@ pub async fn run(config: config::ServerConfig) -> anyhow::Result<()> { config.database_url ); - let state = api::AppState { + let state = std::sync::Arc::new(api::AppState { storage, iris_client, config: config.clone(), - }; + }); - let app = api::build_router(state); + // Shutdown signal — a watch channel the scheduler monitors. axum's graceful + // shutdown drives it via the `with_graceful_shutdown` closure below. + let (shutdown_tx, shutdown_rx) = watch::channel(false); + + // Bind the listener BEFORE spawning the scheduler, so a port-in-use failure + // returns early without leaking a detached polling task. + let app = api::build_router_from_arc(state.clone()); let addr: SocketAddr = format!("{}:{}", config.host, config.port).parse()?; let listener = tokio::net::TcpListener::bind(addr).await?; tracing::info!("PostGhost server listening on {}", addr); - axum::serve(listener, app).await?; + // Spawn the background scheduler now that the listener is bound. + let sched = tokio::spawn(scheduler::run(state.clone(), shutdown_rx)); + + axum::serve(listener, app) + .with_graceful_shutdown(async move { + // Wait for SIGINT (Ctrl+C) OR SIGTERM — either signals shutdown. + wait_for_shutdown_signal().await; + tracing::info!("shutdown signal received — stopping scheduler"); + let _ = shutdown_tx.send(true); + }) + .await?; + + // After axum returns, wait for the scheduler task to finish so we don't + // exit mid-publish. Bound by a timeout so a hung Iris call can't pin the + // process forever. + match tokio::time::timeout(std::time::Duration::from_secs(10), sched).await { + Ok(Ok(())) => tracing::info!("scheduler stopped cleanly"), + Ok(Err(e)) => tracing::error!("scheduler task error: {e:#}"), + Err(_) => tracing::warn!("scheduler did not stop within 10s — exiting anyway"), + } Ok(()) } + +/// Wait for either SIGINT (Ctrl+C) or SIGTERM. +/// +/// `tokio::signal::ctrl_c()` only catches SIGINT. Production shutdown paths +/// (docker stop, systemd, k8s) send SIGTERM, so we must handle both. +#[cfg(unix)] +async fn wait_for_shutdown_signal() { + use tokio::signal::unix::{signal, SignalKind}; + + let mut term = signal(SignalKind::terminate()).expect("install SIGTERM handler"); + let mut int = signal(SignalKind::interrupt()).expect("install SIGINT handler"); + + tokio::select! { + _ = term.recv() => tracing::info!("received SIGTERM"), + _ = int.recv() => tracing::info!("received SIGINT"), + } +} + +/// Non-Unix fallback (Windows) — only SIGINT is available. +#[cfg(not(unix))] +async fn wait_for_shutdown_signal() { + let _ = tokio::signal::ctrl_c().await; + tracing::info!("received interrupt signal"); +} diff --git a/crates/postghost-server/src/scheduler.rs b/crates/postghost-server/src/scheduler.rs new file mode 100644 index 0000000..62c5778 --- /dev/null +++ b/crates/postghost-server/src/scheduler.rs @@ -0,0 +1,334 @@ +//! Background scheduler — polls for due schedules and publishes content via Iris. + +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use tokio::sync::watch; +use tokio::time::{interval, Interval}; + +use crate::api::AppState; +use postghost::ScheduleStatus; + +/// How often the scheduler wakes up to check for due schedules. +const POLL_INTERVAL: Duration = Duration::from_secs(30); + +/// Run the scheduler loop until `shutdown` is signalled. +/// +/// Every [`POLL_INTERVAL`], queries pending schedules and publishes any whose +/// `scheduled_for` time has arrived. Errors per-schedule are logged and the +/// schedule marked `Failed`; the loop itself never panics on individual +/// schedule failures. +pub async fn run(state: Arc, mut shutdown: watch::Receiver) { + tracing::info!("scheduler started — polling every {:?}", POLL_INTERVAL); + let mut ticker = interval_at_aligned(); + // First tick completes immediately; consume it so we don't process on startup. + ticker.tick().await; + + loop { + // Wait for either the next tick or a shutdown signal. + tokio::select! { + _ = ticker.tick() => {} + _ = shutdown.changed() => { + if *shutdown.borrow() { + break; + } + } + } + + if let Err(e) = poll_once(&state).await { + tracing::error!("scheduler poll error: {e:#}"); + } + } + + tracing::info!("scheduler stopped"); +} + +/// Create an interval whose first real tick fires one POLL_INTERVAL from now. +fn interval_at_aligned() -> Interval { + let mut t = interval(POLL_INTERVAL); + // Don't fire the first tick immediately — we consumed one above, but set + // `set_missed_tick_behavior` so backpressure doesn't pile up. + t.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + t +} + +/// One poll cycle: fetch due pending schedules and process them. +async fn poll_once(state: &Arc) -> anyhow::Result<()> { + let pending = state.storage.list_pending_schedules()?; + if pending.is_empty() { + return Ok(()); + } + + let now = Utc::now(); + for row in pending { + let due_at = match chrono::DateTime::parse_from_rfc3339(&row.scheduled_for) { + Ok(dt) => dt.with_timezone(&Utc), + Err(e) => { + tracing::error!( + "schedule {} has invalid scheduled_for {:?}: {e}", + row.id, + row.scheduled_for + ); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + continue; + } + }; + + if due_at > now { + continue; + } + + tracing::info!("schedule {} is due — processing", row.id); + process_schedule(state, &row).await; + } + + Ok(()) +} + +/// Process a single due schedule: claim it atomically, fetch content, send to +/// Iris, mark Published/Failed. +async fn process_schedule(state: &Arc, row: &crate::storage::ScheduleRow) { + // Atomically claim the schedule: only transition Pending → InProgress if + // the schedule is still pending. This prevents double-processing under + // concurrent poll cycles or multiple server instances. + match state.storage.claim_schedule(&row.id) { + Ok(true) => tracing::info!("schedule {} claimed — processing", row.id), + Ok(false) => { + tracing::debug!("schedule {} already claimed or terminal — skipping", row.id); + return; + } + Err(e) => { + tracing::error!("schedule {}: failed to claim: {e:#}", row.id); + return; + } + } + + // Fetch the associated content. + let content_id = match uuid::Uuid::parse_str(&row.content_id) { + Ok(id) => id, + Err(e) => { + tracing::error!( + "schedule {}: invalid content_id {:?}: {e}", + row.id, + row.content_id + ); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + return; + } + }; + + let content = match state.storage.get_content(content_id) { + Ok(Some(c)) => c, + Ok(None) => { + tracing::error!( + "schedule {}: content {} not found — marking Failed", + row.id, + row.content_id + ); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + return; + } + Err(e) => { + tracing::error!( + "schedule {}: failed to fetch content {}: {e:#}", + row.id, + row.content_id + ); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + return; + } + }; + + // Select the best text for this platform: prefer the matching variant, fall + // back to the raw body. (COD-378 will add full platform-aware formatting.) + let platform = postghost::Platform::from_db_key(&row.platform); + let text = content + .variants + .iter() + .find(|v| v.platform == platform) + .map(|v| v.formatted_text.clone()) + .unwrap_or_else(|| content.body.clone()); + + // Dispatch to Iris, with a timeout so a single hung publish can't block + // the scheduler loop indefinitely. + let iris_result = tokio::time::timeout( + Duration::from_secs(30), + state + .iris_client + .send_message(&content_id.to_string(), &text), + ) + .await; + + match iris_result { + Ok(Ok(resp)) => { + tracing::info!("schedule {}: published via Iris — {:?}", row.id, resp); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Published); + } + Ok(Err(e)) => { + tracing::error!("schedule {}: Iris publish failed: {e:#}", row.id); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + } + Err(_) => { + tracing::error!("schedule {}: Iris publish timed out after 30s", row.id); + let _ = state + .storage + .update_schedule_status(&row.id, ScheduleStatus::Failed); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::ServerConfig; + use crate::iris::IrisClient; + use crate::storage::SqliteStorage; + use postghost::{Content, ContentFormat, Platform, Schedule}; + use uuid::Uuid; + + fn test_state(storage: SqliteStorage) -> Arc { + Arc::new(AppState { + storage, + iris_client: IrisClient::new("http://127.0.0.1:1".to_string()), + config: ServerConfig::default(), + }) + } + + fn save_content(storage: &SqliteStorage) -> Uuid { + let now = Utc::now(); + let id = Uuid::new_v4(); + let content = Content { + id, + title: "Scheduled".to_string(), + body: "hello world".to_string(), + format: ContentFormat::Markdown, + tags: vec![], + created_at: now, + updated_at: now, + variants: vec![], + }; + storage.save_content(&content).unwrap(); + id + } + + fn save_due_schedule(storage: &SqliteStorage, content_id: Uuid) -> String { + let now = Utc::now(); + let schedule = Schedule { + id: Uuid::new_v4(), + content_id, + platform: Platform::Twitter, + scheduled_for: now - chrono::Duration::minutes(1), // due + status: ScheduleStatus::Pending, + created_at: now, + }; + let id = schedule.id.to_string(); + storage.save_schedule(&schedule).unwrap(); + id + } + + #[test] + fn test_update_schedule_status_round_trip() { + let storage = SqliteStorage::in_memory().unwrap(); + let content_id = save_content(&storage); + let schedule_id = save_due_schedule(&storage, content_id); + + // Existing row updates. + assert!(storage + .update_schedule_status(&schedule_id, ScheduleStatus::InProgress) + .unwrap()); + + // Non-existent row returns false. + assert!(!storage + .update_schedule_status("deadbeef", ScheduleStatus::Failed) + .unwrap()); + } + + #[tokio::test] + async fn test_poll_once_marks_failed_when_iris_unreachable() { + // Iris points at a closed port — send_message will fail, schedule must + // end up Failed, not stuck InProgress. + let storage = SqliteStorage::in_memory().unwrap(); + let content_id = save_content(&storage); + let schedule_id = save_due_schedule(&storage, content_id); + + let state = test_state(storage); + poll_once(&state).await.unwrap(); + + // The schedule should now be Failed (Iris unreachable). + let status = state + .storage + .get_schedule_status(&schedule_id) + .unwrap() + .unwrap(); + assert_eq!(status, ScheduleStatus::Failed); + } + + #[tokio::test] + async fn test_poll_once_skips_not_yet_due() { + let storage = SqliteStorage::in_memory().unwrap(); + let now = Utc::now(); + let content_id = save_content(&storage); + + // Future schedule — should remain Pending. + let schedule = Schedule { + id: Uuid::new_v4(), + content_id, + platform: Platform::Twitter, + scheduled_for: now + chrono::Duration::hours(1), + status: ScheduleStatus::Pending, + created_at: now, + }; + let id = schedule.id.to_string(); + storage.save_schedule(&schedule).unwrap(); + + let state = test_state(storage); + poll_once(&state).await.unwrap(); + + let status = state.storage.get_schedule_status(&id).unwrap().unwrap(); + assert_eq!(status, ScheduleStatus::Pending); + } + + #[tokio::test] + async fn test_poll_once_marks_failed_when_content_missing() -> anyhow::Result<()> { + // The schedules FK (`content_id TEXT NOT NULL REFERENCES content(id)`) + // normally prevents this scenario — you can't delete content while a + // schedule references it. But if FK enforcement is off, or content is + // removed via raw SQL, the scheduler must handle `get_content` returning + // `None` gracefully: mark the schedule Failed, don't panic. + // + // We simulate this by inserting a schedule with FK checks disabled. + let storage = SqliteStorage::in_memory()?; + let now = Utc::now(); + let ghost_id = Uuid::new_v4(); + let sched_id = Uuid::new_v4(); + + storage.insert_schedule_raw_for_test( + &sched_id.to_string(), + &ghost_id.to_string(), + "twitter", + &(now - chrono::Duration::minutes(1)).to_rfc3339(), + &now.to_rfc3339(), + )?; + + let state = test_state(storage); + poll_once(&state).await?; + + // The schedule referencing ghost content should no longer be pending. + let pending = state.storage.list_pending_schedules()?; + assert!(pending.is_empty(), "schedule should not be pending anymore"); + Ok(()) + } +} diff --git a/crates/postghost-server/src/storage.rs b/crates/postghost-server/src/storage.rs index 16f8bde..8af62e0 100644 --- a/crates/postghost-server/src/storage.rs +++ b/crates/postghost-server/src/storage.rs @@ -3,7 +3,8 @@ use std::sync::Mutex; use anyhow::{Context, Result}; use chrono::Utc; use postghost::{ - Content, ContentId, Platform, PlatformVariant, Schedule, WorkflowState, WorkflowTransition, + Content, ContentId, Platform, PlatformVariant, Schedule, ScheduleStatus, WorkflowState, + WorkflowTransition, }; use rusqlite::Connection; use uuid::Uuid; @@ -308,6 +309,76 @@ impl SqliteStorage { rows.collect::>>() .map_err(Into::into) } + + /// Transition a schedule's status. + /// Returns `Ok(true)` if the row was updated, `Ok(false)` if the ID doesn't exist. + pub fn update_schedule_status(&self, id: &str, status: ScheduleStatus) -> Result { + let conn = self.conn.lock().unwrap(); + let rows = conn.execute( + "UPDATE schedules SET status = ?1 WHERE id = ?2", + rusqlite::params![enum_to_str(&status)?, id], + )?; + Ok(rows > 0) + } + + /// Atomically claim a pending schedule for processing by transitioning it + /// to `InProgress` — but ONLY if its current status is `pending`. + /// + /// Returns `Ok(true)` if this caller won the claim (the schedule was + /// pending and is now in_progress), `Ok(false)` if it was already claimed + /// by another poll cycle or is in a terminal state. + /// + /// This prevents double-publishing when two scheduler instances or poll + /// cycles race on the same schedule. + pub fn claim_schedule(&self, id: &str) -> Result { + let conn = self.conn.lock().unwrap(); + let rows = conn.execute( + "UPDATE schedules SET status = ?1 WHERE id = ?2 AND status = 'pending'", + rusqlite::params![enum_to_str(&ScheduleStatus::InProgress)?, id], + )?; + Ok(rows > 0) + } + + /// Read the current status of a schedule by ID. + /// Returns `Ok(None)` if the schedule doesn't exist. + pub fn get_schedule_status(&self, id: &str) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare("SELECT status FROM schedules WHERE id = ?1")?; + let mut rows = stmt.query_map([id], |row| row.get::<_, String>(0))?; + match rows.next() { + Some(s) => Ok(Some(str_to_enum(&s?)?)), + None => Ok(None), + } + } + + /// Test-only helper: insert a schedule row that references a non-existent + /// content_id, bypassing the FK constraint. Used to test scheduler + /// robustness when content is removed out-of-band. + #[cfg(test)] + pub(crate) fn insert_schedule_raw_for_test( + &self, + schedule_id: &str, + content_id: &str, + platform: &str, + scheduled_for_rfc3339: &str, + created_at_rfc3339: &str, + ) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute_batch("PRAGMA foreign_keys = OFF")?; + conn.execute( + "INSERT INTO schedules (id, content_id, platform, scheduled_for, status, created_at) \ + VALUES (?1, ?2, ?3, ?4, 'pending', ?5)", + rusqlite::params![ + schedule_id, + content_id, + platform, + scheduled_for_rfc3339, + created_at_rfc3339, + ], + )?; + conn.execute_batch("PRAGMA foreign_keys = ON")?; + Ok(()) + } } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]