starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-12 16:02:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: poll a background fork by the id of its own attempt

Browse Source 
create_fork?background=true answers with a creation id, and the status
route reads that attempt only, for the user who started it. A retry that
reuses the fork id is a new attempt, so a poller never reads another
attempt's outcome. The AI-session fork no longer adopts a creation in flight,
which it could not tell apart from someone else's.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-29 14:59:37 +02:00
co-authored by Claude Opus 5.5
parent a6e2f7abf2
commit 4cf481fa34
8 changed files with 77 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20260929092707_workspace_fork_creation.up.sql
+3
View File
@@ -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(),
backend/windmill-api-integration-tests/tests/fork_background.rs
+30 -13
View File
@@ -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<Postgres>) -> 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<Postgres>) -> 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(())
}
backend/windmill-api-workspaces/src/workspaces.rs
+25 -28
View File
@@ -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::<chrono::DateTime<Utc>, _>("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::<Uuid, _>("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<tokio_util::task::TaskTracker>
#[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 <id>`, which clients take as a server without background forks.
background: Option<bool>,
}
@@ -9197,7 +9198,7 @@ struct ForkCreationStatus {
async fn get_fork_creation_status(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path((parent_workspace_id, fork_id)): Path<(String, String)>,
Path((parent_workspace_id, creation_id)): Path<(String, Uuid)>,
) -> JsonResult<ForkCreationStatus> {
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::<String, _>("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<String> = 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))
}
backend/windmill-api/openapi.yaml
+6 -5
View File
@@ -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
cli/src/commands/workspace/fork.ts
+8 -5
View File
@@ -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;
}
frontend/src/lib/components/sessions/sessionState.svelte.ts
+2 -9
View File
@@ -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<string | undefined> {
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
frontend/src/lib/utils/forkCreation.test.ts
+1 -1
View File
@@ -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())
frontend/src/lib/utils/forkCreation.ts
+2 -10
View File
@@ -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<void> {
const creationId = response
let failingSince: number | undefined
while (true) {
let result: Awaited<ReturnType<typeof WorkspaceService.getForkCreationStatus>> | undefined
try {
result = await WorkspaceService.getForkCreationStatus({
workspace: parentWorkspace,
forkWorkspaceId: forkId
creationId
})
failingSince = undefined
} catch (e) {