fix(datatables): hold the fork lock across a rename's settings copy

Fork cleanup of the old id could otherwise drop a copy the renamed workspace goes on using. Also
states the authorization contract of the instance database drop helpers.

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

backend/windmill-api-settings/src/lib.rs

@@ -42,7 +42,6 @@ use axum::{
     routing::{get, post},
     Json, Router,
 };
-use serde_json::json;
 
 use serde::{Deserialize, Serialize};
 use windmill_ai::ai_cache::bump_instance_ai_config_revision;
@@ -1690,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<String> = 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;
         }
@@ -1767,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();
@@ -1793,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))

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<serde_json::Value>>(
+        "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<String, Ducklake> =
         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() {
@@ -8246,6 +8284,15 @@ async fn apply_forked_datatable(
         })?,
     };
 
+    if database.resource_type == DataTableCatalogResourceType::Instance {
+        // Held until the fork commits, as every save newly naming a database takes it: none may
+        // claim the copy between the check below and this fork's entry landing on it.
+        windmill_common::datatable_roles::lock_instance_databases_governance(
+            &mut **tx,
+            [fdt.new_dbname.as_str()],
+        )
+        .await?;
+    }
     if database.resource_type == DataTableCatalogResourceType::Instance
         && !windmill_api_auth::is_super_admin_authed(db, authed).await?
     {

backend/windmill-api-workspaces/src/workspaces_extra.rs

@@ -56,6 +56,11 @@ pub(crate) async fn change_workspace_id(
 
     let mut tx = db.begin().await?;
 
+    // The settings copy below carries every data table entry to the new id, which fork cleanup of
+    // the old id cannot see until this commits: without the lock it could drop a copy the renamed
+    // workspace goes on using. Before the pairing lock, as forking takes the two in that order.
+    windmill_common::workspaces::lock_fork_datatables(&mut tx, &old_id).await?;
+
     // A rename rewrites the workspace's dev flag and reparents its children, so it decides on the
     // same state the pairing handlers do: without this lock a concurrent create/attach could commit
     // an active dev workspace under the shell this rename is about to archive. Both ids, since the
@@ -110,21 +115,6 @@ 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?;
-
     // Duplicate workspace settings (keep copy in old workspace for reference)
     info!("Duplicating workspace_settings table");
     sqlx::query!(
@@ -865,6 +855,10 @@ pub(crate) async fn change_workspace_id(
         }
     }
 
+    // After every workspace_settings write above: fork cleanup locks a settings row before the
+    // registry, so taking the registry first here would deadlock with it.
+    migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?;
+
     // Audit log in the same transaction as the workspace changes
     audit_log(
         &mut *tx,
@@ -933,6 +927,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 +946,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<bool>,
@@ -1438,49 +1462,87 @@ pub async fn drop_forked_datatable_databases(
             // The fork's own entry is what is going away; anything else still reaching the copy,
             // a child fork's pointer at this entry included, keeps it. The lock keeps a child fork
             // from gaining such a pointer before the drop.
-            let dropped = async {
-                let mut tx = db.begin().await?;
-                // The three locks a settings save takes, in its order: this workspace's data
-                // 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")
+            // A task of its own, so a client going away cannot stop it between dropping the
+            // database and committing the entry's removal.
+            let dropped = tokio::spawn({
+                let (db, w_id, dt_name, db_to_drop) = (
+                    db.clone(),
+                    w_id.clone(),
+                    dt_name.clone(),
+                    db_to_drop.clone(),
+                );
+                let resource_type = database.resource_type;
+                async move {
+                    let mut tx = db.begin().await?;
+                    // The three locks a settings save takes, in its order: this workspace's data
+                    // 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?;
+                    // 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<serde_json::Value>>(
+                        "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::<DataTable>(v).ok());
+                    if !current.is_some_and(|dt| {
+                        dt.forked_from.is_some()
+                            && dt.database.is_some_and(|d| {
+                                d.resource_type == 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()],
+                    )
                     .await?;
-                windmill_common::datatable_roles::lock_instance_databases_governance(
-                    &mut tx,
-                    [db_to_drop.as_str()],
-                )
-                .await?;
-                let uses = windmill_common::workspaces::managed_database_uses(
-                    &mut tx,
-                    windmill_common::workspaces::DataTableCatalogResourceType::Instance,
-                    db_to_drop,
-                    Some((&w_id, dt_name)),
-                )
-                .await?;
-                if !uses.is_empty() {
-                    return Err(Error::BadRequest(format!(
-                        "it is still used by {}",
-                        uses.join(", ")
-                    )));
-                }
-                // The entry goes with the database: a fork this one is cloned into afterwards must
-                // not inherit a pointer at a data table whose database is gone.
-                sqlx::query(
+                    let uses = windmill_common::workspaces::managed_database_uses(
+                        &mut tx,
+                        windmill_common::workspaces::DataTableCatalogResourceType::Instance,
+                        &db_to_drop,
+                        Some((w_id.as_str(), dt_name.as_str())),
+                    )
+                    .await?;
+                    if !uses.is_empty() {
+                        return Err(Error::BadRequest(format!(
+                            "it is still used by {}",
+                            uses.join(", ")
+                        )));
+                    }
+                    // The entry goes with the database: a fork this one is cloned into afterwards must
+                    // not inherit a pointer at a data table whose database is gone.
+                    sqlx::query(
                     "UPDATE workspace_settings SET datatable = datatable #- ARRAY['datatables', $2]
                      WHERE workspace_id = $1",
                 )
                 .bind(&w_id)
-                .bind(dt_name)
+                .bind(&dt_name)
                 .execute(&mut *tx)
                 .await?;
-                windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
-                tx.commit().await?;
-                Ok::<_, Error>(())
-            }
-            .await;
+                    windmill_common::drop_custom_instance_database_keep_entry(&db, &db_to_drop)
+                        .await?;
+                    sqlx::query(
+                        "UPDATE global_settings SET value = value #- ARRAY['databases', $1]
+                     WHERE name = 'custom_instance_pg_databases'",
+                    )
+                    .bind(&db_to_drop)
+                    .execute(&mut *tx)
+                    .await?;
+                    tx.commit().await?;
+                    Ok::<_, Error>(())
+                }
+            })
+            .await
+            .unwrap_or_else(|e| Err(Error::internal_err(format!("cleanup task failed: {e}"))));
             if let Err(e) = dropped {
                 errors.push(format!(
                     "Could not drop instance database '{}' for datatable://{}: {}",

backend/windmill-common/src/lib.rs

@@ -1475,7 +1475,26 @@ pub fn validate_dbname(dbname: &str) -> error::Result<()> {
 }
 
 /// Drop a custom instance database: validate, terminate connections, DROP DATABASE, remove from global_settings.
+///
+/// Authorization: drops any instance database but Windmill's own and checks nothing. Callers MUST
+/// be superadmin, or have established the caller may drop this one — a fork's owner cleaning up
+/// its own copy that nothing else uses.
 pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Result<()> {
+    drop_custom_instance_database_keep_entry(db, dbname).await?;
+    sqlx::query!(
+        r#"UPDATE global_settings SET value = value #- ARRAY['databases', $1] WHERE name = 'custom_instance_pg_databases'"#,
+        dbname.trim()
+    )
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+/// [`drop_custom_instance_database`] leaving its registry entry, for a caller holding row locks in
+/// a transaction: the registry write has to go through that transaction, as waiting on another
+/// connection for a lock the transaction's own peers hold is a deadlock Postgres cannot see. Same
+/// authorization contract.
+pub async fn drop_custom_instance_database_keep_entry(db: &DB, dbname: &str) -> error::Result<()> {
     let dbname = dbname.trim();
     validate_dbname(dbname)?;
 
@@ -1521,14 +1540,6 @@ pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Resu
         tracing::info!("Database '{}' does not exist, skipping drop", dbname);
     }
 
-    // Always remove from global_settings
-    sqlx::query!(
-        r#"UPDATE global_settings SET value = value #- ARRAY['databases', $1] WHERE name = 'custom_instance_pg_databases'"#,
-        dbname
-    )
-    .execute(db)
-    .await?;
-
     Ok(())
 }