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