From ec11d577c6089df0b6019cd05064f5ea63fb317c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 24 Mar 2025 22:00:15 +0100 Subject: [PATCH] fix: improve cancel for flows with many substeps --- ...bb36bdece81385380a42ca06ca6be19694896.json | 25 +++++++ ...f2e6606bfc0f440fbdbbdf4dd3234068c9345.json | 29 ++++++++ ...c538fbb54ad269ec81aea89e431a511245a1e.json | 23 ------- backend/windmill-queue/src/jobs.rs | 68 ++++++++++++++----- .../src/lib/components/FlowBuilder.svelte | 4 +- 5 files changed, 107 insertions(+), 42 deletions(-) create mode 100644 backend/.sqlx/query-21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896.json create mode 100644 backend/.sqlx/query-5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345.json delete mode 100644 backend/.sqlx/query-eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e.json diff --git a/backend/.sqlx/query-21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896.json b/backend/.sqlx/query-21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896.json new file mode 100644 index 0000000000..5454fdf323 --- /dev/null +++ b/backend/.sqlx/query-21204693fa8608c78151f63fa76bb36bdece81385380a42ca06ca6be19694896.json @@ -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" +} diff --git a/backend/.sqlx/query-5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345.json b/backend/.sqlx/query-5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345.json new file mode 100644 index 0000000000..a3751998fb --- /dev/null +++ b/backend/.sqlx/query-5e8ba1850b2520bd4bf030f53f1f2e6606bfc0f440fbdbbdf4dd3234068c9345.json @@ -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" +} diff --git a/backend/.sqlx/query-eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e.json b/backend/.sqlx/query-eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e.json deleted file mode 100644 index 60a98ba32a..0000000000 --- a/backend/.sqlx/query-eb68469026be39048c5f42a80a2c538fbb54ad269ec81aea89e431a511245a1e.json +++ /dev/null @@ -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" -} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 394d04e366..865b7d47bc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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?; diff --git a/frontend/src/lib/components/FlowBuilder.svelte b/frontend/src/lib/components/FlowBuilder.svelte index adb5331a6e..412a5ee88f 100644 --- a/frontend/src/lib/components/FlowBuilder.svelte +++ b/frontend/src/lib/components/FlowBuilder.svelte @@ -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,