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.

backend/src/monitor.rs

@@ -562,12 +562,29 @@ 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;
+
 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,
-    };
+    // 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.
@@ -592,7 +609,11 @@ 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);
+            // 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 clamped = u.clamp(1, WORKSPACE_FAIRNESS_DURATION_SECS_MAX) as u32;
+            WORKSPACE_FAIRNESS_DURATION_SECS.store(clamped, Ordering::Relaxed);
         }
     }
     Ok(())
@@ -602,7 +623,12 @@ 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);
+            // Clamp before narrowing — same reasoning as `_duration_secs`, just for the
+            // counting threshold rather than the interval.
+            WORKSPACE_FAIRNESS_MIN_TOTAL.store(
+                u.min(WORKSPACE_FAIRNESS_MIN_TOTAL_MAX) as u32,
+                Ordering::Relaxed,
+            );
         }
     }
     Ok(())

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

@@ -61,8 +61,7 @@ use windmill_common::{
     },
     instance_config::{self, ApplyMode, InstanceConfig},
     server::Smtp,
-    worker::CLOUD_HOSTED,
-    BASE_URL,
+    worker::is_cloud_production_host,
 };
 use windmill_common::{error::to_anyhow, PgDatabase};
 
@@ -466,23 +465,7 @@ fn is_workspace_fairness_setting(key: &str) -> bool {
 /// 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"
+    is_cloud_production_host()
 }
 
 pub async fn set_global_setting(
@@ -776,6 +759,23 @@ 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.
+        let touched_fairness_keys = settings_diff
+            .upserts
+            .keys()
+            .chain(settings_diff.deletes.iter())
+            .any(|k| is_workspace_fairness_setting(k));
+        if touched_fairness_keys && !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/worker.rs

@@ -258,6 +258,9 @@ lazy_static::lazy_static! {
 
 
     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()
@@ -302,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);
@@ -537,7 +568,10 @@ pub fn make_pull_query(tags: &[String]) -> String {
 // 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 {
+//
+// `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

backend/windmill-queue/src/workspace_fairness.rs

@@ -45,16 +45,13 @@ use sqlx::{Pool, Postgres};
 
 use windmill_common::error::Result;
 use windmill_common::worker::{
-    CLOUD_HOSTED, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
+    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,
 };
-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;
@@ -66,32 +63,12 @@ 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`.
+///   - `BASE_URL` host is the production cloud host.
 pub fn fairness_active() -> bool {
-    WORKSPACE_FAIRNESS_ENABLED.load(Ordering::Relaxed) && *CLOUD_HOSTED && is_app_windmill_dev()
+    WORKSPACE_FAIRNESS_ENABLED.load(Ordering::Relaxed) && is_cloud_production_host()
 }
 
 #[derive(serde::Deserialize)]
@@ -133,15 +110,17 @@ pub fn maybe_refresh_overloaded(db: &Pool<Postgres>) {
     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:#}");
-                // 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);
             }
         }
     });
@@ -156,74 +135,106 @@ fn current_refresh_interval_micros() -> i64 {
     (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.
+/// 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)
-        .max(1) as i32;
+        .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 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.
+    // 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;
 
-    // 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(
+    // 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#"
-        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
+        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(duration_secs)
-    .bind(min_total)
-    .bind(max_percent)
-    .bind(MAX_OVERLOADED_RETURNED)
     .bind(refresh_secs)
     .fetch_optional(db)
-    .await?;
+    .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) {