diff --git a/backend/.sqlx/query-0600e0d5cf4aa29b287ce57cf379cc84f7281371b2cccbe2bb59437f645a559c.json b/backend/.sqlx/query-0600e0d5cf4aa29b287ce57cf379cc84f7281371b2cccbe2bb59437f645a559c.json deleted file mode 100644 index 5e9dd78b02..0000000000 --- a/backend/.sqlx/query-0600e0d5cf4aa29b287ce57cf379cc84f7281371b2cccbe2bb59437f645a559c.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-0f141ca6a58901dee1ddcf25694d76a4e5702c0324a27ef3a5740ef5d98d8946.json b/backend/.sqlx/query-0f141ca6a58901dee1ddcf25694d76a4e5702c0324a27ef3a5740ef5d98d8946.json deleted file mode 100644 index 0dbcf04c6e..0000000000 --- a/backend/.sqlx/query-0f141ca6a58901dee1ddcf25694d76a4e5702c0324a27ef3a5740ef5d98d8946.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7.json b/backend/.sqlx/query-847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7.json deleted file mode 100644 index da439f8af3..0000000000 --- a/backend/.sqlx/query-847974abf5011c82a32e0a66fc4c0b9a773cd864688808435df8077e54605dc7.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e1e0e444e6620e3ddb3d22ffbd24a3b637230a4125314b89688d7dc07acb2f4c.json b/backend/.sqlx/query-e1e0e444e6620e3ddb3d22ffbd24a3b637230a4125314b89688d7dc07acb2f4c.json new file mode 100644 index 0000000000..bca8231fec --- /dev/null +++ b/backend/.sqlx/query-e1e0e444e6620e3ddb3d22ffbd24a3b637230a4125314b89688d7dc07acb2f4c.json @@ -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" +} diff --git a/backend/windmill-api-schedule/src/lib.rs b/backend/windmill-api-schedule/src/lib.rs index fb77a823ec..f6a087e3ae 100644 --- a/backend/windmill-api-schedule/src/lib.rs +++ b/backend/windmill-api-schedule/src/lib.rs @@ -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>, -} - -/// 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, + /// 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], +/// 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 { - if push_times.len() <= DRIFT_MIN_MISSED_SLOTS { - return None; - } +) -> Option { 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)) - }; - - // 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 = 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>> { - 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> { 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>, + schedule: String, + timezone: String, + cron_version: Option, + is_flow: bool, + dynamic_skip: 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. "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, - Extension(db): Extension, - Path((w_id, path)): Path<(String, StripPath)>, -) -> JsonResult> { - 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> { - // 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, Path((w_id, path)): Path<(String, StripPath)>, @@ -1911,80 +1747,3 @@ 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 }) - ); - } - - /// 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 - ); - } -} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 0e6aa75951..573e68115a 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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 diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 1f267cb748..62d47a34d7 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -980,27 +980,14 @@ 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. 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, - ) -> Option> { 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"), } } diff --git a/frontend/src/lib/components/schedules/scheduleDrift.ts b/frontend/src/lib/components/schedules/scheduleDrift.ts new file mode 100644 index 0000000000..d719deaeb1 --- /dev/null +++ b/frontend/src/lib/components/schedules/scheduleDrift.ts @@ -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 +} diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte index 94f2bc7d26..7aa83cbbee 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte @@ -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 ): Promise<{ overlay: Record | 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} - - 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} - - {/if}
{/if} - - {#if itemKind == 'script' && dynamicSkipPath == undefined} + {#if itemKind == 'script'}
Script runs never overlap: the next run is queued once the previous one has - completed.Currently, overlapping scripts' executions is not supported. The next execution + will be scheduled only after the previous iteration has completed.
{/if} diff --git a/frontend/src/routes/(root)/(logged)/schedules/+page.svelte b/frontend/src/routes/(root)/(logged)/schedules/+page.svelte index e403c981de..1b26f487a8 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, + 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}
- {#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 + })}