use std::fs::{File, OpenOptions};
use std::io::{self, Read};
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use fs2::FileExt;
use crate::graph::dir_graph::DirGraph;
use crate::graph::io::file::load_file;
use crate::graph::storage::mode::{
convert_dir_graph_to_mode, live_storage_mode, new_dir_graph_in_mode, StorageMode,
};
use crate::graph::wal::DurabilityLevel;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OpenDisposition {
Opened,
Created,
}
pub struct OpenGraphResult {
pub graph: Arc<DirGraph>,
pub disposition: OpenDisposition,
pub identity: GraphFileIdentity,
pub converted_from: Option<StorageMode>,
}
pub struct GraphWriterLease {
file: File,
}
#[derive(Debug)]
pub struct LeaseRefusal {
pub holder: Option<LeaseHolder>,
pub error: io::Error,
}
impl From<LeaseRefusal> for io::Error {
fn from(refusal: LeaseRefusal) -> Self {
refusal.error
}
}
impl GraphWriterLease {
pub fn acquire(graph_path: &Path, timeout: Duration) -> io::Result<Self> {
Self::acquire_ex(graph_path, timeout).map_err(io::Error::from)
}
pub fn acquire_ex(graph_path: &Path, timeout: Duration) -> Result<Self, LeaseRefusal> {
let path = writer_lease_path(graph_path);
let started = Instant::now();
loop {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)
.map_err(LeaseRefusal::io)?;
match file.try_lock_exclusive() {
Ok(()) => {
publish_owner_record(&writer_owner_path(graph_path));
crate::graph::io::file::reap_stale_save_temps(graph_path);
return Ok(Self { file });
}
Err(error) if is_lock_contended(&error) => {
if started.elapsed() >= timeout {
let holder = LeaseHolder::read(&writer_owner_path(graph_path));
let message = contended_message(graph_path, &path, &holder);
return Err(LeaseRefusal {
holder: Some(holder),
error: io::Error::new(io::ErrorKind::WouldBlock, message),
});
}
std::thread::sleep(Duration::from_millis(50));
}
Err(error) => return Err(LeaseRefusal::io(error)),
}
}
}
}
impl LeaseRefusal {
fn io(error: io::Error) -> Self {
Self {
holder: None,
error,
}
}
}
impl Drop for GraphWriterLease {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
fn writer_lease_path(graph_path: &Path) -> std::path::PathBuf {
let mut lock = graph_path.as_os_str().to_os_string();
lock.push(".lock");
lock.into()
}
fn is_lock_contended(error: &io::Error) -> bool {
error.raw_os_error() == fs2::lock_contended_error().raw_os_error()
|| error.kind() == io::ErrorKind::WouldBlock
}
fn writer_owner_path(graph_path: &Path) -> std::path::PathBuf {
let mut owner = graph_path.as_os_str().to_os_string();
owner.push(".lock-owner");
owner.into()
}
fn publish_owner_record(owner_path: &Path) {
let record = format!(
"pid={}\nsince={}\n",
std::process::id(),
chrono::Local::now().to_rfc3339()
);
let _ = std::fs::write(owner_path, record);
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct LeaseHolder {
pub pid: Option<u32>,
pub since: Option<String>,
}
impl LeaseHolder {
fn read(owner_path: &Path) -> Self {
const ATTEMPTS: u32 = 10;
for attempt in 0..ATTEMPTS {
let holder = Self::read_once(owner_path);
if holder.pid.is_some() {
return holder;
}
if attempt + 1 < ATTEMPTS {
std::thread::sleep(Duration::from_millis(20));
}
}
Self::default()
}
fn read_once(owner_path: &Path) -> Self {
let Ok(text) = std::fs::read_to_string(owner_path) else {
return Self::default();
};
let mut holder = Self::default();
for line in text.lines() {
match line.split_once('=') {
Some(("pid", value)) => holder.pid = value.trim().parse().ok(),
Some(("since", value)) => holder.since = Some(value.trim().to_string()),
_ => {}
}
}
holder
}
pub fn is_self(&self) -> bool {
self.pid == Some(std::process::id())
}
fn describe(&self) -> String {
if self.is_self() {
return format!(
"this same process (pid {}), which has not closed an earlier open() of it",
std::process::id()
);
}
match (self.pid, self.since.as_deref()) {
(Some(pid), Some(since)) => format!("pid {pid} (since {since})"),
(Some(pid), None) => format!("pid {pid}"),
_ => "another process".to_string(),
}
}
}
fn contended_message(graph_path: &Path, lock_path: &Path, holder: &LeaseHolder) -> String {
format!(
"{} is open for writing by {}; only one process may write a graph at a time. \
The lock is released automatically when that process exits, even on a crash — \
deleting {} does not release it.",
graph_path.display(),
holder.describe(),
lock_path.display()
)
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct MetadataIdentity {
len: u64,
modified: SystemTime,
#[cfg(unix)]
device: u64,
#[cfg(unix)]
inode: u64,
#[cfg(windows)]
handle: Arc<same_file::Handle>,
}
impl MetadataIdentity {
fn capture(path: &Path) -> io::Result<(Self, std::fs::Metadata)> {
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
#[cfg(not(windows))]
let metadata = std::fs::metadata(path)?;
#[cfg(windows)]
let (metadata, handle) = {
let handle = Arc::new(same_file::Handle::from_path(path)?);
let metadata = handle.as_file().metadata()?;
(metadata, handle)
};
Ok((
Self {
len: metadata.len(),
modified: metadata.modified()?,
#[cfg(unix)]
device: metadata.dev(),
#[cfg(unix)]
inode: metadata.ino(),
#[cfg(windows)]
handle,
},
metadata,
))
}
fn open_snapshot(&self, _path: &Path) -> io::Result<File> {
#[cfg(windows)]
return self.handle.as_file().try_clone();
#[cfg(not(windows))]
File::open(_path)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GraphFileIdentity {
root: Option<MetadataIdentity>,
current: Option<(MetadataIdentity, Vec<u8>)>,
}
impl GraphFileIdentity {
pub fn capture(path: &Path) -> io::Result<Self> {
let (root, metadata) = match MetadataIdentity::capture(path) {
Ok(captured) => captured,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
return Ok(Self {
root: None,
current: None,
});
}
Err(error) => return Err(error),
};
if !metadata.is_dir() {
return Ok(Self {
root: Some(root),
current: None,
});
}
let current_path = path.join("CURRENT");
let (current_identity, current_metadata) = match MetadataIdentity::capture(¤t_path) {
Ok(captured) => captured,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
return Ok(Self {
root: Some(root),
current: None,
});
}
Err(error) => return Err(error),
};
if current_metadata.len() > 4096 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"disk graph CURRENT pointer exceeds 4096 bytes",
));
}
let mut bytes = Vec::with_capacity(current_metadata.len() as usize);
current_identity
.open_snapshot(¤t_path)?
.read_to_end(&mut bytes)?;
Ok(Self {
root: Some(root),
current: Some((current_identity, bytes)),
})
}
}
pub fn open_or_create_graph(
path: &Path,
create_mode: Option<StorageMode>,
) -> io::Result<OpenGraphResult> {
open_or_create_graph_logged(path, create_mode, DurabilityLevel::Off)
}
fn open_or_create_graph_logged(
path: &Path,
create_mode: Option<StorageMode>,
attaching_log: DurabilityLevel,
) -> io::Result<OpenGraphResult> {
match std::fs::metadata(path) {
Ok(_) => {
let before = GraphFileIdentity::capture(path)?;
let graph = load_file(&path.to_string_lossy())?;
let identity = GraphFileIdentity::capture(path)?;
if identity != before {
return Err(io::Error::new(
io::ErrorKind::WouldBlock,
format!("graph path {} changed while it was loading", path.display()),
));
}
if !attaching_log.logs() {
unrecovered_sidecar_check(path, graph.checkpoint_lsn)?;
}
return Ok(OpenGraphResult {
graph,
disposition: OpenDisposition::Opened,
identity,
converted_from: None,
});
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(error),
}
let mode = create_mode.ok_or_else(|| {
io::Error::new(
io::ErrorKind::NotFound,
format!(
"graph path '{}' does not exist and no creation storage mode was provided",
path.display()
),
)
})?;
if !attaching_log.logs() {
unrecovered_sidecar_check(path, 0)?;
}
let graph = new_dir_graph_in_mode(mode, Some(path))
.map_err(|message| io::Error::new(io::ErrorKind::InvalidInput, message))?;
Ok(OpenGraphResult {
graph: Arc::new(graph),
disposition: OpenDisposition::Created,
identity: GraphFileIdentity::capture(path)?,
converted_from: None,
})
}
fn unrecovered_sidecar_check(path: &Path, checkpoint_lsn: u64) -> io::Result<()> {
crate::graph::durability::ensure_recovered(path, checkpoint_lsn).map_err(|e| match e {
crate::graph::durability::DurableOpenError::Io(message) => io::Error::other(message),
other => io::Error::new(io::ErrorKind::InvalidData, other.to_string()),
})
}
pub fn open_or_create_graph_in_mode(
path: &Path,
requested: Option<StorageMode>,
attaching_log: DurabilityLevel,
) -> io::Result<OpenGraphResult> {
let mut opened = open_or_create_graph_logged(path, requested, attaching_log)?;
let Some(requested) = requested else {
return Ok(opened);
};
if opened.disposition == OpenDisposition::Created {
return Ok(opened);
}
let current = live_storage_mode(&opened.graph);
if current == requested {
return Ok(opened);
}
convert_dir_graph_to_mode(
crate::graph::handle::make_dir_graph_mut(&mut opened.graph),
requested,
)
.map_err(|message| io::Error::new(io::ErrorKind::InvalidInput, message))?;
opened.converted_from = Some(current);
Ok(opened)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::datatypes::Value;
use crate::graph::storage::GraphRead;
use crate::graph::wal::{wal_path, DurabilityLevel, MutationOp, SyncMode, Wal, WalFrame};
use std::process::Command;
fn checkpoint_with_pending_frame(path: &Path, checkpoint_lsn: u64, age: i64, lsn: u64) {
let mut graph = Arc::new(DirGraph::new());
crate::graph::mutation::wal_replay::apply_frames(
crate::graph::handle::make_dir_graph_mut(&mut graph),
&[person_frame(1, age)],
0,
)
.unwrap();
crate::graph::handle::make_dir_graph_mut(&mut graph).checkpoint_lsn = checkpoint_lsn;
crate::graph::io::file::save_graph(&mut graph, &path.to_string_lossy()).unwrap();
let mut wal = Wal::open(wal_path(path), SyncMode::Barrier).unwrap();
wal.append(&person_frame(lsn, age + 1)).unwrap();
}
fn person_frame(lsn: u64, age: i64) -> WalFrame {
WalFrame {
lsn,
ops: vec![MutationOp::UpsertNode {
node_type: "Person".into(),
id: Value::Int64(1),
title: Value::String("Alice".into()),
properties: vec![("age".to_string(), Value::Int64(age))],
}],
}
}
fn age_of(graph: &mut Arc<DirGraph>) -> Option<Value> {
let dir = crate::graph::handle::make_dir_graph_mut(graph);
let idx = dir.lookup_by_id("Person", &Value::Int64(1))?;
dir.graph
.node_view(idx)
.and_then(|n| n.get_field_ref("age").map(|c| c.into_owned()))
}
#[test]
fn a_stale_sidecar_cannot_be_replayed_over_a_newer_save() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("graph.kgl");
checkpoint_with_pending_frame(&path, 0, 1, 1);
let refusal =
open_or_create_graph_in_mode(&path, Some(StorageMode::Memory), DurabilityLevel::Off)
.err()
.expect("an open that attaches no log must not proceed over unfolded frames");
assert_eq!(refusal.kind(), io::ErrorKind::InvalidData);
let message = refusal.to_string();
assert!(message.contains("graph.kgl-wal"), "{message}");
assert!(message.contains("holds commits this checkpoint does not contain"));
assert!(message.contains("'full' or 'normal'"), "{message}");
assert!(message.contains("move the sidecar aside"), "{message}");
let mut recovered = crate::graph::io::file::load_file(&path.to_string_lossy()).unwrap();
crate::graph::durability::open_log(&mut recovered, &path, DurabilityLevel::Full).unwrap();
assert_eq!(age_of(&mut recovered), Some(Value::Int64(2)));
}
#[test]
fn a_caller_attaching_a_log_opens_the_same_stale_sidecar() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("graph.kgl");
checkpoint_with_pending_frame(&path, 0, 1, 1);
for level in [DurabilityLevel::Full, DurabilityLevel::Normal] {
let opened = open_or_create_graph_in_mode(&path, Some(StorageMode::Memory), level)
.unwrap_or_else(|e| panic!("durable={} must open to recover: {e}", level.name()));
let mut graph = opened.graph;
assert_eq!(age_of(&mut graph), Some(Value::Int64(1)));
crate::graph::durability::open_log(&mut graph, &path, level).unwrap();
assert_eq!(age_of(&mut graph), Some(Value::Int64(2)));
}
}
#[test]
fn an_orphaned_sidecar_refuses_creation_but_replays_for_a_log() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("orphan.kgl");
let mut wal = Wal::open(wal_path(&path), SyncMode::Barrier).unwrap();
wal.append(&person_frame(1, 42)).unwrap();
drop(wal);
open_or_create_graph_in_mode(&path, Some(StorageMode::Memory), DurabilityLevel::Off)
.err()
.expect("creating over an orphaned sidecar strands its commits");
let opened =
open_or_create_graph_in_mode(&path, Some(StorageMode::Memory), DurabilityLevel::Full)
.expect("a durable creator recovers the orphaned commits");
assert_eq!(opened.disposition, OpenDisposition::Created);
let mut graph = opened.graph;
crate::graph::durability::open_log(&mut graph, &path, DurabilityLevel::Full).unwrap();
assert_eq!(age_of(&mut graph), Some(Value::Int64(42)));
}
#[test]
fn frames_at_or_below_the_checkpoint_still_open() {
let tmp = tempfile::tempdir().unwrap();
let folded = tmp.path().join("folded.kgl");
checkpoint_with_pending_frame(&folded, 7, 1, 7);
let opened = open_or_create_graph(&folded, Some(StorageMode::Memory))
.expect("a frame the checkpoint already contains is harmless residue");
assert_eq!(opened.disposition, OpenDisposition::Opened);
let ahead = tmp.path().join("ahead.kgl");
checkpoint_with_pending_frame(&ahead, 7, 1, 8);
assert!(
open_or_create_graph(&ahead, Some(StorageMode::Memory)).is_err(),
"one frame past the checkpoint is unrecovered data"
);
}
#[test]
fn a_live_sidecar_beside_a_missing_checkpoint_refuses_creation() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("gone.kgl");
let mut wal = Wal::open(wal_path(&path), SyncMode::Barrier).unwrap();
wal.append(&person_frame(1, 2)).unwrap();
let refusal = open_or_create_graph(&path, Some(StorageMode::Memory))
.err()
.expect("creating over a live sidecar must be refused");
assert_eq!(refusal.kind(), io::ErrorKind::InvalidData);
assert!(!path.exists(), "a refused open must leave no graph behind");
}
#[test]
fn missing_path_requires_explicit_create_mode() {
let tmp = tempfile::tempdir().unwrap();
let missing = tmp.path().join("missing.kgl");
let err = open_or_create_graph(&missing, None)
.err()
.expect("missing path without create mode should fail");
assert_eq!(err.kind(), io::ErrorKind::NotFound);
assert!(err.to_string().contains("no creation storage mode"));
}
#[test]
fn creates_requested_storage_mode_when_path_is_missing() {
let tmp = tempfile::tempdir().unwrap();
let memory =
open_or_create_graph(&tmp.path().join("memory.kgl"), Some(StorageMode::Memory))
.unwrap();
assert_eq!(memory.disposition, OpenDisposition::Created);
assert!(!memory.graph.graph.is_mapped());
assert!(!memory.graph.graph.is_disk());
let disk_path = tmp.path().join("disk");
let disk = open_or_create_graph(&disk_path, Some(StorageMode::Disk)).unwrap();
assert_eq!(disk.disposition, OpenDisposition::Created);
assert!(disk.graph.graph.is_disk());
assert!(disk_path.is_dir());
}
#[test]
fn existing_graph_opens_in_the_mode_it_recorded() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("mapped.kgl");
let mut mapped = Arc::new(
crate::graph::storage::mode::new_dir_graph_in_mode(StorageMode::Mapped, None).unwrap(),
);
crate::graph::io::file::save_graph(&mut mapped, &path.to_string_lossy()).unwrap();
let opened = open_or_create_graph(&path, None).unwrap();
assert_eq!(opened.disposition, OpenDisposition::Opened);
assert!(
opened.graph.graph.is_mapped(),
"a mapped-saved checkpoint must reopen mapped with no storage argument at all"
);
let memory_path = tmp.path().join("memory.kgl");
let mut memory = Arc::new(DirGraph::new());
crate::graph::io::file::save_graph(&mut memory, &memory_path.to_string_lossy()).unwrap();
let opened = open_or_create_graph(&memory_path, Some(StorageMode::Mapped)).unwrap();
assert!(!opened.graph.graph.is_mapped());
}
#[test]
fn existing_graph_is_loaded_regardless_of_create_mode() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("existing.kgl");
let mut graph = Arc::new(DirGraph::new());
crate::graph::io::file::save_graph(&mut graph, &path.to_string_lossy()).unwrap();
let loaded = open_or_create_graph(&path, Some(StorageMode::Disk)).unwrap();
assert_eq!(loaded.disposition, OpenDisposition::Opened);
assert!(!loaded.graph.graph.is_disk());
}
#[test]
fn disk_identity_tracks_current_generation_content() {
let tmp = tempfile::tempdir().unwrap();
std::fs::write(tmp.path().join("CURRENT"), b"gen_00000000000000000001\n").unwrap();
let first = GraphFileIdentity::capture(tmp.path()).unwrap();
std::fs::write(tmp.path().join("CURRENT"), b"gen_00000000000000000002\n").unwrap();
let second = GraphFileIdentity::capture(tmp.path()).unwrap();
assert_ne!(first, second);
}
#[test]
fn writer_lease_child() {
let Some(graph_path) = std::env::var_os("KGLITE_LEASE_CHILD_GRAPH") else {
return;
};
let ready = std::env::var_os("KGLITE_LEASE_CHILD_READY").unwrap();
let _lease = GraphWriterLease::acquire(Path::new(&graph_path), Duration::ZERO).unwrap();
std::fs::write(ready, b"ready").unwrap();
std::thread::sleep(Duration::from_secs(60));
}
#[test]
fn crashed_process_releases_writer_lease() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("graph.kgl");
let ready = tmp.path().join("ready");
let mut child = Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"graph::io::open::tests::writer_lease_child",
"--nocapture",
])
.env("KGLITE_LEASE_CHILD_GRAPH", &graph)
.env("KGLITE_LEASE_CHILD_READY", &ready)
.spawn()
.unwrap();
let started = Instant::now();
while !ready.exists() && started.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(20));
}
assert!(ready.exists(), "child did not acquire lease");
assert!(GraphWriterLease::acquire(&graph, Duration::ZERO).is_err());
child.kill().unwrap();
child.wait().unwrap();
GraphWriterLease::acquire(&graph, Duration::from_secs(2)).unwrap();
}
#[test]
fn writer_lease_serializes_open_create_and_publish() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("barrier.kgl");
{
let _lease = GraphWriterLease::acquire(&path, Duration::ZERO).unwrap();
let mut created = open_or_create_graph(&path, Some(StorageMode::Memory)).unwrap();
assert_eq!(created.disposition, OpenDisposition::Created);
crate::graph::io::file::save_graph(&mut created.graph, &path.to_string_lossy())
.unwrap();
}
let _lease = GraphWriterLease::acquire(&path, Duration::ZERO).unwrap();
let opened = open_or_create_graph(&path, Some(StorageMode::Memory)).unwrap();
assert_eq!(opened.disposition, OpenDisposition::Opened);
}
#[test]
fn contended_message_names_the_holding_process() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("named.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let error = GraphWriterLease::acquire(&graph, Duration::ZERO)
.err()
.expect("second acquire must be refused");
let diagnosis = format!(
"got {error:?} (raw OS error {:?}, kind {:?}). A raw platform lock \
error here means `is_lock_contended` failed to classify it — the \
errno differs per platform (EWOULDBLOCK on Unix, \
ERROR_LOCK_VIOLATION on Windows), so an `ErrorKind` comparison \
recognises only Unix.",
error.raw_os_error(),
error.kind()
);
let message = error.to_string();
assert!(
message.contains(&format!("pid {}", std::process::id())),
"the refusal must name the holding process; {diagnosis}"
);
assert!(message.contains("named.kgl"), "{diagnosis}");
assert!(message.contains("does not release it"), "{diagnosis}");
}
#[test]
fn contention_is_classified_from_the_platform_lock_error() {
assert!(
is_lock_contended(&fs2::lock_contended_error()),
"the error fs2 returns for a contended lock must be recognised as \
contention on every platform"
);
assert!(is_lock_contended(&io::Error::from(
io::ErrorKind::WouldBlock
)));
assert!(!is_lock_contended(&io::Error::from(
io::ErrorKind::PermissionDenied
)));
}
#[test]
fn a_contended_acquire_waits_for_its_timeout() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("waiting.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let started = Instant::now();
let error = GraphWriterLease::acquire(&graph, Duration::from_millis(300))
.err()
.expect("a held lease must still be refused after the timeout");
assert!(
started.elapsed() >= Duration::from_millis(250),
"acquire returned after {:?}, so it never waited — contention was \
not recognised and the retry loop was skipped ({error})",
started.elapsed()
);
}
#[test]
fn owner_record_stays_readable_while_the_lease_is_held() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("readable.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let owner = writer_owner_path(&graph);
let text = std::fs::read_to_string(&owner).unwrap_or_else(|error| {
panic!(
"the owner record must stay readable while the lease is held, but \
reading {} failed: {error} (raw OS error {:?}). A mandatory-lock \
platform reports ERROR_LOCK_VIOLATION (33) here the moment the \
record is moved back inside the locked file.",
owner.display(),
error.raw_os_error()
)
});
assert!(
text.contains(&format!("pid={}", std::process::id())),
"{text}"
);
let lock = writer_lease_path(&graph);
assert_eq!(
std::fs::metadata(&lock).unwrap().len(),
0,
"the lock file must stay empty; its contents are unreadable to \
contenders on mandatory-lock platforms"
);
}
#[test]
fn lease_record_carries_pid_and_acquisition_time() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("record.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let holder = LeaseHolder::read(&writer_owner_path(&graph));
assert_eq!(holder.pid, Some(std::process::id()));
assert!(holder.since.is_some(), "acquisition time must be recorded");
}
#[test]
fn a_refusal_carries_the_holder_structured_not_only_in_prose() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("structured.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let refusal = GraphWriterLease::acquire_ex(&graph, Duration::ZERO)
.err()
.expect("a held lease must be refused");
let holder = refusal.holder.expect("a contention refusal names a holder");
assert_eq!(holder.pid, Some(std::process::id()));
assert!(
holder.since.is_some(),
"acquisition time must be structured"
);
assert!(holder.is_self(), "this process is the holder");
assert_eq!(refusal.error.kind(), io::ErrorKind::WouldBlock);
assert!(refusal.error.to_string().contains("this same process"));
}
#[test]
fn acquire_is_acquire_ex_projected_to_its_error() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("projection.kgl");
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let flat = GraphWriterLease::acquire(&graph, Duration::ZERO)
.err()
.unwrap();
let structured = GraphWriterLease::acquire_ex(&graph, Duration::ZERO)
.err()
.unwrap();
assert_eq!(flat.kind(), structured.error.kind());
assert_eq!(flat.to_string(), structured.error.to_string());
}
#[test]
fn a_non_contention_refusal_reports_no_holder() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("missing-dir").join("nowhere.kgl");
let refusal = GraphWriterLease::acquire_ex(&graph, Duration::ZERO)
.err()
.expect("an uncreatable lock sidecar must refuse");
assert!(refusal.holder.is_none(), "nobody holds an unopenable lock");
assert_ne!(refusal.error.kind(), io::ErrorKind::WouldBlock);
}
#[test]
fn owner_record_is_replaced_by_each_new_holder() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("succession.kgl");
let owner = writer_owner_path(&graph);
std::fs::write(&owner, b"pid=999999\nsince=long-ago\n").unwrap();
let _lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
let holder = LeaseHolder::read(&owner);
assert_eq!(
holder.pid,
Some(std::process::id()),
"a stale predecessor's pid must not survive a fresh acquisition"
);
assert_ne!(holder.since.as_deref(), Some("long-ago"));
}
#[test]
fn holder_description_degrades_without_a_readable_record() {
let tmp = tempfile::tempdir().unwrap();
let missing = tmp.path().join("absent.lock");
assert_eq!(
LeaseHolder::read_once(&missing).describe(),
"another process"
);
let legacy = tmp.path().join("legacy.lock");
std::fs::write(&legacy, b"pid=999999\n").unwrap();
assert_eq!(LeaseHolder::read_once(&legacy).describe(), "pid 999999");
}
#[test]
fn dropping_lease_does_not_delete_replacement_path() {
let tmp = tempfile::tempdir().unwrap();
let graph = tmp.path().join("replacement.kgl");
let lock = writer_lease_path(&graph);
let moved = tmp.path().join("moved.lock");
let lease = GraphWriterLease::acquire(&graph, Duration::ZERO).unwrap();
std::fs::rename(&lock, &moved).unwrap();
std::fs::write(&lock, b"replacement\n").unwrap();
drop(lease);
assert_eq!(std::fs::read(&lock).unwrap(), b"replacement\n");
}
}