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

feat: no overlap for flows togglable on schedules

Browse Source
This commit is contained in:
Ruben Fiszel
2023-12-26 16:31:04 +01:00
parent b290cbb758
commit aa65093e94
6 changed files with 98 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json → backend/.sqlx/query-4422b7183ede17a9cbde4afae41be4da5447020e39398deedbcca9121492834a.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-aeedd3d92f7c50fe49d14053afebd82837093636da2950915ce4364106f5679c.json
Generated
+24
View File
@@ -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"
}
backend/.sqlx/query-f2173822c1b3d82afedf0f7dba31e218539f0b259eb5f2c98e4e8ada6665f5ee.json
Generated
+23
View File
@@ -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"
}
backend/windmill-api/src/schedule.rs
+1 -1
View File
@@ -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,
backend/windmill-worker/src/worker_flow.rs
+38 -35
View File
@@ -1149,41 +1149,44 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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(),
frontend/src/lib/components/ScheduleEditorInner.svelte
+10 -2
View File
@@ -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'}
<Toggle options={{ right: 'no overlap of flows' }} bind:checked={no_flow_overlap} />
{/if}
<div class="mt-6">
{#if runnable}
{#if runnable?.schema && runnable.schema.properties && Object.keys(runnable.schema.properties).length > 0}