From aa65093e94fc98f5923a09ea681eb8ee6453bdfe Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 26 Dec 2023 16:31:04 +0100 Subject: [PATCH] feat: no overlap for flows togglable on schedules --- ...e4da5447020e39398deedbcca9121492834a.json} | 4 +- ...bd82837093636da2950915ce4364106f5679c.json | 24 ++++++ ...1e218539f0b259eb5f2c98e4e8ada6665f5ee.json | 23 ++++++ backend/windmill-api/src/schedule.rs | 2 +- backend/windmill-worker/src/worker_flow.rs | 73 ++++++++++--------- .../lib/components/ScheduleEditorInner.svelte | 12 ++- 6 files changed, 98 insertions(+), 40 deletions(-) rename backend/.sqlx/{query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json => query-4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a.json} (92%) create mode 100644 backend/.sqlx/query-aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c.json create mode 100644 backend/.sqlx/query-f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee.json diff --git a/backend/.sqlx/query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json b/backend/.sqlx/query-4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a.json similarity index 92% rename from backend/.sqlx/query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json rename to backend/.sqlx/query-4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a.json index 1cc71540c8..f89c4d0491 100644 --- a/backend/.sqlx/query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json +++ b/backend/.sqlx/query-4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT schedule.*, t.jobs FROM schedule, LATERAL ( SELECT ARRAY (SELECT json_build_object('id', id, 'success', success, 'duration_ms', duration_ms) FROM completed_job WHERE\n completed_job.schedule_path = schedule.path AND completed_job.workspace_id = $1 AND parent_job IS NULL ORDER BY started_at DESC LIMIT 20) AS jobs ) t\n WHERE schedule.workspace_id = $1 ORDER BY schedule.edited_at desc LIMIT $2 OFFSET $3", + "query": "SELECT schedule.*, t.jobs FROM schedule, LATERAL ( SELECT ARRAY (SELECT json_build_object('id', id, 'success', success, 'duration_ms', duration_ms) FROM completed_job WHERE\n completed_job.schedule_path = schedule.path AND completed_job.workspace_id = $1 AND parent_job IS NULL AND is_skipped = False ORDER BY started_at DESC LIMIT 20) AS jobs ) t\n WHERE schedule.workspace_id = $1 ORDER BY schedule.edited_at desc LIMIT $2 OFFSET $3", "describe": { "columns": [ { @@ -164,5 +164,5 @@ null ] }, - "hash": "7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097" + "hash": "4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a" } diff --git a/backend/.sqlx/query-aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c.json b/backend/.sqlx/query-aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c.json new file mode 100644 index 0000000000..82438b8f91 --- /dev/null +++ b/backend/.sqlx/query-aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(*) FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c" +} diff --git a/backend/.sqlx/query-f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee.json b/backend/.sqlx/query-f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee.json new file mode 100644 index 0000000000..4490ba97a1 --- /dev/null +++ b/backend/.sqlx/query-f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT no_flow_overlap FROM schedule WHERE path = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "no_flow_overlap", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee" +} diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index 7d909b098d..57428e0d03 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -356,7 +356,7 @@ async fn list_schedule_with_jobs( let (per_page, offset) = paginate(pagination); let rows = sqlx::query_as!(ScheduleWJobs, "SELECT schedule.*, t.jobs FROM schedule, LATERAL ( SELECT ARRAY (SELECT json_build_object('id', id, 'success', success, 'duration_ms', duration_ms) FROM completed_job WHERE - completed_job.schedule_path = schedule.path AND completed_job.workspace_id = $1 AND parent_job IS NULL ORDER BY started_at DESC LIMIT 20) AS jobs ) t + completed_job.schedule_path = schedule.path AND completed_job.workspace_id = $1 AND parent_job IS NULL AND is_skipped = False ORDER BY started_at DESC LIMIT 20) AS jobs ) t WHERE schedule.workspace_id = $1 ORDER BY schedule.edited_at desc LIMIT $2 OFFSET $3", w_id, per_page as i64, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 92e1a8dede..9050958315 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1149,41 +1149,44 @@ async fn push_next_flow_job let arc_flow_job_args = Arc::new(flow_job_args.clone()); if i == 0 { - // if !flow_job.is_flow_step && flow_job.schedule_path.is_some() { - // let not_allow_overlap = sqlx::query_scalar!( - // "SELECT not_allow_overlap FROM schedule WHERE path = $1 AND workspace_id = $2", - // flow_job.schedule_path.as_ref().unwrap(), - // flow_job.workspace_id.as_str() - // ) - // .fetch_one(db) - // .await?; - // if not_allow_overlap.fetch_one(db).await? { - // let count = sqlx::query_scalar!( - // "SELECT count(*) FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3 AND (flow_status->>'step')::int = 0", - // flow_job.schedule_path.as_ref().unwrap(), - // flow_job.workspace_id.as_str(), - // flow_job.id - // ).fetch_one(db).await?; - // if count > 1 { - // return update_flow_status_after_job_completion( - // db, - // client, - // flow_job.id, - // &Uuid::nil(), - // flow_job.workspace_id.as_str(), - // true, - // serde_json::from_str("\"not allowed to overlap\"").unwrap(), - // true, - // same_worker_tx, - // worker_dir, - // Some(true), - // rsmq, - // worker_name, - // ) - // .await; - // } - // } - // } + if !flow_job.is_flow_step && flow_job.schedule_path.is_some() { + let no_flow_overlap = sqlx::query_scalar!( + "SELECT no_flow_overlap FROM schedule WHERE path = $1 AND workspace_id = $2", + flow_job.schedule_path.as_ref().unwrap(), + flow_job.workspace_id.as_str() + ) + .fetch_one(db) + .await?; + if no_flow_overlap { + let count = sqlx::query_scalar!( + "SELECT count(*) FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3", + flow_job.schedule_path.as_ref().unwrap(), + flow_job.workspace_id.as_str(), + flow_job.id + ).fetch_one(db).await?.unwrap_or(0); + if count > 0 { + return update_flow_status_after_job_completion( + db, + client, + flow_job.id, + &Uuid::nil(), + flow_job.workspace_id.as_str(), + true, + serde_json::from_str( + "\"not allowed to overlap, scheduling next iteration\"", + ) + .unwrap(), + true, + same_worker_tx, + worker_dir, + Some(true), + rsmq, + worker_name, + ) + .await; + } + } + } if let Some(skip_expr) = &flow.skip_expr { let skip = compute_bool_from_expr( skip_expr.to_string(), diff --git a/frontend/src/lib/components/ScheduleEditorInner.svelte b/frontend/src/lib/components/ScheduleEditorInner.svelte index bb35c51de3..c3ae3f3c97 100644 --- a/frontend/src/lib/components/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/ScheduleEditorInner.svelte @@ -84,6 +84,7 @@ itemKind = is_flow ? 'flow' : 'script' initialScriptPath = initial_script_path ?? '' summary = '' + no_flow_overlap = false path = initialScriptPath initialPath = initialScriptPath script_path = initialScriptPath @@ -142,6 +143,7 @@ let enabled: boolean = false let pathError = '' let summary = '' + let no_flow_overlap = false let validCRON = true $: allowSchedule = isValid && validCRON && script_path != '' @@ -237,6 +239,7 @@ summary = s.summary ?? '' script_path = s.script_path ?? '' is_flow = s.is_flow + no_flow_overlap = s.no_flow_overlap ?? false wsErrorHandlerMuted = s.ws_error_handler_muted ?? false retry = s.retry if (s.on_failure) { @@ -299,7 +302,8 @@ on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {}, ws_error_handler_muted: wsErrorHandlerMuted, retry: retry, - summary: summary != '' ? summary : undefined + summary: summary != '' ? summary : undefined, + no_flow_overlap: no_flow_overlap } }) sendUserToast(`Schedule ${path} updated`) @@ -325,7 +329,8 @@ on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {}, ws_error_handler_muted: wsErrorHandlerMuted, retry: retry, - summary: summary != '' ? summary : undefined + summary: summary != '' ? summary : undefined, + no_flow_overlap: no_flow_overlap } }) sendUserToast(`Schedule ${path} created`) @@ -502,6 +507,9 @@ {itemKind} /> {/if} + {#if itemKind == 'flow'} + + {/if}
{#if runnable} {#if runnable?.schema && runnable.schema.properties && Object.keys(runnable.schema.properties).length > 0}