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).

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"
 }

backend/.sqlx/query-23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5.json

@@ -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"
-}

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"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-3984a0f87d24770f860a315061a26c9e53368b2e
+a1322ea50a7e8a0952f19d45b2c6a45936dc39a4

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;

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;

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)

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<Postgres>, 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<Postgres>, n: usize, gated_secs_ago: Option<i64>) {
         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<Postgres>) {
         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<Postgres>) {
         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<Postgres>) {
+        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]
+        );
+    }
 }