mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-13 00:03:00 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
c0c266ea87
commit
eab3749a71
@@ -103,6 +103,22 @@ async fn test_fork_created_in_background(db: Pool<Postgres>) -> 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'")
|
||||
|
||||
@@ -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<string | undefined> {
|
||||
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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<void> {
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user