From e45797cdc239e7d9f9a319924220273b73412692 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 29 Sep 2026 12:37:41 +0200 Subject: [PATCH] fix: join a retried fork creation server-side, settle status by the fork's existence A retry from the same user and parent joins the creation in flight instead of being refused, so clients no longer match the refusal's wording, and another requester can never adopt it. The status route reports a fork that exists under its parent as completed, whatever its run's record says. Clients give up on failing polls after a time window rather than a count, which a rolling deploy can exhaust. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/fork_background.rs | 11 +++++- .../windmill-api-workspaces/src/workspaces.rs | 34 ++++++++++++++++--- cli/src/commands/workspace/fork.ts | 11 +++--- frontend/src/lib/utils/forkCreation.test.ts | 13 ++----- frontend/src/lib/utils/forkCreation.ts | 25 ++++++-------- 5 files changed, 56 insertions(+), 38 deletions(-) diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs index 5a62cca68c..74d3c0e424 100644 --- a/backend/windmill-api-integration-tests/tests/fork_background.rs +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -70,8 +70,17 @@ 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. let resp = create("wm-fork-bg-err").await?; - assert_eq!(resp.status(), 400, "a second creation of the same fork"); + 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')") .execute(&mut *blocker) .await?; diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 3c4220a9eb..fbbb0e59e3 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -9083,6 +9083,23 @@ 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" ))); @@ -9200,10 +9217,13 @@ 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, step, - heartbeat_at < now() - $3 * interval '1 second' AS stale - FROM workspace_fork_creation - WHERE fork_workspace_id = $1 AND parent_workspace_id = $2", + "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 + FROM workspace_fork_creation c + WHERE c.fork_workspace_id = $1 AND c.parent_workspace_id = $2", ) .bind(&fork_id) .bind(&parent_workspace_id) @@ -9222,7 +9242,11 @@ async fn get_fork_creation_status( ))); } let error: Option = row.get("error"); - let status = if row.get::("finished") { + // 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") { match error { Some(error) => ForkCreationStatus { status: "failed", step: None, error: Some(error) }, None => ForkCreationStatus { status: "completed", step: None, error: None }, diff --git a/cli/src/commands/workspace/fork.ts b/cli/src/commands/workspace/fork.ts index 580112bbfb..aaa05dba1b 100644 --- a/cli/src/commands/workspace/fork.ts +++ b/cli/src/commands/workspace/fork.ts @@ -29,7 +29,7 @@ import { * before answering. */ async function waitForForkCreation(parentWorkspace: string, forkId: string) { - let failedPolls = 0; + let failingSince: number | undefined; let lastStep: string | undefined; while (true) { let result: @@ -40,13 +40,14 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) { workspace: parentWorkspace, forkWorkspaceId: forkId, }); - failedPolls = 0; + failingSince = undefined; } catch (e) { const exists = await wmill .existsWorkspace({ requestBody: { id: forkId } }) .catch(() => false); if (exists) return; - if (++failedPolls >= 10) throw e; + failingSince ??= Date.now(); + if (Date.now() - failingSince >= 3 * 60 * 1000) throw e; } if (result?.status === "running" && result.step && result.step !== lastStep) { lastStep = result.step; @@ -424,10 +425,6 @@ async function createWorkspaceFork( color: forkColor, forked_datatables: forkedDatatables, }, - }) - .catch((e) => { - // Started by an earlier run: wait for that creation instead. - if (!String((e as { body?: unknown })?.body ?? "").includes("is already being created")) throw e; }); await waitForForkCreation(workspace.workspaceId, trueWorkspaceId); diff --git a/frontend/src/lib/utils/forkCreation.test.ts b/frontend/src/lib/utils/forkCreation.test.ts index a454560b8f..4f83451cfa 100644 --- a/frontend/src/lib/utils/forkCreation.test.ts +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -52,17 +52,8 @@ describe('createWorkspaceForkAndWait', () => { expect(await run()).toBe('completed') }) - it('gives up after repeated failed polls for a fork that never appears', async () => { + 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') - expect(svc.getForkCreationStatus).toHaveBeenCalledTimes(10) - }) - - it('waits for a creation an earlier request already started', async () => { - svc.createWorkspaceFork.mockRejectedValue({ - body: "Bad request: workspace 'wm-fork-x' is already being created" - }) - svc.getForkCreationStatus.mockResolvedValue({ status: 'completed' }) - expect(await run()).toBe('completed') - }) + }) }) diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index bd37b100a2..849684629f 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -1,7 +1,8 @@ import { WorkspaceService, type CreateWorkspaceFork } from '$lib/gen' const POLL_INTERVAL_MS = 1500 -const MAX_FAILED_POLLS = 10 +// How long polls may keep failing, which a rolling deploy's replicas without the status route cause. +const MAX_FAILING_POLLS_MS = 3 * 60 * 1000 /** * Create a fork and resolve once it exists, rejecting with the reason it could not be created. @@ -15,20 +16,15 @@ export async function createWorkspaceForkAndWait( fork: CreateWorkspaceFork, onStep?: (step: string) => void ): Promise { - try { - await WorkspaceService.createWorkspaceFork({ - workspace: parentWorkspace, - background: true, - requestBody: fork - }) - } catch (e) { - // A retry of a request whose response was lost: wait for the creation it started. - if (!String(e?.body ?? '').includes('is already being created')) throw e - } + 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. - let failedPolls = 0 + let failingSince: number | undefined while (true) { let result: Awaited> | undefined try { @@ -36,10 +32,11 @@ export async function createWorkspaceForkAndWait( workspace: parentWorkspace, forkWorkspaceId: fork.id }) - failedPolls = 0 + failingSince = undefined } catch (e) { if (await forkExists(fork.id)) return - if (++failedPolls >= MAX_FAILED_POLLS) throw e + failingSince ??= Date.now() + if (Date.now() - failingSince >= MAX_FAILING_POLLS_MS) throw e } if (result?.status === 'running' && result.step) onStep?.(result.step) if (result?.status === 'completed') return