[scheduler] Gate cluster feed on an awake-set scan (#2465)

The cluster feed round-robined the full cluster universe, whose size
scales with host/tag count rather than workload: a single show could fan
out to ~180 clusters (mostly hostname/manual) of which ~90% were idle,
so most passes ended in no_jobs and the round-trip p50 degraded as the
lap grew.

Add a periodic awake-gate scan that recomputes which (facility, show,
tag) tuples have plausibly-dispatchable work (JobDao::scan_active_tags),
and have the producer round-robin only that awake subset. Clusters with
no work stay out of the rotation until a later scan re-activates them;
the producer is woken immediately on an empty -> non-empty transition
via active_notify rather than waiting out the idle poll.

The scan query is a provable superset of the per-cluster dispatch query:
it keeps only a strict subset of that query's conjuncts
(PENDING/not-paused/ facility, a waiting layer carrying the tag, show
active+scheduler-managed) and drops the folder/job caps, so it can never
produce a false negative. Its subscription gate is the cheap static
`int_burst > 0` only -- the dynamic live-headroom check is deliberately
omitted to avoid an O(farm) full-`proc` aggregation every interval and
the over-count/false-exclude foot-gun that a global proc-sum would
carry. Burst enforcement stays in the per-cluster path, where an
over-burst-but-busy show backs off on cluster_saturated_sleep.

The existing per-pass sleep_map backoff is unchanged and still handles
the superset residual (clusters that are awake but turn out
capped/saturated on a real pass); the awake gate sits on top of it.

Adds config knob `active_scan_interval` (default 2s, the idle->dispatch
latency lever) and metrics scheduler_clusters_active,
scheduler_active_tags, and scheduler_active_scan_duration_seconds.
Covered by a new stress-test that asserts the scan is a superset of the
per-cluster query against live PG.

The remaining diff is rustfmt normalization.

## LLM Usage Disclosure

Some of the work of this PR was done using Claude Code Opus. 


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Scheduler now periodically focuses on clusters with dispatchable work,
reducing unnecessary re-checks and improving dispatch responsiveness.
  * Added new monitoring for the active cluster subset and scan timing.
* Configuration now includes a new interval to control how often active
clusters are re-scanned.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: DiegoTavares

rust/crates/scheduler/src/accounting/error.rs

@@ -41,7 +41,9 @@ pub enum AccountingError {
     /// per design §2.4), but the bootstrap reseed surfaces it as a startup gate so a
     /// scheduler never begins booking against an unseeded Redis. Carries the number of
     /// attempts made (`cas_max_retries + 1`) for diagnostics.
-    #[error("CAS contention exceeded retry budget after {attempts} attempts; reseed cycle skipped")]
+    #[error(
+        "CAS contention exceeded retry budget after {attempts} attempts; reseed cycle skipped"
+    )]
     CasContentionExceeded { attempts: u32 },
 
     #[error("accounting redis error: {0}")]

rust/crates/scheduler/src/accounting/managed_shows.rs

@@ -107,8 +107,7 @@ impl ManagedShowsCache {
                                     let added_set: HashSet<Uuid> = added.iter().copied().collect();
                                     let deferred: HashSet<Uuid> =
                                         new_set.difference(&added_set).copied().collect();
-                                    let mut lock =
-                                        inner.write().unwrap_or_else(|p| p.into_inner());
+                                    let mut lock = inner.write().unwrap_or_else(|p| p.into_inner());
                                     *lock = deferred;
                                     return;
                                 }

rust/crates/scheduler/src/accounting/recompute.rs

