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

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,52 @@ pub async fn load_preview_tags_override(db: &DB) -> error::Result<()> {
     Ok(())
 }
 
+pub async fn load_workspace_fairness_enabled(db: &DB) -> error::Result<()> {
+    let v = load_value_from_global_settings(db, WORKSPACE_FAIRNESS_ENABLED_SETTING).await;
+    let new_enabled = match v {
+        Ok(Some(serde_json::Value::Bool(t))) => t,
+        _ => 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<()> {
+    let v = load_value_from_global_settings(db, WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING).await;
+    if let Ok(Some(serde_json::Value::Number(n))) = v {
+        if let Some(u) = n.as_u64() {
+            WORKSPACE_FAIRNESS_MAX_PERCENT.store(u.clamp(1, 100) as u32, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
+pub async fn load_workspace_fairness_duration_secs(db: &DB) -> error::Result<()> {
+    let v = load_value_from_global_settings(db, WORKSPACE_FAIRNESS_DURATION_SECS_SETTING).await;
+    if let Ok(Some(serde_json::Value::Number(n))) = v {
+        if let Some(u) = n.as_u64() {
+            WORKSPACE_FAIRNESS_DURATION_SECS.store(u.max(1) as u32, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
+pub async fn load_workspace_fairness_min_total(db: &DB) -> error::Result<()> {
+    let v = load_value_from_global_settings(db, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING).await;
+    if let Ok(Some(serde_json::Value::Number(n))) = v {
+        if let Some(u) = n.as_u64() {
+            WORKSPACE_FAIRNESS_MIN_TOTAL.store(u as u32, 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,15 @@ 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::CLOUD_HOSTED,
+    BASE_URL,
 };
 use windmill_common::{error::to_anyhow, PgDatabase};
 
@@ -446,6 +451,40 @@ 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 {
+    if !*CLOUD_HOSTED {
+        return false;
+    }
+    let base = BASE_URL.load();
+    let s = base.as_str();
+    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 == "app.windmill.dev"
+}
+
 pub async fn set_global_setting(
     Extension(db): Extension<DB>,
     authed: ApiAuthed,
@@ -468,6 +507,17 @@ 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`.
+    if is_workspace_fairness_setting(&key) && !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 {

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,8 +237,21 @@ 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());
@@ -520,17 +533,40 @@ 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 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,251 @@
+//! 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::{
+    CLOUD_HOSTED, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
+    WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS, WORKSPACE_FAIRNESS_MAX_PERCENT,
+    WORKSPACE_FAIRNESS_MIN_TOTAL, WORKSPACE_FAIRNESS_OVERLOADED,
+};
+use windmill_common::BASE_URL;
+
+pub const TASK_STATE_NAME: &str = "workspace_fairness";
+
+const APP_WINDMILL_DEV_HOST: &str = "app.windmill.dev";
+
+/// 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;
+
+/// Read `BASE_URL` and check whether the cluster's base URL points at
+/// `app.windmill.dev`. The setting is configured by the admin via `/base_url`
+/// and is empty until the first settings load completes.
+fn is_app_windmill_dev() -> bool {
+    let base = BASE_URL.load();
+    let s = base.as_str();
+    if s.is_empty() {
+        return false;
+    }
+    // Strip scheme then check the leading host segment. Avoids a new dependency
+    // for what is a fixed-string match against one production host.
+    let after_scheme = s
+        .strip_prefix("https://")
+        .or_else(|| s.strip_prefix("http://"))
+        .unwrap_or(s);
+    let host = after_scheme.split('/').next().unwrap_or("");
+    let host = host.split(':').next().unwrap_or("");
+    host == APP_WINDMILL_DEV_HOST
+}
+
+/// 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 `app.windmill.dev`.
+pub fn fairness_active() -> bool {
+    WORKSPACE_FAIRNESS_ENABLED.load(Ordering::Relaxed) && *CLOUD_HOSTED && is_app_windmill_dev()
+}
+
+#[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(())) => {}
+            Ok(Err(e)) => {
+                tracing::warn!("workspace fairness refresh failed: {e:#}");
+                // Reset the gate so the next pull can retry, but with the
+                // current "active" interval as a floor to avoid stampeding.
+                WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS.store(0, Ordering::Relaxed);
+            }
+            Err(_) => {
+                tracing::warn!("workspace fairness refresh timed out after 5s");
+                WORKSPACE_FAIRNESS_LAST_REFRESH_MICROS.store(0, Ordering::Relaxed);
+            }
+        }
+    });
+}
+
+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. Idempotent across processes: only one update
+/// per cycle commits real work, everyone else does a cheap read.
+async fn refresh_overloaded(db: &Pool<Postgres>) -> Result<()> {
+    let duration_secs = WORKSPACE_FAIRNESS_DURATION_SECS
+        .load(Ordering::Relaxed)
+        .max(1) 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 refresh predicate; the slower
+    // idle cadence is enforced client-side by the CAS gate, while the SQL guard
+    // protects against multiple in-flight refreshes from different processes.
+    let refresh_secs = ACTIVE_REFRESH_SECS as i32;
+
+    // Single statement: either we win the row-lock and recompute, or we read
+    // whatever was just written by the winner.
+    let row: Option<serde_json::Value> = sqlx::query_scalar(
+        r#"
+        WITH refreshed AS (
+            INSERT INTO background_task_state (name, value, running, owner, updated_at)
+            VALUES (
+                $1,
+                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)
+                    ),
+                    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
+                )),
+                false,
+                NULL,
+                NOW()
+            )
+            ON CONFLICT (name) DO UPDATE
+                SET value = EXCLUDED.value,
+                    updated_at = NOW()
+                WHERE background_task_state.updated_at
+                    < NOW() - make_interval(secs => $6)
+            RETURNING value
+        )
+        SELECT value FROM refreshed
+        UNION ALL
+        SELECT value FROM background_task_state WHERE name = $1
+        LIMIT 1
+        "#,
+    )
+    .bind(TASK_STATE_NAME)
+    .bind(duration_secs)
+    .bind(min_total)
+    .bind(max_percent)
+    .bind(MAX_OVERLOADED_RETURNED)
+    .bind(refresh_secs)
+    .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 = [