diff --git a/backend/.sqlx/query-2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d.json b/backend/.sqlx/query-2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d.json new file mode 100644 index 0000000000..f9cf82a9fd --- /dev/null +++ b/backend/.sqlx/query-2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT s.workspace_id, s.path\n FROM schedule s JOIN workspace w ON w.id = s.workspace_id AND NOT w.deleted\n WHERE s.enabled IS TRUE\n AND NOT EXISTS (\n SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id)\n WHERE j.workspace_id = s.workspace_id\n AND j.trigger_kind = 'schedule'\n AND j.trigger = s.path\n AND j.runnable_path = s.script_path\n AND j.parent_job IS NULL\n )", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false + ] + }, + "hash": "2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 0392e128b3..d1c7a78059 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -108,7 +108,11 @@ use windmill_common::{ }; #[cfg(feature = "parquet")] use windmill_object_store::reload_object_store_setting; -use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload}; +use windmill_queue::{ + cancel_job, get_queued_job_v2, + schedule::{find_unarmed_schedules, rearm_schedule, RearmOutcome}, + SameWorkerPayload, +}; use windmill_worker::{ result_processor::handle_job_error, JobCompletedSender, JobIsolationLevel, OtelTracingProxySettings, SameWorkerSender, WorkspaceRegistryMap, BUNFIG_INSTALL_SCOPES, @@ -3286,6 +3290,15 @@ pub async fn monitor_db( } }; + // run every 30 iterations (~5min at the default LISTEN_NEW_EVENTS_INTERVAL_SEC). + let reconcile_unarmed_schedules_f = async { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(30) { + if let Some(db) = conn.as_sql() { + reconcile_unarmed_schedules(&db).await; + } + } + }; + // Poll git-sync repositories for new commits and pull them into the // workspace (repo → Windmill auto-pull). Runs every 2 iterations. let git_auto_pull_f = async { @@ -3349,9 +3362,144 @@ pub async fn monitor_db( cleanup_scheduled_job_deletions_f, git_auto_pull_f, pipeline_freshness_watchdog_f, + reconcile_unarmed_schedules_f, ); } +/// Advisory lock id ensuring only one server replica reconciles schedules at a +/// time (adjacent to GIT_AUTO_PULL_LOCK_ID). +const SCHEDULE_RECONCILE_LOCK_ID: i64 = 737_483_922; + +/// Consecutive reconciliation passes an enabled schedule must be observed with no +/// queued occurrence before it is re-armed. The next occurrence is pushed in the +/// same transaction that completes the previous one (or, for flows, on entry to +/// step 0), so an unarmed schedule is normally only ever a mid-flight push or a +/// push being retried. Requiring two passes keeps the reconciler from racing +/// those and double-pushing an occurrence. +const SCHEDULE_RECONCILE_STRIKES: u8 = 2; + +/// Most schedules re-armed in one pass, so a large first-pass backlog is drained +/// over several passes instead of enqueuing every occurrence at once. +const SCHEDULE_RECONCILE_MAX_PER_PASS: usize = 50; + +lazy_static::lazy_static! { + /// `(workspace_id, path)` -> consecutive passes seen with no queued occurrence. + /// Bounded by the number of enabled schedules; entries drop as soon as a + /// schedule is seen armed again. + static ref UNARMED_SCHEDULES: Mutex> = + Mutex::new(std::collections::HashMap::new()); +} + +/// Re-arm enabled schedules that have no queued occurrence. +/// +/// Every path that completes a scheduled job is supposed to push the next +/// occurrence atomically, but a run that dies through an abnormal path (a flow +/// whose status update fails and is later force-completed by zombie detection, +/// say) can skip that push and leave the schedule enabled yet dead forever. This +/// is the backstop: without it the only recovery is a manual disable/enable. +/// +/// Replicas each run their own passes (staggered by `rd_shift`) and each keep +/// their own strike tally, so the scan cost is per-replica. That is deliberate: +/// scanning inside the advisory lock is what makes a double-push impossible — +/// whoever holds it re-reads the unarmed set, so a schedule another replica just +/// re-armed is seen armed and its tally dropped, rather than pushed twice. +/// +/// Not an authorization boundary: it re-arms schedules across every workspace, so +/// this is a system caller (the monitor loop) only. +async fn reconcile_unarmed_schedules(db: &Pool) { + // Transaction-scoped advisory lock, not session-scoped: monitor_db runs under a + // 600s timeout, and if it fires the whole future is dropped mid-pass. A session + // lock taken on a pooled connection would then ride that connection back into the + // pool still held, wedging reconciliation on every replica until the process + // restarts. An xact lock is released when its transaction ends — including the + // rollback a dropped `Transaction` performs — so cancellation can't strand it. + // The tx is held open only to own the lock; the scan and re-arm run on separate + // pool connections. + let mut lock_tx = match db.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::error!("schedule reconcile: failed to begin lock tx: {e:#}"); + return; + } + }; + let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(SCHEDULE_RECONCILE_LOCK_ID) + .fetch_one(&mut *lock_tx) + .await + { + Ok(v) => v, + Err(e) => { + tracing::error!("schedule reconcile: advisory lock failed: {e:#}"); + return; + } + }; + if !locked { + // Another replica is already reconciling this tick. + return; + } + + if let Err(e) = reconcile_unarmed_schedules_inner(db).await { + tracing::error!("schedule reconcile: {e:#}"); + } + + // Ends the transaction and releases the xact lock; a plain drop would too. + if let Err(e) = lock_tx.rollback().await { + tracing::error!("schedule reconcile: releasing lock failed: {e:#}"); + } +} + +/// Record this pass's unarmed schedules against `seen` and return those that have +/// now struck out. An armed observation drops the schedule's tally entirely, so +/// the strikes a re-arm rests on are always consecutive. +fn strike_unarmed( + seen: &mut std::collections::HashMap<(String, String), u8>, + current: std::collections::HashSet<(String, String)>, +) -> Vec<(String, String)> { + seen.retain(|k, _| current.contains(k)); + current + .into_iter() + .filter(|k| { + let strikes = seen.entry(k.clone()).or_insert(0); + *strikes = strikes.saturating_add(1); + *strikes >= SCHEDULE_RECONCILE_STRIKES + }) + .collect() +} + +async fn reconcile_unarmed_schedules_inner(db: &Pool) -> error::Result<()> { + let current = find_unarmed_schedules(db).await?.into_iter().collect(); + let mut to_rearm = strike_unarmed(&mut UNARMED_SCHEDULES.lock().unwrap(), current); + + // The first pass on an instance that has never been swept can find a large + // backlog; re-arming it all at once would enqueue that whole backlog in one + // go. The overflow keeps its tally and is picked up next pass. + if to_rearm.len() > SCHEDULE_RECONCILE_MAX_PER_PASS { + tracing::warn!( + "schedule reconcile: {} schedules have no queued occurrence, re-arming {} this pass and the rest on later passes", + to_rearm.len(), + SCHEDULE_RECONCILE_MAX_PER_PASS + ); + to_rearm.truncate(SCHEDULE_RECONCILE_MAX_PER_PASS); + } + + for (w_id, path) in to_rearm { + match rearm_schedule(db, &w_id, &path).await { + Ok(outcome) => { + if outcome == RearmOutcome::Rearmed { + tracing::warn!( + "schedule reconcile: re-armed enabled schedule {path} in {w_id}, which had no queued occurrence" + ); + } + UNARMED_SCHEDULES.lock().unwrap().remove(&(w_id, path)); + } + Err(e) => tracing::error!( + "schedule reconcile: could not re-arm schedule {path} in {w_id}: {e:#}" + ), + } + } + Ok(()) +} + /// Advisory lock id ensuring only one server replica runs the git auto-pull /// poll at a time (adjacent to RESTART_LOCK_ID used for restart coordination). #[cfg(feature = "private")] @@ -5390,3 +5538,46 @@ mod retention_overrides_tests { assert!(parse_retention_overrides(over_cap).is_err()); } } + +#[cfg(test)] +mod strike_unarmed_tests { + use super::strike_unarmed; + use std::collections::{HashMap, HashSet}; + + fn key(path: &str) -> (String, String) { + ("ws".to_string(), path.to_string()) + } + + fn set(paths: &[&str]) -> HashSet<(String, String)> { + paths.iter().map(|p| key(p)).collect() + } + + /// The strike threshold is the only thing keeping the reconciler from racing + /// an in-flight push: `push_scheduled_job`'s own `already_exists` guard keys + /// on the same columns as the scan, so it is false by construction whenever a + /// schedule is found unarmed. + #[test] + fn rearms_only_after_consecutive_unarmed_passes() { + let mut seen = HashMap::new(); + assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty()); + assert_eq!(strike_unarmed(&mut seen, set(&["a"])), vec![key("a")]); + } + + #[test] + fn armed_observation_resets_the_tally() { + let mut seen = HashMap::new(); + assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty()); + // `a` is armed again on this pass, so its strike must not carry over. + assert!(strike_unarmed(&mut seen, set(&[])).is_empty()); + assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty()); + assert_eq!(strike_unarmed(&mut seen, set(&["a"])), vec![key("a")]); + } + + #[test] + fn tallies_are_per_schedule() { + let mut seen = HashMap::new(); + assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty()); + assert_eq!(strike_unarmed(&mut seen, set(&["a", "b"])), vec![key("a")]); + assert_eq!(strike_unarmed(&mut seen, set(&["b"])), vec![key("b")]); + } +} diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 2b5aef96a5..55944b08ca 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -597,6 +597,121 @@ pub async fn push_scheduled_job<'c>( Ok(tx) // TODO: Bubble up pushed UUID from here } +/// Enabled schedules with no occurrence in the queue, as `(workspace_id, path)`. +/// +/// Every path that completes a scheduled job pushes the next occurrence in the +/// same transaction (for flows, on entry to step 0), so an enabled schedule +/// always has a queued occurrence — a run in progress is itself one. A run that +/// dies through an abnormal path can skip that push though, leaving the schedule +/// enabled yet dead until it is manually disabled and re-enabled. This is how the +/// monitor spots that state; see `rearm_schedule` for the recovery. +/// +/// Not an authorization boundary: it reports schedules across every workspace, so +/// this is for system callers (the monitor's reconciliation pass) only and its +/// result must never be returned to a user unfiltered. +pub async fn find_unarmed_schedules(db: &DB) -> Result> { + let rows = sqlx::query!( + // Query plan: the anti-join builds from `v2_job_queue` (only pending and + // running jobs) rather than probing `v2_job` once per schedule. + "SELECT s.workspace_id, s.path + FROM schedule s JOIN workspace w ON w.id = s.workspace_id AND NOT w.deleted + WHERE s.enabled IS TRUE + AND NOT EXISTS ( + SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id) + WHERE j.workspace_id = s.workspace_id + AND j.trigger_kind = 'schedule' + AND j.trigger = s.path + AND j.runnable_path = s.script_path + AND j.parent_job IS NULL + )" + ) + .fetch_all(db) + .await?; + Ok(rows.into_iter().map(|r| (r.workspace_id, r.path)).collect()) +} + +#[derive(Debug, PartialEq, Eq)] +pub enum RearmOutcome { + /// The next occurrence was pushed. + Rearmed, + /// Nothing to do: the schedule was deleted or disabled since it was found. + NoOp, +} + +/// Push the next occurrence of a schedule that has none queued. +/// +/// Only ever starts a schedule, never stops one: re-arming something that did not +/// need it costs one extra run, whereas wrongly disabling one is the silent +/// permanent stoppage this whole mechanism exists to prevent. So an occurrence +/// that cannot be pushed is logged and left alone — the schedule is already not +/// running, and `try_schedule_next_job` still disables on the completion path, +/// where the population is limited to actively-cycling schedules. Keep it that +/// way: this sweeps *every* enabled schedule, including ones broken long before +/// this code existed and never swept before. +/// +/// Not an authorization boundary: it pushes under the schedule's own +/// `permissioned_as` identity for any `(w_id, path)`, so this is for system +/// callers (the monitor's reconciliation pass) only. A caller acting for a user +/// MUST already have enforced their permissions on `w_id` and `path`. +pub async fn rearm_schedule(db: &DB, w_id: &str, path: &str) -> Result { + let mut tx = db.begin().await?; + // Lock the row for the whole push: an edit or a disable committing between the + // read and the push would otherwise leave a queued occurrence for a schedule + // that is disabled, or one built from superseded settings. + let schedule = sqlx::query_as::<_, Schedule>( + "SELECT workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, args, extra_perms, email, permissioned_as, error, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, on_success, on_success_extra_args, ws_error_handler_muted, retry, no_flow_overlap, summary, description, tag, paused_until, cron_version, dynamic_skip, labels FROM schedule WHERE path = $1 AND workspace_id = $2 FOR UPDATE", + ) + .bind(path) + .bind(w_id) + .fetch_optional(&mut *tx) + .await?; + let Some(schedule) = schedule else { + return Ok(RearmOutcome::NoOp); + }; + if !schedule.enabled { + return Ok(RearmOutcome::NoOp); + } + // Re-check for a queued occurrence now that the row is locked: a normal + // completion, an edit, or a re-enable could have pushed one between the unarmed + // scan and this lock. push_scheduled_job only dedups the exact computed + // scheduled_for, so re-arming a schedule that has since become armed and crossed a + // cron boundary would queue a second root occurrence. Mirrors the anti-join in + // find_unarmed_schedules. + let already_armed: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id) + WHERE j.workspace_id = $1 + AND j.trigger_kind = 'schedule' + AND j.trigger = $2 + AND j.runnable_path = $3 + AND j.parent_job IS NULL + )", + ) + .bind(w_id) + .bind(path) + .bind(&schedule.script_path) + .fetch_one(&mut *tx) + .await?; + if already_armed { + return Ok(RearmOutcome::NoOp); + } + match push_scheduled_job(db, tx, &schedule, None, None).await { + Ok(tx) => { + tx.commit().await?; + Ok(RearmOutcome::Rearmed) + } + // An occurrence that can never be pushed (runnable gone, quota blown) is + // reported, not acted on — see the note above on why this never disables. + Err(err @ (error::Error::NotFound(_) | error::Error::QuotaExceeded(_))) => { + tracing::error!( + "Could not re-arm schedule {path} in {w_id}: {err}. Leaving it enabled; it will not run until the cause is fixed." + ); + Ok(RearmOutcome::NoOp) + } + Err(err) => Err(err), + } +} + pub async fn get_schedule_opt<'c>( e: impl PgExecutor<'c>, w_id: &str, diff --git a/backend/windmill-queue/tests/schedule_push.rs b/backend/windmill-queue/tests/schedule_push.rs index e11ced23bc..9ac2e3a143 100644 --- a/backend/windmill-queue/tests/schedule_push.rs +++ b/backend/windmill-queue/tests/schedule_push.rs @@ -10,7 +10,9 @@ mod schedule_push { use windmill_common::scripts::ScriptHash; use windmill_common::users::username_to_permissioned_as; use windmill_queue::jobs::{try_schedule_next_job, MiniCompletedJob}; - use windmill_queue::schedule::push_scheduled_job; + use windmill_queue::schedule::{ + find_unarmed_schedules, push_scheduled_job, rearm_schedule, RearmOutcome, + }; fn make_schedule(overrides: impl FnOnce(&mut Schedule)) -> Schedule { let mut s = Schedule { @@ -1762,4 +1764,121 @@ mod schedule_push { assert!(!row_exists, "managed schedule row must be deleted"); Ok(()) } + + // ----------------------------------------------------------------------- + // find_unarmed_schedules / rearm_schedule: recovery for a schedule left + // enabled with no queued occurrence (a run that died on an abnormal path + // skipped its next-occurrence push). Without this the chain stays dead + // until the schedule is manually disabled and re-enabled. + // ----------------------------------------------------------------------- + + async fn insert_schedule(db: &Pool, path: &str, script_path: &str, enabled: bool) { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap, permissioned_as) + VALUES ('test-workspace', $1, 'test-user', now(), '0 0 */5 * * *', 'UTC', $3, $2, false, 'test@windmill.dev', '{}', false, true, 'u/test-user')", + ) + .bind(path) + .bind(script_path) + .bind(enabled) + .execute(db) + .await + .unwrap(); + } + + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn test_find_unarmed_schedules(db: Pool) -> anyhow::Result<()> { + insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await; + insert_schedule(&db, "f/system/disabled", "f/system/test_script", false).await; + + // No occurrence queued yet: the enabled schedule is unarmed, the disabled one is ignored. + assert_eq!( + find_unarmed_schedules(&db).await?, + vec![( + "test-workspace".to_string(), + "f/system/test_schedule".to_string() + )] + ); + + // Once an occurrence is queued it is armed and must not be reported — + // re-arming it would double-push the occurrence. + let tx = db.begin().await?; + let tx = push_scheduled_job(&db, tx, &make_schedule(|_| {}), None, None).await?; + tx.commit().await?; + assert_eq!(count_queued_jobs(&db).await, 1); + assert!(find_unarmed_schedules(&db).await?.is_empty()); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn test_rearm_schedule_pushes_next_occurrence(db: Pool) -> anyhow::Result<()> { + insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await; + + assert_eq!( + rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?, + RearmOutcome::Rearmed + ); + + assert_eq!(count_queued_jobs(&db).await, 1); + assert!(find_unarmed_schedules(&db).await?.is_empty()); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn test_rearm_schedule_skips_disabled(db: Pool) -> anyhow::Result<()> { + // A disable that lands between the scan and the re-arm must win: pushing an + // occurrence for a disabled schedule would resurrect a schedule the user + // just turned off. + insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", false).await; + + assert_eq!( + rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?, + RearmOutcome::NoOp + ); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn test_rearm_schedule_never_disables(db: Pool) -> anyhow::Result<()> { + insert_schedule(&db, "f/system/bad_schedule", "f/system/nonexistent", true).await; + + // Reconciliation only ever starts a schedule. An unpushable occurrence is + // reported and left alone: this sweeps every enabled schedule in the + // instance, so disabling here would turn a wrong invariant into the exact + // silent stoppage the reconciler exists to undo. + assert_eq!( + rearm_schedule(&db, "test-workspace", "f/system/bad_schedule").await?, + RearmOutcome::NoOp + ); + + assert_eq!(count_queued_jobs(&db).await, 0); + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "reconciliation must never disable a schedule"); + assert!(error.is_none()); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn test_rearm_schedule_skips_already_armed(db: Pool) -> anyhow::Result<()> { + // An occurrence can be queued (a normal completion, an edit, a re-enable) + // between the unarmed scan and rearm_schedule acquiring the row lock. Re-arming + // then would double-push, since push_scheduled_job only dedups the exact + // computed scheduled_for. + insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await; + let tx = db.begin().await?; + let tx = push_scheduled_job(&db, tx, &make_schedule(|_| {}), None, None).await?; + tx.commit().await?; + assert_eq!(count_queued_jobs(&db).await, 1); + + assert_eq!( + rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?, + RearmOutcome::NoOp + ); + assert_eq!(count_queued_jobs(&db).await, 1); + Ok(()) + } }