From 1d52d20dee00cfb0100a603798b01ed35b9d25ea Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 29 Sep 2026 12:33:08 +0200 Subject: [PATCH] feat: show which part of the copy a fork being created is in The background copy reports its phase (data tables, settings, resources, scripts, flows, apps, drafts, triggers) through a watch channel. The heartbeat task records it on the fork's creation row as soon as it changes, and fork_creation_status returns it. The fork wizard shows it on its button, and the CLI logs each step. Co-Authored-By: Claude Opus 5.5 (1M context) --- ...60929092707_workspace_fork_creation.up.sql | 3 +- .../windmill-api-workspaces/src/workspaces.rs | 67 ++++++++++++++++--- backend/windmill-api/openapi.yaml | 3 + cli/src/commands/workspace/fork.ts | 5 ++ .../CreateWorkspaceInner.svelte | 34 ++++++---- frontend/src/lib/utils/forkCreation.test.ts | 12 ++-- frontend/src/lib/utils/forkCreation.ts | 5 +- 7 files changed, 100 insertions(+), 29 deletions(-) diff --git a/backend/migrations/20260929092707_workspace_fork_creation.up.sql b/backend/migrations/20260929092707_workspace_fork_creation.up.sql index 560dac6cf5..ac3218ca20 100644 --- a/backend/migrations/20260929092707_workspace_fork_creation.up.sql +++ b/backend/migrations/20260929092707_workspace_fork_creation.up.sql @@ -1,13 +1,14 @@ -- A fork being created in the background, and how that ended. The fork's own workspace row only -- exists once the copy commits, so this is where a failure is reported to whoever polls for it. -- `heartbeat_at` is refreshed while the copy runs: an unfinished row that stops being refreshed --- belongs to a server that went away mid-copy. +-- belongs to a server that went away mid-copy. `step` names the part of the copy it is in. CREATE TABLE workspace_fork_creation ( fork_workspace_id VARCHAR(50) PRIMARY KEY, parent_workspace_id VARCHAR(50) NOT NULL, created_by VARCHAR(255) NOT NULL, started_at TIMESTAMPTZ NOT NULL DEFAULT now(), heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), + step TEXT, finished_at TIMESTAMPTZ, error TEXT ); diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 70b7293dcf..3c4220a9eb 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -6667,7 +6667,9 @@ async fn clone_workspace_data( source_workspace_id: &str, target_workspace_id: &str, authed: &ApiAuthed, + progress: &ForkProgress, ) -> Result<()> { + progress.step("Copying settings, folders and groups"); // Clone workspace settings (merge with existing basic settings) update_workspace_settings(tx, source_workspace_id, target_workspace_id).await?; @@ -6689,6 +6691,7 @@ async fn clone_workspace_data( // Clone groups clone_groups(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying resources and variables"); // Clone resource types clone_resource_types(tx, source_workspace_id, target_workspace_id).await?; @@ -6698,6 +6701,7 @@ async fn clone_workspace_data( // Clone variables (including external secret backend replication) clone_variables(tx, db, source_workspace_id, target_workspace_id).await?; + progress.step("Copying scripts"); // Clone scripts with new hashes clone_scripts(tx, source_workspace_id, target_workspace_id).await?; @@ -6715,18 +6719,21 @@ async fn clone_workspace_data( clone_metric_catalog(tx, source_workspace_id, target_workspace_id).await?; clone_asset_usages_and_triggers(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying flows"); // Clone flows with new versions clone_flows(tx, source_workspace_id, target_workspace_id).await?; // Clone flow nodes clone_flow_nodes(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying apps"); // Clone apps with new IDs and app scripts let _app_id_mapping = clone_apps(tx, source_workspace_id, target_workspace_id, authed).await?; // Clone raw apps clone_raw_apps(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying drafts and dependencies"); // Clone the forker's own per-user drafts (plus the legacy NULL-email // workspace draft, if any) so they keep their pending edits in the // fork. Other users' drafts are intentionally NOT cloned — they don't @@ -9039,6 +9046,7 @@ async fn create_workspace_fork( parent_workspace_id, nw, dev_workspace_label, + ForkProgress::default(), )) .await .map_err(|e| { @@ -9060,7 +9068,8 @@ async fn create_workspace_fork( ON CONFLICT (fork_workspace_id) DO UPDATE SET parent_workspace_id = EXCLUDED.parent_workspace_id, created_by = EXCLUDED.created_by, - started_at = now(), heartbeat_at = now(), finished_at = NULL, error = NULL + started_at = now(), heartbeat_at = now(), step = NULL, finished_at = NULL, + error = NULL WHERE workspace_fork_creation.finished_at IS NOT NULL OR workspace_fork_creation.heartbeat_at < now() - $4 * interval '1 second' RETURNING started_at", @@ -9080,19 +9089,28 @@ async fn create_workspace_fork( }; BACKGROUND_FORKS.spawn(async move { + let (progress_tx, mut progress_rx) = tokio::sync::watch::channel(""); + // The one writer of the run's row while it copies: it records a new step as soon as the + // copy reports it, and refreshes the heartbeat in between. let heartbeat = { let (db, fork_id) = (db.clone(), fork_id.clone()); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(FORK_HEARTBEAT_SECS)); + let mut reporting = true; loop { - interval.tick().await; + tokio::select! { + _ = interval.tick() => {} + changed = progress_rx.changed(), if reporting => reporting = changed.is_ok(), + } + let step = *progress_rx.borrow_and_update(); if let Err(e) = sqlx::query( - "UPDATE workspace_fork_creation SET heartbeat_at = now() + "UPDATE workspace_fork_creation SET heartbeat_at = now(), step = $3 WHERE fork_workspace_id = $1 AND started_at = $2", ) .bind(&fork_id) .bind(started_at) + .bind((!step.is_empty()).then_some(step)) .execute(&db) .await { @@ -9107,6 +9125,7 @@ async fn create_workspace_fork( parent_workspace_id, nw, dev_workspace_label, + ForkProgress(Some(progress_tx)), )) .await .map_err(|e| Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}"))) @@ -9152,10 +9171,25 @@ struct CreateWorkspaceForkQuery { background: Option, } +/// Where the copy of a fork created in the background is, for its status to report. +#[derive(Default)] +struct ForkProgress(Option>); + +impl ForkProgress { + fn step(&self, step: &'static str) { + if let Some(tx) = &self.0 { + tx.send_replace(step); + } + } +} + #[derive(Serialize)] struct ForkCreationStatus { /// `running`, `completed` or `failed`. status: &'static str, + /// The part of the copy a running fork is in. + #[serde(skip_serializing_if = "Option::is_none")] + step: Option, #[serde(skip_serializing_if = "Option::is_none")] error: Option, } @@ -9166,7 +9200,7 @@ async fn get_fork_creation_status( Path((parent_workspace_id, fork_id)): Path<(String, String)>, ) -> JsonResult { let row = sqlx::query( - "SELECT created_by, finished_at IS NOT NULL AS finished, error, + "SELECT created_by, finished_at IS NOT NULL AS finished, error, step, heartbeat_at < now() - $3 * interval '1 second' AS stale FROM workspace_fork_creation WHERE fork_workspace_id = $1 AND parent_workspace_id = $2", @@ -9190,12 +9224,13 @@ async fn get_fork_creation_status( let error: Option = row.get("error"); let status = if row.get::("finished") { match error { - Some(error) => ForkCreationStatus { status: "failed", error: Some(error) }, - None => ForkCreationStatus { status: "completed", error: None }, + Some(error) => ForkCreationStatus { status: "failed", step: None, error: Some(error) }, + None => ForkCreationStatus { status: "completed", step: None, error: None }, } } else if row.get::("stale") { ForkCreationStatus { status: "failed", + step: None, error: Some( "the server creating the fork stopped before it finished; check the server logs \ and whether the fork exists before retrying" @@ -9203,7 +9238,7 @@ async fn get_fork_creation_status( ), } } else { - ForkCreationStatus { status: "running", error: None } + ForkCreationStatus { status: "running", step: row.get("step"), error: None } }; Ok(Json(status)) } @@ -9216,8 +9251,12 @@ async fn make_workspace_fork( parent_workspace_id: String, nw: CreateWorkspaceFork, dev_workspace_label: Option, + progress: ForkProgress, ) -> Result { let fork_id = nw.id.clone(); + if !nw.forked_datatables.is_empty() { + progress.step("Copying data tables"); + } let copies = crate::datatable_clone::make_copies( &db, &authed, @@ -9237,6 +9276,7 @@ async fn make_workspace_fork( nw, dev_workspace_label, &copies, + &progress, ) .await { @@ -9362,6 +9402,7 @@ async fn write_workspace_fork( nw: CreateWorkspaceFork, dev_workspace_label: Option, copies: &[crate::datatable_clone::MadeCopy], + progress: &ForkProgress, ) -> Result { let mut tx: Transaction<'_, Postgres> = db.begin().await?; // Held until this transaction ends: after an error, the copies' cleanup waits on it, so it reads @@ -9459,8 +9500,15 @@ async fn write_workspace_fork( .await?; // Clone all data from the parent workspace using Rust implementation - if let Err(e) = - clone_workspace_data(&mut tx, &db, &parent_workspace_id, &forked_id, &authed).await + if let Err(e) = clone_workspace_data( + &mut tx, + &db, + &parent_workspace_id, + &forked_id, + &authed, + progress, + ) + .await { // A genuine `\u0000` in a source `json` value (`app_version.value` / // `flow_version.schema`) aborts the clone when it is re-encoded to jsonb: @@ -9491,6 +9539,7 @@ async fn write_workspace_fork( // enabled=false. Disabled rows have no side effects (no listener // attaches, no cron fires) so this is safe by construction. The user // re-enables in the fork, with parent-conflict warnings on enable. + progress.step("Copying triggers and schedules"); clone_triggers_and_schedules(&mut tx, &parent_workspace_id, &forked_id).await?; repoint_unresolvable_cloned_identities(&mut tx, &forked_id, &authed).await?; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index e6c4be9968..4884b74636 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1445,6 +1445,9 @@ paths: status: type: string enum: [running, completed, failed] + step: + type: string + description: the part of the copy a running fork is in error: type: string required: diff --git a/cli/src/commands/workspace/fork.ts b/cli/src/commands/workspace/fork.ts index b9db16aa19..580112bbfb 100644 --- a/cli/src/commands/workspace/fork.ts +++ b/cli/src/commands/workspace/fork.ts @@ -30,6 +30,7 @@ import { */ async function waitForForkCreation(parentWorkspace: string, forkId: string) { let failedPolls = 0; + let lastStep: string | undefined; while (true) { let result: | Awaited> @@ -47,6 +48,10 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) { if (exists) return; if (++failedPolls >= 10) throw e; } + if (result?.status === "running" && result.step && result.step !== lastStep) { + lastStep = result.step; + log.info(colors.gray(`${lastStep}...`)); + } if (result?.status === "completed") return; if (result?.status === "failed") { throw new Error(result.error ?? "Unknown error"); diff --git a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte index a6916e53f9..681f4fe15a 100644 --- a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte +++ b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte @@ -330,6 +330,7 @@ } let forkCreationLoading = $state(false) + let forkCreationStep = $state('Creating branch') let forkCreationError = $state('') let errorMsgs: string[] = $state([]) let failedSyncJobs: string[] = $state([]) @@ -377,6 +378,7 @@ async function forkWorkspace(prefixed_id: string): Promise { if (baseWorkspaceId) { forkCreationLoading = true + forkCreationStep = 'Creating branch' errorMsgs = [] failedSyncJobs = [] forkCreationError = '' @@ -471,18 +473,23 @@ } try { - await createWorkspaceForkAndWait(baseWorkspaceId!, { - id: prefixed_id, - name, - color: colorEnabled && workspaceColor ? workspaceColor : undefined, - forked_datatables: forkedDatatables, - shared_ducklakes: forkDucklakeSection?.getSharedDucklakes() ?? [], - is_dev_workspace: createAsDevWorkspace, - dev_workspace_label: createAsDevWorkspace ? devWorkspaceLabel : undefined, - lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, - lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, - copy_members: copyMembers - }) + forkCreationStep = 'Creating fork' + await createWorkspaceForkAndWait( + baseWorkspaceId!, + { + id: prefixed_id, + name, + color: colorEnabled && workspaceColor ? workspaceColor : undefined, + forked_datatables: forkedDatatables, + shared_ducklakes: forkDucklakeSection?.getSharedDucklakes() ?? [], + is_dev_workspace: createAsDevWorkspace, + dev_workspace_label: createAsDevWorkspace ? devWorkspaceLabel : undefined, + lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, + lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, + copy_members: copyMembers + }, + (step) => (forkCreationStep = step) + ) } catch (e) { const msg = e?.body ?? e?.message ?? e ?? 'Unknown error' forkCreationError = `Failed to create fork '${prefixed_id}'` @@ -1114,7 +1121,8 @@ {:else} {/if} diff --git a/frontend/src/lib/utils/forkCreation.test.ts b/frontend/src/lib/utils/forkCreation.test.ts index a953b37600..a454560b8f 100644 --- a/frontend/src/lib/utils/forkCreation.test.ts +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -14,8 +14,8 @@ import { createWorkspaceForkAndWait } from './forkCreation' const svc = vi.mocked(WorkspaceService) const fork = { id: 'wm-fork-x', name: 'x' } -async function run() { - const p = createWorkspaceForkAndWait('ws', fork) +async function run(onStep?: (step: string) => void) { + const p = createWorkspaceForkAndWait('ws', fork, onStep) const settled = p.then( () => 'completed', (e) => `failed: ${e.message}` @@ -33,12 +33,14 @@ describe('createWorkspaceForkAndWait', () => { }) afterEach(() => vi.useRealTimers()) - it('waits while the fork is running and reports how it ended', async () => { + it('reports each step while the fork runs, then how it ended', async () => { svc.getForkCreationStatus .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValueOnce({ status: 'running', step: 'Copying flows' }) .mockResolvedValueOnce({ status: 'failed', error: 'boom' }) - expect(await run()).toBe('failed: boom') - expect(svc.getForkCreationStatus).toHaveBeenCalledTimes(2) + const steps: string[] = [] + expect(await run((s) => steps.push(s))).toBe('failed: boom') + expect(steps).toEqual(['Copying flows']) }) it('takes a failed poll for success once the fork exists', async () => { diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index d5b5e6e5b5..bd37b100a2 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -8,10 +8,12 @@ const MAX_FAILED_POLLS = 10 * * The copy runs on the server after the request returns: a large workspace takes longer to copy * than the timeout of many proxies in front of Windmill, which would otherwise cut the request. + * `onStep` receives the part of the copy the server reports it is in. */ export async function createWorkspaceForkAndWait( parentWorkspace: string, - fork: CreateWorkspaceFork + fork: CreateWorkspaceFork, + onStep?: (step: string) => void ): Promise { try { await WorkspaceService.createWorkspaceFork({ @@ -39,6 +41,7 @@ export async function createWorkspaceForkAndWait( if (await forkExists(fork.id)) return if (++failedPolls >= MAX_FAILED_POLLS) throw e } + if (result?.status === 'running' && result.step) onStep?.(result.step) if (result?.status === 'completed') return if (result?.status === 'failed') throw new Error(result.error ?? 'Unknown error') await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS))