fix(alerts): exclude cancelled rows from gate liveness, keep recovery scoped

A single freshly cancelled job, which no gate holds, vouched for every
stale mark on that gate, and recovery landed globally instead of with the
workspace the alert was scoped to.

Also fixes a test helper that cancelled every uncancelled job on the tag
rather than the rows it inserted, which made the new liveness regression
pass for the wrong reason.

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

backend/.sqlx/query-06122bae7f6660909e769034843beb98cba56635ae4068be0b8ae1c7665f84d6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id FROM alerts\n             WHERE resource = $1 AND workspace_id IS NOT NULL\n             ORDER BY created_at DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "06122bae7f6660909e769034843beb98cba56635ae4068be0b8ae1c7665f84d6"
+}

backend/.sqlx/query-2a78c3cdda61c13f4928646c671501aff9a6f3d762285c1e37dd54f263f9e0d0.json

@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for,\n                                           concurrency_gated_at, concurrency_gate_id, canceled_by)\n                 VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes',\n                         CASE WHEN $3::bigint IS NULL THEN NULL\n                              ELSE NOW() - INTERVAL '1 second' * $3::bigint END,\n                         $4,\n                         CASE WHEN $5 THEN 'canceller' ELSE NULL END)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Int8",
+        "Varchar",
+        "Bool"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2a78c3cdda61c13f4928646c671501aff9a6f3d762285c1e37dd54f263f9e0d0"
+}

backend/.sqlx/query-3c7fc6489a323edb40c59564a268e24e7029668f5005b198ff7b772cc1b1471f.json

@@ -1,6 +1,6 @@
 {
   "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                ",
+  "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                  AND canceled_by IS NULL\n                GROUP BY concurrency_gate_id\n                HAVING MAX(concurrency_gated_at) >= NOW() - $1::interval\n                ",
   "describe": {
     "columns": [
       {
@@ -18,5 +18,5 @@
       true
     ]
   },
-  "hash": "67ac13428f6d5360197a90e5a2262c2f29bcd88d9117d3504f84e705f38cbe04"
+  "hash": "3c7fc6489a323edb40c59564a268e24e7029668f5005b198ff7b772cc1b1471f"
 }

backend/.sqlx/query-d89df8677fe18414db77ce48f0846ace5841a949fc6adf8afaf314bd8890d2a4.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT workspace_id FROM alerts\n             WHERE alert_type = 'recovered_critical_error' ORDER BY created_at DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "d89df8677fe18414db77ce48f0846ace5841a949fc6adf8afaf314bd8890d2a4"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-a73c310dfda32bc145f9a495be543880a6cb2823
+8df33392d864b87731ab5742247b1aa3f6dacf80

backend/windmill-common/tests/concurrency_gate_alerts.rs

@@ -172,6 +172,20 @@ mod tests {
         free_the_lock(&db).await;
         concurrency_gate_alerts(&db).await;
 
+        let recovery_workspace: Option<String> = sqlx::query_scalar!(
+            "SELECT workspace_id FROM alerts
+             WHERE alert_type = 'recovered_critical_error' ORDER BY created_at DESC LIMIT 1"
+        )
+        .fetch_optional(&db)
+        .await
+        .expect("read recovery alert")
+        .flatten();
+        assert_eq!(
+            recovery_workspace.as_deref(),
+            Some("test-workspace"),
+            "recovery must stay scoped to the workspace the alert was raised for"
+        );
+
         let unresolved: i64 = sqlx::query_scalar!(
             "SELECT COUNT(*) FROM healthchecks
              WHERE check_type LIKE 'concurrency_gate_alert_%' AND healthy = false"

backend/windmill-common/tests/jobs_waiting_alerts.rs

@@ -66,15 +66,7 @@ mod tests {
     /// Soft-cancelled jobs keep whatever mark they carried, but the pull path
     /// skips the limiter for them, so they are waiting for a worker.
     async fn queue_canceled_jobs(db: &Pool<Postgres>, n: usize, gate: Option<(&str, i64)>) {
-        queue_jobs_on_tag(db, TAG, n, gate).await;
-        sqlx::query!(
-            "UPDATE v2_job_queue SET canceled_by = 'canceller'
-             WHERE tag = $1 AND canceled_by IS NULL",
-            TAG,
-        )
-        .execute(db)
-        .await
-        .expect("cancel jobs");
+        queue_jobs_inner(db, TAG, n, gate, true).await
     }
 
     async fn queue_jobs_on_tag(
@@ -82,19 +74,31 @@ mod tests {
         tag: &str,
         n: usize,
         gate: Option<(&str, i64)>,
+    ) {
+        queue_jobs_inner(db, tag, n, gate, false).await
+    }
+
+    async fn queue_jobs_inner(
+        db: &Pool<Postgres>,
+        tag: &str,
+        n: usize,
+        gate: Option<(&str, i64)>,
+        canceled: bool,
     ) {
         for _ in 0..n {
             sqlx::query!(
                 "INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for,
-                                           concurrency_gated_at, concurrency_gate_id)
+                                           concurrency_gated_at, concurrency_gate_id, canceled_by)
                  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,
-                         $4)",
+                         $4,
+                         CASE WHEN $5 THEN 'canceller' ELSE NULL END)",
                 Uuid::new_v4(),
                 tag,
                 gate.map(|(_, secs)| secs),
                 gate.map(|(key, _)| key),
+                canceled,
             )
             .execute(db)
             .await
@@ -255,4 +259,28 @@ mod tests {
             "cancelled jobs bypass the limiter and must still alert, got {messages:?}"
         );
     }
+
+    /// Gate liveness is an aggregate, so it has to exclude cancelled rows for the
+    /// same reason the per-job classifier does. Otherwise one freshly cancelled
+    /// job -- which no gate is holding -- vouches for a whole stale backlog on
+    /// that gate and buries it.
+    #[ignore = "requires database setup - run with --ignored flag"]
+    #[sqlx::test(migrations = "../migrations")]
+    async fn a_cancelled_job_does_not_keep_its_gate_live(db: Pool<Postgres>) {
+        free_the_lock(&db).await;
+        configure_alert(&db, 100).await;
+        // A stale backlog nothing is refreshing any more...
+        queue_jobs(&db, 200, Some((BUSY_GATE, 3600))).await;
+        // ...beside one cancelled job carrying a fresh mark for the same gate.
+        queue_canceled_jobs(&db, 1, Some((BUSY_GATE, 1))).await;
+
+        jobs_waiting_alerts(&db).await;
+
+        let messages = alert_messages(&db).await;
+        assert_eq!(
+            messages.len(),
+            1,
+            "a cancelled job must not vouch for its gate, got {messages:?}"
+        );
+    }
 }