diff --git a/backend/.sqlx/query-043cd99b3da09bca6cd2cf1e0adcd674de140b7f93897286447279e74777e99e.json b/backend/.sqlx/query-043cd99b3da09bca6cd2cf1e0adcd674de140b7f93897286447279e74777e99e.json new file mode 100644 index 0000000000..7f14dee4e2 --- /dev/null +++ b/backend/.sqlx/query-043cd99b3da09bca6cd2cf1e0adcd674de140b7f93897286447279e74777e99e.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH RECURSIVE chain AS (\n SELECT id, parent_workspace_id, is_dev_workspace, 0 AS depth\n FROM workspace WHERE id = $1\n UNION ALL\n SELECT w.id, w.parent_workspace_id, w.is_dev_workspace, chain.depth + 1\n FROM workspace w\n JOIN chain ON w.id = chain.parent_workspace_id\n WHERE chain.depth < 20\n )\n SELECT\n (SELECT id FROM chain\n WHERE parent_workspace_id IS NULL\n OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%')\n ORDER BY depth LIMIT 1) AS id,\n (SELECT parent_workspace_id IS NOT NULL FROM chain WHERE depth = 0) AS has_parent\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "has_parent", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "043cd99b3da09bca6cd2cf1e0adcd674de140b7f93897286447279e74777e99e" +} diff --git a/backend/.sqlx/query-0fb0fea54a1580d4b7ffd37ca43e169f51dbbdc3584d3522f25e97e180e307fd.json b/backend/.sqlx/query-0fb0fea54a1580d4b7ffd37ca43e169f51dbbdc3584d3522f25e97e180e307fd.json new file mode 100644 index 0000000000..2afefc7ab8 --- /dev/null +++ b/backend/.sqlx/query-0fb0fea54a1580d4b7ffd37ca43e169f51dbbdc3584d3522f25e97e180e307fd.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT config FROM config WHERE name LIKE 'worker\\_\\_%'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "config", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + true + ] + }, + "hash": "0fb0fea54a1580d4b7ffd37ca43e169f51dbbdc3584d3522f25e97e180e307fd" +} diff --git a/backend/.sqlx/query-2d22fe27288580cb727e43b2abe3be2e6e6e3b03e61ce8f2879627115224c8cf.json b/backend/.sqlx/query-2d22fe27288580cb727e43b2abe3be2e6e6e3b03e61ce8f2879627115224c8cf.json new file mode 100644 index 0000000000..95fd9a3a2f --- /dev/null +++ b/backend/.sqlx/query-2d22fe27288580cb727e43b2abe3be2e6e6e3b03e61ce8f2879627115224c8cf.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT config FROM config WHERE name = $1 FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "config", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "2d22fe27288580cb727e43b2abe3be2e6e6e3b03e61ce8f2879627115224c8cf" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index f2571ecd5c..15b4e190fe 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16229,6 +16229,7 @@ dependencies = [ "uuid", "windmill-api-auth", "windmill-common", + "windmill-queue", ] [[package]] diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index b40437560a..b68bd04e25 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -664708fff04d4f9e0660bcae3edb307fb2c9481b +71e9c0de8317e839f955ebef6a819062f20ac418 diff --git a/backend/src/main.rs b/backend/src/main.rs index fd953213a4..d5c51de819 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -44,7 +44,8 @@ use windmill_common::{ CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, - DEPENDENCY_JOB_TAG_SETTING, DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING, + DEPENDENCY_JOB_TAG_SETTING, DEV_WORKSPACE_ISOLATION_SETTING, + DEV_WORKSPACE_TAG_SUFFIX_SETTING, DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING, HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING, HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, @@ -129,7 +130,8 @@ use windmill_worker::{ use crate::monitor::{ initial_load, load_concurrency_key_max_queued, load_dependency_job_tag, - load_disable_password_login, load_fork_workspace_tag_append_fork_suffix, load_keep_job_dir, + load_dev_workspace_isolation, load_dev_workspace_tag_suffix, load_disable_password_login, + load_fork_workspace_tag_append_fork_suffix, load_keep_job_dir, load_mcp_disable_token_query_param, load_metrics_debug_enabled, load_preview_tags_override, load_require_preexisting_user, load_retention_period_overrides, load_tag_per_workspace_enabled, load_tag_per_workspace_workspaces, load_workspace_fairness_duration_secs, @@ -1052,12 +1054,8 @@ Windmill Community Edition {GIT_VERSION} // Before the first load: it can sign a token for instance object storage. #[cfg(all(feature = "enterprise", feature = "openidconnect", feature = "private"))] - windmill_common::oidc_ee::init_oidc_signing( - &conn, - server_mode, - killpill_rx.resubscribe(), - ) - .await?; + windmill_common::oidc_ee::init_oidc_signing(&conn, server_mode, killpill_rx.resubscribe()) + .await?; initial_load( &conn, @@ -1881,6 +1879,16 @@ async fn process_notify_event( tracing::error!("Error loading dependency job tag: {e:#}"); } } + DEV_WORKSPACE_ISOLATION_SETTING => { + if let Err(e) = load_dev_workspace_isolation(db).await { + tracing::error!("Error loading dev workspace isolation: {e:#}"); + } + } + DEV_WORKSPACE_TAG_SUFFIX_SETTING => match load_dev_workspace_tag_suffix(db).await { + Ok(true) if worker_mode => reload_worker_config(db, tx.clone(), true).await, + Ok(_) => (), + Err(e) => tracing::error!("Error loading dev workspace tag suffix: {e:#}"), + }, WORKSPACE_FAIRNESS_ENABLED_SETTING => { if let Err(e) = load_workspace_fairness_enabled(db).await { tracing::error!("Error loading workspace fairness enabled: {e:#}"); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 4fae6f34ab..03728c1d7b 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -60,7 +60,8 @@ use windmill_common::{ CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, - DEFAULT_TAGS_WORKSPACES_SETTING, DEPENDENCY_JOB_TAG_SETTING, DISABLE_PASSWORD_LOGIN, + DEFAULT_TAGS_WORKSPACES_SETTING, DEPENDENCY_JOB_TAG_SETTING, + DEV_WORKSPACE_ISOLATION_SETTING, DEV_WORKSPACE_TAG_SUFFIX_SETTING, DISABLE_PASSWORD_LOGIN, DISABLE_PASSWORD_LOGIN_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING, HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING, @@ -89,11 +90,13 @@ use windmill_common::{ users::truncate_token, utils::{empty_as_none, now_from_db, report_critical_error, Mode, HUB_API_SECRET}, worker::{ - load_env_vars, load_init_bash_from_env, load_periodic_bash_script_from_env, - load_periodic_bash_script_interval_from_env, load_whitelist_env_vars_from_env, - load_worker_config, store_pull_query, store_suspended_pull_query, Connection, WorkerConfig, - CLOUD_HOSTED, CONCURRENCY_KEY_MAX_QUEUED, CONCURRENCY_KEY_MAX_QUEUED_DEFAULT, - DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, DEPENDENCY_JOB_TAG, + is_valid_dev_workspace_tag_suffix, load_env_vars, load_init_bash_from_env, + load_periodic_bash_script_from_env, load_periodic_bash_script_interval_from_env, + load_whitelist_env_vars_from_env, load_worker_config, store_pull_query, + store_suspended_pull_query, Connection, WorkerConfig, CLOUD_HOSTED, + CONCURRENCY_KEY_MAX_QUEUED, CONCURRENCY_KEY_MAX_QUEUED_DEFAULT, + DEFAULT_DEV_WORKSPACE_TAG_SUFFIX, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, + DEPENDENCY_JOB_TAG, DEV_WORKSPACE_ISOLATION, DEV_WORKSPACE_TAG_SUFFIX, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, INDEXER_CONFIG, PREVIEW_TAGS_OVERRIDE, SMTP_CONFIG, WINDMILL_DIR, WORKER_CONFIG, WORKER_GROUP, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED, WORKSPACE_FAIRNESS_MAX_PERCENT, WORKSPACE_FAIRNESS_MIN_TOTAL, @@ -321,6 +324,13 @@ pub async fn initial_load( pass.setting(DEPENDENCY_JOB_TAG_SETTING, false, |v| async move { apply_dependency_job_tag(v) }); + // Ahead of the worker config reload below, which expands the tags with it. + pass.setting(DEV_WORKSPACE_TAG_SUFFIX_SETTING, false, |v| async move { + apply_dev_workspace_tag_suffix(v); + }); + pass.setting(DEV_WORKSPACE_ISOLATION_SETTING, false, |v| async move { + apply_dev_workspace_isolation(v) + }); // Load per-workspace retention overrides before the first cleanup tick so a fresh server // never sweeps globally without honoring configured longer-retention workspaces. @@ -1127,6 +1137,39 @@ pub fn apply_fork_workspace_tag_append_fork_suffix(value: Option error::Result { + let v = load_value_from_global_settings(db, DEV_WORKSPACE_TAG_SUFFIX_SETTING).await?; + Ok(apply_dev_workspace_tag_suffix(v)) +} + +pub fn apply_dev_workspace_tag_suffix(value: Option) -> bool { + let suffix = match value { + Some(serde_json::Value::String(t)) if !t.trim().is_empty() => t.trim().to_string(), + _ => DEFAULT_DEV_WORKSPACE_TAG_SUFFIX.to_string(), + }; + if !is_valid_dev_workspace_tag_suffix(&suffix) { + tracing::error!("Ignoring invalid dev workspace tag suffix {suffix:?}"); + return false; + } + let prev = DEV_WORKSPACE_TAG_SUFFIX.swap(std::sync::Arc::new(suffix.clone())); + *prev != suffix +} + +pub async fn load_dev_workspace_isolation(db: &DB) -> error::Result<()> { + let v = load_value_from_global_settings(db, DEV_WORKSPACE_ISOLATION_SETTING).await?; + apply_dev_workspace_isolation(v); + Ok(()) +} + +pub fn apply_dev_workspace_isolation(value: Option) { + DEV_WORKSPACE_ISOLATION.store( + matches!(value, Some(serde_json::Value::Bool(true))), + Ordering::Relaxed, + ); +} + pub async fn reload_critical_alert_mute_ui_setting(conn: &Connection) -> error::Result<()> { let v = load_value_from_global_settings_with_conn(conn, CRITICAL_ALERT_MUTE_UI_SETTING, true) .await?; diff --git a/backend/tests/dev_workspace_tag_suffix.rs b/backend/tests/dev_workspace_tag_suffix.rs new file mode 100644 index 0000000000..0969a0e45b --- /dev/null +++ b/backend/tests/dev_workspace_tag_suffix.rs @@ -0,0 +1,177 @@ +//! Sets the process-wide `DEV_WORKSPACE_ISOLATION`, so it lives in its own test binary: any other +//! test pushing a job in the same process would land on a suffixed tag. + +use std::collections::HashMap; + +use sqlx::{Pool, Postgres}; +use windmill_common::{ + jobs::{JobPayload, RawCode}, + runnable_settings::{ConcurrencySettings, DebouncingSettings}, + scripts::{ScriptHash, ScriptLang}, + worker::DEV_WORKSPACE_ISOLATION, +}; +use windmill_queue::{PushArgs, PushIsolationLevel}; + +async fn pushed_tag( + db: &Pool, + workspace_id: &str, + payload: JobPayload, + tag: Option<&str>, +) -> String { + try_pushed_tag(db, workspace_id, payload, tag) + .await + .expect("push must succeed") +} + +async fn try_pushed_tag( + db: &Pool, + workspace_id: &str, + payload: JobPayload, + tag: Option<&str>, +) -> Result { + let args = HashMap::new(); + let (id, tx) = windmill_queue::push( + db, + PushIsolationLevel::IsolatedRoot(db.clone()), + workspace_id, + payload, + PushArgs::from(&args), + "test-user", + "test@windmill.dev", + "u/test-user".to_string(), + None, + None, + None, + None, + None, + None, + None, + None, + false, + false, + None, + true, + tag.map(str::to_string), + None, + None, + None, + None, + false, + None, + None, + None, + None, + ) + .await?; + tx.commit().await.unwrap(); + Ok(sqlx::query_scalar("SELECT tag FROM v2_job WHERE id = $1") + .bind(id) + .fetch_one(db) + .await + .unwrap()) +} + +fn preview() -> JobPayload { + JobPayload::Code(RawCode { + hash: None, + content: "echo hi".to_string(), + path: None, + language: ScriptLang::Bash, + lock: None, + concurrency_settings: ConcurrencySettings::default().into(), + debouncing_settings: DebouncingSettings::default(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + modules: None, + tag: None, + }) +} + +#[sqlx::test(fixtures("base"))] +async fn dev_workspace_and_fork_jobs_run_on_suffixed_tags(db: Pool) { + DEV_WORKSPACE_ISOLATION.store(true, std::sync::atomic::Ordering::Relaxed); + + // A dev workspace, a fork of it, and a fork whose parent was deleted (`ON DELETE SET NULL`). + sqlx::query( + "INSERT INTO workspace (id, name, owner, parent_workspace_id, is_dev_workspace) VALUES + ('staging', 'staging', 'test-user', 'test-workspace', true), + ('wm-fork-a', 'fork', 'test-user', 'staging', false), + ('wm-fork-orphan', 'orphan', 'test-user', NULL, false)", + ) + .execute(&db) + .await + .unwrap(); + sqlx::query( + "INSERT INTO workspace_settings (workspace_id) + VALUES ('staging'), ('wm-fork-a'), ('wm-fork-orphan')", + ) + .execute(&db) + .await + .unwrap(); + + for (workspace, tag, expected) in [ + ("test-workspace", None, "bash"), + ("test-workspace", Some("gpu"), "gpu"), + ("staging", None, "bash-dev"), + ("staging", Some("gpu"), "gpu-dev"), + // A tag inherited from a dev job (a flow step, an agent's tool) is not suffixed twice. + ("staging", Some("gpu-dev"), "gpu-dev"), + ("wm-fork-a", None, "bash-dev"), + ("wm-fork-orphan", None, "bash-dev"), + ] { + assert_eq!( + pushed_tag(&db, workspace, preview(), tag).await, + expected, + "{workspace} with tag {tag:?}" + ); + } + + // A dedicated worker tag already names the workspace, and its worker is placed by hand. + let dedicated = JobPayload::ScriptHash { + hash: ScriptHash(123412), + path: "f/system/hello".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: Some(true), + language: ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: ConcurrencySettings::default(), + debouncing_settings: DebouncingSettings::default(), + labels: None, + job_token_scopes: None, + }; + assert_eq!( + pushed_tag(&db, "staging", dedicated, None).await, + "staging:f/system/hello" + ); + + // A hashed dedicated tag keeps only a prefix of its workspace id: a dev workspace sharing that + // prefix does not get to use another workspace's dedicated worker. + let root = "a".repeat(40); + let dev = format!("{root}-dev"); + sqlx::query( + "INSERT INTO workspace (id, name, owner, parent_workspace_id, is_dev_workspace) + VALUES ($1, 'root', 'test-user', NULL, false), ($2, 'dev', 'test-user', $1, true)", + ) + .bind(&root) + .bind(&dev) + .execute(&db) + .await + .unwrap(); + sqlx::query("INSERT INTO workspace_settings (workspace_id) VALUES ($1), ($2)") + .bind(&root) + .bind(&dev) + .execute(&db) + .await + .unwrap(); + let root_dedicated = + windmill_common::worker::dedicated_worker_tag(&root, "f/some/long/script/path"); + assert!( + try_pushed_tag(&db, &dev, preview(), Some(&root_dedicated)) + .await + .is_err(), + "{root_dedicated} must not stay unsuffixed, and suffixed it no longer fits a job tag" + ); +} diff --git a/backend/tests/instance_config.rs b/backend/tests/instance_config.rs index 24937a3ce5..18a7466637 100644 --- a/backend/tests/instance_config.rs +++ b/backend/tests/instance_config.rs @@ -1533,6 +1533,32 @@ async fn declarative_sync_rejects_an_unusable_instance_banner(db: Pool ); } +#[sqlx::test(fixtures("base"))] +async fn declarative_sync_rejects_an_unusable_dev_workspace_tag_suffix(db: Pool) { + clear_settings_and_configs(&db).await; + let before = count_global_settings(&db).await; + + let mut desired = BTreeMap::new(); + desired.insert( + "dev_workspace_tag_suffix".to_string(), + serde_json::json!("-dev' OR '1'='1"), + ); + + let err = windmill_common::instance_config::sync_global_settings_declarative( + &db, + &BTreeMap::new(), + &desired, + &(), + ) + .await + .expect_err("a suffix the workers would ignore must fail the sync"); + assert!( + err.to_string().contains("dev_workspace_tag_suffix"), + "the error should name the offending setting, got: {err}" + ); + assert_eq!(count_global_settings(&db).await, before); +} + #[sqlx::test(fixtures("base"))] async fn declarative_sync_rejects_a_malformed_max_token_expiration(db: Pool) { clear_settings_and_configs(&db).await; diff --git a/backend/windmill-api-auth/src/auth.rs b/backend/windmill-api-auth/src/auth.rs index 720111e89d..cb799bec25 100644 --- a/backend/windmill-api-auth/src/auth.rs +++ b/backend/windmill-api-auth/src/auth.rs @@ -6,6 +6,7 @@ use axum::{ }; use chrono::TimeZone; use http::{request::Parts, StatusCode}; +use itertools::Itertools; use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; use sqlx::FromRow; @@ -1022,6 +1023,53 @@ where } } +/// Adds the suffixed form of each tag of an `if_jobs:filter_tags:` scope: in a dev workspace a +/// job allowed on `gpu` is queued on `gpu-dev`, and the token must still reach the jobs it starts. +fn expand_filter_tags_with_dev_suffix(scopes: &mut [String], suffix: &str) { + for scope in scopes.iter_mut() { + if let Some(tags) = scope.strip_prefix("if_jobs:filter_tags:") { + let expanded = tags + .split(',') + .flat_map(|t| { + let suffixed = windmill_common::worker::with_dev_workspace_suffix(t, suffix); + [t.to_string(), suffixed] + }) + .unique() + .join(","); + *scope = format!("if_jobs:filter_tags:{expanded}"); + } + } +} + +/// Only while isolation is on and for a dev workspace or a fork, whose jobs get the suffix then: +/// otherwise the suffixed tags are other tags the token was not given. +async fn expand_filter_tags_for_dev_workspace( + db: &DB, + workspace_id: &str, + scopes: Option<&mut Vec>, +) { + let Some(scopes) = scopes else { return }; + if !windmill_common::worker::DEV_WORKSPACE_ISOLATION.load(std::sync::atomic::Ordering::Relaxed) + || !scopes.iter().any(|s| s.starts_with("if_jobs:filter_tags:")) + { + return; + } + let is_dev = workspace_id.starts_with(windmill_common::workspaces::WM_FORK_PREFIX) + || sqlx::query_scalar::<_, bool>( + "SELECT parent_workspace_id IS NOT NULL FROM workspace WHERE id = $1", + ) + .bind(workspace_id) + .fetch_optional(db) + .await + .ok() + .flatten() + .unwrap_or(false); + if is_dev { + let suffix = windmill_common::worker::DEV_WORKSPACE_TAG_SUFFIX.load(); + expand_filter_tags_with_dev_suffix(scopes, &suffix); + } +} + pub fn transform_old_scope_to_new_scope(scopes: Option<&mut Vec>) { if let Some(scopes) = scopes { for scope in scopes.iter_mut() { @@ -1172,6 +1220,14 @@ pub async fn resolve_opt_job_authed( let authed = &mut opt_job_authed.authed; if authed.scopes.is_some() && !own_job_runtime_route { transform_old_scope_to_new_scope(authed.scopes.as_mut()); + if let Some(w_id) = workspace_id.as_deref() { + expand_filter_tags_for_dev_workspace( + &cache.db, + w_id, + authed.scopes.as_mut(), + ) + .await; + } if let Err(err) = crate::scopes::check_scopes_for_route( authed.scopes.as_deref(), @@ -1359,6 +1415,19 @@ pub async fn list_tokens_internal( mod tests { use super::*; + #[test] + fn filter_tags_reach_their_dev_workspace_form() { + let mut scopes = vec![ + "jobs:read".to_string(), + "if_jobs:filter_tags:gpu,bash-dev".to_string(), + ]; + expand_filter_tags_with_dev_suffix(&mut scopes, "-dev"); + assert_eq!( + scopes, + vec!["jobs:read", "if_jobs:filter_tags:gpu,gpu-dev,bash-dev"] + ); + } + /// Dropping the `AUTH_CACHE` entry between the two resolutions is what makes a memo /// hit observable: only a working memo can answer the second one. #[tokio::test] diff --git a/backend/windmill-api-configs/src/lib.rs b/backend/windmill-api-configs/src/lib.rs index f1f4792e71..890a561679 100644 --- a/backend/windmill-api-configs/src/lib.rs +++ b/backend/windmill-api-configs/src/lib.rs @@ -273,18 +273,22 @@ async fn update_config( } let mut tx = db.begin().await?; - let was_paused = if name.starts_with("worker__") { + let previous = if name.starts_with("worker__") { sqlx::query_scalar!( - "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE", + "SELECT config FROM config WHERE name = $1 FOR UPDATE", &name ) .fetch_optional(&mut *tx) .await? .flatten() - .unwrap_or(false) } else { - false + None }; + let is_set = |c: Option<&serde_json::Value>, k: &str| { + c.and_then(|c| c.get(k)).and_then(|v| v.as_bool()) == Some(true) + }; + let was_paused = is_set(previous.as_ref(), "paused"); + let was_serving_dev = is_set(previous.as_ref(), "serves_dev_workspaces"); sqlx::query!( "INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config", &name, @@ -306,7 +310,7 @@ async fn update_config( tx.commit().await?; if name.starts_with("worker__") { - let paused = config.get("paused").and_then(|v| v.as_bool()) == Some(true); + let paused = is_set(Some(&config), "paused"); if paused != was_paused { windmill_common::feature_usage::log_feature_usage( "worker_group", @@ -314,6 +318,14 @@ async fn update_config( if paused { "pause" } else { "resume" }, ); } + let serving_dev = is_set(Some(&config), "serves_dev_workspaces"); + if serving_dev != was_serving_dev { + windmill_common::feature_usage::log_feature_usage( + "worker_group", + "dev_workspaces_toggle", + if serving_dev { "serve" } else { "stop" }, + ); + } } Ok(format!("Updated config {name}")) } diff --git a/backend/windmill-api-jobs/src/jobs_export.rs b/backend/windmill-api-jobs/src/jobs_export.rs index b2b69367bc..576d33bad5 100644 --- a/backend/windmill-api-jobs/src/jobs_export.rs +++ b/backend/windmill-api-jobs/src/jobs_export.rs @@ -434,6 +434,7 @@ pub async fn import_completed_jobs( pub async fn import_queued_jobs( authed: ApiAuthed, + Extension(db): Extension, Extension(user_db): Extension, Path(w_id): Path, Json(jobs): Json>, @@ -468,7 +469,9 @@ pub async fn import_queued_jobs( let mut tx = user_db.begin(&authed).await?; - for job in jobs { + for mut job in jobs { + // A fork's admin is whoever created it, so an imported tag is as untrusted as a pushed one. + job.tag = windmill_queue::tags::apply_dev_workspace_suffix(job.tag, &w_id, &db).await?; sqlx::query!( r#" INSERT INTO v2_job ( diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index f9ba1ff86f..d6803dc432 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -916,6 +916,7 @@ pub async fn set_global_setting_internal( run_setting_pre_write_hook(db, &key, &value).await?; + let turns_on = value.as_bool() == Some(true); match value { serde_json::Value::Null => { if instance_config::PROTECTED_SETTINGS.contains(&key.as_str()) { @@ -949,6 +950,14 @@ pub async fn set_global_setting_internal( } }; + if key == windmill_common::global_settings::DEV_WORKSPACE_ISOLATION_SETTING { + windmill_common::feature_usage::log_feature_usage( + "dev_workspace_isolation", + "toggle", + if turns_on { "on" } else { "off" }, + ); + } + if should_bump_instance_ai_revision { bump_instance_ai_config_revision(); } @@ -975,6 +984,14 @@ async fn run_setting_pre_write_hook( value: &serde_json::Value, ) -> error::Result<()> { match key { + windmill_common::global_settings::DEV_WORKSPACE_TAG_SUFFIX_SETTING => match value { + serde_json::Value::Null => {} + serde_json::Value::String(s) if s.trim().is_empty() => {} + suffix => windmill_common::global_settings::validate_dev_workspace_tag_suffix(suffix) + .map_err(|e| { + error::Error::BadRequest(format!("The dev workspace tag suffix {e}")) + })?, + }, // The instance AI config is written as an untyped blob through this generic // endpoint, so it never passes the typed check the workspace handler applies. // Rates that reach a cost total unbounded would make it negative or infinite. diff --git a/backend/windmill-api-workers/Cargo.toml b/backend/windmill-api-workers/Cargo.toml index 8e54d84e77..24f2bc5649 100644 --- a/backend/windmill-api-workers/Cargo.toml +++ b/backend/windmill-api-workers/Cargo.toml @@ -11,6 +11,7 @@ path = "src/lib.rs" [dependencies] windmill-api-auth.workspace = true windmill-common = { workspace = true, default-features = false } +windmill-queue.workspace = true axum.workspace = true chrono.workspace = true serde.workspace = true diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index 8ebee2a967..a7f488a7dd 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -21,7 +21,10 @@ use windmill_common::{ jobs::{HIDE_WORKERS_FOR_NON_ADMINS, TAGS_ARE_SENSITIVE}, queue_metrics::{read_queue_metrics_series, QueueMetricsSeries}, utils::{paginate, Pagination}, - worker::{ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE}, + worker::{ + ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE, + DEV_WORKSPACE_ISOLATION, DEV_WORKSPACE_TAG_SUFFIX, + }, workspaces::workspace_with_fork_ancestors, DB, }; @@ -37,6 +40,7 @@ pub fn global_service() -> Router { "/is_default_tags_per_workspace", get(get_default_tags_per_workspace), ) + .route("/dev_workspace_isolation", get(get_dev_workspace_isolation)) .route("/get_default_tags", get(get_default_tags)) .route("/queue_metrics", get(get_queue_metrics)) .route("/queue_metrics_series", get(get_queue_metrics_series)) @@ -186,7 +190,18 @@ async fn exists_workers_with_tags( let chain = workspace_with_fork_ancestors(&db, workspace).await?; let custom_tags = CUSTOM_TAGS_PER_WORKSPACE.load(); let allowed_tags = custom_tags.to_string_vec(Some(&chain)); - tags.retain(|t| allowed_tags.contains(t)); + // A dev workspace's jobs run on the suffixed form of the tags it may use. + let dev_suffix = windmill_queue::tags::dev_workspace_suffix_for(workspace, &db) + .await + .ok() + .flatten(); + tags.retain(|t| { + allowed_tags.contains(t) + || dev_suffix.as_deref().is_some_and(|s| { + t.strip_suffix(s) + .is_some_and(|b| allowed_tags.iter().any(|a| a == b)) + }) + }); } else { // No workspace provided and not superadmin - return empty return Ok(Json(std::collections::HashMap::new())); @@ -259,6 +274,19 @@ async fn get_default_tags_per_workspace() -> JsonResult { )) } +#[derive(Serialize)] +struct DevWorkspaceIsolation { + enabled: bool, + suffix: String, +} + +async fn get_dev_workspace_isolation() -> JsonResult { + Ok(Json(DevWorkspaceIsolation { + enabled: DEV_WORKSPACE_ISOLATION.load(std::sync::atomic::Ordering::Relaxed), + suffix: DEV_WORKSPACE_TAG_SUFFIX.load().to_string(), + })) +} + async fn get_default_tags() -> JsonResult> { Ok(Json(DEFAULT_TAGS.clone())) } diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 67d31f8a77..dc40c72d13 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -868,11 +868,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. diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 943d05dc11..1c7619e1f8 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11413,6 +11413,30 @@ paths: schema: type: boolean + /workers/dev_workspace_isolation: + get: + summary: get whether the jobs of dev workspaces and forks run on their own worker groups + operationId: getDevWorkspaceIsolation + tags: + - worker + responses: + "200": + description: dev workspace isolation + content: + application/json: + schema: + type: object + properties: + enabled: + type: boolean + description: whether new jobs of dev workspaces and forks are queued on suffixed tags + suffix: + type: string + description: the suffix of those tags, which worker groups serving dev workspaces listen with + required: + - enabled + - suffix + /w/{workspace}/workspace_dependencies/create: post: summary: create workspace dependencies diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index c567a593d9..a7e2915571 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -9264,6 +9264,7 @@ async fn run_inline_preview_script( return Err(operator_preview_refusal(job_id)); } check_scopes(&authed, || format!("jobs:run"))?; + windmill_queue::tags::check_inline_run_allowed(&w_id, &db).await?; if let Some(job_id) = job_id { register_potential_assets_on_inline_execution(job_id, &w_id, &preview); } @@ -9383,6 +9384,7 @@ async fn run_inline_script_inner( args: Option>>, user_db: Option, ) -> error::Result { + windmill_queue::tags::check_inline_run_allowed(&w_id, &db).await?; // An inline run executes with the caller's own token, so it cannot honour a script's // `job_token_scopes`: such a script only runs as a job. A path is resolved here, once, and // the version checked is the version run. @@ -10183,17 +10185,22 @@ async fn add_batch_jobs( let language = language.unwrap_or(ScriptLang::Deno); - let tag = if let Some(dedicated_worker) = dedicated_worker { + let tag = match (dedicated_worker, batch_info.tag) { // Same rule as `push`: a dedicated worker would run it with its own unscoped token. - if dedicated_worker && path.is_some() && job_token_scopes.is_none() { + (Some(true), _) if path.is_some() && job_token_scopes.is_none() => { windmill_common::worker::dedicated_worker_tag(&w_id, &path.clone().unwrap()) - } else { - format!("{}", language.as_str()) } - } else if let Some(tag) = batch_info.tag { - tag - } else { - format!("{}", language.as_str()) + (None, Some(tag)) => { + windmill_queue::tags::apply_dev_workspace_suffix(tag, &w_id, &db).await? + } + _ => { + windmill_queue::tags::apply_dev_workspace_suffix( + language.as_str().to_string(), + &w_id, + &db, + ) + .await? + } }; let mut tx = user_db.begin(&authed).await?; diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 2bc303f5c2..9472655ce5 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -10,6 +10,8 @@ pub const FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING: &str = "fork_workspace_tag_append_fork_suffix"; pub const PREVIEW_TAGS_OVERRIDE_SETTING: &str = "preview_tags_override"; pub const DEPENDENCY_JOB_TAG_SETTING: &str = "dependency_job_tag"; +pub const DEV_WORKSPACE_ISOLATION_SETTING: &str = "dev_workspace_isolation"; +pub const DEV_WORKSPACE_TAG_SUFFIX_SETTING: &str = "dev_workspace_tag_suffix"; pub const BASE_URL_SETTING: &str = "base_url"; pub const WS_BASE_URL_SETTING: &str = "ws_base_url"; pub const OAUTH_SETTING: &str = "oauths"; @@ -315,6 +317,21 @@ pub fn validate_accent_color(value: &serde_json::Value) -> Result<(), String> { Ok(()) } +/// Validate a non-empty [`DEV_WORKSPACE_TAG_SUFFIX_SETTING`] value. Every writer must call +/// it: a suffix that fails it is ignored at load, so the workers would listen on a suffix +/// the setting does not show. +pub fn validate_dev_workspace_tag_suffix(value: &serde_json::Value) -> Result<(), String> { + let s = value + .as_str() + .ok_or_else(|| "must be a string".to_string())?; + if !crate::worker::is_valid_dev_workspace_tag_suffix(s.trim()) { + return Err( + "may only contain letters, digits, '-', '_' and '.', up to 32 characters".to_string(), + ); + } + Ok(()) +} + /// The settings every signed-in browser reads on each full page load. #[derive(serde::Serialize, Clone, Debug, Default, PartialEq)] pub struct InstanceUi { diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index d19de41dcc..110384a0d6 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -263,6 +263,8 @@ pub struct GlobalSettings { #[serde(skip_serializing_if = "Option::is_none")] pub default_tags_per_workspace: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub dev_workspace_isolation: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub preview_tags_override: Option, #[serde(skip_serializing_if = "Option::is_none")] pub disable_hub: Option, @@ -279,6 +281,8 @@ pub struct GlobalSettings { #[serde(skip_serializing_if = "Option::is_none")] pub dependency_job_tag: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub dev_workspace_tag_suffix: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub ws_base_url: Option, #[serde(skip_serializing_if = "Option::is_none")] pub github_app_webhook_base_url: Option, @@ -970,6 +974,9 @@ pub struct WorkerGroupConfig { /// Workers of the group pull no new job while this is set; queued jobs wait. #[serde(skip_serializing_if = "Option::is_none")] pub paused: Option, + /// The group listens to its tags suffixed with the dev workspace tag suffix, and only to them. + #[serde(skip_serializing_if = "Option::is_none")] + pub serves_dev_workspaces: Option, /// Object store this group's dependency cache uses instead of the instance one. Same shape /// as the instance `object_store_cache_config`. #[serde(skip_serializing_if = "Option::is_none")] @@ -1499,6 +1506,14 @@ pub async fn sync_global_settings_declarative( .map_err(|e| anyhow::anyhow!("{accent_key}: {e}"))?, } + let suffix_key = crate::global_settings::DEV_WORKSPACE_TAG_SUFFIX_SETTING; + match desired.get(suffix_key) { + None | Some(serde_json::Value::Null) => {} + Some(serde_json::Value::String(s)) if s.trim().is_empty() => {} + Some(suffix) => crate::global_settings::validate_dev_workspace_tag_suffix(suffix) + .map_err(|e| anyhow::anyhow!("{suffix_key}: {e}"))?, + } + // An origin list that cannot be parsed is dropped at boot, leaving the // empty default — which is no restriction at all. Rejecting it here is what // keeps a typo in a ConfigMap from silently widening CORS instance-wide. diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index f5743782e3..2257327867 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -451,6 +451,13 @@ lazy_static::lazy_static! { pub static ref FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX: AtomicBool = AtomicBool::new(false); pub static ref PREVIEW_TAGS_OVERRIDE: AtomicBool = AtomicBool::new(false); pub static ref DEPENDENCY_JOB_TAG: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(None); + /// Whether the jobs of dev workspaces and forks are queued on suffixed tags. + pub static ref DEV_WORKSPACE_ISOLATION: AtomicBool = AtomicBool::new(false); + /// The suffix of those tags. Groups serving dev workspaces listen with it whether isolation is + /// on or not, so turning it off leaves no job already queued on a suffixed tag unserved. + pub static ref DEV_WORKSPACE_TAG_SUFFIX: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(DEFAULT_DEV_WORKSPACE_TAG_SUFFIX.to_string()); + /// Whether this process's worker group serves dev workspaces, set when its config is loaded. + pub static ref WORKER_SERVES_DEV_WORKSPACES: AtomicBool = AtomicBool::new(false); pub static ref MAX_TIMEOUT: u64 = std::env::var("TIMEOUT") .ok() @@ -2418,9 +2425,45 @@ pub async fn update_worker_ping_main_loop_query( // occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus), // memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11 WHERE worker = $6", -const MAX_TAG_LEN: usize = 50; +/// The length of `v2_job.tag`. +pub const MAX_TAG_LEN: usize = 50; const HASH_SUFFIX_LEN: usize = 16; +pub const DEFAULT_DEV_WORKSPACE_TAG_SUFFIX: &str = "-dev"; + +/// The suffix new jobs of dev workspaces and forks get, `None` while isolation is off. +pub fn dev_workspace_tag_suffix() -> Option { + DEV_WORKSPACE_ISOLATION + .load(std::sync::atomic::Ordering::Relaxed) + .then(|| DEV_WORKSPACE_TAG_SUFFIX.load().to_string()) +} + +/// Worker tags are pasted into the pull queries as SQL literals and listed comma-separated, so +/// the suffix must not carry a quote, a comma or whitespace. +pub fn is_valid_dev_workspace_tag_suffix(suffix: &str) -> bool { + !suffix.is_empty() + && suffix.len() <= 32 + && suffix + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) +} + +/// `tag` as the workers serving dev workspaces listen to it. Idempotent, because a tag inherited +/// from a dev job (a flow step, an agent's tool) already carries the suffix. +pub fn with_dev_workspace_suffix(tag: &str, suffix: &str) -> String { + if tag.ends_with(suffix) { + tag.to_string() + } else { + format!("{tag}{suffix}") + } +} + +/// Whether `tag` has the shape of a [`dedicated_worker_tag`] too long to keep whole. Its prefix +/// alone does not tell whose it is: two workspaces may share it. +pub fn is_hashed_dedicated_worker_tag(tag: &str) -> bool { + tag.len() == MAX_TAG_LEN && tag.as_bytes().get(MAX_TAG_LEN - 1 - HASH_SUFFIX_LEN) == Some(&b'#') +} + pub fn dedicated_worker_tag(workspace_id: &str, path: &str) -> String { let full_tag = format!("{}:{}", workspace_id, path); if full_tag.len() <= MAX_TAG_LEN { @@ -2581,8 +2624,32 @@ pub async fn load_worker_config( }) .unwrap_or_else(|| DEFAULT_TAGS.clone()); + // A group serving dev workspaces listens to the suffixed form of its tags only, so it never + // takes a root workspace's job. Dedicated workers are placed by hand and keep their tags. + let dev_suffix = (config.serves_dev_workspaces == Some(true) + && dedicated_worker.is_none() + && dedicated_workers.is_none()) + .then(|| DEV_WORKSPACE_TAG_SUFFIX.load().to_string()); + WORKER_SERVES_DEV_WORKSPACES.store(dev_suffix.is_some(), std::sync::atomic::Ordering::Relaxed); + let worker_tags = match &dev_suffix { + Some(suffix) => worker_tags + .iter() + .map(|t| with_dev_workspace_suffix(t, suffix)) + .unique() + .collect_vec(), + None => worker_tags, + }; + let mut priority_tags_sorted: Vec = Vec::new(); - let priority_tags_map = config.priority_tags.unwrap_or_else(HashMap::new); + let priority_tags_map: HashMap = match &dev_suffix { + Some(suffix) => config + .priority_tags + .unwrap_or_default() + .into_iter() + .map(|(t, p)| (with_dev_workspace_suffix(&t, suffix), p)) + .collect(), + None => config.priority_tags.unwrap_or_else(HashMap::new), + }; if priority_tags_map.len() > 0 { let mut all_tags_set: HashSet = HashSet::from_iter(worker_tags.clone()); @@ -2778,6 +2845,7 @@ pub struct WorkerConfigOpt { pub native_mode: Option, pub object_store_cache_config: Option, pub paused: Option, + pub serves_dev_workspaces: Option, } impl Default for WorkerConfigOpt { @@ -2798,6 +2866,7 @@ impl Default for WorkerConfigOpt { native_mode: Default::default(), object_store_cache_config: Default::default(), paused: Default::default(), + serves_dev_workspaces: Default::default(), } } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 2ed6d034b2..eafb344870 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -4778,6 +4778,9 @@ pub async fn pull( } } }; + let tag = + crate::tags::apply_dev_workspace_suffix(tag, &job.workspace_id, db) + .await?; sqlx::query!( "UPDATE v2_job_queue SET tag = $1, running = false WHERE id = $2", tag, @@ -7372,7 +7375,8 @@ async fn push_inner<'c, 'd>( // A dedicated worker runs every job it serves with its own unscoped worker token, so a // job with a restricted token runs on the regular workers of its language instead. - let tag = if dedicated_worker.is_some_and(|x| x) && job_token_scopes.is_none() { + let on_dedicated_worker = dedicated_worker.is_some_and(|x| x) && job_token_scopes.is_none(); + let tag = if on_dedicated_worker { let flow_prefix = if job_kind == JobKind::Flow || job_kind == JobKind::FlowDependencies { "flow/" } else { @@ -7453,6 +7457,11 @@ async fn push_inner<'c, 'd>( } }) }; + let tag = if on_dedicated_worker { + tag + } else { + crate::tags::apply_dev_workspace_suffix(tag, workspace_id, db).await? + }; let mut tx = tx.into_tx().await?; diff --git a/backend/windmill-queue/src/tags.rs b/backend/windmill-queue/src/tags.rs index 5f076bd3b2..59af8f9e8d 100644 --- a/backend/windmill-queue/src/tags.rs +++ b/backend/windmill-queue/src/tags.rs @@ -1,6 +1,10 @@ use sqlx::{Pool, Postgres}; +use windmill_common::error::Error; use windmill_common::worker::{ - DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, + dedicated_worker_tag, dev_workspace_tag_suffix, is_hashed_dedicated_worker_tag, + with_dev_workspace_suffix, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, + DEV_WORKSPACE_TAG_SUFFIX, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, MAX_TAG_LEN, + WORKER_SERVES_DEV_WORKSPACES, }; use windmill_common::workspaces::WM_FORK_PREFIX; @@ -13,6 +17,10 @@ struct TagWorkspace { /// Whether the `-fork` suffix may apply: the workspace shares an ancestor's id, or carries the /// generated prefix itself. is_fork: bool, + /// Whether the workspace's jobs run on the workers serving dev workspaces: it has a parent, + /// or carries the generated prefix (a fork orphaned by its parent's deletion stays a fork). + /// `None` when the lookup failed. + is_dev: Option, } lazy_static::lazy_static! { @@ -140,9 +148,153 @@ pub async fn per_workspace_tag(workspace_id: &str, db: &Pool) -> Optio }) } +/// `tag`, fully resolved, as a job of `workspace_id` is queued on it: suffixed with the dev +/// workspace tag suffix when isolation is on and the workspace is a dev workspace or a fork, so +/// only the worker groups serving dev workspaces pick it up. A tag no group serves leaves the job +/// queued; it never falls back to the unsuffixed tag. Dedicated worker tags are exempt: they +/// already name the workspace and an admin places each dedicated worker by hand. +pub async fn apply_dev_workspace_suffix( + tag: String, + workspace_id: &str, + db: &Pool, +) -> Result { + let Some(suffix) = dev_workspace_suffix_for(workspace_id, db).await? else { + return Ok(tag); + }; + if is_dedicated_worker_tag_of(workspace_id, &tag, db).await? { + return Ok(tag); + } + let suffixed = with_dev_workspace_suffix(&tag, &suffix); + if suffixed.len() > MAX_TAG_LEN { + return Err(Error::BadRequest(format!( + "tag {tag} of a dev workspace job would be {suffixed} once suffixed with {suffix}, \ + longer than the {MAX_TAG_LEN} characters a job tag holds" + ))); + } + Ok(suffixed) +} + +/// Whether `tag` is a dedicated worker tag of `workspace_id`. A hashed one keeps only a prefix of +/// the workspace id, which another workspace may share, so it counts only when a worker group's +/// dedicated config for this workspace produces it. +async fn is_dedicated_worker_tag_of( + workspace_id: &str, + tag: &str, + db: &Pool, +) -> Result { + if tag.starts_with(&format!("{workspace_id}:")) { + return Ok(true); + } + if !is_hashed_dedicated_worker_tag(tag) { + return Ok(false); + } + let configs = sqlx::query_scalar!("SELECT config FROM config WHERE name LIKE 'worker\\_\\_%'") + .fetch_all(db) + .await?; + Ok(configs.into_iter().flatten().any(|config| { + let single = config.get("dedicated_worker").cloned().into_iter(); + let many = config + .get("dedicated_workers") + .and_then(|v| v.as_array().cloned()) + .unwrap_or_default(); + single.chain(many).any(|entry| { + entry + .as_str() + .and_then(|e| e.split_once(':')) + .is_some_and(|(ws, path)| { + ws == workspace_id && dedicated_worker_tag(ws, path) == tag + }) + }) + })) +} + +/// Gives `new_id` the lineage of `old_id` before a rename commits: the rename re-pushes the +/// workspace's schedules in its own transaction, where the new row is not yet visible to the +/// lookup, and a job queued then would keep a tag resolved without its parent for good. +pub async fn carry_lineage_over_rename( + old_id: &str, + new_id: &str, + db: &Pool, +) -> Result<(), Error> { + let old = lookup_tag_workspace(old_id, db).await; + if old.is_dev.is_none() { + return Err(Error::internal_err(format!( + "could not resolve the lineage of {old_id}" + ))); + } + let new = TagWorkspace { + id: if old.id == old_id { + new_id.to_string() + } else { + old.id + }, + ..old + }; + FORK_PARENT_CACHE.insert(new_id.to_string(), (new, std::time::Instant::now())); + Ok(()) +} + +/// The suffix to strip from a flow job's tag to compare it with the generic flow tags: that of a +/// dev workspace's flow, isolation on or off, so a flow started before isolation was turned off +/// still routes its steps by their own tags. +pub async fn dev_flow_tag_suffix( + workspace_id: &str, + flow_tag: &str, + db: &Pool, +) -> Result, Error> { + let suffix = DEV_WORKSPACE_TAG_SUFFIX.load().to_string(); + if !flow_tag.ends_with(&suffix) { + return Ok(None); + } + match lookup_tag_workspace(workspace_id, db).await.is_dev { + Some(true) => Ok(Some(suffix)), + Some(false) => Ok(None), + None => Err(Error::internal_err(format!( + "could not resolve whether {workspace_id} is a dev workspace" + ))), + } +} + +/// The suffix the jobs of `workspace_id` are queued with: `Some` while isolation is on and the +/// workspace is a dev workspace or a fork. A lineage that cannot be read is an error rather than +/// a guess, as either guess queues a job on the wrong pool for good. +pub async fn dev_workspace_suffix_for( + workspace_id: &str, + db: &Pool, +) -> Result, Error> { + let Some(suffix) = dev_workspace_tag_suffix() else { + return Ok(None); + }; + match lookup_tag_workspace(workspace_id, db).await.is_dev { + Some(true) => Ok(Some(suffix)), + Some(false) => Ok(None), + None => Err(Error::internal_err(format!( + "could not resolve whether {workspace_id} is a dev workspace, which decides the \ + workers its jobs run on" + ))), + } +} + +/// Refuses running code of `workspace_id` in this process, as the inline routes do, when its +/// jobs are isolated on dev workers and this process's group does not serve dev workspaces. +pub async fn check_inline_run_allowed( + workspace_id: &str, + db: &Pool, +) -> Result<(), Error> { + if dev_workspace_suffix_for(workspace_id, db).await?.is_some() + && !WORKER_SERVES_DEV_WORKSPACES.load(std::sync::atomic::Ordering::Relaxed) + { + return Err(Error::BadRequest(format!( + "{workspace_id} is a dev workspace or a fork, whose code runs only on worker groups \ + serving dev workspaces, and this worker's group does not: run it as a job instead" + ))); + } + Ok(()) +} + /// Backed by a short-TTL cache to avoid a DB round-trip per job push. A transient DB error resolves /// to the workspace's own id for this call but is NOT cached, so the next push retries instead of -/// misrouting for the whole TTL. +/// misrouting for the whole TTL. Whether the workspace is dev is then unknown. async fn lookup_tag_workspace(workspace_id: &str, db: &Pool) -> TagWorkspace { if let Some((resolved, cached_at)) = FORK_PARENT_CACHE.get(workspace_id) { if cached_at.elapsed().as_secs() < FORK_PARENT_CACHE_TTL_SECS { @@ -154,7 +306,7 @@ async fn lookup_tag_workspace(workspace_id: &str, db: &Pool) -> TagWor // itself under a generated fork id. The depth bound mirrors the other chain walkers as a // cycle-safety backstop. A `wm-fork-` workspace orphaned by `ON DELETE SET NULL` has no parent // left to borrow from, so it ends the walk on itself. - let resolved = match sqlx::query_scalar!( + let resolved = match sqlx::query!( r#" WITH RECURSIVE chain AS ( SELECT id, parent_workspace_id, is_dev_workspace, 0 AS depth @@ -165,35 +317,47 @@ async fn lookup_tag_workspace(workspace_id: &str, db: &Pool) -> TagWor JOIN chain ON w.id = chain.parent_workspace_id WHERE chain.depth < 20 ) - SELECT id AS "id!" FROM chain - WHERE parent_workspace_id IS NULL - OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%') - ORDER BY depth LIMIT 1 + SELECT + (SELECT id FROM chain + WHERE parent_workspace_id IS NULL + OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%') + ORDER BY depth LIMIT 1) AS id, + (SELECT parent_workspace_id IS NOT NULL FROM chain WHERE depth = 0) AS has_parent "#, workspace_id ) - .fetch_optional(db) + .fetch_one(db) .await { - // No row means the workspace does not exist yet, or its chain is broken: both are "we - // cannot resolve this", so fall back to its own id for this call WITHOUT caching. Caching - // it would pin a fork to its own unserved id for the whole TTL — a rename resolves the new - // id before the row lands, and every job pushed until the TTL expires would queue. - Ok(None) => { - return TagWorkspace { - id: workspace_id.to_string(), - is_fork: workspace_id.starts_with(WM_FORK_PREFIX), + Ok(row) => match row.id { + Some(id) => TagWorkspace { + is_fork: id != workspace_id || workspace_id.starts_with(WM_FORK_PREFIX), + is_dev: Some( + row.has_parent == Some(true) || workspace_id.starts_with(WM_FORK_PREFIX), + ), + id, + }, + // No id means the workspace does not exist yet, or its chain is broken: both are "we + // cannot resolve this", so fall back to its own id for this call WITHOUT caching. + // Caching it would pin a fork to its own unserved id for the whole TTL — a rename + // resolves the new id before the row lands, and every job pushed until the TTL + // expires would queue. + None => { + return TagWorkspace { + id: workspace_id.to_string(), + is_fork: workspace_id.starts_with(WM_FORK_PREFIX), + is_dev: Some( + row.has_parent == Some(true) || workspace_id.starts_with(WM_FORK_PREFIX), + ), + } } - } - Ok(Some(id)) => { - let is_fork = id != workspace_id || workspace_id.starts_with(WM_FORK_PREFIX); - TagWorkspace { id, is_fork } - } + }, Err(e) => { tracing::warn!("failed to resolve tag workspace for {workspace_id}: {e:#}"); return TagWorkspace { id: workspace_id.to_string(), is_fork: workspace_id.starts_with(WM_FORK_PREFIX), + is_dev: None, }; } }; diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 1a478292cb..0bc7bf6777 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -577,6 +577,9 @@ async fn enqueue_windmill_tool( &ctx.job.workspace_id, ctx.preserve_step_tags, job_payload.tag.as_deref(), + windmill_queue::tags::dev_flow_tag_suffix(&ctx.job.workspace_id, &ctx.job.tag, ctx.db) + .await? + .as_deref(), ); if tool_tag.as_deref().is_some_and(tag_reads_flow_expr) { append_logs( diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index a257ca51fa..c188f8de58 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1631,7 +1631,17 @@ pub async fn update_flow_status_after_job_completion_internal( Some(t) => { let no_args = HashMap::new(); let args = PushArgs::from(fetched_args.as_ref().unwrap_or(&no_args)); - resolve_push_tag(t, &args, &flow_job.workspace_id, db).await + match resolve_push_tag(t, &args, &flow_job.workspace_id, db).await { + Some(t) => Some( + windmill_queue::tags::apply_dev_workspace_suffix( + t, + &flow_job.workspace_id, + db, + ) + .await?, + ), + None => None, + } } None => None, }; @@ -2210,7 +2220,8 @@ pub async fn update_flow_status_after_job_completion_internal( None }, has_triggered_error_handler: has_triggered_error_handler || is_failure_step, - recovered_by_child: failure_step_recovers || (recovered_by_child && !success), + recovered_by_child: failure_step_recovers + || (recovered_by_child && !success), }, )); } @@ -3284,8 +3295,17 @@ pub(crate) fn resolve_flow_step_tag( workspace_id: &str, preserve_step_tags: bool, step_tag: Option<&str>, + // The suffix of a dev workspace's flow tag, if any: its flows run on the generic tag suffixed, + // and their steps are suffixed when pushed. + dev_suffix: Option<&str>, ) -> Option { - if is_preprocessor_step || flow_tag == "flow" || flow_tag == format!("flow-{}", workspace_id) { + let generic_flow_tag = dev_suffix + .and_then(|s| flow_tag.strip_suffix(s)) + .unwrap_or(flow_tag); + if is_preprocessor_step + || generic_flow_tag == "flow" + || generic_flow_tag == format!("flow-{}", workspace_id) + { step_tag.map(str::to_string) } else if preserve_step_tags && step_tag.is_some_and(|t| !t.is_empty()) { step_tag.map(str::to_string) @@ -3363,7 +3383,14 @@ mod tag_resolution_tests { // Parent flow on a custom tag, step declares its own tag, preserve disabled: // the step inherits the flow tag (historical behavior). assert_eq!( - resolve_flow_step_tag(false, "worker-group-A", "w1", false, Some("worker-group-B")), + resolve_flow_step_tag( + false, + "worker-group-A", + "w1", + false, + Some("worker-group-B"), + None + ), Some("worker-group-A".to_string()) ); } @@ -3373,7 +3400,14 @@ mod tag_resolution_tests { // The exact customer scenario: a sub-flow tagged worker-group-B run as a step of a // flow tagged worker-group-A now runs on worker-group-B when preserve_step_tags is on. assert_eq!( - resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("worker-group-B")), + resolve_flow_step_tag( + false, + "worker-group-A", + "w1", + true, + Some("worker-group-B"), + None + ), Some("worker-group-B".to_string()) ); } @@ -3381,12 +3415,12 @@ mod tag_resolution_tests { #[test] fn untagged_step_inherits_flow_tag_even_when_preserve_enabled() { assert_eq!( - resolve_flow_step_tag(false, "worker-group-A", "w1", true, None), + resolve_flow_step_tag(false, "worker-group-A", "w1", true, None, None), Some("worker-group-A".to_string()) ); // An empty tag counts as "no tag" and still inherits. assert_eq!( - resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("")), + resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some(""), None), Some("worker-group-A".to_string()) ); } @@ -3395,16 +3429,38 @@ mod tag_resolution_tests { fn generic_flow_tag_always_uses_step_tag() { for flow_tag in ["flow", "flow-w1"] { assert_eq!( - resolve_flow_step_tag(false, flow_tag, "w1", false, Some("worker-group-B")), + resolve_flow_step_tag(false, flow_tag, "w1", false, Some("worker-group-B"), None), Some("worker-group-B".to_string()) ); } } + #[test] + fn dev_workspace_flow_on_suffixed_generic_tag_uses_step_tag() { + for flow_tag in ["flow-dev", "flow-w1-dev"] { + assert_eq!( + resolve_flow_step_tag(false, flow_tag, "w1", false, Some("gpu"), Some("-dev")), + Some("gpu".to_string()) + ); + } + // In a root workspace `flow-dev` is a custom tag the steps inherit. + assert_eq!( + resolve_flow_step_tag(false, "flow-dev", "w1", false, Some("gpu"), None), + Some("flow-dev".to_string()) + ); + } + #[test] fn preprocessor_step_uses_step_tag() { assert_eq!( - resolve_flow_step_tag(true, "worker-group-A", "w1", false, Some("worker-group-B")), + resolve_flow_step_tag( + true, + "worker-group-A", + "w1", + false, + Some("worker-group-B"), + None + ), Some("worker-group-B".to_string()) ); } @@ -4654,6 +4710,9 @@ async fn push_next_flow_job( &flow_job.workspace_id, flow.preserve_step_tags, payload_tag.tag.as_deref(), + windmill_queue::tags::dev_flow_tag_suffix(&flow_job.workspace_id, &flow_job.tag, db) + .await? + .as_deref(), ); // A step whose inputs failed to evaluate, or whose `$flow_expr[...]` tag failed to resolve, @@ -4794,10 +4853,8 @@ async fn push_next_flow_job( // narrows that further. windmill_common::scopes::intersect_job_token_scopes( flow_job.job_token_scopes.as_deref(), - windmill_common::scopes::step_job_token_scopes( - module.job_token_scopes.as_deref(), - ) - .as_deref(), + windmill_common::scopes::step_job_token_scopes(module.job_token_scopes.as_deref()) + .as_deref(), ) .as_deref(), ) diff --git a/frontend/src/lib/components/DefaultTagsInner.svelte b/frontend/src/lib/components/DefaultTagsInner.svelte index 87cfa0f89b..229c0d5e62 100644 --- a/frontend/src/lib/components/DefaultTagsInner.svelte +++ b/frontend/src/lib/components/DefaultTagsInner.svelte @@ -19,6 +19,11 @@ import Section from './Section.svelte' import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte' + import { + devWorkspaceTagSuffix, + loadDevWorkspaceTagSuffix, + withDevSuffix + } from '$lib/devWorkspaceTags' interface Props { defaultTagPerWorkspace?: boolean | undefined defaultTagWorkspaces?: string[] @@ -134,6 +139,7 @@ loadDefaultTags() loadWorkspaces() + loadDevWorkspaceTagSuffix()
@@ -208,6 +214,22 @@ /> {/snippet} + {#if $devWorkspaceTagSuffix} + + Dev workspace isolation (Workers page) is on and applies after this: a fork's + {forkAppendForkSuffix ? 'python3-{parent_id}-fork' : 'python3-{parent_id}'} + runs as + {withDevSuffix( + forkAppendForkSuffix ? 'python3-{parent_id}-fork' : 'python3-{parent_id}', + $devWorkspaceTagSuffix + )}, on the worker groups serving dev workspaces. To split root and dev jobs across + worker pools, isolation alone is enough. + + {/if} + import { Alert, Button, Section } from './common' + import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte' + import Toggle from './Toggle.svelte' + import TextInput from './text_input/TextInput.svelte' + import Select from './select/Select.svelte' + import Tooltip from './meltComponents/Tooltip.svelte' + import { ArrowRight, ChevronDown, ChevronRight, Save, Settings } from 'lucide-svelte' + import { + SettingService, + WorkerService, + WorkspaceService, + type GetQueueStatusResponse, + type Workspace + } from '$lib/gen' + import { DEV_WORKSPACE_ISOLATION_SETTING, DEV_WORKSPACE_TAG_SUFFIX_SETTING } from '$lib/consts' + import { + DEFAULT_DEV_WORKSPACE_TAG_SUFFIX, + DEV_WORKSPACE_TAG_SUFFIX_PATTERN, + devWorkspaceIsolation, + devWorkspaceListenSuffix, + devWorkspaceTagSuffix, + isDevOrForkWorkspace, + loadDevWorkspaceTagSuffix, + withDevSuffix + } from '$lib/devWorkspaceTags' + import { enterpriseLicense } from '$lib/stores' + import { sendUserToast } from '$lib/toast' + import { pluralize } from '$lib/utils' + import { onMount } from 'svelte' + + interface Props { + /** Worker group configs, by group name. */ + workerGroups: Record | undefined + /** Workers of each group that pinged in the last 15 seconds. */ + liveWorkers: Record + canEdit: boolean + onConfigureGroup: (name: string) => void + } + + let { workerGroups, liveWorkers, canEdit, onConfigureGroup }: Props = $props() + + let suffix = $derived($devWorkspaceTagSuffix) + let enabled = $derived(suffix != null) + let effectiveSuffix = $derived($devWorkspaceListenSuffix) + + // Overridden while a confirmation is pending, reset to the saved state otherwise. + let toggleChecked = $derived(enabled) + let suffixDraft = $derived(effectiveSuffix) + let suffixError = $derived( + DEV_WORKSPACE_TAG_SUFFIX_PATTERN.test(suffixDraft.trim()) + ? undefined + : "Use 1 to 32 letters, digits, '-', '_' or '.'" + ) + + let workspaces = $state(undefined) + let devWorkspaces = $derived( + (workspaces ?? []).filter((w) => !w.deleted && isDevOrForkWorkspace(w)) + ) + let workspaceNames = $derived(Object.fromEntries((workspaces ?? []).map((w) => [w.id, w.name]))) + + async function loadWorkspaces() { + const all: Workspace[] = [] + for (let page = 1; ; page++) { + const batch = await WorkspaceService.listWorkspacesAsSuperAdmin({ page, perPage: 100 }) + all.push(...batch) + if (batch.length < 100) break + } + workspaces = all + } + + let queueStatus = $state(undefined) + async function loadQueueStatus() { + try { + queueStatus = await WorkerService.getQueueStatus() + } catch { + queueStatus = undefined + } + } + + onMount(() => { + loadDevWorkspaceTagSuffix(true) + loadWorkspaces().catch((e) => console.error('Could not list workspaces', e)) + loadQueueStatus() + const interval = setInterval(loadQueueStatus, 15_000) + return () => clearInterval(interval) + }) + + type GroupEntry = { name: string; live: number; tags: string[] } + + function isDedicated(config: any): boolean { + return config?.dedicated_worker != undefined || (config?.dedicated_workers?.length ?? 0) > 0 + } + + let flaggedGroups: GroupEntry[] = $derived( + Object.entries(workerGroups ?? {}) + .filter(([_, c]) => c?.serves_dev_workspaces === true && !isDedicated(c)) + .map(([name, c]) => ({ name, live: liveWorkers[name] ?? 0, tags: c?.worker_tags ?? [] })) + ) + let liveFlaggedWorkers = $derived(flaggedGroups.reduce((acc, g) => acc + g.live, 0)) + let unflaggedGroupNames = $derived( + Object.entries(workerGroups ?? {}) + .filter(([_, c]) => c?.serves_dev_workspaces !== true && !isDedicated(c)) + .map(([name]) => name) + ) + // The only way to break isolation: an unflagged group listing a suffixed tag takes dev jobs. + let leakingGroups = $derived( + Object.entries(workerGroups ?? {}) + .filter(([_, c]) => c?.serves_dev_workspaces !== true) + .map(([name, c]) => ({ + name, + tags: ((c?.worker_tags ?? []) as string[]).filter((t) => t.endsWith(effectiveSuffix)) + })) + .filter((g) => g.tags.length > 0) + ) + let unservedDevTags = $derived( + enabled + ? (queueStatus ?? []).filter( + (s) => s.tag.endsWith(effectiveSuffix) && s.waiting > 0 && s.workers === 0 + ) + : [] + ) + + let groupToFlag = $state(undefined) + let collapsed = $state(true) + let showWorkspaces = $state(false) + let showAdvanced = $state(false) + + let confirmEnable = $state(false) + let confirmDisable = $state(false) + let saving = $state(false) + + async function setIsolation(on: boolean) { + saving = true + try { + await SettingService.setGlobal({ + key: DEV_WORKSPACE_ISOLATION_SETTING, + requestBody: { value: on } + }) + devWorkspaceIsolation.update((i) => ({ suffix: i?.suffix ?? effectiveSuffix, enabled: on })) + sendUserToast( + on + ? `Dev workspace and fork jobs now run on tags suffixed with ${effectiveSuffix}` + : 'Dev workspace isolation turned off' + ) + } catch (e) { + sendUserToast(`Could not save dev workspace isolation: ${e}`, true) + toggleChecked = enabled + } finally { + saving = false + } + } + + async function saveSuffix(value: string) { + saving = true + try { + await SettingService.setGlobal({ + key: DEV_WORKSPACE_TAG_SUFFIX_SETTING, + requestBody: { value } + }) + devWorkspaceIsolation.update((i) => ({ enabled: i?.enabled ?? false, suffix: value })) + sendUserToast(`Dev workspace tag suffix set to ${value}`) + } catch (e) { + sendUserToast(`Could not save the dev workspace tag suffix: ${e}`, true) + } finally { + saving = false + } + } + + function onToggle(checked: boolean) { + if (checked === enabled) return + collapsed = false + if (checked) { + if (liveFlaggedWorkers === 0) { + confirmEnable = true + } else { + setIsolation(true) + } + } else { + confirmDisable = true + } + } + + function workspaceKind(w: Workspace): string { + if (w.parent_workspace_id) { + const parent = workspaceNames[w.parent_workspace_id] ?? w.parent_workspace_id + return w.is_dev_workspace ? `dev workspace of ${parent}` : `fork of ${parent}` + } + return 'fork whose parent was deleted' + } + + + { + confirmEnable = false + toggleChecked = enabled + }} + onConfirmed={async () => { + confirmEnable = false + await setIsolation(true) + }} +> +
+

+ {#if workspaces === undefined} + Every dev workspace and fork + {:else} + {pluralize(devWorkspaces.length, 'dev workspace or fork', 'dev workspaces and forks')} + {/if} + will have their jobs queued until a worker group serves them. +

+

+ {#if flaggedGroups.length === 0} + No worker group is flagged as serving dev workspaces yet. + {:else} + The worker groups serving dev workspaces ({flaggedGroups.map((g) => g.name).join(', ')}) + have no live worker. + {/if} + Jobs never fall back to the other worker groups: they wait in the queue and run as soon as a serving + worker is up. +

+ {#if unflaggedGroupNames.length > 0 && flaggedGroups.length === 0} +
+ +
+ {/if} +
+ + + { + confirmDisable = false + toggleChecked = enabled + }} + onConfirmed={async () => { + confirmDisable = false + await setIsolation(false) + }} +> +

+ New jobs of dev workspaces and forks will run on the same worker groups as root workspaces + again. Jobs already queued on a {effectiveSuffix} tag keep it, and the worker groups + serving dev workspaces still run them. +

+ + +{#snippet isolationToggle()} + onToggle(e.detail)} + disabled={!canEdit || !$enterpriseLicense || saving || suffix === undefined} + options={{ right: 'Run dev workspaces and forks on separate workers' }} + size="xs" + /> +{/snippet} + +
+ {#snippet action()} + {@render isolationToggle()} + {/snippet} + {#snippet badge()} + {@render isolationToggle()} + {/snippet} + +
+
+ Root workspaces + + + Worker groups not serving dev workspaces, on their usual tags: + python3, gpu + + Dev workspaces and forks + + + {#if enabled} + Worker groups serving dev workspaces, on tags suffixed with + {effectiveSuffix}: + {:else} + Once on, worker groups serving dev workspaces, on suffixed tags: + {/if} + {withDevSuffix('python3', effectiveSuffix)}, + {withDevSuffix('gpu', effectiveSuffix)} + +
+

+ A workspace with a parent, and every fork (including a fork whose parent was deleted), counts + as a dev workspace. Every job is covered: runs, previews, schedules, triggers, flow steps, AI + agent tools and dependency jobs, and inline runs are refused on workers not serving dev + workspaces. A job whose tag no serving worker listens to stays queued: it never falls back to + the other worker groups. Dedicated workers keep their own tags. +

+ + +
+ Worker groups serving dev workspaces + {#if flaggedGroups.length > 0} +
+ {#each flaggedGroups as g (g.name)} + + {/each} +
+ {:else} + None yet. + {/if} + {#if canEdit && unflaggedGroupNames.length > 0} +
+ + + + Jobs in this workspace run on dev workers (tags suffixed with + {$devWorkspaceTagSuffix}, e.g. + {withDevSuffix('python3', $devWorkspaceTagSuffix)}), set by the instance's dev + workspace isolation. + +

+ {/if} +{/snippet} + {#if isThrowawayFork}

Dev workspace pairing is not available on a throwaway fork. This workspace is a fork of {parentId}.

+
{@render devWorkersNotice()}
{@render deployTarget?.()} {:else if isDev && parentId}
@@ -563,6 +593,7 @@ {currentLabel} branch.
+ {@render devWorkersNotice()}
{@render deployTarget?.()}
diff --git a/frontend/src/lib/components/WorkerGroup.svelte b/frontend/src/lib/components/WorkerGroup.svelte index 5e8957c31a..da9e5cf4a9 100644 --- a/frontend/src/lib/components/WorkerGroup.svelte +++ b/frontend/src/lib/components/WorkerGroup.svelte @@ -56,6 +56,12 @@ import DedicatedWorkersSelector from './DedicatedWorkersSelector.svelte' import { computeHashedTag } from './dedicated_worker' import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' + import { + devWorkspaceListenSuffix, + devWorkspaceTagSuffix, + loadDevWorkspaceTagSuffix, + withDevSuffix + } from '$lib/devWorkspaceTags' import ObjectStoreConfigSettings, { type ObjectStoreConfig } from './ObjectStoreConfigSettings.svelte' @@ -106,6 +112,7 @@ native_mode?: boolean object_store_cache_config?: ObjectStoreConfig paused?: boolean + serves_dev_workspaces?: boolean } = $state({}) function loadNConfig() { @@ -214,6 +221,7 @@ native_mode?: boolean object_store_cache_config?: ObjectStoreConfig paused?: boolean + serves_dev_workspaces?: boolean } activeWorkers: number customTags: string[] | undefined @@ -319,7 +327,8 @@ 'priority_tags', 'additional_python_paths', 'pip_local_dependencies', - 'object_store_cache_config' + 'object_store_cache_config', + 'serves_dev_workspaces' ] as const function restartsWorkers(body: typeof nconfig): boolean { if (activeWorkers === 0) return false @@ -381,6 +390,13 @@ !t.startsWith('flow-') ) ) + loadDevWorkspaceTagSuffix() + let devSuffix = $derived($devWorkspaceListenSuffix) + let servesDev = $derived( + config?.serves_dev_workspaces === true && + config?.dedicated_worker == undefined && + !config?.dedicated_workers?.length + ) let isAutoNativeMode = $derived(name === 'native') let isNativeModeEnabled = $derived(nconfig?.native_mode === true || isAutoNativeMode) $effect(() => { @@ -647,6 +663,63 @@ {/if} + {#if nconfig !== undefined} + {@const flagged = nconfig.serves_dev_workspaces === true} + {@const draftTags = nconfig.worker_tags ?? []} + {@const handSuffixed = draftTags.filter((t) => t.endsWith(devSuffix))} +
+ + {/if} + {#if nconfig?.worker_tags !== undefined && nconfig?.worker_tags.length > 0}