From c210a404e03f2a857c2fabbbff2ce23189852eff Mon Sep 17 00:00:00 2001 From: dieriba Date: Mon, 8 Sep 2025 17:42:21 +0200 Subject: [PATCH] feat: timeout as expression (#6509) * done * add ctx and flow input * backward compatible * fix typo * fix --- backend/windmill-common/src/flows.rs | 102 +++++++++--- backend/windmill-worker/src/worker.rs | 2 +- backend/windmill-worker/src/worker_flow.rs | 145 +++++++++++++----- .../flows/content/FlowModuleComponent.svelte | 5 +- .../flows/content/FlowModuleSleep.svelte | 18 ++- .../flows/content/FlowModuleTimeout.svelte | 77 ++++++++-- openflow.openapi.yaml | 2 +- 7 files changed, 275 insertions(+), 76 deletions(-) diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 1be402612b..962ffd6c11 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -14,7 +14,8 @@ use std::{ use anyhow::Context; use rand::Rng; -use serde::{Deserialize, Serialize, Serializer}; +use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::value::RawValue; use sqlx::types::Json; use sqlx::types::JsonRawValue; @@ -324,7 +325,7 @@ pub struct Mock { pub struct FlowModule { #[serde(default = "default_id")] pub id: String, - pub value: Box, + pub value: Box, #[serde(skip_serializing_if = "Option::is_none")] pub stop_after_if: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -341,8 +342,12 @@ pub struct FlowModule { pub sleep: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub timeout: Option, + #[serde( + default, + deserialize_with = "raw_value_to_input_transform::<_, i32>", + skip_serializing_if = "Option::is_none" + )] + pub timeout: Option, #[serde(skip_serializing_if = "Option::is_none")] // Priority at the flow step level pub priority: Option, @@ -435,7 +440,7 @@ impl FlowModule { pub struct UntaggedInputTransform { #[serde(rename = "type")] pub type_: String, - pub value: Option>, + pub value: Option>, pub expr: Option, } @@ -446,20 +451,10 @@ impl<'de> Deserialize<'de> for InputTransform { { let untagged: UntaggedInputTransform = UntaggedInputTransform::deserialize(deserializer)?; - match untagged.type_.as_str() { - "static" => { - let value = untagged.value.unwrap_or_else(default_null); - Ok(InputTransform::Static { value }) - } - "javascript" => { - let expr = untagged.expr.unwrap_or_else(default_empty_string); - Ok(InputTransform::Javascript { expr }) - } - other => Err(serde::de::Error::unknown_variant( - other, - &["static", "javascript"], - )), - } + let input_transform = TryInto::::try_into(untagged) + .map_err(|e| serde::de::Error::custom(e))?; + + Ok(input_transform) } } @@ -471,7 +466,7 @@ impl<'de> Deserialize<'de> for InputTransform { pub enum InputTransform { Static { #[serde(default = "default_null")] - value: Box, + value: Box, }, Javascript { #[serde(default = "default_empty_string")] @@ -479,6 +474,67 @@ pub enum InputTransform { }, } +impl InputTransform { + pub fn new_static_value(value: Box) -> InputTransform { + InputTransform::Static { value } + } + + pub fn new_javascript_expr(expr: &str) -> InputTransform { + InputTransform::Javascript { expr: expr.to_owned() } + } +} + +impl TryFrom for InputTransform { + type Error = anyhow::Error; + fn try_from(value: UntaggedInputTransform) -> Result { + let input_transform = match value.type_.as_str() { + "static" => InputTransform::new_static_value(value.value.unwrap_or_else(default_null)), + "javascript" => InputTransform::new_javascript_expr(&value.expr.unwrap_or_default()), + other => { + return Err(anyhow::anyhow!( + "got value: {other} for field `type`, expected value: `static` or `javascript`" + )) + } + }; + + Ok(input_transform) + } +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum RawValueOrFormatted { + RawValue(T), + Formatted { r#type: String, value: Option, expr: Option }, +} + +fn raw_value_to_input_transform<'de, D, T>( + deserializer: D, +) -> Result, D::Error> +where + D: Deserializer<'de>, + T: DeserializeOwned + Serialize, +{ + let val = Option::>::deserialize(deserializer)?; + let input_tranform = match val { + Some(RawValueOrFormatted::RawValue(v)) => { + Some(InputTransform::new_static_value(to_raw_value(&v))) + } + Some(RawValueOrFormatted::Formatted { r#type, expr, value }) => { + let untaged_input_transform = UntaggedInputTransform { + type_: r#type, + expr, + value: value.map(|val| to_raw_value(&val)), + }; + let input_transform = TryInto::::try_into(untaged_input_transform) + .map_err(|e| serde::de::Error::custom(e))?; + Some(input_transform) + } + _ => None, + }; + Ok(input_tranform) +} + /// Id in the `flow_node` table. #[derive(Serialize, Deserialize, Debug, Copy, Clone, Hash, Eq, PartialEq)] #[serde(transparent)] @@ -751,9 +807,9 @@ impl<'de> Deserialize<'de> for FlowModuleValue { } } -impl Into> for FlowModuleValue { - fn into(self) -> Box { - crate::worker::to_raw_value(&self) +impl Into> for FlowModuleValue { + fn into(self) -> Box { + to_raw_value(&self) } } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 46522388e6..b18e39480a 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2197,7 +2197,7 @@ pub enum SendResultPayload { UpdateFlow(UpdateFlow), } -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct UpdateFlow { pub flow: Uuid, pub w_id: String, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 2c9518255d..70b72c3ecc 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1698,6 +1698,55 @@ struct FailureContext { flow_job_id: Uuid, } +#[instrument(level = "trace", skip_all)] +pub async fn evaluate_input_transform( + transform: &InputTransform, + last_result: Arc>, + flow_args: Option>>>, + authed_client: Option<&AuthedClient>, + by_id: Option<&IdContext>, +) -> error::Result +where + T: for<'de> serde::Deserialize<'de> + Send, +{ + let mut context = HashMap::with_capacity(2); + context.insert("result".to_string(), last_result.clone()); + context.insert("previous_result".to_string(), last_result.clone()); + match transform { + InputTransform::Static { value } => serde_json::from_str(value.get()).map_err(|e| { + Error::ExecutionErr(format!( + "Error parsing static value as {}: {e:#}", + std::any::type_name::() + )) + }), + InputTransform::Javascript { expr } => { + let result = eval_timeout( + expr.to_string(), + context, + flow_args, + authed_client, + by_id, + None, + ) + .warn_after_seconds(3) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Error during evaluation of expression `{expr}`:\n{e:#}" + )) + })?; + + serde_json::from_str(result.get()).map_err(|e| { + Error::ExecutionErr(format!( + "Error parsing result as {}: {e:#}. Value was: {}", + std::any::type_name::(), + result.get() + )) + }) + } + } +} + /// resumes should be in order of timestamp ascending, so that more recent are at the end #[instrument(level = "trace", skip_all)] async fn transform_input( @@ -1901,10 +1950,13 @@ lazy_static::lazy_static! { pub static ref EHM: HashMap> = HashMap::new(); } +#[derive(Debug)] enum PushNextFlowJob { Rec(PushNextFlowJobRec), Done(Option), } + +#[derive(Debug)] struct PushNextFlowJobRec { flow_job: Arc, status: FlowStatus, @@ -2143,9 +2195,10 @@ async fn push_next_flow_job( let user_auth_required = suspend.user_auth_required.unwrap_or(false); if user_auth_required { let self_approval_disabled = suspend.self_approval_disabled.unwrap_or(false); - let mut user_groups_required: Vec = Vec::new(); - if suspend.user_groups_required.is_some() { - match suspend.user_groups_required.unwrap() { + let user_groups_required: Vec; + if let Some(user_groups_required_as_input_transform) = suspend.user_groups_required + { + match user_groups_required_as_input_transform { InputTransform::Static { value } => { user_groups_required = serde_json::from_str::>(value.get()) .expect("Unable to deserialize group names"); @@ -2184,7 +2237,10 @@ async fn push_next_flow_job( } } } - } + } else { + user_groups_required = Vec::new(); + }; + let approval_conditions = ApprovalConditions { user_auth_required, user_groups_required, @@ -2408,35 +2464,16 @@ async fn push_next_flow_job( None }; - if let Some(it) = sleep_input_transform { - let json_value = match it { - InputTransform::Static { value } => Ok(value), - InputTransform::Javascript { expr } => { - let mut context = HashMap::with_capacity(2); - context.insert("result".to_string(), arc_last_job_result.clone()); - context.insert("previous_result".to_string(), arc_last_job_result.clone()); - - serde_json::from_str( - eval_timeout( - expr.to_string(), - context, - Some(arc_flow_job_args.clone()), - None, - None, - None, - ) - .warn_after_seconds(3) - .await - .map_err(|e| { - Error::ExecutionErr(format!( - "Error during isolated evaluation of expression `{expr}`:\n{e:#}" - )) - })? - .get(), - ) - } - }; - match json_value.and_then(|x| serde_json::from_str::(x.get())) { + if let Some(input_transform) = sleep_input_transform { + let timeout_value = evaluate_input_transform::( + &input_transform, + arc_last_job_result.clone(), + Some(arc_flow_job_args.clone()), + Some(client), + None, + ) + .await; + match timeout_value { Ok(serde_json::Value::Number(n)) => { if n.is_f64() { n.as_f64() @@ -2959,6 +2996,31 @@ async fn push_next_flow_job( ) }; + let evaluated_timeout = if let Some(timeout_transform) = &module.timeout { + let ctx = get_transform_context(&flow_job, &previous_id, &status) + .warn_after_seconds(3) + .await?; + + let timeout_value = evaluate_input_transform::( + timeout_transform, + arc_last_job_result.clone(), + Some(arc_flow_job_args.clone()), + Some(client), + Some(&ctx), + ) + .await?; + + if timeout_value < 0 { + return Err(Error::ExecutionErr( + "Timeout value cannot be negative".to_string(), + )); + } + + Some(timeout_value) + } else { + payload_tag.timeout + }; + let tx2 = PushIsolationLevel::Transaction(tx); let (uuid, mut inner_tx) = push( &db, @@ -2984,7 +3046,7 @@ async fn push_next_flow_job( err, flow_job.visible_to_owner, tag, - payload_tag.timeout, + evaluated_timeout, Some(module.id.clone()), new_job_priority_override, job_perms.as_ref(), @@ -3363,7 +3425,7 @@ enum NextStatus { }, } -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct JobPayloadWithTag { pub payload: JobPayload, pub tag: Option, @@ -3585,7 +3647,7 @@ async fn compute_next_flow_transform( }, tag: tag.clone(), delete_after_use, - timeout: module.timeout, + timeout: None, on_behalf_of: None, }; Ok(NextFlowTransform::Continue( @@ -4205,7 +4267,7 @@ async fn payload_from_simple_module( }, tag, delete_after_use, - timeout: module.timeout, + timeout: None, // timeout evaluation handled at higher level on_behalf_of: None, }, _ => unreachable!("is simple flow"), @@ -4239,7 +4301,7 @@ pub fn raw_script_to_payload( }), tag, delete_after_use, - timeout: module.timeout, + timeout: None, // timeout evaluation handled at higher level on_behalf_of: None, } } @@ -4341,7 +4403,12 @@ pub async fn script_to_payload( // the module value overrides the value set at the script level. Defaults to false if both are unset. let final_delete_after_user = module.delete_after_use.unwrap_or(false) || delete_after_use.unwrap_or(false); - let flow_step_timeout = module.timeout.or(script_timeout); + + let flow_step_timeout = if module.timeout.is_some() { + None + } else { + script_timeout + }; Ok(JobPayloadWithTag { payload, tag, diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 2d180631fe..67554c0efa 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -710,7 +710,10 @@ {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'timeout'}
- +
{:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'priority'}
diff --git a/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte b/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte index 14e5bd48cf..e6c967aee9 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte @@ -11,6 +11,7 @@ import { SecondsInput } from '../../common' import Section from '$lib/components/Section.svelte' import Label from '$lib/components/Label.svelte' + import { getStepPropPicker } from '../previousResults' interface Props { flowModule: FlowModule @@ -19,8 +20,8 @@ let { flowModule = $bindable(), previousModuleId }: Props = $props() - const { selectedId, flowStateStore } = getContext('FlowEditorContext') - + const { selectedId, flowStore, flowStateStore, previewArgs } = + getContext('FlowEditorContext') let schema = $state(emptySchema()) schema.properties['sleep'] = { type: 'number' @@ -28,6 +29,18 @@ let editor: SimpleEditor | undefined = $state(undefined) + let stepPropPicker = $derived( + getStepPropPicker( + flowStateStore.val, + undefined, + undefined, + flowModule.id, + flowStore.val, + previewArgs.val, + false + ) + ) + const result = flowStateStore.val[$selectedId]?.previewResult ?? {} let isSleepEnabled = $derived(Boolean(flowModule.sleep)) @@ -62,6 +75,7 @@
+ import InputTransformForm from '$lib/components/InputTransformForm.svelte' + import type SimpleEditor from '$lib/components/SimpleEditor.svelte' import Section from '$lib/components/Section.svelte' import Toggle from '$lib/components/Toggle.svelte' import Tooltip from '$lib/components/Tooltip.svelte' + import Label from '$lib/components/Label.svelte' import type { FlowModule } from '$lib/gen' import { Alert, SecondsInput } from '../../common' + import { emptySchema } from '$lib/utils' + import { getContext } from 'svelte' + import PropPickerWrapper from '$lib/components/flows/propPicker/PropPickerWrapper.svelte' + import type { FlowEditorContext } from '../types' + import { getStepPropPicker } from '../previousResults' interface Props { flowModule: FlowModule + previousModuleId: string | undefined } - let { flowModule = $bindable() }: Props = $props() + let { flowModule = $bindable(), previousModuleId }: Props = $props() + + const { flowStore, flowStateStore, previewArgs } = + getContext('FlowEditorContext') + + let schema = $state(emptySchema()) + schema.properties['timeout'] = { + type: 'number' + } + + if (typeof flowModule.timeout === 'number') { + flowModule.timeout = { + type: 'static', + value: flowModule.timeout + } + } + + let stepPropPicker = $derived( + getStepPropPicker( + flowStateStore.val, + undefined, + undefined, + flowModule.id, + flowStore.val, + previewArgs.val, + false + ) + ) + + let editor: SimpleEditor | undefined = $state(undefined) let istimeoutEnabled = $derived(Boolean(flowModule.timeout)) @@ -29,24 +67,45 @@ if (istimeoutEnabled && flowModule.timeout != undefined) { flowModule.timeout = undefined } else { - flowModule.timeout = 300 + flowModule.timeout = { + type: 'static', + value: 300 + } } }} options={{ right: 'Add a custom timeout for this step' }} /> -
- Timeout duration - - {#if flowModule.timeout} - + -
- +

The timeout will be ignored when running "Test this step"

diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 81848b83ee..0496115f09 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -134,7 +134,7 @@ components: cache_ttl: type: number timeout: - type: number + $ref: "#/components/schemas/InputTransform" delete_after_use: type: boolean summary: