#![allow(dead_code)]
use crate::admission::AdmissionGate;
use crate::disk_pressure::PressureState;
use crate::metrics::{
self,
latency::{DecisionTimer, DecisionType},
};
use crate::ui::workers::{debug_routing_enabled, log_routing_decision};
use crate::workers::{WorkerPool, WorkerState};
use rand::RngExt;
use rch_common::{
CircuitBreakerConfig, CircuitState, CommandPriority, CompilationKind, RequiredRuntime,
SelectionConfig, SelectionDiagnostics, SelectionReason, SelectionRequest, SelectionStrategy,
SelectionWeightConfig, SshClient, SshOptions, ToolchainInfo, WorkerCapabilities, WorkerId,
WorkerSelectionDiagnostic, WorkerSelectionDiagnosticDecision, WorkerStatus, classify_command,
};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
const DEFAULT_NETWORK_SCORE: f64 = 0.5;
const NETWORK_LATENCY_HALF_LIFE_MS: f64 = 200.0;
const PRIORITY_BUCKET_SCORE: f64 = 10_000.0;
const PRIORITY_CACHE_TIEBREAK_SCORE: f64 = 1_000.0;
const PRIORITY_SPEED_TIEBREAK_SCORE: f64 = 10.0;
const TEST_CACHE_BOOST: f64 = 1.5;
const TEST_BUILD_FALLBACK_FACTOR: f64 = 0.4;
const TOOLCHAIN_PREFLIGHT_TTL: Duration = Duration::from_secs(600);
const TOOLCHAIN_PREFLIGHT_TRANSIENT_TTL: Duration = Duration::from_secs(60);
const TOOLCHAIN_PREFLIGHT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
const PRE_V3_MICROARCH_PENALTY: f64 = 0.5;
const TOOLCHAIN_PREFLIGHT_COMMAND_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Debug, Clone)]
pub struct SelectionWeights {
pub slots: f64,
pub speed: f64,
pub locality: f64,
pub priority: f64,
pub disk: f64,
pub half_open_penalty: f64,
}
impl Default for SelectionWeights {
fn default() -> Self {
Self {
slots: 0.4,
speed: 0.5,
locality: 0.1,
priority: 0.1,
disk: 0.2,
half_open_penalty: 0.5, }
}
}
impl From<&SelectionWeightConfig> for SelectionWeights {
fn from(config: &SelectionWeightConfig) -> Self {
Self {
slots: config.slots,
speed: config.speedscore,
locality: config.cache,
priority: config.priority,
disk: config.disk,
half_open_penalty: config.half_open_penalty,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CacheUse {
Build,
Test,
}
#[derive(Debug, Default, Clone)]
struct CacheState {
last_build: Option<Instant>,
last_test: Option<Instant>,
last_success: Option<Instant>,
}
impl CacheState {
fn last_activity(&self) -> Option<Instant> {
match (self.last_build, self.last_test) {
(Some(build), Some(test)) => Some(build.max(test)),
(Some(build), None) => Some(build),
(None, Some(test)) => Some(test),
(None, None) => None,
}
}
}
#[derive(Debug, Clone)]
pub struct LastSuccessEntry {
pub worker_id: String,
pub timestamp: Instant,
}
#[derive(Debug, Default)]
pub struct CacheTracker {
workers: HashMap<String, HashMap<String, CacheState>>,
max_projects_per_worker: usize,
last_success_by_project: HashMap<String, LastSuccessEntry>,
max_success_projects: usize,
}
impl CacheTracker {
pub fn new() -> Self {
Self {
workers: HashMap::new(),
max_projects_per_worker: 50,
last_success_by_project: HashMap::new(),
max_success_projects: 200,
}
}
pub fn record_build(&mut self, worker_id: &str, project_id: &str, cache_use: CacheUse) {
let cache = self.workers.entry(worker_id.to_string()).or_default();
let state = cache.entry(project_id.to_string()).or_default();
let now = Instant::now();
match cache_use {
CacheUse::Build => {
state.last_build = Some(now);
}
CacheUse::Test => {
state.last_test = Some(now);
state.last_build = Some(now);
}
}
while cache.len() > self.max_projects_per_worker {
let oldest = cache
.iter()
.min_by_key(|(_, state)| state.last_activity())
.map(|(k, _)| k.clone());
if let Some(key) = oldest {
cache.remove(&key);
}
}
}
pub fn estimate_warmth(&self, worker_id: &str, project_id: &str, cache_use: CacheUse) -> f64 {
self.workers
.get(worker_id)
.and_then(|c| c.get(project_id))
.map(|state| match cache_use {
CacheUse::Build => state
.last_activity()
.map(Self::warmth_from_instant)
.unwrap_or(0.0),
CacheUse::Test => {
let test_warmth = state
.last_test
.map(Self::warmth_from_instant)
.unwrap_or(0.0);
let build_warmth = state
.last_build
.map(Self::warmth_from_instant)
.unwrap_or(0.0)
* TEST_BUILD_FALLBACK_FACTOR;
test_warmth.max(build_warmth)
}
})
.unwrap_or(0.0)
}
pub fn has_recent_build(
&self,
worker_id: &str,
project_id: &str,
cache_use: CacheUse,
max_age: Duration,
) -> bool {
self.workers
.get(worker_id)
.and_then(|c| c.get(project_id))
.map(|state| match cache_use {
CacheUse::Build => state
.last_activity()
.map(|instant| instant.elapsed() < max_age)
.unwrap_or(false),
CacheUse::Test => {
if let Some(last_test) = state.last_test
&& last_test.elapsed() < max_age
{
return true;
}
state
.last_build
.map(|instant| instant.elapsed() < max_age)
.unwrap_or(false)
}
})
.unwrap_or(false)
}
fn warmth_from_instant(last_build: Instant) -> f64 {
let age = last_build.elapsed();
let hours = age.as_secs_f64() / 3600.0;
if hours < 1.0 {
1.0
} else if hours < 6.0 {
1.0 - (hours - 1.0) * 0.1 } else if hours < 24.0 {
0.5 - (hours - 6.0) * (0.3 / 18.0) } else {
0.2 * (-((hours - 24.0) / 24.0)).exp() }
}
pub fn record_success(&mut self, worker_id: &str, project_id: &str) {
let now = Instant::now();
{
let cache = self.workers.entry(worker_id.to_string()).or_default();
let state = cache.entry(project_id.to_string()).or_default();
state.last_build = Some(now);
state.last_success = Some(now);
while cache.len() > self.max_projects_per_worker {
let oldest = cache
.iter()
.min_by_key(|(_, state)| state.last_activity())
.map(|(key, _)| key.clone());
if let Some(key) = oldest {
cache.remove(&key);
}
}
}
self.last_success_by_project.insert(
project_id.to_string(),
LastSuccessEntry {
worker_id: worker_id.to_string(),
timestamp: now,
},
);
while self.last_success_by_project.len() > self.max_success_projects {
let oldest = self
.last_success_by_project
.iter()
.min_by_key(|(_, entry)| entry.timestamp)
.map(|(k, _)| k.clone());
if let Some(key) = oldest {
self.last_success_by_project.remove(&key);
}
}
}
pub fn get_pinned_worker(&self, project_id: &str, pin_window: Duration) -> Option<&str> {
self.last_success_by_project
.get(project_id)
.filter(|entry| entry.timestamp.elapsed() < pin_window)
.map(|entry| entry.worker_id.as_str())
}
pub fn get_last_success_worker(&self, project_id: &str) -> Option<&LastSuccessEntry> {
self.last_success_by_project.get(project_id)
}
pub fn has_recent_success(&self, worker_id: &str, project_id: &str, max_age: Duration) -> bool {
self.workers
.get(worker_id)
.and_then(|c| c.get(project_id))
.and_then(|state| state.last_success)
.map(|instant| instant.elapsed() < max_age)
.unwrap_or(false)
}
}
#[derive(Debug)]
pub struct SelectionHistory {
selections: HashMap<String, VecDeque<Instant>>,
max_history_per_worker: usize,
}
impl Default for SelectionHistory {
fn default() -> Self {
Self {
selections: HashMap::new(),
max_history_per_worker: 100,
}
}
}
impl SelectionHistory {
pub fn new() -> Self {
Self::default()
}
pub fn record_selection(&mut self, worker_id: &str) {
let history = self.selections.entry(worker_id.to_string()).or_default();
history.push_back(Instant::now());
while history.len() > self.max_history_per_worker {
history.pop_front();
}
}
pub fn recent_selections(&self, worker_id: &str, window: Duration) -> usize {
let now = Instant::now();
self.selections
.get(worker_id)
.map(|history| {
history
.iter()
.filter(|&&t| now.saturating_duration_since(t) < window)
.count()
})
.unwrap_or(0)
}
pub fn prune(&mut self, max_age: Duration) {
let now = Instant::now();
for history in self.selections.values_mut() {
while history
.front()
.map(|&t| now.saturating_duration_since(t) >= max_age)
.unwrap_or(false)
{
history.pop_front();
}
}
}
}
const DEFAULT_AUDIT_LOG_SIZE: usize = 100;
#[derive(Debug, Clone, serde::Serialize)]
pub struct WorkerScoreBreakdown {
pub worker_id: String,
pub total_score: f64,
pub speed_score: f64,
pub slot_availability: f64,
pub disk_headroom: f64,
pub cache_affinity: f64,
pub priority_score: f64,
pub circuit_state: String,
pub convergence_state: Option<String>,
pub reliability_state: Option<String>,
pub selected: bool,
pub skip_reason: Option<String>,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct SelectionAuditEntry {
pub id: u64,
pub timestamp_ms: u64,
pub project: String,
pub command: Option<String>,
pub strategy: String,
pub command_priority: String,
pub required_runtime: Option<String>,
pub eligible_count: usize,
pub workers_evaluated: Vec<WorkerScoreBreakdown>,
pub selected_worker_id: Option<String>,
pub reason: String,
pub classification_duration_us: Option<u64>,
pub selection_duration_us: u64,
}
#[derive(Debug)]
pub struct SelectionAuditLog {
entries: VecDeque<SelectionAuditEntry>,
max_entries: usize,
next_id: u64,
}
impl Default for SelectionAuditLog {
fn default() -> Self {
Self::new(DEFAULT_AUDIT_LOG_SIZE)
}
}
impl SelectionAuditLog {
pub fn new(max_entries: usize) -> Self {
Self {
entries: VecDeque::with_capacity(max_entries),
max_entries,
next_id: 1,
}
}
pub fn push(&mut self, mut entry: SelectionAuditEntry) {
entry.id = self.next_id;
self.next_id += 1;
if self.entries.len() >= self.max_entries {
self.entries.pop_front();
}
self.entries.push_back(entry);
}
pub fn entries(&self) -> &VecDeque<SelectionAuditEntry> {
&self.entries
}
pub fn last_n(&self, n: usize) -> Vec<&SelectionAuditEntry> {
self.entries.iter().rev().take(n).collect()
}
pub fn last(&self) -> Option<&SelectionAuditEntry> {
self.entries.back()
}
pub fn get(&self, id: u64) -> Option<&SelectionAuditEntry> {
self.entries.iter().find(|e| e.id == id)
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn clear(&mut self) {
self.entries.clear();
}
}
#[derive(Clone)]
pub struct WorkerSelector {
pub config: SelectionConfig,
pub circuit_config: CircuitBreakerConfig,
pub cache_tracker: Arc<RwLock<CacheTracker>>,
pub selection_history: Arc<RwLock<SelectionHistory>>,
pub audit_log: Arc<RwLock<SelectionAuditLog>>,
pub admission_gate: Option<Arc<AdmissionGate>>,
pub repo_convergence: Option<Arc<crate::repo_convergence::RepoConvergenceService>>,
pub reliability: Option<Arc<crate::reliability::ReliabilityAggregator>>,
pub ssh_pool: Option<Arc<rch_common::SshPool>>,
preview: bool,
}
pub struct SelectionResult {
pub worker: Option<Arc<WorkerState>>,
pub reason: SelectionReason,
pub diagnostics: Option<SelectionDiagnostics>,
}
fn cache_use_for_request(request: &SelectionRequest) -> CacheUse {
request
.command
.as_deref()
.map_or(CacheUse::Build, cache_use_for_command)
}
fn cache_use_for_command(command: &str) -> CacheUse {
let classification = classify_command(command);
if classification
.kind
.map(|kind| kind.is_test_command())
.unwrap_or(false)
{
CacheUse::Test
} else {
CacheUse::Build
}
}
impl WorkerSelector {
pub fn new() -> Self {
Self {
config: SelectionConfig::default(),
circuit_config: CircuitBreakerConfig::default(),
cache_tracker: Arc::new(RwLock::new(CacheTracker::new())),
selection_history: Arc::new(RwLock::new(SelectionHistory::new())),
audit_log: Arc::new(RwLock::new(SelectionAuditLog::default())),
admission_gate: None,
repo_convergence: None,
reliability: None,
ssh_pool: None,
preview: false,
}
}
pub fn with_config(config: SelectionConfig, circuit_config: CircuitBreakerConfig) -> Self {
Self {
config,
circuit_config,
cache_tracker: Arc::new(RwLock::new(CacheTracker::new())),
selection_history: Arc::new(RwLock::new(SelectionHistory::new())),
audit_log: Arc::new(RwLock::new(SelectionAuditLog::default())),
admission_gate: None,
repo_convergence: None,
reliability: None,
ssh_pool: None,
preview: false,
}
}
pub fn set_admission_gate(&mut self, gate: Arc<AdmissionGate>) {
self.admission_gate = Some(gate);
}
pub fn set_repo_convergence(
&mut self,
svc: Arc<crate::repo_convergence::RepoConvergenceService>,
) {
self.repo_convergence = Some(svc);
}
pub fn set_reliability(&mut self, agg: Arc<crate::reliability::ReliabilityAggregator>) {
self.reliability = Some(agg);
}
pub fn set_ssh_pool(&mut self, pool: Option<Arc<rch_common::SshPool>>) {
self.ssh_pool = pool;
}
pub async fn get_audit_log(&self, limit: Option<usize>) -> Vec<SelectionAuditEntry> {
let log = self.audit_log.read().await;
match limit {
Some(n) => log.last_n(n).into_iter().cloned().collect(),
None => log.entries().iter().cloned().collect(),
}
}
pub async fn get_last_audit_entry(&self) -> Option<SelectionAuditEntry> {
let log = self.audit_log.read().await;
log.last().cloned()
}
pub async fn select(&self, pool: &WorkerPool, request: &SelectionRequest) -> SelectionResult {
self.select_with_exclusions(pool, request, &HashSet::new())
.await
}
pub async fn select_with_exclusions(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> SelectionResult {
let mut round = self.clone();
round.admission_gate = self
.admission_gate
.as_ref()
.map(|gate| Arc::new(gate.for_selection()));
round
.select_in_round(pool, request, excluded_worker_ids)
.await
}
pub async fn preview_with_exclusions(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> SelectionResult {
let mut round = self.clone();
round.preview = true;
round.admission_gate = self
.admission_gate
.as_ref()
.map(|gate| Arc::new(gate.for_selection()));
round
.select_in_round(pool, request, excluded_worker_ids)
.await
}
async fn select_in_round(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> SelectionResult {
let _timer = DecisionTimer::new(DecisionType::WorkerSelection);
let select_start = Instant::now();
let correlation_id = request
.hook_pid
.map(|pid| format!("hook-{pid}"))
.unwrap_or_else(|| "hook-unknown".to_string());
debug!(
correlation_id = %correlation_id,
project = %request.project,
strategy = ?self.config.strategy,
command_priority = ?request.command_priority,
"Starting worker selection"
);
let cache_use = cache_use_for_request(request);
let eligible = match self
.get_eligible_workers(pool, request, excluded_worker_ids)
.await
{
Ok(workers) => workers,
Err(reason) => {
let diagnostics = self
.build_selection_diagnostics(pool, request, excluded_worker_ids)
.await;
self.record_audit_entry(request, Vec::new(), None, &reason, select_start.elapsed())
.await;
record_selection_metrics(&reason, select_start.elapsed());
return SelectionResult {
worker: None,
reason,
diagnostics: Some(diagnostics),
};
}
};
if eligible.is_empty() {
if !request.job_mode
&& request.preferred_workers.is_empty()
&& let Some(fallback_worker_id) =
self.try_fallback(pool, request, excluded_worker_ids).await
{
self.record_audit_entry(
request,
Vec::new(),
Some(fallback_worker_id.clone()),
&SelectionReason::AffinityFallback,
select_start.elapsed(),
)
.await;
let worker_id = WorkerId::new(&fallback_worker_id);
if let Some(worker) = pool.get(&worker_id).await {
debug!(
"Using affinity fallback worker {} for project {}",
fallback_worker_id, request.project
);
record_selection_metrics(
&SelectionReason::AffinityFallback,
select_start.elapsed(),
);
return SelectionResult {
worker: Some(worker),
reason: SelectionReason::AffinityFallback,
diagnostics: None,
};
}
}
let diagnostics = self
.build_selection_diagnostics(pool, request, excluded_worker_ids)
.await;
self.record_audit_entry(
request,
Vec::new(),
None,
&SelectionReason::AllWorkersBusy,
select_start.elapsed(),
)
.await;
record_selection_metrics(&SelectionReason::AllWorkersBusy, select_start.elapsed());
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersBusy,
diagnostics: Some(diagnostics),
};
}
let pinned_selection = self
.try_pinned_worker(&eligible, request, excluded_worker_ids)
.await;
if let Some((worker, circuit_state)) = pinned_selection {
let worker_id = worker.config.read().await.id.as_str().to_string();
debug!(
"Using affinity-pinned worker {} for project {}",
worker_id, request.project
);
if self.preview {
return SelectionResult {
worker: Some(worker),
reason: SelectionReason::AffinityPinned,
diagnostics: None,
};
}
let mut history = self.selection_history.write().await;
history.record_selection(&worker_id);
let breakdowns = self
.build_score_breakdowns(&eligible, request, cache_use, Some(&worker_id))
.await;
self.record_audit_entry(
request,
breakdowns,
Some(worker_id.clone()),
&SelectionReason::AffinityPinned,
select_start.elapsed(),
)
.await;
record_selection_metrics(&SelectionReason::AffinityPinned, select_start.elapsed());
if circuit_state == CircuitState::HalfOpen {
worker.start_probe(&self.circuit_config).await;
}
return SelectionResult {
worker: Some(worker),
reason: SelectionReason::AffinityPinned,
diagnostics: None,
};
}
let selected = match self.config.strategy {
SelectionStrategy::Priority => {
self.select_by_priority(&eligible, request, cache_use).await
}
SelectionStrategy::Fastest => {
if matches!(request.command_priority, CommandPriority::Low) {
self.select_fair_fastest(&eligible).await
} else {
self.select_by_fastest(&eligible).await
}
}
SelectionStrategy::Balanced => {
self.select_balanced(&eligible, request, cache_use).await
}
SelectionStrategy::CacheAffinity => match request.command_priority {
CommandPriority::High => self.select_by_fastest(&eligible).await,
CommandPriority::Normal | CommandPriority::Low => {
self.select_cache_affinity(&eligible, request, cache_use)
.await
}
},
SelectionStrategy::FairFastest => {
if matches!(request.command_priority, CommandPriority::High) {
self.select_by_fastest(&eligible).await
} else {
self.select_fair_fastest(&eligible).await
}
}
};
if let Some((worker, circuit_state)) = selected {
let worker_id = worker.config.read().await.id.as_str().to_string();
if debug_routing_enabled() {
let scores = self.debug_scores(&eligible, request, cache_use).await;
let circuit_label = format!("{:?}", circuit_state);
log_routing_decision(
self.config.strategy,
&worker_id,
Some(circuit_label.as_str()),
&scores,
eligible.len(),
);
}
if self.preview {
return SelectionResult {
worker: Some(worker),
reason: SelectionReason::Success,
diagnostics: None,
};
}
let breakdowns = self
.build_score_breakdowns(&eligible, request, cache_use, Some(&worker_id))
.await;
self.record_audit_entry(
request,
breakdowns,
Some(worker_id.clone()),
&SelectionReason::Success,
select_start.elapsed(),
)
.await;
if circuit_state == CircuitState::HalfOpen {
worker.start_probe(&self.circuit_config).await;
debug!(
"Worker {} selected (half-open probe), strategy={:?}",
worker_id, self.config.strategy
);
} else {
debug!(
"Worker {} selected, strategy={:?}",
worker_id, self.config.strategy
);
}
let mut history = self.selection_history.write().await;
history.record_selection(&worker_id);
record_selection_metrics(&SelectionReason::Success, select_start.elapsed());
SelectionResult {
worker: Some(worker),
reason: SelectionReason::Success,
diagnostics: None,
}
} else {
let diagnostics = self
.build_selection_diagnostics(pool, request, excluded_worker_ids)
.await;
let breakdowns = self
.build_score_breakdowns(&eligible, request, cache_use, None)
.await;
self.record_audit_entry(
request,
breakdowns,
None,
&SelectionReason::AllWorkersBusy,
select_start.elapsed(),
)
.await;
record_selection_metrics(&SelectionReason::AllWorkersBusy, select_start.elapsed());
SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersBusy,
diagnostics: Some(diagnostics),
}
}
}
pub async fn record_build(&self, worker_id: &str, project_id: &str, is_test: bool) {
let mut cache = self.cache_tracker.write().await;
let cache_use = if is_test {
CacheUse::Test
} else {
CacheUse::Build
};
cache.record_build(worker_id, project_id, cache_use);
}
pub async fn record_remote_completion(
&self,
worker_id: &str,
project_id: &str,
command: &str,
exit_code: i32,
remote_command_started: bool,
) {
if exit_code == 0 {
self.record_success(worker_id, project_id).await;
} else if remote_command_started {
let cache_use = cache_use_for_command(command);
self.cache_tracker
.write()
.await
.record_build(worker_id, project_id, cache_use);
}
}
#[cfg(test)]
pub(crate) async fn cache_warmth(
&self,
worker_id: &str,
project_id: &str,
cache_use: CacheUse,
) -> f64 {
self.cache_tracker
.read()
.await
.estimate_warmth(worker_id, project_id, cache_use)
}
pub async fn record_success(&self, worker_id: &str, project_id: &str) {
let mut cache = self.cache_tracker.write().await;
cache.record_success(worker_id, project_id);
}
pub async fn get_pinned_worker(&self, project_id: &str) -> Option<String> {
if !self.config.affinity.enabled {
return None;
}
let pin_window = Duration::from_secs(self.config.affinity.pin_minutes * 60);
let cache = self.cache_tracker.read().await;
cache
.get_pinned_worker(project_id, pin_window)
.map(String::from)
}
pub async fn get_fallback_worker(&self, project_id: &str) -> Option<String> {
if !self.config.affinity.enable_last_success_fallback {
return None;
}
let cache = self.cache_tracker.read().await;
cache
.get_last_success_worker(project_id)
.map(|entry| entry.worker_id.clone())
}
async fn try_pinned_worker(
&self,
eligible: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> Option<(Arc<WorkerState>, CircuitState)> {
if !self.config.affinity.enabled {
return None;
}
let pinned_worker_id = self.get_pinned_worker(&request.project).await?;
if excluded_worker_ids.contains(&pinned_worker_id) {
return None;
}
let mut pinned: Option<(Arc<WorkerState>, CircuitState, u32)> = None;
for (worker, circuit_state) in eligible {
let config = worker.config.read().await;
if config.id.as_str() == pinned_worker_id {
pinned = Some((worker.clone(), *circuit_state, config.priority));
break;
}
}
let (worker, circuit_state, pinned_priority) = pinned?;
for (candidate, _) in eligible {
let available_slots = candidate.available_slots().await;
let config = candidate.config.read().await;
if config.priority > pinned_priority && available_slots >= request.estimated_cores {
debug!(
"Skipping affinity pin {} for project {}: higher-priority worker {} (priority {}) has capacity",
pinned_worker_id, request.project, config.id, config.priority
);
return None;
}
}
Some((worker, circuit_state))
}
async fn try_fallback(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> Option<String> {
if !self.config.affinity.enable_last_success_fallback {
return None;
}
let fallback_id = self.get_fallback_worker(&request.project).await?;
if excluded_worker_ids.contains(&fallback_id) {
return None;
}
let worker_id = WorkerId::new(&fallback_id);
let worker = pool.get(&worker_id).await?;
let declared_os = rch_common::declared_os(&worker.config.read().await.tags);
if !os_gate_admits(
declared_os.as_deref(),
required_os_for_request(request).as_deref(),
) {
debug!(
"Affinity fallback worker {} skipped: OS gate (declared={:?})",
fallback_id, declared_os
);
return None;
}
let status = worker.status().await;
if !matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded) {
debug!(
"Affinity fallback worker {} skipped: status is {:?}",
fallback_id, status
);
return None;
}
let available = worker.available_slots().await;
if available < request.estimated_cores {
debug!(
"Affinity fallback worker {} skipped: available slots {} < requested {}",
fallback_id, available, request.estimated_cores
);
return None;
}
if let Some(circuit_state) = worker.circuit_state().await
&& circuit_state == CircuitState::Open
{
return None;
}
let success_rate = self.health_score(&worker).await;
if success_rate < self.config.affinity.fallback_min_success_rate {
debug!(
"Affinity fallback worker {} skipped: success_rate {:.2} < fallback_min {:.2}",
fallback_id, success_rate, self.config.affinity.fallback_min_success_rate
);
return None;
}
if let Some(gate) = &self.admission_gate {
if let crate::admission::AdmissionVerdict::Reject { reason_code, .. } =
gate.evaluate(&worker, &fallback_id, &request.project).await
&& matches!(
reason_code.as_str(),
"admission_critical_pressure" | "admission_disk_floor"
)
{
return None;
}
} else if worker.pressure_assessment().await.state == PressureState::Critical {
return None;
}
Some(fallback_id)
}
async fn build_score_breakdowns(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
cache_use: CacheUse,
selected_id: Option<&str>,
) -> Vec<WorkerScoreBreakdown> {
let cache = self.cache_tracker.read().await;
let (min_priority, max_priority) = Self::priority_range(workers).await;
let mut breakdowns = Vec::with_capacity(workers.len());
let weights = adjust_weights_for_priority(&self.config.weights, request);
for (worker, circuit_state) in workers {
let total_slots = worker.effective_total_slots().await;
let config = worker.config.read().await;
let worker_id = config.id.as_str().to_string();
let speed_score = worker.get_speed_score();
let used_slots = worker.used_slots();
let slot_availability = if total_slots > 0 {
total_slots.saturating_sub(used_slots) as f64 / total_slots as f64
} else {
0.0
};
let cache_affinity = cache.estimate_warmth(&worker_id, &request.project, cache_use);
let priority_score =
Self::normalize_priority(config.priority, min_priority, max_priority);
drop(config);
let disk_headroom = worker.disk_headroom().await;
let total_score = if self.config.strategy == SelectionStrategy::Balanced {
self.compute_balanced_score(
worker,
*circuit_state,
&request.project,
&cache,
cache_use,
&weights,
min_priority,
max_priority,
)
.await
} else {
speed_score * slot_availability
};
let skip_reason = if *circuit_state == CircuitState::Open {
Some("circuit open".to_string())
} else if used_slots >= total_slots {
Some("no slots available".to_string())
} else {
None
};
let selected = selected_id.is_some_and(|id| id.eq(worker_id.as_str()));
let convergence_state = if let Some(ref convergence_svc) = self.repo_convergence {
let wid = rch_common::WorkerId::new(worker_id.as_str());
Some(convergence_svc.get_drift_state(&wid).await.to_string())
} else {
None
};
let reliability_state = if let Some(ref agg) = self.reliability {
agg.get_assessment(worker_id.as_str())
.await
.map(|a| a.health_state.to_string())
} else {
None
};
breakdowns.push(WorkerScoreBreakdown {
worker_id,
total_score,
speed_score,
slot_availability,
disk_headroom,
cache_affinity,
priority_score,
circuit_state: format!("{:?}", circuit_state),
convergence_state,
reliability_state,
selected,
skip_reason,
});
}
breakdowns
}
async fn record_audit_entry(
&self,
request: &SelectionRequest,
workers_evaluated: Vec<WorkerScoreBreakdown>,
selected_worker_id: Option<String>,
reason: &SelectionReason,
duration: Duration,
) {
use std::time::{SystemTime, UNIX_EPOCH};
let timestamp_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let mut audit_reason = format!("{reason:?}");
if let Some(gate) = &self.admission_gate {
for (worker_id, reason_code) in gate.rejection_codes().await {
audit_reason.push_str(&format!("; {worker_id}={reason_code}"));
}
}
let entry = SelectionAuditEntry {
id: 0, timestamp_ms,
project: request.project.clone(),
command: request.command.clone(),
strategy: format!("{:?}", self.config.strategy),
command_priority: format!("{:?}", request.command_priority),
required_runtime: match request.required_runtime {
RequiredRuntime::None => None,
ref rt => Some(format!("{:?}", rt)),
},
eligible_count: workers_evaluated.len(),
workers_evaluated,
selected_worker_id,
reason: audit_reason,
classification_duration_us: request.classification_duration_us,
selection_duration_us: duration.as_micros() as u64,
};
let mut log = self.audit_log.write().await;
log.push(entry);
}
async fn debug_scores(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
cache_use: CacheUse,
) -> Vec<(String, f64)> {
let mut scores = Vec::new();
match self.config.strategy {
SelectionStrategy::Priority => {
let cache = self.cache_tracker.read().await;
for (worker, _) in workers {
let config = worker.config.read().await;
let warmth =
cache.estimate_warmth(config.id.as_str(), &request.project, cache_use);
let score = Self::priority_selection_score(
config.priority,
warmth,
worker.get_speed_score(),
request.command_priority,
);
scores.push((config.id.as_str().to_string(), score));
}
}
SelectionStrategy::Fastest | SelectionStrategy::FairFastest => {
for (worker, _) in workers {
let score = worker.get_speed_score();
let id = worker.config.read().await.id.as_str().to_string();
scores.push((id, score));
}
}
SelectionStrategy::Balanced => {
let cache = self.cache_tracker.read().await;
let weights = adjust_weights_for_priority(&self.config.weights, request);
let (min_priority, max_priority) = Self::priority_range(workers).await;
for (worker, circuit_state) in workers {
let score = self
.compute_balanced_score(
worker,
*circuit_state,
&request.project,
&cache,
cache_use,
&weights,
min_priority,
max_priority,
)
.await;
let id = worker.config.read().await.id.as_str().to_string();
scores.push((id, score));
}
}
SelectionStrategy::CacheAffinity => {
let cache = self.cache_tracker.read().await;
for (worker, _) in workers {
let id = worker.config.read().await.id.as_str().to_string();
let score = cache.estimate_warmth(id.as_str(), &request.project, cache_use);
scores.push((id, score));
}
}
}
scores
}
async fn build_selection_diagnostics(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> SelectionDiagnostics {
let all_workers = pool.all_workers().await;
let mut diagnostics = Vec::with_capacity(all_workers.len());
let mut active_project_exclusion_count = 0usize;
let required_os = required_os_for_request(request);
for worker in all_workers {
let config = worker.config.read().await;
let worker_id = config.id.clone();
let declared_os = rch_common::declared_os(&config.tags);
drop(config);
let total_slots = worker.effective_total_slots().await;
let status = worker.status().await;
let circuit_state = worker.circuit_state().await.unwrap_or(CircuitState::Closed);
let available_slots = worker.available_slots().await;
let capabilities = worker.capabilities().await;
let pressure = worker.pressure_assessment().await;
let success_rate = self.health_score(&worker).await;
let active_project_excluded = excluded_worker_ids.contains(worker_id.as_str());
if active_project_excluded {
active_project_exclusion_count += 1;
}
let runtime_available = match request.required_runtime {
RequiredRuntime::None => true,
RequiredRuntime::Rust => capabilities.has_rust(),
RequiredRuntime::Bun => capabilities.has_bun(),
RequiredRuntime::Node => capabilities.has_node(),
RequiredRuntime::Nix => capabilities.has_nix(),
RequiredRuntime::Go => capabilities.has_go(),
RequiredRuntime::Zig => capabilities.has_zig(),
};
let toolchain_mismatch =
toolchain_capability_mismatch(request.toolchain.as_ref(), &capabilities);
let component_mismatch = rustup_component_capability_mismatch(request, &capabilities);
let tool_mismatch = required_tool_capability_mismatch(request, &capabilities);
let cached_toolchain_failure = if let Some(toolchain) = request.toolchain.as_ref() {
let toolchain_name = toolchain.rustup_toolchain();
worker
.toolchain_preflight_status(&toolchain_name)
.await
.filter(|status| {
status
.is_reusable(TOOLCHAIN_PREFLIGHT_TTL, TOOLCHAIN_PREFLIGHT_TRANSIENT_TTL)
})
.and_then(|status| {
(!status.usable).then(|| {
status
.reason
.unwrap_or_else(|| "cached_toolchain_unusable".to_string())
})
})
} else {
None
};
let mut reason_codes = Vec::new();
let mut soft_reason: Option<String> = None;
let hard_admission = if let Some(ref gate) = self.admission_gate {
match gate.cached_verdict(worker_id.as_str()).await {
Some(crate::admission::AdmissionVerdict::Reject {
reason_code,
reason,
}) if matches!(
reason_code.as_str(),
"admission_critical_pressure" | "admission_disk_floor"
) =>
{
Some((reason_code, reason))
}
_ => None,
}
} else {
None
};
let (final_decision, final_reason) =
if !matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded) {
push_reason_code(&mut reason_codes, "worker.status_not_assignable");
(
WorkerSelectionDiagnosticDecision::Deny,
format!("worker status is {status:?}"),
)
} else if circuit_state == CircuitState::Open {
push_reason_code(&mut reason_codes, "worker.circuit_open");
(
WorkerSelectionDiagnosticDecision::Deny,
"circuit is open".to_string(),
)
} else if circuit_state == CircuitState::HalfOpen
&& !worker.can_probe(&self.circuit_config).await
{
push_reason_code(&mut reason_codes, "worker.half_open_no_probe_budget");
(
WorkerSelectionDiagnosticDecision::Deny,
"half-open circuit has no probe budget".to_string(),
)
} else if active_project_excluded {
push_reason_code(&mut reason_codes, "active_project_exclusion");
(
WorkerSelectionDiagnosticDecision::Deny,
"worker already has an active build for this project".to_string(),
)
} else if !os_gate_admits(declared_os.as_deref(), required_os.as_deref()) {
push_reason_code(&mut reason_codes, "os.declared_mismatch");
(
WorkerSelectionDiagnosticDecision::Deny,
match (declared_os.as_deref(), required_os.as_deref()) {
(Some(declared), Some(required)) => format!(
"worker declares os={declared}, command requires os={required}"
),
(Some(declared), None) => format!(
"worker declares os={declared}, so it takes only commands \
targeting that OS"
),
(None, Some(required)) => {
format!("command requires os={required}, worker declares no os")
}
(None, None) => "OS gate denied with no OS on either side".to_string(),
},
)
} else if !runtime_available {
push_reason_code(&mut reason_codes, "runtime.unavailable");
(
WorkerSelectionDiagnosticDecision::Deny,
format!(
"required runtime {:?} is unavailable",
request.required_runtime
),
)
} else if let Some(reason) = tool_mismatch {
push_reason_code(&mut reason_codes, "tool.required_missing");
(WorkerSelectionDiagnosticDecision::Deny, reason)
} else if let Some(reason) = component_mismatch {
push_reason_code(&mut reason_codes, "toolchain.component_missing");
(WorkerSelectionDiagnosticDecision::Deny, reason)
} else if let Some(reason) = toolchain_mismatch {
push_reason_code(&mut reason_codes, "toolchain.version_mismatch");
(WorkerSelectionDiagnosticDecision::Deny, reason)
} else if let Some(reason) = cached_toolchain_failure {
push_reason_code(&mut reason_codes, "toolchain.preflight_failed");
(WorkerSelectionDiagnosticDecision::Deny, reason)
} else if let Some((code, reason)) = hard_admission {
push_reason_code(
&mut reason_codes,
if code == "admission_disk_floor" {
"pressure.disk_floor"
} else {
"pressure.critical"
},
);
(
WorkerSelectionDiagnosticDecision::Deny,
format!("admission rejected: {reason}"),
)
} else if pressure.state == PressureState::Critical {
push_reason_code(&mut reason_codes, "pressure.critical");
(
WorkerSelectionDiagnosticDecision::Deny,
format!("critical pressure: {}", pressure.reason_code),
)
} else if total_slots < request.estimated_cores {
push_reason_code(&mut reason_codes, "slots.request_exceeds_capacity");
(
WorkerSelectionDiagnosticDecision::Deny,
format!(
"requested cores {} exceed total slots {total_slots}",
request.estimated_cores
),
)
} else if available_slots < request.estimated_cores {
push_reason_code(&mut reason_codes, "slots.insufficient");
(
WorkerSelectionDiagnosticDecision::Deny,
format!(
"available slots {available_slots} < {}",
request.estimated_cores
),
)
} else if let Some(false) = capabilities.is_topology_healthy() {
push_reason_code(&mut reason_codes, "topology.preflight_failed");
(
WorkerSelectionDiagnosticDecision::Deny,
capabilities
.projects_root_issue
.clone()
.unwrap_or_else(|| "topology preflight failed".to_string()),
)
} else {
let mut hard_decision = None;
if let Some(ref convergence_svc) = self.repo_convergence {
let drift_state = convergence_svc.get_drift_state(&worker_id).await;
match drift_state {
crate::repo_convergence::ConvergenceDriftState::Ready => {}
crate::repo_convergence::ConvergenceDriftState::Drifting => {
push_reason_code(&mut reason_codes, "convergence.drifting");
}
crate::repo_convergence::ConvergenceDriftState::Stale => {
push_reason_code(&mut reason_codes, "convergence.stale");
}
crate::repo_convergence::ConvergenceDriftState::Converging => {
push_reason_code(&mut reason_codes, "convergence.in_progress");
hard_decision = Some((
WorkerSelectionDiagnosticDecision::Deny,
"repo convergence is in progress".to_string(),
));
}
crate::repo_convergence::ConvergenceDriftState::Failed => {
push_reason_code(&mut reason_codes, "convergence.failed");
hard_decision = Some((
WorkerSelectionDiagnosticDecision::Deny,
"repo convergence failed".to_string(),
));
}
}
}
if hard_decision.is_none()
&& let Some(max_load) = self.config.max_load_per_core
&& let Some(true) = capabilities.is_high_load(max_load)
{
let load_per_core = capabilities.load_per_core().unwrap_or(0.0);
push_reason_code(&mut reason_codes, "preflight.high_load");
soft_reason = Some(format!(
"high load {:.2} > {:.2} per core",
load_per_core, max_load
));
}
if let Some(min_disk) = self.config.min_free_gb
&& let Some(true) = capabilities.is_low_disk(min_disk)
{
let free_gb = capabilities.build_disk_gb().0.unwrap_or(0.0);
push_reason_code(&mut reason_codes, "preflight.low_disk");
soft_reason =
Some(format!("low disk {:.1} GB < {:.1} GB", free_gb, min_disk));
}
let admission_decision: Option<(
WorkerSelectionDiagnosticDecision,
String,
&'static str,
)> = if let Some(ref gate) = self.admission_gate {
use crate::admission::AdmissionVerdict;
match gate.cached_verdict(worker_id.as_str()).await {
Some(AdmissionVerdict::Reject {
reason_code,
reason,
}) => {
if matches!(
reason_code.as_str(),
"admission_critical_pressure" | "admission_disk_floor"
) {
Some((
WorkerSelectionDiagnosticDecision::Deny,
format!("disk admission rejected: {reason}"),
if reason_code == "admission_disk_floor" {
"pressure.disk_floor"
} else {
"pressure.critical"
},
))
} else {
Some((
WorkerSelectionDiagnosticDecision::FallbackCandidate,
format!("admission rejected: {reason}"),
"admission.rejected",
))
}
}
Some(AdmissionVerdict::Admit { .. }) => None,
None => {
if pressure.state == PressureState::Critical {
Some((
WorkerSelectionDiagnosticDecision::Deny,
format!("critical pressure: {}", pressure.reason_code),
"pressure.critical",
))
} else {
None
}
}
}
} else if pressure.state == PressureState::Critical {
Some((
WorkerSelectionDiagnosticDecision::Deny,
format!("critical pressure: {}", pressure.reason_code),
"pressure.critical",
))
} else {
None
};
if let Some(decision) = hard_decision {
decision
} else if let Some((decision, reason, code)) = admission_decision {
push_reason_code(&mut reason_codes, code);
(decision, reason)
} else if let Some(ref agg) = self.reliability {
let assessment = match agg.get_assessment(worker_id.as_str()).await {
Some(assessment) => assessment,
None => agg.evaluate(worker.as_ref(), worker_id.as_str()).await,
};
if assessment.hard_exclude {
push_reason_code(&mut reason_codes, "reliability.quarantined");
(
WorkerSelectionDiagnosticDecision::Deny,
format!("reliability quarantined: {}", assessment.health_state),
)
} else if success_rate < self.config.min_success_rate {
push_reason_code(&mut reason_codes, "health.below_min_success_rate");
if success_rate >= self.config.affinity.fallback_min_success_rate {
(
WorkerSelectionDiagnosticDecision::FallbackCandidate,
format!(
"success_rate {:.2} < min {:.2} but >= fallback {:.2}",
success_rate,
self.config.min_success_rate,
self.config.affinity.fallback_min_success_rate
),
)
} else {
push_reason_code(
&mut reason_codes,
"health.below_fallback_min_success_rate",
);
(
WorkerSelectionDiagnosticDecision::Deny,
format!(
"success_rate {:.2} < fallback {:.2}",
success_rate,
self.config.affinity.fallback_min_success_rate
),
)
}
} else if let Some(reason) = soft_reason {
(WorkerSelectionDiagnosticDecision::FallbackCandidate, reason)
} else {
(
WorkerSelectionDiagnosticDecision::Allow,
"eligible".to_string(),
)
}
} else if success_rate < self.config.min_success_rate {
push_reason_code(&mut reason_codes, "health.below_min_success_rate");
if success_rate >= self.config.affinity.fallback_min_success_rate {
(
WorkerSelectionDiagnosticDecision::FallbackCandidate,
format!(
"success_rate {:.2} < min {:.2} but >= fallback {:.2}",
success_rate,
self.config.min_success_rate,
self.config.affinity.fallback_min_success_rate
),
)
} else {
push_reason_code(
&mut reason_codes,
"health.below_fallback_min_success_rate",
);
(
WorkerSelectionDiagnosticDecision::Deny,
format!(
"success_rate {:.2} < fallback {:.2}",
success_rate, self.config.affinity.fallback_min_success_rate
),
)
}
} else if let Some(reason) = soft_reason {
(WorkerSelectionDiagnosticDecision::FallbackCandidate, reason)
} else {
(
WorkerSelectionDiagnosticDecision::Allow,
"eligible".to_string(),
)
}
};
diagnostics.push(WorkerSelectionDiagnostic {
worker_id,
status: worker_status_label(status).to_string(),
circuit_state: circuit_state_label(circuit_state).to_string(),
pressure_state: pressure.state.to_string(),
pressure_reason_code: pressure.reason_code,
success_rate: Some(success_rate),
min_success_rate: self.config.min_success_rate,
fallback_min_success_rate: self.config.affinity.fallback_min_success_rate,
required_runtime: request.required_runtime,
runtime_available,
available_slots,
total_slots,
estimated_cores: request.estimated_cores,
active_project_excluded,
final_decision,
final_reason,
reason_codes,
});
}
SelectionDiagnostics {
required_runtime: request.required_runtime,
estimated_cores: request.estimated_cores,
min_success_rate: self.config.min_success_rate,
fallback_min_success_rate: self.config.affinity.fallback_min_success_rate,
active_project_exclusion_count,
workers: diagnostics,
}
}
async fn get_eligible_workers(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
) -> Result<Vec<(Arc<WorkerState>, CircuitState)>, SelectionReason> {
self.get_eligible_workers_with_refresh(pool, request, excluded_worker_ids, true)
.await
}
async fn refresh_component_capabilities(
&self,
request: &SelectionRequest,
component_excluded: &[Arc<WorkerState>],
) -> bool {
if self.ssh_pool.is_none() {
return false;
}
let mut handles = Vec::with_capacity(component_excluded.len());
for worker in component_excluded {
let worker = Arc::clone(worker);
let ssh_pool = self.ssh_pool.clone();
handles.push(tokio::spawn(async move {
let capabilities = crate::health::probe_worker_capabilities(
&worker,
TOOLCHAIN_PREFLIGHT_COMMAND_TIMEOUT,
ssh_pool.as_ref(),
)
.await?;
worker.set_capabilities(capabilities.clone()).await;
Some((worker, capabilities))
}));
}
let mut any_recovered = false;
for handle in handles {
let Ok(Some((worker, capabilities))) = handle.await else {
continue;
};
if rustup_component_capability_mismatch(request, &capabilities).is_none() {
let worker_id = worker.config.read().await.id.clone();
info!(
"Worker {} capability inventory was stale: live probe shows the required \
rustup component installed; re-running admission (issue #63)",
worker_id
);
any_recovered = true;
}
}
any_recovered
}
async fn get_eligible_workers_with_refresh(
&self,
pool: &WorkerPool,
request: &SelectionRequest,
excluded_worker_ids: &HashSet<String>,
allow_component_refresh: bool,
) -> Result<Vec<(Arc<WorkerState>, CircuitState)>, SelectionReason> {
if let Some(ref gate) = self.admission_gate {
gate.begin_round().await;
}
let workers = pool.healthy_workers().await;
if workers.is_empty() {
if pool.is_empty() {
return Err(SelectionReason::NoWorkersConfigured);
}
let all_workers = pool.all_workers().await;
let mut all_circuits_open = true;
let mut all_unreachable = true;
for worker in &all_workers {
if let Some(state) = worker.circuit_state().await
&& state != CircuitState::Open
{
all_circuits_open = false;
}
if !matches!(
worker.status().await,
rch_common::WorkerStatus::Unreachable
| rch_common::WorkerStatus::Drained
| rch_common::WorkerStatus::Disabled
) {
all_unreachable = false;
}
}
if all_circuits_open {
return Err(SelectionReason::AllCircuitsOpen);
}
if all_unreachable {
return Err(SelectionReason::AllWorkersUnreachable);
}
return Err(SelectionReason::AllWorkersUnreachable);
}
let preferred_set: HashSet<&str> = request
.preferred_workers
.iter()
.map(|id| id.as_str())
.collect();
let has_preferred = !preferred_set.is_empty();
let mut matched_preferred_worker = false;
let mut eligible: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut preferred: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut eligible_without_health: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut preferred_without_health: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut capacity_degraded: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut filtered_by_health = 0usize;
let mut filtered_by_hard_preflight = 0usize;
let mut filtered_by_component = 0usize;
let mut component_excluded: Vec<Arc<WorkerState>> = Vec::new();
let mut filtered_by_convergence = 0usize;
let mut filtered_by_pressure = 0usize;
let mut filtered_by_os_gate = 0usize;
let mut filtered_by_slots = 0usize;
let mut filtered_by_capacity = 0usize;
let mut filtered_by_active_project = 0usize;
let mut any_has_runtime = false;
let required_os = required_os_for_request(request);
for worker in workers {
let circuit_state = worker.circuit_state().await.unwrap_or(CircuitState::Closed);
let (worker_id, declared_os) = {
let config = worker.config.read().await;
(config.id.clone(), rch_common::declared_os(&config.tags))
};
if has_preferred && !preferred_set.contains(worker_id.as_str()) {
continue;
}
matched_preferred_worker = true;
if !os_gate_admits(declared_os.as_deref(), required_os.as_deref()) {
debug!(
"Worker {} excluded by OS gate: declared={:?} required={:?}",
worker_id, declared_os, required_os
);
filtered_by_os_gate += 1;
continue;
}
match circuit_state {
CircuitState::Open => continue,
CircuitState::HalfOpen => {
if !worker.can_probe(&self.circuit_config).await {
continue;
}
}
CircuitState::Closed => {}
}
let has_required_runtime = match &request.required_runtime {
RequiredRuntime::None => true,
RequiredRuntime::Rust => worker.has_rust().await,
RequiredRuntime::Bun => worker.has_bun().await,
RequiredRuntime::Node => worker.has_node().await,
RequiredRuntime::Nix => worker.has_nix().await,
RequiredRuntime::Go => worker.has_go().await,
RequiredRuntime::Zig => worker.has_zig().await,
};
if excluded_worker_ids.contains(worker_id.as_str()) {
debug!(
"Worker {} excluded: already active for project {}",
worker_id, request.project
);
if has_required_runtime {
any_has_runtime = true;
filtered_by_active_project += 1;
}
continue;
}
if !has_required_runtime {
continue;
}
any_has_runtime = true;
let capabilities = worker.capabilities().await;
if let Some(reason) = required_tool_capability_mismatch(request, &capabilities) {
debug!("Worker {} excluded: {}", worker_id, reason);
metrics::inc_reliability_error("selection", "required_tool_missing");
filtered_by_hard_preflight += 1;
continue;
}
if let Some(reason) = rustup_component_capability_mismatch(request, &capabilities) {
debug!("Worker {} excluded: {}", worker_id, reason);
metrics::inc_reliability_error("selection", "toolchain_component_missing");
filtered_by_hard_preflight += 1;
filtered_by_component += 1;
component_excluded.push(Arc::clone(&worker));
continue;
}
if let Some(reason) =
toolchain_capability_mismatch(request.toolchain.as_ref(), &capabilities)
{
debug!("Worker {} excluded: {}", worker_id, reason);
metrics::inc_reliability_error("selection", "toolchain_version_mismatch");
filtered_by_hard_preflight += 1;
continue;
}
if let Some(reason) = self
.toolchain_preflight_failure(worker.as_ref(), worker_id.as_str(), request)
.await
{
debug!(
"Worker {} excluded: toolchain preflight failed ({})",
worker_id, reason
);
metrics::inc_reliability_error("selection", "toolchain_preflight_failed");
filtered_by_hard_preflight += 1;
continue;
}
let mut undersized_but_free = false;
let total_slots = worker.effective_total_slots().await;
let available_slots = worker.available_slots().await;
if total_slots == 0 {
let hard_pressure = if let Some(ref gate) = self.admission_gate {
matches!(
gate.evaluate(worker.as_ref(), worker_id.as_str(), &request.project).await,
crate::admission::AdmissionVerdict::Reject { reason_code, .. }
if matches!(reason_code.as_str(), "admission_critical_pressure" | "admission_disk_floor")
)
} else {
worker.pressure_assessment().await.state == PressureState::Critical
};
if hard_pressure {
filtered_by_pressure += 1;
filtered_by_hard_preflight += 1;
metrics::inc_reliability_error("preflight_pressure", "critical_pressure");
continue;
}
}
if available_slots < request.estimated_cores {
if total_slots < request.estimated_cores {
if available_slots > 0 {
undersized_but_free = true;
} else {
filtered_by_capacity += 1;
debug!(
"Worker {} excluded: no free slots and total below estimate \
(available={}, total={}, requested={})",
worker_id, available_slots, total_slots, request.estimated_cores
);
continue;
}
} else {
filtered_by_slots += 1;
debug!(
"Worker {} excluded: insufficient free slots (available={}, total={}, requested={})",
worker_id, available_slots, total_slots, request.estimated_cores
);
continue;
}
}
if let Some(false) = capabilities.is_topology_healthy() {
let reason = capabilities
.projects_root_issue
.clone()
.unwrap_or_else(|| "unknown_topology_error".to_string());
debug!(
"Worker {} excluded: topology preflight failed ({})",
worker_id, reason
);
filtered_by_hard_preflight += 1;
continue;
}
let mut passes_preflight = true;
let mut hard_preflight_block = false;
if let Some(ref convergence_svc) = self.repo_convergence {
let convergence_start = Instant::now();
let wid = rch_common::WorkerId::new(worker_id.as_str());
let drift_state = convergence_svc.get_drift_state(&wid).await;
let convergence_outcome = match drift_state {
crate::repo_convergence::ConvergenceDriftState::Ready => "ready",
crate::repo_convergence::ConvergenceDriftState::Drifting => "drifting",
crate::repo_convergence::ConvergenceDriftState::Converging => "converging",
crate::repo_convergence::ConvergenceDriftState::Failed => "failed",
crate::repo_convergence::ConvergenceDriftState::Stale => "stale",
};
metrics::observe_reliability_decision(
"preflight_convergence",
convergence_outcome,
convergence_start.elapsed(),
);
match drift_state {
crate::repo_convergence::ConvergenceDriftState::Ready => {
}
crate::repo_convergence::ConvergenceDriftState::Drifting => {
debug!(
"Worker {} convergence drifting: some repos may be stale",
worker_id
);
}
crate::repo_convergence::ConvergenceDriftState::Converging => {
debug!(
"Worker {} excluded: repo convergence sync in progress",
worker_id
);
passes_preflight = false;
hard_preflight_block = true;
filtered_by_convergence += 1;
}
crate::repo_convergence::ConvergenceDriftState::Failed => {
debug!(
"Worker {} excluded: repo convergence failed (budgets exhausted)",
worker_id
);
metrics::inc_reliability_error(
"preflight_convergence",
"convergence_failed",
);
passes_preflight = false;
hard_preflight_block = true;
filtered_by_convergence += 1;
}
crate::repo_convergence::ConvergenceDriftState::Stale => {
debug!(
"Worker {} convergence stale: no recent status check, fail-open allows",
worker_id
);
}
}
}
if let Some(max_load) = self.config.max_load_per_core
&& let Some(true) = capabilities.is_high_load(max_load)
{
let load_per_core = capabilities.load_per_core().unwrap_or(0.0);
debug!(
"Worker {} excluded: high load ({:.2} > {:.2} per core)",
worker_id, load_per_core, max_load
);
passes_preflight = false;
}
if let Some(min_disk) = self.config.min_free_gb
&& let Some(true) = capabilities.is_low_disk(min_disk)
{
let free_gb = capabilities.build_disk_gb().0.unwrap_or(0.0);
debug!(
"Worker {} excluded: low disk ({:.1} GB < {:.1} GB)",
worker_id, free_gb, min_disk
);
passes_preflight = false;
}
if let Some(ref gate) = self.admission_gate {
use crate::admission::AdmissionVerdict;
let admission_start = Instant::now();
let verdict = gate
.evaluate(worker.as_ref(), worker_id.as_str(), &request.project)
.await;
match verdict {
AdmissionVerdict::Reject {
reason_code,
reason,
} => {
debug!(
"Worker {} admission rejected: {} ({})",
worker_id, reason, reason_code
);
metrics::observe_reliability_decision(
"preflight_pressure",
"reject",
admission_start.elapsed(),
);
if matches!(
reason_code.as_str(),
"admission_critical_pressure" | "admission_disk_floor"
) {
metrics::inc_reliability_error(
"preflight_pressure",
"critical_pressure",
);
filtered_by_pressure += 1;
hard_preflight_block = true;
}
passes_preflight = false;
}
AdmissionVerdict::Admit {
pressure_penalty,
headroom_score,
} => {
debug!(
"Worker {} admitted: penalty={:.2}, headroom={:.2}",
worker_id, pressure_penalty, headroom_score
);
metrics::observe_reliability_decision(
"preflight_pressure",
"admit",
admission_start.elapsed(),
);
}
}
} else {
let pressure_start = Instant::now();
let pressure = worker.pressure_assessment().await;
let pressure_outcome = match pressure.state {
PressureState::Healthy => "healthy",
PressureState::Warning => "warning",
PressureState::Critical => "critical",
PressureState::TelemetryGap => "telemetry_gap",
};
metrics::observe_reliability_decision(
"preflight_pressure",
pressure_outcome,
pressure_start.elapsed(),
);
match pressure.state {
PressureState::Critical => {
debug!(
"Worker {} excluded: pressure {} (confidence={}, reason={}, rule={})",
worker_id,
pressure.state,
pressure.confidence,
pressure.reason_code,
pressure.policy_rule
);
metrics::inc_reliability_error("preflight_pressure", "critical_pressure");
passes_preflight = false;
filtered_by_pressure += 1;
hard_preflight_block = true;
}
PressureState::Warning => {
debug!(
"Worker {} pressure warning (confidence={}, reason={}, rule={})",
worker_id,
pressure.confidence,
pressure.reason_code,
pressure.policy_rule
);
}
PressureState::TelemetryGap => {
debug!(
"Worker {} pressure telemetry gap (confidence={}, reason={}, rule={}); fail-open keeps worker eligible",
worker_id,
pressure.confidence,
pressure.reason_code,
pressure.policy_rule
);
}
PressureState::Healthy => {}
}
}
if let Some(ref agg) = self.reliability {
let reliability_start = Instant::now();
let assessment = agg.evaluate(worker.as_ref(), worker_id.as_str()).await;
let reliability_outcome = if assessment.hard_exclude {
"quarantined"
} else if assessment.penalty > 0.0 {
"degraded"
} else {
"healthy"
};
metrics::observe_reliability_decision(
"reliability_preflight",
reliability_outcome,
reliability_start.elapsed(),
);
if assessment.hard_exclude {
debug!(
"Worker {} excluded: reliability quarantined (debt={:.2}, state={})",
worker_id, assessment.aggregated_debt, assessment.health_state
);
metrics::inc_reliability_error("reliability_preflight", "quarantined");
passes_preflight = false;
hard_preflight_block = true;
}
}
if hard_preflight_block {
metrics::inc_reliability_error("selection", "preflight_blocked");
filtered_by_hard_preflight += 1;
continue;
}
if !passes_preflight {
if undersized_but_free {
filtered_by_capacity += 1;
continue;
}
if has_preferred && preferred_set.contains(worker_id.as_str()) {
preferred_without_health.push((worker.clone(), circuit_state));
}
eligible_without_health.push((worker.clone(), circuit_state));
continue;
}
let success_rate = self.health_score(&worker).await;
if success_rate < self.config.min_success_rate {
filtered_by_health += 1;
debug!(
"Worker {} excluded: success_rate {:.2} < min {:.2}",
worker_id, success_rate, self.config.min_success_rate
);
if !undersized_but_free
&& success_rate >= self.config.affinity.fallback_min_success_rate
{
if has_preferred && preferred_set.contains(worker_id.as_str()) {
preferred_without_health.push((worker.clone(), circuit_state));
} else {
eligible_without_health.push((worker.clone(), circuit_state));
}
}
continue;
}
if undersized_but_free {
filtered_by_capacity += 1;
debug!(
"Worker {} held as degraded candidate: passes admission but total slots {} < requested {}",
worker_id, total_slots, request.estimated_cores
);
capacity_degraded.push((worker.clone(), circuit_state));
continue;
}
if has_preferred && preferred_set.contains(worker_id.as_str()) {
preferred.push((worker.clone(), circuit_state));
}
eligible.push((worker, circuit_state));
}
if has_preferred && !matched_preferred_worker {
return Err(SelectionReason::NoMatchingWorkers);
}
let filtered_by_anything_else = filtered_by_health
+ filtered_by_hard_preflight
+ filtered_by_pressure
+ filtered_by_slots
+ filtered_by_capacity
+ filtered_by_active_project
+ filtered_by_convergence
+ filtered_by_component;
if filtered_by_os_gate > 0
&& filtered_by_anything_else == 0
&& eligible.is_empty()
&& preferred.is_empty()
&& eligible_without_health.is_empty()
&& preferred_without_health.is_empty()
{
let detail = match required_os.as_deref() {
Some(os) => format!(
"os_gate_excluded={filtered_by_os_gate} required_os={os} \
(no worker declares `os = \"{os}\"` in workers.toml)"
),
None => format!(
"os_gate_excluded={filtered_by_os_gate} required_os=none \
(every candidate declares an `os`, which restricts it to \
commands targeting that OS; an unqualified build needs at \
least one worker with no `os` set)"
),
};
return Err(SelectionReason::NoAdmissibleWorkers(detail));
}
if !any_has_runtime && !matches!(request.required_runtime, RequiredRuntime::None) {
return Err(SelectionReason::NoWorkersWithRuntime(format!(
"{:?}",
request.required_runtime
)));
}
if has_preferred && !preferred.is_empty() {
return Ok(preferred);
}
if has_preferred && !preferred_without_health.is_empty() {
return Ok(preferred_without_health);
}
if has_preferred && filtered_by_slots > 0 {
return Ok(Vec::new());
}
if !eligible.is_empty() {
return Ok(eligible);
}
if request.job_mode
&& filtered_by_active_project > 0
&& preferred.is_empty()
&& eligible_without_health.is_empty()
&& preferred_without_health.is_empty()
{
return Ok(Vec::new());
}
if !has_preferred
&& filtered_by_slots == 0
&& filtered_by_active_project == 0
&& eligible.is_empty()
&& preferred.is_empty()
&& eligible_without_health.is_empty()
&& preferred_without_health.is_empty()
&& !capacity_degraded.is_empty()
{
debug!(
"No worker can satisfy estimated_cores={}; degrading to {} worker(s) \
with free capacity below the estimate rather than falling back to local",
request.estimated_cores,
capacity_degraded.len()
);
metrics::inc_reliability_error("selection", "capacity_degraded_admission");
return Ok(capacity_degraded);
}
if (filtered_by_active_project > 0 || (filtered_by_capacity > 0 && filtered_by_slots == 0))
&& preferred_without_health.is_empty()
&& eligible_without_health.is_empty()
{
return Err(SelectionReason::NoAdmissibleWorkers(append_os_gate(
no_admissible_workers_summary_with_active_project(
filtered_by_pressure,
filtered_by_slots,
filtered_by_capacity,
filtered_by_hard_preflight,
filtered_by_health,
filtered_by_active_project,
),
filtered_by_os_gate,
)));
}
if filtered_by_hard_preflight > 0
&& preferred_without_health.is_empty()
&& eligible_without_health.is_empty()
{
debug!(
"All candidate workers failed hard preflight checks (count={})",
filtered_by_hard_preflight
);
if filtered_by_component > 0 {
if allow_component_refresh
&& !component_excluded.is_empty()
&& self
.refresh_component_capabilities(request, &component_excluded)
.await
{
return Box::pin(self.get_eligible_workers_with_refresh(
pool,
request,
excluded_worker_ids,
false,
))
.await;
}
return Err(SelectionReason::NoAdmissibleWorkers(format!(
"missing_toolchain_component={filtered_by_component}"
)));
}
if filtered_by_pressure > 0
|| filtered_by_slots > 0
|| filtered_by_capacity > 0
|| filtered_by_health > 0
{
return Err(SelectionReason::NoAdmissibleWorkers(append_os_gate(
no_admissible_workers_summary(
filtered_by_pressure,
filtered_by_slots,
filtered_by_capacity,
filtered_by_hard_preflight,
filtered_by_health,
),
filtered_by_os_gate,
)));
}
if filtered_by_convergence > 0 && filtered_by_convergence >= filtered_by_hard_preflight
{
debug!(
"All candidate workers failed repo convergence checks (count={})",
filtered_by_convergence
);
return Err(SelectionReason::AllWorkersFailedConvergence);
}
return Err(SelectionReason::AllWorkersFailedPreflight);
}
if filtered_by_health > 0
&& preferred_without_health.is_empty()
&& eligible_without_health.is_empty()
{
debug!(
"No workers passed selection health thresholds (min_success_rate {:.2}, fallback_min_success_rate {:.2})",
self.config.min_success_rate, self.config.affinity.fallback_min_success_rate
);
return Err(SelectionReason::NoWorkersPassedHealth);
}
if filtered_by_health > 0 {
debug!(
"No workers meet min_success_rate {:.2}; falling back to workers below threshold",
self.config.min_success_rate
);
}
if has_preferred && eligible_without_health.is_empty() {
return Err(SelectionReason::NoMatchingWorkers);
}
Ok(eligible_without_health)
}
async fn toolchain_preflight_failure(
&self,
worker: &WorkerState,
worker_id: &str,
request: &SelectionRequest,
) -> Option<String> {
let toolchain = request.toolchain.as_ref()?;
let toolchain_name = toolchain.rustup_toolchain();
if let Some(cached) = worker.toolchain_preflight_status(&toolchain_name).await
&& cached.is_reusable(TOOLCHAIN_PREFLIGHT_TTL, TOOLCHAIN_PREFLIGHT_TRANSIENT_TTL)
{
return (!cached.usable).then(|| {
cached
.reason
.unwrap_or_else(|| "cached_toolchain_unusable".to_string())
});
}
let result = probe_worker_toolchain(worker, &toolchain_name, self.ssh_pool.as_ref()).await;
match result {
Ok(()) => {
worker
.record_toolchain_preflight(toolchain_name, true, None)
.await;
None
}
Err(reason) => {
warn!(
worker = %worker_id,
toolchain = %toolchain_name,
reason = %reason,
"Worker toolchain preflight failed"
);
worker
.record_toolchain_preflight(toolchain_name, false, Some(reason.clone()))
.await;
Some(reason)
}
}
}
async fn select_by_priority(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
cache_use: CacheUse,
) -> Option<(Arc<WorkerState>, CircuitState)> {
let cache = self.cache_tracker.read().await;
let mut best: Option<(Arc<WorkerState>, CircuitState)> = None;
let mut best_score = f64::NEG_INFINITY;
for (worker, circuit_state) in workers {
let config = worker.config.read().await;
let warmth = cache.estimate_warmth(config.id.as_str(), &request.project, cache_use);
let score = Self::priority_selection_score(
config.priority,
warmth,
worker.get_speed_score(),
request.command_priority,
);
drop(config);
if score > best_score {
best_score = score;
best = Some((worker.clone(), *circuit_state));
}
}
best
}
fn priority_selection_score(
priority: u32,
warmth: f64,
speed_score: f64,
command_priority: CommandPriority,
) -> f64 {
let warmth = if warmth.is_finite() {
warmth.clamp(0.0, 1.0)
} else {
0.0
};
let speed_score = if speed_score.is_finite() {
speed_score.clamp(0.0, 100.0)
} else {
0.0
};
let tie_break = match command_priority {
CommandPriority::High => (speed_score * PRIORITY_SPEED_TIEBREAK_SCORE) + warmth,
CommandPriority::Normal | CommandPriority::Low => {
(warmth * PRIORITY_CACHE_TIEBREAK_SCORE) + speed_score
}
};
(priority as f64 * PRIORITY_BUCKET_SCORE) + tie_break
}
async fn select_by_fastest(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
) -> Option<(Arc<WorkerState>, CircuitState)> {
let mut best: Option<&(Arc<WorkerState>, CircuitState)> = None;
for pair in workers {
let (worker, _) = pair;
let score = worker.get_speed_score();
match best {
None => best = Some(pair),
Some((best_worker, _)) => {
let best_score = best_worker.get_speed_score();
if score > best_score {
best = Some(pair);
}
}
}
}
best.cloned()
}
async fn select_balanced(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
cache_use: CacheUse,
) -> Option<(Arc<WorkerState>, CircuitState)> {
let cache = self.cache_tracker.read().await;
let weights = adjust_weights_for_priority(&self.config.weights, request);
let (min_priority, max_priority) = Self::priority_range(workers).await;
let mut best: Option<&(Arc<WorkerState>, CircuitState)> = None;
let mut best_score = f64::NEG_INFINITY;
for pair in workers {
let (worker, circuit_state) = pair;
let score = self
.compute_balanced_score(
worker,
*circuit_state,
&request.project,
&cache,
cache_use,
&weights,
min_priority,
max_priority,
)
.await;
if score > best_score {
best_score = score;
best = Some(pair);
}
}
best.cloned()
}
#[allow(clippy::too_many_arguments)]
async fn compute_balanced_score(
&self,
worker: &WorkerState,
circuit_state: CircuitState,
project: &str,
cache: &CacheTracker,
cache_use: CacheUse,
weights: &SelectionWeightConfig,
min_priority: u32,
max_priority: u32,
) -> f64 {
let speed_score = (worker.get_speed_score() / 100.0).clamp(0.0, 1.0);
let total_slots = worker.effective_total_slots().await;
let total_slots = if total_slots == 0 {
return 0.0; } else {
total_slots as f64
};
let utilization = (worker.used_slots() as f64 / total_slots).min(1.0);
let load_factor = 1.0 - (utilization * 0.5);
let slot_score = 1.0 - utilization;
let disk_headroom = worker.disk_headroom().await;
let config = worker.config.read().await;
let mut cache_score = cache.estimate_warmth(config.id.as_str(), project, cache_use);
if cache_use == CacheUse::Test {
cache_score = (cache_score * TEST_CACHE_BOOST).min(1.0);
}
let health_score = self.health_score(worker).await;
let network_score = self.network_score(worker);
let priority_score = Self::normalize_priority(config.priority, min_priority, max_priority);
let base_score = weights.speedscore * speed_score
+ weights.slots * slot_score * load_factor
+ weights.health * health_score
+ weights.cache * cache_score
+ weights.network * network_score
+ weights.priority * priority_score;
let mut final_score = if circuit_state == CircuitState::HalfOpen {
base_score * weights.half_open_penalty
} else {
base_score
};
let admission_penalty = if let Some(ref gate) = self.admission_gate {
let p = gate.get_pressure_penalty(config.id.as_str()).await;
if p > 0.0 {
final_score *= 1.0 - p;
}
p
} else {
0.0
};
let reliability_penalty = if let Some(ref agg) = self.reliability {
let assessment = match agg.get_assessment(config.id.as_str()).await {
Some(assessment) => assessment,
None => agg.evaluate(worker, config.id.as_str()).await,
};
let p = assessment.penalty;
if p > 0.0 {
final_score *= 1.0 - p;
}
p
} else {
0.0
};
let capabilities = worker.capabilities().await;
if capabilities.is_pre_v3_x86() {
final_score *= 1.0 - PRE_V3_MICROARCH_PENALTY;
}
final_score += weights.disk * disk_headroom;
debug!(
"Worker {} balanced score: {:.3} (speed={:.2}, load={:.2}, slots={:.2}, disk={:.2}, health={:.2}, cache={:.2}, network={:.2}, priority={:.2}, half_open={:?}, admission_penalty={:.2}, reliability_penalty={:.2}, cache_use={:?})",
config.id,
final_score,
speed_score,
load_factor,
slot_score,
disk_headroom,
health_score,
cache_score,
network_score,
priority_score,
circuit_state == CircuitState::HalfOpen,
admission_penalty,
reliability_penalty,
cache_use
);
final_score
}
async fn health_score(&self, worker: &WorkerState) -> f64 {
let stats = worker.circuit_stats().await;
let success_rate = 1.0 - stats.error_rate();
success_rate.clamp(0.0, 1.0)
}
fn network_score(&self, worker: &WorkerState) -> f64 {
match worker.last_latency_ms() {
Some(latency_ms) => Self::normalize_latency_ms(latency_ms),
None => DEFAULT_NETWORK_SCORE,
}
}
fn normalize_latency_ms(latency_ms: u64) -> f64 {
let latency = latency_ms as f64;
let score = 1.0 / (1.0 + (latency / NETWORK_LATENCY_HALF_LIFE_MS));
score.clamp(0.0, 1.0)
}
async fn select_cache_affinity(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
request: &SelectionRequest,
cache_use: CacheUse,
) -> Option<(Arc<WorkerState>, CircuitState)> {
let cache = self.cache_tracker.read().await;
let mut warm_workers = Vec::new();
for pair in workers {
let (w, _) = pair;
let id = w.config.read().await.id.clone();
if cache.estimate_warmth(id.as_str(), &request.project, cache_use) > 0.5 {
warm_workers.push(pair.clone());
}
}
if !warm_workers.is_empty() {
return self.select_by_fastest(&warm_workers).await;
}
self.select_by_fastest(workers).await
}
async fn select_fair_fastest(
&self,
workers: &[(Arc<WorkerState>, CircuitState)],
) -> Option<(Arc<WorkerState>, CircuitState)> {
if workers.is_empty() {
return None;
}
let history = self.selection_history.read().await;
let lookback = Duration::from_secs(self.config.fairness.lookback_secs);
let mut weights = Vec::with_capacity(workers.len());
for (w, _) in workers {
let speed = w.get_speed_score().max(10.0);
let id = w.config.read().await.id.clone();
let recent = history.recent_selections(id.as_str(), lookback);
weights.push(speed / (1.0 + recent as f64));
}
let total_weight: f64 = weights.iter().sum();
if total_weight <= 0.0 {
return workers.first().map(|(w, s)| (w.clone(), *s));
}
let mut rng = rand::rng();
let threshold = rng.random_range(0.0..total_weight);
let mut cumulative = 0.0;
for (i, weight) in weights.iter().enumerate() {
cumulative += weight;
if cumulative >= threshold {
return Some((workers[i].0.clone(), workers[i].1));
}
}
workers.last().map(|(w, s)| (w.clone(), *s))
}
async fn priority_range(workers: &[(Arc<WorkerState>, CircuitState)]) -> (u32, u32) {
let mut min_priority = u32::MAX;
let mut max_priority = 0u32;
for (worker, _) in workers {
let priority = worker.config.read().await.priority;
min_priority = min_priority.min(priority);
max_priority = max_priority.max(priority);
}
if min_priority == u32::MAX {
(0, 0)
} else {
(min_priority, max_priority)
}
}
fn normalize_priority(priority: u32, min_priority: u32, max_priority: u32) -> f64 {
if max_priority == min_priority {
1.0
} else {
(priority.saturating_sub(min_priority)) as f64 / (max_priority - min_priority) as f64
}
}
}
impl Default for WorkerSelector {
fn default() -> Self {
Self::new()
}
}
fn adjust_weights_for_priority(
weights: &SelectionWeightConfig,
request: &SelectionRequest,
) -> SelectionWeightConfig {
let mut adjusted = weights.clone();
match request.command_priority {
CommandPriority::Normal => {}
CommandPriority::High => {
adjusted.speedscore *= 1.25;
adjusted.network *= 1.15;
adjusted.cache *= 0.85;
}
CommandPriority::Low => {
adjusted.cache *= 1.25;
adjusted.slots *= 1.15;
adjusted.speedscore *= 0.85;
adjusted.network *= 0.85;
}
}
adjusted
}
pub async fn select_worker(
pool: &WorkerPool,
request: &SelectionRequest,
weights: &SelectionWeights,
) -> Option<Arc<WorkerState>> {
let config = CircuitBreakerConfig::default();
select_worker_with_config(pool, request, weights, &config)
.await
.worker
}
fn required_os_for_request(request: &SelectionRequest) -> Option<String> {
request
.command
.as_deref()
.and_then(rch_common::admit_preflight::required_os_for_command)
}
fn os_gate_admits(declared_os: Option<&str>, required_os: Option<&str>) -> bool {
match (declared_os, required_os) {
(None, None) => true,
(Some(declared), Some(required)) => declared.eq_ignore_ascii_case(required),
_ => false,
}
}
pub async fn select_worker_with_config(
pool: &WorkerPool,
request: &SelectionRequest,
weights: &SelectionWeights,
circuit_config: &CircuitBreakerConfig,
) -> SelectionResult {
let _timer = DecisionTimer::new(DecisionType::WorkerSelection);
let workers = pool.healthy_workers().await;
if workers.is_empty() {
if pool.is_empty() {
return SelectionResult {
worker: None,
reason: SelectionReason::NoWorkersConfigured,
diagnostics: None,
};
}
let all_workers = pool.all_workers().await;
let mut all_circuits_open = true;
let mut all_unreachable = true;
for worker in &all_workers {
if let Some(state) = worker.circuit_state().await
&& state != CircuitState::Open
{
all_circuits_open = false;
}
if !matches!(
worker.status().await,
rch_common::WorkerStatus::Unreachable
| rch_common::WorkerStatus::Drained
| rch_common::WorkerStatus::Disabled
) {
all_unreachable = false;
}
}
if all_circuits_open {
return SelectionResult {
worker: None,
reason: SelectionReason::AllCircuitsOpen,
diagnostics: None,
};
}
if all_unreachable {
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersUnreachable,
diagnostics: None,
};
}
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersUnreachable,
diagnostics: None,
};
}
let preferred_set: HashSet<&str> = request
.preferred_workers
.iter()
.map(|id| id.as_str())
.collect();
let has_preferred = !preferred_set.is_empty();
let mut matched_preferred_worker = false;
let required_os = required_os_for_request(request);
let mut eligible: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut preferred: Vec<(Arc<WorkerState>, CircuitState)> = Vec::new();
let mut all_circuits_open = true;
let mut any_has_slots = false;
let mut any_has_capacity = false;
let mut any_has_runtime = false;
let mut filtered_by_capacity = 0usize;
let mut filtered_by_component = 0usize;
for worker in workers {
let circuit_state = worker.circuit_state().await.unwrap_or(CircuitState::Closed);
let (worker_id, declared_os) = {
let config = worker.config.read().await;
(config.id.clone(), rch_common::declared_os(&config.tags))
};
let total_slots = worker.effective_total_slots().await;
if has_preferred && !preferred_set.contains(worker_id.as_str()) {
continue;
}
matched_preferred_worker = true;
if !os_gate_admits(declared_os.as_deref(), required_os.as_deref()) {
debug!(
"Worker {} excluded by OS gate: declared={:?} required={:?}",
worker_id, declared_os, required_os
);
continue;
}
match circuit_state {
CircuitState::Open => {
debug!("Worker {} excluded: circuit open", worker_id);
continue;
}
CircuitState::HalfOpen => {
if !worker.can_probe(circuit_config).await {
debug!("Worker {} excluded: half-open, no probe budget", worker_id);
continue;
}
all_circuits_open = false;
}
CircuitState::Closed => {
all_circuits_open = false;
}
}
let has_required_runtime = match &request.required_runtime {
RequiredRuntime::None => true,
RequiredRuntime::Rust => worker.has_rust().await,
RequiredRuntime::Bun => worker.has_bun().await,
RequiredRuntime::Node => worker.has_node().await,
RequiredRuntime::Nix => worker.has_nix().await,
RequiredRuntime::Go => worker.has_go().await,
RequiredRuntime::Zig => worker.has_zig().await,
};
if !has_required_runtime {
debug!(
"Worker {} excluded: missing required runtime {:?}",
worker_id, request.required_runtime
);
continue;
}
any_has_runtime = true;
let capabilities = worker.capabilities().await;
if let Some(reason) = required_tool_capability_mismatch(request, &capabilities) {
debug!("Worker {} excluded: {}", worker_id, reason);
continue;
}
if let Some(reason) = rustup_component_capability_mismatch(request, &capabilities) {
filtered_by_component += 1;
debug!("Worker {} excluded: {}", worker_id, reason);
continue;
}
if total_slots < request.estimated_cores {
filtered_by_capacity += 1;
debug!(
"Worker {} excluded: requested cores exceed capacity ({} > {})",
worker_id, request.estimated_cores, total_slots
);
continue;
}
any_has_capacity = true;
if worker.available_slots().await < request.estimated_cores {
debug!(
"Worker {} excluded: insufficient slots ({} < {})",
worker_id,
worker.available_slots().await,
request.estimated_cores
);
continue;
}
any_has_slots = true;
if has_preferred && preferred_set.contains(worker_id.as_str()) {
preferred.push((worker.clone(), circuit_state));
}
eligible.push((worker, circuit_state));
}
let mut candidates = if has_preferred { preferred } else { eligible };
if candidates.is_empty() {
if has_preferred && !matched_preferred_worker {
return SelectionResult {
worker: None,
reason: SelectionReason::NoMatchingWorkers,
diagnostics: None,
};
}
if !any_has_runtime && !matches!(request.required_runtime, RequiredRuntime::None) {
return SelectionResult {
worker: None,
reason: SelectionReason::NoWorkersWithRuntime(format!(
"{:?}",
request.required_runtime
)),
diagnostics: None,
};
}
if all_circuits_open {
return SelectionResult {
worker: None,
reason: SelectionReason::AllCircuitsOpen,
diagnostics: None,
};
}
if !any_has_capacity && filtered_by_capacity > 0 {
return SelectionResult {
worker: None,
reason: SelectionReason::NoAdmissibleWorkers(format!(
"insufficient_total_slots={filtered_by_capacity}"
)),
diagnostics: None,
};
}
if filtered_by_component > 0 {
return SelectionResult {
worker: None,
reason: SelectionReason::NoAdmissibleWorkers(format!(
"missing_toolchain_component={filtered_by_component}"
)),
diagnostics: None,
};
}
if !any_has_slots {
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersBusy,
diagnostics: None,
};
}
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersBusy,
diagnostics: None,
};
}
let (min_priority, max_priority) = priority_range(&candidates).await;
let mut scored: Vec<(Arc<WorkerState>, CircuitState, f64)> =
Vec::with_capacity(candidates.len());
for (worker, circuit_state) in candidates.drain(..) {
let config = worker.config.read().await;
let priority_score = normalize_priority(config.priority, min_priority, max_priority);
let id = config.id.clone();
drop(config);
let final_score =
compute_score(&worker, request, weights, priority_score, circuit_state).await;
debug!(
"Worker {} candidate: circuit={:?}, final_score={:.3}, priority_score={:.2}",
id, circuit_state, final_score, priority_score
);
scored.push((worker, circuit_state, final_score));
}
scored.sort_by(|a, b| b.2.partial_cmp(&a.2).unwrap_or(Ordering::Equal));
let Some((selected_worker, circuit_state, score)) = scored.into_iter().next() else {
warn!("Bug: scored list empty despite non-empty candidates - this should not happen");
return SelectionResult {
worker: None,
reason: SelectionReason::AllWorkersBusy,
diagnostics: None,
};
};
if circuit_state == CircuitState::HalfOpen {
selected_worker.start_probe(circuit_config).await;
debug!(
"Worker {} selected (half-open probe started), score={:.3}",
selected_worker.config.read().await.id,
score
);
} else {
debug!(
"Worker {} selected, score={:.3}",
selected_worker.config.read().await.id,
score
);
}
SelectionResult {
worker: Some(selected_worker),
reason: SelectionReason::Success,
diagnostics: None,
}
}
async fn compute_score(
worker: &WorkerState,
request: &SelectionRequest,
weights: &SelectionWeights,
priority_score: f64,
circuit_state: CircuitState,
) -> f64 {
let total_slots = worker.effective_total_slots().await.max(1) as f64;
let slot_score = (worker.available_slots().await as f64 / total_slots).min(1.0);
let disk_headroom = worker.disk_headroom().await;
let config = worker.config.read().await;
let speed_score = worker.get_speed_score() / 100.0;
let locality_score = if worker.has_cached_project(&request.project).await {
1.0
} else {
0.5
};
let base_score = weights.slots * slot_score
+ weights.speed * speed_score
+ weights.locality * locality_score;
let priority_factor = 1.0 + (weights.priority * priority_score);
let circuit_factor = if circuit_state == CircuitState::HalfOpen {
weights.half_open_penalty
} else {
1.0
};
let score = base_score * priority_factor * circuit_factor + weights.disk * disk_headroom;
debug!(
"Worker {} score: {:.3} (slots: {:.2}, speed: {:.2}, locality: {:.2}, disk: {:.2}, priority: {:.2})",
config.id, score, slot_score, speed_score, locality_score, disk_headroom, priority_score
);
score
}
async fn priority_range(candidates: &[(Arc<WorkerState>, CircuitState)]) -> (u32, u32) {
let mut min_priority = u32::MAX;
let mut max_priority = 0u32;
for (worker, _) in candidates {
let priority = worker.config.read().await.priority;
min_priority = min_priority.min(priority);
max_priority = max_priority.max(priority);
}
if min_priority == u32::MAX {
(0, 0)
} else {
(min_priority, max_priority)
}
}
fn normalize_priority(priority: u32, min_priority: u32, max_priority: u32) -> f64 {
if max_priority == min_priority {
1.0
} else {
(priority.saturating_sub(min_priority)) as f64 / (max_priority - min_priority) as f64
}
}
fn selection_reason_label(reason: &SelectionReason) -> &'static str {
match reason {
SelectionReason::Success => "success",
SelectionReason::NoWorkersConfigured => "no_workers_configured",
SelectionReason::AllWorkersUnreachable => "all_workers_unreachable",
SelectionReason::AllCircuitsOpen => "all_circuits_open",
SelectionReason::AllWorkersBusy => "all_workers_busy",
SelectionReason::NoWorkersPassedHealth => "no_workers_passed_health",
SelectionReason::AllWorkersFailedPreflight => "all_workers_failed_preflight",
SelectionReason::AllWorkersFailedConvergence => "all_workers_failed_convergence",
SelectionReason::NoAdmissibleWorkers(_) => "no_admissible_workers",
SelectionReason::NoMatchingWorkers => "no_matching_workers",
SelectionReason::NoWorkersWithRuntime(_) => "no_workers_with_runtime",
SelectionReason::SelectionError(_) => "selection_error",
SelectionReason::AffinityPinned => "affinity_pinned",
SelectionReason::AffinityFallback => "affinity_fallback",
}
}
fn worker_status_label(status: WorkerStatus) -> &'static str {
match status {
WorkerStatus::Healthy => "healthy",
WorkerStatus::Degraded => "degraded",
WorkerStatus::Unreachable => "unreachable",
WorkerStatus::Draining => "draining",
WorkerStatus::Drained => "drained",
WorkerStatus::Disabled => "disabled",
}
}
fn circuit_state_label(state: CircuitState) -> &'static str {
match state {
CircuitState::Closed => "closed",
CircuitState::Open => "open",
CircuitState::HalfOpen => "half_open",
}
}
fn push_reason_code(reason_codes: &mut Vec<String>, reason_code: &'static str) {
if !reason_codes.iter().any(|existing| existing.eq(reason_code)) {
reason_codes.push(reason_code.to_string());
}
}
fn append_os_gate(mut summary: String, os_gate: usize) -> String {
if os_gate == 0 {
return summary;
}
if !summary.is_empty() {
summary.push(',');
}
summary.push_str(&format!("os_gate_excluded={os_gate}"));
summary
}
fn no_admissible_workers_summary(
critical_pressure: usize,
insufficient_slots: usize,
insufficient_capacity: usize,
hard_preflight: usize,
health_below_fallback: usize,
) -> String {
let mut parts = Vec::new();
if critical_pressure > 0 {
parts.push(format!("critical_pressure={critical_pressure}"));
}
if insufficient_slots > 0 {
parts.push(format!("insufficient_slots={insufficient_slots}"));
}
if insufficient_capacity > 0 {
parts.push(format!("insufficient_total_slots={insufficient_capacity}"));
}
if health_below_fallback > 0 {
parts.push(format!("health_below_fallback={health_below_fallback}"));
}
let generic_hard_preflight = hard_preflight.saturating_sub(critical_pressure);
if generic_hard_preflight > 0 {
parts.push(format!("hard_preflight={generic_hard_preflight}"));
}
parts.join(",")
}
fn no_admissible_workers_summary_with_active_project(
critical_pressure: usize,
insufficient_slots: usize,
insufficient_capacity: usize,
hard_preflight: usize,
health_below_fallback: usize,
active_project_exclusion: usize,
) -> String {
let mut summary = no_admissible_workers_summary(
critical_pressure,
insufficient_slots,
insufficient_capacity,
hard_preflight,
health_below_fallback,
);
if active_project_exclusion == 0 {
return summary;
}
if !summary.is_empty() {
summary.push(',');
}
summary.push_str(&format!(
"active_project_exclusion={active_project_exclusion}"
));
summary
}
fn selection_outcome_label(reason: &SelectionReason) -> &'static str {
match reason {
SelectionReason::Success | SelectionReason::AffinityPinned => "success",
SelectionReason::AffinityFallback => "fallback",
_ => "failure",
}
}
struct NoWorkerTracker {
recent: VecDeque<bool>,
alarm_active: bool,
}
const NO_WORKER_WINDOW: usize = 20;
const NO_WORKER_FIRE_FRACTION: f64 = 0.75;
const NO_WORKER_CLEAR_FRACTION: f64 = 0.40;
static NO_WORKER_TRACKER: std::sync::Mutex<NoWorkerTracker> =
std::sync::Mutex::new(NoWorkerTracker {
recent: VecDeque::new(),
alarm_active: false,
});
fn record_no_worker_outcome(got_worker: bool) {
let mut t = NO_WORKER_TRACKER
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
t.recent.push_back(got_worker);
while t.recent.len() > NO_WORKER_WINDOW {
t.recent.pop_front();
}
if t.recent.len() < NO_WORKER_WINDOW {
return;
}
let no_worker = t.recent.iter().filter(|&&g| !g).count();
let frac = no_worker as f64 / t.recent.len() as f64;
if !t.alarm_active && frac >= NO_WORKER_FIRE_FRACTION {
t.alarm_active = true;
warn!(
"⚠️ RCH FLEET DEGRADED: {:.0}% of the last {} offload requests got NO worker — \
compilation is falling back to LOCAL fleet-wide. Likely stranded workers or \
rustc-toolchain skew. Check `rch status` (worker health) and worker nightly parity.",
frac * 100.0,
NO_WORKER_WINDOW
);
} else if t.alarm_active && frac <= NO_WORKER_CLEAR_FRACTION {
t.alarm_active = false;
info!(
"RCH fleet offloading recovered ({:.0}% no-worker over last {} requests)",
frac * 100.0,
NO_WORKER_WINDOW
);
}
}
fn record_selection_metrics(reason: &SelectionReason, duration: Duration) {
metrics::observe_reliability_decision("selection", selection_outcome_label(reason), duration);
let got_worker = matches!(
reason,
SelectionReason::Success
| SelectionReason::AffinityPinned
| SelectionReason::AffinityFallback
);
record_no_worker_outcome(got_worker);
if !got_worker {
metrics::inc_local_fallback_reason(selection_reason_label(reason));
rch_telemetry::remediation::record_admission(
rch_telemetry::remediation::AdmissionDecision::Local,
selection_reason_label(reason),
);
if matches!(reason, SelectionReason::SelectionError(_)) {
metrics::inc_reliability_error("selection", "selection_error");
}
}
}
fn toolchain_capability_mismatch(
toolchain: Option<&ToolchainInfo>,
capabilities: &WorkerCapabilities,
) -> Option<String> {
let toolchain = toolchain?;
if toolchain.date.is_some()
|| toolchain.full_version.trim().is_empty()
|| toolchain.full_version.trim().eq(toolchain.channel.as_str())
{
return None;
}
let worker_raw = capabilities.rustc_version.as_deref()?;
let channel = toolchain.channel.trim();
if is_floating_rust_channel(channel) {
return (!worker_rustc_matches_channel(worker_raw, channel)).then(|| {
format!(
"rustc_channel_mismatch:local={channel}:worker={}",
rustc_version_key(worker_raw).unwrap_or_else(|| worker_raw.to_string())
)
});
}
let local = rustc_version_key(&toolchain.full_version)?;
let worker = rustc_version_key(worker_raw)?;
(local != worker).then(|| format!("rustc_version_mismatch:local={local}:worker={worker}"))
}
fn required_rustup_component(request: &SelectionRequest) -> Option<&'static str> {
let command = request.command.as_deref()?;
let classification = classify_command(command);
if matches!(classification.kind, Some(CompilationKind::CargoClippy)) {
return Some("clippy");
}
if request.project.contains("::clean-overlay::") && direct_cargo_fmt_check(command) {
return Some("rustfmt");
}
None
}
fn direct_cargo_fmt_check(command: &str) -> bool {
let tokens = command
.split_whitespace()
.map(|token| token.trim_matches(['\'', '"']))
.collect::<Vec<_>>();
let Some(cargo_index) = tokens.iter().position(|token| {
*token == "cargo"
|| token
.rsplit_once('/')
.is_some_and(|(_, name)| name == "cargo")
}) else {
return false;
};
let mut subcommand_index = cargo_index + 1;
if tokens
.get(subcommand_index)
.is_some_and(|token| token.starts_with('+'))
{
subcommand_index += 1;
}
tokens.get(subcommand_index) == Some(&"fmt")
&& tokens[subcommand_index + 1..].contains(&"--check")
}
fn toolchain_from_command_text(command: &str) -> Option<String> {
let mut prev_is_cargo = false;
for raw in command.split_whitespace() {
let token = raw.trim_matches(['\'', '"']);
if let Some(value) = token.strip_prefix("RUSTUP_TOOLCHAIN=") {
let value = value.trim_matches(['\'', '"']);
if !value.is_empty() {
return Some(value.to_string());
}
}
if prev_is_cargo
&& let Some(value) = token.strip_prefix('+')
&& !value.is_empty()
{
return Some(value.to_string());
}
prev_is_cargo = token == "cargo"
|| token
.rsplit_once('/')
.is_some_and(|(_, name)| name == "cargo");
}
None
}
fn required_component_toolchain(request: &SelectionRequest) -> Option<String> {
request
.command
.as_deref()
.and_then(toolchain_from_command_text)
.or_else(|| {
request
.toolchain
.as_ref()
.map(ToolchainInfo::rustup_toolchain)
})
}
fn required_tool_capability_mismatch(
request: &SelectionRequest,
capabilities: &WorkerCapabilities,
) -> Option<String> {
request.required_tools.iter().find_map(|tool| {
(!capabilities.tools_present.iter().any(|t| t == tool)).then(|| {
if capabilities.tools_absent.iter().any(|t| t == tool) {
format!("capability_missing:tool:{tool}:probe_failed")
} else {
format!("capability_missing:tool:{tool}:not_declared")
}
})
})
}
fn rustup_component_capability_mismatch(
request: &SelectionRequest,
capabilities: &WorkerCapabilities,
) -> Option<String> {
let component = required_rustup_component(request)?;
let Some(toolchain) = required_component_toolchain(request) else {
return Some(format!(
"capability_missing:rustup_component:<unknown>:{component}"
));
};
(!capabilities.has_rustup_component(&toolchain, component))
.then(|| format!("capability_missing:rustup_component:{toolchain}:{component}"))
}
fn is_floating_rust_channel(channel: &str) -> bool {
matches!(channel, "stable" | "beta" | "nightly")
}
fn worker_rustc_matches_channel(worker_raw: &str, channel: &str) -> bool {
let Some(worker) = rustc_version_key(worker_raw) else {
return false;
};
match channel {
"nightly" => worker.contains("-nightly"),
"beta" => worker.contains("-beta"),
"stable" => !worker.contains("-nightly") && !worker.contains("-beta"),
_ => false,
}
}
fn rustc_version_key(value: &str) -> Option<String> {
let mut parts = value.split_whitespace();
let first = parts.next()?;
let version = if first.eq("rustc") {
parts.next()?
} else {
first
};
let version = version.trim();
(!version.is_empty()).then(|| version.to_string())
}
async fn probe_worker_toolchain(
worker: &WorkerState,
toolchain_name: &str,
ssh_pool: Option<&Arc<rch_common::SshPool>>,
) -> Result<(), String> {
let worker_config = worker.config.read().await.clone();
let ssh_options = SshOptions {
connect_timeout: TOOLCHAIN_PREFLIGHT_CONNECT_TIMEOUT,
command_timeout: TOOLCHAIN_PREFLIGHT_COMMAND_TIMEOUT,
control_master: false,
..Default::default()
};
let escaped_toolchain = shell_escape::escape(std::borrow::Cow::from(toolchain_name));
let command = format!(
"rustup run {escaped_toolchain} rustc --version >/dev/null && rustup run {escaped_toolchain} cargo --version >/dev/null"
);
let result = if let Some(pool) = ssh_pool {
pool.run_with_timeout(
&worker_config,
&command,
TOOLCHAIN_PREFLIGHT_COMMAND_TIMEOUT,
)
.await
.map_err(|err| format!("toolchain_preflight_connect_failed:{err}"))
} else {
let mut client = SshClient::new(worker_config, ssh_options);
client
.connect()
.await
.map_err(|err| format!("toolchain_preflight_connect_failed:{err}"))?;
let result = client.execute(&command).await;
let _ = client.disconnect().await;
result.map_err(|err| format!("toolchain_preflight_command_error:{err}"))
};
match result {
Ok(output) if output.success() => Ok(()),
Ok(output) => Err(format!(
"toolchain_preflight_command_failed:{}:{}",
output.exit_code,
short_preflight_error(&output.stderr)
)),
Err(reason) => Err(reason),
}
}
fn short_preflight_error(stderr: &str) -> String {
let compact = stderr.trim().replace(['\n', '\r'], " ");
if compact.is_empty() {
return "stderr_empty".to_string();
}
const LIMIT: usize = 240;
if compact.len() > LIMIT {
let truncated: String = compact.chars().take(LIMIT).collect();
format!("{truncated}...")
} else {
compact
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::process_triage::{RemediationPipeline, RemediationPipelineConfig};
use crate::reliability::{
ReliabilityAggregator, ReliabilityConfig, SignalWeights, WorkerHealthState,
};
use crate::workers::WorkerState;
use rch_common::WorkerStatus;
use rch_common::e2e::process_triage::{
PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION, ProcessClassification, ProcessDescriptor,
ProcessTriageActionClass, ProcessTriageActionRequest, ProcessTriageContract,
ProcessTriageRequest, ProcessTriageResponseStatus, ProcessTriageTrigger,
};
use rch_common::test_guard;
use rch_common::{
CommandPriority, RequiredRuntime, SelectionWeightConfig, ToolchainInfo, WorkerCapabilities,
WorkerConfig, WorkerId,
};
use sha2::{Digest, Sha256};
use std::sync::Arc;
fn assert_golden_json(
actual: serde_json::Value,
expected: &serde_json::Value,
fixture_path: &str,
) {
if &actual == expected {
return;
}
let actual_json = serde_json::to_string_pretty(&actual).expect("actual JSON renders");
let expected_json = serde_json::to_string_pretty(expected).expect("expected JSON renders");
let expected_hash = sha256_hex(&expected_json);
let actual_hash = sha256_hex(&actual_json);
assert_eq!(
&actual, expected,
"golden mismatch in {fixture_path}\n\
expected_sha256={expected_hash}\n\
actual_sha256={actual_hash}\n\
bless path: review the semantic diff, update the fixture expected_* field, \
and record both hashes in the Beads closeout.\n\
expected:\n{expected_json}\n\
actual:\n{actual_json}",
);
}
fn sha256_hex(input: &str) -> String {
let digest = Sha256::digest(input.as_bytes());
let mut output = String::with_capacity(digest.len() * 2);
for byte in digest {
use std::fmt::Write as _;
write!(&mut output, "{byte:02x}").expect("write to string");
}
output
}
fn make_worker(id: &str, total_slots: u32, speed: f64) -> WorkerState {
make_worker_with_os(id, total_slots, speed, None)
}
fn make_worker_with_os(
id: &str,
total_slots: u32,
speed: f64,
os: Option<&str>,
) -> WorkerState {
let config = WorkerConfig {
id: WorkerId::new(id),
host: "localhost".to_string(),
user: "user".to_string(),
identity_file: "~/.ssh/id_rsa".to_string(),
total_slots,
priority: 100,
tags: os.map(rch_common::os_tag).into_iter().collect(),
tools: Vec::new(),
};
let state = WorkerState::new(config);
state.set_speed_score(speed);
state
}
async fn prepare_fixture_worker(
worker: &WorkerState,
rustc_version: Option<&str>,
pressure_state: PressureState,
pressure_reason_code: &str,
) {
worker
.set_capabilities(WorkerCapabilities {
rustc_version: rustc_version.map(str::to_string),
num_cpus: Some(8),
disk_free_gb: Some(100.0),
disk_total_gb: Some(200.0),
projects_root_ok: Some(true),
..Default::default()
})
.await;
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: pressure_state,
confidence: crate::disk_pressure::PressureConfidence::High,
reason_code: pressure_reason_code.to_string(),
policy_rule: "ft_4tp7g_golden_fixture".to_string(),
disk_free_gb: Some(100.0),
disk_total_gb: Some(200.0),
disk_free_ratio: Some(0.5),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: None,
memory_pressure: None,
telemetry_age_secs: Some(0),
telemetry_fresh: true,
evaluated_at_unix_ms: 0,
})
.await;
}
fn process_only_weights() -> SignalWeights {
SignalWeights {
circuit: 0.0,
convergence: 0.0,
pressure: 0.0,
process: 1.0,
cancellation: 0.0,
}
}
fn make_hard_terminate_request(
worker_id: &str,
correlation_id: &str,
pid: u32,
) -> ProcessTriageRequest {
ProcessTriageRequest {
schema_version: PROCESS_TRIAGE_CONTRACT_SCHEMA_VERSION.to_string(),
correlation_id: correlation_id.to_string(),
worker_id: worker_id.to_string(),
observed_at_unix_ms: chrono::Utc::now().timestamp_millis(),
trigger: ProcessTriageTrigger::BuildTimeout,
detector_confidence_percent: 99,
retry_attempt: 0,
candidate_processes: vec![ProcessDescriptor {
pid,
ppid: Some(1),
owner: "ubuntu".to_string(),
command: "cargo build --workspace".to_string(),
classification: ProcessClassification::BuildRelated,
cpu_percent_milli: 98_000,
rss_mb: 2048,
runtime_secs: 240,
}],
requested_actions: vec![ProcessTriageActionRequest {
action_class: ProcessTriageActionClass::HardTerminate,
pid,
reason_code: "stuck_hard".to_string(),
signal: Some("KILL".to_string()),
}],
}
}
async fn seeded_remediation_pipeline(
worker_id: &str,
pids: &[u32],
) -> Arc<RemediationPipeline> {
let mut contract = ProcessTriageContract::default();
contract
.safe_action_policy
.allow_action_classes
.push(ProcessTriageActionClass::HardTerminate);
contract
.safe_action_policy
.deny_action_classes
.retain(|class| *class != ProcessTriageActionClass::HardTerminate);
let pipeline = Arc::new(RemediationPipeline::new(
contract,
crate::events::EventBus::new(64),
RemediationPipelineConfig {
dry_run: true,
worker_cooldown: Duration::from_millis(0),
..RemediationPipelineConfig::default()
},
));
for (idx, pid) in pids.iter().enumerate() {
let request = make_hard_terminate_request(worker_id, &format!("corr-hard-{idx}"), *pid);
let response = pipeline.execute(&request).await;
assert_eq!(response.status, ProcessTriageResponseStatus::Applied);
}
pipeline
}
#[test]
fn test_selection_score() {
let _guard = test_guard!();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let worker = make_worker("test", 16, 80.0);
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let score = compute_score(&worker, &request, &weights, 0.5, CircuitState::Closed).await;
assert!(score > 0.0);
assert!(score <= 1.5);
});
}
#[tokio::test]
async fn test_disk_weight_balanced_prefers_headroom_and_audits_actual_score() {
let pool = WorkerPool::new();
for (id, free) in [("fuller", 90.0), ("emptier", 200.0)] {
let worker = make_worker(id, 8, 80.0);
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(free),
disk_total_gb: Some(400.0),
..Default::default()
})
.await;
assert_eq!(worker.available_slots().await, 8);
pool.add_worker_state(worker).await;
}
let request = SelectionRequest {
job_mode: false,
project: "disk-ranking".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let mut config = SelectionConfig::default();
config.weights.disk = 0.0;
let no_disk = WorkerSelector::with_config(config.clone(), CircuitBreakerConfig::default());
let candidates: Vec<_> = pool
.all_workers()
.await
.into_iter()
.map(|worker| (worker, CircuitState::Closed))
.collect();
let baseline = no_disk
.build_score_breakdowns(&candidates, &request, CacheUse::Build, None)
.await;
assert!((baseline[0].total_score - baseline[1].total_score).abs() < f64::EPSILON);
config.weights.disk = 0.2;
let selector = WorkerSelector::with_config(config, CircuitBreakerConfig::default());
let selected = selector.select(&pool, &request).await.worker.unwrap();
assert_eq!(selected.config.read().await.id.as_str(), "emptier");
let audit = selector
.build_score_breakdowns(&candidates, &request, CacheUse::Build, Some("emptier"))
.await;
for entry in &audit {
let before = baseline
.iter()
.find(|value| value.worker_id == entry.worker_id)
.unwrap();
assert!(
(entry.total_score - before.total_score - 0.2 * entry.disk_headroom).abs() < 1e-12
);
}
let winner = audit.iter().find(|entry| entry.selected).unwrap();
let loser = audit.iter().find(|entry| !entry.selected).unwrap();
assert!(winner.disk_headroom > loser.disk_headroom);
assert!(winner.total_score > loser.total_score);
}
#[tokio::test]
async fn test_disk_weight_preserves_ample_legacy_priority_order_and_supports_disk_only() {
let a = make_worker("fast", 8, 100.0);
let b = make_worker("priority", 8, 90.0);
for worker in [&a, &b] {
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(200.0),
disk_total_gb: Some(400.0),
..Default::default()
})
.await;
}
let request = SelectionRequest {
job_mode: false,
project: "disk-legacy".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights {
slots: 0.0,
speed: 1.0,
locality: 0.0,
priority: 0.1,
disk: 0.2,
half_open_penalty: 0.5,
};
let score_a = compute_score(&a, &request, &weights, 0.0, CircuitState::Closed).await;
let score_b = compute_score(&b, &request, &weights, 1.0, CircuitState::Closed).await;
assert!((score_a - 1.2).abs() < 1e-12);
assert!((score_b - 1.19).abs() < 1e-12);
assert!(score_a > score_b);
assert!(
(compute_score(&a, &request, &weights, 0.0, CircuitState::HalfOpen).await - 0.7).abs()
< 1e-12
);
b.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(90.0),
disk_total_gb: Some(400.0),
..Default::default()
})
.await;
let disk_only = SelectionWeights {
speed: 0.0,
priority: 0.0,
disk: 1.0,
..weights
};
assert!(
compute_score(&a, &request, &disk_only, 0.0, CircuitState::Closed).await
> compute_score(&b, &request, &disk_only, 0.0, CircuitState::Closed).await
);
}
#[test]
fn test_selection_score_zero_slots_safe() {
let _guard = test_guard!();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let worker = make_worker("zero", 0, 80.0);
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let score = compute_score(&worker, &request, &weights, 0.5, CircuitState::Closed).await;
assert!(score.is_finite());
});
}
#[tokio::test]
async fn test_select_worker_ignores_unhealthy() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_worker("healthy", 8, 50.0).config.read().await.clone())
.await;
pool.add_worker(
make_worker("unreachable", 16, 90.0)
.config
.read()
.await
.clone(),
)
.await;
pool.set_status(&WorkerId::new("unreachable"), WorkerStatus::Unreachable)
.await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let selected = select_worker(&pool, &request, &weights).await;
let selected = selected.expect("Expected a healthy worker to be selected");
assert_eq!(selected.config.read().await.id.as_str(), "healthy");
}
#[tokio::test]
async fn test_selection_metrics_capture_local_fallback_reason() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "metrics-fallback-project".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: Some(123),
hook_pid: Some(4321),
required_tools: Vec::new(),
};
let decision_before = crate::metrics::RELIABILITY_DECISIONS_TOTAL
.with_label_values(&["selection", "failure"])
.get();
let fallback_before = crate::metrics::LOCAL_FALLBACK_REASON_TOTAL
.with_label_values(&["no_workers_configured"])
.get();
let result = selector.select(&pool, &request).await;
assert_eq!(result.reason, SelectionReason::NoWorkersConfigured);
let decision_after = crate::metrics::RELIABILITY_DECISIONS_TOTAL
.with_label_values(&["selection", "failure"])
.get();
let fallback_after = crate::metrics::LOCAL_FALLBACK_REASON_TOTAL
.with_label_values(&["no_workers_configured"])
.get();
assert!(decision_after > decision_before);
assert!(fallback_after > fallback_before);
}
#[tokio::test]
async fn test_select_worker_respects_slot_availability() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("full", 4, 70.0).config.read().await.clone())
.await;
pool.add_worker(
make_worker("available", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
let full_worker = pool.get(&WorkerId::new("full")).await.unwrap();
assert!(full_worker.reserve_slots(4).await);
assert_eq!(full_worker.available_slots().await, 0);
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let selected = select_worker(&pool, &request, &weights).await;
let selected = selected.expect("Expected a worker with available slots");
assert_eq!(selected.config.read().await.id.as_str(), "available");
}
#[tokio::test]
async fn test_select_worker_ignores_open_circuit() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("closed", 8, 50.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("open", 16, 90.0).config.read().await.clone())
.await;
let open_worker = pool.get(&WorkerId::new("open")).await.unwrap();
open_worker.open_circuit().await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result.worker.expect("Expected a worker to be selected");
assert_eq!(result.reason, SelectionReason::Success);
assert_eq!(selected.config.read().await.id.as_str(), "closed");
}
#[tokio::test]
async fn test_select_worker_returns_all_circuits_open() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("open1", 8, 50.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("open2", 16, 90.0).config.read().await.clone())
.await;
let worker1 = pool.get(&WorkerId::new("open1")).await.unwrap();
let worker2 = pool.get(&WorkerId::new("open2")).await.unwrap();
worker1.open_circuit().await;
worker2.open_circuit().await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllCircuitsOpen);
}
#[tokio::test]
async fn preview_selection_spends_no_half_open_probe_slot() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("half_open", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
let worker = pool.get(&WorkerId::new("half_open")).await.unwrap();
worker.open_circuit().await;
worker.half_open_circuit().await;
let circuit = CircuitBreakerConfig {
half_open_max_probes: 1,
..Default::default()
};
let selector = WorkerSelector::with_config(SelectionConfig::default(), circuit.clone());
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let none = HashSet::new();
for _ in 0..2 {
let preview = selector
.preview_with_exclusions(&pool, &request, &none)
.await;
assert!(preview.worker.is_some(), "{:?}", preview.reason);
}
assert!(
worker.can_probe(&circuit).await,
"a preview consumed the probe slot"
);
let real = selector
.select_with_exclusions(&pool, &request, &none)
.await;
assert!(real.worker.is_some());
assert!(
!worker.can_probe(&circuit).await,
"a real selection takes the probe"
);
worker.close_circuit().await;
worker.open_circuit().await;
worker.half_open_circuit().await;
let mut config = SelectionConfig::default();
config.affinity.enabled = true;
config.affinity.pin_minutes = 60;
let pinned = WorkerSelector::with_config(config, circuit.clone());
pinned.record_success("half_open", "myproject").await;
let preview = pinned.preview_with_exclusions(&pool, &request, &none).await;
assert_eq!(preview.reason, SelectionReason::AffinityPinned);
assert!(
worker.can_probe(&circuit).await,
"a pinned preview consumed the probe slot"
);
}
#[tokio::test]
async fn test_select_worker_allows_half_open_with_probe_budget() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("half_open", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
let worker = pool.get(&WorkerId::new("half_open")).await.unwrap();
worker.open_circuit().await;
worker.half_open_circuit().await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig {
half_open_max_probes: 1,
..Default::default()
};
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result
.worker
.expect("Expected half-open worker to be selected");
assert_eq!(result.reason, SelectionReason::Success);
assert_eq!(selected.config.read().await.id.as_str(), "half_open");
}
#[tokio::test]
async fn test_select_worker_excludes_half_open_at_probe_limit() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("half_open", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("closed", 4, 40.0).config.read().await.clone())
.await;
let half_open_worker = pool.get(&WorkerId::new("half_open")).await.unwrap();
half_open_worker.open_circuit().await;
half_open_worker.half_open_circuit().await;
let config = CircuitBreakerConfig {
half_open_max_probes: 1,
..Default::default()
};
half_open_worker.start_probe(&config).await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result
.worker
.expect("Expected closed worker to be selected");
assert_eq!(result.reason, SelectionReason::Success);
assert_eq!(selected.config.read().await.id.as_str(), "closed");
}
#[tokio::test]
async fn test_select_worker_prefers_closed_over_half_open() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("half_open", 16, 90.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("closed", 8, 50.0).config.read().await.clone())
.await;
let half_open_worker = pool.get(&WorkerId::new("half_open")).await.unwrap();
half_open_worker.open_circuit().await;
half_open_worker.half_open_circuit().await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result.worker.expect("Expected a worker to be selected");
assert_eq!(result.reason, SelectionReason::Success);
assert_eq!(selected.config.read().await.id.as_str(), "closed");
}
#[tokio::test]
async fn test_select_worker_prefers_preferred_list() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("preferred", 8, 60.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("other", 8, 90.0).config.read().await.clone())
.await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("preferred")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result.worker.expect("Expected a worker to be selected");
assert_eq!(selected.config.read().await.id.as_str(), "preferred");
}
#[tokio::test]
async fn test_select_worker_refuses_when_requested_worker_unavailable() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("available", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("missing")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoMatchingWorkers);
}
#[tokio::test]
async fn test_select_worker_prefers_higher_priority() {
let pool = WorkerPool::new();
let high = make_worker("high", 8, 70.0);
{
let mut config = high.config.write().await;
config.priority = 200;
}
let low = make_worker("low", 8, 50.0);
{
let mut config = low.config.write().await;
config.priority = 50;
}
pool.add_worker(high.config.read().await.clone()).await;
pool.add_worker(low.config.read().await.clone()).await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let weights = SelectionWeights::default();
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result.worker.expect("Expected a worker to be selected");
assert_eq!(selected.config.read().await.id.as_str(), "high");
}
#[tokio::test]
async fn test_half_open_penalty_applied() {
let weights = SelectionWeights {
slots: 0.0,
speed: 1.0,
locality: 0.0,
priority: 0.0,
disk: 0.0,
half_open_penalty: 0.5,
};
let pool = WorkerPool::new();
pool.add_worker(
make_worker("half_open", 16, 80.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("closed", 8, 50.0).config.read().await.clone())
.await;
let half_open_worker = pool.get(&WorkerId::new("half_open")).await.unwrap();
half_open_worker.open_circuit().await;
half_open_worker.half_open_circuit().await;
let request = SelectionRequest {
job_mode: false,
project: "myproject".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let config = CircuitBreakerConfig::default();
let result = select_worker_with_config(&pool, &request, &weights, &config).await;
let selected = result.worker.expect("Expected a worker to be selected");
assert_eq!(selected.config.read().await.id.as_str(), "closed");
}
#[test]
fn test_cache_tracker_record_and_warmth() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_build("worker1", "project-a", CacheUse::Build);
let warmth = tracker.estimate_warmth("worker1", "project-a", CacheUse::Build);
assert!(warmth > 0.9, "Expected warmth > 0.9, got {}", warmth);
let unknown = tracker.estimate_warmth("worker1", "project-b", CacheUse::Build);
assert_eq!(unknown, 0.0);
let unknown = tracker.estimate_warmth("worker2", "project-a", CacheUse::Build);
assert_eq!(unknown, 0.0);
}
#[test]
fn test_cache_tracker_has_recent_build() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_build("worker1", "project-a", CacheUse::Build);
assert!(tracker.has_recent_build(
"worker1",
"project-a",
CacheUse::Build,
Duration::from_secs(60)
));
assert!(!tracker.has_recent_build(
"worker1",
"project-b",
CacheUse::Build,
Duration::from_secs(60)
));
}
#[test]
fn test_cache_tracker_has_recent_build_fallback_for_test() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_build("worker1", "project-a", CacheUse::Build);
assert!(tracker.has_recent_build(
"worker1",
"project-a",
CacheUse::Test,
Duration::from_secs(60)
));
}
#[test]
fn test_selection_history_record_and_count() {
let _guard = test_guard!();
let mut history = SelectionHistory::new();
history.record_selection("worker1");
history.record_selection("worker1");
history.record_selection("worker2");
let count1 = history.recent_selections("worker1", Duration::from_secs(60));
assert_eq!(count1, 2);
let count2 = history.recent_selections("worker2", Duration::from_secs(60));
assert_eq!(count2, 1);
let count3 = history.recent_selections("worker3", Duration::from_secs(60));
assert_eq!(count3, 0);
}
#[test]
fn test_selection_history_prune() {
let _guard = test_guard!();
let mut history = SelectionHistory::new();
history.record_selection("worker1");
history.prune(Duration::ZERO);
let count = history.recent_selections("worker1", Duration::from_secs(60));
let _ = count; }
#[tokio::test]
async fn test_worker_selector_priority_strategy() {
let pool = WorkerPool::new();
let high_priority = make_worker("high-priority", 8, 50.0);
{
let mut config = high_priority.config.write().await;
config.priority = 200;
}
pool.add_worker(high_priority.config.read().await.clone())
.await;
let low_priority = make_worker("low-priority", 8, 90.0);
{
let mut config = low_priority.config.write().await;
config.priority = 50;
}
pool.add_worker(low_priority.config.read().await.clone())
.await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Priority,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "high-priority");
}
async fn os_gate_fixture(command: &str) -> (WorkerPool, WorkerSelector, SelectionRequest) {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("linux-worker", 8, 50.0))
.await;
pool.add_worker_state(make_worker_with_os(
"windows-worker",
8,
50.0,
Some("windows"),
))
.await;
let selector = WorkerSelector::with_config(
SelectionConfig::default(),
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: Some(command.to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
(pool, selector, request)
}
#[tokio::test]
async fn test_os_declared_worker_excluded_from_unqualified_command() {
let (pool, selector, request) = os_gate_fixture("cargo check --workspace").await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "linux-worker");
}
#[tokio::test]
async fn test_os_declared_worker_selected_for_matching_target() {
let (pool, selector, request) =
os_gate_fixture("cargo build --target x86_64-pc-windows-msvc").await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "windows-worker");
}
#[tokio::test]
async fn test_undeclared_worker_cannot_satisfy_os_requirement() {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("linux-worker", 8, 50.0))
.await;
let selector = WorkerSelector::with_config(
SelectionConfig::default(),
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: Some("cargo build --target x86_64-pc-windows-msvc".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"a worker with no declared OS must not satisfy an OS requirement"
);
}
#[test]
fn test_os_gate_admits_truth_table() {
assert!(os_gate_admits(None, None));
assert!(os_gate_admits(Some("windows"), Some("windows")));
assert!(!os_gate_admits(Some("windows"), Some("darwin")));
assert!(!os_gate_admits(Some("windows"), None));
assert!(!os_gate_admits(None, Some("windows")));
assert!(os_gate_admits(Some("Windows"), Some("windows")));
assert!(os_gate_admits(Some("windows"), Some("WINDOWS")));
}
#[tokio::test]
async fn test_diagnostics_report_os_gate_exclusion() {
let (pool, selector, request) = os_gate_fixture("cargo check --workspace").await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("linux worker should still be chosen");
assert_eq!(selected.config.read().await.id.as_str(), "linux-worker");
let diagnostics = selector
.build_selection_diagnostics(&pool, &request, &HashSet::new())
.await;
let windows = diagnostics
.workers
.iter()
.find(|w| w.worker_id.as_str() == "windows-worker")
.expect("windows worker present in diagnostics");
assert_eq!(
windows.final_decision,
WorkerSelectionDiagnosticDecision::Deny
);
assert!(
windows
.reason_codes
.iter()
.any(|code| code == "os.declared_mismatch"),
"expected os.declared_mismatch, got {:?}",
windows.reason_codes
);
}
#[tokio::test]
async fn test_windows_gnu_target_does_not_require_windows_host() {
let (pool, selector, request) =
os_gate_fixture("cargo build --target x86_64-pc-windows-gnu").await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "linux-worker");
}
#[tokio::test]
async fn test_worker_selector_priority_strategy_prefers_warm_worker_for_normal_commands() {
let pool = WorkerPool::new();
let warm = make_worker("warm-worker", 8, 50.0);
{
let mut config = warm.config.write().await;
config.priority = 100;
}
pool.add_worker_state(warm).await;
let cold = make_worker("cold-worker", 8, 95.0);
{
let mut config = cold.config.write().await;
config.priority = 100;
}
pool.add_worker_state(cold).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Priority,
affinity: rch_common::AffinityConfig {
enabled: false,
..Default::default()
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_success("warm-worker", "test-project").await;
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: Some("cargo check".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "warm-worker");
}
#[tokio::test]
async fn test_worker_selector_fastest_strategy() {
let pool = WorkerPool::new();
let high_priority = make_worker("high-priority", 8, 50.0);
{
let mut config = high_priority.config.write().await;
config.priority = 200;
}
pool.add_worker_state(high_priority).await;
let fastest = make_worker("fastest", 8, 90.0);
{
let mut config = fastest.config.write().await;
config.priority = 50;
}
pool.add_worker_state(fastest).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Fastest,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "fastest");
}
#[tokio::test]
async fn test_worker_selector_filters_low_success_rate() {
let pool = WorkerPool::new();
let fast_unhealthy = make_worker("fast-unhealthy", 8, 95.0);
fast_unhealthy.record_failure(None).await;
fast_unhealthy.record_failure(None).await;
fast_unhealthy.record_failure(None).await;
pool.add_worker_state(fast_unhealthy).await;
let slow_healthy = make_worker("slow-healthy", 8, 40.0);
slow_healthy.record_success().await;
pool.add_worker_state(slow_healthy).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Fastest,
min_success_rate: 0.8,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "slow-healthy");
}
#[tokio::test]
async fn test_worker_selector_reports_no_workers_passed_health_thresholds() {
let pool = WorkerPool::new();
let unhealthy = make_worker("unhealthy", 8, 95.0);
unhealthy.record_failure(None).await;
pool.add_worker_state(unhealthy).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Fastest,
min_success_rate: 0.8,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoWorkersPassedHealth);
let diagnostics = result
.diagnostics
.expect("failed selection should include per-worker diagnostics");
assert_eq!(diagnostics.required_runtime, RequiredRuntime::None);
assert_eq!(diagnostics.estimated_cores, 2);
assert_eq!(diagnostics.min_success_rate, 0.8);
assert_eq!(diagnostics.workers.len(), 1);
let worker = &diagnostics.workers[0];
assert_eq!(worker.worker_id.as_str(), "unhealthy");
assert_eq!(
worker.final_decision,
WorkerSelectionDiagnosticDecision::Deny
);
assert!(
worker
.reason_codes
.iter()
.any(|code| code == "health.below_min_success_rate")
);
assert!(
worker
.reason_codes
.iter()
.any(|code| code == "health.below_fallback_min_success_rate")
);
}
#[tokio::test]
async fn test_requested_worker_below_health_fallback_threshold_is_refused() {
let pool = WorkerPool::new();
let preferred = make_worker("preferred", 8, 95.0);
preferred.record_failure(None).await;
pool.add_worker_state(preferred).await;
let fallback = make_worker("fallback", 8, 40.0);
fallback.record_success().await;
pool.add_worker_state(fallback).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Fastest,
min_success_rate: 0.8,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("preferred")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoWorkersPassedHealth);
let diagnostics = result
.diagnostics
.expect("health refusal should include diagnostics");
let requested = diagnostics
.workers
.iter()
.find(|worker| worker.worker_id.as_str() == "preferred")
.expect("requested worker diagnostic");
assert_eq!(
requested.final_decision,
WorkerSelectionDiagnosticDecision::Deny
);
assert!(
requested
.reason_codes
.iter()
.any(|code| code == "health.below_fallback_min_success_rate")
);
}
fn tool_gate_request() -> SelectionRequest {
SelectionRequest {
project: "tool-gate".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
hook_pid: None,
job_mode: true,
required_tools: Vec::new(),
}
}
#[test]
fn required_tool_gate_needs_verified_evidence_not_a_declaration() {
let caps = WorkerCapabilities {
tools_present: vec!["clang".to_string()],
tools_absent: vec!["ld.lld".to_string()],
..Default::default()
};
let mut request = tool_gate_request();
assert!(required_tool_capability_mismatch(&request, &caps).is_none());
request.required_tools = vec!["clang".to_string()];
assert!(required_tool_capability_mismatch(&request, &caps).is_none());
request.required_tools = vec!["ld.lld".to_string()];
assert_eq!(
required_tool_capability_mismatch(&request, &caps).as_deref(),
Some("capability_missing:tool:ld.lld:probe_failed")
);
request.required_tools = vec!["clanggg".to_string()];
assert_eq!(
required_tool_capability_mismatch(&request, &caps).as_deref(),
Some("capability_missing:tool:clanggg:not_declared")
);
request.required_tools = vec!["clang".to_string(), "ld.lld".to_string()];
assert!(required_tool_capability_mismatch(&request, &caps).is_some());
request.required_tools = vec!["clang".to_string()];
assert!(
required_tool_capability_mismatch(&request, &WorkerCapabilities::default()).is_some()
);
}
#[test]
fn test_toolchain_capability_mismatch_for_floating_channels() {
let local = ToolchainInfo {
channel: "nightly".to_string(),
date: None,
full_version: "rustc 1.97.0-nightly (abcdef 2026-05-01)".to_string(),
};
let caps = WorkerCapabilities {
rustc_version: Some("rustc 1.95.0-nightly (abcdef 2026-03-01)".to_string()),
..Default::default()
};
assert!(toolchain_capability_mismatch(Some(&local), &caps).is_none());
let stable = ToolchainInfo {
channel: "stable".to_string(),
..local.clone()
};
assert!(toolchain_capability_mismatch(Some(&stable), &caps).is_some());
let dated = ToolchainInfo {
date: Some("2026-05-01".to_string()),
..local
};
assert!(toolchain_capability_mismatch(Some(&dated), &caps).is_none());
}
#[test]
fn test_toolchain_capability_mismatch_skips_channel_only_toolchain_file() {
let local = ToolchainInfo {
channel: "nightly".to_string(),
date: None,
full_version: "nightly".to_string(),
};
let caps = WorkerCapabilities {
rustc_version: Some("rustc 1.95.0-nightly (abcdef 2026-03-01)".to_string()),
..Default::default()
};
assert!(toolchain_capability_mismatch(Some(&local), &caps).is_none());
}
fn pinned_component_request(project: &str, command: &str) -> SelectionRequest {
SelectionRequest {
job_mode: false,
project: project.to_string(),
command: Some(command.to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: Some(ToolchainInfo {
channel: "nightly".to_string(),
date: Some("2026-07-05".to_string()),
full_version: "nightly-2026-07-05".to_string(),
}),
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
}
}
#[test]
fn rustup_component_requirement_is_command_specific_and_fail_closed() {
let clippy = pinned_component_request("asupersync", "cargo clippy --workspace");
assert_eq!(required_rustup_component(&clippy), Some("clippy"));
let fmt = pinned_component_request(
"asupersync::clean-overlay::0123456789abcdef",
"cargo fmt --check",
);
assert_eq!(required_rustup_component(&fmt), Some("rustfmt"));
let ordinary_fmt = pinned_component_request("asupersync", "cargo fmt --check");
assert_eq!(required_rustup_component(&ordinary_fmt), None);
let capabilities = WorkerCapabilities {
rustup_components: vec![
"nightly-2026-07-05-x86_64-unknown-linux-gnu:rustfmt".to_string(),
],
..Default::default()
};
assert!(rustup_component_capability_mismatch(&fmt, &capabilities).is_none());
assert_eq!(
rustup_component_capability_mismatch(&clippy, &capabilities).as_deref(),
Some("capability_missing:rustup_component:nightly-2026-07-05:clippy")
);
}
#[test]
fn component_gate_derives_toolchain_from_command_env_prefix() {
let mut request = pinned_component_request(
"frankensearch",
"env RCH_CARGO_WRAPPER_BYPASS=1 RUSTUP_TOOLCHAIN=nightly-2026-08-25 \
CARGO_TARGET_DIR=/data/projects/frankensearch/.rch-target cargo clippy \
--workspace --all-targets --locked -- -D warnings",
);
request.toolchain = None;
assert_eq!(
toolchain_from_command_text(request.command.as_deref().unwrap()).as_deref(),
Some("nightly-2026-08-25")
);
assert_eq!(
required_component_toolchain(&request).as_deref(),
Some("nightly-2026-08-25")
);
let with_clippy = WorkerCapabilities {
rustup_components: vec![
"nightly-2026-08-25-x86_64-unknown-linux-gnu:clippy".to_string(),
],
..Default::default()
};
assert!(
rustup_component_capability_mismatch(&request, &with_clippy).is_none(),
"a worker with the pinned clippy installed must be admissible"
);
let without_clippy = WorkerCapabilities::default();
assert_eq!(
rustup_component_capability_mismatch(&request, &without_clippy).as_deref(),
Some("capability_missing:rustup_component:nightly-2026-08-25:clippy"),
"the derived toolchain (never <unknown>) names the mismatch"
);
let mut pinned_both = pinned_component_request(
"frankensearch",
"env RUSTUP_TOOLCHAIN=nightly-2026-08-25 cargo clippy --workspace",
);
assert!(pinned_both.toolchain.is_some());
assert_eq!(
required_component_toolchain(&pinned_both).as_deref(),
Some("nightly-2026-08-25")
);
pinned_both.command = Some("cargo clippy --workspace".to_string());
assert_eq!(
required_component_toolchain(&pinned_both).as_deref(),
Some("nightly-2026-07-05")
);
assert_eq!(
toolchain_from_command_text("cargo +nightly-2026-08-25 clippy --workspace").as_deref(),
Some("nightly-2026-08-25")
);
assert_eq!(
toolchain_from_command_text("cargo build --features +weird"),
None
);
let mut unknown = pinned_component_request("proj", "cargo clippy --workspace");
unknown.toolchain = None;
assert_eq!(
rustup_component_capability_mismatch(&unknown, &with_clippy).as_deref(),
Some("capability_missing:rustup_component:<unknown>:clippy")
);
}
#[tokio::test]
async fn clippy_component_selection_refuses_worker_without_pinned_component() {
let pool = WorkerPool::new();
let worker = make_worker("missing-clippy", 8, 80.0);
worker
.set_capabilities(WorkerCapabilities {
rustc_version: Some("rustc 1.88.0-nightly".to_string()),
rustup_toolchains: vec!["nightly-2026-07-05-x86_64-unknown-linux-gnu".to_string()],
rustup_components: vec![
"nightly-2026-07-05-x86_64-unknown-linux-gnu:rustfmt".to_string(),
],
projects_root_ok: Some(true),
..Default::default()
})
.await;
pool.add_worker_state(worker).await;
let request = pinned_component_request("asupersync", "cargo clippy --workspace");
let result = WorkerSelector::default().select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers("missing_toolchain_component=1".to_string())
);
let diagnostics = result.diagnostics.expect("component refusal diagnostics");
assert!(
diagnostics.workers[0]
.reason_codes
.iter()
.any(|code| code == "toolchain.component_missing")
);
}
#[tokio::test]
async fn test_worker_selector_balanced_strategy() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("worker1", 8, 70.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("worker2", 8, 80.0).config.read().await.clone())
.await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_some());
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_command_priority_hint_influences_balanced_selection() {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("fast", 8, 90.0)).await;
pool.add_worker_state(make_worker("cached", 8, 60.0)).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_build("cached", "proj", false).await;
let base_request = SelectionRequest {
job_mode: false,
project: "proj".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let mut high = base_request.clone();
high.command_priority = CommandPriority::High;
let result = selector.select(&pool, &high).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "fast");
let mut low = base_request.clone();
low.command_priority = CommandPriority::Low;
let result = selector.select(&pool, &low).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "cached");
}
#[tokio::test]
async fn test_select_worker_balanced_health_weight_prefers_healthy() {
let pool = WorkerPool::new();
let unhealthy = make_worker("unhealthy", 8, 70.0);
unhealthy.record_failure(None).await;
pool.add_worker_state(unhealthy).await;
let healthy = make_worker("healthy", 8, 70.0);
healthy.record_success().await;
pool.add_worker_state(healthy).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
min_success_rate: 0.0,
weights: SelectionWeightConfig {
speedscore: 0.0,
slots: 0.0,
health: 1.0,
cache: 0.0,
network: 0.0,
priority: 0.0,
disk: 0.0,
half_open_penalty: 1.0,
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "healthy");
}
#[tokio::test]
async fn test_bug_repro_no_workers_with_runtime_when_busy() {
let pool = WorkerPool::new();
let worker = make_worker("busy-rust", 4, 80.0);
worker
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
..Default::default()
})
.await;
assert!(worker.reserve_slots(4).await);
pool.add_worker_state(worker).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::AllWorkersBusy,
"Expected AllWorkersBusy, got {:?}",
result.reason
);
}
#[tokio::test]
async fn test_pre_v3_microarch_worker_deprioritized_but_still_selectable() {
let pool = WorkerPool::new();
let pre_v3 = make_worker("ivy-bridge", 8, 80.0);
pre_v3
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
cpu_microarch_level: Some(2),
..Default::default()
})
.await;
pool.add_worker_state(pre_v3).await;
let v3 = make_worker("modern", 8, 80.0);
v3.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
cpu_microarch_level: Some(3),
..Default::default()
})
.await;
pool.add_worker_state(v3).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("a worker is selected");
assert_eq!(
selected.config.read().await.id.as_str(),
"modern",
"v3-capable worker must beat the equal pre-v3 worker"
);
let solo_pool = WorkerPool::new();
let solo = make_worker("ivy-bridge", 8, 80.0);
solo.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
cpu_microarch_level: Some(2),
..Default::default()
})
.await;
solo_pool.add_worker_state(solo).await;
let result = selector.select(&solo_pool, &request).await;
let selected = result.worker.expect("solo pre-v3 worker still selected");
assert_eq!(selected.config.read().await.id.as_str(), "ivy-bridge");
}
#[tokio::test]
async fn test_pinned_busy_worker_maps_to_all_workers_busy_for_queueing() {
let pool = WorkerPool::new();
let busy = make_worker("pinned-busy", 4, 80.0);
busy.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
..Default::default()
})
.await;
assert!(busy.reserve_slots(4).await);
pool.add_worker_state(busy).await;
let free = make_worker("free-other", 8, 90.0);
free.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.75.0".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(free).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![rch_common::WorkerId::new("pinned-busy")],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"pin must not select another worker"
);
assert_eq!(
result.reason,
SelectionReason::AllWorkersBusy,
"pinned-busy must be queueable (AllWorkersBusy), got {:?}",
result.reason
);
let worker = pool
.get(&rch_common::WorkerId::new("pinned-busy"))
.await
.unwrap();
worker.release_slots(4).await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("freed pinned worker is selected");
assert_eq!(selected.config.read().await.id.as_str(), "pinned-busy");
}
#[tokio::test]
async fn test_nix_runtime_gating() {
let pool = WorkerPool::new();
let rust_only = make_worker("rust-only", 8, 5.0);
rust_only
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(rust_only).await;
let selector = WorkerSelector::default();
let nix_request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: Some("nix build .#foo".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Nix,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &nix_request).await;
assert!(
result.worker.is_none(),
"nix build must not route to a nix-less worker"
);
assert!(
matches!(result.reason, SelectionReason::NoWorkersWithRuntime(_)),
"Expected NoWorkersWithRuntime, got {:?}",
result.reason
);
let nix_worker = make_worker("nix-1", 8, 5.0);
nix_worker
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
nix_version: Some("nix (Nix) 2.24.9".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(nix_worker).await;
let result = selector.select(&pool, &nix_request).await;
let selected = result
.worker
.expect("a nix-capable worker should be selected");
assert_eq!(selected.config.read().await.id.as_str(), "nix-1");
}
#[tokio::test]
async fn test_active_project_exclusion_preserves_runtime_reason() {
let pool = WorkerPool::new();
let active = make_worker("active-rust", 4, 80.0);
active
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.97.0-nightly".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(active).await;
let non_rust = make_worker("non-rust", 4, 70.0);
pool.add_worker_state(non_rust).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "frankenterm".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let mut excluded_worker_ids = HashSet::new();
excluded_worker_ids.insert("active-rust".to_string());
let result = selector
.select_with_exclusions(&pool, &request, &excluded_worker_ids)
.await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers("active_project_exclusion=1".to_string())
);
let diagnostics = result
.diagnostics
.expect("active-project rejection should include per-worker diagnostics");
assert_eq!(diagnostics.required_runtime, RequiredRuntime::Rust);
assert_eq!(diagnostics.active_project_exclusion_count, 1);
assert_eq!(diagnostics.workers.len(), 2);
let active = diagnostics
.workers
.iter()
.find(|worker| worker.worker_id.as_str() == "active-rust")
.expect("active worker diagnostic");
assert!(active.runtime_available);
assert!(active.active_project_excluded);
assert_eq!(
active.final_decision,
WorkerSelectionDiagnosticDecision::Deny
);
assert!(
active
.reason_codes
.iter()
.any(|code| code == "active_project_exclusion")
);
let non_rust = diagnostics
.workers
.iter()
.find(|worker| worker.worker_id.as_str() == "non-rust")
.expect("non-rust worker diagnostic");
assert!(!non_rust.runtime_available);
assert!(
non_rust
.reason_codes
.iter()
.any(|code| code == "runtime.unavailable")
);
}
#[tokio::test]
async fn test_job_mode_queues_when_active_project_exclusion_excludes_all() {
let pool = WorkerPool::new();
let busy_a = make_worker("busy-a", 4, 80.0);
busy_a
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.97.0-nightly".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(busy_a).await;
let busy_b = make_worker("busy-b", 4, 80.0);
busy_b
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.97.0-nightly".to_string()),
..Default::default()
})
.await;
pool.add_worker_state(busy_b).await;
let selector = WorkerSelector::default();
let mut excluded_worker_ids = HashSet::new();
excluded_worker_ids.insert("busy-a".to_string());
excluded_worker_ids.insert("busy-b".to_string());
let job_request = SelectionRequest {
job_mode: true,
project: "sharded".to_string(),
command: Some("./run_shards.sh".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector
.select_with_exclusions(&pool, &job_request, &excluded_worker_ids)
.await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllWorkersBusy);
let compile_request = SelectionRequest {
job_mode: false,
project: "sharded".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector
.select_with_exclusions(&pool, &compile_request, &excluded_worker_ids)
.await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers("active_project_exclusion=2".to_string())
);
}
#[tokio::test]
async fn golden_selector_diagnostics_for_rejected_worker_mix() {
let _guard = test_guard!();
const FIXTURE_PATH: &str =
"../../tests/goldens/ft_4tp7g/selector_diagnostics_rejected_worker_mix.json";
let fixture: serde_json::Value = serde_json::from_str(include_str!(
"../../tests/goldens/ft_4tp7g/selector_diagnostics_rejected_worker_mix.json"
))
.expect("selector diagnostics golden fixture parses");
assert_eq!(
fixture["schema_version"],
"rch.golden.selector_diagnostics.v1"
);
let pool = WorkerPool::new();
let active = make_worker("active-rust", 4, 80.0);
prepare_fixture_worker(
&active,
Some("rustc 1.97.0-nightly (fixture 2026-05-16)"),
PressureState::Healthy,
"pressure.ok",
)
.await;
pool.add_worker_state(active).await;
let busy = make_worker("busy-rust", 1, 75.0);
prepare_fixture_worker(
&busy,
Some("rustc 1.97.0-nightly (fixture 2026-05-16)"),
PressureState::Healthy,
"pressure.ok",
)
.await;
pool.add_worker_state(busy).await;
let circuit_open = make_worker("circuit-open", 4, 70.0);
prepare_fixture_worker(
&circuit_open,
Some("rustc 1.97.0-nightly (fixture 2026-05-16)"),
PressureState::Healthy,
"pressure.ok",
)
.await;
circuit_open.open_circuit().await;
pool.add_worker_state(circuit_open).await;
let no_rust = make_worker("no-rust", 4, 65.0);
prepare_fixture_worker(&no_rust, None, PressureState::Healthy, "pressure.ok").await;
pool.add_worker_state(no_rust).await;
let pressure_critical = make_worker("pressure-critical", 4, 60.0);
prepare_fixture_worker(
&pressure_critical,
Some("rustc 1.97.0-nightly (fixture 2026-05-16)"),
PressureState::Critical,
"pressure.disk_critical",
)
.await;
pool.add_worker_state(pressure_critical).await;
let toolchain_mismatch = make_worker("toolchain-mismatch", 4, 55.0);
prepare_fixture_worker(
&toolchain_mismatch,
Some("rustc 1.95.0 (fixture 2026-03-01)"),
PressureState::Healthy,
"pressure.ok",
)
.await;
pool.add_worker_state(toolchain_mismatch).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "frankenterm".to_string(),
command: Some("cargo test -p rchd --lib".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: Some(ToolchainInfo {
channel: "nightly".to_string(),
date: None,
full_version: "rustc 1.97.0-nightly (fixture 2026-05-16)".to_string(),
}),
required_runtime: RequiredRuntime::Rust,
classification_duration_us: Some(42),
hook_pid: Some(4242),
required_tools: Vec::new(),
};
let mut excluded_worker_ids = std::collections::HashSet::new();
excluded_worker_ids.insert("active-rust".to_string());
let mut diagnostics = selector
.build_selection_diagnostics(&pool, &request, &excluded_worker_ids)
.await;
diagnostics
.workers
.sort_by(|left, right| left.worker_id.as_str().cmp(right.worker_id.as_str()));
assert_golden_json(
serde_json::to_value(&diagnostics).expect("diagnostics serialize"),
&fixture["expected_diagnostics"],
FIXTURE_PATH,
);
}
#[tokio::test]
async fn test_topology_preflight_prefers_healthy_worker() {
let pool = WorkerPool::new();
let missing_alias = make_worker("missing-alias", 8, 95.0);
missing_alias
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(false),
projects_root_issue: Some("alias_missing".to_string()),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
pool.add_worker_state(missing_alias).await;
let healthy = make_worker("healthy-topology", 8, 80.0);
healthy
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
pool.add_worker_state(healthy).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "topology-project".to_string(),
command: Some("cargo test --no-run".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("expected healthy topology worker");
assert_eq!(selected.config.read().await.id.as_str(), "healthy-topology");
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_topology_preflight_rejects_wrong_target_only_pool() {
let pool = WorkerPool::new();
let wrong_target = make_worker("wrong-target", 8, 95.0);
wrong_target
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(false),
projects_root_issue: Some("alias_wrong_target:/tmp/other".to_string()),
projects_root_checked_at_unix_ms: Some(1_700_000_000_123),
..Default::default()
})
.await;
pool.add_worker_state(wrong_target).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "topology-project".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllWorkersFailedPreflight);
}
#[tokio::test]
async fn test_pressure_and_busy_pool_reports_concrete_admission_blockers() {
let pool = WorkerPool::new();
let critical = make_worker("critical-pressure", 8, 80.0);
critical
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
critical
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: crate::disk_pressure::PressureState::Critical,
confidence: crate::disk_pressure::PressureConfidence::High,
reason_code: "disk_ratio_below_critical".to_string(),
policy_rule: "disk_free_ratio<=critical_free_ratio".to_string(),
disk_free_gb: Some(38.0),
disk_total_gb: Some(774.0),
disk_free_ratio: Some(0.049),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: Some(0.0),
memory_pressure: Some(15.0),
telemetry_age_secs: Some(8),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
pool.add_worker_state(critical).await;
let busy = make_worker("busy-rust", 8, 70.0);
busy.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
disk_free_gb: Some(90.0),
disk_total_gb: Some(774.0),
..Default::default()
})
.await;
assert!(busy.reserve_slots(8).await);
pool.add_worker_state(busy).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "pressure-plus-busy".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers(
"critical_pressure=1,insufficient_slots=1".to_string()
)
);
}
#[tokio::test]
async fn test_fleet_derated_below_estimate_degrades_instead_of_falling_local() {
let pool = WorkerPool::new();
for id in ["derated-a", "derated-b"] {
let worker = make_worker(id, 2, 70.0);
worker
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
disk_free_gb: Some(400.0),
disk_total_gb: Some(900.0),
..Default::default()
})
.await;
pool.add_worker_state(worker).await;
}
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "derated-fleet".to_string(),
command: Some("cargo build --release".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"a healthy idle worker below the core estimate must be admitted as a \
degraded candidate instead of silently falling back to local; reason={:?}",
result.reason
);
}
#[tokio::test]
async fn test_capacity_degraded_still_honours_critical_pressure() {
let pool = WorkerPool::new();
let small_critical = make_worker("small-critical", 1, 90.0);
small_critical
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
small_critical
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: crate::disk_pressure::PressureState::Critical,
confidence: crate::disk_pressure::PressureConfidence::High,
reason_code: "disk_free_below_critical_gb".to_string(),
policy_rule: "disk_free<=critical_free_gb".to_string(),
disk_free_gb: Some(2.0),
disk_total_gb: Some(240.0),
disk_free_ratio: Some(0.008),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: Some(0.0),
memory_pressure: Some(10.0),
telemetry_age_secs: Some(5),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
pool.add_worker_state(small_critical).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "degraded-must-respect-pressure".to_string(),
command: Some("cargo build --release".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"a critically-pressured worker must never be handed out, even as a \
last-resort degraded candidate; reason was {:?}",
result.reason
);
}
#[tokio::test]
async fn test_os_gate_does_not_hijack_unrelated_admission_failure() {
let pool = WorkerPool::new();
let windows = make_worker_with_os("wsurf", 8, 60.0, Some("windows"));
windows
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
disk_free_gb: Some(400.0),
disk_total_gb: Some(900.0),
..Default::default()
})
.await;
pool.add_worker_state(windows).await;
let busy = make_worker("busy-linux", 8, 70.0);
busy.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
disk_free_gb: Some(400.0),
disk_total_gb: Some(900.0),
..Default::default()
})
.await;
assert!(busy.reserve_slots(8).await);
pool.add_worker_state(busy).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "os-gate-hijack".to_string(),
command: Some("cargo build --release".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
if let SelectionReason::NoAdmissibleWorkers(detail) = &result.reason {
assert!(
!detail.contains("every candidate declares an `os`"),
"OS gate must not claim sole causality when a worker was filtered \
for an unrelated reason; detail={detail}"
);
}
assert_eq!(
result.reason,
SelectionReason::AllWorkersBusy,
"a busy Linux worker plus an OS-gated Windows worker is a busy pool"
);
}
#[tokio::test]
async fn test_os_gate_is_reported_alongside_other_admission_blockers() {
let pool = WorkerPool::new();
let windows = make_worker_with_os("wsurf", 8, 60.0, Some("windows"));
windows
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
pool.add_worker_state(windows).await;
let critical = make_worker("critical-pressure", 8, 80.0);
critical
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
critical
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: crate::disk_pressure::PressureState::Critical,
confidence: crate::disk_pressure::PressureConfidence::High,
reason_code: "disk_ratio_below_critical".to_string(),
policy_rule: "disk_free_ratio<=critical_free_ratio".to_string(),
disk_free_gb: Some(38.0),
disk_total_gb: Some(774.0),
disk_free_ratio: Some(0.049),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: Some(0.0),
memory_pressure: Some(15.0),
telemetry_age_secs: Some(8),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
pool.add_worker_state(critical).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "os-gate-plus-pressure".to_string(),
command: Some("cargo build --release".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers(
"critical_pressure=1,os_gate_excluded=1".to_string()
)
);
}
#[tokio::test]
async fn test_topology_preflight_flapping_recovery_requires_revalidation() {
let pool = WorkerPool::new();
let flapping = make_worker("flapping", 8, 70.0);
flapping
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(false),
projects_root_issue: Some("alias_missing".to_string()),
projects_root_checked_at_unix_ms: Some(1_700_000_000_200),
..Default::default()
})
.await;
pool.add_worker_state(flapping).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "topology-flap".to_string(),
command: Some("cargo check".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let first = selector.select(&pool, &request).await;
assert!(first.worker.is_none());
assert_eq!(first.reason, SelectionReason::AllWorkersFailedPreflight);
let state = pool.get(&WorkerId::new("flapping")).await.unwrap();
state
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_100_000),
..Default::default()
})
.await;
let second = selector.select(&pool, &request).await;
let selected = second
.worker
.expect("expected worker to recover after explicit revalidation");
assert_eq!(selected.config.read().await.id.as_str(), "flapping");
assert_eq!(second.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_toolchain_preflight_rejects_cached_broken_toolchain_only_pool() {
let pool = WorkerPool::new();
let broken = make_worker("broken-toolchain", 8, 95.0);
broken
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
disk_free_gb: Some(60.0),
disk_total_gb: Some(120.0),
..Default::default()
})
.await;
broken
.record_toolchain_preflight(
"nightly-2026-04-30".to_string(),
false,
Some("cargo binary not applicable".to_string()),
)
.await;
pool.add_worker_state(broken).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "toolchain-project".to_string(),
command: Some("cargo check".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: Some(ToolchainInfo {
channel: "nightly".to_string(),
date: Some("2026-04-30".to_string()),
full_version: "nightly-2026-04-30".to_string(),
}),
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllWorkersFailedPreflight);
}
#[tokio::test]
async fn test_toolchain_preflight_prefers_cached_healthy_worker() {
let pool = WorkerPool::new();
let broken_fast = make_worker("broken-fast", 8, 99.0);
broken_fast
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
disk_free_gb: Some(60.0),
disk_total_gb: Some(120.0),
..Default::default()
})
.await;
broken_fast
.record_toolchain_preflight(
"nightly-2026-04-30".to_string(),
false,
Some("rustup run failed".to_string()),
)
.await;
pool.add_worker_state(broken_fast).await;
let usable_slow = make_worker("usable-slow", 8, 40.0);
usable_slow
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
disk_free_gb: Some(60.0),
disk_total_gb: Some(120.0),
..Default::default()
})
.await;
usable_slow
.record_toolchain_preflight("nightly-2026-04-30".to_string(), true, None)
.await;
pool.add_worker_state(usable_slow).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "toolchain-project".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: Some(ToolchainInfo {
channel: "nightly".to_string(),
date: Some("2026-04-30".to_string()),
full_version: "nightly-2026-04-30".to_string(),
}),
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result
.worker
.expect("expected selector to skip broken toolchain worker");
assert_eq!(selected.config.read().await.id.as_str(), "usable-slow");
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_pressure_preflight_rejects_critical_worker_only_pool() {
let pool = WorkerPool::new();
let critical = make_worker("critical-pressure", 8, 80.0);
critical
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
..Default::default()
})
.await;
critical
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: crate::disk_pressure::PressureState::Critical,
confidence: crate::disk_pressure::PressureConfidence::High,
reason_code: "disk_free_below_critical_gb".to_string(),
policy_rule: "disk_free_gb<=critical_free_gb".to_string(),
disk_free_gb: Some(4.0),
disk_total_gb: Some(120.0),
disk_free_ratio: Some(0.033),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: Some(80.0),
memory_pressure: Some(55.0),
telemetry_age_secs: Some(8),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
pool.add_worker_state(critical).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "pressure-critical".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers("critical_pressure=1".to_string())
);
}
#[tokio::test]
async fn test_no_admissible_summary_reports_pressure_and_health_filters() {
let pool = WorkerPool::new();
let critical = make_worker("critical-pressure", 8, 95.0);
prepare_fixture_worker(
&critical,
Some("1.87.0"),
PressureState::Critical,
"disk_ratio_below_critical",
)
.await;
pool.add_worker_state(critical).await;
let unhealthy = make_worker("unhealthy-rust", 8, 70.0);
prepare_fixture_worker(
&unhealthy,
Some("1.87.0"),
PressureState::Healthy,
"pressure.ok",
)
.await;
unhealthy.record_failure(None).await;
pool.add_worker_state(unhealthy).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "pressure-and-health".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers(
"critical_pressure=1,health_below_fallback=1".to_string()
)
);
}
#[tokio::test]
async fn test_pressure_preflight_does_not_globalize_single_critical_worker() {
let pool = WorkerPool::new();
let critical = make_worker("critical-pressure", 8, 95.0);
prepare_fixture_worker(
&critical,
Some("1.87.0"),
PressureState::Critical,
"disk_ratio_below_critical",
)
.await;
pool.add_worker_state(critical).await;
let telemetry_gap = make_worker("telemetry-gap-rust", 8, 70.0);
prepare_fixture_worker(
&telemetry_gap,
Some("1.87.0"),
PressureState::TelemetryGap,
"telemetry_unavailable",
)
.await;
pool.add_worker_state(telemetry_gap).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "pressure-mixed".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result
.worker
.expect("critical pressure on one worker must not block another admissible worker");
assert_eq!(
selected.config.read().await.id.as_str(),
"telemetry-gap-rust"
);
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_pressure_preflight_allows_telemetry_gap_fail_open() {
let pool = WorkerPool::new();
let telemetry_gap = make_worker("telemetry-gap", 8, 80.0);
telemetry_gap
.set_capabilities(rch_common::WorkerCapabilities {
rustc_version: Some("1.87.0".to_string()),
projects_root_ok: Some(true),
projects_root_checked_at_unix_ms: Some(1_700_000_000_000),
disk_free_gb: Some(60.0),
disk_total_gb: Some(120.0),
..Default::default()
})
.await;
telemetry_gap
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: crate::disk_pressure::PressureState::TelemetryGap,
confidence: crate::disk_pressure::PressureConfidence::Low,
reason_code: "telemetry_unavailable".to_string(),
policy_rule: "fail_open_telemetry_gap".to_string(),
disk_free_gb: Some(60.0),
disk_total_gb: Some(120.0),
disk_free_ratio: Some(0.5),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: None,
memory_pressure: None,
telemetry_age_secs: Some(600),
telemetry_fresh: false,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
pool.add_worker_state(telemetry_gap).await;
let selector = WorkerSelector::default();
let request = SelectionRequest {
job_mode: false,
project: "pressure-gap".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result
.worker
.expect("expected telemetry-gap worker to remain eligible (fail-open)");
assert_eq!(selected.config.read().await.id.as_str(), "telemetry-gap");
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_worker_selector_balanced_network_weight_prefers_low_latency() {
let pool = WorkerPool::new();
let fast_net = make_worker("fast-net", 8, 70.0);
fast_net.set_last_latency_ms(Some(20));
pool.add_worker_state(fast_net).await;
let slow_net = make_worker("slow-net", 8, 70.0);
slow_net.set_last_latency_ms(Some(600));
pool.add_worker_state(slow_net).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
min_success_rate: 0.0,
weights: SelectionWeightConfig {
speedscore: 0.0,
slots: 0.0,
health: 0.0,
cache: 0.0,
network: 1.0,
priority: 0.0,
disk: 0.0,
half_open_penalty: 1.0,
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "fast-net");
}
#[tokio::test]
async fn test_worker_selector_cache_affinity_strategy() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("warm-cache", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(
make_worker("cold-cache", 8, 90.0)
.config
.read()
.await
.clone(),
)
.await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::CacheAffinity,
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector
.record_build("warm-cache", "test-project", false)
.await;
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "warm-cache");
}
#[tokio::test]
async fn test_cache_affinity_prefers_test_cache_for_test_commands() {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("test-warm", 8, 50.0))
.await;
pool.add_worker_state(make_worker("fast", 8, 90.0)).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::CacheAffinity,
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_build("test-warm", "proj", false).await;
let request = SelectionRequest {
job_mode: false,
project: "proj".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "fast");
selector.record_build("test-warm", "proj", true).await;
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "test-warm");
}
#[tokio::test]
async fn test_worker_selector_cache_affinity_fallback_to_fastest() {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("slow", 8, 50.0)).await;
pool.add_worker_state(make_worker("fast", 8, 90.0)).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::CacheAffinity,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "new-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "fast");
}
#[tokio::test]
async fn test_worker_selector_fair_fastest_distributes() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("worker1", 8, 80.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("worker2", 8, 80.0).config.read().await.clone())
.await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::FairFastest,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let mut worker1_count = 0;
let mut worker2_count = 0;
for _ in 0..20 {
let result = selector.select(&pool, &request).await;
if let Some(worker) = result.worker {
if worker.config.read().await.id.as_str() == "worker1" {
worker1_count += 1;
} else {
worker2_count += 1;
}
}
}
assert!(
worker1_count > 0,
"Worker1 should be selected at least once"
);
assert!(
worker2_count > 0,
"Worker2 should be selected at least once"
);
}
#[tokio::test]
async fn test_worker_selector_records_build_for_cache() {
let selector = WorkerSelector::new();
selector.record_build("worker1", "project-a", false).await;
let cache = selector.cache_tracker.read().await;
let warmth = cache.estimate_warmth("worker1", "project-a", CacheUse::Build);
assert!(warmth > 0.9);
}
#[tokio::test]
async fn test_worker_selector_handles_empty_pool() {
let pool = WorkerPool::new();
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoWorkersConfigured);
}
#[tokio::test]
async fn test_worker_selector_respects_preferred_workers() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("preferred", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("faster", 8, 90.0).config.read().await.clone())
.await;
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Fastest,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("preferred")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
let selected = result.worker.expect("Expected a worker");
assert_eq!(selected.config.read().await.id.as_str(), "preferred");
}
#[tokio::test]
async fn test_worker_selector_refuses_outside_requested_worker_set() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("available", 8, 90.0)
.config
.read()
.await
.clone(),
)
.await;
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("missing")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoMatchingWorkers);
assert!(result.diagnostics.is_some());
}
#[tokio::test]
async fn test_worker_selector_busy_request_never_escapes_to_affinity_fallback() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("requested", 2, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("fallback", 8, 90.0).config.read().await.clone())
.await;
let requested = pool.get(&WorkerId::new("requested")).await.unwrap();
assert!(requested.reserve_slots(2).await);
let selector = WorkerSelector::new();
selector.record_success("fallback", "test-project").await;
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllWorkersBusy);
assert!(result.diagnostics.is_some());
}
#[tokio::test]
async fn test_worker_selector_active_project_exclusion_never_escapes_request() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("requested", 8, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(
make_worker("available", 8, 90.0)
.config
.read()
.await
.clone(),
)
.await;
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let excluded = HashSet::from(["requested".to_string()]);
let result = selector
.select_with_exclusions(&pool, &request, &excluded)
.await;
assert!(result.worker.is_none());
assert!(matches!(
result.reason,
SelectionReason::NoAdmissibleWorkers(ref summary)
if summary.contains("active_project_exclusion=1")
));
assert!(result.diagnostics.is_some());
}
#[tokio::test]
async fn test_worker_selector_impossible_requested_capacity_is_terminal() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("requested", 1, 50.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(
make_worker("available", 8, 90.0)
.config
.read()
.await
.clone(),
)
.await;
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert!(matches!(
result.reason,
SelectionReason::NoAdmissibleWorkers(ref summary)
if summary.contains("insufficient_total_slots=1")
));
let diagnostics = result.diagnostics.expect("capacity refusal diagnostics");
let requested = diagnostics
.workers
.iter()
.find(|worker| worker.worker_id.as_str() == "requested")
.expect("requested worker diagnostic");
assert!(
requested
.reason_codes
.iter()
.any(|code| code == "slots.request_exceeds_capacity")
);
}
#[tokio::test]
async fn test_worker_selector_automatic_mixed_busy_and_undersized_remains_queueable() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("busy", 4, 50.0).config.read().await.clone())
.await;
pool.add_worker(
make_worker("undersized", 1, 90.0)
.config
.read()
.await
.clone(),
)
.await;
let busy = pool.get(&WorkerId::new("busy")).await.unwrap();
assert!(busy.reserve_slots(4).await);
let selector = WorkerSelector::new();
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::AllWorkersBusy);
}
#[test]
fn test_audit_log_push_and_retrieve() {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(10);
assert!(log.is_empty());
assert_eq!(log.len(), 0);
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: 1234567890,
project: "test-project".to_string(),
command: Some("cargo build".to_string()),
strategy: "Balanced".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 3,
workers_evaluated: vec![],
selected_worker_id: Some("worker-1".to_string()),
reason: "Success".to_string(),
classification_duration_us: Some(500),
selection_duration_us: 1200,
};
log.push(entry);
assert_eq!(log.len(), 1);
assert!(!log.is_empty());
let last = log.last().unwrap();
assert_eq!(last.id, 1); assert_eq!(last.project, "test-project");
assert_eq!(last.selected_worker_id, Some("worker-1".to_string()));
}
#[test]
fn test_audit_log_eviction() {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(3);
for i in 0..5 {
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: i as u64,
project: format!("project-{}", i),
command: None,
strategy: "Fastest".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 1,
workers_evaluated: vec![],
selected_worker_id: None,
reason: "AllWorkersBusy".to_string(),
classification_duration_us: None,
selection_duration_us: 100,
};
log.push(entry);
}
assert_eq!(log.len(), 3);
let entries: Vec<_> = log.entries().iter().collect();
assert_eq!(entries[0].project, "project-2");
assert_eq!(entries[1].project, "project-3");
assert_eq!(entries[2].project, "project-4");
}
#[test]
fn test_audit_log_last_n() {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(10);
for i in 0..5 {
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: i as u64,
project: format!("project-{}", i),
command: None,
strategy: "Priority".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 1,
workers_evaluated: vec![],
selected_worker_id: Some(format!("worker-{}", i)),
reason: "Success".to_string(),
classification_duration_us: None,
selection_duration_us: 50,
};
log.push(entry);
}
let last_2 = log.last_n(2);
assert_eq!(last_2.len(), 2);
assert_eq!(last_2[0].project, "project-4"); assert_eq!(last_2[1].project, "project-3");
}
#[test]
fn test_audit_log_get_by_id() {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(10);
for i in 0..3 {
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: i as u64,
project: format!("project-{}", i),
command: None,
strategy: "CacheAffinity".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 2,
workers_evaluated: vec![],
selected_worker_id: None,
reason: "AllWorkersBusy".to_string(),
classification_duration_us: None,
selection_duration_us: 75,
};
log.push(entry);
}
assert!(log.get(1).is_some());
assert_eq!(log.get(1).unwrap().project, "project-0");
assert!(log.get(2).is_some());
assert!(log.get(3).is_some());
assert!(log.get(4).is_none()); assert!(log.get(0).is_none()); }
#[test]
fn test_audit_log_clear() {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(10);
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: 0,
project: "test".to_string(),
command: None,
strategy: "FairFastest".to_string(),
command_priority: "High".to_string(),
required_runtime: Some("Rust".to_string()),
eligible_count: 0,
workers_evaluated: vec![],
selected_worker_id: None,
reason: "NoWorkersConfigured".to_string(),
classification_duration_us: None,
selection_duration_us: 25,
};
log.push(entry);
assert_eq!(log.len(), 1);
log.clear();
assert_eq!(log.len(), 0);
assert!(log.is_empty());
}
#[test]
fn test_worker_score_breakdown_serialization() {
let _guard = test_guard!();
let breakdown = WorkerScoreBreakdown {
worker_id: "worker-1".to_string(),
total_score: 0.85,
speed_score: 0.9,
slot_availability: 0.75,
disk_headroom: 0.6,
cache_affinity: 0.5,
priority_score: 0.8,
circuit_state: "Closed".to_string(),
convergence_state: None,
reliability_state: None,
selected: true,
skip_reason: None,
};
let json = serde_json::to_string(&breakdown).unwrap();
assert!(json.contains("\"worker_id\":\"worker-1\""));
assert!(json.contains("\"selected\":true"));
}
#[tokio::test]
async fn test_selection_records_audit_entry() {
let selector = WorkerSelector::new();
let pool = WorkerPool::new();
let worker_config = WorkerConfig {
id: WorkerId::new("audit-test-worker"),
host: "localhost".to_string(),
user: "test".to_string(),
identity_file: "/tmp/test".to_string(),
total_slots: 4,
priority: 1,
tags: vec![],
tools: Vec::new(),
};
pool.add_worker(worker_config).await;
let request = SelectionRequest {
job_mode: false,
project: "audit-test-project".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::Rust,
classification_duration_us: Some(250),
hook_pid: Some(12345),
required_tools: Vec::new(),
};
let _result = selector.select(&pool, &request).await;
let audit_entries = selector.get_audit_log(Some(1)).await;
assert_eq!(audit_entries.len(), 1);
let entry = &audit_entries[0];
assert_eq!(entry.project, "audit-test-project");
assert_eq!(entry.command, Some("cargo test".to_string()));
assert_eq!(entry.strategy, "Balanced"); assert_eq!(entry.classification_duration_us, Some(250));
}
#[tokio::test]
async fn failed_remote_build_warms_its_worker_without_pinning() {
let pool = WorkerPool::new();
for (id, speed) in [("worker-a", 80.0), ("worker-b", 90.0)] {
pool.add_worker(make_worker(id, 8, speed).config.read().await.clone())
.await;
pool.get(&WorkerId::new(id))
.await
.unwrap()
.set_speed_score(speed);
}
let selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
..Default::default()
},
CircuitBreakerConfig::default(),
);
let request = SelectionRequest {
job_mode: false,
project: "poolrepro".to_string(),
command: Some("cargo build -j 2".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
async fn pick(
selector: &WorkerSelector,
pool: &WorkerPool,
request: &SelectionRequest,
) -> String {
let result = selector.select(pool, request).await;
let worker = result.worker.expect("a worker is eligible");
let id = worker.config.read().await.id.to_string();
worker.release_slots(request.estimated_cores).await;
id
}
assert_eq!(
pick(&selector, &pool, &request).await,
"worker-b",
"cold: faster worker"
);
selector
.record_remote_completion("worker-a", "poolrepro", "cargo build -j 2", 1, false)
.await;
assert_eq!(
selector
.cache_warmth("worker-a", "poolrepro", CacheUse::Build)
.await,
0.0
);
selector
.record_remote_completion("worker-a", "poolrepro", "cargo build -j 2", 101, true)
.await;
assert_eq!(
selector
.cache_warmth("worker-a", "poolrepro", CacheUse::Build)
.await,
1.0
);
assert_eq!(selector.get_pinned_worker("poolrepro").await, None);
assert_eq!(selector.get_fallback_worker("poolrepro").await, None);
assert_eq!(
pick(&selector, &pool, &request).await,
"worker-a",
"warm pool wins"
);
}
#[tokio::test]
async fn remote_completion_records_test_warmth_and_success_pins() {
let selector = WorkerSelector::new();
selector
.record_remote_completion("w1", "proj", "cargo test --workspace", 130, true)
.await;
assert_eq!(
selector.cache_warmth("w1", "proj", CacheUse::Test).await,
1.0
);
assert_eq!(selector.get_pinned_worker("proj").await, None);
selector
.record_remote_completion("w2", "proj", "cargo build", 0, true)
.await;
assert_eq!(
selector.get_pinned_worker("proj").await.as_deref(),
Some("w2")
);
}
#[test]
fn test_cache_tracker_record_success() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
let pin_window = Duration::from_secs(3600);
assert!(tracker.get_pinned_worker("project-a", pin_window).is_none());
tracker.record_build("worker1", "project-a", CacheUse::Build);
tracker.record_success("worker1", "project-a");
let pinned = tracker.get_pinned_worker("project-a", pin_window);
assert_eq!(pinned, Some("worker1"));
}
#[test]
fn test_cache_tracker_record_success_creates_cache_entry() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_success("worker1", "project-a");
let warmth = tracker.estimate_warmth("worker1", "project-a", CacheUse::Build);
assert!(
warmth > 0.0,
"successful runs should establish build warmth"
);
}
#[test]
fn test_cache_tracker_last_success_entry() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
assert!(tracker.get_last_success_worker("project-x").is_none());
tracker.record_build("worker2", "project-x", CacheUse::Build);
tracker.record_success("worker2", "project-x");
let entry = tracker.get_last_success_worker("project-x");
assert!(entry.is_some());
assert_eq!(entry.unwrap().worker_id, "worker2");
}
#[test]
fn test_cache_tracker_pin_updates_on_new_success() {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
let pin_window = Duration::from_secs(3600);
tracker.record_build("worker1", "project-a", CacheUse::Build);
tracker.record_success("worker1", "project-a");
tracker.record_build("worker2", "project-a", CacheUse::Build);
tracker.record_success("worker2", "project-a");
let pinned = tracker.get_pinned_worker("project-a", pin_window);
assert_eq!(pinned, Some("worker2"));
}
#[tokio::test]
async fn test_selector_record_success_updates_cache() {
let selector = WorkerSelector::new();
selector.record_build("worker1", "project-a", false).await;
selector.record_success("worker1", "project-a").await;
let pinned = selector.get_pinned_worker("project-a").await;
assert_eq!(pinned, Some("worker1".to_string()));
}
#[tokio::test]
async fn test_selector_affinity_disabled_returns_none() {
use rch_common::AffinityConfig;
let selector = WorkerSelector::with_config(
SelectionConfig {
affinity: AffinityConfig {
enabled: false,
..Default::default()
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_build("worker1", "project-a", false).await;
selector.record_success("worker1", "project-a").await;
let pinned = selector.get_pinned_worker("project-a").await;
assert!(pinned.is_none());
}
#[tokio::test]
async fn test_selector_fallback_disabled_returns_none() {
use rch_common::AffinityConfig;
let selector = WorkerSelector::with_config(
SelectionConfig {
affinity: AffinityConfig {
enable_last_success_fallback: false,
..Default::default()
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_build("worker1", "project-a", false).await;
selector.record_success("worker1", "project-a").await;
let fallback = selector.get_fallback_worker("project-a").await;
assert!(fallback.is_none());
}
#[tokio::test]
async fn test_selector_skips_low_success_last_success_fallback() {
use rch_common::AffinityConfig;
let pool = WorkerPool::new();
let worker = make_worker("worker1", 8, 50.0);
worker
.record_failure(Some("toolchain probe failed".to_string()))
.await;
worker
.record_failure(Some("toolchain probe failed".to_string()))
.await;
pool.add_worker_state(worker).await;
let selector = WorkerSelector::with_config(
SelectionConfig {
min_success_rate: 0.8,
affinity: AffinityConfig {
fallback_min_success_rate: 0.5,
..Default::default()
},
..Default::default()
},
CircuitBreakerConfig::default(),
);
selector.record_success("worker1", "project-a").await;
let request = SelectionRequest {
job_mode: false,
project: "project-a".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(result.reason, SelectionReason::NoWorkersPassedHealth);
}
#[tokio::test]
async fn test_daemon_admission_fallback_rechecks_disk_floor() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("worker1", 8, 90.0)).await;
let mut config = rch_common::RchConfig::default();
config.selection.min_free_gb = Some(40.0);
let selector = crate::daemon_worker_selector(
&config,
Arc::new(crate::history::BuildHistory::new(10)),
None,
);
selector.record_success("worker1", "project-a").await;
let request = SelectionRequest {
project: "project-a".to_string(),
estimated_cores: 1,
job_mode: false,
command: None,
command_priority: CommandPriority::Normal,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert_eq!(
selector
.try_fallback(&pool, &request, &HashSet::new())
.await,
Some("worker1".to_string())
);
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
state: PressureState::Healthy,
disk_free_gb: Some(39.0),
..Default::default()
})
.await;
assert_eq!(
selector
.try_fallback(&pool, &request, &HashSet::new())
.await,
None
);
}
#[tokio::test]
async fn test_try_fallback_rejects_non_assignable_worker() {
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("worker1", 8, 90.0)).await;
let selector = WorkerSelector::new();
selector.record_success("worker1", "project-a").await;
let request = SelectionRequest {
job_mode: false,
project: "project-a".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let empty = std::collections::HashSet::new();
assert_eq!(
selector
.try_fallback(&pool, &request, &empty)
.await
.as_deref(),
Some("worker1"),
"healthy last-success worker should be a valid fallback"
);
for status in [WorkerStatus::Drained, WorkerStatus::Disabled] {
pool.set_status(&WorkerId::new("worker1"), status).await;
assert_eq!(
selector.try_fallback(&pool, &request, &empty).await,
None,
"non-assignable worker ({status:?}) must not be an affinity fallback"
);
}
}
#[tokio::test]
async fn test_selector_skips_last_success_fallback_without_enough_slots() {
let pool = WorkerPool::new();
let worker = make_worker("worker1", 4, 50.0);
assert!(worker.reserve_slots(3).await);
assert_eq!(worker.available_slots().await, 1);
pool.add_worker_state(worker).await;
let selector = WorkerSelector::new();
selector.record_success("worker1", "project-a").await;
let request = SelectionRequest {
job_mode: false,
project: "project-a".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"last-success fallback must not return a worker that cannot reserve the requested slots"
);
assert_eq!(result.reason, SelectionReason::AllWorkersBusy);
let diagnostics = result
.diagnostics
.expect("slot rejection should include diagnostics");
assert!(
diagnostics.workers.iter().any(|worker| worker
.reason_codes
.iter()
.any(|code| code == "slots.insufficient")),
"diagnostics should preserve the slot-capacity root cause: {diagnostics:?}"
);
}
mod proptest_selection {
use super::*;
use proptest::prelude::*;
use std::time::Duration;
fn worker_status_strategy() -> impl Strategy<Value = WorkerStatus> {
prop_oneof![
Just(WorkerStatus::Healthy),
Just(WorkerStatus::Degraded),
Just(WorkerStatus::Unreachable),
Just(WorkerStatus::Draining),
Just(WorkerStatus::Drained),
Just(WorkerStatus::Disabled),
]
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(128))]
#[test]
fn worker_selection_only_selects_admissible_worker_statuses(
statuses in prop::collection::vec(worker_status_strategy(), 0..10),
estimated_cores in 1u32..=4,
) {
let _guard = test_guard!();
let expected_selectable = statuses
.iter()
.filter(|status| matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded))
.count();
let statuses_for_runtime = statuses.clone();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("test runtime should build");
let (healthy_statuses, selected_status) = runtime.block_on(async move {
let pool = WorkerPool::new();
for (index, status) in statuses_for_runtime.iter().copied().enumerate() {
let id = format!("status-{index}");
pool.add_worker_state(make_worker(&id, 8, 50.0 + index as f64))
.await;
pool.set_status(&WorkerId::new(&id), status).await;
}
let mut healthy_statuses = Vec::new();
for worker in pool.healthy_workers().await {
healthy_statuses.push(worker.status().await);
}
let request = SelectionRequest {
project: "worker-status-proptest".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores,
preferred_workers: vec![],
job_mode: false,
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = select_worker_with_config(
&pool,
&request,
&SelectionWeights::default(),
&CircuitBreakerConfig::default(),
)
.await;
let selected_status = if let Some(worker) = result.worker {
Some(worker.status().await)
} else {
None
};
(healthy_statuses, selected_status)
});
prop_assert_eq!(healthy_statuses.len(), expected_selectable);
prop_assert!(healthy_statuses
.iter()
.all(|status| matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded)));
if let Some(status) = selected_status {
prop_assert!(matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded));
} else {
prop_assert_eq!(expected_selectable, 0);
}
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(1000))]
#[test]
fn test_normalize_latency_output_range(latency_ms in 0u64..=u64::MAX) {
let _guard = test_guard!();
let score = WorkerSelector::normalize_latency_ms(latency_ms);
prop_assert!(score >= 0.0, "Score {score} below 0.0 for latency {latency_ms}");
prop_assert!(score <= 1.0, "Score {score} above 1.0 for latency {latency_ms}");
}
#[test]
fn test_normalize_latency_monotonic(
latency_a in 0u64..=1_000_000_000u64,
latency_b in 0u64..=1_000_000_000u64,
) {
let _guard = test_guard!();
let score_a = WorkerSelector::normalize_latency_ms(latency_a);
let score_b = WorkerSelector::normalize_latency_ms(latency_b);
if latency_a <= latency_b {
prop_assert!(
score_a >= score_b,
"Expected score_a ({score_a}) >= score_b ({score_b}) for latency_a ({latency_a}) <= latency_b ({latency_b})"
);
}
}
#[test]
fn test_normalize_latency_zero(_seed in 0u64..100u64) {
let _guard = test_guard!();
let score = WorkerSelector::normalize_latency_ms(0);
prop_assert!(
(score - 1.0).abs() < f64::EPSILON,
"Zero latency should give score 1.0, got {score}"
);
}
#[test]
fn test_normalize_latency_half_life(_seed in 0u64..100u64) {
let _guard = test_guard!();
let score = WorkerSelector::normalize_latency_ms(200);
prop_assert!(
(score - 0.5).abs() < 0.01,
"Half-life latency (200ms) should give score ~0.5, got {score}"
);
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(1000))]
#[test]
fn test_normalize_priority_output_range(
priority in 0u32..=1000u32,
min_priority in 0u32..=500u32,
max_priority in 500u32..=1000u32,
) {
let _guard = test_guard!();
let clamped_priority = priority.clamp(min_priority, max_priority);
let score = WorkerSelector::normalize_priority(clamped_priority, min_priority, max_priority);
prop_assert!(score >= 0.0, "Score {score} below 0.0");
prop_assert!(score <= 1.0, "Score {score} above 1.0");
}
#[test]
fn test_normalize_priority_equal_bounds(value in 0u32..=1000u32) {
let _guard = test_guard!();
let score = WorkerSelector::normalize_priority(value, value, value);
prop_assert!(
(score - 1.0).abs() < f64::EPSILON,
"Equal bounds should give score 1.0, got {score}"
);
}
#[test]
fn test_normalize_priority_boundary_values(
min_priority in 0u32..=500u32,
max_priority in 501u32..=1000u32,
) {
let _guard = test_guard!();
let min_score = WorkerSelector::normalize_priority(min_priority, min_priority, max_priority);
let max_score = WorkerSelector::normalize_priority(max_priority, min_priority, max_priority);
prop_assert!(
min_score.abs() < f64::EPSILON,
"Min priority should give score 0.0, got {min_score}"
);
prop_assert!(
(max_score - 1.0).abs() < f64::EPSILON,
"Max priority should give score 1.0, got {max_score}"
);
}
#[test]
fn test_normalize_priority_monotonic(
priority_a in 0u32..=1000u32,
priority_b in 0u32..=1000u32,
min_priority in 0u32..=100u32,
max_priority in 900u32..=1000u32,
) {
let _guard = test_guard!();
let clamped_a = priority_a.clamp(min_priority, max_priority);
let clamped_b = priority_b.clamp(min_priority, max_priority);
let score_a = WorkerSelector::normalize_priority(clamped_a, min_priority, max_priority);
let score_b = WorkerSelector::normalize_priority(clamped_b, min_priority, max_priority);
if clamped_a <= clamped_b {
prop_assert!(
score_a <= score_b,
"Expected score_a ({score_a}) <= score_b ({score_b}) for priority_a ({clamped_a}) <= priority_b ({clamped_b})"
);
}
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(500))]
#[test]
fn test_selection_weights_from_config(
slots in 0.0f64..=1.0f64,
speedscore in 0.0f64..=1.0f64,
cache in 0.0f64..=1.0f64,
priority in 0.0f64..=1.0f64,
disk in 0.0f64..=1.0f64,
half_open_penalty in 0.0f64..=1.0f64,
) {
let _guard = test_guard!();
let config = SelectionWeightConfig {
slots,
speedscore,
cache,
priority,
disk,
half_open_penalty,
health: 0.1,
network: 0.1,
};
let weights = SelectionWeights::from(&config);
prop_assert!((weights.slots - slots).abs() < f64::EPSILON);
prop_assert!((weights.speed - speedscore).abs() < f64::EPSILON);
prop_assert!((weights.locality - cache).abs() < f64::EPSILON);
prop_assert!((weights.priority - priority).abs() < f64::EPSILON);
prop_assert!((weights.disk - disk).abs() < f64::EPSILON);
prop_assert!((weights.half_open_penalty - half_open_penalty).abs() < f64::EPSILON);
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(200))]
#[test]
fn test_audit_log_capacity(
max_entries in 1usize..=100usize,
num_entries in 0usize..=200usize,
) {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(max_entries);
for i in 0..num_entries {
let entry = SelectionAuditEntry {
id: 0, timestamp_ms: i as u64,
project: format!("project-{i}"),
command: None,
strategy: "Priority".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 1,
workers_evaluated: vec![],
selected_worker_id: Some("worker-1".to_string()),
reason: "Success".to_string(),
classification_duration_us: None,
selection_duration_us: 100,
};
log.push(entry);
}
prop_assert!(
log.len() <= max_entries,
"Log size {} exceeds max_entries {max_entries}",
log.len()
);
}
#[test]
fn test_audit_log_ids_increasing(num_entries in 2usize..=50usize) {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(100);
for i in 0..num_entries {
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: i as u64,
project: format!("project-{i}"),
command: None,
strategy: "Priority".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 1,
workers_evaluated: vec![],
selected_worker_id: Some("worker-1".to_string()),
reason: "Success".to_string(),
classification_duration_us: None,
selection_duration_us: 100,
};
log.push(entry);
}
let entries: Vec<_> = log.entries().iter().collect();
for i in 1..entries.len() {
prop_assert!(
entries[i].id > entries[i - 1].id,
"IDs not monotonically increasing: {} vs {}",
entries[i - 1].id,
entries[i].id
);
}
}
#[test]
fn test_audit_log_last_n(
num_entries in 0usize..=50usize,
n in 0usize..=100usize,
) {
let _guard = test_guard!();
let mut log = SelectionAuditLog::new(100);
for i in 0..num_entries {
let entry = SelectionAuditEntry {
id: 0,
timestamp_ms: i as u64,
project: format!("project-{i}"),
command: None,
strategy: "Priority".to_string(),
command_priority: "Normal".to_string(),
required_runtime: None,
eligible_count: 1,
workers_evaluated: vec![],
selected_worker_id: Some("worker-1".to_string()),
reason: "Success".to_string(),
classification_duration_us: None,
selection_duration_us: 100,
};
log.push(entry);
}
let last_n_entries = log.last_n(n);
prop_assert!(
last_n_entries.len() <= n,
"last_n({n}) returned {} entries",
last_n_entries.len()
);
prop_assert!(
last_n_entries.len() <= num_entries,
"last_n returned more entries than exist"
);
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(200))]
#[test]
fn test_selection_history_capacity(num_selections in 0usize..=200usize) {
let _guard = test_guard!();
let mut history = SelectionHistory::default();
for _ in 0..num_selections {
history.record_selection("worker-1");
}
let count = history.recent_selections("worker-1", Duration::from_secs(3600));
prop_assert!(
count <= 100,
"History count {count} exceeds max_history_per_worker"
);
}
#[test]
fn test_selection_history_unknown_worker(
worker_id in "[a-z]{5,10}",
) {
let _guard = test_guard!();
let history = SelectionHistory::new();
let count = history.recent_selections(&worker_id, Duration::from_secs(3600));
prop_assert_eq!(count, 0, "Unknown worker should have 0 selections");
}
#[test]
fn test_selection_history_counts_match(
num_selections in 1usize..=50usize,
) {
let _guard = test_guard!();
let mut history = SelectionHistory::new();
for _ in 0..num_selections {
history.record_selection("test-worker");
}
let count = history.recent_selections("test-worker", Duration::from_secs(3600));
prop_assert_eq!(
count, num_selections,
"Expected {} selections, got {}", num_selections, count
);
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(200))]
#[test]
fn test_cache_tracker_warmth_range(
worker_id in "[a-z]{3,8}",
project_id in "[a-z]{3,8}",
) {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_build(&worker_id, &project_id, CacheUse::Build);
let warmth = tracker.estimate_warmth(&worker_id, &project_id, CacheUse::Build);
prop_assert!(warmth >= 0.0, "Warmth {warmth} below 0.0");
prop_assert!(warmth <= 1.0, "Warmth {warmth} above 1.0");
}
#[test]
fn test_cache_tracker_unknown_warmth(
worker_id in "[a-z]{3,8}",
project_id in "[a-z]{3,8}",
) {
let _guard = test_guard!();
let tracker = CacheTracker::new();
let warmth = tracker.estimate_warmth(&worker_id, &project_id, CacheUse::Build);
prop_assert!(
warmth.abs() < f64::EPSILON,
"Unknown worker/project should have 0.0 warmth, got {warmth}"
);
}
#[test]
fn test_cache_tracker_project_limit(num_projects in 40usize..=100usize) {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
for i in 0..num_projects {
tracker.record_build("worker-1", &format!("project-{i}"), CacheUse::Build);
}
let worker_projects = tracker.workers.get("worker-1").map(|m| m.len()).unwrap_or(0);
prop_assert!(
worker_projects <= 50,
"Worker has {worker_projects} projects, exceeds limit of 50"
);
}
#[test]
fn test_cache_tracker_immediate_warmth(
worker_id in "[a-z]{3,8}",
project_id in "[a-z]{3,8}",
) {
let _guard = test_guard!();
let mut tracker = CacheTracker::new();
tracker.record_build(&worker_id, &project_id, CacheUse::Build);
let warmth = tracker.estimate_warmth(&worker_id, &project_id, CacheUse::Build);
prop_assert!(
(warmth - 1.0).abs() < 0.01,
"Immediate warmth should be ~1.0, got {warmth}"
);
}
}
#[test]
fn test_cache_state_last_activity_none() {
let _guard = test_guard!();
let state = CacheState::default();
assert!(state.last_activity().is_none());
}
#[test]
fn test_cache_state_last_activity_build_only() {
let _guard = test_guard!();
let now = Instant::now();
let state = CacheState {
last_build: Some(now),
last_test: None,
last_success: None,
};
assert_eq!(state.last_activity(), Some(now));
}
#[test]
fn test_cache_state_last_activity_test_only() {
let _guard = test_guard!();
let now = Instant::now();
let state = CacheState {
last_build: None,
last_test: Some(now),
last_success: None,
};
assert_eq!(state.last_activity(), Some(now));
}
#[test]
fn test_cache_state_last_activity_both_returns_max() {
let _guard = test_guard!();
let earlier = Instant::now();
std::thread::sleep(Duration::from_millis(1));
let later = Instant::now();
let state = CacheState {
last_build: Some(earlier),
last_test: Some(later),
last_success: None,
};
assert_eq!(state.last_activity(), Some(later));
let state2 = CacheState {
last_build: Some(later),
last_test: Some(earlier),
last_success: None,
};
assert_eq!(state2.last_activity(), Some(later));
}
#[test]
fn test_normalize_latency_extreme_values() {
let _guard = test_guard!();
let score = WorkerSelector::normalize_latency_ms(u64::MAX);
assert!(score >= 0.0);
assert!(score <= 1.0);
assert!(!score.is_nan());
}
#[test]
fn test_normalize_priority_edge_cases() {
let _guard = test_guard!();
let score = WorkerSelector::normalize_priority(0, 100, 200);
assert!((score - 0.0).abs() < f64::EPSILON);
let score = WorkerSelector::normalize_priority(300, 100, 200);
assert!(score >= 1.0); }
#[test]
fn test_selection_history_prune() {
let _guard = test_guard!();
let mut history = SelectionHistory::new();
history.record_selection("worker-1");
history.record_selection("worker-2");
history.prune(Duration::from_secs(0));
let count = history.recent_selections("worker-1", Duration::from_secs(3600));
assert!(count <= 1);
}
#[test]
fn test_audit_log_empty_operations() {
let _guard = test_guard!();
let log = SelectionAuditLog::new(10);
assert!(log.is_empty());
assert_eq!(log.len(), 0);
assert!(log.last().is_none());
assert!(log.get(1).is_none());
assert!(log.last_n(5).is_empty());
}
#[tokio::test]
async fn test_reliability_degraded_penalty_reorders_balanced_selection() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("risky", 8, 95.0)).await;
pool.add_worker_state(make_worker("steady", 8, 70.0)).await;
let remediation = seeded_remediation_pipeline("risky", &[1001, 1002]).await;
let reliability_cfg = ReliabilityConfig {
weights: process_only_weights(),
quarantine_threshold: 0.95,
degraded_penalty_floor: 0.05,
..ReliabilityConfig::default()
};
let mut reliability = ReliabilityAggregator::new(reliability_cfg);
reliability.set_remediation(remediation);
let mut selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
weights: SelectionWeightConfig {
speedscore: 1.0,
slots: 0.0,
health: 0.0,
cache: 0.0,
network: 0.0,
priority: 0.0,
..SelectionWeightConfig::default()
},
..SelectionConfig::default()
},
CircuitBreakerConfig::default(),
);
selector.set_reliability(Arc::new(reliability));
let request = SelectionRequest {
job_mode: false,
project: "reliability-penalty".to_string(),
command: Some("cargo build --workspace".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert_eq!(result.reason, SelectionReason::Success);
let selected = result.worker.expect("expected selected worker");
assert_eq!(selected.config.read().await.id.as_str(), "steady");
let assessment = selector
.reliability
.as_ref()
.expect("reliability wired")
.get_assessment("risky")
.await
.expect("expected cached reliability assessment");
assert_eq!(assessment.health_state, WorkerHealthState::Degraded);
assert!(!assessment.hard_exclude);
let audit = selector
.get_last_audit_entry()
.await
.expect("expected selection audit entry");
let risky_breakdown = audit
.workers_evaluated
.iter()
.find(|w| w.worker_id == "risky")
.expect("expected risky worker in audit");
assert_eq!(
risky_breakdown.reliability_state.as_deref(),
Some("degraded")
);
}
#[tokio::test]
async fn test_reliability_quarantine_worker_hard_excluded_preflight() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker_state(make_worker("risky", 8, 95.0)).await;
let remediation = seeded_remediation_pipeline("risky", &[2001, 2002]).await;
let reliability_cfg = ReliabilityConfig {
weights: process_only_weights(),
quarantine_threshold: 0.05,
..ReliabilityConfig::default()
};
let mut reliability = ReliabilityAggregator::new(reliability_cfg);
reliability.set_remediation(remediation);
let mut selector = WorkerSelector::new();
selector.set_reliability(Arc::new(reliability));
let request = SelectionRequest {
job_mode: false,
project: "reliability-quarantine".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"quarantined worker should be excluded"
);
assert_eq!(result.reason, SelectionReason::AllWorkersFailedPreflight);
let assessment = selector
.reliability
.as_ref()
.expect("reliability wired")
.get_assessment("risky")
.await
.expect("expected cached reliability assessment");
assert_eq!(assessment.health_state, WorkerHealthState::Quarantined);
assert!(assessment.hard_exclude);
assert_eq!(assessment.penalty, 1.0);
}
#[tokio::test]
async fn test_reliability_scoring_does_not_advance_recovery_hysteresis() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let worker = make_worker("recovering", 8, 90.0);
for _ in 0..10 {
worker
.record_failure(Some("build failed".to_string()))
.await;
}
let reliability_cfg = ReliabilityConfig {
weights: SignalWeights {
circuit: 1.0,
convergence: 0.0,
pressure: 0.0,
process: 0.0,
cancellation: 0.0,
},
quarantine_threshold: 0.9,
recovery_threshold: 0.1,
recovery_ticks: 2,
min_quarantine_duration: Duration::ZERO,
probing_penalty: 0.5,
..ReliabilityConfig::default()
};
let reliability = Arc::new(ReliabilityAggregator::new(reliability_cfg));
let quarantined = reliability.evaluate(&worker, "recovering").await;
assert_eq!(quarantined.health_state, WorkerHealthState::Quarantined);
for _ in 0..500 {
worker.record_success().await;
}
pool.add_worker_state(worker).await;
let mut selector = WorkerSelector::with_config(
SelectionConfig {
strategy: SelectionStrategy::Balanced,
weights: SelectionWeightConfig {
speedscore: 1.0,
slots: 0.0,
health: 0.0,
cache: 0.0,
network: 0.0,
priority: 0.0,
..SelectionWeightConfig::default()
},
..SelectionConfig::default()
},
CircuitBreakerConfig::default(),
);
selector.set_reliability(reliability.clone());
let request = SelectionRequest {
job_mode: false,
project: "reliability-recovery".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let first = selector.select(&pool, &request).await;
assert!(first.worker.is_none());
assert_eq!(first.reason, SelectionReason::AllWorkersFailedPreflight);
let after_first = reliability
.get_assessment("recovering")
.await
.expect("expected reliability assessment after first selection");
assert_eq!(after_first.health_state, WorkerHealthState::Quarantined);
let second = selector.select(&pool, &request).await;
assert_eq!(second.reason, SelectionReason::Success);
assert!(second.worker.is_some());
let after_second = reliability
.get_assessment("recovering")
.await
.expect("expected reliability assessment after second selection");
assert_eq!(
after_second.health_state,
WorkerHealthState::ProbingRecovery
);
assert_eq!(after_second.penalty, 0.5);
}
fn make_convergence_svc() -> Arc<crate::repo_convergence::RepoConvergenceService> {
Arc::new(crate::repo_convergence::RepoConvergenceService::new(
crate::events::EventBus::new(64),
))
}
fn make_selection_request(project: &str) -> SelectionRequest {
SelectionRequest {
job_mode: false,
project: project.to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
}
}
#[tokio::test]
async fn test_convergence_ready_worker_selected() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_some(), "Ready worker should be selected");
assert_eq!(result.reason, SelectionReason::Success);
}
#[tokio::test]
async fn test_convergence_drifting_worker_still_eligible() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence
.update_required_repos(
&wid,
vec!["repo-a".into(), "repo-b".into()],
vec!["repo-a".into()], )
.await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"Drifting worker should still be eligible"
);
}
#[tokio::test]
async fn test_convergence_failed_worker_excluded() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence
.update_required_repos(
&wid,
vec!["repo-a".into()],
vec![], )
.await;
for _ in 0..3 {
let _ = convergence
.record_convergence_attempt(&wid, 0, 1, 0, 10_000, Some("ssh_timeout".into()))
.await;
}
let drift = convergence.get_drift_state(&wid).await;
assert_eq!(
drift,
crate::repo_convergence::ConvergenceDriftState::Failed
);
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"Failed convergence worker should not be selected"
);
assert_eq!(result.reason, SelectionReason::AllWorkersFailedConvergence);
}
#[tokio::test]
async fn test_convergence_converging_worker_excluded() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence.mark_converging(&wid).await;
let drift = convergence.get_drift_state(&wid).await;
assert_eq!(
drift,
crate::repo_convergence::ConvergenceDriftState::Converging
);
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"Converging worker should be excluded"
);
assert_eq!(result.reason, SelectionReason::AllWorkersFailedConvergence);
}
#[tokio::test]
async fn test_preflight_rejects_mixed_convergence_and_busy_pool() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("converging", 8, 80.0)
.config
.read()
.await
.clone(),
)
.await;
let busy = make_worker("busy", 8, 80.0);
assert!(busy.reserve_slots(8).await);
pool.add_worker_state(busy).await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("converging");
convergence.mark_converging(&wid).await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_none());
assert_eq!(
result.reason,
SelectionReason::NoAdmissibleWorkers(
"insufficient_slots=1,hard_preflight=1".to_string()
)
);
}
#[tokio::test]
async fn test_convergence_stale_worker_fail_open() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
let drift = convergence.get_drift_state(&wid).await;
assert_eq!(drift, crate::repo_convergence::ConvergenceDriftState::Stale);
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"Stale convergence should fail-open and allow selection"
);
}
#[tokio::test]
async fn test_convergence_mixed_workers_prefers_ready() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w-ready", 8, 80.0).config.read().await.clone())
.await;
pool.add_worker(
make_worker("w-failed", 16, 95.0)
.config
.read()
.await
.clone(),
)
.await;
let convergence = make_convergence_svc();
let wid_ready = WorkerId::new("w-ready");
convergence
.update_required_repos(&wid_ready, vec!["repo-x".into()], vec!["repo-x".into()])
.await;
let wid_failed = WorkerId::new("w-failed");
convergence
.update_required_repos(&wid_failed, vec!["repo-x".into()], vec![])
.await;
for _ in 0..3 {
let _ = convergence
.record_convergence_attempt(
&wid_failed,
0,
1,
0,
10_000,
Some("timeout".into()),
)
.await;
}
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_some());
let selected_id = result
.worker
.unwrap()
.config
.read()
.await
.id
.as_str()
.to_string();
assert_eq!(
selected_id, "w-ready",
"Should select Ready worker, not the Failed one"
);
}
#[tokio::test]
async fn test_convergence_no_service_wired_all_eligible() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("w2", 8, 80.0).config.read().await.clone())
.await;
let selector = WorkerSelector::new();
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"Without convergence service, workers should be eligible"
);
}
#[tokio::test]
async fn test_convergence_audit_includes_state() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let _ = selector.select(&pool, &request).await;
let entries = selector.get_audit_log(Some(1)).await;
assert!(!entries.is_empty(), "Should have audit log entry");
let entry = &entries[0];
assert!(!entry.workers_evaluated.is_empty());
let breakdown = &entry.workers_evaluated[0];
assert_eq!(
breakdown.convergence_state.as_deref(),
Some("ready"),
"Audit should include convergence state"
);
}
#[tokio::test]
async fn test_integration_drift_detect_converge_reselect() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid = WorkerId::new("w1");
convergence
.update_required_repos(&wid, vec!["repo-x".into()], vec![])
.await;
assert_eq!(
convergence.get_drift_state(&wid).await,
crate::repo_convergence::ConvergenceDriftState::Drifting
);
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence.clone());
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"Drifting worker should still be eligible"
);
convergence.mark_converging(&wid).await;
let mut selector2 = WorkerSelector::new();
selector2.set_repo_convergence(convergence.clone());
let result2 = selector2.select(&pool, &request).await;
assert!(
result2.worker.is_none(),
"Converging worker should be excluded from selection"
);
convergence
.record_convergence_attempt(&wid, 1, 0, 0, 1_000, None)
.await
.unwrap();
assert_eq!(
convergence.get_drift_state(&wid).await,
crate::repo_convergence::ConvergenceDriftState::Ready
);
let mut selector3 = WorkerSelector::new();
selector3.set_repo_convergence(convergence);
let result3 = selector3.select(&pool, &request).await;
assert!(
result3.worker.is_some(),
"Ready worker should be eligible after convergence"
);
}
#[tokio::test]
async fn test_integration_all_workers_failed_convergence_error() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("w2", 8, 85.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
for name in ["w1", "w2"] {
let wid = WorkerId::new(name);
convergence
.update_required_repos(&wid, vec!["repo-x".into()], vec![])
.await;
for _ in 0..3 {
let _ = convergence
.record_convergence_attempt(
&wid,
0,
1,
0,
10_000,
Some("sync_error".into()),
)
.await;
}
assert_eq!(
convergence.get_drift_state(&wid).await,
crate::repo_convergence::ConvergenceDriftState::Failed
);
}
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_none(),
"No worker should be selected when all are Failed"
);
assert_eq!(
result.reason,
rch_common::SelectionReason::AllWorkersFailedConvergence,
"Should return AllWorkersFailedConvergence reason"
);
}
#[tokio::test]
async fn test_integration_stale_ready_prefers_ready() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w-stale", 8, 90.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("w-ready", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid_ready = WorkerId::new("w-ready");
convergence
.update_required_repos(&wid_ready, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_some(), "Should select a worker");
}
#[tokio::test]
async fn test_integration_budget_exhausted_excluded_fresh_selected() {
let pool = WorkerPool::new();
pool.add_worker(
make_worker("w-exhausted", 16, 95.0)
.config
.read()
.await
.clone(),
)
.await;
pool.add_worker(make_worker("w-fresh", 8, 80.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let wid_exhausted = WorkerId::new("w-exhausted");
convergence
.update_required_repos(&wid_exhausted, vec!["repo-x".into()], vec![])
.await;
for _ in 0..3 {
let _ = convergence
.record_convergence_attempt(
&wid_exhausted,
0,
1,
0,
10_000,
Some("fail".into()),
)
.await;
}
let wid_fresh = WorkerId::new("w-fresh");
convergence
.update_required_repos(&wid_fresh, vec!["repo-x".into()], vec!["repo-x".into()])
.await;
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(result.worker.is_some(), "Should select the fresh worker");
let selected_id = result
.worker
.as_ref()
.unwrap()
.config
.read()
.await
.id
.clone();
assert_eq!(
selected_id.as_str(),
"w-fresh",
"Should prefer the Ready worker over the Failed one"
);
}
#[tokio::test]
async fn test_integration_fail_open_all_stale_still_selects() {
let pool = WorkerPool::new();
pool.add_worker(make_worker("w1", 8, 80.0).config.read().await.clone())
.await;
pool.add_worker(make_worker("w2", 8, 85.0).config.read().await.clone())
.await;
let convergence = make_convergence_svc();
let mut selector = WorkerSelector::new();
selector.set_repo_convergence(convergence);
let request = make_selection_request("test-project");
let result = selector.select(&pool, &request).await;
assert!(
result.worker.is_some(),
"All-stale scenario should still select a worker (fail-open)"
);
}
}
}