fix(datatables): drop the serde_json::json import left unused by the fork setup write

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/windmill-api-settings/src/lib.rs

@@ -42,7 +42,6 @@ use axum::{
     routing::{get, post},
     Json, Router,
 };
-use serde_json::json;
 
 use serde::{Deserialize, Serialize};
 use windmill_ai::ai_cache::bump_instance_ai_config_revision;
@@ -1720,7 +1719,15 @@ async fn list_custom_instance_pg_databases(
             ))
         })?;
 
-    if windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
+    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.
+        for entry in result.values_mut() {
+            entry.workspace_id = None;
+        }
+        return Ok(Json(result));
+    }
+    {
         // Enrich each database with the list of workspaces referencing it through
         // either a ducklake catalog or a datatable database whose resource_type is
         // 'instance'. Not stored in DB to avoid drift.
@@ -1919,16 +1926,6 @@ 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?;
-    // A re-run keeps the fork reservation: without it, the workspace the copy was made for could no
-    // longer import into it or finish its fork.
-    let workspace_id = sqlx::query_scalar::<_, Option<String>>(
-        "SELECT value->'databases'->$1->>'workspace_id' FROM global_settings
-         WHERE name = 'custom_instance_pg_databases'",
-    )
-    .bind(&dbname)
-    .fetch_optional(&db)
-    .await?
-    .flatten();
     let mut logs = CustomInstanceDbLogs::default();
     let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
     let success = result.is_ok();
@@ -1939,14 +1936,25 @@ async fn setup_custom_instance_pg_database(
         error,
         tag: body.tag,
         used_by_workspaces: vec![],
-        workspace_id,
+        workspace_id: None,
     };
     let status_json = serde_json::to_value(&status).map_err(to_anyhow)?;
-    // Save that the database was setup successfully
-    sqlx::query!(
-        r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', (COALESCE(value->'databases', '{}'::jsonb) || to_jsonb($1::json))) WHERE name = 'custom_instance_pg_databases'"#,
-        json!({ dbname: status_json })
-    ).execute(&db).await?;
+    // The fork reservation is carried over inside the write, from whatever the row holds then: a
+    // rename migrating it while the setup above ran would otherwise be overwritten with the value
+    // this request started from, stranding the copy under the archived workspace.
+    let saved = sqlx::query_scalar::<_, serde_json::Value>(
+        r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}',
+               COALESCE(value->'databases', '{}'::jsonb)
+                   || jsonb_build_object($1::text, $2::jsonb || jsonb_build_object(
+                          'workspace_id', value->'databases'->$1::text->'workspace_id')))
+           WHERE name = 'custom_instance_pg_databases'
+           RETURNING value->'databases'->$1::text"#,
+    )
+    .bind(&dbname)
+    .bind(&status_json)
+    .fetch_one(&db)
+    .await?;
+    let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;
 
     Ok(Json(status))
 }

backend/windmill-api-workspaces/src/workspaces.rs

@@ -3680,6 +3680,7 @@ async fn import_pg_database(
     }
 
     let schema_only = req.fork_behavior == DataTableForkBehavior::SchemaOnly;
+    let mut fork_lock: Option<Transaction<'_, Postgres>> = None;
     let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
     let mut target_pg =
         resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.target).await?;
@@ -3693,6 +3694,10 @@ 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.
+                let mut tx = db.begin().await?;
+                windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
                 windmill_common::ensure_fork_database_available_to(
                     &db,
                     kind,
@@ -3700,6 +3705,7 @@ async fn import_pg_database(
                     &w_id,
                 )
                 .await?;
+                fork_lock = Some(tx);
             }
         }
         target_pg.dbname = override_dbname.clone();
@@ -3723,6 +3729,9 @@ async fn import_pg_database(
     )
     .await?;
     pg_import_dump(&target_pg, &dump_file).await?;
