mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-13 00:03:00 +00:00
fix: drop fork-creation joins and existence probes
A second request for a fork being created is refused again; only the AI-session fork, whose request never varies, waits for its own earlier one. Clients recognise a server without background forks by its synchronous answer instead of probing for a workspace by id, which could name another parent's fork. 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
e45797cdc2
commit
a6e2f7abf2
@@ -70,15 +70,9 @@ async fn test_fork_created_in_background(db: Pool<Postgres>) -> 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.
|
||||
// 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.
|
||||
let resp = create("wm-fork-bg-err").await?;
|
||||
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')")
|
||||
|
||||
@@ -9083,23 +9083,6 @@ 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::<chrono::DateTime<Utc>, _>("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"
|
||||
)));
|
||||
|
||||
@@ -22,12 +22,7 @@ import {
|
||||
readConfigFile,
|
||||
} from "../../core/conf.ts";
|
||||
|
||||
/**
|
||||
* Wait for a fork requested with `background: true`. A poll that fails says nothing about the
|
||||
* fork: it can be lost to a proxy, or reach a server without background forks (an older version,
|
||||
* possibly one replica during a rolling deploy), which ignored the flag and created the fork
|
||||
* before answering.
|
||||
*/
|
||||
/** Wait for the creation of a fork started in the background by this user. */
|
||||
async function waitForForkCreation(parentWorkspace: string, forkId: string) {
|
||||
let failingSince: number | undefined;
|
||||
let lastStep: string | undefined;
|
||||
@@ -42,10 +37,8 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) {
|
||||
});
|
||||
failingSince = undefined;
|
||||
} catch (e) {
|
||||
const exists = await wmill
|
||||
.existsWorkspace({ requestBody: { id: forkId } })
|
||||
.catch(() => false);
|
||||
if (exists) return;
|
||||
// A failed poll says nothing about the fork: it can be lost to a proxy, or reach a replica
|
||||
// that has no status route yet during a rolling deploy.
|
||||
failingSince ??= Date.now();
|
||||
if (Date.now() - failingSince >= 3 * 60 * 1000) throw e;
|
||||
}
|
||||
@@ -415,18 +408,20 @@ async function createWorkspaceFork(
|
||||
|
||||
// --- Create the fork workspace ---
|
||||
try {
|
||||
await wmill
|
||||
.createWorkspaceFork({
|
||||
workspace: workspace.workspaceId,
|
||||
background: true,
|
||||
requestBody: {
|
||||
id: trueWorkspaceId,
|
||||
name: opts.createWorkspaceName ?? workspaceName ?? trueWorkspaceId,
|
||||
color: forkColor,
|
||||
forked_datatables: forkedDatatables,
|
||||
},
|
||||
});
|
||||
await waitForForkCreation(workspace.workspaceId, trueWorkspaceId);
|
||||
const response = await wmill.createWorkspaceFork({
|
||||
workspace: workspace.workspaceId,
|
||||
background: true,
|
||||
requestBody: {
|
||||
id: trueWorkspaceId,
|
||||
name: opts.createWorkspaceName ?? workspaceName ?? trueWorkspaceId,
|
||||
color: forkColor,
|
||||
forked_datatables: forkedDatatables,
|
||||
},
|
||||
});
|
||||
// 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);
|
||||
}
|
||||
|
||||
log.info(colors.green(`✅ Created forked workspace ${trueWorkspaceId}`));
|
||||
} catch (error) {
|
||||
|
||||
@@ -45,7 +45,7 @@ export function syncWorkspaceTo(workspaceId: string | undefined): void {
|
||||
switchWorkspace(workspaceId)
|
||||
}
|
||||
import { WorkspaceService } from '$lib/gen'
|
||||
import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation'
|
||||
import { createWorkspaceForkAndWait, waitForForkCreation } from '$lib/utils/forkCreation'
|
||||
import { sendUserToast } from '$lib/toast'
|
||||
import { onUserChange } from '$lib/userScopedStorage'
|
||||
|
||||
@@ -1301,7 +1301,14 @@ export function setGeneratedSessionSummary(
|
||||
export async function materializeFork(fork: PendingFork): Promise<string | undefined> {
|
||||
if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id
|
||||
try {
|
||||
await createWorkspaceForkAndWait(fork.parent_workspace_id, { id: fork.id, name: fork.name })
|
||||
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)
|
||||
}
|
||||
usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces())
|
||||
sendUserToast(`Created fork ${fork.name}`)
|
||||
return fork.id
|
||||
|
||||
@@ -3,8 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
vi.mock('$lib/gen', () => ({
|
||||
WorkspaceService: {
|
||||
createWorkspaceFork: vi.fn(),
|
||||
getForkCreationStatus: vi.fn(),
|
||||
existsWorkspace: vi.fn()
|
||||
getForkCreationStatus: vi.fn()
|
||||
}
|
||||
}))
|
||||
|
||||
@@ -28,8 +27,7 @@ describe('createWorkspaceForkAndWait', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
vi.resetAllMocks()
|
||||
svc.createWorkspaceFork.mockResolvedValue('started')
|
||||
svc.existsWorkspace.mockResolvedValue(false)
|
||||
svc.createWorkspaceFork.mockResolvedValue('Creating fork wm-fork-x in the background')
|
||||
})
|
||||
afterEach(() => vi.useRealTimers())
|
||||
|
||||
@@ -43,17 +41,23 @@ describe('createWorkspaceForkAndWait', () => {
|
||||
expect(steps).toEqual(['Copying flows'])
|
||||
})
|
||||
|
||||
it('takes a failed poll for success once the fork exists', async () => {
|
||||
// A server without background forks created it synchronously and has no status route.
|
||||
svc.getForkCreationStatus.mockRejectedValue(
|
||||
Object.assign(new Error('Not Found'), { status: 404 })
|
||||
)
|
||||
svc.existsWorkspace.mockResolvedValue(true)
|
||||
it('returns at once when the server created the fork synchronously', async () => {
|
||||
// A server without background forks ignores the flag.
|
||||
svc.createWorkspaceFork.mockResolvedValue('Created forked workspace wm-fork-x')
|
||||
expect(await run()).toBe('completed')
|
||||
expect(svc.getForkCreationStatus).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('keeps polling through failed polls', async () => {
|
||||
svc.getForkCreationStatus
|
||||
.mockRejectedValueOnce(Object.assign(new Error('Not Found'), { status: 404 }))
|
||||
.mockRejectedValueOnce(new Error('upstream request timeout'))
|
||||
.mockResolvedValueOnce({ status: 'completed' })
|
||||
expect(await run()).toBe('completed')
|
||||
})
|
||||
|
||||
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')
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -16,25 +16,34 @@ export async function createWorkspaceForkAndWait(
|
||||
fork: CreateWorkspaceFork,
|
||||
onStep?: (step: string) => void
|
||||
): Promise<void> {
|
||||
await WorkspaceService.createWorkspaceFork({
|
||||
const response = 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.
|
||||
// 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<void> {
|
||||
let failingSince: number | undefined
|
||||
while (true) {
|
||||
let result: Awaited<ReturnType<typeof WorkspaceService.getForkCreationStatus>> | undefined
|
||||
try {
|
||||
result = await WorkspaceService.getForkCreationStatus({
|
||||
workspace: parentWorkspace,
|
||||
forkWorkspaceId: fork.id
|
||||
forkWorkspaceId: forkId
|
||||
})
|
||||
failingSince = undefined
|
||||
} catch (e) {
|
||||
if (await forkExists(fork.id)) return
|
||||
// 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
|
||||
}
|
||||
@@ -44,11 +53,3 @@ export async function createWorkspaceForkAndWait(
|
||||
await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS))
|
||||
}
|
||||
}
|
||||
|
||||
async function forkExists(id: string): Promise<boolean> {
|
||||
try {
|
||||
return await WorkspaceService.existsWorkspace({ requestBody: { id } })
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user