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

@@ -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(),

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 {