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

fix(datatable): provision the replication user on managed postgres (#10375)

Browse Source 
* fix(datatable): provision the replication user on managed postgres

* fix(datatable): serialize replication user provisioning and sync config schema

* fix(datatable): keep replication cleanup best-effort and self-heal a null password
This commit is contained in:
Ruben Fiszel
2026-07-28 12:17:09 +02:00
committed by GitHub
parent a544dfde9a
commit 1b6b2aa859
6 changed files with 160 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-settings/src/lib.rs
+17 -19
View File
@@ -1506,6 +1506,10 @@ struct CustomInstanceDbLogs {
db_connect: String,
#[serde(skip_serializing_if = "String::is_empty")]
grant_permissions: String,
#[serde(skip_serializing_if = "String::is_empty")]
replication_user: String,
#[serde(skip_serializing_if = "Option::is_none")]
replication_user_error: Option<String>,
}
async fn list_custom_instance_pg_databases(
@@ -1705,30 +1709,24 @@ async fn setup_custom_instance_pg_database_inner(
))
})?;
// The replication attribute lives on a dedicated role used by postgres trigger
// connections. The getter creates the role (with its stored password) when the
// migration couldn't.
if let Err(e) = windmill_common::utils::get_custom_pg_instance_replication_password(db).await {
tracing::error!("Failed to ensure custom_instance_replication_user exists: {e:#}");
}
if let Err(e) = client
.batch_execute(
"ALTER ROLE custom_instance_replication_user REPLICATION;
GRANT custom_instance_user TO custom_instance_replication_user;
ALTER ROLE custom_instance_user NOREPLICATION;",
)
.await
{
tracing::error!(
"Failed to grant replication permission to custom_instance_replication_user: {e:#}"
);
}
logs.grant_permissions = "OK".to_string();
drop(client); // /!\ Drop before joining to avoid deadlock
windmill_common::shutdown_pg_connection(join_handle).await?;
// Roles are cluster-wide, so the dedicated role used by postgres trigger connections is
// provisioned on the main pool rather than on the new database. Reported as its own step
// rather than failing the setup: without the role the database still serves datatables, only
// postgres triggers on them break.
match windmill_common::utils::ensure_custom_instance_replication_user(db).await {
Ok(()) => logs.replication_user = "OK".to_string(),
Err(e) => {
tracing::error!("Failed to provision custom_instance_replication_user: {e:#}");
logs.replication_user = "FAIL".to_string();
logs.replication_user_error = Some(e.to_string());
}
}
Ok(())
}
backend/windmill-api/openapi.yaml
+4
View File
@@ -29180,6 +29180,10 @@ components:
$ref: "#/components/schemas/LoggedWizardStatus"
grant_permissions:
$ref: "#/components/schemas/LoggedWizardStatus"
replication_user:
$ref: "#/components/schemas/LoggedWizardStatus"
replication_user_error:
type: string
CustomInstanceDbTag:
type: string
backend/windmill-api/src/live_migrations.rs
+30
View File
@@ -23,9 +23,39 @@ pub async fn custom_migrations(migrator: &mut CustomMigrator) -> Result<(), Erro
tracing::error!("Could not normalize custom_instance_user attributes: {err:#}");
}
if let Err(err) = ensure_custom_instance_replication_user(migrator).await {
tracing::error!(
"Could not provision custom_instance_replication_user: {err:#}. Postgres triggers on \
custom-instance datatables will not work until the role can open replication \
connections: grant the role owning DATABASE_URL either SUPERUSER or (PG 16+) the \
REPLICATION attribute. On AWS RDS this is handled by granting rds_replication, which \
requires no change; other managed providers are not covered"
);
}
Ok(())
}
// Converged on every boot, not once: the one-shot migration creates the role with the
// REPLICATION attribute, which managed postgres rejects outright, so instances set up before
// the provider-role fallback existed have no role at all. Scoped to instances that actually
// have a custom-instance database, so the rest never see the error.
async fn ensure_custom_instance_replication_user(
migrator: &mut CustomMigrator,
) -> Result<(), Error> {
let has_custom_instance_db = sqlx::query_scalar::<_, bool>(
"SELECT COALESCE(value->'databases', '{}'::jsonb) <> '{}'::jsonb
FROM global_settings WHERE name = 'custom_instance_pg_databases'",
)
.fetch_optional(migrator.connection())
.await?
.unwrap_or(false);
if !has_custom_instance_db {
return Ok(());
}
windmill_common::utils::ensure_custom_instance_replication_user(migrator.connection()).await
}
// Converged on every boot, not once: the one-shot migration swallows errors (it must not
// abort startup without superuser), and an older instance sharing the cluster can re-add
// the attribute. REPLICATION belongs only on custom_instance_replication_user.
backend/windmill-common/src/instance_config.rs
+4
View File
@@ -819,6 +819,10 @@ pub struct CustomInstanceDbLogs {
pub db_connect: String,
#[serde(skip_serializing_if = "String::is_empty")]
pub grant_permissions: String,
#[serde(skip_serializing_if = "String::is_empty")]
pub replication_user: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub replication_user_error: Option<String>,
}
// ---------------------------------------------------------------------------
backend/windmill-common/src/utils.rs
+92 -18
View File
@@ -1058,30 +1058,90 @@ pub async fn get_custom_pg_instance_password(db: &DB) -> Result<String> {
)
}
const REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL: &str = r#"
/// PL/pgSQL granting `custom_instance_replication_user` the ability to open replication
/// connections, inlined into the `DO` blocks below.
///
/// The REPLICATION attribute requires a real superuser on PG <= 15 (PG 16+ accepts
/// CREATEROLE + REPLICATION), and managed postgres never hands one out: on RDS the
/// capability is carried by the `rds_replication` role instead. Both privilege failures
/// raise `insufficient_privilege`, so the attribute is attempted first and the provider
/// role is the fallback.
const GRANT_REPLICATION_CAPABILITY_PLPGSQL: &str = r#"
BEGIN
EXECUTE 'ALTER ROLE custom_instance_replication_user REPLICATION';
EXCEPTION WHEN insufficient_privilege THEN
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'rds_replication') THEN
EXECUTE 'GRANT rds_replication TO custom_instance_replication_user';
ELSE
RAISE;
END IF;
END;
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN
EXECUTE 'GRANT custom_instance_user TO custom_instance_replication_user';
-- Stripping REPLICATION off custom_instance_user is a cleanup, so it is both
-- guarded and best-effort: the clause is superuser-only on PG <= 15 even when the
-- attribute is already unset, and failing it must not roll back the role above.
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user' AND rolreplication) THEN
BEGIN
EXECUTE 'ALTER ROLE custom_instance_user NOREPLICATION';
EXCEPTION WHEN insufficient_privilege THEN
NULL;
END;
END IF;
END IF;
"#;
/// `rotate_condition` decides when a new password is generated: always for the refresh
/// endpoint, only when the role or its stored password is missing for the boot converge.
/// Both variants are a single statement so they can run on any executor, and both are
/// atomic: a role that cannot be granted replication is rolled back rather than left
/// half-provisioned.
fn provision_replication_user_sql(rotate_condition: &str) -> String {
format!(
r#"
DO $$
DECLARE
pwd text;
BEGIN
SELECT gen_random_uuid()::text INTO pwd;
-- Same lock as get_custom_pg_instance_replication_password: without it, every API
-- replica booting onto a version that provisions the role races into CREATE USER,
-- and the losers abort on duplicate_object.
PERFORM pg_advisory_xact_lock(hashtext('custom_instance_replication_pwd'));
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user') THEN
EXECUTE format('ALTER USER custom_instance_replication_user WITH PASSWORD %L REPLICATION', pwd);
ELSE
EXECUTE format('CREATE USER custom_instance_replication_user WITH PASSWORD %L REPLICATION', pwd);
IF {rotate_condition} THEN
SELECT gen_random_uuid()::text INTO pwd;
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user') THEN
EXECUTE format('ALTER USER custom_instance_replication_user WITH PASSWORD %L', pwd);
ELSE
EXECUTE format('CREATE USER custom_instance_replication_user WITH PASSWORD %L', pwd);
END IF;
INSERT INTO global_settings (name, value)
VALUES ('custom_instance_replication_pwd', to_jsonb(pwd::text))
ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value;
END IF;
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN
GRANT custom_instance_user TO custom_instance_replication_user;
ALTER ROLE custom_instance_user NOREPLICATION;
END IF;
INSERT INTO global_settings (name, value)
VALUES ('custom_instance_replication_pwd', to_jsonb(pwd::text))
ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value;
{GRANT_REPLICATION_CAPABILITY_PLPGSQL}
END
$$;
"#;
"#
)
}
lazy_static::lazy_static! {
static ref REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL: String =
provision_replication_user_sql("TRUE");
// The password test mirrors REPLICATION_PWD_READ_SQL, whose readers flatten a NULL away: a row
// holding a JSON null reads back as no password, so it must rotate rather than count as
// provisioned and strand the getter.
static ref ENSURE_CUSTOM_INSTANCE_REPLICATION_USER_SQL: String = provision_replication_user_sql(
"NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user')
OR NOT EXISTS (SELECT 1 FROM global_settings
WHERE name = 'custom_instance_replication_pwd' AND value #>> '{}' IS NOT NULL)"
);
}
const REPLICATION_PWD_READ_SQL: &str =
"SELECT value #>> '{}' FROM global_settings WHERE name = 'custom_instance_replication_pwd'";
@@ -1093,12 +1153,26 @@ const REPLICATION_PWD_READ_SQL: &str =
/// Authorization: rotates a stored database credential and performs no authorization
/// itself — callers MUST restrict this to superadmin or internal server paths.
pub async fn refresh_custom_instance_replication_user_pwd(db: &DB) -> Result<()> {
sqlx::query(REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL)
sqlx::query(REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL.as_str())
.execute(db)
.await?;
Ok(())
}
/// Create `custom_instance_replication_user` if it is missing and (re)grant it replication,
/// leaving an existing password in place. Idempotent, so it can be converged on every boot.
///
/// Authorization: same contract as [`refresh_custom_instance_replication_user_pwd`] —
/// callers MUST restrict this to superadmin or internal server paths.
pub async fn ensure_custom_instance_replication_user<'c>(
executor: impl sqlx::PgExecutor<'c>,
) -> Result<()> {
sqlx::query(ENSURE_CUSTOM_INSTANCE_REPLICATION_USER_SQL.as_str())
.execute(executor)
.await?;
Ok(())
}
/// Authorization: returns a stored database credential and performs no authorization
/// itself — callers MUST restrict this to superadmin or internal server paths (mirrors
/// [`get_custom_pg_instance_password`]).
@@ -1127,7 +1201,7 @@ pub async fn get_custom_pg_instance_replication_password(db: &DB) -> Result<Stri
tx.commit().await?;
return Ok(pwd);
}
sqlx::query(REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL)
sqlx::query(ENSURE_CUSTOM_INSTANCE_REPLICATION_USER_SQL.as_str())
.execute(&mut *tx)
.await?;
let pwd = sqlx::query_scalar::<_, Option<String>>(REPLICATION_PWD_READ_SQL)
frontend/src/lib/components/workspaceSettings/CustomInstanceDbWizardModal.svelte
+13 -4
View File
@@ -165,17 +165,26 @@
title: 'Grant permissions to custom_instance_user',
status: status?.logs.grant_permissions,
description:
'Gives custom_instance_user the required permissions to use the database. custom_instance_user is already created during a migration and has an auto-generated password stored in global_settings.custom_instance_pg_databases.user_pwd. Postgres triggers use custom_instance_replication_user (password in global_settings.custom_instance_replication_pwd). These are the commands : \n\n' +
'Gives custom_instance_user the required permissions to use the database. custom_instance_user is already created during a migration and has an auto-generated password stored in global_settings.custom_instance_pg_databases.user_pwd. These are the commands : \n\n' +
`GRANT CONNECT ON DATABASE "${dbname}" TO custom_instance_user;\n` +
'GRANT USAGE ON SCHEMA public TO custom_instance_user;\n' +
'GRANT CREATE ON SCHEMA public TO custom_instance_user;\n' +
`GRANT CREATE ON DATABASE "${dbname}" TO custom_instance_user;\n` +
'ALTER DEFAULT PRIVILEGES IN SCHEMA public \n' +
' GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES\n TO custom_instance_user;\n' +
'ALTER ROLE custom_instance_user CREATEROLE;\n' +
'ALTER ROLE custom_instance_user CREATEROLE;'
},
{
title: 'Grant replication to custom_instance_replication_user',
status: status?.logs.replication_user,
description:
'Postgres triggers on custom-instance datatables connect as custom_instance_replication_user, whose password is stored in global_settings.custom_instance_replication_pwd. The role is cluster-wide, so it is created on the Windmill PostgreSQL instance rather than on this database : \n\n' +
'ALTER ROLE custom_instance_replication_user REPLICATION;\n' +
'GRANT custom_instance_user TO custom_instance_replication_user;\n' +
'ALTER ROLE custom_instance_user NOREPLICATION;'
'GRANT custom_instance_user TO custom_instance_replication_user;\n\n' +
'Setting REPLICATION requires a superuser on PostgreSQL 15 and older. Managed instances never grant one, so on AWS RDS Windmill falls back to GRANT rds_replication TO custom_instance_replication_user. The database stays usable for datatables if this step fails, but postgres triggers on them do not.' +
(status?.logs.replication_user_error
? `\n\nError: ${status.logs.replication_user_error}`
: '')
}
],
status?.error ?? undefined