From 58300eb6ac52d57f09c83dc481ba0fbf35e86501 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 3 Mar 2023 12:15:34 +0100 Subject: [PATCH] introduce root_job and leaf_jobs for efficient result_by_id --- ...20230303064952_add_root_leaf_jobs.down.sql | 1 + .../20230303064952_add_root_leaf_jobs.up.sql | 3 + backend/sqlx-data.json | 362 ++++++++++-------- backend/tests/worker.rs | 1 + backend/windmill-api/openapi.yaml | 5 - backend/windmill-api/src/apps.rs | 1 + backend/windmill-api/src/flows.rs | 2 + backend/windmill-api/src/jobs.rs | 20 +- backend/windmill-api/src/oauth2.rs | 1 + backend/windmill-api/src/scripts.rs | 1 + backend/windmill-common/src/flow_status.rs | 2 +- backend/windmill-queue/src/jobs.rs | 124 ++---- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/src/js_eval.rs | 2 +- backend/windmill-worker/src/worker_flow.rs | 17 + 15 files changed, 272 insertions(+), 271 deletions(-) create mode 100644 backend/migrations/20230303064952_add_root_leaf_jobs.down.sql create mode 100644 backend/migrations/20230303064952_add_root_leaf_jobs.up.sql diff --git a/backend/migrations/20230303064952_add_root_leaf_jobs.down.sql b/backend/migrations/20230303064952_add_root_leaf_jobs.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230303064952_add_root_leaf_jobs.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230303064952_add_root_leaf_jobs.up.sql b/backend/migrations/20230303064952_add_root_leaf_jobs.up.sql new file mode 100644 index 0000000000..0082c23344 --- /dev/null +++ b/backend/migrations/20230303064952_add_root_leaf_jobs.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE queue ADD COLUMN root_job uuid; +ALTER TABLE queue ADD COLUMN leaf_jobs jsonb; diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index b0111206ec..9296843537 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -645,27 +645,6 @@ }, "query": "SELECT * FROM workspace_settings WHERE workspace_id = $1" }, - "176afff6dfceac701d499663f15fb7f569fd9d449406df4c5ecc811414429d42": { - "describe": { - "columns": [ - { - "name": "flow_status", - "ordinal": 0, - "type_info": "Jsonb" - } - ], - "nullable": [ - null - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - } - }, - "query": "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2 \n UNION ALL SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2 " - }, "187a27d18f78f068319a0ec684f71a76c49dd09c5c74c2c777e35233c4f5a281": { "describe": { "columns": [ @@ -826,75 +805,6 @@ }, "query": "\n SELECT id, flow_status, suspend, script_path\n FROM queue\n WHERE id = $1\n " }, - "1e35c39bc786d638252e5483ca4efae9a041f7e845341f8bfd715ddd9e899499": { - "describe": { - "columns": [ - { - "name": "id", - "ordinal": 0, - "type_info": "Uuid" - } - ], - "nullable": [ - false - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Bool", - "Uuid", - "Varchar", - "Varchar", - "Timestamptz", - "Int8", - "Varchar", - "Text", - "Text", - "Jsonb", - { - "Custom": { - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies" - ] - }, - "name": "job_kind" - } - }, - "Varchar", - "Jsonb", - "Jsonb", - "Bool", - { - "Custom": { - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash" - ] - }, - "name": "script_lang" - } - }, - "Bool", - "Text", - "Varchar", - "Bool" - ] - } - }, - "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22) RETURNING id" - }, "1eaf8d677d520c7f2f303a731de6b6d939918e41ad0d1c748d80db3fd33cb9d3": { "describe": { "columns": [], @@ -1176,6 +1086,20 @@ }, "query": "SELECT set_config('session.folders_read', $1, true)" }, + "2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Text", + "Jsonb", + "Uuid" + ] + } + }, + "query": "\n UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id\n " + }, "2a4be8334db7d39f3d954193a8b0169cc4a4a07e081d2fa61d8764879d6a8ff5": { "describe": { "columns": [], @@ -1395,6 +1319,27 @@ }, "query": "SELECT EXISTS(SELECT 1 FROM resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'admins'))" }, + "39e423f16f2ffb74df27760345a5340f2f89409f2a16247ac92a1949181d0cf0": { + "describe": { + "columns": [ + { + "name": "leaf_jobs", + "ordinal": 0, + "type_info": "Jsonb" + } + ], + "nullable": [ + true + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + } + }, + "query": "SELECT leaf_jobs FROM queue WHERE id = (SELECT root_job FROM queue WHERE id = $1) AND workspace_id = $2" + }, "39f1b99319f48acb95fd4247f70baaf7a38062aa8b4f70fac10215edbbd2d41c": { "describe": { "columns": [], @@ -1893,6 +1838,28 @@ }, "query": "DELETE FROM usr_to_group WHERE group_ = $1 AND workspace_id = $2" }, + "52a96fa4034dec3edb0655c1cd125e6dacc63c10bbc6c0ea1789809807d720a3": { + "describe": { + "columns": [ + { + "name": "?column?", + "ordinal": 0, + "type_info": "Jsonb" + } + ], + "nullable": [ + null + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text" + ] + } + }, + "query": "SELECT leaf_jobs->$1::text FROM queue WHERE id = $2 AND workspace_id = $3" + }, "52c8b4350235bdaab4df79e517d5e42a61a4e1e209d120b2c8bb31ebb7ce1e56": { "describe": { "columns": [], @@ -2661,6 +2628,27 @@ }, "query": "SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND username = $2)" }, + "7165e50c7de4437ea1969aae4db51dfe040ce3b2f8dcf7eb123831697fcdbfd0": { + "describe": { + "columns": [ + { + "name": "leaf_jobs", + "ordinal": 0, + "type_info": "Jsonb" + } + ], + "nullable": [ + true + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + } + }, + "query": "SELECT leaf_jobs FROM queue WHERE id = $1 AND workspace_id = $2" + }, "72098030cab635723a9cecf8b3b1448e69a8afd68342850ef6376352d2897723": { "describe": { "columns": [], @@ -3657,33 +3645,6 @@ }, "query": "SELECT is_admin FROM usr where username = $1 AND workspace_id = $2 AND disabled = false" }, - "92ce38eaaef9e793a49788bf3b7fc88340e83da57b785364a477f08c776f66f3": { - "describe": { - "columns": [ - { - "name": "flow_status", - "ordinal": 0, - "type_info": "Jsonb" - }, - { - "name": "parent_job", - "ordinal": 1, - "type_info": "Uuid" - } - ], - "nullable": [ - null, - null - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - } - }, - "query": "SELECT flow_status, parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 \n UNION ALL SELECT flow_status, parent_job FROM queue WHERE id = $1 AND workspace_id = $2 " - }, "930ad84a4db26fa6d2c8447d447099e944a0ea7b1266b4d02cee620fe3d761a5": { "describe": { "columns": [], @@ -3830,6 +3791,26 @@ }, "query": "UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *" }, + "971175f6169857c3e1cdc08ac8aeed57300b7792e1797a9cdd73c9b3967cd7b9": { + "describe": { + "columns": [ + { + "name": "root_job", + "ordinal": 0, + "type_info": "Uuid" + } + ], + "nullable": [ + true + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "SELECT root_job FROM queue WHERE id = $1" + }, "97e364c703bdcdfb5345810659cbe0477a28b8199ef0b297f9a22c88a43b6b5c": { "describe": { "columns": [ @@ -4009,6 +3990,76 @@ }, "query": "SELECT content FROM script WHERE path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND archived = false AND workspace_id = $2)" }, + "a1c41bbeb2d64fa1e7dfd2ed053191a1de5d786ae8c22e225e450865ecac94e9": { + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Uuid" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Bool", + "Uuid", + "Varchar", + "Varchar", + "Timestamptz", + "Int8", + "Varchar", + "Text", + "Text", + "Jsonb", + { + "Custom": { + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies" + ] + }, + "name": "job_kind" + } + }, + "Varchar", + "Jsonb", + "Jsonb", + "Bool", + { + "Custom": { + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash" + ] + }, + "name": "script_lang" + } + }, + "Bool", + "Text", + "Varchar", + "Bool", + "Uuid" + ] + } + }, + "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23) RETURNING id" + }, "a227548b6604c56bfc15eb780bd8ee72a89dc6701a50f5048e928bd87baa7b9a": { "describe": { "columns": [ @@ -4119,6 +4170,28 @@ }, "query": "UPDATE flow SET dependency_job = $1 WHERE path = $2 AND workspace_id = $3" }, + "a5f9fb82791103e2bbaf9cb6d87e8c50495d12d87f8ed83382068203a8dd7a67": { + "describe": { + "columns": [ + { + "name": "?column?", + "ordinal": 0, + "type_info": "Jsonb" + } + ], + "nullable": [ + null + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text" + ] + } + }, + "query": "SELECT leaf_jobs->$1::text FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3" + }, "a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b": { "describe": { "columns": [], @@ -4821,26 +4894,6 @@ }, "query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *" }, - "c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc": { - "describe": { - "columns": [ - { - "name": "customer_id", - "ordinal": 0, - "type_info": "Varchar" - } - ], - "nullable": [ - true - ], - "parameters": { - "Left": [ - "Text" - ] - } - }, - "query": "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1" - }, "c1149f74ef0cca4a48be80f523d339d78e8e20e2bfacac9ad1b6b75c5ae370fb": { "describe": { "columns": [], @@ -4875,27 +4928,6 @@ }, "query": "SELECT result FROM completed_job WHERE id = $1" }, - "c2d0e44faab6981a21ca28dfd6f4eef9dfcafb471852e701c0bbc8ae11344325": { - "describe": { - "columns": [ - { - "name": "parent_job", - "ordinal": 0, - "type_info": "Uuid" - } - ], - "nullable": [ - null - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - } - }, - "query": "SELECT parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT parent_job FROM queue WHERE id = $1 AND workspace_id = $2" - }, "c2d6cb56c1dea4498e2aab9ea9301dbbaa127602a38f57f5add4108fdc209b1a": { "describe": { "columns": [ @@ -5816,6 +5848,20 @@ }, "query": "DELETE FROM password WHERE email = $1" }, + "e990488d2238fe229a213c975102b85ea2e159142bb2d34bc83835322496f2b9": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Text", + "Jsonb", + "Uuid" + ] + } + }, + "query": "\n UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE (SELECT root_job FROM queue WHERE id = $3) = id\n " + }, "e9c0e331c16312bf086b17c91466c5389d41454fd3f18d73c2e9554845ee9a72": { "describe": { "columns": [], diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 7a3b2d508f..e59d6d9864 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -836,6 +836,7 @@ impl RunJob { /* scheduled_for_o */ None, /* schedule_path */ None, /* parent_job */ None, + /* root job */ None, /* is_flow_step */ false, /* running */ false, None, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c80d82a8c3..a9a3af43a0 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2538,11 +2538,6 @@ paths: required: true schema: type: string - - name: skip_direct - description: Skip checking that the node is part of the given flow. - in: query - schema: - type: boolean responses: "200": description: job result diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index ad068efa1e..374ceba20a 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -670,6 +670,7 @@ async fn execute_component( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 52d24d1077..0976f228de 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -237,6 +237,7 @@ async fn create_flow( None, None, None, + None, false, false, None, @@ -393,6 +394,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 5685a47168..a34297fcf1 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -102,10 +102,9 @@ pub fn global_service() -> Router { async fn get_result_by_id( Extension(db): Extension, - Query(ResultByIdQuery { skip_direct }): Query, - Path((w_id, flow_id, node_id)): Path<(String, String, String)>, + Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>, ) -> windmill_common::error::JsonResult { - let res = windmill_queue::get_result_by_id(db, skip_direct, w_id, flow_id, node_id).await?; + let res = windmill_queue::get_result_by_id(db, w_id, flow_id, node_id).await?; Ok(Json(res)) } @@ -177,11 +176,6 @@ async fn get_job( Ok(Json(job)) } -#[derive(Deserialize)] -pub struct ResultByIdQuery { - pub skip_direct: bool, -} - pub async fn get_job_by_id<'c>( mut tx: Transaction<'c, Postgres>, w_id: &str, @@ -1078,6 +1072,8 @@ impl From for Job { visible_to_owner: uj.visible_to_owner, suspend: uj.suspend, mem_peak: uj.mem_peak, + root_job: None, + leaf_jobs: None, }), t => panic!("job type {} not valid", t), } @@ -1173,6 +1169,7 @@ pub async fn run_flow_by_path( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, @@ -1208,6 +1205,7 @@ pub async fn run_job_by_path( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, @@ -1360,6 +1358,7 @@ pub async fn run_wait_result_job_by_path( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, @@ -1406,6 +1405,7 @@ pub async fn run_wait_result_job_by_hash( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, @@ -1451,6 +1451,7 @@ pub async fn run_wait_result_flow_by_path( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, @@ -1513,6 +1514,7 @@ async fn run_preview_job( scheduled_for, None, None, + None, false, false, None, @@ -1546,6 +1548,7 @@ async fn run_preview_flow_job( scheduled_for, None, None, + None, false, false, None, @@ -1581,6 +1584,7 @@ pub async fn run_job_by_hash( scheduled_for, None, run_query.parent_job, + run_query.parent_job, false, false, None, diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 154a6e8351..b4979fb760 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -785,6 +785,7 @@ async fn slack_command( None, None, None, + None, false, false, None, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index b2e73d1b2e..1e6dec93f2 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -377,6 +377,7 @@ async fn create_script( None, None, None, + None, false, false, None, diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 66f98b1477..12f54c38ea 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -125,7 +125,7 @@ pub enum FlowStatusModule { }, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum JobResult { SingleJob(Uuid), ListJob(Vec), diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 51f8310334..527d38624e 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,10 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{ - collections::{HashMap, VecDeque}, - str::FromStr, -}; +use std::collections::HashMap; use anyhow::Context; use reqwest::Client; @@ -20,7 +17,7 @@ use ulid::Ulid; use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ - error::{self, to_anyhow, Error}, + error::{self, Error}, flow_status::{FlowStatus, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL}, flows::{FlowModule, FlowModuleValue, FlowValue}, scripts::{get_full_hub_script_by_path, HubScript, ScriptHash, ScriptLang}, @@ -153,102 +150,26 @@ pub async fn pull( Ok(job) } -pub async fn find_recursively_downward( - db: &Pool, - w_id: &str, - flow_id: Uuid, - node_id: &str, -) -> windmill_common::error::Result> { - let mut bfs_stack = VecDeque::new(); - bfs_stack.push_back(flow_id); - while bfs_stack.len() > 0 { - let parent_id = bfs_stack.pop_front().unwrap(); - let job = sqlx::query_scalar!( - "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2 - UNION ALL SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2 ", - parent_id, - w_id - ) - .fetch_optional(db) - .await? - .flatten(); - if let Some(r) = job { - let status = serde_json::from_value::(r).map_err(to_anyhow)?; - for m in status.modules.iter() { - let id = m.id(); - if id == node_id { - return Ok(m.job_result()); - } - if let Some(job_id) = m.job() { - bfs_stack.push_back(job_id); - } - } - } - } - Ok(None) -} - pub async fn get_result_by_id( db: Pool, - mut skip_direct: bool, w_id: String, - flow_id: String, + flow_id: Uuid, node_id: String, ) -> error::Result { - let mut result_id: Option = None; - let mut parent_id = Uuid::from_str(&flow_id).ok(); - let mut lparent_id = parent_id.clone(); - while result_id.is_none() && parent_id.is_some() { - if !skip_direct { - let r = sqlx::query!( - "SELECT flow_status, parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 - UNION ALL SELECT flow_status, parent_job FROM queue WHERE id = $1 AND workspace_id = $2 ", - parent_id.unwrap(), - w_id, - ) - .fetch_optional(&db) - .await?; - if let Some(r) = r { - let value = r - .flow_status - .as_ref() - .ok_or_else(|| Error::InternalErr(format!("requiring a flow status value")))? - .to_owned(); - lparent_id = parent_id; - parent_id = r.parent_job; - let status_o = serde_json::from_value::(value).ok(); - result_id = status_o.and_then(|status| { - status - .modules - .iter() - .find(|m| m.id() == node_id) - .and_then(|m| m.job_result()) - }); - } else { - parent_id = None; - } - } else { - let q_parent = sqlx::query_scalar!( - "SELECT parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT parent_job FROM queue WHERE id = $1 AND workspace_id = $2", - parent_id.unwrap(), - w_id, - ) - .fetch_optional(&db) - .await? - .flatten(); - lparent_id = parent_id; - parent_id = q_parent; - skip_direct = false - } - } - // we could not find the node going upward from the flow by looking at all the jobs (in progress or completed) - // we now look downward from the flow root to the all the children completed job for a job that might hide itself - // in a deep non-direct parent job such as in nested branches - if result_id.is_none() && lparent_id.is_some() { - result_id = find_recursively_downward(&db, &w_id, lparent_id.unwrap(), &node_id).await?; - } + let job_result: Option = sqlx::query_scalar!( + "SELECT leaf_jobs->$1::text FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3", + node_id, + flow_id, + w_id, + ) + .fetch_optional(&db) + .await? + .flatten() + .map(|x| serde_json::from_value(x).ok()) + .flatten(); + let result_id = windmill_common::utils::not_found_if_none( - result_id, + job_result, "Flow result by id", format!("{}, {}", flow_id, node_id), )?; @@ -330,6 +251,7 @@ pub async fn push<'c>( scheduled_for_o: Option>, schedule_path: Option, parent_job: Option, + root_job: Option, is_flow_step: bool, mut same_worker: bool, pre_run_error: Option<&windmill_common::error::Error>, @@ -582,12 +504,13 @@ pub async fn push<'c>( .unwrap_or_else(|| (None, None)); let flow_status = raw_flow.as_ref().map(FlowStatus::new); + let uuid = sqlx::query_scalar!( "INSERT INTO queue (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \ - flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22) \ + flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23) \ RETURNING id", workspace_id, job_id, @@ -610,7 +533,8 @@ pub async fn push<'c>( same_worker, pre_run_error.map(|e| e.to_string()), email, - visible_to_owner + visible_to_owner, + root_job ) .fetch_one(&mut tx) .await @@ -723,6 +647,10 @@ pub struct QueuedJob { pub suspend: Option, #[serde(skip_serializing_if = "Option::is_none")] pub mem_peak: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub root_job: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub leaf_jobs: Option, } impl QueuedJob { diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index eb3c5fe2a9..8d260250d8 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -85,6 +85,7 @@ pub async fn push_scheduled_job<'c>( Some(next), Some(schedule.path.clone()), None, + None, false, false, None, diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 97a7851423..ea89e7d03c 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -315,7 +315,7 @@ async fn op_get_id(args: Vec) -> Result, anyho let client = windmill_api_client::create_client(base_url, token.clone()); let result = client - .result_by_id(workspace, flow_job_id, node_id, Some(true)) + .result_by_id(workspace, flow_job_id, node_id) .await .map_or(None, |e| Some(e.into_inner())); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 79d0f415ac..cda3d1baab 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -326,6 +326,21 @@ pub async fn update_flow_status_after_job_completion( ) .execute(&mut tx) .await?; + + if let Some(job_result) = new_status.job_result() { + sqlx::query!( + " + UPDATE queue + SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2) + WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id + ", + new_status.id(), + json!(job_result), + flow + ) + .execute(&mut tx) + .await?; + } } } @@ -1237,6 +1252,7 @@ async fn push_next_flow_job( Ok(v) => (Some(v), None), Err(e) => (None, Some(e)), }; + let root_job = flow_job.root_job.or_else(|| Some(flow_job.id)); let (uuid, inner_tx) = push( tx, &flow_job.workspace_id, @@ -1248,6 +1264,7 @@ async fn push_next_flow_job( scheduled_for_o, flow_job.schedule_path.clone(), Some(flow_job.id), + root_job, true, continue_on_same_worker, err,