From eab3749a7123075e84049419669704ca5b96c8d4 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 29 Sep 2026 15:35:30 +0200 Subject: [PATCH] fix: bound a resumed session fork's wait, keep its intent while the fork is in flight Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/fork_background.rs | 16 ++++++ .../sessions/sessionState.svelte.ts | 55 +++++++++++++------ .../components/sessions/sessionState.test.ts | 37 +++++++++++++ frontend/src/lib/utils/forkCreation.ts | 10 +++- 4 files changed, 98 insertions(+), 20 deletions(-) diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs index 0e82fb0d70..24ef3468ae 100644 --- a/backend/windmill-api-integration-tests/tests/fork_background.rs +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -103,6 +103,22 @@ async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<( .await?; assert_eq!(resp.status(), 404); + // An attempt whose server stopped heartbeating before its fork committed reads as failed. + let abandoned = "5c3e1b1e-0000-4000-8000-000000000000"; + sqlx::query( + "INSERT INTO workspace_fork_creation + (fork_workspace_id, parent_workspace_id, created_by, creation_id, heartbeat_at) + VALUES ('wm-fork-gone', 'test-workspace', 'test@windmill.dev', $1::uuid, + now() - interval '2 minutes')", + ) + .bind(abandoned) + .execute(&db) + .await?; + assert_eq!( + wait_for_fork(&client, &base_url, abandoned).await["status"], + "failed" + ); + // 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'") diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index e81df02120..772a8f2e45 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -1184,10 +1184,16 @@ 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, (creation_id) => { - s.pending_fork = { ...fork, creation_id } - void putSession(s) + let stillCreating = false + const newId = await materializeFork(fork, { + onCreationStarted: (creation_id) => { + s.pending_fork = { ...fork, creation_id } + void putSession(s) + }, + onStillCreating: () => (stillCreating = true) }) + // The fork is on its way: keep the intent so the next send adopts it. + if (!newId && stillCreating) return undefined if (!newId) { // Real failure (not a recovered duplicate). Drop the pending // fork so the session falls through to the workspace-pick @@ -1293,27 +1299,37 @@ export function setGeneratedSessionSummary( return true } -// Create a new fork workspace via the API, refresh the user-workspaces -// store, and return the new fork id. Used by both the first-send commit -// path (commitSessionWorkspace) and the move-session-to-a-new-fork path -// in the unavailable-session banner. Returns undefined on failure (a -// user-facing toast is already emitted). +const RESUMED_FORK_MAX_FAILING_MS = 10_000 + +// Make sure the fork exists, refresh the user-workspaces store, and return +// the fork id. Used by both the first-send commit path +// (commitSessionWorkspace) and the move-session-to-a-new-fork path in the +// unavailable-session banner. Returns undefined on failure (a user-facing +// toast is already emitted). // -// Self-heal: if `fork.id` is already present in the user-workspaces -// store, the previous create succeeded (whose response we apparently -// 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. +// Self-heal: a fork already present in the user-workspaces store was +// created by an earlier request whose response was lost, and is adopted +// without requesting it again; so is one the API reports as a duplicate +// key. A fork whose `creation_id` is set waits for that creation (started +// before a reload). `onCreationStarted` receives the id of a new creation; +// `onStillCreating` is called instead of failing when the server reports +// this fork is being created by a creation the client lost track of. export async function materializeFork( fork: PendingFork, - onCreationStarted?: (creationId: string) => void + opts: { onCreationStarted?: (creationId: string) => void; onStillCreating?: () => void } = {} ): Promise { if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id try { 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( + // An attempt that failed or is gone is requested again below. Its polls fail only when + // it is gone, so they are not waited out as a fresh creation's are. + resumed = await waitForForkCreation( + fork.parent_workspace_id, + fork.creation_id, + undefined, + RESUMED_FORK_MAX_FAILING_MS + ).then( () => true, () => false ) @@ -1322,7 +1338,7 @@ export async function materializeFork( await createWorkspaceForkAndWait( fork.parent_workspace_id, { id: fork.id, name: fork.name }, - { onCreationStarted } + { onCreationStarted: opts.onCreationStarted } ) } usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) @@ -1330,6 +1346,11 @@ export async function materializeFork( return fork.id } catch (e: any) { const msg = String(e?.body ?? e?.message ?? e) + if (opts.onStillCreating && msg.includes('is already being created')) { + sendUserToast(`Fork ${fork.name} is still being created, send again in a moment`) + opts.onStillCreating() + return undefined + } if (/workspace_pkey|duplicate key/i.test(msg)) { // Self-heal: the create likely already succeeded. Refresh + adopt the // existing row. Guard this refresh — a second network failure here must diff --git a/frontend/src/lib/components/sessions/sessionState.test.ts b/frontend/src/lib/components/sessions/sessionState.test.ts index 62d2b8d39f..a609d9a036 100644 --- a/frontend/src/lib/components/sessions/sessionState.test.ts +++ b/frontend/src/lib/components/sessions/sessionState.test.ts @@ -141,6 +141,43 @@ describe('commitSessionWorkspace — fork creation resumed after a reload', () = }) }) +describe('commitSessionWorkspace — fork still being created', () => { + it('keeps the pending fork when its creation is gone but the fork is still in flight', async () => { + const id = 'test-commit-fork-in-flight' + const prevLicense = get(enterpriseLicense) + enterpriseLicense.set('test-license') + vi.useFakeTimers() + vi.mocked(WorkspaceService.getForkCreationStatus).mockRejectedValue( + Object.assign(new Error('Not Found'), { status: 404 }) + ) + vi.mocked(WorkspaceService.createWorkspaceFork).mockRejectedValue({ + body: "Bad request: workspace 'wm-fork-flight' is already being created" + }) + const pending_fork = { + parent_workspace_id: 'parent_ws', + id: 'wm-fork-flight', + name: 'flight', + creation_id: '5c3e1b1e-0000-4000-8000-000000000000' + } + sessionState.sessions.push({ id, name: 'fork-flight', createdAt: 0, pending_fork } as Session) + try { + const committed = commitSessionWorkspace(id, 'parent_ws') + await vi.runAllTimersAsync() + expect(await committed).toBeUndefined() + expect(sessionState.sessions.find((x) => x.id === id)?.pending_fork).toEqual(pending_fork) + } finally { + vi.useRealTimers() + vi.mocked(WorkspaceService.getForkCreationStatus).mockResolvedValue({ status: 'completed' }) + vi.mocked(WorkspaceService.createWorkspaceFork).mockRejectedValue( + new Error('fork creation failed') + ) + 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/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts index d0fb4e5edf..aadd67e70f 100644 --- a/frontend/src/lib/utils/forkCreation.ts +++ b/frontend/src/lib/utils/forkCreation.ts @@ -28,11 +28,15 @@ export async function createWorkspaceForkAndWait( await waitForForkCreation(parentWorkspace, response, opts.onStep) } -/** Wait for a fork creation started in the background, by the id its request answered with. */ +/** + * Wait for a fork creation started in the background, by the id its request answered with. + * `maxFailingMs` bounds how long polls may keep failing before the wait gives up. + */ export async function waitForForkCreation( parentWorkspace: string, creationId: string, - onStep?: (step: string) => void + onStep?: (step: string) => void, + maxFailingMs = MAX_FAILING_POLLS_MS ): Promise { let failingSince: number | undefined while (true) { @@ -47,7 +51,7 @@ export async function waitForForkCreation( // 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 + if (Date.now() - failingSince >= maxFailingMs) throw e } if (result?.status === 'running' && result.step) onStep?.(result.step) if (result?.status === 'completed') return