fix(datatables): a governed database is not cloned, an opt-in restarts its triggers, and a cleanup respects an adopted row

backend/.sqlx/query-1100d1ab19ed3bf73ce29af00d7a06605e7b697324813a1dce1c2fd4c72bc713.json

@@ -0,0 +1,32 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id, path, postgres_resource_path\n           FROM postgres_trigger WHERE postgres_resource_path LIKE 'datatable://%'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "postgres_resource_path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "1100d1ab19ed3bf73ce29af00d7a06605e7b697324813a1dce1c2fd4c72bc713"
+}

backend/.sqlx/query-64d40ff3de929d96a3d03c7d5b363f1284faf50fa3a0c5f4d0beab560d14c7e6.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE postgres_trigger SET server_id = NULL, last_server_ping = NULL\n                 WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "64d40ff3de929d96a3d03c7d5b363f1284faf50fa3a0c5f4d0beab560d14c7e6"
+}

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

@@ -771,3 +771,41 @@ async fn imported_permissions_govern_without_logins(db: Pool<Postgres>) -> anyho
 
     Ok(())
 }
+
+/// A clone of a permissioned data table would be a database of its own that
+/// nothing governs, readable in full by every member of the fork: refused at the
+/// one place a caller cannot go around, the fork creation that wires it in.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn a_governed_datatable_cannot_be_cloned_into_a_fork(
+    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", &["*"]).await;
+    let resp = authed(
+        client().post(format!(
+            "http://localhost:{port}/api/w/test-workspace/workspaces/create_fork"
+        )),
+        "SECRET_TOKEN",
+    )
+    .json(&json!({
+        "id": "wm-fork-clone", "name": "clone",
+        "forked_datatables": [{ "name": "main", "new_dbname": "wm_fork_clone_main" }]
+    }))
+    .send()
+    .await?;
+    let status = resp.status().as_u16();
+    let text = resp.text().await?;
+    assert_eq!(status, 400, "{text}");
+    assert!(text.contains("cannot be cloned"), "{text}");
+    let exists: bool =
+        sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM workspace WHERE id = 'wm-fork-clone')")
+            .fetch_one(&db)
+            .await?;
+    assert!(!exists);
+
+    Ok(())
+}

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

