early stop now bubble up to the top

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), .. }) => {

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<string, GraphModuleState> = {}
+
+	let localFlowModuleStates: Record<string, GraphModuleState> = {}
 
 	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}
 					/>
 				</div>
-				<div class="border-l border-gray-400">
+				<div class="border-l border-gray-400 pt-1">
 					{#if selectedNode}
-						{#if localFlowModuleStates[selectedNode]}
-							<div class="px-2">
-								<ModuleStatus
-									type={localFlowModuleStates[selectedNode]?.type}
-									scheduled_for={localFlowModuleStates[selectedNode]?.['scheduled_for']}
-								/>
+						{@const node = localFlowModuleStates[selectedNode]}
+						{#if node}
+							<div class="px-2 flex gap-2 min-w-0 ">
+								<ModuleStatus type={node.type} scheduled_for={node['scheduled_for']} />
+								{#if node.job_id}
+									<div class="truncate"
+										><div class=" text-gray-900 whitespace-nowrap truncate">
+											<span class="font-bold">Job Id</span>
+											<a
+												rel="noreferrer"
+												target="_blank"
+												href="/run/{node.job_id ?? ''}?workspace={job?.workspace_id}"
+											>
+												{truncateRev(node.job_id ?? '', 10) ?? ''}
+											</a>
+										</div>
+									</div>
+								{/if}
 							</div>
-							<FlowJobResult
-								noBorder
-								col
-								result={localFlowModuleStates[selectedNode].result ?? {}}
-								logs={localFlowModuleStates[selectedNode].logs ?? ''}
-							/>
+							<FlowJobResult noBorder col result={node.result ?? {}} logs={node.logs ?? ''} />
 						{:else}
 							<p class="p-2 text-gray-600 italic"
 								>The execution of this node has no information attached to it. The job likely did

frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte

@@ -117,7 +117,7 @@
 			{#if flowModule}
 				<Tabs bind:selected>
 					<!-- <Tab value="retries">Retries</Tab> -->
-					<Tab value="early-stop">Early Stop</Tab>
+					<Tab value="early-stop">Early Stop/Break</Tab>
 					<Tab value="suspend">Suspend</Tab>
 					<Tab value="sleep">Sleep</Tab>
 

frontend/src/lib/components/flows/content/FlowLoop.svelte

@@ -98,7 +98,7 @@
 			<Pane size={40} minSize={20} class="flex flex-col flex-1">
 				<Tabs bind:selected>
 					<!-- <Tab value="retries">Retries</Tab> -->
-					<Tab value="early-stop">Early Stop</Tab>
+					<Tab value="early-stop">Early Stop/Break</Tab>
 					<Tab value="suspend">Suspend</Tab>
 					<Tab value="sleep">Sleep</Tab>
 

frontend/src/lib/components/flows/content/FlowModuleComponent.svelte

@@ -223,7 +223,7 @@
 								<Tabs bind:selected>
 									<Tab value="advanced-retries">Retries</Tab>
 									{#if !$selectedId.includes('failure')}
-										<Tab value="advanced-early-stop">Early Stop</Tab>
+										<Tab value="advanced-early-stop">Early Stop/Break</Tab>
 										<Tab value="advanced-suspend">Suspend</Tab>
 										<Tab value="advanced-sleep">Sleep</Tab>
 										<Tab value="advanced-same_worker">Shared Directory</Tab>

frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte

@@ -18,10 +18,11 @@
 
 <div class="flex flex-col items-start space-y-2 {$$props.class}">
 	<h2 class="mt-2"
-		>Early stop <Tooltip>
+		>Early stop/Break<Tooltip>
 			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.</Tooltip
+			page. If stop early is run within a forloop, it will just break the for-loop and have it stop
+			at that iteration instead of stopping the whole flow.</Tooltip
 		></h2
 	>
 	<Toggle
@@ -37,7 +38,7 @@
 			}
 		}}
 		options={{
-			right: 'Early stop if condition met'
+			right: 'Early stop or Break if condition met'
 		}}
 	/>
 

frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte

@@ -54,7 +54,7 @@
 					<div class="bg-white rounded border text-gray-700 px-1">
 						<Icon scale={0.8} data={faStop} />
 					</div>
-					<span slot="text">Early Stop</span>
+					<span slot="text">Early Stop/Break</span>
 				</Popover>
 			{/if}
 			{#if sleep}

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<string, GraphModuleState> | undefined = undefined
 
 	let selectedNode: string | undefined = undefined
 

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 {