feat(queue): per-workspace fairness cap on the shared cloud worker pool (#9303)

* feat(queue): cloud-only per-workspace fairness cap on the shared worker pool

On `app.windmill.dev` the cluster runs a single default worker group, so a
single workspace flooding the queue can degrade quality of service for
everyone else. This adds an opt-in mechanism that caps any single workspace
at a configurable share of the shared worker pool when it has been
dominating cluster activity for more than a configurable window.

Detection signal counts both currently-running jobs and jobs completed in
the rolling window, so it catches workspaces hogging slots with long jobs
**and** workspaces spamming many tiny jobs (where no individual job's
started_at is old, but throughput share dominates).

Refresh is coordinated cluster-wide via a single UPDATE on
`background_task_state`: the `WHERE updated_at < now() - interval` predicate
combined with row-level locking means only one process per refresh cycle
actually runs the aggregation, regardless of fleet size. Every other
process gets the freshly written value in the same round trip via
`UNION ALL ... LIMIT 1`. Heavy aggregation rate stays at ~0.2-0.5 qps for
the whole cluster.

Pull queries are split: the existing query string and its bind shape stay
bit-identical to today, so the planner keeps using the same indexes when
fairness is off or no workspace is currently capped. A separate
`WORKER_PULL_QUERIES_FAIRNESS` adds `AND workspace_id <> ALL($2::text[])`
and is only materialized while the feature is enabled.

Hard-gated to `CLOUD_HOSTED=true` + BASE_URL host == app.windmill.dev at
three layers: frontend `cloudonly: true`, API setter rejection in
`set_global_setting_internal`, runtime check in `fairness_active`. Settings
are exposed under Jobs in the instance-settings UI; defaults are off so
the change is a no-op for self-hosted.

Two-pass pull guarantees no worker idling: if every queued job belongs to
a capped workspace, the second pass uses the unmodified pull queries.
Cap re-asserts on the next refresh.

Fixes WIN-1982

* fix(queue): address CI review findings on workspace fairness

Six fixes from the four-reviewer cross-check on #9303:

1. **Aggregation evaluation (Codex P1).** The previous `INSERT ... ON CONFLICT
   DO UPDATE WHERE updated_at < ...` had the heavy `v2_job_queue ∪
   v2_job_completed` aggregation inlined into `VALUES`, which Postgres
   evaluates for every contender to build the proposed row — losing the
   "one heavy aggregation per cycle cluster-wide" property the design
   advertises. Split into three small statements: (a) cheap claim with
   constant `VALUES`, (b) winner-only `UPDATE ... SET value = jsonb_build_object('overloaded', <agg>)`
   (Postgres only evaluates `SET` per row matching `WHERE`, so losers never
   compute the aggregation), (c) read for everyone. Heavy query now truly
   runs ~0.2-0.5 qps cluster-wide regardless of fleet size.

2. **Numeric setting wraparound (cubic P1).** `u64 as u32` and downstream
   `u32 as i32` could silently flip sign and feed `make_interval(secs => -N)`,
   making `now() - interval` a future timestamp and disabling the
   completed-jobs half of the activity signal. Clamp `duration_secs` to
   [1, 86400] and `min_total_jobs` to [0, u32::MAX] before storing.

3. **`/instance_config` bypass (cubic/Claude/Codex P2).** Bulk config endpoint
   sidestepped `set_global_setting_internal`'s gate; a self-hosted superadmin
   could persist `workspace_fairness_*` rows via the bulk path. Mirror the
   per-key check in `set_instance_config` upsert flow.

4. **DB error coerced to false (Claude P2).** `load_workspace_fairness_enabled`
   collapsed `Err(_)` to `false` and unconditionally swapped the atomic — a
   transient DB blip during notify-event propagation toggled the feature off
   cluster-wide (and triggered a `store_pull_query` rebuild precisely when load
   is highest). Now propagates the error so the atomic stays at its prior value.

5. **Refresh failure cooldown (Claude P2).** Storing `0` removed the rate
   limit entirely; every subsequent pull spawned a new refresh task. Leave
   `LAST_REFRESH_MICROS` at `now_us` (already written by the CAS) so the
   natural interval acts as the cooldown.

6. **Visibility + duplication (Pi P2).** Mark `make_pull_query_fairness` as
   `pub(crate)`. Move the duplicated `BASE_URL host == app.windmill.dev`
   parser into `windmill-common::worker::is_cloud_production_host` and share
   it between the API setter and the runtime path.

Verified locally:
- `POST /api/settings/global/workspace_fairness_enabled` → 400 (per-key gate)
- `PUT /api/settings/instance_config` with fairness key → 400 (bulk gate)
- `cargo check --workspace --features=private,enterprise,quickjs` — clean

Refs WIN-1982.

* fix(queue): second round of CI review nits on workspace fairness

Three issues raised by the Codex/Claude re-review of commit 0b38ff2:

1. Non-cloud deletes were rejected (Codex P2). The cloud gate ran before
   the Null / empty-string deletion branches in both `set_global_setting_internal`
   and the bulk `set_instance_config`. A self-hosted instance that inherited
   stale `workspace_fairness_*` rows from a cloned cloud DB couldn't clear
   them through the API — the rows stayed in `global_settings` and continued
   to show up in the YAML export. Now the gate only blocks upserts; Null /
   empty-string deletes pass through on any host.

2. Deleted numeric knobs kept stale runtime values (Codex P2). When a
   cloud admin cleared `workspace_fairness_max_percent`, `..._duration_secs`,
   or `..._min_total_jobs`, the notify-event fired but the numeric loaders
   ignored `Ok(None)` and left the previous in-memory value pinned until
   process restart. Loaders now distinguish three outcomes:
     - `Err(_)`: transient — leave atomic alone (preserves the
       previous-round fix).
     - `Ok(None)` / `Ok(Some(invalid))`: reset to the documented default.
     - `Ok(Some(valid))`: clamp and store.
   Defaults are extracted to `WORKSPACE_FAIRNESS_*_DEFAULT` constants kept
   in sync with the `AtomicU32::new(...)` initialisers in
   `windmill-common/src/worker.rs`.

3. `fairness_active` was `pub` with no cross-crate caller (Claude nit).
   Tightened to module-private.

Verified locally on this non-cloud instance:
  POST .../workspace_fairness_enabled  body=null  → 200 (delete passes)
  POST .../workspace_fairness_enabled  body=true  → 400 (set blocked)
  PUT .../instance_config              {}         → 200 (no-op passes)
  PUT .../instance_config  with fairness key      → 400 (bulk set blocked)

Skipped the partial index on `v2_job_queue WHERE running = true` that
Claude flagged as a residual nit — queue stays under 50k rows per the
operator's measurement, so the seq-scan cost (~10 ms × 0.5 qps =
~0.5% of a DB core) is well below the noise floor and the index isn't
worth the maintenance cost on job transitions.

