diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 12d2b50559..9f041f1655 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -593,6 +593,44 @@ }, "query": "UPDATE queue SET logs = $1 WHERE id = $2" }, + "1c28baaadd7d0c86a92bf9880a4ea33457bf8cff669e983431f1fd26ff275f83": { + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Uuid" + }, + { + "name": "flow_status", + "ordinal": 1, + "type_info": "Jsonb" + }, + { + "name": "suspend", + "ordinal": 2, + "type_info": "Int4" + }, + { + "name": "script_path", + "ordinal": 3, + "type_info": "Varchar" + } + ], + "nullable": [ + false, + true, + false, + true + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "\n SELECT id, flow_status, suspend, script_path\n FROM queue\n WHERE id = $1\n " + }, "1e35c39bc786d638252e5483ca4efae9a041f7e845341f8bfd715ddd9e899499": { "describe": { "columns": [ diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index fc1b538485..2fd1cbe946 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3176,7 +3176,7 @@ paths: /w/{workspace}/jobs/flow/resume/{id}: post: summary: resume a job for a suspended flow as an owner - operationId: resumeSuspendedJobAsOwner + operationId: resumeSuspendedFlowAsOwner tags: - job parameters: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index c655725984..1db5f17029 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -406,6 +406,7 @@ async fn list_jobs( "false as is_skipped", "email", "visible_to_owner", + "suspend", ], ); let sqlc = list_completed_jobs_query( @@ -440,6 +441,7 @@ async fn list_jobs( "is_skipped", "email", "visible_to_owner", + "null as suspend", ], ); let sql = format!( @@ -458,13 +460,13 @@ async fn list_jobs( pub async fn resume_suspended_flow_as_owner( authed: Authed, Extension(db): Extension, - Path((w_id, job_id)): Path<(String, Uuid)>, + Path((w_id, flow_id)): Path<(String, Uuid)>, QueryOrBody(value): QueryOrBody, ) -> error::Result { let value = value.unwrap_or(serde_json::Value::Null); let mut tx = db.begin().await?; - let flow = get_suspended_flow_info(job_id, &mut tx).await?; + let (flow, job_id) = get_suspended_flow_info(flow_id, &mut tx).await?; if !authed.is_admin { require_owner_of_path( @@ -502,7 +504,7 @@ pub async fn resume_suspended_job( } mac.verify_slice(hex::decode(secret)?.as_ref()) .map_err(|_| anyhow::anyhow!("Invalid signature"))?; - let flow = get_suspended_flow_info(job_id, &mut tx).await?; + let flow = get_suspended_prent_flow_info(job_id, &mut tx).await?; let exists = sqlx::query_scalar!( r#" @@ -593,7 +595,7 @@ struct FlowInfo { script_path: Option, } -async fn get_suspended_flow_info<'c>( +async fn get_suspended_prent_flow_info<'c>( job_id: Uuid, tx: &mut Transaction<'c, Postgres>, ) -> error::Result { @@ -613,6 +615,38 @@ async fn get_suspended_flow_info<'c>( Ok(flow) } +async fn get_suspended_flow_info<'c>( + job_id: Uuid, + tx: &mut Transaction<'c, Postgres>, +) -> error::Result<(FlowInfo, Uuid)> { + let flow = sqlx::query_as!( + FlowInfo, + r#" + SELECT id, flow_status, suspend, script_path + FROM queue + WHERE id = $1 + "#, + job_id, + ) + .fetch_optional(tx) + .await? + .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; + let job_id = flow + .flow_status + .as_ref() + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + .and_then(|s| match s.modules.get(s.step as usize) { + Some(FlowStatusModule::WaitingForEvents { job, .. }) => Some(job.to_owned()), + _ => None, + }); + + if let Some(job_id) = job_id { + Ok((flow, job_id)) + } else { + Err(anyhow::anyhow!("the flow is not in a suspended state anymore").into()) + } +} + pub async fn cancel_suspended_job( /* unauthed */ Extension(db): Extension, @@ -885,6 +919,7 @@ struct UnifiedJob { is_skipped: bool, email: String, visible_to_owner: bool, + suspend: Option, } impl From for Job { @@ -950,6 +985,7 @@ impl From for Job { pre_run_error: None, email: uj.email, visible_to_owner: uj.visible_to_owner, + suspend: uj.suspend, }), t => panic!("job type {} not valid", t), } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7d9251a4fa..6327199943 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -598,6 +598,7 @@ pub struct QueuedJob { pub pre_run_error: Option, pub email: String, pub visible_to_owner: bool, + pub suspend: Option, } impl QueuedJob { diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index 01fffc4864..ad0dc8581b 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -97,7 +97,7 @@
diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index 5bd015469e..1f95a0904c 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -13,9 +13,8 @@ import Tabs from './common/tabs/Tabs.svelte' import { FlowGraph, type GraphModuleState } from './graph' import ModuleStatus from './ModuleStatus.svelte' - import { displayDate, isOwner, truncateRev } from '$lib/utils' + import { displayDate, isOwner, pluralize, truncateRev } from '$lib/utils' import JobArgs from './JobArgs.svelte' - import autosize from 'svelte-autosize' import Tooltip from './Tooltip.svelte' import SimpleEditor from './SimpleEditor.svelte' @@ -36,6 +35,7 @@ let localFlowModuleStates: Record = {} export let retry_status: Record = {} + export let suspend_status: number | undefined = undefined export let is_owner = false @@ -69,6 +69,7 @@ } $: updateFailCount(job?.flow_status?.retry?.fail_count) + $: suspend_status = job?.flow_status?.modules?.[job?.flow_status.step]?.count function updateFailCount(count?: number) { if (count) { @@ -191,9 +192,9 @@ color="green" variant="border" on:click={async () => - await JobService.resumeSuspendedJobAsOwner({ + await JobService.resumeSuspendedFlowAsOwner({ workspace: $workspaceStore ?? '', - id: job?.flow_status?.modules?.[job?.flow_status?.step - 1]?.job ?? '', + id: job?.id ?? '', requestBody: JSON.parse(payload) })} >Resume
{#if [FlowStatusModule.type.IN_PROGRESS, FlowStatusModule.type.SUCCESS, FlowStatusModule.type.FAILURE].includes(mod.type)} {/each} + {#if suspend_status} + + Flow suspended, waiting for {pluralize(suspend_status, 'approval')} + + {/if}