[rust-scheduler] Replace round-robin ClusterFeed with priority-queue scheduling to prevent busy-cluster starvation

[ramonfigueiredo]

Describe the enhancement

The Rust scheduler's ClusterFeed (rust/crates/scheduler/src/cluster.rs:374-447) iterates through clusters with a fixed-step round-robin: current_index = (current_index + 1) % clusters.len(). Every cluster
gets the same air-time regardless of how much work it actually has, and at the end of each lap the feed sleeps for a fixed 10 ms / 100 ms / 5 s depending on how many clusters are currently sleeping.

This produces two unwanted behaviors:

1. Throughput-blind fairness. A cluster with 10,000 pending frames gets the same turn as a cluster with 5. Busy clusters drain slowly while light ones consume scheduler cycles with no work to do.
2. Head-of-line blocking from back-off. When entrypoint.rs:93 puts a cluster to sleep for 3 s (no jobs found), the round-robin still walks past it on every lap and the inter-round sleeps stack up. Neighbors
   don't get extra turns to compensate.

for_each_concurrent(cluster_buffer_size=3) downstream limits parallel cluster processing but does nothing for fairness, it just controls how many of the round-robin's emissions are in flight at once.

Proposed redesign

Replace the Vec<Cluster> + AtomicUsize index with a priority queue keyed on (next_eligible_at, -last_dispatched_jobs):

struct ClusterFeed {
    queue: Arc<Mutex<BinaryHeap<Scheduled>>>,
    // ... existing fields
}

struct Scheduled {
    cluster: Cluster,
    next_eligible_at: Instant,    // sleep deadline; Instant::now() for ready
    last_dispatched_jobs: usize,  // bias toward productive clusters
}

// Ord: next_eligible_at asc, then last_dispatched_jobs desc

Feed loop:

1. Pop the min-next_eligible_at entry.
2. If next_eligible_at is in the future, tokio::time::sleep_until(next_eligible_at).
3. Emit the cluster to the consumer.
4. When the consumer reports back (success, sleep, or empty), push the cluster back with updated next_eligible_at and last_dispatched_jobs.

This naturally subsumes FeedMessage::Sleep (just bump next_eligible_at on re-insert), removes the lap-based sleep tiers entirely, and weights busy clusters higher without starving anyone. Once a cluster is
empty its last_dispatched_jobs drops and other clusters get prioritized.

Trade-offs

• One Mutex per pop. With <10k clusters and pops every few ms this is negligible; if it ever shows up in profiling, shard the heap by facility_id.
• Need a small signaling mechanism from the consumer back to the feed so last_dispatched_jobs and next_eligible_at get updated post-processing. Today FeedMessage::Sleep already provides one direction of this. Extend it to carry a processed_jobs count.
• Behavior change: light/idle clusters will be polled less aggressively. Worth a config knob for the productivity-bias weight so operators can tune toward fairness or throughput.

Acceptance criteria

• Busy clusters receive proportionally more turns than light clusters in a mixed workload (verifiable via the existing CLUSTER_ROUNDS counter, with a new cluster_polls_total{cluster} label).
• A sleeping cluster does not delay the next eligible cluster's turn.
• Smoke tests in tests/smoke_tests.rs continue to pass, with new coverage for prioritization under starvation scenarios.

Version Number

• https://github.com/AcademySoftwareFoundation/OpenCue/releases/tag/v1.19.1

Additional context

• Source: rust/crates/scheduler/src/cluster.rs:374-447 (round-robin loop) and rust/crates/scheduler/src/pipeline/entrypoint.rs:93 (3 s back-off on empty cluster).
• Related dead metric: CLUSTER_ROUNDS (cluster.rs:36) is incremented but never exposed via Prometheus. Wiring it up (along with a per-cluster last_dispatched_jobs gauge) would make the impact of this change observable.
• This change is local to the Rust scheduler crate and does not affect the Java Cuebot dispatcher.

[ramonfigueiredo]

@DiegoTavares FYI !