fix(datatables): a deletion takes only the permissions it owns, and the row outlives its owner closed

backend/.sqlx/query-11f66d88f9374cab8dd59132ffb63f3fef248f52d7518389ac52eb3f515d68ac.json

@@ -0,0 +1,44 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id AS \"workspace_id!\", datatable AS \"datatable!\", name AS \"name!\",\n                  code_up AS \"code_up!\", code_down\n           FROM datatable_migrations\n           WHERE code_up LIKE '%-- role %' OR code_down LIKE '%-- role %'\n           ORDER BY workspace_id, datatable, timestamp",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "name!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "code_up!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "code_down",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "11f66d88f9374cab8dd59132ffb63f3fef248f52d7518389ac52eb3f515d68ac"
+}

backend/.sqlx/query-3abcd060b4e1865f17e73ae7e17b0ba654f21458fb4d2d0a55e5c3eab29bb279.json

@@ -26,7 +26,7 @@
     },
     "nullable": [
       false,
-      false,
+      true,
       false
     ]
   },

backend/.sqlx/query-82f7f52f60868a3aa659f2bee6b41f357a09e2c94792542287cbbc810451c47a.json

@@ -26,7 +26,7 @@
     },
     "nullable": [
       false,
-      false,
+      true,
       false
     ]
   },

backend/.sqlx/query-842a5c2694c996897f17e763498bbefaef59a77846cf63e376ba544f6dd9cf65.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO datatable_database_permissions (database_key, owner_workspace_id, permissions)\n           VALUES ($1, $2, $3)\n           ON CONFLICT (database_key) DO UPDATE\n           SET permissions = EXCLUDED.permissions,\n               owner_workspace_id = COALESCE(datatable_database_permissions.owner_workspace_id, EXCLUDED.owner_workspace_id),\n               updated_at = now()",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "842a5c2694c996897f17e763498bbefaef59a77846cf63e376ba544f6dd9cf65"
+}

backend/.sqlx/query-852c7619569dd676b24319267c3ba1ea98aae07a9c2e30de79b3db411e1ef3e9.json

@@ -26,7 +26,7 @@
     },
     "nullable": [
       false,
-      false,
+      true,
       false
     ]
   },

backend/.sqlx/query-95671af1034d05d58fb6e03e555b35534824748db021100a95eb69078d6e0f74.json

@@ -26,7 +26,7 @@
     },
     "nullable": [
       false,
-      false,
+      true,
       false
     ]
   },

backend/.sqlx/query-9deb2c34f0dd9a3c35d48b671b9704fbda8f4429f9c1c97a3f6e6a80d39d335c.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT pg_advisory_xact_lock(hashtext('datatable_database_permissions_owner:' || $1))",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "pg_advisory_xact_lock",
+        "type_info": "Void"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "9deb2c34f0dd9a3c35d48b671b9704fbda8f4429f9c1c97a3f6e6a80d39d335c"
+}

backend/.sqlx/query-f17b405bb47b4633675b75caffe25d51261c12c016218df01b2897abe94096fe.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO datatable_database_permissions (database_key, owner_workspace_id, permissions)\n           VALUES ($1, $2, $3)\n           ON CONFLICT (database_key) DO UPDATE SET permissions = EXCLUDED.permissions, updated_at = now()",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Varchar",
-        "Jsonb"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "f17b405bb47b4633675b75caffe25d51261c12c016218df01b2897abe94096fe"
-}

backend/migrations/20260906215123_datatable_database_permissions.up.sql

@@ -4,10 +4,12 @@
 -- database_key: 'instance:<dbname>' for an instance database, 'pg:<sha256 of
 -- host, port and dbname>' for a resource-backed one. The tenants named in
 -- `permissions` are principals of owner_workspace_id, and only its admins manage
--- the row.
+-- the row. A row whose owner was deleted keeps governing its database with no
+-- owner: every role is refused and only a superadmin reaches it, until one opts
+-- out or saves it from a workspace that then becomes the owner.
 CREATE TABLE datatable_database_permissions (
     database_key TEXT PRIMARY KEY,
-    owner_workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
+    owner_workspace_id VARCHAR(50) REFERENCES workspace(id) ON DELETE SET NULL,
     permissions JSONB NOT NULL,
     updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
 );

backend/windmill-api-integration-tests/tests/datatable_tenants.rs

@@ -518,6 +518,34 @@ async fn a_save_names_only_what_exists(db: Pool<Postgres>) -> anyhow::Result<()>
     assert_eq!(status, 400, "{text}");
     assert!(text.contains("'add_orders' (role 'analyst')"), "{text}");
 
