fix: add index to scheduled_jobs next_run_at column (#12884)

- Adds index only to the next_run_at column to improve the pull method
throughput and time.
- Adds a 80% fillfactor for the postgres table to enhance the HOT% for
updates. One minor caveat is that, since the next_run_at is now indexed,
for rescheduling updates (next_run_at update), it will not be able to
utilize Heap Only Tuples and can add to the index/dead tuple bloating in
the postgres db. The pulls, keep alives and watch timeout can still use
the HOT approach because they don't touch the next_run_at column.

Note in the results that, the adv lock wait time reduced drastically
after using the index.

Another thing to note that there is a difference from the file_list
table where the dead tuples caused bad search performance when there
were huge number of records in the table (1TB+ table size). In that
case, the auto vacuum didn't help much because, the amount of dead
tuples required to trigger auto vacuum, itself was very big - resulting
in bad index bloating. In the scheduled_jobs, in most cases, almost all
the rows are updated frequently, resulting in relatively higher
percentage of dead tuples and hence trigger auto vacuum more frequently.

The above is visible from the below test results as well - for baseline
(current main branch approach), the HOT% is very good (91%) because the
updates didn't touch the indexes at all and hence, low no. of dead
tuples. For index_ff100 (index on next_run_at + default 100%
fillfactor), it already triggers 1 autovacuum. Same for 80% fill factor.


Results (100K rows, 8 clients, 60s)

### Tier 1 — pull query plan

| variant | plan | exec time | shared buffers |

|---------------|-----------------------------------|-----------:|---------------:|
| `baseline` | Parallel Seq Scan + top-N Sort | **13.5 ms**| 4424 |
| `index_ff100` | Index Scan `scheduled_jobs_pull_idx` | **0.039 ms** |
7 |
| `optimized` | Index Scan `scheduled_jobs_pull_idx` | **0.030 ms** | 7
|

→ **~350× faster** at 100K, and the gap **grows with table size** (seq
scan is O(rows), index is
O(log n + limit)). Fillfactor has no effect on the read plan, as
expected.

### Tier 2 — throughput, lock-wait, churn health

| metric (per variant) | `baseline` | `index_ff100` | `optimized` |

|----------------------------------|-----------:|--------------:|------------:|
| **total tps** | 75.7 | **930.6** | 875.3 |
| avg txn latency | 105.6 ms | 8.6 ms | 9.1 ms |
| **advisory-lock wait** (pull) | **177 ms** | **2.4 ms** | 1.7 ms |
| pull claim UPDATE | 26.4 ms | 0.93 ms | 0.80 ms |
| updates in 60s (`n_tup_upd`) | 22.5 K | 278.8 K | 261.9 K |
| **HOT %** (`n_tup_hot_upd`) | 91.3 % | 47.0 % | 50.2 % |
| index size | 18 MB | 28 MB | **25 MB** |
| true dead-tuple % (pgstattuple) | 0.10 % | 0.84 % | **0.11 %** |
| free space % (pgstattuple) | 3.3 % | 9.2 % | **19.1 %** |
| autovacuum runs | 0 | 1 | 1 |

src/infra/src/scheduler/postgres.rs

@@ -81,6 +81,19 @@ CREATE TABLE IF NOT EXISTS scheduled_jobs
         // drop created_at column for old version <= 0.40.0
         drop_column("scheduled_jobs", "created_at").await?;
 
+        // `scheduled_jobs` is a high-churn table: every job cycle issues several UPDATEs to the
+        // same row (pull -> Processing, keep_alive, completion, timeout). Lowering fillfactor
+        // leaves free space in each heap page so updates that don't touch an indexed column can be
+        // applied HOT (Heap-Only Tuple) — no new index entries, on-page dead-tuple pruning —
+        // cutting index/MVCC bloat from the churn. This pays off only because the pull index is
+        // keyed on `next_run_at` alone (see create_table_index): pull, keep_alive, watch_timeout
+        // and status-only completion updates all stay HOT; just the reschedule path (which
+        // rewrites the indexed `next_run_at`) drops out, which is irreducible. Idempotent metadata
+        // change; it governs future page writes and does not rewrite existing pages.
+        sqlx::query("ALTER TABLE scheduled_jobs SET (fillfactor = 80);")
+            .execute(&pool)
+            .await?;
+
         Ok(())
     }
 
