mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-21 08:02:38 +00:00
fix(datatables): let a retried clone reclaim its own leftover database
A clone creates its target database one request before it copies into it, and the fork that would name it is written a request after that. Any failure in between — a pg_dump error, a bad restore, a dropped connection, the source's roles changing mid-flow — left a registered `wm_fork_*` that no entry names, and every retry then failed on its name. This predates data table roles. `create_pg_database` now reclaims such a leftover before creating: only a `wm_fork_*` database Windmill registered as a data table database and that no data table or ducklake entry names, in any workspace, archived ones included. The drop never terminates connections, so a clone still copying into it makes the reclaim fail instead of being cut off. It is limited to callers who administer the source — reaching it is not enough, since on a data table without roles every member reaches it — and anyone else gets the refusal an existing database always got. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
4600942286
commit
7dd3275a10
+22
@@ -0,0 +1,22 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "SELECT\n COALESCE(value->'databases'->$1::text->>'tag' = 'datatable', false)\n AND NOT EXISTS (\n SELECT 1 FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $1::text)\n AND NOT EXISTS (\n SELECT 1 FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.ducklake->'ducklakes', '{}'::jsonb)) dl\n WHERE dl.value->'catalog'->>'resource_type' = 'instance'\n AND dl.value->'catalog'->>'resource_path' = $1::text)\n AS \"reclaimable!\"\n FROM global_settings WHERE name = 'custom_instance_pg_databases'",
|
||||||
|
"describe": {
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"ordinal": 0,
|
||||||
|
"name": "reclaimable!",
|
||||||
|
"type_info": "Bool"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Text"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": [
|
||||||
|
null
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"hash": "661462c1a96cab2367b2b2bf452c5ab54e8b28037dcfc83d8b03a930c8ba6589"
|
||||||
|
}
|
||||||
@@ -598,3 +598,79 @@ async fn a_data_table_under_roles_is_not_copied_into_a_fork(
|
|||||||
);
|
);
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn register_instance_database(
|
||||||
|
db: &Pool<Postgres>,
|
||||||
|
name: &str,
|
||||||
|
tag: &str,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE global_settings SET value = jsonb_set(value, ARRAY['databases', $1], \
|
||||||
|
jsonb_build_object('tag', $2::text)) WHERE name = 'custom_instance_pg_databases'",
|
||||||
|
)
|
||||||
|
.bind(name)
|
||||||
|
.bind(tag)
|
||||||
|
.execute(db)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
|
||||||
|
async fn only_an_unnamed_fork_database_is_reclaimable(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||||
|
initialize_tracing().await;
|
||||||
|
// Reclaiming a clone's leftover is a `DROP DATABASE`, so this guard is its whole safety story:
|
||||||
|
// only a database Windmill registered for a data table, and that no entry anywhere names.
|
||||||
|
for (name, tag) in [
|
||||||
|
("wm_fork_orphan", "datatable"),
|
||||||
|
("wm_fork_named", "datatable"),
|
||||||
|
("wm_fork_lake_catalog", "datatable"),
|
||||||
|
("wm_fork_archived", "datatable"),
|
||||||
|
("wm_fork_ducklake", "ducklake"),
|
||||||
|
("dt_orphan", "datatable"),
|
||||||
|
] {
|
||||||
|
register_instance_database(&db, name, tag).await?;
|
||||||
|
}
|
||||||
|
sqlx::query(
|
||||||
|
r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,clone}',
|
||||||
|
'{"database": {"resource_type": "instance", "resource_path": "wm_fork_named"}}')
|
||||||
|
WHERE workspace_id = 'test-workspace'"#,
|
||||||
|
)
|
||||||
|
.execute(&db)
|
||||||
|
.await?;
|
||||||
|
// A ducklake catalog on an instance database names it through `resource_path` as well.
|
||||||
|
sqlx::query(
|
||||||
|
r#"UPDATE workspace_settings SET ducklake = '{"ducklakes": {"lake": {"catalog":
|
||||||
|
{"resource_type": "instance", "resource_path": "wm_fork_lake_catalog"}}}}'
|
||||||
|
WHERE workspace_id = 'test-workspace'"#,
|
||||||
|
)
|
||||||
|
.execute(&db)
|
||||||
|
.await?;
|
||||||
|
// An archived workspace still owns what it names: unarchiving it must find its data intact.
|
||||||
|
sqlx::query(
|
||||||
|
r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,clone}',
|
||||||
|
'{"database": {"resource_type": "instance", "resource_path": "wm_fork_archived"}}')
|
||||||
|
WHERE workspace_id = 'wm-fork-dt'"#,
|
||||||
|
)
|
||||||
|
.execute(&db)
|
||||||
|
.await?;
|
||||||
|
sqlx::query("UPDATE workspace SET deleted = true WHERE id = 'wm-fork-dt'")
|
||||||
|
.execute(&db)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
for (name, reclaimable) in [
|
||||||
|
("wm_fork_orphan", true),
|
||||||
|
("wm_fork_named", false),
|
||||||
|
("wm_fork_lake_catalog", false),
|
||||||
|
("wm_fork_archived", false),
|
||||||
|
("wm_fork_ducklake", false),
|
||||||
|
("wm_fork_unregistered", false),
|
||||||
|
("dt_orphan", false),
|
||||||
|
] {
|
||||||
|
assert_eq!(
|
||||||
|
windmill_common::is_reclaimable_fork_database(&db, name).await?,
|
||||||
|
reclaimable,
|
||||||
|
"{name}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|||||||
@@ -3261,10 +3261,12 @@ async fn create_pg_database(
|
|||||||
// The copy this database is for is refused a call later, and nothing collects an instance
|
// The copy this database is for is refused a call later, and nothing collects an instance
|
||||||
// database that no data table entry names. Refuse here too, so the clone stops before one
|
// database that no data table entry names. Refuse here too, so the clone stops before one
|
||||||
// exists rather than leaving an empty registered `wm_fork_…` behind.
|
// exists rather than leaving an empty registered `wm_fork_…` behind.
|
||||||
if let Some(reference) = req.source.strip_prefix("datatable://") {
|
let governing = match req.source.strip_prefix("datatable://") {
|
||||||
let name = datatable_ref_name(reference);
|
Some(reference) => {
|
||||||
ensure_datatable_is_clonable(&db, &w_id, name).await?;
|
Some(ensure_datatable_is_clonable(&db, &w_id, datatable_ref_name(reference)).await?)
|
||||||
}
|
}
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
|
||||||
// Non-superadmin: restrict dbname to wm_fork_ prefix
|
// Non-superadmin: restrict dbname to wm_fork_ prefix
|
||||||
if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
|
if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
|
||||||
@@ -3277,6 +3279,21 @@ async fn create_pg_database(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if is_instance_datatable_source(&db, &w_id, &req.source).await? {
|
if is_instance_datatable_source(&db, &w_id, &req.source).await? {
|
||||||
|
// A retry after a clone that failed past this point finds its own leftover here. Reclaiming
|
||||||
|
// it is a `DROP DATABASE`, so it is for whoever administers the source — not merely whoever
|
||||||
|
// reaches it, which on a data table without roles is every member. Anyone else gets the
|
||||||
|
// refusal an existing database always got.
|
||||||
|
let may_reclaim = match &governing {
|
||||||
|
Some(governing) => crate::datatable_permissions::ensure_governs_datatable(
|
||||||
|
&db, &authed, &w_id, governing,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.is_ok(),
|
||||||
|
None => false,
|
||||||
|
};
|
||||||
|
if may_reclaim {
|
||||||
|
windmill_common::reclaim_orphaned_fork_database(&db, &req.target_dbname).await?;
|
||||||
|
}
|
||||||
windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable")
|
windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable")
|
||||||
.await?;
|
.await?;
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -1425,6 +1425,77 @@ pub fn validate_dbname(dbname: &str) -> error::Result<()> {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Whether `dbname` is a clone's leftover: a `wm_fork_*` database Windmill registered as a data
|
||||||
|
/// table database and that no data table or ducklake entry names, in any workspace, archived ones
|
||||||
|
/// included. A clone creates its database a request before any entry names it, so a failure in
|
||||||
|
/// between leaves exactly this, and its name then blocks every retry.
|
||||||
|
pub async fn is_reclaimable_fork_database(db: &DB, dbname: &str) -> error::Result<bool> {
|
||||||
|
if !dbname.starts_with("wm_fork_") {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
Ok(sqlx::query_scalar!(
|
||||||
|
r#"SELECT
|
||||||
|
COALESCE(value->'databases'->$1::text->>'tag' = 'datatable', false)
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM workspace_settings ws
|
||||||
|
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
|
||||||
|
WHERE dt.value->'database'->>'resource_type' = 'instance'
|
||||||
|
AND dt.value->'database'->>'resource_path' = $1::text)
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM workspace_settings ws
|
||||||
|
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.ducklake->'ducklakes', '{}'::jsonb)) dl
|
||||||
|
WHERE dl.value->'catalog'->>'resource_type' = 'instance'
|
||||||
|
AND dl.value->'catalog'->>'resource_path' = $1::text)
|
||||||
|
AS "reclaimable!"
|
||||||
|
FROM global_settings WHERE name = 'custom_instance_pg_databases'"#,
|
||||||
|
dbname
|
||||||
|
)
|
||||||
|
.fetch_optional(db)
|
||||||
|
.await?
|
||||||
|
.unwrap_or(false))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop `dbname` if [`is_reclaimable_fork_database`], so a retried clone can recreate it, and
|
||||||
|
/// return whether it did. Unlike [`drop_custom_instance_database`] this never terminates
|
||||||
|
/// connections: a clone still copying into the database must make the drop fail, not be cut off
|
||||||
|
/// mid-copy.
|
||||||
|
pub async fn reclaim_orphaned_fork_database(db: &DB, dbname: &str) -> error::Result<bool> {
|
||||||
|
let dbname = dbname.trim();
|
||||||
|
validate_dbname(dbname)?;
|
||||||
|
|
||||||
|
let exists = sqlx::query_scalar!(
|
||||||
|
"SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)",
|
||||||
|
dbname
|
||||||
|
)
|
||||||
|
.fetch_one(db)
|
||||||
|
.await?
|
||||||
|
.unwrap_or(false);
|
||||||
|
if !exists || !is_reclaimable_fork_database(db, dbname).await? {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
// SAFETY: `dbname` has been validated via validate_dbname() above.
|
||||||
|
sqlx::query(&format!("DROP DATABASE IF EXISTS \"{}\"", dbname))
|
||||||
|
.execute(db)
|
||||||
|
.await
|
||||||
|
.map_err(|e| {
|
||||||
|
error::Error::BadRequest(format!(
|
||||||
|
"Database '{dbname}' is left over from an earlier clone and could not be \
|
||||||
|
reclaimed, most often because a clone is still copying into it: {e}"
|
||||||
|
))
|
||||||
|
})?;
|
||||||
|
// A registered database that no longer exists makes every later per-database pass fail on it.
|
||||||
|
sqlx::query!(
|
||||||
|
r#"UPDATE global_settings SET value = value #- ARRAY['databases', $1] WHERE name = 'custom_instance_pg_databases'"#,
|
||||||
|
dbname
|
||||||
|
)
|
||||||
|
.execute(db)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
tracing::info!("Reclaimed orphaned fork database '{dbname}'");
|
||||||
|
Ok(true)
|
||||||
|
}
|
||||||
|
|
||||||
/// Drop a custom instance database: validate, terminate connections, DROP DATABASE, remove from global_settings.
|
/// Drop a custom instance database: validate, terminate connections, DROP DATABASE, remove from global_settings.
|
||||||
pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Result<()> {
|
pub async fn drop_custom_instance_database(db: &DB, dbname: &str) -> error::Result<()> {
|
||||||
let dbname = dbname.trim();
|
let dbname = dbname.trim();
|
||||||
|
|||||||
Reference in New Issue
Block a user