diff --git a/backend/.sqlx/query-661462c1a96cab2367b2b2bf452c5ab54e8b28037dcfc83d8b03a930c8ba6589.json b/backend/.sqlx/query-661462c1a96cab2367b2b2bf452c5ab54e8b28037dcfc83d8b03a930c8ba6589.json new file mode 100644 index 0000000000..fc4681c6ca --- /dev/null +++ b/backend/.sqlx/query-661462c1a96cab2367b2b2bf452c5ab54e8b28037dcfc83d8b03a930c8ba6589.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n COALESCE(value->'databases'->$1::text->>'tag' = 'datatable', false)\n AND NOT EXISTS (\n SELECT 1 FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $1::text)\n AND NOT EXISTS (\n SELECT 1 FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.ducklake->'ducklakes', '{}'::jsonb)) dl\n WHERE dl.value->'catalog'->>'resource_type' = 'instance'\n AND dl.value->'catalog'->>'resource_path' = $1::text)\n AS \"reclaimable!\"\n FROM global_settings WHERE name = 'custom_instance_pg_databases'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "reclaimable!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "661462c1a96cab2367b2b2bf452c5ab54e8b28037dcfc83d8b03a930c8ba6589" +} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 787bd950fd..2a9245ce3e 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -598,3 +598,79 @@ async fn a_data_table_under_roles_is_not_copied_into_a_fork( ); Ok(()) } + +async fn register_instance_database( + db: &Pool, + name: &str, + tag: &str, +) -> anyhow::Result<()> { + sqlx::query( + "UPDATE global_settings SET value = jsonb_set(value, ARRAY['databases', $1], \ + jsonb_build_object('tag', $2::text)) WHERE name = 'custom_instance_pg_databases'", + ) + .bind(name) + .bind(tag) + .execute(db) + .await?; + Ok(()) +} + +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn only_an_unnamed_fork_database_is_reclaimable(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + // Reclaiming a clone's leftover is a `DROP DATABASE`, so this guard is its whole safety story: + // only a database Windmill registered for a data table, and that no entry anywhere names. + for (name, tag) in [ + ("wm_fork_orphan", "datatable"), + ("wm_fork_named", "datatable"), + ("wm_fork_lake_catalog", "datatable"), + ("wm_fork_archived", "datatable"), + ("wm_fork_ducklake", "ducklake"), + ("dt_orphan", "datatable"), + ] { + register_instance_database(&db, name, tag).await?; + } + sqlx::query( + r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,clone}', + '{"database": {"resource_type": "instance", "resource_path": "wm_fork_named"}}') + WHERE workspace_id = 'test-workspace'"#, + ) + .execute(&db) + .await?; + // A ducklake catalog on an instance database names it through `resource_path` as well. + sqlx::query( + r#"UPDATE workspace_settings SET ducklake = '{"ducklakes": {"lake": {"catalog": + {"resource_type": "instance", "resource_path": "wm_fork_lake_catalog"}}}}' + WHERE workspace_id = 'test-workspace'"#, + ) + .execute(&db) + .await?; + // An archived workspace still owns what it names: unarchiving it must find its data intact. + sqlx::query( + r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,clone}', + '{"database": {"resource_type": "instance", "resource_path": "wm_fork_archived"}}') + WHERE workspace_id = 'wm-fork-dt'"#, + ) + .execute(&db) + .await?; + sqlx::query("UPDATE workspace SET deleted = true WHERE id = 'wm-fork-dt'") + .execute(&db) + .await?; + + for (name, reclaimable) in [ + ("wm_fork_orphan", true), + ("wm_fork_named", false), + ("wm_fork_lake_catalog", false), + ("wm_fork_archived", false), + ("wm_fork_ducklake", false), + ("wm_fork_unregistered", false), + ("dt_orphan", false), + ] { + assert_eq!( + windmill_common::is_reclaimable_fork_database(&db, name).await?, + reclaimable, + "{name}" + ); + } + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index f0f99c7a73..2b1eb03b45 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3261,10 +3261,12 @@ async fn create_pg_database( // The copy this database is for is refused a call later, and nothing collects an instance // database that no data table entry names. Refuse here too, so the clone stops before one // exists rather than leaving an empty registered `wm_fork_…` behind. - if let Some(reference) = req.source.strip_prefix("datatable://") { - let name = datatable_ref_name(reference); - ensure_datatable_is_clonable(&db, &w_id, name).await?; - } + let governing = match req.source.strip_prefix("datatable://") { + Some(reference) => { + Some(ensure_datatable_is_clonable(&db, &w_id, datatable_ref_name(reference)).await?) + } + None => None, + }; // Non-superadmin: restrict dbname to wm_fork_ prefix if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? { @@ -3277,6 +3279,21 @@ async fn create_pg_database( } if is_instance_datatable_source(&db, &w_id, &req.source).await? { + // A retry after a clone that failed past this point finds its own leftover here. Reclaiming + // it is a `DROP DATABASE`, so it is for whoever administers the source — not merely whoever + // reaches it, which on a data table without roles is every member. Anyone else gets the + // refusal an existing database always got. + let may_reclaim = match &governing { + Some(governing) => crate::datatable_permissions::ensure_governs_datatable( + &db, &authed, &w_id, governing, + ) + .await + .is_ok(), + None => false, + }; + if may_reclaim { + windmill_common::reclaim_orphaned_fork_database(&db, &req.target_dbname).await?; + } windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable") .await?; } else { diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index e0a60e498b..d98581fb2a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1425,6 +1425,77 @@ pub fn validate_dbname(dbname: &str) -> error::Result<()> { Ok(()) } +/// Whether `dbname` is a clone's leftover: a `wm_fork_*` database Windmill registered as a data +/// table database and that no data table or ducklake entry names, in any workspace, archived ones +/// included. A clone creates its database a request before any entry names it, so a failure in +/// between leaves exactly this, and its name then blocks every retry. +pub async fn is_reclaimable_fork_database(db: &DB, dbname: &str) -> error::Result { + if !dbname.starts_with("wm_fork_") { + return Ok(false); + } + Ok(sqlx::query_scalar!( + r#"SELECT + COALESCE(value->'databases'->$1::text->>'tag' = 'datatable', false) + AND NOT EXISTS ( + SELECT 1 FROM workspace_settings ws + CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt + WHERE dt.value->'database'->>'resource_type' = 'instance' + AND dt.value->'database'->>'resource_path' = $1::text) + AND NOT EXISTS ( + SELECT 1 FROM workspace_settings ws + CROSS JOIN LATERAL jsonb_each(COALESCE(ws.ducklake->'ducklakes', '{}'::jsonb)) dl + WHERE dl.value->'catalog'->>'resource_type' = 'instance' + AND dl.value->'catalog'->>'resource_path' = $1::text) + AS "reclaimable!" + FROM global_settings WHERE name = 'custom_instance_pg_databases'"#, + dbname + ) + .fetch_optional(db) + .await? + .unwrap_or(false)) +} + +/// Drop `dbname` if [`is_reclaimable_fork_database`], so a retried clone can recreate it, and +/// return whether it did. Unlike [`drop_custom_instance_database`] this never terminates +/// connections: a clone still copying into the database must make the drop fail, not be cut off +/// mid-copy. +pub async fn reclaim_orphaned_fork_database(db: &DB, dbname: &str) -> error::Result { + let dbname = dbname.trim(); + validate_dbname(dbname)?; + + let exists = sqlx::query_scalar!( + "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)", + dbname + ) + .fetch_one(db) + .await? + .unwrap_or(false); + if !exists || !is_reclaimable_fork_database(db, dbname).await? { + return Ok(false); + } + + // SAFETY: `dbname` has been validated via validate_dbname() above. + sqlx::query(&format!("DROP DATABASE IF EXISTS \"{}\"", dbname)) + .execute(db) + .await + .map_err(|e| { + error::Error::BadRequest(format!( + "Database '{dbname}' is left over from an earlier clone and could not be \ + reclaimed, most often because a clone is still copying into it: {e}" + )) + })?; + // A registered database that no longer exists makes every later per-database pass fail on it. + sqlx::query!( + r#"UPDATE global_settings SET value = value #- ARRAY['databases', $1] WHERE name = 'custom_instance_pg_databases'"#, + dbname + ) + .execute(db) + .await?; + + tracing::info!("Reclaimed orphaned fork database '{dbname}'"); + Ok(true) +} + /// Drop a custom instance database: validate, terminate connections, DROP DATABASE, remove from global_settings. pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Result<()> { let dbname = dbname.trim();