use std::fmt;
use std::num::NonZeroU16;
use crate::attempt::RunnerAttempt;
use crate::model::{Host, HostId, PolicyId};
use crate::policy::ScalePolicy;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LimitingFactor {
Demand,
MinCapacity,
MaxCapacity,
HostCapacity,
MonitorOnly,
NotReconciling,
ForeignHost,
}
impl fmt::Display for LimitingFactor {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
LimitingFactor::Demand => "demand",
LimitingFactor::MinCapacity => "min_capacity",
LimitingFactor::MaxCapacity => "max_capacity",
LimitingFactor::HostCapacity => "host_capacity",
LimitingFactor::MonitorOnly => "monitor_only",
LimitingFactor::NotReconciling => "not_reconciling",
LimitingFactor::ForeignHost => "foreign_host",
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Allocation {
pub policy_id: PolicyId,
pub demand: u32,
pub desired: u16,
pub active_owned: u16,
pub headroom_before: u16,
pub to_start: u16,
pub limiting_factor: LimitingFactor,
}
impl Allocation {
#[must_use]
pub const fn starts_nothing(&self) -> bool {
self.to_start == 0
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HostAllocator<'a> {
host_id: HostId,
host_capacity: NonZeroU16,
attempts: Vec<&'a RunnerAttempt>,
active_total: u16,
}
impl<'a> HostAllocator<'a> {
#[must_use]
pub fn from_attempts(
host: &Host,
attempts: impl IntoIterator<Item = &'a RunnerAttempt>,
) -> Self {
let attempts: Vec<&'a RunnerAttempt> = attempts.into_iter().collect();
let active_total = crate::attempt::active_count(attempts.iter().copied());
Self {
host_id: host.id,
host_capacity: host.host_capacity,
attempts,
active_total,
}
}
#[must_use]
pub fn host_capacity(&self) -> u16 {
self.host_capacity.get()
}
#[must_use]
pub const fn active_total(&self) -> u16 {
self.active_total
}
#[must_use]
pub fn headroom(&self) -> u16 {
self.host_capacity.get().saturating_sub(self.active_total)
}
pub fn allocate(&mut self, policy: &ScalePolicy, demand: u32) -> Allocation {
let active_owned =
crate::attempt::active_count_for(policy.id, self.attempts.iter().copied());
let headroom_before = self.headroom();
let refuse = |limiting_factor| Allocation {
policy_id: policy.id,
demand,
desired: 0,
active_owned,
headroom_before,
to_start: 0,
limiting_factor,
};
if !policy.is_owned_by(self.host_id) {
return refuse(LimitingFactor::ForeignHost);
}
if !policy.owns_runners() {
return refuse(LimitingFactor::MonitorOnly);
}
if !policy.may_start_runners() {
return refuse(LimitingFactor::NotReconciling);
}
let min = policy.min_capacity();
let max = policy
.max_capacity()
.expect("an Autoscale policy always has a max_capacity (D19)")
.get();
debug_assert!(min <= max, "PolicyMode invariant");
let desired = demand.clamp(u32::from(min), u32::from(max)) as u16;
let limiting_factor = if demand > u32::from(max) {
LimitingFactor::MaxCapacity
} else if demand < u32::from(min) {
LimitingFactor::MinCapacity
} else {
LimitingFactor::Demand
};
let wanted = desired.saturating_sub(active_owned);
let to_start = wanted.min(headroom_before);
let limiting_factor = if to_start < wanted {
LimitingFactor::HostCapacity
} else {
limiting_factor
};
self.active_total = self.active_total.saturating_add(to_start);
Allocation {
policy_id: policy.id,
demand,
desired,
active_owned,
headroom_before,
to_start,
limiting_factor,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::attempt::{
AttemptOutcome, AttemptState, FailureReason, PersistedAttempt, RunnerAttempt,
};
use crate::model::{
Arch, AttemptId, CachePolicy, HostLabel, Os, PolicyId, ScaleTarget, Timestamp,
};
use crate::policy::{PolicyMode, RoutingLabels, RunsOn, ScalePolicy};
use crate::workspace::WorkspaceKind;
fn ts(secs: i64) -> Timestamp {
chrono::DateTime::from_timestamp(secs, 0).expect("valid timestamp")
}
fn nz(v: u16) -> NonZeroU16 {
NonZeroU16::new(v).expect("non-zero")
}
const HOST: HostId = HostId::from_u128(7);
const NO_ATTEMPTS: &[RunnerAttempt] = &[];
fn host(capacity: u16) -> Host {
Host::new(HOST, "home-pc", Os::Windows, Arch::X64, nz(capacity), ts(0)).expect("valid host")
}
fn labels(name: &str) -> RoutingLabels {
RoutingLabels::derive(&HostLabel::new(name).unwrap(), Os::Windows, Arch::X64)
}
fn active_policy(id: u128, host_label: &str, max: u16) -> ScalePolicy {
let mut policy = ScalePolicy::new(
PolicyId::from_u128(id),
ScaleTarget::repository("o/r").unwrap(),
1,
HOST,
PolicyMode::autoscale(labels(host_label), 0, nz(max)).unwrap(),
CachePolicy::default(),
);
policy.activate().expect("pending -> active");
policy
}
fn attempt_in(state: AttemptState, id: u128, policy: u128) -> RunnerAttempt {
let outcome = state.is_terminal().then(|| match state {
AttemptState::Failed => {
AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
}
AttemptState::Orphaned => AttemptOutcome::Orphaned,
_ => AttemptOutcome::CompletedJob,
});
RunnerAttempt::from_persisted(PersistedAttempt {
id: AttemptId::from_u128(id),
policy_id: PolicyId::from_u128(policy),
github_runner_id: None,
state,
outcome,
process_id: None,
runtime_path: "runtime/p/a".into(),
workspace_kind: WorkspaceKind::Ephemeral,
workspace_slot: None,
created_at: ts(0),
terminal_at: state.is_terminal().then(|| ts(0)),
last_state_change_at: ts(0),
})
.expect("a state/outcome pair the domain accepts")
}
#[test]
fn desired_clamps_above_and_below() {
let host = host(100);
let policy = active_policy(1, "home", 3);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let above = alloc.allocate(&policy, 10);
assert_eq!(above.desired, 3, "max_capacity beats reported demand");
assert_eq!(above.to_start, 3);
assert_eq!(above.limiting_factor, LimitingFactor::MaxCapacity);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let inside = alloc.allocate(&policy, 2);
assert_eq!(inside.desired, 2);
assert_eq!(inside.to_start, 2);
assert_eq!(inside.limiting_factor, LimitingFactor::Demand);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let none = alloc.allocate(&policy, 0);
assert_eq!(none.desired, 0);
assert_eq!(none.to_start, 0);
assert!(none.starts_nothing());
}
#[test]
fn a_non_zero_min_capacity_raises_desired_above_demand() {
let host = host(10);
let mut policy = ScalePolicy::new(
PolicyId::from_u128(1),
ScaleTarget::organization("acme").unwrap(),
1,
HOST,
PolicyMode::autoscale(labels("home"), 2, nz(5)).unwrap(),
CachePolicy::default(),
);
policy.activate().unwrap();
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&policy, 0);
assert_eq!(got.desired, 2);
assert_eq!(got.to_start, 2);
assert_eq!(got.limiting_factor, LimitingFactor::MinCapacity);
}
#[test]
fn the_same_queued_job_on_two_polls_yields_one_attempt_not_two() {
let host = host(4);
let policy = active_policy(1, "home", 4);
let queued = vec![RunsOn::Single("rm-home-win-x64".into())];
let demand = policy.tally(&queued).demand();
assert_eq!(demand, 1);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let first = alloc.allocate(&policy, demand);
assert_eq!(first.to_start, 1);
let attempts = vec![attempt_in(AttemptState::Starting, 1, 1)];
for poll in 2..=3 {
let demand = policy.tally(&queued).demand();
assert_eq!(demand, 1, "poll {poll}: the job has not left the queue");
let mut alloc = HostAllocator::from_attempts(&host, &attempts);
let again = alloc.allocate(&policy, demand);
assert_eq!(
again.to_start, 0,
"poll {poll} started another runner for a job already being \
served; the `- active_owned_runners` term was dropped from the \
formula"
);
assert_eq!(again.desired, 1);
assert_eq!(again.active_owned, 1);
}
}
#[test]
fn mutant_ignoring_in_flight_attempts_is_detected() {
let host = host(4);
let policy = active_policy(1, "home", 4);
let attempts = vec![attempt_in(AttemptState::Starting, 1, 1)];
let mut allocator = HostAllocator::from_attempts(&host, &attempts);
let protected = allocator.allocate(&policy, 1);
assert_eq!(protected.active_owned, 1);
assert_eq!(protected.to_start, 0);
let mutant_active_owned = 0_u16;
let mutant_to_start = protected
.desired
.saturating_sub(mutant_active_owned)
.min(protected.headroom_before);
assert_eq!(
mutant_to_start, 1,
"removing the in-flight term must make the duplicate-poll gate red"
);
}
#[test]
fn an_attempt_stops_counting_once_it_is_terminal() {
let host = host(4);
let policy = active_policy(1, "home", 4);
let in_flight = vec![
attempt_in(AttemptState::Allocated, 1, 1),
attempt_in(AttemptState::Starting, 2, 1),
attempt_in(AttemptState::Busy, 3, 1),
];
let mut alloc = HostAllocator::from_attempts(&host, &in_flight);
assert_eq!(alloc.active_total(), 3);
assert_eq!(alloc.headroom(), 1);
assert_eq!(alloc.allocate(&policy, 4).to_start, 1);
let done = vec![
attempt_in(AttemptState::Finished, 1, 1),
attempt_in(AttemptState::Failed, 2, 1),
attempt_in(AttemptState::Cleaned, 3, 1),
];
let mut alloc = HostAllocator::from_attempts(&host, &done);
assert_eq!(alloc.active_total(), 0);
assert_eq!(alloc.headroom(), 4);
assert_eq!(alloc.allocate(&policy, 4).to_start, 4);
}
#[test]
fn one_attempt_set_answers_both_ceilings() {
let host = host(10);
let mine = active_policy(1, "home", 9);
let theirs = active_policy(2, "office", 9);
let on_the_machine = vec![
attempt_in(AttemptState::Busy, 1, 1),
attempt_in(AttemptState::Starting, 2, 1),
attempt_in(AttemptState::Idle, 3, 2),
attempt_in(AttemptState::Finished, 4, 1),
];
let mut alloc = HostAllocator::from_attempts(&host, &on_the_machine);
assert_eq!(alloc.active_total(), 3, "host-wide (D9), from the one set");
let got = alloc.allocate(&mine, 9);
assert_eq!(
got.active_owned, 2,
"per-policy (D7), from the same set and with no second argument that \
could have said otherwise"
);
assert_eq!(got.headroom_before, 7);
assert_eq!(got.to_start, 7, "9 wanted, 2 already in flight, 7 free");
let got = alloc.allocate(&theirs, 9);
assert_eq!(got.active_owned, 1);
assert_eq!(
got.to_start, 0,
"the first grant spent the headroom the second would have used"
);
}
#[test]
fn the_host_ceiling_binds_across_two_policies_whose_max_capacities_sum_higher() {
let host = host(3);
let a = active_policy(1, "home", 3);
let b = active_policy(2, "home", 3);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let first = alloc.allocate(&a, 10);
let second = alloc.allocate(&b, 10);
assert_eq!(first.to_start, 3, "the first policy takes the whole host");
assert_eq!(
second.to_start, 0,
"the second gets nothing; each policy is individually within its own \
max_capacity of 3, and 3 + 3 > host_capacity of 3"
);
assert_eq!(second.limiting_factor, LimitingFactor::HostCapacity);
assert_eq!(
first.to_start + second.to_start,
3,
"the sum across policies must never exceed host_capacity"
);
assert_eq!(alloc.headroom(), 0);
}
#[test]
fn the_host_ceiling_splits_headroom_between_policies_in_call_order() {
let host = host(5);
let a = active_policy(1, "home", 4);
let b = active_policy(2, "home", 4);
let c = active_policy(3, "home", 4);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let first = alloc.allocate(&a, 4);
let second = alloc.allocate(&b, 4);
let third = alloc.allocate(&c, 4);
assert_eq!(first.to_start, 4);
assert_eq!(second.to_start, 1, "one slot of headroom left");
assert_eq!(second.limiting_factor, LimitingFactor::HostCapacity);
assert_eq!(third.to_start, 0);
assert_eq!(
first.to_start + second.to_start + third.to_start,
5,
"12 requested across three policies, 5 granted, which is host_capacity"
);
}
#[test]
fn zero_headroom_starts_nothing_even_at_maximum_demand() {
let host = host(2);
let policy = active_policy(1, "home", 2);
let full = vec![
attempt_in(AttemptState::Busy, 1, 1),
attempt_in(AttemptState::Busy, 2, 1),
];
let mut alloc = HostAllocator::from_attempts(&host, &full);
assert_eq!(alloc.headroom(), 0);
let got = alloc.allocate(&policy, u32::from(u16::MAX));
assert_eq!(got.to_start, 0);
assert_eq!(got.headroom_before, 0);
assert_eq!(got.limiting_factor, LimitingFactor::MaxCapacity);
}
#[test]
fn headroom_smaller_than_the_per_policy_allowance_wins() {
let host = host(6);
let policy = active_policy(1, "home", 5);
let others = vec![
attempt_in(AttemptState::Busy, 1, 99),
attempt_in(AttemptState::Busy, 2, 99),
attempt_in(AttemptState::Idle, 3, 99),
attempt_in(AttemptState::Starting, 4, 99),
];
let mut alloc = HostAllocator::from_attempts(&host, &others);
assert_eq!(alloc.headroom(), 2, "four slots are held by another policy");
let got = alloc.allocate(&policy, 5);
assert_eq!(got.desired, 5, "the policy's own ceiling would allow five");
assert_eq!(got.to_start, 2, "but the host has only two slots free");
assert_eq!(got.limiting_factor, LimitingFactor::HostCapacity);
assert_eq!(alloc.headroom(), 0);
}
#[test]
fn an_over_subscribed_host_reports_zero_headroom_rather_than_wrapping() {
let host = host(2);
let policy = active_policy(1, "home", 10);
let oversubscribed: Vec<RunnerAttempt> = (1..=9)
.map(|id| attempt_in(AttemptState::Busy, id, 99))
.collect();
let mut alloc = HostAllocator::from_attempts(&host, &oversubscribed);
assert_eq!(
alloc.active_total(),
9,
"the raw count is reported, over-subscription included"
);
assert_eq!(alloc.headroom(), 0);
assert_eq!(alloc.allocate(&policy, 10).to_start, 0);
}
#[test]
fn a_monitor_only_policy_under_maximum_demand_starts_nothing() {
let host = host(10);
let mut policy = ScalePolicy::new(
PolicyId::from_u128(1),
ScaleTarget::organization("acme").unwrap(),
1,
HOST,
PolicyMode::monitor_only(),
CachePolicy::default(),
);
policy.activate().unwrap();
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&policy, 1_000);
assert_eq!(got.to_start, 0);
assert_eq!(got.limiting_factor, LimitingFactor::MonitorOnly);
assert_eq!(
alloc.headroom(),
10,
"and it consumes no headroom, so an autoscale policy on the same host \
is unaffected"
);
}
#[test]
fn a_policy_that_is_not_active_and_enabled_starts_nothing() {
let host = host(10);
let pending = ScalePolicy::new(
PolicyId::from_u128(1),
ScaleTarget::repository("o/r").unwrap(),
1,
HOST,
PolicyMode::autoscale(labels("home"), 0, nz(5)).unwrap(),
CachePolicy::default(),
);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&pending, 5);
assert_eq!(got.to_start, 0);
assert_eq!(got.limiting_factor, LimitingFactor::NotReconciling);
let mut draining = active_policy(2, "home", 5);
draining.request_disable().unwrap();
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&draining, 5);
assert_eq!(got.to_start, 0);
assert_eq!(got.limiting_factor, LimitingFactor::NotReconciling);
assert_eq!(alloc.headroom(), 10);
}
#[test]
fn a_policy_belonging_to_another_host_is_refused_before_any_headroom_is_spent() {
let host = host(4);
let mut theirs = ScalePolicy::new(
PolicyId::from_u128(1),
ScaleTarget::repository("o/r").unwrap(),
1,
HostId::from_u128(8),
PolicyMode::autoscale(labels("office"), 0, nz(4)).unwrap(),
CachePolicy::default(),
);
theirs.activate().unwrap();
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&theirs, 4);
assert_eq!(got.to_start, 0);
assert_eq!(got.limiting_factor, LimitingFactor::ForeignHost);
assert_eq!(alloc.headroom(), 4);
}
#[test]
fn max_capacity_beats_demand_and_host_capacity_beats_max_capacity() {
let host = host(2);
let policy = active_policy(1, "home", 4);
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
let got = alloc.allocate(&policy, 9);
assert_eq!(got.demand, 9);
assert_eq!(got.desired, 4, "max_capacity beats reported demand");
assert_eq!(got.to_start, 2, "host_capacity beats max_capacity");
assert_eq!(got.limiting_factor, LimitingFactor::HostCapacity);
}
#[test]
fn an_idle_host_with_no_demand_starts_no_runners() {
let host = host(8);
let policies = [active_policy(1, "home", 4), active_policy(2, "home", 4)];
let mut alloc = HostAllocator::from_attempts(&host, NO_ATTEMPTS);
for policy in &policies {
let got = alloc.allocate(policy, 0);
assert_eq!(got.to_start, 0);
assert_eq!(got.desired, 0);
}
assert_eq!(alloc.active_total(), 0);
assert_eq!(alloc.headroom(), 8);
}
}