starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-26 08:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: allow resume urls at flow level for pre-generation (#7582)

Browse Source
This commit is contained in:
Ruben Fiszel
2026-01-15 08:56:09 +00:00
committed by GitHub
parent cebb47ea20
commit ea7a9f8d01
11 changed files with 238 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
true
null
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-66e66da2ed6eace5d7ec2a41a7b11ae255f5dc212d1ff41c2905b303c8c13b18.json
Generated
+46
View File
@@ -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"
}
backend/.sqlx/query-83785ee2f7dcc7f2252b0e8bcc8322dfd7689d615a34b63e29c9c6699b7e5514.json
Generated
-40
View File
@@ -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"
}
backend/.sqlx/query-a87e7b098b523176916daf5b1685b218dfbfe752aa730b421e9229dadd6ae587.json
Generated
+28
View File
@@ -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"
}
backend/windmill-api/openapi.yaml
+5
View File
@@ -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
backend/windmill-api/src/approvals.rs
+2 -2
View File
@@ -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?;
backend/windmill-api/src/jobs.rs
+132 -39
View File
@@ -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<String>,
}
async fn get_suspended_parent_flow_info(job_id: Uuid, db: &DB) -> error::Result<FlowInfo> {
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<String>,
/// 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<bool>,
}
pub async fn get_suspended_job_flow(
@@ -3084,8 +3141,17 @@ pub async fn get_resume_urls_internal(
Query(approver): Query<QueryApprover>,
) -> error::JsonResult<ResumeUrls> {
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<Uuid> {
// 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<T: JobCommon> {
#[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?
backend/windmill-worker/src/worker_flow.rs
+4 -1
View File
@@ -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
python-client/wmill/wmill/client.py
+13 -4
View File
@@ -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
typescript-client/client.d.ts
Vendored
+3 -1
View File
@@ -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;
typescript-client/client.ts
+4 -1
View File
@@ -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",
});
}