perf: index the suspended-job resume test instead of filtering it (#10863)

* perf: index the suspended-job resume test instead of filtering it

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SEUq14Wz4cC2NzcRyo6CNj

* fix: keep the legacy suspended index until the replacement is recorded

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SEUq14Wz4cC2NzcRyo6CNj

* perf: drop the redundant suspend_until column from the suspended index

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SEUq14Wz4cC2NzcRyo6CNj

---------

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

backend/migrations/20260826202939_queue_suspended_resume_at_index.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS queue_suspended_v2;

backend/migrations/20260826202939_queue_suspended_resume_at_index.up.sql

@@ -0,0 +1,20 @@
+-- Serves the suspended-job pull in windmill-common/src/worker.rs, whose resume test is the
+-- indexed CASE expression. Two things about the shape are load-bearing:
+--   * (priority DESC NULLS LAST, created_at) leads, so the scan yields that query's ORDER BY
+--     and stops at the first match rather than sorting.
+--   * the index is dropped before it is built rather than relying on IF NOT EXISTS. The
+--     OVERRIDDEN_MIGRATIONS rewrite in windmill-api/src/db.rs runs these CONCURRENTLY, and an
+--     interrupted concurrent build leaves the index present but invalid, which IF NOT EXISTS
+--     would then skip rebuilding. Retiring the index this replaces is left to the migration
+--     that follows, so this one can only ever be replayed while that index is still there to
+--     cover the rebuild.
+DROP INDEX IF EXISTS queue_suspended_v2;
+
+CREATE INDEX IF NOT EXISTS queue_suspended_v2
+    ON v2_job_queue (
+        priority DESC NULLS LAST,
+        created_at,
+        (CASE WHEN suspend <= 0 THEN '-infinity'::timestamptz ELSE suspend_until END),
+        tag
+    )
+    WHERE suspend_until IS NOT NULL;

backend/migrations/20260826214706_queue_suspended_drop_legacy_index.down.sql

@@ -0,0 +1,3 @@
+CREATE INDEX IF NOT EXISTS queue_suspended
+    ON v2_job_queue (priority DESC NULLS LAST, created_at, suspend_until, suspend, tag)
+    WHERE suspend_until IS NOT NULL;

backend/migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql

@@ -0,0 +1,6 @@
+-- Retires the index queue_suspended_v2 replaces. Separate from the migration that builds it
+-- so that one is only ever replayed while this index still exists: sqlx records a migration
+-- only after all its statements run, so a process that dies before the record is written
+-- replays the build, and its leading DROP would otherwise be destroying the sole usable
+-- index rather than an interrupted build.
+DROP INDEX IF EXISTS queue_suspended;

backend/tests/suspended_pull_index.rs

@@ -0,0 +1,77 @@
+//! Pins the plan of the suspended-job pull. Its resume test degrades silently: once the
+//! query expression and `queue_suspended_v2` stop matching, Postgres still returns the right
+//! job, just by falling back to a heap filter and fetching one tuple per suspended row on
+//! every worker poll. No functional test can see that, so assert on the plan instead.
+
+use serde_json::Value;
+use sqlx::{Pool, Postgres};
+use windmill_common::worker::make_suspended_pull_query;
+
+/// Depth-first walk of an `EXPLAIN (FORMAT JSON)` plan tree.
+fn nodes(plan: &Value, out: &mut Vec<Value>) {
+    out.push(plan.clone());
+    for child in plan["Plans"].as_array().unwrap_or(&vec![]) {
+        nodes(child, out);
+    }
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn suspended_pull_tests_resume_time_inside_the_index(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO v2_job_queue (id, workspace_id, created_at, scheduled_for, running, suspend, suspend_until, tag)
+         SELECT gen_random_uuid(), 'test-workspace', now() - make_interval(secs => i),
+                now(), true, 1 + (i % 3), now() + interval '7 day', 'flow'
+         FROM generate_series(1, 2000) i",
+    )
+    .execute(&db)
+    .await?;
+    sqlx::query("ANALYZE v2_job_queue").execute(&db).await?;
+
+    // Both plans are cheap on a 2000-row table, and which one wins there says nothing
+    // about a queue with a large suspended backlog. Force the index path, which is the
+    // one production takes, and assert on how it evaluates the resume test.
+    let mut conn = db.acquire().await?;
+    sqlx::query("SET enable_seqscan = off")
+        .execute(&mut *conn)
+        .await?;
+    let version: String = sqlx::query_scalar("SELECT version()")
+        .fetch_one(&mut *conn)
+        .await?;
+    // FORMAT JSON rather than the default: `Index Cond` and `Filter` are separate keys on the
+    // node, so this does not ride on EXPLAIN's line layout staying put across a major bump.
+    let explained: Value = sqlx::query_scalar(&format!(
+        "EXPLAIN (FORMAT JSON) {}",
+        make_suspended_pull_query(&["flow".to_string()])
+    ))
+    .bind("test-worker")
+    .fetch_one(&mut *conn)
+    .await?;
+
+    let mut all = vec![];
+    nodes(&explained[0]["Plan"], &mut all);
+    let pretty = serde_json::to_string_pretty(&explained)?;
+    let scan = all
+        .iter()
+        .find(|n| n["Index Name"] == "queue_suspended_v2")
+        .unwrap_or_else(|| {
+            panic!("suspended pull did not scan queue_suspended_v2 on {version}:\n{pretty}")
+        });
+    // Only `Index Cond` is checked against the index tuple, so that is where the resume test
+    // has to land — as a `Filter` it would cost a heap fetch per suspended row. The residual
+    // `suspend_until IS NOT NULL` filter is not that: it is always true for rows the partial
+    // index holds, and only ever runs on the row LIMIT 1 already fetched.
+    let cond = scan["Index Cond"].as_str().unwrap_or_else(|| {
+        panic!("no Index Cond on the suspended pull scan on {version}:\n{pretty}")
+    });
+    assert!(
+        cond.contains("CASE WHEN"),
+        "resume test is not an index condition on {version}:\n{pretty}"
+    );
+    assert!(
+        !scan["Filter"].as_str().unwrap_or("").contains("CASE WHEN"),
+        "resume test fell back to a heap filter on {version}:\n{pretty}"
+    );
+    Ok(())
+}

backend/windmill-api/src/db.rs

@@ -102,6 +102,12 @@ lazy_static::lazy_static! {
                     (20260727151319, include_str!(
                         "../../migrations/20260727151319_draft_only_listing_indexes.up.sql"
                     ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY")),
+                    (20260826202939, include_str!(
+                        "../../migrations/20260826202939_queue_suspended_resume_at_index.up.sql"
+                    ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")),
+                    (20260826214706, include_str!(
+                        "../../migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql"
+                    ).replace("DROP INDEX", "DROP INDEX CONCURRENTLY")),
                     ].into_iter().collect();
 }
 
@@ -228,6 +234,8 @@ impl Migrate for CustomMigrator {
                     // CONCURRENTLY operations cannot run inside a transaction block
                     // or a multi-statement query (PostgreSQL requires top-level execution).
                     // Split into individual statements and execute each separately.
+                    // The split is naive, so a `;` anywhere in an overridden migration —
+                    // inside a comment or a string literal included — splits mid-statement.
                     for stmt in migration_sql.split(';') {
                         let stmt = stmt.trim();
                         if !stmt.is_empty()

backend/windmill-common/src/worker.rs

@@ -734,11 +734,17 @@ fn format_pull_query(peek: String) -> String {
     r
 }
 
+// The `CASE` is `suspend <= 0 OR suspend_until <= now()` written as one indexable
+// expression, equivalent only under the `suspend_until IS NOT NULL` guard. It must stay in
+// sync with `queue_suspended_v2` (migration 20260826202939): if it no longer matches, the
+// test silently reverts to a heap filter over every suspended row on every worker poll.
 pub fn make_suspended_pull_query(tags: &[String]) -> String {
     format_pull_query(format!(
         "SELECT id
         FROM v2_job_queue
-        WHERE suspend_until IS NOT NULL AND (suspend <= 0 OR suspend_until <= now()) AND tag IN ({})
+        WHERE suspend_until IS NOT NULL
+          AND (CASE WHEN suspend <= 0 THEN '-infinity'::timestamptz ELSE suspend_until END) <= now()
+          AND tag IN ({})
         ORDER BY priority DESC NULLS LAST, created_at
         FOR UPDATE SKIP LOCKED
         LIMIT 1",