mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-25 16:02:11 +00:00
fix(pipelines): pipeline-level run control, tables label, data-test rollback + fork badges (#9944)
* fix(pipelines): pipeline-level run control, tables label, data-test rollback + fork badges - Add always-visible "Run pipeline" header control (edit mode) that runs every script in dependency order via the bounded-cascade engine, so a run no longer requires hovering a node's play button. - Header summary counts ducklake/datatable assets as "tables" (and s3object as "files") instead of the raw kind, collapsing shared nouns. - Surface a data-test outcome badge on guarded asset nodes: EE shows a rolled-back (previous version left live) state, CE shows published-despite- failure — driven by the producer's last run state and the edition. - Make the fork data-environment marker a prominent labeled chip (⑂ fork / ↗ parent) instead of a bare icon. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(pipelines): address CI review — scope Run pipeline to members, anchor guard badge, spin loader - Run pipeline now filters to `in_pipeline` script runnables, so it never launches dependency-only endpoints the graph shows for context (macro libraries, custom data-test scripts, out-of-folder producers). - Data-test guard badge only attaches to the producer's declared `// materialize` target, so a multi-output producer no longer badges its other ducklake writes. - Spin the Loader2 icon in the "Run pipeline" button while a run is in progress (startIcon classes), matching every other loading affordance. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(pipelines): data-test badge copy speaks to write policy, not failure cause producerFailed is a generic job-failure signal, so the failed-state tooltip no longer claims the run "failed its data tests" (it could be a runtime/worker error). It now states the edition's behavior on any failed materialize: EE rolls back (previous version left live), CE may leave a failing write live. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(pipelines): Run pipeline keeps independent branches running after a failure runSelection used a single global fail-fast flag, so once any node failed it refused to schedule *any* newly-ready node — a failure in one branch could strand an unrelated healthy branch as 'skipped' depending on job timing. Now a failure poisons only its transitive descendants; independent branches finish. Add regression tests: independent-branch-survives-failure and join-node-skipped -when-one-upstream-fails. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
GitHub
@@ -323,8 +323,44 @@
|
||||
// own adjacency (boundedCascade.buildLineageDownstreamMap).
|
||||
const hasLineageDownstream = new Set<string>(buildLineageDownstreamMap(g).keys())
|
||||
|
||||
// Producers that declare `// data_test` checks, keyed by runnable id — the
|
||||
// asset node uses this (plus the producer's run state) to render the
|
||||
// data-test outcome badge. Tests only assert on ducklake `// materialize`
|
||||
// targets (v1), so guard status is a ducklake-only concept below.
|
||||
const producerHasTests = new Set<string>()
|
||||
// Producer → its declared `// materialize` target, so a multi-output
|
||||
// producer's guard badge lands only on the table its tests assert on —
|
||||
// not on its other ducklake outputs. Mirrors the write-edge badge anchor.
|
||||
const guardMaterializeTarget = new Map<
|
||||
string,
|
||||
NonNullable<AssetGraphResponse['runnables'][number]['materialize_target']>
|
||||
>()
|
||||
for (const r of g.runnables) {
|
||||
if (r.data_tests && r.data_tests.length > 0) producerHasTests.add(`${r.usage_kind}:${r.path}`)
|
||||
if (r.materialize_target)
|
||||
guardMaterializeTarget.set(`${r.usage_kind}:${r.path}`, r.materialize_target)
|
||||
}
|
||||
|
||||
for (const a of g.assets) {
|
||||
const assetId = `asset:${a.kind}:${a.path}`
|
||||
// Guard/outcome badge inputs: is a producer of this (ducklake) asset
|
||||
// test-guarded, and did that guarded producer's latest run fail?
|
||||
let dataTestGuarded = false
|
||||
let producerFailed = false
|
||||
if (a.kind === 'ducklake') {
|
||||
for (const p of producersByAsset.get(`${a.kind}:${a.path}`) ?? []) {
|
||||
const rid = `${p.kind}:${p.path}`
|
||||
if (!producerHasTests.has(rid)) continue
|
||||
// Tests assert on the producer's `// materialize` target; when the
|
||||
// producer declares one, only that table is guarded (a multi-output
|
||||
// producer must not badge its other ducklake writes). No declared
|
||||
// target → single-output producer, so its lone ducklake write is it.
|
||||
const mt = guardMaterializeTarget.get(rid)
|
||||
if (mt && !(mt.kind === a.kind && mt.path === a.path)) continue
|
||||
dataTestGuarded = true
|
||||
if (runStates?.get(rid)?.status === 'failure') producerFailed = true
|
||||
}
|
||||
}
|
||||
nodes.push({
|
||||
id: assetId,
|
||||
type: 'asset',
|
||||
@@ -337,7 +373,9 @@
|
||||
pathPrefix,
|
||||
defaultPathSuffix,
|
||||
producers: producersByAsset.get(`${a.kind}:${a.path}`) ?? [],
|
||||
onRunProducer
|
||||
onRunProducer,
|
||||
dataTestGuarded,
|
||||
producerFailed
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,9 +5,19 @@
|
||||
import { formatShortAssetPath, type AssetKind } from '$lib/components/assets/lib'
|
||||
import { NODE } from '$lib/components/graph/util'
|
||||
import PipelineInsertMenu, { type PipelineInsertPick } from './PipelineInsertMenu.svelte'
|
||||
import { ArrowUpRight, Code2, GitFork, History, Play, Loader2, Plus } from 'lucide-svelte'
|
||||
import {
|
||||
ArrowUpRight,
|
||||
Code2,
|
||||
GitFork,
|
||||
History,
|
||||
Play,
|
||||
Loader2,
|
||||
Plus,
|
||||
ShieldCheck,
|
||||
ShieldAlert
|
||||
} from 'lucide-svelte'
|
||||
import type { ScriptLang } from '$lib/gen'
|
||||
import { workspaceStore } from '$lib/stores'
|
||||
import { enterpriseLicense, workspaceStore } from '$lib/stores'
|
||||
import { sendUserToast } from '$lib/utils'
|
||||
import { PIPELINE_LANGUAGES } from './pipelineLanguages'
|
||||
import type { PipelineOutputKind } from './pipelineTemplates'
|
||||
@@ -55,6 +65,15 @@
|
||||
// cached draft content). Without this callback, the play button
|
||||
// is hidden — runs only make sense in editor contexts.
|
||||
onRunProducer?: (producer: AssetProducer) => Promise<string | undefined>
|
||||
// True when a producer materializing this asset declares `// data_test`
|
||||
// checks — the asset's write is guarded. Drives the data-test outcome
|
||||
// badge, whose meaning differs by edition (EE rolls a failing write
|
||||
// back; CE publishes it anyway).
|
||||
dataTestGuarded?: boolean
|
||||
// True when the latest observed run of a producer materializing this
|
||||
// asset failed. Escalates the guard badge from "protected" to a
|
||||
// failed-run outcome (rolled-back on EE, published-anyway on CE).
|
||||
producerFailed?: boolean
|
||||
}
|
||||
// SvelteFlow injects this on the node component when the user clicks
|
||||
// the node. Combined with our own `hovered` state to drive the
|
||||
@@ -109,6 +128,37 @@
|
||||
}
|
||||
|
||||
let showAdd = $derived(data.onAddScript != undefined)
|
||||
|
||||
// Data-test outcome badge. Only guarded assets show it. The write's fate on a
|
||||
// failing test differs by edition — surface which one applies so a shared
|
||||
// parent/fork table name can't hide a silently-published bad version.
|
||||
let isEE = $derived(!!$enterpriseLicense)
|
||||
let showGuardBadge = $derived(data.dataTestGuarded === true)
|
||||
let guardFailed = $derived(data.producerFailed === true)
|
||||
let GuardIcon = $derived(isEE ? ShieldCheck : ShieldAlert)
|
||||
// Filled + colored when the last run failed (the actionable state); a quiet
|
||||
// ring at rest so the badge doesn't shout on every healthy guarded asset.
|
||||
let guardClass = $derived(
|
||||
guardFailed
|
||||
? isEE
|
||||
? 'bg-amber-500 text-white border-amber-600'
|
||||
: 'bg-red-500 text-white border-red-600'
|
||||
: isEE
|
||||
? 'bg-surface-secondary text-emerald-600 dark:text-emerald-400 border-emerald-500/60'
|
||||
: 'bg-surface-secondary text-amber-600 dark:text-amber-400 border-amber-500/60'
|
||||
)
|
||||
// Failed copy speaks to the edition's write policy, not the failure cause:
|
||||
// `producerFailed` is a generic job failure (could be a runtime/worker error,
|
||||
// not a data-test violation), so we don't assert "failed its data tests".
|
||||
let guardTitle = $derived(
|
||||
guardFailed
|
||||
? isEE
|
||||
? 'Last run failed — Enterprise rolls a failed materialize back, so the previous version is left live.'
|
||||
: 'Last run failed — Community Edition does not roll a failed materialize back (Enterprise does), so a failing write may be left live. Verify the table.'
|
||||
: isEE
|
||||
? 'Guarded by data tests: a failing write is rolled back, keeping the previous version live.'
|
||||
: 'Guarded by data tests, but Community Edition does not block on failure — a failing write is still published. Rollback is Enterprise-only.'
|
||||
)
|
||||
</script>
|
||||
|
||||
<!-- onmouseenter/leave on the wrapper (not the inner card) so the run
|
||||
@@ -142,23 +192,26 @@
|
||||
<span class="flex-1 min-w-0 pr-1 py-0.5 text-2xs font-mono text-emphasis truncate">
|
||||
{formatShortAssetPath(asset)}
|
||||
</span>
|
||||
<!-- Fork data-environment marker: amber ↗ = deferred (reads the parent
|
||||
workspace's current table via a view), emerald fork glyph = the fork
|
||||
materialized its own copy. Icon-only — the pill truncates its path
|
||||
already, a labeled chip wouldn't fit; the title carries the meaning. -->
|
||||
<!-- Fork data-environment chip: in a fork every asset shares its parent's
|
||||
name, so the env it resolves to must read at a glance. Labeled + tinted
|
||||
(amber "parent" = deferred read of the parent's current table via a
|
||||
view; emerald "fork" = the fork's own materialized copy) rather than a
|
||||
bare icon, which was too easy to miss. The title carries the detail. -->
|
||||
{#if data.fork_materialization === 'deferred'}
|
||||
<span
|
||||
class="shrink-0 mr-1.5 text-amber-600 dark:text-amber-400"
|
||||
class="shrink-0 mr-1.5 flex items-center gap-0.5 rounded px-1 py-px text-3xs font-semibold uppercase tracking-wide bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300 border border-amber-300 dark:border-amber-700"
|
||||
title="Deferred to parent workspace: reads the parent's current data. Materialize it in this fork to iterate on it."
|
||||
>
|
||||
<ArrowUpRight size={12} />
|
||||
<ArrowUpRight size={10} />
|
||||
parent
|
||||
</span>
|
||||
{:else if data.fork_materialization === 'fork'}
|
||||
<span
|
||||
class="shrink-0 mr-1.5 text-emerald-600 dark:text-emerald-500"
|
||||
class="shrink-0 mr-1.5 flex items-center gap-0.5 rounded px-1 py-px text-3xs font-semibold uppercase tracking-wide bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300 border border-emerald-300 dark:border-emerald-700"
|
||||
title="Materialized in this fork: reads and writes use the fork's isolated copy."
|
||||
>
|
||||
<GitFork size={12} />
|
||||
<GitFork size={10} />
|
||||
fork
|
||||
</span>
|
||||
{/if}
|
||||
<!-- SCD2 companion marker: this node is the `<dim>_current` "latest row
|
||||
@@ -173,6 +226,21 @@
|
||||
</span>
|
||||
{/if}
|
||||
</div>
|
||||
{#if showGuardBadge}
|
||||
<!-- Data-test outcome badge. Floats off the TOP-RIGHT corner (opposite the
|
||||
left-edge run button and the bottom + inserter) so it never collides
|
||||
with the other node affordances. Shield = guarded; its fill escalates to
|
||||
amber/red when the last run failed, encoding the edition's write policy. -->
|
||||
<div
|
||||
class={twMerge(
|
||||
'absolute -top-2 -right-2 z-10 rounded-full w-5 h-5 grid place-items-center border shadow-sm',
|
||||
guardClass
|
||||
)}
|
||||
title={guardTitle}
|
||||
>
|
||||
<GuardIcon size={12} strokeWidth={2.25} />
|
||||
</div>
|
||||
{/if}
|
||||
{#if showActions}
|
||||
<!-- Run button revealed on hover/select. Floats off the LEFT edge —
|
||||
visually closer to the upstream producer it triggers (which lays
|
||||
|
||||
@@ -201,7 +201,48 @@ describe('runSelection', () => {
|
||||
expect(r.launched.indexOf('c')).toBeGreaterThan(r.launched.indexOf('b'))
|
||||
})
|
||||
|
||||
it('stops scheduling after a failure', async () => {
|
||||
it('skips a failed node’s descendants but keeps independent branches running', async () => {
|
||||
// Two independent chains: a → b and c → d. `a` fails; `b` (its descendant)
|
||||
// must be skipped, but `d` depends only on the successful `c`, so it must
|
||||
// still run — a failure must not stall unrelated branches.
|
||||
const sched = schedule(
|
||||
[
|
||||
['a', 'b'],
|
||||
['c', 'd']
|
||||
],
|
||||
['a', 'b', 'c', 'd'],
|
||||
['a', 'c']
|
||||
)
|
||||
const r = fakeRunner({ a: 'failure' })
|
||||
const res = await runSelection({ schedule: sched, ...r })
|
||||
expect(res.ok).toBe(false)
|
||||
expect(res.statuses.get('a')?.status).toBe('failure')
|
||||
expect(res.statuses.get('b')?.status).toBe('skipped')
|
||||
expect(res.statuses.get('c')?.status).toBe('success')
|
||||
expect(res.statuses.get('d')?.status).toBe('success')
|
||||
expect(r.launched).toContain('d')
|
||||
expect(r.launched).not.toContain('b')
|
||||
})
|
||||
|
||||
it('skips a join node when any one of its upstreams fails', async () => {
|
||||
// {a, b} → c. `a` fails; `c` needs both, so it must be skipped even though
|
||||
// `b` succeeds — a poisoned lineage isn’t rescued by a sibling success.
|
||||
const sched = schedule(
|
||||
[
|
||||
['a', 'c'],
|
||||
['b', 'c']
|
||||
],
|
||||
['a', 'b', 'c'],
|
||||
['a', 'b']
|
||||
)
|
||||
const r = fakeRunner({ a: 'failure' })
|
||||
const res = await runSelection({ schedule: sched, ...r })
|
||||
expect(res.ok).toBe(false)
|
||||
expect(res.statuses.get('c')?.status).toBe('skipped')
|
||||
expect(r.launched).not.toContain('c')
|
||||
})
|
||||
|
||||
it('stops scheduling a failed node’s chain', async () => {
|
||||
const sched = schedule([['a', 'b']], ['a', 'b'], ['a'])
|
||||
const r = fakeRunner({ a: 'failure' })
|
||||
const res = await runSelection({ schedule: sched, ...r })
|
||||
|
||||
@@ -128,8 +128,9 @@ export type SelectionRunOptions = {
|
||||
* Execute an arbitrary selected set of scripts (e.g. a bounded-cascade
|
||||
* selection) in topological order. Unlike `runCascade` there is no single
|
||||
* privileged root — every `schedule.roots` entry is seeded at once and a node
|
||||
* runs as soon as its in-set upstreams all succeed. Failure stops *scheduling*
|
||||
* (in-flight jobs finish); everything not yet started ends 'skipped'.
|
||||
* runs as soon as its in-set upstreams all succeed. A failure abandons only
|
||||
* that node's lineage (its transitive descendants end 'skipped'); INDEPENDENT
|
||||
* branches keep running. In-flight jobs always finish.
|
||||
*/
|
||||
export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRunResult> {
|
||||
const { schedule, launch, waitTerminal, onUpdate } = opts
|
||||
@@ -137,10 +138,29 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
|
||||
for (const n of schedule.nodes) statuses.set(n, { status: 'pending' })
|
||||
const remaining = new Map(schedule.indegree)
|
||||
let failed = false
|
||||
// Nodes a failed prerequisite has made unrunnable — the transitive descendants
|
||||
// of every failure. Gating scheduling on this (rather than a single global
|
||||
// fail-fast flag) is what lets independent branches finish: only the lineage
|
||||
// below a failure is skipped, not every node that happens to become ready
|
||||
// afterwards. Poisoned nodes are never scheduled, so they end 'skipped' below.
|
||||
const poisoned = new Set<string>()
|
||||
const inFlight = new Set<Promise<void>>()
|
||||
|
||||
const emit = () => onUpdate?.(new Map(statuses))
|
||||
|
||||
function poison(path: string) {
|
||||
const stack = [path]
|
||||
while (stack.length > 0) {
|
||||
const n = stack.pop()!
|
||||
for (const s of schedule.edges.get(n) ?? []) {
|
||||
if (!poisoned.has(s)) {
|
||||
poisoned.add(s)
|
||||
stack.push(s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function schedule_(path: string) {
|
||||
const p = runNode(path).finally(() => inFlight.delete(p))
|
||||
inFlight.add(p)
|
||||
@@ -159,6 +179,7 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
|
||||
emit()
|
||||
if (term === 'failure') {
|
||||
failed = true
|
||||
poison(path)
|
||||
return
|
||||
}
|
||||
} catch (e) {
|
||||
@@ -169,12 +190,13 @@ export async function runSelection(opts: SelectionRunOptions): Promise<CascadeRu
|
||||
})
|
||||
emit()
|
||||
failed = true
|
||||
poison(path)
|
||||
return
|
||||
}
|
||||
for (const s of schedule.edges.get(path) ?? []) {
|
||||
const d = (remaining.get(s) ?? 0) - 1
|
||||
remaining.set(s, d)
|
||||
if (d === 0 && !failed) schedule_(s)
|
||||
if (d === 0 && !poisoned.has(s)) schedule_(s)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1643,6 +1643,25 @@
|
||||
}
|
||||
}
|
||||
|
||||
// Every pipeline-member script, for the always-visible header "Run pipeline"
|
||||
// control. Per-node runs are hover/select-gated on the canvas; this
|
||||
// pipeline-level affordance runs the whole graph without hunting for a root
|
||||
// node to hover. Gated on `in_pipeline` so it never launches dependency-only
|
||||
// endpoints the graph surfaces for context (macro libraries, custom
|
||||
// data-test scripts, out-of-folder producers) — only actual pipeline steps.
|
||||
let allPipelineScripts = $derived(
|
||||
displayGraph.runnables
|
||||
.filter((r) => r.usage_kind === 'script' && r.in_pipeline)
|
||||
.map((r) => r.path)
|
||||
)
|
||||
// Run the whole pipeline: hand every script to the bounded-cascade engine,
|
||||
// which topo-orders them (roots first) and fans downstream — i.e. run + all
|
||||
// downstream from every source at once. Reuses the same per-hop launch/poll
|
||||
// and one-cascade-at-a-time guard as the node-level chain runs.
|
||||
async function runWholePipeline() {
|
||||
await runBoundedCascade(allPipelineScripts)
|
||||
}
|
||||
|
||||
// Counter bumped when the canvas Run button targets the currently-open
|
||||
// script — the pane intercepts and routes through ScriptEditor.runTest
|
||||
// so logs/result/cancel land in the test panel instead of going off
|
||||
@@ -2004,6 +2023,18 @@
|
||||
return `${n} ${singular}${n === 1 ? '' : 's'}`
|
||||
}
|
||||
|
||||
// Human noun per asset kind for the header summary. The raw `ducklake` kind
|
||||
// counts materialized tables/views (including `_current` history views), so
|
||||
// "N ducklakes" mis-reads as a count of lakes — surface "table" instead. Other
|
||||
// kinds keep their own noun; unmapped kinds fall back to the raw kind.
|
||||
const ASSET_KIND_NOUN: Record<string, string> = {
|
||||
ducklake: 'table',
|
||||
datatable: 'table',
|
||||
s3object: 'file',
|
||||
volume: 'volume',
|
||||
resource: 'resource'
|
||||
}
|
||||
|
||||
let summary = $derived.by<string[]>(() => {
|
||||
const g = graphRes.current
|
||||
if (!g) return []
|
||||
@@ -2012,9 +2043,14 @@
|
||||
const flows = g.runnables.filter((r) => r.usage_kind === 'flow').length
|
||||
if (scripts) parts.push(pluralize(scripts, 'script'))
|
||||
if (flows) parts.push(pluralize(flows, 'flow'))
|
||||
const byKind = new Map<string, number>()
|
||||
for (const a of g.assets) byKind.set(a.kind, (byKind.get(a.kind) ?? 0) + 1)
|
||||
for (const [kind, n] of byKind) parts.push(pluralize(n, kind))
|
||||
// Collapse kinds that share a noun (ducklake + datatable → "table") into a
|
||||
// single tally so the summary reads "5 tables", not "3 tables · 2 tables".
|
||||
const byNoun = new Map<string, number>()
|
||||
for (const a of g.assets) {
|
||||
const noun = ASSET_KIND_NOUN[a.kind] ?? a.kind
|
||||
byNoun.set(noun, (byNoun.get(noun) ?? 0) + 1)
|
||||
}
|
||||
for (const [noun, n] of byNoun) parts.push(pluralize(n, noun))
|
||||
return parts
|
||||
})
|
||||
</script>
|
||||
@@ -2098,6 +2134,28 @@
|
||||
</div>
|
||||
{/if}
|
||||
<div class="flex flex-row items-center gap-2 flex-1 justify-end">
|
||||
{#if mode === 'edit' && !isOperator && allPipelineScripts.length > 0}
|
||||
<!-- Pipeline-level run: always visible so a run doesn't require
|
||||
hovering a specific node's play button. Runs every script in
|
||||
dependency order (roots first, cascading downstream). -->
|
||||
<Button
|
||||
variant="accent-secondary"
|
||||
unifiedSize="sm"
|
||||
startIcon={{
|
||||
icon: cascadeRunningRoot ? Loader2 : Play,
|
||||
classes: cascadeRunningRoot ? 'animate-spin' : undefined
|
||||
}}
|
||||
onclick={runWholePipeline}
|
||||
disabled={!!cascadeRunningRoot}
|
||||
title={cascadeRunningRoot
|
||||
? 'A pipeline run is already in progress'
|
||||
: `Run all ${allPipelineScripts.length} script${
|
||||
allPipelineScripts.length === 1 ? '' : 's'
|
||||
} in dependency order (roots first, cascading downstream)`}
|
||||
>
|
||||
{cascadeRunningRoot ? 'Running…' : 'Run pipeline'}
|
||||
</Button>
|
||||
{/if}
|
||||
{#if mode === 'edit' && saveErrors.size > 0}
|
||||
<!-- Compact errors popover anchored next to Save all so users
|
||||
can see exactly which drafts failed and why without losing
|
||||
|
||||
Reference in New Issue
Block a user