mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-19 16:03:29 +00:00
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 <noreply@anthropic.com> * fix: drop the boto3 config key and expose the proxied bucket to polars Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: proxy S3 settings only for server-signed tokens and keep tls verify Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat: isolate fork object storage under a per-fork prefix with read-through Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: guard fork storage isolation against traversal, shared lakes and orphaning Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: serialize fork purges with same-id creation and guard shared middle forks Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: drop fork storage mappings on every replica and lock fork deletion against child creation Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * 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 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5.5
windmill-internal-app[bot]
Claude Opus 5.5
windmill-internal-app[bot]
parent
e0138a7374
commit
9375805fdf
@@ -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/<segment>/` 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
|
||||
|
||||
+23
@@ -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"
|
||||
}
|
||||
+22
@@ -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"
|
||||
}
|
||||
+28
@@ -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"
|
||||
}
|
||||
+28
@@ -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"
|
||||
}
|
||||
+35
@@ -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"
|
||||
}
|
||||
+22
@@ -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"
|
||||
}
|
||||
+15
@@ -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"
|
||||
}
|
||||
+16
@@ -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"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
cd3395013d8484cd3917cdb6ee7837b6ac91e577
|
||||
455681e55eb7cdfcfdb7cd79cbae75b333e07943
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS fork_storage_namespace;
|
||||
@@ -0,0 +1,24 @@
|
||||
-- Workspace storages a fork/dev workspace isolates from its parent: its writes land under
|
||||
-- `__wm_forks/<fork segment>/` 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 (`<lfs type>:<resource path or root
|
||||
-- path>`), 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;
|
||||
@@ -515,6 +515,11 @@ struct CreateWorkspaceFork {
|
||||
/// fork namespace + read-defer.
|
||||
#[serde(default)]
|
||||
shared_ducklakes: Vec<String>,
|
||||
/// 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<String>,
|
||||
/// 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<DB>,
|
||||
@@ -2198,6 +2235,7 @@ async fn edit_large_file_storage_config(
|
||||
let serialized_lfs_config =
|
||||
serde_json::to_value::<LargeFileStorageWithSecondary>(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
|
||||
|
||||
@@ -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<Vec<PreparedForkStoragePurge>> {
|
||||
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<PreparedForkStoragePurge>,
|
||||
) -> Result<Vec<String>> {
|
||||
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/<segment>/` 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<PreparedForkStoragePurge>,
|
||||
) -> Vec<String> {
|
||||
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`
|
||||
|
||||
@@ -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)"
|
||||
|
||||
@@ -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<dyn windmill_object_store::object_store_reexports::ObjectStore>;
|
||||
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::<i64>().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);
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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/<fork segment>/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/<segment>/` in an isolating fork, `""` otherwise.
|
||||
write_prefix: String,
|
||||
/// Prefixes a read tries in order, nearest first.
|
||||
read_prefixes: Vec<String>,
|
||||
/// `__wm_forks/<segment>/` of this workspace, writable as is.
|
||||
own_area: String,
|
||||
/// The fork areas of this workspace and its ancestors, readable as is.
|
||||
readable_areas: Vec<String>,
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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: `<fork area>../../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<String>,
|
||||
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<String> {
|
||||
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<Vec<String>> {
|
||||
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<ListMapping> {
|
||||
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<ForkStorageKeys>, 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<Arc<ForkStorageKeys>> {
|
||||
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<ChainLink> = 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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<Arc<dyn ObjectStore>> {
|
||||
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<Arc<ForkStore>> {
|
||||
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<dyn ObjectStore>,
|
||||
keys: Arc<ForkStorageKeys>,
|
||||
}
|
||||
|
||||
impl ForkStore {
|
||||
pub fn new(inner: Arc<dyn ObjectStore>, keys: Arc<ForkStorageKeys>) -> 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<dyn ObjectStore> {
|
||||
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> {
|
||||
Path::parse(key).map_err(|source| object_store::Error::InvalidPath { source })
|
||||
}
|
||||
|
||||
impl ForkStore {
|
||||
fn write_path(&self, location: &Path) -> Result<Path> {
|
||||
parse(
|
||||
self.keys
|
||||
.write_key(location.as_ref())
|
||||
.map_err(|e| denied(location, e))?,
|
||||
)
|
||||
}
|
||||
|
||||
fn read_paths(&self, location: &Path) -> Result<Vec<Path>> {
|
||||
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<Path> {
|
||||
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<String> {
|
||||
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<PutResult> {
|
||||
self.inner
|
||||
.put_opts(&self.write_path(location)?, payload, opts)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn put_multipart_opts(
|
||||
&self,
|
||||
location: &Path,
|
||||
opts: PutMultipartOpts,
|
||||
) -> Result<Box<dyn MultipartUpload>> {
|
||||
self.inner
|
||||
.put_multipart_opts(&self.write_path(location)?, opts)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
|
||||
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<ObjectMeta> {
|
||||
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<ObjectMeta>> {
|
||||
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<ObjectMeta>> {
|
||||
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<ListResult> {
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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")]
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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/<fork segment>/<key>`,
|
||||
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:<id>` 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.
|
||||
@@ -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<typeof ForkDatatableSection> | undefined = $state(undefined)
|
||||
let forkDucklakeSection: ReturnType<typeof ForkDucklakeSection> | undefined = $state(undefined)
|
||||
let forkStorageSection: ReturnType<typeof ForkStorageSection> | 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 @@
|
||||
}}
|
||||
/>
|
||||
<ForkDucklakeSection bind:this={forkDucklakeSection} sourceWorkspace={baseWorkspaceId} />
|
||||
<ForkStorageSection bind:this={forkStorageSection} sourceWorkspace={baseWorkspaceId} />
|
||||
{/if}
|
||||
{#if !automateUsernameCreation}
|
||||
<Label label="Your username in that workspace">
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
<script lang="ts">
|
||||
import { SettingService } from '$lib/gen'
|
||||
import { workspaceStore } from '$lib/stores'
|
||||
import { resource } from 'runed'
|
||||
import Select from '../select/Select.svelte'
|
||||
import Label from '../Label.svelte'
|
||||
import { TriangleAlert } from 'lucide-svelte'
|
||||
|
||||
// The workspace the fork is created from, like ForkDucklakeSection: the storages
|
||||
// listed here must match the ones the create-fork call submits to.
|
||||
let { sourceWorkspace }: { sourceWorkspace?: string } = $props()
|
||||
let effectiveSource = $derived(sourceWorkspace ?? $workspaceStore ?? undefined)
|
||||
|
||||
let storages = resource(
|
||||
() => effectiveSource,
|
||||
async (ws) =>
|
||||
ws
|
||||
? SettingService.getSecondaryStorageNames({ workspace: ws, includeDefault: true })
|
||||
: undefined
|
||||
)
|
||||
|
||||
let behaviors: Record<string, 'isolated' | 'shared'> = $state({})
|
||||
|
||||
// Sent as `shared_storages` on the create-fork request; unlisted storages are isolated.
|
||||
export function getSharedStorages(): string[] {
|
||||
return (storages.current ?? []).filter((name) => behaviors[name] === 'shared')
|
||||
}
|
||||
|
||||
let anyShared = $derived((storages.current ?? []).some((n) => behaviors[n] === 'shared'))
|
||||
|
||||
function storageLabel(name: string): string {
|
||||
return name === '_default_' ? 'Primary storage' : name
|
||||
}
|
||||
</script>
|
||||
|
||||
<!-- Self-hides when the workspace has no object storage configured. -->
|
||||
{#if storages.current && storages.current.length > 0}
|
||||
<Label label="Object storage">
|
||||
<span class="text-xs text-secondary">
|
||||
By default each storage is isolated in the fork: the fork's writes go to its own prefix, and
|
||||
it reads this workspace's files until it writes its own. Deleting the fork deletes what it
|
||||
wrote.
|
||||
</span>
|
||||
<div class="border rounded-md divide-y">
|
||||
{#each storages.current as name (name)}
|
||||
<div class="flex items-center gap-2 justify-between px-4 py-1.5">
|
||||
<span class="text-xs font-medium">{storageLabel(name)}</span>
|
||||
<Select
|
||||
dropdownClass="max-w-96"
|
||||
bind:value={() => behaviors[name] ?? 'isolated', (v) => (behaviors[name] = v)}
|
||||
items={[
|
||||
{ value: 'isolated', label: 'Isolated (read through to parent)' },
|
||||
{ value: 'shared', label: 'Shared with parent' }
|
||||
]}
|
||||
/>
|
||||
</div>
|
||||
{/each}
|
||||
</div>
|
||||
{#if anyShared}
|
||||
<div
|
||||
class="flex items-start gap-2 px-3 py-2 text-xs rounded-md border border-amber-200 dark:border-amber-900/60 bg-amber-50 dark:bg-amber-950/40 text-amber-700 dark:text-amber-300"
|
||||
>
|
||||
<TriangleAlert size={14} class="shrink-0 mt-0.5" />
|
||||
<span>
|
||||
A shared storage is NOT isolated: scripts in the fork read, overwrite and delete this
|
||||
workspace's files directly.
|
||||
</span>
|
||||
</div>
|
||||
{/if}
|
||||
</Label>
|
||||
{/if}
|
||||
Reference in New Issue
Block a user