use crate::DaemonContext;
use crate::alerts::AlertInfo;
use crate::events::EventBus;
use crate::history::{QueueSelectionContract, QueuedWaiterClaim};
use crate::metrics;
use crate::metrics::budget::{self, BudgetStatusResponse};
use crate::reload;
use crate::telemetry::collect_telemetry_from_worker;
use crate::workers::{
WorkerCapabilitiesResponse, get_workers_capabilities, handle_worker_disable,
handle_worker_drain, handle_worker_enable,
};
use anyhow::{Context, Result, anyhow};
use chrono::{Duration as ChronoDuration, Utc};
use rch_common::job_identity::{
DurableJobLease, LOCAL_WRAPPER_ID_PREFIX, default_job_lease_directory,
};
use rch_common::{
ApiError, BuildHeartbeatRequest, BuildRecord, BuildStats, BypassRecord, BypassRecordStore,
CircuitBreakerConfig, CircuitState, CommandPriority, ErrorCode, ReleaseRequest,
RequiredRuntime, SELECTION_RESPONSE_PROTOCOL_VERSION, SavedTimeStats, SelectedWorker,
SelectionReason, SelectionRequest, SelectionResponse, WorkerId, WorkerStatus,
default_bypass_record_path,
};
use rch_telemetry::protocol::{TelemetrySource, TestRunRecord, TestRunStats, WorkerTelemetry};
use rch_telemetry::speedscore::SpeedScore;
use serde::Serialize;
use sha2::{Digest, Sha256};
use std::path::PathBuf;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
use tracing::{debug, info, warn};
use uuid::Uuid;
const MAX_LINE_SIZE: usize = 65536;
async fn read_line_with_limit<R: AsyncBufReadExt + Unpin>(
reader: &mut R,
buf: &mut String,
limit: usize,
) -> Result<usize> {
let mut bytes = Vec::new();
let mut count = 0;
while count < limit {
let byte = reader.read_u8().await;
match byte {
Ok(b) => {
count += 1;
bytes.push(b);
if b == b'\n' {
break;
}
}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
break;
}
Err(e) => return Err(e.into()),
}
}
if count >= limit && bytes.last() != Some(&b'\n') {
return Err(anyhow!("Request line exceeded limit of {} bytes", limit));
}
let s = String::from_utf8(bytes).map_err(|e| anyhow!("Invalid UTF-8 in request: {}", e))?;
buf.push_str(&s);
Ok(count)
}
fn format_wait_time(secs: u64) -> String {
if secs < 60 {
format!("{}s", secs)
} else if secs < 3600 {
let mins = secs / 60;
let s = secs % 60;
if s == 0 {
format!("{}m", mins)
} else {
format!("{}m {}s", mins, s)
}
} else {
let hours = secs / 3600;
let mins = (secs % 3600) / 60;
if mins == 0 {
format!("{}h", hours)
} else {
format!("{}h {}m", hours, mins)
}
}
}
#[derive(Debug)]
enum ApiRequest {
SelectWorker {
request: SelectionRequest,
local_wrapper_id: Option<String>,
wait_for_worker: bool,
wait_timeout_secs: Option<u64>,
dry_run: bool,
resume_queued: bool,
},
ReleaseWorker(ReleaseRequest),
RecordBuild {
worker_id: WorkerId,
project: String,
is_test: bool,
},
BuildHeartbeat,
IngestTelemetry(TelemetrySource),
TestRun,
TelemetryPoll {
worker_id: WorkerId,
},
SpeedScore {
worker_id: WorkerId,
},
SpeedScoreHistory {
worker_id: WorkerId,
days: u32,
limit: usize,
offset: usize,
},
SpeedScores,
WorkersCapabilities {
refresh: bool,
},
BenchmarkTrigger {
worker_id: WorkerId,
},
Events,
Status,
Metrics,
Health,
Ready,
Budget,
SelfTestStatus,
SelfTestHistory {
limit: usize,
},
SelfTestRun(SelfTestRunRequest),
Shutdown,
RestartAdmission {
close: bool,
},
RestartAdmissionStatus,
Reload,
CancelBuild {
build_id: u64,
force: bool,
local_wrapper_id: Option<String>,
},
CancelJob {
local_wrapper_id: String,
},
CancelAllBuilds {
force: bool,
},
BuildRecovery {
build_id: u64,
local_wrapper_id: String,
},
WorkerDrain {
worker_id: WorkerId,
},
WorkerEnable {
worker_id: WorkerId,
},
WorkerDisable {
worker_id: WorkerId,
reason: Option<String>,
drain_first: bool,
},
RepoConvergenceStatus {
worker_id: Option<WorkerId>,
},
RepoConvergenceDryRun {
worker_id: WorkerId,
},
RepoConvergenceRepair {
worker_id: WorkerId,
},
}
#[derive(Debug, Serialize)]
pub struct DaemonFullStatus {
pub daemon: DaemonStatusInfo,
pub workers: Vec<WorkerStatusInfo>,
pub active_builds: Vec<ActiveBuild>,
pub queued_builds: Vec<QueuedBuild>,
pub recent_builds: Vec<BuildRecord>,
pub issues: Vec<Issue>,
pub alerts: Vec<AlertInfo>,
pub stats: BuildStats,
pub test_stats: TestRunStats,
pub saved_time: SavedTimeStats,
pub remediation: rch_common::remediation_view::RemediationView,
}
#[derive(Debug, Serialize)]
pub struct DaemonStatusInfo {
pub pid: u32,
pub uptime_secs: u64,
pub version: String,
pub socket_path: String,
pub started_at: String,
pub workers_total: usize,
pub workers_healthy: usize,
pub slots_total: u32,
pub slots_available: u32,
}
#[derive(Debug, Serialize)]
pub struct WorkerStatusInfo {
pub id: String,
pub host: String,
pub user: String,
pub status: String,
pub circuit_state: String,
pub used_slots: u32,
pub total_slots: u32,
pub speed_score: f64,
pub last_error: Option<String>,
pub consecutive_failures: u32,
pub recovery_in_secs: Option<u64>,
pub failure_history: Vec<bool>,
pub pressure_state: String,
pub pressure_confidence: String,
pub pressure_reason_code: String,
pub pressure_policy_rule: String,
pub pressure_disk_free_gb: Option<f64>,
pub pressure_disk_total_gb: Option<f64>,
pub pressure_disk_free_ratio: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub pressure_build_disk_free_gb: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub pressure_build_disk_total_gb: Option<f64>,
pub pressure_disk_io_util_pct: Option<f64>,
pub pressure_memory_pressure: Option<f64>,
pub pressure_telemetry_age_secs: Option<u64>,
pub pressure_telemetry_fresh: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bypass: Option<BypassRecord>,
}
#[derive(Debug, Serialize)]
pub struct ActiveBuild {
pub id: u64,
pub project_id: String,
pub worker_id: String,
pub command: String,
pub started_at: String,
pub last_heartbeat_at: String,
pub heartbeat_age_secs: u64,
pub last_progress_at: String,
pub progress_age_secs: u64,
pub heartbeat_phase: String,
pub heartbeat_detail: Option<String>,
pub heartbeat_counter: u64,
pub heartbeat_percent: Option<f64>,
pub slots: u32,
pub detector_hook_alive: bool,
pub detector_heartbeat_stale: bool,
pub detector_progress_stale: bool,
pub detector_confidence: f64,
pub detector_build_age_secs: u64,
pub detector_slots_owned: u32,
pub detector_last_evaluated_at: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct QueuedBuild {
pub id: u64,
pub id_text: String,
pub project_id: String,
pub command: String,
pub queued_at: String,
pub position: usize,
pub slots_needed: u32,
pub estimated_start: Option<String>,
pub wait_time: String,
}
#[derive(Debug, Serialize)]
pub struct Issue {
pub severity: String,
pub summary: String,
pub remediation: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct HealthResponse {
pub status: String,
pub version: String,
pub uptime_seconds: u64,
}
#[derive(Debug, Serialize)]
pub struct ReadyResponse {
pub status: String,
pub workers_available: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,
}
#[derive(Debug)]
pub struct SelfTestRunRequest {
pub worker_ids: Vec<String>,
pub project: Option<String>,
pub timeout_secs: Option<u64>,
pub release_mode: bool,
pub scheduled: bool,
pub retries: Option<u32>,
}
#[derive(Debug, Serialize)]
pub struct SelfTestStatusResponse {
pub enabled: bool,
pub schedule: Option<String>,
pub interval: Option<String>,
pub last_run: Option<crate::self_test::SelfTestRunRecord>,
pub next_run: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct SelfTestHistoryResponse {
pub runs: Vec<crate::self_test::SelfTestRunRecord>,
pub results: Vec<crate::self_test::SelfTestResultRecord>,
}
#[derive(Debug, Serialize)]
pub struct SelfTestRunResponse {
pub run: crate::self_test::SelfTestRunRecord,
pub results: Vec<crate::self_test::SelfTestResultRecord>,
}
#[derive(Debug, Serialize)]
pub struct BuildHeartbeatResponse {
pub status: String,
pub build_id: u64,
pub worker_id: String,
pub phase: String,
}
#[derive(Debug, Serialize)]
pub struct RestartAdmissionResponse {
pub admission_closed: bool,
pub restart_permitted: bool,
pub active_build_ids: Vec<u64>,
pub queued_build_ids: Vec<u64>,
pub client_lease_ids: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub client_lease_scan_error: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct CancelBuildResponse {
pub status: String,
pub build_id: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub project_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
pub slots_released: u32,
}
#[derive(Debug, Serialize)]
pub struct CancelAllBuildsResponse {
pub status: String,
pub cancelled_count: usize,
pub cancelled: Vec<CancelledBuildInfo>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct CancelledBuildInfo {
pub build_id: u64,
pub worker_id: String,
pub project_id: String,
pub slots_released: u32,
}
#[derive(Debug, Serialize)]
pub struct SpeedScoreView {
pub total: f64,
pub cpu_score: f64,
pub memory_score: f64,
pub disk_score: f64,
pub network_score: f64,
pub compilation_score: f64,
pub measured_at: String,
pub version: u32,
}
#[derive(Debug, Serialize)]
pub struct SpeedScoreResponse {
pub worker_id: String,
pub speedscore: Option<SpeedScoreView>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct PaginationInfo {
pub total: u64,
pub offset: usize,
pub limit: usize,
pub has_more: bool,
}
#[derive(Debug, Serialize)]
pub struct SpeedScoreHistoryResponse {
pub worker_id: String,
pub history: Vec<SpeedScoreView>,
pub pagination: PaginationInfo,
}
#[derive(Debug, Serialize)]
pub struct SpeedScoreListResponse {
pub workers: Vec<SpeedScoreWorker>,
}
#[derive(Debug, Serialize)]
pub struct SpeedScoreWorker {
pub worker_id: String,
pub speedscore: Option<SpeedScoreView>,
pub status: WorkerStatus,
}
#[derive(Debug, Serialize)]
pub struct BenchmarkTriggerResponse {
pub status: String,
pub worker_id: String,
pub request_id: String,
}
#[derive(Debug, Serialize)]
pub struct ConvergenceWorkerView {
pub worker_id: String,
pub drift_state: String,
pub drift_confidence: f64,
pub required_repos: Vec<String>,
pub synced_repos: Vec<String>,
pub missing_repos: Vec<String>,
pub attempt_budget_remaining: u32,
pub time_budget_remaining_ms: u64,
pub last_status_check_unix_ms: i64,
pub remediation: Vec<String>,
}
#[derive(Debug, Serialize)]
pub struct RepoConvergenceStatusResponse {
pub status: String,
pub workers: Vec<ConvergenceWorkerView>,
pub recent_outcomes: Vec<crate::repo_convergence::ConvergenceOutcome>,
pub summary: ConvergenceSummary,
}
#[derive(Debug, Serialize)]
pub struct ConvergenceSummary {
pub total_workers: usize,
pub ready: usize,
pub drifting: usize,
pub converging: usize,
pub failed: usize,
pub stale: usize,
}
#[derive(Debug, Serialize)]
pub struct RepoConvergenceDryRunResponse {
pub status: String,
pub worker_id: String,
pub current_state: String,
pub missing_repos: Vec<String>,
pub has_budget: bool,
pub attempt_budget_remaining: u32,
pub time_budget_remaining_ms: u64,
pub would_attempt: bool,
pub reason: String,
pub remediation: Vec<String>,
}
#[derive(Debug, Serialize)]
pub struct RepoConvergenceRepairResponse {
pub status: String,
pub worker_id: String,
pub action: String,
pub previous_state: String,
pub new_state: String,
pub message: String,
}
#[derive(Debug, Serialize)]
#[serde(untagged)]
pub(crate) enum ApiResponse<T: Serialize> {
Ok(T),
Error(ApiError),
}
fn selection_response_json(response: &SelectionResponse) -> Result<String> {
let mut value = serde_json::to_value(response)?;
let object = value
.as_object_mut()
.ok_or_else(|| anyhow!("selection response did not serialize as an object"))?;
object.insert(
"selection_protocol_version".to_string(),
serde_json::json!(SELECTION_RESPONSE_PROTOCOL_VERSION),
);
Ok(serde_json::to_string(&value)?)
}
async fn log_worker_admin_action(
action: &str,
worker_id: &WorkerId,
outcome: &str,
peer_pid: Option<i32>,
) {
let caller = match peer_pid {
Some(pid) => describe_caller(pid).await,
None => "unknown caller".to_string(),
};
info!(
action,
worker = %worker_id,
outcome,
caller = %caller,
"Worker admin action"
);
}
async fn describe_caller(pid: i32) -> String {
async fn ps(fields: &str, pid: &str) -> Option<String> {
let output = tokio::time::timeout(
Duration::from_secs(1),
tokio::process::Command::new("ps")
.args(["-o", fields, "-p", pid])
.kill_on_drop(true)
.output(),
)
.await
.ok()?
.ok()?;
let text = String::from_utf8_lossy(&output.stdout).trim().to_string();
(!text.is_empty()).then(|| text.chars().take(160).collect())
}
let pid_text = pid.to_string();
let Some(line) = ps("ppid=,command=", &pid_text).await else {
return format!("pid={pid}");
};
let (ppid, command) = line
.trim_start()
.split_once(char::is_whitespace)
.unwrap_or((line.as_str(), ""));
let parent = ps("command=", ppid.trim()).await.unwrap_or_default();
format!(
"pid={pid} cmd={:?} ppid={} parent={parent:?}",
command.trim(),
ppid.trim()
)
}
pub async fn handle_connection(
stream: UnixStream,
ctx: DaemonContext,
shutdown_tx: tokio::sync::mpsc::Sender<()>,
) -> Result<()> {
handle_connection_with_metrics(
stream,
ctx,
shutdown_tx,
metrics::tracing::request_metrics(),
)
.await
}
async fn handle_connection_with_metrics(
stream: UnixStream,
ctx: DaemonContext,
shutdown_tx: tokio::sync::mpsc::Sender<()>,
request_metrics: Option<rch_telemetry::metrics::Metrics>,
) -> Result<()> {
let peer_pid = stream.peer_cred().ok().and_then(|cred| cred.pid());
let (reader, mut writer) = stream.into_split();
let mut reader = BufReader::new(reader);
let mut line = String::new();
let n = read_line_with_limit(&mut reader, &mut line, MAX_LINE_SIZE).await?;
if n == 0 {
return Ok(()); }
let line = line.trim();
debug!("Received request: {}", line);
let request = parse_request(line);
let _request_duration = (!matches!(&request, Ok(ApiRequest::Events)))
.then(|| metrics::tracing::RequestDuration::start(request_metrics));
let (response_json, content_type) = match request {
Ok(ApiRequest::SelectWorker {
request,
local_wrapper_id,
wait_for_worker,
wait_timeout_secs,
dry_run,
resume_queued,
}) => {
metrics::inc_requests("select-worker");
let response = if dry_run {
handle_select_worker_dry_run(&ctx, &request).await
} else {
handle_select_worker_mode(
&ctx,
request,
wait_for_worker,
wait_timeout_secs,
local_wrapper_id,
resume_queued,
)
.await?
};
(selection_response_json(&response)?, "application/json")
}
Ok(ApiRequest::ReleaseWorker(mut request)) => {
metrics::inc_requests("release-worker");
let mut body_line = String::new();
if let Ok(Ok(_)) = tokio::time::timeout(
Duration::from_millis(50),
read_line_with_limit(&mut reader, &mut body_line, MAX_LINE_SIZE),
)
.await
{
let body = body_line.trim();
if !body.is_empty()
&& let Ok(timing) =
serde_json::from_str::<rch_common::CommandTimingBreakdown>(body)
{
request.timing = Some(timing);
}
}
handle_release_worker(&ctx, request).await?;
("{}".to_string(), "application/json")
}
Ok(ApiRequest::RecordBuild {
worker_id,
project,
is_test,
}) => {
metrics::inc_requests("record-build");
handle_record_build(&ctx, &worker_id, &project, is_test).await?;
("{}".to_string(), "application/json")
}
Ok(ApiRequest::BuildHeartbeat) => {
metrics::inc_requests("build-heartbeat");
let mut body = String::new();
match tokio::time::timeout(
Duration::from_secs(5),
read_line_with_limit(&mut reader, &mut body, MAX_LINE_SIZE),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
warn!("Build heartbeat body read timed out");
return Ok(());
}
}
let payload = body.trim();
if payload.is_empty() {
(
"{\"status\":\"error\",\"error\":\"empty build heartbeat payload\"}"
.to_string(),
"application/json",
)
} else {
match serde_json::from_str::<BuildHeartbeatRequest>(payload) {
Ok(request) => {
let response = handle_build_heartbeat(&ctx, request).await;
(serde_json::to_string(&response)?, "application/json")
}
Err(e) => {
warn!("Failed to parse build heartbeat JSON: {}", e);
(
"{\"status\":\"error\",\"error\":\"invalid build heartbeat payload\"}"
.to_string(),
"application/json",
)
}
}
}
}
Ok(ApiRequest::IngestTelemetry(source)) => {
metrics::inc_requests("telemetry");
let mut body = String::new();
match tokio::time::timeout(
Duration::from_secs(5),
read_line_with_limit(&mut reader, &mut body, MAX_LINE_SIZE),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
warn!("Telemetry body read timed out");
return Ok(());
}
}
let payload = body.trim();
if payload.is_empty() {
warn!("Telemetry ingestion received empty body");
(
"{\"status\":\"error\",\"error\":\"empty telemetry payload\"}".to_string(),
"application/json",
)
} else {
match WorkerTelemetry::from_json(payload) {
Ok(telemetry) => {
if !telemetry.is_compatible() {
warn!(
"Telemetry protocol version mismatch for worker {}",
telemetry.worker_id
);
}
ctx.telemetry.ingest(telemetry, source);
("{\"status\":\"ok\"}".to_string(), "application/json")
}
Err(e) => {
warn!("Failed to parse telemetry JSON: {}", e);
(
"{\"status\":\"error\",\"error\":\"invalid telemetry payload\"}"
.to_string(),
"application/json",
)
}
}
}
}
Ok(ApiRequest::TestRun) => {
metrics::inc_requests("test-run");
let mut body = String::new();
match tokio::time::timeout(
Duration::from_secs(5),
read_line_with_limit(&mut reader, &mut body, MAX_LINE_SIZE),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
warn!("Test run body read timed out");
return Ok(());
}
}
let payload = body.trim();
if payload.is_empty() {
warn!("Test run ingestion received empty body");
(
"{\"status\":\"error\",\"error\":\"empty test run payload\"}".to_string(),
"application/json",
)
} else {
match TestRunRecord::from_json(payload) {
Ok(record) => {
ctx.telemetry.record_test_run(record);
("{\"status\":\"ok\"}".to_string(), "application/json")
}
Err(e) => {
warn!("Failed to parse test run JSON: {}", e);
(
"{\"status\":\"error\",\"error\":\"invalid test run payload\"}"
.to_string(),
"application/json",
)
}
}
}
}
Ok(ApiRequest::TelemetryPoll { worker_id }) => {
metrics::inc_requests("telemetry-poll");
let response = handle_telemetry_poll(&ctx, &worker_id).await;
let response_json = serde_json::to_string(&response)?;
(response_json, "application/json")
}
Ok(ApiRequest::SpeedScore { worker_id }) => {
metrics::inc_requests("speedscore");
let response = handle_speedscore(&ctx, &worker_id).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::SpeedScoreHistory {
worker_id,
days,
limit,
offset,
}) => {
metrics::inc_requests("speedscore-history");
let response = handle_speedscore_history(&ctx, &worker_id, days, limit, offset).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::SpeedScores) => {
metrics::inc_requests("speedscore-list");
let response = handle_speedscore_list(&ctx).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::WorkersCapabilities { refresh }) => {
metrics::inc_requests("workers-capabilities");
let response = handle_workers_capabilities(&ctx, refresh).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::BenchmarkTrigger { worker_id }) => {
metrics::inc_requests("benchmark-trigger");
let response = handle_benchmark_trigger(&ctx, &worker_id).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::Events) => {
metrics::inc_requests("events");
handle_event_stream(&mut writer, ctx.events.clone()).await?;
return Ok(());
}
Ok(ApiRequest::Status) => {
metrics::inc_requests("status");
let status = handle_status(&ctx).await?;
(serde_json::to_string(&status)?, "application/json")
}
Ok(ApiRequest::Metrics) => {
metrics::inc_requests("metrics");
let metrics_text = handle_metrics()?;
(metrics_text, "text/plain; version=0.0.4")
}
Ok(ApiRequest::Health) => {
metrics::inc_requests("health");
let health = handle_health(&ctx);
(serde_json::to_string(&health)?, "application/json")
}
Ok(ApiRequest::Ready) => {
metrics::inc_requests("ready");
let ready = handle_ready(&ctx).await;
(serde_json::to_string(&ready)?, "application/json")
}
Ok(ApiRequest::Budget) => {
metrics::inc_requests("budget");
let budget_status = handle_budget();
(serde_json::to_string(&budget_status)?, "application/json")
}
Ok(ApiRequest::SelfTestStatus) => {
metrics::inc_requests("self-test-status");
let status = ctx.self_test.status();
let response = SelfTestStatusResponse {
enabled: status.enabled,
schedule: status.schedule,
interval: status.interval,
last_run: status.last_run,
next_run: status.next_run,
};
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::SelfTestHistory { limit }) => {
metrics::inc_requests("self-test-history");
let runs = ctx.self_test.history().recent_runs(limit);
let run_ids: Vec<u64> = runs.iter().map(|run| run.id).collect();
let results = ctx.self_test.history().results_for_runs(&run_ids);
let response = SelfTestHistoryResponse { runs, results };
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::SelfTestRun(request)) => {
metrics::inc_requests("self-test-run");
let mut options = crate::self_test::SelfTestRunOptions {
run_type: if request.scheduled {
crate::self_test::SelfTestRunType::Scheduled
} else {
crate::self_test::SelfTestRunType::Manual
},
..Default::default()
};
if !request.worker_ids.is_empty() {
options.worker_ids =
Some(request.worker_ids.into_iter().map(WorkerId::new).collect());
}
if let Some(project) = request.project {
options.project_path = Some(PathBuf::from(project));
}
if let Some(timeout) = request.timeout_secs {
options.timeout = Duration::from_secs(timeout);
}
options.release_mode = request.release_mode;
options.retry_count_override = request.retries;
let report = if request.scheduled {
ctx.self_test.run_scheduled_now().await?
} else {
ctx.self_test.run_manual(options).await?
};
let response = SelfTestRunResponse {
run: report.run,
results: report.results,
};
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::Shutdown) => {
metrics::inc_requests("shutdown");
let admission = ctx.admission_barrier.read().await;
let active_build_ids: Vec<_> = ctx
.history
.active_builds()
.into_iter()
.map(|build| build.id)
.collect();
let queued_build_ids: Vec<_> = ctx
.history
.queued_builds()
.into_iter()
.map(|build| build.id)
.collect();
if !*admission || !active_build_ids.is_empty() || !queued_build_ids.is_empty() {
(
serde_json::json!({
"status": "shutdown_blocked",
"admission_closed": *admission,
"active_build_ids": active_build_ids,
"queued_build_ids": queued_build_ids,
})
.to_string(),
"application/json",
)
} else {
let _ = shutdown_tx.send(()).await;
(
"{\"status\":\"shutting_down\"}".to_string(),
"application/json",
)
}
}
Ok(ApiRequest::RestartAdmission { close }) => {
metrics::inc_requests("restart-admission");
let response = handle_restart_admission(&ctx, close).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::RestartAdmissionStatus) => {
metrics::inc_requests("restart-admission-status");
let response = restart_admission_status(&ctx).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::Reload) => {
metrics::inc_requests("reload");
let result =
reload::reload_workers(&ctx.pool, ctx.workers_config_path.as_deref(), true).await;
match result {
Ok(reload_result) => {
let response = serde_json::json!({
"success": true,
"added": reload_result.added,
"updated": reload_result.updated,
"removed": reload_result.removed,
"warnings": reload_result.warnings
});
(response.to_string(), "application/json")
}
Err(e) => {
warn!("Configuration reload failed: {}", e);
let response = serde_json::json!({
"success": false,
"error": e.to_string()
});
(response.to_string(), "application/json")
}
}
}
Ok(ApiRequest::BuildRecovery {
build_id,
local_wrapper_id,
}) => {
let response = if let Some(active) = ctx.history.active_build(build_id) {
if active.local_wrapper_id.as_deref() == Some(local_wrapper_id.as_str()) {
serde_json::json!({"status":"active","active":active})
} else {
serde_json::json!({"status":"identity_mismatch"})
}
} else if let Some(record) = ctx.history.terminal_build(build_id, &local_wrapper_id) {
serde_json::json!({"status":"completed","record":record,"local_wrapper_id":local_wrapper_id})
} else if ctx.history.has_terminal_build(build_id) {
serde_json::json!({"status":"identity_mismatch"})
} else {
serde_json::json!({"status":"not_found"})
};
(response.to_string(), "application/json")
}
Ok(ApiRequest::CancelJob { local_wrapper_id }) => {
let response = handle_cancel_job(&ctx, &local_wrapper_id, false).await?;
(response.to_string(), "application/json")
}
Ok(ApiRequest::CancelBuild {
build_id,
force,
local_wrapper_id,
}) => {
metrics::inc_requests("cancel-build");
if let Some(queued) = ctx.history.queued_build(build_id).filter(|_| {
ctx.history.active_build(build_id).is_none()
&& !ctx.history.has_terminal_build(build_id)
}) {
let response = match queued.local_wrapper_id.as_deref() {
Some(wrapper)
if local_wrapper_id
.as_deref()
.is_none_or(|requested| requested == wrapper) =>
{
let mut response = handle_cancel_job(&ctx, wrapper, force).await?;
response["queue_id"] = serde_json::json!(build_id);
if response["status"] == "cancelled_before_start" {
response["status"] = serde_json::json!("cancelled");
response["build_id"] = serde_json::json!(build_id);
response["message"] =
serde_json::json!("Queued job cancelled before start");
}
response
}
_ => {
serde_json::json!({"status":"error", "build_id":build_id, "slots_released":0, "message":"Queued job has no matching durable wrapper identity; cancellation refused"})
}
};
(response.to_string(), "application/json")
} else if let Some(record) = local_wrapper_id
.as_deref()
.and_then(|wrapper| ctx.history.terminal_build(build_id, wrapper))
{
(serde_json::json!({"status":"completed","record":record,"local_wrapper_id":local_wrapper_id}).to_string(), "application/json")
} else {
let owner = ctx
.history
.active_build(build_id)
.filter(|state| state.local_wrapper_id != local_wrapper_id);
let response = if owner.is_some() || ctx.history.has_terminal_build(build_id) {
CancelBuildResponse {
status: "error".to_string(),
build_id,
worker_id: None,
project_id: None,
message: Some(ownership_mismatch_message(
build_id,
owner
.as_ref()
.map(|state| state.local_wrapper_id.as_deref()),
)),
slots_released: 0,
}
} else {
handle_cancel_build(&ctx, build_id, force).await
};
(serde_json::to_string(&response)?, "application/json")
}
}
Ok(ApiRequest::CancelAllBuilds { force }) => {
metrics::inc_requests("cancel-all-builds");
let response = handle_cancel_all_builds(&ctx, force).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::WorkerDrain { worker_id }) => {
metrics::inc_requests("worker-drain");
let response = handle_worker_drain(&ctx, &worker_id).await;
log_worker_admin_action("drain", &worker_id, &response.status, peer_pid).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::WorkerEnable { worker_id }) => {
metrics::inc_requests("worker-enable");
let response = handle_worker_enable(&ctx, &worker_id).await;
log_worker_admin_action("enable", &worker_id, &response.status, peer_pid).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::WorkerDisable {
worker_id,
reason,
drain_first,
}) => {
metrics::inc_requests("worker-disable");
let response = handle_worker_disable(&ctx, &worker_id, reason, drain_first).await;
log_worker_admin_action("disable", &worker_id, &response.status, peer_pid).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::RepoConvergenceStatus { worker_id }) => {
metrics::inc_requests("repo-convergence-status");
let response = handle_repo_convergence_status(&ctx, worker_id.as_ref()).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::RepoConvergenceDryRun { worker_id }) => {
metrics::inc_requests("repo-convergence-dry-run");
let response = handle_repo_convergence_dry_run(&ctx, &worker_id).await;
(serde_json::to_string(&response)?, "application/json")
}
Ok(ApiRequest::RepoConvergenceRepair { worker_id }) => {
metrics::inc_requests("repo-convergence-repair");
let response = handle_repo_convergence_repair(&ctx, &worker_id).await;
(serde_json::to_string(&response)?, "application/json")
}
Err(e) => return Err(e),
};
let response_bytes = format!(
"HTTP/1.0 200 OK\r\nContent-Type: {}\r\n\r\n{}\n",
content_type, response_json
);
writer.write_all(response_bytes.as_bytes()).await?;
writer.flush().await?;
Ok(())
}
fn parse_request(line: &str) -> Result<ApiRequest> {
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() < 2 {
return Err(anyhow!("Invalid request format"));
}
let method = parts[0];
let path = parts[1];
if method != "GET" && method != "POST" {
return Err(anyhow!("Only GET and POST methods supported"));
}
if path == "/events" && method == "GET" {
return Ok(ApiRequest::Events);
}
if path == "/speedscores" && method == "GET" {
return Ok(ApiRequest::SpeedScores);
}
if path.starts_with("/workers/capabilities") && method == "GET" {
let (path_only, query) = split_path_query(path);
if path_only == "/workers/capabilities" {
let mut refresh = false;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "refresh" && (value == "1" || value.eq_ignore_ascii_case("true")) {
refresh = true;
}
}
return Ok(ApiRequest::WorkersCapabilities { refresh });
}
}
if path.starts_with("/speedscore") && method == "GET" {
let (path_only, query) = split_path_query(path);
if path_only == "/speedscore/history" {
let mut worker_id = None;
let mut days = 30u32;
let mut limit = 100usize;
let mut offset = 0usize;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"worker" => worker_id = Some(percent_unescape_query_value(value)),
"days" => days = value.parse().unwrap_or(days),
"limit" => limit = value.parse().unwrap_or(limit).min(10_000),
"offset" => offset = value.parse().unwrap_or(offset).min(1_000_000),
_ => {}
}
}
if let Some(worker_id) = worker_id {
return Ok(ApiRequest::SpeedScoreHistory {
worker_id: WorkerId::new(worker_id),
days,
limit,
offset,
});
}
}
if let Some(rest) = path_only.strip_prefix("/speedscore/") {
let rest = rest.trim_matches('/');
if rest.is_empty() {
return Err(anyhow!("Missing worker id"));
}
if let Some(worker_part) = rest.strip_suffix("/history") {
let worker_part = worker_part.trim_end_matches('/');
let mut days = 30u32;
let mut limit = 100usize;
let mut offset = 0usize;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"days" => days = value.parse().unwrap_or(days),
"limit" => limit = value.parse().unwrap_or(limit).min(10_000),
"offset" => offset = value.parse().unwrap_or(offset).min(1_000_000),
_ => {}
}
}
return Ok(ApiRequest::SpeedScoreHistory {
worker_id: WorkerId::new(percent_unescape_query_value(worker_part)),
days,
limit,
offset,
});
}
return Ok(ApiRequest::SpeedScore {
worker_id: WorkerId::new(percent_unescape_query_value(rest)),
});
}
}
if route_matches_exact_or_child(path, "/benchmark/trigger") {
if method != "POST" {
return Err(anyhow!("Only POST method supported for benchmark trigger"));
}
let (path_only, query) = split_path_query(path);
let mut worker_id = path_only
.strip_prefix("/benchmark/trigger/")
.map(percent_unescape_query_value);
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "worker" {
worker_id = Some(percent_unescape_query_value(value));
}
}
let Some(worker_id) = worker_id.filter(|value| !value.is_empty()) else {
return Err(anyhow!("Missing worker id"));
};
return Ok(ApiRequest::BenchmarkTrigger {
worker_id: WorkerId::new(worker_id),
});
}
if path == "/shutdown" && method == "POST" {
return Ok(ApiRequest::Shutdown);
}
if path == "/restart-admission" && method == "POST" {
return Ok(ApiRequest::RestartAdmission { close: true });
}
if path == "/restart-admission" && method == "GET" {
return Ok(ApiRequest::RestartAdmissionStatus);
}
if path == "/restart-admission/release" && method == "POST" {
return Ok(ApiRequest::RestartAdmission { close: false });
}
if path == "/reload" && method == "POST" {
return Ok(ApiRequest::Reload);
}
if method == "POST"
&& let Some(wrapper) = path
.strip_prefix("/jobs/")
.and_then(|rest| rest.strip_suffix("/cancel"))
{
let uuid = wrapper
.strip_prefix(LOCAL_WRAPPER_ID_PREFIX)
.ok_or_else(|| anyhow!("Invalid local wrapper id"))?;
Uuid::parse_str(uuid).map_err(|_| anyhow!("Invalid local wrapper id"))?;
return Ok(ApiRequest::CancelJob {
local_wrapper_id: wrapper.to_owned(),
});
}
if method == "GET" && path.starts_with("/builds/") {
let (route, query) = path.split_once('?').unwrap_or((path, ""));
let build_id = route
.trim_start_matches("/builds/")
.trim_end_matches('/')
.parse::<u64>()
.map_err(|_| anyhow!("Invalid build id"))?;
let local_wrapper_id = query
.split('&')
.find_map(|pair| pair.strip_prefix("local_wrapper_id="))
.map(percent_unescape_query_value)
.filter(|id| !id.is_empty())
.ok_or_else(|| anyhow!("Missing local_wrapper_id"))?;
return Ok(ApiRequest::BuildRecovery {
build_id,
local_wrapper_id,
});
}
if method == "POST" && path.starts_with("/builds") {
let (path_only, query) = split_path_query(path);
if path_only == "/builds/cancel-all" {
let mut force = false;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "force" {
force = value == "1" || value.eq_ignore_ascii_case("true");
}
}
return Ok(ApiRequest::CancelAllBuilds { force });
}
if let Some(rest) = path_only.strip_prefix("/builds/") {
let rest = rest.trim_matches('/');
let parts: Vec<&str> = rest.split('/').collect();
if parts.len() == 2 && parts[1] == "cancel" {
let build_id = parts[0]
.parse::<u64>()
.map_err(|_| anyhow!("Invalid build id: {}", parts[0]))?;
let mut force = false;
let mut local_wrapper_id = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "force" {
force = value == "1" || value.eq_ignore_ascii_case("true");
} else if key == "local_wrapper_id" {
local_wrapper_id = Some(percent_unescape_query_value(value));
}
}
return Ok(ApiRequest::CancelBuild {
build_id,
force,
local_wrapper_id,
});
}
}
}
if path == "/status" {
return Ok(ApiRequest::Status);
}
if path == "/metrics" {
return Ok(ApiRequest::Metrics);
}
if path == "/health" {
return Ok(ApiRequest::Health);
}
if path == "/ready" {
return Ok(ApiRequest::Ready);
}
if path == "/budget" {
return Ok(ApiRequest::Budget);
}
if path == "/self-test/status" {
return Ok(ApiRequest::SelfTestStatus);
}
if let Some(query) = query_for_exact_route(path, "/self-test/history") {
let mut limit = 10usize;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "limit" {
limit = value.parse().unwrap_or(limit).min(10_000);
}
}
return Ok(ApiRequest::SelfTestHistory { limit });
}
if let Some(query) = query_for_exact_route(path, "/self-test/run") {
if method != "POST" {
return Err(anyhow!("Only POST method supported for self-test run"));
}
let mut worker_ids = Vec::new();
let mut project = None;
let mut timeout_secs = None;
let mut release_mode = true;
let mut scheduled = false;
let mut retries = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"worker" => worker_ids.push(percent_unescape_query_value(value)),
"project" => project = Some(percent_unescape_query_value(value)),
"timeout" => timeout_secs = value.parse().ok(),
"retries" => retries = value.parse().ok(),
"debug" if value == "1" || value.eq_ignore_ascii_case("true") => {
release_mode = false;
}
"scheduled" if value == "1" || value.eq_ignore_ascii_case("true") => {
scheduled = true;
}
"all" if value == "1" || value.eq_ignore_ascii_case("true") => {
worker_ids.clear();
}
_ => {}
}
}
return Ok(ApiRequest::SelfTestRun(SelfTestRunRequest {
worker_ids,
project,
timeout_secs,
release_mode,
scheduled,
retries,
}));
}
if let Some(query) = query_for_exact_route(path, "/release-worker") {
if method != "POST" {
return Err(anyhow!("Only POST method supported for release"));
}
let mut worker_id = None;
let mut slots = None;
let mut build_id = None;
let mut exit_code = None;
let mut duration_ms = None;
let mut bytes_transferred = None;
let mut local_wrapper_id = None;
let mut worker_fault = false;
let mut worker_disk_full = false;
let mut worker_disk_roots = Vec::new();
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"worker" => worker_id = Some(percent_unescape_query_value(value)),
"slots" => slots = value.parse().ok(),
"build_id" => build_id = value.parse().ok(),
"exit_code" => exit_code = value.parse().ok(),
"duration_ms" => duration_ms = value.parse().ok(),
"bytes_transferred" => bytes_transferred = value.parse().ok(),
"local_wrapper_id" => local_wrapper_id = Some(percent_unescape_query_value(value)),
"worker_fault" => worker_fault = matches!(value, "1" | "true"),
"worker_disk_full" => worker_disk_full = matches!(value, "1" | "true"),
"worker_disk_roots" => {
worker_disk_roots =
serde_json::from_str::<Vec<String>>(&percent_unescape_query_value(value))
.context("Invalid worker disk roots")?;
}
_ => {} }
}
let worker_id = worker_id.ok_or_else(|| anyhow!("Missing 'worker' parameter"))?;
let slots = slots.unwrap_or(0);
return Ok(ApiRequest::ReleaseWorker(ReleaseRequest {
worker_id: rch_common::WorkerId::new(worker_id),
slots,
build_id,
exit_code,
duration_ms,
bytes_transferred,
local_wrapper_id,
timing: None,
worker_fault,
worker_disk_full,
worker_disk_roots,
}));
}
if let Some(query) = query_for_exact_route(path, "/record-build") {
if method != "POST" {
return Err(anyhow!("Only POST method supported for record-build"));
}
let mut worker_id = None;
let mut project = None;
let mut is_test = false;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"worker" => worker_id = Some(percent_unescape_query_value(value)),
"project" => project = Some(percent_unescape_query_value(value)),
"is_test" => {
let query_value = percent_unescape_query_value(value);
is_test = matches!(query_value.as_str(), "1" | "true" | "yes" | "y" | "on");
}
_ => {}
}
}
let worker_id = worker_id.ok_or_else(|| anyhow!("Missing 'worker' parameter"))?;
let project = project.ok_or_else(|| anyhow!("Missing 'project' parameter"))?;
return Ok(ApiRequest::RecordBuild {
worker_id: WorkerId::new(worker_id),
project,
is_test,
});
}
if query_for_exact_route(path, "/build-heartbeat").is_some() {
if method != "POST" {
return Err(anyhow!("Only POST method supported for build heartbeat"));
}
return Ok(ApiRequest::BuildHeartbeat);
}
if query_for_exact_route(path, "/test-run").is_some() {
if method != "POST" {
return Err(anyhow!("Only POST method supported for test run"));
}
return Ok(ApiRequest::TestRun);
}
if path.starts_with("/telemetry") {
let (path_only, query) = split_path_query(path);
return match path_only {
"/telemetry/poll" => {
if method != "POST" {
return Err(anyhow!("Only POST method supported for telemetry polling"));
}
let mut worker_id = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "worker" {
worker_id = Some(percent_unescape_query_value(value));
}
}
let worker_id = worker_id.ok_or_else(|| anyhow!("Missing 'worker' parameter"))?;
Ok(ApiRequest::TelemetryPoll {
worker_id: WorkerId::new(worker_id),
})
}
"/telemetry/ingest" => {
if method != "POST" {
return Err(anyhow!(
"Only POST method supported for telemetry ingestion"
));
}
let mut source = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "source" {
source = parse_telemetry_source(&percent_unescape_query_value(value));
}
}
Ok(ApiRequest::IngestTelemetry(
source.unwrap_or(TelemetrySource::Piggyback),
))
}
_ => Err(anyhow!("Unknown endpoint: {}", path_only)),
};
}
if path.starts_with("/workers/") && method == "POST" {
let rest = path.strip_prefix("/workers/").unwrap_or("");
let (path_part, query) = split_path_query(rest);
let parts: Vec<&str> = path_part.split('/').collect();
if parts.len() == 2 {
let worker_id = percent_unescape_query_value(parts[0]);
let action = parts[1];
match action {
"drain" => {
return Ok(ApiRequest::WorkerDrain {
worker_id: WorkerId::new(worker_id),
});
}
"enable" => {
return Ok(ApiRequest::WorkerEnable {
worker_id: WorkerId::new(worker_id),
});
}
"disable" => {
let mut reason = None;
let mut drain_first = false;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"reason" => reason = Some(percent_unescape_query_value(value)),
"drain" => {
drain_first = value == "1" || value.eq_ignore_ascii_case("true")
}
_ => {}
}
}
return Ok(ApiRequest::WorkerDisable {
worker_id: WorkerId::new(worker_id),
reason,
drain_first,
});
}
_ => {}
}
}
}
if path.starts_with("/repo-convergence") {
let (path_only, query) = split_path_query(path);
match path_only {
"/repo-convergence/status" => {
let mut worker_id = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "worker" {
worker_id = Some(WorkerId::new(percent_unescape_query_value(value)));
}
}
return Ok(ApiRequest::RepoConvergenceStatus { worker_id });
}
"/repo-convergence/dry-run" => {
let mut worker_id = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "worker" {
worker_id = Some(WorkerId::new(percent_unescape_query_value(value)));
}
}
return match worker_id {
Some(id) => Ok(ApiRequest::RepoConvergenceDryRun { worker_id: id }),
None => Err(anyhow!("Missing required 'worker' parameter for dry-run")),
};
}
"/repo-convergence/repair" => {
if method != "POST" {
return Err(anyhow!("Repair endpoint requires POST method"));
}
let mut worker_id = None;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
if key == "worker" {
worker_id = Some(WorkerId::new(percent_unescape_query_value(value)));
}
}
return match worker_id {
Some(id) => Ok(ApiRequest::RepoConvergenceRepair { worker_id: id }),
None => Err(anyhow!("Missing required 'worker' parameter for repair")),
};
}
_ => return Err(anyhow!("Unknown convergence endpoint: {}", path_only)),
}
}
let Some((query, resume_queued, disk_budget_route, tools_route)) = [
("/select-worker", false, false, false),
("/select-worker/resume-queued", true, false, false),
("/select-worker/disk-budget", false, true, false),
(
"/select-worker/resume-queued/disk-budget",
true,
true,
false,
),
("/select-verified-tools", false, false, true),
("/select-verified-tools/resume-queued", true, false, true),
("/select-verified-tools/disk-budget", false, true, true),
(
"/select-verified-tools/resume-queued/disk-budget",
true,
true,
true,
),
]
.into_iter()
.find_map(|(route, resume, disk_budget, tools)| {
query_for_exact_route(path, route).map(|query| (query, resume, disk_budget, tools))
}) else {
return Err(anyhow!("Unknown endpoint: {}", path));
};
let mut project = None;
let mut command = None;
let mut cores = None;
let mut disk_headroom_gib = None;
let mut wait_for_worker = false;
let mut job_mode = false;
let mut wait_timeout_secs = None;
let mut toolchain = None;
let mut required_runtime = RequiredRuntime::default();
let mut command_priority = CommandPriority::Normal;
let mut classification_duration_us = None;
let mut hook_pid = None;
let mut local_wrapper_id = None;
let mut preferred_workers = Vec::new();
let mut required_tools: Vec<String> = Vec::new();
let mut dry_run = false;
for param in query.split('&') {
if param.is_empty() {
continue;
}
let mut kv = param.splitn(2, '=');
let key = kv.next().unwrap_or("");
let value = kv.next().unwrap_or("");
match key {
"project" => project = Some(percent_unescape_query_value(value)),
"command" => command = Some(percent_unescape_query_value(value)),
"cores" => cores = value.parse::<u32>().ok().filter(|cores| *cores > 0),
"disk_headroom_gib" => {
anyhow::ensure!(
disk_headroom_gib.is_none(),
"Duplicate disk_headroom_gib parameter"
);
disk_headroom_gib = Some(
value
.parse::<u32>()
.context("disk_headroom_gib must be an unsigned whole GiB count")?,
);
}
"wait" | "queue" => {
wait_for_worker = value == "1" || value.eq_ignore_ascii_case("true");
}
"wait_timeout_secs" => {
wait_timeout_secs = value.parse::<u64>().ok().filter(|secs| *secs > 0);
}
"toolchain" => {
let json = percent_unescape_query_value(value);
toolchain = serde_json::from_str(&json).ok();
}
"runtime" => {
let rt_str = percent_unescape_query_value(value);
required_runtime = serde_json::from_str(&format!("\"{}\"", rt_str))
.ok()
.unwrap_or_default();
}
"priority" => {
let pr_str = percent_unescape_query_value(value);
command_priority = pr_str.parse().unwrap_or(CommandPriority::Normal);
}
"job_mode" => {
job_mode = value == "1" || value.eq_ignore_ascii_case("true");
}
"require_tool" => {
let name = percent_unescape_query_value(value);
let name = name.trim();
if !name.is_empty() && !required_tools.iter().any(|t| t == name) {
required_tools.push(name.to_string());
}
}
"classification_us" => {
classification_duration_us = value.parse().ok();
}
"hook_pid" => {
hook_pid = value.parse().ok();
}
"dry_run" => {
dry_run = value == "1" || value == "true";
}
"local_wrapper_id" => {
let candidate = percent_unescape_query_value(value);
if candidate.starts_with(LOCAL_WRAPPER_ID_PREFIX) {
local_wrapper_id = Some(candidate);
}
}
"worker" | "preferred_worker" | "preferred" => {
preferred_workers.extend(parse_worker_id_list(value));
}
"workers" | "preferred_workers" => {
preferred_workers.extend(parse_worker_id_list(value));
}
_ => {} }
}
let project = project.ok_or_else(|| anyhow!("Missing 'project' parameter"))?;
let estimated_cores = cores.unwrap_or(1);
let disk_headroom_gib = disk_headroom_gib.unwrap_or(0);
anyhow::ensure!(
disk_budget_route == (disk_headroom_gib > 0),
"A positive disk_headroom_gib requires the exact disk-budget selection route"
);
anyhow::ensure!(
!tools_route || !required_tools.is_empty(),
"The verified-tools selection route requires explicit tool requirements"
);
if resume_queued
&& (dry_run || !wait_for_worker || local_wrapper_id.is_none() || hook_pid.is_none())
{
return Err(anyhow!(
"queued resume requires the original waiting wrapper and process"
));
}
Ok(ApiRequest::SelectWorker {
request: SelectionRequest {
project,
command,
command_priority,
estimated_cores,
disk_headroom_gib,
preferred_workers,
toolchain,
required_runtime,
classification_duration_us,
hook_pid,
job_mode,
required_tools,
},
wait_for_worker,
wait_timeout_secs,
local_wrapper_id,
dry_run,
resume_queued,
})
}
fn parse_worker_id_list(encoded_value: &str) -> Vec<WorkerId> {
percent_unescape_query_value(encoded_value)
.split(',')
.map(str::trim)
.filter(|id| !id.is_empty())
.map(WorkerId::new)
.collect()
}
fn query_for_exact_route<'a>(path: &'a str, expected_path: &str) -> Option<&'a str> {
let (path_only, query) = split_path_query(path);
(path_only == expected_path).then_some(query)
}
fn route_matches_exact_or_child(path: &str, expected_path: &str) -> bool {
let (path_only, _) = split_path_query(path);
path_only == expected_path
|| path_only
.strip_prefix(expected_path)
.is_some_and(|rest| rest.starts_with('/'))
}
fn parse_telemetry_source(value: &str) -> Option<TelemetrySource> {
match value.trim().to_lowercase().as_str() {
"piggyback" => Some(TelemetrySource::Piggyback),
"ssh-poll" | "ssh_poll" | "ssh" => Some(TelemetrySource::SshPoll),
"on-demand" | "on_demand" | "ondemand" => Some(TelemetrySource::OnDemand),
_ => None,
}
}
fn split_path_query(path: &str) -> (&str, &str) {
match path.split_once('?') {
Some((path, query)) => (path, query),
None => (path, ""),
}
}
fn percent_unescape_query_value(s: &str) -> String {
let mut bytes: Vec<u8> = Vec::with_capacity(s.len());
let mut chars = s.chars().peekable();
while let Some(c) = chars.next() {
if c == '%' {
let hex: String = chars.by_ref().take(2).collect();
if hex.len() == 2
&& let Ok(byte) = u8::from_str_radix(&hex, 16)
{
bytes.push(byte);
continue;
}
bytes.push(b'%');
bytes.extend_from_slice(hex.as_bytes());
} else if c == '+' {
bytes.push(b' ');
} else {
let mut buf = [0; 4];
let s = c.encode_utf8(&mut buf);
bytes.extend_from_slice(s.as_bytes());
}
}
String::from_utf8_lossy(&bytes).into_owned()
}
#[derive(Debug, Serialize)]
struct TelemetryPollResponse {
status: String,
#[serde(skip_serializing_if = "Option::is_none")]
telemetry: Option<WorkerTelemetry>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
worker_id: Option<String>,
}
async fn handle_telemetry_poll(ctx: &DaemonContext, worker_id: &WorkerId) -> TelemetryPollResponse {
let worker = match ctx.pool.get(worker_id).await {
Some(worker) => worker,
None => {
return TelemetryPollResponse {
status: "error".to_string(),
telemetry: None,
error: Some("worker not found".to_string()),
worker_id: Some(worker_id.to_string()),
};
}
};
let status = worker.status().await;
if matches!(
status,
WorkerStatus::Unreachable | WorkerStatus::Drained | WorkerStatus::Disabled
) {
return TelemetryPollResponse {
status: "error".to_string(),
telemetry: None,
error: Some("worker unavailable".to_string()),
worker_id: Some(worker_id.to_string()),
};
}
match collect_telemetry_from_worker(&worker, Duration::from_secs(20)).await {
Ok((endpoint, telemetry)) => {
let Some(_endpoint_guard) = worker.lock_current_endpoint(&endpoint).await else {
return TelemetryPollResponse {
status: "error".to_string(),
telemetry: None,
error: Some("worker endpoint changed during telemetry collection".to_string()),
worker_id: Some(worker_id.to_string()),
};
};
ctx.telemetry.ingest_for_endpoint(
telemetry.clone(),
TelemetrySource::OnDemand,
&endpoint,
);
TelemetryPollResponse {
status: "ok".to_string(),
telemetry: Some(telemetry),
error: None,
worker_id: None,
}
}
Err(e) => TelemetryPollResponse {
status: "error".to_string(),
telemetry: None,
error: Some(e.to_string()),
worker_id: Some(worker_id.to_string()),
},
}
}
fn speedscore_view(score: SpeedScore) -> SpeedScoreView {
SpeedScoreView {
total: score.total,
cpu_score: score.cpu_score,
memory_score: score.memory_score,
disk_score: score.disk_score,
network_score: score.network_score,
compilation_score: score.compilation_score,
measured_at: score.calculated_at.to_rfc3339(),
version: score.version,
}
}
fn error_response(
code: ErrorCode,
message: impl Into<String>,
worker_id: Option<&WorkerId>,
retry_after: Option<ChronoDuration>,
) -> ApiError {
let mut error = ApiError::new(code, message);
if let Some(id) = worker_id {
error = error.with_context("worker_id", id.as_str());
}
if let Some(duration) = retry_after {
error = error.with_retry_after(duration.num_seconds().max(1) as u64);
}
error
}
async fn handle_speedscore(
ctx: &DaemonContext,
worker_id: &WorkerId,
) -> ApiResponse<SpeedScoreResponse> {
if ctx.pool.get(worker_id).await.is_none() {
return ApiResponse::Error(error_response(
ErrorCode::ConfigInvalidWorker,
format!("Worker '{}' not found", worker_id),
Some(worker_id),
None,
));
}
match ctx.telemetry.latest_speedscore(worker_id.as_str()).await {
Ok(Some(score)) => ApiResponse::Ok(SpeedScoreResponse {
worker_id: worker_id.to_string(),
speedscore: Some(speedscore_view(score)),
message: None,
}),
Ok(None) => ApiResponse::Ok(SpeedScoreResponse {
worker_id: worker_id.to_string(),
speedscore: None,
message: Some("Worker has not been benchmarked yet".to_string()),
}),
Err(err) => {
warn!("Failed to load SpeedScore for {}: {}", worker_id, err);
ApiResponse::Error(error_response(
ErrorCode::InternalStateError,
"Failed to retrieve SpeedScore",
Some(worker_id),
None,
))
}
}
}
async fn handle_speedscore_history(
ctx: &DaemonContext,
worker_id: &WorkerId,
days: u32,
limit: usize,
offset: usize,
) -> ApiResponse<SpeedScoreHistoryResponse> {
if ctx.pool.get(worker_id).await.is_none() {
return ApiResponse::Error(error_response(
ErrorCode::ConfigInvalidWorker,
format!("Worker '{}' not found", worker_id),
Some(worker_id),
None,
));
}
let days = days.clamp(1, 365);
let limit = limit.clamp(1, 1000);
let since = Utc::now() - ChronoDuration::days(days as i64);
match ctx
.telemetry
.speedscore_history(worker_id.as_str(), since, limit, offset)
.await
{
Ok(page) => {
let has_more = ((offset + page.entries.len()) as u64) < page.total;
ApiResponse::Ok(SpeedScoreHistoryResponse {
worker_id: worker_id.to_string(),
history: page.entries.into_iter().map(speedscore_view).collect(),
pagination: PaginationInfo {
total: page.total,
offset,
limit,
has_more,
},
})
}
Err(err) => {
warn!(
"Failed to load SpeedScore history for {}: {}",
worker_id, err
);
ApiResponse::Error(error_response(
ErrorCode::InternalStateError,
"Failed to retrieve SpeedScore history",
Some(worker_id),
None,
))
}
}
}
async fn handle_speedscore_list(ctx: &DaemonContext) -> ApiResponse<SpeedScoreListResponse> {
let workers = ctx.pool.all_workers().await;
let mut entries = Vec::with_capacity(workers.len());
for worker in workers {
let status = worker.status().await;
let worker_id = worker.config.read().await.id.clone();
let speedscore = match ctx.telemetry.latest_speedscore(worker_id.as_str()).await {
Ok(score) => score.map(speedscore_view),
Err(err) => {
warn!("Failed to load SpeedScore for {}: {}", worker_id, err);
None
}
};
entries.push(SpeedScoreWorker {
worker_id: worker_id.to_string(),
speedscore,
status,
});
}
ApiResponse::Ok(SpeedScoreListResponse { workers: entries })
}
async fn handle_workers_capabilities(
ctx: &DaemonContext,
refresh: bool,
) -> ApiResponse<WorkerCapabilitiesResponse> {
ApiResponse::Ok(get_workers_capabilities(ctx, refresh).await)
}
async fn handle_benchmark_trigger(
ctx: &DaemonContext,
worker_id: &WorkerId,
) -> ApiResponse<BenchmarkTriggerResponse> {
if ctx.pool.get(worker_id).await.is_none() {
return ApiResponse::Error(error_response(
ErrorCode::ConfigInvalidWorker,
format!("Worker '{}' not found", worker_id),
Some(worker_id),
None,
));
}
let request_id = Uuid::new_v4().to_string();
match ctx
.benchmark_queue
.enqueue(worker_id.clone(), request_id.clone())
{
Ok(request) => {
if let Err(err) = ctx
.benchmark_trigger
.trigger(worker_id.clone(), request.request_id.clone(), None)
.await
{
warn!(
worker = %worker_id,
request_id = %request.request_id,
"Benchmark trigger dispatch failed: {}",
err
);
let _ = ctx.benchmark_queue.pop();
return ApiResponse::Error(error_response(
ErrorCode::InternalStateError,
"Benchmark scheduler unavailable",
Some(worker_id),
None,
));
}
let _ = ctx.benchmark_queue.pop();
ctx.events.emit(
"benchmark_queued",
&serde_json::json!({
"worker_id": worker_id.as_str(),
"request_id": request.request_id,
"queued_at": request.requested_at.to_rfc3339(),
}),
);
ApiResponse::Ok(BenchmarkTriggerResponse {
status: "queued".to_string(),
worker_id: worker_id.to_string(),
request_id,
})
}
Err(rate) => ApiResponse::Error(error_response(
ErrorCode::WorkerAtCapacity,
"Benchmark trigger rate limited",
Some(worker_id),
Some(rate.retry_after),
)),
}
}
async fn handle_event_stream(
writer: &mut tokio::net::unix::OwnedWriteHalf,
events: EventBus,
) -> Result<()> {
let mut rx = events.subscribe();
let header = "HTTP/1.0 200 OK\r\nContent-Type: application/json\r\n\r\n";
writer.write_all(header.as_bytes()).await?;
writer.flush().await?;
loop {
match rx.recv().await {
Ok(message) => {
if let Err(err) = writer.write_all(message.as_bytes()).await {
warn!("Event stream write failed: {}", err);
break;
}
if let Err(err) = writer.write_all(b"\n").await {
warn!("Event stream write failed: {}", err);
break;
}
if let Err(err) = writer.flush().await {
warn!("Event stream flush failed: {}", err);
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
warn!("Event stream lagged, skipped {} messages", skipped);
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
Ok(())
}
fn cancelled_selection() -> SelectionResponse {
SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError("job_cancelled_before_start".to_owned()),
build_id: None,
diagnostics: None,
}
}
async fn handle_select_worker_dry_run(
ctx: &DaemonContext,
request: &SelectionRequest,
) -> SelectionResponse {
let excluded_worker_ids = ctx.history.active_workers_for_project(&request.project);
let result = ctx
.worker_selector
.preview_with_exclusions(&ctx.pool, request, &excluded_worker_ids)
.await;
let worker = match result.worker {
Some(worker) => {
let slots_available = worker.available_slots().await;
let speed_score = worker.get_speed_score();
let config = worker.config.read().await;
Some(SelectedWorker {
id: config.id.clone(),
host: config.host.clone(),
user: config.user.clone(),
identity_file: config.identity_file.clone(),
slots_available,
speed_score,
declared_os: rch_common::declared_os(&config.tags),
})
}
None => None,
};
SelectionResponse {
worker,
reason: result.reason,
build_id: None,
diagnostics: result.diagnostics,
}
}
#[cfg(test)]
async fn handle_select_worker(
ctx: &DaemonContext,
request: SelectionRequest,
wait_for_worker: bool,
wait_timeout_secs: Option<u64>,
) -> Result<SelectionResponse> {
handle_select_worker_with_wrapper(ctx, request, wait_for_worker, wait_timeout_secs, None).await
}
#[cfg(test)]
async fn handle_select_worker_with_wrapper(
ctx: &DaemonContext,
request: SelectionRequest,
wait_for_worker: bool,
wait_timeout_secs: Option<u64>,
local_wrapper_id: Option<String>,
) -> Result<SelectionResponse> {
handle_select_worker_mode(
ctx,
request,
wait_for_worker,
wait_timeout_secs,
local_wrapper_id,
false,
)
.await
}
fn queue_selection_digest(
request: &SelectionRequest,
wait_timeout_secs: Option<u64>,
) -> Result<[u8; 32]> {
let mut digest = Sha256::new();
digest.update(b"rch-queued-selection-v1\0");
digest.update(serde_json::to_vec(&(request, wait_timeout_secs))?);
Ok(digest.finalize().into())
}
async fn handle_select_worker_mode(
ctx: &DaemonContext,
request: SelectionRequest,
wait_for_worker: bool,
wait_timeout_secs: Option<u64>,
local_wrapper_id: Option<String>,
resume_queued: bool,
) -> Result<SelectionResponse> {
let resumed = if resume_queued {
anyhow::ensure!(wait_for_worker, "queued resume requires waiting selection");
anyhow::ensure!(
!*ctx.admission_barrier.read().await,
"restart admission barrier active"
);
let wrapper = local_wrapper_id
.as_deref()
.ok_or_else(|| anyhow!("missing queued owner"))?;
Some(
ctx.history.resume_queued_build(
wrapper,
request
.hook_pid
.ok_or_else(|| anyhow!("missing queued process"))?,
&queue_selection_digest(&request, wait_timeout_secs)?,
)?,
)
} else {
anyhow::ensure!(
!local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.has_queued_wrapper(id)),
"wrapper already owns a queued selection; explicit recovery is required"
);
None
};
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
return Ok(cancelled_selection());
}
if *ctx.admission_barrier.read().await || ctx.history.ownership_failed() {
anyhow::ensure!(
!resume_queued,
"queued resume blocked by restart admission barrier"
);
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError("restart_admission_barrier_active".to_string()),
build_id: None,
diagnostics: None,
});
}
debug!(
"Selecting worker for project '{}' with {} cores",
request.project, request.estimated_cores
);
if let Some(classification_us) = request.classification_duration_us {
let duration_secs = classification_us as f64 / 1_000_000.0;
metrics::DECISION_LATENCY
.with_label_values(&["compilation"])
.observe(duration_secs);
let duration_ms = classification_us as f64 / 1000.0;
if duration_ms > 5.0 {
metrics::DECISION_BUDGET_VIOLATIONS
.with_label_values(&["compilation"])
.inc();
warn!(
"Classification latency budget violation: {:.3}ms (budget: 5ms)",
duration_ms
);
}
if duration_ms > 10.0 {
metrics::DECISION_PANIC_THRESHOLD_VIOLATIONS
.with_label_values(&["compilation"])
.inc();
tracing::error!(
"Classification latency exceeded panic threshold: {:.3}ms (threshold: 10ms)",
duration_ms
);
}
}
if std::env::var("RCH_MOCK_CIRCUIT_OPEN").is_ok() {
anyhow::ensure!(!resume_queued, "queued resume blocked by circuit override");
debug!("RCH_MOCK_CIRCUIT_OPEN set, returning AllCircuitsOpen");
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::AllCircuitsOpen,
build_id: None,
diagnostics: None,
});
}
async fn attempt_select_and_reserve(
ctx: &DaemonContext,
request: &SelectionRequest,
local_wrapper_id: Option<String>,
waiter: Option<&QueuedWaiterClaim>,
) -> Result<SelectionResponse> {
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
return Ok(cancelled_selection());
}
let admission = ctx.admission_barrier.read().await;
if *admission || ctx.history.ownership_failed() {
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError(
"restart_admission_barrier_active".to_string(),
),
build_id: None,
diagnostics: None,
});
}
let response = async {
let mut reservation_attempts = 0;
const MAX_ATTEMPTS: u32 = 3;
let mut excluded_worker_ids = ctx.history.active_workers_for_project(&request.project);
loop {
let result = ctx
.worker_selector
.select_with_exclusions(&ctx.pool, request, &excluded_worker_ids)
.await;
let selection_reason = result.reason;
let selection_diagnostics = result.diagnostics;
let Some(worker) = result.worker else {
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
return Ok(cancelled_selection());
}
debug!("No worker selected: {}", selection_reason);
return Ok(SelectionResponse {
worker: None,
reason: selection_reason,
build_id: None,
diagnostics: selection_diagnostics,
});
};
let selected_endpoint = worker.endpoint_snapshot().await;
let selected_worker_id = selected_endpoint.config.id.clone();
reservation_attempts += 1;
let reserve_slots = {
let total = worker.effective_total_slots().await;
if request.estimated_cores > total {
total.min(worker.available_slots().await.max(1))
} else {
request.estimated_cores
}
};
if worker.reserve_slots(reserve_slots).await {
let Some(config) = worker.lock_current_endpoint(&selected_endpoint).await
else {
worker.release_slots(reserve_slots).await;
excluded_worker_ids.insert(selected_worker_id.as_str().to_string());
continue;
};
let (id, host, user, identity_file, declared_os) = (
config.id.clone(),
config.host.clone(),
config.user.clone(),
config.identity_file.clone(),
rch_common::declared_os(&config.tags),
);
let command = request
.command
.clone()
.unwrap_or_else(|| "<unknown>".to_string());
let admission = ctx.history.try_start_active_build_with_waiter(
request.project.clone(),
id.as_str().to_string(),
command.clone(),
request.hook_pid.unwrap_or(0),
local_wrapper_id.clone(),
reserve_slots,
rch_common::BuildLocation::Remote,
waiter,
crate::disk_pressure::DiskHeadroomAdmission {
requested_gib: request.disk_headroom_gib,
capacity: worker.disk_capacity_observation().await,
},
Some(selected_endpoint.clone()),
);
drop(config);
let state = match admission {
Ok(Some(state)) => state,
Ok(None) => {
worker.release_slots(reserve_slots).await;
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
return Ok(cancelled_selection());
}
if ctx.history.ownership_failed() {
anyhow::bail!(
"durable ownership uncertain; admission closed until restart"
);
}
anyhow::ensure!(
waiter.is_none_or(|claim| ctx.history.owns_queued_waiter(claim)),
"queued waiter no longer owns admission"
);
anyhow::ensure!(
waiter.is_some()
|| !local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.has_queued_wrapper(id)),
"wrapper already owns a queued selection"
);
excluded_worker_ids.insert(id.as_str().to_string());
continue;
}
Err(error) => {
return Err(error.into());
}
};
let build_id = Some(state.id);
if !cfg!(test) {
metrics::inc_active_builds("remote");
}
ctx.events.emit(
"build_started",
&serde_json::json!({
"build_id": state.id, "project_id": request.project.clone(),
"worker_id": id.as_str(), "command": command,
"local_wrapper_id": local_wrapper_id.clone(), "slots": reserve_slots,
}),
);
let slots_available = worker.available_slots().await;
let speed_score = worker.get_speed_score();
if request.command_priority != CommandPriority::Normal {
ctx.events.emit(
"priority_hint",
&serde_json::json!({
"project": request.project.clone(),
"worker_id": id.as_str(),
"priority": request.command_priority.to_string(),
"estimated_cores": request.estimated_cores,
"command": request.command.clone(),
}),
);
}
return Ok(SelectionResponse {
worker: Some(SelectedWorker {
id,
host,
user,
identity_file,
slots_available,
speed_score,
declared_os,
}),
reason: selection_reason,
build_id,
diagnostics: selection_diagnostics,
});
}
warn!(
"Failed to reserve {} slots on {} (race condition), attempt {}/{}",
reserve_slots, selected_worker_id, reservation_attempts, MAX_ATTEMPTS
);
if reservation_attempts >= MAX_ATTEMPTS {
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::AllWorkersBusy,
build_id: None,
diagnostics: None,
});
}
}
}
.await;
drop(admission);
response
}
let initial = if resumed.is_some() {
SelectionResponse {
worker: None,
reason: SelectionReason::AllWorkersBusy,
build_id: None,
diagnostics: None,
}
} else {
attempt_select_and_reserve(ctx, &request, local_wrapper_id.clone(), None).await?
};
if initial.worker.is_some()
|| !wait_for_worker
|| initial.reason != SelectionReason::AllWorkersBusy
{
return Ok(initial);
}
let hook_pid = request.hook_pid.unwrap_or(0);
let command = request
.command
.clone()
.unwrap_or_else(|| "<unknown>".to_string());
let daemon_queue_timeout_secs = ctx.queue_timeout_secs.max(1);
let effective_queue_timeout_secs = wait_timeout_secs
.filter(|secs| *secs > 0)
.map(|client_secs| client_secs.min(daemon_queue_timeout_secs))
.unwrap_or(daemon_queue_timeout_secs);
let selection_digest = queue_selection_digest(&request, wait_timeout_secs)?;
let admission = ctx.admission_barrier.read().await;
if *admission || ctx.history.ownership_failed() {
anyhow::ensure!(
!resume_queued,
"queued resume blocked by restart admission barrier"
);
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError("restart_admission_barrier_active".to_string()),
build_id: None,
diagnostics: None,
});
}
let queued_result = resumed.or_else(|| {
ctx.history.enqueue_selection_build(
request.project.clone(),
command.clone(),
hook_pid,
request.estimated_cores,
local_wrapper_id.clone(),
QueueSelectionContract {
digest: selection_digest,
timeout_secs: effective_queue_timeout_secs,
},
)
});
drop(admission);
let Some((queued, waiter)) = queued_result else {
if ctx.history.ownership_failed() {
anyhow::bail!("durable ownership uncertain; queue admission closed until restart");
}
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
return Ok(cancelled_selection());
}
anyhow::ensure!(
!local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.has_queued_wrapper(id)),
"wrapper already owns a queued selection"
);
return Ok(initial);
};
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
ctx.events.emit(
if resume_queued {
"build_queue_resumed"
} else {
"build_queued"
},
&serde_json::json!({
"queue_id": queued.id,
"project_id": queued.project_id,
"command": queued.command,
"queued_at": queued.queued_at,
"position": ctx.history.queue_position(queued.id),
"slots_needed": queued.slots_needed,
}),
);
const QUEUE_POLL_INTERVAL: Duration = Duration::from_secs(1);
let queue_timeout = Duration::from_secs(
queued
.selection_contract
.as_ref()
.expect("selection queue has a contract")
.timeout_secs
.min(effective_queue_timeout_secs),
);
loop {
if local_wrapper_id
.as_deref()
.is_some_and(|id| ctx.history.wrapper_cancelled(id))
{
ctx.history
.finish_queued_build(queued.id, local_wrapper_id.as_deref())?;
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
return Ok(cancelled_selection());
}
if queued.queued_at_mono.elapsed() > queue_timeout {
if ctx
.history
.finish_queued_build(queued.id, local_wrapper_id.as_deref())?
{
return Ok(cancelled_selection());
}
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
ctx.events.emit(
"build_queue_removed",
&serde_json::json!({
"queue_id": queued.id,
"project_id": queued.project_id,
"reason": "timeout",
"waited_secs": queued.queued_at_mono.elapsed().as_secs(),
}),
);
warn!(
"Queue timeout after {}s for project {}, falling back to local",
queued.queued_at_mono.elapsed().as_secs(),
queued.project_id
);
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError("queue_timeout".to_string()),
build_id: None,
diagnostics: None,
});
}
if hook_pid > 0 && !is_process_alive(hook_pid) {
if ctx
.history
.finish_queued_build(queued.id, local_wrapper_id.as_deref())?
{
return Ok(cancelled_selection());
}
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
ctx.events.emit(
"build_queue_removed",
&serde_json::json!({
"queue_id": queued.id,
"project_id": queued.project_id,
"reason": "hook_exited",
}),
);
return Ok(SelectionResponse {
worker: None,
reason: SelectionReason::SelectionError("hook_exited".to_string()),
build_id: None,
diagnostics: None,
});
}
anyhow::ensure!(
ctx.history.owns_queued_waiter(&waiter),
"queued waiter ownership lost"
);
let response =
attempt_select_and_reserve(ctx, &request, local_wrapper_id.clone(), Some(&waiter))
.await?;
if response.worker.is_some() {
let _ = ctx.history.remove_queued_build(queued.id);
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
return Ok(response);
}
if response.reason != SelectionReason::AllWorkersBusy {
if ctx
.history
.finish_queued_build(queued.id, local_wrapper_id.as_deref())?
{
return Ok(cancelled_selection());
}
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
ctx.events.emit(
"build_queue_removed",
&serde_json::json!({
"queue_id": queued.id,
"project_id": queued.project_id,
"reason": response.reason.to_string(),
}),
);
return Ok(response);
}
tokio::time::sleep(QUEUE_POLL_INTERVAL).await;
}
}
async fn handle_restart_admission(ctx: &DaemonContext, close: bool) -> RestartAdmissionResponse {
let mut admission = ctx.admission_barrier.write().await;
let was_closed = *admission;
let close = close || ctx.history.ownership_failed();
*admission = close;
let active_build_ids: Vec<u64> = ctx
.history
.active_builds()
.into_iter()
.map(|build| build.id)
.collect();
let queued_build_ids: Vec<u64> = ctx
.history
.queued_builds()
.into_iter()
.map(|build| build.id)
.collect();
let client_lease_scan = tokio::task::spawn_blocking(nonterminal_client_lease_ids)
.await
.unwrap_or_else(|error| Err(anyhow!("lease scan task failed: {error}")));
let (client_lease_ids, client_lease_scan_error) = match client_lease_scan {
Ok(ids) => (ids, None),
Err(error) => (Vec::new(), Some(error.to_string())),
};
let restart_permitted = close
&& !was_closed
&& active_build_ids.is_empty()
&& queued_build_ids.is_empty()
&& client_lease_ids.is_empty()
&& client_lease_scan_error.is_none();
RestartAdmissionResponse {
admission_closed: close,
restart_permitted,
active_build_ids,
queued_build_ids,
client_lease_ids,
client_lease_scan_error,
}
}
async fn restart_admission_status(ctx: &DaemonContext) -> RestartAdmissionResponse {
let admission = ctx.admission_barrier.read().await;
let active_build_ids: Vec<u64> = ctx
.history
.active_builds()
.into_iter()
.map(|build| build.id)
.collect();
let queued_build_ids: Vec<u64> = ctx
.history
.queued_builds()
.into_iter()
.map(|build| build.id)
.collect();
let client_lease_scan = tokio::task::spawn_blocking(nonterminal_client_lease_ids)
.await
.unwrap_or_else(|error| Err(anyhow!("lease scan task failed: {error}")));
let (client_lease_ids, client_lease_scan_error) = match client_lease_scan {
Ok(ids) => (ids, None),
Err(error) => (Vec::new(), Some(error.to_string())),
};
RestartAdmissionResponse {
admission_closed: *admission || ctx.history.ownership_failed(),
restart_permitted: false,
active_build_ids,
queued_build_ids,
client_lease_ids,
client_lease_scan_error,
}
}
fn nonterminal_client_lease_ids() -> Result<Vec<String>> {
scan_client_leases(&default_job_lease_directory())
}
fn scan_client_leases(lease_dir: &std::path::Path) -> Result<Vec<String>> {
let entries = match std::fs::read_dir(lease_dir) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(anyhow!("cannot inspect {}: {error}", lease_dir.display())),
};
let now_unix_ms = u64::try_from(chrono::Utc::now().timestamp_millis()).unwrap_or(0);
let mut blocked = Vec::new();
for entry in entries {
let entry =
entry.map_err(|error| anyhow!("cannot enumerate {}: {error}", lease_dir.display()))?;
if entry
.path()
.extension()
.is_none_or(|extension| extension != "json")
{
continue;
}
let bytes = std::fs::read(entry.path())
.map_err(|error| anyhow!("cannot read {}: {error}", entry.path().display()))?;
let lease: DurableJobLease = serde_json::from_slice(&bytes)
.map_err(|error| anyhow!("cannot parse {}: {error}", entry.path().display()))?;
if lease_blocks_restart(&lease, now_unix_ms, || is_process_alive(lease.wrapper_pid)) {
blocked.push(lease.identity.local_wrapper_id);
} else if now_unix_ms.saturating_sub(lease.heartbeat_unix_ms) >= LEASE_REAP_RETENTION_MS
&& !lease_owns_unretired_source(&lease)
{
let _ = std::fs::remove_file(entry.path());
}
}
blocked.sort();
Ok(blocked)
}
const LEASE_REAP_RETENTION_MS: u64 = 60 * 60 * 1000;
const DEAD_WRAPPER_HEARTBEAT_STALE_MS: u64 = 15 * 60 * 1000;
pub(crate) fn lease_owns_unretired_source(lease: &DurableJobLease) -> bool {
lease.recovery.as_ref().is_some_and(|recipe| {
let owns = recipe
.get("source_roots")
.and_then(serde_json::Value::as_array)
.is_some_and(|roots| !roots.is_empty())
|| recipe.get("pair").is_some_and(|pair| !pair.is_null())
|| recipe
.get("retire_root")
.is_some_and(|root| !root.is_null());
owns && recipe.get("retired").and_then(serde_json::Value::as_bool) != Some(true)
})
}
pub(crate) fn lease_blocks_restart(
lease: &DurableJobLease,
now_unix_ms: u64,
wrapper_alive: impl FnOnce() -> bool,
) -> bool {
if lease.state.is_terminal() && lease.terminal_acknowledged {
return false;
}
let heartbeat_stale =
now_unix_ms.saturating_sub(lease.heartbeat_unix_ms) >= DEAD_WRAPPER_HEARTBEAT_STALE_MS;
if !heartbeat_stale {
return true;
}
if lease.wrapper_pid == 0 {
return true;
}
wrapper_alive()
}
async fn handle_release_worker(ctx: &DaemonContext, request: ReleaseRequest) -> Result<()> {
let _release_guard = ctx.history.lock_releases().await;
let started = std::time::Instant::now();
let exit_code = request.exit_code.unwrap_or(0);
let worker_disk_full = exit_code != 0 && request.worker_disk_full;
let (release_worker_id, release_slots, record, remote_command_started, worker_endpoint) =
if let Some(build_id) = request.build_id {
let completion = ctx.history.complete_durable_with_disk_fault(
build_id,
request.worker_id.as_str(),
request.local_wrapper_id.as_deref(),
crate::history::BuildCompletion {
exit_code,
duration_ms: request.duration_ms,
bytes_transferred: request.bytes_transferred,
timing: request.timing,
cancellation: None,
},
worker_disk_full.then_some(request.worker_disk_roots),
)?;
let Some((state, record)) = completion else {
if let Some(fault) = ctx.history.pending_disk_fault(build_id) {
anyhow::ensure!(
apply_pending_disk_fault(ctx, &fault).await?,
"disk-fault endpoint is absent; durable incident is pending reintroduction"
);
}
return Ok(());
};
let remote_command_started = state.remote_command_started();
(
WorkerId::new(state.worker_id),
state.slots,
Some(record),
remote_command_started,
state.worker_endpoint,
)
} else {
anyhow::bail!("release requires durable build_id; unowned slot release refused")
};
let ownership_done = started.elapsed();
let quarantine_result = if let Some(fault) = ctx.history.pending_disk_fault(
request
.build_id
.expect("durable build identity checked above"),
) {
apply_pending_disk_fault(ctx, &fault)
.await
.and_then(|resolved| {
anyhow::ensure!(
resolved,
"disk-fault endpoint is absent; durable incident is pending reintroduction"
);
Ok(())
})
} else {
Ok(())
};
debug!(
"Releasing {} slots on worker {}",
release_slots, release_worker_id
);
ctx.pool
.release_slots(&release_worker_id, release_slots)
.await;
let slots_done = started.elapsed();
if let Some(ref rec) = record {
if !cfg!(test) {
metrics::dec_active_builds("remote");
let outcome = if exit_code == 0 { "success" } else { "failure" };
metrics::inc_build_total(outcome, "remote");
}
ctx.events.emit(
"build_completed",
&serde_json::json!({
"build_id": rec.id,
"project_id": rec.project_id,
"worker_id": rec.worker_id,
"command": rec.command,
"exit_code": rec.exit_code,
"duration_ms": rec.duration_ms,
"location": format!("{:?}", rec.location),
}),
);
if let Some(endpoint) = worker_endpoint.as_ref()
&& let Some(worker) = ctx.pool.get(&endpoint.config.id).await
{
if exit_code == 0
&& let Some(_endpoint_guard) = worker.lock_current_endpoint(endpoint).await
{
worker.record_success().await;
}
ctx.worker_selector
.record_bound_remote_completion(
&worker,
endpoint,
&rec.project_id,
&rec.command,
exit_code,
remote_command_started && !request.worker_fault && !worker_disk_full,
)
.await;
}
}
let total = started.elapsed();
if total >= SLOW_RELEASE_WARN {
warn!(
build_id = ?request.build_id,
worker_id = %release_worker_id,
ownership_ms = ownership_done.as_millis() as u64,
slots_ms = (slots_done - ownership_done).as_millis() as u64,
completion_ms = (total - slots_done).as_millis() as u64,
"Slow release-worker handling ({} ms)",
total.as_millis()
);
}
quarantine_result
}
async fn apply_pending_disk_fault(
ctx: &DaemonContext,
fault: &crate::history::PendingDiskFault,
) -> Result<bool> {
let worker = ctx.pool.get(&WorkerId::new(&fault.worker_id)).await;
let Some(store) = &ctx.bypass_store else {
let Some(endpoint) = fault.worker_endpoint.as_ref() else {
ctx.history
.archive_disk_fault(fault.build_id, &fault.incident_id)?;
return Ok(true);
};
let Some(worker) = worker else {
return Ok(false);
};
let Some(_config) = worker
.lock_disk_fault_endpoint(endpoint, fault.runtime_endpoint.as_ref())
.await
else {
if worker.is_endpoint_retired() {
return Ok(false);
}
ctx.history
.archive_disk_fault(fault.build_id, &fault.incident_id)?;
return Ok(true);
};
worker
.enter_bypass(rch_common::BypassFailureClass::DiskInodePressure)
.await;
anyhow::bail!("worker disk fault cannot be acknowledged without durable bypass storage");
};
crate::bypass_recovery_service::apply_owned_disk_fault(
store,
worker.as_ref(),
&ctx.history,
fault,
)
.await
}
pub(crate) async fn replay_pending_disk_faults(ctx: &DaemonContext) -> Result<()> {
let _release_guard = ctx.history.lock_releases().await;
for fault in ctx.history.pending_disk_faults() {
apply_pending_disk_fault(ctx, &fault).await?;
}
Ok(())
}
fn ownership_mismatch_message(build_id: u64, owner: Option<Option<&str>>) -> String {
match owner {
Some(Some(wrapper)) => format!(
"build ownership mismatch: build {build_id} is owned by wrapper {wrapper}. \
Cancel it with `rch jobs cancel {wrapper}`. If that wrapper and its lease are \
gone, the stuck detector releases the build on its own once it has been silent \
for 15 min (never started remote work) or 6h (started)"
),
Some(None) => format!(
"build ownership mismatch: build {build_id} has no wrapper identity; cancel it \
by build id alone"
),
None => format!("build ownership mismatch: build {build_id} has already completed"),
}
}
const SLOW_RELEASE_WARN: Duration = Duration::from_secs(1);
async fn handle_record_build(
ctx: &DaemonContext,
worker_id: &WorkerId,
project: &str,
is_test: bool,
) -> Result<()> {
debug!(
"Recording build for project '{}' on worker {}",
project, worker_id
);
ctx.worker_selector
.record_build(worker_id.as_str(), project, is_test)
.await;
Ok(())
}
fn heartbeat_phase_to_str(phase: &rch_common::BuildHeartbeatPhase) -> &'static str {
match phase {
rch_common::BuildHeartbeatPhase::SyncUp => "sync_up",
rch_common::BuildHeartbeatPhase::Execute => "execute",
rch_common::BuildHeartbeatPhase::SyncDown => "sync_down",
rch_common::BuildHeartbeatPhase::Finalize => "finalize",
}
}
async fn handle_build_heartbeat(
ctx: &DaemonContext,
request: BuildHeartbeatRequest,
) -> BuildHeartbeatResponse {
let build_id = request.build_id;
let worker_id = request.worker_id.as_str().to_string();
let phase = heartbeat_phase_to_str(&request.phase).to_string();
let detail = request.detail.clone();
let progress_counter = request.progress_counter;
let progress_percent = request.progress_percent;
if ctx.history.ownership_failed() {
return BuildHeartbeatResponse {
status: "error: durable ownership uncertain; restart required".to_string(),
build_id,
worker_id,
phase,
};
}
if let Some(state) = ctx.history.record_build_heartbeat(request) {
let event_phase = heartbeat_phase_to_str(&state.heartbeat_phase).to_string();
let event_detail = state.heartbeat_detail.clone();
let event_counter = state.heartbeat_counter;
let event_percent = state.heartbeat_percent;
ctx.events.emit(
"build_heartbeat",
&serde_json::json!({
"build_id": state.id,
"project_id": state.project_id,
"worker_id": state.worker_id,
"phase": event_phase,
"detail": event_detail,
"progress_counter": event_counter,
"progress_percent": event_percent,
}),
);
BuildHeartbeatResponse {
status: "ok".to_string(),
build_id,
worker_id,
phase,
}
} else {
if ctx.history.ownership_failed() {
return BuildHeartbeatResponse {
status: "error: failed to persist heartbeat ownership; restart required"
.to_string(),
build_id,
worker_id,
phase,
};
}
warn!(
"Ignoring heartbeat for unknown build {} on worker {}",
build_id, worker_id
);
ctx.events.emit(
"build_heartbeat_ignored",
&serde_json::json!({
"build_id": build_id,
"worker_id": worker_id.clone(),
"phase": phase.clone(),
"detail": detail,
"progress_counter": progress_counter,
"progress_percent": progress_percent,
}),
);
BuildHeartbeatResponse {
status: "ignored".to_string(),
build_id,
worker_id,
phase,
}
}
}
fn handle_metrics() -> Result<String> {
metrics::encode_metrics()
}
fn handle_health(ctx: &DaemonContext) -> HealthResponse {
HealthResponse {
status: "healthy".to_string(),
version: ctx.version.to_string(),
uptime_seconds: ctx.started_at.elapsed().as_secs(),
}
}
pub(crate) fn worker_accepts_new_builds(
status: WorkerStatus,
circuit_state: CircuitState,
pressure_state: crate::disk_pressure::PressureState,
) -> bool {
matches!(status, WorkerStatus::Healthy | WorkerStatus::Degraded)
&& circuit_state != CircuitState::Open
&& pressure_state != crate::disk_pressure::PressureState::Critical
}
async fn handle_ready(ctx: &DaemonContext) -> ReadyResponse {
let workers = ctx.pool.all_workers().await;
let mut workers_available = false;
for w in workers {
let Some(circuit_state) = w.circuit_state().await else {
continue;
};
let pressure_state = w.pressure_assessment().await.state;
if !worker_accepts_new_builds(w.status().await, circuit_state, pressure_state) {
continue;
}
if w.available_slots().await > 0 {
workers_available = true;
break;
}
}
if workers_available {
ReadyResponse {
status: "ready".to_string(),
workers_available: true,
reason: None,
}
} else {
ReadyResponse {
status: "not_ready".to_string(),
workers_available: false,
reason: Some("no_workers_available".to_string()),
}
}
}
fn handle_budget() -> BudgetStatusResponse {
budget::get_budget_status()
}
pub(crate) fn is_process_alive(pid: u32) -> bool {
if pid == 0 {
return false;
}
#[cfg(unix)]
{
let Ok(pid) = i32::try_from(pid) else {
return false;
};
match nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None) {
Ok(()) => true,
Err(nix::errno::Errno::EPERM) => true,
Err(_) => false,
}
}
#[cfg(not(unix))]
{
std::process::Command::new("kill")
.arg("-0")
.arg(pid.to_string())
.status()
.map(|status| status.success())
.unwrap_or(false)
}
}
async fn handle_cancel_job(
ctx: &DaemonContext,
wrapper: &str,
force: bool,
) -> Result<serde_json::Value> {
use crate::history::WrapperCancellation;
match ctx.history.cancel_wrapper(wrapper)? {
WrapperCancellation::BeforeStart => {
ctx.history.update_queue_estimates();
if !cfg!(test) {
metrics::set_build_queue_depth(ctx.history.queue_depth());
}
ctx.events.emit(
"job_cancelled_before_start",
&serde_json::json!({"local_wrapper_id":wrapper}),
);
Ok(
serde_json::json!({"status":"cancelled_before_start", "local_wrapper_id":wrapper, "exit_code":130, "slots_released":0}),
)
}
WrapperCancellation::Active(build_id) => {
let response = handle_cancel_build(ctx, build_id, force).await;
let mut response = serde_json::to_value(response)?;
response["local_wrapper_id"] = serde_json::json!(wrapper);
Ok(response)
}
WrapperCancellation::Completed(record) => Ok(
serde_json::json!({"status":"completed", "local_wrapper_id":wrapper, "record":record}),
),
WrapperCancellation::NotQueued => Ok(
serde_json::json!({"status":"not_queued", "local_wrapper_id":wrapper, "message":"Job is not currently queued; no cancellation-before-start was acknowledged"}),
),
}
}
async fn handle_cancel_build(
ctx: &DaemonContext,
build_id: u64,
force: bool,
) -> CancelBuildResponse {
ctx.cancellation
.cancel_build(
ctx,
build_id,
crate::cancellation::CancelReason::User,
force,
)
.await
}
async fn handle_cancel_all_builds(ctx: &DaemonContext, force: bool) -> CancelAllBuildsResponse {
ctx.cancellation.cancel_all_builds(ctx, force).await
}
fn cancellation_issues_from_recent_builds(recent_builds: &[BuildRecord]) -> Vec<Issue> {
let mut cancelled = 0usize;
let mut cleanup_failures = 0usize;
let mut sigkill_escalations = 0usize;
let mut unreachable_workers = 0usize;
let mut latest_operation: Option<String> = None;
for build in recent_builds {
let Some(cancellation) = &build.cancellation else {
continue;
};
cancelled += 1;
latest_operation = Some(cancellation.operation_id.clone());
if !cancellation.cleanup_ok {
cleanup_failures += 1;
}
if cancellation.escalation_stage == "sigkill" {
sigkill_escalations += 1;
}
if cancellation
.worker_health
.as_ref()
.is_some_and(|health| health.status == "unreachable")
{
unreachable_workers += 1;
}
}
let mut issues = Vec::new();
if cleanup_failures > 0 {
let operation_hint = latest_operation
.map(|op| format!(" Last operation: {op}."))
.unwrap_or_default();
issues.push(Issue {
severity: "error".to_string(),
summary: format!(
"{} recent cancellation(s) finished with cleanup failures.{operation_hint}",
cleanup_failures
),
remediation: Some(
"Run `rch workers probe --all`, then inspect daemon logs for `cancellation_failed` events before retrying affected builds.".to_string(),
),
});
}
if sigkill_escalations > 0 {
issues.push(Issue {
severity: "warning".to_string(),
summary: format!(
"{} recent cancellation(s) escalated to SIGKILL (stuck or unresponsive process trees).",
sigkill_escalations
),
remediation: Some(
"Check for stuck toolchain phases with `rch status --jobs` and investigate long-running remote processes.".to_string(),
),
});
}
if unreachable_workers > 0 {
issues.push(Issue {
severity: "warning".to_string(),
summary: format!(
"{} cancellation(s) ended while worker health reported unreachable.",
unreachable_workers
),
remediation: Some(
"Validate worker reachability with `rch workers probe --all` and confirm SSH connectivity before resuming remote builds.".to_string(),
),
});
}
if cancelled > 0 && issues.is_empty() {
issues.push(Issue {
severity: "info".to_string(),
summary: format!(
"{} recent cancellation(s) completed cleanly with deterministic cleanup.",
cancelled
),
remediation: Some(
"No action required. Use `rch status --json` to inspect cancellation metadata if debugging build interruptions.".to_string(),
),
});
}
issues
}
fn active_build_issues_from_active_builds(
active_builds: &[crate::history::ActiveBuildState],
) -> Vec<Issue> {
let stalled_live_hook_builds: Vec<_> = active_builds
.iter()
.filter(|build| {
build.detector_progress_stale
&& build.detector_hook_alive
&& !build.detector_heartbeat_stale
})
.collect();
if stalled_live_hook_builds.is_empty() {
return Vec::new();
}
let representative = stalled_live_hook_builds
.iter()
.max_by_key(|build| build.detector_build_age_secs)
.expect("non-empty stalled live-hook build list");
vec![Issue {
severity: "info".to_string(),
summary: format!(
"{} active build(s) have stale progress while hook heartbeats remain fresh (longest: build {} on worker {}); usually a large crate compiling silently.",
stalled_live_hook_builds.len(),
representative.id,
representative.worker_id
),
remediation: Some(format!(
"No action needed while the hook stays alive. Inspect with `rch queue --json`; cancel build {} only if it is past its build/test timeout budget. Do not drain {} for this: the worker is healthy.",
representative.id, representative.worker_id
)),
}]
}
fn build_remediation_view(
worker_infos: &[WorkerStatusInfo],
active_builds: &[crate::history::ActiveBuildState],
queued_count: usize,
alerts: &[AlertInfo],
now: chrono::DateTime<Utc>,
) -> rch_common::remediation_view::RemediationView {
use rch_common::bypass_record::BypassState;
use rch_common::fleet_diff::WorkerObservation;
use rch_common::fleet_status::DEFAULT_ABSENCE_THRESHOLD_SECS;
use rch_common::remediation_view::{
DiskLevel, JobsInput, MAX_VIEW_INCIDENTS, ProofQueueInput, RemediationIncidentLine,
RemediationWorkerRow, assemble, build_inputs,
};
let rows: Vec<RemediationWorkerRow> = worker_infos
.iter()
.map(|w| {
let facts_known = w.pressure_state != "telemetry_gap";
RemediationWorkerRow {
observation: WorkerObservation {
worker_id: w.id.clone(),
configured: true,
in_daemon_pool: true,
reachable: w.status != "unreachable",
admin_disabled: w.status == "disabled",
temporarily_bypassed: w.bypass.is_some(),
facts_known,
command_admissible: true,
},
disk_level: DiskLevel::from_pressure_state(&w.pressure_state),
reclaiming: false,
free_ratio: w.pressure_disk_free_ratio,
slots_used: w.used_slots,
slots_total: w.total_slots,
telemetry_known: facts_known,
telemetry_fresh: w.pressure_telemetry_fresh,
telemetry_age_secs: w.pressure_telemetry_age_secs,
recovered_pending_canary: w
.bypass
.as_ref()
.is_some_and(|b| b.state == BypassState::RecoveredPendingCanary),
absent_secs: None,
}
})
.collect();
let stuck = active_builds
.iter()
.filter(|b| {
!b.detector_hook_alive || (b.detector_heartbeat_stale && b.detector_progress_stale)
})
.count();
let jobs = JobsInput {
active: active_builds.len(),
queued: queued_count,
stuck,
};
let incidents: Vec<RemediationIncidentLine> = alerts
.iter()
.filter(|a| a.state == "active")
.take(MAX_VIEW_INCIDENTS)
.map(|a| {
let age = chrono::DateTime::parse_from_rfc3339(&a.last_seen)
.ok()
.map(|t| {
u64::try_from((now - t.with_timezone(&Utc)).num_seconds().max(0)).unwrap_or(0)
})
.unwrap_or(0);
RemediationIncidentLine::new(
a.kind.clone(),
"alert",
a.worker_id.clone(),
age,
a.message.clone(),
)
})
.collect();
let inputs = build_inputs(
&rows,
jobs,
ProofQueueInput::default(),
incidents,
DEFAULT_ABSENCE_THRESHOLD_SECS,
);
let now_ms = u64::try_from(now.timestamp_millis()).unwrap_or(0);
assemble(&inputs, now_ms)
}
fn worker_issue(
worker_id: &str,
status: WorkerStatus,
circuit_state: CircuitState,
pressure_state: crate::disk_pressure::PressureState,
pressure_reason_code: &str,
disabled_reason: Option<&str>,
) -> Option<Issue> {
if status == WorkerStatus::Disabled {
let reason = disabled_reason.unwrap_or("no reason recorded");
return Some(Issue {
severity: "warning".to_string(),
summary: format!(
"Worker '{worker_id}' is administratively disabled ({reason}); its slots are withheld, it is not polled for telemetry, and the disable is durable — restarting the daemon re-applies it"
),
remediation: Some(format!("rch workers enable {worker_id}")),
});
}
if status == WorkerStatus::Drained {
return Some(Issue {
severity: "warning".to_string(),
summary: format!(
"Worker '{worker_id}' is drained; its slots are withheld and it is not polled for telemetry"
),
remediation: Some(format!("rch workers enable {worker_id}")),
});
}
if circuit_state == CircuitState::Open {
return Some(Issue {
severity: "error".to_string(),
summary: format!("Circuit open for worker '{worker_id}'"),
remediation: Some(format!("rch workers probe {worker_id} --force")),
});
}
if pressure_state == crate::disk_pressure::PressureState::Critical {
return Some(Issue {
severity: "error".to_string(),
summary: format!(
"Worker '{worker_id}' in critical pressure state ({pressure_reason_code})"
),
remediation: Some(
"rch workers capabilities --refresh (inspect pressure metrics and ballast policy)"
.to_string(),
),
});
}
if pressure_state == crate::disk_pressure::PressureState::TelemetryGap {
return Some(Issue {
severity: "warning".to_string(),
summary: format!(
"Worker '{worker_id}' has stale/missing pressure telemetry ({pressure_reason_code})"
),
remediation: Some(
"telemetry ingest is daemon-driven; wait for next poll (~poll-interval) or run `rch daemon restart` to force a fresh poll cycle"
.to_string(),
),
});
}
if status == WorkerStatus::Unreachable {
return Some(Issue {
severity: "error".to_string(),
summary: format!("Worker '{worker_id}' is unreachable"),
remediation: Some(format!("rch workers probe {worker_id}")),
});
}
if status == WorkerStatus::Degraded {
return Some(Issue {
severity: "warning".to_string(),
summary: format!("Worker '{worker_id}' is degraded (slow response)"),
remediation: None,
});
}
None
}
pub(crate) async fn handle_status(ctx: &DaemonContext) -> Result<DaemonFullStatus> {
let workers = ctx.pool.all_workers().await;
let bypass_store = BypassRecordStore::load(default_bypass_record_path());
let mut workers_healthy = 0;
let mut slots_total = 0u32;
let mut slots_available = 0u32;
let mut worker_infos = Vec::with_capacity(workers.len());
let mut issues = Vec::new();
for worker in &workers {
let status = worker.status().await;
let (worker_id, host, user) = {
let config = worker.config.read().await;
(
config.id.to_string(),
config.host.clone(),
config.user.clone(),
)
};
let total_slots = worker.effective_total_slots().await;
let used_slots = worker.used_slots();
let available_slots = total_slots.saturating_sub(used_slots);
let circuit_stats = worker.circuit_stats().await;
let circuit_state = circuit_stats.state();
let pressure = worker.pressure_assessment().await;
let assignable_slots = if worker_accepts_new_builds(status, circuit_state, pressure.state) {
available_slots
} else {
0
};
if status == WorkerStatus::Healthy {
workers_healthy += 1;
}
slots_total = slots_total.saturating_add(total_slots);
slots_available = slots_available.saturating_add(assignable_slots);
let status_str = match status {
WorkerStatus::Healthy => "healthy",
WorkerStatus::Degraded => "degraded",
WorkerStatus::Unreachable => "unreachable",
WorkerStatus::Draining => "draining",
WorkerStatus::Drained => "drained",
WorkerStatus::Disabled => "disabled",
};
let circuit_str = match circuit_state {
CircuitState::Closed => "closed",
CircuitState::Open => "open",
CircuitState::HalfOpen => "half_open",
};
let circuit_config = CircuitBreakerConfig::default();
let recovery_in_secs = circuit_stats.recovery_remaining_secs(&circuit_config);
worker_infos.push(WorkerStatusInfo {
id: worker_id.clone(),
host,
user,
status: status_str.to_string(),
circuit_state: circuit_str.to_string(),
used_slots,
total_slots,
speed_score: worker.get_speed_score(),
last_error: worker.last_error().await,
consecutive_failures: circuit_stats.consecutive_failures(),
recovery_in_secs,
failure_history: circuit_stats.recent_results().to_vec(),
pressure_state: pressure.state.to_string(),
pressure_confidence: pressure.confidence.to_string(),
pressure_reason_code: pressure.reason_code.clone(),
pressure_policy_rule: pressure.policy_rule.clone(),
pressure_disk_free_gb: pressure.disk_free_gb,
pressure_disk_total_gb: pressure.disk_total_gb,
pressure_disk_free_ratio: pressure.disk_free_ratio,
pressure_build_disk_free_gb: pressure.build_disk_free_gb,
pressure_build_disk_total_gb: pressure.build_disk_total_gb,
pressure_disk_io_util_pct: pressure.disk_io_util_pct,
pressure_memory_pressure: pressure.memory_pressure,
pressure_telemetry_age_secs: pressure.telemetry_age_secs,
pressure_telemetry_fresh: pressure.telemetry_fresh,
bypass: bypass_store.get(&worker_id).cloned(),
});
if let Some(issue) = worker_issue(
&worker_id,
status,
circuit_state,
pressure.state,
&pressure.reason_code,
worker.disabled_reason().await.as_deref(),
) {
issues.push(issue);
}
}
let uptime_secs = ctx.started_at.elapsed().as_secs();
let started_at = {
use std::time::{SystemTime, UNIX_EPOCH};
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let start = now.saturating_sub(uptime_secs);
let dt = chrono::DateTime::from_timestamp(start as i64, 0).unwrap_or_else(chrono::Utc::now);
dt.to_rfc3339()
};
let recent_builds = ctx.history.recent(20);
issues.extend(cancellation_issues_from_recent_builds(&recent_builds));
let active_builds = ctx.history.active_builds();
issues.extend(active_build_issues_from_active_builds(&active_builds));
let stats = ctx.history.stats();
let test_stats = ctx.telemetry.test_run_stats().await;
let alerts = ctx.alert_manager.active_alerts();
let queue_depth = ctx.history.queue_depth();
if !cfg!(test) {
metrics::set_build_queue_depth(queue_depth);
}
let remediation = build_remediation_view(
&worker_infos,
&active_builds,
queue_depth,
&alerts,
Utc::now(),
);
Ok(DaemonFullStatus {
daemon: DaemonStatusInfo {
pid: ctx.pid,
uptime_secs,
version: ctx.version.to_string(),
socket_path: ctx.socket_path.clone(),
started_at,
workers_total: workers.len(),
workers_healthy,
slots_total,
slots_available,
},
workers: worker_infos,
active_builds: active_builds
.into_iter()
.map(|b| ActiveBuild {
id: b.id,
project_id: b.project_id,
worker_id: b.worker_id,
command: b.command,
started_at: b.started_at,
last_heartbeat_at: b.last_heartbeat_at,
heartbeat_age_secs: b.last_heartbeat_mono.elapsed().as_secs(),
last_progress_at: b.last_progress_at,
progress_age_secs: b.last_progress_mono.elapsed().as_secs(),
heartbeat_phase: heartbeat_phase_to_str(&b.heartbeat_phase).to_string(),
heartbeat_detail: b.heartbeat_detail,
heartbeat_counter: b.heartbeat_counter,
heartbeat_percent: b.heartbeat_percent,
slots: b.slots,
detector_hook_alive: b.detector_hook_alive,
detector_heartbeat_stale: b.detector_heartbeat_stale,
detector_progress_stale: b.detector_progress_stale,
detector_confidence: b.detector_confidence,
detector_build_age_secs: b.detector_build_age_secs,
detector_slots_owned: b.detector_slots_owned,
detector_last_evaluated_at: b.detector_last_evaluated_at,
})
.collect(),
queued_builds: ctx
.history
.queued_builds()
.into_iter()
.enumerate()
.map(|(i, b)| {
let wait_secs = b.queued_at_mono.elapsed().as_secs();
QueuedBuild {
id: b.id,
id_text: b.id.to_string(),
project_id: b.project_id,
command: b.command,
queued_at: b.queued_at,
position: i + 1,
slots_needed: b.slots_needed,
estimated_start: b.estimated_start,
wait_time: format_wait_time(wait_secs),
}
})
.collect(),
recent_builds,
issues,
alerts,
stats,
test_stats,
saved_time: ctx.history.saved_time_stats(),
remediation,
})
}
fn convergence_remediation(state: &crate::repo_convergence::WorkerConvergenceState) -> Vec<String> {
use crate::repo_convergence::ConvergenceDriftState;
let mut hints = Vec::new();
match state.current_state {
ConvergenceDriftState::Ready => {}
ConvergenceDriftState::Drifting => {
if !state.missing_repos.is_empty() {
hints.push(format!(
"Missing {} repo(s): {}",
state.missing_repos.len(),
state.missing_repos.join(", ")
));
}
hints.push("Run convergence repair to sync missing repos.".into());
}
ConvergenceDriftState::Converging => {
hints.push("Sync in progress — wait for completion.".into());
}
ConvergenceDriftState::Failed => {
hints.push(format!(
"Budget exhausted (attempts: {}, time: {}ms remaining).",
state.attempt_budget_remaining, state.time_budget_remaining_ms
));
hints.push("Reset worker state with a new repo set update or manual repair.".into());
}
ConvergenceDriftState::Stale => {
hints.push("No recent status check — run status refresh.".into());
}
}
hints
}
fn build_convergence_worker_view(
state: &crate::repo_convergence::WorkerConvergenceState,
) -> ConvergenceWorkerView {
let drift_confidence = state.drift_confidence();
ConvergenceWorkerView {
worker_id: state.worker_id.clone(),
drift_state: state.current_state.to_string(),
drift_confidence,
required_repos: state.required_repos.clone(),
synced_repos: state.synced_repos.clone(),
missing_repos: state.missing_repos.clone(),
attempt_budget_remaining: state.attempt_budget_remaining,
time_budget_remaining_ms: state.time_budget_remaining_ms,
last_status_check_unix_ms: state.last_status_check_unix_ms,
remediation: convergence_remediation(state),
}
}
pub(crate) async fn handle_repo_convergence_status(
ctx: &DaemonContext,
worker_id: Option<&WorkerId>,
) -> ApiResponse<RepoConvergenceStatusResponse> {
let svc = &ctx.repo_convergence;
let workers: Vec<ConvergenceWorkerView> = if let Some(wid) = worker_id {
match svc.get_worker_state(wid).await {
Some(ws) => vec![build_convergence_worker_view(&ws)],
None => {
vec![ConvergenceWorkerView {
worker_id: wid.as_str().to_string(),
drift_state: "stale".into(),
drift_confidence: 0.0,
required_repos: vec![],
synced_repos: vec![],
missing_repos: vec![],
attempt_budget_remaining: 3,
time_budget_remaining_ms: 120_000,
last_status_check_unix_ms: 0,
remediation: vec!["No convergence data — run status refresh.".into()],
}]
}
}
} else {
let all = svc.get_all_worker_states().await;
all.iter().map(build_convergence_worker_view).collect()
};
let mut summary = ConvergenceSummary {
total_workers: workers.len(),
ready: 0,
drifting: 0,
converging: 0,
failed: 0,
stale: 0,
};
for w in &workers {
match w.drift_state.as_str() {
"ready" => summary.ready += 1,
"drifting" => summary.drifting += 1,
"converging" => summary.converging += 1,
"failed" => summary.failed += 1,
_ => summary.stale += 1,
}
}
let recent_outcomes = svc.get_recent_outcomes(20).await;
let status = if summary.failed > 0 {
"degraded"
} else if summary.drifting > 0 || summary.converging > 0 {
"converging"
} else if summary.total_workers == 0 || summary.stale == summary.total_workers {
"unknown"
} else {
"healthy"
};
ApiResponse::Ok(RepoConvergenceStatusResponse {
status: status.into(),
workers,
recent_outcomes,
summary,
})
}
async fn handle_repo_convergence_dry_run(
ctx: &DaemonContext,
worker_id: &WorkerId,
) -> ApiResponse<RepoConvergenceDryRunResponse> {
use crate::repo_convergence::ConvergenceDriftState;
let svc = &ctx.repo_convergence;
let has_budget = svc.has_budget(worker_id).await;
match svc.get_worker_state(worker_id).await {
Some(ws) => {
let (would_attempt, reason) = match ws.current_state {
ConvergenceDriftState::Ready => (false, "Worker is already converged.".into()),
ConvergenceDriftState::Drifting => {
if has_budget {
(true, "Would attempt sync for missing repos.".into())
} else {
(false, "Budget exhausted — cannot attempt sync.".into())
}
}
ConvergenceDriftState::Converging => (false, "Sync already in progress.".into()),
ConvergenceDriftState::Failed => {
(false, "Budget exhausted — manual reset required.".into())
}
ConvergenceDriftState::Stale => {
(true, "Would attempt fresh status check and sync.".into())
}
};
let remediation = convergence_remediation(&ws);
ApiResponse::Ok(RepoConvergenceDryRunResponse {
status: "ok".into(),
worker_id: worker_id.as_str().to_string(),
current_state: ws.current_state.to_string(),
missing_repos: ws.missing_repos.clone(),
has_budget,
attempt_budget_remaining: ws.attempt_budget_remaining,
time_budget_remaining_ms: ws.time_budget_remaining_ms,
would_attempt,
reason,
remediation,
})
}
None => ApiResponse::Ok(RepoConvergenceDryRunResponse {
status: "ok".into(),
worker_id: worker_id.as_str().to_string(),
current_state: "stale".into(),
missing_repos: vec![],
has_budget: true,
attempt_budget_remaining: 3,
time_budget_remaining_ms: 120_000,
would_attempt: true,
reason: "No convergence data — would attempt fresh sync.".into(),
remediation: vec!["No convergence data — run status refresh.".into()],
}),
}
}
async fn handle_repo_convergence_repair(
ctx: &DaemonContext,
worker_id: &WorkerId,
) -> ApiResponse<RepoConvergenceRepairResponse> {
let svc = &ctx.repo_convergence;
match svc.repair_worker(worker_id).await {
Some(previous_state) => {
let new_state = svc.get_drift_state(worker_id).await;
ApiResponse::Ok(RepoConvergenceRepairResponse {
status: "ok".into(),
worker_id: worker_id.as_str().to_string(),
action: "reset_convergence".into(),
previous_state: previous_state.to_string(),
new_state: new_state.to_string(),
message: format!(
"Worker convergence reset from {} to {}. Next convergence cycle will attempt sync.",
previous_state, new_state
),
})
}
None => ApiResponse::Ok(RepoConvergenceRepairResponse {
status: "noop".into(),
worker_id: worker_id.as_str().to_string(),
action: "none".into(),
previous_state: "stale".into(),
new_state: "stale".into(),
message: "Worker not tracked — nothing to repair.".into(),
}),
}
}
#[cfg(test)]
#[allow(clippy::assertions_on_constants)]
mod tests {
use super::*;
use crate::disk_pressure::{PressureAssessment, PressureConfidence, PressureState};
use crate::history::BuildHistory;
use crate::selection::WorkerSelector;
use crate::self_test::{SelfTestHistory, SelfTestService};
use crate::telemetry::TelemetryStore;
use crate::workers::{WorkerCapabilitiesInfo, WorkerPool, WorkerStateResponse};
use crate::{
benchmark_queue::BenchmarkQueue,
benchmark_scheduler::{BenchmarkScheduler, BenchmarkTriggerHandle, SchedulerConfig},
events::EventBus,
};
use chrono::Duration as ChronoDuration;
use rch_common::test_guard;
use rch_common::{
BuildCancellationMetadata, BuildCancellationWorkerHealth, SelfTestConfig,
WorkerCapabilities,
};
use std::sync::Arc;
use std::time::{Duration, Instant};
fn make_test_context(pool: WorkerPool) -> DaemonContext {
let history = Arc::new(SelfTestHistory::new(
crate::self_test::DEFAULT_RUN_CAPACITY,
crate::self_test::DEFAULT_RESULT_CAPACITY,
));
let self_test = Arc::new(SelfTestService::new(
pool.clone(),
SelfTestConfig::default(),
history,
));
let alert_manager = Arc::new(crate::alerts::AlertManager::new(
crate::alerts::AlertConfig::default(),
));
let events = EventBus::new(16);
DaemonContext {
pool,
worker_selector: Arc::new(WorkerSelector::new()),
history: Arc::new(BuildHistory::new(100)),
telemetry: Arc::new(TelemetryStore::new(Duration::from_secs(300), None)),
benchmark_queue: Arc::new(BenchmarkQueue::new(ChronoDuration::minutes(5))),
benchmark_trigger: make_test_benchmark_trigger(),
repo_convergence: Arc::new(crate::repo_convergence::RepoConvergenceService::new(
events.clone(),
)),
cancellation: Arc::new(crate::cancellation::CancellationOrchestrator::new(
crate::cancellation::CancellationConfig::default(),
events.clone(),
)),
events,
self_test,
alert_manager,
started_at: Instant::now(),
socket_path: "/tmp/test.sock".to_string(),
version: "0.1.0",
pid: 1234,
queue_timeout_secs: 300,
bypass_store: None,
admin_disable_store: None,
workers_config_path: None,
admission_barrier: Arc::new(tokio::sync::RwLock::new(false)),
}
}
fn make_test_benchmark_trigger() -> BenchmarkTriggerHandle {
let pool = WorkerPool::new();
let telemetry = Arc::new(TelemetryStore::new(Duration::from_secs(300), None));
let (scheduler, handle) = BenchmarkScheduler::new(
SchedulerConfig::default(),
pool,
telemetry,
EventBus::new(16),
);
let scheduler = Arc::new(scheduler);
if let Ok(runtime) = tokio::runtime::Handle::try_current() {
runtime.spawn(scheduler.run());
}
handle
}
async fn queue_resume_fixture(
queue_timeout_secs: u64,
) -> (
tempfile::TempDir,
DaemonContext,
SelectionRequest,
String,
u64,
) {
let hook_pid = crate::history::observable_test_process_id();
assert!(
crate::history::process_identity(hook_pid).is_some() && is_process_alive(hook_pid),
"queue resume integration requires one PID namespace for procfs and kill(0): \
getpid={}, procfs owner={hook_pid}; the live test owner must pass both unchanged production checks",
std::process::id(),
);
let tmp = tempfile::TempDir::new().unwrap();
let path = tmp.path().join("history.jsonl");
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("requested", 4)).await;
assert!(
pool.get(&WorkerId::new("requested"))
.await
.unwrap()
.reserve_slots(4)
.await
);
let mut ctx = make_test_context(pool);
ctx.queue_timeout_secs = queue_timeout_secs;
ctx.history = Arc::new(BuildHistory::new(10).with_persistence(path.clone()));
let request = SelectionRequest {
project: "resume-project".into(),
command: Some("cargo build".into()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: Some(9),
job_mode: false,
required_tools: vec![],
hook_pid: Some(hook_pid),
};
let wrapper = format!("rchw-{}", Uuid::new_v4());
let old_ctx = ctx.clone();
let old_request = request.clone();
let old_wrapper = wrapper.clone();
let waiter = tokio::spawn(async move {
handle_select_worker_with_wrapper(
&old_ctx,
old_request,
true,
Some(30),
Some(old_wrapper),
)
.await
});
tokio::time::timeout(Duration::from_secs(2), async {
while ctx.history.queue_depth() == 0 {
tokio::task::yield_now().await;
}
})
.await
.unwrap();
let queue_id = ctx.history.queued_builds()[0].id;
waiter.abort();
assert!(waiter.await.unwrap_err().is_cancelled());
ctx.history = Arc::new(BuildHistory::load_from_file(&path, 10).unwrap());
assert!(ctx.history.queued_build(queue_id).unwrap().recovered);
(tmp, ctx, request, wrapper, queue_id)
}
#[tokio::test]
async fn queue_resume_socket_route_continues_the_original_waiter_once() {
let _guard = test_guard!();
let (tmp, ctx, request, wrapper, queue_id) = queue_resume_fixture(300).await;
let original = ctx.history.queued_build(queue_id).unwrap();
let worker = ctx.pool.get(&WorkerId::new("requested")).await.unwrap();
worker.release_slots(4).await;
assert!(
handle_select_worker_with_wrapper(
&ctx,
request.clone(),
true,
Some(30),
Some(wrapper.clone())
)
.await
.is_err()
);
assert_eq!(worker.available_slots().await, 4);
let (mut client, server) = UnixStream::pair().unwrap();
let (shutdown_tx, _) = tokio::sync::mpsc::channel(1);
let connection = tokio::spawn(handle_connection(server, ctx.clone(), shutdown_tx));
let line = format!(
"GET /select-worker/resume-queued?project=resume-project&command=cargo%20build&cores=2&worker=requested&classification_us=9&hook_pid={}&local_wrapper_id={wrapper}&wait=1&wait_timeout_secs=30\n",
crate::history::observable_test_process_id(),
);
client.write_all(line.as_bytes()).await.unwrap();
let mut response = String::new();
tokio::time::timeout(Duration::from_secs(2), client.read_to_string(&mut response))
.await
.unwrap()
.unwrap();
connection.await.unwrap().unwrap();
assert!(response.starts_with("HTTP/1.0 200"), "{response}");
let body: serde_json::Value =
serde_json::from_str(response.split_once("\r\n\r\n").unwrap().1).unwrap();
let build_id = body["build_id"].as_u64().unwrap();
assert_eq!(body["worker"]["id"], "requested");
assert_eq!(worker.available_slots().await, 2);
assert_eq!(ctx.history.queue_depth(), 0);
assert_eq!(ctx.history.active_builds().len(), 1);
assert_eq!(
ctx.history
.active_build(build_id)
.unwrap()
.local_wrapper_id
.as_deref(),
Some(wrapper.as_str())
);
let restored = BuildHistory::load_from_file(&tmp.path().join("history.jsonl"), 10).unwrap();
assert!(restored.queued_build(original.id).is_none());
assert!(restored.active_build(build_id).is_some());
assert!(
handle_select_worker_mode(&ctx, request, true, Some(30), Some(wrapper), true)
.await
.is_err()
);
assert_eq!(worker.available_slots().await, 2);
}
#[tokio::test]
async fn queue_resume_rejects_changed_selection_constraints_without_claiming() {
let _guard = test_guard!();
let (_tmp, ctx, request, wrapper, queue_id) = queue_resume_fixture(300).await;
let worker = ctx.pool.get(&WorkerId::new("requested")).await.unwrap();
worker.release_slots(4).await;
let original = serde_json::to_value(&request).unwrap();
for (field, value) in [
("project", serde_json::json!("different-project")),
("command", serde_json::json!("cargo test")),
("estimated_cores", serde_json::json!(1)),
("disk_headroom_gib", serde_json::json!(64)),
("command_priority", serde_json::json!("high")),
("preferred_workers", serde_json::json!(["alternate"])),
(
"toolchain",
serde_json::json!({"channel":"nightly", "date":"2026-09-01", "full_version":"different compiler"}),
),
("required_runtime", serde_json::json!("rust")),
("job_mode", serde_json::json!(true)),
("required_tools", serde_json::json!(["git"])),
("classification_duration_us", serde_json::json!(10)),
("hook_pid", serde_json::json!(1)),
] {
let mut changed = original.clone();
changed[field] = value;
let changed: SelectionRequest = serde_json::from_value(changed).unwrap();
assert!(
handle_select_worker_mode(
&ctx,
changed,
true,
Some(30),
Some(wrapper.clone()),
true
)
.await
.is_err(),
"accepted changed {field}"
);
assert!(ctx.history.queued_build(queue_id).unwrap().recovered);
assert_eq!(worker.available_slots().await, 4);
}
for timeout in [None, Some(29), Some(31)] {
assert!(
handle_select_worker_mode(
&ctx,
request.clone(),
true,
timeout,
Some(wrapper.clone()),
true
)
.await
.is_err()
);
}
let response =
handle_select_worker_mode(&ctx, request, true, Some(30), Some(wrapper), true)
.await
.unwrap();
assert!(response.worker.is_some());
assert_eq!(worker.available_slots().await, 2);
}
#[tokio::test]
async fn queue_resume_expired_original_timeout_never_reserves_free_worker() {
let _guard = test_guard!();
let (tmp, mut ctx, request, wrapper, queue_id) = queue_resume_fixture(1).await;
let ownership = tmp.path().join("history.ownership.json");
let mut snapshot: serde_json::Value =
serde_json::from_slice(&std::fs::read(&ownership).unwrap()).unwrap();
snapshot["queued"][0]["queued_at"] =
serde_json::json!((Utc::now() - ChronoDuration::seconds(2)).to_rfc3339());
std::fs::write(&ownership, serde_json::to_vec(&snapshot).unwrap()).unwrap();
ctx.history =
Arc::new(BuildHistory::load_from_file(&tmp.path().join("history.jsonl"), 10).unwrap());
ctx.queue_timeout_secs = 600;
let worker = ctx.pool.get(&WorkerId::new("requested")).await.unwrap();
worker.release_slots(4).await;
let response =
handle_select_worker_mode(&ctx, request, true, Some(30), Some(wrapper), true)
.await
.unwrap();
assert_eq!(
response.reason,
SelectionReason::SelectionError("queue_timeout".into())
);
assert!(response.worker.is_none());
assert!(ctx.history.queued_build(queue_id).is_none());
assert!(ctx.history.active_builds().is_empty());
assert_eq!(worker.available_slots().await, 4);
let restored = BuildHistory::load_from_file(&tmp.path().join("history.jsonl"), 10).unwrap();
assert!(restored.queued_build(queue_id).is_none());
assert!(restored.active_builds().is_empty());
}
#[tokio::test]
async fn queue_resume_cancelled_owner_and_duplicate_waiter_never_select() {
let _guard = test_guard!();
let (_tmp, ctx, request, wrapper, queue_id) = queue_resume_fixture(300).await;
let waiting_ctx = ctx.clone();
let waiting_request = request.clone();
let waiting_wrapper = wrapper.clone();
let waiter = tokio::spawn(async move {
handle_select_worker_mode(
&waiting_ctx,
waiting_request,
true,
Some(30),
Some(waiting_wrapper),
true,
)
.await
});
tokio::time::timeout(Duration::from_secs(2), async {
while ctx.history.queued_build(queue_id).unwrap().recovered {
tokio::task::yield_now().await;
}
})
.await
.unwrap();
assert!(
handle_select_worker_mode(
&ctx,
request.clone(),
true,
Some(30),
Some(wrapper.clone()),
true
)
.await
.is_err()
);
assert!(matches!(
ctx.history.cancel_wrapper(&wrapper).unwrap(),
crate::history::WrapperCancellation::BeforeStart
));
let response = tokio::time::timeout(Duration::from_secs(2), waiter)
.await
.unwrap()
.unwrap()
.unwrap();
assert_eq!(
response.reason,
SelectionReason::SelectionError("job_cancelled_before_start".into())
);
assert!(
handle_select_worker_mode(&ctx, request, true, Some(30), Some(wrapper), true)
.await
.is_err()
);
assert!(ctx.history.active_builds().is_empty());
assert_eq!(ctx.history.queue_depth(), 0);
assert_eq!(
ctx.pool
.get(&WorkerId::new("requested"))
.await
.unwrap()
.available_slots()
.await,
0
);
}
#[test]
fn queue_resume_route_requires_an_identified_waiter_and_never_aliases_selection() {
let base = "GET /select-worker/resume-queued?project=p&wait=1&local_wrapper_id=rchw-owner&hook_pid=123";
assert!(matches!(
parse_request(base).unwrap(),
ApiRequest::SelectWorker {
resume_queued: true,
..
}
));
for line in [
"GET /select-worker/resume-queued?project=p&local_wrapper_id=rchw-owner&hook_pid=123",
"GET /select-worker/resume-queued?project=p&wait=1&hook_pid=123",
"GET /select-worker/resume-queued?project=p&wait=1&local_wrapper_id=rchw-owner",
"GET /select-worker/resume-queued-extra?project=p&wait=1&local_wrapper_id=rchw-owner&hook_pid=123",
] {
assert!(parse_request(line).is_err(), "{line}");
}
assert!(parse_request(&format!("{base}&dry_run=1")).is_err());
}
#[test]
fn verified_tools_routes_bind_requirements_to_selection_and_queue_recovery() {
for (route, resume, budget) in [
("/select-verified-tools", false, 0),
("/select-verified-tools/resume-queued", true, 0),
("/select-verified-tools/disk-budget", false, 64),
("/select-verified-tools/resume-queued/disk-budget", true, 64),
] {
let disk = if budget == 0 {
String::new()
} else {
format!("&disk_headroom_gib={budget}")
};
let base = format!(
"GET {route}?project=p&command=cargo%20xwin%20build&wait=1&local_wrapper_id=rchw-owner&hook_pid=123{disk}"
);
assert!(parse_request(&base).is_err(), "missing tools: {base}");
let ApiRequest::SelectWorker {
mut request,
resume_queued,
..
} = parse_request(&format!(
"{base}&require_tool=cargo-xwin&require_tool=llvm-ar&require_tool=cargo-xwin"
))
.expect("explicit verified tool requirements")
else {
panic!("expected selection")
};
assert_eq!(request.required_tools, vec!["cargo-xwin", "llvm-ar"]);
assert_eq!(request.command.as_deref(), Some("cargo xwin build"));
assert_eq!(request.disk_headroom_gib, budget);
assert_eq!(resume_queued, resume);
assert!(!request.job_mode);
let constrained = queue_selection_digest(&request, Some(30)).unwrap();
request.required_tools.pop();
assert_ne!(
constrained,
queue_selection_digest(&request, Some(30)).unwrap()
);
}
for line in [
"GET /select-verified-tools-extra?project=p&require_tool=cargo-xwin",
"GET /select-verified-tools?project=p&require_tool=&require_tool=%20",
"GET /select-verified-tools?project=p&require_tool=cargo-xwin&disk_headroom_gib=64",
"GET /select-verified-tools/disk-budget?project=p&require_tool=cargo-xwin",
"GET /select-verified-tools/resume-queued?project=p&require_tool=cargo-xwin",
] {
assert!(parse_request(line).is_err(), "accepted {line}");
}
}
#[test]
fn disk_headroom_routes_require_explicit_valid_budget_and_preserve_resume() {
for (route, resume) in [
("/select-worker/disk-budget", false),
("/select-worker/resume-queued/disk-budget", true),
] {
let line = format!(
"GET {route}?project=p&disk_headroom_gib=64&wait=1&local_wrapper_id=rchw-owner&hook_pid=123"
);
let ApiRequest::SelectWorker {
request,
resume_queued,
..
} = parse_request(&line).expect("budgeted selection")
else {
panic!("expected selection")
};
assert_eq!(request.disk_headroom_gib, 64);
assert_eq!(resume_queued, resume);
}
for invalid in ["", "0", "-1", "1.5", "4294967296", "no"] {
let line =
format!("GET /select-worker/disk-budget?project=p&disk_headroom_gib={invalid}");
assert!(parse_request(&line).is_err(), "accepted {line}");
}
for line in [
"GET /select-worker/disk-budget?project=p",
"GET /select-worker?project=p&disk_headroom_gib=64",
"GET /select-worker/disk-budget?project=p&disk_headroom_gib=64&disk_headroom_gib=1",
"GET /select-worker/disk-budget-extra?project=p&disk_headroom_gib=64",
"GET /select-worker/resume-queued/disk-budget?project=p&disk_headroom_gib=64",
"GET /select-worker/resume-queued?project=p&disk_headroom_gib=64&wait=1&local_wrapper_id=rchw-owner&hook_pid=123",
] {
assert!(parse_request(line).is_err(), "accepted {line}");
}
}
#[test]
fn disk_headroom_is_bound_to_queue_identity_without_changing_legacy_requests() {
let ApiRequest::SelectWorker { mut request, .. } =
parse_request("GET /select-worker?project=p&cores=4").unwrap()
else {
panic!("expected selection")
};
let old = queue_selection_digest(&request, Some(30)).unwrap();
assert_eq!(request.disk_headroom_gib, 0);
let json = serde_json::to_value(&request).unwrap();
assert!(json.get("disk_headroom_gib").is_none());
let restored = serde_json::from_value(json).unwrap();
assert_eq!(old, queue_selection_digest(&restored, Some(30)).unwrap());
request.disk_headroom_gib = 64;
let budgeted = queue_selection_digest(&request, Some(30)).unwrap();
assert_ne!(old, budgeted);
request.disk_headroom_gib = 63;
assert_ne!(
budgeted,
queue_selection_digest(&request, Some(30)).unwrap()
);
}
#[tokio::test]
async fn disk_headroom_api_selects_capacity_and_releases_only_owned_budget() {
let _guard = test_guard!();
let tmp = tempfile::tempdir().unwrap();
let history_path = tmp.path().join("history.jsonl");
let pool = WorkerPool::new();
for (id, free) in [("small", 51.0), ("large", 100.0)] {
pool.add_worker(make_test_worker(id, 8)).await;
pool.get(&WorkerId::new(id))
.await
.unwrap()
.set_capabilities(WorkerCapabilities {
rustc_version: Some("1.97.0-nightly".into()),
projects_root_ok: Some(true),
disk_free_gb: Some(free),
disk_total_gb: Some(200.0),
build_disk_free_gb: Some(free),
build_disk_total_gb: Some(200.0),
..Default::default()
})
.await;
}
let mut ctx = make_test_context(pool);
ctx.history = Arc::new(BuildHistory::new(100).with_persistence(history_path.clone()));
ctx.worker_selector = Arc::new(crate::daemon_worker_selector(
&rch_common::RchConfig::default(),
Arc::clone(&ctx.history),
None,
));
let ApiRequest::SelectWorker { request, .. } = parse_request(
"GET /select-worker/disk-budget?project=large-build&command=cargo%20test&cores=1&disk_headroom_gib=64&runtime=none"
).unwrap() else { panic!("expected selection") };
let preview = handle_select_worker_dry_run(&ctx, &request).await;
assert_eq!(
preview.worker.as_ref().map(|worker| worker.id.as_str()),
Some("large")
);
assert!(preview.build_id.is_none());
assert_eq!(ctx.history.reserved_disk_headroom_gib("large"), 0);
let response = handle_select_worker_with_wrapper(
&ctx,
request.clone(),
false,
None,
Some("rchw-disk-owner".into()),
)
.await
.unwrap();
assert_eq!(response.worker.unwrap().id.as_str(), "large");
let build_id = response.build_id.unwrap();
assert_eq!(ctx.history.reserved_disk_headroom_gib("large"), 64);
assert_eq!(ctx.history.reserved_disk_headroom_gib("small"), 0);
let restored = BuildHistory::load_from_file(&history_path, 100).unwrap();
assert_eq!(restored.reserved_disk_headroom_gib("large"), 64);
let mut second = request.clone();
second.project = "another-large-build".into();
let refused = handle_select_worker_with_wrapper(
&ctx,
second.clone(),
false,
None,
Some("rchw-second-disk-owner".into()),
)
.await
.unwrap();
assert!(refused.worker.is_none());
assert!(
refused.reason.to_string().contains("disk_headroom"),
"{:?}",
refused.reason
);
assert_eq!(ctx.history.active_builds().len(), 1);
let worker = ctx.pool.get(&WorkerId::new("large")).await.unwrap();
assert_eq!(worker.used_slots(), 1);
let release = |owner: &str| ReleaseRequest {
local_wrapper_id: Some(owner.into()),
worker_id: WorkerId::new("large"),
slots: 99,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
assert!(
handle_release_worker(&ctx, release("rchw-wrong-owner"))
.await
.is_err()
);
assert_eq!(ctx.history.reserved_disk_headroom_gib("large"), 64);
assert_eq!(worker.used_slots(), 1);
handle_release_worker(&ctx, release("rchw-disk-owner"))
.await
.unwrap();
assert_eq!(ctx.history.reserved_disk_headroom_gib("large"), 0);
assert_eq!(worker.used_slots(), 0);
let stale = handle_select_worker_with_wrapper(
&ctx,
second.clone(),
false,
None,
Some("rchw-second-disk-owner".into()),
)
.await
.unwrap();
assert!(stale.worker.is_none());
assert!(stale.build_id.is_none());
let diagnostic = stale
.diagnostics
.unwrap()
.workers
.into_iter()
.find(|entry| entry.worker_id.as_str() == "large")
.unwrap();
assert!(
diagnostic
.reason_codes
.iter()
.any(|reason| reason == "disk_headroom_stale")
);
assert!(ctx.history.active_builds().is_empty());
assert_eq!(worker.used_slots(), 0);
let refreshed = worker.capabilities().await;
worker.set_capabilities(refreshed).await;
let next = handle_select_worker_with_wrapper(
&ctx,
second,
false,
None,
Some("rchw-second-disk-owner".into()),
)
.await
.unwrap();
assert_eq!(next.worker.unwrap().id.as_str(), "large");
assert_eq!(ctx.history.reserved_disk_headroom_gib("large"), 64);
handle_release_worker(&ctx, release("rchw-disk-owner"))
.await
.unwrap();
assert_eq!(
ctx.history.reserved_disk_headroom_gib("large"),
64,
"duplicate completion must not release the new build's budget"
);
assert_eq!(worker.used_slots(), 1);
}
#[test]
fn test_parse_request_basic() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=myproject&cores=4").unwrap();
assert!(matches!(req, ApiRequest::SelectWorker { .. }));
let ApiRequest::SelectWorker { request: req, .. } = req else {
return;
};
assert_eq!(req.project, "myproject");
assert_eq!(req.estimated_cores, 4);
}
#[test]
fn test_parse_request_project_only() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=test").unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.project, "test");
assert_eq!(req.estimated_cores, 1); }
#[test]
fn test_parse_request_dry_run_flag() {
let _guard = test_guard!();
for (line, expected) in [
("GET /select-worker?project=p&cores=2&dry_run=1", true),
("GET /select-worker?project=p&cores=2&dry_run=true", true),
("GET /select-worker?project=p&cores=2&dry_run=0", false),
("GET /select-worker?project=p&cores=2", false),
] {
let ApiRequest::SelectWorker { dry_run, .. } = parse_request(line).unwrap() else {
panic!("expected select-worker request for {line}");
};
assert_eq!(dry_run, expected, "{line}");
}
}
#[tokio::test]
async fn test_dry_run_selection_reserves_nothing() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("dry-worker", 8)).await;
let worker = pool.get(&WorkerId::new("dry-worker")).await.unwrap();
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "dry-run-project".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 4,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: None,
};
let response = handle_select_worker_dry_run(&ctx, &request).await;
let selected = response.worker.expect("dry run should report a worker");
assert_eq!(selected.id.as_str(), "dry-worker");
assert_eq!(selected.slots_available, 8);
assert!(response.build_id.is_none());
assert_eq!(worker.used_slots(), 0);
assert!(
ctx.history
.active_workers_for_project("dry-run-project")
.is_empty()
);
}
#[test]
fn test_parse_request_zero_cores_defaults_to_one() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=test&cores=0").unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.estimated_cores, 1);
}
#[test]
fn test_parse_request_with_spaces() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=my%20project&cores=2").unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.project, "my project");
assert_eq!(req.estimated_cores, 2);
}
#[test]
fn test_parse_request_carries_valid_local_wrapper_id_separately_from_command() {
let _guard = test_guard!();
let request = parse_request(
"GET /select-worker?project=test&command=cargo%20build&local_wrapper_id=rchw-lease-1",
)
.unwrap();
let ApiRequest::SelectWorker {
request,
local_wrapper_id,
..
} = request
else {
panic!("expected select-worker request");
};
assert_eq!(request.command.as_deref(), Some("cargo build"));
assert_eq!(local_wrapper_id.as_deref(), Some("rchw-lease-1"));
}
#[test]
fn test_parse_request_with_priority_hint() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=test&cores=2&priority=high").unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.command_priority, rch_common::CommandPriority::High);
}
#[test]
fn test_parse_request_with_preferred_workers() {
let _guard = test_guard!();
let req = parse_request(
"GET /select-worker?project=test&worker=ts2&workers=vmi1%2Cvmi2&preferred=fast",
)
.unwrap();
assert!(matches!(req, ApiRequest::SelectWorker { .. }));
let ApiRequest::SelectWorker { request: req, .. } = req else {
return;
};
let ids: Vec<&str> = req
.preferred_workers
.iter()
.map(|worker| worker.as_str())
.collect();
assert_eq!(ids, vec!["ts2", "vmi1", "vmi2", "fast"]);
}
#[test]
fn test_parse_request_invalid_priority_defaults_to_normal() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=test&cores=2&priority=urgent").unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.command_priority, rch_common::CommandPriority::Normal);
}
#[test]
fn test_parse_request_status() {
let _guard = test_guard!();
let req = parse_request("GET /status").unwrap();
assert!(matches!(req, ApiRequest::Status), "expected status request");
}
#[test]
fn test_parse_request_test_run() {
let _guard = test_guard!();
let req = parse_request("POST /test-run").unwrap();
assert!(
matches!(req, ApiRequest::TestRun),
"expected test run request"
);
}
#[test]
fn test_parse_request_build_heartbeat() {
let _guard = test_guard!();
let req = parse_request("POST /build-heartbeat").unwrap();
assert!(
matches!(req, ApiRequest::BuildHeartbeat),
"expected build-heartbeat request"
);
}
#[test]
fn test_parse_request_speedscore() {
let _guard = test_guard!();
let req = parse_request("GET /speedscore/css").unwrap();
match req {
ApiRequest::SpeedScore { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected speedscore request"),
}
}
#[test]
fn test_parse_request_speedscore_history() {
let _guard = test_guard!();
let req = parse_request("GET /speedscore/css/history?days=7&limit=25&offset=5").unwrap();
match req {
ApiRequest::SpeedScoreHistory {
worker_id,
days,
limit,
offset,
} => {
assert_eq!(worker_id.as_str(), "css");
assert_eq!(days, 7);
assert_eq!(limit, 25);
assert_eq!(offset, 5);
}
_ => assert!(false, "expected speedscore history request"),
}
}
#[test]
fn test_parse_request_speedscores() {
let _guard = test_guard!();
let req = parse_request("GET /speedscores").unwrap();
assert!(
matches!(req, ApiRequest::SpeedScores),
"expected speedscores request"
);
}
#[test]
fn test_parse_request_workers_capabilities() {
let _guard = test_guard!();
let req = parse_request("GET /workers/capabilities").unwrap();
match req {
ApiRequest::WorkersCapabilities { refresh } => {
assert!(!refresh);
}
_ => assert!(false, "expected workers capabilities request"),
}
let req = parse_request("GET /workers/capabilities?refresh=true").unwrap();
match req {
ApiRequest::WorkersCapabilities { refresh } => {
assert!(refresh);
}
_ => assert!(false, "expected workers capabilities request"),
}
}
#[test]
fn test_parse_request_benchmark_trigger() {
let _guard = test_guard!();
let req = parse_request("POST /benchmark/trigger/css").unwrap();
match req {
ApiRequest::BenchmarkTrigger { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected benchmark trigger request"),
}
let req = parse_request("POST /benchmark/trigger?worker=css").unwrap();
match req {
ApiRequest::BenchmarkTrigger { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected benchmark trigger query request"),
}
}
#[test]
fn test_parse_request_events() {
let _guard = test_guard!();
let req = parse_request("GET /events").unwrap();
assert!(matches!(req, ApiRequest::Events), "expected events request");
}
#[tokio::test]
async fn socket_request_duration_records_real_responses_and_errors() {
use prometheus::Encoder;
let _guard = test_guard!();
let registry = prometheus::Registry::new();
let observations = rch_telemetry::metrics::Metrics::new().expect("metrics");
observations.register(®istry).expect("register metrics");
let histogram = observations
.request_duration_seconds
.with_label_values(&["rchd_api"]);
for request in ["GET /health\n", "GET /unknown-endpoint\n"] {
let (mut client, server) = UnixStream::pair().expect("socket pair");
let (shutdown_tx, _shutdown_rx) = tokio::sync::mpsc::channel(1);
let task = tokio::spawn(handle_connection_with_metrics(
server,
make_test_context(WorkerPool::new()),
shutdown_tx,
Some(observations.clone()),
));
client.write_all(request.as_bytes()).await.expect("request");
let mut response = String::new();
tokio::time::timeout(Duration::from_secs(2), client.read_to_string(&mut response))
.await
.expect("response deadline")
.expect("response bytes");
let result = task.await.expect("join handler");
if request.contains("/health") {
result.expect("write health response");
assert!(response.starts_with("HTTP/1.0"), "{response}");
let body = response.split_once("\r\n\r\n").expect("HTTP body").1;
let body: serde_json::Value = serde_json::from_str(body).expect("JSON response");
assert!(body.get("status").is_some(), "{body}");
} else {
let error = result.expect_err("unknown route returns early");
assert!(error.to_string().contains("Unknown endpoint"), "{error}");
assert!(response.is_empty(), "no response was written: {response}");
}
}
assert_eq!(histogram.get_sample_count(), 2);
let mut text = Vec::new();
prometheus::TextEncoder::new()
.encode(®istry.gather(), &mut text)
.expect("encode registered collector");
assert!(
String::from_utf8(text)
.unwrap()
.contains("rch_request_duration_seconds_count{entrypoint=\"rchd_api\"} 2\n")
);
}
#[tokio::test]
async fn socket_request_duration_observes_cancellation_but_excludes_event_streams() {
let _guard = test_guard!();
let observations = rch_telemetry::metrics::Metrics::new().expect("metrics");
let histogram = observations
.request_duration_seconds
.with_label_values(&["rchd_api"]);
let (mut client, server) = UnixStream::pair().expect("socket pair");
let (shutdown_tx, _shutdown_rx) = tokio::sync::mpsc::channel(1);
client
.write_all(b"POST /build-heartbeat\n")
.await
.expect("request without body");
let request = handle_connection_with_metrics(
server,
make_test_context(WorkerPool::new()),
shutdown_tx,
Some(observations.clone()),
);
assert!(
tokio::time::timeout(Duration::from_millis(20), request)
.await
.is_err(),
"cancel the handler while it waits for the heartbeat body"
);
assert_eq!(histogram.get_sample_count(), 1, "cancelled finite request");
let (mut client, server) = UnixStream::pair().expect("socket pair");
let (shutdown_tx, _shutdown_rx) = tokio::sync::mpsc::channel(1);
let task = tokio::spawn(handle_connection_with_metrics(
server,
make_test_context(WorkerPool::new()),
shutdown_tx,
Some(observations),
));
client.write_all(b"GET /events\n").await.expect("subscribe");
let mut reader = BufReader::new(client);
let mut header = String::new();
tokio::time::timeout(Duration::from_secs(2), reader.read_line(&mut header))
.await
.expect("stream header deadline")
.expect("stream header");
assert_eq!(header, "HTTP/1.0 200 OK\r\n");
task.abort();
assert!(task.await.expect_err("stream cancelled").is_cancelled());
assert_eq!(histogram.get_sample_count(), 1, "stream lifetime excluded");
}
#[tokio::test]
async fn test_reload_uses_launch_time_workers_config_path() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let _guard = test_guard!();
let dir = tempfile::tempdir().expect("tempdir");
let workers_path = dir.path().join("launch-workers.toml");
std::fs::write(
&workers_path,
"[[workers]]\nid = \"reload-launch-path-w1\"\nhost = \"127.0.0.1\"\ntotal_slots = 2\n",
)
.expect("write workers.toml");
let pool = WorkerPool::new();
let mut ctx = make_test_context(pool.clone());
ctx.workers_config_path = Some(workers_path);
let (client, server) = tokio::net::UnixStream::pair().expect("socket pair");
let (shutdown_tx, _shutdown_rx) = tokio::sync::mpsc::channel(1);
let server_task = tokio::spawn(handle_connection(server, ctx, shutdown_tx));
let (mut reader, mut writer) = client.into_split();
writer
.write_all(b"POST /reload\n")
.await
.expect("send reload");
let mut raw = String::new();
reader
.read_to_string(&mut raw)
.await
.expect("read response");
server_task.await.expect("join").expect("handle_connection");
let body = raw
.split_once("\r\n\r\n")
.map(|(_, body)| body)
.unwrap_or(&raw);
let response: serde_json::Value =
serde_json::from_str(body.trim()).expect("reload response body is JSON");
assert_eq!(response["success"], true, "response: {response}");
assert_eq!(response["added"], 1, "response: {response}");
assert!(
pool.get(&WorkerId::new("reload-launch-path-w1"))
.await
.is_some(),
"worker from the launch-time workers.toml must be in the pool"
);
}
#[test]
fn test_parse_request_reload() {
let _guard = test_guard!();
let req = parse_request("POST /reload").unwrap();
assert!(matches!(req, ApiRequest::Reload), "expected reload request");
}
#[test]
fn test_parse_request_reload_get_fails() {
let _guard = test_guard!();
let result = parse_request("GET /reload");
assert!(result.is_err());
}
#[test]
fn test_parse_request_telemetry_poll() {
let _guard = test_guard!();
let req = parse_request("POST /telemetry/poll?worker=css").unwrap();
match req {
ApiRequest::TelemetryPoll { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected telemetry poll request"),
}
}
#[test]
fn test_parse_request_self_test_status() {
let _guard = test_guard!();
let req = parse_request("GET /self-test/status").unwrap();
assert!(
matches!(req, ApiRequest::SelfTestStatus),
"expected self-test status request"
);
}
#[test]
fn test_parse_request_self_test_history() {
let _guard = test_guard!();
let req = parse_request("GET /self-test/history?limit=5").unwrap();
match req {
ApiRequest::SelfTestHistory { limit } => assert_eq!(limit, 5),
_ => assert!(false, "expected self-test history request"),
}
}
#[test]
fn test_parse_request_self_test_run() {
let _guard = test_guard!();
let req =
parse_request("POST /self-test/run?worker=css&timeout=120&debug=1&scheduled=false")
.unwrap();
let ApiRequest::SelfTestRun(req) = req else {
assert!(false, "expected self-test run request");
return;
};
assert_eq!(req.worker_ids, vec!["css".to_string()]);
assert_eq!(req.timeout_secs, Some(120));
assert!(!req.release_mode);
assert!(!req.scheduled);
}
#[test]
fn test_parse_request_with_toolchain() {
let _guard = test_guard!();
let toolchain_json = r###"{"channel":"nightly","date":"2024-01-01","full_version":"rustc 1.76.0-nightly"}"###;
let encoded = urlencoding_encode(toolchain_json);
let query = format!(
"GET /select-worker?project=test&cores=4&toolchain={}",
encoded
);
let req = parse_request(&query).unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.project, "test");
assert_eq!(req.estimated_cores, 4);
assert!(req.toolchain.is_some());
let tc = req.toolchain.unwrap();
assert_eq!(tc.channel, "nightly");
assert_eq!(tc.date, Some("2024-01-01".to_string()));
}
#[test]
fn test_parse_request_with_runtime() {
let _guard = test_guard!();
let query = "GET /select-worker?project=test&runtime=bun";
let req = parse_request(query).unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.required_runtime, RequiredRuntime::Bun);
let query = "GET /select-worker?project=test&runtime=rust";
let req = parse_request(query).unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.required_runtime, RequiredRuntime::Rust);
let query = "GET /select-worker?project=test&runtime=invalid";
let req = parse_request(query).unwrap();
let ApiRequest::SelectWorker { request: req, .. } = req else {
assert!(false, "expected select-worker request");
return;
};
assert_eq!(req.required_runtime, RequiredRuntime::None);
}
#[test]
fn test_parse_request_missing_project() {
let _guard = test_guard!();
let result = parse_request("GET /select-worker?cores=4");
assert!(result.is_err());
}
#[test]
fn test_parse_request_invalid_method() {
let _guard = test_guard!();
let result = parse_request("PUT /select-worker?project=test");
assert!(result.is_err());
}
#[test]
fn test_parse_request_unknown_endpoint() {
let _guard = test_guard!();
let result = parse_request("GET /unknown?project=test");
assert!(result.is_err());
}
#[test]
fn test_parse_request_rejects_route_prefix_typos() {
let _guard = test_guard!();
let cases = [
"POST /benchmark/trigger-extra?worker=css",
"GET /self-test/history-extra?limit=5",
"POST /self-test/run-extra?worker=css",
"POST /release-worker-extra?worker=css&slots=4",
"POST /record-build-extra?worker=css&project=myproject",
"POST /build-heartbeat-extra",
"POST /test-run-extra",
"GET /select-worker-extra?project=test",
];
for request in cases {
assert!(
parse_request(request).is_err(),
"route typo should be rejected: {request}"
);
}
}
#[test]
fn test_percent_unescape_query_value_basic() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("hello%20world"), "hello world");
assert_eq!(
percent_unescape_query_value("path%2Fto%2Ffile"),
"path/to/file"
);
assert_eq!(percent_unescape_query_value("foo%3Abar"), "foo:bar");
}
#[test]
fn test_percent_unescape_query_value_special_chars() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("a%26b%3Dc"), "a&b=c");
assert_eq!(percent_unescape_query_value("100%25"), "100%");
assert_eq!(percent_unescape_query_value("hello%2Bworld"), "hello+world");
}
#[test]
fn test_percent_unescape_query_value_plus_as_space() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("hello+world"), "hello world");
}
#[test]
fn test_percent_unescape_query_value_no_escapes() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("simple"), "simple");
assert_eq!(
percent_unescape_query_value("with-dash_underscore"),
"with-dash_underscore"
);
}
#[test]
fn test_percent_unescape_query_value_invalid() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("foo%GGbar"), "foo%GGbar");
assert_eq!(percent_unescape_query_value("foo%"), "foo%");
}
#[test]
fn test_percent_unescape_query_value_utf8() {
let _guard = test_guard!();
assert_eq!(percent_unescape_query_value("%C3%A9"), "é");
assert_eq!(
percent_unescape_query_value("%E3%81%93%E3%82%93%E3%81%AB%E3%81%A1%E3%81%AF"),
"こんにちは"
);
assert_eq!(
percent_unescape_query_value("hello%20%F0%9F%8C%8D"),
"hello 🌍"
);
}
fn urlencoding_encode(s: &str) -> String {
let mut result = String::with_capacity(s.len() * 3);
for c in s.chars() {
match c {
'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => result.push(c),
_ => {
for byte in c.to_string().as_bytes() {
result.push('%');
result.push_str(&format!("{:02X}", byte));
}
}
}
}
result
}
use rch_common::{CommandPriority, RequiredRuntime, WorkerConfig, WorkerId, WorkerStatus};
fn make_test_worker(id: &str, total_slots: u32) -> WorkerConfig {
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: vec![],
tools: Vec::new(),
}
}
#[tokio::test]
async fn test_handle_select_worker_no_workers_configured() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 4,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(response.worker.is_none());
assert_eq!(response.reason, SelectionReason::NoWorkersConfigured);
}
#[tokio::test]
async fn test_handle_select_worker_all_unreachable() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
pool.set_status(&WorkerId::new("worker1"), WorkerStatus::Unreachable)
.await;
pool.set_status(&WorkerId::new("worker2"), WorkerStatus::Unreachable)
.await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(response.worker.is_none());
assert_eq!(response.reason, SelectionReason::AllWorkersUnreachable);
}
#[tokio::test]
async fn test_handle_select_worker_all_busy() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 4)).await;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
worker.reserve_slots(4).await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(response.worker.is_none());
assert_eq!(response.reason, SelectionReason::AllWorkersBusy);
}
#[tokio::test]
async fn test_handle_select_worker_requested_busy_does_not_use_available_alternate() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("requested", 4)).await;
pool.add_worker(make_test_worker("alternate", 8)).await;
let requested = pool.get(&WorkerId::new("requested")).await.unwrap();
assert!(requested.reserve_slots(4).await);
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(response.worker.is_none());
assert_eq!(response.reason, SelectionReason::AllWorkersBusy);
assert!(response.diagnostics.is_some());
}
#[tokio::test]
async fn queued_job_cancellation_reaches_waiter_without_reserving_slots() {
let _guard = test_guard!();
for by_queue_id in [false, true] {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("requested", 4)).await;
let worker = pool.get(&WorkerId::new("requested")).await.unwrap();
assert!(worker.reserve_slots(4).await);
let tmp = tempfile::TempDir::new().unwrap();
let path = tmp.path().join("history.jsonl");
let mut ctx = make_test_context(pool);
ctx.history = Arc::new(BuildHistory::new(10).with_persistence(path.clone()));
let wrapper = format!("rchw-{}", Uuid::new_v4());
let request = SelectionRequest {
job_mode: false,
project: "cancel-queued".into(),
command: Some("cargo build".into()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: Some(std::process::id()),
required_tools: Vec::new(),
};
let waiting_ctx = ctx.clone();
let waiting_request = request.clone();
let waiting_wrapper = wrapper.clone();
let waiter = tokio::spawn(async move {
handle_select_worker_with_wrapper(
&waiting_ctx,
waiting_request,
true,
Some(5),
Some(waiting_wrapper),
)
.await
});
tokio::time::timeout(Duration::from_secs(2), async {
while ctx.history.queue_depth() == 0 {
tokio::time::sleep(Duration::from_millis(5)).await;
}
})
.await
.unwrap();
let queue_id = ctx.history.queued_builds()[0].id;
let route = if by_queue_id {
format!("POST /builds/{queue_id}/cancel\n")
} else {
format!("POST /jobs/{wrapper}/cancel\n")
};
let (mut client, server) = UnixStream::pair().unwrap();
let (shutdown_tx, _shutdown_rx) = tokio::sync::mpsc::channel(1);
let cancellation = tokio::spawn(handle_connection(server, ctx.clone(), shutdown_tx));
client.write_all(route.as_bytes()).await.unwrap();
let mut response = String::new();
tokio::time::timeout(Duration::from_secs(2), client.read_to_string(&mut response))
.await
.unwrap()
.unwrap();
cancellation.await.unwrap().unwrap();
let body: serde_json::Value =
serde_json::from_str(response.split_once("\r\n\r\n").unwrap().1).unwrap();
assert_eq!(
body["status"],
if by_queue_id {
"cancelled"
} else {
"cancelled_before_start"
}
);
assert_eq!(body["local_wrapper_id"], wrapper);
assert_eq!(body["slots_released"], 0);
worker.release_slots(4).await;
let response = tokio::time::timeout(Duration::from_secs(3), waiter)
.await
.unwrap()
.unwrap()
.unwrap();
assert_eq!(response.reason, cancelled_selection().reason);
assert!(response.worker.is_none());
assert!(response.build_id.is_none());
assert!(ctx.history.active_builds().is_empty());
assert_eq!(ctx.history.queue_depth(), 0);
assert_eq!(worker.available_slots().await, 4);
ctx.history = Arc::new(BuildHistory::load_from_file(&path, 10).unwrap());
let retry = handle_select_worker_with_wrapper(
&ctx,
request,
false,
None,
Some(wrapper.clone()),
)
.await
.unwrap();
assert_eq!(retry.reason, cancelled_selection().reason);
assert_eq!(worker.available_slots().await, 4);
assert_eq!(
handle_cancel_job(&ctx, &wrapper, false).await.unwrap()["status"],
"cancelled_before_start"
);
assert_eq!(
handle_cancel_job(&ctx, "unknown-wrapper", false)
.await
.unwrap()["status"],
"not_queued"
);
}
}
#[test]
fn queued_job_cancel_route_requires_complete_wrapper_identity() {
let wrapper = format!("rchw-{}", Uuid::new_v4());
assert!(
matches!(parse_request(&format!("POST /jobs/{wrapper}/cancel")).unwrap(), ApiRequest::CancelJob { local_wrapper_id } if local_wrapper_id == wrapper)
);
for route in [
"POST /jobs/rchw-invalid/cancel",
"POST /jobs/1/cancel",
"POST /jobs/../cancel",
] {
assert!(parse_request(route).is_err());
}
}
#[tokio::test]
async fn test_handle_select_worker_requested_busy_queue_waits_for_requested_worker() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("requested", 4)).await;
pool.add_worker(make_test_worker("alternate", 8)).await;
let requested = pool.get(&WorkerId::new("requested")).await.unwrap();
assert!(requested.reserve_slots(4).await);
let release_target = requested.clone();
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(50)).await;
release_target.release_slots(4).await;
});
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, true, Some(2))
.await
.unwrap();
let selected = response
.worker
.expect("requested worker should become available");
assert_eq!(selected.id.as_str(), "requested");
}
#[tokio::test]
async fn test_handle_select_worker_impossible_requested_capacity_never_queues() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("requested", 1)).await;
pool.add_worker(make_test_worker("alternate", 8)).await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("requested")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = tokio::time::timeout(
Duration::from_millis(200),
handle_select_worker(&ctx, request, true, Some(2)),
)
.await
.expect("impossible capacity must return without entering the queue")
.unwrap();
assert!(response.worker.is_none());
assert!(matches!(
response.reason,
SelectionReason::NoAdmissibleWorkers(ref summary)
if summary.contains("insufficient_total_slots=1")
));
}
#[tokio::test]
async fn test_handle_select_worker_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(response.worker.is_some());
assert_eq!(response.reason, SelectionReason::Success);
let worker = response.worker.unwrap();
assert_eq!(worker.id.as_str(), "worker1");
assert_eq!(worker.slots_available, 6);
}
#[tokio::test]
async fn test_admission_barrier_refuses_new_worker_selection() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
*ctx.admission_barrier.write().await = true;
let request = SelectionRequest {
job_mode: false,
project: "restart-guard".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 1,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(1234),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.expect("barrier returns a structured refusal");
assert!(response.worker.is_none());
assert_eq!(
response.reason,
SelectionReason::SelectionError("restart_admission_barrier_active".to_string())
);
assert!(ctx.history.active_builds().is_empty());
}
fn make_test_lease(heartbeat_unix_ms: u64, wrapper_pid: u32) -> DurableJobLease {
let mut identity = rch_common::job_identity::JobIdentity::new_local();
identity.admit(42);
let mut lease = DurableJobLease::new(
identity,
wrapper_pid,
None,
None,
heartbeat_unix_ms,
false,
true,
"blake3:test".to_string(),
);
lease.admit(42, "worker1".to_string(), heartbeat_unix_ms);
lease
}
#[test]
fn scan_reaps_only_nonblocking_stale_leases_and_reports_blockers() {
let dir = std::env::temp_dir().join(format!("rch-lease-scan-test-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let now = u64::try_from(chrono::Utc::now().timestamp_millis()).unwrap();
let write = |name: &str, lease: &DurableJobLease| {
std::fs::write(dir.join(name), serde_json::to_vec(lease).unwrap()).unwrap();
};
let fresh = make_test_lease(now, 1);
write("fresh.json", &fresh);
let mut done = make_test_lease(now - LEASE_REAP_RETENTION_MS - 1, 1);
done.acknowledge_terminal(now - LEASE_REAP_RETENTION_MS - 1);
write("done-old.json", &done);
let mut recent = make_test_lease(now - 1000, 1);
recent.acknowledge_terminal(now - 1000);
write("done-recent.json", &recent);
std::fs::write(dir.join("README.txt"), b"not a lease").unwrap();
let blocked = scan_client_leases(&dir).unwrap();
assert_eq!(blocked.len(), 1, "only the fresh lease blocks: {blocked:?}");
assert!(
!dir.join("done-old.json").exists(),
"stale non-blocker reaped"
);
assert!(
dir.join("done-recent.json").exists(),
"recent non-blocker kept"
);
assert!(dir.join("fresh.json").exists(), "blocker kept");
assert!(dir.join("README.txt").exists(), "non-json untouched");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn scan_never_reaps_a_dead_lease_that_still_owns_worker_source() {
let dir = tempfile::tempdir().unwrap();
let now = u64::try_from(chrono::Utc::now().timestamp_millis()).unwrap();
let stale = now - LEASE_REAP_RETENTION_MS - 1;
let dead_pid = 4_194_304 * 2;
let write = |name: &str, pid: u32, recipe: serde_json::Value| {
let mut lease = make_test_lease(stale, pid);
lease.recovery = Some(recipe);
std::fs::write(dir.path().join(name), serde_json::to_vec(&lease).unwrap()).unwrap();
};
let owning = |retired: bool| serde_json::json!({ "source_roots": ["/p"], "pair": null, "retire_root": null, "retired": retired });
write("owns-source.json", dead_pid, owning(false));
write("retired.json", dead_pid, owning(true));
write(
"owns-nothing.json",
dead_pid,
serde_json::json!({ "source_roots": [], "pair": null, "retire_root": null, "retired": false }),
);
let blocked = scan_client_leases(dir.path()).unwrap();
assert!(blocked.is_empty(), "{blocked:?}");
assert!(
dir.path().join("owns-source.json").exists(),
"the only authority able to release a worker claim was reaped"
);
assert!(
!dir.path().join("retired.json").exists(),
"retired history is reaped"
);
assert!(
!dir.path().join("owns-nothing.json").exists(),
"a recipe owning nothing on the worker is plain history"
);
}
#[test]
fn is_process_alive_uses_the_syscall_truthfully() {
assert!(is_process_alive(std::process::id()));
assert!(!is_process_alive(0));
}
#[test]
fn test_lease_blocks_restart_acknowledged_terminal_never_blocks() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let mut lease = make_test_lease(1, 1);
lease.acknowledge_terminal(now);
assert!(!lease_blocks_restart(&lease, now, || unreachable!(
"terminal+acknowledged must not probe liveness"
)));
}
#[test]
fn test_lease_blocks_restart_refused_before_admission_never_blocks() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let mut lease = DurableJobLease::new(
rch_common::job_identity::JobIdentity::new_local(),
1,
None,
None,
now,
false,
true,
"blake3:test".to_string(),
);
lease.state = rch_common::job_identity::JobLifecycleState::FailedBeforeAdmission;
lease.heartbeat("failed_before_admission", now);
lease.terminal_acknowledged = true;
assert!(!lease_blocks_restart(&lease, now, || unreachable!(
"a retired refusal must not probe liveness"
)));
}
#[test]
fn test_lease_blocks_restart_fresh_heartbeat_blocks_without_liveness_probe() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let lease = make_test_lease(now - DEAD_WRAPPER_HEARTBEAT_STALE_MS + 1, 1);
assert!(lease_blocks_restart(&lease, now, || unreachable!(
"a fresh heartbeat must not probe liveness"
)));
}
#[test]
fn test_lease_blocks_restart_stale_heartbeat_alive_pid_blocks() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let lease = make_test_lease(now - DEAD_WRAPPER_HEARTBEAT_STALE_MS, 1);
assert!(lease_blocks_restart(&lease, now, || true));
}
#[test]
fn test_lease_blocks_restart_stale_heartbeat_dead_pid_unblocks() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let lease = make_test_lease(now - DEAD_WRAPPER_HEARTBEAT_STALE_MS, 1);
assert!(!lease_blocks_restart(&lease, now, || false));
}
#[test]
fn test_lease_blocks_restart_zero_pid_stays_blocking_even_when_stale() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let lease = make_test_lease(1, 0);
assert!(lease_blocks_restart(&lease, now, || unreachable!(
"a zero pid must not probe liveness"
)));
}
#[test]
fn test_lease_blocks_restart_unacknowledged_terminal_dead_wrapper_unblocks() {
let now = 10 * DEAD_WRAPPER_HEARTBEAT_STALE_MS;
let mut lease = make_test_lease(1, 1);
lease.state = rch_common::job_identity::JobLifecycleState::Finished;
assert!(!lease.terminal_acknowledged);
assert!(!lease_blocks_restart(&lease, now, || false));
assert!(lease_blocks_restart(&lease, now, || true));
}
#[tokio::test]
async fn test_restart_admission_reports_active_build_and_keeps_barrier_closed() {
let ctx = make_test_context(WorkerPool::new());
let active = ctx.history.start_active_build(
"restart-proof".to_string(),
"worker-a".to_string(),
"cargo test -p rchd".to_string(),
4242,
1,
rch_common::BuildLocation::Remote,
);
let response = handle_restart_admission(&ctx, true).await;
assert!(!response.restart_permitted);
assert_eq!(response.active_build_ids, vec![active.id]);
assert!(response.queued_build_ids.is_empty());
assert!(*ctx.admission_barrier.read().await);
}
#[tokio::test]
async fn test_handle_select_worker_preserves_affinity_pin_reason() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
let ctx = make_test_context(pool);
ctx.worker_selector
.record_success("worker2", "test-project")
.await;
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert_eq!(response.reason, SelectionReason::AffinityPinned);
let worker = response.worker.expect("affinity-pinned worker");
assert_eq!(worker.id.as_str(), "worker2");
assert_eq!(worker.slots_available, 6);
}
#[tokio::test]
async fn test_handle_select_worker_tracks_active_build_when_hook_pid_present() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test-project".to_string(),
command: Some("cargo build --release".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(4242),
};
let response = handle_select_worker_with_wrapper(
&ctx,
request,
false,
None,
Some("rchw-daemon-correlation".to_string()),
)
.await
.unwrap();
assert_eq!(response.reason, SelectionReason::Success);
let build_id = response.build_id.expect("build_id should be assigned");
let active = ctx.history.active_build(build_id).expect("active build");
assert_eq!(active.project_id, "test-project");
assert_eq!(active.worker_id, "worker1");
assert_eq!(active.command, "cargo build --release");
assert_eq!(active.hook_pid, 4242);
assert_eq!(
active.local_wrapper_id.as_deref(),
Some("rchw-daemon-correlation")
);
assert_eq!(active.slots, 2);
handle_release_worker(
&ctx,
ReleaseRequest {
local_wrapper_id: Some("rchw-daemon-correlation".to_string()),
worker_id: WorkerId::new("worker1"),
slots: 2,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
},
)
.await
.unwrap();
assert!(ctx.history.active_build(build_id).is_none());
let recent = ctx.history.recent(10);
assert_eq!(recent.len(), 1);
assert_eq!(recent[0].id, build_id);
assert_eq!(recent[0].project_id, "test-project");
assert_eq!(recent[0].worker_id.as_deref(), Some("worker1"));
assert_eq!(recent[0].exit_code, 0);
assert_eq!(recent[0].location, rch_common::BuildLocation::Remote);
}
#[tokio::test]
async fn test_handle_select_worker_blocks_same_project_on_same_worker() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let first_request = SelectionRequest {
job_mode: false,
project: "shared-project".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(1001),
};
let first_response = handle_select_worker(&ctx, first_request, false, None)
.await
.unwrap();
assert_eq!(first_response.reason, SelectionReason::Success);
assert_eq!(
first_response.worker.as_ref().map(|w| w.id.as_str()),
Some("worker1")
);
let second_request = SelectionRequest {
job_mode: false,
project: "shared-project".to_string(),
command: Some("cargo test".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(1002),
};
let second_response = handle_select_worker(&ctx, second_request, false, None)
.await
.unwrap();
assert!(second_response.worker.is_none());
assert_eq!(
second_response.reason,
SelectionReason::NoAdmissibleWorkers("active_project_exclusion=1".to_string())
);
let diagnostics = second_response
.diagnostics
.expect("daemon response should surface selector diagnostics");
assert_eq!(diagnostics.active_project_exclusion_count, 1);
assert_eq!(diagnostics.workers.len(), 1);
assert!(
diagnostics.workers[0]
.reason_codes
.iter()
.any(|code| code == "active_project_exclusion")
);
}
#[tokio::test]
async fn test_handle_select_worker_routes_same_project_to_different_worker() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
let ctx = make_test_context(pool);
let first_request = SelectionRequest {
job_mode: false,
project: "shared-project".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("worker1")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(2001),
};
let first_response = handle_select_worker(&ctx, first_request, false, None)
.await
.unwrap();
assert_eq!(first_response.reason, SelectionReason::Success);
assert_eq!(
first_response.worker.as_ref().map(|w| w.id.as_str()),
Some("worker1")
);
let second_request = SelectionRequest {
job_mode: false,
project: "shared-project".to_string(),
command: Some("cargo build".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(2002),
};
let second_response = handle_select_worker(&ctx, second_request, false, None)
.await
.unwrap();
assert_eq!(second_response.reason, SelectionReason::Success);
assert_eq!(
second_response.worker.as_ref().map(|w| w.id.as_str()),
Some("worker2")
);
}
#[tokio::test]
async fn test_handle_select_worker_job_mode_burst_spreads_then_queues() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
let ctx = make_test_context(pool);
let shard = |pid: u32, job_mode: bool| SelectionRequest {
job_mode,
project: "burst-project".to_string(),
command: Some("./run_shard.sh".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(pid),
};
let s1 = handle_select_worker(&ctx, shard(101, true), false, None)
.await
.unwrap();
assert_eq!(s1.reason, SelectionReason::Success);
let s2 = handle_select_worker(&ctx, shard(102, true), false, None)
.await
.unwrap();
assert_eq!(s2.reason, SelectionReason::Success);
let w1 = s1.worker.as_ref().map(|w| w.id.as_str()).unwrap();
let w2 = s2.worker.as_ref().map(|w| w.id.as_str()).unwrap();
assert_ne!(w1, w2);
let s3 = handle_select_worker(&ctx, shard(103, true), false, None)
.await
.unwrap();
assert!(s3.worker.is_none());
assert_eq!(s3.reason, SelectionReason::AllWorkersBusy);
let compile = handle_select_worker(&ctx, shard(104, false), false, None)
.await
.unwrap();
assert!(matches!(
compile.reason,
SelectionReason::NoAdmissibleWorkers(_)
));
}
#[tokio::test]
async fn test_handle_select_worker_allows_different_projects_on_same_worker() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let first_request = SelectionRequest {
job_mode: false,
project: "project-a".to_string(),
command: Some("cargo check".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(3001),
};
let first_response = handle_select_worker(&ctx, first_request, false, None)
.await
.unwrap();
assert_eq!(first_response.reason, SelectionReason::Success);
let second_request = SelectionRequest {
job_mode: false,
project: "project-b".to_string(),
command: Some("cargo check".to_string()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(3002),
};
let second_response = handle_select_worker(&ctx, second_request, false, None)
.await
.unwrap();
assert_eq!(second_response.reason, SelectionReason::Success);
assert_eq!(
second_response.worker.as_ref().map(|w| w.id.as_str()),
Some("worker1")
);
}
#[tokio::test]
async fn test_handle_select_worker_preferred() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "test".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("worker2")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
let worker = response.worker.unwrap();
assert_eq!(worker.id.as_str(), "worker2");
let absent_request = SelectionRequest {
job_mode: false,
project: "test-absent".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![WorkerId::new("no-such-worker")],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
};
let refusal = handle_select_worker(&ctx, absent_request, false, None)
.await
.unwrap();
assert!(refusal.worker.is_none());
assert_eq!(refusal.reason, SelectionReason::NoMatchingWorkers);
}
#[test]
fn test_parse_request_reload_requires_post() {
let _guard = test_guard!();
let result = parse_request("POST /reload");
assert!(result.is_ok());
assert!(
matches!(result.unwrap(), ApiRequest::Reload),
"expected reload request"
);
let result = parse_request("GET /reload");
assert!(result.is_err());
}
#[test]
fn test_format_wait_time_seconds_only() {
let _guard = test_guard!();
assert_eq!(format_wait_time(0), "0s");
assert_eq!(format_wait_time(1), "1s");
assert_eq!(format_wait_time(30), "30s");
assert_eq!(format_wait_time(59), "59s");
}
#[test]
fn test_format_wait_time_minutes() {
let _guard = test_guard!();
assert_eq!(format_wait_time(60), "1m");
assert_eq!(format_wait_time(61), "1m 1s");
assert_eq!(format_wait_time(90), "1m 30s");
assert_eq!(format_wait_time(120), "2m");
assert_eq!(format_wait_time(3599), "59m 59s");
}
#[test]
fn test_format_wait_time_hours() {
let _guard = test_guard!();
assert_eq!(format_wait_time(3600), "1h");
assert_eq!(format_wait_time(3660), "1h 1m");
assert_eq!(format_wait_time(7200), "2h");
assert_eq!(format_wait_time(7260), "2h 1m");
assert_eq!(format_wait_time(9000), "2h 30m");
}
#[test]
fn test_daemon_status_info_serialization() {
let _guard = test_guard!();
let info = DaemonStatusInfo {
pid: 1234,
uptime_secs: 3600,
version: "1.0.0".to_string(),
socket_path: "/tmp/test.sock".to_string(),
started_at: "2025-01-01T00:00:00Z".to_string(),
workers_total: 4,
workers_healthy: 3,
slots_total: 32,
slots_available: 24,
};
let json = serde_json::to_string(&info).unwrap();
assert!(json.contains("\"pid\":1234"));
assert!(json.contains("\"uptime_secs\":3600"));
assert!(json.contains("\"workers_total\":4"));
}
#[test]
fn test_worker_status_info_serialization() {
let _guard = test_guard!();
let info = WorkerStatusInfo {
id: "worker1".to_string(),
host: "localhost".to_string(),
user: "user".to_string(),
status: "healthy".to_string(),
circuit_state: "closed".to_string(),
used_slots: 2,
total_slots: 8,
speed_score: 95.5,
last_error: None,
consecutive_failures: 0,
recovery_in_secs: None,
failure_history: vec![true, true, true],
pressure_state: "healthy".to_string(),
pressure_confidence: "high".to_string(),
pressure_reason_code: "pressure_healthy".to_string(),
pressure_policy_rule: "all_pressure_rules_within_threshold".to_string(),
pressure_disk_free_gb: Some(42.0),
pressure_disk_total_gb: Some(128.0),
pressure_disk_free_ratio: Some(0.328),
pressure_build_disk_free_gb: None,
pressure_build_disk_total_gb: None,
pressure_disk_io_util_pct: Some(18.0),
pressure_memory_pressure: Some(44.0),
pressure_telemetry_age_secs: Some(7),
pressure_telemetry_fresh: true,
bypass: None,
};
let json = serde_json::to_string(&info).unwrap();
assert!(json.contains("\"id\":\"worker1\""));
assert!(json.contains("\"used_slots\":2"));
assert!(json.contains("\"speed_score\":95.5"));
}
#[test]
fn test_health_response_serialization() {
let _guard = test_guard!();
let response = HealthResponse {
status: "healthy".to_string(),
version: "1.0.0".to_string(),
uptime_seconds: 3600,
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"status\":\"healthy\""));
assert!(json.contains("\"version\":\"1.0.0\""));
}
#[test]
fn test_ready_response_serialization_ready() {
let _guard = test_guard!();
let response = ReadyResponse {
status: "ready".to_string(),
workers_available: true,
reason: None,
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"status\":\"ready\""));
assert!(json.contains("\"workers_available\":true"));
assert!(!json.contains("reason"));
}
#[test]
fn test_ready_response_serialization_not_ready() {
let _guard = test_guard!();
let response = ReadyResponse {
status: "not_ready".to_string(),
workers_available: false,
reason: Some("no_workers_available".to_string()),
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"status\":\"not_ready\""));
assert!(json.contains("\"reason\":\"no_workers_available\""));
}
#[test]
fn test_active_build_serialization() {
let _guard = test_guard!();
let build = ActiveBuild {
id: 42,
project_id: "my-project".to_string(),
worker_id: "worker1".to_string(),
command: "cargo build".to_string(),
started_at: "2025-01-01T00:00:00Z".to_string(),
last_heartbeat_at: "2025-01-01T00:00:03Z".to_string(),
heartbeat_age_secs: 2,
last_progress_at: "2025-01-01T00:00:02Z".to_string(),
progress_age_secs: 3,
heartbeat_phase: "execute".to_string(),
heartbeat_detail: Some("Compiling crates".to_string()),
heartbeat_counter: 9,
heartbeat_percent: Some(42.0),
slots: 4,
detector_hook_alive: false,
detector_heartbeat_stale: true,
detector_progress_stale: true,
detector_confidence: 0.91,
detector_build_age_secs: 120,
detector_slots_owned: 4,
detector_last_evaluated_at: Some("2025-01-01T00:00:04Z".to_string()),
};
let json = serde_json::to_string(&build).unwrap();
assert!(json.contains("\"id\":42"));
assert!(json.contains("\"project_id\":\"my-project\""));
assert!(json.contains("\"command\":\"cargo build\""));
assert!(json.contains("\"heartbeat_phase\":\"execute\""));
assert!(json.contains("\"heartbeat_counter\":9"));
assert!(json.contains("\"detector_confidence\":0.91"));
}
#[test]
fn test_queued_build_serialization() {
let _guard = test_guard!();
let build = QueuedBuild {
id: (1_u64 << 63) + 1,
id_text: "9223372036854775809".to_string(),
project_id: "test".to_string(),
command: "cargo test".to_string(),
queued_at: "2025-01-01T00:00:00Z".to_string(),
position: 1,
slots_needed: 4,
estimated_start: Some("2025-01-01T00:01:00Z".to_string()),
wait_time: "1m 30s".to_string(),
};
let json = serde_json::to_string(&build).unwrap();
assert!(json.contains("\"id\":9223372036854775809"));
assert!(json.contains("\"id_text\":\"9223372036854775809\""));
assert!(json.contains("\"position\":1"));
assert!(json.contains("\"slots_needed\":4"));
assert!(json.contains("\"wait_time\":\"1m 30s\""));
}
#[test]
fn test_issue_serialization() {
let _guard = test_guard!();
let issue = Issue {
severity: "warning".to_string(),
summary: "Worker w1 is unreachable".to_string(),
remediation: Some("rch doctor".to_string()),
};
let json = serde_json::to_string(&issue).unwrap();
assert!(json.contains("\"severity\":\"warning\""));
assert!(json.contains("\"remediation\":\"rch doctor\""));
}
#[test]
fn test_cancel_build_response_serialization() {
let _guard = test_guard!();
let response = CancelBuildResponse {
status: "cancelled".to_string(),
build_id: 123,
worker_id: Some("worker1".to_string()),
project_id: Some("test".to_string()),
message: Some("Build cancelled".to_string()),
slots_released: 4,
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"build_id\":123"));
assert!(json.contains("\"slots_released\":4"));
}
#[test]
fn test_cancel_all_builds_response_serialization() {
let _guard = test_guard!();
let response = CancelAllBuildsResponse {
status: "ok".to_string(),
cancelled_count: 2,
cancelled: vec![CancelledBuildInfo {
build_id: 1,
worker_id: "w1".to_string(),
project_id: "p1".to_string(),
slots_released: 4,
}],
message: None,
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"cancelled_count\":2"));
}
#[test]
fn test_worker_state_response_serialization() {
let _guard = test_guard!();
let response = WorkerStateResponse {
status: "ok".to_string(),
worker_id: "worker1".to_string(),
action: "drain".to_string(),
new_status: Some("draining".to_string()),
reason: None,
message: Some("Worker draining started".to_string()),
active_slots: Some(4),
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"action\":\"drain\""));
assert!(json.contains("\"new_status\":\"draining\""));
}
#[test]
fn test_speedscore_view_serialization() {
let _guard = test_guard!();
let view = SpeedScoreView {
total: 85.5,
cpu_score: 90.0,
memory_score: 80.0,
disk_score: 85.0,
network_score: 88.0,
compilation_score: 82.0,
measured_at: "2025-01-01T00:00:00Z".to_string(),
version: 1,
};
let json = serde_json::to_string(&view).unwrap();
assert!(json.contains("\"total\":85.5"));
assert!(json.contains("\"cpu_score\":90"));
}
#[test]
fn test_speedscore_response_serialization() {
let _guard = test_guard!();
let response = SpeedScoreResponse {
worker_id: "worker1".to_string(),
speedscore: None,
message: Some("No speedscore available".to_string()),
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"worker_id\":\"worker1\""));
}
#[test]
fn test_pagination_info_serialization() {
let _guard = test_guard!();
let info = PaginationInfo {
total: 100,
offset: 10,
limit: 20,
has_more: true,
};
let json = serde_json::to_string(&info).unwrap();
assert!(json.contains("\"total\":100"));
assert!(json.contains("\"has_more\":true"));
}
#[test]
fn test_benchmark_trigger_response_serialization() {
let _guard = test_guard!();
let response = BenchmarkTriggerResponse {
status: "queued".to_string(),
worker_id: "worker1".to_string(),
request_id: "abc-123".to_string(),
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"status\":\"queued\""));
assert!(json.contains("\"request_id\":\"abc-123\""));
}
#[test]
fn test_parse_request_health() {
let _guard = test_guard!();
let req = parse_request("GET /health").unwrap();
assert!(matches!(req, ApiRequest::Health));
}
#[test]
fn test_parse_request_ready() {
let _guard = test_guard!();
let req = parse_request("GET /ready").unwrap();
assert!(matches!(req, ApiRequest::Ready));
}
#[test]
fn test_parse_request_metrics() {
let _guard = test_guard!();
let req = parse_request("GET /metrics").unwrap();
assert!(matches!(req, ApiRequest::Metrics));
}
#[test]
fn test_parse_request_budget() {
let _guard = test_guard!();
let req = parse_request("GET /budget").unwrap();
assert!(matches!(req, ApiRequest::Budget));
}
#[test]
fn test_parse_request_shutdown() {
let _guard = test_guard!();
let req = parse_request("POST /shutdown").unwrap();
assert!(matches!(req, ApiRequest::Shutdown));
}
#[test]
fn test_parse_request_restart_admission_routes() {
let _guard = test_guard!();
assert!(matches!(
parse_request("POST /restart-admission").unwrap(),
ApiRequest::RestartAdmission { close: true }
));
assert!(matches!(
parse_request("POST /restart-admission/release").unwrap(),
ApiRequest::RestartAdmission { close: false }
));
assert!(matches!(
parse_request("GET /restart-admission").unwrap(),
ApiRequest::RestartAdmissionStatus
));
}
#[test]
fn test_parse_request_cancel_build() {
let _guard = test_guard!();
let req = parse_request("POST /builds/123/cancel").unwrap();
match req {
ApiRequest::CancelBuild {
build_id, force, ..
} => {
assert_eq!(build_id, 123);
assert!(!force);
}
_ => assert!(false, "expected cancel build request"),
}
let req = parse_request("POST /builds/456/cancel?force=true").unwrap();
match req {
ApiRequest::CancelBuild {
build_id, force, ..
} => {
assert_eq!(build_id, 456);
assert!(force);
}
_ => assert!(false, "expected cancel build request with force"),
}
}
#[test]
fn test_parse_request_cancel_all_builds() {
let _guard = test_guard!();
let req = parse_request("POST /builds/cancel-all").unwrap();
match req {
ApiRequest::CancelAllBuilds { force } => {
assert!(!force);
}
_ => assert!(false, "expected cancel all builds request"),
}
let req = parse_request("POST /builds/cancel-all?force=true").unwrap();
match req {
ApiRequest::CancelAllBuilds { force } => {
assert!(force);
}
_ => assert!(false, "expected cancel all builds with force"),
}
}
#[test]
fn test_parse_request_worker_drain() {
let _guard = test_guard!();
let req = parse_request("POST /workers/css/drain").unwrap();
match req {
ApiRequest::WorkerDrain { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected worker drain request"),
}
}
#[test]
fn test_parse_request_worker_enable() {
let _guard = test_guard!();
let req = parse_request("POST /workers/css/enable").unwrap();
match req {
ApiRequest::WorkerEnable { worker_id } => {
assert_eq!(worker_id.as_str(), "css");
}
_ => assert!(false, "expected worker enable request"),
}
}
#[test]
fn test_parse_request_worker_disable() {
let _guard = test_guard!();
let req = parse_request("POST /workers/css/disable").unwrap();
match req {
ApiRequest::WorkerDisable {
worker_id,
reason,
drain_first,
} => {
assert_eq!(worker_id.as_str(), "css");
assert!(reason.is_none());
assert!(!drain_first);
}
_ => assert!(false, "expected worker disable request"),
}
let req = parse_request("POST /workers/css/disable?reason=maintenance&drain=true").unwrap();
match req {
ApiRequest::WorkerDisable {
worker_id,
reason,
drain_first,
} => {
assert_eq!(worker_id.as_str(), "css");
assert_eq!(reason, Some("maintenance".to_string()));
assert!(drain_first);
}
_ => assert!(false, "expected worker disable request with options"),
}
}
#[test]
fn test_parse_request_release_worker() {
let _guard = test_guard!();
let req = parse_request("POST /release-worker?worker=css&slots=4").unwrap();
match req {
ApiRequest::ReleaseWorker(req) => {
assert_eq!(req.worker_id.as_str(), "css");
assert_eq!(req.slots, 4);
}
_ => assert!(false, "expected release worker request"),
}
}
#[test]
fn test_parse_request_release_worker_with_build_id() {
let _guard = test_guard!();
let req = parse_request("POST /release-worker?worker=css&slots=4&build_id=123").unwrap();
match req {
ApiRequest::ReleaseWorker(req) => {
assert_eq!(req.worker_id.as_str(), "css");
assert_eq!(req.slots, 4);
assert_eq!(req.build_id, Some(123));
}
_ => assert!(false, "expected release worker request with build_id"),
}
}
#[test]
fn test_parse_request_release_worker_with_exit_code() {
let _guard = test_guard!();
let req = parse_request("POST /release-worker?worker=css&slots=4&exit_code=0").unwrap();
match req {
ApiRequest::ReleaseWorker(req) => {
assert_eq!(req.exit_code, Some(0));
}
_ => assert!(false, "expected release worker request with exit_code"),
}
}
#[test]
fn test_parse_request_release_worker_fault_flag() {
let _guard = test_guard!();
for (query, expected) in [
("&exit_code=101&worker_fault=1", true),
("&exit_code=101&worker_fault=true", true),
("&exit_code=101&worker_fault=0", false),
("&exit_code=101", false),
] {
let req =
parse_request(&format!("POST /release-worker?worker=css&slots=4{query}")).unwrap();
let ApiRequest::ReleaseWorker(req) = req else {
unreachable!("expected release worker request for {query}");
};
assert_eq!(req.worker_fault, expected, "{query}");
}
}
#[test]
fn test_parse_request_record_build() {
let _guard = test_guard!();
let req =
parse_request("POST /record-build?worker=css&project=myproject&is_test=true").unwrap();
match req {
ApiRequest::RecordBuild {
worker_id,
project,
is_test,
} => {
assert_eq!(worker_id.as_str(), "css");
assert_eq!(project, "myproject");
assert!(is_test);
}
_ => assert!(false, "expected record build request"),
}
}
#[test]
fn test_parse_request_ingest_telemetry() {
let _guard = test_guard!();
let req = parse_request("POST /telemetry/ingest?source=piggyback").unwrap();
match req {
ApiRequest::IngestTelemetry(source) => {
assert_eq!(source, TelemetrySource::Piggyback);
}
_ => assert!(false, "expected ingest telemetry request"),
}
let req = parse_request("POST /telemetry/ingest?source=ssh_poll").unwrap();
match req {
ApiRequest::IngestTelemetry(source) => {
assert_eq!(source, TelemetrySource::SshPoll);
}
_ => assert!(false, "expected ingest telemetry request"),
}
}
#[test]
fn test_parse_request_wait_for_worker() {
let _guard = test_guard!();
let req = parse_request("GET /select-worker?project=test&wait=true&wait_timeout_secs=45")
.unwrap();
match req {
ApiRequest::SelectWorker {
wait_for_worker,
wait_timeout_secs,
..
} => {
assert!(wait_for_worker);
assert_eq!(wait_timeout_secs, Some(45));
}
_ => assert!(false, "expected select worker request"),
}
let req = parse_request("GET /select-worker?project=test&wait=false").unwrap();
match req {
ApiRequest::SelectWorker {
wait_for_worker,
wait_timeout_secs,
..
} => {
assert!(!wait_for_worker);
assert_eq!(wait_timeout_secs, None);
}
_ => assert!(false, "expected select worker request"),
}
let req = parse_request("GET /select-worker?project=test&wait_timeout_secs=0").unwrap();
match req {
ApiRequest::SelectWorker {
wait_timeout_secs, ..
} => {
assert_eq!(wait_timeout_secs, None);
}
_ => assert!(false, "expected select worker request"),
}
}
#[tokio::test]
async fn test_handle_ready_with_available_workers() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_ready(&ctx).await;
assert_eq!(response.status, "ready");
assert!(response.workers_available);
assert!(response.reason.is_none());
}
#[tokio::test]
async fn test_handle_ready_no_workers() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_ready(&ctx).await;
assert_eq!(response.status, "not_ready");
assert!(!response.workers_available);
assert!(response.reason.is_some());
}
#[tokio::test]
async fn test_handle_ready_all_slots_used() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 4)).await;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
worker.reserve_slots(4).await;
let ctx = make_test_context(pool);
let response = handle_ready(&ctx).await;
assert_eq!(response.status, "not_ready");
assert!(!response.workers_available);
}
#[tokio::test]
async fn test_handle_ready_ignores_non_assignable_workers_with_free_slots() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("drained", 4)).await;
pool.add_worker(make_test_worker("disabled", 4)).await;
pool.add_worker(make_test_worker("unreachable", 4)).await;
pool.set_status(&WorkerId::new("drained"), WorkerStatus::Drained)
.await;
pool.set_status(&WorkerId::new("disabled"), WorkerStatus::Disabled)
.await;
pool.set_status(&WorkerId::new("unreachable"), WorkerStatus::Unreachable)
.await;
let ctx = make_test_context(pool);
let response = handle_ready(&ctx).await;
assert_eq!(response.status, "not_ready");
assert!(!response.workers_available);
assert_eq!(response.reason.as_deref(), Some("no_workers_available"));
}
#[tokio::test]
async fn test_handle_ready_ignores_open_circuit_workers_with_free_slots() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 4)).await;
let worker = pool
.get(&WorkerId::new("worker1"))
.await
.expect("worker should exist");
worker.open_circuit().await;
let ctx = make_test_context(pool);
let response = handle_ready(&ctx).await;
assert_eq!(response.status, "not_ready");
assert!(!response.workers_available);
assert_eq!(response.reason.as_deref(), Some("no_workers_available"));
}
#[tokio::test]
async fn test_handle_cancel_build_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_cancel_build(&ctx, 9999, false).await;
assert_eq!(response.status, "error");
assert_eq!(response.build_id, 9999);
assert!(response.message.unwrap().contains("not found"));
assert_eq!(response.slots_released, 0);
}
#[tokio::test]
async fn test_handle_cancel_all_builds_empty() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_cancel_all_builds(&ctx, false).await;
assert_eq!(response.status, "ok");
assert_eq!(response.cancelled_count, 0);
assert!(response.cancelled.is_empty());
}
#[test]
fn ownership_mismatch_names_the_owner_and_what_works() {
let owned = ownership_mismatch_message(7, Some(Some("rchw-abc")));
assert!(owned.contains("rchw-abc"), "{owned}");
assert!(owned.contains("rch jobs cancel rchw-abc"), "{owned}");
assert!(owned.contains("6h"), "{owned}");
let anonymous = ownership_mismatch_message(7, Some(None));
assert!(anonymous.contains("no wrapper identity"), "{anonymous}");
let done = ownership_mismatch_message(7, None);
assert!(done.contains("already completed"), "{done}");
for message in [owned, anonymous, done] {
assert!(message.starts_with("build ownership mismatch"), "{message}");
}
}
#[tokio::test]
async fn describe_caller_names_the_calling_process() {
let pid = std::process::id();
let described = describe_caller(pid as i32).await;
assert!(described.starts_with(&format!("pid={pid} ")), "{described}");
assert!(
described.contains("cmd=") && described.contains("parent="),
"{described}"
);
assert_eq!(describe_caller(i32::MAX).await, format!("pid={}", i32::MAX));
}
#[tokio::test]
async fn test_handle_worker_drain_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_worker_drain(&ctx, &WorkerId::new("nonexistent")).await;
assert_eq!(response.status, "error");
assert!(response.message.unwrap().contains("not found"));
}
#[tokio::test]
async fn test_handle_worker_drain_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_worker_drain(&ctx, &WorkerId::new("worker1")).await;
assert_eq!(response.status, "ok");
assert_eq!(response.worker_id, "worker1");
assert_eq!(response.action, "drain");
assert_eq!(response.new_status, Some("drained".to_string()));
}
#[tokio::test]
async fn test_handle_worker_enable_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_worker_enable(&ctx, &WorkerId::new("nonexistent")).await;
assert_eq!(response.status, "error");
assert!(response.message.unwrap().contains("not found"));
}
#[tokio::test]
async fn test_handle_worker_enable_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.set_status(&WorkerId::new("worker1"), WorkerStatus::Draining)
.await;
let ctx = make_test_context(pool);
let response = handle_worker_enable(&ctx, &WorkerId::new("worker1")).await;
assert_eq!(response.status, "ok");
assert_eq!(response.worker_id, "worker1");
assert_eq!(response.action, "enable");
assert_eq!(response.new_status, Some("healthy".to_string()));
}
#[tokio::test]
async fn test_handle_worker_disable_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response =
handle_worker_disable(&ctx, &WorkerId::new("nonexistent"), None, false).await;
assert_eq!(response.status, "error");
assert!(response.message.unwrap().contains("not found"));
}
#[tokio::test]
async fn test_handle_worker_disable_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_worker_disable(
&ctx,
&WorkerId::new("worker1"),
Some("maintenance".to_string()),
false,
)
.await;
assert_eq!(response.status, "ok");
assert_eq!(response.worker_id, "worker1");
assert_eq!(response.action, "disable");
assert_eq!(response.new_status, Some("disabled".to_string()));
}
#[tokio::test]
async fn test_handle_worker_disable_with_drain() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let worker = ctx.pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(1).await);
let response = handle_worker_disable(&ctx, &WorkerId::new("worker1"), None, true).await;
assert_eq!(response.status, "ok");
assert_eq!(response.action, "disable");
assert_eq!(response.new_status, Some("draining".to_string()));
}
#[tokio::test]
async fn test_worker_disable_persists_durable_record_and_enable_removes_it() {
use rch_common::bypass_record::AdminDisableStore;
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let mut ctx = make_test_context(pool);
let dir = tempfile::tempdir().unwrap();
let store_path = dir.path().join("admin_disables.json");
ctx.admin_disable_store = Some(Arc::new(tokio::sync::Mutex::new(
AdminDisableStore::with_path(&store_path),
)));
let response = handle_worker_disable(
&ctx,
&WorkerId::new("worker1"),
Some("cpu-capability-fault (SIGILL)".to_string()),
false,
)
.await;
assert_eq!(response.status, "ok");
let reloaded = AdminDisableStore::load(&store_path);
assert_eq!(
reloaded
.get("worker1")
.expect("record persisted")
.reason
.as_deref(),
Some("cpu-capability-fault (SIGILL)"),
);
let response = handle_worker_enable(&ctx, &WorkerId::new("worker1")).await;
assert_eq!(response.status, "ok");
assert!(
AdminDisableStore::load(&store_path)
.get("worker1")
.is_none()
);
}
#[tokio::test]
async fn test_worker_disable_with_drain_also_persists_durable_record() {
use rch_common::bypass_record::AdminDisableStore;
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let mut ctx = make_test_context(pool);
let dir = tempfile::tempdir().unwrap();
let store_path = dir.path().join("admin_disables.json");
ctx.admin_disable_store = Some(Arc::new(tokio::sync::Mutex::new(
AdminDisableStore::with_path(&store_path),
)));
let worker = ctx.pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(1).await);
let response = handle_worker_disable(&ctx, &WorkerId::new("worker1"), None, true).await;
assert_eq!(response.new_status, Some("draining".to_string()));
assert!(
AdminDisableStore::load(&store_path)
.get("worker1")
.is_some()
);
}
#[test]
fn test_worker_capabilities_info_serialization() {
let _guard = test_guard!();
let info = WorkerCapabilitiesInfo {
id: "worker1".to_string(),
host: "localhost".to_string(),
user: "test".to_string(),
capabilities: WorkerCapabilities::default(),
pressure_assessment: crate::disk_pressure::PressureAssessment::default(),
refresh: None,
};
let json = serde_json::to_string(&info).unwrap();
assert!(json.contains("\"id\":\"worker1\""));
}
#[test]
fn test_worker_capabilities_response_serialization() {
let _guard = test_guard!();
let response = WorkerCapabilitiesResponse { workers: vec![] };
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"workers\":[]"));
}
#[test]
fn test_self_test_status_response_serialization() {
let _guard = test_guard!();
let response = SelfTestStatusResponse {
enabled: true,
schedule: Some("daily".to_string()),
interval: Some("24h".to_string()),
last_run: None,
next_run: Some("2025-01-02T00:00:00Z".to_string()),
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"enabled\":true"));
assert!(json.contains("\"schedule\":\"daily\""));
}
#[test]
fn test_speedscore_list_response_serialization() {
let _guard = test_guard!();
let response = SpeedScoreListResponse {
workers: vec![SpeedScoreWorker {
worker_id: "worker1".to_string(),
speedscore: None,
status: WorkerStatus::Healthy,
}],
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"worker_id\":\"worker1\""));
}
#[test]
fn test_speedscore_history_response_serialization() {
let _guard = test_guard!();
let response = SpeedScoreHistoryResponse {
worker_id: "worker1".to_string(),
history: vec![],
pagination: PaginationInfo {
total: 0,
offset: 0,
limit: 20,
has_more: false,
},
};
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"worker_id\":\"worker1\""));
assert!(json.contains("\"has_more\":false"));
}
#[test]
fn test_selection_response_json_includes_protocol_version() {
let _guard = test_guard!();
let response = SelectionResponse {
worker: None,
reason: SelectionReason::AllWorkersBusy,
build_id: None,
diagnostics: None,
};
let json = selection_response_json(&response).expect("selection response serializes");
let value: serde_json::Value = serde_json::from_str(&json).expect("valid json");
assert_eq!(
value["selection_protocol_version"],
rch_common::SELECTION_RESPONSE_PROTOCOL_VERSION
);
assert_eq!(value["reason"], "all_workers_busy");
}
#[test]
fn test_api_response_ok_serialization() {
let _guard = test_guard!();
let response: ApiResponse<HealthResponse> = ApiResponse::Ok(HealthResponse {
status: "healthy".to_string(),
version: "1.0.0".to_string(),
uptime_seconds: 100,
});
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"status\":\"healthy\""));
}
#[test]
fn test_api_response_error_serialization() {
let _guard = test_guard!();
let response: ApiResponse<HealthResponse> =
ApiResponse::Error(ApiError::new(ErrorCode::ConfigNotFound, "Not found"));
let json = serde_json::to_string(&response).unwrap();
assert!(json.contains("\"details\":\"Not found\""));
}
#[tokio::test]
async fn test_handle_telemetry_poll_worker_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_telemetry_poll(&ctx, &WorkerId::new("nonexistent")).await;
assert!(response.error.is_some());
assert!(response.error.unwrap().contains("not found"));
}
#[tokio::test]
async fn test_handle_speedscore_worker_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_speedscore(&ctx, &WorkerId::new("nonexistent")).await;
match response {
ApiResponse::Error(e) => {
assert!(
e.details
.as_deref()
.unwrap_or_default()
.contains("not found")
);
}
_ => assert!(false, "expected error response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_speedscore(&ctx, &WorkerId::new("worker1")).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.worker_id, "worker1");
assert!(r.speedscore.is_none());
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_list() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 4)).await;
let ctx = make_test_context(pool);
let response = handle_speedscore_list(&ctx).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.workers.len(), 2);
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_benchmark_trigger_worker_not_found() {
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_benchmark_trigger(&ctx, &WorkerId::new("nonexistent")).await;
match response {
ApiResponse::Error(e) => {
assert!(
e.details
.as_deref()
.unwrap_or_default()
.contains("not found")
);
}
_ => assert!(false, "expected error response"),
}
}
#[tokio::test]
async fn test_handle_benchmark_trigger_success() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_benchmark_trigger(&ctx, &WorkerId::new("worker1")).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.status, "queued");
assert_eq!(r.worker_id, "worker1");
assert!(!r.request_id.is_empty());
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_benchmark_trigger_rate_limited() {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let first = handle_benchmark_trigger(&ctx, &WorkerId::new("worker1")).await;
assert!(matches!(first, ApiResponse::Ok(_)));
let second = handle_benchmark_trigger(&ctx, &WorkerId::new("worker1")).await;
match second {
ApiResponse::Error(err) => {
assert!(
err.details
.as_deref()
.unwrap_or_default()
.contains("rate limited")
);
assert!(err.retry_after_secs.is_some());
}
_ => assert!(false, "expected rate-limited error response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_history_worker_not_found() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response =
handle_speedscore_history(&ctx, &WorkerId::new("nonexistent"), 7, 20, 0).await;
match response {
ApiResponse::Error(e) => {
assert!(
e.details
.as_deref()
.unwrap_or_default()
.contains("not found")
);
}
_ => assert!(false, "expected error response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_history_success_empty() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_speedscore_history(&ctx, &WorkerId::new("worker1"), 7, 20, 0).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.worker_id, "worker1");
assert!(r.history.is_empty()); assert!(!r.pagination.has_more);
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_history_clamps_days() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_speedscore_history(&ctx, &WorkerId::new("worker1"), 0, 20, 0).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.worker_id, "worker1");
}
_ => assert!(false, "expected ok response"),
}
let response =
handle_speedscore_history(&ctx, &WorkerId::new("worker1"), 1000, 20, 0).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.worker_id, "worker1");
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_speedscore_history_clamps_limit() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let response = handle_speedscore_history(&ctx, &WorkerId::new("worker1"), 7, 0, 0).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.pagination.limit, 1);
}
_ => assert!(false, "expected ok response"),
}
let response = handle_speedscore_history(&ctx, &WorkerId::new("worker1"), 7, 5000, 0).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.pagination.limit, 1000);
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_workers_capabilities_empty_pool() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_workers_capabilities(&ctx, false).await;
match response {
ApiResponse::Ok(r) => {
assert!(r.workers.is_empty());
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn test_handle_workers_capabilities_with_workers() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 4)).await;
let ctx = make_test_context(pool);
let response = handle_workers_capabilities(&ctx, false).await;
match response {
ApiResponse::Ok(r) => {
assert_eq!(r.workers.len(), 2);
let w1 = r.workers.iter().find(|w| w.id == "worker1").unwrap();
assert_eq!(w1.host, "localhost");
assert_eq!(w1.user, "user");
}
_ => assert!(false, "expected ok response"),
}
}
#[tokio::test]
async fn disk_full_release_quarantines_only_the_completed_build_owner() {
let _guard = test_guard!();
for (exit, flag, request_worker, wrapper, quarantine) in [
(101, true, "worker1", "owner", true),
(0, true, "worker1", "owner", false),
(101, false, "worker1", "owner", false),
(101, true, "worker2", "owner", false),
(101, true, "worker1", "wrong-owner", false),
] {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 8)).await;
let mut ctx = make_test_context(pool.clone());
let directory = tempfile::tempdir().unwrap();
let store_path = directory.path().join("bypasses.json");
let store = Arc::new(tokio::sync::Mutex::new(BypassRecordStore::with_path(
&store_path,
)));
ctx.bypass_store = Some(store.clone());
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
let build = ctx
.history
.try_start_active_build_with_waiter(
"disk-project".into(),
"worker1".into(),
"cargo build".into(),
12345,
Some("owner".into()),
2,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
Some(worker.endpoint_snapshot().await),
)
.unwrap()
.unwrap();
assert!(worker.reserve_slots(2).await);
ctx.history
.record_build_heartbeat(BuildHeartbeatRequest {
build_id: build.id,
worker_id: WorkerId::new("worker1"),
hook_pid: None,
local_wrapper_id: Some("owner".into()),
remote_pgid_file: None,
phase: rch_common::BuildHeartbeatPhase::Execute,
detail: None,
progress_counter: Some(1),
progress_percent: None,
})
.unwrap();
let request = parse_request(&format!(
"POST /release-worker?worker={request_worker}&build_id={}&local_wrapper_id={wrapper}&exit_code={exit}&worker_disk_full={flag}&worker_disk_roots=%5B%22%2Fbuild-volume%2Frch%22%5D", build.id
)).unwrap();
let ApiRequest::ReleaseWorker(request) = request else {
panic!("release request expected")
};
assert_eq!(request.worker_disk_full, flag);
assert_eq!(request.worker_disk_roots, ["/build-volume/rch"]);
let result = handle_release_worker(&ctx, request.clone()).await;
let owned = request_worker == "worker1" && wrapper == "owner";
assert_eq!(result.is_ok(), owned, "{request:?}: {result:?}");
assert_eq!(worker.used_slots(), if owned { 0 } else { 2 });
assert_eq!(
worker.lifecycle().await.admin,
crate::workers::AdminIntent::Active
);
assert_eq!(store.lock().await.contains("worker1"), quarantine);
assert!(!store.lock().await.contains("worker2"));
if quarantine {
let persisted = BypassRecordStore::load(&store_path);
let before = persisted.get("worker1").unwrap().clone();
assert_eq!(
before.failure_class,
rch_common::BypassFailureClass::DiskInodePressure
);
assert_eq!(before.disk_roots, ["/build-volume/rch"]);
assert_eq!(
worker.eligibility().await,
crate::workers::EligibilityState::TemporaryBypass
);
worker.apply_health_status(WorkerStatus::Healthy).await;
assert_eq!(worker.status().await, WorkerStatus::Unreachable);
pool.get(&WorkerId::new("worker2"))
.await
.unwrap()
.disable(Some("test maintenance".into()))
.await;
assert!(
!handle_ready(&ctx).await.workers_available,
"quarantined free slots must not advertise ready capacity"
);
handle_release_worker(&ctx, request).await.unwrap();
assert_eq!(
store.lock().await.get("worker1"),
Some(&before),
"duplicate completion must not advance failure backoff"
);
assert_eq!(
ctx.worker_selector
.cache_warmth("worker1", "disk-project", crate::selection::CacheUse::Build)
.await,
0.0
);
} else {
assert!(worker.lifecycle().await.is_schedulable());
if owned {
assert_eq!(
ctx.worker_selector
.cache_warmth(
"worker1",
"disk-project",
crate::selection::CacheUse::Build
)
.await,
1.0,
"successful builds and ordinary project failures retain useful cache warmth"
);
}
}
}
}
fn disk_fault_release_request(build_id: u64) -> ReleaseRequest {
ReleaseRequest {
local_wrapper_id: Some("disk-owner".into()),
worker_id: WorkerId::new("worker1"),
slots: 2,
build_id: Some(build_id),
exit_code: Some(101),
duration_ms: Some(50),
bytes_transferred: None,
timing: None,
worker_fault: true,
worker_disk_full: true,
worker_disk_roots: vec!["/build-volume/rch".into()],
}
}
#[tokio::test]
async fn disk_fault_restart_replays_either_commit_boundary_without_duplicate_quarantine() {
let _guard = test_guard!();
for bypass_was_committed in [false, true] {
let temporary = tempfile::tempdir().unwrap();
let history_path = temporary.path().join("history.jsonl");
let store_path = temporary.path().join("bypasses.json");
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let mut ctx = make_test_context(pool);
ctx.history = Arc::new(BuildHistory::new(10).with_persistence(history_path.clone()));
ctx.bypass_store = Some(Arc::new(tokio::sync::Mutex::new(
BypassRecordStore::with_path(&store_path),
)));
let worker = ctx.pool.get(&WorkerId::new("worker1")).await.unwrap();
let build = ctx
.history
.try_start_active_build_with_waiter(
"disk-project".into(),
"worker1".into(),
"cargo build".into(),
12345,
Some("disk-owner".into()),
2,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
Some(worker.endpoint_snapshot().await),
)
.unwrap()
.unwrap();
ctx.history
.complete_durable_with_disk_fault(
build.id,
"worker1",
Some("disk-owner"),
crate::history::BuildCompletion {
exit_code: 101,
duration_ms: Some(50),
bytes_transferred: None,
timing: None,
cancellation: None,
},
Some(vec!["/build-volume/rch".into()]),
)
.unwrap()
.unwrap();
let fault = ctx.history.pending_disk_fault(build.id).unwrap();
let before = if bypass_was_committed {
let worker = ctx.pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(
crate::bypass_recovery_service::record_worker_disk_bypass(
ctx.bypass_store.as_ref().unwrap(),
Some(&worker),
"worker1",
&fault.incident_id,
&fault.roots,
"disk exhausted",
fault.reported_unix_ms,
|| Err(std::io::Error::other("interrupted before acknowledgment")),
)
.await
.is_err()
);
Some(
BypassRecordStore::load(&store_path)
.get("worker1")
.unwrap()
.clone(),
)
} else {
None
};
drop(ctx);
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(3).await);
let mut restored = make_test_context(pool);
restored.history = Arc::new(BuildHistory::load_from_file(&history_path, 10).unwrap());
let store = Arc::new(tokio::sync::Mutex::new(BypassRecordStore::load(
&store_path,
)));
restored.bypass_store = Some(store.clone());
assert!(restored.history.active_builds().is_empty());
replay_pending_disk_faults(&restored).await.unwrap();
assert_eq!(worker.used_slots(), 3);
assert!(!worker.lifecycle().await.is_schedulable());
let after = store.lock().await.get("worker1").unwrap().clone();
assert_eq!(after.disk_roots, ["/build-volume/rch"]);
if let Some(before) = before {
assert_eq!(
after, before,
"replay must not advance failure/backoff or replace evidence"
);
}
assert!(!restored.history.has_pending_disk_fault("worker1"));
assert!(
BuildHistory::load_from_file(&history_path, 10)
.unwrap()
.pending_disk_faults()
.is_empty()
);
let mut duplicate = disk_fault_release_request(build.id);
duplicate.worker_disk_roots = vec!["../invalid-new-retry-root".into()];
handle_release_worker(&restored, duplicate.clone())
.await
.unwrap();
assert_eq!(store.lock().await.get("worker1"), Some(&after));
store.lock().await.remove("worker1").unwrap();
worker.recover_to_canary().await.unwrap();
worker.promote_from_canary().await.unwrap();
restored.history = Arc::new(BuildHistory::load_from_file(&history_path, 10).unwrap());
replay_pending_disk_faults(&restored).await.unwrap();
handle_release_worker(&restored, duplicate).await.unwrap();
assert!(!store.lock().await.contains("worker1"));
assert!(worker.lifecycle().await.is_schedulable());
assert_eq!(worker.used_slots(), 3);
}
}
#[tokio::test]
async fn disk_fault_failed_bypass_write_retains_intent_and_retry_releases_no_other_slots() {
let _guard = test_guard!();
let temporary = tempfile::tempdir().unwrap();
let history_path = temporary.path().join("history.jsonl");
let store_path = temporary.path().join("bypasses.json");
std::fs::create_dir(&store_path).unwrap();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(5).await);
let mut ctx = make_test_context(pool);
ctx.history = Arc::new(BuildHistory::new(10).with_persistence(history_path.clone()));
let store = Arc::new(tokio::sync::Mutex::new(BypassRecordStore::with_path(
&store_path,
)));
ctx.bypass_store = Some(store.clone());
let build = ctx
.history
.try_start_active_build_with_waiter(
"disk-project".into(),
"worker1".into(),
"cargo build".into(),
12345,
Some("disk-owner".into()),
2,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
Some(worker.endpoint_snapshot().await),
)
.unwrap()
.unwrap();
let mut request = disk_fault_release_request(build.id);
request.worker_disk_roots = vec!["../bad".into()];
assert!(handle_release_worker(&ctx, request).await.is_err());
assert!(ctx.history.active_build(build.id).is_some());
assert_eq!(worker.used_slots(), 5);
let request = disk_fault_release_request(build.id);
assert!(handle_release_worker(&ctx, request.clone()).await.is_err());
assert_eq!(worker.used_slots(), 3);
assert!(ctx.history.active_build(build.id).is_none());
assert!(ctx.history.has_pending_disk_fault("worker1"));
assert!(
BuildHistory::load_from_file(&history_path, 10)
.unwrap()
.has_pending_disk_fault("worker1")
);
assert!(!worker.lifecycle().await.is_schedulable());
let recorded = store.lock().await.get("worker1").unwrap().clone();
std::fs::rename(
&store_path,
temporary.path().join("preserved-rename-obstacle"),
)
.unwrap();
handle_release_worker(&ctx, request).await.unwrap();
assert_eq!(worker.used_slots(), 3);
assert!(!ctx.history.has_pending_disk_fault("worker1"));
assert_eq!(
BypassRecordStore::load(&store_path).get("worker1"),
Some(&recorded)
);
}
#[tokio::test]
async fn late_disk_faults_archive_exact_incidents_without_quarantining_replacements() {
for change in [
"retarget",
"aba",
"restart",
"legacy",
"retarget_then_remove",
] {
for durable_store in [false, true] {
let temporary = tempfile::tempdir().unwrap();
let history_path = temporary.path().join("history.jsonl");
let store_path = temporary.path().join("bypasses.json");
let original = make_test_worker("worker1", 8);
let pool = WorkerPool::new();
pool.add_worker(original.clone()).await;
let mut worker = pool.get(&original.id).await.unwrap();
let mut ctx = make_test_context(pool);
ctx.history =
Arc::new(BuildHistory::new(100).with_persistence(history_path.clone()));
if durable_store {
ctx.bypass_store = Some(Arc::new(tokio::sync::Mutex::new(
BypassRecordStore::with_path(&store_path),
)));
}
let build = ctx
.history
.try_start_active_build_with_waiter(
"old-volume".into(),
original.id.to_string(),
"cargo build".into(),
0,
Some("disk-owner".into()),
2,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
if change == "legacy" {
None
} else {
Some(worker.endpoint_snapshot().await)
},
)
.unwrap()
.unwrap();
let other = ctx.history.start_active_build(
"other-build".into(),
original.id.to_string(),
"cargo check".into(),
0,
1,
rch_common::BuildLocation::Remote,
);
assert!(worker.reserve_slots(3).await);
let mut replacement = original.clone();
replacement.host = "replacement.example".into();
if change == "restart" {
ctx.history
.complete_durable_with_disk_fault(
build.id,
"worker1",
Some("disk-owner"),
crate::history::BuildCompletion {
exit_code: 101,
duration_ms: None,
bytes_transferred: None,
timing: None,
cancellation: None,
},
Some(vec!["/build-volume/rch".into()]),
)
.unwrap()
.unwrap();
ctx.history =
Arc::new(BuildHistory::load_from_file(&history_path, 100).unwrap());
ctx.pool = WorkerPool::new();
ctx.pool.add_worker(replacement).await;
worker = ctx.pool.get(&original.id).await.unwrap();
for active in ctx.history.active_builds() {
ctx.pool
.restore_recovered_slots(&original.id, active.slots)
.await
.unwrap();
}
replay_pending_disk_faults(&ctx).await.unwrap();
} else {
ctx.pool.add_worker(replacement).await;
if change == "aba" {
ctx.pool.add_worker(original.clone()).await;
} else if change == "retarget_then_remove" {
assert!(ctx.pool.remove_worker(&original.id).await);
ctx.pool.add_worker(original.clone()).await;
worker = ctx.pool.get(&original.id).await.unwrap();
for active in ctx.history.active_builds() {
ctx.pool
.restore_recovered_slots(&original.id, active.slots)
.await
.unwrap();
}
}
worker.apply_health_status(WorkerStatus::Healthy).await;
handle_release_worker(&ctx, disk_fault_release_request(build.id))
.await
.unwrap();
}
assert!(
worker.lifecycle().await.is_schedulable(),
"{change}/{durable_store}"
);
assert_eq!(worker.used_slots(), 1);
assert!(ctx.history.active_build(other.id).is_some());
assert!(ctx.history.active_build(build.id).is_none());
assert!(ctx.history.pending_disk_fault(build.id).is_none());
let archived = ctx.history.unapplied_disk_fault(build.id).unwrap();
assert_eq!(archived.roots, ["/build-volume/rch"]);
assert_eq!(archived.worker_endpoint.is_none(), change == "legacy");
if let Some(endpoint) = archived.worker_endpoint.as_ref() {
assert!(endpoint.matches_config(&original));
}
assert!(!BypassRecordStore::load(&store_path).contains("worker1"));
ctx.history = Arc::new(BuildHistory::load_from_file(&history_path, 100).unwrap());
ctx.pool = WorkerPool::new();
ctx.pool.add_worker(original.clone()).await;
worker = ctx.pool.get(&original.id).await.unwrap();
for active in ctx.history.active_builds() {
ctx.pool
.restore_recovered_slots(&original.id, active.slots)
.await
.unwrap();
}
replay_pending_disk_faults(&ctx).await.unwrap();
let mut duplicate = disk_fault_release_request(build.id);
duplicate.worker_disk_roots = vec!["../injected-retry-root".into()];
handle_release_worker(&ctx, duplicate).await.unwrap();
assert!(worker.lifecycle().await.is_schedulable());
assert_eq!(worker.used_slots(), 1);
assert_eq!(ctx.history.unapplied_disk_fault(build.id), Some(archived));
assert!(!BypassRecordStore::load(&store_path).contains("worker1"));
}
}
}
#[tokio::test]
async fn test_handle_release_worker_basic() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
let reserved = worker.reserve_slots(4).await;
assert!(reserved);
let ctx = make_test_context(pool.clone());
let request = ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: None,
exit_code: None,
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
let result = handle_release_worker(&ctx, request).await;
assert!(result.is_err());
assert_eq!(worker.available_slots().await, 4);
}
#[tokio::test]
async fn test_handle_release_worker_with_build_completion() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
let build = ctx
.history
.try_start_active_build_with_waiter(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
None,
4,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
Some(worker.endpoint_snapshot().await),
)
.unwrap()
.unwrap();
let build_id = build.id;
worker.reserve_slots(4).await;
let request = ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: Some(5000),
bytes_transferred: Some(1024 * 1024),
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
let result = handle_release_worker(&ctx, request).await;
assert!(result.is_ok());
let recent = ctx.history.recent(10);
let completed = recent.iter().find(|b| b.id == build_id);
assert!(completed.is_some());
assert_eq!(completed.unwrap().exit_code, 0);
let stats = worker.circuit_stats().await;
assert_eq!(stats.error_rate(), 0.0);
assert_eq!(stats.recent_results(), &[true]);
assert_eq!(
ctx.worker_selector
.get_fallback_worker("test-project")
.await,
Some("worker1".to_string())
);
}
#[tokio::test]
async fn completed_build_feedback_is_bound_but_slot_release_survives_retarget_and_restart() {
for change in ["unchanged", "retarget", "aba", "restart", "legacy"] {
let directory = tempfile::tempdir().unwrap();
let history_path = directory.path().join("history.jsonl");
let original = make_test_worker("bound-completion", 8);
let pool = WorkerPool::new();
pool.add_worker(original.clone()).await;
let mut worker = pool.get(&original.id).await.unwrap();
let mut ctx = make_test_context(pool.clone());
ctx.history = Arc::new(BuildHistory::new(100).with_persistence(history_path.clone()));
let other = ctx.history.start_active_build(
"other-build".into(),
original.id.to_string(),
"cargo check".into(),
0,
1,
rch_common::BuildLocation::Remote,
);
assert!(worker.reserve_slots(1).await);
let build_id = if change == "legacy" {
assert!(worker.reserve_slots(2).await);
ctx.history
.start_active_build(
"finished-build".into(),
original.id.to_string(),
"cargo build".into(),
0,
2,
rch_common::BuildLocation::Remote,
)
.id
} else {
let response = handle_select_worker(
&ctx,
SelectionRequest {
job_mode: false,
project: "finished-build".into(),
command: Some("cargo build".into()),
command_priority: CommandPriority::Normal,
estimated_cores: 2,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::default(),
classification_duration_us: None,
hook_pid: None,
required_tools: Vec::new(),
},
false,
None,
)
.await
.unwrap();
let selected = response.worker.unwrap();
let id = response.build_id.unwrap();
let active = ctx.history.active_build(id).unwrap();
let endpoint = active.worker_endpoint.as_ref().unwrap();
assert_eq!(endpoint.config.id, selected.id);
assert_eq!(endpoint.config.host, selected.host);
assert_eq!(endpoint.config.user, selected.user);
assert_eq!(endpoint.config.identity_file, selected.identity_file);
assert_eq!(
rch_common::declared_os(&endpoint.config.tags),
selected.declared_os
);
id
};
assert_eq!(worker.used_slots(), 3);
if matches!(change, "retarget" | "aba") {
let mut replacement = original.clone();
replacement.host = "replacement.host".into();
pool.add_worker(replacement).await;
if change == "aba" {
pool.add_worker(original.clone()).await;
}
} else if change == "restart" {
ctx.history = Arc::new(BuildHistory::load_from_file(&history_path, 100).unwrap());
ctx.pool = WorkerPool::new();
ctx.pool.add_worker(original.clone()).await;
worker = ctx.pool.get(&original.id).await.unwrap();
for active in ctx.history.active_builds() {
worker.restore_slots(active.slots).unwrap();
}
}
worker
.record_failure(Some("current endpoint evidence".into()))
.await;
let release = ReleaseRequest {
local_wrapper_id: None,
worker_id: original.id.clone(),
slots: 999,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
handle_release_worker(&ctx, release.clone()).await.unwrap();
assert_eq!(
worker.used_slots(),
1,
"{change}: release follows exact ownership"
);
let stats = worker.circuit_stats().await;
assert_eq!(
stats.consecutive_failures(),
u32::from(change != "unchanged"),
"{change}"
);
assert_eq!(
ctx.worker_selector
.get_fallback_worker("finished-build")
.await,
(change == "unchanged").then(|| original.id.to_string()),
"{change}: old ownership cannot pin the current worker"
);
handle_release_worker(&ctx, release).await.unwrap();
assert_eq!(
worker.used_slots(),
1,
"{change}: duplicate completion released another build"
);
assert!(ctx.history.active_build(other.id).is_some());
assert!(ctx.history.active_build(build_id).is_none());
assert_eq!(ctx.history.recent(10).len(), 1);
}
}
#[tokio::test]
async fn test_handle_release_worker_uses_active_build_slots() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
4,
rch_common::BuildLocation::Remote,
);
let build_id = build.id;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(4).await);
let request = ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 0,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: Some(5000),
bytes_transferred: Some(1024 * 1024),
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
handle_release_worker(&ctx, request).await.unwrap();
assert_eq!(worker.available_slots().await, 8);
assert!(ctx.history.active_build(build_id).is_none());
}
#[tokio::test]
async fn test_handle_release_worker_build_id_is_idempotent() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
4,
rch_common::BuildLocation::Remote,
);
let build_id = build.id;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(4).await);
let release = || ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(1),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
handle_release_worker(&ctx, release()).await.unwrap();
assert_eq!(worker.available_slots().await, 8);
assert!(worker.reserve_slots(2).await);
handle_release_worker(&ctx, release()).await.unwrap();
assert_eq!(worker.available_slots().await, 6);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_handle_release_worker_concurrent_duplicates_claim_build_once() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
4,
rch_common::BuildLocation::Remote,
);
let build_id = build.id;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
assert!(worker.reserve_slots(4).await);
let barrier = Arc::new(tokio::sync::Barrier::new(3));
let spawn_release = |ctx: DaemonContext, barrier: Arc<tokio::sync::Barrier>| {
tokio::spawn(async move {
barrier.wait().await;
handle_release_worker(
&ctx,
ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(1),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
},
)
.await
})
};
let first = spawn_release(ctx.clone(), barrier.clone());
let second = spawn_release(ctx.clone(), barrier.clone());
barrier.wait().await;
first.await.unwrap().unwrap();
second.await.unwrap().unwrap();
assert_eq!(worker.available_slots().await, 8);
assert!(ctx.history.active_build(build_id).is_none());
assert_eq!(
ctx.history
.recent(10)
.iter()
.filter(|record| record.id == build_id)
.count(),
1
);
assert!(worker.reserve_slots(2).await);
handle_release_worker(
&ctx,
ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(1),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
},
)
.await
.unwrap();
assert_eq!(worker.available_slots().await, 6);
}
#[tokio::test]
async fn test_release_after_remote_failure_warms_cache_without_health_or_pin() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
let endpoint = worker.endpoint_snapshot().await;
let start = |project: &str| {
let build = ctx
.history
.try_start_active_build_with_waiter(
project.to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
None,
2,
rch_common::BuildLocation::Remote,
None,
crate::disk_pressure::DiskHeadroomAdmission::default(),
Some(endpoint.clone()),
)
.unwrap()
.unwrap();
build.id
};
let release = |build_id| ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 2,
build_id: Some(build_id),
exit_code: Some(101),
duration_ms: Some(5000),
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
let ran = start("ran-project");
assert!(worker.reserve_slots(2).await);
ctx.history
.record_build_heartbeat(rch_common::BuildHeartbeatRequest {
build_id: ran,
worker_id: WorkerId::new("worker1"),
hook_pid: None,
local_wrapper_id: None,
remote_pgid_file: None,
phase: rch_common::BuildHeartbeatPhase::Execute,
detail: None,
progress_counter: Some(1),
progress_percent: None,
})
.expect("heartbeat accepted");
handle_release_worker(&ctx, release(ran)).await.unwrap();
assert_eq!(
ctx.worker_selector
.cache_warmth("worker1", "ran-project", crate::selection::CacheUse::Build)
.await,
1.0
);
assert_eq!(
ctx.worker_selector.get_pinned_worker("ran-project").await,
None
);
assert!(worker.circuit_stats().await.recent_results().is_empty());
let unsynced = start("unsynced-project");
assert!(worker.reserve_slots(2).await);
handle_release_worker(&ctx, release(unsynced))
.await
.unwrap();
assert_eq!(
ctx.worker_selector
.cache_warmth(
"worker1",
"unsynced-project",
crate::selection::CacheUse::Build
)
.await,
0.0
);
assert_eq!(worker.available_slots().await, 8);
let broken = start("broken-worker-project");
assert!(worker.reserve_slots(2).await);
ctx.history
.record_build_heartbeat(rch_common::BuildHeartbeatRequest {
build_id: broken,
worker_id: WorkerId::new("worker1"),
hook_pid: None,
local_wrapper_id: None,
remote_pgid_file: None,
phase: rch_common::BuildHeartbeatPhase::Execute,
detail: None,
progress_counter: Some(1),
progress_percent: None,
})
.expect("heartbeat accepted");
let mut faulted = release(broken);
faulted.worker_fault = true;
handle_release_worker(&ctx, faulted).await.unwrap();
assert_eq!(
ctx.worker_selector
.cache_warmth(
"worker1",
"broken-worker-project",
crate::selection::CacheUse::Build
)
.await,
0.0
);
assert_eq!(worker.available_slots().await, 8);
}
#[tokio::test]
async fn test_handle_release_worker_nonzero_exit_preserves_worker_health() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool.clone());
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo test".to_string(),
12345,
4,
rch_common::BuildLocation::Remote,
);
let build_id = build.id;
let worker = pool.get(&WorkerId::new("worker1")).await.unwrap();
worker.reserve_slots(4).await;
let request = ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("worker1"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(101),
duration_ms: Some(5000),
bytes_transferred: Some(1024 * 1024),
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
};
let result = handle_release_worker(&ctx, request).await;
assert!(result.is_ok());
let recent = ctx.history.recent(10);
let completed = recent
.iter()
.find(|b| b.id == build_id)
.expect("build should be completed");
assert_eq!(completed.exit_code, 101);
let stats = worker.circuit_stats().await;
assert_eq!(stats.error_rate(), 0.0);
assert!(stats.recent_results().is_empty());
assert_eq!(
ctx.worker_selector
.get_fallback_worker("test-project")
.await,
None
);
}
#[tokio::test]
async fn test_handle_build_heartbeat_updates_active_build_state() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
43210,
4,
rch_common::BuildLocation::Remote,
);
let response = handle_build_heartbeat(
&ctx,
BuildHeartbeatRequest {
build_id: build.id,
worker_id: WorkerId::new("worker1"),
hook_pid: Some(43210),
local_wrapper_id: Some("rchw-test".to_string()),
remote_pgid_file: Some("/tmp/rch/test-project/hash/.rch-run/99.pgid".to_string()),
phase: rch_common::BuildHeartbeatPhase::Execute,
detail: Some("Compiling".to_string()),
progress_counter: Some(4),
progress_percent: Some(18.0),
},
)
.await;
assert_eq!(response.status, "ok");
let active = ctx.history.active_build(build.id).expect("active build");
assert_eq!(
active.heartbeat_phase,
rch_common::BuildHeartbeatPhase::Execute
);
assert_eq!(
active.remote_pgid_file.as_deref(),
Some("/tmp/rch/test-project/hash/.rch-run/99.pgid")
);
assert_eq!(active.heartbeat_counter, 4);
assert_eq!(active.heartbeat_percent, Some(18.0));
}
#[tokio::test]
async fn test_handle_build_heartbeat_rejects_worker_mismatch() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let build = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
5555,
2,
rch_common::BuildLocation::Remote,
);
let response = handle_build_heartbeat(
&ctx,
BuildHeartbeatRequest {
build_id: build.id,
worker_id: WorkerId::new("worker-x"),
hook_pid: Some(5555),
local_wrapper_id: None,
remote_pgid_file: None,
phase: rch_common::BuildHeartbeatPhase::Execute,
detail: Some("Mismatch".to_string()),
progress_counter: Some(2),
progress_percent: None,
},
)
.await;
assert_eq!(response.status, "ignored");
}
#[tokio::test]
async fn test_handle_record_build_basic() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let result =
handle_record_build(&ctx, &WorkerId::new("worker1"), "my-project", false).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_handle_record_build_test_flag() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let result = handle_record_build(&ctx, &WorkerId::new("worker1"), "my-project", true).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_handle_status_empty_pool() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let result = handle_status(&ctx).await;
assert!(result.is_ok());
let status = result.unwrap();
assert!(status.workers.is_empty());
assert!(status.active_builds.is_empty());
assert!(status.queued_builds.is_empty());
assert_eq!(status.daemon.workers_total, 0);
assert_eq!(status.daemon.workers_healthy, 0);
}
#[tokio::test]
async fn test_handle_status_with_workers() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
pool.add_worker(make_test_worker("worker2", 4)).await;
let ctx = make_test_context(pool);
let result = handle_status(&ctx).await;
assert!(result.is_ok());
let status = result.unwrap();
assert_eq!(status.workers.len(), 2);
assert_eq!(status.daemon.workers_total, 2);
assert_eq!(status.daemon.slots_total, 12); assert_eq!(status.daemon.version, "0.1.0");
assert_eq!(status.daemon.pid, 1234);
}
#[tokio::test]
async fn test_handle_status_slots_available_counts_only_assignable_capacity() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("healthy", 8)).await;
pool.add_worker(make_test_worker("degraded", 4)).await;
pool.add_worker(make_test_worker("drained", 6)).await;
pool.add_worker(make_test_worker("disabled", 10)).await;
pool.add_worker(make_test_worker("open-circuit", 5)).await;
let healthy = pool
.get(&WorkerId::new("healthy"))
.await
.expect("healthy worker should exist");
assert!(healthy.reserve_slots(2).await);
let degraded = pool
.get(&WorkerId::new("degraded"))
.await
.expect("degraded worker should exist");
degraded.set_status(WorkerStatus::Degraded).await;
assert!(degraded.reserve_slots(1).await);
pool.set_status(&WorkerId::new("drained"), WorkerStatus::Drained)
.await;
pool.set_status(&WorkerId::new("disabled"), WorkerStatus::Disabled)
.await;
let open_circuit = pool
.get(&WorkerId::new("open-circuit"))
.await
.expect("open-circuit worker should exist");
open_circuit.open_circuit().await;
let ctx = make_test_context(pool);
let status = handle_status(&ctx).await.expect("status should succeed");
assert_eq!(status.daemon.slots_total, 33);
assert_eq!(
status.daemon.slots_available, 9,
"only healthy/degraded, non-open-circuit capacity should be assignable"
);
}
#[tokio::test]
async fn test_capacity_surfaces_exclude_critical_pressure_workers() {
use tower::ServiceExt;
let _guard = test_guard!();
async fn http_ready_status(pool: &WorkerPool) -> axum::http::StatusCode {
crate::http_api::create_router(crate::http_api::HttpState {
pool: pool.clone(),
version: "test",
started_at: Instant::now(),
pid: 1,
})
.oneshot(
axum::http::Request::builder()
.uri("/ready")
.body(axum::body::Body::empty())
.unwrap(),
)
.await
.unwrap()
.status()
}
for case in ["disk_critical", "memory_critical"] {
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("w1", 8)).await;
let worker = pool.get(&WorkerId::new("w1")).await.unwrap();
if case == "disk_critical" {
worker
.set_capabilities(WorkerCapabilities {
disk_free_gb: Some(45.0),
disk_total_gb: Some(926.0),
..Default::default()
})
.await;
} else {
worker
.set_pressure_assessment(PressureAssessment {
state: PressureState::Critical,
confidence: PressureConfidence::High,
reason_code: "memory_pressure_critical".to_string(),
policy_rule: "memory_pressure>=critical_memory_pressure".to_string(),
disk_free_gb: Some(500.0),
disk_total_gb: Some(926.0),
memory_pressure: Some(95.0),
telemetry_fresh: true,
..Default::default()
})
.await;
}
assert_eq!(
worker.pressure_assessment().await.state,
PressureState::Critical,
"{case}"
);
let ctx = make_test_context(pool.clone());
let status = handle_status(&ctx).await.unwrap();
assert_eq!(status.daemon.slots_available, 0, "{case}");
assert!(status.daemon.slots_total > 0, "{case}");
let ready = handle_ready(&ctx).await;
assert_eq!(ready.status, "not_ready", "{case}");
assert!(!ready.workers_available, "{case}");
assert_eq!(
http_ready_status(&pool).await,
axum::http::StatusCode::SERVICE_UNAVAILABLE,
"{case}"
);
pool.add_worker(make_test_worker("w2", 4)).await;
let ctx = make_test_context(pool.clone());
let status = handle_status(&ctx).await.unwrap();
assert_eq!(status.daemon.slots_available, 4, "{case}");
assert!(handle_ready(&ctx).await.workers_available, "{case}");
assert_eq!(
http_ready_status(&pool).await,
axum::http::StatusCode::OK,
"{case}"
);
}
}
#[tokio::test]
async fn test_disk_slots_status_preserves_usage_after_derating() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("disk-worker", 8)).await;
let worker = pool.get(&WorkerId::new("disk-worker")).await.unwrap();
assert!(worker.reserve_slots(4).await);
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(20.0),
..Default::default()
})
.await;
let ctx = make_test_context(pool);
let status = handle_status(&ctx).await.unwrap();
assert_eq!(status.workers[0].total_slots, 1);
assert_eq!(status.workers[0].used_slots, 4);
assert_eq!(status.daemon.slots_total, 1);
assert_eq!(status.daemon.slots_available, 0);
let snapshots = crate::ui::workers::WorkerStatusPanel::collect_snapshot(&ctx.pool).await;
assert_eq!(snapshots[0].total_slots, 1);
assert_eq!(snapshots[0].used_slots, 4);
}
#[tokio::test]
async fn test_disk_slots_selection_clamps_reservation_to_effective_capacity() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("disk-worker", 8)).await;
let worker = pool.get(&WorkerId::new("disk-worker")).await.unwrap();
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(40.0),
..Default::default()
})
.await;
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "disk-capacity".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 4,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(98765),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert_eq!(response.worker.unwrap().slots_available, 0);
assert_eq!(worker.used_slots(), 3);
let build_id = response.build_id.unwrap();
assert_eq!(ctx.history.active_build(build_id).unwrap().slots, 3);
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(70.0),
..Default::default()
})
.await;
assert!(worker.reserve_slots(1).await);
handle_release_worker(
&ctx,
ReleaseRequest {
local_wrapper_id: None,
worker_id: WorkerId::new("disk-worker"),
slots: 4,
build_id: Some(build_id),
exit_code: Some(0),
duration_ms: None,
bytes_transferred: None,
timing: None,
worker_fault: false,
worker_disk_full: false,
worker_disk_roots: Vec::new(),
},
)
.await
.unwrap();
assert_eq!(worker.used_slots(), 1);
assert_eq!(worker.available_slots().await, 5);
}
#[tokio::test]
async fn test_degraded_selection_reserves_free_slots_on_partly_busy_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("small-worker", 8)).await;
let worker = pool.get(&WorkerId::new("small-worker")).await.unwrap();
worker
.set_pressure_assessment(crate::disk_pressure::PressureAssessment {
disk_free_gb: Some(40.0),
..Default::default()
})
.await;
assert_eq!(worker.effective_total_slots().await, 3);
assert!(worker.reserve_slots(1).await);
let ctx = make_test_context(pool);
let request = SelectionRequest {
job_mode: false,
project: "degraded-busy".to_string(),
command: None,
command_priority: CommandPriority::Normal,
estimated_cores: 4,
disk_headroom_gib: 0,
preferred_workers: vec![],
toolchain: None,
required_runtime: RequiredRuntime::None,
classification_duration_us: None,
required_tools: Vec::new(),
hook_pid: Some(98766),
};
let response = handle_select_worker(&ctx, request, false, None)
.await
.unwrap();
assert!(
response.worker.is_some(),
"degraded worker with free slots must be admitted: {}",
response.reason
);
assert_eq!(worker.used_slots(), 3);
let build_id = response.build_id.unwrap();
assert_eq!(ctx.history.active_build(build_id).unwrap().slots, 2);
}
#[tokio::test]
async fn test_handle_status_includes_pressure_metadata() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let worker = pool
.get(&WorkerId::new("worker1"))
.await
.expect("worker should exist");
worker
.set_pressure_assessment(PressureAssessment {
state: PressureState::Warning,
confidence: PressureConfidence::High,
reason_code: "disk_free_below_warning_gb".to_string(),
policy_rule: "disk_free_gb<=warning_free_gb".to_string(),
disk_free_gb: Some(18.0),
disk_total_gb: Some(120.0),
disk_free_ratio: Some(0.15),
build_disk_free_gb: None,
build_disk_total_gb: None,
disk_io_util_pct: Some(72.0),
memory_pressure: Some(44.0),
telemetry_age_secs: Some(8),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
let ctx = make_test_context(pool);
let status = handle_status(&ctx).await.expect("status should succeed");
let worker = status
.workers
.iter()
.find(|entry| entry.id == "worker1")
.expect("worker1 status entry should exist");
assert_eq!(worker.pressure_state, "warning");
assert_eq!(worker.pressure_confidence, "high");
assert_eq!(worker.pressure_reason_code, "disk_free_below_warning_gb");
assert_eq!(worker.pressure_policy_rule, "disk_free_gb<=warning_free_gb");
assert_eq!(worker.pressure_disk_free_gb, Some(18.0));
assert_eq!(worker.pressure_disk_total_gb, Some(120.0));
assert_eq!(worker.pressure_disk_free_ratio, Some(0.15));
assert_eq!(worker.pressure_disk_io_util_pct, Some(72.0));
assert_eq!(worker.pressure_memory_pressure, Some(44.0));
assert_eq!(worker.pressure_telemetry_age_secs, Some(8));
assert!(worker.pressure_telemetry_fresh);
}
#[tokio::test]
async fn test_handle_status_emits_critical_pressure_issue() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let worker = pool
.get(&WorkerId::new("worker1"))
.await
.expect("worker should exist");
worker
.set_pressure_assessment(PressureAssessment {
state: PressureState::Critical,
confidence: 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(91.0),
memory_pressure: Some(82.0),
telemetry_age_secs: Some(7),
telemetry_fresh: true,
evaluated_at_unix_ms: 1_700_000_000_000,
})
.await;
let ctx = make_test_context(pool);
let status = handle_status(&ctx).await.expect("status should succeed");
assert!(
status
.issues
.iter()
.any(|issue| issue.summary.contains("critical pressure state")),
"status issues should include critical pressure signal"
);
}
#[tokio::test]
async fn test_handle_status_emits_cancellation_cleanup_issue() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let active = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo test".to_string(),
0,
4,
rch_common::BuildLocation::Remote,
);
let cancellation = BuildCancellationMetadata {
operation_id: "cancel-1".to_string(),
origin: "timeout".to_string(),
reason_code: "timeout".to_string(),
decision_path: vec![
"requested".to_string(),
"term_sent".to_string(),
"remote_kill_sent".to_string(),
"escalated".to_string(),
"completed".to_string(),
],
escalation_stage: "sigkill".to_string(),
escalation_count: 2,
remote_kill_attempted: true,
cleanup_ok: false,
history_cancelled: true,
final_state: "completed".to_string(),
worker_health: Some(BuildCancellationWorkerHealth {
status: "unreachable".to_string(),
speed_score: 0.0,
used_slots: 4,
available_slots: 0,
pressure_state: "critical".to_string(),
pressure_reason_code: "disk_free_below_critical_gb".to_string(),
}),
};
let _ = ctx
.history
.cancel_active_build(active.id, None, Some(cancellation));
let status = handle_status(&ctx).await.expect("status should succeed");
assert!(
status
.issues
.iter()
.any(|issue| issue.summary.contains("cleanup failures")),
"status issues should include cancellation cleanup failures"
);
assert!(
status
.issues
.iter()
.any(|issue| issue.summary.contains("SIGKILL")),
"status issues should include cancellation escalation warnings"
);
}
#[tokio::test]
async fn test_handle_status_emits_live_hook_progress_stall_issue() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
let active = ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo test".to_string(),
0,
4,
rch_common::BuildLocation::Remote,
);
let updated = ctx.history.record_stuck_detector_snapshot(
active.id,
crate::history::StuckDetectorSnapshot {
hook_alive: true,
heartbeat_stale: false,
progress_stale: true,
confidence: 0.25,
build_age_secs: 240,
slots_owned: 4,
},
);
assert!(updated.is_some());
let status = handle_status(&ctx).await.expect("status should succeed");
let issue = status
.issues
.iter()
.find(|issue| issue.summary.contains("stale progress"))
.expect("status issues should include live-hook progress stall");
assert_eq!(issue.severity, "info");
assert!(issue.summary.contains(&active.id.to_string()));
assert!(issue.summary.contains("worker1"));
let remediation = issue
.remediation
.as_ref()
.expect("stall issue should include remediation");
assert!(!remediation.contains("rch workers drain"));
assert!(!remediation.contains("rch cancel"));
assert!(remediation.contains("rch queue --json"));
}
#[tokio::test]
async fn test_handle_status_with_active_build() {
let _guard = test_guard!();
let pool = WorkerPool::new();
pool.add_worker(make_test_worker("worker1", 8)).await;
let ctx = make_test_context(pool);
ctx.history.start_active_build(
"test-project".to_string(),
"worker1".to_string(),
"cargo build".to_string(),
12345,
4,
rch_common::BuildLocation::Remote,
);
let result = handle_status(&ctx).await;
assert!(result.is_ok());
let status = result.unwrap();
assert_eq!(status.active_builds.len(), 1);
assert_eq!(status.active_builds[0].project_id, "test-project");
assert_eq!(status.active_builds[0].command, "cargo build");
assert_eq!(status.active_builds[0].heartbeat_phase, "sync_up");
assert_eq!(status.active_builds[0].heartbeat_counter, 0);
assert_eq!(status.active_builds[0].slots, 4);
assert_eq!(status.active_builds[0].detector_confidence, 0.0);
}
#[test]
fn test_format_wait_time_edge_cases() {
let _guard = test_guard!();
assert_eq!(format_wait_time(60), "1m");
assert_eq!(format_wait_time(3600), "1h");
assert_eq!(format_wait_time(5400), "1h 30m");
assert_eq!(format_wait_time(7200), "2h");
assert_eq!(format_wait_time(0), "0s");
assert_eq!(format_wait_time(59), "59s");
assert_eq!(format_wait_time(61), "1m 1s");
}
#[test]
fn test_handle_health() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_health(&ctx);
assert_eq!(response.status, "healthy");
assert_eq!(response.version, "0.1.0");
assert!(response.uptime_seconds < 10);
}
#[test]
fn test_handle_budget() {
let _guard = test_guard!();
let response = handle_budget();
assert!(!response.computed_at.is_empty());
}
#[tokio::test]
async fn test_convergence_status_empty() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let response = handle_repo_convergence_status(&ctx, None).await;
match response {
ApiResponse::Ok(status) => {
assert_eq!(status.workers.len(), 0);
assert_eq!(status.summary.total_workers, 0);
assert_eq!(status.status, "unknown");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_status_with_workers() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("w1");
ctx.repo_convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let response = handle_repo_convergence_status(&ctx, None).await;
match response {
ApiResponse::Ok(status) => {
assert_eq!(status.workers.len(), 1);
assert_eq!(status.workers[0].worker_id, "w1");
assert_eq!(status.workers[0].drift_state, "ready");
assert_eq!(status.summary.ready, 1);
assert_eq!(status.status, "healthy");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_status_filtered_by_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid1 = WorkerId::new("w1");
let wid2 = WorkerId::new("w2");
ctx.repo_convergence
.update_required_repos(&wid1, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
ctx.repo_convergence
.update_required_repos(&wid2, vec!["repo-b".into()], vec![])
.await;
let response = handle_repo_convergence_status(&ctx, Some(&wid1)).await;
match response {
ApiResponse::Ok(status) => {
assert_eq!(status.workers.len(), 1);
assert_eq!(status.workers[0].worker_id, "w1");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_status_unknown_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("nonexistent");
let response = handle_repo_convergence_status(&ctx, Some(&wid)).await;
match response {
ApiResponse::Ok(status) => {
assert_eq!(status.workers.len(), 1);
assert_eq!(status.workers[0].drift_state, "stale");
assert!(!status.workers[0].remediation.is_empty());
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_dry_run_ready_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("w1");
ctx.repo_convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let response = handle_repo_convergence_dry_run(&ctx, &wid).await;
match response {
ApiResponse::Ok(dr) => {
assert!(!dr.would_attempt, "Ready worker should not attempt sync");
assert_eq!(dr.current_state, "ready");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_dry_run_drifting_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("w1");
ctx.repo_convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec![])
.await;
let response = handle_repo_convergence_dry_run(&ctx, &wid).await;
match response {
ApiResponse::Ok(dr) => {
assert!(
dr.would_attempt,
"Drifting worker with budget should attempt"
);
assert_eq!(dr.current_state, "drifting");
assert!(!dr.missing_repos.is_empty());
assert!(dr.has_budget);
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_dry_run_unknown_worker() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("nonexistent");
let response = handle_repo_convergence_dry_run(&ctx, &wid).await;
match response {
ApiResponse::Ok(dr) => {
assert!(dr.would_attempt);
assert_eq!(dr.current_state, "stale");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_repair_resets_to_drifting() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("w1");
ctx.repo_convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec!["repo-a".into()])
.await;
let response = handle_repo_convergence_repair(&ctx, &wid).await;
match response {
ApiResponse::Ok(repair) => {
assert_eq!(repair.status, "ok");
assert_eq!(repair.previous_state, "ready");
assert_eq!(repair.new_state, "drifting");
assert_eq!(repair.action, "reset_convergence");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_repair_no_repos_noop() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("unknown");
let response = handle_repo_convergence_repair(&ctx, &wid).await;
match response {
ApiResponse::Ok(repair) => {
assert_eq!(repair.status, "noop");
assert_eq!(repair.action, "none");
}
ApiResponse::Error(e) => assert!(false, "Expected Ok, got error: {}", e.message),
}
}
#[tokio::test]
async fn test_convergence_status_json_serializable() {
let _guard = test_guard!();
let pool = WorkerPool::new();
let ctx = make_test_context(pool);
let wid = WorkerId::new("w1");
ctx.repo_convergence
.update_required_repos(&wid, vec!["repo-a".into()], vec![])
.await;
let response = handle_repo_convergence_status(&ctx, None).await;
let json = serde_json::to_string(&response).expect("Should serialize to JSON");
assert!(json.contains("drifting"));
assert!(json.contains("w1"));
}
#[test]
fn test_parse_convergence_status_route() {
let _guard = test_guard!();
let req = parse_request("GET /repo-convergence/status").unwrap();
assert!(
matches!(req, ApiRequest::RepoConvergenceStatus { worker_id: None }),
"Should parse status route without worker filter"
);
}
#[test]
fn test_parse_convergence_status_with_worker() {
let _guard = test_guard!();
let req = parse_request("GET /repo-convergence/status?worker=w1").unwrap();
match req {
ApiRequest::RepoConvergenceStatus {
worker_id: Some(wid),
} => {
assert_eq!(wid.as_str(), "w1");
}
_ => assert!(false, "Expected RepoConvergenceStatus with worker_id"),
}
}
#[test]
fn test_parse_convergence_dry_run_route() {
let _guard = test_guard!();
let req = parse_request("GET /repo-convergence/dry-run?worker=w1").unwrap();
match req {
ApiRequest::RepoConvergenceDryRun { worker_id } => {
assert_eq!(worker_id.as_str(), "w1");
}
_ => assert!(false, "Expected RepoConvergenceDryRun"),
}
}
#[test]
fn test_parse_convergence_dry_run_missing_worker() {
let _guard = test_guard!();
let result = parse_request("GET /repo-convergence/dry-run");
assert!(result.is_err(), "Should fail without worker parameter");
}
#[test]
fn test_parse_convergence_repair_route() {
let _guard = test_guard!();
let req = parse_request("POST /repo-convergence/repair?worker=w1").unwrap();
match req {
ApiRequest::RepoConvergenceRepair { worker_id } => {
assert_eq!(worker_id.as_str(), "w1");
}
_ => assert!(false, "Expected RepoConvergenceRepair"),
}
}
#[test]
fn test_parse_convergence_repair_requires_post() {
let _guard = test_guard!();
let result = parse_request("GET /repo-convergence/repair?worker=w1");
assert!(result.is_err(), "Repair should require POST method");
}
#[test]
fn test_parse_convergence_unknown_subpath() {
let _guard = test_guard!();
let result = parse_request("GET /repo-convergence/unknown");
assert!(result.is_err(), "Unknown subpath should error");
}
#[test]
fn worker_issue_disabled_beats_the_telemetry_gap_it_causes() {
let _guard = test_guard!();
let issue = worker_issue(
"hz3",
WorkerStatus::Disabled,
CircuitState::Closed,
PressureState::TelemetryGap,
"telemetry_unavailable",
Some("e104-timeout-orphan-unverified: remote build process group not verified dead"),
)
.expect("a disabled worker must surface an issue");
assert_eq!(issue.severity, "warning");
assert!(
issue.summary.contains("administratively disabled"),
"summary must name the real cause, got: {}",
issue.summary
);
assert!(
issue.summary.contains("e104-timeout-orphan-unverified"),
"summary must carry the recorded disable reason, got: {}",
issue.summary
);
assert!(
!issue.summary.contains("stale/missing pressure telemetry"),
"the telemetry gap must not mask the disable, got: {}",
issue.summary
);
let fix = issue.remediation.expect("a disable is actionable");
assert!(
fix.contains("rch workers enable hz3"),
"must point at the only command that clears it, got: {fix}"
);
assert!(
!fix.contains("daemon restart"),
"a restart re-applies the durable disable; never suggest it here, got: {fix}"
);
}
#[test]
fn worker_issue_drained_beats_the_telemetry_gap_it_causes() {
let _guard = test_guard!();
let issue = worker_issue(
"hz4",
WorkerStatus::Drained,
CircuitState::Closed,
PressureState::TelemetryGap,
"telemetry_unavailable",
None,
)
.expect("a drained worker must surface an issue");
assert_eq!(issue.severity, "warning");
assert!(issue.summary.contains("drained"), "got: {}", issue.summary);
assert!(
!issue.summary.contains("stale/missing pressure telemetry"),
"got: {}",
issue.summary
);
assert!(
issue
.remediation
.expect("drain is actionable")
.contains("rch workers enable hz4")
);
}
#[test]
fn worker_issue_disabled_without_a_recorded_reason_is_still_explicit() {
let _guard = test_guard!();
let issue = worker_issue(
"omarchy",
WorkerStatus::Disabled,
CircuitState::Closed,
PressureState::Healthy,
"pressure_healthy",
None,
)
.expect("a disabled worker must surface an issue");
assert!(
issue.summary.contains("no reason recorded"),
"got: {}",
issue.summary
);
}
#[test]
fn worker_issue_disabled_beats_an_open_circuit() {
let _guard = test_guard!();
let issue = worker_issue(
"wsurf",
WorkerStatus::Disabled,
CircuitState::Open,
PressureState::TelemetryGap,
"telemetry_unavailable",
Some("maintenance"),
)
.expect("issue expected");
assert!(
issue.summary.contains("administratively disabled"),
"got: {}",
issue.summary
);
assert!(
!issue.summary.contains("Circuit open"),
"got: {}",
issue.summary
);
}
#[test]
fn worker_issue_active_worker_keeps_the_telemetry_gap_warning() {
let _guard = test_guard!();
let issue = worker_issue(
"vmi1264463",
WorkerStatus::Healthy,
CircuitState::Closed,
PressureState::TelemetryGap,
"disk_metrics_unavailable",
None,
)
.expect("an active worker in a telemetry gap must still warn");
assert!(
issue.summary.contains("stale/missing pressure telemetry"),
"got: {}",
issue.summary
);
assert!(
issue
.remediation
.expect("gap advice retained")
.contains("daemon-driven")
);
}
#[test]
fn worker_issue_preserves_the_pre_existing_arms() {
let _guard = test_guard!();
let circuit = worker_issue(
"w1",
WorkerStatus::Healthy,
CircuitState::Open,
PressureState::Healthy,
"pressure_healthy",
None,
)
.expect("issue expected");
assert_eq!(circuit.severity, "error");
assert!(circuit.summary.contains("Circuit open"));
let critical = worker_issue(
"w2",
WorkerStatus::Healthy,
CircuitState::Closed,
PressureState::Critical,
"disk_ratio_below_critical",
None,
)
.expect("issue expected");
assert_eq!(critical.severity, "error");
assert!(critical.summary.contains("critical pressure state"));
let unreachable = worker_issue(
"w3",
WorkerStatus::Unreachable,
CircuitState::Closed,
PressureState::Healthy,
"pressure_healthy",
None,
)
.expect("issue expected");
assert_eq!(unreachable.severity, "error");
assert!(unreachable.summary.contains("is unreachable"));
let degraded = worker_issue(
"w4",
WorkerStatus::Degraded,
CircuitState::Closed,
PressureState::Healthy,
"pressure_healthy",
None,
)
.expect("issue expected");
assert_eq!(degraded.severity, "warning");
assert!(degraded.remediation.is_none());
assert!(
worker_issue(
"w5",
WorkerStatus::Healthy,
CircuitState::Closed,
PressureState::Healthy,
"pressure_healthy",
None,
)
.is_none(),
"a healthy worker must report no issue"
);
}
}