starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-17 16:03:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/datatable-external-instance' into datatable-external-instance-2

Browse Source 
# Conflicts:
#	backend/windmill-api-workspaces/src/workspaces.rs
#	backend/windmill-api-workspaces/src/workspaces_extra.rs
#	backend/windmill-common/src/external_instance_pg.rs
This commit is contained in:
Diego Imbert
2026-09-20 13:22:03 +02:00
parent cd4f6edad3 c96ddc0333
commit 36069773c3
6 changed files with 329 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-workspaces/src/workspaces.rs
+74 -26
View File
@@ -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<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() {
@@ -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?
{
backend/windmill-api-workspaces/src/workspaces_extra.rs
+132 -65
View File
@@ -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>,
@@ -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<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?;
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::<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::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://{}: {}",