diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index d97cb33909..3d2853284d 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -1594,13 +1594,13 @@ async fn cleanup_waits_out_a_save_racing_it_for_an_instance_database( }; assert!( - race(false).await?.is_empty(), + matches!(race(false).await?, windmill_common::Cleanup::Dropped), "a save that rolled back kept the database, whose name then blocks every retry" ); - assert_eq!( - race(true).await?, - vec!["test-workspace".to_string()], - "the database was dropped under a save that committed a reference to it" - ); + let kept = race(true).await?; + let windmill_common::Cleanup::InUse(users) = kept else { + anyhow::bail!("the database was dropped under a save that committed a reference to it") + }; + assert_eq!(users, vec!["test-workspace".to_string()]); Ok(()) } diff --git a/backend/windmill-api-workspaces/src/datatable_clone.rs b/backend/windmill-api-workspaces/src/datatable_clone.rs index be728782af..a57a2216e5 100644 --- a/backend/windmill-api-workspaces/src/datatable_clone.rs +++ b/backend/windmill-api-workspaces/src/datatable_clone.rs @@ -83,14 +83,13 @@ pub(crate) async fn connection_snapshot( if snapshot.resources.contains_key(&path) { continue; } - let value: Option = sqlx::query_scalar( - "SELECT value FROM resource WHERE workspace_id = $1 AND path = $2", - ) - .bind(w_id) - .bind(&path) - .fetch_optional(&mut *conn) - .await? - .flatten(); + let value: Option = + sqlx::query_scalar("SELECT value FROM resource WHERE workspace_id = $1 AND path = $2") + .bind(w_id) + .bind(&path) + .fetch_optional(&mut *conn) + .await? + .flatten(); let mut strings = vec![]; collect_strings(value.as_ref(), &mut strings); for reference in strings { @@ -155,7 +154,9 @@ impl ConnectionSnapshot { { Box::pin(async move { if depth > 32 { - return Err(Error::BadRequest("resource references nest too deeply".to_string())); + return Err(Error::BadRequest( + "resource references nest too deeply".to_string(), + )); } Ok(match value { serde_json::Value::Object(map) => { @@ -174,7 +175,8 @@ impl ConnectionSnapshot { } serde_json::Value::String(s) if s.starts_with("$res:") => { let path = &s[5..]; - self.substitute(db, w_id, self.resource(path)?, depth + 1).await? + self.substitute(db, w_id, self.resource(path)?, depth + 1) + .await? } serde_json::Value::String(s) if s.starts_with("$var:") => { let path = &s[5..]; @@ -182,7 +184,9 @@ impl ConnectionSnapshot { .variables .get(path) .and_then(|v| v.as_ref()) - .ok_or_else(|| Error::NotFound(format!("variable {path} does not exist")))?; + .ok_or_else(|| { + Error::NotFound(format!("variable {path} does not exist")) + })?; serde_json::Value::String(match (secret, self.external_secrets.get(path)) { (false, _) => stored.clone(), (true, Some(external)) => external.clone(), @@ -311,14 +315,15 @@ async fn drop_copy(db: &DB, source_database: &DataTableDatabase, dbname: &str) - if source_database.resource_type == windmill_common::workspaces::DataTableCatalogResourceType::Instance { - let users = windmill_common::drop_unused_instance_database(db, dbname).await?; - if users.is_empty() { - Ok(()) - } else { - Err(Error::BadRequest(format!( + match windmill_common::drop_unused_instance_database(db, dbname).await? { + windmill_common::Cleanup::Dropped => Ok(()), + windmill_common::Cleanup::InUse(users) => Err(Error::BadRequest(format!( "kept, since workspaces {} now use it", users.join(", ") - ))) + ))), + windmill_common::Cleanup::Waiter(pid) => Err(Error::BadRequest(format!( + "kept, since a request (pid {pid}) is still waiting to name it" + ))), } } else { // On a server of the workspace's own, where a resource edited meanwhile can already name it @@ -371,8 +376,8 @@ async fn make_copy( } let server = serde_json::from_value(snapshot.resolve(db, &governing.workspace_id).await?) .map_err(|e| { - Error::internal_err(format!("Failed to parse database credentials: {e}")) - })?; + Error::internal_err(format!("Failed to parse database credentials: {e}")) + })?; (server, Some(snapshot)) }; @@ -455,10 +460,11 @@ async fn fill( // On a connection of its own: the checks below take theirs from the pool, and a lock holder // drawn from the pool too could leave a small one with nothing to give them. let database_url = windmill_common::get_database_url().await?; - let mut lock_holder = - ::connect_with(&database_url.connect_options().await?) - .await - .map_err(|e| Error::internal_err(format!("Failed to connect to the database: {e}")))?; + let mut lock_holder = ::connect_with( + &database_url.connect_options().await?, + ) + .await + .map_err(|e| Error::internal_err(format!("Failed to connect to the database: {e}")))?; let mut tx = sqlx::Connection::begin(&mut lock_holder).await?; lock_role_catalog(&mut tx).await?; lock_settings_rows(&mut tx, parent_w_id, name).await?; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 87e8ebbc4d..e29f4918fb 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1492,23 +1492,24 @@ pub async fn lock_instance_databases<'a>( } /// Drop instance database `dbname`, which a request just created, unless a workspace names it as a -/// data table or a ducklake catalog — the workspaces returned, with nothing dropped. A database is -/// registered on the instance as soon as it is created, so a superadmin can point a workspace at it -/// before the request that created it gives up on it. +/// data table or a ducklake catalog — then it is kept, and who keeps it is what comes back. A +/// database is registered on the instance as soon as it is created, so a superadmin can point a +/// workspace at it before the request that created it gives up on it. /// /// Authorization: none. Callers MUST pass only a database the same request created and has not /// handed to anything yet. /// /// The lock, the check and the drop share one connection: a second one taken from the pool while /// the first is held could wait forever on a small pool. -pub async fn drop_unused_instance_database(db: &DB, dbname: &str) -> error::Result> { +pub async fn drop_unused_instance_database(db: &DB, dbname: &str) -> error::Result { let mut conn = db.acquire().await?; let key = format!("instance_database:{dbname}"); // A save blocked on this lock is about to name the database, but it may also roll back — a // later validation of its own, a superadmin check, a cancelled request. So it is let through // and what it committed is read, rather than taken as a user: trusting it would strand the // database, whose name then blocks every retry. A fresh waiter can always arrive, so after a - // few rounds the database is kept instead, which a retry can still clean up. + // few rounds the database is kept instead; the name is then a superadmin's to drop from + // instance settings, since a retry fails on the name before reaching this cleanup. let mut rounds = 0; let dropped = loop { if let Err(e) = sqlx::query("SELECT pg_advisory_lock(hashtext($1))") @@ -1527,9 +1528,7 @@ pub async fn drop_unused_instance_database(db: &DB, dbname: &str) -> error::Resu break Err(e); } } - Ok(Cleanup::Waiter(waiter)) => break Ok(vec![waiter]), - Ok(Cleanup::InUse(users)) => break Ok(users), - Ok(Cleanup::Dropped) => break Ok(vec![]), + Ok(outcome) => break Ok(outcome), Err(e) => break Err(e), } }; @@ -1549,12 +1548,14 @@ async fn unlock_instance_database(conn: &mut sqlx::PgConnection, key: &str) -> e Ok(()) } -enum Cleanup { +/// What [`drop_unused_instance_database`] did, and for whom when it kept the database. +pub enum Cleanup { Dropped, /// Workspaces that name the database, as [`instance_database_users`] reports them. InUse(Vec), - /// A transaction is blocked on the lock, so what it will commit is not known yet. - Waiter(String), + /// A transaction still blocked on the lock after the rounds above, so what it will commit + /// stays unknown, named as its `pid`. + Waiter(i32), } async fn drop_if_unused_on(conn: &mut sqlx::PgConnection, dbname: &str) -> error::Result { @@ -1571,14 +1572,14 @@ async fn drop_if_unused_on(conn: &mut sqlx::PgConnection, dbname: &str) -> error Ok(Cleanup::Dropped) } -/// A transaction blocked on `dbname`'s instance-database lock, named as its `pid`. The lock is +/// The `pid` of a transaction blocked on `dbname`'s instance-database lock. The lock is /// taken by key, so `pg_locks` reports it split across `classid` and `objid`, and it lists every /// database on the cluster — another Windmill on the same one holds its own locks under the same /// key. async fn waiting_for_instance_database( conn: &mut sqlx::PgConnection, dbname: &str, -) -> error::Result> { +) -> error::Result> { let pid: Option = sqlx::query_scalar( "SELECT l.pid FROM pg_locks l WHERE l.locktype = 'advisory' AND NOT l.granted @@ -1590,7 +1591,7 @@ async fn waiting_for_instance_database( .bind(dbname) .fetch_optional(&mut *conn) .await?; - Ok(pid.map(|pid| format!("a request waiting to name it (pid {pid})"))) + Ok(pid) } /// The workspaces naming instance database `dbname` as a data table or a ducklake catalog. Taken @@ -1756,11 +1757,15 @@ pub async fn create_custom_instance_database( // rather than leave it behind. if let Err(e) = finish_custom_instance_database(db, dbname, tag).await { match drop_unused_instance_database(db, dbname).await { - Ok(users) if !users.is_empty() => tracing::warn!( + Ok(Cleanup::InUse(users)) => tracing::warn!( "Kept '{dbname}' after failing to set it up: workspaces {} use it", users.join(", ") ), - Ok(_) => {} + Ok(Cleanup::Waiter(pid)) => tracing::warn!( + "Kept '{dbname}' after failing to set it up: a request (pid {pid}) is still \ + waiting to name it. Drop it from instance settings once it is unused." + ), + Ok(Cleanup::Dropped) => {} Err(drop_err) => { tracing::error!("Could not drop '{dbname}' after failing to set it up: {drop_err}") }