diff --git a/backend/migrations/20260929092707_workspace_fork_creation.up.sql b/backend/migrations/20260929092707_workspace_fork_creation.up.sql index ac3218ca20..8e3a678419 100644 --- a/backend/migrations/20260929092707_workspace_fork_creation.up.sql +++ b/backend/migrations/20260929092707_workspace_fork_creation.up.sql @@ -2,9 +2,12 @@ -- 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. `step` names the part of the copy it is in. +-- `creation_id` names one attempt: the row is reused when a failed or abandoned one is retried, and +-- whoever polls reads only the attempt it started. CREATE TABLE workspace_fork_creation ( fork_workspace_id VARCHAR(50) PRIMARY KEY, parent_workspace_id VARCHAR(50) NOT NULL, + creation_id UUID NOT NULL, created_by VARCHAR(255) NOT NULL, started_at TIMESTAMPTZ NOT NULL DEFAULT now(), heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs index f9a13df9c8..ac3b7f624e 100644 --- a/backend/windmill-api-integration-tests/tests/fork_background.rs +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -3,10 +3,10 @@ use sqlx::{Pool, Postgres}; use windmill_test_utils::*; -async fn wait_for_fork(client: &reqwest::Client, base_url: &str, fork_id: &str) -> Value { +async fn wait_for_fork(client: &reqwest::Client, base_url: &str, creation_id: &str) -> Value { for _ in 0..100 { let status: Value = client - .get(format!("{base_url}/fork_creation_status/{fork_id}")) + .get(format!("{base_url}/fork_creation_status/{creation_id}")) .header("Authorization", "Bearer SECRET_TOKEN") .send() .await @@ -19,7 +19,7 @@ async fn wait_for_fork(client: &reqwest::Client, base_url: &str, fork_id: &str) } tokio::time::sleep(std::time::Duration::from_millis(100)).await; } - panic!("fork {fork_id} still running"); + panic!("fork creation {creation_id} still running"); } /// A fork created in the background is reported by `fork_creation_status`, including a failure @@ -42,12 +42,13 @@ async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<( }; let resp = create("wm-fork-bg-ok").await?; - assert_eq!(resp.status(), 200, "{}", resp.text().await?); - let status = wait_for_fork(&client, &base_url, "wm-fork-bg-ok").await; + assert_eq!(resp.status(), 200); + let ok_creation = resp.text().await?; + let status = wait_for_fork(&client, &base_url, &ok_creation).await; assert_eq!(status, json!({ "status": "completed" })); - // A member who neither started the fork nor administers the workspace cannot read its outcome. + // Nobody but the user who started a creation reads it. let resp = client - .get(format!("{base_url}/fork_creation_status/wm-fork-bg-ok")) + .get(format!("{base_url}/fork_creation_status/{ok_creation}")) .header("Authorization", "Bearer SECRET_TOKEN_2") .send() .await?; @@ -69,22 +70,38 @@ async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<( .execute(&mut *blocker) .await?; let resp = create("wm-fork-bg-err").await?; - assert_eq!(resp.status(), 200, "{}", resp.text().await?); - // 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. + assert_eq!(resp.status(), 200); + let err_creation = resp.text().await?; let resp = create("wm-fork-bg-err").await?; - assert_eq!(resp.status(), 400); - assert!(resp.text().await?.contains("is already being created")); + assert_eq!(resp.status(), 400, "a second creation while the first is in flight"); sqlx::query("INSERT INTO workspace (id, name, owner) VALUES ('wm-fork-bg-err', 'x', 'x')") .execute(&mut *blocker) .await?; blocker.commit().await?; - let status = wait_for_fork(&client, &base_url, "wm-fork-bg-err").await; + let status = wait_for_fork(&client, &base_url, &err_creation).await; assert_eq!(status["status"], "failed"); assert!( status["error"].as_str().unwrap().contains("workspace_pkey"), "{status}" ); + // A retry of the same id is a new attempt: the failed one does not read its outcome. + sqlx::query("DELETE FROM workspace WHERE id = 'wm-fork-bg-err'") + .execute(&db) + .await?; + let resp = create("wm-fork-bg-err").await?; + assert_eq!(resp.status(), 200); + let retry_creation = resp.text().await?; + assert_eq!( + wait_for_fork(&client, &base_url, &retry_creation).await, + json!({ "status": "completed" }) + ); + let resp = client + .get(format!("{base_url}/fork_creation_status/{err_creation}")) + .header("Authorization", "Bearer SECRET_TOKEN") + .send() + .await?; + assert_eq!(resp.status(), 404); + Ok(()) } diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 3bbaef20ca..e5d2fdf91f 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -177,7 +177,7 @@ pub fn workspaced_service() -> Router { .route("/get_workspace_name", get(get_workspace_name)) .route("/create_fork", post(create_workspace_fork)) .route( - "/fork_creation_status/{fork_workspace_id}", + "/fork_creation_status/{creation_id}", get(get_fork_creation_status), ) .route("/attach_dev_workspace", post(attach_dev_workspace)) @@ -9055,7 +9055,6 @@ async fn create_workspace_fork( } let fork_id = nw.id.clone(); - let response_fork_id = fork_id.clone(); sqlx::query( "DELETE FROM workspace_fork_creation WHERE COALESCE(finished_at, heartbeat_at) < now() - interval '7 days'", @@ -9063,26 +9062,29 @@ async fn create_workspace_fork( .execute(&db) .await?; let claimed = sqlx::query( - "INSERT INTO workspace_fork_creation (fork_workspace_id, parent_workspace_id, created_by) - VALUES ($1, $2, $3) + "INSERT INTO workspace_fork_creation + (fork_workspace_id, parent_workspace_id, created_by, creation_id) + VALUES ($1, $2, $3, $5) ON CONFLICT (fork_workspace_id) DO UPDATE SET parent_workspace_id = EXCLUDED.parent_workspace_id, created_by = EXCLUDED.created_by, + creation_id = EXCLUDED.creation_id, 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", + RETURNING creation_id", ) .bind(&fork_id) .bind(&parent_workspace_id) .bind(&authed.email) .bind(FORK_HEARTBEAT_STALE_SECS as f64) + .bind(Uuid::new_v4()) .fetch_optional(&db) .await?; - // 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 { + // Every write of this attempt is scoped to its `creation_id`: one taken over as abandoned must + // not report over the attempt that took its place. + let Some(creation_id) = claimed.map(|r| r.get::("creation_id")) else { return Err(Error::BadRequest(format!( "workspace '{fork_id}' is already being created" ))); @@ -9106,10 +9108,10 @@ async fn create_workspace_fork( let step = *progress_rx.borrow_and_update(); if let Err(e) = sqlx::query( "UPDATE workspace_fork_creation SET heartbeat_at = now(), step = $3 - WHERE fork_workspace_id = $1 AND started_at = $2", + WHERE fork_workspace_id = $1 AND creation_id = $2", ) .bind(&fork_id) - .bind(started_at) + .bind(creation_id) .bind((!step.is_empty()).then_some(step)) .execute(&db) .await @@ -9137,10 +9139,10 @@ async fn create_workspace_fork( }); if let Err(e) = sqlx::query( "UPDATE workspace_fork_creation SET finished_at = now(), error = $3 - WHERE fork_workspace_id = $1 AND started_at = $2", + WHERE fork_workspace_id = $1 AND creation_id = $2", ) .bind(&fork_id) - .bind(started_at) + .bind(creation_id) .bind(error) .execute(&db) .await @@ -9148,9 +9150,7 @@ async fn create_workspace_fork( tracing::error!("fork '{fork_id}': could not record how its creation ended: {e}"); } }); - Ok(format!( - "Creating fork {response_fork_id} in the background" - )) + Ok(creation_id.to_string()) } /// How often a fork created in the background records that its copy is still running, and how long @@ -9166,8 +9166,9 @@ pub static BACKGROUND_FORKS: std::sync::LazyLock #[derive(Deserialize)] struct CreateWorkspaceForkQuery { - /// Return once the request is validated and create the fork in the background; its progress - /// is read from `fork_creation_status`. + /// Return once the request is validated and create the fork in the background, answering with + /// the id `fork_creation_status` reads this attempt by. Without it the answer is + /// `Created forked workspace `, which clients take as a server without background forks. background: Option, } @@ -9197,7 +9198,7 @@ struct ForkCreationStatus { async fn get_fork_creation_status( authed: ApiAuthed, Extension(db): Extension, - Path((parent_workspace_id, fork_id)): Path<(String, String)>, + Path((parent_workspace_id, creation_id)): Path<(String, Uuid)>, ) -> JsonResult { let row = sqlx::query( "SELECT c.created_by, c.finished_at IS NOT NULL AS finished, c.error, c.step, @@ -9206,24 +9207,19 @@ async fn get_fork_creation_status( 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", + WHERE c.creation_id = $1 AND c.parent_workspace_id = $2", ) - .bind(&fork_id) + .bind(creation_id) .bind(&parent_workspace_id) .bind(FORK_HEARTBEAT_STALE_SECS as f64) .fetch_optional(&db) .await?; - let Some(row) = row else { + // Only the user who started it reads an attempt: its id is not handed to anyone else. + let Some(row) = row.filter(|r| r.get::("created_by") == authed.email) else { return Err(Error::NotFound(format!( - "no creation of fork '{fork_id}' from '{parent_workspace_id}'" + "no fork creation {creation_id} from '{parent_workspace_id}'" ))); }; - let created_by: String = row.get("created_by"); - if created_by != authed.email && !authed.is_admin { - return Err(Error::NotFound(format!( - "no creation of fork '{fork_id}' from '{parent_workspace_id}'" - ))); - } 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. @@ -9681,6 +9677,7 @@ async fn write_workspace_fork( windmill_common::workspaces::invalidate_protection_rules_cache(&parent_workspace_id); } + // Clients that ask for a background fork recognise a server without it by this answer. Ok(format!("Created forked workspace {}", &forked_id)) } diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 4884b74636..7e2612a10d 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1400,8 +1400,8 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - name: background description: | - return once the request is validated and create the fork in the background; - poll getForkCreationStatus for how it ends + return once the request is validated and create the fork in the background, answering + with the creation id getForkCreationStatus reads in: query required: false schema: @@ -1415,13 +1415,13 @@ paths: $ref: "#/components/schemas/CreateWorkspaceFork" responses: "201": - description: forked workspace created, or its creation started when background is set + description: forked workspace created, or with background set, the id of its creation content: text/plain: schema: type: string - /w/{workspace}/workspaces/fork_creation_status/{fork_workspace_id}: + /w/{workspace}/workspaces/fork_creation_status/{creation_id}: get: summary: get the status of a fork created in the background operationId: getForkCreationStatus @@ -1429,11 +1429,12 @@ paths: - workspace parameters: - $ref: "#/components/parameters/WorkspaceId" - - name: fork_workspace_id + - name: creation_id in: path required: true schema: type: string + format: uuid responses: "200": description: fork creation status diff --git a/cli/src/commands/workspace/fork.ts b/cli/src/commands/workspace/fork.ts index b8041f5e7f..eedd9c0842 100644 --- a/cli/src/commands/workspace/fork.ts +++ b/cli/src/commands/workspace/fork.ts @@ -13,6 +13,7 @@ import { renameCurrentGitBranch, } from "../../utils/git.ts"; import process from "node:process"; +import { apiErrorMessage } from "../../utils/utils.ts"; import { WM_FORK_PREFIX } from "../../core/constants.ts"; import { tryResolveBranchWorkspace } from "../../core/context.ts"; import { @@ -22,8 +23,8 @@ import { readConfigFile, } from "../../core/conf.ts"; -/** Wait for the creation of a fork started in the background by this user. */ -async function waitForForkCreation(parentWorkspace: string, forkId: string) { +/** Wait for a fork creation started in the background, by the id its request answered with. */ +async function waitForForkCreation(parentWorkspace: string, creationId: string) { let failingSince: number | undefined; let lastStep: string | undefined; while (true) { @@ -33,7 +34,7 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) { try { result = await wmill.getForkCreationStatus({ workspace: parentWorkspace, - forkWorkspaceId: forkId, + creationId, }); failingSince = undefined; } catch (e) { @@ -420,13 +421,15 @@ async function createWorkspaceFork( }); // 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); + await waitForForkCreation(workspace.workspaceId, response); } log.info(colors.green(`✅ Created forked workspace ${trueWorkspaceId}`)); } catch (error) { log.error( - colors.red(`Failed to create forked workspace: ${(error as Error).message}`), + colors.red( + `Failed to create forked workspace: ${apiErrorMessage(error) ?? (error as Error).message}`, + ), ); throw error; } diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 5fcf95babf..9f8ea077ce 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, waitForForkCreation } from '$lib/utils/forkCreation' +import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation' import { sendUserToast } from '$lib/toast' import { onUserChange } from '$lib/userScopedStorage' @@ -1301,14 +1301,7 @@ export function setGeneratedSessionSummary( export async function materializeFork(fork: PendingFork): Promise { if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id try { - 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) - } + await createWorkspaceForkAndWait(fork.parent_workspace_id, { id: fork.id, name: fork.name }) 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 cde82663b5..e71d1b2a1e 100644 --- a/frontend/src/lib/utils/forkCreation.test.ts +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -27,7 +27,7 @@ describe('createWorkspaceForkAndWait', () => { beforeEach(() => { vi.useFakeTimers() vi.resetAllMocks() - svc.createWorkspaceFork.mockResolvedValue('Creating fork wm-fork-x in the background') + svc.createWorkspaceFork.mockResolvedValue('5c3e1b1e-0000-4000-8000-000000000000') }) afterEach(() => vi.useRealTimers()) diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index 1d2f511df0..ae0d6fbe1a 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -23,22 +23,14 @@ 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 - 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 { + const creationId = response let failingSince: number | undefined while (true) { let result: Awaited> | undefined try { result = await WorkspaceService.getForkCreationStatus({ workspace: parentWorkspace, - forkWorkspaceId: forkId + creationId }) failingSince = undefined } catch (e) {