diff --git a/backend/.sqlx/query-4add520575717d449a3758326777fee03012a3fa0fb943239e004cc4a7067501.json b/backend/.sqlx/query-4add520575717d449a3758326777fee03012a3fa0fb943239e004cc4a7067501.json new file mode 100644 index 0000000000..64330e91cf --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-d4f047b0a672d5bfe9fd7db60f11563ef05db7cc54a958a72384b08a69eb784e.json b/backend/.sqlx/query-d4f047b0a672d5bfe9fd7db60f11563ef05db7cc54a958a72384b08a69eb784e.json new file mode 100644 index 0000000000..3a5531b9a6 --- /dev/null +++ b/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" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 0ef135d961..5e76f395d2 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -bdd4ba4dfd05c8ca7db365e530812dc914517d7f +c3852ecb36bd0be1a74c63169e513888f3347850 diff --git a/backend/windmill-api-assets/Cargo.toml b/backend/windmill-api-assets/Cargo.toml index c6e7676b6b..c75715a3ac 100644 --- a/backend/windmill-api-assets/Cargo.toml +++ b/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 } diff --git a/backend/windmill-api-assets/src/backfill_oss.rs b/backend/windmill-api-assets/src/backfill_oss.rs new file mode 100644 index 0000000000..b5f0da1d58 --- /dev/null +++ b/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 { + Err(Error::BadRequest( + "Backfilling a range of partitions is an enterprise feature".to_string(), + )) +} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 29122bd793..19647c71d7 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/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 (`/`). + 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, +} + +// 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, + Extension(user_db): Extension, + Query(q): Query, +) -> JsonResult { + 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 diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 56893a8737..ebbc98e83e 100644 --- a/backend/windmill-api/Cargo.toml +++ b/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"] diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 0fcbb9132e..e7b5badb8c 100644 --- a/backend/windmill-api/openapi.yaml +++ b/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 (`/
`) + 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] diff --git a/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte b/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte index 1826ff6b41..e4390fe503 100644 --- a/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte +++ b/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 + 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(undefined) let toDate = $state(undefined) - let loading = $state(false) - let error = $state(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 = { + 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 = { + 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' } - open ?? false, (v) => (open = v)} title={`Backfill ${assetPath}`}> + open ?? false, (v) => (open = v)} + title={`Backfill ${assetPath}`} + cancelText="Close" +>
{#if !$enterpriseLicense}

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.

+ {:else if slices?.length} + +

+ {#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} +

+
+ {#each slices as s (s.partition)} +
+ + {s.status} + + {s.partition} + {#if s.status === 'running'} + + {/if} + {#if s.error} + {s.error} + {/if} +
+ {/each} +
{:else}

- 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 + partition arg. Re-running a partition is idempotent, so this is + safe to repeat.

-
+
From - +
To - +
- {#if error} -

{error}

+ {#if preview.loading} +
+ Computing partitions in range… +
+ {:else if preview.error} +

{errText(preview.error)}

+ {:else if preview.current} +
+

+ {preview.current.partitions.length} + {preview.current.partition_kind} partitions in range — {counts.missing} missing, + {counts.failed} failed, {counts.materialized} materialized. Producer: + {preview.current.producer_path} +

+
+ {#each preview.current.partitions as p (p.partition)} + + {p.partition} + + {/each} +
+ +
{/if} {/if}
{#snippet actions()} - - + {#if running} + + {:else if slices?.length} + + + {:else} + + {/if} {/snippet} diff --git a/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte b/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte index 4b7226e753..4be0bb8b6c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte @@ -1,10 +1,12 @@
Materialized partitions -
- +
+ {#if backfillRunning} + + {:else if backfillSlices?.length} + + {/if} +
+ +
@@ -122,4 +196,14 @@
- + (backfillSlices = undefined)} +/> diff --git a/frontend/src/lib/components/assets/AssetGraph/backfillRun.test.ts b/frontend/src/lib/components/assets/AssetGraph/backfillRun.test.ts new file mode 100644 index 0000000000..69fc6f9728 --- /dev/null +++ b/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:`, +// waitTerminal resolves per the `results` table (default success), recording +// launch order. +function fakeRunner(results: Record = {}) { + 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') + }) +}) diff --git a/frontend/src/lib/components/assets/AssetGraph/backfillRun.ts b/frontend/src/lib/components/assets/AssetGraph/backfillRun.ts new file mode 100644 index 0000000000..51fbc9664b --- /dev/null +++ b/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 + /** 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 +} + +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 { + 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 + } +}