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

feat: while loop as new flow primitive

Browse Source
This commit is contained in:
Ruben Fiszel
2024-04-02 17:38:03 +02:00
parent 921f6a85ce
commit ff26c8e42d
17 changed files with 553 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flow_status.rs
+4
View File
@@ -137,8 +137,12 @@ pub enum FlowStatusModule {
branch_chosen: Option<BranchChosen>,
#[serde(skip_serializing_if = "Option::is_none")]
branchall: Option<BranchAllStatus>,
#[serde(skip_serializing_if = "std::ops::Not::not")]
#[serde(default = "default_false")]
parallel: bool,
#[serde(skip_serializing_if = "std::ops::Not::not")]
#[serde(default = "default_false")]
while_loop: bool,
},
Success {
id: String,
backend/windmill-common/src/flows.rs
+5
View File
@@ -316,6 +316,11 @@ pub enum FlowModuleValue {
#[serde(skip_serializing_if = "Option::is_none")]
parallelism: Option<u16>,
},
WhileloopFlow {
modules: Vec<FlowModule>,
#[serde(default = "default_false")]
skip_failures: bool,
},
BranchOne {
branches: Vec<BranchOneModules>,
default: Vec<FlowModule>,
backend/windmill-queue/src/jobs.rs
+2
View File
@@ -3519,6 +3519,7 @@ async fn restarted_flows_resolution(
len: branches.len(),
}),
parallel: parallel,
while_loop: false,
});
}
FlowModuleValue::ForloopFlow { parallel, .. } => {
@@ -3551,6 +3552,7 @@ async fn restarted_flows_resolution(
branch_chosen: None,
branchall: None,
parallel: parallel,
while_loop: false,
});
}
_ => {
backend/windmill-worker/src/worker.rs
+16
View File
@@ -1863,6 +1863,22 @@ async fn spawn_dedicated_workers_for_flow(
.await;
workers.extend(w);
}
FlowModuleValue::WhileloopFlow { modules, .. } => {
let w = spawn_dedicated_workers_for_flow(
&modules,
path,
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
)
.await;
workers.extend(w);
}
FlowModuleValue::BranchOne { branches, default } => {
for modules in branches
.iter()
backend/windmill-worker/src/worker_flow.rs
+290 -194
View File
@@ -422,8 +422,12 @@ pub async fn update_flow_status_after_job_completion_internal<
}
FlowStatusModule::InProgress {
iterator: Some(windmill_common::flow_status::Iterator { index, itered, .. }),
while_loop,
..
} if (*index + 1 < itered.len() && (success || skip_loop_failures)) && !stop_early => {
} if (*while_loop
|| (*index + 1 < itered.len()) && (success || skip_loop_failures))
&& !stop_early =>
{
(false, None)
}
FlowStatusModule::InProgress {
@@ -1919,7 +1923,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
..
} => args.as_ref().map(|args| args.clone()),
NextStatus::NextLoopIteration {
next: NextIteration { new_args, .. },
next: ForloopNextIteration { new_args, .. },
simple_input_transforms,
} => {
let mut args = if let Ok(args) = args.as_ref() {
@@ -2087,7 +2091,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
let first_uuid = uuids[0];
let new_status = match next_status {
NextStatus::NextLoopIteration {
next: NextIteration { index, itered, mut flow_jobs, .. },
next: ForloopNextIteration { index, itered, mut flow_jobs, while_loop, .. },
..
} => {
let uuid = one_uuid?;
@@ -2102,6 +2106,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
branchall: None,
id: status_module.id(),
parallel: false,
while_loop,
}
}
NextStatus::AllFlowJobs { iterator, branchall, .. } => FlowStatusModule::InProgress {
@@ -2112,6 +2117,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
branchall,
id: status_module.id(),
parallel: true,
while_loop: false,
},
NextStatus::NextBranchStep(NextBranch { mut flow_jobs, status, .. }) => {
let uuid = one_uuid?;
@@ -2125,6 +2131,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
branchall: Some(status),
id: status_module.id(),
parallel: false,
while_loop: false,
}
}
@@ -2136,6 +2143,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
branchall: None,
id: status_module.id(),
parallel: false,
while_loop: false,
},
NextStatus::NextStep => {
FlowStatusModule::WaitingForExecutor { id: status_module.id(), job: one_uuid? }
@@ -2269,16 +2277,17 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
/// Some state about the current/last forloop FlowStatusModule used to initialized the next
/// iteration's FlowStatusModule after pushing a job
#[derive(Debug)]
struct NextIteration {
struct ForloopNextIteration {
index: usize,
itered: Vec<serde_json::Value>,
flow_jobs: Vec<Uuid>,
new_args: Iter,
while_loop: bool,
}
enum LoopStatus {
enum ForLoopStatus {
ParallelIteration { itered: Vec<serde_json::Value> },
NextIteration(NextIteration),
NextIteration(ForloopNextIteration),
EmptyIterator,
}
@@ -2294,7 +2303,7 @@ enum NextStatus {
BranchChosen(BranchChosen),
NextBranchStep(NextBranch),
NextLoopIteration {
next: NextIteration,
next: ForloopNextIteration,
simple_input_transforms: Option<HashMap<String, InputTransform>>,
},
AllFlowJobs {
@@ -2429,200 +2438,74 @@ async fn compute_next_flow_transform(
NextStatus::NextStep,
))
}
FlowModuleValue::WhileloopFlow { modules, .. } => {
// 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);
let flow_jobs = match status_module {
FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), .. } => {
flow_jobs.clone()
}
_ => vec![],
};
let next_loop_idx = flow_jobs.len();
next_loop_iteration(
flow,
status,
ForloopNextIteration {
index: next_loop_idx,
itered: vec![],
flow_jobs: flow_jobs.clone(),
new_args: Iter { index: next_loop_idx as i32, value: json!(next_loop_idx) },
while_loop: true,
},
modules,
flow_job,
is_simple,
db,
module,
delete_after_use,
)
.await
}
/* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */
FlowModuleValue::ForloopFlow { modules, 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 = modules.len() == 1
&& modules[0].value.is_simple()
&& modules[0].sleep.is_none()
&& modules[0].suspend.is_none()
&& modules[0].cache_ttl.is_none()
&& (modules[0].mock.is_none()
&& modules[0].mock.as_ref().is_some_and(|m| !m.enabled)
&& flow.failure_module.is_none());
let is_simple = is_simple_modules(modules, flow);
let next_loop_status = match status_module {
FlowStatusModule::WaitingForPriorSteps { .. }
| FlowStatusModule::WaitingForEvents { .. }
| FlowStatusModule::WaitingForExecutor { .. } => {
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
};
/* Iterator is an InputTransform, evaluate it into an array. */
let itered_raw = match iterator {
InputTransform::Static { value } => to_raw_value(value),
InputTransform::Javascript { expr } => {
let mut context = HashMap::with_capacity(5);
context.insert("result".to_string(), arc_last_job_result.clone());
context.insert("previous_result".to_string(), arc_last_job_result);
context.insert("resumes".to_string(), resumes);
context.insert("resume".to_string(), resume);
context.insert("approvers".to_string(), approvers);
eval_timeout(
expr.to_string(),
context,
Some(arc_flow_job_args),
Some(client),
Some(by_id),
)
.await?
}
};
let itered = serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get())
.map_err(|not_array| {
Error::ExecutionErr(format!("Expected an array value in the iterator expression, found: {not_array}"))
})?;
if itered.is_empty() {
LoopStatus::EmptyIterator
} else if *parallel {
LoopStatus::ParallelIteration { itered }
} else if let Some(first) = itered.first() {
let iter = Iter { index: 0 as i32, value: first.to_owned() };
LoopStatus::NextIteration(NextIteration {
index: 0,
itered,
flow_jobs: vec![],
new_args: iter,
})
} else {
panic!("itered cannot be empty")
}
}
FlowStatusModule::InProgress {
iterator: Some(windmill_common::flow_status::Iterator { itered, index }),
flow_jobs: Some(flow_jobs),
..
} if !*parallel => {
let itered_new = if itered.is_empty() {
// it's possible we need to re-compute the iterator Input Transforms here, in particular if the flow is being restarted inside the loop
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
};
let itered_raw = match iterator {
InputTransform::Static { value } => to_raw_value(value),
InputTransform::Javascript { expr } => {
let mut context = HashMap::with_capacity(5);
context.insert("result".to_string(), arc_last_job_result.clone());
context.insert("previous_result".to_string(), arc_last_job_result);
context.insert("resumes".to_string(), resumes);
context.insert("resume".to_string(), resume);
context.insert("approvers".to_string(), approvers);
eval_timeout(
expr.to_string(),
context,
Some(arc_flow_job_args),
Some(client),
Some(by_id),
)
.await?
}
};
serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get()).map_err(
|not_array| {
Error::ExecutionErr(format!(
"Expected an array value, found: {not_array}"
))
},
)?
} else {
itered.clone()
};
let (index, next) = index
.checked_add(1)
.and_then(|i| itered_new.get(i).map(|next| (i, next)))
.with_context(|| {
format!("Could not find iteration number {index} restarting inside the for-loop flow. It's possible the itered-array has changed and this value isn't available anymore.")
})?;
LoopStatus::NextIteration(NextIteration {
index,
itered: itered_new.clone(),
flow_jobs: flow_jobs.clone(),
new_args: Iter { index: index as i32, value: next.to_owned() },
})
}
_ => Err(Error::BadRequest(format!(
"Unrecognized module status for ForloopFlow {status_module:?}"
)))?,
};
let next_loop_status = next_forloop_status(
status_module,
by_id,
flow_job,
previous_id,
status,
iterator,
arc_last_job_result,
resumes,
resume,
approvers,
arc_flow_job_args,
client,
parallel,
)
.await?;
match next_loop_status {
LoopStatus::EmptyIterator => Ok(NextFlowTransform::EmptyInnerFlows),
LoopStatus::NextIteration(ns) => {
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);
}
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(
&modules[0].value,
db,
flow_job,
module,
inner_path,
)
.await?;
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
NextStatus::NextLoopIteration {
next: ns,
simple_input_transforms: if is_simple {
match &modules[0].value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
Some(input_transforms.clone())
}
_ => None,
}
} else {
None
},
},
))
} 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,
},
path: inner_path,
restarted_from: None,
},
tag: None,
delete_after_use: delete_after_use,
timeout: None,
}),
NextStatus::NextLoopIteration {
next: ns,
simple_input_transforms: None,
},
))
}
ForLoopStatus::EmptyIterator => Ok(NextFlowTransform::EmptyInnerFlows),
ForLoopStatus::NextIteration(ns) => {
next_loop_iteration(
flow,
status,
ns,
modules,
flow_job,
is_simple,
db,
module,
delete_after_use,
)
.await
}
LoopStatus::ParallelIteration { itered, .. } => {
ForLoopStatus::ParallelIteration { itered, .. } => {
let inner_path = Some(format!("{}/loop-parrallel", flow_job.script_path(),));
let continue_payload = if is_simple {
let payload = payload_from_simple_module(
@@ -2882,6 +2765,219 @@ async fn compute_next_flow_transform(
}
}
async fn next_loop_iteration(
flow: &FlowValue,
status: &FlowStatus,
ns: ForloopNextIteration,
modules: &Vec<FlowModule>,
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);
}
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(&modules[0].value, db, flow_job, module, inner_path).await?;
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
NextStatus::NextLoopIteration {
next: ns,
simple_input_transforms: if is_simple {
match &modules[0].value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
Some(input_transforms.clone())
}
_ => None,
}
} else {
None
},
},
))
} 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,
},
path: inner_path,
restarted_from: None,
},
tag: None,
delete_after_use,
timeout: None,
}),
NextStatus::NextLoopIteration { next: ns, simple_input_transforms: None },
))
}
}
fn is_simple_modules(modules: &Vec<FlowModule>, flow: &FlowValue) -> bool {
let is_simple = modules.len() == 1
&& modules[0].value.is_simple()
&& modules[0].sleep.is_none()
&& modules[0].suspend.is_none()
&& modules[0].cache_ttl.is_none()
&& (modules[0].mock.is_none()
&& modules[0].mock.as_ref().is_some_and(|m| !m.enabled)
&& flow.failure_module.is_none());
is_simple
}
async fn next_forloop_status(
status_module: &FlowStatusModule,
by_id: Option<IdContext>,
flow_job: &QueuedJob,
previous_id: &str,
status: &FlowStatus,
iterator: &InputTransform,
arc_last_job_result: Arc<Box<RawValue>>,
resumes: Arc<Box<RawValue>>,
resume: Arc<Box<RawValue>>,
approvers: Arc<Box<RawValue>>,
arc_flow_job_args: Arc<HashMap<String, Box<RawValue>>>,
client: &AuthedClient,
parallel: &bool,
) -> Result<ForLoopStatus, Error> {
let next_loop_status = match status_module {
FlowStatusModule::WaitingForPriorSteps { .. }
| FlowStatusModule::WaitingForEvents { .. }
| FlowStatusModule::WaitingForExecutor { .. } => {
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
};
/* Iterator is an InputTransform, evaluate it into an array. */
let itered_raw = match iterator {
InputTransform::Static { value } => to_raw_value(value),
InputTransform::Javascript { expr } => {
let mut context = HashMap::with_capacity(5);
context.insert("result".to_string(), arc_last_job_result.clone());
context.insert("previous_result".to_string(), arc_last_job_result);
context.insert("resumes".to_string(), resumes);
context.insert("resume".to_string(), resume);
context.insert("approvers".to_string(), approvers);
eval_timeout(
expr.to_string(),
context,
Some(arc_flow_job_args),
Some(client),
Some(by_id),
)
.await?
}
};
let itered = serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get()).map_err(
|not_array| {
Error::ExecutionErr(format!(
"Expected an array value in the iterator expression, found: {not_array}"
))
},
)?;
if itered.is_empty() {
ForLoopStatus::EmptyIterator
} else if *parallel {
ForLoopStatus::ParallelIteration { itered }
} else if let Some(first) = itered.first() {
let iter = Iter { index: 0 as i32, value: first.to_owned() };
ForLoopStatus::NextIteration(ForloopNextIteration {
index: 0,
itered,
flow_jobs: vec![],
new_args: iter,
while_loop: false,
})
} else {
panic!("itered cannot be empty")
}
}
FlowStatusModule::InProgress {
iterator: Some(windmill_common::flow_status::Iterator { itered, index }),
flow_jobs: Some(flow_jobs),
..
} if !*parallel => {
let itered_new = if itered.is_empty() {
// it's possible we need to re-compute the iterator Input Transforms here, in particular if the flow is being restarted inside the loop
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(&flow_job, previous_id, &status).await?
};
let itered_raw = match iterator {
InputTransform::Static { value } => to_raw_value(value),
InputTransform::Javascript { expr } => {
let mut context = HashMap::with_capacity(5);
context.insert("result".to_string(), arc_last_job_result.clone());
context.insert("previous_result".to_string(), arc_last_job_result);
context.insert("resumes".to_string(), resumes);
context.insert("resume".to_string(), resume);
context.insert("approvers".to_string(), approvers);
eval_timeout(
expr.to_string(),
context,
Some(arc_flow_job_args),
Some(client),
Some(by_id),
)
.await?
}
};
serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get()).map_err(
|not_array| {
Error::ExecutionErr(format!("Expected an array value, found: {not_array}"))
},
)?
} else {
itered.clone()
};
let (index, next) = index
.checked_add(1)
.and_then(|i| itered_new.get(i).map(|next| (i, next)))
.with_context(|| {
format!("Could not find iteration number {index} restarting inside the for-loop flow. It's possible the itered-array has changed and this value isn't available anymore.")
})?;
ForLoopStatus::NextIteration(ForloopNextIteration {
index,
itered: itered_new.clone(),
flow_jobs: flow_jobs.clone(),
new_args: Iter { index: index as i32, value: next.to_owned() },
while_loop: false,
})
}
_ => Err(Error::BadRequest(format!(
"Unrecognized module status for ForloopFlow {status_module:?}"
)))?,
};
Ok(next_loop_status)
}
async fn payload_from_simple_module(
value: &FlowModuleValue,
db: &sqlx::Pool<sqlx::Postgres>,
frontend/src/lib/components/FlowLoopIterationPreview.svelte
+19 -2
View File
@@ -17,8 +17,9 @@
export let job: Job | undefined = undefined
export let modules: FlowModule[]
export let previewArgs: Record<string, any> = {}
export let whileLoop = false
const schema: Schema = {
export const forloopSchema: Schema = {
$schema: 'https://json-schema.org/draft/2020-12/schema' as string | undefined,
properties: {
iter: {
@@ -37,6 +38,22 @@
type: 'object'
}
export const whileLoopSchema: Schema = {
$schema: 'https://json-schema.org/draft/2020-12/schema' as string | undefined,
properties: {
iter: {
type: 'object',
properties: {
index: {
type: 'number'
}
}
}
},
required: [],
type: 'object'
}
let selectedJobStep: string | undefined = undefined
let isRunning: boolean = false
@@ -140,7 +157,7 @@
noVariablePicker
compact
class="py-4 max-w-3xl"
{schema}
schema={whileLoop ? whileLoopSchema : forloopSchema}
bind:args={previewArgs}
/>
</div>
frontend/src/lib/components/flows/content/FlowLoop.svelte
-1
View File
@@ -80,7 +80,6 @@
<Alert
type="info"
title="For loops"
tooltip="For loops"
documentationLink="https://www.windmill.dev/docs/flows/flow_loops"
class="mb-4"
size="xs"
frontend/src/lib/components/flows/content/FlowModuleWrapper.svelte
+4 -1
View File
@@ -17,6 +17,7 @@
import FlowInputsFlow from './FlowInputsFlow.svelte'
import FlowBranchesAllWrapper from './FlowBranchesAllWrapper.svelte'
import FlowBranchesOneWrapper from './FlowBranchesOneWrapper.svelte'
import FlowWhileLoop from './FlowWhileLoop.svelte'
export let flowModule: FlowModule
export let noEditor: boolean = false
@@ -66,6 +67,8 @@
{#if flowModule.id === $selectedId}
{#if flowModule.value.type === 'forloopflow'}
<FlowLoop {noEditor} bind:mod={flowModule} {parentModule} {previousModule} {enableAi} />
{:else if flowModule.value.type === 'whileloopflow'}
<FlowWhileLoop {noEditor} bind:mod={flowModule} {previousModule} />
{:else if flowModule.value.type === 'branchone'}
<FlowBranchesOneWrapper {noEditor} {previousModule} bind:flowModule {enableAi} />
{:else if flowModule.value.type === 'branchall'}
@@ -147,7 +150,7 @@
{enableAi}
/>
{/if}
{:else if flowModule.value.type === 'forloopflow'}
{:else if flowModule.value.type === 'forloopflow' || flowModule.value.type == 'whileloopflow'}
{#each flowModule.value.modules as submodule, index (index)}
<svelte:self
{noEditor}
frontend/src/lib/components/flows/content/FlowWhileLoop.svelte
+147
View File
@@ -0,0 +1,147 @@
<script lang="ts">
import { getContext } from 'svelte'
import FlowCard from '../common/FlowCard.svelte'
import type { FlowEditorContext } from '../types'
import Toggle from '$lib/components/Toggle.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte'
import FlowModuleSuspend from './FlowModuleSuspend.svelte'
// import FlowRetries from './FlowRetries.svelte'
import { Button, Drawer, Tab, TabContent, Tabs, Alert } from '$lib/components/common'
import type { FlowModule } from '$lib/gen/models/FlowModule'
import { Pane, Splitpanes } from 'svelte-splitpanes'
import { enterpriseLicense } from '$lib/stores'
import FlowModuleSleep from './FlowModuleSleep.svelte'
import FlowModuleMock from './FlowModuleMock.svelte'
import { Play } from 'lucide-svelte'
import type { Job } from '$lib/gen'
import FlowLoopIterationPreview from '$lib/components/FlowLoopIterationPreview.svelte'
import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte'
const { flowStateStore } = getContext<FlowEditorContext>('FlowEditorContext')
export let mod: FlowModule
export let previousModule: FlowModule | undefined
export let noEditor: boolean
let selected: string = 'early-stop'
let previewOpen = false
let jobId: string | undefined = undefined
let job: Job | undefined = undefined
$: previewIterationArgs = $flowStateStore[mod.id]?.previewArgs ?? {}
</script>
<Drawer bind:open={previewOpen} alwaysOpen size="75%">
<FlowLoopIterationPreview
modules={mod.value.type == 'forloopflow' ? mod.value.modules : []}
open={previewOpen}
previewArgs={previewIterationArgs}
bind:job
bind:jobId
on:close={() => {
previewOpen = false
}}
/>
</Drawer>
<div class="h-full flex flex-col">
<FlowCard {noEditor} title="For loop">
<div slot="header" class="grow">
<input bind:value={mod.summary} placeholder={'Summary'} />
</div>
<Splitpanes horizontal class="!max-h-[calc(100%-48px)]">
<Pane size={60} minSize={20} class="p-4">
{#if !noEditor}
<Alert type="info" title="While loops" class="mb-4" size="xs">
Add steps inside the while loop but have one of them use early stop/break in their
Advanced settings to break out of the while loop (otherwise it will loop forever and you
will have to cancel the flow manually)
</Alert>
{/if}
{#if mod.value.type === 'whileloopflow'}
<div class="flex flex-row gap-8 mt-2 mb-6">
<div>
<div class="mb-2 text-sm font-bold"
>Skip failures <Tooltip
documentationLink="https://www.windmill.dev/docs/flows/flow_loops"
>If disabled, the flow will fail as soon as one of the iteration fail. Otherwise,
the error will be collected as the result of the iteration. Regardless of this
setting, if an error handler is defined, it will process the error.</Tooltip
></div
>
<Toggle
bind:checked={mod.value.skip_failures}
options={{
right: 'Skip failures'
}}
/>
</div>
</div>
<div class="my-2 flex flex-row gap-2 items-center">
<div class="flex w-full justify-end">
<Button
on:click={() => (previewOpen = true)}
startIcon={{ icon: Play }}
color="dark"
size="sm">Test an iteration</Button
>
</div>
</div>
{/if}
</Pane>
<Pane size={40} minSize={20} class="flex flex-col flex-1">
<Tabs bind:selected>
<!-- <Tab value="retries">Retries</Tab> -->
<Tab value="early-stop">Early Stop/Break</Tab>
<Tab value="suspend">Suspend/Approval/Prompt</Tab>
<Tab value="sleep">Sleep</Tab>
<Tab value="mock">Mock</Tab>
<Tab value="lifetime">Lifetime</Tab>
<svelte:fragment slot="content">
<div class="overflow-hidden bg-surface" style="height:calc(100% - 32px);">
<!-- <TabContent value="retries" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowRetries bind:flowModule={mod} />
</div>
</TabContent> -->
<TabContent value="early-stop" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowModuleEarlyStop bind:flowModule={mod} />
</div>
</TabContent>
<TabContent value="suspend" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowModuleSuspend previousModuleId={previousModule?.id} bind:flowModule={mod} />
</div>
</TabContent>
<TabContent value="sleep" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowModuleSleep previousModuleId={previousModule?.id} bind:flowModule={mod} />
</div>
</TabContent>
<TabContent value="mock" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowModuleMock bind:flowModule={mod} />
</div>
</TabContent>
<TabContent value="lifetime" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowModuleDeleteAfterUse bind:flowModule={mod} disabled={!$enterpriseLicense} />
</div>
</TabContent>
</div>
</svelte:fragment>
</Tabs>
</Pane>
</Splitpanes>
</FlowCard>
</div>
frontend/src/lib/components/flows/flowExplorer.ts
+1 -1
View File
@@ -3,7 +3,7 @@ import type { FlowModule, InputTransform, OpenFlow } from '$lib/gen'
type ModuleBranches = FlowModule[][]
export function getSubModules(flowModule: FlowModule): ModuleBranches {
if (flowModule.value.type === 'forloopflow') {
if (flowModule.value.type === 'forloopflow' || flowModule.value.type === 'whileloopflow') {
return [flowModule.value.modules]
} else if (flowModule.value.type === 'branchall') {
return flowModule.value.branches.map((branch) => branch.modules)
frontend/src/lib/components/flows/flowStateUtils.ts
+14
View File
@@ -102,6 +102,20 @@ export async function createLoop(
return [loopFlowModule, flowModuleState]
}
export async function createWhileLoop(id: string): Promise<[FlowModule, FlowModuleState]> {
const loopFlowModule: FlowModule = {
id,
value: {
type: 'whileloopflow',
modules: [],
skip_failures: false
}
}
const flowModuleState = await loadFlowModuleState(loopFlowModule)
return [loopFlowModule, flowModuleState]
}
export async function createBranches(id: string): Promise<[FlowModule, FlowModuleState]> {
const branchesFlowModules: FlowModule = {
id,
frontend/src/lib/components/flows/map/FlowModuleSchemaMap.svelte
+5 -1
View File
@@ -5,6 +5,7 @@
createBranchAll,
createBranches,
createLoop,
createWhileLoop,
deleteFlowStateById,
emptyModule,
pickScript
@@ -47,6 +48,7 @@
kind:
| 'script'
| 'forloop'
| 'whileloop'
| 'branchone'
| 'branchall'
| 'flow'
@@ -65,6 +67,8 @@
module.id,
!disableAi && $copilotInfo.exists_openai_resource_path
)
} else if (kind == 'whileloop') {
;[module, state] = await createWhileLoop(module.id)
} else if (kind == 'branchone') {
;[module, state] = await createBranches(module.id)
} else if (kind == 'branchall') {
@@ -93,7 +97,7 @@
return modules
}
return modules.map((mod) => {
if (mod.value.type == 'forloopflow') {
if (mod.value.type == 'forloopflow' || mod.value.type == 'whileloopflow') {
mod.value.modules = removeAtId(mod.value.modules, id)
} else if (mod.value.type == 'branchall') {
mod.value.branches = mod.value.branches.map((branch) => {
frontend/src/lib/components/flows/map/InsertModuleButton.svelte
+12
View File
@@ -90,6 +90,18 @@
For Loop
</button>
<button
class="w-full text-left py-2 px-3 hover:bg-surface-hover whitespace-nowrap flex flex-row gap-2 items-center"
on:pointerdown={() => {
close()
dispatch('new', 'whileloop')
}}
role="menuitem"
>
<Repeat size={14} />
While Loop
</button>
<button
class="w-full text-left py-2 px-3 hover:bg-surface-hover whitespace-nowrap flex flex-row gap-2 items-center"
frontend/src/lib/components/flows/map/MapItem.svelte
+6 -6
View File
@@ -41,7 +41,7 @@
insert: {
modules: FlowModule[]
index: number
detail: 'script' | 'forloop' | 'branchone' | 'branchall' | 'move'
detail: 'script' | 'forloop' | 'whileloop' | 'branchone' | 'branchall' | 'move'
script?: { path: string; summary: string; hash: string | undefined }
}
select: string
@@ -124,13 +124,13 @@
{/if}
<div class={moving == mod.id ? 'opacity-50' : ''}>
{#if mod.value.type === 'forloopflow'}
{#if mod.value.type === 'forloopflow' || mod.value.type === 'whileloopflow'}
<FlowModuleSchemaItem
deletable={insertable}
label={mod.summary ||
`For loop ${mod.value.parallel ? '(parallel)' : ''} ${
mod.value.skip_failures ? '(skip failures)' : ''
}`}
`${mod.value.type == 'forloopflow' ? 'For' : 'While'} loop ${
mod.value.parallel ? '(parallel)' : ''
} ${mod.value.skip_failures ? '(skip failures)' : ''}`}
id={mod.id}
on:move={() => dispatch('move')}
on:delete={onDelete}
@@ -165,7 +165,7 @@
on:click={() => dispatch('select', mod.id)}
id={mod.id}
{...itemProps}
label={mod.summary || `Run all branches${ mod.value.parallel ? ' (parallel)' : ''}`}
label={mod.summary || `Run all branches${mod.value.parallel ? ' (parallel)' : ''}`}
{bgColor}
>
<div slot="icon">
frontend/src/lib/components/graph/FlowGraph.svelte
+2 -2
View File
@@ -221,7 +221,7 @@
): GraphItem | undefined {
const type = module.value.type
const parentIds = getParentIds(parent)
if (type === 'forloopflow') {
if (type === 'forloopflow' || type == 'whileloopflow') {
//@ts-ignore
return flowModuleToLoop(modules, module, parent, loopDepth)
} else if (type === 'branchone') {
@@ -358,7 +358,7 @@
function flowModuleToLoop(
modules: FlowModule[],
module: FlowModule & { value: { type: 'forloopflow' } },
module: FlowModule & { value: { type: 'forloopflow' | 'whileloopflow' } },
parent: NestedNodes | string | undefined,
loopDepth: number
): Loop {
frontend/src/lib/components/graph/svelvet/container/views/GraphView.svelte
+2
View File
@@ -141,6 +141,7 @@
<div class="zoomable" id={`zoomable-${canvasId}`}>
<!-- This is the container that holds GraphView and we have disabled right click functionality to prevent a sticking behavior -->
<div id="graphview-container">
<!-- svelte-ignore a11y-no-static-element-interactions -->
<div class={`Nodes Nodes-${canvasId}`} on:contextmenu|preventDefault>
<!-- This container is transformed by d3zoom -->
<div class={`Node Node-${canvasId}`}>
@@ -175,6 +176,7 @@
</div>
</div>
<!-- rendering dots on the background depending on the zoom level -->
<!-- svelte-ignore a11y-no-static-element-interactions -->
<svg
class={`Edges Edges-${canvasId}`}
viewBox="0 0 {$widthStore} {$heightStore}"
openflow.openapi.yaml
+24
View File
@@ -188,6 +188,7 @@ components:
- $ref: "#/components/schemas/PathScript"
- $ref: "#/components/schemas/PathFlow"
- $ref: "#/components/schemas/ForloopFlow"
- $ref: "#/components/schemas/WhileloopFlow"
- $ref: "#/components/schemas/BranchOne"
- $ref: "#/components/schemas/BranchAll"
- $ref: "#/components/schemas/Identity"
@@ -198,6 +199,7 @@ components:
script: "#/components/schemas/PathScript"
flow: "#/components/schemas/PathFlow"
forloopflow: "#/components/schemas/ForloopFlow"
whileloopflow: "#/components/schemas/WhileloopFlow"
branchone: "#/components/schemas/BranchOne"
branchall: "#/components/schemas/BranchAll"
identity: "#/components/schemas/Identity"
@@ -311,6 +313,28 @@ components:
- skip_failures
- type
WhileloopFlow:
type: object
properties:
modules:
type: array
items:
$ref: "#/components/schemas/FlowModule"
skip_failures:
type: boolean
type:
type: string
enum:
- forloopflow
parallel:
type: boolean
parallelism:
type: integer
required:
- modules
- skip_failures
- type
BranchOne:
type: object
properties: