diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 6b3c89db32..dadfd2b092 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -5433,6 +5433,7 @@ name = "windmill-queue" version = "1.117.0" dependencies = [ "anyhow", + "async-recursion", "chrono", "chrono-tz", "cron", diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index aa4336c763..762d8e9798 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -4403,6 +4403,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" + }, "97966407e9f1fa80fd227f75686cc9ecbb767c69ee294638c1c0957f29fd440f": { "describe": { "columns": [ @@ -4709,28 +4729,6 @@ }, "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": [], @@ -5831,6 +5829,34 @@ }, "query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n )\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms" }, + "c8ac658002423906f2a6e43e423a9e24561dfaa6b07ce89063d0910d2342a83b": { + "describe": { + "columns": [ + { + "name": "leaf_jobs", + "ordinal": 0, + "type_info": "Jsonb" + }, + { + "name": "parent_job", + "ordinal": 1, + "type_info": "Uuid" + } + ], + "nullable": [ + null, + true + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text" + ] + } + }, + "query": "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3" + }, "c9d97800eb0ec87df8e8959b283dacb2c6cce422365ed394375641488ceb6b65": { "describe": { "columns": [], diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index e34b7a2b29..f9d4a12945 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -35,3 +35,4 @@ rsmq_async.workspace = true tokio.workspace = true futures-core.workspace = true itertools.workspace = true +async-recursion.workspace = true \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ae243a73ec..9121b153ed 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -8,6 +8,7 @@ use std::collections::HashMap; +use async_recursion::async_recursion; use itertools::Itertools; use reqwest::Client; use sqlx::{Pool, Postgres, Transaction}; @@ -220,23 +221,42 @@ pub async fn pull( Ok(job) } +#[async_recursion] pub async fn get_result_by_id( db: Pool, w_id: String, flow_id: Uuid, node_id: String, ) -> error::Result { - 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", + let flow_job_result = sqlx::query!( + "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job 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(); + .await?; + + let flow_job_result = windmill_common::utils::not_found_if_none( + flow_job_result, + "Flow result by id", + format!("{}, {}", flow_id, node_id), + )?; + + let job_result = flow_job_result + .leaf_jobs + .map(|x| serde_json::from_value(x).ok()) + .flatten(); + + if job_result.is_none() && flow_job_result.parent_job.is_some() { + let parent_job = flow_job_result.parent_job.unwrap(); + let root_job = sqlx::query_scalar!("SELECT root_job FROM queue WHERE id = $1", parent_job) + .fetch_optional(&db) + .await? + .flatten() + .unwrap_or(parent_job); + return get_result_by_id(db, w_id, root_job, node_id).await; + } let result_id = windmill_common::utils::not_found_if_none( job_result, @@ -572,7 +592,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( retry: None, sleep: None, suspend: None, - cache_ttl: None + cache_ttl: None, }); raw_flow = Some(FlowValue { modules, ..flow.clone() }); } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index b3e394e235..4418cc3e95 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1375,7 +1375,10 @@ async fn push_next_flow_job Ok(v) => (Some(v), None), Err(e) => (None, Some(e)), }; - let root_job = if matches!(module.value, FlowModuleValue::Flow { .. }) { + let root_job = if matches!( + module.value, + FlowModuleValue::Flow { .. } | FlowModuleValue::ForloopFlow { parallel: true, .. } + ) { None } else { flow_job.root_job.or_else(|| Some(flow_job.id))