feat(jobs): cap total queued jobs per workspace on cloud (#10218)

* feat(jobs): cap total queued jobs per workspace on cloud

A workspace could flood the queue with an unbounded number of jobs across
many concurrency keys and scripts (or keyless jobs), which the per-key
cap from #10197 does not bound. Add a companion instance-wide ceiling on
a workspace's total queued jobs.

check_workspace_queue_cap rejects a push once the workspace has
WORKSPACE_MAX_QUEUED_JOBS (default 20000, superadmin-configurable, 0 to
disable) jobs queued, cloud-only and runtime-gated on CLOUD_HOSTED like
the per-key cap. It runs on every push, so it applies even to premium
workspaces and catches parallel for-loop floods. Jobs already queued
still drain; only new pushes past the ceiling are rejected, so an
in-flight flow only fails to push further work while at the ceiling.

The setting loader self-gates on CLOUD_HOSTED so it is never loaded off
cloud, from initial load or a settings-change reload. The depth count is
bounded by the cap via LIMIT so a runaway backlog never costs an
unbounded scan on the push path.

* docs(jobs): note the workspace cap is a soft ceiling and the depth helper is count-only

Records the two review points as constraints: the cap does not serialize
admission (a soft ceiling by design, like the per-key cap), and
workspace_queue_depth is pub only for the test, returns a count not job
data, and leaves authorization to the caller.

backend/.sqlx/query-1ce0728a6b0942fecf8ce5f4a06857a2601b10b057877c3b2795b8ea9c928c1b.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag, running)\n             VALUES ($1, $2, now() + ($3::bigint::text || ' s')::interval, 'other', $4)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Int8",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1ce0728a6b0942fecf8ce5f4a06857a2601b10b057877c3b2795b8ea9c928c1b"
+}

backend/.sqlx/query-750eb7365e2590a42d9f3c4cbbf193ea4129874c36e2c334e9e73f4b4c54ad07.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job (id, workspace_id, tag) VALUES ($1, $2, 'other')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "750eb7365e2590a42d9f3c4cbbf193ea4129874c36e2c334e9e73f4b4c54ad07"
+}

backend/.sqlx/query-d977778801cc5efdfc4051d5b77d75eda5fc20669d598f0dbbb301030e58d702.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT count(*) FROM (\n            SELECT 1 FROM v2_job_queue\n            WHERE workspace_id = $1 AND running = false\n            LIMIT $2\n        ) s",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "d977778801cc5efdfc4051d5b77d75eda5fc20669d598f0dbbb301030e58d702"
+}

backend/src/main.rs

@@ -64,7 +64,7 @@ use windmill_common::{
         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,
-        WORKSPACE_REGISTRIES_SETTING,
+        WORKSPACE_MAX_QUEUED_JOBS_SETTING, WORKSPACE_REGISTRIES_SETTING,
     },
     scripts::ScriptLang,
     stats_oss::schedule_stats,
@@ -127,16 +127,16 @@ use crate::monitor::{
     load_preview_tags_override, load_require_preexisting_user, load_retention_period_overrides,
     load_tag_per_workspace_enabled, load_tag_per_workspace_workspaces,
     load_workspace_fairness_duration_secs, 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, reload_critical_error_channels_setting,
-    reload_extra_pip_index_url_setting, reload_http_route_workspaced_route_setting,
-    reload_hub_api_secret_setting, reload_hub_base_url_setting,
-    reload_instance_events_webhook_setting, reload_job_default_timeout_setting,
-    reload_job_isolation_setting, reload_jwt_secret_setting, reload_license_key,
-    reload_npm_config_registry_setting, reload_nsjail_tmp_backing_setting,
+    load_workspace_fairness_max_percent, load_workspace_fairness_min_total,
+    load_workspace_max_queued_jobs, 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,
+    reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
+    reload_http_route_workspaced_route_setting, reload_hub_api_secret_setting,
+    reload_hub_base_url_setting, reload_instance_events_webhook_setting,
+    reload_job_default_timeout_setting, reload_job_isolation_setting, reload_jwt_secret_setting,
+    reload_license_key, reload_npm_config_registry_setting, reload_nsjail_tmp_backing_setting,
     reload_nsjail_tmpfs_size_setting, reload_otel_tracing_proxy_setting,
     reload_pip_index_url_setting, reload_retention_period_setting,
     reload_sandbox_image_cache_max_setting, reload_sandbox_image_default_registry_setting,
@@ -1892,6 +1892,11 @@ async fn process_notify_event(
                         tracing::error!("Error loading concurrency key max queued: {e:#}");
                     }
                 }
