use std::fmt;
use std::io::Write;
use std::path::PathBuf;
use runner_manager_domain::attempt::RunnerAttempt;
use runner_manager_domain::model::{Host, ScaleTarget, StartMode, TargetScope};
use runner_manager_domain::path::LocalAbsolutePath;
use runner_manager_domain::policy::ScalePolicy;
use runner_manager_domain::store::{Store, StoreError};
use runner_manager_domain::workspace::{WorkspaceKind, WorkspacePolicy};
use runner_manager_platform::paths::AppPaths;
use runner_manager_platform::runner_root::{
RootOwner, RootPreflight, default_runner_root, is_on_privacy_gated_volume,
};
use runner_manager_platform::service::{InstallRecord, ServiceError};
use super::{CliError, Context, Failure, Styling, write_failed};
pub const PERSISTENT_TRUST_WARNING: &[&str] = &[
"warning: a persistent workspace is a trusted-workflow optimization, not isolation.",
" - files under _work are an input to later jobs on the same slot;",
" - executable and generated content can cross branch and job boundaries;",
" - do not enable it for untrusted fork or pull-request workflows;",
" - changing or disabling persistence does not delete old directories;",
" - `actions/checkout` still cleans the workspace, including Git-ignored files,",
" unless the workflow sets `clean: false`.",
];
pub fn write_trust_warning(out: &mut dyn Write) -> Result<(), CliError> {
let failed = write_failed("this workspace trust warning");
for line in PERSISTENT_TRUST_WARNING {
writeln!(out, "{line}").map_err(&failed)?;
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootSource {
PlatformDefault,
Configured,
}
impl RootSource {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
RootSource::PlatformDefault => "platform-default",
RootSource::Configured => "configured",
}
}
#[must_use]
pub const fn as_token(self) -> &'static str {
match self {
RootSource::PlatformDefault => "platform_default",
RootSource::Configured => "configured",
}
}
}
impl fmt::Display for RootSource {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HostRoot {
pub configured: Option<LocalAbsolutePath>,
pub effective: Option<LocalAbsolutePath>,
pub unavailable: Option<String>,
}
impl HostRoot {
#[must_use]
pub const fn source(&self) -> RootSource {
if self.configured.is_some() {
RootSource::Configured
} else {
RootSource::PlatformDefault
}
}
#[must_use]
pub fn effective_text(&self) -> Option<&str> {
self.effective.as_ref().map(LocalAbsolutePath::as_str)
}
#[must_use]
pub fn rendered(&self) -> String {
match (&self.effective, &self.unavailable) {
(Some(root), _) => root.as_str().to_string(),
(None, Some(reason)) => format!("unavailable ({reason})"),
(None, None) => "unavailable".to_string(),
}
}
}
#[must_use]
pub fn host_root(app_paths: &AppPaths, host: Option<&Host>) -> HostRoot {
if let Some(configured) = host.and_then(|host| host.runner_root_override.clone()) {
return HostRoot {
effective: Some(configured.clone()),
configured: Some(configured),
unavailable: None,
};
}
match default_runner_root(app_paths) {
Ok(root) => HostRoot {
configured: None,
effective: Some(root),
unavailable: None,
},
Err(source) => HostRoot {
configured: None,
effective: None,
unavailable: Some(source.to_string()),
},
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AffectedAttempts {
pub active: u16,
pub cleanup_blocked: u16,
}
impl AffectedAttempts {
#[must_use]
pub const fn total(&self) -> u16 {
self.active.saturating_add(self.cleanup_blocked)
}
#[must_use]
pub const fn is_empty(&self) -> bool {
self.total() == 0
}
#[must_use]
pub fn of<'a>(attempts: impl IntoIterator<Item = &'a RunnerAttempt>) -> Self {
let mut counts = Self::default();
for attempt in attempts {
if attempt.counts_against_capacity() {
counts.active = counts.active.saturating_add(1);
} else {
counts.cleanup_blocked = counts.cleanup_blocked.saturating_add(1);
}
}
counts
}
#[must_use]
pub fn refusal(&self, subject: &str) -> String {
format!(
"{subject} cannot change while attempts still own it: {} active and {} awaiting \
cleanup. Nothing was changed.",
self.active, self.cleanup_blocked
)
}
}
pub fn host_affected_attempts(store: &dyn Store) -> Result<AffectedAttempts, CliError> {
Ok(AffectedAttempts::of(
store
.uncleaned_ephemeral_attempts()
.map_err(read_failure)?
.iter(),
))
}
pub fn policy_affected_attempts(
store: &dyn Store,
policy: &ScalePolicy,
) -> Result<AffectedAttempts, CliError> {
Ok(AffectedAttempts::of(
store
.uncleaned_attempts_for_policy(policy.id)
.map_err(read_failure)?
.iter(),
))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SlotLease {
pub slot: u16,
pub attempt: String,
pub state: String,
pub cleanup_blocked: bool,
}
pub fn slot_leases(store: &dyn Store, policy: &ScalePolicy) -> Result<Vec<SlotLease>, CliError> {
Ok(store
.slot_leases_for_policy(policy.id)
.map_err(read_failure)?
.iter()
.filter_map(|attempt| {
Some(SlotLease {
slot: attempt.workspace().slot_number()?,
attempt: attempt.id.to_string(),
state: attempt.state().to_string(),
cleanup_blocked: attempt.is_terminal(),
})
})
.collect())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RepositoryWorkspace {
pub target: ScaleTarget,
pub policy: WorkspacePolicy,
pub host_root: HostRoot,
pub attempts: AffectedAttempts,
pub leases: Vec<SlotLease>,
}
impl RepositoryWorkspace {
#[must_use]
pub const fn kind(&self) -> WorkspaceKind {
self.policy.kind()
}
#[must_use]
pub fn effective_root(&self) -> Option<&str> {
self.policy
.root()
.map(LocalAbsolutePath::as_str)
.or_else(|| self.host_root.effective_text())
}
#[must_use]
pub fn root_source(&self) -> &'static str {
if self.policy.is_persistent() {
"repository"
} else {
self.host_root.source().as_token()
}
}
#[must_use]
pub fn root_source_badge(&self) -> &'static str {
if self.policy.is_persistent() {
"repository-specific"
} else {
self.host_root.source().as_str()
}
}
}
pub fn repository_workspace(
store: &dyn Store,
host_root: &HostRoot,
policy: &ScalePolicy,
) -> Result<RepositoryWorkspace, CliError> {
Ok(RepositoryWorkspace {
target: policy.target.clone(),
policy: policy.workspace_policy().clone(),
host_root: host_root.clone(),
attempts: policy_affected_attempts(store, policy)?,
leases: slot_leases(store, policy)?,
})
}
fn preflight_against_everything<'a>(
app_paths: &'a AppPaths,
host: &HostRoot,
policies: &[ScalePolicy],
) -> RootPreflight<'a> {
let mut preflight = RootPreflight::new(app_paths);
if let Some(root) = host.effective.clone() {
preflight = preflight.against(RootOwner::Host, root);
}
for policy in policies {
if let Some(root) = policy.workspace_policy().root() {
preflight =
preflight.against(RootOwner::Repository(policy.target.slug()), root.clone());
}
}
preflight
}
fn validated_leaf(
app_paths: &AppPaths,
host_root: &HostRoot,
policies: &[ScalePolicy],
owner: &RootOwner,
root: &LocalAbsolutePath,
) -> Result<Option<PathBuf>, CliError> {
let checked = preflight_against_everything(app_paths, host_root, policies)
.check(owner, root)
.map_err(|source| unusable(source, owner))?;
let Some(leaf) = checked.leaf_to_create() else {
return Ok(None);
};
std::fs::create_dir(leaf).map_err(|source| {
CliError::with_remedy(
Failure::LocalState,
format!("cannot create {}: {source}", leaf.display()),
owner.remediation(),
)
})?;
Ok(Some(leaf.to_path_buf()))
}
#[must_use]
pub fn ephemeral_rejects_a_path(target: &ScaleTarget) -> CliError {
CliError::with_remedy(
Failure::InvalidArgument,
"--path names where persistent slots live, and an ephemeral workspace has none; \
nothing was changed",
format!("runner-manager repo set-workspace {target} --mode ephemeral"),
)
}
pub fn parse_root(raw: &str, owner: &RootOwner) -> Result<LocalAbsolutePath, CliError> {
LocalAbsolutePath::new(raw).map_err(|source| {
CliError::with_remedy(
Failure::InvalidArgument,
format!("{raw:?} cannot be used as {owner}: {source}"),
owner.remediation(),
)
})
}
pub fn check_root(
context: &Context,
store: &dyn Store,
owner: &RootOwner,
raw: &str,
) -> Result<LocalAbsolutePath, CliError> {
let root = parse_root(raw, owner)?;
let host = super::host::local_host(store)?;
let current = host_root(context.paths(), host.as_ref());
let policies = store.policies().map_err(read_failure)?;
preflight_against_everything(context.paths(), ¤t, &policies)
.check(owner, &root)
.map_err(|source| unusable(source, owner))?;
Ok(root)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RootChange {
pub previous: HostRoot,
pub current: HostRoot,
pub created: Option<PathBuf>,
pub retained: Option<LocalAbsolutePath>,
pub service_access: Option<ServiceAccessWarning>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServiceAccessWarning {
pub program: Option<PathBuf>,
}
const fn service_may_be_denied(start_mode: StartMode, on_gated_volume: bool) -> bool {
matches!(start_mode, StartMode::Boot) && on_gated_volume
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum RegisteredService {
None,
Known {
start_mode: StartMode,
binary: PathBuf,
},
Unreadable,
}
fn registered_service(paths: &AppPaths) -> RegisteredService {
match InstallRecord::read(paths) {
Ok(Some(record)) => RegisteredService::Known {
start_mode: record.start_mode,
binary: record.binary,
},
Ok(None) => RegisteredService::None,
Err(ServiceError::RecordNotPermitted { .. }) => RegisteredService::Unreadable,
Err(_) => RegisteredService::None,
}
}
fn service_access_warning(
paths: &AppPaths,
host: &Host,
root: &LocalAbsolutePath,
) -> Option<ServiceAccessWarning> {
if !is_on_privacy_gated_volume(root.as_path()) {
return None;
}
let (start_mode, program) = match registered_service(paths) {
RegisteredService::None => return None,
RegisteredService::Known { start_mode, binary } => (start_mode, Some(binary)),
RegisteredService::Unreadable => (host.service_start_mode, None),
};
if !service_may_be_denied(start_mode, true) {
return None;
}
Some(ServiceAccessWarning { program })
}
pub fn set_host_runner_root(
context: &Context,
store: &dyn Store,
requested: Option<LocalAbsolutePath>,
) -> Result<RootChange, CliError> {
let host = super::host::local_host_or_create(context, store)?;
let previous = host_root(context.paths(), Some(&host));
let affected = host_affected_attempts(store)?;
if !affected.is_empty() {
return Err(CliError::with_remedy(
Failure::Conflict,
affected.refusal("the host runner root"),
"runner-manager status",
));
}
let created = match &requested {
Some(root) => {
let policies = store.policies().map_err(read_failure)?;
validated_leaf(
context.paths(),
&previous,
&policies,
&RootOwner::Host,
root,
)?
}
None => None,
};
store
.set_runner_root_override(
host.id,
previous.configured.as_ref(),
requested.as_ref(),
affected.total(),
)
.map_err(|source| write_failure(source, created.as_deref()))?;
let service_access = requested
.as_ref()
.and_then(|root| service_access_warning(context.paths(), &host, root));
let current = host_root(
context.paths(),
Some(&Host {
runner_root_override: requested,
..host
}),
);
Ok(RootChange {
retained: retained_between(&previous, ¤t),
previous,
current,
created,
service_access,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkspaceChange {
pub target: ScaleTarget,
pub previous: WorkspacePolicy,
pub current: WorkspacePolicy,
pub host_root: HostRoot,
pub created: Option<PathBuf>,
pub retained: Option<LocalAbsolutePath>,
}
impl WorkspaceChange {
#[must_use]
pub fn effective_root(&self) -> String {
self.current.root().map_or_else(
|| self.host_root.rendered(),
|root| root.as_str().to_string(),
)
}
}
pub fn set_repository_workspace(
context: &Context,
store: &dyn Store,
target: &ScaleTarget,
kind: WorkspaceKind,
path: Option<LocalAbsolutePath>,
) -> Result<WorkspaceChange, CliError> {
let policies = store.policies().map_err(read_failure)?;
let mut policy = policies
.iter()
.find(|policy| &policy.target == target)
.cloned()
.ok_or_else(|| {
CliError::with_remedy(
Failure::NotFound,
format!("no policy for {target} exists"),
"runner-manager repo list",
)
})?;
let previous = policy.workspace_policy().clone();
let expected_revision = policy.revision();
let host = super::host::local_host(store)?;
let host_root = host_root(context.paths(), host.as_ref());
let affected = policy_affected_attempts(store, &policy)?;
if !affected.is_empty() {
return Err(CliError::with_remedy(
Failure::Conflict,
affected.refusal(&format!("the workspace setting for {target}")),
"runner-manager status",
));
}
let owner = RootOwner::Repository(target.slug());
let requested = match (kind, path) {
(WorkspaceKind::Ephemeral, None) => WorkspacePolicy::Ephemeral,
(WorkspaceKind::Ephemeral, Some(_)) => {
return Err(ephemeral_rejects_a_path(target));
}
(WorkspaceKind::Persistent, Some(root)) => {
WorkspacePolicy::persistent(root, target.scope()).map_err(|source| {
CliError::with_remedy(
Failure::InvalidArgument,
source.to_string(),
"runner-manager repo set-workspace OWNER/REPO --mode ephemeral",
)
})?
}
(WorkspaceKind::Persistent, None) => {
return Err(CliError::with_remedy(
Failure::InvalidArgument,
"a persistent workspace needs the directory its slots live in",
"runner-manager repo set-workspace OWNER/REPO --mode persistent --path <PATH>",
));
}
};
let created = match requested.root() {
Some(root) => validated_leaf(context.paths(), &host_root, &policies, &owner, root)?,
None => None,
};
policy
.set_workspace_policy(requested.clone())
.map_err(|source| CliError::new(Failure::InvalidArgument, source.to_string()))?;
if policy.revision() != expected_revision {
store
.update_policy_confirming_uncleaned_count(&policy, expected_revision, affected.total())
.map_err(|source| write_failure(source, created.as_deref()))?;
}
Ok(WorkspaceChange {
target: target.clone(),
retained: retained_root(&previous, &requested),
previous,
current: requested,
host_root,
created,
})
}
pub fn write_root_change(out: &mut dyn Write, change: &RootChange) -> Result<(), CliError> {
let failed = write_failed("this runner root result");
writeln!(
out,
"Runner root {}.",
if change.current.configured.is_some() {
"configured"
} else {
"reset to the platform default"
}
)
.map_err(&failed)?;
writeln!(
out,
"Previous: {} ({})",
change.previous.rendered(),
change.previous.source()
)
.map_err(&failed)?;
writeln!(
out,
"Current: {} ({})",
change.current.rendered(),
change.current.source()
)
.map_err(&failed)?;
if let Some(created) = &change.created {
writeln!(out, "Created: {}", created.display()).map_err(&failed)?;
}
writeln!(
out,
"New ephemeral attempts will use this path. No existing directory was moved or deleted."
)
.map_err(&failed)?;
if let Some(retained) = &change.retained {
writeln!(
out,
"Retained: {retained} still holds whatever was left there."
)
.map_err(&failed)?;
}
Ok(())
}
pub fn write_service_access_warning(
out: &mut dyn Write,
styling: Styling,
warning: &ServiceAccessWarning,
settings_opened: bool,
) -> Result<(), CliError> {
let failed = write_failed("this runner root warning");
let mut line = |text: &str| writeln!(out, "{text}").map_err(&failed);
line("")?;
line(&styling.caution("The service cannot use this path yet."))?;
line("This root is on a separate volume, and this host starts the agent at boot as")?;
line("`root`. macOS withholds such volumes from a background service until the")?;
line("program is granted Full Disk Access, and a service cannot ask for it: the")?;
line("refusal is silent, and every launch fails with nothing on screen to say why.")?;
line("")?;
line(&styling.step("To fix it now:"))?;
if settings_opened {
line(" 1. In the window that just opened -- System Settings > Privacy &")?;
line(" Security > Full Disk Access -- select `+` and add this exact program:")?;
} else {
line(" 1. Open System Settings > Privacy & Security > Full Disk Access, then")?;
line(" select `+` and add this exact program:")?;
}
match &warning.program {
Some(program) => {
line(&format!(
" {}",
styling.code(&program.display().to_string())
))?;
line(" It is the copy the service runs, not the one on your PATH.")?;
}
None => {
line(" the binary this host's service is registered to run, which")?;
line(&format!(
" {} prints as `binary`. It is not the one on your PATH.",
styling.code("sudo runner-manager service status")
))?;
}
}
line(" 2. Restart the service so it picks the grant up:")?;
line(&format!(
" {}",
styling
.code("sudo runner-manager service uninstall && sudo runner-manager service install")
))?;
line("")?;
line("The grant follows the binary, not the path: an upgrade that replaces the service")?;
line("binary revokes it, and it has to be granted again to the new one. Once the agent")?;
line("has tried and been refused, that refusal is reported by")?;
line(&format!(
" {}",
styling.code("runner-manager service status")
))?;
line("")?;
line("Or avoid the grant entirely by running the agent as you, in your own session:")?;
line(&format!(
" {}",
styling.code("runner-manager service install --start-at login")
))?;
Ok(())
}
pub fn write_workspace_change(
out: &mut dyn Write,
change: &WorkspaceChange,
) -> Result<(), CliError> {
let failed = write_failed("this workspace result");
writeln!(out, "Workspace mode: {}", change.current.kind()).map_err(&failed)?;
writeln!(out, "Workspace root: {}", change.effective_root()).map_err(&failed)?;
if change.current.is_persistent() {
if let Some(created) = &change.created {
writeln!(out, "Created: {}", created.display()).map_err(&failed)?;
}
writeln!(out, "Slots: created on demand as s1, s2, ...").map_err(&failed)?;
writeln!(out, "Retained: each slot's _work directory").map_err(&failed)?;
writeln!(
out,
"Disposable: runner binaries, JIT handoff, and lifecycle files"
)
.map_err(&failed)?;
} else {
writeln!(
out,
"Root source: {} ({})",
change.host_root.rendered(),
change.host_root.source()
)
.map_err(&failed)?;
}
if let Some(retained) = &change.retained {
writeln!(
out,
"Left in place: every slot under {retained} remains on disk, including its _work \
directory."
)
.map_err(&failed)?;
}
writeln!(out, "No existing directory was moved or deleted.").map_err(&failed)?;
if change.current.is_persistent() {
write_trust_warning(out)?;
}
Ok(())
}
fn retained_between(previous: &HostRoot, current: &HostRoot) -> Option<LocalAbsolutePath> {
let old = previous.effective.clone()?;
match ¤t.effective {
Some(new) if new == &old => None,
_ => Some(old),
}
}
fn retained_root(
previous: &WorkspacePolicy,
current: &WorkspacePolicy,
) -> Option<LocalAbsolutePath> {
let old = previous.root()?.clone();
match current.root() {
Some(new) if new == &old => None,
_ => Some(old),
}
}
fn read_failure(source: StoreError) -> CliError {
CliError::with_remedy(
Failure::LocalState,
format!("cannot read this host's local database: {source}"),
"runner-manager host show",
)
}
fn write_failure(source: StoreError, created: Option<&std::path::Path>) -> CliError {
let class = if source.is_conflict() {
Failure::Conflict
} else {
Failure::LocalState
};
let message = format!("{source}{}", leftover_note(created));
CliError::with_remedy(class, message, "runner-manager status, then retry")
}
fn leftover_note(created: Option<&std::path::Path>) -> String {
created.map_or_else(String::new, |leaf| {
format!(
" The empty directory {} was created before the write was refused and has been \
left in place; remove it yourself if you do not want it.",
leaf.display()
)
})
}
fn unusable(
source: runner_manager_platform::runner_root::RunnerRootError,
owner: &RootOwner,
) -> CliError {
CliError::with_remedy(
Failure::InvalidArgument,
source.to_string(),
owner.remediation(),
)
}
#[must_use]
pub const fn scope_token(scope: TargetScope) -> &'static str {
match scope {
TargetScope::Repository => "repository",
TargetScope::Organization => "organization",
}
}
#[cfg(test)]
mod tests {
use super::*;
use runner_manager_domain::attempt::{AttemptOutcome, FailureReason, RunnerAttempt};
use runner_manager_domain::model::{AttemptId, PolicyId};
#[test]
fn a_boot_service_pointed_off_the_startup_volume_is_warned_about() {
assert!(
service_may_be_denied(StartMode::Boot, true),
"a boot-mode daemon is the one identity that cannot ask for consent"
);
}
#[test]
fn no_install_record_means_no_registered_service() {
let data_dir = tempfile::tempdir().expect("a temporary directory");
let paths = AppPaths::rooted_at(data_dir.path());
let bin = paths.state_dir().join("bin");
std::fs::create_dir_all(&bin).expect("the service binary directory");
std::fs::write(bin.join("runner-manager"), b"leftover").expect("the leftover copy");
assert_eq!(
registered_service(&paths),
RegisteredService::None,
"only the install record says a service is registered"
);
}
#[test]
fn an_unknown_program_is_described_rather_than_guessed() {
let mut out = Vec::new();
write_service_access_warning(
&mut out,
Styling::plain(),
&ServiceAccessWarning { program: None },
false,
)
.expect("the block is written");
let text = String::from_utf8(out).expect("utf-8");
assert!(
text.contains("service status"),
"an unknown path must be described, not omitted: {text}"
);
assert!(
!text.contains("bin/runner-manager"),
"a guessed path is worse than none -- granting it changes nothing: {text}"
);
assert!(
!text.contains("window that just opened"),
"nothing opened, so nothing may claim it did: {text}"
);
}
#[test]
fn every_other_combination_is_left_alone() {
assert!(
!service_may_be_denied(StartMode::Boot, false),
"the startup disk is not gated at all"
);
assert!(
!service_may_be_denied(StartMode::Login, true),
"a login-mode agent runs as the operator, in a session that can still prompt"
);
assert!(
!service_may_be_denied(StartMode::Login, false),
"nothing to warn about"
);
}
fn ts(secs: i64) -> runner_manager_domain::model::Timestamp {
chrono::DateTime::from_timestamp(secs, 0).expect("a valid timestamp")
}
fn an_attempt(terminal: bool) -> RunnerAttempt {
let mut attempt = RunnerAttempt::allocate(
AttemptId::new_random(),
PolicyId::new_random(),
std::path::PathBuf::from(a_root(&root_text("rman")).as_path()),
ts(0),
);
if terminal {
attempt
.conclude(
AttemptOutcome::failed(FailureReason::ProcessStartFailed),
ts(1),
)
.expect("allocated -> failed is a legal conclusion");
}
attempt
}
#[test]
fn the_two_counts_follow_the_domains_own_terminal_predicate() {
let attempts = [an_attempt(false), an_attempt(false), an_attempt(true)];
let counts = AffectedAttempts::of(attempts.iter());
assert_eq!(counts.active, 2);
assert_eq!(counts.cleanup_blocked, 1);
assert_eq!(
counts.total(),
3,
"the total is what the store's fence is given, so it must be the whole uncleaned \
set and not just the active half"
);
assert!(!counts.is_empty());
assert!(AffectedAttempts::default().is_empty());
}
#[test]
fn the_refusal_names_both_counts_separately() {
let only_active = AffectedAttempts {
active: 1,
cleanup_blocked: 0,
};
let text = only_active.refusal("the host runner root");
assert!(text.contains("1 active"), "{text}");
assert!(text.contains("0 awaiting cleanup"), "{text}");
assert!(text.contains("Nothing was changed."), "{text}");
let only_blocked = AffectedAttempts {
active: 0,
cleanup_blocked: 2,
};
let text = only_blocked.refusal("the workspace setting for octo/repo");
assert!(text.contains("0 active"), "{text}");
assert!(text.contains("2 awaiting cleanup"), "{text}");
assert!(text.contains("octo/repo"), "{text}");
}
#[test]
fn the_source_has_one_spelling_for_people_and_one_for_scripts() {
assert_eq!(RootSource::PlatformDefault.as_str(), "platform-default");
assert_eq!(RootSource::PlatformDefault.as_token(), "platform_default");
assert_eq!(RootSource::Configured.as_str(), "configured");
assert_eq!(RootSource::Configured.as_token(), "configured");
assert_eq!(RootSource::Configured.to_string(), "configured");
for source in [RootSource::PlatformDefault, RootSource::Configured] {
assert!(
!source.as_token().contains('-'),
"every other enumerated value in status --json is snake_case: {}",
source.as_token()
);
}
}
fn a_root(text: &str) -> LocalAbsolutePath {
LocalAbsolutePath::new(text).expect("a valid fixture path")
}
fn root_text(leaf: &str) -> String {
if cfg!(windows) {
format!(r"C:\{leaf}")
} else {
format!("/srv/{leaf}")
}
}
#[test]
fn a_configured_root_is_its_own_effective_value_and_says_so() {
let configured = a_root(&root_text("elsewhere"));
let root = HostRoot {
configured: Some(configured.clone()),
effective: Some(configured.clone()),
unavailable: None,
};
assert_eq!(root.source(), RootSource::Configured);
assert_eq!(root.effective_text(), Some(configured.as_str()));
assert_eq!(root.rendered(), configured.as_str());
}
#[test]
fn an_unresolvable_default_renders_its_reason_instead_of_a_path() {
let root = HostRoot {
configured: None,
effective: None,
unavailable: Some("the system directory is unreadable".to_string()),
};
assert_eq!(root.source(), RootSource::PlatformDefault);
assert_eq!(root.effective_text(), None);
assert!(
root.rendered().contains("unavailable"),
"{}",
root.rendered()
);
assert!(
root.rendered()
.contains("the system directory is unreadable"),
"the reason has to travel with the row, or the operator learns nothing: {}",
root.rendered()
);
}
#[test]
fn only_a_root_that_actually_moved_is_reported_as_retained() {
let old = a_root(&root_text("rman"));
let new = a_root(&root_text("elsewhere"));
let at = |root: &LocalAbsolutePath| HostRoot {
configured: Some(root.clone()),
effective: Some(root.clone()),
unavailable: None,
};
assert_eq!(retained_between(&at(&old), &at(&old)), None);
assert_eq!(retained_between(&at(&old), &at(&new)), Some(old.clone()));
let ephemeral = WorkspacePolicy::Ephemeral;
let persistent = WorkspacePolicy::persistent(
old.clone(),
runner_manager_domain::model::TargetScope::Repository,
)
.expect("a repository may hold a persistent workspace");
assert_eq!(retained_root(&ephemeral, &persistent), None);
assert_eq!(retained_root(&persistent, &persistent), None);
assert_eq!(
retained_root(&persistent, &ephemeral),
Some(old),
"returning to ephemeral leaves every slot on disk, and the operator is told so"
);
}
#[test]
fn a_lost_race_names_the_directory_it_left_behind() {
assert_eq!(leftover_note(None), "");
let note = leftover_note(Some(std::path::Path::new("/srv/ws")));
assert!(note.contains("/srv/ws"), "{note}");
assert!(note.contains("left in place"), "{note}");
assert!(
!note.contains("removed it") && !note.contains("deleted"),
"the note must not claim a deletion this command is forbidden to perform: {note}"
);
}
#[test]
fn the_trust_warning_states_every_required_clause() {
let text = PERSISTENT_TRUST_WARNING.join("\n");
for clause in [
"_work",
"branch and job boundaries",
"untrusted fork or pull-request",
"does not delete old directories",
"clean: false",
] {
assert!(text.contains(clause), "missing {clause:?} from:\n{text}");
}
let mut buffer = Vec::new();
write_trust_warning(&mut buffer).expect("writing to a Vec");
assert_eq!(
String::from_utf8(buffer).expect("utf-8"),
format!("{text}\n"),
"the rendered warning must be the constant, so a screen and a command cannot \
paraphrase it differently"
);
}
#[test]
fn the_scope_token_is_the_one_the_status_document_already_emits() {
assert_eq!(scope_token(TargetScope::Repository), "repository");
assert_eq!(scope_token(TargetScope::Organization), "organization");
}
fn snapshot_of(
render: impl Fn(&mut Vec<u8>) -> Result<(), CliError>,
paths: &[(&str, &str)],
) -> String {
let mut buffer = Vec::new();
render(&mut buffer).expect("writing to a Vec");
let mut text = String::from_utf8(buffer).expect("utf-8");
for (actual, placeholder) in paths {
text = text.replace(actual, placeholder);
}
text
}
#[test]
fn the_host_root_success_block_is_journey_2s() {
let previous = a_root(&root_text("rman"));
let current = a_root(&root_text("runners"));
let change = RootChange {
service_access: None,
previous: HostRoot {
configured: None,
effective: Some(previous.clone()),
unavailable: None,
},
current: HostRoot {
configured: Some(current.clone()),
effective: Some(current.clone()),
unavailable: None,
},
created: Some(current.as_path().to_path_buf()),
retained: Some(previous.clone()),
};
insta::assert_snapshot!(
snapshot_of(
|out| write_root_change(out, &change),
&[
(current.as_str(), "<NEW>"),
(previous.as_str(), "<OLD>"),
],
),
@r###"
Runner root configured.
Previous: <OLD> (platform-default)
Current: <NEW> (configured)
Created: <NEW>
New ephemeral attempts will use this path. No existing directory was moved or deleted.
Retained: <OLD> still holds whatever was left there.
"###
);
}
#[test]
fn the_persistent_success_block_is_journey_3s_and_carries_the_whole_warning() {
let old = a_root(&root_text("old-cache"));
let new = a_root(&root_text("ci-cache"));
let change = WorkspaceChange {
target: ScaleTarget::repository("octo/repo").expect("a valid slug"),
previous: WorkspacePolicy::persistent(old.clone(), TargetScope::Repository)
.expect("a repository may hold one"),
current: WorkspacePolicy::persistent(new.clone(), TargetScope::Repository)
.expect("a repository may hold one"),
host_root: HostRoot {
configured: None,
effective: Some(a_root(&root_text("rman"))),
unavailable: None,
},
created: Some(new.as_path().to_path_buf()),
retained: Some(old.clone()),
};
insta::assert_snapshot!(
snapshot_of(
|out| write_workspace_change(out, &change),
&[(new.as_str(), "<NEW>"), (old.as_str(), "<OLD>")],
),
@r###"
Workspace mode: persistent
Workspace root: <NEW>
Created: <NEW>
Slots: created on demand as s1, s2, ...
Retained: each slot's _work directory
Disposable: runner binaries, JIT handoff, and lifecycle files
Left in place: every slot under <OLD> remains on disk, including its _work directory.
No existing directory was moved or deleted.
warning: a persistent workspace is a trusted-workflow optimization, not isolation.
- files under _work are an input to later jobs on the same slot;
- executable and generated content can cross branch and job boundaries;
- do not enable it for untrusted fork or pull-request workflows;
- changing or disabling persistence does not delete old directories;
- `actions/checkout` still cleans the workspace, including Git-ignored files,
unless the workflow sets `clean: false`.
"###
);
}
#[test]
fn the_ephemeral_success_block_is_journey_4s() {
let old = a_root(&root_text("ci-cache"));
let host = a_root(&root_text("rman"));
let change = WorkspaceChange {
target: ScaleTarget::repository("octo/repo").expect("a valid slug"),
previous: WorkspacePolicy::persistent(old.clone(), TargetScope::Repository)
.expect("a repository may hold one"),
current: WorkspacePolicy::Ephemeral,
host_root: HostRoot {
configured: None,
effective: Some(host.clone()),
unavailable: None,
},
created: None,
retained: Some(old.clone()),
};
insta::assert_snapshot!(
snapshot_of(
|out| write_workspace_change(out, &change),
&[(old.as_str(), "<OLD>"), (host.as_str(), "<HOST>")],
),
@r###"
Workspace mode: ephemeral
Workspace root: <HOST>
Root source: <HOST> (platform-default)
Left in place: every slot under <OLD> remains on disk, including its _work directory.
No existing directory was moved or deleted.
"###
);
}
}
#[cfg(test)]
mod race {
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
use runner_manager_domain::attempt::RunnerAttempt;
use runner_manager_domain::model::{AttemptId, HostId, PolicyId, ScaleTarget};
use runner_manager_domain::policy::ScalePolicy;
use runner_manager_domain::store::{SqliteStore, Store, StoreError};
use runner_manager_domain::workspace::{WorkspaceKind, WorkspacePolicy};
use runner_manager_testkit::fixtures;
use super::{Failure, LocalAbsolutePath, set_host_runner_root, set_repository_workspace};
use crate::cli::Context;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum At {
HostCount,
PolicyCount,
}
type ConcurrentWrite = Box<dyn FnOnce(&SqliteStore) + Send>;
struct Interleaved {
inner: SqliteStore,
at: At,
concurrent: Mutex<Option<ConcurrentWrite>>,
put_hosts: AtomicUsize,
}
impl std::fmt::Debug for Interleaved {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Interleaved").field("at", &self.at).finish()
}
}
impl Interleaved {
fn new(
inner: SqliteStore,
at: At,
concurrent: impl FnOnce(&SqliteStore) + Send + 'static,
) -> Self {
Self {
inner,
at,
concurrent: Mutex::new(Some(Box::new(concurrent))),
put_hosts: AtomicUsize::new(0),
}
}
fn interleave(&self, at: At) {
if at != self.at {
return;
}
let taken = self
.concurrent
.lock()
.expect("no test panics while holding this")
.take();
if let Some(concurrent) = taken {
concurrent(&self.inner);
}
}
}
impl Store for Interleaved {
fn put_host(&self, host: &runner_manager_domain::model::Host) -> Result<(), StoreError> {
self.put_hosts.fetch_add(1, Ordering::Relaxed);
self.inner.put_host(host)
}
fn host(
&self,
id: HostId,
) -> Result<Option<runner_manager_domain::model::Host>, StoreError> {
self.inner.host(id)
}
fn hosts(&self) -> Result<Vec<runner_manager_domain::model::Host>, StoreError> {
self.inner.hosts()
}
fn set_runner_root_override(
&self,
id: HostId,
expected: Option<&LocalAbsolutePath>,
new_root: Option<&LocalAbsolutePath>,
expected_uncleaned: u16,
) -> Result<(), StoreError> {
self.inner
.set_runner_root_override(id, expected, new_root, expected_uncleaned)
}
fn insert_policy(&self, policy: &ScalePolicy) -> Result<(), StoreError> {
self.inner.insert_policy(policy)
}
fn update_policy(
&self,
policy: &ScalePolicy,
expected_revision: u64,
) -> Result<(), StoreError> {
self.inner.update_policy(policy, expected_revision)
}
fn update_policy_confirming_active_count(
&self,
policy: &ScalePolicy,
expected_revision: u64,
expected_active: u16,
) -> Result<(), StoreError> {
self.inner.update_policy_confirming_active_count(
policy,
expected_revision,
expected_active,
)
}
fn update_policy_confirming_uncleaned_count(
&self,
policy: &ScalePolicy,
expected_revision: u64,
expected_uncleaned: u16,
) -> Result<(), StoreError> {
self.inner.update_policy_confirming_uncleaned_count(
policy,
expected_revision,
expected_uncleaned,
)
}
fn remove_policy(&self, id: PolicyId, expected_revision: u64) -> Result<(), StoreError> {
self.inner.remove_policy(id, expected_revision)
}
fn policy(&self, id: PolicyId) -> Result<Option<ScalePolicy>, StoreError> {
self.inner.policy(id)
}
fn policies(&self) -> Result<Vec<ScalePolicy>, StoreError> {
self.inner.policies()
}
fn record_attempt(&self, attempt: &RunnerAttempt) -> Result<(), StoreError> {
self.inner.record_attempt(attempt)
}
fn attempt(&self, id: AttemptId) -> Result<Option<RunnerAttempt>, StoreError> {
self.inner.attempt(id)
}
fn attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError> {
self.inner.attempts()
}
fn attempts_for_policy(
&self,
policy_id: PolicyId,
) -> Result<Vec<RunnerAttempt>, StoreError> {
self.inner.attempts_for_policy(policy_id)
}
fn active_attempts_for_policy(
&self,
policy_id: PolicyId,
) -> Result<Vec<RunnerAttempt>, StoreError> {
self.inner.active_attempts_for_policy(policy_id)
}
fn uncleaned_attempts_for_policy(
&self,
policy_id: PolicyId,
) -> Result<Vec<RunnerAttempt>, StoreError> {
let answer = self.inner.uncleaned_attempts_for_policy(policy_id);
self.interleave(At::PolicyCount);
answer
}
fn slot_leases_for_policy(
&self,
policy_id: PolicyId,
) -> Result<Vec<RunnerAttempt>, StoreError> {
self.inner.slot_leases_for_policy(policy_id)
}
fn uncleaned_ephemeral_attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError> {
let answer = self.inner.uncleaned_ephemeral_attempts();
self.interleave(At::HostCount);
answer
}
fn remove_attempt(&self, id: AttemptId) -> Result<bool, StoreError> {
self.inner.remove_attempt(id)
}
}
fn a_context(data_dir: &std::path::Path) -> Context {
Context::resolve(Some(data_dir), &mut std::io::sink())
.expect("a temporary data directory must resolve")
}
fn a_host() -> runner_manager_domain::model::Host {
fixtures::host().capacity(1).build()
}
fn a_policy(host: HostId) -> ScalePolicy {
fixtures::policy()
.host(host)
.repository("octo/repo")
.autoscale("home-win", 1)
.build()
}
fn a_root(parent: &tempfile::TempDir, leaf: &str) -> LocalAbsolutePath {
LocalAbsolutePath::new(
parent
.path()
.join(leaf)
.to_str()
.expect("a temporary path must be UTF-8"),
)
.expect("a temporary path is absolute and local")
}
#[test]
fn a_host_root_change_does_not_roll_back_a_concurrent_capacity_change() {
let data_dir = tempfile::tempdir().expect("a temporary directory");
let roots = tempfile::tempdir().expect("a temporary directory");
let context = a_context(data_dir.path());
let inner = SqliteStore::open_in_memory().expect("an in-memory database");
let host = a_host();
inner.put_host(&host).expect("a fresh host");
let id = host.id;
let store = Interleaved::new(inner, At::HostCount, move |inner| {
let mut concurrent = inner
.host(id)
.expect("readable")
.expect("the host is there");
concurrent.host_capacity = std::num::NonZeroU16::new(9).expect("non-zero");
inner.put_host(&concurrent).expect("the concurrent write");
});
let root = a_root(&roots, "runners");
let change = set_host_runner_root(&context, &store, Some(root.clone()))
.expect("the override did not move, so the fenced write is accepted");
assert_eq!(change.current.configured, Some(root));
let after = store.host(id).expect("readable").expect("still there");
assert_eq!(
after.host_capacity.get(),
9,
"the capacity that landed inside the window must survive: a whole-record write \
would have restored the 1 this handler read"
);
assert_eq!(
store.put_hosts.load(Ordering::Relaxed),
0,
"the handler must reach `set_runner_root_override` and never `put_host`, which \
is the only shape that cannot roll a concurrent column back"
);
}
#[test]
fn a_host_root_change_refuses_when_the_override_moved_under_it() {
let data_dir = tempfile::tempdir().expect("a temporary directory");
let roots = tempfile::tempdir().expect("a temporary directory");
let context = a_context(data_dir.path());
let inner = SqliteStore::open_in_memory().expect("an in-memory database");
let host = a_host();
inner.put_host(&host).expect("a fresh host");
let id = host.id;
let theirs = a_root(&roots, "theirs");
let stolen = theirs.clone();
let store = Interleaved::new(inner, At::HostCount, move |inner| {
inner
.set_runner_root_override(id, None, Some(&stolen), 0)
.expect("the concurrent operator got there first");
});
let mine = a_root(&roots, "mine");
let refused = set_host_runner_root(&context, &store, Some(mine.clone()))
.expect_err("the stored override is no longer the one that was read");
assert_eq!(refused.class(), Failure::Conflict);
assert!(
refused
.message()
.contains("another process changed it first"),
"the refusal must say a race was lost rather than look like an I/O failure: {}",
refused.message()
);
assert!(
refused.message().contains("left in place"),
"and it must name the empty directory it created before the write was refused, \
which `03-migration-rollout.md` forbids it to delete: {}",
refused.message()
);
assert!(
mine.as_path().is_dir(),
"named, and left: the handler may not remove a directory it did not prove it \
created empty in this invocation"
);
let after = store.host(id).expect("readable").expect("still there");
assert_eq!(
after.runner_root_override,
Some(theirs),
"the concurrent value stands; the refused one was never written"
);
}
#[test]
fn a_workspace_change_refuses_when_the_policy_moved_under_it() {
let data_dir = tempfile::tempdir().expect("a temporary directory");
let roots = tempfile::tempdir().expect("a temporary directory");
let context = a_context(data_dir.path());
let inner = SqliteStore::open_in_memory().expect("an in-memory database");
let host = a_host();
inner.put_host(&host).expect("a fresh host");
let policy = a_policy(host.id);
inner.insert_policy(&policy).expect("a fresh policy");
let id = policy.id;
let store = Interleaved::new(inner, At::PolicyCount, move |inner| {
let mut concurrent = inner
.policy(id)
.expect("readable")
.expect("the policy is there");
let revision = concurrent.revision();
concurrent
.set_max_capacity(std::num::NonZeroU16::new(4).expect("non-zero"))
.expect("a legal ceiling change");
inner
.update_policy(&concurrent, revision)
.expect("the concurrent write");
});
let target = ScaleTarget::repository("octo/repo").expect("a valid slug");
let root = a_root(&roots, "ci-cache");
let refused = set_repository_workspace(
&context,
&store,
&target,
WorkspaceKind::Persistent,
Some(root.clone()),
)
.expect_err("the revision this write was built from no longer exists");
assert_eq!(refused.class(), Failure::Conflict);
assert!(
refused.message().contains("left in place"),
"the created leaf is reported: {}",
refused.message()
);
let after = store
.policy(id)
.expect("readable")
.expect("still there")
.clone();
assert_eq!(
after.workspace_policy(),
&WorkspacePolicy::Ephemeral,
"nothing was written, so the repository is still disposable"
);
assert_eq!(
after.max_capacity().map(std::num::NonZeroU16::get),
Some(4),
"and the concurrent ceiling change was not rolled back either"
);
}
#[test]
fn the_shared_handler_refuses_a_path_for_an_ephemeral_workspace() {
let data_dir = tempfile::tempdir().expect("a temporary directory");
let roots = tempfile::tempdir().expect("a temporary directory");
let context = a_context(data_dir.path());
let store = SqliteStore::open_in_memory().expect("an in-memory database");
let host = a_host();
store.put_host(&host).expect("a fresh host");
let policy = a_policy(host.id);
store.insert_policy(&policy).expect("a fresh policy");
let target = ScaleTarget::repository("octo/repo").expect("a valid slug");
let root = a_root(&roots, "ci-cache");
let refused = set_repository_workspace(
&context,
&store,
&target,
WorkspaceKind::Ephemeral,
Some(root.clone()),
)
.expect_err("an ephemeral workspace has no slots to place");
assert_eq!(refused.class(), Failure::InvalidArgument);
assert!(
refused.message().contains("nothing was changed"),
"the refusal must say the setting is untouched: {}",
refused.message()
);
assert!(
!root.as_path().exists(),
"and a refused mutation creates no directory"
);
assert_eq!(
store
.policy(policy.id)
.expect("readable")
.expect("still there")
.workspace_policy(),
&WorkspacePolicy::Ephemeral,
);
}
}