diff --git a/backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json b/backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json new file mode 100644 index 0000000000..ae95dc079e --- /dev/null +++ b/backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH ping AS (\n UPDATE v2_job_runtime SET ping = null WHERE id = $2\n )\n UPDATE v2_job_queue SET\n running = false,\n started_at = null,\n scheduled_for = $1,\n concurrency_gated = true\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Timestamptz", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405" +} diff --git a/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json b/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json new file mode 100644 index 0000000000..0811b1f053 --- /dev/null +++ b/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT COUNT(*) FILTER (WHERE concurrency_gated IS NOT TRUE) AS count,\n COUNT(*) FILTER (WHERE concurrency_gated) AS gated_count,\n MIN(scheduled_for) FILTER (WHERE concurrency_gated IS NOT TRUE)\n AS oldest_job,\n MIN(scheduled_for) AS oldest_any\n FROM v2_job_queue\n WHERE tag = $1\n AND scheduled_for <= NOW() - $2::interval\n AND running = false\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "gated_count", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "oldest_job", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "oldest_any", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Interval" + ] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5" +} diff --git a/backend/.sqlx/query-3f9dd35d19e6badc92397fdb476d39ffadf1ff97931a9506772f76c3f36bd2c7.json b/backend/.sqlx/query-3f9dd35d19e6badc92397fdb476d39ffadf1ff97931a9506772f76c3f36bd2c7.json deleted file mode 100644 index e51ceff529..0000000000 --- a/backend/.sqlx/query-3f9dd35d19e6badc92397fdb476d39ffadf1ff97931a9506772f76c3f36bd2c7.json +++ /dev/null @@ -1,41 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT COUNT(*) AS count,\n MIN(scheduled_for) AS oldest_job,\n COUNT(*) FILTER (WHERE canceled_by IS NOT NULL) AS canceled_count,\n MIN(scheduled_for) FILTER (WHERE canceled_by IS NOT NULL) AS canceled_oldest\n FROM v2_job_queue\n WHERE tag = $1\n AND scheduled_for <= NOW() - $2::interval\n AND running = false\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "oldest_job", - "type_info": "Timestamptz" - }, - { - "ordinal": 2, - "name": "canceled_count", - "type_info": "Int8" - }, - { - "ordinal": 3, - "name": "canceled_oldest", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text", - "Interval" - ] - }, - "nullable": [ - null, - null, - null, - null - ] - }, - "hash": "3f9dd35d19e6badc92397fdb476d39ffadf1ff97931a9506772f76c3f36bd2c7" -} diff --git a/backend/.sqlx/query-dbacbbbe27d1589dd26d6bcbd131a4ce32944cb2b36323117bbe92460e88b767.json b/backend/.sqlx/query-dbacbbbe27d1589dd26d6bcbd131a4ce32944cb2b36323117bbe92460e88b767.json deleted file mode 100644 index 3f0b84d9ac..0000000000 --- a/backend/.sqlx/query-dbacbbbe27d1589dd26d6bcbd131a4ce32944cb2b36323117bbe92460e88b767.json +++ /dev/null @@ -1,35 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n WITH waiting AS (\n SELECT ck.key AS concurrency_key, cs.concurrent_limit,\n cs.concurrency_time_window_s,\n COUNT(*) AS waiting_count, MIN(q.scheduled_for) AS oldest_job\n FROM v2_job_queue q\n LEFT JOIN concurrency_key ck ON ck.job_id = q.id\n LEFT JOIN runnable_settings rs ON rs.hash = q.runnable_settings_handle\n LEFT JOIN concurrency_settings cs ON cs.hash = rs.concurrency_settings\n WHERE q.tag = $1\n AND q.scheduled_for <= NOW() - $2::interval\n AND q.running = false\n AND q.canceled_by IS NULL\n GROUP BY ck.key, cs.concurrent_limit, cs.concurrency_time_window_s\n ),\n classified AS MATERIALIZED (\n SELECT w.waiting_count, w.oldest_job,\n w.concurrent_limit > 0 AND w.concurrency_key <> '' AND (\n (SELECT COUNT(*) FROM jsonb_object_keys(cc.job_uuids))\n + (SELECT COUNT(*) FROM concurrency_key done\n WHERE done.key = w.concurrency_key\n AND done.ended_at >= NOW() - INTERVAL '1 second'\n * COALESCE(w.concurrency_time_window_s, 0))\n ) >= w.concurrent_limit AS gated\n FROM waiting w\n LEFT JOIN concurrency_counter cc ON cc.concurrency_id = w.concurrency_key\n )\n SELECT COALESCE(SUM(waiting_count) FILTER (WHERE gated IS NOT TRUE), 0)::bigint AS count,\n COALESCE(SUM(waiting_count) FILTER (WHERE gated), 0)::bigint AS gated_count,\n MIN(oldest_job) FILTER (WHERE gated IS NOT TRUE) AS oldest_job\n FROM classified\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "gated_count", - "type_info": "Int8" - }, - { - "ordinal": 2, - "name": "oldest_job", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text", - "Interval" - ] - }, - "nullable": [ - null, - null, - null - ] - }, - "hash": "dbacbbbe27d1589dd26d6bcbd131a4ce32944cb2b36323117bbe92460e88b767" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index af1bf852b3..14b15ccfc5 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -d54657fe4641a910cb2396de9cf83e850f118d2c \ No newline at end of file +3984a0f87d24770f860a315061a26c9e53368b2e \ No newline at end of file diff --git a/backend/migrations/20260720055519_concurrency_gated_flag.down.sql b/backend/migrations/20260720055519_concurrency_gated_flag.down.sql new file mode 100644 index 0000000000..e2f0d4a7b1 --- /dev/null +++ b/backend/migrations/20260720055519_concurrency_gated_flag.down.sql @@ -0,0 +1 @@ +ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gated; diff --git a/backend/migrations/20260720055519_concurrency_gated_flag.up.sql b/backend/migrations/20260720055519_concurrency_gated_flag.up.sql new file mode 100644 index 0000000000..a825e2e886 --- /dev/null +++ b/backend/migrations/20260720055519_concurrency_gated_flag.up.sql @@ -0,0 +1,7 @@ +-- Set by the concurrency limiter when it re-queues a job it could not admit. +-- Records that the job is parked behind its own concurrency gate rather than +-- waiting for a worker, which the queue row cannot otherwise show: the +-- limiter's admission test spans concurrency_counter, the completed-window rows +-- in concurrency_key, per-version setting fallbacks and several bypass paths. +-- Nullable so adding it does not rewrite the table. +ALTER TABLE v2_job_queue ADD COLUMN IF NOT EXISTS concurrency_gated BOOLEAN; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index e8bf76d763..b21d1d26d4 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -174,7 +174,7 @@ usr_to_group: workspace_id(char), group_(char), usr(char) v2_job: id(uuid), raw_code(text), raw_lock(text), raw_flow(jsonb), tag(char), workspace_id(char), created_at(ts), created_by(char), permissioned_as(char), permissioned_as_email(char), kind(job_kind), runnable_id(bigint), runnable_path(char), parent_job(uuid), root_job(uuid), script_lang(script_lang), script_entrypoint_override(char), flow_step(int), flow_step_id(char), flow_innermost_root_job(uuid), trigger(char), trigger_kind(job_trigger_kind), same_worker(bool), visible_to_owner(bool), concurrent_limit(int), concurrency_time_window_s(int), cache_ttl(int), timeout(int), priority(smallint), preprocessed(bool), args(jsonb), labels(text[]), pre_run_error(text) v2_job_completed: id(uuid), workspace_id(char), duration_ms(bigint), result(jsonb), deleted(bool), canceled_by(char), canceled_reason(text), flow_status(jsonb), started_at(ts), memory_peak(int), status(job_status), completed_at(ts), worker(char), workflow_as_code_status(jsonb), result_columns(text[]), retries(uuid[]), extras(jsonb) v2_job_debounce_batch: id(uuid), debounce_batch(bigint) -v2_job_queue: id(uuid), workspace_id(char), created_at(ts), started_at(ts), scheduled_for(ts), running(bool), canceled_by(char), canceled_reason(text), suspend(int), suspend_until(ts), tag(char), priority(smallint), worker(char), extras(jsonb), cache_ignore_s3_path(bool), runnable_settings_handle(bigint) +v2_job_queue: id(uuid), workspace_id(char), created_at(ts), started_at(ts), scheduled_for(ts), running(bool), canceled_by(char), canceled_reason(text), suspend(int), suspend_until(ts), tag(char), priority(smallint), worker(char), extras(jsonb), cache_ignore_s3_path(bool), runnable_settings_handle(bigint), concurrency_gated(bool) v2_job_runtime: id(uuid), ping(ts), memory_peak(int) FK: (id) -> v2_job_queue(id) v2_job_status: id(uuid), flow_status(jsonb), flow_leaf_jobs(jsonb), workflow_as_code_status(jsonb) diff --git a/backend/windmill-common/tests/jobs_waiting_alerts.rs b/backend/windmill-common/tests/jobs_waiting_alerts.rs index d727a3272c..d3622408e1 100644 --- a/backend/windmill-common/tests/jobs_waiting_alerts.rs +++ b/backend/windmill-common/tests/jobs_waiting_alerts.rs @@ -19,7 +19,6 @@ mod tests { use windmill_common::ee::jobs_waiting_alerts; const TAG: &str = "python3"; - const GATE_KEY: &str = "test-workspace/u/user/gated_script"; /// `jobs_waiting_alerts` bails out unless it can take the shared lock, and a /// freshly inserted lock row is never old enough to be claimed on the first @@ -54,108 +53,20 @@ mod tests { .expect("seed alert config"); } - /// A concurrency gate as the push path lays it out: the limit hangs off - /// `runnable_settings_handle` -> `concurrency_settings`, and the live - /// admission count is the number of job uuids in `concurrency_counter`. - /// `running` is how many slots are currently taken. - /// - /// The window is wide so a completion seeded by `seed_recent_completions` - /// stays inside it for the length of the test. - async fn seed_gate(db: &Pool, key: &str, handle: i64, limit: i32, running: usize) { - let job_uuids: serde_json::Value = (0..running) - .map(|_| (Uuid::new_v4().hyphenated().to_string(), json!({}))) - .collect::>() - .into(); - - sqlx::query!( - "INSERT INTO concurrency_settings (hash, concurrency_key, concurrent_limit, concurrency_time_window_s) - VALUES ($1, $2, $3, 600)", - handle, - key, - limit, - ) - .execute(db) - .await - .expect("seed concurrency settings"); - - sqlx::query!( - "INSERT INTO runnable_settings (hash, concurrency_settings) VALUES ($1, $1)", - handle, - ) - .execute(db) - .await - .expect("seed runnable settings"); - - sqlx::query!( - "INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, $2)", - key, - job_uuids, - ) - .execute(db) - .await - .expect("seed concurrency counter"); - } - - /// Queue `n` jobs on TAG that have been due for 5 minutes. When `gate` is - /// set, each job also gets the `concurrency_key` row that - /// `insert_concurrency_key` writes at push time for any runnable carrying a - /// concurrent_limit, plus the settings handle that carries the limit. - async fn queue_jobs(db: &Pool, n: usize, gate: Option<(&str, i64)>) { - queue_jobs_inner(db, n, gate, None).await - } - - /// Soft-cancelled jobs stay queued until a worker picks them up, and the - /// pull path skips concurrency limiting for them entirely. - async fn queue_canceled_jobs(db: &Pool, n: usize, gate: Option<(&str, i64)>) { - queue_jobs_inner(db, n, gate, Some("canceller")).await - } - - async fn queue_jobs_inner( - db: &Pool, - n: usize, - gate: Option<(&str, i64)>, - canceled_by: Option<&str>, - ) { + /// Queue `n` jobs on TAG that have been due for 5 minutes. `gated` mirrors + /// what the limiter writes when it re-queues a job it could not admit. + async fn queue_jobs(db: &Pool, n: usize, gated: bool) { for _ in 0..n { - let id = Uuid::new_v4(); sqlx::query!( - "INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for, runnable_settings_handle, canceled_by) - VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes', $3, $4)", - id, + "INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for, concurrency_gated) + VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes', $3)", + Uuid::new_v4(), TAG, - gate.map(|(_, handle)| handle), - canceled_by, + gated.then_some(true), ) .execute(db) .await .expect("queue job"); - - if let Some((key, _)) = gate { - sqlx::query!( - "INSERT INTO concurrency_key (job_id, key) VALUES ($1, $2)", - id, - key, - ) - .execute(db) - .await - .expect("gate job"); - } - } - } - - /// Jobs that already finished inside the gate's time window. The limiter - /// counts these toward the limit, so they hold the gate shut even though - /// they are gone from `concurrency_counter`. - async fn seed_recent_completions(db: &Pool, key: &str, n: usize) { - for _ in 0..n { - sqlx::query!( - "INSERT INTO concurrency_key (job_id, key, ended_at) VALUES ($1, $2, NOW())", - Uuid::new_v4(), - key, - ) - .execute(db) - .await - .expect("seed completion"); } } @@ -166,7 +77,7 @@ mod tests { .expect("read alerts") } - /// Jobs parked by their own workspace's concurrency gate must not page + /// Jobs the limiter parked behind their own concurrency gate must not page /// on-call: they are waiting by design, not for want of a worker, and no /// operator action drains them. Their re-queue timestamps mature in bursts, /// so counting them produces a stream of alerts whose count swings wildly @@ -176,101 +87,26 @@ mod tests { async fn concurrency_gated_backlog_does_not_alert(db: Pool) { free_the_lock(&db).await; configure_alert(&db, 100).await; - seed_gate(&db, GATE_KEY, 1, 1, 1).await; - queue_jobs(&db, 200, Some((GATE_KEY, 1))).await; + queue_jobs(&db, 200, true).await; jobs_waiting_alerts(&db).await; assert!( alert_messages(&db).await.is_empty(), - "a backlog held entirely behind a saturated concurrency gate must not raise a critical alert" + "a backlog the limiter parked behind a concurrency gate must not raise a critical alert" ); } - /// A gate can be shut with nothing running: the limiter admits on running - /// jobs *plus* those that ended inside concurrency_time_window_s, and a - /// completion clears the job from `concurrency_counter` while still holding - /// the window. Reading the counter alone reports the gate free, and the - /// backlog behind it pages -- the same false alert, one completion later. - #[ignore = "requires database setup - run with --ignored flag"] - #[sqlx::test(migrations = "../migrations")] - async fn time_window_saturated_gate_does_not_alert(db: Pool) { - free_the_lock(&db).await; - configure_alert(&db, 100).await; - seed_gate(&db, GATE_KEY, 1, 1, 0).await; - seed_recent_completions(&db, GATE_KEY, 1).await; - queue_jobs(&db, 200, Some((GATE_KEY, 1))).await; - - jobs_waiting_alerts(&db).await; - - assert!( - alert_messages(&db).await.is_empty(), - "a gate saturated by completions inside its time window must not raise a critical alert" - ); - } - - /// Cancellation is a limiter bypass too: the pull path hands a job with - /// `canceled_by` set straight to a worker without consulting the counter. - /// Such jobs stay queued until a worker processes them, so a worker outage - /// piles them up behind whatever key they carry -- and classifying them off - /// that key hides an unbounded cancelled backlog from the alert. - #[ignore = "requires database setup - run with --ignored flag"] - #[sqlx::test(migrations = "../migrations")] - async fn canceled_jobs_behind_saturated_gate_still_alert(db: Pool) { - free_the_lock(&db).await; - configure_alert(&db, 100).await; - seed_gate(&db, GATE_KEY, 1, 1, 1).await; - queue_canceled_jobs(&db, 200, Some((GATE_KEY, 1))).await; - - jobs_waiting_alerts(&db).await; - - let messages = alert_messages(&db).await; - assert_eq!( - messages.len(), - 1, - "soft-cancelled jobs bypass the limiter and must still alert, got {messages:?}" - ); - } - - /// An empty concurrency key is a limiter bypass: `apply_concurrency_limit` - /// admits unconditionally instead of consulting the counter, so these jobs - /// are runnable and any backlog of them is genuine capacity starvation. - /// Classifying them off the counter alone would read the gate as full and - /// silence the alert during a real outage. - #[ignore = "requires database setup - run with --ignored flag"] - #[sqlx::test(migrations = "../migrations")] - async fn bypassed_empty_key_still_alerts(db: Pool) { - free_the_lock(&db).await; - configure_alert(&db, 100).await; - seed_gate(&db, "", 1, 1, 1).await; - queue_jobs(&db, 200, Some(("", 1))).await; - - jobs_waiting_alerts(&db).await; - - let messages = alert_messages(&db).await; - assert_eq!( - messages.len(), - 1, - "jobs whose concurrency key is empty bypass the limiter and must still \ - alert, got {messages:?}" - ); - } - - /// The exclusion keys off the gate being *full*, not off the job merely - /// carrying a concurrent_limit -- every such job gets a `concurrency_key` - /// row at push time, so excluding on that row alone would stop this alert - /// from ever firing for a rate-limited runnable. Half the backlog here sits - /// under a gate with free slots: it is starving for workers like the plain - /// jobs beside it and must be counted. Excluding it drops the count under - /// the threshold and the alert goes silent. + /// The complement: jobs no gate is holding are waiting on capacity and must + /// page, counted apart from the gated ones so the number can be reconciled + /// against the much larger raw queue depth. #[ignore = "requires database setup - run with --ignored flag"] #[sqlx::test(migrations = "../migrations")] async fn jobs_waiting_on_capacity_still_alert(db: Pool) { free_the_lock(&db).await; - configure_alert(&db, 150).await; - seed_gate(&db, GATE_KEY, 1, 10, 2).await; - queue_jobs(&db, 100, None).await; - queue_jobs(&db, 100, Some((GATE_KEY, 1))).await; + configure_alert(&db, 100).await; + queue_jobs(&db, 150, false).await; + queue_jobs(&db, 200, true).await; jobs_waiting_alerts(&db).await; @@ -280,9 +116,14 @@ mod tests { 1, "expected exactly one critical alert, got {messages:?}" ); + assert!( + messages[0].contains("150"), + "alert must count only the jobs actually waiting on capacity: {}", + messages[0] + ); assert!( messages[0].contains("200"), - "every job waiting on capacity must be counted, gated or not: {}", + "alert must report the gated jobs it excluded: {}", messages[0] ); }