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 =