From 3b46eb643c5fc9cf66ce499e020967b79ec6e1ee Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 6 Dec 2022 17:39:07 +0100 Subject: [PATCH] early stop now bubble up to the top --- backend/windmill-worker/src/worker_flow.rs | 77 +++++++++++-------- .../lib/components/FlowStatusViewer.svelte | 62 ++++++++------- .../flows/content/FlowBranchesWrapper.svelte | 2 +- .../components/flows/content/FlowLoop.svelte | 2 +- .../flows/content/FlowModuleComponent.svelte | 2 +- .../flows/content/FlowModuleEarlyStop.svelte | 7 +- .../flows/map/FlowModuleSchemaItem.svelte | 2 +- .../src/lib/components/graph/FlowGraph.svelte | 10 +-- frontend/src/lib/components/graph/model.ts | 9 +++ 9 files changed, 98 insertions(+), 75 deletions(-) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 75d07eca47..fa15a9ec0d 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -95,6 +95,38 @@ pub async fn update_flow_status_after_job_completion( false }; + let is_failure_step = old_status.step >= old_status.modules.len() as i32; + + let (mut stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override { + (true, se) + } else if is_failure_step { + (false, false) + } else { + let r = sqlx::query!( + " + SELECT raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr, + (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped, + args + FROM queue + WHERE id = $2 + ", + old_status.step, + flow + ) + .fetch_one(&mut tx) + .await + .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; + + let stop_early = success + && if let Some(expr) = r.stop_early_expr.clone() { + compute_bool_from_expr(expr, &r.args, result.clone(), base_internal_url, None, None) + .await? + } else { + false + }; + (stop_early, r.skip_if_stopped.unwrap_or(false)) + }; + let skip_branch_failure = match module_status { FlowStatusModule::InProgress { branchall: Some(BranchAllStatus { branch, .. }), .. @@ -190,12 +222,23 @@ pub async fn update_flow_status_after_job_completion( FlowStatusModule::InProgress { iterator: Some(windmill_common::flow_status::Iterator { index, itered, .. }), .. - } if (*index + 1 < itered.len() && (success || skip_loop_failures)) => (false, None), + } if (*index + 1 < itered.len() && (success || skip_loop_failures)) && !stop_early => { + (false, None) + } FlowStatusModule::InProgress { branchall: Some(BranchAllStatus { branch, len, .. }), .. } if branch.to_owned() < len - 1 && (success || skip_branch_failure) => (false, None), _ => { + if stop_early + && matches!( + module_status, + FlowStatusModule::InProgress { iterator: Some(_), .. } + ) + { + // if we're stopping early inside a loop, we just want to break the loop instead + stop_early = false; + } let (flow_jobs, branch_chosen) = match module_status { FlowStatusModule::InProgress { flow_jobs, branch_chosen, .. } => { (flow_jobs.clone(), branch_chosen.clone()) @@ -249,8 +292,6 @@ pub async fn update_flow_status_after_job_completion( .map(|i| !(..old_status.modules.len()).contains(&i)) .unwrap_or(true); - let is_failure_step = old_status.step >= old_status.modules.len() as i32; - if let Some(new_status) = new_status.as_ref() { if is_failure_step { sqlx::query!( @@ -280,36 +321,6 @@ pub async fn update_flow_status_after_job_completion( } } - let (stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override { - (true, se) - } else if is_failure_step { - (false, false) - } else { - let r = sqlx::query!( - " - SELECT raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr, - (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped, - args - FROM queue - WHERE id = $2 - ", - old_status.step, - flow - ) - .fetch_one(&mut tx) - .await - .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; - - let stop_early = success - && if let Some(expr) = r.stop_early_expr.clone() { - compute_bool_from_expr(expr, &r.args, result.clone(), base_internal_url, None, None) - .await? - } else { - false - }; - (stop_early, r.skip_if_stopped.unwrap_or(false)) - }; - let result = match &new_status { Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. }) | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => { diff --git a/frontend/src/lib/components/FlowStatusViewer.svelte b/frontend/src/lib/components/FlowStatusViewer.svelte index fea6256912..0b210b7e00 100644 --- a/frontend/src/lib/components/FlowStatusViewer.svelte +++ b/frontend/src/lib/components/FlowStatusViewer.svelte @@ -11,9 +11,9 @@ import { Button, Tab } from './common' import DisplayResult from './DisplayResult.svelte' import Tabs from './common/tabs/Tabs.svelte' - import { FlowGraph } from './graph' + import { FlowGraph, type GraphModuleState } from './graph' import ModuleStatus from './ModuleStatus.svelte' - import { displayDate } from '$lib/utils' + import { displayDate, truncateRev } from '$lib/utils' const dispatch = createEventDispatcher() @@ -27,15 +27,10 @@ } | undefined = undefined export let job: Job | undefined = undefined - export let flowModuleStates: Record< - string, - { type: FlowStatusModule.type; logs?: string; result?: any; scheduled_for?: string } - > = {} - let localFlowModuleStates: Record< - string, - { type: FlowStatusModule.type; logs?: string; result?: any; scheduled_for?: string } - > = {} + export let flowModuleStates: Record = {} + + let localFlowModuleStates: Record = {} let selectedNode: string | undefined = undefined @@ -93,7 +88,8 @@ }).then((job) => { localFlowModuleStates[module.id ?? ''] = { type: module.type, - scheduled_for: 'scheduled for ' + displayDate(job?.['scheduled_for'], true) + scheduled_for: 'scheduled for ' + displayDate(job?.['scheduled_for'], true), + job_id: job?.id } }) } @@ -228,15 +224,17 @@ if (e.detail.type == 'QueuedJob') { localFlowModuleStates[flowJobIds.moduleId] = { type: FlowStatusModule.type.IN_PROGRESS, - logs: e.detail.logs + logs: e.detail.logs, + job_id: e.detail.id } } else { localFlowModuleStates[flowJobIds.moduleId] = { type: e.detail.success ? FlowStatusModule.type.SUCCESS : FlowStatusModule.type.FAILURE, - logs: e.detail.logs, - result: e.detail.result + logs: 'All jobs completed', + result: jobResults, + job_id: e.detail.id } } } @@ -299,7 +297,8 @@ ? FlowStatusModule.type.SUCCESS : FlowStatusModule.type.FAILURE, logs: e.detail.logs, - result: e.detail.result + result: e.detail.result, + job_id: e.detail.id } } } @@ -333,21 +332,28 @@ failureModule={job.raw_flow?.failure_module} /> -
+
{#if selectedNode} - {#if localFlowModuleStates[selectedNode]} -
- + {@const node = localFlowModuleStates[selectedNode]} + {#if node} +
+ + {#if node.job_id} +
+ Job Id + + {truncateRev(node.job_id ?? '', 10) ?? ''} + +
+
+ {/if}
- + {:else}

The execution of this node has no information attached to it. The job likely did diff --git a/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte b/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte index e6989e1067..4be8bfdbc4 100644 --- a/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte +++ b/frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte @@ -117,7 +117,7 @@ {#if flowModule} - Early Stop + Early Stop/Break Suspend Sleep diff --git a/frontend/src/lib/components/flows/content/FlowLoop.svelte b/frontend/src/lib/components/flows/content/FlowLoop.svelte index dc1eb23298..4c934d0b25 100644 --- a/frontend/src/lib/components/flows/content/FlowLoop.svelte +++ b/frontend/src/lib/components/flows/content/FlowLoop.svelte @@ -98,7 +98,7 @@ - Early Stop + Early Stop/Break Suspend Sleep diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 934dd4e132..100ac73a83 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -223,7 +223,7 @@ Retries {#if !$selectedId.includes('failure')} - Early Stop + Early Stop/Break Suspend Sleep Shared Directory diff --git a/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte b/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte index d6ea6a49ed..c09f98b469 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte @@ -18,10 +18,11 @@

Early stop + >Early stop/Break If defined, at the end of the step, the predicate expression will be evaluated to decide if the flow should stop early. Skipped flows are just a label useful to not see them in the runs - page.

diff --git a/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte b/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte index da85a561b8..c81277c384 100644 --- a/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte +++ b/frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte @@ -54,7 +54,7 @@
- Early Stop + Early Stop/Break {/if} {#if sleep} diff --git a/frontend/src/lib/components/graph/FlowGraph.svelte b/frontend/src/lib/components/graph/FlowGraph.svelte index 1d4c016d52..f11b162d5e 100644 --- a/frontend/src/lib/components/graph/FlowGraph.svelte +++ b/frontend/src/lib/components/graph/FlowGraph.svelte @@ -13,7 +13,8 @@ type Loop, type Branch, type NestedNodes, - type ModuleHost + type ModuleHost, + type GraphModuleState } from '.' import { defaultIfEmptyString, truncateRev } from '$lib/utils' import { createEventDispatcher } from 'svelte' @@ -23,12 +24,7 @@ export let failureModule: FlowModule | undefined = undefined export let minHeight: number = 0 export let notSelectable = false - export let flowModuleStates: - | Record< - string, - { type: FlowStatusModule.type; logs?: string; result?: any; scheduled_for?: string } - > - | undefined = undefined + export let flowModuleStates: Record | undefined = undefined let selectedNode: string | undefined = undefined diff --git a/frontend/src/lib/components/graph/model.ts b/frontend/src/lib/components/graph/model.ts index d9999ebc26..93687a8a4e 100644 --- a/frontend/src/lib/components/graph/model.ts +++ b/frontend/src/lib/components/graph/model.ts @@ -1,3 +1,4 @@ +import type { FlowStatusModule } from "$lib/gen" import type { Node as SvelvetNode } from "svelvet" export type ModuleHost = 'workspace' | 'inline' | 'hub' @@ -17,6 +18,14 @@ export type Branch = { export type GraphItem = Node | Loop | Branch +export type GraphModuleState = { + type: FlowStatusModule.type + logs?: string + result?: any + scheduled_for?: string + job_id?: string +} + export type NestedNodes = GraphItem[] export function isNode(item: GraphItem | NestedNodes | undefined): item is Node {