use std::fmt;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use crate::model::{AttemptId, Clock, Elapsed, HostId, PolicyId, Timestamp};
use crate::policy::ScalePolicy;
use crate::workspace::{AttemptWorkspace, WorkspaceError, WorkspaceKind};
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum AttemptError {
#[error("{to} is not a legal transition from {from}")]
IllegalTransition {
from: AttemptState,
to: AttemptState,
},
#[error(transparent)]
Workspace(#[from] WorkspaceError),
#[error(
"a busy attempt must not be cleaned; capacity is reclaimed only when an \
attempt reaches a terminal state, never by stopping a runner that is \
executing a job"
)]
BusyCannotBeCleaned,
#[error("the outcome {outcome:?} cannot be reached from {from}")]
OutcomeUnreachable {
from: AttemptState,
outcome: AttemptOutcome,
},
#[error("{state} is a terminal state and requires an outcome, but none was recorded")]
TerminalWithoutOutcome { state: AttemptState },
#[error("{state} is not terminal, so it must not carry the outcome {outcome:?}")]
NonTerminalWithOutcome {
state: AttemptState,
outcome: AttemptOutcome,
},
#[error("the recorded outcome {outcome:?} does not belong to the recorded state {state}")]
OutcomeStateMismatch {
state: AttemptState,
outcome: AttemptOutcome,
},
#[error("{state} is a terminal state and requires a terminal_at, but none was recorded")]
TerminalWithoutTimestamp { state: AttemptState },
#[error("{state} is not terminal, so it must not carry a terminal_at")]
NonTerminalWithTimestamp { state: AttemptState },
#[error(
"{state} carries {field} at {found}, which is before its created_at of \
{created_at}; an attempt cannot have changed state or concluded before \
it was allocated"
)]
TimestampsOutOfOrder {
state: AttemptState,
field: &'static str,
created_at: Timestamp,
found: Timestamp,
},
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum OwnershipError {
#[error(
"attempt {attempt} belongs to policy {attempt_policy}, but was checked \
against policy {policy}"
)]
PolicyMismatch {
attempt: AttemptId,
attempt_policy: PolicyId,
policy: PolicyId,
},
#[error(
"policy {policy} belongs to host {owner}; this agent runs on host \
{agent} and must not act on its attempts"
)]
ForeignHost {
policy: PolicyId,
owner: HostId,
agent: HostId,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptState {
Allocated,
JitReceived,
Starting,
Idle,
Busy,
Finished,
Failed,
Orphaned,
Cleaned,
}
impl AttemptState {
pub const ALL: [AttemptState; 9] = [
AttemptState::Allocated,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
AttemptState::Finished,
AttemptState::Failed,
AttemptState::Orphaned,
AttemptState::Cleaned,
];
pub const LEGAL: &'static [(AttemptState, AttemptState)] = &[
(AttemptState::Allocated, AttemptState::JitReceived),
(AttemptState::JitReceived, AttemptState::Starting),
(AttemptState::Starting, AttemptState::Idle),
(AttemptState::Starting, AttemptState::Busy),
(AttemptState::Idle, AttemptState::Busy),
(AttemptState::Allocated, AttemptState::Failed),
(AttemptState::Allocated, AttemptState::Orphaned),
(AttemptState::JitReceived, AttemptState::Failed),
(AttemptState::JitReceived, AttemptState::Orphaned),
(AttemptState::Starting, AttemptState::Failed),
(AttemptState::Starting, AttemptState::Orphaned),
(AttemptState::Idle, AttemptState::Finished),
(AttemptState::Idle, AttemptState::Failed),
(AttemptState::Idle, AttemptState::Orphaned),
(AttemptState::Busy, AttemptState::Finished),
(AttemptState::Busy, AttemptState::Failed),
(AttemptState::Busy, AttemptState::Orphaned),
(AttemptState::Finished, AttemptState::Cleaned),
(AttemptState::Failed, AttemptState::Cleaned),
(AttemptState::Orphaned, AttemptState::Cleaned),
];
pub const CONCLUDABLE_FROM: &'static [AttemptState] = &[
AttemptState::Allocated,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
];
#[must_use]
pub fn can_transition_to(self, next: AttemptState) -> bool {
Self::LEGAL.contains(&(self, next))
}
#[must_use]
pub const fn is_terminal(self) -> bool {
matches!(
self,
AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Cleaned
)
}
#[must_use]
pub const fn counts_against_capacity(self) -> bool {
!self.is_terminal()
}
#[must_use]
pub const fn is_concluded(self) -> bool {
matches!(
self,
AttemptState::Finished | AttemptState::Failed | AttemptState::Orphaned
)
}
}
impl fmt::Display for AttemptState {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
AttemptState::Allocated => "allocated",
AttemptState::JitReceived => "jit_received",
AttemptState::Starting => "starting",
AttemptState::Idle => "idle",
AttemptState::Busy => "busy",
AttemptState::Finished => "finished",
AttemptState::Failed => "failed",
AttemptState::Orphaned => "orphaned",
AttemptState::Cleaned => "cleaned",
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FailureReason {
JitRequestFailed,
JitExpired,
RunnerPackageUnverified,
RunnerVersionRejected,
ProcessStartFailed,
ProcessExitedUnexpectedly,
RegistrationTimedOut,
TerminatedAfterRegistrationTimeout,
Other(String),
}
impl FailureReason {
pub const ALL: [FailureReason; 9] = [
FailureReason::JitRequestFailed,
FailureReason::JitExpired,
FailureReason::RunnerPackageUnverified,
FailureReason::RunnerVersionRejected,
FailureReason::ProcessStartFailed,
FailureReason::ProcessExitedUnexpectedly,
FailureReason::RegistrationTimedOut,
FailureReason::TerminatedAfterRegistrationTimeout,
FailureReason::Other(String::new()),
];
}
impl fmt::Display for FailureReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
FailureReason::JitRequestFailed => f.write_str("the JIT configuration request failed"),
FailureReason::JitExpired => f.write_str("the JIT configuration expired unclaimed"),
FailureReason::RunnerPackageUnverified => {
f.write_str("the runner package could not be verified")
}
FailureReason::RunnerVersionRejected => {
f.write_str("GitHub rejected the runner version")
}
FailureReason::ProcessStartFailed => f.write_str("the runner process failed to start"),
FailureReason::ProcessExitedUnexpectedly => {
f.write_str("the runner process exited unexpectedly")
}
FailureReason::RegistrationTimedOut => f.write_str(
"the runner process is running but did not register with GitHub \
before its startup deadline",
),
FailureReason::TerminatedAfterRegistrationTimeout => f.write_str(
"the agent stopped the runner process after it failed to \
register with GitHub before its startup deadline",
),
FailureReason::Other(detail) => write!(f, "{detail}"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum AttemptOutcome {
CompletedJob,
ExitedIdleWithoutWork,
Failed { reason: FailureReason },
Orphaned,
}
impl AttemptOutcome {
#[must_use]
pub fn failed(reason: FailureReason) -> Self {
Self::Failed { reason }
}
#[must_use]
pub const fn terminal_state(&self) -> AttemptState {
match self {
AttemptOutcome::CompletedJob | AttemptOutcome::ExitedIdleWithoutWork => {
AttemptState::Finished
}
AttemptOutcome::Failed { .. } => AttemptState::Failed,
AttemptOutcome::Orphaned => AttemptState::Orphaned,
}
}
#[must_use]
pub const fn is_failure(&self) -> bool {
matches!(
self,
AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned
)
}
#[must_use]
pub const fn is_idle_exit(&self) -> bool {
matches!(self, AttemptOutcome::ExitedIdleWithoutWork)
}
#[must_use]
pub const fn ran_a_job(&self) -> bool {
matches!(self, AttemptOutcome::CompletedJob)
}
const fn required_from(&self) -> &'static [AttemptState] {
match self {
AttemptOutcome::CompletedJob => &[AttemptState::Busy],
AttemptOutcome::ExitedIdleWithoutWork => &[AttemptState::Idle],
AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned => {
AttemptState::CONCLUDABLE_FROM
}
}
}
}
impl fmt::Display for AttemptOutcome {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
AttemptOutcome::CompletedJob => f.write_str("ran a job"),
AttemptOutcome::ExitedIdleWithoutWork => f.write_str("exited idle without work"),
AttemptOutcome::Failed { reason } => write!(f, "failed: {reason}"),
AttemptOutcome::Orphaned => f.write_str("orphaned"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunnerAttempt {
pub id: AttemptId,
pub policy_id: PolicyId,
github_runner_id: Option<u64>,
state: AttemptState,
outcome: Option<AttemptOutcome>,
process_id: Option<u32>,
runtime_path: PathBuf,
workspace: AttemptWorkspace,
pub created_at: Timestamp,
terminal_at: Option<Timestamp>,
last_state_change_at: Timestamp,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PersistedAttempt {
pub id: AttemptId,
pub policy_id: PolicyId,
pub github_runner_id: Option<u64>,
pub state: AttemptState,
pub outcome: Option<AttemptOutcome>,
pub process_id: Option<u32>,
pub runtime_path: PathBuf,
pub workspace_kind: WorkspaceKind,
pub workspace_slot: Option<u16>,
pub created_at: Timestamp,
pub terminal_at: Option<Timestamp>,
pub last_state_change_at: Timestamp,
}
impl RunnerAttempt {
#[must_use]
pub fn to_persisted(&self) -> PersistedAttempt {
PersistedAttempt {
id: self.id,
policy_id: self.policy_id,
github_runner_id: self.github_runner_id,
state: self.state,
outcome: self.outcome.clone(),
process_id: self.process_id,
runtime_path: self.runtime_path.clone(),
workspace_kind: self.workspace.kind(),
workspace_slot: self.workspace.slot_number(),
created_at: self.created_at,
terminal_at: self.terminal_at,
last_state_change_at: self.last_state_change_at,
}
}
#[must_use]
pub fn allocate(
id: AttemptId,
policy_id: PolicyId,
runtime_path: impl Into<PathBuf>,
now: Timestamp,
) -> Self {
Self::allocate_in(
id,
policy_id,
runtime_path,
AttemptWorkspace::Ephemeral,
now,
)
}
#[must_use]
pub fn allocate_in(
id: AttemptId,
policy_id: PolicyId,
runtime_path: impl Into<PathBuf>,
workspace: AttemptWorkspace,
now: Timestamp,
) -> Self {
Self {
id,
policy_id,
github_runner_id: None,
state: AttemptState::Allocated,
outcome: None,
process_id: None,
runtime_path: runtime_path.into(),
workspace,
created_at: now,
terminal_at: None,
last_state_change_at: now,
}
}
pub fn from_persisted(fields: PersistedAttempt) -> Result<Self, AttemptError> {
let PersistedAttempt {
id,
policy_id,
github_runner_id,
state,
outcome,
process_id,
runtime_path,
workspace_kind,
workspace_slot,
created_at,
terminal_at,
last_state_change_at,
} = fields;
let workspace = AttemptWorkspace::from_persisted(workspace_kind, workspace_slot)?;
match (&outcome, state.is_terminal()) {
(None, true) => return Err(AttemptError::TerminalWithoutOutcome { state }),
(Some(outcome), false) => {
return Err(AttemptError::NonTerminalWithOutcome {
state,
outcome: outcome.clone(),
});
}
(Some(outcome), true) => {
let expected = outcome.terminal_state();
if state != expected && state != AttemptState::Cleaned {
return Err(AttemptError::OutcomeStateMismatch {
state,
outcome: outcome.clone(),
});
}
}
(None, false) => {}
}
match (terminal_at, state.is_terminal()) {
(None, true) => return Err(AttemptError::TerminalWithoutTimestamp { state }),
(Some(_), false) => return Err(AttemptError::NonTerminalWithTimestamp { state }),
_ => {}
}
if last_state_change_at < created_at {
return Err(AttemptError::TimestampsOutOfOrder {
state,
field: "last_state_change_at",
created_at,
found: last_state_change_at,
});
}
if let Some(terminal_at) = terminal_at
&& terminal_at < created_at
{
return Err(AttemptError::TimestampsOutOfOrder {
state,
field: "terminal_at",
created_at,
found: terminal_at,
});
}
Ok(Self {
id,
policy_id,
github_runner_id,
state,
outcome,
process_id,
runtime_path,
workspace,
created_at,
terminal_at,
last_state_change_at,
})
}
#[must_use]
pub const fn state(&self) -> AttemptState {
self.state
}
#[must_use]
pub const fn outcome(&self) -> Option<&AttemptOutcome> {
self.outcome.as_ref()
}
#[must_use]
pub const fn github_runner_id(&self) -> Option<u64> {
self.github_runner_id
}
#[must_use]
pub const fn process_id(&self) -> Option<u32> {
self.process_id
}
#[must_use]
pub fn runtime_path(&self) -> &Path {
&self.runtime_path
}
#[must_use]
pub const fn workspace(&self) -> AttemptWorkspace {
self.workspace
}
#[must_use]
pub const fn holds_slot_lease(&self) -> bool {
self.workspace.is_persistent() && !matches!(self.state, AttemptState::Cleaned)
}
#[must_use]
pub const fn terminal_at(&self) -> Option<Timestamp> {
self.terminal_at
}
#[must_use]
pub const fn last_state_change_at(&self) -> Timestamp {
self.last_state_change_at
}
#[must_use]
pub const fn is_terminal(&self) -> bool {
self.state.is_terminal()
}
#[must_use]
pub const fn counts_against_capacity(&self) -> bool {
self.state.counts_against_capacity()
}
fn move_to(&mut self, next: AttemptState, now: Timestamp) -> Result<(), AttemptError> {
if !self.state.can_transition_to(next) {
return Err(AttemptError::IllegalTransition {
from: self.state,
to: next,
});
}
self.state = next;
self.last_state_change_at = now;
Ok(())
}
pub fn jit_received(&mut self, now: Timestamp) -> Result<(), AttemptError> {
self.move_to(AttemptState::JitReceived, now)
}
pub fn started(&mut self, process_id: u32, now: Timestamp) -> Result<(), AttemptError> {
self.move_to(AttemptState::Starting, now)?;
self.process_id = Some(process_id);
Ok(())
}
pub fn registered_idle(
&mut self,
github_runner_id: u64,
now: Timestamp,
) -> Result<(), AttemptError> {
self.move_to(AttemptState::Idle, now)?;
self.github_runner_id = Some(github_runner_id);
Ok(())
}
pub fn assigned_job(
&mut self,
github_runner_id: u64,
now: Timestamp,
) -> Result<(), AttemptError> {
self.move_to(AttemptState::Busy, now)?;
self.github_runner_id = Some(github_runner_id);
Ok(())
}
pub fn conclude(
&mut self,
outcome: AttemptOutcome,
now: Timestamp,
) -> Result<(), AttemptError> {
if !outcome.required_from().contains(&self.state) {
return Err(AttemptError::OutcomeUnreachable {
from: self.state,
outcome,
});
}
self.move_to(outcome.terminal_state(), now)?;
self.terminal_at = Some(now);
self.outcome = Some(outcome);
Ok(())
}
pub fn clean(&mut self, now: Timestamp) -> Result<(), AttemptError> {
if self.state == AttemptState::Busy {
return Err(AttemptError::BusyCannotBeCleaned);
}
self.move_to(AttemptState::Cleaned, now)
}
}
pub fn authorize(
agent_host: HostId,
policy: &ScalePolicy,
attempt: &RunnerAttempt,
) -> Result<(), OwnershipError> {
if attempt.policy_id != policy.id {
return Err(OwnershipError::PolicyMismatch {
attempt: attempt.id,
attempt_policy: attempt.policy_id,
policy: policy.id,
});
}
if !policy.is_owned_by(agent_host) {
return Err(OwnershipError::ForeignHost {
policy: policy.id,
owner: policy.host_id,
agent: agent_host,
});
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RecoveryTimeouts {
pub jit_handoff: Elapsed,
pub startup: Elapsed,
pub idle: Elapsed,
}
impl RecoveryTimeouts {
#[must_use]
pub const fn new(jit_handoff: Elapsed, startup: Elapsed, idle: Elapsed) -> Self {
Self {
jit_handoff,
startup,
idle,
}
}
#[must_use]
pub fn provisional() -> Self {
Self {
jit_handoff: Elapsed::seconds(120),
startup: Elapsed::seconds(300),
idle: Elapsed::seconds(300),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GithubRunnerObservation {
Unreachable,
NotRegistered,
Registered { busy: bool },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RecoveryObservation {
pub process_alive: bool,
pub github: GithubRunnerObservation,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RecoveryDecision {
Nothing,
Clean,
Adopt,
Wait,
Defer,
Observe(AttemptState),
Conclude(AttemptOutcome),
Terminate(AttemptOutcome),
}
#[must_use]
pub fn recovery_decision(
attempt: &RunnerAttempt,
observation: RecoveryObservation,
timeouts: RecoveryTimeouts,
clock: &dyn Clock,
) -> RecoveryDecision {
use AttemptState as S;
use GithubRunnerObservation as G;
let state = attempt.state();
if state == S::Cleaned {
return RecoveryDecision::Nothing;
}
if state.is_terminal() {
return RecoveryDecision::Clean;
}
if observation.github == G::Unreachable {
return RecoveryDecision::Defer;
}
let elapsed = clock.now() - attempt.last_state_change_at();
match state {
S::Allocated | S::JitReceived => {
if observation.process_alive {
RecoveryDecision::Adopt
} else {
match observation.github {
G::Registered { .. } => RecoveryDecision::Observe(if state == S::Allocated {
S::JitReceived
} else {
S::Starting
}),
G::NotRegistered => {
if elapsed >= timeouts.jit_handoff {
RecoveryDecision::Conclude(AttemptOutcome::failed(
if state == S::Allocated {
FailureReason::JitRequestFailed
} else {
FailureReason::JitExpired
},
))
} else {
RecoveryDecision::Wait
}
}
G::Unreachable => unreachable!("handled above"),
}
}
}
S::Starting => match observation.github {
G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
G::Registered { busy: false } => RecoveryDecision::Observe(S::Idle),
G::NotRegistered => {
if !observation.process_alive {
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly,
))
} else if elapsed < timeouts.startup {
RecoveryDecision::Adopt
} else {
RecoveryDecision::Terminate(AttemptOutcome::failed(
FailureReason::RegistrationTimedOut,
))
}
}
G::Unreachable => unreachable!("handled above"),
},
S::Idle => match observation.github {
G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
G::Registered { busy: false } => {
if !observation.process_alive {
RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
} else if elapsed >= timeouts.idle {
RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork)
} else {
RecoveryDecision::Adopt
}
}
G::NotRegistered => {
if observation.process_alive {
RecoveryDecision::Adopt
} else if elapsed >= timeouts.idle {
RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork)
} else {
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly,
))
}
}
G::Unreachable => unreachable!("handled above"),
},
S::Busy => {
if observation.process_alive {
RecoveryDecision::Adopt
} else {
RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
}
}
S::Finished | S::Failed | S::Orphaned | S::Cleaned => {
unreachable!("terminal states are handled above")
}
}
}
#[must_use]
pub fn active_count<'a>(attempts: impl IntoIterator<Item = &'a RunnerAttempt>) -> u16 {
attempts
.into_iter()
.filter(|a| a.counts_against_capacity())
.fold(0u16, |acc, _| acc.saturating_add(1))
}
#[must_use]
pub fn active_count_for<'a>(
policy_id: PolicyId,
attempts: impl IntoIterator<Item = &'a RunnerAttempt>,
) -> u16 {
attempts
.into_iter()
.filter(|a| a.policy_id == policy_id && a.counts_against_capacity())
.fold(0u16, |acc, _| acc.saturating_add(1))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{Arch, CachePolicy, HostLabel, Os, ScaleTarget};
use crate::policy::{PolicyMode, RoutingLabels};
use std::num::NonZeroU16;
#[derive(Debug)]
struct StubClock(std::sync::Mutex<Timestamp>);
impl StubClock {
fn at(secs: i64) -> Self {
Self(std::sync::Mutex::new(ts(secs)))
}
fn set(&self, secs: i64) {
*self.0.lock().unwrap() = ts(secs);
}
}
impl Clock for StubClock {
fn now(&self) -> Timestamp {
*self.0.lock().unwrap()
}
}
fn ts(secs: i64) -> Timestamp {
chrono::DateTime::from_timestamp(secs, 0).expect("valid timestamp")
}
fn attempt_in(state: AttemptState, entered_at: i64) -> RunnerAttempt {
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"runtime/p/a",
ts(0),
);
attempt.state = state;
attempt.last_state_change_at = ts(entered_at);
attempt
}
fn attempt_for(policy_id: PolicyId, state: AttemptState, entered_at: i64) -> RunnerAttempt {
let mut attempt = attempt_in(state, entered_at);
attempt.policy_id = policy_id;
attempt
}
fn a_policy(host: HostId, policy_id: PolicyId) -> ScalePolicy {
ScalePolicy::new(
policy_id,
ScaleTarget::repository("o/r").unwrap(),
1,
host,
PolicyMode::autoscale(
RoutingLabels::derive(&HostLabel::new("home").unwrap(), Os::Windows, Arch::X64),
0,
NonZeroU16::new(1).unwrap(),
)
.unwrap(),
CachePolicy::default(),
)
}
fn diagram_edges() -> Vec<(AttemptState, AttemptState)> {
use AttemptState::*;
let mut edges = vec![
(Allocated, JitReceived),
(JitReceived, Starting),
(Starting, Idle),
(Starting, Busy),
];
edges.push((Idle, Busy));
for from in [Allocated, JitReceived, Starting] {
for to in [Failed, Orphaned] {
edges.push((from, to));
}
}
for from in [Idle, Busy] {
for to in [Finished, Failed, Orphaned] {
edges.push((from, to));
}
}
for from in [Finished, Failed, Orphaned] {
edges.push((from, Cleaned));
}
edges
}
#[test]
fn every_attempt_state_transition_is_legal_exactly_where_the_diagram_says() {
let expected = diagram_edges();
assert_eq!(
expected.len(),
20,
"the transcription itself changed; check it against the diagram"
);
let mut legal_seen = 0usize;
let mut illegal_seen = 0usize;
for from in AttemptState::ALL {
for to in AttemptState::ALL {
let expected_legal = expected.contains(&(from, to));
let mut attempt = attempt_in(from, 0);
let result = attempt.move_to(to, ts(10));
if expected_legal {
legal_seen += 1;
assert!(
result.is_ok(),
"{from} -> {to} is in the diagram and must be accepted"
);
assert_eq!(attempt.state(), to);
assert_eq!(attempt.last_state_change_at(), ts(10));
} else {
illegal_seen += 1;
assert!(
matches!(result, Err(AttemptError::IllegalTransition { .. })),
"{from} -> {to} is not in the diagram and must be rejected"
);
assert_eq!(
attempt.state(),
from,
"a refused transition changes nothing"
);
assert_eq!(attempt.last_state_change_at(), ts(0));
}
}
}
assert_eq!(legal_seen, 20);
assert_eq!(illegal_seen, 81 - 20);
let mut published = AttemptState::LEGAL.to_vec();
let mut transcribed = expected;
published.sort_unstable();
transcribed.sort_unstable();
assert_eq!(published, transcribed);
}
#[test]
fn every_live_state_has_a_terminal_edge_so_no_attempt_can_hold_a_slot_forever() {
for state in AttemptState::ALL {
if state.is_terminal() {
continue;
}
assert_eq!(
active_count([&attempt_in(state, 0)]),
1,
"{state} is non-terminal, so the reconciliation formula must \
still be subtracting it"
);
assert!(
state.can_transition_to(AttemptState::Failed)
|| state.can_transition_to(AttemptState::Finished),
"{state} has no way to conclude, so an attempt in it holds a host \
capacity slot permanently"
);
assert!(
state.can_transition_to(AttemptState::Orphaned),
"{state} has no orphan edge, so an attempt found in it after a \
restart cannot be recorded"
);
assert!(
AttemptState::CONCLUDABLE_FROM.contains(&state),
"{state} is live but missing from CONCLUDABLE_FROM"
);
}
assert_eq!(
AttemptState::CONCLUDABLE_FROM.len(),
AttemptState::ALL
.iter()
.filter(|s| !s.is_terminal())
.count(),
"CONCLUDABLE_FROM must list every non-terminal state and nothing else"
);
}
#[test]
fn an_attempt_state_cannot_transition_to_itself() {
for state in AttemptState::ALL {
assert!(
!state.can_transition_to(state),
"{state} -> {state} is not an edge in the diagram"
);
}
}
#[test]
fn cleaned_is_absorbing() {
for to in AttemptState::ALL {
assert!(
!AttemptState::Cleaned.can_transition_to(to),
"cleaned -> {to} must not exist; a cleaned runtime directory is gone"
);
}
}
#[test]
fn the_documented_happy_path_walks_allocated_to_cleaned() {
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"runtime/p/a",
ts(0),
);
assert_eq!(attempt.state(), AttemptState::Allocated);
assert!(attempt.counts_against_capacity());
attempt.jit_received(ts(1)).unwrap();
attempt.started(4242, ts(2)).unwrap();
assert_eq!(attempt.process_id(), Some(4242));
attempt.assigned_job(73, ts(3)).unwrap();
assert_eq!(attempt.state(), AttemptState::Busy);
assert_eq!(attempt.github_runner_id(), Some(73));
attempt
.conclude(AttemptOutcome::CompletedJob, ts(9))
.unwrap();
assert_eq!(attempt.state(), AttemptState::Finished);
assert_eq!(attempt.terminal_at(), Some(ts(9)));
assert!(!attempt.counts_against_capacity());
attempt.clean(ts(10)).unwrap();
assert_eq!(attempt.state(), AttemptState::Cleaned);
}
#[test]
fn an_attempt_that_exits_idle_without_work_is_terminal_and_not_a_failure() {
let mut surplus = RunnerAttempt::allocate(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"runtime/p/a",
ts(0),
);
surplus.jit_received(ts(1)).unwrap();
surplus.started(1, ts(2)).unwrap();
surplus.registered_idle(73, ts(3)).unwrap();
surplus
.conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(300))
.unwrap();
assert!(surplus.is_terminal());
assert_eq!(surplus.state(), AttemptState::Finished);
let outcome = surplus.outcome().expect("a terminal attempt carries one");
assert!(outcome.is_idle_exit());
assert!(
!outcome.is_failure(),
"the surplus runner is an accepted, bounded cost of having no job \
reservation -- presenting it as a fault sends an operator hunting \
something that did not happen"
);
assert!(!outcome.ran_a_job());
surplus.clean(ts(301)).unwrap();
assert_eq!(surplus.state(), AttemptState::Cleaned);
}
#[test]
fn an_idle_exit_is_distinguishable_from_a_failure_and_from_a_completed_job() {
let idle = AttemptOutcome::ExitedIdleWithoutWork;
let failed = AttemptOutcome::failed(FailureReason::ProcessStartFailed);
let done = AttemptOutcome::CompletedJob;
assert_ne!(idle, failed);
assert_ne!(idle, done);
assert_ne!(failed, done);
assert_eq!(idle.terminal_state(), AttemptState::Finished);
assert_eq!(done.terminal_state(), AttemptState::Finished);
assert_eq!(failed.terminal_state(), AttemptState::Failed);
assert_eq!(
AttemptOutcome::Orphaned.terminal_state(),
AttemptState::Orphaned
);
assert_eq!(idle.terminal_state(), done.terminal_state());
assert_ne!(idle, done);
assert!(!idle.is_failure());
assert!(failed.is_failure());
assert!(AttemptOutcome::Orphaned.is_failure());
}
#[test]
fn an_outcome_cannot_be_recorded_from_a_state_that_could_not_produce_it() {
let mut idle = attempt_in(AttemptState::Idle, 0);
assert!(matches!(
idle.conclude(AttemptOutcome::CompletedJob, ts(1)),
Err(AttemptError::OutcomeUnreachable {
from: AttemptState::Idle,
..
})
));
assert_eq!(idle.state(), AttemptState::Idle);
assert!(idle.outcome().is_none());
let mut busy = attempt_in(AttemptState::Busy, 0);
assert!(matches!(
busy.conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(1)),
Err(AttemptError::OutcomeUnreachable {
from: AttemptState::Busy,
..
})
));
for state in AttemptState::CONCLUDABLE_FROM {
attempt_in(*state, 0)
.conclude(AttemptOutcome::failed(FailureReason::JitExpired), ts(1))
.unwrap_or_else(|e| panic!("{state} must be able to fail: {e}"));
attempt_in(*state, 0)
.conclude(AttemptOutcome::Orphaned, ts(1))
.unwrap_or_else(|e| panic!("{state} must be able to orphan: {e}"));
}
for state in AttemptState::ALL.iter().filter(|s| s.is_terminal()) {
assert!(
attempt_in(*state, 0)
.conclude(AttemptOutcome::Orphaned, ts(1))
.is_err(),
"{state} has already concluded"
);
}
}
fn earliest_state_producing(reason: &FailureReason) -> AttemptState {
match reason {
FailureReason::RunnerPackageUnverified => AttemptState::Allocated,
FailureReason::JitRequestFailed => AttemptState::Allocated,
FailureReason::JitExpired => AttemptState::JitReceived,
FailureReason::ProcessStartFailed => AttemptState::JitReceived,
FailureReason::RunnerVersionRejected => AttemptState::Starting,
FailureReason::ProcessExitedUnexpectedly => AttemptState::Starting,
FailureReason::RegistrationTimedOut => AttemptState::Starting,
FailureReason::TerminatedAfterRegistrationTimeout => AttemptState::Starting,
FailureReason::Other(_) => AttemptState::Busy,
}
}
#[test]
fn all_failure_reasons_are_reachable_from_the_state_that_produces_them() {
let cases: [(FailureReason, AttemptState); 9] = [
(
FailureReason::RunnerPackageUnverified,
AttemptState::Allocated,
),
(FailureReason::JitRequestFailed, AttemptState::Allocated),
(FailureReason::JitExpired, AttemptState::JitReceived),
(FailureReason::ProcessStartFailed, AttemptState::JitReceived),
(FailureReason::RunnerVersionRejected, AttemptState::Starting),
(
FailureReason::ProcessExitedUnexpectedly,
AttemptState::Starting,
),
(FailureReason::RegistrationTimedOut, AttemptState::Starting),
(
FailureReason::TerminatedAfterRegistrationTimeout,
AttemptState::Starting,
),
(
FailureReason::Other("a reason b1 did not anticipate".into()),
AttemptState::Busy,
),
];
assert_eq!(
cases.len(),
FailureReason::ALL.len(),
"every FailureReason variant needs a state it is reachable from"
);
for listed in FailureReason::ALL {
assert!(
cases.iter().any(|(reason, _)| {
std::mem::discriminant(reason) == std::mem::discriminant(&listed)
}),
"{listed:?} is in FailureReason::ALL but has no case here"
);
}
for (reason, from) in &cases {
assert_eq!(
earliest_state_producing(reason),
*from,
"{reason:?}: the table and the exhaustive match disagree"
);
}
for (reason, from) in cases {
let mut attempt = attempt_in(from, 0);
attempt
.conclude(AttemptOutcome::failed(reason.clone()), ts(5))
.unwrap_or_else(|e| panic!("{reason:?} must be recordable from {from}, got {e}"));
assert_eq!(attempt.state(), AttemptState::Failed);
assert_eq!(attempt.terminal_at(), Some(ts(5)));
assert_eq!(
attempt.outcome(),
Some(&AttemptOutcome::failed(reason.clone()))
);
assert!(
!attempt.counts_against_capacity(),
"{reason:?} from {from} must give the host capacity slot back; \
that is what the amendment was for"
);
let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
.unwrap_or_else(|e| panic!("{reason:?} from {from} must reload: {e}"));
assert_eq!(restored, attempt);
}
}
#[test]
fn a_busy_attempt_cannot_be_cleaned() {
let mut busy = attempt_in(AttemptState::Busy, 0);
let err = busy.clean(ts(1)).unwrap_err();
assert_eq!(
err,
AttemptError::BusyCannotBeCleaned,
"the refusal must be named, not a generic transition error, so that a \
scale-down that tried it is legible in a log"
);
assert_eq!(busy.state(), AttemptState::Busy);
assert!(busy.counts_against_capacity());
}
#[test]
fn only_a_terminal_attempt_can_be_cleaned() {
for state in AttemptState::ALL {
let mut attempt = attempt_in(state, 0);
let result = attempt.clean(ts(1));
if state.is_concluded() {
assert!(result.is_ok(), "{state} is terminal and must be cleanable");
assert_eq!(attempt.state(), AttemptState::Cleaned);
} else {
assert!(
result.is_err(),
"{state} is not terminal and must not be cleanable"
);
assert_eq!(attempt.state(), state);
}
}
}
#[test]
fn capacity_is_reclaimed_exactly_at_the_terminal_states() {
for state in AttemptState::ALL {
assert_eq!(
attempt_in(state, 0).counts_against_capacity(),
!state.is_terminal(),
"{state}"
);
}
let attempts = vec![
attempt_in(AttemptState::Allocated, 0),
attempt_in(AttemptState::Starting, 0),
attempt_in(AttemptState::Busy, 0),
attempt_in(AttemptState::Finished, 0),
attempt_in(AttemptState::Cleaned, 0),
];
assert_eq!(active_count(&attempts), 3);
}
#[test]
fn the_per_policy_active_count_is_not_the_host_wide_one() {
let mine = PolicyId::from_u128(1);
let theirs = PolicyId::from_u128(2);
let mut attempts = vec![
attempt_for(mine, AttemptState::Starting, 0),
attempt_for(mine, AttemptState::Busy, 0),
attempt_for(theirs, AttemptState::Busy, 0),
attempt_for(theirs, AttemptState::Idle, 0),
attempt_for(theirs, AttemptState::Allocated, 0),
];
assert_eq!(active_count(&attempts), 5, "the host-wide total, for D9");
assert_eq!(active_count_for(mine, &attempts), 2, "this policy, for D7");
assert_eq!(active_count_for(theirs, &attempts), 3);
assert_eq!(
active_count_for(PolicyId::from_u128(3), &attempts),
0,
"a policy with nothing in flight"
);
attempts.push(attempt_for(mine, AttemptState::Finished, 0));
attempts.push(attempt_for(mine, AttemptState::Cleaned, 0));
assert_eq!(active_count_for(mine, &attempts), 2);
}
fn slot(n: u16) -> AttemptWorkspace {
AttemptWorkspace::persistent_slot(std::num::NonZeroU16::new(n).expect("a positive slot"))
}
fn terminal_outcome(state: AttemptState) -> AttemptOutcome {
match state {
AttemptState::Failed => {
AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
}
AttemptState::Orphaned => AttemptOutcome::Orphaned,
_ => AttemptOutcome::CompletedJob,
}
}
#[test]
fn the_ordinary_constructor_still_allocates_a_disposable_workspace() {
let attempt = RunnerAttempt::allocate(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"runtime/p/a",
ts(0),
);
assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
assert_eq!(attempt.workspace().slot(), None);
assert!(!attempt.holds_slot_lease());
assert_eq!(
attempt.to_persisted().workspace_kind,
WorkspaceKind::Ephemeral
);
assert_eq!(attempt.to_persisted().workspace_slot, None);
}
#[test]
fn a_persistent_attempt_journals_the_slot_it_leased() {
let attempt = RunnerAttempt::allocate_in(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"/srv/rman/acme/s2",
slot(2),
ts(0),
);
assert_eq!(attempt.workspace(), slot(2));
assert_eq!(attempt.workspace().slot_number(), Some(2));
assert_eq!(
attempt.workspace().slot_directory_name().as_deref(),
Some("s2")
);
assert_eq!(attempt.runtime_path(), Path::new("/srv/rman/acme/s2"));
assert!(attempt.holds_slot_lease());
}
#[test]
fn the_workspace_kind_and_slot_do_not_change_after_allocation() {
let mut attempt = RunnerAttempt::allocate_in(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"/srv/rman/acme/s1",
slot(1),
ts(0),
);
attempt
.jit_received(ts(1))
.expect("allocated -> jit_received");
assert_eq!(attempt.workspace(), slot(1));
attempt
.started(4242, ts(2))
.expect("jit_received -> starting");
assert_eq!(attempt.workspace(), slot(1));
attempt
.registered_idle(73, ts(3))
.expect("starting -> idle");
assert_eq!(attempt.workspace(), slot(1));
attempt.assigned_job(73, ts(4)).expect("idle -> busy");
assert_eq!(attempt.workspace(), slot(1));
attempt
.conclude(AttemptOutcome::CompletedJob, ts(5))
.expect("busy -> finished");
assert_eq!(attempt.workspace(), slot(1));
assert!(
attempt.holds_slot_lease(),
"a terminal attempt still holds its slot until cleanup succeeds"
);
attempt.clean(ts(6)).expect("finished -> cleaned");
assert_eq!(attempt.workspace(), slot(1));
assert!(
!attempt.holds_slot_lease(),
"a cleaned attempt releases the slot for reuse"
);
}
#[test]
fn an_uncleaned_terminal_attempt_keeps_its_lease() {
for state in [
AttemptState::Finished,
AttemptState::Failed,
AttemptState::Orphaned,
] {
let mut fields = row(state, Some(terminal_outcome(state)));
fields.workspace_kind = WorkspaceKind::Persistent;
fields.workspace_slot = Some(3);
let attempt = RunnerAttempt::from_persisted(fields).expect("a row the domain accepts");
assert!(attempt.holds_slot_lease(), "state {state}");
}
}
#[test]
fn a_workspace_allocation_round_trips_through_the_journal() {
for workspace in [AttemptWorkspace::Ephemeral, slot(1), slot(u16::MAX)] {
let attempt = RunnerAttempt::allocate_in(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"runtime/p/a",
workspace,
ts(0),
);
let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
.expect("a row this crate wrote must load");
assert_eq!(restored, attempt);
assert_eq!(restored.workspace(), workspace);
}
}
#[test]
fn a_journal_row_whose_workspace_columns_disagree_is_rejected() {
let mut persistent_without_slot = row(AttemptState::Allocated, None);
persistent_without_slot.workspace_kind = WorkspaceKind::Persistent;
assert_eq!(
RunnerAttempt::from_persisted(persistent_without_slot),
Err(AttemptError::Workspace(
WorkspaceError::PersistentWithoutSlot
))
);
let mut zero_slot = row(AttemptState::Allocated, None);
zero_slot.workspace_kind = WorkspaceKind::Persistent;
zero_slot.workspace_slot = Some(0);
assert_eq!(
RunnerAttempt::from_persisted(zero_slot),
Err(AttemptError::Workspace(WorkspaceError::SlotNotPositive))
);
let mut ephemeral_with_slot = row(AttemptState::Allocated, None);
ephemeral_with_slot.workspace_slot = Some(1);
assert_eq!(
RunnerAttempt::from_persisted(ephemeral_with_slot),
Err(AttemptError::Workspace(WorkspaceError::EphemeralWithSlot {
slot: 1
}))
);
}
#[test]
fn an_attempt_serialises_its_workspace_without_credentials() {
let attempt = RunnerAttempt::allocate_in(
AttemptId::from_u128(1),
PolicyId::from_u128(1),
"/srv/rman/acme/s2",
slot(2),
ts(0),
);
let encoded = serde_json::to_string(&attempt).expect("serialisable");
let decoded: RunnerAttempt = serde_json::from_str(&encoded).expect("deserialisable");
assert_eq!(decoded, attempt);
for needle in ["token", "secret", "jitconfig", "password"] {
assert!(
!encoded.to_ascii_lowercase().contains(needle),
"an attempt leaked {needle:?}: {encoded}"
);
}
}
fn row(state: AttemptState, outcome: Option<AttemptOutcome>) -> PersistedAttempt {
PersistedAttempt {
id: AttemptId::from_u128(1),
policy_id: PolicyId::from_u128(1),
github_runner_id: Some(73),
state,
outcome,
process_id: Some(9),
runtime_path: "runtime/p/a".into(),
workspace_kind: WorkspaceKind::Ephemeral,
workspace_slot: None,
created_at: ts(0),
terminal_at: state.is_terminal().then(|| ts(9)),
last_state_change_at: ts(9),
}
}
#[test]
fn a_hand_edited_journal_row_with_an_impossible_outcome_is_rejected() {
assert!(
RunnerAttempt::from_persisted(row(
AttemptState::Finished,
Some(AttemptOutcome::ExitedIdleWithoutWork)
))
.is_ok()
);
assert!(matches!(
RunnerAttempt::from_persisted(row(AttemptState::Failed, None)),
Err(AttemptError::TerminalWithoutOutcome { .. })
));
assert!(matches!(
RunnerAttempt::from_persisted(row(
AttemptState::Busy,
Some(AttemptOutcome::CompletedJob)
)),
Err(AttemptError::NonTerminalWithOutcome { .. })
));
assert!(matches!(
RunnerAttempt::from_persisted(row(
AttemptState::Failed,
Some(AttemptOutcome::CompletedJob)
)),
Err(AttemptError::OutcomeStateMismatch { .. })
));
assert!(
RunnerAttempt::from_persisted(row(
AttemptState::Cleaned,
Some(AttemptOutcome::Orphaned)
))
.is_ok()
);
}
#[test]
fn a_hand_edited_journal_row_with_an_impossible_terminal_at_is_rejected() {
let mut terminal_without = row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
terminal_without.terminal_at = None;
assert!(
matches!(
RunnerAttempt::from_persisted(terminal_without),
Err(AttemptError::TerminalWithoutTimestamp {
state: AttemptState::Finished
})
),
"a terminal row with no terminal_at must be refused, exactly as a \
terminal row with no outcome is"
);
let mut live_with = row(AttemptState::Busy, None);
live_with.terminal_at = Some(ts(9));
assert!(matches!(
RunnerAttempt::from_persisted(live_with),
Err(AttemptError::NonTerminalWithTimestamp {
state: AttemptState::Busy
})
));
let mut cleaned = row(AttemptState::Cleaned, Some(AttemptOutcome::Orphaned));
cleaned.terminal_at = Some(ts(4));
assert!(RunnerAttempt::from_persisted(cleaned).is_ok());
}
#[test]
fn a_hand_edited_journal_row_whose_timestamps_run_backwards_is_rejected() {
let mut concluded_before_created =
row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
concluded_before_created.created_at = ts(100);
concluded_before_created.last_state_change_at = ts(100);
concluded_before_created.terminal_at = Some(ts(0));
assert_eq!(
RunnerAttempt::from_persisted(concluded_before_created),
Err(AttemptError::TimestampsOutOfOrder {
state: AttemptState::Finished,
field: "terminal_at",
created_at: ts(100),
found: ts(0),
})
);
let mut changed_before_created = row(AttemptState::Busy, None);
changed_before_created.created_at = ts(100);
changed_before_created.last_state_change_at = ts(0);
assert_eq!(
RunnerAttempt::from_persisted(changed_before_created),
Err(AttemptError::TimestampsOutOfOrder {
state: AttemptState::Busy,
field: "last_state_change_at",
created_at: ts(100),
found: ts(0),
})
);
let mut same_instant = row(AttemptState::Failed, Some(AttemptOutcome::Orphaned));
same_instant.outcome = Some(AttemptOutcome::failed(FailureReason::JitRequestFailed));
same_instant.created_at = ts(7);
same_instant.last_state_change_at = ts(7);
same_instant.terminal_at = Some(ts(7));
assert!(RunnerAttempt::from_persisted(same_instant).is_ok());
assert!(
RunnerAttempt::from_persisted(row(
AttemptState::Finished,
Some(AttemptOutcome::CompletedJob)
))
.is_ok()
);
}
#[test]
fn an_attempt_round_trips_through_its_persisted_form() {
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(3),
PolicyId::from_u128(4),
"runtime/p/a",
ts(0),
);
attempt.jit_received(ts(1)).unwrap();
attempt.started(7, ts(2)).unwrap();
attempt.registered_idle(7, ts(3)).unwrap();
attempt
.conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(4))
.unwrap();
let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
.expect("a row this crate produced must load");
assert_eq!(restored, attempt);
assert_eq!(restored.terminal_at(), Some(ts(4)));
assert_ne!(
restored.last_state_change_at(),
restored.created_at,
"the two timestamps must not collapse onto each other; that \
transposition is what PersistedAttempt exists to prevent"
);
}
#[test]
fn an_attempt_round_trips_through_serde() {
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(3),
PolicyId::from_u128(4),
"runtime/p/a",
ts(0),
);
attempt.jit_received(ts(1)).unwrap();
attempt.started(7, ts(2)).unwrap();
attempt.registered_idle(73, ts(3)).unwrap();
attempt
.conclude(
AttemptOutcome::failed(FailureReason::Other("no detail".into())),
ts(4),
)
.unwrap();
let json = serde_json::to_string(&attempt).unwrap();
let back: RunnerAttempt = serde_json::from_str(&json).unwrap();
assert_eq!(attempt, back);
}
#[test]
fn an_attempt_belonging_to_another_host_is_rejected() {
let mine = HostId::from_u128(7);
let theirs = HostId::from_u128(8);
let policy_id = PolicyId::from_u128(11);
let their_policy = a_policy(theirs, policy_id);
let attempt =
RunnerAttempt::allocate(AttemptId::from_u128(1), policy_id, "runtime/p/a", ts(0));
let err = authorize(mine, &their_policy, &attempt).unwrap_err();
assert!(
matches!(
err,
OwnershipError::ForeignHost {
owner,
agent,
..
} if owner == theirs && agent == mine
),
"got {err:?}"
);
let my_policy = a_policy(mine, policy_id);
assert!(authorize(mine, &my_policy, &attempt).is_ok());
}
#[test]
fn an_attempt_checked_against_the_wrong_policy_is_rejected() {
let host = HostId::from_u128(7);
let policy = a_policy(host, PolicyId::from_u128(11));
let attempt = RunnerAttempt::allocate(
AttemptId::from_u128(1),
PolicyId::from_u128(12),
"runtime/p/a",
ts(0),
);
assert!(matches!(
authorize(host, &policy, &attempt),
Err(OwnershipError::PolicyMismatch { .. })
));
}
#[test]
fn recovery_never_reads_the_system_clock() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let attempt = attempt_in(AttemptState::Idle, 1_000);
let observation = RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::NotRegistered,
};
let clock = StubClock::at(1_000 + 299);
assert_eq!(
recovery_decision(&attempt, observation, timeouts, &clock),
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly
)),
"one second before its idle timeout, a vanished runner crashed"
);
clock.set(1_000 + 300);
assert_eq!(
recovery_decision(&attempt, observation, timeouts, &clock),
RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
"at its idle timeout, the same runner is the surplus case from flow 2.7"
);
}
#[test]
fn a_live_process_is_adopted_rather_than_duplicated() {
let timeouts = RecoveryTimeouts::provisional();
let clock = StubClock::at(1_000_000);
for state in [
AttemptState::Allocated,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
] {
let attempt = attempt_in(state, 1_000_000);
assert_eq!(
recovery_decision(
&attempt,
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::NotRegistered,
},
timeouts,
&clock,
),
RecoveryDecision::Adopt,
"{state} with a live process"
);
}
}
#[test]
fn a_dead_busy_attempt_is_orphaned_rather_than_guessed_into_a_success() {
let clock = StubClock::at(1_000_000);
for github in [
GithubRunnerObservation::NotRegistered,
GithubRunnerObservation::Registered { busy: true },
GithubRunnerObservation::Registered { busy: false },
] {
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Busy, 0),
RecoveryObservation {
process_alive: false,
github,
},
RecoveryTimeouts::provisional(),
&clock,
),
RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
"{github:?}"
);
}
}
#[test]
fn an_unreachable_github_defers_every_decision() {
let clock = StubClock::at(1_000_000);
for state in [
AttemptState::Allocated,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
] {
assert_eq!(
recovery_decision(
&attempt_in(state, 0),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::Unreachable,
},
RecoveryTimeouts::provisional(),
&clock,
),
RecoveryDecision::Defer,
"{state} while GitHub is unreachable"
);
}
}
#[test]
fn a_starting_attempt_follows_what_github_reports() {
let clock = StubClock::at(1_000);
let attempt = attempt_in(AttemptState::Starting, 0);
assert_eq!(
recovery_decision(
&attempt,
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: true },
},
RecoveryTimeouts::provisional(),
&clock,
),
RecoveryDecision::Observe(AttemptState::Busy)
);
assert_eq!(
recovery_decision(
&attempt,
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: false },
},
RecoveryTimeouts::provisional(),
&clock,
),
RecoveryDecision::Observe(AttemptState::Idle)
);
}
#[test]
fn a_pre_registration_attempt_believes_github_over_its_own_stale_journal() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let clock = StubClock::at(10_000);
for busy in [true, false] {
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::JitReceived, 0),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::Registered { busy },
},
timeouts,
&clock,
),
RecoveryDecision::Observe(AttemptState::Starting),
"jit_received + Registered{{busy:{busy}}}: the runner cannot have \
registered without starting, and `jit_received -> starting` is \
an edge the diagram already has"
);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Allocated, 0),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::Registered { busy },
},
timeouts,
&clock,
),
RecoveryDecision::Observe(AttemptState::JitReceived),
"allocated + Registered{{busy:{busy}}}"
);
}
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::JitReceived, 0),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::NotRegistered,
},
timeouts,
&clock,
),
RecoveryDecision::Conclude(AttemptOutcome::failed(FailureReason::JitExpired))
);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::JitReceived, 0),
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: false },
},
timeouts,
&clock,
),
RecoveryDecision::Adopt,
"precedence rule 3's other half: local process state is \
authoritative for a child process this agent owns"
);
}
#[test]
fn a_terminal_attempt_is_cleaned_and_a_cleaned_one_is_left_alone() {
let clock = StubClock::at(1_000_000);
let observation = RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::NotRegistered,
};
for state in [
AttemptState::Finished,
AttemptState::Failed,
AttemptState::Orphaned,
] {
assert_eq!(
recovery_decision(
&attempt_in(state, 0),
observation,
RecoveryTimeouts::provisional(),
&clock
),
RecoveryDecision::Clean,
"{state}"
);
}
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Cleaned, 0),
observation,
RecoveryTimeouts::provisional(),
&clock
),
RecoveryDecision::Nothing
);
}
#[test]
fn a_pre_registration_attempt_waits_until_its_deadline_then_concludes() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let clock = StubClock::at(1_059);
let observation = RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::NotRegistered,
};
for (state, reason) in [
(AttemptState::Allocated, FailureReason::JitRequestFailed),
(AttemptState::JitReceived, FailureReason::JitExpired),
] {
let attempt = attempt_in(state, 1_000);
assert_eq!(
recovery_decision(&attempt, observation, timeouts, &clock),
RecoveryDecision::Wait,
"{state} inside its handoff window"
);
clock.set(1_060);
assert_eq!(
recovery_decision(&attempt, observation, timeouts, &clock),
RecoveryDecision::Conclude(AttemptOutcome::failed(reason)),
"{state} past its handoff window concludes rather than stranding \
the attempt"
);
clock.set(1_059);
}
clock.set(1_121);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Starting, 1_000),
observation,
timeouts,
&clock
),
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly
))
);
for (state, decision) in [
(
AttemptState::Allocated,
AttemptOutcome::failed(FailureReason::JitRequestFailed),
),
(
AttemptState::JitReceived,
AttemptOutcome::failed(FailureReason::JitExpired),
),
(
AttemptState::Starting,
AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
),
] {
let mut attempt = attempt_in(state, 1_000);
attempt
.conclude(decision, ts(1_200))
.unwrap_or_else(|e| panic!("recovery decided {state} concludes, but: {e}"));
assert!(!attempt.counts_against_capacity());
}
}
#[test]
fn an_idle_runner_that_github_reports_as_busy_is_recorded_as_busy() {
let timeouts = RecoveryTimeouts::provisional();
let clock = StubClock::at(1_121);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::Registered { busy: true },
},
timeouts,
&clock,
),
RecoveryDecision::Observe(AttemptState::Busy)
);
let mut attempt = attempt_in(AttemptState::Idle, 1_000);
attempt.assigned_job(73, ts(1_200)).unwrap();
assert_eq!(attempt.state(), AttemptState::Busy);
assert_eq!(attempt.github_runner_id(), Some(73));
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: true },
},
timeouts,
&clock,
),
RecoveryDecision::Observe(AttemptState::Busy),
"a live process must not hide a job GitHub is reporting"
);
}
#[test]
fn a_restart_during_a_job_is_recorded_as_busy_and_not_left_reading_idle() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let clock = StubClock::at(1_000);
let mid_job = RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: true },
};
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
mid_job,
timeouts,
&clock
),
recovery_decision(
&attempt_in(AttemptState::Starting, 1_000),
mid_job,
timeouts,
&clock
),
"the same observation must not resolve one way at idle and the \
opposite way one state earlier"
);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
mid_job,
timeouts,
&clock
),
RecoveryDecision::Observe(AttemptState::Busy)
);
let mut attempt = attempt_in(AttemptState::Idle, 1_000);
attempt.assigned_job(73, ts(1_000)).expect("idle -> busy");
assert_eq!(attempt.last_state_change_at(), ts(1_000));
clock.set(1_600);
let crashed = RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::NotRegistered,
};
let decision = recovery_decision(&attempt, crashed, timeouts, &clock);
assert_eq!(
decision,
RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
"a crash during a job is a lost supervision, never the surplus exit"
);
let RecoveryDecision::Conclude(outcome) = decision else {
unreachable!("asserted just above")
};
assert!(
outcome.is_failure() && !outcome.is_idle_exit(),
"`g2` reads these two flags to decide whether to alarm an operator, \
and a mid-job crash must alarm one"
);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
crashed,
timeouts,
&clock
),
RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
"which is exactly why the journal must not be left saying `idle`"
);
}
#[test]
fn a_registered_runner_that_never_gets_a_job_is_stopped_at_its_idle_timeout() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let clock = StubClock::at(1_000);
let attempt = attempt_in(AttemptState::Idle, 1_000);
let idle_registered = |alive| RecoveryObservation {
process_alive: alive,
github: GithubRunnerObservation::Registered { busy: false },
};
clock.set(1_299);
assert_eq!(
recovery_decision(&attempt, idle_registered(true), timeouts, &clock),
RecoveryDecision::Adopt,
"a runner one second inside its idle timeout may still be given a job"
);
clock.set(1_300);
let decision = recovery_decision(&attempt, idle_registered(true), timeouts, &clock);
assert_eq!(
decision,
RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork),
"past the idle timeout a registered, unassigned runner is flow 2.7's surplus case"
);
let mut attempt = attempt;
assert!(attempt.counts_against_capacity());
let RecoveryDecision::Terminate(outcome) = decision else {
unreachable!("asserted above")
};
assert!(
outcome.is_idle_exit(),
"the surplus exit is a normal outcome, and `g2` renders it as one -- an operator \
sent to hunt a fault here would find nothing"
);
attempt
.conclude(outcome, ts(1_310))
.expect("the payload must be applicable to the attempt it was made about");
assert!(!attempt.counts_against_capacity());
clock.set(1_400);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
idle_registered(false),
timeouts,
&clock
),
RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
"the idle deadline governs a live runner; a dead one is orphaned however long it sat"
);
for offset in [0, 299, 300, 100_000] {
clock.set(1_000 + offset);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 1_000),
RecoveryObservation {
process_alive: true,
github: GithubRunnerObservation::Registered { busy: true },
},
timeouts,
&clock
),
RecoveryDecision::Observe(AttemptState::Busy),
"at +{offset}s a runner that took a job was stopped by an idle deadline"
);
}
}
#[test]
fn a_live_unregistered_runner_past_its_deadline_is_stopped_before_its_slot_returns() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let clock = StubClock::at(1_000);
let attempt = attempt_in(AttemptState::Starting, 1_000);
let unregistered = |alive| RecoveryObservation {
process_alive: alive,
github: GithubRunnerObservation::NotRegistered,
};
clock.set(1_119);
assert_eq!(
recovery_decision(&attempt, unregistered(true), timeouts, &clock),
RecoveryDecision::Adopt
);
assert_eq!(
recovery_decision(&attempt, unregistered(false), timeouts, &clock),
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly
)),
"the one case that really is flow 2's runner exit"
);
clock.set(1_120);
let decision = recovery_decision(&attempt, unregistered(true), timeouts, &clock);
assert_eq!(
decision,
RecoveryDecision::Terminate(AttemptOutcome::failed(
FailureReason::RegistrationTimedOut
))
);
assert_ne!(
decision,
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly
)),
"a process visible in a task manager did not exit unexpectedly, and \
an operator told that it did stops believing the next message too"
);
let mut attempt = attempt;
assert!(attempt.counts_against_capacity());
assert_eq!(active_count([&attempt]), 1);
let RecoveryDecision::Terminate(outcome) = decision else {
unreachable!("asserted above")
};
attempt
.conclude(outcome, ts(1_130))
.expect("the payload must be applicable to the attempt it was made about");
assert_eq!(attempt.state(), AttemptState::Failed);
assert!(!attempt.counts_against_capacity());
assert_eq!(active_count([&attempt]), 0);
let pending = attempt_in(AttemptState::Starting, 1_000);
assert_eq!(
recovery_decision(&pending, unregistered(true), timeouts, &clock),
RecoveryDecision::Terminate(AttemptOutcome::failed(
FailureReason::RegistrationTimedOut
))
);
let crashed_on_its_own = unregistered(false);
let killed_by_us_then_lost = unregistered(false);
assert_eq!(
crashed_on_its_own, killed_by_us_then_lost,
"this equality *is* the defect: the fact that separates the two \
cases is not an input to `recovery_decision`, which is why it \
cannot be repaired inside it"
);
assert_eq!(
recovery_decision(&pending, killed_by_us_then_lost, timeouts, &clock),
RecoveryDecision::Conclude(AttemptOutcome::failed(
FailureReason::ProcessExitedUnexpectedly
)),
"pinned as the current behaviour of a known-wrong window, not as a \
correct answer: `e3` closes it by journalling terminate-intent \
before signalling and concluding the marked attempt with \
`TerminatedAfterRegistrationTimeout`, which is true of the dead \
process this pass is looking at. `RecoveryDecision::Terminate` \
names the trade and the owner"
);
}
#[test]
fn no_decision_calls_a_dead_process_live() {
let timeouts = RecoveryTimeouts::new(
Elapsed::seconds(60),
Elapsed::seconds(120),
Elapsed::seconds(300),
);
let entered_at = 1_000i64;
let offsets = [0, 59, 60, 61, 119, 120, 121, 299, 300, 301, 100_000];
let observations = [
GithubRunnerObservation::NotRegistered,
GithubRunnerObservation::Registered { busy: false },
GithubRunnerObservation::Registered { busy: true },
GithubRunnerObservation::Unreachable,
];
let mut agent_termination_derived_here = false;
for state in AttemptState::ALL {
for github in observations {
for process_alive in [true, false] {
for offset in offsets {
let clock = StubClock::at(entered_at + offset);
let observation = RecoveryObservation {
process_alive,
github,
};
let decision = recovery_decision(
&attempt_in(state, entered_at),
observation,
timeouts,
&clock,
);
let reason = match &decision {
RecoveryDecision::Conclude(AttemptOutcome::Failed { reason })
| RecoveryDecision::Terminate(AttemptOutcome::Failed { reason }) => {
Some(reason)
}
_ => None,
};
match reason {
Some(FailureReason::RegistrationTimedOut) => assert!(
process_alive,
"{state} at +{offset}s with {github:?} and a process the \
observation says is gone was reported as still running"
),
Some(FailureReason::ProcessExitedUnexpectedly) => assert!(
!process_alive,
"{state} at +{offset}s with {github:?} reported an \
unexpected exit for a process that is alive"
),
Some(FailureReason::TerminatedAfterRegistrationTimeout) => {
agent_termination_derived_here = true;
assert!(
!process_alive,
"{state} at +{offset}s with {github:?} said the agent had \
stopped a process the observation reports as alive"
);
}
_ => {}
}
}
}
}
}
assert!(
!agent_termination_derived_here,
"`recovery_decision` derived `TerminatedAfterRegistrationTimeout`, \
which is `e3`'s to record from journalled terminate-intent and not \
this function's to infer from an observation that cannot know"
);
}
#[test]
fn a_terminated_runner_is_never_described_as_running() {
let terminated = FailureReason::TerminatedAfterRegistrationTimeout.to_string();
assert!(
!terminated.contains("running"),
"this reason is only ever recorded about a process the agent has \
already stopped: {terminated}"
);
assert!(
!terminated.contains("exited unexpectedly"),
"and nothing exited unexpectedly either -- the agent stopped it on \
purpose: {terminated}"
);
assert!(
terminated.contains("stopped") && terminated.contains("register"),
"it has to say both what happened to the process and why, or it \
sends an operator to a crash investigation: {terminated}"
);
assert_ne!(
terminated,
FailureReason::RegistrationTimedOut.to_string(),
"the live and the stopped case must read differently"
);
assert_ne!(
terminated,
FailureReason::ProcessExitedUnexpectedly.to_string()
);
assert!(
!matches!(
FailureReason::TerminatedAfterRegistrationTimeout,
FailureReason::Other(_)
),
"a known, named condition must not travel through the escape hatch"
);
}
#[test]
fn the_two_starting_failures_read_differently_to_an_operator() {
let exited = FailureReason::ProcessExitedUnexpectedly.to_string();
let timed_out = FailureReason::RegistrationTimedOut.to_string();
assert_ne!(exited, timed_out);
assert!(
timed_out.contains("running") && timed_out.contains("register"),
"the message must say the process is up and unregistered: {timed_out}"
);
assert!(
!timed_out.contains("exited"),
"the whole point is that nothing exited: {timed_out}"
);
assert!(
!matches!(FailureReason::RegistrationTimedOut, FailureReason::Other(_)),
"a known, named condition must not travel through the escape hatch"
);
}
#[test]
fn a_dead_idle_runner_still_registered_at_github_is_orphaned() {
let clock = StubClock::at(1_000_000);
assert_eq!(
recovery_decision(
&attempt_in(AttemptState::Idle, 0),
RecoveryObservation {
process_alive: false,
github: GithubRunnerObservation::Registered { busy: false },
},
RecoveryTimeouts::provisional(),
&clock,
),
RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
"the remote registration outlived our process and needs removing"
);
}
}