Refs WIN-1982.

backend/src/main.rs

@@ -59,7 +59,9 @@ use windmill_common::{
         RETENTION_PERIOD_SECS_SETTING, RUBY_REPOS_SETTING, SAML_METADATA_SETTING,
         SCIM_TOKEN_SETTING, SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, TEAMS_SETTING,
         TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING,
-        UV_PYTHON_INSTALL_MIRROR_SETTING, WORKSPACE_REGISTRIES_SETTING,
+        UV_PYTHON_INSTALL_MIRROR_SETTING, WORKSPACE_FAIRNESS_DURATION_SECS_SETTING,
+        WORKSPACE_FAIRNESS_ENABLED_SETTING, WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING,
+        WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING, WORKSPACE_REGISTRIES_SETTING,
     },
     scripts::ScriptLang,
     stats_oss::schedule_stats,
@@ -120,7 +122,9 @@ use crate::monitor::{
     initial_load, load_disable_password_login, load_fork_workspace_tag_append_fork_suffix,
     load_keep_job_dir, load_metrics_debug_enabled, load_preview_tags_override,
     load_require_preexisting_user, load_tag_per_workspace_enabled,
-    load_tag_per_workspace_workspaces, monitor_db, reload_app_workspaced_route_setting,
+    load_tag_per_workspace_workspaces, load_workspace_fairness_duration_secs,
+    load_workspace_fairness_enabled, load_workspace_fairness_max_percent,
+    load_workspace_fairness_min_total, monitor_db, reload_app_workspaced_route_setting,
     reload_audit_log_retention_days_setting, reload_base_url_setting,
     reload_bun_install_min_release_age_setting, reload_bunfig_install_scopes_setting,
     reload_critical_alert_mute_ui_setting, reload_critical_alerts_on_token_expiry_setting,
@@ -1765,6 +1769,26 @@ async fn process_notify_event(
                         tracing::error!("Error loading preview tags override: {e:#}");
                     }
                 }
+                WORKSPACE_FAIRNESS_ENABLED_SETTING => {
+                    if let Err(e) = load_workspace_fairness_enabled(db).await {
+                        tracing::error!("Error loading workspace fairness enabled: {e:#}");
+                    }
+                }
+                WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING => {
+                    if let Err(e) = load_workspace_fairness_max_percent(db).await {
+                        tracing::error!("Error loading workspace fairness max percent: {e:#}");
+                    }
+                }
+                WORKSPACE_FAIRNESS_DURATION_SECS_SETTING => {
+                    if let Err(e) = load_workspace_fairness_duration_secs(db).await {
+                        tracing::error!("Error loading workspace fairness duration secs: {e:#}");
+                    }
+                }
+                WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING => {
+                    if let Err(e) = load_workspace_fairness_min_total(db).await {
+                        tracing::error!("Error loading workspace fairness min total: {e:#}");
+                    }
+                }
                 SMTP_SETTING => {
                     reload_smtp_config(db).await;
                 }

backend/src/monitor.rs

@@ -69,6 +69,8 @@ use windmill_common::{
         RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
         STORE_AUDIT_LOGS_S3_SETTING, TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING,
         UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
+        WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
+        WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
     },
     indexer::load_indexer_config,
     jwt::JWT_SECRET,
@@ -84,7 +86,8 @@ use windmill_common::{
         store_suspended_pull_query, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE,
         DEFAULT_TAGS_WORKSPACES, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, INDEXER_CONFIG,
         PREVIEW_TAGS_OVERRIDE, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, WINDMILL_DIR, WORKER_CONFIG,
-        WORKER_GROUP,
+        WORKER_GROUP, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
+        WORKSPACE_FAIRNESS_MAX_PERCENT, WORKSPACE_FAIRNESS_MIN_TOTAL,
     },
     KillpillSender, AUDIT_LOG_RETENTION_DAYS, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE,
     CRITICAL_ALERTS_ON_TOKEN_EXPIRY, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB,
