From e96c5ca670f2d1efb698f99fa76d0ecc4f01fac1 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 30 Oct 2022 02:32:35 +0200 Subject: [PATCH] early stop now bubble up to the top --- backend/windmill-worker/src/worker.rs | 3 ++ backend/windmill-worker/src/worker_flow.rs | 37 +++++++++++++------ .../flows/content/FlowBranchesWrapper.svelte | 8 ++-- .../components/flows/content/FlowLoop.svelte | 8 ++-- 4 files changed, 37 insertions(+), 19 deletions(-) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index f51f827e90..c2ba1dbb85 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -503,6 +503,7 @@ async fn handle_job_error( worker_dir, keep_job_dir, base_internal_url, + None, ) .await; if let Err(err) = updated_flow { @@ -656,6 +657,7 @@ async fn handle_queued_job( worker_dir, worker_config.keep_job_dir, &worker_config.base_internal_url, + None, ) .await?; } @@ -705,6 +707,7 @@ async fn handle_queued_job( worker_dir, worker_config.keep_job_dir, &worker_config.base_internal_url, + None, ) .await?; } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index d059b7c01c..aa9babec0c 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -50,6 +50,7 @@ pub async fn update_flow_status_after_job_completion( worker_dir: &str, keep_job_dir: bool, base_internal_url: &str, + stop_early_override: Option, ) -> error::Result<()> { tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id}"); @@ -152,8 +153,13 @@ pub async fn update_flow_status_after_job_completion( .map(|i| !(..old_status.modules.len()).contains(&i)) .unwrap_or(true); - let (stop_early_expr, skip_if_stop_early) = - sqlx::query_as::<_, (Option, Option)>( + let (stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override { + (true, se) + } else { + let (stop_early_expr, skip_if_stop_early) = sqlx::query_as::< + _, + (Option, Option), + >( " UPDATE queue SET flow_status = JSONB_SET( @@ -173,12 +179,16 @@ pub async fn update_flow_status_after_job_completion( .await .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; - let stop_early = success - && if let Some(expr) = stop_early_expr.clone() { - compute_bool_from_expr(expr, result.clone(), base_internal_url).await? - } else { - false - }; + ( + success + && if let Some(expr) = stop_early_expr.clone() { + compute_bool_from_expr(expr, result.clone(), base_internal_url).await? + } else { + false + }, + skip_if_stop_early.unwrap_or(false), + ) + }; let result = match &new_status { FlowStatusModule::Success { flow_jobs: Some(jobs), .. } => { @@ -266,8 +276,7 @@ pub async fn update_flow_status_after_job_completion( let logs = if flow_job.canceled { "Flow job canceled".to_string() } else if stop_early { - let stop_early_expr = stop_early_expr.unwrap(); - format!("Flow job stopped early based on the stop_early_expr predicate: {stop_early_expr} returning true") + format!("Flow job stopped early because of a stop early predicate returning true") } else { "Flow job completed".to_string() }; @@ -288,7 +297,7 @@ pub async fn update_flow_status_after_job_completion( client, &flow_job, success, - stop_early && skip_if_stop_early.unwrap_or(false), + stop_early && skip_if_stop_early, result.clone(), logs, ) @@ -343,6 +352,11 @@ pub async fn update_flow_status_after_job_completion( worker_dir, keep_job_dir, base_internal_url, + if stop_early { + Some(skip_if_stop_early) + } else { + None + }, ) .await?); } @@ -589,6 +603,7 @@ pub async fn handle_flow( worker_dir, false, base_internal_url, + None, ) .await?; return Ok(()); diff --git a/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte b/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte index 3de91865e7..6f9205760a 100644 --- a/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte +++ b/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte @@ -6,7 +6,7 @@ import FlowCard from '../common/FlowCard.svelte' import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte' import FlowModuleSuspend from './FlowModuleSuspend.svelte' - import FlowRetries from './FlowRetries.svelte' + // import FlowRetries from './FlowRetries.svelte' export let flowModule: FlowModule export let type: 'branchall' | 'branchone' @@ -50,17 +50,17 @@ {#if flowModule} - Retries + Early Stop Sleep/Suspend
- +
diff --git a/frontend/src/lib/components/flows/content/FlowLoop.svelte b/frontend/src/lib/components/flows/content/FlowLoop.svelte index a3aacd2182..9045567ae9 100644 --- a/frontend/src/lib/components/flows/content/FlowLoop.svelte +++ b/frontend/src/lib/components/flows/content/FlowLoop.svelte @@ -8,7 +8,7 @@ import PropPickerWrapper from '../propPicker/PropPickerWrapper.svelte' import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte' import FlowModuleSuspend from './FlowModuleSuspend.svelte' - import FlowRetries from './FlowRetries.svelte' + // import FlowRetries from './FlowRetries.svelte' import { Button, Tab, TabContent, Tabs } from '$lib/components/common' import type { FlowModule } from '$lib/gen/models/FlowModule' import { Pane, Splitpanes } from 'svelte-splitpanes' @@ -85,17 +85,17 @@ - Retries + Early Stop Sleep/Suspend
- +