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

feat: timeout as expression (#6509)

Browse Source 
* done

* add ctx and flow input

* backward compatible

* fix typo

* fix
This commit is contained in:
dieriba
2025-09-08 15:42:21 +00:00
committed by GitHub
parent 5aac5fa136
commit c210a404e0
7 changed files with 275 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flows.rs
+79 -23
View File
@@ -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<serde_json::value::RawValue>,
pub value: Box<RawValue>,
#[serde(skip_serializing_if = "Option::is_none")]
pub stop_after_if: Option<StopAfterIf>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -341,8 +342,12 @@ pub struct FlowModule {
pub sleep: Option<InputTransform>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(
default,
deserialize_with = "raw_value_to_input_transform::<_, i32>",
skip_serializing_if = "Option::is_none"
)]
pub timeout: Option<InputTransform>,
#[serde(skip_serializing_if = "Option::is_none")]
// Priority at the flow step level
pub priority: Option<i16>,
@@ -435,7 +440,7 @@ impl FlowModule {
pub struct UntaggedInputTransform {
#[serde(rename = "type")]
pub type_: String,
pub value: Option<Box<serde_json::value::RawValue>>,
pub value: Option<Box<RawValue>>,
pub expr: Option<String>,
}
@@ -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::<InputTransform>::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<serde_json::value::RawValue>,
value: Box<RawValue>,
},
Javascript {
#[serde(default = "default_empty_string")]
@@ -479,6 +474,67 @@ pub enum InputTransform {
},
}
impl InputTransform {
pub fn new_static_value(value: Box<RawValue>) -> InputTransform {
InputTransform::Static { value }
}
pub fn new_javascript_expr(expr: &str) -> InputTransform {
InputTransform::Javascript { expr: expr.to_owned() }
}
}
impl TryFrom<UntaggedInputTransform> for InputTransform {
type Error = anyhow::Error;
fn try_from(value: UntaggedInputTransform) -> Result<Self, Self::Error> {
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<T> {
RawValue(T),
Formatted { r#type: String, value: Option<T>, expr: Option<String> },
}
fn raw_value_to_input_transform<'de, D, T>(
deserializer: D,
) -> Result<Option<InputTransform>, D::Error>
where
D: Deserializer<'de>,
T: DeserializeOwned + Serialize,
{
let val = Option::<RawValueOrFormatted<T>>::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::<InputTransform>::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<Box<serde_json::value::RawValue>> for FlowModuleValue {
fn into(self) -> Box<serde_json::value::RawValue> {
crate::worker::to_raw_value(&self)
impl Into<Box<RawValue>> for FlowModuleValue {
fn into(self) -> Box<RawValue> {
to_raw_value(&self)
}
}
backend/windmill-worker/src/worker.rs
+1 -1
View File
@@ -2197,7 +2197,7 @@ pub enum SendResultPayload {
UpdateFlow(UpdateFlow),
}
#[derive(Clone)]
#[derive(Debug, Clone)]
pub struct UpdateFlow {
pub flow: Uuid,
pub w_id: String,
backend/windmill-worker/src/worker_flow.rs
+106 -39
View File
@@ -1698,6 +1698,55 @@ struct FailureContext {
flow_job_id: Uuid,
}
#[instrument(level = "trace", skip_all)]
pub async fn evaluate_input_transform<T>(
transform: &InputTransform,
last_result: Arc<Box<RawValue>>,
flow_args: Option<Marc<HashMap<String, Box<RawValue>>>>,
authed_client: Option<&AuthedClient>,
by_id: Option<&IdContext>,
) -> error::Result<T>
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::<T>()
))
}),
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::<T>(),
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<String, Box<RawValue>> = HashMap::new();
}
#[derive(Debug)]
enum PushNextFlowJob {
Rec(PushNextFlowJobRec),
Done(Option<UpdateFlow>),
}
#[derive(Debug)]
struct PushNextFlowJobRec {
flow_job: Arc<MiniPulledJob>,
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<String> = Vec::new();
if suspend.user_groups_required.is_some() {
match suspend.user_groups_required.unwrap() {
let user_groups_required: Vec<String>;
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::<Vec<String>>(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::<serde_json::Value>(x.get())) {
if let Some(input_transform) = sleep_input_transform {
let timeout_value = evaluate_input_transform::<serde_json::Value>(
&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::<i32>(
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<String>,
@@ -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,
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+4 -1
View File
@@ -710,7 +710,10 @@
</Section>
{:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'timeout'}
<div>
<FlowModuleTimeout bind:flowModule />
<FlowModuleTimeout
previousModuleId={previousModule?.id}
bind:flowModule
/>
</div>
{:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'priority'}
<Section label="Priority" class="flex flex-col gap-4">
frontend/src/lib/components/flows/content/FlowModuleSleep.svelte
+16 -2
View File
@@ -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>('FlowEditorContext')
const { selectedId, flowStore, flowStateStore, previewArgs } =
getContext<FlowEditorContext>('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 @@
<div class="border">
<PropPickerWrapper
noFlowPlugConnect={true}
flow_input={stepPropPicker.pickableProperties.flow_input}
notSelectable
{result}
displayContext={false}
frontend/src/lib/components/flows/content/FlowModuleTimeout.svelte
+68 -9
View File
@@ -1,16 +1,54 @@
<script lang="ts">
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>('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))
</script>
@@ -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'
}}
/>
<div class="mb-4">
<span class="text-xs font-bold">Timeout duration</span>
{#if flowModule.timeout}
<SecondsInput bind:seconds={flowModule.timeout} />
<Label label="Timeout duration">
{#if flowModule.timeout && schema.properties['timeout']}
<div class="border">
<PropPickerWrapper
flow_input={stepPropPicker.pickableProperties.flow_input}
notSelectable
pickableProperties={stepPropPicker.pickableProperties}
on:select={({ detail }) => {
editor?.insertAtCursor(detail)
editor?.focus()
}}
>
<InputTransformForm
bind:arg={flowModule.timeout}
argName="timeout"
{schema}
{previousModuleId}
argExtra={{ seconds: true }}
bind:editor
/>
</PropPickerWrapper>
</div>
{:else}
<SecondsInput disabled />
<div class="text-secondary">OR use a dynamic expression</div>
{/if}
</Label>
<div class="mt-4"></div>
<div class="mt-4">
<Alert title="Only used when testing the full flow" type="info">
<p class="text-sm"> The timeout will be ignored when running "Test this step" </p>
</Alert>
openflow.openapi.yaml
+1 -1
View File
@@ -134,7 +134,7 @@ components:
cache_ttl:
type: number
timeout:
type: number
$ref: "#/components/schemas/InputTransform"
delete_after_use:
type: boolean
summary: