[scheduler] Avoid starvation (#2463)

* **Bug Fixes**
* Enforced per-job core caps during layer dispatching by clamping frame
core reservations to the job’s remaining allowance
* Added an early job-at-cap pre-check to avoid checkout attempts when a
job is already at or over its configured core limit
* Updated job pending/capacity calculations to use live “booked”
core/GPU usage from active processes

* **New Features**
* Added a Prometheus counter to track when job-level core-cap pre-checks
skip layer matching

* **Chores**
  * Improved formatting and updated/added unit tests for cap behavior


## LLM Usage Disclaimer

Claude Code Opus was used to help design and implement the changes on
this PR

---------

Signed-off-by: Diego Tavares <dtavares@imageworks.com>

Author: DiegoTavares

rust/crates/scheduler/src/cluster.rs

@@ -682,8 +682,7 @@ impl ClusterFeed {
                         {
                             let mut by_type: HashMap<&'static str, i64> = HashMap::new();
                             {
-                                let clusters =
-                                    feed.read().unwrap_or_else(|p| p.into_inner());
+                                let clusters = feed.read().unwrap_or_else(|p| p.into_inner());
                                 for c in clusters.iter() {
                                     *by_type.entry(c.cluster_type()).or_default() += 1;
                                 }

rust/crates/scheduler/src/dao/job_dao.rs

@@ -65,18 +65,58 @@ impl DispatchJob {
 }
 
 static QUERY_PENDING_BY_SHOW_FACILITY_TAG: &str = r#"
--- bookable_shows: shows that still have room in at least one subscription.
-WITH bookable_shows AS (
+-- LIVE booked-core CTEs. The job/folder/subscription caps must be gated on CURRENT
+-- usage, but the PG columns job_resource.int_cores / folder_resource.int_cores /
+-- subscription.int_cores are only materialized by the ~120s recompute loop, so they
+-- lag reality. `proc` is the transactionally-accurate source (the scheduler inserts on
+-- booking, Cuebot deletes on frame completion, compensation deletes on a failed launch),
+-- so we sum it directly. Each CTE is scoped to this show ($1) via i_proc_pkshow and
+-- mirrors the recompute aggregation (so the value equals a fresher copy of the PG
+-- column), and joins on indexed pk_host / pk_job. Gating on stale PG would otherwise
+-- (a) over-fetch jobs for caps that are full in Redis -> wasted rejections, and worse
+-- (b) FALSE-EXCLUDE: a frame completes and frees burst live, but the lagged PG column
+-- stays high for up to a cycle, dropping the show/folder/job from the fetch and starving
+-- its (esp. low-priority) jobs until the next recompute.
+WITH job_live AS (
+    -- per-job booked cores (mirrors RECOMPUTE_JOB_RESOURCE_FROM_PROC)
+    SELECT p.pk_job, COALESCE(SUM(p.int_cores_reserved), 0)::int AS cores
+    FROM proc p
+    WHERE p.pk_show = $1
+    GROUP BY p.pk_job
+),
+folder_live AS (
+    -- per-folder booked cores/gpus (mirrors RECOMPUTE_FOLDER_RESOURCE_FROM_PROC)
+    SELECT j2.pk_folder,
+           COALESCE(SUM(p.int_cores_reserved), 0)::int AS cores,
+           COALESCE(SUM(p.int_gpus_reserved), 0)::int  AS gpus
+    FROM proc p
+    JOIN job j2 ON j2.pk_job = p.pk_job AND j2.str_state <> 'FINISHED'
+    WHERE p.pk_show = $1
+    GROUP BY j2.pk_folder
+),
+sub_live AS (
+    -- per-alloc booked cores for this show (mirrors RECOMPUTE_SUBSCRIPTION_FROM_PROC:
+    -- proc -> host -> alloc, excluding local/desktop bookings)
+    SELECT h.pk_alloc, COALESCE(SUM(p.int_cores_reserved), 0)::int AS cores
+    FROM proc p
+    JOIN host h ON h.pk_host = p.pk_host
+    WHERE p.pk_show = $1 AND p.b_local = false
+    GROUP BY h.pk_alloc
+),
+-- bookable_shows: shows that still have room in at least one subscription (LIVE).
+bookable_shows AS (
     SELECT DISTINCT w.pk_show, sh.str_name AS show_name
     FROM subscription s
+    -- LEFT JOIN: an alloc with no booked procs has full headroom.
+    LEFT JOIN sub_live sl ON sl.pk_alloc = s.pk_alloc
     INNER JOIN vs_waiting w ON s.pk_show = w.pk_show
     INNER JOIN show sh ON sh.pk_show = w.pk_show
     WHERE s.pk_show = $1
         -- Burst == 0 is used to freeze a subscription.
         AND s.int_burst > 0
-        -- At least one core unit available.
-        AND s.int_burst - s.int_cores >= $2
-        AND s.int_cores < s.int_burst
+        -- At least one core unit available, measured against LIVE usage.
+        AND s.int_burst - COALESCE(sl.cores, 0) >= $2
+        AND COALESCE(sl.cores, 0) < s.int_burst
 )
 SELECT
     j.pk_job,
@@ -87,15 +127,19 @@ INNER JOIN bookable_shows bs ON j.pk_show = bs.pk_show
 INNER JOIN job_resource jr ON j.pk_job = jr.pk_job
 INNER JOIN folder f ON j.pk_folder = f.pk_folder
 INNER JOIN folder_resource fr ON f.pk_folder = fr.pk_folder
+-- LEFT JOIN the live CTEs: no row => 0 booked => full headroom.
+LEFT JOIN job_live jl ON jl.pk_job = j.pk_job
+LEFT JOIN folder_live fl ON fl.pk_folder = f.pk_folder
 WHERE j.str_state = 'PENDING'
     AND j.b_paused = false
     AND j.pk_facility = $4
-    -- Folder must have any room at all; per-layer fit is checked below.
-    AND (fr.int_max_cores = -1 OR fr.int_cores < fr.int_max_cores)
-    AND (fr.int_max_gpus = -1 OR fr.int_gpus < fr.int_max_gpus)
+    -- Folder must have any room at all (LIVE); per-layer fit is checked below.
+    AND (fr.int_max_cores <= 0 OR COALESCE(fl.cores, 0) < fr.int_max_cores)
+    AND (fr.int_max_gpus <= 0 OR COALESCE(fl.gpus, 0) < fr.int_max_gpus)
     -- The job must have at least one layer that matches the tag set, has waiting
-    -- frames, and fits within the folder cap. EXISTS short-circuits per job and
-    -- avoids the cardinality blowup of joining layer + layer_stat at the outer level.
+    -- frames, and fits within the folder AND job caps (both LIVE). EXISTS short-circuits
+    -- per job and avoids the cardinality blowup of joining layer + layer_stat at the
+    -- outer level.
     AND EXISTS (
         SELECT 1
         FROM layer l
@@ -104,8 +148,9 @@ WHERE j.str_state = 'PENDING'
           AND ls.int_waiting_count > 0
           AND string_to_array(REPLACE($3, ' ', ''), '|')
               && string_to_array(REPLACE(l.str_tags, ' ', ''), '|')
-          AND (fr.int_max_cores = -1 OR fr.int_cores + l.int_cores_min <= fr.int_max_cores)
-          AND (fr.int_max_gpus = -1 OR fr.int_gpus + l.int_gpus_min <= fr.int_max_gpus)
+          AND (fr.int_max_cores <= 0 OR COALESCE(fl.cores, 0) + l.int_cores_min <= fr.int_max_cores)
+          AND (fr.int_max_gpus <= 0 OR COALESCE(fl.gpus, 0) + l.int_gpus_min <= fr.int_max_gpus)
+          AND (jr.int_max_cores <= 0 OR COALESCE(jl.cores, 0) + l.int_cores_min <= jr.int_max_cores)
     )
 ORDER BY jr.int_priority DESC
 LIMIT $5

rust/crates/scheduler/src/metrics/mod.rs

@@ -13,9 +13,9 @@
 use axum::{response::IntoResponse, routing::get, Router};
 use lazy_static::lazy_static;
 use prometheus::{
-    register_counter, register_counter_vec, register_gauge, register_gauge_vec,
-    register_histogram, register_histogram_vec, Counter, CounterVec, Encoder, Gauge, GaugeVec,
-    Histogram, HistogramVec, TextEncoder,
+    register_counter, register_counter_vec, register_gauge, register_gauge_vec, register_histogram,
+    register_histogram_vec, Counter, CounterVec, Encoder, Gauge, GaugeVec, Histogram, HistogramVec,
+    TextEncoder,
 };
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
@@ -79,6 +79,18 @@ lazy_static! {
     )
     .expect("Failed to register accounting_limit_exceeded_total counter");
 
+    // Counts layers skipped by the matcher's job-level at-cap pre-check
+    // (`pipeline/matcher.rs`, `placement::job_at_core_cap`). A job already at its
+    // `job_max_cores` cap would otherwise be re-checked-out and re-rejected up to
+    // `host_candidate_attempts_per_layer` times every pass; the pre-check returns
+    // early instead. Kept distinct from `ACCOUNTING_LIMIT_EXCEEDED_TOTAL{table="job"}`
+    // (Lua-rejection pressure) so this reads cleanly as "wasted attempts avoided".
+    pub static ref JOB_CAP_PRECHECK_SKIP_TOTAL: Counter = register_counter!(
+        "scheduler_job_cap_precheck_skip_total",
+        "Layers skipped pre-checkout because the job is already at its core cap"
+    )
+    .expect("Failed to register job_cap_precheck_skip_total counter");
+
     // Job query metrics from dao/job_dao.rs
     pub static ref JOB_QUERY_DURATION_SECONDS: Histogram = register_histogram!(
         "scheduler_job_query_duration_seconds",
@@ -326,6 +338,12 @@ pub fn increment_accounting_limit_exceeded(table: &str) {
         .inc();
 }
 
+/// Records a layer skipped by the job-level at-cap pre-check.
+#[inline]
+pub fn increment_job_cap_precheck_skip() {
+    JOB_CAP_PRECHECK_SKIP_TOTAL.inc();
+}
+
 /// Helper function to observe job query duration
 #[inline]
 pub fn observe_job_query_duration(duration: Duration) {