feat: clone a data table under roles with its owners and grants, governed by its source

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UbrtwiYNfayrmqouBJHwGV

backend/.sqlx/query-d0548225d92e6e7d0eb9a0227a6522398d2a3ae99d0568b785e63749d6f3225a.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings ws\n           SET datatable = (\n               SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(\n                   dt.key,\n                   (SELECT CASE WHEN r.v->'governed_by'->>'workspace_id' = $2\n                               THEN jsonb_set(r.v, '{governed_by,workspace_id}', to_jsonb($1::text))\n                               ELSE r.v END\n                    FROM (SELECT CASE WHEN dt.value->'reference'->>'workspace_id' = $2\n                                     THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))\n                                     ELSE dt.value END AS v) r)\n               ))\n               FROM jsonb_each(ws.datatable->'datatables') dt\n           )\n           WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'\n             AND (ws.datatable::text LIKE '%\"reference\"%'\n                  OR ws.datatable::text LIKE '%\"governed_by\"%')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d0548225d92e6e7d0eb9a0227a6522398d2a3ae99d0568b785e63749d6f3225a"
+}

backend/.sqlx/query-e524e89440004953cc0b1cb57b8df6bc09f19cf53b141fac5aefa981463d39fa.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n               FROM workspace_settings ws\n               CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n               WHERE EXISTS (\n                   SELECT 1 FROM (VALUES ('reference'), ('governed_by')) k(link)\n                   WHERE dt.value->k.link->>'workspace_id' = $1\n                     AND dt.value->k.link->>'datatable' = $2\n               )",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "datatable!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "e524e89440004953cc0b1cb57b8df6bc09f19cf53b141fac5aefa981463d39fa"
+}

backend/.sqlx/query-f7ece5036ad92e485b5e15a70e5052b42aaf5c87e5be6333b2df67a81990c8a6.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings ws\n           SET datatable = (\n               SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(\n                   dt.key,\n                   (SELECT CASE WHEN r.v->'governed_by'->>'workspace_id' = $1\n                                     AND r.v->'governed_by'->>'datatable' = $2\n                               THEN jsonb_set(r.v, '{governed_by,datatable}', to_jsonb($3::text))\n                               ELSE r.v END\n                    FROM (SELECT CASE WHEN dt.value->'reference'->>'workspace_id' = $1\n                                           AND dt.value->'reference'->>'datatable' = $2\n                                     THEN jsonb_set(dt.value, '{reference,datatable}', to_jsonb($3::text))\n                                     ELSE dt.value END AS v) r)\n               ))\n               FROM jsonb_each(ws.datatable->'datatables') dt\n           )\n           WHERE EXISTS (\n               SELECT 1 FROM jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) d,\n                    LATERAL (VALUES ('reference'), ('governed_by')) k(link)\n               WHERE d.value->k.link->>'workspace_id' = $1\n                 AND d.value->k.link->>'datatable' = $2\n           )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f7ece5036ad92e485b5e15a70e5052b42aaf5c87e5be6333b2df67a81990c8a6"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-6350c1c98965043d06b2cc27a97de8b5d7c8e8a2
+e7c31069e956fa7ae873edf64c1ea5f6343d9b27

backend/migrations/20260914133544_datatable_clone.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS datatable_clone;

backend/migrations/20260914133544_datatable_clone.up.sql

@@ -0,0 +1,19 @@
+-- A database created to hold a copy of a data table, until a fork takes it.
+--
+-- Creating the copy and creating the fork are separate requests, and the fork request names the
+-- database it takes. Without a record of what each copy was made from and for whom, a fork could
+-- name any `wm_fork_*` database — another workspace's copy, full of rows its members were never
+-- given. A fork takes one only when it was copied from the data table it forks, by the user
+-- creating the fork, and no fork has taken it yet.
+CREATE TABLE datatable_clone (
+    dbname VARCHAR(63) PRIMARY KEY,
+    source_workspace_id VARCHAR(50) NOT NULL,
+    source_datatable VARCHAR(255) NOT NULL,
+    created_by VARCHAR(255) NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    -- The fork that took it. NULL while nothing has.
+    claimed_by_workspace_id VARCHAR(50)
+);
+
+GRANT ALL ON datatable_clone TO windmill_user;
+GRANT ALL ON datatable_clone TO windmill_admin;

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

@@ -619,21 +619,183 @@ async fn a_rename_has_to_match_the_save_it_claims_to_describe(
     Ok(())
 }
 
+async fn database_exists(db: &Pool<Postgres>, name: &str) -> anyhow::Result<bool> {
+    Ok(
+        sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)")
+            .bind(name)
+            .fetch_one(db)
+            .await?,
+    )
+}
+
 #[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
