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

feat(backend): store flow inner modules into flow_node table (#4778)

Browse Source
This commit is contained in:
Lucas Abel
2024-11-27 16:08:31 +01:00
committed by GitHub
parent ff92e2f897
commit 2911775d0a
16 changed files with 389 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json
Generated
+2 -1
View File
@@ -43,7 +43,8 @@
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript"
"flowscript",
"flownode"
]
}
}
backend/.sqlx/query-113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow as \"flow!: sqlx::types::Json<Box<RawValue>>\" FROM flow_node WHERE id = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "flow!: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true
]
},
"hash": "113b120ae10ea4469ec3575dc3506aaa6d6a8940017a1172403ca9851d0f13a7"
}
backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json
Generated
+2 -1
View File
@@ -33,7 +33,8 @@
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript"
"flowscript",
"flownode"
]
}
}
backend/.sqlx/query-3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow AS \"flow!: Json<Box<JsonRawValue>>\" FROM flow_node WHERE id = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "flow!: Json<Box<JsonRawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true
]
},
"hash": "3d77e5b691dab38b3e39477ed980560bacd55c6ae9fafcbb0239163a3d7f3c0c"
}
backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json
Generated
+2 -1
View File
@@ -47,7 +47,8 @@
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript"
"flowscript",
"flownode"
]
}
}
backend/migrations/20241122143108_flow_node_kind.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20241122143108_flow_node_kind.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'flownode';
backend/tests/worker.rs
+2
View File
@@ -1170,6 +1170,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
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<Postgres>) {
skip_if: None,
},
],
modules_node: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
backend/windmill-api/src/flows.rs
+1
View File
@@ -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,
backend/windmill-api/src/jobs.rs
+1 -1
View File
@@ -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
)
}
backend/windmill-common/src/flows.rs
+60 -7
View File
@@ -413,6 +413,8 @@ pub struct Branch {
#[serde(default = "default_empty_string")]
pub expr: String,
pub modules: Vec<FlowModule>,
#[serde(skip_serializing_if = "Option::is_none")]
pub modules_node: Option<FlowNodeId>,
#[serde(default = "default_true")]
pub skip_failure: bool,
#[serde(default = "default_true")]
@@ -446,6 +448,8 @@ pub enum FlowModuleValue {
ForloopFlow {
iterator: InputTransform,
modules: Vec<FlowModule>,
#[serde(skip_serializing_if = "Option::is_none")]
modules_node: Option<FlowNodeId>,
#[serde(default = "default_true")]
skip_failures: bool,
parallel: bool,
@@ -454,12 +458,16 @@ pub enum FlowModuleValue {
},
WhileloopFlow {
modules: Vec<FlowModule>,
#[serde(skip_serializing_if = "Option::is_none")]
modules_node: Option<FlowNodeId>,
#[serde(default = "default_false")]
skip_failures: bool,
},
BranchOne {
branches: Vec<Branch>,
default: Vec<FlowModule>,
#[serde(skip_serializing_if = "Option::is_none")]
default_node: Option<FlowNodeId>,
},
BranchAll {
branches: Vec<Branch>,
@@ -537,6 +545,8 @@ struct UntaggedFlowModuleValue {
concurrency_time_window_s: Option<i32>,
is_trigger: Option<bool>,
id: Option<FlowNodeId>,
default_node: Option<FlowNodeId>,
modules_node: Option<FlowNodeId>,
}
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<FlowModule>,
modules_node: Option<FlowNodeId>,
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<Vec<FlowModule>, Error> {
#[derive(Deserialize)]
struct FlowModulesOnly { modules: Vec<FlowModule> }
sqlx::query_scalar!(
"SELECT flow AS \"flow!: Json<Box<JsonRawValue>>\" 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::<FlowModulesOnly>(value.get())
.map(|x| x.modules)
.map_err(|err| Error::InternalErr(format!("Failed to parse flow node value: {}", err)))
})
}
backend/windmill-common/src/jobs.rs
+6 -1
View File
@@ -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<i32>,
dedicated_worker: Option<bool>,
},
FlowNode {
id: FlowNodeId, // flow_node(id).
path: String, // flow node inner path (e.g. `outer/branchall-42`).
},
Code(RawCode),
Dependencies {
path: String,
backend/windmill-queue/src/jobs.rs
+30 -1
View File
@@ -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<Box<RawValue>>\" FROM flow_node WHERE id = $1 LIMIT 1",
id.0
).fetch_one(_db)
.await?;
let value = serde_json::from_str::<FlowValue>(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 {
backend/windmill-worker/src/dedicated_worker.rs
+1 -1
View File
@@ -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)
backend/windmill-worker/src/worker_flow.rs
+168 -248
View File
@@ -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<JobPayloadWithTag>),
ParallelJobs(Vec<JobPayloadWithTag>),
}
enum NextFlowTransform {
@@ -2950,6 +2928,36 @@ fn insert_iter_arg(
args.insert(desired_key, value);
}
fn payload_from_modules<'a>(
mut modules: Vec<FlowModule>,
modules_node: Option<FlowNodeId>,
failure_module: Option<&Box<FlowModule>>,
same_worker: bool,
id: impl FnOnce() -> String,
path: impl FnOnce() -> String,
) -> Option<JobPayload> {
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<HashMap<String, Box<RawValue>>>,
arc_last_job_result: Arc<Box<RawValue>>,
@@ -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::<Vec<_>>();
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::<Vec<_>>();
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<FlowModule>,
modules: Vec<FlowModule>,
modules_node: Option<FlowNodeId>,
flow_job: &QueuedJob,
is_simple: bool,
db: &sqlx::Pool<sqlx::Postgres>,
module: &FlowModule,
delete_after_use: bool,
) -> Result<NextFlowTransform, Error> {
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<FlowModule>, failure_module: Option<&Box<FlowModule>>) -> bool {
pub(super) fn is_simple_modules(modules: &Vec<FlowModule>, failure_module: Option<&Box<FlowModule>>) -> bool {
let is_simple = modules.len() == 1
&& modules[0].is_simple()
&& modules[0].sleep.is_none()
backend/windmill-worker/src/worker_lockfiles.rs
+67 -11
View File
@@ -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<FlowModule>>,
same_worker: bool,
modules: &mut Vec<FlowModule>,
modules_node: &mut Option<FlowNodeId>,
) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
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<FlowModule>,
path: &str,
workspace_id: &str,
failure_module: Option<&Box<FlowModule>>,
same_worker: bool,
) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
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?;
}
}
_ => {}