fix: parallel branchall disordered skip failure retrieval (#3975)

* fix: parallel branchall disordered skip failure retrieval

* fix: use regex for branchall index

* patch(frontend): default to parallel branch all

backend/.sqlx/query-e03b8e0360ed7c282742b9b8657abdeecb3fec75bf10773544339bd025fc45bb.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT script_path FROM completed_job WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "script_path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "e03b8e0360ed7c282742b9b8657abdeecb3fec75bf10773544339bd025fc45bb"
+}

backend/windmill-worker/src/worker_flow.rs

@@ -303,10 +303,18 @@ pub async fn update_flow_status_after_job_completion_internal<
         let skip_branch_failure = match module_status {
             FlowStatusModule::InProgress {
                 branchall: Some(BranchAllStatus { branch, .. }),
+                parallel,
                 ..
-            } => compute_skip_branchall_failure(flow, old_status.step, *branch, db)
-                .await?
-                .unwrap_or(false),
+            } => compute_skip_branchall_failure(
+                flow,
+                job_id_for_status,
+                old_status.step,
+                *branch,
+                *parallel,
+                db,
+            )
+            .await?
+            .unwrap_or(false),
             _ => false,
         };
 
@@ -964,17 +972,41 @@ async fn compute_skip_loop_failures_and_parallelism(
 
 async fn compute_skip_branchall_failure<'c>(
     flow: Uuid,
+    job: &Uuid,
     step: i32,
     branch: usize,
+    parallel: bool,
     db: &DB,
 ) -> Result<Option<bool>, Error> {
+    let branch = if parallel {
+        sqlx::query_scalar!("SELECT script_path FROM completed_job WHERE id = $1", job)
+            .fetch_one(db)
+            .await
+            .map_err(|e| {
+                Error::InternalErr(format!("error during retrieval of branchall index: {e:#}"))
+            })?
+            .map(|p| {
+                BRANCHALL_INDEX_RE
+                    .captures(&p)
+                    .map(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
+                    .flatten()
+                    .ok_or(Error::InternalErr(format!(
+                        "could not parse branchall index from path: {p}"
+                    )))
+            })
+            .ok_or_else(|| {
+                Error::InternalErr(format!("no branchall script path found for job {job}"))
+            })??
+    } else {
+        branch as i32
+    };
     sqlx::query_as(
         "SELECT (raw_flow->'modules'->$1->'value'->'branches'->$2->>'skip_failure')::bool
         FROM queue
         WHERE id = $3",
     )
     .bind(step)
-    .bind(branch as i32)
+    .bind(branch)
     .bind(flow)
     .fetch_one(db)
     .await
@@ -1281,6 +1313,8 @@ lazy_static::lazy_static! {
         .and_then(|x| x.parse::<usize>().ok());
 
     static ref CRASH_STEP_COUNTER: AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
+    static ref BRANCHALL_INDEX_RE: regex::Regex = regex::Regex::new(r"/branchall-(\d+)$").unwrap();
 }
 
 #[inline(always)]

frontend/src/lib/components/flows/flowStateUtils.ts

@@ -141,7 +141,8 @@ export async function createBranchAll(id: string): Promise<[FlowModule, FlowModu
 		id,
 		value: {
 			type: 'branchall',
-			branches: [{ modules: [] }]
+			branches: [{ modules: [] }],
+			parallel: true
 		},
 		summary: ''
 	}