From dc3ebfbe74c11a554e4518e91d304e052bc20612 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 10 Sep 2026 19:15:33 +0200 Subject: [PATCH] fix(datatables): keep the fork schema baseline, and bounce streams on every removal Three fixes from review. `edit_datatable_config` took `forked_from` wholesale from the stored entry, so the fork schema diff's save of an advanced baseline was silently discarded and an applied change was offered again. Whether an entry carries a clone stamp is still carried from the store, since that is what marks its database droppable, but the baseline inside it is now taken from the request. The stranded-pointer warning and the stream bounce ran over the optional `deleted_datatables` hint, which the settings-sync CLI never sends, so removing a governing data table through `wmill` bounced nothing. Removals are now derived from the stored configuration against the saved one. `delete_workspace` read the pointers to bounce before its transaction, so a fork committing a pointer during the deletion was missed. The read now happens inside the transaction, after the workspace row is deleted: a fork's insert key-share locks that row through its parent foreign key, so it is either seen or fails on the missing parent. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/datatable_roles.rs | 76 ++++++++++++++++++- .../windmill-api-workspaces/src/workspaces.rs | 34 ++++++--- .../src/workspaces_extra.rs | 31 ++++---- 3 files changed, 115 insertions(+), 26 deletions(-) diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 43c768cb73..f91495ecef 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -599,6 +599,79 @@ async fn a_data_table_under_roles_is_not_copied_into_a_fork( Ok(()) } +/// The fork's `forked_from` for one of its entries; `None` whether it is absent or `null`. +async fn forked_from_of(db: &Pool, name: &str) -> Option { + sqlx::query_scalar::<_, Option>( + "SELECT datatable->'datatables'->$1::text->'forked_from' + FROM workspace_settings WHERE workspace_id = 'wm-fork-dt'", + ) + .bind(name) + .fetch_one(db) + .await + .unwrap() + .filter(|v| !v.is_null()) +} + +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn a_clone_stamp_is_carried_but_its_schema_baseline_advances( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + // Whether an entry is a clone is what marks its database droppable, so a save can neither + // stamp nor unstamp one. The schema baseline inside the stamp is what the fork's schema diff + // advances after applying a change; dropping it would offer that same change again. + sqlx::query( + r#"UPDATE workspace_settings SET datatable = '{"datatables": { + "clone": {"database": {"resource_type": "instance", "resource_path": "wm_fork_dt__clone"}, + "forked_from": {"schema": {}}}, + "plain": {"database": {"resource_type": "instance", "resource_path": "dt_plain"}}}}'::jsonb + WHERE workspace_id = 'wm-fork-dt'"#, + ) + .execute(&db) + .await?; + let server = ApiServer::start(db.clone()).await?; + let url = format!( + "http://localhost:{}/api/w/wm-fork-dt/workspaces/edit_datatable_config", + server.addr.port() + ); + let clone_db = json!({"resource_type": "instance", "resource_path": "wm_fork_dt__clone"}); + let plain_db = json!({"resource_type": "instance", "resource_path": "dt_plain"}); + let baseline = json!({"schema": {"public": {"orders": {"id": "int4"}}}}); + + let resp = authed(client().post(&url), "SECRET_TOKEN_2") + .json(&json!({"settings": {"datatables": { + "clone": {"database": clone_db, "forked_from": baseline}, + "plain": {"database": plain_db, "forked_from": {"schema": {}}} + }}})) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + assert_eq!( + forked_from_of(&db, "clone").await, + Some(baseline.clone()), + "the schema diff's baseline did not advance" + ); + assert_eq!( + forked_from_of(&db, "plain").await, + None, + "a save stamped a clone" + ); + + let resp = authed(client().post(&url), "SECRET_TOKEN_2") + .json(&json!({"settings": {"datatables": { + "clone": {"database": clone_db}, "plain": {"database": plain_db} + }}})) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + assert_eq!( + forked_from_of(&db, "clone").await, + Some(baseline), + "a save unstamped a clone" + ); + Ok(()) +} + /// Live replication listeners on `main`: a fork's through its pointer, the governing workspace's /// own (in the `?role=` form), and a fork capture — plus one on another data table, which no /// deletion of `main` may touch. Each is held by a server, as a running stream would be. @@ -659,7 +732,8 @@ async fn deleting_a_governing_data_table_bounces_the_streams_reading_it( )), "SECRET_TOKEN", ) - .json(&json!({"settings": {"datatables": {}}, "renames": [], "deleted_datatables": ["main"]})) + // The way the settings-sync CLI saves: the new document alone, with no deletion hint. + .json(&json!({"settings": {"datatables": {}}})) .send() .await?; assert_eq!(resp.status(), 200, "{}", resp.text().await?); diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index ca17629f58..6ac96e3f0b 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3799,15 +3799,18 @@ async fn edit_datatable_config( Some(true) } }; - // Three fields this form does not own, carried across from the stored entry rather than - // taken from the request. `permissions` is an access decision, edited through its own - // endpoint; `reference` is what makes a fork answer to the workspace that governs its data - // table, and letting a save clear it would hand the fork the database outright; and - // `forked_from` is the clone stamp the fork flow writes. Only fork creation writes any of - // them, so a settings save can neither widen nor lose them. + // Carried across from the stored entry rather than taken from the request. `permissions` + // is an access decision, edited through its own endpoint; `reference` is what makes a fork + // answer to the workspace that governs its data table, and letting a save clear it would + // hand the fork the database outright. `forked_from` is the clone stamp the fork flow + // writes: whether an entry has one is carried the same way, since it is what marks the + // database droppable, but the schema baseline inside it is the diff view's to advance. dt.permissions = old.and_then(|old| old.permissions.clone()); dt.reference = old.and_then(|old| old.reference.clone()); - dt.forked_from = old.and_then(|old| old.forked_from.clone()); + dt.forked_from = match old.and_then(|old| old.forked_from.as_ref()) { + Some(stored) => Some(dt.forked_from.take().unwrap_or_else(|| stored.clone())), + None => None, + }; // Carrying the block onto a resource-backed entry would produce a data table the chokepoint // refuses on every job — a save that succeeds and breaks everything afterwards. Refuse it // instead: turning roles off first is one step, and it keeps discarding an access decision @@ -3868,6 +3871,18 @@ async fn edit_datatable_config( } } + // Every entry this save removes, derived rather than taken from `deleted_datatables`: that list + // is a hint the settings-sync CLI never sends, and the stranded-pointer warning and the stream + // bounce below must run for a removal whether or not the caller named it. + let removed: Vec = old_datatables + .keys() + .filter(|name| { + !new_config.settings.datatables.contains_key(*name) + && !new_config.renames.iter().any(|r| &r.from == *name) + }) + .cloned() + .collect(); + let config: serde_json::Value = serde_json::to_value(new_config.settings) .map_err(|err| Error::internal_err(err.to_string()))?; @@ -3907,7 +3922,7 @@ async fn edit_datatable_config( // A deletion cannot be followed the same way — there is nothing to point at any more. Read who // is left stranded so the caller is told, the way deleting a workspace does. let mut stranded: Vec = Vec::new(); - for name in &new_config.deleted_datatables { + for name in &removed { let rows = sqlx::query!( r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" FROM workspace_settings ws @@ -3932,8 +3947,7 @@ async fn edit_datatable_config( // a listener that reconnects finds the entry gone instead of streaming on. crate::datatable_permissions::restart_streams_named( &mut *tx, - new_config - .deleted_datatables + removed .iter() .map(|name| (w_id.clone(), name.clone())) .chain( diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index d66f2a13b0..371b80555c 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -995,21 +995,6 @@ pub(crate) async fn delete_workspace( // but the destructive cleanup itself runs only after the commit below: a delete that // fails mid-way must never leave a live workspace with its fork data destroyed and no // registry row to retry from. Read-only: nothing is dropped here. - // Read before the delete: another workspace's data table entry can point at one of this - // workspace's, and deleting the workspace it names leaves that pointer resolving to nothing. - // Nothing sweeps them — turning them back into copies would hand each fork the database - // outright — so the deleter is told which data tables they just stranded. - let stranded_pointers = sqlx::query!( - r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" - FROM workspace_settings ws - CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt - WHERE dt.value->'reference'->>'workspace_id' = $1 - ORDER BY ws.workspace_id, dt.key"#, - &w_id, - ) - .fetch_all(&db) - .await - .unwrap_or_default(); let fork_ducklake_cleanups = prepare_fork_ducklake_cleanups(&db, &w_id, None) .await @@ -1265,6 +1250,22 @@ pub(crate) async fn delete_workspace( None, ) .await?; + // Who points here, read after the delete above and inside this transaction. A fork writes its + // pointer in the transaction that inserts it, and that insert key-share locks this row through + // the parent foreign key: so the fork either committed before the delete and is seen here, or + // waits on it and then fails on the missing parent. No pointer can escape this list. Nothing + // sweeps them afterwards — turning them back into copies would hand each fork the database + // outright — so the deleter is told which data tables they stranded. + let stranded_pointers = sqlx::query!( + r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" + FROM workspace_settings ws + CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt + WHERE dt.value->'reference'->>'workspace_id' = $1 + ORDER BY ws.workspace_id, dt.key"#, + &w_id, + ) + .fetch_all(&mut *tx) + .await?; // Each fork pointing here keeps the replication stream it opened while this workspace // governed it. Bounced in this transaction, so a listener that reconnects finds its pointer // dangling instead of streaming on.