From de47d61d7182d1ef525e101ff23be484e135184d Mon Sep 17 00:00:00 2001 From: hugocasa Date: Tue, 2 Sep 2025 11:44:09 +0200 Subject: [PATCH] fix(backend): retrieve root_job for MiniPulledJob + fix root job for flow jobs (#6490) * 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 --- ...46d79167747742de236ce57f395561685adc0.json | 77 +++++++++++++++++++ ...335f578c622eb335cfc7cd0969633d85c9f73.json | 23 ++++++ ...ed2d4534e08471a60005bff85b952874c54c2.json | 71 ----------------- ...7b578bc63792cdb0f7ddaeb82e9771287c22.json} | 26 ++++--- ...dfdfc49981568e8c496fb6a163c499c3e4ad1.json | 23 ------ backend/ee-repo-ref.txt | 2 +- backend/tests/worker.rs | 1 + backend/windmill-api/src/apps.rs | 3 + backend/windmill-api/src/flows.rs | 2 + backend/windmill-api/src/jobs.rs | 39 ++++++---- backend/windmill-api/src/resources.rs | 23 ++++-- backend/windmill-api/src/scripts.rs | 1 + backend/windmill-api/src/trigger_helpers.rs | 1 + backend/windmill-api/src/variables.rs | 1 + backend/windmill-common/src/variables.rs | 13 +++- backend/windmill-common/src/worker.rs | 4 +- backend/windmill-queue/src/jobs.rs | 27 ++++++- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/src/ai_executor.rs | 1 + backend/windmill-worker/src/bun_executor.rs | 1 + backend/windmill-worker/src/common.rs | 32 +++----- backend/windmill-worker/src/deno_executor.rs | 1 + .../windmill-worker/src/python_executor.rs | 2 + backend/windmill-worker/src/worker_flow.rs | 28 +++---- .../windmill-worker/src/worker_lockfiles.rs | 1 + 25 files changed, 230 insertions(+), 174 deletions(-) create mode 100644 backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json create mode 100644 backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json delete mode 100644 backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json rename backend/.sqlx/{query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json => query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json} (90%) delete mode 100644 backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json diff --git a/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json b/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json new file mode 100644 index 0000000000..8fd07f3bf9 --- /dev/null +++ b/backend/.sqlx/query-02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0.json @@ -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\"\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", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + true, + false, + true, + null, + true, + true, + true, + false + ] + }, + "hash": "02c945b5f18a56a826721f6884846d79167747742de236ce57f395561685adc0" +} diff --git a/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json b/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json new file mode 100644 index 0000000000..34208d501c --- /dev/null +++ b/backend/.sqlx/query-0e14ab95a08572f0672db266187335f578c622eb335cfc7cd0969633d85c9f73.json @@ -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" +} diff --git a/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json b/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json deleted file mode 100644 index fabcec9b2a..0000000000 --- a/backend/.sqlx/query-b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2.json +++ /dev/null @@ -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\"\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", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - }, - "nullable": [ - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "b2e4dfaaee713604d0700ea4675ed2d4534e08471a60005bff85b952874c54c2" -} diff --git a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json similarity index 90% rename from backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json rename to backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json index fa33c3b8e0..28c0756b0f 100644 --- a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json +++ b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json @@ -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>>\",\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>\",\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>>\",\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>\",\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" } diff --git a/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json b/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json deleted file mode 100644 index 57ae674eb4..0000000000 --- a/backend/.sqlx/query-fc243af1bc70f04e28c006364d6dfdfc49981568e8c496fb6a163c499c3e4ad1.json +++ /dev/null @@ -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" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index dfa2a01850..76cf87e58c 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -d16e52d570f10dfdabb04a9061fe7ebdbf5a2aa1 \ No newline at end of file +a5911739330d09cdc084bb277e7e3db8f8d86801 \ No newline at end of file diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 85d7555e1b..dcf20ae402 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -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, diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 37743eae34..d20c4052ed 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -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, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 9ea840a2d1..4e9b3e4a18 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -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, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5332bcb6c4..b29d683a71 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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 { - // 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 { + 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) -> windmill_common::error::JsonResult { @@ -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, diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index 85f09491e1..f07c3b6760 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -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\" - 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\" + 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, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index f0b2c4c483..9f55d22d65 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -1007,6 +1007,7 @@ async fn create_script_internal<'c>( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/trigger_helpers.rs b/backend/windmill-api/src/trigger_helpers.rs index 1c2bdc029c..e8211d471c 100644 --- a/backend/windmill-api/src/trigger_helpers.rs +++ b/backend/windmill-api/src/trigger_helpers.rs @@ -774,6 +774,7 @@ async fn trigger_script_with_retry_and_error_handler( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index a14e954517..2250bdfd34 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -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)), ) diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index e60b8d9174..ae81512a0a 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -210,7 +210,8 @@ pub async fn get_reserved_variables( flow_path: Option, schedule_path: Option, step_id: Option, - root_flow_id: Option, + flow_innermost_root_job: Option, + root_job_id: Option, scheduled_for: Option>, runnable_id: Option, ) -> Vec { @@ -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 { diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7ec4cd09f5..745102bf66 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -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, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 3667239119..d84e5cdc97 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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, pub concurrency_time_window_s: Option, pub flow_innermost_root_job: Option, + pub root_job: Option, pub timeout: Option, pub flow_step_id: Option, pub cache_ttl: Option, @@ -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, parent_job: Option, root_job: Option, + flow_innermost_root_job: Option, job_id: Option, _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, 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, diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index ae3653881a..cd682629c2 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -294,6 +294,7 @@ pub async fn push_scheduled_job<'c>( None, None, None, + None, false, false, None, diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index c8ee996bc7..b9a7cfcd88 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -484,6 +484,7 @@ async fn call_tool( agent_job.schedule_path(), Some(agent_job.id), None, + None, Some(job_id), false, false, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index d0ad991bf2..d64926b580 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1590,6 +1590,7 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 1df44c34df..ce1619da8b 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -211,25 +211,6 @@ pub fn parse_npm_config(s: &str) -> (String, Option) { return (url, token_opt); } -#[async_recursion] -pub async fn get_root_job_id(job: &Uuid, db: &Pool) -> anyhow::Result { - 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, diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 607e6375d2..a61b581262 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -545,6 +545,7 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 70f980052b..e8291f771a 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -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; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index f4874f6456..36f255a836 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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 = { - 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 = + 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) diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 358e93e0a1..38a451d8fa 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -739,6 +739,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( None, None, None, + None, false, false, None,