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

fix(datatables): tell a waiting request apart from the workspaces using a database

Browse Source 
Both callers render what cleanup returns as the workspaces that keep the
database, so a waiting request's pid read as one of them. Each now words that
case itself, and the give-up comment names where the kept name actually goes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-21 10:29:49 +02:00
co-authored by Claude Opus 5
parent 8d53d7e385
commit 4f81392f1a
3 changed files with 56 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-integration-tests/tests/datatable_roles.rs
+6 -6
View File
@@ -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(())
}
backend/windmill-api-workspaces/src/datatable_clone.rs
+29 -23
View File
@@ -83,14 +83,13 @@ pub(crate) async fn connection_snapshot(
if snapshot.resources.contains_key(&path) {
continue;
}
let value: Option<serde_json::Value> = 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<serde_json::Value> =
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 =
<sqlx::PgConnection as sqlx::Connection>::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 = <sqlx::PgConnection as sqlx::Connection>::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?;
backend/windmill-common/src/lib.rs
+21 -16
View File
@@ -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<Vec<String>> {
pub async fn drop_unused_instance_database(db: &DB, dbname: &str) -> error::Result<Cleanup> {
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<String>),
/// 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<Cleanup> {
@@ -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<Option<String>> {
) -> error::Result<Option<i32>> {
let pid: Option<i32> = 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}")
}