runner_manager_domain/attempt.rs
1// owner: b1-domain-core
2
3//! One runner attempt: its lifecycle, its outcome, who may act on it, and what
4//! to do about it after a restart.
5//!
6//! Two things here are easy to get subtly wrong, so they are stated before the
7//! code:
8//!
9//! **The idle exit is a normal outcome, not a failure.** Because there is no
10//! `AcquireJobs` equivalent on the REST path, nothing reserves a queued job for
11//! this host. A runner may start, find that another host took the work, and exit
12//! on its idle timeout (`03-control-flows.md`, flow 2.7). That attempt is
13//! terminal and is cleaned like any other, but [`AttemptOutcome`] records *which*
14//! terminal thing happened, because `g2` must render it distinctly from a
15//! failure and can only do so if the domain wrote the distinction down. Showing a
16//! normal surplus exit as an error sends an operator hunting a fault that does
17//! not exist.
18//!
19//! **A `busy` attempt is never cleaned to free capacity.** Scale-down reclaims a
20//! slot when an attempt reaches a terminal state and at no other time
21//! (`04-subsystem-contracts.md`: "`busy` cannot transition to cleanup due to a
22//! scale-down request"). [`RunnerAttempt::clean`] refuses it by name rather than
23//! by a generic transition error, so the refusal is legible in a log.
24
25use std::fmt;
26use std::path::{Path, PathBuf};
27
28use serde::{Deserialize, Serialize};
29
30use crate::execution::{AttemptExecution, ExecutionError};
31use crate::model::{AttemptId, Clock, Elapsed, HostId, PolicyId, Timestamp};
32use crate::policy::ScalePolicy;
33use crate::workspace::{AttemptWorkspace, WorkspaceError, WorkspaceKind};
34
35// ---------------------------------------------------------------------------
36// Errors
37// ---------------------------------------------------------------------------
38
39#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
40pub enum AttemptError {
41 #[error(transparent)]
42 Execution(#[from] ExecutionError),
43 #[error("{to} is not a legal transition from {from}")]
44 IllegalTransition {
45 from: AttemptState,
46 to: AttemptState,
47 },
48
49 /// A stored workspace kind and slot that cannot describe one allocation, so
50 /// the legal cleanup algorithm for the journalled path is undecidable.
51 #[error(transparent)]
52 Workspace(#[from] WorkspaceError),
53
54 #[error(
55 "a busy attempt must not be cleaned; capacity is reclaimed only when an \
56 attempt reaches a terminal state, never by stopping a runner that is \
57 executing a job"
58 )]
59 BusyCannotBeCleaned,
60
61 #[error("the outcome {outcome:?} cannot be reached from {from}")]
62 OutcomeUnreachable {
63 from: AttemptState,
64 outcome: AttemptOutcome,
65 },
66
67 #[error("{state} is a terminal state and requires an outcome, but none was recorded")]
68 TerminalWithoutOutcome { state: AttemptState },
69
70 #[error("{state} is not terminal, so it must not carry the outcome {outcome:?}")]
71 NonTerminalWithOutcome {
72 state: AttemptState,
73 outcome: AttemptOutcome,
74 },
75
76 #[error("the recorded outcome {outcome:?} does not belong to the recorded state {state}")]
77 OutcomeStateMismatch {
78 state: AttemptState,
79 outcome: AttemptOutcome,
80 },
81
82 #[error("{state} is a terminal state and requires a terminal_at, but none was recorded")]
83 TerminalWithoutTimestamp { state: AttemptState },
84
85 #[error("{state} is not terminal, so it must not carry a terminal_at")]
86 NonTerminalWithTimestamp { state: AttemptState },
87
88 #[error(
89 "{state} carries {field} at {found}, which is before its created_at of \
90 {created_at}; an attempt cannot have changed state or concluded before \
91 it was allocated"
92 )]
93 TimestampsOutOfOrder {
94 state: AttemptState,
95 /// Which of the two orderable timestamps is out of order.
96 field: &'static str,
97 created_at: Timestamp,
98 found: Timestamp,
99 },
100}
101
102/// Ownership rule 2: "A host agent may act only on attempts persisted under its
103/// `host_id`."
104#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
105pub enum OwnershipError {
106 #[error(
107 "attempt {attempt} belongs to policy {attempt_policy}, but was checked \
108 against policy {policy}"
109 )]
110 PolicyMismatch {
111 attempt: AttemptId,
112 attempt_policy: PolicyId,
113 policy: PolicyId,
114 },
115
116 #[error(
117 "policy {policy} belongs to host {owner}; this agent runs on host \
118 {agent} and must not act on its attempts"
119 )]
120 ForeignHost {
121 policy: PolicyId,
122 owner: HostId,
123 agent: HostId,
124 },
125}
126
127// ---------------------------------------------------------------------------
128// AttemptState
129// ---------------------------------------------------------------------------
130
131/// The runner-attempt lifecycle, exactly as `04-subsystem-contracts.md` draws
132/// it after its 2026-08-21 amendment:
133///
134/// ```text
135/// allocated -> jit_received -> starting -> idle | busy
136/// idle -> busy
137/// allocated | jit_received | starting -> failed | orphaned
138/// idle | busy -> finished | failed | orphaned
139/// finished | failed | orphaned -> cleaned
140/// ```
141///
142/// `idle` means the runner process is registered and awaiting its single job
143/// assignment. It is short-lived and is **not** an idle persistent runner — this
144/// product has none, by D7.
145///
146/// **Every transition outside that diagram is rejected**, which `b1`'s
147/// Definition of Done requires. `cleaned` is absorbing, which is correct and
148/// intended.
149///
150/// **The last two edge sets were added by amendment, and the reasons matter to
151/// anyone reading this state machine.** `b1` first implemented the original
152/// diagram faithfully and surfaced its gaps as an explicit
153/// `NoLegalTransition` decision rather than inventing edges; the amendment
154/// closed them at the design level, and that decision value is gone with them:
155///
156/// * **`idle -> busy`.** `e3`'s Scope step 4 moves an attempt through
157/// `jit_received`, `starting`, `idle`, `busy` *in sequence*, and the
158/// definition of `idle` above describes a state that by construction precedes
159/// a job. Without this edge a runner observed idle and then assigned a job had
160/// nowhere legal to go.
161/// * **Terminal edges out of the three pre-registration states.** An attempt
162/// counts against host capacity for exactly as long as it is non-terminal
163/// ([`Self::counts_against_capacity`]), so an attempt that could not reach a
164/// terminal state **held a host capacity slot permanently**: two failed JIT
165/// requests on a `host_capacity: 2` host wedged that host into starting zero
166/// runners, with no error state and no cleanup path. `orphaned` is included
167/// for the restart case, where a pre-registration attempt is found after the
168/// agent restarts.
169///
170/// Those edges are also what make seven of the nine [`FailureReason`] variants
171/// reachable at all — `JitRequestFailed`, `JitExpired`,
172/// `RunnerPackageUnverified`, `RunnerVersionRejected`, `ProcessStartFailed`,
173/// `RegistrationTimedOut` and `TerminatedAfterRegistrationTimeout` each occur at
174/// a pre-registration state. `03-control-flows.md` flow 2 names the first five
175/// as conditions the agent must record; the last two are named by no document
176/// and are this crate's own, added because [`recovery_decision`] needs to say
177/// "alive, past its deadline, still unregistered" without calling it a crash,
178/// and because `e3` then needs to say what became of that runner without
179/// calling a process it stopped itself either a crash or still running.
180#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
181#[serde(rename_all = "snake_case")]
182pub enum AttemptState {
183 Allocated,
184 Preparing,
185 Prepared,
186 JitReceived,
187 Starting,
188 Idle,
189 Busy,
190 Finished,
191 Failed,
192 Orphaned,
193 Destroying,
194 CleanupDeferred,
195 Cleaned,
196}
197
198impl AttemptState {
199 pub const ALL: [AttemptState; 13] = [
200 AttemptState::Allocated,
201 AttemptState::Preparing,
202 AttemptState::Prepared,
203 AttemptState::JitReceived,
204 AttemptState::Starting,
205 AttemptState::Idle,
206 AttemptState::Busy,
207 AttemptState::Finished,
208 AttemptState::Failed,
209 AttemptState::Orphaned,
210 AttemptState::Destroying,
211 AttemptState::CleanupDeferred,
212 AttemptState::Cleaned,
213 ];
214
215 /// The complete legal transition list. A self-transition is not in it.
216 pub const LEGAL: &'static [(AttemptState, AttemptState)] = &[
217 // `allocated -> jit_received -> starting -> idle | busy`.
218 (AttemptState::Allocated, AttemptState::JitReceived),
219 (AttemptState::Allocated, AttemptState::Preparing),
220 (AttemptState::Preparing, AttemptState::Prepared),
221 (AttemptState::Prepared, AttemptState::JitReceived),
222 (AttemptState::JitReceived, AttemptState::Starting),
223 (AttemptState::Starting, AttemptState::Idle),
224 (AttemptState::Starting, AttemptState::Busy),
225 // `idle -> busy`.
226 (AttemptState::Idle, AttemptState::Busy),
227 // `busy -> idle`: GitHub showed the runner assigned, and the runner
228 // never started the job. See [`RunnerAttempt::assignment_not_taken`].
229 (AttemptState::Busy, AttemptState::Idle),
230 // `allocated | jit_received | starting -> failed | orphaned`.
231 (AttemptState::Allocated, AttemptState::Failed),
232 (AttemptState::Allocated, AttemptState::Orphaned),
233 (AttemptState::Preparing, AttemptState::Failed),
234 (AttemptState::Preparing, AttemptState::Orphaned),
235 (AttemptState::Prepared, AttemptState::Failed),
236 (AttemptState::Prepared, AttemptState::Orphaned),
237 (AttemptState::JitReceived, AttemptState::Failed),
238 (AttemptState::JitReceived, AttemptState::Orphaned),
239 (AttemptState::Starting, AttemptState::Failed),
240 (AttemptState::Starting, AttemptState::Orphaned),
241 // `idle | busy -> finished | failed | orphaned`.
242 (AttemptState::Idle, AttemptState::Finished),
243 (AttemptState::Idle, AttemptState::Failed),
244 (AttemptState::Idle, AttemptState::Orphaned),
245 (AttemptState::Busy, AttemptState::Finished),
246 (AttemptState::Busy, AttemptState::Failed),
247 (AttemptState::Busy, AttemptState::Orphaned),
248 // `finished | failed | orphaned -> cleaned`.
249 (AttemptState::Finished, AttemptState::Cleaned),
250 (AttemptState::Failed, AttemptState::Cleaned),
251 (AttemptState::Orphaned, AttemptState::Cleaned),
252 (AttemptState::Finished, AttemptState::Destroying),
253 (AttemptState::Failed, AttemptState::Destroying),
254 (AttemptState::Orphaned, AttemptState::Destroying),
255 (AttemptState::Destroying, AttemptState::CleanupDeferred),
256 (AttemptState::CleanupDeferred, AttemptState::Destroying),
257 (AttemptState::Destroying, AttemptState::Cleaned),
258 ];
259
260 /// The five states an attempt can still be concluded from: every
261 /// non-terminal state. After the amendment this is exactly
262 /// "not [`Self::is_terminal`]", which is what closed the permanently-held
263 /// capacity slot — but it is written out rather than derived, because the
264 /// two are only equal while the diagram gives every live state a terminal
265 /// edge, and that equality is a property worth failing loudly on.
266 pub const CONCLUDABLE_FROM: &'static [AttemptState] = &[
267 AttemptState::Allocated,
268 AttemptState::Preparing,
269 AttemptState::Prepared,
270 AttemptState::JitReceived,
271 AttemptState::Starting,
272 AttemptState::Idle,
273 AttemptState::Busy,
274 ];
275
276 #[must_use]
277 pub fn can_transition_to(self, next: AttemptState) -> bool {
278 Self::LEGAL.contains(&(self, next))
279 }
280
281 /// Terminal states, at which capacity is reclaimed.
282 ///
283 /// `e1`: "capacity is reclaimed only when an attempt reaches a terminal
284 /// state." `cleaned` is included because it follows a terminal state.
285 #[must_use]
286 pub const fn is_terminal(self) -> bool {
287 matches!(
288 self,
289 AttemptState::Finished
290 | AttemptState::Failed
291 | AttemptState::Orphaned
292 | AttemptState::Destroying
293 | AttemptState::CleanupDeferred
294 | AttemptState::Cleaned
295 )
296 }
297
298 /// Whether this attempt still occupies one of the host's capacity slots.
299 ///
300 /// This is the term the reconciliation formula subtracts, and getting it
301 /// wrong is silent: counting a terminal attempt starves the host, and failing
302 /// to count a `starting` one starts a second runner for a job already being
303 /// served.
304 #[must_use]
305 pub const fn counts_against_capacity(self) -> bool {
306 !self.is_terminal()
307 }
308
309 /// A runner that holds a slot and is still waiting for a job: the supply
310 /// `HostAllocator` subtracts from demand.
311 #[must_use]
312 pub const fn is_unassigned(self) -> bool {
313 matches!(
314 self,
315 AttemptState::Allocated
316 | AttemptState::Preparing
317 | AttemptState::Prepared
318 | AttemptState::JitReceived
319 | AttemptState::Starting
320 | AttemptState::Idle
321 )
322 }
323
324 /// The three terminal states that require an outcome before `cleaned`.
325 #[must_use]
326 pub const fn is_concluded(self) -> bool {
327 matches!(
328 self,
329 AttemptState::Finished | AttemptState::Failed | AttemptState::Orphaned
330 )
331 }
332}
333
334impl fmt::Display for AttemptState {
335 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
336 f.write_str(match self {
337 AttemptState::Allocated => "allocated",
338 AttemptState::Preparing => "preparing",
339 AttemptState::Prepared => "prepared",
340 AttemptState::JitReceived => "jit_received",
341 AttemptState::Starting => "starting",
342 AttemptState::Idle => "idle",
343 AttemptState::Busy => "busy",
344 AttemptState::Finished => "finished",
345 AttemptState::Failed => "failed",
346 AttemptState::Orphaned => "orphaned",
347 AttemptState::Destroying => "destroying",
348 AttemptState::CleanupDeferred => "cleanup_deferred",
349 AttemptState::Cleaned => "cleaned",
350 })
351 }
352}
353
354// ---------------------------------------------------------------------------
355// Outcome
356// ---------------------------------------------------------------------------
357
358/// Why an attempt failed.
359///
360/// `Other` exists so `e3` is not blocked by a reason this task did not
361/// anticipate; `crates/domain/src/attempt.rs` belongs to `b1` and `e3` cannot
362/// extend this enum itself.
363///
364/// **Nothing that reaches this type may contain a credential.** It is written to
365/// the journal by `b2` and rendered by `g2`, and `07-security.md`'s log scan runs
366/// over both. A JIT blob, an `Authorization` header, or a token in an `Other`
367/// string would defeat that gate from inside the domain, where the redacting log
368/// sink (`d1`) never gets a chance to see it.
369#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
370#[serde(rename_all = "snake_case")]
371pub enum FailureReason {
372 /// `generate-jitconfig` did not return a configuration.
373 JitRequestFailed,
374 /// The configuration was never claimed and expired (flow 4.4).
375 JitExpired,
376 /// The runner package's published checksum was absent or did not match
377 /// (`05-infrastructure.md`: the agent fails closed).
378 RunnerPackageUnverified,
379 /// GitHub rejects runners more than 30 days behind the latest release
380 /// (`01-current-architecture.md`, edge case 7). Terminal and
381 /// operator-actionable, never retried.
382 RunnerVersionRejected,
383 /// The child process could not be spawned.
384 ProcessStartFailed,
385 /// The child process exited before it could do its one job.
386 ///
387 /// **Only for a process that is actually gone.** A runner still running past
388 /// its startup deadline is [`Self::RegistrationTimedOut`], not this: `g2`
389 /// renders these strings to an operator, and telling one that a process
390 /// "exited unexpectedly" while it is visible in Task Manager spends the
391 /// credibility of every other message this product prints.
392 ProcessExitedUnexpectedly,
393 /// The runner process is up but never registered with GitHub inside its
394 /// startup window.
395 ///
396 /// Split from [`Self::ProcessExitedUnexpectedly`] because it is accurate and
397 /// because it points an operator somewhere else entirely. A process that
398 /// exited is a crash to investigate — logs, exit code, a corrupt runner
399 /// package. A process that is alive and unregistered has almost always
400 /// failed to *reach* GitHub: a proxy, a firewall, a DNS answer, an expired
401 /// or wrong-scoped configuration. Those are configuration and networking
402 /// fixes, and an operator sent to the wrong one of the two loses the time
403 /// this distinction exists to save.
404 RegistrationTimedOut,
405 /// The agent stopped a runner process that had not registered inside its
406 /// startup window.
407 ///
408 /// **The dead-process counterpart of [`Self::RegistrationTimedOut`], and it
409 /// exists because that one cannot be reused here.** By the time this reason
410 /// is recorded the process is gone — the agent signalled it — so rendering
411 /// "the runner process is running but did not register" would tell an
412 /// operator a process is up that they can see is not. That is the same false
413 /// liveness claim [`Self::ProcessExitedUnexpectedly`]'s documentation says
414 /// spends the credibility of every other message this product prints, and
415 /// `tests::no_decision_calls_a_dead_process_live` is what holds the line.
416 ///
417 /// **It points where [`Self::RegistrationTimedOut`] points, not where
418 /// [`Self::ProcessExitedUnexpectedly`] points.** The runner never reached
419 /// GitHub — a proxy, a firewall, a DNS answer, an expired or wrong-scoped
420 /// configuration — and the exit is the agent's own doing rather than
421 /// evidence of a crash. An operator sent to logs and exit codes for this is
422 /// investigating the wrong machine.
423 ///
424 /// **Who records it.** `e3`, and only `e3`. [`recovery_decision`] cannot
425 /// derive it: a process this agent killed and a process that crashed on its
426 /// own present the *same* [`RecoveryObservation`], so the distinguishing
427 /// fact has to be journalled as terminate-intent before the signal is sent
428 /// and read back afterwards. See [`RecoveryDecision::Terminate`] for the
429 /// window that obligation closes.
430 TerminatedAfterRegistrationTimeout,
431 /// A closed, credential-free isolation provider failure. Its category is
432 /// preserved in the attempt journal and operator activity rather than
433 /// flattened into free-form subprocess or adapter text.
434 IsolationProvider(IsolationProviderFailure),
435 /// An ephemeral attempt's workspace could not be removed yet, typically
436 /// because a late child still holds a file open or a directory refuses to
437 /// be listed; the reconciler retries it with backoff.
438 ///
439 /// **A cleanup report, never an attempt outcome.** It names why the
440 /// reconciler's cleanup of an already concluded attempt failed, so the
441 /// event and the log say *which* failure it was instead of `other`. No code
442 /// path concludes an attempt with it, which also keeps it out of the
443 /// journal that an older binary might read.
444 WorkspaceCleanupDeferred,
445 /// A required host capability is missing (`runner-manager host doctor`),
446 /// so the daemon refused to start a runner at all.
447 ///
448 /// **A launch refusal, never an attempt outcome**, for the reason
449 /// [`Self::WorkspaceCleanupDeferred`] gives: no attempt exists yet when
450 /// the daemon refuses, no runner is registered with GitHub, and the event
451 /// and the log name the refusal instead of `other`.
452 HostUnfit,
453 /// Anything else. Must carry no credential.
454 Other(String),
455}
456
457/// Operator-actionable isolation failures. No variant carries provider output,
458/// image text, JIT content, or another string that could contain a credential.
459#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
460#[serde(rename_all = "snake_case")]
461pub enum IsolationProviderFailure {
462 Unsupported,
463 NotInstalled,
464 PermissionDenied,
465 ImageUnavailableOrIncompatible,
466 DiskQuotaUnavailable,
467 Degraded,
468 RuntimeOperationFailed,
469 OwnershipMismatch,
470 UnsafeRuntimePath,
471 JitHandoffRejected,
472}
473
474impl fmt::Display for IsolationProviderFailure {
475 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
476 f.write_str(match self {
477 Self::Unsupported => "isolation provider unsupported on this host",
478 Self::NotInstalled => "isolation provider not installed",
479 Self::PermissionDenied => "isolation provider permission denied",
480 Self::ImageUnavailableOrIncompatible => "isolated image unavailable or incompatible",
481 Self::DiskQuotaUnavailable => "rootless OCI writable-layer disk quota unavailable",
482 Self::Degraded => "isolation provider degraded",
483 Self::RuntimeOperationFailed => "rootless OCI runtime operation failed",
484 Self::OwnershipMismatch => "OCI environment ownership mismatch",
485 Self::UnsafeRuntimePath => "OCI runtime must be on the Linux filesystem",
486 Self::JitHandoffRejected => "OCI JIT handoff rejected before start",
487 })
488 }
489}
490
491impl FailureReason {
492 /// One value of every variant, the counterpart of [`AttemptState::ALL`].
493 ///
494 /// `Other`'s detail is empty because what a caller enumerates is the
495 /// *variant*; no consumer should read the string out of this constant.
496 ///
497 /// **This list is hand-written, and what keeps it honest is not its own
498 /// length.** A length written as `9` next to nine elements asserts
499 /// nothing — that was the defect in the assertion this constant replaced.
500 /// What catches a new variant is the exhaustive, wildcard-free `match` in
501 /// `tests::earliest_state_producing`, which stops the test target compiling
502 /// the moment one is added and so puts the author in front of this list.
503 ///
504 /// **The residual gap, measured rather than assumed.** An author who adds a
505 /// variant, writes its `Display` arm and its `earliest_state_producing` arm,
506 /// and then adds it to *neither* this constant nor the test's `cases` table,
507 /// gets a green suite with the variant untested. Adding it to exactly one of
508 /// the two fails the length check; adding it to neither does not.
509 ///
510 /// Re-measured when `TerminatedAfterRegistrationTimeout` was added, because
511 /// a gap described once and never re-run is a gap nobody knows still exists.
512 /// With the variant declared and both match arms written but neither list
513 /// touched, `cargo test -p runner-manager-domain` was green — its lib target
514 /// reported `128 passed; 0 failed` — with the ninth variant unreachable and
515 /// untested. Adding it to this constant alone then failed the length check
516 /// with `left: 8 / right: 9`.
517 ///
518 /// **And the compiler never points at this constant.** A `const` array is
519 /// unaffected by a new variant, so nothing here errors. What stops the
520 /// author is `Display::fmt`'s match (`E0004`) and then
521 /// `tests::earliest_state_producing`'s, and neither of those mentions this
522 /// list — which is why a note pointing back here sits at each of those two
523 /// match sites, where the author is actually standing.
524 ///
525 /// **This is closable in stable Rust, and is hand-written anyway.** A local
526 /// `macro_rules!` that declares the enum and emits `ALL` from the same
527 /// variant list needs no dependency and no unstable feature
528 /// (`std::mem::variant_count` is unstable, but a declarative macro is not
529 /// the same thing). It is not used here because every variant of this enum
530 /// carries several paragraphs of its own documentation explaining what an
531 /// operator should do about it, and variants declared inside a macro
532 /// invocation are markedly worse to read and to `rustdoc`. That is a
533 /// legibility trade, deliberately taken — not an impossibility. If the
534 /// documentation ever thins out, the macro is the better answer.
535 pub const ALL: [FailureReason; 12] = [
536 FailureReason::JitRequestFailed,
537 FailureReason::JitExpired,
538 FailureReason::RunnerPackageUnverified,
539 FailureReason::RunnerVersionRejected,
540 FailureReason::ProcessStartFailed,
541 FailureReason::ProcessExitedUnexpectedly,
542 FailureReason::RegistrationTimedOut,
543 FailureReason::TerminatedAfterRegistrationTimeout,
544 FailureReason::IsolationProvider(IsolationProviderFailure::RuntimeOperationFailed),
545 FailureReason::WorkspaceCleanupDeferred,
546 FailureReason::HostUnfit,
547 FailureReason::Other(String::new()),
548 ];
549}
550
551impl fmt::Display for FailureReason {
552 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
553 // Adding a variant? This `E0004` is where the compiler stops you, and it
554 // is the *only* place it does for `FailureReason::ALL`, which is a const
555 // array and errors nowhere. Add the variant to `ALL` and to the `cases`
556 // table in `tests::all_failure_reasons_are_reachable_from_the_state_that
557 // _produces_them` as well, or it ships untested and unreachable.
558 match self {
559 FailureReason::JitRequestFailed => f.write_str("the JIT configuration request failed"),
560 FailureReason::JitExpired => f.write_str("the JIT configuration expired unclaimed"),
561 FailureReason::RunnerPackageUnverified => {
562 f.write_str("the runner package could not be verified")
563 }
564 FailureReason::RunnerVersionRejected => {
565 f.write_str("GitHub rejected the runner version")
566 }
567 FailureReason::ProcessStartFailed => f.write_str("the runner process failed to start"),
568 FailureReason::ProcessExitedUnexpectedly => {
569 f.write_str("the runner process exited unexpectedly")
570 }
571 FailureReason::RegistrationTimedOut => f.write_str(
572 "the runner process is running but did not register with GitHub \
573 before its startup deadline",
574 ),
575 // Past tense, and no claim that anything is still running: by the
576 // time this is recorded the agent has signalled the process and the
577 // operator can see it is gone. `tests::a_terminated_runner_is_never
578 // _described_as_running` pins that.
579 FailureReason::TerminatedAfterRegistrationTimeout => f.write_str(
580 "the agent stopped the runner process after it failed to \
581 register with GitHub before its startup deadline",
582 ),
583 FailureReason::IsolationProvider(category) => category.fmt(f),
584 FailureReason::WorkspaceCleanupDeferred => {
585 f.write_str("the attempt workspace could not be removed yet; cleanup will retry")
586 }
587 FailureReason::HostUnfit => f.write_str(
588 "a required host check fails, so no runner is started; run \
589 `runner-manager host doctor`",
590 ),
591 FailureReason::Other(detail) => write!(f, "{detail}"),
592 }
593 }
594}
595
596/// What terminally happened to an attempt.
597///
598/// The state and the outcome are not two independent fields that a caller has to
599/// keep consistent: [`AttemptOutcome::terminal_state`] derives the state from the
600/// outcome, and [`RunnerAttempt::conclude`] is the only way to set either. A
601/// `failed` attempt whose outcome says it ran a job is therefore not a bug this
602/// code can have.
603#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
604#[serde(tag = "outcome", rename_all = "snake_case")]
605pub enum AttemptOutcome {
606 /// The runner accepted its one job and the job ended. This says the *runner*
607 /// finished, never that the workflow succeeded — GitHub remains the source of
608 /// truth for workflow outcome (`03-control-flows.md`, flow 2, Failure).
609 CompletedJob,
610 /// The surplus case. The runner registered, no job arrived, and it exited on
611 /// its idle timeout (flow 2.7). Normal, bounded, and **not** a failure.
612 ExitedIdleWithoutWork,
613 /// The attempt failed.
614 Failed { reason: FailureReason },
615 /// Supervision was lost: the process is gone and the attempt could not be
616 /// reconciled with GitHub (`e3`, restart recovery).
617 Orphaned,
618}
619
620impl AttemptOutcome {
621 #[must_use]
622 pub fn failed(reason: FailureReason) -> Self {
623 Self::Failed { reason }
624 }
625
626 /// The one terminal state this outcome corresponds to.
627 #[must_use]
628 pub const fn terminal_state(&self) -> AttemptState {
629 match self {
630 AttemptOutcome::CompletedJob | AttemptOutcome::ExitedIdleWithoutWork => {
631 AttemptState::Finished
632 }
633 AttemptOutcome::Failed { .. } => AttemptState::Failed,
634 AttemptOutcome::Orphaned => AttemptState::Orphaned,
635 }
636 }
637
638 /// True for the outcomes an operator should be told to look at.
639 ///
640 /// `g2` renders this differently from [`Self::is_idle_exit`], and `e1`'s
641 /// Definition of Done requires that a surplus attempt "is not reported as a
642 /// failure".
643 #[must_use]
644 pub const fn is_failure(&self) -> bool {
645 matches!(
646 self,
647 AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned
648 )
649 }
650
651 /// True only for the surplus case.
652 #[must_use]
653 pub const fn is_idle_exit(&self) -> bool {
654 matches!(self, AttemptOutcome::ExitedIdleWithoutWork)
655 }
656
657 /// True only when this runner actually took a job.
658 #[must_use]
659 pub const fn ran_a_job(&self) -> bool {
660 matches!(self, AttemptOutcome::CompletedJob)
661 }
662
663 /// The states an attempt must be in for this outcome to be reachable.
664 ///
665 /// Two of the four outcomes are narrower than the diagram's terminal edges,
666 /// and deliberately so: `finished` is reachable from both `idle` and `busy`,
667 /// but only one of them can have produced each of the two outcomes that lead
668 /// there. Failure and orphaning are as wide as the diagram — every live
669 /// state has a `-> failed | orphaned` edge since the amendment, and
670 /// narrowing them per [`FailureReason`] here would invent product rules no
671 /// document states.
672 const fn required_from(&self) -> &'static [AttemptState] {
673 match self {
674 // Only a runner that was assigned a job can have run one.
675 AttemptOutcome::CompletedJob => &[AttemptState::Busy],
676 // Only a runner that was registered and waiting can have exited idle.
677 AttemptOutcome::ExitedIdleWithoutWork => &[AttemptState::Idle],
678 AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned => {
679 AttemptState::CONCLUDABLE_FROM
680 }
681 }
682 }
683}
684
685impl fmt::Display for AttemptOutcome {
686 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
687 match self {
688 AttemptOutcome::CompletedJob => f.write_str("ran a job"),
689 AttemptOutcome::ExitedIdleWithoutWork => f.write_str("exited idle without work"),
690 AttemptOutcome::Failed { reason } => write!(f, "failed: {reason}"),
691 AttemptOutcome::Orphaned => f.write_str("orphaned"),
692 }
693 }
694}
695
696// ---------------------------------------------------------------------------
697// RunnerAttempt
698// ---------------------------------------------------------------------------
699
700/// One ephemeral runner, from directory allocation to cleanup.
701///
702/// The fields `04-subsystem-contracts.md` names are all here. Two are not in that
703/// list and are added deliberately:
704///
705/// * `outcome`, because `b1`'s Scope requires the attempt to "carry an outcome
706/// distinguishing 'ran a job' from 'exited idle without work'".
707/// * `last_state_change_at`, because recovery decisions measure elapsed time in
708/// the *current* state — an idle timeout runs from entering `idle`, not from
709/// `created_at` — and `b1`'s Definition of Done requires those decisions to be
710/// testable against a fake clock.
711///
712/// `b2` persists both.
713#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
714pub struct RunnerAttempt {
715 pub id: AttemptId,
716 pub policy_id: PolicyId,
717 github_runner_id: Option<u64>,
718 state: AttemptState,
719 outcome: Option<AttemptOutcome>,
720 process_id: Option<u32>,
721 execution: AttemptExecution,
722 runtime_path: PathBuf,
723 /// Which cleanup algorithm this attempt's directory is entitled to, and the
724 /// slot it leases if any.
725 ///
726 /// Immutable after allocation, by `02-target-architecture.md`: "The
727 /// workspace kind and slot number tell recovery which cleanup algorithm is
728 /// legal. Neither may change after allocation." Private with no setter is
729 /// how that is enforced — a mutable kind would let a running attempt convert
730 /// a disposable directory into a retained one, which is exactly the
731 /// two-job contamination path `04-security-recovery.md` measures.
732 workspace: AttemptWorkspace,
733 pub created_at: Timestamp,
734 terminal_at: Option<Timestamp>,
735 last_state_change_at: Timestamp,
736}
737
738/// Every stored column of one attempt, named rather than positional.
739///
740/// **Why this is a struct.** [`RunnerAttempt::from_persisted`] took ten
741/// positional arguments, and two of them — `created_at` and
742/// `last_state_change_at` — are both `Timestamp`. Transposing them type-checked,
743/// compiled, and silently reverted `last_state_change_at` to `created_at`, which
744/// is precisely the bug that field exists to prevent: every recovery timeout
745/// would then have measured from allocation rather than from the current state,
746/// so a long-running `busy` attempt would be read as a stuck one. `terminal_at`
747/// is a third `Option<Timestamp>` in the same list.
748///
749/// `b2` maps database columns onto this type. With a struct that mapping is
750/// checked by name at compile time; positionally it was checked by nothing.
751///
752/// **That guarantee covers the Rust side of the mapping and no more.** It is
753/// the *field* names that the compiler checks, not the column names they are
754/// read from: `PersistedAttempt { created_at: row.get("last_state_change_at")?,
755/// … }` compiles exactly as happily as the correct version, and reintroduces the
756/// very transposition described above. `b2` still owes a test that loads a row
757/// whose columns hold distinguishable values and asserts each landed in the
758/// field of the same name; this type does not supply one.
759///
760/// Construct it with a struct literal so every field is written down at the call
761/// site — that is the whole point, and a builder or a `Default` would give the
762/// omission back.
763#[derive(Debug, Clone, PartialEq, Eq)]
764pub struct PersistedAttempt {
765 pub id: AttemptId,
766 pub policy_id: PolicyId,
767 pub github_runner_id: Option<u64>,
768 pub state: AttemptState,
769 pub outcome: Option<AttemptOutcome>,
770 pub process_id: Option<u32>,
771 pub execution: AttemptExecution,
772 pub runtime_path: PathBuf,
773 /// `ephemeral` or `persistent`, stored beside the slot below.
774 pub workspace_kind: WorkspaceKind,
775 /// The leased slot: `Some` exactly when `workspace_kind` is `persistent`,
776 /// and positive. It is a raw `u16` here rather than a `NonZeroU16` so that
777 /// the `0` a hand-edited row can hold is refused by
778 /// [`AttemptWorkspace::from_persisted`] instead of being unrepresentable at
779 /// the column boundary and panicking somewhere else.
780 pub workspace_slot: Option<u16>,
781 /// When the runtime directory was allocated. Never moves.
782 pub created_at: Timestamp,
783 /// Set when, and only when, the attempt concluded.
784 pub terminal_at: Option<Timestamp>,
785 /// When the attempt entered its **current** state. Recovery timeouts run
786 /// from here, not from `created_at`.
787 pub last_state_change_at: Timestamp,
788}
789
790impl RunnerAttempt {
791 /// Every stored column of this attempt, for `b2` to write back.
792 ///
793 /// The exact inverse of [`Self::from_persisted`], so a round trip through
794 /// the journal is expressible without reaching for a field accessor per
795 /// column and without this type exposing its private fields for writing.
796 #[must_use]
797 pub fn to_persisted(&self) -> PersistedAttempt {
798 PersistedAttempt {
799 id: self.id,
800 policy_id: self.policy_id,
801 github_runner_id: self.github_runner_id,
802 state: self.state,
803 outcome: self.outcome.clone(),
804 process_id: self.process_id,
805 execution: self.execution.clone(),
806 runtime_path: self.runtime_path.clone(),
807 workspace_kind: self.workspace.kind(),
808 workspace_slot: self.workspace.slot_number(),
809 created_at: self.created_at,
810 terminal_at: self.terminal_at,
811 last_state_change_at: self.last_state_change_at,
812 }
813 }
814
815 /// The first step of `e3`'s per-attempt flow: a runtime directory is
816 /// allocated and journalled **before** anything remote happens, so a crash
817 /// leaves a recoverable trace rather than an invisible one.
818 /// D3 keeps this the disposable path: an attempt allocated through it is
819 /// [`AttemptWorkspace::Ephemeral`], so every existing caller and test goes on
820 /// producing the behaviour it produced before persistent slots existed.
821 /// [`Self::allocate_in`] is the one that leases a slot.
822 #[must_use]
823 pub fn allocate(
824 id: AttemptId,
825 policy_id: PolicyId,
826 runtime_path: impl Into<PathBuf>,
827 now: Timestamp,
828 ) -> Self {
829 Self::allocate_in(
830 id,
831 policy_id,
832 runtime_path,
833 AttemptWorkspace::Ephemeral,
834 now,
835 )
836 }
837
838 /// The same first step, recording which workspace the directory came from.
839 ///
840 /// `c2` calls this with [`AttemptWorkspace::PersistentSlot`] while holding
841 /// the host allocation lock, so the slot lease is journalled "before package
842 /// or GitHub effects" and a crash between the two leaves a recoverable trace
843 /// rather than an orphaned slot.
844 #[must_use]
845 pub fn allocate_in(
846 id: AttemptId,
847 policy_id: PolicyId,
848 runtime_path: impl Into<PathBuf>,
849 workspace: AttemptWorkspace,
850 now: Timestamp,
851 ) -> Self {
852 Self {
853 id,
854 policy_id,
855 github_runner_id: None,
856 state: AttemptState::Allocated,
857 outcome: None,
858 process_id: None,
859 execution: AttemptExecution::Native { process_id: None },
860 runtime_path: runtime_path.into(),
861 workspace,
862 created_at: now,
863 terminal_at: None,
864 last_state_change_at: now,
865 }
866 }
867
868 /// Rebuild a journalled attempt.
869 ///
870 /// # Errors
871 /// Any state/outcome/timestamp combination that this crate's own transitions
872 /// cannot produce, so a hand-edited journal cannot inject a `failed` attempt
873 /// that claims to have run a job, or a `finished` one that never reached a
874 /// terminal state.
875 pub fn from_persisted(fields: PersistedAttempt) -> Result<Self, AttemptError> {
876 let PersistedAttempt {
877 id,
878 policy_id,
879 github_runner_id,
880 state,
881 outcome,
882 process_id,
883 execution,
884 runtime_path,
885 workspace_kind,
886 workspace_slot,
887 created_at,
888 terminal_at,
889 last_state_change_at,
890 } = fields;
891
892 // The workspace pair is checked first because it decides which cleanup
893 // algorithm recovery is allowed to run on `runtime_path`. A row that
894 // claims `persistent` with no slot, or `ephemeral` with one, names a
895 // directory whose safe cleanup is undecidable, and
896 // `04-security-recovery.md` requires that to fail closed rather than to
897 // fall back to the destructive branch.
898 let workspace = AttemptWorkspace::from_persisted(workspace_kind, workspace_slot)?;
899 execution.validate()?;
900 match &execution {
901 AttemptExecution::Native {
902 process_id: identity_pid,
903 } if *identity_pid == process_id => {}
904 AttemptExecution::Isolated { .. } if process_id.is_none() => {}
905 _ => return Err(ExecutionError::AttemptIdentityMismatch.into()),
906 }
907
908 match (&outcome, state.is_terminal()) {
909 (None, true) => return Err(AttemptError::TerminalWithoutOutcome { state }),
910 (Some(outcome), false) => {
911 return Err(AttemptError::NonTerminalWithOutcome {
912 state,
913 outcome: outcome.clone(),
914 });
915 }
916 (Some(outcome), true) => {
917 let expected = outcome.terminal_state();
918 let allowed_isolated_cleanup = !execution.is_native()
919 && matches!(
920 state,
921 AttemptState::Destroying | AttemptState::CleanupDeferred
922 );
923 if ![expected, AttemptState::Cleaned].contains(&state) && !allowed_isolated_cleanup
924 {
925 return Err(AttemptError::OutcomeStateMismatch {
926 state,
927 outcome: outcome.clone(),
928 });
929 }
930 }
931 (None, false) => {}
932 }
933
934 // `terminal_at` is validated on exactly the same footing as `outcome`,
935 // and for the same stated reason. `conclude` is the only writer of both
936 // and sets them together, so `state.is_terminal()` and
937 // `terminal_at.is_some()` are equivalent in anything this crate
938 // produced; a row where they disagree was edited by hand. Without this,
939 // a `finished` attempt with no `terminal_at` loaded cleanly and every
940 // consumer of `terminal_at()` -- retention, reporting, `g2`'s ordering
941 // -- silently saw an attempt that had never concluded.
942 match (terminal_at, state.is_terminal()) {
943 (None, true) => return Err(AttemptError::TerminalWithoutTimestamp { state }),
944 (Some(_), false) => return Err(AttemptError::NonTerminalWithTimestamp { state }),
945 _ => {}
946 }
947
948 if matches!(
949 state,
950 AttemptState::Preparing
951 | AttemptState::Prepared
952 | AttemptState::Destroying
953 | AttemptState::CleanupDeferred
954 ) && execution.is_native()
955 {
956 return Err(ExecutionError::AttemptIdentityMismatch.into());
957 }
958 if matches!(
959 state,
960 AttemptState::Prepared
961 | AttemptState::JitReceived
962 | AttemptState::Starting
963 | AttemptState::Idle
964 | AttemptState::Busy
965 ) && matches!(
966 &execution,
967 AttemptExecution::Isolated {
968 environment_id: None,
969 ..
970 }
971 ) {
972 return Err(ExecutionError::AttemptIdentityMismatch.into());
973 }
974
975 // Presence was checked above; *ordering* is checked here, and it is a
976 // separate hazard. `created_at` never moves and every other timestamp is
977 // written by a transition that happens after it, so a row where either
978 // precedes it is one this crate cannot have produced. Without this, a
979 // `finished` row with `created_at: ts(100), terminal_at: Some(ts(0))`
980 // loaded cleanly -- an attempt that concluded a hundred seconds before
981 // it was created -- and every duration computed from the pair came out
982 // negative or wrapped. The hand-edited-journal threat model that
983 // motivates the presence gate covers this equally.
984 if last_state_change_at < created_at {
985 return Err(AttemptError::TimestampsOutOfOrder {
986 state,
987 field: "last_state_change_at",
988 created_at,
989 found: last_state_change_at,
990 });
991 }
992 if let Some(terminal_at) = terminal_at
993 && terminal_at < created_at
994 {
995 return Err(AttemptError::TimestampsOutOfOrder {
996 state,
997 field: "terminal_at",
998 created_at,
999 found: terminal_at,
1000 });
1001 }
1002
1003 Ok(Self {
1004 id,
1005 policy_id,
1006 github_runner_id,
1007 state,
1008 outcome,
1009 process_id,
1010 execution,
1011 runtime_path,
1012 workspace,
1013 created_at,
1014 terminal_at,
1015 last_state_change_at,
1016 })
1017 }
1018
1019 #[must_use]
1020 pub const fn state(&self) -> AttemptState {
1021 self.state
1022 }
1023
1024 /// When the attempt was allocated.
1025 #[must_use]
1026 pub const fn created_at(&self) -> Timestamp {
1027 self.created_at
1028 }
1029
1030 #[must_use]
1031 pub const fn outcome(&self) -> Option<&AttemptOutcome> {
1032 self.outcome.as_ref()
1033 }
1034
1035 #[must_use]
1036 pub const fn github_runner_id(&self) -> Option<u64> {
1037 self.github_runner_id
1038 }
1039
1040 #[must_use]
1041 pub const fn process_id(&self) -> Option<u32> {
1042 self.process_id
1043 }
1044
1045 #[must_use]
1046 pub const fn execution(&self) -> &AttemptExecution {
1047 &self.execution
1048 }
1049
1050 /// Choose the provider intent before the first journal write.
1051 pub fn allocate_execution(&mut self, execution: AttemptExecution) -> Result<(), AttemptError> {
1052 if self.state != AttemptState::Allocated || self.process_id.is_some() {
1053 return Err(ExecutionError::AttemptIdentityMismatch.into());
1054 }
1055 execution.validate()?;
1056 if matches!(
1057 execution,
1058 AttemptExecution::Native {
1059 process_id: Some(_)
1060 }
1061 ) {
1062 return Err(ExecutionError::AttemptIdentityMismatch.into());
1063 }
1064 self.execution = execution;
1065 Ok(())
1066 }
1067
1068 /// Fill the provider identity through a durable journal write before an
1069 /// isolated runner receives JIT. A second or changed identity is refused.
1070 pub fn prepared_environment(&mut self, environment_id: String) -> Result<(), AttemptError> {
1071 if !matches!(
1072 self.state,
1073 AttemptState::Allocated
1074 | AttemptState::Preparing
1075 | AttemptState::Failed
1076 | AttemptState::Orphaned
1077 | AttemptState::Destroying
1078 | AttemptState::CleanupDeferred
1079 ) {
1080 return Err(ExecutionError::AttemptIdentityMismatch.into());
1081 }
1082 let AttemptExecution::Isolated {
1083 environment_id: stored,
1084 ..
1085 } = &mut self.execution
1086 else {
1087 return Err(ExecutionError::AttemptIdentityMismatch.into());
1088 };
1089 if stored.is_some() {
1090 return Err(ExecutionError::AttemptIdentityMismatch.into());
1091 }
1092 *stored = Some(environment_id);
1093 if let Err(error) = self.execution.validate() {
1094 if let AttemptExecution::Isolated {
1095 environment_id: stored,
1096 ..
1097 } = &mut self.execution
1098 {
1099 *stored = None;
1100 }
1101 return Err(error.into());
1102 }
1103 Ok(())
1104 }
1105
1106 /// Persist intent before a provider performs an external prepare effect.
1107 pub fn begin_prepare(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1108 if self.execution.is_native() {
1109 return Err(ExecutionError::AttemptIdentityMismatch.into());
1110 }
1111 self.move_to(AttemptState::Preparing, now)
1112 }
1113
1114 /// Persist the returned provider identity before requesting JIT.
1115 pub fn mark_prepared(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1116 if !matches!(
1117 self.execution,
1118 AttemptExecution::Isolated {
1119 environment_id: Some(_),
1120 ..
1121 }
1122 ) {
1123 return Err(ExecutionError::AttemptIdentityMismatch.into());
1124 }
1125 self.move_to(AttemptState::Prepared, now)
1126 }
1127
1128 #[must_use]
1129 pub fn runtime_path(&self) -> &Path {
1130 &self.runtime_path
1131 }
1132
1133 /// The immutable allocation fact: disposable, or the persistent slot leased.
1134 ///
1135 /// There is no setter. Cleanup and recovery dispatch on this
1136 /// (`02-target-architecture.md`, "Cleanup and recovery"), so a value that
1137 /// could be changed after allocation would let the algorithm chosen for a
1138 /// directory disagree with the one it was created under.
1139 #[must_use]
1140 pub const fn workspace(&self) -> AttemptWorkspace {
1141 self.workspace
1142 }
1143
1144 /// Whether this attempt holds a persistent slot lease.
1145 ///
1146 /// Every uncleaned persistent attempt is a lease, including a terminal one
1147 /// whose cleanup failed — which is why this asks about the workspace and not
1148 /// about the state.
1149 #[must_use]
1150 pub const fn holds_slot_lease(&self) -> bool {
1151 self.workspace.is_persistent() && !matches!(self.state, AttemptState::Cleaned)
1152 }
1153
1154 #[must_use]
1155 pub const fn terminal_at(&self) -> Option<Timestamp> {
1156 self.terminal_at
1157 }
1158
1159 /// When the attempt entered its current state. Recovery timeouts run from
1160 /// here.
1161 #[must_use]
1162 pub const fn last_state_change_at(&self) -> Timestamp {
1163 self.last_state_change_at
1164 }
1165
1166 #[must_use]
1167 pub const fn is_terminal(&self) -> bool {
1168 self.state.is_terminal()
1169 }
1170
1171 /// Whether this attempt still holds one of the host's capacity slots.
1172 /// Isolated environments retain the slot through terminal runner states
1173 /// until provider absence has been proved and cleanup is journalled.
1174 #[must_use]
1175 pub const fn counts_against_capacity(&self) -> bool {
1176 self.state.counts_against_capacity()
1177 || (!self.execution.is_native() && !matches!(self.state, AttemptState::Cleaned))
1178 }
1179
1180 fn move_to(&mut self, next: AttemptState, now: Timestamp) -> Result<(), AttemptError> {
1181 if !self.state.can_transition_to(next) {
1182 return Err(AttemptError::IllegalTransition {
1183 from: self.state,
1184 to: next,
1185 });
1186 }
1187 self.state = next;
1188 self.last_state_change_at = now;
1189 Ok(())
1190 }
1191
1192 /// `allocated -> jit_received`.
1193 ///
1194 /// # Errors
1195 /// [`AttemptError::IllegalTransition`] from any other state.
1196 pub fn jit_received(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1197 self.move_to(AttemptState::JitReceived, now)
1198 }
1199
1200 /// `jit_received -> starting`, recording the child process identity.
1201 ///
1202 /// # Errors
1203 /// [`AttemptError::IllegalTransition`] from any other state.
1204 pub fn started(&mut self, process_id: u32, now: Timestamp) -> Result<(), AttemptError> {
1205 if !self.execution.is_native() {
1206 return Err(ExecutionError::AttemptIdentityMismatch.into());
1207 }
1208 self.move_to(AttemptState::Starting, now)?;
1209 self.process_id = Some(process_id);
1210 self.execution = AttemptExecution::Native {
1211 process_id: Some(process_id),
1212 };
1213 Ok(())
1214 }
1215
1216 /// An isolated environment has no host PID; its immutable identity is already journalled.
1217 pub fn started_isolated(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1218 if !matches!(
1219 self.execution,
1220 AttemptExecution::Isolated {
1221 environment_id: Some(_),
1222 ..
1223 }
1224 ) {
1225 return Err(ExecutionError::AttemptIdentityMismatch.into());
1226 }
1227 self.move_to(AttemptState::Starting, now)
1228 }
1229
1230 /// `starting -> idle`: the runner registered and is awaiting its one
1231 /// assignment.
1232 ///
1233 /// # Errors
1234 /// [`AttemptError::IllegalTransition`] from any other state.
1235 pub fn registered_idle(
1236 &mut self,
1237 github_runner_id: u64,
1238 now: Timestamp,
1239 ) -> Result<(), AttemptError> {
1240 self.move_to(AttemptState::Idle, now)?;
1241 self.github_runner_id = Some(github_runner_id);
1242 Ok(())
1243 }
1244
1245 /// `starting | idle -> busy`: the runner was assigned its one job.
1246 ///
1247 /// Both sources are real. A runner may be observed taking a job directly out
1248 /// of `starting`, or it may be seen `idle` first and pick the job up on a
1249 /// later pass — `e3`'s Scope step 4 walks the second sequence explicitly.
1250 ///
1251 /// # Errors
1252 /// [`AttemptError::IllegalTransition`] from any other state.
1253 pub fn assigned_job(
1254 &mut self,
1255 github_runner_id: u64,
1256 now: Timestamp,
1257 ) -> Result<(), AttemptError> {
1258 self.move_to(AttemptState::Busy, now)?;
1259 self.github_runner_id = Some(github_runner_id);
1260 Ok(())
1261 }
1262
1263 /// `busy -> idle`: GitHub showed this runner assigned, but the runner never
1264 /// started the job.
1265 ///
1266 /// GitHub's inventory marks a runner busy when it hands it a job, before the
1267 /// runner has acquired it. An assignment the runner does not take up is
1268 /// re-queued to another runner, and this one goes back to waiting (or, as a
1269 /// one-shot JIT runner, exits). Without this edge the journal went on saying
1270 /// `busy`, so the attempt was later recorded `completed_job` for a job it
1271 /// never ran, and a live runner in that position was never held to the idle
1272 /// timeout. The caller takes the edge only with local proof that no job ran:
1273 /// the runner's own diagnostics show its listener but no job worker.
1274 ///
1275 /// # Errors
1276 /// [`AttemptError::IllegalTransition`] from any state but `busy`.
1277 pub fn assignment_not_taken(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1278 if self.state != AttemptState::Busy {
1279 return Err(AttemptError::IllegalTransition {
1280 from: self.state,
1281 to: AttemptState::Idle,
1282 });
1283 }
1284 self.move_to(AttemptState::Idle, now)
1285 }
1286
1287 /// Record the terminal outcome, moving to the state it implies.
1288 ///
1289 /// # Errors
1290 /// [`AttemptError::OutcomeUnreachable`] when the outcome does not belong to
1291 /// the current state — `CompletedJob` from anything but `busy`, or
1292 /// `ExitedIdleWithoutWork` from anything but `idle` — and
1293 /// [`AttemptError::IllegalTransition`] otherwise.
1294 pub fn conclude(
1295 &mut self,
1296 outcome: AttemptOutcome,
1297 now: Timestamp,
1298 ) -> Result<(), AttemptError> {
1299 if !outcome.required_from().contains(&self.state) {
1300 return Err(AttemptError::OutcomeUnreachable {
1301 from: self.state,
1302 outcome,
1303 });
1304 }
1305 self.move_to(outcome.terminal_state(), now)?;
1306 self.terminal_at = Some(now);
1307 self.outcome = Some(outcome);
1308 Ok(())
1309 }
1310
1311 /// `finished | failed | orphaned -> cleaned`.
1312 ///
1313 /// # Errors
1314 /// [`AttemptError::BusyCannotBeCleaned`] for a `busy` attempt — the
1315 /// scale-down case `04-subsystem-contracts.md` forbids — and
1316 /// [`AttemptError::IllegalTransition`] for any other non-terminal state or
1317 /// for an already-cleaned attempt.
1318 pub fn clean(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1319 if self.state == AttemptState::Busy {
1320 return Err(AttemptError::BusyCannotBeCleaned);
1321 }
1322 if !self.execution.is_native() {
1323 return Err(ExecutionError::IsolatedCleanupUnproven.into());
1324 }
1325 if !self.state.is_concluded() {
1326 return Err(AttemptError::IllegalTransition {
1327 from: self.state,
1328 to: AttemptState::Cleaned,
1329 });
1330 }
1331 self.move_to(AttemptState::Cleaned, now)
1332 }
1333
1334 /// Durable cleanup intent, recorded before stop or destroy is attempted.
1335 pub fn begin_destroy(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1336 if self.execution.is_native() || self.outcome.is_none() {
1337 return Err(ExecutionError::AttemptIdentityMismatch.into());
1338 }
1339 self.move_to(AttemptState::Destroying, now)
1340 }
1341
1342 /// The provider still owns resources; preserve the capacity lease for a retry.
1343 pub fn defer_cleanup(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1344 if self.execution.is_native() {
1345 return Err(ExecutionError::AttemptIdentityMismatch.into());
1346 }
1347 self.move_to(AttemptState::CleanupDeferred, now)
1348 }
1349
1350 /// Only the lifecycle may call this after an independent missing observation.
1351 pub fn clean_isolated(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1352 if self.execution.is_native() {
1353 return Err(ExecutionError::AttemptIdentityMismatch.into());
1354 }
1355 self.move_to(AttemptState::Cleaned, now)
1356 }
1357}
1358
1359// ---------------------------------------------------------------------------
1360// Ownership
1361// ---------------------------------------------------------------------------
1362
1363/// Ownership rules 1 and 2 (`04-subsystem-contracts.md`).
1364///
1365/// An attempt records its `policy_id`, and the policy records its `host_id`, so
1366/// authorising an attempt is a two-link check and both links matter. `e3`'s
1367/// restart recovery runs this before it touches a process it found on the
1368/// machine: adopting, terminating, or cleaning another host's attempt is the
1369/// failure this rule exists to prevent.
1370///
1371/// # Errors
1372/// [`OwnershipError::PolicyMismatch`] when the attempt does not belong to the
1373/// policy, and [`OwnershipError::ForeignHost`] when the policy does not belong to
1374/// this agent's host.
1375pub fn authorize(
1376 agent_host: HostId,
1377 policy: &ScalePolicy,
1378 attempt: &RunnerAttempt,
1379) -> Result<(), OwnershipError> {
1380 if attempt.policy_id != policy.id {
1381 return Err(OwnershipError::PolicyMismatch {
1382 attempt: attempt.id,
1383 attempt_policy: attempt.policy_id,
1384 policy: policy.id,
1385 });
1386 }
1387 if !policy.is_owned_by(agent_host) {
1388 return Err(OwnershipError::ForeignHost {
1389 policy: policy.id,
1390 owner: policy.host_id,
1391 agent: agent_host,
1392 });
1393 }
1394 Ok(())
1395}
1396
1397// ---------------------------------------------------------------------------
1398// Recovery
1399// ---------------------------------------------------------------------------
1400
1401/// How long an attempt may sit in each pre-terminal state before recovery treats
1402/// it as stuck.
1403///
1404/// **No document in this taskflow states these durations.** `03-control-flows.md`
1405/// flow 2.7 says a surplus runner "exits on its idle timeout" without giving one,
1406/// and flow 4.4 says an expired JIT configuration is discarded without saying
1407/// when it expires. [`RecoveryTimeouts::provisional`] therefore returns values
1408/// chosen here, named so that a caller cannot mistake them for a product
1409/// decision, and there is deliberately no `Default` impl — `e1` and `e3` should
1410/// have to write the numbers down.
1411#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1412pub struct RecoveryTimeouts {
1413 /// How long `allocated` or `jit_received` may last before the JIT
1414 /// configuration is assumed lost.
1415 pub jit_handoff: Elapsed,
1416 /// How long `starting` may last before the runner is assumed not to be
1417 /// coming up.
1418 pub startup: Elapsed,
1419 /// How long `idle` may last, in both directions it is read.
1420 ///
1421 /// For an attempt whose process is already gone it separates the two
1422 /// readings of that exit: past this, the surplus case; before it, a crash.
1423 ///
1424 /// For one whose process is still alive and still registered it is a
1425 /// deadline rather than a reading — the point at which the agent stops the
1426 /// runner itself. Nothing else does: `Runner.Listener run` long-polls for
1427 /// an assignment indefinitely, so this value, and only this value, bounds
1428 /// how long a runner that never gets a job holds its capacity slot and its
1429 /// entry in the target's runner settings.
1430 pub idle: Elapsed,
1431}
1432
1433impl RecoveryTimeouts {
1434 #[must_use]
1435 pub const fn new(jit_handoff: Elapsed, startup: Elapsed, idle: Elapsed) -> Self {
1436 Self {
1437 jit_handoff,
1438 startup,
1439 idle,
1440 }
1441 }
1442
1443 /// Placeholder values, with one exception — see the type documentation.
1444 ///
1445 /// `idle` is no longer a placeholder. It stopped being one when it became
1446 /// the only thing that bounds a live runner: five minutes is long enough
1447 /// that a runner GitHub is about to assign is not stopped out from under
1448 /// the assignment, and short enough that a machine does not carry an
1449 /// unusable slot, and a repository an unusable runner row, for hours. The
1450 /// other two still only separate readings of an event that already
1451 /// happened, and nothing here has had to decide what they should be.
1452 #[must_use]
1453 pub fn provisional() -> Self {
1454 Self {
1455 jit_handoff: Elapsed::seconds(120),
1456 startup: Elapsed::seconds(300),
1457 idle: Elapsed::seconds(300),
1458 }
1459 }
1460}
1461
1462/// What GitHub says about this attempt's runner.
1463///
1464/// Precedence rule 3: "GitHub runner status is authoritative for remote job
1465/// status; local process state is authoritative only for a child process owned by
1466/// this agent." Both halves are inputs here, and neither is allowed to stand in
1467/// for the other.
1468#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1469pub enum GithubRunnerObservation {
1470 /// GitHub could not be reached this cycle. Flow 3.3: start nothing, retain
1471 /// what is running, back off. It is emphatically **not** the same as
1472 /// `NotRegistered`, and conflating the two would delete live runners during
1473 /// an outage.
1474 Unreachable,
1475 /// GitHub knows no runner for this attempt.
1476 NotRegistered,
1477 /// GitHub knows the runner.
1478 Registered { busy: bool },
1479}
1480
1481/// One attempt's observed reality at recovery time.
1482#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1483pub struct RecoveryObservation {
1484 /// Whether the child process this agent recorded is still alive. `d1` supplies
1485 /// a process identity that survives a reboot, because a bare PID is reused.
1486 pub process_alive: bool,
1487 pub github: GithubRunnerObservation,
1488}
1489
1490/// What to do about one attempt found in the journal at startup.
1491#[derive(Debug, Clone, PartialEq, Eq)]
1492pub enum RecoveryDecision {
1493 /// Already cleaned; there is nothing left to do.
1494 Nothing,
1495 /// Terminal but not yet cleaned: remove the runtime directory and mark it
1496 /// `cleaned`.
1497 Clean,
1498 /// The process is alive and this attempt is still ours. `e3`: "an attempt
1499 /// whose process still runs is adopted, not duplicated."
1500 Adopt,
1501 /// Nothing is decidable yet; look again next cycle.
1502 Wait,
1503 /// GitHub is unreachable. Decide nothing destructive during an outage
1504 /// (flow 3.3).
1505 Defer,
1506 /// Move the attempt to this state to match what was observed.
1507 ///
1508 /// **Where the observed state is live, the caller must adopt supervision of
1509 /// the process independently of this decision.** `RecoveryDecision` has no
1510 /// way to express "adopt *and* observe" — the two are separate variants —
1511 /// so `Starting` with `process_alive` and `Registered { busy: false }`
1512 /// returns `Observe(Idle)` and never [`Self::Adopt`], because GitHub is
1513 /// authoritative for the remote status while the local process is
1514 /// authoritative for supervision, and both facts are true at once.
1515 /// `Idle` with `process_alive` and `Registered { busy: true }` returns
1516 /// `Observe(Busy)` for exactly the same reason; it did not always, and the
1517 /// arm's own comment records what that cost.
1518 ///
1519 /// **An `Observe` decision must be applied *and persisted*, or it repeats
1520 /// forever.** The decision is a pure function of the journalled state and
1521 /// the observation, so a caller that moves the in-memory attempt without
1522 /// writing the new state back will read the same stale state on the next
1523 /// pass and be handed the same decision, indefinitely.
1524 Observe(AttemptState),
1525 /// Conclude the attempt with this outcome.
1526 ///
1527 /// **Terminal, and therefore a capacity release.** An attempt is only
1528 /// concluded where the thing it was supervising is already gone; see
1529 /// [`Self::Terminate`] for the case where it is not.
1530 Conclude(AttemptOutcome),
1531 /// The process is still alive but the attempt cannot go on: **stop the
1532 /// process, and only once it is gone record this outcome.**
1533 ///
1534 /// **Two producers, and the payload is what separates them.** `starting`
1535 /// past its registration timeout carries
1536 /// [`FailureReason::RegistrationTimedOut`]; `idle` past its idle timeout,
1537 /// with GitHub still reporting the runner registered and unassigned,
1538 /// carries [`AttemptOutcome::ExitedIdleWithoutWork`] — flow 2.7's surplus
1539 /// runner, which is a normal outcome and not a failure at all. The two
1540 /// share this variant because they need the identical *sequence* — signal,
1541 /// confirm the process is gone, then record — and differ only in what is
1542 /// recorded at the end. A caller that hardcodes either reason will
1543 /// mislabel the other, so the payload is the caller's instruction, not
1544 /// decoration.
1545 ///
1546 /// **Why this is not a [`Self::Conclude`].** `Conclude` moves the attempt to
1547 /// a terminal state, and a terminal attempt no longer
1548 /// [counts against capacity](AttemptState::counts_against_capacity) — so
1549 /// concluding one whose process is still running hands the host back a slot
1550 /// it is still using. The agent then starts a replacement runner beside a
1551 /// live, unregistered one that may yet register and take a job, for an
1552 /// attempt the journal already calls `failed`. There is no
1553 /// `RecoveryDecision` that would have expressed the fix: [`Self::Adopt`]
1554 /// means take over supervision and [`Self::Clean`] means delete a runtime
1555 /// directory, and neither stops anything.
1556 ///
1557 /// **What happens to the capacity slot.** Nothing, until the process is
1558 /// actually gone. The attempt stays in its current, non-terminal state and
1559 /// keeps holding its slot for as long as the runner it started is running,
1560 /// which is the honest answer — the resources are genuinely occupied. The
1561 /// slot comes back at the moment the caller applies the payload through
1562 /// [`RunnerAttempt::conclude`], which it does only after the process has
1563 /// exited.
1564 ///
1565 /// **It is safe to re-derive in one direction, and owes a debt in the
1566 /// other.** The decision is a pure function of the journalled state and the
1567 /// observation, so an agent that dies *before* terminating sees a live
1568 /// process and a larger `elapsed` on the next pass and is handed this same
1569 /// decision again. That half costs nothing and needs nothing.
1570 ///
1571 /// **The other half is a known defect, and it is not harmless.** An agent
1572 /// that terminates the process and then dies *before* writing the outcome
1573 /// observes `process_alive: false` next time and reaches the ordinary
1574 /// [`Self::Conclude`] arm, which records
1575 /// [`FailureReason::ProcessExitedUnexpectedly`]. The process did not exit
1576 /// unexpectedly; this agent killed it. That is precisely the diagnosis
1577 /// [`FailureReason::RegistrationTimedOut`] was split out to prevent — the
1578 /// two reasons send an operator to different places, and this window sends
1579 /// them to a crash investigation (logs, exit code, a corrupt package) for
1580 /// what was a registration failure.
1581 ///
1582 /// **It cannot be fixed here, and the alternative was weighed rather than
1583 /// waved off.** This function's inputs are the journalled state and a
1584 /// [`RecoveryObservation`], and neither carries the fact that separates the
1585 /// two cases: a process this agent killed and a process that crashed on its
1586 /// own present the *same observation*. Using `RegistrationTimedOut` in the
1587 /// dead-process arm once `elapsed >= startup` would close this window at the
1588 /// price of a wider one — every genuine early crash first observed after a
1589 /// restart longer than the startup window would then be reported as a runner
1590 /// that "is running but did not register", to an operator who can see that
1591 /// it is not running. That trades a rare wrong reason for a common false
1592 /// claim about liveness, in the one direction
1593 /// [`FailureReason::ProcessExitedUnexpectedly`]'s own documentation says
1594 /// spends the credibility of every other message this product prints.
1595 /// `tests::no_decision_calls_a_dead_process_live` is what stops that trade
1596 /// being made by accident later.
1597 ///
1598 /// **So the obligation is `e3`'s, and it is a persistence one.** The
1599 /// distinguishing fact exists only at the moment terminate-intent is formed,
1600 /// and the only way to carry it across a crash is to write it down. `e3`
1601 /// journals the intent *before* it signals the process, and on a later pass
1602 /// concludes an attempt it finds so marked with
1603 /// [`FailureReason::TerminatedAfterRegistrationTimeout`] rather than with
1604 /// whatever this function derived from an observation that could not know.
1605 /// Until `e3` does that, the window stands.
1606 ///
1607 /// **Why that closure needs a reason of its own, and not
1608 /// `RegistrationTimedOut`.** On the pass where `e3` reads the mark back, the
1609 /// process is dead — `e3` killed it, which is the whole reason the mark is
1610 /// there. `RegistrationTimedOut` renders as "the runner process is running
1611 /// but did not register", so concluding with it would print exactly the
1612 /// false liveness claim the paragraph above rejects option A for: the same
1613 /// sentence, about a process an operator can see is gone, moved one pass
1614 /// later. Rewording that string instead is not open either —
1615 /// `tests::the_two_starting_failures_read_differently_to_an_operator` pins
1616 /// it on "running", which is correct for the live case it names.
1617 ///
1618 /// [`FailureReason::TerminatedAfterRegistrationTimeout`] is that reason.
1619 /// It is true of a dead process, it says who stopped it and why, and it
1620 /// sends an operator to the networking and configuration fix rather than to
1621 /// a crash investigation — which is the whole distinction
1622 /// [`FailureReason::RegistrationTimedOut`] was split out to draw. Nothing in
1623 /// this function derives it, and nothing should: it is a claim about an
1624 /// action this agent took, not about anything a [`RecoveryObservation`]
1625 /// reports. `tests::no_decision_calls_a_dead_process_live` covers it
1626 /// alongside the other two liveness-claiming reasons, so the day something
1627 /// here does start deriving it — which it legitimately might, once the mark
1628 /// is journalled where this function can read it — it may only do so beside
1629 /// a process the observation says is gone.
1630 ///
1631 /// **The one thing this does not bound** is a process that refuses to die.
1632 /// The slot is held until it does. That is a worse outcome than concluding
1633 /// early only if the runner was never going to register, and a better one in
1634 /// every case where it was — and unlike the early conclusion it cannot
1635 /// oversubscribe the host.
1636 Terminate(AttemptOutcome),
1637}
1638
1639/// Decide what to do about one journalled attempt after a restart, or on any
1640/// reconciliation pass.
1641///
1642/// Time enters only through `clock`, and only as `now - attempt.
1643/// last_state_change_at()`, so the whole function is exercised by advancing a
1644/// [`crate::model::Clock`] the test controls.
1645#[must_use]
1646pub fn recovery_decision(
1647 attempt: &RunnerAttempt,
1648 observation: RecoveryObservation,
1649 timeouts: RecoveryTimeouts,
1650 clock: &dyn Clock,
1651) -> RecoveryDecision {
1652 use AttemptState as S;
1653 use GithubRunnerObservation as G;
1654
1655 let state = attempt.state();
1656
1657 if state == S::Cleaned {
1658 return RecoveryDecision::Nothing;
1659 }
1660 if state.is_terminal() {
1661 return RecoveryDecision::Clean;
1662 }
1663 if observation.github == G::Unreachable {
1664 return RecoveryDecision::Defer;
1665 }
1666
1667 let elapsed = clock.now() - attempt.last_state_change_at();
1668
1669 match state {
1670 S::Allocated | S::JitReceived => {
1671 if observation.process_alive {
1672 // Precedence rule 3's other half: local process state is
1673 // authoritative for a child process this agent owns. `e3`: an
1674 // attempt whose process still runs is adopted, not duplicated.
1675 RecoveryDecision::Adopt
1676 } else {
1677 match observation.github {
1678 // The crash window `e3` exists for. A runner cannot register
1679 // without having received its JIT configuration and started,
1680 // so a registration is proof the attempt got further than
1681 // the journal records: the `starting` write was lost, not
1682 // the attempt. This branch used to consult only
1683 // `process_alive` and the clock, which abandoned a live
1684 // runner as `failed` and left its registration at GitHub
1685 // unreconciled and unremoved.
1686 //
1687 // Each state takes the one *forward* edge out of itself, so
1688 // recovery walks the diagram rather than jumping across it:
1689 // `allocated -> jit_received` (the registration proves the
1690 // JIT configuration arrived) and `jit_received -> starting`.
1691 // `busy` is deliberately not consulted here -- neither state
1692 // has an edge to it, and the next pass, from `starting`, is
1693 // where that distinction becomes legal and is drawn.
1694 //
1695 // This is a genuine recovery path and not a workaround for a
1696 // missing edge: GitHub really is reporting a live runner,
1697 // and each step really did happen. It survived the diagram
1698 // amendment unchanged.
1699 G::Registered { .. } => RecoveryDecision::Observe(if state == S::Allocated {
1700 S::JitReceived
1701 } else {
1702 S::Starting
1703 }),
1704 // Past the handoff deadline with nothing at GitHub and no
1705 // process: the attempt died before registering, and since
1706 // the amendment it can say so directly. It previously had to
1707 // report `NoLegalTransition` and hold its host capacity slot
1708 // for ever.
1709 //
1710 // The two states get different reasons because they know
1711 // different things. At `allocated` no configuration was ever
1712 // recorded as arriving, so the request itself did not
1713 // complete; at `jit_received` one arrived and was never
1714 // claimed, which is flow 4.4's expiry.
1715 G::NotRegistered => {
1716 if elapsed >= timeouts.jit_handoff {
1717 RecoveryDecision::Conclude(AttemptOutcome::failed(
1718 if state == S::Allocated {
1719 FailureReason::JitRequestFailed
1720 } else {
1721 FailureReason::JitExpired
1722 },
1723 ))
1724 } else {
1725 RecoveryDecision::Wait
1726 }
1727 }
1728 G::Unreachable => unreachable!("handled above"),
1729 }
1730 }
1731 }
1732
1733 S::Starting => match observation.github {
1734 // GitHub is authoritative for remote job status, and both of these
1735 // are legal edges out of `starting`.
1736 G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
1737 G::Registered { busy: false } => RecoveryDecision::Observe(S::Idle),
1738 // Three cases, not two, and the third is the one that matters.
1739 //
1740 // A live process inside its startup window is adopted, the same as
1741 // in every other pre-terminal state: `e3` must take over supervision
1742 // rather than start a second runner for the same work.
1743 //
1744 // A process that is *gone* is flow 2's "runner exit before job
1745 // acceptance". Since the amendment that is recordable, so the
1746 // attempt concludes and gives its slot back.
1747 //
1748 // A process that is alive and past its deadline is neither, and
1749 // collapsing it into the second was a real defect: `Conclude` makes
1750 // the attempt terminal, a terminal attempt stops counting against
1751 // capacity, and the host therefore got its slot back while the
1752 // runner was still running -- free to register late and take a job
1753 // for an attempt the journal already called `failed`, beside the
1754 // replacement the freed slot let the agent start. It also read
1755 // `ProcessExitedUnexpectedly` to an operator looking at the process
1756 // in a task manager. `Terminate` says the true thing and keeps the
1757 // slot until the process is actually gone.
1758 //
1759 // The dead-process arm below carries a residual this arm cannot
1760 // close: an agent that terminated the process and died before
1761 // writing the outcome lands there and records
1762 // `ProcessExitedUnexpectedly` for a process it killed itself. The
1763 // fact that would separate the two is not in this function's
1764 // inputs -- a killed process and a crashed one are the same
1765 // observation -- so the fix is `e3` journalling terminate-intent
1766 // before signalling and concluding the marked attempt with
1767 // `FailureReason::TerminatedAfterRegistrationTimeout`, not a
1768 // rearrangement here. That reason exists precisely because the
1769 // process is gone by then, so the closure cannot reuse
1770 // `RegistrationTimedOut` without printing the same false liveness
1771 // claim one pass later. See `RecoveryDecision::Terminate`, which
1772 // names the trade and the owner, and
1773 // `tests::no_decision_calls_a_dead_process_live`, which reds if
1774 // somebody closes it here instead.
1775 G::NotRegistered => {
1776 if !observation.process_alive {
1777 RecoveryDecision::Conclude(AttemptOutcome::failed(
1778 FailureReason::ProcessExitedUnexpectedly,
1779 ))
1780 } else if elapsed < timeouts.startup {
1781 RecoveryDecision::Adopt
1782 } else {
1783 RecoveryDecision::Terminate(AttemptOutcome::failed(
1784 FailureReason::RegistrationTimedOut,
1785 ))
1786 }
1787 }
1788 G::Unreachable => unreachable!("handled above"),
1789 },
1790
1791 // GitHub is consulted **first**, exactly as at `starting` above, and the
1792 // symmetry is load-bearing rather than tidy. This arm used to
1793 // short-circuit on `process_alive` before looking at GitHub, so the same
1794 // conflict -- a live process that GitHub reports as `busy` -- resolved
1795 // one way here and the opposite way one state earlier. What that cost
1796 // was not an inconsistency but a wrong outcome: `Adopt` left the journal
1797 // saying `idle`, so `last_state_change_at` went on pointing at the idle
1798 // entry, and a runner that later crashed *during its job* had its idle
1799 // timeout elapse and was concluded `ExitedIdleWithoutWork` -- the benign
1800 // surplus exit, recorded for a mid-job crash, inverting the one
1801 // distinction this module exists to keep. Nothing downstream could
1802 // catch it either: `required_from` for that outcome is `&[Idle]`, and
1803 // the attempt really was `idle`.
1804 //
1805 // `Observe(Busy)` is returned whether or not the process is alive: the
1806 // caller adopts supervision independently of this decision, which is
1807 // what [`RecoveryDecision::Observe`] already instructs and what
1808 // `starting` has always relied on.
1809 S::Idle => match observation.github {
1810 // GitHub says this runner took a job, and GitHub is authoritative
1811 // for remote job status (precedence rule 3). `idle -> busy` was
1812 // added by the amendment *for this observation*; before this change
1813 // the only arm that could reach it was the one where the process is
1814 // already dead.
1815 G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
1816 // GitHub agrees with the journal, so there is no state to move to.
1817 // A dead process means supervision is lost while the remote
1818 // registration outlived it and needs removing.
1819 //
1820 // A *live* one is adopted only while it is still inside the idle
1821 // timeout. Past it, this is flow 2.7's surplus runner and the agent
1822 // has to end it, because nothing else will: the runner is spawned as
1823 // a bare `Runner.Listener run`, and that process has no idle timeout
1824 // of its own -- it long-polls for an assignment until something
1825 // stops it. This arm used to answer `Adopt` for every elapsed time,
1826 // which is why it never was stopped. A registered, unassigned runner
1827 // then holds its capacity slot and its entry in the target's runner
1828 // settings for as long as the host stays up; one observed in the
1829 // field sat here for 27 hours across two restarts of nothing.
1830 //
1831 // `Terminate`, not `Conclude`: the process is alive, so the slot may
1832 // not be returned until it is gone. The caller signals it, re-reads
1833 // liveness, and only then applies the payload -- the same sequence
1834 // `starting` above relies on, and the reason that decision carries
1835 // its outcome rather than the caller deriving one.
1836 G::Registered { busy: false } => {
1837 if !observation.process_alive {
1838 RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
1839 } else if elapsed >= timeouts.idle {
1840 RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork)
1841 } else {
1842 RecoveryDecision::Adopt
1843 }
1844 }
1845 // Nothing at GitHub. A live process is still ours to supervise.
1846 // Otherwise the surplus case and the crash case, separated by the
1847 // clock: a runner that sat out its whole idle timeout and then
1848 // exited with no registration left behind did what flow 2.7
1849 // describes. One that vanished early did not.
1850 G::NotRegistered => {
1851 if observation.process_alive {
1852 RecoveryDecision::Adopt
1853 } else if elapsed >= timeouts.idle {
1854 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork)
1855 } else {
1856 RecoveryDecision::Conclude(AttemptOutcome::failed(
1857 FailureReason::ProcessExitedUnexpectedly,
1858 ))
1859 }
1860 }
1861 G::Unreachable => unreachable!("handled above"),
1862 },
1863
1864 S::Busy => {
1865 if observation.process_alive {
1866 RecoveryDecision::Adopt
1867 } else {
1868 // `e3`: "an attempt whose process is gone and whose runner is
1869 // unknown to GitHub is `orphaned` and cleaned". The agent never
1870 // reports a job as complete, so a lost supervision is recorded as
1871 // exactly that and not guessed into a success.
1872 RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
1873 }
1874 }
1875
1876 S::Preparing | S::Prepared => RecoveryDecision::Wait,
1877 S::Finished | S::Failed | S::Orphaned | S::Destroying | S::CleanupDeferred | S::Cleaned => {
1878 unreachable!("terminal states are handled above")
1879 }
1880 }
1881}
1882
1883/// How many of these attempts still occupy a host capacity slot.
1884///
1885/// Saturating rather than wrapping: a host cannot hold more than `u16::MAX`
1886/// attempts, and if some caller ever produced that many, reporting the ceiling is
1887/// safe where wrapping to zero would let the allocator start `u16::MAX` more.
1888#[must_use]
1889pub fn active_count<'a>(attempts: impl IntoIterator<Item = &'a RunnerAttempt>) -> u16 {
1890 attempts
1891 .into_iter()
1892 .filter(|a| a.counts_against_capacity())
1893 .fold(0u16, |acc, _| acc.saturating_add(1))
1894}
1895
1896/// How many of these attempts still occupy a slot **and belong to one policy**.
1897///
1898/// The per-policy counterpart of [`active_count`], and the term
1899/// [`crate::capacity::HostAllocator::allocate`] subtracts. The two are not
1900/// interchangeable: [`active_count`] is the host-wide total that bounds D9's
1901/// ceiling, this one is the per-policy figure that bounds D7's, and substituting
1902/// either for the other is silent — the host-wide count in a per-policy slot
1903/// starves every policy but the first, and a zero in place of this one starts a
1904/// duplicate runner on every poll.
1905#[must_use]
1906pub fn active_count_for<'a>(
1907 policy_id: PolicyId,
1908 attempts: impl IntoIterator<Item = &'a RunnerAttempt>,
1909) -> u16 {
1910 attempts
1911 .into_iter()
1912 .filter(|a| a.policy_id == policy_id && a.counts_against_capacity())
1913 .fold(0u16, |acc, _| acc.saturating_add(1))
1914}
1915
1916#[cfg(test)]
1917mod tests {
1918 use super::*;
1919 use crate::model::{Arch, CachePolicy, HostLabel, Os, ScaleTarget};
1920 use crate::policy::{PolicyMode, RoutingLabels};
1921 use std::num::NonZeroU16;
1922
1923 #[derive(Debug)]
1924 struct StubClock(std::sync::Mutex<Timestamp>);
1925
1926 impl StubClock {
1927 fn at(secs: i64) -> Self {
1928 Self(std::sync::Mutex::new(ts(secs)))
1929 }
1930 fn set(&self, secs: i64) {
1931 *self.0.lock().unwrap() = ts(secs);
1932 }
1933 }
1934
1935 impl Clock for StubClock {
1936 fn now(&self) -> Timestamp {
1937 *self.0.lock().unwrap()
1938 }
1939 }
1940
1941 fn ts(secs: i64) -> Timestamp {
1942 chrono::DateTime::from_timestamp(secs, 0).expect("valid timestamp")
1943 }
1944
1945 fn attempt_in(state: AttemptState, entered_at: i64) -> RunnerAttempt {
1946 let mut attempt = RunnerAttempt::allocate(
1947 AttemptId::from_u128(1),
1948 PolicyId::from_u128(1),
1949 "runtime/p/a",
1950 ts(0),
1951 );
1952 // Set the state directly: only possible from inside the module, which is
1953 // exactly why the state-machine tests live here.
1954 attempt.state = state;
1955 attempt.last_state_change_at = ts(entered_at);
1956 attempt
1957 }
1958
1959 /// The same, under a nominated policy, for the per-policy count.
1960 fn attempt_for(policy_id: PolicyId, state: AttemptState, entered_at: i64) -> RunnerAttempt {
1961 let mut attempt = attempt_in(state, entered_at);
1962 attempt.policy_id = policy_id;
1963 attempt
1964 }
1965
1966 fn a_policy(host: HostId, policy_id: PolicyId) -> ScalePolicy {
1967 ScalePolicy::new(
1968 policy_id,
1969 ScaleTarget::repository("o/r").unwrap(),
1970 1,
1971 host,
1972 PolicyMode::autoscale(
1973 RoutingLabels::derive(&HostLabel::new("home").unwrap(), Os::Windows, Arch::X64),
1974 0,
1975 NonZeroU16::new(1).unwrap(),
1976 )
1977 .unwrap(),
1978 CachePolicy::default(),
1979 )
1980 }
1981
1982 // =======================================================================
1983 // The state machine, both directions
1984 // =======================================================================
1985
1986 /// The diagram from `04-subsystem-contracts.md`, transcribed by hand.
1987 ///
1988 /// **Deliberately a second copy of [`AttemptState::LEGAL`]**, for the reason
1989 /// given on `policy::tests::diagram_edges`: a test whose expectation is read
1990 /// out of the constant under test asserts only that the constant equals
1991 /// itself, and would accept any edge someone added to it.
1992 ///
1993 /// ```text
1994 /// allocated -> jit_received -> starting -> idle | busy
1995 /// idle -> busy
1996 /// busy -> idle
1997 /// allocated | jit_received | starting -> failed | orphaned
1998 /// idle | busy -> finished | failed | orphaned
1999 /// finished | failed | orphaned -> cleaned
2000 /// ```
2001 fn diagram_edges() -> Vec<(AttemptState, AttemptState)> {
2002 use AttemptState::*;
2003 // Line 1.
2004 let mut edges = vec![
2005 (Allocated, JitReceived),
2006 (Allocated, Preparing),
2007 (Preparing, Prepared),
2008 (Prepared, JitReceived),
2009 (JitReceived, Starting),
2010 (Starting, Idle),
2011 (Starting, Busy),
2012 ];
2013 // Line 2, added by the 2026-08-21 amendment.
2014 edges.push((Idle, Busy));
2015 // Added when an assignment GitHub reported was shown not to have been
2016 // taken up: the runner's diagnostics hold no job worker.
2017 edges.push((Busy, Idle));
2018 // Line 3, added by the same amendment.
2019 for from in [Allocated, Preparing, Prepared, JitReceived, Starting] {
2020 for to in [Failed, Orphaned] {
2021 edges.push((from, to));
2022 }
2023 }
2024 // Line 4.
2025 for from in [Idle, Busy] {
2026 for to in [Finished, Failed, Orphaned] {
2027 edges.push((from, to));
2028 }
2029 }
2030 // Line 5.
2031 for from in [Finished, Failed, Orphaned] {
2032 edges.push((from, Cleaned));
2033 edges.push((from, Destroying));
2034 }
2035 edges.extend([
2036 (Destroying, CleanupDeferred),
2037 (CleanupDeferred, Destroying),
2038 (Destroying, Cleaned),
2039 ]);
2040 edges
2041 }
2042
2043 #[test]
2044 fn every_attempt_state_transition_is_legal_exactly_where_the_diagram_says() {
2045 let expected = diagram_edges();
2046 assert_eq!(
2047 expected.len(),
2048 34,
2049 "the transcription itself changed; check it against the diagram"
2050 );
2051
2052 let mut legal_seen = 0usize;
2053 let mut illegal_seen = 0usize;
2054
2055 for from in AttemptState::ALL {
2056 for to in AttemptState::ALL {
2057 let expected_legal = expected.contains(&(from, to));
2058 let mut attempt = attempt_in(from, 0);
2059 let result = attempt.move_to(to, ts(10));
2060
2061 if expected_legal {
2062 legal_seen += 1;
2063 assert!(
2064 result.is_ok(),
2065 "{from} -> {to} is in the diagram and must be accepted"
2066 );
2067 assert_eq!(attempt.state(), to);
2068 assert_eq!(attempt.last_state_change_at(), ts(10));
2069 } else {
2070 illegal_seen += 1;
2071 assert!(
2072 matches!(result, Err(AttemptError::IllegalTransition { .. })),
2073 "{from} -> {to} is not in the diagram and must be rejected"
2074 );
2075 assert_eq!(
2076 attempt.state(),
2077 from,
2078 "a refused transition changes nothing"
2079 );
2080 assert_eq!(attempt.last_state_change_at(), ts(0));
2081 }
2082 }
2083 }
2084
2085 assert_eq!(legal_seen, 34);
2086 assert_eq!(illegal_seen, 13 * 13 - 34);
2087
2088 // And the published constant matches the transcription.
2089 let mut published = AttemptState::LEGAL.to_vec();
2090 let mut transcribed = expected;
2091 published.sort_unstable();
2092 transcribed.sort_unstable();
2093 assert_eq!(published, transcribed);
2094 }
2095
2096 #[test]
2097 fn every_live_state_has_a_terminal_edge_so_no_attempt_can_hold_a_slot_forever() {
2098 // The operational half of the amendment, asserted as the property it is
2099 // rather than as a list of edges. An attempt occupies a host capacity
2100 // slot for exactly as long as it is non-terminal, so a live state with
2101 // no terminal edge is a state an attempt can be stranded in -- which is
2102 // how two failed JIT requests wedged a `host_capacity: 2` host into
2103 // starting zero runners, with no error and no cleanup path.
2104 for state in AttemptState::ALL {
2105 if state.is_terminal() {
2106 continue;
2107 }
2108 // Read through `active_count`, which is the function the
2109 // reconciliation formula actually calls, rather than through
2110 // `state.counts_against_capacity()`. The latter is defined as
2111 // `!self.is_terminal()`, so asserting it one line under
2112 // `if state.is_terminal() { continue; }` is `assert!(true)` and
2113 // catches nothing; this asserts the same property against a second
2114 // reader that can regress on its own -- an inverted or mistyped
2115 // filter predicate in `active_count` fails here and nowhere else in
2116 // this test.
2117 assert_eq!(
2118 active_count([&attempt_in(state, 0)]),
2119 1,
2120 "{state} is non-terminal, so the reconciliation formula must \
2121 still be subtracting it"
2122 );
2123 assert!(
2124 state.can_transition_to(AttemptState::Failed)
2125 || state.can_transition_to(AttemptState::Finished),
2126 "{state} has no way to conclude, so an attempt in it holds a host \
2127 capacity slot permanently"
2128 );
2129 assert!(
2130 state.can_transition_to(AttemptState::Orphaned),
2131 "{state} has no orphan edge, so an attempt found in it after a \
2132 restart cannot be recorded"
2133 );
2134 assert!(
2135 AttemptState::CONCLUDABLE_FROM.contains(&state),
2136 "{state} is live but missing from CONCLUDABLE_FROM"
2137 );
2138 }
2139 assert_eq!(
2140 AttemptState::CONCLUDABLE_FROM.len(),
2141 AttemptState::ALL
2142 .iter()
2143 .filter(|s| !s.is_terminal())
2144 .count(),
2145 "CONCLUDABLE_FROM must list every non-terminal state and nothing else"
2146 );
2147 }
2148
2149 #[test]
2150 fn an_attempt_state_cannot_transition_to_itself() {
2151 for state in AttemptState::ALL {
2152 assert!(
2153 !state.can_transition_to(state),
2154 "{state} -> {state} is not an edge in the diagram"
2155 );
2156 }
2157 }
2158
2159 #[test]
2160 fn cleaned_is_absorbing() {
2161 for to in AttemptState::ALL {
2162 assert!(
2163 !AttemptState::Cleaned.can_transition_to(to),
2164 "cleaned -> {to} must not exist; a cleaned runtime directory is gone"
2165 );
2166 }
2167 }
2168
2169 #[test]
2170 fn the_documented_happy_path_walks_allocated_to_cleaned() {
2171 // `e3`: "A full attempt runs allocated -> jit_received -> starting ->
2172 // busy -> finished -> cleaned".
2173 let mut attempt = RunnerAttempt::allocate(
2174 AttemptId::from_u128(1),
2175 PolicyId::from_u128(1),
2176 "runtime/p/a",
2177 ts(0),
2178 );
2179 assert_eq!(attempt.state(), AttemptState::Allocated);
2180 assert!(attempt.counts_against_capacity());
2181
2182 attempt.jit_received(ts(1)).unwrap();
2183 attempt.started(4242, ts(2)).unwrap();
2184 assert_eq!(attempt.process_id(), Some(4242));
2185
2186 attempt.assigned_job(73, ts(3)).unwrap();
2187 assert_eq!(attempt.state(), AttemptState::Busy);
2188 assert_eq!(attempt.github_runner_id(), Some(73));
2189
2190 attempt
2191 .conclude(AttemptOutcome::CompletedJob, ts(9))
2192 .unwrap();
2193 assert_eq!(attempt.state(), AttemptState::Finished);
2194 assert_eq!(attempt.terminal_at(), Some(ts(9)));
2195 assert!(!attempt.counts_against_capacity());
2196
2197 attempt.clean(ts(10)).unwrap();
2198 assert_eq!(attempt.state(), AttemptState::Cleaned);
2199 }
2200
2201 // =======================================================================
2202 // Outcome: the surplus attempt
2203 // =======================================================================
2204
2205 #[test]
2206 fn an_attempt_that_exits_idle_without_work_is_terminal_and_not_a_failure() {
2207 // `b1`: "An attempt that exits idle without work reaches a terminal state
2208 // carrying an outcome distinguishable from a failure."
2209 let mut surplus = RunnerAttempt::allocate(
2210 AttemptId::from_u128(1),
2211 PolicyId::from_u128(1),
2212 "runtime/p/a",
2213 ts(0),
2214 );
2215 surplus.jit_received(ts(1)).unwrap();
2216 surplus.started(1, ts(2)).unwrap();
2217 surplus.registered_idle(73, ts(3)).unwrap();
2218 surplus
2219 .conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(300))
2220 .unwrap();
2221
2222 assert!(surplus.is_terminal());
2223 assert_eq!(surplus.state(), AttemptState::Finished);
2224
2225 let outcome = surplus.outcome().expect("a terminal attempt carries one");
2226 assert!(outcome.is_idle_exit());
2227 assert!(
2228 !outcome.is_failure(),
2229 "the surplus runner is an accepted, bounded cost of having no job \
2230 reservation -- presenting it as a fault sends an operator hunting \
2231 something that did not happen"
2232 );
2233 assert!(!outcome.ran_a_job());
2234
2235 // And it is cleaned like any other terminal attempt.
2236 surplus.clean(ts(301)).unwrap();
2237 assert_eq!(surplus.state(), AttemptState::Cleaned);
2238 }
2239
2240 #[test]
2241 fn an_idle_exit_is_distinguishable_from_a_failure_and_from_a_completed_job() {
2242 // The three outcomes `g2` must render apart.
2243 let idle = AttemptOutcome::ExitedIdleWithoutWork;
2244 let failed = AttemptOutcome::failed(FailureReason::ProcessStartFailed);
2245 let done = AttemptOutcome::CompletedJob;
2246
2247 assert_ne!(idle, failed);
2248 assert_ne!(idle, done);
2249 assert_ne!(failed, done);
2250
2251 assert_eq!(idle.terminal_state(), AttemptState::Finished);
2252 assert_eq!(done.terminal_state(), AttemptState::Finished);
2253 assert_eq!(failed.terminal_state(), AttemptState::Failed);
2254 assert_eq!(
2255 AttemptOutcome::Orphaned.terminal_state(),
2256 AttemptState::Orphaned
2257 );
2258
2259 // `finished` alone does not say which happened; the outcome does. This is
2260 // the assertion that would fail if someone dropped the outcome field and
2261 // let `g2` infer from the state.
2262 assert_eq!(idle.terminal_state(), done.terminal_state());
2263 assert_ne!(idle, done);
2264
2265 assert!(!idle.is_failure());
2266 assert!(failed.is_failure());
2267 assert!(AttemptOutcome::Orphaned.is_failure());
2268 }
2269
2270 #[test]
2271 fn an_outcome_cannot_be_recorded_from_a_state_that_could_not_produce_it() {
2272 // A runner that never got a job cannot have run one.
2273 let mut idle = attempt_in(AttemptState::Idle, 0);
2274 assert!(matches!(
2275 idle.conclude(AttemptOutcome::CompletedJob, ts(1)),
2276 Err(AttemptError::OutcomeUnreachable {
2277 from: AttemptState::Idle,
2278 ..
2279 })
2280 ));
2281 assert_eq!(idle.state(), AttemptState::Idle);
2282 assert!(idle.outcome().is_none());
2283
2284 // And a runner that was executing a job did not exit idle without work.
2285 let mut busy = attempt_in(AttemptState::Busy, 0);
2286 assert!(matches!(
2287 busy.conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(1)),
2288 Err(AttemptError::OutcomeUnreachable {
2289 from: AttemptState::Busy,
2290 ..
2291 })
2292 ));
2293
2294 // Failure and orphaning are reachable from every live state, including
2295 // the three pre-registration ones the amendment gave terminal edges.
2296 for state in AttemptState::CONCLUDABLE_FROM {
2297 attempt_in(*state, 0)
2298 .conclude(AttemptOutcome::failed(FailureReason::JitExpired), ts(1))
2299 .unwrap_or_else(|e| panic!("{state} must be able to fail: {e}"));
2300 attempt_in(*state, 0)
2301 .conclude(AttemptOutcome::Orphaned, ts(1))
2302 .unwrap_or_else(|e| panic!("{state} must be able to orphan: {e}"));
2303 }
2304
2305 // And from a terminal state nothing can be concluded at all -- a second
2306 // conclusion would overwrite the first.
2307 for state in AttemptState::ALL.iter().filter(|s| s.is_terminal()) {
2308 assert!(
2309 attempt_in(*state, 0)
2310 .conclude(AttemptOutcome::Orphaned, ts(1))
2311 .is_err(),
2312 "{state} has already concluded"
2313 );
2314 }
2315 }
2316
2317 /// The earliest state at which the agent learns each fact, from flow 2's own
2318 /// step order.
2319 ///
2320 /// **This match is exhaustive and carries no wildcard, and that is the whole
2321 /// mechanism.** The assertion it replaced read
2322 /// `assert_eq!(cases.len(), 7, "FailureReason has seven variants; ...")`,
2323 /// and `cases.len()` on a `[_; 7]` is the compile-time constant `7`: the
2324 /// assertion was `7 == 7` and could not fail. Measured on the code as it
2325 /// stood, adding an eighth variant produced exactly one error — `E0004` from
2326 /// `Display`'s match — and once that arm was written the suite was green
2327 /// with the new variant unreachable and untested. Here, a tenth variant
2328 /// stops this file compiling until somebody says which state produces it,
2329 /// and `all_failure_reasons_are_reachable_from_the_state_that_produces_them`
2330 /// then proves the answer.
2331 ///
2332 /// Re-measured on the ninth. Declaring
2333 /// `TerminatedAfterRegistrationTimeout` and changing nothing else produced
2334 /// exactly one error, `E0004` at `Display`'s match; writing that arm
2335 /// produced exactly one more, `E0004` here; writing this one left
2336 /// `cargo check --all-targets --workspace` clean. Two stops, in that order,
2337 /// and no third — which is what the note above each of them promises.
2338 fn earliest_state_producing(reason: &FailureReason) -> AttemptState {
2339 // The second place a new variant stops the compiler, and the last one.
2340 // `FailureReason::ALL` is a const array, so it errors nowhere at all:
2341 // add the variant to `ALL` and to the `cases` table below too, or the
2342 // suite goes green with it untested.
2343 match reason {
2344 // Step 5: the package is verified before the JIT request is made.
2345 FailureReason::RunnerPackageUnverified => AttemptState::Allocated,
2346 // Step 5: `generate-jitconfig` did not return a configuration.
2347 FailureReason::JitRequestFailed => AttemptState::Allocated,
2348 // Flow 4.4: a configuration arrived and was never claimed.
2349 FailureReason::JitExpired => AttemptState::JitReceived,
2350 // Step 6: the child process could not be spawned.
2351 FailureReason::ProcessStartFailed => AttemptState::JitReceived,
2352 // Edge case 7: GitHub refuses the registration on version grounds.
2353 FailureReason::RunnerVersionRejected => AttemptState::Starting,
2354 // Flow 2's "runner exit before job acceptance".
2355 FailureReason::ProcessExitedUnexpectedly => AttemptState::Starting,
2356 // The live-but-unregistered process past its startup deadline; see
2357 // the `S::Starting` / `G::NotRegistered` arm of `recovery_decision`.
2358 FailureReason::RegistrationTimedOut => AttemptState::Starting,
2359 // The same attempt one step later: still `starting`, because the
2360 // agent acts on the `Terminate` payload without moving the state
2361 // first, and the process it signalled is the one that never
2362 // registered.
2363 FailureReason::TerminatedAfterRegistrationTimeout => AttemptState::Starting,
2364 FailureReason::IsolationProvider(_) => AttemptState::Allocated,
2365 // `WorkspaceCleanupDeferred` is never recorded as an outcome (see
2366 // the variant), so the only claim made for it is the one `Other`
2367 // makes: the type would accept it.
2368 FailureReason::WorkspaceCleanupDeferred
2369 | FailureReason::HostUnfit
2370 | FailureReason::Other(_) => AttemptState::Busy,
2371 }
2372 }
2373
2374 #[test]
2375 fn all_failure_reasons_are_reachable_from_the_state_that_produces_them() {
2376 // `03-control-flows.md` flow 2 names most of these by name as conditions
2377 // the agent must record. Before the diagram amendment five were
2378 // unreachable: each occurs at a pre-registration state, and `allocated`,
2379 // `jit_received` and `starting` had no terminal edge, so `conclude`
2380 // answered `OutcomeUnreachable` from every state that could actually
2381 // have produced them. A `FailureReason` variant that nothing can reach
2382 // is a variant `g2` renders a match arm for and no test can cover.
2383 //
2384 // The table is written out rather than derived so each pairing carries
2385 // its reason; `earliest_state_producing` above is what makes a new
2386 // variant a compile error, and the two are cross-checked below.
2387 let cases: [(FailureReason, AttemptState); 12] = [
2388 // Step 5: the package is verified before the JIT request is made.
2389 (
2390 FailureReason::RunnerPackageUnverified,
2391 AttemptState::Allocated,
2392 ),
2393 // Step 5: `generate-jitconfig` did not return a configuration.
2394 (FailureReason::JitRequestFailed, AttemptState::Allocated),
2395 // Flow 4.4: a configuration arrived and was never claimed.
2396 (FailureReason::JitExpired, AttemptState::JitReceived),
2397 // Step 6: the child process could not be spawned.
2398 (FailureReason::ProcessStartFailed, AttemptState::JitReceived),
2399 // Edge case 7: GitHub refuses the registration on version grounds.
2400 (FailureReason::RunnerVersionRejected, AttemptState::Starting),
2401 // Flow 2's "runner exit before job acceptance".
2402 (
2403 FailureReason::ProcessExitedUnexpectedly,
2404 AttemptState::Starting,
2405 ),
2406 // A live runner that never reached GitHub inside its startup window.
2407 (FailureReason::RegistrationTimedOut, AttemptState::Starting),
2408 // The same runner after `e3` acted on the `Terminate` payload: the
2409 // attempt never left `starting`, so this is where it concludes from.
2410 (
2411 FailureReason::TerminatedAfterRegistrationTimeout,
2412 AttemptState::Starting,
2413 ),
2414 (
2415 FailureReason::IsolationProvider(IsolationProviderFailure::NotInstalled),
2416 AttemptState::Allocated,
2417 ),
2418 (FailureReason::WorkspaceCleanupDeferred, AttemptState::Busy),
2419 (FailureReason::HostUnfit, AttemptState::Busy),
2420 (
2421 FailureReason::Other("a reason b1 did not anticipate".into()),
2422 AttemptState::Busy,
2423 ),
2424 ];
2425
2426 // Every variant is covered exactly once. Three separate things have to
2427 // hold, and none of them is a length compared against itself:
2428 //
2429 // * `FailureReason::ALL` and this table are the same size, so a variant
2430 // added to one and not the other is caught;
2431 // * every variant in `ALL` appears here -- by discriminant, because
2432 // `Other`'s payload differs between the two lists;
2433 // * each pairing agrees with `earliest_state_producing`, whose
2434 // wildcard-free match is what stops this file compiling when a
2435 // variant is added at all.
2436 assert_eq!(
2437 cases.len(),
2438 FailureReason::ALL.len(),
2439 "every FailureReason variant needs a state it is reachable from"
2440 );
2441 for listed in FailureReason::ALL {
2442 assert!(
2443 cases.iter().any(|(reason, _)| {
2444 std::mem::discriminant(reason) == std::mem::discriminant(&listed)
2445 }),
2446 "{listed:?} is in FailureReason::ALL but has no case here"
2447 );
2448 }
2449 for (reason, from) in &cases {
2450 assert_eq!(
2451 earliest_state_producing(reason),
2452 *from,
2453 "{reason:?}: the table and the exhaustive match disagree"
2454 );
2455 }
2456
2457 for (reason, from) in cases {
2458 let mut attempt = attempt_in(from, 0);
2459 attempt
2460 .conclude(AttemptOutcome::failed(reason.clone()), ts(5))
2461 .unwrap_or_else(|e| panic!("{reason:?} must be recordable from {from}, got {e}"));
2462
2463 assert_eq!(attempt.state(), AttemptState::Failed);
2464 assert_eq!(attempt.terminal_at(), Some(ts(5)));
2465 assert_eq!(
2466 attempt.outcome(),
2467 Some(&AttemptOutcome::failed(reason.clone()))
2468 );
2469 assert!(
2470 !attempt.counts_against_capacity(),
2471 "{reason:?} from {from} must give the host capacity slot back; \
2472 that is what the amendment was for"
2473 );
2474
2475 // And it survives the persistence gate, so the row `b2` writes for
2476 // it can be read back.
2477 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2478 .unwrap_or_else(|e| panic!("{reason:?} from {from} must reload: {e}"));
2479 assert_eq!(restored, attempt);
2480 }
2481 }
2482
2483 #[test]
2484 fn isolation_failure_categories_round_trip_without_free_text() {
2485 let categories = [
2486 IsolationProviderFailure::NotInstalled,
2487 IsolationProviderFailure::PermissionDenied,
2488 IsolationProviderFailure::ImageUnavailableOrIncompatible,
2489 IsolationProviderFailure::DiskQuotaUnavailable,
2490 IsolationProviderFailure::RuntimeOperationFailed,
2491 ];
2492 let mut rendered = std::collections::BTreeSet::new();
2493 for category in categories {
2494 let reason = FailureReason::IsolationProvider(category);
2495 let journal = serde_json::to_string(&reason).unwrap();
2496 assert!(!journal.contains("other"));
2497 assert!(!journal.contains("ghp_secret"));
2498 assert_eq!(
2499 serde_json::from_str::<FailureReason>(&journal).unwrap(),
2500 reason
2501 );
2502 rendered.insert(reason.to_string());
2503 }
2504 assert_eq!(rendered.len(), categories.len());
2505 }
2506
2507 // =======================================================================
2508 // Busy protection
2509 // =======================================================================
2510
2511 #[test]
2512 fn a_busy_attempt_cannot_be_cleaned() {
2513 // `04-subsystem-contracts.md`: "`busy` cannot transition to cleanup due
2514 // to a scale-down request."
2515 let mut busy = attempt_in(AttemptState::Busy, 0);
2516 let err = busy.clean(ts(1)).unwrap_err();
2517 assert_eq!(
2518 err,
2519 AttemptError::BusyCannotBeCleaned,
2520 "the refusal must be named, not a generic transition error, so that a \
2521 scale-down that tried it is legible in a log"
2522 );
2523 assert_eq!(busy.state(), AttemptState::Busy);
2524 assert!(busy.counts_against_capacity());
2525 }
2526
2527 #[test]
2528 fn only_a_terminal_attempt_can_be_cleaned() {
2529 for state in AttemptState::ALL {
2530 let mut attempt = attempt_in(state, 0);
2531 let result = attempt.clean(ts(1));
2532 if state.is_concluded() {
2533 assert!(result.is_ok(), "{state} is terminal and must be cleanable");
2534 assert_eq!(attempt.state(), AttemptState::Cleaned);
2535 } else {
2536 assert!(
2537 result.is_err(),
2538 "{state} is not terminal and must not be cleanable"
2539 );
2540 assert_eq!(attempt.state(), state);
2541 }
2542 }
2543 }
2544
2545 #[test]
2546 fn capacity_is_reclaimed_exactly_at_the_terminal_states() {
2547 for state in AttemptState::ALL {
2548 assert_eq!(
2549 attempt_in(state, 0).counts_against_capacity(),
2550 !state.is_terminal(),
2551 "{state}"
2552 );
2553 }
2554
2555 let attempts = vec![
2556 attempt_in(AttemptState::Allocated, 0),
2557 attempt_in(AttemptState::Starting, 0),
2558 attempt_in(AttemptState::Busy, 0),
2559 attempt_in(AttemptState::Finished, 0),
2560 attempt_in(AttemptState::Cleaned, 0),
2561 ];
2562 assert_eq!(active_count(&attempts), 3);
2563 }
2564
2565 #[test]
2566 fn the_per_policy_active_count_is_not_the_host_wide_one() {
2567 // Substituting either for the other is silent, so the difference is
2568 // pinned rather than left to the reader of two similar names.
2569 let mine = PolicyId::from_u128(1);
2570 let theirs = PolicyId::from_u128(2);
2571
2572 let mut attempts = vec![
2573 attempt_for(mine, AttemptState::Starting, 0),
2574 attempt_for(mine, AttemptState::Busy, 0),
2575 attempt_for(theirs, AttemptState::Busy, 0),
2576 attempt_for(theirs, AttemptState::Idle, 0),
2577 attempt_for(theirs, AttemptState::Allocated, 0),
2578 ];
2579
2580 assert_eq!(active_count(&attempts), 5, "the host-wide total, for D9");
2581 assert_eq!(active_count_for(mine, &attempts), 2, "this policy, for D7");
2582 assert_eq!(active_count_for(theirs, &attempts), 3);
2583 assert_eq!(
2584 active_count_for(PolicyId::from_u128(3), &attempts),
2585 0,
2586 "a policy with nothing in flight"
2587 );
2588
2589 // Terminal attempts drop out of the per-policy count on the same rule as
2590 // the host-wide one.
2591 attempts.push(attempt_for(mine, AttemptState::Finished, 0));
2592 attempts.push(attempt_for(mine, AttemptState::Cleaned, 0));
2593 assert_eq!(active_count_for(mine, &attempts), 2);
2594 }
2595
2596 // =======================================================================
2597 // Workspace allocation (D3, D5, D6)
2598 // =======================================================================
2599
2600 fn slot(n: u16) -> AttemptWorkspace {
2601 AttemptWorkspace::persistent_slot(std::num::NonZeroU16::new(n).expect("a positive slot"))
2602 }
2603
2604 /// The unremarkable outcome for a terminal state, so a fixture does not have
2605 /// to restate the state/outcome pairing the loader enforces.
2606 fn terminal_outcome(state: AttemptState) -> AttemptOutcome {
2607 match state {
2608 AttemptState::Failed => {
2609 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
2610 }
2611 AttemptState::Orphaned => AttemptOutcome::Orphaned,
2612 _ => AttemptOutcome::CompletedJob,
2613 }
2614 }
2615
2616 #[test]
2617 fn the_ordinary_constructor_still_allocates_a_disposable_workspace() {
2618 // D3: disposable mode remains the default, so every existing caller of
2619 // `allocate` keeps the cleanup behaviour it had.
2620 let attempt = RunnerAttempt::allocate(
2621 AttemptId::from_u128(1),
2622 PolicyId::from_u128(1),
2623 "runtime/p/a",
2624 ts(0),
2625 );
2626 assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
2627 assert_eq!(attempt.workspace().slot(), None);
2628 assert!(!attempt.holds_slot_lease());
2629 assert_eq!(
2630 attempt.to_persisted().workspace_kind,
2631 WorkspaceKind::Ephemeral
2632 );
2633 assert_eq!(attempt.to_persisted().workspace_slot, None);
2634 }
2635
2636 #[test]
2637 fn a_persistent_attempt_journals_the_slot_it_leased() {
2638 let attempt = RunnerAttempt::allocate_in(
2639 AttemptId::from_u128(1),
2640 PolicyId::from_u128(1),
2641 "/srv/rman/acme/s2",
2642 slot(2),
2643 ts(0),
2644 );
2645 assert_eq!(attempt.workspace(), slot(2));
2646 assert_eq!(attempt.workspace().slot_number(), Some(2));
2647 assert_eq!(
2648 attempt.workspace().slot_directory_name().as_deref(),
2649 Some("s2")
2650 );
2651 assert_eq!(attempt.runtime_path(), Path::new("/srv/rman/acme/s2"));
2652 assert!(attempt.holds_slot_lease());
2653 }
2654
2655 #[test]
2656 fn the_workspace_kind_and_slot_do_not_change_after_allocation() {
2657 // `02-target-architecture.md`: "Neither may change after allocation."
2658 // There is no setter, so the property is proved by driving the whole
2659 // lifecycle and reading the value back at every step.
2660 let mut attempt = RunnerAttempt::allocate_in(
2661 AttemptId::from_u128(1),
2662 PolicyId::from_u128(1),
2663 "/srv/rman/acme/s1",
2664 slot(1),
2665 ts(0),
2666 );
2667 attempt
2668 .jit_received(ts(1))
2669 .expect("allocated -> jit_received");
2670 assert_eq!(attempt.workspace(), slot(1));
2671 attempt
2672 .started(4242, ts(2))
2673 .expect("jit_received -> starting");
2674 assert_eq!(attempt.workspace(), slot(1));
2675 attempt
2676 .registered_idle(73, ts(3))
2677 .expect("starting -> idle");
2678 assert_eq!(attempt.workspace(), slot(1));
2679 attempt.assigned_job(73, ts(4)).expect("idle -> busy");
2680 assert_eq!(attempt.workspace(), slot(1));
2681 attempt
2682 .conclude(AttemptOutcome::CompletedJob, ts(5))
2683 .expect("busy -> finished");
2684 assert_eq!(attempt.workspace(), slot(1));
2685 assert!(
2686 attempt.holds_slot_lease(),
2687 "a terminal attempt still holds its slot until cleanup succeeds"
2688 );
2689
2690 attempt.clean(ts(6)).expect("finished -> cleaned");
2691 assert_eq!(attempt.workspace(), slot(1));
2692 assert!(
2693 !attempt.holds_slot_lease(),
2694 "a cleaned attempt releases the slot for reuse"
2695 );
2696 }
2697
2698 #[test]
2699 fn an_uncleaned_terminal_attempt_keeps_its_lease() {
2700 // `04-security-recovery.md`: "Attempt remains not-cleaned and continues
2701 // to hold the slot through the unique lease index".
2702 for state in [
2703 AttemptState::Finished,
2704 AttemptState::Failed,
2705 AttemptState::Orphaned,
2706 ] {
2707 let mut fields = row(state, Some(terminal_outcome(state)));
2708 fields.workspace_kind = WorkspaceKind::Persistent;
2709 fields.workspace_slot = Some(3);
2710 let attempt = RunnerAttempt::from_persisted(fields).expect("a row the domain accepts");
2711 assert!(attempt.holds_slot_lease(), "state {state}");
2712 }
2713 }
2714
2715 #[test]
2716 fn a_workspace_allocation_round_trips_through_the_journal() {
2717 for workspace in [AttemptWorkspace::Ephemeral, slot(1), slot(u16::MAX)] {
2718 let attempt = RunnerAttempt::allocate_in(
2719 AttemptId::from_u128(1),
2720 PolicyId::from_u128(1),
2721 "runtime/p/a",
2722 workspace,
2723 ts(0),
2724 );
2725 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2726 .expect("a row this crate wrote must load");
2727 assert_eq!(restored, attempt);
2728 assert_eq!(restored.workspace(), workspace);
2729 }
2730 }
2731
2732 #[test]
2733 fn a_journal_row_whose_workspace_columns_disagree_is_rejected() {
2734 // The kind decides which cleanup algorithm is legal, so an undecidable
2735 // pair must fail closed rather than fall back to the destructive branch.
2736 let mut persistent_without_slot = row(AttemptState::Allocated, None);
2737 persistent_without_slot.workspace_kind = WorkspaceKind::Persistent;
2738 assert_eq!(
2739 RunnerAttempt::from_persisted(persistent_without_slot),
2740 Err(AttemptError::Workspace(
2741 WorkspaceError::PersistentWithoutSlot
2742 ))
2743 );
2744
2745 let mut zero_slot = row(AttemptState::Allocated, None);
2746 zero_slot.workspace_kind = WorkspaceKind::Persistent;
2747 zero_slot.workspace_slot = Some(0);
2748 assert_eq!(
2749 RunnerAttempt::from_persisted(zero_slot),
2750 Err(AttemptError::Workspace(WorkspaceError::SlotNotPositive))
2751 );
2752
2753 let mut ephemeral_with_slot = row(AttemptState::Allocated, None);
2754 ephemeral_with_slot.workspace_slot = Some(1);
2755 assert_eq!(
2756 RunnerAttempt::from_persisted(ephemeral_with_slot),
2757 Err(AttemptError::Workspace(WorkspaceError::EphemeralWithSlot {
2758 slot: 1
2759 }))
2760 );
2761 }
2762
2763 #[test]
2764 fn an_attempt_serialises_its_workspace_without_credentials() {
2765 let attempt = RunnerAttempt::allocate_in(
2766 AttemptId::from_u128(1),
2767 PolicyId::from_u128(1),
2768 "/srv/rman/acme/s2",
2769 slot(2),
2770 ts(0),
2771 );
2772 let encoded = serde_json::to_string(&attempt).expect("serialisable");
2773 let decoded: RunnerAttempt = serde_json::from_str(&encoded).expect("deserialisable");
2774 assert_eq!(decoded, attempt);
2775 for needle in ["token", "secret", "jitconfig", "password"] {
2776 assert!(
2777 !encoded.to_ascii_lowercase().contains(needle),
2778 "an attempt leaked {needle:?}: {encoded}"
2779 );
2780 }
2781 }
2782
2783 // =======================================================================
2784 // Persistence gate
2785 // =======================================================================
2786
2787 /// A journal row the domain accepts, for a test to spoil one field of.
2788 fn row(state: AttemptState, outcome: Option<AttemptOutcome>) -> PersistedAttempt {
2789 PersistedAttempt {
2790 id: AttemptId::from_u128(1),
2791 policy_id: PolicyId::from_u128(1),
2792 github_runner_id: Some(73),
2793 state,
2794 outcome,
2795 process_id: Some(9),
2796 execution: AttemptExecution::Native {
2797 process_id: Some(9),
2798 },
2799 runtime_path: "runtime/p/a".into(),
2800 workspace_kind: WorkspaceKind::Ephemeral,
2801 workspace_slot: None,
2802 created_at: ts(0),
2803 terminal_at: state.is_terminal().then(|| ts(9)),
2804 last_state_change_at: ts(9),
2805 }
2806 }
2807
2808 #[test]
2809 fn a_hand_edited_journal_row_with_an_impossible_outcome_is_rejected() {
2810 assert!(
2811 RunnerAttempt::from_persisted(row(
2812 AttemptState::Finished,
2813 Some(AttemptOutcome::ExitedIdleWithoutWork)
2814 ))
2815 .is_ok()
2816 );
2817
2818 // Terminal with no outcome.
2819 assert!(matches!(
2820 RunnerAttempt::from_persisted(row(AttemptState::Failed, None)),
2821 Err(AttemptError::TerminalWithoutOutcome { .. })
2822 ));
2823
2824 // Non-terminal carrying one.
2825 assert!(matches!(
2826 RunnerAttempt::from_persisted(row(
2827 AttemptState::Busy,
2828 Some(AttemptOutcome::CompletedJob)
2829 )),
2830 Err(AttemptError::NonTerminalWithOutcome { .. })
2831 ));
2832
2833 // A `failed` row that claims to have run a job.
2834 assert!(matches!(
2835 RunnerAttempt::from_persisted(row(
2836 AttemptState::Failed,
2837 Some(AttemptOutcome::CompletedJob)
2838 )),
2839 Err(AttemptError::OutcomeStateMismatch { .. })
2840 ));
2841
2842 // `cleaned` keeps whichever outcome preceded it.
2843 assert!(
2844 RunnerAttempt::from_persisted(row(
2845 AttemptState::Cleaned,
2846 Some(AttemptOutcome::Orphaned)
2847 ))
2848 .is_ok()
2849 );
2850 }
2851
2852 #[test]
2853 fn a_hand_edited_journal_row_with_an_impossible_terminal_at_is_rejected() {
2854 // `terminal_at` is on the same footing as `outcome`: `conclude` is the
2855 // only writer of either and sets them together, so a row where
2856 // `state.is_terminal()` and `terminal_at.is_some()` disagree is one this
2857 // crate cannot have produced. Before this gate existed, a `finished` row
2858 // with `terminal_at: None` loaded cleanly and every reader of
2859 // `terminal_at()` saw an attempt that had never concluded.
2860 let mut terminal_without = row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
2861 terminal_without.terminal_at = None;
2862 assert!(
2863 matches!(
2864 RunnerAttempt::from_persisted(terminal_without),
2865 Err(AttemptError::TerminalWithoutTimestamp {
2866 state: AttemptState::Finished
2867 })
2868 ),
2869 "a terminal row with no terminal_at must be refused, exactly as a \
2870 terminal row with no outcome is"
2871 );
2872
2873 // The other direction: a live attempt that claims to have concluded.
2874 let mut live_with = row(AttemptState::Busy, None);
2875 live_with.terminal_at = Some(ts(9));
2876 assert!(matches!(
2877 RunnerAttempt::from_persisted(live_with),
2878 Err(AttemptError::NonTerminalWithTimestamp {
2879 state: AttemptState::Busy
2880 })
2881 ));
2882
2883 // `cleaned` follows a concluded state, so it keeps that state's
2884 // timestamp and is not a special case.
2885 let mut cleaned = row(AttemptState::Cleaned, Some(AttemptOutcome::Orphaned));
2886 cleaned.terminal_at = Some(ts(4));
2887 assert!(RunnerAttempt::from_persisted(cleaned).is_ok());
2888 }
2889
2890 #[test]
2891 fn a_hand_edited_journal_row_whose_timestamps_run_backwards_is_rejected() {
2892 // Presence was already gated; ordering was not, so a row saying an
2893 // attempt concluded a hundred seconds before it was created loaded
2894 // cleanly and every duration derived from the pair came out negative.
2895 // The same hand-edited-journal threat model that motivates the presence
2896 // gate covers this, and `b2` is the task that will meet it.
2897 let mut concluded_before_created =
2898 row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
2899 concluded_before_created.created_at = ts(100);
2900 concluded_before_created.last_state_change_at = ts(100);
2901 concluded_before_created.terminal_at = Some(ts(0));
2902 assert_eq!(
2903 RunnerAttempt::from_persisted(concluded_before_created),
2904 Err(AttemptError::TimestampsOutOfOrder {
2905 state: AttemptState::Finished,
2906 field: "terminal_at",
2907 created_at: ts(100),
2908 found: ts(0),
2909 })
2910 );
2911
2912 // The same for the timestamp every recovery timeout is measured from. A
2913 // `last_state_change_at` before `created_at` makes `now - it` larger
2914 // than the attempt's whole life, so every timeout reads as expired.
2915 let mut changed_before_created = row(AttemptState::Busy, None);
2916 changed_before_created.created_at = ts(100);
2917 changed_before_created.last_state_change_at = ts(0);
2918 assert_eq!(
2919 RunnerAttempt::from_persisted(changed_before_created),
2920 Err(AttemptError::TimestampsOutOfOrder {
2921 state: AttemptState::Busy,
2922 field: "last_state_change_at",
2923 created_at: ts(100),
2924 found: ts(0),
2925 })
2926 );
2927
2928 // Equal is not out of order: `allocate` writes the same instant to both,
2929 // so the very first row of every attempt has `created_at ==
2930 // last_state_change_at`, and an attempt concluded in the same second it
2931 // was allocated has all three equal.
2932 let mut same_instant = row(AttemptState::Failed, Some(AttemptOutcome::Orphaned));
2933 same_instant.outcome = Some(AttemptOutcome::failed(FailureReason::JitRequestFailed));
2934 same_instant.created_at = ts(7);
2935 same_instant.last_state_change_at = ts(7);
2936 same_instant.terminal_at = Some(ts(7));
2937 assert!(RunnerAttempt::from_persisted(same_instant).is_ok());
2938
2939 // And a well-ordered row is untouched by the new arm.
2940 assert!(
2941 RunnerAttempt::from_persisted(row(
2942 AttemptState::Finished,
2943 Some(AttemptOutcome::CompletedJob)
2944 ))
2945 .is_ok()
2946 );
2947 }
2948
2949 #[test]
2950 fn an_attempt_round_trips_through_its_persisted_form() {
2951 let mut attempt = RunnerAttempt::allocate(
2952 AttemptId::from_u128(3),
2953 PolicyId::from_u128(4),
2954 "runtime/p/a",
2955 ts(0),
2956 );
2957 attempt.jit_received(ts(1)).unwrap();
2958 attempt.started(7, ts(2)).unwrap();
2959 attempt.registered_idle(7, ts(3)).unwrap();
2960 attempt
2961 .conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(4))
2962 .unwrap();
2963
2964 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2965 .expect("a row this crate produced must load");
2966 assert_eq!(restored, attempt);
2967 assert_eq!(restored.terminal_at(), Some(ts(4)));
2968 assert_ne!(
2969 restored.last_state_change_at(),
2970 restored.created_at,
2971 "the two timestamps must not collapse onto each other; that \
2972 transposition is what PersistedAttempt exists to prevent"
2973 );
2974 }
2975
2976 #[test]
2977 fn an_attempt_round_trips_through_serde() {
2978 let mut attempt = RunnerAttempt::allocate(
2979 AttemptId::from_u128(3),
2980 PolicyId::from_u128(4),
2981 "runtime/p/a",
2982 ts(0),
2983 );
2984 attempt.jit_received(ts(1)).unwrap();
2985 attempt.started(7, ts(2)).unwrap();
2986 attempt.registered_idle(73, ts(3)).unwrap();
2987 attempt
2988 .conclude(
2989 AttemptOutcome::failed(FailureReason::Other("no detail".into())),
2990 ts(4),
2991 )
2992 .unwrap();
2993
2994 let json = serde_json::to_string(&attempt).unwrap();
2995 let back: RunnerAttempt = serde_json::from_str(&json).unwrap();
2996 assert_eq!(attempt, back);
2997 }
2998
2999 // =======================================================================
3000 // Ownership
3001 // =======================================================================
3002
3003 #[test]
3004 fn an_attempt_belonging_to_another_host_is_rejected() {
3005 // Ownership rule 2: "A host agent may act only on attempts persisted
3006 // under its `host_id`."
3007 let mine = HostId::from_u128(7);
3008 let theirs = HostId::from_u128(8);
3009 let policy_id = PolicyId::from_u128(11);
3010
3011 let their_policy = a_policy(theirs, policy_id);
3012 let attempt =
3013 RunnerAttempt::allocate(AttemptId::from_u128(1), policy_id, "runtime/p/a", ts(0));
3014
3015 let err = authorize(mine, &their_policy, &attempt).unwrap_err();
3016 assert!(
3017 matches!(
3018 err,
3019 OwnershipError::ForeignHost {
3020 owner,
3021 agent,
3022 ..
3023 } if owner == theirs && agent == mine
3024 ),
3025 "got {err:?}"
3026 );
3027
3028 // The same attempt under our own policy is fine.
3029 let my_policy = a_policy(mine, policy_id);
3030 assert!(authorize(mine, &my_policy, &attempt).is_ok());
3031 }
3032
3033 #[test]
3034 fn an_attempt_checked_against_the_wrong_policy_is_rejected() {
3035 let host = HostId::from_u128(7);
3036 let policy = a_policy(host, PolicyId::from_u128(11));
3037 let attempt = RunnerAttempt::allocate(
3038 AttemptId::from_u128(1),
3039 PolicyId::from_u128(12),
3040 "runtime/p/a",
3041 ts(0),
3042 );
3043 assert!(matches!(
3044 authorize(host, &policy, &attempt),
3045 Err(OwnershipError::PolicyMismatch { .. })
3046 ));
3047 }
3048
3049 // =======================================================================
3050 // Recovery, against a controlled clock
3051 // =======================================================================
3052
3053 #[test]
3054 fn recovery_never_reads_the_system_clock() {
3055 // The whole decision surface moves when the fake clock moves, and by
3056 // nothing else. If any branch called `Utc::now()` the two decisions below
3057 // would be identical.
3058 let timeouts = RecoveryTimeouts::new(
3059 Elapsed::seconds(60),
3060 Elapsed::seconds(120),
3061 Elapsed::seconds(300),
3062 );
3063 let attempt = attempt_in(AttemptState::Idle, 1_000);
3064 let observation = RecoveryObservation {
3065 process_alive: false,
3066 github: GithubRunnerObservation::NotRegistered,
3067 };
3068
3069 let clock = StubClock::at(1_000 + 299);
3070 assert_eq!(
3071 recovery_decision(&attempt, observation, timeouts, &clock),
3072 RecoveryDecision::Conclude(AttemptOutcome::failed(
3073 FailureReason::ProcessExitedUnexpectedly
3074 )),
3075 "one second before its idle timeout, a vanished runner crashed"
3076 );
3077
3078 clock.set(1_000 + 300);
3079 assert_eq!(
3080 recovery_decision(&attempt, observation, timeouts, &clock),
3081 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
3082 "at its idle timeout, the same runner is the surplus case from flow 2.7"
3083 );
3084 }
3085
3086 #[test]
3087 fn a_live_process_is_adopted_rather_than_duplicated() {
3088 // `e3`: "an attempt whose process still runs is adopted, not duplicated."
3089 let timeouts = RecoveryTimeouts::provisional();
3090 let clock = StubClock::at(1_000_000);
3091 for state in [
3092 AttemptState::Allocated,
3093 AttemptState::JitReceived,
3094 AttemptState::Starting,
3095 AttemptState::Idle,
3096 AttemptState::Busy,
3097 ] {
3098 // `entered_at` is the clock's own instant, so every state is inside
3099 // its window and the only thing being asserted is the live-process
3100 // rule.
3101 let attempt = attempt_in(state, 1_000_000);
3102 assert_eq!(
3103 recovery_decision(
3104 &attempt,
3105 RecoveryObservation {
3106 process_alive: true,
3107 github: GithubRunnerObservation::NotRegistered,
3108 },
3109 timeouts,
3110 &clock,
3111 ),
3112 RecoveryDecision::Adopt,
3113 "{state} with a live process"
3114 );
3115 }
3116 }
3117
3118 #[test]
3119 fn a_dead_busy_attempt_is_orphaned_rather_than_guessed_into_a_success() {
3120 // The agent never reports a job as complete; GitHub remains the source of
3121 // truth for workflow outcome (flow 2, Failure).
3122 let clock = StubClock::at(1_000_000);
3123 for github in [
3124 GithubRunnerObservation::NotRegistered,
3125 GithubRunnerObservation::Registered { busy: true },
3126 GithubRunnerObservation::Registered { busy: false },
3127 ] {
3128 assert_eq!(
3129 recovery_decision(
3130 &attempt_in(AttemptState::Busy, 0),
3131 RecoveryObservation {
3132 process_alive: false,
3133 github,
3134 },
3135 RecoveryTimeouts::provisional(),
3136 &clock,
3137 ),
3138 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3139 "{github:?}"
3140 );
3141 }
3142 }
3143
3144 #[test]
3145 fn an_unreachable_github_defers_every_decision() {
3146 // Flow 3.3: while offline, start nothing and retain what is running. An
3147 // agent that treated "unreachable" as "not registered" would conclude and
3148 // clean every live attempt during a network outage.
3149 let clock = StubClock::at(1_000_000);
3150 for state in [
3151 AttemptState::Allocated,
3152 AttemptState::JitReceived,
3153 AttemptState::Starting,
3154 AttemptState::Idle,
3155 AttemptState::Busy,
3156 ] {
3157 assert_eq!(
3158 recovery_decision(
3159 &attempt_in(state, 0),
3160 RecoveryObservation {
3161 process_alive: false,
3162 github: GithubRunnerObservation::Unreachable,
3163 },
3164 RecoveryTimeouts::provisional(),
3165 &clock,
3166 ),
3167 RecoveryDecision::Defer,
3168 "{state} while GitHub is unreachable"
3169 );
3170 }
3171 }
3172
3173 #[test]
3174 fn a_starting_attempt_follows_what_github_reports() {
3175 // Precedence rule 3: GitHub runner status is authoritative for remote job
3176 // status. Both edges are in the diagram.
3177 let clock = StubClock::at(1_000);
3178 let attempt = attempt_in(AttemptState::Starting, 0);
3179
3180 assert_eq!(
3181 recovery_decision(
3182 &attempt,
3183 RecoveryObservation {
3184 process_alive: true,
3185 github: GithubRunnerObservation::Registered { busy: true },
3186 },
3187 RecoveryTimeouts::provisional(),
3188 &clock,
3189 ),
3190 RecoveryDecision::Observe(AttemptState::Busy)
3191 );
3192 assert_eq!(
3193 recovery_decision(
3194 &attempt,
3195 RecoveryObservation {
3196 process_alive: true,
3197 github: GithubRunnerObservation::Registered { busy: false },
3198 },
3199 RecoveryTimeouts::provisional(),
3200 &clock,
3201 ),
3202 RecoveryDecision::Observe(AttemptState::Idle)
3203 );
3204 }
3205
3206 #[test]
3207 fn a_pre_registration_attempt_believes_github_over_its_own_stale_journal() {
3208 // The crash window `e3` exists for: the process started and registered
3209 // at GitHub, but the write recording it was lost. GitHub is telling the
3210 // agent the runner is alive.
3211 //
3212 // This branch used to consult only `process_alive` and the clock, so the
3213 // observation below produced
3214 // `NoLegalTransition { from: JitReceived, wanted: Failed }` -- the
3215 // attempt was abandoned as failed while its registration stayed at
3216 // GitHub, never reconciled and never removed. Nothing in the suite
3217 // noticed, because nothing asked.
3218 let timeouts = RecoveryTimeouts::new(
3219 Elapsed::seconds(60),
3220 Elapsed::seconds(120),
3221 Elapsed::seconds(300),
3222 );
3223 // Well past the JIT handoff deadline, so the old code's timeout arm is
3224 // the one being displaced.
3225 let clock = StubClock::at(10_000);
3226
3227 for busy in [true, false] {
3228 assert_eq!(
3229 recovery_decision(
3230 &attempt_in(AttemptState::JitReceived, 0),
3231 RecoveryObservation {
3232 process_alive: false,
3233 github: GithubRunnerObservation::Registered { busy },
3234 },
3235 timeouts,
3236 &clock,
3237 ),
3238 RecoveryDecision::Observe(AttemptState::Starting),
3239 "jit_received + Registered{{busy:{busy}}}: the runner cannot have \
3240 registered without starting, and `jit_received -> starting` is \
3241 an edge the diagram already has"
3242 );
3243
3244 // `allocated` has no `-> starting` edge, so it takes the one legal
3245 // step it does have. The registration proves the JIT configuration
3246 // arrived, which is exactly what that edge records; the next pass
3247 // continues from `jit_received`.
3248 assert_eq!(
3249 recovery_decision(
3250 &attempt_in(AttemptState::Allocated, 0),
3251 RecoveryObservation {
3252 process_alive: false,
3253 github: GithubRunnerObservation::Registered { busy },
3254 },
3255 timeouts,
3256 &clock,
3257 ),
3258 RecoveryDecision::Observe(AttemptState::JitReceived),
3259 "allocated + Registered{{busy:{busy}}}"
3260 );
3261 }
3262
3263 // Unchanged where GitHub knows nothing: the timeout still decides, and a
3264 // live process is still adopted rather than duplicated. What changed at
3265 // the amendment is only what the timeout produces -- a conclusion
3266 // instead of a report that none was expressible.
3267 assert_eq!(
3268 recovery_decision(
3269 &attempt_in(AttemptState::JitReceived, 0),
3270 RecoveryObservation {
3271 process_alive: false,
3272 github: GithubRunnerObservation::NotRegistered,
3273 },
3274 timeouts,
3275 &clock,
3276 ),
3277 RecoveryDecision::Conclude(AttemptOutcome::failed(FailureReason::JitExpired))
3278 );
3279 assert_eq!(
3280 recovery_decision(
3281 &attempt_in(AttemptState::JitReceived, 0),
3282 RecoveryObservation {
3283 process_alive: true,
3284 github: GithubRunnerObservation::Registered { busy: false },
3285 },
3286 timeouts,
3287 &clock,
3288 ),
3289 RecoveryDecision::Adopt,
3290 "precedence rule 3's other half: local process state is \
3291 authoritative for a child process this agent owns"
3292 );
3293 }
3294
3295 #[test]
3296 fn a_terminal_attempt_is_cleaned_and_a_cleaned_one_is_left_alone() {
3297 let clock = StubClock::at(1_000_000);
3298 let observation = RecoveryObservation {
3299 process_alive: false,
3300 github: GithubRunnerObservation::NotRegistered,
3301 };
3302 for state in [
3303 AttemptState::Finished,
3304 AttemptState::Failed,
3305 AttemptState::Orphaned,
3306 ] {
3307 assert_eq!(
3308 recovery_decision(
3309 &attempt_in(state, 0),
3310 observation,
3311 RecoveryTimeouts::provisional(),
3312 &clock
3313 ),
3314 RecoveryDecision::Clean,
3315 "{state}"
3316 );
3317 }
3318 assert_eq!(
3319 recovery_decision(
3320 &attempt_in(AttemptState::Cleaned, 0),
3321 observation,
3322 RecoveryTimeouts::provisional(),
3323 &clock
3324 ),
3325 RecoveryDecision::Nothing
3326 );
3327 }
3328
3329 #[test]
3330 fn a_pre_registration_attempt_waits_until_its_deadline_then_concludes() {
3331 // This test used to document a gap in the state diagram: `allocated`,
3332 // `jit_received` and `starting` had no edge to `failed` or `orphaned`,
3333 // so a dead attempt that never registered could not be concluded at all
3334 // and the decision reported the missing edge instead of inventing one.
3335 // The 2026-08-21 amendment added those edges, and the whole point of
3336 // adding them was that this attempt now gives its capacity slot back.
3337 let timeouts = RecoveryTimeouts::new(
3338 Elapsed::seconds(60),
3339 Elapsed::seconds(120),
3340 Elapsed::seconds(300),
3341 );
3342 let clock = StubClock::at(1_059);
3343 let observation = RecoveryObservation {
3344 process_alive: false,
3345 github: GithubRunnerObservation::NotRegistered,
3346 };
3347
3348 // The two pre-JIT states conclude with different reasons, because they
3349 // know different things: at `allocated` no configuration was ever
3350 // recorded as arriving, at `jit_received` one arrived and was never
3351 // claimed.
3352 for (state, reason) in [
3353 (AttemptState::Allocated, FailureReason::JitRequestFailed),
3354 (AttemptState::JitReceived, FailureReason::JitExpired),
3355 ] {
3356 let attempt = attempt_in(state, 1_000);
3357 assert_eq!(
3358 recovery_decision(&attempt, observation, timeouts, &clock),
3359 RecoveryDecision::Wait,
3360 "{state} inside its handoff window"
3361 );
3362 clock.set(1_060);
3363 assert_eq!(
3364 recovery_decision(&attempt, observation, timeouts, &clock),
3365 RecoveryDecision::Conclude(AttemptOutcome::failed(reason)),
3366 "{state} past its handoff window concludes rather than stranding \
3367 the attempt"
3368 );
3369 clock.set(1_059);
3370 }
3371
3372 // `starting` past its own, longer deadline: flow 2's "runner exit before
3373 // job acceptance".
3374 clock.set(1_121);
3375 assert_eq!(
3376 recovery_decision(
3377 &attempt_in(AttemptState::Starting, 1_000),
3378 observation,
3379 timeouts,
3380 &clock
3381 ),
3382 RecoveryDecision::Conclude(AttemptOutcome::failed(
3383 FailureReason::ProcessExitedUnexpectedly
3384 ))
3385 );
3386
3387 // Every one of those decisions must be applicable to the attempt it was
3388 // made about; a decision the state machine then refuses would strand the
3389 // attempt just as the missing edges did.
3390 for (state, decision) in [
3391 (
3392 AttemptState::Allocated,
3393 AttemptOutcome::failed(FailureReason::JitRequestFailed),
3394 ),
3395 (
3396 AttemptState::JitReceived,
3397 AttemptOutcome::failed(FailureReason::JitExpired),
3398 ),
3399 (
3400 AttemptState::Starting,
3401 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
3402 ),
3403 ] {
3404 let mut attempt = attempt_in(state, 1_000);
3405 attempt
3406 .conclude(decision, ts(1_200))
3407 .unwrap_or_else(|e| panic!("recovery decided {state} concludes, but: {e}"));
3408 assert!(!attempt.counts_against_capacity());
3409 }
3410 }
3411
3412 #[test]
3413 fn an_idle_runner_that_github_reports_as_busy_is_recorded_as_busy() {
3414 // The `idle -> busy` half of the amendment, at the point it bites in
3415 // recovery. GitHub is authoritative for remote job status (precedence
3416 // rule 3), and before the amendment this observation was not
3417 // representable at all: the decision reported that no legal transition
3418 // existed and the attempt stayed `idle` for ever while a job ran on it.
3419 let timeouts = RecoveryTimeouts::provisional();
3420 let clock = StubClock::at(1_121);
3421 assert_eq!(
3422 recovery_decision(
3423 &attempt_in(AttemptState::Idle, 1_000),
3424 RecoveryObservation {
3425 process_alive: false,
3426 github: GithubRunnerObservation::Registered { busy: true },
3427 },
3428 timeouts,
3429 &clock,
3430 ),
3431 RecoveryDecision::Observe(AttemptState::Busy)
3432 );
3433
3434 // And the decision is applicable: `idle -> busy` is an edge, so the
3435 // caller can actually carry it out.
3436 let mut attempt = attempt_in(AttemptState::Idle, 1_000);
3437 attempt.assigned_job(73, ts(1_200)).unwrap();
3438 assert_eq!(attempt.state(), AttemptState::Busy);
3439 assert_eq!(attempt.github_runner_id(), Some(73));
3440
3441 // And a live process does not change the answer. GitHub is authoritative
3442 // for remote job status; the caller adopts supervision independently of
3443 // the decision, which is what `Observe`'s own documentation says and
3444 // what `starting` has always relied on.
3445 assert_eq!(
3446 recovery_decision(
3447 &attempt_in(AttemptState::Idle, 1_000),
3448 RecoveryObservation {
3449 process_alive: true,
3450 github: GithubRunnerObservation::Registered { busy: true },
3451 },
3452 timeouts,
3453 &clock,
3454 ),
3455 RecoveryDecision::Observe(AttemptState::Busy),
3456 "a live process must not hide a job GitHub is reporting"
3457 );
3458 }
3459
3460 #[test]
3461 fn a_restart_during_a_job_is_recorded_as_busy_and_not_left_reading_idle() {
3462 // A1. The same conflict -- a live process that GitHub reports as `busy`
3463 // -- must resolve the same way at `idle` as at `starting`, and this test
3464 // is written as the damage rather than as the symmetry, because the
3465 // symmetry is not what it costs to get wrong.
3466 //
3467 // Before the fix, `idle` short-circuited on `process_alive` and answered
3468 // `Adopt`. `Adopt` writes nothing, so the journal stayed `idle` and
3469 // `last_state_change_at` stayed pointed at the idle entry. Ten minutes
3470 // later the runner crashed mid-job, GitHub reaped its registration, and
3471 // the idle timeout had long since elapsed -- so the crash was concluded
3472 // `ExitedIdleWithoutWork`: the benign surplus exit, which `g2` renders
3473 // as a normal end and never alarms an operator about. Nothing
3474 // downstream could refuse it, because `required_from` for that outcome
3475 // is `&[Idle]` and the attempt genuinely was `idle`.
3476 let timeouts = RecoveryTimeouts::new(
3477 Elapsed::seconds(60),
3478 Elapsed::seconds(120),
3479 Elapsed::seconds(300),
3480 );
3481 let clock = StubClock::at(1_000);
3482
3483 // The two arms answer alike, which is the property. `starting` was
3484 // already right; `idle` was not.
3485 let mid_job = RecoveryObservation {
3486 process_alive: true,
3487 github: GithubRunnerObservation::Registered { busy: true },
3488 };
3489 assert_eq!(
3490 recovery_decision(
3491 &attempt_in(AttemptState::Idle, 1_000),
3492 mid_job,
3493 timeouts,
3494 &clock
3495 ),
3496 recovery_decision(
3497 &attempt_in(AttemptState::Starting, 1_000),
3498 mid_job,
3499 timeouts,
3500 &clock
3501 ),
3502 "the same observation must not resolve one way at idle and the \
3503 opposite way one state earlier"
3504 );
3505 assert_eq!(
3506 recovery_decision(
3507 &attempt_in(AttemptState::Idle, 1_000),
3508 mid_job,
3509 timeouts,
3510 &clock
3511 ),
3512 RecoveryDecision::Observe(AttemptState::Busy)
3513 );
3514
3515 // And now the consequence, walked end to end. The caller applies the
3516 // decision -- which `Observe` requires, on pain of repeating for ever --
3517 // and the attempt is `busy` with `last_state_change_at` moved to the
3518 // moment the job was observed.
3519 let mut attempt = attempt_in(AttemptState::Idle, 1_000);
3520 attempt.assigned_job(73, ts(1_000)).expect("idle -> busy");
3521 assert_eq!(attempt.last_state_change_at(), ts(1_000));
3522
3523 // Ten minutes on, the process is gone and GitHub has reaped the runner.
3524 // Well past the idle timeout, and irrelevant: a crash from `busy` is
3525 // `Orphaned`, and `ExitedIdleWithoutWork` is not reachable from `busy`
3526 // at all.
3527 clock.set(1_600);
3528 let crashed = RecoveryObservation {
3529 process_alive: false,
3530 github: GithubRunnerObservation::NotRegistered,
3531 };
3532 let decision = recovery_decision(&attempt, crashed, timeouts, &clock);
3533 assert_eq!(
3534 decision,
3535 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3536 "a crash during a job is a lost supervision, never the surplus exit"
3537 );
3538 let RecoveryDecision::Conclude(outcome) = decision else {
3539 unreachable!("asserted just above")
3540 };
3541 assert!(
3542 outcome.is_failure() && !outcome.is_idle_exit(),
3543 "`g2` reads these two flags to decide whether to alarm an operator, \
3544 and a mid-job crash must alarm one"
3545 );
3546
3547 // The old behaviour, spelled out so the regression is unmistakable: had
3548 // the attempt been left at `idle`, this is what the same crash would
3549 // have produced.
3550 assert_eq!(
3551 recovery_decision(
3552 &attempt_in(AttemptState::Idle, 1_000),
3553 crashed,
3554 timeouts,
3555 &clock
3556 ),
3557 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
3558 "which is exactly why the journal must not be left saying `idle`"
3559 );
3560 }
3561
3562 #[test]
3563 fn a_registered_runner_that_never_gets_a_job_is_stopped_at_its_idle_timeout() {
3564 // The surplus case of flow 2.7, in the shape it actually occurs in:
3565 // GitHub still lists the runner, it is not busy, and the process is
3566 // very much alive -- `Runner.Listener run` long-polls for an assignment
3567 // and has no idle timeout of its own, so it never leaves on its own.
3568 // This arm answered `Adopt` at every elapsed time, which is why a runner
3569 // observed in the field held its slot and its row in the target's runner
3570 // settings for 27 hours.
3571 let timeouts = RecoveryTimeouts::new(
3572 Elapsed::seconds(60),
3573 Elapsed::seconds(120),
3574 Elapsed::seconds(300),
3575 );
3576 let clock = StubClock::at(1_000);
3577 let attempt = attempt_in(AttemptState::Idle, 1_000);
3578 let idle_registered = |alive| RecoveryObservation {
3579 process_alive: alive,
3580 github: GithubRunnerObservation::Registered { busy: false },
3581 };
3582
3583 // Inside the window the runner is still plausibly about to be assigned,
3584 // so it is adopted and nothing is disturbed.
3585 clock.set(1_299);
3586 assert_eq!(
3587 recovery_decision(&attempt, idle_registered(true), timeouts, &clock),
3588 RecoveryDecision::Adopt,
3589 "a runner one second inside its idle timeout may still be given a job"
3590 );
3591
3592 // At the deadline it is surplus, and the agent has to end it.
3593 clock.set(1_300);
3594 let decision = recovery_decision(&attempt, idle_registered(true), timeouts, &clock);
3595 assert_eq!(
3596 decision,
3597 RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork),
3598 "past the idle timeout a registered, unassigned runner is flow 2.7's surplus case"
3599 );
3600
3601 // `Terminate`, not `Conclude`, and the difference is the capacity slot:
3602 // the process is alive, so the slot may not come back until it is gone.
3603 let mut attempt = attempt;
3604 assert!(attempt.counts_against_capacity());
3605 let RecoveryDecision::Terminate(outcome) = decision else {
3606 unreachable!("asserted above")
3607 };
3608 assert!(
3609 outcome.is_idle_exit(),
3610 "the surplus exit is a normal outcome, and `g2` renders it as one -- an operator \
3611 sent to hunt a fault here would find nothing"
3612 );
3613 attempt
3614 .conclude(outcome, ts(1_310))
3615 .expect("the payload must be applicable to the attempt it was made about");
3616 assert!(!attempt.counts_against_capacity());
3617
3618 // A dead process is the pre-existing reading and is left alone: the
3619 // registration outlived supervision, which is what `Orphaned` names.
3620 clock.set(1_400);
3621 assert_eq!(
3622 recovery_decision(
3623 &attempt_in(AttemptState::Idle, 1_000),
3624 idle_registered(false),
3625 timeouts,
3626 &clock
3627 ),
3628 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3629 "the idle deadline governs a live runner; a dead one is orphaned however long it sat"
3630 );
3631
3632 // And a runner GitHub reports busy is never stopped, at any elapsed
3633 // time. This is the assertion that stops the deadline above from ever
3634 // being applied to a runner in the middle of a job.
3635 for offset in [0, 299, 300, 100_000] {
3636 clock.set(1_000 + offset);
3637 assert_eq!(
3638 recovery_decision(
3639 &attempt_in(AttemptState::Idle, 1_000),
3640 RecoveryObservation {
3641 process_alive: true,
3642 github: GithubRunnerObservation::Registered { busy: true },
3643 },
3644 timeouts,
3645 &clock
3646 ),
3647 RecoveryDecision::Observe(AttemptState::Busy),
3648 "at +{offset}s a runner that took a job was stopped by an idle deadline"
3649 );
3650 }
3651 }
3652
3653 #[test]
3654 fn a_live_unregistered_runner_past_its_deadline_is_stopped_before_its_slot_returns() {
3655 // A2. Three cases at `starting` with nothing at GitHub, and the middle
3656 // one used to be folded into the last.
3657 let timeouts = RecoveryTimeouts::new(
3658 Elapsed::seconds(60),
3659 Elapsed::seconds(120),
3660 Elapsed::seconds(300),
3661 );
3662 let clock = StubClock::at(1_000);
3663 let attempt = attempt_in(AttemptState::Starting, 1_000);
3664 let unregistered = |alive| RecoveryObservation {
3665 process_alive: alive,
3666 github: GithubRunnerObservation::NotRegistered,
3667 };
3668
3669 // Inside the window, alive: adopted.
3670 clock.set(1_119);
3671 assert_eq!(
3672 recovery_decision(&attempt, unregistered(true), timeouts, &clock),
3673 RecoveryDecision::Adopt
3674 );
3675
3676 // Gone: an accurate `ProcessExitedUnexpectedly`, and terminal, because
3677 // there is nothing left running to hold the slot.
3678 assert_eq!(
3679 recovery_decision(&attempt, unregistered(false), timeouts, &clock),
3680 RecoveryDecision::Conclude(AttemptOutcome::failed(
3681 FailureReason::ProcessExitedUnexpectedly
3682 )),
3683 "the one case that really is flow 2's runner exit"
3684 );
3685
3686 // Alive and past the deadline: neither of the above.
3687 clock.set(1_120);
3688 let decision = recovery_decision(&attempt, unregistered(true), timeouts, &clock);
3689 assert_eq!(
3690 decision,
3691 RecoveryDecision::Terminate(AttemptOutcome::failed(
3692 FailureReason::RegistrationTimedOut
3693 ))
3694 );
3695 assert_ne!(
3696 decision,
3697 RecoveryDecision::Conclude(AttemptOutcome::failed(
3698 FailureReason::ProcessExitedUnexpectedly
3699 )),
3700 "a process visible in a task manager did not exit unexpectedly, and \
3701 an operator told that it did stops believing the next message too"
3702 );
3703
3704 // The substance: the slot is not returned while the process runs. The
3705 // decision itself moves nothing, so the attempt is still `starting` and
3706 // still counted -- which is what stops the agent starting a replacement
3707 // beside a runner that could still register and take a job.
3708 let mut attempt = attempt;
3709 assert!(attempt.counts_against_capacity());
3710 assert_eq!(active_count([&attempt]), 1);
3711
3712 // The slot comes back at the moment the caller applies the payload,
3713 // which it does only after the process is gone.
3714 let RecoveryDecision::Terminate(outcome) = decision else {
3715 unreachable!("asserted above")
3716 };
3717 attempt
3718 .conclude(outcome, ts(1_130))
3719 .expect("the payload must be applicable to the attempt it was made about");
3720 assert_eq!(attempt.state(), AttemptState::Failed);
3721 assert!(!attempt.counts_against_capacity());
3722 assert_eq!(active_count([&attempt]), 0);
3723
3724 // Half the re-derivation is free: an agent that died between deciding
3725 // and terminating still sees a live process, and is handed the same
3726 // answer next pass.
3727 let pending = attempt_in(AttemptState::Starting, 1_000);
3728 assert_eq!(
3729 recovery_decision(&pending, unregistered(true), timeouts, &clock),
3730 RecoveryDecision::Terminate(AttemptOutcome::failed(
3731 FailureReason::RegistrationTimedOut
3732 ))
3733 );
3734
3735 // The other half is not, and this is the assertion that says so rather
3736 // than blessing it. An agent that terminated the process and died
3737 // before writing the outcome is, on the next pass, indistinguishable
3738 // from one whose runner crashed on its own: both present exactly the
3739 // same `RecoveryObservation`. The decision below is therefore the same
3740 // for both, and for the terminate-then-crash case the reason it carries
3741 // is wrong -- nothing exited unexpectedly, this agent killed it.
3742 let crashed_on_its_own = unregistered(false);
3743 let killed_by_us_then_lost = unregistered(false);
3744 assert_eq!(
3745 crashed_on_its_own, killed_by_us_then_lost,
3746 "this equality *is* the defect: the fact that separates the two \
3747 cases is not an input to `recovery_decision`, which is why it \
3748 cannot be repaired inside it"
3749 );
3750 assert_eq!(
3751 recovery_decision(&pending, killed_by_us_then_lost, timeouts, &clock),
3752 RecoveryDecision::Conclude(AttemptOutcome::failed(
3753 FailureReason::ProcessExitedUnexpectedly
3754 )),
3755 "pinned as the current behaviour of a known-wrong window, not as a \
3756 correct answer: `e3` closes it by journalling terminate-intent \
3757 before signalling and concluding the marked attempt with \
3758 `TerminatedAfterRegistrationTimeout`, which is true of the dead \
3759 process this pass is looking at. `RecoveryDecision::Terminate` \
3760 names the trade and the owner"
3761 );
3762 }
3763
3764 #[test]
3765 fn no_decision_calls_a_dead_process_live() {
3766 // `RegistrationTimedOut` renders as "the runner process is running but
3767 // did not register", `ProcessExitedUnexpectedly` is documented "only for
3768 // a process that is actually gone", and
3769 // `TerminatedAfterRegistrationTimeout` says the agent stopped the
3770 // process. All three are claims about liveness, and
3771 // `RecoveryObservation::process_alive` is the only source of truth for
3772 // it, so each reason may only ever appear beside the observation that
3773 // supports it.
3774 //
3775 // This is also the guard on the `Terminate` window's residual. The
3776 // rejected fix for that window -- reach for `RegistrationTimedOut` in
3777 // the dead-process arm at `starting` once `elapsed >= startup` --
3778 // compiles, and reds three tests. Measured by applying it:
3779 //
3780 // * here, at `starting at +120s with NotRegistered and a process the
3781 // observation says is gone was reported as still running`;
3782 // * `a_live_unregistered_runner_past_its_deadline_is_stopped_before_
3783 // its_slot_returns`, at the assertion that pins the known-wrong
3784 // window as current behaviour;
3785 // * `a_pre_registration_attempt_waits_until_its_deadline_then_
3786 // concludes`, with `left: Conclude(Failed { reason:
3787 // RegistrationTimedOut })` against `right: Conclude(Failed { reason:
3788 // ProcessExitedUnexpectedly })`.
3789 //
3790 // The third is the strongest of the three and is worth naming ahead of
3791 // this one. It is not a guard written to catch this mistake: it is a
3792 // real scenario already in the suite -- flow 2's "runner exit before job
3793 // acceptance", process dead, elapsed 121s against a 120s startup window
3794 // -- which the rejected fix relabels `RegistrationTimedOut`. So the
3795 // objection is not only that a synthetic sweep dislikes the change; an
3796 // ordinary early crash, observed after a restart longer than the startup
3797 // window, gets reported as a runner that is still running. This test
3798 // remains the argument's *general* form, because it holds the claim at
3799 // every state, every observation and both sides of every deadline rather
3800 // than at one scenario.
3801 let timeouts = RecoveryTimeouts::new(
3802 Elapsed::seconds(60),
3803 Elapsed::seconds(120),
3804 Elapsed::seconds(300),
3805 );
3806 let entered_at = 1_000i64;
3807 // Both sides of each deadline, the deadlines themselves, and one
3808 // instant far past all three.
3809 let offsets = [0, 59, 60, 61, 119, 120, 121, 299, 300, 301, 100_000];
3810 let observations = [
3811 GithubRunnerObservation::NotRegistered,
3812 GithubRunnerObservation::Registered { busy: false },
3813 GithubRunnerObservation::Registered { busy: true },
3814 GithubRunnerObservation::Unreachable,
3815 ];
3816
3817 // Set by the third arm below, and asserted `false` after the sweep. It
3818 // is what keeps that arm from being a vacuous assertion nobody notices
3819 // has stopped meaning anything: today the arm is unreached, and this
3820 // says so out loud rather than leaving it to be assumed.
3821 let mut agent_termination_derived_here = false;
3822
3823 for state in AttemptState::ALL {
3824 for github in observations {
3825 for process_alive in [true, false] {
3826 for offset in offsets {
3827 let clock = StubClock::at(entered_at + offset);
3828 let observation = RecoveryObservation {
3829 process_alive,
3830 github,
3831 };
3832 let decision = recovery_decision(
3833 &attempt_in(state, entered_at),
3834 observation,
3835 timeouts,
3836 &clock,
3837 );
3838 let reason = match &decision {
3839 RecoveryDecision::Conclude(AttemptOutcome::Failed { reason })
3840 | RecoveryDecision::Terminate(AttemptOutcome::Failed { reason }) => {
3841 Some(reason)
3842 }
3843 _ => None,
3844 };
3845 match reason {
3846 Some(FailureReason::RegistrationTimedOut) => assert!(
3847 process_alive,
3848 "{state} at +{offset}s with {github:?} and a process the \
3849 observation says is gone was reported as still running"
3850 ),
3851 Some(FailureReason::ProcessExitedUnexpectedly) => assert!(
3852 !process_alive,
3853 "{state} at +{offset}s with {github:?} reported an \
3854 unexpected exit for a process that is alive"
3855 ),
3856 // Nothing here derives this one today -- it is
3857 // `e3`'s, recorded after reading its own journalled
3858 // terminate-intent back, which is why the flag above
3859 // records that the arm was reached at all. It is not
3860 // idle cover: once the mark is journalled somewhere
3861 // `recovery_decision` can read it, deriving the
3862 // reason here becomes legitimate, and this is what
3863 // stops that derivation landing beside a process the
3864 // observation reports as still running.
3865 Some(FailureReason::TerminatedAfterRegistrationTimeout) => {
3866 agent_termination_derived_here = true;
3867 assert!(
3868 !process_alive,
3869 "{state} at +{offset}s with {github:?} said the agent had \
3870 stopped a process the observation reports as alive"
3871 );
3872 }
3873 _ => {}
3874 }
3875 }
3876 }
3877 }
3878 }
3879
3880 // The division of labour, asserted rather than described. The reason
3881 // that names an agent-initiated stop is not derivable from an
3882 // observation -- a process this agent killed and one that crashed on its
3883 // own look identical to `recovery_decision` -- so it must not appear
3884 // anywhere in the sweep above. If it ever does, the fact that separates
3885 // the two cases has become an input, and `RecoveryDecision::Terminate`'s
3886 // deferred obligation is either discharged or wrong; either way somebody
3887 // should be reading that doc rather than passing this test by accident.
3888 assert!(
3889 !agent_termination_derived_here,
3890 "`recovery_decision` derived `TerminatedAfterRegistrationTimeout`, \
3891 which is `e3`'s to record from journalled terminate-intent and not \
3892 this function's to infer from an observation that cannot know"
3893 );
3894 }
3895
3896 #[test]
3897 fn a_terminated_runner_is_never_described_as_running() {
3898 // The `Display` half of the same argument. `e3` concludes the window in
3899 // `RecoveryDecision::Terminate`'s doc with this reason, on a pass where
3900 // the process is already dead, so its rendering may not claim otherwise.
3901 // Reusing `RegistrationTimedOut` there -- the closure this file used to
3902 // promise -- would have printed "the runner process is running but did
3903 // not register" about a process the operator can see is gone: the same
3904 // false liveness claim option A is rejected for, one pass later.
3905 let terminated = FailureReason::TerminatedAfterRegistrationTimeout.to_string();
3906 assert!(
3907 !terminated.contains("running"),
3908 "this reason is only ever recorded about a process the agent has \
3909 already stopped: {terminated}"
3910 );
3911 assert!(
3912 !terminated.contains("exited unexpectedly"),
3913 "and nothing exited unexpectedly either -- the agent stopped it on \
3914 purpose: {terminated}"
3915 );
3916 assert!(
3917 terminated.contains("stopped") && terminated.contains("register"),
3918 "it has to say both what happened to the process and why, or it \
3919 sends an operator to a crash investigation: {terminated}"
3920 );
3921
3922 // Distinct from both neighbours, in the variant and in the string. A
3923 // reason that renders identically to another is a reason `g2` cannot
3924 // use to send an operator anywhere different.
3925 assert_ne!(
3926 terminated,
3927 FailureReason::RegistrationTimedOut.to_string(),
3928 "the live and the stopped case must read differently"
3929 );
3930 assert_ne!(
3931 terminated,
3932 FailureReason::ProcessExitedUnexpectedly.to_string()
3933 );
3934 assert!(
3935 !matches!(
3936 FailureReason::TerminatedAfterRegistrationTimeout,
3937 FailureReason::Other(_)
3938 ),
3939 "a known, named condition must not travel through the escape hatch"
3940 );
3941 }
3942
3943 #[test]
3944 fn the_two_starting_failures_read_differently_to_an_operator() {
3945 // `g2` renders `FailureReason` directly, and these two send an operator
3946 // in different directions: one to a crash investigation, the other to a
3947 // networking or configuration fix. Reaching for `Other` here instead of
3948 // a named variant would have put a free-form string on the same path.
3949 let exited = FailureReason::ProcessExitedUnexpectedly.to_string();
3950 let timed_out = FailureReason::RegistrationTimedOut.to_string();
3951 assert_ne!(exited, timed_out);
3952 assert!(
3953 timed_out.contains("running") && timed_out.contains("register"),
3954 "the message must say the process is up and unregistered: {timed_out}"
3955 );
3956 assert!(
3957 !timed_out.contains("exited"),
3958 "the whole point is that nothing exited: {timed_out}"
3959 );
3960 assert!(
3961 !matches!(FailureReason::RegistrationTimedOut, FailureReason::Other(_)),
3962 "a known, named condition must not travel through the escape hatch"
3963 );
3964 }
3965
3966 #[test]
3967 fn a_dead_idle_runner_still_registered_at_github_is_orphaned() {
3968 let clock = StubClock::at(1_000_000);
3969 assert_eq!(
3970 recovery_decision(
3971 &attempt_in(AttemptState::Idle, 0),
3972 RecoveryObservation {
3973 process_alive: false,
3974 github: GithubRunnerObservation::Registered { busy: false },
3975 },
3976 RecoveryTimeouts::provisional(),
3977 &clock,
3978 ),
3979 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3980 "the remote registration outlived our process and needs removing"
3981 );
3982 }
3983}