From f3f01fb15c33c3604c2ffbca259e37bd20651fcb Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 28 Aug 2026 12:02:31 +0200 Subject: [PATCH] feat: surface when a schedule runs less often than its cron configures --- ...7a8b99225bd6256eb1be2877c67c919b1955c.json | 26 +++ ...8c2abb9cbc6021bac99b52d75fb883e386f73.json | 62 +++++ backend/windmill-api-schedule/src/lib.rs | 212 +++++++++++++++++- backend/windmill-api/openapi.yaml | 23 ++ backend/windmill-common/src/utils.rs | 18 +- .../schedules/ScheduleEditorInner.svelte | 39 +++- .../(root)/(logged)/schedules/+page.svelte | 29 ++- 7 files changed, 390 insertions(+), 19 deletions(-) create mode 100644 backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json create mode 100644 backend/.sqlx/query-84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73.json diff --git a/backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json b/backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json new file mode 100644 index 0000000000..c1424a6ed0 --- /dev/null +++ b/backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT created_at FROM v2_job\n WHERE workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2\n AND parent_job IS NULL AND runnable_path = $3\n AND created_at > $4\n ORDER BY created_at DESC\n LIMIT $5", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "created_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Timestamptz", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c" +} diff --git a/backend/.sqlx/query-84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73.json b/backend/.sqlx/query-84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73.json new file mode 100644 index 0000000000..e45703bc2f --- /dev/null +++ b/backend/.sqlx/query-84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73.json @@ -0,0 +1,62 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n schedule.path, schedule.enabled, schedule.schedule, schedule.timezone,\n schedule.cron_version, t.jobs, p.push_times FROM schedule,\n LATERAL(SELECT ARRAY(\n SELECT json_build_object('id', id, 'success', status = 'success', 'duration_ms', duration_ms)\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE trigger_kind = 'schedule'\n AND trigger = schedule.path\n AND c.workspace_id = $1\n AND j.workspace_id = $1\n AND parent_job IS NULL AND runnable_path = schedule.script_path\n AND status <> 'skipped'\n ORDER BY completed_at DESC\n LIMIT 20\n ) AS jobs) t,\n LATERAL(SELECT ARRAY(\n SELECT created_at\n FROM v2_job\n WHERE trigger_kind = 'schedule'\n AND trigger = schedule.path\n AND v2_job.workspace_id = $1\n AND parent_job IS NULL AND runnable_path = schedule.script_path\n AND created_at > schedule.edited_at\n ORDER BY created_at DESC\n LIMIT $5\n ) AS push_times) p\n WHERE workspace_id = $1 AND NOT starts_with(schedule.path, $4)\n ORDER BY edited_at DESC\n LIMIT $2 OFFSET $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "enabled", + "type_info": "Bool" + }, + { + "ordinal": 2, + "name": "schedule", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "timezone", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "cron_version", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "jobs", + "type_info": "JsonArray" + }, + { + "ordinal": 6, + "name": "push_times", + "type_info": "TimestamptzArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8", + "Int8", + "Text", + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + null, + null + ] + }, + "hash": "84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73" +} diff --git a/backend/windmill-api-schedule/src/lib.rs b/backend/windmill-api-schedule/src/lib.rs index 6ae902e984..75eeb4c70a 100644 --- a/backend/windmill-api-schedule/src/lib.rs +++ b/backend/windmill-api-schedule/src/lib.rs @@ -1000,6 +1000,71 @@ async fn list_schedule( pub struct ScheduleWJobs { pub path: String, pub jobs: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub interval_drift: Option, +} + +/// How often a schedule's runs are actually landing, reported only while that +/// is consistently slower than its cron asks for. A run still going when its +/// next slot comes round moves the run after it to a later slot, and nothing +/// else on the schedule records that its cadence changed. +#[derive(Serialize, Deserialize, Debug, PartialEq)] +pub struct IntervalDrift { + pub effective_s: i64, + pub configured_s: i64, +} + +/// Consecutive slots a schedule has to miss before we call it drift: one slow +/// run is not a change of cadence. +const DRIFT_MIN_MISSED_SLOTS: usize = 3; + +/// Runs to read back per schedule: one more than the slots to compare, since +/// each comparison spans a pair. +const DRIFT_SAMPLE_SIZE: i64 = DRIFT_MIN_MISSED_SLOTS as i64 + 1; + +/// `push_times` are the moments the last runs were queued, newest first. The +/// pusher anchors `find_next` on exactly that moment, so recomputing from it +/// recovers each run's slot, which the queue row no longer holds once the run +/// has completed. +fn detect_interval_drift( + push_times: &[DateTime], + schedule: &str, + cron_version: Option<&str>, + timezone: &str, +) -> Option { + if push_times.len() <= DRIFT_MIN_MISSED_SLOTS { + return None; + } + let tz = chrono_tz::Tz::from_str(timezone).ok()?; + let cron = ScheduleType::from_str(schedule, cron_version, false).ok()?; + let next_slot = |after: &DateTime| { + cron.find_next_opt(&after.with_timezone(&tz)) + .map(|slot| slot.with_timezone(&Utc)) + }; + + let mut effective = Vec::with_capacity(DRIFT_MIN_MISSED_SLOTS); + let mut configured = Vec::with_capacity(DRIFT_MIN_MISSED_SLOTS); + for pair in push_times[..=DRIFT_MIN_MISSED_SLOTS].windows(2) { + let previous = next_slot(&pair[1])?; + let ran_at = next_slot(&pair[0])?; + let kept_cadence = next_slot(&previous)?; + if ran_at <= kept_cadence { + return None; + } + effective.push((ran_at - previous).num_seconds()); + configured.push((kept_cadence - previous).num_seconds()); + } + Some(IntervalDrift { + effective_s: median(&mut effective), + configured_s: median(&mut configured), + }) +} + +/// Slot gaps vary under an irregular cron, so both intervals are reported as +/// the middle of the sample rather than an average a single outlier moves. +fn median(values: &mut [i64]) -> i64 { + values.sort_unstable(); + values[values.len() / 2] } async fn list_schedule_with_jobs( @@ -1010,13 +1075,25 @@ async fn list_schedule_with_jobs( ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let (per_page, offset) = paginate(pagination); - let rows = sqlx::query_as!(ScheduleWJobs, + struct ScheduleWJobsRow { + path: String, + jobs: Option>, + push_times: Option>>, + enabled: bool, + schedule: String, + timezone: String, + cron_version: Option, + } + let rows = sqlx::query_as!(ScheduleWJobsRow, // Query plan: // - use of the `ix_completed_job_workspace_id_started_at_new_2` index first, then; // - use of the `ix_v2_job_root_by_path` index; hence the `parent_job IS NULL` clause. // - both `workspace_id = $1` checks are required to hit both indexes. + // - `push_times` rides the `(workspace_id, runnable_path, created_at DESC)` index, + // hence its own `parent_job IS NULL` clause. "SELECT - schedule.path, t.jobs FROM schedule, + schedule.path, schedule.enabled, schedule.schedule, schedule.timezone, + schedule.cron_version, t.jobs, p.push_times FROM schedule, LATERAL(SELECT ARRAY( SELECT json_build_object('id', id, 'success', status = 'success', 'duration_ms', duration_ms) FROM v2_job_completed c JOIN v2_job j USING (id) @@ -1028,21 +1105,49 @@ async fn list_schedule_with_jobs( AND status <> 'skipped' ORDER BY completed_at DESC LIMIT 20 - ) AS jobs) t + ) AS jobs) t, + LATERAL(SELECT ARRAY( + SELECT created_at + FROM v2_job + WHERE trigger_kind = 'schedule' + AND trigger = schedule.path + AND v2_job.workspace_id = $1 + AND parent_job IS NULL AND runnable_path = schedule.script_path + AND created_at > schedule.edited_at + ORDER BY created_at DESC + LIMIT $5 + ) AS push_times) p WHERE workspace_id = $1 AND NOT starts_with(schedule.path, $4) ORDER BY edited_at DESC LIMIT $2 OFFSET $3", w_id, per_page as i64, offset as i64, - windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX + windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX, + DRIFT_SAMPLE_SIZE ) .fetch_all(&mut *tx) .await?; tx.commit().await?; let allowed = build_scope_path_predicate(&authed, "schedules", "read"); Ok(Json( - rows.into_iter().filter(|r| allowed(&r.path)).collect(), + rows.into_iter() + .filter(|r| allowed(&r.path)) + .map(|r| ScheduleWJobs { + interval_drift: r.push_times.as_deref().filter(|_| r.enabled).and_then( + |push_times| { + detect_interval_drift( + push_times, + &r.schedule, + r.cron_version.as_deref(), + &r.timezone, + ) + }, + ), + path: r.path, + jobs: r.jobs, + }) + .collect(), )) } @@ -1069,6 +1174,13 @@ async fn get_schedule( let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?; tx.commit().await?; + let deployed = match schedule_o { + Some(schedule) => Some(ScheduleWithIntervalDrift { + interval_drift: fetch_interval_drift(&db, &w_id, &schedule).await?, + schedule, + }), + None => None, + }; let overlay = overlay_or_draft_only( &db, &w_id, @@ -1076,13 +1188,58 @@ async fn get_schedule( UserDraftItemKind::TriggerSchedule, path, q.get_draft, - schedule_o, + deployed, || Error::NotFound(format!("Schedule not found at path {path}")), ) .await?; Ok(Json(overlay)) } +/// The schedule editor reads a schedule through this, so its drift surfaces +/// next to the cron expression that is running behind. +#[derive(Serialize)] +struct ScheduleWithIntervalDrift { + #[serde(flatten)] + schedule: Schedule, + #[serde(skip_serializing_if = "Option::is_none")] + interval_drift: Option, +} + +async fn fetch_interval_drift( + db: &DB, + w_id: &str, + schedule: &Schedule, +) -> Result> { + // A schedule that is off is not running behind, it is not running. + if !schedule.enabled { + return Ok(None); + } + // Query plan: `(workspace_id, runnable_path, created_at DESC)` index, hence the + // `parent_job IS NULL` clause. Runs from before the last edit are left out: they + // were queued under whatever cron the schedule had then. + let push_times = sqlx::query_scalar!( + "SELECT created_at FROM v2_job + WHERE workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2 + AND parent_job IS NULL AND runnable_path = $3 + AND created_at > $4 + ORDER BY created_at DESC + LIMIT $5", + w_id, + &schedule.path, + &schedule.script_path, + schedule.edited_at, + DRIFT_SAMPLE_SIZE + ) + .fetch_all(db) + .await?; + Ok(detect_interval_drift( + &push_times, + &schedule.schedule, + schedule.cron_version.as_deref(), + &schedule.timezone, + )) +} + async fn exists_schedule( Extension(db): Extension, Path((w_id, path)): Path<(String, StripPath)>, @@ -1696,3 +1853,46 @@ pub struct SetEnabled { // pub from: DateTime, // pub to: Option>, // } + +#[cfg(test)] +mod tests { + use super::*; + + fn at(ts: &str) -> DateTime { + DateTime::parse_from_rfc3339(ts) + .unwrap() + .with_timezone(&Utc) + } + + /// A 50s run on a 20s cron is queued again only once the next two slots + /// have gone by, so the cadence settles at 60s and stays there. + #[test] + fn reports_the_cadence_a_slow_run_settles_into() { + let push_times = [ + at("2024-01-01T00:03:10Z"), + at("2024-01-01T00:02:10Z"), + at("2024-01-01T00:01:10Z"), + at("2024-01-01T00:00:10Z"), + ]; + assert_eq!( + detect_interval_drift(&push_times, "*/20 * * * * *", Some("v2"), "UTC"), + Some(IntervalDrift { effective_s: 60, configured_s: 20 }) + ); + } + + /// One run overrunning its slot is not a change of cadence: the two runs + /// before it kept up. + #[test] + fn stays_quiet_when_only_the_last_run_overran() { + let push_times = [ + at("2024-01-01T00:01:50Z"), + at("2024-01-01T00:00:50Z"), + at("2024-01-01T00:00:30Z"), + at("2024-01-01T00:00:10Z"), + ]; + assert_eq!( + detect_interval_drift(&push_times, "*/20 * * * * *", Some("v2"), "UTC"), + None + ); + } +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 517c1d5105..71df7c75a3 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -17035,6 +17035,10 @@ paths: allOf: - $ref: "#/components/schemas/Schedule" - $ref: "#/components/schemas/UserDraftOverlay" + - type: object + properties: + interval_drift: + $ref: "#/components/schemas/ScheduleIntervalDrift" /w/{workspace}/schedules/exists/{path}: get: @@ -29771,6 +29775,25 @@ components: - id - success - duration_ms + interval_drift: + $ref: "#/components/schemas/ScheduleIntervalDrift" + + ScheduleIntervalDrift: + type: object + description: >- + How often the schedule's runs are actually landing, present only while + the last runs have consistently missed their next slot and the schedule + is therefore running less often than its cron asks for. + properties: + effective_s: + type: number + description: Observed seconds between the last runs + configured_s: + type: number + description: Seconds the cron expression puts between those same slots + required: + - effective_s + - configured_s ErrorHandler: type: string diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 62d47a34d7..aa9354b69a 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -980,14 +980,22 @@ impl ScheduleType { &self, starting_from: &chrono::DateTime, ) -> chrono::DateTime { + self.find_next_opt(starting_from) + .expect("cron: a schedule should have a next event") + } + + /// An expression can be parseable and still have no next occurrence (Feb + /// 30th), which the pusher has no answer for; a reader recomputing past + /// occurrences does, so it takes the fallible form. + pub fn find_next_opt( + &self, + starting_from: &chrono::DateTime, + ) -> Option> { match self { ScheduleType::Croner(croner_schedule) => croner_schedule .find_next_occurrence(starting_from, false) - .expect("cron: a schedule should have a next event"), - ScheduleType::Cron(schedule) => schedule - .after(starting_from) - .next() - .expect("cron: a schedule should have a next event"), + .ok(), + ScheduleType::Cron(schedule) => schedule.after(starting_from).next(), } } diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte index 7aa83cbbee..52217163e0 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte @@ -27,10 +27,18 @@ SettingService, type Retry, type Schedule, + type ScheduleIntervalDrift, type ErrorHandler } from '$lib/gen' import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores' - import { canWrite, emptyString, formatCron, sendUserToast, cronV1toV2 } from '$lib/utils' + import { + canWrite, + emptyString, + formatCron, + msToReadableTime, + sendUserToast, + cronV1toV2 + } from '$lib/utils' import { base } from '$lib/base' import Section from '$lib/components/Section.svelte' import { List, Loader2, Save, AlertTriangle } from 'lucide-svelte' @@ -123,6 +131,8 @@ let labels: string[] | undefined = $state(undefined) let description = $state('') let no_flow_overlap = $state(false) + // Measured on the deployed schedule, so it is read back rather than edited. + let intervalDrift: ScheduleIntervalDrift | undefined = $state(undefined) let tag: string | undefined = $state(undefined) let validCRON = $state(true) let isValid = $state(true) @@ -334,6 +344,9 @@ drawer?.openDrawer() runnable = undefined edit = false + // A new schedule has no run history, even when its fields were read + // from an existing one. + intervalDrift = undefined // No deployed baseline for a brand-new schedule. The editor instance // is reused across open() calls, so clear any baseline left by a prior // openEdit — otherwise the "unsaved changes" banner / dirty check would @@ -526,6 +539,7 @@ async function loadScheduleCfg(cfg: Record): Promise { loading = true + intervalDrift = cfg.interval_drift ?? undefined cronVersion = cfg.cron_version ?? 'v2' initialCronVersion = cronVersion isLatestCron = cronVersion == 'v2' @@ -919,6 +933,15 @@ bind:validCRON bind:cronVersion /> + {#if intervalDrift} + + This schedule is running about every {msToReadableTime( + intervalDrift.effective_s * 1000 + )} instead of every {msToReadableTime(intervalDrift.configured_s * 1000)}: each of the + last runs was still going when its next slot came round, so the run after it started + at a later slot. + + {/if}
{/if} {#if itemKind == 'script'} -
- Currently, overlapping scripts' executions is not supported. The next execution - will be scheduled only after the previous iteration has completed. +
+ +

+ Script runs never overlap: the next run is scheduled once the previous one has + completed, so a run that outlasts its interval pushes the next one to a later slot. + To keep the configured cadence, schedule a flow instead: a flow starts on time, and + its "no overlap of flows" setting skips a slot while the previous run is still + going. +

{/if} {/if} diff --git a/frontend/src/routes/(root)/(logged)/schedules/+page.svelte b/frontend/src/routes/(root)/(logged)/schedules/+page.svelte index e403c981de..509d718443 100644 --- a/frontend/src/routes/(root)/(logged)/schedules/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/schedules/+page.svelte @@ -6,7 +6,14 @@ type WorkspaceDeployUISettings, WorkspaceService } from '$lib/gen' - import { canWrite, displayDate, getLocalSetting, storeLocalSetting } from '$lib/utils' + import { + canWrite, + displayDate, + getLocalSetting, + msToReadableTime, + msToReadableTimeShort, + storeLocalSetting + } from '$lib/utils' import { withForkConflictRetry } from '$lib/utils/forkConflict' import { base } from '$app/paths' import CenteredPage from '$lib/components/CenteredPage.svelte' @@ -149,6 +156,7 @@ for (let schedule of schedules) { if (schedulesWithJobsByPath[schedule.path]) { schedule.jobs = schedulesWithJobsByPath[schedule.path].jobs + schedule.interval_drift = schedulesWithJobsByPath[schedule.path].interval_drift } } loadingSchedulesWithJobStats = false @@ -400,7 +408,7 @@ {/if} {:else if items?.length}
- {#each items.slice(0, nbDisplayed) as { path, error, summary, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, extra_perms, canWrite, jobs, paused_until, labels, inherited_labels, draft_only, is_draft } (path)} + {#each items.slice(0, nbDisplayed) as { path, error, summary, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, extra_perms, canWrite, jobs, interval_drift, paused_until, labels, inherited_labels, draft_only, is_draft } (path)} {@const hasDraft = getLocalDraftHint($workspaceStore, 'trigger_schedule', path) ?? is_draft} {@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`} @@ -466,6 +474,23 @@