kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: conditional retry workflow (#6461)

Browse Source 
* feat retry

* fix

* fix

* fix and nits

* fix query

* fix test

* nits and perf

* update sqlx

* once

* remove

* update sqlx
This commit is contained in:
dieriba
2025-09-03 22:24:38 +02:00
committed by GitHub
parent d400fe76c0
commit f10cac1c4b
14 changed files with 338 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/.sqlx/query-903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625.json → backend/.sqlx/query-456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47.json generated
View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job WHERE id = $1",
"query": "\n SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job\n WHERE id = $1\n ",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "903cf23d6b620388c645d5b8ac7d106bb6eea8af03e350d4ba19a4aba2cb9625"
"hash": "456835b1dfc95ebd5feb7e2b4e156d4973d17c983038c7633026d2458dfc4c47"
}

2
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json generated
View File

@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"

22
backend/.sqlx/query-b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149.json generated
View File

@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n args AS \"args: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job\n WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "args: Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "b41fa341e65ee348f468ed04ac1160770b19c0a00cd333abc48b29c54f863149"
}

1
backend/Cargo.lock generated
View File

@@ -15753,6 +15753,7 @@ name = "windmill-worker"
version = "1.538.0"
dependencies = [
"anyhow",
"async-once-cell",
"async-recursion",
"async-stream",
"backon",

2
backend/Cargo.toml
View File

@@ -367,7 +367,7 @@ aws-sdk-sts = "=1.79.0"
aws-sdk-sso = "=1.77.0"
aws-sdk-ssooidc = "=1.78.0"
rustls = "=0.23.29"
async-once-cell = "0.5.4"
aws-smithy-types-convert = { version = "^0", features = ["convert-chrono"] }

11
backend/windmill-api/src/flows.rs
View File

@@ -1653,7 +1653,8 @@ mod tests {
"exponential": {
"multiplier": 1,
"seconds": 0
}
},
"retry_if": null
}
"#
)
@@ -1668,13 +1669,15 @@ mod tests {
multiplier: 1,
seconds: 123,
random_factor: None
}
},
retry_if: None
},
serde_json::from_str(
r#"
{
"constant": {},
"exponential": { "seconds": 123 }
"exponential": { "seconds": 123 },
"retry_if" : null
}
"#
)
@@ -1692,6 +1695,7 @@ mod tests {
seconds: 3,
random_factor: None,
},
retry_if: None,
};
assert_eq!(
vec![
@@ -1718,6 +1722,7 @@ mod tests {
seconds: 3,
random_factor: None,
},
retry_if: None,
};
assert_eq!(
vec![

69
backend/windmill-common/src/flows.rs
View File

@@ -12,6 +12,7 @@ use std::{
u8,
};
use anyhow::Context;
use rand::Rng;
use serde::{Deserialize, Serialize, Serializer};
use sqlx::types::Json;
@@ -137,6 +138,65 @@ pub struct FlowValue {
pub concurrency_key: Option<String>,
}
impl FlowValue {
pub fn get_flow_module_at_step(&self, step: Step) -> anyhow::Result<&FlowModule> {
let flow_module = match step {
Step::PreprocessorStep => self
.preprocessor_module
.as_deref()
.with_context(|| format!("no preprocessor module")),
Step::Step(i) => self
.modules
.get(i)
.with_context(|| format!("no module found at index: {i}")),
Step::FailureStep => self
.failure_module
.as_deref()
.with_context(|| format!("no failure module")),
};
flow_module
}
}
#[derive(Debug, Copy, Clone)]
pub enum Step {
Step(usize),
PreprocessorStep,
FailureStep,
}
impl Step {
pub fn from_i32_and_len(step: i32, len: usize) -> Self {
if step < 0 {
Step::PreprocessorStep
} else if (step as usize) < len {
Step::Step(step as usize)
} else {
Step::FailureStep
}
}
pub fn get_step_index(&self) -> Option<usize> {
match self {
Step::Step(index) => Some(*index),
_ => None,
}
}
pub fn is_index_step(&self) -> bool {
matches!(self, Step::Step(_))
}
pub fn is_preprocessor_step(&self) -> bool {
matches!(self, Step::PreprocessorStep)
}
pub fn is_failure_step(&self) -> bool {
matches!(self, Step::FailureStep)
}
}
#[derive(Default, Deserialize, Serialize, Debug, Clone)]
pub struct StopAfterIf {
pub expr: String,
@@ -144,11 +204,18 @@ pub struct StopAfterIf {
pub error_message: Option<String>,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
pub struct RetryIf {
pub expr: String,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
#[serde(default)]
pub struct Retry {
pub constant: ConstantDelay,
pub exponential: ExponentialDelay,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry_if: Option<RetryIf>,
}
impl Retry {
@@ -156,7 +223,7 @@ impl Retry {
///
/// May return [`Duration::ZERO`] to retry immediately.
pub fn interval(&self, previous_attempts: u32, silent: bool) -> Option<Duration> {
let Self { constant, exponential } = self;
let Self { constant, exponential, .. } = self;
if previous_attempts < constant.attempts {
Some(Duration::from_secs(constant.seconds as u64))

23
backend/windmill-queue/src/flow_status.rs
View File

@@ -1,29 +1,8 @@
use uuid::Uuid;
use windmill_common::{
error::{self, Error},
utils::WarnAfterExt,
DB,
error::{self, Error}, flows::Step, utils::WarnAfterExt, DB
};
#[derive(Debug, Copy, Clone)]
pub enum Step {
Step(usize),
PreprocessorStep,
FailureStep,
}
impl Step {
pub fn from_i32_and_len(step: i32, len: usize) -> Self {
if step < 0 {
Step::PreprocessorStep
} else if (step as usize) < len {
Step::Step(step as usize)
} else {
Step::FailureStep
}
}
}
pub async fn update_flow_status_in_progress(
db: &DB,
_w_id: &str,

1
backend/windmill-worker/Cargo.toml
View File

@@ -124,6 +124,7 @@ backon.workspace = true
winapi = { workspace = true, optional = true }
pep440_rs.workspace = true
process-wrap.workspace = true
async-once-cell.workspace = true
opentelemetry = { workspace = true, optional = true }
bollard = { workspace = true, optional = true }

7
backend/windmill-worker/src/ai_executor.rs
View File

@@ -12,7 +12,7 @@ use windmill_common::{
db::DB,
error::{self, to_anyhow, Error},
flow_status::AgentAction,
flows::{FlowModule, FlowModuleValue},
flows::{FlowModule, FlowModuleValue, Step},
get_latest_hash_for_path,
jobs::JobKind,
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
@@ -21,9 +21,8 @@ use windmill_common::{
};
use windmill_parser::Typ;
use windmill_queue::{
flow_status::{get_step_of_flow_status, Step},
get_mini_pulled_job, push, CanceledBy, JobCompleted, MiniPulledJob, PushArgs,
PushIsolationLevel,
flow_status::get_step_of_flow_status, get_mini_pulled_job, push, CanceledBy, JobCompleted,
MiniPulledJob, PushArgs, PushIsolationLevel,
};
use crate::{

191
backend/windmill-worker/src/worker_flow.rs
View File

@@ -20,6 +20,7 @@ use crate::{
};
use anyhow::Context;
use async_once_cell::Lazy;
use futures::TryFutureExt;
use mappable_rc::Marc;
use serde::{Deserialize, Serialize};
@@ -56,9 +57,8 @@ use windmill_common::{
Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Step, Suspend},
};
use windmill_queue::flow_status::Step;
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
@@ -222,6 +222,7 @@ fn result_has_recover_true(nresult: Arc<Box<RawValue>>) -> bool {
let recover = serde_json::from_str::<RecoveryObject>(nresult.get());
return recover.map(|r| r.recover.unwrap_or(false)).unwrap_or(false);
}
// #[instrument(level = "trace", skip_all)]
pub async fn update_flow_status_after_job_completion_internal(
db: &DB,
@@ -344,14 +345,33 @@ pub async fn update_flow_status_after_job_completion_internal(
let is_failure_step =
old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;
let args = Arc::pin(Lazy::new(async move {
let args = sqlx::query_scalar!(
r#"
SELECT args AS "args: Json<HashMap<String, Box<RawValue>>>"
FROM v2_job
WHERE id = $1
"#,
flow
)
.fetch_one(db)
.await;
args
}));
let from_result_to_args =
|args: &Result<Option<Json<HashMap<String, Box<RawValue>>>>, sqlx::Error>| {
let args = args.as_ref().map_err(|e| {
Error::internal_err(format!("retrieval of args from state: {e:#}"))
})?;
Ok::<_, Error>(args.clone())
};
let (mut stop_early, mut stop_early_err_msg, mut skip_if_stop_early, continue_on_error) =
if let Some(se) = stop_early_override {
//do not stop early if module is a flow step
let step = match module_step {
Step::PreprocessorStep => None,
Step::FailureStep => None,
Step::Step(i) => Some(i),
};
let step = module_step.get_step_index();
let is_flow = if let Some(_) = step {
#[derive(Deserialize)]
@@ -375,7 +395,7 @@ pub async fn update_flow_status_after_job_completion_internal(
} else {
(true, None, se, false)
}
} else if is_failure_step || matches!(module_step, Step::PreprocessorStep) {
} else if is_failure_step || module_step.is_preprocessor_step() {
(false, None, false, false)
} else if let Some(current_module) = current_module {
let stop_early = success
@@ -394,18 +414,7 @@ pub async fn update_flow_status_after_job_completion_internal(
)),
_ => None,
};
let args = sqlx::query_scalar!(
"SELECT
args AS \"args: Json<HashMap<String, Box<RawValue>>>\"
FROM v2_job
WHERE id = $1",
flow
)
.fetch_one(db)
.await
.map_err(|e| {
Error::internal_err(format!("retrieval of args from state: {e:#}"))
})?;
let args = from_result_to_args(args.as_ref().await.get_ref())?;
compute_bool_from_expr(
&expr,
Marc::new(args.unwrap_or_default().0),
@@ -820,7 +829,17 @@ pub async fn update_flow_status_after_job_completion_internal(
.unwrap_or_default();
tracing::info!("update flow status on retry: {retry:#?} ");
next_retry(&retry, &old_status.retry).is_none()
let args = from_result_to_args(args.as_ref().await.get_ref())?;
evaluate_retry(
&retry,
&old_status.retry,
result.clone(),
Marc::new(args.unwrap_or_default().0),
Some(client),
)
.await?
.is_none()
} else {
false
};
@@ -901,7 +920,7 @@ pub async fn update_flow_status_after_job_completion_internal(
"error while setting flow status in failure step: {e:#}"
))
})?;
} else if matches!(module_step, Step::PreprocessorStep) {
} else if module_step.is_preprocessor_step() {
sqlx::query!(
"UPDATE v2_job_status
SET flow_status = JSONB_SET(flow_status, ARRAY['preprocessor_module'], $1)
@@ -968,16 +987,7 @@ pub async fn update_flow_status_after_job_completion_internal(
.as_ref()
.and_then(|m| m.stop_after_all_iters_if.as_ref())
{
let args = sqlx::query_scalar!(
"SELECT args AS \"args: Json<HashMap<String, Box<RawValue>>>\"
FROM v2_job WHERE id = $1",
flow
)
.fetch_one(db)
.await
.map_err(|e| {
Error::internal_err(format!("retrieval of args from state: {e:#}"))
})?;
let args = from_result_to_args(args.as_ref().await.get_ref())?;
let should_stop = compute_bool_from_expr(
&stop_after_all_iters_if.expr,
@@ -1025,7 +1035,7 @@ pub async fn update_flow_status_after_job_completion_internal(
.ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?;
tx.commit().await?;
if matches!(module_step, Step::PreprocessorStep) && success {
if module_step.is_preprocessor_step() && success {
let tag_and_concurrency_key = get_tag_and_concurrency(&flow, db).await;
let require_args = tag_and_concurrency_key.as_ref().is_some_and(|x| {
x.tag.as_ref().is_some_and(|t| t.contains("$args"))
@@ -1150,6 +1160,28 @@ pub async fn update_flow_status_after_job_completion_internal(
.map(|x| x.to_string())
.unwrap_or_else(|| "none".to_string());
let should_retry = async move || -> error::Result<bool> {
let default_retry = Retry::default();
let retry_config = flow_value
.get_flow_module_at_step(module_step)
.ok()
.and_then(|flow_module| flow_module.retry.as_ref())
.unwrap_or(&default_retry);
let args = from_result_to_args(args.as_ref().await.get_ref())?;
let should_retry = evaluate_retry(
retry_config,
&old_status.retry,
result.clone(),
Marc::new(args.unwrap_or_default().0),
Some(client),
)
.await?
.is_some();
Ok(should_retry)
};
let should_continue_flow = match success {
_ if stop_early => false,
_ if flow_job.is_canceled() => false,
@@ -1158,30 +1190,7 @@ pub async fn update_flow_status_after_job_completion_internal(
false if skip_seq_branch_failure || skip_loop_failures || continue_on_error => {
!is_last_step
}
false
if next_retry(
match module_step {
Step::PreprocessorStep => flow_value
.preprocessor_module
.as_ref()
.and_then(|m| m.retry.as_ref()),
Step::Step(i) => flow_value
.modules
.get(i)
.as_ref()
.and_then(|m| m.retry.as_ref()),
Step::FailureStep => flow_value
.failure_module
.as_ref()
.and_then(|m| m.retry.as_ref()),
}
.unwrap_or(&Retry::default()),
&old_status.retry,
)
.is_some() =>
{
true
}
false if should_retry().await? => true,
false
if !is_failure_step
&& !has_triggered_error_handler
@@ -1532,11 +1541,39 @@ async fn compute_skip_branchall_failure<'c>(
// )))
// }
fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u32, Duration)> {
(status.fail_count <= MAX_RETRY_ATTEMPTS)
.then(|| &retry)
.and_then(|retry| retry.interval(status.fail_count, false))
.map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL)))
async fn evaluate_retry(
retry: &Retry,
status: &RetryStatus,
result: Arc<Box<RawValue>>,
flow_args: Marc<HashMap<String, Box<RawValue>>>,
client: Option<&AuthedClient>,
) -> anyhow::Result<Option<(u32, Duration)>> {
if status.fail_count > MAX_RETRY_ATTEMPTS {
return Ok(None);
}
if let Some(retry_if) = &retry.retry_if {
let should_retry = compute_bool_from_expr(
&retry_if.expr,
flow_args,
result,
None,
None,
client,
None,
None,
)
.await?;
if !should_retry {
tracing::debug!("Retry condition evaluated to false, not retrying");
return Ok(None);
}
}
Ok(retry
.interval(status.fail_count, false)
.map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL))))
}
async fn compute_bool_from_expr(
@@ -1959,7 +1996,7 @@ async fn push_next_flow_job(
let arc_last_job_result = if status_module.is_failure() {
// if job is being retried, pass the result of its previous failure
last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}"))))
} else if matches!(step, Step::Step(0)) || matches!(step, Step::PreprocessorStep) {
} else if matches!(step, Step::Step(0)) || step.is_preprocessor_step() {
// if it's the first job executed in the flow, pass the flow args
Arc::new(to_raw_value(&flow_job.args))
} else {
@@ -2270,20 +2307,7 @@ async fn push_next_flow_job(
}
}
let mut module = match step {
Step::Step(i) => flow
.modules
.get(i)
.with_context(|| format!("no module at index {}", i))?,
Step::PreprocessorStep => flow
.preprocessor_module
.as_ref()
.with_context(|| format!("no preprocessor module"))?,
Step::FailureStep => flow
.failure_module
.as_deref()
.with_context(|| format!("no failure module"))?,
};
let mut module = flow.get_flow_module_at_step(step)?;
let current_id = &module.id;
let mut previous_id = match step {
@@ -2365,7 +2389,14 @@ async fn push_next_flow_job(
let retry = if matches!(&status_module, FlowStatusModule::Failure { .. },) {
let retry = &module.retry.clone().unwrap_or_default();
next_retry(retry, &status.retry)
evaluate_retry(
retry,
&status.retry,
arc_last_job_result.clone(),
arc_flow_job_args.clone(),
Some(client),
)
.await?
} else {
None
};
@@ -2497,7 +2528,7 @@ async fn push_next_flow_job(
} else {
Ok(Marc::new(HashMap::new()))
}
} else if matches!(step, Step::PreprocessorStep) {
} else if step.is_preprocessor_step() {
let mut hm = (*arc_flow_job_args).clone();
hm.insert(
ENTRYPOINT_OVERRIDE.to_string(),
@@ -2835,7 +2866,7 @@ async fn push_next_flow_job(
.map(|x| x.into());
tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
let tag = if !matches!(step, Step::PreprocessorStep)
let tag = if !step.is_preprocessor_step()
&& (flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id))
{
payload_tag.tag.clone()

2
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
View File

@@ -648,7 +648,7 @@
</Tooltip>
</div>
<div class="my-8"></div>
<FlowRetries bind:flowModuleRetry={flowModule.retry} />
<FlowRetries bind:flowModuleRetry={flowModule.retry} bind:flowModule />
</Section>
{:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'concurrency'}
<Section label="Concurrency limits" class="flex flex-col gap-4" eeOnly>

134
frontend/src/lib/components/flows/content/FlowRetries.svelte
View File

@@ -1,22 +1,64 @@
<script lang="ts">
import type { Retry } from '$lib/gen'
import type { Retry, FlowModule } from '$lib/gen'
import { SecondsInput } from '$lib/components/common'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
import { enterpriseLicense } from '$lib/stores'
import { AlertTriangle } from 'lucide-svelte'
import { untrack } from 'svelte'
import { untrack, getContext } from 'svelte'
import Toggle from '$lib/components/Toggle.svelte'
import SimpleEditor from '$lib/components/SimpleEditor.svelte'
import PropPickerWrapper from '$lib/components/flows/propPicker/PropPickerWrapper.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import Section from '$lib/components/Section.svelte'
import type { FlowEditorContext } from '../types'
import { getStepPropPicker } from '../previousResults'
import { NEVER_TESTED_THIS_FAR } from '../models'
interface Props {
flowModuleRetry: Retry | undefined
disabled?: boolean
flowModule?: FlowModule
}
let { flowModuleRetry = $bindable(), disabled = false }: Props = $props()
let {
flowModule = $bindable(),
flowModuleRetry = $bindable(),
disabled = false
}: Props = $props()
const flowEditorContext = getContext<FlowEditorContext>('FlowEditorContext')
const { flowStateStore, flowStore, previewArgs } = flowEditorContext || {
flowStateStore: null,
flowStore: null,
previewArgs: null
}
let delayType = $state() as 'disabled' | 'constant' | 'exponential' | undefined
let loaded = $state(false)
let editor: SimpleEditor | undefined = $state(undefined)
let stepPropPicker = $derived(
flowModule && flowStateStore?.val && flowStore?.val && previewArgs?.val
? getStepPropPicker(
flowStateStore.val,
undefined,
undefined,
flowModule.id,
flowStore.val,
previewArgs.val,
false
)
: null
)
let isRetryConditionEnabled = $derived(Boolean(flowModuleRetry?.retry_if))
let result = $derived(
flowModule && flowStateStore?.val
? (flowStateStore.val[flowModule.id]?.previewResult ?? NEVER_TESTED_THIS_FAR)
: NEVER_TESTED_THIS_FAR
)
function setConstantRetries() {
flowModuleRetry = {
...flowModuleRetry,
@@ -63,7 +105,7 @@
const u32Max = 4294967295
</script>
<div class="h-full flex flex-col">
<div class="h-full flex flex-col gap-4">
<ToggleButtonGroup
bind:selected={delayType}
class={`h-10 ${disabled ? 'disabled' : ''}`}
@@ -84,6 +126,90 @@
<ToggleButton light value="exponential" label="Exponential" {item} />
{/snippet}
</ToggleButtonGroup>
{#if delayType === 'constant' || delayType === 'exponential'}
<Section label="Retry Condition" class="w-full">
{#snippet header()}
<Tooltip>
Optional condition to determine when to retry. If not specified, will retry on any failure
within the configured attempt limits.
</Tooltip>
{/snippet}
<Toggle
checked={isRetryConditionEnabled}
on:change={() => {
if (!flowModuleRetry) {
return
}
if (isRetryConditionEnabled && flowModuleRetry.retry_if) {
const { retry_if, ...rest } = flowModuleRetry
flowModuleRetry = rest
} else {
flowModuleRetry = {
...flowModuleRetry,
retry_if: {
expr: 'error && error.name !== "PERMANENT_FAILURE"'
}
}
}
}}
options={{
right: 'Only retry if condition is met'
}}
/>
<div
class="w-full border p-2 mt-2 flex flex-col {flowModuleRetry?.retry_if
? ''
: 'bg-surface-secondary'}"
>
{#if flowModuleRetry?.retry_if}
<span class="mt-2 text-xs font-bold">Retry condition expression</span>
<span class="text-xs text-tertiary mb-2"
>Expression should return true to retry, false to skip retry</span
>
<div class="border w-full">
{#if stepPropPicker}
<PropPickerWrapper
notSelectable
pickableProperties={stepPropPicker.pickableProperties}
{result}
on:select={({ detail }) => {
editor?.insertAtCursor(detail)
editor?.focus()
}}
>
<SimpleEditor
bind:this={editor}
lang="javascript"
bind:code={flowModuleRetry.retry_if.expr}
class="few-lines-editor"
extraLib={`declare const result = ${JSON.stringify(result)};` +
`\ndeclare const flow_input = ${JSON.stringify(stepPropPicker.pickableProperties.flow_input || {})};`}
/>
</PropPickerWrapper>
{:else}
<SimpleEditor
bind:this={editor}
lang="javascript"
bind:code={flowModuleRetry.retry_if.expr}
class="few-lines-editor"
extraLib={`declare const result = ${JSON.stringify(result)};`}
/>
{/if}
</div>
{:else}
<span class="mt-2 text-xs font-bold">Retry condition expression</span>
<span class="text-xs text-tertiary mb-2"
>Expression should return true to retry, false to skip retry</span
>
<textarea disabled rows="3" class="min-h-[80px]"></textarea>
{/if}
</div>
</Section>
{/if}
<div class="flex h-[calc(100%-22px)]">
<div class="w-1/2 h-full overflow-auto pr-2">
{#if delayType === 'constant'}

10
openflow.openapi.yaml
View File

@@ -88,6 +88,16 @@ components:
type: integer
minimum: 0
maximum: 100
retry_if:
$ref: '#/components/schemas/RetryIf'
RetryIf:
type: object
properties:
expr:
type: string
required:
- expr
StopAfterIf:
type: object