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

improve retries max value handling

Browse Source
This commit is contained in:
Ruben Fiszel
2024-12-06 01:03:46 +01:00
parent 2230c729d5
commit b4011be2dc
4 changed files with 105 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flow_status.rs
+2 -2
View File
@@ -17,7 +17,7 @@ use crate::flows::FlowValue;
const MINUTES: Duration = Duration::from_secs(60);
const HOURS: Duration = MINUTES.saturating_mul(60);
pub const MAX_RETRY_ATTEMPTS: u16 = 1000;
pub const MAX_RETRY_ATTEMPTS: u32 = u32::MAX;
pub const MAX_RETRY_INTERVAL: Duration = HOURS.saturating_mul(6);
pub fn is_retry_default(v: &RetryStatus) -> bool {
@@ -48,7 +48,7 @@ pub struct FlowStatus {
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
#[serde(default)]
pub struct RetryStatus {
pub fail_count: u16,
pub fail_count: u32,
pub failed_jobs: Vec<Uuid>,
}
backend/windmill-common/src/flows.rs
+69 -23
View File
@@ -147,7 +147,7 @@ impl Retry {
/// Takes the number of previous retries and returns the interval until the next retry if any.
///
/// May return [`Duration::ZERO`] to retry immediately.
pub fn interval(&self, previous_attempts: u16, silent: bool) -> Option<Duration> {
pub fn interval(&self, previous_attempts: u32, silent: bool) -> Option<Duration> {
let Self { constant, exponential } = self;
if previous_attempts < constant.attempts {
@@ -178,7 +178,7 @@ impl Retry {
self.constant.attempts != 0 || self.exponential.attempts != 0
}
pub fn max_attempts(&self) -> u16 {
pub fn max_attempts(&self) -> u32 {
self.constant
.attempts
.saturating_add(self.exponential.attempts)
@@ -194,7 +194,7 @@ impl Retry {
#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
#[serde(default)]
pub struct ConstantDelay {
pub attempts: u16,
pub attempts: u32,
pub seconds: u16,
}
@@ -202,7 +202,7 @@ pub struct ConstantDelay {
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
#[serde(default)]
pub struct ExponentialDelay {
pub attempts: u16,
pub attempts: u32,
pub multiplier: u16,
pub seconds: u16,
pub random_factor: Option<i8>, // percentage, defaults to 0 for no jitter
@@ -713,10 +713,14 @@ pub async fn resolve_maybe_value<T>(
workspace_id: &str,
with_code: bool,
maybe: Option<T>,
value_mut: impl FnOnce(&mut T) -> Option<&mut Json<Box<JsonRawValue>>>
value_mut: impl FnOnce(&mut T) -> Option<&mut Json<Box<JsonRawValue>>>,
) -> Result<Option<T>, Error> {
let Some(mut container) = maybe else { return Ok(None); };
let Some(value) = value_mut(&mut container) else { return Ok(Some(container)); };
let Some(mut container) = maybe else {
return Ok(None);
};
let Some(value) = value_mut(&mut container) else {
return Ok(Some(container));
};
resolve_value(e, workspace_id, &mut value.0, with_code).await?;
Ok(Some(container))
}
@@ -728,8 +732,9 @@ pub async fn resolve_value(
value: &mut Box<JsonRawValue>,
with_code: bool,
) -> Result<(), Error> {
let mut val = serde_json::from_str::<FlowValue>(value.get())
.map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow value: {}", err)))?;
let mut val = serde_json::from_str::<FlowValue>(value.get()).map_err(|err| {
Error::InternalErr(format!("resolve: Failed to parse flow value: {}", err))
})?;
for module in &mut val.modules {
resolve_module(e, workspace_id, &mut module.value, with_code).await?;
}
@@ -746,16 +751,29 @@ pub async fn resolve_module(
) -> Result<(), Error> {
use FlowModuleValue::*;
let mut val = serde_json::from_str::<FlowModuleValue>(value.get())
.map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow module value: {}", err)))?;
let mut val = serde_json::from_str::<FlowModuleValue>(value.get()).map_err(|err| {
Error::InternalErr(format!(
"resolve: Failed to parse flow module value: {}",
err
))
})?;
match &mut val {
FlowScript { .. } => {
// In order to avoid an unnecessary `.clone()` of `val`, take ownership of it's content
// using `std::mem::replace`.
let FlowScript {
input_transforms, id, tag, language,
custom_concurrency_key, concurrent_limit, concurrency_time_window_s, is_trigger
} = std::mem::replace(&mut val, Identity) else { unreachable!() };
input_transforms,
id,
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
} = std::mem::replace(&mut val, Identity)
else {
unreachable!()
};
// Load script lock file and code content.
let (lock, content) = if !with_code {
(Some("...".to_string()), "...".to_string())
@@ -763,22 +781,44 @@ pub async fn resolve_module(
cache::flow::fetch_script(e, id).await?
};
val = RawScript {
input_transforms, content, lock, path: None, tag, language, custom_concurrency_key,
concurrent_limit, concurrency_time_window_s, is_trigger
input_transforms,
content,
lock,
path: None,
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
};
},
ForloopFlow { modules, modules_node, .. } | WhileloopFlow { modules, modules_node, .. } => {
}
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?;
resolve_modules(
e,
workspace_id,
&mut branch.modules,
branch.modules_node.take(),
with_code,
)
.await?;
}
},
}
BranchAll { branches, .. } => {
for branch in branches {
resolve_modules(e, workspace_id, &mut branch.modules, branch.modules_node.take(), with_code).await?;
resolve_modules(
e,
workspace_id,
&mut branch.modules,
branch.modules_node.take(),
with_code,
)
.await?;
}
}
_ => {}
@@ -801,7 +841,13 @@ pub async fn resolve_modules(
.map(|flow| flow.modules)?;
}
for module in modules.iter_mut() {
Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?;
Box::pin(resolve_module(
e,
workspace_id,
&mut module.value,
with_code,
))
.await?;
}
Ok(())
}
backend/windmill-worker/src/worker_flow.rs
+1 -1
View File
@@ -1292,7 +1292,7 @@ async fn compute_skip_branchall_failure<'c>(
// )))
// }
fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> {
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))
frontend/src/lib/components/flows/content/FlowRetries.svelte
+33 -5
View File
@@ -50,6 +50,8 @@
$: flowModuleRetry === undefined && resetDelayType()
$: !loaded && initialLoad()
const u32Max = 4294967295
</script>
<div class="h-full flex flex-col {$$props.class ?? ''}">
@@ -76,14 +78,34 @@
{#if delayType === 'constant'}
{#if flowModuleRetry?.constant}
<div class="text-xs font-bold !mt-2">Attempts</div>
<input bind:value={flowModuleRetry.constant.attempts} type="number" />
<div class="flex gap-1">
<input
max={u32Max.toString()}
bind:value={flowModuleRetry.constant.attempts}
type="number"
/>
<button
class="text-xs"
on:click={() =>
flowModuleRetry?.constant && (flowModuleRetry.constant.attempts = u32Max)}
>max</button
>
</div>
<div class="text-xs font-bold !mt-2">Delay</div>
<SecondsInput bind:seconds={flowModuleRetry.constant.seconds} />
{/if}
{:else if delayType === 'exponential'}
{#if flowModuleRetry?.exponential}
<div class="text-xs font-bold !mt-2">Attempts</div>
<input bind:value={flowModuleRetry.exponential.attempts} type="number" />
<div class="flex gap-1">
<input max="100" bind:value={flowModuleRetry.exponential.attempts} type="number" />
<button
class="text-xs"
on:click={() =>
flowModuleRetry?.exponential && (flowModuleRetry.exponential.attempts = 100)}
>max</button
>
</div>
<div class="text-xs font-bold !mt-2">Multiplier</div>
<span class="text-xs text-tertiary">delay = multiplier * base ^ (number of attempt)</span>
<input bind:value={flowModuleRetry.exponential.multiplier} type="number" />
@@ -127,9 +149,9 @@
multiplier,
random_factor
} = flowModuleRetry?.exponential || {}}
{@const cArray = Array.from({ length: cAttempts || 0 }, () => cSeconds)}
{@const cArray = Array.from({ length: Math.min(cAttempts || 0, 100) }, () => cSeconds)}
{@const eArray = Array.from(
{ length: eAttempts || 0 },
{ length: Math.min(eAttempts || 0, 100) },
(_, i) => (multiplier || 0) * (eSeconds || 0) ** (i + cArray.length + 1)
)}
{@const array = [...cArray, ...eArray]}
@@ -146,7 +168,7 @@
seconds){/if}</td
>
</tr>
{#each array.slice(1) as delay, i}
{#each array.slice(1, 100) as delay, i}
{@const index = i + 2}
<tr>
<td class="font-semibold pr-1 align-top">{index}:</td>
@@ -163,6 +185,12 @@
</td>
</tr>
{/each}
{#if (cAttempts ?? 0) > 100 || (eAttempts ?? 0) > 100}
<tr>
<td class="font-semibold pr-1 align-top">...</td>
<td class="pb-1">...</td>
</tr>
{/if}
</table>
{:else}
<div class="text-xs">No retries</div>