From fa1e8bd3aa1ebfef324305989591e17741960b15 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Fri, 11 Sep 2026 10:35:36 +0200 Subject: [PATCH] refactor(datatables): keep Postgres triggers and data table roles apart A replication stream reads every row of every table whatever the data table's roles grant, and its listener checks access only when it connects. Rather than chase every way access can change and bounce the streams each one affects, a data table now carries one or the other: - a Postgres trigger or capture cannot be created on, or connect to, a data table under roles; - roles cannot be turned on while an enabled trigger or a live capture reads the data table, its own or a fork's through its pointer. The refusal names each one to disable. This removes the stream bounces on roles edits and on data table and workspace deletion, and the trigger gate that admitted admins. The fork schema baseline fix from the same review round is kept. Co-Authored-By: Claude Opus 5 (1M context) --- ...8961cfbd0171f835faa756762919c021bfd2b.json | 24 +++ .../tests/datatable_roles.rs | 156 ++++++------------ .../src/datatable_permissions.rs | 111 ++++++------- .../windmill-api-workspaces/src/workspaces.rs | 31 +--- .../src/workspaces_extra.rs | 42 ++--- backend/windmill-trigger-postgres/src/lib.rs | 28 ++-- 6 files changed, 148 insertions(+), 244 deletions(-) create mode 100644 backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json diff --git a/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json b/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json new file mode 100644 index 0000000000..b452b6fc16 --- /dev/null +++ b/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id || '/' || path AS \"stream!\" FROM postgres_trigger\n WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE\n AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3))\n UNION ALL\n SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config\n WHERE workspace_id = $1 AND trigger_kind = 'postgres'\n AND last_client_ping > now() - interval '10 seconds'\n AND (trigger_config->>'postgres_resource_path' = $2\n OR starts_with(trigger_config->>'postgres_resource_path', $3))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "stream!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b" +} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index f91495ecef..e619e73f23 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -672,123 +672,59 @@ async fn a_clone_stamp_is_carried_but_its_schema_baseline_advances( Ok(()) } -/// Live replication listeners on `main`: a fork's through its pointer, the governing workspace's -/// own (in the `?role=` form), and a fork capture — plus one on another data table, which no -/// deletion of `main` may touch. Each is held by a server, as a running stream would be. -async fn plant_live_streams(db: &Pool) -> anyhow::Result<()> { +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn roles_cannot_be_turned_on_while_a_trigger_streams_the_data_table( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + // A replication stream reads every row whatever the roles grant, so a data table carries one + // or the other. An enabled trigger on it — here a fork's, through its pointer — keeps roles + // from being turned on, and disabling it is what lets them on. + sqlx::query( + "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}' + WHERE workspace_id = 'test-workspace'", + ) + .execute(&db) + .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, - server_id) - VALUES - ('u/test-user-2/fork_stream', 'u/test-user-2/s', false, 'wm-fork-dt', 'test-user-2', - 'datatable://main', 'slot_fork', 'pub_fork', 'u/test-user-2', 'srv'), - ('u/test-user/own_stream', 'u/test-user/s', false, 'test-workspace', 'test-user', - 'datatable://main?role=admin', 'slot_own', 'pub_own', 'u/test-user', 'srv'), - ('u/test-user-2/unrelated', 'u/test-user-2/s', false, 'wm-fork-dt', 'test-user-2', - 'datatable://other', 'slot_other', 'pub_other', 'u/test-user-2', 'srv')"#, + postgres_resource_path, replication_slot_name, publication_name, permissioned_as, mode) + VALUES ('u/test-user-2/fork_stream', 'u/test-user-2/s', false, 'wm-fork-dt', + 'test-user-2', 'datatable://main', 'slot_fork', 'pub_fork', 'u/test-user-2', + 'enabled')"#, ) - .execute(db) + .execute(&db) .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 turn_on = json!({"permissioned": true, "default_role": "admin", + "roles": [{"id": "admin", "tenants": ["*"]}]}); + + let resp = authed(client().post(&url), "SECRET_TOKEN") + .json(&turn_on) + .send() + .await?; + assert_eq!(resp.status(), 400); + assert!( + resp.text() + .await? + .contains("wm-fork-dt/u/test-user-2/fork_stream"), + "the refusal does not name the trigger to disable" + ); + sqlx::query( - r#"INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email, - trigger_config, server_id) - VALUES ('wm-fork-dt', 'u/test-user-2/s', false, 'postgres', 'u/test-user-2', - 'test2@windmill.dev', '{"postgres_resource_path": "datatable://main"}', 'srv')"#, + "UPDATE postgres_trigger SET mode = 'disabled' WHERE path = 'u/test-user-2/fork_stream'", ) - .execute(db) - .await?; - Ok(()) -} - -/// Which server holds each planted listener; `None` is a bounced one, free for a reconnect. -async fn stream_servers( - db: &Pool, -) -> anyhow::Result>> { - let rows: Vec<(String, Option)> = sqlx::query_as( - "SELECT path, server_id FROM postgres_trigger - UNION ALL - SELECT 'capture:' || path, server_id FROM capture_config", - ) - .fetch_all(db) - .await?; - Ok(rows.into_iter().collect()) -} - -#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] -async fn deleting_a_governing_data_table_bounces_the_streams_reading_it( - db: Pool, -) -> anyhow::Result<()> { - initialize_tracing().await; - // A replication stream keeps the connection it opened while its entry resolved. Unbounced, a - // fork's stream reads on through a pointer that no longer resolves to anything. - plant_live_streams(&db).await?; - let server = ApiServer::start(db.clone()).await?; - let port = server.addr.port(); - - let resp = authed( - client().post(format!( - "http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config" - )), - "SECRET_TOKEN", - ) - // The way the settings-sync CLI saves: the new document alone, with no deletion hint. - .json(&json!({"settings": {"datatables": {}}})) - .send() + .execute(&db) .await?; + let resp = authed(client().post(&url), "SECRET_TOKEN") + .json(&turn_on) + .send() + .await?; assert_eq!(resp.status(), 200, "{}", resp.text().await?); - - let servers = stream_servers(&db).await?; - for bounced in [ - "u/test-user-2/fork_stream", - "u/test-user/own_stream", - "capture:u/test-user-2/s", - ] { - assert_eq!( - servers[bounced], None, - "{bounced} kept streaming: {servers:?}" - ); - } - assert_eq!( - servers["u/test-user-2/unrelated"].as_deref(), - Some("srv"), - "a stream on another data table was bounced: {servers:?}" - ); - Ok(()) -} - -#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] -async fn deleting_a_governing_workspace_bounces_its_forks_streams( - db: Pool, -) -> anyhow::Result<()> { - initialize_tracing().await; - // The governing workspace's own listeners go with it; each fork's stays behind on the - // connection it opened while this workspace still governed the pointer. - plant_live_streams(&db).await?; - let server = ApiServer::start(db.clone()).await?; - let port = server.addr.port(); - - let resp = authed( - client().delete(format!( - "http://localhost:{port}/api/workspaces/delete/test-workspace" - )), - "SECRET_TOKEN", - ) - .send() - .await?; - assert_eq!(resp.status(), 200, "{}", resp.text().await?); - - let servers = stream_servers(&db).await?; - for bounced in ["u/test-user-2/fork_stream", "capture:u/test-user-2/s"] { - assert_eq!( - servers[bounced], None, - "{bounced} kept streaming: {servers:?}" - ); - } - assert_eq!( - servers["u/test-user-2/unrelated"].as_deref(), - Some("srv"), - "a stream on another data table was bounced: {servers:?}" - ); Ok(()) } diff --git a/backend/windmill-api-workspaces/src/datatable_permissions.rs b/backend/windmill-api-workspaces/src/datatable_permissions.rs index 1457be5322..b04fb209e7 100644 --- a/backend/windmill-api-workspaces/src/datatable_permissions.rs +++ b/backend/windmill-api-workspaces/src/datatable_permissions.rs @@ -346,6 +346,12 @@ 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() { + ensure_no_streams_reaching(&db, &governing).await?; + } + let permissions = if req.permissioned { let catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?; let mut roles: BTreeMap = BTreeMap::new(); @@ -436,15 +442,6 @@ async fn set_datatable_permissions( } } - // A live replication stream holds a connection it opened under the old decision. Bouncing the - // rows makes every listener reconnect and re-authorize. - restart_streams_reaching( - &mut *db.acquire().await?, - &governing.workspace_id, - &governing.name, - ) - .await?; - windmill_common::feature_usage::log_feature_usage( "datatable", "roles_toggled", @@ -458,75 +455,61 @@ async fn set_datatable_permissions( }) } -/// Make every Postgres trigger and capture reading this data table reconnect, so a revoked tenant -/// stops streaming rather than living on inside an already-open replication connection. -pub(crate) async fn restart_streams_reaching( - conn: &mut sqlx::PgConnection, - governing_workspace_id: &str, - governing_name: &str, -) -> Result<()> { +/// Refuse to put a data table under roles while a Postgres trigger or capture streams it. A +/// replication stream reads every row whatever the roles grant, so a data table carries one or the +/// other; the listener side refuses a data table already under roles. +async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) -> Result<()> { // Every workspace holding an entry that resolves here, under the name it calls it: the // governing one, plus each fork pointing at it. A fork's trigger names its own local entry, so - // filtering on the governing workspace alone would leave its stream running on the connection - // it already opened under the old decision — which is the one window this function exists to - // close. - let mut reached = vec![( - governing_workspace_id.to_string(), - governing_name.to_string(), - )]; + // looking in the governing workspace alone would miss every stream a fork opened. + let mut reached = vec![(governing.workspace_id.clone(), governing.name.clone())]; let pointers = sqlx::query!( r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" FROM workspace_settings ws CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt WHERE dt.value->'reference'->>'workspace_id' = $1 AND dt.value->'reference'->>'datatable' = $2"#, - governing_workspace_id, - governing_name, + &governing.workspace_id, + &governing.name, ) - .fetch_all(&mut *conn) + .fetch_all(db) .await?; reached.extend(pointers.into_iter().map(|r| (r.workspace_id, r.datatable))); - restart_streams_named(conn, reached).await -} -/// Make every Postgres trigger and capture reading one of `entries` — each a workspace and the name -/// that workspace gives the data table — drop its connection and re-resolve the data table as it -/// now stands. A caller deciding inside a transaction passes it, so the bounce and the decision -/// become visible together and no listener reconnects in between. -pub(crate) async fn restart_streams_named( - conn: &mut sqlx::PgConnection, - entries: Vec<(String, String)>, -) -> Result<()> { - for (w_id, name) in entries { + let mut streams = Vec::new(); + for (w_id, name) in reached { let reference = format!("datatable://{name}"); - let prefix = format!("{reference}?%"); - - sqlx::query!( - "UPDATE postgres_trigger SET server_id = NULL, last_server_ping = NULL - WHERE workspace_id = $1 - AND (postgres_resource_path = $2 OR postgres_resource_path LIKE $3)", - &w_id, - &reference, - &prefix, - ) - .execute(&mut *conn) - .await?; - - // A capture keeps the reference inside its `trigger_config` blob rather than in a column - // of its own, and only a postgres capture has one there at all. - sqlx::query!( - "UPDATE capture_config SET server_id = NULL, last_server_ping = NULL - WHERE workspace_id = $1 AND trigger_kind = 'postgres' - AND (trigger_config->>'postgres_resource_path' = $2 - OR trigger_config->>'postgres_resource_path' LIKE $3)", - &w_id, - &reference, - &prefix, - ) - .execute(&mut *conn) - .await?; + let with_query = format!("{reference}?"); + // A suspended trigger keeps its listener, so only a disabled one is not streaming; a + // capture streams for as long as its client keeps pinging. + streams.extend( + sqlx::query_scalar!( + r#"SELECT workspace_id || '/' || path AS "stream!" FROM postgres_trigger + WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE + AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3)) + UNION ALL + SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config + WHERE workspace_id = $1 AND trigger_kind = 'postgres' + AND last_client_ping > now() - interval '10 seconds' + AND (trigger_config->>'postgres_resource_path' = $2 + OR starts_with(trigger_config->>'postgres_resource_path', $3))"#, + &w_id, + &reference, + &with_query, + ) + .fetch_all(db) + .await?, + ); + } + if !streams.is_empty() { + return Err(Error::BadRequest(format!( + "Data table '{}' cannot be put under roles while a Postgres trigger or capture streams \ + it: a replication stream reads every row whatever the roles grant. Disable them \ + first: {}", + governing.name, + streams.join(", ") + ))); } - Ok(()) } diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 6ac96e3f0b..ab81e50b9d 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3871,18 +3871,6 @@ async fn edit_datatable_config( } } - // Every entry this save removes, derived rather than taken from `deleted_datatables`: that list - // is a hint the settings-sync CLI never sends, and the stranded-pointer warning and the stream - // bounce below must run for a removal whether or not the caller named it. - let removed: Vec = old_datatables - .keys() - .filter(|name| { - !new_config.settings.datatables.contains_key(*name) - && !new_config.renames.iter().any(|r| &r.from == *name) - }) - .cloned() - .collect(); - let config: serde_json::Value = serde_json::to_value(new_config.settings) .map_err(|err| Error::internal_err(err.to_string()))?; @@ -3922,7 +3910,7 @@ async fn edit_datatable_config( // A deletion cannot be followed the same way — there is nothing to point at any more. Read who // is left stranded so the caller is told, the way deleting a workspace does. let mut stranded: Vec = Vec::new(); - for name in &removed { + for name in &new_config.deleted_datatables { let rows = sqlx::query!( r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" FROM workspace_settings ws @@ -3942,23 +3930,6 @@ async fn edit_datatable_config( ); } - // A stream reading a deleted entry keeps the connection it opened while the entry resolved — - // this workspace's own, and every fork's through its pointer. Bounced in this transaction, so - // a listener that reconnects finds the entry gone instead of streaming on. - crate::datatable_permissions::restart_streams_named( - &mut *tx, - removed - .iter() - .map(|name| (w_id.clone(), name.clone())) - .chain( - stranded - .iter() - .map(|s| (s.workspace_id.clone(), s.datatable.clone())), - ) - .collect(), - ) - .await?; - tx.commit().await?; for substrate in created_substrates { diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 371b80555c..1cb5188e5b 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -995,6 +995,21 @@ pub(crate) async fn delete_workspace( // but the destructive cleanup itself runs only after the commit below: a delete that // fails mid-way must never leave a live workspace with its fork data destroyed and no // registry row to retry from. Read-only: nothing is dropped here. + // Read before the delete: another workspace's data table entry can point at one of this + // workspace's, and deleting the workspace it names leaves that pointer resolving to nothing. + // Nothing sweeps them — turning them back into copies would hand each fork the database + // outright — so the deleter is told which data tables they just stranded. + let stranded_pointers = sqlx::query!( + r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" + FROM workspace_settings ws + CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt + WHERE dt.value->'reference'->>'workspace_id' = $1 + ORDER BY ws.workspace_id, dt.key"#, + &w_id, + ) + .fetch_all(&db) + .await + .unwrap_or_default(); let fork_ducklake_cleanups = prepare_fork_ducklake_cleanups(&db, &w_id, None) .await @@ -1250,33 +1265,6 @@ pub(crate) async fn delete_workspace( None, ) .await?; - // Who points here, read after the delete above and inside this transaction. A fork writes its - // pointer in the transaction that inserts it, and that insert key-share locks this row through - // the parent foreign key: so the fork either committed before the delete and is seen here, or - // waits on it and then fails on the missing parent. No pointer can escape this list. Nothing - // sweeps them afterwards — turning them back into copies would hand each fork the database - // outright — so the deleter is told which data tables they stranded. - let stranded_pointers = sqlx::query!( - r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!" - FROM workspace_settings ws - CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt - WHERE dt.value->'reference'->>'workspace_id' = $1 - ORDER BY ws.workspace_id, dt.key"#, - &w_id, - ) - .fetch_all(&mut *tx) - .await?; - // Each fork pointing here keeps the replication stream it opened while this workspace - // governed it. Bounced in this transaction, so a listener that reconnects finds its pointer - // dangling instead of streaming on. - crate::datatable_permissions::restart_streams_named( - &mut *tx, - stranded_pointers - .iter() - .map(|r| (r.workspace_id.clone(), r.datatable.clone())) - .collect(), - ) - .await?; tx.commit().await?; // Physical ducklake-namespace cleanup, post-commit, from the pre-read snapshot: fork diff --git a/backend/windmill-trigger-postgres/src/lib.rs b/backend/windmill-trigger-postgres/src/lib.rs index c77f7ce0d3..41e82614a8 100644 --- a/backend/windmill-trigger-postgres/src/lib.rs +++ b/backend/windmill-trigger-postgres/src/lib.rs @@ -11,10 +11,7 @@ use serde::{Deserialize, Deserializer, Serialize}; use serde_json::value::RawValue; use sqlx::FromRow; use windmill_api_auth::ApiAuthed; -use windmill_common::workspaces::{ - ensure_datatable_admin_access, get_datatable_replication_resource_from_db_unchecked, - DatatableAccess, -}; +use windmill_common::workspaces::get_datatable_replication_resource_from_db_unchecked; use windmill_common::{ db::UserDB, error::{to_anyhow, Error, Result}, @@ -386,15 +383,20 @@ pub async fn resolve_postgres_resource( ) -> 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 it is not something a role can be tenanted into: only someone who could have - // connected as `admin` may open one. - ensure_datatable_admin_access( - db, - w_id, - datatable_name, - &DatatableAccess::Authed(authed.to_authed_ref()), - ) - .await?; + // 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." + ))); + } // Trigger connections (publication/slot management + logical replication) run // as the dedicated replication user on custom-instance databases. let resource_value =