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) <noreply@anthropic.com>

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<Postgres>) -> 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(())
+}

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?;
     }
 

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#"

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

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<Self::TriggerConfigRequest>,
     ) -> 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<Self::TriggerConfigRequest>,
     ) -> 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 {