diff --git a/backend/Cargo.lock b/backend/Cargo.lock index e80d0edf08..4d73a953b8 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15551,6 +15551,7 @@ dependencies = [ "strum", "tokio", "tokio-postgres", + "tokio-util", "tracing", "url", "uuid", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 8571b0b99c..27d8aeac8e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -519,7 +519,7 @@ serde_urlencoded = "^0" astral-tokio-tar = "^0.6.2" tempfile = "^3" x509-parser = "^0.16" -tokio-util = { version = "=0.7.17", features = ["io"] } +tokio-util = { version = "=0.7.17", features = ["io", "rt"] } json-pointer = "^0" itertools = "^0.14.0" regex = "^1" 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..8e3a678419 --- /dev/null +++ b/backend/migrations/20260929092707_workspace_fork_creation.up.sql @@ -0,0 +1,20 @@ +-- 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. `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(), + step TEXT, + 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..0e82fb0d70 --- /dev/null +++ b/backend/windmill-api-integration-tests/tests/fork_background.rs @@ -0,0 +1,120 @@ +use serde_json::{json, Value}; +use sqlx::{Pool, Postgres}; + +use windmill_test_utils::*; + +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/{creation_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 creation {creation_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); + let ok_creation = resp.text().await?; + let status = wait_for_fork(&client, &base_url, &ok_creation).await; + assert_eq!(status, json!({ "status": "completed" })); + // Nobody but the user who started a creation reads it. + let resp = client + .get(format!("{base_url}/fork_creation_status/{ok_creation}")) + .header("Authorization", "Bearer SECRET_TOKEN_2") + .send() + .await?; + assert_eq!(resp.status(), 404); + 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); + let err_creation = resp.text().await?; + let resp = create("wm-fork-bg-err").await?; + 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, &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); + + // 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'") + .execute(&db) + .await?; + let resp = client + .post(format!("{base_url}/create_fork")) + .header("Authorization", "Bearer SECRET_TOKEN") + .json(&json!({ "id": "wm-fork-sync", "name": "wm-fork-sync" })) + .send() + .await?; + assert_eq!(resp.text().await?, "Created forked workspace wm-fork-sync"); + + Ok(()) +} diff --git a/backend/windmill-api-workspaces/Cargo.toml b/backend/windmill-api-workspaces/Cargo.toml index 7ca1f692ad..9f05a26f28 100644 --- a/backend/windmill-api-workspaces/Cargo.toml +++ b/backend/windmill-api-workspaces/Cargo.toml @@ -50,6 +50,7 @@ sha2.workspace = true sqlx.workspace = true tokio.workspace = true tokio-postgres.workspace = true +tokio-util.workspace = true tracing.workspace = true url.workspace = true uuid.workspace = true diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index a4f230457f..95080d7d67 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -174,6 +174,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/{creation_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)) @@ -6922,7 +6926,9 @@ async fn clone_workspace_data( source_workspace_id: &str, target_workspace_id: &str, authed: &ApiAuthed, + progress: &ForkProgress, ) -> Result<()> { + progress.step("Copying settings, folders and groups"); // Clone workspace settings (merge with existing basic settings) update_workspace_settings(tx, source_workspace_id, target_workspace_id).await?; @@ -6944,6 +6950,7 @@ async fn clone_workspace_data( // Clone groups clone_groups(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying resources and variables"); // Clone resource types clone_resource_types(tx, source_workspace_id, target_workspace_id).await?; @@ -6953,6 +6960,7 @@ async fn clone_workspace_data( // Clone variables (including external secret backend replication) clone_variables(tx, db, source_workspace_id, target_workspace_id).await?; + progress.step("Copying scripts"); // Clone scripts with new hashes clone_scripts(tx, source_workspace_id, target_workspace_id).await?; @@ -6970,18 +6978,21 @@ async fn clone_workspace_data( clone_metric_catalog(tx, source_workspace_id, target_workspace_id).await?; clone_asset_usages_and_triggers(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying flows"); // Clone flows with new versions clone_flows(tx, source_workspace_id, target_workspace_id).await?; // Clone flow nodes clone_flow_nodes(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying apps"); // Clone apps with new IDs and app scripts let _app_id_mapping = clone_apps(tx, source_workspace_id, target_workspace_id, authed).await?; // Clone raw apps clone_raw_apps(tx, source_workspace_id, target_workspace_id).await?; + progress.step("Copying drafts and dependencies"); // Clone the forker's own per-user drafts (plus the legacy NULL-email // workspace draft, if any) so they keep their pending edits in the // fork. Other users' drafts are intentionally NOT cloned — they don't @@ -9181,6 +9192,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?; @@ -9286,63 +9298,279 @@ 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. - 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( + 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, + ForkProgress::default(), + )) + .await + .map_err(|e| { + Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}")) + })?; + } + + let fork_id = nw.id.clone(); + sqlx::query( + "DELETE FROM workspace_fork_creation + WHERE COALESCE(finished_at, heartbeat_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, 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 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 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::("creation_id")) else { + return Err(Error::BadRequest(format!( + "workspace '{fork_id}' is already being created" + ))); + }; + + BACKGROUND_FORKS.spawn(async move { + let (progress_tx, mut progress_rx) = tokio::sync::watch::channel(""); + // The one writer of the run's row while it copies: it records a new step as soon as the + // copy reports it, and refreshes the heartbeat in between. + 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)); + let mut reporting = true; + loop { + tokio::select! { + _ = interval.tick() => {} + changed = progress_rx.changed(), if reporting => reporting = changed.is_ok(), + } + 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 creation_id = $2", + ) + .bind(&fork_id) + .bind(creation_id) + .bind((!step.is_empty()).then_some(step)) + .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, + ForkProgress { creation_id: Some(creation_id), steps: Some(progress_tx) }, + )) + .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() + }); + // A fork that committed is already recorded complete, even when its commit's + // acknowledgement was lost and the attempt reads as an error here. + if let Err(e) = sqlx::query( + "UPDATE workspace_fork_creation SET finished_at = now(), error = $3 + WHERE fork_workspace_id = $1 AND creation_id = $2 AND finished_at IS NULL", ) + .bind(&fork_id) + .bind(creation_id) + .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(creation_id.to_string()) +} + +/// 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; + +/// Forks being created in the background. The server waits for them on shutdown: a copy cut short +/// never drops the data table databases it already made, and a retry under the same id is refused +/// while they exist. +pub static BACKGROUND_FORKS: std::sync::LazyLock = + std::sync::LazyLock::new(tokio_util::task::TaskTracker::new); + +#[derive(Deserialize)] +struct CreateWorkspaceForkQuery { + /// 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 `, which clients take as a server without background forks. + background: Option, +} + +/// The attempt a fork created in the background belongs to, and where its copy is, for its status +/// to report. +#[derive(Default)] +struct ForkProgress { + creation_id: Option, + steps: Option>, +} + +impl ForkProgress { + fn step(&self, step: &'static str) { + if let Some(tx) = &self.steps { + tx.send_replace(step); + } + } +} + +#[derive(Serialize)] +struct ForkCreationStatus { + /// `running`, `completed` or `failed`. + status: &'static str, + /// The part of the copy a running fork is in. + #[serde(skip_serializing_if = "Option::is_none")] + step: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, +} + +async fn get_fork_creation_status( + authed: ApiAuthed, + Extension(db): Extension, + Path((parent_workspace_id, creation_id)): Path<(String, Uuid)>, +) -> JsonResult { + let row = sqlx::query( + "SELECT c.created_by, c.finished_at IS NOT NULL AS finished, c.error, c.step, + c.heartbeat_at < now() - $3 * interval '1 second' AS stale + FROM workspace_fork_creation c + WHERE c.creation_id = $1 AND c.parent_workspace_id = $2", + ) + .bind(creation_id) + .bind(&parent_workspace_id) + .bind(FORK_HEARTBEAT_STALE_SECS as f64) + .fetch_optional(&db) + .await?; + // 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::("created_by") == authed.email) else { + return Err(Error::NotFound(format!( + "no fork creation {creation_id} from '{parent_workspace_id}'" + ))); + }; + let error: Option = row.get("error"); + let status = if row.get::("finished") { + match error { + Some(error) => ForkCreationStatus { status: "failed", step: None, error: Some(error) }, + None => ForkCreationStatus { status: "completed", step: None, error: None }, + } + } else if row.get::("stale") { + ForkCreationStatus { + status: "failed", + step: None, + 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", step: row.get("step"), 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, + progress: ForkProgress, +) -> Result { + let fork_id = nw.id.clone(); + if !nw.forked_datatables.is_empty() { + progress.step("Copying data tables"); + } + 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, + &progress, + ) .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 @@ -9438,6 +9666,7 @@ async fn write_workspace_fork( nw: CreateWorkspaceFork, dev_workspace_label: Option, copies: &[crate::datatable_clone::MadeCopy], + progress: &ForkProgress, ) -> Result { let mut tx: Transaction<'_, Postgres> = db.begin().await?; // Held until this transaction ends: after an error, the copies' cleanup waits on it, so it reads @@ -9540,8 +9769,15 @@ async fn write_workspace_fork( .await?; // Clone all data from the parent workspace using Rust implementation - if let Err(e) = - clone_workspace_data(&mut tx, &db, &parent_workspace_id, &forked_id, &authed).await + if let Err(e) = clone_workspace_data( + &mut tx, + &db, + &parent_workspace_id, + &forked_id, + &authed, + progress, + ) + .await { // A genuine `\u0000` in a source `json` value (`app_version.value` / // `flow_version.schema`) aborts the clone when it is re-encoded to jsonb: @@ -9572,6 +9808,7 @@ async fn write_workspace_fork( // enabled=false. Disabled rows have no side effects (no listener // attaches, no cron fires) so this is safe by construction. The user // re-enables in the fork, with parent-conflict warnings on enable. + progress.step("Copying triggers and schedules"); clone_triggers_and_schedules(&mut tx, &parent_workspace_id, &forked_id).await?; repoint_unresolvable_cloned_identities(&mut tx, &forked_id, &authed).await?; @@ -9719,6 +9956,16 @@ async fn write_workspace_fork( (!copied.is_empty()).then(|| [("copied_datatables", copied.as_str())].into()), ) .await?; + // Committed with the fork, so its attempt reads complete exactly when the fork exists. + if let Some(creation_id) = progress.creation_id { + sqlx::query( + "UPDATE workspace_fork_creation SET finished_at = now(), error = NULL + WHERE creation_id = $1", + ) + .bind(creation_id) + .execute(&mut *tx) + .await?; + } tx.commit().await?; // A pre-creation lookup could have cached an EMPTY ancestor chain for this id, which @@ -9732,6 +9979,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)) } diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index cf373241d4..e7afd78ceb 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, answering + with the creation id getForkCreationStatus reads + in: query + required: false + schema: + type: boolean requestBody: description: new forked workspace required: true @@ -1407,12 +1415,45 @@ paths: $ref: "#/components/schemas/CreateWorkspaceFork" responses: "201": - description: forked workspace created + 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/{creation_id}: + get: + summary: get the status of a fork created in the background + operationId: getForkCreationStatus + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: creation_id + in: path + required: true + schema: + type: string + format: uuid + responses: + "200": + description: fork creation status + content: + application/json: + schema: + type: object + properties: + status: + type: string + enum: [running, completed, failed] + step: + type: string + description: the part of the copy a running fork is in + 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/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 156c6ca898..556163ce1c 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -1279,6 +1279,15 @@ pub async fn run_server( }); server.await?; + let forks = &windmill_api_workspaces::workspaces::BACKGROUND_FORKS; + forks.close(); + if !forks.is_empty() { + tracing::info!( + "server off. waiting for {} fork(s) being created", + forks.len() + ); + forks.wait().await; + } #[cfg(feature = "agent_worker_server")] for (i, bg_processor) in agent_workers_bg_processor.into_iter().enumerate() { tracing::info!("server off. shutting down agent worker bg processor {i}"); diff --git a/cli/src/commands/workspace/fork.ts b/cli/src/commands/workspace/fork.ts index 0aca0984ff..eedd9c0842 100644 --- a/cli/src/commands/workspace/fork.ts +++ b/cli/src/commands/workspace/fork.ts @@ -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,6 +23,38 @@ import { readConfigFile, } from "../../core/conf.ts"; +/** 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) { + let result: + | Awaited> + | undefined; + try { + result = await wmill.getForkCreationStatus({ + workspace: parentWorkspace, + creationId, + }); + failingSince = undefined; + } catch (e) { + // 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; + } + if (result?.status === "running" && result.step && result.step !== lastStep) { + lastStep = result.step; + log.info(colors.gray(`${lastStep}...`)); + } + if (result?.status === "completed") return; + if (result?.status === "failed") { + throw new Error(result.error ?? "Unknown error"); + } + await new Promise((resolve) => setTimeout(resolve, 1500)); + } +} + async function createWorkspaceFork( opts: GlobalOptions & { createWorkspaceName: string | undefined; @@ -376,8 +409,9 @@ async function createWorkspaceFork( // --- Create the fork workspace --- try { - const result = await wmill.createWorkspaceFork({ + const response = await wmill.createWorkspaceFork({ workspace: workspace.workspaceId, + background: true, requestBody: { id: trueWorkspaceId, name: opts.createWorkspaceName ?? workspaceName ?? trueWorkspaceId, @@ -385,11 +419,17 @@ async function createWorkspaceFork( 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, response); + } - log.info(colors.green(`✅ ${result}`)); + 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; } diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index c5a2f28455..e81df02120 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, waitForForkCreation } from '$lib/utils/forkCreation' import { sendUserToast } from '$lib/toast' import { onUserChange } from '$lib/userScopedStorage' @@ -76,6 +77,9 @@ export type PendingFork = { id: string // Display name shown in the workspace bar. name: string + // The background creation this fork started, so a reload mid-copy waits for it rather than + // requesting the fork again, which the server refuses while that creation runs. + creation_id?: string } export type SessionSummarySource = 'placeholder' | 'generated' | 'manual' @@ -1180,7 +1184,10 @@ 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) + const newId = await materializeFork(fork, (creation_id) => { + s.pending_fork = { ...fork, creation_id } + void putSession(s) + }) if (!newId) { // Real failure (not a recovered duplicate). Drop the pending // fork so the session falls through to the workspace-pick @@ -1297,13 +1304,27 @@ export function setGeneratedSessionSummary( // 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. -export async function materializeFork(fork: PendingFork): Promise { +export async function materializeFork( + fork: PendingFork, + onCreationStarted?: (creationId: string) => void +): 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 } - }) + 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( + () => true, + () => false + ) + } + if (!resumed) { + await createWorkspaceForkAndWait( + fork.parent_workspace_id, + { id: fork.id, name: fork.name }, + { onCreationStarted } + ) + } usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) sendUserToast(`Created fork ${fork.name}`) return fork.id diff --git a/frontend/src/lib/components/sessions/sessionState.test.ts b/frontend/src/lib/components/sessions/sessionState.test.ts index 5cf3e3269e..62d2b8d39f 100644 --- a/frontend/src/lib/components/sessions/sessionState.test.ts +++ b/frontend/src/lib/components/sessions/sessionState.test.ts @@ -37,6 +37,7 @@ vi.mock('$lib/gen', async (orig) => { WorkspaceService: { ...actual.WorkspaceService, createWorkspaceFork: vi.fn().mockRejectedValue(new Error('fork creation failed')), + getForkCreationStatus: vi.fn().mockResolvedValue({ status: 'completed' }), listUserWorkspaces: vi.fn().mockResolvedValue([]) } } @@ -108,6 +109,38 @@ describe('commitSessionWorkspace — fork-creation failure', () => { }) }) +describe('commitSessionWorkspace — fork creation resumed after a reload', () => { + it('waits for the creation the pending fork started instead of requesting it again', async () => { + const id = 'test-commit-fork-resume' + const prevLicense = get(enterpriseLicense) + enterpriseLicense.set('test-license') + vi.mocked(WorkspaceService.createWorkspaceFork).mockClear() + sessionState.sessions.push({ + id, + name: 'fork-resume', + createdAt: 0, + pending_fork: { + parent_workspace_id: 'parent_ws', + id: 'wm-fork-resume', + name: 'resume', + creation_id: '5c3e1b1e-0000-4000-8000-000000000000' + } + } as Session) + try { + expect(await commitSessionWorkspace(id, 'parent_ws')).toBe('wm-fork-resume') + expect(WorkspaceService.getForkCreationStatus).toHaveBeenCalledWith({ + workspace: 'parent_ws', + creationId: '5c3e1b1e-0000-4000-8000-000000000000' + }) + expect(WorkspaceService.createWorkspaceFork).not.toHaveBeenCalled() + } finally { + 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/components/workspaceSettings/CreateWorkspaceInner.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte index d0f86b36ad..e6a0dce4fd 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 @@ -329,6 +330,7 @@ } let forkCreationLoading = $state(false) + let forkCreationStep = $state('Creating branch') let forkCreationError = $state('') let errorMsgs: string[] = $state([]) let failedSyncJobs: string[] = $state([]) @@ -376,6 +378,7 @@ async function forkWorkspace(prefixed_id: string): Promise { if (baseWorkspaceId) { forkCreationLoading = true + forkCreationStep = 'Creating branch' errorMsgs = [] failedSyncJobs = [] forkCreationError = '' @@ -470,9 +473,10 @@ } try { - await WorkspaceService.createWorkspaceFork({ - workspace: baseWorkspaceId!, - requestBody: { + forkCreationStep = 'Creating fork' + await createWorkspaceForkAndWait( + baseWorkspaceId!, + { id: prefixed_id, name, color: colorEnabled && workspaceColor ? workspaceColor : undefined, @@ -483,13 +487,15 @@ lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, lock_prod_forking: createAsDevWorkspace && effectiveLockProdForking, copy_members: copyMembers - } - }) + }, + { onStep: (step) => (forkCreationStep = step) } + ) } 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 } @@ -1115,7 +1121,8 @@ {:else} {/if} diff --git a/frontend/src/lib/utils/forkCreation.test.ts b/frontend/src/lib/utils/forkCreation.test.ts new file mode 100644 index 0000000000..bf180c6c6f --- /dev/null +++ b/frontend/src/lib/utils/forkCreation.test.ts @@ -0,0 +1,63 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('$lib/gen', () => ({ + WorkspaceService: { + createWorkspaceFork: vi.fn(), + getForkCreationStatus: vi.fn() + } +})) + +import { WorkspaceService } from '$lib/gen' +import { createWorkspaceForkAndWait } from './forkCreation' + +const svc = vi.mocked(WorkspaceService) +const fork = { id: 'wm-fork-x', name: 'x' } + +async function run(onStep?: (step: string) => void) { + const p = createWorkspaceForkAndWait('ws', fork, { onStep }) + const settled = p.then( + () => 'completed', + (e) => `failed: ${e.message}` + ) + await vi.runAllTimersAsync() + return settled +} + +describe('createWorkspaceForkAndWait', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.resetAllMocks() + svc.createWorkspaceFork.mockResolvedValue('5c3e1b1e-0000-4000-8000-000000000000') + }) + afterEach(() => vi.useRealTimers()) + + it('reports each step while the fork runs, then how it ended', async () => { + svc.getForkCreationStatus + .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValueOnce({ status: 'running', step: 'Copying flows' }) + .mockResolvedValueOnce({ status: 'failed', error: 'boom' }) + const steps: string[] = [] + expect(await run((s) => steps.push(s))).toBe('failed: boom') + expect(steps).toEqual(['Copying flows']) + }) + + 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') + }) +}) diff --git a/frontend/src/lib/utils/forkCreation.ts b/frontend/src/lib/utils/forkCreation.ts new file mode 100644 index 0000000000..d0fb4e5edf --- /dev/null +++ b/frontend/src/lib/utils/forkCreation.ts @@ -0,0 +1,57 @@ +import { WorkspaceService, type CreateWorkspaceFork } from '$lib/gen' + +const POLL_INTERVAL_MS = 1500 +// How long polls may keep failing, which a rolling deploy's replicas without the status route cause. +const MAX_FAILING_POLLS_MS = 3 * 60 * 1000 + +/** + * 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. + * `onCreationStarted` receives the id `waitForForkCreation` resumes the wait by, and `onStep` the + * part of the copy the server reports it is in. + */ +export async function createWorkspaceForkAndWait( + parentWorkspace: string, + fork: CreateWorkspaceFork, + opts: { onCreationStarted?: (creationId: string) => void; onStep?: (step: string) => void } = {} +): Promise { + const response = await WorkspaceService.createWorkspaceFork({ + workspace: parentWorkspace, + background: true, + requestBody: fork + }) + // A server without background forks ignores the flag and answers once the fork is created. + if (response.startsWith('Created forked workspace')) return + opts.onCreationStarted?.(response) + await waitForForkCreation(parentWorkspace, response, opts.onStep) +} + +/** Wait for a fork creation started in the background, by the id its request answered with. */ +export async function waitForForkCreation( + parentWorkspace: string, + creationId: string, + onStep?: (step: string) => void +): Promise { + let failingSince: number | undefined + while (true) { + let result: Awaited> | undefined + try { + result = await WorkspaceService.getForkCreationStatus({ + workspace: parentWorkspace, + creationId + }) + failingSince = undefined + } catch (e) { + // 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 (result?.status === 'running' && result.step) onStep?.(result.step) + if (result?.status === 'completed') return + if (result?.status === 'failed') throw new Error(result.error ?? 'Unknown error') + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)) + } +}