From 9375805fdf7501347a4dfb37e58f1be7fa2c5658 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 16 Oct 2026 22:21:38 +0200 Subject: [PATCH] feat: isolate fork object storage under a per-fork prefix (#11829) * feat: route the python SDK's S3 client settings through the S3 proxy Co-Authored-By: Claude Opus 5.5 * fix: drop the boto3 config key and expose the proxied bucket to polars Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 * fix: proxy S3 settings only for server-signed tokens and keep tls verify Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 * feat: isolate fork object storage under a per-fork prefix with read-through Co-Authored-By: Claude Opus 5.5 * fix: guard fork storage isolation against traversal, shared lakes and orphaning Co-Authored-By: Claude Opus 5.5 * fix: serialize fork purges with same-id creation and guard shared middle forks Co-Authored-By: Claude Opus 5.5 * fix: drop fork storage mappings on every replica and lock fork deletion against child creation Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref to 455681e55eb7cdfcfdb7cd79cbae75b333e07943 This commit updates the EE repository reference after PR #893 was merged in windmill-ee-private. Previous ee-repo-ref: 93f26e3cf816e4eb97b7d4dbbb5fc97fdf9821af New ee-repo-ref: 455681e55eb7cdfcfdb7cd79cbae75b333e07943 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: windmill-internal-app[bot] --- AGENTS.md | 3 + ...b2d57a07e2047a704d9a7d8eab3416ff1d9d5.json | 23 + ...a49857237cd57134dbc467b53601c1ec942d2.json | 22 + ...2224ab49659213638230be7f9acbf82d8e1b0.json | 28 ++ ...6309b24b658cac3c99d8d0bdb84cdd00c413a.json | 28 ++ ...72ae5b7858a087cb4eb5869573fff69c1c6b9.json | 35 ++ ...f6db33d63ff7bf55329452dc84dae2560c23e.json | 22 + ...415fe710c5fbbc7380c5088c7069fa2caf990.json | 15 + ...92249e32b1184a146463094289c604bbd0332.json | 16 + backend/ee-repo-ref.txt | 2 +- ...1016160346_fork_storage_namespace.down.sql | 1 + ...261016160346_fork_storage_namespace.up.sql | 24 ++ .../windmill-api-workspaces/src/workspaces.rs | 159 +++++++ .../src/workspaces_extra.rs | 156 +++++++ backend/windmill-api/openapi.yaml | 5 + backend/windmill-api/src/apps.rs | 59 ++- backend/windmill-api/src/args.rs | 4 +- .../windmill-api/src/triggers/http/handler.rs | 13 +- backend/windmill-common/src/fork_storage.rs | 394 ++++++++++++++++++ backend/windmill-common/src/lib.rs | 1 + .../windmill-object-store/src/fork_store.rs | 260 ++++++++++++ backend/windmill-object-store/src/lib.rs | 2 + backend/windmill-queue/src/tags.rs | 6 +- docs/fork-object-storage.md | 56 +++ .../CreateWorkspaceInner.svelte | 4 + .../ForkStorageSection.svelte | 71 ++++ 26 files changed, 1391 insertions(+), 18 deletions(-) create mode 100644 backend/.sqlx/query-45b73345f2009122142f3410736b2d57a07e2047a704d9a7d8eab3416ff1d9d5.json create mode 100644 backend/.sqlx/query-4640ac69b297127cd0f5c369fa4a49857237cd57134dbc467b53601c1ec942d2.json create mode 100644 backend/.sqlx/query-7796f40e349bf0771cceff037fa2224ab49659213638230be7f9acbf82d8e1b0.json create mode 100644 backend/.sqlx/query-98aebe8ead7d3f6e84ad88e6d006309b24b658cac3c99d8d0bdb84cdd00c413a.json create mode 100644 backend/.sqlx/query-a37dec4e86f537880d45da01ce872ae5b7858a087cb4eb5869573fff69c1c6b9.json create mode 100644 backend/.sqlx/query-bdc39bdd63c38217834a0c0d858f6db33d63ff7bf55329452dc84dae2560c23e.json create mode 100644 backend/.sqlx/query-c37f34ea3603af1496825ec54a3415fe710c5fbbc7380c5088c7069fa2caf990.json create mode 100644 backend/.sqlx/query-eb6dc43ac53a853eab20ef804a092249e32b1184a146463094289c604bbd0332.json create mode 100644 backend/migrations/20261016160346_fork_storage_namespace.down.sql create mode 100644 backend/migrations/20261016160346_fork_storage_namespace.up.sql create mode 100644 backend/windmill-common/src/fork_storage.rs create mode 100644 backend/windmill-object-store/src/fork_store.rs create mode 100644 docs/fork-object-storage.md create mode 100644 frontend/src/lib/components/workspaceSettings/ForkStorageSection.svelte diff --git a/AGENTS.md b/AGENTS.md index 8c30fed2f9..6516e0e095 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,6 +33,9 @@ Open-source platform for internal tools, workflows, API integrations, background administers behind `external_instance` data tables and Ducklake catalogs: its invariants (one lifecycle lock, managed-object markers, the setup gate, per-cluster roles, fork copy ownership) and how to run one locally +- **Fork object storage**: `docs/fork-object-storage.md` — how a fork's object keys map under + `__wm_forks//` with read-through to its ancestors, where the mapping must be applied, + and why `__wm_forks/` is reachable only through a workspace's own chain - **Data table viewer roles**: `docs/datatable-viewer-roles.md`, on what a connection naming no role holds: the default role as a floor, whose roles are added (the job's own, or its viewer's for an inline app runnable that opts in), why an admin's roles are not added, the per-role-set diff --git a/backend/.sqlx/query-45b73345f2009122142f3410736b2d57a07e2047a704d9a7d8eab3416ff1d9d5.json b/backend/.sqlx/query-45b73345f2009122142f3410736b2d57a07e2047a704d9a7d8eab3416ff1d9d5.json new file mode 100644 index 0000000000..2c22ae1d7e --- /dev/null +++ b/backend/.sqlx/query-45b73345f2009122142f3410736b2d57a07e2047a704d9a7d8eab3416ff1d9d5.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (SELECT 1 FROM fork_storage_namespace WHERE workspace_id = ANY($1))\n AND EXISTS (SELECT 1 FROM workspace WHERE parent_workspace_id = $2)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "45b73345f2009122142f3410736b2d57a07e2047a704d9a7d8eab3416ff1d9d5" +} diff --git a/backend/.sqlx/query-4640ac69b297127cd0f5c369fa4a49857237cd57134dbc467b53601c1ec942d2.json b/backend/.sqlx/query-4640ac69b297127cd0f5c369fa4a49857237cd57134dbc467b53601c1ec942d2.json new file mode 100644 index 0000000000..5ad778d3c0 --- /dev/null +++ b/backend/.sqlx/query-4640ac69b297127cd0f5c369fa4a49857237cd57134dbc467b53601c1ec942d2.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (SELECT 1 FROM fork_storage_namespace WHERE workspace_id = $1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "4640ac69b297127cd0f5c369fa4a49857237cd57134dbc467b53601c1ec942d2" +} diff --git a/backend/.sqlx/query-7796f40e349bf0771cceff037fa2224ab49659213638230be7f9acbf82d8e1b0.json b/backend/.sqlx/query-7796f40e349bf0771cceff037fa2224ab49659213638230be7f9acbf82d8e1b0.json new file mode 100644 index 0000000000..cf9dc458c2 --- /dev/null +++ b/backend/.sqlx/query-7796f40e349bf0771cceff037fa2224ab49659213638230be7f9acbf82d8e1b0.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT storage, storage_ref FROM fork_storage_namespace WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "storage", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "storage_ref", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "7796f40e349bf0771cceff037fa2224ab49659213638230be7f9acbf82d8e1b0" +} diff --git a/backend/.sqlx/query-98aebe8ead7d3f6e84ad88e6d006309b24b658cac3c99d8d0bdb84cdd00c413a.json b/backend/.sqlx/query-98aebe8ead7d3f6e84ad88e6d006309b24b658cac3c99d8d0bdb84cdd00c413a.json new file mode 100644 index 0000000000..f81b66ccb0 --- /dev/null +++ b/backend/.sqlx/query-98aebe8ead7d3f6e84ad88e6d006309b24b658cac3c99d8d0bdb84cdd00c413a.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT large_file_storage, ducklake FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "large_file_storage", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "ducklake", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "98aebe8ead7d3f6e84ad88e6d006309b24b658cac3c99d8d0bdb84cdd00c413a" +} diff --git a/backend/.sqlx/query-a37dec4e86f537880d45da01ce872ae5b7858a087cb4eb5869573fff69c1c6b9.json b/backend/.sqlx/query-a37dec4e86f537880d45da01ce872ae5b7858a087cb4eb5869573fff69c1c6b9.json new file mode 100644 index 0000000000..7733ad9fa5 --- /dev/null +++ b/backend/.sqlx/query-a37dec4e86f537880d45da01ce872ae5b7858a087cb4eb5869573fff69c1c6b9.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT c.id AS \"id!\", ws.large_file_storage,\n EXISTS (SELECT 1 FROM fork_storage_namespace f\n WHERE f.workspace_id = c.id AND f.storage = $2) AS \"isolated!\"\n FROM unnest($1::text[]) WITH ORDINALITY AS c(id, ord)\n LEFT JOIN workspace_settings ws ON ws.workspace_id = c.id\n ORDER BY c.ord", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "large_file_storage", + "type_info": "Jsonb" + }, + { + "ordinal": 2, + "name": "isolated!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Text" + ] + }, + "nullable": [ + null, + true, + null + ] + }, + "hash": "a37dec4e86f537880d45da01ce872ae5b7858a087cb4eb5869573fff69c1c6b9" +} diff --git a/backend/.sqlx/query-bdc39bdd63c38217834a0c0d858f6db33d63ff7bf55329452dc84dae2560c23e.json b/backend/.sqlx/query-bdc39bdd63c38217834a0c0d858f6db33d63ff7bf55329452dc84dae2560c23e.json new file mode 100644 index 0000000000..cb595ad5b3 --- /dev/null +++ b/backend/.sqlx/query-bdc39bdd63c38217834a0c0d858f6db33d63ff7bf55329452dc84dae2560c23e.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (SELECT 1 FROM workspace WHERE id = $1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "bdc39bdd63c38217834a0c0d858f6db33d63ff7bf55329452dc84dae2560c23e" +} diff --git a/backend/.sqlx/query-c37f34ea3603af1496825ec54a3415fe710c5fbbc7380c5088c7069fa2caf990.json b/backend/.sqlx/query-c37f34ea3603af1496825ec54a3415fe710c5fbbc7380c5088c7069fa2caf990.json new file mode 100644 index 0000000000..bb82fbf97e --- /dev/null +++ b/backend/.sqlx/query-c37f34ea3603af1496825ec54a3415fe710c5fbbc7380c5088c7069fa2caf990.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM fork_storage_namespace WHERE workspace_id = $1 AND storage = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "c37f34ea3603af1496825ec54a3415fe710c5fbbc7380c5088c7069fa2caf990" +} diff --git a/backend/.sqlx/query-eb6dc43ac53a853eab20ef804a092249e32b1184a146463094289c604bbd0332.json b/backend/.sqlx/query-eb6dc43ac53a853eab20ef804a092249e32b1184a146463094289c604bbd0332.json new file mode 100644 index 0000000000..b69c453bef --- /dev/null +++ b/backend/.sqlx/query-eb6dc43ac53a853eab20ef804a092249e32b1184a146463094289c604bbd0332.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO fork_storage_namespace (workspace_id, storage, storage_ref)\n VALUES ($1, $2, $3)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "eb6dc43ac53a853eab20ef804a092249e32b1184a146463094289c604bbd0332" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 6f07da341c..f6aca2078d 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -cd3395013d8484cd3917cdb6ee7837b6ac91e577 +455681e55eb7cdfcfdb7cd79cbae75b333e07943 diff --git a/backend/migrations/20261016160346_fork_storage_namespace.down.sql b/backend/migrations/20261016160346_fork_storage_namespace.down.sql new file mode 100644 index 0000000000..47c6b58649 --- /dev/null +++ b/backend/migrations/20261016160346_fork_storage_namespace.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS fork_storage_namespace; diff --git a/backend/migrations/20261016160346_fork_storage_namespace.up.sql b/backend/migrations/20261016160346_fork_storage_namespace.up.sql new file mode 100644 index 0000000000..247a9e2da9 --- /dev/null +++ b/backend/migrations/20261016160346_fork_storage_namespace.up.sql @@ -0,0 +1,24 @@ +-- Workspace storages a fork/dev workspace isolates from its parent: its writes land under +-- `__wm_forks//` of that storage, and its reads fall back to the parent's keys. +-- One row per (fork, storage) chosen isolated at fork creation; forks created before this +-- table have none and keep sharing the parent's keys. +-- +-- Deliberately NO foreign key to workspace(id), like fork_ducklake_namespace: rows are also the +-- cleanup ledger and must OUTLIVE the workspace row when deleting the prefix fails after the +-- delete commits. A recreated same-id fork would otherwise read the deleted fork's objects +-- under the deterministic prefix; fork creation retries leftover rows and refuses to proceed +-- while they cannot be cleaned. +CREATE TABLE fork_storage_namespace ( + workspace_id VARCHAR(50) NOT NULL, + -- '' = the default storage, else a `secondary_storage` name. + storage TEXT NOT NULL, + -- The storage's resolved identity at fork creation (`:`), so cleanup deletes the prefix from the storage the fork wrote to even if the + -- name is repointed later. + storage_ref TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workspace_id, storage) +); + +GRANT ALL ON fork_storage_namespace TO windmill_user; +GRANT ALL ON fork_storage_namespace TO windmill_admin; diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index d76f5c9cc2..17db877834 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -515,6 +515,11 @@ struct CreateWorkspaceFork { /// fork namespace + read-defer. #[serde(default)] shared_ducklakes: Vec, + /// Workspace storages the fork reads and writes at the parent's keys (`_default_` for the + /// primary one). Every other storage is isolated: the fork writes under its own prefix + /// and reads through to the parent's objects. + #[serde(default)] + shared_storages: Vec, /// Create the fork as a persistent dev workspace: the id is not required to carry the /// `wm-fork-` prefix, and at most one dev workspace may exist per parent. #[serde(default)] @@ -2131,6 +2136,38 @@ async fn edit_dbt_warehouses( Ok("Updated the workspace's dbt warehouses".to_string()) } +/// A fork's isolated storage is purged at deletion from the storage recorded at creation: under +/// another location, its objects would outlive the fork untracked, and a same-id fork would +/// read them. +async fn reject_isolated_storage_repoint( + tx: &mut Transaction<'_, Postgres>, + w_id: &str, + new_lfs: &serde_json::Value, +) -> Result<()> { + let rows = sqlx::query!( + "SELECT storage, storage_ref FROM fork_storage_namespace WHERE workspace_id = $1", + w_id + ) + .fetch_all(&mut **tx) + .await?; + for row in rows { + let new_ref = windmill_common::fork_storage::lfs_entry(new_lfs, &row.storage) + .and_then(windmill_common::workspaces::lfs_entry_storage_ref); + if new_ref.is_some_and(|r| r != row.storage_ref) { + let name = if row.storage.is_empty() { + windmill_types::s3::DEFAULT_STORAGE + } else { + row.storage.as_str() + }; + return Err(Error::BadRequest(format!( + "storage `{name}` is isolated from the parent in this fork, so where it points \ + cannot be changed" + ))); + } + } + Ok(()) +} + async fn edit_large_file_storage_config( authed: ApiAuthed, Extension(db): Extension, @@ -2198,6 +2235,7 @@ async fn edit_large_file_storage_config( let serialized_lfs_config = serde_json::to_value::(lfs_config) .map_err(|err| Error::internal_err(err.to_string()))?; + reject_isolated_storage_repoint(&mut tx, &w_id, &serialized_lfs_config).await?; // A workspace whose AI session backups fell back to the instance store leaves it // here: the generation moves on, so nothing it left in any instance store is read @@ -9571,6 +9609,25 @@ async fn create_workspace_fork( i.msg ); } + // Unlike a dropped lake's files, a deleted fork's objects are live to a same-id fork: it + // reads its own prefix first. + let leftover_storages = crate::workspaces_extra::prepare_fork_storage_purges( + &db, + &nw.id, + Some(&parent_workspace_id), + ) + .await?; + let storage_issues = + crate::workspaces_extra::purge_fork_storages_unless_live(&db, &nw.id, leftover_storages) + .await?; + if !storage_issues.is_empty() { + return Err(Error::BadRequest(format!( + "a previously deleted workspace with id '{}' left objects in its fork prefix that \ + could not be deleted: {}; retry once the storage is reachable again", + nw.id, + storage_issues.join("; ") + ))); + } #[cfg(not(feature = "enterprise"))] _check_nb_of_workspaces(&db).await?; @@ -9964,6 +10021,97 @@ fn copy_requests( .collect() } +/// Record every storage of the fork's cloned settings as isolated, except `shared` ones. +/// Isolation is a per-fork-creation choice, never inherited: a fork of a shared fork is +/// isolated unless it asks otherwise. +async fn isolate_fork_storages( + tx: &mut Transaction<'_, Postgres>, + fork_id: &str, + shared: &[String], + shared_ducklakes: &[String], +) -> Result<()> { + use windmill_common::fork_storage::{lfs_entry, registry_storage_name}; + + let settings = sqlx::query!( + "SELECT large_file_storage, ducklake FROM workspace_settings WHERE workspace_id = $1", + fork_id + ) + .fetch_optional(&mut **tx) + .await?; + let (lfs, ducklake) = settings + .map(|s| (s.large_file_storage, s.ducklake)) + .unwrap_or_default(); + let lfs = lfs.filter(|v| v.get("type").is_some()); + + // A shared lake keeps the parent's data path: its files must land at the parent's keys, or + // the parent's catalog would point at objects in the fork's prefix, purged with the fork. + if lfs.is_some() { + for lake in shared_ducklakes { + let lake_storage = ducklake + .as_ref() + .and_then(|d| { + d.get("ducklakes")? + .get(lake)? + .get("storage")? + .get("storage") + }) + .and_then(|s| s.as_str()); + let storage = registry_storage_name(lake_storage); + if !shared + .iter() + .any(|s| registry_storage_name(Some(s)) == storage) + { + return Err(Error::BadRequest(format!( + "ducklake `{lake}` is shared with the parent, so the storage holding its \ + data (`{}`) must be shared too", + lake_storage.unwrap_or(windmill_types::s3::DEFAULT_STORAGE) + ))); + } + } + } + let mut storages: Vec<&str> = vec![]; + if let Some(lfs) = &lfs { + storages.push(""); + if let Some(secondary) = lfs.get("secondary_storage").and_then(|s| s.as_object()) { + storages.extend(secondary.keys().map(String::as_str)); + } + } + for name in shared { + if !storages.contains(®istry_storage_name(Some(name))) { + return Err(Error::BadRequest(format!( + "cannot share storage `{name}` with the parent: no such storage in the workspace \ + settings" + ))); + } + } + let Some(lfs) = &lfs else { + return Ok(()); + }; + for storage in storages { + if shared + .iter() + .any(|s| registry_storage_name(Some(s)) == storage) + { + continue; + } + let Some(storage_ref) = + lfs_entry(lfs, storage).and_then(windmill_common::workspaces::lfs_entry_storage_ref) + else { + continue; + }; + sqlx::query!( + "INSERT INTO fork_storage_namespace (workspace_id, storage, storage_ref) + VALUES ($1, $2, $3)", + fork_id, + storage, + storage_ref + ) + .execute(&mut **tx) + .await?; + } + Ok(()) +} + /// Write the fork, with its data tables pointed at `copies`. async fn write_workspace_fork( db: DB, @@ -10229,6 +10377,14 @@ async fn write_workspace_fork( } } + isolate_fork_storages( + &mut tx, + &forked_id, + &nw.shared_storages, + &nw.shared_ducklakes, + ) + .await?; + // Lock the parent ("prod") so edits are funneled through this dev workspace. let locked_prod = nw.is_dev_workspace && (nw.lock_prod_deploy || nw.lock_prod_forking); if locked_prod { @@ -10276,6 +10432,9 @@ async fn write_workspace_fork( .await?; } tx.commit().await?; + // Same for a mapping cached before the isolation rows were visible: identity, writing the + // root's keys. + windmill_common::fork_storage::invalidate_fork_storage_keys_cache(&forked_id); // A pre-creation lookup could have cached an EMPTY ancestor chain for this id, which // would bypass ducklake fork isolation for the TTL. The same lookup could have cached the id diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index dc40c72d13..0cae90952b 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -54,6 +54,23 @@ pub(crate) async fn change_workspace_id( require_admin(authed.is_admin, &authed.username)?; } + // An isolating fork's objects live under a prefix derived from its id: under a new id it + // would see none of them, and descendants would lose them from their read-through. + let isolates_storage = sqlx::query_scalar!( + "SELECT EXISTS (SELECT 1 FROM fork_storage_namespace WHERE workspace_id = $1)", + &old_id + ) + .fetch_one(&db) + .await? + .unwrap_or(false); + if isolates_storage { + return Err(Error::BadRequest( + "This workspace isolates its object storage from its parent under a prefix derived \ + from its id, so its id cannot be changed" + .to_string(), + )); + } + let mut tx = db.begin().await?; // The settings copy below carries every data table entry to the new id, which fork cleanup of @@ -1109,6 +1126,34 @@ pub(crate) async fn delete_workspace( tracing::warn!("deleting workspace {w_id}: preparing ducklake cleanup: {e:#}"); vec![] }); + // A child fork reads through to this fork's prefix, or to the one this fork shares, and a + // child sharing the storage writes into it: deleting this link would destroy the child's + // data, or leave its chain without the isolating fork, writing the root's own keys. Under the + // lock forking this workspace takes, so no child can commit between the check and the delete. + windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?; + let chain = windmill_common::workspaces::workspace_with_fork_ancestors(&db, &w_id).await?; + let isolating_with_children = sqlx::query_scalar!( + "SELECT EXISTS (SELECT 1 FROM fork_storage_namespace WHERE workspace_id = ANY($1)) + AND EXISTS (SELECT 1 FROM workspace WHERE parent_workspace_id = $2)", + &chain, + &w_id + ) + .fetch_one(&mut *tx) + .await? + .unwrap_or(false); + if isolating_with_children { + return Err(Error::BadRequest(format!( + "the forks of {w_id} reach its fork chain's isolated object storage through it: \ + delete them first" + ))); + } + // Nothing can reach a deleted fork's prefix once it is gone, so it is always purged. + let fork_storage_purges = prepare_fork_storage_purges(&db, &w_id, None) + .await + .unwrap_or_else(|e| { + tracing::warn!("deleting workspace {w_id}: preparing storage purge: {e:#}"); + vec![] + }); sqlx::query!("DELETE FROM ai_agent_memory WHERE workspace_id = $1", &w_id) .execute(&mut *tx) @@ -1379,6 +1424,14 @@ pub(crate) async fn delete_workspace( e.msg ); } + match purge_fork_storages_unless_live(&db, &w_id, fork_storage_purges).await { + Ok(errors) => { + for e in errors { + tracing::warn!("deleted workspace {w_id}: storage purge: {e}"); + } + } + Err(e) => tracing::warn!("deleted workspace {w_id}: storage purge: {e:#}"), + } if let Some(parent) = dev_lock_parent { windmill_common::workspaces::invalidate_protection_rules_cache(&parent); @@ -2308,6 +2361,109 @@ async fn delete_fork_ducklake_data( )) } +/// A storage a fork isolated, resolved while its `$res:` still exists. +pub(crate) struct PreparedForkStoragePurge { + storage: String, + store: std::result::Result<(windmill_types::s3::LargeFileStorage, serde_json::Value), String>, +} + +/// Read-only: one entry per `fork_storage_namespace` row of `w_id`. `res_fallback_w_id` as in +/// [`prepare_fork_ducklake_cleanups`]. +pub(crate) async fn prepare_fork_storage_purges( + db: &DB, + w_id: &str, + res_fallback_w_id: Option<&str>, +) -> Result> { + let rows = sqlx::query!( + "SELECT storage, storage_ref FROM fork_storage_namespace WHERE workspace_id = $1", + w_id + ) + .fetch_all(db) + .await?; + let mut prepared = Vec::with_capacity(rows.len()); + for row in rows { + let storage = Some(row.storage.as_str()).filter(|s| !s.is_empty()); + let store = resolve_fork_storage(db, w_id, res_fallback_w_id, storage, &row.storage_ref) + .await + .map_err(|e| e.to_string()); + prepared.push(PreparedForkStoragePurge { storage: row.storage, store }); + } + Ok(prepared) +} + +/// [`purge_fork_storages`] under the lock same-id fork creation takes, unless a workspace with +/// that id exists again: the rows are then the new fork's, and its creation already purged the +/// ones it found. +pub(crate) async fn purge_fork_storages_unless_live( + db: &DB, + w_id: &str, + prepared: Vec, +) -> Result> { + let mut lock = db.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtext('fork:' || $1))") + .bind(w_id) + .execute(&mut *lock) + .await?; + let live = sqlx::query_scalar!( + "SELECT EXISTS (SELECT 1 FROM workspace WHERE id = $1)", + w_id + ) + .fetch_one(&mut *lock) + .await? + .unwrap_or(false); + let errors = if live { + vec![] + } else { + purge_fork_storages(db, w_id, prepared).await + }; + lock.commit().await?; + Ok(errors) +} + +/// Delete everything under the fork's `__wm_forks//` in each prepared storage, then its +/// registry row. Returns one message per storage that could not be purged; its row survives. +async fn purge_fork_storages( + db: &DB, + w_id: &str, + prepared: Vec, +) -> Vec { + let area = format!( + "{}/{}", + windmill_common::workspaces::FORK_DUCKLAKE_DATA_DIR, + windmill_common::workspaces::fork_data_dir_segment(w_id) + ); + let mut errors = vec![]; + for p in prepared { + let name = if p.storage.is_empty() { + "default storage" + } else { + p.storage.as_str() + }; + let purged = match p.store { + Ok((lfs, value)) => delete_fork_ducklake_data(lfs, value, &area) + .await + .map_err(|e| e.to_string()), + Err(e) => Err(e), + }; + if let Err(e) = purged { + errors.push(format!("{name}: {e}")); + continue; + } + if let Err(e) = sqlx::query!( + "DELETE FROM fork_storage_namespace WHERE workspace_id = $1 AND storage = $2", + w_id, + p.storage + ) + .execute(db) + .await + { + errors.push(format!("{name}: removing registry row: {e}")); + } + } + windmill_common::fork_storage::invalidate_fork_storage_keys_cache(w_id); + errors +} + /// Destroying an ATTACHED dev workspace (or its data environments) must be a prod-admin /// action, not just the dev's own owner (dev ownership can diverge from prod's) — mirrors /// detach_dev_workspace. Shared by `delete_workspace` and `drop_forked_ducklake_namespaces` diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 81a4e3274a..806d91e8b8 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -36678,6 +36678,11 @@ components: items: type: string description: "Lake names the fork SHARES with the parent (reads and writes the parent's lake directly). Every lake not listed gets the default isolated fork namespace with read-defer to the parent." + shared_storages: + type: array + items: + type: string + description: "Workspace storages the fork SHARES with the parent (`_default_` for the primary one): it reads and writes the parent's objects directly. Every storage not listed is isolated: the fork writes under its own prefix and reads through to the parent's objects." is_dev_workspace: type: boolean description: "Create the fork as a persistent dev workspace (id not required to carry the wm-fork- prefix; at most one per parent)" diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 8343b4595d..944535dd84 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -5317,10 +5317,14 @@ async fn upload_s3_file_from_app( let user_db = UserDB::new(db.clone()); - let (s3_resource_opt, file_key, on_behalf_of_email, permissioned_as, username) = if policy - .as_ref() - .is_some_and(|p| p.s3_inputs.is_some()) - { + let ( + s3_resource_opt, + file_key, + on_behalf_of_email, + permissioned_as, + username, + is_workspace_storage, + ) = if policy.as_ref().is_some_and(|p| p.s3_inputs.is_some()) { let policy = policy.unwrap(); let s3_inputs = policy.s3_inputs.as_ref().unwrap(); @@ -5389,6 +5393,7 @@ async fn upload_s3_file_from_app( email, permissioned_as, username, + false, ) } else { return Err(Error::BadRequest( @@ -5417,6 +5422,7 @@ async fn upload_s3_file_from_app( email, permissioned_as, username, + false, ) } } else { @@ -5434,7 +5440,14 @@ async fn upload_s3_file_from_app( None, ) .await?; - (s3_resource_opt, file_key, email, permissioned_as, username) + ( + s3_resource_opt, + file_key, + email, + permissioned_as, + username, + true, + ) } } else { return Err(Error::BadRequest( @@ -5468,7 +5481,14 @@ async fn upload_s3_file_from_app( ) .await?; - (s3_resource_opt, file_key, email, permissioned_as, username) + ( + s3_resource_opt, + file_key, + email, + permissioned_as, + username, + true, + ) } } else { // backward compatibility (no policy) @@ -5500,6 +5520,7 @@ async fn upload_s3_file_from_app( on_behalf_of_email, permissioned_as, username, + false, ) } else { let db_with_opt_authed = DbWithOptAuthed::from_authed(&authed, db.clone(), None); @@ -5519,6 +5540,7 @@ async fn upload_s3_file_from_app( on_behalf_of_email, permissioned_as, username, + true, ) } } else { @@ -5529,7 +5551,19 @@ async fn upload_s3_file_from_app( let s3_resource = s3_resource_opt.ok_or(Error::internal_err( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = windmill_object_store::build_object_store_client(&s3_resource).await?; + // The quota sizes an overwrite by the fork's own copy: a fork's write never replaces the + // ancestor's object its reads fall through to. + type Client = std::sync::Arc; + let (s3_client, _quota_client): (Client, Client) = if is_workspace_storage { + let store = + windmill_object_store::fork_store::workspace_fork_store(&db, &w_id, None, &s3_resource) + .await?; + let own_only = store.own_only(); + (store, own_only) + } else { + let client = windmill_object_store::build_object_store_client(&s3_resource).await?; + (client.clone(), client) + }; let options = Attributes::from_iter(vec![ ( @@ -5562,7 +5596,8 @@ async fn upload_s3_file_from_app( .get(http::header::CONTENT_LENGTH) .and_then(|h| h.to_str().ok()) .and_then(|s| s.parse::().ok()); - let budget = ce_upload_budget(&db, &w_id, &s3_client, &file_key, content_length).await?; + let budget = + ce_upload_budget(&db, &w_id, &_quota_client, &file_key, content_length).await?; (Some(budget.max_size), budget.existing_size) } else { (None, 0) @@ -5644,6 +5679,7 @@ async fn delete_s3_file_from_app( ) .await?; + let is_workspace_storage = s3_resource_path.is_none(); let s3_resource = if let Some(s3_resource_path) = s3_resource_path { let db_with_opt_authed = DbWithOptAuthed::from_authed(&on_behalf_authed, db.clone(), Some(user_db.clone())); @@ -5672,7 +5708,12 @@ async fn delete_s3_file_from_app( ))? }; - let s3_client = windmill_object_store::build_object_store_client(&s3_resource).await?; + let s3_client = if is_workspace_storage { + windmill_object_store::fork_store::workspace_object_store(&db, &w_id, None, &s3_resource) + .await? + } else { + windmill_object_store::build_object_store_client(&s3_resource).await? + }; s3_client.delete(&path).await.map_err(|err| { tracing::error!("Error deleting file: {:?}", err); diff --git a/backend/windmill-api/src/args.rs b/backend/windmill-api/src/args.rs index 7e676989b2..6c1c9824bd 100644 --- a/backend/windmill-api/src/args.rs +++ b/backend/windmill-api/src/args.rs @@ -139,13 +139,13 @@ impl RawWebhookArgs { get_random_file_name, get_workspace_s3_resource, upload_file_internal, }; use futures::TryStreamExt; - use windmill_object_store::build_object_store_client; + use windmill_object_store::fork_store::workspace_object_store; use windmill_object_store::object_store_reexports::{Attribute, Attributes}; let (_, s3_resource) = get_workspace_s3_resource(authed, db, None, w_id, None).await?; if let Some(s3_resource) = s3_resource { - let s3_client = build_object_store_client(&s3_resource).await?; + let s3_client = workspace_object_store(db, w_id, None, &s3_resource).await?; let mut parts: Vec<(String, MultipartValue)> = Vec::new(); diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 3d85904e50..5db6bf303d 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -34,10 +34,7 @@ use windmill_store::resources::try_get_resource_from_db_as; use windmill_trigger::TriggerMode; #[cfg(feature = "parquet")] -use { - crate::job_helpers_oss::get_workspace_s3_resource, - windmill_object_store::build_object_store_client, -}; +use crate::job_helpers_oss::get_workspace_s3_resource; /// Which router a request's CORS decision must be looked up in. /// @@ -624,7 +621,13 @@ async fn route_job( let s3_resource = s3_resource_opt.ok_or_else(|| { windmill_object_store::workspace_storage_misconfigured(config.storage.as_deref()) })?; - let s3_client = build_object_store_client(&s3_resource).await?; + let s3_client = windmill_object_store::fork_store::workspace_object_store( + &db, + &trigger.workspace_id, + config.storage.as_deref(), + &s3_resource, + ) + .await?; let path = if trigger.is_static_website { let subpath = params diff --git a/backend/windmill-common/src/fork_storage.rs b/backend/windmill-common/src/fork_storage.rs new file mode 100644 index 0000000000..8d2797ab3c --- /dev/null +++ b/backend/windmill-common/src/fork_storage.rs @@ -0,0 +1,394 @@ +//! Where a workspace's object keys live in a storage it may share with its fork ancestors. +//! +//! A fork that isolates a storage (a `fork_storage_namespace` row) writes logical key `k` at +//! `__wm_forks//k` and reads `k` from the first of its own and its ancestors' +//! locations that has it. Everything under `__wm_forks/` is reachable only through the areas of +//! the workspace's own chain, which is also what DuckLake's fork namespaces live in: those keys +//! arrive already prefixed and pass through unchanged. + +use std::sync::Arc; + +use quick_cache::sync::Cache; + +use crate::error::{Error, Result}; +use crate::workspaces::{ + fork_data_dir_segment, lfs_entry_storage_ref, workspace_with_fork_ancestors, + FORK_DUCKLAKE_DATA_DIR, +}; +use crate::DB; + +/// How one workspace's logical keys map to physical keys in one of its storages. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ForkStorageKeys { + /// Prefix of every write: `__wm_forks//` in an isolating fork, `""` otherwise. + write_prefix: String, + /// Prefixes a read tries in order, nearest first. + read_prefixes: Vec, + /// `__wm_forks//` of this workspace, writable as is. + own_area: String, + /// The fork areas of this workspace and its ancestors, readable as is. + readable_areas: Vec, +} + +/// A listing request translated to the physical storage. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ListMapping { + pub physical_prefix: String, + /// Stripped from each physical key to get the logical key returned to the caller. + strip: String, + /// Logical keys under `__wm_forks/` are dropped: reached only by listing that area itself. + hide_fork_area: bool, +} + +impl ListMapping { + /// Whether listed keys differ from what the caller may see: false lets a listing response + /// be passed through untouched. + pub fn rewrites(&self) -> bool { + let fork_dir = format!("{FORK_DUCKLAKE_DATA_DIR}/"); + !self.strip.is_empty() + || (self.hide_fork_area && fork_dir.starts_with(&self.physical_prefix)) + } + + /// The physical form of a logical key of this listing, e.g. a start-after cursor. + pub fn physical_key(&self, logical: &str) -> String { + format!("{}{logical}", self.strip) + } + + /// The logical key for a listed physical key, or None when the caller must not see it. + pub fn logical_key(&self, physical: &str) -> Option { + let logical = physical.strip_prefix(self.strip.as_str())?; + if self.hide_fork_area && in_fork_area(logical) { + return None; + } + Some(logical.to_string()) + } +} + +fn in_fork_area(key: &str) -> bool { + key == FORK_DUCKLAKE_DATA_DIR + || key + .strip_prefix(FORK_DUCKLAKE_DATA_DIR) + .is_some_and(|rest| rest.starts_with('/')) +} + +fn fork_area(w_id: &str) -> String { + format!("{FORK_DUCKLAKE_DATA_DIR}/{}/", fork_data_dir_segment(w_id)) +} + +fn forbidden(key: &str) -> Error { + Error::NotAuthorized(format!( + "`{key}` is in another workspace's fork namespace (`{FORK_DUCKLAKE_DATA_DIR}/`)" + )) +} + +/// The S3 proxy forwards mapped keys in a URL, where clients and signers collapse `.` and `..` +/// segments: `../../k` would reach the parent's `k`. +fn reject_dot_segments(key: &str) -> Result<&str> { + let key = key.trim_start_matches('/'); + if key.split('/').any(|s| s == "." || s == "..") { + return Err(Error::BadRequest(format!( + "`{key}` has a `.` or `..` path segment, which object storage keys may not use here" + ))); + } + Ok(key) +} + +/// One workspace of a fork chain, nearest first, as the storage mapping needs it. +#[derive(Debug, Clone)] +pub struct ChainLink { + pub w_id: String, + /// Identity of the storage in this workspace's settings (`lfs_entry_storage_ref`). + pub storage_ref: Option, + pub isolated: bool, +} + +impl ForkStorageKeys { + /// The mapping of `chain[0]`. Reads stop at the first ancestor whose storage is a different + /// one: its keys are not in this bucket. + pub fn from_chain(chain: &[ChainLink]) -> Self { + let own = &chain[0]; + let mut read_prefixes = vec![]; + let mut reached_root = true; + for link in chain { + if link.storage_ref != own.storage_ref { + reached_root = false; + break; + } + if link.isolated { + read_prefixes.push(fork_area(&link.w_id)); + } + } + if reached_root || read_prefixes.is_empty() { + read_prefixes.push(String::new()); + } + Self { + write_prefix: read_prefixes[0].clone(), + read_prefixes, + own_area: fork_area(&own.w_id), + readable_areas: chain.iter().map(|l| fork_area(&l.w_id)).collect(), + } + } + + /// Whether logical keys map to themselves (only the fork-area boundary applies). + pub fn is_identity(&self) -> bool { + self.write_prefix.is_empty() && self.read_prefixes.len() == 1 + } + + /// The same mapping with reads confined to the workspace's own copy, for sizing what a + /// write replaces: an ancestor's object is not overwritten by it. + pub fn own_only(&self) -> Self { + Self { read_prefixes: vec![self.write_prefix.clone()], ..self.clone() } + } + + /// The physical key a write (put, multipart, delete, copy destination) of `key` goes to. + pub fn write_key(&self, key: &str) -> Result { + let key = reject_dot_segments(key)?; + if in_fork_area(key) { + return if key.starts_with(&self.own_area) { + Ok(key.to_string()) + } else { + Err(forbidden(key)) + }; + } + Ok(format!("{}{key}", self.write_prefix)) + } + + /// The physical keys a read of `key` tries, in order, until one exists. + pub fn read_keys(&self, key: &str) -> Result> { + let key = reject_dot_segments(key)?; + if in_fork_area(key) { + return if self.readable_areas.iter().any(|a| key.starts_with(a)) { + Ok(vec![key.to_string()]) + } else { + Err(forbidden(key)) + }; + } + Ok(self + .read_prefixes + .iter() + .map(|p| format!("{p}{key}")) + .collect()) + } + + /// A listing of `prefix` covers this workspace's own objects only: an isolating fork does + /// not list its ancestors' keys. + pub fn list(&self, prefix: &str) -> Result { + let prefix = reject_dot_segments(prefix)?; + if in_fork_area(prefix) { + if !self.readable_areas.iter().any(|a| prefix.starts_with(a)) { + return Err(forbidden(prefix)); + } + return Ok(ListMapping { + physical_prefix: prefix.to_string(), + strip: String::new(), + hide_fork_area: false, + }); + } + Ok(ListMapping { + physical_prefix: format!("{}{prefix}", self.write_prefix), + strip: self.write_prefix.clone(), + hide_fork_area: true, + }) + } +} + +lazy_static::lazy_static! { + static ref FORK_STORAGE_KEYS_CACHE: Cache<(String, String), (Arc, i64)> = + Cache::new(1000); +} + +/// Drop the cached mappings of a workspace whose isolation rows or lineage just changed. +pub fn invalidate_fork_storage_keys_cache(w_id: &str) { + FORK_STORAGE_KEYS_CACHE.retain(|(w, _), _| w != w_id); +} + +pub fn clear_fork_storage_keys_cache() { + FORK_STORAGE_KEYS_CACHE.clear(); +} + +/// `storage` as the registry keys it: '' for the default storage. +pub fn registry_storage_name(storage: Option<&str>) -> &str { + match storage { + None | Some("") | Some(windmill_types::s3::DEFAULT_STORAGE) => "", + Some(s) => s, + } +} + +/// The storage entry named `storage` (registry form) in a `large_file_storage` value. +pub fn lfs_entry<'a>(lfs: &'a serde_json::Value, storage: &str) -> Option<&'a serde_json::Value> { + if storage.is_empty() { + Some(lfs) + } else { + lfs.get("secondary_storage")?.get(storage) + } +} + +/// How `w_id`'s keys map in `storage`. Reads lineage for any `w_id` with no authorization +/// check, like [`workspace_with_fork_ancestors`]: callers must be authorized for `w_id`. +pub async fn fork_storage_keys( + db: &DB, + w_id: &str, + storage: Option<&str>, +) -> Result> { + let storage = registry_storage_name(storage).to_string(); + let cache_key = (w_id.to_string(), storage.clone()); + let now = chrono::Utc::now().timestamp(); + if let Some((keys, expiry)) = FORK_STORAGE_KEYS_CACHE.get(&cache_key) { + if expiry > now { + return Ok(keys); + } + } + + let chain_ids = workspace_with_fork_ancestors(db, w_id).await?; + let rows = sqlx::query!( + r#"SELECT c.id AS "id!", ws.large_file_storage, + EXISTS (SELECT 1 FROM fork_storage_namespace f + WHERE f.workspace_id = c.id AND f.storage = $2) AS "isolated!" + FROM unnest($1::text[]) WITH ORDINALITY AS c(id, ord) + LEFT JOIN workspace_settings ws ON ws.workspace_id = c.id + ORDER BY c.ord"#, + &chain_ids, + &storage, + ) + .fetch_all(db) + .await?; + let chain: Vec = rows + .into_iter() + .map(|r| ChainLink { + storage_ref: r + .large_file_storage + .as_ref() + .and_then(|lfs| lfs_entry(lfs, &storage)) + .and_then(lfs_entry_storage_ref), + w_id: r.id, + isolated: r.isolated, + }) + .collect(); + let keys = Arc::new(if chain.is_empty() { + ForkStorageKeys::from_chain(&[ChainLink { + w_id: w_id.to_string(), + storage_ref: None, + isolated: false, + }]) + } else { + ForkStorageKeys::from_chain(&chain) + }); + FORK_STORAGE_KEYS_CACHE.insert(cache_key, (keys.clone(), now + 30)); + Ok(keys) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn link(w_id: &str, storage_ref: &str, isolated: bool) -> ChainLink { + ChainLink { w_id: w_id.into(), storage_ref: Some(storage_ref.into()), isolated } + } + + #[test] + fn root_is_identity_but_keeps_fork_areas_out_of_reach() { + let keys = ForkStorageKeys::from_chain(&[link("prod", "S3Storage:u/a/s3", false)]); + assert!(keys.is_identity()); + assert_eq!(keys.write_key("a/b.csv").unwrap(), "a/b.csv"); + assert_eq!(keys.read_keys("a/b.csv").unwrap(), vec!["a/b.csv"]); + let other = fork_area("wm-fork-x"); + assert!(keys.write_key(&format!("{other}a")).is_err()); + assert!(keys.read_keys(&format!("{other}a")).is_err()); + let listing = keys.list("").unwrap(); + assert_eq!(listing.logical_key("a/b.csv").as_deref(), Some("a/b.csv")); + assert_eq!(listing.logical_key(&format!("{other}a")), None); + } + + #[test] + fn isolated_fork_writes_its_area_and_reads_through_to_ancestors() { + let s = "S3Storage:u/a/s3"; + let chain = [ + link("wm-fork-b", s, true), + link("wm-fork-a", s, false), + link("dev", s, true), + link("prod", s, false), + ]; + let keys = ForkStorageKeys::from_chain(&chain); + let (b, dev) = (fork_area("wm-fork-b"), fork_area("dev")); + assert_eq!(keys.write_key("x.csv").unwrap(), format!("{b}x.csv")); + assert_eq!( + keys.read_keys("x.csv").unwrap(), + vec![ + format!("{b}x.csv"), + format!("{dev}x.csv"), + "x.csv".to_string() + ] + ); + // DuckLake's own fork namespace and its ancestors' pass through as is. + assert_eq!( + keys.write_key(&format!("{b}lake/f")).unwrap(), + format!("{b}lake/f") + ); + let a = fork_area("wm-fork-a"); + assert_eq!( + keys.read_keys(&format!("{a}lake/f")).unwrap(), + vec![format!("{a}lake/f")] + ); + assert!(keys.write_key(&format!("{a}lake/f")).is_err()); + let listing = keys.list("dir/").unwrap(); + assert_eq!(listing.physical_prefix, format!("{b}dir/")); + assert_eq!( + listing.logical_key(&format!("{b}dir/f")).as_deref(), + Some("dir/f") + ); + } + + #[test] + fn shared_fork_writes_where_its_parent_does() { + let s = "S3Storage:u/a/s3"; + let keys = ForkStorageKeys::from_chain(&[ + link("wm-fork-c", s, false), + link("dev", s, true), + link("prod", s, false), + ]); + let dev = fork_area("dev"); + assert_eq!(keys.write_key("x").unwrap(), format!("{dev}x")); + assert_eq!( + keys.read_keys("x").unwrap(), + vec![format!("{dev}x"), "x".into()] + ); + } + + #[test] + fn reads_stop_at_an_ancestor_on_another_storage() { + let keys = ForkStorageKeys::from_chain(&[ + link("wm-fork-d", "S3Storage:u/a/other", true), + link("prod", "S3Storage:u/a/s3", false), + ]); + let d = fork_area("wm-fork-d"); + assert_eq!(keys.read_keys("x").unwrap(), vec![format!("{d}x")]); + } + + #[test] + fn dot_segments_cannot_climb_out_of_the_fork_area() { + let keys = ForkStorageKeys::from_chain(&[ + link("wm-fork-a", "S3Storage:u/a/s3", true), + link("prod", "S3Storage:u/a/s3", false), + ]); + let area = fork_area("wm-fork-a"); + assert!(keys.write_key("../../prod.csv").is_err()); + assert!(keys.write_key(&format!("{area}lake/../../x")).is_err()); + assert!(keys.read_keys("a/./b").is_err()); + assert!(keys.list("..").is_err()); + assert_eq!( + keys.write_key("a..b/c.d").unwrap(), + format!("{area}a..b/c.d") + ); + } + + #[test] + fn fork_area_match_is_on_whole_segments() { + assert!(in_fork_area("__wm_forks/a")); + assert!(!in_fork_area("__wm_forks_backup/a")); + let keys = ForkStorageKeys::from_chain(&[link("prod", "S3Storage:u/a/s3", false)]); + assert_eq!( + keys.write_key("__wm_forks_backup/a").unwrap(), + "__wm_forks_backup/a" + ); + } +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 3b0a8e207f..0b5929025e 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -73,6 +73,7 @@ pub use feature_usage_oss as feature_usage; pub mod flow_conversations; pub mod flow_status; pub mod flows; +pub mod fork_storage; pub mod folders; pub mod global_settings; pub mod hub_resource_types; diff --git a/backend/windmill-object-store/src/fork_store.rs b/backend/windmill-object-store/src/fork_store.rs new file mode 100644 index 0000000000..a7d283a09e --- /dev/null +++ b/backend/windmill-object-store/src/fork_store.rs @@ -0,0 +1,260 @@ +//! An [`ObjectStore`] over one workspace storage that applies the workspace's +//! [`ForkStorageKeys`]: callers keep using logical keys. + +use std::sync::Arc; + +use futures::{stream::BoxStream, StreamExt, TryStreamExt}; +use object_store::path::Path; +use object_store::{ + GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutMultipartOpts, + PutOptions, PutPayload, PutResult, Result, +}; +use windmill_common::fork_storage::{fork_storage_keys, ForkStorageKeys}; + +use crate::{build_object_store_client, ObjectStoreResource}; + +/// The client for `w_id`'s storage `storage` (None = default). Every server-side access to a +/// workspace storage by logical key goes through this, so fork isolation holds on each path. +/// +/// It checks no authorization: the caller must already have checked that the requester may +/// use `w_id`'s storage, and its permission rules for each key it reads or writes. +pub async fn workspace_object_store( + db: &windmill_common::DB, + w_id: &str, + storage: Option<&str>, + resource: &ObjectStoreResource, +) -> windmill_common::error::Result> { + Ok(workspace_fork_store(db, w_id, storage, resource).await?) +} + +/// [`workspace_object_store`] as its concrete type, under the same authorization contract. +pub async fn workspace_fork_store( + db: &windmill_common::DB, + w_id: &str, + storage: Option<&str>, + resource: &ObjectStoreResource, +) -> windmill_common::error::Result> { + let inner = build_object_store_client(resource).await?; + let keys = fork_storage_keys(db, w_id, storage).await?; + Ok(Arc::new(ForkStore { inner, keys })) +} + +#[derive(Debug)] +pub struct ForkStore { + inner: Arc, + keys: Arc, +} + +impl ForkStore { + pub fn new(inner: Arc, keys: Arc) -> Self { + Self { inner, keys } + } + + pub fn keys(&self) -> &ForkStorageKeys { + &self.keys + } + + /// This store with reads confined to the workspace's own copies. + pub fn own_only(&self) -> Arc { + Arc::new(ForkStore { inner: self.inner.clone(), keys: Arc::new(self.keys.own_only()) }) + } +} + +impl std::fmt::Display for ForkStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "ForkStore({})", self.inner) + } +} + +fn denied(location: &Path, e: windmill_common::error::Error) -> object_store::Error { + object_store::Error::PermissionDenied { path: location.to_string(), source: Box::new(e) } +} + +fn parse(key: String) -> Result { + Path::parse(key).map_err(|source| object_store::Error::InvalidPath { source }) +} + +impl ForkStore { + fn write_path(&self, location: &Path) -> Result { + parse( + self.keys + .write_key(location.as_ref()) + .map_err(|e| denied(location, e))?, + ) + } + + fn read_paths(&self, location: &Path) -> Result> { + self.keys + .read_keys(location.as_ref()) + .map_err(|e| denied(location, e))? + .into_iter() + .map(parse) + .collect() + } + + /// The physical path of the first of `location`'s read paths that exists (the last one + /// when none does). + pub async fn resolve_read(&self, location: &Path) -> Result { + parse(self.resolve_read_key(location.as_ref()).await?) + } + + /// [`Self::resolve_read`] on a raw key, for callers that forward the key themselves. + pub async fn resolve_read_key(&self, key: &str) -> Result { + let location = Path::from(key); + let mut keys = self.keys.read_keys(key).map_err(|e| denied(&location, e))?; + let last = keys.pop().expect("read_keys returns at least one key"); + for candidate in keys { + match self.inner.head(&parse(candidate.clone())?).await { + Ok(_) => return Ok(candidate), + Err(object_store::Error::NotFound { .. }) => continue, + Err(e) => return Err(e), + } + } + Ok(last) + } +} + +#[async_trait::async_trait] +impl ObjectStore for ForkStore { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> Result { + self.inner + .put_opts(&self.write_path(location)?, payload, opts) + .await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOpts, + ) -> Result> { + self.inner + .put_multipart_opts(&self.write_path(location)?, opts) + .await + } + + async fn get_opts(&self, location: &Path, options: GetOptions) -> Result { + let paths = self.read_paths(location)?; + let last = paths.len() - 1; + for (i, path) in paths.into_iter().enumerate() { + match self.inner.get_opts(&path, options.clone()).await { + Err(object_store::Error::NotFound { .. }) if i < last => continue, + Ok(mut result) => { + result.meta.location = location.clone(); + return Ok(result); + } + Err(e) => return Err(e), + } + } + unreachable!("read_keys returns at least one key") + } + + async fn head(&self, location: &Path) -> Result { + let paths = self.read_paths(location)?; + let last = paths.len() - 1; + for (i, path) in paths.into_iter().enumerate() { + match self.inner.head(&path).await { + Err(object_store::Error::NotFound { .. }) if i < last => continue, + Ok(meta) => return Ok(ObjectMeta { location: location.clone(), ..meta }), + Err(e) => return Err(e), + } + } + unreachable!("read_keys returns at least one key") + } + + /// Removes this workspace's copy only: an ancestor's object stays readable through it. + async fn delete(&self, location: &Path) -> Result<()> { + self.inner.delete(&self.write_path(location)?).await + } + + fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result> { + let prefix = prefix.map(|p| p.as_ref().to_string()).unwrap_or_default(); + let mapping = match self.keys.list(&prefix) { + Ok(m) => m, + Err(e) => { + let err = denied(&Path::from(prefix), e); + return futures::stream::once(async move { Err(err) }).boxed(); + } + }; + let physical = Path::from(mapping.physical_prefix.clone()); + self.inner + .list(Some(&physical)) + .try_filter_map(move |meta| { + let logical = mapping.logical_key(meta.location.as_ref()); + futures::future::ready(Ok(logical + .and_then(|l| Path::parse(l).ok()) + .map(|location| ObjectMeta { location, ..meta }))) + }) + .boxed() + } + + fn list_with_offset( + &self, + prefix: Option<&Path>, + offset: &Path, + ) -> BoxStream<'static, Result> { + let prefix = prefix.map(|p| p.as_ref().to_string()).unwrap_or_default(); + let mapping = match self.keys.list(&prefix) { + Ok(m) => m, + Err(e) => { + let err = denied(&Path::from(prefix), e); + return futures::stream::once(async move { Err(err) }).boxed(); + } + }; + let physical = Path::from(mapping.physical_prefix.clone()); + let physical_offset = match parse(mapping.physical_key(offset.as_ref())) { + Ok(p) => p, + Err(e) => return futures::stream::once(async move { Err(e) }).boxed(), + }; + self.inner + .list_with_offset(Some(&physical), &physical_offset) + .try_filter_map(move |meta| { + let logical = mapping.logical_key(meta.location.as_ref()); + futures::future::ready(Ok(logical + .and_then(|l| Path::parse(l).ok()) + .map(|location| ObjectMeta { location, ..meta }))) + }) + .boxed() + } + + async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result { + let prefix_str = prefix.map(|p| p.as_ref().to_string()).unwrap_or_default(); + let mapping = self + .keys + .list(&prefix_str) + .map_err(|e| denied(&Path::from(prefix_str.clone()), e))?; + let physical = Path::from(mapping.physical_prefix.clone()); + let listed = self.inner.list_with_delimiter(Some(&physical)).await?; + let logical = |p: &Path| { + mapping + .logical_key(p.as_ref()) + .and_then(|l| Path::parse(l).ok()) + }; + Ok(ListResult { + common_prefixes: listed.common_prefixes.iter().filter_map(logical).collect(), + objects: listed + .objects + .into_iter() + .filter_map(|meta| { + logical(&meta.location).map(|location| ObjectMeta { location, ..meta }) + }) + .collect(), + }) + } + + async fn copy(&self, from: &Path, to: &Path) -> Result<()> { + let from = self.resolve_read(from).await?; + self.inner.copy(&from, &self.write_path(to)?).await + } + + async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> { + let from = self.resolve_read(from).await?; + self.inner + .copy_if_not_exists(&from, &self.write_path(to)?) + .await + } +} diff --git a/backend/windmill-object-store/src/lib.rs b/backend/windmill-object-store/src/lib.rs index a9551ad835..4e40d6e313 100644 --- a/backend/windmill-object-store/src/lib.rs +++ b/backend/windmill-object-store/src/lib.rs @@ -59,6 +59,8 @@ use windmill_common::utils::rd_string; pub mod job_s3_helpers_ee; #[cfg(feature = "parquet")] pub mod job_s3_helpers_oss; +#[cfg(feature = "parquet")] +pub mod fork_store; // Re-export object_store types so consumers don't need a direct object_store dep #[cfg(feature = "parquet")] diff --git a/backend/windmill-queue/src/tags.rs b/backend/windmill-queue/src/tags.rs index ba045ad451..b31e3d99c3 100644 --- a/backend/windmill-queue/src/tags.rs +++ b/backend/windmill-queue/src/tags.rs @@ -61,16 +61,20 @@ const CLEAR_ALL: &str = "*"; pub fn invalidate_fork_parent_cache(workspace_id: &str) { FORK_PARENT_CACHE.remove(workspace_id); windmill_common::workspaces::invalidate_root_workspace_cache(workspace_id); + windmill_common::fork_storage::invalidate_fork_storage_keys_cache(workspace_id); } -/// Apply a broadcast lineage change to this process's caches. +/// Apply a broadcast lineage change to this process's caches. The object storage key mapping is +/// among them: a replica keeping a recreated fork's old mapping would write the wrong keys. pub fn apply_fork_lineage_change(payload: &str) { if payload == CLEAR_ALL { FORK_PARENT_CACHE.clear(); windmill_common::workspaces::clear_root_workspace_cache(); + windmill_common::fork_storage::clear_fork_storage_keys_cache(); } else { FORK_PARENT_CACHE.remove(payload); windmill_common::workspaces::invalidate_root_workspace_cache(payload); + windmill_common::fork_storage::invalidate_fork_storage_keys_cache(payload); } } diff --git a/docs/fork-object-storage.md b/docs/fork-object-storage.md new file mode 100644 index 0000000000..6544edf902 --- /dev/null +++ b/docs/fork-object-storage.md @@ -0,0 +1,56 @@ +# Fork-isolated workspace object storage + +A fork clones its parent's `large_file_storage` settings, so without isolation it reads and +writes the parent's objects: a script iterating in a fork overwrites production's files. Each +storage the fork isolates (a `fork_storage_namespace` row, chosen at fork creation, default +isolated) instead maps logical keys: + +- **Writes** (put, multipart, delete, copy destination) go to `__wm_forks//`, + the same one-component segment DuckLake's fork namespaces use (`fork_data_dir_segment`). +- **Reads** try the fork's own key, then each ancestor's mapping, nearest first, then the root's + raw key. They stop at the first ancestor whose storage is a different one + (`lfs_entry_storage_ref`): its keys are not in this bucket. A shared fork maps like its parent. +- **Listings** show the workspace's own objects only, as logical keys. +- **Deletes** remove the fork's copy; an ancestor's object stays readable through it. + +The mapping (`windmill_common::fork_storage::ForkStorageKeys`) is applied wherever Windmill +turns a workspace-storage key into a storage operation: the `ForkStore` object-store wrapper +(`windmill_object_store::fork_store::workspace_object_store`) for every server-side helper, and +key rewriting in the S3 proxy's raw S3 forwarding (URI key, list `prefix`/`start-after`/`marker` +and response keys, `x-amz-copy-source`). A named resource (`s3_resource_path`) is not workspace +storage and is never mapped. A workspace whose keys are mapped is refused its workspace +storage's own credentials (`s3_resource_info`, the duckdb/polars settings endpoints): a client +holding them would write past the mapping. + +Not mapped, on purpose: AI sessions and volumes (their keys already carry the workspace id), +SMTP attachments (job-id keys), the storage usage count (it counts the whole bucket). Known +gaps: the worker's result-cache hash HEADs the logical key unmapped, and a read of an object +the fork does not own costs one HEAD per chain level before the GET. + +## Invariants + +- **`__wm_forks/` is reachable only through the workspace's own chain.** A key under it maps to + itself when it is in the workspace's own area (writes) or an area of the chain (reads), and is + refused otherwise; listings hide it. This is also what keeps DuckLake working: its fork data + paths arrive already prefixed, and its defer attaches read ancestors' raw keys through the + read chain. Every workspace gets the rule, isolating or not, so pre-existing forks' lakes keep + working. +- **No `.` or `..` key segments**, in any workspace: the proxy forwards mapped keys in a URL, + where they collapse and would climb out of the fork's prefix. +- **The prefix derives from the workspace id**, so an isolating fork's id cannot be changed + (`change_workspace_id` refuses): under a new id it would see none of its objects. Nor can an + isolated storage be repointed, since the purge targets the storage recorded at creation. +- **A shared lake needs its storage shared**: its files must land at the parent's keys, where the + parent's catalog looks for them. Fork creation refuses the other combination. +- **Children before parents**: deleting a fork that has forks of its own is refused when anything + in its chain isolates a storage, shared forks included. A child reads through, or writes into, + the isolating prefix via that link, and without it the child's chain would map like the root's. +- **Deletion and same-id creation are serialized** on the `fork:` advisory lock, and a purge + skips an id that exists again: the rows are then the new fork's. The proxy caches only the raw + client, never a mapping, so fork creation and deletion invalidate the one mapping cache. +- **Deletion always purges** the fork's area of each isolated storage, post-commit, from the + storage identity recorded at creation: nothing can reach it once the fork is gone. Rows have + no FK and outlive the workspace when the purge fails; creating a fork with the same id retries + them and refuses while they cannot be purged, since a same-id fork would read them. +- Isolation is a per-fork-creation choice (`shared_storages` on create-fork), never inherited. + Forks created before the registry existed have no rows and keep sharing. diff --git a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte index 378a96733f..33543efa82 100644 --- a/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte +++ b/frontend/src/lib/components/workspaceSettings/CreateWorkspaceInner.svelte @@ -76,6 +76,7 @@ import WorkspaceScopeTrigger from '../WorkspaceScopeTrigger.svelte' import DarkModeToggle from '../sidebar/DarkModeToggle.svelte' import ForkDucklakeSection from './ForkDucklakeSection.svelte' + import ForkStorageSection from './ForkStorageSection.svelte' import { createWorkspaceForkAndWait } from '$lib/utils/forkCreation' interface Props { @@ -277,6 +278,7 @@ let forkDatatableSection: ReturnType | undefined = $state(undefined) let forkDucklakeSection: ReturnType | undefined = $state(undefined) + let forkStorageSection: ReturnType | undefined = $state(undefined) let workspaceColor: string | undefined = $state(undefined) let colorEnabled = $state(false) @@ -499,6 +501,7 @@ color: colorEnabled && workspaceColor ? workspaceColor : undefined, forked_datatables: forkedDatatables, shared_ducklakes: forkDucklakeSection?.getSharedDucklakes() ?? [], + shared_storages: forkStorageSection?.getSharedStorages() ?? [], is_dev_workspace: createAsDevWorkspace, dev_workspace_label: createAsDevWorkspace ? devWorkspaceLabel : undefined, lock_prod_deploy: createAsDevWorkspace && effectiveLockProdDeploy, @@ -1021,6 +1024,7 @@ }} /> + {/if} {#if !automateUsernameCreation}