@@ -662,19 +662,32 @@ async fn drop_roles_the_record_no_longer_names(
 /// 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.
+/// Run under the row's lock, so no save plans against these roles meanwhile.
+/// The plan was made before the deletion committed; the row is read again here
+/// and the plan runs only while the row is still `expected_owner`'s — none, after
+/// a workspace deletion — so a superadmin who adopted the row from another
+/// workspace in between keeps the roles their save now names.
 pub(crate) async fn run_planned_drop_keeping_record(
     db: &DB,
     w_id: &str,
     datatable_name: &str,
+    expected_owner: Option<&str>,
     (mut client, plan, database_key): PlannedRoleDrop,
 ) -> bool {
     let statements: Vec<&PlannedStatement> = plan.statements.iter().collect();
     let ran = async {
         let mut tx = db.begin().await?;
-        lock_database_permissions(&mut tx, &database_key).await?;
+        let record = lock_database_permissions(&mut tx, &database_key).await?;
+        if record
+            .as_ref()
+            .is_some_and(|r| r.owner_workspace_id.as_deref() != expected_owner)
+        {
+            tracing::warn!(
+                "The roles behind data table {datatable_name} in {w_id} were adopted by workspace {:?} before they could be dropped; left in place",
+                record.and_then(|r| r.owner_workspace_id)
+            );
+            return Ok(false);
+        }
         client
             .batch_execute("SET statement_timeout = '60s'")
             .await
@@ -686,11 +699,11 @@ pub(crate) async fn run_planned_drop_keeping_record(
             })?;
         run_statements(&mut client, &statements).await?;
         tx.commit().await?;
-        Ok::<(), Error>(())
+        Ok::<bool, Error>(true)
     }
     .await;
     match ran {
-        Ok(()) => true,
+        Ok(dropped) => dropped,
         Err(e) => {
             tracing::error!(
                 "Could not drop the Postgres roles behind data table {datatable_name} in {w_id}: {e:#}"
@@ -1115,8 +1128,9 @@ async fn preview_datatable_permissions(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((w_id, datatable_name)): Path<(String, String)>,
-    Json(req): Json<SetDatatablePermissions>,
+    Json(mut req): Json<SetDatatablePermissions>,
 ) -> JsonResult<DatatablePermissionsPreview> {
+    forget_client_login_names(&mut req);
     let (_, _, key) = resolve_datatable_database_unchecked(&db, &w_id, &datatable_name).await?;
     let record = database_permissions_by_key(&db, &key).await?;
     // Refused here too: offering a plan that the save will not run is its own
@@ -1145,9 +1159,10 @@ async fn set_datatable_permissions(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((w_id, datatable_name)): Path<(String, String)>,
-    Json(req): Json<SetDatatablePermissions>,
+    Json(mut req): Json<SetDatatablePermissions>,
 ) -> Result<String> {
     require_admin(authed.is_admin, &authed.username)?;
+    forget_client_login_names(&mut req);
     let (_, _, key) = resolve_datatable_database_unchecked(&db, &w_id, &datatable_name).await?;
 
     // Reading the permissions, planning against them, running the plan and
@@ -1243,6 +1258,14 @@ async fn set_datatable_permissions(
 
     tx.commit().await?;
 
+    // Turned on just now: a trigger streaming this database was opened while
+    // nothing governed it, and is made to ask again.
+    if req.enabled && record.is_none() {
+        if let Err(e) = restart_triggers_reaching(&db, &key).await {
+            tracing::error!("Could not restart the triggers replicating {key}: {e:#}");
+        }
+    }
+
     // What the row no longer names, now that it says so. A failure here is the
     // end of the line for these logins: the save that stopped naming them has
     // committed, so no later plan diffs against them and nothing will try again.
@@ -1328,6 +1351,15 @@ async fn import_datatable_permissions(
     Ok(Json(skipped))
 }
 
+/// A login name is a cluster-wide identifier the planner renames and drops; the
+/// request shape carries the field because it is also the response shape, and
+/// the planner reads a role's login from the stored row alone. Never from here.
+fn forget_client_login_names(req: &mut SetDatatablePermissions) {
+    for role in req.roles.iter_mut() {
+        role.pg_rolename = None;
+    }
+}
+
 /// The shape a save would have refused: role and tenant names as the planner
 /// and the tenant matcher read them.
 fn validate_imported_permissions(permissions: &DataTablePermissions) -> Result<()> {
@@ -1362,3 +1394,71 @@ fn validate_imported_permissions(permissions: &DataTablePermissions) -> Result<(
     }
     Ok(())
 }
+
+/// Refuse to clone a data table whose database has role permissions: the copy
+/// lands in a new database, keyed on its own, where nothing governs it — and a
+/// fork cannot turn permissions on — so every member of the fork would read, in
+/// full, the data the roles existed to divide.
+pub(crate) async fn refuse_clone_of_governed_datatable(
+    db: &DB,
+    w_id: &str,
+    datatable_name: &str,
+) -> Result<()> {
+    let (_, _, key) = resolve_datatable_database_unchecked(db, w_id, datatable_name).await?;
+    if database_permissions_by_key(db, &key)
+        .await?
+        .is_some_and(|r| r.permissions.enabled)
+    {
+        return Err(Error::BadRequest(format!(
+            "Data table '{datatable_name}' has role permissions enabled and cannot be cloned: \
+             the copy would be a database of its own that nothing governs, readable in full by \
+             every member of the fork. Keep the original, which shares the roles."
+        )));
+    }
+    Ok(())
+}
+
+/// Make every Postgres trigger replicating a data table that reaches `database_key`
+/// reconnect, so the admin check runs against the permissions just turned on: a
+/// stream opened while the database was unpermissioned would otherwise keep
+/// receiving every change. Clearing the listener's claim is what stops it; the
+/// next claim resolves the resource again, and refuses where it must.
+pub(crate) async fn restart_triggers_reaching(db: &DB, database_key: &str) -> Result<()> {
+    let triggers = sqlx::query!(
+        r#"SELECT workspace_id, path, postgres_resource_path
+           FROM postgres_trigger WHERE postgres_resource_path LIKE 'datatable://%'"#
+    )
+    .fetch_all(db)
+    .await?;
+    let mut reaches: std::collections::HashMap<(String, String), bool> =
+        std::collections::HashMap::new();
+    for t in triggers {
+        let Some(datatable) = t.postgres_resource_path.strip_prefix("datatable://") else {
+            continue;
+        };
+        let (datatable, _) = windmill_common::workspaces::parse_datatable_ref(datatable);
+        let entry = (t.workspace_id.clone(), datatable.to_string());
+        let reached = match reaches.get(&entry) {
+            Some(reached) => *reached,
+            None => {
+                let reached = resolve_datatable_database_unchecked(db, &t.workspace_id, datatable)
+                    .await
+                    .map(|(_, _, key)| key == database_key)
+                    .unwrap_or(false);
+                reaches.insert(entry, reached);
+                reached
+            }
+        };
+        if reached {
+            sqlx::query!(
+                "UPDATE postgres_trigger SET server_id = NULL, last_server_ping = NULL
+                 WHERE workspace_id = $1 AND path = $2",
+                t.workspace_id,
+                t.path
+            )
+            .execute(db)
+            .await?;
+        }
+    }
+    Ok(())
+}

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

@@ -3457,6 +3457,9 @@ async fn create_pg_database(
     Json(req): Json<CreatePgDatabaseRequest>,
 ) -> Result<String> {
     windmill_common::validate_dbname(&req.target_dbname)?;
+    if let Some(name) = req.source.strip_prefix("datatable://") {
+        crate::datatable_permissions::refuse_clone_of_governed_datatable(&db, &w_id, name).await?;
+    }
 
     // Non-superadmin: restrict dbname to wm_fork_ prefix
     if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
@@ -3554,6 +3557,13 @@ async fn import_pg_database(
         resolve_pg_source_for_copy(&db, &user_db, &authed, &w_id, &req.target).await?;
 
     if let Some(ref override_dbname) = req.target_dbname_override {
+        // Only the fork clone flow overrides the target database name; a plain
+        // database-to-database import is an admin moving data between databases
+        // they already reach.
+        if let Some(name) = req.source.strip_prefix("datatable://") {
+            crate::datatable_permissions::refuse_clone_of_governed_datatable(&db, &w_id, name)
+                .await?;
+        }
         if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
             if !override_dbname.starts_with("wm_fork_") {
                 return Err(Error::BadRequest(
@@ -7721,6 +7731,10 @@ async fn apply_forked_datatable(
     fdt: &ForkedDatatableInfo,
 ) -> Result<()> {
     windmill_common::validate_dbname(&fdt.new_dbname)?;
+    // The clone endpoints refuse this too; this is the one a caller cannot go
+    // around, since it is what wires the fork's config to the copied database.
+    crate::datatable_permissions::refuse_clone_of_governed_datatable(db, parent_w_id, &fdt.name)
+        .await?;
     if !fdt.new_dbname.starts_with("wm_fork_") {
         return Err(Error::BadRequest(format!(
             "Forked datatable database name '{}' must start with 'wm_fork_'",

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

@@ -1289,8 +1289,10 @@ pub(crate) async fn delete_workspace(
     // 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_keeping_record(&db, &w_id, &name, planned)
-            .await;
+        crate::datatable_permissions::run_planned_drop_keeping_record(
+            &db, &w_id, &name, None, planned,
+        )
+        .await;
     }
 
     if let Some(parent) = dev_lock_parent {
@@ -1377,7 +1379,11 @@ async fn drop_datatable_roles_before_its_database(
         crate::datatable_permissions::plan_drop_of_datatable_roles(db, w_id, dt_name).await
     {
         if !crate::datatable_permissions::run_planned_drop_keeping_record(
-            db, w_id, dt_name, planned,
+            db,
+            w_id,
+            dt_name,
+            Some(w_id),
+            planned,
         )
         .await
         {

frontend/src/lib/components/workspaceSettings/ForkDatatableSection.svelte

@@ -41,11 +41,29 @@
 
 	let effectiveSource = $derived(sourceWorkspace ?? $workspaceStore ?? undefined)
 
+	// Listed with whether each is permissioned: a clone of a permissioned data
+	// table would be a database of its own that nothing governs, so the backend
+	// refuses it and the choice is not offered. Where the check could not
+	// answer, the safe reading is "permissioned".
 	let allDatatables = resource(
 		() => effectiveSource,
 		async (ws) => {
 			if (!ws) return undefined
-			return await WorkspaceService.listDataTables({ workspace: ws })
+			const datatables = await WorkspaceService.listDataTables({ workspace: ws })
+			return await Promise.all(
+				datatables.map(async (dt) => {
+					try {
+						const roles = await WorkspaceService.listUsableDatatableRoles({
+							workspace: ws,
+							datatableName: dt.name
+						})
+						return { ...dt, permissioned: roles.enabled as boolean | undefined }
+					} catch (e) {
+						console.error('Failed to read datatable permissions:', e)
+						return { ...dt, permissioned: undefined }
+					}
+				})
+			)
 		}
 	)
 
@@ -192,10 +210,17 @@
 							(v) => (datatableBehaviors[dt.name] = v)
 						}
 						items={[
-							{ value: 'keep_original', label: 'Keep original' },
-							{ value: 'schema_only', label: 'Clone schema only' },
-							...(!isCloudHosted() && $userStore?.is_admin
-								? [{ value: 'schema_and_data', label: 'Clone schema and data' }]
+							{
+								value: 'keep_original',
+								label: dt.permissioned ? 'Keep original (permissions enabled)' : 'Keep original'
+							},
+							...(dt.permissioned === false
+								? [
+										{ value: 'schema_only', label: 'Clone schema only' },
+										...(!isCloudHosted() && $userStore?.is_admin
+											? [{ value: 'schema_and_data', label: 'Clone schema and data' }]
+											: [])
+									]
 								: [])
 						]}
 					/>