diff --git a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json index 71680b0922..8fdca62eca 100644 --- a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json +++ b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json @@ -43,7 +43,8 @@ "appdependencies", "deploymentcallback", "singlescriptflow", - "flowscript" + "flowscript", + "flownode" ] } } diff --git a/backend/.sqlx/query-113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7.json b/backend/.sqlx/query-113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7.json new file mode 100644 index 0000000000..64ec588ffa --- /dev/null +++ b/backend/.sqlx/query-113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow as \"flow!: sqlx::types::Json>\" FROM flow_node WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow!: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true + ] + }, + "hash": "113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7" +} diff --git a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json index 695607059e..20d76cbed4 100644 --- a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json +++ b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json @@ -33,7 +33,8 @@ "appdependencies", "deploymentcallback", "singlescriptflow", - "flowscript" + "flowscript", + "flownode" ] } } diff --git a/backend/.sqlx/query-3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c.json b/backend/.sqlx/query-3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c.json new file mode 100644 index 0000000000..edaec50386 --- /dev/null +++ b/backend/.sqlx/query-3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow AS \"flow!: Json>\" FROM flow_node WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow!: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true + ] + }, + "hash": "3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c" +} diff --git a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json index a3465b1c1f..4092b60b71 100644 --- a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json +++ b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json @@ -47,7 +47,8 @@ "appdependencies", "deploymentcallback", "singlescriptflow", - "flowscript" + "flowscript", + "flownode" ] } } diff --git a/backend/migrations/20241122143108_flow_node_kind.down.sql b/backend/migrations/20241122143108_flow_node_kind.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20241122143108_flow_node_kind.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20241122143108_flow_node_kind.up.sql b/backend/migrations/20241122143108_flow_node_kind.up.sql new file mode 100644 index 0000000000..0f34aff1ec --- /dev/null +++ b/backend/migrations/20241122143108_flow_node_kind.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'flownode'; diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 88ac603c19..a229450651 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1170,6 +1170,7 @@ async fn test_deno_flow(db: Pool) { continue_on_error: None, skip_if: None, }], + modules_node: None, } .into(), stop_after_if: Default::default(), @@ -1386,6 +1387,7 @@ async fn test_deno_flow_same_worker(db: Pool) { skip_if: None, }, ], + modules_node: None, }.into(), stop_after_if: Default::default(), stop_after_all_iters_if: Default::default(), diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 66560ac37b..df0562ee47 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -1257,6 +1257,7 @@ mod tests { value: windmill_common::worker::to_raw_value(&[1, 2, 3]), }, modules: vec![], + modules_node: None, skip_failures: true, parallel: false, parallelism: None, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 2ba8139f70..ad59742f88 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2430,7 +2430,7 @@ impl Job { pub fn is_flow(&self) -> bool { matches!( self.job_kind(), - JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow + JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow | JobKind::FlowNode ) } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 7b819f550a..df423eb0ae 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -413,6 +413,8 @@ pub struct Branch { #[serde(default = "default_empty_string")] pub expr: String, pub modules: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub modules_node: Option, #[serde(default = "default_true")] pub skip_failure: bool, #[serde(default = "default_true")] @@ -446,6 +448,8 @@ pub enum FlowModuleValue { ForloopFlow { iterator: InputTransform, modules: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + modules_node: Option, #[serde(default = "default_true")] skip_failures: bool, parallel: bool, @@ -454,12 +458,16 @@ pub enum FlowModuleValue { }, WhileloopFlow { modules: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + modules_node: Option, #[serde(default = "default_false")] skip_failures: bool, }, BranchOne { branches: Vec, default: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + default_node: Option, }, BranchAll { branches: Vec, @@ -537,6 +545,8 @@ struct UntaggedFlowModuleValue { concurrency_time_window_s: Option, is_trigger: Option, id: Option, + default_node: Option, + modules_node: Option, } impl<'de> Deserialize<'de> for FlowModuleValue { @@ -569,6 +579,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue { modules: untagged .modules .ok_or_else(|| serde::de::Error::missing_field("modules"))?, + modules_node: untagged.modules_node, skip_failures: untagged.skip_failures.unwrap_or(true), parallel: untagged.parallel.unwrap_or(false), parallelism: untagged.parallelism, @@ -577,6 +588,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue { modules: untagged .modules .ok_or_else(|| serde::de::Error::missing_field("modules"))?, + modules_node: untagged.modules_node, skip_failures: untagged.skip_failures.unwrap_or(false), }), "branchone" => Ok(FlowModuleValue::BranchOne { @@ -586,6 +598,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue { default: untagged .default .ok_or_else(|| serde::de::Error::missing_field("default"))?, + default_node: untagged.default_node, }), "branchall" => Ok(FlowModuleValue::BranchAll { branches: untagged @@ -757,16 +770,18 @@ pub async fn resolve_module( concurrent_limit, concurrency_time_window_s, is_trigger }; }, - ForloopFlow { modules, .. } | WhileloopFlow { modules, .. } => { - for module in modules { - Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?; + ForloopFlow { modules, modules_node, .. } | WhileloopFlow { modules, modules_node, .. } => { + resolve_modules(e, workspace_id, modules, modules_node.take(), with_code).await?; + }, + BranchOne { branches, default, default_node } => { + resolve_modules(e, workspace_id, default, default_node.take(), with_code).await?; + for branch in branches { + resolve_modules(e, workspace_id, &mut branch.modules, branch.modules_node.take(), with_code).await?; } }, - BranchOne { branches, .. } | BranchAll { branches, .. } => { + BranchAll { branches, .. } => { for branch in branches { - for module in &mut branch.modules { - Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?; - } + resolve_modules(e, workspace_id, &mut branch.modules, branch.modules_node.take(), with_code).await?; } } _ => {} @@ -774,3 +789,41 @@ pub async fn resolve_module( *value = to_raw_value(&val); Ok(()) } + +pub async fn resolve_modules( + e: &sqlx::PgPool, + workspace_id: &str, + modules: &mut Vec, + modules_node: Option, + with_code: bool, +) -> Result<(), Error> { + // Replace the `modules_node` with the actual modules. + if let Some(id) = modules_node { + *modules = load_flow_modules(e, id).await?; + } + for module in modules.iter_mut() { + Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?; + } + Ok(()) +} + +pub async fn load_flow_modules( + e: &sqlx::PgPool, + id: FlowNodeId, +) -> Result, Error> { + #[derive(Deserialize)] + struct FlowModulesOnly { modules: Vec } + + sqlx::query_scalar!( + "SELECT flow AS \"flow!: Json>\" FROM flow_node WHERE id = $1 LIMIT 1", + id.0 + ) + .fetch_one(e) + .await + .map_err(Error::SqlErr) + .and_then(|value| { + serde_json::from_str::(value.get()) + .map(|x| x.modules) + .map_err(|err| Error::InternalErr(format!("Failed to parse flow node value: {}", err))) + }) +} diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 5a8fd24fb1..5b687b54fb 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -40,6 +40,7 @@ pub enum JobKind { Noop, DeploymentCallback, FlowScript, + FlowNode, } #[derive(sqlx::FromRow, Debug, Serialize, Clone)] @@ -115,7 +116,7 @@ impl QueuedJob { pub fn is_flow(&self) -> bool { matches!( self.job_kind, - JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow + JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow | JobKind::FlowNode ) } @@ -273,6 +274,10 @@ pub enum JobPayload { cache_ttl: Option, dedicated_worker: Option, }, + FlowNode { + id: FlowNodeId, // flow_node(id). + path: String, // flow node inner path (e.g. `outer/branchall-42`). + }, Code(RawCode), Dependencies { path: String, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d6f035294f..7fa7a33b52 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2139,7 +2139,7 @@ fn fullpath_with_workspace( let path = script_path.map(String::as_str).unwrap_or("tmp/main"); let is_flow = matches!( job_kind, - &JobKind::Flow | &JobKind::FlowPreview | &JobKind::SingleScriptFlow + &JobKind::Flow | &JobKind::FlowPreview | &JobKind::SingleScriptFlow | &JobKind::FlowNode ); format!( "{}/{}/{}", @@ -3208,6 +3208,34 @@ pub async fn push<'c, 'd>( dedicated_worker, None, ), + JobPayload::FlowNode { id, path } => { + let flow_value = sqlx::query_scalar!( + "SELECT flow as \"flow!: sqlx::types::Json>\" FROM flow_node WHERE id = $1 LIMIT 1", + id.0 + ).fetch_one(_db) + .await?; + let value = serde_json::from_str::(flow_value.get()).map_err(|err| { + Error::InternalErr(format!( + "could not convert json to flow for node={}: {err:?}", id.0 + )) + })?; + let status = Some(FlowStatus::new(&value)); + ( + Some(id.0), + Some(path), + None, + JobKind::FlowNode, + Some(value), + status, + None, + None, + None, + None, + None, + None, + None, + ) + }, JobPayload::ScriptHub { path } => { if path == "hub/7771/slack" || path == "hub/7836/slack" { permissioned_as = SUPERADMIN_NOTIFICATION_EMAIL.to_string(); @@ -3965,6 +3993,7 @@ pub async fn push<'c, 'd>( JobKind::AppDependencies => "jobs.run.app_dependencies", JobKind::DeploymentCallback => "jobs.run.deployment_callback", JobKind::FlowScript => "jobs.run.flow_script", + JobKind::FlowNode => "jobs.run.flow_node", }; let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as { diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 05f23a2bc6..e68d217a36 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -329,7 +329,7 @@ async fn spawn_dedicated_workers_for_flow( .await; workers.extend(w); } - FlowModuleValue::BranchOne { branches, default } => { + FlowModuleValue::BranchOne { branches, default, .. } => { for modules in branches .iter() .map(|x| &x.modules) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index f339cd6868..147662c7e8 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -36,7 +36,7 @@ use windmill_common::db::Authed; use windmill_common::flow_status::{ ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, }; -use windmill_common::flows::add_virtual_items_if_necessary; +use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId}; use windmill_common::jobs::{ script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob, RawCode, ENTRYPOINT_OVERRIDE, @@ -2340,15 +2340,15 @@ async fn push_next_flow_job( /* Finally, push the job into the queue */ let mut uuids = vec![]; - let len = match &job_payloads { - ContinuePayload::SingleJob(_) => 1, - ContinuePayload::BranchAllJobs(payloads) => payloads.len(), - ContinuePayload::ForloopJobs { n, .. } => *n, + let job_payloads = match job_payloads { + ContinuePayload::SingleJob(payload) => vec![payload], + ContinuePayload::ParallelJobs(payloads) => payloads, }; + let len = job_payloads.len(); let mut tx = db.begin().await?; let nargs = args.as_ref(); - for i in (0..len).into_iter() { + for (i, payload_tag) in job_payloads.into_iter().enumerate() { if i % 100 == 0 && i != 0 { tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}"); sqlx::query!( @@ -2361,27 +2361,6 @@ async fn push_next_flow_job( .await?; } tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushing job {i} of {len}"); - let payload_tag = match &job_payloads { - ContinuePayload::SingleJob(payload) => payload, - ContinuePayload::BranchAllJobs(payloads) => &payloads[i], - ContinuePayload::ForloopJobs { flow_value, delete_after_use, .. } => { - let mut fv = flow_value.clone(); - - if let Some(failure_module) = fv.failure_module.as_mut() { - failure_module.id_append(&format!("{}-{i}", &status.step.to_string())); - } - &JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: fv, - path: Some(format!("{}/forloop-{i}", flow_job.script_path())), - restarted_from: None, - }, - tag: None, - delete_after_use: delete_after_use.clone(), - timeout: None, - } - } - }; // compute job-to-be-pushed priority // The job definition itself might have its own priority, but as we're running @@ -2922,8 +2901,7 @@ struct JobPayloadWithTag { } enum ContinuePayload { SingleJob(JobPayloadWithTag), - ForloopJobs { n: usize, flow_value: FlowValue, delete_after_use: bool }, - BranchAllJobs(Vec), + ParallelJobs(Vec), } enum NextFlowTransform { @@ -2950,6 +2928,36 @@ fn insert_iter_arg( args.insert(desired_key, value); } +fn payload_from_modules<'a>( + mut modules: Vec, + modules_node: Option, + failure_module: Option<&Box>, + same_worker: bool, + id: impl FnOnce() -> String, + path: impl FnOnce() -> String, +) -> Option { + if modules.is_empty() && modules_node.is_none() { + return None; + } + + if let Some(id) = modules_node { + return Some(JobPayload::FlowNode { id, path: path() }); + } + + add_virtual_items_if_necessary(&mut modules); + + let mut failure_module = failure_module.cloned(); + if let Some(failure_module) = failure_module.as_mut() { + failure_module.id_append(&id()); + } + + Some(JobPayload::RawFlow { + value: FlowValue { modules, failure_module, same_worker, ..Default::default() }, + path: Some(path()), + restarted_from: None, + }) +} + async fn compute_next_flow_transform( arc_flow_job_args: Marc>>, arc_last_job_result: Arc>, @@ -3080,7 +3088,7 @@ async fn compute_next_flow_transform( NextStatus::NextStep, )) }, - FlowModuleValue::WhileloopFlow { modules, .. } => { + FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref()); let (flow_jobs, flow_jobs_success) = match status_module { @@ -3107,6 +3115,7 @@ async fn compute_next_flow_transform( while_loop: true, }, modules, + modules_node, flow_job, is_simple, db, @@ -3116,7 +3125,7 @@ async fn compute_next_flow_transform( .await } /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */ - FlowModuleValue::ForloopFlow { modules, iterator, parallel, .. } => { + FlowModuleValue::ForloopFlow { modules, modules_node, iterator, parallel, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg let is_simple = !parallel && is_simple_modules(&modules, flow.failure_module.as_ref()); @@ -3155,6 +3164,7 @@ async fn compute_next_flow_transform( status, ns, modules, + modules_node, flow_job, is_simple, db, @@ -3164,67 +3174,61 @@ async fn compute_next_flow_transform( .await } ForLoopStatus::ParallelIteration { itered, .. } => { - if modules.is_empty() { - return Ok(NextFlowTransform::EmptyInnerFlows); - } else { - // let inner_path = Some(format!("{}/loop-parallel", flow_job.script_path(),)); - // let value = &modules[0].get_value()?; + // let inner_path = Some(format!("{}/loop-parallel", flow_job.script_path(),)); + // let value = &modules[0].get_value()?; - // we removed the is_simple_case - // if is_simple { - // let payload = - // payload_from_simple_module(value, db, flow_job, module, inner_path) - // .await?; - // ContinuePayload::ForloopJobs { n: itered.len(), payload: payload } - // } else { + // we removed the is_simple_case + // if is_simple { + // let payload = + // payload_from_simple_module(value, db, flow_job, module, inner_path) + // .await?; + // ContinuePayload::ForloopJobs { n: itered.len(), payload: payload } + // } else { - let continue_payload = { - let flow_value = FlowValue { - modules, - failure_module: flow.failure_module.clone(), - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - priority: None, - early_return: None, - concurrency_key: None, - preprocessor_module: None, + let payloads = (0..itered.len()) + .into_iter() + .filter_map(|i| { + let Some(payload) = payload_from_modules( + modules.clone(), modules_node, flow.failure_module.as_ref(), flow.same_worker, + || format!("{}-{i}", status.step), + || format!("{}/forloop-{i}", flow_job.script_path()), + ) else { + return None; }; - ContinuePayload::ForloopJobs { - n: itered.len(), - flow_value, - delete_after_use, - } - }; - Ok(NextFlowTransform::Continue( - continue_payload, - NextStatus::AllFlowJobs { - branchall: None, - iterator: Some(windmill_common::flow_status::Iterator { - index: 0, - itered, - }), - // we removed the is_simple_case for simple_input_transforms - // if is_simple { - // match value { - // FlowModuleValue::Script { input_transforms, .. } - // | FlowModuleValue::RawScript { input_transforms, .. } - // | FlowModuleValue::FlowScript { input_transforms, .. } - // | FlowModuleValue::Flow { input_transforms, .. } => { - // Some(input_transforms.clone()) - // } - // _ => None, - // } - simple_input_transforms: None, - }, - )) + Some(JobPayloadWithTag { + payload, tag: None, delete_after_use, timeout: None + }) + }) + .collect::>(); + if payloads.is_empty() { + return Ok(NextFlowTransform::EmptyInnerFlows); } + Ok(NextFlowTransform::Continue( + ContinuePayload::ParallelJobs(payloads), + NextStatus::AllFlowJobs { + branchall: None, + iterator: Some(windmill_common::flow_status::Iterator { + index: 0, + itered, + }), + // we removed the is_simple_case for simple_input_transforms + // if is_simple { + // match value { + // FlowModuleValue::Script { input_transforms, .. } + // | FlowModuleValue::RawScript { input_transforms, .. } + // | FlowModuleValue::FlowScript { input_transforms, .. } + // | FlowModuleValue::Flow { input_transforms, .. } => { + // Some(input_transforms.clone()) + // } + // _ => None, + // } + simple_input_transforms: None, + }, + )) } } } - FlowModuleValue::BranchOne { branches, default, .. } => { + FlowModuleValue::BranchOne { branches, default, default_node } => { let branch = match status_module { FlowStatusModule::WaitingForPriorSteps { .. } | FlowStatusModule::WaitingForEvents { .. } @@ -3256,51 +3260,27 @@ async fn compute_next_flow_transform( )))?, }; - let mut modules = if let BranchChosen::Branch { branch } = branch { - branches - .get(branch) - .map(|b| b.modules.clone()) - .ok_or_else(|| { - Error::BadRequest(format!( - "Unrecognized branch for BranchAll {status_module:?}" - )) - })? - } else { - default.clone() + + let (modules, modules_node) = match branch { + BranchChosen::Default => (default, default_node), + BranchChosen::Branch { branch } => branches + .into_iter() + .nth(branch) + .map(|Branch { modules, modules_node, .. }| (modules, modules_node)) + .ok_or_else(|| Error::BadRequest(format!("Unrecognized branch for BranchOne {status_module:?}")))? + }; + + let Some(payload) = payload_from_modules( + modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + || status.step.to_string(), + || format!("{}/branchone-{}", flow_job.script_path(), status.step), + ) else { + return Ok(NextFlowTransform::EmptyInnerFlows); }; - 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()); - fm = Some(failure_module); - } Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules, - failure_module: fm, - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - priority: None, - early_return: None, - concurrency_key: None, - preprocessor_module: None, - }, - path: Some(format!( - "{}/branchone-{}", - flow_job.script_path(), - status.step - )), - restarted_from: None, - }, - tag: None, - delete_after_use: delete_after_use, - timeout: None, + payload, tag: None, delete_after_use, timeout: None }), NextStatus::BranchChosen(branch), )) @@ -3313,53 +3293,30 @@ async fn compute_next_flow_transform( if branches.is_empty() { return Ok(NextFlowTransform::EmptyInnerFlows); } else if parallel { + let len = branches.len(); + let payloads = branches + .into_iter() + .enumerate() + .filter_map(|(i, Branch { modules, modules_node, .. })| { + let Some(payload) = payload_from_modules( + modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + || format!("{}-{i}", status.step), + || format!("{}/branchall-{}", flow_job.script_path(), i), + ) else { + return None; + }; + Some(JobPayloadWithTag { + payload, tag: None, delete_after_use, timeout: None + }) + }) + .collect::>(); + if payloads.is_empty() { + return Ok(NextFlowTransform::EmptyInnerFlows); + } return Ok(NextFlowTransform::Continue( - ContinuePayload::BranchAllJobs( - branches - .iter() - .enumerate() - .map(|(i, b)| { - let mut fm = flow.failure_module.clone(); - if let Some(mut failure_module) = - flow.failure_module.clone() - { - failure_module - .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, - failure_module: fm.clone(), - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - priority: None, - early_return: None, - concurrency_key: None, - preprocessor_module: None, - }, - path: Some(format!( - "{}/branchall-{}", - flow_job.script_path(), - i - )), - restarted_from: None, - }, - tag: None, - delete_after_use: delete_after_use, - timeout: None, - } - }) - .collect(), - ), + ContinuePayload::ParallelJobs(payloads), NextStatus::AllFlowJobs { - branchall: Some(BranchAllStatus { branch: 0, len: branches.len() }), + branchall: Some(BranchAllStatus { branch: 0, len }), iterator: None, simple_input_transforms: None, }, @@ -3388,47 +3345,22 @@ async fn compute_next_flow_transform( )))?, }; - let mut modules = branches - .get(branch_status.branch) - .map(|b| b.modules.clone()) - .ok_or_else(|| { - Error::BadRequest(format!( - "Unrecognized branch for BranchAll {status_module:?}" - )) - })?; + let Branch { modules, modules_node, .. } = branches + .into_iter() + .nth(branch_status.branch) + .ok_or_else(|| Error::BadRequest(format!("Unrecognized branch for BranchAll {status_module:?}")))?; + + let Some(payload) = payload_from_modules( + modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + || format!("{}-{}", status.step, branch_status.branch), + || format!("{}/branchall-{}", flow_job.script_path(), branch_status.branch), + ) else { + return Ok(NextFlowTransform::EmptyInnerFlows); + }; - 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)); - fm = Some(failure_module); - } Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules, - failure_module: fm.clone(), - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - priority: None, - early_return: None, - concurrency_key: None, - preprocessor_module: None, - }, - path: Some(format!( - "{}/branchall-{}", - flow_job.script_path(), - branch_status.branch - )), - restarted_from: None, - }, - tag: None, - delete_after_use: delete_after_use, - timeout: None, + payload, tag: None, delete_after_use, timeout: None }), NextStatus::NextBranchStep(NextBranch { status: branch_status, @@ -3444,20 +3376,15 @@ async fn next_loop_iteration( flow: &FlowValue, status: &FlowStatus, ns: ForloopNextIteration, - mut modules: Vec, + modules: Vec, + modules_node: Option, flow_job: &QueuedJob, is_simple: bool, db: &sqlx::Pool, module: &FlowModule, delete_after_use: bool, ) -> Result { - let mut fm = flow.failure_module.clone(); - if let Some(mut failure_module) = flow.failure_module.clone() { - failure_module.id_append(&format!("{}-{}", status.step, ns.index)); - fm = Some(failure_module); - } - add_virtual_items_if_necessary(&mut modules); - let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)); + let inner_path = || format!("{}/loop-{}", flow_job.script_path(), ns.index); if is_simple { let mut value = modules[0].get_value()?; let simple_input_transforms = match &mut value { @@ -3469,44 +3396,37 @@ async fn next_loop_iteration( }, _ => None, }; - let payload = payload_from_simple_module(value, db, flow_job, module, inner_path).await?; - Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(payload), + return Ok(NextFlowTransform::Continue( + ContinuePayload::SingleJob( + payload_from_simple_module(value, db, flow_job, module, Some(inner_path())).await? + ), NextStatus::NextLoopIteration { next: ns, simple_input_transforms, }, - )) - } else { - Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules, - failure_module: fm, - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - priority: None, - early_return: None, - concurrency_key: None, - preprocessor_module: None, - }, - path: inner_path, - restarted_from: None, - }, - tag: None, - delete_after_use, - timeout: None, - }), - NextStatus::NextLoopIteration { next: ns, simple_input_transforms: None }, - )) + )); } + + let Some(payload) = payload_from_modules( + modules, modules_node, flow.failure_module.as_ref(), flow.same_worker, + || format!("{}-{}", status.step, ns.index), + inner_path, + ) else { + return Ok(NextFlowTransform::EmptyInnerFlows); + }; + + Ok(NextFlowTransform::Continue( + ContinuePayload::SingleJob(JobPayloadWithTag { + payload, + tag: None, + delete_after_use, + timeout: None, + }), + NextStatus::NextLoopIteration { next: ns, simple_input_transforms: None }, + )) } -fn is_simple_modules(modules: &Vec, failure_module: Option<&Box>) -> bool { +pub(super) fn is_simple_modules(modules: &Vec, failure_module: Option<&Box>) -> bool { let is_simple = modules.len() == 1 && modules[0].is_simple() && modules[0].sleep.is_none() diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 998488537b..ef72292b5e 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -15,7 +15,7 @@ use windmill_common::scripts::ScriptHash; use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file}; use windmill_common::{ error::{self, to_anyhow}, - flows::FlowValue, + flows::{add_virtual_items_if_necessary, FlowValue}, jobs::QueuedJob, scripts::ScriptLang, DB, @@ -648,7 +648,7 @@ pub async fn handle_flow_dependency_job( // Compute a lite version of the flow value (`RawScript` => `FlowScript`). let mut value_lite = flow.clone(); - tx = reduce(tx, &mut value_lite.modules, &job_path, &job.workspace_id).await?; + tx = reduce(tx, &mut value_lite.modules, &job_path, &job.workspace_id, flow.failure_module.as_ref(), flow.same_worker).await?; sqlx::query!( "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", @@ -748,6 +748,7 @@ async fn lock_modules<'c>( FlowModuleValue::ForloopFlow { iterator, modules, + modules_node, skip_failures, parallel, parallelism, @@ -773,6 +774,7 @@ async fn lock_modules<'c>( e.value = FlowModuleValue::ForloopFlow { iterator, modules: nmodules, + modules_node, skip_failures, parallel, parallelism, @@ -808,7 +810,7 @@ async fn lock_modules<'c>( } e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }.into() } - FlowModuleValue::WhileloopFlow { modules, skip_failures } => { + FlowModuleValue::WhileloopFlow { modules, modules_node, skip_failures } => { let nmodules; (nmodules, tx, nmodified_ids) = Box::pin(lock_modules( modules, @@ -828,9 +830,9 @@ async fn lock_modules<'c>( )) .await?; e.value = - FlowModuleValue::WhileloopFlow { modules: nmodules, skip_failures }.into() + FlowModuleValue::WhileloopFlow { modules: nmodules, modules_node, skip_failures }.into() } - FlowModuleValue::BranchOne { branches, default } => { + FlowModuleValue::BranchOne { branches, default, default_node } => { let mut nbranches = vec![]; nmodified_ids = vec![]; for mut b in branches { @@ -876,7 +878,7 @@ async fn lock_modules<'c>( occupancy_metrics, )) .await?; - e.value = FlowModuleValue::BranchOne { branches: nbranches, default: ndefault } + e.value = FlowModuleValue::BranchOne { branches: nbranches, default: ndefault, default_node } .into(); } _ => (), @@ -1042,11 +1044,46 @@ async fn insert_flow_node<'c>( Ok((tx, FlowNodeId(id))) } +async fn insert_flow_modules<'c>( + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + path: &str, + workspace_id: &str, + failure_module: Option<&Box>, + same_worker: bool, + modules: &mut Vec, + modules_node: &mut Option, +) -> Result> { + tx = Box::pin(reduce(tx, modules, path, workspace_id, failure_module, same_worker)).await?; + add_virtual_items_if_necessary(modules); + if modules.is_empty() || crate::worker_flow::is_simple_modules(modules, failure_module) { + return Ok(tx); + } + let id; + (tx, id) = insert_flow_node( + tx, + path, + workspace_id, + None, + None, + Some(&Json(to_raw_value(&FlowValue { + modules: std::mem::take(modules), + failure_module: failure_module.cloned(), + same_worker, + ..Default::default() + }))) + ) + .await?; + *modules_node = Some(id); + Ok(tx) +} + async fn reduce<'c>( mut tx: sqlx::Transaction<'c, sqlx::Postgres>, modules: &mut Vec, path: &str, workspace_id: &str, + failure_module: Option<&Box>, + same_worker: bool, ) -> Result> { use FlowModuleValue::*; for module in &mut *modules { @@ -1081,12 +1118,31 @@ async fn reduce<'c>( is_trigger, }; }, - ForloopFlow { modules, .. } | WhileloopFlow { modules, .. } => { - tx = Box::pin(reduce(tx, &mut *modules, path, workspace_id)).await?; + ForloopFlow { modules, modules_node, .. } + | WhileloopFlow { modules, modules_node, .. } => { + tx = insert_flow_modules( + tx, path, workspace_id, failure_module, same_worker, + modules, modules_node + ).await?; } - BranchOne { branches, .. } | BranchAll { branches, .. } => { - for branch in &mut *branches { - tx = Box::pin(reduce(tx, &mut branch.modules, path, workspace_id)).await?; + BranchOne { branches, default, default_node, .. } => { + for branch in branches.iter_mut() { + tx = insert_flow_modules( + tx, path, workspace_id, failure_module, same_worker, + &mut branch.modules, &mut branch.modules_node + ).await?; + } + tx = insert_flow_modules( + tx, path, workspace_id, failure_module, same_worker, + default, default_node + ).await?; + } + BranchAll { branches, .. } => { + for branch in branches.iter_mut() { + tx = insert_flow_modules( + tx, path, workspace_id, failure_module, same_worker, + &mut branch.modules, &mut branch.modules_node + ).await?; } } _ => {}