#[cfg(test)]
use std::collections::VecDeque;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::ffi::OsStr;
use std::fmt;
use std::fs;
use std::io::Write;
use std::num::NonZeroU16;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use async_trait::async_trait;
use runner_manager_domain::attempt::{
AttemptOutcome, AttemptState, FailureReason, GithubRunnerObservation, IsolationProviderFailure,
RecoveryDecision, RecoveryObservation, RecoveryTimeouts, RunnerAttempt, authorize,
recovery_decision,
};
use runner_manager_domain::execution::{AttemptExecution, Backend, ImageReference};
use runner_manager_domain::model::{AttemptId, Clock, HostId, PolicyId, ScaleTarget};
use runner_manager_domain::path::LocalAbsolutePath;
use runner_manager_domain::policy::ScalePolicy;
use runner_manager_domain::store::{Store, StoreError};
use runner_manager_domain::workspace::{AttemptWorkspace, WorkspacePolicy};
use runner_manager_github::jit::{
DEFAULT_WORK_FOLDER, EncodedJitConfig, JitError, JitGateway, JitRegistration, JitRunnerRequest,
};
use runner_manager_github::rest::{CancelToken, InventoryGateway};
use runner_manager_platform::process::{
Adoption, ChildProcess, ProcessIdentity, RestrictiveHandoff, SpawnSpec, Termination,
};
use runner_manager_platform::runner_env::{self, RunnerEnv, RunnerPlatform};
use runner_manager_platform::runner_root::{
self, RootOwner, RootPreflight, RunnerRootError, default_runner_root,
};
use secrecy::SecretString;
use crate::dependency_caches::DependencyCaches;
use crate::oci::OciProcesses;
use crate::package::{
InstalledPackage, LastCheck, PackageCache, PackageContents, PackageError, RunnerVersion,
};
use crate::reconcile::{
AllocationGuard, CleanBackoff, EventSink, LaunchFailure, LaunchRequest, LifecycleEvent,
OutcomeKind, ReplacementIntent, ReservationWaits, RunnerLauncher, failure_reason_kind,
};
mod windows_hyperv;
pub use windows_hyperv::{WindowsHyperVContainers, WindowsHyperVHostState};
const IDENTITY_FILE: &str = ".runner-process.json";
const FALLBACK_IDENTITY_FILE: &str = ".runner-process.recovery.json";
const UNRESOLVED_PROCESS_FILE: &str = ".runner-process.unresolved";
const RUNNER_ID_FILE: &str = ".github-runner-id";
const TERMINATE_INTENT_FILE: &str = ".terminate-registration-timeout";
const MAX_POST_SPAWN_STOP_ATTEMPTS: usize = 3;
const SENSITIVE_SLOT_ENTRIES: &[&str] = &[
"bin",
"externals",
"run.sh",
"run.cmd",
"config.sh",
"config.cmd",
".runner",
".credentials",
".credentials_rsaparams",
".env",
".path",
"_diag",
IDENTITY_FILE,
FALLBACK_IDENTITY_FILE,
UNRESOLVED_PROCESS_FILE,
RUNNER_ID_FILE,
TERMINATE_INTENT_FILE,
];
#[cfg(test)]
const TEST_LISTENER_READY: &str = ".test-listener-ready";
fn runner_listener_spec(
program: PathBuf,
runtime: &Path,
caches: Vec<(&'static str, std::ffi::OsString)>,
host_env: &RunnerEnv,
) -> SpawnSpec {
let tmp = runtime.join("tmp");
let _ = std::fs::create_dir_all(&tmp);
let defaults = runner_env::platform_defaults(
RunnerPlatform::current(),
runtime,
&runner_env::Inherited::current(),
Path::exists,
);
for (name, dir) in &defaults {
if runner_env::DIRECTORY_VARIABLES.contains(name) {
let _ = std::fs::create_dir_all(dir);
}
}
let mut spec = SpawnSpec::new(program)
.arg("run")
.working_dir(runtime)
.expect_normal_scheduling();
for (name, value) in runner_env::runner_environment(defaults, caches, host_env) {
spec = spec.env(name, value);
}
spec.env("TMPDIR", &tmp).env("TEMP", &tmp).env("TMP", &tmp)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RetryPolicy {
pub max_attempts: u32,
pub initial: Duration,
pub maximum: Duration,
}
impl RetryPolicy {
#[must_use]
pub const fn bounded(max_attempts: u32, initial: Duration, maximum: Duration) -> Self {
Self {
max_attempts,
initial,
maximum,
}
}
pub(crate) fn delay(self, failure_index: u32) -> Duration {
let shift = failure_index.saturating_sub(1).min(31);
self.initial
.saturating_mul(1_u32 << shift)
.min(self.maximum)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttemptEvent {
State {
attempt: AttemptId,
state: AttemptState,
},
Retry {
attempt: AttemptId,
operation: &'static str,
delay: Duration,
},
Launched {
attempt: AttemptId,
started: bool,
timings: LaunchTimings,
},
Adopted {
attempt: AttemptId,
},
RemoteIdentityRecovered {
attempt: AttemptId,
runner_id: u64,
},
TerminateIntent {
attempt: AttemptId,
},
Terminated {
attempt: AttemptId,
},
Deregistered {
attempt: AttemptId,
runner_id: u64,
},
Concluded {
attempt: AttemptId,
outcome: OutcomeKind,
},
Cleaned {
attempt: AttemptId,
outcome: OutcomeKind,
},
QuarantinedOwnedResource {
attempt: AttemptId,
},
}
pub trait AttemptEventSink: fmt::Debug + Send + Sync {
fn emit(&self, event: AttemptEvent);
}
#[derive(Debug, Default)]
pub struct AttemptEventLog(Mutex<Vec<AttemptEvent>>);
impl AttemptEventLog {
#[must_use]
pub fn events(&self) -> Vec<AttemptEvent> {
self.0
.lock()
.map(|events| events.clone())
.unwrap_or_default()
}
}
impl AttemptEventSink for AttemptEventLog {
fn emit(&self, event: AttemptEvent) {
if let Ok(mut events) = self.0.lock() {
events.push(event);
}
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct NoAttemptEvents;
impl AttemptEventSink for NoAttemptEvents {
fn emit(&self, _event: AttemptEvent) {}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct LaunchTimings {
pub order_wait: Duration,
pub lock_wait: Duration,
pub turn_wait: Duration,
pub materialize: Duration,
pub package: PackageTimings,
pub prepare: Duration,
pub register: Duration,
pub spawn: Duration,
}
impl LaunchTimings {
#[must_use]
pub fn total(&self) -> Duration {
self.order_wait
+ self.lock_wait
+ self.turn_wait
+ self.materialize
+ self.prepare
+ self.register
+ self.spawn
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PackageTimings {
pub wait: Duration,
pub install: Duration,
pub verify: Duration,
pub copy: Duration,
pub downloaded: bool,
}
#[derive(Debug, Clone, Copy)]
enum LaunchStep {
Turn,
Materialize,
Prepare,
Register,
Spawn,
}
#[derive(Debug)]
struct LaunchStopwatch {
timings: LaunchTimings,
running: Option<(LaunchStep, std::time::Instant)>,
}
impl LaunchStopwatch {
fn new(waits: ReservationWaits) -> Self {
Self {
timings: LaunchTimings {
order_wait: waits.order,
lock_wait: waits.lock,
..LaunchTimings::default()
},
running: None,
}
}
fn begin(&mut self, step: LaunchStep) {
self.finish();
self.running = Some((step, std::time::Instant::now()));
}
fn finish(&mut self) {
if let Some((step, since)) = self.running.take() {
let slot = match step {
LaunchStep::Turn => &mut self.timings.turn_wait,
LaunchStep::Materialize => &mut self.timings.materialize,
LaunchStep::Prepare => &mut self.timings.prepare,
LaunchStep::Register => &mut self.timings.register,
LaunchStep::Spawn => &mut self.timings.spawn,
};
*slot = since.elapsed();
}
}
}
const SLOW_LAUNCH: Duration = Duration::from_secs(60);
#[derive(Debug, Clone, Copy, Default)]
pub struct TracingAttemptEvents;
impl AttemptEventSink for TracingAttemptEvents {
fn emit(&self, event: AttemptEvent) {
match event {
AttemptEvent::Retry {
attempt,
operation,
delay,
} => tracing::warn!(
event = "attempt_retry",
attempt_id = %attempt,
reason = operation,
retry_in_ms = millis(delay),
),
AttemptEvent::Launched {
attempt,
started,
timings,
} => {
macro_rules! launched {
($level:ident, $name:literal) => {
tracing::$level!(
event = $name,
attempt_id = %attempt,
duration_ms = millis(timings.total()),
order_wait_ms = millis(timings.order_wait),
lock_wait_ms = millis(timings.lock_wait),
turn_wait_ms = millis(timings.turn_wait),
materialize_ms = millis(timings.materialize),
package_wait_ms = millis(timings.package.wait),
package_install_ms = millis(timings.package.install),
package_verify_ms = millis(timings.package.verify),
package_copy_ms = millis(timings.package.copy),
package_downloaded = timings.package.downloaded,
prepare_ms = millis(timings.prepare),
register_ms = millis(timings.register),
spawn_ms = millis(timings.spawn),
)
};
}
if !started {
launched!(warn, "attempt_launch_failed");
} else if timings.total() >= SLOW_LAUNCH {
launched!(warn, "attempt_launch_slow");
} else {
launched!(info, "attempt_launched");
}
}
_ => {}
}
}
}
fn millis(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
#[async_trait]
pub trait DemandPersistence: fmt::Debug + Send + Sync {
async fn persists(&self, policy: PolicyId) -> bool;
}
#[derive(Debug, Clone, Copy, Default)]
pub struct PersistentDemand;
#[async_trait]
impl DemandPersistence for PersistentDemand {
async fn persists(&self, _policy: PolicyId) -> bool {
true
}
}
#[async_trait]
pub trait RetryDelay: fmt::Debug + Send + Sync {
async fn wait(&self, duration: Duration);
}
#[derive(Debug, Clone, Copy, Default)]
pub struct TokioRetryDelay;
#[async_trait]
impl RetryDelay for TokioRetryDelay {
async fn wait(&self, duration: Duration) {
tokio::time::sleep(duration).await;
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JitRequestFailure {
pub terminal: bool,
pub reason: FailureReason,
pub retry_after: Option<Duration>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LifecycleGithubObservation {
pub status: GithubRunnerObservation,
pub runner_id: Option<u64>,
}
impl LifecycleGithubObservation {
#[must_use]
pub const fn unreachable() -> Self {
Self {
status: GithubRunnerObservation::Unreachable,
runner_id: None,
}
}
#[must_use]
pub const fn not_registered() -> Self {
Self {
status: GithubRunnerObservation::NotRegistered,
runner_id: None,
}
}
#[must_use]
pub const fn registered(runner_id: u64, busy: bool) -> Self {
Self {
status: GithubRunnerObservation::Registered { busy },
runner_id: Some(runner_id),
}
}
}
#[async_trait]
pub trait LifecycleGithub: fmt::Debug + Send + Sync {
async fn register(
&self,
target: &ScaleTarget,
request: &JitRunnerRequest,
cancel: &CancelToken,
) -> Result<JitRegistration, JitRequestFailure>;
async fn observe(
&self,
target: &ScaleTarget,
attempt: AttemptId,
cancel: &CancelToken,
) -> LifecycleGithubObservation;
async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool;
}
#[async_trait]
impl<T> LifecycleGithub for T
where
T: JitGateway + InventoryGateway + fmt::Debug + Send + Sync,
{
async fn register(
&self,
target: &ScaleTarget,
request: &JitRunnerRequest,
cancel: &CancelToken,
) -> Result<JitRegistration, JitRequestFailure> {
self.generate_jit_config(target, request, cancel)
.await
.map_err(|error| {
let reason = if matches!(&error, JitError::Forbidden { .. }) {
FailureReason::Other(
"GitHub refused JIT registration with 403; check the App's runner permission and runner-group access"
.into(),
)
} else {
FailureReason::JitRequestFailed
};
JitRequestFailure {
terminal: error.is_terminal(),
reason,
retry_after: error
.rate_limited()
.map(|limit| limit.delay_from(self.now())),
}
})
}
async fn observe(
&self,
target: &ScaleTarget,
attempt: AttemptId,
cancel: &CancelToken,
) -> LifecycleGithubObservation {
let expected_name = runner_name(attempt);
match self.list_runners(target, cancel).await {
Ok(inventory) => inventory
.runners()
.iter()
.find(|runner| runner.name == expected_name)
.map_or(LifecycleGithubObservation::not_registered(), |runner| {
LifecycleGithubObservation::registered(runner.id, runner.busy)
}),
Err(_) => LifecycleGithubObservation::unreachable(),
}
}
async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool {
self.remove_runner(target, runner_id, cancel).await.is_ok()
}
}
#[async_trait]
pub trait RuntimePackages: fmt::Debug + Send + Sync {
async fn materialize(
&self,
attempt: &RunnerAttempt,
timings: &mut PackageTimings,
) -> Result<RunnerVersion, FailureReason>;
fn release(&self, attempt: AttemptId) -> Result<(), FailureReason>;
fn prune_obsolete_guarded(
&self,
authority: PruneAuthority<'_>,
attempts: &[RunnerAttempt],
) -> Result<(), FailureReason>;
}
pub struct PruneAuthority<'a> {
_guard: &'a AllocationGuard,
}
impl fmt::Debug for PruneAuthority<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("PruneAuthority")
}
}
impl<'a> PruneAuthority<'a> {
fn from_launch_request(guard: &'a AllocationGuard) -> Self {
Self { _guard: guard }
}
#[cfg(test)]
pub(crate) const fn for_test(guard: &'a AllocationGuard) -> Self {
Self { _guard: guard }
}
}
#[derive(Debug, Clone, Default)]
pub struct SharedPackages {
trees: Arc<tokio::sync::RwLock<()>>,
trusted: Arc<TrustedTrees>,
last_check: LastCheck,
failed_install: Arc<Mutex<Option<(std::time::Instant, FailureReason)>>>,
}
impl SharedPackages {
fn discard(
&self,
root: &Path,
origin: &Path,
contents: &Arc<PackageContents>,
reason: &'static str,
) {
if self.trusted.forget_if_holding(root, contents) {
tracing::warn!(
event = "package_verify_failed",
reason,
"a shared runner package does not match its verified archive; it is discarded \
and prepared again"
);
self.trusted.forget(origin);
}
}
fn note_install(&self, result: Result<(), &FailureReason>) {
*self
.failed_install
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = result
.err()
.filter(|reason| !package_failure_is_terminal(reason))
.map(|reason| (std::time::Instant::now(), reason.clone()));
}
fn install_failed_since(&self, queued: std::time::Instant) -> Option<FailureReason> {
self.failed_install
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.filter(|(at, _)| *at >= queued)
.map(|(_, reason)| reason.clone())
}
#[must_use]
pub fn last_check(&self) -> LastCheck {
self.last_check.clone()
}
}
#[cfg(test)]
impl SharedPackages {
pub(crate) async fn copying(&self) -> tokio::sync::RwLockReadGuard<'_, ()> {
self.trees.read().await
}
pub(crate) fn blind_to_edits_of(&self, root: &Path) {
let contents = self.trusted.contents_of(root);
self.trusted.record(root, contents).expect("recorded");
}
pub(crate) fn vouches_for(&self, root: &Path) -> bool {
self.trusted.contents_of(root).is_some()
}
}
#[derive(Debug)]
pub struct CachedRuntimePackages {
cache: Arc<PackageCache>,
shared: SharedPackages,
}
async fn timed<T>(slot: &mut Duration, work: impl std::future::Future<Output = T>) -> T {
let began = std::time::Instant::now();
let result = work.await;
*slot += began.elapsed();
result
}
impl CachedRuntimePackages {
async fn blocking<T: Send + 'static>(
&self,
root: &Path,
work: impl FnOnce(&TrustedTrees, &Path) -> T + Send + 'static,
) -> Result<T, FailureReason> {
let trusted = Arc::clone(&self.shared.trusted);
let root = root.to_path_buf();
tokio::task::spawn_blocking(move || work(&trusted, &root))
.await
.map_err(|_| FailureReason::ProcessStartFailed)
}
#[must_use]
pub const fn new(cache: Arc<PackageCache>, shared: SharedPackages) -> Self {
Self { cache, shared }
}
async fn reject_copy(
&self,
attempt: &RunnerAttempt,
root: &Path,
origin: &Path,
contents: &Arc<PackageContents>,
timings: &mut PackageTimings,
) -> Result<(), FailureReason> {
let (attempt_copy, checking, from) =
(attempt.clone(), Arc::clone(contents), origin.to_path_buf());
let differs = timed(
&mut timings.verify,
self.blocking(root, move |_, source| {
let _ = remove_materialized_package(&attempt_copy);
let shared = checking.verify(source, |_| false);
shared.and_then(|()| {
if from == source {
Ok(())
} else {
checking.verify(&from, |_| false)
}
})
}),
)
.await?;
if let Err(mismatch) = differs {
self.shared
.discard(root, origin, contents, mismatch.as_str());
}
Ok(())
}
async fn ready(
&self,
destination: &Path,
timings: &mut PackageTimings,
) -> Result<Option<(InstalledPackage, PathBuf)>, FailureReason> {
let Some(installed) = self.cache.reusable().map_err(package_failure)? else {
return Ok(None);
};
let destination = destination.to_path_buf();
let verifying = self.blocking(installed.root(), move |trusted, source| {
if trusted.holds(source) {
copy_origin_ready(source, &destination, trusted)
} else {
None
}
});
let origin = timed(&mut timings.verify, verifying).await?;
Ok(origin.map(|origin| (installed, origin)))
}
async fn install_trusted(
&self,
destination: &Path,
timings: &mut PackageTimings,
) -> Result<(InstalledPackage, PathBuf), FailureReason> {
let (mut installed, downloaded) = timed(
&mut timings.install,
self.cache.ensure_installed_reporting(),
)
.await
.map_err(package_failure)?;
timings.downloaded |= downloaded;
let fresh = downloaded
|| !timed(
&mut timings.verify,
self.blocking(installed.root(), TrustedTrees::holds),
)
.await?;
if fresh && !downloaded {
if self.shared.trusted.contents_of(installed.root()).is_some() {
tracing::warn!(
event = "package_verify_failed",
reason = "changed",
"a shared runner package changed since it was prepared; it is discarded and \
prepared again"
);
}
installed = timed(&mut timings.install, self.cache.reinstall())
.await
.map_err(package_failure)?;
timings.downloaded = true;
}
let destination = destination.to_path_buf();
let contents = installed.contents();
let origin = timed(
&mut timings.verify,
self.blocking(installed.root(), move |trusted, source| {
if fresh {
trusted.record(source, contents)?;
}
Ok::<_, std::io::Error>(copy_origin(source, &destination, trusted))
}),
)
.await?
.map_err(|_| FailureReason::ProcessStartFailed)?;
Ok((installed, origin))
}
}
#[async_trait]
impl RuntimePackages for CachedRuntimePackages {
async fn materialize(
&self,
attempt: &RunnerAttempt,
timings: &mut PackageTimings,
) -> Result<RunnerVersion, FailureReason> {
let destination = attempt.runtime_path().to_path_buf();
let trees = &self.shared.trees;
let mut installs = 0;
let (hold, (installed, origin)) = loop {
let queued = std::time::Instant::now();
let hold = timed(&mut timings.wait, trees.read()).await;
if let Some(ready) = self.ready(&destination, timings).await? {
break (hold, ready);
}
drop(hold);
let exclusive = timed(&mut timings.wait, trees.write()).await;
if let Some(reason) = self.shared.install_failed_since(queued) {
return Err(reason);
}
let installed = self.install_trusted(&destination, timings).await;
self.shared.note_install(installed.as_ref().map(|_| ()));
let ready = installed?;
installs += 1;
if installs == 2 {
break (exclusive.downgrade(), ready);
}
};
let Some(contents) = self.shared.trusted.contents_of(installed.root()) else {
self.shared.trusted.forget(installed.root());
return Err(FailureReason::ProcessStartFailed);
};
let copied_from = origin.clone();
let copied = timed(
&mut timings.copy,
self.blocking(installed.root(), {
let destination = destination.clone();
move |trusted, source| {
copy_package_from(source, &origin, &destination, trusted)?;
if trusted.holds(source) {
Ok(())
} else {
Err(std::io::Error::other(
"the runner package changed while copied",
))
}
}
}),
)
.await?;
if copied.is_err() {
drop(hold);
self.reject_copy(attempt, installed.root(), &copied_from, &contents, timings)
.await?;
return Err(FailureReason::ProcessStartFailed);
}
if let Err(error) = self.cache.lease(attempt, installed.version()) {
drop(hold);
self.reject_copy(attempt, installed.root(), &copied_from, &contents, timings)
.await?;
return Err(package_failure(error));
}
drop(hold);
let persistent = matches!(attempt.workspace(), AttemptWorkspace::PersistentSlot { .. });
let checking = Arc::clone(&contents);
let verified = timed(
&mut timings.verify,
self.blocking(installed.root(), move |_, _| {
checking.verify(&destination, |name| persistent && is_work_folder(name))
}),
)
.await?;
if let Err(mismatch) = verified {
let _ = self.cache.release(attempt.id);
tracing::warn!(
event = "package_copy_differs",
attempt_id = %attempt.id,
reason = mismatch.as_str(),
"a runner's package copy does not match the package's verified archive; it is \
being removed and will not be used"
);
self.reject_copy(attempt, installed.root(), &copied_from, &contents, timings)
.await?;
return Err(FailureReason::ProcessStartFailed);
}
Ok(installed.version().clone())
}
fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
self.cache.release(attempt).map_err(package_failure)
}
fn prune_obsolete_guarded(
&self,
_authority: PruneAuthority<'_>,
attempts: &[RunnerAttempt],
) -> Result<(), FailureReason> {
let Ok(_exclusive) = self.shared.trees.try_write() else {
return Ok(());
};
let Some(current) = self.cache.newest_installed().map_err(package_failure)? else {
return Ok(());
};
for installed in self.cache.installed().map_err(package_failure)? {
if installed.version() != current.version() {
match self.cache.prune(installed.version(), attempts) {
Ok(()) => self.shared.trusted.forget(installed.root()),
Err(PackageError::VersionInUse { .. }) => {}
Err(error) => return Err(package_failure(error)),
}
}
}
Ok(())
}
}
fn package_failure(error: PackageError) -> FailureReason {
error.failure_reason().unwrap_or(FailureReason::Other(
"runner package cache operation failed".into(),
))
}
fn package_failure_is_terminal(reason: &FailureReason) -> bool {
matches!(
reason,
FailureReason::RunnerPackageUnverified | FailureReason::RunnerVersionRejected
)
}
const WORKSPACE_NAME_LEN: usize = 12;
fn workspace_name(id: AttemptId) -> String {
let full = id.to_string();
full.chars()
.filter(|c| *c != '-')
.take(WORKSPACE_NAME_LEN)
.collect()
}
#[derive(Debug, Clone)]
struct Placement {
runtime: PathBuf,
workspace: AttemptWorkspace,
}
fn root_failure(error: RunnerRootError) -> LifecycleError {
tracing::warn!(
error_kind = error.kind(),
"the runner root refused this launch, so no attempt was created; the host will \
retry every poll until the cause is resolved. Re-running `host set-runtime-root` \
with the same path re-runs this check and prints the directory and the \
remediation in full"
);
LifecycleError::Failed(FailureReason::Other(error.to_string()))
}
fn lowest_free_slot(leases: &[RunnerAttempt], ceiling: NonZeroU16) -> Option<NonZeroU16> {
let held: BTreeSet<u16> = leases
.iter()
.filter_map(|attempt| attempt.workspace().slot_number())
.collect();
(1..=ceiling.get())
.find(|slot| !held.contains(slot))
.and_then(NonZeroU16::new)
}
fn create_or_validate_slot(slot: &Path) -> Result<(), LifecycleError> {
match fs::symlink_metadata(slot) {
Ok(metadata) if is_link_like(&metadata) => Err(slot_refusal(
slot,
"is a symbolic link, junction or other reparse point, which could place runner \
files outside the configured root",
)),
Ok(metadata) if !metadata.is_dir() => Err(slot_refusal(slot, "is not a directory")),
Ok(_) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => fs::create_dir(slot)
.map_err(|source| slot_refusal(slot, format!("could not be created: {source}"))),
Err(source) => Err(slot_refusal(
slot,
format!("could not be inspected: {source}"),
)),
}
}
fn accept_reusable_slot(slot: &Path) -> Result<(), LifecycleError> {
let unreadable =
|source: std::io::Error| slot_refusal(slot, format!("could not be read: {source}"));
let entries = fs::read_dir(slot).map_err(unreadable)?;
let mut refused: Vec<String> = Vec::new();
for entry in entries {
let entry = entry.map_err(unreadable)?;
let name = entry.file_name();
let metadata = fs::symlink_metadata(entry.path()).map_err(|source| {
slot_refusal(
slot,
format!("entry {name:?} could not be inspected: {source}"),
)
})?;
if is_retainable_work_folder(&name, &metadata) {
continue;
}
refused.push(name.to_string_lossy().into_owned());
}
if refused.is_empty() {
return Ok(());
}
refused.sort();
Err(slot_refusal(
slot,
format!(
"holds {} that this attempt may not reuse: [{}]. A reusable slot is empty or holds \
one real `{DEFAULT_WORK_FOLDER}` directory and nothing else; remove or move the \
entries listed, or let cleanup and recovery resolve them",
if refused.len() == 1 {
"an entry"
} else {
"entries"
},
refused.join(", ")
),
))
}
fn slot_refusal(slot: &Path, detail: impl fmt::Display) -> LifecycleError {
LifecycleError::Failed(FailureReason::Other(format!(
"the persistent slot {} {detail}",
slot.display()
)))
}
fn is_work_folder(name: &OsStr) -> bool {
if cfg!(windows) {
name.eq_ignore_ascii_case(DEFAULT_WORK_FOLDER)
} else {
name == OsStr::new(DEFAULT_WORK_FOLDER)
}
}
fn is_link_like(metadata: &fs::Metadata) -> bool {
if metadata.file_type().is_symlink() {
return true;
}
#[cfg(windows)]
{
use std::os::windows::fs::MetadataExt;
const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x0000_0400;
metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0
}
#[cfg(not(windows))]
false
}
fn is_retainable_work_folder(name: &OsStr, metadata: &fs::Metadata) -> bool {
is_work_folder(name) && metadata.is_dir() && !is_link_like(metadata)
}
fn remove_materialized_package(attempt: &RunnerAttempt) -> std::io::Result<()> {
match attempt.workspace() {
AttemptWorkspace::Ephemeral => remove_runtime_tree(attempt.runtime_path()),
AttemptWorkspace::PersistentSlot { .. } => scrub_slot_entries(attempt.runtime_path())
.map_err(|quarantine| std::io::Error::other(quarantine.to_string())),
}
}
fn remove_runtime_tree(path: &Path) -> std::io::Result<()> {
#[cfg(windows)]
{
match remove_dir_all::remove_dir_all(path) {
Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
unlink_reparse_points(path);
remove_dir_all::remove_dir_all(path)
}
other => other,
}
}
#[cfg(not(windows))]
{
fs::remove_dir_all(path)
}
}
#[cfg(windows)]
fn unlink_reparse_points(dir: &Path) {
let Ok(entries) = fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
let Ok(metadata) = entry.metadata() else {
continue;
};
if is_link_like(&metadata) {
let _ = fs::remove_dir(&path).or_else(|_| fs::remove_file(&path));
} else if metadata.is_dir() {
unlink_reparse_points(&path);
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SlotRefusal {
NotTheJournalledSlot,
PolicyRootDisagrees,
Containment,
SlotNotADirectory,
Enumeration,
WorkNotADirectory,
Deletion,
Residue,
}
impl SlotRefusal {
const fn class(self) -> &'static str {
match self {
Self::NotTheJournalledSlot => "slot_path_is_not_the_journalled_slot",
Self::PolicyRootDisagrees => "slot_root_disagrees_with_policy",
Self::Containment => "slot_escapes_its_root",
Self::SlotNotADirectory => "slot_is_not_a_directory",
Self::Enumeration => "slot_could_not_be_enumerated",
Self::WorkNotADirectory => "retained_work_is_not_a_directory",
Self::Deletion => "slot_entry_could_not_be_removed",
Self::Residue => "slot_still_holds_runner_state",
}
}
const fn remediation(self) -> &'static str {
match self {
Self::NotTheJournalledSlot | Self::PolicyRootDisagrees | Self::Containment => {
"the attempt keeps its slot lease and nothing was removed; correct the \
repository's persistent workspace path, or remove the slot directory by hand \
once you have confirmed what is in it"
}
Self::SlotNotADirectory | Self::WorkNotADirectory => {
"the attempt keeps its slot lease and nothing was removed; a job replaced the \
slot or its `_work` with a link, so inspect it before deleting anything and \
treat the retained workspace as untrusted"
}
Self::Enumeration | Self::Deletion | Self::Residue => {
"the attempt keeps its slot lease and will be cleaned again on the next pass; \
release whatever is holding the files open, or remove the slot's contents by \
hand leaving only `_work`"
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct SlotQuarantine {
refusal: SlotRefusal,
detail: String,
}
impl SlotQuarantine {
fn new(refusal: SlotRefusal, detail: impl Into<String>) -> Self {
Self {
refusal,
detail: detail.into(),
}
}
}
impl fmt::Display for SlotQuarantine {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}; {}", self.detail, self.refusal.remediation())
}
}
fn verify_journalled_slot(
runtime: &Path,
slot: NonZeroU16,
configured: Option<&LocalAbsolutePath>,
) -> Result<(), SlotQuarantine> {
let mislaid = || {
SlotQuarantine::new(
SlotRefusal::NotTheJournalledSlot,
format!(
"the journalled runtime {} is not the slot s{slot} this attempt was allocated as",
runtime.display()
),
)
};
let local = |path: &Path| {
path.to_str()
.and_then(|raw| LocalAbsolutePath::new(raw).ok())
.ok_or_else(mislaid)
};
let runtime_path = local(runtime)?;
let root = local(runtime.parent().ok_or_else(mislaid)?)?;
let name = AttemptWorkspace::persistent_slot(slot)
.slot_directory_name()
.expect("a persistent workspace names its slot directory");
let derived = runner_root::derive_child(&root, &name).map_err(|_| mislaid())?;
if derived != runtime_path {
return Err(mislaid());
}
if let Some(configured) = configured
&& configured != &root
{
return Err(SlotQuarantine::new(
SlotRefusal::PolicyRootDisagrees,
format!(
"the journalled slot {} is not under the repository's configured persistent root \
{}",
runtime.display(),
configured.as_str()
),
));
}
runner_root::verify_containment(&root, &derived).map_err(|source| {
SlotQuarantine::new(
SlotRefusal::Containment,
format!("the journalled slot is not inside the root it was allocated from: {source}"),
)
})
}
fn slot_is_present(slot: &Path) -> Result<bool, SlotQuarantine> {
match fs::symlink_metadata(slot) {
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(SlotQuarantine::new(
SlotRefusal::SlotNotADirectory,
format!(
"the slot {} could not be inspected: {:?}",
slot.display(),
source.kind()
),
)),
Ok(metadata) if !metadata.is_dir() || is_link_like(&metadata) => Err(SlotQuarantine::new(
SlotRefusal::SlotNotADirectory,
format!(
"the slot {} is a link or a file rather than a real directory",
slot.display()
),
)),
Ok(_) => Ok(true),
}
}
fn scrub_slot_entries(slot: &Path) -> Result<(), SlotQuarantine> {
let unreadable = |source: std::io::Error| {
SlotQuarantine::new(
SlotRefusal::Enumeration,
format!(
"the entries of {} could not be listed: {:?}",
slot.display(),
source.kind()
),
)
};
for entry in fs::read_dir(slot).map_err(unreadable)? {
let name = entry.map_err(unreadable)?.file_name();
let path = slot.join(&name);
let Some(metadata) = listed_entry_metadata(&path).map_err(unreadable)? else {
continue;
};
if is_work_folder(&name) {
if is_retainable_work_folder(&name, &metadata) {
continue;
}
return Err(SlotQuarantine::new(
SlotRefusal::WorkNotADirectory,
format!(
"the retained `{DEFAULT_WORK_FOLDER}` in {} is a link or a file rather than a \
real directory",
slot.display()
),
));
}
remove_slot_entry(&path, &metadata).map_err(|source| {
SlotQuarantine::new(
SlotRefusal::Deletion,
format!(
"an entry of {} could not be removed: {:?}",
slot.display(),
source.kind()
),
)
})?;
}
Ok(())
}
fn listed_entry_metadata(path: &Path) -> std::io::Result<Option<fs::Metadata>> {
match fs::symlink_metadata(path) {
Ok(metadata) => Ok(Some(metadata)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(source) => Err(source),
}
}
fn remove_slot_entry(path: &Path, metadata: &fs::Metadata) -> std::io::Result<()> {
let removed = if is_link_like(metadata) {
fs::remove_file(path).or_else(|_| fs::remove_dir(path))
} else if metadata.is_dir() {
remove_runtime_tree(path)
} else {
fs::remove_file(path)
};
match removed {
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
other => other,
}
}
fn verify_slot_scrubbed(slot: &Path) -> Result<(), SlotQuarantine> {
let unreadable = |source: std::io::Error| {
SlotQuarantine::new(
SlotRefusal::Enumeration,
format!(
"the entries of {} could not be listed to verify the scrub: {:?}",
slot.display(),
source.kind()
),
)
};
let mut residue = 0_usize;
let mut named: Vec<String> = Vec::new();
for entry in fs::read_dir(slot).map_err(unreadable)? {
let name = entry.map_err(unreadable)?.file_name();
let Some(metadata) = listed_entry_metadata(&slot.join(&name)).map_err(unreadable)? else {
continue;
};
if is_retainable_work_folder(&name, &metadata) {
continue;
}
residue = residue.saturating_add(1);
if name
.to_string_lossy()
.starts_with(RestrictiveHandoff::NAME_PREFIX)
{
named.push("an encoded JIT handoff".to_owned());
}
}
named.extend(
SENSITIVE_SLOT_ENTRIES
.iter()
.filter(|entry| fs::symlink_metadata(slot.join(entry)).is_ok())
.map(|entry| format!("`{entry}`")),
);
if residue == 0 && named.is_empty() {
return Ok(());
}
named.sort_unstable();
named.dedup();
Err(SlotQuarantine::new(
SlotRefusal::Residue,
residue_detail(slot, residue, &named),
))
}
fn residue_detail(slot: &Path, residue: usize, named: &[String]) -> String {
if residue == 0 {
format!(
"the listing of {} reported nothing but `{DEFAULT_WORK_FOLDER}`, yet {} survived \
cleanup",
slot.display(),
named.join(", ")
)
} else {
format!(
"{residue} entr{} other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}{}",
if residue == 1 { "y" } else { "ies" },
slot.display(),
if named.is_empty() {
String::new()
} else {
format!(", including {}", named.join(", "))
}
)
}
}
fn replacement_operation(outcome: &AttemptOutcome) -> Option<&'static str> {
match outcome {
AttemptOutcome::Failed {
reason: FailureReason::JitExpired,
} => Some("jit_expired_replacement"),
AttemptOutcome::Failed {
reason: FailureReason::ProcessExitedUnexpectedly,
} => Some("exit_before_acceptance_replacement"),
_ => None,
}
}
fn copy_origin(source: &Path, destination: &Path, seeds: &TrustedTrees) -> PathBuf {
#[cfg(unix)]
{
package_seed(source, destination, seeds)
.ok()
.filter(|seed| refuse_work_folder(seed).is_ok())
.unwrap_or_else(|| source.to_path_buf())
}
#[cfg(not(unix))]
{
let _ = (destination, seeds);
source.to_path_buf()
}
}
fn copy_origin_ready(source: &Path, destination: &Path, seeds: &TrustedTrees) -> Option<PathBuf> {
#[cfg(unix)]
{
match seed_location(source, destination) {
Ok(None) => Some(source.to_path_buf()),
Ok(Some(seeds_root)) => held_seed(source, &seeds_root, seeds)
.ok()
.flatten()
.filter(|seed| refuse_work_folder(seed).is_ok()),
Err(_) => None,
}
}
#[cfg(not(unix))]
{
let _ = (destination, seeds);
Some(source.to_path_buf())
}
}
fn copy_package_from(
source: &Path,
origin: &Path,
destination: &Path,
seeds: &TrustedTrees,
) -> std::io::Result<()> {
refuse_work_folder(source)?;
#[cfg(unix)]
{
clone_tree(origin, destination)?;
if origin != source && !seeds.holds(origin) {
return Err(std::io::Error::other(
"the package seed changed while cloned",
));
}
Ok(())
}
#[cfg(not(unix))]
{
let _ = (origin, seeds);
copy_package_entries(source, destination, true)
}
}
fn refuse_work_folder(source: &Path) -> std::io::Result<()> {
if source.join(DEFAULT_WORK_FOLDER).exists() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"a cached runner package contains a _work folder, which means it was used \
to run a job before it was archived; the cache must only contain clean \
extracts to prevent data leakage",
));
}
Ok(())
}
#[cfg(unix)]
const PACKAGE_SEED_DIR: &str = ".runner-package";
#[cfg(unix)]
fn package_seed(
source: &Path,
destination: &Path,
seeds: &TrustedTrees,
) -> std::io::Result<PathBuf> {
match seed_location(source, destination)? {
Some(root) => refresh_package_seed(source, &root, seeds),
None => Ok(source.to_path_buf()),
}
}
#[cfg(unix)]
fn seed_location(source: &Path, destination: &Path) -> std::io::Result<Option<PathBuf>> {
use std::os::unix::fs::MetadataExt as _;
if !cfg!(target_os = "macos") {
return Ok(None);
}
let root = destination
.parent()
.ok_or_else(|| std::io::Error::other("a runtime directory has no parent"))?;
if fs::metadata(source)?.dev() == fs::metadata(root)?.dev() {
return Ok(None);
}
Ok(Some(root.join(PACKAGE_SEED_DIR)))
}
#[derive(Debug, Default)]
pub(crate) struct TrustedTrees {
trees: Mutex<HashMap<PathBuf, TrustedTree>>,
}
#[derive(Debug, Clone)]
struct TrustedTree {
#[cfg(unix)]
print: [u8; 32],
contents: Option<Arc<PackageContents>>,
}
impl TrustedTrees {
fn trees(&self) -> std::sync::MutexGuard<'_, HashMap<PathBuf, TrustedTree>> {
self.trees
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn holds(&self, root: &Path) -> bool {
let Some(tree) = self.trees().get(root).cloned() else {
return false;
};
#[cfg(unix)]
{
tree_fingerprint(root).is_ok_and(|current| current == tree.print)
}
#[cfg(not(unix))]
{
let _ = tree;
true
}
}
fn contents_of(&self, root: &Path) -> Option<Arc<PackageContents>> {
self.trees()
.get(root)
.and_then(|tree| tree.contents.clone())
}
fn record(&self, root: &Path, contents: Option<Arc<PackageContents>>) -> std::io::Result<()> {
let tree = TrustedTree {
#[cfg(unix)]
print: tree_fingerprint(root)?,
contents,
};
self.trees().insert(root.to_path_buf(), tree);
Ok(())
}
fn forget_if_holding(&self, root: &Path, contents: &Arc<PackageContents>) -> bool {
let mut trees = self.trees();
let holding = trees
.get(root)
.and_then(|tree| tree.contents.as_ref())
.is_some_and(|current| Arc::ptr_eq(current, contents));
if holding {
trees.remove(root);
}
holding
}
fn forget(&self, root: &Path) {
self.trees().remove(root);
}
}
#[cfg(unix)]
fn tree_fingerprint(root: &Path) -> std::io::Result<[u8; 32]> {
use sha2::{Digest as _, Sha256};
use std::os::unix::fs::MetadataExt as _;
fn walk(root: &Path, path: &Path, hasher: &mut Sha256) -> std::io::Result<()> {
let metadata = fs::symlink_metadata(path)?;
let relative = path.strip_prefix(root).unwrap_or(path);
hasher.update(relative.as_os_str().as_encoded_bytes());
hasher.update([0]);
for field in [
metadata.ino(),
metadata.size(),
u64::from(metadata.mode()),
u64::from(metadata.uid()),
u64::from(metadata.gid()),
] {
hasher.update(field.to_le_bytes());
}
hasher.update(metadata.ctime().to_le_bytes());
hasher.update(metadata.ctime_nsec().to_le_bytes());
if metadata.is_dir() {
let mut children = fs::read_dir(path)?
.map(|entry| entry.map(|entry| entry.path()))
.collect::<std::io::Result<Vec<_>>>()?;
children.sort();
for child in children {
walk(root, &child, hasher)?;
}
}
Ok(())
}
let mut hasher = Sha256::new();
walk(root, root, &mut hasher)?;
Ok(hasher.finalize().into())
}
#[cfg(unix)]
fn seed_path(source: &Path, seeds: &Path) -> std::io::Result<PathBuf> {
let version = source
.file_name()
.ok_or_else(|| std::io::Error::other("a cache entry has no name"))?;
Ok(seeds.join(version))
}
#[cfg(unix)]
fn held_seed(
source: &Path,
seeds: &Path,
trusted: &TrustedTrees,
) -> std::io::Result<Option<PathBuf>> {
use crate::package::MANIFEST_FILE;
let seed = seed_path(source, seeds)?;
let manifest = fs::read(source.join(MANIFEST_FILE))?;
let held = fs::read(seed.join(MANIFEST_FILE)).is_ok_and(|held| held == manifest)
&& trusted.holds(&seed);
Ok(held.then_some(seed))
}
#[cfg(unix)]
fn refresh_package_seed(
source: &Path,
seeds: &Path,
trusted: &TrustedTrees,
) -> std::io::Result<PathBuf> {
if let Some(seed) = held_seed(source, seeds, trusted)? {
return Ok(seed);
}
let seed = seed_path(source, seeds)?;
match remove_runtime_tree(seeds) {
Err(error) if error.kind() != std::io::ErrorKind::NotFound => return Err(error),
_ => {}
}
fs::create_dir_all(seeds)?;
let staging = seeds.join(format!(".staging-{}", uuid::Uuid::new_v4()));
fs::create_dir(&staging)?;
if let Err(error) = clone_tree(source, &staging).and_then(|()| fs::rename(&staging, &seed)) {
let _ = remove_runtime_tree(&staging);
return Err(error);
}
trusted.record(&seed, None)?;
Ok(seed)
}
#[cfg(unix)]
fn clone_tree(source: &Path, destination: &Path) -> std::io::Result<()> {
#[cfg(target_os = "macos")]
const CLONE: &[&str] = &["-c"];
#[cfg(target_os = "linux")]
const CLONE: &[&str] = &["--reflink=auto"];
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
const CLONE: &[&str] = &[];
let from = format!("{}/.", source.display());
let cp = |extra: &[&str]| {
std::process::Command::new("cp")
.arg("-a")
.args(extra)
.arg(&from)
.arg(destination)
.status()
};
if cp(CLONE)?.success() || cp(&[])?.success() {
Ok(())
} else {
Err(std::io::Error::other("cp failed"))
}
}
#[cfg(not(unix))]
fn copy_package_entries(source: &Path, destination: &Path, top_level: bool) -> std::io::Result<()> {
fs::create_dir_all(destination)?;
for entry in fs::read_dir(source)? {
let entry = entry?;
if top_level && is_work_folder(&entry.file_name()) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"the runner package holds a top-level `{DEFAULT_WORK_FOLDER}`; copying \
it would overwrite the job workspace a persistent slot retains"
),
));
}
let target = destination.join(entry.file_name());
if entry.file_type()?.is_dir() {
copy_package_entries(&entry.path(), &target, false)?;
} else {
fs::copy(entry.path(), target)?;
}
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProcessStartFailure {
pub reason: FailureReason,
pub retryable: bool,
pub live_pid: Option<u32>,
}
impl ProcessStartFailure {
pub(crate) fn before_spawn(reason: FailureReason) -> Self {
Self {
reason,
retryable: true,
live_pid: None,
}
}
fn after_spawn_stopped() -> Self {
Self {
reason: FailureReason::ProcessStartFailed,
retryable: false,
live_pid: None,
}
}
fn after_spawn_live(pid: u32) -> Self {
Self::after_spawn_live_with_reason(pid, FailureReason::ProcessStartFailed)
}
fn after_spawn_live_with_reason(pid: u32, reason: FailureReason) -> Self {
Self {
reason,
retryable: false,
live_pid: Some(pid),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderCapability {
Ready,
Unsupported,
NotInstalled,
PermissionDenied,
ImageUnavailableOrIncompatible,
DiskQuotaUnavailable,
Degraded,
}
impl ProviderCapability {
#[must_use]
pub const fn display_name(self) -> &'static str {
match self {
Self::Ready => "ready",
Self::Unsupported => "unsupported",
Self::NotInstalled => "not installed",
Self::PermissionDenied => "permission denied",
Self::ImageUnavailableOrIncompatible => "image unavailable or incompatible",
Self::DiskQuotaUnavailable => "disk quota unavailable",
Self::Degraded => "degraded",
}
}
pub(crate) fn refusal(self) -> FailureReason {
FailureReason::IsolationProvider(match self {
Self::Ready | Self::Degraded => IsolationProviderFailure::Degraded,
Self::Unsupported => IsolationProviderFailure::Unsupported,
Self::NotInstalled => IsolationProviderFailure::NotInstalled,
Self::PermissionDenied => IsolationProviderFailure::PermissionDenied,
Self::ImageUnavailableOrIncompatible => {
IsolationProviderFailure::ImageUnavailableOrIncompatible
}
Self::DiskQuotaUnavailable => IsolationProviderFailure::DiskQuotaUnavailable,
})
}
}
fn safe_isolation_reason(
reason: FailureReason,
fallback: IsolationProviderFailure,
) -> FailureReason {
match reason {
FailureReason::IsolationProvider(_) => reason,
_ => FailureReason::IsolationProvider(fallback),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedEnvironment {
attempt: AttemptId,
identity: Option<EnvironmentIdentity>,
cache_target: Option<runner_manager_domain::model::ScaleTarget>,
}
impl PreparedEnvironment {
#[must_use]
pub fn isolated(attempt: AttemptId, identity: EnvironmentIdentity) -> Self {
Self {
attempt,
identity: Some(identity),
cache_target: None,
}
}
#[must_use]
pub fn identity(&self) -> Option<&EnvironmentIdentity> {
self.identity.as_ref()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedEnvironment {
pub provider_kind: Backend,
pub image: ImageReference,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderDiagnostic {
CapabilityUnavailable,
ImageRejected,
PrepareFailed,
StartFailed,
OwnershipMismatch,
CleanupDeferred,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EnvironmentIdentity {
NativeProcess(u32),
Isolated {
host: HostId,
attempt: AttemptId,
provider_kind: Backend,
environment_id: String,
resolved_image: ImageReference,
generation: String,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EnvironmentState {
Starting,
Running,
Exited,
Missing,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderStop {
Stopped,
StillRunning,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderDestroy {
Destroyed,
Deferred(&'static str),
}
fn identity_matches_intent(
host: HostId,
attempt: &RunnerAttempt,
identity: &EnvironmentIdentity,
) -> bool {
identity_same_allocation(host, attempt, identity)
&& match (attempt.execution(), identity) {
(
AttemptExecution::Isolated { environment_id, .. },
EnvironmentIdentity::Isolated {
environment_id: id, ..
},
) => environment_id
.as_ref()
.is_none_or(|journal_id| journal_id == id),
_ => false,
}
}
fn identity_same_allocation(
host: HostId,
attempt: &RunnerAttempt,
identity: &EnvironmentIdentity,
) -> bool {
match (attempt.execution(), identity) {
(
AttemptExecution::Isolated {
provider_kind,
environment_id: _,
resolved_image,
generation,
},
EnvironmentIdentity::Isolated {
host: owner,
attempt: owner_attempt,
provider_kind: kind,
environment_id: _,
resolved_image: image,
generation: resource_generation,
},
) => {
*owner == host
&& *owner_attempt == attempt.id
&& provider_kind == kind
&& resolved_image == image
&& generation == resource_generation
}
_ => false,
}
}
pub struct OneTimeJitHandoff<'a> {
config: &'a EncodedJitConfig,
}
impl fmt::Debug for OneTimeJitHandoff<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("OneTimeJitHandoff([redacted])")
}
}
impl<'a> OneTimeJitHandoff<'a> {
pub(crate) fn new(config: &'a EncodedJitConfig) -> Self {
Self { config }
}
pub fn consume(self) -> &'a EncodedJitConfig {
self.config
}
}
pub trait ExecutionProvider: fmt::Debug + Send + Sync {
fn probe(&self, policy: &ScalePolicy) -> ProviderCapability {
if policy.execution_policy().is_native() {
ProviderCapability::Ready
} else {
ProviderCapability::Unsupported
}
}
fn resolve(&self, policy: &ScalePolicy) -> Result<Option<ResolvedEnvironment>, FailureReason> {
if policy.execution_policy().is_native() {
Ok(None)
} else {
Err(FailureReason::Other(
"execution provider unavailable".into(),
))
}
}
fn prepare(
&self,
attempt: &RunnerAttempt,
policy: &ScalePolicy,
) -> Result<PreparedEnvironment, FailureReason> {
if !attempt.execution().is_native() || !policy.execution_policy().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
Ok(PreparedEnvironment {
attempt: attempt.id,
identity: None,
cache_target: None,
})
}
fn start(
&self,
prepared: PreparedEnvironment,
attempt: &RunnerAttempt,
handoff: OneTimeJitHandoff<'_>,
) -> Result<EnvironmentIdentity, ProcessStartFailure> {
if prepared.attempt != attempt.id || !attempt.execution().is_native() {
return Err(ProcessStartFailure::before_spawn(FailureReason::Other(
"execution provider identity mismatch".into(),
)));
}
self.spawn(attempt, handoff.consume())
.map(EnvironmentIdentity::NativeProcess)
}
fn inspect(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
if !attempt.execution().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
self.is_alive(attempt).map(|alive| {
if alive {
EnvironmentState::Running
} else {
EnvironmentState::Missing
}
})
}
fn stop(&self, attempt: &RunnerAttempt) -> Result<ProviderStop, FailureReason> {
if !attempt.execution().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
self.terminate(attempt).map(|()| ProviderStop::Stopped)
}
fn destroy(&self, attempt: &RunnerAttempt) -> Result<ProviderDestroy, FailureReason> {
if !attempt.execution().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
Ok(ProviderDestroy::Destroyed)
}
fn recover(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
self.inspect(attempt)
}
fn enumerate_owned(&self, _host_id: HostId) -> Vec<EnvironmentIdentity> {
Vec::new() }
fn diagnostics(&self, _attempt: &RunnerAttempt) -> Vec<ProviderDiagnostic> {
Vec::new() }
fn owns(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
Ok(attempt.execution().is_native())
}
fn spawn(
&self,
attempt: &RunnerAttempt,
config: &EncodedJitConfig,
) -> Result<u32, ProcessStartFailure>;
fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason>;
fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason>;
fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool;
fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool;
fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
}
pub use ExecutionProvider as ProcessSupervisor;
#[derive(Debug, Default)]
pub struct NativeProcesses {
children: Mutex<BTreeMap<AttemptId, ChildProcess>>,
successful_exits: Mutex<BTreeMap<AttemptId, bool>>,
runner_env_file: Option<PathBuf>,
dependency_caches: Option<DependencyCaches>,
docker_cleanup: Option<crate::docker_cleanup::DockerCleanup>,
#[cfg(test)]
post_spawn_faults: Mutex<VecDeque<PostSpawnBoundary>>,
#[cfg(test)]
post_spawn_reaps: std::sync::atomic::AtomicUsize,
#[cfg(test)]
post_spawn_stop_failures: std::sync::atomic::AtomicUsize,
#[cfg(test)]
use_long_lived_test_listener: std::sync::atomic::AtomicBool,
}
#[derive(Debug)]
pub struct PlatformExecutionProvider {
native: NativeProcesses,
oci: OciProcesses,
windows_hyper_v: WindowsHyperVContainers,
}
impl PlatformExecutionProvider {
#[must_use]
pub fn new(host_id: HostId) -> Self {
Self {
native: NativeProcesses::new(),
oci: OciProcesses::new(host_id),
windows_hyper_v: WindowsHyperVContainers::new(host_id),
}
}
#[must_use]
pub fn with_runner_env_file(mut self, path: PathBuf) -> Self {
self.native = self.native.with_runner_env_file(path);
self
}
#[must_use]
pub fn with_dependency_caches(mut self, caches: DependencyCaches) -> Self {
self.native = self.native.with_dependency_caches(caches);
self
}
#[must_use]
pub fn with_docker_cleanup(mut self, cleanup: crate::docker_cleanup::DockerCleanup) -> Self {
self.native.docker_cleanup = Some(cleanup);
self
}
fn isolated_provider(&self, backend: Backend) -> Option<&dyn ExecutionProvider> {
match backend {
Backend::Auto => {
#[cfg(target_os = "windows")]
{
Some(&self.windows_hyper_v)
}
#[cfg(not(target_os = "windows"))]
{
Some(&self.oci)
}
}
Backend::Oci => Some(&self.oci),
Backend::WindowsHyperVContainer => Some(&self.windows_hyper_v),
Backend::VirtualMachine => None,
}
}
fn provider_for_attempt(&self, attempt: &RunnerAttempt) -> Option<&dyn ExecutionProvider> {
match attempt.execution() {
AttemptExecution::Native { .. } => Some(&self.native),
AttemptExecution::Isolated { provider_kind, .. } => {
self.isolated_provider(*provider_kind)
}
}
}
fn required_provider_for_attempt(
&self,
attempt: &RunnerAttempt,
) -> Result<&dyn ExecutionProvider, FailureReason> {
self.provider_for_attempt(attempt)
.ok_or_else(|| FailureReason::Other("execution provider unavailable".into()))
}
}
impl ExecutionProvider for PlatformExecutionProvider {
fn probe(&self, policy: &ScalePolicy) -> ProviderCapability {
match policy.execution_policy() {
runner_manager_domain::execution::ExecutionPolicy::Native => self.native.probe(policy),
runner_manager_domain::execution::ExecutionPolicy::Isolated { backend, .. } => self
.isolated_provider(*backend)
.map_or(ProviderCapability::Unsupported, |provider| {
provider.probe(policy)
}),
}
}
fn resolve(&self, policy: &ScalePolicy) -> Result<Option<ResolvedEnvironment>, FailureReason> {
match policy.execution_policy() {
runner_manager_domain::execution::ExecutionPolicy::Native => {
self.native.resolve(policy)
}
runner_manager_domain::execution::ExecutionPolicy::Isolated { backend, .. } => self
.isolated_provider(*backend)
.ok_or_else(|| FailureReason::Other("execution provider unavailable".into()))?
.resolve(policy),
}
}
fn prepare(
&self,
attempt: &RunnerAttempt,
policy: &ScalePolicy,
) -> Result<PreparedEnvironment, FailureReason> {
match policy.execution_policy() {
runner_manager_domain::execution::ExecutionPolicy::Native => {
self.native.prepare(attempt, policy)
}
runner_manager_domain::execution::ExecutionPolicy::Isolated { backend, .. } => self
.isolated_provider(*backend)
.ok_or_else(|| FailureReason::Other("execution provider unavailable".into()))?
.prepare(attempt, policy),
}
}
fn start(
&self,
prepared: PreparedEnvironment,
attempt: &RunnerAttempt,
handoff: OneTimeJitHandoff<'_>,
) -> Result<EnvironmentIdentity, ProcessStartFailure> {
self.provider_for_attempt(attempt)
.ok_or_else(|| {
ProcessStartFailure::before_spawn(FailureReason::Other(
"execution provider unavailable".into(),
))
})?
.start(prepared, attempt, handoff)
}
fn inspect(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
self.required_provider_for_attempt(attempt)?
.inspect(attempt)
}
fn stop(&self, attempt: &RunnerAttempt) -> Result<ProviderStop, FailureReason> {
self.required_provider_for_attempt(attempt)?.stop(attempt)
}
fn destroy(&self, attempt: &RunnerAttempt) -> Result<ProviderDestroy, FailureReason> {
self.required_provider_for_attempt(attempt)?
.destroy(attempt)
}
fn recover(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
self.required_provider_for_attempt(attempt)?
.recover(attempt)
}
fn enumerate_owned(&self, host_id: HostId) -> Vec<EnvironmentIdentity> {
let mut owned = self.oci.enumerate_owned(host_id);
owned.extend(self.windows_hyper_v.enumerate_owned(host_id));
owned
}
fn diagnostics(&self, attempt: &RunnerAttempt) -> Vec<ProviderDiagnostic> {
self.provider_for_attempt(attempt)
.map_or_else(Vec::new, |provider| provider.diagnostics(attempt))
}
fn owns(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
self.required_provider_for_attempt(attempt)?.owns(attempt)
}
fn spawn(
&self,
attempt: &RunnerAttempt,
config: &EncodedJitConfig,
) -> Result<u32, ProcessStartFailure> {
if !attempt.execution().is_native() {
return Err(ProcessStartFailure::before_spawn(FailureReason::Other(
"native process launch refused for an isolated allocation".into(),
)));
}
self.native.spawn(attempt, config)
}
fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
self.required_provider_for_attempt(attempt)?
.is_alive(attempt)
}
fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
self.required_provider_for_attempt(attempt)?
.recovered_pid(attempt)
}
fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool {
self.provider_for_attempt(attempt)
.is_some_and(|provider| provider.completed_successfully(attempt))
}
fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
self.required_provider_for_attempt(attempt)?
.record_terminate_intent(attempt)
}
fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool {
self.provider_for_attempt(attempt)
.is_some_and(|provider| provider.has_terminate_intent(attempt))
}
fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
self.required_provider_for_attempt(attempt)?
.terminate(attempt)
}
}
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PostSpawnBoundary {
HandoffDelete,
IdentitySerialize,
IdentityWrite,
ChildMapInsert,
}
impl NativeProcesses {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn with_runner_env_file(mut self, path: PathBuf) -> Self {
self.runner_env_file = Some(path);
self
}
#[must_use]
pub fn with_dependency_caches(mut self, caches: DependencyCaches) -> Self {
self.dependency_caches = Some(caches);
self
}
fn cache_env(
&self,
attempt: &RunnerAttempt,
target: Option<&runner_manager_domain::model::ScaleTarget>,
host_env: &RunnerEnv,
) -> Vec<(&'static str, std::ffi::OsString)> {
match (&self.dependency_caches, target) {
(Some(caches), Some(target)) => {
caches.for_launch(target, attempt.runtime_path(), host_env)
}
_ => Vec::new(),
}
}
fn spawn_listener(
&self,
attempt: &RunnerAttempt,
config: &EncodedJitConfig,
cache_target: Option<&runner_manager_domain::model::ScaleTarget>,
) -> Result<u32, ProcessStartFailure> {
let handoff = RestrictiveHandoff::create(
attempt.runtime_path(),
SecretString::from(config.expose().to_owned()),
)
.map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
#[cfg(windows)]
let program = attempt
.runtime_path()
.join("bin")
.join("Runner.Listener.exe");
#[cfg(not(windows))]
let program = attempt.runtime_path().join("bin").join("Runner.Listener");
if !program.is_file() {
return Err(ProcessStartFailure::before_spawn(
FailureReason::ProcessStartFailed,
));
}
let host_env = self.host_env().map_err(ProcessStartFailure::before_spawn)?;
let caches = self.cache_env(attempt, cache_target, &host_env);
#[cfg(test)]
let spec = if self
.use_long_lived_test_listener
.load(std::sync::atomic::Ordering::SeqCst)
{
SpawnSpec::new(program)
.args([
"--ignored",
"--exact",
"lifecycle::tests::long_lived_native_listener_helper",
"--nocapture",
])
.env(
"RUNNER_MANAGER_TEST_LISTENER_READY",
attempt.runtime_path().join(TEST_LISTENER_READY),
)
.working_dir(attempt.runtime_path())
} else {
runner_listener_spec(program, attempt.runtime_path(), caches, &host_env)
};
#[cfg(not(test))]
let spec = runner_listener_spec(program, attempt.runtime_path(), caches, &host_env);
let (spec, docker_proxy) = if let Some(cleanup) = &self.docker_cleanup {
let repository = cache_target.map(ToString::to_string).unwrap_or_default();
match cleanup.launch_environment(attempt.id, &repository, &host_env) {
Ok(Some(endpoint)) => (
spec.env("DOCKER_HOST", endpoint),
Some(cleanup.root.join(attempt.id.to_string())),
),
Ok(None) => (spec, None),
Err(error) => {
tracing::warn!(attempt_id = %attempt.id, event = "docker_accounting_unavailable", reason = ?error.kind(), "native Docker accounting could not start");
if crate::docker_cleanup::Config::load(&cleanup.config).map_or(true, |config| {
config.mode == crate::docker_cleanup::Mode::Apply
}) {
return Err(ProcessStartFailure::before_spawn(
FailureReason::ProcessStartFailed,
));
}
(spec, None)
}
}
} else {
(spec, None)
};
let child = spec
.spawn_runner_with_handoff(&handoff)
.map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
#[cfg(test)]
if self.faults_at(PostSpawnBoundary::HandoffDelete) {
drop(handoff);
return Err(self.abort_spawned_child(child, attempt, false));
}
if handoff.delete().is_err() {
return Err(self.abort_spawned_child(child, attempt, false));
}
#[cfg(test)]
if self.faults_at(PostSpawnBoundary::IdentitySerialize) {
return Err(self.abort_spawned_child(child, attempt, false));
}
let identity = match serde_json::to_vec(child.identity()) {
Ok(identity) => identity,
Err(_) => {
return Err(self.abort_spawned_child(child, attempt, false));
}
};
if self.persist_identity(attempt, &identity).is_err() {
return Err(self.abort_spawned_child(child, attempt, true));
}
if let Some(directory) = docker_proxy
&& crate::docker_cleanup::write_json(&directory.join("runner.json"), child.identity())
.is_err()
{
return Err(self.abort_spawned_child(child, attempt, true));
}
let pid = child.pid();
#[cfg(test)]
if self.faults_at(PostSpawnBoundary::ChildMapInsert) {
return Err(self.abort_spawned_child(child, attempt, true));
}
let mut children = self
.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
children.insert(attempt.id, child);
Ok(pid)
}
fn host_env(&self) -> Result<RunnerEnv, FailureReason> {
let Some(path) = &self.runner_env_file else {
return Ok(RunnerEnv::default());
};
RunnerEnv::load(path).map_err(|error| {
tracing::warn!(
reason = "runner_env_file_invalid",
"runner.env could not be applied, so no runner starts until it is fixed: {error}"
);
FailureReason::Other(format!(
"runner.env cannot be applied ({error}); fix it with `runner-manager host env`"
))
})
}
#[cfg(test)]
fn fail_post_spawn_at(&self, boundary: PostSpawnBoundary) {
self.post_spawn_faults.lock().unwrap().push_back(boundary);
}
#[cfg(test)]
fn faults_at(&self, boundary: PostSpawnBoundary) -> bool {
let mut faults = self.post_spawn_faults.lock().unwrap();
if faults.front() == Some(&boundary) {
faults.pop_front();
true
} else {
false
}
}
#[cfg(test)]
fn fail_post_spawn_stops(&self, count: usize) {
self.post_spawn_stop_failures
.fetch_add(count, std::sync::atomic::Ordering::SeqCst);
}
#[cfg(test)]
fn fail_next_post_spawn_stop(&self) {
self.fail_post_spawn_stops(1);
}
#[cfg(test)]
fn use_long_lived_test_listener(&self) {
self.use_long_lived_test_listener
.store(true, std::sync::atomic::Ordering::SeqCst);
}
fn stop_spawned_child(&self, child: &mut ChildProcess) -> Result<(), FailureReason> {
#[cfg(test)]
if self
.post_spawn_stop_failures
.fetch_update(
std::sync::atomic::Ordering::SeqCst,
std::sync::atomic::Ordering::SeqCst,
|left| if left > 0 { Some(left - 1) } else { None },
)
.is_ok()
{
return Err(FailureReason::Other("injected runner stop failure".into()));
}
child
.stop(Duration::from_secs(1))
.map(|_| ())
.map_err(|_| FailureReason::Other("spawned runner process could not be stopped".into()))
}
fn abort_spawned_child(
&self,
mut child: ChildProcess,
attempt: &RunnerAttempt,
remove_identity: bool,
) -> ProcessStartFailure {
let mut reaped = self.stop_spawned_child(&mut child).is_ok();
if reaped {
if remove_identity {
Self::remove_identity_files(attempt);
}
} else {
let identity_durable =
serde_json::to_vec(child.identity())
.ok()
.is_some_and(|identity| {
self.persist_identity(attempt, &identity).is_ok()
|| self.persist_fallback_identity(attempt, &identity).is_ok()
});
if !identity_durable {
for _ in 1..MAX_POST_SPAWN_STOP_ATTEMPTS {
if self.stop_spawned_child(&mut child).is_ok() {
reaped = true;
break;
}
}
if !reaped {
let pid = child.pid();
let marker = write_durable_file(
&Self::unresolved_process_path(attempt),
pid.to_string().as_bytes(),
);
self.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(attempt.id, child);
let reason = if marker.is_ok() {
FailureReason::Other(
"spawn cleanup exhausted its bounded stop attempts; the live process remains under durable unresolved supervision"
.into(),
)
} else {
FailureReason::Other(
"spawn cleanup exhausted its bounded stop attempts and the unresolved-process marker could not be journalled"
.into(),
)
};
return ProcessStartFailure::after_spawn_live_with_reason(pid, reason);
}
} else {
let pid = child.pid();
self.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(attempt.id, child);
return ProcessStartFailure::after_spawn_live(pid);
}
}
#[cfg(test)]
if reaped {
self.post_spawn_reaps
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
#[cfg(not(test))]
let _ = reaped;
ProcessStartFailure::after_spawn_stopped()
}
fn identity_path(attempt: &RunnerAttempt) -> PathBuf {
attempt.runtime_path().join(IDENTITY_FILE)
}
fn fallback_identity_path(attempt: &RunnerAttempt) -> PathBuf {
attempt.runtime_path().join(FALLBACK_IDENTITY_FILE)
}
fn unresolved_process_path(attempt: &RunnerAttempt) -> PathBuf {
attempt.runtime_path().join(UNRESOLVED_PROCESS_FILE)
}
fn remove_identity_files(attempt: &RunnerAttempt) {
let _ = fs::remove_file(Self::identity_path(attempt));
let _ = fs::remove_file(Self::fallback_identity_path(attempt));
let _ = fs::remove_file(Self::unresolved_process_path(attempt));
}
fn persist_identity(&self, attempt: &RunnerAttempt, bytes: &[u8]) -> std::io::Result<()> {
self.persist_identity_at(&Self::identity_path(attempt), bytes)
}
fn persist_fallback_identity(
&self,
attempt: &RunnerAttempt,
bytes: &[u8],
) -> std::io::Result<()> {
self.persist_identity_at(&Self::fallback_identity_path(attempt), bytes)
}
fn persist_identity_at(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
#[cfg(test)]
if self.faults_at(PostSpawnBoundary::IdentityWrite) {
return Err(std::io::Error::other("injected identity write failure"));
}
write_durable_file(path, bytes)
}
fn intent_path(attempt: &RunnerAttempt) -> PathBuf {
attempt.runtime_path().join(TERMINATE_INTENT_FILE)
}
fn read_identity(attempt: &RunnerAttempt) -> Result<Option<ProcessIdentity>, FailureReason> {
match Self::read_identity_at(&Self::identity_path(attempt))? {
Some(identity) => Ok(Some(identity)),
None => Self::read_identity_at(&Self::fallback_identity_path(attempt)),
}
}
fn read_identity_at(path: &Path) -> Result<Option<ProcessIdentity>, FailureReason> {
match fs::read(path) {
Ok(bytes) => serde_json::from_slice(&bytes)
.map(Some)
.map_err(|_| FailureReason::Other("process identity journal is unreadable".into())),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(_) => Err(FailureReason::Other(
"process identity journal could not be read".into(),
)),
}
}
}
impl ProcessSupervisor for NativeProcesses {
fn destroy(&self, attempt: &RunnerAttempt) -> Result<ProviderDestroy, FailureReason> {
if !attempt.execution().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
if let Some(caches) = &self.dependency_caches {
caches.capture_git(attempt.runtime_path());
}
Ok(ProviderDestroy::Destroyed)
}
fn prepare(
&self,
attempt: &RunnerAttempt,
policy: &ScalePolicy,
) -> Result<PreparedEnvironment, FailureReason> {
if !attempt.execution().is_native() || !policy.execution_policy().is_native() {
return Err(FailureReason::Other(
"execution provider unavailable".into(),
));
}
self.host_env()?;
let cache_target = match &self.dependency_caches {
Some(caches) => {
caches.config()?;
Some(policy.target.clone())
}
None => None,
};
Ok(PreparedEnvironment {
attempt: attempt.id,
identity: None,
cache_target,
})
}
fn start(
&self,
prepared: PreparedEnvironment,
attempt: &RunnerAttempt,
handoff: OneTimeJitHandoff<'_>,
) -> Result<EnvironmentIdentity, ProcessStartFailure> {
if prepared.attempt != attempt.id || !attempt.execution().is_native() {
return Err(ProcessStartFailure::before_spawn(FailureReason::Other(
"execution provider identity mismatch".into(),
)));
}
self.spawn_listener(attempt, handoff.consume(), prepared.cache_target.as_ref())
.map(EnvironmentIdentity::NativeProcess)
}
fn spawn(
&self,
attempt: &RunnerAttempt,
config: &EncodedJitConfig,
) -> Result<u32, ProcessStartFailure> {
self.spawn_listener(attempt, config, None)
}
fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
let mut children = self
.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(child) = children.get_mut(&attempt.id) {
return match child
.try_exit_status()
.map_err(|_| FailureReason::Other("runner process could not be observed".into()))?
{
None => Ok(true),
Some(status) => {
if let Ok(mut exits) = self.successful_exits.lock() {
exits.insert(attempt.id, status.success());
}
Ok(false)
}
};
}
let Some(identity) = Self::read_identity(attempt)? else {
if attempt.process_id().is_some() || Self::unresolved_process_path(attempt).is_file() {
return Err(FailureReason::Other(
"runner process identity is missing; refusing recovery until the process is resolved"
.into(),
));
}
return Ok(false);
};
match identity.recheck() {
Ok(Adoption::Live) => Ok(true),
Ok(Adoption::Gone | Adoption::PidRecycled { .. }) => Ok(false),
Err(_) => Ok(false),
}
}
fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
Ok(Self::read_identity(attempt)?.map(|identity| identity.pid()))
}
fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool {
self.successful_exits
.lock()
.ok()
.and_then(|exits| exits.get(&attempt.id).copied())
.unwrap_or(false)
}
fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
let path = Self::intent_path(attempt);
write_durable_file(&path, b"registration-timeout\n")
.map_err(|_| FailureReason::Other("terminate intent could not be journalled".into()))
}
fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool {
Self::intent_path(attempt).is_file()
}
fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
let mut children = self
.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(child) = children.get_mut(&attempt.id) {
child
.stop(Duration::from_secs(10))
.map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?;
return Ok(());
}
let Some(identity) = Self::read_identity(attempt)? else {
return Ok(());
};
match identity
.terminate(Duration::from_secs(10))
.map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?
{
Termination::Terminated | Termination::AlreadyGone => Ok(()),
Termination::RefusedPidRecycled { .. } => Err(FailureReason::Other(
"runner PID was recycled; refusing to signal it".into(),
)),
}
}
}
pub struct LifecyclePorts {
pub store: Arc<dyn Store>,
pub github: Arc<dyn LifecycleGithub>,
pub packages: Arc<dyn RuntimePackages>,
pub processes: Arc<dyn ExecutionProvider>,
pub clock: Arc<dyn Clock>,
pub demand: Arc<dyn DemandPersistence>,
pub delay: Arc<dyn RetryDelay>,
pub events: Arc<dyn AttemptEventSink>,
pub reconcile_events: Arc<dyn EventSink>,
}
impl fmt::Debug for LifecyclePorts {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("LifecyclePorts")
.field("store", &self.store)
.field("github", &self.github)
.field("packages", &self.packages)
.field("processes", &self.processes)
.finish_non_exhaustive()
}
}
#[derive(Debug, thiserror::Error)]
pub enum LifecycleError {
#[error("attempt journal operation failed")]
Journal,
#[error("attempt {0} is not in the journal")]
Missing(AttemptId),
#[error("attempt lifecycle transition was refused")]
Transition,
#[error("startup recovery has not completed")]
RecoveryIncomplete,
#[error("the persistent slot was not cleaned: {detail}")]
SlotQuarantined {
class: &'static str,
detail: String,
},
#[error("{}", FailureReason::WorkspaceCleanupDeferred)]
WorkspaceCleanupDeferred,
#[error("isolated environment cleanup is deferred")]
EnvironmentCleanupDeferred,
#[error("runner lifecycle failed: {0}")]
Failed(FailureReason),
}
impl LifecycleError {
fn reason(&self) -> FailureReason {
match self {
Self::Failed(reason) => reason.clone(),
Self::RecoveryIncomplete => FailureReason::Other("startup recovery incomplete".into()),
Self::Journal => FailureReason::Other("attempt journal operation failed".into()),
Self::Missing(_) => FailureReason::Other("attempt disappeared from the journal".into()),
Self::Transition => FailureReason::Other("attempt transition was refused".into()),
Self::SlotQuarantined { .. } => FailureReason::Other(self.to_string()),
Self::WorkspaceCleanupDeferred => FailureReason::WorkspaceCleanupDeferred,
Self::EnvironmentCleanupDeferred => FailureReason::Other(self.to_string()),
}
}
}
#[derive(Debug)]
pub struct LifecycleLauncher {
host_id: HostId,
app_paths: runner_manager_platform::paths::AppPaths,
diagnostics_root: PathBuf,
runner_group_id: u64,
timeouts: RecoveryTimeouts,
retry: RetryPolicy,
cancel: CancelToken,
ports: LifecyclePorts,
recovery_complete: Mutex<bool>,
versions: Mutex<BTreeMap<AttemptId, RunnerVersion>>,
pending_replacements: Mutex<BTreeMap<AttemptId, ReplacementIntent>>,
cleanup_retries: Mutex<CleanBackoff>,
turns: Arc<tokio::sync::Semaphore>,
bringing_up: Mutex<BTreeSet<AttemptId>>,
}
struct BroughtUpOnDrop<'a> {
attempts: &'a Mutex<BTreeSet<AttemptId>>,
id: AttemptId,
}
impl Drop for BroughtUpOnDrop<'_> {
fn drop(&mut self) {
self.attempts
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(&self.id);
}
}
#[must_use]
pub fn preparation_turns() -> Arc<tokio::sync::Semaphore> {
let cores = std::thread::available_parallelism().map_or(1, std::num::NonZero::get);
Arc::new(tokio::sync::Semaphore::new((cores / 4).max(2)))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReconcileProgress {
Reconciled,
Deferred,
Replacement {
attempt: AttemptId,
operation: &'static str,
},
}
impl LifecycleLauncher {
#[must_use]
pub fn new(
host_id: HostId,
app_paths: runner_manager_platform::paths::AppPaths,
diagnostics_root: impl Into<PathBuf>,
runner_group_id: u64,
timeouts: RecoveryTimeouts,
retry: RetryPolicy,
ports: LifecyclePorts,
) -> Self {
Self {
host_id,
app_paths,
diagnostics_root: diagnostics_root.into(),
runner_group_id,
timeouts,
retry,
cancel: CancelToken::new(),
ports,
recovery_complete: Mutex::new(false),
versions: Mutex::new(BTreeMap::new()),
pending_replacements: Mutex::new(BTreeMap::new()),
cleanup_retries: Mutex::new(CleanBackoff::default()),
turns: preparation_turns(),
bringing_up: Mutex::default(),
}
}
#[must_use]
pub fn with_preparation_turns(mut self, turns: Arc<tokio::sync::Semaphore>) -> Self {
self.turns = turns;
self
}
pub async fn recover_startup(
&self,
policies: &[ScalePolicy],
) -> Result<Vec<ReplacementIntent>, LifecycleError> {
*self
.recovery_complete
.lock()
.map_err(|_| LifecycleError::Journal)? = false;
let by_id: BTreeMap<_, _> = policies.iter().map(|policy| (policy.id, policy)).collect();
let attempts = self
.ports
.store
.attempts()
.map_err(|_| LifecycleError::Journal)?;
let owned = self.ports.processes.enumerate_owned(self.host_id);
for identity in &owned {
if let EnvironmentIdentity::Isolated { attempt: id, .. } = identity
&& !attempts.iter().any(|attempt| {
attempt.state() != AttemptState::Cleaned
&& identity_matches_intent(self.host_id, attempt, identity)
})
{
self.ports
.events
.emit(AttemptEvent::QuarantinedOwnedResource { attempt: *id });
tracing::warn!(host_id = %self.host_id, attempt_id = %id, reason = "owned_isolated_resource_without_matching_journal", "provider resource quarantined for operator review");
}
}
let mut unresolved = false;
for mut attempt in attempts {
let Some(policy) = by_id.get(&attempt.policy_id) else {
if attempt.counts_against_capacity() {
unresolved = true;
}
continue;
};
authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
let allocation_resources = owned
.iter()
.filter(|identity| identity_same_allocation(self.host_id, &attempt, identity))
.collect::<Vec<_>>();
if allocation_resources.len() > 1 {
self.ports
.events
.emit(AttemptEvent::QuarantinedOwnedResource {
attempt: attempt.id,
});
tracing::warn!(attempt_id = %attempt.id, reason = "duplicate_isolated_resource_ownership", "provider resources quarantined for operator review");
}
if attempt.state() != AttemptState::Cleaned
&& matches!(
attempt.execution(),
AttemptExecution::Isolated {
environment_id: None,
..
}
)
&& allocation_resources.len() == 1
&& let Some(EnvironmentIdentity::Isolated { environment_id, .. }) =
allocation_resources.first().copied()
{
attempt
.prepared_environment(environment_id.clone())
.map_err(|_| LifecycleError::Transition)?;
if attempt.state() == AttemptState::Preparing {
attempt
.mark_prepared(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
}
self.record(&attempt)?;
}
match self.reconcile_one(policy, attempt).await? {
ReconcileProgress::Deferred => {
if policy.execution_policy().is_native() {
unresolved = true;
}
}
ReconcileProgress::Replacement { attempt, operation } => {
self.pending_replacements
.lock()
.map_err(|_| LifecycleError::Journal)?
.insert(
attempt,
ReplacementIntent {
policy: policy.id,
previous_attempt: attempt,
operation,
},
);
}
ReconcileProgress::Reconciled => {}
}
}
if unresolved {
return Err(LifecycleError::RecoveryIncomplete);
}
*self
.recovery_complete
.lock()
.map_err(|_| LifecycleError::Journal)? = true;
Ok(self
.pending_replacements
.lock()
.map_err(|_| LifecycleError::Journal)?
.values()
.copied()
.collect())
}
pub async fn supervise(
&self,
policy: &ScalePolicy,
) -> Result<Vec<ReplacementIntent>, LifecycleError> {
let mut replacements = Vec::new();
self.pending_replacements
.lock()
.map_err(|_| LifecycleError::Journal)?
.retain(|_, intent| {
if intent.policy == policy.id {
replacements.push(*intent);
false
} else {
true
}
});
let attempts = {
let bringing_up = self
.bringing_up
.lock()
.map_err(|_| LifecycleError::Journal)?;
let mut attempts = self
.ports
.store
.attempts_for_policy(policy.id)
.map_err(|_| LifecycleError::Journal)?;
attempts.retain(|attempt| !bringing_up.contains(&attempt.id));
attempts
};
for attempt in attempts {
authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
if let ReconcileProgress::Replacement { attempt, operation } =
self.reconcile_one(policy, attempt).await?
{
replacements.push(ReplacementIntent {
policy: policy.id,
previous_attempt: attempt,
operation,
});
}
}
Ok(replacements)
}
async fn reconcile_one(
&self,
policy: &ScalePolicy,
mut attempt: RunnerAttempt,
) -> Result<ReconcileProgress, LifecycleError> {
if !attempt.execution().is_native() {
return self.reconcile_isolated_one(policy, attempt).await;
}
if attempt.state() == AttemptState::Cleaned {
if matches!(attempt.workspace(), AttemptWorkspace::Ephemeral)
&& attempt.runtime_path().exists()
{
let now = self.ports.clock.now();
if !self.cleanup_retries().is_due(attempt.id, now) {
return Ok(ReconcileProgress::Reconciled);
}
match self.scrub_workspace(&attempt) {
Ok(()) => self.cleanup_retries().succeeded(attempt.id),
Err(LifecycleError::WorkspaceCleanupDeferred) => {
self.cleanup_retries().failed(attempt.id, now);
self.ports
.reconcile_events
.emit(LifecycleEvent::AttemptCleanFailed {
policy: attempt.policy_id,
attempt: attempt.id,
reason: "late_ephemeral_workspace_could_not_be_removed",
});
}
Err(error) => return Err(error),
}
}
return Ok(ReconcileProgress::Reconciled);
}
if attempt.is_terminal() {
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Reconciled);
}
let process_alive = self
.ports
.processes
.inspect(&attempt)
.map_err(LifecycleError::Failed)?
== EnvironmentState::Running;
let github = self
.ports
.github
.observe(&policy.target, attempt.id, &self.cancel)
.await;
if let Some(runner_id) = github.runner_id
&& read_runner_id(attempt.runtime_path()).is_none()
{
write_runner_id(attempt.runtime_path(), runner_id)?;
self.ports
.events
.emit(AttemptEvent::RemoteIdentityRecovered {
attempt: attempt.id,
runner_id,
});
}
if attempt.state() == AttemptState::JitReceived
&& process_alive
&& let Some(pid) = self
.ports
.processes
.recovered_pid(&attempt)
.map_err(LifecycleError::Failed)?
{
attempt
.started(pid, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
}
if !process_alive
&& github.status != GithubRunnerObservation::Unreachable
&& matches!(
attempt.state(),
AttemptState::Starting | AttemptState::Idle | AttemptState::Busy
)
&& self.ports.processes.completed_successfully(&attempt)
{
let state = attempt.state();
let outcome = match (state, JobEvidence::of(attempt.runtime_path())) {
(AttemptState::Starting | AttemptState::Idle, JobEvidence::RanAJob) => {
known_runner_id(&attempt, github.runner_id)
.map(|runner_id| (Some(runner_id), AttemptOutcome::CompletedJob))
}
(AttemptState::Busy, JobEvidence::RanAJob) => {
Some((None, AttemptOutcome::CompletedJob))
}
(AttemptState::Busy, JobEvidence::NoJob) => {
Some((None, AttemptOutcome::ExitedIdleWithoutWork))
}
(AttemptState::Busy, JobEvidence::Unknown)
if github.status == GithubRunnerObservation::NotRegistered =>
{
Some((None, AttemptOutcome::CompletedJob))
}
_ => None,
};
if let Some((walk_to_busy, outcome)) = outcome {
if matches!(github.status, GithubRunnerObservation::Registered { .. }) {
self.deregister_runner(policy, &attempt).await;
}
let now = self.ports.clock.now();
match (walk_to_busy, &outcome) {
(Some(runner_id), _) => {
self.transition(&mut attempt, |a| a.assigned_job(runner_id, now))?;
}
(None, AttemptOutcome::ExitedIdleWithoutWork) => {
self.transition(&mut attempt, |a| a.assignment_not_taken(now))?;
}
(None, _) => {}
}
self.conclude(&mut attempt, outcome)?;
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Reconciled);
}
}
if process_alive
&& attempt.state() == AttemptState::Busy
&& github.status == (GithubRunnerObservation::Registered { busy: false })
&& JobEvidence::of(attempt.runtime_path()) == JobEvidence::NoJob
{
let now = self.ports.clock.now();
self.transition(&mut attempt, |a| a.assignment_not_taken(now))?;
return Ok(ReconcileProgress::Reconciled);
}
if self.ports.processes.has_terminate_intent(&attempt) && !process_alive {
self.deregister_runner(policy, &attempt).await;
self.conclude(
&mut attempt,
AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout),
)?;
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Replacement {
attempt: attempt.id,
operation: "registration_timeout_replacement",
});
}
if matches!(
attempt.state(),
AttemptState::Allocated | AttemptState::JitReceived
) && !process_alive
&& matches!(github.status, GithubRunnerObservation::Registered { .. })
{
if attempt.state() == AttemptState::Allocated {
attempt
.jit_received(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
}
self.deregister_runner(policy, &attempt).await;
self.conclude(
&mut attempt,
AttemptOutcome::failed(FailureReason::JitExpired),
)?;
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Replacement {
attempt: attempt.id,
operation: "jit_expired_replacement",
});
}
match recovery_decision(
&attempt,
RecoveryObservation {
process_alive,
github: github.status,
},
self.timeouts,
self.ports.clock.as_ref(),
) {
RecoveryDecision::Nothing | RecoveryDecision::Wait => Ok(ReconcileProgress::Reconciled),
RecoveryDecision::Defer => Ok(ReconcileProgress::Deferred),
RecoveryDecision::Adopt => {
self.ports.events.emit(AttemptEvent::Adopted {
attempt: attempt.id,
});
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Clean => {
self.clean_or_quarantine(&mut attempt)?;
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Observe(state) => {
let runner_id = known_runner_id(&attempt, github.runner_id)
.ok_or(LifecycleError::Transition)?;
match state {
AttemptState::JitReceived => attempt
.jit_received(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?,
AttemptState::Starting => {
let pid = attempt.process_id().ok_or(LifecycleError::Transition)?;
attempt
.started(pid, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
}
AttemptState::Idle => attempt
.registered_idle(runner_id, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?,
AttemptState::Busy => attempt
.assigned_job(runner_id, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?,
_ => return Err(LifecycleError::Transition),
}
self.record(&attempt)?;
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Conclude(outcome) => {
let replacement = replacement_operation(&outcome);
if matches!(github.status, GithubRunnerObservation::Registered { .. }) {
self.deregister_runner(policy, &attempt).await;
}
self.conclude(&mut attempt, outcome)?;
self.clean_or_quarantine(&mut attempt)?;
Ok(
replacement.map_or(ReconcileProgress::Reconciled, |operation| {
ReconcileProgress::Replacement {
attempt: attempt.id,
operation,
}
}),
)
}
RecoveryDecision::Terminate(payload) => {
let idle_exit = payload.is_idle_exit();
self.ports
.processes
.record_terminate_intent(&attempt)
.map_err(LifecycleError::Failed)?;
self.ports.events.emit(AttemptEvent::TerminateIntent {
attempt: attempt.id,
});
if self
.ports
.processes
.stop(&attempt)
.map_err(LifecycleError::Failed)?
== ProviderStop::StillRunning
{
return Ok(ReconcileProgress::Deferred);
}
if self
.ports
.processes
.inspect(&attempt)
.map_err(LifecycleError::Failed)?
== EnvironmentState::Running
{
return Ok(ReconcileProgress::Deferred);
}
self.ports.events.emit(AttemptEvent::Terminated {
attempt: attempt.id,
});
let outcome = if idle_exit {
AttemptOutcome::ExitedIdleWithoutWork
} else {
AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout)
};
self.deregister_runner(policy, &attempt).await;
self.conclude(&mut attempt, outcome)?;
self.clean_or_quarantine(&mut attempt)?;
if idle_exit {
Ok(ReconcileProgress::Reconciled)
} else {
Ok(ReconcileProgress::Replacement {
attempt: attempt.id,
operation: "registration_timeout_replacement",
})
}
}
}
}
async fn reconcile_isolated_one(
&self,
policy: &ScalePolicy,
mut attempt: RunnerAttempt,
) -> Result<ReconcileProgress, LifecycleError> {
if attempt.state() == AttemptState::Cleaned {
return Ok(ReconcileProgress::Reconciled);
}
if attempt.is_terminal() {
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Reconciled);
}
if matches!(
attempt.state(),
AttemptState::Allocated | AttemptState::Preparing | AttemptState::Prepared
) {
self.conclude(
&mut attempt,
AttemptOutcome::failed(FailureReason::Other(
"isolated preparation interrupted".into(),
)),
)?;
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Reconciled);
}
let Ok(provider_state) = self.ports.processes.recover(&attempt) else {
return Ok(ReconcileProgress::Deferred);
};
if provider_state != EnvironmentState::Missing {
let Ok(owned) = self.ports.processes.owns(&attempt) else {
return Ok(ReconcileProgress::Deferred);
};
if !owned {
self.ports
.events
.emit(AttemptEvent::QuarantinedOwnedResource {
attempt: attempt.id,
});
tracing::warn!(attempt_id = %attempt.id, reason = "isolated_resource_ownership_mismatch", "provider resource quarantined for operator review");
return Ok(ReconcileProgress::Deferred);
}
}
let live = matches!(
provider_state,
EnvironmentState::Starting | EnvironmentState::Running
);
if attempt.state() == AttemptState::JitReceived && live {
attempt
.started_isolated(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
}
let github = self
.ports
.github
.observe(&policy.target, attempt.id, &self.cancel)
.await;
if let Some(runner_id) = github.runner_id
&& read_runner_id(attempt.runtime_path()).is_none()
{
write_runner_id(attempt.runtime_path(), runner_id)?;
}
if attempt.state() == AttemptState::Busy
&& !live
&& github.status == GithubRunnerObservation::NotRegistered
&& self.ports.processes.completed_successfully(&attempt)
{
self.conclude(&mut attempt, AttemptOutcome::CompletedJob)?;
self.clean_or_quarantine(&mut attempt)?;
return Ok(ReconcileProgress::Reconciled);
}
match recovery_decision(
&attempt,
RecoveryObservation {
process_alive: live,
github: github.status,
},
self.timeouts,
self.ports.clock.as_ref(),
) {
RecoveryDecision::Nothing | RecoveryDecision::Wait => Ok(ReconcileProgress::Reconciled),
RecoveryDecision::Defer => Ok(ReconcileProgress::Deferred),
RecoveryDecision::Adopt => {
self.ports.events.emit(AttemptEvent::Adopted {
attempt: attempt.id,
});
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Clean => {
self.clean_or_quarantine(&mut attempt)?;
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Observe(state) => {
let runner_id = github
.runner_id
.or_else(|| read_runner_id(attempt.runtime_path()))
.ok_or(LifecycleError::Transition)?;
match state {
AttemptState::Idle => attempt
.registered_idle(runner_id, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?,
AttemptState::Busy => attempt
.assigned_job(runner_id, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?,
_ => return Err(LifecycleError::Transition),
}
self.record(&attempt)?;
Ok(ReconcileProgress::Reconciled)
}
RecoveryDecision::Conclude(outcome) | RecoveryDecision::Terminate(outcome) => {
let replacement = replacement_operation(&outcome);
if matches!(github.status, GithubRunnerObservation::Registered { .. }) {
self.deregister_runner(policy, &attempt).await;
}
self.conclude(&mut attempt, outcome)?;
self.clean_or_quarantine(&mut attempt)?;
Ok(
replacement.map_or(ReconcileProgress::Reconciled, |operation| {
ReconcileProgress::Replacement {
attempt: attempt.id,
operation,
}
}),
)
}
}
}
fn transition(
&self,
attempt: &mut RunnerAttempt,
edge: impl FnOnce(
&mut RunnerAttempt,
) -> Result<(), runner_manager_domain::attempt::AttemptError>,
) -> Result<(), LifecycleError> {
edge(attempt).map_err(|_| LifecycleError::Transition)?;
self.record(attempt)
}
fn record(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
self.ports
.store
.record_attempt(attempt)
.map_err(|_| LifecycleError::Journal)?;
self.ports.events.emit(AttemptEvent::State {
attempt: attempt.id,
state: attempt.state(),
});
Ok(())
}
async fn deregister_runner(&self, policy: &ScalePolicy, attempt: &RunnerAttempt) {
let Some(runner_id) = attempt
.github_runner_id()
.or_else(|| read_runner_id(attempt.runtime_path()))
else {
return;
};
if self
.ports
.github
.deregister(&policy.target, runner_id, &self.cancel)
.await
{
self.ports.events.emit(AttemptEvent::Deregistered {
attempt: attempt.id,
runner_id,
});
} else {
tracing::warn!(
attempt = %attempt.id,
runner_id,
"the runner registration could not be removed from GitHub; it will show in the \
target's runner settings until GitHub retires it or a later pass removes it"
);
}
}
fn conclude(
&self,
attempt: &mut RunnerAttempt,
outcome: AttemptOutcome,
) -> Result<(), LifecycleError> {
attempt
.conclude(outcome.clone(), self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(attempt)?;
self.ports.events.emit(AttemptEvent::Concluded {
attempt: attempt.id,
outcome: OutcomeKind::of(&outcome),
});
Ok(())
}
fn clean_or_quarantine(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
let now = self.ports.clock.now();
if !self.cleanup_retries().is_due(attempt.id, now) {
return Ok(());
}
let reason = match self.clean_attempt(attempt) {
Err(LifecycleError::SlotQuarantined { class, .. }) => class,
Err(LifecycleError::WorkspaceCleanupDeferred) => {
failure_reason_kind(&FailureReason::WorkspaceCleanupDeferred)
}
Err(LifecycleError::EnvironmentCleanupDeferred) => {
"isolated_environment_cleanup_deferred"
}
other => {
if other.is_ok() {
self.cleanup_retries().succeeded(attempt.id);
}
return other;
}
};
self.cleanup_retries().failed(attempt.id, now);
self.ports
.reconcile_events
.emit(LifecycleEvent::AttemptCleanFailed {
policy: attempt.policy_id,
attempt: attempt.id,
reason,
});
Ok(())
}
fn cleanup_retries(&self) -> std::sync::MutexGuard<'_, CleanBackoff> {
self.cleanup_retries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn clean_attempt(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
if !attempt.execution().is_native() {
return self.clean_isolated_attempt(attempt);
}
match self
.ports
.processes
.destroy(attempt)
.map_err(LifecycleError::Failed)?
{
ProviderDestroy::Destroyed => {}
ProviderDestroy::Deferred(reason) => {
return Err(LifecycleError::Failed(FailureReason::Other(reason.into())));
}
}
let outcome = attempt
.outcome()
.cloned()
.ok_or(LifecycleError::Transition)?;
self.preserve_diagnostics(attempt, &outcome)?;
self.scrub_workspace(attempt)?;
self.ports
.packages
.release(attempt.id)
.map_err(LifecycleError::Failed)?;
attempt
.clean(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(attempt)?;
let kind = OutcomeKind::of(&outcome);
self.ports.events.emit(AttemptEvent::Cleaned {
attempt: attempt.id,
outcome: kind,
});
self.ports
.reconcile_events
.emit(LifecycleEvent::AttemptCleaned {
policy: attempt.policy_id,
attempt: attempt.id,
outcome: kind,
});
Ok(())
}
fn clean_isolated_attempt(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
if attempt.state() != AttemptState::Destroying {
attempt
.begin_destroy(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(attempt)?;
}
let provider = self.ports.processes.as_ref();
let allocation_resources = provider.enumerate_owned(self.host_id);
let matching_allocation = allocation_resources
.iter()
.filter(|identity| identity_same_allocation(self.host_id, attempt, identity))
.collect::<Vec<_>>();
if matching_allocation.len() > 1
|| matching_allocation
.iter()
.any(|identity| !identity_matches_intent(self.host_id, attempt, identity))
{
self.ports
.events
.emit(AttemptEvent::QuarantinedOwnedResource {
attempt: attempt.id,
});
tracing::warn!(attempt_id = %attempt.id, reason = "duplicate_or_mismatched_isolated_ownership", "provider resources quarantined for operator review");
return Err(self.defer_environment_cleanup(attempt)?);
}
let has_identity = matches!(
attempt.execution(),
AttemptExecution::Isolated {
environment_id: Some(_),
..
}
);
if !has_identity
&& allocation_resources
.iter()
.any(|identity| identity_matches_intent(self.host_id, attempt, identity))
{
return Err(self.defer_environment_cleanup(attempt)?);
}
let current = if has_identity {
provider.inspect(attempt)
} else {
provider.recover(attempt)
};
let current = match current {
Ok(state) => state,
Err(_) => return Err(self.defer_environment_cleanup(attempt)?),
};
if !has_identity && current != EnvironmentState::Missing {
return Err(self.defer_environment_cleanup(attempt)?);
}
if current != EnvironmentState::Missing {
if !matches!(provider.owns(attempt), Ok(true)) {
self.ports
.events
.emit(AttemptEvent::QuarantinedOwnedResource {
attempt: attempt.id,
});
tracing::warn!(attempt_id = %attempt.id, reason = "isolated_resource_ownership_mismatch", "provider resource quarantined for operator review");
return Err(self.defer_environment_cleanup(attempt)?);
}
if matches!(
current,
EnvironmentState::Starting | EnvironmentState::Running
) && !matches!(provider.stop(attempt), Ok(ProviderStop::Stopped))
{
return Err(self.defer_environment_cleanup(attempt)?);
}
if !matches!(provider.destroy(attempt), Ok(ProviderDestroy::Destroyed)) {
return Err(self.defer_environment_cleanup(attempt)?);
}
if !matches!(provider.inspect(attempt), Ok(EnvironmentState::Missing)) {
return Err(self.defer_environment_cleanup(attempt)?);
}
}
if provider
.enumerate_owned(self.host_id)
.iter()
.any(|identity| identity_same_allocation(self.host_id, attempt, identity))
{
return Err(self.defer_environment_cleanup(attempt)?);
}
let outcome = attempt
.outcome()
.cloned()
.ok_or(LifecycleError::Transition)?;
for diagnostic in provider.diagnostics(attempt).into_iter().take(32) {
tracing::debug!(attempt_id = %attempt.id, provider_diagnostic = ?diagnostic, "provider lifecycle diagnostic");
}
self.preserve_diagnostics(attempt, &outcome)?;
self.scrub_workspace(attempt)?;
self.ports
.packages
.release(attempt.id)
.map_err(LifecycleError::Failed)?;
attempt
.clean_isolated(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(attempt)?;
self.ports.events.emit(AttemptEvent::Cleaned {
attempt: attempt.id,
outcome: OutcomeKind::of(&outcome),
});
self.ports
.reconcile_events
.emit(LifecycleEvent::AttemptCleaned {
policy: attempt.policy_id,
attempt: attempt.id,
outcome: OutcomeKind::of(&outcome),
});
Ok(())
}
fn defer_environment_cleanup(
&self,
attempt: &mut RunnerAttempt,
) -> Result<LifecycleError, LifecycleError> {
attempt
.defer_cleanup(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(attempt)?;
Ok(LifecycleError::EnvironmentCleanupDeferred)
}
fn scrub_workspace(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
#[cfg(test)]
{
if matches!(
std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref(),
Ok("skip_workspace_cleanup" | "reuse_job_workspace")
) {
return Ok(());
}
}
match attempt.workspace() {
AttemptWorkspace::Ephemeral => match remove_runtime_tree(attempt.runtime_path()) {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(_) => Err(LifecycleError::WorkspaceCleanupDeferred),
},
AttemptWorkspace::PersistentSlot { slot } => self.scrub_persistent_slot(attempt, slot),
}
}
fn scrub_persistent_slot(
&self,
attempt: &RunnerAttempt,
slot: NonZeroU16,
) -> Result<(), LifecycleError> {
let configured = self
.ports
.store
.policy(attempt.policy_id)
.map_err(|_| LifecycleError::Journal)?
.and_then(|policy| match policy.workspace_policy() {
WorkspacePolicy::Persistent { root } => Some(root.clone()),
WorkspacePolicy::Ephemeral => None,
});
let runtime = attempt.runtime_path();
self.quarantine_on_refusal(
attempt,
verify_journalled_slot(runtime, slot, configured.as_ref())
.and_then(|()| slot_is_present(runtime))
.and_then(|present| {
if present {
scrub_slot_entries(runtime).and_then(|()| verify_slot_scrubbed(runtime))
} else {
Ok(())
}
}),
)
}
fn quarantine_on_refusal(
&self,
attempt: &RunnerAttempt,
outcome: Result<(), SlotQuarantine>,
) -> Result<(), LifecycleError> {
let Err(quarantine) = outcome else {
return Ok(());
};
let detail = quarantine.to_string();
tracing::warn!(
attempt = %attempt.id,
policy = %attempt.policy_id,
slot = attempt.workspace().slot_number(),
refusal = quarantine.refusal.class(),
"{detail}"
);
Err(LifecycleError::SlotQuarantined {
class: quarantine.refusal.class(),
detail,
})
}
fn preserve_diagnostics(
&self,
attempt: &RunnerAttempt,
outcome: &AttemptOutcome,
) -> Result<(), LifecycleError> {
fs::create_dir_all(&self.diagnostics_root).map_err(|_| {
LifecycleError::Failed(FailureReason::Other(
"diagnostics directory could not be created".into(),
))
})?;
let reason_kind = match outcome {
AttemptOutcome::Failed { reason } => failure_reason_kind(reason),
_ => "none",
};
let diagnostic = format!(
"attempt_id={}\npolicy_id={}\noutcome={}\nreason={}\n",
attempt.id,
attempt.policy_id,
OutcomeKind::of(outcome).as_str(),
reason_kind,
);
fs::write(
self.diagnostics_root.join(format!("{}.log", attempt.id)),
diagnostic,
)
.map_err(|_| {
LifecycleError::Failed(FailureReason::Other(
"redacted diagnostics could not be preserved".into(),
))
})
}
async fn materialize_with_retry(
&self,
policy: &ScalePolicy,
attempt: &RunnerAttempt,
timings: &mut PackageTimings,
) -> Result<RunnerVersion, FailureReason> {
let mut issued = 0_u32;
loop {
issued = issued.saturating_add(1);
match self.ports.packages.materialize(attempt, timings).await {
Ok(version) => return Ok(version),
Err(reason)
if package_failure_is_terminal(&reason)
|| issued >= self.retry.max_attempts.max(1) =>
{
return Err(reason);
}
Err(reason) => {
if !self.ports.demand.persists(policy.id).await {
return Err(reason);
}
let delay = self.retry.delay(issued);
self.ports.events.emit(AttemptEvent::Retry {
attempt: attempt.id,
operation: "package_materialization",
delay,
});
self.ports.delay.wait(delay).await;
if !self.ports.demand.persists(policy.id).await {
return Err(reason);
}
}
}
}
}
async fn register_with_retry(
&self,
policy: &ScalePolicy,
attempt: AttemptId,
request: &JitRunnerRequest,
) -> Result<JitRegistration, LifecycleError> {
let mut issued = 0_u32;
loop {
issued = issued.saturating_add(1);
match self
.ports
.github
.register(&policy.target, request, &self.cancel)
.await
{
Ok(registration) => return Ok(registration),
Err(error) if error.terminal => {
return Err(LifecycleError::Failed(error.reason));
}
Err(error) => {
if issued >= self.retry.max_attempts.max(1)
|| !self.ports.demand.persists(policy.id).await
{
return Err(LifecycleError::Failed(error.reason));
}
let delay = error
.retry_after
.unwrap_or_else(|| self.retry.delay(issued));
self.ports.events.emit(AttemptEvent::Retry {
attempt,
operation: "jit_request",
delay,
});
self.ports.delay.wait(delay).await;
if !self.ports.demand.persists(policy.id).await {
return Err(LifecycleError::Failed(error.reason));
}
}
}
}
}
fn allocate_workspace(
&self,
policy: &ScalePolicy,
id: AttemptId,
) -> Result<Placement, LifecycleError> {
let placement = match policy.workspace_policy() {
WorkspacePolicy::Persistent { root } => self.allocate_persistent_slot(policy, root),
WorkspacePolicy::Ephemeral => self.allocate_disposable(policy, id),
};
if placement.is_ok() {
self.root_accepted(policy.id);
}
placement
}
fn configured_host_root(&self) -> Result<Option<LocalAbsolutePath>, LifecycleError> {
let host = self
.ports
.store
.host(self.host_id)
.map_err(|_| LifecycleError::Journal)?
.ok_or_else(|| LifecycleError::Failed(FailureReason::Other("host not found".into())))?;
Ok(host.runner_root_override.clone())
}
fn resolvable_host_root(&self) -> Result<Option<LocalAbsolutePath>, LifecycleError> {
Ok(self
.configured_host_root()?
.or_else(|| default_runner_root(&self.app_paths).ok()))
}
fn effective_host_root(
&self,
policy: &ScalePolicy,
) -> Result<LocalAbsolutePath, LifecycleError> {
match self.configured_host_root()? {
Some(configured) => Ok(configured),
None => default_runner_root(&self.app_paths).map_err(|error| {
self.root_refused(policy.id, "the platform default runner root", error)
}),
}
}
fn root_refused(&self, policy: PolicyId, root: &str, error: RunnerRootError) -> LifecycleError {
let _ = runner_manager_platform::service::record_runner_root_refusal(
&self.app_paths,
&policy.to_string(),
self.ports.clock.now(),
error.kind(),
root,
&error.to_string(),
);
root_failure(error)
}
fn root_accepted(&self, policy: PolicyId) {
let _ = runner_manager_platform::service::clear_runner_root_refusal(
&self.app_paths,
&policy.to_string(),
);
}
fn allocate_disposable(
&self,
policy: &ScalePolicy,
id: AttemptId,
) -> Result<Placement, LifecycleError> {
let effective_root = self.effective_host_root(policy)?;
RootPreflight::new(&self.app_paths)
.check(&RootOwner::Host, &effective_root)
.map_err(|error| self.root_refused(policy.id, effective_root.as_str(), error))?;
let runtime = effective_root.as_path().join({
#[cfg(test)]
{
if std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref()
== Ok("reuse_job_workspace")
{
"mutant-shared-workspace".to_owned()
} else {
workspace_name(id)
}
}
#[cfg(not(test))]
{
workspace_name(id)
}
});
fs::create_dir_all(&runtime)
.map_err(|_| LifecycleError::Failed(FailureReason::ProcessStartFailed))?;
Ok(Placement {
runtime,
workspace: AttemptWorkspace::Ephemeral,
})
}
fn allocate_persistent_slot(
&self,
policy: &ScalePolicy,
root: &LocalAbsolutePath,
) -> Result<Placement, LifecycleError> {
let ceiling = policy.max_capacity().ok_or_else(|| {
LifecycleError::Failed(FailureReason::Other(
"a persistent workspace needs the policy's max_capacity to bound its slots"
.to_string(),
))
})?;
let leases = self
.ports
.store
.slot_leases_for_policy(policy.id)
.map_err(|_| LifecycleError::Journal)?;
let slot = lowest_free_slot(&leases, ceiling).ok_or_else(|| {
LifecycleError::Failed(FailureReason::Other(format!(
"every persistent slot s1 to s{ceiling} for {} is leased by an attempt that has \
not been cleaned, so no slot is free; raise the repository's max capacity, or \
finish cleaning a concluded attempt",
policy.target
)))
})?;
let workspace = AttemptWorkspace::persistent_slot(slot);
let name = workspace
.slot_directory_name()
.expect("a persistent allocation names its slot directory");
let host_root = self.resolvable_host_root()?;
let mut preflight = RootPreflight::new(&self.app_paths);
if let Some(host_root) = host_root {
preflight = preflight.against(RootOwner::Host, host_root);
}
let checked = preflight
.check(&RootOwner::Repository(policy.target.to_string()), root)
.map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
if let Some(leaf) = checked.leaf_to_create() {
fs::create_dir(leaf).map_err(|source| {
LifecycleError::Failed(FailureReason::Other(format!(
"the persistent workspace root {} could not be created: {source}",
leaf.display()
)))
})?;
}
let slot_path = runner_root::derive_child(root, &name)
.map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
create_or_validate_slot(slot_path.as_path())?;
runner_root::verify_containment(root, &slot_path)
.map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
accept_reusable_slot(slot_path.as_path())?;
Ok(Placement {
runtime: slot_path.as_path().to_path_buf(),
workspace,
})
}
fn record_allocation(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
match self.ports.store.record_attempt(attempt) {
Ok(()) => {
self.ports.events.emit(AttemptEvent::State {
attempt: attempt.id,
state: attempt.state(),
});
Ok(())
}
Err(error @ StoreError::SlotAlreadyLeased { .. }) => Err(LifecycleError::Failed(
FailureReason::Other(error.to_string()),
)),
Err(_) => Err(LifecycleError::Journal),
}
}
fn reserve_attempt(
&self,
policy: &ScalePolicy,
allocation_guard: &AllocationGuard,
) -> Result<RunnerAttempt, LifecycleError> {
if let Err(error) = self.prune_under_allocation_lock(allocation_guard) {
tracing::warn!(
event = "package_prune_failed",
reason = failure_reason_kind(&error.reason()),
);
}
self.allocate_attempt(policy)
}
async fn finish_attempt(
&self,
policy: &ScalePolicy,
attempt: RunnerAttempt,
waits: ReservationWaits,
) -> Result<RunnerAttempt, LifecycleError> {
let id = attempt.id;
let _bringing_up = BroughtUpOnDrop {
attempts: &self.bringing_up,
id,
};
let mut stopwatch = LaunchStopwatch::new(waits);
let result = self.launch_steps(policy, attempt, &mut stopwatch).await;
stopwatch.finish();
self.ports.events.emit(AttemptEvent::Launched {
attempt: id,
started: result.is_ok(),
timings: stopwatch.timings,
});
result
}
fn allocate_attempt(&self, policy: &ScalePolicy) -> Result<RunnerAttempt, LifecycleError> {
if !*self
.recovery_complete
.lock()
.map_err(|_| LifecycleError::Journal)?
{
return Err(LifecycleError::RecoveryIncomplete);
}
let capability = self.ports.processes.probe(policy);
if capability != ProviderCapability::Ready {
return Err(LifecycleError::Failed(capability.refusal()));
}
let resolved = self.ports.processes.resolve(policy).map_err(|reason| {
LifecycleError::Failed(if policy.execution_policy().is_native() {
reason
} else {
safe_isolation_reason(reason, IsolationProviderFailure::RuntimeOperationFailed)
})
})?;
if policy.execution_policy().is_native() != resolved.is_none() {
return Err(LifecycleError::Failed(FailureReason::Other(
"provider resolution mismatch".into(),
)));
}
if let Some(resolution) = &resolved {
if resolution.provider_kind == Backend::Auto {
return Err(LifecycleError::Failed(FailureReason::Other(
"provider resolution mismatch".into(),
)));
}
AttemptExecution::Isolated {
provider_kind: resolution.provider_kind,
environment_id: None,
resolved_image: resolution.image.clone(),
generation: "generation-check".into(),
}
.validate()
.map_err(|_| {
LifecycleError::Failed(FailureReason::Other(
"resolved image is incompatible".into(),
))
})?;
}
policy
.routing_labels()
.ok_or(LifecycleError::Failed(FailureReason::JitRequestFailed))?;
let id = AttemptId::new_random();
let placement = self.allocate_workspace(policy, id)?;
let mut attempt = RunnerAttempt::allocate_in(
id,
policy.id,
placement.runtime,
placement.workspace,
self.ports.clock.now(),
);
if let Some(resolution) = &resolved {
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: resolution.provider_kind,
environment_id: None,
resolved_image: resolution.image.clone(),
generation: uuid::Uuid::new_v4().to_string(),
})
.map_err(|_| LifecycleError::Transition)?;
}
self.bringing_up
.lock()
.map_err(|_| LifecycleError::Journal)?
.insert(id);
if let Err(error) = self.record_allocation(&attempt) {
self.bringing_up
.lock()
.map_err(|_| LifecycleError::Journal)?
.remove(&id);
return Err(error);
}
Ok(attempt)
}
async fn launch_steps(
&self,
policy: &ScalePolicy,
mut attempt: RunnerAttempt,
stopwatch: &mut LaunchStopwatch,
) -> Result<RunnerAttempt, LifecycleError> {
let id = attempt.id;
let isolated = !attempt.execution().is_native();
let Some(labels) = policy.routing_labels() else {
return self.fail_launch(&mut attempt, FailureReason::JitRequestFailed);
};
stopwatch.begin(LaunchStep::Turn);
let turn = self.turns.acquire().await.ok();
stopwatch.begin(LaunchStep::Materialize);
let package_timings = &mut stopwatch.timings.package;
let version = match self
.materialize_with_retry(policy, &attempt, package_timings)
.await
{
Ok(version) => version,
Err(reason) => return self.fail_launch(&mut attempt, reason),
};
self.versions
.lock()
.map_err(|_| LifecycleError::Journal)?
.insert(id, version);
if isolated {
attempt
.begin_prepare(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
}
stopwatch.begin(LaunchStep::Prepare);
let preparing = {
let (processes, attempt, policy) = (
Arc::clone(&self.ports.processes),
attempt.clone(),
policy.clone(),
);
tokio::task::spawn_blocking(move || processes.prepare(&attempt, &policy)).await
};
drop(turn);
let prepared = match preparing.unwrap_or(Err(FailureReason::ProcessStartFailed)) {
Ok(prepared) => prepared,
Err(reason) => {
let safe_reason = if isolated {
safe_isolation_reason(reason, IsolationProviderFailure::RuntimeOperationFailed)
} else {
reason
};
return self.fail_launch(&mut attempt, safe_reason);
}
};
if isolated {
let Some(identity) = prepared.identity() else {
return Err(LifecycleError::Failed(FailureReason::IsolationProvider(
IsolationProviderFailure::OwnershipMismatch,
)));
};
if !identity_matches_intent(self.host_id, &attempt, identity) {
return Err(LifecycleError::Failed(FailureReason::IsolationProvider(
IsolationProviderFailure::OwnershipMismatch,
)));
}
let EnvironmentIdentity::Isolated { environment_id, .. } = identity else {
unreachable!()
};
attempt
.prepared_environment(environment_id.clone())
.map_err(|_| LifecycleError::Transition)?;
attempt
.mark_prepared(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
}
stopwatch.begin(LaunchStep::Register);
let jit_request =
JitRunnerRequest::for_policy(runner_name(id), self.runner_group_id, labels);
let registration = match self.register_with_retry(policy, id, &jit_request).await {
Ok(registration) => registration,
Err(error) => return self.fail_launch(&mut attempt, error.reason()),
};
stopwatch.begin(LaunchStep::Spawn);
let runner_id = registration.runner().id;
write_runner_id(attempt.runtime_path(), runner_id)?;
attempt
.jit_received(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
let config = registration.into_config();
if isolated {
let started = self
.ports
.processes
.start(prepared, &attempt, OneTimeJitHandoff::new(&config))
.map_err(|failure| {
LifecycleError::Failed(safe_isolation_reason(
failure.reason,
IsolationProviderFailure::RuntimeOperationFailed,
))
})?;
if !identity_matches_intent(self.host_id, &attempt, &started) {
return Err(LifecycleError::Failed(FailureReason::IsolationProvider(
IsolationProviderFailure::OwnershipMismatch,
)));
}
attempt
.started_isolated(self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
return Ok(attempt);
}
let mut issued = 0_u32;
let pid = loop {
issued = issued.saturating_add(1);
match self.ports.processes.start(
prepared.clone(),
&attempt,
OneTimeJitHandoff::new(&config),
) {
Ok(EnvironmentIdentity::NativeProcess(pid)) => break pid,
Ok(EnvironmentIdentity::Isolated { .. }) => {
return Err(LifecycleError::Failed(FailureReason::Other(
"unexpected isolated environment identity".into(),
)));
}
Err(error) => {
if let Some(pid) = error.live_pid {
attempt
.started(pid, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
return Err(LifecycleError::Failed(error.reason));
}
if !error.retryable
|| issued >= self.retry.max_attempts.max(1)
|| !self.ports.demand.persists(policy.id).await
{
return self.fail_launch(&mut attempt, error.reason);
}
let delay = self.retry.delay(issued);
self.ports.events.emit(AttemptEvent::Retry {
attempt: attempt.id,
operation: "process_start",
delay,
});
self.ports.delay.wait(delay).await;
if !self.ports.demand.persists(policy.id).await {
return self.fail_launch(&mut attempt, error.reason);
}
}
}
};
attempt
.started(pid, self.ports.clock.now())
.map_err(|_| LifecycleError::Transition)?;
self.record(&attempt)?;
Ok(attempt)
}
fn fail_launch<T>(
&self,
attempt: &mut RunnerAttempt,
reason: FailureReason,
) -> Result<T, LifecycleError> {
self.conclude(attempt, AttemptOutcome::failed(reason.clone()))?;
Err(LifecycleError::Failed(reason))
}
fn prune_under_allocation_lock(&self, guard: &AllocationGuard) -> Result<(), LifecycleError> {
let attempts = self
.ports
.store
.attempts()
.map_err(|_| LifecycleError::Journal)?;
self.ports
.packages
.prune_obsolete_guarded(PruneAuthority::from_launch_request(guard), &attempts)
.map_err(LifecycleError::Failed)?;
#[cfg(target_os = "macos")]
self.prune_abandoned_seeds(&attempts);
Ok(())
}
#[cfg(target_os = "macos")]
fn prune_abandoned_seeds(&self, attempts: &[RunnerAttempt]) {
let Ok(policies) = self.ports.store.policies() else {
return;
};
let Ok(Some(host_root)) = self.resolvable_host_root() else {
return;
};
let in_use: BTreeSet<PathBuf> = policies
.iter()
.filter_map(|policy| match policy.workspace_policy() {
WorkspacePolicy::Persistent { root } => Some(canonical_or_raw(root.as_path())),
WorkspacePolicy::Ephemeral => None,
})
.chain(std::iter::once(canonical_or_raw(host_root.as_path())))
.collect();
let used: BTreeSet<&Path> = attempts
.iter()
.filter_map(|attempt| attempt.runtime_path().parent())
.collect();
for root in used.into_iter().map(canonical_or_raw) {
if !in_use.contains(&root) {
let _ = remove_runtime_tree(&root.join(PACKAGE_SEED_DIR));
}
}
}
}
#[cfg(target_os = "macos")]
fn canonical_or_raw(path: &Path) -> PathBuf {
fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf())
}
#[async_trait]
impl RunnerLauncher for LifecycleLauncher {
async fn host_capacity(
&self,
host: &runner_manager_domain::model::Host,
) -> Result<std::num::NonZeroU16, crate::reconcile::HostCapacityError> {
self.ports
.store
.host(host.id)
.map_err(|_| crate::reconcile::HostCapacityError::JournalUnreadable)?
.map(|current| current.host_capacity)
.ok_or(crate::reconcile::HostCapacityError::MissingHost)
}
async fn supervise(
&self,
policy: &ScalePolicy,
) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
LifecycleLauncher::supervise(self, policy)
.await
.map_err(|error| LaunchFailure::new(error.reason()))
}
async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
self.ports.store.attempts().map_err(|_| {
LaunchFailure::new(FailureReason::Other(
"attempt journal could not be read".into(),
))
})
}
async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
self.reserve_attempt(request.policy, request.allocation_guard)
.map_err(|error| LaunchFailure::new(error.reason()))
}
async fn finish_launch(
&self,
policy: &ScalePolicy,
attempt: RunnerAttempt,
waits: ReservationWaits,
) -> Result<RunnerAttempt, LaunchFailure> {
self.finish_attempt(policy, attempt, waits)
.await
.map_err(|error| LaunchFailure::new(error.reason()))
}
async fn clean(&self, id: AttemptId) -> Result<(), LaunchFailure> {
let mut attempt = self
.ports
.store
.attempt(id)
.map_err(|_| {
LaunchFailure::new(FailureReason::Other(
"attempt journal could not be read".into(),
))
})?
.ok_or_else(|| {
LaunchFailure::new(FailureReason::Other(
"attempt disappeared from the journal".into(),
))
})?;
self.clean_attempt(&mut attempt)
.map_err(|error| LaunchFailure::new(error.reason()))?;
self.cleanup_retries().succeeded(id);
Ok(())
}
}
#[must_use]
pub fn runner_name(attempt: AttemptId) -> String {
format!("runner-manager-{attempt}")
}
const EVIDENCE_DEADLINE: Duration = Duration::from_secs(2);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum JobEvidence {
RanAJob,
NoJob,
Unknown,
}
impl JobEvidence {
fn of(runtime: &Path) -> Self {
let diag = runtime.join("_diag");
if runner_manager_platform::host_fitness::directory_responds(&diag, EVIDENCE_DEADLINE)
!= runner_manager_platform::host_fitness::Responsiveness::Responds
{
return Self::Unknown;
}
let Ok(entries) = fs::read_dir(diag) else {
return Self::Unknown;
};
let mut listener = false;
for entry in entries.flatten() {
let name = entry.file_name();
let name = name.to_string_lossy();
if !name.ends_with(".log") {
continue;
}
if name.starts_with("Worker_") {
return Self::RanAJob;
}
listener |= name.starts_with("Runner_");
}
if listener { Self::NoJob } else { Self::Unknown }
}
}
fn known_runner_id(attempt: &RunnerAttempt, observed: Option<u64>) -> Option<u64> {
attempt
.github_runner_id()
.or(observed)
.or_else(|| read_runner_id(attempt.runtime_path()))
}
fn read_runner_id(runtime: &Path) -> Option<u64> {
fs::read_to_string(runtime.join(RUNNER_ID_FILE))
.ok()?
.trim()
.parse()
.ok()
}
fn write_runner_id(runtime: &Path, runner_id: u64) -> Result<(), LifecycleError> {
let target = runtime.join(RUNNER_ID_FILE);
if let Some(existing) = read_runner_id(runtime) {
return (existing == runner_id)
.then_some(())
.ok_or(LifecycleError::Journal);
}
let temporary = runtime.join(format!("{RUNNER_ID_FILE}.{}.tmp", uuid::Uuid::new_v4()));
write_durable_file(&temporary, runner_id.to_string().as_bytes())
.map_err(|_| LifecycleError::Journal)?;
match fs::rename(&temporary, &target) {
Ok(()) => sync_directory(runtime).map_err(|_| LifecycleError::Journal),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
let _ = fs::remove_file(&temporary);
(read_runner_id(runtime) == Some(runner_id))
.then_some(())
.ok_or(LifecycleError::Journal)
}
Err(_) => {
let _ = fs::remove_file(&temporary);
Err(LifecycleError::Journal)
}
}
}
fn write_durable_file(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
let mut file = fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(path)?;
file.write_all(bytes)?;
file.sync_all()?;
let parent = path.parent().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"file has no parent directory",
)
})?;
sync_directory(parent)
}
#[cfg(unix)]
fn sync_directory(path: &Path) -> std::io::Result<()> {
fs::File::open(path)?.sync_all()
}
#[cfg(windows)]
fn sync_directory(path: &Path) -> std::io::Result<()> {
use std::os::windows::fs::OpenOptionsExt;
const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
const FILE_SHARE_ALL: u32 = 0x0000_0007;
const GENERIC_WRITE: u32 = 0x4000_0000;
fs::OpenOptions::new()
.access_mode(GENERIC_WRITE)
.share_mode(FILE_SHARE_ALL)
.custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
.open(path)?
.sync_all()
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeSet;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use crate::reconcile::{AllocationLock, InProcessAllocationLock};
use runner_manager_domain::model::{Elapsed, TargetScope};
use runner_manager_domain::store::SqliteStore;
use runner_manager_github::jit::JitRunner;
use runner_manager_testkit::clock::FakeClock;
use runner_manager_testkit::fixtures;
#[test]
fn production_provider_set_routes_every_implemented_isolated_backend() {
let providers = PlatformExecutionProvider::new(HostId::from_u128(1));
assert!(providers.isolated_provider(Backend::Auto).is_some());
assert!(providers.isolated_provider(Backend::Oci).is_some());
assert!(
providers
.isolated_provider(Backend::WindowsHyperVContainer)
.is_some()
);
assert!(
providers
.isolated_provider(Backend::VirtualMachine)
.is_none()
);
}
type CapturedFields = Vec<(String, String)>;
#[derive(Clone, Default)]
struct CapturedEvents(std::sync::Arc<std::sync::Mutex<Vec<CapturedFields>>>);
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedEvents {
fn on_event(
&self,
event: &tracing::Event<'_>,
_context: tracing_subscriber::layer::Context<'_, S>,
) {
struct Collect(Vec<(String, String)>);
impl tracing::field::Visit for Collect {
fn record_debug(
&mut self,
field: &tracing::field::Field,
value: &dyn std::fmt::Debug,
) {
self.0.push((
field.name().to_owned(),
format!("{value:?}").trim_matches('"').to_owned(),
));
}
}
let mut collected = Collect(Vec::new());
event.record(&mut collected);
self.0
.lock()
.expect("the capture mutex is not poisoned")
.push(collected.0);
}
}
#[test]
fn a_launch_the_runner_root_refused_names_the_cause_in_the_log_that_ships() {
use runner_manager_platform::logging;
use tracing_subscriber::layer::SubscriberExt as _;
let captured = CapturedEvents::default();
let error = RunnerRootError::DeniedByPrivacyPolicy {
requested: PathBuf::from("/Volumes/NVME/runners"),
refused: PathBuf::from("/Volumes/NVME"),
remediation: RootOwner::Host.remediation(),
};
let kind = error.kind();
let failure = tracing::subscriber::with_default(
tracing_subscriber::registry().with(captured.clone()),
|| root_failure(error),
);
assert!(
matches!(
&failure,
LifecycleError::Failed(FailureReason::Other(detail))
if detail.contains("/Volumes/NVME/runners")
&& detail.contains("Full Disk Access")
),
"the reason must still carry the detail: {failure:?}"
);
let events = captured
.0
.lock()
.expect("the capture mutex is not poisoned")
.clone();
let event = events
.iter()
.find(|fields| fields.iter().any(|(_, value)| value.contains(kind)))
.unwrap_or_else(|| panic!("the refusal did not name its cause: {events:?}"));
for (name, value) in event {
assert!(
logging::is_field_allowed(name),
"`{name}` is not allow-listed, so it ships as `{}`: {event:?}",
logging::REDACTION
);
assert_eq!(
&logging::redact(value),
value,
"`{name}` does not survive value-shape scrubbing: {event:?}"
);
}
}
fn nz(slot: u16) -> NonZeroU16 {
NonZeroU16::new(slot).expect("a positive slot")
}
const JIT: &str = "eyJzZWNyZXQiOiJnaHBfRE9fTk9UX0xFQUsifQ==";
#[derive(Debug, Default)]
struct FakeGithubLifecycle {
registration_failures: Mutex<VecDeque<bool>>,
observations: Mutex<VecDeque<LifecycleGithubObservation>>,
registrations: AtomicUsize,
remaining_runners: AtomicUsize,
deregistrations: Mutex<Vec<u64>>,
deregistration_fails: AtomicBool,
journal: Mutex<Option<Arc<SqliteStore>>>,
registration_facts: Mutex<Vec<RegistrationFact>>,
}
#[derive(Debug, Clone)]
struct RegistrationFact {
leased_slots: Vec<u16>,
work_folder: String,
runner_name: String,
journal_states: Vec<AttemptState>,
}
impl FakeGithubLifecycle {
fn fail(mut self, terminal: bool) -> Self {
self.registration_failures
.get_mut()
.expect("unpoisoned")
.push_back(terminal);
self
}
fn watch_journal(&self, store: Arc<SqliteStore>) {
*self.journal.lock().unwrap() = Some(store);
}
fn registration_facts(&self) -> Vec<RegistrationFact> {
self.registration_facts.lock().unwrap().clone()
}
fn observe(&self, observation: GithubRunnerObservation) {
let observation = match observation {
GithubRunnerObservation::Unreachable => LifecycleGithubObservation::unreachable(),
GithubRunnerObservation::NotRegistered => {
LifecycleGithubObservation::not_registered()
}
GithubRunnerObservation::Registered { busy } => {
LifecycleGithubObservation::registered(73, busy)
}
};
self.observations.lock().unwrap().push_back(observation);
}
}
#[async_trait]
impl LifecycleGithub for FakeGithubLifecycle {
async fn register(
&self,
_target: &ScaleTarget,
request: &JitRunnerRequest,
_cancel: &CancelToken,
) -> Result<JitRegistration, JitRequestFailure> {
self.registrations.fetch_add(1, Ordering::SeqCst);
if let Some(store) = self.journal.lock().unwrap().as_ref() {
let attempts = store.attempts().expect("the journal is readable");
let slots = attempts
.iter()
.filter_map(|attempt| attempt.workspace().slot_number())
.collect();
self.registration_facts
.lock()
.unwrap()
.push(RegistrationFact {
leased_slots: slots,
work_folder: request.work_folder().to_string(),
runner_name: request.name().to_string(),
journal_states: attempts.iter().map(RunnerAttempt::state).collect(),
});
}
if let Some(terminal) = self.registration_failures.lock().unwrap().pop_front() {
return Err(JitRequestFailure {
terminal,
reason: if terminal {
FailureReason::Other("GitHub refused JIT registration with 403".into())
} else {
FailureReason::JitRequestFailed
},
retry_after: None,
});
}
self.remaining_runners.store(1, Ordering::SeqCst);
Ok(JitRegistration::new(
EncodedJitConfig::new(JIT),
JitRunner {
id: 73,
name: request.name().to_string(),
os: "windows".into(),
status: "offline".into(),
busy: false,
runner_group_id: Some(1),
labels: request.labels().to_vec(),
},
))
}
async fn observe(
&self,
_target: &ScaleTarget,
_attempt: AttemptId,
_cancel: &CancelToken,
) -> LifecycleGithubObservation {
let observation = self
.observations
.lock()
.unwrap()
.pop_front()
.unwrap_or(LifecycleGithubObservation::not_registered());
if observation.status == GithubRunnerObservation::NotRegistered {
self.remaining_runners.store(0, Ordering::SeqCst);
}
observation
}
async fn deregister(
&self,
_target: &ScaleTarget,
runner_id: u64,
_cancel: &CancelToken,
) -> bool {
self.deregistrations.lock().unwrap().push(runner_id);
if self.deregistration_fails.load(Ordering::SeqCst) {
return false;
}
self.remaining_runners.store(0, Ordering::SeqCst);
true
}
}
#[derive(Debug)]
struct FakePackages {
version: RunnerVersion,
leases: Mutex<BTreeSet<AttemptId>>,
materializations: AtomicUsize,
materializing: AtomicUsize,
peak_materializing: AtomicUsize,
materialization_failures: AtomicUsize,
release_failures: AtomicUsize,
releases: AtomicUsize,
prunes: AtomicUsize,
}
impl Default for FakePackages {
fn default() -> Self {
Self {
version: RunnerVersion::parse("2.330.0").unwrap(),
leases: Mutex::new(BTreeSet::new()),
materializations: AtomicUsize::new(0),
materializing: AtomicUsize::new(0),
peak_materializing: AtomicUsize::new(0),
materialization_failures: AtomicUsize::new(0),
release_failures: AtomicUsize::new(0),
releases: AtomicUsize::new(0),
prunes: AtomicUsize::new(0),
}
}
}
impl FakePackages {
fn fail_materializations(&self, count: usize) {
self.materialization_failures.store(count, Ordering::SeqCst);
}
fn fail_releases(&self, count: usize) {
self.release_failures.store(count, Ordering::SeqCst);
}
}
#[async_trait]
impl RuntimePackages for FakePackages {
async fn materialize(
&self,
attempt: &RunnerAttempt,
_timings: &mut PackageTimings,
) -> Result<RunnerVersion, FailureReason> {
self.materializations.fetch_add(1, Ordering::SeqCst);
let now = self.materializing.fetch_add(1, Ordering::SeqCst) + 1;
self.peak_materializing.fetch_max(now, Ordering::SeqCst);
for _ in 0..5 {
tokio::task::yield_now().await;
}
self.materializing.fetch_sub(1, Ordering::SeqCst);
if self
.materialization_failures
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
if left > 0 { Some(left - 1) } else { None }
})
.is_ok()
{
return Err(FailureReason::Other(
"runner package materialization failed transiently".into(),
));
}
fs::create_dir_all(attempt.runtime_path()).unwrap();
fs::write(attempt.runtime_path().join("runner-package"), b"verified").unwrap();
self.leases.lock().unwrap().insert(attempt.id);
Ok(self.version.clone())
}
fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
if self
.release_failures
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
if left > 0 { Some(left - 1) } else { None }
})
.is_ok()
{
return Err(FailureReason::Other(
"runner package lease could not be released".into(),
));
}
self.leases.lock().unwrap().remove(&attempt);
self.releases.fetch_add(1, Ordering::SeqCst);
Ok(())
}
fn prune_obsolete_guarded(
&self,
_authority: PruneAuthority<'_>,
_attempts: &[RunnerAttempt],
) -> Result<(), FailureReason> {
self.prunes.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
#[derive(Debug, Default)]
struct FakeProcesses {
alive: AtomicBool,
completed_successfully: AtomicBool,
spawns: AtomicUsize,
spawn_failures: AtomicUsize,
live_spawn_failure: AtomicBool,
terminations: AtomicUsize,
intent: AtomicBool,
intent_failure: AtomicBool,
actions: Mutex<Vec<&'static str>>,
saw_secret: AtomicBool,
}
impl FakeProcesses {
fn fail_spawns(&self, count: usize) {
self.spawn_failures.store(count, Ordering::SeqCst);
}
fn fail_spawn_with_live_child(&self) {
self.live_spawn_failure.store(true, Ordering::SeqCst);
}
fn set_alive(&self, alive: bool) {
self.alive.store(alive, Ordering::SeqCst);
}
fn finish_successfully(&self) {
self.completed_successfully.store(true, Ordering::SeqCst);
self.alive.store(false, Ordering::SeqCst);
}
fn fail_intent(&self) {
self.intent_failure.store(true, Ordering::SeqCst);
}
}
impl ProcessSupervisor for FakeProcesses {
fn spawn(
&self,
attempt: &RunnerAttempt,
config: &EncodedJitConfig,
) -> Result<u32, ProcessStartFailure> {
self.spawns.fetch_add(1, Ordering::SeqCst);
let handoff = RestrictiveHandoff::create(
attempt.runtime_path(),
SecretString::from(config.expose().to_owned()),
)
.unwrap();
self.saw_secret
.store(config.expose() == JIT, Ordering::SeqCst);
let handoff_path = handoff.path().to_path_buf();
let failing = self
.spawn_failures
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
if left > 0 { Some(left - 1) } else { None }
})
.is_ok();
drop(handoff);
assert!(!handoff_path.exists(), "handoff must be absent on return");
if self.live_spawn_failure.swap(false, Ordering::SeqCst) {
self.alive.store(true, Ordering::SeqCst);
return Err(ProcessStartFailure::after_spawn_live(4242));
}
if failing {
return Err(ProcessStartFailure::before_spawn(
FailureReason::ProcessStartFailed,
));
}
self.alive.store(true, Ordering::SeqCst);
Ok(4242)
}
fn is_alive(&self, _attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
self.actions.lock().unwrap().push("observe_process");
Ok(self.alive.load(Ordering::SeqCst))
}
fn recovered_pid(&self, _attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
Ok(self.alive.load(Ordering::SeqCst).then_some(4242))
}
fn completed_successfully(&self, _attempt: &RunnerAttempt) -> bool {
self.completed_successfully.load(Ordering::SeqCst)
}
fn record_terminate_intent(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
self.actions.lock().unwrap().push("terminate_intent");
if self.intent_failure.load(Ordering::SeqCst) {
return Err(FailureReason::Other(
"terminate intent directory sync failed".into(),
));
}
self.intent.store(true, Ordering::SeqCst);
Ok(())
}
fn has_terminate_intent(&self, _attempt: &RunnerAttempt) -> bool {
self.intent.load(Ordering::SeqCst)
}
fn terminate(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
assert!(
self.intent.load(Ordering::SeqCst),
"the durable intent must exist before signalling"
);
self.actions.lock().unwrap().push("terminate");
self.terminations.fetch_add(1, Ordering::SeqCst);
self.alive.store(false, Ordering::SeqCst);
Ok(())
}
}
#[derive(Debug, Default)]
struct FakeIsolatedProvider {
resources: Mutex<BTreeMap<AttemptId, (EnvironmentIdentity, EnvironmentState)>>,
extra_resources: Mutex<Vec<EnvironmentIdentity>>,
unsupported_host: AtomicBool,
permission_denied: AtomicBool,
incompatible_image: AtomicBool,
prepare_failure: AtomicBool,
recovery_unavailable: AtomicBool,
destroy_deferred: AtomicBool,
handoffs: AtomicUsize,
completed_successfully: AtomicBool,
native_calls: AtomicUsize,
actions: Mutex<Vec<&'static str>>,
}
impl FakeIsolatedProvider {
fn identity(attempt: &RunnerAttempt, host: HostId) -> EnvironmentIdentity {
let AttemptExecution::Isolated {
provider_kind,
resolved_image,
generation,
..
} = attempt.execution()
else {
panic!("isolated intent required")
};
EnvironmentIdentity::Isolated {
host,
attempt: attempt.id,
provider_kind: *provider_kind,
environment_id: format!("environment-{}", attempt.id),
resolved_image: resolved_image.clone(),
generation: generation.clone(),
}
}
fn resource_count(&self) -> usize {
self.resources.lock().unwrap().len()
}
}
impl ExecutionProvider for FakeIsolatedProvider {
fn probe(&self, policy: &ScalePolicy) -> ProviderCapability {
if self.unsupported_host.load(Ordering::SeqCst) {
ProviderCapability::Unsupported
} else if self.permission_denied.load(Ordering::SeqCst) {
ProviderCapability::PermissionDenied
} else if policy.execution_policy().is_native() {
ProviderCapability::Unsupported
} else {
ProviderCapability::Ready
}
}
fn resolve(
&self,
policy: &ScalePolicy,
) -> Result<Option<ResolvedEnvironment>, FailureReason> {
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
policy.execution_policy()
else {
return Err(FailureReason::Other("isolated policy required".into()));
};
Ok(Some(ResolvedEnvironment {
provider_kind: Backend::Oci,
image: if self.incompatible_image.load(Ordering::SeqCst) {
ImageReference::new("vm-version:wrong-provider")
.expect("valid but incompatible")
} else {
image.clone()
},
}))
}
fn prepare(
&self,
attempt: &RunnerAttempt,
policy: &ScalePolicy,
) -> Result<PreparedEnvironment, FailureReason> {
assert_eq!(
attempt.state(),
AttemptState::Preparing,
"intent must be durable before provider effect"
);
self.actions.lock().unwrap().push("prepare");
let identity = Self::identity(attempt, policy.host_id);
self.resources
.lock()
.unwrap()
.insert(attempt.id, (identity.clone(), EnvironmentState::Starting));
if self.prepare_failure.swap(false, Ordering::SeqCst) {
return Err(FailureReason::Other("provider-secret-needle".into()));
}
Ok(PreparedEnvironment::isolated(attempt.id, identity))
}
fn start(
&self,
prepared: PreparedEnvironment,
attempt: &RunnerAttempt,
handoff: OneTimeJitHandoff<'_>,
) -> Result<EnvironmentIdentity, ProcessStartFailure> {
assert_eq!(prepared.attempt, attempt.id);
assert_eq!(attempt.state(), AttemptState::JitReceived);
assert_eq!(handoff.consume().expose(), JIT);
self.handoffs.fetch_add(1, Ordering::SeqCst);
self.actions.lock().unwrap().push("start");
let mut resources = self.resources.lock().unwrap();
let (identity, state) = resources.get_mut(&attempt.id).expect("prepared resource");
*state = EnvironmentState::Running;
Ok(identity.clone())
}
fn inspect(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
self.actions.lock().unwrap().push("inspect");
Ok(self
.resources
.lock()
.unwrap()
.get(&attempt.id)
.map_or(EnvironmentState::Missing, |(_, state)| *state))
}
fn recover(&self, attempt: &RunnerAttempt) -> Result<EnvironmentState, FailureReason> {
if self.recovery_unavailable.load(Ordering::SeqCst) {
return Err(FailureReason::Other(
"provider discovery unavailable".into(),
));
}
self.inspect(attempt)
}
fn owns(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
Ok(self
.resources
.lock()
.unwrap()
.get(&attempt.id)
.is_some_and(|(identity, _)| {
identity_matches_intent(attempt_host(identity), attempt, identity)
}))
}
fn stop(&self, attempt: &RunnerAttempt) -> Result<ProviderStop, FailureReason> {
assert!(self.owns(attempt)?);
self.actions.lock().unwrap().push("stop");
if let Some((_, state)) = self.resources.lock().unwrap().get_mut(&attempt.id) {
*state = EnvironmentState::Exited;
}
Ok(ProviderStop::Stopped)
}
fn destroy(&self, attempt: &RunnerAttempt) -> Result<ProviderDestroy, FailureReason> {
assert!(self.owns(attempt)?);
self.actions.lock().unwrap().push("destroy");
if self.destroy_deferred.load(Ordering::SeqCst) {
return Ok(ProviderDestroy::Deferred("provider busy"));
}
self.resources.lock().unwrap().remove(&attempt.id);
Ok(ProviderDestroy::Destroyed)
}
fn enumerate_owned(&self, host: HostId) -> Vec<EnvironmentIdentity> {
let mut owned = self
.resources
.lock()
.unwrap()
.values()
.filter(|(identity, _)| attempt_host(identity) == host)
.map(|(identity, _)| identity.clone())
.collect::<Vec<_>>();
owned.extend(
self.extra_resources
.lock()
.unwrap()
.iter()
.filter(|identity| attempt_host(identity) == host)
.cloned(),
);
owned
}
fn diagnostics(&self, _attempt: &RunnerAttempt) -> Vec<ProviderDiagnostic> {
vec![ProviderDiagnostic::CleanupDeferred; 40]
}
fn spawn(
&self,
_attempt: &RunnerAttempt,
_config: &EncodedJitConfig,
) -> Result<u32, ProcessStartFailure> {
self.native_calls.fetch_add(1, Ordering::SeqCst);
panic!("isolated provider must never spawn natively")
}
fn is_alive(&self, _attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
panic!("native fallback")
}
fn recovered_pid(&self, _attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
panic!("native fallback")
}
fn completed_successfully(&self, _attempt: &RunnerAttempt) -> bool {
self.completed_successfully.load(Ordering::SeqCst)
}
fn record_terminate_intent(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
panic!("native fallback")
}
fn has_terminate_intent(&self, _attempt: &RunnerAttempt) -> bool {
false
}
fn terminate(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
panic!("native fallback")
}
}
fn attempt_host(identity: &EnvironmentIdentity) -> HostId {
let EnvironmentIdentity::Isolated { host, .. } = identity else {
panic!("isolated identity")
};
*host
}
#[derive(Debug, Default)]
struct FakeDemand {
answers: Mutex<VecDeque<bool>>,
}
impl FakeDemand {
fn answering(answers: impl IntoIterator<Item = bool>) -> Self {
Self {
answers: Mutex::new(answers.into_iter().collect()),
}
}
}
#[async_trait]
impl DemandPersistence for FakeDemand {
async fn persists(&self, _policy: PolicyId) -> bool {
self.answers.lock().unwrap().pop_front().unwrap_or(true)
}
}
#[derive(Debug, Default)]
struct FakeDelay(Mutex<Vec<Duration>>);
#[async_trait]
impl RetryDelay for FakeDelay {
async fn wait(&self, duration: Duration) {
self.0.lock().unwrap().push(duration);
}
}
#[tokio::test]
async fn live_capacity_reads_the_updated_sqlite_host_and_rejects_a_missing_host() {
let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let path = harness._root.path().join("live-capacity.sqlite3");
harness.store = Arc::new(SqliteStore::open(&path).unwrap());
harness.store.put_host(&harness.host).unwrap();
harness.launcher = harness.restart();
assert_eq!(
harness.launcher.host_capacity(&harness.host).await.unwrap(),
harness.host.host_capacity
);
let writer = SqliteStore::open(&path).unwrap();
let mut updated = harness.host.clone();
updated.host_capacity = std::num::NonZeroU16::new(12).unwrap();
writer.put_host(&updated).unwrap();
assert_eq!(
harness
.launcher
.host_capacity(&harness.host)
.await
.unwrap()
.get(),
12
);
updated.host_capacity = std::num::NonZeroU16::new(2).unwrap();
writer.put_host(&updated).unwrap();
assert_eq!(
harness
.launcher
.host_capacity(&harness.host)
.await
.unwrap()
.get(),
2
);
let mut missing = harness.host.clone();
missing.id = HostId::from_u128(999);
assert_eq!(
harness.launcher.host_capacity(&missing).await,
Err(crate::reconcile::HostCapacityError::MissingHost)
);
}
struct Harness {
_root: tempfile::TempDir,
app_paths: runner_manager_platform::paths::AppPaths,
launcher: LifecycleLauncher,
demand: Arc<dyn DemandPersistence>,
store: Arc<SqliteStore>,
github: Arc<FakeGithubLifecycle>,
packages: Arc<FakePackages>,
processes: Arc<FakeProcesses>,
clock: Arc<FakeClock>,
events: Arc<AttemptEventLog>,
reconcile_events: Arc<crate::reconcile::EventLog>,
delay: Arc<FakeDelay>,
host: runner_manager_domain::model::Host,
policy: ScalePolicy,
allocation_lock: InProcessAllocationLock,
workspace_root: Option<LocalAbsolutePath>,
}
impl Harness {
fn new(github: FakeGithubLifecycle, demand: Arc<dyn DemandPersistence>) -> Self {
let root = tempfile::tempdir().unwrap();
let paths = runner_manager_platform::paths::AppPaths::rooted_at(root.path());
paths.create_all().unwrap();
let policy = fixtures::policy()
.repository("octo/repo")
.autoscale("home", 2)
.active()
.build();
let host_root = root.path().join("host-root");
fs::create_dir_all(&host_root).unwrap();
let mut host = fixtures::host().build();
host.runner_root_override = Some(
LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
.expect("a local absolute host root"),
);
let store = Arc::new(SqliteStore::open_in_memory().unwrap());
store.put_host(&host).unwrap();
let github = Arc::new(github);
let packages = Arc::new(FakePackages::default());
let processes = Arc::new(FakeProcesses::default());
let clock = Arc::new(FakeClock::default());
let events = Arc::new(AttemptEventLog::default());
let reconcile_events = Arc::new(crate::reconcile::EventLog::new());
let delay = Arc::new(FakeDelay::default());
let ports = LifecyclePorts {
store: Arc::clone(&store) as Arc<dyn Store>,
github: Arc::clone(&github) as Arc<dyn LifecycleGithub>,
packages: Arc::clone(&packages) as Arc<dyn RuntimePackages>,
processes: Arc::clone(&processes) as Arc<dyn ProcessSupervisor>,
clock: Arc::clone(&clock) as Arc<dyn Clock>,
demand: Arc::clone(&demand),
delay: Arc::clone(&delay) as Arc<dyn RetryDelay>,
events: Arc::clone(&events) as Arc<dyn AttemptEventSink>,
reconcile_events: Arc::clone(&reconcile_events) as Arc<dyn EventSink>,
};
let launcher = Self::launcher_over(policy.host_id, &paths, ports);
Self {
_root: root,
app_paths: paths,
launcher,
demand,
store,
github,
packages,
processes,
clock,
events,
reconcile_events,
delay,
host,
policy,
allocation_lock: InProcessAllocationLock::new(),
workspace_root: None,
}
}
fn with_host_runner_root(mut self) -> Self {
let host_root = self.host_root();
fs::create_dir_all(&host_root).unwrap();
self.host.runner_root_override = Some(
LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
.expect("a local absolute host root"),
);
self.store.put_host(&self.host).unwrap();
self
}
fn with_persistent_workspace(mut self, capacity: u16) -> Self {
self = self.with_host_runner_root();
let root = self._root.path().join("persist");
let root = LocalAbsolutePath::new(root.to_str().expect("a UTF-8 temporary path"))
.expect("a local absolute workspace root");
self.policy = fixtures::policy()
.repository("octo/repo")
.autoscale("home", capacity)
.active()
.build();
self.policy
.set_workspace_policy(
WorkspacePolicy::persistent(root.clone(), TargetScope::Repository)
.expect("a repository may be persistent"),
)
.expect("a repository may be persistent");
self.workspace_root = Some(root);
self.store.insert_policy(&self.policy).unwrap();
self
}
fn workspace_root(&self) -> &LocalAbsolutePath {
self.workspace_root
.as_ref()
.expect("this harness configured a persistent workspace")
}
fn slot_path(&self, slot: u16) -> PathBuf {
self.workspace_root().as_path().join(format!("s{slot}"))
}
fn host_root(&self) -> PathBuf {
self._root.path().join("host-root")
}
fn attempt(&self, id: AttemptId) -> RunnerAttempt {
self.store
.attempt(id)
.unwrap()
.expect("the attempt is journalled")
}
fn conclude(&self, id: AttemptId) -> RunnerAttempt {
let mut attempt = self.attempt(id);
attempt
.conclude(
AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
self.clock.now(),
)
.unwrap();
self.store.record_attempt(&attempt).unwrap();
attempt
}
async fn cleanup_retaining_work(&self, id: AttemptId) {
self.conclude(id);
self.launcher
.clean(id)
.await
.expect("the slot is scrubbed and the lease released");
}
fn launcher_over(
host: HostId,
paths: &runner_manager_platform::paths::AppPaths,
ports: LifecyclePorts,
) -> LifecycleLauncher {
LifecycleLauncher::new(
host,
paths.clone(),
paths.logs_dir(),
1,
RecoveryTimeouts::new(
Elapsed::seconds(10),
Elapsed::seconds(10),
Elapsed::seconds(10),
),
RetryPolicy::bounded(3, Duration::from_millis(10), Duration::from_millis(25)),
ports,
)
}
fn restart(&self) -> LifecycleLauncher {
Self::launcher_over(
self.policy.host_id,
&self.app_paths,
LifecyclePorts {
store: Arc::clone(&self.store) as Arc<dyn Store>,
github: Arc::clone(&self.github) as Arc<dyn LifecycleGithub>,
packages: Arc::clone(&self.packages) as Arc<dyn RuntimePackages>,
processes: Arc::clone(&self.processes) as Arc<dyn ProcessSupervisor>,
clock: Arc::clone(&self.clock) as Arc<dyn Clock>,
demand: Arc::clone(&self.demand),
delay: Arc::clone(&self.delay) as Arc<dyn RetryDelay>,
events: Arc::clone(&self.events) as Arc<dyn AttemptEventSink>,
reconcile_events: Arc::clone(&self.reconcile_events) as Arc<dyn EventSink>,
},
)
}
async fn ready(&self) {
self.launcher
.recover_startup(std::slice::from_ref(&self.policy))
.await
.unwrap();
}
async fn launch(&self) -> RunnerAttempt {
self.launch_result().await.unwrap()
}
async fn launch_result(&self) -> Result<RunnerAttempt, LaunchFailure> {
self.launch_after_waiting(Duration::ZERO).await
}
async fn launch_after_waiting(
&self,
lock_wait: Duration,
) -> Result<RunnerAttempt, LaunchFailure> {
let reserved = self.reserve().await?;
self.launcher
.finish_launch(
&self.policy,
reserved,
ReservationWaits {
lock: lock_wait,
order: Duration::from_secs(2),
},
)
.await
}
async fn launch_through(
&self,
launcher: &LifecycleLauncher,
) -> Result<RunnerAttempt, LaunchFailure> {
let reserved = self.reserve_through(launcher).await?;
launcher
.finish_launch(&self.policy, reserved, ReservationWaits::default())
.await
}
async fn reserve(&self) -> Result<RunnerAttempt, LaunchFailure> {
self.reserve_through(&self.launcher).await
}
async fn reserve_through(
&self,
launcher: &LifecycleLauncher,
) -> Result<RunnerAttempt, LaunchFailure> {
let guard = self.allocation_lock.acquire().await.unwrap();
launcher
.launch(LaunchRequest {
host: &self.host,
policy: &self.policy,
allocation_guard: &guard,
})
.await
}
fn only_attempt(&self) -> RunnerAttempt {
self.store.attempts().unwrap().into_iter().next().unwrap()
}
}
fn isolated_harness() -> Harness {
use runner_manager_domain::execution::{ExecutionPolicy, ResourceLimits};
let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_host_runner_root();
let image =
ImageReference::new(format!("registry.example/runner@sha256:{}", "a".repeat(64)))
.expect("pinned image");
harness
.policy
.set_execution_policy(ExecutionPolicy::Isolated {
backend: Backend::Oci,
image,
resources: ResourceLimits {
cpu_millis: 1000,
memory_mib: 1024,
disk_mib: 4096,
},
})
.expect("isolated policy");
harness.github.watch_journal(Arc::clone(&harness.store));
harness
}
fn launcher_with_isolated(
harness: &Harness,
provider: Arc<FakeIsolatedProvider>,
) -> LifecycleLauncher {
Harness::launcher_over(
harness.policy.host_id,
&harness.app_paths,
LifecyclePorts {
store: Arc::clone(&harness.store) as Arc<dyn Store>,
github: Arc::clone(&harness.github) as Arc<dyn LifecycleGithub>,
packages: Arc::clone(&harness.packages) as Arc<dyn RuntimePackages>,
processes: provider as Arc<dyn ExecutionProvider>,
clock: Arc::clone(&harness.clock) as Arc<dyn Clock>,
demand: Arc::clone(&harness.demand),
delay: Arc::clone(&harness.delay) as Arc<dyn RetryDelay>,
events: Arc::clone(&harness.events) as Arc<dyn AttemptEventSink>,
reconcile_events: Arc::clone(&harness.reconcile_events) as Arc<dyn EventSink>,
},
)
}
#[tokio::test]
async fn isolated_policy_and_journal_never_enter_native_provider_or_jit() {
use runner_manager_domain::execution::{
AttemptExecution, Backend, ExecutionPolicy, ImageReference, ResourceLimits,
};
let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_host_runner_root();
let image =
ImageReference::new(format!("registry.example/runner@sha256:{}", "a".repeat(64)))
.expect("pinned image");
harness
.policy
.set_execution_policy(ExecutionPolicy::Isolated {
backend: Backend::Oci,
image: image.clone(),
resources: ResourceLimits {
cpu_millis: 1000,
memory_mib: 1024,
disk_mib: 4096,
},
})
.expect("valid isolated profile");
harness.ready().await;
assert_eq!(
harness.processes.probe(&harness.policy),
ProviderCapability::Unsupported
);
let failure = harness
.launch_result()
.await
.expect_err("provider is unavailable");
assert_eq!(
failure.reason,
FailureReason::IsolationProvider(IsolationProviderFailure::Unsupported)
);
let runtime = harness.host_root().join("isolated-owned-environment");
fs::create_dir_all(&runtime).expect("provider resource fixture");
fs::write(runtime.join("owned-marker"), "non-secret").expect("fixture marker");
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(0x123),
harness.policy.id,
runtime.clone(),
harness.clock.now(),
);
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image,
generation: "generation-123".into(),
})
.expect("isolated intent");
attempt
.prepared_environment("environment-123".into())
.expect("prepared identity");
harness
.store
.record_attempt(&attempt)
.expect("durable isolated journal");
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("unsupported isolated cleanup is profile-local");
let failure = harness
.launch_result()
.await
.expect_err("provider unavailable for this policy");
assert_eq!(
failure.reason,
FailureReason::IsolationProvider(IsolationProviderFailure::Unsupported)
);
harness
.launcher
.supervise(&harness.policy)
.await
.expect("unavailable attempt is deferred");
assert!(harness.launcher.clean(attempt.id).await.is_err());
assert_eq!(
harness.store.attempt(attempt.id).unwrap().unwrap().state(),
AttemptState::CleanupDeferred
);
assert!(matches!(
harness.launcher.recover_startup(&[]).await,
Err(LifecycleError::RecoveryIncomplete)
));
assert!(
runtime.join("owned-marker").is_file(),
"native cleanup touched the isolated environment"
);
assert!(
harness.processes.actions.lock().unwrap().is_empty(),
"native recovery inspected an isolated identity"
);
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert_eq!(harness.packages.materializations.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn isolated_prepare_is_durable_before_jit_and_start_uses_one_handoff() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("ready");
let attempt = harness
.launch_through(&launcher)
.await
.expect("isolated launch");
assert_eq!(attempt.state(), AttemptState::Starting);
assert_eq!(provider.resource_count(), 1);
assert_eq!(provider.handoffs.load(Ordering::SeqCst), 1);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
let fact = harness
.github
.registration_facts()
.pop()
.expect("JIT registration");
assert_eq!(
fact.journal_states,
vec![AttemptState::Prepared],
"provider readiness preceded GitHub JIT"
);
assert_eq!(
harness
.events
.events()
.into_iter()
.filter_map(|event| match event {
AttemptEvent::State { state, .. } => Some(state),
_ => None,
})
.collect::<Vec<_>>(),
vec![
AttemptState::Allocated,
AttemptState::Preparing,
AttemptState::Prepared,
AttemptState::JitReceived,
AttemptState::Starting
]
);
let restarted = launcher_with_isolated(&harness, Arc::clone(&provider));
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("adopt live sandbox");
assert_eq!(provider.resource_count(), 1);
assert_eq!(
harness.github.registrations.load(Ordering::SeqCst),
1,
"restart cannot issue duplicate JIT"
);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
restarted
.supervise(&harness.policy)
.await
.expect("job observed");
assert_eq!(
harness.store.attempt(attempt.id).unwrap().unwrap().state(),
AttemptState::Busy
);
provider
.resources
.lock()
.unwrap()
.get_mut(&attempt.id)
.unwrap()
.1 = EnvironmentState::Exited;
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
restarted
.supervise(&harness.policy)
.await
.expect("terminal cleanup");
assert_eq!(provider.resource_count(), 0);
assert_eq!(
harness.store.attempt(attempt.id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
#[tokio::test]
async fn attached_isolated_job_exit_is_recorded_as_completed() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let attempt = harness.launch_through(&launcher).await.unwrap();
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
launcher.supervise(&harness.policy).await.unwrap();
provider
.resources
.lock()
.unwrap()
.get_mut(&attempt.id)
.unwrap()
.1 = EnvironmentState::Exited;
provider
.completed_successfully
.store(true, Ordering::SeqCst);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
launcher.supervise(&harness.policy).await.unwrap();
let cleaned = harness.store.attempt(attempt.id).unwrap().unwrap();
assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
assert_eq!(cleaned.state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn isolated_prepare_failure_stays_local_and_crash_gap_recovers_owned_resource() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
provider.prepare_failure.store(true, Ordering::SeqCst);
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("ready");
let failure = harness
.launch_through(&launcher)
.await
.expect_err("prepare failed after creating resource");
assert_eq!(
failure.reason,
FailureReason::IsolationProvider(IsolationProviderFailure::RuntimeOperationFailed)
);
assert_eq!(provider.resource_count(), 1);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert!(
!harness
.store
.dump_text()
.unwrap()
.contains("provider-secret-needle")
);
let id = harness.store.attempts().unwrap()[0].id;
assert_eq!(
harness.store.attempt(id).unwrap().unwrap().outcome(),
Some(&AttemptOutcome::failed(FailureReason::IsolationProvider(
IsolationProviderFailure::RuntimeOperationFailed
)))
);
let restarted = launcher_with_isolated(&harness, Arc::clone(&provider));
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("profile-local recovery");
assert_eq!(
provider.resource_count(),
0,
"the exact owned resource was destroyed once"
);
assert_eq!(
harness.store.attempt(id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
let diagnostic =
fs::read_to_string(restarted.diagnostics_root.join(format!("{id}.log"))).unwrap();
assert!(diagnostic.contains("reason=isolation_runtime_operation_failed"));
assert!(!diagnostic.contains("provider-secret-needle"));
assert_eq!(
provider
.actions
.lock()
.unwrap()
.iter()
.filter(|action| **action == "destroy")
.count(),
1
);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
}
#[test]
fn isolation_capability_and_adapter_failures_keep_closed_categories() {
let cases = [
(
ProviderCapability::NotInstalled,
IsolationProviderFailure::NotInstalled,
),
(
ProviderCapability::PermissionDenied,
IsolationProviderFailure::PermissionDenied,
),
(
ProviderCapability::ImageUnavailableOrIncompatible,
IsolationProviderFailure::ImageUnavailableOrIncompatible,
),
(
ProviderCapability::DiskQuotaUnavailable,
IsolationProviderFailure::DiskQuotaUnavailable,
),
(
ProviderCapability::Degraded,
IsolationProviderFailure::Degraded,
),
];
for (capability, category) in cases {
assert_eq!(
capability.refusal(),
FailureReason::IsolationProvider(category)
);
}
let raw = FailureReason::Other("ghp_secret_from_provider".into());
let safe = safe_isolation_reason(raw, IsolationProviderFailure::RuntimeOperationFailed);
assert_eq!(
safe,
FailureReason::IsolationProvider(IsolationProviderFailure::RuntimeOperationFailed)
);
assert!(!safe.to_string().contains("ghp_secret"));
assert_eq!(
safe_isolation_reason(
FailureReason::IsolationProvider(IsolationProviderFailure::PermissionDenied),
IsolationProviderFailure::RuntimeOperationFailed,
),
FailureReason::IsolationProvider(IsolationProviderFailure::PermissionDenied)
);
}
#[tokio::test]
async fn isolated_permission_and_image_failures_stop_before_jit_without_native_fallback() {
for image_failure in [false, true] {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
provider
.permission_denied
.store(!image_failure, Ordering::SeqCst);
provider
.incompatible_image
.store(image_failure, Ordering::SeqCst);
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let error = harness
.launch_through(&launcher)
.await
.expect_err("pre-JIT refusal");
let expected = if image_failure {
"resolved image is incompatible"
} else {
"permission denied"
};
assert!(error.reason.to_string().contains(expected));
assert!(
harness.store.attempts().unwrap().is_empty(),
"no allocation was journalled"
);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert_eq!(provider.handoffs.load(Ordering::SeqCst), 0);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
}
}
#[tokio::test]
async fn unsupported_host_stops_before_jit_without_allocation_or_native_fallback() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
provider.unsupported_host.store(true, Ordering::SeqCst);
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let error = harness
.launch_through(&launcher)
.await
.expect_err("unsupported host must be refused before JIT");
assert_eq!(
error.reason,
FailureReason::IsolationProvider(IsolationProviderFailure::Unsupported)
);
assert!(harness.store.attempts().unwrap().is_empty());
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert_eq!(provider.handoffs.load(Ordering::SeqCst), 0);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
assert!(provider.actions.lock().unwrap().is_empty());
}
#[tokio::test]
async fn no_journalled_environment_id_waits_for_provider_absence_proof() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
let runtime = harness.host_root().join("prepare-without-identity");
fs::create_dir_all(&runtime).unwrap();
let mut attempt = RunnerAttempt::allocate(
AttemptId::from_u128(0x9191),
harness.policy.id,
runtime,
harness.clock.now(),
);
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
harness.policy.execution_policy()
else {
unreachable!()
};
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image.clone(),
generation: "prepare-generation".into(),
})
.unwrap();
attempt.begin_prepare(harness.clock.now()).unwrap();
attempt
.conclude(
AttemptOutcome::failed(FailureReason::Other("prepare interrupted".into())),
harness.clock.now(),
)
.unwrap();
harness.store.record_attempt(&attempt).unwrap();
provider.recovery_unavailable.store(true, Ordering::SeqCst);
launcher
.clean(attempt.id)
.await
.expect_err("absence unproven");
let deferred = harness.store.attempt(attempt.id).unwrap().unwrap();
assert_eq!(deferred.state(), AttemptState::CleanupDeferred);
assert!(deferred.counts_against_capacity());
provider.recovery_unavailable.store(false, Ordering::SeqCst);
launcher.clean(attempt.id).await.expect("absence proven");
assert_eq!(
harness.store.attempt(attempt.id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
#[tokio::test]
async fn isolated_terminal_capacity_waits_for_confirmed_absence_and_generation_matches() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("ready");
let mut attempt = harness.launch_through(&launcher).await.expect("launch");
attempt
.conclude(
AttemptOutcome::failed(FailureReason::ProcessStartFailed),
harness.clock.now(),
)
.unwrap();
harness.store.record_attempt(&attempt).unwrap();
provider.destroy_deferred.store(true, Ordering::SeqCst);
launcher
.clean(attempt.id)
.await
.expect_err("provider deferred destroy");
let deferred = harness.store.attempt(attempt.id).unwrap().unwrap();
assert_eq!(deferred.state(), AttemptState::CleanupDeferred);
assert!(deferred.counts_against_capacity());
assert_eq!(provider.resource_count(), 1);
provider.destroy_deferred.store(false, Ordering::SeqCst);
let restarted = launcher_with_isolated(&harness, Arc::clone(&provider));
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("retry cleanup");
assert_eq!(provider.resource_count(), 0);
assert!(
!harness
.store
.attempt(attempt.id)
.unwrap()
.unwrap()
.counts_against_capacity()
);
let second_id = AttemptId::from_u128(0x9999);
let second_runtime = harness.host_root().join("generation-mismatch");
fs::create_dir_all(&second_runtime).unwrap();
let mut terminal = RunnerAttempt::allocate(
second_id,
harness.policy.id,
second_runtime,
harness.clock.now(),
);
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
harness.policy.execution_policy()
else {
unreachable!()
};
terminal
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image.clone(),
generation: "expected-generation".into(),
})
.unwrap();
let mut stranger = FakeIsolatedProvider::identity(&terminal, harness.policy.host_id);
let EnvironmentIdentity::Isolated { environment_id, .. } = &stranger else {
unreachable!()
};
terminal.begin_prepare(harness.clock.now()).unwrap();
terminal
.prepared_environment(environment_id.clone())
.unwrap();
terminal.mark_prepared(harness.clock.now()).unwrap();
terminal.jit_received(harness.clock.now()).unwrap();
terminal.started_isolated(harness.clock.now()).unwrap();
terminal
.conclude(
AttemptOutcome::failed(FailureReason::ProcessStartFailed),
harness.clock.now(),
)
.unwrap();
harness.store.record_attempt(&terminal).unwrap();
let EnvironmentIdentity::Isolated { generation, .. } = &mut stranger else {
unreachable!()
};
*generation = "another-generation".into();
provider
.resources
.lock()
.unwrap()
.insert(second_id, (stranger, EnvironmentState::Running));
restarted
.clean(terminal.id)
.await
.expect_err("ownership mismatch quarantined");
assert_eq!(provider.resource_count(), 1);
assert_eq!(
harness.store.attempt(terminal.id).unwrap().unwrap().state(),
AttemptState::CleanupDeferred
);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn isolated_restart_at_each_transition_adopts_or_destroys_one_resource() {
for stage in [
AttemptState::Allocated,
AttemptState::Preparing,
AttemptState::Prepared,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
AttemptState::Finished,
AttemptState::Failed,
AttemptState::Orphaned,
AttemptState::Destroying,
AttemptState::CleanupDeferred,
] {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
let id = AttemptId::from_u128(0x1000 + stage as u128);
let runtime = harness.host_root().join(format!("phase-{stage}"));
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
harness.policy.execution_policy()
else {
unreachable!()
};
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image.clone(),
generation: format!("generation-{stage}"),
})
.unwrap();
if stage != AttemptState::Allocated {
attempt.begin_prepare(harness.clock.now()).unwrap();
}
let identity = FakeIsolatedProvider::identity(&attempt, harness.policy.host_id);
if stage != AttemptState::Allocated {
let resource_state = if matches!(
stage,
AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Destroying
| AttemptState::CleanupDeferred
) {
EnvironmentState::Exited
} else {
EnvironmentState::Running
};
provider
.resources
.lock()
.unwrap()
.insert(id, (identity.clone(), resource_state));
}
if !matches!(stage, AttemptState::Allocated | AttemptState::Preparing) {
let EnvironmentIdentity::Isolated { environment_id, .. } = &identity else {
unreachable!()
};
attempt
.prepared_environment(environment_id.clone())
.unwrap();
attempt.mark_prepared(harness.clock.now()).unwrap();
}
if matches!(
stage,
AttemptState::JitReceived
| AttemptState::Starting
| AttemptState::Idle
| AttemptState::Busy
| AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Destroying
| AttemptState::CleanupDeferred
) {
attempt.jit_received(harness.clock.now()).unwrap();
}
if matches!(
stage,
AttemptState::Starting
| AttemptState::Idle
| AttemptState::Busy
| AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Destroying
| AttemptState::CleanupDeferred
) {
attempt.started_isolated(harness.clock.now()).unwrap();
}
if matches!(
stage,
AttemptState::Idle | AttemptState::Busy | AttemptState::Finished
) {
attempt.registered_idle(73, harness.clock.now()).unwrap();
}
if matches!(stage, AttemptState::Busy | AttemptState::Finished) {
attempt.assigned_job(73, harness.clock.now()).unwrap();
}
if matches!(
stage,
AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Destroying
| AttemptState::CleanupDeferred
) {
let outcome = match stage {
AttemptState::Finished => AttemptOutcome::CompletedJob,
AttemptState::Orphaned => AttemptOutcome::Orphaned,
_ => AttemptOutcome::failed(FailureReason::ProcessStartFailed),
};
attempt.conclude(outcome, harness.clock.now()).unwrap();
}
if matches!(
stage,
AttemptState::Destroying | AttemptState::CleanupDeferred
) {
attempt.begin_destroy(harness.clock.now()).unwrap();
}
if stage == AttemptState::CleanupDeferred {
attempt.defer_cleanup(harness.clock.now()).unwrap();
}
harness.store.record_attempt(&attempt).unwrap();
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap_or_else(|error| panic!("restart from {stage}: {error}"));
let resumed = harness.store.attempt(id).unwrap().unwrap();
if matches!(
stage,
AttemptState::Allocated
| AttemptState::Preparing
| AttemptState::Prepared
| AttemptState::Finished
| AttemptState::Failed
| AttemptState::Orphaned
| AttemptState::Destroying
| AttemptState::CleanupDeferred
) {
assert_eq!(
resumed.state(),
AttemptState::Cleaned,
"{stage} must retire after restart"
);
} else {
assert!(resumed.counts_against_capacity(), "live {stage} is adopted");
assert_eq!(provider.resource_count(), 1);
provider.resources.lock().unwrap().get_mut(&id).unwrap().1 =
EnvironmentState::Exited;
launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
harness.store.attempt(id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
assert_eq!(
provider.resource_count(),
0,
"{stage} resource cleaned exactly once"
);
let destroyed = provider
.actions
.lock()
.unwrap()
.iter()
.filter(|action| **action == "destroy")
.count();
assert_eq!(
destroyed,
usize::from(stage != AttemptState::Allocated),
"{stage} destroy count"
);
assert_eq!(
harness.github.registrations.load(Ordering::SeqCst),
0,
"restart from {stage} cannot duplicate JIT"
);
assert_eq!(provider.native_calls.load(Ordering::SeqCst), 0);
}
}
#[tokio::test]
async fn duplicate_provider_ownership_is_quarantined_until_unambiguous() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
let id = AttemptId::from_u128(0x7777);
let runtime = harness.host_root().join("duplicate-provider-ownership");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
harness.policy.execution_policy()
else {
unreachable!()
};
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image.clone(),
generation: "duplicate-generation".into(),
})
.unwrap();
attempt.begin_prepare(harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
let first = FakeIsolatedProvider::identity(&attempt, harness.policy.host_id);
let mut second = first.clone();
let EnvironmentIdentity::Isolated { environment_id, .. } = &mut second else {
unreachable!()
};
*environment_id = "another-environment".into();
provider
.resources
.lock()
.unwrap()
.insert(id, (first, EnvironmentState::Running));
provider.extra_resources.lock().unwrap().push(second);
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("duplicate is local");
let deferred = harness.store.attempt(id).unwrap().unwrap();
assert_eq!(deferred.state(), AttemptState::CleanupDeferred);
assert!(deferred.counts_against_capacity());
assert_eq!(provider.resource_count(), 1);
assert_eq!(
provider
.actions
.lock()
.unwrap()
.iter()
.filter(|action| **action == "destroy")
.count(),
0
);
provider.extra_resources.lock().unwrap().clear();
harness.clock.advance_secs(31);
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("single identity recovered");
assert_eq!(provider.resource_count(), 0);
assert_eq!(
harness.store.attempt(id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
#[tokio::test]
async fn provider_resource_without_live_journal_is_surfaced_but_never_deleted() {
let harness = isolated_harness();
let provider = Arc::new(FakeIsolatedProvider::default());
let launcher = launcher_with_isolated(&harness, Arc::clone(&provider));
let id = AttemptId::from_u128(0x8888);
let runtime = harness.host_root().join("orphan-provider-resource");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
let runner_manager_domain::execution::ExecutionPolicy::Isolated { image, .. } =
harness.policy.execution_policy()
else {
unreachable!()
};
attempt
.allocate_execution(AttemptExecution::Isolated {
provider_kind: Backend::Oci,
environment_id: None,
resolved_image: image.clone(),
generation: "orphan-generation".into(),
})
.unwrap();
let identity = FakeIsolatedProvider::identity(&attempt, harness.policy.host_id);
provider
.resources
.lock()
.unwrap()
.insert(id, (identity, EnvironmentState::Running));
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("orphan cannot block host");
assert_eq!(provider.resource_count(), 1);
assert!(
harness
.events
.events()
.contains(&AttemptEvent::QuarantinedOwnedResource { attempt: id })
);
assert_eq!(
provider
.actions
.lock()
.unwrap()
.iter()
.filter(|action| **action == "destroy")
.count(),
0
);
let EnvironmentIdentity::Isolated { environment_id, .. } = provider
.resources
.lock()
.unwrap()
.get(&id)
.unwrap()
.0
.clone()
else {
unreachable!()
};
attempt.begin_prepare(harness.clock.now()).unwrap();
attempt.prepared_environment(environment_id).unwrap();
attempt.mark_prepared(harness.clock.now()).unwrap();
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started_isolated(harness.clock.now()).unwrap();
attempt
.conclude(
AttemptOutcome::failed(FailureReason::ProcessStartFailed),
harness.clock.now(),
)
.unwrap();
attempt.begin_destroy(harness.clock.now()).unwrap();
attempt.clean_isolated(harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("cleaned row cannot authorize deletion");
assert_eq!(provider.resource_count(), 1);
assert_eq!(
provider
.actions
.lock()
.unwrap()
.iter()
.filter(|action| **action == "destroy")
.count(),
0
);
}
#[tokio::test]
async fn a_root_that_refuses_a_launch_is_recorded_and_cleared_when_one_succeeds() {
use runner_manager_platform::service::{clear_runner_root_refusal, runner_root_refusals};
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_host_runner_root();
harness.ready().await;
let unusable = harness
._root
.path()
.join("absent")
.join("deeper")
.join("runners");
let mut host = harness.host.clone();
host.runner_root_override = Some(
LocalAbsolutePath::new(unusable.to_str().expect("a UTF-8 temporary path"))
.expect("a local absolute host root"),
);
harness.store.put_host(&host).unwrap();
let failure = harness
.launch_result()
.await
.expect_err("a root whose parents are missing cannot hold a runner");
assert!(
matches!(failure.reason, FailureReason::Other(_)),
"{failure:?}"
);
let refusals = runner_root_refusals(&harness.app_paths).expect("readable");
let refusal = refusals
.first()
.expect("the refusal reached the one surface that can hold it");
assert_eq!(refusal.policy, harness.policy.id.to_string());
assert_eq!(refusal.kind, "missing_parents");
assert!(
refusal.root.contains("runners") && refusal.detail.contains("runners"),
"the directory must be named in full: {refusal:?}"
);
harness.store.put_host(&harness.host).unwrap();
harness.launch().await;
assert!(
runner_root_refusals(&harness.app_paths)
.expect("readable")
.is_empty(),
"a successful placement clears that policy's record"
);
clear_runner_root_refusal(&harness.app_paths, &harness.policy.id.to_string())
.expect("cleanup");
}
#[tokio::test]
async fn a_launch_reports_how_long_each_step_took() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
let events = harness.events.events();
let launched = events
.iter()
.position(|event| matches!(event, AttemptEvent::Launched { attempt, .. } if *attempt == started.id))
.expect("a successful launch reports its step timings");
let starting = events
.iter()
.position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Starting } if *attempt == started.id))
.unwrap();
assert!(starting < launched, "timings follow the start: {events:?}");
}
#[tokio::test]
async fn a_failed_launch_still_reports_its_step_timings() {
let harness = Harness::new(
FakeGithubLifecycle::default().fail(true),
Arc::new(PersistentDemand),
);
harness.ready().await;
let failed = harness.launch_after_waiting(Duration::from_secs(90)).await;
assert!(failed.is_err());
let timings = harness
.events
.events()
.into_iter()
.find_map(|event| match event {
AttemptEvent::Launched {
started: false,
timings,
..
} => Some(timings),
_ => None,
})
.expect("a launch that failed at registration reports its timings");
assert_eq!(timings.lock_wait, Duration::from_secs(90));
assert_eq!(timings.order_wait, Duration::from_secs(2));
assert_eq!(timings.spawn, Duration::ZERO, "spawn was never reached");
assert!(timings.total() >= Duration::from_secs(92));
}
#[cfg(target_os = "macos")]
#[tokio::test]
async fn a_launch_removes_the_seed_of_a_root_no_longer_in_use() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let old_root = harness._root.path().join("old-root");
let old_runtime = old_root.join("abandoned");
fs::create_dir_all(old_root.join(PACKAGE_SEED_DIR).join("2.335.0")).unwrap();
fs::create_dir_all(&old_runtime).unwrap();
let old = RunnerAttempt::allocate(
AttemptId::new_random(),
harness.policy.id,
&old_runtime,
harness.clock.now(),
);
harness.store.record_attempt(&old).unwrap();
let kept = harness.host_root().join(PACKAGE_SEED_DIR);
fs::create_dir_all(kept.join("2.336.0")).unwrap();
harness.launch().await;
assert!(
!old_root.join(PACKAGE_SEED_DIR).exists(),
"the abandoned root's seed is removed"
);
assert!(old_runtime.exists(), "only the seed is touched");
assert!(kept.exists(), "the host root's own seed is kept");
}
#[tokio::test]
async fn a_job_walks_every_state_and_cleans_every_artifact() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
assert_eq!(started.state(), AttemptState::Starting);
assert_eq!(read_runner_id(started.runtime_path()), Some(73));
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
let cleaned = harness.only_attempt();
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
assert!(!started.runtime_path().exists());
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
assert_eq!(harness.github.remaining_runners.load(Ordering::SeqCst), 0);
let states: Vec<_> = harness
.events
.events()
.into_iter()
.filter_map(|event| match event {
AttemptEvent::State { state, .. } => Some(state),
_ => None,
})
.collect();
assert_eq!(
states,
vec![
AttemptState::Allocated,
AttemptState::JitReceived,
AttemptState::Starting,
AttemptState::Idle,
AttemptState::Busy,
AttemptState::Finished,
AttemptState::Cleaned,
]
);
}
#[tokio::test]
async fn a_cleaned_ephemeral_attempt_reaps_a_directory_recreated_after_cleanup() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
let runtime = started.runtime_path().to_path_buf();
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.attempt(started.id).state(), AttemptState::Cleaned);
assert!(!runtime.exists());
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
let residue = runtime.join("_work").join("late-node-process");
fs::create_dir_all(&residue).unwrap();
fs::write(residue.join("node_modules.lock"), b"late residue").unwrap();
harness.launcher.supervise(&harness.policy).await.unwrap();
assert!(!runtime.exists(), "late ephemeral residue is reaped");
assert_eq!(harness.attempt(started.id).state(), AttemptState::Cleaned);
assert_eq!(
harness.packages.releases.load(Ordering::SeqCst),
1,
"reaping residue does not release the package lease twice"
);
}
#[tokio::test]
async fn locked_late_residue_never_blocks_startup_recovery() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
let runtime = started.runtime_path().to_path_buf();
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.attempt(started.id).state(), AttemptState::Cleaned);
let held = runtime.join("_work").join("late-node-process");
fs::create_dir_all(&held).unwrap();
fs::write(held.join("node_modules.lock"), b"late residue").unwrap();
let Some(block) = BlockedDeletion::inject(&held) else {
eprintln!(
"skipped: this account cannot be refused a deletion, so locked late residue cannot be injected"
);
return;
};
let restarted = harness.restart();
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("late residue cannot take the daemon offline");
assert!(runtime.exists(), "the locked residue remains for a retry");
assert!(
harness
.reconcile_events
.events()
.iter()
.any(|event| matches!(
event,
LifecycleEvent::AttemptCleanFailed {
attempt,
reason: "late_ephemeral_workspace_could_not_be_removed",
..
} if *attempt == started.id
)),
"the non-blocking cleanup failure remains visible"
);
block.release();
harness.clock.advance_secs(31);
restarted
.supervise(&harness.policy)
.await
.expect("the ordinary retry succeeds after the lock is released");
assert!(!runtime.exists(), "the retry removes the late residue");
assert_eq!(harness.attempt(started.id).state(), AttemptState::Cleaned);
assert_eq!(
harness.packages.releases.load(Ordering::SeqCst),
1,
"retrying residue never releases the package lease twice"
);
}
#[tokio::test]
async fn mixed_locked_cleanup_residue_isolated_from_the_whole_startup_pass() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let terminal = harness.launch().await;
let terminal_runtime = terminal.runtime_path().to_path_buf();
harness.conclude(terminal.id);
let Some(terminal_block) = BlockedDeletion::inject(&terminal_runtime.join("late-child"))
else {
eprintln!("skipped: this account cannot inject the two independent deletion refusals");
return;
};
let cleaned = harness.launch().await;
let cleaned_runtime = cleaned.runtime_path().to_path_buf();
harness.conclude(cleaned.id);
harness
.launcher
.clean(cleaned.id)
.await
.expect("the second attempt initially cleans");
let Some(cleaned_block) =
BlockedDeletion::inject(&cleaned_runtime.join("_work").join("late-child"))
else {
terminal_block.release();
eprintln!("skipped: this account cannot inject the two independent deletion refusals");
return;
};
let restarted = harness.restart();
for _ in 0..2 {
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("repeated recovery remains ready while both residues are locked");
}
assert_ne!(
harness.attempt(terminal.id).state(),
AttemptState::Cleaned,
"the journal does not claim the terminal workspace was removed"
);
assert_eq!(
harness.attempt(cleaned.id).state(),
AttemptState::Cleaned,
"late residue does not undo an already durable cleaned transition"
);
assert!(terminal_runtime.exists());
assert!(cleaned_runtime.exists());
assert_eq!(
harness.packages.releases.load(Ordering::SeqCst),
1,
"repeated recovery neither loses nor duplicates package leases"
);
terminal_block.release();
cleaned_block.release();
harness.clock.advance_secs(31);
restarted
.supervise(&harness.policy)
.await
.expect("one ordinary pass clears both residues after their locks disappear");
assert!(!terminal_runtime.exists());
assert!(!cleaned_runtime.exists());
assert_eq!(harness.attempt(terminal.id).state(), AttemptState::Cleaned);
assert_eq!(harness.attempt(cleaned.id).state(), AttemptState::Cleaned);
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn a_terminal_attempt_whose_runtime_is_already_gone_recovers_cleanly() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
let runtime = attempt.runtime_path().to_path_buf();
harness.conclude(attempt.id);
fs::remove_dir_all(&runtime).unwrap();
harness
.restart()
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("an already absent disposable runtime is successful cleanup");
assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn a_non_workspace_cleanup_failure_still_fails_closed_and_can_retry() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
let runtime = attempt.runtime_path().to_path_buf();
harness.conclude(attempt.id);
harness.packages.fail_releases(1);
let restarted = harness.restart();
let failure = restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect_err("a package-accounting failure is not safe to downgrade to residue");
assert!(
failure
.reason()
.to_string()
.contains("package lease could not be released")
);
assert!(
!runtime.exists(),
"workspace removal completed before release failed"
);
assert_ne!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("the same journal entry retries safely after the transient failure");
assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn startup_never_scrubs_a_cleaned_persistent_work_directory() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let attempt = harness.launch().await;
let retained = attempt.runtime_path().join("_work").join("checkout-marker");
fs::create_dir_all(retained.parent().unwrap()).unwrap();
fs::write(&retained, b"persistent checkout").unwrap();
harness.cleanup_retaining_work(attempt.id).await;
harness
.restart()
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("a cleaned persistent slot is already reconciled");
assert_eq!(fs::read(&retained).unwrap(), b"persistent checkout");
assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn an_idle_exit_is_not_a_failure_in_the_journal_or_events() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
let cleaned = harness.only_attempt();
assert!(cleaned.outcome().unwrap().is_idle_exit());
assert!(!cleaned.outcome().unwrap().is_failure());
assert!(!started.runtime_path().exists());
assert!(
harness
.reconcile_events
.events()
.iter()
.any(|event| matches!(
event,
LifecycleEvent::AttemptCleaned {
outcome: OutcomeKind::IdleExit,
..
}
))
);
assert!(!harness.events.events().iter().any(|event| matches!(
event,
AttemptEvent::Concluded {
outcome: OutcomeKind::Failed,
..
}
)));
}
#[tokio::test]
async fn handoff_is_absent_after_success_and_every_failed_spawn_retry() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.processes.fail_spawns(2);
harness.ready().await;
let attempt = harness.launch().await;
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 3);
assert!(harness.processes.saw_secret.load(Ordering::SeqCst));
let names: Vec<_> = fs::read_dir(attempt.runtime_path())
.unwrap()
.map(|entry| entry.unwrap().file_name())
.collect();
assert!(
names.iter().all(|name| {
!name
.to_string_lossy()
.starts_with(RestrictiveHandoff::NAME_PREFIX)
}),
"JIT artifact survived: {names:?}"
);
assert_eq!(
*harness.delay.0.lock().unwrap(),
vec![Duration::from_millis(10), Duration::from_millis(20)]
);
}
#[tokio::test]
async fn jit_retry_stops_with_demand_and_a_terminal_403_never_retries() {
let gone = Harness::new(
FakeGithubLifecycle::default().fail(false),
Arc::new(FakeDemand::answering([false])),
);
gone.ready().await;
assert!(gone.launch_result().await.is_err());
assert_eq!(gone.github.registrations.load(Ordering::SeqCst), 1);
assert!(gone.delay.0.lock().unwrap().is_empty());
let forbidden = Harness::new(
FakeGithubLifecycle::default().fail(true),
Arc::new(PersistentDemand),
);
forbidden.ready().await;
assert!(forbidden.launch_result().await.is_err());
assert_eq!(forbidden.github.registrations.load(Ordering::SeqCst), 1);
assert!(forbidden.delay.0.lock().unwrap().is_empty());
assert!(matches!(
forbidden.only_attempt().outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::Other(action)
}) if action.contains("403")
));
let transient = Harness::new(
FakeGithubLifecycle::default().fail(false).fail(false),
Arc::new(PersistentDemand),
);
transient.ready().await;
transient.launch().await;
assert_eq!(transient.github.registrations.load(Ordering::SeqCst), 3);
assert_eq!(
*transient.delay.0.lock().unwrap(),
vec![Duration::from_millis(10), Duration::from_millis(20)]
);
}
#[test]
fn a_workspace_leaves_room_for_the_deepest_path_a_checkout_writes() {
const MAX_PATH: usize = 260;
let root = r"C:\Users\IvanD\AppData\Local\IvanMurzak\runner-manager\data\runtime";
let repo = "GitHub-Runner-Scaler-UI";
let deepest = format!(
r"_work\{repo}\{repo}\.git\objects\pack\pack-{}.keep",
"0".repeat(40)
);
let name = workspace_name(AttemptId::new_random());
assert_eq!(name.len(), WORKSPACE_NAME_LEN, "{name}");
assert!(
name.chars().all(|c| c.is_ascii_hexdigit()),
"a directory name must not carry the identifier's dashes: {name}"
);
let full = format!(r"{root}\{name}\{deepest}");
assert!(
full.len() < MAX_PATH,
"the deepest path a checkout writes must fit: {} characters, limit {MAX_PATH}",
full.len()
);
let old = format!(
r"{root}\{}\{}\{deepest}",
PolicyId::new_random(),
AttemptId::new_random()
);
assert!(
old.len() > MAX_PATH,
"the old layout is supposed to be the thing that did not fit: {} characters",
old.len()
);
}
#[tokio::test]
async fn two_attempts_never_share_a_workspace_even_after_failure() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let first = harness.launch().await;
fs::write(first.runtime_path().join("hostile-leftover"), b"first job").unwrap();
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert!(!first.runtime_path().exists());
let second = harness.launch().await;
assert_ne!(first.runtime_path(), second.runtime_path());
assert!(!second.runtime_path().join("hostile-leftover").exists());
fs::write(
second.runtime_path().join("hostile-on-failure"),
b"second job",
)
.unwrap();
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert!(
!second.runtime_path().exists(),
"failed workspace was retained"
);
}
#[tokio::test]
async fn a_runner_that_never_gets_a_job_is_stopped_deregistered_and_not_replaced() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
harness.clock.advance_secs(9);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
let none_yet = harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
assert!(none_yet.is_empty());
assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
harness.clock.advance_secs(1);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
assert_eq!(
concluded.outcome(),
Some(&AttemptOutcome::ExitedIdleWithoutWork),
"a surplus runner did not fail; recording one as a failure sends an operator \
hunting a fault that does not exist"
);
assert_eq!(concluded.state(), AttemptState::Cleaned);
assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
assert!(!attempt.runtime_path().exists());
assert_eq!(
*harness.github.deregistrations.lock().unwrap(),
vec![73],
"the attempt's own runner id, deleted exactly once"
);
assert!(
replacements.is_empty(),
"a surplus exit must not request a replacement"
);
}
#[tokio::test]
async fn a_registration_github_will_not_delete_still_concludes_the_attempt() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness
.github
.deregistration_fails
.store(true, Ordering::SeqCst);
harness.clock.advance_secs(11);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
*harness.github.deregistrations.lock().unwrap(),
vec![73],
"the delete was attempted"
);
let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
assert_eq!(
concluded.outcome(),
Some(&AttemptOutcome::ExitedIdleWithoutWork),
"the attempt concluded anyway"
);
assert_eq!(concluded.state(), AttemptState::Cleaned);
assert!(!attempt.runtime_path().exists());
}
#[tokio::test]
async fn exit_before_acceptance_returns_replacement_intent_without_launching() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let first = harness.launch().await;
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
let failed = harness.store.attempt(first.id).unwrap().unwrap();
assert!(matches!(
failed.outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::ProcessExitedUnexpectedly
})
));
assert!(!first.runtime_path().exists());
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: first.id,
operation: "exit_before_acceptance_replacement",
}]
);
assert_eq!(harness.store.attempts().unwrap().len(), 1);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
assert!(harness.delay.0.lock().unwrap().is_empty());
}
fn write_runner_diagnostics(runtime: &Path, ran_a_job: bool) {
let diag = runtime.join("_diag");
fs::create_dir_all(&diag).unwrap();
fs::write(diag.join("Runner_20261007-211941-utc.log"), b"listener").unwrap();
if ran_a_job {
fs::write(diag.join("Worker_20261007-211950-utc.log"), b"worker").unwrap();
}
}
#[tokio::test]
async fn a_job_too_short_to_be_seen_busy_is_recorded_as_completed() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
assert_eq!(started.state(), AttemptState::Starting);
write_runner_diagnostics(started.runtime_path(), true);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
let cleaned = harness.only_attempt();
assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert_eq!(cleaned.github_runner_id(), Some(73));
assert!(
replacements.is_empty(),
"a finished job needs no replacement"
);
let busy_seen = harness.events.events().into_iter().any(|event| {
matches!(
event,
AttemptEvent::State {
state: AttemptState::Busy,
..
}
)
});
assert!(busy_seen, "the journal walks the diagram to `busy` first");
}
#[tokio::test]
async fn a_job_taken_between_two_passes_after_idle_is_recorded_as_completed() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
write_runner_diagnostics(started.runtime_path(), true);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
harness.only_attempt().outcome(),
Some(&AttemptOutcome::CompletedJob)
);
assert!(replacements.is_empty());
}
#[tokio::test]
async fn an_assignment_the_runner_never_took_up_is_not_a_completed_job() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
write_runner_diagnostics(started.runtime_path(), false);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
let cleaned = harness.only_attempt();
assert_eq!(
cleaned.outcome(),
Some(&AttemptOutcome::ExitedIdleWithoutWork)
);
assert_eq!(cleaned.state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn a_finished_job_still_listed_at_github_is_completed_and_deregistered() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
write_runner_diagnostics(started.runtime_path(), true);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
harness.only_attempt().outcome(),
Some(&AttemptOutcome::CompletedJob)
);
assert_eq!(*harness.github.deregistrations.lock().unwrap(), vec![73]);
}
#[tokio::test]
async fn a_live_runner_whose_assignment_was_withdrawn_goes_back_to_idle() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
write_runner_diagnostics(started.runtime_path(), false);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
harness.launcher.supervise(&harness.policy).await.unwrap();
write_runner_diagnostics(started.runtime_path(), true);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
}
#[tokio::test]
async fn a_clean_exit_waits_for_github_to_answer() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
write_runner_diagnostics(started.runtime_path(), true);
harness.processes.finish_successfully();
harness.github.observe(GithubRunnerObservation::Unreachable);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.only_attempt().state(), AttemptState::Starting);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
harness.only_attempt().outcome(),
Some(&AttemptOutcome::CompletedJob)
);
}
#[tokio::test]
async fn a_short_job_with_no_runner_id_still_concludes() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let started = harness.launch().await;
fs::remove_file(started.runtime_path().join(RUNNER_ID_FILE)).unwrap();
write_runner_diagnostics(started.runtime_path(), true);
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness
.launcher
.supervise(&harness.policy)
.await
.expect("supervision goes on");
assert!(harness.only_attempt().is_terminal());
}
#[tokio::test]
async fn without_runner_diagnostics_an_early_exit_is_still_a_failure() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
harness.launch().await;
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
assert!(matches!(
harness.only_attempt().outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::ProcessExitedUnexpectedly
})
));
}
#[tokio::test]
async fn expired_jit_is_removed_and_does_not_reregister_after_demand_disappears() {
let harness = Harness::new(
FakeGithubLifecycle::default(),
Arc::new(FakeDemand::answering([false])),
);
let id = AttemptId::new_random();
let runtime = harness
.launcher
.app_paths
.runtime_dir()
.join(harness.policy.id.to_string())
.join(id.to_string());
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.clock.advance_secs(11);
let replacements = harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: id,
operation: "jit_expired_replacement",
}]
);
let cleaned = harness.store.attempt(id).unwrap().unwrap();
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert!(matches!(
cleaned.outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::JitExpired
})
));
assert!(!runtime.exists());
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert!(harness.delay.0.lock().unwrap().is_empty());
}
#[tokio::test]
async fn expired_jit_returns_intent_but_never_launches_inside_lifecycle() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = harness
.launcher
.app_paths
.runtime_dir()
.join("expired-with-demand");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.clock.advance_secs(11);
let replacements = harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let attempts = harness.store.attempts().unwrap();
assert_eq!(attempts.len(), 1);
assert_eq!(
attempts
.iter()
.find(|attempt| attempt.id == id)
.unwrap()
.state(),
AttemptState::Cleaned
);
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: id,
operation: "jit_expired_replacement",
}]
);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
assert!(harness.delay.0.lock().unwrap().is_empty());
}
#[tokio::test]
async fn package_materialization_retries_are_bounded_and_demand_adjacent() {
let persistent = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
persistent.packages.fail_materializations(2);
persistent.ready().await;
persistent.launch().await;
assert_eq!(
persistent.packages.materializations.load(Ordering::SeqCst),
3
);
assert_eq!(
*persistent.delay.0.lock().unwrap(),
vec![Duration::from_millis(10), Duration::from_millis(20)]
);
let gone_before_wait = Harness::new(
FakeGithubLifecycle::default(),
Arc::new(FakeDemand::answering([false])),
);
gone_before_wait.packages.fail_materializations(3);
gone_before_wait.ready().await;
assert!(gone_before_wait.launch_result().await.is_err());
assert_eq!(
gone_before_wait
.packages
.materializations
.load(Ordering::SeqCst),
1
);
assert!(gone_before_wait.delay.0.lock().unwrap().is_empty());
let gone_during_wait = Harness::new(
FakeGithubLifecycle::default(),
Arc::new(FakeDemand::answering([true, false])),
);
gone_during_wait.packages.fail_materializations(3);
gone_during_wait.ready().await;
assert!(gone_during_wait.launch_result().await.is_err());
assert_eq!(
gone_during_wait
.packages
.materializations
.load(Ordering::SeqCst),
1
);
assert_eq!(
*gone_during_wait.delay.0.lock().unwrap(),
vec![Duration::from_millis(10)]
);
}
#[tokio::test]
async fn replacement_is_intent_only_and_never_launches_inside_lifecycle() {
let harness = Harness::new(
FakeGithubLifecycle::default(),
Arc::new(FakeDemand::answering([true, false])),
);
harness.ready().await;
let first = harness.launch().await;
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(harness.store.attempts().unwrap().len(), 1);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
assert!(harness.delay.0.lock().unwrap().is_empty());
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: first.id,
operation: "exit_before_acceptance_replacement",
}]
);
assert_eq!(
harness.store.attempt(first.id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
#[tokio::test]
async fn startup_adopts_a_live_process_and_refuses_launch_before_recovery() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let before = harness.launch_result().await;
assert!(before.is_err());
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
let id = AttemptId::new_random();
let runtime = harness.launcher.app_paths.runtime_dir().join("adopt");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.processes.set_alive(true);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
assert!(replacements.is_empty());
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
assert!(
harness
.events
.events()
.contains(&AttemptEvent::Adopted { attempt: id })
);
}
#[tokio::test]
async fn spawn_before_starting_crash_recovers_pid_then_completes_and_cleans() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = harness
.launcher
.app_paths
.runtime_dir()
.join("spawn-before-starting");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.processes.set_alive(true);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: true });
let replacements = harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
assert!(replacements.is_empty());
let recovered = harness.store.attempt(id).unwrap().unwrap();
assert_eq!(recovered.state(), AttemptState::Busy);
assert_eq!(recovered.process_id(), Some(4242));
assert_eq!(recovered.github_runner_id(), Some(73));
let events = harness.events.events();
let starting = events
.iter()
.position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Starting } if *attempt == id))
.unwrap();
let busy = events
.iter()
.position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Busy } if *attempt == id))
.unwrap();
assert!(starting < busy, "recovery skipped a legal edge: {events:?}");
harness.processes.finish_successfully();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
assert!(
harness
.launcher
.supervise(&harness.policy)
.await
.unwrap()
.is_empty()
);
let cleaned = harness.store.attempt(id).unwrap().unwrap();
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
assert!(!runtime.exists());
}
#[tokio::test]
async fn failed_post_spawn_stop_keeps_capacity_until_supervision_proves_death() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.processes.fail_spawn_with_live_child();
harness.ready().await;
assert!(harness.launch_result().await.is_err());
let attempt = harness.only_attempt();
assert_eq!(attempt.state(), AttemptState::Starting);
assert_eq!(attempt.process_id(), Some(4242));
assert!(attempt.outcome().is_none());
assert!(attempt.state().counts_against_capacity());
assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
assert!(harness.delay.0.lock().unwrap().is_empty());
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(replacements.len(), 1);
assert_eq!(
harness.store.attempt(attempt.id).unwrap().unwrap().state(),
AttemptState::Cleaned
);
}
#[tokio::test]
async fn remote_runner_identity_closes_both_sides_of_the_registration_crash_boundary() {
for sidecar_already_present in [false, true] {
let harness = Harness::new(
FakeGithubLifecycle::default(),
Arc::new(FakeDemand::answering([false])),
);
let id = AttemptId::new_random();
let runtime =
harness
.launcher
.app_paths
.runtime_dir()
.join(if sidecar_already_present {
"after-id-sidecar"
} else {
"before-id-sidecar"
});
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
if sidecar_already_present {
write_runner_id(&runtime, 73).unwrap();
attempt.jit_received(harness.clock.now()).unwrap();
}
harness.store.record_attempt(&attempt).unwrap();
harness.processes.set_alive(true);
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
assert_eq!(read_runner_id(&runtime), Some(73));
assert!(
harness
.store
.attempt(id)
.unwrap()
.unwrap()
.outcome()
.is_none()
);
let events = harness.events.events();
let recovered = events.iter().position(|event| {
matches!(
event,
AttemptEvent::RemoteIdentityRecovered {
attempt,
runner_id: 73
} if *attempt == id
)
});
assert_eq!(recovered.is_some(), !sidecar_already_present);
if let Some(recovered) = recovered {
let adopted = events
.iter()
.position(|event| matches!(event, AttemptEvent::Adopted { attempt } if *attempt == id))
.unwrap();
assert!(
recovered < adopted,
"identity was not durable before adoption: {events:?}"
);
}
assert!(runtime.exists());
}
}
#[tokio::test]
async fn recovery_stays_closed_for_unknown_policy_and_unreachable_attempts() {
let unknown = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let unknown_attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
PolicyId::from_u128(0xfeed),
unknown
.launcher
.app_paths
.runtime_dir()
.join("unknown-policy"),
unknown.clock.now(),
);
unknown.store.record_attempt(&unknown_attempt).unwrap();
let expired_id = AttemptId::new_random();
let expired_runtime = unknown
.launcher
.app_paths
.runtime_dir()
.join("expired-beside-unknown");
fs::create_dir_all(&expired_runtime).unwrap();
let mut expired = RunnerAttempt::allocate(
expired_id,
unknown.policy.id,
expired_runtime,
unknown.clock.now(),
);
expired.jit_received(unknown.clock.now()).unwrap();
unknown.store.record_attempt(&expired).unwrap();
unknown.clock.advance_secs(11);
assert!(matches!(
unknown
.launcher
.recover_startup(std::slice::from_ref(&unknown.policy))
.await,
Err(LifecycleError::RecoveryIncomplete)
));
assert!(unknown.launch_result().await.is_err());
assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
let recovered_policy = fixtures::policy()
.id(PolicyId::from_u128(0xfeed))
.repository("octo/repo")
.autoscale("home", 2)
.active()
.build();
let pending = unknown
.launcher
.recover_startup(&[unknown.policy.clone(), recovered_policy])
.await
.unwrap();
assert_eq!(
pending,
vec![ReplacementIntent {
policy: unknown.policy.id,
previous_attempt: expired_id,
operation: "jit_expired_replacement",
}]
);
assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
let unreachable = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = unreachable
.launcher
.app_paths
.runtime_dir()
.join("unreachable");
fs::create_dir_all(&runtime).unwrap();
unreachable
.store
.record_attempt(&RunnerAttempt::allocate(
id,
unreachable.policy.id,
runtime,
unreachable.clock.now(),
))
.unwrap();
unreachable
.github
.observe(GithubRunnerObservation::Unreachable);
assert!(matches!(
unreachable
.launcher
.recover_startup(std::slice::from_ref(&unreachable.policy))
.await,
Err(LifecycleError::RecoveryIncomplete)
));
assert!(unreachable.launch_result().await.is_err());
assert_eq!(unreachable.processes.spawns.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn a_dead_busy_process_unknown_to_github_is_orphaned_and_cleaned() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = harness.launcher.app_paths.runtime_dir().join("orphan");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
attempt.assigned_job(73, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let cleaned = harness.store.attempt(id).unwrap().unwrap();
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::Orphaned));
assert!(!runtime.exists());
}
#[tokio::test]
async fn registration_timeout_journals_intent_stops_then_concludes_with_dead_reason() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let id = AttemptId::new_random();
let runtime = harness.launcher.app_paths.runtime_dir().join("timeout");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(true);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: id,
operation: "registration_timeout_replacement",
}]
);
assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
assert!(!harness.processes.alive.load(Ordering::SeqCst));
let actions = harness.processes.actions.lock().unwrap().clone();
let intent = actions
.iter()
.position(|action| *action == "terminate_intent")
.unwrap();
let signal = actions
.iter()
.position(|action| *action == "terminate")
.unwrap();
assert!(
intent < signal,
"intent was not durable before signal: {actions:?}"
);
let cleaned = harness.store.attempt(id).unwrap().unwrap();
assert!(matches!(
cleaned.outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::TerminatedAfterRegistrationTimeout
})
));
let events = harness.events.events();
let intent = events
.iter()
.position(|event| matches!(event, AttemptEvent::TerminateIntent { .. }))
.unwrap();
let stopped = events
.iter()
.position(|event| matches!(event, AttemptEvent::Terminated { .. }))
.unwrap();
let concluded = events
.iter()
.position(|event| matches!(event, AttemptEvent::Concluded { .. }))
.unwrap();
assert!(intent < stopped && stopped < concluded, "{events:?}");
}
#[tokio::test]
async fn timeout_crash_recovery_returns_the_same_replacement_intent() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = harness
.launcher
.app_paths
.runtime_dir()
.join("timeout-after-crash");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.processes.intent.store(true, Ordering::SeqCst);
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
let replacements = harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
assert_eq!(
replacements,
vec![ReplacementIntent {
policy: harness.policy.id,
previous_attempt: id,
operation: "registration_timeout_replacement",
}]
);
let consumed = RunnerLauncher::supervise(&harness.launcher, &harness.policy)
.await
.unwrap();
assert_eq!(consumed, replacements);
assert!(
RunnerLauncher::supervise(&harness.launcher, &harness.policy)
.await
.unwrap()
.is_empty(),
"startup replacement evidence must be consumed exactly once by e1"
);
assert!(matches!(
harness.store.attempt(id).unwrap().unwrap().outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::TerminatedAfterRegistrationTimeout
})
));
}
#[tokio::test]
async fn terminate_intent_sync_failure_prevents_signal_and_conclusion() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
let id = AttemptId::new_random();
let runtime = harness
.launcher
.app_paths
.runtime_dir()
.join("timeout-sync-failure");
fs::create_dir_all(&runtime).unwrap();
let mut attempt =
RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(true);
harness.processes.fail_intent();
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
assert!(
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.is_err()
);
assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
assert!(harness.processes.alive.load(Ordering::SeqCst));
assert_eq!(
harness.store.attempt(id).unwrap().unwrap().state(),
AttemptState::Starting
);
assert!(!harness.events.events().iter().any(|event| matches!(
event,
AttemptEvent::Terminated { attempt } | AttemptEvent::Concluded { attempt, .. }
if *attempt == id
)));
}
#[tokio::test]
async fn diagnostics_survive_cleanup_without_the_jit_or_a_token() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
harness
.github
.observe(GithubRunnerObservation::Registered { busy: false });
harness.launcher.supervise(&harness.policy).await.unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(false);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
let diagnostic = fs::read_to_string(
harness
.launcher
.diagnostics_root
.join(format!("{}.log", attempt.id)),
)
.unwrap();
assert!(diagnostic.contains("exited_idle_without_work"));
assert!(!diagnostic.contains(JIT));
assert!(!diagnostic.contains("ghp_"));
assert!(!attempt.runtime_path().exists());
}
#[test]
fn native_process_listing_never_contains_jit_and_handoffs_never_survive() {
let root = tempfile::tempdir().unwrap();
let policy = fixtures::policy()
.repository("octo/repo")
.autoscale("home", 1)
.active()
.build();
let runtime = root.path().join("successful");
fs::create_dir_all(&runtime).unwrap();
let processes = NativeProcesses::new();
let config = EncodedJitConfig::new(JIT);
let handoff =
RestrictiveHandoff::create(&runtime, SecretString::from(config.expose().to_owned()))
.unwrap();
let mut child = native_inspection_spec()
.spawn_runner_with_handoff(&handoff)
.expect("native child starts");
let pid = child.pid();
handoff.delete().unwrap();
let command_line = native_command_line(pid);
assert!(
!command_line.contains(JIT),
"the encoded JIT configuration appeared in the native process listing"
);
assert_no_jit_file(&runtime);
child
.stop(Duration::from_secs(1))
.expect("native child stops");
let failed_runtime = root.path().join("failed");
fs::create_dir_all(&failed_runtime).unwrap();
let failed = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&failed_runtime,
FakeClock::default().now(),
);
assert!(
processes
.spawn(&failed, &EncodedJitConfig::new(JIT))
.is_err(),
"a runtime with no runner executable must fail"
);
assert_no_jit_file(&failed_runtime);
processes
.record_terminate_intent(&failed)
.expect("the intent file and its directory entry are durably synced");
assert_eq!(
fs::read(NativeProcesses::intent_path(&failed)).unwrap(),
b"registration-timeout\n"
);
}
#[test]
fn post_spawn_boundaries_are_bounded_durable_and_never_retry_jit() {
let root = tempfile::tempdir().unwrap();
let policy = fixtures::policy()
.repository("octo/repo")
.autoscale("home", 1)
.active()
.build();
let processes = NativeProcesses::new();
processes.use_long_lived_test_listener();
for (index, boundary) in [
PostSpawnBoundary::HandoffDelete,
PostSpawnBoundary::IdentitySerialize,
PostSpawnBoundary::IdentityWrite,
PostSpawnBoundary::ChildMapInsert,
]
.into_iter()
.enumerate()
{
let runtime = root.path().join(format!("post-spawn-{index}"));
let bin = runtime.join("bin");
fs::create_dir_all(&bin).unwrap();
#[cfg(windows)]
let listener = bin.join("Runner.Listener.exe");
#[cfg(not(windows))]
let listener = bin.join("Runner.Listener");
fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
let attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&runtime,
FakeClock::default().now(),
);
processes.fail_post_spawn_at(boundary);
let failure = processes
.spawn(&attempt, &EncodedJitConfig::new(JIT))
.expect_err("fault must cross the post-spawn cleanup path");
assert!(!failure.retryable, "{boundary:?} allowed duplicate retry");
assert!(
!processes.is_alive(&attempt).unwrap(),
"{boundary:?} left a child"
);
assert!(!NativeProcesses::identity_path(&attempt).exists());
assert_no_jit_file(&runtime);
}
assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
let runtime = root.path().join("identity-and-stop-fail");
let bin = runtime.join("bin");
fs::create_dir_all(&bin).unwrap();
#[cfg(windows)]
let listener = bin.join("Runner.Listener.exe");
#[cfg(not(windows))]
let listener = bin.join("Runner.Listener");
fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
let attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&runtime,
FakeClock::default().now(),
);
processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
processes.fail_next_post_spawn_stop();
let failure = processes
.spawn(&attempt, &EncodedJitConfig::new(JIT))
.expect_err("the identity boundary must fail closed");
assert!(failure.live_pid.is_some());
assert_long_lived_listener_ready(&processes, &attempt);
assert!(processes.is_alive(&attempt).unwrap());
assert!(!NativeProcesses::identity_path(&attempt).exists());
assert!(NativeProcesses::fallback_identity_path(&attempt).is_file());
assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
processes.terminate(&attempt).unwrap();
let runtime = root.path().join("persistent-stop-and-identity-failures");
let bin = runtime.join("bin");
fs::create_dir_all(&bin).unwrap();
#[cfg(windows)]
let listener = bin.join("Runner.Listener.exe");
#[cfg(not(windows))]
let listener = bin.join("Runner.Listener");
fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
let mut unresolved = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&runtime,
FakeClock::default().now(),
);
for _ in 0..3 {
processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
}
processes.fail_post_spawn_stops(MAX_POST_SPAWN_STOP_ATTEMPTS);
let failure = processes
.spawn(&unresolved, &EncodedJitConfig::new(JIT))
.expect_err("bounded cleanup must return even when every stop errors");
let pid = failure
.live_pid
.expect("the owned child remains supervised in this invocation");
assert!(matches!(failure.reason, FailureReason::Other(_)));
assert_long_lived_listener_ready(&processes, &unresolved);
unresolved.jit_received(FakeClock::default().now()).unwrap();
unresolved.started(pid, FakeClock::default().now()).unwrap();
let journal = SqliteStore::open_in_memory().unwrap();
journal.record_attempt(&unresolved).unwrap();
let recovered = journal.attempt(unresolved.id).unwrap().unwrap();
assert_eq!(recovered.process_id(), Some(pid));
assert_eq!(recovered.state(), AttemptState::Starting);
assert!(processes.is_alive(&unresolved).unwrap());
assert!(!NativeProcesses::identity_path(&unresolved).exists());
assert!(!NativeProcesses::fallback_identity_path(&unresolved).exists());
assert_eq!(
fs::read_to_string(NativeProcesses::unresolved_process_path(&unresolved)).unwrap(),
pid.to_string(),
"bounded cleanup must leave durable unresolved-process evidence before returning"
);
assert!(
NativeProcesses::new().is_alive(&recovered).is_err(),
"restart must fail closed on the durable starting/PID journal rather than trust a bare PID"
);
processes.terminate(&unresolved).unwrap();
let runtime = root.path().join("post-spawn-stop-failed");
let bin = runtime.join("bin");
fs::create_dir_all(&bin).unwrap();
#[cfg(windows)]
let listener = bin.join("Runner.Listener.exe");
#[cfg(not(windows))]
let listener = bin.join("Runner.Listener");
fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
let attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&runtime,
FakeClock::default().now(),
);
processes.fail_post_spawn_at(PostSpawnBoundary::ChildMapInsert);
processes.fail_next_post_spawn_stop();
let failure = processes
.spawn(&attempt, &EncodedJitConfig::new(JIT))
.expect_err("the injected stop failure must preserve supervision");
let live_pid = failure
.live_pid
.expect("live PID is returned to the journal");
assert!(!failure.retryable);
assert_long_lived_listener_ready(&processes, &attempt);
assert!(NativeProcesses::identity_path(&attempt).is_file());
assert_eq!(
NativeProcesses::read_identity(&attempt)
.unwrap()
.unwrap()
.pid(),
live_pid
);
assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
processes.terminate(&attempt).unwrap();
}
#[test]
#[ignore = "spawned only as the platform-stable native listener fixture"]
fn long_lived_native_listener_helper() {
let ready = std::env::var_os("RUNNER_MANAGER_TEST_LISTENER_READY")
.map(PathBuf::from)
.expect("the parent supplies the readiness path");
runner_manager_platform::host_fitness::write_atomically(&ready, b"ready\n")
.expect("the listener publishes readiness");
std::thread::sleep(Duration::from_secs(30));
}
fn assert_long_lived_listener_ready(processes: &NativeProcesses, attempt: &RunnerAttempt) {
let ready = attempt.runtime_path().join(TEST_LISTENER_READY);
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
if ready.is_file() {
assert_eq!(fs::read(&ready).unwrap(), b"ready\n");
return;
}
assert!(
processes.is_alive(attempt).unwrap(),
"the native listener exited before publishing readiness"
);
assert!(
std::time::Instant::now() < deadline,
"the native listener stayed alive but never published readiness"
);
std::thread::sleep(Duration::from_millis(10));
}
}
#[tokio::test]
async fn a_reservation_is_journalled_before_any_effect_and_holds_its_slot() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let reserved = harness.reserve().await.unwrap();
assert_eq!(reserved.state(), AttemptState::Allocated);
assert!(reserved.counts_against_capacity());
let journalled = harness.store.attempt(reserved.id).unwrap().unwrap();
assert_eq!(journalled.state(), AttemptState::Allocated);
assert_eq!(harness.packages.materializations.load(Ordering::SeqCst), 0);
assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
let started = harness
.launcher
.finish_launch(&harness.policy, reserved, ReservationWaits::default())
.await
.unwrap();
assert_eq!(started.state(), AttemptState::Starting);
}
#[tokio::test]
async fn a_pass_leaves_a_runner_coming_up_alone_and_supervises_it_once_up() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let reserved = harness.reserve().await.unwrap();
let looks_left = || harness.github.observations.lock().unwrap().len();
harness
.github
.observations
.lock()
.unwrap()
.push_back(LifecycleGithubObservation::not_registered());
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(looks_left(), 1, "it did not look");
let journalled = harness.store.attempt(reserved.id).unwrap().unwrap();
assert_eq!(journalled.state(), AttemptState::Allocated);
harness
.launcher
.finish_launch(&harness.policy, reserved, ReservationWaits::default())
.await
.unwrap();
assert!(harness.launcher.bringing_up.lock().unwrap().is_empty());
harness.launcher.supervise(&harness.policy).await.unwrap();
assert_eq!(looks_left(), 0, "up, it is this pass's to supervise");
}
#[tokio::test]
async fn a_crash_between_reservation_and_spawn_recovers_the_slot() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let reserved = harness.reserve().await.unwrap();
harness.clock.advance_secs(31);
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let recovered = harness.store.attempt(reserved.id).unwrap().unwrap();
assert!(
!recovered.counts_against_capacity(),
"the slot is free again: {:?}",
recovered.state()
);
}
#[tokio::test]
async fn a_crash_after_the_package_is_copied_recovers_the_slot() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let reserved = harness.reserve().await.unwrap();
harness
.packages
.materialize(&reserved, &mut PackageTimings::default())
.await
.unwrap();
harness.clock.advance_secs(31);
harness
.launcher
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.unwrap();
let recovered = harness.store.attempt(reserved.id).unwrap().unwrap();
assert!(!recovered.counts_against_capacity());
}
#[tokio::test]
async fn runners_prepare_at_once_within_the_bound() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let turns = harness.launcher.turns.available_permits();
harness.launcher.turns.forget_permits(turns - 2);
let mut reserved = Vec::new();
for _ in 0..4 {
reserved.push(harness.reserve().await.unwrap());
}
let finished = futures::future::join_all(reserved.into_iter().map(|attempt| {
harness
.launcher
.finish_launch(&harness.policy, attempt, ReservationWaits::default())
}))
.await;
assert!(finished.iter().all(Result::is_ok));
assert_eq!(
harness.packages.peak_materializing.load(Ordering::SeqCst),
2
);
}
#[tokio::test]
async fn every_production_launch_prunes_under_the_same_allocation_guard() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 0);
harness.launch().await;
assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 1);
}
fn copy_package_tree(
source: &Path,
destination: &Path,
seeds: &TrustedTrees,
) -> std::io::Result<()> {
let origin = copy_origin(source, destination, seeds);
copy_package_from(source, &origin, destination, seeds)
}
fn assert_no_jit_file(runtime: &Path) {
for entry in fs::read_dir(runtime).unwrap() {
let path = entry.unwrap().path();
if path.is_file() {
let bytes = fs::read(&path).unwrap();
assert!(
!bytes
.windows(JIT.len())
.any(|window| window == JIT.as_bytes()),
"a JIT payload survived in a runtime file"
);
}
}
}
#[test]
fn production_listener_command_uses_the_supported_jit_contract() {
let runtime = Path::new("runtime");
let spec = runner_listener_spec(
PathBuf::from("Runner.Listener"),
runtime,
Vec::new(),
&RunnerEnv::default(),
);
let arguments: Vec<_> = spec
.arguments()
.iter()
.map(|argument| argument.to_string_lossy().into_owned())
.collect();
assert_eq!(arguments, ["run"]);
assert!(
!arguments
.iter()
.any(|argument| argument == "--jit-config-file"),
"the obsolete file option would be rejected by Runner.Listener 2.336.0"
);
}
#[test]
fn production_listener_layers_runner_env_over_the_platform_defaults() {
let root = tempfile::tempdir().unwrap();
let runtime = root.path().join("attempt");
let host_env =
RunnerEnv::parse("ELECTRON_CACHE=/cache/electron\nHOME=/operator/chose/this\n")
.unwrap();
let caches = vec![
(
"ELECTRON_CACHE",
std::ffi::OsString::from("/rman/_cache/o/r/electron"),
),
(
"npm_config_cache",
std::ffi::OsString::from("/rman/_cache/o/r/npm"),
),
];
let spec = runner_listener_spec(
PathBuf::from("Runner.Listener"),
&runtime,
caches,
&host_env,
);
let env = |name: &str| spec.configured_env(name).map(PathBuf::from);
assert!(spec.expects_normal_scheduling());
assert_eq!(
env("ELECTRON_CACHE"),
Some(PathBuf::from("/cache/electron")),
"runner.env must win over a dependency cache of the same name"
);
assert_eq!(
env("npm_config_cache"),
Some(PathBuf::from("/rman/_cache/o/r/npm"))
);
assert_eq!(
env("HOME"),
Some(PathBuf::from("/operator/chose/this")),
"runner.env must win over a platform default of the same name"
);
for tmp in ["TMPDIR", "TEMP", "TMP"] {
assert_eq!(env(tmp), Some(runtime.join("tmp")), "{tmp}");
}
let home = runtime.join(runner_env::RUNNER_HOME_DIR);
if cfg!(windows) {
assert_eq!(env("USERPROFILE"), Some(home.clone()));
assert_eq!(env("APPDATA"), Some(home.join(r"AppData\Roaming")));
assert_eq!(env("LOCALAPPDATA"), Some(home.join(r"AppData\Local")));
assert!(home.join(r"AppData\Roaming").is_dir());
assert!(home.join(r"AppData\Local").is_dir());
} else {
assert_eq!(env("USERPROFILE"), None);
assert!(!home.exists());
}
}
#[test]
fn production_listener_opts_powershell_out_of_telemetry() {
let root = tempfile::tempdir().unwrap();
let runtime = root.path().join("attempt");
let listener = |host_env: &RunnerEnv| {
runner_listener_spec(
PathBuf::from("Runner.Listener"),
&runtime,
Vec::new(),
host_env,
)
};
let optout = |spec: &SpawnSpec| {
spec.configured_env(runner_env::POWERSHELL_TELEMETRY_OPTOUT)
.map(OsStr::to_os_string)
};
let spec = listener(&RunnerEnv::default());
if std::env::var_os(runner_env::POWERSHELL_TELEMETRY_OPTOUT)
.is_some_and(|value| !value.is_empty())
{
assert_eq!(optout(&spec), None);
} else {
assert_eq!(
optout(&spec),
Some(runner_env::POWERSHELL_TELEMETRY_OPTOUT_VALUE.into())
);
}
let operator = RunnerEnv::parse("POWERSHELL_TELEMETRY_OPTOUT=0\n").unwrap();
assert_eq!(
optout(&listener(&operator)),
Some("0".into()),
"runner.env must win over the telemetry opt-out"
);
}
#[test]
fn an_invalid_runner_env_fails_the_launch_without_echoing_a_value() {
let root = tempfile::tempdir().unwrap();
let path = runner_env::path_in(root.path());
fs::write(&path, "GOOD=1\nghp_notarealtokenbutshapedlikeone\n").unwrap();
let processes = NativeProcesses::new().with_runner_env_file(path.clone());
let failure = processes.host_env().unwrap_err().to_string();
assert!(
failure.contains("runner.env") && failure.contains("line 2"),
"{failure}"
);
assert!(
!failure.contains("ghp_"),
"the failure echoed a value: {failure}"
);
let policy = fixtures::policy()
.repository("octo/repo")
.autoscale("home", 1)
.active()
.build();
let attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
root.path(),
FakeClock::default().now(),
);
assert!(processes.prepare(&attempt, &policy).is_err());
fs::write(&path, "GOOD=1\n").unwrap();
assert!(processes.prepare(&attempt, &policy).is_ok());
assert_eq!(processes.host_env().unwrap().get("GOOD"), Some("1"));
assert!(NativeProcesses::new().host_env().unwrap().is_empty());
}
#[test]
fn prepare_hands_the_policys_cache_namespace_to_the_launch() {
let root = tempfile::tempdir().unwrap();
let runner_root = root.path().join("rman");
let runtime = runner_root.join("a1");
fs::create_dir_all(runtime.join("bin")).unwrap();
let config = runner_manager_platform::dependency_cache::config_path_in(root.path());
let source_root = runner_root.clone();
let processes = NativeProcesses::new().with_dependency_caches(DependencyCaches::new(
config.clone(),
Arc::new(move || Some(source_root.clone())),
None,
));
let policy = fixtures::policy()
.repository("Octo/Repo")
.autoscale("home", 1)
.active()
.build();
let attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
policy.id,
&runtime,
FakeClock::default().now(),
);
assert!(
processes
.cache_env(&attempt, None, &RunnerEnv::default())
.is_empty(),
"a launch with no prepared target, such as a bare `spawn`, has no cache"
);
let prepared = processes.prepare(&attempt, &policy).unwrap();
let variables = processes.cache_env(
&attempt,
prepared.cache_target.as_ref(),
&RunnerEnv::default(),
);
let npm = variables
.iter()
.find(|(name, _)| *name == "npm_config_cache")
.map(|(_, value)| PathBuf::from(value));
assert_eq!(
npm,
Some(
runner_root
.join("_cache")
.join("octo")
.join("repo")
.join("npm")
)
);
fs::write(&config, "[tools]\nnot-a-tool = true\n").unwrap();
let refused = processes
.prepare(&attempt, &policy)
.unwrap_err()
.to_string();
assert!(refused.contains("caches.toml"), "{refused}");
}
#[cfg(windows)]
fn native_inspection_spec() -> SpawnSpec {
SpawnSpec::new("powershell.exe").args([
"-NoProfile",
"-NonInteractive",
"-Command",
"Start-Sleep -Seconds 30",
])
}
#[cfg(unix)]
fn native_inspection_spec() -> SpawnSpec {
SpawnSpec::new("/bin/sh").args(["-c", "sleep 30"])
}
#[cfg(windows)]
fn native_command_line(pid: u32) -> String {
let output = std::process::Command::new("powershell.exe")
.args([
"-NoProfile",
"-NonInteractive",
"-Command",
&format!("(Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}').CommandLine"),
])
.output()
.expect("PowerShell can inspect the native child");
assert!(output.status.success(), "native process inspection failed");
String::from_utf8(output.stdout).expect("Windows command lines are Unicode")
}
#[cfg(target_os = "linux")]
fn native_command_line(pid: u32) -> String {
fs::read(format!("/proc/{pid}/cmdline"))
.map(|bytes| String::from_utf8_lossy(&bytes).replace('\0', " "))
.expect("/proc exposes the native child command line")
}
#[cfg(target_os = "macos")]
fn native_command_line(pid: u32) -> String {
let output = std::process::Command::new("ps")
.args(["-o", "command=", "-p", &pid.to_string()])
.output()
.expect("ps can inspect the native child");
assert!(output.status.success(), "native process inspection failed");
String::from_utf8(output.stdout).expect("the command line is UTF-8")
}
#[tokio::test]
async fn a_persistent_repository_leases_s1_and_journals_it_before_any_github_effect() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.github.watch_journal(Arc::clone(&harness.store));
harness.ready().await;
let attempt = harness.launch().await;
assert_eq!(
attempt.workspace(),
AttemptWorkspace::persistent_slot(nz(1)),
"the lowest free slot is leased"
);
assert_eq!(attempt.runtime_path(), harness.slot_path(1));
assert!(attempt.holds_slot_lease());
assert_eq!(
harness.attempt(attempt.id).runtime_path(),
harness.slot_path(1)
);
let facts = harness.github.registration_facts();
assert_eq!(facts.len(), 1);
assert_eq!(
facts[0].leased_slots,
vec![1],
"the lease was journalled first"
);
assert_eq!(facts[0].work_folder, DEFAULT_WORK_FOLDER);
}
#[tokio::test]
async fn a_terminal_but_uncleaned_attempt_keeps_its_slot_without_holding_capacity() {
let harness = Harness::new(
FakeGithubLifecycle::default().fail(true),
Arc::new(PersistentDemand),
)
.with_persistent_workspace(2);
harness.ready().await;
harness.launch_result().await.unwrap_err();
let first = harness.store.attempts().unwrap().remove(0);
assert_eq!(first.state(), AttemptState::Failed);
assert!(
!first.state().counts_against_capacity(),
"a concluded attempt is invisible to host capacity"
);
assert!(
first.holds_slot_lease(),
"and still owns its directory, so its slot is not free"
);
let second = harness.launch().await;
assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
assert_eq!(second.runtime_path(), harness.slot_path(2));
assert_eq!(
harness
.store
.slot_leases_for_policy(harness.policy.id)
.unwrap()
.len(),
2
);
}
#[tokio::test]
async fn two_sequential_allocations_at_capacity_one_reuse_s1_and_its_retained_work() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let first = harness.launch().await;
assert_eq!(first.runtime_path(), harness.slot_path(1));
let checkout = harness.slot_path(1).join(DEFAULT_WORK_FOLDER).join("repo");
fs::create_dir_all(&checkout).unwrap();
fs::write(checkout.join("checkout.txt"), b"from the first job").unwrap();
harness.cleanup_retaining_work(first.id).await;
let second = harness.launch().await;
assert_ne!(second.id, first.id);
assert_eq!(
second.workspace(),
AttemptWorkspace::persistent_slot(nz(1)),
"a released slot is leased again rather than skipped"
);
assert_eq!(
second.runtime_path(),
first.runtime_path(),
"the same slot is the same exact path"
);
assert_eq!(
fs::read_to_string(checkout.join("checkout.txt")).unwrap(),
"from the first job",
"the retained job workspace survived the second allocation"
);
assert!(harness.slot_path(1).join("runner-package").exists());
}
#[tokio::test]
async fn lowering_capacity_leaves_higher_slots_alone_and_raising_it_permits_them_again() {
let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.ready().await;
let first = harness.launch().await;
let second = harness.launch().await;
assert_eq!(second.runtime_path(), harness.slot_path(2));
let kept = harness
.slot_path(2)
.join(DEFAULT_WORK_FOLDER)
.join("kept.txt");
fs::create_dir_all(kept.parent().unwrap()).unwrap();
fs::write(&kept, b"s2 was here").unwrap();
harness.cleanup_retaining_work(second.id).await;
harness.policy.set_max_capacity(nz(1)).unwrap();
let refusal = harness.launch_result().await.unwrap_err().to_string();
assert!(
refusal.contains("s1 to s1"),
"the refusal names the ceiling it reached: {refusal}"
);
assert!(
harness.slot_path(2).exists() && kept.exists(),
"lowering capacity deletes nothing; the higher slot is merely unusable"
);
harness.policy.set_max_capacity(nz(2)).unwrap();
let third = harness.launch().await;
assert_eq!(third.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
assert_eq!(third.runtime_path(), harness.slot_path(2));
assert_eq!(fs::read_to_string(&kept).unwrap(), "s2 was here");
assert!(first.holds_slot_lease(), "s1 was never disturbed");
}
#[tokio::test]
async fn organization_and_ephemeral_policies_never_enter_slot_allocation() {
for policy in [
fixtures::policy()
.organization("octo")
.autoscale("home", 2)
.active()
.build(),
fixtures::policy()
.repository("octo/repo")
.autoscale("home", 2)
.active()
.build(),
] {
let mut harness =
Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_host_runner_root();
assert_eq!(policy.workspace_policy(), &WorkspacePolicy::Ephemeral);
harness.policy = policy;
harness.ready().await;
let attempt = harness.launch().await;
assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
assert_eq!(attempt.workspace().slot_number(), None);
assert!(!attempt.holds_slot_lease());
assert_eq!(
attempt.runtime_path().parent().unwrap(),
harness.host_root(),
"a disposable attempt is a child of the host root, never of a slot"
);
assert!(
harness
.store
.slot_leases_for_policy(harness.policy.id)
.unwrap()
.is_empty()
);
}
}
#[tokio::test]
async fn two_concurrent_allocations_never_share_a_slot() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.github.watch_journal(Arc::clone(&harness.store));
harness.ready().await;
let (first, second) = tokio::join!(harness.launch_result(), harness.launch_result());
let first = first.unwrap();
let second = second.unwrap();
let slots: BTreeSet<u16> = [&first, &second]
.iter()
.map(|attempt| {
attempt
.workspace()
.slot_number()
.expect("a persistent attempt leases a slot")
})
.collect();
assert_eq!(slots, BTreeSet::from([1, 2]), "one slot each, never shared");
assert_ne!(first.runtime_path(), second.runtime_path());
assert_eq!(
harness
.store
.slot_leases_for_policy(harness.policy.id)
.unwrap()
.len(),
2
);
let facts = harness.github.registration_facts();
assert_eq!(facts.len(), 2);
for fact in facts {
let attempt = [&first, &second]
.into_iter()
.find(|attempt| runner_name(attempt.id) == fact.runner_name)
.expect("every registration belongs to one of the two attempts");
let slot = attempt
.workspace()
.slot_number()
.expect("a persistent attempt leases a slot");
assert!(
fact.leased_slots.contains(&slot),
"a JIT request never precedes its own lease: s{slot} not in {:?}",
fact.leased_slots
);
assert_eq!(fact.work_folder, DEFAULT_WORK_FOLDER);
}
}
#[tokio::test]
async fn the_database_is_the_final_fence_against_two_attempts_in_one_slot() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.ready().await;
let first = harness.launch().await;
let clash = RunnerAttempt::allocate_in(
AttemptId::new_random(),
harness.policy.id,
first.runtime_path(),
AttemptWorkspace::persistent_slot(nz(1)),
harness.clock.now(),
);
assert!(matches!(
harness.store.record_attempt(&clash).unwrap_err(),
StoreError::SlotAlreadyLeased { slot: 1, .. }
));
let error = harness.launcher.record_allocation(&clash).unwrap_err();
let rendered = error.to_string();
assert!(rendered.contains("slot s1"), "{rendered}");
assert!(rendered.contains("nothing was written"), "{rendered}");
assert_eq!(
harness.store.attempts().unwrap().len(),
1,
"the losing allocator journalled nothing"
);
}
#[test]
fn slot_selection_fills_the_lowest_gap_and_stops_at_the_ceiling() {
let leased = |slots: &[u16]| -> Vec<RunnerAttempt> {
slots
.iter()
.map(|slot| {
RunnerAttempt::allocate_in(
AttemptId::new_random(),
fixtures::POLICY_ID,
format!("/srv/rman/acme/s{slot}"),
AttemptWorkspace::persistent_slot(nz(*slot)),
fixtures::created_at(),
)
})
.collect()
};
assert_eq!(lowest_free_slot(&[], nz(1)), Some(nz(1)));
assert_eq!(lowest_free_slot(&leased(&[1]), nz(4)), Some(nz(2)));
assert_eq!(lowest_free_slot(&leased(&[1, 3]), nz(4)), Some(nz(2)));
assert_eq!(lowest_free_slot(&leased(&[1]), nz(1)), None);
assert_eq!(lowest_free_slot(&leased(&[1, 2]), nz(2)), None);
let ephemeral = vec![RunnerAttempt::allocate(
AttemptId::new_random(),
fixtures::POLICY_ID,
"/srv/rman/host/abc",
fixtures::created_at(),
)];
assert_eq!(lowest_free_slot(&ephemeral, nz(1)), Some(nz(1)));
}
#[test]
fn a_slot_is_reusable_only_when_it_is_empty_or_holds_one_real_work_directory() {
let root = tempfile::tempdir().unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
accept_reusable_slot(&slot).expect("an empty slot is reusable");
fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
accept_reusable_slot(&slot).expect("a retained job workspace is reusable");
fs::create_dir(slot.join("bin")).unwrap();
fs::write(slot.join(".github-runner-id"), b"73").unwrap();
let refusal = accept_reusable_slot(&slot).unwrap_err().to_string();
assert!(refusal.contains("bin"), "{refusal}");
assert!(refusal.contains(".github-runner-id"), "{refusal}");
let file_work = root.path().join("s2");
fs::create_dir(&file_work).unwrap();
fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
assert!(accept_reusable_slot(&file_work).is_err());
}
#[cfg(unix)]
#[test]
fn a_link_shaped_work_directory_is_refused_rather_than_followed() {
let root = tempfile::tempdir().unwrap();
let elsewhere = root.path().join("elsewhere");
fs::create_dir(&elsewhere).unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
std::os::unix::fs::symlink(&elsewhere, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
assert!(accept_reusable_slot(&slot).is_err());
let linked_slot = root.path().join("s2");
std::os::unix::fs::symlink(&elsewhere, &linked_slot).unwrap();
assert!(create_or_validate_slot(&linked_slot).is_err());
}
#[test]
fn a_slot_standing_where_a_file_is_refuses_rather_than_replacing_it() {
let root = tempfile::tempdir().unwrap();
let occupied = root.path().join("s1");
fs::write(&occupied, b"an operator's file").unwrap();
let refusal = create_or_validate_slot(&occupied).unwrap_err().to_string();
assert!(refusal.contains("is not a directory"), "{refusal}");
assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
let fresh = root.path().join("s2");
create_or_validate_slot(&fresh).expect("a missing slot is created");
assert!(fresh.is_dir());
create_or_validate_slot(&fresh).expect("an existing directory is accepted");
}
#[test]
fn the_retained_work_directory_is_matched_the_way_the_filesystem_matches_it() {
assert!(is_work_folder(OsStr::new(DEFAULT_WORK_FOLDER)));
assert!(!is_work_folder(OsStr::new("_work2")));
assert_eq!(is_work_folder(OsStr::new("_Work")), cfg!(windows));
}
#[test]
fn package_materialization_never_overwrites_or_follows_a_retained_work_directory() {
let root = tempfile::tempdir().unwrap();
let package = root.path().join("package");
fs::create_dir_all(package.join("bin")).unwrap();
fs::write(package.join("bin").join("Runner.Listener"), b"binary").unwrap();
fs::create_dir_all(package.join("externals").join(DEFAULT_WORK_FOLDER)).unwrap();
let slot = root.path().join("s1");
let retained = slot.join(DEFAULT_WORK_FOLDER).join("repo");
fs::create_dir_all(&retained).unwrap();
fs::write(retained.join("checkout.txt"), b"from the first job").unwrap();
copy_package_tree(&package, &slot, &TrustedTrees::default())
.expect("the package lays out around `_work`");
assert!(slot.join("bin").join("Runner.Listener").exists());
assert!(
slot.join("externals").join(DEFAULT_WORK_FOLDER).is_dir(),
"the guard is top-level only"
);
assert_eq!(
fs::read_to_string(retained.join("checkout.txt")).unwrap(),
"from the first job"
);
fs::create_dir(package.join(DEFAULT_WORK_FOLDER)).unwrap();
let error = copy_package_tree(&package, &slot, &TrustedTrees::default()).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
assert_eq!(
fs::read_to_string(retained.join("checkout.txt")).unwrap(),
"from the first job"
);
}
#[test]
fn rolling_back_a_materialization_keeps_a_slot_but_removes_a_disposable_directory() {
let root = tempfile::tempdir().unwrap();
let slot = root.path().join("s1");
let retained = slot.join(DEFAULT_WORK_FOLDER);
fs::create_dir_all(retained.join("repo")).unwrap();
fs::write(retained.join("repo").join("checkout.txt"), b"kept").unwrap();
fs::create_dir_all(slot.join("bin")).unwrap();
fs::write(slot.join(".github-runner-id"), b"73").unwrap();
let persistent = RunnerAttempt::allocate_in(
AttemptId::new_random(),
fixtures::POLICY_ID,
&slot,
AttemptWorkspace::persistent_slot(nz(1)),
fixtures::created_at(),
);
remove_materialized_package(&persistent).unwrap();
assert!(slot.is_dir(), "the slot itself is not removed");
assert!(!slot.join("bin").exists());
assert!(!slot.join(".github-runner-id").exists());
assert_eq!(
fs::read_to_string(retained.join("repo").join("checkout.txt")).unwrap(),
"kept"
);
let disposable_path = root.path().join("abcdef012345");
fs::create_dir_all(disposable_path.join(DEFAULT_WORK_FOLDER)).unwrap();
let disposable = RunnerAttempt::allocate(
AttemptId::new_random(),
fixtures::POLICY_ID,
&disposable_path,
fixtures::created_at(),
);
remove_materialized_package(&disposable).unwrap();
assert!(
!disposable_path.exists(),
"a disposable directory is still removed whole"
);
}
fn litter_the_slot(slot: &Path) {
for directory in ["bin", "externals", "_diag"] {
fs::create_dir_all(slot.join(directory)).unwrap();
}
fs::write(slot.join("bin").join("Runner.Listener"), b"binary").unwrap();
for file in SENSITIVE_SLOT_ENTRIES
.iter()
.filter(|entry| !slot.join(entry).is_dir())
{
fs::write(slot.join(file), b"runner state").unwrap();
}
fs::write(
slot.join(format!(
"{}0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0.tmp",
RestrictiveHandoff::NAME_PREFIX
)),
JIT.as_bytes(),
)
.unwrap();
}
fn retain_under_work(slot: &Path) -> PathBuf {
let checkout = slot.join(DEFAULT_WORK_FOLDER).join("repo").join("target");
fs::create_dir_all(&checkout).unwrap();
let marker = checkout.join("build-output.bin");
fs::write(&marker, RETAINED).unwrap();
marker
}
const RETAINED: &str = "a Git-ignored build output the next job reuses";
fn entries_of(directory: &Path) -> Vec<String> {
let mut names: Vec<String> = fs::read_dir(directory)
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.collect();
names.sort();
names
}
fn only_the_job_workspace() -> Vec<String> {
vec![DEFAULT_WORK_FOLDER.to_owned()]
}
#[cfg(unix)]
#[test]
fn disposable_tree_removal_does_not_open_a_dotnet_diagnostic_fifo() {
use std::sync::mpsc;
let temporary = tempfile::tempdir().unwrap();
let tree = temporary.path().join("attempt");
let diagnostic = tree.join("tmp/clr-debug-pipe-runner-in");
fs::create_dir_all(diagnostic.parent().unwrap()).unwrap();
assert!(
std::process::Command::new("mkfifo")
.arg(&diagnostic)
.status()
.unwrap()
.success()
);
let (finished, result) = mpsc::channel();
std::thread::spawn(move || {
let removed = remove_runtime_tree(&tree);
let _ = finished.send(removed);
});
result
.recv_timeout(Duration::from_secs(2))
.expect("runtime deletion must not wait for a FIFO peer")
.unwrap();
assert!(!diagnostic.exists());
}
#[cfg(windows)]
#[test]
fn disposable_tree_removal_unlinks_a_junction_nobody_may_list() {
let temporary = tempfile::tempdir().unwrap();
let tree = temporary.path().join("attempt");
let cache = tree.join(r"tmp\home\AppData\Local\Microsoft\Windows\INetCache");
let outside = temporary.path().join("outside");
fs::create_dir_all(&cache).unwrap();
fs::create_dir_all(&outside).unwrap();
fs::write(outside.join("keep.txt"), b"not the attempt's").unwrap();
fs::write(tree.join("tmp").join("job.log"), b"job output").unwrap();
let junction = cache.join("Content.IE5");
let created = std::process::Command::new("cmd")
.arg("/C")
.arg("mklink")
.arg("/J")
.arg(&junction)
.arg(&outside)
.output()
.unwrap();
assert!(created.status.success(), "{created:?}");
let denied = std::process::Command::new("icacls")
.arg(&junction)
.args(["/L", "/deny", "*S-1-1-0:(RD)"])
.output()
.unwrap();
assert!(denied.status.success(), "{denied:?}");
let unfixed = remove_dir_all::remove_dir_all(&tree)
.expect_err("the fixture no longer reproduces the WinINet junction");
assert_eq!(unfixed.kind(), std::io::ErrorKind::PermissionDenied);
assert!(junction.symlink_metadata().is_ok());
remove_runtime_tree(&tree).expect("a tree holding a deny-list junction is removed");
assert!(!tree.exists(), "the attempt tree survived");
assert_eq!(
fs::read(outside.join("keep.txt")).unwrap(),
b"not the attempt's",
"removing the junction followed it into the directory it points at"
);
}
struct BlockedDeletion {
directory: PathBuf,
#[cfg(windows)]
_handle: fs::File,
}
impl BlockedDeletion {
const HELD: &'static str = "held-open";
fn inject(directory: &Path) -> Option<Self> {
#[cfg(unix)]
if !Self::refusal_is_possible() {
return None;
}
fs::create_dir_all(directory).unwrap();
fs::write(
directory.join(Self::HELD),
b"a file the scrub cannot remove",
)
.unwrap();
#[cfg(windows)]
let handle = {
use std::os::windows::fs::OpenOptionsExt;
fs::OpenOptions::new()
.read(true)
.share_mode(0)
.open(directory.join(Self::HELD))
.expect("the blocking handle opens")
};
#[cfg(unix)]
Self::set_mode(directory, 0o555);
Some(Self {
directory: directory.to_path_buf(),
#[cfg(windows)]
_handle: handle,
})
}
fn release(self) {
drop(self);
}
#[cfg(unix)]
fn refusal_is_possible() -> bool {
let probe = tempfile::tempdir().unwrap();
let directory = probe.path().join("probe");
fs::create_dir(&directory).unwrap();
fs::write(directory.join("file"), b"probe").unwrap();
Self::set_mode(&directory, 0o555);
let refused = fs::remove_dir_all(&directory).is_err();
Self::set_mode(&directory, 0o755);
refused
}
#[cfg(unix)]
fn set_mode(directory: &Path, mode: u32) {
use std::os::unix::fs::PermissionsExt;
let mut permissions = fs::metadata(directory).unwrap().permissions();
permissions.set_mode(mode);
fs::set_permissions(directory, permissions).unwrap();
}
}
impl Drop for BlockedDeletion {
fn drop(&mut self) {
#[cfg(unix)]
Self::set_mode(&self.directory, 0o755);
#[cfg(not(unix))]
let _ = &self.directory;
}
}
#[tokio::test]
async fn two_sequential_jobs_keep_the_checkout_and_start_without_the_earlier_runner_state() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let first = harness.launch().await;
let slot = harness.slot_path(1);
assert_eq!(first.runtime_path(), slot);
assert_eq!(
read_runner_id(&slot),
Some(73),
"the attempt registered, so its identity is on disk"
);
let marker = retain_under_work(&slot);
litter_the_slot(&slot);
harness.cleanup_retaining_work(first.id).await;
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
assert_eq!(
read_runner_id(&slot),
None,
"the first attempt's registration identity is gone before the second starts"
);
assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
assert!(!harness.attempt(first.id).holds_slot_lease());
let second = harness.launch().await;
assert_ne!(second.id, first.id);
assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(1)));
assert_eq!(
second.runtime_path(),
slot,
"the same slot, so the same retained `_work`"
);
assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
}
#[tokio::test]
async fn cleaning_a_persistent_slot_needs_no_policy_and_scans_no_directory_for_ownership() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let attempt = harness.launch().await;
let slot = harness.slot_path(1);
let marker = retain_under_work(&slot);
litter_the_slot(&slot);
harness.conclude(attempt.id);
harness
.store
.remove_policy(harness.policy.id, harness.policy.revision())
.unwrap();
assert!(harness.store.policy(harness.policy.id).unwrap().is_none());
harness
.launcher
.clean(attempt.id)
.await
.expect("journal facts alone are enough to clean the slot");
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert!(marker.exists());
assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn an_injected_partial_deletion_quarantines_the_slot_across_a_restart() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.ready().await;
let first = harness.launch().await;
let slot = harness.slot_path(1);
let marker = retain_under_work(&slot);
litter_the_slot(&slot);
harness.conclude(first.id);
let Some(block) = BlockedDeletion::inject(&slot.join("bin")) else {
eprintln!(
"skipped: this account cannot be refused a deletion, so no partial deletion can \
be injected"
);
return;
};
let refusal = harness
.launcher
.clean(first.id)
.await
.expect_err("a deletion that failed may not report a cleaned slot");
let rendered = refusal.reason.to_string();
assert!(rendered.contains("could not be removed"), "{rendered}");
let held = harness.attempt(first.id);
assert_eq!(held.state(), AttemptState::Failed, "still not cleaned");
assert!(held.holds_slot_lease(), "so the slot is still leased");
assert!(
!held.state().counts_against_capacity(),
"and a concluded attempt still costs the host no capacity"
);
let restarted = harness.restart();
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("one quarantined slot does not stop the host recovering");
assert_eq!(
harness.attempt(first.id).state(),
AttemptState::Failed,
"the quarantine survived the restart"
);
assert!(
harness
.reconcile_events
.events()
.iter()
.any(|event| matches!(
event,
LifecycleEvent::AttemptCleanFailed {
reason: "slot_entry_could_not_be_removed",
..
}
)),
"the refusal is reported rather than retried in silence"
);
let second = harness
.launch_through(&restarted)
.await
.expect("the host can still launch");
assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
block.release();
restarted
.clean(first.id)
.await
.expect("the retried cleanup completes");
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert!(marker.exists());
assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn an_injected_deletion_failure_leaves_a_disposable_attempt_uncleaned() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
let runtime = attempt.runtime_path().to_path_buf();
assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
harness.conclude(attempt.id);
let Some(block) = BlockedDeletion::inject(&runtime.join("held-open-subdirectory")) else {
eprintln!(
"skipped: this account cannot be refused a deletion, so no partial deletion can be injected"
);
return;
};
let refusal = harness
.launcher
.clean(attempt.id)
.await
.expect_err("a deletion that failed may not report a removed workspace");
let rendered = refusal.reason.to_string();
assert!(
rendered.contains("could not be removed"),
"the refusal names what happened: {rendered}"
);
assert_ne!(
harness.attempt(attempt.id).state(),
AttemptState::Cleaned,
"an attempt whose directory is still on disk is not cleaned"
);
assert!(
runtime.is_dir(),
"the directory the removal could not finish is still there, which is the fact the journal must keep agreeing with"
);
let restarted = harness.restart();
restarted
.recover_startup(std::slice::from_ref(&harness.policy))
.await
.expect("one locked ephemeral workspace does not stop startup recovery");
assert!(
harness
.reconcile_events
.events()
.iter()
.any(|event| matches!(
event,
LifecycleEvent::AttemptCleanFailed {
attempt: failed_attempt,
reason: "ephemeral_workspace_could_not_be_removed",
..
} if *failed_attempt == attempt.id
)),
"the deferred cleanup remains visible while the daemon keeps running"
);
block.release();
harness
.launcher
.clean(attempt.id)
.await
.expect("the retried cleanup completes");
assert!(!runtime.exists(), "the whole attempt directory goes");
assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn supervision_backs_off_a_workspace_it_cannot_remove() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
harness.ready().await;
let attempt = harness.launch().await;
let runtime = attempt.runtime_path().to_path_buf();
harness.conclude(attempt.id);
let Some(block) = BlockedDeletion::inject(&runtime.join("held-open-subdirectory")) else {
eprintln!(
"skipped: this account cannot be refused a deletion, so no partial deletion can be injected"
);
return;
};
let failures = || {
harness
.reconcile_events
.events()
.iter()
.filter(|event| {
matches!(
event,
LifecycleEvent::AttemptCleanFailed { attempt: failed, .. }
if *failed == attempt.id
)
})
.count()
};
let supervise = async || {
harness
.launcher
.supervise(&harness.policy)
.await
.expect("a stuck workspace does not fail supervision");
};
supervise().await;
assert_eq!(failures(), 1);
supervise().await;
assert_eq!(
failures(),
1,
"the failed cleanup was retried on the very next pass"
);
harness.clock.advance_secs(31);
supervise().await;
assert_eq!(
failures(),
2,
"the cleanup was not retried once its delay passed"
);
block.release();
harness.clock.advance_secs(61);
supervise().await;
assert!(
!runtime.exists(),
"the retry removes the workspace once it can"
);
assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
}
#[tokio::test]
async fn changing_a_repository_back_to_ephemeral_leaves_every_old_slot_untouched() {
let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let first = harness.launch().await;
let slot = harness.slot_path(1);
let marker = retain_under_work(&slot);
harness.cleanup_retaining_work(first.id).await;
harness
.policy
.set_workspace_policy(WorkspacePolicy::Ephemeral)
.unwrap();
let second = harness.launch().await;
assert_eq!(second.workspace(), AttemptWorkspace::Ephemeral);
assert_eq!(
second.runtime_path().parent().unwrap(),
harness.host_root(),
"a disposable attempt is a child of the host root"
);
assert!(slot.is_dir(), "the old slot is left where it stands");
assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
harness.conclude(second.id);
harness.launcher.clean(second.id).await.unwrap();
assert!(!second.runtime_path().exists());
assert!(marker.exists());
}
#[tokio::test]
async fn a_persistent_slot_is_scrubbed_only_after_the_process_is_signalled_and_gone() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(1);
harness.ready().await;
let slot = harness.slot_path(1);
fs::create_dir_all(&slot).unwrap();
let marker = retain_under_work(&slot);
litter_the_slot(&slot);
let id = AttemptId::new_random();
let mut attempt = RunnerAttempt::allocate_in(
id,
harness.policy.id,
&slot,
AttemptWorkspace::persistent_slot(nz(1)),
harness.clock.now(),
);
attempt.jit_received(harness.clock.now()).unwrap();
attempt.started(4242, harness.clock.now()).unwrap();
harness.store.record_attempt(&attempt).unwrap();
harness.clock.advance_secs(11);
harness.processes.set_alive(true);
harness
.github
.observe(GithubRunnerObservation::NotRegistered);
harness.launcher.supervise(&harness.policy).await.unwrap();
let actions = harness.processes.actions.lock().unwrap().clone();
let intent = actions
.iter()
.position(|action| *action == "terminate_intent")
.unwrap();
let signal = actions
.iter()
.position(|action| *action == "terminate")
.unwrap();
assert!(intent < signal, "{actions:?}");
assert!(!harness.processes.alive.load(Ordering::SeqCst));
let cleaned = harness.attempt(id);
assert_eq!(cleaned.state(), AttemptState::Cleaned);
assert!(matches!(
cleaned.outcome(),
Some(AttemptOutcome::Failed {
reason: FailureReason::TerminatedAfterRegistrationTimeout
})
));
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert!(marker.exists());
}
#[test]
fn a_scrub_retains_one_real_work_directory_and_removes_every_other_entry() {
let root = tempfile::tempdir().unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
let marker = retain_under_work(&slot);
litter_the_slot(&slot);
fs::write(slot.join("runner-package"), b"verified").unwrap();
scrub_slot_entries(&slot).expect("a slot of ordinary runner state scrubs");
verify_slot_scrubbed(&slot).expect("and proves it afterwards");
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert!(marker.exists());
}
#[test]
fn a_residue_refusal_never_reports_the_under_count_as_the_fact() {
let slot = Path::new("/runners/s1");
let counted = residue_detail(slot, 2, &["`bin`".to_owned()]);
assert!(counted.contains("2 entries other than"), "{counted}");
assert!(counted.contains("including `bin`"), "{counted}");
assert_eq!(
residue_detail(slot, 1, &[]),
format!(
"1 entry other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}",
slot.display()
)
);
let raced = residue_detail(slot, 0, &["`.credentials`".to_owned()]);
assert!(!raced.contains('0'), "{raced}");
assert!(raced.contains("reported nothing but"), "{raced}");
assert!(raced.contains("`.credentials` survived cleanup"), "{raced}");
}
#[test]
fn verification_asks_the_filesystem_rather_than_the_listing_that_missed_an_entry() {
let root = tempfile::tempdir().unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
verify_slot_scrubbed(&slot).expect("only `_work` is a clean slot");
for survivor in ["bin", ".credentials", IDENTITY_FILE, RUNNER_ID_FILE] {
fs::write(slot.join(survivor), b"left behind").unwrap();
let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
assert_eq!(quarantine.refusal, SlotRefusal::Residue);
assert!(
quarantine.detail.contains(&format!("`{survivor}`")),
"{quarantine}"
);
fs::remove_file(slot.join(survivor)).unwrap();
}
let handoff = slot.join(format!("{}whatever.tmp", RestrictiveHandoff::NAME_PREFIX));
fs::write(&handoff, JIT.as_bytes()).unwrap();
let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
assert!(
quarantine.detail.contains("an encoded JIT handoff"),
"{quarantine}"
);
assert!(!quarantine.detail.contains(JIT), "{quarantine}");
fs::remove_file(&handoff).unwrap();
fs::write(slot.join("ghp_DO_NOT_LEAK"), b"named by the job").unwrap();
let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
assert!(
quarantine.detail.contains("1 entry other than"),
"{quarantine}"
);
assert!(
!quarantine.detail.contains("ghp_DO_NOT_LEAK"),
"{quarantine}"
);
}
#[test]
fn a_slot_is_derived_from_the_journal_and_refused_when_it_disagrees() {
let root = tempfile::tempdir().unwrap();
let configured =
LocalAbsolutePath::new(root.path().to_str().unwrap()).expect("a local absolute root");
let slot = configured.as_path().join("s1");
fs::create_dir(&slot).unwrap();
verify_journalled_slot(&slot, nz(1), Some(&configured))
.expect("the journalled slot agrees");
verify_journalled_slot(&slot, nz(1), None)
.expect("and a policy that is gone removes a check, not the ability to clean");
assert_eq!(
verify_journalled_slot(&slot, nz(2), None)
.unwrap_err()
.refusal,
SlotRefusal::NotTheJournalledSlot
);
for stray in ["s1/nested", "not-a-slot", "s01"] {
let path = configured.as_path().join(stray);
assert_eq!(
verify_journalled_slot(&path, nz(1), None)
.unwrap_err()
.refusal,
SlotRefusal::NotTheJournalledSlot,
"{}",
path.display()
);
}
let elsewhere = tempfile::tempdir().unwrap();
let other =
LocalAbsolutePath::new(elsewhere.path().to_str().unwrap()).expect("a second root");
assert_eq!(
verify_journalled_slot(&slot, nz(1), Some(&other))
.unwrap_err()
.refusal,
SlotRefusal::PolicyRootDisagrees
);
}
#[cfg(unix)]
#[test]
fn a_substituted_work_directory_quarantines_the_slot_and_deletes_nothing_outside_it() {
let root = tempfile::tempdir().unwrap();
let outside = root.path().join("operator-data");
fs::create_dir(&outside).unwrap();
let sentinel = outside.join("do-not-delete.txt");
fs::write(
&sentinel,
b"an operator's data, outside every approved root",
)
.unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
fs::create_dir(slot.join("bin")).unwrap();
std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
let quarantine = scrub_slot_entries(&slot).unwrap_err();
assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
assert!(
sentinel.exists(),
"the deletion followed the link out of the slot"
);
assert!(outside.is_dir());
assert!(
slot.join(DEFAULT_WORK_FOLDER).symlink_metadata().is_ok(),
"the substituted link is left for the operator, never unlinked as if it were ours"
);
let file_work = root.path().join("s2");
fs::create_dir(&file_work).unwrap();
fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
assert_eq!(
scrub_slot_entries(&file_work).unwrap_err().refusal,
SlotRefusal::WorkNotADirectory
);
}
#[cfg(unix)]
#[test]
fn a_slot_replaced_by_a_link_out_of_its_root_is_refused_before_anything_is_read() {
let root = tempfile::tempdir().unwrap();
let outside = root.path().join("operator-data");
fs::create_dir(&outside).unwrap();
let sentinel = outside.join("do-not-delete.txt");
fs::write(
&sentinel,
b"an operator's data, outside every approved root",
)
.unwrap();
let inside = root.path().join("inside");
fs::create_dir(&inside).unwrap();
let slot = inside.join("s1");
std::os::unix::fs::symlink(&outside, &slot).unwrap();
assert_eq!(
verify_journalled_slot(&slot, nz(1), None)
.unwrap_err()
.refusal,
SlotRefusal::Containment
);
assert!(sentinel.exists());
assert!(
slot.symlink_metadata().is_ok(),
"the link is left for the operator rather than removed as if it were ours"
);
}
#[cfg(windows)]
#[test]
fn a_slot_root_replaced_by_a_junction_is_refused_before_anything_is_read() {
let root = tempfile::tempdir().unwrap();
let outside = root.path().join("operator-data");
fs::create_dir(&outside).unwrap();
let sentinel = outside.join("do-not-delete.txt");
fs::write(
&sentinel,
b"an operator's data, outside every approved root",
)
.unwrap();
let inside = root.path().join("inside");
fs::create_dir(&inside).unwrap();
let slot = inside.join("s1");
let Some(()) = plant_junction(&slot, &outside) else {
eprintln!("skipped: this machine would not create a directory junction");
return;
};
assert_eq!(
verify_journalled_slot(&slot, nz(1), None)
.unwrap_err()
.refusal,
SlotRefusal::Containment
);
assert!(
sentinel.exists(),
"the refusal resolved the junction and reached the operator's data"
);
assert!(
slot.symlink_metadata().is_ok(),
"the junction is left for the operator rather than removed as if it were ours"
);
}
#[cfg(windows)]
fn plant_junction(link: &Path, target: &Path) -> Option<()> {
let made = std::process::Command::new("cmd")
.arg("/C")
.arg("mklink")
.arg("/J")
.arg(link)
.arg(target)
.output()
.ok()?;
(made.status.success() && link.symlink_metadata().is_ok()).then_some(())
}
#[test]
fn removing_an_attempt_never_reaches_into_the_dependency_caches_it_linked() {
let root = tempfile::tempdir().unwrap();
let cache = root.path().join("_cache").join("o").join("r").join("npm");
fs::create_dir_all(&cache).unwrap();
let cached = cache.join("package.tgz");
fs::write(&cached, b"kept between jobs").unwrap();
let attempt = root.path().join("0123456789abcdef");
let work = attempt.join("_work").join("repo");
fs::create_dir_all(&work).unwrap();
fs::hard_link(&cached, work.join("hard-linked.tgz")).unwrap();
#[cfg(unix)]
std::os::unix::fs::symlink(&cache, work.join("linked-cache")).unwrap();
#[cfg(windows)]
if plant_junction(&work.join("linked-cache"), &cache).is_none() {
eprintln!("this machine would not create a junction; the hard link is still checked");
}
remove_runtime_tree(&attempt).unwrap();
assert!(!attempt.exists());
assert_eq!(fs::read(&cached).unwrap(), b"kept between jobs");
}
#[cfg(windows)]
#[test]
fn a_work_directory_replaced_by_a_junction_fails_closed_and_deletes_nothing_beyond_it() {
let root = tempfile::tempdir().unwrap();
let outside = root.path().join("operator-data");
fs::create_dir(&outside).unwrap();
let sentinel = outside.join("do-not-delete.txt");
fs::write(
&sentinel,
b"an operator's data, outside every approved root",
)
.unwrap();
let slot = root.path().join("s1");
fs::create_dir(&slot).unwrap();
fs::create_dir(slot.join("bin")).unwrap();
let Some(()) = plant_junction(&slot.join(DEFAULT_WORK_FOLDER), &outside) else {
eprintln!("skipped: this machine would not create a directory junction");
return;
};
let work = fs::symlink_metadata(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
assert!(is_link_like(&work), "a junction is a reparse point");
let quarantine = scrub_slot_entries(&slot).unwrap_err();
assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
assert!(
sentinel.exists(),
"the deletion followed the junction out of the slot"
);
assert!(outside.is_dir());
let elsewhere = root.path().join("s2");
fs::create_dir(&elsewhere).unwrap();
fs::create_dir(elsewhere.join(DEFAULT_WORK_FOLDER)).unwrap();
if plant_junction(&elsewhere.join("externals"), &outside).is_some() {
scrub_slot_entries(&elsewhere).expect("an ordinary entry is removed, junction or not");
verify_slot_scrubbed(&elsewhere).expect("and the slot verifies");
assert!(sentinel.exists(), "the junction was followed, not unlinked");
assert_eq!(entries_of(&elsewhere), only_the_job_workspace());
}
}
#[cfg(unix)]
#[tokio::test]
async fn a_substituted_work_directory_leaves_the_attempt_uncleaned_and_still_leased() {
let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
.with_persistent_workspace(2);
harness.ready().await;
let first = harness.launch().await;
let slot = harness.slot_path(1);
harness.conclude(first.id);
let outside = harness._root.path().join("operator-data");
fs::create_dir_all(&outside).unwrap();
let sentinel = outside.join("do-not-delete.txt");
fs::write(&sentinel, b"outside every approved root").unwrap();
std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
harness
.launcher
.clean(first.id)
.await
.expect_err("a slot whose `_work` was substituted is quarantined");
assert!(sentinel.exists());
let held = harness.attempt(first.id);
assert_eq!(held.state(), AttemptState::Failed);
assert!(held.holds_slot_lease());
let second = harness.launch().await;
assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
}
#[test]
fn a_slot_that_is_a_file_is_refused_and_a_slot_that_is_gone_is_not() {
let root = tempfile::tempdir().unwrap();
let occupied = root.path().join("s1");
fs::write(&occupied, b"an operator's file").unwrap();
verify_journalled_slot(&occupied, nz(1), None).expect("the path is the journalled slot");
assert_eq!(
slot_is_present(&occupied).unwrap_err().refusal,
SlotRefusal::SlotNotADirectory
);
assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
assert!(!slot_is_present(&root.path().join("s2")).unwrap());
let present = root.path().join("s3");
fs::create_dir(&present).unwrap();
assert!(slot_is_present(&present).unwrap());
}
#[test]
fn cleanup_dispatches_on_the_journalled_kind_and_not_on_what_the_directory_holds() {
let root = tempfile::tempdir().unwrap();
let disposable = root.path().join("abcdef012345");
fs::create_dir_all(disposable.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
let ephemeral = RunnerAttempt::allocate(
AttemptId::new_random(),
fixtures::POLICY_ID,
&disposable,
fixtures::created_at(),
);
remove_materialized_package(&ephemeral).unwrap();
assert!(!disposable.exists());
let slot = root.path().join("s1");
fs::create_dir_all(slot.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
fs::create_dir_all(slot.join("bin")).unwrap();
let persistent = RunnerAttempt::allocate_in(
AttemptId::new_random(),
fixtures::POLICY_ID,
&slot,
AttemptWorkspace::persistent_slot(nz(1)),
fixtures::created_at(),
);
remove_materialized_package(&persistent).unwrap();
assert_eq!(entries_of(&slot), only_the_job_workspace());
assert!(slot.join(DEFAULT_WORK_FOLDER).join("repo").is_dir());
}
#[test]
fn every_slot_refusal_names_a_distinct_event_class_and_keeps_the_lease() {
let refusals = [
SlotRefusal::NotTheJournalledSlot,
SlotRefusal::PolicyRootDisagrees,
SlotRefusal::Containment,
SlotRefusal::SlotNotADirectory,
SlotRefusal::Enumeration,
SlotRefusal::WorkNotADirectory,
SlotRefusal::Deletion,
SlotRefusal::Residue,
];
let classes: BTreeSet<&str> = refusals.iter().map(|refusal| refusal.class()).collect();
assert_eq!(
classes.len(),
refusals.len(),
"an event class shared by two refusals tells an operator less than it appears to"
);
for refusal in refusals {
assert!(
refusal
.class()
.chars()
.all(|c| c.is_ascii_lowercase() || c == '_'),
"{}",
refusal.class()
);
assert!(
refusal.remediation().contains("slot lease"),
"every refusal has to say the lease is still held: {}",
refusal.class()
);
}
}
#[test]
fn copy_package_tree_copies_files_and_preserves_paths_with_spaces() {
let root = tempfile::tempdir().unwrap();
let source = root.path().join("source with spaces");
let dest = root.path().join("dest with spaces");
fs::create_dir_all(&source).unwrap();
fs::write(source.join("file1.txt"), b"hello").unwrap();
let nested = source.join("nested dir");
fs::create_dir_all(&nested).unwrap();
fs::write(nested.join("file2.txt"), b"world").unwrap();
let nested_work = nested.join(DEFAULT_WORK_FOLDER);
fs::create_dir_all(&nested_work).unwrap();
fs::write(nested_work.join("allowed.txt"), b"allowed").unwrap();
copy_package_tree(&source, &dest, &TrustedTrees::default()).unwrap();
assert_eq!(fs::read_to_string(dest.join("file1.txt")).unwrap(), "hello");
assert_eq!(
fs::read_to_string(dest.join("nested dir").join("file2.txt")).unwrap(),
"world"
);
assert_eq!(
fs::read_to_string(
dest.join("nested dir")
.join(DEFAULT_WORK_FOLDER)
.join("allowed.txt")
)
.unwrap(),
"allowed"
);
}
#[test]
fn every_field_name_the_attempt_sink_emits_is_allowed() {
use runner_manager_platform::logging::is_field_allowed;
for field in [
"event",
"attempt_id",
"reason",
"retry_in_ms",
"duration_ms",
"order_wait_ms",
"lock_wait_ms",
"turn_wait_ms",
"materialize_ms",
"package_wait_ms",
"package_install_ms",
"package_verify_ms",
"package_copy_ms",
"package_downloaded",
"prepare_ms",
"register_ms",
"spawn_ms",
] {
assert!(
is_field_allowed(field),
"`{field}` would be logged as `[redacted]`"
);
}
}
#[cfg(unix)]
fn package_entry(root: &Path, version: &str, manifest: &str) -> PathBuf {
let entry = root.join("packages").join(version);
fs::create_dir_all(entry.join("bin")).unwrap();
fs::write(entry.join(crate::package::MANIFEST_FILE), manifest).unwrap();
fs::write(entry.join("bin").join("Runner.Listener"), version).unwrap();
entry
}
#[cfg(unix)]
fn listener_inode(seed: &Path) -> u64 {
use std::os::unix::fs::MetadataExt as _;
fs::metadata(seed.join("bin").join("Runner.Listener"))
.unwrap()
.ino()
}
#[cfg(unix)]
#[test]
fn a_package_seed_is_built_once_and_reused_while_untouched() {
let root = tempfile::tempdir().unwrap();
let source = package_entry(root.path(), "2.336.0", "{\"digest\":\"a\"}");
let seeds = root.path().join("runners").join(PACKAGE_SEED_DIR);
let trusted = TrustedTrees::default();
let seed = refresh_package_seed(&source, &seeds, &trusted).unwrap();
assert_eq!(seed, seeds.join("2.336.0"));
assert_eq!(
fs::read_to_string(seed.join("bin").join("Runner.Listener")).unwrap(),
"2.336.0"
);
let built = listener_inode(&seed);
assert_eq!(
refresh_package_seed(&source, &seeds, &trusted).unwrap(),
seed
);
assert_eq!(listener_inode(&seed), built, "an untouched seed is reused");
}
#[cfg(unix)]
#[test]
fn a_tampered_or_unrecorded_seed_is_rebuilt() {
let root = tempfile::tempdir().unwrap();
let source = package_entry(root.path(), "2.336.0", "{\"digest\":\"a\"}");
let seeds = root.path().join("runners").join(PACKAGE_SEED_DIR);
let trusted = TrustedTrees::default();
let seed = refresh_package_seed(&source, &seeds, &trusted).unwrap();
fs::write(seed.join("bin").join("Runner.Listener"), b"planted").unwrap();
refresh_package_seed(&source, &seeds, &trusted).unwrap();
assert_eq!(
fs::read_to_string(seed.join("bin").join("Runner.Listener")).unwrap(),
"2.336.0",
"a seed whose print moved is rebuilt from the cache entry"
);
let old_listener = fs::File::open(seed.join("bin").join("Runner.Listener")).unwrap();
let built = listener_inode(&seed);
refresh_package_seed(&source, &seeds, &TrustedTrees::default()).unwrap();
assert_ne!(
listener_inode(&seed),
built,
"an unrecorded seed is rebuilt"
);
drop(old_listener);
}
#[cfg(unix)]
#[test]
fn a_package_seed_is_rebuilt_for_a_new_entry_and_drops_every_other() {
let root = tempfile::tempdir().unwrap();
let seeds = root.path().join("runners").join(PACKAGE_SEED_DIR);
let trusted = TrustedTrees::default();
let old = package_entry(root.path(), "2.335.0", "{\"digest\":\"old\"}");
refresh_package_seed(&old, &seeds, &trusted).unwrap();
fs::create_dir_all(seeds.join(".staging-interrupted")).unwrap();
fs::write(seeds.join("stray"), b"").unwrap();
let new = package_entry(root.path(), "2.336.0", "{\"digest\":\"new\"}");
let seed = refresh_package_seed(&new, &seeds, &trusted).unwrap();
let mut left: Vec<_> = fs::read_dir(&seeds)
.unwrap()
.map(|entry| entry.unwrap().file_name().into_string().unwrap())
.collect();
left.sort();
assert_eq!(left, ["2.336.0"]);
fs::write(
new.join(crate::package::MANIFEST_FILE),
"{\"digest\":\"again\"}",
)
.unwrap();
refresh_package_seed(&new, &seeds, &trusted).unwrap();
assert_eq!(
fs::read_to_string(seed.join(crate::package::MANIFEST_FILE)).unwrap(),
"{\"digest\":\"again\"}",
"a stale seed is rebuilt from the changed entry"
);
}
#[cfg(unix)]
#[test]
fn a_package_on_the_runtime_volume_is_its_own_seed() {
let root = tempfile::tempdir().unwrap();
let source = package_entry(root.path(), "2.336.0", "{}");
let runtime = root.path().join("runners").join("attempt");
fs::create_dir_all(&runtime).unwrap();
let trusted = TrustedTrees::default();
assert_eq!(package_seed(&source, &runtime, &trusted).unwrap(), source);
copy_package_tree(&source, &runtime, &trusted).unwrap();
assert!(runtime.join("bin").join("Runner.Listener").exists());
assert!(!root.path().join("runners").join(PACKAGE_SEED_DIR).exists());
}
#[test]
fn copy_package_tree_refuses_top_level_work_folder() {
let root = tempfile::tempdir().unwrap();
let source = root.path().join("source");
let dest = root.path().join("dest");
fs::create_dir_all(&source).unwrap();
fs::write(source.join("file1.txt"), b"hello").unwrap();
let top_work = source.join(DEFAULT_WORK_FOLDER);
fs::create_dir_all(&top_work).unwrap();
let err = copy_package_tree(&source, &dest, &TrustedTrees::default()).unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
}
}