use crate::{
config::{self, Config, Paths},
docker::Docker,
github::{Github, RemoteRunner},
ipc,
protocol::{Request, Response},
state::{PoolState, RetryScope, Run, State, Store, now},
};
use anyhow::{Context, Result, ensure};
use fs2::FileExt;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
#[cfg(unix)]
use std::os::unix::fs::OpenOptionsExt;
use std::{fs, future::Future, time::Duration};
use tokio::{
io::{AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader},
sync::{mpsc, oneshot, watch},
};
use uuid::Uuid;
const UNEXPECTED_EXIT_ERROR: &str =
"runner exited before completing a job; inspect archived diagnostics";
const MAX_RETRY_SECONDS: u64 = 600;
struct Envelope {
request: Request,
reply: oneshot::Sender<Result<Value>>,
}
#[derive(Debug)]
struct ReconcileInterrupted;
impl std::fmt::Display for ReconcileInterrupted {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("reconciliation interrupted by a control request")
}
}
impl std::error::Error for ReconcileInterrupted {}
pub fn hash(text: &str) -> String {
format!("{:x}", Sha256::digest(text.as_bytes()))
}
pub fn lock(paths: &Paths) -> Result<fs::File> {
let mut options = fs::OpenOptions::new();
options.create(true).truncate(false).read(true).write(true);
#[cfg(unix)]
options.mode(0o600).custom_flags(libc::O_NOFOLLOW);
let f = options.open(paths.home.join("manager.lock"))?;
f.try_lock_exclusive()
.context("manager is already running or another command holds the lock")?;
Ok(f)
}
pub struct Engine {
pub state: State,
store: Store,
paths: Paths,
github: Github,
cursor: usize,
boot_time: Option<u64>,
commands: Option<mpsc::Receiver<Envelope>>,
snapshots: Option<watch::Sender<State>>,
epoch: u64,
reconcile_epoch: Option<u64>,
}
impl Engine {
pub fn open(paths: Paths) -> Result<Self> {
let store = Store::open(&paths.home.join("state.sqlite"))?;
let state = match store.load()? {
Some(state) => state,
None => {
let text =
fs::read_to_string(paths.config()).context("run runnerctl init first")?;
let state = State::new(Config::parse(&text)?, hash(&text));
store.save(&state)?;
state
}
};
Ok(Self {
state,
store,
paths,
github: Github::new()?,
cursor: 0,
boot_time: None,
commands: None,
snapshots: None,
epoch: 0,
reconcile_epoch: None,
})
}
fn save(&self) -> Result<()> {
self.store.save(&self.state)?;
if let Some(snapshots) = &self.snapshots {
snapshots.send_replace(self.state.clone());
}
Ok(())
}
fn clean_file(&self) -> Result<()> {
ensure!(
hash(&fs::read_to_string(self.paths.config())?) == self.state.file_hash,
"config.toml has unapplied changes; run config apply first"
);
Ok(())
}
fn commit_config(&mut self, next: Config) -> Result<()> {
next.validate()?;
let text = toml::to_string_pretty(&next)?;
let mut state = self.state.clone();
state.config = next;
state.revision += 1;
state.file_hash = hash(&text);
// SQLite is authoritative. If interrupted before file write, config apply can
// recover explicitly; startup never silently overwrites user-edited TOML.
self.store.save(&state)?;
self.state = state;
config::atomic_write(&self.paths.config(), text.as_bytes())?;
self.github.clear();
if let Some(snapshots) = &self.snapshots {
snapshots.send_replace(self.state.clone());
}
Ok(())
}
fn targets(&self, pool: Option<String>, all: bool) -> Result<Vec<String>> {
ensure!(pool.is_some() != all, "specify a pool or --all");
if let Some(pool) = pool {
ensure!(self.state.pools.contains_key(&pool), "unknown pool: {pool}");
Ok(vec![pool])
} else {
Ok(self.state.pools.keys().cloned().collect())
}
}
pub async fn handle(&mut self, request: Request) -> Result<Value> {
match &request {
Request::Doctor => return diagnose(&self.state, &self.github, &self.paths).await,
Request::Logs { pool, id } => return runner_logs(&self.state, pool, id).await,
Request::Upgrade { image, .. } => {
config::validate_image(image)?;
Docker::image(image).await?;
}
_ => {}
}
self.handle_local(request)
}
fn handle_local(&mut self, request: Request) -> Result<Value> {
let mutating = !matches!(&request, Request::Status { .. });
let result = self.handle_inner(request);
if result.is_err() {
// A failed mutation must not leave uncommitted changes in memory.
self.state = self.store.load()?.context("state disappeared")?;
if let Some(snapshots) = &self.snapshots {
snapshots.send_replace(self.state.clone());
}
}
if mutating {
// Even a failed file write can follow a successful SQLite commit.
self.epoch = self.epoch.wrapping_add(1);
}
result
}
// State has a single writer. External work remains in flight while local
// requests are accepted; a mutation restarts reconciliation after it finishes.
async fn external<T>(&mut self, future: impl Future<Output = Result<T>>) -> Result<T> {
if self.reconcile_epoch.is_some_and(|e| e != self.epoch) {
return Err(ReconcileInterrupted.into());
}
tokio::pin!(future);
let result = loop {
let Some(commands) = &mut self.commands else {
break future.await;
};
tokio::select! {
result = &mut future => break result,
envelope = commands.recv() => {
if let Some(envelope) = envelope {
let result = self.handle_local(envelope.request);
let _ = envelope.reply.send(result);
} else { break future.await; }
}
}
};
if self.reconcile_epoch.is_some_and(|e| e != self.epoch) {
return Err(ReconcileInterrupted.into());
}
result
}
fn handle_inner(&mut self, request: Request) -> Result<Value> {
if request.changes_config() {
self.clean_file()?;
}
match request {
Request::Status { pool } => {
if let Some(p) = &pool {
ensure!(self.state.pools.contains_key(p), "unknown pool: {p}");
}
let pools: Vec<Value> = self.state.config.pools.iter().filter(|p| pool.as_ref().is_none_or(|n| n == &p.name)).map(|p| {
let state = &self.state.pools[&p.name];
let runs: Vec<&Run> = self.state.runs.values().filter(|r| r.pool_id == state.id).collect();
let count = |phase: &str| runs.iter().filter(|r| r.phase == phase).count();
json!({"name":p.name,"organization":p.organization,"repository":p.repository,"target":p.target(),"replicas":p.replicas,"intent":if state.deleting {"deleting"} else if state.running {"running"} else {"stopped"},"active":runs.len(),"starting":count("creating")+count("starting"),"idle":count("idle"),"busy":count("busy"),"unknown":count("unknown"),"draining":runs.iter().filter(|r|r.retiring).count(),"generation":state.generation,"error":state.error,"retry_at":state.retry_at,"observed_at":state.observed_at,"config":p,"runners":runs.iter().map(|r|json!({"id":r.id,"phase":r.phase,"retiring":r.retiring,"generation":r.generation,"drain_reason":if !r.retiring {None} else if r.phase=="idle" {Some("waiting_for_next_job")} else if r.phase=="busy" {Some("job_in_progress")} else {Some("awaiting_observation_or_cleanup")} })).collect::<Vec<_>>()})
}).collect();
Ok(json!({"revision":self.state.revision,"pools":pools}))
}
Request::Add { pool } => {
ensure!(
!self.state.pools.contains_key(&pool.name),
"pool already exists"
);
let mut next = self.state.config.clone();
next.pools.push(pool.clone());
next.validate()?;
self.state
.pools
.insert(pool.name.clone(), PoolState::default());
self.commit_config(next)?;
Ok(json!({"added":pool.name,"intent":"stopped"}))
}
Request::Start { pool, all } => {
let targets = self.targets(pool, all)?;
ensure!(
targets.iter().all(|n| !self.state.pools[n].deleting),
"a target pool is being deleted"
);
for name in &targets {
let p = self.state.pools.get_mut(name).unwrap();
p.running = true;
p.retry_at = 0;
p.failures = 0;
p.error = None;
}
self.github.clear();
self.save()?;
Ok(json!({"started":targets}))
}
Request::Stop { pool, all, force } => {
let targets = self.targets(pool, all)?;
for name in &targets {
let p = self.state.pools.get_mut(name).unwrap();
p.running = false;
if force {
p.retry_at = 0;
}
for r in self.state.runs.values_mut().filter(|r| r.pool_id == p.id) {
r.retiring = true;
r.force |= force;
}
}
self.save()?;
let pending = targets.iter().any(|n| {
self.state
.runs
.values()
.any(|r| r.pool_id == self.state.pools[n].id)
});
Ok(
json!({"stopped":targets,"pending":pending,"mode":if force {"force"} else {"drain"},"message":if force {"Termination requested; running jobs may be interrupted. Registration cleanup waits for GitHub connectivity."} else {"Replacements disabled. Existing idle runners can still accept one job before removal; busy jobs finish naturally. Use --wait to watch cleanup, or --force to terminate work."}}),
)
}
Request::Scale { pool, replicas } => {
let mut next = self.state.config.clone();
let p = next
.pools
.iter_mut()
.find(|p| p.name == pool)
.context("unknown pool")?;
ensure!(!self.state.pools[&pool].deleting, "pool is being deleted");
p.replicas = replicas;
self.commit_config(next)?;
self.mark_retiring(&pool);
self.save()?;
Ok(
json!({"pool":pool,"replicas":replicas,"message":"Capacity updated. Excess runners drain after their next job; stopped pools remain stopped."}),
)
}
Request::Upgrade { pool, image } => {
let mut next = self.state.config.clone();
let p = next
.pools
.iter_mut()
.find(|p| p.name == pool)
.context("unknown pool")?;
ensure!(!self.state.pools[&pool].deleting, "pool is being deleted");
p.image = image;
next.validate()?;
self.state.pools.get_mut(&pool).unwrap().generation += 1;
self.commit_config(next)?;
self.mark_retiring(&pool);
self.save()?;
Ok(
json!({"upgrading":pool,"message":"Image updated. Existing runners drain after their next job; replacements use the new image."}),
)
}
Request::Remove { pool, force } => {
let p = self.state.pools.get_mut(&pool).context("unknown pool")?;
p.running = false;
p.deleting = true;
if force {
p.retry_at = 0;
}
for r in self.state.runs.values_mut().filter(|r| r.pool_id == p.id) {
r.retiring = true;
r.force |= force;
}
self.save()?;
Ok(
json!({"removing":pool,"pending":true,"mode":if force {"force"} else {"drain"},"message":if force {"Removal and termination requested; running jobs may be interrupted. Registration cleanup waits for GitHub connectivity."} else {"Removal requested. Idle runners can accept one final job; cleanup completes after all runners exit."}}),
)
}
Request::Auth {
profile,
credential,
app,
} => {
ensure!(config::valid_name(&profile), "invalid profile");
let mut next = self.state.config.clone();
next.auth
.insert(profile.clone(), config::Auth { credential, app });
self.commit_config(next)?;
for (name, p) in &mut self.state.pools {
if self
.state
.config
.pools
.iter()
.any(|c| c.name == *name && c.auth == profile)
{
p.retry_at = 0;
p.error = None;
p.failures = 0;
}
}
self.save()?;
Ok(json!({"profile":profile}))
}
Request::Apply { text, revision } => {
ensure!(
self.state.revision == revision,
"stale revision; read status and retry"
);
ensure!(
fs::read_to_string(self.paths.config())? == text,
"config changed while applying; retry"
);
let next = Config::parse(&text)?;
ensure!(
self.state
.config
.pools
.iter()
.all(|p| next.pools.iter().any(|n| n.name == p.name)),
"use pool remove to drain and delete pools before removing configuration"
);
let mut pool_states = self.state.pools.clone();
for new in &next.pools {
if let Some(old) = self.state.config.pools.iter().find(|p| p.name == new.name) {
let ps = pool_states.get_mut(&new.name).unwrap();
ensure!(!ps.deleting, "cannot edit a deleting pool");
if old.api_scope() != new.api_scope() || old.auth != new.auth {
ensure!(
!ps.running
&& !self.state.runs.values().any(|r| r.pool_id == ps.id),
"stop and fully clean pool before changing scope/auth"
);
}
if old.image != new.image
|| old.labels != new.labels
|| old.runner_group != new.runner_group
|| old.docker_socket != new.docker_socket
|| old.cpus != new.cpus
|| old.memory != new.memory
{
ps.generation += 1;
}
} else {
pool_states.insert(new.name.clone(), PoolState::default());
}
}
for run in self.state.runs.values() {
ensure!(
next.auth.contains_key(&run.pool.auth),
"auth profile still required for cleanup"
);
}
self.state.pools = pool_states;
self.commit_config(next)?;
for name in self.state.pools.keys().cloned().collect::<Vec<_>>() {
self.mark_retiring(&name);
}
self.save()?;
Ok(json!({"revision":self.state.revision,"applied":true}))
}
Request::Logs { .. } | Request::Doctor => {
anyhow::bail!("external request must use the query handler")
}
}
}
fn mark_retiring(&mut self, name: &str) {
let ps = &self.state.pools[name];
let pool = self
.state
.config
.pools
.iter()
.find(|p| p.name == name)
.unwrap();
let keep = if ps.running && !ps.deleting {
pool.replicas as usize
} else {
0
};
let mut order: Vec<_> = self
.state
.runs
.values()
.filter(|r| r.pool_id == ps.id)
.map(|r| (r.generation, r.created_at, r.id.clone()))
.collect();
order.sort_by(|a, b| b.cmp(a));
for (index, (_, _, id)) in order.into_iter().enumerate() {
let run = self.state.runs.get_mut(&id).unwrap();
if index >= keep || run.generation != ps.generation {
run.retiring = true;
}
}
}
pub async fn tick(&mut self) -> Result<()> {
let names: Vec<String> = self.state.pools.keys().cloned().collect();
if names.is_empty() {
return Ok(());
}
let name = names[self.cursor % names.len()].clone();
self.cursor = self.cursor.wrapping_add(1);
let at = now();
let retry_at = self.state.pools[&name].retry_at;
let retrying = retry_at > at;
if retrying && self.state.pools[&name].retry_scope == RetryScope::Observation {
return Ok(());
}
// Unexpected exits delay provisioning, not observation or cleanup.
// Dependency failures still back off the entire reconciliation attempt.
self.reconcile_epoch = Some(self.epoch);
let result = self
.reconcile(&name, replenishment_allowed(retry_at, at))
.await;
let interrupted = self.reconcile_epoch != Some(self.epoch);
self.reconcile_epoch = None;
if interrupted {
self.save()?;
return Ok(());
}
match result {
Ok(report) => {
if report.unexpected_exits > 0 {
self.record_failure(&name, UNEXPECTED_EXIT_ERROR, RetryScope::Provisioning)?;
} else if !retrying {
if let Some(p) = self.state.pools.get_mut(&name) {
p.error = None;
p.retry_at = 0;
}
}
}
Err(error) => {
if error.is::<ReconcileInterrupted>() {
self.save()?;
return Ok(());
}
self.record_failure(&name, &error.to_string(), RetryScope::Observation)?;
if let Some(p) = self.state.pools.get(&name) {
// Last known busy/idle is not evidence after an observation failure.
let pool_id = p.id.clone();
for r in self.state.runs.values_mut().filter(|r| {
r.pool_id == pool_id
&& matches!(r.phase.as_str(), "idle" | "busy" | "starting")
}) {
r.phase = "unknown".into();
}
}
}
}
if let Some(pool) = self.state.pools.get_mut(&name) {
pool.observed_at = now();
}
self.save()?;
Ok(())
}
fn record_failure(&mut self, name: &str, error: &str, scope: RetryScope) -> Result<()> {
let at = now();
let p = self.state.pools.get_mut(name).context("pool disappeared")?;
if p.retry_at > at && p.retry_scope == scope {
return Ok(());
}
p.failures = p.failures.saturating_add(1);
let jitter = u64::from(Uuid::new_v4().as_bytes()[0] % 7);
p.retry_at = at.saturating_add(retry_delay(p.failures, jitter));
p.retry_scope = scope;
p.error = Some(error.to_owned());
Ok(())
}
async fn reconcile(&mut self, name: &str, allow_replenish: bool) -> Result<ReconcileReport> {
let pool = self
.state
.config
.pools
.iter()
.find(|p| p.name == name)
.context("missing pool config")?
.clone();
let ps = self.state.pools[name].clone();
let mut report = ReconcileReport::default();
let manager_id = self.state.manager_id.clone();
let paths = self.paths.clone();
let github = self.github.clone();
let scope = pool.api_scope();
let containers = self.external(Docker::list(&manager_id)).await?;
// An owned but unrecorded container counts against capacity and is never deleted blindly.
let unknown_containers = containers
.iter()
.filter(|c| {
c["Config"]["Labels"]["io.runnerctl.run"]
.as_str()
.is_none_or(|id| !self.state.runs.contains_key(id))
})
.count();
let no_runs = !self.state.runs.values().any(|r| r.pool_id == ps.id);
if no_runs && !ps.running {
if ps.deleting {
self.clean_file()?;
let mut next = self.state.config.clone();
next.pools.retain(|p| p.name != name);
self.state.pools.remove(name);
self.commit_config(next)?;
}
return Ok(report);
}
// Explicit force can stop local work even when the GitHub credential is revoked.
let forced: Vec<Run> = self
.state
.runs
.values()
.filter(|r| r.pool_id == ps.id && r.force)
.cloned()
.collect();
for mut run in forced {
if let Some(container) = containers
.iter()
.find(|c| Docker::owned(c, &self.state.manager_id, &run))
{
run.phase = "cleanup".into();
self.state.runs.insert(run.id.clone(), run.clone());
self.save()?;
self.archive(&mut run, container).await?;
self.external(Docker::remove(&run.id, true)).await?;
}
}
let containers = if self
.state
.runs
.values()
.any(|r| r.pool_id == ps.id && r.force)
{
self.external(Docker::list(&manager_id)).await?
} else {
containers
};
let auth = self.state.config.auth[&pool.auth].clone();
let ttl = self.state.config.manager.poll_interval_seconds;
let remotes = self
.external(github.list(&pool.auth, &auth, &scope, ttl))
.await?;
let ids: Vec<String> = self
.state
.runs
.values()
.filter(|r| r.pool_id == ps.id)
.map(|r| r.id.clone())
.collect();
let keep = if ps.running && !ps.deleting {
pool.replicas as usize
} else {
0
};
// Use the same stable selection as scale/apply, including timestamp ties.
self.mark_retiring(name);
self.save()?;
let mut cleanup_error: Option<anyhow::Error> = None;
for id in ids {
let mut run = self.state.runs[&id].clone();
let remote = remotes.iter().find(|r| r.name == id);
if let Some(remote) = remote {
run.remote_id = Some(remote.id);
}
let container = containers
.iter()
.find(|c| Docker::owned(c, &self.state.manager_id, &run));
let status = container.and_then(|c| c["State"]["Status"].as_str());
let mut interrupted = false;
if status == Some("running") || status == Some("restarting") || status == Some("paused")
{
run.phase = observed_phase(remote).into();
// Remove short-lived bootstrap secret after registration is observed.
if remote.is_some() {
let _ = fs::remove_file(self.paths.home.join("runs").join(&id).join("token"));
}
self.state.runs.insert(id, run);
continue;
} else if status == Some("created") && !run.retiring {
if !allow_replenish {
self.state.runs.insert(id, run);
continue;
}
// Token might have expired during manager downtime. Refresh before start.
let token = self
.external(github.token(&pool.auth, &auth, &scope))
.await?;
config::atomic_write(
&self.paths.home.join("runs").join(&id).join("token"),
token.as_bytes(),
)?;
self.external(Docker::start(&id)).await?;
run.phase = "starting".into();
self.state.runs.insert(id, run);
self.save()?;
continue;
} else if container.is_none() && run.phase == "creating" && !run.retiring {
if !allow_replenish {
self.state.runs.insert(id, run);
continue;
}
// Generation was saved before Docker create; deterministic name recovers retries.
self.external(Docker::image(&run.pool.image)).await?;
let dir = self.paths.home.join("runs").join(&id);
config::secure_dir(&dir)?;
let token = self
.external(github.token(&pool.auth, &auth, &scope))
.await?;
config::atomic_write(&dir.join("token"), token.as_bytes())?;
self.external(Docker::create(&manager_id, &run, &paths))
.await?;
self.external(Docker::start(&id)).await?;
run.phase = "starting".into();
self.state.runs.insert(id, run);
self.save()?;
continue;
} else if let Some(container) = container {
run.completed_job = self
.paths
.home
.join("runs")
.join(&id)
.join("completed")
.exists();
if unexpected_exit(&run) {
interrupted = self.stopped_by_reboot(container, &run.pool.image).await?;
}
run.phase = "cleanup".into();
self.state.runs.insert(id.clone(), run.clone());
self.save()?;
if let Err(error) = self.archive(&mut run, container).await {
cleanup_error.get_or_insert(error);
continue;
}
if let Err(error) = self.external(Docker::remove(&id, false)).await {
cleanup_error.get_or_insert(error);
continue;
}
}
// No live container remains. Fresh lookup avoids leaving a registration that
// appeared after the cached list (or a lost response from config.sh).
let fresh = match self
.external(github.list(&pool.auth, &auth, &scope, 0))
.await
{
Ok(fresh) => fresh,
Err(error) => {
cleanup_error.get_or_insert(error);
continue;
}
};
if let Some(remote) = fresh.iter().find(|r| r.name == id) {
// No live local worker exists, so stale busy state cannot represent running work here.
if let Err(error) = self
.external(github.remove(&pool.auth, &auth, &scope, remote.id))
.await
{
cleanup_error.get_or_insert(error);
continue;
}
}
let directory = paths.home.join("runs").join(&id);
if directory.exists() {
if let Err(error) = fs::remove_dir_all(&directory) {
if run.pool.docker_socket {
if let Err(error) = self
.external(Docker::repair_runtime_permissions(&run, &paths))
.await
{
cleanup_error.get_or_insert(error);
continue;
}
if let Err(error) = fs::remove_dir_all(&directory) {
cleanup_error.get_or_insert(error.into());
continue;
}
} else {
cleanup_error.get_or_insert(error.into());
continue;
}
}
}
if run.completed_job {
self.state.pools.get_mut(name).unwrap().failures = 0;
}
report.note_exit(&run, interrupted);
self.state.runs.remove(&id);
self.save()?;
}
if let Some(error) = cleanup_error {
return Err(error);
}
let active = self
.state
.runs
.values()
.filter(|r| r.pool_id == ps.id)
.count();
if ps.deleting && active == 0 {
self.clean_file()?;
let mut next = self.state.config.clone();
next.pools.retain(|p| p.name != name);
self.state.pools.remove(name);
self.commit_config(next)?;
return Ok(report);
}
let total = self.state.runs.len() + unknown_containers;
if allow_replenish
&& report.unexpected_exits == 0
&& active < keep
&& total < self.state.config.manager.max_runners as usize
{
// Only one create per pool turn; the single writer is below max_parallel_creates.
self.external(Docker::image(&pool.image)).await?;
let id = format!(
"rc-{}-{}",
&self.state.manager_id[..8],
Uuid::new_v4().simple()
);
let run = Run {
id: id.clone(),
pool_id: ps.id,
pool,
generation: ps.generation,
created_at: now(),
phase: "creating".into(),
remote_id: None,
retiring: false,
force: false,
completed_job: false,
log_saved: false,
};
self.state.runs.insert(id, run);
self.save()?;
}
self.expire_logs()?;
Ok(report)
}
// Docker Desktop, OrbStack and WSL2 VMs restart independently of this manager, so a
// cached boot time is refreshed before an exit is counted as a failure.
async fn stopped_by_reboot(&mut self, container: &Value, image: &str) -> Result<bool> {
if started_at(container).is_some() && !started_before_boot(container, self.boot_time) {
match self.external(Docker::boot_time(image)).await {
Ok(boot_time) => self.boot_time = Some(boot_time),
Err(error) if error.is::<ReconcileInterrupted>() => return Err(error),
// Without a boot time the exit counts as a failure.
Err(_) => {}
}
}
Ok(started_before_boot(container, self.boot_time))
}
async fn archive(&mut self, run: &mut Run, container: &Value) -> Result<()> {
if run.log_saved {
return Ok(());
}
let logs = self.external(Docker::logs(&run.id)).await?;
config::atomic_write(
&self.paths.home.join("logs").join(format!("{}.log", run.id)),
logs.as_bytes(),
)?;
// Never-started containers do not have diagnostic files.
if container["State"]["StartedAt"]
.as_str()
.is_some_and(|s| !s.starts_with("0001-"))
{
let paths = self.paths.clone();
self.external(Docker::archive(&run.id, &paths)).await?;
}
run.log_saved = true;
self.state.runs.insert(run.id.clone(), run.clone());
self.save()?;
Ok(())
}
fn expire_logs(&self) -> Result<()> {
let age = Duration::from_secs(
self.state
.config
.manager
.log_retention_days
.saturating_mul(86400),
);
for entry in fs::read_dir(self.paths.home.join("logs"))? {
let entry = entry?;
let meta = entry.metadata()?;
if meta.is_file() && meta.modified()?.elapsed().unwrap_or_default() > age {
let filename = entry.file_name().to_string_lossy().into_owned();
if !self.state.runs.keys().any(|id| filename.starts_with(id)) {
fs::remove_file(entry.path())?;
}
}
}
Ok(())
}
}
async fn runner_logs(state: &State, pool: &str, id: &str) -> Result<Value> {
let ps = state.pools.get(pool).context("unknown pool")?;
let run = state
.runs
.get(id)
.context("runner not active; archived logs are in the logs directory")?;
ensure!(run.pool_id == ps.id, "runner does not belong to pool");
let containers = Docker::list(&state.manager_id).await?;
ensure!(
containers
.iter()
.any(|c| Docker::owned(c, &state.manager_id, run)),
"owned container not found"
);
Ok(json!({"logs":Docker::logs(id).await?}))
}
async fn diagnose(state: &State, github: &Github, paths: &Paths) -> Result<Value> {
let mut checks = vec![json!({"check":"config", "ok":true, "warnings":state.config.warnings()})];
let docker = Docker::info().await;
checks.push(
json!({"check":"docker", "ok":docker.is_ok(), "runtime":docker.as_ref().ok(),
"error":docker.as_ref().err().map(|e|e.to_string())}),
);
let latest = if state.config.pools.is_empty() {
None
} else {
github.latest_runner_release().await.ok()
};
let mut checked_images = std::collections::BTreeMap::new();
for p in &state.config.pools {
let auth = &state.config.auth[&p.auth];
let gh = github.list(&p.auth, auth, &p.api_scope(), 0).await;
checks.push(json!({"check":format!("github:{}",p.name),"ok":gh.is_ok(),"error":gh.err().map(|e|e.to_string())}));
if !checked_images.contains_key(&p.image) {
let image = Docker::image_version(&p.image)
.await
.map_err(|e| e.to_string());
if let Ok(version) = &image {
checks.push(crate::image::version_check(
&p.image,
version.as_deref(),
latest.as_ref(),
));
}
checked_images.insert(p.image.clone(), image);
}
let image = &checked_images[&p.image];
checks.push(json!({"check":format!("image:{}",p.name),"image":p.image,"ok":image.is_ok(),"error":image.as_ref().err()}));
let ps = &state.pools[&p.name];
let runs: Vec<&Run> = state.runs.values().filter(|r| r.pool_id == ps.id).collect();
let health = pool_health(p, ps, &runs, now());
checks.push(json!({"check":format!("pool:{}",p.name),"ok":health.ok,
"intent":if ps.deleting {"deleting"} else if ps.running {"running"} else {"stopped"},
"replicas":p.replicas,"available":health.available,"unknown":health.unknown,"error":ps.error,
"retry_at":ps.retry_at,"observed_at":ps.observed_at,"reasons":health.reasons}));
}
// Runner bootstrap directories are bind-mounted from here; checked with a local image.
if let Some(image) = checked_images
.iter()
.find_map(|(image, version)| version.is_ok().then_some(image))
{
let shared = Docker::shares(&paths.home.join("runs"), image).await;
checks.push(json!({"check":"docker:bind","ok":shared.is_ok(),"error":shared.err().map(|e|e.to_string())}));
}
Ok(
json!({"healthy":checks.iter().all(|c| c["ok"]==true),"revision":state.revision,"checks":checks}),
)
}
#[derive(Default)]
struct ReconcileReport {
unexpected_exits: usize,
}
impl ReconcileReport {
fn note_exit(&mut self, run: &Run, interrupted: bool) {
// A host reboot stops every runner at once; that is not evidence of a broken image.
if unexpected_exit(run) && !interrupted {
self.unexpected_exits += 1;
}
}
}
struct PoolHealth {
ok: bool,
available: usize,
unknown: usize,
reasons: Vec<String>,
}
fn retry_delay(failures: u32, jitter: u64) -> u64 {
let base = 5u64
.saturating_mul(1u64 << failures.min(7))
.min(MAX_RETRY_SECONDS);
base.saturating_add(jitter.min(6)).min(MAX_RETRY_SECONDS)
}
fn replenishment_allowed(retry_at: u64, at: u64) -> bool {
retry_at <= at
}
fn pool_health(pool: &config::Pool, state: &PoolState, runs: &[&Run], at: u64) -> PoolHealth {
let available = runs
.iter()
.filter(|r| matches!(r.phase.as_str(), "idle" | "busy"))
.count();
let unknown = runs.iter().filter(|r| r.phase == "unknown").count();
let managed = state.running && !state.deleting && pool.replicas > 0;
let mut reasons = vec![];
if managed {
if let Some(error) = &state.error {
reasons.push(format!("error: {error}"));
}
if state.retry_at > at {
reasons.push(format!("retry scheduled at {}", state.retry_at));
}
if unknown > 0 {
reasons.push(format!("{unknown} unknown runner(s)"));
}
if available < pool.replicas as usize {
reasons.push(format!("available capacity {available}/{}", pool.replicas));
}
}
PoolHealth {
ok: reasons.is_empty(),
available,
unknown,
reasons,
}
}
fn observed_phase(remote: Option<&RemoteRunner>) -> &'static str {
match remote {
Some(r) if r.busy => "busy",
Some(r) if r.status == "online" => "idle",
_ => "unknown",
}
}
fn unexpected_exit(run: &Run) -> bool {
!run.completed_job && !run.retiring && !run.force
}
fn started_at(container: &Value) -> Option<u64> {
humantime::parse_rfc3339(container["State"]["StartedAt"].as_str()?)
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|t| t.as_secs())
}
// Containers use --restart=no, so one started before this boot was stopped by the reboot.
fn started_before_boot(container: &Value, boot_time: Option<u64>) -> bool {
boot_time.is_some_and(|boot_time| started_at(container).is_some_and(|t| t < boot_time))
}
pub async fn serve(paths: Paths) -> Result<()> {
paths.prepare()?;
let _lock = lock(&paths)?;
let mut engine = Engine::open(paths.clone())?;
let mut listener = ipc::Listener::bind(&paths)?;
eprintln!(
"runnerctl manager listening at {}",
paths.socket().display()
);
let (commands, receiver) = mpsc::channel::<Envelope>(128);
let (snapshots, snapshot_receiver) = watch::channel(engine.state.clone());
engine.commands = Some(receiver);
engine.snapshots = Some(snapshots);
let github = engine.github.clone();
let connection_paths = paths.clone();
let connections = tokio::spawn(async move {
let permits = std::sync::Arc::new(tokio::sync::Semaphore::new(64));
loop {
let permit = permits.clone().acquire_owned().await?;
let stream = listener.accept().await?;
let commands = commands.clone();
let snapshots = snapshot_receiver.clone();
let github = github.clone();
let paths = connection_paths.clone();
tokio::spawn(async move {
let _permit = permit;
serve_connection(stream, commands, snapshots, github, paths).await;
});
}
#[allow(unreachable_code)]
Ok::<(), anyhow::Error>(())
});
let mut interval = tokio::time::interval(Duration::from_secs(1));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let term = terminated()?;
tokio::pin!(term);
let result = loop {
tokio::select! {
_ = &mut term => break Ok(()),
_ = tokio::signal::ctrl_c() => break Ok(()),
envelope = engine.commands.as_mut().unwrap().recv() => {
let Some(envelope) = envelope else { break Err(anyhow::anyhow!("request listener stopped")); };
let _ = envelope.reply.send(engine.handle_local(envelope.request));
}
_ = interval.tick() => {
if let Err(error) = engine.tick().await { break Err(error); }
}
}
};
connections.abort();
fs::remove_file(paths.socket())?;
result
}
#[cfg(unix)]
fn terminated() -> Result<impl Future<Output = ()>> {
use tokio::signal::unix::{SignalKind, signal};
let mut term = signal(SignalKind::terminate())?;
Ok(async move {
term.recv().await;
})
}
// Windows has no SIGTERM; a closed console, Ctrl+Break and shutdown stop the manager.
#[cfg(windows)]
fn terminated() -> Result<impl Future<Output = ()>> {
use tokio::signal::windows::{ctrl_break, ctrl_close, ctrl_shutdown};
let (mut brk, mut close, mut shutdown) = (ctrl_break()?, ctrl_close()?, ctrl_shutdown()?);
Ok(async move {
tokio::select! {
_ = brk.recv() => {}
_ = close.recv() => {}
_ = shutdown.recv() => {}
}
})
}
async fn serve_connection(
mut stream: ipc::Server,
commands: mpsc::Sender<Envelope>,
snapshots: watch::Receiver<State>,
github: Github,
paths: Paths,
) {
let result: Result<Value> = async {
let request = tokio::time::timeout(Duration::from_secs(5), read_request(&mut stream))
.await
.context("request timeout")??;
let state = snapshots.borrow().clone();
match request {
Request::Doctor => diagnose(&state, &github, &paths).await,
Request::Logs { pool, id } => runner_logs(&state, &pool, &id).await,
request => {
if let Request::Upgrade { image, .. } = &request {
config::validate_image(image)?;
Docker::image(image).await?;
}
let (reply, receiver) = oneshot::channel();
commands
.send(Envelope { request, reply })
.await
.context("manager stopped")?;
receiver.await.context("manager stopped before replying")?
}
}
}
.await;
let response = match result {
Ok(data) => Response { ok: true, data },
Err(error) => Response {
ok: false,
data: json!({"error":error.to_string()}),
},
};
if let Ok(mut bytes) = serde_json::to_vec(&response) {
bytes.push(b'\n');
let _ = tokio::time::timeout(Duration::from_secs(5), stream.write_all(&bytes)).await;
}
}
async fn read_request(stream: &mut (impl AsyncRead + Unpin)) -> Result<Request> {
use tokio::io::AsyncReadExt;
let mut line = String::new();
BufReader::new(stream.take(1024 * 1024 + 1))
.read_line(&mut line)
.await?;
ensure!(
line.len() <= 1024 * 1024 && line.ends_with('\n'),
"request too large or incomplete"
);
Ok(serde_json::from_str(&line)?)
}
pub async fn send(paths: &Paths, request: &Request) -> Result<Value> {
let mut stream = ipc::connect(paths)
.await
.context("manager is not running; run runnerctl manager or service install")?;
let mut bytes = serde_json::to_vec(request)?;
bytes.push(b'\n');
stream.write_all(&bytes).await?;
let mut line = String::new();
tokio::time::timeout(
Duration::from_secs(300),
BufReader::new(stream).read_line(&mut line),
)
.await
.context("manager request timed out; check status before retrying changes")??;
let response: Response = serde_json::from_str(&line)?;
ensure!(
response.ok,
"{}",
response.data["error"]
.as_str()
.unwrap_or("manager request failed")
);
Ok(response.data)
}
#[cfg(test)]
mod recovery_tests;
#[cfg(test)]
mod tests {
use super::*;
fn fixture() -> (tempfile::TempDir, Engine) {
let dir = tempfile::tempdir().unwrap();
let paths = Paths::new(dir.path().to_owned()).unwrap();
paths.prepare().unwrap();
let mut c = Config::default();
c.auth.insert(
"team".into(),
config::Auth {
credential: "env:TEST_TOKEN".into(),
app: None,
},
);
for (name, replicas) in [("validate", 4), ("build", 2)] {
c.pools.push(config::Pool {
name: name.into(),
organization: "test-org".into(),
repository: None,
auth: "team".into(),
labels: vec![name.into()],
runner_group: "Default".into(),
replicas,
image: config::default_image(),
docker_socket: false,
cpus: None,
memory: None,
});
}
config::atomic_write(&paths.config(), toml::to_string(&c).unwrap().as_bytes()).unwrap();
let e = Engine::open(paths).unwrap();
(dir, e)
}
fn test_run(e: &Engine, pool: usize, phase: &str) -> Run {
Run {
id: format!("run-{pool}-{phase}"),
pool_id: e.state.pools[&e.state.config.pools[pool].name].id.clone(),
pool: e.state.config.pools[pool].clone(),
generation: 1,
created_at: now(),
phase: phase.into(),
remote_id: None,
retiring: false,
force: false,
completed_job: false,
log_saved: false,
}
}
#[test]
fn unexpected_exits_exclude_completed_retiring_and_forced_runs() {
let (_dir, e) = fixture();
let exited = test_run(&e, 0, "exited");
let exited_again = test_run(&e, 0, "exited");
let completed = Run {
completed_job: true,
..exited.clone()
};
let retiring = Run {
retiring: true,
..exited.clone()
};
let forced = Run {
force: true,
..exited.clone()
};
let mut report = ReconcileReport::default();
for run in [&exited, &exited_again, &completed, &retiring, &forced] {
report.note_exit(run, false);
}
report.note_exit(&exited, true);
assert_eq!(report.unexpected_exits, 2);
}
#[test]
fn only_containers_started_before_boot_are_interrupted() {
let started = |at: &str| json!({"State":{"Status":"exited","StartedAt":at}});
// 2026-09-30T03:20:00Z
let boot = Some(1_790_738_400);
assert!(started_before_boot(
&started("2026-09-29T07:23:50.123456789Z"),
boot
));
assert!(!started_before_boot(
&started("2026-09-30T03:20:36.5Z"),
boot
));
assert!(!started_before_boot(&started("0001-01-01T00:00:00Z"), boot));
assert!(!started_before_boot(&started("not a time"), boot));
assert!(!started_before_boot(&json!({"State":{}}), boot));
assert!(!started_before_boot(&started("2026-09-29T07:23:50Z"), None));
}
#[test]
fn replenishment_backoff_is_bounded() {
assert_eq!(retry_delay(0, 0), 5);
assert!(retry_delay(3, 6) > retry_delay(2, 0));
assert_eq!(retry_delay(u32::MAX, 6), MAX_RETRY_SECONDS);
let at = 100;
assert!(!replenishment_allowed(at + 1, at));
assert!(replenishment_allowed(at, at));
}
#[tokio::test]
async fn dependency_backoff_survives_restart_and_skips_observation() {
let (_dir, mut e) = fixture();
e.record_failure(
"validate",
"dependency unavailable",
RetryScope::Observation,
)
.unwrap();
e.save().unwrap();
let deadline = e.state.pools["validate"].retry_at;
let paths = e.paths.clone();
drop(e);
let mut e = Engine::open(paths).unwrap();
e.cursor = 1; // BTreeMap order: build, validate.
e.tick().await.unwrap();
let p = &e.state.pools["validate"];
assert_eq!(p.retry_at, deadline);
assert_eq!(p.failures, 1);
assert_eq!(p.error.as_deref(), Some("dependency unavailable"));
assert!(p.retry_scope == RetryScope::Observation);
}
#[test]
fn provisioning_backoff_is_persisted_and_not_extended_by_more_exits() {
let (_dir, mut e) = fixture();
e.record_failure("validate", UNEXPECTED_EXIT_ERROR, RetryScope::Provisioning)
.unwrap();
let deadline = e.state.pools["validate"].retry_at;
e.record_failure("validate", UNEXPECTED_EXIT_ERROR, RetryScope::Provisioning)
.unwrap();
assert_eq!(e.state.pools["validate"].retry_at, deadline);
assert_eq!(e.state.pools["validate"].failures, 1);
e.save().unwrap();
let restored = Engine::open(e.paths.clone()).unwrap();
assert!(restored.state.pools["validate"].retry_scope == RetryScope::Provisioning);
e.record_failure("validate", "Docker unavailable", RetryScope::Observation)
.unwrap();
assert!(e.state.pools["validate"].retry_scope == RetryScope::Observation);
assert_eq!(e.state.pools["validate"].failures, 2);
}
#[test]
fn busy_capacity_is_healthy_but_unknown_or_missing_capacity_is_not() {
let (_dir, e) = fixture();
let mut pool = e.state.config.pools[0].clone();
pool.replicas = 1;
let mut state = e.state.pools[&pool.name].clone();
state.running = true;
let busy = test_run(&e, 0, "busy");
let health = pool_health(&pool, &state, &[&busy], now());
assert!(health.ok);
assert_eq!(health.available, 1);
state.error = Some("runner failure".into());
assert!(!pool_health(&pool, &state, &[&busy], now()).ok);
state.error = None;
state.retry_at = now() + 60;
assert!(!pool_health(&pool, &state, &[&busy], now()).ok);
state.retry_at = 0;
let unknown = test_run(&e, 0, "unknown");
let health = pool_health(&pool, &state, &[&busy, &unknown], now());
assert!(!health.ok);
assert_eq!(health.unknown, 1);
let mut pool = pool.clone();
pool.replicas = 2;
let health = pool_health(&pool, &state, &[&busy], now());
assert!(!health.ok);
assert!(
health
.reasons
.iter()
.any(|reason| reason.contains("available capacity"))
);
}
#[test]
fn stopped_and_zero_replica_pools_are_not_degraded() {
let (_dir, e) = fixture();
let mut pool = e.state.config.pools[0].clone();
let mut state = e.state.pools[&pool.name].clone();
state.error = Some("old failure".into());
state.retry_at = now() + 60;
let unknown = test_run(&e, 0, "unknown");
assert!(pool_health(&pool, &state, &[&unknown], now()).ok);
state.running = true;
pool.replicas = 0;
assert!(pool_health(&pool, &state, &[&unknown], now()).ok);
}
#[test]
fn remote_observation_preserves_busy_state() {
let busy = RemoteRunner {
id: 1,
name: "runner".into(),
status: "offline".into(),
busy: true,
};
let idle = RemoteRunner {
status: "online".into(),
busy: false,
..busy.clone()
};
let unknown = RemoteRunner {
status: "offline".into(),
busy: false,
..busy.clone()
};
assert_eq!(observed_phase(Some(&busy)), "busy");
assert_eq!(observed_phase(Some(&idle)), "idle");
assert_eq!(observed_phase(Some(&unknown)), "unknown");
assert_eq!(observed_phase(None), "unknown");
}
#[tokio::test]
async fn pool_stop_scale_and_restart_are_independent() {
let (_dir, mut e) = fixture();
e.handle(Request::Start {
pool: None,
all: true,
})
.await
.unwrap();
e.handle(Request::Stop {
pool: Some("build".into()),
all: false,
force: false,
})
.await
.unwrap();
e.handle(Request::Scale {
pool: "build".into(),
replicas: 3,
})
.await
.unwrap();
let paths = e.paths.clone();
drop(e);
let e = Engine::open(paths).unwrap();
assert!(e.state.pools["validate"].running);
assert!(!e.state.pools["build"].running);
assert_eq!(e.state.config.pools[1].replicas, 3);
}
#[tokio::test]
async fn rejected_changes_preserve_state_and_file() {
let (_dir, mut e) = fixture();
let previous = fs::read_to_string(e.paths.config()).unwrap();
assert!(
e.handle(Request::Scale {
pool: "build".into(),
replicas: 100
})
.await
.is_err()
);
assert_eq!(e.state.config.pools[1].replicas, 2);
assert_eq!(fs::read_to_string(e.paths.config()).unwrap(), previous);
assert!(
e.handle(Request::Start {
pool: None,
all: false
})
.await
.is_err()
);
assert!(e.state.pools.values().all(|p| !p.running));
}
#[tokio::test]
async fn manual_edits_and_revision_conflicts_are_preserved() {
let (_dir, mut e) = fixture();
let mut c = e.state.config.clone();
c.pools[0].replicas = 5;
let text = toml::to_string(&c).unwrap();
config::atomic_write(&e.paths.config(), text.as_bytes()).unwrap();
assert!(
e.handle(Request::Scale {
pool: "build".into(),
replicas: 1
})
.await
.is_err()
);
assert!(
e.handle(Request::Apply {
text: text.clone(),
revision: 0
})
.await
.is_err()
);
e.handle(Request::Apply { text, revision: 1 })
.await
.unwrap();
assert_eq!(e.state.config.pools[0].replicas, 5);
}
#[tokio::test]
async fn organization_change_requires_fully_stopped_pool() {
let (_dir, mut e) = fixture();
e.handle(Request::Start {
pool: Some("build".into()),
all: false,
})
.await
.unwrap();
let mut c = e.state.config.clone();
c.pools[1].organization = "different".into();
let text = toml::to_string(&c).unwrap();
config::atomic_write(&e.paths.config(), text.as_bytes()).unwrap();
assert!(
e.handle(Request::Apply { text, revision: 1 })
.await
.is_err()
);
assert_eq!(e.state.config.pools[1].organization, "test-org");
}
#[test]
fn manager_lock_is_exclusive() {
let (_dir, e) = fixture();
let _lock = lock(&e.paths).unwrap();
assert!(lock(&e.paths).is_err());
}
#[test]
fn ownership_requires_manager_pool_and_run_ids() {
let (_dir, e) = fixture();
let run = Run {
id: "test".into(),
pool_id: e.state.pools["build"].id.clone(),
pool: e.state.config.pools[1].clone(),
generation: 1,
created_at: now(),
phase: "busy".into(),
remote_id: None,
retiring: false,
force: false,
completed_job: false,
log_saved: false,
};
let mut c = json!({"Config":{"Labels":{"io.runnerctl.manager":e.state.manager_id,"io.runnerctl.pool":run.pool_id,"io.runnerctl.run":run.id}}});
assert!(Docker::owned(&c, &e.state.manager_id, &run));
c["Config"]["Labels"]["io.runnerctl.pool"] = json!("another");
assert!(!Docker::owned(&c, &e.state.manager_id, &run));
}
#[tokio::test]
#[ignore = "requires Docker and the local/runnerctl-test:1 image; see tests/README.md"]
async fn docker_replacement_and_restart_recovery() {
let (_dir, mut e) = fixture();
let (base, server) = crate::github::tests::server().await;
let path = e.paths.home.join("credentials/test");
config::atomic_write(&path, b"test-pat").unwrap();
e.state.config.auth.get_mut("team").unwrap().credential =
format!("file:{}", path.display());
for pool in &mut e.state.config.pools {
pool.image = "local/runnerctl-test:1".into();
pool.replicas = 1;
}
e.github = Github::testing(base.clone());
e.handle(Request::Start {
pool: None,
all: true,
})
.await
.unwrap();
struct Cleanup(String);
impl Drop for Cleanup {
fn drop(&mut self) {
if let Ok(out) = std::process::Command::new("docker")
.args([
"ps",
"-aq",
"--filter",
&format!("label=io.runnerctl.manager={}", self.0),
])
.output()
{
for id in String::from_utf8_lossy(&out.stdout).split_whitespace() {
let _ = std::process::Command::new("docker")
.args(["rm", "-f", id])
.output();
}
}
}
}
let _cleanup = Cleanup(e.state.manager_id.clone());
for _ in 0..4 {
e.tick().await.unwrap();
}
assert_eq!(
e.state.runs.len(),
2,
"{:?}",
e.state.pools.values().map(|p| &p.error).collect::<Vec<_>>()
);
let initial: Vec<String> = e.state.runs.keys().cloned().collect();
assert!(e.state.runs.values().all(|r| r.phase == "starting"));
e.handle(Request::Stop {
pool: Some("build".into()),
all: false,
force: false,
})
.await
.unwrap();
let paths = e.paths.clone();
drop(e);
let mut e = Engine::open(paths).unwrap();
e.github = Github::testing(base);
tokio::time::sleep(Duration::from_secs(3)).await;
for _ in 0..4 {
e.tick().await.unwrap();
}
assert!(e.state.runs.values().all(|r| r.pool.name == "validate"));
assert!(e.state.runs.keys().all(|id| !initial.contains(id)));
assert!(e.state.pools["validate"].error.is_none());
assert!(!e.state.pools["build"].running);
server.abort();
}
}