Merge commit '8c554ef1337aa6cd99ab869443dfd30becc65a36' into HEAD

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

backend/Cargo.lock

@@ -15653,6 +15653,7 @@ dependencies = [
  "pin-project-lite",
  "pkcs1",
  "postgres-native-tls 0.5.3",
+ "postgres-protocol",
  "prometheus",
  "quick_cache",
  "rand 0.9.0",

backend/Cargo.toml

@@ -624,6 +624,7 @@ wasm-bindgen-test = "^0"
 convert_case = "0.6.0"
 getrandom = "0.2"
 tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]}
+postgres-protocol = "0.6"
 rust-postgres = { package = "tokio-postgres", git = "https://github.com/MaterializeInc/rust-postgres", rev = "78c1222577bb091d69bc22b1bc7ad01c14675abe"}
 rust-postgres-native-tls = { package = "postgres-native-tls", git = "https://github.com/MaterializeInc/rust-postgres", features = ["runtime"], rev = "78c1222577bb091d69bc22b1bc7ad01c14675abe" }
 bit-vec = "=0.6.3"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-c2e43f5b5ff753d339b70e232fd17b6ffecb054d
+a97b5a5982d67e977dde6222380903f019da39c2

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

@@ -60,7 +60,7 @@ use windmill_common::{
     global_settings::{
         AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, AUTOMATE_USERNAME_CREATION_SETTING,
         CRITICAL_ALERT_MUTE_UI_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
-        DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS,
+        DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, EXTERNAL_INSTANCE_PG_SETTING,
         GITHUB_APP_WEBHOOK_BASE_URL_SETTING, HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING,
         HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING,
         INSTANCE_BANNER_SETTING, MAX_RETENTION_OVERRIDE_WORKSPACES,
@@ -168,6 +168,22 @@ pub fn global_service() -> Router {
             "/refresh_custom_instance_user_pwd",
             post(refresh_custom_instance_user_pwd),
         )
+        .route(
+            "/external_instance_pg/status",
+            get(get_external_instance_pg_status),
+        )
+        .route(
+            "/external_instance_pg/setup",
+            post(setup_external_instance_pg),
+        )
+        .route(
+            "/external_instance_pg/databases",
+            get(list_external_instance_pg_databases),
+        )
+        .route(
+            "/external_instance_pg/databases/{name}",
+            post(create_external_instance_pg_database).delete(drop_external_instance_pg_database),
+        )
         .route(
             "/setup_custom_instance_pg_database/{name}",
             post(setup_custom_instance_pg_database),
@@ -877,6 +893,14 @@ pub async fn set_global_setting_internal(
         )));
     }
 
