From c27ae4e0afb36f7b53d801786d330a3bf6bf2877 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 10 Sep 2026 18:49:05 +0200 Subject: [PATCH] fix(datatables): bounce the streams reading a data table when it is deleted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deleting a governing data table, or the workspace that holds it, only collected the fork pointers it stranded, for the warning. A Postgres trigger or capture already streaming through one of those pointers kept the replication connection it opened while the pointer still resolved, so it went on dispatching the governing database's rows after the fork lost access — until its connection happened to restart. The governing workspace's own streams on a deleted entry did the same. Both deletion paths now bounce the affected listeners inside their own transaction, through the helper a permission change already uses, so a listener that reconnects re-resolves the entry and finds it gone. The helper is split so a caller can pass the (workspace, local name) pairs it already holds. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/datatable_roles.rs | 120 ++++++++++++++++++ .../src/datatable_permissions.rs | 39 ++++-- .../windmill-api-workspaces/src/workspaces.rs | 18 +++ .../src/workspaces_extra.rs | 11 ++ 4 files changed, 178 insertions(+), 10 deletions(-) diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 787bd950fd..43c768cb73 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -598,3 +598,123 @@ async fn a_data_table_under_roles_is_not_copied_into_a_fork( ); 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::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')"#, + ) + .execute(db) + .await?; + 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')"#, + ) + .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", + ) + .json(&json!({"settings": {"datatables": {}}, "renames": [], "deleted_datatables": ["main"]})) + .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 a1493c4b98..1457be5322 100644 --- a/backend/windmill-api-workspaces/src/datatable_permissions.rs +++ b/backend/windmill-api-workspaces/src/datatable_permissions.rs @@ -438,7 +438,12 @@ 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(&db, &governing).await?; + restart_streams_reaching( + &mut *db.acquire().await?, + &governing.workspace_id, + &governing.name, + ) + .await?; windmill_common::feature_usage::log_feature_usage( "datatable", @@ -456,29 +461,43 @@ 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( - db: &DB, - governing: &GoverningDatatable, + conn: &mut sqlx::PgConnection, + governing_workspace_id: &str, + governing_name: &str, ) -> 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.clone(), governing.name.clone())]; + let mut reached = vec![( + governing_workspace_id.to_string(), + governing_name.to_string(), + )]; 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(db) + .fetch_all(&mut *conn) .await?; reached.extend(pointers.into_iter().map(|r| (r.workspace_id, r.datatable))); + restart_streams_named(conn, reached).await +} - for (w_id, name) in reached { +/// 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 reference = format!("datatable://{name}"); let prefix = format!("{reference}?%"); @@ -490,7 +509,7 @@ pub(crate) async fn restart_streams_reaching( &reference, &prefix, ) - .execute(db) + .execute(&mut *conn) .await?; // A capture keeps the reference inside its `trigger_config` blob rather than in a column @@ -504,7 +523,7 @@ pub(crate) async fn restart_streams_reaching( &reference, &prefix, ) - .execute(db) + .execute(&mut *conn) .await?; } diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index c5567373db..f9f22fcaa6 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3876,6 +3876,24 @@ 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, + new_config + .deleted_datatables + .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 1cb5188e5b..d66f2a13b0 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -1265,6 +1265,17 @@ pub(crate) async fn delete_workspace( None, ) .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