From 1afb7a2156570eadb780be00a40eb9395cfb4deb Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 20 Sep 2025 11:00:45 +0000 Subject: [PATCH] improve job cancelling with new flow jobs locks --- ...a83a2e1f07007b500b556b841181b3adebfb8.json | 18 --------- ...e03a6d51c24fc9a2d8ac872aaf41a20af5045.json | 27 ++++++++++++++ ...a39101520409c5f241bc8d1b4190bb13260a9.json | 25 ------------- backend/windmill-worker/src/worker_flow.rs | 37 ++++++------------- .../lib/components/FlowPreviewContent.svelte | 2 +- .../lib/components/FlowStatusViewer.svelte | 8 ++-- .../components/FlowStatusViewerInner.svelte | 2 +- .../lib/components/flows/flowModuleNextId.ts | 1 - frontend/src/lib/components/graph/model.ts | 2 +- 9 files changed, 45 insertions(+), 77 deletions(-) delete mode 100644 backend/.sqlx/query-054fc3f9d6776b9068816b7568ca83a2e1f07007b500b556b841181b3adebfb8.json create mode 100644 backend/.sqlx/query-ac5dd4a4d7991159e053f1c11bce03a6d51c24fc9a2d8ac872aaf41a20af5045.json delete mode 100644 backend/.sqlx/query-d70ddf86eb77a1fdac1e03d2c1aa39101520409c5f241bc8d1b4190bb13260a9.json diff --git a/backend/.sqlx/query-054fc3f9d6776b9068816b7568ca83a2e1f07007b500b556b841181b3adebfb8.json b/backend/.sqlx/query-054fc3f9d6776b9068816b7568ca83a2e1f07007b500b556b841181b3adebfb8.json deleted file mode 100644 index 461f79a1f2..0000000000 --- a/backend/.sqlx/query-054fc3f9d6776b9068816b7568ca83a2e1f07007b500b556b841181b3adebfb8.json +++ /dev/null @@ -1,18 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET flow_status = JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5)\n WHERE id = $2 AND flow_status->'modules'->$1::int->'flow_jobs_duration' IS NOT NULL", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "054fc3f9d6776b9068816b7568ca83a2e1f07007b500b556b841181b3adebfb8" -} diff --git a/backend/.sqlx/query-ac5dd4a4d7991159e053f1c11bce03a6d51c24fc9a2d8ac872aaf41a20af5045.json b/backend/.sqlx/query-ac5dd4a4d7991159e053f1c11bce03a6d51c24fc9a2d8ac872aaf41a20af5045.json new file mode 100644 index 0000000000..8de90d69d7 --- /dev/null +++ b/backend/.sqlx/query-ac5dd4a4d7991159e053f1c11bce03a6d51c24fc9a2d8ac872aaf41a20af5045.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE v2_job_status SET flow_status = \n JSONB_SET(JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "ac5dd4a4d7991159e053f1c11bce03a6d51c24fc9a2d8ac872aaf41a20af5045" +} diff --git a/backend/.sqlx/query-d70ddf86eb77a1fdac1e03d2c1aa39101520409c5f241bc8d1b4190bb13260a9.json b/backend/.sqlx/query-d70ddf86eb77a1fdac1e03d2c1aa39101520409c5f241bc8d1b4190bb13260a9.json deleted file mode 100644 index 91074b72a6..0000000000 --- a/backend/.sqlx/query-d70ddf86eb77a1fdac1e03d2c1aa39101520409c5f241bc8d1b4190bb13260a9.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE v2_job_status SET flow_status = \n JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [ - null - ] - }, - "hash": "d70ddf86eb77a1fdac1e03d2c1aa39101520409c5f241bc8d1b4190bb13260a9" -} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 0c2da9f761..d55ce51be9 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -28,6 +28,7 @@ use serde_json::value::RawValue; use serde_json::{json, Value}; use sqlx::types::Json; use sqlx::{FromRow, Postgres, Transaction}; +use tokio::spawn; use tracing::instrument; use uuid::Uuid; use windmill_common::auth::get_job_perms; @@ -531,16 +532,20 @@ pub async fn update_flow_status_after_job_completion_internal( sqlx::query_scalar!( " UPDATE v2_job_status SET flow_status = - JSONB_SET(JSONB_SET( + JSONB_SET(JSONB_SET(JSONB_SET(JSONB_SET( flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), - ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb) + ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6) WHERE id = $2 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", old_status.step, flow, position as i32, json!(success), + flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default(), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(), ) } else { sqlx::query_scalar!( @@ -563,26 +568,6 @@ pub async fn update_flow_status_after_job_completion_internal( )) })?.ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress for flow {flow} at step {}", old_status.step)))?; - if let Some(position) = position { - // if we do this in a single statement, somehow cancelling the flow job will have issues because the RETURNING clause will return None - // Could not figure out why, so we do it in two steps - if let Err(e) = sqlx::query!( - "UPDATE v2_job_status SET flow_status = JSONB_SET(JSONB_SET( - flow_status, - ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $4), - ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5) - WHERE id = $2 AND flow_status->'modules'->$1::int->'flow_jobs_duration' IS NOT NULL", - old_status.step as i32, - flow, - position as i32, - flow_job_duration.as_ref().map(|x| json!(x.started_at)), - flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) - ) - .execute(&mut *tx) - .await { - tracing::error!("error while updating flow jobs duration: {e:#}"); - } - } // let status_for_debug = sqlx::query!( // "SELECT flow_status FROM v2_job_status WHERE id = $1", // flow @@ -636,8 +621,8 @@ pub async fn update_flow_status_after_job_completion_internal( flow, position as i32, json!(success), - flow_job_duration.as_ref().map(|x| json!(x.started_at)), - flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) + flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default(), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(), ) } else { sqlx::query_scalar!( @@ -1617,8 +1602,8 @@ async fn set_success_and_duration_in_flow_job_success<'c>( flow, position as i32, json!(success), - flow_job_duration.as_ref().map(|x| json!(x.duration_ms)), - flow_job_duration.as_ref().map(|x| json!(x.started_at)) + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)).unwrap_or_default(), + flow_job_duration.as_ref().map(|x| json!(x.started_at)).unwrap_or_default() ) .execute(&mut **tx) .await diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index 558a854bca..f2340066e9 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -583,7 +583,7 @@ bind:suspendStatus hideDownloadInGraph={customUi?.downloadLogs === false} wideResults - bind:flowStateStore={flowStateStore.val} + bind:flowState={flowStateStore.val} {jobId} onDone={() => { isRunning = false diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index 6ca2c8fd0f..7d5c1c4a12 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -11,7 +11,7 @@ jobId: string initialJob?: Job | undefined workspaceId?: string | undefined - flowStateStore?: FlowState + flowState?: FlowState selectedJobStep?: string | undefined hideFlowResult?: boolean hideTimeline?: boolean @@ -39,7 +39,7 @@ jobId, initialJob = undefined, workspaceId = undefined, - flowStateStore = $bindable({}), + flowState = $bindable({}), selectedJobStep = $bindable(undefined), hideFlowResult = false, hideTimeline = false, @@ -67,7 +67,7 @@ let globalRefreshes: Record Promise)[]> = $state({}) setContext('FlowStatusViewer', { - flowStateStore, + flowState: flowState, suspendStatus, retryStatus, hideDownloadInGraph, @@ -88,7 +88,7 @@ retryStatus.val = {} suspendStatus.val = {} globalRefreshes = {} - flowStateStore.val = {} + flowState.val = {} localDurationStatuses = {} localModuleStates = {} } diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index caf8c7083b..dc101fa520 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -48,7 +48,7 @@ import JobAssetsViewer from './assets/JobAssetsViewer.svelte' let { - flowStateStore, + flowState: flowStateStore, retryStatus, suspendStatus, hideDownloadInGraph, diff --git a/frontend/src/lib/components/flows/flowModuleNextId.ts b/frontend/src/lib/components/flows/flowModuleNextId.ts index 1d0cd14ec6..34bed860f8 100644 --- a/frontend/src/lib/components/flows/flowModuleNextId.ts +++ b/frontend/src/lib/components/flows/flowModuleNextId.ts @@ -6,7 +6,6 @@ import { charsToNumber, numberToChars } from './idUtils' // Computes the next available id export function nextId(flowState: FlowState, fullFlow: OpenFlow): string { const allIds = dfs(fullFlow.value.modules, (fm) => fm.id) - console.log('allIds', allIds) const max = allIds.concat(Object.keys(flowState)).reduce((acc, key) => { if (key.length >= 4) { diff --git a/frontend/src/lib/components/graph/model.ts b/frontend/src/lib/components/graph/model.ts index 8ed90faeeb..990d5cac53 100644 --- a/frontend/src/lib/components/graph/model.ts +++ b/frontend/src/lib/components/graph/model.ts @@ -28,7 +28,7 @@ export type DurationStatus = { } export type FlowStatusViewerContext = { - flowStateStore?: FlowState + flowState?: FlowState retryStatus: StateStore> suspendStatus: StateStore> hideDownloadInGraph?: boolean