@@ -248,6 +251,22 @@ pub async fn initial_load(
         if let Err(e) = load_preview_tags_override(db).await {
             tracing::error!("Error loading preview tags override: {e:#}");
         }
+
+        // Workspace fairness (cloud-only). Load the percentage/duration/min knobs
+        // *before* the enabled flag so that `load_workspace_fairness_enabled` reads
+        // current values when re-storing the pull queries.
+        if let Err(e) = load_workspace_fairness_max_percent(db).await {
+            tracing::error!("Error loading workspace fairness max percent: {e:#}");
+        }
+        if let Err(e) = load_workspace_fairness_duration_secs(db).await {
+            tracing::error!("Error loading workspace fairness duration secs: {e:#}");
+        }
+        if let Err(e) = load_workspace_fairness_min_total(db).await {
+            tracing::error!("Error loading workspace fairness min total: {e:#}");
+        }
+        if let Err(e) = load_workspace_fairness_enabled(db).await {
+            tracing::error!("Error loading workspace fairness enabled: {e:#}");
+        }
     }
 
     if server_mode {
@@ -543,6 +562,112 @@ pub async fn load_preview_tags_override(db: &DB) -> error::Result<()> {
     Ok(())
 }
 
+// Upper bound on the duration window. Postgres `make_interval(secs => $1::int4)` is the consumer
+// downstream, so this stays comfortably below `i32::MAX` and the subsequent `u32 -> i32` cast in
+// `workspace_fairness::refresh_overloaded` cannot wrap into a negative interval (which would
+// silently turn `now() - interval` into a future timestamp and disable the completed-jobs half
+// of the activity signal). A day is the practical ceiling for a "rolling window" knob.
+const WORKSPACE_FAIRNESS_DURATION_SECS_MAX: u64 = 86_400;
+
+/// Min-total floor is a counting threshold; cap at `u32::MAX` to make wraparound impossible
+/// while still leaving more headroom than any realistic cluster will need.
+const WORKSPACE_FAIRNESS_MIN_TOTAL_MAX: u64 = u32::MAX as u64;
+
+// Defaults used when a fairness knob is unset (row missing or row deleted via NULL/empty value).
+// Must stay in sync with the `AtomicU32::new(...)` initialisers in `windmill-common/src/worker.rs`
+// so a process that has never seen the setting reads the same value as one that just saw it
+// cleared.
+const WORKSPACE_FAIRNESS_MAX_PERCENT_DEFAULT: u32 = 50;
+const WORKSPACE_FAIRNESS_DURATION_SECS_DEFAULT: u32 = 10;
+const WORKSPACE_FAIRNESS_MIN_TOTAL_DEFAULT: u32 = 4;
+
+pub async fn load_workspace_fairness_enabled(db: &DB) -> error::Result<()> {
+    // Match the convention used by `load_preview_tags_override` /
+    // `load_fork_workspace_tag_append_fork_suffix`: on transient DB errors, leave the in-memory
+    // atomic untouched rather than silently toggling the feature off across the whole cluster
+    // (which would also trigger an unnecessary `store_pull_query` rebuild — exactly when DB load
+    // is probably highest).
+    let new_enabled =
+        match load_value_from_global_settings(db, WORKSPACE_FAIRNESS_ENABLED_SETTING).await? {
+            Some(serde_json::Value::Bool(t)) => t,
+            // Setting unset / non-bool → explicit off.
+            _ => false,
+        };
+    let prev = WORKSPACE_FAIRNESS_ENABLED.swap(new_enabled, Ordering::Relaxed);
+    // Re-store the pull queries so the fairness variants appear/disappear in
+    // lockstep with the toggle.
+    if prev != new_enabled {
+        let wc = windmill_common::worker::WORKER_CONFIG.load_full();
+        store_pull_query(&wc).await;
+    }
+    Ok(())
+}
+
+pub async fn load_workspace_fairness_max_percent(db: &DB) -> error::Result<()> {
+    // Distinguish three outcomes:
+    //   - `Err(_)`: transient DB issue. Leave the atomic alone (don't clobber a known-good value
+    //     because of a network blip during a notify-event propagation).
+    //   - `Ok(None)` or `Ok(Some(invalid))`: setting is unset / explicitly cleared / corrupt.
+    //     Restore the default so a deletion via the admin UI actually takes effect at runtime
+    //     instead of leaving the stale in-memory value pinned until restart.
+    //   - `Ok(Some(valid))`: clamp and store.
+    match load_value_from_global_settings(db, WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING).await? {
+        Some(serde_json::Value::Number(n)) => {
+            let v = n
+                .as_u64()
+                .map(|u| u.clamp(1, 100) as u32)
+                .unwrap_or(WORKSPACE_FAIRNESS_MAX_PERCENT_DEFAULT);
+            WORKSPACE_FAIRNESS_MAX_PERCENT.store(v, Ordering::Relaxed);
+        }
+        _ => {
+            WORKSPACE_FAIRNESS_MAX_PERCENT
+                .store(WORKSPACE_FAIRNESS_MAX_PERCENT_DEFAULT, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
+pub async fn load_workspace_fairness_duration_secs(db: &DB) -> error::Result<()> {
+    // See `load_workspace_fairness_max_percent` for the Err / None / invalid policy.
+    match load_value_from_global_settings(db, WORKSPACE_FAIRNESS_DURATION_SECS_SETTING).await? {
+        Some(serde_json::Value::Number(n)) => {
+            // Clamp to the safe range before narrowing. The downstream `u32 -> i32` cast in
+            // `workspace_fairness::refresh_overloaded` makes any value above `i32::MAX` toxic
+            // (sign flip → negative interval → silent disable of the completed-jobs scan).
+            let v = n
+                .as_u64()
+                .map(|u| u.clamp(1, WORKSPACE_FAIRNESS_DURATION_SECS_MAX) as u32)
+                .unwrap_or(WORKSPACE_FAIRNESS_DURATION_SECS_DEFAULT);
+            WORKSPACE_FAIRNESS_DURATION_SECS.store(v, Ordering::Relaxed);
+        }
+        _ => {
+            WORKSPACE_FAIRNESS_DURATION_SECS
+                .store(WORKSPACE_FAIRNESS_DURATION_SECS_DEFAULT, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
+pub async fn load_workspace_fairness_min_total(db: &DB) -> error::Result<()> {
+    // See `load_workspace_fairness_max_percent` for the Err / None / invalid policy.
+    match load_value_from_global_settings(db, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING).await? {
+        Some(serde_json::Value::Number(n)) => {
+            // Clamp before narrowing — same reasoning as `_duration_secs`, just for the
+            // counting threshold rather than the interval.
+            let v = n
+                .as_u64()
+                .map(|u| u.min(WORKSPACE_FAIRNESS_MIN_TOTAL_MAX) as u32)
+                .unwrap_or(WORKSPACE_FAIRNESS_MIN_TOTAL_DEFAULT);
+            WORKSPACE_FAIRNESS_MIN_TOTAL.store(v, Ordering::Relaxed);
+        }
+        _ => {
+            WORKSPACE_FAIRNESS_MIN_TOTAL
+                .store(WORKSPACE_FAIRNESS_MIN_TOTAL_DEFAULT, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
 pub async fn load_fork_workspace_tag_append_fork_suffix(db: &DB) -> error::Result<()> {
     let value =
         load_value_from_global_settings(db, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING).await;

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

@@ -54,10 +54,14 @@ use windmill_common::{
         AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, AUTOMATE_USERNAME_CREATION_SETTING,
         CRITICAL_ALERT_MUTE_UI_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, DISABLE_HUB_SETTING,
         EMAIL_DOMAIN_SETTING, ENV_SETTINGS, HTTP_ROUTE_WORKSPACED_ROUTE_SETTING,
-        HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING, RUFF_CONFIG_SETTING, WS_BASE_URL_SETTING,
+        HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING, RUFF_CONFIG_SETTING,
+        WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
+        WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
+        WS_BASE_URL_SETTING,
     },
     instance_config::{self, ApplyMode, InstanceConfig},
     server::Smtp,
+    worker::is_cloud_production_host,
 };
 use windmill_common::{error::to_anyhow, PgDatabase};
 
@@ -446,6 +450,24 @@ pub async fn delete_global_setting(db: &DB, key: &str) -> error::Result<()> {
     tracing::info!("Unset global setting {}", key);
     Ok(())
 }
+/// Returns true when `key` is one of the workspace-fairness settings whose
+/// writes must be gated to cloud only.
+fn is_workspace_fairness_setting(key: &str) -> bool {
+    matches!(
+        key,
+        WORKSPACE_FAIRNESS_ENABLED_SETTING
+            | WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING
+            | WORKSPACE_FAIRNESS_DURATION_SECS_SETTING
+            | WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING
+    )
+}
+
+/// Cloud-and-app.windmill.dev gate for workspace fairness. Must hold to persist
+/// the setting; the runtime path additionally verifies before applying the cap.
+fn workspace_fairness_settings_allowed() -> bool {
+    is_cloud_production_host()
+}
+
 pub async fn set_global_setting(
     Extension(db): Extension<DB>,
     authed: ApiAuthed,
@@ -468,6 +490,27 @@ pub async fn set_global_setting_internal(
         value
     };
 
+    // Hard-gate the cloud-only workspace fairness settings: refuse to persist
+    // them on any instance that is not CLOUD_HOSTED + app.windmill.dev. This is
+    // belt-and-suspenders alongside the frontend `{#if isCloudHosted()}` wrap
+    // and the runtime check in `workspace_fairness::fairness_active`.
+    //
+    // Deletes (Null / empty-string) are *allowed* on non-cloud so admins can clear
+    // stale rows that ended up in `global_settings` via a cloned cloud DB. Without
+    // this exception a self-hosted instance would be stuck with cloud-only rows
+    // showing up in its instance-config YAML export.
+    let is_clearing_value = matches!(&value, serde_json::Value::Null)
+        || matches!(&value, serde_json::Value::String(s) if s.trim().is_empty());
+    if is_workspace_fairness_setting(&key)
+        && !is_clearing_value
+        && !workspace_fairness_settings_allowed()
+    {
+        return Err(error::Error::BadRequest(format!(
+            "{} is only configurable on app.windmill.dev cloud (CLOUD_HOSTED + BASE_URL match required)",
+            key
+        )));
+    }
+
     run_setting_pre_write_hook(db, &key, &value).await?;
 
     match value {
@@ -726,6 +769,25 @@ async fn set_instance_config(
             .iter()
             .any(|(key, _)| key == AI_CONFIG_SETTING);
 
+        // Mirror the per-key cloud gate in `set_global_setting_internal`. Without this, the
+        // bulk endpoint would let a self-hosted superadmin persist `workspace_fairness_*` rows
+        // even though the per-key API rejects them. The runtime check in
+        // `workspace_fairness::fairness_active` still keeps the cap inert there, but persisting
+        // the rows would be a leak of cloud-only config into non-cloud DBs and would advertise
+        // the feature in the YAML export.
+        //
+        // Only block *upserts*; deletes are allowed everywhere so admins can clean up stale
+        // rows (e.g. from a cloned cloud DB) without flipping `CLOUD_HOSTED` on temporarily.
+        let upserts_touch_fairness = settings_diff
+            .upserts
+            .keys()
+            .any(|k| is_workspace_fairness_setting(k));
+        if upserts_touch_fairness && !workspace_fairness_settings_allowed() {
+            return Err(error::Error::BadRequest(
+                "Workspace fairness settings are only configurable on app.windmill.dev cloud (CLOUD_HOSTED + BASE_URL match required)".to_string(),
+            ));
+        }
+
         for (key, value) in &settings_diff.upserts {
             run_setting_pre_write_hook(&db, key, value).await?;
         }

backend/windmill-common/src/global_settings.rs

@@ -86,6 +86,15 @@ pub const WORKSPACE_REGISTRIES_SETTING: &str = "workspace_registries";
 pub const RESTART_COORDINATION_SETTING: &str = "_restart_coordination";
 pub const ALERT_CONFIG_SETTING: &str = "alert_job_queue_waiting";
 
+// Workspace fairness: cloud-only mechanism that caps any single workspace at
+// `workspace_fairness_max_percent`% of the shared worker pool once it has been
+// occupying it for more than `workspace_fairness_duration_secs` seconds. See
+// `windmill-queue/src/workspace_fairness.rs`.
+pub const WORKSPACE_FAIRNESS_ENABLED_SETTING: &str = "workspace_fairness_enabled";
+pub const WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING: &str = "workspace_fairness_max_percent";
+pub const WORKSPACE_FAIRNESS_DURATION_SECS_SETTING: &str = "workspace_fairness_duration_secs";
+pub const WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING: &str = "workspace_fairness_min_total_jobs";
+
 use std::sync::atomic::AtomicBool;
 
 lazy_static::lazy_static! {

backend/windmill-common/src/worker.rs

@@ -17,7 +17,7 @@ use std::{
     panic::Location,
     path::{Component, Path, PathBuf},
     str::FromStr,
-    sync::atomic::AtomicBool,
+    sync::atomic::{AtomicBool, AtomicI64, AtomicU32},
     time::Duration,
 };
 #[cfg(windows)]
@@ -237,14 +237,30 @@ lazy_static::lazy_static! {
     });
 
     pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
+    pub static ref WORKER_PULL_QUERIES_FAIRNESS: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
     pub static ref WORKER_SUSPENDED_PULL_QUERY: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee("".to_string());
 
+    // Workspace fairness (cloud-only). When enabled, a workspace whose footprint over the rolling
+    // `WORKSPACE_FAIRNESS_DURATION_SECS` window represents >= `WORKSPACE_FAIRNESS_MAX_PERCENT`% of
+    // all worker activity gets excluded from the pull query, freeing slots for other workspaces.
+    // The list of overloaded workspaces is computed cluster-wide via a single coordinated UPDATE
+    // on `background_task_state` so only one process per refresh interval runs the aggregation.
+    pub static ref WORKSPACE_FAIRNESS_ENABLED: AtomicBool = AtomicBool::new(false);
+    pub static ref WORKSPACE_FAIRNESS_MAX_PERCENT: AtomicU32 = AtomicU32::new(50);
+    pub static ref WORKSPACE_FAIRNESS_DURATION_SECS: AtomicU32 = AtomicU32::new(10);
+    pub static ref WORKSPACE_FAIRNESS_MIN_TOTAL: AtomicU32 = AtomicU32::new(4);
+    pub static ref WORKSPACE_FAIRNESS_OVERLOADED: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
+    pub static ref WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS: AtomicI64 = AtomicI64::new(0);
+
 
     pub static ref SMTP_CONFIG: arc_swap::ArcSwap<Option<Smtp>> = arc_swap::ArcSwap::from_pointee(None);
     pub static ref INDEXER_CONFIG: arc_swap::ArcSwap<TantivyIndexerSettings> = arc_swap::ArcSwap::from_pointee(TantivyIndexerSettings::default());
 
 
     pub static ref CLOUD_HOSTED: bool = std::env::var("CLOUD_HOSTED").is_ok();
+    /// Host used to gate cloud-only features that must only ever run on the
+    /// production `app.windmill.dev` cluster, not on staging or self-hosted.
+    pub static ref CLOUD_PRODUCTION_HOST: &'static str = "app.windmill.dev";
 
     pub static ref CUSTOM_TAGS: Vec<String> = std::env::var("CUSTOM_TAGS")
         .ok()
@@ -289,6 +305,34 @@ pub fn is_native_mode_from_env() -> bool {
     *NATIVE_MODE || *WORKER_GROUP == "native"
 }
 
+/// True iff this process is configured to act as the production cloud cluster:
+/// `CLOUD_HOSTED=true` AND `BASE_URL`'s host matches `CLOUD_PRODUCTION_HOST`.
+/// Centralized so the API setter, the runtime pull path, and any future cloud-
+/// only feature share one canonical check (rather than re-implementing the
+/// scheme/host parser at each call site).
+pub fn is_cloud_production_host() -> bool {
+    if !*CLOUD_HOSTED {
+        return false;
+    }
+    let base = crate::BASE_URL.load();
+    let s = base.as_str();
+    if s.is_empty() {
+        return false;
+    }
+    let after_scheme = s
+        .strip_prefix("https://")
+        .or_else(|| s.strip_prefix("http://"))
+        .unwrap_or(s);
+    let host = after_scheme
+        .split('/')
+        .next()
+        .unwrap_or("")
+        .split(':')
+        .next()
+        .unwrap_or("");
+    host == *CLOUD_PRODUCTION_HOST
+}
+
 /// Cached resolved native mode flag, updated when worker config is reloaded.
 /// Use this for hot-path checks (e.g. per-job dispatch) to avoid read-locking WORKER_CONFIG.
 pub static NATIVE_MODE_RESOLVED: AtomicBool = AtomicBool::new(false);
@@ -520,17 +564,43 @@ pub fn make_pull_query(tags: &[String]) -> String {
     query
 }
 
+// Variant of `make_pull_query` that additionally excludes jobs whose workspace_id is in the
+// overloaded-list bind parameter ($2::text[]). Built as a separate string (rather than reusing
+// `make_pull_query` with an always-bound array) so the planner can keep using the same indexes
+// when fairness is off — the default `make_pull_query` text stays bit-identical to today's.
+//
+// `pub(crate)` because only `store_pull_query` consumes it; the resulting query string is what
+// crosses crate boundaries via `WORKER_PULL_QUERIES_FAIRNESS`.
+pub(crate) fn make_pull_query_fairness(tags: &[String]) -> String {
+    let query = format_pull_query(format!(
+        "SELECT id
+        FROM v2_job_queue
+        WHERE running = false AND tag IN ({}) AND scheduled_for <= now()
+          AND workspace_id <> ALL($2::text[])
+        ORDER BY priority DESC NULLS LAST, scheduled_for
+        FOR UPDATE SKIP LOCKED
+        LIMIT 1",
+        tags.iter().map(|x| format!("'{x}'")).join(", ")
+    ));
+    query
+}
+
 pub async fn store_pull_query(wc: &WorkerConfig) {
     let mut queries = vec![];
+    let mut fairness_queries = vec![];
+    let fairness_enabled = WORKSPACE_FAIRNESS_ENABLED.load(std::sync::atomic::Ordering::Relaxed);
     for tags in wc.priority_tags_sorted.iter() {
         if tags.tags.len() == 0 {
             tracing::error!("Empty tags in priority tags, skipping");
             continue;
         }
-        let query = make_pull_query(&tags.tags);
-        queries.push(query);
+        queries.push(make_pull_query(&tags.tags));
+        if fairness_enabled {
+            fairness_queries.push(make_pull_query_fairness(&tags.tags));
+        }
     }
     WORKER_PULL_QUERIES.store(std::sync::Arc::new(queries));
+    WORKER_PULL_QUERIES_FAIRNESS.store(std::sync::Arc::new(fairness_queries));
 }
 
 lazy_static::lazy_static! {

backend/windmill-queue/src/jobs.rs

@@ -78,7 +78,8 @@ use windmill_common::{
     utils::{not_found_if_none, report_critical_error, StripPath, WarnAfterExt},
     worker::{
         to_raw_value, CLOUD_HOSTED, DISABLE_FLOW_SCRIPT, NO_LOGS, PREVIEW_TAGS_OVERRIDE,
-        WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY,
+        WORKER_PULL_QUERIES, WORKER_PULL_QUERIES_FAIRNESS, WORKER_SUSPENDED_PULL_QUERY,
+        WORKSPACE_FAIRNESS_OVERLOADED,
     },
     DB, METRICS_ENABLED,
 };
@@ -3632,28 +3633,67 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>(
                 return Ok((None, false));
             }
 
-            for query in queries.iter() {
-                // tracing::info!("Pulling job with query: {}", query);
-                // let instant = std::time::Instant::now();
+            // Workspace fairness (cloud-only): if the fairness refresh has flagged any
+            // overloaded workspaces, try the fairness-aware pull queries first (which
+            // exclude those workspace_ids). When fairness is off or no workspace is
+            // currently capped, this branch is skipped and the hot path is identical
+            // to today's. Lazy refresh is fired from the same place; it runs at most
+            // once per process per refresh interval and never blocks this pull.
+            crate::workspace_fairness::maybe_refresh_overloaded(db);
+            let overloaded = WORKSPACE_FAIRNESS_OVERLOADED.load_full();
+            let fairness_active = !overloaded.is_empty();
 
-                #[cfg(feature = "benchmark")]
-                add_time!(bench, "pre pull");
+            if fairness_active {
+                let fairness_queries = WORKER_PULL_QUERIES_FAIRNESS.load();
+                let overloaded_slice: &[String] = overloaded.as_slice();
+                for query in fairness_queries.iter() {
+                    #[cfg(feature = "benchmark")]
+                    add_time!(bench, "pre pull (fairness)");
 
-                let r = sqlx::query_as::<_, PulledJob>(query)
-                    .bind(worker_name)
-                    .fetch_optional(db)
-                    .await?;
+                    let r = sqlx::query_as::<_, PulledJob>(query)
+                        .bind(worker_name)
+                        .bind(overloaded_slice)
+                        .fetch_optional(db)
+                        .await?;
 
-                #[cfg(feature = "benchmark")]
-                add_time!(bench, "post pull");
+                    #[cfg(feature = "benchmark")]
+                    add_time!(bench, "post pull (fairness)");
 
-                if let Some(pulled_job) = r {
-                    // tracing::info!("pulled job: {:?}", instant.elapsed().as_micros());
-
-                    highest_priority_job = Some(pulled_job);
-                    break;
+                    if let Some(pulled_job) = r {
+                        highest_priority_job = Some(pulled_job);
+                        break;
+                    }
+                }
+            }
+
+            if highest_priority_job.is_none() {
+                // Standard pull path. Also acts as the fallback when fairness filtered
+                // out every candidate: prefer running a capped workspace's job over
+                // leaving a worker idle. The cap re-engages on the next refresh as
+                // soon as the workspace's footprint exceeds the threshold again.
+                for query in queries.iter() {
+                    // tracing::info!("Pulling job with query: {}", query);
+                    // let instant = std::time::Instant::now();
+
+                    #[cfg(feature = "benchmark")]
+                    add_time!(bench, "pre pull");
+
+                    let r = sqlx::query_as::<_, PulledJob>(query)
+                        .bind(worker_name)
+                        .fetch_optional(db)
+                        .await?;
+
+                    #[cfg(feature = "benchmark")]
+                    add_time!(bench, "post pull");
+
+                    if let Some(pulled_job) = r {
+                        // tracing::info!("pulled job: {:?}", instant.elapsed().as_micros());
+
+                        highest_priority_job = Some(pulled_job);
+                        break;
+                    }
+                    // else continue pulling for lower priority tags
                 }
-                // else continue pulling for lower priority tags
             }
 
             // #[cfg(feature = "benchmark")]

backend/windmill-queue/src/lib.rs

@@ -14,6 +14,7 @@ pub mod schedule;
 pub use jobs::*;
 pub mod flow_status;
 pub mod tags;
+pub mod workspace_fairness;
 
 #[cfg(feature = "cloud")]
 pub mod cloud_usage;

backend/windmill-queue/src/workspace_fairness.rs

@@ -0,0 +1,262 @@
+//! Per-workspace fairness for the shared worker pool (cloud-only).
+//!
+//! On `app.windmill.dev` the cluster runs a single default worker group, so a
+//! single workspace flooding the queue with jobs can degrade quality of service
+//! for everyone else. This module computes the set of "overloaded" workspaces
+//! that should be temporarily excluded from the pull query.
+//!
+//! ## Detection signal
+//!
+//! A workspace is overloaded when, over the last `WORKSPACE_FAIRNESS_DURATION_SECS`
+//! seconds, it has accounted for at least `WORKSPACE_FAIRNESS_MAX_PERCENT`% of
+//! cluster activity. "Cluster activity" counts both currently-running jobs and
+//! jobs completed within the window — this captures workspaces hogging slots
+//! with long-running jobs **and** workspaces spamming many small short-lived
+//! jobs (where no individual job's `started_at` is old, but the aggregate
+//! throughput share dominates).
+//!
+//! ## Coordinated refresh
+//!
+//! The aggregation runs **at most once every `refresh_interval` seconds
+//! cluster-wide**, regardless of fleet size. A single `UPDATE` statement on
+//! `background_task_state` does double duty:
+//!   1. The `WHERE updated_at < now() - $interval` predicate, combined with
+//!      row-level locking, ensures only the first process to commit per cycle
+//!      actually recomputes the value. Other processes that race in see the
+//!      `WHERE` re-evaluated against the now-fresh row and update zero rows.
+//!   2. The same round trip falls through to a plain `SELECT` (via
+//!      `UNION ALL ... LIMIT 1`) so every caller reads the current value.
+//!
+//! Each process mirrors the result into [`WORKSPACE_FAIRNESS_OVERLOADED`]
+//! which the pull path reads at near-zero cost.
+//!
+//! ## Cloud gating
+//!
+//! The feature is hard-gated to `CLOUD_HOSTED=true` **and** `BASE_URL` matching
+//! `app.windmill.dev` (belt-and-suspenders against an on-prem instance importing
+//! cloud's `global_settings` row). When either check fails, [`maybe_refresh_overloaded`]
+//! and the pull-side dispatch both treat the feature as disabled.
+
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Duration;
+
+use sqlx::{Pool, Postgres};
+
+use windmill_common::error::Result;
+use windmill_common::worker::{
+    is_cloud_production_host, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
+    WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS, WORKSPACE_FAIRNESS_MAX_PERCENT,
+    WORKSPACE_FAIRNESS_MIN_TOTAL, WORKSPACE_FAIRNESS_OVERLOADED,
+};
+
+pub const TASK_STATE_NAME: &str = "workspace_fairness";
+
+/// Refresh interval when no workspace is currently capped. Slower cadence to
+/// keep DB load minimal during normal operation.
+const IDLE_REFRESH_SECS: u32 = 5;
+
+/// Refresh interval when at least one workspace is capped. Faster cadence so
+/// the cap lifts promptly once load drops below threshold.
+const ACTIVE_REFRESH_SECS: u32 = 2;
+
+/// Hard cap on the size of the overloaded list bound into the pull query.
+const MAX_OVERLOADED_RETURNED: i64 = 64;
+
+/// Whether the feature can be active in this process. Combined gate:
+///   - `WORKSPACE_FAIRNESS_ENABLED` setting toggled on, AND
+///   - `CLOUD_HOSTED=true`, AND
+///   - `BASE_URL` host is the production cloud host.
+fn fairness_active() -> bool {
+    WORKSPACE_FAIRNESS_ENABLED.load(Ordering::Relaxed) && is_cloud_production_host()
+}
+
+#[derive(serde::Deserialize)]
+struct FairnessState {
+    #[serde(default)]
+    overloaded: Vec<String>,
+}
+
+/// Lazy, non-blocking refresh entry point called from the pull path.
+///
+/// Cost on the hot path: one atomic load, optionally one compare-exchange. If
+/// this process wins the per-interval CAS, the actual refresh is spawned as a
+/// `tokio` task — the caller does not wait on it.
+pub fn maybe_refresh_overloaded(db: &Pool<Postgres>) {
+    if !fairness_active() {
+        // Drain the cached list so the dispatch in jobs.rs falls back to the
+        // unmodified pull queries within at most one pull cycle.
+        if !WORKSPACE_FAIRNESS_OVERLOADED.load().is_empty() {
+            WORKSPACE_FAIRNESS_OVERLOADED.store(Arc::new(vec![]));
+        }
+        return;
+    }
+
+    let interval_us = current_refresh_interval_micros();
+    let now_us = chrono::Utc::now().timestamp_micros();
+    let last = WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS.load(Ordering::Relaxed);
+    if now_us.saturating_sub(last) < interval_us {
+        return;
+    }
+    // Single in-flight refresh per process per cycle. If someone beat us, give up.
+    if WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS
+        .compare_exchange(last, now_us, Ordering::Relaxed, Ordering::Relaxed)
+        .is_err()
+    {
+        return;
+    }
+
+    let db = db.clone();
+    tokio::spawn(async move {
+        match tokio::time::timeout(Duration::from_secs(5), refresh_overloaded(&db)).await {
+            Ok(Ok(())) => {}
+            // On failure, leave `LAST_REFRESH_MICROS` set to `now_us` (already done by the CAS
+            // above). The next attempt therefore has to wait a full `current_refresh_interval`
+            // — exactly the same cooldown as a successful refresh. Previously we wrote `0`
+            // here, which removed the rate limit entirely and let every subsequent pull spawn
+            // a fresh refresh task while the DB was under pressure (precisely the moment we
+            // most need to back off).
+            Ok(Err(e)) => {
+                tracing::warn!("workspace fairness refresh failed: {e:#}");
+            }
+            Err(_) => {
+                tracing::warn!("workspace fairness refresh timed out after 5s");
+            }
+        }
+    });
+}
+
+fn current_refresh_interval_micros() -> i64 {
+    let secs = if WORKSPACE_FAIRNESS_OVERLOADED.load().is_empty() {
+        IDLE_REFRESH_SECS
+    } else {
+        ACTIVE_REFRESH_SECS
+    };
+    (secs as i64) * 1_000_000
+}
+
+/// Run the coordinated refresh.
+///
+/// The previous implementation used a single `INSERT ... ON CONFLICT DO UPDATE
+/// WHERE updated_at < ...` statement, which had a fatal flaw: Postgres evaluates
+/// the `VALUES` clause (including the expensive `v2_job_queue ∪ v2_job_completed`
+/// aggregation inlined there) **for every contender** to build the proposed row,
+/// before the conflict-row check decides whether to actually apply the update.
+/// So every worker process re-ran the heavy aggregation each cycle, and the
+/// claimed "one heavy aggregation per cycle cluster-wide" property did not hold.
+///
+/// This version splits the refresh into three small statements:
+///   1. Claim: a cheap upsert with only constant `VALUES`. Returns `Some(...)`
+///      iff this process won the right to refresh (row was either missing or
+///      had a stale `updated_at`).
+///   2. Winner-only: an `UPDATE ... SET value = ...` whose `SET` expression
+///      contains the heavy aggregation. Postgres evaluates `SET` per row
+///      matching `WHERE`; we only issue it when `won`, so the aggregation runs
+///      exactly once per refresh cycle cluster-wide.
+///   3. Read: every caller reads the current value (winner sees its own fresh
+///      write; losers see whatever the winner-from-this-or-the-prior-cycle
+///      wrote).
+async fn refresh_overloaded(db: &Pool<Postgres>) -> Result<()> {
+    let duration_secs = WORKSPACE_FAIRNESS_DURATION_SECS
+        .load(Ordering::Relaxed)
+        .clamp(1, i32::MAX as u32) as i32;
+    let max_percent = WORKSPACE_FAIRNESS_MAX_PERCENT
+        .load(Ordering::Relaxed)
+        .clamp(1, 100) as i64;
+    let min_total = WORKSPACE_FAIRNESS_MIN_TOTAL.load(Ordering::Relaxed) as i64;
+    // Use the tighter of the two intervals as the cluster-wide guard. The
+    // slower idle cadence is enforced by the per-process CAS gate in
+    // `maybe_refresh_overloaded`; the DB-side guard only needs to prevent
+    // two processes from racing into a refresh at the same time.
+    let refresh_secs = ACTIVE_REFRESH_SECS as i32;
+
+    // Step 1: claim. The VALUES clause is all constants — Postgres has no
+    // expensive work to do for either the insert-side or the conflict-side.
+    // Returns Some(true) for the unique winner per cycle, None for losers.
+    let won = sqlx::query_scalar::<_, bool>(
+        r#"
+        INSERT INTO background_task_state (name, value, running, owner, updated_at)
+        VALUES ($1, '{"overloaded":[]}'::jsonb, false, NULL, NOW())
+        ON CONFLICT (name) DO UPDATE
+            SET updated_at = NOW()
+            WHERE background_task_state.updated_at
+                < NOW() - make_interval(secs => $2::int)
+        RETURNING true
+        "#,
+    )
+    .bind(TASK_STATE_NAME)
+    .bind(refresh_secs)
+    .fetch_optional(db)
+    .await?
+    .is_some();
+
+    // Step 2: winner-only aggregation + value write. `SET` is evaluated per
+    // updated row, so issuing this statement only when `won` guarantees the
+    // expensive aggregation never runs for a loser.
+    if won {
+        sqlx::query(
+            r#"
+            UPDATE background_task_state
+            SET value = jsonb_build_object('overloaded', (
+                WITH active AS (
+                    SELECT workspace_id FROM v2_job_queue WHERE running = true
+                    UNION ALL
+                    SELECT workspace_id FROM v2_job_completed
+                     WHERE completed_at > NOW() - make_interval(secs => $2::int)
+                ),
+                per_ws AS (
+                    SELECT workspace_id, COUNT(*)::int8 AS c FROM active GROUP BY 1
+                ),
+                total AS (SELECT SUM(c)::int8 AS t FROM per_ws)
+                SELECT COALESCE(jsonb_agg(workspace_id ORDER BY c DESC), '[]'::jsonb)
+                FROM (
+                    SELECT workspace_id, c FROM per_ws, total
+                    WHERE total.t >= $3
+                      AND per_ws.c * 100 >= $4 * total.t
+                    ORDER BY c DESC
+                    LIMIT $5
+                ) capped
+            ))
+            WHERE name = $1
+            "#,
+        )
+        .bind(TASK_STATE_NAME)
+        .bind(duration_secs)
+        .bind(min_total)
+        .bind(max_percent)
+        .bind(MAX_OVERLOADED_RETURNED)
+        .execute(db)
+        .await?;
+    }
+
+    // Step 3: read current state (winner reads its own fresh write).
+    let row: Option<serde_json::Value> =
+        sqlx::query_scalar("SELECT value FROM background_task_state WHERE name = $1")
+            .bind(TASK_STATE_NAME)
+            .fetch_optional(db)
+            .await?;
+
+    let new_list: Vec<String> = match row {
+        Some(value) => match serde_json::from_value::<FairnessState>(value) {
+            Ok(s) => s.overloaded,
+            Err(e) => {
+                tracing::warn!("workspace fairness state parse error: {e:#}");
+                vec![]
+            }
+        },
+        None => vec![],
+    };
+
+    let prev = WORKSPACE_FAIRNESS_OVERLOADED.load();
+    if **prev != new_list {
+        tracing::info!(
+            "workspace fairness overloaded set changed: {} -> {} ({:?})",
+            prev.len(),
+            new_list.len(),
+            &new_list,
+        );
+        WORKSPACE_FAIRNESS_OVERLOADED.store(Arc::new(new_list));
+    }
+
+    Ok(())
+}

frontend/src/lib/components/instanceSettings.ts

@@ -303,6 +303,49 @@ export const settings: Record<string, Setting[]> = {
 			storage: 'setting',
 			ee_only: 'You can only adjust this setting to above 30 days in the EE version',
 			cloudonly: false
+		},
+		{
+			label: 'Workspace fairness — enabled',
+			description:
+				'Cloud-only safeguard against a single workspace dominating the shared worker pool. When a workspace accounts for at least <em>Workspace fairness — max percent</em> of cluster activity over the last <em>Workspace fairness — duration</em> seconds, the pull query temporarily excludes that workspace until its share drops back below the threshold. Idle workers always fall back to running its jobs, so capping never starves the queue.',
+			key: 'workspace_fairness_enabled',
+			fieldType: 'boolean',
+			storage: 'setting',
+			cloudonly: true,
+			hideInQuickSetup: true
+		},
+		{
+			label: 'Workspace fairness — max percent',
+			description:
+				'Maximum percentage of cluster activity a single workspace may sustain before being temporarily excluded from the pull query. Default 50.',
+			key: 'workspace_fairness_max_percent',
+			fieldType: 'number',
+			placeholder: '50',
+			storage: 'setting',
+			cloudonly: true,
+			hideInQuickSetup: true
+		},
+		{
+			label: 'Workspace fairness — duration (seconds)',
+			description:
+				'Rolling window used to measure workspace share. Activity = currently running jobs ∪ jobs completed in the last N seconds. Default 10.',
+			key: 'workspace_fairness_duration_secs',
+			fieldType: 'seconds',
+			placeholder: '10',
+			storage: 'setting',
+			cloudonly: true,
+			hideInQuickSetup: true
+		},
+		{
+			label: 'Workspace fairness — minimum total jobs',
+			description:
+				'Cap is only applied when cluster-wide activity exceeds this floor. Prevents over-eager capping on small clusters or quiet periods. Default 4.',
+			key: 'workspace_fairness_min_total_jobs',
+			fieldType: 'number',
+			placeholder: '4',
+			storage: 'setting',
+			cloudonly: true,
+			hideInQuickSetup: true
 		}
 	],
 	'Object Storage': [

frontend/src/lib/consts.ts

@@ -50,6 +50,11 @@ export const DEFAULT_TAGS_WORKSPACES_SETTING = 'default_tags_workspaces'
 export const FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING = 'fork_workspace_tag_append_fork_suffix'
 export const PREVIEW_TAGS_OVERRIDE_SETTING = 'preview_tags_override'
 
+export const WORKSPACE_FAIRNESS_ENABLED_SETTING = 'workspace_fairness_enabled'
+export const WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING = 'workspace_fairness_max_percent'
+export const WORKSPACE_FAIRNESS_DURATION_SECS_SETTING = 'workspace_fairness_duration_secs'
+export const WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING = 'workspace_fairness_min_total_jobs'
+
 export const WORKSPACE_SLACK_BOT_TOKEN_PATH = 'f/slack_bot/bot_token'
 
 export const POSTGRES_TYPES = [