perf(alerts): resolve live gates once per pass rather than per tag

Gate ids are global, so the liveness aggregate covers the whole queue and
running it per monitored tag let an unrelated backlog scale the cost of
every tag. It is now resolved once per alert config, skipped when marks
cannot be trusted, and backed by a partial index over parked jobs.

Emitter and limiter changes live in windmill-ee-private (see
ee-repo-ref.txt bump).

backend/.sqlx/query-07548bfb29b053e7ea6e915c67c81a89d09e226c4640ea476e05b192856bfbcb.json

@@ -1,41 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n                WITH live_gates AS (\n                    SELECT concurrency_gate_id AS id\n                    FROM v2_job_queue\n                    WHERE running = false AND concurrency_gate_id IS NOT NULL\n                    GROUP BY concurrency_gate_id\n                    HAVING MAX(concurrency_gated_at) >= NOW() - $2::interval\n                )\n                SELECT COUNT(*) FILTER (WHERE overdue AND g.id IS NULL) AS ungated_count,\n                       COUNT(*) FILTER (WHERE overdue AND g.id IS NOT NULL) AS gated_count,\n                       MIN(q.scheduled_for) FILTER (WHERE overdue AND g.id IS NULL)\n                           AS oldest_ungated,\n                       MIN(q.scheduled_for) FILTER (WHERE overdue) AS oldest_overdue\n                FROM (\n                    SELECT scheduled_for, concurrency_gate_id,\n                           scheduled_for <= NOW() - $2::interval AS overdue\n                    FROM v2_job_queue\n                    WHERE tag = $1 AND running = false\n                ) q\n                LEFT JOIN live_gates g ON g.id = q.concurrency_gate_id\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"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Interval"
-      ]
-    },
-    "nullable": [
-      null,
-      null,
-      null,
-      null
-    ]
-  },
-  "hash": "07548bfb29b053e7ea6e915c67c81a89d09e226c4640ea476e05b192856bfbcb"
-}

backend/.sqlx/query-428bd28304df56528604da2a5f83f0e989cb96acbd2eaff99e29110d96625f64.json

@@ -0,0 +1,36 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                SELECT COUNT(*) FILTER (WHERE overdue AND gated IS NOT TRUE) AS ungated_count,\n                       COUNT(*) FILTER (WHERE overdue AND gated) AS gated_count,\n                       MIN(scheduled_for) FILTER (WHERE overdue AND gated IS NOT TRUE)\n                           AS oldest_ungated\n                FROM (\n                    SELECT scheduled_for,\n                           scheduled_for <= NOW() - $2::interval AS overdue,\n                           concurrency_gate_id = ANY($3::text[]) AS gated\n                    FROM v2_job_queue\n                    WHERE tag = $1 AND running = false\n                ) q\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"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Interval",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "428bd28304df56528604da2a5f83f0e989cb96acbd2eaff99e29110d96625f64"
+}

backend/.sqlx/query-67ac13428f6d5360197a90e5a2262c2f29bcd88d9117d3504f84e705f38cbe04.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                SELECT concurrency_gate_id AS \"id!\"\n                FROM v2_job_queue\n                WHERE running = false AND concurrency_gate_id IS NOT NULL\n                GROUP BY concurrency_gate_id\n                HAVING MAX(concurrency_gated_at) >= NOW() - $1::interval\n                ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id!",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Interval"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "67ac13428f6d5360197a90e5a2262c2f29bcd88d9117d3504f84e705f38cbe04"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-5e449c771fc1fed928b40727378cd67c3380b5cb
+d3445d03c1f103f7fc5432a7d898a5ae8db728ee

backend/migrations/20260720055519_concurrency_gated_flag.down.sql

@@ -1,2 +1,3 @@
+DROP INDEX IF EXISTS v2_job_queue_concurrency_gate;
 ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gated_at;
 ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gate_id;

backend/migrations/20260720055519_concurrency_gated_flag.up.sql

@@ -4,3 +4,9 @@
 -- policies, and only jobs under the same one share a gate.
 ALTER TABLE v2_job_queue ADD COLUMN IF NOT EXISTS concurrency_gated_at TIMESTAMPTZ;
 ALTER TABLE v2_job_queue ADD COLUMN IF NOT EXISTS concurrency_gate_id VARCHAR(320);
+
+-- Serves the queue-depth alert's per-gate liveness aggregate. Partial so it only
+-- covers jobs a gate actually parked, leaving the ordinary queue path unindexed.
+CREATE INDEX IF NOT EXISTS v2_job_queue_concurrency_gate
+    ON v2_job_queue (concurrency_gate_id, concurrency_gated_at DESC)
+    WHERE running = false AND concurrency_gate_id IS NOT NULL;