fix(datatables): bounce the streams reading a data table when it is deleted

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

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<Postgres>) -> 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<Postgres>,
+) -> anyhow::Result<std::collections::HashMap<String, Option<String>>> {
+    let rows: Vec<(String, Option<String>)> = 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<Postgres>,
+) -> 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<Postgres>,
+) -> 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(())
+}

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

backend/windmill-api-workspaces/src/workspaces.rs

@@ -3931,6 +3931,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 {

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