diff --git a/backend/migrations/20260929092707_workspace_fork_creation.down.sql b/backend/migrations/20260929092707_workspace_fork_creation.down.sql new file mode 100644 index 0000000000..924855e129 --- /dev/null +++ b/backend/migrations/20260929092707_workspace_fork_creation.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS workspace_fork_creation; diff --git a/backend/migrations/20260929092707_workspace_fork_creation.up.sql b/backend/migrations/20260929092707_workspace_fork_creation.up.sql new file mode 100644 index 0000000000..560dac6cf5 --- /dev/null +++ b/backend/migrations/20260929092707_workspace_fork_creation.up.sql @@ -0,0 +1,16 @@ +-- A fork being created in the background, and how that ended. The fork's own workspace row only +-- 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. +CREATE TABLE workspace_fork_creation ( + fork_workspace_id VARCHAR(50) PRIMARY KEY, + parent_workspace_id VARCHAR(50) NOT NULL, + created_by VARCHAR(255) NOT NULL, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), + finished_at TIMESTAMPTZ, + error TEXT +); + +GRANT ALL ON workspace_fork_creation TO windmill_user; +GRANT ALL ON workspace_fork_creation TO windmill_admin; diff --git a/backend/windmill-api-integration-tests/tests/fork_background.rs b/backend/windmill-api-integration-tests/tests/fork_background.rs new file mode 100644 index 0000000000..ae8a73bfde --- /dev/null +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -0,0 +1,80 @@ +use serde_json::{json, Value}; +use sqlx::{Pool, Postgres}; + +use windmill_test_utils::*; + +async fn wait_for_fork(client: &reqwest::Client, base_url: &str, fork_id: &str) -> Value { + for _ in 0..100 { + let status: Value = client + .get(format!("{base_url}/fork_creation_status/{fork_id}")) + .header("Authorization", "Bearer SECRET_TOKEN") + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + if status["status"] != "running" { + return status; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + panic!("fork {fork_id} still running"); +} + +/// A fork created in the background is reported by `fork_creation_status`, including a failure +/// that happens after the request has already returned. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_fork_created_in_background(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base_url = format!( + "http://localhost:{}/api/w/test-workspace/workspaces", + server.addr.port() + ); + let client = reqwest::Client::new(); + let create = |id: &str| { + client + .post(format!("{base_url}/create_fork?background=true")) + .header("Authorization", "Bearer SECRET_TOKEN") + .json(&json!({ "id": id, "name": id })) + .send() + }; + + 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!(status, json!({ "status": "completed" })); + let exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM workspace WHERE id = 'wm-fork-bg-ok')") + .fetch_one(&db) + .await?; + assert!(exists); + // Under CE's workspace cap. + sqlx::query("UPDATE workspace SET deleted = true WHERE id = 'wm-fork-bg-ok'") + .execute(&db) + .await?; + + // Hold the fork's write lock so the request is validated and returns, then commit a workspace + // with the same id: the background write fails on it. + let mut blocker = db.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtext('fork:wm-fork-bg-err'))") + .execute(&mut *blocker) + .await?; + let resp = create("wm-fork-bg-err").await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let resp = create("wm-fork-bg-err").await?; + assert_eq!(resp.status(), 400, "a second creation of the same fork"); + 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; + assert_eq!(status["status"], "failed"); + assert!( + status["error"].as_str().unwrap().contains("workspace_pkey"), + "{status}" + ); + + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 185fe17b25..121c5989d4 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -176,6 +176,10 @@ pub fn workspaced_service() -> Router { .route("/leave", post(leave_workspace)) .route("/get_workspace_name", get(get_workspace_name)) .route("/create_fork", post(create_workspace_fork)) + .route( + "/fork_creation_status/{fork_workspace_id}", + get(get_fork_creation_status), + ) .route("/attach_dev_workspace", post(attach_dev_workspace)) .route("/detach_dev_workspace", post(detach_dev_workspace)) .route("/get_dev_workspace", get(get_dev_workspace)) @@ -8919,6 +8923,7 @@ async fn create_workspace_fork( authed: ApiAuthed, Extension(db): Extension, Path(parent_workspace_id): Path, + Query(q): Query, Json(nw): Json, ) -> Result { enforce_fork_depth(&db, &parent_workspace_id, 0).await?; @@ -9024,63 +9029,238 @@ async fn create_workspace_fork( // Refused here, before any database exists; `make_copies` checks again under the locks. validate_forked_datatables(&db, &authed, &parent_workspace_id, &nw).await?; - // Detached from the request: a client that goes away while the copies are made must still end - // with the fork created or no copy left behind, and dropping the handler's future would skip - // that cleanup. + if !q.background.unwrap_or(false) { + // Detached from the request: a client that goes away while the copies are made must still + // end with the fork created or no copy left behind, and dropping the handler's future + // would skip that cleanup. + return tokio::spawn(make_workspace_fork( + db, + authed, + parent_workspace_id, + nw, + dev_workspace_label, + )) + .await + .map_err(|e| { + Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}")) + })?; + } + + let fork_id = nw.id.clone(); + let response_fork_id = fork_id.clone(); + sqlx::query( + "DELETE FROM workspace_fork_creation + WHERE finished_at < now() - interval '7 days'", + ) + .execute(&db) + .await?; + let claimed = sqlx::query( + "INSERT INTO workspace_fork_creation (fork_workspace_id, parent_workspace_id, created_by) + VALUES ($1, $2, $3) + ON CONFLICT (fork_workspace_id) DO UPDATE SET + parent_workspace_id = EXCLUDED.parent_workspace_id, + created_by = EXCLUDED.created_by, + started_at = now(), heartbeat_at = now(), 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", + ) + .bind(&fork_id) + .bind(&parent_workspace_id) + .bind(&authed.email) + .bind(FORK_HEARTBEAT_STALE_SECS as f64) + .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 { + return Err(Error::BadRequest(format!( + "workspace '{fork_id}' is already being created" + ))); + }; + tokio::spawn(async move { - let fork_id = nw.id.clone(); - let copies = crate::datatable_clone::make_copies( - &db, - &authed, - &parent_workspace_id, - ©_requests(&nw.forked_datatables), - ) - .await?; - let replayed: Vec = copies - .iter() - .filter(|c| c.replayed) - .map(|c| c.behavior) - .collect(); - match write_workspace_fork( + let heartbeat = { + let (db, fork_id) = (db.clone(), fork_id.clone()); + tokio::spawn(async move { + let mut interval = + tokio::time::interval(std::time::Duration::from_secs(FORK_HEARTBEAT_SECS)); + loop { + interval.tick().await; + if let Err(e) = sqlx::query( + "UPDATE workspace_fork_creation SET heartbeat_at = now() + WHERE fork_workspace_id = $1 AND started_at = $2", + ) + .bind(&fork_id) + .bind(started_at) + .execute(&db) + .await + { + tracing::warn!("fork '{fork_id}': could not record progress: {e}"); + } + } + }) + }; + let made = tokio::spawn(make_workspace_fork( db.clone(), authed, parent_workspace_id, nw, dev_workspace_label, - &copies, + )) + .await + .map_err(|e| Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}"))) + .and_then(|r| r); + heartbeat.abort(); + let error = made.err().map(|e| { + tracing::error!("Creating fork '{fork_id}' failed: {e}"); + e.to_string() + }); + 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", ) + .bind(&fork_id) + .bind(started_at) + .bind(error) + .execute(&db) .await { - Ok(message) => { - for behavior in replayed { - windmill_common::feature_usage::log_feature_usage( - "datatable", - "clone_replayed", - match behavior { - DataTableForkBehavior::SchemaOnly => "schema_only", - _ => "schema_and_data", - }, - ); - } - Ok(message) - } - // A commit whose acknowledgement was lost can still have committed, and a concurrent - // request for the same id can have: once the write has settled, the cleanup keeps - // whichever copies a committed fork names, and drops the rest. - Err(e) => match wait_for_fork_write(&db, &fork_id).await { - Ok(()) => Err(crate::datatable_clone::drop_copies_after(&db, copies, e).await), - Err(settle) => { - tracing::error!( - "Could not tell whether fork '{fork_id}' was created, so its copies were \ - kept: {settle}" - ); - Err(e) - } - }, + tracing::error!("fork '{fork_id}': could not record how its creation ended: {e}"); } - }) + }); + Ok(format!( + "Creating fork {response_fork_id} in the background" + )) +} + +/// How often a fork created in the background records that its copy is still running, and how long +/// without that record its creation counts as abandoned. +const FORK_HEARTBEAT_SECS: u64 = 10; +const FORK_HEARTBEAT_STALE_SECS: u64 = 60; + +#[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`. + background: Option, +} + +#[derive(Serialize)] +struct ForkCreationStatus { + /// `running`, `completed` or `failed`. + status: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, +} + +async fn get_fork_creation_status( + authed: ApiAuthed, + Extension(db): Extension, + Path((parent_workspace_id, fork_id)): Path<(String, String)>, +) -> JsonResult { + let row = sqlx::query( + "SELECT created_by, finished_at IS NOT NULL AS finished, error, + heartbeat_at < now() - $3 * interval '1 second' AS stale + FROM workspace_fork_creation + WHERE fork_workspace_id = $1 AND parent_workspace_id = $2", + ) + .bind(&fork_id) + .bind(&parent_workspace_id) + .bind(FORK_HEARTBEAT_STALE_SECS as f64) + .fetch_optional(&db) + .await?; + let Some(row) = row else { + return Err(Error::NotFound(format!( + "no creation of fork '{fork_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"); + let status = if row.get::("finished") { + match error { + Some(error) => ForkCreationStatus { status: "failed", error: Some(error) }, + None => ForkCreationStatus { status: "completed", error: None }, + } + } else if row.get::("stale") { + ForkCreationStatus { + status: "failed", + error: Some( + "the server creating the fork stopped before it finished; check the server logs \ + and whether the fork exists before retrying" + .to_string(), + ), + } + } else { + ForkCreationStatus { status: "running", error: None } + }; + Ok(Json(status)) +} + +/// Copy the data tables the fork gets, then write the fork itself, dropping the copies again if +/// that write fails. +async fn make_workspace_fork( + db: DB, + authed: ApiAuthed, + parent_workspace_id: String, + nw: CreateWorkspaceFork, + dev_workspace_label: Option, +) -> Result { + let fork_id = nw.id.clone(); + let copies = crate::datatable_clone::make_copies( + &db, + &authed, + &parent_workspace_id, + ©_requests(&nw.forked_datatables), + ) + .await?; + let replayed: Vec = copies + .iter() + .filter(|c| c.replayed) + .map(|c| c.behavior) + .collect(); + match write_workspace_fork( + db.clone(), + authed, + parent_workspace_id, + nw, + dev_workspace_label, + &copies, + ) .await - .map_err(|e| Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}")))? + { + Ok(message) => { + for behavior in replayed { + windmill_common::feature_usage::log_feature_usage( + "datatable", + "clone_replayed", + match behavior { + DataTableForkBehavior::SchemaOnly => "schema_only", + _ => "schema_and_data", + }, + ); + } + Ok(message) + } + // A commit whose acknowledgement was lost can still have committed, and a concurrent + // request for the same id can have: once the write has settled, the cleanup keeps + // whichever copies a committed fork names, and drops the rest. + Err(e) => match wait_for_fork_write(&db, &fork_id).await { + Ok(()) => Err(crate::datatable_clone::drop_copies_after(&db, copies, e).await), + Err(settle) => { + tracing::error!( + "Could not tell whether fork '{fork_id}' was created, so its copies were \ + kept: {settle}" + ); + Err(e) + } + }, + } } /// The database a data table of the fork or dev workspace `fork_id` is copied into, as the wizard diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 85ae9388c8..e6c4be9968 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1398,6 +1398,14 @@ paths: - workspace parameters: - $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 + in: query + required: false + schema: + type: boolean requestBody: description: new forked workspace required: true @@ -1407,12 +1415,41 @@ paths: $ref: "#/components/schemas/CreateWorkspaceFork" responses: "201": - description: forked workspace created + description: forked workspace created, or its creation started when background is set content: text/plain: schema: type: string + /w/{workspace}/workspaces/fork_creation_status/{fork_workspace_id}: + get: + summary: get the status of a fork created in the background + operationId: getForkCreationStatus + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: fork_workspace_id + in: path + required: true + schema: + type: string + responses: + "200": + description: fork creation status + content: + application/json: + schema: + type: object + properties: + status: + type: string + enum: [running, completed, failed] + error: + type: string + required: + - status + /w/{workspace}/workspaces/attach_dev_workspace: post: summary: attach an existing workspace as the dev workspace of this (prod) workspace diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index c5a2f28455..9f8ea077ce 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -45,6 +45,7 @@ export function syncWorkspaceTo(workspaceId: string | undefined): void { switchWorkspace(workspaceId) } import { WorkspaceService } from '$lib/gen' +import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation' import { sendUserToast } from '$lib/toast' import { onUserChange } from '$lib/userScopedStorage' @@ -1300,10 +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 { - await WorkspaceService.createWorkspaceFork({ - workspace: fork.parent_workspace_id, - requestBody: { id: fork.id, name: fork.name } - }) + 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/components/workspaceSettings/CreateWorkspaceInner.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte index d0f86b36ad..a6916e53f9 100644 --- a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte +++ b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte @@ -70,6 +70,7 @@ import WorkspaceScopeTrigger from '../WorkspaceScopeTrigger.svelte' import DarkModeToggle from '../sidebar/DarkModeToggle.svelte' import ForkDucklakeSection from './ForkDucklakeSection.svelte' + import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation' interface Props { isFork?: boolean @@ -470,26 +471,24 @@ } try { - await WorkspaceService.createWorkspaceFork({ - workspace: baseWorkspaceId!, - requestBody: { - id: prefixed_id, - name, - color: colorEnabled && workspaceColor ? workspaceColor : undefined, - forked_datatables: forkedDatatables, - shared_ducklakes: forkDucklakeSection?.getSharedDucklakes() ?? [], - is_dev_workspace: createAsDevWorkspace, - dev_workspace_label: createAsDevWorkspace ? devWorkspaceLabel : undefined, - lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, - lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, - copy_members: copyMembers - } + await createWorkspaceForkAndWait(baseWorkspaceId!, { + id: prefixed_id, + name, + color: colorEnabled && workspaceColor ? workspaceColor : undefined, + forked_datatables: forkedDatatables, + shared_ducklakes: forkDucklakeSection?.getSharedDucklakes() ?? [], + is_dev_workspace: createAsDevWorkspace, + dev_workspace_label: createAsDevWorkspace ? devWorkspaceLabel : undefined, + lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, + lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, + copy_members: copyMembers }) } catch (e) { + const msg = e?.body ?? e?.message ?? e ?? 'Unknown error' forkCreationError = `Failed to create fork '${prefixed_id}'` - errorMsgs.push(e?.body ?? e ?? 'Unknown error') + errorMsgs.push(msg) forkCreationLoading = false - sendUserToast(`Could not create fork '${prefixed_id}' ${e}`, true) + sendUserToast(`Could not create fork '${prefixed_id}' ${msg}`, true) return } diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts new file mode 100644 index 0000000000..bc072afdb9 --- /dev/null +++ b/frontend/src/lib/utils/forkCreation.ts @@ -0,0 +1,39 @@ +import { WorkspaceService, type CreateWorkspaceFork } from '$lib/gen' + +const POLL_INTERVAL_MS = 1500 +const MAX_FAILED_POLLS = 5 + +/** + * Create a fork and resolve once it exists, rejecting with the reason it could not be created. + * + * The copy runs on the server after the request returns: a large workspace takes longer to copy + * than the timeout of many proxies in front of Windmill, which would otherwise cut the request. + */ +export async function createWorkspaceForkAndWait( + parentWorkspace: string, + fork: CreateWorkspaceFork +): Promise { + await WorkspaceService.createWorkspaceFork({ + workspace: parentWorkspace, + background: true, + requestBody: fork + }) + // A poll lost to the same proxy says nothing about the fork, which is still being created. + let failedPolls = 0 + while (true) { + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)) + let result: Awaited> + try { + result = await WorkspaceService.getForkCreationStatus({ + workspace: parentWorkspace, + forkWorkspaceId: fork.id + }) + failedPolls = 0 + } catch (e) { + if (++failedPolls >= MAX_FAILED_POLLS) throw e + continue + } + if (result.status === 'completed') return + if (result.status === 'failed') throw new Error(result.error ?? 'Unknown error') + } +}