Compare commits

..
Author SHA1 Message Date
Diego ImbertandClaude Opus 5 668c101270 docs(datatables): describe the external instance cluster and how to run one locally
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-18 01:05:19 +02:00
Diego Imbert 51224582c1 Merge commit '2661dadba63efcc0c5db88a7942d66caf34f20b9' into HEAD
# Conflicts:
#	backend/ee-repo-ref.txt
2026-09-18 01:00:23 +02:00
Diego Imbert 3fdb3047a7 Merge commit '5dfb2f30f74cedaae9f113383e42d3a35cde5189' into HEAD
# Conflicts:
#	backend/ee-repo-ref.txt
#	backend/windmill-common/src/datatable_roles.rs
2026-09-17 18:23:38 +02:00
Diego Imbert 4a0d7c0c29 Merge commit 'd62807e9edc553b1eb08c10ce28521a71007a263' into HEAD 2026-09-17 18:04:22 +02:00
Diego ImbertandClaude Opus 5 48f026eb0f fix(datatables): refuse rolling back while external data tables are under roles, and type external_instance in the CLI
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-17 16:35:45 +02:00
Diego Imbert 0e659898f3 Merge commit 'a296ca016a' into HEAD
# Conflicts:
#	backend/ee-repo-ref.txt
2026-09-17 16:35:45 +02:00
Diego Imbert c55ff3141c Merge commit '5eec203099ba1034039f8b52e81400078cac919d' into HEAD 2026-09-17 16:17:18 +02:00
Diego ImbertandClaude Opus 5 0b3882f2b2 fix(datatables): keep certificate verification when DuckDB attaches an external data table
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-17 16:17:02 +02:00
Diego Imbert 58731b1828 Merge commit '53d62dd399' into HEAD
# Conflicts:
#	backend/ee-repo-ref.txt
2026-09-17 16:17:02 +02:00
Diego ImbertandClaude Opus 5 0406f133c5 feat(datatables): per-cluster data table role catalogs, with roles on the external instance cluster
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-17 15:59:22 +02:00
23 changed files with 605 additions and 484 deletions
+4
View File
@@ -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
- **Auth surface**: `docs/auth-surface.md` — credential precedence, session/cache invalidation
scope, which token labels email their owner at expiry, how OAuth login matches `login_type`, and
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
+1 -1
View File
@@ -1 +1 @@
a97b5a5982d67e977dde6222380903f019da39c2
151be033ea2bf2f768f381a8cd3b888bdd366cf9
@@ -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;
@@ -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);
@@ -391,7 +391,11 @@ async fn concurrent_role_creations_both_survive(db: Pool<Postgres>) -> 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");
+18 -26
View File
@@ -42,6 +42,7 @@ 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;
@@ -1719,15 +1720,7 @@ async fn list_custom_instance_pg_databases(
))
})?;
if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
// Which workspace reserved a fork copy is nobody else's business: it would enumerate every
// pending fork on the instance.
for entry in result.values_mut() {
entry.workspace_id = None;
}
return Ok(Json(result));
}
{
if windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
// 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.
@@ -1926,6 +1919,16 @@ async fn setup_custom_instance_pg_database(
// Before anything is recorded: the status written below replaces the registry entry, and with it
// the workspace a fork copy is reserved for.
require_super_admin(&db, &authed).await?;
// A re-run keeps the fork reservation: without it, the workspace the copy was made for could no
// longer import into it or finish its fork.
let workspace_id = sqlx::query_scalar::<_, Option<String>>(
"SELECT value->'databases'->$1->>'workspace_id' FROM global_settings
WHERE name = 'custom_instance_pg_databases'",
)
.bind(&dbname)
.fetch_optional(&db)
.await?
.flatten();
let mut logs = CustomInstanceDbLogs::default();
let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
let success = result.is_ok();
@@ -1936,25 +1939,14 @@ async fn setup_custom_instance_pg_database(
error,
tag: body.tag,
used_by_workspaces: vec![],
workspace_id: None,
workspace_id,
};
let status_json = serde_json::to_value(&status).map_err(to_anyhow)?;
// The fork reservation is carried over inside the write, from whatever the row holds then: a
// rename migrating it while the setup above ran would otherwise be overwritten with the value
// this request started from, stranding the copy under the archived workspace.
let saved = sqlx::query_scalar::<_, serde_json::Value>(
r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}',
COALESCE(value->'databases', '{}'::jsonb)
|| jsonb_build_object($1::text, $2::jsonb || jsonb_build_object(
'workspace_id', value->'databases'->$1::text->'workspace_id')))
WHERE name = 'custom_instance_pg_databases'
RETURNING value->'databases'->$1::text"#,
)
.bind(&dbname)
.bind(&status_json)
.fetch_one(&db)
.await?;
let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;
// 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?;
Ok(Json(status))
}
@@ -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};
@@ -295,22 +295,22 @@ fn role_names(catalog: &DatatableRoleCatalog) -> Vec<String> {
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<DatatableRoleCluster> {
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,
@@ -320,21 +320,33 @@ async fn connect_as_admin_unchecked(
mpsc::UnboundedReceiver<DbError>,
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, mut connection) = pg.connect(Some(db)).await?;
// Unbounded: the driver must never wait on the receiver, which only drains once the statement
// the driver is carrying has completed.
@@ -1020,12 +1032,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![]
};
@@ -1310,7 +1322,7 @@ async fn authorize_acl_change(
) -> Result<GoverningDatatable> {
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)
}
@@ -1320,8 +1332,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')
@@ -1333,7 +1351,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 {
@@ -1559,7 +1577,8 @@ async fn plan_datatable_acl(
) -> JsonResult<AclPlan> {
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?,
@@ -1583,6 +1602,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.
@@ -1591,7 +1611,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
@@ -1606,7 +1626,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 {
@@ -2263,7 +2263,8 @@ struct DataTableTables {
schemas: TableListMap,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
/// 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.
@@ -2520,7 +2521,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?;
@@ -3680,7 +3681,6 @@ async fn import_pg_database(
}
let schema_only = req.fork_behavior == DataTableForkBehavior::SchemaOnly;
let mut fork_lock: Option<Transaction<'_, Postgres>> = None;
let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
let mut target_pg =
resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.target).await?;
@@ -3694,10 +3694,6 @@ async fn import_pg_database(
));
}
if let Some(kind) = managed_datatable_source_kind(&db, &w_id, &req.target).await? {
// Held until the restore is done, as fork finalization takes it: a fork must not
// commit this database while `psql` is still filling it.
let mut tx = db.begin().await?;
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
windmill_common::ensure_fork_database_available_to(
&db,
kind,
@@ -3705,7 +3701,6 @@ async fn import_pg_database(
&w_id,
)
.await?;
fork_lock = Some(tx);
}
}
target_pg.dbname = override_dbname.clone();
@@ -3729,9 +3724,6 @@ 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 '{}'",
@@ -3926,8 +3918,6 @@ async fn edit_datatable_config(
let is_superadmin = require_super_admin(&db, &authed).await.is_ok();
let mut tx = db.begin().await?;
// Ahead of the settings row, as fork cleanup of this workspace takes the two.
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
// Read under the row lock this transaction will write with. `permissions`, `reference` and
// `forked_from` are carried across from what this read returns, so a permissions save
@@ -4061,18 +4051,20 @@ async fn edit_datatable_config(
None => None,
};
// 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.
@@ -4117,14 +4109,6 @@ async fn edit_datatable_config(
if unchanged {
continue;
}
// Before the registration check, whose refusal would otherwise tell a workspace admin
// which databases exist on the cluster.
if !is_superadmin {
return Err(Error::BadRequest(
"Only superadmins can create or modify data tables with Instance databases"
.to_string(),
));
}
if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
windmill_common::external_instance_pg::ensure_external_instance_available()?;
windmill_common::external_instance_pg::ensure_external_instance_database_registered(
@@ -4133,6 +4117,12 @@ async fn edit_datatable_config(
)
.await?;
}
if !is_superadmin {
return Err(Error::BadRequest(
"Only superadmins can create or modify data tables with Instance databases"
.to_string(),
));
}
}
// Worked out from the locked entries rather than taken from `deleted_datatables`: a settings
@@ -4151,7 +4141,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()
@@ -4160,7 +4150,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()
@@ -4172,66 +4162,44 @@ 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. 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();
// without this the scan below could read past its uncommitted write.
windmill_common::datatable_roles::lock_instance_databases_governance(
&mut *tx,
newly_pointed
.iter()
.map(|(_, dbname)| *dbname)
.chain(newly_named.iter().copied()),
newly_pointed.iter().map(|(_, _, dbname)| *dbname),
)
.await?;
let governed_elsewhere: Vec<String> = 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'
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.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 \
@@ -1438,19 +1438,7 @@ pub async fn drop_forked_datatable_databases(
// from gaining such a pointer before the drop.
let dropped = async {
let mut tx = db.begin().await?;
// The three locks a settings save takes, in its order: this workspace's data
// tables, its settings row, and the database itself. Without them a save could
// rename this entry, or point another one here, either side of the check below.
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
sqlx::query("SELECT 1 FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE")
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
windmill_common::datatable_roles::lock_instance_databases_governance(
&mut tx,
[db_to_drop.as_str()],
)
.await?;
if database.resource_type
== windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance
{
@@ -1476,16 +1464,6 @@ pub async fn drop_forked_datatable_databases(
}
windmill_common::drop_custom_instance_database(&db, db_to_drop).await?;
}
// The entry goes with the database: a fork this one is cloned into afterwards must
// not inherit a pointer at a data table whose database is gone.
sqlx::query(
"UPDATE workspace_settings SET datatable = datatable #- ARRAY['datatables', $2]
WHERE workspace_id = $1",
)
.bind(&w_id)
.bind(dt_name)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok::<_, Error>(())
}
+27 -7
View File
@@ -1688,13 +1688,19 @@ paths:
/settings/datatable_roles:
get:
summary: list the instance's data table roles
summary: list the data table roles of one Windmill-managed Postgres cluster
operationId: listInstanceDatatableRoles
tags:
- setting
parameters:
- in: query
name: cluster
required: false
schema:
$ref: "#/components/schemas/DatatableRoleCluster"
responses:
"200":
description: the instance role catalog
description: the cluster's role catalog
content:
application/json:
schema:
@@ -1702,7 +1708,7 @@ paths:
items:
$ref: "#/components/schemas/InstanceDatatableRole"
post:
summary: create a data table role on the instance's Postgres cluster
summary: create a data table role on a Windmill-managed Postgres cluster
operationId: createInstanceDatatableRole
tags:
- setting
@@ -1716,6 +1722,8 @@ paths:
properties:
name:
type: string
cluster:
$ref: "#/components/schemas/DatatableRoleCluster"
responses:
"200":
description: the created role
@@ -33757,9 +33765,17 @@ components:
last_setup:
$ref: "#/components/schemas/ExternalInstancePgSetupReport"
DatatableRoleCluster:
type: string
description: >-
The Windmill-managed Postgres cluster a data table role is a login on: Windmill's own
(behind `instance` data tables) or the external instance cluster (behind
`external_instance` ones). Defaults to `instance`.
enum: [instance, external_instance]
InstanceDatatableRole:
type: object
required: [id, name, enabled]
required: [id, name, enabled, cluster]
properties:
id:
type: string
@@ -33767,6 +33783,8 @@ components:
type: string
enabled:
type: boolean
cluster:
$ref: "#/components/schemas/DatatableRoleCluster"
DatatableRoleTenants:
type: object
@@ -33788,8 +33806,10 @@ components:
supported:
type: boolean
description: >-
Whether this data table can be put under roles at all. Only one backed by the
instance database can: a role is a login on that cluster.
Whether this data table can be put under roles at all. Only one on a database Windmill
manages can: a role is a login on that database's cluster.
cluster:
$ref: "#/components/schemas/DatatableRoleCluster"
permissioned:
type: boolean
default_role:
@@ -34089,7 +34109,7 @@ components:
type: array
items:
type: string
description: Workspaces that reference this database through a ducklake catalog or a datatable database of the kind being listed — 'instance' for the instance databases endpoint, 'external_instance' for the external cluster one. Computed at request time, not persisted, and only returned to superadmins.
description: Workspaces that reference this database via a ducklake catalog or datatable database with resource_type 'instance'. Computed at request time, not persisted.
workspace_id:
type: string
description: The workspace a member created this database for as a fork copy. Only that workspace can import into it or point a fork at it.
+127 -44
View File
@@ -6,22 +6,67 @@
* LICENSE-AGPL for a copy of the license.
*/
//! The instance's data table role catalog.
//! The instance's data table role catalogs.
//!
//! A data table role is a real Postgres login role on the Windmill cluster, named exactly as the
//! user named it, shared by every instance database. Windmill decides who may ask for a role (the
//! per-data-table tenant lists in [`crate::workspaces`]); Postgres decides what the role may then
//! touch. The catalog here is only the first half's vocabulary plus the cluster provisioning.
//! A data table role is a real Postgres login role on one cluster — Windmill's own, or the external
//! instance cluster — named exactly as the user named it, shared by every database Windmill manages
//! on that cluster. Each cluster has its own catalog: a role exists where it was created and nowhere
//! else. Windmill decides who may ask for a role (the per-data-table tenant lists in
//! [`crate::workspaces`]); Postgres decides what the role may then touch. The catalog here is only
//! the first half's vocabulary plus the cluster provisioning.
//!
//! Entries are keyed by a generated id so a rename moves nothing else: tenants name the id.
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use crate::{
error::{Error, Result},
workspaces::DataTableCatalogResourceType,
DB,
};
/// The cluster a role catalog belongs to.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DatatableRoleCluster {
/// Windmill's own Postgres, behind `instance` data tables.
#[default]
Instance,
/// The external instance cluster, behind `external_instance` data tables.
ExternalInstance,
}
impl DatatableRoleCluster {
pub fn as_str(self) -> &'static str {
match self {
Self::Instance => "instance",
Self::ExternalInstance => "external_instance",
}
}
pub fn parse(value: &str) -> Result<Self> {
match value {
"instance" => Ok(Self::Instance),
"external_instance" => Ok(Self::ExternalInstance),
other => Err(Error::BadRequest(format!(
"Unknown data table role cluster '{other}': expected instance or external_instance"
))),
}
}
/// The cluster whose roles a data table on `kind` can use. `None` for a resource-backed one,
/// which is never under roles.
pub fn of(kind: DataTableCatalogResourceType) -> Option<Self> {
match kind {
DataTableCatalogResourceType::Instance => Some(Self::Instance),
DataTableCatalogResourceType::ExternalInstance => Some(Self::ExternalInstance),
DataTableCatalogResourceType::Postgresql => None,
}
}
}
/// The connection every data table resolved to before roles existed (`custom_instance_user`). It
/// owns every pre-existing object, so it is a reserved name rather than a catalog entry: never
/// created, renamed or dropped.
@@ -164,28 +209,42 @@ pub async fn lock_instance_databases_governance<'a>(
/// need the names — but callers MUST NOT let `pwd` reach a response, a log line, an audit record
/// or an export. Nothing about who may call it: the credential is the whole risk, and `Debug` is
/// hand-written to redact it for the same reason.
pub async fn read_role_catalog(db: &DB) -> Result<DatatableRoleCatalog> {
crate::datatable_roles_oss::read_role_catalog(db).await
pub async fn read_role_catalog(
db: &DB,
cluster: DatatableRoleCluster,
) -> Result<DatatableRoleCatalog> {
crate::datatable_roles_oss::read_role_catalog(db, cluster).await
}
/// As [`read_role_catalog`], reading inside the caller's transaction so the value is the one
/// [`lock_role_catalog`] is protecting. Same disclosure contract.
pub async fn read_role_catalog_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
cluster: DatatableRoleCluster,
) -> Result<DatatableRoleCatalog> {
crate::datatable_roles_oss::read_role_catalog_tx(tx).await
crate::datatable_roles_oss::read_role_catalog_tx(tx, cluster).await
}
/// Record a role, in the caller's transaction so it commits with the `CREATE ROLE` it describes.
/// The cluster a role belongs to, or `None` if no role has this id.
pub async fn role_cluster(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
id: &str,
) -> Result<Option<DatatableRoleCluster>> {
crate::datatable_roles_oss::role_cluster(tx, id).await
}
/// Record a role, in the caller's transaction. On Windmill's own cluster that commits it with the
/// `CREATE ROLE` it describes; on the external cluster the role already exists by then.
///
/// Authorization: writes a generated Postgres credential. Callers MUST restrict this to superadmin
/// paths and MUST hold [`lock_role_catalog`] on `tx`.
pub async fn insert_role_catalog_entry(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
id: &str,
cluster: DatatableRoleCluster,
role: &InstanceDatatableRole,
) -> Result<()> {
crate::datatable_roles_oss::insert_role_catalog_entry(tx, id, role).await
crate::datatable_roles_oss::insert_role_catalog_entry(tx, id, cluster, role).await
}
/// Update a role's recorded name, login flag and password. Same contract as
@@ -215,7 +274,7 @@ pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Res
.find(|(_, role)| role.name == name)
.ok_or_else(|| {
Error::NotFound(format!(
"'{name}' is not a data table role of this instance. Defined roles: {}.",
"'{name}' is not a data table role of this database's cluster. Defined roles: {}.",
catalog
.values()
.map(|r| r.name.as_str())
@@ -231,70 +290,90 @@ pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Res
Ok(entry.0.as_str())
}
/// Every instance database the registry knows about. Role provisioning has to reach all of them:
/// a role that cannot `CONNECT` to a database is refused by Postgres before any grant matters.
/// Every database Windmill manages on `cluster`. Role provisioning has to reach all of them: a role
/// that cannot `CONNECT` to a database is refused by Postgres before any grant matters.
///
/// Authorization: checks nothing, and names every instance database across all workspaces. Callers
/// Authorization: checks nothing, and names every managed database across all workspaces. Callers
/// MUST be superadmin-gated or keep the names server-side; never return them to a workspace caller.
pub async fn registered_instance_databases(db: &DB) -> Result<Vec<String>> {
crate::datatable_roles_oss::registered_instance_databases(db).await
pub async fn registered_instance_databases(
db: &DB,
cluster: DatatableRoleCluster,
) -> Result<Vec<String>> {
crate::datatable_roles_oss::registered_instance_databases(db, cluster).await
}
/// `CONNECT` on `dbname` for every enabled role, and none for `PUBLIC`. Run at role creation, at
/// database creation, and lazily whenever an instance data table is administered, so a database
/// provisioned before a role existed is repaired rather than left silently unreachable.
/// `CONNECT` on `dbname` for every enabled role of `cluster`, and none for `PUBLIC`. Run at role
/// creation, at database creation, and lazily whenever a managed data table is administered, so a
/// database provisioned before a role existed is repaired rather than left silently unreachable.
///
/// Authorization: rewrites a database's ACL with the server's own credentials and checks nothing.
/// Callers MUST have authorized administration of `dbname` — superadmin, or an admin of the
/// workspace governing a data table on it.
pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> {
crate::datatable_roles_oss::converge_connect_grants(db, dbname).await
pub async fn converge_connect_grants(
db: &DB,
cluster: DatatableRoleCluster,
dbname: &str,
) -> Result<()> {
crate::datatable_roles_oss::converge_connect_grants(db, cluster, dbname).await
}
/// As [`converge_connect_grants`], with a catalog the caller already read. Same contract.
/// As [`converge_connect_grants`], with the catalog of `cluster` the caller already read. Same
/// contract.
pub async fn converge_connect_grants_with(
db: &DB,
cluster: DatatableRoleCluster,
dbname: &str,
catalog: &DatatableRoleCatalog,
) -> Result<()> {
crate::datatable_roles_oss::converge_connect_grants_with(db, dbname, catalog).await
crate::datatable_roles_oss::converge_connect_grants_with(db, cluster, dbname, catalog).await
}
/// `CREATE ROLE <name> LOGIN PASSWORD ...; GRANT <name> TO custom_instance_user`, and `CONNECT` on
/// every registered database. No privileges beyond that — an admin grants them through SQL or the
/// ACL editor.
/// `CREATE ROLE <name> LOGIN PASSWORD ...; GRANT <name> TO custom_instance_user` on `cluster`. No
/// privileges beyond that — an admin grants them through SQL or the ACL editor.
///
/// On Windmill's own cluster the DDL runs on `tx`, so it commits with the catalog row. The external
/// cluster is another server: the role is created there before `tx` commits, and callers MUST drop
/// it again ([`drop_datatable_role`]) if `tx` then fails to commit.
///
/// Authorization: creates a cluster-wide Postgres login. Callers MUST restrict this to superadmin
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
pub async fn create_instance_role(
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
pub async fn create_datatable_role(
db: &DB,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
cluster: DatatableRoleCluster,
name: &str,
password: &str,
) -> Result<()> {
crate::datatable_roles_oss::create_instance_role(tx, name, password).await
crate::datatable_roles_oss::create_datatable_role(db, tx, cluster, name, password).await
}
/// Authorization: alters a cluster-wide Postgres login. Callers MUST restrict this to superadmin
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
pub async fn set_instance_role_login(
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
pub async fn set_datatable_role_login(
db: &DB,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
cluster: DatatableRoleCluster,
name: &str,
enabled: bool,
) -> Result<()> {
crate::datatable_roles_oss::set_instance_role_login(tx, name, enabled).await
crate::datatable_roles_oss::set_datatable_role_login(db, tx, cluster, name, enabled).await
}
/// A rename discards an md5-hashed password, so the caller has to hand over a fresh one.
/// A rename discards an md5-hashed password, so the caller has to hand over a fresh one. On the
/// external cluster the rename lands before `tx` commits, and callers MUST rename it back if `tx`
/// then fails to commit.
///
/// Authorization: renames a cluster-wide Postgres login. Callers MUST restrict this to superadmin
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
pub async fn rename_instance_role(
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
pub async fn rename_datatable_role(
db: &DB,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
cluster: DatatableRoleCluster,
from: &str,
to: &str,
password: &str,
) -> Result<()> {
crate::datatable_roles_oss::rename_instance_role(tx, from, to, password).await
crate::datatable_roles_oss::rename_datatable_role(db, tx, cluster, from, to, password).await
}
/// A role owning anything in any database blocks its own `DROP ROLE`, and both its objects and the
@@ -302,8 +381,9 @@ pub async fn rename_instance_role(
/// registry. An unreachable database aborts the whole delete: dropping the role while one database
/// still holds objects owned by it leaves those objects owned by a numeric OID nobody can name.
///
/// Each pass runs as the instance's own Postgres user rather than `custom_instance_user`, which
/// owns the databases and can therefore revoke a grant whoever made it. `custom_instance_user`
/// Each pass runs as the cluster's administrator rather than `custom_instance_user`: on Windmill's
/// own cluster the instance's Postgres user, on the external one its configured admin login. Both
/// own the databases and can therefore revoke a grant whoever made it. `custom_instance_user`
/// could only undo what it granted itself, so a privilege planted by an operator in psql — the
/// ordinary way privileges reach a role — would survive and block the drop.
///
@@ -311,16 +391,19 @@ pub async fn rename_instance_role(
/// MUST restrict this to superadmin paths, and MUST hold [`lock_role_catalog`] on `tx`.
///
/// The per-database passes open their own connections and cannot join `tx`; the lock is what keeps
/// a concurrent mutation out while they run. Only the final `DROP ROLE` is on `tx`, so it commits
/// or rolls back with the catalog write that forgets the role. Those passes commit as they go, so
/// callers MUST have disabled the role in an earlier committed transaction: a failure part-way
/// then leaves a disabled role to retry, not an enabled one already stripped in some databases.
pub async fn drop_instance_role(
/// a concurrent mutation out while they run. On Windmill's own cluster only the final `DROP ROLE`
/// is on `tx`, so it commits or rolls back with the catalog write that forgets the role; on the
/// external cluster it runs there, and tolerates a role already gone so a retry after a failed
/// commit can finish. The passes commit as they go, so callers MUST have disabled the role in an
/// earlier committed transaction: a failure part-way then leaves a disabled role to retry, not an
/// enabled one already stripped in some databases.
pub async fn drop_datatable_role(
db: &DB,
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
cluster: DatatableRoleCluster,
name: &str,
) -> Result<()> {
crate::datatable_roles_oss::drop_instance_role(db, tx, name).await
crate::datatable_roles_oss::drop_datatable_role(db, tx, cluster, name).await
}
#[cfg(test)]
@@ -24,12 +24,12 @@ pub fn datatable_roles_unavailable() -> Error {
#[cfg(all(feature = "private", feature = "enterprise"))]
pub(crate) use crate::datatable_roles_ee::{
can_use_datatable_role, can_use_datatable_role_in_governing_workspace, converge_connect_grants,
converge_connect_grants_with, create_instance_role, delete_role_catalog_entry,
drop_instance_role, ensure_can_use_datatable_role, ensure_datatable_admin_access,
converge_connect_grants_with, create_datatable_role, delete_role_catalog_entry,
drop_datatable_role, ensure_can_use_datatable_role, ensure_datatable_admin_access,
ensure_instance_db_grant_options_unchecked, forget_datatable_role_everywhere,
insert_role_catalog_entry, read_role_catalog, read_role_catalog_tx,
registered_instance_databases, rename_instance_role, resolve_datatable_role_connection,
set_instance_role_login, update_role_catalog_entry,
registered_instance_databases, rename_datatable_role, resolve_datatable_role_connection,
role_cluster, set_datatable_role_login, update_role_catalog_entry,
};
#[cfg(not(all(feature = "private", feature = "enterprise")))]
@@ -39,7 +39,7 @@ pub(crate) use ce::*;
mod ce {
use super::datatable_roles_unavailable as unavailable;
use crate::{
datatable_roles::{DatatableRoleCatalog, InstanceDatatableRole},
datatable_roles::{DatatableRoleCatalog, DatatableRoleCluster, InstanceDatatableRole},
db::AuthedRef,
error::Result,
workspaces::{
@@ -50,17 +50,31 @@ mod ce {
type Tx<'a> = sqlx::Transaction<'a, sqlx::Postgres>;
pub(crate) async fn read_role_catalog(_db: &DB) -> Result<DatatableRoleCatalog> {
pub(crate) async fn read_role_catalog(
_db: &DB,
_cluster: DatatableRoleCluster,
) -> Result<DatatableRoleCatalog> {
Err(unavailable())
}
pub(crate) async fn read_role_catalog_tx(_tx: &mut Tx<'_>) -> Result<DatatableRoleCatalog> {
pub(crate) async fn read_role_catalog_tx(
_tx: &mut Tx<'_>,
_cluster: DatatableRoleCluster,
) -> Result<DatatableRoleCatalog> {
Err(unavailable())
}
pub(crate) async fn role_cluster(
_tx: &mut Tx<'_>,
_id: &str,
) -> Result<Option<DatatableRoleCluster>> {
Err(unavailable())
}
pub(crate) async fn insert_role_catalog_entry(
_tx: &mut Tx<'_>,
_id: &str,
_cluster: DatatableRoleCluster,
_role: &InstanceDatatableRole,
) -> Result<()> {
Err(unavailable())
@@ -78,43 +92,57 @@ mod ce {
Err(unavailable())
}
pub(crate) async fn registered_instance_databases(_db: &DB) -> Result<Vec<String>> {
pub(crate) async fn registered_instance_databases(
_db: &DB,
_cluster: DatatableRoleCluster,
) -> Result<Vec<String>> {
Err(unavailable())
}
/// Nothing to converge: with no roles to admit, an instance database keeps the `CONNECT`
/// grants it was created with, `PUBLIC`'s included, as it did before roles existed.
pub(crate) async fn converge_connect_grants(_db: &DB, _dbname: &str) -> Result<()> {
/// Nothing to converge: with no roles to admit, a managed database keeps the `CONNECT` grants
/// it was created with, as it did before roles existed.
pub(crate) async fn converge_connect_grants(
_db: &DB,
_cluster: DatatableRoleCluster,
_dbname: &str,
) -> Result<()> {
Ok(())
}
/// As [`converge_connect_grants`].
pub(crate) async fn converge_connect_grants_with(
_db: &DB,
_cluster: DatatableRoleCluster,
_dbname: &str,
_catalog: &DatatableRoleCatalog,
) -> Result<()> {
Ok(())
}
pub(crate) async fn create_instance_role(
pub(crate) async fn create_datatable_role(
_db: &DB,
_tx: &mut Tx<'_>,
_cluster: DatatableRoleCluster,
_name: &str,
_password: &str,
) -> Result<()> {
Err(unavailable())
}
pub(crate) async fn set_instance_role_login(
pub(crate) async fn set_datatable_role_login(
_db: &DB,
_tx: &mut Tx<'_>,
_cluster: DatatableRoleCluster,
_name: &str,
_enabled: bool,
) -> Result<()> {
Err(unavailable())
}
pub(crate) async fn rename_instance_role(
pub(crate) async fn rename_datatable_role(
_db: &DB,
_tx: &mut Tx<'_>,
_cluster: DatatableRoleCluster,
_from: &str,
_to: &str,
_password: &str,
@@ -122,12 +150,18 @@ mod ce {
Err(unavailable())
}
pub(crate) async fn drop_instance_role(_db: &DB, _tx: &mut Tx<'_>, _name: &str) -> Result<()> {
pub(crate) async fn drop_datatable_role(
_db: &DB,
_tx: &mut Tx<'_>,
_cluster: DatatableRoleCluster,
_name: &str,
) -> Result<()> {
Err(unavailable())
}
pub(crate) async fn ensure_instance_db_grant_options_unchecked(
_db: &DB,
_cluster: DatatableRoleCluster,
_dbname: &str,
) -> Result<()> {
Err(unavailable())
@@ -192,27 +192,48 @@ pub async fn external_instance_database_usages<'c>(
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
/// Refuse to unset the cluster while Windmill still has databases or data table roles on it, or a
/// workspace still points at one: every data table there would stop resolving, and every role
/// would be a login nothing can drop any more. 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(conn: &mut sqlx::PgConnection) -> Result<()> {
let state = read_external_instance_pg_state(&mut *conn).await?;
let usages = external_instance_database_usages(&mut *conn).await?;
if state.databases.is_empty() && usages.is_empty() {
pub async fn ensure_external_instance_pg_removable(db: &DB) -> Result<()> {
ensure_external_instance_pg_unused(db, &format!("removing {EXTERNAL_INSTANCE_PG_SETTING}"))
.await
}
/// Refuse while Windmill has databases or data table roles on the cluster, or a workspace points
/// at one of its databases. `before` finishes the sentence saying what to do first.
async fn ensure_external_instance_pg_unused(db: &DB, before: &str) -> Result<()> {
let state = read_external_instance_pg_state(db).await?;
let usages = external_instance_database_usages(db).await?;
let roles = sqlx::query_scalar::<_, String>(
"SELECT name FROM datatable_role WHERE cluster = 'external_instance' ORDER BY name",
)
.fetch_all(db)
.await?;
if state.databases.is_empty() && usages.is_empty() && roles.is_empty() {
return Ok(());
}
let names = state
.databases
.keys()
.chain(usages.keys())
.collect::<BTreeSet<_>>()
.into_iter()
.cloned()
.collect::<Vec<_>>()
.join(", ");
let mut held = vec![];
if !(state.databases.is_empty() && usages.is_empty()) {
let names = state
.databases
.keys()
.chain(usages.keys())
.collect::<BTreeSet<_>>()
.into_iter()
.cloned()
.collect::<Vec<_>>()
.join(", ");
held.push(format!("databases in use ({names})"));
}
if !roles.is_empty() {
held.push(format!("data table roles ({})", roles.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}."
"The external instance cluster still holds {}. Drop them and repoint the data tables and \
Ducklake catalogs using them before {before}.",
held.join(" and ")
)))
}
@@ -287,9 +308,6 @@ pub async fn lock_external_instance_pg_state(
/// Refuse a data table naming `dbname` unless Windmill created it on the external cluster. Takes
/// the lock drops take, so none can remove the database before `tx`, which saves the data table,
/// commits.
///
/// Authorization: its refusal says whether Windmill created a database of that name, which is
/// instance-wide knowledge. Callers MUST have authorized the caller as superadmin first.
pub async fn ensure_external_instance_database_registered(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
dbname: &str,
@@ -329,12 +347,9 @@ pub async fn write_external_instance_pg_setting(
};
let mut tx = db.begin().await?;
lock_external_instance_pg_state(&mut tx).await?;
// Every check runs on this transaction's own connection: it holds the advisory lock, and
// taking a second connection from the pool while other writers queue on that lock is how a
// small pool deadlocks.
match value {
None => {
ensure_external_instance_pg_removable(&mut tx).await?;
ensure_external_instance_pg_removable(db).await?;
sqlx::query("DELETE FROM global_settings WHERE name = $1")
.bind(EXTERNAL_INSTANCE_PG_SETTING)
.execute(&mut *tx)
@@ -342,7 +357,7 @@ pub async fn write_external_instance_pg_setting(
}
Some(value) => {
crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?;
ensure_external_instance_pg_not_repointed(&mut tx, value).await?;
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()",
@@ -384,14 +399,15 @@ pub async fn write_external_instance_pg_from_diff(
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.
/// Refuse pointing the setting at another host or port while databases or data table roles live on
/// the current one. Data tables name databases, and the role catalog names logins, not clusters, so
/// both 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(
conn: &mut sqlx::PgConnection,
db: &DB,
value: &serde_json::Value,
) -> Result<()> {
let Some(current) = read_external_instance_pg_config(&mut *conn).await? else {
let Some(current) = read_external_instance_pg_config(db).await? else {
return Ok(());
};
let Ok(desired) = serde_json::from_value::<ExternalInstancePg>(value.clone()) else {
@@ -400,17 +416,11 @@ async fn ensure_external_instance_pg_not_repointed(
if external_instance_pg_address(&current) == external_instance_pg_address(&desired) {
return Ok(());
}
let state = read_external_instance_pg_state(&mut *conn).await?;
let usages = external_instance_database_usages(&mut *conn).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)
)))
ensure_external_instance_pg_unused(
db,
&format!("pointing {EXTERNAL_INSTANCE_PG_SETTING} at another cluster"),
)
.await
}
/// Converge the external cluster on the configured login: check what it can do, create or update
+14 -6
View File
@@ -1563,18 +1563,20 @@ pub(crate) fn instance_db_grants(dbname: &str) -> String {
)
}
/// Re-apply [`instance_db_grants`] to an instance database provisioned before data table roles
/// existed, whose grants carry no grant option. Connects as the instance's own Postgres user —
/// the database and `public` schema owner — since only it can hand out an option it holds.
/// Re-apply [`instance_db_grants`] to a managed database provisioned before data table roles
/// existed, whose grants carry no grant option. Connects as the cluster's administrator — the
/// database and `public` schema owner — since only it can hand out an option it holds.
///
/// Authorization: reaches an instance database with the server's own credentials and checks
/// Authorization: reaches a managed database with the server's own credentials and checks
/// nothing. Callers MUST have authorized administration of `dbname` — superadmin, or an admin of
/// the workspace governing a data table on it.
pub async fn ensure_instance_db_grant_options_unchecked(
db: &DB,
cluster: crate::datatable_roles::DatatableRoleCluster,
dbname: &str,
) -> error::Result<()> {
crate::datatable_roles_oss::ensure_instance_db_grant_options_unchecked(db, dbname).await
crate::datatable_roles_oss::ensure_instance_db_grant_options_unchecked(db, cluster, dbname)
.await
}
/// Create a custom instance database: CREATE DATABASE, grant permissions, register in global_settings.
@@ -1651,7 +1653,13 @@ pub async fn create_custom_instance_database(
// A data table role can only reach a database it may CONNECT to, and PUBLIC's default CONNECT
// would otherwise let every role in regardless of what this instance defines. Best-effort: a
// failure here leaves the database usable as `admin`, and the next role change repairs it.
if let Err(e) = crate::datatable_roles::converge_connect_grants(db, dbname).await {
if let Err(e) = crate::datatable_roles::converge_connect_grants(
db,
crate::datatable_roles::DatatableRoleCluster::Instance,
dbname,
)
.await
{
tracing::warn!("Could not set CONNECT grants on instance database '{dbname}': {e}");
}
+5 -4
View File
@@ -1551,13 +1551,14 @@ pub async fn lock_fork_datatables(conn: &mut sqlx::PgConnection, w_id: &str) ->
}
impl GoverningDatatable {
/// Backed by the Windmill instance's own Postgres, which is the only substrate data table
/// roles apply to.
pub fn is_instance(&self) -> bool {
/// The Windmill-managed cluster whose data table roles this entry can use. `None` for a
/// resource-backed one: roles are logins Windmill creates, and it creates none on a host a
/// workspace admin chose.
pub fn role_cluster(&self) -> Option<crate::datatable_roles::DatatableRoleCluster> {
self.datatable
.database
.as_ref()
.is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
.and_then(|d| crate::datatable_roles::DatatableRoleCluster::of(d.resource_type))
}
}
+90 -87
View File
@@ -1475,7 +1475,6 @@ pub async fn do_duckdb(
&job.id,
client,
&mut hidden_passwords,
job_dir,
)
.await?,
);
@@ -1491,13 +1490,12 @@ pub async fn do_duckdb(
&mut hidden_passwords,
&job.workspace_id,
materialize.as_ref().map(|(_, m)| m.asset_path.as_str()),
job_dir,
)
.await?
{
probe_blocks.extend(q);
} else if let Some(q) =
transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job, job_dir)
transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
.await?
{
probe_blocks.extend(q);
@@ -1554,7 +1552,6 @@ pub async fn do_duckdb(
&job.id,
client,
&mut hidden_passwords,
job_dir,
)
.await?,
);
@@ -1570,13 +1567,12 @@ pub async fn do_duckdb(
&mut hidden_passwords,
&job.workspace_id,
materialize.as_ref().map(|(_, m)| m.asset_path.as_str()),
job_dir,
)
.await?
{
v.extend(ducklake_query);
} else if let Some(datatable_query) =
transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job, job_dir)
transform_attach_datatable(&query_block, conn, &mut hidden_passwords, job)
.await?
{
v.extend(datatable_query);
@@ -2250,16 +2246,10 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option<ParsedAttachDbResource
/// A connection that explicitly refuses invalid certificates — the external instance cluster's —
/// keeps its mode instead: under `require` its shared password would go to whichever server
/// answers. DuckDB's libpq takes one root file, so it gets the system bundle plus the configured
/// certificate, written in the job directory: a resource's certificate is workspace-controlled, so
/// a file per distinct one has to go with the job rather than pile up on the worker.
fn pg_attach_verification<'a>(
res: &'a PgDatabase,
job_dir: &str,
) -> Result<Option<(&'a str, std::path::PathBuf)>> {
/// 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
}
Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode,
_ => return Ok(None),
};
let bundle = windmill_common::system_ca_bundle()
@@ -2275,20 +2265,58 @@ fn pg_attach_verification<'a>(
}
let roots = format!("{bundle}\n{pem}\n");
use sha2::Digest;
let path = std::path::Path::new(job_dir).join(format!(
"pg_roots_{}.pem",
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])
));
if !path.is_file() {
std::fs::write(&path, &roots)
.map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?;
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)))
}
fn pg_attach_uri(res: &PgDatabase, job_dir: &str) -> Result<String> {
/// 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, job_dir)? else {
let Some((mode, roots)) = pg_attach_verification(res)? else {
return Ok(uri);
};
let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| {
@@ -2300,11 +2328,11 @@ fn pg_attach_uri(res: &PgDatabase, job_dir: &str) -> Result<String> {
))
}
fn format_attach_db_conn_str(db_resource: Value, db_type: &str, job_dir: &str) -> Result<String> {
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)?;
pg_attach_uri(&res, job_dir)?
pg_attach_uri(&res)?
}
#[cfg(feature = "mysql")]
"mysql" => {
@@ -2376,7 +2404,6 @@ async fn transform_attach_db_resource_query(
job_id: &Uuid,
client: &AuthedClient,
hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
job_dir: &str,
) -> Result<Vec<String>> {
let db_resource: Value = client
.get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string()))
@@ -2384,14 +2411,8 @@ async fn transform_attach_db_resource_query(
if let Some(pwd) = db_resource.get("password").and_then(|p| p.as_str()) {
hidden_passwords.lock().unwrap().push(pwd.to_string());
}
db_resource_to_attach_statements(
db_resource,
parsed.name,
parsed.db_type,
parsed.extra_args,
job_dir,
)
.await
db_resource_to_attach_statements(db_resource, parsed.name, parsed.db_type, parsed.extra_args)
.await
}
async fn db_resource_to_attach_statements(
@@ -2399,12 +2420,11 @@ async fn db_resource_to_attach_statements(
ident_name: &str,
db_type: &str,
extra_args: Option<&str>,
job_dir: &str,
) -> Result<Vec<String>> {
// Escape single quotes: the connection string is built from resource fields
// (host/db/user/password) and embedded in a single-quoted DuckDB literal, so an
// unescaped quote in any field would otherwise break out of the ATTACH statement.
let conn_str = format_attach_db_conn_str(db_resource, db_type, job_dir)?.replace('\'', "''");
let conn_str = format_attach_db_conn_str(db_resource, db_type)?.replace('\'', "''");
let attach_str = format!(
"ATTACH '{}' as {} (TYPE {}{});",
conn_str,
@@ -2427,7 +2447,6 @@ async fn transform_attach_ducklake(
hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
w_id: &str,
materialize_target: Option<&str>,
job_dir: &str,
) -> Result<Option<Vec<String>>> {
lazy_static::lazy_static! {
static ref RE: regex::Regex = regex::Regex::new(r"(?i)ATTACH\s*'ducklake(://[^':]+)?'\s*AS\s+([^ ;]+)\s*(\([^)]*\))?").unwrap();
@@ -2478,7 +2497,7 @@ async fn transform_attach_ducklake(
// single-quoted DuckDB literals below, so an unescaped quote in a resource
// field would break out of the ATTACH statement.
let db_conn_str =
format_attach_db_conn_str(ducklake.catalog_resource, db_type, job_dir)?.replace('\'', "''");
format_attach_db_conn_str(ducklake.catalog_resource, db_type)?.replace('\'', "''");
let storage = ducklake
.storage
.storage
@@ -2533,7 +2552,6 @@ async fn transform_attach_ducklake(
defer,
materialize_target,
hidden_passwords,
job_dir,
)?);
}
Ok(Some(statements))
@@ -2567,7 +2585,6 @@ fn fork_defer_statements(
defer: &windmill_common::workspaces::DucklakeForkDefer,
materialize_target: Option<&str>,
hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
job_dir: &str,
) -> Result<Vec<String>> {
let mut stmts = vec![];
if defer.ancestors.is_empty() {
@@ -2586,7 +2603,7 @@ fn fork_defer_statements(
};
stmts.push(get_attach_db_install_str(db_type)?.to_string());
let conn_str =
format_attach_db_conn_str(a.catalog_resource.clone(), db_type, job_dir)?.replace('\'', "''");
format_attach_db_conn_str(a.catalog_resource.clone(), db_type)?.replace('\'', "''");
let storage = a
.storage
.storage
@@ -2705,7 +2722,6 @@ async fn transform_attach_datatable(
conn: &Connection,
hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
job: &MiniPulledJob,
job_dir: &str,
) -> Result<Option<Vec<String>>> {
let Some(attached) = parse_attach_datatable(query) else {
return Ok(None);
@@ -2748,7 +2764,6 @@ async fn transform_attach_datatable(
Ok(Some(pg_secret_attach_statements(
db_resource,
attached.alias,
job_dir,
)?))
}
@@ -2769,18 +2784,14 @@ fn datatable_secret_name(alias: &str) -> String {
/// ATTACH a datatable's postgres database through a DuckDB TEMPORARY SECRET holding
/// the connection parameters; only sslmode rides in the ATTACH string.
fn pg_secret_attach_statements(
db_resource: Value,
alias_name: &str,
job_dir: &str,
) -> Result<Vec<String>> {
fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<Vec<String>> {
let res: PgDatabase = serde_json::from_value(db_resource)?;
// Escape single quotes: each field is embedded in a single-quoted DuckDB literal,
// so an unescaped quote would break out of the CREATE SECRET statement.
let esc = |s: &str| s.replace('\'', "''");
// The postgres secret type has no sslmode parameter, so it goes in the ATTACH
// string; only the libpq values PgDatabase::to_uri collapses to are forwarded.
let sslmode = match pg_attach_verification(&res, job_dir)? {
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=''{}''",
@@ -2889,9 +2900,6 @@ mod tests {
#[test]
fn pg_attach_keeps_verification_only_when_required() {
let job_dir = std::env::temp_dir().join(format!("wm-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&job_dir).unwrap();
let job_dir = job_dir.to_string_lossy().to_string();
let pg = |sslmode: &str, accept_invalid_certs: Option<bool>| PgDatabase {
host: "db.internal".to_string(),
user: Some("custom_instance_user".to_string()),
@@ -2904,35 +2912,30 @@ mod tests {
use_iam_auth: None,
region: None,
};
let uri = pg_attach_uri(&pg("verify-full", Some(false)), &job_dir).unwrap();
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();
assert!(std::fs::read_to_string(root.as_ref())
.unwrap()
.contains("-----BEGIN CERTIFICATE-----test"));
// Every certificate a job attaches keeps its own file: one attach must not evict another's.
for i in 0..40 {
let 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}"));
let other = pg_attach_uri(&other, &job_dir).unwrap();
let path = urlencoding::decode(other.split("sslrootcert=").nth(1).unwrap()).unwrap();
assert!(std::path::Path::new(path.as_ref()).is_file(), "{path}");
pg_attach_uri(&other).unwrap();
}
assert!(std::path::Path::new(root.as_ref()).is_file(), "the first file is still there");
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", &job_dir).unwrap()[3];
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), &job_dir)
.unwrap()
.ends_with("?sslmode=require"));
assert!(pg_attach_uri(&pg("require", Some(false)), &job_dir)
.unwrap()
.ends_with("?sslmode=require"));
std::fs::remove_dir_all(&job_dir).unwrap();
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]
@@ -3086,7 +3089,7 @@ mod tests {
let mut defer = test_fork_defer(vec![("orders", false)], vec![]);
defer.ancestors[0].extra_args = Some("ENCRYPTED true".to_string());
let mut hp = Arc::new(Mutex::new(vec![]));
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
let attach = stmts
.iter()
.find(|s| s.starts_with("ATTACH IF NOT EXISTS"))
@@ -3110,7 +3113,7 @@ mod tests {
vec![],
);
let mut hp = Arc::new(Mutex::new(vec![]));
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
let joined = stmts.join("\n");
assert!(
joined.contains(
@@ -3142,7 +3145,7 @@ mod tests {
fn test_fork_defer_statements_shape() {
let defer = test_fork_defer(vec![("orders", false), ("dim", true)], vec![]);
let mut hp = Arc::new(Mutex::new(vec![]));
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
let joined = stmts.join("\n");
// Ancestor attach: read-only, idempotent, never auto-migrating or auto-creating.
assert!(joined.contains("ATTACH IF NOT EXISTS"), "{joined}");
@@ -3169,7 +3172,7 @@ mod tests {
let defer = test_fork_defer(vec![("orders", false)], vec!["orders", "orders_current"]);
let mut hp = Arc::new(Mutex::new(vec![]));
let stmts =
fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp, "/tmp")
fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp)
.unwrap();
let joined = stmts.join("\n");
assert!(!joined.contains("CREATE VIEW"), "{joined}");
@@ -3186,14 +3189,14 @@ mod tests {
// status can't be trusted) → no DROP VIEW, or the job would wedge on a type mismatch.
let defer = test_fork_defer(vec![("orders", false)], vec![]);
let stmts =
fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp, "/tmp")
fork_defer_statements("lake", "_wm_target", &defer, Some("lake/orders"), &mut hp)
.unwrap();
assert!(!stmts.join("\n").contains("DROP VIEW"), "{stmts:?}");
// Target in a different lake → this lake's defer views are untouched.
let defer = test_fork_defer(vec![("orders", false)], vec!["orders"]);
let stmts =
fork_defer_statements("lake", "dl", &defer, Some("other/orders"), &mut hp, "/tmp").unwrap();
fork_defer_statements("lake", "dl", &defer, Some("other/orders"), &mut hp).unwrap();
let joined = stmts.join("\n");
assert!(
joined.contains("CREATE VIEW IF NOT EXISTS dl.\"orders\""),
@@ -3206,7 +3209,7 @@ mod tests {
fn test_fork_defer_statements_schema_qualified() {
let defer = test_fork_defer(vec![("staging.raw", false)], vec![]);
let mut hp = Arc::new(Mutex::new(vec![]));
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp, "/tmp").unwrap();
let stmts = fork_defer_statements("lake", "dl", &defer, None, &mut hp).unwrap();
let joined = stmts.join("\n");
assert!(
joined.contains("CREATE SCHEMA IF NOT EXISTS dl.\"staging\";"),
@@ -4032,7 +4035,7 @@ mod tests {
"dbname": "mydb",
"sslmode": "require"
});
let result = format_attach_db_conn_str(db_resource, "postgres", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "postgres").unwrap();
// Should be in URI format: postgres://user:password@host:port/dbname?sslmode=require
assert!(result.starts_with("postgres://"));
assert!(result.contains("admin:secret123@localhost:5432/mydb"));
@@ -4045,7 +4048,7 @@ mod tests {
"host": "db.example.com",
"dbname": "production"
});
let result = format_attach_db_conn_str(db_resource, "postgres", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "postgres").unwrap();
// Should be in URI format with defaults: postgres://postgres:@host:5432/dbname?sslmode=prefer
assert!(result.starts_with("postgres://"));
assert!(result.contains("@db.example.com:5432/production"));
@@ -4058,7 +4061,7 @@ mod tests {
"host": "localhost",
"dbname": "test"
});
let result = format_attach_db_conn_str(db_resource, "postgresql", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "postgresql").unwrap();
// Should be in URI format (postgresql is treated the same as postgres)
assert!(result.starts_with("postgres://"));
assert!(result.contains("@localhost:5432/test"));
@@ -4075,7 +4078,7 @@ mod tests {
"dbname": "wm_datatables",
"sslmode": "require"
});
let stmts = pg_secret_attach_statements(db_resource, "dt", "/tmp").unwrap();
let stmts = pg_secret_attach_statements(db_resource, "dt").unwrap();
assert_eq!(stmts[0], "INSTALL postgres;");
assert_eq!(stmts[1], "LOAD postgres;");
let secret_name = datatable_secret_name("dt");
@@ -4106,7 +4109,7 @@ mod tests {
if let Some(s) = input {
db_resource["sslmode"] = json!(s);
}
let stmts = pg_secret_attach_statements(db_resource, "dt", "/tmp").unwrap();
let stmts = pg_secret_attach_statements(db_resource, "dt").unwrap();
assert!(
stmts[3].starts_with(&format!("ATTACH 'sslmode={expected}'")),
"sslmode {input:?} → {}",
@@ -4129,7 +4132,7 @@ mod tests {
let db_resource = json!({
"project_id": "my-gcp-project"
});
let result = format_attach_db_conn_str(db_resource, "bigquery", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "bigquery").unwrap();
assert_eq!(result, "project=my-gcp-project");
}
@@ -4138,7 +4141,7 @@ mod tests {
let db_resource = json!({
"other_field": "value"
});
let result = format_attach_db_conn_str(db_resource, "bigquery", "/tmp");
let result = format_attach_db_conn_str(db_resource, "bigquery");
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("project_id"));
}
@@ -4146,7 +4149,7 @@ mod tests {
#[test]
fn test_format_attach_db_conn_str_unsupported_type() {
let db_resource = json!({});
let result = format_attach_db_conn_str(db_resource, "oracle", "/tmp");
let result = format_attach_db_conn_str(db_resource, "oracle");
assert!(result.is_err());
assert!(result
.unwrap_err()
@@ -4160,7 +4163,7 @@ mod tests {
"host": "localhost",
"dbname": "test"
});
let result = format_attach_db_conn_str(db_resource, "POSTGRES", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "POSTGRES").unwrap();
// Should be in URI format
assert!(result.starts_with("postgres://"));
assert!(result.contains("@localhost:5432/test"));
@@ -4177,7 +4180,7 @@ mod tests {
"database": "app_db",
"ssl": true
});
let result = format_attach_db_conn_str(db_resource, "mysql", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "mysql").unwrap();
assert!(result.contains("database=app_db"));
assert!(result.contains("host=mysql.example.com"));
assert!(result.contains("ssl_mode=required"));
@@ -4194,7 +4197,7 @@ mod tests {
"database": "test",
"ssl": false
});
let result = format_attach_db_conn_str(db_resource, "mysql", "/tmp").unwrap();
let result = format_attach_db_conn_str(db_resource, "mysql").unwrap();
assert!(result.contains("ssl_mode=disabled"));
}
+4 -2
View File
@@ -111,6 +111,8 @@ const migrateCommand = new Command()
)
.action(migrateDown as any);
type DataTableResourceType = "postgresql" | "instance" | "external_instance";
async function create(
opts: GlobalOptions & { resource?: string; force?: boolean },
name?: string,
@@ -139,12 +141,12 @@ async function create(
const datatables: Record<
string,
{ database: { resource_type: "postgresql" | "instance"; resource_path?: string } }
{ database: { resource_type: DataTableResourceType; resource_path?: string } }
> = {};
for (const d of existing) {
datatables[d.name] = {
database: {
resource_type: d.resource_type as "postgresql" | "instance",
resource_type: d.resource_type as DataTableResourceType,
resource_path: d.resource_path ?? undefined,
},
};
+88
View File
@@ -0,0 +1,88 @@
# External instance data tables
A data table is backed by one of three things: a Postgres resource a workspace brings
(`postgresql`), a database on Windmill's own cluster (`instance`), or a database on a separate
cluster Windmill administers (`external_instance`, Enterprise Edition). The third is what this
document covers; Ducklake catalogs take the same three shapes.
Windmill administers the external cluster the way it administers its own: it creates and drops
databases there, owns `custom_instance_user` and `custom_instance_replication_user`, and creates
the data table roles of that cluster. It logs in as the admin in the `external_instance_pg`
instance setting, and keeps what it generates in the hidden `external_instance_pg_state` setting.
## Code
| Where | What |
|---|---|
| `windmill-common/src/external_instance_pg.rs` | Setting, state, usage accounting, the lifecycle lock, the OSS forwarders |
| `windmill-common/src/external_instance_pg_ee.rs` | Setup, database create and drop, the admin connection |
| `windmill-common/src/datatable_roles.rs` | Per-cluster role catalogs (`DatatableRoleCluster`) |
| `windmill-common/src/workspaces.rs` | Resolution (`resolve_datatable_connection_unchecked`), `managed_database_uses` |
| `windmill-api-settings/src/lib.rs` | `/settings/external_instance_pg/*`, `/settings/datatable_roles` |
## What holds it together
- **One lifecycle lock.** `lock_external_instance_pg_state` serializes everything that changes
which databases exist on the cluster or which entries name them: setup, create, drop, data table
and Ducklake saves, external role DDL, and writes to the setting itself. Anything reading the
configuration to reach the cluster reads it under that lock, so a database is never created on
one cluster and registered while the setting names another.
- **Windmill only touches what it made.** Databases it creates carry a comment, and a drop
requires it. The two managed roles and every data table role carry their own comment, and setup
refuses a `custom_instance_user` without it rather than resetting the password of someone else's
role.
- **Creation needs a successful setup.** `set_up_for` records the `host:port` the last successful
setup converged. Creating a database on a cluster that setup has not succeeded on is refused.
- **Nothing is dropped from under a user.** `managed_database_uses` lists every data table naming
a database, every fork pointing at those, every Ducklake catalog on it, and every fork Ducklake
metadata schema still to be dropped. Fork cleanup exempts exactly the entry it is cleaning up.
- **Fork copies belong to a workspace.** `wm_fork_*` is a name, not an authorization: every
database of a cluster answers to the same `custom_instance_user`. The registry records the
workspace a copy was created for, and a member can only import into or fork onto a copy of their
own workspace.
- **Roles are per cluster.** `datatable_role.cluster` splits the catalog, so the same role name can
exist on both clusters. Role names are unique per cluster, as they are in Postgres.
## Running one locally
```bash
docker run -d --name wm-external-pg -e POSTGRES_PASSWORD=external -p 5497:5432 postgres:18 \
-c wal_level=logical
psql "postgresql://postgres:external@127.0.0.1:5497/postgres" \
-c "CREATE ROLE wm_admin LOGIN PASSWORD 'adminpw' CREATEDB CREATEROLE REPLICATION"
```
A non-superuser admin with `CREATEDB` and `CREATEROLE` is the realistic case: managed Postgres
gives nothing more. `REPLICATION` is only needed for Postgres triggers on external data tables.
Then, as superadmin (`$T` is a token):
```bash
api=http://localhost:8000/api
curl -s -X POST $api/settings/global/external_instance_pg -H "Authorization: Bearer $T" \
-H 'Content-Type: application/json' \
--data '{"value":{"host":"127.0.0.1","port":5497,"user":"wm_admin","password":"adminpw","sslmode":"disable"}}'
curl -s -X POST $api/settings/external_instance_pg/setup -H "Authorization: Bearer $T" \
-H 'Content-Type: application/json' --data '{}' # report per step
curl -s -X POST $api/settings/external_instance_pg/databases/dt_demo -H "Authorization: Bearer $T" \
-H 'Content-Type: application/json' --data '{}'
curl -s -X POST $api/w/admins/workspaces/edit_datatable_config -H "Authorization: Bearer $T" \
-H 'Content-Type: application/json' \
--data '{"settings":{"datatables":{"demo":{"database":{"resource_type":"external_instance","resource_path":"dt_demo"}}}}}'
```
`sslmode` defaults to `verify-full`; `disable` is for a local container only. With `verify-full`
against a server with a private CA, put the CA in `root_certificate_pem``pg_dump`, `psql` and
DuckDB attaches all verify against the system trust store plus that certificate.
Jobs then reach it as any data table: `ATTACH 'datatable://demo' AS d` from DuckDB, or
`datatable://demo` as the database of a PostgreSQL script, with `-- role <name>` to connect as a
data table role of that cluster.
Worth knowing while testing:
- A worker needs the `postgresql` and `duckdb` tags for those jobs
(`update config set config = jsonb_set(config, '{worker_tags}', …) where name = 'worker__default'`).
- DuckDB jobs load `libwindmill_duckdb_ffi_internal.so` by name, so a binary built into its own
`CARGO_TARGET_DIR` needs that library on `LD_LIBRARY_PATH`.
- Setup holds the lifecycle lock for its whole run, so a settings save during it waits.