+    if key == EXTERNAL_INSTANCE_PG_SETTING {
+        return windmill_common::external_instance_pg::write_external_instance_pg_setting(
+            db,
+            Some(&value),
+        )
+        .await;
+    }
+
     run_setting_pre_write_hook(db, &key, &value).await?;
 
     match value {
@@ -1258,7 +1282,7 @@ async fn set_instance_config(
     let desired_map = desired.global_settings.to_settings_map();
     if !desired_map.is_empty() {
         let current_map = current.global_settings.to_settings_map();
-        let settings_diff =
+        let mut settings_diff =
             instance_config::diff_global_settings(&current_map, &desired_map, ApplyMode::Merge);
         let ai_config_changed = settings_diff
             .upserts
@@ -1287,8 +1311,15 @@ async fn set_instance_config(
         }
 
         for (key, value) in &settings_diff.upserts {
-            run_setting_pre_write_hook(&db, key, value).await?;
+            if key != EXTERNAL_INSTANCE_PG_SETTING {
+                run_setting_pre_write_hook(&db, key, value).await?;
+            }
         }
+        windmill_common::external_instance_pg::write_external_instance_pg_from_diff(
+            &db,
+            &mut settings_diff,
+        )
+        .await?;
 
         instance_config::apply_settings_diff(&db, &settings_diff)
             .await
@@ -1745,6 +1776,135 @@ async fn refresh_custom_instance_user_pwd(
     Ok(Json(()))
 }
 
+async fn get_external_instance_pg_status(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgStatus> {
+    require_super_admin(&db, &authed).await?;
+    Ok(Json(
+        windmill_common::external_instance_pg::external_instance_pg_status(&db).await?,
+    ))
+}
+
+#[derive(Deserialize)]
+struct SetupExternalInstancePgBody {
+    #[serde(default)]
+    rotate_passwords: bool,
+}
+
+async fn setup_external_instance_pg(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Json(body): Json<SetupExternalInstancePgBody>,
+) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgSetupReport> {
+    require_super_admin(&db, &authed).await?;
+    let report = windmill_common::external_instance_pg::setup_external_instance_pg_unchecked(
+        &db,
+        body.rotate_passwords,
+    )
+    .await?;
+    let rotated = body.rotate_passwords.to_string();
+    let success = report.success.to_string();
+    windmill_audit::audit_oss::audit_log(
+        &db,
+        &authed,
+        "settings.setup_external_instance_pg",
+        windmill_audit::ActionKind::Update,
+        "global",
+        Some(&authed.email),
+        Some(
+            [
+                ("rotate_passwords", rotated.as_str()),
+                ("success", success.as_str()),
+            ]
+            .into(),
+        ),
+    )
+    .await?;
+    Ok(Json(report))
+}
+
+#[derive(Serialize)]
+struct ExternalInstancePgDatabase {
+    #[serde(flatten)]
+    status: windmill_common::instance_config::CustomInstanceDb,
+    used_by_workspaces: Vec<String>,
+}
+
+async fn list_external_instance_pg_databases(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<std::collections::BTreeMap<String, ExternalInstancePgDatabase>> {
+    require_super_admin(&db, &authed).await?;
+    let databases = windmill_common::external_instance_pg::external_instance_databases(&db).await?;
+    let mut usages =
+        windmill_common::external_instance_pg::external_instance_database_usages(&db).await?;
+    Ok(Json(
+        databases
+            .into_iter()
+            .map(|(name, status)| {
+                let used_by_workspaces = usages.remove(&name).unwrap_or_default();
+                (
+                    name,
+                    ExternalInstancePgDatabase {
+                        status,
+                        used_by_workspaces: used_by_workspaces.into_iter().collect(),
+                    },
+                )
+            })
+            .collect(),
+    ))
+}
+
+async fn create_external_instance_pg_database(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(dbname): Path<String>,
+    Json(body): Json<SetupCustomInstanceDbBody>,
+) -> JsonResult<()> {
+    require_super_admin(&db, &authed).await?;
+    let tag = body.tag.as_deref().unwrap_or("datatable");
+    windmill_common::external_instance_pg::create_external_instance_database_unchecked(
+        &db, &dbname, tag, None,
+    )
+    .await?;
+    windmill_audit::audit_oss::audit_log(
+        &db,
+        &authed,
+        "settings.create_external_instance_pg_database",
+        windmill_audit::ActionKind::Create,
+        "global",
+        Some(&authed.email),
+        Some([("dbname", dbname.as_str()), ("tag", tag)].into()),
+    )
+    .await?;
+    Ok(Json(()))
+}
+
+async fn drop_external_instance_pg_database(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(dbname): Path<String>,
+) -> JsonResult<()> {
+    require_super_admin(&db, &authed).await?;
+    // A data table naming a dropped database fails on every job, far from the drop that caused it.
+    windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
+        &db, &dbname, None,
+    )
+    .await?;
+    windmill_audit::audit_oss::audit_log(
+        &db,
+        &authed,
+        "settings.drop_external_instance_pg_database",
+        windmill_audit::ActionKind::Delete,
+        "global",
+        Some(&authed.email),
+        Some([("dbname", dbname.as_str())].into()),
+    )
+    .await?;
+    Ok(Json(()))
+}
+
 #[derive(Deserialize)]
 struct SetupCustomInstanceDbBody {
     tag: Option<String>,

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

@@ -11,7 +11,7 @@
 //! to keep that file focused on core workspace configuration.
 
 use crate::workspaces::{
-    is_instance_datatable, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
+    managed_datatable_kind, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
     PgDumpOptions,
 };
 
@@ -1556,7 +1556,9 @@ async fn generate_initial_datatable_migration(
     // without what a replay elsewhere cannot run: the replaying user owns none of this
     // database's objects, and the grants Windmill plants in an instance database (`ALTER
     // DEFAULT PRIVILEGES FOR ROLE ...`) fail even replaying onto the same server.
-    let no_acl = is_instance_datatable(&db, &w_id, &datatable_name).await?;
+    let no_acl = managed_datatable_kind(&db, &w_id, &datatable_name)
+        .await?
+        .is_some();
     let dump_file = pg_dump_database(
         &pg_db,
         PgDumpOptions {

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

@@ -3078,23 +3078,32 @@ pub(crate) async fn resolve_pg_source_checked(
         .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))
 }
 
-/// Whether the data table `name` is backed by the Windmill instance's own PostgreSQL
-/// rather than a user resource.
-pub(crate) async fn is_instance_datatable(db: &DB, w_id: &str, name: &str) -> Result<bool> {
+/// The kind of the database backing the data table `name` when Windmill manages it (on its own
+/// cluster or the external one), `None` when it is a user resource.
+pub(crate) async fn managed_datatable_kind(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+) -> Result<Option<DataTableCatalogResourceType>> {
     // Resolved rather than read: a pointer entry owns no database of its own, so only the entry it
-    // lands on can answer. A name that resolves to nothing keeps the historical `false`.
+    // lands on can answer. A name that resolves to nothing keeps the historical `None`.
     Ok(resolve_governing_datatable(db, w_id, name)
         .await
         .ok()
         .and_then(|g| g.datatable.database)
-        .is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance))
+        .map(|d| d.resource_type)
+        .filter(|kind| kind.is_windmill_managed()))
 }
 
 /// Same, for the `datatable://<name>` / `$res:<path>` form the import endpoints take.
-async fn is_instance_datatable_source(db: &DB, w_id: &str, source: &str) -> Result<bool> {
+async fn managed_datatable_source_kind(
+    db: &DB,
+    w_id: &str,
+    source: &str,
+) -> Result<Option<DataTableCatalogResourceType>> {
     match source.strip_prefix("datatable://") {
-        Some(name) => is_instance_datatable(db, w_id, name).await,
-        None => Ok(false),
+        Some(name) => managed_datatable_kind(db, w_id, name).await,
+        None => Ok(None),
     }
 }
 
@@ -3193,10 +3202,7 @@ pub(crate) async fn pg_dump_database(
     if let Some(ref password) = pg_db.password {
         cmd.env("PGPASSWORD", password);
     }
-
-    if let Some(ref sslmode) = pg_db.sslmode {
-        cmd.env("PGSSLMODE", sslmode);
-    }
+    let _root_cert = apply_pg_tls_env(&mut cmd, pg_db)?;
 
     let output = cmd
         .output()
@@ -3314,7 +3320,7 @@ async fn comment_out_unsupported_settings(
 
 /// A psql invocation against `pg_db`, carrying the connection settings the CLI reads
 /// from the environment.
-fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
+fn psql_command(pg_db: &PgDatabase) -> Result<(tokio::process::Command, Option<DumpFile>)> {
     let mut cmd = tokio::process::Command::new("psql");
     cmd.arg("--host")
         .arg(&pg_db.host)
@@ -3331,10 +3337,88 @@ fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
     if let Some(ref password) = pg_db.password {
         cmd.env("PGPASSWORD", password);
     }
+    let root_cert = apply_pg_tls_env(&mut cmd, pg_db)?;
+    Ok((cmd, root_cert))
+}
+
+/// Give libpq the TLS settings `PgDatabase::connect` applies. The returned file holds the root
+/// certificate `PGSSLROOTCERT` names, so it must outlive the command.
+fn apply_pg_tls_env(
+    cmd: &mut tokio::process::Command,
+    pg_db: &PgDatabase,
+) -> Result<Option<DumpFile>> {
     if let Some(ref sslmode) = pg_db.sslmode {
         cmd.env("PGSSLMODE", sslmode);
     }
-    cmd
+    if let Some(pem) = pg_db
+        .root_certificate_pem
+        .as_deref()
+        .filter(|p| !p.is_empty())
+    {
+        let file = DumpFile::new()?;
+        std::fs::write(&file.path, pem)
+            .map_err(|e| Error::internal_err(format!("Failed to write root certificate: {e}")))?;
+        cmd.env("PGSSLROOTCERT", &file.path);
+        return Ok(Some(file));
+    }
+    // Only a connection that asked to be verified against the system trust store. Without a file,
+    // libpq's own default would look for `~/.postgresql/root.crt` and refuse a verify-* mode. libpq
+    // takes the special `system` value with verify-full only, so verify-ca needs the bundle itself.
+    if pg_db.accept_invalid_certs == Some(false) {
+        match pg_db.sslmode.as_deref() {
+            Some("verify-full") => {
+                cmd.env("PGSSLROOTCERT", "system");
+            }
+            Some("verify-ca") => {
+                if let Some(bundle) = windmill_common::system_ca_bundle() {
+                    cmd.env("PGSSLROOTCERT", bundle);
+                }
+            }
+            _ => {}
+        }
+    }
+    Ok(None)
+}
+
+
+#[cfg(test)]
+mod pg_tls_env_tests {
+    use super::apply_pg_tls_env;
+    use windmill_common::PgDatabase;
+
+    fn root_cert_env(sslmode: &str) -> Option<std::ffi::OsString> {
+        let pg_db = PgDatabase {
+            host: "db".to_string(),
+            user: None,
+            password: None,
+            port: None,
+            sslmode: Some(sslmode.to_string()),
+            dbname: "d".to_string(),
+            root_certificate_pem: None,
+            accept_invalid_certs: Some(false),
+            use_iam_auth: None,
+            region: None,
+        };
+        let mut cmd = tokio::process::Command::new("psql");
+        apply_pg_tls_env(&mut cmd, &pg_db).unwrap();
+        cmd.as_std()
+            .get_envs()
+            .find(|(k, _)| *k == "PGSSLROOTCERT")
+            .and_then(|(_, v)| v.map(|v| v.to_os_string()))
+    }
+
+    #[test]
+    fn system_roots_only_through_verify_full() {
+        assert_eq!(
+            root_cert_env("verify-full").as_deref(),
+            Some("system".as_ref())
+        );
+        // libpq refuses `sslrootcert=system` with verify-ca, which would fail every dump and restore.
+        assert_ne!(
+            root_cert_env("verify-ca").as_deref(),
+            Some("system".as_ref())
+        );
+    }
 }
 
 /// GUC names the server backing `pg_db` knows about.
@@ -3344,7 +3428,8 @@ fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
 /// and an unset mode, where `PgDatabase::connect` would hand a TLS-only server a
 /// plaintext socket and fail before the import ever starts.
 async fn server_setting_names(pg_db: &PgDatabase) -> Result<HashSet<String>> {
-    let output = psql_command(pg_db)
+    let (mut cmd, _root_cert) = psql_command(pg_db)?;
+    let output = cmd
         .arg("--tuples-only")
         .arg("--no-align")
         .arg("--command")
@@ -3378,7 +3463,8 @@ async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) -> Result<
     let supported_settings = server_setting_names(target_db).await?;
     comment_out_unsupported_settings(dump_file, &supported_settings).await?;
 
-    let output = psql_command(target_db)
+    let (mut cmd, _root_cert) = psql_command(target_db)?;
+    let output = cmd
         .arg("--set")
         .arg("ON_ERROR_STOP=1")
         .arg("--single-transaction")
@@ -3436,7 +3522,16 @@ async fn create_pg_database(
         }
     }
 
-    if is_instance_datatable_source(&db, &w_id, &req.source).await? {
+    let source_kind = managed_datatable_source_kind(&db, &w_id, &req.source).await?;
+    if source_kind == Some(DataTableCatalogResourceType::ExternalInstance) {
+        windmill_common::external_instance_pg::create_external_instance_database_unchecked(
+            &db,
+            &req.target_dbname,
+            "datatable",
+            Some(&w_id),
+        )
+        .await?;
+    } else if source_kind == Some(DataTableCatalogResourceType::Instance) {
         windmill_common::create_custom_instance_database(
             &db,
             &req.target_dbname,
@@ -3544,12 +3639,12 @@ async fn ensure_datatable_is_clonable(
     }
     // The copy has to name a database of its own. A resource-backed entry reached through a
     // pointer names one this workspace does not own, so there is nothing here to repoint.
-    let is_instance = governing
+    let is_managed = governing
         .datatable
         .database
         .as_ref()
-        .is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance);
-    if governing.workspace_id != w_id && !is_instance {
+        .is_some_and(|d| d.resource_type.is_windmill_managed());
+    if governing.workspace_id != w_id && !is_managed {
         return Err(Error::BadRequest(format!(
             "Data table '{name}' points at a resource-backed data table in another workspace \
              and cannot be copied; fork it from the workspace that owns it."
@@ -3597,9 +3692,14 @@ async fn import_pg_database(
                         .to_string(),
                 ));
             }
-            if is_instance_datatable_source(&db, &w_id, &req.target).await? {
-                windmill_common::ensure_fork_database_available_to(&db, override_dbname, &w_id)
-                    .await?;
+            if let Some(kind) = managed_datatable_source_kind(&db, &w_id, &req.target).await? {
+                windmill_common::ensure_fork_database_available_to(
+                    &db,
+                    kind,
+                    override_dbname,
+                    &w_id,
+                )
+                .await?;
             }
         }
         target_pg.dbname = override_dbname.clone();
@@ -3610,8 +3710,12 @@ async fn import_pg_database(
     // what it creates it owns. Grants do, except around an instance data table — Windmill
     // plants `custom_instance_user` grants in one, which nothing else can replay. Elsewhere
     // the ACLs are user intent (`REVOKE ... FROM PUBLIC`) and dropping them widens access.
-    let no_acl = is_instance_datatable_source(&db, &w_id, &req.target).await?
-        || is_instance_datatable_source(&db, &w_id, &req.source).await?;
+    let no_acl = managed_datatable_source_kind(&db, &w_id, &req.target)
+        .await?
+        .is_some()
+        || managed_datatable_source_kind(&db, &w_id, &req.source)
+            .await?
+            .is_some();
 
     let dump_file = pg_dump_database(
         &source_pg,
@@ -3724,22 +3828,37 @@ async fn edit_ducklake_config(
     let old_ducklakes: HashMap<String, Ducklake> =
         serde_json::from_value(old_ducklakes).unwrap_or_default();
 
-    // Check that non-superadmins are not abusing Instance databases
-    if !is_superadmin {
-        for (name, dl) in new_config.settings.ducklakes.iter() {
-            if dl.catalog.resource_type == DucklakeCatalogResourceType::Instance {
-                let old_dl = old_ducklakes.get(name);
-                if old_dl.is_none()
-                    || old_dl.unwrap().catalog.resource_type
-                        != DucklakeCatalogResourceType::Instance
-                    || old_dl.unwrap().catalog.resource_path != dl.catalog.resource_path
-                {
-                    return Err(Error::BadRequest(
-                        "Only superadmins can create or modify ducklakes with Instance databases"
-                            .to_string(),
-                    ));
-                }
-            }
+    // Check that non-superadmins are not abusing Instance databases. An unchanged catalog is left
+    // alone either way, so a downgraded instance can still save lakes that already name an
+    // external instance database.
+    for (name, dl) in new_config.settings.ducklakes.iter() {
+        let kind = &dl.catalog.resource_type;
+        if !matches!(
+            kind,
+            DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance
+        ) {
+            continue;
+        }
+        let unchanged = old_ducklakes.get(name).is_some_and(|old| {
+            &old.catalog.resource_type == kind
+                && old.catalog.resource_path == dl.catalog.resource_path
+        });
+        if unchanged {
+            continue;
+        }
+        if *kind == DucklakeCatalogResourceType::ExternalInstance {
+            windmill_common::external_instance_pg::ensure_external_instance_available()?;
+            windmill_common::external_instance_pg::ensure_external_instance_database_registered(
+                &mut tx,
+                &dl.catalog.resource_path,
+            )
+            .await?;
+        }
+        if !is_superadmin {
+            return Err(Error::BadRequest(
+                "Only superadmins can create or modify ducklakes with Instance databases"
+                    .to_string(),
+            ));
         }
     }
 
@@ -3911,6 +4030,7 @@ async fn edit_datatable_config(
                 // so these line up with the `datatable_configured` adoption counts.
                 created_substrates.push(match dt.database.as_ref().map(|d| d.resource_type) {
                     Some(DataTableCatalogResourceType::Instance) => "instance",
+                    Some(DataTableCatalogResourceType::ExternalInstance) => "external_instance",
                     Some(DataTableCatalogResourceType::Postgresql) => "postgresql",
                     None => "reference",
                 });
@@ -3966,26 +4086,39 @@ async fn edit_datatable_config(
     // Check that non-superadmins are not abusing Instance databases, which reach a database this
     // workspace does not own. Pointing an entry at another workspace's data table is not checked
     // here because it cannot be requested at all: `reference` is overwritten from the stored entry
-    // above, for every caller.
-    if !is_superadmin {
-        for (name, dt) in new_config.settings.datatables.iter() {
-            let old_dt = old_datatables.get(name);
-            if dt
-                .database
-                .as_ref()
-                .is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
-            {
-                let unchanged = old_dt.and_then(|o| o.database.as_ref()).is_some_and(|o| {
-                    o.resource_type == DataTableCatalogResourceType::Instance
-                        && Some(&o.resource_path) == dt.database.as_ref().map(|d| &d.resource_path)
-                });
-                if !unchanged {
-                    return Err(Error::BadRequest(
-                        "Only superadmins can create or modify data tables with Instance databases"
-                            .to_string(),
-                    ));
-                }
-            }
+    // above, for every caller. An unchanged entry is left alone either way, so a downgraded
+    // instance can still save settings that already name an external instance database.
+    for (name, dt) in new_config.settings.datatables.iter() {
+        let Some(database) = dt
+            .database
+            .as_ref()
+            .filter(|d| d.resource_type.is_windmill_managed())
+        else {
+            continue;
+        };
+        let unchanged = old_datatables
+            .get(name)
+            .and_then(|o| o.database.as_ref())
+            .is_some_and(|o| {
+                o.resource_type == database.resource_type
+                    && o.resource_path == database.resource_path
+            });
+        if unchanged {
+            continue;
+        }
+        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(
+                &mut tx,
+                &database.resource_path,
+            )
+            .await?;
+        }
+        if !is_superadmin {
+            return Err(Error::BadRequest(
+                "Only superadmins can create or modify data tables with Instance databases"
+                    .to_string(),
+            ));
         }
     }
 
@@ -8071,13 +8204,14 @@ async fn point_kept_datatables_at_parent(
         if dt.reference.is_some() {
             continue;
         }
-        // Only instance databases. A resource-backed data table names a resource, and the settings
-        // clone gave the fork its own copy of that resource in its own workspace — pointing at the
-        // parent's entry would silently move the fork onto the parent's resource instead.
+        // Only instance databases, on either cluster. A resource-backed data table names a
+        // resource, and the settings clone gave the fork its own copy of that resource in its own
+        // workspace — pointing at the parent's entry would silently move the fork onto the
+        // parent's resource instead.
         if dt
             .database
             .as_ref()
-            .is_none_or(|d| d.resource_type != DataTableCatalogResourceType::Instance)
+            .is_none_or(|d| !d.resource_type.is_windmill_managed())
         {
             continue;
         }
@@ -8207,17 +8341,30 @@ async fn apply_forked_datatable(
         })?,
     };
 
-    if database.resource_type == DataTableCatalogResourceType::Instance
+    if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
+        windmill_common::external_instance_pg::ensure_external_instance_database_registered(
+            tx,
+            &fdt.new_dbname,
+        )
+        .await?;
+    }
+    if database.resource_type.is_windmill_managed()
         && !windmill_api_auth::is_super_admin_authed(db, authed).await?
     {
-        windmill_common::ensure_fork_database_available_to(db, &fdt.new_dbname, parent_w_id)
-            .await?;
+        windmill_common::ensure_fork_database_available_to(
+            db,
+            database.resource_type,
+            &fdt.new_dbname,
+            parent_w_id,
+        )
+        .await?;
     }
-    if database.resource_type == DataTableCatalogResourceType::Instance {
+    if database.resource_type.is_windmill_managed() {
         // The whole `database` object, not just its `resource_path`: a pointer entry has none to
-        // patch. `reference` goes with it — exactly one of the two may be set.
+        // patch. `reference` goes with it — exactly one of the two may be set. The copy was created
+        // on the same cluster as its source, so it keeps the source's kind.
         let new_database = serde_json::json!({
-            "resource_type": "instance",
+            "resource_type": database.resource_type,
             "resource_path": &fdt.new_dbname,
         });
         sqlx::query!(
@@ -8608,6 +8755,8 @@ async fn create_workspace_fork(
     let mut tx: Transaction<'_, Postgres> = db.begin().await?;
     // Before the settings clone reads the parent's data tables: a pointer this fork ends up with
     // must not be written after cleanup of the parent decided that nothing points at its copies.
+    // Also before the external cluster's lifecycle lock, which finalizing an external copy takes:
+    // fork cleanup takes the two in this order.
     windmill_common::workspaces::lock_fork_datatables(&mut tx, &parent_workspace_id).await?;
 
     if nw.is_dev_workspace {

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

@@ -1424,9 +1424,7 @@ pub async fn drop_forked_datatable_databases(
             _ => continue,
         };
 
-        if database.resource_type
-            == windmill_common::workspaces::DataTableCatalogResourceType::Instance
-        {
+        if database.resource_type.is_windmill_managed() {
             let db_to_drop = &database.resource_path;
             if !db_to_drop.starts_with("wm_fork_") {
                 errors.push(format!(
@@ -1441,20 +1439,31 @@ pub async fn drop_forked_datatable_databases(
             let dropped = async {
                 let mut tx = db.begin().await?;
                 windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
-                let uses = windmill_common::workspaces::managed_database_uses(
-                    &mut tx,
-                    windmill_common::workspaces::DataTableCatalogResourceType::Instance,
-                    db_to_drop,
-                    Some((&w_id, dt_name)),
-                )
-                .await?;
-                if !uses.is_empty() {
-                    return Err(Error::BadRequest(format!(
-                        "it is still used by {}",
-                        uses.join(", ")
-                    )));
+                if database.resource_type
+                    == windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance
+                {
+                    windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
+                        &db,
+                        db_to_drop,
+                        Some((&w_id, dt_name)),
+                    )
+                    .await?;
+                } else {
+                    let uses = windmill_common::workspaces::managed_database_uses(
+                        &mut tx,
+                        windmill_common::workspaces::DataTableCatalogResourceType::Instance,
+                        db_to_drop,
+                        Some((&w_id, dt_name)),
+                    )
+                    .await?;
+                    if !uses.is_empty() {
+                        return Err(Error::BadRequest(format!(
+                            "it is still used by {}",
+                            uses.join(", ")
+                        )));
+                    }
+                    windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
                 }
-                windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
                 tx.commit().await?;
                 Ok::<_, Error>(())
             }
@@ -1826,7 +1835,17 @@ async fn resolve_fork_catalog_pg(
             "ducklake://{ducklake_name}: malformed registry catalog identity `{catalog}`"
         ))
     })?;
-    let catalog_resource = if resource_type == "instance" {
+    let catalog_resource = if resource_type == "external_instance" {
+        serde_json::to_value(
+            windmill_common::external_instance_pg::external_instance_connection_unchecked(
+                db,
+                resource_path,
+                false,
+            )
+            .await?,
+        )
+        .map_err(|e| Error::internal_err(format!("serializing pg creds: {e}")))?
+    } else if resource_type == "instance" {
         let mut pg_creds = windmill_common::PgDatabase::parse_uri(
             &windmill_common::get_database_url().await?.as_str().await,
         )?;

backend/windmill-api/openapi.yaml

@@ -1572,6 +1572,104 @@ paths:
               schema:
                 type: object
 
+  /settings/external_instance_pg/status:
+    get:
+      summary: Returns whether the external instance cluster is configured and how its last setup went
+      operationId: getExternalInstancePgStatus
+      tags:
+        - setting
+      responses:
+        "200":
+          description: external instance cluster status
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ExternalInstancePgStatus"
+
+  /settings/external_instance_pg/setup:
+    post:
+      summary: Sets up the external instance cluster with its saved admin login, optionally rotating the passwords Windmill manages on it (enterprise edition only)
+      operationId: setupExternalInstancePg
+      tags:
+        - setting
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                rotate_passwords:
+                  type: boolean
+      responses:
+        "200":
+          description: the setup report, also stored as the last setup
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ExternalInstancePgSetupReport"
+
+  /settings/external_instance_pg/databases:
+    get:
+      summary: Lists the databases Windmill created on the external instance cluster, with the workspaces whose data tables, Ducklake catalogs or pending fork cleanups use each
+      operationId: listExternalInstancePgDatabases
+      tags:
+        - setting
+      responses:
+        "200":
+          description: databases by name
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties:
+                  $ref: "#/components/schemas/CustomInstanceDb"
+
+  /settings/external_instance_pg/databases/{name}:
+    post:
+      summary: Creates a database on the external instance cluster (enterprise edition only)
+      operationId: createExternalInstancePgDatabase
+      tags:
+        - setting
+      parameters:
+        - name: name
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                tag:
+                  $ref: "#/components/schemas/CustomInstanceDbTag"
+      responses:
+        "200":
+          description: database created
+          content:
+            application/json:
+              schema: {}
+    delete:
+      summary: Drops a database Windmill created on the external instance cluster, refused while a data table, Ducklake catalog or pending fork cleanup uses it
+      operationId: dropExternalInstancePgDatabase
+      tags:
+        - setting
+      parameters:
+        - name: name
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: database dropped
+          content:
+            application/json:
+              schema: {}
+
   /settings/list_custom_instance_pg_databases:
     post:
       summary: Returns the set-up statuses of custom instance pg databases
@@ -5230,7 +5328,7 @@ paths:
                       type: string
                     resource_type:
                       type: string
-                      enum: [postgres, instance]
+                      enum: [postgres, instance, external_instance]
                     resource_path:
                       type: string
                     governing_workspace_id:
@@ -33621,6 +33719,44 @@ components:
         - ducklake
         - datatable
 
+    ExternalInstancePgSetupStep:
+      type: object
+      required: [name, status, message]
+      properties:
+        name:
+          type: string
+        status:
+          type: string
+          enum: [ok, warning, error]
+        message:
+          type: string
+
+    ExternalInstancePgSetupReport:
+      type: object
+      required: [success, finished_at, steps]
+      properties:
+        success:
+          type: boolean
+          description: no step failed; warnings leave it true
+        finished_at:
+          type: string
+          format: date-time
+        steps:
+          type: array
+          items:
+            $ref: "#/components/schemas/ExternalInstancePgSetupStep"
+
+    ExternalInstancePgStatus:
+      type: object
+      required: [configured, database_count]
+      properties:
+        configured:
+          type: boolean
+        database_count:
+          type: integer
+        last_setup:
+          $ref: "#/components/schemas/ExternalInstancePgSetupReport"
+
     InstanceDatatableRole:
       type: object
       required: [id, name, enabled]
@@ -35930,6 +36066,7 @@ components:
                       - postgresql
                       - mysql
                       - instance
+                      - external_instance
                   resource_path:
                     type: string
                 required:
@@ -35993,6 +36130,7 @@ components:
                     enum:
                       - postgresql
                       - instance
+                      - external_instance
                   resource_path:
                     type: string
                 required:

backend/windmill-common/Cargo.toml

@@ -75,6 +75,7 @@ bitflags.workspace = true
 once_cell.workspace = true
 phf.workspace = true
 tokio-postgres.workspace = true
+postgres-protocol.workspace = true
 postgres-native-tls.workspace = true
 native-tls.workspace = true
 

backend/windmill-common/src/external_instance_pg.rs

@@ -0,0 +1,423 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! The external Postgres cluster behind `external_instance` data tables and Ducklake catalogs.
+//!
+//! Windmill administers that cluster itself, logged in as the user in
+//! [`EXTERNAL_INSTANCE_PG_SETTING`]. It creates `custom_instance_user` and
+//! `custom_instance_replication_user` there, with passwords it generates and keeps in
+//! [`EXTERNAL_INSTANCE_PG_STATE_SETTING`]. They share their names with the roles on Windmill's own
+//! cluster, but they are different roles with different passwords.
+//!
+//! The cluster may hold data Windmill did not create. Two Windmill instances sharing one is not
+//! supported: each would keep resetting the passwords the other depends on.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    error::{Error, Result},
+    global_settings::{EXTERNAL_INSTANCE_PG_SETTING, EXTERNAL_INSTANCE_PG_STATE_SETTING},
+    instance_config::{CustomInstanceDb, ExternalInstancePg},
+    DB,
+};
+
+/// What Windmill keeps about the external cluster. Server-managed and hidden: never part of the
+/// instance config, never readable by an agent worker. No `Debug`: it carries live passwords.
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub struct ExternalInstancePgState {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub user_pwd: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub replication_pwd: Option<String>,
+    /// The databases Windmill created on the cluster. It only ever drops one of these.
+    #[serde(default)]
+    pub databases: BTreeMap<String, CustomInstanceDb>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_setup: Option<ExternalInstancePgSetupReport>,
+    /// The cluster ([`external_instance_pg_address`]) the last successful setup converged. Databases
+    /// are only created on a cluster setup succeeded on: the passwords above exist as soon as setup
+    /// first runs, whether or not the cluster accepted them.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub set_up_for: Option<String>,
+}
+
+/// What identifies the cluster a configuration points at. Other fields (admin login, sslmode) can
+/// change without it becoming another cluster.
+pub fn external_instance_pg_address(config: &ExternalInstancePg) -> String {
+    format!(
+        "{}:{}",
+        config.host.trim().to_lowercase(),
+        config.port.unwrap_or(5432)
+    )
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct ExternalInstancePgSetupReport {
+    /// No step failed. Warnings leave it true.
+    pub success: bool,
+    pub finished_at: chrono::DateTime<chrono::Utc>,
+    pub steps: Vec<ExternalInstancePgSetupStep>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct ExternalInstancePgSetupStep {
+    pub name: String,
+    pub status: SetupStepStatus,
+    pub message: String,
+}
+
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+pub enum SetupStepStatus {
+    Ok,
+    Warning,
+    Error,
+}
+
+/// The status the settings page shows without running anything.
+#[derive(Serialize, Debug)]
+pub struct ExternalInstancePgStatus {
+    pub configured: bool,
+    pub database_count: usize,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_setup: Option<ExternalInstancePgSetupReport>,
+}
+
+/// Authorization: returns the cluster's admin password and checks nothing. Callers MUST be
+/// superadmin or an internal server path.
+pub(crate) async fn read_external_instance_pg_config<'c>(
+    executor: impl sqlx::PgExecutor<'c>,
+) -> Result<Option<ExternalInstancePg>> {
+    let value = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        EXTERNAL_INSTANCE_PG_SETTING
+    )
+    .fetch_optional(executor)
+    .await?;
+    value
+        .map(|v| {
+            serde_json::from_value(v).map_err(|e| {
+                Error::internal_err(format!("reading {EXTERNAL_INSTANCE_PG_SETTING}: {e}"))
+            })
+        })
+        .transpose()
+}
+
+/// Authorization: returns the passwords Windmill generated on the cluster and checks nothing.
+/// Callers MUST be superadmin or an internal server path.
+pub(crate) async fn read_external_instance_pg_state<'c>(
+    executor: impl sqlx::PgExecutor<'c>,
+) -> Result<ExternalInstancePgState> {
+    let value = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        EXTERNAL_INSTANCE_PG_STATE_SETTING
+    )
+    .fetch_optional(executor)
+    .await?;
+    match value {
+        None => Ok(ExternalInstancePgState::default()),
+        Some(v) => serde_json::from_value(v).map_err(|e| {
+            Error::internal_err(format!("reading {EXTERNAL_INSTANCE_PG_STATE_SETTING}: {e}"))
+        }),
+    }
+}
+
+pub async fn external_instance_pg_status(db: &DB) -> Result<ExternalInstancePgStatus> {
+    let configured = read_external_instance_pg_config(db).await?.is_some();
+    let state = read_external_instance_pg_state(db).await?;
+    Ok(ExternalInstancePgStatus {
+        configured,
+        database_count: state.databases.len(),
+        last_setup: state.last_setup,
+    })
+}
+
+/// The databases Windmill created on the external cluster, without the passwords kept beside them.
+///
+/// Authorization: names every database across all workspaces, and the workspace each fork copy is
+/// reserved for, and checks nothing. Callers MUST be superadmin or an internal authorization or
+/// lifecycle path that does not return the names to a workspace caller.
+pub async fn external_instance_databases(db: &DB) -> Result<BTreeMap<String, CustomInstanceDb>> {
+    Ok(read_external_instance_pg_state(db).await?.databases)
+}
+
+/// The workspaces whose data tables or Ducklake catalogs name each database on the external cluster,
+/// and the forks whose Ducklake metadata schemas there are still waiting to be dropped: those rows
+/// outlive a settings change, and cleanup cannot drop a schema in a database that is gone. A row
+/// whose schema is already dropped only waits on object storage, which needs no database.
+///
+/// Authorization: reads every workspace's settings and checks nothing. Callers MUST be superadmin
+/// or an internal lifecycle path.
+pub async fn external_instance_database_usages<'c>(
+    db: impl sqlx::PgExecutor<'c>,
+) -> Result<BTreeMap<String, BTreeSet<String>>> {
+    let rows = sqlx::query_as::<_, (String, String)>(
+        "SELECT ws.workspace_id, entry->'database'->>'resource_path'
+         FROM workspace_settings ws
+         CROSS JOIN LATERAL jsonb_each(
+             CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
+                 THEN ws.datatable->'datatables'
+                 ELSE '{}'::jsonb END
+         ) AS dt(k, entry)
+         WHERE entry->'database'->>'resource_type' = 'external_instance'
+           AND entry->'database'->>'resource_path' IS NOT NULL
+         UNION ALL
+         SELECT ws.workspace_id, entry->'catalog'->>'resource_path'
+         FROM workspace_settings ws
+         CROSS JOIN LATERAL jsonb_each(
+             CASE WHEN jsonb_typeof(ws.ducklake->'ducklakes') = 'object'
+                 THEN ws.ducklake->'ducklakes'
+                 ELSE '{}'::jsonb END
+         ) AS dl(k, entry)
+         WHERE entry->'catalog'->>'resource_type' = 'external_instance'
+           AND entry->'catalog'->>'resource_path' IS NOT NULL
+         UNION ALL
+         SELECT workspace_id, substring(catalog FROM length('external_instance:') + 1)
+         FROM fork_ducklake_namespace
+         WHERE catalog LIKE 'external\\_instance:%' AND NOT schema_dropped",
+    )
+    .fetch_all(db)
+    .await?;
+    let mut usages: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+    for (workspace_id, dbname) in rows {
+        usages.entry(dbname).or_default().insert(workspace_id);
+    }
+    Ok(usages)
+}
+
+/// 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?;
+    if state.databases.is_empty() && usages.is_empty() {
+        return Ok(());
+    }
+    let names = state
+        .databases
+        .keys()
+        .chain(usages.keys())
+        .collect::<BTreeSet<_>>()
+        .into_iter()
+        .cloned()
+        .collect::<Vec<_>>()
+        .join(", ");
+    Err(Error::BadRequest(format!(
+        "The external instance cluster still holds databases in use ({names}). Drop them and \
+         repoint the data tables and Ducklake catalogs using them before removing {EXTERNAL_INSTANCE_PG_SETTING}."
+    )))
+}
+
+/// Refuse a workspace setting that newly names an `external_instance` database on an edition
+/// without them.
+pub fn ensure_external_instance_available() -> Result<()> {
+    crate::external_instance_pg_oss::ensure_external_instance_available()
+}
+
+/// The connection an `external_instance` database resolves to: `custom_instance_user`, or the
+/// replication user, on the external cluster.
+///
+/// Authorization: returns live credentials and checks nothing. Callers MUST have authorized access
+/// to the data table that names `dbname`.
+pub async fn external_instance_connection_unchecked(
+    db: &DB,
+    dbname: &str,
+    replication: bool,
+) -> Result<crate::PgDatabase> {
+    crate::external_instance_pg_oss::external_instance_connection_unchecked(db, dbname, replication)
+        .await
+}
+
+/// Create `dbname` on the external cluster and register it. Refuses a name already taken there,
+/// whoever took it.
+///
+/// Authorization: checks nothing. Callers MUST be superadmin, or be cloning a data table they may
+/// fork into a `wm_fork_` database.
+pub async fn create_external_instance_database_unchecked(
+    db: &DB,
+    dbname: &str,
+    tag: &str,
+    for_workspace: Option<&str>,
+) -> Result<()> {
+    crate::external_instance_pg_oss::create_external_instance_database_unchecked(
+        db,
+        dbname,
+        tag,
+        for_workspace,
+    )
+    .await
+}
+
+/// Drop `dbname` from the external cluster: only a database Windmill registered creating, and still
+/// carries the mark it set there. Refused while anything uses it
+/// ([`crate::workspaces::managed_database_uses`]), except the `exempt` data table entry: the fork
+/// copy being cleaned up.
+///
+/// Authorization: checks nothing. Callers MUST be superadmin, or be deleting the fork that owns
+/// this `wm_fork_` database.
+pub async fn drop_external_instance_database_unchecked(
+    db: &DB,
+    dbname: &str,
+    exempt: Option<(&str, &str)>,
+) -> Result<()> {
+    crate::external_instance_pg_oss::drop_external_instance_database_unchecked(db, dbname, exempt)
+    .await
+}
+
+/// Serializes everything that changes which databases exist on the external cluster, or which data
+/// tables name them: setup, creates, drops, and data table saves. Held until `tx` ends.
+pub async fn lock_external_instance_pg_state(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+) -> Result<()> {
+    sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1))")
+        .bind(EXTERNAL_INSTANCE_PG_STATE_SETTING)
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
+/// 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.
+pub async fn ensure_external_instance_database_registered(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    dbname: &str,
+) -> Result<()> {
+    lock_external_instance_pg_state(tx).await?;
+    if read_external_instance_pg_state(&mut **tx)
+        .await?
+        .databases
+        .contains_key(dbname)
+    {
+        return Ok(());
+    }
+    Err(Error::BadRequest(format!(
+        "Windmill did not create a database named '{dbname}' on the external instance cluster. \
+         Create it from the instance settings first."
+    )))
+}
+
+/// Write [`EXTERNAL_INSTANCE_PG_SETTING`]: `None`, null or an empty string unsets it. Every writer
+/// of global settings goes through this for that key — the per-key and bulk endpoints as well as
+/// the declarative sync — instead of writing the row itself.
+///
+/// The checks and the write share one transaction holding [`lock_external_instance_pg_state`]. A
+/// check taken outside it could pass while a database create still reads the old cluster, which
+/// would then register a database there after the setting names another one.
+///
+/// Authorization: checks nothing. Callers MUST be superadmin, or the declarative instance config
+/// sync, which applies what the operator deployed.
+pub async fn write_external_instance_pg_setting(
+    db: &DB,
+    value: Option<&serde_json::Value>,
+) -> Result<()> {
+    let value = match value {
+        None | Some(serde_json::Value::Null) => None,
+        Some(serde_json::Value::String(s)) if s.trim().is_empty() => None,
+        Some(value) => Some(value),
+    };
+    let mut tx = db.begin().await?;
+    lock_external_instance_pg_state(&mut tx).await?;
+    match value {
+        None => {
+            ensure_external_instance_pg_removable(db).await?;
+            sqlx::query("DELETE FROM global_settings WHERE name = $1")
+                .bind(EXTERNAL_INSTANCE_PG_SETTING)
+                .execute(&mut *tx)
+                .await?;
+        }
+        Some(value) => {
+            crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?;
+            ensure_external_instance_pg_not_repointed(db, 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()",
+            )
+            .bind(EXTERNAL_INSTANCE_PG_SETTING)
+            .bind(value)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+    tx.commit().await?;
+    tracing::info!(
+        "{} global setting {EXTERNAL_INSTANCE_PG_SETTING}",
+        if value.is_some() { "Set" } else { "Unset" }
+    );
+    Ok(())
+}
+
+/// [`write_external_instance_pg_setting`] for a settings diff: writes the key if the diff touches
+/// it, and takes it out of the diff so the generic apply does not write it again.
+///
+/// Authorization: checks nothing. Callers MUST be superadmin, or the declarative instance config
+/// sync, which applies what the operator deployed.
+pub async fn write_external_instance_pg_from_diff(
+    db: &DB,
+    diff: &mut crate::instance_config::SettingsDiff,
+) -> Result<()> {
+    if let Some(value) = diff.upserts.remove(EXTERNAL_INSTANCE_PG_SETTING) {
+        write_external_instance_pg_setting(db, Some(&value)).await?;
+    }
+    if let Some(i) = diff
+        .deletes
+        .iter()
+        .position(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
+    {
+        diff.deletes.remove(i);
+        write_external_instance_pg_setting(db, None).await?;
+    }
+    Ok(())
+}
+
+/// Refuse pointing the setting at another host or port while databases live on the current one.
+/// 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,
+    value: &serde_json::Value,
+) -> Result<()> {
+    let Some(current) = read_external_instance_pg_config(db).await? else {
+        return Ok(());
+    };
+    let Ok(desired) = serde_json::from_value::<ExternalInstancePg>(value.clone()) else {
+        return Ok(());
+    };
+    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?;
+    if state.databases.is_empty() && usages.is_empty() {
+        return Ok(());
+    }
+    Err(Error::BadRequest(format!(
+        "The external instance cluster at {}:{} still holds databases in use. Drop them and repoint \
+         what uses them before pointing {EXTERNAL_INSTANCE_PG_SETTING} at another cluster.",
+        current.host.trim(),
+        current.port.unwrap_or(5432)
+    )))
+}
+
+/// Converge the external cluster on the configured login: check what it can do, create or update
+/// Windmill's two roles with the stored passwords, and report anything that would get in the way.
+/// With `rotate_passwords`, generate new passwords first. Safe to run again; running it again is
+/// how a failed rotation is repaired.
+///
+/// Authorization: administers the external cluster with its admin credentials and checks nothing.
+/// Callers MUST be superadmin.
+pub async fn setup_external_instance_pg_unchecked(
+    db: &DB,
+    rotate_passwords: bool,
+) -> Result<ExternalInstancePgSetupReport> {
+    crate::external_instance_pg_oss::setup_external_instance_pg_unchecked(db, rotate_passwords)
+        .await
+}

backend/windmill-common/src/external_instance_pg_oss.rs

@@ -0,0 +1,81 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the external instance cluster comes from: the enterprise implementation, or a refusal.
+//! `private` alone is not that edition: community builds carry it.
+
+use crate::error::Error;
+
+pub fn external_instance_pg_unavailable() -> Error {
+    Error::BadRequest(
+        "External instance databases are a Windmill Enterprise Edition feature".to_string(),
+    )
+}
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::external_instance_pg_ee::{
+    create_external_instance_database_unchecked, drop_external_instance_database_unchecked,
+    external_instance_connection_unchecked, setup_external_instance_pg_unchecked,
+    validate_external_instance_pg_setting,
+};
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) fn ensure_external_instance_available() -> crate::error::Result<()> {
+    Ok(())
+}
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) use ce::*;
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+mod ce {
+    use super::external_instance_pg_unavailable as unavailable;
+    use crate::{
+        error::Result, external_instance_pg::ExternalInstancePgSetupReport, PgDatabase, DB,
+    };
+
+    pub(crate) fn validate_external_instance_pg_setting(_value: &serde_json::Value) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) fn ensure_external_instance_available() -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn setup_external_instance_pg_unchecked(
+        _db: &DB,
+        _rotate_passwords: bool,
+    ) -> Result<ExternalInstancePgSetupReport> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn external_instance_connection_unchecked(
+        _db: &DB,
+        _dbname: &str,
+        _replication: bool,
+    ) -> Result<PgDatabase> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn create_external_instance_database_unchecked(
+        _db: &DB,
+        _dbname: &str,
+        _tag: &str,
+        _for_workspace: Option<&str>,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+
+    pub(crate) async fn drop_external_instance_database_unchecked(
+        _db: &DB,
+        _dbname: &str,
+        _exempt: Option<(&str, &str)>,
+    ) -> Result<()> {
+        Err(unavailable())
+    }
+}

backend/windmill-common/src/global_settings.rs

@@ -57,6 +57,8 @@ pub const SAML_METADATA_SETTING: &str = "saml_metadata";
 pub const SMTP_SETTING: &str = "smtp_settings";
 pub const TEAMS_SETTING: &str = "teams";
 pub const INDEXER_SETTING: &str = "indexer_settings";
+pub const EXTERNAL_INSTANCE_PG_SETTING: &str = "external_instance_pg";
+pub const EXTERNAL_INSTANCE_PG_STATE_SETTING: &str = "external_instance_pg_state";
 pub const TIMEOUT_WAIT_RESULT_SETTING: &str = "timeout_wait_result";
 
 pub const UNIQUE_ID_SETTING: &str = "uid";
@@ -355,6 +357,9 @@ pub const AGENT_WORKER_BLOCKED_SETTINGS: &[&str] = &[
     // resolve datatable connections through the dedicated datatable endpoints, never these.
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    // The external cluster's admin login, and the passwords Windmill generated on it.
+    EXTERNAL_INSTANCE_PG_SETTING,
+    EXTERNAL_INSTANCE_PG_STATE_SETTING,
 ];
 
 /// Whether an agent worker may read the given global setting over HTTP.

backend/windmill-common/src/instance_config.rs

@@ -350,6 +350,8 @@ pub struct GlobalSettings {
     pub ducklake_settings: Option<DucklakeSettings>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub custom_instance_pg_databases: Option<CustomInstancePgDatabases>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub external_instance_pg: Option<ExternalInstancePg>,
 
     // Opaque settings (EE-private structs or no clear schema)
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -836,6 +838,36 @@ pub struct CustomInstanceDbLogs {
     pub replication_user_error: Option<String>,
 }
 
+// ---------------------------------------------------------------------------
+// External instance PG cluster
+// ---------------------------------------------------------------------------
+
+/// The external Postgres cluster Windmill manages for `external_instance` data tables and Ducklake
+/// catalogs. `user` logs in as the cluster's administrator: it needs `CREATEDB` and `CREATEROLE`.
+/// `dbname` is only where that login connects to run cluster-wide statements.
+///
+/// Every field defaults rather than being required: this deserializes as part of the whole
+/// instance config, and one malformed row must not make every other setting unreadable. The
+/// write path and every use reject an incomplete value instead.
+#[derive(Deserialize, Serialize, Clone, Debug, Default)]
+#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
+pub struct ExternalInstancePg {
+    #[serde(default)]
+    pub host: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub port: Option<u16>,
+    #[serde(default)]
+    pub user: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub password: Option<StringOrSecretRef>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dbname: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sslmode: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub root_certificate_pem: Option<String>,
+}
+
 // ---------------------------------------------------------------------------
 // Autoscaling (worker config)
 // ---------------------------------------------------------------------------
@@ -972,6 +1004,7 @@ pub const PROTECTED_SETTINGS: &[&str] = &[
     "ducklake_settings",
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    "external_instance_pg_state",
     "uid",
     "rsa_keys",
     "jwt_secret",
@@ -997,6 +1030,8 @@ pub const HIDDEN_SETTINGS: &[&str] = &[
     // Server-only (written by setup/refresh via direct SQL), never operator-authored —
     // hidden so the config machinery can't read, rewrite, or drop it.
     "custom_instance_replication_pwd",
+    // Same for the passwords and database registry Windmill keeps for the external cluster.
+    "external_instance_pg_state",
 ];
 
 /// Top-level settings whose entire value is sensitive and must be fully redacted in logs.
@@ -1008,6 +1043,7 @@ const SENSITIVE_SETTINGS: &[&str] = &[
     "license_key",
     "ducklake_user_pg_pwd",
     "custom_instance_replication_pwd",
+    "external_instance_pg_state",
     "pip_index_url",
     "pip_extra_index_url",
     "npm_config_registry",
@@ -1033,6 +1069,7 @@ const NESTED_SENSITIVE_FIELDS: &[(&str, &[&str])] = &[
         &["secret_key", "serviceAccountKey"],
     ),
     ("custom_instance_pg_databases", &["user_pwd"]),
+    ("external_instance_pg", &["password"]),
 ];
 
 fn redact_json_value(value: &serde_json::Value) -> serde_json::Value {
@@ -1361,7 +1398,8 @@ pub async fn sync_global_settings_declarative(
     crate::global_settings::parse_allowed_origins_setting(desired.get(origins_key))
         .map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?;
 
-    let diff = diff_global_settings(current, desired, ApplyMode::Replace);
+    let mut diff = diff_global_settings(current, desired, ApplyMode::Replace);
+    crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff).await?;
     apply_settings_diff(db, &diff).await?;
 
     Ok(())
@@ -1494,6 +1532,10 @@ pub fn resolve_env_refs(settings: &mut GlobalSettings) -> Result<(), String> {
         resolve_env_option(&mut pg.user_pwd)?;
     }
 
+    if let Some(pg) = &mut settings.external_instance_pg {
+        resolve_env_option(&mut pg.password)?;
+    }
+
     Ok(())
 }
 
@@ -2463,39 +2505,33 @@ mod tests {
     }
 
     #[test]
-    fn custom_instance_replication_pwd_is_isolated_from_config() {
-        // The replication-role password is server-only: written by setup/refresh via direct
-        // SQL, never operator-authored. It must stay out of the declarative config surface
-        // (hidden on read) and be undeletable, so config sync can't read, rewrite, or drop it.
-        assert!(HIDDEN_SETTINGS.contains(&"custom_instance_replication_pwd"));
-        assert!(PROTECTED_SETTINGS.contains(&"custom_instance_replication_pwd"));
-        assert!(SENSITIVE_SETTINGS.contains(&"custom_instance_replication_pwd"));
+    fn server_generated_db_passwords_are_isolated_from_config() {
+        // These hold passwords the server generates: written by setup/refresh via direct SQL,
+        // never operator-authored. They must stay out of the declarative config surface
+        // (hidden on read) and be undeletable, so config sync can't read, rewrite, or drop them.
+        for key in [
+            "custom_instance_replication_pwd",
+            "external_instance_pg_state",
+        ] {
+            assert!(HIDDEN_SETTINGS.contains(&key), "{key}");
+            assert!(PROTECTED_SETTINGS.contains(&key), "{key}");
+            assert!(SENSITIVE_SETTINGS.contains(&key), "{key}");
 
-        // A stray desired value (e.g. flattened into `extra`) is ignored, not upserted.
-        let mut desired = BTreeMap::new();
-        desired.insert(
-            "custom_instance_replication_pwd".to_string(),
-            serde_json::json!("attacker-set"),
-        );
-        let diff = diff_global_settings(&BTreeMap::new(), &desired, ApplyMode::Merge);
-        assert!(
-            diff.upserts.is_empty(),
-            "hidden setting must not be upserted"
-        );
+            // A stray desired value (e.g. flattened into `extra`) is ignored, not upserted.
+            let mut desired = BTreeMap::new();
+            desired.insert(key.to_string(), serde_json::json!("attacker-set"));
+            let diff = diff_global_settings(&BTreeMap::new(), &desired, ApplyMode::Merge);
+            assert!(diff.upserts.is_empty(), "{key} must not be upserted");
 
-        // A current value is never deleted by a Replace that omits it.
-        let mut current = BTreeMap::new();
-        current.insert(
-            "custom_instance_replication_pwd".to_string(),
-            serde_json::json!("live"),
-        );
-        let diff = diff_global_settings(&current, &BTreeMap::new(), ApplyMode::Replace);
-        assert!(
-            !diff
-                .deletes
-                .contains(&"custom_instance_replication_pwd".to_string()),
-            "hidden setting must not be deleted"
-        );
+            // A current value is never deleted by a Replace that omits it.
+            let mut current = BTreeMap::new();
+            current.insert(key.to_string(), serde_json::json!("live"));
+            let diff = diff_global_settings(&current, &BTreeMap::new(), ApplyMode::Replace);
+            assert!(
+                !diff.deletes.contains(&key.to_string()),
+                "{key} must not be deleted"
+            );
+        }
     }
 
     #[test]

backend/windmill-common/src/lib.rs

@@ -58,6 +58,10 @@ pub mod ee_oss;
 pub mod email_ee;
 pub mod email_oss;
 pub mod error;
+pub mod external_instance_pg;
+#[cfg(all(feature = "private", feature = "enterprise"))]
+mod external_instance_pg_ee;
+pub mod external_instance_pg_oss;
 pub mod external_ip;
 #[cfg(feature = "private")]
 pub mod feature_usage_ee;
@@ -1082,7 +1086,13 @@ impl PgDatabase {
                 if err_str.contains("password authentication failed for user")
                     && err_str.contains("custom_instance_user")
                 {
-                    if let Some(db) = main_db {
+                    // The external instance cluster has a `custom_instance_user` of its own, whose
+                    // password setup manages. Rotating the local one would break every instance
+                    // data table and fix nothing.
+                    let local = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
+                    let on_local_cluster = local.host == self.host
+                        && local.port.unwrap_or(5432) == self.port.unwrap_or(5432);
+                    if let Some(db) = main_db.filter(|_| on_local_cluster) {
                         tracing::warn!(
                             "custom_instance_user password auth failed, refreshing and retrying..."
                         );
@@ -1649,39 +1659,62 @@ pub async fn create_custom_instance_database(
     Ok(())
 }
 
-/// Refuse a workspace member writing a fork copy into, or pointing a fork at, the instance database
-/// `dbname`, unless `w_id` created it for that ([`create_custom_instance_database`]) and nothing uses
-/// it yet. The `wm_fork_` prefix is no authorization: every instance database answers to the same
-/// `custom_instance_user`, so a name is all it takes to reach another workspace's copy.
+/// The system's CA bundle file, for libpq clients that cannot take `sslrootcert=system`: that value
+/// needs libpq 16, and verify-full only.
+pub fn system_ca_bundle() -> Option<std::path::PathBuf> {
+    std::env::var_os("SSL_CERT_FILE")
+        .map(std::path::PathBuf::from)
+        .into_iter()
+        .chain(
+            [
+                "/etc/ssl/certs/ca-certificates.crt",
+                "/etc/pki/tls/certs/ca-bundle.crt",
+                "/etc/ssl/cert.pem",
+                "/etc/ssl/ca-bundle.pem",
+            ]
+            .map(std::path::PathBuf::from),
+        )
+        .find(|path| path.is_file())
+}
+
+/// Refuse a workspace member writing a fork copy into, or pointing a fork at, the managed database
+/// `dbname` of `kind`, unless `w_id` created it for that ([`create_custom_instance_database`], or
+/// its external instance counterpart) and nothing uses it yet. The `wm_fork_` prefix is no
+/// authorization: every database of a cluster answers to the same `custom_instance_user`, so a name
+/// is all it takes to reach another workspace's copy.
 ///
-/// Authorization: reads the global registry and every workspace's settings, and names other
+/// Authorization: reads the global registries and every workspace's settings, and names other
 /// workspaces in its refusal. Callers MUST have authorized `w_id` for the caller first — a member
 /// of it forking or importing there — and MUST NOT call it on a workspace the caller is not in.
 pub async fn ensure_fork_database_available_to(
     db: &DB,
+    kind: workspaces::DataTableCatalogResourceType,
     dbname: &str,
     w_id: &str,
 ) -> error::Result<()> {
-    let created_for = 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 created_for = match kind {
+        workspaces::DataTableCatalogResourceType::ExternalInstance => {
+            external_instance_pg::external_instance_databases(db)
+                .await?
+                .remove(dbname)
+                .and_then(|entry| entry.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(),
+    };
     if created_for.as_deref() != Some(w_id) {
         return Err(Error::BadRequest(format!(
             "Database '{dbname}' was not created for a fork of workspace '{w_id}'"
         )));
     }
-    let uses = workspaces::managed_database_uses(
-        &mut *db.acquire().await?,
-        workspaces::DataTableCatalogResourceType::Instance,
-        dbname,
-        None,
-    )
-    .await?;
+    let uses =
+        workspaces::managed_database_uses(&mut *db.acquire().await?, kind, dbname, None).await?;
     if !uses.is_empty() {
         return Err(Error::BadRequest(format!(
             "Database '{dbname}' is already in use: {}",

backend/windmill-common/src/workspaces.rs

@@ -1397,6 +1397,18 @@ pub enum DataTableCatalogResourceType {
     #[strum(serialize = "postgres")]
     Postgresql,
     Instance,
+    /// On the external instance cluster ([`crate::external_instance_pg`]). Enterprise Edition.
+    #[serde(rename = "external_instance")]
+    #[strum(serialize = "external_instance")]
+    ExternalInstance,
+}
+
+impl DataTableCatalogResourceType {
+    /// A database Windmill created and administers, on its own cluster or the external one, as
+    /// opposed to one a user brought as a resource.
+    pub fn is_windmill_managed(self) -> bool {
+        matches!(self, Self::Instance | Self::ExternalInstance)
+    }
 }
 
 /// Build a self-teaching error for an unresolved `datatable://<name>` reference.
@@ -1680,7 +1692,8 @@ pub async fn resolve_workspace_governing_datatables(
 }
 
 /// Build the `admin` connection for a governing entry: `custom_instance_user` for an instance
-/// database, the user's own resource for a BYO-postgres one.
+/// database, on Windmill's cluster or the external one; the user's own resource for a BYO-postgres
+/// one.
 async fn resolve_datatable_connection_unchecked(
     db: &DB,
     governing: &GoverningDatatable,
@@ -1691,7 +1704,16 @@ async fn resolve_datatable_connection_unchecked(
         .database
         .as_ref()
         .expect("a governing entry owns a database");
-    if database.resource_type == DataTableCatalogResourceType::Instance {
+    if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
+        let pg_creds = crate::external_instance_pg::external_instance_connection_unchecked(
+            db,
+            &database.resource_path,
+            replication,
+        )
+        .await?;
+        serde_json::to_value(&pg_creds)
+            .map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))
+    } else if database.resource_type == DataTableCatalogResourceType::Instance {
         let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
         pg_creds.dbname = database.resource_path.clone();
         if replication {
@@ -2235,6 +2257,10 @@ pub enum DucklakeCatalogResourceType {
     Postgresql,
     Mysql,
     Instance,
+    /// On the external instance cluster ([`crate::external_instance_pg`]). Enterprise Edition.
+    #[serde(rename = "external_instance")]
+    #[strum(serialize = "external_instance")]
+    ExternalInstance,
 }
 
 #[derive(Deserialize, Serialize)]
@@ -2762,7 +2788,16 @@ async fn ducklake_conn_data(
     let ducklake = serde_json::from_value::<Ducklake>(ducklake)?;
 
     let catalog_resource =
-        if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
+        if ducklake.catalog.resource_type == DucklakeCatalogResourceType::ExternalInstance {
+            let pg_creds = crate::external_instance_pg::external_instance_connection_unchecked(
+                db,
+                &ducklake.catalog.resource_path,
+                false,
+            )
+            .await?;
+            serde_json::to_value(&pg_creds)
+                .map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))?
+        } else if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
             let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
             pg_creds.dbname = ducklake.catalog.resource_path.clone();
             pg_creds.user = Some("custom_instance_user".to_string());
@@ -3143,6 +3178,14 @@ async fn register_fork_ducklake_namespace(
     {
         return Ok(());
     }
+    let mut tx = db.begin().await?;
+    // A row naming an external database counts as a use of it. Written under the lock a drop takes,
+    // and only while the database is still registered, so a drop cannot slip in between the
+    // settings this attach resolved and the row that protects the database.
+    if let Some(dbname) = catalog.strip_prefix("external_instance:") {
+        crate::external_instance_pg::ensure_external_instance_database_registered(&mut tx, dbname)
+            .await?;
+    }
     sqlx::query!(
         "INSERT INTO fork_ducklake_namespace
            (workspace_id, ducklake_name, metadata_schema, catalog, storage, storage_ref, data_path)
@@ -3157,9 +3200,10 @@ async fn register_fork_ducklake_namespace(
         &storage_ref,
         data_path,
     )
-    .execute(db)
+    .execute(&mut *tx)
     .await
     .map_err(|e| Error::internal_err(format!("registering fork ducklake namespace: {e:#}")))?;
+    tx.commit().await?;
     let mut locations = FORK_DUCKLAKE_REGISTERED
         .get(w_id)
         .filter(|(_, exp)| *exp > now)

backend/windmill-worker/src/duckdb_executor.rs

@@ -2240,11 +2240,99 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option<ParsedAttachDbResource
     None
 }
 
+/// The verification a DuckDB postgres attach keeps, as its libpq `sslmode` and `sslrootcert`.
+///
+/// Attaches have always turned verify-ca and verify-full into `require`, which resources rely on.
+/// 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)>> {
+    let mode = match res.sslmode.as_deref() {
+        Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode,
+        _ => return Ok(None),
+    };
+    let bundle = windmill_common::system_ca_bundle()
+        .map(std::fs::read_to_string)
+        .transpose()
+        .map_err(|e| Error::ExecutionErr(format!("Failed to read the system CA bundle: {e}")))?
+        .unwrap_or_default();
+    let pem = res.root_certificate_pem.as_deref().unwrap_or_default();
+    if bundle.is_empty() && pem.is_empty() {
+        return Err(Error::ExecutionErr(format!(
+            "sslmode {mode} needs a root certificate, and this worker has no system CA bundle"
+        )));
+    }
+    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",
+        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);
+    }
+    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> {
+    let uri = res.to_uri();
+    let Some((mode, roots)) = pg_attach_verification(res)? else {
+        return Ok(uri);
+    };
+    let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| {
+        Error::internal_err("unexpected sslmode in a postgres connection URI".to_string())
+    })?;
+    Ok(format!(
+        "{base}?sslmode={mode}&sslrootcert={}",
+        urlencoding::encode(&roots.to_string_lossy())
+    ))
+}
+
 fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result<String> {
     let s = match db_type.to_lowercase().as_str() {
         "postgres" | "postgresql" => {
             let res: PgDatabase = serde_json::from_value(db_resource)?;
-            res.to_uri()
+            pg_attach_uri(&res)?
         }
         #[cfg(feature = "mysql")]
         "mysql" => {
@@ -2391,7 +2479,9 @@ async fn transform_attach_ducklake(
         format!(", {}", user_extra_args)
     };
     let db_type = match ducklake.catalog.resource_type {
-        DucklakeCatalogResourceType::Instance => "postgres",
+        DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance => {
+            "postgres"
+        }
         _ => ducklake.catalog.resource_type.as_ref(),
     };
 
@@ -2507,7 +2597,8 @@ fn fork_defer_statements(
             hidden_passwords.lock().unwrap().push(pwd.to_string());
         }
         let db_type = match a.catalog.resource_type {
-            DucklakeCatalogResourceType::Instance => "postgres",
+            DucklakeCatalogResourceType::Instance
+            | DucklakeCatalogResourceType::ExternalInstance => "postgres",
             _ => a.catalog.resource_type.as_ref(),
         };
         stmts.push(get_attach_db_install_str(db_type)?.to_string());
@@ -2700,10 +2791,21 @@ fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<V
     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 res.sslmode.as_deref() {
-        Some("disable") => "disable",
-        Some("require") | Some("verify-ca") | Some("verify-full") => "require",
-        _ => "prefer",
+    let sslmode = match pg_attach_verification(&res)? {
+        // A libpq keyword/value string: the path is quoted for libpq, then for the DuckDB literal.
+        Some((mode, roots)) => format!(
+            "{mode} sslrootcert=''{}''",
+            roots
+                .to_string_lossy()
+                .replace('\\', "\\\\")
+                .replace('\'', "\\''")
+        ),
+        None => match res.sslmode.as_deref() {
+            Some("disable") => "disable",
+            Some("require") | Some("verify-ca") | Some("verify-full") => "require",
+            _ => "prefer",
+        }
+        .to_string(),
     };
     let secret_name = datatable_secret_name(alias_name);
     Ok(vec![
@@ -2796,6 +2898,46 @@ pub struct Arg {
 mod tests {
     use super::*;
 
+    #[test]
+    fn pg_attach_keeps_verification_only_when_required() {
+        let pg = |sslmode: &str, accept_invalid_certs: Option<bool>| PgDatabase {
+            host: "db.internal".to_string(),
+            user: Some("custom_instance_user".to_string()),
+            password: Some("pw".to_string()),
+            port: None,
+            sslmode: Some(sslmode.to_string()),
+            dbname: "dt".to_string(),
+            root_certificate_pem: Some("-----BEGIN CERTIFICATE-----test".to_string()),
+            accept_invalid_certs,
+            use_iam_auth: None,
+            region: None,
+        };
+        let uri = pg_attach_uri(&pg("verify-full", Some(false))).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) {
+            let mut other = pg("verify-full", Some(false));
+            other.root_certificate_pem = Some(format!("-----BEGIN CERTIFICATE-----{i}"));
+            pg_attach_uri(&other).unwrap();
+        }
+        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"));
+        let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap();
+        let attach = &pg_secret_attach_statements(external, "dt").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"));
+    }
+
     #[test]
     fn attach_datatable_parses_name_and_role() {
         let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;