1#[cfg(test)]
12use std::collections::VecDeque;
13use std::collections::{BTreeMap, BTreeSet};
14use std::ffi::OsStr;
15use std::fmt;
16use std::fs;
17use std::io::Write;
18use std::num::NonZeroU16;
19use std::path::{Path, PathBuf};
20use std::sync::{Arc, Mutex};
21use std::time::Duration;
22
23use async_trait::async_trait;
24use runner_manager_domain::attempt::{
25 AttemptOutcome, AttemptState, FailureReason, GithubRunnerObservation, RecoveryDecision,
26 RecoveryObservation, RecoveryTimeouts, RunnerAttempt, authorize, recovery_decision,
27};
28use runner_manager_domain::model::{AttemptId, Clock, HostId, PolicyId, ScaleTarget};
29use runner_manager_domain::path::LocalAbsolutePath;
30use runner_manager_domain::policy::ScalePolicy;
31use runner_manager_domain::store::{Store, StoreError};
32use runner_manager_domain::workspace::{AttemptWorkspace, WorkspacePolicy};
33use runner_manager_github::jit::{
34 DEFAULT_WORK_FOLDER, EncodedJitConfig, JitError, JitGateway, JitRegistration, JitRunnerRequest,
35};
36use runner_manager_github::rest::{CancelToken, InventoryGateway};
37use runner_manager_platform::process::{
38 Adoption, ChildProcess, ProcessIdentity, RestrictiveHandoff, SpawnSpec, Termination,
39};
40use runner_manager_platform::runner_root::{
41 self, RootOwner, RootPreflight, RunnerRootError, default_runner_root,
42};
43use secrecy::SecretString;
44
45use crate::package::{PackageCache, PackageError, RunnerVersion};
46use crate::reconcile::{
47 AllocationGuard, EventSink, LaunchFailure, LaunchRequest, LifecycleEvent, OutcomeKind,
48 ReplacementIntent, RunnerLauncher,
49};
50
51const IDENTITY_FILE: &str = ".runner-process.json";
52const FALLBACK_IDENTITY_FILE: &str = ".runner-process.recovery.json";
53const UNRESOLVED_PROCESS_FILE: &str = ".runner-process.unresolved";
54const RUNNER_ID_FILE: &str = ".github-runner-id";
55const TERMINATE_INTENT_FILE: &str = ".terminate-registration-timeout";
56const MAX_POST_SPAWN_STOP_ATTEMPTS: usize = 3;
57
58const SENSITIVE_SLOT_ENTRIES: &[&str] = &[
71 "bin",
73 "externals",
74 "run.sh",
75 "run.cmd",
76 "config.sh",
77 "config.cmd",
78 ".runner",
81 ".credentials",
82 ".credentials_rsaparams",
83 ".env",
84 ".path",
85 "_diag",
86 IDENTITY_FILE,
88 FALLBACK_IDENTITY_FILE,
89 UNRESOLVED_PROCESS_FILE,
90 RUNNER_ID_FILE,
91 TERMINATE_INTENT_FILE,
92];
93#[cfg(test)]
94const TEST_LISTENER_READY: &str = ".test-listener-ready";
95
96fn runner_listener_spec(program: PathBuf, runtime: &Path) -> SpawnSpec {
101 let tmp = runtime.join("tmp");
102 let _ = std::fs::create_dir_all(&tmp);
103 SpawnSpec::new(program)
104 .arg("run")
105 .working_dir(runtime)
106 .env("TMPDIR", &tmp)
107 .env("TEMP", &tmp)
108 .env("TMP", &tmp)
109}
110
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub struct RetryPolicy {
114 pub max_attempts: u32,
115 pub initial: Duration,
116 pub maximum: Duration,
117}
118
119impl RetryPolicy {
120 #[must_use]
121 pub const fn bounded(max_attempts: u32, initial: Duration, maximum: Duration) -> Self {
122 Self {
123 max_attempts,
124 initial,
125 maximum,
126 }
127 }
128
129 fn delay(self, failure_index: u32) -> Duration {
130 let shift = failure_index.saturating_sub(1).min(31);
131 self.initial
132 .saturating_mul(1_u32 << shift)
133 .min(self.maximum)
134 }
135}
136
137#[derive(Debug, Clone, PartialEq, Eq)]
140pub enum AttemptEvent {
141 State {
142 attempt: AttemptId,
143 state: AttemptState,
144 },
145 Retry {
146 attempt: AttemptId,
147 operation: &'static str,
148 delay: Duration,
149 },
150 Adopted {
151 attempt: AttemptId,
152 },
153 RemoteIdentityRecovered {
154 attempt: AttemptId,
155 runner_id: u64,
156 },
157 TerminateIntent {
158 attempt: AttemptId,
159 },
160 Terminated {
161 attempt: AttemptId,
162 },
163 Deregistered {
167 attempt: AttemptId,
168 runner_id: u64,
169 },
170 Concluded {
171 attempt: AttemptId,
172 outcome: OutcomeKind,
173 },
174 Cleaned {
175 attempt: AttemptId,
176 outcome: OutcomeKind,
177 },
178}
179
180pub trait AttemptEventSink: fmt::Debug + Send + Sync {
181 fn emit(&self, event: AttemptEvent);
182}
183
184#[derive(Debug, Default)]
185pub struct AttemptEventLog(Mutex<Vec<AttemptEvent>>);
186
187impl AttemptEventLog {
188 #[must_use]
189 pub fn events(&self) -> Vec<AttemptEvent> {
190 self.0
191 .lock()
192 .map(|events| events.clone())
193 .unwrap_or_default()
194 }
195}
196
197impl AttemptEventSink for AttemptEventLog {
198 fn emit(&self, event: AttemptEvent) {
199 if let Ok(mut events) = self.0.lock() {
200 events.push(event);
201 }
202 }
203}
204
205#[derive(Debug, Clone, Copy, Default)]
206pub struct NoAttemptEvents;
207
208impl AttemptEventSink for NoAttemptEvents {
209 fn emit(&self, _event: AttemptEvent) {}
210}
211
212#[async_trait]
214pub trait DemandPersistence: fmt::Debug + Send + Sync {
215 async fn persists(&self, policy: PolicyId) -> bool;
216}
217
218#[derive(Debug, Clone, Copy, Default)]
219pub struct PersistentDemand;
220
221#[async_trait]
222impl DemandPersistence for PersistentDemand {
223 async fn persists(&self, _policy: PolicyId) -> bool {
224 true
225 }
226}
227
228#[async_trait]
229pub trait RetryDelay: fmt::Debug + Send + Sync {
230 async fn wait(&self, duration: Duration);
231}
232
233#[derive(Debug, Clone, Copy, Default)]
234pub struct TokioRetryDelay;
235
236#[async_trait]
237impl RetryDelay for TokioRetryDelay {
238 async fn wait(&self, duration: Duration) {
239 tokio::time::sleep(duration).await;
240 }
241}
242
243#[derive(Debug, Clone, PartialEq, Eq)]
244pub struct JitRequestFailure {
245 pub terminal: bool,
246 pub reason: FailureReason,
247 pub retry_after: Option<Duration>,
248}
249
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
254pub struct LifecycleGithubObservation {
255 pub status: GithubRunnerObservation,
256 pub runner_id: Option<u64>,
257}
258
259impl LifecycleGithubObservation {
260 #[must_use]
261 pub const fn unreachable() -> Self {
262 Self {
263 status: GithubRunnerObservation::Unreachable,
264 runner_id: None,
265 }
266 }
267
268 #[must_use]
269 pub const fn not_registered() -> Self {
270 Self {
271 status: GithubRunnerObservation::NotRegistered,
272 runner_id: None,
273 }
274 }
275
276 #[must_use]
277 pub const fn registered(runner_id: u64, busy: bool) -> Self {
278 Self {
279 status: GithubRunnerObservation::Registered { busy },
280 runner_id: Some(runner_id),
281 }
282 }
283}
284
285#[async_trait]
288pub trait LifecycleGithub: fmt::Debug + Send + Sync {
289 async fn register(
290 &self,
291 target: &ScaleTarget,
292 request: &JitRunnerRequest,
293 cancel: &CancelToken,
294 ) -> Result<JitRegistration, JitRequestFailure>;
295
296 async fn observe(
297 &self,
298 target: &ScaleTarget,
299 attempt: AttemptId,
300 cancel: &CancelToken,
301 ) -> LifecycleGithubObservation;
302
303 async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool;
310}
311
312#[async_trait]
313impl<T> LifecycleGithub for T
314where
315 T: JitGateway + InventoryGateway + fmt::Debug + Send + Sync,
316{
317 async fn register(
318 &self,
319 target: &ScaleTarget,
320 request: &JitRunnerRequest,
321 cancel: &CancelToken,
322 ) -> Result<JitRegistration, JitRequestFailure> {
323 self.generate_jit_config(target, request, cancel)
324 .await
325 .map_err(|error| {
326 let reason = if matches!(&error, JitError::Forbidden { .. }) {
327 FailureReason::Other(
328 "GitHub refused JIT registration with 403; check the App's runner permission and runner-group access"
329 .into(),
330 )
331 } else {
332 FailureReason::JitRequestFailed
333 };
334 JitRequestFailure {
335 terminal: error.is_terminal(),
336 reason,
337 retry_after: error
338 .rate_limited()
339 .map(|limit| limit.delay_from(self.now())),
340 }
341 })
342 }
343
344 async fn observe(
345 &self,
346 target: &ScaleTarget,
347 attempt: AttemptId,
348 cancel: &CancelToken,
349 ) -> LifecycleGithubObservation {
350 let expected_name = runner_name(attempt);
351 match self.list_runners(target, cancel).await {
352 Ok(inventory) => inventory
353 .runners()
354 .iter()
355 .find(|runner| runner.name == expected_name)
356 .map_or(LifecycleGithubObservation::not_registered(), |runner| {
357 LifecycleGithubObservation::registered(runner.id, runner.busy)
358 }),
359 Err(_) => LifecycleGithubObservation::unreachable(),
360 }
361 }
362
363 async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool {
364 self.remove_runner(target, runner_id, cancel).await.is_ok()
365 }
366}
367
368#[async_trait]
370pub trait RuntimePackages: fmt::Debug + Send + Sync {
371 async fn materialize(&self, attempt: &RunnerAttempt) -> Result<RunnerVersion, FailureReason>;
372 fn release(&self, attempt: AttemptId) -> Result<(), FailureReason>;
373 fn prune_obsolete_guarded(
374 &self,
375 authority: PruneAuthority<'_>,
376 current: &RunnerVersion,
377 attempts: &[RunnerAttempt],
378 ) -> Result<(), FailureReason>;
379}
380
381pub struct PruneAuthority<'a> {
386 _guard: &'a AllocationGuard,
387}
388
389impl fmt::Debug for PruneAuthority<'_> {
390 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
391 f.write_str("PruneAuthority")
392 }
393}
394
395impl<'a> PruneAuthority<'a> {
396 fn from_launch_request(guard: &'a AllocationGuard) -> Self {
397 Self { _guard: guard }
398 }
399}
400
401#[derive(Debug)]
404pub struct CachedRuntimePackages {
405 cache: Arc<PackageCache>,
406}
407
408impl CachedRuntimePackages {
409 #[must_use]
410 pub fn new(cache: Arc<PackageCache>) -> Self {
411 Self { cache }
412 }
413}
414
415#[async_trait]
416impl RuntimePackages for CachedRuntimePackages {
417 async fn materialize(&self, attempt: &RunnerAttempt) -> Result<RunnerVersion, FailureReason> {
418 let installed = self
419 .cache
420 .ensure_installed()
421 .await
422 .map_err(package_failure)?;
423 copy_package_tree(installed.root(), attempt.runtime_path())
424 .map_err(|_| FailureReason::ProcessStartFailed)?;
425 if let Err(error) = self.cache.lease(attempt, installed.version()) {
426 let _ = remove_materialized_package(attempt);
430 return Err(package_failure(error));
431 }
432 Ok(installed.version().clone())
433 }
434
435 fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
436 self.cache.release(attempt).map_err(package_failure)
437 }
438
439 fn prune_obsolete_guarded(
440 &self,
441 _authority: PruneAuthority<'_>,
442 current: &RunnerVersion,
443 attempts: &[RunnerAttempt],
444 ) -> Result<(), FailureReason> {
445 for installed in self.cache.installed().map_err(package_failure)? {
446 if installed.version() != current {
447 match self.cache.prune(installed.version(), attempts) {
448 Ok(()) | Err(PackageError::VersionInUse { .. }) => {}
449 Err(error) => return Err(package_failure(error)),
450 }
451 }
452 }
453 Ok(())
454 }
455}
456
457fn package_failure(error: PackageError) -> FailureReason {
458 error.failure_reason().unwrap_or(FailureReason::Other(
459 "runner package cache operation failed".into(),
460 ))
461}
462
463fn package_failure_is_terminal(reason: &FailureReason) -> bool {
464 matches!(
465 reason,
466 FailureReason::RunnerPackageUnverified | FailureReason::RunnerVersionRejected
467 )
468}
469
470const WORKSPACE_NAME_LEN: usize = 12;
493
494fn workspace_name(id: AttemptId) -> String {
496 let full = id.to_string();
497 full.chars()
498 .filter(|c| *c != '-')
499 .take(WORKSPACE_NAME_LEN)
500 .collect()
501}
502
503#[derive(Debug, Clone)]
510struct Placement {
511 runtime: PathBuf,
512 workspace: AttemptWorkspace,
513}
514
515fn root_failure(error: RunnerRootError) -> LifecycleError {
522 tracing::warn!(
539 error_kind = error.kind(),
540 "the runner root refused this launch, so no attempt was created; the host will \
541 retry every poll until the cause is resolved. Re-running `host set-runtime-root` \
542 with the same path re-runs this check and prints the directory and the \
543 remediation in full"
544 );
545 LifecycleError::Failed(FailureReason::Other(error.to_string()))
546}
547
548fn lowest_free_slot(leases: &[RunnerAttempt], ceiling: NonZeroU16) -> Option<NonZeroU16> {
557 let held: BTreeSet<u16> = leases
558 .iter()
559 .filter_map(|attempt| attempt.workspace().slot_number())
560 .collect();
561 (1..=ceiling.get())
562 .find(|slot| !held.contains(slot))
563 .and_then(NonZeroU16::new)
564}
565
566fn create_or_validate_slot(slot: &Path) -> Result<(), LifecycleError> {
574 match fs::symlink_metadata(slot) {
575 Ok(metadata) if is_link_like(&metadata) => Err(slot_refusal(
581 slot,
582 "is a symbolic link, junction or other reparse point, which could place runner \
583 files outside the configured root",
584 )),
585 Ok(metadata) if !metadata.is_dir() => Err(slot_refusal(slot, "is not a directory")),
586 Ok(_) => Ok(()),
587 Err(error) if error.kind() == std::io::ErrorKind::NotFound => fs::create_dir(slot)
588 .map_err(|source| slot_refusal(slot, format!("could not be created: {source}"))),
589 Err(source) => Err(slot_refusal(
590 slot,
591 format!("could not be inspected: {source}"),
592 )),
593 }
594}
595
596fn accept_reusable_slot(slot: &Path) -> Result<(), LifecycleError> {
608 let unreadable =
609 |source: std::io::Error| slot_refusal(slot, format!("could not be read: {source}"));
610 let entries = fs::read_dir(slot).map_err(unreadable)?;
611 let mut refused: Vec<String> = Vec::new();
612 for entry in entries {
613 let entry = entry.map_err(unreadable)?;
614 let name = entry.file_name();
615 let metadata = fs::symlink_metadata(entry.path()).map_err(|source| {
616 slot_refusal(
617 slot,
618 format!("entry {name:?} could not be inspected: {source}"),
619 )
620 })?;
621 if is_retainable_work_folder(&name, &metadata) {
626 continue;
627 }
628 refused.push(name.to_string_lossy().into_owned());
629 }
630 if refused.is_empty() {
631 return Ok(());
632 }
633 refused.sort();
634 Err(slot_refusal(
635 slot,
636 format!(
637 "holds {} that this attempt may not reuse: [{}]. A reusable slot is empty or holds \
638 one real `{DEFAULT_WORK_FOLDER}` directory and nothing else; remove or move the \
639 entries listed, or let cleanup and recovery resolve them",
640 if refused.len() == 1 {
641 "an entry"
642 } else {
643 "entries"
644 },
645 refused.join(", ")
646 ),
647 ))
648}
649
650fn slot_refusal(slot: &Path, detail: impl fmt::Display) -> LifecycleError {
651 LifecycleError::Failed(FailureReason::Other(format!(
652 "the persistent slot {} {detail}",
653 slot.display()
654 )))
655}
656
657fn is_work_folder(name: &OsStr) -> bool {
667 if cfg!(windows) {
668 name.eq_ignore_ascii_case(DEFAULT_WORK_FOLDER)
669 } else {
670 name == OsStr::new(DEFAULT_WORK_FOLDER)
671 }
672}
673
674fn is_link_like(metadata: &fs::Metadata) -> bool {
683 if metadata.file_type().is_symlink() {
684 return true;
685 }
686 #[cfg(windows)]
687 {
688 use std::os::windows::fs::MetadataExt;
689
690 const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x0000_0400;
691 metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0
692 }
693 #[cfg(not(windows))]
694 false
695}
696
697fn is_retainable_work_folder(name: &OsStr, metadata: &fs::Metadata) -> bool {
705 is_work_folder(name) && metadata.is_dir() && !is_link_like(metadata)
706}
707
708fn remove_materialized_package(attempt: &RunnerAttempt) -> std::io::Result<()> {
715 match attempt.workspace() {
716 AttemptWorkspace::Ephemeral => remove_runtime_tree(attempt.runtime_path()),
717 AttemptWorkspace::PersistentSlot { .. } => scrub_slot_entries(attempt.runtime_path())
718 .map_err(|quarantine| std::io::Error::other(quarantine.to_string())),
719 }
720}
721
722fn remove_runtime_tree(path: &Path) -> std::io::Result<()> {
730 #[cfg(windows)]
731 {
732 remove_dir_all::remove_dir_all(path)
733 }
734 #[cfg(not(windows))]
735 {
736 fs::remove_dir_all(path)
737 }
738}
739
740#[derive(Debug, Clone, Copy, PartialEq, Eq)]
755enum SlotRefusal {
756 NotTheJournalledSlot,
758 PolicyRootDisagrees,
760 Containment,
762 SlotNotADirectory,
764 Enumeration,
766 WorkNotADirectory,
768 Deletion,
770 Residue,
772}
773
774impl SlotRefusal {
775 const fn class(self) -> &'static str {
777 match self {
778 Self::NotTheJournalledSlot => "slot_path_is_not_the_journalled_slot",
779 Self::PolicyRootDisagrees => "slot_root_disagrees_with_policy",
780 Self::Containment => "slot_escapes_its_root",
781 Self::SlotNotADirectory => "slot_is_not_a_directory",
782 Self::Enumeration => "slot_could_not_be_enumerated",
783 Self::WorkNotADirectory => "retained_work_is_not_a_directory",
784 Self::Deletion => "slot_entry_could_not_be_removed",
785 Self::Residue => "slot_still_holds_runner_state",
786 }
787 }
788
789 const fn remediation(self) -> &'static str {
791 match self {
792 Self::NotTheJournalledSlot | Self::PolicyRootDisagrees | Self::Containment => {
793 "the attempt keeps its slot lease and nothing was removed; correct the \
794 repository's persistent workspace path, or remove the slot directory by hand \
795 once you have confirmed what is in it"
796 }
797 Self::SlotNotADirectory | Self::WorkNotADirectory => {
798 "the attempt keeps its slot lease and nothing was removed; a job replaced the \
799 slot or its `_work` with a link, so inspect it before deleting anything and \
800 treat the retained workspace as untrusted"
801 }
802 Self::Enumeration | Self::Deletion | Self::Residue => {
803 "the attempt keeps its slot lease and will be cleaned again on the next pass; \
804 release whatever is holding the files open, or remove the slot's contents by \
805 hand leaving only `_work`"
806 }
807 }
808 }
809}
810
811#[derive(Debug, Clone, PartialEq, Eq)]
818struct SlotQuarantine {
819 refusal: SlotRefusal,
820 detail: String,
821}
822
823impl SlotQuarantine {
824 fn new(refusal: SlotRefusal, detail: impl Into<String>) -> Self {
825 Self {
826 refusal,
827 detail: detail.into(),
828 }
829 }
830}
831
832impl fmt::Display for SlotQuarantine {
833 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
834 write!(f, "{}; {}", self.detail, self.refusal.remediation())
835 }
836}
837
838fn verify_journalled_slot(
852 runtime: &Path,
853 slot: NonZeroU16,
854 configured: Option<&LocalAbsolutePath>,
855) -> Result<(), SlotQuarantine> {
856 let mislaid = || {
857 SlotQuarantine::new(
858 SlotRefusal::NotTheJournalledSlot,
859 format!(
860 "the journalled runtime {} is not the slot s{slot} this attempt was allocated as",
861 runtime.display()
862 ),
863 )
864 };
865 let local = |path: &Path| {
866 path.to_str()
867 .and_then(|raw| LocalAbsolutePath::new(raw).ok())
868 .ok_or_else(mislaid)
869 };
870
871 let runtime_path = local(runtime)?;
872 let root = local(runtime.parent().ok_or_else(mislaid)?)?;
873 let name = AttemptWorkspace::persistent_slot(slot)
880 .slot_directory_name()
881 .expect("a persistent workspace names its slot directory");
882 let derived = runner_root::derive_child(&root, &name).map_err(|_| mislaid())?;
883 if derived != runtime_path {
884 return Err(mislaid());
885 }
886 if let Some(configured) = configured
887 && configured != &root
888 {
889 return Err(SlotQuarantine::new(
890 SlotRefusal::PolicyRootDisagrees,
891 format!(
892 "the journalled slot {} is not under the repository's configured persistent root \
893 {}",
894 runtime.display(),
895 configured.as_str()
896 ),
897 ));
898 }
899 runner_root::verify_containment(&root, &derived).map_err(|source| {
902 SlotQuarantine::new(
903 SlotRefusal::Containment,
904 format!("the journalled slot is not inside the root it was allocated from: {source}"),
905 )
906 })
907}
908
909fn slot_is_present(slot: &Path) -> Result<bool, SlotQuarantine> {
920 match fs::symlink_metadata(slot) {
921 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
922 Err(source) => Err(SlotQuarantine::new(
923 SlotRefusal::SlotNotADirectory,
924 format!(
925 "the slot {} could not be inspected: {:?}",
926 slot.display(),
927 source.kind()
928 ),
929 )),
930 Ok(metadata) if !metadata.is_dir() || is_link_like(&metadata) => Err(SlotQuarantine::new(
931 SlotRefusal::SlotNotADirectory,
932 format!(
933 "the slot {} is a link or a file rather than a real directory",
934 slot.display()
935 ),
936 )),
937 Ok(_) => Ok(true),
938 }
939}
940
941fn scrub_slot_entries(slot: &Path) -> Result<(), SlotQuarantine> {
956 let unreadable = |source: std::io::Error| {
957 SlotQuarantine::new(
958 SlotRefusal::Enumeration,
959 format!(
960 "the entries of {} could not be listed: {:?}",
961 slot.display(),
962 source.kind()
963 ),
964 )
965 };
966 for entry in fs::read_dir(slot).map_err(unreadable)? {
967 let name = entry.map_err(unreadable)?.file_name();
968 let path = slot.join(&name);
969 let Some(metadata) = listed_entry_metadata(&path).map_err(unreadable)? else {
972 continue;
973 };
974 if is_work_folder(&name) {
975 if is_retainable_work_folder(&name, &metadata) {
976 continue;
977 }
978 return Err(SlotQuarantine::new(
979 SlotRefusal::WorkNotADirectory,
980 format!(
981 "the retained `{DEFAULT_WORK_FOLDER}` in {} is a link or a file rather than a \
982 real directory",
983 slot.display()
984 ),
985 ));
986 }
987 remove_slot_entry(&path, &metadata).map_err(|source| {
988 SlotQuarantine::new(
989 SlotRefusal::Deletion,
990 format!(
991 "an entry of {} could not be removed: {:?}",
992 slot.display(),
993 source.kind()
994 ),
995 )
996 })?;
997 }
998 Ok(())
999}
1000
1001fn listed_entry_metadata(path: &Path) -> std::io::Result<Option<fs::Metadata>> {
1007 match fs::symlink_metadata(path) {
1008 Ok(metadata) => Ok(Some(metadata)),
1009 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
1010 Err(source) => Err(source),
1011 }
1012}
1013
1014fn remove_slot_entry(path: &Path, metadata: &fs::Metadata) -> std::io::Result<()> {
1019 let removed = if is_link_like(metadata) {
1020 fs::remove_file(path).or_else(|_| fs::remove_dir(path))
1023 } else if metadata.is_dir() {
1024 remove_runtime_tree(path)
1025 } else {
1026 fs::remove_file(path)
1027 };
1028 match removed {
1029 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
1030 other => other,
1031 }
1032}
1033
1034fn verify_slot_scrubbed(slot: &Path) -> Result<(), SlotQuarantine> {
1051 let unreadable = |source: std::io::Error| {
1052 SlotQuarantine::new(
1053 SlotRefusal::Enumeration,
1054 format!(
1055 "the entries of {} could not be listed to verify the scrub: {:?}",
1056 slot.display(),
1057 source.kind()
1058 ),
1059 )
1060 };
1061 let mut residue = 0_usize;
1062 let mut named: Vec<String> = Vec::new();
1063 for entry in fs::read_dir(slot).map_err(unreadable)? {
1064 let name = entry.map_err(unreadable)?.file_name();
1065 let Some(metadata) = listed_entry_metadata(&slot.join(&name)).map_err(unreadable)? else {
1069 continue;
1070 };
1071 if is_retainable_work_folder(&name, &metadata) {
1072 continue;
1073 }
1074 residue = residue.saturating_add(1);
1075 if name
1076 .to_string_lossy()
1077 .starts_with(RestrictiveHandoff::NAME_PREFIX)
1078 {
1079 named.push("an encoded JIT handoff".to_owned());
1080 }
1081 }
1082 named.extend(
1083 SENSITIVE_SLOT_ENTRIES
1084 .iter()
1085 .filter(|entry| fs::symlink_metadata(slot.join(entry)).is_ok())
1086 .map(|entry| format!("`{entry}`")),
1087 );
1088 if residue == 0 && named.is_empty() {
1089 return Ok(());
1090 }
1091 named.sort_unstable();
1092 named.dedup();
1093 Err(SlotQuarantine::new(
1094 SlotRefusal::Residue,
1095 residue_detail(slot, residue, &named),
1096 ))
1097}
1098
1099fn residue_detail(slot: &Path, residue: usize, named: &[String]) -> String {
1109 if residue == 0 {
1110 format!(
1114 "the listing of {} reported nothing but `{DEFAULT_WORK_FOLDER}`, yet {} survived \
1115 cleanup",
1116 slot.display(),
1117 named.join(", ")
1118 )
1119 } else {
1120 format!(
1121 "{residue} entr{} other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}{}",
1122 if residue == 1 { "y" } else { "ies" },
1123 slot.display(),
1124 if named.is_empty() {
1125 String::new()
1126 } else {
1127 format!(", including {}", named.join(", "))
1128 }
1129 )
1130 }
1131}
1132
1133fn replacement_operation(outcome: &AttemptOutcome) -> Option<&'static str> {
1134 match outcome {
1135 AttemptOutcome::Failed {
1136 reason: FailureReason::JitExpired,
1137 } => Some("jit_expired_replacement"),
1138 AttemptOutcome::Failed {
1139 reason: FailureReason::ProcessExitedUnexpectedly,
1140 } => Some("exit_before_acceptance_replacement"),
1141 _ => None,
1142 }
1143}
1144
1145fn copy_package_tree(source: &Path, destination: &Path) -> std::io::Result<()> {
1160 if source.join(DEFAULT_WORK_FOLDER).exists() {
1161 return Err(std::io::Error::new(
1162 std::io::ErrorKind::InvalidData,
1163 "a cached runner package contains a _work folder, which means it was used \
1164 to run a job before it was archived; the cache must only contain clean \
1165 extracts to prevent data leakage",
1166 ));
1167 }
1168
1169 #[cfg(unix)]
1170 {
1171 let status = std::process::Command::new("cp")
1172 .arg("-a")
1173 .arg(format!("{}/.", source.display()))
1174 .arg(destination)
1175 .status()?;
1176 if status.success() {
1177 Ok(())
1178 } else {
1179 Err(std::io::Error::other("cp failed"))
1180 }
1181 }
1182 #[cfg(not(unix))]
1183 copy_package_entries(source, destination, true)
1184}
1185
1186#[cfg(not(unix))]
1187fn copy_package_entries(source: &Path, destination: &Path, top_level: bool) -> std::io::Result<()> {
1188 fs::create_dir_all(destination)?;
1189 for entry in fs::read_dir(source)? {
1190 let entry = entry?;
1191 if top_level && is_work_folder(&entry.file_name()) {
1192 return Err(std::io::Error::new(
1193 std::io::ErrorKind::InvalidData,
1194 format!(
1195 "the runner package holds a top-level `{DEFAULT_WORK_FOLDER}`; copying \
1196 it would overwrite the job workspace a persistent slot retains"
1197 ),
1198 ));
1199 }
1200 let target = destination.join(entry.file_name());
1201 if entry.file_type()?.is_dir() {
1202 copy_package_entries(&entry.path(), &target, false)?;
1203 } else {
1204 fs::copy(entry.path(), target)?;
1205 }
1206 }
1207 Ok(())
1208}
1209
1210#[derive(Debug, Clone, PartialEq, Eq)]
1213pub struct ProcessStartFailure {
1214 pub reason: FailureReason,
1215 pub retryable: bool,
1218 pub live_pid: Option<u32>,
1221}
1222
1223impl ProcessStartFailure {
1224 fn before_spawn(reason: FailureReason) -> Self {
1225 Self {
1226 reason,
1227 retryable: true,
1228 live_pid: None,
1229 }
1230 }
1231
1232 fn after_spawn_stopped() -> Self {
1233 Self {
1234 reason: FailureReason::ProcessStartFailed,
1235 retryable: false,
1236 live_pid: None,
1237 }
1238 }
1239
1240 fn after_spawn_live(pid: u32) -> Self {
1241 Self::after_spawn_live_with_reason(pid, FailureReason::ProcessStartFailed)
1242 }
1243
1244 fn after_spawn_live_with_reason(pid: u32, reason: FailureReason) -> Self {
1245 Self {
1246 reason,
1247 retryable: false,
1248 live_pid: Some(pid),
1249 }
1250 }
1251}
1252
1253pub trait ProcessSupervisor: fmt::Debug + Send + Sync {
1254 fn spawn(
1255 &self,
1256 attempt: &RunnerAttempt,
1257 config: &EncodedJitConfig,
1258 ) -> Result<u32, ProcessStartFailure>;
1259 fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason>;
1260 fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason>;
1263 fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool;
1266 fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
1267 fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool;
1268 fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
1269}
1270
1271#[derive(Debug, Default)]
1274pub struct NativeProcesses {
1275 children: Mutex<BTreeMap<AttemptId, ChildProcess>>,
1276 successful_exits: Mutex<BTreeMap<AttemptId, bool>>,
1277 #[cfg(test)]
1278 post_spawn_faults: Mutex<VecDeque<PostSpawnBoundary>>,
1279 #[cfg(test)]
1280 post_spawn_reaps: std::sync::atomic::AtomicUsize,
1281 #[cfg(test)]
1282 post_spawn_stop_failures: std::sync::atomic::AtomicUsize,
1283 #[cfg(test)]
1284 use_long_lived_test_listener: std::sync::atomic::AtomicBool,
1285}
1286
1287#[cfg(test)]
1288#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1289enum PostSpawnBoundary {
1290 HandoffDelete,
1291 IdentitySerialize,
1292 IdentityWrite,
1293 ChildMapInsert,
1294}
1295
1296impl NativeProcesses {
1297 #[must_use]
1298 pub fn new() -> Self {
1299 Self::default()
1300 }
1301
1302 #[cfg(test)]
1303 fn fail_post_spawn_at(&self, boundary: PostSpawnBoundary) {
1304 self.post_spawn_faults.lock().unwrap().push_back(boundary);
1305 }
1306
1307 #[cfg(test)]
1308 fn faults_at(&self, boundary: PostSpawnBoundary) -> bool {
1309 let mut faults = self.post_spawn_faults.lock().unwrap();
1310 if faults.front() == Some(&boundary) {
1311 faults.pop_front();
1312 true
1313 } else {
1314 false
1315 }
1316 }
1317
1318 #[cfg(test)]
1319 fn fail_post_spawn_stops(&self, count: usize) {
1320 self.post_spawn_stop_failures
1321 .fetch_add(count, std::sync::atomic::Ordering::SeqCst);
1322 }
1323
1324 #[cfg(test)]
1325 fn fail_next_post_spawn_stop(&self) {
1326 self.fail_post_spawn_stops(1);
1327 }
1328
1329 #[cfg(test)]
1330 fn use_long_lived_test_listener(&self) {
1331 self.use_long_lived_test_listener
1332 .store(true, std::sync::atomic::Ordering::SeqCst);
1333 }
1334
1335 fn stop_spawned_child(&self, child: &mut ChildProcess) -> Result<(), FailureReason> {
1336 #[cfg(test)]
1337 if self
1338 .post_spawn_stop_failures
1339 .fetch_update(
1340 std::sync::atomic::Ordering::SeqCst,
1341 std::sync::atomic::Ordering::SeqCst,
1342 |left| if left > 0 { Some(left - 1) } else { None },
1343 )
1344 .is_ok()
1345 {
1346 return Err(FailureReason::Other("injected runner stop failure".into()));
1347 }
1348 child
1349 .stop(Duration::from_secs(1))
1350 .map(|_| ())
1351 .map_err(|_| FailureReason::Other("spawned runner process could not be stopped".into()))
1352 }
1353
1354 fn abort_spawned_child(
1355 &self,
1356 mut child: ChildProcess,
1357 attempt: &RunnerAttempt,
1358 remove_identity: bool,
1359 ) -> ProcessStartFailure {
1360 let mut reaped = self.stop_spawned_child(&mut child).is_ok();
1361 if reaped {
1362 if remove_identity {
1363 Self::remove_identity_files(attempt);
1364 }
1365 } else {
1366 let identity_durable =
1369 serde_json::to_vec(child.identity())
1370 .ok()
1371 .is_some_and(|identity| {
1372 self.persist_identity(attempt, &identity).is_ok()
1373 || self.persist_fallback_identity(attempt, &identity).is_ok()
1374 });
1375 if !identity_durable {
1376 for _ in 1..MAX_POST_SPAWN_STOP_ATTEMPTS {
1381 if self.stop_spawned_child(&mut child).is_ok() {
1382 reaped = true;
1383 break;
1384 }
1385 }
1386 if !reaped {
1387 let pid = child.pid();
1392 let marker = write_durable_file(
1393 &Self::unresolved_process_path(attempt),
1394 pid.to_string().as_bytes(),
1395 );
1396 self.children
1397 .lock()
1398 .unwrap_or_else(std::sync::PoisonError::into_inner)
1399 .insert(attempt.id, child);
1400 let reason = if marker.is_ok() {
1401 FailureReason::Other(
1402 "spawn cleanup exhausted its bounded stop attempts; the live process remains under durable unresolved supervision"
1403 .into(),
1404 )
1405 } else {
1406 FailureReason::Other(
1407 "spawn cleanup exhausted its bounded stop attempts and the unresolved-process marker could not be journalled"
1408 .into(),
1409 )
1410 };
1411 return ProcessStartFailure::after_spawn_live_with_reason(pid, reason);
1412 }
1413 } else {
1414 let pid = child.pid();
1415 self.children
1416 .lock()
1417 .unwrap_or_else(std::sync::PoisonError::into_inner)
1418 .insert(attempt.id, child);
1419 return ProcessStartFailure::after_spawn_live(pid);
1420 }
1421 }
1422 #[cfg(test)]
1423 if reaped {
1424 self.post_spawn_reaps
1425 .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1426 }
1427 #[cfg(not(test))]
1428 let _ = reaped;
1429 ProcessStartFailure::after_spawn_stopped()
1430 }
1431
1432 fn identity_path(attempt: &RunnerAttempt) -> PathBuf {
1433 attempt.runtime_path().join(IDENTITY_FILE)
1434 }
1435
1436 fn fallback_identity_path(attempt: &RunnerAttempt) -> PathBuf {
1437 attempt.runtime_path().join(FALLBACK_IDENTITY_FILE)
1438 }
1439
1440 fn unresolved_process_path(attempt: &RunnerAttempt) -> PathBuf {
1441 attempt.runtime_path().join(UNRESOLVED_PROCESS_FILE)
1442 }
1443
1444 fn remove_identity_files(attempt: &RunnerAttempt) {
1445 let _ = fs::remove_file(Self::identity_path(attempt));
1446 let _ = fs::remove_file(Self::fallback_identity_path(attempt));
1447 let _ = fs::remove_file(Self::unresolved_process_path(attempt));
1448 }
1449
1450 fn persist_identity(&self, attempt: &RunnerAttempt, bytes: &[u8]) -> std::io::Result<()> {
1451 self.persist_identity_at(&Self::identity_path(attempt), bytes)
1452 }
1453
1454 fn persist_fallback_identity(
1455 &self,
1456 attempt: &RunnerAttempt,
1457 bytes: &[u8],
1458 ) -> std::io::Result<()> {
1459 self.persist_identity_at(&Self::fallback_identity_path(attempt), bytes)
1460 }
1461
1462 fn persist_identity_at(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
1463 #[cfg(test)]
1464 if self.faults_at(PostSpawnBoundary::IdentityWrite) {
1465 return Err(std::io::Error::other("injected identity write failure"));
1466 }
1467 write_durable_file(path, bytes)
1468 }
1469
1470 fn intent_path(attempt: &RunnerAttempt) -> PathBuf {
1471 attempt.runtime_path().join(TERMINATE_INTENT_FILE)
1472 }
1473
1474 fn read_identity(attempt: &RunnerAttempt) -> Result<Option<ProcessIdentity>, FailureReason> {
1475 match Self::read_identity_at(&Self::identity_path(attempt))? {
1476 Some(identity) => Ok(Some(identity)),
1477 None => Self::read_identity_at(&Self::fallback_identity_path(attempt)),
1478 }
1479 }
1480
1481 fn read_identity_at(path: &Path) -> Result<Option<ProcessIdentity>, FailureReason> {
1482 match fs::read(path) {
1483 Ok(bytes) => serde_json::from_slice(&bytes)
1484 .map(Some)
1485 .map_err(|_| FailureReason::Other("process identity journal is unreadable".into())),
1486 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
1487 Err(_) => Err(FailureReason::Other(
1488 "process identity journal could not be read".into(),
1489 )),
1490 }
1491 }
1492}
1493
1494impl ProcessSupervisor for NativeProcesses {
1495 fn spawn(
1496 &self,
1497 attempt: &RunnerAttempt,
1498 config: &EncodedJitConfig,
1499 ) -> Result<u32, ProcessStartFailure> {
1500 let handoff = RestrictiveHandoff::create(
1501 attempt.runtime_path(),
1502 SecretString::from(config.expose().to_owned()),
1503 )
1504 .map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
1505 #[cfg(windows)]
1506 let program = attempt
1507 .runtime_path()
1508 .join("bin")
1509 .join("Runner.Listener.exe");
1510 #[cfg(not(windows))]
1511 let program = attempt.runtime_path().join("bin").join("Runner.Listener");
1512 if !program.is_file() {
1515 return Err(ProcessStartFailure::before_spawn(
1516 FailureReason::ProcessStartFailed,
1517 ));
1518 }
1519 #[cfg(test)]
1520 let spec = if self
1521 .use_long_lived_test_listener
1522 .load(std::sync::atomic::Ordering::SeqCst)
1523 {
1524 SpawnSpec::new(program)
1525 .args([
1526 "--ignored",
1527 "--exact",
1528 "lifecycle::tests::long_lived_native_listener_helper",
1529 "--nocapture",
1530 ])
1531 .env(
1532 "RUNNER_MANAGER_TEST_LISTENER_READY",
1533 attempt.runtime_path().join(TEST_LISTENER_READY),
1534 )
1535 .working_dir(attempt.runtime_path())
1536 } else {
1537 runner_listener_spec(program, attempt.runtime_path())
1538 };
1539 #[cfg(not(test))]
1540 let spec = runner_listener_spec(program, attempt.runtime_path());
1541 let child = spec
1542 .spawn_runner_with_handoff(&handoff)
1543 .map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
1544 #[cfg(test)]
1546 if self.faults_at(PostSpawnBoundary::HandoffDelete) {
1547 drop(handoff);
1548 return Err(self.abort_spawned_child(child, attempt, false));
1549 }
1550 if handoff.delete().is_err() {
1551 return Err(self.abort_spawned_child(child, attempt, false));
1552 }
1553 #[cfg(test)]
1554 if self.faults_at(PostSpawnBoundary::IdentitySerialize) {
1555 return Err(self.abort_spawned_child(child, attempt, false));
1556 }
1557 let identity = match serde_json::to_vec(child.identity()) {
1558 Ok(identity) => identity,
1559 Err(_) => {
1560 return Err(self.abort_spawned_child(child, attempt, false));
1561 }
1562 };
1563 if self.persist_identity(attempt, &identity).is_err() {
1564 return Err(self.abort_spawned_child(child, attempt, true));
1565 }
1566 let pid = child.pid();
1567 #[cfg(test)]
1568 if self.faults_at(PostSpawnBoundary::ChildMapInsert) {
1569 return Err(self.abort_spawned_child(child, attempt, true));
1570 }
1571 let mut children = self
1572 .children
1573 .lock()
1574 .unwrap_or_else(std::sync::PoisonError::into_inner);
1575 children.insert(attempt.id, child);
1576 Ok(pid)
1577 }
1578
1579 fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
1580 let mut children = self
1581 .children
1582 .lock()
1583 .unwrap_or_else(std::sync::PoisonError::into_inner);
1584 if let Some(child) = children.get_mut(&attempt.id) {
1585 return match child
1586 .try_exit_status()
1587 .map_err(|_| FailureReason::Other("runner process could not be observed".into()))?
1588 {
1589 None => Ok(true),
1590 Some(status) => {
1591 if let Ok(mut exits) = self.successful_exits.lock() {
1592 exits.insert(attempt.id, status.success());
1593 }
1594 Ok(false)
1595 }
1596 };
1597 }
1598 let Some(identity) = Self::read_identity(attempt)? else {
1599 if attempt.process_id().is_some() || Self::unresolved_process_path(attempt).is_file() {
1600 return Err(FailureReason::Other(
1601 "runner process identity is missing; refusing recovery until the process is resolved"
1602 .into(),
1603 ));
1604 }
1605 return Ok(false);
1606 };
1607 match identity.recheck() {
1608 Ok(Adoption::Live) => Ok(true),
1609 Ok(Adoption::Gone | Adoption::PidRecycled { .. }) => Ok(false),
1610 Err(_) => Ok(false),
1611 }
1612 }
1613
1614 fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
1615 Ok(Self::read_identity(attempt)?.map(|identity| identity.pid()))
1616 }
1617
1618 fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool {
1619 self.successful_exits
1620 .lock()
1621 .ok()
1622 .and_then(|exits| exits.get(&attempt.id).copied())
1623 .unwrap_or(false)
1624 }
1625
1626 fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
1627 let path = Self::intent_path(attempt);
1628 write_durable_file(&path, b"registration-timeout\n")
1629 .map_err(|_| FailureReason::Other("terminate intent could not be journalled".into()))
1630 }
1631
1632 fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool {
1633 Self::intent_path(attempt).is_file()
1634 }
1635
1636 fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
1637 let mut children = self
1638 .children
1639 .lock()
1640 .unwrap_or_else(std::sync::PoisonError::into_inner);
1641 if let Some(child) = children.get_mut(&attempt.id) {
1642 child
1643 .stop(Duration::from_secs(10))
1644 .map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?;
1645 return Ok(());
1646 }
1647 let Some(identity) = Self::read_identity(attempt)? else {
1648 return Ok(());
1649 };
1650 match identity
1651 .terminate(Duration::from_secs(10))
1652 .map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?
1653 {
1654 Termination::Terminated | Termination::AlreadyGone => Ok(()),
1655 Termination::RefusedPidRecycled { .. } => Err(FailureReason::Other(
1656 "runner PID was recycled; refusing to signal it".into(),
1657 )),
1658 }
1659 }
1660}
1661
1662pub struct LifecyclePorts {
1663 pub store: Arc<dyn Store>,
1664 pub github: Arc<dyn LifecycleGithub>,
1665 pub packages: Arc<dyn RuntimePackages>,
1666 pub processes: Arc<dyn ProcessSupervisor>,
1667 pub clock: Arc<dyn Clock>,
1668 pub demand: Arc<dyn DemandPersistence>,
1669 pub delay: Arc<dyn RetryDelay>,
1670 pub events: Arc<dyn AttemptEventSink>,
1671 pub reconcile_events: Arc<dyn EventSink>,
1672}
1673
1674impl fmt::Debug for LifecyclePorts {
1675 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1676 f.debug_struct("LifecyclePorts")
1677 .field("store", &self.store)
1678 .field("github", &self.github)
1679 .field("packages", &self.packages)
1680 .field("processes", &self.processes)
1681 .finish_non_exhaustive()
1682 }
1683}
1684
1685#[derive(Debug, thiserror::Error)]
1686pub enum LifecycleError {
1687 #[error("attempt journal operation failed")]
1688 Journal,
1689 #[error("attempt {0} is not in the journal")]
1690 Missing(AttemptId),
1691 #[error("attempt lifecycle transition was refused")]
1692 Transition,
1693 #[error("startup recovery has not completed")]
1694 RecoveryIncomplete,
1695 #[error("the persistent slot was not cleaned: {detail}")]
1705 SlotQuarantined {
1706 class: &'static str,
1708 detail: String,
1709 },
1710 #[error("runner lifecycle failed: {0}")]
1711 Failed(FailureReason),
1712}
1713
1714impl LifecycleError {
1715 fn reason(&self) -> FailureReason {
1716 match self {
1717 Self::Failed(reason) => reason.clone(),
1718 Self::RecoveryIncomplete => FailureReason::Other("startup recovery incomplete".into()),
1719 Self::Journal => FailureReason::Other("attempt journal operation failed".into()),
1720 Self::Missing(_) => FailureReason::Other("attempt disappeared from the journal".into()),
1721 Self::Transition => FailureReason::Other("attempt transition was refused".into()),
1722 Self::SlotQuarantined { .. } => FailureReason::Other(self.to_string()),
1725 }
1726 }
1727}
1728
1729#[derive(Debug)]
1731pub struct LifecycleLauncher {
1732 host_id: HostId,
1733 app_paths: runner_manager_platform::paths::AppPaths,
1734 diagnostics_root: PathBuf,
1735 runner_group_id: u64,
1736 timeouts: RecoveryTimeouts,
1737 retry: RetryPolicy,
1738 cancel: CancelToken,
1739 ports: LifecyclePorts,
1740 recovery_complete: Mutex<bool>,
1741 versions: Mutex<BTreeMap<AttemptId, RunnerVersion>>,
1742 pending_replacements: Mutex<BTreeMap<AttemptId, ReplacementIntent>>,
1743}
1744
1745#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1746enum ReconcileProgress {
1747 Reconciled,
1748 Deferred,
1749 Replacement {
1750 attempt: AttemptId,
1751 operation: &'static str,
1752 },
1753}
1754
1755impl LifecycleLauncher {
1756 #[must_use]
1757 pub fn new(
1758 host_id: HostId,
1759 app_paths: runner_manager_platform::paths::AppPaths,
1760 diagnostics_root: impl Into<PathBuf>,
1761 runner_group_id: u64,
1762 timeouts: RecoveryTimeouts,
1763 retry: RetryPolicy,
1764 ports: LifecyclePorts,
1765 ) -> Self {
1766 Self {
1767 host_id,
1768 app_paths,
1769 diagnostics_root: diagnostics_root.into(),
1770 runner_group_id,
1771 timeouts,
1772 retry,
1773 cancel: CancelToken::new(),
1774 ports,
1775 recovery_complete: Mutex::new(false),
1776 versions: Mutex::new(BTreeMap::new()),
1777 pending_replacements: Mutex::new(BTreeMap::new()),
1778 }
1779 }
1780
1781 pub async fn recover_startup(
1785 &self,
1786 policies: &[ScalePolicy],
1787 ) -> Result<Vec<ReplacementIntent>, LifecycleError> {
1788 let by_id: BTreeMap<_, _> = policies.iter().map(|policy| (policy.id, policy)).collect();
1789 let attempts = self
1790 .ports
1791 .store
1792 .attempts()
1793 .map_err(|_| LifecycleError::Journal)?;
1794 let mut unresolved = false;
1795 for attempt in attempts {
1796 let Some(policy) = by_id.get(&attempt.policy_id) else {
1797 if !attempt.is_terminal() && attempt.state() != AttemptState::Cleaned {
1798 unresolved = true;
1799 }
1800 continue;
1801 };
1802 authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
1803 match self.reconcile_one(policy, attempt).await? {
1804 ReconcileProgress::Deferred => unresolved = true,
1805 ReconcileProgress::Replacement { attempt, operation } => {
1806 self.pending_replacements
1807 .lock()
1808 .map_err(|_| LifecycleError::Journal)?
1809 .insert(
1810 attempt,
1811 ReplacementIntent {
1812 policy: policy.id,
1813 previous_attempt: attempt,
1814 operation,
1815 },
1816 );
1817 }
1818 ReconcileProgress::Reconciled => {}
1819 }
1820 }
1821 if unresolved {
1822 return Err(LifecycleError::RecoveryIncomplete);
1823 }
1824 *self
1825 .recovery_complete
1826 .lock()
1827 .map_err(|_| LifecycleError::Journal)? = true;
1828 Ok(self
1829 .pending_replacements
1830 .lock()
1831 .map_err(|_| LifecycleError::Journal)?
1832 .values()
1833 .copied()
1834 .collect())
1835 }
1836
1837 pub async fn supervise(
1839 &self,
1840 policy: &ScalePolicy,
1841 ) -> Result<Vec<ReplacementIntent>, LifecycleError> {
1842 let mut replacements = Vec::new();
1843 self.pending_replacements
1844 .lock()
1845 .map_err(|_| LifecycleError::Journal)?
1846 .retain(|_, intent| {
1847 if intent.policy == policy.id {
1848 replacements.push(*intent);
1849 false
1850 } else {
1851 true
1852 }
1853 });
1854 let attempts = self
1855 .ports
1856 .store
1857 .attempts_for_policy(policy.id)
1858 .map_err(|_| LifecycleError::Journal)?;
1859 for attempt in attempts {
1860 authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
1861 if let ReconcileProgress::Replacement { attempt, operation } =
1862 self.reconcile_one(policy, attempt).await?
1863 {
1864 replacements.push(ReplacementIntent {
1865 policy: policy.id,
1866 previous_attempt: attempt,
1867 operation,
1868 });
1869 }
1870 }
1871 Ok(replacements)
1872 }
1873
1874 async fn reconcile_one(
1875 &self,
1876 policy: &ScalePolicy,
1877 mut attempt: RunnerAttempt,
1878 ) -> Result<ReconcileProgress, LifecycleError> {
1879 if attempt.state() == AttemptState::Cleaned {
1880 return Ok(ReconcileProgress::Reconciled);
1881 }
1882 if attempt.is_terminal() {
1883 self.clean_or_quarantine(&mut attempt)?;
1884 return Ok(ReconcileProgress::Reconciled);
1885 }
1886 let process_alive = self
1887 .ports
1888 .processes
1889 .is_alive(&attempt)
1890 .map_err(LifecycleError::Failed)?;
1891 let github = self
1892 .ports
1893 .github
1894 .observe(&policy.target, attempt.id, &self.cancel)
1895 .await;
1896
1897 if let Some(runner_id) = github.runner_id
1902 && read_runner_id(attempt.runtime_path()).is_none()
1903 {
1904 write_runner_id(attempt.runtime_path(), runner_id)?;
1905 self.ports
1906 .events
1907 .emit(AttemptEvent::RemoteIdentityRecovered {
1908 attempt: attempt.id,
1909 runner_id,
1910 });
1911 }
1912
1913 if attempt.state() == AttemptState::JitReceived
1918 && process_alive
1919 && let Some(pid) = self
1920 .ports
1921 .processes
1922 .recovered_pid(&attempt)
1923 .map_err(LifecycleError::Failed)?
1924 {
1925 attempt
1926 .started(pid, self.ports.clock.now())
1927 .map_err(|_| LifecycleError::Transition)?;
1928 self.record(&attempt)?;
1929 }
1930
1931 if attempt.state() == AttemptState::Busy
1936 && !process_alive
1937 && github.status == GithubRunnerObservation::NotRegistered
1938 && self.ports.processes.completed_successfully(&attempt)
1939 {
1940 self.conclude(&mut attempt, AttemptOutcome::CompletedJob)?;
1941 self.clean_or_quarantine(&mut attempt)?;
1942 return Ok(ReconcileProgress::Reconciled);
1943 }
1944
1945 if self.ports.processes.has_terminate_intent(&attempt) && !process_alive {
1948 self.deregister_runner(policy, &attempt).await;
1949 self.conclude(
1950 &mut attempt,
1951 AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout),
1952 )?;
1953 self.clean_or_quarantine(&mut attempt)?;
1954 return Ok(ReconcileProgress::Replacement {
1955 attempt: attempt.id,
1956 operation: "registration_timeout_replacement",
1957 });
1958 }
1959
1960 if matches!(
1966 attempt.state(),
1967 AttemptState::Allocated | AttemptState::JitReceived
1968 ) && !process_alive
1969 && matches!(github.status, GithubRunnerObservation::Registered { .. })
1970 {
1971 if attempt.state() == AttemptState::Allocated {
1972 attempt
1973 .jit_received(self.ports.clock.now())
1974 .map_err(|_| LifecycleError::Transition)?;
1975 self.record(&attempt)?;
1976 }
1977 self.deregister_runner(policy, &attempt).await;
1978 self.conclude(
1979 &mut attempt,
1980 AttemptOutcome::failed(FailureReason::JitExpired),
1981 )?;
1982 self.clean_or_quarantine(&mut attempt)?;
1983 return Ok(ReconcileProgress::Replacement {
1984 attempt: attempt.id,
1985 operation: "jit_expired_replacement",
1986 });
1987 }
1988
1989 match recovery_decision(
1990 &attempt,
1991 RecoveryObservation {
1992 process_alive,
1993 github: github.status,
1994 },
1995 self.timeouts,
1996 self.ports.clock.as_ref(),
1997 ) {
1998 RecoveryDecision::Nothing | RecoveryDecision::Wait => Ok(ReconcileProgress::Reconciled),
1999 RecoveryDecision::Defer => Ok(ReconcileProgress::Deferred),
2000 RecoveryDecision::Adopt => {
2001 self.ports.events.emit(AttemptEvent::Adopted {
2002 attempt: attempt.id,
2003 });
2004 Ok(ReconcileProgress::Reconciled)
2005 }
2006 RecoveryDecision::Clean => {
2007 self.clean_or_quarantine(&mut attempt)?;
2008 Ok(ReconcileProgress::Reconciled)
2009 }
2010 RecoveryDecision::Observe(state) => {
2011 let runner_id = attempt
2012 .github_runner_id()
2013 .or(github.runner_id)
2014 .or_else(|| read_runner_id(attempt.runtime_path()))
2015 .ok_or(LifecycleError::Transition)?;
2016 match state {
2017 AttemptState::JitReceived => attempt
2018 .jit_received(self.ports.clock.now())
2019 .map_err(|_| LifecycleError::Transition)?,
2020 AttemptState::Starting => {
2021 let pid = attempt.process_id().ok_or(LifecycleError::Transition)?;
2022 attempt
2023 .started(pid, self.ports.clock.now())
2024 .map_err(|_| LifecycleError::Transition)?;
2025 }
2026 AttemptState::Idle => attempt
2027 .registered_idle(runner_id, self.ports.clock.now())
2028 .map_err(|_| LifecycleError::Transition)?,
2029 AttemptState::Busy => attempt
2030 .assigned_job(runner_id, self.ports.clock.now())
2031 .map_err(|_| LifecycleError::Transition)?,
2032 _ => return Err(LifecycleError::Transition),
2033 }
2034 self.record(&attempt)?;
2035 Ok(ReconcileProgress::Reconciled)
2036 }
2037 RecoveryDecision::Conclude(outcome) => {
2038 let replacement = replacement_operation(&outcome);
2039 if matches!(github.status, GithubRunnerObservation::Registered { .. }) {
2044 self.deregister_runner(policy, &attempt).await;
2045 }
2046 self.conclude(&mut attempt, outcome)?;
2047 self.clean_or_quarantine(&mut attempt)?;
2048 Ok(
2049 replacement.map_or(ReconcileProgress::Reconciled, |operation| {
2050 ReconcileProgress::Replacement {
2051 attempt: attempt.id,
2052 operation,
2053 }
2054 }),
2055 )
2056 }
2057 RecoveryDecision::Terminate(payload) => {
2058 let idle_exit = payload.is_idle_exit();
2068 self.ports
2069 .processes
2070 .record_terminate_intent(&attempt)
2071 .map_err(LifecycleError::Failed)?;
2072 self.ports.events.emit(AttemptEvent::TerminateIntent {
2073 attempt: attempt.id,
2074 });
2075 self.ports
2076 .processes
2077 .terminate(&attempt)
2078 .map_err(LifecycleError::Failed)?;
2079 if self
2080 .ports
2081 .processes
2082 .is_alive(&attempt)
2083 .map_err(LifecycleError::Failed)?
2084 {
2085 return Ok(ReconcileProgress::Deferred);
2086 }
2087 self.ports.events.emit(AttemptEvent::Terminated {
2088 attempt: attempt.id,
2089 });
2090 let outcome = if idle_exit {
2096 AttemptOutcome::ExitedIdleWithoutWork
2097 } else {
2098 AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout)
2099 };
2100 self.deregister_runner(policy, &attempt).await;
2101 self.conclude(&mut attempt, outcome)?;
2102 self.clean_or_quarantine(&mut attempt)?;
2103 if idle_exit {
2107 Ok(ReconcileProgress::Reconciled)
2108 } else {
2109 Ok(ReconcileProgress::Replacement {
2110 attempt: attempt.id,
2111 operation: "registration_timeout_replacement",
2112 })
2113 }
2114 }
2115 }
2116 }
2117
2118 fn record(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2119 self.ports
2120 .store
2121 .record_attempt(attempt)
2122 .map_err(|_| LifecycleError::Journal)?;
2123 self.ports.events.emit(AttemptEvent::State {
2124 attempt: attempt.id,
2125 state: attempt.state(),
2126 });
2127 Ok(())
2128 }
2129
2130 async fn deregister_runner(&self, policy: &ScalePolicy, attempt: &RunnerAttempt) {
2151 let Some(runner_id) = attempt
2152 .github_runner_id()
2153 .or_else(|| read_runner_id(attempt.runtime_path()))
2154 else {
2155 return;
2156 };
2157 if self
2158 .ports
2159 .github
2160 .deregister(&policy.target, runner_id, &self.cancel)
2161 .await
2162 {
2163 self.ports.events.emit(AttemptEvent::Deregistered {
2164 attempt: attempt.id,
2165 runner_id,
2166 });
2167 } else {
2168 tracing::warn!(
2169 attempt = %attempt.id,
2170 runner_id,
2171 "the runner registration could not be removed from GitHub; it will show in the \
2172 target's runner settings until GitHub retires it or a later pass removes it"
2173 );
2174 }
2175 }
2176
2177 fn conclude(
2178 &self,
2179 attempt: &mut RunnerAttempt,
2180 outcome: AttemptOutcome,
2181 ) -> Result<(), LifecycleError> {
2182 attempt
2183 .conclude(outcome.clone(), self.ports.clock.now())
2184 .map_err(|_| LifecycleError::Transition)?;
2185 self.record(attempt)?;
2186 self.ports.events.emit(AttemptEvent::Concluded {
2187 attempt: attempt.id,
2188 outcome: OutcomeKind::of(&outcome),
2189 });
2190 Ok(())
2191 }
2192
2193 fn clean_or_quarantine(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
2208 match self.clean_attempt(attempt) {
2209 Err(LifecycleError::SlotQuarantined { class, .. }) => {
2210 self.ports
2211 .reconcile_events
2212 .emit(LifecycleEvent::AttemptCleanFailed {
2213 policy: attempt.policy_id,
2214 attempt: attempt.id,
2215 reason: class,
2216 });
2217 Ok(())
2218 }
2219 other => other,
2220 }
2221 }
2222
2223 fn clean_attempt(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
2224 let outcome = attempt
2225 .outcome()
2226 .cloned()
2227 .ok_or(LifecycleError::Transition)?;
2228 self.preserve_diagnostics(attempt, &outcome)?;
2229 self.scrub_workspace(attempt)?;
2230 self.ports
2231 .packages
2232 .release(attempt.id)
2233 .map_err(LifecycleError::Failed)?;
2234 attempt
2235 .clean(self.ports.clock.now())
2236 .map_err(|_| LifecycleError::Transition)?;
2237 self.record(attempt)?;
2238 let kind = OutcomeKind::of(&outcome);
2239 self.ports.events.emit(AttemptEvent::Cleaned {
2240 attempt: attempt.id,
2241 outcome: kind,
2242 });
2243 self.ports
2244 .reconcile_events
2245 .emit(LifecycleEvent::AttemptCleaned {
2246 policy: attempt.policy_id,
2247 attempt: attempt.id,
2248 outcome: kind,
2249 });
2250 Ok(())
2251 }
2252
2253 fn scrub_workspace(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2263 #[cfg(test)]
2264 {
2265 if matches!(
2269 std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref(),
2270 Ok("skip_workspace_cleanup" | "reuse_job_workspace")
2271 ) {
2272 return Ok(());
2273 }
2274 }
2275 match attempt.workspace() {
2276 AttemptWorkspace::Ephemeral => match remove_runtime_tree(attempt.runtime_path()) {
2277 Ok(()) => Ok(()),
2278 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
2279 Err(_) => Err(LifecycleError::Failed(FailureReason::Other(
2280 "attempt workspace could not be removed".into(),
2281 ))),
2282 },
2283 AttemptWorkspace::PersistentSlot { slot } => self.scrub_persistent_slot(attempt, slot),
2284 }
2285 }
2286
2287 fn scrub_persistent_slot(
2300 &self,
2301 attempt: &RunnerAttempt,
2302 slot: NonZeroU16,
2303 ) -> Result<(), LifecycleError> {
2304 let configured = self
2309 .ports
2310 .store
2311 .policy(attempt.policy_id)
2312 .map_err(|_| LifecycleError::Journal)?
2313 .and_then(|policy| match policy.workspace_policy() {
2314 WorkspacePolicy::Persistent { root } => Some(root.clone()),
2315 WorkspacePolicy::Ephemeral => None,
2316 });
2317 let runtime = attempt.runtime_path();
2318 self.quarantine_on_refusal(
2319 attempt,
2320 verify_journalled_slot(runtime, slot, configured.as_ref())
2321 .and_then(|()| slot_is_present(runtime))
2322 .and_then(|present| {
2323 if present {
2324 scrub_slot_entries(runtime).and_then(|()| verify_slot_scrubbed(runtime))
2325 } else {
2326 Ok(())
2327 }
2328 }),
2329 )
2330 }
2331
2332 fn quarantine_on_refusal(
2341 &self,
2342 attempt: &RunnerAttempt,
2343 outcome: Result<(), SlotQuarantine>,
2344 ) -> Result<(), LifecycleError> {
2345 let Err(quarantine) = outcome else {
2346 return Ok(());
2347 };
2348 let detail = quarantine.to_string();
2349 tracing::warn!(
2350 attempt = %attempt.id,
2351 policy = %attempt.policy_id,
2352 slot = attempt.workspace().slot_number(),
2353 refusal = quarantine.refusal.class(),
2354 "{detail}"
2355 );
2356 Err(LifecycleError::SlotQuarantined {
2357 class: quarantine.refusal.class(),
2358 detail,
2359 })
2360 }
2361
2362 fn preserve_diagnostics(
2363 &self,
2364 attempt: &RunnerAttempt,
2365 outcome: &AttemptOutcome,
2366 ) -> Result<(), LifecycleError> {
2367 fs::create_dir_all(&self.diagnostics_root).map_err(|_| {
2368 LifecycleError::Failed(FailureReason::Other(
2369 "diagnostics directory could not be created".into(),
2370 ))
2371 })?;
2372 let diagnostic = format!(
2375 "attempt_id={}\npolicy_id={}\noutcome={}\n",
2376 attempt.id,
2377 attempt.policy_id,
2378 OutcomeKind::of(outcome).as_str()
2379 );
2380 fs::write(
2381 self.diagnostics_root.join(format!("{}.log", attempt.id)),
2382 diagnostic,
2383 )
2384 .map_err(|_| {
2385 LifecycleError::Failed(FailureReason::Other(
2386 "redacted diagnostics could not be preserved".into(),
2387 ))
2388 })
2389 }
2390
2391 async fn materialize_with_retry(
2392 &self,
2393 policy: &ScalePolicy,
2394 attempt: &RunnerAttempt,
2395 ) -> Result<RunnerVersion, FailureReason> {
2396 let mut issued = 0_u32;
2397 loop {
2398 issued = issued.saturating_add(1);
2399 match self.ports.packages.materialize(attempt).await {
2400 Ok(version) => return Ok(version),
2401 Err(reason)
2402 if package_failure_is_terminal(&reason)
2403 || issued >= self.retry.max_attempts.max(1) =>
2404 {
2405 return Err(reason);
2406 }
2407 Err(reason) => {
2408 if !self.ports.demand.persists(policy.id).await {
2409 return Err(reason);
2410 }
2411 let delay = self.retry.delay(issued);
2412 self.ports.events.emit(AttemptEvent::Retry {
2413 attempt: attempt.id,
2414 operation: "package_materialization",
2415 delay,
2416 });
2417 self.ports.delay.wait(delay).await;
2418 if !self.ports.demand.persists(policy.id).await {
2419 return Err(reason);
2420 }
2421 }
2422 }
2423 }
2424 }
2425
2426 async fn register_with_retry(
2427 &self,
2428 policy: &ScalePolicy,
2429 attempt: AttemptId,
2430 request: &JitRunnerRequest,
2431 ) -> Result<JitRegistration, LifecycleError> {
2432 let mut issued = 0_u32;
2433 loop {
2434 issued = issued.saturating_add(1);
2435 match self
2436 .ports
2437 .github
2438 .register(&policy.target, request, &self.cancel)
2439 .await
2440 {
2441 Ok(registration) => return Ok(registration),
2442 Err(error) if error.terminal => {
2443 return Err(LifecycleError::Failed(error.reason));
2444 }
2445 Err(error) => {
2446 if issued >= self.retry.max_attempts.max(1)
2447 || !self.ports.demand.persists(policy.id).await
2448 {
2449 return Err(LifecycleError::Failed(error.reason));
2450 }
2451 let delay = error
2452 .retry_after
2453 .unwrap_or_else(|| self.retry.delay(issued));
2454 self.ports.events.emit(AttemptEvent::Retry {
2455 attempt,
2456 operation: "jit_request",
2457 delay,
2458 });
2459 self.ports.delay.wait(delay).await;
2460 if !self.ports.demand.persists(policy.id).await {
2461 return Err(LifecycleError::Failed(error.reason));
2462 }
2463 }
2464 }
2465 }
2466 }
2467
2468 fn allocate_workspace(
2478 &self,
2479 policy: &ScalePolicy,
2480 id: AttemptId,
2481 ) -> Result<Placement, LifecycleError> {
2482 let placement = match policy.workspace_policy() {
2483 WorkspacePolicy::Persistent { root } => self.allocate_persistent_slot(policy, root),
2488 WorkspacePolicy::Ephemeral => self.allocate_disposable(policy, id),
2489 };
2490 if placement.is_ok() {
2493 self.root_accepted(policy.id);
2494 }
2495 placement
2496 }
2497
2498 fn configured_host_root(&self) -> Result<Option<LocalAbsolutePath>, LifecycleError> {
2505 let host = self
2506 .ports
2507 .store
2508 .host(self.host_id)
2509 .map_err(|_| LifecycleError::Journal)?
2510 .ok_or_else(|| LifecycleError::Failed(FailureReason::Other("host not found".into())))?;
2511 Ok(host.runner_root_override.clone())
2512 }
2513
2514 fn effective_host_root(
2519 &self,
2520 policy: &ScalePolicy,
2521 ) -> Result<LocalAbsolutePath, LifecycleError> {
2522 match self.configured_host_root()? {
2523 Some(configured) => Ok(configured),
2524 None => default_runner_root(&self.app_paths).map_err(|error| {
2525 self.root_refused(policy.id, "the platform default runner root", error)
2528 }),
2529 }
2530 }
2531
2532 fn root_refused(&self, policy: PolicyId, root: &str, error: RunnerRootError) -> LifecycleError {
2545 let _ = runner_manager_platform::service::record_runner_root_refusal(
2546 &self.app_paths,
2547 &policy.to_string(),
2548 self.ports.clock.now(),
2549 error.kind(),
2550 root,
2551 &error.to_string(),
2552 );
2553 root_failure(error)
2554 }
2555
2556 fn root_accepted(&self, policy: PolicyId) {
2563 let _ = runner_manager_platform::service::clear_runner_root_refusal(
2564 &self.app_paths,
2565 &policy.to_string(),
2566 );
2567 }
2568
2569 fn allocate_disposable(
2572 &self,
2573 policy: &ScalePolicy,
2574 id: AttemptId,
2575 ) -> Result<Placement, LifecycleError> {
2576 let effective_root = self.effective_host_root(policy)?;
2577 RootPreflight::new(&self.app_paths)
2578 .check(&RootOwner::Host, &effective_root)
2579 .map_err(|error| self.root_refused(policy.id, effective_root.as_str(), error))?;
2580 let runtime = effective_root.as_path().join({
2581 #[cfg(test)]
2582 {
2583 if std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref()
2584 == Ok("reuse_job_workspace")
2585 {
2586 "mutant-shared-workspace".to_owned()
2587 } else {
2588 workspace_name(id)
2589 }
2590 }
2591 #[cfg(not(test))]
2592 {
2593 workspace_name(id)
2594 }
2595 });
2596 fs::create_dir_all(&runtime)
2597 .map_err(|_| LifecycleError::Failed(FailureReason::ProcessStartFailed))?;
2598 Ok(Placement {
2599 runtime,
2600 workspace: AttemptWorkspace::Ephemeral,
2601 })
2602 }
2603
2604 fn allocate_persistent_slot(
2616 &self,
2617 policy: &ScalePolicy,
2618 root: &LocalAbsolutePath,
2619 ) -> Result<Placement, LifecycleError> {
2620 let ceiling = policy.max_capacity().ok_or_else(|| {
2623 LifecycleError::Failed(FailureReason::Other(
2624 "a persistent workspace needs the policy's max_capacity to bound its slots"
2625 .to_string(),
2626 ))
2627 })?;
2628 let leases = self
2629 .ports
2630 .store
2631 .slot_leases_for_policy(policy.id)
2632 .map_err(|_| LifecycleError::Journal)?;
2633 let slot = lowest_free_slot(&leases, ceiling).ok_or_else(|| {
2634 LifecycleError::Failed(FailureReason::Other(format!(
2635 "every persistent slot s1 to s{ceiling} for {} is leased by an attempt that has \
2636 not been cleaned, so no slot is free; raise the repository's max capacity, or \
2637 finish cleaning a concluded attempt",
2638 policy.target
2639 )))
2640 })?;
2641 let workspace = AttemptWorkspace::persistent_slot(slot);
2642 let name = workspace
2643 .slot_directory_name()
2644 .expect("a persistent allocation names its slot directory");
2645
2646 let host_root = self
2658 .configured_host_root()?
2659 .or_else(|| default_runner_root(&self.app_paths).ok());
2660 let mut preflight = RootPreflight::new(&self.app_paths);
2661 if let Some(host_root) = host_root {
2662 preflight = preflight.against(RootOwner::Host, host_root);
2663 }
2664 let checked = preflight
2665 .check(&RootOwner::Repository(policy.target.to_string()), root)
2666 .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2667 if let Some(leaf) = checked.leaf_to_create() {
2668 fs::create_dir(leaf).map_err(|source| {
2669 LifecycleError::Failed(FailureReason::Other(format!(
2670 "the persistent workspace root {} could not be created: {source}",
2671 leaf.display()
2672 )))
2673 })?;
2674 }
2675
2676 let slot_path = runner_root::derive_child(root, &name)
2680 .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2681 create_or_validate_slot(slot_path.as_path())?;
2682 runner_root::verify_containment(root, &slot_path)
2683 .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2684 accept_reusable_slot(slot_path.as_path())?;
2685 Ok(Placement {
2686 runtime: slot_path.as_path().to_path_buf(),
2687 workspace,
2688 })
2689 }
2690
2691 fn record_allocation(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2703 match self.ports.store.record_attempt(attempt) {
2704 Ok(()) => {
2705 self.ports.events.emit(AttemptEvent::State {
2706 attempt: attempt.id,
2707 state: attempt.state(),
2708 });
2709 Ok(())
2710 }
2711 Err(error @ StoreError::SlotAlreadyLeased { .. }) => Err(LifecycleError::Failed(
2712 FailureReason::Other(error.to_string()),
2713 )),
2714 Err(_) => Err(LifecycleError::Journal),
2715 }
2716 }
2717
2718 async fn launch_attempt(
2719 &self,
2720 policy: &ScalePolicy,
2721 allocation_guard: &AllocationGuard,
2722 ) -> Result<RunnerAttempt, LifecycleError> {
2723 if !*self
2724 .recovery_complete
2725 .lock()
2726 .map_err(|_| LifecycleError::Journal)?
2727 {
2728 return Err(LifecycleError::RecoveryIncomplete);
2729 }
2730 let labels = policy
2731 .routing_labels()
2732 .ok_or(LifecycleError::Failed(FailureReason::JitRequestFailed))?;
2733 let id = AttemptId::new_random();
2734 let placement = self.allocate_workspace(policy, id)?;
2735 let mut attempt = RunnerAttempt::allocate_in(
2736 id,
2737 policy.id,
2738 placement.runtime,
2739 placement.workspace,
2740 self.ports.clock.now(),
2741 );
2742 self.record_allocation(&attempt)?;
2747
2748 let version = match self.materialize_with_retry(policy, &attempt).await {
2749 Ok(version) => version,
2750 Err(reason) => return self.fail_launch(&mut attempt, reason),
2751 };
2752 self.prune_under_allocation_lock(allocation_guard, &version)?;
2753 self.versions
2754 .lock()
2755 .map_err(|_| LifecycleError::Journal)?
2756 .insert(id, version);
2757
2758 let jit_request =
2759 JitRunnerRequest::for_policy(runner_name(id), self.runner_group_id, labels);
2760 let registration = match self.register_with_retry(policy, id, &jit_request).await {
2761 Ok(registration) => registration,
2762 Err(error) => return self.fail_launch(&mut attempt, error.reason()),
2763 };
2764 let runner_id = registration.runner().id;
2765 write_runner_id(attempt.runtime_path(), runner_id)?;
2766 attempt
2767 .jit_received(self.ports.clock.now())
2768 .map_err(|_| LifecycleError::Transition)?;
2769 self.record(&attempt)?;
2770 let config = registration.into_config();
2771 let mut issued = 0_u32;
2772 let pid = loop {
2773 issued = issued.saturating_add(1);
2774 match self.ports.processes.spawn(&attempt, &config) {
2775 Ok(pid) => break pid,
2776 Err(error) => {
2777 if let Some(pid) = error.live_pid {
2778 attempt
2779 .started(pid, self.ports.clock.now())
2780 .map_err(|_| LifecycleError::Transition)?;
2781 self.record(&attempt)?;
2782 return Err(LifecycleError::Failed(error.reason));
2783 }
2784 if !error.retryable
2785 || issued >= self.retry.max_attempts.max(1)
2786 || !self.ports.demand.persists(policy.id).await
2787 {
2788 return self.fail_launch(&mut attempt, error.reason);
2789 }
2790 let delay = self.retry.delay(issued);
2791 self.ports.events.emit(AttemptEvent::Retry {
2792 attempt: attempt.id,
2793 operation: "process_start",
2794 delay,
2795 });
2796 self.ports.delay.wait(delay).await;
2797 if !self.ports.demand.persists(policy.id).await {
2798 return self.fail_launch(&mut attempt, error.reason);
2799 }
2800 }
2801 }
2802 };
2803 attempt
2804 .started(pid, self.ports.clock.now())
2805 .map_err(|_| LifecycleError::Transition)?;
2806 self.record(&attempt)?;
2807 Ok(attempt)
2808 }
2809
2810 fn fail_launch<T>(
2811 &self,
2812 attempt: &mut RunnerAttempt,
2813 reason: FailureReason,
2814 ) -> Result<T, LifecycleError> {
2815 self.conclude(attempt, AttemptOutcome::failed(reason.clone()))?;
2816 Err(LifecycleError::Failed(reason))
2817 }
2818
2819 fn prune_under_allocation_lock(
2822 &self,
2823 guard: &AllocationGuard,
2824 version: &RunnerVersion,
2825 ) -> Result<(), LifecycleError> {
2826 let attempts = self
2827 .ports
2828 .store
2829 .attempts()
2830 .map_err(|_| LifecycleError::Journal)?;
2831 self.ports
2832 .packages
2833 .prune_obsolete_guarded(
2834 PruneAuthority::from_launch_request(guard),
2835 version,
2836 &attempts,
2837 )
2838 .map_err(LifecycleError::Failed)
2839 }
2840}
2841
2842#[async_trait]
2843impl RunnerLauncher for LifecycleLauncher {
2844 async fn supervise(
2845 &self,
2846 policy: &ScalePolicy,
2847 ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
2848 LifecycleLauncher::supervise(self, policy)
2849 .await
2850 .map_err(|error| LaunchFailure::new(error.reason()))
2851 }
2852
2853 async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
2854 self.ports.store.attempts().map_err(|_| {
2855 LaunchFailure::new(FailureReason::Other(
2856 "attempt journal could not be read".into(),
2857 ))
2858 })
2859 }
2860
2861 async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
2862 self.launch_attempt(request.policy, request.allocation_guard)
2863 .await
2864 .map_err(|error| LaunchFailure::new(error.reason()))
2865 }
2866
2867 async fn clean(&self, id: AttemptId) -> Result<(), LaunchFailure> {
2868 let mut attempt = self
2869 .ports
2870 .store
2871 .attempt(id)
2872 .map_err(|_| {
2873 LaunchFailure::new(FailureReason::Other(
2874 "attempt journal could not be read".into(),
2875 ))
2876 })?
2877 .ok_or_else(|| {
2878 LaunchFailure::new(FailureReason::Other(
2879 "attempt disappeared from the journal".into(),
2880 ))
2881 })?;
2882 self.clean_attempt(&mut attempt)
2883 .map_err(|error| LaunchFailure::new(error.reason()))
2884 }
2885}
2886
2887fn runner_name(attempt: AttemptId) -> String {
2888 format!("runner-manager-{attempt}")
2889}
2890
2891fn read_runner_id(runtime: &Path) -> Option<u64> {
2892 fs::read_to_string(runtime.join(RUNNER_ID_FILE))
2893 .ok()?
2894 .trim()
2895 .parse()
2896 .ok()
2897}
2898
2899fn write_runner_id(runtime: &Path, runner_id: u64) -> Result<(), LifecycleError> {
2900 let target = runtime.join(RUNNER_ID_FILE);
2901 if let Some(existing) = read_runner_id(runtime) {
2902 return (existing == runner_id)
2903 .then_some(())
2904 .ok_or(LifecycleError::Journal);
2905 }
2906 let temporary = runtime.join(format!("{RUNNER_ID_FILE}.{}.tmp", uuid::Uuid::new_v4()));
2907 write_durable_file(&temporary, runner_id.to_string().as_bytes())
2908 .map_err(|_| LifecycleError::Journal)?;
2909 match fs::rename(&temporary, &target) {
2910 Ok(()) => sync_directory(runtime).map_err(|_| LifecycleError::Journal),
2911 Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
2912 let _ = fs::remove_file(&temporary);
2913 (read_runner_id(runtime) == Some(runner_id))
2914 .then_some(())
2915 .ok_or(LifecycleError::Journal)
2916 }
2917 Err(_) => {
2918 let _ = fs::remove_file(&temporary);
2919 Err(LifecycleError::Journal)
2920 }
2921 }
2922}
2923
2924fn write_durable_file(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
2925 let mut file = fs::OpenOptions::new()
2926 .create(true)
2927 .truncate(true)
2928 .write(true)
2929 .open(path)?;
2930 file.write_all(bytes)?;
2931 file.sync_all()?;
2932 let parent = path.parent().ok_or_else(|| {
2933 std::io::Error::new(
2934 std::io::ErrorKind::InvalidInput,
2935 "file has no parent directory",
2936 )
2937 })?;
2938 sync_directory(parent)
2939}
2940
2941#[cfg(unix)]
2942fn sync_directory(path: &Path) -> std::io::Result<()> {
2943 fs::File::open(path)?.sync_all()
2944}
2945
2946#[cfg(windows)]
2947fn sync_directory(path: &Path) -> std::io::Result<()> {
2948 use std::os::windows::fs::OpenOptionsExt;
2949
2950 const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
2951 const FILE_SHARE_ALL: u32 = 0x0000_0007;
2952 const GENERIC_WRITE: u32 = 0x4000_0000;
2953 fs::OpenOptions::new()
2954 .access_mode(GENERIC_WRITE)
2955 .share_mode(FILE_SHARE_ALL)
2956 .custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
2957 .open(path)?
2958 .sync_all()
2959}
2960
2961#[cfg(test)]
2962mod tests {
2963 use super::*;
2964 use std::collections::BTreeSet;
2965 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2966
2967 use crate::reconcile::{AllocationLock, InProcessAllocationLock};
2968 use runner_manager_domain::model::{Elapsed, TargetScope};
2969 use runner_manager_domain::store::SqliteStore;
2970 use runner_manager_github::jit::JitRunner;
2971 use runner_manager_testkit::clock::FakeClock;
2972 use runner_manager_testkit::fixtures;
2973
2974 type CapturedFields = Vec<(String, String)>;
2976
2977 #[derive(Clone, Default)]
2985 struct CapturedEvents(std::sync::Arc<std::sync::Mutex<Vec<CapturedFields>>>);
2986
2987 impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedEvents {
2988 fn on_event(
2989 &self,
2990 event: &tracing::Event<'_>,
2991 _context: tracing_subscriber::layer::Context<'_, S>,
2992 ) {
2993 struct Collect(Vec<(String, String)>);
2994 impl tracing::field::Visit for Collect {
2995 fn record_debug(
2996 &mut self,
2997 field: &tracing::field::Field,
2998 value: &dyn std::fmt::Debug,
2999 ) {
3000 self.0.push((
3003 field.name().to_owned(),
3004 format!("{value:?}").trim_matches('"').to_owned(),
3005 ));
3006 }
3007 }
3008 let mut collected = Collect(Vec::new());
3009 event.record(&mut collected);
3010 self.0
3011 .lock()
3012 .expect("the capture mutex is not poisoned")
3013 .push(collected.0);
3014 }
3015 }
3016
3017 #[test]
3035 fn a_launch_the_runner_root_refused_names_the_cause_in_the_log_that_ships() {
3036 use runner_manager_platform::logging;
3037 use tracing_subscriber::layer::SubscriberExt as _;
3038
3039 let captured = CapturedEvents::default();
3040 let error = RunnerRootError::DeniedByPrivacyPolicy {
3041 requested: PathBuf::from("/Volumes/NVME/runners"),
3042 refused: PathBuf::from("/Volumes/NVME"),
3043 remediation: RootOwner::Host.remediation(),
3044 };
3045 let kind = error.kind();
3046
3047 let failure = tracing::subscriber::with_default(
3048 tracing_subscriber::registry().with(captured.clone()),
3049 || root_failure(error),
3050 );
3051
3052 assert!(
3057 matches!(
3058 &failure,
3059 LifecycleError::Failed(FailureReason::Other(detail))
3060 if detail.contains("/Volumes/NVME/runners")
3061 && detail.contains("Full Disk Access")
3062 ),
3063 "the reason must still carry the detail: {failure:?}"
3064 );
3065
3066 let events = captured
3067 .0
3068 .lock()
3069 .expect("the capture mutex is not poisoned")
3070 .clone();
3071 let event = events
3072 .iter()
3073 .find(|fields| fields.iter().any(|(_, value)| value.contains(kind)))
3074 .unwrap_or_else(|| panic!("the refusal did not name its cause: {events:?}"));
3075
3076 for (name, value) in event {
3081 assert!(
3082 logging::is_field_allowed(name),
3083 "`{name}` is not allow-listed, so it ships as `{}`: {event:?}",
3084 logging::REDACTION
3085 );
3086 assert_eq!(
3087 &logging::redact(value),
3088 value,
3089 "`{name}` does not survive value-shape scrubbing: {event:?}"
3090 );
3091 }
3092 }
3093
3094 fn nz(slot: u16) -> NonZeroU16 {
3096 NonZeroU16::new(slot).expect("a positive slot")
3097 }
3098
3099 const JIT: &str = "eyJzZWNyZXQiOiJnaHBfRE9fTk9UX0xFQUsifQ==";
3100
3101 #[derive(Debug, Default)]
3102 struct FakeGithubLifecycle {
3103 registration_failures: Mutex<VecDeque<bool>>,
3104 observations: Mutex<VecDeque<LifecycleGithubObservation>>,
3105 registrations: AtomicUsize,
3106 remaining_runners: AtomicUsize,
3107 deregistrations: Mutex<Vec<u64>>,
3111 deregistration_fails: AtomicBool,
3114 journal: Mutex<Option<Arc<SqliteStore>>>,
3119 registration_facts: Mutex<Vec<RegistrationFact>>,
3121 }
3122
3123 #[derive(Debug, Clone)]
3130 struct RegistrationFact {
3131 leased_slots: Vec<u16>,
3133 work_folder: String,
3135 runner_name: String,
3138 }
3139
3140 impl FakeGithubLifecycle {
3141 fn fail(mut self, terminal: bool) -> Self {
3142 self.registration_failures
3143 .get_mut()
3144 .expect("unpoisoned")
3145 .push_back(terminal);
3146 self
3147 }
3148
3149 fn watch_journal(&self, store: Arc<SqliteStore>) {
3150 *self.journal.lock().unwrap() = Some(store);
3151 }
3152
3153 fn registration_facts(&self) -> Vec<RegistrationFact> {
3154 self.registration_facts.lock().unwrap().clone()
3155 }
3156
3157 fn observe(&self, observation: GithubRunnerObservation) {
3158 let observation = match observation {
3159 GithubRunnerObservation::Unreachable => LifecycleGithubObservation::unreachable(),
3160 GithubRunnerObservation::NotRegistered => {
3161 LifecycleGithubObservation::not_registered()
3162 }
3163 GithubRunnerObservation::Registered { busy } => {
3164 LifecycleGithubObservation::registered(73, busy)
3165 }
3166 };
3167 self.observations.lock().unwrap().push_back(observation);
3168 }
3169 }
3170
3171 #[async_trait]
3172 impl LifecycleGithub for FakeGithubLifecycle {
3173 async fn register(
3174 &self,
3175 _target: &ScaleTarget,
3176 request: &JitRunnerRequest,
3177 _cancel: &CancelToken,
3178 ) -> Result<JitRegistration, JitRequestFailure> {
3179 self.registrations.fetch_add(1, Ordering::SeqCst);
3180 if let Some(store) = self.journal.lock().unwrap().as_ref() {
3181 let slots = store
3182 .attempts()
3183 .expect("the journal is readable")
3184 .iter()
3185 .filter_map(|attempt| attempt.workspace().slot_number())
3186 .collect();
3187 self.registration_facts
3188 .lock()
3189 .unwrap()
3190 .push(RegistrationFact {
3191 leased_slots: slots,
3192 work_folder: request.work_folder().to_string(),
3193 runner_name: request.name().to_string(),
3194 });
3195 }
3196 if let Some(terminal) = self.registration_failures.lock().unwrap().pop_front() {
3197 return Err(JitRequestFailure {
3198 terminal,
3199 reason: if terminal {
3200 FailureReason::Other("GitHub refused JIT registration with 403".into())
3201 } else {
3202 FailureReason::JitRequestFailed
3203 },
3204 retry_after: None,
3205 });
3206 }
3207 self.remaining_runners.store(1, Ordering::SeqCst);
3208 Ok(JitRegistration::new(
3209 EncodedJitConfig::new(JIT),
3210 JitRunner {
3211 id: 73,
3212 name: request.name().to_string(),
3213 os: "windows".into(),
3214 status: "offline".into(),
3215 busy: false,
3216 runner_group_id: Some(1),
3217 labels: request.labels().to_vec(),
3218 },
3219 ))
3220 }
3221
3222 async fn observe(
3223 &self,
3224 _target: &ScaleTarget,
3225 _attempt: AttemptId,
3226 _cancel: &CancelToken,
3227 ) -> LifecycleGithubObservation {
3228 let observation = self
3229 .observations
3230 .lock()
3231 .unwrap()
3232 .pop_front()
3233 .unwrap_or(LifecycleGithubObservation::not_registered());
3234 if observation.status == GithubRunnerObservation::NotRegistered {
3235 self.remaining_runners.store(0, Ordering::SeqCst);
3236 }
3237 observation
3238 }
3239
3240 async fn deregister(
3241 &self,
3242 _target: &ScaleTarget,
3243 runner_id: u64,
3244 _cancel: &CancelToken,
3245 ) -> bool {
3246 self.deregistrations.lock().unwrap().push(runner_id);
3247 if self.deregistration_fails.load(Ordering::SeqCst) {
3248 return false;
3249 }
3250 self.remaining_runners.store(0, Ordering::SeqCst);
3251 true
3252 }
3253 }
3254
3255 #[derive(Debug)]
3256 struct FakePackages {
3257 version: RunnerVersion,
3258 leases: Mutex<BTreeSet<AttemptId>>,
3259 materializations: AtomicUsize,
3260 materialization_failures: AtomicUsize,
3261 releases: AtomicUsize,
3262 prunes: AtomicUsize,
3263 prune_currents: Mutex<Vec<RunnerVersion>>,
3264 }
3265
3266 impl Default for FakePackages {
3267 fn default() -> Self {
3268 Self {
3269 version: RunnerVersion::parse("2.330.0").unwrap(),
3270 leases: Mutex::new(BTreeSet::new()),
3271 materializations: AtomicUsize::new(0),
3272 materialization_failures: AtomicUsize::new(0),
3273 releases: AtomicUsize::new(0),
3274 prunes: AtomicUsize::new(0),
3275 prune_currents: Mutex::new(Vec::new()),
3276 }
3277 }
3278 }
3279
3280 impl FakePackages {
3281 fn fail_materializations(&self, count: usize) {
3282 self.materialization_failures.store(count, Ordering::SeqCst);
3283 }
3284 }
3285
3286 #[async_trait]
3287 impl RuntimePackages for FakePackages {
3288 async fn materialize(
3289 &self,
3290 attempt: &RunnerAttempt,
3291 ) -> Result<RunnerVersion, FailureReason> {
3292 self.materializations.fetch_add(1, Ordering::SeqCst);
3293 if self
3294 .materialization_failures
3295 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
3296 if left > 0 { Some(left - 1) } else { None }
3297 })
3298 .is_ok()
3299 {
3300 return Err(FailureReason::Other(
3301 "runner package materialization failed transiently".into(),
3302 ));
3303 }
3304 fs::create_dir_all(attempt.runtime_path()).unwrap();
3305 fs::write(attempt.runtime_path().join("runner-package"), b"verified").unwrap();
3306 self.leases.lock().unwrap().insert(attempt.id);
3307 Ok(self.version.clone())
3308 }
3309
3310 fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
3311 self.leases.lock().unwrap().remove(&attempt);
3312 self.releases.fetch_add(1, Ordering::SeqCst);
3313 Ok(())
3314 }
3315
3316 fn prune_obsolete_guarded(
3317 &self,
3318 _authority: PruneAuthority<'_>,
3319 current: &RunnerVersion,
3320 _attempts: &[RunnerAttempt],
3321 ) -> Result<(), FailureReason> {
3322 self.prunes.fetch_add(1, Ordering::SeqCst);
3323 self.prune_currents.lock().unwrap().push(current.clone());
3324 Ok(())
3325 }
3326 }
3327
3328 #[derive(Debug, Default)]
3329 struct FakeProcesses {
3330 alive: AtomicBool,
3331 completed_successfully: AtomicBool,
3332 spawns: AtomicUsize,
3333 spawn_failures: AtomicUsize,
3334 live_spawn_failure: AtomicBool,
3335 terminations: AtomicUsize,
3336 intent: AtomicBool,
3337 intent_failure: AtomicBool,
3338 actions: Mutex<Vec<&'static str>>,
3339 saw_secret: AtomicBool,
3340 }
3341
3342 impl FakeProcesses {
3343 fn fail_spawns(&self, count: usize) {
3344 self.spawn_failures.store(count, Ordering::SeqCst);
3345 }
3346
3347 fn fail_spawn_with_live_child(&self) {
3348 self.live_spawn_failure.store(true, Ordering::SeqCst);
3349 }
3350
3351 fn set_alive(&self, alive: bool) {
3352 self.alive.store(alive, Ordering::SeqCst);
3353 }
3354
3355 fn finish_successfully(&self) {
3356 self.completed_successfully.store(true, Ordering::SeqCst);
3357 self.alive.store(false, Ordering::SeqCst);
3358 }
3359
3360 fn fail_intent(&self) {
3361 self.intent_failure.store(true, Ordering::SeqCst);
3362 }
3363 }
3364
3365 impl ProcessSupervisor for FakeProcesses {
3366 fn spawn(
3367 &self,
3368 attempt: &RunnerAttempt,
3369 config: &EncodedJitConfig,
3370 ) -> Result<u32, ProcessStartFailure> {
3371 self.spawns.fetch_add(1, Ordering::SeqCst);
3372 let handoff = RestrictiveHandoff::create(
3375 attempt.runtime_path(),
3376 SecretString::from(config.expose().to_owned()),
3377 )
3378 .unwrap();
3379 self.saw_secret
3380 .store(config.expose() == JIT, Ordering::SeqCst);
3381 let handoff_path = handoff.path().to_path_buf();
3382 let failing = self
3383 .spawn_failures
3384 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
3385 if left > 0 { Some(left - 1) } else { None }
3386 })
3387 .is_ok();
3388 drop(handoff);
3389 assert!(!handoff_path.exists(), "handoff must be absent on return");
3390 if self.live_spawn_failure.swap(false, Ordering::SeqCst) {
3391 self.alive.store(true, Ordering::SeqCst);
3392 return Err(ProcessStartFailure::after_spawn_live(4242));
3393 }
3394 if failing {
3395 return Err(ProcessStartFailure::before_spawn(
3396 FailureReason::ProcessStartFailed,
3397 ));
3398 }
3399 self.alive.store(true, Ordering::SeqCst);
3400 Ok(4242)
3401 }
3402
3403 fn is_alive(&self, _attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
3404 self.actions.lock().unwrap().push("observe_process");
3405 Ok(self.alive.load(Ordering::SeqCst))
3406 }
3407
3408 fn recovered_pid(&self, _attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
3409 Ok(self.alive.load(Ordering::SeqCst).then_some(4242))
3410 }
3411
3412 fn completed_successfully(&self, _attempt: &RunnerAttempt) -> bool {
3413 self.completed_successfully.load(Ordering::SeqCst)
3414 }
3415
3416 fn record_terminate_intent(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
3417 self.actions.lock().unwrap().push("terminate_intent");
3418 if self.intent_failure.load(Ordering::SeqCst) {
3419 return Err(FailureReason::Other(
3420 "terminate intent directory sync failed".into(),
3421 ));
3422 }
3423 self.intent.store(true, Ordering::SeqCst);
3424 Ok(())
3425 }
3426
3427 fn has_terminate_intent(&self, _attempt: &RunnerAttempt) -> bool {
3428 self.intent.load(Ordering::SeqCst)
3429 }
3430
3431 fn terminate(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
3432 assert!(
3433 self.intent.load(Ordering::SeqCst),
3434 "the durable intent must exist before signalling"
3435 );
3436 self.actions.lock().unwrap().push("terminate");
3437 self.terminations.fetch_add(1, Ordering::SeqCst);
3438 self.alive.store(false, Ordering::SeqCst);
3439 Ok(())
3440 }
3441 }
3442
3443 #[derive(Debug, Default)]
3444 struct FakeDemand {
3445 answers: Mutex<VecDeque<bool>>,
3446 }
3447
3448 impl FakeDemand {
3449 fn answering(answers: impl IntoIterator<Item = bool>) -> Self {
3450 Self {
3451 answers: Mutex::new(answers.into_iter().collect()),
3452 }
3453 }
3454 }
3455
3456 #[async_trait]
3457 impl DemandPersistence for FakeDemand {
3458 async fn persists(&self, _policy: PolicyId) -> bool {
3459 self.answers.lock().unwrap().pop_front().unwrap_or(true)
3460 }
3461 }
3462
3463 #[derive(Debug, Default)]
3464 struct FakeDelay(Mutex<Vec<Duration>>);
3465
3466 #[async_trait]
3467 impl RetryDelay for FakeDelay {
3468 async fn wait(&self, duration: Duration) {
3469 self.0.lock().unwrap().push(duration);
3470 }
3471 }
3472
3473 struct Harness {
3474 _root: tempfile::TempDir,
3475 app_paths: runner_manager_platform::paths::AppPaths,
3476 launcher: LifecycleLauncher,
3477 demand: Arc<dyn DemandPersistence>,
3478 store: Arc<SqliteStore>,
3479 github: Arc<FakeGithubLifecycle>,
3480 packages: Arc<FakePackages>,
3481 processes: Arc<FakeProcesses>,
3482 clock: Arc<FakeClock>,
3483 events: Arc<AttemptEventLog>,
3484 reconcile_events: Arc<crate::reconcile::EventLog>,
3485 delay: Arc<FakeDelay>,
3486 host: runner_manager_domain::model::Host,
3487 policy: ScalePolicy,
3488 allocation_lock: InProcessAllocationLock,
3489 workspace_root: Option<LocalAbsolutePath>,
3491 }
3492
3493 impl Harness {
3494 fn new(github: FakeGithubLifecycle, demand: Arc<dyn DemandPersistence>) -> Self {
3495 let root = tempfile::tempdir().unwrap();
3496 let paths = runner_manager_platform::paths::AppPaths::rooted_at(root.path());
3497 paths.create_all().unwrap();
3498 let policy = fixtures::policy()
3499 .repository("octo/repo")
3500 .autoscale("home", 2)
3501 .active()
3502 .build();
3503 let host_root = root.path().join("host-root");
3509 fs::create_dir_all(&host_root).unwrap();
3510 let mut host = fixtures::host().build();
3511 host.runner_root_override = Some(
3512 LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
3513 .expect("a local absolute host root"),
3514 );
3515 let store = Arc::new(SqliteStore::open_in_memory().unwrap());
3516 store.put_host(&host).unwrap();
3517 let github = Arc::new(github);
3518 let packages = Arc::new(FakePackages::default());
3519 let processes = Arc::new(FakeProcesses::default());
3520 let clock = Arc::new(FakeClock::default());
3521 let events = Arc::new(AttemptEventLog::default());
3522 let reconcile_events = Arc::new(crate::reconcile::EventLog::new());
3523 let delay = Arc::new(FakeDelay::default());
3524 let ports = LifecyclePorts {
3525 store: Arc::clone(&store) as Arc<dyn Store>,
3526 github: Arc::clone(&github) as Arc<dyn LifecycleGithub>,
3527 packages: Arc::clone(&packages) as Arc<dyn RuntimePackages>,
3528 processes: Arc::clone(&processes) as Arc<dyn ProcessSupervisor>,
3529 clock: Arc::clone(&clock) as Arc<dyn Clock>,
3530 demand: Arc::clone(&demand),
3531 delay: Arc::clone(&delay) as Arc<dyn RetryDelay>,
3532 events: Arc::clone(&events) as Arc<dyn AttemptEventSink>,
3533 reconcile_events: Arc::clone(&reconcile_events) as Arc<dyn EventSink>,
3534 };
3535 let launcher = Self::launcher_over(policy.host_id, &paths, ports);
3536 Self {
3537 _root: root,
3538 app_paths: paths,
3539 launcher,
3540 demand,
3541 store,
3542 github,
3543 packages,
3544 processes,
3545 clock,
3546 events,
3547 reconcile_events,
3548 delay,
3549 host,
3550 policy,
3551 allocation_lock: InProcessAllocationLock::new(),
3552 workspace_root: None,
3553 }
3554 }
3555
3556 fn with_host_runner_root(mut self) -> Self {
3563 let host_root = self.host_root();
3564 fs::create_dir_all(&host_root).unwrap();
3565 self.host.runner_root_override = Some(
3566 LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
3567 .expect("a local absolute host root"),
3568 );
3569 self.store.put_host(&self.host).unwrap();
3570 self
3571 }
3572
3573 fn with_persistent_workspace(mut self, capacity: u16) -> Self {
3575 self = self.with_host_runner_root();
3576 let root = self._root.path().join("persist");
3577 let root = LocalAbsolutePath::new(root.to_str().expect("a UTF-8 temporary path"))
3578 .expect("a local absolute workspace root");
3579 self.policy = fixtures::policy()
3580 .repository("octo/repo")
3581 .autoscale("home", capacity)
3582 .active()
3583 .build();
3584 self.policy
3585 .set_workspace_policy(
3586 WorkspacePolicy::persistent(root.clone(), TargetScope::Repository)
3587 .expect("a repository may be persistent"),
3588 )
3589 .expect("a repository may be persistent");
3590 self.workspace_root = Some(root);
3591 self.store.insert_policy(&self.policy).unwrap();
3594 self
3595 }
3596
3597 fn workspace_root(&self) -> &LocalAbsolutePath {
3598 self.workspace_root
3599 .as_ref()
3600 .expect("this harness configured a persistent workspace")
3601 }
3602
3603 fn slot_path(&self, slot: u16) -> PathBuf {
3604 self.workspace_root().as_path().join(format!("s{slot}"))
3605 }
3606
3607 fn host_root(&self) -> PathBuf {
3608 self._root.path().join("host-root")
3609 }
3610
3611 fn attempt(&self, id: AttemptId) -> RunnerAttempt {
3612 self.store
3613 .attempt(id)
3614 .unwrap()
3615 .expect("the attempt is journalled")
3616 }
3617
3618 fn conclude(&self, id: AttemptId) -> RunnerAttempt {
3620 let mut attempt = self.attempt(id);
3621 attempt
3622 .conclude(
3623 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
3624 self.clock.now(),
3625 )
3626 .unwrap();
3627 self.store.record_attempt(&attempt).unwrap();
3628 attempt
3629 }
3630
3631 async fn cleanup_retaining_work(&self, id: AttemptId) {
3632 self.conclude(id);
3633 self.launcher
3634 .clean(id)
3635 .await
3636 .expect("the slot is scrubbed and the lease released");
3637 }
3638
3639 fn launcher_over(
3642 host: HostId,
3643 paths: &runner_manager_platform::paths::AppPaths,
3644 ports: LifecyclePorts,
3645 ) -> LifecycleLauncher {
3646 LifecycleLauncher::new(
3647 host,
3648 paths.clone(),
3649 paths.logs_dir(),
3650 1,
3651 RecoveryTimeouts::new(
3652 Elapsed::seconds(10),
3653 Elapsed::seconds(10),
3654 Elapsed::seconds(10),
3655 ),
3656 RetryPolicy::bounded(3, Duration::from_millis(10), Duration::from_millis(25)),
3657 ports,
3658 )
3659 }
3660
3661 fn restart(&self) -> LifecycleLauncher {
3664 Self::launcher_over(
3665 self.policy.host_id,
3666 &self.app_paths,
3667 LifecyclePorts {
3668 store: Arc::clone(&self.store) as Arc<dyn Store>,
3669 github: Arc::clone(&self.github) as Arc<dyn LifecycleGithub>,
3670 packages: Arc::clone(&self.packages) as Arc<dyn RuntimePackages>,
3671 processes: Arc::clone(&self.processes) as Arc<dyn ProcessSupervisor>,
3672 clock: Arc::clone(&self.clock) as Arc<dyn Clock>,
3673 demand: Arc::clone(&self.demand),
3674 delay: Arc::clone(&self.delay) as Arc<dyn RetryDelay>,
3675 events: Arc::clone(&self.events) as Arc<dyn AttemptEventSink>,
3676 reconcile_events: Arc::clone(&self.reconcile_events) as Arc<dyn EventSink>,
3677 },
3678 )
3679 }
3680
3681 async fn ready(&self) {
3682 self.launcher
3683 .recover_startup(std::slice::from_ref(&self.policy))
3684 .await
3685 .unwrap();
3686 }
3687
3688 async fn launch(&self) -> RunnerAttempt {
3689 self.launch_result().await.unwrap()
3690 }
3691
3692 async fn launch_result(&self) -> Result<RunnerAttempt, LaunchFailure> {
3693 let guard = self.allocation_lock.acquire().await.unwrap();
3694 self.launcher
3695 .launch(LaunchRequest {
3696 host: &self.host,
3697 policy: &self.policy,
3698 allocation_guard: &guard,
3699 })
3700 .await
3701 }
3702
3703 fn only_attempt(&self) -> RunnerAttempt {
3704 self.store.attempts().unwrap().into_iter().next().unwrap()
3705 }
3706 }
3707
3708 #[tokio::test]
3717 async fn a_root_that_refuses_a_launch_is_recorded_and_cleared_when_one_succeeds() {
3718 use runner_manager_platform::service::{clear_runner_root_refusal, runner_root_refusals};
3719
3720 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
3721 .with_host_runner_root();
3722 harness.ready().await;
3723
3724 let unusable = harness
3728 ._root
3729 .path()
3730 .join("absent")
3731 .join("deeper")
3732 .join("runners");
3733 let mut host = harness.host.clone();
3734 host.runner_root_override = Some(
3735 LocalAbsolutePath::new(unusable.to_str().expect("a UTF-8 temporary path"))
3736 .expect("a local absolute host root"),
3737 );
3738 harness.store.put_host(&host).unwrap();
3739
3740 let failure = harness
3741 .launch_result()
3742 .await
3743 .expect_err("a root whose parents are missing cannot hold a runner");
3744 assert!(
3745 matches!(failure.reason, FailureReason::Other(_)),
3746 "{failure:?}"
3747 );
3748
3749 let refusals = runner_root_refusals(&harness.app_paths).expect("readable");
3750 let refusal = refusals
3751 .first()
3752 .expect("the refusal reached the one surface that can hold it");
3753 assert_eq!(refusal.policy, harness.policy.id.to_string());
3754 assert_eq!(refusal.kind, "missing_parents");
3755 assert!(
3756 refusal.root.contains("runners") && refusal.detail.contains("runners"),
3757 "the directory must be named in full: {refusal:?}"
3758 );
3759
3760 harness.store.put_host(&harness.host).unwrap();
3763 harness.launch().await;
3764 assert!(
3765 runner_root_refusals(&harness.app_paths)
3766 .expect("readable")
3767 .is_empty(),
3768 "a successful placement clears that policy's record"
3769 );
3770
3771 clear_runner_root_refusal(&harness.app_paths, &harness.policy.id.to_string())
3772 .expect("cleanup");
3773 }
3774
3775 #[tokio::test]
3776 async fn a_job_walks_every_state_and_cleans_every_artifact() {
3777 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3778 harness.ready().await;
3779 let started = harness.launch().await;
3780 assert_eq!(started.state(), AttemptState::Starting);
3781 assert_eq!(read_runner_id(started.runtime_path()), Some(73));
3782
3783 harness
3784 .github
3785 .observe(GithubRunnerObservation::Registered { busy: false });
3786 harness.launcher.supervise(&harness.policy).await.unwrap();
3787 assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
3788
3789 harness
3790 .github
3791 .observe(GithubRunnerObservation::Registered { busy: true });
3792 harness.launcher.supervise(&harness.policy).await.unwrap();
3793 assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
3794
3795 harness.processes.finish_successfully();
3796 harness
3797 .github
3798 .observe(GithubRunnerObservation::NotRegistered);
3799 harness.launcher.supervise(&harness.policy).await.unwrap();
3800 let cleaned = harness.only_attempt();
3801 assert_eq!(cleaned.state(), AttemptState::Cleaned);
3802 assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
3803 assert!(!started.runtime_path().exists());
3804 assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
3805 assert_eq!(harness.github.remaining_runners.load(Ordering::SeqCst), 0);
3806
3807 let states: Vec<_> = harness
3808 .events
3809 .events()
3810 .into_iter()
3811 .filter_map(|event| match event {
3812 AttemptEvent::State { state, .. } => Some(state),
3813 _ => None,
3814 })
3815 .collect();
3816 assert_eq!(
3817 states,
3818 vec![
3819 AttemptState::Allocated,
3820 AttemptState::JitReceived,
3821 AttemptState::Starting,
3822 AttemptState::Idle,
3823 AttemptState::Busy,
3824 AttemptState::Finished,
3825 AttemptState::Cleaned,
3826 ]
3827 );
3828 }
3829
3830 #[tokio::test]
3831 async fn an_idle_exit_is_not_a_failure_in_the_journal_or_events() {
3832 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3833 harness.ready().await;
3834 let started = harness.launch().await;
3835 harness
3836 .github
3837 .observe(GithubRunnerObservation::Registered { busy: false });
3838 harness.launcher.supervise(&harness.policy).await.unwrap();
3839 harness.clock.advance_secs(11);
3840 harness.processes.set_alive(false);
3841 harness
3842 .github
3843 .observe(GithubRunnerObservation::NotRegistered);
3844 harness.launcher.supervise(&harness.policy).await.unwrap();
3845
3846 let cleaned = harness.only_attempt();
3847 assert!(cleaned.outcome().unwrap().is_idle_exit());
3848 assert!(!cleaned.outcome().unwrap().is_failure());
3849 assert!(!started.runtime_path().exists());
3850 assert!(
3851 harness
3852 .reconcile_events
3853 .events()
3854 .iter()
3855 .any(|event| matches!(
3856 event,
3857 LifecycleEvent::AttemptCleaned {
3858 outcome: OutcomeKind::IdleExit,
3859 ..
3860 }
3861 ))
3862 );
3863 assert!(!harness.events.events().iter().any(|event| matches!(
3864 event,
3865 AttemptEvent::Concluded {
3866 outcome: OutcomeKind::Failed,
3867 ..
3868 }
3869 )));
3870 }
3871
3872 #[tokio::test]
3873 async fn handoff_is_absent_after_success_and_every_failed_spawn_retry() {
3874 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3875 harness.processes.fail_spawns(2);
3876 harness.ready().await;
3877 let attempt = harness.launch().await;
3878 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 3);
3879 assert!(harness.processes.saw_secret.load(Ordering::SeqCst));
3880 let names: Vec<_> = fs::read_dir(attempt.runtime_path())
3881 .unwrap()
3882 .map(|entry| entry.unwrap().file_name())
3883 .collect();
3884 assert!(
3885 names.iter().all(|name| {
3886 !name
3887 .to_string_lossy()
3888 .starts_with(RestrictiveHandoff::NAME_PREFIX)
3889 }),
3890 "JIT artifact survived: {names:?}"
3891 );
3892 assert_eq!(
3893 *harness.delay.0.lock().unwrap(),
3894 vec![Duration::from_millis(10), Duration::from_millis(20)]
3895 );
3896 }
3897
3898 #[tokio::test]
3899 async fn jit_retry_stops_with_demand_and_a_terminal_403_never_retries() {
3900 let gone = Harness::new(
3901 FakeGithubLifecycle::default().fail(false),
3902 Arc::new(FakeDemand::answering([false])),
3903 );
3904 gone.ready().await;
3905 assert!(gone.launch_result().await.is_err());
3906 assert_eq!(gone.github.registrations.load(Ordering::SeqCst), 1);
3907 assert!(gone.delay.0.lock().unwrap().is_empty());
3908
3909 let forbidden = Harness::new(
3910 FakeGithubLifecycle::default().fail(true),
3911 Arc::new(PersistentDemand),
3912 );
3913 forbidden.ready().await;
3914 assert!(forbidden.launch_result().await.is_err());
3915 assert_eq!(forbidden.github.registrations.load(Ordering::SeqCst), 1);
3916 assert!(forbidden.delay.0.lock().unwrap().is_empty());
3917 assert!(matches!(
3918 forbidden.only_attempt().outcome(),
3919 Some(AttemptOutcome::Failed {
3920 reason: FailureReason::Other(action)
3921 }) if action.contains("403")
3922 ));
3923
3924 let transient = Harness::new(
3925 FakeGithubLifecycle::default().fail(false).fail(false),
3926 Arc::new(PersistentDemand),
3927 );
3928 transient.ready().await;
3929 transient.launch().await;
3930 assert_eq!(transient.github.registrations.load(Ordering::SeqCst), 3);
3931 assert_eq!(
3932 *transient.delay.0.lock().unwrap(),
3933 vec![Duration::from_millis(10), Duration::from_millis(20)]
3934 );
3935 }
3936
3937 #[test]
3945 fn a_workspace_leaves_room_for_the_deepest_path_a_checkout_writes() {
3946 const MAX_PATH: usize = 260;
3947 let root = r"C:\Users\IvanD\AppData\Local\IvanMurzak\runner-manager\data\runtime";
3949 let repo = "GitHub-Runner-Scaler-UI";
3953 let deepest = format!(
3954 r"_work\{repo}\{repo}\.git\objects\pack\pack-{}.keep",
3955 "0".repeat(40)
3956 );
3957
3958 let name = workspace_name(AttemptId::new_random());
3959 assert_eq!(name.len(), WORKSPACE_NAME_LEN, "{name}");
3960 assert!(
3961 name.chars().all(|c| c.is_ascii_hexdigit()),
3962 "a directory name must not carry the identifier's dashes: {name}"
3963 );
3964
3965 let full = format!(r"{root}\{name}\{deepest}");
3966 assert!(
3967 full.len() < MAX_PATH,
3968 "the deepest path a checkout writes must fit: {} characters, limit {MAX_PATH}",
3969 full.len()
3970 );
3971
3972 let old = format!(
3975 r"{root}\{}\{}\{deepest}",
3976 PolicyId::new_random(),
3977 AttemptId::new_random()
3978 );
3979 assert!(
3980 old.len() > MAX_PATH,
3981 "the old layout is supposed to be the thing that did not fit: {} characters",
3982 old.len()
3983 );
3984 }
3985
3986 #[tokio::test]
3987 async fn two_attempts_never_share_a_workspace_even_after_failure() {
3988 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3989 harness.ready().await;
3990 let first = harness.launch().await;
3991 fs::write(first.runtime_path().join("hostile-leftover"), b"first job").unwrap();
3992 harness
3993 .github
3994 .observe(GithubRunnerObservation::Registered { busy: false });
3995 harness.launcher.supervise(&harness.policy).await.unwrap();
3996 harness.clock.advance_secs(11);
3997 harness.processes.set_alive(false);
3998 harness
3999 .github
4000 .observe(GithubRunnerObservation::NotRegistered);
4001 harness.launcher.supervise(&harness.policy).await.unwrap();
4002 assert!(!first.runtime_path().exists());
4003
4004 let second = harness.launch().await;
4005 assert_ne!(first.runtime_path(), second.runtime_path());
4006 assert!(!second.runtime_path().join("hostile-leftover").exists());
4007
4008 fs::write(
4009 second.runtime_path().join("hostile-on-failure"),
4010 b"second job",
4011 )
4012 .unwrap();
4013 harness.processes.set_alive(false);
4014 harness
4015 .github
4016 .observe(GithubRunnerObservation::NotRegistered);
4017 harness.launcher.supervise(&harness.policy).await.unwrap();
4018 assert!(
4019 !second.runtime_path().exists(),
4020 "failed workspace was retained"
4021 );
4022 }
4023
4024 #[tokio::test]
4025 async fn a_runner_that_never_gets_a_job_is_stopped_deregistered_and_not_replaced() {
4026 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4027 harness.ready().await;
4028 let attempt = harness.launch().await;
4029
4030 harness
4034 .github
4035 .observe(GithubRunnerObservation::Registered { busy: false });
4036 harness.launcher.supervise(&harness.policy).await.unwrap();
4037 assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
4038
4039 harness.clock.advance_secs(9);
4042 harness
4043 .github
4044 .observe(GithubRunnerObservation::Registered { busy: false });
4045 let none_yet = harness.launcher.supervise(&harness.policy).await.unwrap();
4046 assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
4047 assert!(none_yet.is_empty());
4048 assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
4049
4050 harness.clock.advance_secs(1);
4052 harness
4053 .github
4054 .observe(GithubRunnerObservation::Registered { busy: false });
4055 let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4056
4057 let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
4058 assert_eq!(
4059 concluded.outcome(),
4060 Some(&AttemptOutcome::ExitedIdleWithoutWork),
4061 "a surplus runner did not fail; recording one as a failure sends an operator \
4062 hunting a fault that does not exist"
4063 );
4064 assert_eq!(concluded.state(), AttemptState::Cleaned);
4065 assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
4066 assert!(!attempt.runtime_path().exists());
4067
4068 assert_eq!(
4071 *harness.github.deregistrations.lock().unwrap(),
4072 vec![73],
4073 "the attempt's own runner id, deleted exactly once"
4074 );
4075
4076 assert!(
4079 replacements.is_empty(),
4080 "a surplus exit must not request a replacement"
4081 );
4082 }
4083
4084 #[tokio::test]
4085 async fn a_registration_github_will_not_delete_still_concludes_the_attempt() {
4086 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4090 harness.ready().await;
4091 let attempt = harness.launch().await;
4092 harness
4093 .github
4094 .observe(GithubRunnerObservation::Registered { busy: false });
4095 harness.launcher.supervise(&harness.policy).await.unwrap();
4096
4097 harness
4098 .github
4099 .deregistration_fails
4100 .store(true, Ordering::SeqCst);
4101 harness.clock.advance_secs(11);
4102 harness
4103 .github
4104 .observe(GithubRunnerObservation::Registered { busy: false });
4105 harness.launcher.supervise(&harness.policy).await.unwrap();
4106
4107 assert_eq!(
4108 *harness.github.deregistrations.lock().unwrap(),
4109 vec![73],
4110 "the delete was attempted"
4111 );
4112 let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
4113 assert_eq!(
4114 concluded.outcome(),
4115 Some(&AttemptOutcome::ExitedIdleWithoutWork),
4116 "the attempt concluded anyway"
4117 );
4118 assert_eq!(concluded.state(), AttemptState::Cleaned);
4119 assert!(!attempt.runtime_path().exists());
4120 }
4121
4122 #[tokio::test]
4123 async fn exit_before_acceptance_returns_replacement_intent_without_launching() {
4124 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4125 harness.ready().await;
4126 let first = harness.launch().await;
4127 harness.processes.set_alive(false);
4128 harness
4129 .github
4130 .observe(GithubRunnerObservation::NotRegistered);
4131 let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4132 let failed = harness.store.attempt(first.id).unwrap().unwrap();
4133 assert!(matches!(
4134 failed.outcome(),
4135 Some(AttemptOutcome::Failed {
4136 reason: FailureReason::ProcessExitedUnexpectedly
4137 })
4138 ));
4139 assert!(!first.runtime_path().exists());
4140
4141 assert_eq!(
4142 replacements,
4143 vec![ReplacementIntent {
4144 policy: harness.policy.id,
4145 previous_attempt: first.id,
4146 operation: "exit_before_acceptance_replacement",
4147 }]
4148 );
4149 assert_eq!(harness.store.attempts().unwrap().len(), 1);
4150 assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
4151 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4152 assert!(harness.delay.0.lock().unwrap().is_empty());
4153 }
4154
4155 #[tokio::test]
4156 async fn expired_jit_is_removed_and_does_not_reregister_after_demand_disappears() {
4157 let harness = Harness::new(
4158 FakeGithubLifecycle::default(),
4159 Arc::new(FakeDemand::answering([false])),
4160 );
4161 let id = AttemptId::new_random();
4162 let runtime = harness
4163 .launcher
4164 .app_paths
4165 .runtime_dir()
4166 .join(harness.policy.id.to_string())
4167 .join(id.to_string());
4168 fs::create_dir_all(&runtime).unwrap();
4169 let mut attempt =
4170 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4171 attempt.jit_received(harness.clock.now()).unwrap();
4172 harness.store.record_attempt(&attempt).unwrap();
4173 harness.clock.advance_secs(11);
4174 let replacements = harness
4175 .launcher
4176 .recover_startup(std::slice::from_ref(&harness.policy))
4177 .await
4178 .unwrap();
4179 assert_eq!(
4180 replacements,
4181 vec![ReplacementIntent {
4182 policy: harness.policy.id,
4183 previous_attempt: id,
4184 operation: "jit_expired_replacement",
4185 }]
4186 );
4187
4188 let cleaned = harness.store.attempt(id).unwrap().unwrap();
4189 assert_eq!(cleaned.state(), AttemptState::Cleaned);
4190 assert!(matches!(
4191 cleaned.outcome(),
4192 Some(AttemptOutcome::Failed {
4193 reason: FailureReason::JitExpired
4194 })
4195 ));
4196 assert!(!runtime.exists());
4197 assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
4198 assert!(harness.delay.0.lock().unwrap().is_empty());
4199 }
4200
4201 #[tokio::test]
4202 async fn expired_jit_returns_intent_but_never_launches_inside_lifecycle() {
4203 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4204 let id = AttemptId::new_random();
4205 let runtime = harness
4206 .launcher
4207 .app_paths
4208 .runtime_dir()
4209 .join("expired-with-demand");
4210 fs::create_dir_all(&runtime).unwrap();
4211 let mut attempt =
4212 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4213 attempt.jit_received(harness.clock.now()).unwrap();
4214 harness.store.record_attempt(&attempt).unwrap();
4215 harness.clock.advance_secs(11);
4216 let replacements = harness
4217 .launcher
4218 .recover_startup(std::slice::from_ref(&harness.policy))
4219 .await
4220 .unwrap();
4221
4222 let attempts = harness.store.attempts().unwrap();
4223 assert_eq!(attempts.len(), 1);
4224 assert_eq!(
4225 attempts
4226 .iter()
4227 .find(|attempt| attempt.id == id)
4228 .unwrap()
4229 .state(),
4230 AttemptState::Cleaned
4231 );
4232 assert_eq!(
4233 replacements,
4234 vec![ReplacementIntent {
4235 policy: harness.policy.id,
4236 previous_attempt: id,
4237 operation: "jit_expired_replacement",
4238 }]
4239 );
4240 assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
4241 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4242 assert!(harness.delay.0.lock().unwrap().is_empty());
4243 }
4244
4245 #[tokio::test]
4246 async fn package_materialization_retries_are_bounded_and_demand_adjacent() {
4247 let persistent = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4248 persistent.packages.fail_materializations(2);
4249 persistent.ready().await;
4250 persistent.launch().await;
4251 assert_eq!(
4252 persistent.packages.materializations.load(Ordering::SeqCst),
4253 3
4254 );
4255 assert_eq!(
4256 *persistent.delay.0.lock().unwrap(),
4257 vec![Duration::from_millis(10), Duration::from_millis(20)]
4258 );
4259
4260 let gone_before_wait = Harness::new(
4261 FakeGithubLifecycle::default(),
4262 Arc::new(FakeDemand::answering([false])),
4263 );
4264 gone_before_wait.packages.fail_materializations(3);
4265 gone_before_wait.ready().await;
4266 assert!(gone_before_wait.launch_result().await.is_err());
4267 assert_eq!(
4268 gone_before_wait
4269 .packages
4270 .materializations
4271 .load(Ordering::SeqCst),
4272 1
4273 );
4274 assert!(gone_before_wait.delay.0.lock().unwrap().is_empty());
4275
4276 let gone_during_wait = Harness::new(
4277 FakeGithubLifecycle::default(),
4278 Arc::new(FakeDemand::answering([true, false])),
4279 );
4280 gone_during_wait.packages.fail_materializations(3);
4281 gone_during_wait.ready().await;
4282 assert!(gone_during_wait.launch_result().await.is_err());
4283 assert_eq!(
4284 gone_during_wait
4285 .packages
4286 .materializations
4287 .load(Ordering::SeqCst),
4288 1
4289 );
4290 assert_eq!(
4291 *gone_during_wait.delay.0.lock().unwrap(),
4292 vec![Duration::from_millis(10)]
4293 );
4294 }
4295
4296 #[tokio::test]
4297 async fn replacement_is_intent_only_and_never_launches_inside_lifecycle() {
4298 let harness = Harness::new(
4299 FakeGithubLifecycle::default(),
4300 Arc::new(FakeDemand::answering([true, false])),
4301 );
4302 harness.ready().await;
4303 let first = harness.launch().await;
4304 harness.processes.set_alive(false);
4305 harness
4306 .github
4307 .observe(GithubRunnerObservation::NotRegistered);
4308 let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4309
4310 assert_eq!(harness.store.attempts().unwrap().len(), 1);
4311 assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
4312 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4313 assert!(harness.delay.0.lock().unwrap().is_empty());
4314 assert_eq!(
4315 replacements,
4316 vec![ReplacementIntent {
4317 policy: harness.policy.id,
4318 previous_attempt: first.id,
4319 operation: "exit_before_acceptance_replacement",
4320 }]
4321 );
4322 assert_eq!(
4323 harness.store.attempt(first.id).unwrap().unwrap().state(),
4324 AttemptState::Cleaned
4325 );
4326 }
4327
4328 #[tokio::test]
4329 async fn startup_adopts_a_live_process_and_refuses_launch_before_recovery() {
4330 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4331 let before = harness.launch_result().await;
4332 assert!(before.is_err());
4333 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4334
4335 let id = AttemptId::new_random();
4336 let runtime = harness.launcher.app_paths.runtime_dir().join("adopt");
4337 fs::create_dir_all(&runtime).unwrap();
4338 let mut attempt =
4339 RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4340 attempt.jit_received(harness.clock.now()).unwrap();
4341 attempt.started(4242, harness.clock.now()).unwrap();
4342 harness.store.record_attempt(&attempt).unwrap();
4343 harness.processes.set_alive(true);
4344 harness
4345 .github
4346 .observe(GithubRunnerObservation::NotRegistered);
4347 let replacements = harness
4348 .launcher
4349 .recover_startup(std::slice::from_ref(&harness.policy))
4350 .await
4351 .unwrap();
4352 assert!(replacements.is_empty());
4353 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4354 assert!(
4355 harness
4356 .events
4357 .events()
4358 .contains(&AttemptEvent::Adopted { attempt: id })
4359 );
4360 }
4361
4362 #[tokio::test]
4363 async fn spawn_before_starting_crash_recovers_pid_then_completes_and_cleans() {
4364 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4365 let id = AttemptId::new_random();
4366 let runtime = harness
4367 .launcher
4368 .app_paths
4369 .runtime_dir()
4370 .join("spawn-before-starting");
4371 fs::create_dir_all(&runtime).unwrap();
4372 let mut attempt =
4373 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4374 attempt.jit_received(harness.clock.now()).unwrap();
4375 harness.store.record_attempt(&attempt).unwrap();
4376 harness.processes.set_alive(true);
4377 harness
4378 .github
4379 .observe(GithubRunnerObservation::Registered { busy: true });
4380
4381 let replacements = harness
4382 .launcher
4383 .recover_startup(std::slice::from_ref(&harness.policy))
4384 .await
4385 .unwrap();
4386 assert!(replacements.is_empty());
4387 let recovered = harness.store.attempt(id).unwrap().unwrap();
4388 assert_eq!(recovered.state(), AttemptState::Busy);
4389 assert_eq!(recovered.process_id(), Some(4242));
4390 assert_eq!(recovered.github_runner_id(), Some(73));
4391 let events = harness.events.events();
4392 let starting = events
4393 .iter()
4394 .position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Starting } if *attempt == id))
4395 .unwrap();
4396 let busy = events
4397 .iter()
4398 .position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Busy } if *attempt == id))
4399 .unwrap();
4400 assert!(starting < busy, "recovery skipped a legal edge: {events:?}");
4401
4402 harness.processes.finish_successfully();
4403 harness
4404 .github
4405 .observe(GithubRunnerObservation::NotRegistered);
4406 assert!(
4407 harness
4408 .launcher
4409 .supervise(&harness.policy)
4410 .await
4411 .unwrap()
4412 .is_empty()
4413 );
4414 let cleaned = harness.store.attempt(id).unwrap().unwrap();
4415 assert_eq!(cleaned.state(), AttemptState::Cleaned);
4416 assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
4417 assert!(!runtime.exists());
4418 }
4419
4420 #[tokio::test]
4421 async fn failed_post_spawn_stop_keeps_capacity_until_supervision_proves_death() {
4422 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4423 harness.processes.fail_spawn_with_live_child();
4424 harness.ready().await;
4425 assert!(harness.launch_result().await.is_err());
4426
4427 let attempt = harness.only_attempt();
4428 assert_eq!(attempt.state(), AttemptState::Starting);
4429 assert_eq!(attempt.process_id(), Some(4242));
4430 assert!(attempt.outcome().is_none());
4431 assert!(attempt.state().counts_against_capacity());
4432 assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4433 assert!(harness.delay.0.lock().unwrap().is_empty());
4434
4435 harness.processes.set_alive(false);
4436 harness
4437 .github
4438 .observe(GithubRunnerObservation::NotRegistered);
4439 let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4440 assert_eq!(replacements.len(), 1);
4441 assert_eq!(
4442 harness.store.attempt(attempt.id).unwrap().unwrap().state(),
4443 AttemptState::Cleaned
4444 );
4445 }
4446
4447 #[tokio::test]
4448 async fn remote_runner_identity_closes_both_sides_of_the_registration_crash_boundary() {
4449 for sidecar_already_present in [false, true] {
4450 let harness = Harness::new(
4451 FakeGithubLifecycle::default(),
4452 Arc::new(FakeDemand::answering([false])),
4453 );
4454 let id = AttemptId::new_random();
4455 let runtime =
4456 harness
4457 .launcher
4458 .app_paths
4459 .runtime_dir()
4460 .join(if sidecar_already_present {
4461 "after-id-sidecar"
4462 } else {
4463 "before-id-sidecar"
4464 });
4465 fs::create_dir_all(&runtime).unwrap();
4466 let mut attempt =
4467 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4468 if sidecar_already_present {
4469 write_runner_id(&runtime, 73).unwrap();
4470 attempt.jit_received(harness.clock.now()).unwrap();
4471 }
4472 harness.store.record_attempt(&attempt).unwrap();
4473 harness.processes.set_alive(true);
4474 harness
4475 .github
4476 .observe(GithubRunnerObservation::Registered { busy: false });
4477 harness
4478 .launcher
4479 .recover_startup(std::slice::from_ref(&harness.policy))
4480 .await
4481 .unwrap();
4482
4483 assert_eq!(read_runner_id(&runtime), Some(73));
4484 assert!(
4485 harness
4486 .store
4487 .attempt(id)
4488 .unwrap()
4489 .unwrap()
4490 .outcome()
4491 .is_none()
4492 );
4493 let events = harness.events.events();
4494 let recovered = events.iter().position(|event| {
4495 matches!(
4496 event,
4497 AttemptEvent::RemoteIdentityRecovered {
4498 attempt,
4499 runner_id: 73
4500 } if *attempt == id
4501 )
4502 });
4503 assert_eq!(recovered.is_some(), !sidecar_already_present);
4504 if let Some(recovered) = recovered {
4505 let adopted = events
4506 .iter()
4507 .position(|event| matches!(event, AttemptEvent::Adopted { attempt } if *attempt == id))
4508 .unwrap();
4509 assert!(
4510 recovered < adopted,
4511 "identity was not durable before adoption: {events:?}"
4512 );
4513 }
4514 assert!(runtime.exists());
4515 }
4516 }
4517
4518 #[tokio::test]
4519 async fn recovery_stays_closed_for_unknown_policy_and_unreachable_attempts() {
4520 let unknown = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4521 let unknown_attempt = RunnerAttempt::allocate(
4522 AttemptId::new_random(),
4523 PolicyId::from_u128(0xfeed),
4524 unknown
4525 .launcher
4526 .app_paths
4527 .runtime_dir()
4528 .join("unknown-policy"),
4529 unknown.clock.now(),
4530 );
4531 unknown.store.record_attempt(&unknown_attempt).unwrap();
4532 let expired_id = AttemptId::new_random();
4533 let expired_runtime = unknown
4534 .launcher
4535 .app_paths
4536 .runtime_dir()
4537 .join("expired-beside-unknown");
4538 fs::create_dir_all(&expired_runtime).unwrap();
4539 let mut expired = RunnerAttempt::allocate(
4540 expired_id,
4541 unknown.policy.id,
4542 expired_runtime,
4543 unknown.clock.now(),
4544 );
4545 expired.jit_received(unknown.clock.now()).unwrap();
4546 unknown.store.record_attempt(&expired).unwrap();
4547 unknown.clock.advance_secs(11);
4548 assert!(matches!(
4549 unknown
4550 .launcher
4551 .recover_startup(std::slice::from_ref(&unknown.policy))
4552 .await,
4553 Err(LifecycleError::RecoveryIncomplete)
4554 ));
4555 assert!(unknown.launch_result().await.is_err());
4556 assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
4557 let recovered_policy = fixtures::policy()
4558 .id(PolicyId::from_u128(0xfeed))
4559 .repository("octo/repo")
4560 .autoscale("home", 2)
4561 .active()
4562 .build();
4563 let pending = unknown
4564 .launcher
4565 .recover_startup(&[unknown.policy.clone(), recovered_policy])
4566 .await
4567 .unwrap();
4568 assert_eq!(
4569 pending,
4570 vec![ReplacementIntent {
4571 policy: unknown.policy.id,
4572 previous_attempt: expired_id,
4573 operation: "jit_expired_replacement",
4574 }]
4575 );
4576 assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
4577
4578 let unreachable = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4579 let id = AttemptId::new_random();
4580 let runtime = unreachable
4581 .launcher
4582 .app_paths
4583 .runtime_dir()
4584 .join("unreachable");
4585 fs::create_dir_all(&runtime).unwrap();
4586 unreachable
4587 .store
4588 .record_attempt(&RunnerAttempt::allocate(
4589 id,
4590 unreachable.policy.id,
4591 runtime,
4592 unreachable.clock.now(),
4593 ))
4594 .unwrap();
4595 unreachable
4596 .github
4597 .observe(GithubRunnerObservation::Unreachable);
4598 assert!(matches!(
4599 unreachable
4600 .launcher
4601 .recover_startup(std::slice::from_ref(&unreachable.policy))
4602 .await,
4603 Err(LifecycleError::RecoveryIncomplete)
4604 ));
4605 assert!(unreachable.launch_result().await.is_err());
4606 assert_eq!(unreachable.processes.spawns.load(Ordering::SeqCst), 0);
4607 }
4608
4609 #[tokio::test]
4610 async fn a_dead_busy_process_unknown_to_github_is_orphaned_and_cleaned() {
4611 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4612 let id = AttemptId::new_random();
4613 let runtime = harness.launcher.app_paths.runtime_dir().join("orphan");
4614 fs::create_dir_all(&runtime).unwrap();
4615 let mut attempt =
4616 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4617 attempt.jit_received(harness.clock.now()).unwrap();
4618 attempt.started(4242, harness.clock.now()).unwrap();
4619 attempt.assigned_job(73, harness.clock.now()).unwrap();
4620 harness.store.record_attempt(&attempt).unwrap();
4621 harness.processes.set_alive(false);
4622 harness
4623 .github
4624 .observe(GithubRunnerObservation::NotRegistered);
4625 harness
4626 .launcher
4627 .recover_startup(std::slice::from_ref(&harness.policy))
4628 .await
4629 .unwrap();
4630 let cleaned = harness.store.attempt(id).unwrap().unwrap();
4631 assert_eq!(cleaned.state(), AttemptState::Cleaned);
4632 assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::Orphaned));
4633 assert!(!runtime.exists());
4634 }
4635
4636 #[tokio::test]
4637 async fn registration_timeout_journals_intent_stops_then_concludes_with_dead_reason() {
4638 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4639 harness.ready().await;
4640 let id = AttemptId::new_random();
4641 let runtime = harness.launcher.app_paths.runtime_dir().join("timeout");
4642 fs::create_dir_all(&runtime).unwrap();
4643 let mut attempt =
4644 RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4645 attempt.jit_received(harness.clock.now()).unwrap();
4646 attempt.started(4242, harness.clock.now()).unwrap();
4647 harness.store.record_attempt(&attempt).unwrap();
4648 harness.clock.advance_secs(11);
4649 harness.processes.set_alive(true);
4650 harness
4651 .github
4652 .observe(GithubRunnerObservation::NotRegistered);
4653 let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4654 assert_eq!(
4655 replacements,
4656 vec![ReplacementIntent {
4657 policy: harness.policy.id,
4658 previous_attempt: id,
4659 operation: "registration_timeout_replacement",
4660 }]
4661 );
4662
4663 assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
4664 assert!(!harness.processes.alive.load(Ordering::SeqCst));
4665 let actions = harness.processes.actions.lock().unwrap().clone();
4666 let intent = actions
4667 .iter()
4668 .position(|action| *action == "terminate_intent")
4669 .unwrap();
4670 let signal = actions
4671 .iter()
4672 .position(|action| *action == "terminate")
4673 .unwrap();
4674 assert!(
4675 intent < signal,
4676 "intent was not durable before signal: {actions:?}"
4677 );
4678
4679 let cleaned = harness.store.attempt(id).unwrap().unwrap();
4680 assert!(matches!(
4681 cleaned.outcome(),
4682 Some(AttemptOutcome::Failed {
4683 reason: FailureReason::TerminatedAfterRegistrationTimeout
4684 })
4685 ));
4686 let events = harness.events.events();
4687 let intent = events
4688 .iter()
4689 .position(|event| matches!(event, AttemptEvent::TerminateIntent { .. }))
4690 .unwrap();
4691 let stopped = events
4692 .iter()
4693 .position(|event| matches!(event, AttemptEvent::Terminated { .. }))
4694 .unwrap();
4695 let concluded = events
4696 .iter()
4697 .position(|event| matches!(event, AttemptEvent::Concluded { .. }))
4698 .unwrap();
4699 assert!(intent < stopped && stopped < concluded, "{events:?}");
4700 }
4701
4702 #[tokio::test]
4703 async fn timeout_crash_recovery_returns_the_same_replacement_intent() {
4704 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4705 let id = AttemptId::new_random();
4706 let runtime = harness
4707 .launcher
4708 .app_paths
4709 .runtime_dir()
4710 .join("timeout-after-crash");
4711 fs::create_dir_all(&runtime).unwrap();
4712 let mut attempt =
4713 RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4714 attempt.jit_received(harness.clock.now()).unwrap();
4715 attempt.started(4242, harness.clock.now()).unwrap();
4716 harness.store.record_attempt(&attempt).unwrap();
4717 harness.processes.intent.store(true, Ordering::SeqCst);
4718 harness.processes.set_alive(false);
4719 harness
4720 .github
4721 .observe(GithubRunnerObservation::NotRegistered);
4722
4723 let replacements = harness
4724 .launcher
4725 .recover_startup(std::slice::from_ref(&harness.policy))
4726 .await
4727 .unwrap();
4728 assert_eq!(
4729 replacements,
4730 vec![ReplacementIntent {
4731 policy: harness.policy.id,
4732 previous_attempt: id,
4733 operation: "registration_timeout_replacement",
4734 }]
4735 );
4736 let consumed = RunnerLauncher::supervise(&harness.launcher, &harness.policy)
4737 .await
4738 .unwrap();
4739 assert_eq!(consumed, replacements);
4740 assert!(
4741 RunnerLauncher::supervise(&harness.launcher, &harness.policy)
4742 .await
4743 .unwrap()
4744 .is_empty(),
4745 "startup replacement evidence must be consumed exactly once by e1"
4746 );
4747 assert!(matches!(
4748 harness.store.attempt(id).unwrap().unwrap().outcome(),
4749 Some(AttemptOutcome::Failed {
4750 reason: FailureReason::TerminatedAfterRegistrationTimeout
4751 })
4752 ));
4753 }
4754
4755 #[tokio::test]
4756 async fn terminate_intent_sync_failure_prevents_signal_and_conclusion() {
4757 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4758 let id = AttemptId::new_random();
4759 let runtime = harness
4760 .launcher
4761 .app_paths
4762 .runtime_dir()
4763 .join("timeout-sync-failure");
4764 fs::create_dir_all(&runtime).unwrap();
4765 let mut attempt =
4766 RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4767 attempt.jit_received(harness.clock.now()).unwrap();
4768 attempt.started(4242, harness.clock.now()).unwrap();
4769 harness.store.record_attempt(&attempt).unwrap();
4770 harness.clock.advance_secs(11);
4771 harness.processes.set_alive(true);
4772 harness.processes.fail_intent();
4773 harness
4774 .github
4775 .observe(GithubRunnerObservation::NotRegistered);
4776
4777 assert!(
4778 harness
4779 .launcher
4780 .recover_startup(std::slice::from_ref(&harness.policy))
4781 .await
4782 .is_err()
4783 );
4784 assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
4785 assert!(harness.processes.alive.load(Ordering::SeqCst));
4786 assert_eq!(
4787 harness.store.attempt(id).unwrap().unwrap().state(),
4788 AttemptState::Starting
4789 );
4790 assert!(!harness.events.events().iter().any(|event| matches!(
4791 event,
4792 AttemptEvent::Terminated { attempt } | AttemptEvent::Concluded { attempt, .. }
4793 if *attempt == id
4794 )));
4795 }
4796
4797 #[tokio::test]
4798 async fn diagnostics_survive_cleanup_without_the_jit_or_a_token() {
4799 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4800 harness.ready().await;
4801 let attempt = harness.launch().await;
4802 harness
4803 .github
4804 .observe(GithubRunnerObservation::Registered { busy: false });
4805 harness.launcher.supervise(&harness.policy).await.unwrap();
4806 harness.clock.advance_secs(11);
4807 harness.processes.set_alive(false);
4808 harness
4809 .github
4810 .observe(GithubRunnerObservation::NotRegistered);
4811 harness.launcher.supervise(&harness.policy).await.unwrap();
4812 let diagnostic = fs::read_to_string(
4813 harness
4814 .launcher
4815 .diagnostics_root
4816 .join(format!("{}.log", attempt.id)),
4817 )
4818 .unwrap();
4819 assert!(diagnostic.contains("exited_idle_without_work"));
4820 assert!(!diagnostic.contains(JIT));
4821 assert!(!diagnostic.contains("ghp_"));
4822 assert!(!attempt.runtime_path().exists());
4823 }
4824
4825 #[test]
4826 fn native_process_listing_never_contains_jit_and_handoffs_never_survive() {
4827 let root = tempfile::tempdir().unwrap();
4828 let policy = fixtures::policy()
4829 .repository("octo/repo")
4830 .autoscale("home", 1)
4831 .active()
4832 .build();
4833 let runtime = root.path().join("successful");
4834 fs::create_dir_all(&runtime).unwrap();
4835 let processes = NativeProcesses::new();
4836 let config = EncodedJitConfig::new(JIT);
4837 let handoff =
4838 RestrictiveHandoff::create(&runtime, SecretString::from(config.expose().to_owned()))
4839 .unwrap();
4840 let mut child = native_inspection_spec()
4841 .spawn_runner_with_handoff(&handoff)
4842 .expect("native child starts");
4843 let pid = child.pid();
4844 handoff.delete().unwrap();
4845 let command_line = native_command_line(pid);
4846 assert!(
4847 !command_line.contains(JIT),
4848 "the encoded JIT configuration appeared in the native process listing"
4849 );
4850 assert_no_jit_file(&runtime);
4851 child
4852 .stop(Duration::from_secs(1))
4853 .expect("native child stops");
4854
4855 let failed_runtime = root.path().join("failed");
4856 fs::create_dir_all(&failed_runtime).unwrap();
4857 let failed = RunnerAttempt::allocate(
4858 AttemptId::new_random(),
4859 policy.id,
4860 &failed_runtime,
4861 FakeClock::default().now(),
4862 );
4863 assert!(
4864 processes
4865 .spawn(&failed, &EncodedJitConfig::new(JIT))
4866 .is_err(),
4867 "a runtime with no runner executable must fail"
4868 );
4869 assert_no_jit_file(&failed_runtime);
4870 processes
4871 .record_terminate_intent(&failed)
4872 .expect("the intent file and its directory entry are durably synced");
4873 assert_eq!(
4874 fs::read(NativeProcesses::intent_path(&failed)).unwrap(),
4875 b"registration-timeout\n"
4876 );
4877 }
4878
4879 #[test]
4880 fn post_spawn_boundaries_are_bounded_durable_and_never_retry_jit() {
4881 let root = tempfile::tempdir().unwrap();
4882 let policy = fixtures::policy()
4883 .repository("octo/repo")
4884 .autoscale("home", 1)
4885 .active()
4886 .build();
4887 let processes = NativeProcesses::new();
4888 processes.use_long_lived_test_listener();
4889 for (index, boundary) in [
4890 PostSpawnBoundary::HandoffDelete,
4891 PostSpawnBoundary::IdentitySerialize,
4892 PostSpawnBoundary::IdentityWrite,
4893 PostSpawnBoundary::ChildMapInsert,
4894 ]
4895 .into_iter()
4896 .enumerate()
4897 {
4898 let runtime = root.path().join(format!("post-spawn-{index}"));
4899 let bin = runtime.join("bin");
4900 fs::create_dir_all(&bin).unwrap();
4901 #[cfg(windows)]
4902 let listener = bin.join("Runner.Listener.exe");
4903 #[cfg(not(windows))]
4904 let listener = bin.join("Runner.Listener");
4905 fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4906 let attempt = RunnerAttempt::allocate(
4907 AttemptId::new_random(),
4908 policy.id,
4909 &runtime,
4910 FakeClock::default().now(),
4911 );
4912 processes.fail_post_spawn_at(boundary);
4913 let failure = processes
4914 .spawn(&attempt, &EncodedJitConfig::new(JIT))
4915 .expect_err("fault must cross the post-spawn cleanup path");
4916 assert!(!failure.retryable, "{boundary:?} allowed duplicate retry");
4917 assert!(
4918 !processes.is_alive(&attempt).unwrap(),
4919 "{boundary:?} left a child"
4920 );
4921 assert!(!NativeProcesses::identity_path(&attempt).exists());
4922 assert_no_jit_file(&runtime);
4923 }
4924 assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
4925
4926 let runtime = root.path().join("identity-and-stop-fail");
4927 let bin = runtime.join("bin");
4928 fs::create_dir_all(&bin).unwrap();
4929 #[cfg(windows)]
4930 let listener = bin.join("Runner.Listener.exe");
4931 #[cfg(not(windows))]
4932 let listener = bin.join("Runner.Listener");
4933 fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4934 let attempt = RunnerAttempt::allocate(
4935 AttemptId::new_random(),
4936 policy.id,
4937 &runtime,
4938 FakeClock::default().now(),
4939 );
4940 processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4944 processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4945 processes.fail_next_post_spawn_stop();
4946 let failure = processes
4947 .spawn(&attempt, &EncodedJitConfig::new(JIT))
4948 .expect_err("the identity boundary must fail closed");
4949 assert!(failure.live_pid.is_some());
4950 assert_long_lived_listener_ready(&processes, &attempt);
4951 assert!(processes.is_alive(&attempt).unwrap());
4952 assert!(!NativeProcesses::identity_path(&attempt).exists());
4953 assert!(NativeProcesses::fallback_identity_path(&attempt).is_file());
4954 assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
4955 processes.terminate(&attempt).unwrap();
4956
4957 let runtime = root.path().join("persistent-stop-and-identity-failures");
4958 let bin = runtime.join("bin");
4959 fs::create_dir_all(&bin).unwrap();
4960 #[cfg(windows)]
4961 let listener = bin.join("Runner.Listener.exe");
4962 #[cfg(not(windows))]
4963 let listener = bin.join("Runner.Listener");
4964 fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4965 let mut unresolved = RunnerAttempt::allocate(
4966 AttemptId::new_random(),
4967 policy.id,
4968 &runtime,
4969 FakeClock::default().now(),
4970 );
4971 for _ in 0..3 {
4972 processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4973 }
4974 processes.fail_post_spawn_stops(MAX_POST_SPAWN_STOP_ATTEMPTS);
4975 let failure = processes
4976 .spawn(&unresolved, &EncodedJitConfig::new(JIT))
4977 .expect_err("bounded cleanup must return even when every stop errors");
4978 let pid = failure
4979 .live_pid
4980 .expect("the owned child remains supervised in this invocation");
4981 assert!(matches!(failure.reason, FailureReason::Other(_)));
4982 assert_long_lived_listener_ready(&processes, &unresolved);
4983 unresolved.jit_received(FakeClock::default().now()).unwrap();
4984 unresolved.started(pid, FakeClock::default().now()).unwrap();
4985 let journal = SqliteStore::open_in_memory().unwrap();
4986 journal.record_attempt(&unresolved).unwrap();
4987 let recovered = journal.attempt(unresolved.id).unwrap().unwrap();
4988 assert_eq!(recovered.process_id(), Some(pid));
4989 assert_eq!(recovered.state(), AttemptState::Starting);
4990 assert!(processes.is_alive(&unresolved).unwrap());
4991 assert!(!NativeProcesses::identity_path(&unresolved).exists());
4992 assert!(!NativeProcesses::fallback_identity_path(&unresolved).exists());
4993 assert_eq!(
4994 fs::read_to_string(NativeProcesses::unresolved_process_path(&unresolved)).unwrap(),
4995 pid.to_string(),
4996 "bounded cleanup must leave durable unresolved-process evidence before returning"
4997 );
4998 assert!(
4999 NativeProcesses::new().is_alive(&recovered).is_err(),
5000 "restart must fail closed on the durable starting/PID journal rather than trust a bare PID"
5001 );
5002 processes.terminate(&unresolved).unwrap();
5003
5004 let runtime = root.path().join("post-spawn-stop-failed");
5005 let bin = runtime.join("bin");
5006 fs::create_dir_all(&bin).unwrap();
5007 #[cfg(windows)]
5008 let listener = bin.join("Runner.Listener.exe");
5009 #[cfg(not(windows))]
5010 let listener = bin.join("Runner.Listener");
5011 fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
5012 let attempt = RunnerAttempt::allocate(
5013 AttemptId::new_random(),
5014 policy.id,
5015 &runtime,
5016 FakeClock::default().now(),
5017 );
5018 processes.fail_post_spawn_at(PostSpawnBoundary::ChildMapInsert);
5019 processes.fail_next_post_spawn_stop();
5020 let failure = processes
5021 .spawn(&attempt, &EncodedJitConfig::new(JIT))
5022 .expect_err("the injected stop failure must preserve supervision");
5023 let live_pid = failure
5024 .live_pid
5025 .expect("live PID is returned to the journal");
5026 assert!(!failure.retryable);
5027 assert_long_lived_listener_ready(&processes, &attempt);
5028 assert!(NativeProcesses::identity_path(&attempt).is_file());
5029 assert_eq!(
5030 NativeProcesses::read_identity(&attempt)
5031 .unwrap()
5032 .unwrap()
5033 .pid(),
5034 live_pid
5035 );
5036 assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
5037 processes.terminate(&attempt).unwrap();
5038 }
5039
5040 #[test]
5041 #[ignore = "spawned only as the platform-stable native listener fixture"]
5042 fn long_lived_native_listener_helper() {
5043 let ready = std::env::var_os("RUNNER_MANAGER_TEST_LISTENER_READY")
5044 .map(PathBuf::from)
5045 .expect("the parent supplies the readiness path");
5046 fs::write(ready, b"ready\n").expect("the listener publishes readiness");
5047 std::thread::sleep(Duration::from_secs(30));
5048 }
5049
5050 fn assert_long_lived_listener_ready(processes: &NativeProcesses, attempt: &RunnerAttempt) {
5051 let ready = attempt.runtime_path().join(TEST_LISTENER_READY);
5052 let deadline = std::time::Instant::now() + Duration::from_secs(5);
5053 loop {
5054 if ready.is_file() {
5055 assert_eq!(fs::read(&ready).unwrap(), b"ready\n");
5056 return;
5057 }
5058 assert!(
5059 processes.is_alive(attempt).unwrap(),
5060 "the native listener exited before publishing readiness"
5061 );
5062 assert!(
5063 std::time::Instant::now() < deadline,
5064 "the native listener stayed alive but never published readiness"
5065 );
5066 std::thread::sleep(Duration::from_millis(10));
5067 }
5068 }
5069
5070 #[tokio::test]
5071 async fn every_production_launch_prunes_under_the_same_allocation_guard() {
5072 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
5073 harness.ready().await;
5074 assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 0);
5075 harness.launch().await;
5076 assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 1);
5077 assert_eq!(
5078 *harness.packages.prune_currents.lock().unwrap(),
5079 vec![harness.packages.version.clone()],
5080 "the leased current version is an exclusion, never the prune target"
5081 );
5082 }
5083
5084 fn assert_no_jit_file(runtime: &Path) {
5085 for entry in fs::read_dir(runtime).unwrap() {
5086 let path = entry.unwrap().path();
5087 if path.is_file() {
5088 let bytes = fs::read(&path).unwrap();
5089 assert!(
5090 !bytes
5091 .windows(JIT.len())
5092 .any(|window| window == JIT.as_bytes()),
5093 "a JIT payload survived in a runtime file"
5094 );
5095 }
5096 }
5097 }
5098
5099 #[test]
5100 fn production_listener_command_uses_the_supported_jit_contract() {
5101 let runtime = Path::new("runtime");
5102 let spec = runner_listener_spec(PathBuf::from("Runner.Listener"), runtime);
5103 let arguments: Vec<_> = spec
5104 .arguments()
5105 .iter()
5106 .map(|argument| argument.to_string_lossy().into_owned())
5107 .collect();
5108
5109 assert_eq!(arguments, ["run"]);
5110 assert!(
5111 !arguments
5112 .iter()
5113 .any(|argument| argument == "--jit-config-file"),
5114 "the obsolete file option would be rejected by Runner.Listener 2.336.0"
5115 );
5116 }
5117
5118 #[cfg(windows)]
5119 fn native_inspection_spec() -> SpawnSpec {
5120 SpawnSpec::new("powershell.exe").args([
5121 "-NoProfile",
5122 "-NonInteractive",
5123 "-Command",
5124 "Start-Sleep -Seconds 30",
5125 ])
5126 }
5127
5128 #[cfg(unix)]
5129 fn native_inspection_spec() -> SpawnSpec {
5130 SpawnSpec::new("/bin/sh").args(["-c", "sleep 30"])
5131 }
5132
5133 #[cfg(windows)]
5134 fn native_command_line(pid: u32) -> String {
5135 let output = std::process::Command::new("powershell.exe")
5136 .args([
5137 "-NoProfile",
5138 "-NonInteractive",
5139 "-Command",
5140 &format!("(Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}').CommandLine"),
5141 ])
5142 .output()
5143 .expect("PowerShell can inspect the native child");
5144 assert!(output.status.success(), "native process inspection failed");
5145 String::from_utf8(output.stdout).expect("Windows command lines are Unicode")
5146 }
5147
5148 #[cfg(target_os = "linux")]
5149 fn native_command_line(pid: u32) -> String {
5150 fs::read(format!("/proc/{pid}/cmdline"))
5151 .map(|bytes| String::from_utf8_lossy(&bytes).replace('\0', " "))
5152 .expect("/proc exposes the native child command line")
5153 }
5154
5155 #[cfg(target_os = "macos")]
5156 fn native_command_line(pid: u32) -> String {
5157 let output = std::process::Command::new("ps")
5158 .args(["-o", "command=", "-p", &pid.to_string()])
5159 .output()
5160 .expect("ps can inspect the native child");
5161 assert!(output.status.success(), "native process inspection failed");
5162 String::from_utf8(output.stdout).expect("the command line is UTF-8")
5163 }
5164
5165 #[tokio::test]
5168 async fn a_persistent_repository_leases_s1_and_journals_it_before_any_github_effect() {
5169 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5170 .with_persistent_workspace(2);
5171 harness.github.watch_journal(Arc::clone(&harness.store));
5172 harness.ready().await;
5173
5174 let attempt = harness.launch().await;
5175
5176 assert_eq!(
5177 attempt.workspace(),
5178 AttemptWorkspace::persistent_slot(nz(1)),
5179 "the lowest free slot is leased"
5180 );
5181 assert_eq!(attempt.runtime_path(), harness.slot_path(1));
5182 assert!(attempt.holds_slot_lease());
5183 assert_eq!(
5185 harness.attempt(attempt.id).runtime_path(),
5186 harness.slot_path(1)
5187 );
5188
5189 let facts = harness.github.registration_facts();
5193 assert_eq!(facts.len(), 1);
5194 assert_eq!(
5195 facts[0].leased_slots,
5196 vec![1],
5197 "the lease was journalled first"
5198 );
5199 assert_eq!(facts[0].work_folder, DEFAULT_WORK_FOLDER);
5200 }
5201
5202 #[tokio::test]
5203 async fn a_terminal_but_uncleaned_attempt_keeps_its_slot_without_holding_capacity() {
5204 let harness = Harness::new(
5205 FakeGithubLifecycle::default().fail(true),
5206 Arc::new(PersistentDemand),
5207 )
5208 .with_persistent_workspace(2);
5209 harness.ready().await;
5210
5211 harness.launch_result().await.unwrap_err();
5213 let first = harness.store.attempts().unwrap().remove(0);
5214 assert_eq!(first.state(), AttemptState::Failed);
5215 assert!(
5216 !first.state().counts_against_capacity(),
5217 "a concluded attempt is invisible to host capacity"
5218 );
5219 assert!(
5220 first.holds_slot_lease(),
5221 "and still owns its directory, so its slot is not free"
5222 );
5223
5224 let second = harness.launch().await;
5225 assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5226 assert_eq!(second.runtime_path(), harness.slot_path(2));
5227 assert_eq!(
5228 harness
5229 .store
5230 .slot_leases_for_policy(harness.policy.id)
5231 .unwrap()
5232 .len(),
5233 2
5234 );
5235 }
5236
5237 #[tokio::test]
5238 async fn two_sequential_allocations_at_capacity_one_reuse_s1_and_its_retained_work() {
5239 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5240 .with_persistent_workspace(1);
5241 harness.ready().await;
5242
5243 let first = harness.launch().await;
5244 assert_eq!(first.runtime_path(), harness.slot_path(1));
5245
5246 let checkout = harness.slot_path(1).join(DEFAULT_WORK_FOLDER).join("repo");
5248 fs::create_dir_all(&checkout).unwrap();
5249 fs::write(checkout.join("checkout.txt"), b"from the first job").unwrap();
5250
5251 harness.cleanup_retaining_work(first.id).await;
5252
5253 let second = harness.launch().await;
5254 assert_ne!(second.id, first.id);
5255 assert_eq!(
5256 second.workspace(),
5257 AttemptWorkspace::persistent_slot(nz(1)),
5258 "a released slot is leased again rather than skipped"
5259 );
5260 assert_eq!(
5261 second.runtime_path(),
5262 first.runtime_path(),
5263 "the same slot is the same exact path"
5264 );
5265 assert_eq!(
5266 fs::read_to_string(checkout.join("checkout.txt")).unwrap(),
5267 "from the first job",
5268 "the retained job workspace survived the second allocation"
5269 );
5270 assert!(harness.slot_path(1).join("runner-package").exists());
5272 }
5273
5274 #[tokio::test]
5275 async fn lowering_capacity_leaves_higher_slots_alone_and_raising_it_permits_them_again() {
5276 let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5277 .with_persistent_workspace(2);
5278 harness.ready().await;
5279
5280 let first = harness.launch().await;
5281 let second = harness.launch().await;
5282 assert_eq!(second.runtime_path(), harness.slot_path(2));
5283 let kept = harness
5284 .slot_path(2)
5285 .join(DEFAULT_WORK_FOLDER)
5286 .join("kept.txt");
5287 fs::create_dir_all(kept.parent().unwrap()).unwrap();
5288 fs::write(&kept, b"s2 was here").unwrap();
5289 harness.cleanup_retaining_work(second.id).await;
5290
5291 harness.policy.set_max_capacity(nz(1)).unwrap();
5293 let refusal = harness.launch_result().await.unwrap_err().to_string();
5294 assert!(
5295 refusal.contains("s1 to s1"),
5296 "the refusal names the ceiling it reached: {refusal}"
5297 );
5298 assert!(
5299 harness.slot_path(2).exists() && kept.exists(),
5300 "lowering capacity deletes nothing; the higher slot is merely unusable"
5301 );
5302
5303 harness.policy.set_max_capacity(nz(2)).unwrap();
5305 let third = harness.launch().await;
5306 assert_eq!(third.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5307 assert_eq!(third.runtime_path(), harness.slot_path(2));
5308 assert_eq!(fs::read_to_string(&kept).unwrap(), "s2 was here");
5309 assert!(first.holds_slot_lease(), "s1 was never disturbed");
5310 }
5311
5312 #[tokio::test]
5313 async fn organization_and_ephemeral_policies_never_enter_slot_allocation() {
5314 for policy in [
5315 fixtures::policy()
5316 .organization("octo")
5317 .autoscale("home", 2)
5318 .active()
5319 .build(),
5320 fixtures::policy()
5321 .repository("octo/repo")
5322 .autoscale("home", 2)
5323 .active()
5324 .build(),
5325 ] {
5326 let mut harness =
5327 Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5328 .with_host_runner_root();
5329 assert_eq!(policy.workspace_policy(), &WorkspacePolicy::Ephemeral);
5330 harness.policy = policy;
5331 harness.ready().await;
5332
5333 let attempt = harness.launch().await;
5334 assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
5335 assert_eq!(attempt.workspace().slot_number(), None);
5336 assert!(!attempt.holds_slot_lease());
5337 assert_eq!(
5338 attempt.runtime_path().parent().unwrap(),
5339 harness.host_root(),
5340 "a disposable attempt is a child of the host root, never of a slot"
5341 );
5342 assert!(
5343 harness
5344 .store
5345 .slot_leases_for_policy(harness.policy.id)
5346 .unwrap()
5347 .is_empty()
5348 );
5349 }
5350 }
5351
5352 #[tokio::test]
5353 async fn two_concurrent_allocations_never_share_a_slot() {
5354 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5355 .with_persistent_workspace(2);
5356 harness.github.watch_journal(Arc::clone(&harness.store));
5357 harness.ready().await;
5358
5359 let (first, second) = tokio::join!(harness.launch_result(), harness.launch_result());
5363 let first = first.unwrap();
5364 let second = second.unwrap();
5365
5366 let slots: BTreeSet<u16> = [&first, &second]
5367 .iter()
5368 .map(|attempt| {
5369 attempt
5370 .workspace()
5371 .slot_number()
5372 .expect("a persistent attempt leases a slot")
5373 })
5374 .collect();
5375 assert_eq!(slots, BTreeSet::from([1, 2]), "one slot each, never shared");
5376 assert_ne!(first.runtime_path(), second.runtime_path());
5377 assert_eq!(
5378 harness
5379 .store
5380 .slot_leases_for_policy(harness.policy.id)
5381 .unwrap()
5382 .len(),
5383 2
5384 );
5385
5386 let facts = harness.github.registration_facts();
5391 assert_eq!(facts.len(), 2);
5392 for fact in facts {
5393 let attempt = [&first, &second]
5394 .into_iter()
5395 .find(|attempt| runner_name(attempt.id) == fact.runner_name)
5396 .expect("every registration belongs to one of the two attempts");
5397 let slot = attempt
5398 .workspace()
5399 .slot_number()
5400 .expect("a persistent attempt leases a slot");
5401 assert!(
5402 fact.leased_slots.contains(&slot),
5403 "a JIT request never precedes its own lease: s{slot} not in {:?}",
5404 fact.leased_slots
5405 );
5406 assert_eq!(fact.work_folder, DEFAULT_WORK_FOLDER);
5407 }
5408 }
5409
5410 #[tokio::test]
5411 async fn the_database_is_the_final_fence_against_two_attempts_in_one_slot() {
5412 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5413 .with_persistent_workspace(2);
5414 harness.ready().await;
5415 let first = harness.launch().await;
5416
5417 let clash = RunnerAttempt::allocate_in(
5421 AttemptId::new_random(),
5422 harness.policy.id,
5423 first.runtime_path(),
5424 AttemptWorkspace::persistent_slot(nz(1)),
5425 harness.clock.now(),
5426 );
5427 assert!(matches!(
5428 harness.store.record_attempt(&clash).unwrap_err(),
5429 StoreError::SlotAlreadyLeased { slot: 1, .. }
5430 ));
5431
5432 let error = harness.launcher.record_allocation(&clash).unwrap_err();
5435 let rendered = error.to_string();
5436 assert!(rendered.contains("slot s1"), "{rendered}");
5437 assert!(rendered.contains("nothing was written"), "{rendered}");
5438 assert_eq!(
5439 harness.store.attempts().unwrap().len(),
5440 1,
5441 "the losing allocator journalled nothing"
5442 );
5443 }
5444
5445 #[test]
5446 fn slot_selection_fills_the_lowest_gap_and_stops_at_the_ceiling() {
5447 let leased = |slots: &[u16]| -> Vec<RunnerAttempt> {
5448 slots
5449 .iter()
5450 .map(|slot| {
5451 RunnerAttempt::allocate_in(
5452 AttemptId::new_random(),
5453 fixtures::POLICY_ID,
5454 format!("/srv/rman/acme/s{slot}"),
5455 AttemptWorkspace::persistent_slot(nz(*slot)),
5456 fixtures::created_at(),
5457 )
5458 })
5459 .collect()
5460 };
5461
5462 assert_eq!(lowest_free_slot(&[], nz(1)), Some(nz(1)));
5463 assert_eq!(lowest_free_slot(&leased(&[1]), nz(4)), Some(nz(2)));
5464 assert_eq!(lowest_free_slot(&leased(&[1, 3]), nz(4)), Some(nz(2)));
5466 assert_eq!(lowest_free_slot(&leased(&[1]), nz(1)), None);
5468 assert_eq!(lowest_free_slot(&leased(&[1, 2]), nz(2)), None);
5469 let ephemeral = vec![RunnerAttempt::allocate(
5471 AttemptId::new_random(),
5472 fixtures::POLICY_ID,
5473 "/srv/rman/host/abc",
5474 fixtures::created_at(),
5475 )];
5476 assert_eq!(lowest_free_slot(&ephemeral, nz(1)), Some(nz(1)));
5477 }
5478
5479 #[test]
5480 fn a_slot_is_reusable_only_when_it_is_empty_or_holds_one_real_work_directory() {
5481 let root = tempfile::tempdir().unwrap();
5482 let slot = root.path().join("s1");
5483 fs::create_dir(&slot).unwrap();
5484 accept_reusable_slot(&slot).expect("an empty slot is reusable");
5485
5486 fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
5487 accept_reusable_slot(&slot).expect("a retained job workspace is reusable");
5488
5489 fs::create_dir(slot.join("bin")).unwrap();
5492 fs::write(slot.join(".github-runner-id"), b"73").unwrap();
5493 let refusal = accept_reusable_slot(&slot).unwrap_err().to_string();
5494 assert!(refusal.contains("bin"), "{refusal}");
5495 assert!(refusal.contains(".github-runner-id"), "{refusal}");
5496
5497 let file_work = root.path().join("s2");
5499 fs::create_dir(&file_work).unwrap();
5500 fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
5501 assert!(accept_reusable_slot(&file_work).is_err());
5502 }
5503
5504 #[cfg(unix)]
5505 #[test]
5506 fn a_link_shaped_work_directory_is_refused_rather_than_followed() {
5507 let root = tempfile::tempdir().unwrap();
5511 let elsewhere = root.path().join("elsewhere");
5512 fs::create_dir(&elsewhere).unwrap();
5513
5514 let slot = root.path().join("s1");
5515 fs::create_dir(&slot).unwrap();
5516 std::os::unix::fs::symlink(&elsewhere, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
5517 assert!(accept_reusable_slot(&slot).is_err());
5518
5519 let linked_slot = root.path().join("s2");
5520 std::os::unix::fs::symlink(&elsewhere, &linked_slot).unwrap();
5521 assert!(create_or_validate_slot(&linked_slot).is_err());
5522 }
5523
5524 #[test]
5525 fn a_slot_standing_where_a_file_is_refuses_rather_than_replacing_it() {
5526 let root = tempfile::tempdir().unwrap();
5527 let occupied = root.path().join("s1");
5528 fs::write(&occupied, b"an operator's file").unwrap();
5529 let refusal = create_or_validate_slot(&occupied).unwrap_err().to_string();
5530 assert!(refusal.contains("is not a directory"), "{refusal}");
5531 assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
5532
5533 let fresh = root.path().join("s2");
5534 create_or_validate_slot(&fresh).expect("a missing slot is created");
5535 assert!(fresh.is_dir());
5536 create_or_validate_slot(&fresh).expect("an existing directory is accepted");
5537 }
5538
5539 #[test]
5540 fn the_retained_work_directory_is_matched_the_way_the_filesystem_matches_it() {
5541 assert!(is_work_folder(OsStr::new(DEFAULT_WORK_FOLDER)));
5542 assert!(!is_work_folder(OsStr::new("_work2")));
5543 assert_eq!(is_work_folder(OsStr::new("_Work")), cfg!(windows));
5547 }
5548
5549 #[test]
5550 fn package_materialization_never_overwrites_or_follows_a_retained_work_directory() {
5551 let root = tempfile::tempdir().unwrap();
5552 let package = root.path().join("package");
5553 fs::create_dir_all(package.join("bin")).unwrap();
5554 fs::write(package.join("bin").join("Runner.Listener"), b"binary").unwrap();
5555 fs::create_dir_all(package.join("externals").join(DEFAULT_WORK_FOLDER)).unwrap();
5557
5558 let slot = root.path().join("s1");
5559 let retained = slot.join(DEFAULT_WORK_FOLDER).join("repo");
5560 fs::create_dir_all(&retained).unwrap();
5561 fs::write(retained.join("checkout.txt"), b"from the first job").unwrap();
5562
5563 copy_package_tree(&package, &slot).expect("the package lays out around `_work`");
5564 assert!(slot.join("bin").join("Runner.Listener").exists());
5565 assert!(
5566 slot.join("externals").join(DEFAULT_WORK_FOLDER).is_dir(),
5567 "the guard is top-level only"
5568 );
5569 assert_eq!(
5570 fs::read_to_string(retained.join("checkout.txt")).unwrap(),
5571 "from the first job"
5572 );
5573
5574 fs::create_dir(package.join(DEFAULT_WORK_FOLDER)).unwrap();
5576 let error = copy_package_tree(&package, &slot).unwrap_err();
5577 assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
5578 assert_eq!(
5579 fs::read_to_string(retained.join("checkout.txt")).unwrap(),
5580 "from the first job"
5581 );
5582 }
5583
5584 #[test]
5585 fn rolling_back_a_materialization_keeps_a_slot_but_removes_a_disposable_directory() {
5586 let root = tempfile::tempdir().unwrap();
5587
5588 let slot = root.path().join("s1");
5589 let retained = slot.join(DEFAULT_WORK_FOLDER);
5590 fs::create_dir_all(retained.join("repo")).unwrap();
5591 fs::write(retained.join("repo").join("checkout.txt"), b"kept").unwrap();
5592 fs::create_dir_all(slot.join("bin")).unwrap();
5593 fs::write(slot.join(".github-runner-id"), b"73").unwrap();
5594 let persistent = RunnerAttempt::allocate_in(
5595 AttemptId::new_random(),
5596 fixtures::POLICY_ID,
5597 &slot,
5598 AttemptWorkspace::persistent_slot(nz(1)),
5599 fixtures::created_at(),
5600 );
5601
5602 remove_materialized_package(&persistent).unwrap();
5603 assert!(slot.is_dir(), "the slot itself is not removed");
5604 assert!(!slot.join("bin").exists());
5605 assert!(!slot.join(".github-runner-id").exists());
5606 assert_eq!(
5607 fs::read_to_string(retained.join("repo").join("checkout.txt")).unwrap(),
5608 "kept"
5609 );
5610
5611 let disposable_path = root.path().join("abcdef012345");
5612 fs::create_dir_all(disposable_path.join(DEFAULT_WORK_FOLDER)).unwrap();
5613 let disposable = RunnerAttempt::allocate(
5614 AttemptId::new_random(),
5615 fixtures::POLICY_ID,
5616 &disposable_path,
5617 fixtures::created_at(),
5618 );
5619 remove_materialized_package(&disposable).unwrap();
5620 assert!(
5621 !disposable_path.exists(),
5622 "a disposable directory is still removed whole"
5623 );
5624 }
5625
5626 fn litter_the_slot(slot: &Path) {
5636 for directory in ["bin", "externals", "_diag"] {
5637 fs::create_dir_all(slot.join(directory)).unwrap();
5638 }
5639 fs::write(slot.join("bin").join("Runner.Listener"), b"binary").unwrap();
5640 for file in SENSITIVE_SLOT_ENTRIES
5641 .iter()
5642 .filter(|entry| !slot.join(entry).is_dir())
5643 {
5644 fs::write(slot.join(file), b"runner state").unwrap();
5645 }
5646 fs::write(
5648 slot.join(format!(
5649 "{}0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0.tmp",
5650 RestrictiveHandoff::NAME_PREFIX
5651 )),
5652 JIT.as_bytes(),
5653 )
5654 .unwrap();
5655 }
5656
5657 fn retain_under_work(slot: &Path) -> PathBuf {
5659 let checkout = slot.join(DEFAULT_WORK_FOLDER).join("repo").join("target");
5660 fs::create_dir_all(&checkout).unwrap();
5661 let marker = checkout.join("build-output.bin");
5662 fs::write(&marker, RETAINED).unwrap();
5663 marker
5664 }
5665
5666 const RETAINED: &str = "a Git-ignored build output the next job reuses";
5668
5669 fn entries_of(directory: &Path) -> Vec<String> {
5671 let mut names: Vec<String> = fs::read_dir(directory)
5672 .unwrap()
5673 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
5674 .collect();
5675 names.sort();
5676 names
5677 }
5678
5679 fn only_the_job_workspace() -> Vec<String> {
5681 vec![DEFAULT_WORK_FOLDER.to_owned()]
5682 }
5683
5684 #[cfg(unix)]
5685 #[test]
5686 fn disposable_tree_removal_does_not_open_a_dotnet_diagnostic_fifo() {
5687 use std::sync::mpsc;
5688
5689 let temporary = tempfile::tempdir().unwrap();
5690 let tree = temporary.path().join("attempt");
5691 let diagnostic = tree.join("tmp/clr-debug-pipe-runner-in");
5692 fs::create_dir_all(diagnostic.parent().unwrap()).unwrap();
5693 assert!(
5694 std::process::Command::new("mkfifo")
5695 .arg(&diagnostic)
5696 .status()
5697 .unwrap()
5698 .success()
5699 );
5700
5701 let (finished, result) = mpsc::channel();
5702 std::thread::spawn(move || {
5703 let removed = remove_runtime_tree(&tree);
5704 let _ = finished.send(removed);
5705 });
5706
5707 result
5708 .recv_timeout(Duration::from_secs(2))
5709 .expect("runtime deletion must not wait for a FIFO peer")
5710 .unwrap();
5711 assert!(!diagnostic.exists());
5712 }
5713
5714 struct BlockedDeletion {
5730 directory: PathBuf,
5731 #[cfg(windows)]
5732 _handle: fs::File,
5733 }
5734
5735 impl BlockedDeletion {
5736 const HELD: &'static str = "held-open";
5737
5738 fn inject(directory: &Path) -> Option<Self> {
5741 #[cfg(unix)]
5742 if !Self::refusal_is_possible() {
5743 return None;
5744 }
5745 fs::create_dir_all(directory).unwrap();
5746 fs::write(
5747 directory.join(Self::HELD),
5748 b"a file the scrub cannot remove",
5749 )
5750 .unwrap();
5751 #[cfg(windows)]
5752 let handle = {
5753 use std::os::windows::fs::OpenOptionsExt;
5754
5755 fs::OpenOptions::new()
5756 .read(true)
5757 .share_mode(0)
5758 .open(directory.join(Self::HELD))
5759 .expect("the blocking handle opens")
5760 };
5761 #[cfg(unix)]
5762 Self::set_mode(directory, 0o555);
5763 Some(Self {
5764 directory: directory.to_path_buf(),
5765 #[cfg(windows)]
5766 _handle: handle,
5767 })
5768 }
5769
5770 fn release(self) {
5771 drop(self);
5772 }
5773
5774 #[cfg(unix)]
5775 fn refusal_is_possible() -> bool {
5776 let probe = tempfile::tempdir().unwrap();
5777 let directory = probe.path().join("probe");
5778 fs::create_dir(&directory).unwrap();
5779 fs::write(directory.join("file"), b"probe").unwrap();
5780 Self::set_mode(&directory, 0o555);
5781 let refused = fs::remove_dir_all(&directory).is_err();
5782 Self::set_mode(&directory, 0o755);
5783 refused
5784 }
5785
5786 #[cfg(unix)]
5787 fn set_mode(directory: &Path, mode: u32) {
5788 use std::os::unix::fs::PermissionsExt;
5789
5790 let mut permissions = fs::metadata(directory).unwrap().permissions();
5791 permissions.set_mode(mode);
5792 fs::set_permissions(directory, permissions).unwrap();
5793 }
5794 }
5795
5796 impl Drop for BlockedDeletion {
5797 fn drop(&mut self) {
5798 #[cfg(unix)]
5799 Self::set_mode(&self.directory, 0o755);
5800 #[cfg(not(unix))]
5801 let _ = &self.directory;
5802 }
5803 }
5804
5805 #[tokio::test]
5806 async fn two_sequential_jobs_keep_the_checkout_and_start_without_the_earlier_runner_state() {
5807 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5808 .with_persistent_workspace(1);
5809 harness.ready().await;
5810
5811 let first = harness.launch().await;
5812 let slot = harness.slot_path(1);
5813 assert_eq!(first.runtime_path(), slot);
5814 assert_eq!(
5815 read_runner_id(&slot),
5816 Some(73),
5817 "the attempt registered, so its identity is on disk"
5818 );
5819 let marker = retain_under_work(&slot);
5820 litter_the_slot(&slot);
5821
5822 harness.cleanup_retaining_work(first.id).await;
5823
5824 assert_eq!(entries_of(&slot), only_the_job_workspace());
5827 assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
5828 assert_eq!(
5829 read_runner_id(&slot),
5830 None,
5831 "the first attempt's registration identity is gone before the second starts"
5832 );
5833 assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
5834 assert!(!harness.attempt(first.id).holds_slot_lease());
5835
5836 let second = harness.launch().await;
5837 assert_ne!(second.id, first.id);
5838 assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(1)));
5839 assert_eq!(
5840 second.runtime_path(),
5841 slot,
5842 "the same slot, so the same retained `_work`"
5843 );
5844 assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
5845 }
5846
5847 #[tokio::test]
5848 async fn cleaning_a_persistent_slot_needs_no_policy_and_scans_no_directory_for_ownership() {
5849 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5850 .with_persistent_workspace(1);
5851 harness.ready().await;
5852 let attempt = harness.launch().await;
5853 let slot = harness.slot_path(1);
5854 let marker = retain_under_work(&slot);
5855 litter_the_slot(&slot);
5856 harness.conclude(attempt.id);
5857
5858 harness
5864 .store
5865 .remove_policy(harness.policy.id, harness.policy.revision())
5866 .unwrap();
5867 assert!(harness.store.policy(harness.policy.id).unwrap().is_none());
5868
5869 harness
5870 .launcher
5871 .clean(attempt.id)
5872 .await
5873 .expect("journal facts alone are enough to clean the slot");
5874
5875 assert_eq!(entries_of(&slot), only_the_job_workspace());
5876 assert!(marker.exists());
5877 assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
5878 }
5879
5880 #[tokio::test]
5881 async fn an_injected_partial_deletion_quarantines_the_slot_across_a_restart() {
5882 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5883 .with_persistent_workspace(2);
5884 harness.ready().await;
5885 let first = harness.launch().await;
5886 let slot = harness.slot_path(1);
5887 let marker = retain_under_work(&slot);
5888 litter_the_slot(&slot);
5889 harness.conclude(first.id);
5890
5891 let Some(block) = BlockedDeletion::inject(&slot.join("bin")) else {
5892 eprintln!(
5893 "skipped: this account cannot be refused a deletion, so no partial deletion can \
5894 be injected"
5895 );
5896 return;
5897 };
5898
5899 let refusal = harness
5900 .launcher
5901 .clean(first.id)
5902 .await
5903 .expect_err("a deletion that failed may not report a cleaned slot");
5904 let rendered = refusal.reason.to_string();
5905 assert!(rendered.contains("could not be removed"), "{rendered}");
5906
5907 let held = harness.attempt(first.id);
5908 assert_eq!(held.state(), AttemptState::Failed, "still not cleaned");
5909 assert!(held.holds_slot_lease(), "so the slot is still leased");
5910 assert!(
5911 !held.state().counts_against_capacity(),
5912 "and a concluded attempt still costs the host no capacity"
5913 );
5914
5915 let restarted = harness.restart();
5920 restarted
5921 .recover_startup(std::slice::from_ref(&harness.policy))
5922 .await
5923 .expect("one quarantined slot does not stop the host recovering");
5924 assert_eq!(
5925 harness.attempt(first.id).state(),
5926 AttemptState::Failed,
5927 "the quarantine survived the restart"
5928 );
5929 assert!(
5930 harness
5931 .reconcile_events
5932 .events()
5933 .iter()
5934 .any(|event| matches!(
5935 event,
5936 LifecycleEvent::AttemptCleanFailed {
5937 reason: "slot_entry_could_not_be_removed",
5938 ..
5939 }
5940 )),
5941 "the refusal is reported rather than retried in silence"
5942 );
5943
5944 let guard = harness.allocation_lock.acquire().await.unwrap();
5947 let second = restarted
5948 .launch(LaunchRequest {
5949 host: &harness.host,
5950 policy: &harness.policy,
5951 allocation_guard: &guard,
5952 })
5953 .await
5954 .expect("the host can still launch");
5955 assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5956 drop(guard);
5957
5958 block.release();
5961 restarted
5962 .clean(first.id)
5963 .await
5964 .expect("the retried cleanup completes");
5965 assert_eq!(entries_of(&slot), only_the_job_workspace());
5966 assert!(marker.exists());
5967 assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
5968 }
5969
5970 #[tokio::test]
5981 async fn an_injected_deletion_failure_leaves_a_disposable_attempt_uncleaned() {
5982 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
5983 harness.ready().await;
5984 let attempt = harness.launch().await;
5985 let runtime = attempt.runtime_path().to_path_buf();
5986 assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
5987 harness.conclude(attempt.id);
5988
5989 let Some(block) = BlockedDeletion::inject(&runtime.join("held-open-subdirectory")) else {
5990 eprintln!(
5991 "skipped: this account cannot be refused a deletion, so no partial deletion can be injected"
5992 );
5993 return;
5994 };
5995
5996 let refusal = harness
5997 .launcher
5998 .clean(attempt.id)
5999 .await
6000 .expect_err("a deletion that failed may not report a removed workspace");
6001 let rendered = refusal.reason.to_string();
6002 assert!(
6003 rendered.contains("could not be removed"),
6004 "the refusal names what happened: {rendered}"
6005 );
6006 assert_ne!(
6007 harness.attempt(attempt.id).state(),
6008 AttemptState::Cleaned,
6009 "an attempt whose directory is still on disk is not cleaned"
6010 );
6011 assert!(
6012 runtime.is_dir(),
6013 "the directory the removal could not finish is still there, which is the fact the journal must keep agreeing with"
6014 );
6015
6016 block.release();
6019 harness
6020 .launcher
6021 .clean(attempt.id)
6022 .await
6023 .expect("the retried cleanup completes");
6024 assert!(!runtime.exists(), "the whole attempt directory goes");
6025 assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
6026 }
6027
6028 #[tokio::test]
6029 async fn changing_a_repository_back_to_ephemeral_leaves_every_old_slot_untouched() {
6030 let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6031 .with_persistent_workspace(1);
6032 harness.ready().await;
6033 let first = harness.launch().await;
6034 let slot = harness.slot_path(1);
6035 let marker = retain_under_work(&slot);
6036 harness.cleanup_retaining_work(first.id).await;
6037
6038 harness
6042 .policy
6043 .set_workspace_policy(WorkspacePolicy::Ephemeral)
6044 .unwrap();
6045
6046 let second = harness.launch().await;
6047 assert_eq!(second.workspace(), AttemptWorkspace::Ephemeral);
6048 assert_eq!(
6049 second.runtime_path().parent().unwrap(),
6050 harness.host_root(),
6051 "a disposable attempt is a child of the host root"
6052 );
6053 assert!(slot.is_dir(), "the old slot is left where it stands");
6054 assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
6055
6056 harness.conclude(second.id);
6059 harness.launcher.clean(second.id).await.unwrap();
6060 assert!(!second.runtime_path().exists());
6061 assert!(marker.exists());
6062 }
6063
6064 #[tokio::test]
6065 async fn a_persistent_slot_is_scrubbed_only_after_the_process_is_signalled_and_gone() {
6066 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6067 .with_persistent_workspace(1);
6068 harness.ready().await;
6069 let slot = harness.slot_path(1);
6070 fs::create_dir_all(&slot).unwrap();
6071 let marker = retain_under_work(&slot);
6072 litter_the_slot(&slot);
6073
6074 let id = AttemptId::new_random();
6075 let mut attempt = RunnerAttempt::allocate_in(
6076 id,
6077 harness.policy.id,
6078 &slot,
6079 AttemptWorkspace::persistent_slot(nz(1)),
6080 harness.clock.now(),
6081 );
6082 attempt.jit_received(harness.clock.now()).unwrap();
6083 attempt.started(4242, harness.clock.now()).unwrap();
6084 harness.store.record_attempt(&attempt).unwrap();
6085 harness.clock.advance_secs(11);
6086 harness.processes.set_alive(true);
6087 harness
6088 .github
6089 .observe(GithubRunnerObservation::NotRegistered);
6090
6091 harness.launcher.supervise(&harness.policy).await.unwrap();
6092
6093 let actions = harness.processes.actions.lock().unwrap().clone();
6097 let intent = actions
6098 .iter()
6099 .position(|action| *action == "terminate_intent")
6100 .unwrap();
6101 let signal = actions
6102 .iter()
6103 .position(|action| *action == "terminate")
6104 .unwrap();
6105 assert!(intent < signal, "{actions:?}");
6106 assert!(!harness.processes.alive.load(Ordering::SeqCst));
6107
6108 let cleaned = harness.attempt(id);
6109 assert_eq!(cleaned.state(), AttemptState::Cleaned);
6110 assert!(matches!(
6111 cleaned.outcome(),
6112 Some(AttemptOutcome::Failed {
6113 reason: FailureReason::TerminatedAfterRegistrationTimeout
6114 })
6115 ));
6116 assert_eq!(entries_of(&slot), only_the_job_workspace());
6117 assert!(marker.exists());
6118 }
6119
6120 #[test]
6121 fn a_scrub_retains_one_real_work_directory_and_removes_every_other_entry() {
6122 let root = tempfile::tempdir().unwrap();
6123 let slot = root.path().join("s1");
6124 fs::create_dir(&slot).unwrap();
6125 let marker = retain_under_work(&slot);
6126 litter_the_slot(&slot);
6127 fs::write(slot.join("runner-package"), b"verified").unwrap();
6128
6129 scrub_slot_entries(&slot).expect("a slot of ordinary runner state scrubs");
6130 verify_slot_scrubbed(&slot).expect("and proves it afterwards");
6131
6132 assert_eq!(entries_of(&slot), only_the_job_workspace());
6133 assert!(marker.exists());
6134 }
6135
6136 #[test]
6137 fn a_residue_refusal_never_reports_the_under_count_as_the_fact() {
6138 let slot = Path::new("/runners/s1");
6139
6140 let counted = residue_detail(slot, 2, &["`bin`".to_owned()]);
6143 assert!(counted.contains("2 entries other than"), "{counted}");
6144 assert!(counted.contains("including `bin`"), "{counted}");
6145 assert_eq!(
6146 residue_detail(slot, 1, &[]),
6147 format!(
6148 "1 entry other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}",
6149 slot.display()
6150 )
6151 );
6152
6153 let raced = residue_detail(slot, 0, &["`.credentials`".to_owned()]);
6158 assert!(!raced.contains('0'), "{raced}");
6159 assert!(raced.contains("reported nothing but"), "{raced}");
6160 assert!(raced.contains("`.credentials` survived cleanup"), "{raced}");
6161 }
6162
6163 #[test]
6164 fn verification_asks_the_filesystem_rather_than_the_listing_that_missed_an_entry() {
6165 let root = tempfile::tempdir().unwrap();
6166 let slot = root.path().join("s1");
6167 fs::create_dir(&slot).unwrap();
6168 fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6169 verify_slot_scrubbed(&slot).expect("only `_work` is a clean slot");
6170
6171 for survivor in ["bin", ".credentials", IDENTITY_FILE, RUNNER_ID_FILE] {
6175 fs::write(slot.join(survivor), b"left behind").unwrap();
6176 let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6177 assert_eq!(quarantine.refusal, SlotRefusal::Residue);
6178 assert!(
6179 quarantine.detail.contains(&format!("`{survivor}`")),
6180 "{quarantine}"
6181 );
6182 fs::remove_file(slot.join(survivor)).unwrap();
6183 }
6184
6185 let handoff = slot.join(format!("{}whatever.tmp", RestrictiveHandoff::NAME_PREFIX));
6188 fs::write(&handoff, JIT.as_bytes()).unwrap();
6189 let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6190 assert!(
6191 quarantine.detail.contains("an encoded JIT handoff"),
6192 "{quarantine}"
6193 );
6194 assert!(!quarantine.detail.contains(JIT), "{quarantine}");
6195 fs::remove_file(&handoff).unwrap();
6196
6197 fs::write(slot.join("ghp_DO_NOT_LEAK"), b"named by the job").unwrap();
6201 let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6202 assert!(
6203 quarantine.detail.contains("1 entry other than"),
6204 "{quarantine}"
6205 );
6206 assert!(
6207 !quarantine.detail.contains("ghp_DO_NOT_LEAK"),
6208 "{quarantine}"
6209 );
6210 }
6211
6212 #[test]
6213 fn a_slot_is_derived_from_the_journal_and_refused_when_it_disagrees() {
6214 let root = tempfile::tempdir().unwrap();
6215 let configured =
6216 LocalAbsolutePath::new(root.path().to_str().unwrap()).expect("a local absolute root");
6217 let slot = configured.as_path().join("s1");
6218 fs::create_dir(&slot).unwrap();
6219
6220 verify_journalled_slot(&slot, nz(1), Some(&configured))
6221 .expect("the journalled slot agrees");
6222 verify_journalled_slot(&slot, nz(1), None)
6223 .expect("and a policy that is gone removes a check, not the ability to clean");
6224
6225 assert_eq!(
6228 verify_journalled_slot(&slot, nz(2), None)
6229 .unwrap_err()
6230 .refusal,
6231 SlotRefusal::NotTheJournalledSlot
6232 );
6233 for stray in ["s1/nested", "not-a-slot", "s01"] {
6234 let path = configured.as_path().join(stray);
6235 assert_eq!(
6236 verify_journalled_slot(&path, nz(1), None)
6237 .unwrap_err()
6238 .refusal,
6239 SlotRefusal::NotTheJournalledSlot,
6240 "{}",
6241 path.display()
6242 );
6243 }
6244
6245 let elsewhere = tempfile::tempdir().unwrap();
6248 let other =
6249 LocalAbsolutePath::new(elsewhere.path().to_str().unwrap()).expect("a second root");
6250 assert_eq!(
6251 verify_journalled_slot(&slot, nz(1), Some(&other))
6252 .unwrap_err()
6253 .refusal,
6254 SlotRefusal::PolicyRootDisagrees
6255 );
6256 }
6257
6258 #[cfg(unix)]
6259 #[test]
6260 fn a_substituted_work_directory_quarantines_the_slot_and_deletes_nothing_outside_it() {
6261 let root = tempfile::tempdir().unwrap();
6265 let outside = root.path().join("operator-data");
6266 fs::create_dir(&outside).unwrap();
6267 let sentinel = outside.join("do-not-delete.txt");
6268 fs::write(
6269 &sentinel,
6270 b"an operator's data, outside every approved root",
6271 )
6272 .unwrap();
6273
6274 let slot = root.path().join("s1");
6275 fs::create_dir(&slot).unwrap();
6276 fs::create_dir(slot.join("bin")).unwrap();
6277 std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6278
6279 let quarantine = scrub_slot_entries(&slot).unwrap_err();
6280 assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
6281 assert!(
6282 sentinel.exists(),
6283 "the deletion followed the link out of the slot"
6284 );
6285 assert!(outside.is_dir());
6286 assert!(
6287 slot.join(DEFAULT_WORK_FOLDER).symlink_metadata().is_ok(),
6288 "the substituted link is left for the operator, never unlinked as if it were ours"
6289 );
6290
6291 let file_work = root.path().join("s2");
6295 fs::create_dir(&file_work).unwrap();
6296 fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
6297 assert_eq!(
6298 scrub_slot_entries(&file_work).unwrap_err().refusal,
6299 SlotRefusal::WorkNotADirectory
6300 );
6301 }
6302
6303 #[cfg(unix)]
6304 #[test]
6305 fn a_slot_replaced_by_a_link_out_of_its_root_is_refused_before_anything_is_read() {
6306 let root = tempfile::tempdir().unwrap();
6307 let outside = root.path().join("operator-data");
6308 fs::create_dir(&outside).unwrap();
6309 let sentinel = outside.join("do-not-delete.txt");
6310 fs::write(
6311 &sentinel,
6312 b"an operator's data, outside every approved root",
6313 )
6314 .unwrap();
6315
6316 let inside = root.path().join("inside");
6319 fs::create_dir(&inside).unwrap();
6320 let slot = inside.join("s1");
6321 std::os::unix::fs::symlink(&outside, &slot).unwrap();
6322
6323 assert_eq!(
6324 verify_journalled_slot(&slot, nz(1), None)
6325 .unwrap_err()
6326 .refusal,
6327 SlotRefusal::Containment
6328 );
6329 assert!(sentinel.exists());
6330 assert!(
6331 slot.symlink_metadata().is_ok(),
6332 "the link is left for the operator rather than removed as if it were ours"
6333 );
6334 }
6335
6336 #[cfg(windows)]
6346 #[test]
6347 fn a_slot_root_replaced_by_a_junction_is_refused_before_anything_is_read() {
6348 let root = tempfile::tempdir().unwrap();
6349 let outside = root.path().join("operator-data");
6350 fs::create_dir(&outside).unwrap();
6351 let sentinel = outside.join("do-not-delete.txt");
6352 fs::write(
6353 &sentinel,
6354 b"an operator's data, outside every approved root",
6355 )
6356 .unwrap();
6357
6358 let inside = root.path().join("inside");
6361 fs::create_dir(&inside).unwrap();
6362 let slot = inside.join("s1");
6363 let Some(()) = plant_junction(&slot, &outside) else {
6364 eprintln!("skipped: this machine would not create a directory junction");
6365 return;
6366 };
6367
6368 assert_eq!(
6369 verify_journalled_slot(&slot, nz(1), None)
6370 .unwrap_err()
6371 .refusal,
6372 SlotRefusal::Containment
6373 );
6374 assert!(
6375 sentinel.exists(),
6376 "the refusal resolved the junction and reached the operator's data"
6377 );
6378 assert!(
6379 slot.symlink_metadata().is_ok(),
6380 "the junction is left for the operator rather than removed as if it were ours"
6381 );
6382 }
6383
6384 #[cfg(windows)]
6393 fn plant_junction(link: &Path, target: &Path) -> Option<()> {
6394 let made = std::process::Command::new("cmd")
6395 .arg("/C")
6396 .arg("mklink")
6397 .arg("/J")
6398 .arg(link)
6399 .arg(target)
6400 .output()
6401 .ok()?;
6402 (made.status.success() && link.symlink_metadata().is_ok()).then_some(())
6403 }
6404
6405 #[cfg(windows)]
6406 #[test]
6407 fn a_work_directory_replaced_by_a_junction_fails_closed_and_deletes_nothing_beyond_it() {
6408 let root = tempfile::tempdir().unwrap();
6409 let outside = root.path().join("operator-data");
6410 fs::create_dir(&outside).unwrap();
6411 let sentinel = outside.join("do-not-delete.txt");
6412 fs::write(
6413 &sentinel,
6414 b"an operator's data, outside every approved root",
6415 )
6416 .unwrap();
6417
6418 let slot = root.path().join("s1");
6419 fs::create_dir(&slot).unwrap();
6420 fs::create_dir(slot.join("bin")).unwrap();
6421 let Some(()) = plant_junction(&slot.join(DEFAULT_WORK_FOLDER), &outside) else {
6422 eprintln!("skipped: this machine would not create a directory junction");
6423 return;
6424 };
6425
6426 let work = fs::symlink_metadata(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6430 assert!(is_link_like(&work), "a junction is a reparse point");
6431 let quarantine = scrub_slot_entries(&slot).unwrap_err();
6432 assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
6433 assert!(
6434 sentinel.exists(),
6435 "the deletion followed the junction out of the slot"
6436 );
6437 assert!(outside.is_dir());
6438
6439 let elsewhere = root.path().join("s2");
6442 fs::create_dir(&elsewhere).unwrap();
6443 fs::create_dir(elsewhere.join(DEFAULT_WORK_FOLDER)).unwrap();
6444 if plant_junction(&elsewhere.join("externals"), &outside).is_some() {
6445 scrub_slot_entries(&elsewhere).expect("an ordinary entry is removed, junction or not");
6446 verify_slot_scrubbed(&elsewhere).expect("and the slot verifies");
6447 assert!(sentinel.exists(), "the junction was followed, not unlinked");
6448 assert_eq!(entries_of(&elsewhere), only_the_job_workspace());
6449 }
6450 }
6451
6452 #[cfg(unix)]
6453 #[tokio::test]
6454 async fn a_substituted_work_directory_leaves_the_attempt_uncleaned_and_still_leased() {
6455 let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6456 .with_persistent_workspace(2);
6457 harness.ready().await;
6458 let first = harness.launch().await;
6459 let slot = harness.slot_path(1);
6460 harness.conclude(first.id);
6461
6462 let outside = harness._root.path().join("operator-data");
6463 fs::create_dir_all(&outside).unwrap();
6464 let sentinel = outside.join("do-not-delete.txt");
6465 fs::write(&sentinel, b"outside every approved root").unwrap();
6466 std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6467
6468 harness
6469 .launcher
6470 .clean(first.id)
6471 .await
6472 .expect_err("a slot whose `_work` was substituted is quarantined");
6473 assert!(sentinel.exists());
6474
6475 let held = harness.attempt(first.id);
6476 assert_eq!(held.state(), AttemptState::Failed);
6477 assert!(held.holds_slot_lease());
6478
6479 let second = harness.launch().await;
6481 assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
6482 }
6483
6484 #[test]
6485 fn a_slot_that_is_a_file_is_refused_and_a_slot_that_is_gone_is_not() {
6486 let root = tempfile::tempdir().unwrap();
6487 let occupied = root.path().join("s1");
6488 fs::write(&occupied, b"an operator's file").unwrap();
6489 verify_journalled_slot(&occupied, nz(1), None).expect("the path is the journalled slot");
6492 assert_eq!(
6493 slot_is_present(&occupied).unwrap_err().refusal,
6494 SlotRefusal::SlotNotADirectory
6495 );
6496 assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
6497
6498 assert!(!slot_is_present(&root.path().join("s2")).unwrap());
6501 let present = root.path().join("s3");
6502 fs::create_dir(&present).unwrap();
6503 assert!(slot_is_present(&present).unwrap());
6504 }
6505
6506 #[test]
6507 fn cleanup_dispatches_on_the_journalled_kind_and_not_on_what_the_directory_holds() {
6508 let root = tempfile::tempdir().unwrap();
6509
6510 let disposable = root.path().join("abcdef012345");
6514 fs::create_dir_all(disposable.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
6515 let ephemeral = RunnerAttempt::allocate(
6516 AttemptId::new_random(),
6517 fixtures::POLICY_ID,
6518 &disposable,
6519 fixtures::created_at(),
6520 );
6521 remove_materialized_package(&ephemeral).unwrap();
6522 assert!(!disposable.exists());
6523
6524 let slot = root.path().join("s1");
6526 fs::create_dir_all(slot.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
6527 fs::create_dir_all(slot.join("bin")).unwrap();
6528 let persistent = RunnerAttempt::allocate_in(
6529 AttemptId::new_random(),
6530 fixtures::POLICY_ID,
6531 &slot,
6532 AttemptWorkspace::persistent_slot(nz(1)),
6533 fixtures::created_at(),
6534 );
6535 remove_materialized_package(&persistent).unwrap();
6536 assert_eq!(entries_of(&slot), only_the_job_workspace());
6537 assert!(slot.join(DEFAULT_WORK_FOLDER).join("repo").is_dir());
6538 }
6539
6540 #[test]
6541 fn every_slot_refusal_names_a_distinct_event_class_and_keeps_the_lease() {
6542 let refusals = [
6543 SlotRefusal::NotTheJournalledSlot,
6544 SlotRefusal::PolicyRootDisagrees,
6545 SlotRefusal::Containment,
6546 SlotRefusal::SlotNotADirectory,
6547 SlotRefusal::Enumeration,
6548 SlotRefusal::WorkNotADirectory,
6549 SlotRefusal::Deletion,
6550 SlotRefusal::Residue,
6551 ];
6552 let classes: BTreeSet<&str> = refusals.iter().map(|refusal| refusal.class()).collect();
6553 assert_eq!(
6554 classes.len(),
6555 refusals.len(),
6556 "an event class shared by two refusals tells an operator less than it appears to"
6557 );
6558 for refusal in refusals {
6559 assert!(
6562 refusal
6563 .class()
6564 .chars()
6565 .all(|c| c.is_ascii_lowercase() || c == '_'),
6566 "{}",
6567 refusal.class()
6568 );
6569 assert!(
6570 refusal.remediation().contains("slot lease"),
6571 "every refusal has to say the lease is still held: {}",
6572 refusal.class()
6573 );
6574 }
6575 }
6576
6577 #[test]
6578 fn copy_package_tree_copies_files_and_preserves_paths_with_spaces() {
6579 let root = tempfile::tempdir().unwrap();
6580 let source = root.path().join("source with spaces");
6581 let dest = root.path().join("dest with spaces");
6582
6583 fs::create_dir_all(&source).unwrap();
6584 fs::write(source.join("file1.txt"), b"hello").unwrap();
6585
6586 let nested = source.join("nested dir");
6587 fs::create_dir_all(&nested).unwrap();
6588 fs::write(nested.join("file2.txt"), b"world").unwrap();
6589
6590 let nested_work = nested.join(DEFAULT_WORK_FOLDER);
6592 fs::create_dir_all(&nested_work).unwrap();
6593 fs::write(nested_work.join("allowed.txt"), b"allowed").unwrap();
6594
6595 copy_package_tree(&source, &dest).unwrap();
6596
6597 assert_eq!(fs::read_to_string(dest.join("file1.txt")).unwrap(), "hello");
6598 assert_eq!(
6599 fs::read_to_string(dest.join("nested dir").join("file2.txt")).unwrap(),
6600 "world"
6601 );
6602 assert_eq!(
6603 fs::read_to_string(
6604 dest.join("nested dir")
6605 .join(DEFAULT_WORK_FOLDER)
6606 .join("allowed.txt")
6607 )
6608 .unwrap(),
6609 "allowed"
6610 );
6611 }
6612
6613 #[test]
6614 fn copy_package_tree_refuses_top_level_work_folder() {
6615 let root = tempfile::tempdir().unwrap();
6616 let source = root.path().join("source");
6617 let dest = root.path().join("dest");
6618
6619 fs::create_dir_all(&source).unwrap();
6620 fs::write(source.join("file1.txt"), b"hello").unwrap();
6621
6622 let top_work = source.join(DEFAULT_WORK_FOLDER);
6624 fs::create_dir_all(&top_work).unwrap();
6625
6626 let err = copy_package_tree(&source, &dest).unwrap_err();
6627 assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
6628 }
6629}