From ec95817741a9ef57dd94d826326e0468bcde26b0 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 20 Jul 2026 06:15:18 +0000 Subject: [PATCH] fix(alerts): expire gate marks so a freed gate cannot hide an outage A mark records that a gate held the job when the limiter last looked at it, and only a worker re-attempting the job refreshes it. Nothing clears it when the gate later frees, so if workers then disappear the backlog is genuinely starved while still carrying marks. Store concurrency_gated_at rather than a boolean, so the alert can tell a mark the limiter is still refreshing from one that stopped advancing. Emitter and limiter changes live in windmill-ee-private (see ee-repo-ref.txt bump). --- ...39710761e6f5e9bff87e2a3250f8fd131e68.json} | 4 +- ...1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json | 41 -------------- ...e1c46dfb34550ffb103ff722e1579b1b346ac.json | 47 ++++++++++++++++ backend/ee-repo-ref.txt | 2 +- ...0720055519_concurrency_gated_flag.down.sql | 2 +- ...260720055519_concurrency_gated_flag.up.sql | 12 ++-- backend/summarized_schema.txt | 2 +- .../tests/jobs_waiting_alerts.rs | 56 ++++++++++++++----- 8 files changed, 100 insertions(+), 66 deletions(-) rename backend/.sqlx/{query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json => query-0e9776d85ff7c7cd7fcafbbe813839710761e6f5e9bff87e2a3250f8fd131e68.json} (72%) delete mode 100644 backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json create mode 100644 backend/.sqlx/query-44ae7df82f205b11530de9911a5e1c46dfb34550ffb103ff722e1579b1b346ac.json diff --git a/backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json b/backend/.sqlx/query-0e9776d85ff7c7cd7fcafbbe813839710761e6f5e9bff87e2a3250f8fd131e68.json similarity index 72% rename from backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json rename to backend/.sqlx/query-0e9776d85ff7c7cd7fcafbbe813839710761e6f5e9bff87e2a3250f8fd131e68.json index ae95dc079e..cc8b8e88f3 100644 --- a/backend/.sqlx/query-1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405.json +++ b/backend/.sqlx/query-0e9776d85ff7c7cd7fcafbbe813839710761e6f5e9bff87e2a3250f8fd131e68.json @@ -1,6 +1,6 @@ { "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", + "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_at = now()\n WHERE id = $2", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405" + "hash": "0e9776d85ff7c7cd7fcafbbe813839710761e6f5e9bff87e2a3250f8fd131e68" } diff --git a/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json b/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json deleted file mode 100644 index 0811b1f053..0000000000 --- a/backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json +++ /dev/null @@ -1,41 +0,0 @@ -{ - "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-44ae7df82f205b11530de9911a5e1c46dfb34550ffb103ff722e1579b1b346ac.json b/backend/.sqlx/query-44ae7df82f205b11530de9911a5e1c46dfb34550ffb103ff722e1579b1b346ac.json new file mode 100644 index 0000000000..f2ebc8f7e9 --- /dev/null +++ b/backend/.sqlx/query-44ae7df82f205b11530de9911a5e1c46dfb34550ffb103ff722e1579b1b346ac.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT COUNT(*) FILTER (WHERE overdue AND concurrency_gated_at IS NULL)\n AS ungated_count,\n COUNT(*) FILTER (WHERE overdue AND concurrency_gated_at IS NOT NULL)\n AS gated_count,\n MIN(scheduled_for) FILTER (WHERE overdue AND concurrency_gated_at IS NULL)\n AS oldest_ungated,\n MIN(scheduled_for) FILTER (WHERE overdue) AS oldest_overdue,\n MAX(concurrency_gated_at) AS newest_mark\n FROM (\n SELECT scheduled_for, concurrency_gated_at,\n scheduled_for <= NOW() - $2::interval AS overdue\n FROM v2_job_queue\n WHERE tag = $1 AND running = false\n ) t\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "ungated_count", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "gated_count", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "oldest_ungated", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "oldest_overdue", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "newest_mark", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Interval" + ] + }, + "nullable": [ + null, + null, + null, + null, + null + ] + }, + "hash": "44ae7df82f205b11530de9911a5e1c46dfb34550ffb103ff722e1579b1b346ac" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 14b15ccfc5..0e5025d236 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -3984a0f87d24770f860a315061a26c9e53368b2e \ No newline at end of file +a1322ea50a7e8a0952f19d45b2c6a45936dc39a4 \ 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 index e2f0d4a7b1..8f2f216d67 100644 --- a/backend/migrations/20260720055519_concurrency_gated_flag.down.sql +++ b/backend/migrations/20260720055519_concurrency_gated_flag.down.sql @@ -1 +1 @@ -ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gated; +ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gated_at; diff --git a/backend/migrations/20260720055519_concurrency_gated_flag.up.sql b/backend/migrations/20260720055519_concurrency_gated_flag.up.sql index a825e2e886..2b7864500f 100644 --- a/backend/migrations/20260720055519_concurrency_gated_flag.up.sql +++ b/backend/migrations/20260720055519_concurrency_gated_flag.up.sql @@ -1,7 +1,5 @@ --- 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; +-- When the concurrency limiter last re-queued this job for want of a free slot. +-- Records that a gate parked the job rather than a worker being unavailable. +-- A timestamp rather than a flag: the gate can free later, and only a fresh +-- mark means the limiter is still parking. Nullable so adding it is instant. +ALTER TABLE v2_job_queue ADD COLUMN IF NOT EXISTS concurrency_gated_at TIMESTAMPTZ; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index b21d1d26d4..60e0120c45 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), concurrency_gated(bool) +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_at(ts) 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 d3622408e1..bbeed2c491 100644 --- a/backend/windmill-common/tests/jobs_waiting_alerts.rs +++ b/backend/windmill-common/tests/jobs_waiting_alerts.rs @@ -53,16 +53,19 @@ mod tests { .expect("seed alert config"); } - /// 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) { + /// Queue `n` jobs on TAG that have been due for 5 minutes. `gated_secs_ago` + /// mirrors the stamp the limiter writes when it re-queues a job it could not + /// admit; `None` is a job no gate has parked. + async fn queue_jobs(db: &Pool, n: usize, gated_secs_ago: Option) { for _ in 0..n { sqlx::query!( - "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)", + "INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for, concurrency_gated_at) + VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes', + CASE WHEN $3::bigint IS NULL THEN NULL + ELSE NOW() - INTERVAL '1 second' * $3::bigint END)", Uuid::new_v4(), TAG, - gated.then_some(true), + gated_secs_ago, ) .execute(db) .await @@ -78,16 +81,15 @@ mod tests { } /// 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 - /// while the fleet sits idle. + /// on-call: they wait by design, not for want of a worker, and no operator + /// action drains them. Their re-queue timestamps mature in bursts, so + /// counting them alerts repeatedly while the fleet sits idle. #[ignore = "requires database setup - run with --ignored flag"] #[sqlx::test(migrations = "../migrations")] async fn concurrency_gated_backlog_does_not_alert(db: Pool) { free_the_lock(&db).await; configure_alert(&db, 100).await; - queue_jobs(&db, 200, true).await; + queue_jobs(&db, 200, Some(1)).await; jobs_waiting_alerts(&db).await; @@ -105,8 +107,8 @@ mod tests { async fn jobs_waiting_on_capacity_still_alert(db: Pool) { free_the_lock(&db).await; configure_alert(&db, 100).await; - queue_jobs(&db, 150, false).await; - queue_jobs(&db, 200, true).await; + queue_jobs(&db, 150, None).await; + queue_jobs(&db, 200, Some(1)).await; jobs_waiting_alerts(&db).await; @@ -127,4 +129,32 @@ mod tests { messages[0] ); } + + /// A mark is only evidence while the limiter keeps refreshing it. Once a gate + /// frees, nothing clears the marks it left, so if workers then disappear the + /// backlog is genuinely starved while still carrying them. Marks that stopped + /// advancing must not be trusted, or the outage stays silent forever. + #[ignore = "requires database setup - run with --ignored flag"] + #[sqlx::test(migrations = "../migrations")] + async fn stale_gate_marks_do_not_suppress_the_alert(db: Pool) { + free_the_lock(&db).await; + configure_alert(&db, 100).await; + // Marked well beyond alert_time_threshold_seconds and never refreshed. + queue_jobs(&db, 200, Some(3600)).await; + + jobs_waiting_alerts(&db).await; + + let messages = alert_messages(&db).await; + assert_eq!( + messages.len(), + 1, + "a backlog whose gate marks stopped advancing is waiting on workers and \ + must alert, got {messages:?}" + ); + assert!( + messages[0].contains("200"), + "every job must be counted once its marks go stale: {}", + messages[0] + ); + } }