From a6e2f7abf2a309264023f1187198f84df87049be Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 29 Sep 2026 14:43:14 +0200 Subject: [PATCH] fix: drop fork-creation joins and existence probes A second request for a fork being created is refused again; only the AI-session fork, whose request never varies, waits for its own earlier one. Clients recognise a server without background forks by its synchronous answer instead of probing for a workspace by id, which could name another parent's fork. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/fork_background.rs | 10 +---- .../windmill-api-workspaces/src/workspaces.rs | 17 -------- cli/src/commands/workspace/fork.ts | 39 ++++++++----------- .../sessions/sessionState.svelte.ts | 11 +++++- frontend/src/lib/utils/forkCreation.test.ts | 26 +++++++------ frontend/src/lib/utils/forkCreation.ts | 29 +++++++------- 6 files changed, 58 insertions(+), 74 deletions(-) diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs index 74d3c0e424..f9a13df9c8 100644 --- a/backend/windmill-api-integration-tests/tests/fork_background.rs +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -70,15 +70,9 @@ async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<( .await?; let resp = create("wm-fork-bg-err").await?; assert_eq!(resp.status(), 200, "{}", resp.text().await?); - // A retry by the same user joins the creation in flight; anyone else is refused. + // A second request is refused while the first is in flight, with the wording the AI-session + // fork matches to wait for its own earlier request. let resp = create("wm-fork-bg-err").await?; - assert_eq!(resp.status(), 200, "{}", resp.text().await?); - let resp = client - .post(format!("{base_url}/create_fork?background=true")) - .header("Authorization", "Bearer SECRET_TOKEN_2") - .json(&json!({ "id": "wm-fork-bg-err", "name": "wm-fork-bg-err" })) - .send() - .await?; assert_eq!(resp.status(), 400); assert!(resp.text().await?.contains("is already being created")); sqlx::query("INSERT INTO workspace (id, name, owner) VALUES ('wm-fork-bg-err', 'x', 'x')") diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index fbbb0e59e3..3bbaef20ca 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -9083,23 +9083,6 @@ async fn create_workspace_fork( // Every write of this run is scoped to its `started_at`: a run taken over as abandoned must // not report over the run that took its place. let Some(started_at) = claimed.map(|r| r.get::, _>("started_at")) else { - // A retry whose response was lost joins the creation it started; its status is polled the - // same way. Anyone else is refused rather than handed a fork of another request. - let same_request = sqlx::query_scalar::<_, bool>( - "SELECT EXISTS(SELECT 1 FROM workspace_fork_creation - WHERE fork_workspace_id = $1 AND parent_workspace_id = $2 AND created_by = $3 - AND finished_at IS NULL)", - ) - .bind(&fork_id) - .bind(&parent_workspace_id) - .bind(&authed.email) - .fetch_one(&db) - .await?; - if same_request { - return Ok(format!( - "Creating fork {response_fork_id} in the background" - )); - } return Err(Error::BadRequest(format!( "workspace '{fork_id}' is already being created" ))); diff --git a/cli/src/commands/workspace/fork.ts b/cli/src/commands/workspace/fork.ts index aaa05dba1b..b8041f5e7f 100644 --- a/cli/src/commands/workspace/fork.ts +++ b/cli/src/commands/workspace/fork.ts @@ -22,12 +22,7 @@ import { readConfigFile, } from "../../core/conf.ts"; -/** - * Wait for a fork requested with `background: true`. A poll that fails says nothing about the - * fork: it can be lost to a proxy, or reach a server without background forks (an older version, - * possibly one replica during a rolling deploy), which ignored the flag and created the fork - * before answering. - */ +/** Wait for the creation of a fork started in the background by this user. */ async function waitForForkCreation(parentWorkspace: string, forkId: string) { let failingSince: number | undefined; let lastStep: string | undefined; @@ -42,10 +37,8 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) { }); failingSince = undefined; } catch (e) { - const exists = await wmill - .existsWorkspace({ requestBody: { id: forkId } }) - .catch(() => false); - if (exists) return; + // A failed poll says nothing about the fork: it can be lost to a proxy, or reach a replica + // that has no status route yet during a rolling deploy. failingSince ??= Date.now(); if (Date.now() - failingSince >= 3 * 60 * 1000) throw e; } @@ -415,18 +408,20 @@ async function createWorkspaceFork( // --- Create the fork workspace --- try { - await wmill - .createWorkspaceFork({ - workspace: workspace.workspaceId, - background: true, - requestBody: { - id: trueWorkspaceId, - name: opts.createWorkspaceName ?? workspaceName ?? trueWorkspaceId, - color: forkColor, - forked_datatables: forkedDatatables, - }, - }); - await waitForForkCreation(workspace.workspaceId, trueWorkspaceId); + const response = await wmill.createWorkspaceFork({ + workspace: workspace.workspaceId, + background: true, + requestBody: { + id: trueWorkspaceId, + name: opts.createWorkspaceName ?? workspaceName ?? trueWorkspaceId, + color: forkColor, + forked_datatables: forkedDatatables, + }, + }); + // A server without background forks ignores the flag and answers once the fork is created. + if (!response.startsWith("Created forked workspace")) { + await waitForForkCreation(workspace.workspaceId, trueWorkspaceId); + } log.info(colors.green(`✅ Created forked workspace ${trueWorkspaceId}`)); } catch (error) { diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 9f8ea077ce..5fcf95babf 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' @@ -1301,7 +1301,14 @@ export function setGeneratedSessionSummary( export async function materializeFork(fork: PendingFork): 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 }) + try { + await createWorkspaceForkAndWait(fork.parent_workspace_id, { id: fork.id, name: fork.name }) + } catch (e: any) { + // This fork's request is always the same, so a creation of it already in flight (a + // lost response, a reload) is this one: the status only answers its own creator. + if (!String(e?.body ?? '').includes('is already being created')) throw e + await waitForForkCreation(fork.parent_workspace_id, fork.id) + } usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) sendUserToast(`Created fork ${fork.name}`) return fork.id diff --git a/frontend/src/lib/utils/forkCreation.test.ts b/frontend/src/lib/utils/forkCreation.test.ts index 4f83451cfa..cde82663b5 100644 --- a/frontend/src/lib/utils/forkCreation.test.ts +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -3,8 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('$lib/gen', () => ({ WorkspaceService: { createWorkspaceFork: vi.fn(), - getForkCreationStatus: vi.fn(), - existsWorkspace: vi.fn() + getForkCreationStatus: vi.fn() } })) @@ -28,8 +27,7 @@ describe('createWorkspaceForkAndWait', () => { beforeEach(() => { vi.useFakeTimers() vi.resetAllMocks() - svc.createWorkspaceFork.mockResolvedValue('started') - svc.existsWorkspace.mockResolvedValue(false) + svc.createWorkspaceFork.mockResolvedValue('Creating fork wm-fork-x in the background') }) afterEach(() => vi.useRealTimers()) @@ -43,17 +41,23 @@ describe('createWorkspaceForkAndWait', () => { expect(steps).toEqual(['Copying flows']) }) - it('takes a failed poll for success once the fork exists', async () => { - // A server without background forks created it synchronously and has no status route. - svc.getForkCreationStatus.mockRejectedValue( - Object.assign(new Error('Not Found'), { status: 404 }) - ) - svc.existsWorkspace.mockResolvedValue(true) + it('returns at once when the server created the fork synchronously', async () => { + // A server without background forks ignores the flag. + svc.createWorkspaceFork.mockResolvedValue('Created forked workspace wm-fork-x') + expect(await run()).toBe('completed') + expect(svc.getForkCreationStatus).not.toHaveBeenCalled() + }) + + it('keeps polling through failed polls', async () => { + svc.getForkCreationStatus + .mockRejectedValueOnce(Object.assign(new Error('Not Found'), { status: 404 })) + .mockRejectedValueOnce(new Error('upstream request timeout')) + .mockResolvedValueOnce({ status: 'completed' }) expect(await run()).toBe('completed') }) it('gives up once polls keep failing for a fork that never appears', async () => { svc.getForkCreationStatus.mockRejectedValue(new Error('upstream request timeout')) expect(await run()).toBe('failed: upstream request timeout') - }) + }) }) diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index 849684629f..1d2f511df0 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -16,25 +16,34 @@ export async function createWorkspaceForkAndWait( fork: CreateWorkspaceFork, onStep?: (step: string) => void ): Promise { - await WorkspaceService.createWorkspaceFork({ + const response = await WorkspaceService.createWorkspaceFork({ workspace: parentWorkspace, background: true, requestBody: fork }) - // A poll that fails says nothing about the fork: it can be lost to the same proxy, or reach a - // server without background forks (one of an older version during a rolling deploy), which - // ignored the flag and created the fork before answering. + // A server without background forks ignores the flag and answers once the fork is created. + if (response.startsWith('Created forked workspace')) return + await waitForForkCreation(parentWorkspace, fork.id, onStep) +} + +/** Wait for the creation of a fork started in the background by this user from `parentWorkspace`. */ +export async function waitForForkCreation( + parentWorkspace: string, + forkId: string, + onStep?: (step: string) => void +): Promise { let failingSince: number | undefined while (true) { let result: Awaited> | undefined try { result = await WorkspaceService.getForkCreationStatus({ workspace: parentWorkspace, - forkWorkspaceId: fork.id + forkWorkspaceId: forkId }) failingSince = undefined } catch (e) { - if (await forkExists(fork.id)) return + // A failed poll says nothing about the fork: it can be lost to the same proxy the fork + // runs in the background to avoid, or reach a replica that has no status route yet. failingSince ??= Date.now() if (Date.now() - failingSince >= MAX_FAILING_POLLS_MS) throw e } @@ -44,11 +53,3 @@ export async function createWorkspaceForkAndWait( await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)) } } - -async function forkExists(id: string): Promise { - try { - return await WorkspaceService.existsWorkspace({ requestBody: { id } }) - } catch { - return false - } -}