kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: improve cancel for flows with many substeps

Browse Source
This commit is contained in:
Ruben Fiszel
2025-03-24 22:00:15 +01:00
parent 3edca4bc91
commit ec11d577c6
5 changed files with 107 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
backend/.sqlx/query-21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896.json generated Normal file
View File

@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_queue SET canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = ANY($3) AND workspace_id = $4 AND (canceled_by IS NULL OR canceled_reason != $2) RETURNING id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"UuidArray",
"Text"
]
},
"nullable": [
false
]
},
"hash": "21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896"
}

29
backend/.sqlx/query-5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345.json generated Normal file
View File

@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "\nWITH RECURSIVE job_tree AS (\n -- Base case: direct children of the given parent job\n SELECT id, parent_job, 1 AS depth\n FROM v2_job_queue \n INNER JOIN v2_job USING (id)\n WHERE parent_job = $1 AND v2_job.workspace_id = $2\n\n UNION ALL\n\n -- Recursive case: fetch children of previously found jobs\n SELECT q.id, j.parent_job, t.depth + 1\n FROM v2_job_queue q\n INNER JOIN v2_job j USING (id)\n INNER JOIN job_tree t ON t.id = j.parent_job\n WHERE j.workspace_id = $2 AND t.depth < 500 -- Limit recursion depth to 500\n)\nSELECT id AS id, depth\nFROM job_tree\nORDER BY depth, id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "depth",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null,
null
]
},
"hash": "5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345"
}

23
backend/.sqlx/query-eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e.json generated
View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id AS \"id!\" FROM v2_job_queue INNER JOIN v2_job USING (id) WHERE parent_job = $1 AND v2_job.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false
]
},
"hash": "eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e"
}

68
backend/windmill-queue/src/jobs.rs
View File

@@ -245,22 +245,40 @@ pub async fn cancel_job<'c>(
let job = Arc::new(job);
// get all children
let mut jobs = vec![job.id];
let mut jobs_to_cancel = vec![];
while !jobs.is_empty() {
let p_job = jobs.pop();
let new_jobs = sqlx::query_scalar!(
"SELECT id AS \"id!\" FROM v2_job_queue INNER JOIN v2_job USING (id) WHERE parent_job = $1 AND v2_job.workspace_id = $2",
p_job,
w_id
)
.fetch_all(&mut *tx)
.await?;
jobs.extend(new_jobs.clone());
jobs_to_cancel.extend(new_jobs);
}
jobs.reverse();
// get all children using recursive CTE
let mut jobs_to_cancel = sqlx::query!(
r#"
WITH RECURSIVE job_tree AS (
-- Base case: direct children of the given parent job
SELECT id, parent_job, 1 AS depth
FROM v2_job_queue
INNER JOIN v2_job USING (id)
WHERE parent_job = $1 AND v2_job.workspace_id = $2
UNION ALL
-- Recursive case: fetch children of previously found jobs
SELECT q.id, j.parent_job, t.depth + 1
FROM v2_job_queue q
INNER JOIN v2_job j USING (id)
INNER JOIN job_tree t ON t.id = j.parent_job
WHERE j.workspace_id = $2 AND t.depth < 500 -- Limit recursion depth to 500
)
SELECT id AS id, depth
FROM job_tree
ORDER BY depth, id
"#,
job.id,
w_id
)
.fetch_all(&mut *tx)
.await?
.into_iter()
.filter_map(|r| r.id.clone())
.collect_vec();
jobs_to_cancel.reverse();
tracing::info!("Found {} child jobs to cancel", jobs_to_cancel.len());
let (ntx, _) = cancel_single_job(
username,
@@ -274,7 +292,23 @@ pub async fn cancel_job<'c>(
.await?;
tx = ntx;
// cancel children
if !force_cancel {
// cancel children in batch first
if !jobs_to_cancel.is_empty() {
let updated = sqlx::query_scalar!(
"UPDATE v2_job_queue SET canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = ANY($3) AND workspace_id = $4 AND (canceled_by IS NULL OR canceled_reason != $2) RETURNING id",
username,
reason,
jobs_to_cancel.as_slice(),
w_id
)
.fetch_all(&mut *tx)
.await?;
// Remove any jobs that were successfully updated
jobs_to_cancel.retain(|id| !updated.contains(&id));
}
}
for job_id in jobs_to_cancel {
let job = get_queued_job_tx(job_id, &w_id, &mut tx).await?;

4
frontend/src/lib/components/FlowBuilder.svelte
View File

@@ -248,7 +248,7 @@
workspace: $workspaceStore!,
requestBody: {
path: $pathStore,
summary: flow.summary,
summary: flow.summary ?? '',
description: flow.description ?? '',
value: flow.value,
schema: flow.schema,
@@ -379,7 +379,7 @@
workspace: $workspaceStore!,
requestBody: {
path: $pathStore,
summary: flow.summary,
summary: flow.summary ?? '',
description: flow.description ?? '',
value: flow.value,
schema: flow.schema,