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/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());
+                    if let Some(pulled_job) = r {
+                        highest_priority_job = Some(pulled_job);
+                        break;
+                    }
+                }
+            }
 
-                    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(())
+}