diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 36ddb8ab9f..713ccb9dd3 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-66e66da2ed6eace5d7ec2a41a7b11ae255f5dc212d1ff41c2905b303c8c13b18.json b/backend/.sqlx/query-66e66da2ed6eace5d7ec2a41a7b11ae255f5dc212d1ff41c2905b303c8c13b18.json new file mode 100644 index 0000000000..6350d8a181 --- /dev/null +++ b/backend/.sqlx/query-66e66da2ed6eace5d7ec2a41a7b11ae255f5dc212d1ff41c2905b303c8c13b18.json @@ -0,0 +1,46 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH job_info AS (\n SELECT id, kind::text AS kind, parent_job\n FROM v2_job\n WHERE id = $1\n )\n SELECT\n q.id AS \"id!\",\n s.flow_status,\n q.suspend AS \"suspend!\",\n j.runnable_path AS script_path,\n (ji.kind IN ('flow', 'flowpreview')) AS \"is_flow_level!\"\n FROM job_info ji\n JOIN v2_job_queue q ON q.id = CASE\n WHEN ji.kind IN ('flow', 'flowpreview') THEN ji.id\n ELSE ji.parent_job\n END\n JOIN v2_job j ON j.id = q.id\n JOIN v2_job_status s ON s.id = q.id\n FOR UPDATE OF q\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "flow_status", + "type_info": "Jsonb" + }, + { + "ordinal": 2, + "name": "suspend!", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "is_flow_level!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + true, + false, + true, + null + ] + }, + "hash": "66e66da2ed6eace5d7ec2a41a7b11ae255f5dc212d1ff41c2905b303c8c13b18" +} diff --git a/backend/.sqlx/query-83785ee2f7dcc7f2252b0e8bcc8322dfd7689d615a34b63e29c9c6699b7e5514.json b/backend/.sqlx/query-83785ee2f7dcc7f2252b0e8bcc8322dfd7689d615a34b63e29c9c6699b7e5514.json deleted file mode 100644 index 1c1288e21c..0000000000 --- a/backend/.sqlx/query-83785ee2f7dcc7f2252b0e8bcc8322dfd7689d615a34b63e29c9c6699b7e5514.json +++ /dev/null @@ -1,40 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT q.id, f.flow_status, q.suspend, j.runnable_path AS script_path\n FROM v2_job_queue q\n JOIN v2_job j USING (id)\n JOIN v2_job_status f USING (id)\n WHERE id = ( SELECT parent_job FROM v2_job WHERE id = $1 )\n FOR UPDATE\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "flow_status", - "type_info": "Jsonb" - }, - { - "ordinal": 2, - "name": "suspend", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "script_path", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - false, - true, - false, - true - ] - }, - "hash": "83785ee2f7dcc7f2252b0e8bcc8322dfd7689d615a34b63e29c9c6699b7e5514" -} diff --git a/backend/.sqlx/query-a87e7b098b523176916daf5b1685b218dfbfe752aa730b421e9229dadd6ae587.json b/backend/.sqlx/query-a87e7b098b523176916daf5b1685b218dfbfe752aa730b421e9229dadd6ae587.json new file mode 100644 index 0000000000..9dbb0ac3cf --- /dev/null +++ b/backend/.sqlx/query-a87e7b098b523176916daf5b1685b218dfbfe752aa730b421e9229dadd6ae587.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT kind::text as \"kind!\", parent_job\n FROM v2_job\n WHERE id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "parent_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null, + true + ] + }, + "hash": "a87e7b098b523176916daf5b1685b218dfbfe752aa730b421e9229dadd6ae587" +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 6b1ad78b9c..544492d4e3 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -10030,6 +10030,11 @@ paths: in: query schema: type: string + - name: flow_level + in: query + description: If true, generate resume URLs for the parent flow instead of the specific step. This allows pre-approvals that can be consumed by any later suspend step in the same flow. + schema: + type: boolean responses: "200": description: url endpoints diff --git a/backend/windmill-api/src/approvals.rs b/backend/windmill-api/src/approvals.rs index 1296000481..e2344adc4e 100644 --- a/backend/windmill-api/src/approvals.rs +++ b/backend/windmill-api/src/approvals.rs @@ -129,7 +129,7 @@ pub async fn handle_resume_action( captures.name("approver").map(|m| m.as_str().to_string()), ); - let approver = QueryApprover { approver }; + let approver = QueryApprover { approver, flow_level: None }; // Convert job_id and resume_id to appropriate types let job_uuid = Uuid::from_str(job_id) @@ -191,7 +191,7 @@ pub async fn get_approval_form_details( let res = get_resume_urls_internal( axum::Extension(db.clone()), Path((w_id.to_string(), job_id, resume_id)), - Query(QueryApprover { approver: approver.map(|a| a.to_string()) }), + Query(QueryApprover { approver: approver.map(|a| a.to_string()), flow_level: None }), ) .await?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 81a6055d22..d5b71e9140 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2553,22 +2553,28 @@ async fn resume_suspended_job_internal( let value = value.unwrap_or(serde_json::Value::Null); verify_suspended_secret(&w_id, &db, job_id, resume_id, &approver, secret).await?; - let parent_flow_info = get_suspended_parent_flow_info(job_id, &db).await?; - let parent_flow = GetQuery::new() - .without_logs() - .without_code() - .without_flow() - .fetch(&db, &parent_flow_info.id, &w_id) - .await?; - let flow_status = parent_flow - .flow_status() - .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?; + // Get flow info - works for both step-level (job_id is a step) and flow-level (job_id is the flow) + let (flow_info, is_flow_level) = get_flow_info_for_resume(job_id, &db).await?; - let trigger_email = match &parent_flow { - Job::CompletedJob(job) => &job.email, - Job::QueuedJob(job) => &job.email, - }; - conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?; + // For step-level resumes, verify user auth and flow status + // For flow-level resumes (pre-approvals), the flow might not be at a suspended step yet + if !is_flow_level { + let parent_flow = GetQuery::new() + .without_logs() + .without_code() + .without_flow() + .fetch(&db, &flow_info.id, &w_id) + .await?; + let flow_status = parent_flow + .flow_status() + .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?; + + let trigger_email = match &parent_flow { + Job::CompletedJob(job) => &job.email, + Job::QueuedJob(job) => &job.email, + }; + conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?; + } let exists = sqlx::query_scalar!( r#" @@ -2584,7 +2590,7 @@ async fn resume_suspended_job_internal( return Err(anyhow::anyhow!("resume request already sent").into()); } - let approver = if authed.as_ref().is_none() + let approver_value = if authed.as_ref().is_none() || (approver .approver .clone() @@ -2599,9 +2605,9 @@ async fn resume_suspended_job_internal( insert_resume_job( resume_id, job_id, - &parent_flow_info, + &flow_info, value, - approver.clone(), + approver_value.clone(), approved, &mut tx, ) @@ -2610,15 +2616,20 @@ async fn resume_suspended_job_internal( if !approved { sqlx::query!( "UPDATE v2_job_queue SET suspend = 0 WHERE id = $1", - parent_flow_info.id + flow_info.id ) .execute(&mut *tx) .await?; + } else if is_flow_level { + // For flow-level resumes, decrement the suspend counter if the flow is currently suspended + // The approval will be matched when the worker checks for resumes (both step-level and flow-level) + resume_immediately_for_flow_level(&flow_info, &mut tx).await?; } else { - resume_immediately_if_relevant(parent_flow_info, job_id, &mut tx).await?; + // For step-level resumes, try to resume immediately if the step is waiting + resume_immediately_if_relevant(flow_info, job_id, &mut tx).await?; } - let approver = approver.unwrap_or_else(|| "anonymous".to_string()); + let approver = approver_value.unwrap_or_else(|| "anonymous".to_string()); let audit_author = match authed { Some(authed) => (&authed).into(), @@ -2709,6 +2720,26 @@ async fn resume_immediately_if_relevant<'c>( ) } +/// For flow-level resumes, decrement the suspend counter if the flow is currently suspended. +/// Unlike step-level resumes, we don't check if the job_id matches - we just need the flow +/// to be in a suspended state. +async fn resume_immediately_for_flow_level<'c>( + flow: &FlowInfo, + tx: &mut Transaction<'c, Postgres>, +) -> error::Result<()> { + if flow.suspend > 0 { + let new_suspend = flow.suspend - 1; + sqlx::query!( + "UPDATE v2_job_queue SET suspend = $1 WHERE id = $2", + new_suspend, + flow.id, + ) + .execute(&mut **tx) + .await?; + } + Ok(()) +} + async fn insert_resume_job<'c>( resume_id: u32, job_id: Uuid, @@ -2745,23 +2776,46 @@ struct FlowInfo { script_path: Option, } -async fn get_suspended_parent_flow_info(job_id: Uuid, db: &DB) -> error::Result { - let flow = sqlx::query_as!( - FlowInfo, +/// Get flow info from either a step job (by looking up its parent) or a flow job directly. +/// Returns (FlowInfo, is_flow_level) where is_flow_level indicates if job_id was a flow job. +async fn get_flow_info_for_resume(job_id: Uuid, db: &DB) -> error::Result<(FlowInfo, bool)> { + // Single query that determines if job_id is a flow or step, and fetches the appropriate flow info + let result = sqlx::query!( r#" - SELECT q.id, f.flow_status, q.suspend, j.runnable_path AS script_path - FROM v2_job_queue q - JOIN v2_job j USING (id) - JOIN v2_job_status f USING (id) - WHERE id = ( SELECT parent_job FROM v2_job WHERE id = $1 ) - FOR UPDATE - "#, + WITH job_info AS ( + SELECT id, kind::text AS kind, parent_job + FROM v2_job + WHERE id = $1 + ) + SELECT + q.id AS "id!", + s.flow_status, + q.suspend AS "suspend!", + j.runnable_path AS script_path, + (ji.kind IN ('flow', 'flowpreview')) AS "is_flow_level!" + FROM job_info ji + JOIN v2_job_queue q ON q.id = CASE + WHEN ji.kind IN ('flow', 'flowpreview') THEN ji.id + ELSE ji.parent_job + END + JOIN v2_job j ON j.id = q.id + JOIN v2_job_status s ON s.id = q.id + FOR UPDATE OF q + "#, job_id, ) .fetch_optional(db) .await? - .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; - Ok(flow) + .ok_or_else(|| anyhow::anyhow!("job not found or parent flow not in queue: {}", job_id))?; + + let flow_info = FlowInfo { + id: result.id, + flow_status: result.flow_status, + suspend: result.suspend, + script_path: result.script_path, + }; + + Ok((flow_info, result.is_flow_level)) } async fn get_suspended_flow_info<'c>( @@ -2828,6 +2882,9 @@ pub struct SuspendedJobFlow { #[derive(Deserialize, Debug)] pub struct QueryApprover { pub approver: Option, + /// If true, generate/verify resume URLs for the parent flow instead of the specific step. + /// This allows pre-approvals that can be consumed by any later suspend step in the same flow. + pub flow_level: Option, } pub async fn get_suspended_job_flow( @@ -3084,8 +3141,17 @@ pub async fn get_resume_urls_internal( Query(approver): Query, ) -> error::JsonResult { let key = get_workspace_key(&w_id, &db).await?; - let signature = create_signature(key, job_id, resume_id, approver.approver.clone())?; - let approver = approver + + // If flow_level is true, use the parent flow ID for the signature and URLs + // This allows pre-approvals that can be consumed by any later suspend step + let target_job_id = if approver.flow_level.unwrap_or(false) { + get_flow_id_for_job(&db, job_id).await? + } else { + job_id + }; + + let signature = create_signature(key, target_job_id, resume_id, approver.approver.clone())?; + let approver_query = approver .approver .as_ref() .map(|x| format!("?approver={}", encode(x))) @@ -3095,19 +3161,46 @@ pub async fn get_resume_urls_internal( let base_url = base_url_str.as_str(); let res = ResumeUrls { approvalPage: format!( - "{base_url}/approve/{w_id}/{job_id}/{resume_id}/{signature}{approver}" + "{base_url}/approve/{w_id}/{target_job_id}/{resume_id}/{signature}{approver_query}" ), cancel: build_resume_url( - "cancel", &w_id, &job_id, &resume_id, &signature, &approver, &base_url, + "cancel", &w_id, &target_job_id, &resume_id, &signature, &approver_query, &base_url, ), resume: build_resume_url( - "resume", &w_id, &job_id, &resume_id, &signature, &approver, &base_url, + "resume", &w_id, &target_job_id, &resume_id, &signature, &approver_query, &base_url, ), }; Ok(Json(res)) } +/// Get the flow ID for a job. If the job is a flow, returns the job_id. +/// If the job is a step in a flow, returns the parent flow ID. +async fn get_flow_id_for_job(db: &DB, job_id: Uuid) -> error::Result { + // First check if the job is a flow itself (kind = 'flow' or 'flowpreview') + let job_info = sqlx::query!( + r#" + SELECT kind::text as "kind!", parent_job + FROM v2_job + WHERE id = $1 + "#, + job_id + ) + .fetch_optional(db) + .await? + .ok_or_else(|| anyhow::anyhow!("job not found: {}", job_id))?; + + // If it's a flow job, return the job_id itself + if job_info.kind == "flow" || job_info.kind == "flowpreview" { + return Ok(job_id); + } + + // Otherwise, return the parent flow ID + job_info + .parent_job + .ok_or_else(|| anyhow::anyhow!("job {} has no parent flow", job_id).into()) +} + #[derive(sqlx::FromRow, Debug, Serialize)] pub struct JobExtended { #[sqlx(flatten)] @@ -8417,7 +8510,7 @@ async fn get_completed_job_result( &db, suspended_job, resume_id, - &QueryApprover { approver }, + &QueryApprover { approver, flow_level: None }, secret, ) .await? diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 63b7c493ef..73e3559345 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2518,10 +2518,13 @@ async fn push_next_flow_job( .await .context("lock flow in queue")?; + // Query for both step-level resumes (job = step_id) and flow-level resumes (job = flow_id) + // Flow-level resumes allow pre-approvals that can be consumed by any suspend step let resumes = sqlx::query_as::<_, ResumeRow>( - "SELECT value, approver, resume_id, approved FROM resume_job WHERE job = $1 ORDER BY created_at ASC", + "SELECT value, approver, resume_id, approved FROM resume_job WHERE job = $1 OR job = $2 ORDER BY created_at ASC", ) .bind(last) + .bind(flow_job.id) .fetch_all(&mut *tx) .warn_after_seconds(3) .await diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 28f4da28c4..56f6c6d6fc 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -1179,20 +1179,26 @@ class Windmill: with open(f"/shared/{path}", "r", encoding="utf-8") as f: return json.load(f) - def get_resume_urls(self, approver: str = None) -> dict: + def get_resume_urls(self, approver: str = None, flow_level: bool = None) -> dict: """Get URLs needed for resuming a flow after suspension. Args: approver: Optional approver name + flow_level: If True, generate resume URLs for the parent flow instead of the + specific step. This allows pre-approvals that can be consumed by any later + suspend step in the same flow. Returns: Dictionary with approvalPage, resume, and cancel URLs """ nonce = random.randint(0, 1000000000) job_id = os.environ.get("WM_JOB_ID") or "NO_ID" + params = {"approver": approver} + if flow_level is not None: + params["flow_level"] = flow_level return self.get( f"/w/{self.workspace}/jobs/resume_urls/{job_id}/{nonce}", - params={"approver": approver}, + params=params, ).json() def request_interactive_slack_approval( @@ -1887,16 +1893,19 @@ def get_state_path() -> str: @init_global_client -def get_resume_urls(approver: str = None) -> dict: +def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict: """Get URLs needed for resuming a flow after suspension. Args: approver: Optional approver name + flow_level: If True, generate resume URLs for the parent flow instead of the + specific step. This allows pre-approvals that can be consumed by any later + suspend step in the same flow. Returns: Dictionary with approvalPage, resume, and cancel URLs """ - return _client.get_resume_urls(approver) + return _client.get_resume_urls(approver, flow_level) @init_global_client diff --git a/typescript-client/client.d.ts b/typescript-client/client.d.ts index 1ab545f3f7..01803b6fb4 100644 --- a/typescript-client/client.d.ts +++ b/typescript-client/client.d.ts @@ -230,9 +230,11 @@ export declare function getPresignedS3PublicUrl( /** * Get URLs needed for resuming a flow after this step * @param approver approver name + * @param flowLevel if true, generate resume URLs for the parent flow instead of the specific step. + * This allows pre-approvals that can be consumed by any later suspend step in the same flow. * @returns approval page UI URL, resume and cancel API URLs for resuming the flow */ -export declare function getResumeUrls(approver?: string): Promise<{ +export declare function getResumeUrls(approver?: string, flowLevel?: boolean): Promise<{ approvalPage: string; resume: string; cancel: string; diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 7f3b12baf0..d959f8a970 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1043,9 +1043,11 @@ export async function getPresignedS3PublicUrl( /** * Get URLs needed for resuming a flow after this step * @param approver approver name + * @param flowLevel if true, generate resume URLs for the parent flow instead of the specific step. + * This allows pre-approvals that can be consumed by any later suspend step in the same flow. * @returns approval page UI URL, resume and cancel API URLs for resuming the flow */ -export async function getResumeUrls(approver?: string): Promise<{ +export async function getResumeUrls(approver?: string, flowLevel?: boolean): Promise<{ approvalPage: string; resume: string; cancel: string; @@ -1056,6 +1058,7 @@ export async function getResumeUrls(approver?: string): Promise<{ workspace, resumeId: nonce, approver, + flowLevel, id: getEnv("WM_JOB_ID") ?? "NO_JOB_ID", }); }