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

fix(backend): retrieve root_job for MiniPulledJob + fix root job for flow jobs (#6490)

Browse Source 
* feat(backend): retrieve root_job for MiniPulledJob + fix root job for flow jobs

* nit

* nits

* set root_job only if additional information meaning if not equal to innermost or parent

* nit

* feat: add root job env var

* nits

* fix build

* nit

* fix build

* nit

* sqlx
This commit is contained in:
hugocasa
2025-09-02 11:44:09 +02:00
committed by GitHub
parent 8c6053b575
commit de47d61d71
25 changed files with 230 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json
Generated
+77
View File
@@ -0,0 +1,77 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n v2_job.permissioned_as_email,\n v2_job.created_by,\n v2_job.parent_job,\n v2_job.permissioned_as,\n v2_job.runnable_path,\n CASE WHEN v2_job.trigger_kind = 'schedule'::job_trigger_kind THEN v2_job.trigger END AS schedule_path,\n v2_job.flow_step_id,\n v2_job.flow_innermost_root_job,\n v2_job.root_job,\n v2_job_queue.scheduled_for AS \"scheduled_for: chrono::DateTime<chrono::Utc>\"\n FROM v2_job INNER JOIN v2_job_queue ON v2_job.id = v2_job_queue.id\n WHERE v2_job.id = $1 AND v2_job.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "permissioned_as_email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "parent_job",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "permissioned_as",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "runnable_path",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "schedule_path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "flow_step_id",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "flow_innermost_root_job",
"type_info": "Uuid"
},
{
"ordinal": 8,
"name": "root_job",
"type_info": "Uuid"
},
{
"ordinal": 9,
"name": "scheduled_for: chrono::DateTime<chrono::Utc>",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false,
false,
true,
false,
true,
null,
true,
true,
true,
false
]
},
"hash": "02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0"
}
backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as \"root_job!\" FROM v2_job WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "root_job!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73"
}
backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json
Generated
-71
View File
@@ -1,71 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n email AS \"email!\",\n created_by AS \"created_by!\",\n parent_job, permissioned_as AS \"permissioned_as!\",\n script_path, schedule_path, flow_step_id, root_job,\n scheduled_for AS \"scheduled_for!: chrono::DateTime<chrono::Utc>\"\n FROM v2_as_queue WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email!",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "parent_job",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "permissioned_as!",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "schedule_path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "flow_step_id",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "root_job",
"type_info": "Uuid"
},
{
"ordinal": 8,
"name": "scheduled_for!: chrono::DateTime<chrono::Utc>",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true,
true,
true,
true,
true,
true,
true
]
},
"hash": "b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2"
}
backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json → backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json
Generated
+16 -10
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"describe": {
"columns": [
{
@@ -173,41 +173,46 @@
},
{
"ordinal": 22,
"name": "root_job",
"type_info": "Uuid"
},
{
"ordinal": 23,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 23,
"ordinal": 24,
"name": "flow_step_id",
"type_info": "Varchar"
},
{
"ordinal": 24,
"ordinal": 25,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 25,
"ordinal": 26,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 26,
"ordinal": 27,
"name": "preprocessed",
"type_info": "Bool"
},
{
"ordinal": 27,
"ordinal": 28,
"name": "script_entrypoint_override",
"type_info": "Varchar"
},
{
"ordinal": 28,
"ordinal": 29,
"name": "trigger",
"type_info": "Varchar"
},
{
"ordinal": 29,
"ordinal": 30,
"name": "trigger_kind: JobTriggerKind",
"type_info": {
"Custom": {
@@ -233,7 +238,7 @@
}
},
{
"ordinal": 30,
"ordinal": 31,
"name": "visible_to_owner",
"type_info": "Bool"
}
@@ -274,8 +279,9 @@
true,
true,
true,
true,
false
]
},
"hash": "4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0"
"hash": "f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22"
}
backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow_version.value->>'early_return' as early_return\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "early_return",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
d16e52d570f10dfdabb04a9061fe7ebdbf5a2aa1
a5911739330d09cdc084bb277e7e3db8f8d86801
backend/tests/worker.rs
+1
View File
@@ -949,6 +949,7 @@ impl RunJob {
/* schedule_path */ None,
/* parent_job */ None,
/* root job */ None,
/* flow_innermost_root_job */ None,
/* job_id */ None,
/* is_flow_step */ false,
/* same_worker */ false,
backend/windmill-api/src/apps.rs
+3
View File
@@ -1089,6 +1089,7 @@ async fn create_app_internal<'a>(
None,
None,
None,
None,
false,
false,
None,
@@ -1467,6 +1468,7 @@ async fn update_app_internal<'a>(
None,
None,
None,
None,
false,
false,
None,
@@ -1780,6 +1782,7 @@ async fn execute_component(
None,
None,
None,
None,
job_id,
false,
false,
backend/windmill-api/src/flows.rs
+2
View File
@@ -510,6 +510,7 @@ async fn create_flow(
None,
None,
None,
None,
false,
false,
None,
@@ -977,6 +978,7 @@ async fn update_flow(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/jobs.rs
+24 -15
View File
@@ -332,21 +332,16 @@ async fn get_root_job(
Ok(Json(res))
}
async fn compute_root_job_for_flow(db: &DB, w_id: &str, mut job_id: Uuid) -> error::Result<String> {
// TODO: use `root_job` ?
loop {
job_id = match sqlx::query_scalar!(
"SELECT parent_job FROM v2_job WHERE id = $1 AND workspace_id = $2",
job_id,
w_id
)
.fetch_one(db)
.await
{
Ok(Some(job_id)) => job_id,
_ => return Ok(job_id.to_string()),
}
}
async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result<String> {
let root_job = sqlx::query_scalar!(
r#"SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as "root_job!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#,
job_id,
w_id
)
.fetch_one(db)
.await?;
Ok(root_job.to_string())
}
async fn get_db_clock(Extension(db): Extension<DB>) -> windmill_common::error::JsonResult<i64> {
@@ -3875,6 +3870,7 @@ pub async fn run_flow_by_path_inner(
scheduled_for,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -3970,6 +3966,7 @@ pub async fn restart_flow(
scheduled_for,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -4066,6 +4063,7 @@ pub async fn run_script_by_path_inner(
scheduled_for,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -4215,6 +4213,7 @@ pub async fn run_workflow_as_code(
scheduled_for,
None,
Some(job_id),
None,
job.root_job.or(Some(job_id)),
run_query.job_id,
false,
@@ -4746,6 +4745,7 @@ pub async fn run_wait_result_job_by_path_get(
None,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -4897,6 +4897,7 @@ pub async fn run_wait_result_script_by_path_internal(
None,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -5012,6 +5013,7 @@ pub async fn run_wait_result_script_by_hash(
None,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -5130,6 +5132,7 @@ pub async fn run_wait_result_flow_by_path_internal(
scheduled_for,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
@@ -5201,6 +5204,7 @@ async fn run_preview_script(
None,
None,
None,
None,
run_query.job_id,
false,
false,
@@ -5312,6 +5316,7 @@ async fn run_bundle_preview_script(
None,
None,
None,
None,
run_query.job_id,
false,
false,
@@ -5479,6 +5484,7 @@ async fn run_dependencies_job(
None,
None,
None,
None,
false,
false,
None,
@@ -5545,6 +5551,7 @@ async fn run_flow_dependencies_job(
None,
None,
None,
None,
false,
false,
None,
@@ -5885,6 +5892,7 @@ async fn run_preview_flow_job(
None,
None,
None,
None,
run_query.job_id,
false,
false,
@@ -6024,6 +6032,7 @@ pub async fn run_job_by_hash_inner(
scheduled_for,
None,
run_query.parent_job,
None,
run_query.root_job,
run_query.job_id,
false,
backend/windmill-api/src/resources.rs
+15 -8
View File
@@ -573,12 +573,18 @@ pub async fn transform_json_value<'c>(
let job_id = job_id.unwrap();
let job = sqlx::query!(
"SELECT
email AS \"email!\",
created_by AS \"created_by!\",
parent_job, permissioned_as AS \"permissioned_as!\",
script_path, schedule_path, flow_step_id, root_job,
scheduled_for AS \"scheduled_for!: chrono::DateTime<chrono::Utc>\"
FROM v2_as_queue WHERE id = $1 AND workspace_id = $2",
v2_job.permissioned_as_email,
v2_job.created_by,
v2_job.parent_job,
v2_job.permissioned_as,
v2_job.runnable_path,
CASE WHEN v2_job.trigger_kind = 'schedule'::job_trigger_kind THEN v2_job.trigger END AS schedule_path,
v2_job.flow_step_id,
v2_job.flow_innermost_root_job,
v2_job.root_job,
v2_job_queue.scheduled_for AS \"scheduled_for: chrono::DateTime<chrono::Utc>\"
FROM v2_job INNER JOIN v2_job_queue ON v2_job.id = v2_job_queue.id
WHERE v2_job.id = $1 AND v2_job.workspace_id = $2",
job_id,
workspace
)
@@ -605,15 +611,16 @@ pub async fn transform_json_value<'c>(
&db.into(),
workspace,
token,
&job.email,
&job.permissioned_as_email,
&job.created_by,
&job_id.to_string(),
&job.permissioned_as,
job.script_path.clone(),
job.runnable_path.clone(),
job.parent_job.map(|x| x.to_string()),
flow_path,
job.schedule_path.clone(),
job.flow_step_id.clone(),
job.flow_innermost_root_job.map(|x| x.to_string()),
job.root_job.map(|x| x.to_string()),
Some(job.scheduled_for.clone()),
None,
backend/windmill-api/src/scripts.rs
+1
View File
@@ -1007,6 +1007,7 @@ async fn create_script_internal<'c>(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/trigger_helpers.rs
+1
View File
@@ -774,6 +774,7 @@ async fn trigger_script_with_retry_and_error_handler(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/variables.rs
+1
View File
@@ -78,6 +78,7 @@ async fn list_contextual_variables(
Some("u/user/triggering_flow_path".to_string()),
Some("c".to_string()),
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
Some(chrono::offset::Utc::now()),
Some(ScriptHash(1234567890)),
)
backend/windmill-common/src/variables.rs
+10 -3
View File
@@ -210,7 +210,8 @@ pub async fn get_reserved_variables(
flow_path: Option<String>,
schedule_path: Option<String>,
step_id: Option<String>,
root_flow_id: Option<String>,
flow_innermost_root_job: Option<String>,
root_job_id: Option<String>,
scheduled_for: Option<chrono::DateTime<Utc>>,
runnable_id: Option<ScriptHash>,
) -> Vec<ContextualVariable> {
@@ -322,8 +323,14 @@ pub async fn get_reserved_variables(
},
ContextualVariable {
name: "WM_ROOT_FLOW_JOB_ID".to_string(),
value: root_flow_id.unwrap_or_else(|| "".to_string()),
description: "Job id of the root flow if the job is a flow step".to_string(),
value: flow_innermost_root_job.unwrap_or_else(|| "".to_string()),
description: "Job id of the innermost root flow if the job is a flow step".to_string(),
is_custom: false,
},
ContextualVariable {
name: "WM_ROOT_JOB_ID".to_string(),
value: root_job_id.unwrap_or_else(|| "".to_string()),
description: "Job id of the root job".to_string(),
is_custom: false,
},
ContextualVariable {
backend/windmill-common/src/worker.rs
+2 -2
View File
@@ -380,7 +380,7 @@ fn format_pull_query(peek: String) -> String {
id, workspace_id, parent_job, created_by, created_at, runnable_id,
runnable_path, args, kind, trigger, trigger_kind,
permissioned_as, permissioned_as_email, script_lang,
flow_innermost_root_job, flow_step_id,
flow_innermost_root_job, root_job, flow_step_id,
same_worker, pre_run_error, visible_to_owner, tag, concurrent_limit,
concurrency_time_window_s, timeout, cache_ttl, priority, raw_code, raw_lock,
raw_flow, script_entrypoint_override, preprocessed
@@ -391,7 +391,7 @@ fn format_pull_query(peek: String) -> String {
canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as,
flow_status, j.script_lang,
j.same_worker, j.pre_run_error, j.visible_to_owner,
j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job,
j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job,
j.timeout, j.flow_step_id, j.cache_ttl, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
j.script_entrypoint_override, j.preprocessed, pj.runnable_path as parent_runnable_path,
COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
backend/windmill-queue/src/jobs.rs
+24 -3
View File
@@ -439,6 +439,7 @@ pub async fn push_init_job<'c>(
None,
None,
None,
None,
false,
true,
None,
@@ -492,6 +493,7 @@ pub async fn push_periodic_bash_job<'c>(
None,
None,
None,
None,
false,
true,
None,
@@ -1267,6 +1269,7 @@ async fn restart_job_if_perpetual_inner(
None,
None,
None,
None,
false,
false,
None,
@@ -2029,6 +2032,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>(
None,
None,
Some(job_id),
None,
Some(job_id),
None,
false,
@@ -2138,6 +2142,7 @@ async fn handle_recovered_schedule<'a, 'c, T: Serialize + Send + Sync>(
None,
None,
Some(job_id),
None,
Some(job_id),
None,
false,
@@ -2228,6 +2233,7 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>(
None,
None,
Some(job_id),
None,
Some(job_id),
None,
false,
@@ -2311,6 +2317,7 @@ pub struct MiniPulledJob {
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub flow_innermost_root_job: Option<Uuid>,
pub root_job: Option<Uuid>,
pub timeout: Option<i32>,
pub flow_step_id: Option<String>,
pub cache_ttl: Option<i32>,
@@ -2369,7 +2376,8 @@ impl MiniPulledJob {
pre_run_error: job.pre_run_error.clone(),
concurrent_limit: job.concurrent_limit.clone(),
concurrency_time_window_s: job.concurrency_time_window_s.clone(),
flow_innermost_root_job: job.root_job.clone(),
flow_innermost_root_job: job.root_job.clone(), // QueuedJob is taken from v2_as_queue, where root_job corresponds to flow_innermost_root_job in v2_job
root_job: None,
timeout: job.timeout.clone(),
flow_step_id: job.flow_step_id.clone(),
cache_ttl: job.cache_ttl.clone(),
@@ -2574,6 +2582,7 @@ pub async fn get_mini_pulled_job<'c>(
concurrent_limit,
concurrency_time_window_s,
flow_innermost_root_job,
root_job,
timeout,
flow_step_id,
cache_ttl,
@@ -3623,6 +3632,7 @@ pub async fn push<'c, 'd>(
schedule_path: Option<String>,
parent_job: Option<Uuid>,
root_job: Option<Uuid>,
flow_innermost_root_job: Option<Uuid>,
job_id: Option<Uuid>,
_is_flow_step: bool,
mut same_worker: bool, // whether the job will be executed on the same worker: if true, the job will be set to running but started_at will not be set.
@@ -4683,6 +4693,16 @@ pub async fn push<'c, 'd>(
None
};
let root_job = if root_job.is_some()
&& (root_job == flow_innermost_root_job.or(parent_job).or(Some(job_id)))
{
// We only save the root job if it's not the innermost root job, parent job, or the job itself as an optimization
// Reference: see [`windmill_worker::common::get_root_job_id`] for logic on determining the root job.
None
} else {
root_job
};
sqlx::query!(
"WITH inserted_job AS (
INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,
@@ -4724,7 +4744,7 @@ pub async fn push<'c, 'd>(
pre_run_error.map(|e| e.to_string()),
email,
visible_to_owner,
root_job,
flow_innermost_root_job,
concurrent_limit,
if concurrent_limit.is_some() {
concurrency_time_window_s
@@ -4746,7 +4766,7 @@ pub async fn push<'c, 'd>(
job_authed.is_operator,
folders.as_slice(),
job_authed.groups.as_slice(),
root_job.or(parent_job),
root_job,
trigger_kind as Option<JobTriggerKind>,
running,
)
@@ -5137,6 +5157,7 @@ pub async fn get_same_worker_job(
v2_job.concurrent_limit,
v2_job.concurrency_time_window_s,
v2_job.flow_innermost_root_job,
v2_job.root_job,
v2_job.timeout,
v2_job.flow_step_id,
v2_job.cache_ttl,
backend/windmill-queue/src/schedule.rs
+1
View File
@@ -294,6 +294,7 @@ pub async fn push_scheduled_job<'c>(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-worker/src/ai_executor.rs
+1
View File
@@ -484,6 +484,7 @@ async fn call_tool(
agent_job.schedule_path(),
Some(agent_job.id),
None,
None,
Some(job_id),
false,
false,
backend/windmill-worker/src/bun_executor.rs
+1
View File
@@ -1590,6 +1590,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/common.rs
+10 -22
View File
@@ -211,25 +211,6 @@ pub fn parse_npm_config(s: &str) -> (String, Option<String>) {
return (url, token_opt);
}
#[async_recursion]
pub async fn get_root_job_id(job: &Uuid, db: &Pool<Postgres>) -> anyhow::Result<Uuid> {
let njob = sqlx::query_scalar!(
"SELECT flow_innermost_root_job FROM v2_job WHERE id = $1",
job
)
.fetch_optional(db)
.await?
.flatten();
if let Some(root_job) = njob {
if root_job == *job {
return Ok(job.to_owned());
}
get_root_job_id(&root_job, db).await
} else {
Ok(job.to_owned())
}
}
#[async_recursion]
pub async fn transform_json_value(
name: &str,
@@ -272,9 +253,7 @@ pub async fn transform_json_value(
Connection::Sql(db) => {
let encrypted = y.strip_prefix("$encrypted:").unwrap();
let root_job_id =
get_root_job_id(&job.flow_innermost_root_job.unwrap_or_else(|| job.id), db)
.await?;
let root_job_id = get_root_job_id(&job);
let mc = build_crypt_with_key_suffix(
&db,
&job.workspace_id,
@@ -463,6 +442,7 @@ pub async fn get_reserved_variables(
job.schedule_path(),
job.flow_step_id.clone(),
job.flow_innermost_root_job.clone().map(|x| x.to_string()),
Some(get_root_job_id(job).to_string()),
Some(job.scheduled_for.clone()),
job.runnable_id,
)
@@ -1073,6 +1053,14 @@ pub fn build_http_client(timeout_duration: std::time::Duration) -> error::Result
.map_err(|e| Error::internal_err(format!("Error building http client: {e:#}")))
}
pub fn get_root_job_id(job: &MiniPulledJob) -> uuid::Uuid {
// fallback to flow_innermost_root_job and parent_job as root_job is not set if equal to innermost root job or parent job
job.root_job
.or(job.flow_innermost_root_job)
.or(job.parent_job)
.unwrap_or(job.id)
}
#[derive(Clone)]
pub struct S3ModeWorkerData {
pub client: AuthedClient,
backend/windmill-worker/src/deno_executor.rs
+1
View File
@@ -545,6 +545,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/python_executor.rs
+2
View File
@@ -2146,6 +2146,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await
.to_vec();
@@ -2266,6 +2267,7 @@ for line in sys.stdin:
None,
None,
None,
None,
)
.await;
backend/windmill-worker/src/worker_flow.rs
+12 -16
View File
@@ -11,7 +11,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use crate::common::{cached_result_path, save_in_cache};
use crate::common::{cached_result_path, get_root_job_id, save_in_cache};
use crate::js_eval::{eval_timeout, IdContext};
use crate::worker_utils::get_tag_and_concurrency;
use crate::{
@@ -2816,7 +2816,7 @@ async fn push_next_flow_job(
let value_with_parallel = module.get_value_with_parallel()?;
let root_job = if {
let flow_innermost_root_job = if {
value_with_parallel.type_ == "flow"
|| (value_with_parallel.type_ == "forloopflow"
&& value_with_parallel.parallel.is_some_and(|x| x))
@@ -2828,19 +2828,13 @@ async fn push_next_flow_job(
.or_else(|| Some(flow_job.id))
};
let flow_root_job = get_root_job_id(&flow_job);
// forward root job permissions to the new job
let job_perms: Option<Authed> = {
if let Some(root_job) = &flow_job
.flow_innermost_root_job
.or_else(|| Some(flow_job.id))
{
get_job_perms(&mut *tx, root_job, &flow_job.workspace_id)
.await?
.map(|x| x.into())
} else {
None
}
};
let job_perms: Option<Authed> =
get_job_perms(&mut *tx, &flow_root_job, &flow_job.workspace_id)
.await?
.map(|x| x.into());
tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
let tag = if !matches!(step, Step::PreprocessorStep)
@@ -2860,6 +2854,7 @@ async fn push_next_flow_job(
flow_job.permissioned_as.to_owned(),
)
};
let tx2 = PushIsolationLevel::Transaction(tx);
let (uuid, mut inner_tx) = push(
&db,
@@ -2877,7 +2872,8 @@ async fn push_next_flow_job(
scheduled_for_o,
flow_job.schedule_path(),
Some(flow_job.id),
root_job,
Some(flow_root_job),
flow_innermost_root_job,
None,
true,
continue_on_same_worker,
@@ -2938,7 +2934,7 @@ async fn push_next_flow_job(
SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)
WHERE id = $2",
uuid_singleton_json,
root_job.unwrap_or(flow_job.id)
flow_innermost_root_job.unwrap_or(flow_job.id)
)
.execute(&mut *inner_tx)
.warn_after_seconds(3)
backend/windmill-worker/src/worker_lockfiles.rs
+1
View File
@@ -739,6 +739,7 @@ pub async fn trigger_dependents_to_recompute_dependencies(
None,
None,
None,
None,
false,
false,
None,