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

perf: cap the drift sample's scan and restore it when a schedule is re-enabled

Browse Source
This commit is contained in:
hugocasa
2026-08-28 13:18:58 +02:00
parent 3cbd3359cf
commit da150d57a4
6 changed files with 85 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-543607315f3bd60037b275c5b6e7a8b99225bd6256eb1be2877c67c919b1955c.json
Generated
-26
View File
@@ -1,26 +0,0 @@
{
"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-7dc5ca6aca1ba55c7448d1788f5afc84afb4604bf6a85797e781347734ef3cd8.json → backend/.sqlx/query-7d67aebe32cc5310a92f510bd2dd742198c8a492783348d457e6811601b21371.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n schedule.path, 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 schedule.enabled\n AND 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",
"query": "SELECT\n schedule.path, 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 FROM (\n SELECT created_at, trigger, trigger_kind\n FROM v2_job\n WHERE schedule.enabled\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 $6\n ) recent\n WHERE trigger_kind = 'schedule' AND trigger = schedule.path\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": [
{
@@ -40,6 +40,7 @@
"Int8",
"Int8",
"Text",
"Int8",
"Int8"
]
},
@@ -52,5 +53,5 @@
null
]
},
"hash": "7dc5ca6aca1ba55c7448d1788f5afc84afb4604bf6a85797e781347734ef3cd8"
"hash": "7d67aebe32cc5310a92f510bd2dd742198c8a492783348d457e6811601b21371"
}
backend/.sqlx/query-847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7.json
Generated
+27
View File
@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_at FROM (\n SELECT created_at, trigger, trigger_kind\n FROM v2_job\n WHERE workspace_id = $1\n AND parent_job IS NULL AND runnable_path = $3\n AND created_at > $4\n ORDER BY created_at DESC\n LIMIT $6\n ) recent\n WHERE trigger_kind = 'schedule' AND trigger = $2\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",
"Int8"
]
},
"nullable": [
false
]
},
"hash": "847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7"
}
backend/windmill-api-schedule/src/lib.rs
+38 -21
View File
@@ -1022,6 +1022,13 @@ const DRIFT_MIN_MISSED_SLOTS: usize = 3;
/// each comparison spans a pair.
const DRIFT_SAMPLE_SIZE: i64 = DRIFT_MIN_MISSED_SLOTS as i64 + 1;
/// Root jobs of the runnable the sample may walk before giving up. There is no
/// index on `trigger`, so the sample rides the one on `runnable_path` and finds
/// the schedule's own runs by filter: a runnable that is also busy outside this
/// schedule would otherwise be read back as far as its fourth scheduled run.
/// Past the budget the schedule reports nothing rather than costing the walk.
const DRIFT_SCAN_BUDGET: i64 = 100;
/// `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
@@ -1137,11 +1144,10 @@ async fn list_schedule_with_jobs(
// - 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. `trigger` is only a filter on that
// scan, so the `edited_at` bound is what ends the walk: without it, a path that
// also carries non-schedule traffic is read to the end of its range whenever it
// holds fewer than the sampled number of scheduled runs. Nothing writes
// `edited_at` after the insert, so no run of the schedule can predate it.
// hence its own `parent_job IS NULL` clause, and is capped by DRIFT_SCAN_BUDGET.
// The `edited_at` bound is not that cap: it drops the runs of whatever schedule
// last held this path, which carry the same `trigger`. Nothing writes `edited_at`
// after the insert, so no run of this schedule can predate it.
"SELECT
schedule.path, schedule.schedule, schedule.timezone,
schedule.cron_version, t.jobs, p.push_times FROM schedule,
@@ -1158,14 +1164,17 @@ async fn list_schedule_with_jobs(
LIMIT 20
) AS jobs) t,
LATERAL(SELECT ARRAY(
SELECT created_at
FROM v2_job
WHERE schedule.enabled
AND 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
SELECT created_at FROM (
SELECT created_at, trigger, trigger_kind
FROM v2_job
WHERE schedule.enabled
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 $6
) recent
WHERE trigger_kind = 'schedule' AND trigger = schedule.path
ORDER BY created_at DESC
LIMIT $5
) AS push_times) p
@@ -1176,7 +1185,8 @@ async fn list_schedule_with_jobs(
per_page as i64,
offset as i64,
windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX,
DRIFT_SAMPLE_SIZE
DRIFT_SAMPLE_SIZE,
DRIFT_SCAN_BUDGET
)
.fetch_all(&mut *tx)
.await?;
@@ -1282,20 +1292,27 @@ async fn fetch_interval_drift(
if !schedule.enabled {
return Ok(None);
}
// Same index, and the same `edited_at` bound for the same reason, as the sample
// in `list_schedule_with_jobs`.
// Same index, same budget and same `edited_at` bound, for the same reasons as
// the sample in `list_schedule_with_jobs`.
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
"SELECT created_at FROM (
SELECT created_at, trigger, trigger_kind
FROM v2_job
WHERE workspace_id = $1
AND parent_job IS NULL AND runnable_path = $3
AND created_at > $4
ORDER BY created_at DESC
LIMIT $6
) recent
WHERE trigger_kind = 'schedule' AND trigger = $2
ORDER BY created_at DESC
LIMIT $5",
w_id,
&schedule.path,
&schedule.script_path,
schedule.edited_at,
DRIFT_SAMPLE_SIZE
DRIFT_SAMPLE_SIZE,
DRIFT_SCAN_BUDGET
)
.fetch_all(db)
.await?;
backend/windmill-common/src/utils.rs
+8 -10
View File
@@ -980,20 +980,15 @@ impl ScheduleType {
&self,
starting_from: &chrono::DateTime<chrono_tz::Tz>,
) -> 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"),
}
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.
/// occurrences does, so it takes the fallible form. What croner had to say
/// about it is logged here rather than carried out, so that the pusher's
/// panic is still preceded by the reason.
pub fn find_next_opt(
&self,
starting_from: &chrono::DateTime<chrono_tz::Tz>,
@@ -1001,6 +996,9 @@ impl ScheduleType {
match self {
ScheduleType::Croner(croner_schedule) => croner_schedule
.find_next_occurrence(starting_from, false)
.inspect_err(|err| {
tracing::error!("cron: no occurrence after {starting_from}: {err:?}")
})
.ok(),
ScheduleType::Cron(schedule) => schedule.after(starting_from).next(),
}
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+9 -2
View File
@@ -733,8 +733,15 @@
enabled = previousEnabled
return
}
if (!nEnabled) {
// The measurement describes a schedule that is running.
// The measurement describes the deployed schedule, which just changed:
// disabling drops it, and enabling brings back what the runs still show.
try {
const deployed = await ScheduleService.getSchedule({
workspace: wsId ?? '',
path: initialPath
})
intervalDrift = (deployed as any).interval_drift
} catch {
intervalDrift = undefined
}
sendUserToast(`${nEnabled ? 'enabled' : 'disabled'} schedule ${initialPath}`)