---
AGENTS.md | 4 +
...d368347c59f36fd87df6dc7996152ccb84af0.json | 38 --
...9ad2db940c2455ab19cc95e5198baf96d5629.json | 29 -
...df2a63695f8cbf8af648da5a0a77a5b9d02ba.json | 17 -
...12373638be391f884f81dc387ffc465badac6.json | 20 -
...8dc45020a2a553d8874c49f9eafedea5a9d40.json | 24 -
backend/Cargo.lock | 1 +
backend/Cargo.toml | 1 +
backend/ee-repo-ref.txt | 2 +-
...0917133824_datatable_role_cluster.down.sql | 22 +
...260917133824_datatable_role_cluster.up.sql | 8 +
backend/tests/instance_config.rs | 46 ++
.../tests/datatable_roles.rs | 12 +-
backend/windmill-api-settings/src/lib.rs | 249 +++++++-
.../src/datatable_acl.rs | 78 ++-
.../src/datatable_clone.rs | 100 +++-
.../src/datatable_migrations.rs | 6 +-
.../windmill-api-workspaces/src/workspaces.rs | 533 ++++++++++++++----
.../src/workspaces_extra.rs | 184 +++++-
backend/windmill-api/openapi.yaml | 177 +++++-
backend/windmill-common/Cargo.toml | 1 +
.../windmill-common/src/datatable_roles.rs | 171 ++++--
.../src/datatable_roles_oss.rs | 64 ++-
.../src/external_instance_pg.rs | 456 +++++++++++++++
.../src/external_instance_pg_oss.rs | 82 +++
.../windmill-common/src/global_settings.rs | 8 +
.../windmill-common/src/instance_config.rs | 103 ++--
backend/windmill-common/src/lib.rs | 153 ++++-
backend/windmill-common/src/workspaces.rs | 197 ++++++-
.../windmill-worker/src/duckdb_executor.rs | 285 ++++++++--
cli/src/commands/datatable/datatable.ts | 6 +-
docs/external-instance-datatables.md | 88 +++
.../src/lib/components/InstanceSetting.svelte | 22 +-
.../lib/components/InstanceSettings.svelte | 13 +
.../src/lib/components/instanceSettings.ts | 31 +
.../ExternalInstancePgSettings.svelte | 471 ++++++++++++++++
.../InstancePgSettings.svelte | 65 +++
.../AddDataTableWizard.svelte | 153 ++++-
.../CustomInstanceDbWizardModal.svelte | 5 +-
.../DataTablePermissionsButton.svelte | 20 +-
.../DataTableRolesSection.svelte | 71 ++-
.../DataTableSettings.svelte | 137 ++++-
.../workspaceSettings/DucklakeSettings.svelte | 399 +++++++------
.../ExternalInstanceDbSelect.svelte | 117 ++++
.../InstanceRolesButton.svelte | 8 +-
.../addDataTableModel.test.ts | 48 +-
.../workspaceSettings/addDataTableModel.ts | 99 +++-
.../workspaceSettings/datatableTelemetry.ts | 2 +-
.../workspaceSettings/utils.svelte.ts | 24 +
.../workspaceSettings/wizardParking.ts | 3 +
50 files changed, 4121 insertions(+), 732 deletions(-)
delete mode 100644 backend/.sqlx/query-71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0.json
delete mode 100644 backend/.sqlx/query-79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629.json
delete mode 100644 backend/.sqlx/query-86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba.json
delete mode 100644 backend/.sqlx/query-b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6.json
delete mode 100644 backend/.sqlx/query-d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40.json
create mode 100644 backend/migrations/20260917133824_datatable_role_cluster.down.sql
create mode 100644 backend/migrations/20260917133824_datatable_role_cluster.up.sql
create mode 100644 backend/windmill-common/src/external_instance_pg.rs
create mode 100644 backend/windmill-common/src/external_instance_pg_oss.rs
create mode 100644 docs/external-instance-datatables.md
create mode 100644 frontend/src/lib/components/instanceSettings/ExternalInstancePgSettings.svelte
create mode 100644 frontend/src/lib/components/instanceSettings/InstancePgSettings.svelte
create mode 100644 frontend/src/lib/components/workspaceSettings/ExternalInstanceDbSelect.svelte
diff --git a/AGENTS.md b/AGENTS.md
index 858e9329ff..850538f080 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -29,6 +29,10 @@ Open-source platform for internal tools, workflows, API integrations, background
- **Agent workers**: `docs/agent-worker-e2e.md` — building and running one locally. An agent
reaches the DB only through the API, so `Connection::Http` paths are never taken by a plain
`cargo run`; a normal build cannot start one at all.
+- **External instance data tables**: `docs/external-instance-datatables.md` — the cluster Windmill
+ administers behind `external_instance` data tables and Ducklake catalogs: its invariants (one
+ lifecycle lock, managed-object markers, the setup gate, per-cluster roles, fork copy ownership)
+ and how to run one locally
- **Enterprise**: `docs/enterprise.md` — EE file conventions and PR workflow
- **Operator write rights**: `docs/operator-write-rights.md` — which `operator_settings` flags are
enforced rather than cosmetic, and why a right that is granted-unless-withdrawn needs `Option`
diff --git a/backend/.sqlx/query-71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0.json b/backend/.sqlx/query-71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0.json
deleted file mode 100644
index a03fab7081..0000000000
--- a/backend/.sqlx/query-71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0.json
+++ /dev/null
@@ -1,38 +0,0 @@
-{
- "db_name": "PostgreSQL",
- "query": "SELECT id, name, enabled, pwd FROM datatable_role",
- "describe": {
- "columns": [
- {
- "ordinal": 0,
- "name": "id",
- "type_info": "Varchar"
- },
- {
- "ordinal": 1,
- "name": "name",
- "type_info": "Varchar"
- },
- {
- "ordinal": 2,
- "name": "enabled",
- "type_info": "Bool"
- },
- {
- "ordinal": 3,
- "name": "pwd",
- "type_info": "Text"
- }
- ],
- "parameters": {
- "Left": []
- },
- "nullable": [
- false,
- false,
- false,
- true
- ]
- },
- "hash": "71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0"
-}
diff --git a/backend/.sqlx/query-79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629.json b/backend/.sqlx/query-79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629.json
deleted file mode 100644
index c36b6fb361..0000000000
--- a/backend/.sqlx/query-79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629.json
+++ /dev/null
@@ -1,29 +0,0 @@
-{
- "db_name": "PostgreSQL",
- "query": "\n SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n FROM workspace_settings ws\n JOIN workspace w ON w.id = ws.workspace_id AND w.deleted = false\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE ws.workspace_id <> $1\n AND dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $2\n ORDER BY ws.workspace_id, dt.key\n ",
- "describe": {
- "columns": [
- {
- "ordinal": 0,
- "name": "workspace_id!",
- "type_info": "Varchar"
- },
- {
- "ordinal": 1,
- "name": "datatable!",
- "type_info": "Text"
- }
- ],
- "parameters": {
- "Left": [
- "Text",
- "Text"
- ]
- },
- "nullable": [
- false,
- null
- ]
- },
- "hash": "79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629"
-}
diff --git a/backend/.sqlx/query-86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba.json b/backend/.sqlx/query-86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba.json
deleted file mode 100644
index 7627d9828d..0000000000
--- a/backend/.sqlx/query-86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba.json
+++ /dev/null
@@ -1,17 +0,0 @@
-{
- "db_name": "PostgreSQL",
- "query": "INSERT INTO datatable_role (id, name, enabled, pwd) VALUES ($1, $2, $3, $4)",
- "describe": {
- "columns": [],
- "parameters": {
- "Left": [
- "Varchar",
- "Varchar",
- "Bool",
- "Text"
- ]
- },
- "nullable": []
- },
- "hash": "86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba"
-}
diff --git a/backend/.sqlx/query-b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6.json b/backend/.sqlx/query-b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6.json
deleted file mode 100644
index 4b3a2f33e9..0000000000
--- a/backend/.sqlx/query-b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6.json
+++ /dev/null
@@ -1,20 +0,0 @@
-{
- "db_name": "PostgreSQL",
- "query": "SELECT jsonb_object_keys(value->'databases') FROM global_settings\n WHERE name = 'custom_instance_pg_databases'",
- "describe": {
- "columns": [
- {
- "ordinal": 0,
- "name": "jsonb_object_keys",
- "type_info": "Text"
- }
- ],
- "parameters": {
- "Left": []
- },
- "nullable": [
- null
- ]
- },
- "hash": "b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6"
-}
diff --git a/backend/.sqlx/query-d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40.json b/backend/.sqlx/query-d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40.json
deleted file mode 100644
index 49875acd78..0000000000
--- a/backend/.sqlx/query-d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40.json
+++ /dev/null
@@ -1,24 +0,0 @@
-{
- "db_name": "PostgreSQL",
- "query": "SELECT dt.key AS \"datatable!\"\n FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE ws.workspace_id = $1\n AND dt.key <> $2\n AND NOT dt.value ? 'permissions'\n AND dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $3\n ORDER BY dt.key",
- "describe": {
- "columns": [
- {
- "ordinal": 0,
- "name": "datatable!",
- "type_info": "Text"
- }
- ],
- "parameters": {
- "Left": [
- "Text",
- "Text",
- "Text"
- ]
- },
- "nullable": [
- null
- ]
- },
- "hash": "d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40"
-}
diff --git a/backend/Cargo.lock b/backend/Cargo.lock
index 9182843789..e80d0edf08 100644
--- a/backend/Cargo.lock
+++ b/backend/Cargo.lock
@@ -15660,6 +15660,7 @@ dependencies = [
"pin-project-lite",
"pkcs1",
"postgres-native-tls 0.5.3",
+ "postgres-protocol",
"prometheus",
"quick_cache",
"rand 0.9.5",
diff --git a/backend/Cargo.toml b/backend/Cargo.toml
index 77f512a3dd..8571b0b99c 100644
--- a/backend/Cargo.toml
+++ b/backend/Cargo.toml
@@ -626,6 +626,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"
diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt
index 58f9aaf07d..21e445394e 100644
--- a/backend/ee-repo-ref.txt
+++ b/backend/ee-repo-ref.txt
@@ -1 +1 @@
-9855e1b7a43a0a33e04f8accf1c497af3fd9b139
+dfe7b8b9204d21e0264fbea1c6f6eedf9e738d56
diff --git a/backend/migrations/20260917133824_datatable_role_cluster.down.sql b/backend/migrations/20260917133824_datatable_role_cluster.down.sql
new file mode 100644
index 0000000000..379f1e3d34
--- /dev/null
+++ b/backend/migrations/20260917133824_datatable_role_cluster.down.sql
@@ -0,0 +1,22 @@
+-- Roles on the external cluster are live logins there; dropping the column would forget them.
+LOCK TABLE datatable_role;
+DO $$
+BEGIN
+ IF EXISTS (SELECT 1 FROM datatable_role WHERE cluster <> 'instance') THEN
+ RAISE EXCEPTION 'datatable_role holds roles on the external instance cluster. Delete them in instance settings first.';
+ END IF;
+ -- Before this, only data tables on Windmill's own cluster could be under roles, and a role
+ -- block left with just `admin` survives deleting every external role.
+ IF EXISTS (
+ SELECT 1 FROM workspace_settings ws,
+ jsonb_each(CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
+ THEN ws.datatable->'datatables' ELSE '{}'::jsonb END) dt
+ WHERE dt.value->'database'->>'resource_type' = 'external_instance'
+ AND dt.value ? 'permissions'
+ ) THEN
+ RAISE EXCEPTION 'external instance data tables are still under roles. Turn their roles off first.';
+ END IF;
+END $$;
+ALTER TABLE datatable_role DROP CONSTRAINT datatable_role_cluster_name_key;
+ALTER TABLE datatable_role ADD CONSTRAINT datatable_role_name_key UNIQUE (name);
+ALTER TABLE datatable_role DROP COLUMN cluster;
diff --git a/backend/migrations/20260917133824_datatable_role_cluster.up.sql b/backend/migrations/20260917133824_datatable_role_cluster.up.sql
new file mode 100644
index 0000000000..d63f06f1f4
--- /dev/null
+++ b/backend/migrations/20260917133824_datatable_role_cluster.up.sql
@@ -0,0 +1,8 @@
+-- A data table role is a Postgres login on one cluster: Windmill's own ('instance'), or the external
+-- instance cluster ('external_instance'). Role names are the cluster's own key, so they are unique
+-- per cluster rather than across the instance.
+ALTER TABLE datatable_role
+ ADD COLUMN cluster VARCHAR(20) NOT NULL DEFAULT 'instance'
+ CHECK (cluster IN ('instance', 'external_instance'));
+ALTER TABLE datatable_role DROP CONSTRAINT datatable_role_name_key;
+ALTER TABLE datatable_role ADD CONSTRAINT datatable_role_cluster_name_key UNIQUE (cluster, name);
diff --git a/backend/tests/instance_config.rs b/backend/tests/instance_config.rs
index 26a46d272e..52ebbee7e7 100644
--- a/backend/tests/instance_config.rs
+++ b/backend/tests/instance_config.rs
@@ -1609,6 +1609,52 @@ async fn declarative_sync_rejects_an_unusable_default_allowed_origins(db: Pool) {
+ clear_settings_and_configs(&db).await;
+ let cluster = |host: &str, password: &str| serde_json::json!({ "host": host, "port": 5432, "user": "wm_admin", "password": password });
+ sqlx::query(
+ "INSERT INTO global_settings (name, value) VALUES
+ ('external_instance_pg', $1),
+ ('external_instance_pg_state', '{\"databases\": {\"dt_a\": {\"success\": true}}}')",
+ )
+ .bind(cluster("pg-a.internal", "one"))
+ .execute(&db)
+ .await
+ .unwrap();
+ let current = BTreeMap::from([(
+ "external_instance_pg".to_string(),
+ cluster("pg-a.internal", "one"),
+ )]);
+ let sync = |value: serde_json::Value| {
+ let desired = BTreeMap::from([("external_instance_pg".to_string(), value)]);
+ let (db, current) = (db.clone(), current.clone());
+ async move {
+ windmill_common::instance_config::sync_global_settings_declarative(
+ &db, ¤t, &desired,
+ )
+ .await
+ }
+ };
+
+ let err = sync(cluster("pg-b.internal", "one"))
+ .await
+ .expect_err("another host must be refused while dt_a is registered");
+ assert!(err.to_string().contains("dt_a"), "got: {err}");
+ assert_eq!(
+ get_global_setting(&db, "external_instance_pg").await,
+ Some(cluster("pg-a.internal", "one"))
+ );
+
+ sync(cluster("PG-A.internal ", "two"))
+ .await
+ .expect("a new login on the same cluster must sync");
+}
+
/// The accent color is interpolated into a stylesheet every user loads, so the operator
/// path must refuse anything but `#rrggbb` just like the settings API does.
#[sqlx::test(fixtures("base"))]
diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs
index 4948a0098a..ddb94f8c2b 100644
--- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs
+++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs
@@ -391,7 +391,11 @@ async fn concurrent_role_creations_both_survive(db: Pool) -> anyhow::R
assert_eq!(a.0, 200, "{}", a.1);
assert_eq!(b.0, 200, "{}", b.1);
- let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?;
+ let catalog = windmill_common::datatable_roles::read_role_catalog(
+ &db,
+ windmill_common::datatable_roles::DatatableRoleCluster::Instance,
+ )
+ .await?;
let recorded: Vec<&str> = catalog.values().map(|r| r.name.as_str()).collect();
for name in &names {
assert!(
@@ -460,7 +464,11 @@ async fn a_role_delete_that_fails_part_way_leaves_the_role_disabled(
let body = resp.text().await?;
assert_eq!(status, 400, "{body}");
- let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?;
+ let catalog = windmill_common::datatable_roles::read_role_catalog(
+ &db,
+ windmill_common::datatable_roles::DatatableRoleCluster::Instance,
+ )
+ .await?;
let role = catalog
.get(&id)
.expect("a failed delete keeps the entry to retry");
diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs
index ca93d3c682..71cbcc8460 100644
--- a/backend/windmill-api-settings/src/lib.rs
+++ b/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;
@@ -61,6 +60,7 @@ use windmill_common::{
ACCENT_COLOR_SETTING, 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,
+ 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,
@@ -181,6 +181,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),
@@ -853,6 +869,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 {
@@ -1248,7 +1272,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(¤t_map, &desired_map, ApplyMode::Merge);
let ai_config_changed = settings_diff
.upserts
@@ -1277,8 +1301,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
@@ -1643,6 +1674,8 @@ struct CustomInstanceDb {
tag: Option,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
used_by_workspaces: Vec,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ workspace_id: Option,
}
#[derive(Deserialize, Debug, Serialize, Default)]
@@ -1685,7 +1718,40 @@ 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? {
+ // A fork copy's name gives away the workspace it was reserved for, so every pending fork on
+ // the instance would be listed. Kept for members of that workspace, and wherever the
+ // caller's workspaces use it, e.g. the fork it was finalized into.
+ let reserved_visible: BTreeSet = sqlx::query_scalar(
+ r#"SELECT e.k FROM global_settings gs
+ CROSS JOIN LATERAL jsonb_each(gs.value->'databases') AS e(k, v)
+ WHERE gs.name = 'custom_instance_pg_databases' AND e.v->>'workspace_id' IS NOT NULL
+ AND (EXISTS (SELECT 1 FROM usr WHERE usr.email = $1
+ AND usr.workspace_id = e.v->>'workspace_id')
+ OR EXISTS (SELECT 1 FROM usr JOIN workspace_settings ws
+ ON ws.workspace_id = usr.workspace_id
+ CROSS JOIN LATERAL jsonb_each(
+ CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
+ THEN ws.datatable->'datatables' ELSE '{}'::jsonb END) dt
+ WHERE usr.email = $1
+ AND dt.value->'database'->>'resource_type' = 'instance'
+ AND dt.value->'database'->>'resource_path' = e.k))"#,
+ )
+ .bind(&authed.email)
+ .fetch_all(&db)
+ .await?
+ .into_iter()
+ .collect();
+ result.retain(|dbname, entry| {
+ entry.workspace_id.is_none() || reserved_visible.contains(dbname)
+ });
+ // Which workspace reserved a copy is still only for superadmins.
+ 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.
@@ -1741,6 +1807,139 @@ async fn refresh_custom_instance_user_pwd(
Ok(Json(()))
}
+async fn get_external_instance_pg_status(
+ authed: ApiAuthed,
+ Extension(db): Extension,
+) -> JsonResult {
+ 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,
+ Json(body): Json,
+) -> JsonResult {
+ 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,
+}
+
+async fn list_external_instance_pg_databases(
+ authed: ApiAuthed,
+ Extension(db): Extension,
+) -> JsonResult> {
+ 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,
+ Path(dbname): Path,
+ Json(body): Json,
+) -> JsonResult<()> {
+ require_super_admin(&db, &authed).await?;
+ let tag = body.tag.as_deref().unwrap_or("datatable");
+ let mut tx = db.begin().await?;
+ windmill_common::external_instance_pg::create_external_instance_database_unchecked(
+ &db, &mut tx, &dbname, tag, None,
+ )
+ .await?;
+ tx.commit().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,
+ Path(dbname): Path,
+) -> 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.
+ let mut tx = db.begin().await?;
+ windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
+ &mut tx, &dbname, None,
+ )
+ .await?;
+ tx.commit().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,
@@ -1752,18 +1951,46 @@ async fn setup_custom_instance_pg_database(
Path(dbname): Path,
Json(body): Json,
) -> JsonResult {
+ // 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?;
+ windmill_common::workspaces::ensure_instance_pg_available(&db).await?;
+ // Fork cleanup checks and drops the database and its entry under this lock. Held from before
+ // the setup creates the database to after its entry is written, neither lands on the other's
+ // half-done state: a dropped database with its entry written back, or the reverse.
+ let mut tx = db.begin().await?;
+ windmill_common::datatable_roles::lock_instance_databases_governance(&mut tx, [dbname.trim()])
+ .await?;
let mut logs = CustomInstanceDbLogs::default();
let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
let success = result.is_ok();
let error = result.err().map(|e| e.to_string());
- let status =
- CustomInstanceDb { logs, success, error, tag: body.tag, used_by_workspaces: vec![] };
+ let status = CustomInstanceDb {
+ logs,
+ success,
+ error,
+ tag: body.tag,
+ used_by_workspaces: vec![],
+ 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(&mut *tx)
+ .await?;
+ tx.commit().await?;
+ let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;
Ok(Json(status))
}
diff --git a/backend/windmill-api-workspaces/src/datatable_acl.rs b/backend/windmill-api-workspaces/src/datatable_acl.rs
index 04dc3d4e80..ebab6214ce 100644
--- a/backend/windmill-api-workspaces/src/datatable_acl.rs
+++ b/backend/windmill-api-workspaces/src/datatable_acl.rs
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
-//! Ownership and grants on the objects of an instance data table.
+//! Ownership and grants on the objects of a data table on a cluster Windmill manages.
//!
//! [`datatable_permissions`](crate::datatable_permissions) decides who may connect as which role;
//! this decides what each role may then touch. Every change is a real `GRANT`, `REVOKE`,
@@ -33,7 +33,7 @@ use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::datatable_roles::{
lock_role_catalog, quote_ident, read_role_catalog, read_role_catalog_tx, DatatableRoleCatalog,
- ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER,
+ DatatableRoleCluster, ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER,
};
use windmill_common::error::{pg_error_message, Error, JsonResult, Result};
use windmill_common::workspaces::{resolve_governing_datatable, DataTable, GoverningDatatable};
@@ -297,22 +297,22 @@ fn role_names(catalog: &DatatableRoleCatalog) -> Vec {
names
}
-fn ensure_instance(governing: &GoverningDatatable) -> Result<()> {
- if governing.is_instance() {
- return Ok(());
- }
- Err(Error::BadRequest(format!(
- "Data table '{}' is backed by a Postgres resource, so its access is managed on that \
- server directly. Only a data table on the Windmill instance's own database has data \
- table roles to grant to.",
- governing.name
- )))
+/// The cluster whose roles the data table's grants name.
+fn ensure_managed(governing: &GoverningDatatable) -> Result {
+ governing.role_cluster().ok_or_else(|| {
+ Error::BadRequest(format!(
+ "Data table '{}' is backed by a Postgres resource, so its access is managed on that \
+ server directly. Only a data table on a database Windmill manages has data table \
+ roles to grant to.",
+ governing.name
+ ))
+ })
}
/// The data table's `admin` connection, and the notices Postgres sends on it.
///
-/// Authorization: connects as `custom_instance_user` with the instance's own credentials and checks
-/// nothing. Callers MUST have authorized the request first — a request about to be refused must
+/// Authorization: connects as `custom_instance_user` with the cluster's stored credentials and
+/// checks nothing. Callers MUST have authorized the request first — a request about to be refused must
/// not get as far as this connection.
async fn connect_as_admin_unchecked(
db: &DB,
@@ -322,21 +322,33 @@ async fn connect_as_admin_unchecked(
mpsc::UnboundedReceiver,
String,
)> {
- ensure_instance(governing)?;
+ let cluster = ensure_managed(governing)?;
// Built from the authorized entry, never by resolving the settings again: a save in between
// could point the entry at a resource on another server and back, and this connection would
// then alter a database the later checks of the entry never see.
- let mut pg = PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
- pg.dbname = governing
+ let dbname = governing
.datatable
.database
.as_ref()
.expect("a governing entry owns a database")
.resource_path
.clone();
- pg.user = Some(CUSTOM_INSTANCE_USER.to_string());
- pg.password = Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
- let dbname = pg.dbname.clone();
+ let pg = match cluster {
+ DatatableRoleCluster::Instance => {
+ let mut pg =
+ PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
+ pg.dbname = dbname.clone();
+ pg.user = Some(CUSTOM_INSTANCE_USER.to_string());
+ pg.password = Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
+ pg
+ }
+ DatatableRoleCluster::ExternalInstance => {
+ windmill_common::external_instance_pg::external_instance_connection_unchecked(
+ db, &dbname, false,
+ )
+ .await?
+ }
+ };
let (client, notices) = connect_with_notices(db, &pg).await?;
Ok((client, notices, dbname))
}
@@ -1085,12 +1097,12 @@ async fn get_datatable_acl(
let target: AclTarget = query.try_into()?;
let governing = resolve_governing_datatable(&db, &w_id, &datatable_name).await?;
ensure_reaches_governing_datatable(&db, &w_id, &datatable_name, &governing, &authed).await?;
- ensure_instance(&governing)?;
+ let cluster = ensure_managed(&governing)?;
let editable = ensure_governs_datatable(&db, &authed, &w_id, &governing)
.await
.is_ok();
let roles = if editable {
- role_names(&read_role_catalog(&db).await?)
+ role_names(&read_role_catalog(&db, cluster).await?)
} else {
vec![]
};
@@ -1376,7 +1388,7 @@ async fn authorize_acl_change(
) -> Result {
let governing = resolve_governing_datatable(db, w_id, datatable_name).await?;
ensure_governs_datatable(db, authed, w_id, &governing).await?;
- ensure_instance(&governing)?;
+ ensure_managed(&governing)?;
Ok(governing)
}
@@ -1386,8 +1398,14 @@ static APPLY_SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1)
/// provisioned before data table roles gave `custom_instance_user` none. Adds that option to its
/// database and `public` privileges, and nothing else: default privileges are left alone, since a
/// schema's change of owner is planned against them. Best-effort, as a grant it fails to enable is
-/// refused when it runs.
-async fn ensure_grant_options(client: &tokio_postgres::Client, db: &DB, dbname: &str) {
+/// refused when it runs. An external instance database was created with the options, so one
+/// missing there is someone's deliberate revoke and is left alone.
+async fn ensure_grant_options(
+ client: &tokio_postgres::Client,
+ db: &DB,
+ cluster: DatatableRoleCluster,
+ dbname: &str,
+) {
let held = client
.query_one(
"SELECT has_database_privilege(current_database(), 'CONNECT WITH GRANT OPTION')
@@ -1399,7 +1417,7 @@ async fn ensure_grant_options(client: &tokio_postgres::Client, db: &DB, dbname:
)
.await
.is_ok_and(|row| row.get::<_, bool>(0));
- if held {
+ if held || cluster != DatatableRoleCluster::Instance {
return;
}
if let Err(e) = grant_options_as_server(db, dbname).await {
@@ -1625,7 +1643,8 @@ async fn plan_datatable_acl(
) -> JsonResult {
crate::datatable_acl_oss::ensure_datatable_acl_available()?;
let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
- let catalog = read_role_catalog(&db).await?;
+ let cluster = ensure_managed(&governing)?;
+ let catalog = read_role_catalog(&db, cluster).await?;
let (client, _notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
Ok(Json(
build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?,
@@ -1649,6 +1668,7 @@ async fn apply_datatable_acl(
// connection could wait forever on a pool that concurrent applies, queued on the same locks,
// have exhausted.
let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
+ let cluster = ensure_managed(&governing)?;
// Applies queue on an instance-wide lock while each holds a direct connection to the instance's
// Postgres; unbounded, the queue alone could exhaust its connection limit. One at a time per
// server, and the ones waiting hold no connection at all.
@@ -1657,7 +1677,7 @@ async fn apply_datatable_acl(
.await
.map_err(|e| Error::internal_err(format!("ACL apply slot closed: {e}")))?;
let (mut client, mut notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
- ensure_grant_options(&client, &db, &dbname).await;
+ ensure_grant_options(&client, &db, cluster, &dbname).await;
// Held until the change is committed: a role renamed or dropped meanwhile would change what
// the plan names, and a settings save could move the entry onto another database. Taken in the
@@ -1672,7 +1692,7 @@ async fn apply_datatable_acl(
.fetch_optional(&mut *tx)
.await?
.flatten();
- let catalog = read_role_catalog_tx(&mut tx).await?;
+ let catalog = read_role_catalog_tx(&mut tx, cluster).await?;
let plan = build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?;
if !entry_unchanged(&governing, entry_now) || &plan.statements != confirmed {
diff --git a/backend/windmill-api-workspaces/src/datatable_clone.rs b/backend/windmill-api-workspaces/src/datatable_clone.rs
index a57a2216e5..6083b7b619 100644
--- a/backend/windmill-api-workspaces/src/datatable_clone.rs
+++ b/backend/windmill-api-workspaces/src/datatable_clone.rs
@@ -11,8 +11,8 @@
//! The fork request makes its copies before it writes the fork: each is created, filled and — for a
//! data table under roles — given the source's owners and grants. What each copy was made from
//! stays in the request ([`MadeCopy`]), so the fork's transaction checks it against the source as it
-//! is then, and a fork that fails drops the instance copies it made ([`drop_copies_after`]) and names
-//! the others, which live on servers of the workspace's own.
+//! is then, and a fork that fails drops the copies it made on a cluster Windmill manages
+//! ([`drop_copies_after`]) and names the others, which live on servers of the workspace's own.
use std::collections::BTreeSet;
@@ -22,8 +22,8 @@ use windmill_common::error::{pg_error_message, Error, Result};
use windmill_common::utils::require_admin;
use windmill_common::worker::CLOUD_HOSTED;
use windmill_common::workspaces::{
- get_datatable_resource_from_db_unchecked, DataTableDatabase, DataTableForkBehavior,
- GoverningDatatable,
+ get_datatable_resource_from_db_unchecked, DataTableCatalogResourceType, DataTableDatabase,
+ DataTableForkBehavior, GoverningDatatable,
};
use windmill_common::{PgDatabase, DB};
@@ -312,27 +312,38 @@ pub(crate) async fn drop_copies_after(db: &DB, copies: Vec, error: Err
}
async fn drop_copy(db: &DB, source_database: &DataTableDatabase, dbname: &str) -> Result<()> {
- if source_database.resource_type
- == windmill_common::workspaces::DataTableCatalogResourceType::Instance
- {
- match windmill_common::drop_unused_instance_database(db, dbname).await? {
- windmill_common::Cleanup::Dropped => Ok(()),
- windmill_common::Cleanup::InUse(users) => Err(Error::BadRequest(format!(
- "kept, since workspaces {} now use it",
- users.join(", ")
- ))),
- windmill_common::Cleanup::Waiter(pid) => Err(Error::BadRequest(format!(
- "kept, since a request (pid {pid}) is still waiting to name it"
- ))),
+ match source_database.resource_type {
+ DataTableCatalogResourceType::Instance => {
+ match windmill_common::drop_unused_instance_database(db, dbname).await? {
+ windmill_common::Cleanup::Dropped => Ok(()),
+ windmill_common::Cleanup::InUse(users) => Err(Error::BadRequest(format!(
+ "kept, since workspaces {} now use it",
+ users.join(", ")
+ ))),
+ windmill_common::Cleanup::Waiter(pid) => Err(Error::BadRequest(format!(
+ "kept, since a request (pid {pid}) is still waiting to name it"
+ ))),
+ }
}
- } else {
+ // Refused, and the copy kept, when anything names it by then.
+ DataTableCatalogResourceType::ExternalInstance => {
+ let mut tx = db.begin().await?;
+ windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
+ &mut tx, dbname, None,
+ )
+ .await?;
+ tx.commit().await?;
+ Ok(())
+ }
+ DataTableCatalogResourceType::Postgresql => {
// On a server of the workspace's own, where a resource edited meanwhile can already name it
// and nothing locks such an edit: dropping it could take someone's data.
- Err(Error::BadRequest(
- "kept on its PostgreSQL server, where it may already be in use; drop it there once it \
- is not, before forking the same data table under this id again"
- .to_string(),
- ))
+ Err(Error::BadRequest(
+ "kept on its PostgreSQL server, where it may already be in use; drop it there once \
+ it is not, before forking the same data table under this id again"
+ .to_string(),
+ ))
+ }
}
}
@@ -349,10 +360,10 @@ async fn make_copy(
request.name
))
})?;
- let is_instance = governing.is_instance();
+ let managed = source_database.resource_type.is_windmill_managed();
// Resolved from the snapshot, not read again: the connection the copy is made on is then
// exactly what these rows describe, which the fork's own clone of them is checked against.
- let (server, connection): (PgDatabase, Option) = if is_instance {
+ let (server, connection): (PgDatabase, Option) = if managed {
let server = serde_json::from_value(
get_datatable_resource_from_db_unchecked(db, parent_w_id, request.name).await?,
)
@@ -383,22 +394,44 @@ async fn make_copy(
// Dumped before the database exists, so a source that cannot be read leaves nothing behind.
// Ownership never carries over: the restore runs as the target's connection user. Grants do,
- // except on the instance, where the replay below is what reproduces them.
+ // except on a cluster Windmill manages, which plants its own and where the replay below is what
+ // reproduces the roles'.
let dump = pg_dump_database(
&server,
PgDumpOptions {
schema_only: request.behavior == DataTableForkBehavior::SchemaOnly,
no_owner: true,
- no_acl: is_instance,
+ no_acl: managed,
..Default::default()
},
)
.await?;
- if is_instance {
- windmill_common::create_custom_instance_database(db, request.dbname, "datatable").await?;
- } else {
- create_database_on_server(db, &server, request.dbname).await?;
+ match source_database.resource_type {
+ DataTableCatalogResourceType::Instance => {
+ windmill_common::create_custom_instance_database(
+ db,
+ request.dbname,
+ "datatable",
+ Some(parent_w_id),
+ )
+ .await?
+ }
+ DataTableCatalogResourceType::ExternalInstance => {
+ let mut tx = db.begin().await?;
+ windmill_common::external_instance_pg::create_external_instance_database_unchecked(
+ db,
+ &mut tx,
+ request.dbname,
+ "datatable",
+ Some(parent_w_id),
+ )
+ .await?;
+ tx.commit().await?;
+ }
+ DataTableCatalogResourceType::Postgresql => {
+ create_database_on_server(db, &server, request.dbname).await?
+ }
}
let target = PgDatabase { dbname: request.dbname.to_string(), ..server.clone() };
@@ -481,11 +514,16 @@ async fn fill(
let replayed = now.datatable.permissions.is_some();
if replayed {
crate::datatable_replay_oss::ensure_replay()?;
- let catalog = read_role_catalog_tx(&mut tx).await?;
+ // The copy is on the source's cluster, so its roles come from that cluster's catalog.
+ let cluster = governing
+ .role_cluster()
+ .ok_or_else(|| Error::internal_err("a replayed clone is on a managed database"))?;
+ let catalog = read_role_catalog_tx(&mut tx, cluster).await?;
// The replay leaves `CONNECT` to the catalog, and creating the database only tried to set
// it: a copy `PUBLIC` could still connect to would admit logins the source turns away.
windmill_common::datatable_roles::converge_connect_grants_with(
db,
+ cluster,
&target.dbname,
&catalog,
)
diff --git a/backend/windmill-api-workspaces/src/datatable_migrations.rs b/backend/windmill-api-workspaces/src/datatable_migrations.rs
index df1fac203b..f762fda175 100644
--- a/backend/windmill-api-workspaces/src/datatable_migrations.rs
+++ b/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,
};
@@ -1669,7 +1669,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 {
diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs
index 185fe17b25..458e63a973 100644
--- a/backend/windmill-api-workspaces/src/workspaces.rs
+++ b/backend/windmill-api-workspaces/src/workspaces.rs
@@ -2300,7 +2300,8 @@ struct DataTableTables {
schemas: TableListMap,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option,
- /// On the instance database: the only kind that can be under roles or have its access edited.
+ /// On a database Windmill manages, on its own cluster or the external one: the only kinds that
+ /// can be under roles or have their access edited.
instance: bool,
permissioned: bool,
/// The roles this caller may connect as, by name; empty when not under roles.
@@ -2557,7 +2558,7 @@ async fn list_one_datatable_tables(
};
let result: Result<()> = async {
let governing = resolve_governing_datatable(db, w_id, &entry.datatable_name).await?;
- entry.instance = governing.is_instance();
+ entry.instance = governing.role_cluster().is_some();
let usable =
crate::datatable_permissions_oss::usable_datatable_roles(db, authed, w_id, &governing)
.await?;
@@ -3082,6 +3083,25 @@ pub(crate) async fn resolve_pg_source_checked(
w_id: &str,
source: &str,
) -> Result {
+ Ok(
+ resolve_pg_source_checked_with_kind(db, user_db, authed, w_id, source)
+ .await?
+ .0,
+ )
+}
+
+/// [`resolve_pg_source_checked`], also reporting the kind of Windmill-managed database behind the
+/// source, `None` for a user resource. Callers that guard a managed connection MUST take the kind
+/// from here rather than ask separately: between two reads a save can flip the entry, leaving the guards of one kind
+/// applied to the connection of the other.
+pub(crate) async fn resolve_pg_source_checked_with_kind(
+ db: &DB,
+ user_db: &UserDB,
+ authed: &ApiAuthed,
+ w_id: &str,
+ source: &str,
+) -> Result<(PgDatabase, Option)> {
+ let mut managed_kind = None;
let db_resource = if let Some(name) = source.strip_prefix("datatable://") {
windmill_common::workspaces::ensure_datatable_admin_access(
db,
@@ -3090,7 +3110,13 @@ pub(crate) async fn resolve_pg_source_checked(
&DatatableAccess::Authed(authed.to_authed_ref()),
)
.await?;
- get_datatable_resource_from_db_unchecked(db, w_id, name).await?
+ let (connection, kind) =
+ windmill_common::workspaces::get_datatable_connection_and_kind_unchecked(
+ db, w_id, name,
+ )
+ .await?;
+ managed_kind = kind;
+ connection
} else if let Some(path) = source.strip_prefix("$res:") {
let db_with_authed = windmill_common::db::DbWithOptAuthed::from_authed(
authed,
@@ -3123,27 +3149,37 @@ pub(crate) async fn resolve_pg_source_checked(
)));
};
- serde_json::from_value(db_resource)
- .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))
+ let pg: PgDatabase = serde_json::from_value(db_resource)
+ .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;
+ Ok((pg, managed_kind))
}
-/// 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 {
+/// 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> {
// 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://` / `$res:` form the import endpoints take.
-async fn is_instance_datatable_source(db: &DB, w_id: &str, source: &str) -> Result {
+async fn managed_datatable_source_kind(
+ db: &DB,
+ w_id: &str,
+ source: &str,
+) -> Result> {
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),
}
}
@@ -3242,13 +3278,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);
- }
- if let Some(options) = pg_db.non_empty_options() {
- cmd.env("PGOPTIONS", options);
- }
+ let _root_cert = apply_pg_tls_env(&mut cmd, pg_db)?;
let output = cmd
.output()
@@ -3366,7 +3396,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)> {
let mut cmd = tokio::process::Command::new("psql");
cmd.arg("--host")
.arg(&pg_db.host)
@@ -3383,13 +3413,92 @@ 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> {
if let Some(ref sslmode) = pg_db.sslmode {
cmd.env("PGSSLMODE", sslmode);
}
if let Some(options) = pg_db.non_empty_options() {
cmd.env("PGOPTIONS", options);
}
- 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 {
+ 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,
+ options: 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.
@@ -3399,7 +3508,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> {
- 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")
@@ -3436,7 +3546,8 @@ pub(crate) async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile)
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")
@@ -3494,9 +3605,39 @@ async fn create_pg_database(
}
}
- if is_instance_datatable_source(&db, &w_id, &req.source).await? {
- windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable")
+ if let Some(source_kind) = managed_datatable_source_kind(&db, &w_id, &req.source).await? {
+ // Held until the copy is registered, as a rename migrates reservations to the new id under
+ // it once the old one is archived: a copy registered after that would be reserved for an
+ // id nothing answers on.
+ let mut tx = db.begin().await?;
+ windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
+ let live = sqlx::query_scalar::<_, bool>("SELECT NOT deleted FROM workspace WHERE id = $1")
+ .bind(&w_id)
+ .fetch_optional(&mut *tx)
+ .await?
+ .unwrap_or(false);
+ if !live {
+ return Err(Error::BadRequest(format!("Workspace '{w_id}' is archived")));
+ }
+ if source_kind == DataTableCatalogResourceType::ExternalInstance {
+ windmill_common::external_instance_pg::create_external_instance_database_unchecked(
+ &db,
+ &mut tx,
+ &req.target_dbname,
+ "datatable",
+ Some(&w_id),
+ )
.await?;
+ } else {
+ windmill_common::create_custom_instance_database(
+ &db,
+ &req.target_dbname,
+ "datatable",
+ Some(&w_id),
+ )
+ .await?;
+ }
+ tx.commit().await?;
} else {
let source_pg =
resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
@@ -3583,12 +3724,12 @@ pub(crate) async fn ensure_datatable_is_clonable(
let governing = resolve_governing_datatable(db, w_id, name).await?;
// 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."
@@ -3642,9 +3783,11 @@ async fn import_pg_database(
}
let schema_only = req.fork_behavior == DataTableForkBehavior::SchemaOnly;
- 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?;
+ let mut fork_lock: Option> = None;
+ let (source_pg, source_kind) =
+ resolve_pg_source_checked_with_kind(&db, &user_db, &authed, &w_id, &req.source).await?;
+ let (mut target_pg, target_kind) =
+ resolve_pg_source_checked_with_kind(&db, &user_db, &authed, &w_id, &req.target).await?;
if let Some(ref override_dbname) = req.target_dbname_override {
if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
@@ -3654,6 +3797,29 @@ async fn import_pg_database(
.to_string(),
));
}
+ // The kind the connection above was built from, never a second read: an entry flipped
+ // to a resource between the two would keep the managed connection here and lose the
+ // check that decides which copy it may fill.
+ if let Some(kind) = target_kind {
+ // Held until the restore is done: fork finalization takes the first, and every
+ // save newly naming a database, in any workspace, the second. Nothing may start
+ // using 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::datatable_roles::lock_instance_databases_governance(
+ &mut tx,
+ [override_dbname.as_str()],
+ )
+ .await?;
+ windmill_common::ensure_fork_database_available_to(
+ &mut tx,
+ kind,
+ override_dbname,
+ &w_id,
+ )
+ .await?;
+ fork_lock = Some(tx);
+ }
}
target_pg.dbname = override_dbname.clone();
}
@@ -3663,8 +3829,7 @@ 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 = target_kind.is_some() || source_kind.is_some();
let dump_file = pg_dump_database(
&source_pg,
@@ -3672,6 +3837,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 '{}'",
@@ -3772,39 +3940,76 @@ async fn edit_ducklake_config(
)
.await?;
- let old_ducklakes = sqlx::query_scalar!(
- r#"
- SELECT ws.ducklake->'ducklakes' AS ducklake_name
- FROM workspace_settings ws
- WHERE ws.workspace_id = $1
- "#,
- &w_id
+ // Under the row lock the save writes with, taken before the database locks below as fork
+ // cleanup takes the two.
+ let old_ducklakes = sqlx::query_scalar::<_, Option>(
+ "SELECT ws.ducklake->'ducklakes' FROM workspace_settings ws
+ WHERE ws.workspace_id = $1 FOR UPDATE",
)
+ .bind(&w_id)
.fetch_one(&mut *tx)
.await?
.unwrap_or(serde_json::Value::Null);
let old_ducklakes: HashMap =
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;
+ }
+ // 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 ducklakes with Instance databases"
+ .to_string(),
+ ));
+ }
+ 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?;
+ } else {
+ windmill_common::workspaces::ensure_instance_pg_available(&mut *tx).await?;
}
}
+ // Fork cleanup decides nothing uses an instance database under this lock, so a catalog newly
+ // put on one must not commit between its check and its drop.
+ windmill_common::datatable_roles::lock_instance_databases_governance(
+ &mut *tx,
+ new_config
+ .settings
+ .ducklakes
+ .iter()
+ .filter(|(name, dl)| {
+ dl.catalog.resource_type == DucklakeCatalogResourceType::Instance
+ && old_ducklakes.get(name.as_str()).is_none_or(|old| {
+ old.catalog.resource_type != DucklakeCatalogResourceType::Instance
+ || old.catalog.resource_path != dl.catalog.resource_path
+ })
+ })
+ .map(|(_, dl)| dl.catalog.resource_path.as_str()),
+ )
+ .await?;
+
let config: serde_json::Value = serde_json::to_value(&new_config.settings)
.map_err(|err| Error::internal_err(err.to_string()))?;
@@ -3860,6 +4065,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?;
windmill_common::lock_instance_databases(
&mut tx,
new_config.settings.datatables.values().filter_map(|dt| {
@@ -3983,6 +4190,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",
});
@@ -4018,18 +4226,20 @@ async fn edit_datatable_config(
)));
}
// Carrying the block onto a resource-backed entry would produce a data table the chokepoint
- // refuses on every job — a save that succeeds and breaks everything afterwards. Refuse it
- // instead: turning roles off first is one step, and it keeps discarding an access decision
- // something somebody chose rather than a side effect of moving a database.
+ // refuses on every job — a save that succeeds and breaks everything afterwards — and onto
+ // the other managed cluster, one whose role ids name nothing in that cluster's catalog.
+ // Refuse it instead: turning roles off first is one step, and it keeps discarding an access
+ // decision something somebody chose rather than a side effect of moving a database.
+ let old_kind = old.and_then(|old| old.database.as_ref()).map(|d| d.resource_type);
if dt.permissions.is_some()
&& dt
.database
.as_ref()
- .is_some_and(|d| d.resource_type != DataTableCatalogResourceType::Instance)
+ .is_some_and(|d| Some(d.resource_type) != old_kind)
{
return Err(Error::BadRequest(format!(
- "Data table '{name}' is under roles, which only a data table on the instance \
- database can be. Turn its roles off before moving it to a PostgreSQL resource."
+ "Data table '{name}' is under roles, which belong to the cluster its database is \
+ on. Turn its roles off before moving it to another kind of database."
)));
}
// A pointer names no database of its own, so the form's empty `database` is correct there.
@@ -4054,35 +4264,52 @@ 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.
+ // 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.
//
// Compared against the entry the carried fields came from, not the one stored under the same
// name: otherwise swapping two names keeps each database in place while moving a clone's
// `governed_by` off the copy it governs.
- if !is_superadmin {
- for (name, dt) in new_config.settings.datatables.iter() {
- let old_dt = old_datatables.get(
+ 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(
rename_src
.get(name.as_str())
.copied()
.unwrap_or(name.as_str()),
- );
- 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(),
- ));
- }
- }
+ )
+ .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;
+ }
+ // 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(
+ &mut tx,
+ &database.resource_path,
+ )
+ .await?;
+ } else {
+ windmill_common::workspaces::ensure_instance_pg_available(&mut *tx).await?;
}
}
@@ -4102,7 +4329,7 @@ async fn edit_datatable_config(
// entry through a declared rename alone, and a settings sync never declares one, so an entry
// without roles that newly points at such a database — a name added, or an existing one
// repointed — would answer everyone there as `admin`. That holds whichever workspace governs it.
- let newly_pointed: Vec<(&String, &str)> = new_config
+ let newly_pointed: Vec<(&String, DataTableCatalogResourceType, &str)> = new_config
.settings
.datatables
.iter()
@@ -4112,7 +4339,7 @@ async fn edit_datatable_config(
let db = dt
.database
.as_ref()
- .filter(|d| d.resource_type == DataTableCatalogResourceType::Instance)?;
+ .filter(|d| d.resource_type.is_windmill_managed())?;
let lookup = rename_src
.get(name.as_str())
.copied()
@@ -4124,38 +4351,72 @@ async fn edit_datatable_config(
old_db.resource_type != db.resource_type
|| old_db.resource_path != db.resource_path
});
- repointed.then_some((name, db.resource_path.as_str()))
+ repointed.then_some((name, db.resource_type, db.resource_path.as_str()))
})
.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 = if newly_pointed.is_empty() {
+ let governed_elsewhere: Vec<(String, String)> = if newly_pointed.is_empty() {
vec![]
} else {
- sqlx::query_scalar(
- "SELECT DISTINCT dt.value->'database'->>'resource_path' FROM workspace_settings ws
+ sqlx::query_as(
+ "SELECT DISTINCT dt.value->'database'->>'resource_type',
+ dt.value->'database'->>'resource_path'
+ FROM workspace_settings ws
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
WHERE ws.workspace_id <> $1 AND (dt.value ? 'permissions' OR dt.value ? 'governed_by')
- AND dt.value->'database'->>'resource_type' = 'instance'",
+ AND dt.value->'database'->>'resource_type' IN ('instance', 'external_instance')",
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?
};
- for (name, dbname) in newly_pointed {
+ for (name, kind, dbname) in newly_pointed {
let governed_here = old_datatables.values().any(|old| {
(old.permissions.is_some() || old.governed_by.is_some())
- && old.database.as_ref().is_some_and(|d| {
- d.resource_type == DataTableCatalogResourceType::Instance
- && d.resource_path == dbname
- })
+ && old
+ .database
+ .as_ref()
+ .is_some_and(|d| d.resource_type == kind && d.resource_path == dbname)
});
- if governed_here || governed_elsewhere.iter().any(|g| g == dbname) {
+ if governed_here
+ || governed_elsewhere
+ .iter()
+ .any(|(k, p)| k == kind.as_ref() && p == dbname)
+ {
return Err(Error::BadRequest(format!(
"Data table '{name}' would point at database '{dbname}', which a data table under \
roles uses, without carrying those roles: everyone reaching '{name}' would connect \
@@ -8429,13 +8690,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;
}
@@ -8609,11 +8871,39 @@ 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() {
+ // Held until the fork commits, as every save newly naming a database takes it: none may
+ // claim the copy between the check below and this fork's entry landing on it.
+ windmill_common::datatable_roles::lock_instance_databases_governance(
+ &mut **tx,
+ [fdt.new_dbname.as_str()],
+ )
+ .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(
+ &mut **tx,
+ database.resource_type,
+ &fdt.new_dbname,
+ parent_w_id,
+ )
+ .await?;
+ }
+ 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(
@@ -9184,6 +9474,11 @@ async fn write_workspace_fork(
.bind(&nw.id)
.execute(&mut *tx)
.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 {
// The checks above ran outside a transaction, so the parent's eligibility and the chain's
@@ -9328,6 +9623,32 @@ async fn write_workspace_fork(
)));
}
}
+ // Saves name an external database under the external cluster's lifecycle lock instead.
+ let external_copies: Vec<&str> = copies
+ .iter()
+ .filter(|c| {
+ c.source_database.resource_type == DataTableCatalogResourceType::ExternalInstance
+ })
+ .map(|c| c.dbname.as_str())
+ .collect();
+ if !external_copies.is_empty() {
+ windmill_common::external_instance_pg::lock_external_instance_pg_state(&mut tx).await?;
+ }
+ for dbname in &external_copies {
+ let uses = windmill_common::workspaces::managed_database_uses(
+ &mut tx,
+ DataTableCatalogResourceType::ExternalInstance,
+ dbname,
+ None,
+ )
+ .await?;
+ if !uses.is_empty() {
+ return Err(Error::BadRequest(format!(
+ "Database '{dbname}' copied for this fork is already used by {}; fork again",
+ uses.join(", ")
+ )));
+ }
+ }
// Update forked datatable settings to point to new databases
for fdt in &nw.forked_datatables {
diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs
index 9039bda2ef..8ca98f66d1 100644
--- a/backend/windmill-api-workspaces/src/workspaces_extra.rs
+++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs
@@ -56,6 +56,11 @@ pub(crate) async fn change_workspace_id(
let mut tx = db.begin().await?;
+ // The settings copy below carries every data table entry to the new id, which fork cleanup of
+ // the old id cannot see until this commits: without the lock it could drop a copy the renamed
+ // workspace goes on using. Before the pairing lock, as forking takes the two in that order.
+ windmill_common::workspaces::lock_fork_datatables(&mut tx, &old_id).await?;
+
// A rename rewrites the workspace's dev flag and reparents its children, so it decides on the
// same state the pairing handlers do: without this lock a concurrent create/attach could commit
// an active dev workspace under the shell this rename is about to archive. Both ids, since the
@@ -869,6 +874,10 @@ pub(crate) async fn change_workspace_id(
}
}
+ // After every workspace_settings write above: fork cleanup locks a settings row before the
+ // registry, so taking the registry first here would deadlock with it.
+ migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?;
+
// Audit log in the same transaction as the workspace changes
audit_log(
&mut *tx,
@@ -937,6 +946,14 @@ pub(crate) async fn change_workspace_id(
let (_schedules_count, canceled_count, _deleted_tokens_count) =
archive_workspace_impl(&db, &old_id, &authed.username, None).await?;
+ // The old id stays live between the commit above and the archive, and a fork copy created for
+ // it in that window registers under it. Creation checks the workspace is live under the fork
+ // lock, so once this has run under it, no copy can be reserved for the old id any more.
+ let mut tx = db.begin().await?;
+ windmill_common::workspaces::lock_fork_datatables(&mut tx, &old_id).await?;
+ migrate_fork_reservations(&mut tx, &old_id, &rw.new_id).await?;
+ tx.commit().await?;
+
info!(
"Workspace id change completed: moved {} to {}, archived old workspace",
old_id, rw.new_id
@@ -948,6 +965,50 @@ pub(crate) async fn change_workspace_id(
))
}
+/// A fork copy reserved for the old id would otherwise be unreachable: its creator cannot import
+/// into it or finish its fork under the new id, and nothing else would ever drop it.
+async fn migrate_fork_reservations(
+ tx: &mut Transaction<'_, Postgres>,
+ old_id: &str,
+ new_id: &str,
+) -> Result<()> {
+ const MIGRATE: &str = r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', (
+ SELECT COALESCE(jsonb_object_agg(k, CASE WHEN v->>'workspace_id' = $1
+ THEN jsonb_set(v, '{workspace_id}', to_jsonb($2::text)) ELSE v END), '{}'::jsonb)
+ FROM jsonb_each(COALESCE(value->'databases', '{}'::jsonb)) AS e(k, v)
+ ))
+ WHERE name = $3"#;
+ sqlx::query(MIGRATE)
+ .bind(old_id)
+ .bind(new_id)
+ .bind("custom_instance_pg_databases")
+ .execute(&mut **tx)
+ .await?;
+ // External copies are registered in the cluster state, which its writers rewrite whole under
+ // the lifecycle lock. Only taken when there is something to move, as setup can hold it for as
+ // long as the cluster takes to answer.
+ let state = windmill_common::global_settings::EXTERNAL_INSTANCE_PG_STATE_SETTING;
+ let reserved = sqlx::query_scalar::<_, bool>(
+ "SELECT EXISTS (SELECT 1 FROM global_settings, jsonb_each(value->'databases') AS e(k, v)
+ WHERE name = $1 AND jsonb_typeof(value->'databases') = 'object'
+ AND v->>'workspace_id' = $2)",
+ )
+ .bind(state)
+ .bind(old_id)
+ .fetch_one(&mut **tx)
+ .await?;
+ if reserved {
+ windmill_common::external_instance_pg::lock_external_instance_pg_state(tx).await?;
+ sqlx::query(MIGRATE)
+ .bind(old_id)
+ .bind(new_id)
+ .bind(state)
+ .execute(&mut **tx)
+ .await?;
+ }
+ Ok(())
+}
+
#[derive(Deserialize)]
pub(crate) struct DeleteWorkspaceQuery {
pub(crate) only_delete_forks: Option,
@@ -1430,9 +1491,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,7 +1500,112 @@ pub async fn drop_forked_datatable_databases(
));
continue;
}
- if let Err(e) = windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
+ // The fork's own entry is what is going away; anything else still reaching the copy,
+ // a child fork's pointer at this entry included, keeps it. The lock keeps a child fork
+ // from gaining such a pointer before the drop.
+ // A task of its own, so a client going away cannot stop it between dropping the
+ // database and committing the entry's removal.
+ let dropped = tokio::spawn({
+ let (db, w_id, dt_name, db_to_drop) = (
+ db.clone(),
+ w_id.clone(),
+ dt_name.clone(),
+ db_to_drop.clone(),
+ );
+ let resource_type = database.resource_type;
+ async move {
+ 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?;
+ // The snapshot above was read unlocked: a save committing since could have
+ // repointed this entry, and the entry is removed below whatever it names by then.
+ let current = sqlx::query_scalar::<_, Option>(
+ "SELECT datatable->'datatables'->$2 FROM workspace_settings
+ WHERE workspace_id = $1 FOR UPDATE",
+ )
+ .bind(&w_id)
+ .bind(&dt_name)
+ .fetch_optional(&mut *tx)
+ .await?
+ .flatten()
+ .and_then(|v| serde_json::from_value::