@@ -107,6 +120,31 @@ CREATE TABLE IF NOT EXISTS scheduled_jobs
         ))
         .await?;
 
+        // Index for the hot pull predicate: `WHERE status = Waiting AND next_run_at <= now
+        // ORDER BY next_run_at`. Without it the puller does a seq scan + sort that grows with the
+        // table (and with multi-tenancy), and every autovacuum has more to do.
+        //
+        // Keyed on `next_run_at` ALONE (no `status`), deliberately:
+        //   * `status` is non-selective here — between runs a job sits in Waiting, so Waiting is
+        //     the dominant state; a partial `WHERE status = 0` index would be ~the same size as
+        //     this one while buying nothing.
+        //   * Leaving `status` out of the key/predicate/INCLUDE keeps HOT updates alive (see the
+        //     fillfactor note in create_table). The pull, watch_timeout and status-only completion
+        //     updates flip `status` but NOT `next_run_at`, so they stay HOT. Only the reschedule
+        //     path (which rewrites `next_run_at`) drops HOT — unavoidable, since the pull needs
+        //     `next_run_at` indexed. Putting `status` in the index (key, partial predicate, or
+        //     INCLUDE) would block HOT on every status flip and defeat the fillfactor change.
+        //
+        // The pull pays a few extra heap visits to filter `status` (bounded by the due +
+        // Processing set) — cheap next to the seq-scan + sort it replaces. Raw SQL since
+        // `create_index` can't emit it; `IF NOT EXISTS` keeps it idempotent across restarts.
+        let pool = CLIENT_DDL.clone();
+        sqlx::query(
+            "CREATE INDEX IF NOT EXISTS scheduled_jobs_pull_idx ON scheduled_jobs (next_run_at);",
+        )
+        .execute(&pool)
+        .await?;
+
         Ok(())
     }
 
@@ -576,12 +614,19 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s
 
     /// Returns the Trigger jobs with "Waiting" status.
     /// Steps:
-    /// - Read the records with status "Waiting", oldest createdAt, next_run_at <= now and limit =
-    ///   `concurrency`
-    /// - Skip locked rows and lock the read rows with "FOR UPDATE SKIP LOCKED"
+    /// - Take a global `pg_advisory_xact_lock` so only one puller across all nodes claims jobs at a
+    ///   time. This serializes the pull cluster-wide; the query itself does **not** use `FOR UPDATE
+    ///   SKIP LOCKED`.
+    /// - Read the records with status "Waiting", next_run_at <= now, ordered by next_run_at and
+    ///   limited to `concurrency`
     /// - Changes their statuses from "Waiting" to "Processing"
     /// - Commits as a single transaction
     /// - Returns the Trigger jobs
+    ///
+    /// Lock semantics: `pg_advisory_xact_lock` is **transaction-scoped** — it is released
+    /// automatically on COMMIT/ROLLBACK, never outliving the transaction. This is the safe variant
+    /// on pooled sqlx connections; a session-scoped lock (`pg_advisory_lock`) could leak onto a
+    /// connection returned to the pool and is deliberately avoided here.
     async fn pull(
         &self,
         concurrency: i64,
@@ -618,7 +663,8 @@ RETURNING *;"#;
 
         let mut tx = pool.begin().await?;
 
-        // Lock the table for the duration of the transaction
+        // Take a transaction-scoped advisory lock (not a table lock) for the duration of the
+        // transaction; auto-released on commit/rollback. Serializes the pull across all nodes.
         let lock_key = "scheduler_pull_lock";
         let lock_id = config::utils::hash::gxhash::new().sum64(lock_key);
         let lock_id = if lock_id > i64::MAX as u64 {