starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-20 00:02:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve suspended flow handling when missing next step

Browse Source
This commit is contained in:
Ruben Fiszel
2023-11-17 13:12:57 +01:00
parent 859cc14f67
commit f1571fd07e
3 changed files with 50 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flows.rs
+21
View File
@@ -347,3 +347,24 @@ pub struct ListFlowQuery {
pub order_desc: Option<bool>,
pub starred_only: Option<bool>,
}
pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
if modules.len() > 0
&& (modules[modules.len() - 1].sleep.is_some()
|| modules[modules.len() - 1].suspend.is_some())
{
modules.push(FlowModule {
id: format!("{}-v", modules[modules.len() - 1].id),
value: FlowModuleValue::Identity,
stop_after_if: None,
summary: Some("Virtual module needed for suspend/sleep when last module".to_string()),
mock: None,
retry: None,
sleep: None,
suspend: None,
cache_ttl: None,
timeout: None,
priority: None,
});
}
}
backend/windmill-queue/src/jobs.rs
+18 -35
View File
@@ -48,7 +48,7 @@ use windmill_common::{
BranchAllStatus, FlowStatus, FlowStatusModule, FlowStatusModuleWParent, Iterator,
JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
},
flows::{FlowModule, FlowModuleValue, FlowValue},
flows::{add_virtual_items_if_necessary, FlowModuleValue, FlowValue},
jobs::{
get_payload_tag_from_prefixed_path, script_path_to_payload, CompletedJob, JobKind,
JobPayload, QueuedJob, RawCode,
@@ -2135,7 +2135,6 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
0
};
if is_overquota || !premium_workspace {
let is_super_admin =
sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
@@ -2192,7 +2191,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
script_path,
raw_code_tuple,
job_kind,
mut raw_flow,
raw_flow,
flow_status,
language,
concurrent_limit,
@@ -2322,7 +2321,9 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
None,
None,
),
JobPayload::RawFlow { value, path, restarted_from } => {
JobPayload::RawFlow { mut value, path, restarted_from } => {
add_virtual_items_if_necessary(&mut value.modules);
let flow_status: FlowStatus = match restarted_from {
Some(restarted_from_val) => {
let (_, _, step_n, truncated_modules, _) = restarted_flows_resolution(
@@ -2382,24 +2383,30 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
tx
)?
.ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
let value = serde_json::from_value::<FlowValue>(value_json).map_err(|err| {
let mut value = serde_json::from_value::<FlowValue>(value_json).map_err(|err| {
Error::InternalErr(format!(
"could not convert json to flow for {path}: {err:?}"
))
})?;
let priority = value.priority;
add_virtual_items_if_necessary(&mut value.modules);
let cache_ttl = value.cache_ttl.map(|x| x as i32).clone();
let concurrency_time_window_s = value.concurrency_time_window_s.clone();
let concurrent_limit = value.concurrent_limit.clone();
let status = Some(FlowStatus::new(&value));
(
None,
Some(path),
None,
JobKind::Flow,
Some(value.clone()),
Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value
Some(value),
status, // this is a new flow being pushed, flow_status is set to flow_value
None,
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
dedicated_worker,
value.priority,
priority,
)
}
JobPayload::RestartedFlow { completed_job_id, step_id, branch_or_iteration_n } => {
@@ -2517,30 +2524,6 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
}
}
}
// If last module has a sleep or suspend, we insert a virtual identity module
if flow.modules.len() > 0
&& (flow.modules[flow.modules.len() - 1].sleep.is_some()
|| flow.modules[flow.modules.len() - 1].suspend.is_some())
{
let mut modules = flow.modules.clone();
modules.push(FlowModule {
id: format!("{}-v", flow.modules[flow.modules.len() - 1].id),
value: FlowModuleValue::Identity,
stop_after_if: None,
summary: Some(
"Virtual module needed for suspend/sleep when last module".to_string(),
),
mock: None,
retry: None,
sleep: None,
suspend: None,
cache_ttl: None,
timeout: None,
priority: None,
});
raw_flow = Some(FlowValue { modules, ..flow.clone() });
}
}
let (raw_code, raw_lock) = raw_code_tuple
backend/windmill-worker/src/worker_flow.rs
+11 -4
View File
@@ -29,6 +29,7 @@ use uuid::Uuid;
use windmill_common::flow_status::{
ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
};
use windmill_common::flows::add_virtual_items_if_necessary;
use windmill_common::jobs::{
script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob,
RawCode,
@@ -2305,7 +2306,8 @@ async fn compute_next_flow_transform(
failure_module.id_append(&format!("{}/{}", status.step, ns.index));
fm = Some(failure_module);
}
let modules = (*modules).clone();
let mut modules = (*modules).clone();
add_virtual_items_if_necessary(&mut modules);
let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index));
if is_simple {
let payload = payload_from_simple_module(
@@ -2451,7 +2453,7 @@ async fn compute_next_flow_transform(
)))?,
};
let modules = if let BranchChosen::Branch { branch } = branch {
let mut modules = if let BranchChosen::Branch { branch } = branch {
branches
.get(branch)
.map(|b| b.modules.clone())
@@ -2463,6 +2465,8 @@ async fn compute_next_flow_transform(
} else {
default.clone()
};
add_virtual_items_if_necessary(&mut modules);
let mut fm = flow.failure_module.clone();
if let Some(mut failure_module) = flow.failure_module.clone() {
failure_module.id_append(&status.step.to_string());
@@ -2516,10 +2520,12 @@ async fn compute_next_flow_transform(
.id_append(&format!("{}/{i}", status.step));
fm = Some(failure_module);
}
let mut modules = b.modules.clone();
add_virtual_items_if_necessary(&mut modules);
JobPayloadWithTag {
payload: JobPayload::RawFlow {
value: FlowValue {
modules: b.modules.clone(),
modules,
failure_module: fm.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
@@ -2565,7 +2571,7 @@ async fn compute_next_flow_transform(
)))?,
};
let modules = branches
let mut modules = branches
.get(branch_status.branch)
.map(|b| b.modules.clone())
.ok_or_else(|| {
@@ -2574,6 +2580,7 @@ async fn compute_next_flow_transform(
))
})?;
add_virtual_items_if_necessary(&mut modules);
let mut fm = flow.failure_module.clone();
if let Some(mut failure_module) = flow.failure_module.clone() {
failure_module.id_append(&format!("{}/{}", status.step, branch_status.branch));