feat(pipeline): backfill a range of partitions from the asset drawer (#9885)

* feat(pipeline): backfill a range of partitions from the asset drawer (ee)

* feat(pipeline): cancel in-flight backfill job and show cancelling state

* refactor(pipeline): move backfill range logic behind private feature

* fix(pipeline): close backfill cancel-launch race and record dispatch intent

* docs(openapi): producer_path also covers SDK write-edge producers

* chore: update ee-repo-ref to c3852ecb36bd0be1a74c63169e513888f3347850

This commit updates the EE repository reference after PR #641 was merged in windmill-ee-private.

Previous ee-repo-ref: 7c1450ef89fbc9e844a121b39cafe0d7235d704b

New ee-repo-ref: c3852ecb36bd0be1a74c63169e513888f3347850

Automated by sync-ee-ref workflow.

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

backend/.sqlx/query-4add520575717d449a3758326777fee03012a3fa0fb943239e004cc4a7067501.json

@@ -0,0 +1,30 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT DISTINCT ON (path) path AS \"path!\", content AS \"content!\"\n        FROM script\n        WHERE workspace_id = $1\n          AND auto_kind = 'pipeline'\n          AND archived = false\n          AND deleted = false\n          AND (content LIKE '%' || $2 || '%' OR path = ANY($3))\n        ORDER BY path, created_at DESC\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "content!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "4add520575717d449a3758326777fee03012a3fa0fb943239e004cc4a7067501"
+}

backend/.sqlx/query-d4f047b0a672d5bfe9fd7db60f11563ef05db7cc54a958a72384b08a69eb784e.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT DISTINCT usage_path FROM asset\n        WHERE workspace_id = $1\n          AND kind = 'ducklake'\n          AND path = $2\n          AND usage_kind = 'script'\n          AND usage_access_type IN ('w', 'rw')\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "usage_path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "d4f047b0a672d5bfe9fd7db60f11563ef05db7cc54a958a72384b08a69eb784e"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-bdd4ba4dfd05c8ca7db365e530812dc914517d7f
+c3852ecb36bd0be1a74c63169e513888f3347850

backend/windmill-api-assets/Cargo.toml

@@ -8,6 +8,9 @@ edition.workspace = true
 name = "windmill_api_assets"
 path = "src/lib.rs"
 
+[features]
+private = ["windmill-common/private"]
+
 [dependencies]
 windmill-api-auth.workspace = true
 windmill-common = { workspace = true, default-features = false }

backend/windmill-api-assets/src/backfill_oss.rs

@@ -0,0 +1,18 @@
+//! OSS fallback: backfilling a range of partitions is an enterprise
+//! feature (the resolution/enumeration logic lives in `windmill-ee-private`,
+//! `windmill-api-assets/src/backfill_ee.rs`). Single-partition runs with an
+//! explicit `partition` arg remain available in OSS.
+
+use windmill_common::error::{Error, Result};
+
+use crate::{PartitionsInRangeQuery, PartitionsInRangeResponse};
+
+pub(crate) async fn partitions_in_range(
+    _tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    _w_id: &str,
+    _q: &PartitionsInRangeQuery,
+) -> Result<PartitionsInRangeResponse> {
+    Err(Error::BadRequest(
+        "Backfilling a range of partitions is an enterprise feature".to_string(),
+    ))
+}

backend/windmill-api-assets/src/lib.rs

