diff --git a/backend/.sqlx/query-e03b8e0360ed7c282742b9b8657abdeecb3fec75bf10773544339bd025fc45bb.json b/backend/.sqlx/query-e03b8e0360ed7c282742b9b8657abdeecb3fec75bf10773544339bd025fc45bb.json new file mode 100644 index 0000000000..e17fd3f203 --- /dev/null +++ b/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" +} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9acf336ec2..ba8d5ada54 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/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, 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::().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::().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)] diff --git a/frontend/src/lib/components/flows/flowStateUtils.ts b/frontend/src/lib/components/flows/flowStateUtils.ts index b8939d73cd..6c2b9895f4 100644 --- a/frontend/src/lib/components/flows/flowStateUtils.ts +++ b/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: '' }