kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

do not stop early if module is a flow step

Browse Source
This commit is contained in:
Ruben Fiszel
2023-09-30 00:54:49 +02:00
parent 40477b8803
commit 6efee0af20
1 changed files with 30 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
backend/windmill-worker/src/worker_flow.rs
View File

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