fix(datatables): let a retried clone reclaim its own leftover database

A clone creates its target database one request before it copies into it, and
the fork that would name it is written a request after that. Any failure in
between — a pg_dump error, a bad restore, a dropped connection, the source's
roles changing mid-flow — left a registered `wm_fork_*` that no entry names,
and every retry then failed on its name. This predates data table roles.

`create_pg_database` now reclaims such a leftover before creating: only 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.
The drop never terminates connections, so a clone still copying into it makes
the reclaim fail instead of being cut off. It is limited to callers who
administer the source — reaching it is not enough, since on a data table
without roles every member reaches it — and anyone else gets the refusal an
existing database always got.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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"
+}

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

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 {

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<bool> {
+    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<bool> {
+    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();