use std::fmt;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::paths::AppPaths;
use crate::process::{Adoption, ProcessIdentity};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum LockKind {
SingleInstance,
Allocation,
}
impl LockKind {
#[must_use]
pub const fn file_name(self) -> &'static str {
match self {
Self::SingleInstance => "agent.lock",
Self::Allocation => "allocation.lock",
}
}
#[must_use]
pub const fn description(self) -> &'static str {
match self {
Self::SingleInstance => "the single-instance agent lock",
Self::Allocation => "the runtime allocation lock",
}
}
#[must_use]
pub const fn advice(self) -> &'static str {
match self {
Self::SingleInstance => {
"Only one agent may reconcile policies on a host. Stop the other agent, or wait \
for it to exit; the operating system releases this lock when that process ends, \
including after a crash, so there is never anything to clean up by hand."
}
Self::Allocation => {
"This lock is held only for as long as it takes to create one runtime. Retry \
shortly."
}
}
}
}
impl fmt::Display for LockKind {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.description())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LockHolder {
pub identity: ProcessIdentity,
pub executable: Option<PathBuf>,
pub acquired_at: DateTime<Utc>,
pub lock: LockKind,
}
impl fmt::Display for LockHolder {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "process {}", self.identity.pid())?;
if let Some(executable) = &self.executable {
write!(f, " ({})", executable.display())?;
}
write!(f, ", holding since {}", self.acquired_at.to_rfc3339())
}
}
#[derive(Debug, thiserror::Error)]
pub enum LockError {
#[error(
"{kind} ({}) is already held on this host by {}. {} [refused with {}]",
path.display(),
describe(holder.as_deref()),
kind.advice(),
describe_refusal(*refused_with)
)]
Held {
kind: LockKind,
path: PathBuf,
holder: Option<Box<LockHolder>>,
refused_with: Option<i32>,
},
#[error("cannot use the lock file {}: {source}", path.display())]
Io {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("cannot record this process as the holder of {kind}: {source}")]
Identity {
kind: LockKind,
#[source]
source: crate::process::ProcessError,
},
}
fn describe(holder: Option<&LockHolder>) -> String {
match holder {
Some(holder) => holder.to_string(),
None => "a process that has not finished identifying itself".to_string(),
}
}
#[derive(Debug)]
pub struct HostLock {
file: File,
path: PathBuf,
kind: LockKind,
}
impl HostLock {
pub fn try_acquire(paths: &AppPaths, kind: LockKind) -> Result<Self, LockError> {
Self::try_acquire_at(&paths.state_dir().join(kind.file_name()), kind)
}
pub fn acquire(paths: &AppPaths, kind: LockKind, wait: Duration) -> Result<Self, LockError> {
Self::acquire_at(&paths.state_dir().join(kind.file_name()), kind, wait)
}
pub fn try_acquire_at(path: &Path, kind: LockKind) -> Result<Self, LockError> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|source| LockError::Io {
path: path.to_path_buf(),
source,
})?;
}
let held = |path: &Path, refused_with: Option<i32>| LockError::Held {
kind,
path: path.to_path_buf(),
holder: read_holder(path).ok().flatten().map(Box::new),
refused_with,
};
let file = match open_for_locking(path) {
Ok(file) => file,
Err(source) if sys::is_contention(&source) => {
return Err(held(path, source.raw_os_error()));
}
Err(source) => return Err(io_error(path, source)),
};
match sys::try_lock(&file).map_err(|source| io_error(path, source))? {
Acquired::Yes => {}
Acquired::No { refused_with } => return Err(held(path, refused_with)),
}
let lock = Self {
file,
path: path.to_path_buf(),
kind,
};
lock.record_holder()?;
Ok(lock)
}
pub fn acquire_at(path: &Path, kind: LockKind, wait: Duration) -> Result<Self, LockError> {
let deadline = Instant::now() + wait;
loop {
match Self::try_acquire_at(path, kind) {
Ok(lock) => return Ok(lock),
Err(error @ LockError::Held { .. }) => {
if Instant::now() >= deadline {
return Err(error);
}
std::thread::sleep(RETRY_INTERVAL);
}
Err(other) => return Err(other),
}
}
}
pub fn holder_of(path: &Path) -> Result<Option<LockHolder>, LockError> {
read_holder(path)
}
#[must_use]
pub fn path(&self) -> &Path {
&self.path
}
#[must_use]
pub const fn kind(&self) -> LockKind {
self.kind
}
pub fn recorded_holder_is_live(path: &Path) -> Result<bool, LockError> {
let Some(holder) = read_holder(path)? else {
return Ok(false);
};
Ok(matches!(holder.identity.recheck(), Ok(Adoption::Live)))
}
fn record_holder(&self) -> Result<(), LockError> {
let holder = LockHolder {
identity: ProcessIdentity::of_current_process().map_err(|source| {
LockError::Identity {
kind: self.kind,
source,
}
})?,
executable: std::env::current_exe().ok(),
acquired_at: Utc::now(),
lock: self.kind,
};
let encoded = serde_json::to_vec_pretty(&holder).map_err(|source| {
io_error(
&self.path,
std::io::Error::new(std::io::ErrorKind::InvalidData, source),
)
})?;
let mut file = &self.file;
file.seek(SeekFrom::Start(0))
.map_err(|source| io_error(&self.path, source))?;
file.set_len(0)
.map_err(|source| io_error(&self.path, source))?;
file.write_all(&encoded)
.map_err(|source| io_error(&self.path, source))?;
file.flush()
.map_err(|source| io_error(&self.path, source))?;
file.sync_all()
.map_err(|source| io_error(&self.path, source))
}
}
fn io_error(path: &Path, source: std::io::Error) -> LockError {
LockError::Io {
path: path.to_path_buf(),
source,
}
}
const RETRY_INTERVAL: Duration = Duration::from_millis(25);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(windows, allow(dead_code))]
pub(crate) enum Acquired {
Yes,
No { refused_with: Option<i32> },
}
fn describe_refusal(code: Option<i32>) -> String {
match code {
Some(code) => format!("os error {code}"),
None => "the open itself, which is how this platform excludes".to_string(),
}
}
fn open_for_locking(path: &Path) -> std::io::Result<File> {
let mut options = OpenOptions::new();
options.read(true).write(true).create(true).truncate(false);
sys::prepare_for_locking(&mut options);
options.open(path)
}
fn read_holder(path: &Path) -> Result<Option<LockHolder>, LockError> {
let mut options = OpenOptions::new();
options.read(true);
sys::prepare_for_reading(&mut options);
let mut file = match options.open(path) {
Ok(file) => file,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => return Err(io_error(path, source)),
};
let mut contents = String::new();
file.read_to_string(&mut contents)
.map_err(|source| io_error(path, source))?;
Ok(serde_json::from_str(&contents).ok())
}
#[cfg(windows)]
mod sys {
use std::fs::{File, OpenOptions};
use std::io;
use std::os::windows::fs::OpenOptionsExt;
use windows::Win32::Storage::FileSystem::{
FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE,
};
pub(super) fn prepare_for_locking(options: &mut OpenOptions) {
options.share_mode(FILE_SHARE_READ.0);
}
pub(super) fn prepare_for_reading(options: &mut OpenOptions) {
options.share_mode(FILE_SHARE_READ.0 | FILE_SHARE_WRITE.0 | FILE_SHARE_DELETE.0);
}
pub(super) fn try_lock(_file: &File) -> io::Result<super::Acquired> {
Ok(super::Acquired::Yes)
}
pub(super) const SHARING_VIOLATION: i32 =
windows::Win32::Foundation::ERROR_SHARING_VIOLATION.0 as i32;
pub(super) fn is_contention(error: &io::Error) -> bool {
error.raw_os_error() == Some(SHARING_VIOLATION)
}
}
#[cfg(unix)]
mod sys {
use std::fs::{File, OpenOptions};
use std::io;
use std::os::unix::io::AsRawFd;
pub(super) fn prepare_for_locking(_options: &mut OpenOptions) {}
pub(super) fn prepare_for_reading(_options: &mut OpenOptions) {}
pub(super) fn try_lock(file: &File) -> io::Result<super::Acquired> {
let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
if result == 0 {
return Ok(super::Acquired::Yes);
}
let error = io::Error::last_os_error();
if is_contention(&error) {
return Ok(super::Acquired::No {
refused_with: error.raw_os_error(),
});
}
Err(error)
}
pub(super) fn is_contention(error: &io::Error) -> bool {
matches!(
error.raw_os_error(),
Some(code) if code == libc::EWOULDBLOCK || code == libc::EAGAIN
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::process::{OutputMode, SpawnSpec};
const HELPER_PATH: &str = "RUNNER_MANAGER_LOCK_HELPER_PATH";
const HELPER_READY: &str = "@@LOCK-HELD@@";
fn lock_path(directory: &tempfile::TempDir, kind: LockKind) -> PathBuf {
directory.path().join(kind.file_name())
}
fn check_mutual_exclusion<G>(acquire: impl Fn() -> Result<G, LockError>) -> Result<(), String> {
let first = acquire().map_err(|error| format!("the first acquisition failed: {error}"))?;
let outcome = match acquire() {
Ok(_) => Err("both acquisitions succeeded; nothing is being excluded".to_string()),
Err(LockError::Held { .. }) => Ok(()),
Err(other) => Err(format!(
"the second acquisition failed, but not because the lock was held: {other}"
)),
};
drop(first);
outcome
}
#[test]
fn two_contenders_produce_exactly_one_holder() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
check_mutual_exclusion(|| HostLock::try_acquire_at(&path, LockKind::SingleInstance))
.expect("the single-instance lock must admit exactly one holder");
}
#[test]
fn the_contention_check_catches_a_lock_that_never_excludes() {
let complaint = check_mutual_exclusion(|| Ok::<(), LockError>(()))
.expect_err("a lock that excludes nothing must be caught");
assert!(
complaint.contains("nothing is being excluded"),
"the complaint must name the failure mode, got: {complaint}"
);
}
#[test]
fn releasing_the_lock_lets_the_next_contender_take_it() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let first = HostLock::try_acquire_at(&path, LockKind::SingleInstance).expect("acquired");
assert!(HostLock::try_acquire_at(&path, LockKind::SingleInstance).is_err());
drop(first);
let second = HostLock::try_acquire_at(&path, LockKind::SingleInstance)
.expect("the lock must be free once the holder drops it");
assert_eq!(second.path(), path);
assert_eq!(second.kind(), LockKind::SingleInstance);
}
#[test]
fn the_two_locks_do_not_contend_with_each_other() {
let directory = tempfile::tempdir().expect("a temporary directory");
let instance = HostLock::try_acquire_at(
&lock_path(&directory, LockKind::SingleInstance),
LockKind::SingleInstance,
)
.expect("the instance lock is free");
let allocation = HostLock::try_acquire_at(
&lock_path(&directory, LockKind::Allocation),
LockKind::Allocation,
)
.expect("the allocation lock is a different lock and must be free");
assert_ne!(instance.path(), allocation.path());
assert_ne!(
LockKind::SingleInstance.file_name(),
LockKind::Allocation.file_name()
);
}
#[test]
fn the_loser_gets_a_message_naming_the_holder_and_saying_what_to_do() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let _held = HostLock::try_acquire_at(&path, LockKind::SingleInstance).expect("acquired");
let error = HostLock::try_acquire_at(&path, LockKind::SingleInstance)
.expect_err("the second acquisition must fail");
let LockError::Held { holder, kind, .. } = &error else {
panic!("expected a contention error, got {error}");
};
assert_eq!(*kind, LockKind::SingleInstance);
let holder = holder.as_ref().expect("the holder recorded itself");
assert_eq!(holder.identity.pid(), std::process::id());
assert_eq!(holder.lock, LockKind::SingleInstance);
assert_eq!(
holder.executable.as_deref(),
std::env::current_exe().ok().as_deref()
);
let message = error.to_string();
assert!(
message.contains(&std::process::id().to_string()),
"the message must name the holding process: {message}"
);
assert!(
message.contains("Stop the other agent"),
"the message must say what to do, not only what happened: {message}"
);
}
#[test]
fn the_recorded_holder_survives_a_read_by_a_second_process_shape() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let _held = HostLock::try_acquire_at(&path, LockKind::SingleInstance).expect("acquired");
let holder = HostLock::holder_of(&path)
.expect("the record must be readable while the lock is held")
.expect("a record must be there");
assert_eq!(holder.identity.pid(), std::process::id());
assert!(HostLock::recorded_holder_is_live(&path).expect("readable"));
}
#[test]
fn a_stale_record_is_reported_as_not_live() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let mut child = SpawnSpec::new(if cfg!(windows) { "cmd" } else { "sh" })
.args(if cfg!(windows) {
vec!["/C", "exit", "0"]
} else {
vec!["-c", "sleep 0.3"]
})
.spawn()
.expect("the child starts");
let identity = child.identity().clone();
child.wait().expect("the child exits");
let stale = LockHolder {
identity,
executable: None,
acquired_at: Utc::now(),
lock: LockKind::SingleInstance,
};
std::fs::write(&path, serde_json::to_vec(&stale).expect("serialisable")).expect("writable");
assert!(
!HostLock::recorded_holder_is_live(&path).expect("readable"),
"a record naming a dead process must not be reported as live"
);
HostLock::try_acquire_at(&path, LockKind::SingleInstance)
.expect("a stale record must not keep the lock held");
}
#[test]
fn an_unidentified_holder_still_reads_as_a_holder() {
let no_record = describe(None);
assert!(
no_record.contains("not finished identifying itself"),
"an unidentified holder must still read as a holder: {no_record}"
);
let error = LockError::Held {
kind: LockKind::SingleInstance,
path: PathBuf::from("/var/lib/runner-manager/state/agent.lock"),
holder: None,
refused_with: Some(35),
};
let message = error.to_string();
assert!(message.contains("already held"), "{message}");
assert!(message.contains("agent.lock"), "{message}");
assert!(message.contains("Stop the other agent"), "{message}");
assert!(message.contains("os error 35"), "{message}");
let at_open = LockError::Held {
kind: LockKind::SingleInstance,
path: PathBuf::from("/var/lib/runner-manager/state/agent.lock"),
holder: None,
refused_with: None,
};
assert!(
at_open.to_string().contains("the open itself"),
"a platform that excludes at the open says so rather than showing a bare `None`: {at_open}"
);
}
#[test]
fn acquire_waits_and_then_reports_the_holder() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::Allocation);
let _held = HostLock::try_acquire_at(&path, LockKind::Allocation).expect("acquired");
let wait = Duration::from_millis(200);
let started = Instant::now();
let error = HostLock::acquire_at(&path, LockKind::Allocation, wait)
.expect_err("the lock is held for the whole window");
let elapsed = started.elapsed();
assert!(matches!(error, LockError::Held { .. }), "{error}");
assert!(
elapsed >= wait,
"acquire must actually wait; it returned after {elapsed:?} of a {wait:?} window"
);
}
#[test]
fn acquire_returns_immediately_when_the_lock_is_free() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::Allocation);
let started = Instant::now();
let lock = HostLock::acquire_at(&path, LockKind::Allocation, Duration::from_secs(30))
.expect("the lock is free");
assert!(
started.elapsed() < Duration::from_secs(5),
"a free lock must not be waited for"
);
drop(lock);
}
#[test]
fn try_acquire_uses_the_state_directory() {
let root = tempfile::tempdir().expect("a temporary directory");
let paths = AppPaths::rooted_at(root.path());
let lock = HostLock::try_acquire(&paths, LockKind::SingleInstance).expect("acquired");
assert_eq!(lock.path(), paths.state_dir().join("agent.lock"));
assert!(
lock.path().exists(),
"the lock file must have been created under state/, as 05-infrastructure.md says"
);
}
#[test]
fn lock_holder_helper() {
let Some(path) = std::env::var_os(HELPER_PATH) else {
return;
};
let _lock = HostLock::try_acquire_at(Path::new(&path), LockKind::SingleInstance)
.expect("the helper must be able to take the lock");
println!("{HELPER_READY} {}", std::process::id());
let _ = std::io::stdout().flush();
std::thread::sleep(Duration::from_secs(120));
}
fn start_helper(path: &Path) -> (crate::process::ChildProcess, u32) {
let executable = std::env::current_exe().expect("the test binary's own path");
let mut child = SpawnSpec::new(executable)
.args([
"--exact",
"lock::tests::lock_holder_helper",
"--nocapture",
"--test-threads=1",
])
.env(HELPER_PATH, path)
.output(OutputMode::Capture)
.spawn()
.expect("the helper process starts");
let stdout = child.take_stdout().expect("captured");
let (sender, receiver) = std::sync::mpsc::channel();
std::thread::spawn(move || {
use std::io::BufRead as _;
for line in std::io::BufReader::new(stdout).lines() {
let Ok(line) = line else { break };
if let Some((_, rest)) = line.split_once(HELPER_READY) {
let _ = sender.send(rest.trim().to_string());
return;
}
}
let _ = sender.send(String::new());
});
let announced = receiver
.recv_timeout(Duration::from_secs(60))
.expect("the helper must announce that it holds the lock");
let pid: u32 = announced
.parse()
.unwrap_or_else(|_| panic!("the helper announced {announced:?} instead of a PID"));
(child, pid)
}
#[test]
fn two_processes_contending_produce_exactly_one_holder() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let (mut helper, helper_pid) = start_helper(&path);
assert_ne!(helper_pid, std::process::id());
let error = HostLock::try_acquire_at(&path, LockKind::SingleInstance)
.expect_err("a second agent must not get the lock");
let LockError::Held { holder, .. } = &error else {
panic!("expected a contention error, got {error}");
};
let holder = holder.as_ref().expect("the helper recorded itself");
assert_eq!(
holder.identity.pid(),
helper_pid,
"the record must name the process that actually holds it"
);
assert!(
error.to_string().contains(&helper_pid.to_string()),
"the loser's message must name the holder: {error}"
);
helper.stop(Duration::ZERO).expect("cleanup");
}
#[test]
fn killing_the_holder_releases_the_lock_with_no_manual_cleanup() {
let directory = tempfile::tempdir().expect("a temporary directory");
let path = lock_path(&directory, LockKind::SingleInstance);
let (mut helper, helper_pid) = start_helper(&path);
assert!(
HostLock::try_acquire_at(&path, LockKind::SingleInstance).is_err(),
"the helper holds it"
);
helper.stop(Duration::ZERO).expect("the helper is killed");
assert!(!helper.is_running().expect("observable"));
let recovered = HostLock::try_acquire_at(&path, LockKind::SingleInstance)
.unwrap_or_else(|error| panic!("the lock leaked after its holder was killed: {error}"));
assert!(
path.exists(),
"the lock file itself must survive; deleting it is how two processes end up \
locking different inodes"
);
let holder = HostLock::holder_of(&path)
.expect("readable")
.expect("the new holder recorded itself");
assert_eq!(holder.identity.pid(), std::process::id());
assert_ne!(
holder.identity.pid(),
helper_pid,
"the record must have been replaced, not inherited"
);
drop(recovered);
}
}