use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
const LOCK_FILE: &str = crate::data_dir::WRITER_LOCK_FILE;
const READERS_DIR: &str = crate::data_dir::READERS_DIR;
static READER_NONCE: AtomicU64 = AtomicU64::new(0);
#[cfg(test)]
#[allow(clippy::type_complexity)]
static BEFORE_FLOCK_HOOKS: std::sync::Mutex<Vec<(u32, Box<dyn FnOnce() + Send>)>> =
std::sync::Mutex::new(Vec::new());
#[cfg(test)]
fn install_before_flock_hook<F: FnOnce() + Send + 'static>(pid: u32, hook: F) {
BEFORE_FLOCK_HOOKS
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.push((pid, Box::new(hook)));
}
#[cfg(test)]
fn run_before_flock_hook(me: u32) {
let hook = {
let mut hooks = BEFORE_FLOCK_HOOKS
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
hooks
.iter()
.position(|(pid, _)| *pid == me)
.map(|index| hooks.swap_remove(index).1)
};
if let Some(hook) = hook {
hook();
}
}
#[cfg(not(test))]
#[inline(always)]
fn run_before_flock_hook(_me: u32) {}
#[derive(Debug)]
enum LockKind {
Writer,
Reader,
ReaderLockless,
}
#[derive(Debug)]
pub struct DirLock {
path: PathBuf,
pid: u32,
kind: LockKind,
flock: Option<fs::File>,
}
impl DirLock {
pub fn acquire(data_dir: &Path) -> io::Result<DirLock> {
Self::acquire_as(data_dir, std::process::id())
}
fn acquire_as(data_dir: &Path, me: u32) -> io::Result<DirLock> {
let path = data_dir.join(LOCK_FILE);
let claimed_owner = live_writer_pid(&path, me);
run_before_flock_hook(me);
let admission = match take_lock_flock(&path) {
FlockAttempt::Held(file) => {
if let Some(ghost) = claimed_owner {
tracing::warn!(
data_dir = %data_dir.display(),
stale_pid = ghost,
"reclaimed stale lock from pid {ghost}: it holds no lock on {}",
path.display()
);
}
Admission::Locked(file)
}
FlockAttempt::Busy => {
let owner = lock_file_pid(&path);
if owner != Some(me) {
return Err(writer_busy_err(data_dir, owner.unwrap_or(0), &path));
}
Admission::OwnHandle
}
FlockAttempt::Unavailable(error) => return Err(error),
FlockAttempt::Unsupported => {
if let Some(owner) = claimed_owner {
return Err(writer_busy_err(data_dir, owner, &path));
}
Admission::PidOnly
}
};
if let Some(reader_pid) = live_reader_pid(data_dir, me) {
admission.back_off(&path, me);
return Err(reader_present_err(data_dir, reader_pid));
}
match &admission {
Admission::Locked(file) => write_pid_in_place(file, me)?,
Admission::OwnHandle => {}
Admission::PidOnly => write_pid_atomically(&path, me)?,
}
if let Some(reader_pid) = live_reader_pid(data_dir, me) {
admission.back_off(&path, me);
return Err(reader_present_err(data_dir, reader_pid));
}
if matches!(admission, Admission::PidOnly) {
if let Some(owner) = live_writer_pid(&path, me) {
return Err(writer_busy_err(data_dir, owner, &path));
}
}
Ok(DirLock {
path,
pid: me,
kind: LockKind::Writer,
flock: admission.into_flock(),
})
}
pub fn acquire_reader(data_dir: &Path) -> io::Result<DirLock> {
Self::acquire_reader_as(data_dir, std::process::id())
}
fn acquire_reader_as(data_dir: &Path, me: u32) -> io::Result<DirLock> {
let lock_path = data_dir.join(LOCK_FILE);
if let Some(owner) = live_writer_pid(&lock_path, me) {
return Err(writer_serving_err(data_dir, owner));
}
let readers_dir = data_dir.join(READERS_DIR);
if let Err(e) = fs::create_dir_all(&readers_dir) {
if is_readonly_dir_error(&e) {
return Ok(lockless_reader_fallback(data_dir));
}
return Err(e);
}
let path = match publish_reader_file(&readers_dir, me) {
Ok(path) => path,
Err(e) if is_readonly_dir_error(&e) => {
return Ok(lockless_reader_fallback(data_dir));
}
Err(e) => return Err(e),
};
if let Some(owner) = live_writer_pid(&lock_path, me) {
let _ = fs::remove_file(&path);
return Err(writer_serving_err(data_dir, owner));
}
Ok(DirLock {
path,
pid: me,
kind: LockKind::Reader,
flock: None,
})
}
}
enum Admission {
Locked(fs::File),
OwnHandle,
PidOnly,
}
impl Admission {
fn back_off(&self, path: &Path, me: u32) {
match self {
Self::Locked(_) => {
let _ = fs::remove_file(path);
}
Self::OwnHandle => {}
Self::PidOnly => remove_lock_if_owned(path, me),
}
}
fn into_flock(self) -> Option<fs::File> {
match self {
Self::Locked(file) => Some(file),
Self::OwnHandle | Self::PidOnly => None,
}
}
}
fn write_pid_in_place(mut file: &fs::File, pid: u32) -> io::Result<()> {
use std::io::{Seek, SeekFrom};
file.set_len(0)?;
file.seek(SeekFrom::Start(0))?;
file.write_all(pid.to_string().as_bytes())?;
let _ = file.sync_all();
Ok(())
}
fn lock_file_pid(lock_path: &Path) -> Option<u32> {
fs::read_to_string(lock_path)
.ok()?
.trim()
.parse::<u32>()
.ok()
}
const MAX_READER_NAME_ATTEMPTS: u32 = 32;
fn write_pid_atomically(final_path: &Path, pid: u32) -> io::Result<()> {
let dir = final_path.parent().unwrap_or_else(|| Path::new("."));
let file_name = final_path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("");
let tmp = dir.join(format!(".{file_name}.tmp.{:032x}", reader_entropy()));
{
let mut f = fs::File::create(&tmp)?;
f.write_all(pid.to_string().as_bytes())?;
let _ = f.sync_all();
}
fs::rename(&tmp, final_path)
}
fn publish_reader_file(readers_dir: &Path, me: u32) -> io::Result<PathBuf> {
for _ in 0..MAX_READER_NAME_ATTEMPTS {
let entropy = format!("{:032x}", reader_entropy());
let tmp = readers_dir.join(format!(".tmp.{entropy}"));
match fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&tmp)
{
Ok(mut f) => {
f.write_all(me.to_string().as_bytes())?;
let _ = f.sync_all();
let final_path = readers_dir.join(format!("{me}.{entropy}"));
fs::rename(&tmp, &final_path)?;
return Ok(final_path);
}
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
Err(e) => return Err(e),
}
}
Err(io::Error::new(
io::ErrorKind::AlreadyExists,
"could not allocate a unique reader lock file name",
))
}
fn reader_entropy() -> u128 {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let counter = u128::from(READER_NONCE.fetch_add(1, Ordering::Relaxed));
nanos ^ (counter << 96) ^ counter.wrapping_mul(0x9E37_79B9_7F4A_7C15)
}
fn lockless_reader_fallback(data_dir: &Path) -> DirLock {
tracing::warn!(
data_dir = %data_dir.display(),
"data directory is not writable; reader lock skipped, writer exclusion is \
not enforced for this process (safe: a writer cannot start against a \
non-writable directory)"
);
DirLock {
path: data_dir.to_path_buf(),
pid: std::process::id(),
kind: LockKind::ReaderLockless,
flock: None,
}
}
fn is_readonly_dir_error(e: &io::Error) -> bool {
matches!(
e.kind(),
io::ErrorKind::PermissionDenied | io::ErrorKind::ReadOnlyFilesystem
) || e.raw_os_error() == Some(libc::EROFS)
|| e.raw_os_error() == Some(libc::EACCES)
}
fn live_writer_pid(lock_path: &Path, me: u32) -> Option<u32> {
let owner = lock_file_pid(lock_path)?;
if owner != me && pid_is_alive(owner) {
Some(owner)
} else {
None
}
}
fn remove_lock_if_owned(lock_path: &Path, me: u32) {
if let Ok(contents) = fs::read_to_string(lock_path) {
if contents.trim().parse::<u32>().ok() == Some(me) {
let _ = fs::remove_file(lock_path);
}
}
}
fn writer_busy_err(data_dir: &Path, owner: u32, lock_path: &Path) -> io::Error {
io::Error::new(
io::ErrorKind::AddrInUse,
format!(
"data directory {} is already open by process {owner}; \
close that instance first (or delete {} if it is gone)",
data_dir.display(),
lock_path.display()
),
)
}
fn reader_present_err(data_dir: &Path, reader_pid: u32) -> io::Error {
io::Error::new(
io::ErrorKind::AddrInUse,
format!(
"data directory {} is being served read-only by process {reader_pid}; \
stop the read-only reader(s) before opening it for writing",
data_dir.display()
),
)
}
fn writer_serving_err(data_dir: &Path, owner: u32) -> io::Error {
io::Error::new(
io::ErrorKind::AddrInUse,
format!(
"data directory {} is open for writing by process {owner}; \
read-only serving requires a quiescent directory",
data_dir.display()
),
)
}
impl Drop for DirLock {
fn drop(&mut self) {
match self.kind {
LockKind::Writer => {
if let Ok(contents) = fs::read_to_string(&self.path) {
if contents.trim().parse::<u32>().ok() == Some(self.pid) {
let _ = fs::remove_file(&self.path);
}
}
drop(self.flock.take());
}
LockKind::Reader => {
let _ = fs::remove_file(&self.path);
}
LockKind::ReaderLockless => {
}
}
}
}
fn live_reader_pid(data_dir: &Path, me: u32) -> Option<u32> {
let readers_dir = data_dir.join(READERS_DIR);
let entries = fs::read_dir(&readers_dir).ok()?;
for entry in entries.flatten() {
let path = entry.path();
if entry
.file_name()
.to_str()
.is_some_and(|n| n.starts_with('.'))
{
continue;
}
let Ok(contents) = fs::read_to_string(&path) else {
continue;
};
let Ok(owner) = contents.trim().parse::<u32>() else {
let _ = fs::remove_file(&path);
continue;
};
if owner == me {
continue;
}
if pid_is_alive(owner) {
return Some(owner);
}
let _ = fs::remove_file(&path);
}
None
}
enum FlockAttempt {
Held(fs::File),
Busy,
Unavailable(io::Error),
Unsupported,
}
#[cfg(unix)]
const MAX_FLOCK_ATTEMPTS: u32 = 32;
#[cfg(unix)]
fn take_lock_flock(lock_path: &Path) -> FlockAttempt {
use std::os::unix::fs::MetadataExt;
use std::os::unix::io::AsRawFd;
for _ in 0..MAX_FLOCK_ATTEMPTS {
let file = match fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)
{
Ok(file) => file,
Err(error) => return FlockAttempt::Unavailable(error),
};
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
let error = io::Error::last_os_error();
let code = error.raw_os_error();
if code == Some(libc::EWOULDBLOCK) {
return FlockAttempt::Busy;
}
if code == Some(libc::ENOLCK)
|| code == Some(libc::EOPNOTSUPP)
|| code == Some(libc::ENOTSUP)
|| code == Some(libc::ENOSYS)
{
tracing::warn!(
lock_file = %lock_path.display(),
%error,
"this filesystem does not support advisory file locks; falling back to \
the PID heuristic, which cannot tell a live writer in another PID \
namespace from a stale lock"
);
return FlockAttempt::Unsupported;
}
return FlockAttempt::Unavailable(error);
}
if let (Ok(held), Ok(named)) = (file.metadata(), fs::metadata(lock_path)) {
if held.ino() == named.ino() && held.dev() == named.dev() {
return FlockAttempt::Held(file);
}
}
}
FlockAttempt::Unavailable(io::Error::new(
io::ErrorKind::AddrInUse,
format!(
"{}: the writer lock file kept being replaced while we were locking it",
lock_path.display()
),
))
}
#[cfg(not(unix))]
fn take_lock_flock(_lock_path: &Path) -> FlockAttempt {
FlockAttempt::Unsupported
}
#[cfg(unix)]
fn pid_is_alive(pid: u32) -> bool {
if pid == 0 {
return false;
}
if unsafe { libc::kill(pid as libc::pid_t, 0) } == 0 {
return true;
}
io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}
#[cfg(not(unix))]
fn pid_is_alive(_pid: u32) -> bool {
true
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn acquire_then_drop_removes_lock_file() {
let dir = tempfile::tempdir().unwrap();
{
let _lock = DirLock::acquire(dir.path()).unwrap();
assert!(dir.path().join(LOCK_FILE).exists());
}
assert!(
!dir.path().join(LOCK_FILE).exists(),
"drop should remove it"
);
}
#[test]
fn same_process_reacquire_is_allowed() {
let dir = tempfile::tempdir().unwrap();
let first = DirLock::acquire(dir.path()).unwrap();
std::mem::forget(first); assert!(DirLock::acquire(dir.path()).is_ok());
}
#[cfg(unix)]
#[test]
fn a_lock_naming_a_live_pid_that_holds_no_flock_is_reclaimed() {
let dir = tempfile::tempdir().unwrap();
let ghost = unsafe { libc::getppid() } as u32;
assert_ne!(ghost, std::process::id());
assert!(pid_is_alive(ghost), "the test needs a live stand-in pid");
fs::write(dir.path().join(LOCK_FILE), ghost.to_string()).unwrap();
let lock = DirLock::acquire(dir.path()).expect("a lock nobody holds must be reclaimable");
assert_eq!(
fs::read_to_string(dir.path().join(LOCK_FILE)).unwrap(),
std::process::id().to_string(),
"the reclaimed lock must now name us"
);
drop(lock);
}
#[cfg(unix)]
#[test]
fn a_lock_whose_flock_is_held_is_still_refused() {
let dir = tempfile::tempdir().unwrap();
let other = unsafe { libc::getppid() } as u32;
assert_ne!(other, std::process::id());
let held = DirLock::acquire_as(dir.path(), other).expect("stand-in writer");
assert!(held.flock.is_some(), "the writer must hold the flock");
let err = DirLock::acquire(dir.path())
.expect_err("a directory a live process holds must stay refused");
assert_eq!(err.kind(), io::ErrorKind::AddrInUse);
assert!(
err.to_string().contains(&other.to_string()),
"the refusal must name the holder, got: {err}"
);
drop(held);
}
#[cfg(unix)]
#[test]
fn two_writers_racing_for_the_flock_are_not_both_admitted() {
use std::sync::mpsc;
use std::time::Duration;
let dir = tempfile::tempdir().unwrap();
let mut first_proc = std::process::Command::new("sleep")
.arg("120")
.spawn()
.unwrap();
let mut second_proc = std::process::Command::new("sleep")
.arg("120")
.spawn()
.unwrap();
let first_pid = first_proc.id();
let second_pid = second_proc.id();
let (parked_tx, parked_rx) = mpsc::channel::<()>();
let (resume_tx, resume_rx) = mpsc::channel::<()>();
install_before_flock_hook(first_pid, move || {
let _ = parked_tx.send(());
let _ = resume_rx.recv();
});
let path = dir.path().to_path_buf();
let first = std::thread::spawn(move || DirLock::acquire_as(&path, first_pid));
parked_rx
.recv_timeout(Duration::from_secs(60))
.expect("the first acquisition never reached the flock");
let second = DirLock::acquire_as(dir.path(), second_pid);
let _ = resume_tx.send(());
let first = first.join().expect("the acquiring thread must not panic");
let admitted = usize::from(first.is_ok()) + usize::from(second.is_ok());
assert_eq!(
admitted, 1,
"exactly one of two racing writers may hold the directory, \
but {admitted} were admitted (first: {first:?}, second: {second:?})"
);
drop(first);
drop(second);
let _ = first_proc.kill();
let _ = second_proc.kill();
let _ = first_proc.wait();
let _ = second_proc.wait();
}
#[cfg(unix)]
#[test]
fn the_pid_is_published_only_after_the_lock_is_held() {
use std::sync::mpsc;
use std::time::Duration;
let dir = tempfile::tempdir().unwrap();
let lock_path = dir.path().join(LOCK_FILE);
let mut stand_in = std::process::Command::new("sleep")
.arg("120")
.spawn()
.unwrap();
let pid = stand_in.id();
let (parked_tx, parked_rx) = mpsc::channel::<()>();
let (resume_tx, resume_rx) = mpsc::channel::<()>();
install_before_flock_hook(pid, move || {
let _ = parked_tx.send(());
let _ = resume_rx.recv();
});
let path = dir.path().to_path_buf();
let acquiring = std::thread::spawn(move || DirLock::acquire_as(&path, pid));
parked_rx
.recv_timeout(Duration::from_secs(60))
.expect("the acquisition never reached the flock");
assert_eq!(
lock_file_pid(&lock_path),
None,
"LOCK named a writer that did not hold the lock yet"
);
let _ = resume_tx.send(());
let lock = acquiring
.join()
.expect("the acquiring thread must not panic")
.expect("nothing else holds the directory");
assert_eq!(
lock_file_pid(&lock_path),
Some(pid),
"once the lock is held, LOCK must name its holder"
);
drop(lock);
let _ = stand_in.kill();
let _ = stand_in.wait();
}
#[test]
fn dead_owner_pid_is_a_dead_pid() {
let mut child = std::process::Command::new("true").spawn().unwrap();
let pid = child.id();
child.wait().unwrap();
assert!(!pid_is_alive(pid));
}
#[test]
fn two_readers_coexist() {
let dir = tempfile::tempdir().unwrap();
let r1 = DirLock::acquire_reader(dir.path()).unwrap();
let r2 = DirLock::acquire_reader(dir.path()).unwrap();
let count = fs::read_dir(dir.path().join(READERS_DIR)).unwrap().count();
assert_eq!(count, 2, "two readers should leave two reader files");
drop(r1);
drop(r2);
let count = fs::read_dir(dir.path().join(READERS_DIR)).unwrap().count();
assert_eq!(count, 0, "reader drops should remove their files");
}
#[test]
fn reader_excludes_live_writer() {
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join(LOCK_FILE), "1").unwrap();
let err = DirLock::acquire_reader(dir.path()).unwrap_err();
assert_eq!(err.kind(), io::ErrorKind::AddrInUse);
}
#[test]
fn writer_excludes_live_reader() {
let dir = tempfile::tempdir().unwrap();
let readers = dir.path().join(READERS_DIR);
fs::create_dir_all(&readers).unwrap();
fs::write(readers.join("1.0"), "1").unwrap();
let err = DirLock::acquire(dir.path()).unwrap_err();
assert_eq!(err.kind(), io::ErrorKind::AddrInUse);
}
#[test]
fn writer_reclaims_dead_reader_files() {
let dir = tempfile::tempdir().unwrap();
let readers = dir.path().join(READERS_DIR);
fs::create_dir_all(&readers).unwrap();
let mut child = std::process::Command::new("true").spawn().unwrap();
let dead_pid = child.id();
child.wait().unwrap();
fs::write(readers.join(format!("{dead_pid}.0")), dead_pid.to_string()).unwrap();
let _w = DirLock::acquire(dir.path()).unwrap();
assert!(
!readers.join(format!("{dead_pid}.0")).exists(),
"dead reader file should be reclaimed"
);
}
#[test]
fn reader_reclaims_dead_writer_lock() {
let dir = tempfile::tempdir().unwrap();
let mut child = std::process::Command::new("true").spawn().unwrap();
let dead_pid = child.id();
child.wait().unwrap();
fs::write(dir.path().join(LOCK_FILE), dead_pid.to_string()).unwrap();
let _r = DirLock::acquire_reader(dir.path()).unwrap();
}
#[cfg(unix)]
#[test]
fn reader_falls_back_lock_less_on_non_writable_dir() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("catalog"), b"x").unwrap();
let orig = fs::metadata(dir.path()).unwrap().permissions();
fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o555)).unwrap();
let result = DirLock::acquire_reader(dir.path());
fs::set_permissions(dir.path(), orig).unwrap();
let lock = result.expect("read-only open must succeed on a non-writable dir");
assert!(
matches!(lock.kind, LockKind::ReaderLockless),
"a non-writable dir must fall back to a lock-less reader"
);
assert!(
!dir.path().join(READERS_DIR).exists(),
"the lock-less fallback must not mutate the directory"
);
}
#[test]
fn reader_names_are_pid_reuse_proof() {
let dir = tempfile::tempdir().unwrap();
let readers = dir.path().join(READERS_DIR);
fs::create_dir_all(&readers).unwrap();
let recycled_pid = std::process::id();
let stale = readers.join(format!("{recycled_pid}.0"));
fs::write(&stale, recycled_pid.to_string()).unwrap();
let lock = DirLock::acquire_reader_as(dir.path(), recycled_pid).unwrap();
assert!(
stale.exists(),
"stale reader file must not be adopted/truncated"
);
assert_ne!(
lock.path, stale,
"a recycled PID must not reuse the stale file's name"
);
let count = fs::read_dir(&readers).unwrap().count();
assert_eq!(
count, 2,
"stale + live reader files coexist without collision"
);
}
#[cfg(unix)]
#[test]
fn reader_and_writer_are_never_both_admitted() {
let writer_child = std::process::Command::new("sleep")
.arg("30")
.spawn()
.unwrap();
let reader_child = std::process::Command::new("sleep")
.arg("30")
.spawn()
.unwrap();
let writer_pid = writer_child.id();
let reader_pid = reader_child.id();
let mut both_admitted = 0u32;
for _ in 0..200 {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().to_path_buf();
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let (w_res, r_res) = std::thread::scope(|scope| {
let wb = barrier.clone();
let wp = path.clone();
let w = scope.spawn(move || {
wb.wait();
DirLock::acquire_as(&wp, writer_pid)
});
let rb = barrier.clone();
let rp = path.clone();
let r = scope.spawn(move || {
rb.wait();
DirLock::acquire_reader_as(&rp, reader_pid)
});
(w.join().unwrap(), r.join().unwrap())
});
if w_res.is_ok() && r_res.is_ok() {
both_admitted += 1;
}
}
let mut writer_child = writer_child;
let mut reader_child = reader_child;
let _ = writer_child.kill();
let _ = reader_child.kill();
let _ = writer_child.wait();
let _ = reader_child.wait();
assert_eq!(
both_admitted, 0,
"a reader and a writer were both admitted in {both_admitted} race(s)"
);
}
}