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<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(())
+}
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 f1a315f787..69a7bae3cf 100644
--- a/backend/windmill-api-workspaces/src/workspaces.rs
+++ b/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 {
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