diff --git a/Cargo.lock b/Cargo.lock index f66fada4..80cbece1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -895,6 +895,17 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "croner" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4aa42bcd3d846ebf66e15bd528d1087f75d1c6c1c66ebff626178a106353c576" +dependencies = [ + "chrono", + "derive_builder", + "strum", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -996,6 +1007,41 @@ dependencies = [ "cmov", ] +[[package]] +name = "darling" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d00b9596d185e565c2207a0b01f8bd1a135483d02d9b7b0a54b11da8d53412e" +dependencies = [ + "fnv", + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn", +] + +[[package]] +name = "darling_macro" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" +dependencies = [ + "darling_core", + "quote", + "syn", +] + [[package]] name = "dashmap" version = "6.2.1" @@ -1025,6 +1071,37 @@ dependencies = [ "powerfmt", ] +[[package]] +name = "derive_builder" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "507dfb09ea8b7fa618fcf76e953f4f5e192547945816d5358edffe39f6f94947" +dependencies = [ + "derive_builder_macro", +] + +[[package]] +name = "derive_builder_core" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d5bcf7b024d6835cfb3d473887cd966994907effbe9227e8c8219824d06c4e8" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "derive_builder_macro" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" +dependencies = [ + "derive_builder_core", + "syn", +] + [[package]] name = "derive_more" version = "2.1.1" @@ -1874,6 +1951,12 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + [[package]] name = "idna" version = "1.1.0" @@ -2904,6 +2987,7 @@ dependencies = [ "base64", "brotli", "chrono", + "croner", "dashmap", "flate2", "fluent-bundle", @@ -3636,6 +3720,27 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" diff --git a/crates/ruscker-admin/Cargo.toml b/crates/ruscker-admin/Cargo.toml index 6a6ee849..702075ca 100644 --- a/crates/ruscker-admin/Cargo.toml +++ b/crates/ruscker-admin/Cargo.toml @@ -40,6 +40,8 @@ http-body-util = { workspace = true } # the connector must be built with one explicitly. hyper-rustls = "0.27" rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } +# Cron parsing for scheduled jobs (#986) — verified on crates.io (3.0.1). +croner = "3" futures-util = { workspace = true } async-stream = { workspace = true } lol_html = { workspace = true } diff --git a/crates/ruscker-admin/migrations-pg/0028_schedules.sql b/crates/ruscker-admin/migrations-pg/0028_schedules.sql new file mode 100644 index 00000000..d1011b6b --- /dev/null +++ b/crates/ruscker-admin/migrations-pg/0028_schedules.sql @@ -0,0 +1,24 @@ +-- Postgres twin of migrations/0028 (#986 slice B). +CREATE TABLE schedules ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + spec_id TEXT NOT NULL, + cron TEXT NOT NULL, + cmd_json TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + timeout_secs BIGINT, + last_run_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL +); + +CREATE TABLE schedule_runs ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + schedule_id BIGINT NOT NULL REFERENCES schedules(id) ON DELETE CASCADE, + started_at TIMESTAMPTZ NOT NULL, + finished_at TIMESTAMPTZ, + status TEXT NOT NULL, + exit_code BIGINT, + log_tail TEXT, + duration_ms BIGINT +); +CREATE INDEX idx_schedule_runs_schedule ON schedule_runs(schedule_id, started_at DESC); diff --git a/crates/ruscker-admin/migrations/0028_schedules.sql b/crates/ruscker-admin/migrations/0028_schedules.sql new file mode 100644 index 00000000..065017fd --- /dev/null +++ b/crates/ruscker-admin/migrations/0028_schedules.sql @@ -0,0 +1,30 @@ +-- Scheduled jobs (#986 slice B): cron-triggered run-to-completion +-- executions of a spec's image (ETL, reports). `cmd_json` is an +-- optional argv override (JSON array) — NULL runs the spec's own +-- command. `last_run_at` is the scheduler's fire marker (set BEFORE +-- the job runs, so a crash mid-job never double-fires). +CREATE TABLE schedules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + spec_id TEXT NOT NULL, + cron TEXT NOT NULL, + cmd_json TEXT, + enabled INTEGER NOT NULL DEFAULT 1, + timeout_secs INTEGER, + last_run_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +-- Run history. `status`: 'ok' (exit 0) | 'failed' (non-zero exit) | +-- 'error' (could not run: pull/create/start/timeout). +CREATE TABLE schedule_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + schedule_id INTEGER NOT NULL REFERENCES schedules(id) ON DELETE CASCADE, + started_at TEXT NOT NULL, + finished_at TEXT, + status TEXT NOT NULL, + exit_code INTEGER, + log_tail TEXT, + duration_ms INTEGER +); +CREATE INDEX idx_schedule_runs_schedule ON schedule_runs(schedule_id, started_at DESC); diff --git a/crates/ruscker-admin/src/alerts.rs b/crates/ruscker-admin/src/alerts.rs index ba9bb6aa..447ba4b0 100644 --- a/crates/ruscker-admin/src/alerts.rs +++ b/crates/ruscker-admin/src/alerts.rs @@ -69,6 +69,8 @@ pub enum AlertKind { /// A spec is saturated with every replica at `max-replicas` — /// visitors are being turned away and Ruscker can't scale further. Saturated, + /// A scheduled job (#986) exited non-zero or could not run. + JobFailed, /// Operator-triggered delivery check. Test, } @@ -79,6 +81,7 @@ impl AlertKind { AlertKind::SpawnFailed => "spawn-failed", AlertKind::ReplicaDown => "replica-down", AlertKind::Saturated => "saturated", + AlertKind::JobFailed => "job-failed", AlertKind::Test => "test", } } diff --git a/crates/ruscker-admin/src/db.rs b/crates/ruscker-admin/src/db.rs index dc80322e..e8fb557c 100644 --- a/crates/ruscker-admin/src/db.rs +++ b/crates/ruscker-admin/src/db.rs @@ -165,6 +165,7 @@ pub mod images; pub mod landing; pub mod landing_blocks; pub mod ruscker_images; +pub mod schedules; pub mod settings; pub mod showcase; pub mod spec_access; diff --git a/crates/ruscker-admin/src/db/schedules.rs b/crates/ruscker-admin/src/db/schedules.rs new file mode 100644 index 00000000..6a4b0ddc --- /dev/null +++ b/crates/ruscker-admin/src/db/schedules.rs @@ -0,0 +1,336 @@ +//! Scheduled-job storage (#986 slice B): `schedules` (a cron per spec, +//! with an optional argv override) and `schedule_runs` (the history the +//! admin UI shows in slice C). The SCHEDULER decides what's due — this +//! module only stores; `mark_fired` sets `last_run_at` **before** the +//! job runs so a crash mid-job can never double-fire. + +use super::ConfigDb; +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; + +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct Schedule { + pub id: i64, + pub spec_id: String, + pub cron: String, + /// JSON argv array, or `None` = run the spec's own command. + pub cmd_json: Option, + pub enabled: bool, + pub timeout_secs: Option, + pub last_run_at: Option>, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +impl Schedule { + /// The parsed argv override, if any. A malformed stored JSON is + /// treated as "no override" (and logged by the caller). + pub fn cmd_override(&self) -> Option> { + self.cmd_json + .as_deref() + .and_then(|j| serde_json::from_str(j).ok()) + } +} + +/// Outcome bucket for a run row. `Error` = the job could not run at +/// all; `Failed` = ran and exited non-zero. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RunStatus { + Ok, + Failed, + Error, +} + +impl RunStatus { + pub fn as_str(self) -> &'static str { + match self { + RunStatus::Ok => "ok", + RunStatus::Failed => "failed", + RunStatus::Error => "error", + } + } +} + +pub async fn list_all(db: &ConfigDb) -> Result> { + let rows = match db { + ConfigDb::Sqlite(pool) => { + sqlx::query_as::<_, Schedule>("SELECT * FROM schedules ORDER BY spec_id, id") + .fetch_all(pool) + .await + .context("list schedules (sqlite)")? + } + ConfigDb::Postgres(pool) => { + sqlx::query_as::<_, Schedule>("SELECT * FROM schedules ORDER BY spec_id, id") + .fetch_all(pool) + .await + .context("list schedules (postgres)")? + } + }; + Ok(rows) +} + +pub async fn insert( + db: &ConfigDb, + spec_id: &str, + cron: &str, + cmd_json: Option<&str>, + timeout_secs: Option, + actor: Option<&str>, +) -> Result<()> { + let now = Utc::now(); + let target = format!("schedule:{spec_id}"); + match db { + ConfigDb::Sqlite(pool) => { + let mut tx = pool.begin().await.context("begin schedule insert")?; + sqlx::query( + "INSERT INTO schedules (spec_id, cron, cmd_json, enabled, timeout_secs, created_at, updated_at) + VALUES (?, ?, ?, 1, ?, ?, ?)", + ) + .bind(spec_id) + .bind(cron) + .bind(cmd_json) + .bind(timeout_secs) + .bind(now) + .bind(now) + .execute(&mut *tx) + .await + .context("insert schedule (sqlite)")?; + sqlx::query( + "INSERT INTO audit_log (actor, action, target, diff_json, occurred_at) + VALUES (?, 'schedule.create', ?, ?, ?)", + ) + .bind(actor) + .bind(&target) + .bind(serde_json::json!({ "cron": cron }).to_string()) + .bind(now) + .execute(&mut *tx) + .await + .context("audit schedule insert")?; + tx.commit().await.context("commit schedule insert")?; + } + ConfigDb::Postgres(pool) => { + let mut tx = pool.begin().await.context("begin schedule insert")?; + sqlx::query( + "INSERT INTO schedules (spec_id, cron, cmd_json, enabled, timeout_secs, created_at, updated_at) + VALUES ($1, $2, $3, TRUE, $4, $5, $6)", + ) + .bind(spec_id) + .bind(cron) + .bind(cmd_json) + .bind(timeout_secs) + .bind(now) + .bind(now) + .execute(&mut *tx) + .await + .context("insert schedule (postgres)")?; + sqlx::query( + "INSERT INTO audit_log (actor, action, target, diff_json, occurred_at) + VALUES ($1, 'schedule.create', $2, $3, $4)", + ) + .bind(actor) + .bind(&target) + .bind(serde_json::json!({ "cron": cron }).to_string()) + .bind(now) + .execute(&mut *tx) + .await + .context("audit schedule insert")?; + tx.commit().await.context("commit schedule insert")?; + } + } + Ok(()) +} + +pub async fn set_enabled(db: &ConfigDb, id: i64, enabled: bool, actor: Option<&str>) -> Result<()> { + let now = Utc::now(); + match db { + ConfigDb::Sqlite(pool) => { + sqlx::query("UPDATE schedules SET enabled = ?, updated_at = ? WHERE id = ?") + .bind(enabled) + .bind(now) + .bind(id) + .execute(pool) + .await + .context("set schedule enabled (sqlite)")?; + super::audit::record(db, actor.unwrap_or("system"), "schedule.update", &format!("schedule:{id}"), None) + .await?; + } + ConfigDb::Postgres(pool) => { + sqlx::query("UPDATE schedules SET enabled = $1, updated_at = $2 WHERE id = $3") + .bind(enabled) + .bind(now) + .bind(id) + .execute(pool) + .await + .context("set schedule enabled (postgres)")?; + super::audit::record(db, actor.unwrap_or("system"), "schedule.update", &format!("schedule:{id}"), None) + .await?; + } + } + Ok(()) +} + +pub async fn delete(db: &ConfigDb, id: i64, actor: Option<&str>) -> Result<()> { + match db { + ConfigDb::Sqlite(pool) => { + sqlx::query("DELETE FROM schedules WHERE id = ?") + .bind(id) + .execute(pool) + .await + .context("delete schedule (sqlite)")?; + } + ConfigDb::Postgres(pool) => { + sqlx::query("DELETE FROM schedules WHERE id = $1") + .bind(id) + .execute(pool) + .await + .context("delete schedule (postgres)")?; + } + } + super::audit::record(db, actor.unwrap_or("system"), "schedule.delete", &format!("schedule:{id}"), None).await +} + +/// Set the fire marker. Called BEFORE the job runs; the WHERE clause +/// on the previous value makes the claim atomic, so even a split-brain +/// second scheduler can't double-fire the same tick (mirrors the +/// scaler's belt-and-braces posture, #596). Returns whether this +/// caller won the claim. +pub async fn mark_fired( + db: &ConfigDb, + id: i64, + previous: Option>, + now: DateTime, +) -> Result { + let n = match db { + ConfigDb::Sqlite(pool) => sqlx::query( + "UPDATE schedules SET last_run_at = ? WHERE id = ? AND last_run_at IS ?", + ) + .bind(now) + .bind(id) + .bind(previous) + .execute(pool) + .await + .context("mark schedule fired (sqlite)")? + .rows_affected(), + ConfigDb::Postgres(pool) => sqlx::query( + "UPDATE schedules SET last_run_at = $1 WHERE id = $2 AND last_run_at IS NOT DISTINCT FROM $3", + ) + .bind(now) + .bind(id) + .bind(previous) + .execute(pool) + .await + .context("mark schedule fired (postgres)")? + .rows_affected(), + }; + Ok(n == 1) +} + +/// Record one finished (or failed-to-start) run. +#[allow(clippy::too_many_arguments)] +pub async fn record_run( + db: &ConfigDb, + schedule_id: i64, + started_at: DateTime, + status: RunStatus, + exit_code: Option, + log_tail: &str, + duration_ms: Option, +) -> Result<()> { + let finished = Utc::now(); + match db { + ConfigDb::Sqlite(pool) => { + sqlx::query( + "INSERT INTO schedule_runs (schedule_id, started_at, finished_at, status, exit_code, log_tail, duration_ms) + VALUES (?, ?, ?, ?, ?, ?, ?)", + ) + .bind(schedule_id) + .bind(started_at) + .bind(finished) + .bind(status.as_str()) + .bind(exit_code) + .bind(log_tail) + .bind(duration_ms) + .execute(pool) + .await + .context("record schedule run (sqlite)")?; + } + ConfigDb::Postgres(pool) => { + sqlx::query( + "INSERT INTO schedule_runs (schedule_id, started_at, finished_at, status, exit_code, log_tail, duration_ms) + VALUES ($1, $2, $3, $4, $5, $6, $7)", + ) + .bind(schedule_id) + .bind(started_at) + .bind(finished) + .bind(status.as_str()) + .bind(exit_code) + .bind(log_tail) + .bind(duration_ms) + .execute(pool) + .await + .context("record schedule run (postgres)")?; + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn mem_db() -> ConfigDb { + ConfigDb::Sqlite(crate::db::open_memory().await.expect("open in-memory")) + } + + #[tokio::test] + async fn schedule_crud_and_run_history_roundtrip() { + let db = mem_db().await; + insert(&db, "etl-app", "0 3 * * *", Some(r#"["run.sh"]"#), Some(600), Some("root")) + .await + .unwrap(); + let all = list_all(&db).await.unwrap(); + assert_eq!(all.len(), 1); + let s = &all[0]; + assert!(s.enabled); + assert_eq!(s.cmd_override(), Some(vec!["run.sh".to_string()])); + assert_eq!(s.timeout_secs, Some(600)); + assert!(s.last_run_at.is_none()); + + // The fire claim is atomic on the previous value: the first + // caller wins, a concurrent second (same previous) loses. + let now = Utc::now(); + assert!(mark_fired(&db, s.id, None, now).await.unwrap()); + assert!(!mark_fired(&db, s.id, None, now).await.unwrap(), "stale claim loses"); + + record_run(&db, s.id, now, RunStatus::Failed, Some(3), "boom", Some(1200)) + .await + .unwrap(); + set_enabled(&db, s.id, false, Some("root")).await.unwrap(); + assert!(!list_all(&db).await.unwrap()[0].enabled); + + delete(&db, s.id, Some("root")).await.unwrap(); + assert!(list_all(&db).await.unwrap().is_empty()); + } + + // Dual-dialect check (the `IS NOT DISTINCT FROM` claim + BOOLEAN + // columns). Gated on `postgres-it`. + #[cfg(feature = "postgres-it")] + #[tokio::test] + async fn schedules_against_real_postgres() { + let _guard = crate::db::pg_test_lock().lock().await; + let url = std::env::var("RUSCKER_TEST_PG_URL") + .expect("set RUSCKER_TEST_PG_URL to a reachable postgres:// DSN"); + let pg = crate::db::open_pg(&url).await.unwrap(); + sqlx::query("DELETE FROM schedule_runs").execute(&pg).await.unwrap(); + sqlx::query("DELETE FROM schedules").execute(&pg).await.unwrap(); + let db = ConfigDb::Postgres(pg); + + insert(&db, "etl-app", "*/5 * * * *", None, None, None).await.unwrap(); + let s = list_all(&db).await.unwrap().remove(0); + let now = Utc::now(); + assert!(mark_fired(&db, s.id, None, now).await.unwrap()); + assert!(!mark_fired(&db, s.id, None, now).await.unwrap()); + record_run(&db, s.id, now, RunStatus::Ok, Some(0), "", Some(10)).await.unwrap(); + delete(&db, s.id, None).await.unwrap(); + } +} diff --git a/crates/ruscker-admin/src/jobs.rs b/crates/ruscker-admin/src/jobs.rs new file mode 100644 index 00000000..7abfe75a --- /dev/null +++ b/crates/ruscker-admin/src/jobs.rs @@ -0,0 +1,288 @@ +//! The scheduled-jobs runner (#986 slice B). +//! +//! One task per process, LEADER-ONLY in HA (same posture as the +//! scaler): every [`TICK`] it loads the enabled schedules and, for each +//! one that is DUE (its cron has an occurrence between the last fire +//! marker and now), atomically claims the fire (`mark_fired` compares +//! the previous marker, so a split-brain second scheduler loses the +//! race instead of double-firing), then runs the spec's image to +//! completion via [`ruscker_core::ContainerBackend::run_job`] on a +//! DETACHED task — a long ETL never blocks the tick. The outcome lands +//! in `schedule_runs`, and a non-`Ok` outcome emits a `job-failed` +//! alert through the webhook sink (#930). +//! +//! Missed windows collapse: if the server was down over three nightly +//! occurrences, the schedule fires ONCE on the next tick (the marker +//! then moves past all three) — ETL semantics, not a message queue. + +use crate::db::schedules::{RunStatus, Schedule}; +use crate::AppState; +use chrono::{DateTime, Utc}; +use std::time::Duration; +use tokio::task::JoinHandle; + +/// Scheduler cadence. Cron's resolution is the minute, so half-minute +/// ticks keep worst-case lateness ~30 s without busy-waiting. +pub const TICK: Duration = Duration::from_secs(30); + +/// The next occurrence of `cron` strictly after `after`, or `None` +/// for an unparseable expression (surfaced by the caller once — a bad +/// expression must not wedge the tick). +fn next_occurrence(cron: &str, after: DateTime) -> Option> { + let parsed: croner::Cron = cron.parse().ok()?; + parsed.find_next_occurrence(&after, false).ok() +} + +/// Whether `schedule` is due at `now`: its cron has an occurrence in +/// `(anchor, now]`, where the anchor is the last fire marker (or the +/// schedule's creation, so a brand-new "every day at 03:00" waits for +/// 03:00 instead of firing on creation). +fn is_due(schedule: &Schedule, now: DateTime) -> bool { + let anchor = schedule.last_run_at.unwrap_or(schedule.created_at); + match next_occurrence(&schedule.cron, anchor) { + Some(next) => next <= now, + None => false, + } +} + +/// Build the run-to-completion request for a schedule: the SPEC's +/// image/env/volumes/limits/creds (same resolution the spawn path +/// uses — `${VAR}` interpolated now, failing loudly), with the +/// schedule's argv override when set. No public-path injection: a job +/// serves no HTTP. +async fn job_request( + state: &AppState, + spec: &ruscker_config::Spec, + schedule: &Schedule, +) -> anyhow::Result { + let image = spec + .container_image + .as_deref() + .ok_or_else(|| anyhow::anyhow!("spec {} has no container-image", spec.id))?; + let creds = crate::routes::proxy::resolve_creds(state, spec).await?; + let limits = crate::routes::proxy::limits_from_spec(spec); + let env = spec + .resolved_env_pairs() + .map_err(|e| anyhow::anyhow!("spec {} container-env: {e}", spec.id))?; + let cmd = schedule.cmd_override().or_else(|| spec.container_cmd.clone()); + + let mut req = ruscker_core::SpawnRequest::new(&spec.id, image) + .with_limits(limits) + .with_volumes(spec.volumes.clone().unwrap_or_default()) + .with_env(env); + if let Some(platform) = spec.platform.as_deref() { + req = req.with_platform(platform); + } + if let Some(cmd) = cmd { + req = req.with_cmd(cmd); + } + if let Some(net) = spec.effective_container_network() { + req = req.with_network(net); + } + let labels = spec.effective_labels(); + if !labels.is_empty() { + req = req.with_labels(labels); + } + if let Some(c) = creds { + req = req.with_creds(c); + } + Ok(req) +} + +/// One scheduler pass. Public-for-tests; `spawn` loops it. +async fn tick(state: &AppState) { + // HA: only the leader fires schedules (the DB claim in + // `mark_fired` backstops a split brain). + if !state.leader.is_leader().await { + return; + } + let (Some(db), Some(backend)) = (state.db.as_ref(), state.backend.as_ref()) else { + return; + }; + let schedules = match crate::db::schedules::list_all(db).await { + Ok(s) => s, + Err(e) => { + tracing::warn!(error = ?e, "scheduler: list schedules failed; skipping tick"); + return; + } + }; + let now = Utc::now(); + for schedule in schedules.into_iter().filter(|s| s.enabled) { + if next_occurrence(&schedule.cron, now).is_none() { + tracing::warn!( + schedule = schedule.id, + spec = %schedule.spec_id, + cron = %schedule.cron, + "scheduler: unparseable cron expression; schedule skipped" + ); + continue; + } + if !is_due(&schedule, now) { + continue; + } + // Claim BEFORE running — a crash mid-job must not double-fire, + // and a concurrent second scheduler loses this compare-and-set. + match crate::db::schedules::mark_fired(db, schedule.id, schedule.last_run_at, now).await { + Ok(true) => {} + Ok(false) => continue, // someone else claimed it + Err(e) => { + tracing::warn!(error = ?e, schedule = schedule.id, "scheduler: fire claim failed"); + continue; + } + } + let Some(spec) = crate::catalog::effective_specs_cached(state) + .await + .iter() + .find(|s| s.id == schedule.spec_id) + .cloned() + else { + tracing::warn!( + schedule = schedule.id, + spec = %schedule.spec_id, + "scheduler: schedule references a spec that no longer exists" + ); + let _ = crate::db::schedules::record_run( + db, schedule.id, now, RunStatus::Error, None, + "spec no longer exists in the catalog", None, + ) + .await; + continue; + }; + + let req = match job_request(state, &spec, &schedule).await { + Ok(r) => r, + Err(e) => { + report(state, &schedule, now, RunStatus::Error, None, &format!("{e}"), None).await; + continue; + } + }; + + // Detached: a long ETL must not block the next tick. The run + // itself is bounded by run_job's cap (per-schedule timeout is + // slice C, the column already exists). + let state = state.clone(); + let backend = backend.clone(); + tokio::spawn(async move { + tracing::info!(schedule = schedule.id, spec = %schedule.spec_id, "scheduled job starting"); + match backend.run_job(&req).await { + Ok(out) => { + let status = if out.exit_code == 0 { RunStatus::Ok } else { RunStatus::Failed }; + report( + &state, &schedule, now, status, Some(out.exit_code), + &out.log_tail.join("\n"), Some(out.duration_ms as i64), + ) + .await; + } + Err(e) => { + report(&state, &schedule, now, RunStatus::Error, None, &format!("{e}"), None) + .await; + } + } + }); + } +} + +/// Persist the run row and, for anything but `Ok`, raise the +/// `job-failed` alert (#930). +async fn report( + state: &AppState, + schedule: &Schedule, + started: DateTime, + status: RunStatus, + exit_code: Option, + log_tail: &str, + duration_ms: Option, +) { + if let Some(db) = state.db.as_ref() { + if let Err(e) = crate::db::schedules::record_run( + db, schedule.id, started, status, exit_code, log_tail, duration_ms, + ) + .await + { + tracing::warn!(error = ?e, schedule = schedule.id, "scheduler: record run failed"); + } + } + match status { + RunStatus::Ok => { + tracing::info!(schedule = schedule.id, spec = %schedule.spec_id, "scheduled job succeeded"); + } + RunStatus::Failed | RunStatus::Error => { + tracing::warn!( + schedule = schedule.id, + spec = %schedule.spec_id, + ?exit_code, + "scheduled job failed" + ); + state.alerts.notify(crate::alerts::AlertEvent { + kind: crate::alerts::AlertKind::JobFailed, + spec: schedule.spec_id.clone(), + replica: None, + message: match (status, exit_code) { + (RunStatus::Failed, Some(code)) => format!( + "scheduled job for `{}` (cron `{}`) exited with code {code}", + schedule.spec_id, schedule.cron + ), + _ => format!( + "scheduled job for `{}` (cron `{}`) could not run: {log_tail}", + schedule.spec_id, schedule.cron + ), + }, + }); + } + } +} + +/// Start the scheduler loop. Detached like the scaler — every tick is +/// idempotent (the DB claim makes fires exactly-once per occurrence). +pub fn spawn(state: AppState) -> JoinHandle<()> { + tokio::spawn(async move { + tracing::info!(tick = ?TICK, "job scheduler started"); + loop { + tokio::time::sleep(TICK).await; + tick(&state).await; + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sched(cron: &str, last: Option<&str>, created: &str) -> Schedule { + Schedule { + id: 1, + spec_id: "etl".into(), + cron: cron.into(), + cmd_json: None, + enabled: true, + timeout_secs: None, + last_run_at: last.map(|s| s.parse().unwrap()), + created_at: created.parse().unwrap(), + updated_at: created.parse().unwrap(), + } + } + + #[test] + fn due_when_an_occurrence_passed_since_the_anchor() { + let now: DateTime = "2026-07-13T03:05:00Z".parse().unwrap(); + // Daily at 03:00, last fired yesterday 03:00 → due. + assert!(is_due(&sched("0 3 * * *", Some("2026-07-12T03:00:30Z"), "2026-07-01T00:00:00Z"), now)); + // Already fired today at 03:00 → not due again. + assert!(!is_due(&sched("0 3 * * *", Some("2026-07-13T03:00:30Z"), "2026-07-01T00:00:00Z"), now)); + // Brand new schedule created at 02:00 today → 03:00 passed → due. + assert!(is_due(&sched("0 3 * * *", None, "2026-07-13T02:00:00Z"), now)); + // Brand new created at 04:00 → next is tomorrow → NOT due (no + // fire-on-create). + let later: DateTime = "2026-07-13T04:10:00Z".parse().unwrap(); + assert!(!is_due(&sched("0 3 * * *", None, "2026-07-13T04:00:00Z"), later)); + // Downtime over several occurrences collapses to one firing. + assert!(is_due(&sched("0 3 * * *", Some("2026-07-09T03:00:00Z"), "2026-07-01T00:00:00Z"), now)); + } + + #[test] + fn unparseable_cron_is_never_due() { + let now = Utc::now(); + assert!(!is_due(&sched("not a cron", None, "2026-07-01T00:00:00Z"), now)); + assert!(next_occurrence("61 99 * * *", now).is_none()); + } +} diff --git a/crates/ruscker-admin/src/lib.rs b/crates/ruscker-admin/src/lib.rs index 1301b866..8a3f6e88 100644 --- a/crates/ruscker-admin/src/lib.rs +++ b/crates/ruscker-admin/src/lib.rs @@ -29,6 +29,7 @@ pub mod crypto; pub mod db; pub mod i18n; pub mod images; +pub mod jobs; pub mod leader; pub mod logbuf; pub mod markdown; @@ -485,6 +486,14 @@ impl AdminServer { let _ = access_counter::spawn(self.state.access_counter.clone(), db); } + // Job scheduler (#986 slice B): leader-only cron runner for + // run-to-completion jobs. Needs BOTH a DB (schedules live + // there) and a backend (something must run the container). + if self.state.db.is_some() && self.state.backend.is_some() { + #[allow(clippy::let_underscore_future)] + let _ = jobs::spawn(self.state.clone()); + } + // Alert-webhook sender (#930): one task drains the alert queue // and POSTs to the operator-configured URL (settings table). // Detached like the loops above; no shutdown flush — alerts