From 445bf965eddc6da39a125fce60b53e0903698664 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Fri, 29 Sep 2023 00:28:11 +0200
Subject: [PATCH] fix: error handler does not recover flow anymore and error
 handler is called only once up the flow

---
 backend/windmill-worker/src/worker_flow.rs    | 21 +++++++++++++------
 .../lib/init_scripts/python_failure_module.ts |  3 ---
 frontend/src/lib/script_helpers.ts            |  3 +--
 3 files changed, 16 insertions(+), 11 deletions(-)

diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index b574425b3b..a4b855214c 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -70,6 +70,7 @@ pub async fn update_flow_status_after_job_completion<
         same_worker_tx.clone(),
         worker_dir,
         stop_early_override,
+        false,
         rsmq.clone(),
     )
     .await?;
@@ -87,6 +88,7 @@ pub async fn update_flow_status_after_job_completion<
             same_worker_tx.clone(),
             worker_dir,
             nrec.stop_early_override,
+            nrec.skip_error_handler,
             rsmq.clone(),
         )
         .await?;
@@ -99,6 +101,7 @@ pub struct RecUpdateFlowStatusAfterJobCompletion {
     success: bool,
     result: serde_json::Value,
     stop_early_override: Option<bool>,
+    skip_error_handler: bool,
 }
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion_internal<
@@ -116,9 +119,10 @@ pub async fn update_flow_status_after_job_completion_internal<
     same_worker_tx: Sender<Uuid>,
     worker_dir: &str,
     stop_early_override: Option<bool>,
+    skip_error_handler: bool,
     rsmq: Option<R>,
 ) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
-    let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult) = {
+    let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult, is_failure_step) = {
         // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}");
 
         let old_status_json = sqlx::query_scalar!(
@@ -204,8 +208,6 @@ pub async fn update_flow_status_after_job_completion_internal<
             _ => false,
         };
 
-        let skip_failure = skip_branch_failure || skip_loop_failures;
-
         let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into();
 
         let (inc_step_counter, new_status) = match module_status {
@@ -327,6 +329,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                     _ => (None, None),
                 };
                 if success || (flow_jobs.is_some() && (skip_loop_failures || skip_branch_failure)) {
+                    success = true;
                     (
                         true,
                         Some(FlowStatusModule::Success {
@@ -510,7 +513,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             _ if flow_job.canceled => false,
             true => !is_last_step,
             false if unrecoverable => false,
-            false if skip_failure => !is_last_step,
+            false if skip_branch_failure || skip_loop_failures => !is_last_step,
             false
                 if next_retry(
                     &module.and_then(|m| m.retry.clone()).unwrap_or_default(),
@@ -520,7 +523,11 @@ pub async fn update_flow_status_after_job_completion_internal<
             {
                 true
             }
-            false if has_failure_module(flow, tx.transaction_mut()).await? && !is_failure_step => {
+            false
+                if !is_failure_step
+                    && !skip_error_handler
+                    && has_failure_module(flow, tx.transaction_mut()).await? =>
+            {
                 true
             }
             false => false,
@@ -548,6 +555,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             stop_early,
             skip_if_stop_early,
             nresult,
+            is_failure_step,
         )
     };
 
@@ -584,7 +592,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             add_completed_job(
                 db,
                 &flow_job,
-                success,
+                success && !is_failure_step && !skip_error_handler,
                 stop_early && skip_if_stop_early,
                 &nresult,
                 logs,
@@ -638,6 +646,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                 } else {
                     None
                 },
+                skip_error_handler: skip_error_handler || is_failure_step,
             }));
         }
         Ok(None)
diff --git a/frontend/src/lib/init_scripts/python_failure_module.ts b/frontend/src/lib/init_scripts/python_failure_module.ts
index e29457f209..f6acb652f5 100644
--- a/frontend/src/lib/init_scripts/python_failure_module.ts
+++ b/frontend/src/lib/init_scripts/python_failure_module.ts
@@ -1,8 +1,5 @@
 export default `import os
 
-# flow is considered recovered and a success unless an exception is raised
-
-
 def main(message: str, name: str):
     flow_id = os.environ.get("WM_FLOW_JOB_ID")
     print("message", message)
diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts
index de7d800ab2..22829cf522 100644
--- a/frontend/src/lib/script_helpers.ts
+++ b/frontend/src/lib/script_helpers.ts
@@ -128,8 +128,7 @@ export async function main(x: string) {
 }
 `
 
-export const DENO_FAILURE_MODULE_CODE = `// flow is considered recovered and a success unless an exception is thrown
-
+export const DENO_FAILURE_MODULE_CODE = `
 export async function main(message: string, name: string) {
   const flow_id = Deno.env.get("WM_FLOW_JOB_ID")
   console.log("message", message)