+    // The roles are the database's: a migration of another entry reaching it,
+    // here an alias in the same workspace, is stranded all the same.
+    sqlx::query(
+        r#"UPDATE workspace_settings
+           SET datatable = jsonb_set(datatable, '{datatables,alias}',
+               '{"database": {"resource_type": "instance", "resource_path": "dt_main"}}')
+           WHERE workspace_id = 'test-workspace'"#,
+    )
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "UPDATE datatable_migrations SET datatable = 'alias' WHERE workspace_id = 'test-workspace'",
+    )
+    .execute(&db)
+    .await?;
+    let (status, text) = preview(json!({ "enabled": true,
+        "roles": [{ "name": "admin", "tenants": [] }]
+    }))
+    .await;
+    assert_eq!(status, 400, "{text}");
+    assert!(
+        text.contains("test-workspace/alias: 'add_orders' (role 'analyst')"),
+        "{text}"
+    );
+    sqlx::query("DELETE FROM datatable_migrations WHERE workspace_id = 'test-workspace'")
+        .execute(&db)
+        .await?;
+
     // Turning permissions off ignores the submitted roles and is never refused,
     // stale tenant or not: it gets as far as the database this test lacks.
     let (_, text) = preview(json!({ "enabled": false, "roles": [
@@ -544,3 +572,181 @@ async fn a_save_names_only_what_exists(db: Pool<Postgres>) -> anyhow::Result<()>
 
     Ok(())
 }
+
+/// Deleting a workspace takes only the roles it owns with it. A fork that held a
+/// copy of the parent's data table owns nothing there, so its deletion leaves the
+/// parent's row alone; the owner's own deletion drops its logins but leaves the
+/// row, ownerless and closed: every role refused, only a superadmin through.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn a_deletion_takes_only_the_permissions_it_owns(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    plant_main(&db, "test-workspace").await;
+    plant_permissions(&db, MAIN_KEY, "test-workspace", &["u/test-user-3"]).await;
+    sqlx::query(
+        "INSERT INTO workspace (id, name, owner, parent_workspace_id)
+         VALUES ('wm-fork-t', 'wm-fork-t', 'test2@windmill.dev', 'test-workspace')",
+    )
+    .execute(&db)
+    .await?;
+    plant_main(&db, "wm-fork-t").await;
+    sqlx::query(
+        "INSERT INTO usr (workspace_id, email, username, is_admin, role) VALUES
+           ('wm-fork-t', 'test2@windmill.dev', 'test-user-2', true, 'Admin'),
+           ('wm-fork-t', 'test3@windmill.dev', 'test-user-3', false, 'User')",
+    )
+    .execute(&db)
+    .await?;
+
+    // The fork's owner deletes it: the parent's row is untouched.
+    let resp = authed(
+        client().delete(format!(
+            "http://localhost:{port}/api/workspaces/delete/wm-fork-t"
+        )),
+        "SECRET_TOKEN_2",
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
+    let owner: Option<String> = sqlx::query_scalar(
+        "SELECT owner_workspace_id FROM datatable_database_permissions WHERE database_key = $1",
+    )
+    .bind(MAIN_KEY)
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(owner.as_deref(), Some("test-workspace"));
+    let roles = usable_roles(port, "test-workspace", "main", "SECRET_TOKEN_3").await;
+    assert_eq!(roles["roles"], json!(["analyst"]), "{roles}");
+
+    // Another workspace reaching the same database, then the owner is deleted.
+    sqlx::query(
+        "INSERT INTO workspace (id, name, owner) VALUES ('elsewhere', 'elsewhere', 'test-user')",
+    )
+    .execute(&db)
+    .await?;
+    plant_main(&db, "elsewhere").await;
+    sqlx::query(
+        "INSERT INTO usr (workspace_id, email, username, is_admin, role)
+         VALUES ('elsewhere', 'test3@windmill.dev', 'test-user-3', true, 'Admin')",
+    )
+    .execute(&db)
+    .await?;
+    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 owner: Option<String> = sqlx::query_scalar(
+        "SELECT owner_workspace_id FROM datatable_database_permissions WHERE database_key = $1",
+    )
+    .bind(MAIN_KEY)
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(owner, None);
+    // Admin of `elsewhere`, and the tenant the row names: neither counts without an owner.
+    let closed = usable_roles(port, "elsewhere", "main", "SECRET_TOKEN_3").await;
+    assert_eq!(closed["enabled"], json!(true));
+    assert_eq!(closed["roles"], json!([]), "{closed}");
+    let superadmin = usable_roles(port, "elsewhere", "main", "SECRET_TOKEN").await;
+    assert_eq!(
+        superadmin["roles"],
+        json!(["admin", "analyst"]),
+        "{superadmin}"
+    );
+    // Nobody but a superadmin manages it now: an admin of `elsewhere` reads it, changes nothing.
+    let resp = authed(
+        client().get(format!(
+            "http://localhost:{port}/api/w/elsewhere/workspaces/datatable_permissions/main"
+        )),
+        "SECRET_TOKEN_3",
+    )
+    .send()
+    .await?;
+    let status = resp.status().as_u16();
+    let text = resp.text().await?;
+    assert_eq!(status, 200, "{text}");
+    let info: serde_json::Value = serde_json::from_str(&text)?;
+    assert_eq!(info["editable"], json!(false), "{info}");
+
+    Ok(())
+}
+
+/// An export carries a database's roles, tenants and login names, never its
+/// passwords; importing them governs a database nobody governs yet, owned by
+/// the importing workspace, with every role refused until a save recreates the
+/// logins. A database already governed is left alone and reported.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn imported_permissions_govern_without_logins(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let ws = format!("http://localhost:{port}/api/w/test-workspace");
+
+    plant_main(&db, "test-workspace").await;
+    plant_permissions(&db, "instance:dt_other", "test-workspace", &["*"]).await;
+    let import = |rows: serde_json::Value| {
+        let ws = ws.clone();
+        async move {
+            let resp = authed(
+                client().post(format!("{ws}/workspaces/datatable_permissions_import")),
+                "SECRET_TOKEN",
+            )
+            .json(&rows)
+            .send()
+            .await
+            .unwrap();
+            let status = resp.status().as_u16();
+            let text = resp.text().await.unwrap();
+            (status, text)
+        }
+    };
+    let exported = json!([
+        { "database_key": MAIN_KEY, "permissions": { "enabled": true, "roles": {
+            "admin": { "tenants": [] },
+            "analyst": { "tenants": ["u/test-user-3"], "pg_rolename": "wm_analyst_x", "pg_password": "leaked?" }
+        }}},
+        { "database_key": "instance:dt_other", "permissions": { "enabled": true, "roles": { "admin": { "tenants": [] } } } }
+    ]);
+    let (status, text) = import(exported).await;
+    assert_eq!(status, 200, "{text}");
+    assert_eq!(text, "[\"instance:dt_other\"]");
+
+    let row: (Option<String>, serde_json::Value) = sqlx::query_as(
+        "SELECT owner_workspace_id, permissions FROM datatable_database_permissions WHERE database_key = $1",
+    )
+    .bind(MAIN_KEY)
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(row.0.as_deref(), Some("test-workspace"));
+    assert_eq!(
+        row.1["roles"]["analyst"]["pg_rolename"],
+        json!("wm_analyst_x")
+    );
+    assert!(
+        row.1["roles"]["analyst"].get("pg_password").is_none(),
+        "{}",
+        row.1
+    );
+    // The tenant is listed as usable; resolving the role, which has no stored
+    // credential, is refused rather than falling back to the owning connection.
+    let roles = usable_roles(port, "test-workspace", "main", "SECRET_TOKEN_3").await;
+    assert_eq!(roles["roles"], json!(["analyst"]), "{roles}");
+    let resp = authed(
+        client().get(format!(
+            "{ws}/workspaces/get_datatable_table_schema?datatable_name=main&schema_name=public&table_name=t&role=analyst"
+        )),
+        "SECRET_TOKEN_3",
+    )
+    .send()
+    .await?;
+    let text = resp.text().await?;
+    assert!(text.contains("no stored credential"), "{text}");
+
+    Ok(())
+}

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

@@ -34,7 +34,8 @@ use windmill_common::utils::require_admin;
 use windmill_common::worker::SqlAnnotations;
 use windmill_common::workspaces::{
     can_use_datatable_role_in_owner_workspace, database_permissions_by_key,
-    delete_database_permissions, lock_database_permissions, resolve_datatable_database_unchecked,
+    delete_database_permissions, lock_database_permissions, lock_database_permissions_key,
+    lock_datatable_permissions_unchecked, resolve_datatable_database_unchecked,
     upsert_database_permissions, DataTable, DataTableCatalogResourceType, DataTablePermissions,
     DatabasePermissions, DatatableAccess, ADMIN_DATATABLE_ROLE,
 };
@@ -54,6 +55,10 @@ pub(crate) fn routes() -> Router {
             "/datatable_usable_roles/{datatable_name}",
             get(list_usable_datatable_roles),
         )
+        .route(
+            "/datatable_permissions_import",
+            post(import_datatable_permissions),
+        )
 }
 
 /// A data table role as the UI sees it: the generated password never leaves the
@@ -84,6 +89,14 @@ pub struct DatatablePermissionsInfo {
     pub editable: bool,
 }
 
+/// A database's permissions as a workspace export carries them: the roles and
+/// their tenants, the login names, never the passwords.
+#[derive(Deserialize, Debug)]
+pub struct ImportedDatabasePermissions {
+    pub database_key: String,
+    pub permissions: DataTablePermissions,
+}
+
 #[derive(Deserialize, Debug)]
 pub struct SetDatatablePermissions {
     pub enabled: bool,
@@ -470,7 +483,11 @@ async fn build_plan(
     req: &SetDatatablePermissions,
 ) -> Result<(tokio_postgres::Client, AdminConnection, RolePlan)> {
     require_datatable_permissions_license().await?;
-    ensure_save_names_what_exists(db, w_id, datatable_name, old, req).await?;
+    // Validated against the database the entry names before anything connects to
+    // it: a stale tenant or a stranded migration is refused without a round trip,
+    // and the connection is then checked to have reached that same database.
+    let (_, _, key) = resolve_datatable_database_unchecked(db, w_id, datatable_name).await?;
+    ensure_save_names_what_exists(db, w_id, &key, old, req).await?;
     let (client, conn) = connect_as_admin_unchecked(db, w_id, datatable_name).await?;
     let mut plan = crate::datatable_permissions_oss::plan_role_changes(
         &conn.database_key,
@@ -490,7 +507,7 @@ async fn build_plan(
             .flatten()
             .filter(|name| *name != ADMIN_DATATABLE_ROLE)
             .collect();
-        let stranded = migrations_naming(db, w_id, datatable_name, &roles).await?;
+        let stranded = migrations_naming(db, &conn.database_key, &roles).await?;
         if !stranded.is_empty() {
             plan.warnings.push(format!(
                 "Migration(s) {} name a role in a `-- role` annotation; they will not run until \
@@ -513,7 +530,9 @@ pub(crate) type PlannedRoleDrop = (tokio_postgres::Client, RolePlan, String);
 
 /// Plan the removal of every Postgres role of the database a data table reaches,
 /// for a database that is going away with the data table — a workspace being
-/// deleted, a fork's clone being dropped.
+/// deleted, a fork's clone being dropped — when `w_id` owns its permissions. A
+/// workspace that merely reaches the database, a fork holding a copy above all,
+/// has no say over roles another workspace turned on.
 ///
 /// A database that is already unreachable must not block the deletion, so a
 /// failure here is logged and the deletion goes ahead without a plan — leaving
@@ -531,10 +550,11 @@ pub(crate) async fn plan_drop_of_datatable_roles(
     };
     let res = async {
         let (_, _, key) = resolve_datatable_database_unchecked(db, w_id, datatable_name).await?;
-        let Some(record) = database_permissions_by_key(db, &key)
-            .await?
-            .filter(|r| r.permissions.enabled && r.permissions.roles.len() > 1)
-        else {
+        let Some(record) = database_permissions_by_key(db, &key).await?.filter(|r| {
+            r.owner_workspace_id.as_deref() == Some(w_id)
+                && r.permissions.enabled
+                && r.permissions.roles.len() > 1
+        }) else {
             return Ok::<_, Error>(None);
         };
         let (client, _, plan) =
@@ -622,24 +642,15 @@ async fn drop_roles_the_record_no_longer_names(
     Ok(())
 }
 
-/// Drop the roles of a database that is going away, once the deletion saying so
-/// has committed, and forget its permissions.
-pub(crate) async fn run_planned_drop(
-    db: &DB,
-    w_id: &str,
-    datatable_name: &str,
-    planned: PlannedRoleDrop,
-) {
-    let key = planned.2.clone();
-    if run_planned_drop_keeping_record(db, w_id, datatable_name, planned).await {
-        forget_database_permissions(db, &key).await;
-    }
-}
-
 /// Drop the roles a plan names, leaving the database's permissions row in place:
 /// with its logins gone every role is refused and `admin` stays the owning
 /// workspace's alone, which is the safe state for a database that was meant to go
-/// and did not. Returns whether the roles were dropped.
+/// and did not — and for one whose owner is gone, which is what a workspace
+/// deletion leaves behind. Returns whether the roles were dropped.
+///
+/// Run under the row's lock, so no save plans against these roles meanwhile,
+/// and without asking the row whether it still names them: it does, and the
+/// plan is what says they go.
 pub(crate) async fn run_planned_drop_keeping_record(
     db: &DB,
     w_id: &str,
@@ -647,7 +658,24 @@ pub(crate) async fn run_planned_drop_keeping_record(
     (mut client, plan, database_key): PlannedRoleDrop,
 ) -> bool {
     let statements: Vec<&PlannedStatement> = plan.statements.iter().collect();
-    match drop_roles_the_record_no_longer_names(db, &database_key, &mut client, &statements).await {
+    let ran = async {
+        let mut tx = db.begin().await?;
+        lock_database_permissions(&mut tx, &database_key).await?;
+        client
+            .batch_execute("SET statement_timeout = '60s'")
+            .await
+            .map_err(|e| {
+                Error::internal_err(format!(
+                    "Failed to bound the role changes: {}",
+                    pg_error_message(&e)
+                ))
+            })?;
+        run_statements(&mut client, &statements).await?;
+        tx.commit().await?;
+        Ok::<(), Error>(())
+    }
+    .await;
+    match ran {
         Ok(()) => true,
         Err(e) => {
             tracing::error!(
@@ -716,13 +744,17 @@ async fn ensure_can_manage_permissions(
 ) -> Result<()> {
     require_admin(authed.is_admin, &authed.username)?;
     match record {
-        Some(record) if record.owner_workspace_id != w_id => {
+        Some(record) if record.owner_workspace_id.as_deref() != Some(w_id) => {
             if !windmill_common::auth::is_super_admin_email(db, &authed.email).await? {
-                return Err(Error::NotAuthorized(format!(
-                    "The permissions of this database are managed from workspace '{}', \
-                     which turned them on.",
-                    record.owner_workspace_id
-                )));
+                return Err(Error::NotAuthorized(match &record.owner_workspace_id {
+                    Some(owner) => format!(
+                        "The permissions of this database are managed from workspace '{owner}', \
+                         which turned them on."
+                    ),
+                    None => "The workspace that turned this database's permissions on was \
+                             deleted; only a superadmin can change them now."
+                        .to_string(),
+                }));
             }
         }
         Some(_) => {}
@@ -757,7 +789,7 @@ fn permissions_info(
                 pg_rolename: role.pg_rolename,
             })
             .collect(),
-        owner_workspace_id: record.map(|r| r.owner_workspace_id.clone()),
+        owner_workspace_id: record.and_then(|r| r.owner_workspace_id.clone()),
         editable,
     }
 }
@@ -810,7 +842,7 @@ pub(crate) async fn ensure_can_use_datatable_role(
     })?;
     let allowed = can_use_datatable_role_in_owner_workspace(
         db,
-        &record.owner_workspace_id,
+        record.owner_workspace_id.as_deref(),
         w_id,
         entry,
         &DatatableAccess::Authed(authed.to_authed_ref()),
@@ -835,7 +867,7 @@ pub(crate) async fn usable_roles(
     for (name, role) in record.permissions.roles.iter() {
         if can_use_datatable_role_in_owner_workspace(
             db,
-            &record.owner_workspace_id,
+            record.owner_workspace_id.as_deref(),
             w_id,
             role,
             access,
@@ -868,7 +900,7 @@ pub(crate) async fn usable_roles(
 async fn ensure_save_names_what_exists(
     db: &DB,
     w_id: &str,
-    datatable_name: &str,
+    database_key: &str,
     old: Option<&DataTablePermissions>,
     req: &SetDatatablePermissions,
 ) -> Result<()> {
@@ -951,7 +983,7 @@ async fn ensure_save_names_what_exists(
         .flatten()
         .filter(|name| !kept.contains(name) && *name != ADMIN_DATATABLE_ROLE)
         .collect();
-    let blocking = migrations_naming(db, w_id, datatable_name, &gone).await?;
+    let blocking = migrations_naming(db, database_key, &gone).await?;
     if !blocking.is_empty() {
         return Err(Error::BadRequest(format!(
             "Migration(s) {} name a role this save removes or renames, in a `-- role` \
@@ -963,27 +995,35 @@ async fn ensure_save_names_what_exists(
     Ok(())
 }
 
-/// The stored migrations of a data table whose `-- role` annotation names one of
-/// `roles`, as `'<migration>' (role '<role>')`.
+/// The stored migrations, of every data table entry on the instance that
+/// reaches the database `database_key` names, whose `-- role` annotation names
+/// one of `roles`, as `<workspace>/<data table>: '<migration>' (role '<role>')`.
+///
+/// Roles are the database's, so a migration of another entry reaching it — in
+/// this workspace or a fork's copy — is stranded by a removal exactly as this
+/// entry's own would be. Only migrations carrying an annotation are read, and
+/// only the entries those name are resolved; an entry that does not resolve
+/// reaches nothing.
 async fn migrations_naming(
     db: &DB,
-    w_id: &str,
-    datatable_name: &str,
+    database_key: &str,
     roles: &HashSet<&str>,
 ) -> Result<Vec<String>> {
     if roles.is_empty() {
         return Ok(vec![]);
     }
     let migrations = sqlx::query!(
-        r#"SELECT name AS "name!", code_up AS "code_up!", code_down
-           FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2
-           ORDER BY timestamp"#,
-        w_id,
-        datatable_name,
+        r#"SELECT workspace_id AS "workspace_id!", datatable AS "datatable!", name AS "name!",
+                  code_up AS "code_up!", code_down
+           FROM datatable_migrations
+           WHERE code_up LIKE '%-- role %' OR code_down LIKE '%-- role %'
+           ORDER BY workspace_id, datatable, timestamp"#,
     )
     .fetch_all(db)
     .await?;
     let mut naming = Vec::new();
+    let mut reaches: std::collections::HashMap<(String, String), bool> =
+        std::collections::HashMap::new();
     for m in migrations {
         let names = [Some(m.code_up.as_str()), m.code_down.as_deref()]
             .into_iter()
@@ -991,8 +1031,30 @@ async fn migrations_naming(
             .filter_map(SqlAnnotations::datatable_role)
             .filter(|role| roles.contains(role.as_str()))
             .collect::<HashSet<String>>();
+        if names.is_empty() {
+            continue;
+        }
+        let entry = (m.workspace_id.clone(), m.datatable.clone());
+        let reached = match reaches.get(&entry) {
+            Some(reached) => *reached,
+            None => {
+                let reached =
+                    resolve_datatable_database_unchecked(db, &m.workspace_id, &m.datatable)
+                        .await
+                        .map(|(_, _, key)| key == database_key)
+                        .unwrap_or(false);
+                reaches.insert(entry, reached);
+                reached
+            }
+        };
+        if !reached {
+            continue;
+        }
         for role in names {
-            naming.push(format!("'{}' (role '{role}')", m.name));
+            naming.push(format!(
+                "{}/{}: '{}' (role '{role}')",
+                m.workspace_id, m.datatable, m.name
+            ));
         }
     }
     Ok(naming)
@@ -1071,12 +1133,18 @@ async fn set_datatable_permissions(
     // everything touching its permissions takes, so taking it here is what
     // serializes them — whether or not the row exists yet.
     let mut tx = db.begin().await?;
-    let record = lock_database_permissions(&mut tx, &key).await?;
+    lock_database_permissions_key(&mut tx, &key).await?;
+    let record = database_permissions_by_key(&mut *tx, &key).await?;
     ensure_can_manage_permissions(&db, &authed, &w_id, record.as_ref(), req.enabled).await?;
+    // A row without an owner — its workspace deleted — is adopted by the
+    // workspace a superadmin saves it from.
     let owner_workspace_id = record
         .as_ref()
-        .map(|r| r.owner_workspace_id.clone())
+        .and_then(|r| r.owner_workspace_id.clone())
         .unwrap_or_else(|| w_id.clone());
+    // The tenants are the owning workspace's principals: its lock is what a
+    // principal's deletion takes before stripping them, first save or not.
+    lock_datatable_permissions_unchecked(&mut tx, &owner_workspace_id).await?;
 
     // The roles about to be created are handed privileges by this connection,
     // which cannot pass on what it holds without the grant option.
@@ -1084,7 +1152,7 @@ async fn set_datatable_permissions(
 
     // The plan is rebuilt here rather than trusted from the preview: the client
     // never gets to choose what runs against the database.
-    let (mut client, _, plan) = build_plan(
+    let (mut client, conn, plan) = build_plan(
         &db,
         &w_id,
         &datatable_name,
@@ -1092,6 +1160,16 @@ async fn set_datatable_permissions(
         &req,
     )
     .await?;
+    // The plan was built against the database the entry resolves to now; the
+    // row it is about to be written under is the one locked above. An entry
+    // pointed elsewhere in between would leave the database it left open and
+    // the one it reached ungoverned.
+    if conn.database_key != key {
+        return Err(Error::BadRequest(format!(
+            "Data table '{datatable_name}' was pointed at another database while its \
+             permissions were being saved. Reload and save again."
+        )));
+    }
 
     // Creating and renaming roles is committed before the row: a Windmill-side
     // failure after this point leaves roles the row does not know about, which
@@ -1158,3 +1236,57 @@ async fn set_datatable_permissions(
         "Permissions of data table {datatable_name} updated"
     ))
 }
+
+/// Restore exported permissions on databases nobody governs yet, owned by this
+/// workspace. The export carries no passwords, so every role is refused until an
+/// admin saves the drawer again, which recreates the logins; a database that is
+/// already governed is left as it is and reported.
+async fn import_datatable_permissions(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Json(rows): Json<Vec<ImportedDatabasePermissions>>,
+) -> JsonResult<Vec<String>> {
+    require_admin(authed.is_admin, &authed.username)?;
+    require_datatable_permissions_license().await?;
+    if crate::workspaces_extra::workspace_is_fork(&db, &w_id).await? {
+        return Err(Error::BadRequest(
+            "Data table permissions cannot be imported into a fork workspace.".to_string(),
+        ));
+    }
+    let mut skipped = Vec::new();
+    for row in rows {
+        let mut tx = db.begin().await?;
+        lock_database_permissions_key(&mut tx, &row.database_key).await?;
+        if database_permissions_by_key(&mut *tx, &row.database_key)
+            .await?
+            .is_some()
+        {
+            skipped.push(row.database_key);
+            continue;
+        }
+        let mut permissions = row.permissions;
+        for role in permissions.roles.values_mut() {
+            role.pg_password = None;
+        }
+        if !permissions.roles.contains_key(ADMIN_DATATABLE_ROLE) {
+            permissions
+                .roles
+                .insert(ADMIN_DATATABLE_ROLE.to_string(), Default::default());
+        }
+        lock_datatable_permissions_unchecked(&mut tx, &w_id).await?;
+        upsert_database_permissions(&mut tx, &row.database_key, &w_id, &permissions).await?;
+        audit_log(
+            &mut *tx,
+            &authed,
+            "workspaces.import_datatable_permissions",
+            ActionKind::Create,
+            &w_id,
+            Some(&authed.email),
+            Some([("database", row.database_key.as_str())].into()),
+        )
+        .await?;
+        tx.commit().await?;
+    }
+    Ok(Json(skipped))
+}

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

@@ -1285,11 +1285,12 @@ pub(crate) async fn delete_workspace(
         );
     }
 
-    // The workspace is gone, so nothing names these logins any more and the drop
-    // finds them unclaimed. A workspace id is reusable, and so are the names
-    // generated under it, which is what makes leaving them behind more than litter.
+    // The workspace that owned these roles is gone. Their logins go with it; the
+    // permissions rows stay, ownerless — every entry still reaching those databases
+    // finds every role refused rather than the owning connection.
     for (name, planned) in planned_role_drops {
-        crate::datatable_permissions::run_planned_drop(&db, &w_id, &name, planned).await;
+        crate::datatable_permissions::run_planned_drop_keeping_record(&db, &w_id, &name, planned)
+            .await;
     }
 
     if let Some(parent) = dev_lock_parent {

backend/windmill-api/openapi.yaml

@@ -5318,6 +5318,38 @@ paths:
                     items:
                       type: string
 
+  /w/{workspace}/workspaces/datatable_permissions_import:
+    post:
+      summary: restore exported datatable permissions on databases nobody governs yet (admins only); passwords are not restored, so every role is refused until saved again
+      operationId: importDatatablePermissions
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: array
+              items:
+                type: object
+                required: [database_key, permissions]
+                properties:
+                  database_key:
+                    type: string
+                  permissions:
+                    type: object
+      responses:
+        "200":
+          description: the database keys that were already governed and left as they were
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: string
+
   /w/{workspace}/workspaces/datatable_acl/{datatable_name}:
     get:
       summary: read the owner and grants of a datatable schema or table

backend/windmill-api/src/workspaces_export.rs

@@ -1759,6 +1759,31 @@ pub(crate) async fn tarball_workspace(
         }
     }
 
+    // The permissions of the databases this workspace governs, as the import
+    // endpoint takes them back: roles, tenants and login names, never passwords —
+    // those are direct database logins, and a re-save recreates them.
+    let permissions =
+        windmill_common::workspaces::database_permissions_owned_by(&mut *tx, &w_id).await?;
+    if !permissions.is_empty() {
+        let exported: Vec<serde_json::Value> = permissions
+            .into_iter()
+            .map(|mut row| {
+                for role in row.permissions.roles.values_mut() {
+                    role.pg_password = None;
+                }
+                serde_json::json!({
+                    "database_key": row.database_key,
+                    "permissions": row.permissions,
+                })
+            })
+            .collect();
+        let json = serde_json::to_string_pretty(&exported)
+            .map_err(|e| Error::internal_err(format!("serializing permissions: {e}")))?;
+        archive
+            .write_to_archive(&json, "datatable_permissions.json")
+            .await?;
+    }
+
     archive.finish().await?;
 
     let file = tokio::fs::File::open(&file_path).await?;

backend/windmill-common/src/workspaces.rs

@@ -1265,11 +1265,15 @@ pub struct DataTableRole {
 /// resolves to this one row rather than carrying a copy of its own. The tenants
 /// in `permissions` are principals of `owner_workspace_id`, and a caller from
 /// another workspace is evaluated as a member of that one (see
-/// [`can_use_datatable_role_in_owner_workspace`]).
+/// [`can_use_datatable_role_in_owner_workspace`]). A row whose owner was
+/// deleted keeps governing its database with none: every role is refused and
+/// only a superadmin reaches it, until one opts out or saves it from a workspace
+/// that then becomes the owner — the database stays closed rather than falling
+/// open to every entry that still reaches it.
 #[derive(Debug, Clone)]
 pub struct DatabasePermissions {
     pub database_key: String,
-    pub owner_workspace_id: String,
+    pub owner_workspace_id: Option<String>,
     pub permissions: DataTablePermissions,
 }
 
@@ -1291,15 +1295,29 @@ pub fn datatable_database_key(
         }
         DataTableCatalogResourceType::Postgresql => {
             use sha2::{Digest, Sha256};
+            // As the connection reads them, not as the JSON spells them: a port
+            // left out is 5432, a number and its string are one port.
+            let text = |field: &str| {
+                resolved
+                    .get(field)
+                    .map(|v| match v.as_str() {
+                        Some(s) => s.to_string(),
+                        None => v.to_string(),
+                    })
+                    .unwrap_or_default()
+            };
+            let port = resolved
+                .get("port")
+                .and_then(|v| {
+                    v.as_u64()
+                        .or_else(|| v.as_str().and_then(|s| s.trim().parse::<u64>().ok()))
+                })
+                .unwrap_or(5432);
             let mut hasher = Sha256::new();
             // NUL-joined so a value cannot be replayed by moving characters across
             // the field boundaries.
-            for field in ["host", "port", "dbname"] {
-                let value = resolved
-                    .get(field)
-                    .map(|v| v.to_string())
-                    .unwrap_or_default();
-                hasher.update(value.as_bytes());
+            for part in [text("host"), port.to_string(), text("dbname")] {
+                hasher.update(part.as_bytes());
                 hasher.update([0u8]);
             }
             format!("pg:{}", hex::encode(hasher.finalize()))
@@ -1309,7 +1327,7 @@ pub fn datatable_database_key(
 
 fn database_permissions_row(
     database_key: String,
-    owner_workspace_id: String,
+    owner_workspace_id: Option<String>,
     permissions: serde_json::Value,
 ) -> Result<DatabasePermissions> {
     Ok(DatabasePermissions {
@@ -1341,15 +1359,34 @@ pub async fn database_permissions_by_key<'e, E: sqlx::PgExecutor<'e>>(
         .transpose()
 }
 
-/// Take the database's permissions row for the length of the caller's
-/// transaction, whether or not it exists yet: an advisory lock on the key
-/// serializes two opt-ins racing to create it, and the row lock serializes a
-/// save with the principal deletions that strip tenants from it.
+/// Take the key of a database's permissions for the length of the caller's
+/// transaction, whether or not its row exists yet: what serializes two opt-ins
+/// racing to create it, and every change to an existing row with every other.
 ///
-/// **Take it before the transaction locks anything else.** One lock, always
-/// acquired first, cannot deadlock. A caller that also needs a workspace's rows
-/// (a rename, a deletion) takes those through
-/// [`lock_datatable_permissions_unchecked`] and nothing else first.
+/// **The first lock a transaction that changes a database's permissions takes.**
+/// A save then takes the owning workspace's lock
+/// ([`lock_datatable_permissions_unchecked`]) before touching the row itself, so
+/// it serializes with the principal deletions that strip tenants — those take
+/// the workspace lock and then the rows, so a save must not hold the row while
+/// it waits for the workspace, which is why the row is not taken here.
+///
+/// Authorization: performs none.
+pub async fn lock_database_permissions_key(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    database_key: &str,
+) -> Result<()> {
+    sqlx::query!(
+        "SELECT pg_advisory_xact_lock(hashtext('datatable_database_permissions:' || $1))",
+        database_key
+    )
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+/// [`lock_database_permissions_key`] and the row itself, for a change that
+/// needs no workspace lock: an ACL change, a drop of the roles of a database
+/// that is going away.
 ///
 /// Authorization: performs none; the row carries login passwords, so callers
 /// MUST have authorized the read and MUST NOT pass the value outward.
@@ -1357,12 +1394,7 @@ pub async fn lock_database_permissions(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     database_key: &str,
 ) -> Result<Option<DatabasePermissions>> {
-    sqlx::query!(
-        "SELECT pg_advisory_xact_lock(hashtext('datatable_database_permissions:' || $1))",
-        database_key
-    )
-    .execute(&mut **tx)
-    .await?;
+    lock_database_permissions_key(tx, database_key).await?;
     let row = sqlx::query!(
         r#"SELECT database_key, owner_workspace_id, permissions
            FROM datatable_database_permissions WHERE database_key = $1 FOR UPDATE"#,
@@ -1375,7 +1407,7 @@ pub async fn lock_database_permissions(
 }
 
 /// Write the database's permissions, creating the row for `owner_workspace_id`
-/// when none exists. An existing row keeps its owner.
+/// when none exists. An existing row keeps its owner, unless it lost it.
 pub async fn upsert_database_permissions(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     database_key: &str,
@@ -1387,7 +1419,10 @@ pub async fn upsert_database_permissions(
     sqlx::query!(
         r#"INSERT INTO datatable_database_permissions (database_key, owner_workspace_id, permissions)
            VALUES ($1, $2, $3)
-           ON CONFLICT (database_key) DO UPDATE SET permissions = EXCLUDED.permissions, updated_at = now()"#,
+           ON CONFLICT (database_key) DO UPDATE
+           SET permissions = EXCLUDED.permissions,
+               owner_workspace_id = COALESCE(datatable_database_permissions.owner_workspace_id, EXCLUDED.owner_workspace_id),
+               updated_at = now()"#,
         database_key,
         owner_workspace_id,
         permissions
@@ -1432,22 +1467,25 @@ pub async fn database_permissions_owned_by<'e, E: sqlx::PgExecutor<'e>>(
         .collect()
 }
 
-/// Take every permissions row `w_id` owns for the length of the caller's
-/// transaction, and read them under the lock.
+/// Take the workspace's lock over the permissions it owns, and every row it
+/// owns, for the length of the caller's transaction, and read them under it.
 ///
-/// The rows are the one lock over what a database's permissions depend on: the
+/// This is the one lock over what a workspace's permissions depend on: the
 /// roles themselves, and the users, groups and folders they name as tenants.
-/// Every path that touches either — a role save, an ACL change, a principal's
-/// rename or deletion — takes it, which is what stops one of them persisting a
-/// tenant list it computed before another committed: a save that planned with
-/// `g/devs` would otherwise put the tenant back after the group's deletion took
-/// it away.
+/// Every path that touches either — a role save, a principal's rename or
+/// deletion — takes it, which is what stops one of them persisting a tenant list
+/// it computed before another committed: a save that planned with `g/devs`
+/// would otherwise put the tenant back after the group's deletion took it away.
+/// The lock is an advisory one on the workspace id, not the rows alone: the
+/// first save of a workspace has no row to lock yet, and a deletion committing
+/// between its validation and its insert would leave a freed name on a role.
 ///
-/// **Take it before the transaction locks anything else**, with one exception:
-/// the dev-pairing advisory lock (`lock_dev_pairing`) comes first where a path
-/// needs both, and nothing takes these rows and then reaches for that one. One
-/// order, held everywhere, cannot deadlock. A transaction spanning workspaces
-/// takes them in `workspace_id` order, for the same reason.
+/// **Take it before the transaction locks anything else**, with two exceptions
+/// that come first where a path needs them: the dev-pairing advisory lock
+/// (`lock_dev_pairing`), and a database key ([`lock_database_permissions_key`]).
+/// Nothing takes these rows and then reaches for either. One order, held
+/// everywhere, cannot deadlock. A transaction spanning workspaces takes them in
+/// `workspace_id` order, for the same reason.
 ///
 /// Authorization: performs none, for any workspace it is handed. The rows carry
 /// login passwords, so callers MUST have authorized the read and MUST NOT pass
@@ -1457,6 +1495,12 @@ pub async fn lock_datatable_permissions_unchecked(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     w_id: &str,
 ) -> Result<Vec<DatabasePermissions>> {
+    sqlx::query!(
+        "SELECT pg_advisory_xact_lock(hashtext('datatable_database_permissions_owner:' || $1))",
+        w_id
+    )
+    .execute(&mut **tx)
+    .await?;
     let rows = sqlx::query!(
         r#"SELECT database_key, owner_workspace_id, permissions
            FROM datatable_database_permissions WHERE owner_workspace_id = $1
@@ -1797,6 +1841,49 @@ pub async fn get_datatable_replication_resource_from_db_unchecked(
     .await
 }
 
+/// Refuse `access`, made from `w_id`, the data table's owning connection when
+/// its database is permissioned, unless it may run as `admin` there: what a
+/// Postgres trigger needs, since replication streams every change of the
+/// database whatever roles say.
+///
+/// Authorization: this is the check; an unpermissioned database refuses nobody.
+pub async fn ensure_datatable_admin_access(
+    db: &DB,
+    w_id: &str,
+    name: &str,
+    access: &DatatableAccess<'_>,
+) -> Result<()> {
+    let (_, _, key) = resolve_datatable_database_unchecked(db, w_id, name).await?;
+    let Some(record) = database_permissions_by_key(db, &key)
+        .await?
+        .filter(|r| r.permissions.enabled)
+    else {
+        return Ok(());
+    };
+    let admin = record
+        .permissions
+        .roles
+        .get(ADMIN_DATATABLE_ROLE)
+        .cloned()
+        .unwrap_or_default();
+    if can_use_datatable_role_in_owner_workspace(
+        db,
+        record.owner_workspace_id.as_deref(),
+        w_id,
+        &admin,
+        access,
+    )
+    .await?
+    {
+        Ok(())
+    } else {
+        Err(Error::NotAuthorized(format!(
+            "Data table '{name}' reaches a database with role permissions enabled; replicating \
+             it is for the admins of the workspace that manages them."
+        )))
+    }
+}
+
 /// Look up the role a resolution asks for, without authorizing it.
 ///
 /// `Ok(None)` means the database is unpermissioned and the data table resolves
@@ -1835,7 +1922,8 @@ fn datatable_role_entry<'a>(
 }
 
 /// Whether `access`, made from workspace `w_id`, may run as `role` of a database
-/// whose permissions `owner_w_id` owns.
+/// whose permissions `owner_w_id` owns — or nobody, once the owner was deleted:
+/// then only a superadmin does.
 ///
 /// Tenants are principals of the owning workspace, and so is the admin bypass:
 /// a caller from another workspace — a fork's copy of the data table, a
@@ -1850,7 +1938,7 @@ fn datatable_role_entry<'a>(
 /// through, for callers that have authorized already.
 pub async fn can_use_datatable_role_in_owner_workspace(
     db: &DB,
-    owner_w_id: &str,
+    owner_w_id: Option<&str>,
     w_id: &str,
     role: &DataTableRole,
     access: &DatatableAccess<'_>,
@@ -1859,7 +1947,7 @@ pub async fn can_use_datatable_role_in_owner_workspace(
         DatatableAccess::Unchecked => return Ok(true),
         DatatableAccess::NoIdentity => return Ok(false),
         DatatableAccess::Authed(authed) => {
-            if w_id == owner_w_id {
+            if Some(w_id) == owner_w_id {
                 return Ok(can_use_datatable_role(role, authed));
             }
             (format!("u/{}", authed.username), authed.email.to_string())
@@ -1879,7 +1967,7 @@ pub async fn can_use_datatable_role_in_owner_workspace(
             (job.permissioned_as, job.permissioned_as_email)
         }
     };
-    if w_id == owner_w_id {
+    if Some(w_id) == owner_w_id {
         let authed =
             crate::auth::fetch_authed_from_permissioned_as(&permissioned_as, &email, w_id, db)
                 .await?;
@@ -1888,6 +1976,9 @@ pub async fn can_use_datatable_role_in_owner_workspace(
     if crate::auth::is_super_admin_email(db, &email).await? {
         return Ok(true);
     }
+    let Some(owner_w_id) = owner_w_id else {
+        return Ok(false);
+    };
     if !permissioned_as.starts_with("u/") {
         return Ok(false);
     }
@@ -1930,9 +2021,10 @@ async fn resolve_datatable_role(
     let Some((role_name, role_entry)) = datatable_role_entry(record, name, role)? else {
         return Ok(None);
     };
-    let owner_w_id = &record
+    let owner_w_id = record
         .expect("a role entry comes from a record")
-        .owner_workspace_id;
+        .owner_workspace_id
+        .as_deref();
     if !can_use_datatable_role_in_owner_workspace(db, owner_w_id, w_id, role_entry, access).await? {
         return Err(Error::NotAuthorized(format!(
             "Not allowed to use role '{role_name}' of data table '{name}'"
@@ -3345,6 +3437,22 @@ mod tests {
             datatable_database_key(&pg("u/a/pg"), &resolved("db", "prod", "app")),
             datatable_database_key(&pg("u/a/pg"), &resolved("db", "staging", "app"))
         );
+        // A port left out is 5432, as the connection reads it, and a number and
+        // its string are one port.
+        assert_eq!(
+            datatable_database_key(&pg("u/a/pg"), &resolved("db", "prod", "app")),
+            datatable_database_key(
+                &pg("u/a/pg"),
+                &serde_json::json!({ "host": "db", "dbname": "prod", "user": "app" })
+            )
+        );
+        assert_eq!(
+            datatable_database_key(&pg("u/a/pg"), &resolved("db", "prod", "app")),
+            datatable_database_key(
+                &pg("u/a/pg"),
+                &serde_json::json!({ "host": "db", "port": "5432", "dbname": "prod" })
+            )
+        );
     }
 
     fn record(roles: &[(&str, &[&str])], default_role: Option<&str>) -> DatabasePermissions {
@@ -3362,7 +3470,7 @@ mod tests {
         }
         DatabasePermissions {
             database_key: "instance:dt_main".to_string(),
-            owner_workspace_id: "acme".to_string(),
+            owner_workspace_id: Some("acme".to_string()),
             permissions: DataTablePermissions {
                 enabled: true,
                 roles: map,

backend/windmill-trigger-postgres/src/lib.rs

@@ -383,7 +383,15 @@ pub async fn resolve_postgres_resource(
 ) -> Result<Postgres> {
     if let Some(datatable_name) = postgres_resource_path.strip_prefix("datatable://") {
         // Trigger connections (publication/slot management + logical replication) run
-        // as the dedicated replication user on custom-instance databases.
+        // as the dedicated replication user on custom-instance databases — and stream
+        // every change of the database, so a permissioned one is for its admins.
+        windmill_common::workspaces::ensure_datatable_admin_access(
+            db,
+            w_id,
+            datatable_name,
+            &windmill_common::workspaces::DatatableAccess::Authed(authed.to_authed_ref()),
+        )
+        .await?;
         let resource_value =
             get_datatable_replication_resource_from_db_unchecked(db, w_id, datatable_name).await?;
         serde_json::from_value::<Postgres>(resource_value).map_err(|e| Error::SerdeJson {