From e6061189438fb3a7e630d2e390075fc3eded984c Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Tue, 16 Aug 2022 23:03:55 +0200
Subject: [PATCH]  fix(backend): collecting result when for loop is not the
 last step #422

* wip: step after forloop results

Adding a failing test so I don't forget out about it.

In the last step, `items` is `4`, the last item in iteration, rather
than the collected list.  My guess is this is because the results aren't
collected unless the flow quits early or the forloop module is the last
module so that `last_step` is true.

* test

Co-authored-by: sqwishy <somebody@froghat.ca>
---
 backend/src/worker.rs      | 52 ++++++++++++++++++++++++++++++++++++++
 backend/src/worker_flow.rs | 34 ++++++++++++-------------
 2 files changed, 69 insertions(+), 17 deletions(-)

diff --git a/backend/src/worker.rs b/backend/src/worker.rs
index b740c77a36..4ed160a0e9 100644
--- a/backend/src/worker.rs
+++ b/backend/src/worker.rs
@@ -1390,6 +1390,58 @@ def main():
         assert_eq!(result, serde_json::json!("hello world"));
     }
 
+    #[sqlx::test(fixtures("base"))]
+    async fn test_step_after_loop(db: DB) {
+        initialize_tracing().await;
+
+        let flow: FlowValue = serde_json::from_value(serde_json::json!({
+            "modules": [
+                {
+                    "value": {
+                        "type": "forloopflow",
+                        "iterator": { "type": "static", "value": [2,3,4] },
+                        "value": {
+                            "modules": [
+                                {
+                                    "input_transform": {
+                                        "n": {
+                                            "type": "javascript",
+                                            "expr": "previous_result.iter.value",
+                                        },
+                                    },
+                                    "value": {
+                                        "type": "rawscript",
+                                        "language": "python3",
+                                        "content": "def main(n): return n",
+                                    },
+                                }
+                            ],
+                        }
+                    },
+                },
+                {
+                    "input_transform": {
+                        "items": {
+                            "type": "javascript",
+                            "expr": "previous_result",
+                        },
+                    },
+                    "value": {
+                        "type": "rawscript",
+                        "language": "python3",
+                        "content": "def main(items): return sum(items)",
+                    },
+                },
+            ],
+        }))
+        .unwrap();
+
+        let flow = JobPayload::RawFlow { value: flow, path: None };
+        let result = run_job_in_new_worker_until_complete(&db, flow).await;
+
+        assert_eq!(result, serde_json::json!(9));
+    }
+
     async fn run_job_in_new_worker_until_complete(db: &DB, job: JobPayload) -> serde_json::Value {
         let (uuid, tx) = push(
             db.begin().await.unwrap(),
diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs
index e3a5f4d9d6..e7eb3bb31f 100644
--- a/backend/src/worker_flow.rs
+++ b/backend/src/worker_flow.rs
@@ -147,12 +147,11 @@ pub async fn update_flow_status_after_job_completion(
             false
         };
 
-    let done = if !(success || skip_loop_failures) || last_step || stop_early {
-        let result = match new_status {
-            FlowStatusModule::Success { forloop_jobs: Some(jobs), .. } => {
-                use futures::TryStreamExt;
-                let results = sqlx::query_as(
-                    "
+    let result = match new_status {
+        FlowStatusModule::Success { forloop_jobs: Some(jobs), .. } => {
+            use futures::TryStreamExt;
+            let results = sqlx::query_as(
+                "
                   SELECT result
                     FROM completed_job
                    WHERE id = ANY($1)
@@ -160,17 +159,18 @@ pub async fn update_flow_status_after_job_completion(
                      AND success = true
                 ORDER BY args->>'_index'
                     ",
-                )
-                .bind(jobs.as_slice())
-                .bind(w_id)
-                .fetch(&mut tx)
-                .map_ok(|(v,)| v)
-                .try_collect::<Vec<serde_json::Value>>()
-                .await?;
-                serde_json::json!(results)
-            }
-            _ => result.clone(),
-        };
+            )
+            .bind(jobs.as_slice())
+            .bind(w_id)
+            .fetch(&mut tx)
+            .map_ok(|(v,)| v)
+            .try_collect::<Vec<serde_json::Value>>()
+            .await?;
+            serde_json::json!(results)
+        }
+        _ => result.clone(),
+    };
+    let done = if !(success || skip_loop_failures) || last_step || stop_early {
         tx.commit().await?;
 
         let logs = if stop_early {