@@ -15,6 +15,18 @@ use windmill_common::{
 
 use windmill_api_auth::ApiAuthed;
 
+// Partition-range backfill preview. The logic (producer resolution, range
+// enumeration, status join) is enterprise: the `private` build compiles the
+// EE module, the public build a stub that errors.
+#[cfg(feature = "private")]
+mod backfill_ee;
+#[cfg(feature = "private")]
+use backfill_ee as backfill;
+#[cfg(not(feature = "private"))]
+mod backfill_oss;
+#[cfg(not(feature = "private"))]
+use backfill_oss as backfill;
+
 pub fn workspaced_service() -> Router {
     Router::new()
         .route("/list", get(list_assets))
@@ -23,6 +35,7 @@ pub fn workspaced_service() -> Router {
         .route("/graph", get(asset_graph))
         .route("/pipelines", get(list_pipeline_folders))
         .route("/partitions", get(list_partitions))
+        .route("/partitions_in_range", get(list_partitions_in_range))
         .route("/asset_schemas", get(list_asset_schemas))
         .route("/record_materialization", post(record_materialization))
 }
@@ -54,6 +67,55 @@ async fn list_partitions(
     Ok(Json(rows))
 }
 
+// Only the EE `backfill` module reads the fields; the OSS stub errors without
+// touching them.
+#[cfg_attr(not(feature = "private"), allow(dead_code))]
+#[derive(Deserialize)]
+struct PartitionsInRangeQuery {
+    // The materialized ducklake asset path (`<ducklake>/<table>`).
+    path: String,
+    // Inclusive calendar-day range (YYYY-MM-DD), local to the producer's
+    // partition tz.
+    from: chrono::NaiveDate,
+    to: chrono::NaiveDate,
+}
+
+#[derive(Serialize)]
+struct PartitionInRange {
+    partition: String,
+    // `missing` | `running` | `materialized` | `failed` — `missing` means no
+    // materialization was ever recorded for the slice.
+    status: &'static str,
+}
+
+#[derive(Serialize)]
+struct PartitionsInRangeResponse {
+    // The pipeline script that materializes the asset (managed `// materialize`
+    // target, or a partitioned writer using the SDK helpers) — the runnable a
+    // backfill launches (with an explicit `partition` arg per slice).
+    producer_path: String,
+    partition_kind: String,
+    partitions: Vec<PartitionInRange>,
+}
+
+// Backfill range preview: every partition the producer's `// partitioned` spec
+// expects in `[from, to]`, joined with what `materialized_partition` records —
+// the missing/failed subset is the backfill worklist. The logic is in the
+// `backfill` module pair: EE resolves and enumerates, the OSS stub errors
+// (single-partition runs stay available everywhere; fanning out over a range
+// is enterprise).
+async fn list_partitions_in_range(
+    authed: ApiAuthed,
+    Path(w_id): Path<String>,
+    Extension(user_db): Extension<UserDB>,
+    Query(q): Query<PartitionsInRangeQuery>,
+) -> JsonResult<PartitionsInRangeResponse> {
+    let mut tx = user_db.begin(&authed).await?;
+    let res = backfill::partitions_in_range(&mut tx, &w_id, &q).await?;
+    tx.commit().await?;
+    Ok(Json(res))
+}
+
 // Per-asset captured output schema versions for a ducklake asset (gap #2a) —
 // the schema-evolution history persisted after each managed `// materialize`.
 // Newest version first; materialization targets are ducklake-only in v1, so the

backend/windmill-api/Cargo.toml

@@ -10,7 +10,7 @@ path = "src/lib.rs"
 
 [features]
 default = []
-private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private"]
+private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-assets/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private"]
 enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-api-scripts/enterprise", "windmill-api-flows/enterprise", "windmill-api-users/enterprise", "windmill-api-workspaces/enterprise", "windmill-api-groups/enterprise", "windmill-api-configs/enterprise", "windmill-api-settings/enterprise", "windmill-api-schedule/enterprise", "windmill-api-agent-workers?/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-azure?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "license"]
 stripe = []
 run_inline = ["dep:windmill-worker", "windmill-api-configs/run_inline"]

backend/windmill-api/openapi.yaml

@@ -21498,6 +21498,42 @@ paths:
                 items:
                   $ref: "#/components/schemas/MaterializedPartition"
 
+  /w/{workspace}/assets/partitions_in_range:
+    get:
+      summary: List expected partitions of a ducklake asset in a date range with their materialization status (enterprise)
+      operationId: listAssetPartitionsInRange
+      tags:
+        - asset
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: path
+          in: query
+          required: true
+          description: The materialized ducklake asset path (`<ducklake>/<table>`)
+          schema:
+            type: string
+        - name: from
+          in: query
+          required: true
+          description: Inclusive range start (YYYY-MM-DD), local to the producer's partition tz
+          schema:
+            type: string
+            format: date
+        - name: to
+          in: query
+          required: true
+          description: Inclusive range end (YYYY-MM-DD), local to the producer's partition tz
+          schema:
+            type: string
+            format: date
+      responses:
+        "200":
+          description: expected partitions in range with per-slice status — the missing/failed subset is the backfill worklist
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/PartitionsInRange"
+
   /w/{workspace}/assets/asset_schemas:
     get:
       summary: List captured output-schema versions for a ducklake asset
@@ -29692,6 +29728,27 @@ components:
         error:
           type: string
           nullable: true
+    PartitionsInRange:
+      type: object
+      required: [producer_path, partition_kind, partitions]
+      properties:
+        producer_path:
+          type: string
+          description: the pipeline script that materializes the asset (managed `// materialize` target, or a partitioned writer using the SDK helpers) — the runnable a backfill launches
+        partition_kind:
+          type: string
+          enum: [daily, hourly, weekly, monthly, dynamic]
+        partitions:
+          type: array
+          items:
+            type: object
+            required: [partition, status]
+            properties:
+              partition:
+                type: string
+              status:
+                type: string
+                enum: [missing, running, materialized, failed]
     AssetSchemaVersion:
       type: object
       required: [version, columns, captured_at]

frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte

@@ -2,78 +2,217 @@
 	import { Button } from '$lib/components/common'
 	import Modal from '$lib/components/common/modal/Modal.svelte'
 	import DateInput from '$lib/components/DateInput.svelte'
+	import Toggle from '$lib/components/Toggle.svelte'
+	import { Loader2 } from 'lucide-svelte'
+	import { resource } from 'runed'
+	import { AssetService } from '$lib/gen'
 	import { enterpriseLicense } from '$lib/stores'
+	import type { BackfillSliceState } from './backfillRun'
 
-	// Backfill re-runs the CE materialization once per partition in [from, to].
-	// It is an enterprise feature (orchestration over a range); the dialog is
-	// only reachable when licensed, but we guard here too so the action can
-	// never fire in CE.
+	// Range backfill of a partitioned ducklake asset (enterprise). The dialog
+	// previews the expected partitions in [from, to] against what
+	// `materialized_partition` records (the missing/failed set is the
+	// worklist), then hands the worklist to the parent, which runs the
+	// producer once per slice (see `backfillRun.ts`). Progress state lives in
+	// the parent so a run survives closing the dialog.
 	interface Props {
 		// Controlled by the parent. `$bindable()` without a default per the
 		// AGENTS.md ban on `$bindable(default)` for optional props.
 		open?: boolean
 		assetPath: string
-		// Invoked with the inclusive ISO date range; the parent performs the
-		// actual fan-out (EE backfill endpoint).
-		onBackfill: (from: string, to: string) => Promise<void>
+		workspace: string
+		// Live per-slice states of the in-flight (or last finished) backfill.
+		slices?: BackfillSliceState[]
+		running?: boolean
+		// Cancellation was requested but the in-flight slice hasn't finished yet.
+		cancelRequested?: boolean
+		onStart: (producerPath: string, partitions: string[]) => void
+		onCancel: () => void
+		// Clear the finished run so the range picker shows again.
+		onReset: () => void
 	}
 
-	let { open = $bindable(), assetPath, onBackfill }: Props = $props()
+	let {
+		open = $bindable(),
+		assetPath,
+		workspace,
+		slices,
+		running,
+		cancelRequested,
+		onStart,
+		onCancel,
+		onReset
+	}: Props = $props()
 
 	let fromDate = $state<string | undefined>(undefined)
 	let toDate = $state<string | undefined>(undefined)
-	let loading = $state(false)
-	let error = $state<string | undefined>(undefined)
+	let onlyMissing = $state(true)
 
-	let canSubmit = $derived(!!$enterpriseLicense && !!fromDate && !!toDate && !loading)
-
-	async function submit() {
-		if (!fromDate || !toDate) return
-		loading = true
-		error = undefined
-		try {
-			await onBackfill(fromDate, toDate)
-			open = false
-		} catch (e) {
-			error = e instanceof Error ? e.message : String(e)
-		} finally {
-			loading = false
+	let preview = resource(
+		[() => workspace, () => assetPath, () => fromDate, () => toDate, () => open ?? false],
+		async ([ws, path, from, to, isOpen]) => {
+			if (!isOpen || !ws || !path || !from || !to) return undefined
+			return await AssetService.listAssetPartitionsInRange({ workspace: ws, path, from, to })
 		}
+	)
+
+	let worklist = $derived.by(() => {
+		const parts = preview.current?.partitions ?? []
+		const picked = onlyMissing
+			? parts.filter((p) => p.status === 'missing' || p.status === 'failed')
+			: parts
+		return picked.map((p) => p.partition)
+	})
+	let counts = $derived.by(() => {
+		const c = { missing: 0, materialized: 0, failed: 0, running: 0 }
+		for (const p of preview.current?.partitions ?? []) c[p.status]++
+		return c
+	})
+	let canStart = $derived(
+		!!$enterpriseLicense && !running && !preview.loading && worklist.length > 0
+	)
+
+	// Backend errors come back as a plain-text body (windmill_common::error).
+	function errText(e: unknown): string {
+		const err = e as { body?: unknown; message?: string }
+		if (typeof err?.body === 'string' && err.body) return err.body
+		return err?.message ?? String(e)
+	}
+
+	const previewChipClass: Record<string, string> = {
+		missing: 'bg-amber-100 text-amber-800 dark:bg-amber-900/40 dark:text-amber-300',
+		failed: 'bg-red-100 text-red-800 dark:bg-red-900/40 dark:text-red-300',
+		materialized: 'bg-green-100 text-green-800 dark:bg-green-900/40 dark:text-green-300',
+		running: 'bg-blue-100 text-blue-800 dark:bg-blue-900/40 dark:text-blue-300'
+	}
+	const sliceChipClass: Record<BackfillSliceState['status'], string> = {
+		pending: 'bg-surface-secondary text-tertiary',
+		running: 'bg-blue-100 text-blue-800 dark:bg-blue-900/40 dark:text-blue-300',
+		success: 'bg-green-100 text-green-800 dark:bg-green-900/40 dark:text-green-300',
+		failure: 'bg-red-100 text-red-800 dark:bg-red-900/40 dark:text-red-300'
 	}
 </script>
 
-<Modal bind:open={() => open ?? false, (v) => (open = v)} title={`Backfill ${assetPath}`}>
+<Modal
+	bind:open={() => open ?? false, (v) => (open = v)}
+	title={`Backfill ${assetPath}`}
+	cancelText="Close"
+>
 	<div class="flex flex-col gap-4">
 		{#if !$enterpriseLicense}
 			<p class="text-sm text-secondary">
 				Partition backfill is an enterprise feature. Materializing a single partition is available
 				in the open-source edition; reprocessing a historical range requires an enterprise license.
 			</p>
+		{:else if slices?.length}
+			<!-- A run is in flight (or just finished): show per-slice progress. -->
+			<p class="text-sm text-secondary">
+				{#if running && cancelRequested}
+					Cancelling — waiting for the current partition to finish; the rest will not run.
+				{:else if running}
+					Materializing {slices.filter((s) => s.status === 'success' || s.status === 'failure')
+						.length}/{slices.length} partitions sequentially — each run gets its partition as an explicit
+					arg.
+				{:else}
+					Backfill finished: {slices.filter((s) => s.status === 'success').length} succeeded,
+					{slices.filter((s) => s.status === 'failure').length} failed,
+					{slices.filter((s) => s.status === 'pending').length} not run.
+				{/if}
+			</p>
+			<div class="flex flex-col gap-1 max-h-64 overflow-auto">
+				{#each slices as s (s.partition)}
+					<div class="flex items-center gap-2 text-xs">
+						<span class="px-1.5 py-0.5 rounded text-3xs font-medium {sliceChipClass[s.status]}">
+							{s.status}
+						</span>
+						<span class="font-mono">{s.partition}</span>
+						{#if s.status === 'running'}
+							<Loader2 size={12} class="animate-spin text-tertiary" />
+						{/if}
+						{#if s.error}
+							<span class="text-3xs text-red-600 truncate" title={s.error}>{s.error}</span>
+						{/if}
+					</div>
+				{/each}
+			</div>
 		{:else}
 			<p class="text-sm text-secondary">
-				Re-runs the materialization for each partition in the range. Re-running a partition is
-				idempotent, so this is safe to repeat.
+				Re-runs the producing script once per partition in the range, each with an explicit
+				<span class="font-mono">partition</span> arg. Re-running a partition is idempotent, so this is
+				safe to repeat.
 			</p>
-			<div class="flex gap-3">
+			<div class="flex gap-3 items-end">
 				<div class="flex flex-col gap-1">
 					<span class="text-xs font-semibold">From</span>
-					<DateInput bind:value={fromDate} />
+					<DateInput bind:value={fromDate} dateFormat="yyyy-MM-dd" />
 				</div>
 				<div class="flex flex-col gap-1">
 					<span class="text-xs font-semibold">To</span>
-					<DateInput bind:value={toDate} />
+					<DateInput bind:value={toDate} dateFormat="yyyy-MM-dd" />
 				</div>
 			</div>
-			{#if error}
-				<p class="text-sm text-red-600">{error}</p>
+			{#if preview.loading}
+				<div class="flex items-center gap-2 text-tertiary text-xs">
+					<Loader2 size={14} class="animate-spin" /> Computing partitions in range…
+				</div>
+			{:else if preview.error}
+				<p class="text-sm text-red-600">{errText(preview.error)}</p>
+			{:else if preview.current}
+				<div class="flex flex-col gap-2">
+					<p class="text-xs text-secondary">
+						{preview.current.partitions.length}
+						{preview.current.partition_kind} partitions in range — {counts.missing} missing,
+						{counts.failed} failed, {counts.materialized} materialized. Producer:
+						<span class="font-mono">{preview.current.producer_path}</span>
+					</p>
+					<div class="flex flex-wrap gap-1 max-h-40 overflow-auto">
+						{#each preview.current.partitions as p (p.partition)}
+							<span
+								class="px-1.5 py-0.5 rounded text-3xs font-mono {previewChipClass[p.status]}"
+								title={p.status}
+							>
+								{p.partition}
+							</span>
+						{/each}
+					</div>
+					<Toggle
+						bind:checked={onlyMissing}
+						size="xs"
+						options={{ right: 'Only run missing and failed partitions' }}
+					/>
+				</div>
 			{/if}
 		{/if}
 	</div>
 
 	{#snippet actions()}
-		<Button variant="subtle" onclick={() => (open = false)}>Cancel</Button>
-		<Button variant="accent" disabled={!canSubmit} {loading} onclick={submit}>
-			Start backfill
-		</Button>
+		{#if running}
+			<Button variant="default" disabled={cancelRequested} onclick={onCancel}>
+				{cancelRequested ? 'Cancelling…' : 'Cancel backfill'}
+			</Button>
+		{:else if slices?.length}
+			<!-- The finished run just changed partition statuses — refresh the
+			     range preview along with clearing the run. -->
+			<Button
+				variant="default"
+				onclick={() => {
+					onReset()
+					preview.refetch()
+				}}
+			>
+				New backfill
+			</Button>
+		{:else}
+			<Button
+				variant="accent"
+				disabled={!canStart}
+				onclick={() => {
+					const producer = preview.current?.producer_path
+					if (producer) onStart(producer, worklist)
+				}}
+			>
+				Backfill {worklist.length} partition{worklist.length === 1 ? '' : 's'}
+			</Button>
+		{/if}
 	{/snippet}
 </Modal>

frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte

@@ -1,10 +1,12 @@
 <script lang="ts">
 	import { resource } from 'runed'
-	import { OpenAPI, AssetService, type MaterializedPartition } from '$lib/gen'
+	import { AssetService, JobService, type MaterializedPartition } from '$lib/gen'
 	import { Button } from '$lib/components/common'
 	import { Loader2, RefreshCw, History } from 'lucide-svelte'
-	import { sendUserToast } from '$lib/utils'
+	import { enterpriseLicense } from '$lib/stores'
 	import BackfillRangeDialog from './BackfillRangeDialog.svelte'
+	import { runBackfill, type BackfillSliceState } from './backfillRun'
+	import { makeWaitJobTerminal } from './cascadeRun'
 
 	interface Props {
 		// The materialized ducklake asset path (`<ducklake>/<table>`).
@@ -19,59 +21,131 @@
 	})
 
 	let backfillOpen = $state(false)
+	// Slice states of the in-flight (or last finished) backfill. Owned here —
+	// not in the dialog — so progress keeps streaming in the header when the
+	// dialog is closed mid-run.
+	let backfillSlices = $state<BackfillSliceState[] | undefined>(undefined)
+	let backfillRunning = $state(false)
+	let backfillCancelRequested = $state(false)
+
+	let backfillDone = $derived(
+		backfillSlices?.filter((s) => s.status === 'success' || s.status === 'failure').length ?? 0
+	)
+	let backfillFailed = $derived(backfillSlices?.filter((s) => s.status === 'failure').length ?? 0)
+
+	// Never throws: the job may already be terminal, and the sequential loop
+	// stops on the cancel flag regardless.
+	async function cancelJobById(jobId: string) {
+		try {
+			await JobService.cancelQueuedJob({
+				workspace,
+				id: jobId,
+				requestBody: { reason: 'backfill cancelled' }
+			})
+		} catch {}
+	}
+
+	// Stop scheduling further slices and cancel the in-flight run (its slice
+	// then reports failure); already-materialized slices are unaffected. A
+	// cancel that lands while a launch is in flight (no job id yet) is
+	// finished by the runner via its `cancelJob` hook.
+	async function cancelBackfill() {
+		backfillCancelRequested = true
+		const running = backfillSlices?.find((s) => s.status === 'running')
+		if (running?.jobId) {
+			await cancelJobById(running.jobId)
+		}
+	}
+
+	async function startBackfill(producerPath: string, worklist: string[]) {
+		if (backfillRunning || worklist.length === 0) return
+		backfillRunning = true
+		backfillCancelRequested = false
+		try {
+			await runBackfill({
+				partitions: worklist,
+				// Backend asset dispatch stays ON (unlike client-orchestrated dev
+				// cascades, which pass `_wmill_skip_asset_dispatch`): a backfilled
+				// slice refreshes its deployed consumers like any deployed run,
+				// carrying its partition down the chain.
+				launch: async (partition) =>
+					await JobService.runScriptByPath({
+						workspace,
+						path: producerPath,
+						requestBody: { partition }
+					}),
+				waitTerminal: async (jobId) => {
+					const term = await makeWaitJobTerminal(workspace)(jobId)
+					// The run just recorded (or failed to record) its
+					// materialized_partition row — stream it into the grid.
+					partitions.refetch()
+					return term
+				},
+				onUpdate: (s) => (backfillSlices = s),
+				isCancelled: () => backfillCancelRequested,
+				cancelJob: cancelJobById
+			})
+		} finally {
+			backfillRunning = false
+			partitions.refetch()
+		}
+	}
 
 	const statusClass: Record<MaterializedPartition['status'], string> = {
 		materialized: 'bg-green-100 text-green-800 dark:bg-green-900/40 dark:text-green-300',
 		running: 'bg-blue-100 text-blue-800 dark:bg-blue-900/40 dark:text-blue-300',
 		failed: 'bg-red-100 text-red-800 dark:bg-red-900/40 dark:text-red-300'
 	}
-
-	async function onBackfill(from: string, to: string) {
-		// The fan-out runner is an enterprise feature; the dialog only submits
-		// when licensed. This posts the intent to the (EE) backfill endpoint — a raw
-		// fetch since it's not in the OSS OpenAPI, so carry the bearer token
-		// explicitly (matches how `/pipeline_dev` authenticates: no session cookie).
-		const token = OpenAPI.TOKEN
-		const authHeader: Record<string, string> =
-			typeof token === 'string' && token ? { Authorization: `Bearer ${token}` } : {}
-		const res = await fetch(`${OpenAPI.BASE ?? ''}/w/${workspace}/assets/backfill`, {
-			method: 'POST',
-			credentials: 'include',
-			headers: { 'content-type': 'application/json', ...authHeader },
-			body: JSON.stringify({ path, from, to })
-		})
-		if (!res.ok) throw new Error(`backfill → ${res.status}`)
-		sendUserToast(`Backfill queued for ${path} (${from} → ${to})`)
-		await partitions.refetch()
-	}
 </script>
 
 <div class="flex flex-col h-full">
 	<div class="flex items-center justify-between gap-2 px-3 py-2 border-b shrink-0">
 		<span class="text-xs font-semibold text-secondary">Materialized partitions</span>
-		<div class="flex items-center gap-1">
-			<Button
-				variant="subtle"
-				unifiedSize="sm"
-				startIcon={{ icon: RefreshCw }}
-				iconOnly
-				onclick={() => partitions.refetch()}
-				title="Refresh"
-			/>
-			<!-- The backfill range runner (POST /assets/backfill) is a planned
-			     enterprise follow-up and not yet implemented on the backend, so the
-			     button stays disabled — enabling it would 404. Re-enable when the
-			     endpoint lands. -->
-			<Button
-				variant="default"
-				unifiedSize="sm"
-				startIcon={{ icon: History }}
-				disabled
-				onclick={() => (backfillOpen = true)}
-				title="Backfill a range of partitions — coming soon (enterprise)"
-			>
-				Backfill
-			</Button>
+		<div class="flex items-center gap-2">
+			{#if backfillRunning}
+				<button
+					class="flex items-center gap-1 text-3xs text-tertiary hover:text-primary"
+					onclick={() => (backfillOpen = true)}
+					title="Show backfill progress"
+				>
+					<Loader2 size={12} class="animate-spin" />
+					Backfilling {backfillDone}/{backfillSlices?.length ?? 0}
+				</button>
+			{:else if backfillSlices?.length}
+				<button
+					class="text-3xs {backfillFailed > 0
+						? 'text-red-600'
+						: 'text-tertiary'} hover:text-primary"
+					onclick={() => (backfillOpen = true)}
+					title="Show backfill result"
+				>
+					Backfill: {backfillDone - backfillFailed}/{backfillSlices.length} ok{backfillFailed > 0
+						? `, ${backfillFailed} failed`
+						: ''}
+				</button>
+			{/if}
+			<div class="flex items-center gap-1">
+				<Button
+					variant="subtle"
+					unifiedSize="sm"
+					startIcon={{ icon: RefreshCw }}
+					iconOnly
+					onclick={() => partitions.refetch()}
+					title="Refresh"
+				/>
+				<Button
+					variant="default"
+					unifiedSize="sm"
+					startIcon={{ icon: History }}
+					disabled={!$enterpriseLicense}
+					onclick={() => (backfillOpen = true)}
+					title={$enterpriseLicense
+						? 'Backfill a range of partitions'
+						: 'Backfill a range of partitions (enterprise feature)'}
+				>
+					Backfill
+				</Button>
+			</div>
 		</div>
 	</div>
 
@@ -122,4 +196,14 @@
 	</div>
 </div>
 
-<BackfillRangeDialog bind:open={backfillOpen} assetPath={path} {onBackfill} />
+<BackfillRangeDialog
+	bind:open={backfillOpen}
+	assetPath={path}
+	{workspace}
+	slices={backfillSlices}
+	running={backfillRunning}
+	cancelRequested={backfillCancelRequested}
+	onStart={startBackfill}
+	onCancel={cancelBackfill}
+	onReset={() => (backfillSlices = undefined)}
+/>

frontend/src/lib/components/assets/AssetGraph/backfillRun.test.ts

@@ -0,0 +1,118 @@
+import { describe, expect, it } from 'vitest'
+import { runBackfill, type BackfillSliceState } from './backfillRun'
+
+// Deterministic fake backend: launch resolves with `job:<partition>`,
+// waitTerminal resolves per the `results` table (default success), recording
+// launch order.
+function fakeRunner(results: Record<string, 'success' | 'failure'> = {}) {
+	const launched: string[] = []
+	return {
+		launched,
+		launch: async (partition: string) => {
+			launched.push(partition)
+			return `job:${partition}`
+		},
+		waitTerminal: async (jobId: string) => results[jobId.slice(4)] ?? ('success' as const)
+	}
+}
+
+describe('runBackfill', () => {
+	it('runs slices sequentially in order and reports ok', async () => {
+		const r = fakeRunner()
+		const res = await runBackfill({
+			partitions: ['2026-06-26', '2026-06-27', '2026-06-29'],
+			launch: r.launch,
+			waitTerminal: r.waitTerminal
+		})
+		expect(r.launched).toEqual(['2026-06-26', '2026-06-27', '2026-06-29'])
+		expect(res.ok).toBe(true)
+		expect(res.cancelled).toBe(false)
+		expect(res.slices.map((s) => s.status)).toEqual(['success', 'success', 'success'])
+		expect(res.slices.map((s) => s.jobId)).toEqual([
+			'job:2026-06-26',
+			'job:2026-06-27',
+			'job:2026-06-29'
+		])
+	})
+
+	it('continues past a failed slice — each slice is independent', async () => {
+		const r = fakeRunner({ '2026-06-27': 'failure' })
+		const res = await runBackfill({
+			partitions: ['2026-06-26', '2026-06-27', '2026-06-29'],
+			launch: r.launch,
+			waitTerminal: r.waitTerminal
+		})
+		expect(r.launched).toEqual(['2026-06-26', '2026-06-27', '2026-06-29'])
+		expect(res.ok).toBe(false)
+		expect(res.slices.map((s) => s.status)).toEqual(['success', 'failure', 'success'])
+	})
+
+	it('records a launch error as slice failure and keeps going', async () => {
+		const r = fakeRunner()
+		const res = await runBackfill({
+			partitions: ['a', 'b'],
+			launch: async (p) => {
+				if (p === 'a') throw new Error('boom')
+				return r.launch(p)
+			},
+			waitTerminal: r.waitTerminal
+		})
+		expect(res.slices[0]).toMatchObject({ status: 'failure', error: 'boom' })
+		expect(res.slices[1].status).toBe('success')
+	})
+
+	it('stops before the next launch when cancelled, leaving the rest pending', async () => {
+		const r = fakeRunner()
+		let done = 0
+		const res = await runBackfill({
+			partitions: ['a', 'b', 'c'],
+			launch: r.launch,
+			waitTerminal: async (id) => {
+				done++
+				return r.waitTerminal(id)
+			},
+			isCancelled: () => done >= 1
+		})
+		expect(r.launched).toEqual(['a'])
+		expect(res.cancelled).toBe(true)
+		expect(res.ok).toBe(false)
+		expect(res.slices.map((s) => s.status)).toEqual(['success', 'pending', 'pending'])
+	})
+
+	it('cancels a job whose launch raced the cancellation', async () => {
+		let cancelled = false
+		const cancelledJobs: string[] = []
+		const res = await runBackfill({
+			partitions: ['a', 'b'],
+			launch: async (p) => {
+				// The user clicks cancel while the launch request is in flight —
+				// there is no job id to cancel yet.
+				cancelled = true
+				return `job:${p}`
+			},
+			waitTerminal: async () => 'failure',
+			isCancelled: () => cancelled,
+			cancelJob: async (id) => {
+				cancelledJobs.push(id)
+			}
+		})
+		expect(cancelledJobs).toEqual(['job:a'])
+		expect(res.cancelled).toBe(true)
+		expect(res.slices.map((s) => s.status)).toEqual(['failure', 'pending'])
+	})
+
+	it('emits a snapshot per transition, never mutating earlier snapshots', async () => {
+		const r = fakeRunner()
+		const snapshots: BackfillSliceState[][] = []
+		await runBackfill({
+			partitions: ['a'],
+			launch: r.launch,
+			waitTerminal: r.waitTerminal,
+			onUpdate: (s) => snapshots.push(s)
+		})
+		// initial pending, running, running+jobId, terminal
+		expect(snapshots.map((s) => s[0].status)).toEqual(['pending', 'running', 'running', 'success'])
+		expect(snapshots[1][0].jobId).toBeUndefined()
+		expect(snapshots[2][0].jobId).toBe('job:a')
+	})
+})

frontend/src/lib/components/assets/AssetGraph/backfillRun.ts

@@ -0,0 +1,83 @@
+// Client-side orchestration of a partition-range backfill (enterprise): one
+// deployed run of the producing script per slice, launched with an explicit
+// `partition` arg (the worker only resolves a partition when the arg is
+// absent, so the caller-provided value wins). Slices run sequentially —
+// concurrent materializations of the same ducklake table would contend on
+// the catalog commit — and a failed slice does not stop the rest: each slice
+// is independent, and the missing/failed set is simply the next worklist.
+//
+// Pure module (no Svelte runes) so the loop is unit-testable; reactive
+// progress is delivered via `onUpdate` snapshots, mirroring
+// `cascadeOrchestrator.ts`.
+
+export type BackfillSliceStatus = 'pending' | 'running' | 'success' | 'failure'
+
+export type BackfillSliceState = {
+	partition: string
+	status: BackfillSliceStatus
+	jobId?: string
+	error?: string
+}
+
+export type BackfillRunOptions = {
+	/** Partition values to materialize, in run order. */
+	partitions: string[]
+	/** Launch one run of the producer with the given partition arg; returns the job id. */
+	launch: (partition: string) => Promise<string>
+	/** Resolve once the job reaches a terminal state. */
+	waitTerminal: (jobId: string) => Promise<'success' | 'failure'>
+	/** Snapshot of all slice states, emitted on every transition. */
+	onUpdate?: (slices: BackfillSliceState[]) => void
+	/** Checked before each launch; a true stop leaves the remaining slices 'pending'. */
+	isCancelled?: () => boolean
+	/**
+	 * Cancel a job whose launch raced the cancellation — the cancel click had
+	 * no job id to act on yet, so the loop cancels it as soon as the id
+	 * arrives. Must not throw (the job may already be terminal).
+	 */
+	cancelJob?: (jobId: string) => Promise<void>
+}
+
+export type BackfillRunResult = {
+	/** True when every slice ran and succeeded. */
+	ok: boolean
+	/** True when the loop stopped early on `isCancelled`. */
+	cancelled: boolean
+	slices: BackfillSliceState[]
+}
+
+export async function runBackfill(opts: BackfillRunOptions): Promise<BackfillRunResult> {
+	const { partitions, launch, waitTerminal, onUpdate, isCancelled, cancelJob } = opts
+	const slices: BackfillSliceState[] = partitions.map((partition) => ({
+		partition,
+		status: 'pending'
+	}))
+	const emit = () => onUpdate?.(slices.map((s) => ({ ...s })))
+	emit()
+	let cancelled = false
+	for (const slice of slices) {
+		if (isCancelled?.()) {
+			cancelled = true
+			break
+		}
+		slice.status = 'running'
+		emit()
+		try {
+			slice.jobId = await launch(slice.partition)
+			emit()
+			if (isCancelled?.() && cancelJob) {
+				await cancelJob(slice.jobId)
+			}
+			slice.status = await waitTerminal(slice.jobId)
+		} catch (e) {
+			slice.status = 'failure'
+			slice.error = e instanceof Error ? e.message : String(e)
+		}
+		emit()
+	}
+	return {
+		ok: !cancelled && slices.every((s) => s.status === 'success'),
+		cancelled,
+		slices
+	}
+}