@@ -70,8 +70,7 @@ pub fn spawn_loop(service: Arc<AccountingService>) {
             last_dispatched = current_dispatched;
 
             let current_limit_exceeded = metrics::resource_limit_exceeded_session();
-            let limit_exceeded_delta =
-                current_limit_exceeded.saturating_sub(last_limit_exceeded);
+            let limit_exceeded_delta = current_limit_exceeded.saturating_sub(last_limit_exceeded);
             last_limit_exceeded = current_limit_exceeded;
 
             info!(

rust/crates/scheduler/src/cluster.rs

@@ -146,6 +146,14 @@ pub struct QueueConfig {
     /// The reload only swaps the live set when it actually changed.
     #[serde(with = "humantime_serde")]
     pub cluster_reload_interval: Duration,
+    /// Interval between awake-gate scans. Each scan runs a single query that
+    /// computes which clusters have plausibly-dispatchable work; clusters not in
+    /// that set are skipped by the producer until a later scan re-activates them.
+    /// This bounds idle->dispatch latency: a newly-launched job starts being
+    /// dispatched within at most one scan interval. Smaller = lower wake latency,
+    /// larger = fewer scan queries.
+    #[serde(with = "humantime_serde")]
+    pub active_scan_interval: Duration,
     /// Duration a cluster sleeps after a pass found jobs but dispatched zero
     /// frames (saturated farm: no host candidate fits any pending layer).
     /// Keeps the loop from re-querying jobs and layers continuously while
@@ -183,6 +191,7 @@ impl Default for QueueConfig {
             job_back_off_duration: Duration::from_secs(300),
             cluster_empty_sleep: Duration::from_secs(30),
             cluster_reload_interval: Duration::from_secs(120),
+            active_scan_interval: Duration::from_secs(2),
             cluster_saturated_sleep: Duration::from_secs(5),
             stream: StreamConfig::default(),
             max_jobs_per_cluster_pass: 20,

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

@@ -12,11 +12,11 @@
 
 use std::sync::Arc;
 
+use crate::pgpool::connection_pool;
 use futures::Stream;
 use miette::{IntoDiagnostic, Result};
 use serde::{Deserialize, Serialize};
 use sqlx::{Pool, Postgres};
-use crate::pgpool::connection_pool;
 
 /// Data Access Object for host operations in the job dispatch system.
 ///

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

@@ -44,6 +44,21 @@ pub struct JobModel {
     pub show_name: String,
 }
 
+/// One active (facility, show, tag) tuple from [`JobDao::scan_active_tags`].
+///
+/// A tag is "active" for a (facility, show) when at least one PENDING job in
+/// that facility/show has a waiting layer carrying the tag, and the show still
+/// has subscription headroom. Used to gate the cluster feed's awake set: a
+/// cluster is eligible iff one of its tags appears here under its
+/// (facility, show) key. See `cluster::ClusterFeed` and the superset argument
+/// on [`QUERY_ACTIVE_TAGS`].
+#[derive(sqlx::FromRow, Serialize, Deserialize)]
+pub struct ActiveTagModel {
+    pub pk_show: String,
+    pub pk_facility: String,
+    pub tag: String,
+}
+
 impl DispatchJob {
     /// Creates a new DispatchJob from a database model and cluster assignment.
     ///
@@ -156,6 +171,39 @@ ORDER BY jr.int_priority DESC
 LIMIT $5
 "#;
 
+// Awake-gate scan. Returns the set of (facility, show, tag) tuples that *could*
+// yield a dispatchable job, so the cluster feed can keep clusters with no work
+// asleep instead of re-querying them every backoff window.
+//
+// SUPERSET INVARIANT (must never produce a false negative): every (job, layer)
+// the per-cluster query QUERY_PENDING_BY_SHOW_FACILITY_TAG would return must be
+// captured here, otherwise the owning cluster is gated to permanent sleep and
+// its jobs starve. Two rules keep this true:
+//
+// `unnest` splits each layer's pipe-joined str_tags into individual tags; the
+// feed re-applies tag-set overlap on the cluster side, matching the per-cluster
+// `&&` array-overlap semantics.
+static QUERY_ACTIVE_TAGS: &str = r#"
+SELECT DISTINCT
+    j.pk_show,
+    j.pk_facility,
+    unnest(string_to_array(REPLACE(l.str_tags, ' ', ''), '|')) AS tag
+FROM job j
+INNER JOIN show sh ON sh.pk_show = j.pk_show
+INNER JOIN layer l ON l.pk_job = j.pk_job
+INNER JOIN layer_stat ls ON ls.pk_layer = l.pk_layer
+WHERE j.str_state = 'PENDING'
+    AND j.b_paused = false
+    AND sh.b_active = true
+    AND sh.b_scheduler_managed = true
+    AND ls.int_waiting_count > 0
+    AND EXISTS (
+        SELECT 1 FROM subscription s
+        WHERE s.pk_show = j.pk_show AND s.int_burst > 0
+    )
+    AND ($1::text IS NULL OR j.pk_facility = $1)
+"#;
+
 impl JobDao {
     /// Creates a new JobDao from database configuration.
     ///
@@ -230,4 +278,33 @@ impl JobDao {
         observe_job_query_duration(start.elapsed());
         result
     }
+
+    /// Scans for the set of (facility, show, tag) tuples that currently have
+    /// plausibly-dispatchable work, used to gate the cluster feed's awake set.
+    ///
+    /// This is a strict superset of what `query_pending_jobs_by_show_facility_and_tags`
+    /// would return (see [`QUERY_ACTIVE_TAGS`]): a cluster whose tag is absent
+    /// here is guaranteed to have no dispatchable job and can stay asleep. The
+    /// reverse does not hold — a returned tag may still yield a no_jobs or
+    /// saturated pass once the folder/job caps and live subscription headroom are
+    /// evaluated per-cluster, which is intentionally cheap to absorb.
+    ///
+    /// # Arguments
+    /// * `facility_id` - Optional facility filter; `None` scans every facility.
+    ///
+    /// # Returns
+    /// All active tuples (unordered). One database round-trip regardless of the
+    /// number of clusters.
+    pub async fn scan_active_tags(
+        &self,
+        facility_id: Option<&str>,
+    ) -> Result<Vec<ActiveTagModel>, sqlx::Error> {
+        let start = std::time::Instant::now();
+        let result = sqlx::query_as::<_, ActiveTagModel>(QUERY_ACTIVE_TAGS)
+            .bind(facility_id)
+            .fetch_all(&*self.connection_pool)
+            .await;
+        crate::metrics::observe_active_scan_duration(start.elapsed());
+        result
+    }
 }

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

@@ -22,7 +22,7 @@ mod resource_accounting_dao;
 pub use cluster_dao::ClusterDao;
 pub use frame_dao::FrameDao;
 pub use host_dao::HostDao;
-pub use job_dao::JobDao;
+pub use job_dao::{ActiveTagModel, JobDao};
 pub use layer_dao::LayerDao;
 pub use proc_dao::{ProcDao, ProcDaoError};
 

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

@@ -964,7 +964,10 @@ mod tests {
             .values()
             .flat_map(|by_memory| by_memory.values())
             .any(|hosts| hosts.contains(&dropped_id));
-        assert!(!still_referenced, "pruned host must not linger in any bucket");
+        assert!(
+            !still_referenced,
+            "pruned host must not linger in any bucket"
+        );
         drop(hosts_index);
 
         // A second prune with the same live set is a no-op.

rust/crates/scheduler/src/host_cache/cache.rs

@@ -10,8 +10,8 @@
 // or implied. See the License for the specific language governing permissions and limitations under
 // the License.
 
-pub(crate) mod cache;
 mod actor;
+pub(crate) mod cache;
 pub mod messages;
 mod store;