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

feat: show which part of the copy a fork being created is in

Browse Source 
The background copy reports its phase (data tables, settings, resources,
scripts, flows, apps, drafts, triggers) through a watch channel. The heartbeat
task records it on the fork's creation row as soon as it changes, and
fork_creation_status returns it. The fork wizard shows it on its button, and
the CLI logs each step.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-29 12:33:08 +02:00
co-authored by Claude Opus 5.5
parent fd032a16c5
commit 1d52d20dee
7 changed files with 100 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20260929092707_workspace_fork_creation.up.sql
+2 -1
View File
@@ -1,13 +1,14 @@
-- 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.
-- belongs to a server that went away mid-copy. `step` names the part of the copy it is in.
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(),
step TEXT,
finished_at TIMESTAMPTZ,
error TEXT
);
backend/windmill-api-workspaces/src/workspaces.rs
+58 -9
View File
@@ -6667,7 +6667,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?;
@@ -6689,6 +6691,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?;
@@ -6698,6 +6701,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?;
@@ -6715,18 +6719,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
@@ -9039,6 +9046,7 @@ async fn create_workspace_fork(
parent_workspace_id,
nw,
dev_workspace_label,
ForkProgress::default(),
))
.await
.map_err(|e| {
@@ -9060,7 +9068,8 @@ async fn create_workspace_fork(
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
started_at = now(), heartbeat_at = now(), step = NULL, finished_at = NULL,
error = NULL
WHERE workspace_fork_creation.finished_at IS NOT NULL
OR workspace_fork_creation.heartbeat_at < now() - $4 * interval '1 second'
RETURNING started_at",
@@ -9080,19 +9089,28 @@ async fn create_workspace_fork(
};
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 {
interval.tick().await;
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()
"UPDATE workspace_fork_creation SET heartbeat_at = now(), step = $3
WHERE fork_workspace_id = $1 AND started_at = $2",
)
.bind(&fork_id)
.bind(started_at)
.bind((!step.is_empty()).then_some(step))
.execute(&db)
.await
{
@@ -9107,6 +9125,7 @@ async fn create_workspace_fork(
parent_workspace_id,
nw,
dev_workspace_label,
ForkProgress(Some(progress_tx)),
))
.await
.map_err(|e| Error::internal_err(format!("Creating the fork stopped unexpectedly: {e}")))
@@ -9152,10 +9171,25 @@ struct CreateWorkspaceForkQuery {
background: Option<bool>,
}
/// Where the copy of a fork created in the background is, for its status to report.
#[derive(Default)]
struct ForkProgress(Option<tokio::sync::watch::Sender<&'static str>>);
impl ForkProgress {
fn step(&self, step: &'static str) {
if let Some(tx) = &self.0 {
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>,
}
@@ -9166,7 +9200,7 @@ async fn get_fork_creation_status(
Path((parent_workspace_id, fork_id)): Path<(String, String)>,
) -> JsonResult<ForkCreationStatus> {
let row = sqlx::query(
"SELECT created_by, finished_at IS NOT NULL AS finished, error,
"SELECT created_by, finished_at IS NOT NULL AS finished, error, step,
heartbeat_at < now() - $3 * interval '1 second' AS stale
FROM workspace_fork_creation
WHERE fork_workspace_id = $1 AND parent_workspace_id = $2",
@@ -9190,12 +9224,13 @@ async fn get_fork_creation_status(
let error: Option<String> = row.get("error");
let status = if row.get::<bool, _>("finished") {
match error {
Some(error) => ForkCreationStatus { status: "failed", error: Some(error) },
None => ForkCreationStatus { status: "completed", error: None },
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"
@@ -9203,7 +9238,7 @@ async fn get_fork_creation_status(
),
}
} else {
ForkCreationStatus { status: "running", error: None }
ForkCreationStatus { status: "running", step: row.get("step"), error: None }
};
Ok(Json(status))
}
@@ -9216,8 +9251,12 @@ async fn make_workspace_fork(
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,
@@ -9237,6 +9276,7 @@ async fn make_workspace_fork(
nw,
dev_workspace_label,
&copies,
&progress,
)
.await
{
@@ -9362,6 +9402,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
@@ -9459,8 +9500,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:
@@ -9491,6 +9539,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?;
backend/windmill-api/openapi.yaml
+3
View File
@@ -1445,6 +1445,9 @@ paths:
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:
cli/src/commands/workspace/fork.ts
+5
View File
@@ -30,6 +30,7 @@ import {
*/
async function waitForForkCreation(parentWorkspace: string, forkId: string) {
let failedPolls = 0;
let lastStep: string | undefined;
while (true) {
let result:
| Awaited<ReturnType<typeof wmill.getForkCreationStatus>>
@@ -47,6 +48,10 @@ async function waitForForkCreation(parentWorkspace: string, forkId: string) {
if (exists) return;
if (++failedPolls >= 10) 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");
frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte
+21 -13
View File
@@ -330,6 +330,7 @@
}
let forkCreationLoading = $state(false)
let forkCreationStep = $state('Creating branch')
let forkCreationError = $state('')
let errorMsgs: string[] = $state([])
let failedSyncJobs: string[] = $state([])
@@ -377,6 +378,7 @@
async function forkWorkspace(prefixed_id: string): Promise<void> {
if (baseWorkspaceId) {
forkCreationLoading = true
forkCreationStep = 'Creating branch'
errorMsgs = []
failedSyncJobs = []
forkCreationError = ''
@@ -471,18 +473,23 @@
}
try {
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
})
forkCreationStep = 'Creating fork'
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
},
(step) => (forkCreationStep = step)
)
} catch (e) {
const msg = e?.body ?? e?.message ?? e ?? 'Unknown error'
forkCreationError = `Failed to create fork '${prefixed_id}'`
@@ -1114,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
+7 -5
View File
@@ -14,8 +14,8 @@ import { createWorkspaceForkAndWait } from './forkCreation'
const svc = vi.mocked(WorkspaceService)
const fork = { id: 'wm-fork-x', name: 'x' }
async function run() {
const p = createWorkspaceForkAndWait('ws', fork)
async function run(onStep?: (step: string) => void) {
const p = createWorkspaceForkAndWait('ws', fork, onStep)
const settled = p.then(
() => 'completed',
(e) => `failed: ${e.message}`
@@ -33,12 +33,14 @@ describe('createWorkspaceForkAndWait', () => {
})
afterEach(() => vi.useRealTimers())
it('waits while the fork is running and reports how it ended', async () => {
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' })
expect(await run()).toBe('failed: boom')
expect(svc.getForkCreationStatus).toHaveBeenCalledTimes(2)
const steps: string[] = []
expect(await run((s) => steps.push(s))).toBe('failed: boom')
expect(steps).toEqual(['Copying flows'])
})
it('takes a failed poll for success once the fork exists', async () => {
frontend/src/lib/utils/forkCreation.ts
+4 -1
View File
@@ -8,10 +8,12 @@ const MAX_FAILED_POLLS = 10
*
* 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.
* `onStep` receives the part of the copy the server reports it is in.
*/
export async function createWorkspaceForkAndWait(
parentWorkspace: string,
fork: CreateWorkspaceFork
fork: CreateWorkspaceFork,
onStep?: (step: string) => void
): Promise<void> {
try {
await WorkspaceService.createWorkspaceFork({
@@ -39,6 +41,7 @@ export async function createWorkspaceForkAndWait(
if (await forkExists(fork.id)) return
if (++failedPolls >= MAX_FAILED_POLLS) 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))