use std::collections::HashSet;
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};
use crate::error::SqliteStoreError;
fn held_exclusive_locks() -> &'static Mutex<HashSet<PathBuf>> {
static REGISTRY: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
REGISTRY.get_or_init(|| Mutex::new(HashSet::new()))
}
fn registry_key(lock_path: &Path) -> PathBuf {
std::fs::canonicalize(lock_path).unwrap_or_else(|_| lock_path.to_path_buf())
}
fn process_holds_exclusive(lock_path: &Path) -> bool {
held_exclusive_locks()
.lock()
.map(|held| held.contains(®istry_key(lock_path)))
.unwrap_or(false)
}
pub const FENCE_LOCK_SUFFIX: &str = "mfence";
pub const PRESENCE_LOCK_SUFFIX: &str = "mpresence";
pub fn fence_lock_path(db_path: &Path) -> PathBuf {
sibling_lock_path(db_path, FENCE_LOCK_SUFFIX)
}
pub fn presence_lock_path(db_path: &Path) -> PathBuf {
sibling_lock_path(db_path, PRESENCE_LOCK_SUFFIX)
}
fn sibling_lock_path(db_path: &Path, suffix: &str) -> PathBuf {
let mut name = db_path.file_name().unwrap_or_default().to_os_string();
name.push(".");
name.push(suffix);
db_path.with_file_name(name)
}
fn open_lock_file(lock_path: &Path) -> Option<File> {
OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)
.ok()
}
#[derive(Debug)]
pub struct OperationGuard {
_lock: Option<File>,
}
impl OperationGuard {
pub fn try_for_database_strict(db_path: &Path) -> Result<Self, SqliteStoreError> {
if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
return Err(SqliteStoreError::Io(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"strict operation custody requires an existing physical database",
)));
}
crate::profile::validate_database_file(db_path)?;
if !db_path.is_file() {
return Err(SqliteStoreError::Io(std::io::Error::new(
std::io::ErrorKind::NotFound,
"strict operation database is missing",
)));
}
#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
{
Err(SqliteStoreError::Io(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"strict operation custody requires descriptor-scoped locking",
)))
}
#[cfg(any(target_os = "macos", target_os = "linux", windows))]
{
let path = fence_lock_path(db_path);
crate::profile::validate_database_file(&path)?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)?;
crate::profile::validate_database_file(&path)?;
if !file.metadata()?.is_file() {
return Err(SqliteStoreError::Io(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"strict operation lock is not a regular file",
)));
}
match file.try_lock_shared() {
Ok(()) => Ok(Self { _lock: Some(file) }),
Err(error) if is_would_block(&error) => {
Err(SqliteStoreError::MaintenanceFenceHeld {
path: db_path.to_path_buf(),
})
}
Err(error) => Err(SqliteStoreError::Io(std::io::Error::other(error))),
}
}
}
pub fn for_database(db_path: &Path) -> Result<Self, SqliteStoreError> {
if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
return Ok(Self { _lock: None });
}
crate::profile::validate_database_file(db_path)?;
let lock_path = fence_lock_path(db_path);
if process_holds_exclusive(&lock_path) {
return Ok(Self { _lock: None });
}
let Some(file) = open_lock_file(&lock_path) else {
return Ok(Self { _lock: None });
};
match file.try_lock_shared() {
Ok(()) => Ok(Self { _lock: Some(file) }),
Err(err) if is_would_block(&err) => Err(SqliteStoreError::MaintenanceFenceHeld {
path: db_path.to_path_buf(),
}),
Err(_) => Ok(Self { _lock: None }),
}
}
pub fn admit_missing_database(db_path: &Path) -> Result<(), SqliteStoreError> {
if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
return Ok(());
}
let lock_path = fence_lock_path(db_path);
if process_holds_exclusive(&lock_path) {
return Ok(());
}
let Ok(file) = OpenOptions::new().read(true).write(true).open(&lock_path) else {
return Ok(());
};
match file.try_lock_shared() {
Err(err) if is_would_block(&err) => Err(SqliteStoreError::MaintenanceFenceHeld {
path: db_path.to_path_buf(),
}),
Ok(()) | Err(_) => Ok(()),
}
}
}
#[derive(Debug)]
pub struct KeptConnectionPresence {
_lock: File,
}
impl KeptConnectionPresence {
pub fn acquire(db_path: &Path) -> Option<Self> {
if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
return None;
}
if process_holds_exclusive(&fence_lock_path(db_path)) {
return None;
}
let Ok(Some(_)) = crate::profile::validate_database_file(db_path) else {
return None;
};
let file = open_lock_file(&presence_lock_path(db_path))?;
file.try_lock_shared().ok()?;
Some(Self { _lock: file })
}
}
#[derive(Debug)]
pub struct ExclusiveFence {
_lock: File,
_presence: File,
lock_path: PathBuf,
}
impl ExclusiveFence {
pub fn try_acquire(db_path: &Path) -> Result<Option<Self>, SqliteStoreError> {
let lock_path = fence_lock_path(db_path);
let Some(fence) = try_lock_exclusive(&lock_path)? else {
return Ok(None);
};
let Some(presence) = try_lock_exclusive(&presence_lock_path(db_path))? else {
return Ok(None);
};
Ok(Some(Self::held(fence, presence, lock_path)))
}
pub fn acquire(db_path: &Path, deadline: Duration) -> Result<Self, SqliteStoreError> {
let started = Instant::now();
let lock_path = fence_lock_path(db_path);
let fence = loop {
if let Some(fence) = try_lock_exclusive(&lock_path)? {
break fence;
}
if started.elapsed() >= deadline {
return Err(SqliteStoreError::MaintenanceFenceHeld {
path: db_path.to_path_buf(),
});
}
std::thread::sleep(FENCE_POLL);
};
let presence_path = presence_lock_path(db_path);
loop {
if let Some(presence) = try_lock_exclusive(&presence_path)? {
return Ok(Self::held(fence, presence, lock_path));
}
if started.elapsed() >= deadline {
return Err(SqliteStoreError::MaintenanceFenceHeld {
path: db_path.to_path_buf(),
});
}
std::thread::sleep(FENCE_POLL);
}
}
fn held(fence: File, presence: File, lock_path: PathBuf) -> Self {
if let Ok(mut held) = held_exclusive_locks().lock() {
held.insert(registry_key(&lock_path));
}
Self {
_lock: fence,
_presence: presence,
lock_path,
}
}
pub fn lock_path(&self) -> &Path {
&self.lock_path
}
}
const FENCE_POLL: Duration = Duration::from_millis(25);
fn try_lock_exclusive(lock_path: &Path) -> Result<Option<File>, SqliteStoreError> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)?;
match file.try_lock() {
Ok(()) => Ok(Some(file)),
Err(err) if is_would_block(&err) => Ok(None),
Err(err) => Err(SqliteStoreError::Io(std::io::Error::other(err))),
}
}
impl Drop for ExclusiveFence {
fn drop(&mut self) {
if let Ok(mut held) = held_exclusive_locks().lock() {
held.remove(®istry_key(&self.lock_path));
}
}
}
fn is_would_block(err: &std::fs::TryLockError) -> bool {
matches!(err, std::fs::TryLockError::WouldBlock)
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
use super::*;
#[test]
fn shared_guards_coexist() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
let g1 = OperationGuard::for_database(&db).expect("g1");
let g2 = OperationGuard::for_database(&db).expect("g2");
drop((g1, g2));
}
#[test]
fn exclusive_fence_blocks_foreign_operations_typed() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
let lock_path = fence_lock_path(&db);
let foreign = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&lock_path)
.expect("open lock file");
foreign.try_lock().expect("foreign exclusive lock");
let err = OperationGuard::for_database(&db).expect_err("fence must block foreign ops");
assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
drop(foreign);
OperationGuard::for_database(&db).expect("released after drop");
}
#[test]
fn a_missing_database_honours_a_held_fence_and_creates_nothing() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
OperationGuard::admit_missing_database(&db).expect("no lock file: no fence");
assert!(
!fence_lock_path(&db).exists(),
"admission creates no lock file"
);
let lock_path = fence_lock_path(&db);
let foreign = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&lock_path)
.expect("open lock file");
foreign.try_lock().expect("foreign exclusive lock");
let err = OperationGuard::admit_missing_database(&db)
.expect_err("a foreign fence covers the missing database");
assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
drop(foreign);
OperationGuard::admit_missing_database(&db).expect("released after drop");
let own = ExclusiveFence::try_acquire(&db)
.expect("io")
.expect("own fence");
OperationGuard::admit_missing_database(&db).expect("the fence holder is admitted");
drop(own);
assert!(!db.exists(), "nothing created the database");
}
#[test]
fn exclusive_fence_waits_for_guard_drain() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
let guard = OperationGuard::for_database(&db).expect("guard");
assert!(
ExclusiveFence::try_acquire(&db).expect("io").is_none(),
"fence must not be acquirable while an operation is in flight"
);
let handle = std::thread::spawn(move || {
ExclusiveFence::acquire(&db, Duration::from_secs(5)).expect("acquire")
});
std::thread::sleep(Duration::from_millis(100));
drop(guard);
let fence = handle.join().expect("thread");
assert!(fence.lock_path().ends_with("db.sqlite3.mfence"));
}
#[test]
fn fence_holder_operations_self_admit() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
let fence = ExclusiveFence::try_acquire(&db)
.expect("io")
.expect("acquired");
let guard = OperationGuard::for_database(&db).expect("holder self-admission");
drop(guard);
drop(fence);
OperationGuard::for_database(&db).expect("released");
let fence = ExclusiveFence::try_acquire(&db)
.expect("io")
.expect("reacquired");
drop(fence);
}
#[test]
fn in_memory_paths_get_noop_guards() {
let guard = OperationGuard::for_database(Path::new(":memory:")).expect("noop");
drop(guard);
}
#[test]
fn fence_lock_path_shape() {
assert_eq!(
fence_lock_path(Path::new("/a/b/sessions.sqlite3")),
PathBuf::from("/a/b/sessions.sqlite3.mfence")
);
assert_eq!(
presence_lock_path(Path::new("/a/b/sessions.sqlite3")),
PathBuf::from("/a/b/sessions.sqlite3.mpresence")
);
}
fn existing_database(dir: &Path) -> PathBuf {
let db = dir.join("db.sqlite3");
std::fs::write(&db, b"").expect("database file");
db
}
#[test]
fn kept_presence_refuses_the_fence_and_a_refused_fence_holds_nothing() {
let dir = tempfile::tempdir().expect("tempdir");
let db = existing_database(dir.path());
let presence = KeptConnectionPresence::acquire(&db).expect("presence");
assert!(
ExclusiveFence::try_acquire(&db).expect("io").is_none(),
"the fence must not be taken while a connection is kept"
);
OperationGuard::for_database(&db).expect("a refused try holds no fence lock");
let err = ExclusiveFence::acquire(&db, Duration::from_millis(200))
.expect_err("a kept connection outlives the deadline");
assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
OperationGuard::for_database(&db).expect("an expired wait holds no fence lock");
drop(presence);
let fence = ExclusiveFence::try_acquire(&db)
.expect("io")
.expect("taken once the kept connection closed");
drop(fence);
}
#[test]
fn a_waiting_fence_refuses_the_kept_readers_next_read_and_completes_when_it_closes() {
let dir = tempfile::tempdir().expect("tempdir");
let db = existing_database(dir.path());
let presence = KeptConnectionPresence::acquire(&db).expect("presence");
let waiting = {
let db = db.clone();
std::thread::spawn(move || ExclusiveFence::acquire(&db, Duration::from_secs(10)))
};
let started = Instant::now();
loop {
match OperationGuard::for_database(&db) {
Err(SqliteStoreError::MaintenanceFenceHeld { .. }) => break,
Ok(guard) => drop(guard),
Err(other) => panic!("unexpected guard error: {other}"),
}
assert!(
started.elapsed() < Duration::from_secs(5),
"the waiting fence never took its fence lock"
);
std::thread::sleep(Duration::from_millis(5));
}
assert!(
!waiting.is_finished(),
"the fence waits for the kept reader"
);
drop(presence);
let fence = waiting
.join()
.expect("thread")
.expect("fence after the reader closed");
assert!(
KeptConnectionPresence::acquire(&db).is_none(),
"no connection is kept while the fence is held"
);
drop(fence);
KeptConnectionPresence::acquire(&db).expect("presence after release");
}
#[test]
fn presence_is_refused_while_the_fence_is_held() {
let dir = tempfile::tempdir().expect("tempdir");
let db = existing_database(dir.path());
let own = ExclusiveFence::try_acquire(&db)
.expect("io")
.expect("own fence");
assert!(KeptConnectionPresence::acquire(&db).is_none());
OperationGuard::for_database(&db).expect("holder self-admission");
drop(own);
let foreign = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(presence_lock_path(&db))
.expect("open presence lock file");
foreign.try_lock().expect("foreign exclusive presence lock");
assert!(KeptConnectionPresence::acquire(&db).is_none());
drop(foreign);
KeptConnectionPresence::acquire(&db).expect("presence after release");
}
#[test]
fn presence_plants_nothing_next_to_a_refused_database() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
std::fs::create_dir(&db).expect("a directory where the database belongs");
assert!(KeptConnectionPresence::acquire(&db).is_none());
assert!(!presence_lock_path(&db).exists());
assert!(KeptConnectionPresence::acquire(Path::new(":memory:")).is_none());
}
#[test]
fn presence_plants_nothing_next_to_a_missing_database() {
let dir = tempfile::tempdir().expect("tempdir");
let db = dir.path().join("db.sqlite3");
assert!(KeptConnectionPresence::acquire(&db).is_none());
assert!(!presence_lock_path(&db).exists());
assert_eq!(
std::fs::read_dir(dir.path()).expect("read dir").count(),
0,
"nothing planted"
);
}
}