From 6efee0af2004e6c755d013ad80ed3dd511667df6 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Sat, 30 Sep 2023 00:54:49 +0200
Subject: [PATCH] do not stop early if module is a flow step

---
 backend/windmill-worker/src/worker_flow.rs | 47 ++++++++++++++--------
 1 file changed, 30 insertions(+), 17 deletions(-)

diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 21d5df24d7..12c1e76049 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -169,7 +169,18 @@ pub async fn update_flow_status_after_job_completion_internal<
         let is_failure_step = old_status.step >= old_status.modules.len() as i32;
 
         let (mut stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override {
-            (true, se)
+            //do not stop early if module is a flow step
+            let mut tx = db.begin().await?;
+            let flow_job = get_queued_job(flow, w_id, &mut tx)
+                .await?
+                .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
+            tx.commit().await?;
+            let module = get_module(&flow_job, module_index);
+            if module.is_some_and(|x| matches!(x.value, FlowModuleValue::Flow { .. })) {
+                (false, false)
+            } else {
+                (true, se)
+            }
         } else if is_failure_step {
             (false, false)
         } else {
@@ -488,22 +499,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             .unwrap_or_else(|| "none".to_string());
         tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status");
 
-        let module = {
-            let raw_flow = flow_job.parse_raw_flow();
-            if let Some(raw_flow) = raw_flow {
-                if let Some(i) = module_index {
-                    if let Some(module) = raw_flow.modules.get(i) {
-                        Some(module.clone())
-                    } else {
-                        raw_flow.failure_module
-                    }
-                } else {
-                    None
-                }
-            } else {
-                None
-            }
-        };
+        let module = get_module(&flow_job, module_index);
 
         // tracing::error!(
         //     "UPDATE FLOW STATUS 3: {module:#?} {skip_failure} {is_last_step} {success}"
@@ -655,6 +651,23 @@ pub async fn update_flow_status_after_job_completion_internal<
     }
 }
 
+fn get_module(flow_job: &QueuedJob, module_index: Option<usize>) -> Option<FlowModule> {
+    let raw_flow = flow_job.parse_raw_flow();
+    if let Some(raw_flow) = raw_flow {
+        if let Some(i) = module_index {
+            if let Some(module) = raw_flow.modules.get(i) {
+                Some(module.clone())
+            } else {
+                raw_flow.failure_module
+            }
+        } else {
+            None
+        }
+    } else {
+        None
+    }
+}
+
 async fn compute_skip_loop_failures_and_parallelism(
     flow: Uuid,
     step: i32,