From 122a3bd16422bf317161cb0e76f171eb777dc7d1 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Fri, 11 Sep 2026 11:59:27 +0200 Subject: [PATCH] fix(datatables): serialize roles going on with a stream starting Turning roles on looked for enabled triggers and live captures once, without a lock anything starting a stream also took. A trigger enabled in that window could have its listener connect before roles committed, and a healthy listener never checks again. Both transitions now serialize on one advisory lock: roles going on hold it exclusive while they look, and trigger create, edit and enable, and capture setup and ping hold it shared while they commit. Either the look sees the stream, or the listener connects after roles are committed and refuses. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/datatable_roles.rs | 54 +++++++++++++++++++ .../src/datatable_permissions.rs | 1 + backend/windmill-api/src/capture.rs | 6 +++ .../windmill-common/src/datatable_roles.rs | 16 ++++++ .../windmill-trigger-postgres/src/handler.rs | 18 ++++++- 5 files changed, 94 insertions(+), 1 deletion(-) diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index f709637750..9179095c6a 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -798,3 +798,57 @@ async fn roles_cannot_be_turned_on_while_a_trigger_streams_the_data_table( assert_eq!(resp.status(), 200, "{}", resp.text().await?); Ok(()) } + +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn roles_going_on_wait_for_a_trigger_being_enabled(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + sqlx::query( + "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}' + WHERE workspace_id = 'test-workspace'", + ) + .execute(&db) + .await?; + + // A trigger enable in flight: it holds the stream lock and its row is not committed yet, so a + // roles save that looked for streams now would miss it and its listener would connect to a + // data table it is about to be refused. + let mut enabling = db.begin().await?; + windmill_common::datatable_roles::lock_datatable_streams(&mut *enabling, false).await?; + sqlx::query( + r#"INSERT INTO postgres_trigger (path, script_path, is_flow, workspace_id, edited_by, + postgres_resource_path, replication_slot_name, publication_name, permissioned_as, mode) + VALUES ('u/test-user-2/racing_stream', 'u/test-user-2/s', false, 'wm-fork-dt', + 'test-user-2', 'datatable://main', 'slot_race', 'pub_race', 'u/test-user-2', + 'enabled')"#, + ) + .execute(&mut *enabling) + .await?; + + let server = ApiServer::start(db.clone()).await?; + let url = format!( + "http://localhost:{}/api/w/test-workspace/workspaces/datatable_permissions/main", + server.addr.port() + ); + let save = tokio::spawn( + authed(client().post(&url), "SECRET_TOKEN") + .json(&json!({"permissioned": true, "default_role": "admin", + "roles": [{"id": "admin", "tenants": ["*"]}]})) + .send(), + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + assert!( + !save.is_finished(), + "roles went on while a trigger was being enabled" + ); + enabling.commit().await?; + + let resp = save.await??; + assert_eq!(resp.status(), 400); + assert!( + resp.text() + .await? + .contains("wm-fork-dt/u/test-user-2/racing_stream"), + "the roles save missed the trigger enabled while it waited" + ); + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/datatable_permissions.rs b/backend/windmill-api-workspaces/src/datatable_permissions.rs index b04fb209e7..4014e81f1e 100644 --- a/backend/windmill-api-workspaces/src/datatable_permissions.rs +++ b/backend/windmill-api-workspaces/src/datatable_permissions.rs @@ -349,6 +349,7 @@ async fn set_datatable_permissions( // Turning roles on is refused while a replication stream reads this data table. One already // under roles cannot have any: the listener refuses to open a stream on it. if req.permissioned && governing.datatable.permissions.is_none() { + windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, true).await?; ensure_no_streams_reaching(&db, &governing).await?; } diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 2e2058ea80..6255b5eb13 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -567,6 +567,9 @@ async fn set_config( }; let mut tx = user_db.begin(&authed).await?; + if matches!(nc.trigger_kind, TriggerKind::Postgres) { + windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?; + } sqlx::query!( r#" @@ -614,6 +617,9 @@ async fn ping_config( )>, ) -> Result<()> { let mut tx = user_db.begin(&authed).await?; + if matches!(trigger_kind, TriggerKind::Postgres) { + windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?; + } sqlx::query!( r#" diff --git a/backend/windmill-common/src/datatable_roles.rs b/backend/windmill-common/src/datatable_roles.rs index 07e3e16114..3a7709ccd1 100644 --- a/backend/windmill-common/src/datatable_roles.rs +++ b/backend/windmill-common/src/datatable_roles.rs @@ -127,6 +127,22 @@ pub async fn lock_role_catalog(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) - Ok(()) } +/// A replication stream reads every row whatever a data table's roles grant. Turning roles on looks +/// for streams holding this exclusive; whatever can start a Postgres trigger or capture streaming +/// holds it shared on the transaction that commits it. So either the look sees the stream, or the +/// stream's listener connects after roles are committed and refuses. Held for the transaction. +pub async fn lock_datatable_streams(conn: &mut sqlx::PgConnection, exclusive: bool) -> Result<()> { + let lock = if exclusive { + "pg_advisory_xact_lock" + } else { + "pg_advisory_xact_lock_shared" + }; + sqlx::query(&format!("SELECT {lock}(hashtext('datatable_streams'))")) + .execute(conn) + .await?; + Ok(()) +} + /// Disclosure: returns every role's stored Postgres password in plaintext. Any server path that /// has to resolve or name a role may call it — including handlers open to a workspace member, who /// need the names — but callers MUST NOT let `pwd` reach a response, a log line, an audit record diff --git a/backend/windmill-trigger-postgres/src/handler.rs b/backend/windmill-trigger-postgres/src/handler.rs index 6bc23514dd..4294645b89 100644 --- a/backend/windmill-trigger-postgres/src/handler.rs +++ b/backend/windmill-trigger-postgres/src/handler.rs @@ -21,7 +21,7 @@ use windmill_common::{ use windmill_git_sync::DeployedObject; use windmill_api_auth::{check_scopes, ApiAuthed}; -use windmill_trigger::{Trigger, TriggerCrud, TriggerData}; +use windmill_trigger::{Trigger, TriggerCrud, TriggerData, TriggerMode}; use super::{ check_if_valid_publication_for_postgres_version, create_logical_replication_slot, @@ -74,6 +74,20 @@ impl TriggerCrud for PostgresTrigger { ensure_not_under_roles(db, workspace_id, &config.postgres_resource_path).await } + async fn authorize_set_trigger_mode( + &self, + _authed: &ApiAuthed, + tx: &mut PgConnection, + _workspace_id: &str, + _path: &str, + mode: &TriggerMode, + ) -> Result<()> { + if *mode != TriggerMode::Disabled { + windmill_common::datatable_roles::lock_datatable_streams(tx, false).await?; + } + Ok(()) + } + async fn create_trigger( &self, db: &DB, @@ -82,6 +96,7 @@ impl TriggerCrud for PostgresTrigger { w_id: &str, trigger: TriggerData, ) -> Result<()> { + windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?; let resolved_edited_by = trigger.base.resolve_edited_by(authed); let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed); let Self::TriggerConfigRequest { @@ -171,6 +186,7 @@ impl TriggerCrud for PostgresTrigger { path: &str, trigger: TriggerData, ) -> Result<()> { + windmill_common::datatable_roles::lock_datatable_streams(&mut *tx, false).await?; let resolved_edited_by = trigger.base.resolve_edited_by(authed); let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed); let Self::TriggerConfigRequest {