From e29dacb93813105bec61eb5c0cdf90a8e8475cf0 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Fri, 18 Sep 2026 15:27:35 +0200 Subject: [PATCH] fix(datatables): serialize fork database reservations, cleanup, setup and Ducklake saves on the database - A fork copy is created and registered under its workspace's fork lock, and refused once the workspace is archived; a rename re-migrates reservations under that lock after archiving. - Ducklake saves lock every instance database they newly name, as data table saves do. - Instance database setup holds the database's lock until its entry is written. - Fork import also holds the database's lock across the restore. - Cleanup re-reads the entry under its locks before dropping anything. - Non-superadmins no longer see fork copies reserved for workspaces they are not in. Co-Authored-By: Claude Opus 5 (1M context) --- backend/windmill-api-settings/src/lib.rs | 41 ++++++++++- .../windmill-api-workspaces/src/workspaces.rs | 56 ++++++++++++--- .../src/workspaces_extra.rs | 72 ++++++++++++++----- 3 files changed, 139 insertions(+), 30 deletions(-) diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 71e46be9c4..4288684a2c 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -1689,8 +1689,33 @@ async fn list_custom_instance_pg_databases( })?; if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? { - // Which workspace reserved a fork copy is nobody else's business: it would enumerate every - // pending fork on the instance. + // A fork copy's name gives away the workspace it was reserved for, so every pending fork on + // the instance would be listed. Kept for members of that workspace, and wherever the + // caller's workspaces use it, e.g. the fork it was finalized into. + let reserved_visible: BTreeSet = sqlx::query_scalar( + r#"SELECT e.k FROM global_settings gs + CROSS JOIN LATERAL jsonb_each(gs.value->'databases') AS e(k, v) + WHERE gs.name = 'custom_instance_pg_databases' AND e.v->>'workspace_id' IS NOT NULL + AND (EXISTS (SELECT 1 FROM usr WHERE usr.email = $1 + AND usr.workspace_id = e.v->>'workspace_id') + OR EXISTS (SELECT 1 FROM usr JOIN workspace_settings ws + ON ws.workspace_id = usr.workspace_id + CROSS JOIN LATERAL jsonb_each( + CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object' + THEN ws.datatable->'datatables' ELSE '{}'::jsonb END) dt + WHERE usr.email = $1 + AND dt.value->'database'->>'resource_type' = 'instance' + AND dt.value->'database'->>'resource_path' = e.k))"#, + ) + .bind(&authed.email) + .fetch_all(&db) + .await? + .into_iter() + .collect(); + result.retain(|dbname, entry| { + entry.workspace_id.is_none() || reserved_visible.contains(dbname) + }); + // Which workspace reserved a copy is still only for superadmins. for entry in result.values_mut() { entry.workspace_id = None; } @@ -1766,6 +1791,15 @@ async fn setup_custom_instance_pg_database( // Before anything is recorded: the status written below replaces the registry entry, and with it // the workspace a fork copy is reserved for. require_super_admin(&db, &authed).await?; + // Fork cleanup checks and drops the database and its entry under this lock. Held from before + // the setup creates the database to after its entry is written, neither lands on the other's + // half-done state: a dropped database with its entry written back, or the reverse. + let mut tx = db.begin().await?; + windmill_common::datatable_roles::lock_instance_databases_governance( + &mut tx, + [dbname.trim()], + ) + .await?; let mut logs = CustomInstanceDbLogs::default(); let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await; let success = result.is_ok(); @@ -1792,8 +1826,9 @@ async fn setup_custom_instance_pg_database( ) .bind(&dbname) .bind(&status_json) - .fetch_one(&db) + .fetch_one(&mut *tx) .await?; + tx.commit().await?; let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?; Ok(Json(status)) diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 3d258b761b..363a810dda 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3437,6 +3437,19 @@ async fn create_pg_database( } if is_instance_datatable_source(&db, &w_id, &req.source).await? { + // Held until the copy is registered, as a rename migrates reservations to the new id under + // it once the old one is archived: a copy registered after that would be reserved for an + // id nothing answers on. + let mut tx = db.begin().await?; + windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?; + let live = sqlx::query_scalar::<_, bool>("SELECT NOT deleted FROM workspace WHERE id = $1") + .bind(&w_id) + .fetch_optional(&mut *tx) + .await? + .unwrap_or(false); + if !live { + return Err(Error::BadRequest(format!("Workspace '{w_id}' is archived"))); + } windmill_common::create_custom_instance_database( &db, &req.target_dbname, @@ -3444,6 +3457,7 @@ async fn create_pg_database( Some(&w_id), ) .await?; + tx.commit().await?; } else { let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?; @@ -3599,10 +3613,16 @@ async fn import_pg_database( )); } if is_instance_datatable_source(&db, &w_id, &req.target).await? { - // Held until the restore is done, as fork finalization takes it: a fork must not - // commit this database while `psql` is still filling it. + // Held until the restore is done: fork finalization takes the first, and every + // save newly naming a database, in any workspace, the second. Nothing may start + // using this database while `psql` is still filling it. let mut tx = db.begin().await?; windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?; + windmill_common::datatable_roles::lock_instance_databases_governance( + &mut tx, + [override_dbname.as_str()], + ) + .await?; windmill_common::ensure_fork_database_available_to(&db, override_dbname, &w_id) .await?; fork_lock = Some(tx); @@ -3719,20 +3739,38 @@ async fn edit_ducklake_config( ) .await?; - let old_ducklakes = sqlx::query_scalar!( - r#" - SELECT ws.ducklake->'ducklakes' AS ducklake_name - FROM workspace_settings ws - WHERE ws.workspace_id = $1 - "#, - &w_id + // Under the row lock the save writes with, taken before the database locks below as fork + // cleanup takes the two. + let old_ducklakes = sqlx::query_scalar::<_, Option>( + "SELECT ws.ducklake->'ducklakes' FROM workspace_settings ws + WHERE ws.workspace_id = $1 FOR UPDATE", ) + .bind(&w_id) .fetch_one(&mut *tx) .await? .unwrap_or(serde_json::Value::Null); let old_ducklakes: HashMap = serde_json::from_value(old_ducklakes).unwrap_or_default(); + // Fork cleanup decides nothing uses an instance database under this lock, so a catalog newly + // put on one must not commit between its check and its drop. + windmill_common::datatable_roles::lock_instance_databases_governance( + &mut *tx, + new_config + .settings + .ducklakes + .iter() + .filter(|(name, dl)| { + dl.catalog.resource_type == DucklakeCatalogResourceType::Instance + && old_ducklakes.get(name.as_str()).is_none_or(|old| { + old.catalog.resource_type != DucklakeCatalogResourceType::Instance + || old.catalog.resource_path != dl.catalog.resource_path + }) + }) + .map(|(_, dl)| dl.catalog.resource_path.as_str()), + ) + .await?; + // Check that non-superadmins are not abusing Instance databases if !is_superadmin { for (name, dl) in new_config.settings.ducklakes.iter() { diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index c627d99af6..605f82da38 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -110,20 +110,7 @@ pub(crate) async fn change_workspace_id( .execute(&mut *tx) .await?; - // A fork copy reserved for the old id would otherwise be unreachable: its creator cannot - // import into it or finish its fork under the new id, and nothing else would ever drop it. - sqlx::query( - r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', ( - SELECT COALESCE(jsonb_object_agg(k, CASE WHEN v->>'workspace_id' = $1 - THEN jsonb_set(v, '{workspace_id}', to_jsonb($2::text)) ELSE v END), '{}'::jsonb) - FROM jsonb_each(COALESCE(value->'databases', '{}'::jsonb)) AS e(k, v) - )) - WHERE name = 'custom_instance_pg_databases'"#, - ) - .bind(&old_id) - .bind(&rw.new_id) - .execute(&mut *tx) - .await?; + migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?; // Duplicate workspace settings (keep copy in old workspace for reference) info!("Duplicating workspace_settings table"); @@ -933,6 +920,14 @@ pub(crate) async fn change_workspace_id( let (_schedules_count, canceled_count, _deleted_tokens_count) = archive_workspace_impl(&db, &old_id, &authed.username, None).await?; + // The old id stays live between the commit above and the archive, and a fork copy created for + // it in that window registers under it. Creation checks the workspace is live under the fork + // lock, so once this has run under it, no copy can be reserved for the old id any more. + let mut tx = db.begin().await?; + windmill_common::workspaces::lock_fork_datatables(&mut tx, &old_id).await?; + migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?; + tx.commit().await?; + info!( "Workspace id change completed: moved {} to {}, archived old workspace", old_id, rw.new_id @@ -944,6 +939,28 @@ pub(crate) async fn change_workspace_id( )) } +/// A fork copy reserved for the old id would otherwise be unreachable: its creator cannot import +/// into it or finish its fork under the new id, and nothing else would ever drop it. +async fn migrate_fork_reservations( + tx: &mut Transaction<'_, Postgres>, + old_id: &str, + new_id: &str, +) -> Result<()> { + sqlx::query( + r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', ( + SELECT COALESCE(jsonb_object_agg(k, CASE WHEN v->>'workspace_id' = $1 + THEN jsonb_set(v, '{workspace_id}', to_jsonb($2::text)) ELSE v END), '{}'::jsonb) + FROM jsonb_each(COALESCE(value->'databases', '{}'::jsonb)) AS e(k, v) + )) + WHERE name = 'custom_instance_pg_databases'"#, + ) + .bind(old_id) + .bind(new_id) + .execute(&mut **tx) + .await?; + Ok(()) +} + #[derive(Deserialize)] pub(crate) struct DeleteWorkspaceQuery { pub(crate) only_delete_forks: Option, @@ -1444,10 +1461,29 @@ pub async fn drop_forked_datatable_databases( // tables, its settings row, and the database itself. Without them a save could // rename this entry, or point another one here, either side of the check below. windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?; - sqlx::query("SELECT 1 FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE") - .bind(&w_id) - .fetch_optional(&mut *tx) - .await?; + // The snapshot above was read unlocked: a save committing since could have + // repointed this entry, and the entry is removed below whatever it names by then. + let current = sqlx::query_scalar::<_, Option>( + "SELECT datatable->'datatables'->$2 FROM workspace_settings + WHERE workspace_id = $1 FOR UPDATE", + ) + .bind(&w_id) + .bind(dt_name) + .fetch_optional(&mut *tx) + .await? + .flatten() + .and_then(|v| serde_json::from_value::(v).ok()); + if !current.is_some_and(|dt| { + dt.forked_from.is_some() + && dt.database.is_some_and(|d| { + d.resource_type == database.resource_type + && &d.resource_path == db_to_drop + }) + }) { + return Err(Error::BadRequest( + "the data table changed while it was being cleaned up".to_string(), + )); + } windmill_common::datatable_roles::lock_instance_databases_governance( &mut tx, [db_to_drop.as_str()],