perf(monitor): hash active-root exclusion in retention delete (WIN-2088) (#9732)

* perf(monitor): hash active-root exclusion in retention delete

The expired-job retention delete (delete_expired_jobs_batch) excluded jobs
belonging to still-active root flows with
`COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)`. That
ScalarArrayOp is evaluated per candidate row as a linear scan of $3, so cost
grows with the number of active root jobs.

Express the exclusion as `NOT IN (SELECT u FROM unnest($3) u WHERE u IS NOT
NULL)` instead. The subquery form lets Postgres build a one-time hashed SubPlan
and apply it as a filter on the ordered index scan, giving O(1) membership per
candidate while preserving the `ORDER BY completed_at ASC LIMIT` early
termination. The `u IS NOT NULL` guard sidesteps NOT IN's null-trap semantics
($3 holds non-null PK ids).

Measured on a 2M-row synthetic v2_job_completed (batch LIMIT 20000, 5-run min):

  active roots | != ALL (before) | NOT IN hashed (after)
  -------------|-----------------|----------------------
  100          | 108 ms          | 104 ms
  1000         | 168 ms          | 105 ms
  10000        | 719 ms          | 131 ms

Both forms return identical row sets (verified via EXCEPT, 0 diff). Neutral at
small active-root counts, ~5.5x faster when many flows are active.

Relates to WIN-2088

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* perf(monitor): apply hashed active-root exclusion to log_cleanup mirror

windmill-api-settings/log_cleanup.rs::delete_expired_jobs_batch carries a
byte-identical copy of the retention delete and shared its prepared-query
cache. Updating only monitor.rs removed that shared cache entry and broke the
SQLX_OFFLINE build of the mirror. Apply the same NOT IN (hashed SubPlan)
rewrite so both copies converge on one cached query and the mirror gets the
same speedup.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_completed\n         WHERE id IN (\n             SELECT jc.id FROM v2_job_completed jc\n             LEFT JOIN v2_job j ON j.id = jc.id\n             WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)\n             ORDER BY jc.completed_at ASC\n             LIMIT $2\n             FOR UPDATE OF jc SKIP LOCKED\n         )\n         RETURNING id",
+  "query": "DELETE FROM v2_job_completed\n         WHERE id IN (\n             SELECT jc.id FROM v2_job_completed jc\n             LEFT JOIN v2_job j ON j.id = jc.id\n             WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN (\n                   SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL\n               )\n             ORDER BY jc.completed_at ASC\n             LIMIT $2\n             FOR UPDATE OF jc SKIP LOCKED\n         )\n         RETURNING id",
   "describe": {
     "columns": [
       {
@@ -20,5 +20,5 @@
       false
     ]
   },
-  "hash": "45997fcb4d9d62c7f7011966bf59bdb86e12ce1d0c8e925e738d2645121a5c1f"
+  "hash": "fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9"
 }

backend/src/monitor.rs

@@ -1532,14 +1532,22 @@ async fn delete_expired_jobs_batch(
     .await?;
 
     // Use FOR UPDATE SKIP LOCKED to avoid contention between replicas
-    // ORDER BY completed_at ensures we delete oldest jobs first
+    // ORDER BY completed_at ensures we delete oldest jobs first.
+    // Active-root exclusion uses `NOT IN (SELECT ... unnest($3))` rather than
+    // `!= ALL($3)`: the subquery form lets the planner build a one-time hashed
+    // SubPlan and apply it as a filter on the ordered index scan, giving O(1)
+    // membership per candidate instead of a per-row linear array scan (which
+    // degrades sharply when many root jobs are active). The `u IS NOT NULL` guard
+    // sidesteps NOT IN's null-trap semantics ($3 holds non-null PK ids).
     let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
         "DELETE FROM v2_job_completed
          WHERE id IN (
              SELECT jc.id FROM v2_job_completed jc
              LEFT JOIN v2_job j ON j.id = jc.id
              WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
-               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN (
+                   SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL
+               )
              ORDER BY jc.completed_at ASC
              LIMIT $2
              FOR UPDATE OF jc SKIP LOCKED

backend/windmill-api-settings/src/log_cleanup.rs

@@ -395,13 +395,17 @@ async fn delete_expired_jobs_batch(
     .fetch_all(&mut *tx)
     .await?;
 
+    // Active-root exclusion via NOT IN (hashed SubPlan) instead of `!= ALL($3)`;
+    // see backend/src/monitor.rs::delete_expired_jobs_batch for the rationale.
     let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
         "DELETE FROM v2_job_completed
          WHERE id IN (
              SELECT jc.id FROM v2_job_completed jc
              LEFT JOIN v2_job j ON j.id = jc.id
              WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
-               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN (
+                   SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL
+               )
              ORDER BY jc.completed_at ASC
              LIMIT $2
              FOR UPDATE OF jc SKIP LOCKED