starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: surface when a schedule runs less often than its cron configures

Browse Source
This commit is contained in:
hugocasa
2026-08-28 12:02:31 +02:00
parent 9213319a74
commit f3f01fb15c
7 changed files with 390 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json
Generated
+26
View File
@@ -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"
}
backend/.sqlx/query-84e1e33d30e473c343ee75a27428c2abb9cbc6021bac99b52d75fb883e386f73.json
Generated
+62
View File
@@ -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"
}
backend/windmill-api-schedule/src/lib.rs
+206 -6
View File
@@ -1000,6 +1000,71 @@ async fn list_schedule(
pub struct ScheduleWJobs {
pub path: String,
pub jobs: Option<Vec<serde_json::Value>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub interval_drift: Option<IntervalDrift>,
}
/// 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<Utc>],
schedule: &str,
cron_version: Option<&str>,
timezone: &str,
) -> Option<IntervalDrift> {
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<Utc>| {
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<Vec<ScheduleWJobs>> {
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<Vec<serde_json::Value>>,
push_times: Option<Vec<DateTime<Utc>>>,
enabled: bool,
schedule: String,
timezone: String,
cron_version: Option<String>,
}
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<IntervalDrift>,
}
async fn fetch_interval_drift(
db: &DB,
w_id: &str,
schedule: &Schedule,
) -> Result<Option<IntervalDrift>> {
// 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<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
@@ -1696,3 +1853,46 @@ pub struct SetEnabled {
// pub from: DateTime<Utc>,
// pub to: Option<DateTime<Utc>>,
// }
#[cfg(test)]
mod tests {
use super::*;
fn at(ts: &str) -> DateTime<Utc> {
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
);
}
}
backend/windmill-api/openapi.yaml
+23
View File
@@ -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
backend/windmill-common/src/utils.rs
+13 -5
View File
@@ -980,14 +980,22 @@ impl ScheduleType {
&self,
starting_from: &chrono::DateTime<chrono_tz::Tz>,
) -> chrono::DateTime<chrono_tz::Tz> {
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<chrono_tz::Tz>,
) -> Option<chrono::DateTime<chrono_tz::Tz>> {
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(),
}
}
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+33 -6
View File
@@ -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<string, any>): Promise<void> {
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}
<Alert type="warning" size="xs" title="Running less often than configured">
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.
</Alert>
{/if}
<div class="flex flex-col gap-1">
<Toggle
options={{
@@ -983,11 +1006,15 @@
/>
{/if}
{#if itemKind == 'script'}
<div class="flex gap-2 items-center mt-2">
<Toggle options={{ right: 'no overlap' }} checked={true} disabled /><Tooltip
>Currently, overlapping scripts' executions is not supported. The next execution
will be scheduled only after the previous iteration has completed.</Tooltip
>
<div class="flex flex-col gap-1 mt-2">
<Toggle options={{ right: 'no overlap' }} checked={true} disabled />
<p class="text-xs text-secondary">
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.
</p>
</div>
{/if}
{/if}
frontend/src/routes/(root)/(logged)/schedules/+page.svelte
+27 -2
View File
@@ -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}
<div class="border rounded-md divide-y">
{#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 @@
<div class="gap-2 items-center hidden md:flex">
<Badge large color="blue">{schedule}</Badge>
<Badge small color="gray">{timezone}</Badge>
{#if interval_drift}
<Popover notClickable>
<Badge small color="yellow"
>every ~{msToReadableTimeShort(interval_drift.effective_s * 1000)}</Badge
>
{#snippet text()}
<div>
This schedule is running about every {msToReadableTime(
interval_drift.effective_s * 1000
)} instead of every {msToReadableTime(
interval_drift.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.
</div>
{/snippet}
</Popover>
{/if}
</div>
<div class="hidden lg:flex flex-row gap-1 items-center">