starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(datatables): serialize fork database reservations, cleanup, setup and Ducklake saves on the database

Browse Source 
- A fork copy is created and registered under its workspace's fork lock, and refused once the
  workspace is archived; a rename re-migrates reservations under that lock after archiving.
- Ducklake saves lock every instance database they newly name, as data table saves do.
- Instance database setup holds the database's lock until its entry is written.
- Fork import also holds the database's lock across the restore.
- Cleanup re-reads the entry under its locks before dropping anything.
- Non-superadmins no longer see fork copies reserved for workspaces they are not in.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-18 15:27:35 +02:00
co-authored by Claude Opus 5
parent 5aa09bc871
commit e29dacb938
3 changed files with 139 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-settings/src/lib.rs
+38 -3
View File
@@ -1689,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;
}
@@ -1766,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();
@@ -1792,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
+47 -9
View File
@@ -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() {
backend/windmill-api-workspaces/src/workspaces_extra.rs
+54 -18
View File
@@ -110,20 +110,7 @@ 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?;
migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?;
// Duplicate workspace settings (keep copy in old workspace for reference)
info!("Duplicating workspace_settings table");
@@ -933,6 +920,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 +939,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>,
@@ -1444,10 +1461,29 @@ pub async fn drop_forked_datatable_databases(
// 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?;
// 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 == database.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()],