use std::fs::{self, File, OpenOptions};
#[cfg(not(target_os = "wasi"))]
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
#[cfg(unix)]
use std::time::{Duration, Instant};
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
use radixdb_core::{Error, Result};
#[derive(Debug, Clone)]
pub struct FileLock {
inner: Arc<FileLockInner>,
}
#[derive(Debug)]
struct FileLockInner {
#[allow(dead_code)]
file: File,
path: PathBuf,
root: PathBuf,
root_identity: FilesystemIdentity,
lock_identity: FilesystemIdentity,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct FilesystemIdentity {
#[cfg(unix)]
device: u64,
#[cfg(unix)]
inode: u64,
}
impl FileLock {
pub fn acquire(db_path: impl AsRef<Path>) -> Result<Self> {
let db_path = db_path.as_ref();
fs::create_dir_all(db_path)
.map_err(|e| Error::internal(format!("failed to create database directory: {}", e)))?;
let lock_file_path = db_path.join("LOCK");
#[allow(unused_mut)]
let mut file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&lock_file_path)
.map_err(|e| Error::internal(format!("failed to open lock file: {}", e)))?;
acquire_lock_for_open(&file, &lock_file_path)?;
#[cfg(not(target_os = "wasi"))]
{
file.set_len(0)
.map_err(|e| Error::internal(format!("failed to truncate lock file: {}", e)))?;
let pid = std::process::id();
write!(file, "{}", pid).ok();
file.sync_all().ok();
}
let root = fs::canonicalize(db_path)
.map_err(|e| Error::internal(format!("failed to resolve database directory: {e}")))?;
let root_identity =
filesystem_identity(&fs::metadata(&root).map_err(|e| {
Error::internal(format!("failed to inspect database directory: {e}"))
})?);
let lock_identity = filesystem_identity(
&file
.metadata()
.map_err(|e| Error::internal(format!("failed to inspect lock file: {e}")))?,
);
Ok(Self {
inner: Arc::new(FileLockInner {
file,
path: root.join("LOCK"),
root,
root_identity,
lock_identity,
}),
})
}
pub fn path(&self) -> &Path {
&self.inner.path
}
pub(crate) fn root(&self) -> &Path {
&self.inner.root
}
pub(crate) fn validate_root(&self, path: &Path) -> std::io::Result<bool> {
let canonical = fs::canonicalize(path)?;
let root_identity = filesystem_identity(&fs::metadata(&canonical)?);
let lock_identity = filesystem_identity(&fs::metadata(canonical.join("LOCK"))?);
Ok(root_identity == self.inner.root_identity
&& lock_identity == self.inner.lock_identity
&& canonical == self.inner.root)
}
}
#[cfg(unix)]
fn filesystem_identity(metadata: &fs::Metadata) -> FilesystemIdentity {
FilesystemIdentity {
device: metadata.dev(),
inode: metadata.ino(),
}
}
#[cfg(not(unix))]
fn filesystem_identity(_metadata: &fs::Metadata) -> FilesystemIdentity {
FilesystemIdentity {}
}
fn acquire_lock(file: &File) -> Result<()> {
use std::os::unix::io::AsRawFd;
let fd = file.as_raw_fd();
let result = unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) };
if result != 0 {
let errno = std::io::Error::last_os_error();
if errno.raw_os_error() == Some(libc::EWOULDBLOCK) {
return Err(Error::DatabaseLocked);
}
return Err(Error::internal(format!(
"failed to acquire lock: {}",
errno
)));
}
Ok(())
}
#[cfg(unix)]
fn acquire_lock_for_open(file: &File, lock_file_path: &Path) -> Result<()> {
const SAME_PROCESS_HANDOFF_GRACE: Duration = Duration::from_millis(100);
match acquire_lock(file) {
Ok(()) => return Ok(()),
Err(error) if !matches!(error, Error::DatabaseLocked) => return Err(error),
Err(_) => {}
}
let owned_by_current_process = fs::read_to_string(lock_file_path)
.ok()
.and_then(|contents| contents.trim().parse::<u32>().ok())
.is_some_and(|pid| pid == std::process::id());
if !owned_by_current_process {
return Err(Error::DatabaseLocked);
}
let deadline = Instant::now() + SAME_PROCESS_HANDOFF_GRACE;
loop {
std::thread::sleep(Duration::from_millis(1));
match acquire_lock(file) {
Ok(()) => return Ok(()),
Err(error) if !matches!(error, Error::DatabaseLocked) => return Err(error),
Err(_) if Instant::now() < deadline => {}
Err(_) => return Err(Error::DatabaseLocked),
}
}
}
#[cfg(not(unix))]
fn acquire_lock_for_open(file: &File, _lock_file_path: &Path) -> Result<()> {
acquire_lock(file)
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
#[test]
fn test_acquire_lock() {
let dir = tempdir().unwrap();
let db_path = dir.path().join("test_db");
let lock = FileLock::acquire(&db_path).unwrap();
assert!(db_path.join("LOCK").exists());
let contents = fs::read_to_string(db_path.join("LOCK")).unwrap();
assert_eq!(contents, std::process::id().to_string());
drop(lock);
}
#[test]
fn test_lock_prevents_second_acquisition() {
let dir = tempdir().unwrap();
let db_path = dir.path().join("test_db");
let _lock1 = FileLock::acquire(&db_path).unwrap();
let result = FileLock::acquire(&db_path);
assert!(result.is_err());
assert!(result
.unwrap_err()
.to_string()
.contains("locked by another process"));
}
#[test]
fn cloned_owner_keeps_the_same_writer_lock_alive() {
let dir = tempdir().unwrap();
let db_path = dir.path().join("shared_owner");
let lock = FileLock::acquire(&db_path).unwrap();
let publisher_owner = lock.clone();
drop(lock);
assert!(matches!(
FileLock::acquire(&db_path).unwrap_err(),
Error::DatabaseLocked
));
drop(publisher_owner);
FileLock::acquire(&db_path).unwrap();
}
#[cfg(unix)]
#[test]
fn same_process_lock_handoff_waits_for_a_retained_clone() {
let dir = tempdir().unwrap();
let db_path = dir.path().join("same-process-handoff");
let lock = FileLock::acquire(&db_path).unwrap();
let retained = lock.clone();
let release = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(10));
drop(retained);
});
drop(lock);
FileLock::acquire(&db_path)
.expect("same-process lock handoff must tolerate fork-sized lag");
release.join().unwrap();
}
#[test]
fn test_lock_released_on_drop() {
let dir = tempdir().unwrap();
let db_path = dir.path().join("test_db");
{
let _lock = FileLock::acquire(&db_path).unwrap();
}
let _lock2 = FileLock::acquire(&db_path).unwrap();
}
}