From 39c0e7c4aa40e7921535ca1e8d5b4b8f00a47f6d Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 20 Jul 2026 05:07:11 +0000 Subject: [PATCH] fix(alerts): count time-window completions when classifying gates The saturation check read only concurrency_counter.job_uuids, but the limiter admits on running jobs plus those that ended inside concurrency_time_window_s, so a gate held shut purely by recent completions read as free and its backlog raised the same false capacity alert one completion later. Emitter change lives in windmill-ee-private (see ee-repo-ref.txt bump). Adds the regression test for that case and refreshes the query cache. --- ...7e2047df060d61b77f23f730d4a02e7181040.json | 35 -------------- ...5b324c10ddd7aa8ea92efe2a262dc83b80a04.json | 35 ++++++++++++++ backend/ee-repo-ref.txt | 2 +- .../tests/jobs_waiting_alerts.rs | 46 +++++++++++++++++-- 4 files changed, 79 insertions(+), 39 deletions(-) delete mode 100644 backend/.sqlx/query-978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040.json create mode 100644 backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json diff --git a/backend/.sqlx/query-978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040.json b/backend/.sqlx/query-978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040.json deleted file mode 100644 index 05b1772ebf..0000000000 --- a/backend/.sqlx/query-978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040.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 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\n ),\n classified AS (\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 ) >= 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": "978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040" -} diff --git a/backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json b/backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json new file mode 100644 index 0000000000..f2cc5a698d --- /dev/null +++ b/backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json @@ -0,0 +1,35 @@ +{ + "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 ", + "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": "e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index d559ff0d0a..0d980243d9 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0c71f555ed74a1c801cdc8e5e6c063afcfb14384 \ No newline at end of file +6911fbf69af0c7ced2965dc031096184b173d6c6 \ 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 eabfc12c7e..f387e1e60b 100644 --- a/backend/windmill-common/tests/jobs_waiting_alerts.rs +++ b/backend/windmill-common/tests/jobs_waiting_alerts.rs @@ -57,8 +57,10 @@ mod tests { /// 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, so `running == limit` - /// is a saturated gate. + /// `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!({}))) @@ -67,7 +69,7 @@ mod tests { sqlx::query!( "INSERT INTO concurrency_settings (hash, concurrency_key, concurrent_limit, concurrency_time_window_s) - VALUES ($1, $2, $3, 5)", + VALUES ($1, $2, $3, 600)", handle, key, limit, @@ -125,6 +127,22 @@ mod tests { } } + /// 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"); + } + } + async fn alert_messages(db: &Pool) -> Vec { sqlx::query_scalar!("SELECT message FROM alerts WHERE alert_type = 'critical_error'") .fetch_all(db) @@ -153,6 +171,28 @@ mod tests { ); } + /// 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" + ); + } + /// 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