mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-21 00:02:30 +00:00
Merge commit '45e24e509862ecbd87ad9ab7d90910f9a5a163ad' into HEAD
# Conflicts: # backend/windmill-api-workspaces/src/workspaces.rs # backend/windmill-api-workspaces/src/workspaces_extra.rs
This commit is contained in:
@@ -1720,7 +1720,15 @@ async fn list_custom_instance_pg_databases(
|
||||
))
|
||||
})?;
|
||||
|
||||
if windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
|
||||
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.
|
||||
for entry in result.values_mut() {
|
||||
entry.workspace_id = None;
|
||||
}
|
||||
return Ok(Json(result));
|
||||
}
|
||||
{
|
||||
// Enrich each database with the list of workspaces referencing it through
|
||||
// either a ducklake catalog or a datatable database whose resource_type is
|
||||
// 'instance'. Not stored in DB to avoid drift.
|
||||
@@ -1919,16 +1927,6 @@ 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?;
|
||||
// A re-run keeps the fork reservation: without it, the workspace the copy was made for could no
|
||||
// longer import into it or finish its fork.
|
||||
let workspace_id = sqlx::query_scalar::<_, Option<String>>(
|
||||
"SELECT value->'databases'->$1->>'workspace_id' FROM global_settings
|
||||
WHERE name = 'custom_instance_pg_databases'",
|
||||
)
|
||||
.bind(&dbname)
|
||||
.fetch_optional(&db)
|
||||
.await?
|
||||
.flatten();
|
||||
let mut logs = CustomInstanceDbLogs::default();
|
||||
let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
|
||||
let success = result.is_ok();
|
||||
@@ -1939,14 +1937,25 @@ async fn setup_custom_instance_pg_database(
|
||||
error,
|
||||
tag: body.tag,
|
||||
used_by_workspaces: vec![],
|
||||
workspace_id,
|
||||
workspace_id: None,
|
||||
};
|
||||
let status_json = serde_json::to_value(&status).map_err(to_anyhow)?;
|
||||
// Save that the database was setup successfully
|
||||
sqlx::query!(
|
||||
r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', (COALESCE(value->'databases', '{}'::jsonb) || to_jsonb($1::json))) WHERE name = 'custom_instance_pg_databases'"#,
|
||||
json!({ dbname: status_json })
|
||||
).execute(&db).await?;
|
||||
// The fork reservation is carried over inside the write, from whatever the row holds then: a
|
||||
// rename migrating it while the setup above ran would otherwise be overwritten with the value
|
||||
// this request started from, stranding the copy under the archived workspace.
|
||||
let saved = sqlx::query_scalar::<_, serde_json::Value>(
|
||||
r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}',
|
||||
COALESCE(value->'databases', '{}'::jsonb)
|
||||
|| jsonb_build_object($1::text, $2::jsonb || jsonb_build_object(
|
||||
'workspace_id', value->'databases'->$1::text->'workspace_id')))
|
||||
WHERE name = 'custom_instance_pg_databases'
|
||||
RETURNING value->'databases'->$1::text"#,
|
||||
)
|
||||
.bind(&dbname)
|
||||
.bind(&status_json)
|
||||
.fetch_one(&db)
|
||||
.await?;
|
||||
let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;
|
||||
|
||||
Ok(Json(status))
|
||||
}
|
||||
|
||||
@@ -3695,6 +3695,7 @@ async fn import_pg_database(
|
||||
}
|
||||
|
||||
let schema_only = req.fork_behavior == DataTableForkBehavior::SchemaOnly;
|
||||
let mut fork_lock: Option<Transaction<'_, Postgres>> = None;
|
||||
let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
|
||||
let mut target_pg =
|
||||
resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.target).await?;
|
||||
@@ -3708,6 +3709,10 @@ 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.
|
||||
let mut tx = db.begin().await?;
|
||||
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
|
||||
windmill_common::ensure_fork_database_available_to(
|
||||
&db,
|
||||
kind,
|
||||
@@ -3715,6 +3720,7 @@ async fn import_pg_database(
|
||||
&w_id,
|
||||
)
|
||||
.await?;
|
||||
fork_lock = Some(tx);
|
||||
}
|
||||
}
|
||||
target_pg.dbname = override_dbname.clone();
|
||||
@@ -3738,6 +3744,9 @@ async fn import_pg_database(
|
||||
)
|
||||
.await?;
|
||||
pg_import_dump(&target_pg, &dump_file).await?;
|
||||
if let Some(tx) = fork_lock {
|
||||
tx.commit().await?;
|
||||
}
|
||||
|
||||
Ok(format!(
|
||||
"Imported from '{}' into '{}'",
|
||||
@@ -3917,6 +3926,8 @@ async fn edit_datatable_config(
|
||||
let is_superadmin = require_super_admin(&db, &authed).await.is_ok();
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
// Ahead of the settings row, as fork cleanup of this workspace takes the two.
|
||||
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
|
||||
|
||||
// Read under the row lock this transaction will write with. `permissions`, `reference` and
|
||||
// `forked_from` are carried across from what this read returns, so a permissions save
|
||||
@@ -4165,10 +4176,38 @@ async fn edit_datatable_config(
|
||||
})
|
||||
.collect();
|
||||
// Another workspace turning roles on for the same database holds only its own settings row, so
|
||||
// without this the scan below could read past its uncommitted write.
|
||||
// without this the scan below could read past its uncommitted write. Every managed database
|
||||
// this save newly names is locked, not just the ones the scan is about: fork cleanup takes the
|
||||
// same lock to decide nothing uses the database it is dropping.
|
||||
let newly_named: std::collections::BTreeSet<&str> = new_config
|
||||
.settings
|
||||
.datatables
|
||||
.iter()
|
||||
.filter_map(|(name, dt)| {
|
||||
let db = dt
|
||||
.database
|
||||
.as_ref()
|
||||
.filter(|d| d.resource_type == DataTableCatalogResourceType::Instance)?;
|
||||
let lookup = rename_src
|
||||
.get(name.as_str())
|
||||
.copied()
|
||||
.unwrap_or(name.as_str());
|
||||
old_datatables
|
||||
.get(lookup)
|
||||
.and_then(|old| old.database.as_ref())
|
||||
.is_none_or(|old_db| {
|
||||
old_db.resource_type != db.resource_type
|
||||
|| old_db.resource_path != db.resource_path
|
||||
})
|
||||
.then_some(db.resource_path.as_str())
|
||||
})
|
||||
.collect();
|
||||
windmill_common::datatable_roles::lock_instance_databases_governance(
|
||||
&mut *tx,
|
||||
newly_pointed.iter().map(|(_, dbname)| *dbname),
|
||||
newly_pointed
|
||||
.iter()
|
||||
.map(|(_, dbname)| *dbname)
|
||||
.chain(newly_named.iter().copied()),
|
||||
)
|
||||
.await?;
|
||||
let governed_elsewhere: Vec<String> = if newly_pointed.is_empty() {
|
||||
|
||||
@@ -1438,7 +1438,19 @@ pub async fn drop_forked_datatable_databases(
|
||||
// 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")
|
||||
.bind(&w_id)
|
||||
.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
|
||||
{
|
||||
@@ -1464,6 +1476,16 @@ pub async fn drop_forked_datatable_databases(
|
||||
}
|
||||
windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
|
||||
}
|
||||
// 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>(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user