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