diff --git a/backend/Cargo.lock b/backend/Cargo.lock index bb96ba44bb..a970233004 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2557,7 +2557,7 @@ dependencies = [ [[package]] name = "progenitor" version = "0.2.1-dev" -source = "git+https://github.com/oxidecomputer/progenitor#82c979df65476fe4dfc2590970ccdf64113e9e0c" +source = "git+https://github.com/oxidecomputer/progenitor#cbe875ad1a4c650cf3af595c90df6fd7421b47c2" dependencies = [ "anyhow", "built", @@ -2575,7 +2575,7 @@ dependencies = [ [[package]] name = "progenitor-client" version = "0.2.1-dev" -source = "git+https://github.com/oxidecomputer/progenitor#82c979df65476fe4dfc2590970ccdf64113e9e0c" +source = "git+https://github.com/oxidecomputer/progenitor#cbe875ad1a4c650cf3af595c90df6fd7421b47c2" dependencies = [ "bytes", "futures-core", @@ -2589,7 +2589,7 @@ dependencies = [ [[package]] name = "progenitor-impl" version = "0.2.1-dev" -source = "git+https://github.com/oxidecomputer/progenitor#82c979df65476fe4dfc2590970ccdf64113e9e0c" +source = "git+https://github.com/oxidecomputer/progenitor#cbe875ad1a4c650cf3af595c90df6fd7421b47c2" dependencies = [ "getopts", "heck", @@ -2611,7 +2611,7 @@ dependencies = [ [[package]] name = "progenitor-macro" version = "0.2.1-dev" -source = "git+https://github.com/oxidecomputer/progenitor#82c979df65476fe4dfc2590970ccdf64113e9e0c" +source = "git+https://github.com/oxidecomputer/progenitor#cbe875ad1a4c650cf3af595c90df6fd7421b47c2" dependencies = [ "openapiv3", "proc-macro2", @@ -2985,7 +2985,7 @@ dependencies = [ [[package]] name = "rustpython-ast" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#f2515405b59f8fe3b9be59d4454783c2dc2d83d1" +source = "git+https://github.com/RustPython/RustPython#dbfa61a15615a4371e1310dd4f61b007644498c1" dependencies = [ "num-bigint", "rustpython-compiler-core", @@ -2994,7 +2994,7 @@ dependencies = [ [[package]] name = "rustpython-compiler-core" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#f2515405b59f8fe3b9be59d4454783c2dc2d83d1" +source = "git+https://github.com/RustPython/RustPython#dbfa61a15615a4371e1310dd4f61b007644498c1" dependencies = [ "bincode", "bitflags", @@ -3011,7 +3011,7 @@ dependencies = [ [[package]] name = "rustpython-parser" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#f2515405b59f8fe3b9be59d4454783c2dc2d83d1" +source = "git+https://github.com/RustPython/RustPython#dbfa61a15615a4371e1310dd4f61b007644498c1" dependencies = [ "ahash", "anyhow", @@ -4281,7 +4281,7 @@ checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba" [[package]] name = "typify" version = "0.0.11-dev" -source = "git+https://github.com/oxidecomputer/typify#d579a526b3cc2e0c36d17fae8df549a03187f177" +source = "git+https://github.com/oxidecomputer/typify#05d65ea62be9061c1abd0c31d955d0248120d301" dependencies = [ "typify-impl", "typify-macro", @@ -4290,7 +4290,7 @@ dependencies = [ [[package]] name = "typify-impl" version = "0.0.11-dev" -source = "git+https://github.com/oxidecomputer/typify#d579a526b3cc2e0c36d17fae8df549a03187f177" +source = "git+https://github.com/oxidecomputer/typify#05d65ea62be9061c1abd0c31d955d0248120d301" dependencies = [ "heck", "log", @@ -4308,7 +4308,7 @@ dependencies = [ [[package]] name = "typify-macro" version = "0.0.11-dev" -source = "git+https://github.com/oxidecomputer/typify#d579a526b3cc2e0c36d17fae8df549a03187f177" +source = "git+https://github.com/oxidecomputer/typify#05d65ea62be9061c1abd0c31d955d0248120d301" dependencies = [ "proc-macro2", "quote", diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 522c9b47ad..cb33c9895f 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -550,21 +550,6 @@ }, "query": "SELECT * FROM workspace LIMIT $1 OFFSET $2" }, - "15de975d9be141c9ed9647935a508492aabbbddbf986d5c5c0f0c415293c432d": { - "describe": { - "columns": [], - "nullable": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar" - ] - } - }, - "query": "INSERT INTO variable\n (workspace_id, path, value, is_secret, description)\n VALUES ($1, 'g/all/pretty_secret', $2, true, 'This item is secret'), \n ($3, 'g/all/not_secret', $4, false, 'This item is not secret')" - }, "163f00eb8b1a489d5f382cdba22a5744e88a8e6f1532d7cb02af560f5f5d49f7": { "describe": { "columns": [ @@ -660,6 +645,27 @@ }, "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": [ @@ -1183,33 +1189,6 @@ }, "query": "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2" }, - "2be0cfd075df9624ccbcbe5fd645e0a5c25460c2d01493f86dcdd9b2b71f6181": { - "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 UNION ALL SELECT flow_status, parent_job FROM queue WHERE id = $1 AND workspace_id = $2 " - }, "2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755": { "describe": { "columns": [], @@ -3678,6 +3657,33 @@ }, "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": [], @@ -4815,26 +4821,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": [], diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 2d5223489f..51f8310334 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,7 +6,10 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{collections::HashMap, str::FromStr}; +use std::{ + collections::{HashMap, VecDeque}, + str::FromStr, +}; use anyhow::Context; use reqwest::Client; @@ -17,7 +20,7 @@ use ulid::Ulid; use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ - error::{self, Error}, + error::{self, to_anyhow, 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}, @@ -150,6 +153,41 @@ 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, @@ -159,10 +197,12 @@ pub async fn get_result_by_id( ) -> 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 ", + "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, ) @@ -174,6 +214,7 @@ pub async fn get_result_by_id( .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| { @@ -195,10 +236,17 @@ pub async fn get_result_by_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 result_id = windmill_common::utils::not_found_if_none( result_id, "Flow result by id",