diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs index ac3b7f624e..0e82fb0d70 100644 --- a/backend/windmill-api-integration-tests/tests/fork_background.rs +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -103,5 +103,18 @@ async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<( .await?; assert_eq!(resp.status(), 404); + // Without the flag the fork is created before the answer, which clients asking for a + // background fork recognise a server without it by. + sqlx::query("UPDATE workspace SET deleted = true WHERE id = 'wm-fork-bg-err'") + .execute(&db) + .await?; + let resp = client + .post(format!("{base_url}/create_fork")) + .header("Authorization", "Bearer SECRET_TOKEN") + .json(&json!({ "id": "wm-fork-sync", "name": "wm-fork-sync" })) + .send() + .await?; + assert_eq!(resp.text().await?, "Created forked workspace wm-fork-sync"); + Ok(()) } diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index e5d2fdf91f..304017ceda 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -9127,7 +9127,7 @@ async fn create_workspace_fork( parent_workspace_id, nw, dev_workspace_label, - ForkProgress(Some(progress_tx)), + ForkProgress { creation_id: Some(creation_id), steps: Some(progress_tx) }, )) .await .map_err(|e| Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}"))) @@ -9137,9 +9137,11 @@ async fn create_workspace_fork( tracing::error!("Creating fork '{fork_id}' failed: {e}"); e.to_string() }); + // A fork that committed is already recorded complete, even when its commit's + // acknowledgement was lost and the attempt reads as an error here. if let Err(e) = sqlx::query( "UPDATE workspace_fork_creation SET finished_at = now(), error = $3 - WHERE fork_workspace_id = $1 AND creation_id = $2", + WHERE fork_workspace_id = $1 AND creation_id = $2 AND finished_at IS NULL", ) .bind(&fork_id) .bind(creation_id) @@ -9172,13 +9174,17 @@ struct CreateWorkspaceForkQuery { background: Option, } -/// Where the copy of a fork created in the background is, for its status to report. +/// The attempt a fork created in the background belongs to, and where its copy is, for its status +/// to report. #[derive(Default)] -struct ForkProgress(Option>); +struct ForkProgress { + creation_id: Option, + steps: Option>, +} impl ForkProgress { fn step(&self, step: &'static str) { - if let Some(tx) = &self.0 { + if let Some(tx) = &self.steps { tx.send_replace(step); } } @@ -9202,10 +9208,7 @@ async fn get_fork_creation_status( ) -> JsonResult { let row = sqlx::query( "SELECT c.created_by, c.finished_at IS NOT NULL AS finished, c.error, c.step, - c.heartbeat_at < now() - $3 * interval '1 second' AS stale, - EXISTS(SELECT 1 FROM workspace w - WHERE w.id = c.fork_workspace_id - AND w.parent_workspace_id = c.parent_workspace_id) AS fork_exists + c.heartbeat_at < now() - $3 * interval '1 second' AS stale FROM workspace_fork_creation c WHERE c.creation_id = $1 AND c.parent_workspace_id = $2", ) @@ -9221,11 +9224,7 @@ async fn get_fork_creation_status( ))); }; let error: Option = row.get("error"); - // The fork commits whole, so once it exists it is complete: the run's own record can lag behind - // it when the server stopped between the commit and that record. - let status = if error.is_none() && row.get::("fork_exists") { - ForkCreationStatus { status: "completed", step: None, error: None } - } else if row.get::("finished") { + let status = if row.get::("finished") { match error { Some(error) => ForkCreationStatus { status: "failed", step: None, error: Some(error) }, None => ForkCreationStatus { status: "completed", step: None, error: None }, @@ -9664,6 +9663,16 @@ async fn write_workspace_fork( (!copied.is_empty()).then(|| [("copied_datatables", copied.as_str())].into()), ) .await?; + // Committed with the fork, so its attempt reads complete exactly when the fork exists. + if let Some(creation_id) = progress.creation_id { + sqlx::query( + "UPDATE workspace_fork_creation SET finished_at = now(), error = NULL + WHERE creation_id = $1", + ) + .bind(creation_id) + .execute(&mut *tx) + .await?; + } tx.commit().await?; // A pre-creation lookup could have cached an EMPTY ancestor chain for this id, which diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 9f8ea077ce..e81df02120 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -45,7 +45,7 @@ export function syncWorkspaceTo(workspaceId: string | undefined): void { switchWorkspace(workspaceId) } import { WorkspaceService } from '$lib/gen' -import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation' +import { createWorkspaceForkAndWait, waitForForkCreation } from '$lib/utils/forkCreation' import { sendUserToast } from '$lib/toast' import { onUserChange } from '$lib/userScopedStorage' @@ -77,6 +77,9 @@ export type PendingFork = { id: string // Display name shown in the workspace bar. name: string + // The background creation this fork started, so a reload mid-copy waits for it rather than + // requesting the fork again, which the server refuses while that creation runs. + creation_id?: string } export type SessionSummarySource = 'placeholder' | 'generated' | 'manual' @@ -1181,7 +1184,10 @@ export async function commitSessionWorkspace( `Community edition is limited to ${CE_MAX_NON_ADMIN_WORKSPACES + 1} workspaces — archive a workspace or pick one to run in` ) } - const newId = await materializeFork(fork) + const newId = await materializeFork(fork, (creation_id) => { + s.pending_fork = { ...fork, creation_id } + void putSession(s) + }) if (!newId) { // Real failure (not a recovered duplicate). Drop the pending // fork so the session falls through to the workspace-pick @@ -1298,10 +1304,27 @@ export function setGeneratedSessionSummary( // lost). Adopt it silently instead of re-POSTing — the API would // otherwise reject with workspace_pkey. Likewise, if the API returns a // duplicate-key error we refresh the store and adopt the existing row. -export async function materializeFork(fork: PendingFork): Promise { +export async function materializeFork( + fork: PendingFork, + onCreationStarted?: (creationId: string) => void +): Promise { if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id try { - await createWorkspaceForkAndWait(fork.parent_workspace_id, { id: fork.id, name: fork.name }) + let resumed = false + if (fork.creation_id) { + // An attempt that failed or is gone is requested again below. + resumed = await waitForForkCreation(fork.parent_workspace_id, fork.creation_id).then( + () => true, + () => false + ) + } + if (!resumed) { + await createWorkspaceForkAndWait( + fork.parent_workspace_id, + { id: fork.id, name: fork.name }, + { onCreationStarted } + ) + } usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) sendUserToast(`Created fork ${fork.name}`) return fork.id diff --git a/frontend/src/lib/components/sessions/sessionState.test.ts b/frontend/src/lib/components/sessions/sessionState.test.ts index 5cf3e3269e..62d2b8d39f 100644 --- a/frontend/src/lib/components/sessions/sessionState.test.ts +++ b/frontend/src/lib/components/sessions/sessionState.test.ts @@ -37,6 +37,7 @@ vi.mock('$lib/gen', async (orig) => { WorkspaceService: { ...actual.WorkspaceService, createWorkspaceFork: vi.fn().mockRejectedValue(new Error('fork creation failed')), + getForkCreationStatus: vi.fn().mockResolvedValue({ status: 'completed' }), listUserWorkspaces: vi.fn().mockResolvedValue([]) } } @@ -108,6 +109,38 @@ describe('commitSessionWorkspace — fork-creation failure', () => { }) }) +describe('commitSessionWorkspace — fork creation resumed after a reload', () => { + it('waits for the creation the pending fork started instead of requesting it again', async () => { + const id = 'test-commit-fork-resume' + const prevLicense = get(enterpriseLicense) + enterpriseLicense.set('test-license') + vi.mocked(WorkspaceService.createWorkspaceFork).mockClear() + sessionState.sessions.push({ + id, + name: 'fork-resume', + createdAt: 0, + pending_fork: { + parent_workspace_id: 'parent_ws', + id: 'wm-fork-resume', + name: 'resume', + creation_id: '5c3e1b1e-0000-4000-8000-000000000000' + } + } as Session) + try { + expect(await commitSessionWorkspace(id, 'parent_ws')).toBe('wm-fork-resume') + expect(WorkspaceService.getForkCreationStatus).toHaveBeenCalledWith({ + workspace: 'parent_ws', + creationId: '5c3e1b1e-0000-4000-8000-000000000000' + }) + expect(WorkspaceService.createWorkspaceFork).not.toHaveBeenCalled() + } finally { + const i = sessionState.sessions.findIndex((x) => x.id === id) + if (i >= 0) sessionState.sessions.splice(i, 1) + enterpriseLicense.set(prevLicense) + } + }) +}) + // Set the workspace list to a given number of non-'admins' workspaces so the // commit-path's CE workspace-cap check (mirror of backend _check_nb_of_workspaces) // can be exercised. Returns a restore fn. diff --git a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte index 681f4fe15a..e6a0dce4fd 100644 --- a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte +++ b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte @@ -488,7 +488,7 @@ lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, copy_members: copyMembers }, - (step) => (forkCreationStep = step) + { onStep: (step) => (forkCreationStep = step) } ) } catch (e) { const msg = e?.body ?? e?.message ?? e ?? 'Unknown error' diff --git a/frontend/src/lib/utils/forkCreation.test.ts b/frontend/src/lib/utils/forkCreation.test.ts index e71d1b2a1e..bf180c6c6f 100644 --- a/frontend/src/lib/utils/forkCreation.test.ts +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -14,7 +14,7 @@ const svc = vi.mocked(WorkspaceService) const fork = { id: 'wm-fork-x', name: 'x' } async function run(onStep?: (step: string) => void) { - const p = createWorkspaceForkAndWait('ws', fork, onStep) + const p = createWorkspaceForkAndWait('ws', fork, { onStep }) const settled = p.then( () => 'completed', (e) => `failed: ${e.message}` diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index ae0d6fbe1a..d0fb4e5edf 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -9,12 +9,13 @@ const MAX_FAILING_POLLS_MS = 3 * 60 * 1000 * * 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. + * `onCreationStarted` receives the id `waitForForkCreation` resumes the wait by, and `onStep` the + * part of the copy the server reports it is in. */ export async function createWorkspaceForkAndWait( parentWorkspace: string, fork: CreateWorkspaceFork, - onStep?: (step: string) => void + opts: { onCreationStarted?: (creationId: string) => void; onStep?: (step: string) => void } = {} ): Promise { const response = await WorkspaceService.createWorkspaceFork({ workspace: parentWorkspace, @@ -23,7 +24,16 @@ export async function createWorkspaceForkAndWait( }) // A server without background forks ignores the flag and answers once the fork is created. if (response.startsWith('Created forked workspace')) return - const creationId = response + opts.onCreationStarted?.(response) + await waitForForkCreation(parentWorkspace, response, opts.onStep) +} + +/** Wait for a fork creation started in the background, by the id its request answered with. */ +export async function waitForForkCreation( + parentWorkspace: string, + creationId: string, + onStep?: (step: string) => void +): Promise { let failingSince: number | undefined while (true) { let result: Awaited> | undefined