diff --git a/backend/.sqlx/query-7ae1f44d50c262916530d8b45265bf2d8357209ec093db72697e410ce6e4e385.json b/backend/.sqlx/query-7ae1f44d50c262916530d8b45265bf2d8357209ec093db72697e410ce6e4e385.json deleted file mode 100644 index 52650e6de3..0000000000 --- a/backend/.sqlx/query-7ae1f44d50c262916530d8b45265bf2d8357209ec093db72697e410ce6e4e385.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT COUNT(*)\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" - } - ], - "parameters": { - "Left": [ - "Text", - "Interval" - ] - }, - "nullable": [ - null - ] - }, - "hash": "7ae1f44d50c262916530d8b45265bf2d8357209ec093db72697e410ce6e4e385" -} diff --git a/backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json b/backend/.sqlx/query-953566f05e2ada95ad440c42aede50a8265917b5c5a8eecbcfe139102aa48348.json similarity index 57% rename from backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json rename to backend/.sqlx/query-953566f05e2ada95ad440c42aede50a8265917b5c5a8eecbcfe139102aa48348.json index f2cc5a698d..0954e506dc 100644 --- a/backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json +++ b/backend/.sqlx/query-953566f05e2ada95ad440c42aede50a8265917b5c5a8eecbcfe139102aa48348.json @@ -1,6 +1,6 @@ { "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 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 (\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 ", + "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 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": [ { @@ -31,5 +31,5 @@ null ] }, - "hash": "e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04" + "hash": "953566f05e2ada95ad440c42aede50a8265917b5c5a8eecbcfe139102aa48348" } diff --git a/backend/.sqlx/query-dd2752edc8245a636baa187ac6ee05cc57be86a79be09b35223aa8170fb51cd4.json b/backend/.sqlx/query-dd2752edc8245a636baa187ac6ee05cc57be86a79be09b35223aa8170fb51cd4.json new file mode 100644 index 0000000000..fa71794c74 --- /dev/null +++ b/backend/.sqlx/query-dd2752edc8245a636baa187ac6ee05cc57be86a79be09b35223aa8170fb51cd4.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT COUNT(*) AS count, MIN(scheduled_for) AS oldest_job\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" + } + ], + "parameters": { + "Left": [ + "Text", + "Interval" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "dd2752edc8245a636baa187ac6ee05cc57be86a79be09b35223aa8170fb51cd4" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 0d980243d9..93d7ffc9ab 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -6911fbf69af0c7ced2965dc031096184b173d6c6 \ No newline at end of file +f01fc53cad5ec0c598b6eac904e53443f683aae5 \ No newline at end of file diff --git a/backend/windmill-common/tests/jobs_waiting_alerts.rs b/backend/windmill-common/tests/jobs_waiting_alerts.rs index f387e1e60b..290faee269 100644 --- a/backend/windmill-common/tests/jobs_waiting_alerts.rs +++ b/backend/windmill-common/tests/jobs_waiting_alerts.rs @@ -193,6 +193,30 @@ mod tests { ); } + /// 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