+                WORKSPACE_MAX_QUEUED_JOBS_SETTING => {
+                    if let Err(e) = load_workspace_max_queued_jobs(db).await {
+                        tracing::error!("Error loading workspace max queued jobs: {e:#}");
+                    }
+                }
                 SMTP_SETTING => {
                     reload_smtp_config(db).await;
                 }

backend/src/monitor.rs

@@ -74,6 +74,7 @@ use windmill_common::{
         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,
+        WORKSPACE_MAX_QUEUED_JOBS_SETTING,
     },
     indexer::load_indexer_config,
     jobs::delete_jobs,
@@ -92,7 +93,8 @@ use windmill_common::{
         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, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
-        WORKSPACE_FAIRNESS_MAX_PERCENT, WORKSPACE_FAIRNESS_MIN_TOTAL,
+        WORKSPACE_FAIRNESS_MAX_PERCENT, WORKSPACE_FAIRNESS_MIN_TOTAL, WORKSPACE_MAX_QUEUED_JOBS,
+        WORKSPACE_MAX_QUEUED_JOBS_DEFAULT,
     },
     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,
@@ -299,6 +301,9 @@ pub async fn initial_load(
             if let Err(e) = load_concurrency_key_max_queued(db).await {
                 tracing::error!("Error loading concurrency key max queued: {e:#}");
             }
+            if let Err(e) = load_workspace_max_queued_jobs(db).await {
+                tracing::error!("Error loading workspace max queued jobs: {e:#}");
+            }
         }
     }
 
@@ -743,6 +748,39 @@ pub async fn load_concurrency_key_max_queued(db: &DB) -> error::Result<()> {
     Ok(())
 }
 
