fix: correctly handle deeply nested results for out-of-order loops

backend/Cargo.lock

@@ -5433,6 +5433,7 @@ name = "windmill-queue"
 version = "1.117.0"
 dependencies = [
  "anyhow",
+ "async-recursion",
  "chrono",
  "chrono-tz",
  "cron",

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": [],

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

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<R: rsmq_async::RsmqConnection + Clone>(
     Ok(job)
 }
 
+#[async_recursion]
 pub async fn get_result_by_id(
     db: Pool<Postgres>,
     w_id: String,
     flow_id: Uuid,
     node_id: String,
 ) -> error::Result<serde_json::Value> {
-    let job_result: Option<JobResult> = 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() });
         }

backend/windmill-worker/src/worker_flow.rs

@@ -1375,7 +1375,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             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))