+    if let Some(tx) = fork_lock {
+        tx.commit().await?;
+    }
 
     Ok(format!(
         "Imported from '{}' into '{}'",
@@ -3917,6 +3926,8 @@ async fn edit_datatable_config(
     let is_superadmin = require_super_admin(&db, &authed).await.is_ok();
 
     let mut tx = db.begin().await?;
+    // Ahead of the settings row, as fork cleanup of this workspace takes the two.
+    windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
 
     // Read under the row lock this transaction will write with. `permissions`, `reference` and
     // `forked_from` are carried across from what this read returns, so a permissions save
@@ -4106,6 +4117,14 @@ async fn edit_datatable_config(
         if unchanged {
             continue;
         }
+        // Before the registration check, whose refusal would otherwise tell a workspace admin
+        // which databases exist on the cluster.
+        if !is_superadmin {
+            return Err(Error::BadRequest(
+                "Only superadmins can create or modify data tables with Instance databases"
+                    .to_string(),
+            ));
+        }
         if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
             windmill_common::external_instance_pg::ensure_external_instance_available()?;
             windmill_common::external_instance_pg::ensure_external_instance_database_registered(
@@ -4114,12 +4133,6 @@ async fn edit_datatable_config(
             )
             .await?;
         }
-        if !is_superadmin {
-            return Err(Error::BadRequest(
-                "Only superadmins can create or modify data tables with Instance databases"
-                    .to_string(),
-            ));
-        }
     }
 
     // Worked out from the locked entries rather than taken from `deleted_datatables`: a settings
@@ -4163,10 +4176,38 @@ async fn edit_datatable_config(
         })
         .collect();
     // Another workspace turning roles on for the same database holds only its own settings row, so
-    // without this the scan below could read past its uncommitted write.
+    // without this the scan below could read past its uncommitted write. Every managed database
+    // this save newly names is locked, not just the ones the scan is about: fork cleanup takes the
+    // same lock to decide nothing uses the database it is dropping.
+    let newly_named: std::collections::BTreeSet<&str> = new_config
+        .settings
+        .datatables
+        .iter()
+        .filter_map(|(name, dt)| {
+            let db = dt
+                .database
+                .as_ref()
+                .filter(|d| d.resource_type == DataTableCatalogResourceType::Instance)?;
+            let lookup = rename_src
+                .get(name.as_str())
+                .copied()
+                .unwrap_or(name.as_str());
+            old_datatables
+                .get(lookup)
+                .and_then(|old| old.database.as_ref())
+                .is_none_or(|old_db| {
+                    old_db.resource_type != db.resource_type
+                        || old_db.resource_path != db.resource_path
+                })
+                .then_some(db.resource_path.as_str())
+        })
+        .collect();
     windmill_common::datatable_roles::lock_instance_databases_governance(
         &mut *tx,
-        newly_pointed.iter().map(|(_, dbname)| *dbname),
+        newly_pointed
+            .iter()
+            .map(|(_, dbname)| *dbname)
+            .chain(newly_named.iter().copied()),
     )
     .await?;
     let governed_elsewhere: Vec<String> = if newly_pointed.is_empty() {

backend/windmill-api-workspaces/src/workspaces_extra.rs

@@ -1438,7 +1438,19 @@ pub async fn drop_forked_datatable_databases(
             // 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")
+                    .bind(&w_id)
+                    .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
                 {
@@ -1464,6 +1476,16 @@ pub async fn drop_forked_datatable_databases(
                     }
                     windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
                 }
+                // 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>(())
             }

backend/windmill-api/openapi.yaml

@@ -34089,7 +34089,7 @@ components:
           type: array
           items:
             type: string
-          description: Workspaces that reference this database via a ducklake catalog or datatable database with resource_type 'instance'. Computed at request time, not persisted.
+          description: Workspaces that reference this database through a ducklake catalog or a datatable database of the kind being listed — 'instance' for the instance databases endpoint, 'external_instance' for the external cluster one. Computed at request time, not persisted, and only returned to superadmins.
         workspace_id:
           type: string
           description: The workspace a member created this database for as a fork copy. Only that workspace can import into it or point a fork at it.

backend/windmill-common/src/external_instance_pg.rs

@@ -195,9 +195,9 @@ pub async fn external_instance_database_usages<'c>(
 /// Refuse to unset the cluster while Windmill still has databases on it, or a workspace still
 /// points at one: every data table there would stop resolving. Allowed on every edition, so a
 /// downgraded instance can still clear a setting it no longer uses.
-pub async fn ensure_external_instance_pg_removable(db: &DB) -> Result<()> {
-    let state = read_external_instance_pg_state(db).await?;
-    let usages = external_instance_database_usages(db).await?;
+pub async fn ensure_external_instance_pg_removable(conn: &mut sqlx::PgConnection) -> Result<()> {
+    let state = read_external_instance_pg_state(&mut *conn).await?;
+    let usages = external_instance_database_usages(&mut *conn).await?;
     if state.databases.is_empty() && usages.is_empty() {
         return Ok(());
     }
@@ -287,6 +287,9 @@ pub async fn lock_external_instance_pg_state(
 /// Refuse a data table naming `dbname` unless Windmill created it on the external cluster. Takes
 /// the lock drops take, so none can remove the database before `tx`, which saves the data table,
 /// commits.
+///
+/// Authorization: its refusal says whether Windmill created a database of that name, which is
+/// instance-wide knowledge. Callers MUST have authorized the caller as superadmin first.
 pub async fn ensure_external_instance_database_registered(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     dbname: &str,
@@ -326,9 +329,12 @@ pub async fn write_external_instance_pg_setting(
     };
     let mut tx = db.begin().await?;
     lock_external_instance_pg_state(&mut tx).await?;
+    // Every check runs on this transaction's own connection: it holds the advisory lock, and
+    // taking a second connection from the pool while other writers queue on that lock is how a
+    // small pool deadlocks.
     match value {
         None => {
-            ensure_external_instance_pg_removable(db).await?;
+            ensure_external_instance_pg_removable(&mut tx).await?;
             sqlx::query("DELETE FROM global_settings WHERE name = $1")
                 .bind(EXTERNAL_INSTANCE_PG_SETTING)
                 .execute(&mut *tx)
@@ -336,7 +342,7 @@ pub async fn write_external_instance_pg_setting(
         }
         Some(value) => {
             crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?;
-            ensure_external_instance_pg_not_repointed(db, value).await?;
+            ensure_external_instance_pg_not_repointed(&mut tx, value).await?;
             sqlx::query(
                 "INSERT INTO global_settings (name, value) VALUES ($1, $2)
                  ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
@@ -382,10 +388,10 @@ pub async fn write_external_instance_pg_from_diff(
 /// Data tables name databases, not clusters, so they would silently resolve to whatever the new
 /// cluster holds under the same names. Other fields (admin login, sslmode) may change freely.
 async fn ensure_external_instance_pg_not_repointed(
-    db: &DB,
+    conn: &mut sqlx::PgConnection,
     value: &serde_json::Value,
 ) -> Result<()> {
-    let Some(current) = read_external_instance_pg_config(db).await? else {
+    let Some(current) = read_external_instance_pg_config(&mut *conn).await? else {
         return Ok(());
     };
     let Ok(desired) = serde_json::from_value::<ExternalInstancePg>(value.clone()) else {
@@ -394,8 +400,8 @@ async fn ensure_external_instance_pg_not_repointed(
     if external_instance_pg_address(&current) == external_instance_pg_address(&desired) {
         return Ok(());
     }
-    let state = read_external_instance_pg_state(db).await?;
-    let usages = external_instance_database_usages(db).await?;
+    let state = read_external_instance_pg_state(&mut *conn).await?;
+    let usages = external_instance_database_usages(&mut *conn).await?;
     if state.databases.is_empty() && usages.is_empty() {
         return Ok(());
     }

backend/windmill-worker/src/duckdb_executor.rs

@@ -1475,6 +1475,7 @@ pub async fn do_duckdb(
                             &job.id,
                             client,
                             &mut hidden_passwords,
+                            job_dir,
                         )
                         .await?,
                     );
@@ -1490,12 +1491,13 @@ pub async fn do_duckdb(
                     &mut hidden_passwords,
                     &job.workspace_id,
                     materialize.as_ref().map(|(_, m)| m.asset_path.as_str()),
+                    job_dir,
                 )
                 .await?
                 {
                     probe_blocks.extend(q);
                 } else if let Some(q) =
-                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
+                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job, job_dir)
                         .await?
                 {
                     probe_blocks.extend(q);
@@ -1552,6 +1554,7 @@ pub async fn do_duckdb(
                             &job.id,
                             client,
                             &mut hidden_passwords,
+                            job_dir,
                         )
                         .await?,
                     );
@@ -1567,12 +1570,13 @@ pub async fn do_duckdb(
                     &mut hidden_passwords,
                     &job.workspace_id,
                     materialize.as_ref().map(|(_, m)| m.asset_path.as_str()),
+                    job_dir,
                 )
                 .await?
                 {
                     v.extend(ducklake_query);
                 } else if let Some(datatable_query) =
-                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
+                    transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job, job_dir)
                         .await?
                 {
                     v.extend(datatable_query);
@@ -2246,10 +2250,16 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option<ParsedAttachDbResource
 /// A connection that explicitly refuses invalid certificates — the external instance cluster's —
 /// keeps its mode instead: under `require` its shared password would go to whichever server
 /// answers. DuckDB's libpq takes one root file, so it gets the system bundle plus the configured
-/// certificate.
-fn pg_attach_verification(res: &PgDatabase) -> Result<Option<(&str, std::path::PathBuf)>> {
+/// certificate, written in the job directory: a resource's certificate is workspace-controlled, so
+/// a file per distinct one has to go with the job rather than pile up on the worker.
+fn pg_attach_verification<'a>(
+    res: &'a PgDatabase,
+    job_dir: &str,
+) -> Result<Option<(&'a str, std::path::PathBuf)>> {
     let mode = match res.sslmode.as_deref() {
-        Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode,
+        Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => {
+            mode
+        }
         _ => return Ok(None),
     };
     let bundle = windmill_common::system_ca_bundle()
@@ -2265,58 +2275,20 @@ fn pg_attach_verification(res: &PgDatabase) -> Result<Option<(&str, std::path::P
     }
     let roots = format!("{bundle}\n{pem}\n");
     use sha2::Digest;
-    let dir = std::env::temp_dir().join("windmill-pg-roots");
-    let path = dir.join(format!(
-        "{}.pem",
+    let path = std::path::Path::new(job_dir).join(format!(
+        "pg_roots_{}.pem",
         hex::encode(&sha2::Sha256::digest(roots.as_bytes())[..8])
     ));
-    let write_err = |e: std::io::Error| {
-        Error::ExecutionErr(format!("Failed to write root certificates: {e}"))
-    };
-    if path.is_file() {
-        // Marks it recently used, so pruning takes the others first.
-        let _ = std::fs::File::options()
-            .append(true)
-            .open(&path)
-            .and_then(|f| f.set_modified(std::time::SystemTime::now()));
-    } else {
-        std::fs::create_dir_all(&dir).map_err(write_err)?;
-        // Renamed into place: a job attaching concurrently must never read a half-written file.
-        let partial = path.with_extension(format!("{}.partial", Uuid::new_v4()));
-        std::fs::write(&partial, &roots)
-            .and_then(|()| std::fs::rename(&partial, &path))
-            .map_err(write_err)?;
-        prune_pg_roots(&dir, &path);
+    if !path.is_file() {
+        std::fs::write(&path, &roots)
+            .map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?;
     }
     Ok(Some((mode, path)))
 }
 
-/// Root files outlive the job: a resource's certificate is workspace-controlled, so each distinct
-/// one would otherwise add a file forever. Keeps the most recently used ones.
-const PG_ROOTS_KEPT: usize = 32;
-
-fn prune_pg_roots(dir: &std::path::Path, keep: &std::path::Path) {
-    let Ok(entries) = std::fs::read_dir(dir) else {
-        return;
-    };
-    let mut files: Vec<(std::time::SystemTime, std::path::PathBuf)> = entries
-        .filter_map(|e| e.ok())
-        .map(|e| e.path())
-        .filter(|p| p.extension().is_some_and(|x| x == "pem") && p != keep)
-        .filter_map(|p| Some((std::fs::metadata(&p).ok()?.modified().ok()?, p)))
-        .collect();
-    if files.len() < PG_ROOTS_KEPT {
-        return;
-    }
-    files.sort();
-    for (_, p) in &files[..=files.len() - PG_ROOTS_KEPT] {
-        let _ = std::fs::remove_file(p);
-    }
-}
-
-fn pg_attach_uri(res: &PgDatabase) -> Result<String> {
+fn pg_attach_uri(res: &PgDatabase, job_dir: &str) -> Result<String> {
     let uri = res.to_uri();
-    let Some((mode, roots)) = pg_attach_verification(res)? else {
+    let Some((mode, roots)) = pg_attach_verification(res, job_dir)? else {
         return Ok(uri);
     };
     let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| {
@@ -2328,11 +2300,11 @@ fn pg_attach_uri(res: &PgDatabase) -> Result<String> {
     ))
 }
 
-fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result<String> {
+fn format_attach_db_conn_str(db_resource: Value, db_type: &str, job_dir: &str) -> Result<String> {
     let s = match db_type.to_lowercase().as_str() {
         "postgres" | "postgresql" => {
             let res: PgDatabase = serde_json::from_value(db_resource)?;
-            pg_attach_uri(&res)?
+            pg_attach_uri(&res, job_dir)?
         }
         #[cfg(feature = "mysql")]
         "mysql" => {
@@ -2404,6 +2376,7 @@ async fn transform_attach_db_resource_query(
     job_id: &Uuid,
     client: &AuthedClient,
     hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
+    job_dir: &str,
 ) -> Result<Vec<String>> {
     let db_resource: Value = client
         .get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string()))
@@ -2411,8 +2384,14 @@ async fn transform_attach_db_resource_query(
     if let Some(pwd) = db_resource.get("password").and_then(|p| p.as_str()) {
         hidden_passwords.lock().unwrap().push(pwd.to_string());
     }
-    db_resource_to_attach_statements(db_resource, parsed.name, parsed.db_type, parsed.extra_args)
-        .await
+    db_resource_to_attach_statements(
+        db_resource,
+        parsed.name,
+        parsed.db_type,
+        parsed.extra_args,
+        job_dir,
+    )
+    .await
 }
 
 async fn db_resource_to_attach_statements(
@@ -2420,11 +2399,12 @@ async fn db_resource_to_attach_statements(
     ident_name: &str,
     db_type: &str,
     extra_args: Option<&str>,
+    job_dir: &str,
 ) -> Result<Vec<String>> {
     // Escape single quotes: the connection string is built from resource fields
     // (host/db/user/password) and embedded in a single-quoted DuckDB literal, so an
     // unescaped quote in any field would otherwise break out of the ATTACH statement.
-    let conn_str = format_attach_db_conn_str(db_resource, db_type)?.replace('\'', "''");
+    let conn_str = format_attach_db_conn_str(db_resource, db_type, job_dir)?.replace('\'', "''");
     let attach_str = format!(
         "ATTACH '{}' as {} (TYPE {}{});",
         conn_str,
@@ -2447,6 +2427,7 @@ async fn transform_attach_ducklake(
     hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
     w_id: &str,
     materialize_target: Option<&str>,
+    job_dir: &str,
 ) -> Result<Option<Vec<String>>> {
     lazy_static::lazy_static! {
         static ref RE: regex::Regex = regex::Regex::new(r"(?i)ATTACH\s*'ducklake(://[^':]+)?'\s*AS\s+([^ ;]+)\s*(\([^)]*\))?").unwrap();
@@ -2497,7 +2478,7 @@ async fn transform_attach_ducklake(
     // single-quoted DuckDB literals below, so an unescaped quote in a resource
     // field would break out of the ATTACH statement.
     let db_conn_str =
-        format_attach_db_conn_str(ducklake.catalog_resource, db_type)?.replace('\'', "''");
+        format_attach_db_conn_str(ducklake.catalog_resource, db_type, job_dir)?.replace('\'', "''");
     let storage = ducklake
         .storage
         .storage
@@ -2552,6 +2533,7 @@ async fn transform_attach_ducklake(
             defer,
             materialize_target,
             hidden_passwords,
+            job_dir,
         )?);
     }
     Ok(Some(statements))
@@ -2585,6 +2567,7 @@ fn fork_defer_statements(
     defer: &windmill_common::workspaces::DucklakeForkDefer,
     materialize_target: Option<&str>,
     hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
+    job_dir: &str,
 ) -> Result<Vec<String>> {
     let mut stmts = vec![];
     if defer.ancestors.is_empty() {
@@ -2603,7 +2586,7 @@ fn fork_defer_statements(
         };
         stmts.push(get_attach_db_install_str(db_type)?.to_string());
         let conn_str =
-            format_attach_db_conn_str(a.catalog_resource.clone(), db_type)?.replace('\'', "''");
+            format_attach_db_conn_str(a.catalog_resource.clone(), db_type, job_dir)?.replace('\'', "''");
         let storage = a
             .storage
             .storage
@@ -2722,6 +2705,7 @@ async fn transform_attach_datatable(
     conn: &Connection,
     hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
     job: &MiniPulledJob,
+    job_dir: &str,
 ) -> Result<Option<Vec<String>>> {
     let Some(attached) = parse_attach_datatable(query) else {
         return Ok(None);
@@ -2764,6 +2748,7 @@ async fn transform_attach_datatable(
     Ok(Some(pg_secret_attach_statements(
         db_resource,
         attached.alias,
+        job_dir,
     )?))
 }
 
@@ -2784,14 +2769,18 @@ fn datatable_secret_name(alias: &str) -> String {
 
 /// ATTACH a datatable's postgres database through a DuckDB TEMPORARY SECRET holding
 /// the connection parameters; only sslmode rides in the ATTACH string.
-fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<Vec<String>> {
+fn pg_secret_attach_statements(
+    db_resource: Value,
+    alias_name: &str,
+    job_dir: &str,
+) -> Result<Vec<String>> {
     let res: PgDatabase = serde_json::from_value(db_resource)?;
     // Escape single quotes: each field is embedded in a single-quoted DuckDB literal,
     // so an unescaped quote would break out of the CREATE SECRET statement.
     let esc = |s: &str| s.replace('\'', "''");
     // The postgres secret type has no sslmode parameter, so it goes in the ATTACH
     // string; only the libpq values PgDatabase::to_uri collapses to are forwarded.
-    let sslmode = match pg_attach_verification(&res)? {
+    let sslmode = match pg_attach_verification(&res, job_dir)? {
         // A libpq keyword/value string: the path is quoted for libpq, then for the DuckDB literal.
         Some((mode, roots)) => format!(
             "{mode} sslrootcert=''{}''",
@@ -2900,6 +2889,9 @@ mod tests {
 
     #[test]
     fn pg_attach_keeps_verification_only_when_required() {
+        let job_dir = std::env::temp_dir().join(format!("wm-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&job_dir).unwrap();
+        let job_dir = job_dir.to_string_lossy().to_string();
         let pg = |sslmode: &str, accept_invalid_certs: Option<bool>| PgDatabase {
             host: "db.internal".to_string(),
             user: Some("custom_instance_user".to_string()),
@@ -2912,30 +2904,35 @@ mod tests {
             use_iam_auth: None,
             region: None,
         };
-        let uri = pg_attach_uri(&pg("verify-full", Some(false))).unwrap();
+        let uri = pg_attach_uri(&pg("verify-full", Some(false)), &job_dir).unwrap();
         assert!(uri.contains("?sslmode=verify-full&sslrootcert="), "{uri}");
         let root = urlencoding::decode(uri.split("sslrootcert=").nth(1).unwrap()).unwrap();
-        let roots = std::fs::read_to_string(root.as_ref()).unwrap();
-        for i in 0..(PG_ROOTS_KEPT + 5) {
+        assert!(std::fs::read_to_string(root.as_ref())
+            .unwrap()
+            .contains("-----BEGIN CERTIFICATE-----test"));
+        // Every certificate a job attaches keeps its own file: one attach must not evict another's.
+        for i in 0..40 {
             let mut other = pg("verify-full", Some(false));
             other.root_certificate_pem = Some(format!("-----BEGIN CERTIFICATE-----{i}"));
-            pg_attach_uri(&other).unwrap();
+            let other = pg_attach_uri(&other, &job_dir).unwrap();
+            let path = urlencoding::decode(other.split("sslrootcert=").nth(1).unwrap()).unwrap();
+            assert!(std::path::Path::new(path.as_ref()).is_file(), "{path}");
         }
-        let kept = std::fs::read_dir(std::env::temp_dir().join("windmill-pg-roots"))
-            .unwrap()
-            .filter(|e| e.as_ref().unwrap().path().extension().is_some_and(|x| x == "pem"))
-            .count();
-        assert!(kept <= PG_ROOTS_KEPT, "{kept} root files kept");
-        assert!(roots.contains("-----BEGIN CERTIFICATE-----test"));
+        assert!(std::path::Path::new(root.as_ref()).is_file(), "the first file is still there");
         let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap();
-        let attach = &pg_secret_attach_statements(external, "dt").unwrap()[3];
+        let attach = &pg_secret_attach_statements(external, "dt", &job_dir).unwrap()[3];
         assert!(
             attach.starts_with(&format!("ATTACH 'sslmode=verify-full sslrootcert=''{}''", root)),
             "{attach}"
         );
         // A resource that never opted in keeps the historical downgrade.
-        assert!(pg_attach_uri(&pg("verify-full", None)).unwrap().ends_with("?sslmode=require"));
-        assert!(pg_attach_uri(&pg("require", Some(false))).unwrap().ends_with("?sslmode=require"));
+        assert!(pg_attach_uri(&pg("verify-full", None), &job_dir)
+            .unwrap()
+            .ends_with("?sslmode=require"));
+        assert!(pg_attach_uri(&pg("require", Some(false)), &job_dir)
+            .unwrap()
+            .ends_with("?sslmode=require"));
+        std::fs::remove_dir_all(&job_dir).unwrap();
     }
 
     #[test]
@@ -3089,7 +3086,7 @@ mod tests {
         let mut defer = test_fork_defer(vec![("orders", false)], vec![]);
         defer.ancestors[0].extra_args = Some("ENCRYPTED true".to_string());
         let mut hp = Arc::new(Mutex::new(vec![]));
-        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
+        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
         let attach = stmts
             .iter()
             .find(|s| s.starts_with("ATTACH IF NOT EXISTS"))
@@ -3113,7 +3110,7 @@ mod tests {
             vec![],
         );
         let mut hp = Arc::new(Mutex::new(vec![]));
-        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
+        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
         let joined = stmts.join("\n");
         assert!(
             joined.contains(
@@ -3145,7 +3142,7 @@ mod tests {
     fn test_fork_defer_statements_shape() {
         let defer = test_fork_defer(vec![("orders", false), ("dim", true)], vec![]);
         let mut hp = Arc::new(Mutex::new(vec![]));
-        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
+        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
         let joined = stmts.join("\n");
         // Ancestor attach: read-only, idempotent, never auto-migrating or auto-creating.
         assert!(joined.contains("ATTACH IF NOT EXISTS"), "{joined}");
@@ -3172,7 +3169,7 @@ mod tests {
         let defer = test_fork_defer(vec![("orders", false)], vec!["orders", "orders_current"]);
         let mut hp = Arc::new(Mutex::new(vec![]));
         let stmts =
-            fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp)
+            fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp, "/tmp")
                 .unwrap();
         let joined = stmts.join("\n");
         assert!(!joined.contains("CREATE VIEW"), "{joined}");
@@ -3189,14 +3186,14 @@ mod tests {
         // status can't be trusted) → no DROP VIEW, or the job would wedge on a type mismatch.
         let defer = test_fork_defer(vec![("orders", false)], vec![]);
         let stmts =
-            fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp)
+            fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp, "/tmp")
                 .unwrap();
         assert!(!stmts.join("\n").contains("DROP VIEW"), "{stmts:?}");
 
         // Target in a different lake → this lake's defer views are untouched.
         let defer = test_fork_defer(vec![("orders", false)], vec!["orders"]);
         let stmts =
-            fork_defer_statements("lake", "dl", &defer, Some("other/orders"), &mut hp).unwrap();
+            fork_defer_statements("lake", "dl", &defer, Some("other/orders"), &mut hp, "/tmp").unwrap();
         let joined = stmts.join("\n");
         assert!(
             joined.contains("CREATE VIEW IF NOT EXISTS dl.\"orders\""),
@@ -3209,7 +3206,7 @@ mod tests {
     fn test_fork_defer_statements_schema_qualified() {
         let defer = test_fork_defer(vec![("staging.raw", false)], vec![]);
         let mut hp = Arc::new(Mutex::new(vec![]));
-        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
+        let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
         let joined = stmts.join("\n");
         assert!(
             joined.contains("CREATE SCHEMA IF NOT EXISTS dl.\"staging\";"),
@@ -4035,7 +4032,7 @@ mod tests {
             "dbname": "mydb",
             "sslmode": "require"
         });
-        let result = format_attach_db_conn_str(db_resource, "postgres").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "postgres", "/tmp").unwrap();
         // Should be in URI format: postgres://user:password@host:port/dbname?sslmode=require
         assert!(result.starts_with("postgres://"));
         assert!(result.contains("admin:secret123@localhost:5432/mydb"));
@@ -4048,7 +4045,7 @@ mod tests {
             "host": "db.example.com",
             "dbname": "production"
         });
-        let result = format_attach_db_conn_str(db_resource, "postgres").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "postgres", "/tmp").unwrap();
         // Should be in URI format with defaults: postgres://postgres:@host:5432/dbname?sslmode=prefer
         assert!(result.starts_with("postgres://"));
         assert!(result.contains("@db.example.com:5432/production"));
@@ -4061,7 +4058,7 @@ mod tests {
             "host": "localhost",
             "dbname": "test"
         });
-        let result = format_attach_db_conn_str(db_resource, "postgresql").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "postgresql", "/tmp").unwrap();
         // Should be in URI format (postgresql is treated the same as postgres)
         assert!(result.starts_with("postgres://"));
         assert!(result.contains("@localhost:5432/test"));
@@ -4078,7 +4075,7 @@ mod tests {
             "dbname": "wm_datatables",
             "sslmode": "require"
         });
-        let stmts = pg_secret_attach_statements(db_resource, "dt").unwrap();
+        let stmts = pg_secret_attach_statements(db_resource, "dt", "/tmp").unwrap();
         assert_eq!(stmts[0], "INSTALL postgres;");
         assert_eq!(stmts[1], "LOAD postgres;");
         let secret_name = datatable_secret_name("dt");
@@ -4109,7 +4106,7 @@ mod tests {
             if let Some(s) = input {
                 db_resource["sslmode"] = json!(s);
             }
-            let stmts = pg_secret_attach_statements(db_resource, "dt").unwrap();
+            let stmts = pg_secret_attach_statements(db_resource, "dt", "/tmp").unwrap();
             assert!(
                 stmts[3].starts_with(&format!("ATTACH 'sslmode={expected}'")),
                 "sslmode {input:?} → {}",
@@ -4132,7 +4129,7 @@ mod tests {
         let db_resource = json!({
             "project_id": "my-gcp-project"
         });
-        let result = format_attach_db_conn_str(db_resource, "bigquery").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "bigquery", "/tmp").unwrap();
         assert_eq!(result, "project=my-gcp-project");
     }
 
@@ -4141,7 +4138,7 @@ mod tests {
         let db_resource = json!({
             "other_field": "value"
         });
-        let result = format_attach_db_conn_str(db_resource, "bigquery");
+        let result = format_attach_db_conn_str(db_resource, "bigquery", "/tmp");
         assert!(result.is_err());
         assert!(result.unwrap_err().to_string().contains("project_id"));
     }
@@ -4149,7 +4146,7 @@ mod tests {
     #[test]
     fn test_format_attach_db_conn_str_unsupported_type() {
         let db_resource = json!({});
-        let result = format_attach_db_conn_str(db_resource, "oracle");
+        let result = format_attach_db_conn_str(db_resource, "oracle", "/tmp");
         assert!(result.is_err());
         assert!(result
             .unwrap_err()
@@ -4163,7 +4160,7 @@ mod tests {
             "host": "localhost",
             "dbname": "test"
         });
-        let result = format_attach_db_conn_str(db_resource, "POSTGRES").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "POSTGRES", "/tmp").unwrap();
         // Should be in URI format
         assert!(result.starts_with("postgres://"));
         assert!(result.contains("@localhost:5432/test"));
@@ -4180,7 +4177,7 @@ mod tests {
             "database": "app_db",
             "ssl": true
         });
-        let result = format_attach_db_conn_str(db_resource, "mysql").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "mysql", "/tmp").unwrap();
         assert!(result.contains("database=app_db"));
         assert!(result.contains("host=mysql.example.com"));
         assert!(result.contains("ssl_mode=required"));
@@ -4197,7 +4194,7 @@ mod tests {
             "database": "test",
             "ssl": false
         });
-        let result = format_attach_db_conn_str(db_resource, "mysql").unwrap();
+        let result = format_attach_db_conn_str(db_resource, "mysql", "/tmp").unwrap();
         assert!(result.contains("ssl_mode=disabled"));
     }