diff --git a/backend/.sqlx/query-1ce0728a6b0942fecf8ce5f4a06857a2601b10b057877c3b2795b8ea9c928c1b.json b/backend/.sqlx/query-1ce0728a6b0942fecf8ce5f4a06857a2601b10b057877c3b2795b8ea9c928c1b.json new file mode 100644 index 0000000000..7a74d18f48 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-750eb7365e2590a42d9f3c4cbbf193ea4129874c36e2c334e9e73f4b4c54ad07.json b/backend/.sqlx/query-750eb7365e2590a42d9f3c4cbbf193ea4129874c36e2c334e9e73f4b4c54ad07.json new file mode 100644 index 0000000000..17f12041c4 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-d977778801cc5efdfc4051d5b77d75eda5fc20669d598f0dbbb301030e58d702.json b/backend/.sqlx/query-d977778801cc5efdfc4051d5b77d75eda5fc20669d598f0dbbb301030e58d702.json new file mode 100644 index 0000000000..420092f7e8 --- /dev/null +++ b/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" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index 6a12250e53..aa895df3e7 100644 --- a/backend/src/main.rs +++ b/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; } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 86f26aeb33..0a3333e58c 100644 --- a/backend/src/monitor.rs +++ b/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; diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index bbe7dbb9b5..2714d31276 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/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. diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 2c936ff0aa..b9dd29c00a 100644 --- a/backend/windmill-common/src/worker.rs +++ b/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> = arc_swap::ArcSwap::from_pointee(None); pub static ref INDEXER_CONFIG: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(TantivyIndexerSettings::default()); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 934b9b50b4..bea3bc1008 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/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 { + 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>, diff --git a/backend/windmill-queue/tests/workspace_queue_depth_test.rs b/backend/windmill-queue/tests/workspace_queue_depth_test.rs new file mode 100644 index 0000000000..f6969afc4c --- /dev/null +++ b/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, 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) { + 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) { + 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" + ); +} diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index b58eb329f7..3b892049c6 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -438,6 +438,18 @@ export const settings: Record = { 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': [