diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index d29a18c691..e7ed0aee65 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - true, - true, - true, - true, - true, + false, + false, + false, + false, + false, true, true ] diff --git a/backend/.sqlx/query-0bd853be57b43a7820cc1d73d92b8babd50b1092d4ad70e022431bdcf32e378f.json b/backend/.sqlx/query-0bd853be57b43a7820cc1d73d92b8babd50b1092d4ad70e022431bdcf32e378f.json new file mode 100644 index 0000000000..f0e1c853d8 --- /dev/null +++ b/backend/.sqlx/query-0bd853be57b43a7820cc1d73d92b8babd50b1092d4ad70e022431bdcf32e378f.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job (id, workspace_id, tag) VALUES ($1, 'test-workspace', 'other')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "0bd853be57b43a7820cc1d73d92b8babd50b1092d4ad70e022431bdcf32e378f" +} diff --git a/backend/.sqlx/query-20ae664da5ebd70a6d37609f6b9a3e10d8d524fbaaa8eddf74f12a906b3c8306.json b/backend/.sqlx/query-20ae664da5ebd70a6d37609f6b9a3e10d8d524fbaaa8eddf74f12a906b3c8306.json new file mode 100644 index 0000000000..04e663f7c3 --- /dev/null +++ b/backend/.sqlx/query-20ae664da5ebd70a6d37609f6b9a3e10d8d524fbaaa8eddf74f12a906b3c8306.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag, running)\n VALUES ($1, 'test-workspace', now() + ($2::bigint::text || ' s')::interval, 'other', $3)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Int8", + "Bool" + ] + }, + "nullable": [] + }, + "hash": "20ae664da5ebd70a6d37609f6b9a3e10d8d524fbaaa8eddf74f12a906b3c8306" +} diff --git a/backend/.sqlx/query-521367aaea7beefe4ff9cdb66273f8e3cddfbaf598536f1753f0824f84604826.json b/backend/.sqlx/query-521367aaea7beefe4ff9cdb66273f8e3cddfbaf598536f1753f0824f84604826.json new file mode 100644 index 0000000000..9acef34c47 --- /dev/null +++ b/backend/.sqlx/query-521367aaea7beefe4ff9cdb66273f8e3cddfbaf598536f1753f0824f84604826.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM v2_job_queue", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "521367aaea7beefe4ff9cdb66273f8e3cddfbaf598536f1753f0824f84604826" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-694d787e06b5ef9fbb5ca1315652023ecc0b3122501b72ec7f825966419759ac.json b/backend/.sqlx/query-694d787e06b5ef9fbb5ca1315652023ecc0b3122501b72ec7f825966419759ac.json new file mode 100644 index 0000000000..5b69718bfd --- /dev/null +++ b/backend/.sqlx/query-694d787e06b5ef9fbb5ca1315652023ecc0b3122501b72ec7f825966419759ac.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(*) FROM (\n SELECT ck.job_id FROM concurrency_key ck\n WHERE ck.key = $1 AND ck.ended_at IS NULL\n LIMIT $2\n ) s WHERE EXISTS (\n SELECT 1 FROM v2_job_queue q WHERE q.id = s.job_id AND q.running = false\n )", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "694d787e06b5ef9fbb5ca1315652023ecc0b3122501b72ec7f825966419759ac" +} diff --git a/backend/.sqlx/query-7ef0e3f0fdbda1ec514a621c73782f9afac8b6db79668cbdb188719c5a5de6da.json b/backend/.sqlx/query-7ef0e3f0fdbda1ec514a621c73782f9afac8b6db79668cbdb188719c5a5de6da.json new file mode 100644 index 0000000000..fd30017ca3 --- /dev/null +++ b/backend/.sqlx/query-7ef0e3f0fdbda1ec514a621c73782f9afac8b6db79668cbdb188719c5a5de6da.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO concurrency_key (key, job_id) VALUES ($1, $2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "7ef0e3f0fdbda1ec514a621c73782f9afac8b6db79668cbdb188719c5a5de6da" +} diff --git a/backend/.sqlx/query-fc4583d1570f3a2a428bb28390ca72e61719fae68aa4b42730f9076f3bd97441.json b/backend/.sqlx/query-fc4583d1570f3a2a428bb28390ca72e61719fae68aa4b42730f9076f3bd97441.json index eb320264bc..484dd23e2a 100644 --- a/backend/.sqlx/query-fc4583d1570f3a2a428bb28390ca72e61719fae68aa4b42730f9076f3bd97441.json +++ b/backend/.sqlx/query-fc4583d1570f3a2a428bb28390ca72e61719fae68aa4b42730f9076f3bd97441.json @@ -21,7 +21,7 @@ ] }, "nullable": [ - true, + null, null ] }, diff --git a/backend/src/main.rs b/backend/src/main.rs index 43502a1980..2bc536200e 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -40,14 +40,14 @@ use windmill_common::{ global_settings::{ AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, AUDIT_LOG_RETENTION_DAYS_SETTING, BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, BUN_INSTALL_MIN_RELEASE_AGE_SETTING, - CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, - CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, - DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, - DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, - EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, - FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING, HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, - HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING, - INSTANCE_EVENTS_WEBHOOK_SETTING, INSTANCE_PYTHON_VERSION_SETTING, + CONCURRENCY_KEY_MAX_QUEUED_SETTING, CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, + CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, + CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, + DEFAULT_TAGS_WORKSPACES_SETTING, DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING, + ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, + EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING, + HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, + INDEXER_SETTING, INSTANCE_EVENTS_WEBHOOK_SETTING, INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JOB_ISOLATION_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING, MAVEN_SETTINGS_XML_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NO_DEFAULT_MAVEN_SETTING, @@ -122,20 +122,21 @@ use windmill_worker::{ }; 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_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, + initial_load, load_concurrency_key_max_queued, 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_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, 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, @@ -1879,6 +1880,11 @@ async fn process_notify_event( tracing::error!("Error loading workspace fairness min total: {e:#}"); } } + CONCURRENCY_KEY_MAX_QUEUED_SETTING => { + if let Err(e) = load_concurrency_key_max_queued(db).await { + tracing::error!("Error loading concurrency key max queued: {e:#}"); + } + } SMTP_SETTING => { reload_smtp_config(db).await; } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 35d057ab12..58af60c525 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -53,11 +53,12 @@ use windmill_common::{ flow_status::{FlowStatus, FlowStatusModule}, global_settings::{ AUDIT_LOG_RETENTION_DAYS_SETTING, BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, - BUN_INSTALL_MIN_RELEASE_AGE_SETTING, CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, - CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, - CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, - DEFAULT_TAGS_WORKSPACES_SETTING, DISABLE_PASSWORD_LOGIN, DISABLE_PASSWORD_LOGIN_SETTING, - EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, + BUN_INSTALL_MIN_RELEASE_AGE_SETTING, CONCURRENCY_KEY_MAX_QUEUED_SETTING, + CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, + CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, + DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, + DISABLE_PASSWORD_LOGIN, DISABLE_PASSWORD_LOGIN_SETTING, EXPOSE_DEBUG_METRICS_SETTING, + EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING, HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JOB_ISOLATION_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, @@ -86,7 +87,8 @@ use windmill_common::{ load_env_vars, load_init_bash_from_env, load_periodic_bash_script_from_env, load_periodic_bash_script_interval_from_env, load_whitelist_env_vars_from_env, load_worker_config, reload_custom_tags_setting, store_pull_query, - store_suspended_pull_query, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE, + store_suspended_pull_query, Connection, WorkerConfig, CLOUD_HOSTED, + CONCURRENCY_KEY_MAX_QUEUED, CONCURRENCY_KEY_MAX_QUEUED_DEFAULT, 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, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED, @@ -291,6 +293,13 @@ pub async fn initial_load( if let Err(e) = load_workspace_fairness_enabled(db).await { tracing::error!("Error loading workspace fairness enabled: {e:#}"); } + + // Only the cloud reads this cap, so don't spend a query loading it anywhere else. + if *CLOUD_HOSTED { + if let Err(e) = load_concurrency_key_max_queued(db).await { + tracing::error!("Error loading concurrency key max queued: {e:#}"); + } + } } if server_mode { @@ -610,6 +619,10 @@ const WORKSPACE_FAIRNESS_MAX_PERCENT_DEFAULT: u32 = 50; const WORKSPACE_FAIRNESS_DURATION_SECS_DEFAULT: u32 = 10; const WORKSPACE_FAIRNESS_MIN_TOTAL_DEFAULT: u32 = 4; +/// The cap is used as a SQL `LIMIT`, so it must survive the `u32 -> i64` widening without +/// becoming absurd; `u32::MAX` is already far beyond any queue depth worth allowing. +const CONCURRENCY_KEY_MAX_QUEUED_MAX: u64 = u32::MAX as u64; + pub async fn load_workspace_fairness_enabled(db: &DB) -> error::Result<()> { // Match the convention used by `load_preview_tags_override` / // `load_fork_workspace_tag_append_fork_suffix`: on transient DB errors, leave the in-memory @@ -697,6 +710,39 @@ pub async fn load_workspace_fairness_min_total(db: &DB) -> error::Result<()> { Ok(()) } +pub async fn load_concurrency_key_max_queued(db: &DB) -> error::Result<()> { + // See `load_workspace_fairness_max_percent` for the Err / None / invalid policy. + match load_value_from_global_settings(db, CONCURRENCY_KEY_MAX_QUEUED_SETTING).await? { + Some(serde_json::Value::Number(n)) => { + // `0` is a meaningful value here (disable the cap), so unlike the fairness knobs + // the lower bound is 0 rather than 1. + let v = n + .as_u64() + .map(|u| u.min(CONCURRENCY_KEY_MAX_QUEUED_MAX) as u32) + .unwrap_or_else(|| { + // Warn rather than silently defaulting: `-1` and `"0"` are plausible + // attempts to disable the cap, and both would otherwise land on 10000. + tracing::warn!( + "{CONCURRENCY_KEY_MAX_QUEUED_SETTING}={n} is not a non-negative integer, \ + falling back to {CONCURRENCY_KEY_MAX_QUEUED_DEFAULT}. Set 0 to disable." + ); + CONCURRENCY_KEY_MAX_QUEUED_DEFAULT + }); + CONCURRENCY_KEY_MAX_QUEUED.store(v, Ordering::Relaxed); + } + other => { + if let Some(v) = other { + tracing::warn!( + "{CONCURRENCY_KEY_MAX_QUEUED_SETTING}={v} is not a number, falling back to \ + {CONCURRENCY_KEY_MAX_QUEUED_DEFAULT}. Set 0 to disable." + ); + } + CONCURRENCY_KEY_MAX_QUEUED.store(CONCURRENCY_KEY_MAX_QUEUED_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-api-jobs/src/jobs_export.rs b/backend/windmill-api-jobs/src/jobs_export.rs index 0b188c0fa0..4f481fbd54 100644 --- a/backend/windmill-api-jobs/src/jobs_export.rs +++ b/backend/windmill-api-jobs/src/jobs_export.rs @@ -19,6 +19,7 @@ use windmill_common::{ jobs::{is_safe_log_file_path, JobKind, JobStatus, JobTriggerKind}, scripts::ScriptLang, utils::{paginate, paginate_without_limits, require_admin, Pagination}, + worker::CLOUD_HOSTED, }; use windmill_api_auth::ApiAuthed; @@ -435,6 +436,15 @@ pub async fn import_queued_jobs( ) -> error::Result { require_admin(authed.is_admin, &authed.username)?; + // Self-hosted migration/restore tool. It writes queue rows straight to the DB, bypassing the + // push path and every admission check that lives there, so it has no place on multi-tenant + // cloud where those checks are what keep one workspace from swamping the shared pool. + if *CLOUD_HOSTED { + return Err(error::Error::BadRequest( + "Importing queued jobs is not available on the cloud".to_string(), + )); + } + let mut tx = user_db.begin(&authed).await?; for job in jobs { diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 0ee326ce6b..bbe7dbb9b5 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -116,6 +116,11 @@ pub const WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING: &str = "workspace_fairness_max 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"; +// Cloud-only ceiling on how many jobs may sit in the queue behind a single +// concurrency key. `0` disables the cap. See `windmill-queue/src/jobs.rs`, +// `check_concurrency_key_queue_cap`. +pub const CONCURRENCY_KEY_MAX_QUEUED_SETTING: &str = "concurrency_key_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 eb339b56f8..2c936ff0aa 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -210,6 +210,9 @@ impl SpecificTagType { pub const DEFAULT_CLOUD_TIMEOUT: u64 = 900; pub const DEFAULT_SELFHOSTED_TIMEOUT: u64 = 604800; // 7 days 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; lazy_static::lazy_static! { pub static ref WORKER_GROUP: String = std::env::var("WORKER_GROUP").unwrap_or_else(|_| { #[cfg(not(feature = "enterprise"))] @@ -337,6 +340,13 @@ lazy_static::lazy_static! { /// `should_admit_capped` is moot and "admit all" is the correct no-op. pub static ref WORKSPACE_FAIRNESS_ADMISSION_PPM: AtomicU32 = AtomicU32::new(10_000); + /// Cloud-only ceiling on the number of jobs queued behind a single concurrency key. + /// A concurrency-limited key drains at most `concurrent_limit` jobs per window, so a + /// producer pushing faster than that grows an unbounded backlog that no amount of + /// spare worker capacity can absorb. `0` disables the cap. + pub static ref CONCURRENCY_KEY_MAX_QUEUED: AtomicU32 = + AtomicU32::new(CONCURRENCY_KEY_MAX_QUEUED_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 4ce45eb8c0..934b9b50b4 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6281,14 +6281,23 @@ async fn push_inner<'c, 'd>( .unzip(); if concurrency_settings.concurrent_limit.is_some() { - insert_concurrency_key( + let concurrency_key = resolve_concurrency_key( workspace_id, &args, &runnable_path, job_kind, concurrency_settings.concurrency_key.clone(), + ); + #[cfg(feature = "cloud")] + if *CLOUD_HOSTED { + check_concurrency_key_queue_cap(&mut *tx, &concurrency_key).await?; + } + insert_resolved_concurrency_key( + &concurrency_key, &mut *tx, job_id, + &runnable_path, + workspace_id, ) .await?; } @@ -6617,16 +6626,14 @@ async fn push_inner<'c, 'd>( Ok((job_id, tx)) } -pub async fn insert_concurrency_key<'d, 'c>( +fn resolve_concurrency_key<'d>( workspace_id: &str, args: &PushArgs<'d>, script_path: &Option, job_kind: JobKind, custom_concurrency_key: Option, - db: impl PgExecutor<'c>, - job_id: Uuid, -) -> Result<(), Error> { - let concurrency_key = custom_concurrency_key +) -> String { + custom_concurrency_key .map(|x| { let interpolated = interpolate_args(x.clone(), args, workspace_id); // In cloud mode, enforce workspace isolation by prefixing with workspace @@ -6648,7 +6655,50 @@ pub async fn insert_concurrency_key<'d, 'c>( workspace_id, script_path.as_ref(), &job_kind, - )); + )) +} + +/// Resolves the concurrency key, applies the cloud-only queue-depth cap, then registers the key. +/// +/// `concurrent_limit: None` still registers the key but skips the cap: without a limit nothing +/// serializes the key, so there is no backlog to bound. Callers register keys for tag +/// interpolation alone, so the two are not interchangeable. +/// +/// Takes a `Copy` executor because the cap runs a query before the insert on the same one. `push` +/// cannot use this — it holds a `&mut Transaction` — so it performs the same three steps inline. +pub async fn insert_concurrency_key_capped<'d, 'c, E: PgExecutor<'c> + Copy>( + workspace_id: &str, + args: &PushArgs<'d>, + script_path: &Option, + job_kind: JobKind, + custom_concurrency_key: Option, + concurrent_limit: Option, + db: E, + job_id: Uuid, +) -> Result<(), Error> { + let concurrency_key = resolve_concurrency_key( + workspace_id, + args, + script_path, + job_kind, + custom_concurrency_key, + ); + #[cfg(feature = "cloud")] + if *CLOUD_HOSTED && concurrent_limit.is_some() { + check_concurrency_key_queue_cap(db, &concurrency_key).await?; + } + #[cfg(not(feature = "cloud"))] + let _ = concurrent_limit; + insert_resolved_concurrency_key(&concurrency_key, db, job_id, script_path, workspace_id).await +} + +async fn insert_resolved_concurrency_key<'c>( + concurrency_key: &str, + db: impl PgExecutor<'c>, + job_id: Uuid, + script_path: &Option, + workspace_id: &str, +) -> Result<(), Error> { sqlx::query!( "WITH inserted_concurrency_counter AS ( INSERT INTO concurrency_counter (concurrency_id, job_uuids) @@ -6666,6 +6716,81 @@ pub async fn insert_concurrency_key<'d, 'c>( Ok(()) } +/// Counts jobs *waiting* behind `concurrency_key`, scanning at most `limit` rows. +/// +/// Three constraints on the query below, each pinned by a test in +/// `tests/concurrency_key_queue_depth_test.rs`: +/// - Any `scheduled_for`: the limiter parks blocked jobs in the future, so a gated backlog is +/// almost entirely future-dated and `scheduled_for <= now()` would never see it. +/// - `running = false`: running jobs keep `ended_at` NULL, and counting them would charge a key +/// for the concurrency it is licensed to use. +/// - `LIMIT` inside the `EXISTS`, not outside: rows whose job left the queue without +/// `add_completed_job` keep `ended_at` NULL forever (the retention sweep only matches +/// `ended_at <= …`), so an outside `LIMIT` would let them be rescanned on every push. +/// +/// Takes an already-resolved key and performs no authorization; `pub` only so the integration +/// test can reach it. Never call it with a caller-supplied key — it would leak queue depth +/// across workspaces. +pub async fn concurrency_key_queue_depth<'c>( + db: impl PgExecutor<'c>, + concurrency_key: &str, + limit: i64, +) -> Result { + sqlx::query_scalar!( + "SELECT count(*) FROM ( + SELECT ck.job_id FROM concurrency_key ck + WHERE ck.key = $1 AND ck.ended_at IS NULL + LIMIT $2 + ) s WHERE EXISTS ( + SELECT 1 FROM v2_job_queue q WHERE q.id = s.job_id AND q.running = false + )", + concurrency_key, + limit, + ) + .fetch_one(db) + .warn_after_seconds(3) + .await + .map_err(|e| { + Error::internal_err(format!( + "Could not count queued jobs for concurrency_key={concurrency_key}: {e:#}" + )) + }) + .map(|c| c.unwrap_or(0)) +} + +/// Rejects the push when `concurrency_key` already has `CONCURRENCY_KEY_MAX_QUEUED` jobs queued. +/// +/// Caller must runtime-gate this on `*CLOUD_HOSTED`, and must call it from *every push path* +/// that registers a concurrency key: a flow with a preprocessor is pushed with `concurrent_limit` +/// cleared and only registers its key later from `worker_flow`, so gating `push` alone leaves +/// trigger-driven flows uncapped. +/// +/// `add_batch_jobs` writes keys directly and is deliberately exempt: it is superadmin-only, and a +/// superadmin can set the cap to `0` anyway. `import_queued_jobs` is likewise exempt because it +/// is rejected outright on cloud. +#[cfg(feature = "cloud")] +async fn check_concurrency_key_queue_cap<'c>( + db: impl PgExecutor<'c>, + concurrency_key: &str, +) -> Result<(), Error> { + let cap = windmill_common::worker::CONCURRENCY_KEY_MAX_QUEUED + .load(std::sync::atomic::Ordering::Relaxed); + if cap == 0 { + return Ok(()); + } + let cap = cap as i64; + let depth = concurrency_key_queue_depth(db, concurrency_key, cap).await?; + if depth >= cap { + return Err(Error::QuotaExceeded(format!( + "Too many jobs queued behind concurrency key '{concurrency_key}': at least {depth} \ + jobs are already waiting and the limit is {cap}. Jobs sharing a concurrency key run \ + at most `concurrent_limit` at a time, so this queue is growing faster than it can \ + drain. Cancel the backlog, slow down the caller, or raise the concurrency limit." + ))); + } + Ok(()) +} + // pub async fn insert_debounce_key<'d, 'c>( // workspace_id: &str, // args: &PushArgs<'d>, diff --git a/backend/windmill-queue/tests/concurrency_key_queue_depth_test.rs b/backend/windmill-queue/tests/concurrency_key_queue_depth_test.rs new file mode 100644 index 0000000000..3689289bf4 --- /dev/null +++ b/backend/windmill-queue/tests/concurrency_key_queue_depth_test.rs @@ -0,0 +1,105 @@ +//! Regression guard for `concurrency_key_queue_depth`, which backs the cloud-only cap on how +//! many jobs may queue behind a single concurrency key. +//! +//! Run with: +//! cargo test -p windmill-queue --test concurrency_key_queue_depth_test + +use sqlx::{Pool, Postgres}; +use uuid::Uuid; +use windmill_queue::jobs::concurrency_key_queue_depth; + +/// Queues `count` jobs on `key`, all scheduled `offset_secs` from now. +async fn seed_queued(db: &Pool, key: &str, count: usize, offset_secs: i64) { + seed(db, key, count, offset_secs, false).await +} + +async fn seed(db: &Pool, key: &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, 'test-workspace', 'other')", + id, + ) + .execute(db) + .await + .expect("seed v2_job"); + sqlx::query!( + "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag, running) + VALUES ($1, 'test-workspace', now() + ($2::bigint::text || ' s')::interval, 'other', $3)", + id, + offset_secs, + running, + ) + .execute(db) + .await + .expect("seed v2_job_queue"); + sqlx::query!( + "INSERT INTO concurrency_key (key, job_id) VALUES ($1, $2)", + key, + id, + ) + .execute(db) + .await + .expect("seed concurrency_key"); + } +} + +/// The property the cap depends on: jobs scheduled into the future count toward the depth. +/// +/// The concurrency limiter re-queues a blocked job by pushing `scheduled_for` forward one full +/// window at a time, so a backlog gated by a concurrency limit is almost entirely future-dated. +/// A depth query narrowed to `scheduled_for <= now()` would report only the trickle the limiter +/// has released and the cap would never fire on the runaway it exists to stop. +#[sqlx::test(migrations = "../migrations")] +async fn counts_future_scheduled_jobs(db: Pool) { + let key = "future-scheduled-count"; + seed_queued(&db, key, 3, -10).await; // due now + seed_queued(&db, key, 7, 3600).await; // parked an hour out by the limiter + + let depth = concurrency_key_queue_depth(&db, key, 1000) + .await + .expect("count depth"); + + assert_eq!( + depth, 10, + "depth must include future-scheduled jobs; counting only due jobs would report 3" + ); +} + +/// Running jobs are not backlog. They stay in `v2_job_queue` with `ended_at` still NULL, so +/// counting them would charge a key for the concurrency it is licensed to use: a key whose +/// `concurrent_limit` exceeds the cap would reject every push with nothing actually waiting. +#[sqlx::test(migrations = "../migrations")] +async fn ignores_running_jobs(db: Pool) { + let key = "running-not-backlog"; + seed(&db, key, 6, -10, true).await; // executing right now + seed(&db, key, 2, 3600, false).await; // actually waiting + + let depth = concurrency_key_queue_depth(&db, key, 1000) + .await + .expect("count depth"); + + assert_eq!( + depth, 2, + "only waiting jobs count; the 6 running ones must not" + ); +} + +/// A job that left the queue without going through `add_completed_job` leaves its +/// `concurrency_key` row with `ended_at IS NULL` forever. Those must not count, otherwise a mass +/// cancel or manual purge leaves the key permanently wedged above the cap with an empty queue. +#[sqlx::test(migrations = "../migrations")] +async fn ignores_rows_whose_job_left_the_queue(db: Pool) { + let key = "orphaned-rows"; + seed_queued(&db, key, 4, 3600).await; + sqlx::query!("DELETE FROM v2_job_queue") + .execute(&db) + .await + .expect("purge queue"); + + let depth = concurrency_key_queue_depth(&db, key, 1000) + .await + .expect("count depth"); + + assert_eq!(depth, 0, "orphaned concurrency_key rows must not count"); +} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 21f4ce6925..f60fc94c0e 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -70,7 +70,7 @@ use windmill_common::{ use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job, - insert_concurrency_key, interpolate_args, + insert_concurrency_key_capped, interpolate_args, report_error_to_workspace_handler_or_critical_side_channel, try_schedule_next_job, CanceledBy, FlowRunners, MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError, @@ -1552,12 +1552,13 @@ pub async fn update_flow_status_after_job_completion_internal( if concurrency_requires_args { let args = PushArgs::from(fetched_args.as_ref().unwrap()); if let Some(ck) = concurrency_key { - insert_concurrency_key( + insert_concurrency_key_capped( &flow_job.workspace_id, &args, &flow_job.runnable_path, JobKind::Flow, Some(ck), + concurrent_limit, db, flow, ) @@ -1567,12 +1568,13 @@ pub async fn update_flow_status_after_job_completion_internal( tag = Some(interpolate_args(t, &args, &flow_job.workspace_id)); } } else if concurrent_limit.is_some() { - insert_concurrency_key( + insert_concurrency_key_capped( &flow_job.workspace_id, &PushArgs::from(&HashMap::new()), &flow_job.runnable_path, JobKind::Flow, concurrency_key, + concurrent_limit, db, flow, ) diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 0c8ba77beb..b58eb329f7 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -426,6 +426,18 @@ export const settings: Record = { cloudonly: false, ee_only: 'Workspace fairness is an Enterprise feature.', hideInQuickSetup: true + }, + { + label: 'Max jobs queued per concurrency key', + description: + 'Rejects new jobs once this many are already queued behind one concurrency key. Jobs sharing a key run at most concurrent limit at a time regardless of spare worker capacity, so a caller pushing faster than the key drains grows a backlog no capacity can absorb. Scoped per key, so a runaway producer cannot block the rest of the workspace. Set 0 to disable. Default 10000.', + key: 'concurrency_key_max_queued_jobs', + fieldType: 'number', + placeholder: '10000', + storage: 'setting', + cloudonly: true, + ee_only: '', + hideInQuickSetup: true } ], 'Object Storage': [