diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 48e195610d..6221705322 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -347,3 +347,24 @@ pub struct ListFlowQuery { pub order_desc: Option, pub starred_only: Option, } + +pub fn add_virtual_items_if_necessary(modules: &mut Vec) { + 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, + }); + } +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 32568687a2..7772565760 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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::(value_json).map_err(|err| { + let mut value = serde_json::from_value::(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 diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index a7ca556261..a4ea2d0249 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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));