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