diff --git a/backend/windmill-trigger-postgres/src/handler.rs b/backend/windmill-trigger-postgres/src/handler.rs index 92dd2e1e19..6bc23514dd 100644 --- a/backend/windmill-trigger-postgres/src/handler.rs +++ b/backend/windmill-trigger-postgres/src/handler.rs @@ -25,7 +25,8 @@ use windmill_trigger::{Trigger, TriggerCrud, TriggerData}; use super::{ check_if_valid_publication_for_postgres_version, create_logical_replication_slot, - create_pg_publication, drop_publication, generate_random_string, get_default_pg_connection, + create_pg_publication, drop_publication, ensure_not_under_roles, generate_random_string, + get_default_pg_connection, mapper::{Mapper, MappingInfo}, PostgresConfig, PostgresConfigRequest, PostgresPublicationReplication, PostgresTrigger, PublicationData, Relations, Slot, SlotList, TableToTrack, TemplateScript, TestPostgresConfig, @@ -64,6 +65,15 @@ impl TriggerCrud for PostgresTrigger { DeployedObject::PostgresTrigger { path, parent_path } } + async fn validate_config( + &self, + db: &DB, + config: &Self::TriggerConfigRequest, + workspace_id: &str, + ) -> Result<()> { + ensure_not_under_roles(db, workspace_id, &config.postgres_resource_path).await + } + async fn create_trigger( &self, db: &DB, diff --git a/backend/windmill-trigger-postgres/src/lib.rs b/backend/windmill-trigger-postgres/src/lib.rs index 41e82614a8..25b19e6c1b 100644 --- a/backend/windmill-trigger-postgres/src/lib.rs +++ b/backend/windmill-trigger-postgres/src/lib.rs @@ -374,6 +374,32 @@ pub async fn get_raw_postgres_connection( Ok(client) } +/// A replication stream reads every row of every table whatever the data table's roles grant, so +/// the two don't mix: a data table under roles takes no triggers or captures, and roles cannot be +/// turned on while one is enabled on it. +pub async fn ensure_not_under_roles( + db: &DB, + w_id: &str, + postgres_resource_path: &str, +) -> Result<()> { + let Some(datatable_name) = postgres_resource_path.strip_prefix("datatable://") else { + return Ok(()); + }; + if windmill_common::workspaces::resolve_governing_datatable(db, w_id, datatable_name) + .await? + .datatable + .permissions + .is_some() + { + return Err(Error::BadRequest(format!( + "Data table '{datatable_name}' is under roles, and a Postgres trigger or capture \ + cannot read one: a replication stream sees every row whatever the roles grant. \ + Turn its roles off to stream it." + ))); + } + Ok(()) +} + pub async fn resolve_postgres_resource( authed: &ApiAuthed, user_db: Option, @@ -382,21 +408,7 @@ pub async fn resolve_postgres_resource( w_id: &str, ) -> Result { if let Some(datatable_name) = postgres_resource_path.strip_prefix("datatable://") { - // A replication stream reads every row of every table whatever the data table's roles - // grant, so the two don't mix: a data table under roles takes no triggers or captures, and - // roles cannot be turned on while one is enabled on it. - if windmill_common::workspaces::resolve_governing_datatable(db, w_id, datatable_name) - .await? - .datatable - .permissions - .is_some() - { - return Err(Error::BadRequest(format!( - "Data table '{datatable_name}' is under roles, and a Postgres trigger or capture \ - cannot read one: a replication stream sees every row whatever the roles grant. \ - Turn its roles off to stream it." - ))); - } + ensure_not_under_roles(db, w_id, postgres_resource_path).await?; // Trigger connections (publication/slot management + logical replication) run // as the dedicated replication user on custom-instance databases. let resource_value = diff --git a/backend/windmill-trigger-postgres/src/listener.rs b/backend/windmill-trigger-postgres/src/listener.rs index 14d348b5f2..876b75661e 100644 --- a/backend/windmill-trigger-postgres/src/listener.rs +++ b/backend/windmill-trigger-postgres/src/listener.rs @@ -20,7 +20,8 @@ use windmill_common::{ use windmill_trigger::{listener::ListeningTrigger, trigger_helpers::TriggerJobArgs, Listener}; use super::{ - drop_publication, get_default_pg_connection, get_raw_postgres_connection, + drop_publication, ensure_not_under_roles, get_default_pg_connection, + get_raw_postgres_connection, handler::drop_logical_replication_slot, relation::RelationConverter, replication_message::{ @@ -135,8 +136,8 @@ impl PostgresSimpleClient { /// Resolves the Postgres resource, validates that the configured publication and /// replication slot still exist, and opens a fresh logical replication stream. /// -/// Returns `Error::BadConfig` when the publication or slot is missing (an -/// unrecoverable misconfiguration). Any other error is treated as transient +/// Returns `Error::BadConfig` when the publication or slot is missing, or the +/// data table is under roles (unrecoverable misconfigurations). Any other error is treated as transient /// (connection refused, network interruption, ...) and is retried by the caller. /// The resource is re-resolved on every call so credential rotations are picked /// up across reconnections. @@ -149,6 +150,14 @@ async fn connect_logical_replication_stream( let PostgresConfig { postgres_resource_path, publication_name, replication_slot_name, .. } = trigger_config; + // Retrying cannot lift roles, so this disables the trigger like a missing slot does. + ensure_not_under_roles(db, workspace_id, postgres_resource_path) + .await + .map_err(|e| match e { + Error::BadRequest(msg) => Error::BadConfig(msg), + e => e, + })?; + let database = resolve_postgres_resource( authed, Some(UserDB::new(db.clone())),