Merge branch 'main' into di/fork-clone-set-based

backend/Cargo.lock

@@ -15551,6 +15551,7 @@ dependencies = [
  "strum",
  "tokio",
  "tokio-postgres",
+ "tokio-util",
  "tracing",
  "url",
  "uuid",

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"

backend/migrations/20260929092707_workspace_fork_creation.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS workspace_fork_creation;

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;

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<Postgres>) -> 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(())
+}

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

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<DB>,
     Path(parent_workspace_id): Path<String>,
+    Query(q): Query<CreateWorkspaceForkQuery>,
     Json(nw): Json<CreateWorkspaceFork>,
 ) -> Result<String> {
     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,
-            &copy_requests(&nw.forked_datatables),
-        )
-        .await?;
-        let replayed: Vec<DataTableForkBehavior> = 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::<Uuid, _>("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<tokio_util::task::TaskTracker> =
+    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 <id>`, which clients take as a server without background forks.
+    background: Option<bool>,
+}
+
+/// 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<Uuid>,
+    steps: Option<tokio::sync::watch::Sender<&'static str>>,
+}
+
+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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error: Option<String>,
+}
+
+async fn get_fork_creation_status(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    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,
+                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::<String, _>("created_by") == authed.email) else {
+        return Err(Error::NotFound(format!(
+            "no fork creation {creation_id} from '{parent_workspace_id}'"
+        )));
+    };
+    let error: Option<String> = row.get("error");
+    let status = if row.get::<bool, _>("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::<bool, _>("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<String>,
+    progress: ForkProgress,
+) -> Result<String> {
+    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,
+        &copy_requests(&nw.forked_datatables),
+    )
+    .await?;
+    let replayed: Vec<DataTableForkBehavior> = 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<String>,
     copies: &[crate::datatable_clone::MadeCopy],
+    progress: &ForkProgress,
 ) -> Result<String> {
     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))
 }
 

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

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}");

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<ReturnType<typeof wmill.getForkCreationStatus>>
+      | 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;
   }

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<string | undefined> {
+export async function materializeFork(
+	fork: PendingFork,
+	onCreationStarted?: (creationId: string) => void
+): Promise<string | undefined> {
 	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

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.

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<void> {
 		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 @@
 			</Button>
 		{:else}
 			<Button variant="accent" disabled={true}>
-				<LoaderCircle class="animate-spin" /> Creating branch
+				<LoaderCircle class="animate-spin" />
+				{forkCreationStep}
 			</Button>
 		{/if}
 	</div>

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')
+	})
+})

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<void> {
+	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<void> {
+	let failingSince: number | undefined
+	while (true) {
+		let result: Awaited<ReturnType<typeof WorkspaceService.getForkCreationStatus>> | 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))
+	}
+}