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: flag a schedule whose runs outlast the interval between its slots

Browse Source
This commit is contained in:
hugocasa
2026-08-28 17:10:37 +02:00
parent 2966c7a158
commit 17b057f1f3
10 changed files with 184 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0600e0d5cf4aa29b287ce57cf379cc84f7281371b2cccbe2bb59437f645a559c.json
Generated
-24
View File
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_at\n FROM trigger_history\n WHERE workspace_id = $1 AND trigger_kind = $2 AND path = $3\n AND (changes ?| array['schedule', 'timezone', 'cron_version', 'dynamic_skip']\n OR changes -> 'truncated_fields'\n ?| array['schedule', 'timezone', 'cron_version', 'dynamic_skip'])\n ORDER BY id DESC\n LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "0600e0d5cf4aa29b287ce57cf379cc84f7281371b2cccbe2bb59437f645a559c"
}
backend/.sqlx/query-0f141ca6a58901dee1ddcf25694d76a4e5702c0324a27ef3a5740ef5d98d8946.json
Generated
-31
View File
@@ -1,31 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n schedule.path, t.jobs 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 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": "jobs",
"type_info": "JsonArray"
}
],
"parameters": {
"Left": [
"Text",
"Int8",
"Int8",
"Text"
]
},
"nullable": [
false,
null
]
},
"hash": "0f141ca6a58901dee1ddcf25694d76a4e5702c0324a27ef3a5740ef5d98d8946"
}
backend/.sqlx/query-847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7.json
Generated
-27
View File
@@ -1,27 +0,0 @@
{
"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/.sqlx/query-e1e0e444e6620e3ddb3d22ffbd24a3b637230a4125314b89688d7dc07acb2f4c.json
Generated
+61
View File
@@ -0,0 +1,61 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n schedule.path, schedule.schedule, schedule.timezone, schedule.cron_version,\n schedule.is_flow, schedule.dynamic_skip, t.jobs 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 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": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "cron_version",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 5,
"name": "dynamic_skip",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "jobs",
"type_info": "JsonArray"
}
],
"parameters": {
"Left": [
"Text",
"Int8",
"Int8",
"Text"
]
},
"nullable": [
false,
false,
false,
true,
false,
true,
null
]
},
"hash": "e1e0e444e6620e3ddb3d22ffbd24a3b637230a4125314b89688d7dc07acb2f4c"
}
backend/windmill-api-schedule/src/lib.rs
+45 -286
View File
@@ -127,7 +127,6 @@ pub fn workspaced_service() -> Router {
.route("/list", get(list_schedule))
.route("/list_with_jobs", get(list_schedule_with_jobs))
.route("/get/{*path}", get(get_schedule))
.route("/interval_drift/{*path}", get(get_interval_drift))
.route("/exists/{*path}", get(exists_schedule))
.route("/create", post(create_schedule))
.route("/update/{*path}", post(edit_schedule))
@@ -1001,136 +1000,35 @@ async fn list_schedule(
pub struct ScheduleWJobs {
pub path: String,
pub jobs: Option<Vec<serde_json::Value>>,
}
/// 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,
}
/// What the reader needs on top of the numbers: which way out to offer. A flow
/// already queues its next run when the previous one starts, so its runs are
/// starting late rather than overrunning. Answered from the deployed row, so the
/// editor never has to derive it from the form a draft may be sitting in.
#[derive(Serialize, Deserialize, Debug)]
pub struct IntervalDriftReport {
#[serde(flatten)]
pub drift: IntervalDrift,
/// Shortest gap the cron asks for, so a caller can tell whether the runs it
/// was handed are outlasting it. `None` when the expression cannot be read.
#[serde(skip_serializing_if = "Option::is_none")]
pub interval_s: Option<i64>,
/// A flow, or a script with a skip handler (pushed as a single step flow),
/// queues its next run as the previous one starts, so a long run does not
/// push the following one out the way it does for a plain script.
pub queues_next_run_at_start: bool,
}
/// 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;
/// Occurrences to look ahead over when sizing the interval.
const INTERVAL_SAMPLE_SLOTS: usize = 5;
/// 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;
/// 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,
/// which for a slow schedule on a busy script is unbounded. This covers that
/// case with room to spare, and is affordable because the sample is only taken
/// when someone opens one schedule, never once per row of a listing.
const DRIFT_SCAN_BUDGET: i64 = 2000;
/// `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.
///
/// Not every push follows the cadence: enabling, editing and re-arming each
/// queue one against the wall clock, and a push made while the schedule is
/// paused anchors on `paused_until` rather than on its own timestamp. Holding
/// out for a whole run of missed slots is what keeps one of those from reading
/// as a change of cadence.
fn detect_interval_drift(
push_times: &[DateTime<Utc>],
/// The *shortest* gap, not the average: under an irregular expression a run can
/// outlast the tight gaps while still fitting inside the mean, and it is the
/// tight ones it will skip.
fn configured_interval_s(
schedule: &str,
cron_version: Option<&str>,
timezone: &str,
) -> Option<IntervalDrift> {
if push_times.len() <= DRIFT_MIN_MISSED_SLOTS {
return None;
}
) -> Option<i64> {
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))
};
// The two intervals are reported from the pair with the least drift in the
// sample, and always from the same pair so they stay comparable. A sample
// reaching over a pause, or over a stretch with the schedule turned off,
// carries one enormous gap; the mildest pair is the one still describing
// how the schedule runs when it is running.
let mut mildest: Option<IntervalDrift> = None;
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;
}
let pair_drift = IntervalDrift {
effective_s: (ran_at - previous).num_seconds(),
configured_s: (kept_cadence - previous).num_seconds(),
};
if mildest
.as_ref()
.is_none_or(|mildest| pair_drift.effective_s < mildest.effective_s)
{
mildest = Some(pair_drift);
}
}
mildest
}
/// When this schedule's cron expression last changed, if it ever has.
///
/// Runs queued under a previous expression measure as drift against the one
/// that replaced it, so a sample reaching back past the change has to be
/// dropped. `schedule.edited_at` cannot draw that line: nothing updates it
/// after the insert. The trigger history can, and is only read once a sample
/// has already come out as drifting.
///
/// `dynamic_skip` counts as such a change even though it leaves the arithmetic
/// alone: it moves the tick from completion-time to start-time, which is what
/// the reader tells the user their drift is made of.
///
/// Compare strictly against it: the edit queues a tick of its own in the same
/// transaction, so that first run under the new expression carries the very
/// timestamp recorded here.
///
/// Reads on the privileged pool, which the caller has earned with an RLS read
/// of the schedule row itself.
async fn cron_changed_at(db: &DB, w_id: &str, path: &str) -> Result<Option<DateTime<Utc>>> {
let changed_at = sqlx::query_scalar!(
// An entry too large to record field by field keeps only the names, so both
// shapes have to be asked.
"SELECT created_at
FROM trigger_history
WHERE workspace_id = $1 AND trigger_kind = $2 AND path = $3
AND (changes ?| array['schedule', 'timezone', 'cron_version', 'dynamic_skip']
OR changes -> 'truncated_fields'
?| array['schedule', 'timezone', 'cron_version', 'dynamic_skip'])
ORDER BY id DESC
LIMIT 1",
w_id,
SCHEDULE_TRIGGER_KIND,
path
)
.fetch_optional(db)
.await?;
Ok(changed_at)
let slots = cron.upcoming(tz, INTERVAL_SAMPLE_SLOTS).ok()?;
slots
.windows(2)
.map(|pair| (pair[1] - pair[0]).num_seconds())
.min()
.filter(|gap| *gap > 0)
}
async fn list_schedule_with_jobs(
@@ -1141,13 +1039,23 @@ 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>>,
schedule: String,
timezone: String,
cron_version: Option<String>,
is_flow: bool,
dynamic_skip: 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.
"SELECT
schedule.path, t.jobs FROM schedule,
schedule.path, schedule.schedule, schedule.timezone, schedule.cron_version,
schedule.is_flow, schedule.dynamic_skip, t.jobs 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)
@@ -1173,7 +1081,19 @@ async fn list_schedule_with_jobs(
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_s: configured_interval_s(
&r.schedule,
r.cron_version.as_deref(),
&r.timezone,
),
queues_next_run_at_start: r.is_flow || r.dynamic_skip.is_some(),
path: r.path,
jobs: r.jobs,
})
.collect(),
))
}
@@ -1214,90 +1134,6 @@ async fn get_schedule(
Ok(Json(overlay))
}
/// Measured, not configured: kept off `get_schedule`, whose response is the
/// schedule's configuration and is diffed for deploys and read by exporters that
/// fan out over every schedule in a workspace.
async fn get_interval_drift(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Option<IntervalDriftReport>> {
let path = path.to_path();
check_scopes(&authed, || format!("schedules:read:{}", path))?;
let mut tx = user_db.begin(&authed).await?;
let schedule = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?;
tx.commit().await?;
let schedule = not_found_if_none(schedule, "Schedule", path)?;
let report = fetch_interval_drift(&db, &w_id, &schedule)
.await?
.map(|drift| IntervalDriftReport {
drift,
queues_next_run_at_start: schedule.is_flow || schedule.dynamic_skip.is_some(),
});
Ok(Json(report))
}
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. Nor is one
// told to skip: a skip handler and `no_flow_overlap` both exist to drop runs, so
// for those two the cadence the cron asks for was never the promise.
// `no_flow_overlap` is only consulted by the flow runtime, so on a plain script
// schedule it is inert and must not suppress anything.
if !schedule.enabled
|| (schedule.is_flow && schedule.no_flow_overlap)
|| schedule.dynamic_skip.is_some()
{
return Ok(None);
}
// Query plan: `(workspace_id, runnable_path, created_at DESC)` index, hence the
// `parent_job IS NULL` clause, with the walk 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.
let push_times = sqlx::query_scalar!(
"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_SCAN_BUDGET
)
.fetch_all(db)
.await?;
let drift = detect_interval_drift(
&push_times,
&schedule.schedule,
schedule.cron_version.as_deref(),
&schedule.timezone,
);
if drift.is_some() {
if let Some(oldest) = push_times.last() {
let changed_at = cron_changed_at(db, w_id, &schedule.path).await?;
if changed_at.is_some_and(|changed_at| *oldest < changed_at) {
return Ok(None);
}
}
}
Ok(drift)
}
async fn exists_schedule(
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
@@ -1911,80 +1747,3 @@ 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 })
);
}
/// Twice a day, so consecutive runs sit 8 hours apart and then 16. Every
/// run here landed on the slot after its predecessor's, the overnight one
/// included: a gap the cron itself asks for is not drift.
#[test]
fn stays_quiet_on_the_long_gaps_an_irregular_cron_leaves() {
let push_times = [
at("2024-01-02T17:00:05Z"),
at("2024-01-02T09:00:05Z"),
at("2024-01-01T17:00:05Z"),
at("2024-01-01T09:00:05Z"),
];
assert_eq!(
detect_interval_drift(&push_times, "0 0 9,17 * * *", Some("v2"), "UTC"),
None
);
}
/// The schedule spent most of this window turned off. The gap that leaves
/// is not the cadence, so the interval reported is the one the runs on
/// either side of it settled into.
#[test]
fn reports_past_the_gap_a_pause_leaves() {
let push_times = [
at("2024-01-01T04:01:10Z"),
at("2024-01-01T04:00: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
+11 -48
View File
@@ -17036,27 +17036,6 @@ paths:
- $ref: "#/components/schemas/Schedule"
- $ref: "#/components/schemas/UserDraftOverlay"
/w/{workspace}/schedules/interval_drift/{path}:
get:
summary: get how far a schedule's runs have drifted from its cron
operationId: getScheduleIntervalDrift
tags:
- schedule
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"200":
description: >-
The measurement, or null when the schedule is keeping up, is
disabled, is configured to skip runs, or has not run often enough
under its current cron to tell
content:
application/json:
schema:
$ref: "#/components/schemas/ScheduleIntervalDrift"
nullable: true
/w/{workspace}/schedules/exists/{path}:
get:
summary: does schedule exists
@@ -29792,33 +29771,17 @@ components:
- id
- success
- duration_ms
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: integer
description: >-
Observed seconds between two consecutive runs, taken from the pair
that drifted least, so that a sample spanning a pause reports the
cadence rather than the pause
configured_s:
type: integer
description: Seconds the cron expression puts between those same two slots
queues_next_run_at_start:
type: boolean
description: >-
True when the schedule queues its next run as the previous one
starts (a flow, or a script with a skip handler), which means its
runs are starting late rather than overrunning
required:
- effective_s
- configured_s
- queues_next_run_at_start
interval_s:
type: integer
description: >-
Shortest gap the cron expression asks for, in seconds. A run
that outlasts it pushes the next one to a later slot.
queues_next_run_at_start:
type: boolean
description: >-
True when the schedule queues its next run as the previous one
starts (a flow, or a script with a skip handler), so a long run
does not push the following one out
ErrorHandler:
type: string
backend/windmill-common/src/utils.rs
+5 -18
View File
@@ -980,27 +980,14 @@ 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. 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>,
) -> Option<chrono::DateTime<chrono_tz::Tz>> {
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(),
.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"),
}
}
frontend/src/lib/components/schedules/scheduleDrift.ts
+26
View File
@@ -0,0 +1,26 @@
/**
* Whether a schedule's runs are taking longer than the gap between its slots.
*
* A plain script schedule queues its next run only once the previous one has
* completed, so a run that outlasts the interval necessarily pushes the next
* one to a later slot: the schedule quietly runs less often than its cron says.
* Schedules that queue the next run as the previous one starts are exempt, and
* the server says which those are.
*
* Reads the runs the schedules page has already loaded, and asks for a few of
* them so that one slow run is not read as a change of cadence.
*/
const MIN_RUNS = 3
export function scheduleOutlastsItsInterval(schedule: {
queues_next_run_at_start?: boolean
enabled?: boolean
interval_s?: number
jobs?: Array<{ duration_ms: number }>
}): boolean {
const { queues_next_run_at_start, enabled, interval_s, jobs } = schedule
if (queues_next_run_at_start || !enabled || !interval_s || (jobs?.length ?? 0) < MIN_RUNS)
return false
const durations = jobs!.map((j) => j.duration_ms).sort((a, b) => a - b)
return durations[durations.length >> 1] > interval_s * 1000
}
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+4 -58
View File
@@ -27,18 +27,10 @@
SettingService,
type Retry,
type Schedule,
type ScheduleIntervalDrift,
type ErrorHandler
} from '$lib/gen'
import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores'
import {
canWrite,
emptyString,
formatCron,
msToReadableTime,
sendUserToast,
cronV1toV2
} from '$lib/utils'
import { canWrite, emptyString, formatCron, 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'
@@ -131,25 +123,6 @@
let labels: string[] | undefined = $state(undefined)
let description = $state('')
let no_flow_overlap = $state(false)
// Measured rather than configured, so it comes from its own read of the deployed
// schedule: everything else on this form may be a draft that has never run.
let deployedDrift: ScheduleIntervalDrift | undefined = $state(undefined)
async function readDeployedDrift(path: string) {
let drift: ScheduleIntervalDrift | undefined
try {
drift =
(await ScheduleService.getScheduleIntervalDrift({ workspace: wsId ?? '', path })) ??
undefined
} catch {
drift = undefined
}
// The drawer is reused, so a slow answer can arrive once it has moved on to
// another schedule. It describes the one it asked about, not the one on screen.
if (path === initialPath) {
deployedDrift = drift
}
}
let tag: string | undefined = $state(undefined)
let validCRON = $state(true)
let isValid = $state(true)
@@ -361,9 +334,6 @@
drawer?.openDrawer()
runnable = undefined
edit = false
// A new schedule has no run history, even when its fields were read
// from an existing one.
deployedDrift = 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
@@ -528,7 +498,6 @@
async function loadSchedule(
defaultCfg?: Record<string, any>
): Promise<{ overlay: Record<string, any> | undefined; noDeployed: boolean }> {
deployedDrift = undefined
if (defaultCfg) {
await loadScheduleCfg(defaultCfg)
return { overlay: undefined, noDeployed: false }
@@ -540,7 +509,6 @@
getDraft: true
})
const { draft: draftFromBackend, ...deployedSchedule } = s as any
readDeployedDrift(initialPath)
await loadScheduleCfg(deployedSchedule)
return {
overlay: draftFromBackend
@@ -751,9 +719,6 @@
enabled = previousEnabled
return
}
// The measurement describes the deployed schedule, which just changed:
// disabling drops it, and enabling brings back what the runs still show.
await readDeployedDrift(initialPath)
sendUserToast(`${nEnabled ? 'enabled' : 'disabled'} schedule ${initialPath}`)
onUpdate?.(initialPath)
}
@@ -954,23 +919,6 @@
bind:validCRON
bind:cronVersion
/>
{#if deployedDrift}
<Alert type="warning" size="xs" title="Running less often than configured">
This schedule is running about every {msToReadableTime(
deployedDrift.effective_s * 1000
)} instead of every {msToReadableTime(deployedDrift.configured_s * 1000)}: each of the
last runs was queued too late for the slot that would have kept the cadence.
{#if deployedDrift.queues_next_run_at_start}
Its next run is already queued when the previous one starts, so the runs are
starting late rather than overrunning: look at worker capacity or a concurrency
limit.
{:else}
Script runs never overlap, so a run that outlasts its interval pushes the next one
out. To keep the cadence, schedule a flow instead: a flow queues its next run when
the previous one starts.
{/if}
</Alert>
{/if}
<div class="flex flex-col gap-1">
<Toggle
options={{
@@ -1034,13 +982,11 @@
class="mt-2"
/>
{/if}
<!-- Hidden for a skip handler, which makes the tick a single step flow: those
runs do overlap, so the toggle would be asserting the opposite. -->
{#if itemKind == 'script' && dynamicSkipPath == undefined}
{#if itemKind == 'script'}
<div class="flex gap-2 items-center mt-2">
<Toggle options={{ right: 'no overlap' }} checked={true} disabled /><Tooltip
>Script runs never overlap: the next run is queued once the previous one has
completed.</Tooltip
>Currently, overlapping scripts' executions is not supported. The next execution
will be scheduled only after the previous iteration has completed.</Tooltip
>
</div>
{/if}
frontend/src/routes/(root)/(logged)/schedules/+page.svelte
+32 -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,
storeLocalSetting
} from '$lib/utils'
import { scheduleOutlastsItsInterval } from '$lib/components/schedules/scheduleDrift'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -149,6 +156,9 @@
for (let schedule of schedules) {
if (schedulesWithJobsByPath[schedule.path]) {
schedule.jobs = schedulesWithJobsByPath[schedule.path].jobs
schedule.interval_s = schedulesWithJobsByPath[schedule.path].interval_s
schedule.queues_next_run_at_start =
schedulesWithJobsByPath[schedule.path].queues_next_run_at_start
}
}
loadingSchedulesWithJobStats = false
@@ -400,13 +410,19 @@
{/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_s, queues_next_run_at_start, 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}`}
{@const avg_s = jobs
? jobs.reduce((acc, x) => acc + x.duration_ms, 0) / jobs.length
: undefined}
{@const outlastsInterval = scheduleOutlastsItsInterval({
queues_next_run_at_start,
enabled,
interval_s,
jobs
})}
<div
class="bg-surface-tertiary hover:bg-surface-hover w-full items-center px-4 py-2 gap-4 first-of-type:!border-t-0
@@ -466,6 +482,20 @@
<div class="gap-2 items-center hidden md:flex">
<Badge large color="blue">{schedule}</Badge>
<Badge small color="gray">{timezone}</Badge>
{#if outlastsInterval}
<Popover notClickable>
<Badge small color="yellow">runs outlast the interval</Badge>
{#snippet text()}
<div>
Recent runs take longer than the {msToReadableTime(interval_s! * 1000)} between
slots. Script runs never overlap, so the next run is only queued once the
previous one has completed, and this schedule is running less often than its
cron asks for. To keep the cadence, schedule a flow instead: a flow queues its
next run when the previous one starts.
</div>
{/snippet}
</Popover>
{/if}
</div>
<div class="hidden lg:flex flex-row gap-1 items-center">