+pub async fn load_workspace_max_queued_jobs(db: &DB) -> error::Result<()> {
+    // Only the cloud enforces this cap, so never spend the query off-cloud, from any call site.
+    if !*CLOUD_HOSTED {
+        return Ok(());
+    }
+    // Same Err / None / invalid policy as load_concurrency_key_max_queued: 0 disables.
+    match load_value_from_global_settings(db, WORKSPACE_MAX_QUEUED_JOBS_SETTING).await? {
+        Some(serde_json::Value::Number(n)) => {
+            let v = n
+                .as_u64()
+                .map(|u| u.min(CONCURRENCY_KEY_MAX_QUEUED_MAX) as u32)
+                .unwrap_or_else(|| {
+                    tracing::warn!(
+                        "{WORKSPACE_MAX_QUEUED_JOBS_SETTING}={n} is not a non-negative integer, \
+                         falling back to {WORKSPACE_MAX_QUEUED_JOBS_DEFAULT}. Set 0 to disable."
+                    );
+                    WORKSPACE_MAX_QUEUED_JOBS_DEFAULT
+                });
+            WORKSPACE_MAX_QUEUED_JOBS.store(v, Ordering::Relaxed);
+        }
+        other => {
+            if let Some(v) = other {
+                tracing::warn!(
+                    "{WORKSPACE_MAX_QUEUED_JOBS_SETTING}={v} is not a number, falling back to \
+                     {WORKSPACE_MAX_QUEUED_JOBS_DEFAULT}. Set 0 to disable."
+                );
+            }
+            WORKSPACE_MAX_QUEUED_JOBS.store(WORKSPACE_MAX_QUEUED_JOBS_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-common/src/global_settings.rs

@@ -121,6 +121,11 @@ pub const WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING: &str = "workspace_fairness_min_t
 // `check_concurrency_key_queue_cap`.
 pub const CONCURRENCY_KEY_MAX_QUEUED_SETTING: &str = "concurrency_key_max_queued_jobs";
 
+// Cloud-only ceiling on how many jobs a single workspace may have queued in
+// total, across every key and script. Applies even to premium workspaces. `0`
+// disables the cap. See `windmill-queue/src/jobs.rs`, `check_workspace_queue_cap`.
+pub const WORKSPACE_MAX_QUEUED_JOBS_SETTING: &str = "workspace_max_queued_jobs";
+
 /// Global settings an agent worker (a remote worker connected over HTTP instead
 /// of to the database) must NEVER read through
 /// `GET /api/agent_workers/get_global_setting/{key}`. Every other key is served.

backend/windmill-common/src/worker.rs

@@ -213,6 +213,10 @@ pub const MIN_PERIODIC_SCRIPT_INTERVAL_SECONDS: u64 = 60;
 /// Default for [`CONCURRENCY_KEY_MAX_QUEUED`]; also the value the setting loader restores when
 /// the setting is cleared or malformed.
 pub const CONCURRENCY_KEY_MAX_QUEUED_DEFAULT: u32 = 10_000;
+/// Default for [`WORKSPACE_MAX_QUEUED_JOBS`]; also the value the setting loader restores when
+/// the setting is cleared or malformed. A workspace spans many keys, so this sits well above
+/// the per-key cap.
+pub const WORKSPACE_MAX_QUEUED_JOBS_DEFAULT: u32 = 20_000;
 lazy_static::lazy_static! {
     pub static ref WORKER_GROUP: String = std::env::var("WORKER_GROUP").unwrap_or_else(|_| {
         #[cfg(not(feature = "enterprise"))]
@@ -347,6 +351,12 @@ lazy_static::lazy_static! {
     pub static ref CONCURRENCY_KEY_MAX_QUEUED: AtomicU32 =
         AtomicU32::new(CONCURRENCY_KEY_MAX_QUEUED_DEFAULT);
 
+    /// Cloud-only ceiling on the total number of jobs a workspace may have queued at once,
+    /// across every concurrency key and script. Guards against a workspace flooding the queue
+    /// generally (not just behind one key), including from parallel for-loops. `0` disables it.
+    pub static ref WORKSPACE_MAX_QUEUED_JOBS: AtomicU32 =
+        AtomicU32::new(WORKSPACE_MAX_QUEUED_JOBS_DEFAULT);
+
 
     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());

backend/windmill-queue/src/jobs.rs

@@ -6280,6 +6280,11 @@ async fn push_inner<'c, 'd>(
         )
         .unzip();
 
+    #[cfg(feature = "cloud")]
+    if *CLOUD_HOSTED {
+        check_workspace_queue_cap(&mut *tx, workspace_id).await?;
+    }
+
     if concurrency_settings.concurrent_limit.is_some() {
         let concurrency_key = resolve_concurrency_key(
             workspace_id,
@@ -6791,6 +6796,70 @@ async fn check_concurrency_key_queue_cap<'c>(
     Ok(())
 }
 
+/// Bounded count of a workspace's non-running queued jobs, capped at `limit` so the scan
+/// stops once the ceiling is reached rather than counting an entire runaway backlog.
+///
+/// Internal helper for `check_workspace_queue_cap`, `pub` only so the integration test can call
+/// it (like `concurrency_key_queue_depth`). Returns a count, not job contents; the caller is
+/// responsible for any authorization — it takes the workspace id as given.
+pub async fn workspace_queue_depth<'c>(
+    db: impl PgExecutor<'c>,
+    workspace_id: &str,
+    limit: i64,
+) -> Result<i64, Error> {
+    sqlx::query_scalar!(
+        "SELECT count(*) FROM (
+            SELECT 1 FROM v2_job_queue
+            WHERE workspace_id = $1 AND running = false
+            LIMIT $2
+        ) s",
+        workspace_id,
+        limit,
+    )
+    .fetch_one(db)
+    .warn_after_seconds(3)
+    .await
+    .map_err(|e| {
+        Error::internal_err(format!(
+            "Could not count queued jobs for workspace={workspace_id}: {e:#}"
+        ))
+    })
+    .map(|c| c.unwrap_or(0))
+}
+
+/// Rejects the push when the workspace already has `WORKSPACE_MAX_QUEUED_JOBS` jobs queued.
+///
+/// Caller must runtime-gate this on `*CLOUD_HOSTED`. It runs on every push (not only
+/// concurrency-limited ones) because a workspace can flood the queue across many keys or with
+/// keyless jobs. It caps *new* pushes past the ceiling; jobs already queued still drain, so an
+/// in-flight flow only ever fails to push further work while the workspace is at the ceiling.
+///
+/// This is a soft ceiling, like the per-key cap: the count and the insert are not serialized, so
+/// a burst of concurrent pushes can land a handful over the limit. That is fine and intentional
+/// — the cap exists to stop an unbounded runaway, not to enforce an exact quota, and a
+/// per-workspace lock on every push would add hot-path contention for no practical gain.
+#[cfg(feature = "cloud")]
+async fn check_workspace_queue_cap<'c>(
+    db: impl PgExecutor<'c>,
+    workspace_id: &str,
+) -> Result<(), Error> {
+    let cap = windmill_common::worker::WORKSPACE_MAX_QUEUED_JOBS
+        .load(std::sync::atomic::Ordering::Relaxed);
+    if cap == 0 {
+        return Ok(());
+    }
+    let cap = cap as i64;
+    let depth = workspace_queue_depth(db, workspace_id, cap).await?;
+    if depth >= cap {
+        return Err(Error::QuotaExceeded(format!(
+            "Too many jobs queued in workspace '{workspace_id}': at least {depth} jobs are \
+             already waiting and the instance limit is {cap}. Cancel the backlog or slow down \
+             whatever is creating jobs before pushing more."
+        )));
+    }
+    Ok(())
+}
+
 // pub async fn insert_debounce_key<'d, 'c>(
 //     workspace_id: &str,
 //     args: &PushArgs<'d>,

backend/windmill-queue/tests/workspace_queue_depth_test.rs

@@ -0,0 +1,71 @@
+//! Regression guard for `workspace_queue_depth`, which backs the cloud-only cap on how many
+//! jobs a single workspace may have queued in total.
+//!
+//! Run with:
+//!   cargo test -p windmill-queue --test workspace_queue_depth_test
+
+use sqlx::{Pool, Postgres};
+use uuid::Uuid;
+use windmill_queue::jobs::workspace_queue_depth;
+
+/// Queues `count` jobs in `workspace`, scheduled `offset_secs` from now, with `running` state.
+async fn seed(db: &Pool<Postgres>, workspace: &str, count: usize, offset_secs: i64, running: bool) {
+    for _ in 0..count {
+        let id = Uuid::new_v4();
+        sqlx::query!(
+            "INSERT INTO v2_job (id, workspace_id, tag) VALUES ($1, $2, 'other')",
+            id,
+            workspace,
+        )
+        .execute(db)
+        .await
+        .expect("seed v2_job");
+        sqlx::query!(
+            "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag, running)
+             VALUES ($1, $2, now() + ($3::bigint::text || ' s')::interval, 'other', $4)",
+            id,
+            workspace,
+            offset_secs,
+            running,
+        )
+        .execute(db)
+        .await
+        .expect("seed v2_job_queue");
+    }
+}
+
+/// The count is per workspace and ignores running jobs: a runaway in one workspace must not
+/// charge another, and jobs already executing are not backlog. Future-scheduled jobs count,
+/// since a concurrency-parked backlog is almost entirely future-dated.
+#[sqlx::test(migrations = "../migrations")]
+async fn scoped_to_workspace_and_ignores_running(db: Pool<Postgres>) {
+    seed(&db, "ws-a", 4, 3600, false).await; // waiting (parked in the future)
+    seed(&db, "ws-a", 3, -10, false).await; // waiting (due now)
+    seed(&db, "ws-a", 5, -10, true).await; // running — not backlog
+    seed(&db, "ws-b", 9, 3600, false).await; // a different workspace
+
+    let depth = workspace_queue_depth(&db, "ws-a", 1000)
+        .await
+        .expect("count depth");
+
+    assert_eq!(
+        depth, 7,
+        "only ws-a's 7 waiting jobs count; running jobs and ws-b must not"
+    );
+}
+
+/// The scan stops at `limit`, so a runaway backlog does not cost an unbounded count on every
+/// push. The cap only needs to know the depth has reached the ceiling.
+#[sqlx::test(migrations = "../migrations")]
+async fn bounded_by_limit(db: Pool<Postgres>) {
+    seed(&db, "ws-a", 50, -10, false).await;
+
+    let depth = workspace_queue_depth(&db, "ws-a", 10)
+        .await
+        .expect("count depth");
+
+    assert_eq!(
+        depth, 10,
+        "the count must stop at the limit, not scan all 50"
+    );
+}

frontend/src/lib/components/instanceSettings.ts

@@ -438,6 +438,18 @@ export const settings: Record<string, Setting[]> = {
 			cloudonly: true,
 			ee_only: '',
 			hideInQuickSetup: true
+		},
+		{
+			label: 'Max jobs queued per workspace',
+			description:
+				'Rejects new jobs once a workspace has this many queued in total, across every concurrency key and script. Guards against a single workspace flooding the queue generally, including from parallel for-loops. Applies even to premium workspaces. Jobs already queued still drain; only new pushes past the ceiling are rejected. Set 0 to disable. Default 20000.',
+			key: 'workspace_max_queued_jobs',
+			fieldType: 'number',
+			placeholder: '20000',
+			storage: 'setting',
+			cloudonly: true,
+			ee_only: '',
+			hideInQuickSetup: true
 		}
 	],
 	'Object Storage': [