fix(datatables): dedicated owner role per protected instance database

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fBWmrMs2DbpU1iYdCvGX6
2026-08-18 16:02:10 +00:00 · 2026-07-28 12:00:23 +02:00
parent ab14e185db
commit ef3b4d5ebd
74 changed files with 1573 additions and 51 deletions
@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT role_name, password, workspace_id FROM datatable_owner_role WHERE dbname = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "role_name",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "password",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "03cf25fdaa82d4f7e82f64350c16c2c2734c2a8de309891b81454634bbb2b695"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE postgres_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "03fdb7d0d4d95a98fdbdcf7dd1a9289119b9ce719d6a6c2975e96d7785819fc9"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings SET slack_email = $1 WHERE slack_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "099067cf6c08642b205074771ef23d7ab950015d70f7065f5645f1e76d6a25ee"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE azure_trigger SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0da2425ff8ee737104cd9f2650f1ffba1511715ac1f9270939d79fb82f8e88d0"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE schedule SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0f49203698dce2d77724c3d2fbaf729c6b2c2f9dd9475b778dbce66f28b48e69"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO folder(workspace_id, name, display_name, owners, extra_perms, default_permissioned_as)\n         VALUES ('test-workspace', 'fold', 'fold', '{}', '{}'::jsonb,\n                 '[{\"path_glob\": \"a/**\", \"permissioned_as\": \"u/other\"}, {\"path_glob\": \"**\", \"permissioned_as\": \"test2@windmill.dev\"}]'::jsonb)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "1159c2beb23057e3123847efa2b5d2d7cd7c89b47cbf5000fb74a9cdaa4b62df"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_fork_deployment_request_comment SET author_email = $1 WHERE author_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "162ab0cc9094d7031229a65f47d730282d17bc8a71a0e234c9edf81b85199dd1"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE app SET policy = jsonb_set(policy, ARRAY['on_behalf_of'], to_jsonb($1::text)) WHERE policy->>'on_behalf_of' = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1b828adc60798b096ebf5db75fc1ade1ffffff492dc02caafe002fd868d90a8d"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id FROM usr WHERE email = 'renamed@windmill.dev'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "24e8f955d2a9a3bf502736749563e82974a5744dbb4942c1f2b2dff774594252"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE mcp_oauth_server_code SET user_email = $1 WHERE user_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "273e18a8fe284da79b0929678f2e07d2db3240c1501d99e6b3d2fccfd7405d06"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE flow SET on_behalf_of_email = $1 WHERE on_behalf_of_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "287a9f01a5b789b7724d7209f8f399638dd57f24fb407cbcb4fe4a510a74d3af"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM workspace_invite o WHERE o.email = $1 AND EXISTS (SELECT 1 FROM workspace_invite n WHERE n.email = $2 AND n.workspace_id = o.workspace_id)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "299a5d12779185a51b6e6815208dfe0aae64c39637208d081462b84d8636aa42"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT owner FROM workspace WHERE id = 'test-workspace'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "owner",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "2b3ffacd09b0fa4bb458f18adbd7cb0f57455338d28e3f164a105a51510c1cc1"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM tutorial_progress WHERE email = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2f74858ebe82499e06283b0d22cc19ecf4e8c9a9ea6a76977bb17aefe9eb8f27"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE http_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "32e6b8ee457bcf28ed052f56ac8d24ad333ee5c75086ab1f07ed952cf6aecea9"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM usage WHERE id = $1 AND NOT is_workspace",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "34f1e04d3ff0ffe7fe4e142bb54e00431bc7fb4a4b20dacdce79316d7a84ba8f"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job SET permissioned_as_email = $1 WHERE permissioned_as_email = $2 AND id IN (SELECT id FROM v2_job_queue)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "35384f86bf1058a228307655c545bfbe91032050603e882bca0d7352a0f08a62"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE kafka_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "35b7c715362239c09765d3717340216154b4bda0bd5729e93318fd6527fbbbc4"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT username FROM password WHERE email = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "username",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "3b7e46f80cf1e543a76f90c74080a3e2d1cf04812aa1091e3c64f262988da099"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT username FROM password WHERE email = 'renamed@windmill.dev'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "username",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "3c5e00d0e79fa518a95b05a00e75dbd96d7852ad340759252de4dcb57ad22a43"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_invite SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3fec727489a5166c3036593d7089fd252e2c779762cb359f98a7d17e657a926f"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE amqp_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "44356ac660ef1113f21b3ec24d67de7e64cda8f1e8419c231ecc090447614c4e"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM datatable_owner_role WHERE dbname = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "49c01ddc282b4d6fabd8ba2527a42397096f5c6a1bcc608b8c7829c2ccbad2ac"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT permissioned_as FROM schedule WHERE path = 'u/test-user-2/sched'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "permissioned_as",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "5ae3532928770d6bd9fc57a3ca489f819d8845ddb54f09f53880a55cd2304c8c"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE mqtt_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5c3e4a9fd4ac5ce75326230ecbbd40bb126ae3e2dff1824904da91b2f7d900cd"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE usr SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5d1bcca26a78968351da909b23cb5fd0d90ed946263db6ffda99ca1cd9e727f9"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE tutorial_progress SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5e74efd5f9152deb9511d2f1efc084cbf6aa8f252d86a50fa3a4f4e93bf373aa"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO app(workspace_id, path, summary, policy, versions)\n         VALUES ('test-workspace', 'u/test-user-2/app', '', '{\"on_behalf_of\": \"test2@windmill.dev\", \"on_behalf_of_email\": \"test2@windmill.dev\"}'::jsonb, '{}')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "629a6f38411c49c13bfb7773cef87ef0b833f57e23392ba0f9c6d2a165a4c9fe"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job SET permissioned_as = $1 WHERE permissioned_as = $2 AND id IN (SELECT id FROM v2_job_queue)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "639f87dbe115bbd3d9eb9a70e74b4c8b3a71829aa9048fc2459e1f6603956e5d"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE websocket_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "647c12e69360e6a34d01a563e6e84b26192c18cb4ede596233c00187b789d9ec"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE usage SET id = $1 WHERE id = $2 AND NOT is_workspace",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "65ee4ace29b39818594efdfa16453e4806c4dca776f80789a57d4d85e16aa197"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_fork_deployment_request_assignee SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "73910791d9eca97624c1b1b9dd65cc861d0339a4238bce6928af49601a6ad4d1"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT password FROM datatable_owner_role WHERE dbname = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "password",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "7603a9378a9d21bc9c14adef3a1af21de870883b969b27d1c64502c15186133e"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE email_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "774db0fcbc3905a8c80f271612efcaf7268011f8868028cad6d5339419cb7aef"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE password SET username = 'test-user-2' WHERE email = 'test2@windmill.dev'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "883be05a1a02dde8d9056094a4f17491791832d5bd206ab99a8b96ad30a0a3f5"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM email_to_igroup o WHERE o.email = $1 AND EXISTS (SELECT 1 FROM email_to_igroup n WHERE n.email = $2 AND n.igroup = o.igroup)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "8bd1c2f216b4c525888bb94f94e09c3c519eb4bccea78954cb25ff945807883c"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE datatable_owner_role SET password = $1 WHERE dbname = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "8bd6890c2bd0eb3641e30c3022304b1debf68c25ef5ac9f353ec5e4dbe79dfcb"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE gcp_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "9d6375303333eab6424dfba2b8fc4afb0d3fc87b79818b5087919a7ae604545f"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM password WHERE email = 'test2@windmill.dev'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a3233c8a26f4ca47e957d55e77d9e3ccd51dcf9d259527e0b1f20d6e7f43f80b"
+}
@@ -0,0 +1,30 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH RECURSIVE queued_ancestors AS (\n            SELECT j.id, j.parent_job, 0 AS depth\n            FROM v2_job j JOIN v2_job_queue q USING (id)\n            WHERE j.id = $1 AND j.workspace_id = $2\n          UNION ALL\n            SELECT j.id, j.parent_job, a.depth + 1\n            FROM queued_ancestors a\n            JOIN v2_job j ON j.id = a.parent_job AND j.workspace_id = $2\n            JOIN v2_job_queue q ON q.id = j.id\n            WHERE a.depth < $3\n        )\n        SELECT id AS \"id!\", depth AS \"depth!\" FROM queued_ancestors ORDER BY depth DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id!",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "depth!",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text",
+        "Int4"
+      ]
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "a6a9a8013ac8ea8ecba8a39c3c0bfb0b5ccdd31740cd63f2d174b3cd1e18019d"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO notify_event (channel, payload) SELECT 'notify_token_invalidation', token_prefix FROM token WHERE email = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a718367f00f7dc0280ab297ff66a939f930eb2a43beac528fd3b5fb6add5e251"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT policy::text FROM app WHERE path = 'u/test-user-2/app' AND workspace_id = 'test-workspace'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "policy",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "ab07fa69343f1ab5e2c89911c314d027a05302085c838aa1c2d84406f9fffdb2"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE script SET on_behalf_of_email = $1 WHERE on_behalf_of_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "acf96371f08812a15760436f43f05290ec39617e18e65f05fa4a9b002560a6d5"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_fork_deployment_request SET requested_by_email = $1 WHERE requested_by_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "aefab48bd35a3bbfdb6d5a86d017acdd2d3759e33b1c75e54aee140d8db562d6"
+}
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT dbname, password FROM datatable_owner_role WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "dbname",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "password",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "b172cb61ef20f27f99b834875923a2707ac27ce28d70c95fd2ea44cdadfafe02"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT default_permissioned_as::text FROM folder WHERE workspace_id = 'test-workspace' AND name = 'fold'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "default_permissioned_as",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c22b71a3a40710e44e66aa48bda41de76b2cc0d0495c7d32f565d2548bf8c860"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT dbname FROM datatable_owner_role WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "dbname",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "c71a985f40cdf31b6cd3da53b554a96d5cf263db3cd37040a348c1becf4ebf30"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(\n                SELECT 1 FROM workspace WHERE owner = $1\n                UNION ALL SELECT 1 FROM workspace_settings WHERE slack_email = $1\n                UNION ALL SELECT 1 FROM usage WHERE id = $1 AND NOT is_workspace\n                UNION ALL SELECT 1 FROM v2_job WHERE permissioned_as = $1 AND id IN (SELECT id FROM v2_job_queue))",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c9513c08e03089a581a8bcc5178094fc42098ac26facbd010006ae06da284131"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE nats_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c9d78891f8ccdfc12c8315653187cc8d2d4a2df257e5fd723016c6344174d4a1"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE sqs_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d24faa96edde6ff6cb7c443775552419ff11b6671a58e3aa0c75a7b48973437a"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM unique_ext_jwt_token WHERE email = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d5f94950c15ee592d5580cf001b37560f2093f9569a2d8a4a702004041976961"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT role_name FROM datatable_owner_role WHERE dbname = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "role_name",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "d73feada6ac9bc88edd462bfb4768df7eaf24331df7388a5e20d921d6dd52697"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE token SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d7597022d6fc01601bff5f243fb38080fc6b65e95a3aad630c7abe3edcedbc98"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace SET owner = 'test2@windmill.dev' WHERE id = 'test-workspace'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "d94140411f0ddefd74732e9a755249a044397b6fbfe164bc3dfd007ccd490388"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE azure_trigger SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d9b33bb1cf7894d5c61caa599975cc39b961191259bbccc233edca59bc7252b9"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE job_perms SET email = $1 WHERE email = $2 AND job_id IN (SELECT id FROM v2_job_queue)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d9b877b970b8e7e5d895d9b4f83ba945776b17c153b77cb2933864e9cd313439"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE folder SET default_permissioned_as = (\n            SELECT jsonb_agg(\n                CASE WHEN rule->>'permissioned_as' = $2\n                     THEN jsonb_set(rule, ARRAY['permissioned_as'], to_jsonb($1::text))\n                     ELSE rule END\n                ORDER BY ord)\n            FROM jsonb_array_elements(default_permissioned_as) WITH ORDINALITY AS t(rule, ord))\n        WHERE default_permissioned_as @> jsonb_build_array(jsonb_build_object('permissioned_as', $2::text))",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "de902e3eded6c244ea4b4594c30e32b9e86b1902bfa504d60f0b3bef8f1e8ea6"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE password SET email = 'Legacy@windmill.dev' WHERE email = 'renamed@windmill.dev'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "df893ad32fd18b55d975247289d32fb80cbd8668c7c50c1f0b2ebd5faf50dce1"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace SET owner = $1 WHERE owner = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e084010d1ff7492b77156c01bdaebae14f8c39ba0715fc1000fb8ff1bf78169f"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(\n            SELECT 1 FROM password WHERE lower(email) = $1 AND email <> $2\n            UNION ALL SELECT 1 FROM usr WHERE lower(email) = $1 AND email <> $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "ececc8132f985b3e2d8301d5dc24356953e6e6cd71f28384b6f51231a35c7947"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE capture_config SET email = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "eeb0ef2be5db28fa722a977dfbecfe0dddc1d33b3f98c649cd91d9a3a5785e34"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE mcp_oauth_refresh_token SET user_email = $1 WHERE user_email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "eff07725675fa97c91f86564c13658316f5c3dff23704feca654dedc035adacf"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE app SET policy = jsonb_set(policy, ARRAY['on_behalf_of_email'], to_jsonb($1::text)) WHERE policy->>'on_behalf_of_email' = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f8c3c8cc0b3bb90adf611bab39f71cdd9a96e35508c852ab9c1dcbae3decbaba"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO schedule(workspace_id, path, edited_by, schedule, timezone, enabled, script_path, is_flow, args, email, permissioned_as)\n         VALUES ('test-workspace', 'u/test-user-2/sched', 'test-user-2', '0 0 1 1 *', 'UTC', false, 'u/test-user-2/s', false, '{}'::json, 'test2@windmill.dev', 'test2@windmill.dev')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "f8fa9ba552697292752e80a224e9c85bbed7cb0196c40927bf5a542416cf3b15"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE schedule SET permissioned_as = $1 WHERE permissioned_as = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "fbd24c0bc6c364d643f81f0a1baa4db6beaa12b133932ce0fd6c16a771a5a594"
+}
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO datatable_owner_role (dbname, role_name, password, workspace_id)\n         VALUES ($1, $2, $3, $4)\n         ON CONFLICT (dbname) DO UPDATE SET\n            role_name = EXCLUDED.role_name,\n            password = EXCLUDED.password,\n            workspace_id = EXCLUDED.workspace_id",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "fdb0fe445cb882c4010b2581e16d0adcf23343926f9610ed802ea1aaca34ebb3"
+}
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS datatable_owner_role;
@@ -0,0 +1,23 @@
+-- Dedicated per-database owner role for permissions-enabled instance data
+-- tables. The shared `custom_instance_user` is handed to user SQL on every
+-- unprotected instance data table and holds CONNECT on the whole cluster, so a
+-- non-admin could set its password (`ALTER ROLE CURRENT_USER PASSWORD`) and
+-- reconnect to a protected database as the owner, bypassing the per-caller
+-- grants. Protected databases therefore get an owner role that is never handed
+-- to non-admin SQL, and `custom_instance_user` loses CONNECT on them.
+CREATE TABLE datatable_owner_role (
+    dbname TEXT PRIMARY KEY,
+    role_name TEXT NOT NULL,
+    -- Encrypted with the workspace key of `workspace_id`, like the ephemeral
+    -- role passwords. Deprovisioned (ownership reassigned back) whenever the
+    -- data table, its permissions or its workspace go away, so the row never
+    -- outlives the key that decrypts it.
+    password TEXT NOT NULL,
+    workspace_id VARCHAR(50) NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX datatable_owner_role_workspace_idx ON datatable_owner_role (workspace_id);
+
+GRANT ALL ON datatable_owner_role TO windmill_user;
+GRANT ALL ON datatable_owner_role TO windmill_admin;
@@ -12,6 +12,7 @@ use windmill_audit::audit_oss::audit_log;
 use windmill_audit::ActionKind;
 use windmill_common::datatable_permissions::{
    compute_effective_grants, datatable_license_valid, datatable_permissions_enabled,
+    deprovision_instance_owner_role_best_effort, provision_instance_owner_role,
    snapshot_datatable_roles, teardown_snapshot_roles_best_effort, validate_grant_identifier,
    PERMISSIONED_AS_FOLDER_PREFIX,
 };
@@ -240,6 +241,15 @@ async fn set_datatable_permissions(
    let config = get_datatable_config(&db, &w_id, &datatable_name).await?;
    if perms.enabled {
        check_shared_database_forbid(&db, &w_id, &datatable_name, &config.database, true).await?;
+        // Instance databases get a dedicated owner role before the CREATEROLE
+        // preflight, so the preflight validates the role that will actually
+        // mint the ephemeral roles. Failing here (rather than lazily on first
+        // access) gives the admin the error while they are looking at it.
+        if config.database.resource_type
+            == windmill_common::workspaces::DataTableCatalogResourceType::Instance
+        {
+            provision_instance_owner_role(&db, &w_id, &config.database.resource_path).await?;
+        }
        check_owner_can_create_roles(&db, &w_id, &config).await?;
    }
    let args_for_audit = format!("{:?}", perms);
@@ -281,6 +291,16 @@ async fn set_datatable_permissions(
    // nothing outlives the edit (active holders get NOLOGIN'd, the rest drop).
    teardown_snapshot_roles_best_effort(&db, &w_id, pre_edit_roles).await;

+    // Disabling hands the instance database back to `custom_instance_user`, so
+    // the ordinary shared-role path keeps working and no bookkeeping row
+    // outlives the workspace key that decrypts it.
+    if !perms.enabled
+        && config.database.resource_type
+            == windmill_common::workspaces::DataTableCatalogResourceType::Instance
+    {
+        deprovision_instance_owner_role_best_effort(&db, &config.database.resource_path).await;
+    }
+
    Ok(format!(
        "Edited permissions of data table {datatable_name} in workspace {w_id}"
    ))
@@ -3161,6 +3161,24 @@ async fn edit_datatable_config(
        }
    }

+    // Instance databases the saved config no longer uses (deleted data table
+    // or a database change) lose their dedicated owner role after the commit.
+    let mut dbnames_to_release: Vec<String> = vec![];
+    for (name, old) in old_datatables.iter() {
+        if old.database.resource_type != DataTableCatalogResourceType::Instance {
+            continue;
+        }
+        let still_used = new_config.settings.datatables.values().any(|dt| {
+            dt.database.resource_type == DataTableCatalogResourceType::Instance
+                && dt.database.resource_path == old.database.resource_path
+        });
+        if !still_used
+            && (new_config.deleted_datatables.contains(name)
+                || database_changed_names.contains(name))
+        {
+            dbnames_to_release.push(old.database.resource_path.clone());
+        }
+    }
    let args_for_audit = format!("{:?}", new_config.settings);
    audit_log(
        &mut *tx,
@@ -3242,6 +3260,15 @@ async fn edit_datatable_config(
    )
    .await;

+    // Hand back any protected instance database this config no longer points
+    // at: its owner-role row must not outlive the data table that justified it.
+    for dbname in dbnames_to_release {
+        windmill_common::datatable_permissions::deprovision_instance_owner_role_best_effort(
+            &db, &dbname,
+        )
+        .await;
+    }
+
    Ok(format!("Edit datatable config for workspace {}", &w_id))
 }

@@ -4809,6 +4836,37 @@ async fn set_encryption_key(
                .execute(&mut *tx)
                .await?;
        }
+
+        // Same for the dedicated owner roles of this workspace's protected
+        // instance databases (their passwords use the workspace key too).
+        let owner_roles = sqlx::query!(
+            "SELECT dbname, password FROM datatable_owner_role WHERE workspace_id = $1",
+            w_id
+        )
+        .fetch_all(&mut *tx)
+        .await?;
+        for row in owner_roles {
+            sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('wm_dto:' || $1, 0))")
+                .bind(&row.dbname)
+                .execute(&mut *tx)
+                .await?;
+            let password = encrypt(
+                &new_encryption_key,
+                &decrypt(&previous_encryption_key, row.password).map_err(|e| {
+                    Error::internal_err(format!(
+                        "Error decrypting owner role password of {}: {}",
+                        row.dbname, e
+                    ))
+                })?,
+            );
+            sqlx::query!(
+                "UPDATE datatable_owner_role SET password = $1 WHERE dbname = $2",
+                password,
+                row.dbname
+            )
+            .execute(&mut *tx)
+            .await?;
+        }
        for row in dt_roles {
            let password = encrypt(
                &new_encryption_key,
@@ -6586,6 +6644,52 @@ async fn deprecated_create_workspace_fork(_authed: ApiAuthed) -> Result<String>
 }

 /// Return the uuids of the git sync jobs to create the branch before creating the fork
+/// A fork keeping the original database (data table absent from
+/// `forked_datatables`) would query the parent's physical DB through its own
+/// copied config — bypassing or diverging from the parent's fine-grained
+/// grants. Permissions-enabled data tables must be forked, not shared, and
+/// forking one copies its schema/data through the shared role, so it is
+/// admin-only like every other whole-database operation on such a table.
+/// Checked in BOTH fork phases: the branch-creation phase must reject before
+/// any git branch is created.
+async fn check_fork_datatable_permissions(
+    db: &DB,
+    authed: &ApiAuthed,
+    parent_workspace_id: &str,
+    forked_datatable_names: &[String],
+) -> Result<()> {
+    let parent_datatables: HashMap<String, DataTable> = serde_json::from_value(
+        sqlx::query_scalar!(
+            "SELECT ws.datatable->'datatables' FROM workspace_settings ws WHERE ws.workspace_id = $1",
+            parent_workspace_id
+        )
+        .fetch_optional(db)
+        .await?
+        .flatten()
+        .unwrap_or(serde_json::Value::Null),
+    )
+    .unwrap_or_default();
+    for (name, dt) in parent_datatables {
+        if !dt.permissions.as_ref().is_some_and(|p| p.enabled) {
+            continue;
+        }
+        if !forked_datatable_names.iter().any(|n| n == &name) {
+            return Err(Error::BadRequest(format!(
+                "Data table '{name}' has fine-grained permissions enabled and cannot keep \
+                 pointing at the original database in a fork; fork its database \
+                 (schema-only or schema and data) instead."
+            )));
+        }
+        if !authed.is_admin {
+            return Err(Error::PermissionDenied(format!(
+                "Data table '{name}' has fine-grained permissions enabled: forking a \
+                 workspace containing it requires workspace admin."
+            )));
+        }
+    }
+    Ok(())
+}
+
 async fn create_workspace_fork_branch(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
@@ -6641,6 +6745,17 @@ async fn create_workspace_fork_branch(
    }
    validate_workspace_name(&nw.name)?;

+    check_fork_datatable_permissions(
+        &db,
+        &authed,
+        &w_id,
+        &nw.forked_datatables
+            .iter()
+            .map(|f| f.name.clone())
+            .collect::<Vec<_>>(),
+    )
+    .await?;
+
    // Fail before creating any git branch so a name conflict doesn't leave a
    // dangling branch on the synced repos.
    check_fork_w_id_conflict(&db, &nw.id).await?;
@@ -6954,44 +7069,16 @@ async fn create_workspace_fork(
    } else {
        None
    };
-    // A fork keeping the original database (data table absent from
-    // `forked_datatables`) would query the parent's physical DB through its own
-    // copied config — bypassing or diverging from the parent's fine-grained
-    // grants. Permissions-enabled data tables must be forked, not shared.
-    {
-        let parent_datatables: HashMap<String, DataTable> = serde_json::from_value(
-            sqlx::query_scalar!(
-                "SELECT ws.datatable->'datatables' FROM workspace_settings ws WHERE ws.workspace_id = $1",
-                &parent_workspace_id
-            )
-            .fetch_optional(&db)
-            .await?
-            .flatten()
-            .unwrap_or(serde_json::Value::Null),
-        )
-        .unwrap_or_default();
-        for (name, dt) in parent_datatables {
-            if !dt.permissions.as_ref().is_some_and(|p| p.enabled) {
-                continue;
-            }
-            if !nw.forked_datatables.iter().any(|f| f.name == name) {
-                return Err(Error::BadRequest(format!(
-                    "Data table '{name}' has fine-grained permissions enabled and cannot keep \
-                     pointing at the original database in a fork; fork its database \
-                     (schema-only or schema and data) instead."
-                )));
-            }
-            // Forking copies the schema (and possibly data) through the shared
-            // role, sidestepping per-caller grants — admin-only, like every
-            // other whole-database operation on a permissions-enabled table.
-            if !authed.is_admin {
-                return Err(Error::PermissionDenied(format!(
-                    "Data table '{name}' has fine-grained permissions enabled: forking a \
-                     workspace containing it requires workspace admin."
-                )));
-            }
-        }
-    }
+    check_fork_datatable_permissions(
+        &db,
+        &authed,
+        &parent_workspace_id,
+        &nw.forked_datatables
+            .iter()
+            .map(|f| f.name.clone())
+            .collect::<Vec<_>>(),
+    )
+    .await?;

    // Check the id conflict before the CE workspace-count limit so that
    // re-using a taken (possibly archived) fork id reports the actual
@@ -935,6 +935,24 @@ pub(crate) async fn delete_workspace(
        .bind(&w_id)
        .execute(&mut *tx)
        .await?;
+    // Same reason for the dedicated owner roles of this workspace's protected
+    // instance databases: their passwords are encrypted with the workspace key,
+    // so ownership must go back to `custom_instance_user` before it disappears.
+    let owner_dbs = sqlx::query_scalar!(
+        "SELECT dbname FROM datatable_owner_role WHERE workspace_id = $1",
+        &w_id
+    )
+    .fetch_all(&db)
+    .await?;
+    for dbname in owner_dbs {
+        windmill_common::datatable_permissions::deprovision_instance_owner_role(&db, &dbname)
+            .await
+            .map_err(|e| {
+                Error::internal_err(format!(
+                    "cannot delete workspace {w_id}: returning ownership of instance database                      {dbname} to the shared role failed: {e}. Retry when it is reachable."
+                ))
+            })?;
+    }
    windmill_common::datatable_permissions::teardown_datatable_roles_strict(&db, &w_id)
        .await
        .map_err(|e| {
@@ -1318,6 +1336,12 @@ pub async fn drop_forked_datatable_databases(
                ));
                continue;
            }
+            // The owner role owns everything in this database; hand it back
+            // before the drop so no row survives pointing at a gone database.
+            windmill_common::datatable_permissions::deprovision_instance_owner_role_best_effort(
+                &db, db_to_drop,
+            )
+            .await;
            if let Err(e) = windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
                errors.push(format!(
                    "Could not drop instance database '{}' for datatable://{}: {}",
@@ -33,6 +33,16 @@ pub const DATATABLE_EPHEMERAL_ROLE_PREFIX: &str = "wm_dt_";
 pub const WORKSPACE_ROLES_LOCK: &str =
    "SELECT pg_advisory_xact_lock(hashtextextended('wm_dt_ws:' || $1::text, 0))";

+/// Reserved prefix for the dedicated owner roles of protected instance
+/// databases. Like the ephemeral prefix, no drop path may touch a role without
+/// it.
+pub const DATATABLE_OWNER_ROLE_PREFIX: &str = "wm_dto_";
+
+/// Transaction-scoped advisory lock serializing (de)provisioning of one
+/// protected instance database's owner role. Bind the database name.
+const OWNER_ROLE_LOCK: &str =
+    "SELECT pg_advisory_xact_lock(hashtextextended('wm_dto:' || $1::text, 0))";
+
 pub const PERMISSIONED_AS_FOLDER_PREFIX: &str = "f/";

 const EPHEMERAL_ROLE_CONNECTION_LIMIT: u32 = 25;
@@ -394,6 +404,229 @@ pub fn grant_sql_statements(
    sql
 }

+// ---------------------------------------------------------------------------
+// Dedicated owner role for protected instance databases
+// ---------------------------------------------------------------------------
+
+/// Deterministic owner-role name for a protected instance database.
+pub fn instance_owner_role_name(dbname: &str) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(dbname.as_bytes());
+    let hash = hex::encode(&hasher.finalize()[..4]);
+    format!(
+        "{DATATABLE_OWNER_ROLE_PREFIX}{}_{hash}",
+        sanitize_role_part(dbname, 40)
+    )
+}
+
+/// Credentials of a protected instance database's dedicated owner role, if it
+/// has been provisioned.
+pub async fn instance_owner_creds(db: &DB, dbname: &str) -> Result<Option<(String, String)>> {
+    let Some(row) = sqlx::query!(
+        "SELECT role_name, password, workspace_id FROM datatable_owner_role WHERE dbname = $1",
+        dbname
+    )
+    .fetch_optional(db)
+    .await?
+    else {
+        return Ok(None);
+    };
+    let password = decrypt_with_refresh(db, &row.workspace_id, row.password).await?;
+    Ok(Some((row.role_name, password)))
+}
+
+/// Connect to an instance database as the main pool's user (owner of these
+/// databases), which is what can reassign ownership between roles.
+async fn connect_instance_db_as_superuser(db: &DB, dbname: &str) -> Result<tokio_postgres::Client> {
+    let mut pg = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
+    pg.dbname = dbname.to_string();
+    connect_target(&pg, db).await
+}
+
+/// Make sure a protected instance database has its dedicated owner role: a
+/// `LOGIN` role that owns the database's objects, can create the ephemeral
+/// roles, and is never handed to non-admin SQL. `custom_instance_user` loses
+/// CONNECT on the database, so a caller who learns that shared password from
+/// an unprotected data table on the same cluster cannot reach this one.
+/// Idempotent, and re-asserts the revoke on every call (superadmin re-running
+/// the instance-database setup re-grants it).
+pub async fn provision_instance_owner_role(db: &DB, w_id: &str, dbname: &str) -> Result<()> {
+    let role = instance_owner_role_name(dbname);
+    let mut tx = db.begin().await?;
+    sqlx::query(OWNER_ROLE_LOCK)
+        .bind(dbname)
+        .execute(&mut *tx)
+        .await?;
+
+    let existing = sqlx::query_scalar!(
+        "SELECT password FROM datatable_owner_role WHERE dbname = $1",
+        dbname
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let client = connect_instance_db_as_superuser(db, dbname).await?;
+    let role_present = role_exists(&client, &role).await?;
+
+    // A row whose role vanished (manual cleanup, restored cluster) must be
+    // rebuilt with a fresh password rather than trusted.
+    let password = match (&existing, role_present) {
+        (Some(encrypted), true) => decrypt_with_refresh(db, w_id, encrypted.clone()).await?,
+        _ => {
+            let password = rd_string(48);
+            if role_present {
+                client
+                    .batch_execute(&format!(
+                        "ALTER ROLE {} WITH LOGIN PASSWORD {}",
+                        quote_ident(&role),
+                        quote_literal(&password)
+                    ))
+                    .await
+                    .map_err(|e| pg_err(&format!("resetting owner role {role} password"), e))?;
+            } else {
+                client
+                    .batch_execute(&format!(
+                        "CREATE ROLE {} LOGIN NOSUPERUSER NOCREATEDB CREATEROLE NOINHERIT NOREPLICATION PASSWORD {}",
+                        quote_ident(&role),
+                        quote_literal(&password)
+                    ))
+                    .await
+                    .map_err(|e| pg_err(&format!("creating owner role {role}"), e))?;
+            }
+            password
+        }
+    };
+
+    // Take over everything `custom_instance_user` owns in THIS database (the
+    // statement is database-scoped), so the ephemeral-role grants and
+    // `ALTER DEFAULT PRIVILEGES FOR ROLE <owner>` have an owner to hang off,
+    // and give the role what migrations need to keep creating objects.
+    client
+        .batch_execute(&format!(
+            "REASSIGN OWNED BY custom_instance_user TO {role};
+             GRANT ALL ON SCHEMA public TO {role};",
+            role = quote_ident(&role)
+        ))
+        .await
+        .map_err(|e| pg_err(&format!("transferring ownership to {role}"), e))?;
+    drop(client);
+
+    // Database-level grants/revokes are executed from the main pool: any
+    // database on the cluster can carry them and the main user owns these.
+    sqlx::query(&format!(
+        "GRANT CONNECT, CREATE ON DATABASE {} TO {}",
+        quote_ident(dbname),
+        quote_ident(&role)
+    ))
+    .execute(db)
+    .await
+    .map_err(|e| Error::internal_err(format!("granting {role} access to {dbname}: {e:#}")))?;
+    sqlx::query(&format!(
+        "REVOKE CONNECT ON DATABASE {} FROM custom_instance_user",
+        quote_ident(dbname)
+    ))
+    .execute(db)
+    .await
+    .map_err(|e| {
+        Error::internal_err(format!(
+            "revoking custom_instance_user access to {dbname}: {e:#}"
+        ))
+    })?;
+
+    let mc = build_crypt(db, w_id).await?;
+    let encrypted = encrypt(&mc, &password);
+    sqlx::query!(
+        "INSERT INTO datatable_owner_role (dbname, role_name, password, workspace_id)
+         VALUES ($1, $2, $3, $4)
+         ON CONFLICT (dbname) DO UPDATE SET
+            role_name = EXCLUDED.role_name,
+            password = EXCLUDED.password,
+            workspace_id = EXCLUDED.workspace_id",
+        dbname,
+        &role,
+        &encrypted,
+        w_id
+    )
+    .execute(&mut *tx)
+    .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+/// Hand a protected instance database back to `custom_instance_user`: used
+/// when permissions are disabled, the data table is deleted or re-pointed, and
+/// on workspace deletion — the bookkeeping row must never outlive the
+/// workspace key that decrypts its password.
+pub async fn deprovision_instance_owner_role(db: &DB, dbname: &str) -> Result<()> {
+    let mut tx = db.begin().await?;
+    sqlx::query(OWNER_ROLE_LOCK)
+        .bind(dbname)
+        .execute(&mut *tx)
+        .await?;
+    let Some(role) = sqlx::query_scalar!(
+        "SELECT role_name FROM datatable_owner_role WHERE dbname = $1",
+        dbname
+    )
+    .fetch_optional(&mut *tx)
+    .await?
+    else {
+        return Ok(());
+    };
+    if !role.starts_with(DATATABLE_OWNER_ROLE_PREFIX) {
+        return Err(Error::internal_err(format!(
+            "refusing to drop role '{role}': name does not start with the reserved '{DATATABLE_OWNER_ROLE_PREFIX}' prefix"
+        )));
+    }
+
+    // The database may already be gone (dropped fork/instance database); the
+    // row must still be cleared, and the role is then privilege-free.
+    let db_exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1) AS \"e!\"",
+        dbname
+    )
+    .fetch_one(db)
+    .await?;
+    if db_exists {
+        let client = connect_instance_db_as_superuser(db, dbname).await?;
+        client
+            .batch_execute(&format!(
+                "REASSIGN OWNED BY {role} TO custom_instance_user; DROP OWNED BY {role};",
+                role = quote_ident(&role)
+            ))
+            .await
+            .map_err(|e| pg_err(&format!("returning ownership from {role}"), e))?;
+        drop(client);
+        sqlx::query(&format!(
+            "GRANT CONNECT ON DATABASE {} TO custom_instance_user",
+            quote_ident(dbname)
+        ))
+        .execute(db)
+        .await
+        .map_err(|e| {
+            Error::internal_err(format!(
+                "restoring custom_instance_user access to {dbname}: {e:#}"
+            ))
+        })?;
+    }
+    sqlx::query(&format!("DROP ROLE IF EXISTS {}", quote_ident(&role)))
+        .execute(db)
+        .await
+        .map_err(|e| Error::internal_err(format!("dropping owner role {role}: {e:#}")))?;
+    sqlx::query!("DELETE FROM datatable_owner_role WHERE dbname = $1", dbname)
+        .execute(&mut *tx)
+        .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+/// Best-effort [`deprovision_instance_owner_role`] for the paths where a
+/// failure must not block the operation (config saves, database re-points).
+pub async fn deprovision_instance_owner_role_best_effort(db: &DB, dbname: &str) {
+    if let Err(e) = deprovision_instance_owner_role(db, dbname).await {
+        tracing::warn!("deprovisioning owner role of instance database {dbname}: {e:#}");
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Role lifecycle
 // ---------------------------------------------------------------------------
@@ -724,6 +957,16 @@ async fn ensure_ephemeral_role(
        )));
    }
    let is_instance = config.database.resource_type == DataTableCatalogResourceType::Instance;
+    // Protected instance databases must be owned by their dedicated role
+    // before any ephemeral role is minted: `custom_instance_user` is exposed
+    // to non-admin SQL on unprotected data tables of the same cluster, so
+    // leaving it able to CONNECT here would make the grants bypassable. This
+    // is idempotent and also re-asserts the revoke, so a data table whose
+    // permissions were enabled without provisioning (EE default on creation,
+    // instance database created later) is healed on first non-admin access.
+    if is_instance {
+        provision_instance_owner_role(db, w_id, &config.database.resource_path).await?;
+    }
    let owner: PgDatabase =
        serde_json::from_value(datatable_shared_resource(db, w_id, &config).await?)
            .map_err(|e| Error::internal_err(format!("parsing data table owner creds: {e}")))?;
@@ -766,6 +1009,11 @@ async fn ensure_ephemeral_role(
    }

    let password = rd_string(48);
+    // Encrypt with a cipher read fresh under the locks: a concurrent key
+    // rotation (which holds the same locks) may have committed while this
+    // resolution waited, and a cached pre-rotation cipher would write
+    // ciphertext that nothing can decrypt afterwards.
+    crate::variables::WORKSPACE_CRYPT_CACHE.remove(w_id);
    let mc = build_crypt(db, w_id).await?;
    let encrypted = encrypt(&mc, &password);

@@ -914,17 +1162,20 @@ pub async fn cleanup_expired_datatable_roles(db: &DB, limit: i64) {
        }
    };
    for row in rows {
-        if let Err(e) =
-            cleanup_one_expired_role(db, &row.role_name, &row.workspace_id, &row.datatable).await
-        {
+        let outcome =
+            cleanup_one_expired_role(db, &row.role_name, &row.workspace_id, &row.datatable).await;
+        if let Err(e) = &outcome {
            tracing::warn!(
                "cleaning up expired datatable ephemeral role {}: {e:#}",
                row.role_name
            );
-            // Push the row's expiry forward so a persistently failing target
-            // (e.g. an unreachable external database) doesn't monopolize every
-            // sweep batch and starve other expired roles — each failure costs
-            // at most one attempt per backoff window.
+        }
+        // Push the row's expiry forward unless it was actually reaped: a
+        // persistently failing target (unreachable database) or a role holding
+        // long-lived sessions must not monopolize every sweep batch and starve
+        // later expired roles. Skipped roles are already NOLOGIN, so the delay
+        // costs no privilege exposure.
+        if !matches!(outcome, Ok(CleanupOutcome::Reaped)) {
            let _ = sqlx::query!(
                "UPDATE datatable_ephemeral_role SET expires_at = now() + interval '5 minutes'
                 WHERE role_name = $1 AND expires_at < now()",
@@ -936,7 +1187,20 @@ pub async fn cleanup_expired_datatable_roles(db: &DB, limit: i64) {
    }
 }

-async fn cleanup_one_expired_role(db: &DB, role: &str, w_id: &str, datatable: &str) -> Result<()> {
+enum CleanupOutcome {
+    /// Role dropped and its bookkeeping row deleted.
+    Reaped,
+    /// Left for a later sweep (lock contention, still-fresh row, or active
+    /// sessions — in the last case the role has been made NOLOGIN).
+    Deferred,
+}
+
+async fn cleanup_one_expired_role(
+    db: &DB,
+    role: &str,
+    w_id: &str,
+    datatable: &str,
+) -> Result<CleanupOutcome> {
    if !role.starts_with(DATATABLE_EPHEMERAL_ROLE_PREFIX) {
        return Err(Error::internal_err(format!(
            "refusing to clean up role '{role}' without the reserved prefix"
@@ -950,7 +1214,7 @@ async fn cleanup_one_expired_role(db: &DB, role: &str, w_id: &str, datatable: &s
    .fetch_one(&mut *tx)
    .await?;
    if !locked {
-        return Ok(());
+        return Ok(CleanupOutcome::Deferred);
    }
    // Re-check under the lock: a concurrent resolution may have refreshed it.
    let row = sqlx::query!(
@@ -960,7 +1224,7 @@ async fn cleanup_one_expired_role(db: &DB, role: &str, w_id: &str, datatable: &s
    .fetch_optional(&mut *tx)
    .await?;
    let Some(row) = row.filter(|r| r.expired) else {
-        return Ok(());
+        return Ok(CleanupOutcome::Deferred);
    };

    match resolve_role_target(db, w_id, datatable, row.owner_creds).await {
@@ -969,7 +1233,7 @@ async fn cleanup_one_expired_role(db: &DB, role: &str, w_id: &str, datatable: &s
                drop_or_disable_with_instance_fallback(db, &owner, is_instance, role).await?,
                DropOutcome::SkippedActive
            ) {
-                return Ok(());
+                return Ok(CleanupOutcome::Deferred);
            }
        }
        None => {
@@ -991,7 +1255,7 @@ async fn cleanup_one_expired_role(db: &DB, role: &str, w_id: &str, datatable: &s
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;
-    Ok(())
+    Ok(CleanupOutcome::Reaped)
 }

 /// One bookkeeping row captured before a permissions/config edit commits.
@@ -1158,6 +1158,15 @@ async fn datatable_shared_resource_inner(
        if replication {
            pg_creds.user = Some("custom_instance_replication_user".to_string());
            pg_creds.password = Some(get_custom_pg_instance_replication_password(&db).await?);
+        } else if let Some((role, password)) =
+            crate::datatable_permissions::instance_owner_creds(db, &pg_creds.dbname).await?
+        {
+            // Protected database: its dedicated owner role, never
+            // `custom_instance_user` — that shared role is exposed to
+            // non-admin SQL on unprotected data tables and loses CONNECT here
+            // (see `provision_instance_owner_role`).
+            pg_creds.user = Some(role);
+            pg_creds.password = Some(password);
        } else {
            pg_creds.user = Some("custom_instance_user".to_string());
            pg_creds.password = Some(get_custom_pg_instance_password(&db).await?);
@@ -345,7 +345,7 @@
 							</li>
 							<li>
 								Changes to grants, groups or folders take effect on the next access; a connection
-								already open keeps its privileges for up to 5 minutes.
+								that is already open keeps its privileges until it finishes.
 							</li>
 							<li>
 								Creating or editing Postgres triggers on this data table, and running its