diff --git a/backend/tests/instance_config.rs b/backend/tests/instance_config.rs index 64d7854173..882207ebfe 100644 --- a/backend/tests/instance_config.rs +++ b/backend/tests/instance_config.rs @@ -1583,3 +1583,49 @@ async fn declarative_sync_rejects_an_unusable_default_allowed_origins(db: Pool

) { + clear_settings_and_configs(&db).await; + let cluster = |host: &str, password: &str| serde_json::json!({ "host": host, "port": 5432, "user": "wm_admin", "password": password }); + sqlx::query( + "INSERT INTO global_settings (name, value) VALUES + ('external_instance_pg', $1), + ('external_instance_pg_state', '{\"databases\": {\"dt_a\": {\"success\": true}}}')", + ) + .bind(cluster("pg-a.internal", "one")) + .execute(&db) + .await + .unwrap(); + let current = BTreeMap::from([( + "external_instance_pg".to_string(), + cluster("pg-a.internal", "one"), + )]); + let sync = |value: serde_json::Value| { + let desired = BTreeMap::from([("external_instance_pg".to_string(), value)]); + let (db, current) = (db.clone(), current.clone()); + async move { + windmill_common::instance_config::sync_global_settings_declarative( + &db, ¤t, &desired, + ) + .await + } + }; + + let err = sync(cluster("pg-b.internal", "one")) + .await + .expect_err("another host must be refused while dt_a is registered"); + assert!(err.to_string().contains("dt_a"), "got: {err}"); + assert_eq!( + get_global_setting(&db, "external_instance_pg").await, + Some(cluster("pg-a.internal", "one")) + ); + + sync(cluster("PG-A.internal ", "two")) + .await + .expect("a new login on the same cluster must sync"); +} diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 00a97b8977..e96cc31fa5 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -1720,8 +1720,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; } @@ -1926,6 +1951,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(); @@ -1952,8 +1986,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 7ae427dd11..4d50198c1b 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3537,23 +3537,38 @@ async fn create_pg_database( } } - let source_kind = managed_datatable_source_kind(&db, &w_id, &req.source).await?; - if source_kind == Some(DataTableCatalogResourceType::ExternalInstance) { - windmill_common::external_instance_pg::create_external_instance_database_unchecked( - &db, - &req.target_dbname, - "datatable", - Some(&w_id), - ) - .await?; - } else if source_kind == Some(DataTableCatalogResourceType::Instance) { - windmill_common::create_custom_instance_database( - &db, - &req.target_dbname, - "datatable", - Some(&w_id), - ) - .await?; + if let Some(source_kind) = managed_datatable_source_kind(&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"))); + } + if source_kind == DataTableCatalogResourceType::ExternalInstance { + windmill_common::external_instance_pg::create_external_instance_database_unchecked( + &db, + &req.target_dbname, + "datatable", + Some(&w_id), + ) + .await?; + } else { + windmill_common::create_custom_instance_database( + &db, + &req.target_dbname, + "datatable", + Some(&w_id), + ) + .await?; + } + tx.commit().await?; } else { let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?; @@ -3709,10 +3724,16 @@ async fn import_pg_database( )); } if let Some(kind) = managed_datatable_source_kind(&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, kind, @@ -3838,20 +3859,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() { @@ -8389,6 +8428,15 @@ async fn apply_forked_datatable( ) .await?; } + if database.resource_type.is_windmill_managed() { + // 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.is_windmill_managed() && !windmill_api_auth::is_super_admin_authed(db, authed).await? { diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 9210ef7ba1..fd2edad8bf 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/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, @@ -1436,60 +1460,103 @@ 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>( + "SELECT datatable->'datatables'->$2 FROM workspace_settings + WHERE workspace_id = $1 FOR UPDATE", + ) .bind(&w_id) + .bind(&dt_name) .fetch_optional(&mut *tx) - .await?; - windmill_common::datatable_roles::lock_instance_databases_governance( - &mut tx, - [db_to_drop.as_str()], - ) - .await?; - if database.resource_type - == windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance - { - windmill_common::external_instance_pg::drop_external_instance_database_unchecked( - &db, - db_to_drop, - Some((&w_id, dt_name)), - ) - .await?; - } else { - 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(", ") - ))); + .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 == 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::drop_custom_instance_database(&db, db_to_drop).await?; + windmill_common::datatable_roles::lock_instance_databases_governance( + &mut tx, + [db_to_drop.as_str()], + ) + .await?; + if resource_type + != windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance + { + 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) + .execute(&mut *tx) + .await?; + if resource_type + == windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance + { + // Checks the uses of the database itself, and unregisters it. + windmill_common::external_instance_pg::drop_external_instance_database_unchecked( + &db, + &db_to_drop, + Some((w_id.as_str(), dt_name.as_str())), + ) + .await?; + } else { + 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>(()) } - // 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) - .execute(&mut *tx) - .await?; - tx.commit().await?; - Ok::<_, Error>(()) - } - .await; + }) + .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://{}: {}", diff --git a/backend/windmill-common/src/external_instance_pg.rs b/backend/windmill-common/src/external_instance_pg.rs index 263fca7542..2e66dcd8cf 100644 --- a/backend/windmill-common/src/external_instance_pg.rs +++ b/backend/windmill-common/src/external_instance_pg.rs @@ -129,6 +129,7 @@ pub(crate) async fn read_external_instance_pg_state<'c>( } } +/// Authorization: reads the hidden cluster state and checks nothing. Callers MUST be superadmin. pub async fn external_instance_pg_status(db: &DB) -> Result { let configured = read_external_instance_pg_config(db).await?.is_some(); let state = read_external_instance_pg_state(db).await?; @@ -184,7 +185,19 @@ pub async fn ensure_external_instance_pg_removable(conn: &mut sqlx::PgConnection if state.databases.is_empty() && usages.is_empty() { return Ok(()); } - let names = state + Err(Error::BadRequest(format!( + "The external instance cluster still holds databases in use ({}). Drop them and \ + repoint the data tables using them before removing {EXTERNAL_INSTANCE_PG_SETTING}.", + databases_in_use(&state, &usages) + ))) +} + +/// The databases on the cluster, whether Windmill created them or a data table names them. +fn databases_in_use( + state: &ExternalInstancePgState, + usages: &BTreeMap>, +) -> String { + state .databases .keys() .chain(usages.keys()) @@ -192,11 +205,7 @@ pub async fn ensure_external_instance_pg_removable(conn: &mut sqlx::PgConnection .into_iter() .cloned() .collect::>() - .join(", "); - Err(Error::BadRequest(format!( - "The external instance cluster still holds databases in use ({names}). Drop them and \ - repoint the data tables using them before removing {EXTERNAL_INSTANCE_PG_SETTING}." - ))) + .join(", ") } /// Refuse a workspace setting that newly names an `external_instance` database on an edition @@ -252,7 +261,7 @@ pub async fn drop_external_instance_database_unchecked( exempt: Option<(&str, &str)>, ) -> Result<()> { crate::external_instance_pg_oss::drop_external_instance_database_unchecked(db, dbname, exempt) - .await + .await } /// Serializes everything that changes which databases exist on the external cluster, or which data @@ -389,10 +398,11 @@ async fn ensure_external_instance_pg_not_repointed( return Ok(()); } Err(Error::BadRequest(format!( - "The external instance cluster at {}:{} still holds databases in use. Drop them and repoint \ - what uses them before pointing {EXTERNAL_INSTANCE_PG_SETTING} at another cluster.", + "The external instance cluster at {}:{} still holds databases in use ({}). Drop them and \ + repoint what uses them before pointing {EXTERNAL_INSTANCE_PG_SETTING} at another cluster.", current.host.trim(), - current.port.unwrap_or(5432) + current.port.unwrap_or(5432), + databases_in_use(&state, &usages) ))) } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 80d5eccc64..2974df5300 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1485,7 +1485,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)?; @@ -1531,14 +1550,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(()) }