starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-28 00:02:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into migrate-session-to-fork

Browse Source 
# Conflicts:
#	backend/ee-repo-ref.txt
#	frontend/src/lib/components/sessions/SessionWrapper.svelte
#	frontend/src/lib/components/sessions/WorkspaceFamilyPicker.svelte
This commit is contained in:
Guilhem Lemouel
2026-10-23 05:27:00 -07:00
parent bf98ddb460 fb193d7a81
commit e88dd4ae28
1190 changed files with 71933 additions and 14475 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-workspaces/Cargo.toml
+2
View File
@@ -7,6 +7,8 @@ edition.workspace = true
[lib]
name = "windmill_api_workspaces"
path = "src/lib.rs"
# src/ has no doctests: remove this line before adding one, or cargo test never runs it.
doctest = false
[features]
default = []
backend/windmill-api-workspaces/src/datatable_clone.rs
+94 -17
View File
@@ -28,16 +28,22 @@ use windmill_common::workspaces::{
use windmill_common::{PgDatabase, DB};
use crate::datatable_acl::connect_with_notices;
use crate::datatable_migrations::{
lock_migration_runs_on, read_applied_migrations, write_applied_migrations, AppliedMigrations,
};
use crate::datatable_permissions::ensure_reaches_governing_datatable;
use crate::workspaces::{
create_database_on_server, ensure_datatable_is_clonable, pg_dump_database, pg_import_dump,
DumpFile, PgDumpOptions,
copy_database_settings, create_database_on_server, ensure_datatable_is_clonable,
pg_dump_database, pg_import_dump, DumpFile, PgDumpOptions,
};
/// A database this request created and filled for one data table of the fork.
/// A database this request created and filled for the data tables of the fork on one source
/// database.
pub(crate) struct MadeCopy {
/// The data table's name, in the parent and in the fork.
pub(crate) name: String,
/// The data tables' names, in the parent and in the fork, the one it was made for first. Data
/// tables on one database share its copy: a resource-backed fork has one resource for them all
/// to be pointed through.
pub(crate) names: Vec<String>,
pub(crate) dbname: String,
pub(crate) behavior: DataTableForkBehavior,
/// The database the source resolved to when it was copied.
@@ -237,6 +243,7 @@ pub(crate) async fn authorize_copies(
parent_w_id: &str,
requests: &[CopyRequest<'_>],
) -> Result<()> {
let mut sources: Vec<(&CopyRequest<'_>, GoverningDatatable)> = Vec::new();
for request in requests {
match request.behavior {
DataTableForkBehavior::KeepOriginal => {
@@ -265,9 +272,50 @@ pub(crate) async fn authorize_copies(
let governing = ensure_datatable_is_clonable(db, parent_w_id, request.name).await?;
ensure_reaches_governing_datatable(db, parent_w_id, request.name, &governing, authed)
.await?;
if governing.datatable.permissions.is_some() {
let under_roles = governing.datatable.permissions.is_some();
if under_roles {
crate::datatable_replay_oss::ensure_replay()?;
}
if let Some((first, first_governing)) = sources
.iter()
.find(|(_, earlier)| same_source(earlier, &governing))
{
ensure_shares_copy(first.name, first.behavior, request)?;
if first_governing.datatable.permissions.is_some() != under_roles {
return Err(Error::BadRequest(format!(
"Data tables '{}' and '{}' use the same database, which the fork copies once, \
but only one of them is under roles",
first.name, request.name
)));
}
}
sources.push((request, governing));
}
Ok(())
}
/// Whether two data tables are on one database. A resource is the governing workspace's.
fn same_source(a: &GoverningDatatable, b: &GoverningDatatable) -> bool {
same_database(&a.datatable.database, &b.datatable.database)
&& (a.workspace_id == b.workspace_id
|| a.datatable
.database
.as_ref()
.is_some_and(|d| d.resource_type.is_windmill_managed()))
}
/// Data tables on one database get one copy, made as the first of them asks.
fn ensure_shares_copy(
first: &str,
behavior: DataTableForkBehavior,
request: &CopyRequest<'_>,
) -> Result<()> {
if behavior != request.behavior {
return Err(Error::BadRequest(format!(
"Data tables '{}' and '{}' use the same database, which the fork copies once: fork \
them with the same behavior",
first, request.name
)));
}
Ok(())
}
@@ -281,10 +329,28 @@ pub(crate) async fn make_copies(
requests: &[CopyRequest<'_>],
) -> Result<Vec<MadeCopy>> {
authorize_copies(db, authed, parent_w_id, requests).await?;
let mut copies = Vec::with_capacity(requests.len());
let mut copies: Vec<MadeCopy> = Vec::with_capacity(requests.len());
// What each copy was made from, by index.
let mut sources: Vec<GoverningDatatable> = Vec::with_capacity(requests.len());
for request in requests {
// Checked again by the fork against the copy's `source_database`, under the locks.
let governing = match ensure_datatable_is_clonable(db, parent_w_id, request.name).await {
Ok(governing) => governing,
Err(e) => return Err(drop_copies_after(db, copies, e).await),
};
let shared = sources.iter().position(|s| same_source(s, &governing));
if let Some(c) = shared {
if let Err(e) = ensure_shares_copy(&copies[c].names[0], copies[c].behavior, request) {
return Err(drop_copies_after(db, copies, e).await);
}
copies[c].names.push(request.name.to_string());
continue;
}
match make_copy(db, authed, parent_w_id, request).await {
Ok(copy) => copies.push(copy),
Ok(copy) => {
copies.push(copy);
sources.push(governing);
}
Err(e) => return Err(drop_copies_after(db, copies, e).await),
}
}
@@ -336,8 +402,8 @@ async fn drop_copy(db: &DB, source_database: &DataTableDatabase, dbname: &str) -
Ok(())
}
DataTableCatalogResourceType::Postgresql => {
// On a server of the workspace's own, where a resource edited meanwhile can already name it
// and nothing locks such an edit: dropping it could take someone's data.
// On a server of the workspace's own, where a resource edited meanwhile can already name it
// and nothing locks such an edit: dropping it could take someone's data.
Err(Error::BadRequest(
"kept on its PostgreSQL server, where it may already be in use; drop it there once \
it is not, before forking the same data table under this id again"
@@ -392,20 +458,25 @@ async fn make_copy(
(server, Some(snapshot))
};
let schema_only = request.behavior == DataTableForkBehavior::SchemaOnly;
let migrations_lock = if schema_only {
Some(lock_migration_runs_on(db, &server, request.name).await?)
} else {
None
};
// Dumped before the database exists, so a source that cannot be read leaves nothing behind.
// Ownership never carries over: the restore runs as the target's connection user. Grants do,
// except on a cluster Windmill manages, which plants its own and where the replay below is what
// reproduces the roles'.
let dump = pg_dump_database(
&server,
PgDumpOptions {
schema_only: request.behavior == DataTableForkBehavior::SchemaOnly,
no_owner: true,
no_acl: managed,
..Default::default()
},
PgDumpOptions { schema_only, no_owner: true, no_acl: managed, ..Default::default() },
)
.await?;
let applied = match migrations_lock {
Some(client) => Some(read_applied_migrations(&client).await?),
None => None,
};
match source_database.resource_type {
DataTableCatalogResourceType::Instance => {
@@ -444,11 +515,12 @@ async fn make_copy(
&server,
&target,
&dump,
applied.as_ref(),
)
.await;
match filled {
Ok(replayed) => Ok(MadeCopy {
name: request.name.to_string(),
names: vec![request.name.to_string()],
dbname: request.dbname.to_string(),
behavior: request.behavior,
source_database,
@@ -484,8 +556,12 @@ async fn fill(
source: &PgDatabase,
target: &PgDatabase,
dump: &DumpFile,
applied: Option<&AppliedMigrations>,
) -> Result<bool> {
pg_import_dump(target, dump).await?;
if let Some(applied) = applied {
write_applied_migrations(db, target, applied).await?;
}
// Held until the replay commits: a role renamed or dropped meanwhile would change what the
// replay names, and a settings save could move the source onto another database or put it under
@@ -554,6 +630,7 @@ async fn fill(
})?;
}
tx.commit().await?;
copy_database_settings(db, source, target).await?;
Ok(replayed)
}
backend/windmill-api-workspaces/src/datatable_migration_scope.rs
+319
View File
@@ -0,0 +1,319 @@
/*
* Author: Ruben Fiszel
* Copyright: Windmill Labs, Inc 2022
* This file and its contents are licensed under the AGPLv3 License.
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
//! A data table migration's scope: the schemas it declares it touches, written as a
//! `-- schemas sales, billing` annotation in its up SQL. A migration that declares none applies to
//! the whole database.
//!
//! Scope decides ordering, not access: a migration waits on every earlier migration whose scope
//! overlaps its own, and its rollback on every later overlapping one. A whole-database migration
//! overlaps everything, so it is a barrier; migrations on disjoint schemas do not wait on each
//! other. Nothing checks that the SQL stays inside its declared scope.
use std::collections::HashSet;
use windmill_common::error::{Error, Result};
/// The schemas a migration declares, lowercased, or `None` for the whole database.
pub(crate) type Scope = Option<Vec<String>>;
const KEYWORD: &str = "schemas";
/// Read the `-- schemas a, b` annotation from the leading comment block of `code`, the way
/// `SqlAnnotations::datatable_role` reads `-- role`. A leading comment whose first word is
/// `schemas` is an annotation attempt, and a malformed one is an error rather than a silent
/// whole-database scope.
///
/// Names are lowercased: SQL folds unquoted identifiers, and two migrations spelling one schema
/// `Sales` and `sales` must still be ordered against each other. A false overlap only costs
/// waiting; a false disjointness lets a migration run before one it depends on.
pub(crate) fn parse_migration_schemas(code: &str) -> Result<Scope> {
for line in code.lines() {
let line = line.trim();
if line.is_empty() {
continue;
}
if !line.starts_with("--") {
break;
}
let body = line[2..].trim_start();
let Some(after) = body
.get(..KEYWORD.len())
.filter(|kw| kw.eq_ignore_ascii_case(KEYWORD))
.map(|_| &body[KEYWORD.len()..])
else {
continue;
};
if !after.is_empty()
&& !after.starts_with(char::is_whitespace)
&& !after.starts_with([':', '='])
{
continue;
}
let after = after.trim_start();
let after = after.strip_prefix([':', '=']).unwrap_or(after).trim();
let after = after.strip_suffix(';').unwrap_or(after);
let mut schemas: Vec<String> = vec![];
for piece in after.split(',') {
let name = piece.trim();
if !is_annotatable_schema_name(name) {
return Err(Error::BadRequest(format!(
"Malformed data table migration scope annotation: `{line}`. Write it as \
`-- schemas <name>, <name>` on a line of its own, where each name is \
letters, digits, '_', '$' or '-', or leave it out for a migration that \
applies to the whole database. A comment in the leading block that starts \
with the word 'schemas' is read as this annotation; move it below the first \
statement if it is prose."
)));
}
let name = name.to_ascii_lowercase();
if !schemas.contains(&name) {
schemas.push(name);
}
}
return Ok(Some(schemas));
}
Ok(None)
}
/// Whether `name` can be written in a `-- schemas` annotation.
fn is_annotatable_schema_name(name: &str) -> bool {
!name.is_empty()
&& name.len() <= 63
&& name
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '$' | '-'))
}
pub(crate) fn scopes_overlap(a: &Scope, b: &Scope) -> bool {
match (a, b) {
(Some(a), Some(b)) => a.iter().any(|s| b.contains(s)),
_ => true,
}
}
/// One migration of a data table's chain, as ordering sees it.
pub(crate) struct ChainEntry {
pub(crate) timestamp: i64,
pub(crate) name: String,
/// A malformed annotation counts as the whole database, so it orders conservatively. It is
/// refused only when a migration is written: one stored before scopes existed may open with
/// prose that starts with the word, and must keep running as it always did.
pub(crate) scope: Scope,
}
impl ChainEntry {
pub(crate) fn new(timestamp: i64, name: String, code_up: &str) -> Self {
let scope = parse_migration_schemas(code_up).unwrap_or(None);
Self { timestamp, name, scope }
}
fn label(&self) -> String {
format!("{} ({})", self.timestamp, self.name)
}
}
/// Why `later` has to wait for `earlier`, for an error message.
fn overlap_reason(later: &ChainEntry, earlier: &ChainEntry) -> String {
match (&later.scope, &earlier.scope) {
(_, None) => format!("{} applies to the whole database", earlier.label()),
(None, _) => format!("{} applies to the whole database", later.label()),
(Some(a), Some(b)) => {
let shared: Vec<&str> = a
.iter()
.filter(|s| b.contains(s))
.map(String::as_str)
.collect();
format!("both touch schema {}", shared.join(", "))
}
}
}
/// Which pending migrations one run applies.
pub(crate) enum RunSelection {
/// Every pending migration, or those up to and including a version.
UpTo(Option<i64>),
/// One version.
Only(i64),
/// The pending migrations that declare this schema. A whole-database one is never part of
/// it: running for one schema must not change the others.
Schema(String),
}
/// The chain indices one run applies, in order, or why it cannot start. Every selected migration
/// needs each earlier overlapping migration applied or applied by this same run, so the whole run
/// is refused before anything is applied rather than stopping half-way.
///
/// `Only` is exempt: it is the explicit out-of-order run, which the editors confirm with the user
/// (through `blocked_by`) before asking for it.
pub(crate) fn plan_run(
chain: &[ChainEntry],
applied: &HashSet<i64>,
selection: &RunSelection,
) -> Result<Vec<usize>> {
let pending = |i: usize| !applied.contains(&chain[i].timestamp);
let selected: Vec<usize> = (0..chain.len())
.filter(|&i| {
pending(i)
&& match selection {
RunSelection::UpTo(None) => true,
RunSelection::UpTo(Some(up_to)) => chain[i].timestamp <= *up_to,
RunSelection::Only(version) => chain[i].timestamp == *version,
RunSelection::Schema(schema) => {
chain[i].scope.as_ref().is_some_and(|s| s.contains(schema))
}
}
})
.collect();
if matches!(selection, RunSelection::Only(_)) {
return Ok(selected);
}
let in_run: HashSet<usize> = selected.iter().copied().collect();
for &i in &selected {
let blocker = (0..i).find(|&j| {
pending(j) && !in_run.contains(&j) && scopes_overlap(&chain[i].scope, &chain[j].scope)
});
if let Some(j) = blocker {
return Err(Error::BadRequest(format!(
"Migration {} cannot run before {}, which is not applied yet: {}. Apply it first, \
or run all pending migrations.",
chain[i].label(),
chain[j].label(),
overlap_reason(&chain[i], &chain[j]),
)));
}
}
Ok(selected)
}
/// The earliest pending migration `chain[i]` has to wait for, when it is pending; the latest
/// applied migration that has to be rolled back before it, when it is applied.
pub(crate) fn blocked_by(chain: &[ChainEntry], applied: &HashSet<i64>, i: usize) -> Option<usize> {
let is_applied = |j: usize| applied.contains(&chain[j].timestamp);
let overlaps = |j: usize| scopes_overlap(&chain[i].scope, &chain[j].scope);
if is_applied(i) {
(i + 1..chain.len())
.rev()
.find(|&j| is_applied(j) && overlaps(j))
} else {
(0..i).find(|&j| !is_applied(j) && overlaps(j))
}
}
#[cfg(test)]
mod tests {
use super::*;
fn s(names: &[&str]) -> Scope {
Some(names.iter().map(|n| n.to_string()).collect())
}
#[test]
fn schemas_annotation_is_read_from_the_leading_comment_block() {
let p = |code| parse_migration_schemas(code).unwrap();
assert_eq!(p("-- schemas sales\nBEGIN;"), s(&["sales"]));
assert_eq!(
p("-- role analyst\n-- Schemas: Sales, billing,sales;\nBEGIN;"),
s(&["sales", "billing"])
);
assert_eq!(p("-- schemas=a\nBEGIN;"), s(&["a"]));
assert_eq!(p("BEGIN;\n-- schemas sales"), None);
assert_eq!(p("-- schemasx sales\nBEGIN;"), None);
assert_eq!(p("-- up migration: x\nBEGIN;"), None);
for malformed in [
"-- schemas\nBEGIN;",
"-- schemas:\nBEGIN;",
"-- schemas sales billing\nBEGIN;",
"-- schemas: changes for the orders table\nBEGIN;",
"-- schemas a,,b\nBEGIN;",
"-- schemas a.b\nBEGIN;",
] {
assert!(parse_migration_schemas(malformed).is_err(), "{malformed}");
}
}
fn chain(entries: &[(i64, Scope)]) -> Vec<ChainEntry> {
entries
.iter()
.map(|(t, scope)| ChainEntry {
timestamp: *t,
name: format!("m{t}"),
scope: scope.clone(),
})
.collect()
}
fn versions(c: &[ChainEntry], idx: Vec<usize>) -> Vec<i64> {
idx.into_iter().map(|i| c[i].timestamp).collect()
}
#[test]
fn a_run_waits_only_on_overlapping_earlier_migrations() {
let c = chain(&[
(1, None),
(2, s(&["sales"])),
(3, s(&["billing"])),
(4, s(&["sales", "billing"])),
(5, None),
(6, s(&["sales"])),
]);
let applied: HashSet<i64> = [1].into();
// Disjoint schemas don't wait on each other; a cross-schema migration waits on both, a
// whole-database one on everything.
assert_eq!(blocked_by(&c, &applied, 2).map(|j| c[j].timestamp), None);
assert_eq!(blocked_by(&c, &applied, 3).map(|j| c[j].timestamp), Some(2));
assert_eq!(blocked_by(&c, &applied, 4).map(|j| c[j].timestamp), Some(2));
// An explicit single run is the confirmed override.
assert_eq!(
versions(&c, plan_run(&c, &applied, &RunSelection::Only(4)).unwrap()),
vec![4]
);
// Running for a schema takes what declares it, refuses past a pending whole-database
// migration, and refuses when a selected migration also needs another schema's.
let for_schema =
|a: &HashSet<i64>, x: &str| plan_run(&c, a, &RunSelection::Schema(x.into()));
let err = for_schema(&applied, "sales").unwrap_err().to_string();
assert!(err.contains("4 (m4)") && err.contains("3 (m3)"), "{err}");
let applied: HashSet<i64> = [1, 3].into();
let err = for_schema(&applied, "sales").unwrap_err().to_string();
assert!(
err.contains("5 (m5)") && err.contains("whole database"),
"{err}"
);
let applied: HashSet<i64> = [1, 3, 5].into();
assert_eq!(
versions(&c, for_schema(&applied, "sales").unwrap()),
vec![2, 4, 6]
);
// Everything pending in order needs no check: each earlier pending one is in the run.
assert_eq!(
versions(
&c,
plan_run(&c, &[1].into(), &RunSelection::UpTo(None)).unwrap()
),
vec![2, 3, 4, 5, 6]
);
}
#[test]
fn a_rollback_waits_on_later_overlapping_migrations() {
let c = chain(&[
(1, None),
(2, s(&["sales"])),
(3, s(&["billing"])),
(4, s(&["sales"])),
]);
let applied: HashSet<i64> = [1, 2, 3, 4].into();
let blocker = |i| blocked_by(&c, &applied, i).map(|j| c[j].timestamp);
assert_eq!(blocker(2), None);
assert_eq!(blocker(1), Some(4));
assert_eq!(blocker(0), Some(4));
}
}
backend/windmill-api-workspaces/src/datatable_migrations.rs
+213 -38
View File
@@ -10,6 +10,9 @@
//! management, and the workspace-merge diff helper. Split out of `workspaces.rs`
//! to keep that file focused on core workspace configuration.
use crate::datatable_migration_scope::{
blocked_by, parse_migration_schemas, plan_run, ChainEntry, RunSelection,
};
use crate::workspaces::{
managed_datatable_kind, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
PgDumpOptions,
@@ -154,15 +157,30 @@ async fn ensure_may_edit_migration(
Ok(())
}
/// Operators run deployed code only, and a migration is SQL written or run by hand. Checked before
/// anything else, including the role checks: on a data table not under roles those pass everyone.
fn ensure_not_operator(authed: &ApiAuthed) -> Result<()> {
if authed.is_operator {
return Err(Error::NotAuthorized(
"Operators cannot manage or run data table migrations".to_string(),
));
}
Ok(())
/// A migration is SQL written or run by hand, so it needs a role that may run code of its own.
/// Checked before anything else, including the data table role checks: on a data table not under
/// roles those pass everyone.
async fn ensure_not_operator(db: &DB, w_id: &str, authed: &ApiAuthed) -> Result<()> {
windmill_common::workspaces::check_role_can_code(
db,
w_id,
authed,
"manage or run data table migrations",
)
.await
}
/// Whether new migrations declare a scope, which the editors push for but do not require.
fn log_migration_scope(scope: &crate::datatable_migration_scope::Scope) {
windmill_common::feature_usage::log_feature_usage(
"datatable",
"migration_scope",
if scope.is_some() {
"scoped"
} else {
"whole_database"
},
);
}
#[derive(Serialize)]
@@ -182,8 +200,11 @@ struct RunDatatableMigrationsQuery {
up_to: Option<i64>,
/// When set, apply only this specific migration version (if not already
/// applied), ignoring any other pending migrations. Takes precedence over
/// `up_to`.
/// `up_to` and `schema`.
only: Option<i64>,
/// When set, apply only the pending migrations that declare this schema in their
/// `-- schemas` annotation. Takes precedence over `up_to`.
schema: Option<String>,
}
/// Build the `database` argument for a migration job. Both resource-backed and
@@ -509,6 +530,127 @@ async fn hold_migration_lock(
}
}
/// The `_wm_migrations` rows of a database a schema-only copy is being made of, read with
/// [`lock_migration_runs_on`] held so they describe the schema the dump captured.
pub(crate) struct AppliedMigrations {
/// `None` when the source has no `_wm_migrations`.
schema: Option<String>,
datatables: Vec<String>,
versions: Vec<i64>,
installed_at: Vec<String>,
}
/// Hold the migration-run lock on database `pg`, for a copy whose schema has to agree with the
/// `_wm_migrations` rows it carries. The lock is per database, whichever data table names it.
pub(crate) async fn lock_migration_runs_on(
db: &DB,
pg: &PgDatabase,
label: &str,
) -> Result<tokio_postgres::Client> {
let (client, connection) = pg.connect(Some(db)).await?;
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::error!("Datatable connection error: {}", e);
}
});
hold_migration_lock(&client, label, MIGRATION_LOCK_WAIT_SECS).await?;
Ok(client)
}
/// Read every `_wm_migrations` row of the database `client` is connected to, along with the
/// schema the table is in: the copy is written to by a connection whose `search_path` need not
/// resolve the way this one does.
pub(crate) async fn read_applied_migrations(
client: &tokio_postgres::Client,
) -> Result<AppliedMigrations> {
let read_err = |e: tokio_postgres::Error| {
Error::internal_err(format!(
"Failed to read _wm_migrations: {}",
pg_error_message(&e)
))
};
let schema: Option<String> = client
.query_one(
"SELECT n.nspname::text FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \
RIGHT JOIN (SELECT to_regclass('_wm_migrations')::oid AS oid) t ON c.oid = t.oid",
&[],
)
.await
.map_err(read_err)?
.get(0);
let mut applied =
AppliedMigrations { schema, datatables: vec![], versions: vec![], installed_at: vec![] };
let Some(schema) = applied.schema.as_deref() else {
return Ok(applied);
};
let rows = client
.query(
&format!(
// ISO 8601 in UTC: `::text` follows the session's `DateStyle`, which the copy's
// session need not parse the same way.
"SELECT datatable, version, \
to_char(installed_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') \
FROM {}._wm_migrations",
render_db_quoted_identifier(schema, DbType::Postgresql)
),
&[],
)
.await
.map_err(read_err)?;
for row in rows {
applied.datatables.push(row.get(0));
applied.versions.push(row.get(1));
applied.installed_at.push(row.get(2));
}
Ok(applied)
}
/// Write the rows [`read_applied_migrations`] read into a copy restored from a schema-only dump,
/// which brings `_wm_migrations` over empty: without them the copy reports every migration its
/// schema already holds as pending, and running them fails on objects that exist.
pub(crate) async fn write_applied_migrations(
db: &DB,
target: &PgDatabase,
applied: &AppliedMigrations,
) -> Result<()> {
let Some(schema) = applied.schema.as_deref() else {
return Ok(());
};
if applied.versions.is_empty() {
return Ok(());
}
let (client, connection) = target.connect(Some(db)).await?;
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::error!("Datatable connection error: {}", e);
}
});
client
.execute(
&format!(
"INSERT INTO {}._wm_migrations (datatable, version, installed_at) \
SELECT d, v, i::timestamp AT TIME ZONE 'UTC' \
FROM unnest($1::text[], $2::bigint[], $3::text[]) AS t(d, v, i) \
ON CONFLICT DO NOTHING",
render_db_quoted_identifier(schema, DbType::Postgresql)
),
&[
&applied.datatables,
&applied.versions,
&applied.installed_at,
],
)
.await
.map_err(|e| {
Error::internal_err(format!(
"Failed to copy _wm_migrations into '{}': {}",
target.dbname,
pg_error_message(&e)
))
})?;
Ok(())
}
/// Whether `e` is the lock wait running out, as opposed to the database being unreachable.
fn is_migration_lock_busy(e: &Error) -> bool {
matches!(e, Error::Generic(StatusCode::CONFLICT, _))
@@ -541,7 +683,9 @@ async fn read_applied_versions_on_client(
/// Each migration runs as a normal Windmill `postgresql` job (permissioned as
/// the requester, labelled `datatable_migration`); applied versions are then
/// recorded in the data table's own `_wm_migrations` table, so only migrations
/// not recorded there are run, in ascending `timestamp` order.
/// not recorded there are run, in ascending `timestamp` order. A run for one
/// schema is refused while an earlier migration its selection overlaps is pending
/// (see `datatable_migration_scope`).
async fn run_datatable_migrations(
authed: ApiAuthed,
Extension(db): Extension<DB>,
@@ -549,7 +693,7 @@ async fn run_datatable_migrations(
Path((w_id, datatable_name)): Path<(String, String)>,
Query(query): Query<RunDatatableMigrationsQuery>,
) -> JsonResult<RunDatatableMigrationsResult> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
// Before the admin connection is opened at all: the bookkeeping below is created and read
// through it, so a caller no role covers must be refused here rather than after the fact.
crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
@@ -587,30 +731,29 @@ async fn run_datatable_migrations(
let applied_versions = read_applied_versions_on_client(&client, &datatable_name).await?;
let selection = if let Some(only) = query.only {
RunSelection::Only(only)
} else if let Some(schema) = query.schema.as_deref() {
RunSelection::Schema(schema.to_ascii_lowercase())
} else {
RunSelection::UpTo(query.up_to)
};
// How the user scoped the run, for the counter emitted on the first migration
// that lands below.
let scope = if query.only.is_some() {
"only"
} else if query.up_to.is_some() {
"up_to"
} else {
"all"
let scope = match selection {
RunSelection::Only(_) => "only",
RunSelection::Schema(_) => "schema",
RunSelection::UpTo(Some(_)) => "up_to",
RunSelection::UpTo(None) => "all",
};
let chain: Vec<ChainEntry> = migrations
.iter()
.map(|m| ChainEntry::new(m.timestamp, m.name.clone(), &m.code_up))
.collect();
let plan = plan_run(&chain, &applied_versions, &selection)?;
let mut applied = Vec::new();
for m in migrations {
if let Some(only) = query.only {
// Run a single specific migration, skipping every other one.
if m.timestamp != only {
continue;
}
} else if query.up_to.is_some_and(|up_to| m.timestamp > up_to) {
// Migrations are ordered ascending, so once we pass `up_to` we're done.
break;
}
if applied_versions.contains(&m.timestamp) {
continue;
}
for m in plan.into_iter().map(|i| &migrations[i]) {
ensure_migration_role_allowed(
&db,
&w_id,
@@ -642,7 +785,7 @@ async fn run_datatable_migrations(
pg_error_message(&e)
))
})?;
applied.push(AppliedMigration { version: m.timestamp, name: m.name });
applied.push(AppliedMigration { version: m.timestamp, name: m.name.clone() });
// One event per run that moved the data table forward, emitted on the
// first migration that lands rather than after the loop: a later one
// failing returns early, and that run still advanced the data table. A
@@ -686,7 +829,7 @@ async fn rollback_datatable_migrations(
Path((w_id, datatable_name)): Path<(String, String)>,
Query(query): Query<RollbackDatatableMigrationsQuery>,
) -> JsonResult<RollbackDatatableMigrationsResult> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
// Before the admin connection is opened at all: the bookkeeping below is created and read
// through it, so a caller no role covers must be refused here rather than after the fact.
crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
@@ -865,6 +1008,13 @@ struct DatatableMigrationWithStatus {
#[serde(skip_serializing_if = "Option::is_none")]
code_down: Option<String>,
status: DatatableMigrationRunStatus,
/// The schemas its `-- schemas` annotation declares; absent for the whole database.
#[serde(skip_serializing_if = "Option::is_none")]
schemas: Option<Vec<String>>,
/// When not run: the earliest pending migration that has to be applied before it. When run:
/// the latest applied migration that has to be rolled back before it.
#[serde(skip_serializing_if = "Option::is_none")]
blocked_by: Option<i64>,
}
#[derive(Serialize)]
@@ -981,9 +1131,14 @@ async fn datatable_migrations_status(
Err(e) => (None, Some(e.to_string())),
};
let chain: Vec<ChainEntry> = defs
.iter()
.map(|m| ChainEntry::new(m.timestamp, m.name.clone(), &m.code_up))
.collect();
let migrations = defs
.into_iter()
.map(|m| {
.enumerate()
.map(|(i, m)| {
let status = match &applied {
Some(set) if set.contains(&m.timestamp) => DatatableMigrationRunStatus::Ran,
Some(_) => DatatableMigrationRunStatus::NotRun,
@@ -995,6 +1150,11 @@ async fn datatable_migrations_status(
code_up: m.code_up,
code_down: m.code_down,
status,
schemas: chain[i].scope.clone(),
blocked_by: applied
.as_ref()
.and_then(|set| blocked_by(&chain, set, i))
.map(|j| chain[j].timestamp),
}
})
.collect();
@@ -1330,9 +1490,10 @@ async fn create_datatable_migration(
Path((w_id, datatable_name)): Path<(String, String)>,
Json(payload): Json<CreateDatatableMigration>,
) -> JsonResult<DatatableMigration> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
validate_datatable_path_segment(&datatable_name)?;
validate_migration_name(&payload.name)?;
let scope = parse_migration_schemas(&payload.code_up)?;
ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;
ensure_may_edit_migration(
&db,
@@ -1379,6 +1540,7 @@ async fn create_datatable_migration(
.await?;
windmill_common::feature_usage::log_feature_usage("datatable", "migration_created", "manual");
log_migration_scope(&scope);
Ok(Json(DatatableMigration {
datatable: datatable_name,
@@ -1395,7 +1557,7 @@ async fn delete_datatable_migration(
Extension(db): Extension<DB>,
Path((w_id, datatable_name, timestamp)): Path<(String, String, i64)>,
) -> Result<String> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
let deleted_message = format!("Deleted migration {} from {}", timestamp, datatable_name);
// A definition already gone is not an error: `wmill sync push` deletes whatever the repo
@@ -1532,7 +1694,7 @@ async fn upsert_datatable_migration(
Path((w_id, datatable_name)): Path<(String, String)>,
Json(payload): Json<UpsertDatatableMigration>,
) -> Result<String> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
validate_datatable_path_segment(&datatable_name)?;
validate_migration_name(&payload.name)?;
ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;
@@ -1581,6 +1743,16 @@ async fn upsert_datatable_migration(
if unchanged {
return Ok(upserted_message);
}
// Only up SQL being written is held to the annotation: a stored up that opens with prose
// starting with the word `schemas` predates scopes, and must still accept a down added to it.
let up_changed = existing
.as_ref()
.is_none_or(|existing| existing.code_up != payload.code_up);
let scope = if up_changed {
parse_migration_schemas(&payload.code_up)?
} else {
None
};
ensure_may_edit_migration(
&db,
&w_id,
@@ -1709,6 +1881,9 @@ async fn upsert_datatable_migration(
"edited"
},
);
if existing.is_none() {
log_migration_scope(&scope);
}
Ok(upserted_message)
}
@@ -1723,7 +1898,7 @@ async fn generate_initial_datatable_migration(
Extension(db): Extension<DB>,
Path((w_id, datatable_name)): Path<(String, String)>,
) -> JsonResult<DatatableMigration> {
ensure_not_operator(&authed)?;
ensure_not_operator(&db, &w_id, &authed).await?;
// Returns a `pg_dump` of the whole schema and writes into the data table's own bookkeeping, so
// it answers to the workspace that governs it rather than to whoever is asking.
ensure_datatable_admin_access(
backend/windmill-api-workspaces/src/lib.rs
+2
View File
@@ -4,12 +4,14 @@ pub mod data_metrics;
pub mod datatable_acl;
pub mod datatable_acl_oss;
pub mod datatable_clone;
pub mod datatable_migration_scope;
pub mod datatable_migrations;
pub mod datatable_permissions;
pub mod datatable_permissions_oss;
pub mod datatable_replay_oss;
pub mod deployment_requests;
pub mod remote_deploy;
pub mod workspace_roles;
pub mod workspaces;
pub mod workspaces_extra;
pub mod workspaces_oss;
backend/windmill-api-workspaces/src/workspace_roles.rs
+290
View File
@@ -0,0 +1,290 @@
//! The workspace's roles and its overrides of them. See `windmill_common::roles`.
use axum::{
extract::{Extension, Path},
routing::{get, post},
Json, Router,
};
use serde::{Deserialize, Serialize};
use windmill_api_auth::ApiAuthed;
use windmill_audit::{audit_oss::audit_log, ActionKind};
use windmill_common::{
error::{Error, JsonResult, Result},
roles::{
self, instance_roles, load_role_overrides, EffectiveRole, RoleCapabilities, RoleOverrides,
RoleSource, OPERATOR_ROLE,
},
utils::require_admin,
DB,
};
use windmill_git_sync::handle_deployment_metadata;
pub fn workspaced_service() -> Router {
Router::new()
.route("/roles", get(list_roles))
.route("/role_overrides", post(update_role_overrides))
}
#[derive(Serialize)]
struct WorkspaceRoles {
roles: Vec<EffectiveRole>,
overrides: RoleOverrides,
/// Whether this workspace may add roles of its own.
custom_roles_enabled: bool,
/// The caller's role, `None` for a caller with no membership (a superadmin, a guest).
my_role: Option<String>,
my_capabilities: RoleCapabilities,
}
async fn list_roles(
authed: ApiAuthed,
Path(w_id): Path<String>,
Extension(db): Extension<DB>,
) -> JsonResult<WorkspaceRoles> {
let roles = roles::workspace_roles(&db, &w_id).await?;
let overrides = load_role_overrides(&db, &w_id).await?;
let my_role = sqlx::query!(
"SELECT workspace_role, operator FROM usr WHERE workspace_id = $1 AND username = $2",
&w_id,
&authed.username
)
.fetch_optional(&db)
.await?
.map(|r| roles::resolve_role_name(r.workspace_role, r.operator));
// Not `caller_capabilities`: its fast path answers FULL for an unrestricted member, which
// drops the visibility a role may still withhold.
let my_capabilities = match &my_role {
_ if authed.is_admin => roles::RoleCapabilities::FULL,
Some(role) => roles::role_capabilities_by_name(&db, &w_id, role).await?,
None => roles::RoleCapabilities::NONE,
};
Ok(Json(WorkspaceRoles {
roles: roles.as_ref().clone(),
overrides,
custom_roles_enabled: roles::custom_roles_enabled(),
my_role,
my_capabilities,
}))
}
async fn update_role_overrides(
authed: ApiAuthed,
Path(w_id): Path<String>,
Extension(db): Extension<DB>,
Json(overrides): Json<RoleOverrides>,
) -> Result<String> {
require_admin(authed.is_admin, &authed.username)?;
write_role_overrides(&db, &authed, &w_id, overrides, None).await?;
Ok("Role overrides updated".to_string())
}
/// Replaces the workspace's role overrides with `overrides`. Refuses removing a workspace-only role
/// anything still names, and a change that pushes an offline license over its seat cap.
/// `operator_settings` is the legacy `/operator_settings` payload, mirrored into its column for
/// git sync and for servers that predate roles.
pub(crate) async fn write_role_overrides(
db: &DB,
authed: &ApiAuthed,
w_id: &str,
overrides: RoleOverrides,
operator_settings: Option<serde_json::Value>,
) -> Result<()> {
let instance = instance_roles(db).await?;
roles::validate_overrides(&instance, &overrides)?;
roles::check_frozen_overrides(
&instance,
&roles::load_role_overrides(db, w_id).await?,
&overrides,
)?;
let previous = roles::workspace_roles(db, w_id).await?;
let removed: Vec<String> = previous
.iter()
.filter(|r| r.source == RoleSource::Workspace && !overrides.contains_key(&r.name))
.map(|r| r.name.clone())
.collect();
for name in &removed {
roles::ensure_role_unused(db, Some(w_id), name).await?;
}
let overrides_json = serde_json::to_value(&overrides)
.map_err(|e| Error::internal_err(format!("serializing role overrides: {e:#}")))?;
#[cfg(feature = "enterprise")]
let seat_baseline = windmill_common::ee_oss::seat_cap_baseline(db).await?;
let mut tx = db.begin().await?;
sqlx::query!(
"UPDATE workspace_settings SET role_overrides = $1 WHERE workspace_id = $2",
if overrides.is_empty() {
None
} else {
Some(&overrides_json)
},
w_id
)
.execute(&mut *tx)
.await?;
roles::notify_role_overrides_change(&mut tx, w_id).await?;
if let Some(settings) = &operator_settings {
sqlx::query!(
"UPDATE workspace_settings
SET operator_settings = COALESCE(operator_settings, '{}'::jsonb) || $1
WHERE workspace_id = $2",
settings,
w_id
)
.execute(&mut *tx)
.await?;
}
roles::invalidate_workspace_roles_cache(w_id);
roles::sync_role_seats(&mut tx, Some(w_id)).await?;
#[cfg(feature = "enterprise")]
if let Some(msg) =
windmill_common::ee_oss::check_seat_cap_for_role_change(&seat_baseline, &mut tx).await?
{
return Err(Error::BadRequest(msg));
}
audit_log(
&mut *tx,
authed,
if operator_settings.is_some() {
"workspaces.update_operator_settings"
} else {
"workspaces.update_role_overrides"
},
ActionKind::Update,
w_id,
None,
Some([("role_overrides", overrides_json.to_string().as_str())].into()),
)
.await?;
tx.commit().await?;
roles::invalidate_workspace_roles_cache(w_id);
handle_deployment_metadata(
&authed.email,
&authed.username,
db,
w_id,
windmill_git_sync::DeployedObject::Settings {
setting_type: if operator_settings.is_some() {
"operator_settings".to_string()
} else {
"role_overrides".to_string()
},
},
Some("Workspace roles updated".to_string()),
false,
None,
)
.await?;
Ok(())
}
/// The legacy `/operator_settings` payload. Visibility flags always serialize; the rights are
/// `Option` because git-sync posts whole files that may predate them, and an absent key must
/// leave the stored value alone.
#[derive(Deserialize, Serialize)]
pub(crate) struct ChangeOperatorSettings {
#[serde(default)]
runs: bool,
#[serde(default)]
schedules: bool,
#[serde(default)]
resources: bool,
#[serde(default)]
variables: bool,
#[serde(default)]
assets: bool,
#[serde(default)]
triggers: bool,
#[serde(default)]
audit_logs: bool,
#[serde(default)]
groups: bool,
#[serde(default)]
folders: bool,
#[serde(default)]
workers: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
builder_flows: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
builder_apps: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
manage_schedules: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
manage_triggers: Option<bool>,
}
/// Applies a legacy operator settings payload as an override of the `operator` role. A key equal
/// to what the instance gives the role is dropped rather than pinned, so it keeps following the
/// instance definition.
pub(crate) async fn update_operator_settings(
authed: ApiAuthed,
Path(w_id): Path<String>,
Extension(db): Extension<DB>,
Json(settings): Json<ChangeOperatorSettings>,
) -> Result<String> {
require_admin(authed.is_admin, &authed.username)?;
let instance = instance_roles(&db).await?;
let Some(base) = roles::effective_roles(&instance, &RoleOverrides::new())?
.into_iter()
.find(|r| r.name == OPERATOR_ROLE)
else {
return Err(Error::BadRequest(
"this instance defines no `operator` role: edit the workspace roles instead"
.to_string(),
));
};
let base = base.capabilities;
let mut overrides = load_role_overrides(&db, &w_id).await?;
let current = roles::workspace_roles(&db, &w_id)
.await?
.iter()
.find(|r| r.name == OPERATOR_ROLE)
.map(|r| r.capabilities)
.unwrap_or(base);
let mut target = current;
let v = &mut target.visibility;
(v.runs, v.schedules, v.resources, v.variables, v.assets) = (
settings.runs,
settings.schedules,
settings.resources,
settings.variables,
settings.assets,
);
(v.triggers, v.audit_logs, v.groups, v.folders, v.workers) = (
settings.triggers,
settings.audit_logs,
settings.groups,
settings.folders,
settings.workers,
);
let compose_or_keep = |cur: roles::AuthorLevel, grant: Option<bool>| match grant {
Some(true) if cur == roles::AuthorLevel::None => roles::AuthorLevel::Compose,
Some(false) => roles::AuthorLevel::None,
_ => cur,
};
target.flows = compose_or_keep(current.flows, settings.builder_flows);
target.apps = compose_or_keep(current.apps, settings.builder_apps);
if let Some(b) = settings.manage_schedules {
target.manage_schedules = b;
}
if let Some(b) = settings.manage_triggers {
target.manage_triggers = b;
}
let entry = overrides.entry(OPERATOR_ROLE.to_string()).or_default();
entry.capabilities = roles::diff_capabilities(&base, &target);
if entry.label.is_none() && entry.capabilities == Default::default() {
overrides.remove(OPERATOR_ROLE);
}
let settings_json = serde_json::json!(settings);
write_role_overrides(&db, &authed, &w_id, overrides, Some(settings_json)).await?;
Ok("Operator settings updated successfully".to_string())
}
backend/windmill-api-workspaces/src/workspaces.rs
+979 -224
View File
File diff suppressed because it is too large Load Diff
backend/windmill-api-workspaces/src/workspaces_extra.rs
+288 -52
View File
@@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use windmill_api_auth::{require_super_admin, ApiAuthed};
use windmill_common::DB;
@@ -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
@@ -118,7 +135,7 @@ pub(crate) async fn change_workspace_id(
// Duplicate workspace settings (keep copy in old workspace for reference)
info!("Duplicating workspace_settings table");
sqlx::query!(
"INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url, add_admins_and_developers_to_forks, remote_deploy_target, remote_deploy_target_changed_at) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url, add_admins_and_developers_to_forks, remote_deploy_target, remote_deploy_target_changed_at FROM workspace_settings WHERE workspace_id = $2",
"INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, role_overrides, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, slack_signing_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url, add_admins_and_developers_to_forks, remote_deploy_target, remote_deploy_target_changed_at) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, role_overrides, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, slack_signing_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url, add_admins_and_developers_to_forks, remote_deploy_target, remote_deploy_target_changed_at FROM workspace_settings WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
@@ -208,6 +225,14 @@ pub(crate) async fn change_workspace_id(
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE role_seat SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating account table");
sqlx::query!(
"UPDATE account SET workspace_id = $1 WHERE workspace_id = $2",
@@ -839,7 +864,7 @@ pub(crate) async fn change_workspace_id(
info!("Duplicating usr table");
sqlx::query!(
"INSERT INTO usr SELECT $1, username, email, is_admin, created_at, operator, disabled, role FROM usr WHERE workspace_id = $2",
"INSERT INTO usr (workspace_id, username, email, is_admin, created_at, operator, disabled, role, workspace_role) SELECT $1, username, email, is_admin, created_at, operator, disabled, role, workspace_role FROM usr WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
@@ -868,11 +893,21 @@ pub(crate) async fn change_workspace_id(
"Re-pushing jobs for {} enabled schedules",
enabled_schedule_paths.len()
);
for schedule_path in &enabled_schedule_paths {
if let Some(schedule) = get_schedule_opt(&mut *tx, &rw.new_id, schedule_path).await? {
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
windmill_queue::tags::carry_lineage_over_rename(&old_id, &rw.new_id, &db).await?;
let new_id = &rw.new_id;
let repushed = async {
for schedule_path in &enabled_schedule_paths {
if let Some(schedule) = get_schedule_opt(&mut *tx, new_id, schedule_path).await? {
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
}
}
Ok::<_, Error>(tx)
}
.await;
if repushed.is_err() {
windmill_queue::tags::invalidate_fork_parent_cache(new_id);
}
let mut tx = repushed?;
// After every workspace_settings write above: fork cleanup locks a settings row before the
// registry, so taking the registry first here would deadlock with it.
@@ -1099,6 +1134,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)
@@ -1203,6 +1266,10 @@ pub(crate) async fn delete_workspace(
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM role_seat WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM resource_type WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
@@ -1369,6 +1436,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);
@@ -1486,20 +1561,39 @@ pub async fn drop_forked_datatable_databases(
serde_json::from_value(datatable_config).unwrap_or_default();
let mut errors: Vec<String> = Vec::new();
// Data tables cloned from one database share its copy, which is dropped once for all of them.
// Keyed by where the database lives too: copies on other clusters or servers can share a name.
let mut dropped_dbs: HashSet<(String, Option<u16>, String)> = HashSet::new();
// Only a clone is droppable, and a clone is terminal by construction: a kept data table is a
// pointer at the parent's database, which this fork does not own.
let cloned_database = |name: &str| {
datatables
.get(name)
.filter(|dt| dt.forked_from.is_some())
.and_then(|dt| dt.database.as_ref())
};
for dt_name in &req.datatable_names {
// Only a clone is droppable, and a clone is terminal by construction: a kept data table is
// a pointer at the parent's database, which this fork does not own.
let database = match datatables.get(dt_name) {
Some(dt) if dt.forked_from.is_some() => match dt.database.as_ref() {
Some(database) => database,
None => continue,
},
_ => continue,
let Some(database) = cloned_database(dt_name) else {
continue;
};
if database.resource_type.is_windmill_managed() {
let db_to_drop = &database.resource_path;
let cluster = database.resource_type.as_ref().to_string();
if !dropped_dbs.insert((cluster, None, db_to_drop.clone())) {
continue;
}
let sharing: Vec<String> = req
.datatable_names
.iter()
.filter(|name| {
cloned_database(name).is_some_and(|d| {
d.resource_type == database.resource_type && &d.resource_path == db_to_drop
})
})
.cloned()
.collect();
if !db_to_drop.starts_with("wm_fork_") {
errors.push(format!(
"Refusing to drop instance database '{}' for datatable://{}: name does not start with 'wm_fork_'",
@@ -1513,12 +1607,8 @@ pub async fn drop_forked_datatable_databases(
// A task of its own, so a client going away cannot stop it between dropping the
// database and committing the entry's removal.
let dropped = tokio::spawn({
let (db, w_id, dt_name, db_to_drop) = (
db.clone(),
w_id.clone(),
dt_name.clone(),
db_to_drop.clone(),
);
let (db, w_id, sharing, db_to_drop) =
(db.clone(), w_id.clone(), sharing.clone(), db_to_drop.clone());
let resource_type = database.resource_type;
async move {
let mut tx = db.begin().await?;
@@ -1527,26 +1617,29 @@ pub async fn drop_forked_datatable_databases(
// rename this entry, or point another one here, either side of the check below.
windmill_common::workspaces::lock_fork_datatables(&mut tx, &w_id).await?;
// The snapshot above was read unlocked: a save committing since could have
// repointed this entry, and the entry is removed below whatever it names by then.
let current = sqlx::query_scalar::<_, Option<serde_json::Value>>(
"SELECT datatable->'datatables'->$2 FROM workspace_settings
WHERE workspace_id = $1 FOR UPDATE",
)
.bind(&w_id)
.bind(&dt_name)
.fetch_optional(&mut *tx)
.await?
.flatten()
.and_then(|v| serde_json::from_value::<DataTable>(v).ok());
if !current.is_some_and(|dt| {
dt.forked_from.is_some()
&& dt.database.is_some_and(|d| {
d.resource_type == resource_type && d.resource_path == db_to_drop
})
}) {
return Err(Error::BadRequest(
"the data table changed while it was being cleaned up".to_string(),
));
// repointed these entries, and they are removed below whatever they name by then.
for dt_name in &sharing {
let current = sqlx::query_scalar::<_, Option<serde_json::Value>>(
"SELECT datatable->'datatables'->$2 FROM workspace_settings
WHERE workspace_id = $1 FOR UPDATE",
)
.bind(&w_id)
.bind(dt_name)
.fetch_optional(&mut *tx)
.await?
.flatten()
.and_then(|v| serde_json::from_value::<DataTable>(v).ok());
if !current.is_some_and(|dt| {
dt.forked_from.is_some()
&& dt.database.is_some_and(|d| {
d.resource_type == resource_type
&& d.resource_path == db_to_drop
})
}) {
return Err(Error::BadRequest(format!(
"data table '{dt_name}' changed while it was being cleaned up"
)));
}
}
let external = resource_type
== windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance;
@@ -1564,11 +1657,13 @@ pub async fn drop_forked_datatable_databases(
)
.await?;
// Before the entry is removed: child fork pointers are found through it.
let uses = windmill_common::workspaces::managed_database_uses(
let names: Vec<&str> = sharing.iter().map(String::as_str).collect();
let uses = windmill_common::workspaces::managed_database_uses_except(
&mut tx,
resource_type,
&db_to_drop,
Some((w_id.as_str(), dt_name.as_str())),
&w_id,
&names,
)
.await?;
if !uses.is_empty() {
@@ -1579,20 +1674,22 @@ pub async fn drop_forked_datatable_databases(
}
// The entry goes with the database: a fork this one is cloned into afterwards
// must not inherit a pointer at a data table whose database is gone.
sqlx::query(
"UPDATE workspace_settings SET datatable = datatable #- ARRAY['datatables', $2]
WHERE workspace_id = $1",
)
.bind(&w_id)
.bind(&dt_name)
.execute(&mut *tx)
.await?;
for dt_name in &sharing {
sqlx::query(
"UPDATE workspace_settings SET datatable = datatable #- ARRAY['datatables', $2]
WHERE workspace_id = $1",
)
.bind(&w_id)
.bind(dt_name)
.execute(&mut *tx)
.await?;
}
if external {
// Checks the uses of the database itself, and unregisters it.
windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
&mut tx,
&db_to_drop,
Some((w_id.as_str(), dt_name.as_str())),
sharing.first().map(|name| (w_id.as_str(), name.as_str())),
)
.await?;
} else {
@@ -1615,10 +1712,38 @@ pub async fn drop_forked_datatable_databases(
if let Err(e) = dropped {
errors.push(format!(
"Could not drop instance database '{}' for datatable://{}: {}",
db_to_drop, dt_name, e
db_to_drop,
sharing.join(", datatable://"),
e
));
}
} else {
// Clones on one resource share its copy, which stays while any of them is kept. Only
// clones: a kept entry naming the resource cannot be put in the request.
let mut kept: Vec<&String> = datatables
.iter()
.filter(|(name, dt)| {
!req.datatable_names.contains(name)
&& dt.forked_from.is_some()
&& dt.database.as_ref().is_some_and(|d| {
d.resource_type == database.resource_type
&& d.resource_path == database.resource_path
})
})
.map(|(name, _)| name)
.collect();
if !kept.is_empty() {
kept.sort();
errors.push(format!(
"Could not drop the database of datatable://{dt_name}: it is still used by data \
tables {}",
kept.iter()
.map(|name| format!("'{name}'"))
.collect::<Vec<_>>()
.join(", ")
));
continue;
}
let fork_pg = match crate::workspaces::resolve_pg_source_checked(
&db,
&user_db,
@@ -1673,6 +1798,9 @@ pub async fn drop_forked_datatable_databases(
));
continue;
}
if !dropped_dbs.insert((parent_pg.host.clone(), parent_pg.port, db_to_drop.clone())) {
continue;
}
match parent_pg.connect(Some(&db)).await {
Ok((client, connection)) => {
@@ -2245,6 +2373,114 @@ 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>> {
// The lock only orders a purge against a same-id fork write; with nothing to purge, waiting
// on it would only hold fork creation's answer behind such a write.
if prepared.is_empty() {
return Ok(vec![]);
}
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`