-async fn a_data_table_under_roles_is_not_copied_into_a_fork(
+async fn a_data_table_under_roles_is_cloned_only_with_its_grants(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let parent = format!("http://localhost:{port}/api/w/test-workspace/workspaces");
+    // Unique per test database: databases are cluster-wide, and sibling runs share the cluster.
+    let test_db: String = sqlx::query_scalar("SELECT current_database()::text")
+        .fetch_one(&db)
+        .await?;
+    let target = format!("wm_fork_{}", test_db.trim_start_matches('_').to_lowercase());
+
+    // Rows are a workspace admin's to copy, as they were before roles.
+    let resp = authed(
+        client().post(format!("{parent}/clone_pg_database")),
+        "SECRET_TOKEN_2",
+    )
+    .json(
+        &json!({"source": "datatable://main", "target_dbname": target,
+                      "fork_behavior": "schema_and_data"}),
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status(), 403, "{}", resp.text().await?);
+    assert!(!database_exists(&db, &target).await?);
+
+    // The copy an admin asks for goes ahead in an edition that replays grants, and is refused
+    // before any database exists in one that does not. The fixture's database does not exist, so
+    // the dump fails either way — and leaves nothing behind.
+    let resp = authed(
+        client().post(format!("{parent}/clone_pg_database")),
+        "SECRET_TOKEN",
+    )
+    .json(
+        &json!({"source": "datatable://main", "target_dbname": target,
+                      "fork_behavior": "schema_only"}),
+    )
+    .send()
+    .await?;
+    assert_ne!(resp.status(), 200);
+    let body = resp.text().await?;
+    #[cfg(all(feature = "private", feature = "enterprise"))]
+    assert!(!body.contains("Enterprise Edition"), "{body}");
+    #[cfg(not(all(feature = "private", feature = "enterprise")))]
+    assert!(body.contains("Enterprise Edition"), "{body}");
+    assert!(!database_exists(&db, &target).await?);
+
+    Ok(())
+}
+
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_clone_takes_its_roles_from_the_data_table_it_was_cloned_from(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    sqlx::query(
+        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,copy}', '{
+               "database": {"resource_type": "instance", "resource_path": "wm_fork_dt__copy"},
+               "governed_by": {"workspace_id": "test-workspace", "datatable": "main"},
+               "forked_from": {}
+           }'::jsonb) WHERE workspace_id = 'wm-fork-dt'"#,
+    )
+    .execute(&db)
+    .await?;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let fork = format!("http://localhost:{port}/api/w/wm-fork-dt/workspaces");
+
+    // `test-user-2` administers the fork, and is only a tenant of `analytics` in the parent.
+    let resp = authed(
+        client().get(format!("{fork}/datatable_usable_roles/copy")),
+        "SECRET_TOKEN_2",
+    )
+    .send()
+    .await?;
+    let body: Value = resp.json().await?;
+    assert_eq!(body["roles"], json!(["analytics"]), "{body}");
+
+    let resp = authed(
+        client().get(format!("{fork}/datatable_permissions/copy")),
+        "SECRET_TOKEN_2",
+    )
+    .send()
+    .await?;
+    let body: Value = resp.json().await?;
+    assert_eq!(body["editable"], false, "{body}");
+    assert_eq!(body["clone_of"]["workspace_id"], "test-workspace", "{body}");
+
+    // Nobody changes a clone's roles, a superadmin included: they are the source's.
+    for token in ["SECRET_TOKEN_2", "SECRET_TOKEN"] {
+        let resp = authed(
+            client().post(format!("{fork}/datatable_permissions/copy")),
+            token,
+        )
+        .json(&json!({"permissioned": false}))
+        .send()
+        .await?;
+        assert_eq!(resp.status(), 400, "{token} changed a clone's roles");
+    }
+
+    // A settings save cannot clear the link.
+    let resp = authed(
+        client().post(format!("{fork}/edit_datatable_config")),
+        "SECRET_TOKEN_2",
+    )
+    .json(&json!({
+        "settings": {"datatables": {"main": {}, "copy": {
+            "database": {"resource_type": "instance", "resource_path": "wm_fork_dt__copy"}
+        }}},
+        "renames": [], "deleted_datatables": []
+    }))
+    .send()
+    .await?;
+    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
+    let governed_by: Option<Value> = sqlx::query_scalar(
+        "SELECT datatable->'datatables'->'copy'->'governed_by' FROM workspace_settings
+         WHERE workspace_id = 'wm-fork-dt'",
+    )
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(governed_by.unwrap()["datatable"], "main");
+    Ok(())
+}
+
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_fork_takes_only_a_copy_made_for_it(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let fork = |dbname: &str| {
+        authed(
+            client().post(format!(
+                "http://localhost:{port}/api/w/test-workspace/workspaces/create_fork"
+            )),
+            "SECRET_TOKEN",
+        )
+        .json(&json!({"id": "wm-fork-taker", "name": "taker",
+                      "forked_datatables": [{"name": "main", "new_dbname": dbname}]}))
+    };
+
+    // Recorded, but made by someone else.
+    sqlx::query(
+        "INSERT INTO datatable_clone (dbname, source_workspace_id, source_datatable, created_by)
+         VALUES ('wm_fork_other__main', 'test-workspace', 'main', 'test2@windmill.dev')",
+    )
+    .execute(&db)
+    .await?;
+    for dbname in ["wm_fork_never_cloned__main", "wm_fork_other__main"] {
+        let resp = fork(dbname).send().await?;
+        assert_eq!(resp.status(), 400, "{dbname}");
+        assert!(
+            resp.text().await?.contains("not a copy"),
+            "{dbname} was taken for another reason"
+        );
+    }
+    Ok(())
+}
+
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_data_table_under_roles_is_not_copied_without_its_grants(
     db: Pool<Postgres>,
 ) -> anyhow::Result<()> {
     initialize_tracing().await;
-    // `pg_dump` carries no roles and the restore drops ACLs, so a copy would arrive with the
-    // parent's tenants and none of the grants behind them: every role but admin denied by
-    // Postgres in a data table that reads as configured. Refuse the copy rather than ship that,
-    // and refuse it before any data moves.
     let server = ApiServer::start(db.clone()).await?;
     let port = server.addr.port();
 
-    // Both halves of the clone: the database the copy would land in, then the copy itself. The
-    // first has to refuse too, or a permissioned fork leaves an empty registered database that
-    // no data table entry names and nothing collects.
     let resp = authed(
         client().post(format!(
             "http://localhost:{port}/api/w/test-workspace/workspaces/create_pg_database"

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

@@ -247,6 +247,8 @@ pub struct DatatableAclInfo {
     /// Whether this caller may plan and apply changes: they administer the data table, on an
     /// edition that has the planner.
     pub editable: bool,
+    /// Whether the data table is a clone, whose grants stay as they were copied from its source.
+    pub clone: bool,
     /// Whether the server is Postgres 17 or later, which added the `MAINTAIN` table privilege.
     pub supports_maintain: bool,
     /// The database the target lives in, which no target carries itself.
@@ -328,13 +330,40 @@ async fn connect_as_admin_unchecked(
     let pg: PgDatabase = serde_json::from_value(resource)
         .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {e}")))?;
     let dbname = pg.dbname.clone();
-    let (client, mut connection) = pg.connect(Some(db)).await?;
+    let (client, notices) = connect_with_notices(db, &pg).await?;
+    Ok((client, notices, dbname))
+}
+
+/// A connection to `pg`, and the notices Postgres sends on it — which is where a grant or revoke
+/// that changed nothing is reported ([`execute_acl_statements`]).
+pub(crate) async fn connect_with_notices(
+    db: &DB,
+    pg: &PgDatabase,
+) -> Result<(tokio_postgres::Client, mpsc::UnboundedReceiver<DbError>)> {
+    let (client, connection) = pg.connect(Some(db)).await?;
+    Ok((
+        client,
+        drive_with_notices(connection, windmill_common::TokioPgConnection::poll_message),
+    ))
+}
+
+type PollMessage<C> =
+    fn(
+        &mut C,
+        &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<std::result::Result<AsyncMessage, tokio_postgres::Error>>>;
+
+/// Drive `connection` in the background with `poll`, forwarding its notices.
+pub(crate) fn drive_with_notices<C: Send + 'static>(
+    mut connection: C,
+    poll: PollMessage<C>,
+) -> mpsc::UnboundedReceiver<DbError> {
     // Unbounded: the driver must never wait on the receiver, which only drains once the statement
     // the driver is carrying has completed.
     let (notices_tx, notices) = mpsc::unbounded_channel();
     tokio::spawn(async move {
         loop {
-            match std::future::poll_fn(|cx| connection.poll_message(cx)).await {
+            match std::future::poll_fn(|cx| poll(&mut connection, cx)).await {
                 Some(Ok(AsyncMessage::Notice(notice))) => {
                     let _ = notices_tx.send(notice);
                 }
@@ -347,7 +376,38 @@ async fn connect_as_admin_unchecked(
             }
         }
     });
-    Ok((client, notices, dbname))
+    notices
+}
+
+/// Run `statements` in order on `tx`, failing on the first that errors or that Postgres only warns
+/// about. A privilege the connection cannot pass on is a warning to Postgres (`01007` / `01006`),
+/// which then carries on having changed nothing; returning drops the transaction, rolling back
+/// everything before it.
+pub(crate) async fn execute_acl_statements(
+    tx: &tokio_postgres::Transaction<'_>,
+    notices: &mut mpsc::UnboundedReceiver<DbError>,
+    statements: &[String],
+) -> Result<()> {
+    while notices.try_recv().is_ok() {}
+    for statement in statements {
+        tx.batch_execute(statement).await.map_err(|e| {
+            Error::ExecutionErr(format!(
+                "Failed to run `{statement}`: {}",
+                pg_error_message(&e)
+            ))
+        })?;
+        while let Ok(notice) = notices.try_recv() {
+            if *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_GRANTED
+                || *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_REVOKED
+            {
+                return Err(Error::ExecutionErr(format!(
+                    "`{statement}` did not take effect ({}), so nothing was applied",
+                    notice.message()
+                )));
+            }
+        }
+    }
+    Ok(())
 }
 
 /// An object whose ownership follows the schema's.
@@ -392,10 +452,12 @@ macro_rules! schema_owned_objects {
                           AND x.refobjsubid <> 0)))"
     };
 }
+#[allow(unused_imports)]
+pub(crate) use schema_owned_objects;
 
 /// The keyword `ALTER ... OWNER TO` takes for a kind of object, as `pg_identify_object` names the
 /// kind. A kind missing here is refused rather than skipped, which would leave it behind.
-fn owned_keyword(kind: &str) -> Option<&'static str> {
+pub(crate) fn owned_keyword(kind: &str) -> Option<&'static str> {
     Some(match kind {
         "table" => "TABLE",
         "view" => "VIEW",
@@ -1047,6 +1109,7 @@ async fn get_datatable_acl(
         owner: role_name_of(&owner),
         roles,
         editable,
+        clone: governing.governor.is_some(),
         supports_maintain,
         dbname,
         grants,
@@ -1545,27 +1608,7 @@ async fn apply_datatable_acl(
             pg_error_message(&e)
         ))
     })?;
-    for statement in &plan.statements {
-        pg_tx.batch_execute(statement).await.map_err(|e| {
-            Error::ExecutionErr(format!(
-                "Failed to run `{statement}`: {}",
-                pg_error_message(&e)
-            ))
-        })?;
-        // A privilege the connection cannot pass on is only a warning to Postgres, which then
-        // carries on having changed nothing. Returning drops the transaction, rolling back
-        // everything before it.
-        while let Ok(notice) = notices.try_recv() {
-            if *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_GRANTED
-                || *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_REVOKED
-            {
-                return Err(Error::ExecutionErr(format!(
-                    "`{statement}` did not take effect ({}), so nothing was applied",
-                    notice.message()
-                )));
-            }
-        }
-    }
+    execute_acl_statements(&pg_tx, &mut notices, &plan.statements).await?;
 
     // A schema's objects were listed before the transaction opened; one committed since would stay
     // with its old owner. One created while this transaction is still open can still slip past, as

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

@@ -0,0 +1,319 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Copying a data table's database for a fork, as one request.
+//!
+//! The copy is created, filled and — for a data table under roles — given the source's owners and
+//! grants in one server-side operation, and a failure anywhere after the database exists drops it
+//! again. The fork request then takes the copy by name ([`crate::workspaces`]'s
+//! `claim_datatable_clone`), which only a copy recorded here, or by `create_pg_database`, allows.
+
+use std::collections::BTreeSet;
+
+use axum::{
+    extract::{Extension, Path},
+    Json,
+};
+use serde::Deserialize;
+
+use windmill_api_auth::ApiAuthed;
+use windmill_audit::audit_oss::audit_log;
+use windmill_audit::ActionKind;
+use windmill_common::datatable_roles::{lock_role_catalog, read_role_catalog_tx};
+use windmill_common::error::{pg_error_message, Error, Result};
+use windmill_common::utils::require_admin;
+use windmill_common::worker::CLOUD_HOSTED;
+use windmill_common::workspaces::{
+    get_datatable_resource_from_db_unchecked, parse_datatable_ref_for, DataTableForkBehavior,
+    GoverningDatatable,
+};
+use windmill_common::{PgDatabase, DB};
+
+use crate::datatable_acl::connect_with_notices;
+use crate::workspaces::{
+    create_database_on_server, ensure_datatable_is_clonable, pg_dump_database, pg_import_dump,
+    record_datatable_clone, DumpFile, PgDumpOptions,
+};
+
+#[derive(Deserialize)]
+pub struct ClonePgDatabaseRequest {
+    /// `datatable://<name>`: the data table to copy.
+    pub source: String,
+    /// The database the copy lands in. This request creates it.
+    pub target_dbname: String,
+    pub fork_behavior: DataTableForkBehavior,
+}
+
+/// Copy data table `source` of this workspace into a new database `target_dbname`.
+///
+/// Who may copy is what it was before data table roles: anyone for the schema, an admin of this
+/// workspace for the rows. A copy of a data table under roles is safe to hand to a fork because
+/// the fork takes it governed by the source's roles, and the replay gives those roles exactly the
+/// privileges they hold on the source.
+pub(crate) async fn clone_pg_database(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Json(req): Json<ClonePgDatabaseRequest>,
+) -> Result<String> {
+    let reference = req.source.strip_prefix("datatable://").ok_or_else(|| {
+        Error::BadRequest(format!(
+            "A clone copies a data table: expected 'datatable://<name>', got '{}'",
+            req.source
+        ))
+    })?;
+    let (name, role) = parse_datatable_ref_for(&db, &w_id, reference).await?;
+    if role.is_some() {
+        return Err(Error::BadRequest(format!(
+            "A clone copies the whole data table whatever the role; name it without `?role=`: \
+             'datatable://{name}'"
+        )));
+    }
+    let schema_only = match req.fork_behavior {
+        DataTableForkBehavior::KeepOriginal => {
+            return Err(Error::BadRequest(
+                "Keeping the original database copies nothing".to_string(),
+            ))
+        }
+        DataTableForkBehavior::SchemaOnly => true,
+        DataTableForkBehavior::SchemaAndData => {
+            require_admin(authed.is_admin, &authed.username)?;
+            if *CLOUD_HOSTED {
+                return Err(Error::BadRequest(
+                    "Cloning schema and data is not available on cloud".to_string(),
+                ));
+            }
+            false
+        }
+    };
+    windmill_common::validate_dbname(&req.target_dbname)?;
+    if !req.target_dbname.starts_with("wm_fork_")
+        && !windmill_api_auth::is_super_admin_authed(&db, &authed).await?
+    {
+        return Err(Error::BadRequest(
+            "Non-superadmin users can only clone into databases whose names start with 'wm_fork_'"
+                .to_string(),
+        ));
+    }
+    let governing = ensure_datatable_is_clonable(&db, &w_id, &name).await?;
+    if governing.datatable.permissions.is_some() {
+        crate::datatable_replay_oss::ensure_replay()?;
+    }
+
+    // Detached from the request: a client that goes away mid-copy must still leave either a
+    // finished copy or no database at all, and dropping the handler's future would skip the
+    // cleanup.
+    let clone = CloneJob { db, authed, w_id, name, target: req.target_dbname, schema_only };
+    tokio::spawn(clone.run(governing))
+        .await
+        .map_err(|e| Error::internal_err(format!("The clone stopped unexpectedly: {e}")))?
+}
+
+struct CloneJob {
+    db: DB,
+    authed: ApiAuthed,
+    w_id: String,
+    name: String,
+    target: String,
+    schema_only: bool,
+}
+
+impl CloneJob {
+    async fn run(self, governing: GoverningDatatable) -> Result<String> {
+        let source_pg: PgDatabase = serde_json::from_value(
+            get_datatable_resource_from_db_unchecked(&self.db, &self.w_id, &self.name).await?,
+        )
+        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {e}")))?;
+        let is_instance = governing.is_instance();
+
+        // Dumped before the database exists, so a source that cannot be read leaves nothing
+        // behind. Ownership never carries over: the restore runs as the target's connection user.
+        // Grants do, except on the instance, where the replay below is what reproduces them.
+        let dump = pg_dump_database(
+            &source_pg,
+            PgDumpOptions {
+                schema_only: self.schema_only,
+                no_owner: true,
+                no_acl: is_instance,
+                ..Default::default()
+            },
+        )
+        .await?;
+
+        if is_instance {
+            windmill_common::create_custom_instance_database(&self.db, &self.target, "datatable")
+                .await?;
+        } else {
+            create_database_on_server(&self.db, &source_pg, &self.target).await?;
+        }
+
+        let target_pg = PgDatabase { dbname: self.target.clone(), ..source_pg.clone() };
+        if let Err(e) = self.fill(&governing, &source_pg, &target_pg, &dump).await {
+            let dropped = if is_instance {
+                windmill_common::drop_custom_instance_database(&self.db, &self.target).await
+            } else {
+                drop_database_on_server(&self.db, &source_pg, &self.target).await
+            };
+            if let Err(drop_err) = dropped {
+                tracing::error!(
+                    "Could not drop '{}' after a failed clone of data table '{}': {drop_err}",
+                    self.target,
+                    self.name
+                );
+                return Err(Error::ExecutionErr(format!(
+                    "{e}. The database '{}' this clone created could not be dropped: {drop_err}",
+                    self.target
+                )));
+            }
+            return Err(e);
+        }
+        Ok(format!(
+            "Cloned data table '{}' into '{}'",
+            self.name, self.target
+        ))
+    }
+
+    /// Restore the dump into the new database, replay the owners and grants of a data table under
+    /// roles, and record the copy for the fork to take.
+    async fn fill(
+        &self,
+        governing: &GoverningDatatable,
+        source_pg: &PgDatabase,
+        target_pg: &PgDatabase,
+        dump: &DumpFile,
+    ) -> Result<()> {
+        pg_import_dump(target_pg, dump).await?;
+
+        // Held until the copy is recorded: a role renamed or dropped meanwhile would change what
+        // the replay names, and a settings save could move the source onto another database or put
+        // it under roles. Taken in the same order as the permissions save and the ACL apply.
+        let mut tx = self.db.begin().await?;
+        lock_role_catalog(&mut tx).await?;
+        let mut rows = vec![
+            governing.workspace_id.clone(),
+            governing.governing_workspace_id().to_string(),
+        ];
+        rows.sort();
+        rows.dedup();
+        sqlx::query(
+            "SELECT 1 FROM workspace_settings WHERE workspace_id = ANY($1)
+             ORDER BY workspace_id FOR UPDATE",
+        )
+        .bind(&rows)
+        .fetch_all(&mut *tx)
+        .await?;
+
+        // Everything so far was decided before the locks.
+        let now = ensure_datatable_is_clonable(&self.db, &self.w_id, &self.name).await?;
+        let same_database = match (&now.datatable.database, &governing.datatable.database) {
+            (Some(now), Some(then)) => {
+                now.resource_type == then.resource_type && now.resource_path == then.resource_path
+            }
+            _ => false,
+        };
+        if !same_database {
+            return Err(Error::BadRequest(format!(
+                "Data table '{}' moved to another database while it was being copied; clone it \
+                 again",
+                self.name
+            )));
+        }
+
+        let replayed = now.datatable.permissions.is_some();
+        let mut replay = None;
+        if replayed {
+            crate::datatable_replay_oss::ensure_replay()?;
+            let catalog = read_role_catalog_tx(&mut tx).await?;
+            let catalog_roles: BTreeSet<String> =
+                catalog.values().map(|r| r.name.clone()).collect();
+            let (source, _source_notices) = connect_with_notices(&self.db, source_pg).await?;
+            let (target, notices) = connect_with_notices(&self.db, target_pg).await?;
+            replay = Some((source, target, notices, catalog_roles));
+        }
+
+        record_datatable_clone(&mut *tx, &self.target, &self.w_id, &self.name, &self.authed)
+            .await?;
+        let behavior = if self.schema_only {
+            "schema_only"
+        } else {
+            "schema_and_data"
+        };
+        audit_log(
+            &mut *tx,
+            &self.authed,
+            "workspaces.clone_datatable",
+            ActionKind::Create,
+            &self.w_id,
+            Some(&self.name),
+            Some(
+                [
+                    ("database", self.target.as_str()),
+                    ("fork_behavior", behavior),
+                ]
+                .into(),
+            ),
+        )
+        .await?;
+
+        if let Some((source, mut target, mut notices, catalog_roles)) = replay {
+            // One transaction on the copy: a replay that stops halfway leaves objects owned by one
+            // role and granted as another. It commits before the record does; if the record then
+            // fails, the database is dropped all the same.
+            let pg_tx = target.transaction().await.map_err(|e| {
+                Error::internal_err(format!(
+                    "Failed to open a transaction on the copy: {}",
+                    pg_error_message(&e)
+                ))
+            })?;
+            crate::datatable_replay_oss::replay_owners_and_grants(
+                &source,
+                &pg_tx,
+                &mut notices,
+                &catalog_roles,
+            )
+            .await?;
+            pg_tx.commit().await.map_err(|e| {
+                Error::internal_err(format!(
+                    "Failed to commit the replayed grants: {}",
+                    pg_error_message(&e)
+                ))
+            })?;
+        }
+        tx.commit().await?;
+
+        if replayed {
+            windmill_common::feature_usage::log_feature_usage(
+                "datatable",
+                "clone_replayed",
+                behavior,
+            );
+        }
+        Ok(())
+    }
+}
+
+/// `DROP DATABASE` on the server `server` connects to, disconnecting whoever is still on it.
+async fn drop_database_on_server(db: &DB, server: &PgDatabase, dbname: &str) -> Result<()> {
+    windmill_common::validate_dbname(dbname)?;
+    let (client, connection) = server.connect(Some(db)).await?;
+    let join_handle = tokio::spawn(async move { connection.await });
+    let result = client
+        .execute(
+            &format!("DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"),
+            &[],
+        )
+        .await;
+    drop(client);
+    let _ = windmill_common::shutdown_pg_connection(join_handle).await;
+    result.map(|_| ()).map_err(|e| {
+        Error::internal_err(format!(
+            "Failed to drop database '{dbname}': {}",
+            pg_error_message(&e)
+        ))
+    })
+}

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

@@ -30,8 +30,8 @@ use windmill_common::datatable_roles::{read_role_catalog, ADMIN_DATATABLE_ROLE};
 use windmill_common::error::{Error, JsonResult, Result};
 use windmill_common::workspaces::{
     can_use_datatable_role_in_governing_workspace, resolve_governing_datatable,
-    DataTableCatalogResourceType, DataTablePermissions, DataTableRoleTenants, DatatableAccess,
-    GoverningDatatable, DATATABLE_TENANT_WILDCARD,
+    DataTableCatalogResourceType, DataTablePermissions, DataTableReference, DataTableRoleTenants,
+    DatatableAccess, GoverningDatatable, DATATABLE_TENANT_WILDCARD,
 };
 use windmill_common::DB;
 
@@ -72,7 +72,10 @@ struct DatatablePermissionsInfo {
     /// The workspace whose entry this is, when it is not the one asking.
     #[serde(skip_serializing_if = "Option::is_none")]
     governing_workspace_id: Option<String>,
-    /// Whether this caller may save. False from a fork, and for a non-admin.
+    /// For a clone, the data table whose roles it takes. Not editable from anywhere.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    clone_of: Option<DataTableReference>,
+    /// Whether this caller may save. False from a fork, on a clone, and for a non-admin.
     editable: bool,
     /// Every instance role the instance defines, to pick from.
     available_roles: Vec<AvailableRole>,
@@ -117,12 +120,22 @@ struct UsableDatatableRoles {
 /// Administering a data table — its permissions, its migrations that declare no role, its exports
 /// — is for the admins of the workspace that governs it. A fork can use the data table; it never
 /// administers it.
+///
+/// A clone is refused whoever asks: its roles are the data table it was cloned from, changed there,
+/// and its grants stay as they were copied.
 pub(crate) async fn ensure_governs_datatable(
     db: &DB,
     authed: &ApiAuthed,
     w_id: &str,
     governing: &GoverningDatatable,
 ) -> Result<()> {
+    if let Some(governor) = governing.governor.as_ref() {
+        return Err(Error::BadRequest(format!(
+            "Data table '{}' is a clone of data table '{}' of workspace '{}', which decides its \
+             roles; change them there. Its grants stay as they were copied.",
+            governing.name, governor.datatable, governor.workspace_id
+        )));
+    }
     if governing.workspace_id == w_id && authed.is_admin {
         return Ok(());
     }
@@ -162,7 +175,7 @@ pub(crate) async fn ensure_reaches_datatable(
         }
         if can_use_datatable_role_in_governing_workspace(
             db,
-            &governing.workspace_id,
+            governing.governing_workspace_id(),
             w_id,
             tenants,
             &access,
@@ -271,8 +284,9 @@ async fn get_datatable_permissions(
             .map(|p| p.default_role().to_string())
             .unwrap_or_else(|| ADMIN_DATATABLE_ROLE.to_string()),
         roles,
-        governing_workspace_id: (governing.workspace_id != w_id)
-            .then(|| governing.workspace_id.clone()),
+        governing_workspace_id: (governing.governing_workspace_id() != w_id)
+            .then(|| governing.governing_workspace_id().to_string()),
+        clone_of: governing.governor.clone(),
         editable,
         // The instance's role names are only of use to someone who can pick from them, and
         // enumerating them is the first step of anything that wants to name one it shouldn't.
@@ -460,22 +474,36 @@ async fn set_datatable_permissions(
 /// replication stream reads every row whatever the roles grant, so a data table carries one or the
 /// other; the listener side refuses a data table already under roles.
 async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) -> Result<()> {
-    // Every workspace holding an entry that resolves here, under the name it calls it: the
-    // governing one, plus each fork pointing at it. A fork's trigger names its own local entry, so
-    // looking in the governing workspace alone would miss every stream a fork opened.
+    // Every workspace holding an entry that takes its roles from here, under the name it calls it:
+    // the governing one, each fork pointing at it, each clone governed by it, and each fork pointing
+    // at one of those clones. A fork's trigger names its own local entry, so looking in the
+    // governing workspace alone would miss every stream a fork opened — and a clone's stream reads
+    // the copied rows.
     let mut reached = vec![(governing.workspace_id.clone(), governing.name.clone())];
-    let pointers = sqlx::query!(
-        r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!"
-           FROM workspace_settings ws
-           CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
-           WHERE dt.value->'reference'->>'workspace_id' = $1
-             AND dt.value->'reference'->>'datatable' = $2"#,
-        &governing.workspace_id,
-        &governing.name,
-    )
-    .fetch_all(db)
-    .await?;
-    reached.extend(pointers.into_iter().map(|r| (r.workspace_id, r.datatable)));
+    let mut next = 0;
+    while next < reached.len() {
+        let (w_id, name) = reached[next].clone();
+        next += 1;
+        let linked = sqlx::query_as::<_, (String, String)>(
+            r#"SELECT ws.workspace_id, dt.key
+               FROM workspace_settings ws
+               CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
+               WHERE EXISTS (
+                   SELECT 1 FROM (VALUES ('reference'), ('governed_by')) k(link)
+                   WHERE dt.value->k.link->>'workspace_id' = $1
+                     AND dt.value->k.link->>'datatable' = $2
+               )"#,
+        )
+        .bind(&w_id)
+        .bind(&name)
+        .fetch_all(db)
+        .await?;
+        for entry in linked {
+            if !reached.contains(&entry) {
+                reached.push(entry);
+            }
+        }
+    }
 
     let mut streams = Vec::new();
     for (w_id, name) in reached {
@@ -551,7 +579,7 @@ async fn list_usable_datatable_roles(
         };
         if can_use_datatable_role_in_governing_workspace(
             &db,
-            &governing.workspace_id,
+            governing.governing_workspace_id(),
             &w_id,
             tenants,
             &access,

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

@@ -0,0 +1,48 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Where the replay of a data table's owners and grants into its copy comes from: the enterprise
+//! one, or a refusal. Without it a data table under roles is not copied at all: its rows would
+//! arrive owned by the admin connection with no grant for any role.
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) use crate::datatable_replay_ee::replay_owners_and_grants;
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub(crate) fn ensure_replay() -> windmill_common::error::Result<()> {
+    Ok(())
+}
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+use {
+    std::collections::BTreeSet,
+    tokio::sync::mpsc,
+    tokio_postgres::error::DbError,
+    windmill_common::error::{Error, Result},
+};
+
+/// Checked before anything is created.
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) fn ensure_replay() -> Result<()> {
+    Err(Error::BadRequest(
+        "Cloning a data table under roles is a Windmill Enterprise Edition feature: the copy \
+         needs the source's owners and grants replayed. Fork it keeping the original database \
+         instead."
+            .to_string(),
+    ))
+}
+
+#[cfg(not(all(feature = "private", feature = "enterprise")))]
+pub(crate) async fn replay_owners_and_grants(
+    _source: &tokio_postgres::Client,
+    _target: &tokio_postgres::Transaction<'_>,
+    _notices: &mut mpsc::UnboundedReceiver<DbError>,
+    _catalog_roles: &BTreeSet<String>,
+) -> Result<()> {
+    ensure_replay()
+}

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

@@ -1,9 +1,11 @@
+pub mod data_metrics;
 pub mod datatable_acl;
 pub mod datatable_acl_oss;
+pub mod datatable_clone;
 pub mod datatable_migrations;
 pub mod datatable_permissions;
+pub mod datatable_replay_oss;
 pub mod deployment_requests;
-pub mod data_metrics;
 pub mod workspaces;
 pub mod workspaces_extra;
 pub mod workspaces_oss;
@@ -13,3 +15,5 @@ pub mod workspaces_ee;
 
 #[cfg(all(feature = "private", feature = "enterprise"))]
 pub mod datatable_acl_ee;
+#[cfg(all(feature = "private", feature = "enterprise"))]
+pub mod datatable_replay_ee;

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

@@ -198,6 +198,10 @@ pub fn workspaced_service() -> Router {
             post(seed_full_diff_scan),
         )
         .route("/create_pg_database", post(create_pg_database))
+        .route(
+            "/clone_pg_database",
+            post(crate::datatable_clone::clone_pg_database),
+        )
         .route("/import_pg_database", post(import_pg_database))
         .route("/export_pg_schema", post(export_pg_schema))
         .route(
@@ -2174,8 +2178,8 @@ async fn list_datatables(
             name,
             resource_type: database.resource_type.as_ref().to_string(),
             resource_path: database.resource_path.clone(),
-            governing_workspace_id: (governing.workspace_id != w_id)
-                .then(|| governing.workspace_id.clone()),
+            governing_workspace_id: (governing.governing_workspace_id() != w_id)
+                .then(|| governing.governing_workspace_id().to_string()),
             permissioned: governing.datatable.permissions.is_some(),
         });
     }
@@ -3215,7 +3219,7 @@ async fn server_setting_names(pg_db: &PgDatabase) -> Result<HashSet<String>> {
 /// so a dump that breaks partway through imports partially and reads as a success.
 /// ON_ERROR_STOP surfaces the failure and --single-transaction makes the restore
 /// all-or-nothing, leaving the target as it was and the import retryable.
-async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) -> Result<()> {
+pub(crate) async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) -> Result<()> {
     let supported_settings = server_setting_names(target_db).await?;
     comment_out_unsupported_settings(dump_file, &supported_settings).await?;
 
@@ -3262,10 +3266,15 @@ async fn create_pg_database(
     // The copy this database is for is refused a call later, and nothing collects an instance
     // database that no data table entry names. Refuse here too, so the clone stops before one
     // exists rather than leaving an empty registered `wm_fork_…` behind.
-    if let Some(reference) = req.source.strip_prefix("datatable://") {
-        let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?;
-        ensure_datatable_is_clonable(&db, &w_id, &name).await?;
-    }
+    let source_datatable = match req.source.strip_prefix("datatable://") {
+        Some(reference) => {
+            let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?;
+            let governing = ensure_datatable_is_clonable(&db, &w_id, &name).await?;
+            ensure_copied_without_roles(&governing)?;
+            Some(name)
+        }
+        None => None,
+    };
 
     // Non-superadmin: restrict dbname to wm_fork_ prefix
     if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
@@ -3283,50 +3292,134 @@ async fn create_pg_database(
     } else {
         let source_pg =
             resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
-        let (client, connection) = source_pg.connect(Some(&db)).await?;
-        let join_handle = tokio::spawn(async move { connection.await });
+        create_database_on_server(&db, &source_pg, &req.target_dbname).await?;
+    }
 
-        let row = client
-            .query_one(
-                "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)",
-                &[&req.target_dbname],
-            )
-            .await
-            .map_err(|e| {
-                Error::internal_err(format!(
-                    "Failed to check database existence: {}",
-                    pg_error_message(&e)
-                ))
-            })?;
-        let db_exists: bool = row.get(0);
-
-        if db_exists {
-            drop(client);
-            let _ = windmill_common::shutdown_pg_connection(join_handle).await;
-            return Err(Error::BadRequest(format!(
-                "Database '{}' already exists on the resource server",
-                req.target_dbname
-            )));
-        }
-
-        client
-            .execute(&format!("CREATE DATABASE \"{}\"", &req.target_dbname), &[])
-            .await
-            .map_err(|e| {
-                Error::internal_err(format!(
-                    "Failed to create database '{}': {}",
-                    req.target_dbname,
-                    pg_error_message(&e)
-                ))
-            })?;
-
-        drop(client);
-        windmill_common::shutdown_pg_connection(join_handle).await?;
+    if let Some(name) = source_datatable {
+        record_datatable_clone(
+            &mut *db.acquire().await?,
+            &req.target_dbname,
+            &w_id,
+            &name,
+            &authed,
+        )
+        .await?;
     }
 
     Ok(format!("Created database '{}'", req.target_dbname))
 }
 
+/// `CREATE DATABASE` on the server `server` connects to, refusing a name already taken there.
+pub(crate) async fn create_database_on_server(
+    db: &DB,
+    server: &PgDatabase,
+    dbname: &str,
+) -> Result<()> {
+    windmill_common::validate_dbname(dbname)?;
+    let (client, connection) = server.connect(Some(db)).await?;
+    let join_handle = tokio::spawn(async move { connection.await });
+
+    let row = client
+        .query_one(
+            "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)",
+            &[&dbname],
+        )
+        .await
+        .map_err(|e| {
+            Error::internal_err(format!(
+                "Failed to check database existence: {}",
+                pg_error_message(&e)
+            ))
+        })?;
+    let db_exists: bool = row.get(0);
+
+    if db_exists {
+        drop(client);
+        let _ = windmill_common::shutdown_pg_connection(join_handle).await;
+        return Err(Error::BadRequest(format!(
+            "Database '{dbname}' already exists on the resource server"
+        )));
+    }
+
+    client
+        .execute(&format!("CREATE DATABASE \"{dbname}\""), &[])
+        .await
+        .map_err(|e| {
+            Error::internal_err(format!(
+                "Failed to create database '{dbname}': {}",
+                pg_error_message(&e)
+            ))
+        })?;
+
+    drop(client);
+    windmill_common::shutdown_pg_connection(join_handle).await?;
+    Ok(())
+}
+
+/// Record that `dbname` holds a copy of data table `datatable` of workspace `w_id`, made by
+/// `authed`, for [`claim_datatable_clone`] to hand to exactly one fork.
+///
+/// Called only once `dbname` was just created, so a row already under that name describes a
+/// database that no longer exists, and is replaced.
+pub(crate) async fn record_datatable_clone(
+    conn: &mut sqlx::PgConnection,
+    dbname: &str,
+    w_id: &str,
+    datatable: &str,
+    authed: &ApiAuthed,
+) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO datatable_clone (dbname, source_workspace_id, source_datatable, created_by)
+         VALUES ($1, $2, $3, $4)
+         ON CONFLICT (dbname) DO UPDATE SET source_workspace_id = EXCLUDED.source_workspace_id,
+             source_datatable = EXCLUDED.source_datatable, created_by = EXCLUDED.created_by,
+             created_at = now(), claimed_by_workspace_id = NULL",
+    )
+    .bind(dbname)
+    .bind(w_id)
+    .bind(datatable)
+    .bind(&authed.email)
+    .execute(conn)
+    .await?;
+    Ok(())
+}
+
+/// Hand the copy in `dbname` to fork `forked_w_id`, refusing unless it was copied from data table
+/// `datatable` of `parent_w_id` by this same user, and no fork has taken it.
+///
+/// Without this a fork names its database freely, so it could take another workspace's copy — or
+/// one an admin made with data for a fork of their own — and govern rows it was never given.
+async fn claim_datatable_clone(
+    tx: &mut Transaction<'_, Postgres>,
+    dbname: &str,
+    parent_w_id: &str,
+    datatable: &str,
+    forked_w_id: &str,
+    authed: &ApiAuthed,
+) -> Result<()> {
+    let claimed = sqlx::query_scalar::<_, String>(
+        "UPDATE datatable_clone SET claimed_by_workspace_id = $1
+         WHERE dbname = $2 AND source_workspace_id = $3 AND source_datatable = $4
+           AND created_by = $5 AND claimed_by_workspace_id IS NULL
+         RETURNING dbname",
+    )
+    .bind(forked_w_id)
+    .bind(dbname)
+    .bind(parent_w_id)
+    .bind(datatable)
+    .bind(&authed.email)
+    .fetch_optional(&mut **tx)
+    .await?;
+    if claimed.is_none() {
+        return Err(Error::BadRequest(format!(
+            "Database '{dbname}' is not a copy of data table '{datatable}' of workspace \
+             '{parent_w_id}' that you made and no fork has taken yet. Clone the data table \
+             again for this fork."
+        )));
+    }
+    Ok(())
+}
+
 #[derive(Deserialize)]
 struct ImportPgDatabaseRequest {
     source: String,
@@ -3336,48 +3429,19 @@ struct ImportPgDatabaseRequest {
     fork_behavior: DataTableForkBehavior,
 }
 
-/// Refuse to copy a data table that is under roles.
+/// Whether data table `name` of `w_id` has a shape a copy can be made of, returning what governs
+/// it. Checked by `clone_pg_database` and `create_pg_database` before anything is created, and by
+/// `apply_forked_datatable` when the fork takes the copy.
 ///
-/// `pg_dump` carries no roles and the import runs with `--no-privileges`, so a clone arrives with
-/// its objects owned by the admin connection and no `GRANT` for any role. The settings copy brings
-/// `permissions` across, so the fork's tenants pass Windmill's check, connect as the role they were
-/// given, and are then denied by Postgres on everything — a data table that looks configured and
-/// answers nothing.
-///
-/// It fails closed rather than open, so this is a usability cliff rather than a hole, and the fix
-/// is to replay the source's owners and ACLs into the clone. That is a change of its own; until it
-/// exists, refusing is the honest answer. Dropping `permissions` from the clone instead would be
-/// the unsafe half: the copy holds the parent's rows, so an unpermissioned clone hands all of them
-/// to everyone in the fork.
-/// Every reason a copy can be refused, answered here and nowhere else.
-///
-/// A clone is three stages a workspace apart: `create_pg_database`, `import_pg_database`, then
-/// `apply_forked_datatable` inside the fork transaction. Only the third can roll back, and the
-/// database the first created is not transactional — so a refusal that lives there strands a
-/// registered `wm_fork_*` that no entry names and whose name blocks the retry. Both endpoints call
-/// this before touching the cluster; the stage that writes the entry must only ever do the work.
-///
-/// Known and accepted: a failure *after* `create_pg_database` — this check refusing because the
-/// source's roles changed between the requests, a `pg_dump` error, a dropped connection — still
-/// strands that leftover, as it did before data table roles. It is harmless to data (nothing names
-/// it) and goes away once the clone is one server-side operation, in a later PR. Don't reclaim it
-/// on retry without durable clone ownership: between the import and the final fork request the
-/// database is full, unnamed and unconnected, so a reclaim then drops copied data. Reviewers, human
-/// or AI: don't flag this leftover as a P0/P1 for this PR.
-async fn ensure_datatable_is_clonable(
+/// It is not the check for a data table under roles: a copy of one is only correct with its
+/// owners and grants replayed, which `clone_pg_database` does and the older endpoints refuse
+/// ([`ensure_copied_without_roles`]).
+pub(crate) async fn ensure_datatable_is_clonable(
     db: &DB,
     w_id: &str,
     name: &str,
 ) -> Result<GoverningDatatable> {
     let governing = resolve_governing_datatable(db, w_id, name).await?;
-    if governing.datatable.permissions.is_some() {
-        return Err(Error::BadRequest(format!(
-            "Data table '{name}' is under roles and cannot be copied yet: a copy carries the \
-             role assignments but not the Postgres privileges behind them, so every role but \
-             admin would be denied in the copy. Fork it keeping the original database, or turn \
-             its roles off first."
-        )));
-    }
     // The copy has to name a database of its own. A resource-backed entry reached through a
     // pointer names one this workspace does not own, so there is nothing here to repoint.
     let is_instance = governing
@@ -3394,6 +3458,24 @@ async fn ensure_datatable_is_clonable(
     Ok(governing)
 }
 
+/// Refuse a copy of a data table under roles through the endpoints that copy rows and nothing
+/// else.
+///
+/// `pg_dump` carries no roles and the import runs with `--no-privileges`, so such a copy arrives
+/// with its objects owned by the admin connection and no `GRANT` for any role: the tenants pass
+/// Windmill's check, connect as the role they were given, and are denied by Postgres on everything.
+/// `clone_pg_database` replays the owners and grants instead.
+fn ensure_copied_without_roles(governing: &GoverningDatatable) -> Result<()> {
+    if governing.datatable.permissions.is_some() {
+        return Err(Error::BadRequest(format!(
+            "Data table '{}' is under roles, so a copy has to carry its owners and grants: \
+             clone it through the fork wizard or `wmill workspace fork`, which do.",
+            governing.name
+        )));
+    }
+    Ok(())
+}
+
 /// Import (pg_dump/pg_import) from source to target
 async fn import_pg_database(
     authed: ApiAuthed,
@@ -3408,7 +3490,7 @@ async fn import_pg_database(
 
     if let Some(reference) = req.source.strip_prefix("datatable://") {
         let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?;
-        ensure_datatable_is_clonable(&db, &w_id, &name).await?;
+        ensure_copied_without_roles(&ensure_datatable_is_clonable(&db, &w_id, &name).await?)?;
     }
 
     if req.fork_behavior == DataTableForkBehavior::SchemaAndData {
@@ -3755,8 +3837,11 @@ async fn edit_datatable_config(
         // hand the fork the database outright. `forked_from` is the clone stamp the fork flow
         // writes: whether an entry has one is carried the same way, since it is what marks the
         // database droppable, but the schema baseline inside it is the diff view's to advance.
+        // `governed_by` is a clone's `reference` for its roles, and clearing it would hand the
+        // fork the copied rows the same way.
         dt.permissions = old.and_then(|old| old.permissions.clone());
         dt.reference = old.and_then(|old| old.reference.clone());
+        dt.governed_by = old.and_then(|old| old.governed_by.clone());
         dt.forked_from = match old.and_then(|old| old.forked_from.as_ref()) {
             Some(stored) => Some(dt.forked_from.take().unwrap_or_else(|| stored.clone())),
             None => None,
@@ -3799,9 +3884,18 @@ async fn edit_datatable_config(
     // workspace does not own. Pointing an entry at another workspace's data table is not checked
     // here because it cannot be requested at all: `reference` is overwritten from the stored entry
     // above, for every caller.
+    //
+    // Compared against the entry the carried fields came from, not the one stored under the same
+    // name: otherwise swapping two names keeps each database in place while moving a clone's
+    // `governed_by` off the copy it governs.
     if !is_superadmin {
         for (name, dt) in new_config.settings.datatables.iter() {
-            let old_dt = old_datatables.get(name);
+            let old_dt = old_datatables.get(
+                rename_src
+                    .get(name.as_str())
+                    .copied()
+                    .unwrap_or(name.as_str()),
+            );
             if dt
                 .database
                 .as_ref()
@@ -3877,8 +3971,11 @@ async fn edit_datatable_config(
             r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!"
                FROM workspace_settings ws
                CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
-               WHERE dt.value->'reference'->>'workspace_id' = $1
-                 AND dt.value->'reference'->>'datatable' = $2"#,
+               WHERE EXISTS (
+                   SELECT 1 FROM (VALUES ('reference'), ('governed_by')) k(link)
+                   WHERE dt.value->k.link->>'workspace_id' = $1
+                     AND dt.value->k.link->>'datatable' = $2
+               )"#,
             &w_id,
             name,
         )
@@ -7857,6 +7954,7 @@ async fn point_kept_datatables_at_parent(
                 workspace_id: parent_w_id.to_string(),
                 datatable: name.clone(),
             }),
+            governed_by: None,
             forked_from: None,
             migrations_enabled: dt.migrations_enabled,
             permissions: None,
@@ -7877,7 +7975,8 @@ async fn point_kept_datatables_at_parent(
     Ok(())
 }
 
-/// Move every pointer in any workspace that names `(w_id, from)` to `(w_id, to)`.
+/// Move every pointer, and every clone's `governed_by`, in any workspace that names `(w_id, from)`
+/// to `(w_id, to)`.
 ///
 /// `EXISTS` rather than a `LIKE` over the whole document: the update rewrites the row, so matching
 /// every workspace that holds any pointer would rewrite rows to a byte-identical value and hold an
@@ -7893,17 +7992,22 @@ async fn repoint_datatable_references(
            SET datatable = (
                SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(
                    dt.key,
-                   CASE WHEN dt.value->'reference'->>'workspace_id' = $1
-                             AND dt.value->'reference'->>'datatable' = $2
-                       THEN jsonb_set(dt.value, '{reference,datatable}', to_jsonb($3::text))
-                       ELSE dt.value END
+                   (SELECT CASE WHEN r.v->'governed_by'->>'workspace_id' = $1
+                                     AND r.v->'governed_by'->>'datatable' = $2
+                               THEN jsonb_set(r.v, '{governed_by,datatable}', to_jsonb($3::text))
+                               ELSE r.v END
+                    FROM (SELECT CASE WHEN dt.value->'reference'->>'workspace_id' = $1
+                                           AND dt.value->'reference'->>'datatable' = $2
+                                     THEN jsonb_set(dt.value, '{reference,datatable}', to_jsonb($3::text))
+                                     ELSE dt.value END AS v) r)
                ))
                FROM jsonb_each(ws.datatable->'datatables') dt
            )
            WHERE EXISTS (
-               SELECT 1 FROM jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) d
-               WHERE d.value->'reference'->>'workspace_id' = $1
-                 AND d.value->'reference'->>'datatable' = $2
+               SELECT 1 FROM jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) d,
+                    LATERAL (VALUES ('reference'), ('governed_by')) k(link)
+               WHERE d.value->k.link->>'workspace_id' = $1
+                 AND d.value->k.link->>'datatable' = $2
            )"#,
         w_id,
         from,
@@ -7922,15 +8026,6 @@ async fn apply_forked_datatable(
     forked_w_id: &str,
     fdt: &ForkedDatatableInfo,
 ) -> Result<()> {
-    // Cloning reads the parent's whole schema as admin and hands the copy to the fork, so it is
-    // for the workspace that governs the data table — a fork can use one, never duplicate it.
-    windmill_common::workspaces::ensure_datatable_admin_access(
-        db,
-        parent_w_id,
-        &fdt.name,
-        &DatatableAccess::Authed(authed.to_authed_ref()),
-    )
-    .await?;
     let governing = ensure_datatable_is_clonable(db, parent_w_id, &fdt.name).await?;
     windmill_common::validate_dbname(&fdt.new_dbname)?;
     if !fdt.new_dbname.starts_with("wm_fork_") {
@@ -7939,6 +8034,25 @@ async fn apply_forked_datatable(
             fdt.new_dbname
         )));
     }
+    claim_datatable_clone(
+        tx,
+        &fdt.new_dbname,
+        parent_w_id,
+        &fdt.name,
+        forked_w_id,
+        authed,
+    )
+    .await?;
+    // The copy holds the rows of what governs the source, so that entry keeps deciding who reaches
+    // them. Settled from the source as it resolves now: the settings clone may have handed the fork
+    // a pointer, or a clone of its own.
+    let governed_by = serde_json::to_value(governing.governor.clone().unwrap_or_else(|| {
+        windmill_common::workspaces::DataTableReference {
+            workspace_id: governing.workspace_id.clone(),
+            datatable: governing.name.clone(),
+        }
+    }))
+    .map_err(|e| Error::internal_err(format!("serializing a clone's governor: {e}")))?;
 
     // Snapshot the schema from the source (parent) datatable
     let schema = snapshot_datatable_schema(db, parent_w_id, &fdt.name).await?;
@@ -7984,20 +8098,16 @@ async fn apply_forked_datatable(
             "resource_type": "instance",
             "resource_path": &fdt.new_dbname,
         });
-        sqlx::query!(
+        sqlx::query(
             r#"UPDATE workspace_settings
                SET datatable = jsonb_set(
-                   jsonb_set(
-                       datatable #- ARRAY['datatables', $2, 'reference'],
-                       ARRAY['datatables', $2, 'database'], $3::jsonb),
-                   ARRAY['datatables', $2, 'forked_from'], $4::jsonb
-               )
+                   datatable #- ARRAY['datatables', $2, 'reference'],
+                   ARRAY['datatables', $2, 'database'], $3::jsonb)
                WHERE workspace_id = $1"#,
-            forked_w_id,
-            &fdt.name,
-            new_database,
-            forked_from,
         )
+        .bind(forked_w_id)
+        .bind(&fdt.name)
+        .bind(new_database)
         .execute(&mut **tx)
         .await?;
     } else {
@@ -8021,20 +8131,26 @@ async fn apply_forked_datatable(
         )
         .execute(&mut **tx)
         .await?;
-
-        // Set forked_from on the datatable config
-        sqlx::query!(
-            r#"UPDATE workspace_settings
-               SET datatable = jsonb_set(datatable, ARRAY['datatables', $2, 'forked_from'], $3::jsonb)
-               WHERE workspace_id = $1"#,
-            forked_w_id,
-            &fdt.name,
-            forked_from,
-        )
-        .execute(&mut **tx)
-        .await?;
     }
 
+    // The settings clone copied the source's `permissions` along; a clone keeps none of its own.
+    sqlx::query(
+        r#"UPDATE workspace_settings
+           SET datatable = jsonb_set(
+               jsonb_set(
+                   datatable #- ARRAY['datatables', $2, 'permissions'],
+                   ARRAY['datatables', $2, 'governed_by'], $3::jsonb),
+               ARRAY['datatables', $2, 'forked_from'], $4::jsonb
+           )
+           WHERE workspace_id = $1"#,
+    )
+    .bind(forked_w_id)
+    .bind(&fdt.name)
+    .bind(governed_by)
+    .bind(forked_from)
+    .execute(&mut **tx)
+    .await?;
+
     Ok(())
 }
 

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

@@ -495,26 +495,40 @@ pub(crate) async fn change_workspace_id(
     // A fork's data table entry names the workspace that governs it by id, so the rename has to
     // follow there too — anywhere, not just in the reparented children: a detached workspace can
     // point at this one without being its fork. Left behind, the pointer resolves to the archived
-    // shell and every job through it stops.
+    // shell and every job through it stops. A clone's `governed_by` names it the same way.
     info!("Re-pointing data table references to the new workspace id");
     sqlx::query!(
         r#"UPDATE workspace_settings ws
            SET datatable = (
                SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(
                    dt.key,
-                   CASE WHEN dt.value->'reference'->>'workspace_id' = $2
-                       THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))
-                       ELSE dt.value END
+                   (SELECT CASE WHEN r.v->'governed_by'->>'workspace_id' = $2
+                               THEN jsonb_set(r.v, '{governed_by,workspace_id}', to_jsonb($1::text))
+                               ELSE r.v END
+                    FROM (SELECT CASE WHEN dt.value->'reference'->>'workspace_id' = $2
+                                     THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))
+                                     ELSE dt.value END AS v) r)
                ))
                FROM jsonb_each(ws.datatable->'datatables') dt
            )
            WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'
-             AND ws.datatable::text LIKE '%"reference"%'"#,
+             AND (ws.datatable::text LIKE '%"reference"%'
+                  OR ws.datatable::text LIKE '%"governed_by"%')"#,
         &rw.new_id,
         &old_id,
     )
     .execute(&mut *tx)
     .await?;
+    sqlx::query(
+        "UPDATE datatable_clone SET
+             source_workspace_id = CASE WHEN source_workspace_id = $2 THEN $1 ELSE source_workspace_id END,
+             claimed_by_workspace_id = CASE WHEN claimed_by_workspace_id = $2 THEN $1 ELSE claimed_by_workspace_id END
+         WHERE source_workspace_id = $2 OR claimed_by_workspace_id = $2",
+    )
+    .bind(&rw.new_id)
+    .bind(&old_id)
+    .execute(&mut *tx)
+    .await?;
 
     info!("Updating workspace_protection_rule table");
     sqlx::query!(
@@ -1397,6 +1411,7 @@ pub async fn drop_forked_datatable_databases(
         serde_json::from_value(datatable_config).unwrap_or_default();
 
     let mut errors: Vec<String> = Vec::new();
+    let mut dropped: Vec<String> = Vec::new();
 
     for dt_name in &req.datatable_names {
         // Only a clone is droppable, and a clone is terminal by construction: a kept data table is
@@ -1420,11 +1435,12 @@ pub async fn drop_forked_datatable_databases(
                 ));
                 continue;
             }
-            if let Err(e) = windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
-                errors.push(format!(
+            match windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
+                Ok(()) => dropped.push(db_to_drop.clone()),
+                Err(e) => errors.push(format!(
                     "Could not drop instance database '{}' for datatable://{}: {}",
                     db_to_drop, dt_name, e
-                ));
+                )),
             }
         } else {
             let fork_pg = match crate::workspaces::resolve_pg_source_checked(
@@ -1485,14 +1501,15 @@ pub async fn drop_forked_datatable_databases(
             match parent_pg.connect(Some(&db)).await {
                 Ok((client, connection)) => {
                     let join_handle = tokio::spawn(async move { connection.await });
-                    if let Err(e) = client
+                    match client
                         .execute(&format!("DROP DATABASE \"{}\"", db_to_drop), &[])
                         .await
                     {
-                        errors.push(format!(
+                        Ok(_) => dropped.push(db_to_drop.clone()),
+                        Err(e) => errors.push(format!(
                             "Could not drop database '{}' for datatable://{}: {}",
                             db_to_drop, dt_name, e
-                        ));
+                        )),
                     }
                     drop(client);
                     let _ = windmill_common::shutdown_pg_connection(join_handle).await;
@@ -1507,6 +1524,11 @@ pub async fn drop_forked_datatable_databases(
         }
     }
 
+    sqlx::query("DELETE FROM datatable_clone WHERE dbname = ANY($1)")
+        .bind(&dropped)
+        .execute(&db)
+        .await?;
+
     Ok(Json(errors))
 }
 

backend/windmill-api/openapi.yaml

@@ -5785,6 +5785,45 @@ paths:
               schema:
                 type: string
 
+  /w/{workspace}/workspaces/clone_pg_database:
+    post:
+      summary: copy a datatable into a new database for a fork, with its owners and grants
+      description: |
+        Creates `target_dbname`, restores the datatable into it, and for a datatable under roles
+        replays the source's owners, grants and default privileges. Anything that fails after the
+        database is created drops it again. The fork request then takes the copy by name.
+      operationId: clonePgDatabase
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [source, target_dbname, fork_behavior]
+              properties:
+                source:
+                  type: string
+                  description: "The datatable to copy: 'datatable://name'"
+                target_dbname:
+                  type: string
+                  description: "Name of the database to create for the copy"
+                fork_behavior:
+                  type: string
+                  enum:
+                    - schema_only
+                    - schema_and_data
+      responses:
+        "200":
+          description: status
+          content:
+            text/plain:
+              schema:
+                type: string
+
   /w/{workspace}/workspaces/drop_forked_datatable_databases:
     post:
       summary: drop forked datatable databases
@@ -32733,6 +32772,15 @@ components:
             $ref: "#/components/schemas/DatatableRoleTenants"
         governing_workspace_id:
           type: string
+        clone_of:
+          type: object
+          description: for a clone, the data table whose roles it takes
+          required: [workspace_id, datatable]
+          properties:
+            workspace_id:
+              type: string
+            datatable:
+              type: string
         editable:
           type: boolean
         available_roles:
@@ -32983,6 +33031,9 @@ components:
         editable:
           type: boolean
           description: whether the caller may plan and apply changes
+        clone:
+          type: boolean
+          description: whether this is a clone, whose grants stay as they were copied
         supports_maintain:
           type: boolean
           description: whether the server is Postgres 17+, which added the MAINTAIN table privilege
@@ -35071,6 +35122,17 @@ components:
                     type: string
                   datatable:
                     type: string
+              governed_by:
+                description: >-
+                  On a clone, the data table it was copied from, whose roles it takes. Server-owned
+                  like `reference`.
+                type: object
+                required: [workspace_id, datatable]
+                properties:
+                  workspace_id:
+                    type: string
+                  datatable:
+                    type: string
               migrations_enabled:
                 type: boolean
                 description: Whether the SQL migrations feature is opted in for this data table

backend/windmill-common/src/lib.rs

@@ -1578,6 +1578,28 @@ pub async fn create_custom_instance_database(
             error::Error::internal_err(format!("Failed to create database '{}': {}", dbname, e))
         })?;
 
+    // Nothing names a database that failed past this point, and its name blocks the retry: drop it
+    // rather than leave it behind.
+    if let Err(e) = finish_custom_instance_database(db, dbname, tag).await {
+        if let Err(drop_err) = drop_custom_instance_database(db, dbname).await {
+            tracing::error!("Could not drop '{dbname}' after failing to set it up: {drop_err}");
+        }
+        return Err(e);
+    }
+
+    // A data table role can only reach a database it may CONNECT to, and PUBLIC's default CONNECT
+    // would otherwise let every role in regardless of what this instance defines. Best-effort: a
+    // failure here leaves the database usable as `admin`, and the next role change repairs it.
+    if let Err(e) = crate::datatable_roles::converge_connect_grants(db, dbname).await {
+        tracing::warn!("Could not set CONNECT grants on instance database '{dbname}': {e}");
+    }
+
+    tracing::info!("Created custom instance database '{}'", dbname);
+    Ok(())
+}
+
+/// Grant `custom_instance_user` its privileges on a database just created, and register it.
+async fn finish_custom_instance_database(db: &DB, dbname: &str, tag: &str) -> error::Result<()> {
     // Grant permissions to custom_instance_user
     let wmill_pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
     let new_pg_creds = PgDatabase { dbname: dbname.to_string(), ..wmill_pg_creds };
@@ -1612,15 +1634,6 @@ pub async fn create_custom_instance_database(
     )
     .execute(db)
     .await?;
-
-    // A data table role can only reach a database it may CONNECT to, and PUBLIC's default CONNECT
-    // would otherwise let every role in regardless of what this instance defines. Best-effort: a
-    // failure here leaves the database usable as `admin`, and the next role change repairs it.
-    if let Err(e) = crate::datatable_roles::converge_connect_grants(db, dbname).await {
-        tracing::warn!("Could not set CONNECT grants on instance database '{dbname}': {e}");
-    }
-
-    tracing::info!("Created custom instance database '{}'", dbname);
     Ok(())
 }
 

backend/windmill-common/src/workspaces.rs

@@ -1297,6 +1297,12 @@ pub struct DataTable {
     /// nothing local for a fork admin to widen.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub reference: Option<DataTableReference>,
+    /// Set on a *clone* — a terminal entry whose database was copied from the entry this names. The
+    /// copy holds that entry's rows, so who may connect as which role stays that entry's decision:
+    /// the clone carries no `permissions` of its own and is governed like a pointer, while
+    /// connecting to its own database.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub governed_by: Option<DataTableReference>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub forked_from: Option<DataTableForkedFrom>,
     /// Whether the SQL-migrations feature is opted in for this data table.
@@ -1355,9 +1361,16 @@ pub const DATATABLE_TENANT_WILDCARD: &str = "*";
 /// enough to survive a fork of a fork.
 const DATATABLE_REFERENCE_MAX_DEPTH: usize = 20;
 
-/// Exactly one of `database` and `reference` must be set. Called wherever an entry is persisted,
-/// so nothing downstream has to handle an entry that is both or neither.
+/// Exactly one of `database` and `reference` must be set, and a clone's `governed_by` leaves no
+/// `permissions` beside it. Called wherever an entry is persisted, so nothing downstream has to
+/// handle an entry that is both or neither, or a clone with a decision of its own.
 pub fn validate_datatable_shape(name: &str, dt: &DataTable) -> Result<()> {
+    if dt.governed_by.is_some() && (dt.database.is_none() || dt.permissions.is_some()) {
+        return Err(Error::BadRequest(format!(
+            "Data table '{name}' is a clone, which owns a database and takes its roles from the \
+             data table it was cloned from"
+        )));
+    }
     match (&dt.database, &dt.reference) {
         (Some(_), None) | (None, Some(_)) => Ok(()),
         (Some(_), Some(_)) => Err(Error::BadRequest(format!(
@@ -1445,23 +1458,28 @@ pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<Dat
     Ok(serde_json::from_value::<DataTable>(datatable.clone())?)
 }
 
-/// The terminal entry a reference chain lands on: the workspace that governs the data table, the
-/// entry name there, and the entry itself. A terminal entry resolves to itself.
+/// The terminal entry a reference chain lands on, and what governs it: the workspace and name of
+/// the entry that owns the database, the entry itself, and — for a clone — the entry its
+/// `governed_by` chain lands on. A terminal entry that is not a clone resolves to itself and
+/// governs itself.
 ///
-/// Every decision downstream — which database to connect to, whose `permissions` apply, whose
-/// members tenants are evaluated against, who may administer it — is taken on this, never on the
-/// entry the caller named.
+/// Every decision downstream is taken on this, never on the entry the caller named. Which database
+/// to connect to comes from `workspace_id` / `name` / `datatable.database`; whose `permissions`
+/// apply (already in `datatable.permissions`), whose members tenants are evaluated against and who
+/// may administer it come from [`GoverningDatatable::governing_workspace_id`].
 ///
 /// Authorization: resolving deliberately crosses into the governing workspace, so it answers for a
 /// workspace the caller may not belong to and checks nothing itself. It is the input to the
 /// checks, not one of them: callers MUST pass what it returns to
 /// [`can_use_datatable_role_in_governing_workspace`] or [`ensure_datatable_admin_access`] before
-/// acting on it, and MUST NOT return its `permissions` or `workspace_id` to a caller from
+/// acting on it, and MUST NOT return its `permissions` or workspace ids to a caller from
 /// elsewhere without gating on the answer.
 pub struct GoverningDatatable {
     pub workspace_id: String,
     pub name: String,
     pub datatable: DataTable,
+    /// For a clone, the entry whose `permissions` govern it. `None` when the entry governs itself.
+    pub governor: Option<DataTableReference>,
 }
 
 impl GoverningDatatable {
@@ -1473,6 +1491,13 @@ impl GoverningDatatable {
             .as_ref()
             .is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
     }
+
+    /// The workspace whose admins administer the data table and whose members its tenants are.
+    pub fn governing_workspace_id(&self) -> &str {
+        self.governor
+            .as_ref()
+            .map_or(&self.workspace_id, |g| &g.workspace_id)
+    }
 }
 
 pub async fn resolve_governing_datatable(
@@ -1483,6 +1508,9 @@ pub async fn resolve_governing_datatable(
     let mut workspace_id = w_id.to_string();
     let mut name = name.to_string();
     let mut hops = 0;
+    // The clone a `governed_by` chain started from: it keeps its database, and takes the
+    // `permissions` of wherever the chain lands.
+    let mut clone: Option<(String, String, DataTable)> = None;
     for _ in 0..DATATABLE_REFERENCE_MAX_DEPTH {
         let datatable = read_datatable_entry(db, &workspace_id, &name)
             .await
@@ -1501,13 +1529,32 @@ pub async fn resolve_governing_datatable(
             })?;
         hops += 1;
         validate_datatable_shape(&name, &datatable)?;
-        match &datatable.reference {
-            None => return Ok(GoverningDatatable { workspace_id, name, datatable }),
-            Some(reference) => {
-                workspace_id = reference.workspace_id.clone();
-                name = reference.datatable.clone();
+        let next = match (&datatable.reference, &datatable.governed_by) {
+            (Some(reference), _) => reference.clone(),
+            (None, Some(governed_by)) => {
+                let governed_by = governed_by.clone();
+                if clone.is_none() {
+                    clone = Some((workspace_id.clone(), name.clone(), datatable));
+                }
+                governed_by
             }
-        }
+            (None, None) => {
+                return Ok(match clone {
+                    None => GoverningDatatable { workspace_id, name, datatable, governor: None },
+                    Some((clone_w_id, clone_name, mut clone_datatable)) => {
+                        clone_datatable.permissions = datatable.permissions;
+                        GoverningDatatable {
+                            workspace_id: clone_w_id,
+                            name: clone_name,
+                            datatable: clone_datatable,
+                            governor: Some(DataTableReference { workspace_id, datatable: name }),
+                        }
+                    }
+                });
+            }
+        };
+        workspace_id = next.workspace_id;
+        name = next.datatable;
     }
     Err(Error::BadRequest(format!(
         "Data table '{name}' points at another data table through more than \
@@ -1792,7 +1839,7 @@ pub async fn get_datatable_resource_from_db(
 
     if !can_use_datatable_role_in_governing_workspace(
         db,
-        &governing.workspace_id,
+        governing.governing_workspace_id(),
         w_id,
         tenants,
         &access,
@@ -1863,7 +1910,7 @@ pub async fn ensure_can_use_datatable_role(
     };
     if can_use_datatable_role_in_governing_workspace(
         db,
-        &governing.workspace_id,
+        governing.governing_workspace_id(),
         w_id,
         tenants,
         access,
@@ -1905,7 +1952,7 @@ pub async fn ensure_datatable_admin_access(
         .unwrap_or_default();
     if can_use_datatable_role_in_governing_workspace(
         db,
-        &governing.workspace_id,
+        governing.governing_workspace_id(),
         w_id,
         &admin,
         access,
@@ -1917,7 +1964,7 @@ pub async fn ensure_datatable_admin_access(
         Err(Error::NotAuthorized(format!(
             "Data table '{name}' is under roles; this reaches the whole database, so it is for \
              the admins of workspace '{}', which governs it.",
-            governing.workspace_id
+            governing.governing_workspace_id()
         )))
     }
 }
@@ -3411,6 +3458,7 @@ mod tests {
                 resource_path: "dt_main".to_string(),
             }),
             reference: None,
+            governed_by: None,
             forked_from: None,
             migrations_enabled: None,
             permissions: None,

cli/src/commands/workspace/fork.ts

@@ -286,33 +286,44 @@ async function createWorkspaceFork(
         const newDbName = `${trueWorkspaceId.replace(/-/g, "_")}__${dt.name}`;
 
         try {
-          log.info(
-            colors.blue(`  Creating database "${newDbName}" for datatable "${dt.name}"...`)
-          );
-
-          await wmill.createPgDatabase({
-            workspace: workspace.workspaceId,
-            requestBody: {
-              source: `datatable://${dt.name}`,
-              target_dbname: newDbName,
-            },
-          });
-
           log.info(
             colors.blue(
-              `  Importing ${dtBehavior === "schema_only" ? "schema" : "schema + data"}...`
+              `  Cloning datatable "${dt.name}" (${dtBehavior === "schema_only" ? "schema" : "schema + data"}) into "${newDbName}"...`
             )
           );
 
-          await wmill.importPgDatabase({
-            workspace: workspace.workspaceId,
-            requestBody: {
-              source: `datatable://${dt.name}`,
-              target: `datatable://${dt.name}`,
-              target_dbname_override: newDbName,
-              fork_behavior: dtBehavior as "schema_only" | "schema_and_data",
-            },
-          });
+          const forkBehavior = dtBehavior as "schema_only" | "schema_and_data";
+          try {
+            await wmill.clonePgDatabase({
+              workspace: workspace.workspaceId,
+              requestBody: {
+                source: `datatable://${dt.name}`,
+                target_dbname: newDbName,
+                fork_behavior: forkBehavior,
+              },
+            });
+          } catch (e: any) {
+            // A server predating the single clone request: create, then import.
+            if (e?.status !== 404) {
+              throw e;
+            }
+            await wmill.createPgDatabase({
+              workspace: workspace.workspaceId,
+              requestBody: {
+                source: `datatable://${dt.name}`,
+                target_dbname: newDbName,
+              },
+            });
+            await wmill.importPgDatabase({
+              workspace: workspace.workspaceId,
+              requestBody: {
+                source: `datatable://${dt.name}`,
+                target: `datatable://${dt.name}`,
+                target_dbname_override: newDbName,
+                fork_behavior: forkBehavior,
+              },
+            });
+          }
 
           log.info(colors.green(`  ✓ Datatable "${dt.name}" cloned.`));
           forkedDatatables.push({ name: dt.name, new_dbname: newDbName });

frontend/src/lib/components/InstanceSettings.svelte

@@ -1088,8 +1088,9 @@
 							the home page’s create menu and hub-project picker are opened and from which entry
 							point, the name of any public hub project imported from the home page and how far that
 							import got, whether data tables are put under roles and whether callers name a role or
-							take the default, and which kinds of access change (grant, revoke, ownership, default
-							privileges) are applied to data tables, last 30 days)</li
+							take the default, which kinds of access change (grant, revoke, ownership, default
+							privileges) are applied to data tables, and whether a data table under roles is cloned
+							into a fork with its schema only or with its data, last 30 days)</li
 						>
 						<li
 							>feature adoption (counts of which flow, script, trigger, worker and data table
@@ -1154,8 +1155,9 @@
 							the home page’s create menu and hub-project picker are opened and from which entry
 							point, the name of any public hub project imported from the home page and how far that
 							import got, whether data tables are put under roles and whether callers name a role or
-							take the default, and which kinds of access change (grant, revoke, ownership, default
-							privileges) are applied to data tables, last 30 days)</li
+							take the default, which kinds of access change (grant, revoke, ownership, default
+							privileges) are applied to data tables, and whether a data table under roles is cloned
+							into a fork with its schema only or with its data, last 30 days)</li
 						>
 						<li
 							>feature adoption (counts of which flow, script, trigger, worker and data table

frontend/src/lib/components/datatableAcl/PgAclEditor.svelte

@@ -120,7 +120,12 @@
 	<span class="text-xs text-secondary">Loading…</span>
 {:else}
 	<div class="flex flex-col gap-4">
-		{#if !info.editable}
+		{#if info.clone}
+			<span class="text-xs text-secondary">
+				Read only: this data table is a clone, and its owners and grants stay as they were copied
+				from the data table it was cloned from.
+			</span>
+		{:else if !info.editable}
 			<span class="text-xs text-secondary">
 				Read only: access is changed by the admins of the workspace that governs this data table, on
 				Windmill Enterprise Edition.

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

@@ -370,8 +370,7 @@
 			// Clone datatables BEFORE creating the workspace fork
 			if (forkDatatableSection) {
 				const queue = forkDatatableSection.buildCloneQueue(prefixed_id)
-				if (queue.length > 0) {
-					forkDatatableSection.startCloning(queue)
+				if (queue.length > 0 && forkDatatableSection.startCloning(queue)) {
 					return
 				}
 			}

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

@@ -193,6 +193,12 @@
 						data table backed by that database can use one. This one is backed by a PostgreSQL
 						resource — grant access on that server directly.
 					</Alert>
+				{:else if info?.clone_of}
+					<Alert type="info" title="Governed by {info.clone_of.workspace_id}" size="xs">
+						This data table is a clone of <span class="font-mono">{info.clone_of.datatable}</span>
+						in workspace <span class="font-mono">{info.clone_of.workspace_id}</span>, so it takes
+						its roles from there. You are evaluated as a member of that workspace.
+					</Alert>
 				{:else if governing}
 					<Alert type="info" title="Governed by {governing}" size="xs">
 						This data table points at the one in workspace <span class="font-mono">{governing}</span
@@ -227,7 +233,8 @@
 				/>
 
 				{#if permissioned}
-					{#if availableRoles.length === 0}
+					<!-- The catalog is only sent to someone who may pick from it. -->
+					{#if editable && availableRoles.length === 0}
 						<Alert type="warning" title="No role defined on this instance" size="xs">
 							Only <span class="font-mono">admin</span> can be used until a superadmin adds a data table
 							role in the data table settings page.

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

@@ -71,11 +71,7 @@
 
 				const steps: ForkStep[] = [
 					{
-						label: `CREATE DATABASE "${newDbName}"`,
-						status: 'pending'
-					},
-					{
-						label: `pg_dump → pg_import (${behavior === 'schema_only' ? 'schema only' : 'schema + data'})`,
+						label: `Clone into "${newDbName}" (${behavior === 'schema_only' ? 'schema only' : 'schema + data'})`,
 						status: 'pending'
 					}
 				]
@@ -96,62 +92,56 @@
 
 	let completedJobs: DatatableCloneJob[] = $state([])
 
-	export function startCloning(queue: DatatableCloneJob[]) {
-		completedJobs = []
-		cloneQueue = queue
+	/** Returns false when every job of `queue` was already cloned, and there is nothing to run. */
+	export function startCloning(queue: DatatableCloneJob[]): boolean {
+		// A copy made for this same fork is still waiting for it when the fork request failed and is
+		// retried; cloning it again would collide with the database already there.
+		const alreadyCloned = (job: DatatableCloneJob) =>
+			completedJobs.some(
+				(done) =>
+					done.name === job.name &&
+					done._newDbName === job._newDbName &&
+					done.behavior === job.behavior
+			)
+		completedJobs = completedJobs.filter((done) =>
+			queue.some((job) => job.name === done.name && job._newDbName === done._newDbName)
+		)
+		cloneQueue = queue.filter((job) => !alreadyCloned(job))
+		if (cloneQueue.length === 0) {
+			return false
+		}
 		currentCloneJob = cloneQueue[0]
 		cloneModalOpen = true
+		return true
 	}
 
 	export function getCompletedCloneJobs(): DatatableCloneJob[] {
 		return completedJobs
 	}
 
-	async function executeCloneJob(job: DatatableCloneJob) {
+	async function executeCloneJob(job: DatatableCloneJob): Promise<boolean> {
 		cloneRunning = true
-		let stepIdx = 0
-
-		// Step 1: Create the database
-		job.steps[stepIdx].status = 'running'
+		const step = job.steps[0]
+		step.status = 'running'
+		step.error = undefined
 		try {
-			await WorkspaceService.createPgDatabase({
+			await WorkspaceService.clonePgDatabase({
 				workspace: job._sourceWorkspace,
 				requestBody: {
 					source: `datatable://${job.name}`,
-					target_dbname: job._newDbName
-				}
-			})
-			job.steps[stepIdx].status = 'done'
-		} catch (e: any) {
-			job.steps[stepIdx].status = 'error'
-			job.steps[stepIdx].error = e?.body ?? e?.message ?? String(e)
-			cloneRunning = false
-			return
-		}
-		stepIdx++
-
-		// Step 2: Import data
-		job.steps[stepIdx].status = 'running'
-		try {
-			await WorkspaceService.importPgDatabase({
-				workspace: job._sourceWorkspace,
-				requestBody: {
-					source: `datatable://${job.name}`,
-					target: `datatable://${job.name}`,
-					target_dbname_override: job._newDbName,
+					target_dbname: job._newDbName,
 					fork_behavior: job.behavior
 				}
 			})
-			job.steps[stepIdx].status = 'done'
+			step.status = 'done'
+			return true
 		} catch (e: any) {
-			job.steps[stepIdx].status = 'error'
-			job.steps[stepIdx].error = e?.body ?? e?.message ?? String(e)
+			step.status = 'error'
+			step.error = e?.body ?? e?.message ?? String(e)
+			return false
+		} finally {
 			cloneRunning = false
-			return
 		}
-		stepIdx++
-
-		cloneRunning = false
 	}
 
 	function advanceCloneQueue() {
@@ -209,8 +199,11 @@
 		open={cloneModalOpen}
 		loading={cloneRunning}
 		onConfirmed={async () => {
-			await executeCloneJob(currentCloneJob!)
-			advanceCloneQueue()
+			// A failed clone leaves no database behind, so the error stays on screen and confirming
+			// again retries it.
+			if (await executeCloneJob(currentCloneJob!)) {
+				advanceCloneQueue()
+			}
 		}}
 		onCanceled={() => {
 			cloneModalOpen = false
@@ -233,9 +226,9 @@
 
 		{#if currentCloneJob.resourceType === 'instance'}
 			<p class="text-xs text-secondary mt-2">
-				This will run <code
-					>CREATE DATABASE {currentCloneJob.steps[0]?.label.match(/"([^"]+)"/)?.[1] ?? ''}</code
-				> on the Windmill PostgreSQL instance.
+				This will run <code>CREATE DATABASE {currentCloneJob._newDbName}</code> on the Windmill PostgreSQL
+				instance. A data table under roles keeps its owners and grants in the copy, and its roles stay
+				decided where they are decided today.
 			</p>
 		{:else}
 			<p class="text-xs text-secondary mt-2">